This is an automated email from the ASF dual-hosted git repository.
zclllyybb pushed a commit to branch opt_perf_4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/opt_perf_4.1 by this push:
new 1a9a446687e [Improvement](agg) Add a knob to control local exchange
(#65579)
1a9a446687e is described below
commit 1a9a446687ef935a81f89a9eb29c9a65810e1b18
Author: zclllyybb <[email protected]>
AuthorDate: Tue Jul 14 15:10:08 2026 +0800
[Improvement](agg) Add a knob to control local exchange (#65579)
## What problem does this PR solve?
Backport commit `7ec32b4ac59` from master to `opt_perf_4.1`, adding the
session/runtime configuration for enabling local exchange before
aggregation.
## Check List
- Test: No tests run; this is a requested branch backport with conflict
resolution only.
- Behavior changed: Yes, the backported session variable is available on
the target branch.
- Does this need documentation: No
Co-authored-by: Gabriel <[email protected]>
---
be/src/exec/operator/aggregation_sink_operator.h | 3 +++
be/src/exec/operator/distinct_streaming_aggregation_operator.h | 4 ++++
be/src/exec/operator/streaming_aggregation_operator.h | 3 +--
be/src/runtime/query_context.h | 6 ------
be/src/runtime/runtime_state.h | 5 +++++
fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java | 6 ++++++
6 files changed, 19 insertions(+), 8 deletions(-)
diff --git a/be/src/exec/operator/aggregation_sink_operator.h
b/be/src/exec/operator/aggregation_sink_operator.h
index fe0a4023cda..35823d075a2 100644
--- a/be/src/exec/operator/aggregation_sink_operator.h
+++ b/be/src/exec/operator/aggregation_sink_operator.h
@@ -160,6 +160,9 @@ public:
:
DataSinkOperatorX<AggSinkLocalState>::required_data_distribution(
state);
}
+ if (!_needs_finalize && !state->enable_local_exchange_before_agg()) {
+ return
DataSinkOperatorX<AggSinkLocalState>::required_data_distribution(state);
+ }
return _is_colocate && _require_bucket_distribution
? DataDistribution(ExchangeType::BUCKET_HASH_SHUFFLE,
_partition_exprs)
: DataDistribution(ExchangeType::HASH_SHUFFLE,
_partition_exprs);
diff --git a/be/src/exec/operator/distinct_streaming_aggregation_operator.h
b/be/src/exec/operator/distinct_streaming_aggregation_operator.h
index 145ad4c79b0..e0b175d93c5 100644
--- a/be/src/exec/operator/distinct_streaming_aggregation_operator.h
+++ b/be/src/exec/operator/distinct_streaming_aggregation_operator.h
@@ -118,6 +118,10 @@ public:
if (_needs_finalize && _probe_expr_ctxs.empty()) {
return {ExchangeType::NOOP};
}
+ if (!_needs_finalize && !state->enable_local_exchange_before_agg()) {
+ return
StatefulOperatorX<DistinctStreamingAggLocalState>::required_data_distribution(
+ state);
+ }
if (_needs_finalize || (!_probe_expr_ctxs.empty() &&
!_is_streaming_preagg)) {
return _is_colocate && _require_bucket_distribution
?
DataDistribution(ExchangeType::BUCKET_HASH_SHUFFLE, _partition_exprs)
diff --git a/be/src/exec/operator/streaming_aggregation_operator.h
b/be/src/exec/operator/streaming_aggregation_operator.h
index 162b362145e..5ed6d1481d5 100644
--- a/be/src/exec/operator/streaming_aggregation_operator.h
+++ b/be/src/exec/operator/streaming_aggregation_operator.h
@@ -220,8 +220,7 @@ public:
state->enable_streaming_agg_hash_join_force_passthrough()) {
return DataDistribution(ExchangeType::PASSTHROUGH);
}
- if (!state->get_query_ctx()->should_be_shuffled_agg(
- StatefulOperatorX<StreamingAggLocalState>::node_id())) {
+ if (!_needs_finalize && !state->enable_local_exchange_before_agg()) {
return
StatefulOperatorX<StreamingAggLocalState>::required_data_distribution(state);
}
if (_partition_exprs.empty()) {
diff --git a/be/src/runtime/query_context.h b/be/src/runtime/query_context.h
index 5e2b6babe87..3ef113c1fd1 100644
--- a/be/src/runtime/query_context.h
+++ b/be/src/runtime/query_context.h
@@ -190,12 +190,6 @@ public:
return _query_options.__isset.enable_force_spill &&
_query_options.enable_force_spill;
}
const TQueryOptions& query_options() const { return _query_options; }
- bool should_be_shuffled_agg(int node_id) const {
- return _query_options.__isset.shuffled_agg_ids &&
- std::any_of(_query_options.shuffled_agg_ids.begin(),
- _query_options.shuffled_agg_ids.end(),
- [&](const int id) -> bool { return id == node_id;
});
- }
// global runtime filter mgr, the runtime filter have remote target or
// need local merge should regist here. before publish() or
push_to_remote()
diff --git a/be/src/runtime/runtime_state.h b/be/src/runtime/runtime_state.h
index 13e4b7b6355..93a9e660056 100644
--- a/be/src/runtime/runtime_state.h
+++ b/be/src/runtime/runtime_state.h
@@ -581,6 +581,11 @@ public:
_query_options.enable_streaming_agg_hash_join_force_passthrough;
}
+ bool enable_local_exchange_before_agg() const {
+ return _query_options.__isset.enable_local_exchange_before_agg &&
+ _query_options.enable_local_exchange_before_agg;
+ }
+
bool enable_distinct_streaming_agg_force_passthrough() const {
return
_query_options.__isset.enable_distinct_streaming_agg_force_passthrough &&
_query_options.enable_distinct_streaming_agg_force_passthrough;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index a0208174b81..5efec5d589d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -172,6 +172,7 @@ public class SessionVariable implements Serializable,
Writable {
public static final String ENABLE_DISTINCT_STREAMING_AGG_FORCE_PASSTHROUGH
=
"enable_distinct_streaming_agg_force_passthrough";
public static final String ENABLE_BROADCAST_JOIN_FORCE_PASSTHROUGH =
"enable_broadcast_join_force_passthrough";
+ public static final String ENABLE_LOCAL_EXCHANGE_BEFORE_AGG =
"enable_local_exchange_before_agg";
public static final String DISABLE_COLOCATE_PLAN = "disable_colocate_plan";
public static final String COLOCATE_MAX_PARALLEL_NUM =
"colocate_max_parallel_num";
public static final String ENABLE_BUCKET_SHUFFLE_JOIN =
"enable_bucket_shuffle_join";
@@ -1362,6 +1363,9 @@ public class SessionVariable implements Serializable,
Writable {
@VariableMgr.VarAttr(name =
ENABLE_STREAMING_AGG_HASH_JOIN_FORCE_PASSTHROUGH, fuzzy = true)
public boolean enableStreamingAggHashJoinForcePassthrough = true;
+ @VariableMgr.VarAttr(name = ENABLE_LOCAL_EXCHANGE_BEFORE_AGG, fuzzy = true)
+ public boolean enableLocalExchangeBeforeAgg = true;
+
@VariableMgr.VarAttr(name =
ENABLE_DISTINCT_STREAMING_AGG_FORCE_PASSTHROUGH, fuzzy = true)
public boolean enableDistinctStreamingAggForcePassthrough = true;
@@ -3698,6 +3702,7 @@ public class SessionVariable implements Serializable,
Writable {
this.enableCommonExpPushDownForInvertedIndex = random.nextBoolean();
this.disableStreamPreaggregations = random.nextBoolean();
this.enableStreamingAggHashJoinForcePassthrough = random.nextBoolean();
+ this.enableLocalExchangeBeforeAgg = random.nextBoolean();
this.enableDistinctStreamingAggForcePassthrough = random.nextBoolean();
this.enableBroadcastJoinForcePassthrough = random.nextBoolean();
this.enableShareHashTableForBroadcastJoin = random.nextBoolean();
@@ -5481,6 +5486,7 @@ public class SessionVariable implements Serializable,
Writable {
tResult.setDisableStreamPreaggregations(disableStreamPreaggregations);
tResult.setEnableDistinctStreamingAggregation(enableDistinctStreamingAggregation);
tResult.setEnableStreamingAggHashJoinForcePassthrough(enableStreamingAggHashJoinForcePassthrough);
+ tResult.setEnableLocalExchangeBeforeAgg(enableLocalExchangeBeforeAgg);
tResult.setEnableDistinctStreamingAggForcePassthrough(enableDistinctStreamingAggForcePassthrough);
tResult.setEnableBroadcastJoinForcePassthrough(enableBroadcastJoinForcePassthrough);
tResult.setPartitionTopnMaxPartitions(partitionTopNMaxPartitions);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
