Repository navigation
Conversation
Contributor
Author
|
run buildall |
Contributor
Author
|
/review |
Contributor
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
Gabriel39
force-pushed
the
agent/fix-streaming-agg-required-distribution
branch
from
July 29, 2026 06:44
24bcb95 to
d1eeddf
Compare
Gabriel39
changed the title
Contributor
Author
|
run buildall |
Contributor
Author
|
/review |
Gabriel39
force-pushed
the
agent/fix-streaming-agg-required-distribution
branch
from
July 29, 2026 06:48
d1eeddf to
8808186
Compare
Contributor
Author
|
run buildall |
Contributor
Author
|
/review |
Gabriel39
marked this pull request as ready for review
July 29, 2026 06:55
Contributor
Cloud UT Coverage ReportIncrement line coverage Increment coverage report
|
Contributor
FE UT Coverage ReportIncrement line coverage |
![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: one P2 issue remains.
The new enable_local_exchange_before_streaming_agg session variable is not included in follower-to-master session forwarding. For forwarded SELECT or INSERT statements, an explicit true is lost and the master sends the default false to BE, so the switch silently cannot enable the requested exchange.
Critical checkpoints:
- Distribution and lifecycle: the default-false inherited distribution is safe for partial, non-finalizing
StreamingAggOperatorX; downstream shuffle propagation, task-local state, hash-join passthrough precedence, empty-partition fallback, pushed Top-N, and query-cache paths were checked with no additional defect. - FE/Thrift/BE contract: declaration, persistence and cloning, randomization, Thrift field ID 226, Java setter/
__isset, absent-field false default, and mixed-version unknown-field behavior were checked. FE statement forwarding is the blocking gap. - Coverage and validation: no builds or tests were run by this review runner, per the review-only instruction. Existing tests do not directly cover the switch matrix or the FE forwarding round trip; the inline comment requests targeted forwarding coverage. At payload freeze, style, formatter, and Cloud UT checks passed, while compile and FE/BE unit checks were still pending.
- Existing context and user focus: there were no existing inline review threads and no additional user-provided focus; the complete PR was reviewed.
| @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) |
Contributor
There was a problem hiding this comment.
[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.
Contributor
There was a problem hiding this comment.
Follow-up review: no new nonduplicate inline findings.
The existing P2 at discussion_r3672330401 remains valid and blocking: the new
session switch needs needForward = true plus a follower-to-master forwarding
round-trip test. I did not repeat that inline comment.
Critical checkpoints:
- Distribution and lifecycle: every reachable
StreamingAggOperatorXis a
non-finalizing partial aggregate with grouping keys. Default false safely
inherits the current distribution; explicit true requests hash shuffle;
direct hash-join passthrough, serial-child inheritance, downstream
hash/bucket propagation, query-cache wrapping, and group-limit paths retain
their existing contracts. - FE/Thrift/BE contract: SET, persistence, cloning, fuzzy initialization,
toThrift, field ID 226, Java/BE names, false defaults, absent fields, and
mixed-version unknown-field handling are consistent. Follower forwarding is
the one gap already covered by the existing thread. - Compatibility and scope: the definition-only legacy
shuffled_agg_node_idspath and the partitioned hash-join wrapper behavior
both pre-exist at the authoritative base and are not introduced by this PR. - Coverage and validation: this review-only runner performed no builds or
tests. Direct switch-matrix and optional-field tests are absent; only the
missing forwarding round trip corresponds to a substantiated defect. - Existing context and user focus: all existing review context was
deduplicated, and there was no additional user-provided focus.
Review completion: complete for
880818677ff0db6066c0a4e5e4e828536a115424; no unresolved new candidate
remains, but the already-requested forwarding fix is still required.
Contributor
There was a problem hiding this comment.
Final convergence verification for 8808186: request changes remains appropriate. The existing P2 forwarding defect is the only blocker; no new inline comment is added because it is already covered at discussion_r3672330401. A concurrent body-only follow-up review (4806100152) reached the same conclusion.
Critical checkpoints:
- Distribution and lifecycle: the default-false path safely inherits upstream distribution for the reachable non-finalizing StreamingAggOperatorX; explicit true requests the expected hash exchange, while hash-join passthrough precedence, empty grouping, downstream shuffle propagation, and serial/task safeguards remain sound.
- FE/Thrift/BE contract: declaration, reflective SET registration, persistence and cloning, toThrift, unique optional field 226, RuntimeState copy/getter behavior, absent-field false semantics, and mixed-version unknown-field handling are sound. Follower-to-master session forwarding remains the blocking gap.
- Coverage and validation: this review runner performed no builds or tests, as required. Existing tests do not directly cover the switch matrix or follower forwarding round trip; the existing inline thread requests targeted coverage.
- Existing context and user focus: the known forwarding thread was deduplicated, no distinct additional issue was found, and there was no additional user-provided focus. The complete four-file PR was reviewed.
- Review completion: the repository review skill was unavailable; the authoritative bundle prompt was followed as the fallback. Two normal full-coverage reviewers and the separate risk-focused reviewer converged with NO_NEW_VALUABLE_FINDINGS.
Contributor
FE Regression Coverage ReportIncrement line coverage |
yiguolei
approved these changes
Jul 30, 2026
This was referenced Nov 4, 2026
Merged
morningman
added a commit
that referenced
this pull request
Nov 5, 2026
…local shuffle planner (#70177) ### What problem does this PR solve? Issue Number: None Related PR: #66216, #66222, #63766 Problem Summary: **In short.** This brings the FE local shuffle planner in line with `enable_local_exchange_before_streaming_agg`, which branch-4.1 / 4.2 #66216 and master #66222 added on the BE side only. With the FE planner, which master uses by default, a streaming pre-aggregation still asked for a local hash exchange on its grouping keys whatever the variable said. Master plans therefore had a `LOCAL_EXECUTION_HASH_SHUFFLE` local exchange that the BE planner, and branch-4.x, do not add, for example right above a `UNION ALL` or an external table scan. The FE now follows the same rule as BE `StreamingAggOperatorX`. **Background.** #66222 added `enable_local_exchange_before_streaming_agg` (default false) so that a streaming pre-aggregation keeps the distribution it inherits. It replaced `enable_local_exchange_before_agg` for streaming aggregation. In `StreamingAggOperatorX::required_data_distribution()`, the operator asks for a hash local exchange on its partition expressions only when the variable is on, or when `OperatorBase::child_breaks_local_key_distribution()` (#63766) says the child spreads rows without regard to any key: the child is serial, or the child's own required distribution is a non-hash local exchange (PASSTHROUGH, ADAPTIVE_PASSTHROUGH, ...). **The problem, and what it cost.** On master the FE plans local exchanges (`enable_local_shuffle_planner`, default true) and BE does not. `AggregationNode.enforceAndDeriveLocalExchange()` mirrors the BE operators, but its streaming aggregation branch still had the rule from before #66222. It never read the new variable. It skipped the hash requirement only when `enable_local_exchange_before_agg` was off, and then without the child check. With the default settings it required a hash distribution for every streaming pre-aggregation with grouping keys, so: - Below a streaming pre-aggregation whose child keeps the key distribution but does not report a hash distribution, the FE inserted a `LOCAL_EXECUTION_HASH_SHUFFLE` that BE planning does not. Examples are a `UNION ALL` and an external table scan that is not serial. Every row was hashed and moved once more before the pre-aggregation. - `enable_local_exchange_before_streaming_agg=true` had no effect with the FE planner. - With `enable_local_exchange_before_agg=false`, the FE dropped the hash requirement even after a child that breaks the key distribution, where BE keeps it. None of this changes query results: a streaming pre-aggregation only affects performance, and the operators that need hash-distributed input for correctness ask for it themselves. **How this PR fixes it.** The streaming aggregation branch of `AggregationNode.enforceAndDeriveLocalExchange()` now follows BE: after the existing hash join passthrough rule, it keeps the base class requirement (no exchange, or PASSTHROUGH over a serial child) unless `enable_local_exchange_before_streaming_agg` is on or the child breaks the key distribution. `enable_local_exchange_before_agg` no longer affects streaming aggregation. The new `childBreaksLocalKeyDistribution()` mirrors `child_breaks_local_key_distribution()` with each BE operator's `required_data_distribution()`: | Child | BE operator's required distribution | Breaks the key distribution | |---|---|---| | serial on BE (pooling scan, serial exchange, ...) | — | yes | | nested loop join | ADAPTIVE_PASSTHROUGH, NOOP for null-aware anti, right outer / semi / anti and full outer joins | yes, except those joins | | broadcast hash join | PASSTHROUGH when forced (`enable_broadcast_join_force_passthrough`) or when the probe child is serial | in those two cases | | table function, assert num rows | PASSTHROUGH | yes | | anything else (scan, exchange, union, partitioned join, ...) | NOOP or a hash distribution | no | `SelectNode` and `RepeatNode` take the serial flag of their child, so the serial check already covers the base class rule (PASSTHROUGH over a serial child) for them. **Results.** Local exchanges in the plans of three queries on a one-BE cluster, `parallel_pipeline_task_num=4`, non-pooling scans, taken from the query profiles: | Query | BE planner | FE planner before | FE planner after | |---|---|---|---| | `GROUP BY` over `UNION ALL` of two tables | none | LOCAL_HASH_SHUFFLE | none | | `GROUP BY` over a nested loop join | ADAPTIVE_PASSTHROUGH, LOCAL_HASH_SHUFFLE | same | same | | `GROUP BY` over an Iceberg table with 6 data files | none | LOCAL_HASH_SHUFFLE | none | A planner probe over 72 plans (8 query shapes × pooling / non-pooling scans × `agg_phase` 0 / 2 × the variable off / on) found only the `UNION ALL` plans changed with the variable off. A scan of an OLAP table, a nested loop join, `explode`, broadcast and shuffle joins, a `LIMIT` subquery and `GROUPING SETS` gave the same plans as before.
morningman
added a commit
that referenced
this pull request
Nov 6, 2026
…#70197) ### What problem does this PR solve? Issue Number: None Related PR: #69273, #70177 Problem Summary: **In short.** Found while bringing branch-4.2 #66216 to the FE local shuffle planner (#70177): four session variables that decide the local exchanges of a plan are not forwarded to the master. #69273 made the planner and executor variables travel with a statement that a follower or an observer forwards to the master, `enable_local_exchange_before_agg` among them, but not `enable_local_exchange_before_streaming_agg`, `enable_streaming_agg_hash_join_force_passthrough`, `enable_distinct_streaming_agg_force_passthrough` and `force_to_local_shuffle`. A session-level `SET` of one of them is dropped on that path, and the master plans the statement with its own global value. This PR forwards the four as well. **Background.** A statement that has to run on the master FE (`INSERT INTO ... SELECT`, `CREATE TABLE ... AS SELECT`, ...) and arrives at a follower or an observer is forwarded: `FEOpExecutor` sends `SessionVariable#getForwardVariables()`, and `ConnectProcessor` on the master applies them with `setForwardedSessionVariables()` to a session that starts from the master's global variables. Only variables whose `@VarAttr` sets `needForward` (or one of the `affectQueryResult*` attributes) are sent. **The problem, and what it cost.** These four do not set it: | Variable | Default | What it decides | |---|---|---| | `enable_local_exchange_before_streaming_agg` (#66222) | false | whether a streaming pre-aggregation asks for a local hash exchange on its grouping keys: BE `StreamingAggOperatorX`, and the FE planner with #70177 | | `enable_streaming_agg_hash_join_force_passthrough` | true | whether a streaming pre-aggregation right above a hash join asks for a passthrough local exchange: the FE planner (`AggregationNode`) and BE `StreamingAggOperatorX` | | `enable_distinct_streaming_agg_force_passthrough` | true | the same for a distinct streaming pre-aggregation: the FE planner and BE `DistinctStreamingAggOperatorX` | | `force_to_local_shuffle` | false | makes every scan serial (`ScanNode#isSerialNode`), so the FE planner pools the scan and adds a local shuffle above it | So a session that sets one of them gets that plan only for the statements its own FE runs. A forwarded statement, such as an `INSERT INTO ... SELECT` or a CTAS issued through a follower or an observer, is planned with the master's global value, and `EXPLAIN` on the follower shows a plan that the master does not run. Query results stay the same, since these variables only decide how rows move between the pipeline tasks of a BE; what differs is the plan, and with it the speed of the statement. **How this PR fixes it.** Set `needForward = true` on the four, as #69273 does for the others, and add them to the list in `SessionVariablesTest#testForwardPlanAndExecutionKnobs`. **Results.** | Statement forwarded from a follower after `SET <variable> = <non-default value>` | Before | After | |---|---|---| | Value the master plans it with | the master's global value | the session's value | branch-4.1 and branch-4.2 have the same gap, and #69273 is not on them either.
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
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
The historical
QueryContext::should_be_shuffled_agg()interface is no longer available on branch-4.1. Reusingenable_local_exchange_before_aggalso 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