924060929 commented on code in PR #65129:
URL: https://github.com/apache/doris/pull/65129#discussion_r3568004494
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java:
##########
@@ -2496,14 +2497,56 @@ 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;
- // }
- // }
+ // Mark BUCKET_SHUFFLE only when every child is bucket-distributed on
the same storage
+ // layout, i.e. ChildrenPropertiesRegulator actually aligned all
children to one basic
+ // child's buckets: it enforces the non-basic children to the basic
child's storage
+ // layout, and only does so when set-op bucket shuffle is allowed, so
the marker follows
+ // the regulator's decision without re-checking the session variables
here (that would
+ // risk drifting out of sync with the request-derivation gate).
+ //
+ // A single storage-bucketed child is not proof on its own that this
set operation chose
+ // bucket shuffle: under a non-bucket (ANY) request a child can stay
storage-bucketed
+ // independently while the other children are left unaligned, and
marking from that child
+ // alone would drive PlanFragment.hasBucketShuffleNode(), bucket
assignment and non-
+ // INTERSECT receiver fill-up on a set operation whose rows are not
actually bucket-co-
+ // located. Requiring one shared storage layout keeps the marker in
step with
+ // ChildOutputPropertyDeriver, which likewise only claims a bucket
distribution when all
+ // children agree on the layout.
+ //
+ // 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.
+ boolean anyStorageBucketed = false;
+ boolean allChildrenBucketAligned = !setOperation.children().isEmpty();
+ DistributionSpecHash basicLayout = null;
+ for (Plan child : setOperation.children()) {
+ DistributionSpec childSpec = ((PhysicalPlan)
child).getPhysicalProperties().getDistributionSpec();
+ if (!(childSpec instanceof DistributionSpecHash)) {
+ allChildrenBucketAligned = false;
+ break;
+ }
+ DistributionSpecHash childHash = (DistributionSpecHash) childSpec;
+ ShuffleType shuffleType = childHash.getShuffleType();
+ if ((shuffleType != ShuffleType.NATURAL && shuffleType !=
ShuffleType.STORAGE_BUCKETED)
+ || childHash.getTableId() < 0) {
+ allChildrenBucketAligned = false;
+ break;
+ }
+ anyStorageBucketed |= shuffleType == ShuffleType.STORAGE_BUCKETED;
+ if (basicLayout == null) {
+ basicLayout = childHash;
+ } else if (childHash.getTableId() != basicLayout.getTableId()
Review Comment:
Good catch — confirmed and fixed. `shouldFillUpInstances` imported the
Nereids algebra `Intersect`, but the collected nodes are legacy
`org.apache.doris.planner.SetOperationNode`s, so `instanceof Intersect` was
always false and the intersect skip was dead. It now checks `instanceof
IntersectNode` (the legacy planner node).
On a 4-BE cluster I reproduced that, before the fix, a bucket-shuffle
`INTERSECT` with a bucket-pruned anchor did create missing-bucket fill-up
instances; the result stayed correct (an empty anchor bucket intersects to
empty, so the shuffled rows there produce nothing) but the receivers were
wasted work. After the fix the intersect no longer fills up while
`UNION`/`EXCEPT` still do, and results are unchanged.
Added a deterministic unit test
`UnassignedScanBucketOlapTableJobTest#testShouldFillUpInstancesSkipsBucketShuffleIntersectOnly`:
a bucket-shuffle `IntersectNode` → no fill-up, bucket-shuffle `UnionNode` /
`ExceptNode` → fill-up, and a non-bucket-shuffle union → no fill-up. Reverting
the check to the Nereids type makes the intersect assertion fail.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java:
##########
@@ -2496,14 +2497,56 @@ 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;
- // }
- // }
+ // Mark BUCKET_SHUFFLE only when every child is bucket-distributed on
the same storage
+ // layout, i.e. ChildrenPropertiesRegulator actually aligned all
children to one basic
+ // child's buckets: it enforces the non-basic children to the basic
child's storage
+ // layout, and only does so when set-op bucket shuffle is allowed, so
the marker follows
+ // the regulator's decision without re-checking the session variables
here (that would
+ // risk drifting out of sync with the request-derivation gate).
+ //
+ // A single storage-bucketed child is not proof on its own that this
set operation chose
+ // bucket shuffle: under a non-bucket (ANY) request a child can stay
storage-bucketed
+ // independently while the other children are left unaligned, and
marking from that child
+ // alone would drive PlanFragment.hasBucketShuffleNode(), bucket
assignment and non-
+ // INTERSECT receiver fill-up on a set operation whose rows are not
actually bucket-co-
+ // located. Requiring one shared storage layout keeps the marker in
step with
+ // ChildOutputPropertyDeriver, which likewise only claims a bucket
distribution when all
+ // children agree on the layout.
+ //
+ // 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.
+ boolean anyStorageBucketed = false;
+ boolean allChildrenBucketAligned = !setOperation.children().isEmpty();
+ DistributionSpecHash basicLayout = null;
+ for (Plan child : setOperation.children()) {
+ DistributionSpec childSpec = ((PhysicalPlan)
child).getPhysicalProperties().getDistributionSpec();
+ if (!(childSpec instanceof DistributionSpecHash)) {
+ allChildrenBucketAligned = false;
+ break;
+ }
+ DistributionSpecHash childHash = (DistributionSpecHash) childSpec;
+ ShuffleType shuffleType = childHash.getShuffleType();
+ if ((shuffleType != ShuffleType.NATURAL && shuffleType !=
ShuffleType.STORAGE_BUCKETED)
Review Comment:
Made the marker use the deriver's exact proof. Extracted the alignment check
— all children hash NATURAL/STORAGE_BUCKETED on one storage layout **and**
mapping their shuffle columns to the same set-operation output positions — into
a shared `ChildOutputPropertyDeriver.isSetOperationBucketAligned(setOperation,
childDistributions)` (with `mapShuffleColumnsToOutputPositions` now static).
`visitPhysicalSetOperation` (to decide the output property) and
`PhysicalPlanTranslator` (to set the `BUCKET_SHUFFLE` marker) both call it, so
they can never disagree: a same-layout shape whose bucket key feeds output
position `k` in one child and `v` in another fails the shared check, so the
deriver returns `createAnyFromHash(...)` and the marker is not set.
The deriver's own behavior is unchanged — the output-position proof was
already there; I only lifted it into a shared method and made the translator
use it instead of the layout-only check. Verified the full
`bucket_shuffle_set_operation` suite still matches the committed `.out` (no
shape/result delta).
--
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]
