perf(aggregate): specialize fully ordered group keys without hashing - #25150
Open
QinXi-ai wants to merge 1 commit into
Open
perf(aggregate): specialize fully ordered group keys without hashing#25150QinXi-ai wants to merge 1 commit into
QinXi-ai wants to merge 1 commit into
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
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?
GroupValuesOrderedselected only forGroupOrdering::Fulland 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.AggregateExecexecution.unused_asyncexpectation 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_emitscompares group IDs and emitted/preserved keys withGroupValuesColumn<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_executioncompares actualAggregateExecresults with unordered execution for Single and Partial→Final with FILTER, using four batch sizes and asserted sorted execution.Local validation on Windows:
cargo clippy --all-targets --all-features -- -D warningsand formatting checks pass.423df6f4: two datasource path tests, four floating-point interval/statistics tests, and the xtask CI command snapshot. These remain recorded failures.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
AggregateExecbenchmark uses 8,192-row batches. Speedup ranges below aremin(baseline) / max(candidate)throughmax(baseline) / min(candidate)across the two repetitions; they are not confidence intervals for the ratio.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_bytesreports peak retainedGroupValues::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.