github-actions[bot] commented on code in PR #66570:
URL: https://github.com/apache/doris/pull/66570#discussion_r3772880670
##########
fe/fe-core/src/test/java/org/apache/doris/planner/LocalShuffleNodeCoverageTest.java:
##########
@@ -855,6 +860,258 @@ 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 regardless
of the
+ // enableLocalExchangeBeforeAgg flag.
+ AggContext agg =
buildAggContext(Collections.singletonList(multiDistinctFunction("multi_distinct_count")),
/* groupByExprs */ true,
+ /* merge */ true, /* needsFinalize */ false,
LocalExchangeType.NOOP, KEYED_DISTRIBUTE_EXPRS);
Review Comment:
[P2] Exercise FIRST_MERGE with local exchange disabled
This test says the correctness-required FIRST_MERGE phase keeps HASH
regardless of `enableLocalExchangeBeforeAgg`, but it leaves the freshly
constructed `SessionVariable` at its default `true`. In that state both merge
and non-merge keyed aggregates reach the ordinary HASH arm, so removing the
`!aggInfo.isMerge()` exemption at `AggregationNode.java:353` would still leave
this test green even though `enable_local_exchange_before_agg=false` would make
FIRST_MERGE fall back to NoRequire. Please set
`enableLocalExchangeBeforeAgg=false` here, as the adjacent LOCAL-phase test
does, before retaining the HASH assertions.
--
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]
