namanjain24-sudo opened a new issue, #25100: URL: https://github.com/apache/datafusion/issues/25100
### 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`](https://github.com/substrait-io/substrait/blob/v0.87.0/proto/substrait/algebra.proto) and [`Expression.WindowFunction.phase`](https://github.com/substrait-io/substrait/blob/v0.87.0/proto/substrait/algebra.proto) 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. ```rust 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: - #24967 and #25045 are about the consumer ignoring `phase`. #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. - #25049 concerns `output_type` on the same `AggregateFunction` message; this report is about the neighbouring `phase` field. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
