github-actions[bot] commented on code in PR #65129:
URL: https://github.com/apache/doris/pull/65129#discussion_r3567809685
##########
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:
This still treats any storage-bucketed child as proof that this set
operation chose the bucket-shuffle alternative. A reduced reachable shape is an
outer `PhysicalUnion` using its cheap `ANY` request, with one child being a
nested `INTERSECT` that outputs `STORAGE_BUCKETED` under its own all-column
hash request and another child being realigned to `EXECUTION_ANY`.
`ChildOutputPropertyDeriver` correctly returns a non-specific property for the
outer union because its children are not uniformly bucket-aligned, but this
translator loop ignores that proof and marks the legacy `SetOperationNode`
`BUCKET_SHUFFLE` from the nested child's property alone. That marker is
consumed by `PlanFragment.hasBucketShuffleNode()`, bucket assignment, and
non-`INTERSECT` receiver fill-up, so it is not just an explain-label issue.
Please only set the marker when the current set operation itself has proved
bucket-shuffle alignment, for example by mirroring the all-children
same-layout/same-output-position
proof or by tracking that this set operation's regulator branch inserted the
storage-bucketed exchanges.
##########
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:
This change newly preserves `LOCAL_EXECUTION_HASH_SHUFFLE` as a specific
hash requirement, but the added set-operation regression does not exercise that
path: `explain shape plan` runs before `AddLocalExchange`, and nothing here
forces the bucket-to-local-hash upgrade that creates
`requireSpecific(LOCAL_EXECUTION_HASH_SHUFFLE)`. The existing
`LocalShuffleNodeCoverageTest` union coverage also uses generic
`requireHash()`, so reverting this line back to the old global/bucket-only
check would not be caught. Please add a targeted local-exchange unit test for
`parentRequire.requireSpecific(LOCAL_EXECUTION_HASH_SHUFFLE)` through a
union/pass-through node, or an `EXPLAIN DISTRIBUTED PLAN` regression that
deterministically forces the bucket-to-local upgrade.
##########
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"
Review Comment:
This assertion only proves that the union line itself is not
`[bucketShuffle]`; it does not prove the parent hash request was pushed through
`PhysicalUnion`. If the `createHashRequestAccordingToParent(...)` path
regressed or was removed, the optimizer could still keep an unbucketed union
and insert one `PhysicalDistribute[DistributionSpecHash]` above it for the
shuffle join, so this check and the ordered result would still pass. Please
strengthen this to assert the intended shape, e.g. that the `PhysicalUnion`
children are `PhysicalDistribute[DistributionSpecHash]` under
`enable_local_shuffle_planner=false`, or convert this case to a golden
shape/result check.
--
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]
