924060929 commented on code in PR #65129:
URL: https://github.com/apache/doris/pull/65129#discussion_r3541429510
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/RequestPropertyDeriver.java:
##########
@@ -328,25 +329,48 @@ public Void visitPhysicalNestedLoopJoin(
@Override
public Void visitPhysicalSetOperation(PhysicalSetOperation setOperation,
PlanContext context) {
+ // 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. Ask EXECUTION_BUCKETED otherwise.
+ boolean setOperationBucketShuffleAllowed = connectContext != null
Review Comment:
Fixed — the predicate and the downgrade are extracted into shared helpers
(`setOperationBucketShuffleAllowed` /
`downgradeRequireWhenBucketShuffleNotAllowed`) and the union override now
applies the same downgrade to the parent hash request. Verified: with
`enable_nereids_distribute_planner=false` the union no longer chooses bucket
shuffle (the join above may still be a bucket shuffle join, which the legacy
coordinator supports).
##########
regression-test/conf/regression-conf.groovy:
##########
@@ -24,18 +24,18 @@ defaultDb = "regression_test"
// init cmd like: select @@session.tx_read_only
// at each time we connect.
// add allowLoadLocalInfile so that the jdbc can execute mysql load data from
client.
-jdbcUrl =
"jdbc:mysql://127.0.0.1:9030/?useLocalSessionState=true&allowLoadLocalInfile=true&zeroDateTimeBehavior=round"
+jdbcUrl =
"jdbc:mysql://127.0.0.1:29030/?useLocalSessionState=true&allowLoadLocalInfile=true&zeroDateTimeBehavior=round"
Review Comment:
Good catch, that was a local worktree port override committed by mistake —
reverted.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java:
##########
@@ -651,83 +652,105 @@ 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;
+
+ // 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 basic child must have a known storage layout:
e.g. a hash join
+ // output can keep STORAGE_BUCKETED with the layout
cleared to -1
+ // (withShuffleTypeAndForbidColocateJoin). Its bucket
function is
+ // unknown, so the other children cannot be aligned to
it — fall back
+ // to the execution-bucketed shuffle instead.
+ if (childDistribution instanceof DistributionSpecHash
+ && supportBucketShuffleTypes.contains(
+ ((DistributionSpecHash)
childDistribution).getShuffleType())
+ && ((DistributionSpecHash)
childDistribution).getTableId() >= 0
+ &&
!(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 physically re-distributed by the
basic child's
+ // bucket function, so its distribution carries the basic
child's storage
+ // layout (table / index / partitions) instead of its own.
+ // ChildOutputPropertyDeriver relies on this to prove the
bucket-shuffle
+ // alignment structurally: every child of an aligned set
operation shares
+ // the basic child's layout, while an un-enforced
STORAGE_BUCKETED child
+ // coming from another table's bucket layout does not.
+ List<ExprId> shuffleSideIds =
calAnotherSideRequiredShuffleIds(
Review Comment:
Fixed — the basic-child selection now requires every bucket key of the
candidate to be mappable into the child's required hash columns (directly or
through its equivalence sets, mirroring `calAnotherSideRequiredShuffleIds`), so
a child bucketed by (k, v) feeding a set operation on k only falls back to the
execution-bucketed shuffle instead of failing the precondition. Verified with a
(k, v)-bucketed table intersect on k: partitioned plan and correct results.
##########
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:
One addition after re-checking the invariant: the ANY branch of the set-op
regulator enforces NATURAL children to EXECUTION_ANY, so a NATURAL child can
only be observed here after the hash-require branch really aligned the children
— which confirms the simplification is safe for rules 1. The only guard kept is
that rules 1/2 apply when every child is hash-distributed: with a mixed shape
such as an execution-any child plus a storage-bucketed child (reachable through
the ANY child request of union with a nested set operation), the output is not
distributed by that storage layout, so it falls through to the generic
derivation. One edge to confirm: with two nested set operations both reporting
STORAGE_BUCKETED but from different tables' layouts under the ANY request, rule
2 claims the first child's layout — if that matters we may need a layout
equality check there, but it seems unreachable through the current regulator
paths.
--
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]
