Substrait consumer: DuplicateUnqualifiedField when chaining two Window relations with a carried window column

[songkant-aws]

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:

// 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)

#[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)

#[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.

#[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.

[songkant-aws]

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:

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:

// 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.