924060929 opened a new pull request, #66295:
URL: https://github.com/apache/doris/pull/66295
[fix](local shuffle) align bucket-shuffle union branches by the storage
bucket function
Related PR: #65129
Problem Summary:
A `UNION ALL` feeding a window function returned a duplicated
`row_number()` for the same partition key:
SELECT k, v, ROW_NUMBER() OVER (PARTITION BY k ORDER BY v, src) rn
FROM (SELECT k, v, 1 src FROM t_left
UNION ALL
SELECT k, v, 2 src FROM t_right) s
WHERE k IN (2, 4) ORDER BY k, rn;
-- expected -- actual with enable_local_shuffle_planner=true
2 | 20 | 1 2 | 20 | 1
2 | 20 | 2 2 | 20 | 1
4 | 40 | 1 4 | 40 | 1
4 | 40 | 2 4 | 40 | 1
The two duplicated rows are exactly one row per union branch, i.e. the
branches of the same key are landing in different pipeline tasks and the
analytic sink numbers each task from 1.
`SetOperationNode.enforceAndDeriveLocalExchange` handled the colocate /
bucket-shuffle mode only in the intersect / except branch. For a union it
always asked each child for the generic hash requirement, which
`GLOBAL_EXECUTION_HASH_SHUFFLE`, `LOCAL_EXECUTION_HASH_SHUFFLE` and
`BUCKET_HASH_SHUFFLE` all satisfy — it expresses "I need hash-partitioned
input", not "I need the same hash function as my sibling branches". So in a
bucket-shuffle union:
- the branch that scans its own buckets is serial under a pooling scan,
claims NOOP, and gets an `LE(LOCAL_EXECUTION_HASH_SHUFFLE)`
(execution hash modulo the local task count);
- the branch arriving through a bucket-shuffle exchange claims
`BUCKET_HASH_SHUFFLE`, already satisfies the requirement, and keeps the
storage bucket to instance mapping with no local exchange at all.
Two different placement functions feed the same union, so a key present in
both branches sits in two tasks. The failing plan (`explain distributed
plan`, which is where FE-planned local exchanges show up):
3:VUNION
|----2:VEXCHANGE <- no local exchange,
placement
| comes from the sender's
|
BUCKET_SHFFULE_HASH_PARTITIONED
10:VLOCAL-EXCHANGE
| type: LOCAL_EXECUTION_HASH_SHUFFLE <- execution hash
9:VLOCAL-EXCHANGE
| type: PASSTHROUGH
0:VOlapScanNode POOLING-SCAN
The shape only became reachable when #65129 re-enabled the bucket-shuffle
alternative for set operations (union included) and marked every set
operation with a storage-bucketed child as `DistributionMode.BUCKET_SHUFFLE`;
before that both branches entered a fresh fragment through global hash
exchanges, i.e. one consistent placement.
Fix: when a union is propagating a hash requirement downward and is itself
colocate / bucket-shuffle, require `BUCKET_HASH_SHUFFLE` from every branch
and report `BUCKET_HASH_SHUFFLE` upward — the same pattern `HashJoinNode` and
the intersect / except branch already use. Unlike intersect / except, a union
does not need its branches aligned for its own semantics, so the bucket
requirement is gated on `hasShuffleForCorrectnessAncestor`: with no
downstream correctness consumer nothing has to move.
The existing `bucket_shuffle_union_with_all_column` case did not catch this
because it puts the union under a `join[shuffle]`, whose requirement is a
specific GLOBAL hash that neither branch satisfies, so both get re-aligned to
the same placement. Only consumers asking for the generic hash requirement
(the analytic sort, and the aggregation paths that use `requireHash()` /
`autoRequireHash()`) expose the mixed placement.
Verified end to end on a cluster with the BE binary held fixed and only the
FE swapped: wrong result before, correct after, and the same bucket-shuffle
union under a `group by` consumer is correct too. The new regression case was
checked against the unmodified FE and fails there with
`duplicated row_number 1 in window partition 2`.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
https://claude.ai/code/session_012ZmtXicM2eb3Yrn4b4keR5
--
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]
