feat: stream aggregates for group-contiguous input

[xavlee]

Which issue does this PR close?

• Closes Support streaming aggregates when partitions are unsorted but non‑overlapping #24438.

Rationale for this change

Some data sources range-partition data into logical runs that are locally grouped by (key, time) and non-overlapping on (key, date_bin(time)). Combining many such runs into fewer DataFusion output partitions keeps target_partitions near the available CPU count, but the key ranges can reset between runs and therefore no longer form a global sort order.

The input is still streamable: within each output stream, every (key, date_bin(time)) tuple occurs in one contiguous range. Once the next tuple begins, the aggregate knows that the previous group is complete. DataFusion currently has no way to express that guarantee independently of sort order, so AggregateExec falls back to blocking hash aggregation.

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

What changes are included in this PR?

• Add ExecutionPlan::group_contiguous_exprs, PlanProperties::with_group_contiguous_exprs, and a corresponding DataSource hook. The expressions are the components of one composite tuple whose values must each occupy at most one contiguous range within every output stream.
• Store the property in PlanProperties, so row-transparent wrappers that copy cached properties (including CooperativeExec) preserve it without special-case forwarding.
• Project the complete composite tuple through ProjectionExec. If any component cannot be mapped, the property is cleared. A source may declare a derived expression such as date_bin(time) after validating its run boundaries; the projection does not infer that property from time alone.
• Add a private GroupCompletionMode to AggregateExec, separate from InputOrderMode:
  • an exact match between the contiguous tuple and the grouping tuple enables full group completion;
  • a strict subset enables partial group completion for additional grouping expressions;
  • the input remains InputOrderMode::Linear, and the aggregate advertises no false output ordering.
• Reuse the existing ordered aggregate state machines for early emission, including partial, final, single-stage, and spill-replay paths. The first aggregate consumes the property, and grouping sets remain on the conservative path.
• Avoid optimizer-inserted round-robin repartitioning when the source property is what enables streaming, since that exchange would invalidate the guarantee.
• Append the property callback to FFI_PlanProperties, with both local and separately compiled cross-library coverage.
• Add a Criterion benchmark using descending, non-overlapping logical runs so the global key order deliberately resets.

This is complementary to #24501. That PR uses range partitioning to avoid a hash shuffle across DataFusion output partitions; this PR lets aggregation stream within each output partition after the source combines multiple non-overlapping logical runs. Neither property substitutes for the other.

Are these changes tested?

Yes. New coverage includes:

• output before an unbounded input reaches end-of-stream;
• non-monotonic key-range resets;
• exact and partial composite-key matches;
• date_bin projection and all-or-nothing projection propagation;
• aggregate combination, distribution enforcement, and the default cooperative wrapper;
• migrated aggregation, forced spill, and spill replay;
• native and cross-library FFI round trips.

Validation on the final commit:

• cargo test -p datafusion-physical-plan --lib (1,780 tests)
• cargo test -p datafusion-physical-optimizer (34 tests)
• cargo test -p datafusion-datasource --lib (178 tests)
• cargo test -p datafusion --test core_integration physical_optimizer (553 tests)
• cargo test -p datafusion-ffi
• cargo test -p datafusion-ffi --features integration-tests
• cargo clippy -p datafusion-ffi --all-targets --all-features -- -D warnings
• cargo clippy --all-targets --all-features -- -D warnings
• rustdoc for the affected crates with -D warnings
• the feat: skip hash shuffle for date_bin/date_trunc on Range([timestamp]) #24501 range_sorted_time_bin_agg.slt test on a combined branch

The benchmark measured approximately 937 µs for unordered hash aggregation and 746 µs for group-contiguous streaming on the development machine (about 20% lower latency for this workload).

cargo-semver-checks passes for datafusion, datafusion-datasource, datafusion-physical-plan, and datafusion-physical-optimizer. It reports the expected added-field break for the appended FFI_PlanProperties::group_contiguous_exprs callback.

Are there any user-facing changes?

Yes. Data source and execution-plan implementors can opt into the new property. This is a correctness contract: an invalid declaration can cause an aggregate group to be emitted before all of its rows have been observed.

Appending the callback changes the FFI struct layout, so this PR requires the api change label.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 77.14286% with 176 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.31%. Comparing base (f951020) to head (6718bbf).

Files with missing lines	Patch %	Lines
datafusion/physical-plan/src/aggregates/mod.rs	75.76%	70 Missing and 56 partials ⚠️
datafusion/physical-plan/src/test.rs	69.56%	13 Missing and 1 partial ⚠️
datafusion/physical-plan/src/projection.rs	75.00%	5 Missing and 8 partials ⚠️
datafusion/datasource/src/source.rs	50.00%	7 Missing and 1 partial ⚠️
datafusion/physical-optimizer/src/ensure_coop.rs	78.26%	1 Missing and 4 partials ⚠️
...tafusion/physical-plan/src/aggregates/order/mod.rs	75.00%	4 Missing ⚠️
datafusion/ffi/src/execution_plan.rs	91.66%	0 Missing and 2 partials ⚠️
datafusion/ffi/src/plan_properties.rs	96.00%	0 Missing and 1 partial ⚠️
.../aggregates/aggregate_hash_table/common_ordered.rs	50.00%	0 Missing and 1 partial ⚠️
...hysical-plan/src/aggregates/grouped_hash_stream.rs	0.00%	0 Missing and 1 partial ⚠️
... and 1 more

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24497      +/-   ##
==========================================
- Coverage   81.32%   81.31%   -0.01%     
==========================================
  Files        1117     1117              
  Lines      396617   397331     +714     
  Branches   396617   397331     +714     
==========================================
+ Hits       322549   323091     +542     
- Misses      55201    55294      +93     
- Partials    18867    18946      +79     

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

[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 v55.0.0 (current)
       Built [  59.890s] (current)
     Parsing datafusion v55.0.0 (current)
      Parsed [   0.035s] (current)
    Building datafusion v55.0.0 (baseline)
       Built [  58.596s] (baseline)
     Parsing datafusion v55.0.0 (baseline)
      Parsed [   0.035s] (baseline)
    Checking datafusion v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.606s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [ 120.828s] datafusion
    Building datafusion-datasource v55.0.0 (current)
       Built [  43.212s] (current)
     Parsing datafusion-datasource v55.0.0 (current)
      Parsed [   0.031s] (current)
    Building datafusion-datasource v55.0.0 (baseline)
       Built [  43.285s] (baseline)
     Parsing datafusion-datasource v55.0.0 (baseline)
      Parsed [   0.032s] (baseline)
    Checking datafusion-datasource v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.246s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  87.919s] datafusion-datasource
    Building datafusion-ffi v55.0.0 (current)
       Built [  55.029s] (current)
     Parsing datafusion-ffi v55.0.0 (current)
      Parsed [   0.063s] (current)
    Building datafusion-ffi v55.0.0 (baseline)
       Built [  55.499s] (baseline)
     Parsing datafusion-ffi v55.0.0 (baseline)
      Parsed [   0.066s] (baseline)
    Checking datafusion-ffi v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.262s] 223 checks: 222 pass, 1 fail, 0 warn, 31 skip

--- failure constructible_struct_adds_field: struct exhaustively constructible through public API adds field ---

Description:
A pub struct that could be exhaustively constructed with a literal using only public API has a new pub field, breaking existing exhaustive literals.
        ref: https://doc.rust-lang.org/reference/expressions/struct-expr.html
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.50.0/src/lints/constructible_struct_adds_field.ron

Failed in:
  field FFI_PlanProperties.group_contiguous_exprs in /home/runner/work/datafusion/datafusion/datafusion/ffi/src/plan_properties.rs:72

     Summary semver requires new major version: 1 major and 0 minor checks failed
    Finished [ 112.249s] datafusion-ffi
    Building datafusion-physical-optimizer v55.0.0 (current)
       Built [  41.432s] (current)
     Parsing datafusion-physical-optimizer v55.0.0 (current)
      Parsed [   0.021s] (current)
    Building datafusion-physical-optimizer v55.0.0 (baseline)
       Built [  40.683s] (baseline)
     Parsing datafusion-physical-optimizer v55.0.0 (baseline)
      Parsed [   0.023s] (baseline)
    Checking datafusion-physical-optimizer v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.116s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  83.301s] datafusion-physical-optimizer
    Building datafusion-physical-plan v55.0.0 (current)
       Built [  38.462s] (current)
     Parsing datafusion-physical-plan v55.0.0 (current)
      Parsed [   0.151s] (current)
    Building datafusion-physical-plan v55.0.0 (baseline)
       Built [  38.793s] (baseline)
     Parsing datafusion-physical-plan v55.0.0 (baseline)
      Parsed [   0.152s] (baseline)
    Checking datafusion-physical-plan v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.626s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  79.271s] datafusion-physical-plan