This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 2c4b04878b5 [fix](pipeline) Add streaming aggregation local exchange
switch (#66222)
2c4b04878b5 is described below
commit 2c4b04878b5f5c63fcf6df2961f3c8d4b73ef5c1
Author: Gabriel <[email protected]>
AuthorDate: Wed Jul 29 23:29:06 2026 +0800
[fix](pipeline) Add streaming aggregation local exchange switch (#66222)
### 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
Using the generic aggregation switch changes streaming aggregation
whenever it is enabled. A dedicated switch preserves the existing
streaming aggregation distribution by default. The master-specific
non-hash child distribution guard remains in place so grouping keys are
reshuffled when required.
### Validation
- 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 924120d9ad5..b23c79477a1 100644
--- a/be/src/exec/operator/streaming_aggregation_operator.h
+++ b/be/src/exec/operator/streaming_aggregation_operator.h
@@ -226,7 +226,9 @@ public:
state->enable_streaming_agg_hash_join_force_passthrough()) {
return {TLocalPartitionType::PASSTHROUGH};
}
- if (!_needs_finalize && !state->enable_local_exchange_before_agg() &&
+ // Preserve the inherited distribution by default, but still reshuffle
after a child
+ // non-hash exchange because it invalidates the grouping-key
distribution.
+ if (!state->enable_local_exchange_before_streaming_agg() &&
!child_breaks_local_key_distribution(state)) {
return
StatefulOperatorX<StreamingAggLocalState>::required_data_distribution(state);
}
diff --git a/be/src/runtime/runtime_state.h b/be/src/runtime/runtime_state.h
index 2a58dc31fc9..0551c2e5336 100644
--- a/be/src/runtime/runtime_state.h
+++ b/be/src/runtime/runtime_state.h
@@ -587,6 +587,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 7ba64eaf6fa..1938d19670b 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
@@ -174,6 +174,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";
@@ -1408,6 +1410,9 @@ public class SessionVariable implements Serializable,
Writable {
@VarAttrDef.VarAttr(name = ENABLE_LOCAL_EXCHANGE_BEFORE_AGG, fuzzy = true)
public boolean enableLocalExchangeBeforeAgg = true;
+ @VarAttrDef.VarAttr(name = ENABLE_LOCAL_EXCHANGE_BEFORE_STREAMING_AGG,
fuzzy = true)
+ public boolean enableLocalExchangeBeforeStreamingAgg = false;
+
@VarAttrDef.VarAttr(name =
ENABLE_DISTINCT_STREAMING_AGG_FORCE_PASSTHROUGH, fuzzy = true)
public boolean enableDistinctStreamingAggForcePassthrough = true;
@@ -3903,6 +3908,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();
@@ -5719,6 +5725,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 d15765ecf82..e4bde8cea41 100644
--- a/gensrc/thrift/PaloInternalService.thrift
+++ b/gensrc/thrift/PaloInternalService.thrift
@@ -508,6 +508,7 @@ struct TQueryOptions {
226: optional bool enable_prune_nested_column = false;
227: optional bool new_version_bitmap_op_count = false;
+ 228: 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]
