morrySnow commented on code in PR #65129:
URL: https://github.com/apache/doris/pull/65129#discussion_r3540998168
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java:
##########
@@ -2497,14 +2496,38 @@ 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 set-op bucket shuffle was chosen by
+ // ChildrenPropertiesRegulator, which only does so under the FE
local-shuffle planner;
+ // the gate here keeps the two sites explicitly consistent. 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.
+ ConnectContext setOperationConnectContext =
context.getConnectContext();
+ if (setOperationConnectContext != null
+ &&
setOperationConnectContext.getSessionVariable().isEnableLocalShuffle()
+ &&
setOperationConnectContext.getSessionVariable().isEnableLocalShufflePlanner()
+ &&
SessionVariable.canUseNereidsDistributePlanner(setOperationConnectContext)) {
Review Comment:
same in `ChildrenPropertiesRegulator`
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java:
##########
@@ -651,83 +652,115 @@ public List<List<PhysicalProperties>>
visitPhysicalSetOperation(PhysicalSetOpera
} else if (requiredDistributionSpec instanceof DistributionSpecHash) {
// TODO: should use the most common hash spec as basic
DistributionSpecHash basic = (DistributionSpecHash)
requiredDistributionSpec;
- // TODO: open comment when support `enable_local_shuffle_planner`
- // int bucketShuffleBasicIndex = -1;
- // double basicRowCount = -1;
-
- // find the bucket shuffle basic index
- // try {
- // ImmutableSet<ShuffleType> supportBucketShuffleTypes =
ImmutableSet.of(
- // ShuffleType.NATURAL,
- // ShuffleType.STORAGE_BUCKETED
- // );
- // for (int i = 0; i < originChildrenProperties.size(); i++) {
- // PhysicalProperties originChildrenProperty =
originChildrenProperties.get(i);
- // DistributionSpec childDistribution =
originChildrenProperty.getDistributionSpec();
- // if (childDistribution instanceof DistributionSpecHash
- // && supportBucketShuffleTypes.contains(
- // ((DistributionSpecHash)
childDistribution).getShuffleType())
- // &&
!(isBucketShuffleDownGrade(setOperation.child(i)))) {
- // Statistics stats = setOperation.child(i).getStats();
- // double rowCount = stats.getRowCount();
- // if (rowCount > basicRowCount) {
- // basicRowCount = rowCount;
- // bucketShuffleBasicIndex = i;
- // }
- // }
- // }
- // } catch (Throwable t) {
- // // catch stats exception
- // LOG.warn("Can not find the most (bucket num, rowCount): " +
t, t);
- // bucketShuffleBasicIndex = -1;
- // }
-
- // use bucket shuffle
- // if (bucketShuffleBasicIndex >= 0) {
- // DistributionSpecHash notShuffleSideRequire
- // = (DistributionSpecHash)
requiredProperties.get(bucketShuffleBasicIndex)
- // .getDistributionSpec();
- //
- // DistributionSpecHash notNeedShuffleOutput
- // = (DistributionSpecHash)
originChildrenProperties.get(bucketShuffleBasicIndex)
- // .getDistributionSpec();
- //
- // for (int i = 0; i < originChildrenProperties.size(); i++) {
- // DistributionSpecHash current
- // = (DistributionSpecHash)
originChildrenProperties.get(i).getDistributionSpec();
- // if (i == bucketShuffleBasicIndex) {
- // continue;
- // }
- //
- // DistributionSpecHash currentRequire
- // = (DistributionSpecHash)
requiredProperties.get(i).getDistributionSpec();
- //
- // PhysicalProperties target = calAnotherSideRequired(
- // ShuffleType.STORAGE_BUCKETED,
- // notNeedShuffleOutput, current,
- // notShuffleSideRequire,
- // currentRequire);
- // updateChildEnforceAndCost(i, target);
- // }
- // setOperation.setMutableState(
- // PhysicalSetOperation.DISTRIBUTE_TO_CHILD_INDEX,
bucketShuffleBasicIndex);
- // use partitioned shuffle
- // } else {
- for (int i = 0; i < originChildrenProperties.size(); i++) {
- DistributionSpecHash current
- = (DistributionSpecHash)
originChildrenProperties.get(i).getDistributionSpec();
- if (current.getShuffleType() != ShuffleType.EXECUTION_BUCKETED
- || !bothSideShuffleKeysAreSameOrder(basic, current,
- (DistributionSpecHash)
requiredProperties.get(0).getDistributionSpec(),
- (DistributionSpecHash)
requiredProperties.get(i).getDistributionSpec())) {
- PhysicalProperties target = calAnotherSideRequired(
- ShuffleType.EXECUTION_BUCKETED, basic, current,
- (DistributionSpecHash)
requiredProperties.get(0).getDistributionSpec(),
- (DistributionSpecHash)
requiredProperties.get(i).getDistributionSpec());
+ int bucketShuffleBasicIndex = -1;
+ double basicRowCount = -1;
+
+ // Bucket shuffle for set operation is only valid when the FE
plans the local
+ // shuffle: with the BE-side local-shuffle planner the backend
cannot infer the
+ // correct local shuffle type for the set sink/probe and computes
wrong results.
+ // It also requires the nereids distribute planner: the legacy
coordinator only
+ // supports bucket-shuffle-partitioned sinks whose dest fragment
contains a bucket
+ // shuffle join, so a bucket-shuffle set operation fragment cannot
be scheduled there.
+ // Otherwise, keep bucketShuffleBasicIndex = -1 and fall back to
the
+ // execution-bucketed (partitioned) shuffle below.
+ ConnectContext setOperationContext = ConnectContext.get();
+ boolean enableLocalShufflePlanner = setOperationContext != null
+ &&
setOperationContext.getSessionVariable().isEnableLocalShuffle()
+ &&
setOperationContext.getSessionVariable().isEnableLocalShufflePlanner()
+ &&
SessionVariable.canUseNereidsDistributePlanner(setOperationContext);
+
+ // find the bucket shuffle basic index: the largest natural /
storage-bucketed child
+ // keeps its bucket distribution, every other child is
bucket-shuffled to it.
+ // isBucketShuffleDownGrade reuses the join-side heuristics on
purpose, including
+ // the enable_bucket_shuffle_join switch and
bucket_shuffle_downgrade_ratio: bucket
+ // shuffle for set operation belongs to the same optimization
family as bucket
+ // shuffle join, so the join switches govern both instead of
introducing a separate
+ // session variable.
+ if (enableLocalShufflePlanner) {
Review Comment:
maybe `if (enableLocalShufflePlanner)` can be removed? because we already
generate right require in `RequestPropertyDeriver`
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java:
##########
@@ -447,53 +443,136 @@ 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);
- // }
+ // When set-op bucket shuffle is chosen, the set operation keeps the
basic child's bucket
+ // distribution as its own output so the bucket distribution
propagates upward instead of
+ // being flattened to execution-bucketed. The basic child is
recomputed from the children
+ // distributions instead of being carried as mutable planner state:
mutable state does not
+ // survive the with-copies in chooseBestPlan() and the
+ // RecomputePhysicalPropertiesPostProcessor re-derivation, while the
recomputation below is
+ // deterministic on any copy of the plan.
//
- // 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()
- // )
- // );
- // }
- // }
- // }
+ // The bucket-shuffle signature is structural: every child is
hash-distributed by NATURAL
+ // or STORAGE_BUCKETED with the same storage layout (table / index /
partitions — an
+ // enforced bucket-shuffle child carries the basic child's layout, see
+ // ChildrenPropertiesRegulator), and at least one child is
STORAGE_BUCKETED. The layout
+ // equality rejects an un-enforced mix such as a NATURAL scan plus a
lower bucket-shuffle
+ // plan distributed by another table's buckets (reachable through the
ANY child request of
+ // union), and requiring one STORAGE_BUCKETED child rejects children
that merely keep
+ // their own NATURAL distributions without any set-op enforcement. The
basic child keeps
+ // its NATURAL distribution, so prefer the first NATURAL child and
fall back to the first
+ // STORAGE_BUCKETED one (then every child shares the same layout, so
the claim does not
+ // depend on which child the regulator actually picked).
+ int distributeToChildIndex = -1;
+ int firstNaturalIndex = -1;
+ int firstStorageBucketedIndex = -1;
+ boolean allChildrenBucketAligned = true;
+ DistributionSpecHash firstChildHash = null;
+ for (int i = 0; i < childrenDistribution.size(); i++) {
+ DistributionSpec childDistributionSpec =
childrenDistribution.get(i);
+ if (!(childDistributionSpec instanceof DistributionSpecHash)) {
+ allChildrenBucketAligned = false;
+ break;
+ }
+ DistributionSpecHash childHash = (DistributionSpecHash)
childDistributionSpec;
+ if (firstChildHash == null) {
+ // an unknown layout cannot prove alignment
+ if (childHash.getTableId() < 0) {
+ allChildrenBucketAligned = false;
+ break;
+ }
+ firstChildHash = childHash;
+ } else if (childHash.getTableId() != firstChildHash.getTableId()
+ || childHash.getSelectedIndexId() !=
firstChildHash.getSelectedIndexId()
+ ||
!childHash.getPartitionIds().equals(firstChildHash.getPartitionIds())) {
+ allChildrenBucketAligned = false;
+ break;
+ }
+ ShuffleType childShuffleType = childHash.getShuffleType();
+ if (childShuffleType == ShuffleType.NATURAL) {
+ if (firstNaturalIndex < 0) {
+ firstNaturalIndex = i;
+ }
+ } else if (childShuffleType == ShuffleType.STORAGE_BUCKETED) {
+ if (firstStorageBucketedIndex < 0) {
+ firstStorageBucketedIndex = i;
+ }
+ } else {
+ allChildrenBucketAligned = false;
+ break;
+ }
+ }
+ if (allChildrenBucketAligned && firstStorageBucketedIndex >= 0) {
+ distributeToChildIndex = firstNaturalIndex >= 0 ?
firstNaturalIndex : firstStorageBucketedIndex;
+ }
+ if (distributeToChildIndex >= 0) {
+ DistributionSpecHash childDistribution
+ = (DistributionSpecHash)
childrenDistribution.get(distributeToChildIndex);
+ // A shared storage layout alone does not prove alignment: two
children may be
+ // bucketed by columns that feed different set-output positions
(e.g. one child
+ // bucketed by the column feeding output k, another by the column
feeding output v
+ // of the same table layout). Map every child's hash columns to
set-output
+ // positions and require all children to land on the same
positions in the same
+ // order; otherwise fall through to the generic derivation below.
+ List<Integer> outputPositions = null;
+ boolean allChildrenSamePositions = true;
+ for (int i = 0; i < childrenDistribution.size() &&
allChildrenSamePositions; i++) {
+ DistributionSpecHash childHash = (DistributionSpecHash)
childrenDistribution.get(i);
+ List<SlotReference> childOutput =
setOperation.getRegularChildrenOutputs().get(i);
+ Map<ExprId, Integer> idToOutputIndex = new LinkedHashMap<>();
+ for (int j = 0; j < childOutput.size(); j++) {
+ idToOutputIndex.put(childOutput.get(j).getExprId(), j);
+ }
+ List<Integer> positions = new ArrayList<>();
+ for (ExprId shuffledColumnId :
childHash.getOrderedShuffledColumns()) {
+ Integer index = idToOutputIndex.get(shuffledColumnId);
+ if (index == null) {
+ allChildrenSamePositions = false;
+ break;
+ }
+ positions.add(index);
+ }
+ if (!allChildrenSamePositions) {
+ break;
+ }
+ if (outputPositions == null) {
+ outputPositions = positions;
+ } else if (!outputPositions.equals(positions)) {
+ allChildrenSamePositions = false;
+ }
+ }
+ if (allChildrenSamePositions && outputPositions != null &&
!outputPositions.isEmpty()) {
+ List<ExprId> setOperationDistributeColumnIds = new
ArrayList<>();
+ for (int outputPosition : outputPositions) {
+ setOperationDistributeColumnIds.add(
+
setOperation.getOutput().get(outputPosition).getExprId());
+ }
+ boolean isUnion = setOperation instanceof Union;
+ // When no NATURAL child remains (the basic child itself was
storage-bucketed),
+ // the index the regulator actually picked is unknown, so
intersect / except
+ // conservatively downgrade like the shuffle-to-right case.
Union is insensitive
+ // to the picked index because every child shares the same
layout.
+ boolean shuffleToRight = distributeToChildIndex > 0 ||
firstNaturalIndex < 0;
Review Comment:
why `|| firstNaturalIndex < 0` means shuffle to right?
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java:
##########
@@ -447,53 +443,136 @@ 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);
- // }
+ // When set-op bucket shuffle is chosen, the set operation keeps the
basic child's bucket
+ // distribution as its own output so the bucket distribution
propagates upward instead of
+ // being flattened to execution-bucketed. The basic child is
recomputed from the children
+ // distributions instead of being carried as mutable planner state:
mutable state does not
+ // survive the with-copies in chooseBestPlan() and the
+ // RecomputePhysicalPropertiesPostProcessor re-derivation, while the
recomputation below is
+ // deterministic on any copy of the plan.
//
- // 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()
- // )
- // );
- // }
- // }
- // }
+ // The bucket-shuffle signature is structural: every child is
hash-distributed by NATURAL
+ // or STORAGE_BUCKETED with the same storage layout (table / index /
partitions — an
+ // enforced bucket-shuffle child carries the basic child's layout, see
+ // ChildrenPropertiesRegulator), and at least one child is
STORAGE_BUCKETED. The layout
+ // equality rejects an un-enforced mix such as a NATURAL scan plus a
lower bucket-shuffle
+ // plan distributed by another table's buckets (reachable through the
ANY child request of
+ // union), and requiring one STORAGE_BUCKETED child rejects children
that merely keep
+ // their own NATURAL distributions without any set-op enforcement. The
basic child keeps
+ // its NATURAL distribution, so prefer the first NATURAL child and
fall back to the first
+ // STORAGE_BUCKETED one (then every child shares the same layout, so
the claim does not
+ // depend on which child the regulator actually picked).
+ int distributeToChildIndex = -1;
+ int firstNaturalIndex = -1;
+ int firstStorageBucketedIndex = -1;
+ boolean allChildrenBucketAligned = true;
+ DistributionSpecHash firstChildHash = null;
+ for (int i = 0; i < childrenDistribution.size(); i++) {
+ DistributionSpec childDistributionSpec =
childrenDistribution.get(i);
+ if (!(childDistributionSpec instanceof DistributionSpecHash)) {
+ allChildrenBucketAligned = false;
+ break;
+ }
+ DistributionSpecHash childHash = (DistributionSpecHash)
childDistributionSpec;
+ if (firstChildHash == null) {
+ // an unknown layout cannot prove alignment
+ if (childHash.getTableId() < 0) {
+ allChildrenBucketAligned = false;
+ break;
+ }
+ firstChildHash = childHash;
+ } else if (childHash.getTableId() != firstChildHash.getTableId()
+ || childHash.getSelectedIndexId() !=
firstChildHash.getSelectedIndexId()
+ ||
!childHash.getPartitionIds().equals(firstChildHash.getPartitionIds())) {
+ allChildrenBucketAligned = false;
+ break;
+ }
+ ShuffleType childShuffleType = childHash.getShuffleType();
+ if (childShuffleType == ShuffleType.NATURAL) {
+ if (firstNaturalIndex < 0) {
+ firstNaturalIndex = i;
+ }
+ } else if (childShuffleType == ShuffleType.STORAGE_BUCKETED) {
+ if (firstStorageBucketedIndex < 0) {
+ firstStorageBucketedIndex = i;
+ }
+ } else {
+ allChildrenBucketAligned = false;
+ break;
+ }
+ }
+ if (allChildrenBucketAligned && firstStorageBucketedIndex >= 0) {
+ distributeToChildIndex = firstNaturalIndex >= 0 ?
firstNaturalIndex : firstStorageBucketedIndex;
+ }
+ if (distributeToChildIndex >= 0) {
+ DistributionSpecHash childDistribution
+ = (DistributionSpecHash)
childrenDistribution.get(distributeToChildIndex);
+ // A shared storage layout alone does not prove alignment: two
children may be
+ // bucketed by columns that feed different set-output positions
(e.g. one child
Review Comment:
感觉这里整理的实现复杂了。在计算 child output 的时候，已经经过了 regulator。所以当前的孩子输出肯定是合法的。那么只有以下几种情况：
1. 有一个或多个 natural：则按照natural（多个类似 colocate join）输出
2. 没有natural 但是有storage bucketed，则输出storage bucketed
3. 都是 execution bucketed，输出execution hash
4. 都是random，输出random
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/RequestPropertyDeriver.java:
##########
@@ -339,14 +340,22 @@ public Void
visitPhysicalSetOperation(PhysicalSetOperation setOperation, PlanCon
// shuffle all column
// TODO: for wide table, may be we should add a upper limit of
shuffle columns
- // TODO: open comment when support `enable_local_shuffle_planner`
and change to REQUIRE
- // intersect/except always need hash distribution, we use REQUIRE
to auto select
- // bucket shuffle or execution shuffle
+ // intersect/except always need hash distribution. Auto-selecting
bucket shuffle
+ // (ShuffleType.REQUIRE) for set operation is only valid when the
FE plans the local
+ // shuffle: with the BE-side local-shuffle planner the backend
cannot infer the
+ // correct local shuffle type for the set sink/probe and computes
wrong results.
+ // It also requires the nereids distribute planner: the legacy
coordinator only
+ // supports bucket-shuffle-partitioned sinks whose dest fragment
contains a bucket
+ // shuffle join. Fall back to EXECUTION_BUCKETED otherwise.
+ ShuffleType setOperationShuffleType = connectContext != null
Review Comment:
if local shuffle is disable, we can still do bucket shuffle for set
operation. so the condition should be local shuffle disable or using local
shuffle planner to plan local shuffle
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java:
##########
@@ -447,53 +443,136 @@ 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);
- // }
+ // When set-op bucket shuffle is chosen, the set operation keeps the
basic child's bucket
+ // distribution as its own output so the bucket distribution
propagates upward instead of
+ // being flattened to execution-bucketed. The basic child is
recomputed from the children
+ // distributions instead of being carried as mutable planner state:
mutable state does not
+ // survive the with-copies in chooseBestPlan() and the
+ // RecomputePhysicalPropertiesPostProcessor re-derivation, while the
recomputation below is
+ // deterministic on any copy of the plan.
//
- // 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()
- // )
- // );
- // }
- // }
- // }
+ // The bucket-shuffle signature is structural: every child is
hash-distributed by NATURAL
+ // or STORAGE_BUCKETED with the same storage layout (table / index /
partitions — an
+ // enforced bucket-shuffle child carries the basic child's layout, see
+ // ChildrenPropertiesRegulator), and at least one child is
STORAGE_BUCKETED. The layout
+ // equality rejects an un-enforced mix such as a NATURAL scan plus a
lower bucket-shuffle
+ // plan distributed by another table's buckets (reachable through the
ANY child request of
+ // union), and requiring one STORAGE_BUCKETED child rejects children
that merely keep
+ // their own NATURAL distributions without any set-op enforcement. The
basic child keeps
+ // its NATURAL distribution, so prefer the first NATURAL child and
fall back to the first
+ // STORAGE_BUCKETED one (then every child shares the same layout, so
the claim does not
+ // depend on which child the regulator actually picked).
+ int distributeToChildIndex = -1;
+ int firstNaturalIndex = -1;
+ int firstStorageBucketedIndex = -1;
+ boolean allChildrenBucketAligned = true;
+ DistributionSpecHash firstChildHash = null;
+ for (int i = 0; i < childrenDistribution.size(); i++) {
+ DistributionSpec childDistributionSpec =
childrenDistribution.get(i);
+ if (!(childDistributionSpec instanceof DistributionSpecHash)) {
+ allChildrenBucketAligned = false;
+ break;
+ }
+ DistributionSpecHash childHash = (DistributionSpecHash)
childDistributionSpec;
+ if (firstChildHash == null) {
+ // an unknown layout cannot prove alignment
+ if (childHash.getTableId() < 0) {
+ allChildrenBucketAligned = false;
+ break;
+ }
+ firstChildHash = childHash;
+ } else if (childHash.getTableId() != firstChildHash.getTableId()
+ || childHash.getSelectedIndexId() !=
firstChildHash.getSelectedIndexId()
+ ||
!childHash.getPartitionIds().equals(firstChildHash.getPartitionIds())) {
+ allChildrenBucketAligned = false;
+ break;
+ }
+ ShuffleType childShuffleType = childHash.getShuffleType();
+ if (childShuffleType == ShuffleType.NATURAL) {
+ if (firstNaturalIndex < 0) {
+ firstNaturalIndex = i;
+ }
+ } else if (childShuffleType == ShuffleType.STORAGE_BUCKETED) {
+ if (firstStorageBucketedIndex < 0) {
+ firstStorageBucketedIndex = i;
+ }
+ } else {
+ allChildrenBucketAligned = false;
+ break;
+ }
+ }
+ if (allChildrenBucketAligned && firstStorageBucketedIndex >= 0) {
Review Comment:
why must firstStorageBucketedIndex >= 0 ?
--
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]
