Substrait producer emits AGGREGATION_PHASE_UNSPECIFIED for every aggregate and window function

[namanjain24-sudo]

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.

[namanjain24-sudo]

take