[Bug] MultiDistinct can OOM on group-by with high-NDV or unknown-statistics distinct arguments

[Baymine]

Search before asking

• I had searched in the issues and found no similar issues.

Version

master / 4.0

What's Wrong?

The MultiDistinct strategy 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 whose statistics are unknown and could be near-unique), one node can end up holding a whole group's value set and OOM, e.g.:

select count(distinct payment_id), ... from t group by merchant_id

The no-group-by branch of DistinctAggStrategySelector.shouldUseMultiDistinct already guards against this by checking distinct-argument NDV and routing unknown-statistics cases to the CTE split strategy. The group-by branch does not: it neither checks distinct-argument NDV, nor treats unknown distinct-argument statistics as risky, and on unknown group-by statistics it unconditionally chooses MultiDistinct.

What You Expected?

For a group-by query whose distinct argument is near-unique relative to input rows, or whose distinct argument (or a slot it reads) has unknown statistics, the planner should route to the CTE split strategy (which redistributes on the distinct key) to avoid the single-node OOM — mirroring the no-group-by branch.

How to Reproduce?

Run a group-by aggregation with one or more count(distinct ...) over a near-unique column (or a column wrapped in if(cond, col, null) where col is unanalyzed), and observe that MultiDistinct is chosen and can OOM under a low-cardinality group-by key.

Anything Else?

Fix (group-by branch of shouldUseMultiDistinct), matching the no-group-by branch:

1. hasHighNdvDistinctArgument: force CTE split when any distinct argument's estimated NDV >= row * MID_CARDINALITY_THRESHOLD.
2. hasUnknownNdvDistinctArgument: force CTE split when a distinct argument's estimate is unknown, or any input slot it reads has absent/unknown statistics (the per-slot scan is required because ExpressionEstimation.visitIf fabricates a non-unknown NDV for if(cond, col, null) even when col is unanalyzed).
3. When group-by statistics are unknown (reached only when every distinct argument is confirmed low-NDV), prefer CTE split for a single group-by key and keep MultiDistinct for >= 2 keys.

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[Doris-Breakwater]

Breakwater-GitHub-Analysis-Slot: slot_2f0ad5858350

Initial assessment

Confirmed planner safety issue, with high confidence; the proposed fix is directionally correct but incomplete for the issue's stated “one or more count(distinct ...)” scope.

The issue currently has no labels, assignee, or linked pull request. Suggested triage is Bug + Nereids/optimizer (and memory/OOM, if such a label exists in the repository taxonomy).

Code findings

1. In DistinctAggStrategySelector.shouldUseMultiDistinct, the scalar branch rejects MultiDistinct when a distinct argument has absent/unknown statistics or NDV at least inputRows * MID_CARDINALITY_THRESHOLD. The grouped branch does neither check. It only handles a skew hint, group-key statistics, and group NDV; if group-key statistics are unknown it returns true immediately. This directly confirms the selector asymmetry described here on both the local master and branch-4.0 snapshots.

2. The memory mechanism is also confirmed. MultiDistinctFunctionStrategy converts the aggregate to multi_distinct_*; the BE implementation in aggregate_function_uniq.h stores an exact flat_hash_set in each aggregate state and merges the sets. A global grouped aggregation is partitioned by the group key, so the complete state for a hot/low-cardinality group is ultimately owned by one execution instance. A near-unique distinct argument can therefore make one per-group state grow with the group's row count.

3. The proposed input-slot validation is necessary for expressions. ExpressionEstimation.visitIf constructs a fresh statistic with NDV at least 2 and does not propagate the children's unknown flag. Consequently, checking only the estimated statistic for if(cond, col, null) can classify the expression as known even when col has absent/unknown statistics. Every input Slot collected from each distinct argument must also have a present, known statistic.

Scope gap that should be resolved before the PR

DistinctAggStrategySelector only processes aggregates with more than one distinct argument group (distinctArgumentGroupCountUpToTwo(...) > 1 on current master). A single distinct argument group passes through to DistinctAggregateRewriter.

That second rewriter has the same safety exposure:

• on current master, DistinctAggregateRewriter.chooseStrategy returns MULTI_STRATEGY when group-key or distinct-argument statistics are unknown;
• it also returns MULTI_STRATEGY specifically for low group-key NDV plus high distinct-key NDV;
• branch 4.0 has equivalent behavior in shouldUseMultiDistinct;
• both paths convert to the same exact-set multi_distinct_* implementation.

This is not only a separate single-distinct case. CTE splitting turns each distinct argument group into its own one-distinct aggregate branch, and the later CTE-child rewrite can run DistinctAggregateRewriter on those branches. With unknown or low-group/high-distinct statistics, that rewriter can convert a CTE branch back to MultiDistinct and reintroduce the same exact-set risk.

Therefore, changing only DistinctAggStrategySelector does not reliably close the OOM path and does not fix the single-distinct example unless the ellipsis contains another DISTINCT argument group. Please either:

1. apply the safety checks to both strategy selectors, preferably through shared helper logic; or
2. narrow the issue and reproduction explicitly to multiple distinct argument groups and track the single-distinct path separately.

There is also a physical-plan condition to verify. CTE decomposition enables the normal DISTINCT multi-phase plans, including lower shuffles containing the distinct key. However, SplitAggMultiPhase can expose both group-key and distinct-key partition alternatives in some unknown-statistics cases. The regression must assert the final physical plan, not only the presence of LogicalCTEAnchor, and confirm that the risky low/unknown group-cardinality case actually selects a lower shuffle containing the distinct argument. Otherwise the CTE rewrite alone is not a proof that one large group cannot still be concentrated.

Missing runtime evidence

The code is sufficient to confirm the unsafe planner decision, so an OOM profile is not required to accept this as a planner bug. To validate impact and make the reproduction unambiguous, it would still help to attach:

• the complete SQL, showing whether there is one or multiple distinct argument groups;
• EXPLAIN SHAPE PLAN (or full EXPLAIN) plus multi_distinct_strategy, agg_phase, row count, and available column statistics;
• the BE memory-limit error and relevant query-profile/aggregate memory trackers, if an actual OOM was observed;
• input row count, group-key NDV, distinct-argument NDV, and the largest group's row/NDV count.

Recommended implementation and tests

• Evaluate distinct-argument risk before the grouped/scalar heuristic: reject MultiDistinct if an argument estimate is absent/unknown, any input slot has absent/unknown statistics, or estimated NDV is at least inputRows * MID_CARDINALITY_THRESHOLD.
• Reuse the same risk predicate in DistinctAggStrategySelector and DistinctAggregateRewriter so their behavior cannot diverge again.
• Preserve or explicitly document the precedence of forced modes (multi_distinct_strategy, agg_phase) and the existing sourceRepeat/GROUPING SETS constraints.
• Treat “unknown group stats: one key => CTE, two or more keys => MultiDistinct” as a performance heuristic, not a cardinality guarantee. It is only reached after distinct arguments are considered safe and needs plan-shape coverage.
• Add regression cases for: high-NDV direct slots; unknown direct slots; if(cond, unanalyzed_col, null); known low-NDV arguments; unknown one-key and multi-key group-by statistics; unchanged scalar behavior; explicit strategy overrides; and both one and multiple distinct argument groups.
• Verify master and branch 4.0 separately: their selector entry conditions and DistinctAggregateRewriter APIs differ, so the backport will need branch-specific adaptation.

Given the reporter's willingness to submit a PR, the next maintainer action should be to confirm the intended single-distinct scope and the forced-mode precedence, then invite a PR with the above physical-plan regression coverage.

[Doris-Breakwater]

Initial triage: valid FE/Nereids planner safety issue, but the proposed fix is not yet complete end to end. The issue currently has no labels; it should be triaged under the repository's bug/FE-optimizer category.

Code findings

I checked local master (27cb451229c7a42dd089c5248847828eb0536d02) and upstream/branch-4.0; both have the relevant behavior.

• DistinctAggStrategySelector.shouldUseMultiDistinct checks unknown/high-NDV distinct arguments only in the scalar (no-group-by) branch. In the group-by branch it checks only the group keys, and unknown group-key statistics return true immediately. This asymmetry is real.
• The BE implementation confirms the memory mechanism: AggregateFunctionUniqExactData owns a flat_hash_set, and AggregateFunctionUniq.add_batch selects the set through the aggregate-state place for each row/group. Memory for a group therefore grows with that group's distinct values, and merging also materializes those keys in the destination set.
• This selector only runs its strategy decision when there are at least two distinct-argument groups (distinctArgumentGroupCountUpToTwo(agg) > 1). A query with one COUNT(DISTINCT payment_id) plus ordinary aggregates bypasses this selector. Its strategy is controlled by DistinctAggregateRewriter, so the example and the phrase "one or more" need this second code path to be covered explicitly.

Important gap in the proposed fix / PR #66045

Routing the outer multi-distinct aggregate to SplitMultiDistinctStrategy does not by itself guarantee redistribution on the distinct key. The newly created per-distinct CTE aggregate branches are subsequently processed by DistinctAggregateRewriter:

• on unknown group-key or distinct-key statistics, DistinctAggregateRewriter chooses MULTI_STRATEGY;
• with known statistics, it also chooses MULTI_STRATEGY when group NDV is below 0.1% of input rows and distinct NDV is above 10%.

Consequently, an unknown-statistics branch, or the genuinely near-unique/low-cardinality-group case described in this issue, can be converted back to multi_distinct_* inside each CTE consumer. The CTE may separate multiple sets into different branches, but each branch can still keep a whole group's distinct set on one node; that does not establish the claimed OOM protection.

The current PR regression uses d_20 over about 1,003 rows (roughly 2% NDV). That crosses the selector's 1% MID_CARDINALITY_THRESHOLD but remains below DistinctAggregateRewriter's 10% HIGH_CARDINALITY_THRESHOLD, so the downstream rewriter happens to choose a distributed split. It does not exercise a near-unique argument above 10%. The no-statistics expected plan has only one aggregate per CTE consumer, consistent with the downstream MULTI_STRATEGY decision.

There is another unsafe assumption in the proposed unknown-group-key fallback: the number of group-by expressions is not evidence of joint cardinality. Two unknown, constant, or strongly correlated keys can still produce one large group. If safety is the goal, groupByExpressions().size() >= 2 should not be treated as sufficient without statistics; either use the conservative split or document and validate this as a performance heuristic rather than an OOM guarantee.

Recommended implementation direction

Coordinate both strategy selectors rather than changing only DistinctAggStrategySelector:

1. Centralize the high/unknown distinct-argument risk test so the multi-distinct selector and DistinctAggregateRewriter cannot make contradictory decisions.
2. For a risky high/unknown-NDV argument, force SPLIT_IN_REWRITE/SPLIT_IN_CASCADES in each single-distinct branch (or carry an explicit "must split" property into the generated CTE aggregates). The final physical plan must shuffle/deduplicate on a key set containing the distinct key and must not contain multi_distinct_* for the protected branch.
3. Apply the same policy to the single-distinct group-by path, since it bypasses DistinctAggStrategySelector.
4. Decide whether scanning every input slot is intentionally conservative. getInputSlots() also includes condition slots and inputs to bounded transforms such as YEAR; missing statistics there do not necessarily make the result high-NDV. Fixing unknown provenance in expression estimation, or checking only value-producing inputs where possible, would reduce unnecessary CTE plans.

Before merging #66045, add final-plan tests (not only helper/selector tests) for:

• two distinct arguments with group NDV below 0.1% and distinct NDV well above 10%;
• unknown distinct statistics, including IF(cond, col, NULL);
• one distinct argument plus an ordinary aggregate;
• multiple unknown/correlated group keys if the >= 2 fallback is retained;
• assertions that the optimized plan contains no multi_distinct_* in the protected branches and that the lower distribution includes the distinct key.

Missing runtime evidence

The code is sufficient to confirm the planner gap, but the reported production OOM and its severity cannot be validated from the issue as written. Please add:

• the complete SQL (the ... is material because the selector requires multiple distinct-argument groups);
• exact build commit/version, table DDL/distribution, backend count, and relevant session variables (multi_distinct_strategy, agg_phase, parallelism);
• EXPLAIN VERBOSE or EXPLAIN SHAPE PLAN, row counts, and group/distinct column statistics;
• the BE OOM/error log and query runtime profile showing the responsible aggregation operator and its peak memory.

As a temporary diagnostic for the multiple-distinct case, test multi_distinct_strategy=2 together with agg_phase=3 or 4, then verify in EXPLAIN that the distinct branches use distributed distinct phases rather than multi_distinct_*. For a single-distinct query, validate agg_phase=3/4 separately because multi_distinct_strategy does not route that selector path.

Breakwater-GitHub-Analysis-Slot: slot_2f0ad5858350