feat: stream exact group-contiguous aggregates

[xavlee]

Which issue does this PR close?

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

Rationale for this change

Analytics over large amounts of pre-partitioned data can combine many logical runs into fewer DataFusion output partitions. A source may know that every complete grouping tuple occupies one contiguous row range even when tuple values occur in arbitrary order.

For telemetry workloads, a tuple such as (bhandle, date_bin(timestamp)) can satisfy this invariant. Streaming completed groups releases aggregate state as the input advances and produces output before end of input.

What changes are included in this PR?

AggregateExec reads the cached group_contiguous_exprs tuple from its input PlanProperties and compares it with the complete GROUP BY tuple using input equivalence properties. Matching supports equivalent expressions and tuple permutation.

An exact complete-tuple match establishes GroupCompletionMode::Full. The input retains InputOrderMode::Linear for arbitrary tuple order, while DataFusion's existing full group-completion tables and streams emit each completed group when the next tuple begins.

For example, the following tuple order is unsorted, but every complete group occupies one contiguous run:

GROUP BY (key, time_bin)

input tuples:
  (A, 20), (A, 20) | (B, 20), (B, 20) | (A, 0), (A, 0)

AggregateExec.input_order_mode = InputOrderMode::Linear
input.group_contiguous_exprs = [key, time_bin]
input.pipeline_behavior = EmissionType::Incremental

group_contiguous_exprs == complete GROUP BY tuple
  -> 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 narrow group-contiguous source property
4. #24497 — feat: stream exact group-contiguous aggregates ← this PR

Are these changes tested?

Tests cover:

• a globally unsorted, group-contiguous input that retains InputOrderMode::Linear, establishes GroupCompletionMode::Full, reports EmissionType::Incremental, and selects OrderedSingleAggregateStream;
• complete-tuple matching, including permutation, equivalent expressions, and incomplete-tuple rejection;
• grouping-set boundaries;
• output ordering and assertion consumption;
• ordinary and group-contiguous benefits_from_input_partitioning behavior;
• completion-mode recomputation after child property changes;
• a projected date_bin grouping tuple with non-monotonic logical-run resets and correct aggregate output; and
• the range_sorted_time_bin_agg.slt scenario on a branch combined with feat: skip hash shuffle for date_bin/date_trunc on Range([timestamp]) #24501.

Are there any user-facing changes?

Sources and custom physical plans can certify a complete group-contiguous tuple. An aggregate whose complete grouping tuple matches that assertion uses full group completion and emits completed groups incrementally within each input partition.

Review this layer

View only this PR layer

[codecov-commenter]

Codecov Report

❌ Patch coverage is 75.53957% with 136 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.49%. Comparing base (ee59f62) to head (58bef24).
⚠️ Report is 9 commits behind head on main.

Files with missing lines	Patch %	Lines
datafusion/physical-plan/src/aggregates/mod.rs	81.94%	16 Missing and 36 partials ⚠️
datafusion/datasource/src/source.rs	40.98%	36 Missing ⚠️
datafusion/physical-plan/src/projection.rs	79.41%	5 Missing and 9 partials ⚠️
datafusion/physical-plan/src/test.rs	62.16%	13 Missing and 1 partial ⚠️
datafusion/physical-plan/src/execution_plan.rs	70.00%	6 Missing ⚠️
datafusion/physical-plan/src/buffer.rs	72.72%	0 Missing and 3 partials ⚠️
datafusion/physical-plan/src/coop.rs	75.00%	0 Missing and 3 partials ⚠️
datafusion/physical-plan/src/scalar_subquery.rs	76.92%	0 Missing and 3 partials ⚠️
...tafusion/physical-plan/src/aggregates/order/mod.rs	83.33%	2 Missing ⚠️
.../aggregates/aggregate_hash_table/common_ordered.rs	50.00%	0 Missing and 1 partial ⚠️
... and 2 more

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24497      +/-   ##
==========================================
+ Coverage   81.42%   81.49%   +0.06%     
==========================================
  Files        1121     1123       +2     
  Lines      402142   405273    +3131     
  Branches   402142   405273    +3131     
==========================================
+ Hits       327460   330271    +2811     
- Misses      55484    55633     +149     
- Partials    19198    19369     +171     

☔ 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