[fix](pipeline) Add streaming aggregation local exchange switch - #66222
Merged
Gabriel39 merged 1 commit intoJul 29, 2026
Merged
Conversation
Contributor
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
Contributor
Author
|
run buildall |
Contributor
Author
|
/review |
Gabriel39
marked this pull request as ready for review
July 29, 2026 07:10
Contributor
TPC-H: Total hot run time: 29437 ms |
Contributor
TPC-DS: Total hot run time: 177318 ms |
Contributor
ClickBench: Total hot run time: 25.47 s |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Request changes.
The dedicated option is wired through Thrift and BE safely, but two required execution paths are incomplete: default FE local-shuffle planning still uses the generic aggregation flag, and observer-to-master forwarding drops the new session value. The variable therefore does not reliably control streaming aggregation.
Critical checkpoint conclusions:
- Goal and proof: not achieved on the default FE-planned or forwarded-query paths; no new test proves either switch value.
- Scope and parallel paths: the four-file change is focused but omits the FE
AggregationNodeplanner path and forwarding metadata. - Concurrency and lifecycle: no new concurrency, lock, allocation, or lifecycle hazard was found.
StreamingAggOperatorX::init()rejects merge/finalize before distribution planning, so removing_needs_finalizefrom the early return is safe on reachable BE paths. - Conditions and data correctness: BE preserves hash-join passthrough, empty-key behavior, serial-child reshuffle behavior, grouping expressions, and the non-hash-child reshuffle guard. Final merge aggregation preserves rows; the defects are broken session semantics plus avoidable or suppressed local-exchange work.
- Configuration and compatibility: this is a per-query session variable. Thrift ID 228, the false default, BE
__isset, and rolling unknown-field behavior are compatible, but both FE and BE planner paths must consume the same setting and follower forwarding must preserve it. - Tests and results: no test file changed; existing coverage references only the old option, and the PR states tests were not run. Add direct BE branch tests, FE plan-shape/parity tests, and forwarding coverage.
- Observability, persistence, transactions, and writes: no new observability is required; reflective persistence and cloning are sound; no transaction, journal, or data-write behavior changes.
- User focus: no additional focus was provided; the whole PR was reviewed.
| tResult.setEnableDistinctStreamingAggregation(enableDistinctStreamingAggregation); | ||
| tResult.setEnableStreamingAggHashJoinForcePassthrough(enableStreamingAggHashJoinForcePassthrough); | ||
| tResult.setEnableLocalExchangeBeforeAgg(enableLocalExchangeBeforeAgg); | ||
| tResult.setEnableLocalExchangeBeforeStreamingAgg(enableLocalExchangeBeforeStreamingAgg); |
Contributor
There was a problem hiding this comment.
[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.
| @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) |
Contributor
There was a problem hiding this comment.
[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.
Mryange
approved these changes
Jul 29, 2026
github-actions
Bot
added
the
approved
Contributor
|
PR approved by at least one committer and no changes requested. |
Contributor
|
PR approved by anyone and no changes requested. |
Contributor
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
Contributor
FE Regression Coverage ReportIncrement line coverage |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
Add
enable_local_exchange_before_streaming_agg, defaulting tofalse, 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