morningman opened a new pull request, #70177:
URL: https://github.com/apache/doris/pull/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.
### Release note
With the FE local shuffle planner (the default), a streaming pre-aggregation
no longer adds a local hash exchange on its grouping keys above a child that
keeps the key distribution, such as a UNION ALL or an external table scan. This
matches the BE planner and branch-4.x. Set
`enable_local_exchange_before_streaming_agg=true` to add the exchange.
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [x] Regression test
- [x] Unit Test
- [x] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
New tests: 4 in `LocalShuffleNodeCoverageTest` (3 fail without the fix;
the fourth pins the child check), 2 in `LocalExchangePlannerTest` (1 fails
without the fix), and `testSetOperationUnderAggHasHashShuffle` now turns the
variable on. The 9 FE test classes that plan local exchanges or print
distributed plans pass (141 tests). Each of 6 mutations of the new code fails
only the tests aimed at it: dropping the serial check, the nested loop join
case, the hash join case, the table function / assert num rows case, or the
nested loop join's exception for the serial joins, or ignoring the variable.
The new regression
`nereids_p0/local_shuffle/test_local_exchange_before_streaming_agg` passes on a
local one-BE cluster, and its first plan check fails on the FE without this
fix. The profile comparison in Results was taken on the same cluster.
- Behavior changed:
- [ ] No.
- [x] Yes. <!-- Explain the behavior change -->
With the FE local shuffle planner, a streaming pre-aggregation follows
`enable_local_exchange_before_streaming_agg` as described above, and
`enable_local_exchange_before_agg` no longer affects it. Plans change, results
do not.
- Does this need documentation?
- [x] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR should
merge into -->
--
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]
