refactor: use group clustering terminology in aggregate execution by xavlee · Pull Request #24697 · apache/datafusion

xavlee · 2026-08-26T13:24:48Z

Which issue does this PR relate to?

Part of Supporting analytics over large amounts of pre‑partitioned data #24438.
Builds on the baseline test in test: cover unsorted contiguous groups in one partition #24737 (merged).

Rationale for this change

Aggregate execution currently uses InputOrderMode to select execution paths and recognize completed groups.

A group can be completed incrementally when its rows form one contiguous run: once the run ends, that group cannot appear again in the same input partition. Sorted input supplies this guarantee, and unsorted input can supply it too:

AAABBBCCCC -> sorted; groups can be completed incrementally
CCCAAABBB  -> unsorted; groups can still be completed incrementally

This PR describes the input's clustering guarantee with GroupClusteringMode and uses clustering terminology throughout the aggregate execution machinery. Aggregate construction derives the mode directly from the existing ordering analysis. The follow-up in #24497 will also derive it from grouped equivalence properties.

What changes are included in this PR?

Replace the aggregate's InputOrderMode field with GroupClusteringMode::{None, Partial, Full}:
- None: Groups have no apparent clustering guarantee.
- Partial(indices): rows are contiguous on a subset of grouping expressions; a change in that tuple completes all groups in the preceding run.
- Full: rows are contiguous on the complete grouping tuple; a change in that tuple completes the preceding group.
Rename the stateful GroupOrdering tracker and its Partial and Full implementations to GroupClustering, GroupClusteringPartial, and GroupClusteringFull.
Rename the ordered aggregate streams, shared table, and adjacent-key group-values implementation to use clustered terminology,
Preserve actual sort expressions and options through equivalence properties and input-ordering requirements.
Display group_clustering_mode in aggregate plans and document the public API migration in the DataFusion 56 upgrade guide.

Stack

Are there any user-facing changes?

Yes. This PR changes public Rust APIs:

AggregateExec::input_order_mode() is replaced by AggregateExec::group_clustering_mode(), returning the public GroupClusteringMode enum.
AggregateExec::compute_properties accepts &GroupClusteringMode in place of &InputOrderMode.
The public GroupOrdering, GroupOrderingPartial, and GroupOrderingFull types in aggregates::order are renamed to GroupClustering, GroupClusteringPartial, and GroupClusteringFull.
GroupClustering::try_new takes &GroupClusteringMode; new_group_values takes &GroupClustering.

Aggregate plan output uses group_clustering_mode=Full and group_clustering_mode=Partial(indices) in place of ordering_mode=Sorted and ordering_mode=PartiallySorted(indices). SQL results and aggregate execution-path selection retain their existing behavior.

Review this layer

View this PR's diff

codecov-commenter · 2026-08-26T13:59:18Z

Codecov Report

❌ Patch coverage is 85.75419% with 51 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.84%. Comparing base (f0ee13e) to head (2fd4763).
⚠️ Report is 5 commits behind head on main.

Files with missing lines	Patch %	Lines
datafusion/physical-plan/src/aggregates/mod.rs	83.13%	3 Missing and 11 partials ⚠️
...cal-plan/src/aggregates/clustered_single_stream.rs	79.66%	12 Missing ⚠️
...tafusion/physical-plan/src/aggregates/order/mod.rs	84.21%	2 Missing and 7 partials ⚠️
...hysical-plan/src/aggregates/grouped_hash_stream.rs	73.33%	1 Missing and 3 partials ⚠️
...sion/physical-plan/src/aggregates/order/partial.rs	92.50%	0 Missing and 3 partials ⚠️
...ggregates/aggregate_hash_table/common_clustered.rs	86.66%	0 Missing and 2 partials ⚠️
...plan/src/aggregates/aggregate_hash_table/common.rs	0.00%	0 Missing and 1 partial ⚠️
...c/aggregates/aggregate_hash_table/partial_table.rs	0.00%	0 Missing and 1 partial ⚠️
...ical-plan/src/aggregates/clustered_final_stream.rs	95.23%	1 Missing ⚠️
...n/physical-plan/src/aggregates/group_values/mod.rs	87.50%	0 Missing and 1 partial ⚠️
... and 3 more

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24697      +/-   ##
==========================================
- Coverage   82.85%   82.84%   -0.01%     
==========================================
  Files        1151     1151              
  Lines      455961   455880      -81     
  Branches   455961   455880      -81     
==========================================
- Hits       377768   377691      -77     
- Misses      55238    55247       +9     
+ Partials    22955    22942      -13

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

gene-bordegaray

approved with some non blocking suggestoins. Thank you @xavlee

gene-bordegaray · 2026-09-03T00:04:49Z

Rereviewed after changes and everything looks good 👍

NGA-TRAN

Looks great. It is a pleasure to review. Nice explanation.

I wonder if we should add an attribute in the explain to show this property? Maybe in a follow-up PR? If it is not that invasive or very easy to review, maybe adding that in this PR?

NGA-TRAN · 2026-09-03T20:29:18Z

+    ///
+    /// For example, with `GROUP BY (a, b)`, `Partial(vec![0])` means all rows
+    /// for each value of `a` are contiguous, while an `(a, b)` tuple may recur
+    /// within that range.


Does this mean I have 2 keys (a, b) and data is sorted on (a) only?

NGA-TRAN · 2026-09-03T20:31:59Z

+        assert_eq!(
+            aggregate.group_completion_mode,
+            GroupCompletionMode::Partial(vec![0])
+        );


NGA-TRAN · 2026-09-03T20:33:22Z

+        // This captures the behavior before #24438. When the source can declare
+        // `(key, time_bin)` group-contiguous, the corresponding case can use
+        // `EmissionType::Incremental`.
+        assert_eq!(aggregate.cache().emission_type, EmissionType::Final);


gene-bordegaray

this also looks good, just needs a rebase

xudong963

Nit: finish updating the runtime documentation from “ordered” to “group-complete.”

## Which issue does this PR relate to? - Part of apache#24438. ## Rationale for this change A source can concatenate several sorted logical runs into one DataFusion output partition. The resulting stream may be globally unsorted while every distinct `(key, time_bin)` tuple still occupies one contiguous range. This PR records how aggregate planning handles that layout before grouped input properties are available. It provides the behavioral baseline for the remaining PRs in the stack. ## What changes are included in this PR? - Add a single-partition `TestMemoryExec` fixture containing two sorted logical runs whose `(key, time_bin)` order resets at the record-batch boundary. - Aggregate by the complete `(key, time_bin)` tuple and verify the result. - Assert that planning selects `InputOrderMode::Linear`, `EmissionType::Final`, and `SingleHashAggregateStream`. ## Stack 1. [apache#24737 — test: cover unsorted contiguous groups in one partition](apache#24737) ← **this PR** 2. [apache#24697 — refactor: separate aggregate group completion from input ordering](apache#24697) 3. [apache#24698 — feat: add grouped equivalence properties](apache#24698) 4. [apache#24497 — feat: stream aggregates over grouped input](apache#24497) ## Are these changes tested? The new aggregate test executes the single-partition input and snapshot-checks all four grouped sums. ## Are there any user-facing changes? No. ## Review this layer [View only this PR layer](apache@36969e7)

alamb · 2026-10-14T20:40:23Z

What is the status of this PR? If it is ready to merge, perhaps we can resolve the conflicts and get the CI to pass

NGA-TRAN · 2026-10-14T22:00:39Z

@gene-bordegaray : Since Xavier is on vacation, I wonder whether you could help resolve conflicts and have CI passes so we can merge?

xavlee · 2026-10-14T22:05:59Z

Hi folks, I will rebase the PR and resolve the conflicts today.

xavlee · 2026-10-15T02:20:53Z

@alamb I resolved the conflicts and CI just passed so this guy should be ready to merge 🙇🏻

alamb · 2026-10-15T12:56:12Z

🤔 seems another conflict

@xudong963 and @jayzhan211 shall we merge this?

xavlee · 2026-10-15T13:02:44Z

I'll address these shortly. It seems like renaming GroupOrdering -> GroupClustering is prone to quite a lot of conflicts 😅

jayshrivastava · 2026-10-15T15:06:01Z

Instead of a "sorted" property, we instead declare clustering - this means a SortPreservingMerge is not possible between two clustered partitions right? DataFusion doesn't consider them to be sorted.

NGA-TRAN · 2026-10-15T15:18:34Z

Instead of a "sorted" property, we instead declare clustering - this means a SortPreservingMerge is not possible between two clustered partitions right? DataFusion doesn't consider them to be sorted.

Right, data is not sorted and thus I do not think SortPreservingMerge is possible

xavlee · 2026-10-15T23:49:25Z

clustering != sorted, so we cannot assume an input is sorted for SortPreservingMerge based on the clustering. Existing use cases of SortPreservingMerge remain unaffected since ordering metadata is preserved.

Clarification- we could have an input that is clustered and sorted (sorting implies clustering), which would allow us to use SortPreservingMerge.

xudong963 · 2026-10-16T09:30:35Z

Thanks all!

## Which issue does this PR relate to? - Part of apache#24438. - Builds on apache#24737 and apache#24697, both merged. ## Rationale for this change Aggregation can recognize completed groups when all rows with the same grouping-key tuple occur in one contiguous run within a partition. Sorted input provides this guarantee, and the runs can also appear in an arbitrary order: ```text AAABBBCCCC -> ordered and grouped CCCAAABBB -> grouped; each group can be completed when its run ends AAABBBAACC -> A appears in separate runs ``` This PR represents that guarantee in `EquivalenceProperties`, allowing execution-plan producers to declare grouping tuples and consumers to check whether an input satisfies a grouping requirement. ## What changes are included in this PR? ### Semantics and propagation A grouping assertion describes a complete expression tuple within every output partition. `[a, b]` and `[b, a]` identify the same groups. The guarantee applies to the complete pair; a guarantee for a subset such as `[a]` requires its own proof. Grouping assertions are correctness guarantees supplied by producers. DataFusion trusts the asserted row layout. Producers can declare derived tuples such as `[bhandle, date_bin(timestamp)]` directly. - Store grouping tuples alongside ordering, equality, and constant information in `EquivalenceProperties`. - Normalize tuples using expression equivalences and constants, removing duplicate expressions. - Check grouping requirements against declared tuples and grouping implied by lexicographical ordering prefixes. - Project a tuple when every expression can be represented in the output, including explicitly declared derived expressions. - Remap grouping expressions during schema replacement. - Preserve a declared grouping tuple through `UnionExec` when every input satisfies it, explicitly or through ordering. - Clear explicit grouping assertions at aggregation, sorting, repartitioning, and partition-merging boundaries where execution can invalidate them. ### Public API additions - `GroupingEquivalenceClass` represents complete grouping tuples and supports construction, insertion, lookup, clearing, schema rewriting, iteration, and display. - `EquivalenceProperties::{geq_class, normalized_geq_class, add_grouping, add_groupings, clear_groupings, grouping_satisfy}` expose grouping metadata and its analysis. - `TestMemoryExec::try_with_grouping_information` attaches grouping assertions in physical-plan tests and examples. Execution-plan producers attach these assertions through the `EquivalenceProperties` held by `PlanProperties`. The follow-up apache#24497 consumes them to derive `GroupClusteringMode::Full` for eligible aggregates. ## Stack 1. [apache#24737 — test: cover unsorted contiguous groups in one partition](apache#24737) (merged) 2. [apache#24697 — refactor: use group clustering terminology in aggregate execution](apache#24697) (merged) 3. [apache#24698 — feat: add grouping equivalence properties](apache#24698) ← **this PR** 4. [apache#24497 — feat: stream aggregates over grouped input](apache#24497) ## What is the testing strategy for this PR? Tests cover complete-tuple semantics, equivalent permutations, normalization through equivalences and constants, ordering-derived grouping, complete and incomplete projection mappings, explicitly declared derived expressions, operator boundaries, and UNION propagation with selective tuple retention, mixed grouping/ordering proofs, schema rewrites, and input permutations. `aggregate_clears_explicit_grouping` covers partial and single aggregation with reordered and aliased grouping keys. Both cases fail without the aggregate boundary clear and pass with it. Local validation passed the extended workspace suite, including 2,997 physical-plan tests and 546 SQL logic files, and all-target/all-feature Clippy with warnings denied. ## Are there any user-facing changes? Yes. The additive Rust APIs above let plan producers describe contiguous grouping tuples and let consumers evaluate grouping requirements. The streaming aggregate optimization follows in apache#24497. ## Review this layer [View this PR's diff](https://github.com/apache/datafusion/pull/24698/files)

Backport the Apache DataFusion apache#24438 stack onto DataDog branch-55: - apache#24737: unsorted contiguous-group regression coverage - apache#24697: group-clustering terminology and aggregate APIs - apache#24698: grouping equivalence properties and propagation - apache#24497: incremental aggregation over explicitly grouped input - apache#26852: invalidate grouping assertions when clearing orderings Adapt the changes to DF55's existing aggregate streams. Single-stage clustered aggregation continues to use GroupedHashAggregateStream; partial and final aggregation retain the existing specialized streams. Do not import the unrelated DF56 aggregate-stream migration. Include the constant-key, grouping-set, and PartialReduce guards. Extend blocked-input tests across single/final and migrated/legacy execution, and retain sort-option and metadata-invalidation coverage. Update DF55 plan snapshots, the wrapped memory-limit error assertion, and the API upgrade guide. Validation: 503 SQL test files passed, 3347 focused expression/plan unit tests passed, and all-targets/all-features clippy passed. Extended tests exposed a dictionary sort-fuzzer failure also reproduced on the unchanged branch-55 base (root seed 1793724857; dataset 15872832093508421016; query 12465732596605393044; config 11717754459111916353). Upstream-Commit: 59d74ca Upstream-Commit: d93699c Upstream-Commit: 320a7d2 Upstream-Commit: 64a5955 Upstream-Commit: 3c3898e Upstream-Commit: 86f93b6 Co-authored-by: xavlee <52435309+xavlee@users.noreply.github.com>

…ache#24697) Backport apache#24697 Adapt the clustering API and names to DF55's existing aggregate streams. Include the public API guide and plan-display expectation updates. Upstream-Commit: d93699c

DataFusion 56 renames OrderedFinalAggregateStream to ClusteredFinalAggregateStream (apache/datafusion#24697), and the memory consumer with it. The fair pool's exemption for that consumer and the spill replay pool's final-aggregate check match it by name, so on DataFusion 56 neither matched anymore. Match the new name, and update the tests, docs and comments that use the old one.

github-actions Bot added the physical-plan Changes to the physical-plan crate label Aug 26, 2026

xavlee mentioned this pull request Aug 26, 2026

feat: add grouping equivalence properties #24698

Merged

xavlee mentioned this pull request Aug 26, 2026

feat: stream aggregates over grouped input #24497

Open

xavlee force-pushed the refactor/aggregate-group-completion-mode branch 4 times, most recently from f58e128 to 372429f Compare August 27, 2026 20:54

xavlee mentioned this pull request Aug 28, 2026

test: cover unsorted contiguous groups in one partition #24737

Merged

NGA-TRAN mentioned this pull request Aug 31, 2026

Supporting analytics over large amounts of pre‑partitioned data #24438

Open

xavlee force-pushed the refactor/aggregate-group-completion-mode branch 2 times, most recently from d1d1d2f to 0267ad1 Compare August 31, 2026 17:50

gene-bordegaray approved these changes Sep 1, 2026

View reviewed changes

xavlee force-pushed the refactor/aggregate-group-completion-mode branch 4 times, most recently from e7f8b42 to 0108b7c Compare September 2, 2026 15:34

NGA-TRAN approved these changes Sep 3, 2026

View reviewed changes

xavlee force-pushed the refactor/aggregate-group-completion-mode branch from 0108b7c to 6e369ce Compare September 4, 2026 19:21

This was referenced Sep 7, 2026

A fixed GROUP BY column is incorrectly used to decide when groups are complete #25018

Closed

fix: ignore constant aggregate completion boundaries #25019

Merged

xavlee force-pushed the refactor/aggregate-group-completion-mode branch 2 times, most recently from 2fae45d to bb3dc16 Compare September 7, 2026 20:23

xavlee force-pushed the refactor/aggregate-group-completion-mode branch from bb3dc16 to 64a6e55 Compare September 16, 2026 14:24

gene-bordegaray approved these changes Sep 22, 2026

View reviewed changes

xavlee force-pushed the refactor/aggregate-group-completion-mode branch 2 times, most recently from 3a0ecb6 to 1fd156e Compare September 23, 2026 02:04

xavlee marked this pull request as ready for review September 23, 2026 19:30

xudong963 approved these changes Sep 28, 2026

View reviewed changes

xavlee force-pushed the refactor/aggregate-group-completion-mode branch from cca3e3f to 9402397 Compare October 14, 2026 22:56

xavlee added 4 commits October 15, 2026 22:04

refactor: separate aggregate group completion from input ordering

5d88524

docs: clarify aggregate completion migration guidance

68fb326

docs: describe group completion clustering guarantees

3700c42

refactor: unify aggregate group clustering terminology

2fd4763

xavlee force-pushed the refactor/aggregate-group-completion-mode branch from 9402397 to 2fd4763 Compare October 15, 2026 13:17

xudong963 added this pull request to the merge queue Oct 16, 2026

Merged via the queue into apache:main with commit d93699c Oct 16, 2026
43 checks passed

github-actions Bot added the v56.0.0 label Oct 16, 2026

QinXi-ai mentioned this pull request Oct 19, 2026

Resolve upstream clustering refactor conflicts for scalar group keys geoffreyclaude/datafusion#20

Closed

mbutrovich mentioned this pull request Oct 26, 2026

fix: spill a native PartialMerge whose input is sorted on some of its keys apache/datafusion-comet#7222

Merged

andygrove mentioned this pull request Oct 27, 2026

[EPIC] DataFusion 56 upgrade: regressions found by tracking DataFusion branch-56 apache/datafusion-comet#6410

Open

2 of 13 tasks

xavlee mentioned this pull request Oct 28, 2026

fix: clear groupings when clearing orderings #26852

Open

andygrove mentioned this pull request Nov 3, 2026

deps: upgrade to DataFusion 56.0.0 apache/datafusion-comet#7249

Draft

Conversation

xavlee commented Aug 26, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR relate to?

Rationale for this change

What changes are included in this PR?

Stack

Are there any user-facing changes?

Review this layer

Uh oh!

codecov-commenter commented Aug 26, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

gene-bordegaray left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

gene-bordegaray commented Sep 3, 2026

Uh oh!

NGA-TRAN left a comment

Choose a reason for hiding this comment

Uh oh!

NGA-TRAN Sep 3, 2026

Choose a reason for hiding this comment

Uh oh!

xavlee Sep 4, 2026

Choose a reason for hiding this comment

Uh oh!

NGA-TRAN Sep 3, 2026

Choose a reason for hiding this comment

Uh oh!

NGA-TRAN Sep 3, 2026

Choose a reason for hiding this comment

Uh oh!

gene-bordegaray left a comment

Choose a reason for hiding this comment

Uh oh!

xudong963 left a comment

Choose a reason for hiding this comment

Uh oh!

alamb commented Oct 14, 2026

Uh oh!

NGA-TRAN commented Oct 14, 2026

Uh oh!

xavlee commented Oct 14, 2026

Uh oh!

xavlee commented Oct 15, 2026

Uh oh!

alamb commented Oct 15, 2026

Uh oh!

xavlee commented Oct 15, 2026

Uh oh!

jayshrivastava commented Oct 15, 2026

Uh oh!

NGA-TRAN commented Oct 15, 2026

Uh oh!

xavlee commented Oct 15, 2026

Uh oh!

xudong963 commented Oct 16, 2026

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

9 participants

xavlee commented Aug 26, 2026 •

edited

Loading

codecov-commenter commented Aug 26, 2026 •

edited

Loading