timsaucer opened a new issue, #24884:
URL: https://github.com/apache/datafusion/issues/24884
### Describe the bug
When the physical optimizer reverses a window expression to avoid an extra
sort, and that window expression is backed by an **aggregate UDAF** (rather
than a window UDF), the reversed expression's *output field name* is rewritten.
The window exec's schema changes, but parent plan nodes still reference the old
column name, so planning fails with an internal error:
```
EnsureRequirements
caused by
Internal error: Assertion failed: col.name() == matching_name: Input field
name first_value(?table?.v) ORDER BY [?table?.t ASC NULLS FIRST] ROWS BETWEEN
UNBOUNDED PRECEDING AND CURRENT ROW does not match with the projection
expression last_value(?table?.v) ORDER BY [?table?.t DESC NULLS LAST] ROWS
BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW.
```
Mechanism:
1. `EnsureRequirements` sorts the input `t ASC` for the first window. For
the second (`ORDER BY t DESC`) window, `get_window_mode` reports
`should_reverse = true`, so `get_best_fitting_window` swaps in
`e.get_reverse_expr()` to avoid a second sort
(`datafusion/physical-plan/src/windows/mod.rs`, `get_best_fitting_window`).
2. For an aggregate-backed window expr this reaches
`AggregateFunctionExpr::reverse_expr`
(`datafusion/physical-expr/src/aggregate.rs`). Because `human_display_alias` is
`None` (`was_aliased == false`), it rewrites the aggregate's output name via
`replace_order_by_clause` / `replace_fn_name_clause`:
`last_value(v) ORDER BY [t DESC NULLS LAST] ...` → `first_value(v) ORDER
BY [t ASC NULLS FIRST] ...`
3. `adjust_window_sort_removal`
(`datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/mod.rs`)
installs the new window exec. Its schema field is now renamed, but the parent
`ProjectionExec` still holds `Column("last_value(...) ORDER BY [... DESC ...]",
3)`.
4. `ProjectionMapping::try_new`
(`datafusion/physical-expr/src/projection.rs`) asserts that the projection's
column name matches the input schema field name, and fails.
Note the contrast with the window-UDF path: `WindowUDFExpr::reverse_expr`
(`datafusion/physical-plan/src/windows/mod.rs`) explicitly carries over `name:
self.name.clone()`, so reversal there preserves the output field name and
nothing breaks. Renaming on reversal only makes sense for `AggregateExec`,
where no parent node references the aggregate by its schema name.
This is why the equivalent SQL query does **not** fail: in SQL,
`last_value(v) OVER (...)` resolves to the `last_value` *window UDF*
(`datafusion/functions-window/src/nth_value.rs`), which reverses without
renaming. The failure shows up when the aggregate UDAF (`last_value_udaf()`) is
used as a window function, which is reachable through the DataFrame API.
### To Reproduce
Add as `datafusion/core/tests/df_window_reverse_mre.rs` and run `cargo test
-p datafusion --test df_window_reverse_mre`:
```rust
use std::sync::Arc;
use arrow::array::{Int64Array, RecordBatch};
use arrow::datatypes::{DataType, Field, Schema};
use datafusion::prelude::*;
fn last_value_over(ascending: bool) -> Expr {
use datafusion::logical_expr::{ExprFunctionExt as _,
expr::WindowFunction};
Expr::from(WindowFunction::new(
datafusion::functions_aggregate::first_last::last_value_udaf(),
vec![col("v")],
))
.order_by(vec![col("t").sort(ascending, false)])
.build()
.unwrap()
}
#[tokio::test]
async fn window_reverse_rename_breaks_parent_projection() {
let schema = Arc::new(Schema::new(vec![
Field::new("t", DataType::Int64, false), // NOT NULL is required
Field::new("v", DataType::Int64, true),
]));
let batch = RecordBatch::try_new(
Arc::clone(&schema),
vec![
Arc::new(Int64Array::from(vec![1, 2])),
Arc::new(Int64Array::from(vec![None, Some(10)])),
],
)
.unwrap();
let ctx = SessionContext::new();
let df = ctx
.read_batch(batch)
.unwrap()
.with_column("asc_win", last_value_over(true))
.unwrap()
.with_column("desc_win", last_value_over(false))
.unwrap();
match df.collect().await {
Ok(batches) => println!("OK: {} batches", batches.len()),
Err(err) => panic!("FAILED: {err}"),
}
}
```
Two window expressions with opposite `ORDER BY` directions are needed so
that the optimizer reverses the second one; `t` must be non-nullable for the
ordering equivalence to trigger the reversal.
### Expected behavior
Reversing a window expression should not change the window's output field
name; the plan schema must stay stable so parent projections keep resolving.
Concretely, the aggregate-backed window reversal
(`Plain`/`SlidingAggregateWindowExpr::get_reverse_expr`) should preserve the
original output name, matching what `WindowUDFExpr::reverse_expr` already does.
### Additional context
Reproduced on `main` (`d7b8e4fc1`) and originally observed on 54.1.0.
Scope note: this issue is only about the renaming/plan-schema breakage.
There is a separate, independent gap behind it — with the name preserved
locally, the same query then fails at execution with:
```
NotImplemented("Aggregate can not be used as a sliding accumulator because
`retract_batch` is not implemented: last_value(...)")
```
because the reversed frame `ROWS CURRENT ROW .. UNBOUNDED FOLLOWING` is not
`is_ever_expanding()`, so `get_reverse_expr` builds a
`SlidingAggregateWindowExpr` that requires `retract_batch`. That is a distinct
problem (missing `retract_batch` / frame classification on reversal) and is not
what this issue asks to fix.
--
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]
