Repository navigation
[fix](local shuffle) Keep local exchanges before pre-aggregations over key-breaking children - #70296
Conversation
…r key-breaking children ### Background On master the FE plans the local exchanges that redistribute rows among the pipeline tasks of one BE (`enable_local_shuffle_planner`, on by default). Each plan node follows the rule of the BE operator it becomes, `required_data_distribution()`, which BE applies itself when the FE planner is off. `enable_local_exchange_before_agg` (default true) controls the exchange before a LOCAL (first-phase) aggregation. Since apache#63766, BE skips it with the switch off only while the child keeps the grouping-key distribution; a child breaks it when it is serial or asks for a non-hash local exchange itself, such as the probe side of a nested loop join or a table function. ### What problem does this PR solve? Issue Number: None Related PR: apache#63366 (introduced the FE rule without the child check), apache#63766 (the BE rule), apache#70177 (the same fix for streaming aggregation; this PR reuses its `childBreaksLocalKeyDistribution()`), apache#62438 (added the switch) PR Type: Bugfix Problem Summary: With `enable_local_exchange_before_agg=false`, the FE planner dropped the local exchange below a LOCAL `AggSink` and below a streaming distinct pre-aggregation whatever their child was, while BE keeps it after a child that breaks the key distribution. Queries planned by the FE therefore got fewer local exchanges than BE planning gives, for example no hash exchange between a nested loop join and the pre-aggregation above it. This PR adds BE's child check to both cases. Results do not change; most of the diff is tests. ### Details - **Change map:** production code +15 / -7 in `AggregationNode` (two conditions and their comments) and a `PlanNode` javadoc; tests +345 / -34: cases in `LocalShuffleNodeCoverageTest` and `LocalExchangePlannerTest`, and the new regression suite `nereids_p0/local_shuffle/test_local_exchange_before_agg_breaking_child` with its generated `.out` (+169). - **Frequency:** Always, when `enable_local_exchange_before_agg=false` (not the default; P0 randomizes it) and a LOCAL aggregation sits on a child that breaks the key distribution. Only plans and speed change: the FE always requires hash input at FIRST_MERGE and at the non-streaming DISTINCT dedup (a deliberate divergence from BE), so results stay right. - **Root cause:** `AggregationNode.enforceAndDeriveLocalExchange()` mirrors the BE operators, but its DistinctStreamingAgg and AggSink LOCAL branches fall back to the base class on `!enableLeBeforeAgg` alone, without BE's `child_breaks_local_key_distribution()`. apache#63366 wrote them after apache#63766 had merged, from the rule before it. - **Why existing tests missed it:** Unit tests: the two "LE disabled" cases in `LocalShuffleNodeCoverageTest` use a child that keeps the key distribution, no case reached the DistinctStreamingAgg branch, and `LocalExchangePlannerTest` had no plan with the switch off. Regression: `query_p0/join/test_agg_after_nested_loop_join_local_exchange` (apache#63766's suite) and `nereids_p0/local_shuffle/test_enable_local_exchange_before_agg` compare only results, which were right, and no `.out` shows local exchanges. - **Fix:** Both branches fall back to the base class only when `childBreaksLocalKeyDistribution()` is false. After a key-breaking child, the AggSink LOCAL phase goes on to its hash requirement, and the streaming distinct pre-aggregation to `enable_distinct_streaming_agg_force_passthrough` (PASSTHROUGH, else the base class), as on BE; it never asks for hash. - **Other paths and branches:** Streaming aggregation got the check in apache#70177; BE needs no change. One known difference from BE planning remains, with the switch on or off: the FE lets an ADAPTIVE_PASSTHROUGH or PASSTHROUGH child output satisfy a PASSTHROUGH requirement, where BE `Pipeline::need_to_local_exchange()` always adds one (last row of the table in Testing). branch-4.1 and branch-4.2 have no FE local shuffle planner and are not affected. ### Testing - `LocalShuffleNodeCoverageTest` and `LocalExchangePlannerTest`, two new cases each: all four fail without the fix (for example, a serial child got `PASSTHROUGH` instead of `LOCAL_EXECUTION_HASH_SHUFFLE`) and pass with it. The 12 fe-core test classes that plan local exchanges or print distributed plans pass (168 tests). Run locally. - A planner probe over 288 plans (18 query shapes × pooling on and off × `agg_phase` 0 and 2 × `disable_streaming_preaggregations` on and off × the switch): only 32 with the switch off change. 30 of them now equal the plan with the switch on; the other 2 keep the distribution of a `UNION ALL`, as the switch asks. - Local exchanges in the query profiles on a local 1 FE + 1 BE cluster (master `4530f761fa1`; `doris-fe.jar` from `7c4b4462dcc` with apache#70177, and for "after" with only `AggregationNode.class` replaced), switch off, `parallel_pipeline_task_num=4`, `agg_phase=2`. The `AggSink` rows set `disable_streaming_preaggregations=true`; the distinct rows are a `GROUP BY` without aggregate functions: | Query | BE planner | FE planner before | FE planner after | |---|---|---|---| | LOCAL `AggSink` over a nested loop join | ADAPTIVE_PASSTHROUGH, LOCAL_HASH_SHUFFLE | ADAPTIVE_PASSTHROUGH | ADAPTIVE_PASSTHROUGH, LOCAL_HASH_SHUFFLE | | LOCAL `AggSink` over `explode` | PASSTHROUGH, LOCAL_HASH_SHUFFLE | PASSTHROUGH | PASSTHROUGH, LOCAL_HASH_SHUFFLE | | LOCAL `AggSink` over a one-tablet scan (serial) | PASSTHROUGH, LOCAL_HASH_SHUFFLE | PASSTHROUGH | PASSTHROUGH, LOCAL_HASH_SHUFFLE | | LOCAL `AggSink` over `UNION ALL` | none | none | none | | distinct pre-aggregation over `explode` | PASSTHROUGH ×2 | PASSTHROUGH | PASSTHROUGH ×2 | | distinct pre-aggregation over a nested loop join | ADAPTIVE_PASSTHROUGH, PASSTHROUGH | ADAPTIVE_PASSTHROUGH | ADAPTIVE_PASSTHROUGH | - The new regression suite fails on the FE without the fix at its first plan check (`nlj: [VNESTED]`) and passes with it; its 16 result blocks (4 queries × both planners × the switch on and off) match results computed independently from the inserted rows. `nereids_p0/local_shuffle` (11 suites), `query_p0/join/test_agg_after_nested_loop_join_local_exchange` and `nereids_syntax_p0/agg_4_phase` pass on the same cluster with the fix. ### Notes for Reviewers - `baseClassRequire()` stays in both branches although only a non-serial child reaches them now, since a serial child always breaks the key distribution. It keeps the branches shaped like BE's call to the base class, as apache#70177 does for streaming aggregation. - In `LocalShuffleNodeCoverageTest`, apache#70177's five key-breaking children moved into `childrenThatBreakKeyDistribution()` and its aggregate builder into `buildLocalPhaseAgg()`, so the new cases reuse them; apache#70177's case is otherwise unchanged. - The `PlanNode#requiresShuffleForCorrectness` javadoc still named `enable_local_exchange_before_agg` for streaming aggregation, which reads `enable_local_exchange_before_streaming_agg` since apache#70177. ### Release note None ### Check List (For Author) - Behavior changed: No - Does this need documentation: No - Pre-submission status: Ready ### AI Disclosure - Tool: Claude Code (Claude Opus 5.5) - Generated: the fix, the unit and regression tests (the `.out` generated by the framework), and this description - Human review: not stated - Prompts: The author asked to continue moving branch-4.2's changes back to master, where this follow-up of apache#70177 was recorded together with the outdated `PlanNode` javadoc, and chose to verify it on a local cluster before opening the PR. Co-Authored-By: Claude Opus 5.5 <[email protected]>
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
|
run buildall |
|
/review |
There was a problem hiding this comment.
Reviewed all six changed files at 60bcdca1e807049183f085298a9ff071e13a3cc2, including the upstream aggregate translation and downstream BE operator/local-exchange construction. No verified new actionable findings. No existing inline findings require carrying forward. No additional user focus was supplied.
Critical checkpoint conclusions:
- Goal and proof: the two new guards restore the BE child-distribution exception for disabled-switch LOCAL AggSink and streaming-distinct preaggregation. Added node, planner, and regression assertions target the changed behavior; runtime success was not independently verified.
- Scope and reuse: two conditions reuse the existing child predicate; the accompanying javadoc and test helper extraction are focused.
- Concurrency and locks: this is a synchronous per-query planning pass before plan serialization; no new thread, shared mutable state, lock, blocking I/O, or lock-order change.
- Lifecycle and initialization: no new resource ownership, cleanup obligation, or static-initialization dependency.
- Configuration: existing per-session switches are reused; no configuration is added or changed in mutability/propagation.
- Compatibility: no function-signature, storage-format, or wire-schema change; existing local-exchange types and serialization are used.
- Thrift/protobuf IDs: no gensrc definitions change.
- Parallel paths: regular StreamingAgg already uses the predicate; both remaining affected FE branches are updated and compared with BE. Existing FE acceptance of PASSTHROUGH/ADAPTIVE_PASSTHROUGH remains a documented preexisting difference, so this patch does not make every FE/BE plan shape identical.
- Conditions and special cases: child serial flags, NLJ join-kind exclusions, forced broadcast passthrough, table functions, and AssertNumRows match the corresponding BE rules. Finalize, FIRST_MERGE, non-streaming dedup, and empty-effective-key paths retain their requirements. Hash-key selection, explicit hash flavors, serial fanout, and correctness-ancestor propagation were traced without finding a new defect.
- Test coverage: unit cases cover five key-breaking child categories and both distinct force-passthrough values; planner cases cover NLJ, table function, and serial scan. Regression cases check changed shapes plus key-preserving UNION and compare four query shapes across both planners and both switch values. Existing negative/exclusion cases remain; sorted output, drop-before table lifecycle, and session resets were inspected.
- Changed results: all 16 golden blocks and their matrix labels were independently recomputed from the inserted rows and checked against the framework's lexical ordering; they match. The author reports framework generation, but generation provenance was not independently verified.
- Observability: existing EXPLAIN DISTRIBUTED PLAN exchange types and BE profiles expose the affected plan decisions; no additional instrumentation is needed for this scoped change.
- Transactions and persistence: no EditLog, replay, commit, visible-version, or delete-bitmap behavior changes.
- Writes and atomicity: production changes affect query planning only; no data-write or crash-recovery path changes.
- FE/BE variable passing: no new transmitted variable or sending point.
- Performance: the added planner checks and local redistributions implement the existing BE choice after key-breaking children; retained hash output can satisfy the hash requirement. No new execution hot loop or substantiated avoidable exchange was identified; performance was not measured.
- Other issues: production, test, and additional risk reviewers completed their scopes and final missed-case rechecks without a substantiated new finding. No unresolved static candidate or uncovered changed file remains.
Static review only: no builds, Doris tests, cluster runs, benchmarks, source edits, or submodule initialization were performed. Author-reported execution results were inspected as claims, not reproduced. Convergence is limited to this static review scope.
Review converged after 1 round(s).
TPC-H: Total hot run time: 28618 ms |
TPC-DS: Total hot run time: 153522 ms |
ClickBench: Total hot run time: 24.41 s |
FE UT Coverage ReportIncrement line coverage |
924060929
left a comment
There was a problem hiding this comment.
This patch restores the child-distribution exception when local exchanges before preaggregation are disabled: LOCAL AggSink retains its hash requirement after a distribution-breaking child, while streaming distinct preaggregation follows its passthrough setting.
Independent static review of all six changed files at 60bcdca found no actionable P0/P1/P2 regression. I traced aggregate translation, exchange enforcement, hash-key selection, serial boundaries, serialization, and the necessary BE consumers. Existing correctness requirements remain intact, and the tests cover changed shapes and key-preserving cases. APPROVE within the FE review scope and necessary FE/BE boundaries.
No builds or tests were run. The supplied current-head CI snapshot reports successful P0, FE UT, and compilation checks, but failed FE coverage. Merge requirements were not verified.
Automated FE-focused static review (gpt-6.1-sol, high). No build or tests were run.
|
PR approved by anyone and no changes requested. |
|
PR approved by at least one committer and no changes requested. |
Background
On master the FE plans the local exchanges that redistribute rows among the pipeline tasks of one BE (
enable_local_shuffle_planner, on by default). Each plan node follows the rule of the BE operator it becomes,required_data_distribution(), which BE applies itself when the FE planner is off.enable_local_exchange_before_agg(default true) controls the exchange before a LOCAL (first-phase) aggregation. Since #63766, BE skips it with the switch off only while the child keeps the grouping-key distribution; a child breaks it when it is serial or asks for a non-hash local exchange itself, such as the probe side of a nested loop join or a table function.What problem does this PR solve?
Issue Number: None
Related PR: #63366 (introduced the FE rule without the child check), #63766 (the BE rule), #70177 (the same fix for streaming aggregation; this PR reuses its
childBreaksLocalKeyDistribution()), #62438 (added the switch)PR Type: Bugfix
Problem Summary:
With
enable_local_exchange_before_agg=false, the FE planner dropped the local exchange below a LOCALAggSinkand below a streaming distinct pre-aggregation whatever their child was, while BE keeps it after a child that breaks the key distribution. Queries planned by the FE therefore got fewer local exchanges than BE planning gives, for example no hash exchange between a nested loop join and the pre-aggregation above it. This PR adds BE's child check to both cases. Results do not change; most of the diff is tests.Details
AggregationNode(two conditions and their comments) and aPlanNodejavadoc; tests +345 / -34: cases inLocalShuffleNodeCoverageTestandLocalExchangePlannerTest, and the new regression suitenereids_p0/local_shuffle/test_local_exchange_before_agg_breaking_childwith its generated.out(+169).enable_local_exchange_before_agg=false(not the default; P0 randomizes it) and a LOCAL aggregation sits on a child that breaks the key distribution. Only plans and speed change: the FE always requires hash input at FIRST_MERGE and at the non-streaming DISTINCT dedup (a deliberate divergence from BE), so results stay right.AggregationNode.enforceAndDeriveLocalExchange()mirrors the BE operators, but its DistinctStreamingAgg and AggSink LOCAL branches fall back to the base class on!enableLeBeforeAggalone, without BE'schild_breaks_local_key_distribution(). [refactor](local shuffle) Move local exchange planning from BE to FE #63366 wrote them after [fix](be) Preserve agg hash shuffle after non-hash exchange #63766 had merged, from the rule before it.LocalShuffleNodeCoverageTestuse a child that keeps the key distribution, no case reached the DistinctStreamingAgg branch, andLocalExchangePlannerTesthad no plan with the switch off. Regression:query_p0/join/test_agg_after_nested_loop_join_local_exchange([fix](be) Preserve agg hash shuffle after non-hash exchange #63766's suite) andnereids_p0/local_shuffle/test_enable_local_exchange_before_aggcompare only results, which were right, and no.outshows local exchanges.childBreaksLocalKeyDistribution()is false. After a key-breaking child, the AggSink LOCAL phase goes on to its hash requirement, and the streaming distinct pre-aggregation toenable_distinct_streaming_agg_force_passthrough(PASSTHROUGH, else the base class), as on BE; it never asks for hash.Pipeline::need_to_local_exchange()always adds one (last row of the table in Testing). branch-4.1 and branch-4.2 have no FE local shuffle planner and are not affected.Testing
LocalShuffleNodeCoverageTestandLocalExchangePlannerTest, two new cases each: all four fail without the fix (for example, a serial child gotPASSTHROUGHinstead ofLOCAL_EXECUTION_HASH_SHUFFLE) and pass with it. The 12 fe-core test classes that plan local exchanges or print distributed plans pass (168 tests). Run locally.agg_phase0 and 2 ×disable_streaming_preaggregationson and off × the switch): only 32 with the switch off change. 30 of them now equal the plan with the switch on; the other 2 keep the distribution of aUNION ALL, as the switch asks.4530f761fa1;doris-fe.jarfrom7c4b4462dccwith [fix](fe) Honor enable_local_exchange_before_streaming_agg in the FE local shuffle planner #70177, and for "after" with onlyAggregationNode.classreplaced), switch off,parallel_pipeline_task_num=4,agg_phase=2. TheAggSinkrows setdisable_streaming_preaggregations=true; the distinct rows are aGROUP BYwithout aggregate functions:AggSinkover a nested loop joinAggSinkoverexplodeAggSinkover a one-tablet scan (serial)AggSinkoverUNION ALLexplodenlj: [VNESTED]) and passes with it; its 16 result blocks (4 queries × both planners × the switch on and off) match results computed independently from the inserted rows.nereids_p0/local_shuffle(11 suites),query_p0/join/test_agg_after_nested_loop_join_local_exchangeandnereids_syntax_p0/agg_4_phasepass on the same cluster with the fix.Notes for Reviewers
baseClassRequire()stays in both branches although only a non-serial child reaches them now, since a serial child always breaks the key distribution. It keeps the branches shaped like BE's call to the base class, as [fix](fe) Honor enable_local_exchange_before_streaming_agg in the FE local shuffle planner #70177 does for streaming aggregation.LocalShuffleNodeCoverageTest, [fix](fe) Honor enable_local_exchange_before_streaming_agg in the FE local shuffle planner #70177's five key-breaking children moved intochildrenThatBreakKeyDistribution()and its aggregate builder intobuildLocalPhaseAgg(), so the new cases reuse them; [fix](fe) Honor enable_local_exchange_before_streaming_agg in the FE local shuffle planner #70177's case is otherwise unchanged.PlanNode#requiresShuffleForCorrectnessjavadoc still namedenable_local_exchange_before_aggfor streaming aggregation, which readsenable_local_exchange_before_streaming_aggsince [fix](fe) Honor enable_local_exchange_before_streaming_agg in the FE local shuffle planner #70177.Release note
None
Check List (For Author)
AI Disclosure
.outgenerated by the framework), and this descriptionPlanNodejavadoc, and chose to verify it on a local cluster before opening the PR.