From 87f36c2aa7491acf7cb213c728f8ddd444468218 Mon Sep 17 00:00:00 2001 From: 924060929 Date: Fri, 7 Aug 2026 13:44:47 +0800 Subject: [PATCH] [fix](planner) Require hash input for distinct finalize agg without group keys MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The FE local-shuffle planner handed a NoRequire distribution to a finalize merge agg that has no group keys but DISTINCT aggregates (e.g. count(distinct k)). Unlike COUNT(*), such an agg emits per-instance scalar values that the parent sums (sum0(multi_distinct_count(...))), so its input must be hash-partitioned by the distinct key. When a PASSTHROUGH local exchange (e.g. broadcast-join probe fan-out) scatters same-key rows across instances, the parent double-counts overlapping keys — result = correct value × local task count. Mirror BE AggSinkOperatorX::update_operator's _partition_exprs (grouping exprs, or distinct/distribute exprs): aggs with a partition requirement must demand HASH from their child; only partition-less aggs (COUNT(*)-style) keep NoRequire. requiresShuffleForCorrectness now also covers DISTINCT aggregates to match BE is_shuffled_operator(). Regression tests: AggregationNode unit coverage for every phase/flag combination, plus a sql-level distributed-plan test asserting the LOCAL_HASH exchange appears below the distinct finalize agg. --- .../apache/doris/planner/AggregationNode.java | 65 ++++-- .../planner/LocalShuffleNodeCoverageTest.java | 202 ++++++++++++++++++ .../doris/qe/LocalExchangePlannerTest.java | 55 +++++ 3 files changed, 302 insertions(+), 20 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 786e026660af39..ed7f315d17fd56 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 @@ -335,7 +335,13 @@ public Pair enforceAndDeriveLocalExchange( // early return also catches FIRST_MERGE, dropping the HASH requirement and // causing wrong-result (e.g. PASSTHROUGH over serial child breaks the // group-by-key invariant — DORIS-25413). - if (!hasKeys) { + if (!hasPartitionRequirement()) { + // No group keys and no DISTINCT aggregates (e.g. COUNT(*)): the input + // distribution is irrelevant. Any agg with group or distinct keys keeps + // its partition requirement — a finalize agg emits per-instance scalar + // values (sum0(multi_distinct_count(...)) above) that the parent sums, + // so same-key rows must stay in a single instance. This mirrors BE's + // `_partition_exprs` (grouping exprs, or distinct/distribute exprs). requireChild = needsFinalize ? LocalExchangeTypeRequire.noRequire() : baseClassRequire(connectContext); @@ -348,13 +354,16 @@ public Pair enforceAndDeriveLocalExchange( // FIRST_MERGE (correctness) or finalize+colocate → HASH. requireChild = parentRequire.autoRequireHash(); } else if (hasPartitionExprs(parentRequire)) { - // FE-only heuristic: finalize non-colocate with parent hash requirement - // → inherit parent's specific hash type. + // finalize non-colocate with a parent hash requirement → inherit the + // parent's specific hash type. requireChild = parentRequire.autoRequireHash(); } else { - // FE-only heuristic: finalize non-colocate without parent hash → skip - // LE (child Exchange already provides hash distribution). - requireChild = LocalExchangeTypeRequire.noRequire(); + // finalize non-colocate without a parent hash requirement: the input + // must still be key-aligned (group/distinct key), so require HASH + // explicitly instead of trusting the child's distribution. When the + // child already provides hash distribution, satisfy() passes and no + // LE is inserted, so this is safe and free in the common case. + requireChild = LocalExchangeTypeRequire.requireHash(); } } @@ -371,6 +380,32 @@ private LocalExchangeTypeRequire baseClassRequire(ConnectContext connectContext) : LocalExchangeTypeRequire.noRequire(); } + /** + * Whether this agg needs key-aligned (hash-partitioned) input from its child. + * Mirrors BE AggSinkOperatorX::update_operator's `_partition_exprs`: partition + * exprs are the grouping exprs, or the distinct/distribute exprs when the agg + * has DISTINCT functions. A finalize agg that emits per-instance scalar values + * (e.g. the sum0(multi_distinct_count(...)) above) is only correct when + * same-key rows stay in a single instance — overlapping keys across instances + * get summed multiple times by the parent. + */ + private boolean hasPartitionRequirement() { + return !aggInfo.getGroupingExprs().isEmpty() || hasDistinctAggregate(); + } + + private boolean hasDistinctAggregate() { + // Multi-distinct aggregates are detected by function name. Nereids rewrites + // count/sum/group_concat(distinct ...) into dedicated MultiDistinct* functions + // constructed with distinct=false, so by this legacy FunctionCallExpr layer + // isDistinct() is already false and the function name is the only signal. + return aggInfo.getAggregateExprs().stream() + .map(FunctionCallExpr::getFnName) + .filter(name -> name != null) + .map(name -> name.getFunction()) + .filter(name -> name != null) + .anyMatch(name -> name.startsWith("multi_distinct_")); + } + @Override protected List getSemanticPartitionExprs() { return aggInfo.getGroupingExprs(); @@ -386,18 +421,7 @@ protected List getLocalExchangeDistributeExprs(int childIndex, boolean fol // chain scatters same-group rows across N instances, leaving partial_preagg essentially a // no-op and breaking row-arrival order at downstream merge-finalize (e.g. group_concat). List childDist = getChildDistributeExprList(childIndex); - // Multi-distinct aggregates are detected by function name. Nereids rewrites - // count/sum(distinct ...) into dedicated MultiDistinct* functions constructed with - // distinct=false and a "multi_distinct_" name, so by this legacy FunctionCallExpr layer - // isDistinct() is already false and the function name is the only remaining signal — - // there is no structural flag to test here. - boolean hasDistinct = aggInfo.getAggregateExprs().stream() - .map(FunctionCallExpr::getFnName) - .filter(name -> name != null) - .map(name -> name.getFunction()) - .filter(name -> name != null) - .anyMatch(name -> name.startsWith("multi_distinct_")); - if (childDist != null && !childDist.isEmpty() && (followedByShuffled || hasDistinct)) { + if (childDist != null && !childDist.isEmpty() && (followedByShuffled || hasDistinctAggregate())) { return childDist; } return Lists.newArrayList(aggInfo.getGroupingExprs()); @@ -406,13 +430,14 @@ protected List getLocalExchangeDistributeExprs(int childIndex, boolean fol @Override public boolean requiresShuffleForCorrectness() { // Mirrors BE's AggSinkOperatorX::is_shuffled_operator() exactly: - // finalize agg with group keys needs hash-distributed input for correctness. + // finalize agg with partition exprs (group keys or DISTINCT aggregates) + // needs hash-distributed input for correctness. // GLOBAL dedup (!needsFinalize) is intentionally NOT included here — if a // GLOBAL dedup exists, a finalize agg always sits above it (e.g. DISTINCT_GLOBAL // above DISTINCT_LOCAL/GLOBAL_DEDUP), and the finalize agg propagates the flag // down via inheritedShuffled. A solo finalize agg satisfies hash distribution // through its own child requirement. - return needsFinalize && !aggInfo.getGroupingExprs().isEmpty(); + return needsFinalize && hasPartitionRequirement(); } private boolean canUseDistinctStreamingAgg(SessionVariable sessionVariable) { 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 02e68ba6b59def..aaf019f4477537 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 @@ -17,9 +17,11 @@ package org.apache.doris.planner; +import org.apache.doris.analysis.AggregateInfo; import org.apache.doris.analysis.AssertNumRowsElement; import org.apache.doris.analysis.BinaryPredicate; import org.apache.doris.analysis.Expr; +import org.apache.doris.analysis.FunctionCallExpr; import org.apache.doris.analysis.GroupingInfo; import org.apache.doris.analysis.JoinOperator; import org.apache.doris.analysis.OrderByElement; @@ -28,6 +30,7 @@ import org.apache.doris.analysis.SortInfo; import org.apache.doris.analysis.TupleDescriptor; import org.apache.doris.analysis.TupleId; +import org.apache.doris.catalog.FunctionName; import org.apache.doris.common.Pair; import org.apache.doris.common.UserException; import org.apache.doris.nereids.glue.translator.PlanTranslatorContext; @@ -35,6 +38,8 @@ import org.apache.doris.nereids.trees.plans.WindowFuncType; import org.apache.doris.planner.LocalExchangeNode.LocalExchangeType; import org.apache.doris.planner.LocalExchangeNode.LocalExchangeTypeRequire; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.SessionVariable; import org.apache.doris.thrift.TExplainLevel; import org.apache.doris.thrift.TPartitionType; import org.apache.doris.thrift.TPlanNode; @@ -855,6 +860,203 @@ public void testExchangeNodeBranches() { Assertions.assertEquals(LocalExchangeType.NOOP, noopOutput.second); } + @Test + public void testAggregationNodeDistinctFinalizeRequiresHash() { + // count(distinct k) without group-by: the finalize merge agg emits per-instance + // scalar values that the parent sums (sum0(multi_distinct_count(...)) above), so + // the input must be hash-partitioned by the distinct key. Pre-fix this agg got + // NoRequire and a PASSTHROUGH local exchange below scattered same-key rows across + // instances → the parent double-counted (result = correct × task count). + for (String fn : new String[] {"multi_distinct_count", "multi_distinct_sum", + "multi_distinct_group_concat"}) { + AggContext agg = buildAggContext(Collections.singletonList(multiDistinctFunction(fn)), /* groupByExprs */ true, + /* merge */ true, /* needsFinalize */ true, LocalExchangeType.NOOP); + Pair output = agg.node.enforceAndDeriveLocalExchange( + agg.ctx, null, LocalExchangeTypeRequire.noRequire()); + Assertions.assertEquals(LocalExchangeNode.RequireHash.class, agg.child.lastRequire.getClass(), + fn + " finalize agg must require hash input"); + Assertions.assertEquals(LocalExchangeType.LOCAL_EXECUTION_HASH_SHUFFLE, output.second); + assertChildLocalExchangeType(agg.node, 0, LocalExchangeType.LOCAL_EXECUTION_HASH_SHUFFLE); + } + } + + @Test + public void testAggregationNodeDistinctFinalizeWithParentHashRequirement() { + // A parent that already requires hash must not change the agg's own hash demand. + AggContext agg = buildAggContext(Collections.singletonList(multiDistinctFunction("multi_distinct_count")), /* groupByExprs */ true, + /* merge */ true, /* needsFinalize */ true, LocalExchangeType.NOOP); + Pair output = agg.node.enforceAndDeriveLocalExchange( + agg.ctx, null, LocalExchangeTypeRequire.requireHash()); + Assertions.assertEquals(LocalExchangeNode.RequireHash.class, agg.child.lastRequire.getClass()); + Assertions.assertEquals(LocalExchangeType.LOCAL_EXECUTION_HASH_SHUFFLE, output.second); + assertChildLocalExchangeType(agg.node, 0, LocalExchangeType.LOCAL_EXECUTION_HASH_SHUFFLE); + } + + @Test + public void testAggregationNodeNoPartitionFinalizeStaysNoRequire() { + // COUNT(*)-style agg (no group keys, no DISTINCT aggregates) genuinely has no + // partition requirement: the input distribution is irrelevant. + AggContext agg = buildAggContext(Collections.emptyList(), /* groupByExprs */ true, + /* merge */ true, /* needsFinalize */ true, LocalExchangeType.NOOP); + Pair output = agg.node.enforceAndDeriveLocalExchange( + agg.ctx, null, LocalExchangeTypeRequire.noRequire()); + Assertions.assertEquals(LocalExchangeNode.NoRequire.class, agg.child.lastRequire.getClass()); + Assertions.assertEquals(LocalExchangeType.NOOP, output.second); + Assertions.assertSame(agg.child, agg.node.getChild(0)); + } + + @Test + public void testAggregationNodeDistinctLocalPhaseDefaultLeRequiresHash() { + // LOCAL (FIRST/SECOND, non-merge, non-finalize) phase of a distinct agg with the + // default enable_local_exchange_before_agg=true: BE requires HASH here + // (partition_exprs non-empty), so the FE must mirror that. + AggContext agg = buildAggContext(Collections.singletonList(multiDistinctFunction("multi_distinct_count")), /* groupByExprs */ true, + /* merge */ false, /* needsFinalize */ false, LocalExchangeType.NOOP); + Pair output = agg.node.enforceAndDeriveLocalExchange( + agg.ctx, null, LocalExchangeTypeRequire.noRequire()); + Assertions.assertEquals(LocalExchangeNode.RequireHash.class, agg.child.lastRequire.getClass(), + "LOCAL distinct phase with default LE requires hash"); + Assertions.assertEquals(LocalExchangeType.LOCAL_EXECUTION_HASH_SHUFFLE, output.second); + assertChildLocalExchangeType(agg.node, 0, LocalExchangeType.LOCAL_EXECUTION_HASH_SHUFFLE); + } + + @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. + AggContext agg = buildAggContext(Collections.singletonList(multiDistinctFunction("multi_distinct_count")), /* groupByExprs */ true, + /* merge */ false, /* needsFinalize */ false, LocalExchangeType.NOOP); + SessionVariable sessionVariable = new SessionVariable(); + sessionVariable.enableLocalExchangeBeforeAgg = false; + Mockito.when(agg.connectContext.getSessionVariable()).thenReturn(sessionVariable); + Pair output = agg.node.enforceAndDeriveLocalExchange( + agg.ctx, null, LocalExchangeTypeRequire.noRequire()); + Assertions.assertEquals(LocalExchangeNode.NoRequire.class, agg.child.lastRequire.getClass(), + "LOCAL distinct phase with LE disabled keeps no alignment requirement"); + Assertions.assertEquals(LocalExchangeType.NOOP, output.second); + Assertions.assertSame(agg.child, agg.node.getChild(0)); + } + + @Test + public void testAggregationNodeDistinctFirstMergeRequiresHash() { + // FIRST_MERGE (correctness-required) keeps the hash demand regardless of the + // enableLocalExchangeBeforeAgg flag. + AggContext agg = buildAggContext(Collections.singletonList(multiDistinctFunction("multi_distinct_count")), /* groupByExprs */ true, + /* merge */ true, /* needsFinalize */ false, LocalExchangeType.NOOP); + Pair output = agg.node.enforceAndDeriveLocalExchange( + agg.ctx, null, LocalExchangeTypeRequire.noRequire()); + Assertions.assertEquals(LocalExchangeNode.RequireHash.class, agg.child.lastRequire.getClass()); + Assertions.assertEquals(LocalExchangeType.LOCAL_EXECUTION_HASH_SHUFFLE, output.second); + assertChildLocalExchangeType(agg.node, 0, LocalExchangeType.LOCAL_EXECUTION_HASH_SHUFFLE); + } + + @Test + public void testAggregationNodeGroupByFinalizeRequiresHash() { + // GROUP BY finalize agg requires hash input; when the parent has no hash + // requirement the semantic partition exprs (group keys) drive the decision. + AggContext agg = buildAggContext(Collections.emptyList(), /* groupByExprs */ false, + /* merge */ true, /* needsFinalize */ true, LocalExchangeType.NOOP); + Pair output = agg.node.enforceAndDeriveLocalExchange( + agg.ctx, null, LocalExchangeTypeRequire.noRequire()); + Assertions.assertEquals(LocalExchangeNode.RequireHash.class, agg.child.lastRequire.getClass()); + Assertions.assertEquals(LocalExchangeType.LOCAL_EXECUTION_HASH_SHUFFLE, output.second); + assertChildLocalExchangeType(agg.node, 0, LocalExchangeType.LOCAL_EXECUTION_HASH_SHUFFLE); + } + + @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. + // 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, + /* merge */ false, /* needsFinalize */ false, LocalExchangeType.NOOP); + SessionVariable sessionVariable = new SessionVariable(); + sessionVariable.enableLocalExchangeBeforeAgg = false; + Mockito.when(agg.connectContext.getSessionVariable()).thenReturn(sessionVariable); + Pair output = agg.node.enforceAndDeriveLocalExchange( + agg.ctx, null, LocalExchangeTypeRequire.noRequire()); + Assertions.assertEquals(LocalExchangeNode.NoRequire.class, agg.child.lastRequire.getClass()); + Assertions.assertEquals(LocalExchangeType.NOOP, output.second); + Assertions.assertSame(agg.child, agg.node.getChild(0)); + } + + @Test + public void testAggregationNodeRequiresShuffleForCorrectness() { + // Mirrors BE is_shuffled_operator(): finalize agg with partition exprs + // (group keys or DISTINCT aggregates) needs hash-distributed input. + AggContext distinctAgg = buildAggContext(Collections.singletonList(multiDistinctFunction("multi_distinct_count")), + /* groupByExprs */ true, /* merge */ true, /* needsFinalize */ true, + LocalExchangeType.NOOP); + Assertions.assertTrue(distinctAgg.node.requiresShuffleForCorrectness(), + "distinct finalize agg must require shuffle for correctness"); + + AggContext noPartitionAgg = buildAggContext(Collections.emptyList(), /* groupByExprs */ true, + /* merge */ true, /* needsFinalize */ true, LocalExchangeType.NOOP); + Assertions.assertFalse(noPartitionAgg.node.requiresShuffleForCorrectness(), + "COUNT(*) finalize agg has no partition requirement"); + + AggContext groupByAgg = buildAggContext(Collections.emptyList(), /* groupByExprs */ false, + /* merge */ true, /* needsFinalize */ true, LocalExchangeType.NOOP); + Assertions.assertTrue(groupByAgg.node.requiresShuffleForCorrectness(), + "GROUP BY finalize agg must require shuffle for correctness"); + + AggContext localAgg = buildAggContext(Collections.singletonList(multiDistinctFunction("multi_distinct_count")), + /* groupByExprs */ true, /* merge */ false, /* needsFinalize */ false, + LocalExchangeType.NOOP); + Assertions.assertFalse(localAgg.node.requiresShuffleForCorrectness(), + "non-finalize agg does not require shuffle for correctness"); + } + + private static class AggContext { + final AggregationNode node; + final PlanTranslatorContext ctx; + final TrackingPlanNode child; + final ConnectContext connectContext; + + AggContext(AggregationNode node, PlanTranslatorContext ctx, TrackingPlanNode child, + ConnectContext connectContext) { + this.node = node; + this.ctx = ctx; + this.child = child; + this.connectContext = connectContext; + } + } + + /** groupByExprs == true → no group keys (mirrors the RQG scalar COUNT(DISTINCT)). */ + private static AggContext buildAggContext(List aggExprs, boolean groupByExprs, + boolean merge, boolean needsFinalize, LocalExchangeType childProvided) { + PlanTranslatorContext ctx = Mockito.mock(PlanTranslatorContext.class); + ConnectContext connectContext = Mockito.mock(ConnectContext.class); + Mockito.when(connectContext.getSessionVariable()).thenReturn(new SessionVariable()); + Mockito.when(ctx.getConnectContext()).thenReturn(connectContext); + + AggregateInfo aggInfo = Mockito.mock(AggregateInfo.class); + Mockito.when(aggInfo.getOutputTupleId()).thenReturn(new TupleId(NEXT_ID.getAndIncrement())); + ArrayList groupingExprs = new ArrayList<>(); + if (!groupByExprs) { + groupingExprs.add(Mockito.mock(Expr.class)); + } + Mockito.when(aggInfo.getGroupingExprs()).thenReturn(groupingExprs); + Mockito.when(aggInfo.getAggregateExprs()).thenReturn(new ArrayList<>(aggExprs)); + Mockito.when(aggInfo.isMerge()).thenReturn(merge); + + TrackingPlanNode child = new TrackingPlanNode(nextPlanNodeId(), childProvided); + AggregationNode agg = new AggregationNode(nextPlanNodeId(), child, aggInfo); + if (!needsFinalize) { + agg.unsetNeedsFinalize(); + } + return new AggContext(agg, ctx, child, connectContext); + } + + private static FunctionCallExpr multiDistinctFunction(String functionName) { + FunctionCallExpr fce = Mockito.mock(FunctionCallExpr.class); + FunctionName fnName = Mockito.mock(FunctionName.class); + Mockito.when(fnName.getFunction()).thenReturn(functionName); + Mockito.when(fce.getFnName()).thenReturn(fnName); + return fce; + } + private static PlanNodeId nextPlanNodeId() { return new PlanNodeId(NEXT_ID.getAndIncrement()); } 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 275df40cffaa34..abe5db63ef8968 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 @@ -163,6 +163,61 @@ public void testAggWithoutKeyTwoPhase() throws Exception { olapScan("t1"))))))); } + @Test + public void testCountDistinctNoGroupByRequiresHashBeforeAgg() throws Exception { + // count(distinct k2) without group-by: the finalize merge agg emits per-instance + // scalar values that the parent sums (sum0(multi_distinct_count(...))), so its + // input must be hash-partitioned by the distinct key. With the broadcast-join + // probe forced to PASSTHROUGH, rows are scattered below the join — the agg must + // still get a LOCAL_HASH exchange directly beneath it. Pre-fix this agg received + // NoRequire (no hash LE) and the parent double-counted overlapping keys. + // agg_phase=1 forces the multi_distinct_count two-phase shape (the default + // strategy splits into a group-by + count shape which is already safe). + setupLocalShuffleSession(sv -> { + sv.enableBroadcastJoinForcePassthrough = true; + sv.aggPhase = 1; + }); + assertHasLocalExchangeOfType("select count(distinct a.k2) from test.t1 a " + + "left join [shuffle] test.t2 b on a.k2 = b.k2 " + + "left join [broadcast] test.t2 c on b.k1 = c.k1", + LocalExchangeType.LOCAL_EXECUTION_HASH_SHUFFLE); + + } + + @Test + public void testCountDistinctNoGroupByRequiresHashWithoutForcePassthrough() throws Exception { + // Same as above without the broadcast-join force-passthrough: the join output is + // hash-partitioned by the probe key, and the satisfy() check lets the agg keep + // that distribution without inserting a redundant LE — the hash demand must + // still be recognized (a plain hash join probe output satisfies it). + setupLocalShuffleSession(null); + assertPlanShape("select count(distinct a.k2) from test.t1 a " + + "left join [broadcast] test.t2 b on a.k1 = b.k1", + anyTree(agg())); + } + + @Test + public void testCountStarNoGroupByHasNoHashLe() throws Exception { + // COUNT(*) has no partition requirement: no LOCAL_HASH local exchange may appear + // anywhere in the plan (the two-phase agg only gets the PASSTHROUGH fan-out of + // the pooling scan). + setupLocalShuffleSession(null); + assertNoLocalExchangeOfType("select count(*) from test.t1", + LocalExchangeType.LOCAL_EXECUTION_HASH_SHUFFLE); + } + + /** + * Assert that a local exchange of the given type appears somewhere in any + * fragment's plan tree. + */ + protected void assertHasLocalExchangeOfType(String sql, LocalExchangeType expectedType) throws Exception { + StmtExecutor executor = executeNereidsSql("explain distributed plan " + sql); + NereidsPlanner planner = (NereidsPlanner) executor.planner(); + EnumSet types = collectLocalExchangeTypes(planner.getFragments()); + Assertions.assertTrue(types.contains(expectedType), + "expected " + expectedType + " in plan, actual: " + types); + } + @Test public void testBroadcastJoinPoolingShapeDsl() throws Exception { // doc rule "HashJoin / BROADCAST / 池化":