[feature](pipelineX) support paritition tablet sink shuffle #31689
Conversation
|
Thank you for your contribution to Apache Doris. |
|
run buildall |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| return Status::OK(); | ||
| } | ||
|
|
||
| Status ExchangeSinkLocalState::_send_new_partition_batch() { |
There was a problem hiding this comment.
warning: method '_send_new_partition_batch' can be made static [readability-convert-member-functions-to-static]
be/src/pipeline/exec/exchange_sink_operator.h:127:
- Status _send_new_partition_batch();
+ static Status _send_new_partition_batch();| } | ||
|
|
||
| template <typename Channels> | ||
| Status ExchangeSinkOperatorX::channel_add_rows_with_idx( |
There was a problem hiding this comment.
warning: function 'channel_add_rows_with_idx' has cognitive complexity of 52 (threshold 50) [readability-function-cognitive-complexity]
Status ExchangeSinkOperatorX::channel_add_rows_with_idx(
^Additional context
be/src/pipeline/exec/exchange_sink_operator.cpp:577: +1, including nesting penalty of 0, nesting level increased to 1
for (int i = 0; i < num_channels; ++i) {
^be/src/pipeline/exec/exchange_sink_operator.cpp:578: +2, including nesting penalty of 1, nesting level increased to 2
if (!channels[i]->is_receiver_eof() && !channel2rows[i].empty()) {
^be/src/pipeline/exec/exchange_sink_operator.cpp:578: +1
if (!channels[i]->is_receiver_eof() && !channel2rows[i].empty()) {
^be/src/pipeline/exec/exchange_sink_operator.cpp:580: +3, including nesting penalty of 2, nesting level increased to 3
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:419: expanded from macro 'HANDLE_CHANNEL_STATUS'
do { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:580: +4, including nesting penalty of 3, nesting level increased to 4
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:420: expanded from macro 'HANDLE_CHANNEL_STATUS'
if (status.is<ErrorCode::END_OF_FILE>()) { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:580: +1, nesting level increased to 4
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:422: expanded from macro 'HANDLE_CHANNEL_STATUS'
} else { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:580: +5, including nesting penalty of 4, nesting level increased to 5
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:423: expanded from macro 'HANDLE_CHANNEL_STATUS'
RETURN_IF_ERROR(status); \
^be/src/common/status.h:541: expanded from macro 'RETURN_IF_ERROR'
do { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:580: +6, including nesting penalty of 5, nesting level increased to 6
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:423: expanded from macro 'HANDLE_CHANNEL_STATUS'
RETURN_IF_ERROR(status); \
^be/src/common/status.h:543: expanded from macro 'RETURN_IF_ERROR'
if (UNLIKELY(!_status_.ok())) { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:585: +1, including nesting penalty of 0, nesting level increased to 1
if (eos) {
^be/src/pipeline/exec/exchange_sink_operator.cpp:586: +2, including nesting penalty of 1, nesting level increased to 2
for (int i = 0; i < num_channels; ++i) {
^be/src/pipeline/exec/exchange_sink_operator.cpp:587: +3, including nesting penalty of 2, nesting level increased to 3
if (!channels[i]->is_receiver_eof()) {
^be/src/pipeline/exec/exchange_sink_operator.cpp:589: +4, including nesting penalty of 3, nesting level increased to 4
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:419: expanded from macro 'HANDLE_CHANNEL_STATUS'
do { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:589: +5, including nesting penalty of 4, nesting level increased to 5
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:420: expanded from macro 'HANDLE_CHANNEL_STATUS'
if (status.is<ErrorCode::END_OF_FILE>()) { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:589: +1, nesting level increased to 5
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:422: expanded from macro 'HANDLE_CHANNEL_STATUS'
} else { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:589: +6, including nesting penalty of 5, nesting level increased to 6
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:423: expanded from macro 'HANDLE_CHANNEL_STATUS'
RETURN_IF_ERROR(status); \
^be/src/common/status.h:541: expanded from macro 'RETURN_IF_ERROR'
do { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:589: +7, including nesting penalty of 6, nesting level increased to 7
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:423: expanded from macro 'HANDLE_CHANNEL_STATUS'
RETURN_IF_ERROR(status); \
^be/src/common/status.h:543: expanded from macro 'RETURN_IF_ERROR'
if (UNLIKELY(!_status_.ok())) { \
^
|
TeamCity be ut coverage result: |
|
run buildall |
| } | ||
|
|
||
| template <typename Channels> | ||
| Status ExchangeSinkOperatorX::channel_add_rows_with_idx( |
There was a problem hiding this comment.
warning: function 'channel_add_rows_with_idx' has cognitive complexity of 52 (threshold 50) [readability-function-cognitive-complexity]
Status ExchangeSinkOperatorX::channel_add_rows_with_idx(
^Additional context
be/src/pipeline/exec/exchange_sink_operator.cpp:578: +1, including nesting penalty of 0, nesting level increased to 1
for (int i = 0; i < num_channels; ++i) {
^be/src/pipeline/exec/exchange_sink_operator.cpp:579: +2, including nesting penalty of 1, nesting level increased to 2
if (!channels[i]->is_receiver_eof() && !channel2rows[i].empty()) {
^be/src/pipeline/exec/exchange_sink_operator.cpp:579: +1
if (!channels[i]->is_receiver_eof() && !channel2rows[i].empty()) {
^be/src/pipeline/exec/exchange_sink_operator.cpp:581: +3, including nesting penalty of 2, nesting level increased to 3
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:419: expanded from macro 'HANDLE_CHANNEL_STATUS'
do { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:581: +4, including nesting penalty of 3, nesting level increased to 4
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:420: expanded from macro 'HANDLE_CHANNEL_STATUS'
if (status.is<ErrorCode::END_OF_FILE>()) { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:581: +1, nesting level increased to 4
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:422: expanded from macro 'HANDLE_CHANNEL_STATUS'
} else { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:581: +5, including nesting penalty of 4, nesting level increased to 5
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:423: expanded from macro 'HANDLE_CHANNEL_STATUS'
RETURN_IF_ERROR(status); \
^be/src/common/status.h:541: expanded from macro 'RETURN_IF_ERROR'
do { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:581: +6, including nesting penalty of 5, nesting level increased to 6
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:423: expanded from macro 'HANDLE_CHANNEL_STATUS'
RETURN_IF_ERROR(status); \
^be/src/common/status.h:543: expanded from macro 'RETURN_IF_ERROR'
if (UNLIKELY(!_status_.ok())) { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:586: +1, including nesting penalty of 0, nesting level increased to 1
if (eos) {
^be/src/pipeline/exec/exchange_sink_operator.cpp:587: +2, including nesting penalty of 1, nesting level increased to 2
for (int i = 0; i < num_channels; ++i) {
^be/src/pipeline/exec/exchange_sink_operator.cpp:588: +3, including nesting penalty of 2, nesting level increased to 3
if (!channels[i]->is_receiver_eof()) {
^be/src/pipeline/exec/exchange_sink_operator.cpp:590: +4, including nesting penalty of 3, nesting level increased to 4
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:419: expanded from macro 'HANDLE_CHANNEL_STATUS'
do { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:590: +5, including nesting penalty of 4, nesting level increased to 5
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:420: expanded from macro 'HANDLE_CHANNEL_STATUS'
if (status.is<ErrorCode::END_OF_FILE>()) { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:590: +1, nesting level increased to 5
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:422: expanded from macro 'HANDLE_CHANNEL_STATUS'
} else { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:590: +6, including nesting penalty of 5, nesting level increased to 6
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:423: expanded from macro 'HANDLE_CHANNEL_STATUS'
RETURN_IF_ERROR(status); \
^be/src/common/status.h:541: expanded from macro 'RETURN_IF_ERROR'
do { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:590: +7, including nesting penalty of 6, nesting level increased to 7
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:423: expanded from macro 'HANDLE_CHANNEL_STATUS'
RETURN_IF_ERROR(status); \
^be/src/common/status.h:543: expanded from macro 'RETURN_IF_ERROR'
if (UNLIKELY(!_status_.ok())) { \
^
|
TeamCity be ut coverage result: |
|
run buildall |
| } | ||
|
|
||
| template <typename Channels> | ||
| Status ExchangeSinkOperatorX::channel_add_rows_with_idx( |
There was a problem hiding this comment.
warning: function 'channel_add_rows_with_idx' has cognitive complexity of 52 (threshold 50) [readability-function-cognitive-complexity]
Status ExchangeSinkOperatorX::channel_add_rows_with_idx(
^Additional context
be/src/pipeline/exec/exchange_sink_operator.cpp:581: +1, including nesting penalty of 0, nesting level increased to 1
for (int i = 0; i < num_channels; ++i) {
^be/src/pipeline/exec/exchange_sink_operator.cpp:582: +2, including nesting penalty of 1, nesting level increased to 2
if (!channels[i]->is_receiver_eof() && !channel2rows[i].empty()) {
^be/src/pipeline/exec/exchange_sink_operator.cpp:582: +1
if (!channels[i]->is_receiver_eof() && !channel2rows[i].empty()) {
^be/src/pipeline/exec/exchange_sink_operator.cpp:584: +3, including nesting penalty of 2, nesting level increased to 3
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:419: expanded from macro 'HANDLE_CHANNEL_STATUS'
do { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:584: +4, including nesting penalty of 3, nesting level increased to 4
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:420: expanded from macro 'HANDLE_CHANNEL_STATUS'
if (status.is<ErrorCode::END_OF_FILE>()) { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:584: +1, nesting level increased to 4
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:422: expanded from macro 'HANDLE_CHANNEL_STATUS'
} else { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:584: +5, including nesting penalty of 4, nesting level increased to 5
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:423: expanded from macro 'HANDLE_CHANNEL_STATUS'
RETURN_IF_ERROR(status); \
^be/src/common/status.h:541: expanded from macro 'RETURN_IF_ERROR'
do { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:584: +6, including nesting penalty of 5, nesting level increased to 6
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:423: expanded from macro 'HANDLE_CHANNEL_STATUS'
RETURN_IF_ERROR(status); \
^be/src/common/status.h:543: expanded from macro 'RETURN_IF_ERROR'
if (UNLIKELY(!_status_.ok())) { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:589: +1, including nesting penalty of 0, nesting level increased to 1
if (eos) {
^be/src/pipeline/exec/exchange_sink_operator.cpp:590: +2, including nesting penalty of 1, nesting level increased to 2
for (int i = 0; i < num_channels; ++i) {
^be/src/pipeline/exec/exchange_sink_operator.cpp:591: +3, including nesting penalty of 2, nesting level increased to 3
if (!channels[i]->is_receiver_eof()) {
^be/src/pipeline/exec/exchange_sink_operator.cpp:593: +4, including nesting penalty of 3, nesting level increased to 4
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:419: expanded from macro 'HANDLE_CHANNEL_STATUS'
do { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:593: +5, including nesting penalty of 4, nesting level increased to 5
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:420: expanded from macro 'HANDLE_CHANNEL_STATUS'
if (status.is<ErrorCode::END_OF_FILE>()) { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:593: +1, nesting level increased to 5
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:422: expanded from macro 'HANDLE_CHANNEL_STATUS'
} else { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:593: +6, including nesting penalty of 5, nesting level increased to 6
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:423: expanded from macro 'HANDLE_CHANNEL_STATUS'
RETURN_IF_ERROR(status); \
^be/src/common/status.h:541: expanded from macro 'RETURN_IF_ERROR'
do { \
^be/src/pipeline/exec/exchange_sink_operator.cpp:593: +7, including nesting penalty of 6, nesting level increased to 7
HANDLE_CHANNEL_STATUS(state, channels[i], status);
^be/src/vec/sink/vdata_stream_sender.h:423: expanded from macro 'HANDLE_CHANNEL_STATUS'
RETURN_IF_ERROR(status); \
^be/src/common/status.h:543: expanded from macro 'RETURN_IF_ERROR'
if (UNLIKELY(!_status_.ok())) { \
^
|
TeamCity be ut coverage result: |
|
run buildall |
|
TeamCity be ut coverage result: |
|
run buildall |
be/src/olap/tablet.cpp
Outdated
| @@ -181,6 +181,11 @@ Status _load_rowset_segments(const RowsetSharedPtr& rowset, | |||
|
|
|||
| struct WriteCooldownMetaExecutors { | |||
| WriteCooldownMetaExecutors(size_t executor_nums = 5); | |||
| ~WriteCooldownMetaExecutors() { | |||
There was a problem hiding this comment.
warning: use '= default' to define a trivial destructor [modernize-use-equals-default]
~WriteCooldownMetaExecutors() {
^
|
TeamCity be ut coverage result: |
|
run buildall |
1 similar comment
|
run buildall |
zhangstar333
force-pushed
the
tablet_sinke_x
branch
from
343352d to
48dc1c0
Compare
|
run buildall |
TPC-H: Total hot run time: 37782 ms |
TPC-DS: Total hot run time: 186033 ms |
ClickBench: Total hot run time: 31.18 s |
|
Load test result on machine: 'aliyun_ecs.c7a.8xlarge_32C64G' |
|
run buildall |
TPC-H: Total hot run time: 37715 ms |
|
TeamCity be ut coverage result: |
TPC-DS: Total hot run time: 185562 ms |
ClickBench: Total hot run time: 31.08 s |
|
Load test result on machine: 'aliyun_ecs.c7a.8xlarge_32C64G' |
|
run buildall |
TPC-H: Total hot run time: 36443 ms |
TPC-DS: Total hot run time: 186534 ms |
ClickBench: Total hot run time: 29.49 s |
|
Load test result on machine: 'aliyun_ecs.c7a.8xlarge_32C64G' |
|
TeamCity be ut coverage result: |
github-actions
bot
added
the
approved
|
PR approved by at least one committer and no changes requested. |
|
PR approved by anyone and no changes requested. |
Proposed changes
Issue Number: close #xxx
Further comments
If this is a relatively large or complex change, kick off the discussion at dev@doris.apache.org by explaining why you chose the solution you did and what alternatives you considered, etc...