You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "mrhhsg (via GitHub)" <gi...@apache.org> on 2024/10/16 09:00:12 UTC
[PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
mrhhsg opened a new pull request, #41968:
URL: https://github.com/apache/doris/pull/41968
## Proposed changes
Issue Number: close #xxx
<!--Describe your changes.-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "mrhhsg (via GitHub)" <gi...@apache.org>.
mrhhsg commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2417030968
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1804072426
##########
be/src/vec/sink/vdata_stream_sender.cpp:
##########
@@ -309,67 +215,52 @@ Status Channel<Parent>::close_wait(RuntimeState* state) {
return Status::OK();
}
-template <typename Parent>
-Status Channel<Parent>::close_internal(Status exec_status) {
- if (!_need_close) {
+Status Channel::close(RuntimeState* state) {
+ if (_closed) {
return Status::OK();
}
- VLOG_RPC << "Channel::close_internal() instance_id=" << print_id(_fragment_instance_id)
- << " dest_node=" << _dest_node_id << " #rows= "
- << ((_serializer.get_block() == nullptr) ? 0 : _serializer.get_block()->rows())
- << " receiver status: " << _receiver_status << ", exec_status: " << exec_status;
+ _closed = true;
+
if (is_receiver_eof()) {
_serializer.reset_block();
- return Status::OK();
}
- Status status;
- if (_serializer.get_block() != nullptr && _serializer.get_block()->rows() > 0) {
- status = send_current_block(true, exec_status);
- } else {
- if (is_local()) {
- if (_recvr_is_valid()) {
- _local_recvr->remove_sender(_parent->sender_id(), _be_number, exec_status);
- }
- } else {
- // Non pipeline engine will send an empty eos block
- status = send_remote_block((PBlock*)nullptr, true, exec_status);
+
+ if (is_local()) {
+ if (_recvr_is_valid()) {
+ _local_recvr->remove_sender(_parent->sender_id(), _be_number, Status::OK());
}
- }
- // Don't wait for the last packet to finish, left it to close_wait.
- if (status.is<ErrorCode::END_OF_FILE>()) {
- return Status::OK();
} else {
- return status;
+ RETURN_IF_ERROR(send_remote_block(nullptr, true, Status::OK()));
}
+ return Status::OK();
}
-template <typename Parent>
-Status Channel<Parent>::close(RuntimeState* state, Status exec_status) {
- if (_closed) {
+Status Channel::_wait_last_brpc() {
Review Comment:
我们都走exchange sink buffer了，为啥还需要wait rpc，这个操作应该是在exchange sink buffer 里完成的
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2419824953
TeamCity be ut coverage result:
Function Coverage: 37.48% (9709/25905)
Line Coverage: 28.74% (80587/280434)
Region Coverage: 28.19% (41709/147948)
Branch Coverage: 24.76% (21205/85646)
Coverage Report: http://coverage.selectdb-in.cc/coverage/15975aad468e15ff0ce148bfc479e68c07f816ac_15975aad468e15ff0ce148bfc479e68c07f816ac/report/index.html
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei merged PR #41968:
URL: https://github.com/apache/doris/pull/41968
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "mrhhsg (via GitHub)" <gi...@apache.org>.
mrhhsg commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2419063637
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2416172878
clang-tidy review says "All clean, LGTM! :+1:"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2425535281
TeamCity be ut coverage result:
Function Coverage: 37.49% (9709/25899)
Line Coverage: 28.75% (80615/280371)
Region Coverage: 28.20% (41715/147923)
Branch Coverage: 24.77% (21210/85618)
Coverage Report: http://coverage.selectdb-in.cc/coverage/aadf14acbd38302bd3a804e99d5c1686bf70454a_aadf14acbd38302bd3a804e99d5c1686bf70454a/report/index.html
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2426293972
TeamCity be ut coverage result:
Function Coverage: 37.47% (9699/25882)
Line Coverage: 28.71% (80490/280339)
Region Coverage: 28.16% (41647/147909)
Branch Coverage: 24.73% (21168/85612)
Coverage Report: http://coverage.selectdb-in.cc/coverage/ba3097e797a59bb23380739e83eac032a91e8f68_ba3097e797a59bb23380739e83eac032a91e8f68/report/index.html
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "Gabriel39 (via GitHub)" <gi...@apache.org>.
Gabriel39 commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1808013816
##########
be/src/pipeline/exec/exchange_sink_operator.cpp:
##########
@@ -106,6 +111,24 @@ Status ExchangeSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo& inf
return Status::OK();
}
+void ExchangeSinkLocalState::on_channel_finished(InstanceLoId channel_id) {
+ std::lock_guard<std::mutex> lock(_finished_channels_mutex);
+
+ if (_finished_channels.contains(channel_id)) {
+ LOG(WARNING) << "query: " << print_id(_state->query_id())
+ << ", on_channel_finished on already finished channel: " << channel_id;
+ return;
+ } else {
+ _finished_channels.emplace(channel_id);
+ if (_working_channels_count.fetch_sub(1) == 1) {
+ set_reach_limit();
+ if (_finish_dependency) {
+ _finish_dependency->set_ready();
Review Comment:
`_finish_dependency` should never block if only local channels remain
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2422001784
PR approved by anyone and no changes requested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1808662248
##########
be/src/pipeline/exec/exchange_sink_buffer.cpp:
##########
@@ -470,18 +424,13 @@ bool ExchangeSinkBuffer::_is_receiver_eof(InstanceLoId id) {
return _instance_to_receiver_eof[id];
}
-void ExchangeSinkBuffer::_turn_off_channel(InstanceLoId id, bool cleanup) {
+void ExchangeSinkBuffer::_turn_off_channel(InstanceLoId id) {
Review Comment:
Just add the lock as a reference parameter, then we could make sure we locked. like _turn_off_channel(InstanceLoId id, std::unique_lock<std::mutex>&)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1808662248
##########
be/src/pipeline/exec/exchange_sink_buffer.cpp:
##########
@@ -470,18 +424,13 @@ bool ExchangeSinkBuffer::_is_receiver_eof(InstanceLoId id) {
return _instance_to_receiver_eof[id];
}
-void ExchangeSinkBuffer::_turn_off_channel(InstanceLoId id, bool cleanup) {
+void ExchangeSinkBuffer::_turn_off_channel(InstanceLoId id) {
Review Comment:
Just add the lock as a reference parameter, then we could make sure we locked.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1806070765
##########
be/src/pipeline/exec/exchange_sink_operator.cpp:
##########
@@ -358,7 +373,7 @@ void ExchangeSinkOperatorX::_handle_eof_channel(RuntimeState* state, ChannelPtrT
Status st) {
channel->set_receiver_eof(st);
// Chanel will not send RPC to the downstream when eof, so close chanel by OK status.
- static_cast<void>(channel->close(state, Status::OK()));
+ static_cast<void>(channel->close(state));
}
Status ExchangeSinkOperatorX::sink(RuntimeState* state, vectorized::Block* block, bool eos) {
Review Comment:
warning: function 'sink' has cognitive complexity of 236 (threshold 50) [readability-function-cognitive-complexity]
```cpp
Status ExchangeSinkOperatorX::sink(RuntimeState* state, vectorized::Block* block, bool eos) {
^
```
<details>
<summary>Additional context</summary>
**be/src/pipeline/exec/exchange_sink_operator.cpp:390:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (all_receiver_eof) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:398:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (_part_type == TPartitionType::UNPARTITIONED || local_state.channels.size() == 1) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:402:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (local_state.only_local_exchange) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:403:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (!block->empty()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:418:** +1, nesting level increased to 2
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:422:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._serializer.next_serialized_block(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:422:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.next_serialized_block(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:425:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (serialized) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:427:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
if (!cur_block.empty()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:428:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:428:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:431:** +1, nesting level increased to 4
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:457:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
if (moved) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:459:** +1, nesting level increased to 4
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:467:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::RANDOM) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:470:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (!current_channel->is_receiver_eof()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:472:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (current_channel->is_local()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:474:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:233:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:474:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:234:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:474:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:474:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:474:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:475:** +1, nesting level increased to 3
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:477:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:477:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:479:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:233:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:479:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:234:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:479:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:479:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:479:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:484:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::HASH_PARTITIONED ||
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:489:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:489:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:495:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (_part_type == TPartitionType::HASH_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:496:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:496:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:499:** +1, nesting level increased to 2
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:500:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:500:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:512:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLET_SINK_SHUFFLE_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:518:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:518:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:525:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (input_rows > 0) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:530:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._row_distribution.generate_rows_distribution(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:530:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._row_distribution.generate_rows_distribution(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:536:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
for (int idx = 0; idx < row_ids.size(); ++idx) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:544:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (eos) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:546:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:546:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:550:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(state, local_state.channels, num_channels,
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:550:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(state, local_state.channels, num_channels,
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:560:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLE_SINK_HASH_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:567:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:567:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:571:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:571:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:582:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLE_SINK_RANDOM_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:586:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (!current_channel->is_receiver_eof()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:588:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (current_channel->is_local()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:590:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:233:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:590:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:234:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:590:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:590:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:590:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:591:** +1, nesting level increased to 3
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:593:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:593:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:595:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:233:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:595:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:234:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:595:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:595:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:595:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:600:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (_writer_count < local_state.channels.size()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:601:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (_data_processed >=
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:608:** +1, nesting level increased to 1
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:615:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (eos) {
^
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "mrhhsg (via GitHub)" <gi...@apache.org>.
mrhhsg commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2425472147
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "mrhhsg (via GitHub)" <gi...@apache.org>.
mrhhsg commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2425797816
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "mrhhsg (via GitHub)" <gi...@apache.org>.
mrhhsg commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2419758062
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2417338591
TeamCity be ut coverage result:
Function Coverage: 37.47% (9709/25911)
Line Coverage: 28.73% (80581/280503)
Region Coverage: 28.18% (41698/147982)
Branch Coverage: 24.75% (21204/85656)
Coverage Report: http://coverage.selectdb-in.cc/coverage/f86b2320f3decbb6b15e7791c9851d8ed63ce3c2_f86b2320f3decbb6b15e7791c9851d8ed63ce3c2/report/index.html
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1803383750
##########
be/src/vec/sink/vdata_stream_sender.cpp:
##########
@@ -166,131 +166,51 @@ Status PipChannel::send_broadcast_block(std::shared_ptr<BroadcastPBlockHolder>&
return Status::OK();
}
-Status PipChannel::send_current_block(bool eos, Status exec_status) {
- if (Channel<pipeline::ExchangeSinkLocalState>::is_local()) {
- return Channel<pipeline::ExchangeSinkLocalState>::send_local_block(exec_status, eos);
- }
- RETURN_IF_ERROR(send_remote_block(_pblock.release(), eos, exec_status));
- return Status::OK();
-}
-
-template <typename Parent>
-Status Channel<Parent>::send_current_block(bool eos, Status exec_status) {
- // FIXME: Now, local exchange will cause the performance problem is in a multi-threaded scenario
- // so this feature is turned off here by default. We need to re-examine this logic
+Status PipChannel::_send_current_block(bool eos, Status exec_status) {
Review Comment:
把pipchannel和channel 合并了吧，代码特别乱
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2425859233
TeamCity be ut coverage result:
Function Coverage: 37.50% (9709/25893)
Line Coverage: 28.77% (80654/280355)
Region Coverage: 28.21% (41724/147919)
Branch Coverage: 24.78% (21216/85620)
Coverage Report: http://coverage.selectdb-in.cc/coverage/1bd9b09dcf9a2969a3b3fb8fa0a9ca76ee448a34_1bd9b09dcf9a2969a3b3fb8fa0a9ca76ee448a34/report/index.html
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "mrhhsg (via GitHub)" <gi...@apache.org>.
mrhhsg commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2428615109
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "Gabriel39 (via GitHub)" <gi...@apache.org>.
Gabriel39 commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2430676607
Need to summary this PR in description
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "mrhhsg (via GitHub)" <gi...@apache.org>.
mrhhsg commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2421760478
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2429034922
PR approved by at least one committer and no changes requested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2421742050
TeamCity be ut coverage result:
Function Coverage: 37.49% (9711/25903)
Line Coverage: 28.75% (80615/280409)
Region Coverage: 28.20% (41727/147946)
Branch Coverage: 24.76% (21209/85644)
Coverage Report: http://coverage.selectdb-in.cc/coverage/84c44c81889f438aea6cb6abea3754504ece28ab_84c44c81889f438aea6cb6abea3754504ece28ab/report/index.html
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1804068926
##########
be/src/vec/sink/vdata_stream_sender.cpp:
##########
@@ -166,135 +159,48 @@ Status PipChannel::send_broadcast_block(std::shared_ptr<BroadcastPBlockHolder>&
return Status::OK();
}
-Status PipChannel::send_current_block(bool eos, Status exec_status) {
- if (Channel<pipeline::ExchangeSinkLocalState>::is_local()) {
- return Channel<pipeline::ExchangeSinkLocalState>::send_local_block(exec_status, eos);
- }
- RETURN_IF_ERROR(send_remote_block(_pblock.release(), eos, exec_status));
- return Status::OK();
-}
-
-template <typename Parent>
-Status Channel<Parent>::send_current_block(bool eos, Status exec_status) {
- // FIXME: Now, local exchange will cause the performance problem is in a multi-threaded scenario
- // so this feature is turned off here by default. We need to re-examine this logic
+Status Channel::_send_current_block(bool eos, Status exec_status) {
if (is_local()) {
- return send_local_block(exec_status, eos);
+ return _send_local_block(exec_status, eos);
}
- if (eos) {
- RETURN_IF_ERROR(_serializer.serialize_block(_ch_cur_pb_block, 1));
- }
- RETURN_IF_ERROR(send_remote_block(_ch_cur_pb_block, eos, exec_status));
- ch_roll_pb_block();
+ RETURN_IF_ERROR(send_remote_block(std::move(_pblock), eos, exec_status));
return Status::OK();
}
-template <typename Parent>
-Status Channel<Parent>::send_local_block(Status exec_status, bool eos) {
+Status Channel::_send_local_block(Status exec_status, bool eos) {
SCOPED_TIMER(_parent->local_send_timer());
Block block = _serializer.get_block()->to_block();
_serializer.get_block()->set_mutable_columns(block.clone_empty_columns());
- if (_recvr_is_valid()) {
- if constexpr (!std::is_same_v<pipeline::ResultFileSinkLocalState, Parent>) {
- COUNTER_UPDATE(_parent->local_bytes_send_counter(), block.bytes());
- COUNTER_UPDATE(_parent->local_sent_rows(), block.rows());
- COUNTER_UPDATE(_parent->blocks_sent_counter(), 1);
- }
- _local_recvr->add_block(&block, _parent->sender_id(), true);
- if (eos) {
- _local_recvr->remove_sender(_parent->sender_id(), _be_number, exec_status);
- }
- return Status::OK();
- } else {
- _serializer.reset_block();
- return _receiver_status;
+ if (!block.empty() || eos) {
+ RETURN_IF_ERROR(send_local_block(&block, eos, true));
}
+ _serializer.reset_block();
+ return Status::OK();
}
-template <typename Parent>
-Status Channel<Parent>::send_local_block(Block* block, bool can_be_moved) {
+Status Channel::send_local_block(Block* block, bool eos, bool can_be_moved) {
SCOPED_TIMER(_parent->local_send_timer());
if (_recvr_is_valid()) {
- if constexpr (!std::is_same_v<pipeline::ResultFileSinkLocalState, Parent>) {
- COUNTER_UPDATE(_parent->local_bytes_send_counter(), block->bytes());
- COUNTER_UPDATE(_parent->local_sent_rows(), block->rows());
- COUNTER_UPDATE(_parent->blocks_sent_counter(), 1);
- }
- _local_recvr->add_block(block, _parent->sender_id(), can_be_moved);
- return Status::OK();
- } else {
- return _receiver_status;
- }
-}
-
-template <typename Parent>
-Status Channel<Parent>::send_remote_block(PBlock* block, bool eos, Status exec_status) {
- if constexpr (!std::is_same_v<pipeline::ResultFileSinkLocalState, Parent>) {
+ COUNTER_UPDATE(_parent->local_bytes_send_counter(), block->bytes());
+ COUNTER_UPDATE(_parent->local_sent_rows(), block->rows());
COUNTER_UPDATE(_parent->blocks_sent_counter(), 1);
- }
- SCOPED_TIMER(_parent->brpc_send_timer());
- if (_send_remote_block_callback == nullptr) {
- _send_remote_block_callback = DummyBrpcCallback<PTransmitDataResult>::create_shared();
- } else {
- RETURN_IF_ERROR(_wait_last_brpc());
- _send_remote_block_callback->cntl_->Reset();
- }
- VLOG_ROW << "Channel<Parent>::send_batch() instance_id=" << print_id(_fragment_instance_id)
- << " dest_node=" << _dest_node_id << " to_host=" << _brpc_dest_addr.hostname
- << " _packet_seq=" << _packet_seq << " row_desc=" << _row_desc.debug_string();
-
- _brpc_request->set_eos(eos);
- if (!exec_status.ok()) {
- exec_status.to_protobuf(_brpc_request->mutable_exec_status());
- }
- if (block != nullptr && !block->column_metas().empty()) {
- _brpc_request->set_allocated_block(block);
- }
- _brpc_request->set_packet_seq(_packet_seq++);
-
- _send_remote_block_callback->cntl_->set_timeout_ms(_brpc_timeout_ms);
- if (config::exchange_sink_ignore_eovercrowded) {
- _send_remote_block_callback->cntl_->ignore_eovercrowded();
- }
-
- {
- auto send_remote_block_closure =
- AutoReleaseClosure<PTransmitDataParams, DummyBrpcCallback<PTransmitDataResult>>::
- create_unique(_brpc_request, _send_remote_block_callback);
- if (enable_http_send_block(*_brpc_request)) {
- RETURN_IF_ERROR(transmit_block_httpv2(
- _state->exec_env(), std::move(send_remote_block_closure), _brpc_dest_addr));
- } else {
- transmit_blockv2(*_brpc_stub, std::move(send_remote_block_closure));
+ const auto sender_id = _parent->sender_id();
+ if (!block->empty()) [[likely]] {
+ _local_recvr->add_block(block, sender_id, can_be_moved);
}
- }
-
- if (block != nullptr) {
- static_cast<void>(_brpc_request->release_block());
- }
- return Status::OK();
-}
-template <typename Parent>
-Status Channel<Parent>::add_rows(Block* block, const std::vector<uint32_t>& rows, bool eos) {
- if (_fragment_instance_id.lo == -1) {
+ if (eos) [[unlikely]] {
+ _local_recvr->remove_sender(sender_id, _be_number, Status::OK());
+ }
return Status::OK();
+ } else {
+ return _receiver_status;
}
-
- bool serialized = false;
- RETURN_IF_ERROR(
- _serializer.next_serialized_block(block, _ch_cur_pb_block, 1, &serialized, eos, &rows));
- if (serialized) {
- RETURN_IF_ERROR(send_current_block(false, Status::OK()));
- }
-
- return Status::OK();
}
-template <typename Parent>
-Status Channel<Parent>::close_wait(RuntimeState* state) {
+Status Channel::close_wait(RuntimeState* state) {
Review Comment:
这个close wait 函数还有意义吗？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2419121505
TeamCity be ut coverage result:
Function Coverage: 37.49% (9711/25905)
Line Coverage: 28.74% (80594/280433)
Region Coverage: 28.20% (41717/147948)
Branch Coverage: 24.76% (21207/85646)
Coverage Report: http://coverage.selectdb-in.cc/coverage/fd8882c3b52989f328e2aad92880f4edd4e42a2c_fd8882c3b52989f328e2aad92880f4edd4e42a2c/report/index.html
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "mrhhsg (via GitHub)" <gi...@apache.org>.
mrhhsg commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2421647482
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1808221547
##########
be/src/pipeline/exec/exchange_sink_operator.cpp:
##########
@@ -358,7 +383,7 @@ void ExchangeSinkOperatorX::_handle_eof_channel(RuntimeState* state, ChannelPtrT
Status st) {
channel->set_receiver_eof(st);
// Chanel will not send RPC to the downstream when eof, so close chanel by OK status.
- static_cast<void>(channel->close(state, Status::OK()));
+ static_cast<void>(channel->close(state));
}
Status ExchangeSinkOperatorX::sink(RuntimeState* state, vectorized::Block* block, bool eos) {
Review Comment:
warning: function 'sink' has cognitive complexity of 236 (threshold 50) [readability-function-cognitive-complexity]
```cpp
Status ExchangeSinkOperatorX::sink(RuntimeState* state, vectorized::Block* block, bool eos) {
^
```
<details>
<summary>Additional context</summary>
**be/src/pipeline/exec/exchange_sink_operator.cpp:400:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (all_receiver_eof) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:408:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (_part_type == TPartitionType::UNPARTITIONED || local_state.channels.size() == 1) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:412:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (local_state.only_local_exchange) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:413:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (!block->empty()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:428:** +1, nesting level increased to 2
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:432:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._serializer.next_serialized_block(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:432:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.next_serialized_block(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:435:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (serialized) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:437:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
if (!cur_block.empty()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:438:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:438:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:441:** +1, nesting level increased to 4
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:467:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
if (moved) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:469:** +1, nesting level increased to 4
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:477:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::RANDOM) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:480:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (!current_channel->is_receiver_eof()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:482:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (current_channel->is_local()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:484:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:225:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:484:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:226:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:484:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:228:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:484:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:229:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:484:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:229:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:485:** +1, nesting level increased to 3
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:487:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:487:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:489:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:225:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:489:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:226:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:489:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:228:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:489:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:229:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:489:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:229:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:494:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::HASH_PARTITIONED ||
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:499:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:499:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:505:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (_part_type == TPartitionType::HASH_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:506:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:506:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:509:** +1, nesting level increased to 2
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:510:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:510:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:522:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLET_SINK_SHUFFLE_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:528:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:528:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:535:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (input_rows > 0) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:540:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._row_distribution.generate_rows_distribution(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:540:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._row_distribution.generate_rows_distribution(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:546:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
for (int idx = 0; idx < row_ids.size(); ++idx) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:554:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (eos) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:556:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:556:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:560:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(state, local_state.channels, num_channels,
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:560:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(state, local_state.channels, num_channels,
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:570:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLE_SINK_HASH_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:577:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:577:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:581:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:581:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:592:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLE_SINK_RANDOM_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:596:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (!current_channel->is_receiver_eof()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:598:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (current_channel->is_local()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:600:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:225:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:600:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:226:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:600:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:228:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:600:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:229:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:600:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:229:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:601:** +1, nesting level increased to 3
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:603:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:603:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:605:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:225:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:605:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:226:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:605:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:228:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:605:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:229:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:605:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:229:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:610:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (_writer_count < local_state.channels.size()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:611:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (_data_processed >=
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:618:** +1, nesting level increased to 1
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:625:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (eos) {
^
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2418839574
TeamCity be ut coverage result:
Function Coverage: 37.49% (9711/25905)
Line Coverage: 28.75% (80610/280431)
Region Coverage: 28.20% (41717/147944)
Branch Coverage: 24.77% (21214/85640)
Coverage Report: http://coverage.selectdb-in.cc/coverage/a4008dac9d53a65c976079750dc3969150bf3f57_a4008dac9d53a65c976079750dc3969150bf3f57/report/index.html
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1804074075
##########
be/src/vec/sink/vdata_stream_sender.h:
##########
@@ -193,47 +211,23 @@ class Channel {
return false;
}
- Status _wait_last_brpc() {
- SCOPED_TIMER(_parent->brpc_wait_timer());
- if (_send_remote_block_callback == nullptr) {
- return Status::OK();
- }
- _send_remote_block_callback->join();
- if (_send_remote_block_callback->cntl_->Failed()) {
- std::string err = fmt::format(
- "failed to send brpc batch, error={}, error_text={}, client: {}, "
- "latency = {}",
- berror(_send_remote_block_callback->cntl_->ErrorCode()),
- _send_remote_block_callback->cntl_->ErrorText(),
- BackendOptions::get_localhost(),
- _send_remote_block_callback->cntl_->latency_us());
- LOG(WARNING) << err;
- return Status::RpcError(err);
- }
- _receiver_status = Status::create(_send_remote_block_callback->response_->status());
- return _receiver_status;
- }
-
- Status close_internal(Status exec_status);
+ Status _wait_last_brpc();
- Parent* _parent = nullptr;
+ pipeline::ExchangeSinkLocalState* _parent = nullptr;
- const RowDescriptor& _row_desc;
const TUniqueId _fragment_instance_id;
PlanNodeId _dest_node_id;
// the number of RowBatch.data bytes sent successfully
int64_t _num_data_bytes_sent {};
int64_t _packet_seq {};
- bool _need_close;
- bool _closed;
+ bool _need_close {};
Review Comment:
明确写上，是true还是false
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "HappenLee (via GitHub)" <gi...@apache.org>.
HappenLee commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1808508222
##########
be/src/pipeline/exec/exchange_sink_buffer.cpp:
##########
@@ -470,18 +424,13 @@ bool ExchangeSinkBuffer::_is_receiver_eof(InstanceLoId id) {
return _instance_to_receiver_eof[id];
}
-void ExchangeSinkBuffer::_turn_off_channel(InstanceLoId id, bool cleanup) {
+void ExchangeSinkBuffer::_turn_off_channel(InstanceLoId id) {
Review Comment:
if we call the function, we must get the lock of `std::unique_lock<std::mutex> lock(*_instance_to_package_queue_mutex[id]);` so better we can read the important msg from comment or function name
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "mrhhsg (via GitHub)" <gi...@apache.org>.
mrhhsg commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2426796548
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2426885225
TeamCity be ut coverage result:
Function Coverage: 37.47% (9699/25882)
Line Coverage: 28.71% (80501/280378)
Region Coverage: 28.16% (41657/147923)
Branch Coverage: 24.72% (21162/85622)
Coverage Report: http://coverage.selectdb-in.cc/coverage/aa5df39d279774e98568ab011e2548a35ba68f2a_aa5df39d279774e98568ab011e2548a35ba68f2a/report/index.html
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1806016860
##########
be/src/pipeline/exec/exchange_sink_operator.cpp:
##########
@@ -358,7 +373,7 @@ void ExchangeSinkOperatorX::_handle_eof_channel(RuntimeState* state, ChannelPtrT
Status st) {
channel->set_receiver_eof(st);
// Chanel will not send RPC to the downstream when eof, so close chanel by OK status.
- static_cast<void>(channel->close(state, Status::OK()));
+ static_cast<void>(channel->close(state));
}
Status ExchangeSinkOperatorX::sink(RuntimeState* state, vectorized::Block* block, bool eos) {
Review Comment:
warning: function 'sink' has cognitive complexity of 236 (threshold 50) [readability-function-cognitive-complexity]
```cpp
Status ExchangeSinkOperatorX::sink(RuntimeState* state, vectorized::Block* block, bool eos) {
^
```
<details>
<summary>Additional context</summary>
**be/src/pipeline/exec/exchange_sink_operator.cpp:390:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (all_receiver_eof) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:398:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (_part_type == TPartitionType::UNPARTITIONED || local_state.channels.size() == 1) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:402:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (local_state.only_local_exchange) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:403:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (!block->empty()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:418:** +1, nesting level increased to 2
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:422:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._serializer.next_serialized_block(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:422:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.next_serialized_block(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:425:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (serialized) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:427:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
if (!cur_block.empty()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:428:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:428:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:431:** +1, nesting level increased to 4
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:457:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
if (moved) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:459:** +1, nesting level increased to 4
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:467:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::RANDOM) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:470:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (!current_channel->is_receiver_eof()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:472:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (current_channel->is_local()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:474:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:232:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:474:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:233:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:474:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:235:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:474:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:474:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:475:** +1, nesting level increased to 3
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:477:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:477:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:479:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:232:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:479:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:233:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:479:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:235:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:479:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:479:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:484:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::HASH_PARTITIONED ||
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:489:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:489:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:495:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (_part_type == TPartitionType::HASH_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:496:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:496:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:499:** +1, nesting level increased to 2
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:500:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:500:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:512:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLET_SINK_SHUFFLE_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:518:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:518:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:525:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (input_rows > 0) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:530:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._row_distribution.generate_rows_distribution(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:530:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._row_distribution.generate_rows_distribution(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:536:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
for (int idx = 0; idx < row_ids.size(); ++idx) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:544:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (eos) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:546:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:546:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:550:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(state, local_state.channels, num_channels,
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:550:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(state, local_state.channels, num_channels,
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:560:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLE_SINK_HASH_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:567:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:567:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:571:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:571:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:582:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLE_SINK_RANDOM_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:586:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (!current_channel->is_receiver_eof()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:588:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (current_channel->is_local()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:590:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:232:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:590:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:233:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:590:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:235:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:590:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:590:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:591:** +1, nesting level increased to 3
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:593:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:593:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:595:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:232:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:595:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:233:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:595:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:235:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:595:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:595:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:600:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (_writer_count < local_state.channels.size()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:601:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (_data_processed >=
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:608:** +1, nesting level increased to 1
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:615:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (eos) {
^
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2421819595
TeamCity be ut coverage result:
Function Coverage: 37.49% (9711/25903)
Line Coverage: 28.75% (80624/280410)
Region Coverage: 28.20% (41723/147948)
Branch Coverage: 24.76% (21210/85646)
Coverage Report: http://coverage.selectdb-in.cc/coverage/17c2099422e6d59dbac0b260d21100a4035366d0_17c2099422e6d59dbac0b260d21100a4035366d0/report/index.html
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2417045104
clang-tidy review says "All clean, LGTM! :+1:"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1803326549
##########
be/src/vec/sink/vdata_stream_sender.cpp:
##########
@@ -166,131 +166,51 @@ Status PipChannel::send_broadcast_block(std::shared_ptr<BroadcastPBlockHolder>&
return Status::OK();
}
-Status PipChannel::send_current_block(bool eos, Status exec_status) {
- if (Channel<pipeline::ExchangeSinkLocalState>::is_local()) {
- return Channel<pipeline::ExchangeSinkLocalState>::send_local_block(exec_status, eos);
- }
- RETURN_IF_ERROR(send_remote_block(_pblock.release(), eos, exec_status));
- return Status::OK();
-}
-
-template <typename Parent>
-Status Channel<Parent>::send_current_block(bool eos, Status exec_status) {
- // FIXME: Now, local exchange will cause the performance problem is in a multi-threaded scenario
- // so this feature is turned off here by default. We need to re-examine this logic
+Status PipChannel::_send_current_block(bool eos, Status exec_status) {
if (is_local()) {
- return send_local_block(exec_status, eos);
- }
- if (eos) {
- RETURN_IF_ERROR(_serializer.serialize_block(_ch_cur_pb_block, 1));
+ return _send_local_block(exec_status, eos);
}
- RETURN_IF_ERROR(send_remote_block(_ch_cur_pb_block, eos, exec_status));
- ch_roll_pb_block();
+ RETURN_IF_ERROR(send_remote_block(_pblock.release(), eos, exec_status));
return Status::OK();
}
template <typename Parent>
-Status Channel<Parent>::send_local_block(Status exec_status, bool eos) {
+Status Channel<Parent>::_send_local_block(Status exec_status, bool eos) {
SCOPED_TIMER(_parent->local_send_timer());
Block block = _serializer.get_block()->to_block();
_serializer.get_block()->set_mutable_columns(block.clone_empty_columns());
- if (_recvr_is_valid()) {
- if constexpr (!std::is_same_v<pipeline::ResultFileSinkLocalState, Parent>) {
- COUNTER_UPDATE(_parent->local_bytes_send_counter(), block.bytes());
- COUNTER_UPDATE(_parent->local_sent_rows(), block.rows());
- COUNTER_UPDATE(_parent->blocks_sent_counter(), 1);
- }
- _local_recvr->add_block(&block, _parent->sender_id(), true);
- if (eos) {
- _local_recvr->remove_sender(_parent->sender_id(), _be_number, exec_status);
- }
- return Status::OK();
- } else {
- _serializer.reset_block();
- return _receiver_status;
+ if (!block.empty() || eos) {
+ RETURN_IF_ERROR(send_local_block(&block, eos, true));
}
+ _serializer.reset_block();
+ return Status::OK();
}
template <typename Parent>
-Status Channel<Parent>::send_local_block(Block* block, bool can_be_moved) {
+Status Channel<Parent>::send_local_block(Block* block, bool eos, bool can_be_moved) {
SCOPED_TIMER(_parent->local_send_timer());
if (_recvr_is_valid()) {
if constexpr (!std::is_same_v<pipeline::ResultFileSinkLocalState, Parent>) {
COUNTER_UPDATE(_parent->local_bytes_send_counter(), block->bytes());
COUNTER_UPDATE(_parent->local_sent_rows(), block->rows());
COUNTER_UPDATE(_parent->blocks_sent_counter(), 1);
}
- _local_recvr->add_block(block, _parent->sender_id(), can_be_moved);
- return Status::OK();
- } else {
- return _receiver_status;
- }
-}
-
-template <typename Parent>
-Status Channel<Parent>::send_remote_block(PBlock* block, bool eos, Status exec_status) {
- if constexpr (!std::is_same_v<pipeline::ResultFileSinkLocalState, Parent>) {
- COUNTER_UPDATE(_parent->blocks_sent_counter(), 1);
- }
- SCOPED_TIMER(_parent->brpc_send_timer());
- if (_send_remote_block_callback == nullptr) {
- _send_remote_block_callback = DummyBrpcCallback<PTransmitDataResult>::create_shared();
- } else {
- RETURN_IF_ERROR(_wait_last_brpc());
- _send_remote_block_callback->cntl_->Reset();
- }
- VLOG_ROW << "Channel<Parent>::send_batch() instance_id=" << print_id(_fragment_instance_id)
- << " dest_node=" << _dest_node_id << " to_host=" << _brpc_dest_addr.hostname
- << " _packet_seq=" << _packet_seq << " row_desc=" << _row_desc.debug_string();
-
- _brpc_request->set_eos(eos);
- if (!exec_status.ok()) {
- exec_status.to_protobuf(_brpc_request->mutable_exec_status());
- }
- if (block != nullptr && !block->column_metas().empty()) {
- _brpc_request->set_allocated_block(block);
- }
- _brpc_request->set_packet_seq(_packet_seq++);
-
- _send_remote_block_callback->cntl_->set_timeout_ms(_brpc_timeout_ms);
- if (config::exchange_sink_ignore_eovercrowded) {
- _send_remote_block_callback->cntl_->ignore_eovercrowded();
- }
-
- {
- auto send_remote_block_closure =
- AutoReleaseClosure<PTransmitDataParams, DummyBrpcCallback<PTransmitDataResult>>::
- create_unique(_brpc_request, _send_remote_block_callback);
- if (enable_http_send_block(*_brpc_request)) {
- RETURN_IF_ERROR(transmit_block_httpv2(
- _state->exec_env(), std::move(send_remote_block_closure), _brpc_dest_addr));
- } else {
- transmit_blockv2(*_brpc_stub, std::move(send_remote_block_closure));
+ const auto sender_id = _parent->sender_id();
+ if (!block->empty()) [[likely]] {
+ _local_recvr->add_block(block, sender_id, can_be_moved);
}
- }
-
- if (block != nullptr) {
- static_cast<void>(_brpc_request->release_block());
- }
- return Status::OK();
-}
-template <typename Parent>
-Status Channel<Parent>::add_rows(Block* block, const std::vector<uint32_t>& rows, bool eos) {
- if (_fragment_instance_id.lo == -1) {
+ if (eos) [[unlikely]] {
+ _local_recvr->remove_sender(sender_id, _be_number, Status::OK());
+ LOG(INFO) << "****** remove sender: " << sender_id << ", be_number: " << _be_number;
Review Comment:
too many logs
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1803335561
##########
be/src/vec/sink/vdata_stream_sender.cpp:
##########
@@ -166,131 +166,51 @@ Status PipChannel::send_broadcast_block(std::shared_ptr<BroadcastPBlockHolder>&
return Status::OK();
}
-Status PipChannel::send_current_block(bool eos, Status exec_status) {
- if (Channel<pipeline::ExchangeSinkLocalState>::is_local()) {
- return Channel<pipeline::ExchangeSinkLocalState>::send_local_block(exec_status, eos);
- }
- RETURN_IF_ERROR(send_remote_block(_pblock.release(), eos, exec_status));
- return Status::OK();
-}
-
-template <typename Parent>
-Status Channel<Parent>::send_current_block(bool eos, Status exec_status) {
- // FIXME: Now, local exchange will cause the performance problem is in a multi-threaded scenario
- // so this feature is turned off here by default. We need to re-examine this logic
+Status PipChannel::_send_current_block(bool eos, Status exec_status) {
if (is_local()) {
- return send_local_block(exec_status, eos);
- }
- if (eos) {
- RETURN_IF_ERROR(_serializer.serialize_block(_ch_cur_pb_block, 1));
+ return _send_local_block(exec_status, eos);
}
- RETURN_IF_ERROR(send_remote_block(_ch_cur_pb_block, eos, exec_status));
- ch_roll_pb_block();
+ RETURN_IF_ERROR(send_remote_block(_pblock.release(), eos, exec_status));
Review Comment:
这么写有风险，假如有异常，罗指针没释放
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1806215307
##########
be/src/pipeline/exec/exchange_sink_operator.cpp:
##########
@@ -358,7 +384,7 @@ void ExchangeSinkOperatorX::_handle_eof_channel(RuntimeState* state, ChannelPtrT
Status st) {
channel->set_receiver_eof(st);
// Chanel will not send RPC to the downstream when eof, so close chanel by OK status.
- static_cast<void>(channel->close(state, Status::OK()));
+ static_cast<void>(channel->close(state));
}
Status ExchangeSinkOperatorX::sink(RuntimeState* state, vectorized::Block* block, bool eos) {
Review Comment:
warning: function 'sink' has cognitive complexity of 236 (threshold 50) [readability-function-cognitive-complexity]
```cpp
Status ExchangeSinkOperatorX::sink(RuntimeState* state, vectorized::Block* block, bool eos) {
^
```
<details>
<summary>Additional context</summary>
**be/src/pipeline/exec/exchange_sink_operator.cpp:401:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (all_receiver_eof) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:409:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (_part_type == TPartitionType::UNPARTITIONED || local_state.channels.size() == 1) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:413:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (local_state.only_local_exchange) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:414:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (!block->empty()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:429:** +1, nesting level increased to 2
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:433:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._serializer.next_serialized_block(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:433:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.next_serialized_block(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:436:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (serialized) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:438:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
if (!cur_block.empty()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:439:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:439:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:442:** +1, nesting level increased to 4
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:468:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
if (moved) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:470:** +1, nesting level increased to 4
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:478:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::RANDOM) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:481:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (!current_channel->is_receiver_eof()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:483:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (current_channel->is_local()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:485:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:233:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:485:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:234:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:485:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:485:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:485:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:486:** +1, nesting level increased to 3
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:488:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:488:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:490:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:233:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:490:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:234:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:490:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:490:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:490:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:495:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::HASH_PARTITIONED ||
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:500:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:500:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:506:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (_part_type == TPartitionType::HASH_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:507:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:507:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:510:** +1, nesting level increased to 2
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:511:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:511:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:523:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLET_SINK_SHUFFLE_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:529:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:529:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:536:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (input_rows > 0) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:541:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._row_distribution.generate_rows_distribution(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:541:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._row_distribution.generate_rows_distribution(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:547:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
for (int idx = 0; idx < row_ids.size(); ++idx) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:555:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (eos) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:557:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:557:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:561:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(state, local_state.channels, num_channels,
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:561:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(state, local_state.channels, num_channels,
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:571:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLE_SINK_HASH_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:578:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:578:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:582:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:582:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:593:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLE_SINK_RANDOM_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:597:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (!current_channel->is_receiver_eof()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:599:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (current_channel->is_local()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:601:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:233:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:601:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:234:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:601:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:601:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:601:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:602:** +1, nesting level increased to 3
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:604:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:604:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:606:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:233:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:606:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:234:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:606:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:606:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:606:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:611:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (_writer_count < local_state.channels.size()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:612:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (_data_processed >=
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:619:** +1, nesting level increased to 1
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:626:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (eos) {
^
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "HappenLee (via GitHub)" <gi...@apache.org>.
HappenLee commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1808421399
##########
be/src/vec/sink/vdata_stream_sender.h:
##########
@@ -313,12 +182,54 @@ class PipChannel final : public Channel<pipeline::ExchangeSinkLocalState> {
std::shared_ptr<pipeline::Dependency> get_local_channel_dependency();
-private:
+protected:
+ Status _send_local_block(bool eos);
+ Status _send_current_block(bool eos);
+
+ bool _recvr_is_valid() {
Review Comment:
should we return Status in this func and dispose the `_receiver_status` out the function is better ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "mrhhsg (via GitHub)" <gi...@apache.org>.
mrhhsg commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2426231921
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1808051121
##########
be/src/pipeline/exec/exchange_sink_operator.cpp:
##########
@@ -358,7 +383,7 @@ void ExchangeSinkOperatorX::_handle_eof_channel(RuntimeState* state, ChannelPtrT
Status st) {
channel->set_receiver_eof(st);
// Chanel will not send RPC to the downstream when eof, so close chanel by OK status.
- static_cast<void>(channel->close(state, Status::OK()));
+ static_cast<void>(channel->close(state));
}
Status ExchangeSinkOperatorX::sink(RuntimeState* state, vectorized::Block* block, bool eos) {
Review Comment:
warning: function 'sink' has cognitive complexity of 236 (threshold 50) [readability-function-cognitive-complexity]
```cpp
Status ExchangeSinkOperatorX::sink(RuntimeState* state, vectorized::Block* block, bool eos) {
^
```
<details>
<summary>Additional context</summary>
**be/src/pipeline/exec/exchange_sink_operator.cpp:400:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (all_receiver_eof) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:408:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (_part_type == TPartitionType::UNPARTITIONED || local_state.channels.size() == 1) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:412:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (local_state.only_local_exchange) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:413:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (!block->empty()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:428:** +1, nesting level increased to 2
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:432:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._serializer.next_serialized_block(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:432:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.next_serialized_block(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:435:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (serialized) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:437:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
if (!cur_block.empty()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:438:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:438:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:441:** +1, nesting level increased to 4
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:467:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
if (moved) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:469:** +1, nesting level increased to 4
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:477:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::RANDOM) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:480:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (!current_channel->is_receiver_eof()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:482:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (current_channel->is_local()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:484:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:233:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:484:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:234:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:484:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:484:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:484:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:485:** +1, nesting level increased to 3
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:487:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:487:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:489:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:233:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:489:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:234:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:489:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:489:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:489:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:494:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::HASH_PARTITIONED ||
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:499:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:499:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:505:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (_part_type == TPartitionType::HASH_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:506:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:506:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:509:** +1, nesting level increased to 2
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:510:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:510:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:522:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLET_SINK_SHUFFLE_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:528:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:528:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:535:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (input_rows > 0) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:540:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._row_distribution.generate_rows_distribution(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:540:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._row_distribution.generate_rows_distribution(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:546:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
for (int idx = 0; idx < row_ids.size(); ++idx) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:554:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (eos) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:556:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:556:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:560:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(state, local_state.channels, num_channels,
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:560:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(state, local_state.channels, num_channels,
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:570:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLE_SINK_HASH_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:577:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:577:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:581:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:581:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:592:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLE_SINK_RANDOM_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:596:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (!current_channel->is_receiver_eof()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:598:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (current_channel->is_local()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:600:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:233:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:600:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:234:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:600:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:600:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:600:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:601:** +1, nesting level increased to 3
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:603:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:603:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:605:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:233:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:605:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:234:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:605:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:605:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:605:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:237:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:610:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (_writer_count < local_state.channels.size()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:611:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (_data_processed >=
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:618:** +1, nesting level increased to 1
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:625:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (eos) {
^
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2418359094
clang-tidy review says "All clean, LGTM! :+1:"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1803321780
##########
be/src/vec/sink/vdata_stream_sender.cpp:
##########
@@ -166,131 +166,51 @@ Status PipChannel::send_broadcast_block(std::shared_ptr<BroadcastPBlockHolder>&
return Status::OK();
}
-Status PipChannel::send_current_block(bool eos, Status exec_status) {
- if (Channel<pipeline::ExchangeSinkLocalState>::is_local()) {
- return Channel<pipeline::ExchangeSinkLocalState>::send_local_block(exec_status, eos);
- }
- RETURN_IF_ERROR(send_remote_block(_pblock.release(), eos, exec_status));
- return Status::OK();
-}
-
-template <typename Parent>
-Status Channel<Parent>::send_current_block(bool eos, Status exec_status) {
- // FIXME: Now, local exchange will cause the performance problem is in a multi-threaded scenario
- // so this feature is turned off here by default. We need to re-examine this logic
+Status PipChannel::_send_current_block(bool eos, Status exec_status) {
if (is_local()) {
- return send_local_block(exec_status, eos);
- }
- if (eos) {
- RETURN_IF_ERROR(_serializer.serialize_block(_ch_cur_pb_block, 1));
+ return _send_local_block(exec_status, eos);
}
- RETURN_IF_ERROR(send_remote_block(_ch_cur_pb_block, eos, exec_status));
- ch_roll_pb_block();
+ RETURN_IF_ERROR(send_remote_block(_pblock.release(), eos, exec_status));
return Status::OK();
}
template <typename Parent>
-Status Channel<Parent>::send_local_block(Status exec_status, bool eos) {
+Status Channel<Parent>::_send_local_block(Status exec_status, bool eos) {
SCOPED_TIMER(_parent->local_send_timer());
Block block = _serializer.get_block()->to_block();
_serializer.get_block()->set_mutable_columns(block.clone_empty_columns());
- if (_recvr_is_valid()) {
- if constexpr (!std::is_same_v<pipeline::ResultFileSinkLocalState, Parent>) {
- COUNTER_UPDATE(_parent->local_bytes_send_counter(), block.bytes());
- COUNTER_UPDATE(_parent->local_sent_rows(), block.rows());
- COUNTER_UPDATE(_parent->blocks_sent_counter(), 1);
- }
- _local_recvr->add_block(&block, _parent->sender_id(), true);
- if (eos) {
- _local_recvr->remove_sender(_parent->sender_id(), _be_number, exec_status);
- }
- return Status::OK();
- } else {
- _serializer.reset_block();
- return _receiver_status;
+ if (!block.empty() || eos) {
+ RETURN_IF_ERROR(send_local_block(&block, eos, true));
}
+ _serializer.reset_block();
Review Comment:
如果这里调用reset，那么181 行的意义是什么？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "mrhhsg (via GitHub)" <gi...@apache.org>.
mrhhsg commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1804928001
##########
be/src/vec/sink/vdata_stream_sender.cpp:
##########
@@ -166,210 +151,76 @@ Status PipChannel::send_broadcast_block(std::shared_ptr<BroadcastPBlockHolder>&
return Status::OK();
}
-Status PipChannel::send_current_block(bool eos, Status exec_status) {
- if (Channel<pipeline::ExchangeSinkLocalState>::is_local()) {
- return Channel<pipeline::ExchangeSinkLocalState>::send_local_block(exec_status, eos);
- }
- RETURN_IF_ERROR(send_remote_block(_pblock.release(), eos, exec_status));
- return Status::OK();
-}
-
-template <typename Parent>
-Status Channel<Parent>::send_current_block(bool eos, Status exec_status) {
- // FIXME: Now, local exchange will cause the performance problem is in a multi-threaded scenario
- // so this feature is turned off here by default. We need to re-examine this logic
+Status Channel::_send_current_block(bool eos, Status exec_status) {
if (is_local()) {
- return send_local_block(exec_status, eos);
- }
- if (eos) {
- RETURN_IF_ERROR(_serializer.serialize_block(_ch_cur_pb_block, 1));
+ return _send_local_block(exec_status, eos);
}
- RETURN_IF_ERROR(send_remote_block(_ch_cur_pb_block, eos, exec_status));
- ch_roll_pb_block();
- return Status::OK();
+ return send_remote_block(std::move(_pblock), eos, exec_status);
}
-template <typename Parent>
-Status Channel<Parent>::send_local_block(Status exec_status, bool eos) {
+Status Channel::_send_local_block(Status exec_status, bool eos) {
SCOPED_TIMER(_parent->local_send_timer());
Block block = _serializer.get_block()->to_block();
_serializer.get_block()->set_mutable_columns(block.clone_empty_columns());
Review Comment:
prepare for next block to merge with
##########
be/src/vec/sink/vdata_stream_sender.cpp:
##########
@@ -147,13 +132,13 @@ Status PipChannel::send_remote_block(PBlock* block, bool eos, Status exec_status
}
}
if (eos || block->column_metas_size()) {
- RETURN_IF_ERROR(_buffer->add_block({this, std::move(pblock_ptr), eos, exec_status}));
+ RETURN_IF_ERROR(_buffer->add_block({this, std::move(block), eos, exec_status}));
}
return Status::OK();
}
-Status PipChannel::send_broadcast_block(std::shared_ptr<BroadcastPBlockHolder>& block, bool eos) {
- COUNTER_UPDATE(Channel<pipeline::ExchangeSinkLocalState>::_parent->blocks_sent_counter(), 1);
+Status Channel::send_broadcast_block(std::shared_ptr<BroadcastPBlockHolder>& block, bool eos) {
+ COUNTER_UPDATE(_parent->blocks_sent_counter(), 1);
if (eos) {
if (_eos_send) {
Review Comment:
Here use `_eos_send` to avoid sending eos repeatedly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "Gabriel39 (via GitHub)" <gi...@apache.org>.
Gabriel39 commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1805790254
##########
be/src/vec/sink/vdata_stream_sender.h:
##########
@@ -177,151 +140,40 @@ class Channel {
bool is_local() const { return _is_local; }
- virtual void ch_roll_pb_block();
-
bool is_receiver_eof() const { return _receiver_status.is<ErrorCode::END_OF_FILE>(); }
void set_receiver_eof(Status st) { _receiver_status = st; }
-protected:
- bool _recvr_is_valid() {
- if (_local_recvr && !_local_recvr->is_closed()) {
- return true;
- }
- _receiver_status = Status::EndOfFile(
- "local data stream receiver closed"); // local data stream receiver closed
- return false;
- }
-
- Status _wait_last_brpc() {
- SCOPED_TIMER(_parent->brpc_wait_timer());
- if (_send_remote_block_callback == nullptr) {
- return Status::OK();
- }
- _send_remote_block_callback->join();
- if (_send_remote_block_callback->cntl_->Failed()) {
- std::string err = fmt::format(
- "failed to send brpc batch, error={}, error_text={}, client: {}, "
- "latency = {}",
- berror(_send_remote_block_callback->cntl_->ErrorCode()),
- _send_remote_block_callback->cntl_->ErrorText(),
- BackendOptions::get_localhost(),
- _send_remote_block_callback->cntl_->latency_us());
- LOG(WARNING) << err;
- return Status::RpcError(err);
- }
- _receiver_status = Status::create(_send_remote_block_callback->response_->status());
- return _receiver_status;
- }
-
- Status close_internal(Status exec_status);
-
- Parent* _parent = nullptr;
-
- const RowDescriptor& _row_desc;
- const TUniqueId _fragment_instance_id;
- PlanNodeId _dest_node_id;
-
- // the number of RowBatch.data bytes sent successfully
- int64_t _num_data_bytes_sent {};
- int64_t _packet_seq {};
-
- bool _need_close;
- bool _closed;
- int _be_number;
-
- TNetworkAddress _brpc_dest_addr;
-
- PUniqueId _finst_id;
- PUniqueId _query_id;
- PBlock _pb_block;
- std::shared_ptr<PTransmitDataParams> _brpc_request = nullptr;
- std::shared_ptr<PBackendService_Stub> _brpc_stub = nullptr;
- std::shared_ptr<DummyBrpcCallback<PTransmitDataResult>> _send_remote_block_callback;
- Status _receiver_status;
- int32_t _brpc_timeout_ms = 500;
- RuntimeState* _state = nullptr;
-
- bool _is_local;
- std::shared_ptr<VDataStreamRecvr> _local_recvr;
- // serialized blocks for broadcasting; we need two so we can write
- // one while the other one is still being sent.
- // Which is for same reason as `_cur_pb_block`, `_pb_block1` and `_pb_block2`
- // in VDataStreamSender.
- PBlock* _ch_cur_pb_block = nullptr;
- PBlock _ch_pb_block1;
- PBlock _ch_pb_block2;
-
- BlockSerializer<Parent> _serializer;
-};
-
-#define HANDLE_CHANNEL_STATUS(state, channel, status) \
- do { \
- if (status.is<ErrorCode::END_OF_FILE>()) { \
- _handle_eof_channel(state, channel, status); \
- } else { \
- RETURN_IF_ERROR(status); \
- } \
- } while (0)
-
-class PipChannel final : public Channel<pipeline::ExchangeSinkLocalState> {
-public:
- PipChannel(pipeline::ExchangeSinkLocalState* parent, const RowDescriptor& row_desc,
- const TNetworkAddress& brpc_dest, const TUniqueId& fragment_instance_id,
- PlanNodeId dest_node_id)
- : Channel<pipeline::ExchangeSinkLocalState>(parent, row_desc, brpc_dest,
- fragment_instance_id, dest_node_id) {
- ch_roll_pb_block();
- }
-
- ~PipChannel() override { delete Channel<pipeline::ExchangeSinkLocalState>::_ch_cur_pb_block; }
-
int64_t mem_usage() const;
- void ch_roll_pb_block() override {
- // We have two choices here.
- // 1. Use a PBlock pool and fetch an available PBlock if we need one. In this way, we can
- // reuse the memory, but we have to use a lock to synchronize.
- // 2. Create a new PBlock every time. In this way we don't need a lock but have to allocate
- // new memory.
- // Now we use the second way.
- Channel<pipeline::ExchangeSinkLocalState>::_ch_cur_pb_block = new PBlock();
- }
-
// Asynchronously sends a block
// Returns the status of the most recently finished transmit_data
// rpc (or OK if there wasn't one that hasn't been reported yet).
// if batch is nullptr, send the eof packet
- Status send_remote_block(PBlock* block, bool eos = false,
- Status exec_status = Status::OK()) override;
+ Status send_remote_block(std::unique_ptr<PBlock>&& block, bool eos = false,
+ Status exec_status = Status::OK());
- Status send_broadcast_block(std::shared_ptr<BroadcastPBlockHolder>& block,
- bool eos = false) override;
+ Status send_broadcast_block(std::shared_ptr<BroadcastPBlockHolder>& block, bool eos = false);
- Status add_rows(Block* block, const std::vector<uint32_t>& rows, bool eos) override {
- if (Channel<pipeline::ExchangeSinkLocalState>::_fragment_instance_id.lo == -1) {
+ Status add_rows(Block* block, const std::vector<uint32_t>& rows, bool eos) {
+ if (_fragment_instance_id.lo == -1) {
return Status::OK();
}
bool serialized = false;
_pblock = std::make_unique<PBlock>();
Review Comment:
DCHECK(_pblock == nullptr)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "mrhhsg (via GitHub)" <gi...@apache.org>.
mrhhsg commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2418355319
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1804070935
##########
be/src/pipeline/exec/exchange_sink_operator.h:
##########
@@ -112,8 +112,8 @@ class ExchangeSinkLocalState final : public PipelineXSinkLocalState<> {
return Status::OK();
}
Status _send_new_partition_batch();
- std::vector<vectorized::PipChannel*> channels;
- std::vector<std::shared_ptr<vectorized::PipChannel>> channel_shared_ptrs;
+ std::vector<vectorized::Channel*> channels;
Review Comment:
为什么保留2个？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "mrhhsg (via GitHub)" <gi...@apache.org>.
mrhhsg commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2421993215
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2422095936
TeamCity be ut coverage result:
Function Coverage: 37.49% (9710/25903)
Line Coverage: 28.75% (80610/280418)
Region Coverage: 28.20% (41723/147951)
Branch Coverage: 24.76% (21206/85648)
Coverage Report: http://coverage.selectdb-in.cc/coverage/a84a68cad48b603e9dc0cf40c19558edfb31297a_a84a68cad48b603e9dc0cf40c19558edfb31297a/report/index.html
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1808489163
##########
be/src/pipeline/exec/exchange_sink_operator.cpp:
##########
@@ -358,7 +383,7 @@ void ExchangeSinkOperatorX::_handle_eof_channel(RuntimeState* state, ChannelPtrT
Status st) {
channel->set_receiver_eof(st);
// Chanel will not send RPC to the downstream when eof, so close chanel by OK status.
- static_cast<void>(channel->close(state, Status::OK()));
+ static_cast<void>(channel->close(state));
}
Status ExchangeSinkOperatorX::sink(RuntimeState* state, vectorized::Block* block, bool eos) {
Review Comment:
warning: function 'sink' has cognitive complexity of 236 (threshold 50) [readability-function-cognitive-complexity]
```cpp
Status ExchangeSinkOperatorX::sink(RuntimeState* state, vectorized::Block* block, bool eos) {
^
```
<details>
<summary>Additional context</summary>
**be/src/pipeline/exec/exchange_sink_operator.cpp:400:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (all_receiver_eof) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:408:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (_part_type == TPartitionType::UNPARTITIONED || local_state.channels.size() == 1) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:412:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (local_state.only_local_exchange) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:413:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (!block->empty()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:428:** +1, nesting level increased to 2
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:432:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._serializer.next_serialized_block(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:432:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.next_serialized_block(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:435:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (serialized) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:437:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
if (!cur_block.empty()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:438:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:438:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:441:** +1, nesting level increased to 4
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:467:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
if (moved) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:469:** +1, nesting level increased to 4
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:477:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::RANDOM) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:480:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (!current_channel->is_receiver_eof()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:482:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (current_channel->is_local()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:484:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:224:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:484:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:225:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:484:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:227:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:484:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:228:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:484:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:228:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:485:** +1, nesting level increased to 3
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:487:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:487:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:489:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:224:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:489:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:225:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:489:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:227:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:489:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:228:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:489:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:228:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:494:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::HASH_PARTITIONED ||
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:499:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:499:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:505:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (_part_type == TPartitionType::HASH_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:506:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:506:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:509:** +1, nesting level increased to 2
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:510:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:510:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:522:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLET_SINK_SHUFFLE_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:528:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:528:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:535:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (input_rows > 0) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:540:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._row_distribution.generate_rows_distribution(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:540:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._row_distribution.generate_rows_distribution(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:546:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
for (int idx = 0; idx < row_ids.size(); ++idx) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:554:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (eos) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:556:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:556:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:560:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(state, local_state.channels, num_channels,
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:560:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(state, local_state.channels, num_channels,
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:570:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLE_SINK_HASH_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:577:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:577:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:581:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:581:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:592:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLE_SINK_RANDOM_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:596:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (!current_channel->is_receiver_eof()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:598:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (current_channel->is_local()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:600:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:224:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:600:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:225:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:600:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:227:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:600:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:228:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:600:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:228:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:601:** +1, nesting level increased to 3
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:603:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:603:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:605:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:224:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:605:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:225:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:605:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:227:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:605:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:228:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:605:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:228:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:610:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (_writer_count < local_state.channels.size()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:611:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (_data_processed >=
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:618:** +1, nesting level increased to 1
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:625:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (eos) {
^
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2428702406
TeamCity be ut coverage result:
Function Coverage: 37.46% (9706/25908)
Line Coverage: 28.70% (80493/280452)
Region Coverage: 28.15% (41642/147954)
Branch Coverage: 24.71% (21157/85632)
Coverage Report: http://coverage.selectdb-in.cc/coverage/74f65c342ba57db8803ceb5a58a8f3929f056b22_74f65c342ba57db8803ceb5a58a8f3929f056b22/report/index.html
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2418391676
TeamCity be ut coverage result:
Function Coverage: 37.47% (9709/25910)
Line Coverage: 28.74% (80611/280482)
Region Coverage: 28.19% (41708/147968)
Branch Coverage: 24.76% (21204/85654)
Coverage Report: http://coverage.selectdb-in.cc/coverage/0e43eef28e0e5c5fdd297a0293ee8a43472f29ff_0e43eef28e0e5c5fdd297a0293ee8a43472f29ff/report/index.html
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1804264275
##########
be/src/vec/sink/vdata_stream_sender.h:
##########
@@ -337,12 +189,53 @@ class PipChannel final : public Channel<pipeline::ExchangeSinkLocalState> {
std::shared_ptr<pipeline::Dependency> get_local_channel_dependency();
-private:
+protected:
+ Status _send_local_block(Status exec_status, bool eos = false);
+ Status _send_current_block(bool eos, Status exec_status);
+
+ bool _recvr_is_valid() {
+ if (_local_recvr && !_local_recvr->is_closed()) {
+ return true;
Review Comment:
warning: redundant boolean literal in conditional return statement [readability-simplify-boolean-expr]
```cpp
return true;
^
```
##########
be/src/pipeline/exec/exchange_sink_operator.cpp:
##########
@@ -358,7 +360,7 @@ void ExchangeSinkOperatorX::_handle_eof_channel(RuntimeState* state, ChannelPtrT
Status st) {
channel->set_receiver_eof(st);
// Chanel will not send RPC to the downstream when eof, so close chanel by OK status.
- static_cast<void>(channel->close(state, Status::OK()));
+ static_cast<void>(channel->close(state));
}
Status ExchangeSinkOperatorX::sink(RuntimeState* state, vectorized::Block* block, bool eos) {
Review Comment:
warning: function 'sink' has cognitive complexity of 236 (threshold 50) [readability-function-cognitive-complexity]
```cpp
Status ExchangeSinkOperatorX::sink(RuntimeState* state, vectorized::Block* block, bool eos) {
^
```
<details>
<summary>Additional context</summary>
**be/src/pipeline/exec/exchange_sink_operator.cpp:377:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (all_receiver_eof) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:385:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (_part_type == TPartitionType::UNPARTITIONED || local_state.channels.size() == 1) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:389:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (local_state.only_local_exchange) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:390:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (!block->empty()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:405:** +1, nesting level increased to 2
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:409:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._serializer.next_serialized_block(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:409:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.next_serialized_block(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:412:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (serialized) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:414:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
if (!cur_block.empty()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:415:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:415:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:418:** +1, nesting level increased to 4
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:444:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
if (moved) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:446:** +1, nesting level increased to 4
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:454:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::RANDOM) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:457:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (!current_channel->is_receiver_eof()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:459:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (current_channel->is_local()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:461:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:231:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:461:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:232:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:461:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:234:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:461:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:235:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:461:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:235:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:462:** +1, nesting level increased to 3
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:464:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:464:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:466:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:231:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:466:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:232:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:466:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:234:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:466:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:235:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:466:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:235:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:471:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::HASH_PARTITIONED ||
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:476:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:476:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:482:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (_part_type == TPartitionType::HASH_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:483:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:483:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:486:** +1, nesting level increased to 2
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:487:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:487:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:499:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLET_SINK_SHUFFLE_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:505:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:505:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:512:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (input_rows > 0) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:517:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._row_distribution.generate_rows_distribution(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:517:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._row_distribution.generate_rows_distribution(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:523:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
for (int idx = 0; idx < row_ids.size(); ++idx) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:531:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (eos) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:533:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:533:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:537:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(state, local_state.channels, num_channels,
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:537:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(state, local_state.channels, num_channels,
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:547:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLE_SINK_HASH_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:554:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:554:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:558:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:558:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:569:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLE_SINK_RANDOM_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:573:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (!current_channel->is_receiver_eof()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:575:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (current_channel->is_local()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:577:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:231:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:577:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:232:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:577:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:234:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:577:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:235:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:577:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:235:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:578:** +1, nesting level increased to 3
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:580:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:580:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:582:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:231:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:582:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:232:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:582:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:234:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:582:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:235:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:582:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:235:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:587:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (_writer_count < local_state.channels.size()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:588:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (_data_processed >=
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:595:** +1, nesting level increased to 1
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:602:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (eos) {
^
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "mrhhsg (via GitHub)" <gi...@apache.org>.
mrhhsg commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2418779522
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1804462686
##########
be/src/pipeline/exec/exchange_sink_operator.cpp:
##########
@@ -358,7 +360,7 @@ void ExchangeSinkOperatorX::_handle_eof_channel(RuntimeState* state, ChannelPtrT
Status st) {
channel->set_receiver_eof(st);
// Chanel will not send RPC to the downstream when eof, so close chanel by OK status.
- static_cast<void>(channel->close(state, Status::OK()));
+ static_cast<void>(channel->close(state));
}
Status ExchangeSinkOperatorX::sink(RuntimeState* state, vectorized::Block* block, bool eos) {
Review Comment:
warning: function 'sink' has cognitive complexity of 236 (threshold 50) [readability-function-cognitive-complexity]
```cpp
Status ExchangeSinkOperatorX::sink(RuntimeState* state, vectorized::Block* block, bool eos) {
^
```
<details>
<summary>Additional context</summary>
**be/src/pipeline/exec/exchange_sink_operator.cpp:377:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (all_receiver_eof) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:385:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (_part_type == TPartitionType::UNPARTITIONED || local_state.channels.size() == 1) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:389:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (local_state.only_local_exchange) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:390:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (!block->empty()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:405:** +1, nesting level increased to 2
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:409:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._serializer.next_serialized_block(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:409:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.next_serialized_block(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:412:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (serialized) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:414:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
if (!cur_block.empty()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:415:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:415:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:418:** +1, nesting level increased to 4
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:444:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
if (moved) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:446:** +1, nesting level increased to 4
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:454:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::RANDOM) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:457:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (!current_channel->is_receiver_eof()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:459:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (current_channel->is_local()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:461:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:232:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:461:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:233:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:461:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:235:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:461:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:461:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:462:** +1, nesting level increased to 3
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:464:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:464:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:466:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:232:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:466:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:233:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:466:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:235:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:466:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:466:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:471:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::HASH_PARTITIONED ||
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:476:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:476:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:482:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (_part_type == TPartitionType::HASH_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:483:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:483:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:486:** +1, nesting level increased to 2
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:487:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:487:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(channel_add_rows(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:499:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLET_SINK_SHUFFLE_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:505:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:505:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:512:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (input_rows > 0) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:517:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._row_distribution.generate_rows_distribution(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:517:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._row_distribution.generate_rows_distribution(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:523:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
for (int idx = 0; idx < row_ids.size(); ++idx) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:531:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (eos) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:533:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:533:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._send_new_partition_batch());
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:537:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(state, local_state.channels, num_channels,
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:537:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(state, local_state.channels, num_channels,
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:547:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLE_SINK_HASH_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:554:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:554:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._partitioner->do_partitioning(state, block));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:558:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:558:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(channel_add_rows_with_idx(
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:569:** +1, nesting level increased to 1
```cpp
} else if (_part_type == TPartitionType::TABLE_SINK_RANDOM_PARTITIONED) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:573:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (!current_channel->is_receiver_eof()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:575:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (current_channel->is_local()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:577:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:232:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:577:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:233:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:577:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:235:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:577:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:577:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:578:** +1, nesting level increased to 3
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:580:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:580:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(local_state._serializer.serialize_block(block, pblock.get()));
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:582:** +4, including nesting penalty of 3, nesting level increased to 4
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:232:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:582:** +5, including nesting penalty of 4, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:233:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
if (status.is<ErrorCode::END_OF_FILE>()) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:582:** +1, nesting level increased to 5
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:235:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
} else { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:582:** +6, including nesting penalty of 5, nesting level increased to 6
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:629:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:582:** +7, including nesting penalty of 6, nesting level increased to 7
```cpp
HANDLE_CHANNEL_STATUS(state, current_channel, status);
^
```
**be/src/vec/sink/vdata_stream_sender.h:236:** expanded from macro 'HANDLE_CHANNEL_STATUS'
```cpp
RETURN_IF_ERROR(status); \
^
```
**be/src/common/status.h:631:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:587:** +2, including nesting penalty of 1, nesting level increased to 2
```cpp
if (_writer_count < local_state.channels.size()) {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:588:** +3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (_data_processed >=
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:595:** +1, nesting level increased to 1
```cpp
} else {
^
```
**be/src/pipeline/exec/exchange_sink_operator.cpp:602:** +1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (eos) {
^
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1804653078
##########
be/src/vec/sink/vdata_stream_sender.cpp:
##########
@@ -166,210 +151,76 @@ Status PipChannel::send_broadcast_block(std::shared_ptr<BroadcastPBlockHolder>&
return Status::OK();
}
-Status PipChannel::send_current_block(bool eos, Status exec_status) {
- if (Channel<pipeline::ExchangeSinkLocalState>::is_local()) {
- return Channel<pipeline::ExchangeSinkLocalState>::send_local_block(exec_status, eos);
- }
- RETURN_IF_ERROR(send_remote_block(_pblock.release(), eos, exec_status));
- return Status::OK();
-}
-
-template <typename Parent>
-Status Channel<Parent>::send_current_block(bool eos, Status exec_status) {
- // FIXME: Now, local exchange will cause the performance problem is in a multi-threaded scenario
- // so this feature is turned off here by default. We need to re-examine this logic
+Status Channel::_send_current_block(bool eos, Status exec_status) {
if (is_local()) {
- return send_local_block(exec_status, eos);
- }
- if (eos) {
- RETURN_IF_ERROR(_serializer.serialize_block(_ch_cur_pb_block, 1));
+ return _send_local_block(exec_status, eos);
}
- RETURN_IF_ERROR(send_remote_block(_ch_cur_pb_block, eos, exec_status));
- ch_roll_pb_block();
- return Status::OK();
+ return send_remote_block(std::move(_pblock), eos, exec_status);
}
-template <typename Parent>
-Status Channel<Parent>::send_local_block(Status exec_status, bool eos) {
+Status Channel::_send_local_block(Status exec_status, bool eos) {
SCOPED_TIMER(_parent->local_send_timer());
Block block = _serializer.get_block()->to_block();
_serializer.get_block()->set_mutable_columns(block.clone_empty_columns());
Review Comment:
为什么要clone 一把
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1804664930
##########
be/src/vec/sink/vdata_stream_sender.cpp:
##########
@@ -147,13 +132,13 @@ Status PipChannel::send_remote_block(PBlock* block, bool eos, Status exec_status
}
}
if (eos || block->column_metas_size()) {
- RETURN_IF_ERROR(_buffer->add_block({this, std::move(pblock_ptr), eos, exec_status}));
+ RETURN_IF_ERROR(_buffer->add_block({this, std::move(block), eos, exec_status}));
}
return Status::OK();
}
-Status PipChannel::send_broadcast_block(std::shared_ptr<BroadcastPBlockHolder>& block, bool eos) {
- COUNTER_UPDATE(Channel<pipeline::ExchangeSinkLocalState>::_parent->blocks_sent_counter(), 1);
+Status Channel::send_broadcast_block(std::shared_ptr<BroadcastPBlockHolder>& block, bool eos) {
+ COUNTER_UPDATE(_parent->blocks_sent_counter(), 1);
if (eos) {
if (_eos_send) {
Review Comment:
eos_send 是什么意思？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1803299619
##########
be/src/pipeline/exec/exchange_sink_buffer.cpp:
##########
@@ -432,13 +426,13 @@ void ExchangeSinkBuffer::_failed(InstanceLoId id, const std::string& err) {
_is_finishing = true;
_context->cancel(Status::Cancelled(err));
std::unique_lock<std::mutex> lock(*_instance_to_package_queue_mutex[id]);
- _turn_off_channel(id, true);
+ _turn_off_channel(id);
Review Comment:
do not need do anything because you already call context->cancel
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #41968:
URL: https://github.com/apache/doris/pull/41968#discussion_r1803311481
##########
be/src/vec/sink/vdata_stream_sender.cpp:
##########
@@ -166,131 +166,51 @@ Status PipChannel::send_broadcast_block(std::shared_ptr<BroadcastPBlockHolder>&
return Status::OK();
}
-Status PipChannel::send_current_block(bool eos, Status exec_status) {
- if (Channel<pipeline::ExchangeSinkLocalState>::is_local()) {
- return Channel<pipeline::ExchangeSinkLocalState>::send_local_block(exec_status, eos);
- }
- RETURN_IF_ERROR(send_remote_block(_pblock.release(), eos, exec_status));
- return Status::OK();
-}
-
-template <typename Parent>
-Status Channel<Parent>::send_current_block(bool eos, Status exec_status) {
- // FIXME: Now, local exchange will cause the performance problem is in a multi-threaded scenario
- // so this feature is turned off here by default. We need to re-examine this logic
+Status PipChannel::_send_current_block(bool eos, Status exec_status) {
if (is_local()) {
- return send_local_block(exec_status, eos);
- }
- if (eos) {
- RETURN_IF_ERROR(_serializer.serialize_block(_ch_cur_pb_block, 1));
+ return _send_local_block(exec_status, eos);
}
- RETURN_IF_ERROR(send_remote_block(_ch_cur_pb_block, eos, exec_status));
- ch_roll_pb_block();
+ RETURN_IF_ERROR(send_remote_block(_pblock.release(), eos, exec_status));
return Status::OK();
Review Comment:
useless code
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](exchange) Optimize the logic related to sending and closing [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #41968:
URL: https://github.com/apache/doris/pull/41968#issuecomment-2422001718
PR approved by at least one committer and no changes requested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org