Skip to content

feat: enable spillable native grouped integer DISTINCT counts - #27405

Open
QinXi-ai wants to merge 4 commits into
apache:mainfrom
QinXi-ai:feat/25936-distinct-step23
Open

QinXi-ai wants to merge 4 commits into
apache:mainfrom
QinXi-ai:feat/25936-distinct-step23

Conversation

@QinXi-ai

Copy link
Copy Markdown
Contributor

Which issue does this PR close?

Closes #25936 for eligible single-partition integer counts. This completes the spill and physical-selection steps in the issue's follow-up plan. It builds on #27299; that PR's inline-first commit is included here while it awaits merging.

Rationale for this change

The opt-in native path can run out of memory when one group's distinct values exceed the budget. Spilling its List state still requires rebuilding that group's set during replay. This change spills sorted (group keys, value) rows and counts adjacent distinct values during merging, allowing a single group to exceed memory. NULL markers preserve groups whose count is zero.

With bounded replay available, the physical optimizer can select native aggregation for a single integer distinct count whose actual input has one partition. Parallel plans retain the existing rewrite.

What changes are included in this PR?

Drain accumulator state in bounded batches while accounting for retained hash state, conversion buffers, and sorting workspace. Size fixed-width batches from their types, and use the available conversion headroom to reduce the number of spill runs. Sorting omits constant group keys in single-group runs and the value column when a batch has one row per group. Replay compares typed integers and updates its reservation at batch and group boundaries.

Add SingleDistinctAggregation after distribution enforcement and partial/final combination. It fuses eligible adjacent aggregates into the existing AggregateExec(Single). The existing single_distinct_prefer_native setting defaults to true; actual input partitioning determines eligibility. Unsupported shapes, including filtered, ordered, mixed, and multiple aggregates, keep their existing plans.

The cross-crate additions are DistinctCountValues, GroupsAccumulator::supports_distinct_count_spill, and GroupsAccumulator::take_distinct_count_values. They are documented and hidden as internal APIs, and the trait methods have defaults. The execution-plan and aggregate-state formats remain unchanged. The lockfile uses the patched quinn-proto 0.11.19 and quinn-udp 0.5.16 releases.

Performance

Compared with main 55de8615 on a shared Xeon Gold 6238T host, Rust 1.99.1, release-nonlto, CPU affinity 0–7, and four million preloaded rows. Two rounds reverse version order, discarding the first iteration of each run: eight retained samples per normal case and six per spill case. Builds and tests were idle during measurement. Times below are medians.

One input partition, greedy / 1 GiB:

Query Main ms This PR ms Elapsed change Main / PR peak pool MiB
Q01: high NDV 427.45 296.09 -30.7% 191.45 / 337.32
Q02: low NDV 107.09 82.55 -22.9% 1.07 / 1.51
Q03: eight groups, high NDV 388.03 292.19 -24.7% 191.45 / 337.14
Q06: Float64 control 388.85 424.18 +9.1% 191.45 / 191.45
Q07: skew 236.78 196.38 -17.1% 65.51 / 85.14
Q10: repeated singleton values 273.65 197.60 -27.8% 69.88 / 142.01
Q12: 95% NULL-only groups 642.03 637.06 -0.8% 55.19 / 142.01
Q13: one large group 222.44 199.33 -10.4% 48.13 / 85.14

One input partition, greedy / 16 MiB:

Query Main ms This PR ms Elapsed change Main / PR peak pool MiB
Q01 3343.20 3469.61 +3.8% 15.01 / 15.99
Q10 1522.69 1395.60 -8.3% 15.41 / 15.63
Q12 1189.55 1943.90 +63.4% 13.89 / 15.63
Q13 1799.21 1390.76 -22.7% 15.01 / 15.85

Four target partitions, greedy / 1 GiB; these plans retain parallel aggregation:

Source partitions Query Main ms This PR ms
1 Q01 150.51 146.61
1 Q10 107.67 119.84
4 Q01 137.12 153.06
4 Q10 116.19 116.18

The main tradeoffs are the NULL-heavy spill case and extra reservation for hash-table growth and conversion. The memory figures are pool reservations, not process RSS. Feedback on default selection would be particularly useful for that NULL-heavy case.

What is the testing strategy for this PR?

Forced-spill tests exercise groups larger than the budget, all eight integer types, signed and unsigned limits, cross-run duplicates, NULL-only groups, filters, dictionary keys, concurrent wide keys, greedy/fair pools, and resource cleanup. Planner integration tests cover actual versus configured partitions, expressions and compound keys, unsupported shapes, and ordered input. SQL tests cover the default, fallback, and low-memory execution paths.

The extended workspace suite passed 13,366 Rust tests, with 8 ignored, and all 305 SQL files. All 115 CLI tests passed. Validation also includes strict all-targets/all-features Clippy, the full lint suite, and 88 successful benchmark executions.

Are there any user-facing changes?

Eligible single-input-partition integer distinct counts use native aggregation by default and can spill a group larger than memory. Setting single_distinct_prefer_native=false disables the physical fusion. Aggregates that already have a native shape, such as Q12, still use the bounded spill implementation.

@github-actions github-actions Bot added documentation Improvements or additions to documentation logical-expr Logical plan and expressions optimizer Optimizer rules core Core DataFusion crate sqllogictest SQL Logic Tests (.slt) common Related to common crate functions Changes to functions implementation physical-plan Changes to the physical-plan crate labels Nov 13, 2026
@QinXi-ai
QinXi-ai marked this pull request as ready for review November 13, 2026 16:03
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

common Related to common crate core Core DataFusion crate documentation Improvements or additions to documentation functions Changes to functions implementation logical-expr Logical plan and expressions optimizer Optimizer rules physical-plan Changes to the physical-plan crate sqllogictest SQL Logic Tests (.slt)

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Grouped count(DISTINCT x) on numbers takes a slower path

1 participant