Skip to content

fix: Set Substrait aggregation phase to INITIAL_TO_RESULT - #25146

Open
namanjain24-sudo wants to merge 2 commits into
apache:mainfrom
namanjain24-sudo:fix-substrait-aggregation-phase
Open

namanjain24-sudo wants to merge 2 commits into
apache:mainfrom
namanjain24-sudo:fix-substrait-aggregation-phase

Conversation

@namanjain24-sudo

@namanjain24-sudo namanjain24-sudo commented Sep 10, 2026

Copy link
Copy Markdown

Which issue does this PR close?

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.

@github-actions github-actions Bot added the substrait Changes to the substrait crate label Sep 10, 2026

@Xuanwo Xuanwo left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice fix!

@namanjain24-sudo
namanjain24-sudo force-pushed the fix-substrait-aggregation-phase branch 2 times, most recently from 26b02c3 to d50b923 Compare September 11, 2026 12:38
@namanjain24-sudo

Copy link
Copy Markdown
Author

Nice fix!

Thank you, @Xuanwo

The producer left `phase` at its default on every aggregate and window
function call it emits, so each one carried AGGREGATION_PHASE_UNSPECIFIED.
That is not the same as omitting the field: the spec gives the value the
meaning INTERMEDIATE_TO_RESULT, i.e. that the arguments are already
intermediate aggregation state to be combined.

A LogicalPlan::Aggregate is always a complete aggregation over its input
rows. The partial/final split is a physical planning concern that the
logical producer has no notion of. Both `AggregateFunction.phase` and
`Expression.WindowFunction.phase` are documented as required, and as
needing INITIAL_TO_RESULT for a complete invocation, so set that at both
call sites.

This stays invisible to a DataFusion-to-DataFusion round trip because the
consumer never reads `phase`, but a consumer that honours the declaration
reads a complete aggregation as one whose arguments are partial state.
A DataFusion round trip cannot see a field the consumer never reads, so
nothing here catches a wrong AggregationPhase: the consumer rebuilds a
complete aggregation whatever the plan says. substrait-spark does read
it, and maps the default UNSPECIFIED to Spark's Final, which Spark then
rejects because the inputs are raw rows rather than partial buffers.

An ignored Rust test writes the plan for count/sum/avg, and a small
Maven project converts it with substrait-java and runs it in Spark over
t(i) = 1, 2, 3, checking both the aggregation mode and the rows. Two
workarounds are applied on the Java side, each named after the issue it
stands in for: apache#11545 for the extension URN, and apache#25049 for the unset
output_type, which can go once that fix lands.

The job runs only when the Substrait crate changes, because it pulls a
Spark sized dependency set.
@github-actions github-actions Bot added the development-process Related to development process of DataFusion label Sep 16, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

development-process Related to development process of DataFusion substrait Changes to the substrait crate

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Substrait producer emits AGGREGATION_PHASE_UNSPECIFIED for every aggregate and window function

2 participants