You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "songkant-aws (via GitHub)" <gi...@apache.org> on 2026/06/18 06:40:15 UTC
[I] Substrait consumer: DuplicateUnqualifiedField when chaining two Window relations with a carried window column [datafusion]
songkant-aws opened a new issue, #23007:
URL: https://github.com/apache/datafusion/issues/23007
# Substrait consumer fails with `DuplicateUnqualifiedField` when a plan chains two Window relations and the first window's output column is carried into the second window's input
## Which crate
`datafusion-substrait` (consumer / `from_substrait_plan`). Reproduced on `54.0.0` (crates.io) and the code path is unchanged on `branch-54` / `main`.
## What happens
A logical plan that contains **two `WindowAggr` nodes in series**, where a window-derived column produced by the first window survives (via the intervening projections) into the **input schema of the second window**, round-trips through Substrait and then fails to be consumed:
```
SchemaError(DuplicateUnqualifiedField {
name: "avg(data.b) ORDER BY [UInt64(1) ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW"
}, Some(""))
```
(With `row_number()` windows the duplicated name is
`row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING` instead — same mechanism.)
The producer side is fine: `to_substrait_plan` succeeds. The failure is purely on the **consumer** side, when rebuilding the second `Window` relation.
## Why it happens (root cause)
Substrait intermediate `ProjectRel`/`WindowRel` nodes are **positional** — they carry an `output_mapping` of field indices, not names. Field names live only at `ReadRel.base_schema` and at the top-level `Plan.relations[].root.names`. So any alias a producer attached to an intermediate window column (`AS seq1`, `AS ravg1`, …) is **dropped** in the Substrait IR. When the consumer rebuilds the expressions, a window function gets its default schema name back, e.g. `avg(data.b) ... RANGE ...` or `row_number() ROWS BETWEEN ...`.
In the consumer, `from_project_rel`
(`src/logical_plan/consumer/rel/project_rel.rs`) handles a project that
contains window expressions like this:
```rust
// project_rel.rs (54.0.0), lines ~49-82
let mut explicit_exprs: Vec<Expr> = vec![];
let mut window_exprs: HashSet<Expr> = HashSet::new();
for expr in &p.expressions {
let e = consumer.consume_expression(expr, input.clone().schema()).await?;
if let Expr::WindowFunction(_) = &e {
window_exprs.insert(e.clone());
}
explicit_exprs.push(name_tracker.get_uniquely_named_expr(e)?); // (A)
}
let input = if !window_exprs.is_empty() {
LogicalPlanBuilder::window_plan(input, window_exprs)? // (B) <-- error originates here
} else {
input
};
let mut final_exprs: Vec<Expr> = vec![];
for index in 0..original_schema.fields().len() {
let e = Expr::Column(Column::from(original_schema.qualified_field(index)));
final_exprs.push(name_tracker.get_uniquely_named_expr(e)?); // (C)
}
final_exprs.append(&mut explicit_exprs);
project(input, final_exprs) // (D)
```
The `NameTracker` dedup at **(A)** and **(C)** only governs the names of the
**outer projection** built at **(D)**. The collision, however, happens earlier,
at **(B)**, inside `LogicalPlanBuilder::window_plan` → `.window(...)` →
`Window::try_new` (`datafusion-expr` `plan.rs`), which builds
`window_fields = [ ...all input fields... ] ++ [ ...new window expr fields... ]`
and passes it to `DFSchema::new_with_metadata`, whose `check_names()` raises
`DuplicateUnqualifiedField`:
```
[ ...all input fields... ] ++ [ ...new window expr fields... ]
```
When the first window's output column has been carried down into `input`
(here named `avg(data.b) ... RANGE ...`), and the second window re-introduces an
expression with the **same default schema name** (because its alias was lost over
Substrait), the resulting `DFSchema` has two fields with identical unqualified
names → `DuplicateUnqualifiedField`.
`NameTracker` never inspects the schema *inside* `window_plan`, so it cannot
prevent this. It also does not seed itself with the input schema's existing field
names, so even at the project level an inherited window column and a freshly-built
identical one are not deduplicated against each other.
### Relationship to #15211
PR #15211 added `NameTracker` to dedup **duplicate window functions referenced
multiple times within a single project**. That fix is real but orthogonal: it
covers duplicates *within one `from_project_rel` call's explicit expressions*. It
does **not** cover the case here — an inherited window column (from a *previous*
window relation, already in the input schema) colliding with a newly-built window
expression of the same default name. The duplicate is across the input-schema /
new-window boundary inside `window_plan`, which `NameTracker` doesn't reach.
## Minimal reproductions
All four are `datafusion-substrait` integration tests
(`datafusion/substrait/tests/cases/roundtrip_logical_plan.rs`) using the existing
`create_context()` / `roundtrip_logical_plan_with_ctx()` helpers.
### Repro 1 — two identical `row_number()` windows (fails)
```rust
#[tokio::test]
async fn chained_identical_window_functions() -> Result<()> {
use datafusion::functions_window::expr_fn::row_number;
let ctx = create_context().await?;
let scan = ctx.table("data").await?.into_optimized_plan()?;
let plan = LogicalPlanBuilder::from(scan)
.window(vec![row_number().alias("rn1")])?
.window(vec![row_number().alias("rn2")])?
.build()?;
roundtrip_logical_plan_with_ctx(plan, ctx).await?; // FAILS: DuplicateUnqualifiedField
Ok(())
}
```
### Repro 2 — seq window + sort + project-except, chained (PASSES — important contrast)
```rust
#[tokio::test]
async fn chained_seq_with_sort_and_project_except() -> Result<()> {
use datafusion::functions_window::expr_fn::row_number;
let ctx = create_context().await?;
let scan = ctx.table("data").await?.into_optimized_plan()?;
let plan = LogicalPlanBuilder::from(scan)
.window(vec![row_number().alias("seq1")])?
.project(vec![col("a"), col("b"), col("seq1")])?
.sort(vec![col("seq1").sort(true, false)])?
.project(vec![col("a"), col("b")])? // DROP seq1
.window(vec![row_number().alias("seq2")])?
.project(vec![col("a"), col("b"), col("seq2")])?
.sort(vec![col("seq2").sort(true, false)])?
.project(vec![col("a"), col("b")])? // DROP seq2
.build()?;
roundtrip_logical_plan_with_ctx(plan, ctx).await?; // PASSES
Ok(())
}
```
This passes because dropping `seq1` lets the plan flatten so the two
`row_number()` columns never coexist in a single Window input schema. This case
isolates the trigger: it is **not** `sort`/`project-except` that breaks things.
### Repro 3 — faithful shape: window emits seq + running aggregate, the aggregate is carried through (fails)
This mirrors a real chained `streamstats` plan: each window emits a sequence
column **and** a running aggregate; the running aggregate is carried through the
intervening projects (only the seq is dropped). Carrying that window column is
what keeps window #1's output alive in window #2's input.
```rust
#[tokio::test]
async fn chained_window_with_carried_running_aggregate() -> Result<()> {
use datafusion::functions_aggregate::average::avg_udaf;
use datafusion::functions_window::expr_fn::row_number;
use datafusion::logical_expr::expr::WindowFunction;
use datafusion::logical_expr::{Expr, ExprFunctionExt, WindowFrame};
let ctx = create_context().await?;
let scan = ctx.table("data").await?.into_optimized_plan()?;
let running1 = Expr::from(WindowFunction::new(avg_udaf(), vec![col("b")]))
.window_frame(WindowFrame::new(Some(false))).build()?.alias("ravg1");
let running2 = Expr::from(WindowFunction::new(avg_udaf(), vec![col("b")]))
.window_frame(WindowFrame::new(Some(false))).build()?.alias("ravg2");
let plan = LogicalPlanBuilder::from(scan)
.window(vec![row_number().alias("seq1"), running1])?
.project(vec![col("a"), col("b"), col("seq1"), col("ravg1")])?
.sort(vec![col("seq1").sort(true, false)])?
.project(vec![col("a"), col("b"), col("ravg1")])? // drop seq1, KEEP ravg1
.window(vec![row_number().alias("seq2"), running2])?
.project(vec![col("a"), col("b"), col("ravg1"), col("seq2"), col("ravg2")])?
.sort(vec![col("seq2").sort(true, false)])?
.project(vec![col("a"), col("b"), col("ravg1"), col("ravg2")])? // drop seq2
.build()?;
roundtrip_logical_plan_with_ctx(plan, ctx).await?;
// FAILS: DuplicateUnqualifiedField {
// name: "avg(data.b) ORDER BY [UInt64(1) ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW" }
Ok(())
}
```
## Suggested fix (direction, not a final patch)
Extend the existing `NameTracker` mechanism so it also dedups against names that
already exist in the window relation's input schema. Concretely, in
`from_project_rel`, before building `window_exprs`, seed the tracker (or apply
`get_uniquely_named_expr`) with the input schema's field names, and pass uniquely
re-aliased window expressions into `window_plan`, so an inherited window column
forces the newly-built one to receive a `__temp__N` suffix — the same disambiguation
#15211 already uses, just extended across the input-schema boundary. This keeps the
positional `output_mapping` semantics intact (the suffix only affects internal
schema names, not the emitted ordering).
I'm happy to open a PR along these lines and add the four tests above (the passing
one included, as a guard against "fixing" it by breaking the flatten path).
## Environment
- `datafusion` / `datafusion-substrait` `54.0.0` (also present on `branch-54`, HEAD `08da279`, and `main`).
- Discovered via OpenSearch's PPL `streamstats` command, which lowers chained
`streamstats` into exactly this `window → project(carry) → sort → project(drop seq)`
shape and ships the plan to a DataFusion backend over Substrait.
--
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: github-unsubscribe@datafusion.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Substrait consumer: DuplicateUnqualifiedField when chaining two Window relations with a carried window column [datafusion]
Posted by "gabotechs (via GitHub)" <gi...@apache.org>.
gabotechs closed issue #23007: Substrait consumer: DuplicateUnqualifiedField when chaining two Window relations with a carried window column
URL: https://github.com/apache/datafusion/issues/23007
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Substrait consumer: DuplicateUnqualifiedField when chaining two Window relations with a carried window column [datafusion]
Posted by "songkant-aws (via GitHub)" <gi...@apache.org>.
songkant-aws commented on issue #23007:
URL: https://github.com/apache/datafusion/issues/23007#issuecomment-4739160847
## Additional repro: equivalent SQL that executes fine in DataFusion but fails after a Substrait round-trip
Reproduced on **`main` @ `0fc55d067` (2026-06-18)** as well as `54.0.0` — identical error.
This is the SQL form of the streamstats plan (the original report used `LogicalPlanBuilder`). It's a single hand-written query against the existing `data` table:
```sql
SELECT a, b, avg1, avg2 FROM (
SELECT a, b, avg1,
row_number() OVER () AS seq2,
avg(b) OVER (PARTITION BY a ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS avg2
FROM (
SELECT a, b, avg1 FROM (
SELECT a, b,
row_number() OVER () AS seq1,
avg(b) OVER (PARTITION BY a ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS avg1
FROM data
) t1 ORDER BY seq1
) t2
) t3 ORDER BY seq2
```
### The strange part
This SQL is **valid** and **executes successfully** in DataFusion with no Substrait involved:
```rust
// PASSES: planning + optimization + execution all succeed
let plan = ctx.sql(SQL).await?.into_optimized_plan()?;
let _ = ctx.sql(SQL).await?.collect().await?; // 1 batch, ok
// FAILS: the exact same SQL, once round-tripped through Substrait
roundtrip(SQL).await?;
// Error: SchemaError(DuplicateUnqualifiedField {
// name: "avg(data.b) PARTITION BY [data.a] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW" }, Some(""))
```
So a legal, executable plan becomes unconsumable purely because it went through Substrait.
### Why this is counterintuitive — and what actually collides
At first glance you'd expect the duplicate to involve the two `row_number()` columns. It does **not**. Here is the DataFusion-optimized plan (no Substrait), which is what gets serialized:
```
Projection: t3.a, t3.b, t3.avg1, t3.avg2
Sort: t3.seq2 ASC NULLS LAST
Projection: t3.a, t3.b, t3.avg1, t3.avg2, t3.seq2
SubqueryAlias: t3
Projection: t2.a, t2.b, t2.avg1,
row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS seq2,
avg(t2.b) PARTITION BY [t2.a] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS avg2
WindowAggr: windowExpr=[[row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]]
WindowAggr: windowExpr=[[avg(t2.b) PARTITION BY [t2.a] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]
SubqueryAlias: t2
SubqueryAlias: t1
Projection: data.a, data.b,
avg(data.b) PARTITION BY [data.a] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS avg1
WindowAggr: windowExpr=[[avg(data.b) PARTITION BY [data.a] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]
TableScan: data projection=[a, b]
```
Two observations from this plan:
1. **The inner `row_number()` (`seq1`) is gone.** The optimizer pruned the entire inner `seq1` WindowAggr — it was only referenced by an inner `ORDER BY` that doesn't affect the final result, and the outer queries never select it. So `row_number` is not what collides. (`seq2` survives only because it's used by the top-level `ORDER BY`.)
2. **The collision is between the two `avg(...)` columns**, which carry *different aliases* `avg1` and `avg2`:
- inner WindowAggr → `AS avg1`, and `avg1` is then carried up through `t1 → t2 → t3` into the final SELECT;
- middle WindowAggr → `AS avg2`.
In the DataFusion plan they are unambiguous because of those distinct aliases — that's why it executes fine. But their **internal schema names are byte-for-byte identical**: both are `avg(data.b) PARTITION BY [data.a] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW`. They are distinguished *only* by the `AS avg1` / `AS avg2` aliases.
Substrait intermediate rels are positional (`output_mapping` by index); the `avg1`/`avg2` aliases are **not** preserved in the IR. On the consumer side, when `from_project_rel` rebuilds the second window via `window_plan` → `Window::try_new`, the inherited `avg1` column (already in the input schema) and the freshly-built `avg2` both fall back to that same default schema name, and `Window::try_new`'s `DFSchema::new_with_metadata` → `check_names()` raises `DuplicateUnqualifiedField`.
So the precise "strange point": **the two columns are never duplicates semantically** — they have different aliases and different origins, and the column that the outer query selects is `avg1`, not the pruned `row_number`. The duplicate is *manufactured* by Substrait's positional representation dropping the aliases that disambiguated them. The existing `NameTracker` (added in #15211) doesn't catch it because it never seeds itself with the input schema's field names before `window_plan`, so an inherited window column and a new identically-named one are not deduplicated against each other.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Substrait consumer: DuplicateUnqualifiedField when chaining two Window relations with a carried window column [datafusion]
Posted by "ShayanGho (via GitHub)" <gi...@apache.org>.
ShayanGho commented on issue #23007:
URL: https://github.com/apache/datafusion/issues/23007#issuecomment-5549241207
take
This still reproduces on current main: running the SQL from the comment above through `roundtrip()` in `roundtrip_logical_plan.rs` fails with `SchemaError(DuplicateUnqualifiedField { name: "avg(data.b) PARTITION BY [data.a] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW" })`.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Substrait consumer: DuplicateUnqualifiedField when chaining two Window relations with a carried window column [datafusion]
Posted by "ShayanGho (via GitHub)" <gi...@apache.org>.
ShayanGho commented on issue #23007:
URL: https://github.com/apache/datafusion/issues/23007#issuecomment-5595035820
take
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org