Repository navigation
Add GroupValuesFullyOrdered mode to GroupValues trait for aggregate grouping. - #9662
Closed
mustafasrepo wants to merge 15 commits into
Closed
mustafasrepo wants to merge 15 commits into
mustafasrepo wants to merge 15 commits into
Conversation
Contributor
|
I plan to review this carefully over the next day or two |
6 of 7 tasks
alamb
reviewed
Mar 19, 2024
alamb
left a comment
alamb
left a comment
Contributor
There was a problem hiding this comment.
Thank you @mustafasrepo -- I think this is a neat idea, but I am not sure it will always be an improvement. I suspect it will do very well for data with low cardinality groups but may actually perform worse for data with high cardinality groups
In order to merge it I think we should:
- Add a benchmark to DataFusion that we can repeatably run to ensure we don't regress this performance. Maybe it is time to add a "clickbench sorted" type benchmark to
bench.shthat sorts the clickbench dataset and uses that sort order 🤔 - Add some additional tests to cover the edge cases where the previous group is duplicated in the next batch -- perhaps we can extend https://github.com/apache/arrow-datafusion/blob/main/datafusion/core/tests/fuzz_cases/aggregate_fuzz.rs
where source has 1000 rows, with ordering [a ASC, b ASC, c ASC], and group by expression is GROUP BY a, b. Here are the results for different expected hit values for each distinct group
What data types were columns a, b and c?
| use datafusion_expr::EmitTo; | ||
| use datafusion_physical_expr::LexOrdering; | ||
|
|
||
| /// A [`GroupValues`] making use of order |
Contributor
There was a problem hiding this comment.
Can we please expand this description (perhaps using some of the content in the description of this PR
Notable points include that the input to the group values are sure to be totally ordered
| /// The output schema | ||
| schema: SchemaRef, | ||
|
|
||
| /// The actual group by values, stored as `Vec<ScalarValue>` for each distinct group. |
Contributor
There was a problem hiding this comment.
The idea that the overhead of using Vec<ScalarValue> to store the row is better than having to convert all the rows into RowFormat I think makes sense for low cardinality (when there are relatively few groups) but for high cardinality (when there are many distinct groups) this may not hold
I
Contributor
There was a problem hiding this comment.
Another possibility might be to use something like the RecordBatchStore from the TopK implementation:
https://github.com/apache/arrow-datafusion/blob/09747596fd75bfce8903e86472cccb8acc524453/datafusion/physical-plan/src/topk/mod.rs#L537-L551
And totally avoid copying rows at all 🤔
Contributor
|
Marking as draft as I think this PR is no longer waiting on feedback. Please mark it as ready for review when it is ready for another look |
alamb
marked this pull request as draft
March 20, 2024 23:57
Contributor
Author
Sure, once I address them. I will let you know. Thanks for the review @alamb . |
github-actions
Bot
added
logical-expr
|
Thank you for your contribution. Unfortunately, this pull request is stale because it has been open 60 days with no activity. Please remove the stale label or comment or this will be closed in 7 days. |
This was referenced Sep 10, 2026
mohitgurav20
pushed a commit
to mohitgurav20/datafusion
that referenced
this pull request
Oct 1, 2026
…pache#25150) ## Which issue does this PR close? Closes apache#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 apache#9662, particularly its single-column regression and requests for repeatable measurements and cross-batch tests. It is distinct from apache#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. Validation after the September 26 conflict-resolution rebase onto `main` at `6a792c6` (Windows, Rust 1.98.1): `cargo fmt --all -- --check`, `git diff --check`, `cargo test --offline --locked -p datafusion-physical-plan --lib` (**2,323 passed, 0 failed**), focused physical-plan library Clippy with `-D warnings`, and `cargo check --offline --locked -p datafusion-physical-plan --bench ordered_group_values` passed. Full `cargo clippy --all-targets --all-features -- -D warnings` was attempted but stopped while linking `protobuf-src`'s `protoc` on this Windows machine (`LNK2019`/`LNK1120`); it is not a passing full lint result. Fresh Linux CI remains to be seen for the rebased head. Earlier local validation on Windows, before this rebase: - 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. Linux CI passed on the pre-rebase head `a62b640`. The following benchmark measurements predate the September 26 rebase and have not been rerun against `6a792c6`. They were collected 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. --------- Co-authored-by: hushen <190065939+918154429@users.noreply.github.com> Co-authored-by: kosiew <kosiew@gmail.com>
Omega359
pushed a commit
to Omega359/arrow-datafusion
that referenced
this pull request
Oct 11, 2026
…pache#25150) ## Which issue does this PR close? Closes apache#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 apache#9662, particularly its single-column regression and requests for repeatable measurements and cross-batch tests. It is distinct from apache#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. Validation after the September 26 conflict-resolution rebase onto `main` at `6a792c6` (Windows, Rust 1.98.1): `cargo fmt --all -- --check`, `git diff --check`, `cargo test --offline --locked -p datafusion-physical-plan --lib` (**2,323 passed, 0 failed**), focused physical-plan library Clippy with `-D warnings`, and `cargo check --offline --locked -p datafusion-physical-plan --bench ordered_group_values` passed. Full `cargo clippy --all-targets --all-features -- -D warnings` was attempted but stopped while linking `protobuf-src`'s `protoc` on this Windows machine (`LNK2019`/`LNK1120`); it is not a passing full lint result. Fresh Linux CI remains to be seen for the rebased head. Earlier local validation on Windows, before this rebase: - 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. Linux CI passed on the pre-rebase head `a62b640`. The following benchmark measurements predate the September 26 rebase and have not been rerun against `6a792c6`. They were collected 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. --------- Co-authored-by: hushen <190065939+918154429@users.noreply.github.com> Co-authored-by: kosiew <kosiew@gmail.com>
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 #.
Rationale for this change
Currently grouping calculation for the group by expressions is done using 3 modes:
GroupValuesByesspecialization for group by expression that have typeUtf8/LargeUtf8/Binary/LargeBinary(group by expression is single)GroupValuesPrimitivespecialization for single column primitive types.GroupValuesRowsfor other cases(group by expression count is larger than > 1 or underlying type of the group by is not primitive.)All grouping calculation is done using hash calculation for above implementations.
We can also specialize for fully ordered group by expressions. Where all of the group by expressions defines an ordering (This corresponds to
input_order_mode=InputOrderMode::Sortedin theAggregateExec).For these cases, we can calculate distinct groups using the
evaluate_partition_rangeshelper given group by values are ordered at the input. This eliminates the need for hash calculations.This PR adds support for
GroupValuesFullyOrderedgrouping for above cases.I have compared the performance of the
GroupValuesFullyOrderedwithGroupValuesRowsfor following plan:where source has 1000 rows, with ordering
[a ASC, b ASC, c ASC], and group by expression isGROUP BY a, b. Here are the results for different expected hit values for each distinct groupI have compared the performance of the
GroupValuesFullyOrderedwithGroupValuesRowsfor following plan:where source has 1000 rows, with ordering
[a ASC, b ASC, c ASC], and group by expression isGROUP BY a(e.g single expressions). Here are the results for different expected hit values for each distinct groupAccording to these results, for single group by expressions using
GroupValuesPrimitiveis still better even if group by expression is still fully ordered. However, when group by expression length is larger than 1, it is beneficial to useGroupValuesFullyOrderedinstead ofGroupValuesRowswhen group by expressions are fully ordered.What changes are included in this PR?
Are these changes tested?
Yes,
Are there any user-facing changes?