924060929 opened a new pull request, #65835:
URL: https://github.com/apache/doris/pull/65835
### What problem does this PR solve?
Related PR: #65129
Problem Summary:
A regression from #65129, which re-enabled bucket shuffle for set
operations. A UNION
whose branches must be hash distributed for a downstream operator to be
correct returns
wrong results when the union is bucket shuffled and more than one pipeline
task runs.
```sql
SELECT COUNT() OVER (PARTITION BY pk ORDER BY pk)
FROM ( (SELECT pk FROM t21)
UNION ALL
(SELECT t.pk FROM t21 t INNER JOIN t2 ON TRUE) ) subq;
```
`t2` holds 2 rows, so the second branch contributes 2 rows per `pk` and the
first
contributes 1: every row should report 3. It reports a mix of 1, 2 and 3
instead.
The union propagates a generic `requireHash()` to every child. `RequireHash`
accepts
`BUCKET_HASH_SHUFFLE`, `LOCAL_EXECUTION_HASH_SHUFFLE` and
`GLOBAL_EXECUTION_HASH_SHUFFLE` interchangeably, and the satisfy check in
`PlanNode.enforceRequire` runs per child, so the children can settle on
different hash
functions while each still reports the requirement as met. The branch
arriving on a
bucket-shuffle exchange reports `BUCKET_HASH_SHUFFLE`, is accepted as-is and
keeps its
bucket -> instance placement; the branch that provides no hash gets a
freshly inserted
`LOCAL_EXECUTION_HASH_SHUFFLE`:
```
6:VUNION
|----14:VLOCAL-EXCHANGE type: LOCAL_EXECUTION_HASH_SHUFFLE <-
repartitioned by execution hash
| 5:VNESTED LOOP JOIN
| 4:VOlapScanNode (POOLING-SCAN)
1:VEXCHANGE (BUCKET_SHFFULE_HASH_PARTITIONED) <- no local
exchange, keeps bucket placement
```
`bucket(pk) -> owning instance` and `exec_hash(pk) % task_num` are unrelated
mappings, so
the two branches of a given key only meet in the same local task by
coincidence. The
analytic window above then sees a split partition and counts per task.
The requirement degrades to the generic flavour along `AnalyticEvalNode`
(asks nothing of
its child) -> `SortNode` (analytic sort, turns that into
`autoRequireHash()`, and
`NoRequire.autoRequireHash()` is the generic `RequireHash`) -> `UnionNode`
(forwards it
as-is).
A generic hash requirement is sound for a single-input consumer, which only
needs *some*
hash placement. It is not sound for a multi-input one, where the flavours
must also agree
*across* children — something `RequireHash` cannot express. Before #65129
all union
children entered through the same global hash exchange and trivially agreed,
so the hole
was unreachable; keeping the largest child on its own bucket distribution
made
heterogeneous placements possible for the first time.
Fix: require `BUCKET_HASH_SHUFFLE` from every child when the union is
colocate or bucket
shuffle, mirroring the intersect / except branch directly below and
`HashJoinNode`. This
is also the behaviour `PhysicalPlanTranslator` already documents where it
marks the set
operation `BUCKET_SHUFFLE` ("Both routes converge to the same bucket-hash
local exchange
requirement in `SetOperationNode.enforceAndDeriveLocalExchange`") — that
only held for
intersect / except, and the union branch never consulted `isBucketShuffle()`.
A child can only be bucket placed when its distribution is
`STORAGE_BUCKETED`, and that is
exactly what makes the translator mark the set operation `BUCKET_SHUFFLE`
(same
`enableLocalShufflePlanner` gate), so the condition leaves no bucket-placed
child behind.
Union branches that are not bucket aligned keep the previous behaviour: they
all enter
through the same global hash exchange and are therefore already consistent.
Verified on a 4 BE cluster: the reproduction goes from `1x18, 2x36, 3x9` to
`3x63`, and a
second shape from `1x6, 5x30, 6x6` to `6x42`.
### Release note
Fix wrong results when a bucket-shuffled UNION feeds a window function that
partitions by
the union key.
### Check List (For Author)
- Test
- [x] Regression test
- [x] Unit Test
The new `bucket_shuffle_union_analytic_partition` case was checked both
ways: it fails
on the unpatched FE (only that tag; every other case in the suite still
passes) and
passes after the fix. Regenerating the whole `.out` with the fix applied
changes nothing
except the new case, so the plan shapes and results added by #65129 are
untouched.
`LocalShuffleNodeCoverageTest` (17) and `LocalExchangePlannerTest` (28)
pass.
- Behavior changed:
- [x] No.
- Does this need documentation?
- [x] No.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
--
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]
