fix: Set Substrait aggregation phase to INITIAL_TO_RESULT

[namanjain24-sudo]

Which issue does this PR close?

• Closes Substrait producer emits AGGREGATION_PHASE_UNSPECIFIED for every aggregate and window function #25100.

Rationale for this change

The Substrait producer never sets phase on the aggregate and window function
calls it emits, so every call carries AGGREGATION_PHASE_UNSPECIFIED.

That is not the same as leaving the field out. The spec gives the value a
meaning, and it is not the one these plans need:

enum AggregationPhase {
  // Implies `INTERMEDIATE_TO_RESULT`.
  AGGREGATION_PHASE_UNSPECIFIED = 0;
  ...
  // A complete invocation: the function should aggregate the given set of
  // inputs to yield a single return value. This style must be used for
  // aggregate or window functions that are not decomposable.
  AGGREGATION_PHASE_INITIAL_TO_RESULT = 3;

Both AggregateFunction.phase and Expression.WindowFunction.phase are
documented as Required. Must be set to INITIAL_TO_RESULT for ... that are not decomposable.

A LogicalPlan::Aggregate is always a complete aggregation over its input
rows. The partial/final split is a physical planning concern, and the logical
producer has no notion of it, so INITIAL_TO_RESULT is the phase these plans
should declare. What they declare instead carries the spec meaning
INTERMEDIATE_TO_RESULT: that the arguments are already intermediate state to
be combined.

This stays invisible to a DataFusion-to-DataFusion round trip because the
consumer never reads the field (#24967). A consumer that does honour the
declaration reads a complete aggregation as one whose arguments are already
partial state.

What changes are included in this PR?

Set phase to AGGREGATION_PHASE_INITIAL_TO_RESULT at the two producer call
sites that emit it:

• from_aggregate_function in producer/expr/aggregate_function.rs
• make_substrait_window_function in producer/expr/window_function.rs

No other producer site emits the field, and the consumer does not read it, so
nothing else changes.

What is the testing strategy for this PR?

New test aggregate_and_window_functions_declare_initial_to_result in
datafusion/substrait/tests/cases/serialize.rs. It produces plans for a bare
aggregate, a grouped aggregate, and a window function, then walks the produced
protobuf directly and asserts the phase on every aggregate and window function
call, rather than round-tripping through the consumer (which ignores the
field, so a round trip could not catch this).

Verified the test fails without the producer change:

assertion `left == right` failed: phase for `SELECT sum(a) FROM data`
  left: 0
 right: 3

The java interop test

main and this branch both pass every test in the repository, because the
DataFusion consumer never reads phase and rebuilds a complete aggregation
either way. #25100 asked for that gap to be covered against another
implementation, so this PR also adds one:

• tests/cases/java_interop.rs writes the plan for
  SELECT count(i), sum(i), avg(i) FROM t. It is #[ignore]d, so cargo test
  is unchanged for anyone without a JVM.
• java-interop/ converts that plan with substrait-java 0.103.0 and runs it in
  Spark 3.5.4 over t(i) = 1, 2, 3, checking the aggregation mode and the rows.

Run both halves with:

cargo test -p datafusion-substrait --test substrait_integration -- --ignored write_java_interop_plan
mvn -f datafusion/substrait/java-interop/pom.xml test

Against a plan from this branch it passes, with Complete and
count 3, sum 6, avg 2.0. Against a plan from main it fails with
expected: <[Complete, Complete, Complete]> but was: <[Final, Final, Final]>,
which is the failure this change prevents.

Two workarounds are applied on the Java side, each named after the issue it
stands in for: #11545 for the extension URN, and #25049 for the unset
output_type, which can be removed once #25090 lands.

The CI job runs only when datafusion/substrait/** or rust.yml changes,
because it pulls a Spark sized dependency set.

Are there any user-facing changes?

Plans produced by to_substrait_plan now declare
AGGREGATION_PHASE_INITIAL_TO_RESULT instead of
AGGREGATION_PHASE_UNSPECIFIED on aggregate and window function calls. This
is a fix to the emitted protobuf rather than a Rust API change. Consumers that
read phase will now see the correct declaration; consumers that ignore it,
including DataFusion's own, are unaffected.

[Xuanwo]

Nice fix!

[namanjain24-sudo]

Nice fix!

Thank you, @Xuanwo

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 82.47%. Comparing base (ac37adf) to head (caf289f).

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #25146      +/-   ##
==========================================
- Coverage   82.48%   82.47%   -0.01%     
==========================================
  Files        1140     1140              
  Lines      437341   437342       +1     
  Branches   437341   437342       +1     
==========================================
- Hits       360721   360710      -11     
- Misses      54835    54843       +8     
- Partials    21785    21789       +4     

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

[namanjain24-sudo]

@Xuanwo just rebased onto main to clear the conflict, no code changes — but that reset the approval. Could you take another quick look when you get a chance?

[namanjain24-sudo]

@Xuanwo gentle bump on this whenever you get a chance — still just the rebase, no code changes since your approval.

[namanjain24-sudo]

@Xuanwo rebased onto main again to clear the conflict (main independently added Substrait output_type for window functions, which this PR's window-phase change also touches) — no further code changes, just the rebase. Sorry for the repeat ping.