Substrait producer emits AGGREGATION_PHASE_UNSPECIFIED for every aggregate and window function
- Dominant language
- Rust
- Stars
- 9.3k
- Forks
- 2.4k
- Avg merge
- 3d 7h
- Merged PRs (30d)
- 344
Description
### 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) {
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.
Contributor guide
Assessment
This issue has not been assessed yet.