QinXi-ai opened a new pull request, #27405: URL: https://github.com/apache/datafusion/pull/27405
## 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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
