This is an automated email from the ASF dual-hosted git repository.
924060929 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 589fd3e9a1e [opt](local shuffle) support bucket shuffle for set
operation (#65129)
589fd3e9a1e is described below
commit 589fd3e9a1e8bd05d2baba5533e20f164db9854c
Author: 924060929 <[email protected]>
AuthorDate: Fri Jul 17 11:19:10 2026 +0800
[opt](local shuffle) support bucket shuffle for set operation (#65129)
Related PR: #59006, #60823
Problem Summary:
Re-enable bucket shuffle for set operation (union / intersect / except).
This optimization was introduced by #59006 and later disabled by #60823,
because at that time the backend could not plan the correct local
shuffle type for set operation, which produced wrong results. Now that
planning local shuffle in the frontend is supported, the frontend plans
the correct local shuffle type, so it is safe to re-enable this feature.
When `enable_local_shuffle_planner` is enabled (the default), the
optimizer keeps the distribution of the largest natural /
storage-bucketed child of a set operation and bucket-shuffles the other
children onto it (the same idea as bucket shuffle join), which avoids a
full reshuffle of the largest side. When it is disabled, the previous
full-shuffle behavior is kept, so this change is gated and only takes
effect with the frontend local shuffle planner.
Besides re-enabling the planning, two local-exchange alignment problems
that the re-enabled shapes expose are fixed (both verified on a 4-BE
cluster with a stable wrong-result reproduction before the fix and
correct results after):
1. `SetOperationNode.enforceAndDeriveLocalExchange` only handled the
colocate mode. A bucket-shuffle intersect / except whose basic child is
a join output (instead of a direct scan) fell into the partitioned
branch, so the basic side was locally re-partitioned by execution hash
while the other side stayed bucket-distributed, and the set operation
lost rows. Now the bucket-shuffle mode takes the same
`requireBucketHash` path as colocate, mirroring `HashJoinNode`.
2. `RequireSpecific.autoRequireHash()` degraded
`LOCAL_EXECUTION_HASH_SHUFFLE` to the generic hash requirement.
Pass-through operators (union / streaming agg / sort) forward this
requirement to children while claiming the specific type to their
parent, so under a bucket join upgraded to local hash, a
bucket-distributed child could satisfy the generic requirement and keep
its bucket placement while the parent skipped its re-align local
exchange, and the mixed placements computed wrong results. A specific
hash requirement is now forwarded as-is.
The regression suite disables the bucket shuffle downgrade so the chosen
shapes do not depend on the backend count / parallelism of the
environment, and adds a case where the basic child of a bucket-shuffle
intersect is a join output.
---
.../glue/translator/PhysicalPlanTranslator.java | 36 ++-
.../properties/ChildOutputPropertyDeriver.java | 121 +++++---
.../properties/ChildrenPropertiesRegulator.java | 195 ++++++++-----
.../nereids/properties/RequestPropertyDeriver.java | 55 +++-
.../job/UnassignedScanBucketOlapTableJob.java | 12 +-
.../trees/plans/physical/PhysicalSetOperation.java | 1 -
.../apache/doris/planner/LocalExchangeNode.java | 11 +-
.../org/apache/doris/planner/SetOperationNode.java | 10 +-
.../job/UnassignedScanBucketOlapTableJobTest.java | 40 +++
.../planner/LocalShuffleNodeCoverageTest.java | 49 ++++
.../bucket_shuffle_set_operation.out | 185 ++++++++++++
.../bucket_shuffle_set_operation.groovy | 315 ++++++++++++++++++++-
12 files changed, 881 insertions(+), 149 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
index 01d91395775..62e35f74473 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
@@ -2487,14 +2487,34 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
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;
- // }
- // }
+ // Storage-bucketed children only appear when the FE local shuffle
planner is active:
+ // ChildrenPropertiesRegulator and RequestPropertyDeriver both gate
the bucket-shuffle
+ // alternative on enableLocalShufflePlanner. Gate the marker on the
same flag so the
+ // dependency is explicit and a future planner change that produced a
STORAGE_BUCKETED
+ // distribution outside the local-shuffle planner cannot silently mark
BUCKET_SHUFFLE here.
+ //
+ // Within that gate a storage-bucketed child means the regulator chose
the bucket shuffle
+ // alternative (it enforces the other children onto the basic child's
buckets), so the marker
+ // simply follows that decision. It must not re-check the table id
independently: the basic
+ // child selection in the regulator is the single place that vets the
layout, and re-checking
+ // here could suppress a bucket shuffle the property model already
committed to and desync a
+ // parent that aligned to the set operation output.
+ //
+ // 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.
+ if (context.getSessionVariable() != null
+ && context.getSessionVariable().isEnableLocalShufflePlanner())
{
+ for (Plan child : setOperation.children()) {
+ PhysicalPlan childPhysicalPlan = (PhysicalPlan) child;
+ if
(JoinUtils.isStorageBucketed(childPhysicalPlan.getPhysicalProperties())) {
+
setOperationNode.setDistributionMode(DistributionMode.BUCKET_SHUFFLE);
+ break;
+ }
+ }
+ }
return setOperationFragment;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java
index 616bd387616..823231aa317 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java
@@ -71,9 +71,11 @@ import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -447,53 +449,78 @@ public class ChildOutputPropertyDeriver extends
PlanVisitor<PhysicalProperties,
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
already legal, so the
+ // output is derived by describing what the children provide:
+ // 1. one or more NATURAL children: output the first NATURAL child's
distribution
+ // (several NATURAL children behave like colocate);
+ // 2. no NATURAL but some STORAGE_BUCKETED child: output its
STORAGE_BUCKETED distribution;
+ // 3. all EXECUTION_BUCKETED children: output the execution hash (the
generic loop below);
+ // 4. anything else (e.g. random): output a non-specific property
(also below).
+ // When the basic child does not directly output its shuffle columns,
or the children do
+ // not agree, this falls through to the generic loop below, which is
equivalence-set aware
+ // and checks that every child maps its shuffle columns to the same
set-operation output
+ // positions before it claims a bucketed output. The basic child is
recomputed from the
+ // children distributions instead of being carried as mutable planner
state, because mutable
+ // state does not survive the with-copies in chooseBestPlan() and the
+ // RecomputePhysicalPropertiesPostProcessor re-derivation, while this
recomputation is
+ // deterministic on any copy of the plan.
+ int distributeToChildIndex = -1;
+ int firstStorageBucketedIndex = -1;
+ for (int i = 0; i < childrenDistribution.size(); i++) {
+ if (childrenDistribution.get(i) instanceof DistributionSpecHash) {
+ ShuffleType childShuffleType
+ = ((DistributionSpecHash)
childrenDistribution.get(i)).getShuffleType();
+ if (childShuffleType == ShuffleType.NATURAL) {
+ distributeToChildIndex = i;
+ break;
+ } else if (childShuffleType == ShuffleType.STORAGE_BUCKETED
+ && firstStorageBucketedIndex < 0) {
+ firstStorageBucketedIndex = i;
+ }
+ }
+ }
+ if (distributeToChildIndex < 0) {
+ distributeToChildIndex = firstStorageBucketedIndex;
+ }
+ if (distributeToChildIndex >= 0) {
+ 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()) {
+ // Keep the basic child's specific storage layout as the set
operation output. When
+ // the basic child is on the right (shuffleToRight) the output
rows are physically
+ // placed by the right child's storage bucket function, so
advertising that layout is
+ // truthful: a parent can co-locate this set operation against
the same layout, while
+ // a sibling on a different layout is re-aligned rather than
wrongly co-located. (The
+ // earlier "Can not find tablet ... in the bucket" failure
came from advertising a
+ // layout-less EXECUTION_BUCKETED here, which erased the table
id so two different-
+ // layout set operations looked co-locatable; preserving the
layout fixes that.)
+ return new PhysicalProperties(
+ new DistributionSpecHash(
+ setOperationDistributeColumnIds,
+ childDistribution.getShuffleType(),
+ childDistribution.getTableId(),
+ childDistribution.getSelectedIndexId(),
+ childDistribution.getPartitionIds()
+ )
+ );
+ }
+ }
for (int i = 0; i < childrenDistribution.size(); i++) {
DistributionSpec childDistribution = childrenDistribution.get(i);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java
index e5e0d9d1bd0..6aa9d01c7cc 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java
@@ -57,6 +57,7 @@ import org.apache.doris.statistics.util.StatisticsUtil;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -651,83 +652,103 @@ public class ChildrenPropertiesRegulator extends
PlanVisitor<List<List<PhysicalP
} 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;
+
+ // find the bucket shuffle basic index: the largest natural /
storage-bucketed child
+ // keeps its bucket distribution, every other child is
bucket-shuffled to it.
+ // RequestPropertyDeriver only asks ShuffleType.REQUIRE when
set-op bucket shuffle
+ // is allowed, so the required shuffle type is the single source
of truth here:
+ // for any other required type keep bucketShuffleBasicIndex = -1
and fall back to
+ // the execution-bucketed (partitioned) shuffle below.
+ // 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 (basic.getShuffleType() == ShuffleType.REQUIRE) {
+ 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();
+ // The table id is deliberately not checked here:
DistributionSpecHash.satisfy
+ // aligns the other children by shuffle type and
columns regardless of the table
+ // id, and a basic child with an unknown layout (a
hash join output with the table
+ // id cleared to -1 by
withShuffleTypeAndForbidColocateJoin) produces a
+ // STORAGE_BUCKETED output, which couldColocateJoin
never co-locates (it requires
+ // NATURAL on both sides) so it cannot mislead a
parent into a wrong co-location.
+ if (childDistribution instanceof DistributionSpecHash
+ && supportBucketShuffleTypes.contains(
+ ((DistributionSpecHash)
childDistribution).getShuffleType())
+ &&
canMapBucketKeysToRequire((DistributionSpecHash) childDistribution,
+ (DistributionSpecHash)
requiredProperties.get(i).getDistributionSpec())
+ &&
!(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;
+ }
+ }
+
+ if (bucketShuffleBasicIndex >= 0) {
+ // use bucket shuffle
+ DistributionSpecHash notShuffleSideRequire
+ = (DistributionSpecHash)
requiredProperties.get(bucketShuffleBasicIndex)
+ .getDistributionSpec();
+
+ DistributionSpecHash notNeedShuffleOutput
+ = (DistributionSpecHash)
originChildrenProperties.get(bucketShuffleBasicIndex)
+ .getDistributionSpec();
+
+ for (int i = 0; i < originChildrenProperties.size(); i++) {
+ if (i == bucketShuffleBasicIndex) {
+ continue;
+ }
+
+ DistributionSpecHash currentRequire
+ = (DistributionSpecHash)
requiredProperties.get(i).getDistributionSpec();
+
+ // The enforced child is bucket-shuffled to the basic
child's buckets by the
+ // storage hash on shuffleSideIds. Only the shuffle type
and the column order
+ // carry the alignment: DistributionSpecHash.satisfy
compares shuffle type and
+ // columns (the equivalence set), never the storage
layout, and the set operation
+ // output is STORAGE_BUCKETED which couldColocateJoin
never co-locates, so the
+ // basic child's table / index / partition ids are inert
here.
+ List<ExprId> shuffleSideIds =
calAnotherSideRequiredShuffleIds(
+ notNeedShuffleOutput, notShuffleSideRequire,
currentRequire);
+ PhysicalProperties target = new PhysicalProperties(
+ new DistributionSpecHash(shuffleSideIds,
ShuffleType.STORAGE_BUCKETED));
updateChildEnforceAndCost(i, target);
}
+ } else {
+ // use partitioned shuffle
+ 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());
+ updateChildEnforceAndCost(i, target);
+ }
+ }
}
- // }
}
return ImmutableList.of(originChildrenProperties);
}
@@ -801,6 +822,32 @@ public class ChildrenPropertiesRegulator extends
PlanVisitor<List<List<PhysicalP
}
}
+ /**
+ * Whether every bucket key of the candidate basic child can be mapped
into the child's
+ * required hash columns (directly or through its equivalence sets). When
the candidate's
+ * bucket key is wider than the set operation output (e.g. a table
bucketed by (k, v)
+ * feeding INTERSECT on k only), the mapping is impossible and choosing it
as the basic
+ * child would fail calAnotherSideRequiredShuffleIds, so the caller falls
back to the
+ * execution-bucketed shuffle instead.
+ */
+ private boolean canMapBucketKeysToRequire(DistributionSpecHash
childOutput, DistributionSpecHash childRequired) {
+ for (ExprId scanId : childOutput.getOrderedShuffledColumns()) {
+ int index =
childRequired.getOrderedShuffledColumns().indexOf(scanId);
+ if (index == -1) {
+ for (ExprId alternativeExpr :
childOutput.getEquivalenceExprIdsOf(scanId)) {
+ index =
childRequired.getOrderedShuffledColumns().indexOf(alternativeExpr);
+ if (index != -1) {
+ break;
+ }
+ }
+ }
+ if (index == -1) {
+ return false;
+ }
+ }
+ return true;
+ }
+
/**
* calculate the shuffle side hash key right orders.
* For example,
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/RequestPropertyDeriver.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/RequestPropertyDeriver.java
index 70f2b51665b..6ea2601bbee 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/RequestPropertyDeriver.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/RequestPropertyDeriver.java
@@ -67,6 +67,7 @@ import
org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
import org.apache.doris.nereids.util.AggregateUtils;
import org.apache.doris.nereids.util.JoinUtils;
import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.SessionVariable;
import org.apache.doris.statistics.Statistics;
import com.google.common.base.Preconditions;
@@ -333,20 +334,18 @@ public class RequestPropertyDeriver extends
PlanVisitor<Void, PlanContext> {
if (distributionRequestFromParent instanceof DistributionSpecHash) {
// shuffle according to parent require
DistributionSpecHash distributionSpecHash = (DistributionSpecHash)
distributionRequestFromParent;
- addRequestPropertyToChildren(createHashRequestAccordingToParent(
- setOperation, distributionSpecHash, context));
+
addRequestPropertyToChildren(downgradeRequireWhenBucketShuffleNotAllowed(
+ createHashRequestAccordingToParent(setOperation,
distributionSpecHash, context)));
} else {
// 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
+ ShuffleType setOperationShuffleType =
setOperationBucketShuffleAllowed()
+ ? ShuffleType.REQUIRE : ShuffleType.EXECUTION_BUCKETED;
addRequestPropertyToChildren(setOperation.getRegularChildrenOutputs().stream()
.map(childOutputs -> childOutputs.stream()
.map(SlotReference::getExprId)
.collect(ImmutableList.toImmutableList()))
- .map(l -> PhysicalProperties.createHash(l,
ShuffleType.EXECUTION_BUCKETED))
+ .map(l -> PhysicalProperties.createHash(l,
setOperationShuffleType))
.collect(Collectors.toList()));
}
return null;
@@ -366,9 +365,8 @@ public class RequestPropertyDeriver extends
PlanVisitor<Void, PlanContext> {
DistributionSpec distributionRequestFromParent =
requestPropertyFromParent.getDistributionSpec();
if (distributionRequestFromParent instanceof DistributionSpecHash)
{
DistributionSpecHash distributionSpecHash =
(DistributionSpecHash) distributionRequestFromParent;
- List<PhysicalProperties> requestHash
- = createHashRequestAccordingToParent(union,
distributionSpecHash, context);
- addRequestPropertyToChildren(requestHash);
+
addRequestPropertyToChildren(downgradeRequireWhenBucketShuffleNotAllowed(
+ createHashRequestAccordingToParent(union,
distributionSpecHash, context)));
}
}
@@ -581,6 +579,43 @@ public class RequestPropertyDeriver extends
PlanVisitor<Void, PlanContext> {
return combinedNdv > AggregateUtils.LOW_NDV_THRESHOLD;
}
+ /**
+ * A ShuffleType.REQUIRE request lets ChildrenPropertiesRegulator choose
the bucket
+ * shuffle alternative for the set operation. That needs either no local
shuffle at all
+ * (every pipeline runs a single task per instance, so the bucket
alignment holds
+ * naturally) or the FE local-shuffle planner (which plans the correct
bucket-hash
+ * local exchanges): with the BE-planned local shuffle 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.
+ */
+ private boolean setOperationBucketShuffleAllowed() {
+ return connectContext != null
+ &&
SessionVariable.canUseNereidsDistributePlanner(connectContext)
+ && (!connectContext.getSessionVariable().isEnableLocalShuffle()
+ ||
connectContext.getSessionVariable().isEnableLocalShufflePlanner());
+ }
+
+ /**
+ * The parent may pass ShuffleType.REQUIRE down through
+ * {@link #createHashRequestAccordingToParent}; downgrade it to
EXECUTION_BUCKETED so the
+ * regulator does not pick the bucket shuffle alternative when it is not
allowed.
+ */
+ private List<PhysicalProperties>
downgradeRequireWhenBucketShuffleNotAllowed(
+ List<PhysicalProperties> requests) {
+ if (setOperationBucketShuffleAllowed()) {
+ return requests;
+ }
+ return requests.stream().map(request -> {
+ DistributionSpecHash requestHash = (DistributionSpecHash)
request.getDistributionSpec();
+ return requestHash.getShuffleType() == ShuffleType.REQUIRE
+ ? PhysicalProperties.createHash(
+ requestHash.getOrderedShuffledColumns(),
ShuffleType.EXECUTION_BUCKETED)
+ : request;
+ }).collect(Collectors.toList());
+ }
+
private List<PhysicalProperties> createHashRequestAccordingToParent(
SetOperation setOperation, DistributionSpecHash
distributionRequestFromParent, PlanContext context) {
List<PhysicalProperties> requiredPropertyList =
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/UnassignedScanBucketOlapTableJob.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/UnassignedScanBucketOlapTableJob.java
index 390b66076d4..d4d65dfd225 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/UnassignedScanBucketOlapTableJob.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/UnassignedScanBucketOlapTableJob.java
@@ -24,7 +24,6 @@ import org.apache.doris.catalog.Replica;
import org.apache.doris.catalog.Tablet;
import org.apache.doris.common.Pair;
import org.apache.doris.nereids.StatementContext;
-import org.apache.doris.nereids.trees.plans.algebra.Intersect;
import org.apache.doris.nereids.trees.plans.distribute.DistributeContext;
import
org.apache.doris.nereids.trees.plans.distribute.worker.DistributedPlanWorker;
import
org.apache.doris.nereids.trees.plans.distribute.worker.DistributedPlanWorkerManager;
@@ -32,12 +31,14 @@ import
org.apache.doris.nereids.trees.plans.distribute.worker.ScanWorkerSelector
import org.apache.doris.nereids.util.Utils;
import org.apache.doris.planner.ExchangeNode;
import org.apache.doris.planner.HashJoinNode;
+import org.apache.doris.planner.IntersectNode;
import org.apache.doris.planner.OlapScanNode;
import org.apache.doris.planner.PlanFragment;
import org.apache.doris.planner.ScanNode;
import org.apache.doris.planner.SetOperationNode;
import org.apache.doris.qe.ConnectContext;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ImmutableList;
@@ -352,7 +353,8 @@ public class UnassignedScanBucketOlapTableJob extends
AbstractUnassignedScanJob
}
}
- private boolean shouldFillUpInstances(List<HashJoinNode> hashJoinNodes,
List<SetOperationNode> setOperationNodes) {
+ @VisibleForTesting
+ static boolean shouldFillUpInstances(List<HashJoinNode> hashJoinNodes,
List<SetOperationNode> setOperationNodes) {
for (HashJoinNode hashJoinNode : hashJoinNodes) {
if (!hashJoinNode.isBucketShuffle()) {
continue;
@@ -368,7 +370,11 @@ public class UnassignedScanBucketOlapTableJob extends
AbstractUnassignedScanJob
}
for (SetOperationNode setOperationNode : setOperationNodes) {
- if (setOperationNode instanceof Intersect) {
+ // INTERSECT does not need missing-bucket receiver fill-up: a
bucket the basic child
+ // does not scan is empty in the anchor, so the intersect result
for that bucket is
+ // empty and the other children's rows shuffled there produce
nothing. Note the node
+ // here is the legacy planner IntersectNode, not the Nereids
algebra Intersect.
+ if (setOperationNode instanceof IntersectNode) {
continue;
}
if (setOperationNode.isBucketShuffle()) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalSetOperation.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalSetOperation.java
index a0af54c7ed0..1f5c843e292 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalSetOperation.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalSetOperation.java
@@ -44,7 +44,6 @@ import java.util.Optional;
* Physical SetOperation.
*/
public abstract class PhysicalSetOperation extends AbstractPhysicalPlan
implements SetOperation {
- public static final String DISTRIBUTE_TO_CHILD_INDEX =
"DistributeToChildIndex";
protected final Qualifier qualifier;
protected final List<NamedExpression> outputs;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/LocalExchangeNode.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/LocalExchangeNode.java
index f939102d699..66eda400799 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/LocalExchangeNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/LocalExchangeNode.java
@@ -283,8 +283,15 @@ public class LocalExchangeNode extends PlanNode {
@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()) {
return this;
}
return RequireHash.INSTANCE;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/SetOperationNode.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/SetOperationNode.java
index 4074ab3e85a..1396701d3c7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/SetOperationNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/SetOperationNode.java
@@ -223,7 +223,15 @@ public abstract class SetOperationNode extends PlanNode {
: LocalExchangeType.NOOP;
} else {
// Intersect / Except
- if (AddLocalExchange.isColocated(this)) {
+ if (AddLocalExchange.isColocated(this) || isBucketShuffle()) {
+ // COLOCATE / BUCKET_SHUFFLE: every child is distributed by
the basic child's
+ // storage bucket function (basic side scans buckets directly,
other sides come
+ // from bucket-shuffle exchanges), so all children must stay
aligned by that
+ // bucket function locally. requireBucketHash keeps
bucket-distributed children
+ // as-is and re-aligns a serial (NOOP-claim) child with a
BUCKET_HASH_SHUFFLE
+ // local exchange — same pattern as HashJoinNode's
colocate/bucket-shuffle
+ // branch. An execution-hash require here would locally
re-partition one side
+ // by a different hash function and break build/probe
alignment.
requireChild = LocalExchangeTypeRequire.requireBucketHash();
outputType = LocalExchangeType.BUCKET_HASH_SHUFFLE;
} else {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/UnassignedScanBucketOlapTableJobTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/UnassignedScanBucketOlapTableJobTest.java
index db7f55fb2fd..e5020e2b8a6 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/UnassignedScanBucketOlapTableJobTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/UnassignedScanBucketOlapTableJobTest.java
@@ -20,9 +20,14 @@ package
org.apache.doris.nereids.trees.plans.distribute.worker.job;
import org.apache.doris.nereids.StatementContext;
import
org.apache.doris.nereids.trees.plans.distribute.worker.ScanWorkerSelector;
import org.apache.doris.planner.DataPartition;
+import org.apache.doris.planner.ExceptNode;
import org.apache.doris.planner.ExchangeNode;
+import org.apache.doris.planner.HashJoinNode;
+import org.apache.doris.planner.IntersectNode;
import org.apache.doris.planner.OlapScanNode;
import org.apache.doris.planner.PlanFragment;
+import org.apache.doris.planner.SetOperationNode;
+import org.apache.doris.planner.UnionNode;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.OriginStatement;
import org.apache.doris.thrift.TUniqueId;
@@ -34,6 +39,7 @@ import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import java.util.BitSet;
+import java.util.List;
public class UnassignedScanBucketOlapTableJobTest {
@@ -114,4 +120,38 @@ public class UnassignedScanBucketOlapTableJobTest {
int result = unassignedJob.degreeOfParallelism(3, false);
Assertions.assertEquals(100, result);
}
+
+ @Test
+ public void testShouldFillUpInstancesSkipsBucketShuffleIntersectOnly() {
+ List<HashJoinNode> noJoins = ImmutableList.of();
+
+ // A bucket-shuffle INTERSECT must NOT trigger missing-bucket receiver
fill-up: a bucket the
+ // basic child does not scan is empty in the anchor, so the intersect
result for that bucket
+ // is empty and the other children's rows shuffled there produce
nothing. The skip must match
+ // the legacy planner IntersectNode; the translated node is never an
instance of the Nereids
+ // algebra Intersect, so checking that type would silently never skip.
+ IntersectNode bucketShuffleIntersect =
Mockito.mock(IntersectNode.class);
+
Mockito.when(bucketShuffleIntersect.isBucketShuffle()).thenReturn(true);
+
Assertions.assertFalse(UnassignedScanBucketOlapTableJob.shouldFillUpInstances(
+ noJoins,
ImmutableList.<SetOperationNode>of(bucketShuffleIntersect)));
+
+ // Bucket-shuffle UNION / EXCEPT still need fill-up: the other
children's rows shuffled into
+ // buckets the basic child does not scan must be received to be
produced (union) or to
+ // subtract against (except) correctly.
+ UnionNode bucketShuffleUnion = Mockito.mock(UnionNode.class);
+ Mockito.when(bucketShuffleUnion.isBucketShuffle()).thenReturn(true);
+
Assertions.assertTrue(UnassignedScanBucketOlapTableJob.shouldFillUpInstances(
+ noJoins,
ImmutableList.<SetOperationNode>of(bucketShuffleUnion)));
+
+ ExceptNode bucketShuffleExcept = Mockito.mock(ExceptNode.class);
+ Mockito.when(bucketShuffleExcept.isBucketShuffle()).thenReturn(true);
+
Assertions.assertTrue(UnassignedScanBucketOlapTableJob.shouldFillUpInstances(
+ noJoins,
ImmutableList.<SetOperationNode>of(bucketShuffleExcept)));
+
+ // A union that did not choose bucket shuffle does not trigger fill-up.
+ UnionNode plainUnion = Mockito.mock(UnionNode.class);
+ Mockito.when(plainUnion.isBucketShuffle()).thenReturn(false);
+
Assertions.assertFalse(UnassignedScanBucketOlapTableJob.shouldFillUpInstances(
+ noJoins, ImmutableList.<SetOperationNode>of(plainUnion)));
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/planner/LocalShuffleNodeCoverageTest.java
b/fe/fe-core/src/test/java/org/apache/doris/planner/LocalShuffleNodeCoverageTest.java
index 5ffe61ce8f5..e171ebe166c 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/planner/LocalShuffleNodeCoverageTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/planner/LocalShuffleNodeCoverageTest.java
@@ -53,6 +53,33 @@ import java.util.concurrent.atomic.AtomicInteger;
public class LocalShuffleNodeCoverageTest {
private static final AtomicInteger NEXT_ID = new AtomicInteger(1);
+ @Test
+ public void testRequireSpecificAutoRequireHashPreservesSpecificHash() {
+ // Pass-through operators (union / streaming agg / sort) forward their
parent's specific
+ // hash requirement downward via autoRequireHash() while leaving row
placement to their
+ // children. Every hash flavour must be forwarded unchanged: degrading
a specific
+ // LOCAL_EXECUTION_HASH_SHUFFLE requirement to the generic RequireHash
would let a
+ // bucket-distributed child satisfy it and keep its bucket placement
while the operator
+ // still advertised LOCAL_EXECUTION_HASH_SHUFFLE upward, so a bucket
join upgraded to
+ // local hash above it would skip its realign local exchange and
compute wrong results.
+ for (LocalExchangeType hashType : new LocalExchangeType[] {
+ LocalExchangeType.LOCAL_EXECUTION_HASH_SHUFFLE,
+ LocalExchangeType.GLOBAL_EXECUTION_HASH_SHUFFLE,
+ LocalExchangeType.BUCKET_HASH_SHUFFLE}) {
+ LocalExchangeNode.RequireSpecific require = new
LocalExchangeNode.RequireSpecific(hashType);
+ LocalExchangeTypeRequire forwarded = require.autoRequireHash();
+ Assertions.assertSame(require, forwarded,
+ "specific hash require " + hashType + " must be forwarded
unchanged");
+ Assertions.assertEquals(hashType, forwarded.preferType());
+ }
+
+ // A non-hash specific require still relaxes to the generic
RequireHash, whose preferType
+ // is GLOBAL_EXECUTION_HASH_SHUFFLE.
+ LocalExchangeTypeRequire relaxed =
+ new
LocalExchangeNode.RequireSpecific(LocalExchangeType.PASSTHROUGH).autoRequireHash();
+
Assertions.assertEquals(LocalExchangeType.GLOBAL_EXECUTION_HASH_SHUFFLE,
relaxed.preferType());
+ }
+
@Test
public void testSelectNode() {
PlanTranslatorContext ctx = new PlanTranslatorContext();
@@ -613,6 +640,28 @@ public class LocalShuffleNodeCoverageTest {
Assertions.assertSame(exceptLeft, exceptNode.getChild(0));
Assertions.assertSame(exceptRight, exceptNode.getChild(1));
+ // Bucket-shuffle IntersectNode (not colocated): exercises the `||
isBucketShuffle()` leg.
+ // Every child is distributed by the basic child's storage bucket
function (via bucket-shuffle
+ // exchanges), so the output is BUCKET_HASH_SHUFFLE and each serial
(NOOP) child is re-aligned
+ // with a BUCKET_HASH_SHUFFLE local exchange. Without the
isBucketShuffle() branch this would
+ // fall into the partitioned (GLOBAL hash) leg and re-partition one
side by a different hash
+ // function, breaking alignment. setColocate(false) keeps
isColocated() false (SetOperationNode
+ // returns false immediately when isColocate() is false), so the
branch is reached through
+ // isBucketShuffle() alone.
+ IntersectNode bucketIntersect = new IntersectNode(nextPlanNodeId(),
+ new TupleId(NEXT_ID.getAndIncrement()));
+ bucketIntersect.setColocate(false);
+ bucketIntersect.setDistributionMode(DistributionMode.BUCKET_SHUFFLE);
+ TrackingPlanNode bucketLeft = new TrackingPlanNode(nextPlanNodeId(),
LocalExchangeType.NOOP);
+ TrackingPlanNode bucketRight = new TrackingPlanNode(nextPlanNodeId(),
LocalExchangeType.NOOP);
+ bucketIntersect.addChild(bucketLeft);
+ bucketIntersect.addChild(bucketRight);
+ Pair<PlanNode, LocalExchangeType> bucketIntersectOutput =
bucketIntersect.enforceAndDeriveLocalExchange(
+ ctx, null, LocalExchangeTypeRequire.requireHash());
+ Assertions.assertEquals(LocalExchangeType.BUCKET_HASH_SHUFFLE,
bucketIntersectOutput.second);
+ assertChildLocalExchangeType(bucketIntersect, 0,
LocalExchangeType.BUCKET_HASH_SHUFFLE);
+ assertChildLocalExchangeType(bucketIntersect, 1,
LocalExchangeType.BUCKET_HASH_SHUFFLE);
+
TrackingPlanNode assertChild = new TrackingPlanNode(nextPlanNodeId(),
LocalExchangeType.NOOP);
AssertNumRowsElement assertElement =
Mockito.mock(AssertNumRowsElement.class);
Mockito.when(assertElement.getDesiredNumOfRows()).thenReturn(1L);
diff --git
a/regression-test/data/query_p0/set_operations/bucket_shuffle_set_operation.out
b/regression-test/data/query_p0/set_operations/bucket_shuffle_set_operation.out
index 50a3c5cc131..1ec75925ec2 100644
---
a/regression-test/data/query_p0/set_operations/bucket_shuffle_set_operation.out
+++
b/regression-test/data/query_p0/set_operations/bucket_shuffle_set_operation.out
@@ -117,3 +117,188 @@ PhysicalResultSink
2
3
+-- !bucket_shuffle_join_as_basic_child_shape --
+PhysicalResultSink
+--PhysicalIntersect[bucketShuffle]
+----PhysicalProject
+------hashJoin[INNER_JOIN broadcast] hashCondition=((a.id = b.id))
otherCondition=()
+--------PhysicalProject
+----------PhysicalOlapScan[bucket_shuffle_set_operation1(a)]
+--------PhysicalProject
+----------PhysicalOlapScan[bucket_shuffle_set_operation2(b)]
+----PhysicalDistribute[DistributionSpecHash]
+------PhysicalProject
+--------PhysicalOlapScan[bucket_shuffle_set_operation3]
+
+-- !bucket_shuffle_join_as_basic_child_result --
+1
+2
+3
+
+-- !bucket_shuffle_nested_set_operation_shape --
+PhysicalResultSink
+--PhysicalUnion
+----PhysicalDistribute[DistributionSpecExecutionAny]
+------PhysicalProject
+--------PhysicalOlapScan[bucket_shuffle_set_operation3]
+----PhysicalDistribute[DistributionSpecExecutionAny]
+------PhysicalIntersect[bucketShuffle]
+--------PhysicalProject
+----------hashJoin[INNER_JOIN broadcast] hashCondition=((a.id = b.id))
otherCondition=()
+------------PhysicalProject
+--------------PhysicalOlapScan[bucket_shuffle_set_operation1(a)]
+------------PhysicalProject
+--------------PhysicalOlapScan[bucket_shuffle_set_operation2(b)]
+--------PhysicalDistribute[DistributionSpecHash]
+----------PhysicalProject
+------------PhysicalOlapScan[bucket_shuffle_set_operation2]
+
+-- !bucket_shuffle_nested_set_operation_result --
+1
+1
+2
+2
+3
+3
+
+-- !bucket_shuffle_when_local_shuffle_off_shape --
+PhysicalResultSink
+--PhysicalIntersect[bucketShuffle]
+----PhysicalProject
+------PhysicalOlapScan[bucket_shuffle_set_operation1]
+----PhysicalDistribute[DistributionSpecHash]
+------PhysicalProject
+--------PhysicalOlapScan[bucket_shuffle_set_operation2]
+
+-- !bucket_shuffle_when_local_shuffle_off_result --
+1
+2
+3
+
+-- !union_parent_hash_shape_when_local_shuffle_planner_off --
+PhysicalResultSink
+--PhysicalProject
+----hashJoin[INNER_JOIN bucketShuffle] hashCondition=((u.id = b.id))
otherCondition=()
+------PhysicalUnion
+--------PhysicalDistribute[DistributionSpecHash]
+----------PhysicalProject
+------------PhysicalOlapScan[bucket_shuffle_set_operation1]
+--------PhysicalDistribute[DistributionSpecHash]
+----------PhysicalProject
+------------PhysicalOlapScan[bucket_shuffle_set_operation2]
+------PhysicalProject
+--------PhysicalOlapScan[bucket_shuffle_set_operation3(b)]
+
+Hint log:
+Used: [shuffle]_1
+UnUsed:
+SyntaxError:
+
+-- !union_parent_hash_when_local_shuffle_planner_off --
+1
+1
+2
+2
+3
+3
+
+-- !plain_intersect_when_local_shuffle_planner_off --
+1
+2
+3
+
+-- !plain_intersect_when_nereids_distribute_planner_off --
+1
+2
+3
+
+-- !union_parent_hash_when_nereids_distribute_planner_off --
+1
+1
+2
+2
+3
+3
+
+-- !intersect_right_basic_parent_hash --
+1
+2
+3
+
+-- !bucket_shuffle_union_fill_up --
+0
+0
+1
+10
+11
+12
+13
+14
+15
+16
+17
+18
+19
+2
+2
+3
+4
+4
+5
+6
+6
+7
+8
+9
+
+-- !bucket_shuffle_equivalent_key_fill_up --
+0
+1
+10
+11
+12
+13
+14
+15
+16
+17
+18
+19
+2
+2
+3
+4
+5
+6
+7
+8
+9
+
+-- !bucket_shuffle_two_equivalent_keys --
+0 0
+1 1
+10 10
+11 11
+12 12
+13 13
+14 14
+15 15
+16 16
+17 17
+18 18
+19 19
+2 2
+2 2
+3 3
+4 4
+5 5
+6 6
+7 7
+8 8
+9 9
+
+-- !multi_column_bucket_key_intersect --
+1
+2
+3
+
diff --git
a/regression-test/suites/query_p0/set_operations/bucket_shuffle_set_operation.groovy
b/regression-test/suites/query_p0/set_operations/bucket_shuffle_set_operation.groovy
index 5533853eaa9..ee118bccbd5 100644
---
a/regression-test/suites/query_p0/set_operations/bucket_shuffle_set_operation.groovy
+++
b/regression-test/suites/query_p0/set_operations/bucket_shuffle_set_operation.groovy
@@ -16,9 +16,6 @@
// under the License.
suite("bucket_shuffle_set_operation") {
- // TODO: open comment when support `enable_local_shuffle_planner` and
change to REQUIRE
- return
-
multi_sql """
drop table if exists bucket_shuffle_set_operation1;
create table bucket_shuffle_set_operation1(id int, value int)
distributed by hash(id) buckets 10 properties('replication_num'='1');
@@ -37,6 +34,9 @@ suite("bucket_shuffle_set_operation") {
// make bucket shuffle set operation stable
sql "set parallel_pipeline_task_num=5"
+ // disable the bucket shuffle downgrade so the chosen shapes do not depend
on the
+ // backend count / parallelism of the environment running this suite
+ sql "set bucket_shuffle_downgrade_ratio=0"
def checkShapeAndResult = { String tag, String sqlStr ->
quickTest(tag + "_shape", "explain shape plan " + sqlStr)
@@ -95,6 +95,315 @@ 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
+ """
+ sql "set enable_local_shuffle_planner=false"
+ // Golden shape: the union must stay a plain PhysicalUnion whose two
children are each a
+ // PhysicalDistribute[DistributionSpecHash]. That is the actual proof that
the parent hash
+ // request was pushed down into the union
(createHashRequestAccordingToParent) and downgraded
+ // to execution hash, not merely that the PhysicalUnion line lacks the
[bucketShuffle] tag.
+ // If that path regressed, the optimizer could instead keep an unbucketed
union and add a
+ // single PhysicalDistribute[DistributionSpecHash] above it for the
shuffle join; the golden
+ // shape below (union children are distributes, not direct scans) would
catch that.
+ qt_union_parent_hash_shape_when_local_shuffle_planner_off("explain shape
plan " + unionParentHashSql)
+ explain {
+ sql "shape plan " + unionParentHashSql
+ check { String e ->
+ def unionIndex = e.indexOf("PhysicalUnion")
+ assertTrue(unionIndex >= 0)
+ // the union must not be a bucket shuffle union when the FE local
shuffle planner is off
+ assertFalse(e.substring(unionIndex,
+ Math.min(unionIndex + "PhysicalUnion".length() + 20,
e.length())).contains("bucketShuffle"))
+ // and the parent hash request must have been pushed into the
union: each union child
+ // arrives through a PhysicalDistribute[DistributionSpecHash], so
no union child is a
+ // direct scan.
+ def afterUnion = e.substring(unionIndex)
+ def joinProbeIndex =
afterUnion.indexOf("bucket_shuffle_set_operation3")
+ def unionSubtree = joinProbeIndex >= 0 ? afterUnion.substring(0,
joinProbeIndex) : afterUnion
+ assertEquals(2,
unionSubtree.split("PhysicalDistribute\\[DistributionSpecHash\\]", -1).length -
1)
+ }
+ }
+ order_qt_union_parent_hash_when_local_shuffle_planner_off
unionParentHashSql
+ sql "set enable_local_shuffle_planner=true"
+
+ // A plain intersect/except without a parent hash request goes through
+ // visitPhysicalSetOperation directly (not the parent-hash request path);
with the FE local
+ // shuffle planner disabled it must not choose bucket shuffle either, and
the result stays
+ // correct.
+ sql "set enable_local_shuffle_planner=false"
+ explain {
+ sql "shape plan select id from bucket_shuffle_set_operation1 intersect
select id from bucket_shuffle_set_operation2"
+ check { String e ->
+ assertFalse(e.contains("bucketShuffle"))
+ }
+ }
+ order_qt_plain_intersect_when_local_shuffle_planner_off "select id from
bucket_shuffle_set_operation1 intersect select id from
bucket_shuffle_set_operation2"
+ sql "set enable_local_shuffle_planner=true"
+
+ // Set operation bucket shuffle is gated on
setOperationBucketShuffleAllowed() =
+ // enableLocalShufflePlanner && canUseNereidsDistributePlanner. The
enable_local_shuffle_planner
+ // =false cases above only exercise the first conjunct; this block guards
the second one. With
+ // local shuffle and the FE local-shuffle planner both enabled but the
Nereids distribute planner
+ // disabled, the set operation must still downgrade (no [bucketShuffle])
and keep correct results.
+ sql "set enable_local_shuffle=true"
+ sql "set enable_local_shuffle_planner=true"
+ sql "set enable_nereids_distribute_planner=false"
+ // plain intersect goes through visitPhysicalSetOperation directly
+ explain {
+ sql "shape plan select id from bucket_shuffle_set_operation1 intersect
select id from bucket_shuffle_set_operation2"
+ check { String e ->
+ assertFalse(e.contains("bucketShuffle"))
+ }
+ }
+ order_qt_plain_intersect_when_nereids_distribute_planner_off "select id
from bucket_shuffle_set_operation1 intersect select id from
bucket_shuffle_set_operation2"
+ // the parent-hash union path (createHashRequestAccordingToParent) must
downgrade too
+ explain {
+ sql "shape plan " + unionParentHashSql
+ check { String e ->
+ def unionIndex = e.indexOf("PhysicalUnion")
+ assertTrue(unionIndex >= 0)
+ assertFalse(e.substring(unionIndex,
+ Math.min(unionIndex + "PhysicalUnion".length() + 20,
e.length())).contains("bucketShuffle"))
+ }
+ }
+ order_qt_union_parent_hash_when_nereids_distribute_planner_off
unionParentHashSql
+ sql "set enable_nereids_distribute_planner=true"
+
+ // The right child can be selected as the bucket-shuffle basic child
(larger row count). The
+ // set operation output must then advertise a non-specific distribution
rather than a plain
+ // execution hash: otherwise two such set operations with different
storage layouts are
+ // co-located under a join and fail bucket assignment ("Can not find
tablet ... in the
+ // bucket"). r1 / r2 are larger than the small tables so they become the
basic child on the
+ // right, and they are different tables so their bucket layouts differ.
+ sql "drop table if exists bucket_shuffle_set_operation_r1"
+ sql "create table bucket_shuffle_set_operation_r1(id int) distributed by
hash(id) buckets 10 properties('replication_num'='1')"
+ sql "insert into bucket_shuffle_set_operation_r1 select number from
numbers('number'='20')"
+ sql "drop table if exists bucket_shuffle_set_operation_r2"
+ sql "create table bucket_shuffle_set_operation_r2(id int) distributed by
hash(id) buckets 10 properties('replication_num'='1')"
+ sql "insert into bucket_shuffle_set_operation_r2 select number from
numbers('number'='20')"
+ sql "alter table bucket_shuffle_set_operation_r1 modify column id set
stats ('row_count'='1000', 'ndv'='1000', 'min_value'='0', 'max_value'='19')"
+ sql "alter table bucket_shuffle_set_operation_r2 modify column id set
stats ('row_count'='1000', 'ndv'='1000', 'min_value'='0', 'max_value'='19')"
+ def intersectRightBasicSql = """
+ select t.id from
+ (select id from bucket_shuffle_set_operation1 intersect select id
from bucket_shuffle_set_operation_r1) t
+ join[shuffle]
+ (select id from bucket_shuffle_set_operation1 intersect select id
from bucket_shuffle_set_operation_r2) s
+ on t.id = s.id"""
+ // Prove the shuffleToRight branch (distributeToChildIndex > 0) is
actually covered, under a
+ // parent hash consumer. The join[shuffle] hint forces the parent to be a
hash consumer of the
+ // two set-operation outputs. Each INTERSECT must keep the right table (r1
/ r2, the larger row
+ // count) as the direct bucketed basic child while the left op1 is the
enforced
+ // PhysicalDistribute[DistributionSpecHash] side. Because
visitPhysicalSetOperation keeps the
+ // right basic child's specific storage layout, the two intersects carry
different table ids
+ // (r1 vs r2), so the parent re-aligns them with a bucket-shuffle join
instead of co-locating
+ // them (which, if the layout were erased to a layout-less execution hash,
would fail bucket
+ // assignment with "Can not find tablet"). If the right side stopped being
selected as basic, or
+ // the layout were dropped so the join co-located, these assertions fail
even though the result
+ // stays 1, 2, 3.
+ explain {
+ sql "shape plan " + intersectRightBasicSql
+ check { String e ->
+ assertTrue(e.contains("PhysicalIntersect[bucketShuffle]"))
+ // the parent join re-aligns the two different-layout outputs
(bucket-shuffle hash
+ // consumer) and does not co-locate them
+ assertTrue(e.contains("hashJoin[INNER_JOIN bucketShuffle]"),
+ "parent must re-align the two different-layout outputs
with a bucket-shuffle join")
+ assertFalse(e.contains("colocated"), "parent must not co-locate
the two right-basic intersects")
+ def lines = e.split("\n").findAll { it =~ /Physical|hashJoin/ }
+ def depth = { String s -> (s =~ /^-*/)[0].length() }
+ def parentOf = { int idx ->
+ for (int k = idx - 1; k >= 0; k--) {
+ if (depth(lines[k]) < depth(lines[idx])) {
+ return lines[k]
+ }
+ }
+ return ""
+ }
+ // each INTERSECT keeps r1 / r2 (the larger, right side) as its
direct bucketed basic child
+ ["bucket_shuffle_set_operation_r1",
"bucket_shuffle_set_operation_r2"].each { rt ->
+ int i = lines.findIndexOf { it.contains("PhysicalOlapScan[" +
rt + "]") }
+ assertTrue(i >= 0, rt + " scan not found in shape")
+
assertTrue(parentOf(i).contains("PhysicalIntersect[bucketShuffle]"),
+ rt + " must be the direct bucketed basic child of the
intersect (right basic)")
+ }
+ // and each INTERSECT's other (left) child arrives through an
enforced
+ // PhysicalDistribute[DistributionSpecHash] directly under the
intersect
+ int enforcedLeftCount = 0
+ lines.eachWithIndex { String ln, int i ->
+ if (ln.contains("PhysicalDistribute[DistributionSpecHash]")
+ &&
parentOf(i).contains("PhysicalIntersect[bucketShuffle]")) {
+ enforcedLeftCount++
+ }
+ }
+ assertTrue(enforcedLeftCount >= 2,
+ "each intersect must shuffle its left child through a
PhysicalDistribute[DistributionSpecHash]")
+ }
+ }
+ order_qt_intersect_right_basic_parent_hash intersectRightBasicSql
+
+ // A non-intersect bucket-shuffle set operation (UNION ALL) whose basic /
anchor child is a
+ // direct bucketed scan that is bucket-pruned by an IN predicate on the
distribution key, so
+ // it only scans a subset of the buckets. The other child is shuffled onto
the anchor's
+ // storage layout and has rows in the buckets the pruned anchor does not
scan. Because the
+ // union is a non-intersect bucket-shuffle set operation,
UnassignedScanBucketOlapTableJob
+ // must fill up receiver instances for those missing buckets; without the
fill-up the other
+ // child's rows in the missing buckets would have no destination instance
and be lost.
+ //
+ // The setup forces the pruned scan to be the anchor: fill_anchor has huge
injected stats so
+ // the IN-filtered branch still wins the largest-row-count basic-child
selection, while
+ // fill_spread has a mismatched bucket count (11 vs 10) so it cannot be
the natural anchor and
+ // must be shuffled. The bucket-shuffle join above the union supplies the
parent hash request
+ // that makes the union choose bucket shuffle, and the join probe
fill_probe has yet another
+ // bucket count (7) so the join is a real shuffle in its own fragment and
does not co-locate a
+ // full-bucket scan into the union fragment (which would otherwise cover
the missing buckets
+ // and hide the fill-up).
+ sql "drop table if exists bucket_shuffle_set_operation_fill_anchor"
+ sql """create table bucket_shuffle_set_operation_fill_anchor(id int)
+ distributed by hash(id) buckets 10
properties('replication_num'='1')"""
+ sql "insert into bucket_shuffle_set_operation_fill_anchor select number
from numbers('number'='20')"
+ sql "drop table if exists bucket_shuffle_set_operation_fill_spread"
+ sql """create table bucket_shuffle_set_operation_fill_spread(id int)
+ distributed by hash(id) buckets 11
properties('replication_num'='1')"""
+ sql "insert into bucket_shuffle_set_operation_fill_spread select number
from numbers('number'='20')"
+ sql "drop table if exists bucket_shuffle_set_operation_fill_probe"
+ sql """create table bucket_shuffle_set_operation_fill_probe(id int)
+ distributed by hash(id) buckets 7
properties('replication_num'='1')"""
+ sql "insert into bucket_shuffle_set_operation_fill_probe select number
from numbers('number'='20')"
+ sql """alter table bucket_shuffle_set_operation_fill_anchor modify column
id
+ set stats ('row_count'='1000000', 'ndv'='20', 'min_value'='0',
'max_value'='19')"""
+ sql """alter table bucket_shuffle_set_operation_fill_spread modify column
id
+ set stats ('row_count'='50', 'ndv'='20', 'min_value'='0',
'max_value'='19')"""
+ sql """alter table bucket_shuffle_set_operation_fill_probe modify column id
+ set stats ('row_count'='30', 'ndv'='20', 'min_value'='0',
'max_value'='19')"""
+ def bucketShuffleUnionFillUpSql = """
+ select t.id from (
+ select id from bucket_shuffle_set_operation_fill_anchor where id
in (0, 2, 4, 6)
+ union all
+ select id from bucket_shuffle_set_operation_fill_spread
+ ) t join[shuffle] bucket_shuffle_set_operation_fill_probe c on t.id =
c.id"""
+ explain {
+ sql "shape plan " + bucketShuffleUnionFillUpSql
+ check { String e ->
+ assertTrue(e.contains("PhysicalUnion[bucketShuffle]"))
+ }
+ }
+ order_qt_bucket_shuffle_union_fill_up bucketShuffleUnionFillUpSql
+
+ // Same missing-bucket fill-up contract, but the pruned basic child
exposes its storage bucket
+ // key only through an equivalent slot: the union's basic child is a
bucket-shuffle join output
+ // that projects the join key bucket_shuffle_set_operation2.id AS k, so
the storage bucket column
+ // (fill_anchor.id) is hidden and only the equivalent k is visible in the
set-operation output.
+ // The alignment proof must resolve the bucket key through its hash
equivalence set (mirroring
+ // ChildrenPropertiesRegulator.canMapBucketKeysToRequire); a direct ExprId
lookup would report
+ // the union as not bucket-aligned, drop the BUCKET_SHUFFLE marker, and
skip
+ // UnassignedScanBucketOlapTableJob.fillUpInstances(), losing the shuffled
side's rows in the
+ // buckets the pruned basic child does not scan.
+ def bucketShuffleEquivalentKeyFillUpSql = """
+ select t.k from (
+ select b.id as k from bucket_shuffle_set_operation_fill_anchor a
+ join[shuffle] bucket_shuffle_set_operation2 b on a.id = b.id
+ where a.id in (0, 2, 4, 6)
+ union all
+ select id from bucket_shuffle_set_operation_fill_spread
+ ) t join[shuffle] bucket_shuffle_set_operation_fill_probe c on t.k =
c.id"""
+ explain {
+ sql "shape plan " + bucketShuffleEquivalentKeyFillUpSql
+ check { String e ->
+ assertTrue(e.contains("PhysicalUnion[bucketShuffle]"))
+ }
+ }
+ order_qt_bucket_shuffle_equivalent_key_fill_up
bucketShuffleEquivalentKeyFillUpSql
+
+ // Same shape but the hidden bucket key has two visible equivalent
set-output columns: the join
+ // chain a.id = b.id = c.id makes fill_anchor.id equivalent to both
projected columns x (c.id)
+ // and y (b.id), and the parent hashes the union output on the later one
(y). The bucket key
+ // alignment must resolve to the same output position across children (the
enforced sibling is
+ // shuffled by y), which is why the proof intersects each child's
candidate equivalent positions
+ // rather than letting the basic child pick its first visible equivalent
(x) and disagree with
+ // the sibling on y.
+ def bucketShuffleTwoEquivalentKeysSql = """
+ select t.x, t.y from (
+ select c.id as x, b.id as y from
bucket_shuffle_set_operation_fill_anchor a
+ join[shuffle] bucket_shuffle_set_operation2 b on a.id = b.id
+ join[shuffle] bucket_shuffle_set_operation3 c on a.id = c.id
+ where a.id in (0, 2, 4, 6)
+ union all
+ select id, id from bucket_shuffle_set_operation_fill_spread
+ ) t join[shuffle] bucket_shuffle_set_operation_fill_probe p on t.y =
p.id"""
+ explain {
+ sql "shape plan " + bucketShuffleTwoEquivalentKeysSql
+ check { String e ->
+ assertTrue(e.contains("PhysicalUnion[bucketShuffle]"))
+ }
+ }
+ order_qt_bucket_shuffle_two_equivalent_keys
bucketShuffleTwoEquivalentKeysSql
+
+ // The basic child candidate can be bucketed by MORE columns than the set
operation distributes
+ // on: a table distributed by hash(k, v) feeding an INTERSECT on only k.
The (k, v) bucket key is
+ // wider than the single-column requirement, so
canMapBucketKeysToRequire() must reject it as the
+ // bucket-shuffle basic and the set operation falls back to execution-hash
shuffle. Without that
+ // guard the planner hits a checkState in calAnotherSideRequiredShuffleIds
and fails to plan. Two
+ // such tables (different bucket counts so neither is a natural colocate
basic) force the fallback
+ // on both sides.
+ sql "drop table if exists bucket_shuffle_set_operation_kv"
+ sql """create table bucket_shuffle_set_operation_kv(k int, v int)
+ distributed by hash(k, v) buckets 10
properties('replication_num'='1')"""
+ sql "insert into bucket_shuffle_set_operation_kv values (1, 1), (2, 2),
(3, 3)"
+ sql "drop table if exists bucket_shuffle_set_operation_kv2"
+ sql """create table bucket_shuffle_set_operation_kv2(k int, v int)
+ distributed by hash(k, v) buckets 11
properties('replication_num'='1')"""
+ sql "insert into bucket_shuffle_set_operation_kv2 values (1, 1), (2, 2),
(3, 3)"
+ def multiColumnBucketKeyIntersectSql = """
+ select k from bucket_shuffle_set_operation_kv
+ intersect
+ select k from bucket_shuffle_set_operation_kv2"""
+ explain {
+ sql "shape plan " + multiColumnBucketKeyIntersectSql
+ check { String e ->
+ assertFalse(e.contains("bucketShuffle"))
+ }
+ }
+ order_qt_multi_column_bucket_key_intersect multiColumnBucketKeyIntersectSql
+
explain {
sql """
select id, id as id2 from (select nullable(id) as id from
bucket_shuffle_set_operation1)a
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
