924060929 commented on code in PR #65129:
URL: https://github.com/apache/doris/pull/65129#discussion_r3567721015
##########
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:
Fixed — took the "return a non-specific property" option. When the
bucket-shuffle basic child is on the right (`distributeToChildIndex > 0`) and
the set operation is not a `UNION`, `visitPhysicalSetOperation` now returns
`createAnyFromHash(basicChildHash)` instead of a specific bucketed property, so
a parent hash consumer inserts a real realignment exchange rather than
co-locating this set op against an execution-hash sibling.
Added regression `intersect_right_basic_parent_hash`: two `INTERSECT`s whose
right child (made larger via injected stats) is selected as the bucket-shuffle
basic child, joined by a parent hash consumer. Before the fix this reproduced
`Can not find tablet ... in the bucket: 0`; after it the query returns the
correct result, and all existing `.out` shapes are unchanged.
##########
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:
Added `bucket_shuffle_union_fill_up`, a bucket-shuffle `UNION ALL` whose
correctness depends on missing-bucket fill-up:
- the basic/anchor child is a direct bucketed scan pruned by `id IN (0, 2,
4, 6)` so it scans only 4/10 buckets, forced to be the anchor via huge injected
stats;
- the other child has a mismatched bucket count (11) so it cannot be the
natural anchor and is bucket-shuffled onto the anchor's layout, producing rows
for the 6 unscanned buckets;
- a bucket-shuffle join above (probe table with a third bucket count, 7)
supplies the parent hash request and stays in its own fragment, so no
full-bucket scan is co-located into the union fragment to mask the pruning.
On a 4-BE cluster I verified the union fragment's anchor scan is
`tablets=4/10` while `fillUpInstances()` creates receiver instances for all 10
buckets; the 16 ids that live only in the missing buckets appear in the result
and would be lost if the `BUCKET_SHUFFLE` marker or `fillUpInstances()` were
dropped. The case asserts `PhysicalUnion[bucketShuffle]` in the shape and
checks an 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:
Added `plain_intersect_when_local_shuffle_planner_off`: with
`enable_local_shuffle_planner=false`, a plain `INTERSECT` (no parent hash
request, so it goes through `visitPhysicalSetOperation()` and the `REQUIRE` vs
`EXECUTION_BUCKETED` choice directly, not the `visitPhysicalUnion()` /
`createHashRequestAccordingToParent(...)` downgrade path) asserts the shape
contains no `bucketShuffle` and preserves the result via `order_qt`.
--
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]
