This is an automated email from the ASF dual-hosted git repository.
morningman 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 52362a455cd [fix](local shuffle) Keep local exchanges before
pre-aggregations over key-breaking children (#70296)
52362a455cd is described below
commit 52362a455cd0a584431648f2ff16d429ec3bb8d1
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Thu Nov 5 17:26:44 2026 +0000
[fix](local shuffle) Keep local exchanges before pre-aggregations over
key-breaking children (#70296)
### Background
On master the FE plans the local exchanges that redistribute rows among
the pipeline tasks of one BE (`enable_local_shuffle_planner`, on by
default). Each plan node follows the rule of the BE operator it becomes,
`required_data_distribution()`, which BE applies itself when the FE
planner is off. `enable_local_exchange_before_agg` (default true)
controls the exchange before a LOCAL (first-phase) aggregation. Since
#63766, BE skips it with the switch off only while the child keeps the
grouping-key distribution; a child breaks it when it is serial or asks
for a non-hash local exchange itself, such as the probe side of a nested
loop join or a table function.
### What problem does this PR solve?
Issue Number: None
Related PR: #63366 (introduced the FE rule without the child check),
#63766 (the BE rule), #70177 (the same fix for streaming aggregation;
this PR reuses its `childBreaksLocalKeyDistribution()`), #62438 (added
the switch)
PR Type: Bugfix
Problem Summary:
With `enable_local_exchange_before_agg=false`, the FE planner dropped
the local exchange below a LOCAL `AggSink` and below a streaming
distinct pre-aggregation whatever their child was, while BE keeps it
after a child that breaks the key distribution. Queries planned by the
FE therefore got fewer local exchanges than BE planning gives, for
example no hash exchange between a nested loop join and the
pre-aggregation above it. This PR adds BE's child check to both cases.
Results do not change; most of the diff is tests.
### Details
- **Change map:** production code +15 / -7 in `AggregationNode` (two
conditions and their comments) and a `PlanNode` javadoc; tests +345 /
-34: cases in `LocalShuffleNodeCoverageTest` and
`LocalExchangePlannerTest`, and the new regression suite
`nereids_p0/local_shuffle/test_local_exchange_before_agg_breaking_child`
with its generated `.out` (+169).
- **Frequency:** Always, when `enable_local_exchange_before_agg=false`
(not the default; P0 randomizes it) and a LOCAL aggregation sits on a
child that breaks the key distribution. Only plans and speed change: the
FE always requires hash input at FIRST_MERGE and at the non-streaming
DISTINCT dedup (a deliberate divergence from BE), so results stay right.
- **Root cause:** `AggregationNode.enforceAndDeriveLocalExchange()`
mirrors the BE operators, but its DistinctStreamingAgg and AggSink LOCAL
branches fall back to the base class on `!enableLeBeforeAgg` alone,
without BE's `child_breaks_local_key_distribution()`. #63366 wrote them
after #63766 had merged, from the rule before it.
- **Why existing tests missed it:** Unit tests: the two "LE disabled"
cases in `LocalShuffleNodeCoverageTest` use a child that keeps the key
distribution, no case reached the DistinctStreamingAgg branch, and
`LocalExchangePlannerTest` had no plan with the switch off. Regression:
`query_p0/join/test_agg_after_nested_loop_join_local_exchange` (#63766's
suite) and
`nereids_p0/local_shuffle/test_enable_local_exchange_before_agg` compare
only results, which were right, and no `.out` shows local exchanges.
- **Fix:** Both branches fall back to the base class only when
`childBreaksLocalKeyDistribution()` is false. After a key-breaking
child, the AggSink LOCAL phase goes on to its hash requirement, and the
streaming distinct pre-aggregation to
`enable_distinct_streaming_agg_force_passthrough` (PASSTHROUGH, else the
base class), as on BE; it never asks for hash.
- **Other paths and branches:** Streaming aggregation got the check in
#70177; BE needs no change. One known difference from BE planning
remains, with the switch on or off: the FE lets an ADAPTIVE_PASSTHROUGH
or PASSTHROUGH child output satisfy a PASSTHROUGH requirement, where BE
`Pipeline::need_to_local_exchange()` always adds one (last row of the
table in Testing). branch-4.1 and branch-4.2 have no FE local shuffle
planner and are not affected.
### Testing
- `LocalShuffleNodeCoverageTest` and `LocalExchangePlannerTest`, two new
cases each: all four fail without the fix (for example, a serial child
got `PASSTHROUGH` instead of `LOCAL_EXECUTION_HASH_SHUFFLE`) and pass
with it. The 12 fe-core test classes that plan local exchanges or print
distributed plans pass (168 tests). Run locally.
- A planner probe over 288 plans (18 query shapes × pooling on and off ×
`agg_phase` 0 and 2 × `disable_streaming_preaggregations` on and off ×
the switch): only 32 with the switch off change. 30 of them now equal
the plan with the switch on; the other 2 keep the distribution of a
`UNION ALL`, as the switch asks.
- Local exchanges in the query profiles on a local 1 FE + 1 BE cluster
(master `4530f761fa1`; `doris-fe.jar` from `7c4b4462dcc` with #70177,
and for "after" with only `AggregationNode.class` replaced), switch off,
`parallel_pipeline_task_num=4`, `agg_phase=2`. The `AggSink` rows set
`disable_streaming_preaggregations=true`; the distinct rows are a `GROUP
BY` without aggregate functions:
| Query | BE planner | FE planner before | FE planner after |
|---|---|---|---|
| LOCAL `AggSink` over a nested loop join | ADAPTIVE_PASSTHROUGH,
LOCAL_HASH_SHUFFLE | ADAPTIVE_PASSTHROUGH | ADAPTIVE_PASSTHROUGH,
LOCAL_HASH_SHUFFLE |
| LOCAL `AggSink` over `explode` | PASSTHROUGH, LOCAL_HASH_SHUFFLE |
PASSTHROUGH | PASSTHROUGH, LOCAL_HASH_SHUFFLE |
| LOCAL `AggSink` over a one-tablet scan (serial) | PASSTHROUGH,
LOCAL_HASH_SHUFFLE | PASSTHROUGH | PASSTHROUGH, LOCAL_HASH_SHUFFLE |
| LOCAL `AggSink` over `UNION ALL` | none | none | none |
| distinct pre-aggregation over `explode` | PASSTHROUGH ×2 | PASSTHROUGH
| PASSTHROUGH ×2 |
| distinct pre-aggregation over a nested loop join |
ADAPTIVE_PASSTHROUGH, PASSTHROUGH | ADAPTIVE_PASSTHROUGH |
ADAPTIVE_PASSTHROUGH |
- The new regression suite fails on the FE without the fix at its first
plan check (`nlj: [VNESTED]`) and passes with it; its 16 result blocks
(4 queries × both planners × the switch on and off) match results
computed independently from the inserted rows.
`nereids_p0/local_shuffle` (11 suites),
`query_p0/join/test_agg_after_nested_loop_join_local_exchange` and
`nereids_syntax_p0/agg_4_phase` pass on the same cluster with the fix.
### Notes for Reviewers
- `baseClassRequire()` stays in both branches although only a non-serial
child reaches them now, since a serial child always breaks the key
distribution. It keeps the branches shaped like BE's call to the base
class, as #70177 does for streaming aggregation.
- In `LocalShuffleNodeCoverageTest`, #70177's five key-breaking children
moved into `childrenThatBreakKeyDistribution()` and its aggregate
builder into `buildLocalPhaseAgg()`, so the new cases reuse them;
#70177's case is otherwise unchanged.
- The `PlanNode#requiresShuffleForCorrectness` javadoc still named
`enable_local_exchange_before_agg` for streaming aggregation, which
reads `enable_local_exchange_before_streaming_agg` since #70177.
---
.../org/apache/doris/planner/AggregationNode.java | 18 ++-
.../java/org/apache/doris/planner/PlanNode.java | 4 +-
.../planner/LocalShuffleNodeCoverageTest.java | 141 +++++++++++++----
.../apache/doris/qe/LocalExchangePlannerTest.java | 62 ++++++++
...st_local_exchange_before_agg_breaking_child.out | 169 ++++++++++++++++++++
...local_exchange_before_agg_breaking_child.groovy | 176 +++++++++++++++++++++
6 files changed, 529 insertions(+), 41 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/AggregationNode.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/AggregationNode.java
index 8dd12e0a8b4..ed3e9322e35 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/AggregationNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/AggregationNode.java
@@ -280,7 +280,8 @@ public class AggregationNode extends PlanNode {
ConnectContext connectContext = translatorContext.getConnectContext();
SessionVariable sessionVariable = connectContext.getSessionVariable();
- // PR #62438: when false, non-finalize agg falls back to BE base class.
+ // PR #62438: when false, a LOCAL-phase agg falls back to BE base
class while its child
+ // keeps the local key distribution (PR #63766).
boolean enableLeBeforeAgg =
sessionVariable.enableLocalExchangeBeforeAgg;
boolean hasKeys = !aggInfo.getGroupingExprs().isEmpty();
boolean selfOrInheritedShuffled =
translatorContext.hasShuffleForCorrectnessAncestor(this)
@@ -292,7 +293,8 @@ public class AggregationNode extends PlanNode {
if (canUseDistinctStreamingAgg(sessionVariable)) {
// DistinctStreamingAggOperatorX. Two flavors share this operator
class:
// - streaming preagg (useStreamingPreagg=true):
performance-only,
- // flag controls
+ // flag controls while the child keeps the key distribution;
after a
+ // child that breaks it, the force-passthrough rule below
applies
// - non-streaming dedup (useStreamingPreagg=false):
correctness-required,
// always HASH regardless of flag
// Diverges from BE: BE's `!_needs_finalize &&
!enable_local_exchange_before_agg`
@@ -300,7 +302,8 @@ public class AggregationNode extends PlanNode {
// wrong-result bug as AggSink (DORIS-25413).
if (needsFinalize && !hasKeys) {
requireChild = LocalExchangeTypeRequire.noRequire();
- } else if (!needsFinalize && useStreamingPreagg &&
!enableLeBeforeAgg) {
+ } else if (!needsFinalize && useStreamingPreagg &&
!enableLeBeforeAgg
+ && !childBreaksLocalKeyDistribution(connectContext)) {
requireChild = baseClassRequire(connectContext);
} else if (needsFinalize || (hasKeys && !useStreamingPreagg)) {
requireChild = AddLocalExchange.isColocated(this)
@@ -335,6 +338,7 @@ public class AggregationNode extends PlanNode {
//
// Phase semantics for !needsFinalize:
// - FIRST / SECOND (LOCAL phase, !isMerge): performance-only,
flag controls
+ // while the child keeps the key distribution
// - FIRST_MERGE (correctness-required): always HASH regardless
of flag
//
// Diverges from BE here: BE's `!_needs_finalize &&
!enable_local_exchange_before_agg`
@@ -354,10 +358,12 @@ public class AggregationNode extends PlanNode {
requireChild = needsFinalize
? LocalExchangeTypeRequire.noRequire()
: baseClassRequire(connectContext);
- } else if (!needsFinalize && !aggInfo.isMerge() &&
!enableLeBeforeAgg) {
+ } else if (!needsFinalize && !aggInfo.isMerge() &&
!enableLeBeforeAgg
+ && !childBreaksLocalKeyDistribution(connectContext)) {
// LOCAL phase (FIRST preagg / SECOND distinct local) + user
opted out
- // of pre-agg LE → base class decides: serial child →
PASSTHROUGH
- // (parallelism), non-serial child → NOOP (no LE).
+ // of pre-agg LE → base class decides. A serial child breaks
the key
+ // distribution, so only a non-serial child gets here → NOOP
(no LE);
+ // otherwise the next branch still requires HASH, as BE does.
requireChild = baseClassRequire(connectContext);
} else if (!needsFinalize || AddLocalExchange.isColocated(this)) {
// FIRST_MERGE (correctness) or finalize+colocate → HASH.
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
index d680afac953..3fa8a9698a2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
@@ -895,7 +895,9 @@ public abstract class PlanNode extends TreeNode<PlanNode> {
* Default is false; only operators that truly need hash for correctness
override
* (finalize AggSink with group keys, HashJoin PARTITIONED/BUCKET_SHUFFLE,
Intersect,
* Except, analytic SortNode, partition-by AnalyticEvalNode). Operators
that request
- * hash for performance only (StreamingAgg pre-agg with
enable_local_exchange_before_agg)
+ * hash for performance only (the pre-aggregations: StreamingAgg with
+ * enable_local_exchange_before_streaming_agg, the LOCAL phase of AggSink
with
+ * enable_local_exchange_before_agg)
* MUST NOT override — that would cause SetOperationNode to over-insert
HASH LE on
* every union branch even when nothing downstream actually needs
correctness shuffling.
*
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 eef6fbbe0e2..571b1319268 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
@@ -999,7 +999,8 @@ public class LocalShuffleNodeCoverageTest {
@Test
public void
testAggregationNodeDistinctLocalPhaseWithLeDisabledStaysNoRequire() {
// LOCAL distinct phase + enable_local_exchange_before_agg=false →
base class
- // behavior (NOOP for a non-serial child): user explicitly opted out
of pre-agg LE.
+ // behavior (NOOP for a non-serial child that keeps the key
distribution): user
+ // explicitly opted out of pre-agg LE.
AggContext agg =
buildAggContext(Collections.singletonList(multiDistinctFunction("multi_distinct_count")),
/* groupByExprs */ true,
/* merge */ false, /* needsFinalize */ false,
LocalExchangeType.NOOP, KEYED_DISTRIBUTE_EXPRS);
SessionVariable sessionVariable = new SessionVariable();
@@ -1047,7 +1048,8 @@ public class LocalShuffleNodeCoverageTest {
@Test
public void
testAggregationNodeGroupByLocalPhaseWithLeDisabledStaysNoRequire() {
// GROUP BY local phase + enable_local_exchange_before_agg=false →
base class
- // behavior (NOOP for a non-serial child): user explicitly opted out
of pre-agg LE.
+ // behavior (NOOP for a non-serial child that keeps the key
distribution): user
+ // explicitly opted out of pre-agg LE.
// aggExprs is non-empty so the AggSink branch is exercised (an empty
aggExprs
// would route through DistinctStreamingAgg with its own hash logic).
AggContext agg =
buildAggContext(Collections.singletonList(multiDistinctFunction("multi_distinct_count")),
/* groupByExprs */ false,
@@ -1161,37 +1163,30 @@ public class LocalShuffleNodeCoverageTest {
sessionVariable.enableStreamingAggHashJoinForcePassthrough = false;
sessionVariable.enableBroadcastJoinForcePassthrough = true;
- SerialTrackingPlanNode serialChild = new
SerialTrackingPlanNode(nextPlanNodeId(), LocalExchangeType.NOOP);
- serialChild.fragment = Mockito.mock(PlanFragment.class);
-
Mockito.when(serialChild.fragment.useSerialSource(Mockito.any())).thenReturn(true);
-
- NestedLoopJoinNode nestedLoopJoin = new
NestedLoopJoinNode(nextPlanNodeId(),
- new TrackingPlanNode(nextPlanNodeId(), LocalExchangeType.NOOP),
- new TrackingPlanNode(nextPlanNodeId(), LocalExchangeType.NOOP),
- Lists.newArrayList(new TupleId(NEXT_ID.getAndIncrement())),
JoinOperator.INNER_JOIN, false);
- nestedLoopJoin.fragment = Mockito.mock(PlanFragment.class);
-
Mockito.when(nestedLoopJoin.fragment.useSerialSource(Mockito.any())).thenReturn(false);
-
- HashJoinNode broadcastJoin = new HashJoinNode(nextPlanNodeId(),
- new TrackingPlanNode(nextPlanNodeId(), LocalExchangeType.NOOP),
- new TrackingPlanNode(nextPlanNodeId(),
LocalExchangeType.NOOP), JoinOperator.INNER_JOIN,
-
Collections.singletonList(Mockito.mock(BinaryPredicate.class)),
Collections.emptyList(),
- null, null, false);
- broadcastJoin.setDistributionMode(DistributionMode.BROADCAST);
-
- TableFunctionNode tableFunction = new
TableFunctionNode(nextPlanNodeId(),
- new TrackingPlanNode(nextPlanNodeId(), LocalExchangeType.NOOP),
- new TupleId(NEXT_ID.getAndIncrement()), new ArrayList<>(), new
ArrayList<>(), new ArrayList<>());
+ for (PlanNode child : childrenThatBreakKeyDistribution()) {
+ AggregationNode agg = buildStreamingPreAgg(child);
+ Pair<PlanNode, LocalExchangeType> output =
agg.enforceAndDeriveLocalExchange(
+ mockTranslatorContext(sessionVariable), null,
LocalExchangeTypeRequire.noRequire());
+ String name = child.getClass().getSimpleName();
+
Assertions.assertEquals(LocalExchangeType.LOCAL_EXECUTION_HASH_SHUFFLE,
output.second, name);
+ assertChildLocalExchangeType(agg, 0,
LocalExchangeType.LOCAL_EXECUTION_HASH_SHUFFLE);
+ Assertions.assertSame(child, agg.getChild(0).getChild(0), name);
+ }
+ }
- AssertNumRowsElement assertElement =
Mockito.mock(AssertNumRowsElement.class);
-
Mockito.when(assertElement.getAssertion()).thenReturn(AssertNumRowsElement.Assertion.EQ);
- AssertNumRowsNode assertNumRows = new
AssertNumRowsNode(nextPlanNodeId(),
- new TrackingPlanNode(nextPlanNodeId(),
LocalExchangeType.NOOP), assertElement,
- new TupleDescriptor(new TupleId(NEXT_ID.getAndIncrement())));
+ @Test
+ public void
testAggSinkLocalPhaseRequiresHashWhenChildBreaksKeyDistribution() {
+ // Mirrors BE AggSinkOperatorX after #63766: with
enable_local_exchange_before_agg=false the
+ // LOCAL (non-merge, non-finalize) phase falls back to the base class
only while the child
+ // keeps the local key distribution. After a child that is serial, or
whose own BE operator
+ // requires a non-hash local exchange, it still hashes by its grouping
keys.
+ SessionVariable sessionVariable = new SessionVariable();
+ sessionVariable.enableLocalExchangeBeforeAgg = false;
+ sessionVariable.enableBroadcastJoinForcePassthrough = true;
- for (PlanNode child : new PlanNode[] {serialChild, nestedLoopJoin,
broadcastJoin, tableFunction,
- assertNumRows}) {
- AggregationNode agg = buildStreamingPreAgg(child);
+ for (PlanNode child : childrenThatBreakKeyDistribution()) {
+ AggregationNode agg = buildLocalPhaseAgg(child,
+
Collections.singletonList(plainAggregateFunction("count")), /* streaming */
false);
Pair<PlanNode, LocalExchangeType> output =
agg.enforceAndDeriveLocalExchange(
mockTranslatorContext(sessionVariable), null,
LocalExchangeTypeRequire.noRequire());
String name = child.getClass().getSimpleName();
@@ -1201,6 +1196,38 @@ public class LocalShuffleNodeCoverageTest {
}
}
+ @Test
+ public void
testDistinctStreamingPreAggFollowsForcePassthroughWhenChildBreaksKeyDistribution()
{
+ // Mirrors BE DistinctStreamingAggOperatorX after #63766: with
+ // enable_local_exchange_before_agg=false a streaming distinct preagg
falls back to the base
+ // class only while the child keeps the local key distribution. After
a child that breaks
+ // it, the preagg follows the same rule as with the switch on, which
is never a hash:
+ // PASSTHROUGH when enable_distinct_streaming_agg_force_passthrough is
on, else the base
+ // class. The joins and AssertNumRows already report a passthrough
output that satisfies a
+ // PASSTHROUGH requirement, so only the table function (NOOP output)
gets a new exchange.
+ for (boolean forcePassthrough : new boolean[] {true, false}) {
+ SessionVariable sessionVariable = new SessionVariable();
+ sessionVariable.enableLocalExchangeBeforeAgg = false;
+ sessionVariable.enableBroadcastJoinForcePassthrough = true;
+ sessionVariable.enableDistinctStreamingAggForcePassthrough =
forcePassthrough;
+
+ for (PlanNode child : childrenThatBreakKeyDistribution()) {
+ AggregationNode agg = buildLocalPhaseAgg(child,
Collections.emptyList(), /* streaming */ true);
+ agg.enforceAndDeriveLocalExchange(
+ mockTranslatorContext(sessionVariable), null,
LocalExchangeTypeRequire.noRequire());
+ String name = child.getClass().getSimpleName() + ", force
passthrough " + forcePassthrough;
+ // The base class also asks for PASSTHROUGH over the serial
child.
+ if (child instanceof SerialTrackingPlanNode
+ || (forcePassthrough && child instanceof
TableFunctionNode)) {
+ assertChildLocalExchangeType(agg, 0,
LocalExchangeType.PASSTHROUGH);
+ Assertions.assertSame(child, agg.getChild(0).getChild(0),
name);
+ } else {
+ Assertions.assertSame(child, agg.getChild(0), name);
+ }
+ }
+ }
+ }
+
@Test
public void testStreamingPreAggKeepsDistributionOfJoinsThatKeepKeys() {
// The join probes that BE leaves on NOOP do not break the key
distribution, so the
@@ -1307,17 +1334,63 @@ public class LocalShuffleNodeCoverageTest {
/** A LOCAL-phase streaming preagg with one grouping key over {@code
child}. */
private static AggregationNode buildStreamingPreAgg(PlanNode child) {
- FunctionCallExpr count = plainAggregateFunction("count");
+ return buildLocalPhaseAgg(child,
Collections.singletonList(plainAggregateFunction("count")),
+ /* streaming */ true);
+ }
+
+ /**
+ * A LOCAL-phase (non-merge, non-finalize) agg with one grouping key over
{@code child}: a
+ * DistinctStreamingAgg without aggregate functions, otherwise a streaming
preagg or an AggSink.
+ */
+ private static AggregationNode buildLocalPhaseAgg(PlanNode child,
List<FunctionCallExpr> aggExprs,
+ boolean streaming) {
AggregateInfo aggInfo = Mockito.mock(AggregateInfo.class);
Mockito.when(aggInfo.getOutputTupleId()).thenReturn(new
TupleId(NEXT_ID.getAndIncrement()));
Mockito.when(aggInfo.getGroupingExprs()).thenReturn(Lists.newArrayList(Mockito.mock(Expr.class)));
-
Mockito.when(aggInfo.getAggregateExprs()).thenReturn(Lists.newArrayList(count));
+ Mockito.when(aggInfo.getAggregateExprs()).thenReturn(new
ArrayList<>(aggExprs));
AggregationNode agg = new AggregationNode(nextPlanNodeId(), child,
aggInfo);
agg.unsetNeedsFinalize();
- agg.setUseStreamingPreagg(true);
+ agg.setUseStreamingPreagg(streaming);
return agg;
}
+ /**
+ * One child of each kind that breaks the local key distribution on BE: a
serial child, and the
+ * operators whose own BE required_data_distribution() is a non-hash local
exchange. The
+ * broadcast hash join breaks it only with
enable_broadcast_join_force_passthrough on.
+ */
+ private static List<PlanNode> childrenThatBreakKeyDistribution() {
+ SerialTrackingPlanNode serialChild = new
SerialTrackingPlanNode(nextPlanNodeId(), LocalExchangeType.NOOP);
+ serialChild.fragment = Mockito.mock(PlanFragment.class);
+
Mockito.when(serialChild.fragment.useSerialSource(Mockito.any())).thenReturn(true);
+
+ NestedLoopJoinNode nestedLoopJoin = new
NestedLoopJoinNode(nextPlanNodeId(),
+ new TrackingPlanNode(nextPlanNodeId(), LocalExchangeType.NOOP),
+ new TrackingPlanNode(nextPlanNodeId(), LocalExchangeType.NOOP),
+ Lists.newArrayList(new TupleId(NEXT_ID.getAndIncrement())),
JoinOperator.INNER_JOIN, false);
+ nestedLoopJoin.fragment = Mockito.mock(PlanFragment.class);
+
Mockito.when(nestedLoopJoin.fragment.useSerialSource(Mockito.any())).thenReturn(false);
+
+ HashJoinNode broadcastJoin = new HashJoinNode(nextPlanNodeId(),
+ new TrackingPlanNode(nextPlanNodeId(), LocalExchangeType.NOOP),
+ new TrackingPlanNode(nextPlanNodeId(),
LocalExchangeType.NOOP), JoinOperator.INNER_JOIN,
+
Collections.singletonList(Mockito.mock(BinaryPredicate.class)),
Collections.emptyList(),
+ null, null, false);
+ broadcastJoin.setDistributionMode(DistributionMode.BROADCAST);
+
+ TableFunctionNode tableFunction = new
TableFunctionNode(nextPlanNodeId(),
+ new TrackingPlanNode(nextPlanNodeId(), LocalExchangeType.NOOP),
+ new TupleId(NEXT_ID.getAndIncrement()), new ArrayList<>(), new
ArrayList<>(), new ArrayList<>());
+
+ AssertNumRowsElement assertElement =
Mockito.mock(AssertNumRowsElement.class);
+
Mockito.when(assertElement.getAssertion()).thenReturn(AssertNumRowsElement.Assertion.EQ);
+ AssertNumRowsNode assertNumRows = new
AssertNumRowsNode(nextPlanNodeId(),
+ new TrackingPlanNode(nextPlanNodeId(),
LocalExchangeType.NOOP), assertElement,
+ new TupleDescriptor(new TupleId(NEXT_ID.getAndIncrement())));
+
+ return Lists.newArrayList(serialChild, nestedLoopJoin, broadcastJoin,
tableFunction, assertNumRows);
+ }
+
private static FunctionCallExpr plainAggregateFunction(String
functionName) {
FunctionCallExpr fce = Mockito.mock(FunctionCallExpr.class);
FunctionName fnName = Mockito.mock(FunctionName.class);
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 6a474b8896f..4f6446c0cf1 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
@@ -86,6 +86,8 @@ public class LocalExchangePlannerTest extends
TestWithFeService implements PlanS
sv.enableBroadcastJoinForcePassthrough = false;
sv.enableShareHashTableForBroadcastJoin = true;
sv.enableLocalExchangeBeforeStreamingAgg = false;
+ sv.enableLocalExchangeBeforeAgg = true;
+ sv.disableStreamPreaggregations = false;
if (tweaks != null) {
tweaks.accept(sv);
}
@@ -693,6 +695,66 @@ public class LocalExchangePlannerTest extends
TestWithFeService implements PlanS
localExchange(PT,
olapScan("t1")))))));
}
+ @Test
+ public void
testLocalPhaseAggHashShufflesAfterChildThatBreaksKeyDistribution() throws
Exception {
+ // With enable_local_exchange_before_agg=false a LOCAL-phase AggSink
(streaming preagg off)
+ // keeps the inherited distribution only while its child keeps the key
distribution, as BE
+ // AggSinkOperatorX decides since #63766. After the
ADAPTIVE_PASSTHROUGH probe side of a
+ // nested loop join, the PASSTHROUGH input of a table function, or a
serial pooling scan it
+ // still hashes its input by the grouping keys.
+ setupLocalShuffleSession(sv -> {
+ sv.setIgnoreStorageDataDistribution(false);
+ sv.enableLocalExchangeBeforeAgg = false;
+ sv.disableStreamPreaggregations = true;
+ });
+ assertPlanShape(
+ "select t1.k2, count(*) from test.t1 join test.t2 on t1.v1 >
t2.v2 group by t1.k2",
+ anyTree(
+ agg(
+ localExchange(LOCAL_HASH,
+ nestedLoopJoin(
+ localExchange(ADAPTIVE_PT,
olapScan()),
+ exchange())))));
+ assertPlanShape(
+ "select e, count(*) from test.t1 lateral view
explode(array(k1, k2)) tmp as e group by e",
+ anyTree(
+ agg(
+ localExchange(LOCAL_HASH,
+ node(TableFunctionNode.class,
+ localExchange(PT,
olapScan("t1")))))));
+ setupLocalShuffleSession(sv -> {
+ sv.enableLocalExchangeBeforeAgg = false;
+ sv.disableStreamPreaggregations = true;
+ });
+ assertPlanShape(
+ "select k2, count(*) from test.t1 group by k2",
+ anyTree(
+ agg(
+ localExchange(LOCAL_HASH,
+ localExchange(PT, olapScan("t1"))))));
+ }
+
+ @Test
+ public void testDistinctStreamingPreAggPassesThroughTableFunctionOutput()
throws Exception {
+ // A GROUP BY without aggregate functions runs its LOCAL phase as a
streaming distinct
+ // preagg. With enable_local_exchange_before_agg=false BE
DistinctStreamingAggOperatorX
+ // still applies enable_distinct_streaming_agg_force_passthrough
(default true) after a
+ // child that breaks the key distribution (#63766), so above a table
function, whose
+ // output is NOOP, the FE planner adds the PASSTHROUGH local exchange
it adds with the
+ // switch on.
+ setupLocalShuffleSession(sv -> {
+ sv.setIgnoreStorageDataDistribution(false);
+ sv.enableLocalExchangeBeforeAgg = false;
+ });
+ assertPlanShape(
+ "select e from test.t1 lateral view explode(array(k1, k2)) tmp
as e group by e",
+ anyTree(
+ agg(
+ localExchange(PT,
+ node(TableFunctionNode.class,
+ localExchange(PT,
olapScan("t1")))))));
+ }
+
@Test
public void testAnalyticPlanContainsPassthroughAndLocalHashShuffle()
throws Exception {
// doc rule "Analytic / 有 partition by / 池化": LE(LOCAL_HASH) for
partition
diff --git
a/regression-test/data/nereids_p0/local_shuffle/test_local_exchange_before_agg_breaking_child.out
b/regression-test/data/nereids_p0/local_shuffle/test_local_exchange_before_agg_breaking_child.out
new file mode 100644
index 00000000000..93ea7e85a13
--- /dev/null
+++
b/regression-test/data/nereids_p0/local_shuffle/test_local_exchange_before_agg_breaking_child.out
@@ -0,0 +1,169 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !nlj_planner_true_le_false --
+10 7 25
+20 6 17
+30 3 4
+40 4 14
+
+-- !nlj_planner_true_le_true --
+10 7 25
+20 6 17
+30 3 4
+40 4 14
+
+-- !nlj_planner_false_le_false --
+10 7 25
+20 6 17
+30 3 4
+40 4 14
+
+-- !nlj_planner_false_le_true --
+10 7 25
+20 6 17
+30 3 4
+40 4 14
+
+-- !explode_planner_true_le_false --
+1 1
+10 3
+2 1
+20 2
+3 1
+30 2
+4 1
+40 1
+5 1
+6 1
+7 1
+8 1
+
+-- !explode_planner_true_le_true --
+1 1
+10 3
+2 1
+20 2
+3 1
+30 2
+4 1
+40 1
+5 1
+6 1
+7 1
+8 1
+
+-- !explode_planner_false_le_false --
+1 1
+10 3
+2 1
+20 2
+3 1
+30 2
+4 1
+40 1
+5 1
+6 1
+7 1
+8 1
+
+-- !explode_planner_false_le_true --
+1 1
+10 3
+2 1
+20 2
+3 1
+30 2
+4 1
+40 1
+5 1
+6 1
+7 1
+8 1
+
+-- !union_planner_true_le_false --
+10 5 21
+20 3 20
+30 2 5
+40 1 8
+50 1 2
+60 1 5
+
+-- !union_planner_true_le_true --
+10 5 21
+20 3 20
+30 2 5
+40 1 8
+50 1 2
+60 1 5
+
+-- !union_planner_false_le_false --
+10 5 21
+20 3 20
+30 2 5
+40 1 8
+50 1 2
+60 1 5
+
+-- !union_planner_false_le_true --
+10 5 21
+20 3 20
+30 2 5
+40 1 8
+50 1 2
+60 1 5
+
+-- !explode_distinct_planner_true_le_false --
+1
+10
+2
+20
+3
+30
+4
+40
+5
+6
+7
+8
+
+-- !explode_distinct_planner_true_le_true --
+1
+10
+2
+20
+3
+30
+4
+40
+5
+6
+7
+8
+
+-- !explode_distinct_planner_false_le_false --
+1
+10
+2
+20
+3
+30
+4
+40
+5
+6
+7
+8
+
+-- !explode_distinct_planner_false_le_true --
+1
+10
+2
+20
+3
+30
+4
+40
+5
+6
+7
+8
+
diff --git
a/regression-test/suites/nereids_p0/local_shuffle/test_local_exchange_before_agg_breaking_child.groovy
b/regression-test/suites/nereids_p0/local_shuffle/test_local_exchange_before_agg_breaking_child.groovy
new file mode 100644
index 00000000000..f681732c545
--- /dev/null
+++
b/regression-test/suites/nereids_p0/local_shuffle/test_local_exchange_before_agg_breaking_child.groovy
@@ -0,0 +1,176 @@
+// 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.
+
+/**
+ * With enable_local_exchange_before_agg=false the FE local shuffle planner
decides the local
+ * exchange below the LOCAL phase of an aggregation the way BE does since
apache/doris#63766: the
+ * phase keeps the distribution it inherits only while its child keeps the
grouping-key
+ * distribution. After a child that breaks it (a serial child, or one that
spreads rows with a
+ * non-hash local exchange, such as the probe side of a nested loop join or
the input of a table
+ * function):
+ * - a LOCAL AggSink (streaming preaggregation off) still hashes its input
by the grouping keys;
+ * - a streaming distinct preaggregation (a GROUP BY without aggregate
functions) applies
+ * enable_distinct_streaming_agg_force_passthrough, as with the switch on.
+ *
+ * Local exchanges only show up in EXPLAIN DISTRIBUTED PLAN, where the node
printed right below a
+ * VAGGREGATE (update serialize) is the child of that LOCAL phase.
+ */
+suite("test_local_exchange_before_agg_breaking_child", "p0") {
+ sql "DROP TABLE IF EXISTS le_agg_breaking_t1"
+ sql "DROP TABLE IF EXISTS le_agg_breaking_t2"
+
+ sql """
+ CREATE TABLE le_agg_breaking_t1 (
+ k1 INT NOT NULL,
+ k2 INT,
+ v1 INT
+ ) ENGINE=OLAP
+ DUPLICATE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 16
+ PROPERTIES ("replication_num" = "1")
+ """
+
+ sql """
+ CREATE TABLE le_agg_breaking_t2 (
+ k1 INT NOT NULL,
+ k2 INT,
+ v2 INT
+ ) ENGINE=OLAP
+ DUPLICATE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 16
+ PROPERTIES ("replication_num" = "1")
+ """
+
+ sql """INSERT INTO le_agg_breaking_t1 VALUES
+ (1, 10, 1), (2, 20, 5), (3, 10, 9), (4, 30, 2),
+ (5, 20, 7), (6, 10, 4), (7, 40, 8), (8, 30, 3)
+ """
+
+ sql """INSERT INTO le_agg_breaking_t2 VALUES
+ (1, 10, 6), (2, 50, 2), (3, 20, 8), (4, 10, 1), (5, 60, 5)
+ """
+
+ // Pin the fuzzy variables that change the aggregation or exchange shape.
Non-pooling scans of
+ // 16 buckets stay parallel on every BE, so only the child under test can
break the key
+ // distribution.
+ def hints = { String planner, String leBeforeAgg, String disableStreaming
-> """/*+SET_VAR(
+ enable_local_shuffle_planner=${planner},
+ enable_local_exchange_before_agg=${leBeforeAgg},
+ disable_streaming_preaggregations=${disableStreaming},
+ enable_local_shuffle=true,
+ ignore_storage_data_distribution=false,
+ parallel_pipeline_task_num=4,
+ enable_bucketed_hash_agg=false,
+ use_serial_exchange=false,
+ enable_local_exchange_before_streaming_agg=false,
+ enable_distinct_streaming_aggregation=true,
+ enable_distinct_streaming_agg_force_passthrough=true,
+ disable_join_reorder=true,
+ enable_sql_cache=false
+ )*/""" }
+
+ // Each query with the disable_streaming_preaggregations value that gives
the LOCAL phase under
+ // test: an AggSink for the first three, a streaming distinct
preaggregation for the last.
+ def queries = [
+ // The probe side of a nested loop join is ADAPTIVE_PASSTHROUGH.
+ nlj: ["true", { String h -> """SELECT ${h} t1.k2, count(*), sum(t2.v2)
+ FROM le_agg_breaking_t1 t1 JOIN le_agg_breaking_t2 t2 ON t1.v1 >
t2.v2
+ GROUP BY t1.k2""" }],
+ // A table function takes PASSTHROUGH input.
+ explode: ["true", { String h -> """SELECT ${h} e, count(*)
+ FROM le_agg_breaking_t1 LATERAL VIEW explode(array(k1, k2)) tmp AS
e
+ GROUP BY e""" }],
+ // A union keeps the key distribution.
+ union: ["true", { String h -> """SELECT ${h} k2, count(*), sum(v)
+ FROM (SELECT k2, v1 AS v FROM le_agg_breaking_t1
+ UNION ALL
+ SELECT k2, v2 AS v FROM le_agg_breaking_t2) u
+ GROUP BY k2""" }],
+ explode_distinct: ["false", { String h -> """SELECT ${h} e
+ FROM le_agg_breaking_t1 LATERAL VIEW explode(array(k1, k2)) tmp AS
e
+ GROUP BY e""" }],
+ ]
+ def query = { String name, String planner, String leBeforeAgg ->
+ def (disableStreaming, template) = queries[name]
+ return template(hints(planner, leBeforeAgg, disableStreaming))
+ }
+
+ // The plan node right below each LOCAL aggregate, with the type of a
local exchange.
+ def belowLocalAggs = { String plan ->
+ def lines = plan.readLines()
+ def below = []
+ for (int i = 0; i < lines.size(); i++) {
+ if (!(lines[i] =~ /\d+:VAGGREGATE \(update serialize\)/)) {
+ continue
+ }
+ for (int j = i + 1; j < lines.size(); j++) {
+ def node = lines[j] =~ /^[\s|]*(?:----)?\d+:(V[A-Za-z-]+)/
+ if (node.find()) {
+ def name = node.group(1)
+ if (name == "VLOCAL-EXCHANGE" && j + 1 < lines.size()) {
+ def type = lines[j + 1] =~ /type: (\S+)/
+ name += type.find() ? ":" + type.group(1) : ""
+ }
+ below << name
+ break
+ }
+ }
+ }
+ return below
+ }
+ def localHash = "VLOCAL-EXCHANGE:LOCAL_EXECUTION_HASH_SHUFFLE"
+ def passthrough = "VLOCAL-EXCHANGE:PASSTHROUGH"
+
+ // A LOCAL AggSink still hashes after a child that breaks the key
distribution.
+ for (def name : ["nlj", "explode"]) {
+ explain {
+ sql "distributed plan ${query(name, 'true', 'false')}"
+ check { plan ->
+ def below = belowLocalAggs(plan)
+ assertTrue(below.contains(localHash), "${name}:
${below}\n${plan}")
+ }
+ }
+ }
+
+ // It keeps the distribution of a child that does not break it.
+ explain {
+ sql "distributed plan ${query('union', 'true', 'false')}"
+ check { plan ->
+ def below = belowLocalAggs(plan)
+ assertTrue(below.contains("VUNION") && !below.contains(localHash),
"${below}\n${plan}")
+ }
+ }
+
+ // A streaming distinct preaggregation above a table function, whose
output distribution is
+ // unknown, gets the PASSTHROUGH exchange of
enable_distinct_streaming_agg_force_passthrough.
+ explain {
+ sql "distributed plan ${query('explode_distinct', 'true', 'false')}"
+ check { plan ->
+ def below = belowLocalAggs(plan)
+ assertTrue(below.contains(passthrough), "${below}\n${plan}")
+ }
+ }
+
+ // Same results from the FE and BE local shuffle planners, with the switch
off and on.
+ for (def name : queries.keySet()) {
+ for (def planner : ["true", "false"]) {
+ for (def leBeforeAgg : ["false", "true"]) {
+
"order_qt_${name}_planner_${planner}_le_${leBeforeAgg}"(query(name, planner,
leBeforeAgg))
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
