feat: stream aggregates over grouped input

[xavlee]

Which issue does this PR close?

• Closes Supporting analytics over large amounts of pre‑partitioned data #24438.
• Builds on test: cover unsorted contiguous groups in one partition #24737, refactor: separate aggregate group completion from input ordering #24697, and feat: add Grouped equivalence properties #24698.

Rationale for this change

An aggregate can emit a completed group once it knows that the same complete grouping tuple cannot appear again in its input partition. EquivalenceProperties::grouping_satisfy now provides that guarantee for both lexicographically ordered input and explicitly grouped input.

For example, these complete (key, time_bin) groups are contiguous even though their values are not globally sorted:

(A, 20), (A, 20) | (B, 20), (B, 20) | (A, 0), (A, 0)

The aggregate can therefore release each group as its run ends while retaining InputOrderMode::Linear to describe the input order accurately.

What changes are included in this PR?

• Query the input equivalence properties for the complete GROUP BY tuple during AggregateExec construction.
• Select GroupCompletionMode::Full when grouping_satisfy proves that tuple is grouped and at least one grouping expression is non-constant.
• Report incremental emission for grouped input and select the existing ordered aggregate implementation.
• Consume explicit input grouping information at the aggregate boundary when computing output equivalence properties.
• Apply the grouped-input path to ordinary grouping expressions in partial, final, and single-stage aggregation; retain hash-based execution for PartialReduce.

input grouping: [[key, time_bin]]
GROUP BY:       [key, time_bin]
InputOrderMode: Linear

grouping_satisfy(GROUP BY)
  -> GroupCompletionMode::Full
  -> EmissionType::Incremental
  -> OrderedSingleAggregateStream

Stack

1. #24737 — test: cover unsorted contiguous groups in one partition
2. #24697 — refactor: separate aggregate group completion from input ordering
3. #24698 — feat: add grouped equivalence properties
4. #24497 — feat: stream aggregates over grouped input ← this PR

Are these changes tested?

The characterization test from #24737 declares its complete grouping tuple and verifies GroupCompletionMode::Full, EmissionType::Incremental, and OrderedSingleAggregateStream. It uses BarrierExec to hold back the second input batch until the first completed group is emitted, then checks the full result, including a group spanning both batches.

Regression tests cover PartialReduce with ordered and explicitly grouped input, and constant-only grouping with both a literal and a column made constant by a filter. A projection test verifies that a semantically equivalent date_bin grouping tuple reaches AggregateExec and enables incremental emission.

Are there any user-facing changes?

Yes. Aggregates whose complete grouping tuple is satisfied by the input equivalence properties can execute with full group completion and incremental emission on unsorted input.

Review this layer

View only this PR layer

[codecov-commenter]

Codecov Report

❌ Patch coverage is 79.40854% with 188 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.45%. Comparing base (ebdb657) to head (9c62dea).
⚠️ Report is 21 commits behind head on main.

Files with missing lines	Patch %	Lines
datafusion/physical-expr/src/equivalence/mod.rs	58.97%	3 Missing and 61 partials ⚠️
datafusion/physical-plan/src/aggregates/mod.rs	79.75%	12 Missing and 38 partials ⚠️
...tafusion/physical-expr/src/equivalence/grouping.rs	82.17%	13 Missing and 5 partials ⚠️
datafusion/physical-plan/src/test.rs	60.46%	16 Missing and 1 partial ⚠️
...on/physical-expr/src/equivalence/properties/mod.rs	90.00%	7 Missing and 5 partials ⚠️
datafusion/physical-plan/src/repartition/mod.rs	78.57%	0 Missing and 6 partials ⚠️
...atafusion/physical-plan/src/coalesce_partitions.rs	84.61%	0 Missing and 4 partials ⚠️
datafusion/physical-expr/src/expressions/cast.rs	81.25%	0 Missing and 3 partials ⚠️
...tafusion/physical-plan/src/aggregates/order/mod.rs	78.57%	3 Missing ⚠️
...n/physical-plan/src/sorts/sort_preserving_merge.rs	85.00%	0 Missing and 3 partials ⚠️
... and 6 more

Additional details and impacted files

@@           Coverage Diff            @@
##             main   #24497    +/-   ##
========================================
  Coverage   82.45%   82.45%            
========================================
  Files        1140     1141     +1     
  Lines      436434   437417   +983     
  Branches   436434   437417   +983     
========================================
+ Hits       359840   360655   +815     
- Misses      54840    54883    +43     
- Partials    21754    21879   +125     

☔ 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.

[alamb]

This PR models group contiguity as its own correctness property. It is intentionally neither an ordering guarantee nor an output-distribution guarantee.

Why does it need a new property? I think this notion is designed to be covered y the existing ordering / monotonic analyses

[alamb]

Oh, I see, somehow the external system knows the data is not sorted but is non overlapping

I think this is going to be really hard to manage / ensure through the plan -- we will need to ensure that every operator properly reports if it will propagate this property or not

I am not sure this is something we want to complicate datafusion with

[alamb]

@NGA-TRAN can you help evaluate this PR for its impact and if we will be able to keep this property in tact?

[xavlee]

Hi Andrew, thanks for taking an initial look. Apologies that this is still verymuch a draft.

I agree that propagating another physical property through the plan would be a little complicated. I was hoping we would narrow the group_contiguous_exprs assertion s.t. it:

• is declared explicitly by the data source
• defaults to absent on every execution operator
• may pass only through ProjectionExec when every expression maps
• is consumed by the only the first AggregateExec (and not present on the aggregate output)

[NGA-TRAN]

@alamb : I added a comment in the ticket

@NGA-TRAN can you help evaluate this PR for its impact

The impact of this PR is huge for telemetry use cases of AI frontiers as I described in the comment above

and if we will be able to keep this property in tact?

This property, like some properties, will be no longer available after certain operators so I think it would wok the same. I agree the propagation is a bit more complicated than usual but we work together to split this PR into smaller ones and will look into design carefully to avoid a lot of side effect. I think we would be able to make the design simpler

[github-actions]

Thank you for opening this pull request!

Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).

Details

     Cloning apache/main
    Building datafusion-expr v55.1.0 (current)
       Built [  32.058s] (current)
     Parsing datafusion-expr v55.1.0 (current)
      Parsed [   0.080s] (current)
    Building datafusion-expr v55.1.0 (baseline)
       Built [  31.658s] (baseline)
     Parsing datafusion-expr v55.1.0 (baseline)
      Parsed [   0.082s] (baseline)
    Checking datafusion-expr v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   1.292s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  66.275s] datafusion-expr
    Building datafusion-expr-common v55.1.0 (current)
       Built [  21.471s] (current)
     Parsing datafusion-expr-common v55.1.0 (current)
      Parsed [   0.020s] (current)
    Building datafusion-expr-common v55.1.0 (baseline)
       Built [  22.034s] (baseline)
     Parsing datafusion-expr-common v55.1.0 (baseline)
      Parsed [   0.022s] (baseline)
    Checking datafusion-expr-common v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.261s] 223 checks: 222 pass, 1 fail, 0 warn, 31 skip

--- failure enum_variant_added: enum variant added on exhaustive enum ---

Description:
A publicly-visible enum without #[non_exhaustive] has a new variant.
        ref: https://doc.rust-lang.org/cargo/reference/semver.html#enum-variant-new
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.50.0/src/lints/enum_variant_added.ron

Failed in:
  variant SortProperties:Grouped in /home/runner/work/datafusion/datafusion/datafusion/expr-common/src/sort_properties.rs:42

     Summary semver requires new major version: 1 major and 0 minor checks failed
    Finished [  44.520s] datafusion-expr-common
    Building datafusion-ffi v55.1.0 (current)
       Built [  59.826s] (current)
     Parsing datafusion-ffi v55.1.0 (current)
      Parsed [   0.066s] (current)
    Building datafusion-ffi v55.1.0 (baseline)
       Built [  62.193s] (baseline)
     Parsing datafusion-ffi v55.1.0 (baseline)
      Parsed [   0.067s] (baseline)
    Checking datafusion-ffi v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.277s] 223 checks: 222 pass, 1 fail, 0 warn, 31 skip

--- failure enum_variant_added: enum variant added on exhaustive enum ---

Description:
A publicly-visible enum without #[non_exhaustive] has a new variant.
        ref: https://doc.rust-lang.org/cargo/reference/semver.html#enum-variant-new
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.50.0/src/lints/enum_variant_added.ron

Failed in:
  variant FFI_SortProperties:Grouped in /home/runner/work/datafusion/datafusion/datafusion/ffi/src/expr/expr_properties.rs:70

     Summary semver requires new major version: 1 major and 0 minor checks failed
    Finished [ 123.886s] datafusion-ffi
    Building datafusion-functions v55.1.0 (current)
       Built [  36.340s] (current)
     Parsing datafusion-functions v55.1.0 (current)
      Parsed [   0.092s] (current)
    Building datafusion-functions v55.1.0 (baseline)
       Built [  35.650s] (baseline)
     Parsing datafusion-functions v55.1.0 (baseline)
      Parsed [   0.090s] (baseline)
    Checking datafusion-functions v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.408s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  74.095s] datafusion-functions
    Building datafusion-physical-expr v55.1.0 (current)
       Built [  32.777s] (current)
     Parsing datafusion-physical-expr v55.1.0 (current)
      Parsed [   0.052s] (current)
    Building datafusion-physical-expr v55.1.0 (baseline)
       Built [  33.325s] (baseline)
     Parsing datafusion-physical-expr v55.1.0 (baseline)
      Parsed [   0.053s] (baseline)
    Checking datafusion-physical-expr v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.350s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  67.430s] datafusion-physical-expr
    Building datafusion-physical-optimizer v55.1.0 (current)
       Built [  46.260s] (current)
     Parsing datafusion-physical-optimizer v55.1.0 (current)
      Parsed [   0.023s] (current)
    Building datafusion-physical-optimizer v55.1.0 (baseline)
       Built [  46.935s] (baseline)
     Parsing datafusion-physical-optimizer v55.1.0 (baseline)
      Parsed [   0.024s] (baseline)
    Checking datafusion-physical-optimizer v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.117s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  94.447s] datafusion-physical-optimizer
    Building datafusion-physical-plan v55.1.0 (current)
       Built [  42.627s] (current)
     Parsing datafusion-physical-plan v55.1.0 (current)
      Parsed [   0.175s] (current)
    Building datafusion-physical-plan v55.1.0 (baseline)
       Built [  43.587s] (baseline)
     Parsing datafusion-physical-plan v55.1.0 (baseline)
      Parsed [   0.180s] (baseline)
    Checking datafusion-physical-plan v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.652s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  88.462s] datafusion-physical-plan

[gene-bordegaray]

overall looks great, thank you @xavlee

[gene-bordegaray]

Great the last commit addresses. Thank you @xavlee , I think this stack is ready 👍