Substrait producer emits AGGREGATION_PHASE_UNSPECIFIED for every aggregate and window function

Describe the bug

The Substrait producer never sets phase on the aggregate and window function calls it emits. Every call carries AGGREGATION_PHASE_UNSPECIFIED, which is not the same as leaving a field out: the spec gives that enum value a meaning, and it is not the one these plans need.

Reproduced on main at 35c56b020.

AggregateFunction.phase and Expression.WindowFunction.phase are both documented as:

Describes which part of the aggregation to perform within the context of distributed algorithms. Required. Must be set to INITIAL_TO_RESULT for aggregate functions that are not decomposable.

and the enum documents the default as:

// Implies INTERMEDIATE_TO_RESULT.
AGGREGATION_PHASE_UNSPECIFIED = 0;

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 the phase these plans should declare is INITIAL_TO_RESULT. What they declare instead carries the spec meaning INTERMEDIATE_TO_RESULT: that the arguments are already intermediate state to be combined.

Both call sites hardcode the value:

from_aggregate_function — phase: AggregationPhase::Unspecified as i32 (datafusion/substrait/src/logical_plan/producer/expr/aggregate_function.rs:68)
the window function producer — phase: 0, // default to AGGREGATION_PHASE_UNSPECIFIED (datafusion/substrait/src/logical_plan/producer/expr/window_function.rs:111)

To reproduce

Add this as an example under datafusion/substrait/examples/ and run
cargo run --locked -p datafusion-substrait --example phase_probe. It inspects the produced protobuf directly, without converting it back through a consumer.

use datafusion::arrow::datatypes::{DataType, Field, Schema};
use datafusion::common::Result;
use datafusion::datasource::empty::EmptyTable;
use datafusion::prelude::SessionContext;
use datafusion_substrait::logical_plan::producer::to_substrait_plan;
use datafusion_substrait::substrait::proto::expression::RexType;
use datafusion_substrait::substrait::proto::rel::RelType;
use datafusion_substrait::substrait::proto::{plan_rel, Rel};
use std::sync::Arc;

fn name(p: i32) -> &'static str {
    match p {
        0 => "UNSPECIFIED",
        1 => "INITIAL_TO_INTERMEDIATE",
        2 => "INTERMEDIATE_TO_INTERMEDIATE",
        3 => "INITIAL_TO_RESULT",
        4 => "INTERMEDIATE_TO_RESULT",
        _ => "?",
    }
}

fn walk(rel: &Rel, out: &mut Vec<String>) {
    match rel.rel_type.as_ref() {
        Some(RelType::Aggregate(a)) => {
            for m in &a.measures {
                if let Some(f) = &m.measure {
                    out.push(format!("AggregateFunction.phase = {}", name(f.phase)));
                }
            }
            a.input.as_ref().map(|i| walk(i, out));
        }
        Some(RelType::Project(p)) => {
            for e in &p.expressions {
                if let Some(RexType::WindowFunction(w)) = &e.rex_type {
                    out.push(format!("WindowFunction.phase    = {}", name(w.phase)));
                }
            }
            p.input.as_ref().map(|i| walk(i, out));
        }
        _ => {}
    };
}

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<()> {
    let ctx = SessionContext::new();
    ctx.register_table(
        "t",
        Arc::new(EmptyTable::new(Arc::new(Schema::new(vec![Field::new(
            "i",
            DataType::Int64,
            false,
        )])))),
    )?;

    for sql in [
        "SELECT count(i) FROM t",
        "SELECT sum(i) FROM t",
        "SELECT avg(i) FROM t",
        "SELECT count(i) OVER () FROM t",
        "SELECT sum(i) OVER (ORDER BY i) FROM t",
    ] {
        let df = ctx.sql(sql).await?;
        let proto = to_substrait_plan(df.logical_plan(), &ctx.state())?;
        let mut out = vec![];
        for r in &proto.relations {
            if let Some(plan_rel::RelType::Root(root)) = &r.rel_type {
                root.input.as_ref().map(|i| walk(i, &mut out));
            }
        }
        for line in out {
            println!("{sql:<40} {line}");
        }
    }
    Ok(())
}

Output:

SELECT count(i) FROM t                   AggregateFunction.phase = UNSPECIFIED
SELECT sum(i) FROM t                     AggregateFunction.phase = UNSPECIFIED
SELECT avg(i) FROM t                     AggregateFunction.phase = UNSPECIFIED
SELECT count(i) OVER () FROM t           WindowFunction.phase    = UNSPECIFIED
SELECT sum(i) OVER (ORDER BY i) FROM t   WindowFunction.phase    = UNSPECIFIED

Expected behavior

AggregateFunction.phase and Expression.WindowFunction.phase should be set to AGGREGATION_PHASE_INITIAL_TO_RESULT, since the producer only ever emits complete aggregations.

This stays invisible to a DataFusion-to-DataFusion round trip because the consumer never reads the field — the string phase does not appear anywhere under datafusion/substrait/src/logical_plan/consumer/, which is #24967. A consumer that does honour the declaration reads a complete aggregation as one whose arguments are already intermediate state.

Additional context

Related, but distinct:

Substrait consumer ignores AggregateFunction.phase, so an intermediate aggregate runs as a complete one #24967 and fix: reject unsupported Substrait aggregation phases #25045 are about the consumer ignoring phase. fix: reject unsupported Substrait aggregation phases #25045 accepts INITIAL_TO_RESULT and deliberately keeps accepting UNSPECIFIED "for compatibility with existing DataFusion-produced plans" — setting the phase on the producer side is what would let that allowance go away later.
Substrait producer omits the required output_type on aggregate functions #25049 concerns output_type on the same AggregateFunction message; this report is about the neighbouring phase field.

take

Same question: #25190 (comment)

I don't understand what you are doing, or why this is a bug. Can you please fix clarify on this PR what the actual problem is you are solving?

Yes, this is also a compatibility problem.

DataFusion exports a complete logical aggregate over raw input rows, but sets phase to UNSPECIFIED. Current substrait-java maps UNSPECIFIED to Spark Final, because in Substrait it means INTERMEDIATE_TO_RESULT. INITIAL_TO_RESULT maps to Spark Complete. As a result, substrait-java treats the aggregate arguments as intermediate state even though the plan contains expressions over the original input rows.

DataFusion-to-DataFusion round trips do not catch this because the DataFusion consumer ignores phase. #25049 is a separate issue: substrait-java currently fails earlier because output_type is missing. Once that is fixed, the phase is still incorrect.

@alexandrefimov

@alexandrefimov's reading is what the code does — I checked it rather than take it on trust, and the
exact lines are worth pinning because this one never throws.

substrait-java v0.103.0, spark/src/main/scala/io/substrait/spark/expression/ToAggregateFunction.scala:76:

def toSpark(phase: SExpression.AggregationPhase): AggregateMode = phase match {
  case SExpression.AggregationPhase.UNSPECIFIED =>
    Final // UNSPECIFIED implies INTERMEDIATE_TO_RESULT
  case SExpression.AggregationPhase.INITIAL_TO_INTERMEDIATE => Partial
  case SExpression.AggregationPhase.INTERMEDIATE_TO_INTERMEDIATE => PartialMerge
  case SExpression.AggregationPhase.INTERMEDIATE_TO_RESULT => Final
  case SExpression.AggregationPhase.INITIAL_TO_RESULT => Complete
}

So what we export as AGGREGATION_PHASE_UNSPECIFIED arrives as Spark Final, meaning "the inputs
are partial aggregation buffers", when what we actually exported is a complete aggregation over raw
input rows, which is Spark Complete. Nothing is rejected. It is read as a different plan.

algebra.proto agrees at both ends: AGGREGATION_PHASE_UNSPECIFIED = 0 is documented "Implies
INTERMEDIATE_TO_RESULT", and INITIAL_TO_RESULT is "A complete invocation: the function should
aggregate the given set of inputs to yield a single return value". The field's own comment says
Required.

Why no test here catches it: our consumer never reads phase. There are zero occurrences of it
anywhere under logical_plan/consumer/, so producer and consumer agree on the omission and every
DataFusion-to-DataFusion round trip passes.

On the fix in #25146 — setting both sites to INITIAL_TO_RESULT unconditionally is correct for us
and does not need a case analysis, because the Substrait producer converts logical plans and
LogicalPlan::Aggregate has no mode at all (input, group_expr, aggr_expr, schema).
AggregateMode, with its Partial and Final variants, lives in physical-plan. A logical
aggregate is therefore always a complete aggregation, which is exactly INITIAL_TO_RESULT. I
confirmed the diff does set both sites, the aggregate and the window function.

Correcting myself on one point: I wrote on #25090 that this issue was a spec-conformance argument
rather than a reproduction. That was wrong, and the mapping above is why. A silent misread is the
worse of the two failures, not the milder one.

I feel like there is a larger point here -- basically if we want to have better interop between java and datafusion, we should probably be testing that interop directly rather than reasoning about the code / adding incremental unit tests. That way we can ensure the systems interoperate and that we aren't introducing regressions

Maybe we could look into some sort of automated testing -- or see if the substrait project already has one -- that actually creates plans and then verifies if DataFusion can read them successfully, and verifies the plan is as expected in DataFusion (e.g. check the explain plan)

Part of this exists already, for plans DataFusion reads. consumer_integration.rs compares the logical plans DataFusion builds from 22 TPC-H plans taken from substrait-io/consumer-testing with snapshots, and logical_plans.rs does the same for five plans generated by substrait-java's Isthmus. Both read JSON files committed here, the TPC-H ones last refreshed in 2024, so a change in substrait-java never reaches them.

#25049 (output_type), #25190 (CASE) and this issue (phase) are about plans DataFusion writes and substrait-java reads, and neither suite covers that. DataFusion's Substrait tests do not run substrait-java, and consumer-testing runs DataFusion as a producer, but its consumers are Acero, DataFusion and DuckDB.

Until #11545 is fixed, substrait-java rejects any DataFusion plan that calls a function before it reads output_type or phase, so a test in that direction has to work around #11545 first, as the table in #25090 does.

@alexandrefimov

I ran this directly instead of reasoning about it further. Setup: main at c2cf289, substrait-java / Isthmus 0.103.0 (the latest release), the 22 TPC-H queries, SF 0.01 data. @alexandrefimov's comment above already covers what the existing suites test, so this is only what a fresh run shows.

Java → DataFusion. I used plans generated by Isthmus from its own TPC-H queries.

As generated, 0 of 22 load. The plans name tables in upper case and DataFusion registers them in lower case, so every one fails with No table named '…' ('LINEITEM', 'PART', and so on; substrait generated by Apache Calcite does not run in DataFusion #14831).
With identifier normalization turned off so the names match, 7 of 22 run. All 7 return the same rows as DuckDB running the same queries (compared to 4 decimal places).
Of the other 15:
- 11 fail on VarChar literals. The consumer accepts FixedChar but has no VarChar case.
- 3 are correlated subqueries. Isthmus now writes OuterReference.rel_reference, a field the substrait 0.63.0 crate we pin doesn't have, so the consumer sees steps_out = 0.
- 1 has an enum function argument ([substrait] Add support for enum arguments #12493).

The checked-in TPC-H plans from 2024 contain no VarChar literals and no rel_reference, so those tests can't see either gap.

DataFusion → Java. DataFusion produces all 22 plans, and substrait-java 0.103.0 fails on all 22 at the extension URN (#11545).

A correction to my earlier comment on this issue. I wrote "Nothing is rejected." That is true of substrait-java's plan converter, but not end to end. I took an Isthmus plan for SELECT COUNT(I), SUM(I), AVG(I) over the values 1, 2, 3 and changed only AggregateFunction.phase. With INITIAL_TO_RESULT, substrait-spark 0.103.0 on Spark 3.5.4 returns [3, 6, 2.0]. With UNSPECIFIED, it maps the aggregates to Final, and Spark rejects the plan during analysis with MISSING_ATTRIBUTES. So the phase bug is real, but in Spark it shows up as a failure, not as wrong results.

The harness is small: one Java class for Isthmus and substrait-java, a Rust example for DataFusion, and DuckDB for the reference rows. If it's useful, I can turn it into a job that runs on a schedule rather than on every PR, either here or in consumer-testing, whichever you prefer. The VarChar literal and rel_reference gaps don't seem to have issues yet, and I can file them.

I guess what I am saying is that if the goal is to ensure that datafusion produced plans can be consumed by substrait-java, we should be testing that explicitly somehow (and not one offs in an issue comment). Otherwise I predict it will be hard to find people who can review / look at these PRs and both 1) Have confidence they match the java behavior, 2) will continue to work over time

For example, can we make a test that makes DataFusion plans, and then tries to consume it using substrait-java? We could perhaps only run this test on changes to datafusion-substrait.

@namanjain24-sudo

I agree. Could we start with one aggregate test that generates a plan in DataFusion, runs it through substrait-java and Spark, and checks the result? Conversion alone would miss the phase failure described above.

@namanjain24-sudo could you add this test to #25146 using your harness, with an explicit workaround for #11545?

@alexandrefimov

@alexandrefimov @alamb yes, I'll add it to #25146. Before writing it, I checked that such a test really tells the fix apart from main, using a plan DataFusion itself generates. The query is SELECT count(i), sum(i), avg(i) FROM t, run through substrait-java 0.103.0 and substrait-spark on Spark 3.5.4 over t(i) = 1, 2, 3:

producer	workarounds applied	result
#25146	none	substrait-java: `Function 'avg' references URN anchor -1` (#11545)
#25146	#11545	substrait-java: `Type is not set` (#25049)
#25146	#11545 + `output_type`	Spark `Complete`, `[3, 6, 2.0]`
`main` (`f8cc678`, the base of #25146)	#11545 + `output_type`	Spark `Final`, analysis fails with `MISSING_ATTRIBUTES`
#25146 with #25090 applied	#11545	Spark `Complete`, `[3, 6, 2.0]`

So on #25146 alone the test needs a second workaround: substrait-java stops at the missing output_type before it reads phase. I'll include both and mark the output_type one for removal once #25090 lands. If #25090 is merged first, only the #11545 workaround is needed.

What I plan to push to #25146:

a Rust test in datafusion/substrait that writes the plan,
a small Java program (substrait-java and substrait-spark 0.103.0, Spark 3.5.4 in local mode) that applies the workarounds, runs the plan and checks the rows,
a CI job with JDK 17 that runs both only when datafusion/substrait/** changes. The Spark classpath is about 260 MB, so the job would cache it.

This would be the first Java code and JVM job in the repo, so please say if you'd rather it lived in a separate workflow or somewhere else.

@alexandrefimov

@alexandrefimov @alamb the test is in #25146 now, as f0f8429.

It is in two halves, because producing the plan needs no JVM:

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

The two workarounds are applied on the Java side only, each named after the issue it stands in for, so the test fails for the reason it is about rather than an unrelated one:

Provide valid extensionUris and extensionUriReferences in Substrait #11545, the extension URN, which otherwise stops the plan first;
Substrait producer omits the required output_type on aggregate functions #25049, the unset output_type. That one can go once fix: Set Substrait output_type on aggregate functions #25090 lands, and the test then needs only the URN workaround.

What it catches, run both ways:

plan produced by	result
this branch	passes: `Complete`, and `count 3, sum 6, avg 2.0`
`main`	fails: `expected: <[Complete, Complete, Complete]> but was: <[Final, Final, Final]>`

The second row is the point. A round trip cannot see this, because the consumer never reads phase and rebuilds a complete aggregation either way.

On cost, since it pulls a Spark sized dependency set: the job runs only when datafusion/substrait/** or rust.yml changes, using dorny/paths-filter, and caches ~/.m2 through actions/setup-java. Locally the Maven half takes about 9 seconds with a warm cache, and the plan is written by a test that already exists in the crate.

This is the first JVM job in the repo, so if you would rather it lived in its own workflow, or ran on a schedule instead of on Substrait changes, say so and I will move it.

Navigation Menu

Repository navigation

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

Describe the bug

To reproduce

Expected behavior

Additional context

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

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

Description

Describe the bug

To reproduce

Expected behavior

Additional context

Activity

namanjain24-sudo commented on Sep 9, 2026

alamb commented on Sep 11, 2026

alexandrefimov commented on Sep 11, 2026

namanjain24-sudo commented on Sep 11, 2026

alamb commented on Sep 13, 2026

alexandrefimov commented on Sep 14, 2026

namanjain24-sudo commented on Sep 14, 2026

alamb commented on Sep 15, 2026

alexandrefimov commented on Sep 15, 2026

namanjain24-sudo commented on Sep 15, 2026

namanjain24-sudo commented on Sep 16, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions