github-actions[bot] commented on code in PR #65129:
URL: https://github.com/apache/doris/pull/65129#discussion_r3567635128
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java:
##########
@@ -440,53 +443,85 @@ public PhysicalProperties
visitPhysicalSetOperation(PhysicalSetOperation setOper
return PhysicalProperties.GATHER;
}
- // TODO: open comment when support `enable_local_shuffle_planner`
- // int distributeToChildIndex
- // =
setOperation.<Integer>getMutableState(PhysicalSetOperation.DISTRIBUTE_TO_CHILD_INDEX).orElse(-1);
- // if (distributeToChildIndex >= 0
- // && childrenDistribution.get(distributeToChildIndex)
instanceof DistributionSpecHash) {
- // DistributionSpecHash childDistribution
- // = (DistributionSpecHash)
childrenDistribution.get(distributeToChildIndex);
- // List<SlotReference> childToIndex =
setOperation.getRegularChildrenOutputs().get(distributeToChildIndex);
- // Map<ExprId, Integer> idToOutputIndex = new LinkedHashMap<>();
- // for (int j = 0; j < childToIndex.size(); j++) {
- // idToOutputIndex.put(childToIndex.get(j).getExprId(), j);
- // }
- //
- // List<ExprId> orderedShuffledColumns =
childDistribution.getOrderedShuffledColumns();
- // List<ExprId> setOperationDistributeColumnIds = new
ArrayList<>();
- // for (ExprId tableDistributeColumnId : orderedShuffledColumns) {
- // Integer index =
idToOutputIndex.get(tableDistributeColumnId);
- // if (index == null) {
- // break;
- // }
- //
setOperationDistributeColumnIds.add(setOperation.getOutput().get(index).getExprId());
- // }
- // // check whether the set operation output all distribution
columns of the child
- // if (setOperationDistributeColumnIds.size() ==
orderedShuffledColumns.size()) {
- // boolean isUnion = setOperation instanceof Union;
- // boolean shuffleToRight = distributeToChildIndex > 0;
- // if (!isUnion && shuffleToRight) {
- // return new PhysicalProperties(
- // new DistributionSpecHash(
- // setOperationDistributeColumnIds,
- // ShuffleType.EXECUTION_BUCKETED
- // )
- // );
- // } else {
- // // keep the distribution as the child
- // return new PhysicalProperties(
- // new DistributionSpecHash(
- // setOperationDistributeColumnIds,
- // childDistribution.getShuffleType(),
- // childDistribution.getTableId(),
- // childDistribution.getSelectedIndexId(),
- // childDistribution.getPartitionIds()
- // )
- // );
- // }
- // }
- // }
+ // After ChildrenPropertiesRegulator the children distributions are
legal for the
+ // alternative that produced them. When the set operation is bucketed
(some child is
+ // NATURAL or STORAGE_BUCKETED) the output can only be described as
that specific bucket
+ // distribution if EVERY child agrees on it: they must all be
hash-distributed by the
+ // same storage layout (table / index / partitions) and map their
shuffle columns to the
+ // same set-operation output positions. The regulator's bucket-shuffle
branch enforces
+ // exactly this, but the union's cheap ANY child request does not
align its children, so
+ // two children can carry different layouts or positions there; in
that case the output
+ // is not uniformly bucketed and we return a non-specific property so
a parent operator
+ // re-aligns the data. The basic child is recomputed from the children
distributions
+ // rather than carried as mutable state, which does not survive
chooseBestPlan() and the
+ // RecomputePhysicalPropertiesPostProcessor re-derivation.
+ int firstNaturalIndex = -1;
+ int firstStorageBucketedIndex = -1;
+ for (int i = 0; i < childrenDistribution.size(); i++) {
+ if (!(childrenDistribution.get(i) instanceof
DistributionSpecHash)) {
+ continue;
+ }
+ ShuffleType childShuffleType
+ = ((DistributionSpecHash)
childrenDistribution.get(i)).getShuffleType();
+ if (childShuffleType == ShuffleType.NATURAL && firstNaturalIndex <
0) {
+ firstNaturalIndex = i;
+ } else if (childShuffleType == ShuffleType.STORAGE_BUCKETED &&
firstStorageBucketedIndex < 0) {
+ firstStorageBucketedIndex = i;
+ }
+ }
+ int distributeToChildIndex = firstNaturalIndex >= 0 ?
firstNaturalIndex : firstStorageBucketedIndex;
+ if (distributeToChildIndex >= 0) {
+ DistributionSpecHash basicChildHash
+ = (DistributionSpecHash)
childrenDistribution.get(distributeToChildIndex);
+ List<Integer> basicOutputPositions
+ = mapShuffleColumnsToOutputPositions(setOperation,
basicChildHash, distributeToChildIndex);
+ boolean allChildrenAligned = basicOutputPositions != null &&
!basicOutputPositions.isEmpty()
+ && basicChildHash.getTableId() >= 0;
+ for (int i = 0; allChildrenAligned && i <
childrenDistribution.size(); i++) {
+ if (!(childrenDistribution.get(i) instanceof
DistributionSpecHash)) {
+ allChildrenAligned = false;
+ break;
+ }
+ DistributionSpecHash childHash = (DistributionSpecHash)
childrenDistribution.get(i);
+ ShuffleType childShuffleType = childHash.getShuffleType();
+ if (childShuffleType != ShuffleType.NATURAL &&
childShuffleType != ShuffleType.STORAGE_BUCKETED) {
+ allChildrenAligned = false;
+ break;
+ }
+ // same storage layout (a NATURAL basic child and its enforced
STORAGE_BUCKETED
+ // siblings all carry the basic child's layout, so they share
the bucket function)
+ if (childHash.getTableId() != basicChildHash.getTableId()
+ || childHash.getSelectedIndexId() !=
basicChildHash.getSelectedIndexId()
+ ||
!childHash.getPartitionIds().equals(basicChildHash.getPartitionIds())) {
+ allChildrenAligned = false;
+ break;
+ }
+ // same set-operation output positions
+ List<Integer> positions =
mapShuffleColumnsToOutputPositions(setOperation, childHash, i);
+ if (positions == null ||
!positions.equals(basicOutputPositions)) {
+ allChildrenAligned = false;
+ break;
+ }
+ }
+ if (allChildrenAligned) {
+ List<ExprId> setOperationDistributeColumnIds = new
ArrayList<>();
+ for (int outputPosition : basicOutputPositions) {
+
setOperationDistributeColumnIds.add(setOperation.getOutput().get(outputPosition).getExprId());
+ }
+ boolean isUnion = setOperation instanceof Union;
Review Comment:
This downgrade is still unsafe when the bucket-shuffle basic child is on the
right. The regulator physically redistributes the other children with
`ShuffleType.STORAGE_BUCKETED` using the basic child's storage layout, and the
translator sends that exchange as `BUCKET_SHFFULE_HASH_PARTITIONED`. But this
branch then advertises the set-op output as `EXECUTION_BUCKETED` whenever
`distributeToChildIndex > 0`. A parent shuffle join/aggregate can accept that
as a normal execution-hash child and skip the realignment exchange against
another execution-hash side, even though this set-op's rows are placed by the
storage bucket function. Please either preserve the actual storage-bucketed
layout here or return a non-specific property so parent consumers insert a real
exchange; add a regression with a parent hash consumer above an
INTERSECT/EXCEPT whose right child is selected as the bucket-shuffle basic
child.
##########
regression-test/suites/query_p0/set_operations/bucket_shuffle_set_operation.groovy:
##########
@@ -95,6 +95,61 @@ 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)""")
Review Comment:
This still leaves the downstream receiver-fill-up contract untested for the
new `BUCKET_SHUFFLE` marker. The scheduler only creates missing bucket
receivers for non-`INTERSECT` set operations when
`SetOperationNode.isBucketShuffle()` is true, but the new cases do not put a
bucket-shuffle `UNION` or `EXCEPT` in the shape that needs that path: the
direct/basic scan side has some buckets pruned while the other side still
produces rows for those missing buckets. The existing `bucket_shuffle_except_1`
shape filters both sides to `id = 1`, and `bucket_shuffle_except_2` leaves the
direct left side unfiltered, so losing the marker or breaking
`fillUpInstances()` for non-intersect set ops would still pass. Please add a
bucket-shuffle `UNION` or `EXCEPT` case that requires missing-bucket receiver
fill-up for correctness, with a shape assertion and ordered result.
##########
regression-test/suites/query_p0/set_operations/bucket_shuffle_set_operation.groovy:
##########
@@ -95,6 +95,61 @@ 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
Review Comment:
This disabled-planner case exercises the `PhysicalUnion` parent-hash path,
but it still misses the ordinary set-operation request path. Plain
`INTERSECT`/`EXCEPT` without a parent hash request goes through
`visitPhysicalSetOperation()` and directly chooses `ShuffleType.REQUIRE` versus
`EXECUTION_BUCKETED`; the union test below only verifies the
`createHashRequestAccordingToParent(...)` downgrade used by
`visitPhysicalUnion()`. Please add a small disabled
`enable_local_shuffle_planner` or disabled Nereids distribute planner case for
a plain `INTERSECT`/`EXCEPT` that asserts no `[bucketShuffle]` shape and
preserves the result.
--
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]
