fix: [branch-48] Revert "Improve performance of constant aggregate window expression" (apache#16307)

* Revert "Improve performance of constant aggregate window expression (apache#16234)"

This reverts commit 0c30374.

* update changelog

* update changelog

Author: andygrove

datafusion/physical-expr/src/window/aggregate.rs

@@ -34,7 +34,7 @@ use arrow::array::ArrayRef;
 use arrow::datatypes::FieldRef;
 use arrow::record_batch::RecordBatch;
 use datafusion_common::{DataFusionError, Result, ScalarValue};
-use datafusion_expr::{Accumulator, WindowFrame, WindowFrameBound, WindowFrameUnits};
+use datafusion_expr::{Accumulator, WindowFrame};
 use datafusion_physical_expr_common::sort_expr::LexOrdering;
 
 /// A window expr that takes the form of an aggregate function.
@@ -46,7 +46,6 @@ pub struct PlainAggregateWindowExpr {
     partition_by: Vec<Arc<dyn PhysicalExpr>>,
     order_by: LexOrdering,
     window_frame: Arc<WindowFrame>,
-    is_constant_in_partition: bool,
 }
 
 impl PlainAggregateWindowExpr {
@@ -57,14 +56,11 @@ impl PlainAggregateWindowExpr {
         order_by: &LexOrdering,
         window_frame: Arc<WindowFrame>,
     ) -> Self {
-        let is_constant_in_partition =
-            Self::is_window_constant_in_partition(order_by, &window_frame);
         Self {
             aggregate,
             partition_by: partition_by.to_vec(),
             order_by: order_by.clone(),
             window_frame,
-            is_constant_in_partition,
         }
     }
 
@@ -89,30 +85,6 @@ impl PlainAggregateWindowExpr {
             );
         }
     }
-
-    // Returns true if every row in the partition has the same window frame. This allows
-    // for preventing bound + function calculation for every row due to the values being the
-    // same.
-    //
-    // This occurs when both bounds fall under either condition below:
-    //  1. Bound is unbounded (`Preceding` or `Following`)
-    //  2. Bound is `CurrentRow` while using `Range` units with no order by clause
-    //  This results in an invalid range specification. Following PostgreSQL’s convention,
-    //  we interpret this as the entire partition being used for the current window frame.
-    fn is_window_constant_in_partition(
-        order_by: &LexOrdering,
-        window_frame: &WindowFrame,
-    ) -> bool {
-        let is_constant_bound = |bound: &WindowFrameBound| match bound {
-            WindowFrameBound::CurrentRow => {
-                window_frame.units == WindowFrameUnits::Range && order_by.is_empty()
-            }
-            _ => bound.is_unbounded(),
-        };
-
-        is_constant_bound(&window_frame.start_bound)
-            && is_constant_bound(&window_frame.end_bound)
-    }
 }
 
 /// peer based evaluation based on the fact that batch is pre-sorted given the sort columns
@@ -241,8 +213,4 @@ impl AggregateWindowExpr for PlainAggregateWindowExpr {
             accumulator.evaluate()
         }
     }
-
-    fn is_constant_in_partition(&self) -> bool {
-        self.is_constant_in_partition
-    }
 }

datafusion/physical-expr/src/window/sliding_aggregate.rs

@@ -210,8 +210,4 @@ impl AggregateWindowExpr for SlidingAggregateWindowExpr {
             accumulator.evaluate()
         }
     }
-
-    fn is_constant_in_partition(&self) -> bool {
-        false
-    }
 }

datafusion/physical-expr/src/window/window_expr.rs

@@ -186,10 +186,6 @@ pub trait AggregateWindowExpr: WindowExpr {
         accumulator: &mut Box<dyn Accumulator>,
     ) -> Result<ScalarValue>;
 
-    /// Indicates whether this window function always produces the same result
-    /// for all rows in the partition.
-    fn is_constant_in_partition(&self) -> bool;
-
     /// Evaluates the window function against the batch.
     fn aggregate_evaluate(&self, batch: &RecordBatch) -> Result<ArrayRef> {
         let mut accumulator = self.get_accumulator()?;
@@ -276,13 +272,8 @@ pub trait AggregateWindowExpr: WindowExpr {
         not_end: bool,
     ) -> Result<ArrayRef> {
         let values = self.evaluate_args(record_batch)?;
-
-        if self.is_constant_in_partition() {
-            accumulator.update_batch(&values)?;
-            let value = accumulator.evaluate()?;
-            return value.to_array_of_size(record_batch.num_rows());
-        }
         let order_bys = get_orderby_values(self.order_by_columns(record_batch)?);
+
         let most_recent_row_order_bys = most_recent_row
             .map(|batch| self.order_by_columns(batch))
             .transpose()?

dev/changelog/48.0.0.md

@@ -19,7 +19,7 @@ under the License.
 
 # Apache DataFusion 48.0.0 Changelog
 
-This release consists of 267 commits from 89 contributors. See credits at the end of this changelog for more information.
+This release consists of 266 commits from 88 contributors. See credits at the end of this changelog for more information.
 
 **Breaking changes:**
 
@@ -297,7 +297,6 @@ This release consists of 267 commits from 89 contributors. See credits at the en
 - Simplify FileSource / SchemaAdapterFactory API [#16214](https://github.com/apache/datafusion/pull/16214) (alamb)
 - Add dicts to aggregation fuzz testing [#16232](https://github.com/apache/datafusion/pull/16232) (blaginin)
 - chore(deps): bump sysinfo from 0.35.1 to 0.35.2 [#16247](https://github.com/apache/datafusion/pull/16247) (dependabot[bot])
-- Improve performance of constant aggregate window expression [#16234](https://github.com/apache/datafusion/pull/16234) (suibianwanwank)
 - Support compound identifier when parsing tuples [#16225](https://github.com/apache/datafusion/pull/16225) (hozan23)
 - Schema adapter helper [#16108](https://github.com/apache/datafusion/pull/16108) (kosiew)
 - Update tpch, clickbench, sort_tpch to mark failed queries [#16182](https://github.com/apache/datafusion/pull/16182) (ding-young)
@@ -397,7 +396,6 @@ Thank you to everyone who contributed to this release. Here is a breakdown of co
      1	irenjj
      1	jsai28
      1	m09526
-     1	suibianwanwan
      1	the0ninjas
      1	wiedld
 ```