924060929 commented on code in PR #65129:
URL: https://github.com/apache/doris/pull/65129#discussion_r3567882134
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java:
##########
@@ -2496,14 +2496,26 @@ public PlanFragment visitPhysicalSetOperation(
setOperationNode.setColocate(true);
}
- // TODO: open comment when support `enable_local_shuffle_planner`
- // for (Plan child : setOperation.children()) {
- // PhysicalPlan childPhysicalPlan = (PhysicalPlan) child;
- // if
(JoinUtils.isStorageBucketed(childPhysicalPlan.getPhysicalProperties())) {
- //
setOperationNode.setDistributionMode(DistributionMode.BUCKET_SHUFFLE);
- // break;
- // }
- // }
+ // A storage-bucketed child means ChildrenPropertiesRegulator chose
the bucket shuffle
+ // alternative for this set operation (it enforces the other children
to the basic
+ // child's storage layout), which it only does when set-op bucket
shuffle is allowed.
+ // So the marker follows the regulator's decision directly, without
re-checking the
+ // session variables here (that would risk drifting out of sync with
the request
+ // derivation gate). Mark the node BUCKET_SHUFFLE so the set
sink/probe align by bucket
+ // instead of execution-bucketed hash.
+ //
+ // Unlike hash join, BUCKET_SHUFFLE is not exclusive with isColocate
above: for a set
+ // operation isColocate describes the bucket-aligned scheduling of the
fragment (the
+ // basic child scans buckets directly), while BUCKET_SHUFFLE describes
how the other
+ // children arrive (bucket-shuffle exchanges). Both routes converge to
the same
+ // bucket-hash local exchange requirement in
SetOperationNode.enforceAndDeriveLocalExchange.
+ for (Plan child : setOperation.children()) {
+ PhysicalPlan childPhysicalPlan = (PhysicalPlan) child;
+ if
(JoinUtils.isStorageBucketed(childPhysicalPlan.getPhysicalProperties())) {
Review Comment:
Hardened the marker as suggested — it now mirrors the deriver's alignment
proof instead of trusting a single storage-bucketed child. The translator marks
`BUCKET_SHUFFLE` only when **every** child is a hash
`NATURAL`/`STORAGE_BUCKETED` distribution sharing one known storage layout
(same `tableId >= 0` / `selectedIndexId` / `partitionIds`), with at least one
`STORAGE_BUCKETED` child — i.e. exactly when `ChildrenPropertiesRegulator`
aligned all children to one basic child's buckets. A lone storage-bucketed
child with an unaligned / different-layout sibling no longer flips the marker.
On reachability I could not construct a wrong result and it looks
structurally hard to reach: `JoinUtils.isStorageBucketed` is true only for
`STORAGE_BUCKETED` with `tableId >= 0`, which is produced solely as a
regulator-enforced bucket-shuffle exchange (⟹ the set op was aligned). A pure
set-op/scan tree keeps a `NATURAL` basic (its leaves are scans), which the
union `ANY` branch washes to `EXECUTION_ANY`; and a non-aligned multi-layout
set op advertises a non-specific property (the deriver's cross-child guard +
the right-basic fix), so it is re-exchanged to execution hash before reaching
the parent. I tried the outer-`ANY`-union-over-nested-intersect, the
join-output-under-`ANY`-union, and the parent-hash union of two
different-layout intersects — all stay correct. But the per-child check was not
a local proof, so the hardening removes the reliance on that transitive
argument.
Verified zero behavior change: the whole `bucket_shuffle_set_operation`
suite regenerates with the marker guard applied and the only `.out` delta is
the new finding-2 golden shape below; every existing shape and result
(including the bucket-shuffle intersect/except, the fill-up `UNION`, and the
join-as-basic-child cases) is byte-identical.
##########
regression-test/suites/query_p0/set_operations/bucket_shuffle_set_operation.groovy:
##########
@@ -95,6 +95,144 @@ suite("bucket_shuffle_set_operation") {
select id from bucket_shuffle_set_operation2 where id=1
""")
+ // The basic child of a bucket-shuffle set operation can be a join output
instead of a
+ // direct scan. In that shape the local exchange planned for the basic
side must still
+ // partition by the storage bucket function: an execution-hash local
exchange would not
+ // align with the bucket-distributed side and the set operation would
compute wrong results.
+ checkShapeAndResult("bucket_shuffle_join_as_basic_child", """
+ select a.id from bucket_shuffle_set_operation1 a
+ join bucket_shuffle_set_operation2 b on a.id = b.id
+ intersect
+ select id from bucket_shuffle_set_operation3""")
+
+ // a set operation child can itself be a set operation whose output claims
a bucket
+ // distribution; the outer set operation must only treat its children as
bucket-aligned
+ // when they share the same storage layout
+ checkShapeAndResult("bucket_shuffle_nested_set_operation", """
+ select id from bucket_shuffle_set_operation3
+ union all
+ (select a.id from bucket_shuffle_set_operation1 a
+ join bucket_shuffle_set_operation2 b on a.id = b.id
+ intersect
+ select id from bucket_shuffle_set_operation2)""")
+
+ // when local shuffle is disabled entirely, every pipeline runs a single
task per
+ // instance so the bucket alignment holds naturally and bucket shuffle is
still allowed
+ sql "set enable_local_shuffle=false"
+ checkShapeAndResult("bucket_shuffle_when_local_shuffle_off", """
+ select id from bucket_shuffle_set_operation1
+ intersect
+ select id from bucket_shuffle_set_operation2""")
+ sql "set enable_local_shuffle=true"
+
+ // A shuffle join above the union pushes a hash request into the union
+ // (createHashRequestAccordingToParent, the parent-hash request path).
When the FE does not
+ // plan the local shuffle, that request must be downgraded so the union
does not choose
+ // bucket shuffle, while the result stays correct.
+ def unionParentHashSql = """
+ select b.id from (
+ select id from bucket_shuffle_set_operation1
+ union all
+ select id from bucket_shuffle_set_operation2
+ ) u join[shuffle] bucket_shuffle_set_operation3 b on u.id = b.id
+ """
+ sql "set enable_local_shuffle_planner=false"
+ explain {
+ sql "shape plan " + unionParentHashSql
+ check { String e ->
+ def unionIndex = e.indexOf("PhysicalUnion")
+ assertTrue(unionIndex >= 0)
+ // the union must not be a bucket shuffle union when the FE local
shuffle planner is off
+ assertFalse(e.substring(unionIndex,
Review Comment:
Strengthened this case. It now (1) emits a golden shape via
`qt_union_parent_hash_shape_when_local_shuffle_planner_off` and (2) asserts the
union subtree contains exactly two `PhysicalDistribute[DistributionSpecHash]` —
i.e. both `PhysicalUnion` children arrive through a hash distribute, which is
the actual proof that `createHashRequestAccordingToParent(...)` pushed the
parent hash request into the union and downgraded it to execution hash. If that
path regressed and the optimizer instead kept an unbucketed union with a single
distribute above it for the shuffle join, the count would be 1 (or the child
would be a direct scan) and both the golden shape and the assertion would fail.
The ordered result check is kept.
##########
fe/fe-core/src/main/java/org/apache/doris/planner/LocalExchangeNode.java:
##########
@@ -283,8 +283,15 @@ public LocalExchangeType preferType() {
@Override
public LocalExchangeTypeRequire autoRequireHash() {
- if (requireType == LocalExchangeType.GLOBAL_EXECUTION_HASH_SHUFFLE
- || requireType == LocalExchangeType.BUCKET_HASH_SHUFFLE) {
+ // Callers are pass-through operators (union / streaming agg /
sort) that report
+ // resolveExchangeType(requireChild) upward while leaving row
placement to their
+ // children. A specific hash require must therefore be forwarded
as-is: degrading
+ // LOCAL_EXECUTION_HASH_SHUFFLE to the generic RequireHash lets a
bucket-distributed
+ // child satisfy the requirement and keep its bucket placement,
while the operator
+ // still claims LOCAL_EXECUTION_HASH_SHUFFLE to its parent — the
parent (e.g. a
+ // bucket join upgraded to local hash) then skips its re-align
local exchange and
+ // the mixed placements compute wrong results.
+ if (requireType.isHashShuffle()) {
Review Comment:
Added a targeted unit test in `LocalShuffleNodeCoverageTest`
(`testRequireSpecificAutoRequireHashPreservesSpecificHash`) that exercises
`RequireSpecific.autoRequireHash()` directly, independent of
`AddLocalExchange`: for each hash flavour (`LOCAL_EXECUTION_HASH_SHUFFLE`,
`GLOBAL_EXECUTION_HASH_SHUFFLE`, `BUCKET_HASH_SHUFFLE`) it asserts the require
is forwarded unchanged (`assertSame` + `preferType()` equals the input), and
that a non-hash specific require (`PASSTHROUGH`) still relaxes to the generic
`RequireHash` (`preferType()` == `GLOBAL_EXECUTION_HASH_SHUFFLE`). Reverting
this line to the old `GLOBAL_EXECUTION_HASH_SHUFFLE || BUCKET_HASH_SHUFFLE`
check makes the `LOCAL_EXECUTION_HASH_SHUFFLE` iteration return
`RequireHash.INSTANCE`, so `assertSame` (and the `preferType()` equality) fail
— the regression is now caught deterministically.
--
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]
