This is an automated email from the ASF dual-hosted git repository.
gabriellee pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-3.0 by this push:
new abff86aab57 branch-3.0: [opt](exec) Use PASSTHROUGH to improve the
concurrency of the ADAPTIVE_PASSTHROUGH SINK. #44925 (#44964)
abff86aab57 is described below
commit abff86aab5793344ed17fdbea884b6a6f776a704
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Wed Dec 4 17:06:31 2024 +0800
branch-3.0: [opt](exec) Use PASSTHROUGH to improve the concurrency of the
ADAPTIVE_PASSTHROUGH SINK. #44925 (#44964)
Cherry-picked from #44925
Co-authored-by: Mryange <yanxuech...@selectdb.com>
---
be/src/pipeline/pipeline.h | 8 ++++++++
be/src/pipeline/pipeline_fragment_context.cpp | 10 +++++++---
2 files changed, 15 insertions(+), 3 deletions(-)
diff --git a/be/src/pipeline/pipeline.h b/be/src/pipeline/pipeline.h
index 619848110d4..061a62ea99b 100644
--- a/be/src/pipeline/pipeline.h
+++ b/be/src/pipeline/pipeline.h
@@ -73,6 +73,14 @@ public:
return idx == ExchangeType::HASH_SHUFFLE || idx ==
ExchangeType::BUCKET_HASH_SHUFFLE;
}
+ // For HASH_SHUFFLE, BUCKET_HASH_SHUFFLE, and ADAPTIVE_PASSTHROUGH,
+ // data is processed and shuffled on the sink.
+ // Compared to PASSTHROUGH, this is a relatively heavy operation.
+ static bool heavy_operations_on_the_sink(ExchangeType idx) {
+ return idx == ExchangeType::HASH_SHUFFLE || idx ==
ExchangeType::BUCKET_HASH_SHUFFLE ||
+ idx == ExchangeType::ADAPTIVE_PASSTHROUGH;
+ }
+
bool need_to_local_exchange(const DataDistribution
target_data_distribution,
const int idx) const;
void init_data_distribution() {
diff --git a/be/src/pipeline/pipeline_fragment_context.cpp
b/be/src/pipeline/pipeline_fragment_context.cpp
index 01c14f1ddb3..48f2d96597c 100644
--- a/be/src/pipeline/pipeline_fragment_context.cpp
+++ b/be/src/pipeline/pipeline_fragment_context.cpp
@@ -807,7 +807,7 @@ Status PipelineFragmentContext::_add_local_exchange_impl(
}
case ExchangeType::ADAPTIVE_PASSTHROUGH:
shared_state->exchanger = AdaptivePassthroughExchanger::create_unique(
- cur_pipe->num_tasks(), _num_instances,
+ std::max(cur_pipe->num_tasks(), _num_instances),
_num_instances,
_runtime_state->query_options().__isset.local_exchange_free_blocks_limit
?
_runtime_state->query_options().local_exchange_free_blocks_limit
: 0);
@@ -907,9 +907,13 @@ Status PipelineFragmentContext::_add_local_exchange(
<< " cur_pipe->operators().size(): " <<
cur_pipe->operators().size()
<< " new_pip->operators().size(): " << new_pip->operators().size();
- // Add passthrough local exchanger if necessary
+ // There are some local shuffles with relatively heavy operations on the
sink.
+ // If the local sink concurrency is 1 and the local source concurrency is
n, the sink becomes a bottleneck.
+ // Therefore, local passthrough is used to increase the concurrency of the
sink.
+ // op -> local sink(1) -> local source (n)
+ // op -> local passthrough(1) -> local passthrough(n) -> local sink(n) ->
local source (n)
if (cur_pipe->num_tasks() > 1 && new_pip->num_tasks() == 1 &&
- Pipeline::is_hash_exchange(data_distribution.distribution_type)) {
+
Pipeline::heavy_operations_on_the_sink(data_distribution.distribution_type)) {
RETURN_IF_ERROR(_add_local_exchange_impl(
new_pip->operators().size(), pool, new_pip,
add_pipeline(new_pip, pip_idx + 2),
DataDistribution(ExchangeType::PASSTHROUGH),
do_local_exchange, num_buckets,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
