This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 4c031a267e fix: preserve output field name when reversing aggregate
window expressions (#24887)
4c031a267e is described below
commit 4c031a267e8621cb0ec8c5c5a9853c4c0e1c97c0
Author: Tim Saucer <[email protected]>
AuthorDate: Thu Sep 10 02:15:07 2026 +0000
fix: preserve output field name when reversing aggregate window expressions
(#24887)
## Which issue does this PR close?
- Closes #24884.
## Rationale for this change
Using an aggregate UDAF as a window function (reachable through the
DataFrame API) fails to plan whenever the physical optimizer decides to
reverse the window to avoid an extra sort:
```
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.
```
`get_best_fitting_window` swaps in `WindowExpr::get_reverse_expr()`, and
for aggregate-backed window expressions that reaches
`AggregateFunctionExpr::reverse_expr`, which rewrites the aggregate's
output *name* (`last_value(v) ORDER BY [t DESC]` → `first_value(v) ORDER
BY [t ASC]`). A window exec derives its schema from
`WindowExpr::field()`, so the node's output column gets renamed while
the parent `ProjectionExec` still holds a `Column` with the old name.
Renaming on reversal only makes sense for `AggregateExec`, which pins
its schema at construction (`try_new` builds the schema *before*
reversing exprs) — there the renamed name is a useful signal in
`EXPLAIN` of which implementation actually runs, and no parent
references it. The window path has the opposite requirement, which is
why `WindowUDFExpr::reverse_expr` already carries `name` over unchanged.
This makes the aggregate-backed window path behave the same way.
The equivalent SQL query does not fail, because in SQL `last_value(v)
OVER (...)` resolves to the `last_value` *window UDF*, which reverses
without renaming.
### Scope
Reviewed the rest of this bug class while here:
- `first_value` ↔ `last_value` is the only `ReversedUDAF::Reversed` pair
with a different name; `array_agg`, `string_agg` and `nth_value` reverse
to themselves, so the rewrite is a no-op for them.
- Two rules reach `get_best_fitting_window` — `enforce_sorting` and
`enforce_distribution` — so the fix is applied at the `get_reverse_expr`
level to cover both.
- `OptimizeAggregateOrder` also calls `reverse_expr`, but
`AggregateExec::with_new_aggr_exprs` keeps the original schema, so it
cannot rename an output field.
- `OptimizationInvariantChecker` does compare field names, but only for
the **root** plan schema, so a rename on an intermediate node under a
name-preserving projection is invisible to it. Hence the local assertion
below.
## What changes are included in this PR?
- `AggregateFunctionExpr::reverse_expr` is refactored into
`reverse_expr_inner(preserve_name)`, with a new public
`reverse_expr_preserving_name()`. Existing `reverse_expr` behavior is
unchanged.
- `PlainAggregateWindowExpr::get_reverse_expr` and
`SlidingAggregateWindowExpr::get_reverse_expr` use the name-preserving
variant. Their bodies were byte-identical, so they are factored into a
shared `reverse_aggregate_window_expr` helper.
- `get_best_fitting_window` now asserts that reversal did not change any
output field name, so a future renaming `WindowExpr` implementation
fails there — naming the culprit — instead of at a distant
`ProjectionMapping` assertion.
Note that the prefix-based `replace_fn_name_clause` also mangled
user-supplied aliases beginning with `last_value`/`first_value` (e.g. an
alias `last_value_desc` became `first_value_desc`); that is fixed on the
window path too.
## What is the testing strategy for this PR?
Two new regression tests, both verified to fail without the fix and pass
with it:
- `datafusion/core/tests/dataframe/mod.rs` —
`window_reversal_preserves_output_field_names` reproduces the issue
through the DataFrame API and reproduces the reported error exactly. It
asserts on `create_physical_plan()` rather than `collect()`, because
execution then hits the separate missing-`retract_batch` gap tracked by
#24885; a comment marks where to upgrade the assertion once that lands.
- `datafusion/core/tests/physical_optimizer/window_optimize.rs` —
`test_window_reversal_preserves_output_field_names` builds two
opposite-`ORDER BY` aggregate-UDAF windows under a projection, runs
`EnsureRequirements`, and asserts the optimized plan's schema field
names are unchanged. Without the fix it fails with `Input field name
first_value_desc does not match with the projection expression
last_value_desc`.
The existing `test_reverse_expr_preserves_non_aliased_display_path` and
the two neighboring `reverse_expr` display tests still pass, confirming
the plain-aggregate renaming is untouched. All 510 sqllogictest files
pass with no expectation changes — SQL cannot reach the renaming branch,
and existing reversal expectations already show the original name with a
reversed frame, which is exactly the shape this fix produces.
Full extended suite (`--features
avro,json,backtrace,extended_tests,recursive_protection,parquet_encryption`):
11055 passed, 0 failed. `cargo clippy --all-targets --all-features -- -D
warnings` clean.
## Are there any user-facing changes?
Queries that previously failed to plan now plan successfully; no
expected output or plan text changes for anything that worked before.
`AggregateFunctionExpr::reverse_expr_preserving_name` is a new public
method — additive, no breaking API changes.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
---
datafusion/core/tests/dataframe/mod.rs | 63 ++++++++++++
.../tests/physical_optimizer/window_optimize.rs | 110 ++++++++++++++++++++-
datafusion/physical-expr/src/aggregate.rs | 30 +++++-
datafusion/physical-expr/src/window/aggregate.rs | 82 +++++++++------
.../physical-expr/src/window/sliding_aggregate.rs | 40 ++------
datafusion/physical-plan/src/windows/mod.rs | 16 ++-
6 files changed, 272 insertions(+), 69 deletions(-)
diff --git a/datafusion/core/tests/dataframe/mod.rs
b/datafusion/core/tests/dataframe/mod.rs
index 4e32ea3116..9b0ca1b276 100644
--- a/datafusion/core/tests/dataframe/mod.rs
+++ b/datafusion/core/tests/dataframe/mod.rs
@@ -1336,6 +1336,69 @@ async fn window_aggregates_with_filter() -> Result<()> {
Ok(())
}
+// Test issue: https://github.com/apache/datafusion/issues/24884
+//
+// When the physical optimizer reverses a window expression to avoid an extra
+// sort, the reversed expression must keep its output field name. Otherwise the
+// window exec's schema changes while the parent projection still references
the
+// old column name, and planning fails.
+//
+// Note this only asserts on planning: executing the plan currently hits a
+// separate gap (missing `retract_batch` on the reversed sliding frame),
tracked
+// by https://github.com/apache/datafusion/issues/24885. Once that is fixed,
this
+// test can be extended to collect results.
+#[tokio::test]
+async fn window_reversal_preserves_output_field_names() -> Result<()> {
+ fn last_value_over(ascending: bool) -> Expr {
+ 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()
+ }
+
+ // `t` must be non-nullable for the ordering equivalence that makes the
+ // optimizer reverse the second window instead of adding a second sort.
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("t", DataType::Int64, false),
+ 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)])),
+ ],
+ )?;
+
+ let ctx = SessionContext::new();
+ let df = ctx
+ .read_batch(batch)?
+ .with_column("asc_win", last_value_over(true))?
+ .with_column("desc_win", last_value_over(false))?;
+
+ let logical_schema = df.schema().clone();
+ // Planning used to fail here with an internal error from
`EnsureRequirements`.
+ let physical_plan = df.create_physical_plan().await?;
+
+ let physical_names = physical_plan
+ .schema()
+ .fields()
+ .iter()
+ .map(|f| f.name().clone())
+ .collect::<Vec<_>>();
+ let logical_names = logical_schema
+ .fields()
+ .iter()
+ .map(|f| f.name().clone())
+ .collect::<Vec<_>>();
+ assert_eq!(physical_names, logical_names);
+
+ Ok(())
+}
+
// Test issue: https://github.com/apache/datafusion/issues/10346
#[tokio::test]
async fn test_select_over_aggregate_schema() -> Result<()> {
diff --git a/datafusion/core/tests/physical_optimizer/window_optimize.rs
b/datafusion/core/tests/physical_optimizer/window_optimize.rs
index 796f6b6259..909cb8b323 100644
--- a/datafusion/core/tests/physical_optimizer/window_optimize.rs
+++ b/datafusion/core/tests/physical_optimizer/window_optimize.rs
@@ -18,17 +18,23 @@
#[cfg(test)]
mod test {
use arrow::array::{Int32Array, RecordBatch};
- use arrow_schema::{DataType, Field, Schema};
+ use arrow_schema::{DataType, Field, Schema, SortOptions};
use datafusion_common::Result;
+ use datafusion_common::config::ConfigOptions;
use datafusion_datasource::memory::MemorySourceConfig;
use datafusion_datasource::source::DataSourceExec;
use datafusion_execution::TaskContext;
- use datafusion_expr::WindowFrame;
+ use datafusion_expr::{WindowFrame, WindowFunctionDefinition};
use datafusion_functions_aggregate::count::count_udaf;
+ use datafusion_functions_aggregate::first_last::last_value_udaf;
use datafusion_physical_expr::aggregate::AggregateExprBuilder;
use datafusion_physical_expr::expressions::{Column, col};
- use datafusion_physical_expr::window::PlainAggregateWindowExpr;
- use datafusion_physical_plan::windows::BoundedWindowAggExec;
+ use datafusion_physical_expr::window::{PlainAggregateWindowExpr,
WindowExpr};
+ use datafusion_physical_expr::{LexOrdering, PhysicalExpr,
PhysicalSortExpr};
+ use datafusion_physical_optimizer::PhysicalOptimizerRule;
+ use datafusion_physical_optimizer::ensure_requirements::EnsureRequirements;
+ use datafusion_physical_plan::projection::{ProjectionExec, ProjectionExpr};
+ use datafusion_physical_plan::windows::{BoundedWindowAggExec,
create_window_expr};
use datafusion_physical_plan::{ExecutionPlan, InputOrderMode, common};
use std::sync::Arc;
@@ -64,6 +70,102 @@ mod test {
Ok(())
}
+ /// Test case for <https://github.com/apache/datafusion/issues/24884>
+ ///
+ /// `EnsureRequirements` reverses the second window expression to reuse the
+ /// `t ASC` ordering required by the first one. That reversal must not
rename
+ /// the window's output field, otherwise the parent projection -- which
+ /// references those columns by name -- can no longer be resolved.
+ #[tokio::test]
+ async fn test_window_reversal_preserves_output_field_names() -> Result<()>
{
+ // `t` must be non-nullable for the ordering equivalence that triggers
+ // the reversal.
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("t", DataType::Int32, false),
+ Field::new("v", DataType::Int32, true),
+ ]));
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2])),
+ Arc::new(Int32Array::from(vec![None, Some(10)])),
+ ],
+ )?;
+ let ordering =
+ LexOrdering::new([PhysicalSortExpr::new_default(col("t",
&schema)?)])
+ .unwrap();
+ let source: Arc<dyn ExecutionPlan> =
Arc::new(DataSourceExec::new(Arc::new(
+ MemorySourceConfig::try_new(&[vec![batch]], Arc::clone(&schema),
None)?
+ .try_with_sort_information(vec![ordering])?,
+ )));
+
+ // Two `last_value` aggregate-UDAF windows with opposite ORDER BY
+ // directions, so the optimizer reverses the second one.
+ let window_expr = |ascending: bool| -> Result<Arc<dyn WindowExpr>> {
+ let sort_expr = PhysicalSortExpr::new(
+ col("t", &schema)?,
+ SortOptions::new(!ascending, false),
+ );
+ create_window_expr(
+ &WindowFunctionDefinition::AggregateUDF(last_value_udaf()),
+ format!("last_value_{}", if ascending { "asc" } else { "desc"
}),
+ &[col("v", &schema)?],
+ &[],
+ std::slice::from_ref(&sort_expr),
+ Arc::new(WindowFrame::new(Some(false))),
+ Arc::clone(&schema),
+ false,
+ false,
+ None,
+ )
+ };
+
+ let mut plan: Arc<dyn ExecutionPlan> = source;
+ for ascending in [true, false] {
+ plan = Arc::new(BoundedWindowAggExec::try_new(
+ vec![window_expr(ascending)?],
+ plan,
+ InputOrderMode::Sorted,
+ false,
+ )?);
+ }
+
+ // A projection that references the window outputs by name, as the
+ // physical planner would produce.
+ let plan_schema = plan.schema();
+ let projection_exprs = plan_schema
+ .fields()
+ .iter()
+ .enumerate()
+ .map(|(idx, field)| ProjectionExpr {
+ expr: Arc::new(Column::new(field.name(), idx)) as Arc<dyn
PhysicalExpr>,
+ alias: field.name().clone(),
+ })
+ .collect::<Vec<_>>();
+ let plan: Arc<dyn ExecutionPlan> =
+ Arc::new(ProjectionExec::try_new(projection_exprs, plan)?);
+
+ // Used to fail with an internal error from
`ProjectionMapping::try_new`.
+ let optimized = EnsureRequirements::new()
+ .optimize(Arc::clone(&plan), &ConfigOptions::new())?;
+
+ let names = |plan: &Arc<dyn ExecutionPlan>| {
+ plan.schema()
+ .fields()
+ .iter()
+ .map(|f| f.name().clone())
+ .collect::<Vec<_>>()
+ };
+ assert_eq!(names(&optimized), names(&plan));
+
+ // The window exec below the projection must keep its field names too,
+ // otherwise the projection's columns would dangle.
+ let window = optimized.children()[0];
+ assert_eq!(names(window), names(&plan));
+
+ Ok(())
+ }
+
pub fn mock_data() -> Result<Arc<DataSourceExec>> {
let schema = Arc::new(Schema::new(vec![
Field::new("a", DataType::Int32, true),
diff --git a/datafusion/physical-expr/src/aggregate.rs
b/datafusion/physical-expr/src/aggregate.rs
index 5f045ca8c7..df22bc69d8 100644
--- a/datafusion/physical-expr/src/aggregate.rs
+++ b/datafusion/physical-expr/src/aggregate.rs
@@ -933,24 +933,44 @@ impl AggregateFunctionExpr {
/// For aggregates that do not support calculation in reverse,
/// returns None (which is the default value).
pub fn reverse_expr(&self) -> Option<AggregateFunctionExpr> {
+ self.reverse_expr_inner(false)
+ }
+
+ /// Same as [`Self::reverse_expr`], but the output `name` is always carried
+ /// over unchanged.
+ ///
+ /// Window execs derive their output schema from `WindowExpr::field()`,
which
+ /// for aggregate-backed window expressions is this expression's `name`. If
+ /// reversal renamed it, the rebuilt window exec would expose a differently
+ /// named column while parent plan nodes still reference the old one. This
+ /// mirrors `WindowUDFExpr::reverse_expr`, which preserves its name for the
+ /// same reason. `AggregateExec`, by contrast, pins its schema at
+ /// construction, so it can use [`Self::reverse_expr`] and let the name
+ /// reflect the function actually being evaluated.
+ pub(crate) fn reverse_expr_preserving_name(&self) ->
Option<AggregateFunctionExpr> {
+ self.reverse_expr_inner(true)
+ }
+
+ fn reverse_expr_inner(&self, preserve_name: bool) ->
Option<AggregateFunctionExpr> {
match self.fun.reverse_udf() {
ReversedUDAF::NotSupported => None,
ReversedUDAF::Identical => Some(self.clone()),
ReversedUDAF::Reversed(reverse_udf) => {
- let was_aliased = self.human_display_alias().is_some();
+ let keep_name = preserve_name ||
self.human_display_alias().is_some();
let mut name = self.name().to_string();
let mut human_display = self.human_display.clone();
// Reversing display follows two paths:
- // - aliased display keeps the output `name` unchanged and
rewrites only
- // the lowered expression in `human_display`.
+ // - aliased display (or an explicit request to keep the name)
keeps
+ // the output `name` unchanged and rewrites only the lowered
+ // expression in `human_display`.
// - non-aliased display rewrites the canonical `name`, and
rewrites
// `human_display` only when present.
// If the function is changed, we need to reverse order_by
clause as well
// i.e. First(a order by b asc null first) -> Last(a order by
b desc null last)
- if !was_aliased && self.fun().name() != reverse_udf.name() {
+ if !keep_name && self.fun().name() != reverse_udf.name() {
replace_order_by_clause(&mut name);
}
- if !was_aliased {
+ if !keep_name {
replace_fn_name_clause(
&mut name,
self.fun.name(),
diff --git a/datafusion/physical-expr/src/window/aggregate.rs
b/datafusion/physical-expr/src/window/aggregate.rs
index 7cfdcb167f..a021588edd 100644
--- a/datafusion/physical-expr/src/window/aggregate.rs
+++ b/datafusion/physical-expr/src/window/aggregate.rs
@@ -39,6 +39,53 @@ use datafusion_common::{Result, ScalarValue,
exec_datafusion_err};
use datafusion_expr::{Accumulator, WindowFrame, WindowFrameBound,
WindowFrameUnits};
use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
+/// Reverses an aggregate-backed window expression.
+///
+/// Shared by [`PlainAggregateWindowExpr::get_reverse_expr`] and
+/// [`SlidingAggregateWindowExpr::get_reverse_expr`], which are otherwise
+/// identical. The reversed frame decides which of the two variants is
produced:
+/// an ever-expanding frame can be evaluated by the plain (accumulating)
variant,
+/// anything else needs the sliding variant.
+///
+/// The aggregate is reversed with
+/// [`AggregateFunctionExpr::reverse_expr_preserving_name`] so that the
window's
+/// output field name is unchanged -- the window exec's schema is derived from
+/// [`WindowExpr::field`], and parent plan nodes reference those columns by
name.
+pub(crate) fn reverse_aggregate_window_expr(
+ aggregate: &AggregateFunctionExpr,
+ partition_by: &[Arc<dyn PhysicalExpr>],
+ order_by: &[PhysicalSortExpr],
+ window_frame: &WindowFrame,
+ filter: Option<&Arc<dyn PhysicalExpr>>,
+) -> Option<Arc<dyn WindowExpr>> {
+ aggregate
+ .reverse_expr_preserving_name()
+ .map(|reverse_expr| {
+ let reverse_expr = Arc::new(reverse_expr);
+ let reverse_order_by =
+ order_by.iter().map(|e| e.reverse()).collect::<Vec<_>>();
+ let reverse_window_frame = Arc::new(window_frame.reverse());
+ let filter = filter.cloned();
+ if reverse_window_frame.is_ever_expanding() {
+ Arc::new(PlainAggregateWindowExpr::new(
+ reverse_expr,
+ partition_by,
+ &reverse_order_by,
+ reverse_window_frame,
+ filter,
+ )) as _
+ } else {
+ Arc::new(SlidingAggregateWindowExpr::new(
+ reverse_expr,
+ partition_by,
+ &reverse_order_by,
+ reverse_window_frame,
+ filter,
+ )) as _
+ }
+ })
+}
+
/// A window expr that takes the form of an aggregate function.
///
/// See comments on [`WindowExpr`] for more details.
@@ -185,34 +232,13 @@ impl WindowExpr for PlainAggregateWindowExpr {
}
fn get_reverse_expr(&self) -> Option<Arc<dyn WindowExpr>> {
- self.aggregate.reverse_expr().map(|reverse_expr| {
- let reverse_window_frame = self.window_frame.reverse();
- if reverse_window_frame.is_ever_expanding() {
- Arc::new(PlainAggregateWindowExpr::new(
- Arc::new(reverse_expr),
- &self.partition_by.clone(),
- &self
- .order_by
- .iter()
- .map(|e| e.reverse())
- .collect::<Vec<_>>(),
- Arc::new(self.window_frame.reverse()),
- self.filter.clone(),
- )) as _
- } else {
- Arc::new(SlidingAggregateWindowExpr::new(
- Arc::new(reverse_expr),
- &self.partition_by.clone(),
- &self
- .order_by
- .iter()
- .map(|e| e.reverse())
- .collect::<Vec<_>>(),
- Arc::new(self.window_frame.reverse()),
- self.filter.clone(),
- )) as _
- }
- })
+ reverse_aggregate_window_expr(
+ &self.aggregate,
+ &self.partition_by,
+ &self.order_by,
+ &self.window_frame,
+ self.filter.as_ref(),
+ )
}
fn uses_bounded_memory(&self) -> bool {
diff --git a/datafusion/physical-expr/src/window/sliding_aggregate.rs
b/datafusion/physical-expr/src/window/sliding_aggregate.rs
index a39334f057..0954037acb 100644
--- a/datafusion/physical-expr/src/window/sliding_aggregate.rs
+++ b/datafusion/physical-expr/src/window/sliding_aggregate.rs
@@ -22,12 +22,11 @@ use std::ops::Range;
use std::sync::Arc;
use crate::aggregate::AggregateFunctionExpr;
+use crate::window::aggregate::reverse_aggregate_window_expr;
use crate::window::window_expr::{
AggregateWindowExpr, WindowEvalContext, WindowFn, filter_array,
};
-use crate::window::{
- PartitionBatches, PartitionWindowAggStates, PlainAggregateWindowExpr,
WindowExpr,
-};
+use crate::window::{PartitionBatches, PartitionWindowAggStates, WindowExpr};
use crate::{PhysicalExpr, expressions::PhysicalSortExpr};
use arrow::array::{ArrayRef, BooleanArray};
@@ -122,34 +121,13 @@ impl WindowExpr for SlidingAggregateWindowExpr {
}
fn get_reverse_expr(&self) -> Option<Arc<dyn WindowExpr>> {
- self.aggregate.reverse_expr().map(|reverse_expr| {
- let reverse_window_frame = self.window_frame.reverse();
- if reverse_window_frame.is_ever_expanding() {
- Arc::new(PlainAggregateWindowExpr::new(
- Arc::new(reverse_expr),
- &self.partition_by.clone(),
- &self
- .order_by
- .iter()
- .map(|e| e.reverse())
- .collect::<Vec<_>>(),
- Arc::new(self.window_frame.reverse()),
- self.filter.clone(),
- )) as _
- } else {
- Arc::new(SlidingAggregateWindowExpr::new(
- Arc::new(reverse_expr),
- &self.partition_by.clone(),
- &self
- .order_by
- .iter()
- .map(|e| e.reverse())
- .collect::<Vec<_>>(),
- Arc::new(self.window_frame.reverse()),
- self.filter.clone(),
- )) as _
- }
- })
+ reverse_aggregate_window_expr(
+ &self.aggregate,
+ &self.partition_by,
+ &self.order_by,
+ &self.window_frame,
+ self.filter.as_ref(),
+ )
}
fn uses_bounded_memory(&self) -> bool {
diff --git a/datafusion/physical-plan/src/windows/mod.rs
b/datafusion/physical-plan/src/windows/mod.rs
index 3f33dfedfd..7c5f55f661 100644
--- a/datafusion/physical-plan/src/windows/mod.rs
+++ b/datafusion/physical-plan/src/windows/mod.rs
@@ -33,7 +33,7 @@ use crate::{
use arrow::datatypes::{Schema, SchemaRef};
use arrow_schema::{FieldRef, SortOptions};
-use datafusion_common::{Result, exec_err};
+use datafusion_common::{Result, assert_or_internal_err, exec_err};
use datafusion_expr::{
LimitEffect, PartitionEvaluator, ReversedUDWF, SetMonotonicity,
WindowFrame,
WindowFunctionDefinition, WindowUDF,
@@ -623,6 +623,20 @@ pub fn get_best_fitting_window(
.map(|e| e.get_reverse_expr())
.collect::<Option<Vec<_>>>()
{
+ // The rebuilt exec derives its schema from `WindowExpr::field()`,
so a
+ // reversal that renames the output field would silently change
this
+ // node's schema while parent nodes still reference the old column
+ // names. Catch that here, where the culprit is identifiable.
+ for (reversed, original) in
reversed_window_expr.iter().zip(window_exprs) {
+ let (reversed_field, original_field) =
+ (reversed.field()?, original.field()?);
+ assert_or_internal_err!(
+ reversed_field.name() == original_field.name(),
+ "Reversing window expression changed its output field name
from {} to {}",
+ original_field.name(),
+ reversed_field.name()
+ );
+ }
reversed_window_expr
} else {
// Cannot take reverse of any of the window expr
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
