fix: reject unsupported Substrait aggregation phases

[goutamadwant]

Which issue does this PR close?

• Closes Substrait consumer ignores AggregateFunction.phase, so an intermediate aggregate runs as a complete one #24967.

Rationale for this change

The Substrait consumer treats explicit intermediate aggregate phases as complete calls. This can silently return final values when a plan requests intermediate state, or report an unrelated root-schema naming error.

What changes are included in this PR?

• Validate phases in aggregate and window expressions before translating their arguments.
• Accept INITIAL_TO_RESULT and retain UNSPECIFIED for compatibility with existing DataFusion-produced plans.
• Reject explicit intermediate phases and unknown protobuf enum values with clear errors.
• Document the compatibility limitation: Substrait defines UNSPECIFIED as INTERMEDIATE_TO_RESULT, but the consumer cannot distinguish legacy DataFusion complete calls from unspecified intermediate-state calls produced elsewhere. Those calls remain accepted. The producer change in fix: Set Substrait aggregation phase to INITIAL_TO_RESULT #25146 is complementary and is not duplicated here.

What is the testing strategy for this PR?

• Reproduced the original behavior: an INITIAL_TO_INTERMEDIATE average over values 1 and 2 returned 1.5 instead of intermediate state.
• Tests cover supported phases, every explicit unsupported phase, rooted and unrooted aggregates, unknown binary-protobuf enum values, and actual window output.
• The full Substrait integration target passes: 213 tests passed, with six existing tests ignored.
• The extended workspace suite passes 11,267 Rust tests, with eight existing tests ignored, and all 511 SQL logic-test files using an explicit four-thread limit. An initial default-concurrency run failed one ordered-aggregate spill test with a test-memory-pool exhaustion; the complete limited-concurrency rerun passed without source changes.
• cargo clippy --all-targets --all-features -- -D warnings and the complete ./dev/rust_lint.sh pass, including strict workspace documentation checks.

Are there any user-facing changes?

Plans with explicit unsupported aggregate or window phases now fail instead of being interpreted as complete calls. Existing unspecified-phase plans remain accepted, including the ambiguity described above. No public Rust API changes are included. Intermediate-state execution and the separate AVG output-type mismatch are not addressed here.

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 81.90%. Comparing base (1b6dc92) to head (f86799b).
⚠️ Report is 92 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #25045      +/-   ##
==========================================
+ Coverage   81.72%   81.90%   +0.18%     
==========================================
  Files        1127     1134       +7     
  Lines      416310   425269    +8959     
  Branches   416310   425269    +8959     
==========================================
+ Hits       340224   348333    +8109     
- Misses      56095    56294     +199     
- Partials    19991    20642     +651     

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

[kumarUjjawal]

Hi @goutamadwant thank you for the fix, though I see alexandrefimov mentioned opening a PR. Can you please coordinate with the issue author so we don't duplicate the work.

[namanjain24-sudo]

Not a committer, so this is not a binding review. I have been working in the same corner of the producer, so I read this closely.

I checked that the two gates cover every path: from_substrait_agg_func is reached only from rel/aggregate_rel.rs:98, and from_window_function only from substrait_consumer.rs:332, so there is no third place where a phase could slip past. The -1 and 12345 cases in aggregate_invalid_phase are a good touch, since try_from is the only thing standing between an unknown enum value and silent acceptance.

One thing worth pinning down in the code, not just in review. The accept-list treats UNSPECIFIED as a complete call and rejects INTERMEDIATE_TO_RESULT. On the spec's own terms those are the same value:

enum AggregationPhase {
  // Implies `INTERMEDIATE_TO_RESULT`.
  AGGREGATION_PHASE_UNSPECIFIED = 0;

That text is identical in the pinned 0.63.0 crate and on substrait main today, so the rule as written accepts a value the spec defines as meaning exactly the thing the next arm rejects.

I think the carve-out is the right call anyway, because DataFusion's own producer has been emitting UNSPECIFIED for every aggregate and window call, and plans written by released versions have to keep loading. But that is a deliberate, DataFusion-specific exception rather than a reading of the spec, and the comment currently reads as though the two were the same thing. Something like "the spec says UNSPECIFIED implies INTERMEDIATE_TO_RESULT; we accept it regardless because DataFusion's own producer emitted it, so rejecting it would break plans written by released versions" would stop the next reader from treating it as spec-derived.

It also leaves a residual hole worth naming in the comment: a non-DataFusion producer that emits UNSPECIFIED meaning intermediate state still gets executed as a complete aggregation, which is the failure #24967 describes. This PR closes it for the three explicit phases and leaves it open for the default one.

For disclosure, #25146 is mine, and it stops the producer from emitting UNSPECIFIED. It does not conflict with this; if anything the two fit together, since after it the carve-out here covers only plans from older versions rather than everything DataFusion currently writes. If this lands first I am happy to rebase onto it.

Minor, take or leave: validate_aggregation_phase lives in expr/aggregate_function.rs and is pulled into the window path via use super::aggregate_function::.... Since both callers are peers it might read better next to the shared consumer helpers, but that is taste.

The branch is 65 commits behind main now, so it will want a rebase before it can go in.

[goutamadwant]

Not a committer, so this is not a binding review. I have been working in the same corner of the producer, so I read this closely.

I checked that the two gates cover every path: from_substrait_agg_func is reached only from rel/aggregate_rel.rs:98, and from_window_function only from substrait_consumer.rs:332, so there is no third place where a phase could slip past. The -1 and 12345 cases in aggregate_invalid_phase are a good touch, since try_from is the only thing standing between an unknown enum value and silent acceptance.

One thing worth pinning down in the code, not just in review. The accept-list treats UNSPECIFIED as a complete call and rejects INTERMEDIATE_TO_RESULT. On the spec's own terms those are the same value:

enum AggregationPhase {
  // Implies `INTERMEDIATE_TO_RESULT`.
  AGGREGATION_PHASE_UNSPECIFIED = 0;

That text is identical in the pinned 0.63.0 crate and on substrait main today, so the rule as written accepts a value the spec defines as meaning exactly the thing the next arm rejects.

I think the carve-out is the right call anyway, because DataFusion's own producer has been emitting UNSPECIFIED for every aggregate and window call, and plans written by released versions have to keep loading. But that is a deliberate, DataFusion-specific exception rather than a reading of the spec, and the comment currently reads as though the two were the same thing. Something like "the spec says UNSPECIFIED implies INTERMEDIATE_TO_RESULT; we accept it regardless because DataFusion's own producer emitted it, so rejecting it would break plans written by released versions" would stop the next reader from treating it as spec-derived.

It also leaves a residual hole worth naming in the comment: a non-DataFusion producer that emits UNSPECIFIED meaning intermediate state still gets executed as a complete aggregation, which is the failure #24967 describes. This PR closes it for the three explicit phases and leaves it open for the default one.

For disclosure, #25146 is mine, and it stops the producer from emitting UNSPECIFIED. It does not conflict with this; if anything the two fit together, since after it the carve-out here covers only plans from older versions rather than everything DataFusion currently writes. If this lands first I am happy to rebase onto it.

Minor, take or leave: validate_aggregation_phase lives in expr/aggregate_function.rs and is pulled into the window path via use super::aggregate_function::.... Since both callers are peers it might read better next to the shared consumer helpers, but that is taste.

The branch is 65 commits behind main now, so it will want a rebase before it can go in.

Clarified that substrait defines UNSPECIFIED as INTERMEDIATE_TO_RESULT, while this consumer retains the compatibility exception for existing DataFusion plans. The comment now explicitly acknowledges the ambiguity for external producers. The producer change in #25146 is complementary and is not duplicated here.

[goutamadwant]

Hi @goutamadwant thank you for the fix, though I see alexandrefimov mentioned opening a PR. Can you please coordinate with the issue author so we don't duplicate the work.

@alexandrefimov This PR rejects explicit unsupported phases while retaining the documented legacy UNSPECIFIED exception. Are you working on overlapping consumer changes or tests? Happy to coordinate and reuse your cases where helpful. The separate producer fix is in #25146.

[alexandrefimov]

No, I'm not working on anything that overlaps. aggregation_tests.rs in this PR already covers the three plans from #24967 (rel with INITIAL_TO_INTERMEDIATE, rel with a complete phase, and root with three names) and runs every phase through them, so there is nothing of mine left to bring over.

[namanjain24-sudo]

Thanks, f86799b covers both points. The comment now says Substrait defines UNSPECIFIED as INTERMEDIATE_TO_RESULT, and that the exception also lets through other producers' intermediate-state calls, which can't be told apart here. That definition reads the same in the pinned substrait 0.63.0 crate and on substrait main.

I also merged the branch onto current main (c2cf289) locally: no conflicts, cargo fmt and cargo clippy -D warnings are clean, and substrait_integration passes (214 passed, 6 ignored), including the 7 aggregation_tests.