From 63e716c4604b67cedc797f7ba96d98013b94640a Mon Sep 17 00:00:00 2001 From: York Cao <978176728@qq.com> Date: Sat, 25 Jul 2026 09:07:35 +0000 Subject: [PATCH 1/2] [improvement](fe) route high-NDV multi-distinct group-by to CTE split to avoid OOM ### What problem does this PR solve? Issue Number: close #66044 Problem Summary: MultiDistinct keeps a per-group hash set of every distinct value on a single node. Under a group-by with a low- or unknown-cardinality key, a near-unique distinct argument (e.g. count(distinct payment_id) group by merchant_id) forces one node to hold a whole group's value set and can OOM. The no-group-by branch of DistinctAggStrategySelector already guards against this by checking distinct-argument NDV; the group-by branch did not, and on unknown group-by stats it unconditionally returned true (chose MultiDistinct). This change adds two guards to the group-by branch of shouldUseMultiDistinct: 1. AggregateUtils.hasHighNdvDistinctArgument(agg, childStats, row): returns true when any distinct argument's estimated NDV >= row * MID_CARDINALITY_THRESHOLD, forcing CTE split (which redistributes on the distinct key). It estimates via ExpressionEstimation so if(cond, col, null) propagates col's NDV through the then-branch. Unknown stats do not trigger, to avoid pushing safe cases onto the costlier CTE path. This check runs before hasUnknownStatistics so a function group-by key (e.g. format_datetime) with unknown stats is still protected. 2. When group-by stats are unknown, instead of unconditionally choosing MultiDistinct (return true), prefer CTE split for a single group-by key (parallelism would otherwise be capped at the unknown group count) and keep MultiDistinct only for >=2 group-by keys (joint cardinality is usually high enough to spread safely). Mirrors StarRocks' fallback. End-to-end: a query of the form select count(distinct near_unique_col), ... from t group by low_card_key that previously planned as a single MultiDistinct node (and OOMed on large inputs) now plans as a CTE split with a hash join over two redistributed distinct aggregates. ### Release note MultiDistinct is no longer chosen for group-by queries whose distinct argument is near-unique relative to the input row count, or whose single group-by key has unknown statistics; such queries route to the CTE split strategy to avoid potential OOM. ### Check List (For Author) - Test: Unit Test / Regression test - FE unit test: SplitMultiDistinctTest#multiSumWithGby (updated to assert the new CTE split plan shape: physicalCTEAnchor/physicalCTEProducer + hashJoin). - FE unit test: AggregateUtilsHighNdvDistinctTest (new) covers AggregateUtils.hasHighNdvDistinctArgument directly: near-unique argument triggers, low-NDV does not, unknown stats do not trigger, if-wrapped near-unique argument triggers (ExpressionEstimation.visitIf propagation), and NDV exactly at the threshold counts as high. - Behavior changed: Yes (multi-distinct strategy selection for group-by queries with high-NDV distinct arguments or unknown single-key group-by stats now prefers CTE split). - Does this need documentation: No --- .../rewrite/DistinctAggStrategySelector.java | 13 +- .../doris/nereids/util/AggregateUtils.java | 23 +++ .../rules/rewrite/SplitMultiDistinctTest.java | 26 +++- .../AggregateUtilsHighNdvDistinctTest.java | 136 ++++++++++++++++++ 4 files changed, 191 insertions(+), 7 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/nereids/util/AggregateUtilsHighNdvDistinctTest.java diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/DistinctAggStrategySelector.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/DistinctAggStrategySelector.java index 817460a92fe10f..65c2ac2d34b298 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/DistinctAggStrategySelector.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/DistinctAggStrategySelector.java @@ -171,11 +171,22 @@ private boolean shouldUseMultiDistinct(LogicalAggregate agg) { } } } else { + // A near-unique distinct argument makes MultiDistinct hold a whole group's value set on + // one node; force CTE split (redistributes on the distinct key) as the no-group-by branch + // above already does. Checked before hasUnknownStatistics so a function group-by key + // (e.g. format_datetime) with unknown stats still gets protected. + if (AggregateUtils.hasHighNdvDistinctArgument(agg, childStats, row)) { + return false; + } if (agg.hasSkewHint()) { return false; } if (AggregateUtils.hasUnknownStatistics(agg.getGroupByExpressions(), childStats)) { - return true; + // No group-by stats to judge cardinality. Grouping by a single low-cardinality key + // is the common MultiDistinct OOM case (parallelism capped at the group count), so + // prefer CTE split there; with >=2 group-by keys the joint cardinality is usually + // high enough for MultiDistinct to spread safely. Mirrors StarRocks' fallback. + return agg.getGroupByExpressions().size() >= 2; } // The joint ndv of Group by key is high, so multi_distinct is not selected; if (aggStats.getRowCount() >= row * AggregateUtils.LOW_CARDINALITY_THRESHOLD) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/util/AggregateUtils.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/util/AggregateUtils.java index 947da096185821..deb5fca7bfc395 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/util/AggregateUtils.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/util/AggregateUtils.java @@ -148,6 +148,29 @@ public static boolean containsCountDistinctMultiExpr(LogicalAggregate 1); } + /** + * Whether any distinct argument is near-unique (high ndv relative to input rows). + * + * MultiDistinct keeps a per-group hash set of every distinct value on a single node, so a + * near-unique distinct argument under a low-cardinality group by makes one node hold a whole + * group's value set. In that case CTE split (which redistributes on the distinct key) is far + * safer. Estimates via ExpressionEstimation so if(cond, col, null) propagates col's ndv; + * unknown stats do not trigger, to avoid pushing safe cases onto the costlier CTE path. + */ + public static boolean hasHighNdvDistinctArgument( + LogicalAggregate agg, Statistics childStats, double row) { + for (Expression distinctArgument : agg.getDistinctArguments()) { + ColumnStatistic stat = ExpressionEstimation.estimate(distinctArgument, childStats); + if (stat == null || stat.isUnKnown()) { + continue; + } + if (stat.ndv >= row * MID_CARDINALITY_THRESHOLD) { + return true; + } + } + return false; + } + /** count agg function distinct group, up to 2*/ public static int distinctArgumentGroupCountUpToTwo(Aggregate aggregate) { Set distinctArgumentGroup = null; diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/SplitMultiDistinctTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/SplitMultiDistinctTest.java index 94c60586feec96..9800857566fa70 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/SplitMultiDistinctTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/SplitMultiDistinctTest.java @@ -203,16 +203,30 @@ void countMultiColumnsWithGby() { @Test void multiSumWithGby() { + // group by a single column with unknown stats: stage-2 fallback prefers CTE split + // (parallelism would otherwise be capped at the group count). See DistinctAggStrategySelector. String sql = "select sum(distinct b), sum(distinct a) from test_distinct_multi group by c"; PlanChecker.from(connectContext).checkExplain(sql, planner -> { Plan plan = planner.getOptimizedPlan(); MatchingUtils.assertMatches(plan, - physicalResultSink( - physicalDistribute( - physicalProject( - physicalHashAggregate( - physicalDistribute( - physicalHashAggregate(any()))))))); + physicalCTEAnchor( + physicalCTEProducer(any()), + physicalResultSink( + physicalDistribute( + physicalProject( + physicalHashJoin( + physicalProject( + physicalHashAggregate( + physicalDistribute(any()))), + physicalProject( + physicalHashAggregate( + physicalDistribute(any()))) + ).when(join -> join.getJoinType() == JoinType.INNER_JOIN) + ) + ) + ) + ) + ); }); } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/util/AggregateUtilsHighNdvDistinctTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/util/AggregateUtilsHighNdvDistinctTest.java new file mode 100644 index 00000000000000..a6680ae229f809 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/util/AggregateUtilsHighNdvDistinctTest.java @@ -0,0 +1,136 @@ +// 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.nereids.util; + +import org.apache.doris.nereids.memo.Group; +import org.apache.doris.nereids.memo.GroupExpression; +import org.apache.doris.nereids.memo.GroupId; +import org.apache.doris.nereids.trees.expressions.Alias; +import org.apache.doris.nereids.trees.expressions.EqualTo; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator; +import org.apache.doris.nereids.trees.expressions.functions.agg.Count; +import org.apache.doris.nereids.trees.expressions.functions.scalar.If; +import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral; +import org.apache.doris.nereids.trees.expressions.literal.NullLiteral; +import org.apache.doris.nereids.trees.plans.GroupPlan; +import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate; +import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan; +import org.apache.doris.statistics.ColumnStatistic; +import org.apache.doris.statistics.ColumnStatisticBuilder; +import org.apache.doris.statistics.Statistics; +import org.apache.doris.statistics.StatisticsBuilder; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.Lists; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.List; + +class AggregateUtilsHighNdvDistinctTest { + private static final double ROW_COUNT = 10000000; + + private final LogicalOlapScan scan = new LogicalOlapScan( + StatementScopeIdGenerator.newRelationId(), PlanConstructor.student, ImmutableList.of("")); + private final Slot id = scan.getOutput().get(0); + private final Slot name = scan.getOutput().get(2); + private final Slot age = scan.getOutput().get(3); + private final GroupExpression scanGroupExpr = new GroupExpression(scan, ImmutableList.of()); + private final GroupPlan childGroup = new GroupPlan( + new Group(GroupId.createGenerator().getNextId(), scanGroupExpr.getPlan().getLogicalProperties())); + + private LogicalAggregate buildAgg() { + List outputs = Lists.newArrayList( + new Alias(new Count(true, id), "count_distinct_id"), + new Alias(new Count(true, name), "count_distinct_name")); + return new LogicalAggregate<>(Lists.newArrayList(age), outputs, childGroup); + } + + // count(distinct if(age = 1, id, null)) group by age: the distinct argument is an If, not a bare + // slot. This is the shape hasHighNdvDistinctArgument exists for -- ExpressionEstimation.visitIf must + // propagate id's ndv out of the then-branch, otherwise the near-unique payment_id style production + // query would never be detected. + private LogicalAggregate buildAggWithIf() { + If ifExpr = new If(new EqualTo(age, new IntegerLiteral(1)), id, new NullLiteral(id.getDataType())); + List outputs = Lists.newArrayList( + new Alias(new Count(true, ifExpr), "count_distinct_if"), + new Alias(new Count(true, name), "count_distinct_name")); + return new LogicalAggregate<>(Lists.newArrayList(age), outputs, childGroup); + } + + private ColumnStatistic ndvStat(double ndv) { + return new ColumnStatisticBuilder(ROW_COUNT).setNdv(ndv).setAvgSizeByte(4).build(); + } + + @Test + void nearUniqueDistinctArgumentDoesNotUseMultiDistinct() { + Statistics childStats = new StatisticsBuilder() + .setRowCount(ROW_COUNT) + .putColumnStatistics(id, ndvStat(ROW_COUNT * 0.9)) + .putColumnStatistics(name, ndvStat(ROW_COUNT * 0.9)) + .putColumnStatistics(age, ndvStat(12)) + .build(); + Assertions.assertTrue( + AggregateUtils.hasHighNdvDistinctArgument(buildAgg(), childStats, ROW_COUNT)); + } + + @Test + void lowNdvDistinctArgumentStillUsesMultiDistinct() { + Statistics childStats = new StatisticsBuilder() + .setRowCount(ROW_COUNT) + .putColumnStatistics(id, ndvStat(100)) + .putColumnStatistics(name, ndvStat(100)) + .putColumnStatistics(age, ndvStat(12)) + .build(); + Assertions.assertFalse( + AggregateUtils.hasHighNdvDistinctArgument(buildAgg(), childStats, ROW_COUNT)); + } + + @Test + void unknownDistinctArgumentStatsDoNotTrigger() { + Statistics childStats = new StatisticsBuilder().setRowCount(ROW_COUNT).build(); + Assertions.assertFalse( + AggregateUtils.hasHighNdvDistinctArgument(buildAgg(), childStats, ROW_COUNT)); + } + + @Test + void ifWrappedNearUniqueDistinctArgumentDoesNotUseMultiDistinct() { + Statistics childStats = new StatisticsBuilder() + .setRowCount(ROW_COUNT) + .putColumnStatistics(id, ndvStat(ROW_COUNT * 0.9)) + .putColumnStatistics(name, ndvStat(100)) + .putColumnStatistics(age, ndvStat(12)) + .build(); + Assertions.assertTrue( + AggregateUtils.hasHighNdvDistinctArgument(buildAggWithIf(), childStats, ROW_COUNT)); + } + + @Test + void ndvExactlyAtThresholdCountsAsHighNdv() { + Statistics childStats = new StatisticsBuilder() + .setRowCount(ROW_COUNT) + .putColumnStatistics(id, ndvStat(ROW_COUNT * AggregateUtils.MID_CARDINALITY_THRESHOLD)) + .putColumnStatistics(name, ndvStat(100)) + .putColumnStatistics(age, ndvStat(12)) + .build(); + Assertions.assertTrue( + AggregateUtils.hasHighNdvDistinctArgument(buildAgg(), childStats, ROW_COUNT)); + } +} From 8ce576f0d893974570485a2e3c8b2cca980b5771 Mon Sep 17 00:00:00 2001 From: York Cao <978176728@qq.com> Date: Sat, 25 Jul 2026 09:09:18 +0000 Subject: [PATCH 2/2] [improvement](fe) route unknown-NDV multi-distinct args to CTE split ### What problem does this PR solve? Issue Number: close #66044 Problem Summary: Building on the high-NDV guard from the previous commit, this closes a remaining OOM hole in the group-by branch of shouldUseMultiDistinct: a distinct argument whose statistics are UNKNOWN. Because MultiDistinct keeps a per-group hash set of every distinct value on a single node, an unknown distinct argument could be near-unique and OOM one node under a low- or unknown-cardinality group by. The no-group-by branch already routes unknown distinct statistics to CTE; the group-by branch did not. This change adds AggregateUtils.hasUnknownNdvDistinctArgument(agg, childStats), which returns true (forcing CTE split) when a distinct argument's estimated statistics are unknown OR any input Slot it reads has absent/unknown statistics. It is wired into the group-by branch immediately after hasHighNdvDistinctArgument, before hasUnknownStatistics(groupBy), so that only when every distinct argument is confirmed low-NDV (neither high nor unknown) does execution fall through to the group-by cardinality checks. The per-input-slot scan is required because ExpressionEstimation fabricates a small non-unknown NDV for wrappers such as if(cond, col, null) (visitIf) even when col itself is unanalyzed -- without the slot scan, a near-unique column wrapped in if(cond, payment_id, null) would escape both the high-NDV and the unknown-NDV guards. The slot scan treats that case as risky, trading a costlier plan for OOM safety. It is intentionally conservative: a low-NDV derived function like year() over an unanalyzed base slot also routes to CTE. End-to-end: a query of the form select count(distinct if(flag, payment_id, null)), ... from t group by merchant_id where payment_id has no collected statistics previously planned as a single MultiDistinct node (and could OOM) now plans as a CTE split. ### Release note MultiDistinct is no longer chosen for group-by queries whose distinct argument has unknown statistics (including arguments wrapped in if/other expressions whose underlying input slots are unanalyzed); such queries route to the CTE split strategy to avoid potential OOM. ### Check List (For Author) - Test: Unit Test / Regression test - FE unit test: AggregateUtilsHighNdvDistinctTest extended with 6 cases for hasUnknownNdvDistinctArgument: all-unknown triggers, one-unknown-among-known triggers, all-known does not trigger, if-wrapped known does not trigger, if-wrapped unknown triggers (the per-slot-scan gap), and year() over an unanalyzed slot triggers (conservative fallback). - Regression test: distinct_agg_strategy_selector renamed qt_no_stats_should_use_multi_distinct to qt_no_stats_should_use_cte_with_group_by and flipped the expected plan to the CTE split shape. NOTE: the .out was ported from the reference fix's expected EXPLAIN SHAPE plan and must be regenerated by CI (a live cluster with an alive BE was not available locally to run the regression harness). - Behavior changed: Yes (multi-distinct strategy selection for group-by queries with unknown-statistics distinct arguments now prefers CTE split). - Does this need documentation: No --- .../rewrite/DistinctAggStrategySelector.java | 17 +++- .../doris/nereids/util/AggregateUtils.java | 34 ++++++- .../AggregateUtilsHighNdvDistinctTest.java | 98 +++++++++++++++++++ .../distinct_agg_strategy_selector.out | 50 +++++++--- .../distinct_agg_strategy_selector.groovy | 10 +- 5 files changed, 188 insertions(+), 21 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/DistinctAggStrategySelector.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/DistinctAggStrategySelector.java index 65c2ac2d34b298..3f501a34d6dc6d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/DistinctAggStrategySelector.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/DistinctAggStrategySelector.java @@ -178,14 +178,23 @@ private boolean shouldUseMultiDistinct(LogicalAggregate agg) { if (AggregateUtils.hasHighNdvDistinctArgument(agg, childStats, row)) { return false; } + // A distinct argument with unknown statistics could be near-unique. We cannot rule out + // the single-node OOM above, so treat unknown as risky and split, matching the no-group-by + // branch which also routes unknown distinct statistics to CTE. Only when every distinct + // argument is confirmed low-ndv (neither high nor unknown) do we fall through to the + // group-by cardinality checks below. + if (AggregateUtils.hasUnknownNdvDistinctArgument(agg, childStats)) { + return false; + } if (agg.hasSkewHint()) { return false; } if (AggregateUtils.hasUnknownStatistics(agg.getGroupByExpressions(), childStats)) { - // No group-by stats to judge cardinality. Grouping by a single low-cardinality key - // is the common MultiDistinct OOM case (parallelism capped at the group count), so - // prefer CTE split there; with >=2 group-by keys the joint cardinality is usually - // high enough for MultiDistinct to spread safely. Mirrors StarRocks' fallback. + // Reached only when every distinct argument is confirmed low-ndv, so a single node + // holding a whole group's value set is not a concern. With a single group-by key the + // parallelism is still capped at the (unknown) group count, so prefer CTE split; with + // >=2 keys the joint cardinality is usually high enough for MultiDistinct to spread + // safely. Mirrors StarRocks' fallback. return agg.getGroupByExpressions().size() >= 2; } // The joint ndv of Group by key is high, so multi_distinct is not selected; diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/util/AggregateUtils.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/util/AggregateUtils.java index deb5fca7bfc395..883d09cbca869e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/util/AggregateUtils.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/util/AggregateUtils.java @@ -23,6 +23,7 @@ import org.apache.doris.nereids.trees.expressions.Expression; import org.apache.doris.nereids.trees.expressions.IsNull; import org.apache.doris.nereids.trees.expressions.NamedExpression; +import org.apache.doris.nereids.trees.expressions.Slot; import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction; import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateParam; import org.apache.doris.nereids.trees.expressions.functions.agg.Count; @@ -161,7 +162,7 @@ public static boolean hasHighNdvDistinctArgument( LogicalAggregate agg, Statistics childStats, double row) { for (Expression distinctArgument : agg.getDistinctArguments()) { ColumnStatistic stat = ExpressionEstimation.estimate(distinctArgument, childStats); - if (stat == null || stat.isUnKnown()) { + if (stat.isUnKnown()) { continue; } if (stat.ndv >= row * MID_CARDINALITY_THRESHOLD) { @@ -171,6 +172,37 @@ public static boolean hasHighNdvDistinctArgument( return false; } + /** + * Whether any distinct argument has unknown ndv (no usable statistics). + * + * A distinct argument with unknown statistics could be near-unique, and MultiDistinct keeps a + * per-group hash set of every distinct value on a single node, so under a low- or unknown- + * cardinality group by one node may hold a whole group's value set and OOM. Treating unknown as + * risky and routing to CTE split mirrors the no-group-by branch of DistinctAggStrategySelector, + * which also falls back to CTE when a distinct argument's statistics are unknown. + * + *

An argument is unknown when either ExpressionEstimation yields an unknown stat, or any input + * slot it reads has no usable statistics. The per-slot check is required because ExpressionEstimation + * fabricates a small non-unknown ndv for wrappers such as if(cond, col, null) (visitIf) even when col + * itself is unanalyzed, which would otherwise let a near-unique wrapped column escape this guard. + */ + public static boolean hasUnknownNdvDistinctArgument( + LogicalAggregate agg, Statistics childStats) { + for (Expression distinctArgument : agg.getDistinctArguments()) { + ColumnStatistic stat = ExpressionEstimation.estimate(distinctArgument, childStats); + if (stat.isUnKnown()) { + return true; + } + for (Slot inputSlot : distinctArgument.getInputSlots()) { + ColumnStatistic slotStat = childStats.findColumnStatistics(inputSlot); + if (slotStat == null || slotStat.isUnKnown()) { + return true; + } + } + } + return false; + } + /** count agg function distinct group, up to 2*/ public static int distinctArgumentGroupCountUpToTwo(Aggregate aggregate) { Set distinctArgumentGroup = null; diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/util/AggregateUtilsHighNdvDistinctTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/util/AggregateUtilsHighNdvDistinctTest.java index a6680ae229f809..3fa36c82385cbf 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/util/AggregateUtilsHighNdvDistinctTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/util/AggregateUtilsHighNdvDistinctTest.java @@ -24,14 +24,17 @@ import org.apache.doris.nereids.trees.expressions.EqualTo; import org.apache.doris.nereids.trees.expressions.NamedExpression; import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.expressions.SlotReference; import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator; import org.apache.doris.nereids.trees.expressions.functions.agg.Count; import org.apache.doris.nereids.trees.expressions.functions.scalar.If; +import org.apache.doris.nereids.trees.expressions.functions.scalar.Year; import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral; import org.apache.doris.nereids.trees.expressions.literal.NullLiteral; import org.apache.doris.nereids.trees.plans.GroupPlan; import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate; import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan; +import org.apache.doris.nereids.types.DateV2Type; import org.apache.doris.statistics.ColumnStatistic; import org.apache.doris.statistics.ColumnStatisticBuilder; import org.apache.doris.statistics.Statistics; @@ -75,6 +78,18 @@ private LogicalAggregate buildAggWithIf() { return new LogicalAggregate<>(Lists.newArrayList(age), outputs, childGroup); } + // count(distinct year(dt)) group by age, where dt has no statistics. year() is a determinably + // low-ndv function (ExpressionEstimation.visitYear fabricates ndv=69, isUnknown=false), so the + // estimate alone says "not unknown"; only the per-input-slot check sees that dt itself is + // unanalyzed. This documents the intentional conservative fallback: an unknown base slot routes to + // CTE even under a low-ndv wrapper, trading a costlier plan for OOM safety. + private LogicalAggregate buildAggWithYear(Slot dateSlot) { + List outputs = Lists.newArrayList( + new Alias(new Count(true, new Year(dateSlot)), "count_distinct_year"), + new Alias(new Count(true, name), "count_distinct_name")); + return new LogicalAggregate<>(Lists.newArrayList(age), outputs, childGroup); + } + private ColumnStatistic ndvStat(double ndv) { return new ColumnStatisticBuilder(ROW_COUNT).setNdv(ndv).setAvgSizeByte(4).build(); } @@ -133,4 +148,87 @@ void ndvExactlyAtThresholdCountsAsHighNdv() { Assertions.assertTrue( AggregateUtils.hasHighNdvDistinctArgument(buildAgg(), childStats, ROW_COUNT)); } + + @Test + void unknownDistinctArgumentIsTreatedAsRisky() { + // No column statistics at all: every distinct argument is unknown, so we cannot rule out a + // near-unique argument that would OOM MultiDistinct under a low-cardinality group by. This is + // the gap the helper closes -- an unknown distinct argument must route to CTE, matching the + // no-group-by branch of DistinctAggStrategySelector. + Statistics childStats = new StatisticsBuilder().setRowCount(ROW_COUNT).build(); + Assertions.assertTrue( + AggregateUtils.hasUnknownNdvDistinctArgument(buildAgg(), childStats)); + } + + @Test + void oneUnknownDistinctArgumentAmongKnownIsTreatedAsRisky() { + // id has stats but name does not: a single unknown distinct argument is enough to be risky, + // because that one argument alone could be near-unique. + Statistics childStats = new StatisticsBuilder() + .setRowCount(ROW_COUNT) + .putColumnStatistics(id, ndvStat(100)) + .putColumnStatistics(age, ndvStat(12)) + .build(); + Assertions.assertTrue( + AggregateUtils.hasUnknownNdvDistinctArgument(buildAgg(), childStats)); + } + + @Test + void allKnownDistinctArgumentsAreNotUnknown() { + // Both distinct arguments have statistics (regardless of ndv value), so none is unknown and + // the group-by cardinality checks downstream are allowed to run. + Statistics childStats = new StatisticsBuilder() + .setRowCount(ROW_COUNT) + .putColumnStatistics(id, ndvStat(100)) + .putColumnStatistics(name, ndvStat(ROW_COUNT * 0.9)) + .putColumnStatistics(age, ndvStat(12)) + .build(); + Assertions.assertFalse( + AggregateUtils.hasUnknownNdvDistinctArgument(buildAgg(), childStats)); + } + + @Test + void ifWrappedKnownDistinctArgumentIsNotUnknown() { + // if(age = 1, id, null): ExpressionEstimation resolves through the If to id's statistics, so a + // known column wrapped in an If is not treated as unknown. + Statistics childStats = new StatisticsBuilder() + .setRowCount(ROW_COUNT) + .putColumnStatistics(id, ndvStat(100)) + .putColumnStatistics(name, ndvStat(100)) + .putColumnStatistics(age, ndvStat(12)) + .build(); + Assertions.assertFalse( + AggregateUtils.hasUnknownNdvDistinctArgument(buildAggWithIf(), childStats)); + } + + @Test + void ifWrappedUnknownDistinctArgumentIsTreatedAsRisky() { + // if(age = 1, id, null) where id has no statistics. ExpressionEstimation.visitIf fabricates a + // small non-unknown ndv, so the estimate alone would not flag this; the per-input-slot check is + // what catches that id is unanalyzed and could be near-unique. This is the gap the slot scan + // closes -- the production if(cond, payment_id, null) shape. + Statistics childStats = new StatisticsBuilder() + .setRowCount(ROW_COUNT) + .putColumnStatistics(name, ndvStat(100)) + .putColumnStatistics(age, ndvStat(12)) + .build(); + Assertions.assertTrue( + AggregateUtils.hasUnknownNdvDistinctArgument(buildAggWithIf(), childStats)); + } + + @Test + void lowNdvDerivedFunctionOverUnknownSlotIsTreatedAsRisky() { + // count(distinct year(dt)) where dt is unanalyzed. year() has an obviously bounded ndv, yet + // because its base slot has no statistics the strategy still routes to CTE. This is intentional: + // the per-input-slot check favors OOM safety over reusing the low-ndv estimate, so it is a + // conservative fallback, not a regression, if a future reader expects MultiDistinct here. + SlotReference dt = new SlotReference("dt", DateV2Type.INSTANCE); + Statistics childStats = new StatisticsBuilder() + .setRowCount(ROW_COUNT) + .putColumnStatistics(name, ndvStat(100)) + .putColumnStatistics(age, ndvStat(12)) + .build(); + Assertions.assertTrue( + AggregateUtils.hasUnknownNdvDistinctArgument(buildAggWithYear(dt), childStats)); + } } diff --git a/regression-test/data/nereids_rules_p0/agg_strategy/distinct_agg_strategy_selector.out b/regression-test/data/nereids_rules_p0/agg_strategy/distinct_agg_strategy_selector.out index d67a3927da204e..428234e94f0234 100644 --- a/regression-test/data/nereids_rules_p0/agg_strategy/distinct_agg_strategy_selector.out +++ b/regression-test/data/nereids_rules_p0/agg_strategy/distinct_agg_strategy_selector.out @@ -30,13 +30,29 @@ PhysicalCteAnchor ( cteId=CTEId#0 ) --------------------PhysicalDistribute[DistributionSpecExecutionAny] ----------------------PhysicalCteConsumer ( cteId=CTEId#0 ) --- !should_use_multi_distinct_with_group_by -- -PhysicalResultSink ---PhysicalDistribute[DistributionSpecGather] -----hashAgg[GLOBAL] -------PhysicalDistribute[DistributionSpecHash] ---------hashAgg[LOCAL] -----------PhysicalOlapScan[t1000] +-- !should_use_cte_with_group_by_high_ndv -- +PhysicalCteAnchor ( cteId=CTEId#0 ) +--PhysicalCteProducer ( cteId=CTEId#0 ) +----PhysicalOlapScan[t1000] +--PhysicalResultSink +----PhysicalDistribute[DistributionSpecGather] +------hashJoin[INNER_JOIN colocated] hashCondition=((a_1 <=> .a_1)) otherCondition=() +--------hashAgg[GLOBAL] +----------PhysicalDistribute[DistributionSpecHash] +------------hashAgg[LOCAL] +--------------hashAgg[GLOBAL] +----------------PhysicalDistribute[DistributionSpecHash] +------------------hashAgg[LOCAL] +--------------------PhysicalDistribute[DistributionSpecExecutionAny] +----------------------PhysicalCteConsumer ( cteId=CTEId#0 ) +--------hashAgg[GLOBAL] +----------PhysicalDistribute[DistributionSpecHash] +------------hashAgg[LOCAL] +--------------hashAgg[GLOBAL] +----------------PhysicalDistribute[DistributionSpecHash] +------------------hashAgg[LOCAL] +--------------------PhysicalDistribute[DistributionSpecExecutionAny] +----------------------PhysicalCteConsumer ( cteId=CTEId#0 ) -- !no_stats_should_use_cte -- PhysicalCteAnchor ( cteId=CTEId#0 ) @@ -61,13 +77,19 @@ PhysicalCteAnchor ( cteId=CTEId#0 ) ------------------PhysicalDistribute[DistributionSpecExecutionAny] --------------------PhysicalCteConsumer ( cteId=CTEId#0 ) --- !no_stats_should_use_multi_distinct -- -PhysicalResultSink ---PhysicalDistribute[DistributionSpecGather] -----hashAgg[GLOBAL] -------PhysicalDistribute[DistributionSpecHash] ---------hashAgg[LOCAL] -----------PhysicalOlapScan[t1000] +-- !no_stats_should_use_cte_with_group_by -- +PhysicalCteAnchor ( cteId=CTEId#0 ) +--PhysicalCteProducer ( cteId=CTEId#0 ) +----PhysicalOlapScan[t1000] +--PhysicalResultSink +----PhysicalDistribute[DistributionSpecGather] +------hashJoin[INNER_JOIN colocated] hashCondition=((a_1 <=> .a_1)) otherCondition=() +--------hashAgg[GLOBAL] +----------PhysicalDistribute[DistributionSpecHash] +------------PhysicalCteConsumer ( cteId=CTEId#0 ) +--------hashAgg[GLOBAL] +----------PhysicalDistribute[DistributionSpecHash] +------------PhysicalCteConsumer ( cteId=CTEId#0 ) -- !count_distinct_group -- 5 5 diff --git a/regression-test/suites/nereids_rules_p0/agg_strategy/distinct_agg_strategy_selector.groovy b/regression-test/suites/nereids_rules_p0/agg_strategy/distinct_agg_strategy_selector.groovy index 88467c999ba3a0..06f6592650b9bb 100644 --- a/regression-test/suites/nereids_rules_p0/agg_strategy/distinct_agg_strategy_selector.groovy +++ b/regression-test/suites/nereids_rules_p0/agg_strategy/distinct_agg_strategy_selector.groovy @@ -39,12 +39,18 @@ suite("distinct_agg_strategy_selector") { qt_should_use_cte_with_group_by """ explain shape plan select count(distinct a_1) , count(distinct b_5) from t1000 group by d_20;""" - qt_should_use_multi_distinct_with_group_by """explain shape plan + // d_20 has ndv ~20 over ~1000 rows (>= row * MID_CARDINALITY_THRESHOLD), so this near-unique + // distinct argument makes a single MultiDistinct node hold a whole group's value set. Route to + // CTE split, matching the no-group-by branch that also falls back to CTE on a high-ndv argument. + qt_should_use_cte_with_group_by_high_ndv """explain shape plan select count(distinct d_20) , count(distinct b_5) from t1000 group by a_1;""" sql "drop stats t1000" qt_no_stats_should_use_cte """explain shape plan select count(distinct a_1) , count(distinct b_5) from t1000;""" - qt_no_stats_should_use_multi_distinct """explain shape plan + // With statistics dropped the distinct arguments are unknown and could be near-unique, so a single + // MultiDistinct node could hold a whole group's value set and OOM. Route to CTE split instead of + // MultiDistinct, matching the no-group-by branch that also falls back to CTE on unknown statistics. + qt_no_stats_should_use_cte_with_group_by """explain shape plan select count(distinct d_20) , count(distinct b_5) from t1000 group by a_1;""" test {