github-actions[bot] commented on code in PR #65129:
URL: https://github.com/apache/doris/pull/65129#discussion_r3541350925
##########
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:
This looks like a worktree-local port override accidentally committed into
the default regression config. The standard regression runner expects the
source FE on the default ports (`9030`/`9020`/`8030`), and the top of this file
explicitly warns against committing local config changes. With this patch, CI
or a normal local run will try to connect to `127.0.0.1:29030` instead. Please
revert these config edits and keep per-worktree port offsets outside the PR.
##########
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:
The new gate only protects the generic `visitPhysicalSetOperation` path.
`PhysicalUnion` overrides this visitor below and still forwards a parent hash
request with `createHashRequestAccordingToParent(...)` without running the same
downgrade. Because that helper preserves the parent's `ShuffleType.REQUIRE`,
`ChildrenPropertiesRegulator` can still choose bucket-shuffle union children
when `enable_local_shuffle=true && enable_local_shuffle_planner=false` or when
the Nereids distribute planner is disabled, which is exactly the mode this
block says must fall back to `EXECUTION_BUCKETED`. Please share this
predicate/downgrade with the union override as well.
##########
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:
This branch can still choose a storage basic child whose bucket key is wider
than the set operation output. For example, if the basic child comes from a
table bucketed by `(k, v)` but the set operation is `INTERSECT(k)`,
`PhysicalProject` can preserve the child's NATURAL/STORAGE hash spec with
ordered columns `[k, v]` even though the set-op request built from regular
outputs only contains `[k]`. `calAnotherSideRequiredShuffleIds` then iterates
both storage keys and hits `Preconditions.checkState(index != -1)` for `v`, so
planning fails instead of falling back to execution-bucketed shuffle. Please
require the chosen basic child's full bucket key to be mappable through the
set-op output/request before taking this bucket-shuffle branch, or downgrade to
the partitioned path when it is not.
--
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]
