QinXi-ai opened a new pull request, #25150: URL: https://github.com/apache/datafusion/pull/25150
## Which issue does this PR close? Closes #14574 for supported fully ordered multi-column keys. Single-column specializations and additional key types remain on their existing paths. ## Rationale for this change When input is already sorted by all grouping expressions, DataFusion can emit completed groups early, but its multi-column group-key mapping still hashes each row and maintains a hash table. This adds CPU and retained state even though equal keys must be consecutive. ## What changes are included in this PR? - Add an internal columnar `GroupValuesOrdered` selected only for `GroupOrdering::Full` and supported multi-column schemas. Arrow's partition kernel identifies adjacent runs; existing column builders retain representatives and compare the boundary key across batches. Dense IDs, partial emission, preserving selections, and reset behavior follow the existing contracts. - Support boolean, integer, string, and binary keys, including their large/view variants. Keep single-column, float, nested, dictionary, partially ordered, and unordered inputs on established paths. - Add differential correctness tests and a reproducible benchmark covering two key schemas, batch sizes 127/8192, and run lengths 1/8/128/8192. Include both grouping with emission and complete ordered `AggregateExec` execution. - Include two small Windows lint compatibility fixes needed to run the repository's full Clippy command: remove a redundant semicolon in rounding and add a Windows-only `unused_async` expectation to the unsupported FIFO example's uniform async entry point. This builds on the problem and review in #9662, particularly its single-column regression and requests for repeatable measurements and cross-batch tests. It is distinct from #23986's consecutive-key cache for clustered input, which retains the general hash path. This PR optimizes existing streaming aggregation; it does not introduce streaming aggregation or change whole-query asymptotic complexity. ## What is the testing strategy for this PR? `ordered_keys_match_hash_grouping_across_batches_and_emits` compares group IDs and emitted/preserved keys with `GroupValuesColumn<true>` for 15 key types, ascending/descending runs, NULLs, long strings, six batch sizes, and four emission limits. It also exercises empty batches and reset/reuse. A schema gate test covers excluded types. `ordered_single_and_partial_final_match_unordered_execution` compares actual `AggregateExec` results with unordered execution for Single and Partial→Final with FILTER, using four batch sizes and asserted sorted execution. Local validation on Windows: - Rust 1.97.0: full `cargo clippy --all-targets --all-features -- -D warnings` and formatting checks pass. - Rust 1.98.1, CI profile, extended workspace features: 11,333 tests passed across recorded runs, including all 2,050 physical-plan tests, 120 fuzz tests, and the two 5M-row SMJ low-memory runners. Eight tests are ignored, with no pending test programs. Seven baseline-reproduced failures remain, so this is not a full-workspace pass. - Seven failures reproduce with the same assertion differences on unmodified baseline `423df6f4`: two datasource path tests, four floating-point interval/statistics tests, and the xtask CI command snapshot. These remain recorded failures. - Windows library backtrace capture was disabled for the extended run because it made error-path tests exceptionally slow. The common crate's 650 tests, including backtrace behavior, pass separately with capture enabled. A CRLF checkout of the Q15 fixture was normalized to its original LF contents so the existing multi-statement skip works as intended; no test assertion was changed. - An external process interruption required resuming missing tests from Cargo's compiled artifact list. Completed suite summaries and unique individual test results were retained; this does not represent one successful full-workspace Cargo invocation. Upstream Linux CI has not run yet. Repeated benchmark results on an Intel Core Ultra 7 258V / Windows, Rust 1.98.1, `release-nonlto`, pinned to logical CPU 0, with no concurrent task builds/tests. Baseline and candidate were built from separate source directories with identical lockfiles. The run order was baseline, candidate, candidate, baseline; each case used 250 ms warmup, 10 samples, and at least 1 second of measurement. The complete ordered `AggregateExec` benchmark uses 8,192-row batches. Speedup ranges below are `min(baseline) / max(candidate)` through `max(baseline) / min(candidate)` across the two repetitions; they are not confidence intervals for the ratio. | Keys / rows per group | Baseline ms (two runs) | Candidate ms (two runs) | Speedup range | |---|---:|---:|---:| | int / 1 | 6.086, 5.269 | 1.609, 1.857 | 2.84–3.78x | | int / 8 | 2.434, 2.296 | 0.480, 0.609 | 3.77–5.07x | | int / 128 | 2.112, 1.949 | 0.549, 0.598 | 3.26–3.85x | | int / 8192 | 2.162, 2.059 | 0.545, 0.553 | 3.73–3.97x | | string / 1 | 9.616, 9.665 | 4.793, 5.521 | 1.74–2.02x | | string / 8 | 4.333, 4.422 | 1.697, 1.660 | 2.55–2.66x | | string / 128 | 3.723, 3.464 | 1.747, 1.638 | 1.98–2.27x | | string / 8192 | 3.907, 3.745 | 1.613, 1.607 | 2.32–2.43x | All 16 selected grouping microbenchmark cases also have faster point estimates in both repetitions; the weakest cross-repetition ratio is approximately 1.06x for long strings in 127-row batches. Small differences on one machine should not be generalized across platforms. At batch size 8192 / run length 1, peak retained group state drops from 425,792 to 196,608 bytes for integer keys and from 950,080 to 720,896 bytes for integer/long-string keys. The benchmark pre-generates 131,072 rows and excludes input generation, sorting, SQL planning, and I/O. The grouping benchmark includes partial/final emission; the execution benchmark includes SUM and output collection on already ordered input. `group_state_bytes` reports peak retained `GroupValues::size()`, excluding shared inputs and transient kernel allocations; it is not process RSS. ## Are there any user-facing changes? Eligible fully ordered multi-column aggregations use less grouping state and avoid hashing. No SQL syntax, result semantics, configuration, or public API changes. This contribution is AI-assisted. -- 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]
