morningman opened a new pull request, #70296: URL: https://github.com/apache/doris/pull/70296
### 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 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()`. #63366 wrote them after #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` (#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 #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 #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 #70177 does for streaming aggregation. - In `LocalShuffleNodeCoverageTest`, #70177's five key-breaking children moved into `childrenThatBreakKeyDistribution()` and its aggregate builder into `buildLocalPhaseAgg()`, so the new cases reuse them; #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 #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 #70177 was recorded together with the outdated `PlanNode` javadoc, and chose to verify it on a local cluster before opening the PR. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
