Skip to content

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

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
Gabriel39 merged 1 commit into from
Jul 29, 2026
+16 −1
Merged

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

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 @@ -226,7 +226,9 @@ class StreamingAggOperatorX MOCK_REMOVE(final) : public StatefulOperatorX<Stream
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);
}
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 @@ -587,6 +587,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 @@ -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";
Expand Down Expand Up @@ -1408,6 +1410,9 @@ public void checkQuerySlotCount(String slotCnt) {
@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)

@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.

[P1] Forward this query-affecting variable to the master

getForwardVariables() only includes fields marked needForward or one of the query-result-affect flags, and observer queries sent by FEOpExecutor are planned under a fresh master context. Because this annotation sets none of them, SET enable_local_exchange_before_streaming_agg=true on a follower is omitted and the master keeps false, so the knob silently changes behavior depending on the connected FE. Please add needForward=true and cover follower-to-master restoration.

public boolean enableLocalExchangeBeforeStreamingAgg = false;

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

Expand Down Expand Up @@ -3903,6 +3908,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 @@ -5719,6 +5725,7 @@ public TQueryOptions toThrift() {
tResult.setEnableDistinctStreamingAggregation(enableDistinctStreamingAggregation);
tResult.setEnableStreamingAggHashJoinForcePassthrough(enableStreamingAggHashJoinForcePassthrough);
tResult.setEnableLocalExchangeBeforeAgg(enableLocalExchangeBeforeAgg);
tResult.setEnableLocalExchangeBeforeStreamingAgg(enableLocalExchangeBeforeStreamingAgg);

@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.

[P1] Make the FE planner honor this dedicated switch

With enable_local_shuffle_planner=true (the default), NereidsPlanner inserts local exchanges in FE and RuntimeState::plan_local_shuffle() prevents BE from replanning them. AggregationNode.enforceAndDeriveLocalExchange() still gates the useStreamingPreagg branch on enableLocalExchangeBeforeAgg, so new=false/old=true still inserts a hash exchange and new=true/old=false suppresses it—the new variable has no effect on the default path. Please use enableLocalExchangeBeforeStreamingAgg in that branch, retain the old flag for the other aggregation operators, preserve the non-hash-child reshuffle invariant, and add opposite-value FE/BE planner tests plus direct BE branch coverage.

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 @@ -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.
Expand Down
Loading