Support bounded distinct-key spill for high-NDV exact DISTINCT aggregates

[XuPeng-SH]

Motivation

Exact DISTINCT aggregates retain one key per distinct value. For a query with one or a few hot groups, such as:

SELECT COUNT(DISTINCT user_id) FROM events;

the resident aggregate state grows with the global NDV. Existing Group spill partitions by the group hash. It can evict independent groups, but it cannot subdivide the DISTINCT keys owned by one hot group. A spilled hot group must therefore be reconstructed as one complete in-memory argSkl during reload; recursive group spill cannot make progress when that single state still exceeds the memory budget.

This is an existing architectural limitation, not a regression introduced by #27688. #27688 removes the independent O(B²) duplicate scan inside a bounded 256-row preflight work unit, but deliberately leaves persistent DISTINCT state and Group spill unchanged.

Current behavior

• Batch-local DISTINCT admission is bounded and temporary.
• Persistent exact DISTINCT arguments are retained in aggState.argSkl, with O(global NDV) state.
• SaveSpillIntermediateRows serializes argCnt and every retained argument for each selected group.
• UnmarshalSpillFromReader reconstructs the complete skiplist for that group.
• Group spill chooses buckets from group hashes, so every key belonging to one aggregate group remains inseparable.
• Under a hard allocation limit, a single high-NDV group can repeatedly spill and reload without reducing its required resident state, eventually returning the controlled maximum-partition-depth error. Wide VARCHAR, UUID, multi-column keys, and parallel partial states increase the risk.

The behavior preserves correctness, but it does not provide bounded-memory completion for supported exact DISTINCT workloads.

Expected behavior

Provide an exact, bounded-memory execution path for high-NDV DISTINCT state, including the single-global-group case. The implementation should partition retained DISTINCT tuples by a stable hash of the DISTINCT key, or (group key, DISTINCT arguments) when grouping is present, so one hot aggregate group can be processed in independently reloadable partitions.

The result must remain exact. Falling back to approximate counting, truncating keys, dropping partitions, or changing NULL and equality semantics is not acceptable.

Design requirements

This is a design-level executor and spill change. A design document should be reviewed before implementation and should define:

• ownership and lifecycle of resident partitions, spill files, reload scratch, and finalization state;
• partition key and canonical equality for fixed, varlen, multi-column, const-vector, NULL, floating signed zero, and hash-collision cases;
• exact finalization for COUNT(DISTINCT) and the applicability boundary for other saved-argument DISTINCT aggregates such as SUM, AVG, and GROUP_CONCAT;
• integration with local partial aggregation, distributed merge, and existing group spill;
• a hard memory bound including partition metadata, write/read buffers, hash tables, skiplist or sort state, and recovery reserve;
• repartition or external-sort behavior when one partition still exceeds the limit, with a bounded depth and a controlled no-progress error;
• cancellation, I/O failure, truncated/corrupt spill, retry boundaries, cleanup, Reset/Free, and repeated spill/reload behavior;
• performance behavior for low NDV, high NDV, skew, wide keys, many groups, and a single hot group;
• observability for spilled bytes, rows/keys, partitions, repartitions, reloads, peak resident state, and cleanup failures.

Acceptance criteria

1. A single-group high-NDV exact COUNT(DISTINCT) completes under a finite memory limit that cannot hold all distinct keys at once.
2. Results match an independent no-spill exact oracle for fixed, varlen, and multi-column keys, including duplicates, NULL, signed zero, and forced hash collisions.
3. Peak accounted memory remains within the configured limit plus an explicitly bounded recovery reserve.
4. Multiple groups and a skewed hot group complete without requiring the complete hot-group state to be resident during reload.
5. Spill, recursive repartition, finalization, cancellation, injected write/read failures, and cleanup leave no files, goroutines, allocation-account debt, or reusable-state contamination.
6. Low-NDV and no-spill controls do not regress materially and do not allocate partition machinery on the normal path.
7. Unit tests use injected spill thresholds and small deterministic datasets; no sleeps or large-data UTs. Add a constrained-memory integration regression separately.

Related

• perf: linearize DISTINCT aggregate preflight deduplication #27688 optimizes bounded batch-local DISTINCT preflight and does not change spill semantics.
• arenaskl: cached Inserter can admit a duplicate successor key #27691 fixes cached skiplist Inserter duplicate admission.
• [Bug]: ClickBench Q10 is significantly slower in MatrixOne than DuckDB #27672 and perf: parallelize high-cardinality mixed DISTINCT aggregation #27693 cover a high-NDV mixed-aggregation performance path, but not single-hot-group bounded-memory spill.