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 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 intermediate phases and unknown protobuf enum values with explicit errors.

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.
• Added tests for supported phases, every 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 required extended workspace suite passes: 11,267 tests passed, eight existing tests ignored, and all 511 SQL logic-test files completed.
• cargo clippy --all-targets --all-features -- -D warnings passes.
• The complete ./dev/rust_lint.sh passes, including strict workspace documentation checks.

Are there any user-facing changes?

Plans with unsupported aggregate or window phases now fail explicitly instead of being interpreted as complete calls. Existing unspecified-phase plans remain accepted. 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.72%. Comparing base (1b6dc92) to head (6e8b7b3).

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #25045      +/-   ##
==========================================
- Coverage   81.72%   81.72%   -0.01%     
==========================================
  Files        1127     1127              
  Lines      416310   416318       +8     
  Branches   416310   416318       +8     
==========================================
+ Hits       340224   340226       +2     
- Misses      56095    56103       +8     
+ Partials    19991    19989       -2     

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