github-actions[bot] commented on code in PR #66570:
URL: https://github.com/apache/doris/pull/66570#discussion_r3773091220
##########
fe/fe-core/src/test/java/org/apache/doris/planner/LocalShuffleNodeCoverageTest.java:
##########
@@ -855,6 +860,263 @@ 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, KEYED_DISTRIBUTE_EXPRS);
+ Pair<PlanNode, LocalExchangeType> 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, KEYED_DISTRIBUTE_EXPRS);
+ Pair<PlanNode, LocalExchangeType> 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 testAggregationNodeDirectMultiDistinctNoKeyStaysNoRequire() {
+ // A directly called scalar multi_distinct_count(col) has
isDistinct=false and
+ // no child distribute exprs (SplitAggWithoutDistinct builds a LOCAL
aggregate
+ // without partition exprs). It must NOT be given a HASH requirement —
a
+ // zero-key HASH exchange would collapse the whole input onto one task
per BE.
+ AggContext agg =
buildAggContext(Collections.singletonList(multiDistinctFunction("multi_distinct_count")),
+ /* groupByExprs */ true, /* merge */ true, /* needsFinalize */
true,
+ LocalExchangeType.NOOP, null);
+ Pair<PlanNode, LocalExchangeType> output =
agg.node.enforceAndDeriveLocalExchange(
+ agg.ctx, null, LocalExchangeTypeRequire.noRequire());
+ Assertions.assertEquals(LocalExchangeNode.NoRequire.class,
agg.child.lastRequire.getClass(),
+ "direct multi_distinct with no effective key must stay
NoRequire");
+ Assertions.assertEquals(LocalExchangeType.NOOP, output.second);
+ Assertions.assertSame(agg.child, agg.node.getChild(0));
+ }
+
+ @Test
+ public void testAggregationNodeNoPartitionNonFinalizeBaseClassRequire() {
+ // COUNT(*)-style non-finalize (LOCAL) agg: no partition requirement,
so
+ // the non-finalize arm of the first branch falls back to base class
+ // behavior (NOOP for a non-serial child). The agg exprs are non-empty
+ // (a plain count function) so the AggSink branch is exercised rather
+ // than DistinctStreamingAgg.
+ AggContext agg = buildAggContext(
+ Collections.singletonList(plainAggregateFunction("count")), /*
groupByExprs */ true,
+ /* merge */ false, /* needsFinalize */ false,
LocalExchangeType.NOOP, null);
+ Pair<PlanNode, LocalExchangeType> 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 testAggregationNodeNoPartitionFinalizeStaysNoRequire() {
Review Comment:
[P2] Make these finalize fixtures enter AggSink
Both this COUNT(*) case and
`testAggregationNodeGroupByFinalizeRequiresHash()` pass an empty
aggregate-expression list. With the default
`enableDistinctStreamingAggregation=true` and no `sortByGroupKey`,
`canUseDistinctStreamingAgg()` is therefore true, so both tests exercise the
separate DistinctStreamingAgg branch at `AggregationNode.java:291` and never
reach the changed AggSink/`hasPartitionRequirement()` logic they claim to
cover. A translated COUNT(*) has a `count` `FunctionCallExpr`. Please pass
`plainAggregateFunction("count")` in both fixtures (and keep any pure-dedup
checks separately), so removing the changed AggSink finalize behavior actually
fails these tests.
##########
fe/fe-core/src/test/java/org/apache/doris/planner/LocalShuffleNodeCoverageTest.java:
##########
@@ -855,6 +860,263 @@ 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, KEYED_DISTRIBUTE_EXPRS);
+ Pair<PlanNode, LocalExchangeType> 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, KEYED_DISTRIBUTE_EXPRS);
+ Pair<PlanNode, LocalExchangeType> 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 testAggregationNodeDirectMultiDistinctNoKeyStaysNoRequire() {
+ // A directly called scalar multi_distinct_count(col) has
isDistinct=false and
+ // no child distribute exprs (SplitAggWithoutDistinct builds a LOCAL
aggregate
+ // without partition exprs). It must NOT be given a HASH requirement —
a
+ // zero-key HASH exchange would collapse the whole input onto one task
per BE.
+ AggContext agg =
buildAggContext(Collections.singletonList(multiDistinctFunction("multi_distinct_count")),
+ /* groupByExprs */ true, /* merge */ true, /* needsFinalize */
true,
+ LocalExchangeType.NOOP, null);
+ Pair<PlanNode, LocalExchangeType> output =
agg.node.enforceAndDeriveLocalExchange(
+ agg.ctx, null, LocalExchangeTypeRequire.noRequire());
+ Assertions.assertEquals(LocalExchangeNode.NoRequire.class,
agg.child.lastRequire.getClass(),
+ "direct multi_distinct with no effective key must stay
NoRequire");
+ Assertions.assertEquals(LocalExchangeType.NOOP, output.second);
+ Assertions.assertSame(agg.child, agg.node.getChild(0));
+ }
+
+ @Test
+ public void testAggregationNodeNoPartitionNonFinalizeBaseClassRequire() {
+ // COUNT(*)-style non-finalize (LOCAL) agg: no partition requirement,
so
+ // the non-finalize arm of the first branch falls back to base class
+ // behavior (NOOP for a non-serial child). The agg exprs are non-empty
+ // (a plain count function) so the AggSink branch is exercised rather
+ // than DistinctStreamingAgg.
+ AggContext agg = buildAggContext(
+ Collections.singletonList(plainAggregateFunction("count")), /*
groupByExprs */ true,
+ /* merge */ false, /* needsFinalize */ false,
LocalExchangeType.NOOP, null);
+ Pair<PlanNode, LocalExchangeType> 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 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, null);
+ Pair<PlanNode, LocalExchangeType> 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, KEYED_DISTRIBUTE_EXPRS);
+ Pair<PlanNode, LocalExchangeType> 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, KEYED_DISTRIBUTE_EXPRS);
+ SessionVariable sessionVariable = new SessionVariable();
+ sessionVariable.enableLocalExchangeBeforeAgg = false;
+
Mockito.when(agg.connectContext.getSessionVariable()).thenReturn(sessionVariable);
+ Pair<PlanNode, LocalExchangeType> 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 even when
the
+ // user opts out of pre-agg local exchanges
(enable_local_exchange_before_agg
+ // = false): removing the !isMerge() exemption must not weaken it.
+ AggContext agg =
buildAggContext(Collections.singletonList(multiDistinctFunction("multi_distinct_count")),
/* groupByExprs */ true,
+ /* merge */ true, /* needsFinalize */ false,
LocalExchangeType.NOOP, KEYED_DISTRIBUTE_EXPRS);
+ SessionVariable sessionVariable = new SessionVariable();
+ sessionVariable.enableLocalExchangeBeforeAgg = false;
+
Mockito.when(agg.connectContext.getSessionVariable()).thenReturn(sessionVariable);
+ Pair<PlanNode, LocalExchangeType> output =
agg.node.enforceAndDeriveLocalExchange(
+ agg.ctx, null, LocalExchangeTypeRequire.noRequire());
+ Assertions.assertEquals(LocalExchangeNode.RequireHash.class,
agg.child.lastRequire.getClass(),
+ "FIRST_MERGE must keep the hash demand with
enable_local_exchange_before_agg=false");
+
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, null);
+ Pair<PlanNode, LocalExchangeType> 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, KEYED_DISTRIBUTE_EXPRS);
+ SessionVariable sessionVariable = new SessionVariable();
+ sessionVariable.enableLocalExchangeBeforeAgg = false;
+
Mockito.when(agg.connectContext.getSessionVariable()).thenReturn(sessionVariable);
+ Pair<PlanNode, LocalExchangeType> 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() {
Review Comment:
[P2] Exercise the inherited-shuffle aggregate path
This test only calls the node-local `requiresShuffleForCorrectness()`, while
every aggregate created by `buildAggContext()` uses a Mockito context whose
`hasShuffleForCorrectnessAncestor()` remains false. The other true-ancestor
fixtures set the flag directly on Union, and the `agg_phase=1` planner case
reaches the self-detected `multi_distinct_` path. Consequently, dropping the
newly added inherited half of `selfOrInheritedShuffled`, or selecting grouping
expressions instead of the child distribute key in an intermediate aggregate,
leaves all of these tests green even though the documented `AggGlobal ->
DISTINCT_LOCAL -> FIRST_MERGE -> FIRST_LOCAL -> Union` chain relies on it.
Please add an aggregate fixture with inherited shuffle=true and a
non-multi-distinct aggregate, and assert that the HASH LE uses the child
distribute expressions; ideally also cover the real multi-stage plan.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
