This is an automated email from the ASF dual-hosted git repository.
morrySnow 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 d14c6b962ff [fix](local shuffle) make set operation and analytic
advertise the placement their input really is on (#66295)
d14c6b962ff is described below
commit d14c6b962ffcc589a884f238ae703ff91449ded3
Author: 924060929 <[email protected]>
AuthorDate: Mon Aug 3 15:30:48 2026 +0800
[fix](local shuffle) make set operation and analytic advertise the
placement their input really is on (#66295)
Related PR: #65129
Problem Summary:
A `UNION ALL` feeding a window function returned a duplicated `row_number()`
for the same partition key:
SELECT k, v, ROW_NUMBER() OVER (PARTITION BY k ORDER BY v, src) rn
FROM (SELECT k, v, 1 src FROM t_left
UNION ALL
SELECT k, v, 2 src FROM t_right) s
WHERE k IN (2, 4) ORDER BY k, rn;
-- expected -- actual
2 | 20 | 1 2 | 20 | 1
2 | 20 | 2 2 | 20 | 1
4 | 40 | 1 4 | 40 | 1
4 | 40 | 2 4 | 40 | 1
The two duplicated rows are exactly one row per union branch: the branches
of
the same key land in different pipeline tasks and the analytic sink numbers
each task from 1.
Root cause: the generic hash requirement says "I need hash-partitioned
input",
not "I need the same hash function as my siblings". `RequireHash.satisfy()`
accepts GLOBAL_EXECUTION_HASH_SHUFFLE, LOCAL_EXECUTION_HASH_SHUFFLE and
BUCKET_HASH_SHUFFLE alike, and BE implements the first two with
`ShuffleExchanger` (hash modulo the local task count) and the third with
`BucketShuffleExchanger` (storage bucket to instance mapping). So in a
bucket-shuffle union:
- the branch that scans its own buckets is serial under a pooling scan,
claims NOOP, and gets an `LE(LOCAL_EXECUTION_HASH_SHUFFLE)`;
- the branch arriving through a bucket-shuffle exchange claims
`BUCKET_HASH_SHUFFLE`, already satisfies the requirement, and keeps its
placement with no local exchange at all.
The failing plan (`explain distributed plan` — FE-planned local exchanges do
not show up in `explain verbose`, that pass runs after distribute planning):
3:VUNION
|----2:VEXCHANGE <- no local exchange; placement
| comes from the sender's
| BUCKET_SHFFULE_HASH_PARTITIONED
10:VLOCAL-EXCHANGE
| type: LOCAL_EXECUTION_HASH_SHUFFLE <- execution hash
9:VLOCAL-EXCHANGE
| type: PASSTHROUGH
0:VOlapScanNode POOLING-SCAN
The shape only became reachable when #65129 re-enabled the bucket-shuffle
alternative for set operations (union included) and marked every set
operation
with a storage-bucketed child as `DistributionMode.BUCKET_SHUFFLE`; before
that both branches entered a fresh fragment through global hash exchanges,
i.e. one consistent placement. #65129 taught only the intersect / except
branch of `SetOperationNode.enforceAndDeriveLocalExchange` to align on the
bucket function.
Changes:
1. A union that is propagating a hash requirement downward and is itself
colocate / bucket-shuffle now requires `BUCKET_HASH_SHUFFLE` from every
branch and reports it upward — the pattern `HashJoinNode` and the
intersect / except branch already use. Unlike intersect / except a union
does not need its branches aligned for its own semantics, so the bucket
requirement is gated on `hasShuffleForCorrectnessAncestor`: with no
downstream correctness consumer nothing has to move.
2. `SortNode` (analytic sort, colocate) and `AnalyticEvalNode` (partition
by,
no order by, colocate) hand the generic hash requirement to their child
and then hardcoded `LOCAL_EXECUTION_HASH_SHUFFLE` as their own output. A
bucket-distributed child satisfies that requirement and keeps its bucket
placement, so the advertised type was a claim about data that never moved,
and a parent asking for exactly `LOCAL_EXECUTION_HASH_SHUFFLE` (a bucket
join upgraded to local hash) would skip its realign local exchange. Both
now report the placement `enforceRequire` actually produced, matching what
`PartitionSortNode`, `AggregationNode`, `RepeatNode` and `SelectNode`
already do.
3. `SetOperationNode` reports a hash placement only when every branch really
is on it, and NOOP otherwise, so a parent inserts its own aligning exchange
instead of trusting one branch's placement as the whole output's. This only
changes what is advertised, never what is required, so it cannot add a
local exchange.
Tests:
`LocalExchangePlacementAuditTest` is an invariant checker rather than a
shape
snapshot: it walks finished plans and flags any placement-sensitive
multi-input operator (set operation, hash join) whose branches end up on
different hash functions. Adding a shape is one line in its case table.
Running it over 22 shapes x pooling on/off x bucket-upgrade on/off = 88
plans
found 10 mixed-placement plans before this change and 0 after. Those 10 are
five shapes, all of them a window over a bucket-shuffle union: partition-by
with an ORDER BY (the reported case), partition-by without one, unequal
bucket
counts between the branches, three branches, and a nested union. All five
are
now regression cases in `bucket_shuffle_set_operation.groovy`, asserting
that
each window partition contains every `row_number()` exactly once.
Two things the audit disproved and that are therefore *not* claimed here:
aggregation consumers do not reproduce this (the aggregate is pushed below
the
union, so both branches become identical hash exchanges — the consumer has
to
be in the same fragment as the union, which the analytic sort is and a hash
aggregate's finalize phase is not); and pooling is required (with
`ignore_storage_data_distribution=false` none of the 22 shapes mixes).
Verified end to end on a cluster with the BE binary held fixed and only the
FE
swapped: wrong result before, correct after. The new regression cases were
checked against the unmodified FE and fail there with
`duplicated row_number 1 in window partition 2`.
---
.../org/apache/doris/planner/AnalyticEvalNode.java | 8 +-
.../org/apache/doris/planner/SetOperationNode.java | 88 +++++--
.../java/org/apache/doris/planner/SortNode.java | 16 +-
.../planner/LocalShuffleNodeCoverageTest.java | 51 +++++
.../doris/qe/LocalExchangePlacementAuditTest.java | 254 +++++++++++++++++++++
.../apache/doris/qe/LocalExchangePlannerTest.java | 33 +++
.../bucket_shuffle_set_operation.out | 46 ++++
.../bucket_shuffle_set_operation.groovy | 74 +++++-
8 files changed, 535 insertions(+), 35 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/AnalyticEvalNode.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/AnalyticEvalNode.java
index 244913b4055..1698b25495d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/AnalyticEvalNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/AnalyticEvalNode.java
@@ -229,9 +229,13 @@ public class AnalyticEvalNode extends PlanNode {
return Pair.of(this, LocalExchangeType.NOOP);
} else if (orderByElements.isEmpty()) {
if (AddLocalExchange.isColocated(this)) {
+ // requireHash() is the generic hash require, which
BUCKET_HASH_SHUFFLE also
+ // satisfies — a bucket-distributed child then keeps its
bucket placement and no
+ // local exchange is inserted. Leave outputType null so the
real placement is
+ // reported upward; hardcoding LOCAL_EXECUTION_HASH_SHUFFLE
would let a parent
+ // that requires exactly that type (a bucket join upgraded to
local hash) skip
+ // its realign local exchange and pair up mismatched
placements.
requireChild = LocalExchangeTypeRequire.requireHash();
- outputType = AddLocalExchange.resolveExchangeType(
- LocalExchangeTypeRequire.requireHash());
} else {
// Non-colocated analytic with PARTITION BY but no ORDER BY:
// The parent SortNode (mergeByExchange) will insert
PASSTHROUGH above us,
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 1396701d3c7..e1fb75913c8 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
@@ -208,6 +208,14 @@ public abstract class SetOperationNode extends PlanNode {
PlanNode parent, LocalExchangeTypeRequire parentRequire) {
LocalExchangeTypeRequire requireChild;
LocalExchangeType outputType;
+ // 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 the alignment.
+ boolean bucketAligned = AddLocalExchange.isColocated(this) ||
isBucketShuffle();
if (this instanceof UnionNode) {
// Propagate parent's hash requirement to children ONLY when a
downstream operator
// requires shuffle for correctness (not just performance
optimization). Matches BE's
@@ -217,42 +225,72 @@ public abstract class SetOperationNode extends PlanNode {
// intersect/except) through hash/noop links.
// See PlanNode.requiresShuffleForCorrectness() for a
chain-propagation example.
boolean canPropagateHash =
translatorContext.hasShuffleForCorrectnessAncestor(this);
- requireChild = canPropagateHash ? parentRequire.autoRequireHash()
: LocalExchangeTypeRequire.noRequire();
- outputType = canPropagateHash
- ? AddLocalExchange.resolveExchangeType(requireChild)
- : LocalExchangeType.NOOP;
- } else {
- // Intersect / Except
- 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.
+ if (!canPropagateHash) {
+ requireChild = LocalExchangeTypeRequire.noRequire();
+ outputType = LocalExchangeType.NOOP;
+ } else if (bucketAligned) {
+ // A union does not need its branches aligned for its own
semantics — it only
+ // concatenates — but the downstream correctness consumer
does, and the generic
+ // hash require cannot deliver that for a bucket-aligned
union: the branch that
+ // arrives through a bucket-shuffle exchange already satisfies
it and keeps its
+ // bucket placement, while the branch that scans its own
buckets is serial under
+ // a pooling scan, claims NOOP, and gets re-partitioned by
execution hash. The
+ // same key then sits in two different pipeline tasks and the
consumer computes
+ // per-task results — e.g. both rows of a window partition get
row_number()=1.
requireChild = LocalExchangeTypeRequire.requireBucketHash();
outputType = LocalExchangeType.BUCKET_HASH_SHUFFLE;
} else {
- // PARTITIONED intersect/except: all children enter via global
hash
- // exchange. Require GLOBAL so any inserted exchange matches
the
- // cross-fragment instance mapping (same fix as HashJoinNode
DORIS-26101).
- // Exception: serial source → fall back to LOCAL (DORIS-26120).
- boolean serialSource = fragment != null
- &&
fragment.useSerialSource(translatorContext.getConnectContext());
- requireChild = serialSource
- ? LocalExchangeTypeRequire.requireHash()
- :
LocalExchangeTypeRequire.requireGlobalExecutionHash();
+ requireChild = parentRequire.autoRequireHash();
outputType =
AddLocalExchange.resolveExchangeType(requireChild);
}
+ } else if (bucketAligned) {
+ // Intersect / Except, colocate or bucket shuffle. Unlike a union
these always need
+ // their children aligned, so there is no shuffle-for-correctness
gate here.
+ requireChild = LocalExchangeTypeRequire.requireBucketHash();
+ outputType = LocalExchangeType.BUCKET_HASH_SHUFFLE;
+ } else {
+ // PARTITIONED intersect/except: all children enter via global hash
+ // exchange. Require GLOBAL so any inserted exchange matches the
+ // cross-fragment instance mapping, same as HashJoinNode's
partitioned branch.
+ // Exception: a serial source sends to a single BE, so its
+ // shuffle_idx_to_instance_idx has only one entry and GLOBAL would
route rows to
+ // indices that do not exist — fall back to the generic hash
require, which
+ // resolves to LOCAL.
+ boolean serialSource = fragment != null
+ &&
fragment.useSerialSource(translatorContext.getConnectContext());
+ requireChild = serialSource
+ ? LocalExchangeTypeRequire.requireHash()
+ : LocalExchangeTypeRequire.requireGlobalExecutionHash();
+ outputType = AddLocalExchange.resolveExchangeType(requireChild);
}
ArrayList<PlanNode> newChildren = Lists.newArrayList();
+ LocalExchangeType branchPlacement = null;
+ boolean branchesAgree = true;
for (int i = 0; i < children.size(); i++) {
- newChildren.add(enforceRequire(translatorContext, children.get(i),
i, requireChild).first);
+ Pair<PlanNode, LocalExchangeType> branch
+ = enforceRequire(translatorContext, children.get(i), i,
requireChild);
+ newChildren.add(branch.first);
+ if (i == 0) {
+ branchPlacement = branch.second;
+ } else if (branchPlacement != branch.second) {
+ branchesAgree = false;
+ }
}
this.children = newChildren;
+
+ // Only advertise a hash placement the branches really are on.
requireBucketHash /
+ // requireGlobalExecutionHash pin the branches to one type, so those
branches keep the
+ // outputType computed above; the generic requireHash is satisfied by
GLOBAL / LOCAL /
+ // BUCKET alike, so a branch may keep an existing placement and the
hardcoded type would
+ // be a claim about data that never moved. Report what the branches
actually agreed on,
+ // and NOOP when they did not — a parent that needs one placement then
inserts its own
+ // local exchange instead of trusting one branch's placement as the
whole output's.
+ if (outputType.isHashShuffle() && !(branchesAgree && branchPlacement
== outputType)) {
+ outputType = branchesAgree && branchPlacement != null &&
branchPlacement.isHashShuffle()
+ ? branchPlacement
+ : LocalExchangeType.NOOP;
+ }
return Pair.of(this, outputType);
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/SortNode.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/SortNode.java
index 1df6d30110d..716dc1739f3 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/SortNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/SortNode.java
@@ -271,13 +271,15 @@ public class SortNode extends PlanNode {
// BE: SortSink._is_analytic_sort=true →
required_data_distribution() = HASH.
// This sort serves a parent AnalyticEvalNode (window function)
and requires
// data partitioned by the analytic's partition exprs.
- if (AddLocalExchange.isColocated(this)) {
- requireChild = LocalExchangeTypeRequire.requireHash();
- outputType = AddLocalExchange.resolveExchangeType(
- LocalExchangeTypeRequire.requireHash());
- } else {
- requireChild = parentRequire.autoRequireHash();
- }
+ // requireHash() is the generic hash require: BUCKET_HASH_SHUFFLE
satisfies it too,
+ // so a bucket-distributed child keeps its bucket placement and no
local exchange is
+ // inserted. outputType must therefore stay null and be derived
from enforceResult —
+ // hardcoding LOCAL_EXECUTION_HASH_SHUFFLE here would advertise a
placement the data
+ // is not on, and a parent asking for exactly
LOCAL_EXECUTION_HASH_SHUFFLE (a bucket
+ // join upgraded to local hash) would skip its realign local
exchange.
+ requireChild = AddLocalExchange.isColocated(this)
+ ? LocalExchangeTypeRequire.requireHash()
+ : parentRequire.autoRequireHash();
} else if (mergeByexchange) {
// BE: SortSink._merge_by_exchange=true →
required_data_distribution() = PASSTHROUGH.
requireChild = LocalExchangeTypeRequire.requirePassthrough();
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 e171ebe166c..02e68ba6b59 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
@@ -609,9 +609,60 @@ public class LocalShuffleNodeCoverageTest {
ctx.setHasShuffleForCorrectnessAncestor(unionNode, true);
Pair<PlanNode, LocalExchangeType> unionOutput =
unionNode.enforceAndDeriveLocalExchange(
ctx, null, LocalExchangeTypeRequire.requireHash());
+ // The single branch really is re-partitioned by
LOCAL_EXECUTION_HASH_SHUFFLE (it claimed
+ // NOOP, so enforceRequire inserted the exchange), so advertising that
type is truthful.
Assertions.assertEquals(LocalExchangeType.LOCAL_EXECUTION_HASH_SHUFFLE,
unionOutput.second);
Assertions.assertEquals(LocalExchangeNode.RequireHash.class,
unionChild.lastRequire.getClass());
+ // A branch that already is on a hash placement is reported as-is
instead of being
+ // relabelled: the union claimed LOCAL_EXECUTION_HASH_SHUFFLE
unconditionally before, which
+ // was a claim about data that never moved.
+ UnionNode passthroughUnion = new UnionNode(nextPlanNodeId(), new
TupleId(NEXT_ID.getAndIncrement()));
+ TrackingPlanNode globalHashChild = new
TrackingPlanNode(nextPlanNodeId(),
+ LocalExchangeType.GLOBAL_EXECUTION_HASH_SHUFFLE);
+ passthroughUnion.addChild(globalHashChild);
+ ctx.setHasShuffleForCorrectnessAncestor(passthroughUnion, true);
+ Pair<PlanNode, LocalExchangeType> passthroughOutput = passthroughUnion
+ .enforceAndDeriveLocalExchange(ctx, null,
LocalExchangeTypeRequire.requireHash());
+
Assertions.assertEquals(LocalExchangeType.GLOBAL_EXECUTION_HASH_SHUFFLE,
passthroughOutput.second);
+ Assertions.assertSame(globalHashChild, passthroughUnion.getChild(0));
+
+ // Bucket-shuffle UnionNode under a shuffle-for-correctness consumer:
the branches are
+ // aligned by the basic child's storage bucket function, so the union
must require
+ // BUCKET_HASH_SHUFFLE from every branch instead of the generic hash.
With the generic
+ // require the branch arriving through a bucket-shuffle exchange
satisfies it and keeps
+ // its bucket placement while a serial (NOOP-claim) branch is
re-partitioned by execution
+ // hash, splitting one key across two pipeline tasks (duplicate
row_number()=1).
+ UnionNode bucketUnion = new UnionNode(nextPlanNodeId(), new
TupleId(NEXT_ID.getAndIncrement()));
+ bucketUnion.setColocate(false);
+ bucketUnion.setDistributionMode(DistributionMode.BUCKET_SHUFFLE);
+ TrackingPlanNode bucketUnionLeft = new
TrackingPlanNode(nextPlanNodeId(), LocalExchangeType.NOOP);
+ TrackingPlanNode bucketUnionRight = new
TrackingPlanNode(nextPlanNodeId(),
+ LocalExchangeType.BUCKET_HASH_SHUFFLE);
+ bucketUnion.addChild(bucketUnionLeft);
+ bucketUnion.addChild(bucketUnionRight);
+ ctx.setHasShuffleForCorrectnessAncestor(bucketUnion, true);
+ Pair<PlanNode, LocalExchangeType> bucketUnionOutput =
bucketUnion.enforceAndDeriveLocalExchange(
+ ctx, null, LocalExchangeTypeRequire.requireHash());
+ Assertions.assertEquals(LocalExchangeType.BUCKET_HASH_SHUFFLE,
bucketUnionOutput.second);
+ // the serial branch is re-aligned by bucket hash ...
+ assertChildLocalExchangeType(bucketUnion, 0,
LocalExchangeType.BUCKET_HASH_SHUFFLE);
+ // ... and the branch that already is bucket-distributed keeps its
placement untouched
+ Assertions.assertSame(bucketUnionRight, bucketUnion.getChild(1));
+
+ // Without a downstream shuffle-for-correctness consumer a
bucket-shuffle union still
+ // requires nothing: UNION ALL only concatenates, so no branch has to
move.
+ UnionNode bucketUnionNoConsumer = new UnionNode(nextPlanNodeId(),
+ new TupleId(NEXT_ID.getAndIncrement()));
+
bucketUnionNoConsumer.setDistributionMode(DistributionMode.BUCKET_SHUFFLE);
+ TrackingPlanNode noConsumerChild = new
TrackingPlanNode(nextPlanNodeId(),
+ LocalExchangeType.BUCKET_HASH_SHUFFLE);
+ bucketUnionNoConsumer.addChild(noConsumerChild);
+ Pair<PlanNode, LocalExchangeType> noConsumerOutput =
bucketUnionNoConsumer.enforceAndDeriveLocalExchange(
+ ctx, null, LocalExchangeTypeRequire.requireHash());
+ Assertions.assertEquals(LocalExchangeType.NOOP,
noConsumerOutput.second);
+ Assertions.assertSame(noConsumerChild,
bucketUnionNoConsumer.getChild(0));
+
IntersectNode intersectNode = new IntersectNode(nextPlanNodeId(), new
TupleId(NEXT_ID.getAndIncrement()));
intersectNode.setColocate(false);
TrackingScanNode left = new TrackingScanNode(nextPlanNodeId(),
LocalExchangeType.NOOP);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/LocalExchangePlacementAuditTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/LocalExchangePlacementAuditTest.java
new file mode 100644
index 00000000000..5e231e1a243
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/qe/LocalExchangePlacementAuditTest.java
@@ -0,0 +1,254 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.qe;
+
+import org.apache.doris.nereids.NereidsPlanner;
+import org.apache.doris.planner.ExchangeNode;
+import org.apache.doris.planner.HashJoinNode;
+import org.apache.doris.planner.LocalExchangeNode;
+import org.apache.doris.planner.LocalExchangeNode.LocalExchangeType;
+import org.apache.doris.planner.OlapScanNode;
+import org.apache.doris.planner.PlanFragment;
+import org.apache.doris.planner.PlanNode;
+import org.apache.doris.planner.SetOperationNode;
+import org.apache.doris.thrift.TPartitionType;
+import org.apache.doris.utframe.TestWithFeService;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Audit: for every multi-input operator that depends on hash placement, all
of its input
+ * branches must end up on the SAME placement function.
+ *
+ * <p>Motivation: {@code RequireHash} means "I need hash-partitioned input";
it is satisfied by
+ * GLOBAL_EXECUTION_HASH_SHUFFLE, LOCAL_EXECUTION_HASH_SHUFFLE and
BUCKET_HASH_SHUFFLE alike.
+ * It does NOT mean "all my branches must use the same hash function". A
multi-input operator
+ * that hands the generic require to every branch can therefore end up with
one branch keeping
+ * a storage-bucket placement while another is re-partitioned by execution
hash — the same key
+ * then sits in two different pipeline tasks. This audit walks the finished
plan and flags that
+ * mix instead of relying on a reviewer noticing it.
+ */
+public class LocalExchangePlacementAuditTest extends TestWithFeService {
+ @Override
+ protected int backendNum() {
+ return 3;
+ }
+
+ @Override
+ protected void runBeforeAll() throws Exception {
+ createDatabase("test");
+ useDatabase("test");
+ // b1/b2 share bucket count so the optimizer may bucket-shuffle one
onto the other;
+ // b3 has a different bucket count so it must be re-shuffled.
+ createTable("CREATE TABLE test.b1 (k INT, k2 INT, v INT) DISTRIBUTED
BY HASH(k) BUCKETS 6 "
+ + "PROPERTIES ('replication_num'='1')");
+ createTable("CREATE TABLE test.b2 (k INT, k2 INT, v INT) DISTRIBUTED
BY HASH(k) BUCKETS 6 "
+ + "PROPERTIES ('replication_num'='1')");
+ createTable("CREATE TABLE test.b3 (k INT, k2 INT, v INT) DISTRIBUTED
BY HASH(k) BUCKETS 7 "
+ + "PROPERTIES ('replication_num'='1')");
+ }
+
+ /** The placement a branch actually lands on, as observed from the
finished plan tree. */
+ private static LocalExchangeType effectivePlacement(PlanNode node,
ConnectContext ctx) {
+ if (node instanceof LocalExchangeNode) {
+ LocalExchangeType type = ((LocalExchangeNode)
node).getExchangeType();
+ // A PASSTHROUGH/BROADCAST/... wrapper does not decide hash
placement; look through it.
+ return type.isHashShuffle() ? type :
effectivePlacement(node.getChild(0), ctx);
+ }
+ if (node instanceof ExchangeNode) {
+ TPartitionType partitionType = ((ExchangeNode)
node).getPartitionType();
+ if (partitionType == TPartitionType.HASH_PARTITIONED) {
+ return LocalExchangeType.GLOBAL_EXECUTION_HASH_SHUFFLE;
+ }
+ if (partitionType ==
TPartitionType.BUCKET_SHFFULE_HASH_PARTITIONED) {
+ return LocalExchangeType.BUCKET_HASH_SHUFFLE;
+ }
+ return LocalExchangeType.NOOP;
+ }
+ if (node instanceof OlapScanNode) {
+ // Mirrors OlapScanNode.enforceAndDeriveLocalExchange: a pooling
(serial) scan claims
+ // nothing, a non-pooling bucket scan claims its storage bucket
distribution.
+ boolean pooling = node.getFragment() != null &&
node.getFragment().useSerialSource(ctx);
+ return pooling ? LocalExchangeType.NOOP :
LocalExchangeType.BUCKET_HASH_SHUFFLE;
+ }
+ // Pass-through operators report their child's placement upward.
+ if (node.getChildren().size() == 1) {
+ return effectivePlacement(node.getChild(0), ctx);
+ }
+ return LocalExchangeType.NOOP;
+ }
+
+ /** Returns "" when the plan is consistent, otherwise a description of the
mixed placement. */
+ private static String findMixedPlacement(List<PlanFragment> fragments,
ConnectContext ctx) {
+ StringBuilder problems = new StringBuilder();
+ for (PlanFragment fragment : fragments) {
+ walk(fragment.getPlanRoot(), problems, ctx);
+ }
+ return problems.toString();
+ }
+
+ private static void walk(PlanNode node, StringBuilder problems,
ConnectContext ctx) {
+ boolean placementSensitive = node instanceof SetOperationNode || node
instanceof HashJoinNode;
+ if (placementSensitive && node.getChildren().size() > 1) {
+ Map<LocalExchangeType, List<Integer>> byPlacement = new
LinkedHashMap<>();
+ for (int i = 0; i < node.getChildren().size(); i++) {
+ LocalExchangeType placement =
effectivePlacement(node.getChild(i), ctx);
+ if (placement.isHashShuffle()) {
+ byPlacement.computeIfAbsent(placement, k -> new
ArrayList<>()).add(i);
+ }
+ }
+ if (byPlacement.size() > 1) {
+ problems.append(node.getClass().getSimpleName())
+ .append('(').append(node.getId().asInt()).append(")
mixes ")
+ .append(byPlacement).append('\n');
+ }
+ }
+ for (PlanNode child : node.getChildren()) {
+ walk(child, problems, ctx);
+ }
+ }
+
+ private String audit(String label, String sql, boolean pooling, boolean
bucketUpgrade)
+ throws Exception {
+ SessionVariable sv = connectContext.getSessionVariable();
+ sv.setEnableLocalShufflePlanner(true);
+ sv.setEnableLocalShuffle(true);
+ sv.setEnableNereidsDistributePlanner(true);
+ // ignore_storage_data_distribution is the real pooling switch:
useSerialSource() gates on
+ // it. force_to_local_shuffle only forces ScanNode.isSerialNode(),
which a small table
+ // already satisfies via `scanRangeNum < parallelExecInstanceNum *
numScanBackends`, so
+ // flipping it alone would not give a non-pooling arm at all.
+ sv.setIgnoreStorageDataDistribution(pooling);
+ sv.setForceToLocalShuffle(pooling);
+ sv.setPipelineTaskNum(bucketUpgrade ? "16" : "8");
+ sv.setBucketShuffleDowngradeRatio(0);
+ // ratio <= 1 disables the upgrade entirely; 1.01 makes it fire
whenever instances
+ // slightly exceed buckets-with-data, which is what a bucket join
above a mis-claiming
+ // child needs in order to be fooled by that claim.
+ sv.setLocalShuffleBucketUpgradeRatio(bucketUpgrade ? 1.01 : 1.5);
+ sv.disableColocatePlan = true;
+
+ StmtExecutor executor = executeNereidsSql("explain distributed plan "
+ sql);
+ NereidsPlanner planner = (NereidsPlanner) executor.planner();
+ String problems = findMixedPlacement(planner.getFragments(),
connectContext);
+ return problems.isEmpty() ? "" : "MIXED | pooling=" + pooling + "
upgrade=" + bucketUpgrade
+ + " | " + label + " | " + problems.trim().replace('\n', ';');
+ }
+
+ @Test
+ public void auditPlacementConsistency() {
+ List<String> failures = new ArrayList<>();
+ // set operation kind x consumer kind x which side needs re-shuffling
+ String[][] cases = {
+ {"union+window(partition by bucket key)",
+ "select k, row_number() over (partition by k order by v) from "
+ + "(select k, v from test.b1 union all select k, v from
test.b2) u"},
+ {"union+window(partition by non-bucket key)",
+ "select k2, row_number() over (partition by k2 order by v)
from "
+ + "(select k2, v from test.b1 union all select k2, v from
test.b2) u"},
+ {"union+window(no order by)",
+ "select k, row_number() over (partition by k) from "
+ + "(select k from test.b1 union all select k from test.b2)
u"},
+ {"union+agg",
+ "select k, count(*) from (select k from test.b1 union all
select k from test.b2) u group by k"},
+ {"union+agg(distinct)",
+ "select k, count(distinct v) from "
+ + "(select k, v from test.b1 union all select k, v from
test.b2) u group by k"},
+ {"union+shuffle join",
+ "select u.k from (select k from test.b1 union all select k
from test.b2) u "
+ + "join[shuffle] test.b3 t on u.k = t.k"},
+ {"union+bucket join",
+ "select u.k from (select k from test.b1 union all select k
from test.b2) u "
+ + "join test.b1 t on u.k = t.k"},
+ {"union of different bucket counts + window",
+ "select k, row_number() over (partition by k order by v) from "
+ + "(select k, v from test.b1 union all select k, v from
test.b3) u"},
+ {"3-way union + window",
+ "select k, row_number() over (partition by k order by v) from "
+ + "(select k, v from test.b1 union all select k, v from
test.b2 "
+ + "union all select k, v from test.b3) u"},
+ {"union(scan, values) + window",
+ "select k, row_number() over (partition by k order by v) from "
+ + "(select k, v from test.b1 union all select 1, 2) u"},
+ {"intersect+window",
+ "select k, row_number() over (partition by k order by k) from "
+ + "(select k from test.b1 intersect select k from test.b2)
u"},
+ {"except+window",
+ "select k, row_number() over (partition by k order by k) from "
+ + "(select k from test.b1 except select k from test.b2)
u"},
+ {"intersect(join as basic child)+window",
+ "select k, row_number() over (partition by k order by k) from "
+ + "(select a.k from test.b1 a join test.b2 b on a.k=b.k
intersect "
+ + "select k from test.b3) u"},
+ {"nested union under intersect + window",
+ "select k, row_number() over (partition by k order by k) from "
+ + "((select k from test.b1 union all select k from
test.b2) "
+ + "intersect select k from test.b3) u"},
+ {"union under union + window",
+ "select k, row_number() over (partition by k order by v) from "
+ + "(select k, v from test.b1 union all "
+ + "(select k, v from test.b2 union all select k, v from
test.b3)) u"},
+ {"window(partition by bucket key) under bucket join",
+ "select w.k from (select k, row_number() over (partition by k)
rn from test.b1) w "
+ + "join test.b1 t on w.k = t.k"},
+ {"window(partition by bucket key, no order by) under shuffle join",
+ "select w.k from (select k, row_number() over (partition by k)
rn from test.b1) w "
+ + "join[shuffle] test.b3 t on w.k = t.k"},
+ {"window(partition by bucket key, order by) under bucket join",
+ "select w.k from (select k, row_number() over (partition by k
order by v) rn "
+ + "from test.b1) w join test.b1 t on w.k = t.k"},
+ {"window(no order by) under bucket join on 2 tables",
+ "select w.k from (select k, row_number() over (partition by k)
rn from test.b1) w "
+ + "join test.b2 t on w.k = t.k"},
+ {"intersect under bucket join",
+ "select u.k from (select k from test.b1 intersect select k
from test.b3) u "
+ + "join test.b1 t on u.k = t.k"},
+ {"union under bucket join under window",
+ "select k, row_number() over (partition by k order by c) from "
+ + "(select u.k k, count(*) c from "
+ + "(select k from test.b1 union all select k from test.b2)
u "
+ + "join test.b1 t on u.k = t.k group by u.k) x"},
+ {"agg over union under bucket join",
+ "select u.k from (select k, count(*) c from "
+ + "(select k from test.b1 union all select k from test.b2)
x group by k) u "
+ + "join test.b1 t on u.k = t.k"},
+ };
+ for (boolean pooling : new boolean[] {true, false}) {
+ for (boolean bucketUpgrade : new boolean[] {false, true}) {
+ for (String[] c : cases) {
+ try {
+ String failure = audit(c[0], c[1], pooling,
bucketUpgrade);
+ if (!failure.isEmpty()) {
+ failures.add(failure);
+ }
+ } catch (Exception e) {
+ failures.add("PLANFAIL | pooling=" + pooling + "
upgrade=" + bucketUpgrade
+ + " | " + c[0] + " | " + e);
+ }
+ }
+ }
+ }
+ Assertions.assertTrue(failures.isEmpty(), String.join("\n", failures));
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/LocalExchangePlannerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/LocalExchangePlannerTest.java
index 66e7578f0fa..275df40cffa 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/qe/LocalExchangePlannerTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/qe/LocalExchangePlannerTest.java
@@ -531,6 +531,39 @@ public class LocalExchangePlannerTest extends
TestWithFeService implements PlanS
olapScan("t1")))))))));
}
+ @Test
+ public void testBucketShuffleUnionUnderWindowStaysBucketAligned() throws
Exception {
+ // Regression: UNION ALL feeding a window function returned duplicate
row_number()=1 for
+ // the same partition key. The union is bucket-shuffled (t1 keeps its
storage buckets,
+ // t2 arrives through a bucket-shuffle exchange onto them), and the
analytic sort below
+ // the AnalyticEval requires hash-partitioned input. With the generic
hash requirement
+ // the t2 branch satisfied it and kept its bucket placement while the
t1 branch — serial
+ // under the pooling scan, so it claims no distribution — was
re-partitioned by
+ // LOCAL_EXECUTION_HASH_SHUFFLE. The two placements put one partition
key in two
+ // different pipeline tasks and the analytic numbered each task from 1.
+ //
+ // AnalyticEval ← Sort ← Union ← LE(BUCKET_HASH) ← LE(PT) ← scan(t1)
+ // ← Exchange (bucket shuffle) ← scan(t2)
+ setupLocalShuffleSession(sv -> {
+ sv.setForceToLocalShuffle(true);
+ sv.setBucketShuffleDowngradeRatio(0);
+ });
+ String sql = "select k1, row_number() over (partition by k1 order by
k2) from ("
+ + "select k1, k2 from test.t1 union all select k1, k2 from
test.t2) u";
+ assertPlanShape(sql,
+ anyTree(
+ analytic(
+ sort(
+ union(
+ localExchange(BUCKET_HASH,
+ localExchange(PT,
+
olapScan("t1"))),
+ anyTree(exchange()))))));
+ // The mixed-placement signature of the bug: an execution-hash local
exchange sitting
+ // next to a bucket-distributed sibling branch.
+ assertNoLocalExchangeOfType(sql,
LocalExchangeType.LOCAL_EXECUTION_HASH_SHUFFLE);
+ }
+
@Test
public void testGroupingSetsPlanContainsHashShuffle() throws Exception {
// Non-pooling grouping sets keeps the colocated BUCKET_HASH_SHUFFLE
output of
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 1ec75925ec2..dcb9d11753b 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
@@ -161,6 +161,52 @@ PhysicalResultSink
3
3
+-- !union_under_window --
+1 1 1
+1 1 2
+2 2 1
+2 2 2
+3 3 1
+3 3 2
+
+-- !union_under_window_no_order_by --
+1 1 1
+1 1 2
+2 2 1
+2 2 2
+3 3 1
+3 3 2
+
+-- !union_unequal_buckets_under_window --
+1 1 1
+1 1 2
+2 2 1
+2 2 2
+3 3 1
+3 3 2
+
+-- !three_way_union_under_window --
+1 1 1
+1 1 2
+1 1 3
+2 2 1
+2 2 2
+2 2 3
+3 3 1
+3 3 2
+3 3 3
+
+-- !nested_union_under_window --
+1 1 1
+1 1 2
+1 1 3
+2 2 1
+2 2 2
+2 2 3
+3 3 1
+3 3 2
+3 3 3
+
-- !bucket_shuffle_when_local_shuffle_off_shape --
PhysicalResultSink
--PhysicalIntersect[bucketShuffle]
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 ee118bccbd5..6f10c2ca96b 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
@@ -116,6 +116,78 @@ suite("bucket_shuffle_set_operation") {
intersect
select id from bucket_shuffle_set_operation2)""")
+ // A window function above a bucket-shuffle UNION ALL is a
shuffle-for-correctness consumer:
+ // the analytic sink needs every row of a window partition inside one
pipeline task. The union
+ // branches are aligned by the basic child's storage bucket function, so
the union must require
+ // a bucket-hash local exchange from every branch. Requiring the generic
hash instead let the
+ // branch arriving through a bucket-shuffle exchange keep its bucket
placement while the branch
+ // scanning its own buckets (serial under a pooling scan, so it claims no
distribution) was
+ // re-partitioned by execution hash: one window partition ended up split
across two pipeline
+ // tasks and row_number() restarted at 1 in each.
+ // force_to_local_shuffle pins the pooling scan so the shape does not
depend on the backend
+ // count of the environment running this suite.
+ // The variants below were not guessed: LocalExchangePlacementAuditTest
walks finished plans and
+ // flags any set operation whose branches end up on different hash
placements, and these are
+ // every shape it flagged — partition-by with and without ORDER BY,
unequal bucket counts,
+ // three branches, and a nested union.
+ sql "set force_to_local_shuffle=true"
+
+ // The ordered golden results prove both the row count and that every
window partition
+ // contains each row_number() exactly once. A partition split across
pipeline tasks would
+ // produce duplicate row_number values inside one id.
+ order_qt_union_under_window """
+ select id, value, row_number() over (partition by id order by value) rn
+ from (
+ select id, value from bucket_shuffle_set_operation1
+ union all
+ select id, value from bucket_shuffle_set_operation2
+ ) u
+ order by id, value, rn"""
+
+ order_qt_union_under_window_no_order_by """
+ select id, value, row_number() over (partition by id) rn
+ from (
+ select id, value from bucket_shuffle_set_operation1
+ union all
+ select id, value from bucket_shuffle_set_operation2
+ ) u
+ order by id, value, rn"""
+
+ // the branches have different bucket counts, so one must be re-shuffled
onto the other's
+ // buckets rather than keeping its own
+ order_qt_union_unequal_buckets_under_window """
+ select id, value, row_number() over (partition by id order by value) rn
+ from (
+ select id, value from bucket_shuffle_set_operation1
+ union all
+ select id, value from bucket_shuffle_set_operation3
+ ) u
+ order by id, value, rn"""
+
+ order_qt_three_way_union_under_window """
+ select id, value, row_number() over (partition by id order by value) rn
+ from (
+ select id, value from bucket_shuffle_set_operation1
+ union all
+ select id, value from bucket_shuffle_set_operation2
+ union all
+ select id, value from bucket_shuffle_set_operation3
+ ) u
+ order by id, value, rn"""
+
+ order_qt_nested_union_under_window """
+ select id, value, row_number() over (partition by id order by value) rn
+ from (
+ select id, value from bucket_shuffle_set_operation1
+ union all
+ (select id, value from bucket_shuffle_set_operation2
+ union all
+ select id, value from bucket_shuffle_set_operation3)
+ ) u
+ order by id, value, rn"""
+
+ sql "set force_to_local_shuffle=false"
+
// 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"
@@ -439,4 +511,4 @@ suite("bucket_shuffle_set_operation") {
assertTrue(checked)
}
}
-}
\ No newline at end of file
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
