Skip to content

[fix](pipeline) Add streaming aggregation local exchange switch #66216

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yiguolei merged 1 commit into from
Jul 30, 2026
+16 −1
Merged

[fix](pipeline) Add streaming aggregation local exchange switch #66216

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion be/src/exec/operator/streaming_aggregation_operator.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -220,7 +220,9 @@ class StreamingAggOperatorX MOCK_REMOVE(final) : public StatefulOperatorX<Stream
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()) {
Expand Down
5 changes: 5 additions & 0 deletions be/src/runtime/runtime_state.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -590,6 +590,11 @@ class RuntimeState {
_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;
Expand Down
7 changes: 7 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -1377,6 +1379,9 @@ public void checkQuerySlotCount(String slotCnt) {
@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)

@github-actions github-actions Bot Jul 29, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Forward the new switch to the master FE

Please add needForward = true to this VarAttr. FEOpExecutor forwards only SessionVariable.getForwardVariables(), whose filter excludes this field as currently annotated; ConnectProcessor then reconstructs the master-side session from that filtered map. Whenever a follower forwards a SELECT or INSERT (for example because it cannot read), enable_local_exchange_before_streaming_agg=true is therefore dropped and SessionVariable.toThrift() sends the master's default false, so the switch silently has no effect. Please cover this forwarding round trip as well.

public boolean enableLocalExchangeBeforeStreamingAgg = false;

@VariableMgr.VarAttr(name = ENABLE_DISTINCT_STREAMING_AGG_FORCE_PASSTHROUGH, fuzzy = true)
public boolean enableDistinctStreamingAggForcePassthrough = true;

Expand Down Expand Up @@ -3698,6 +3703,7 @@ public void initFuzzyModeVariables() {
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();
Expand Down Expand Up @@ -5506,6 +5512,7 @@ public TQueryOptions toThrift() {
tResult.setEnableDistinctStreamingAggregation(enableDistinctStreamingAggregation);
tResult.setEnableStreamingAggHashJoinForcePassthrough(enableStreamingAggHashJoinForcePassthrough);
tResult.setEnableLocalExchangeBeforeAgg(enableLocalExchangeBeforeAgg);
tResult.setEnableLocalExchangeBeforeStreamingAgg(enableLocalExchangeBeforeStreamingAgg);
tResult.setEnableDistinctStreamingAggForcePassthrough(enableDistinctStreamingAggForcePassthrough);
tResult.setEnableBroadcastJoinForcePassthrough(enableBroadcastJoinForcePassthrough);
tResult.setPartitionTopnMaxPartitions(partitionTopNMaxPartitions);
Expand Down
1 change: 1 addition & 0 deletions gensrc/thrift/PaloInternalService.thrift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
Loading