[opt](aggregate) route multi-distinct group-by to CTE split to avoid OOM#66045
Open
Baymine wants to merge 3 commits into
Open
[opt](aggregate) route multi-distinct group-by to CTE split to avoid OOM#66045Baymine wants to merge 3 commits into
Baymine wants to merge 3 commits into
Conversation
… to avoid OOM ### What problem does this PR solve? Issue Number: close apache#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
### What problem does this PR solve? Issue Number: close apache#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
Baymine
requested review from
924060929,
englefly,
morrySnow and
starocean999
as code owners
July 25, 2026 09:59
Contributor
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
Contributor
Author
|
run buildall |
3 tasks
…istinct-cte-oom # Conflicts: # fe/fe-core/src/main/java/org/apache/doris/nereids/util/AggregateUtils.java
Contributor
Author
|
run buildall |
Contributor
FE UT Coverage ReportIncrement line coverage |
Contributor
TPC-H: Total hot run time: 29535 ms |
Contributor
TPC-DS: Total hot run time: 177654 ms |
Contributor
ClickBench: Total hot run time: 25.12 s |
morrySnow
changed the title
morrySnow
changed the title
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What problem does this PR solve?
Issue Number: close #66044
Problem Summary:
The
MultiDistinctstrategy keeps a per-group hash set of every distinct value on a single node.For a group-by query whose distinct argument is near-unique (or has unknown statistics that could
be near-unique), one node can hold a whole group's value set and OOM. The no-group-by branch of
DistinctAggStrategySelector.shouldUseMultiDistinctalready guards against this; the group-bybranch did not.
This PR adds two coordinated guards to the group-by branch of
shouldUseMultiDistinct, matchingthe no-group-by branch's behavior (two commits, logically distinct):
hasHighNdvDistinctArgument: force CTE split when any distinct argument's estimated NDV >=row * MID_CARDINALITY_THRESHOLD. Estimates viaExpressionEstimationsoif(cond, col, null)propagates
col's NDV. Placed beforehasUnknownStatistics(groupBy)so a function group-by keywith unknown stats is still protected. Unknown stats do not trigger this guard (avoids pushing
safe cases onto the costlier CTE path).
hasUnknownNdvDistinctArgument: force CTE split when a distinct argument's estimate is unknownOR any input slot it reads has absent/unknown statistics. The per-slot scan is required because
ExpressionEstimation.visitIffabricates a small non-unknown NDV forif(cond, col, null)evenwhen
colis unanalyzed, which would otherwise let a near-unique wrapped column escape guard.When group-by stats are unknown (now reached only when every distinct argument is confirmed
low-NDV), prefer CTE split for a single group-by key (parallelism is capped at the unknown group
count) and keep
MultiDistinctfor >= 2 keys (joint cardinality is usually high enough to spreadsafely).
Release note
MultiDistinct is no longer chosen for group-by queries whose distinct argument is near-unique
relative to input row count, whose distinct argument has unknown statistics (including arguments
wrapped in expressions such as
if(...)whose underlying input slots are unanalyzed), or whosesingle group-by key has unknown statistics; such queries route to the CTE split strategy to avoid
potential OOM.
Check List (For Author)
SplitMultiDistinctTest#multiSumWithGby: updated to assert the CTE split shape.AggregateUtilsHighNdvDistinctTest(new, 11 cases): covers both helpers —high-NDV triggers / low-NDV doesn't / unknown doesn't trigger the high-NDV guard / if-wrapped
near-unique triggers / threshold boundary; and for the unknown guard: all-unknown triggers /
one-unknown-among-known triggers / all-known doesn't / if-wrapped-known doesn't /
if-wrapped-unknown triggers (the per-slot-scan gap) / year()-over-unknown-slot triggers.
distinct_agg_strategy_selector: the pre-existing WITH-stats caseshould_use_multi_distinct_with_group_by(querycount(distinct d_20), count(distinct b_5) ... group by a_1, whered_20is near-unique) now correctly routes to CTE split, so it wasrenamed to
should_use_cte_with_group_by_high_ndvwith an explanatory comment; and theno-stats case was renamed
no_stats_should_use_multi_distinct->no_stats_should_use_cte_with_group_by.The
.outwas regenerated by the regression harness against a live cluster (not hand-written).unknown-statistics distinct arguments, or unknown single-key group-by stats, now prefers CTE split).