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 758aa0a7d81 [fix](pipeline) Add streaming aggregation local exchange
switch (#66216)
758aa0a7d81 is described below
commit 758aa0a7d81344dfdacb6d7d1666acc7666c9f8b
Author: Gabriel <[email protected]>
AuthorDate: Thu Jul 30 08:41:40 2026 +0800
[fix](pipeline) Add streaming aggregation local exchange switch (#66216)
### What
Add `enable_local_exchange_before_streaming_agg`, defaulting to `false`,
and use it to control whether streaming aggregation requests a local
hash exchange.
### Why
The historical `QueryContext::should_be_shuffled_agg()` interface is no
longer available on branch-4.1. Reusing
`enable_local_exchange_before_agg` also changes streaming aggregation
whenever the generic aggregation switch is enabled. A dedicated switch
preserves the existing streaming aggregation distribution by default
while allowing the local exchange to be enabled explicitly.
### Validation
- `git diff --check`
- clang-format 16 dry-run
- Local tests not run as requested.
---
be/src/exec/operator/streaming_aggregation_operator.h | 4 +++-
be/src/runtime/runtime_state.h | 5 +++++
fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java | 7 +++++++
gensrc/thrift/PaloInternalService.thrift | 1 +
4 files changed, 16 insertions(+), 1 deletion(-)
diff --git a/be/src/exec/operator/streaming_aggregation_operator.h
b/be/src/exec/operator/streaming_aggregation_operator.h
index 5ed6d1481d5..18dcb4ca906 100644
--- a/be/src/exec/operator/streaming_aggregation_operator.h
+++ b/be/src/exec/operator/streaming_aggregation_operator.h
@@ -220,7 +220,9 @@ public:
state->enable_streaming_agg_hash_join_force_passthrough()) {
return DataDistribution(ExchangeType::PASSTHROUGH);
}
- if (!_needs_finalize && !state->enable_local_exchange_before_agg()) {
+ // Keep streaming aggregation on its inherited distribution unless the
dedicated switch
+ // explicitly enables a local hash exchange.
+ if (!state->enable_local_exchange_before_streaming_agg()) {
return
StatefulOperatorX<StreamingAggLocalState>::required_data_distribution(state);
}
if (_partition_exprs.empty()) {
diff --git a/be/src/runtime/runtime_state.h b/be/src/runtime/runtime_state.h
index fc764085c86..688ae013a94 100644
--- a/be/src/runtime/runtime_state.h
+++ b/be/src/runtime/runtime_state.h
@@ -590,6 +590,11 @@ public:
_query_options.enable_local_exchange_before_agg;
}
+ bool enable_local_exchange_before_streaming_agg() const {
+ return
_query_options.__isset.enable_local_exchange_before_streaming_agg &&
+ _query_options.enable_local_exchange_before_streaming_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 e663c440af4..74caa6d0fad 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,8 @@ public class SessionVariable implements Serializable,
Writable {
"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 ENABLE_LOCAL_EXCHANGE_BEFORE_STREAMING_AGG =
+ "enable_local_exchange_before_streaming_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";
@@ -1378,6 +1380,9 @@ public class SessionVariable implements Serializable,
Writable {
@VariableMgr.VarAttr(name = ENABLE_LOCAL_EXCHANGE_BEFORE_AGG, fuzzy = true)
public boolean enableLocalExchangeBeforeAgg = true;
+ @VariableMgr.VarAttr(name = ENABLE_LOCAL_EXCHANGE_BEFORE_STREAMING_AGG,
fuzzy = true)
+ public boolean enableLocalExchangeBeforeStreamingAgg = false;
+
@VariableMgr.VarAttr(name =
ENABLE_DISTINCT_STREAMING_AGG_FORCE_PASSTHROUGH, fuzzy = true)
public boolean enableDistinctStreamingAggForcePassthrough = true;
@@ -3711,6 +3716,7 @@ public class SessionVariable implements Serializable,
Writable {
this.disableStreamPreaggregations = random.nextBoolean();
this.enableStreamingAggHashJoinForcePassthrough = random.nextBoolean();
this.enableLocalExchangeBeforeAgg = random.nextBoolean();
+ this.enableLocalExchangeBeforeStreamingAgg = random.nextBoolean();
this.enableDistinctStreamingAggForcePassthrough = random.nextBoolean();
this.enableBroadcastJoinForcePassthrough = random.nextBoolean();
this.enableShareHashTableForBroadcastJoin = random.nextBoolean();
@@ -5519,6 +5525,7 @@ public class SessionVariable implements Serializable,
Writable {
tResult.setEnableDistinctStreamingAggregation(enableDistinctStreamingAggregation);
tResult.setEnableStreamingAggHashJoinForcePassthrough(enableStreamingAggHashJoinForcePassthrough);
tResult.setEnableLocalExchangeBeforeAgg(enableLocalExchangeBeforeAgg);
+
tResult.setEnableLocalExchangeBeforeStreamingAgg(enableLocalExchangeBeforeStreamingAgg);
tResult.setEnableDistinctStreamingAggForcePassthrough(enableDistinctStreamingAggForcePassthrough);
tResult.setEnableBroadcastJoinForcePassthrough(enableBroadcastJoinForcePassthrough);
tResult.setPartitionTopnMaxPartitions(partitionTopNMaxPartitions);
diff --git a/gensrc/thrift/PaloInternalService.thrift
b/gensrc/thrift/PaloInternalService.thrift
index 6add308255d..05deef4efa5 100644
--- a/gensrc/thrift/PaloInternalService.thrift
+++ b/gensrc/thrift/PaloInternalService.thrift
@@ -500,6 +500,7 @@ struct TQueryOptions {
224: optional bool enable_expr_zonemap_filter = true
225: optional i64 runtime_filter_tree_publish_max_send_bytes = 268435456
+ 226: optional bool enable_local_exchange_before_streaming_agg = false
// For cloud, to control if the content would be written into file cache
// In write path, to control if the content would be written into file cache.
// In read path, read from file cache or remote storage when execute query.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
