This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new c64618c8131 branch-4.1: [Exec](shuffle) random shuffle firstly use
local channel to improve performance #59431 (#66103)
c64618c8131 is described below
commit c64618c81317d41dabf930846f393ddf06e1ca2b
Author: HappenLee <[email protected]>
AuthorDate: Wed Jul 29 13:53:56 2026 +0800
branch-4.1: [Exec](shuffle) random shuffle firstly use local channel to
improve performance #59431 (#66103)
### What problem does this PR solve?
Issue Number: close #xxx
Related PR: #59431
Problem Summary: Cherry-pick #59431 to branch-4.1. For RANDOM shuffle,
the exchange sink used to send blocks to channels in round-robin order
one by one. This change collects local channel ids during init and
prefers sending to local channels first, reducing network overhead and
improving performance.
### Release note
None
### Check List (For Author)
- Test: No need to test (pure cherry-pick from master with minor
conflict resolution on namespace)
- Behavior changed: No
- Does this need documentation: No
---
be/src/exec/operator/exchange_sink_operator.cpp | 12 ++++++++++++
be/src/exec/operator/exchange_sink_operator.h | 1 +
2 files changed, 13 insertions(+)
diff --git a/be/src/exec/operator/exchange_sink_operator.cpp
b/be/src/exec/operator/exchange_sink_operator.cpp
index 6a64dc212a5..4d767611e1c 100644
--- a/be/src/exec/operator/exchange_sink_operator.cpp
+++ b/be/src/exec/operator/exchange_sink_operator.cpp
@@ -109,6 +109,7 @@ Status ExchangeSinkLocalState::init(RuntimeState* state,
LocalSinkStateInfo& inf
size_t local_size = 0;
for (int i = 0; i < channels.size(); ++i) {
if (channels[i]->is_local()) {
+ local_channel_ids.push_back(i);
local_size++;
_last_local_channel_idx = i;
}
@@ -509,6 +510,17 @@ Status ExchangeSinkOperatorX::sink_impl(RuntimeState*
state, Block* block, bool
}
}
} else if (_part_type == TPartitionType::RANDOM) {
+ if (!local_state.local_channel_ids.empty()) {
+ const auto& ids = local_state.local_channel_ids;
+ // Find the first channel ID >= current_channel_idx
+ auto it = std::lower_bound(ids.begin(), ids.end(),
local_state.current_channel_idx);
+ if (it != ids.end()) {
+ local_state.current_channel_idx = *it;
+ } else {
+ // All IDs are < current_channel_idx; wrap around to the first
one
+ local_state.current_channel_idx = ids[0];
+ }
+ }
RETURN_IF_ERROR(send_to_current_channel());
local_state.current_channel_idx =
(local_state.current_channel_idx + 1) %
local_state.channels.size();
diff --git a/be/src/exec/operator/exchange_sink_operator.h
b/be/src/exec/operator/exchange_sink_operator.h
index c850c13cac4..09f4cf8b047 100644
--- a/be/src/exec/operator/exchange_sink_operator.h
+++ b/be/src/exec/operator/exchange_sink_operator.h
@@ -103,6 +103,7 @@ public:
return _distribute_rows_into_channels_timer;
}
std::vector<std::shared_ptr<Channel>> channels;
+ std::vector<int> local_channel_ids;
int current_channel_idx {0}; // index of current channel to send to if
_random == true
bool _only_local_exchange {false};
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
