Support min max aggregates in window functions with sliding windows

datafusion/core/src/datasource/file_format/parquet.rs

@@ -47,7 +47,7 @@ use crate::config::OPT_PARQUET_SKIP_METADATA;
 use crate::datasource::{create_max_min_accs, get_col_stats};
 use crate::error::Result;
 use crate::logical_expr::Expr;
-use crate::physical_plan::expressions::{MaxAccumulator, MinAccumulator};
+use crate::physical_plan::expressions::{MinAccumulator, SlidingMaxAccumulator};
 use crate::physical_plan::file_format::{ParquetExec, SchemaAdapter};
 use crate::physical_plan::{Accumulator, ExecutionPlan, Statistics};
 
@@ -220,7 +220,7 @@ impl FileFormat for ParquetFormat {
 }
 
 fn summarize_min_max(
-    max_values: &mut [Option<MaxAccumulator>],
+    max_values: &mut [Option<SlidingMaxAccumulator>],
     min_values: &mut [Option<MinAccumulator>],
     fields: &[Field],
     i: usize,

datafusion/core/src/datasource/mod.rs

@@ -41,7 +41,7 @@ pub use self::view::ViewTable;
 use crate::arrow::datatypes::{Schema, SchemaRef};
 use crate::error::Result;
 pub use crate::logical_expr::TableType;
-use crate::physical_plan::expressions::{MaxAccumulator, MinAccumulator};
+use crate::physical_plan::expressions::{MinAccumulator, SlidingMaxAccumulator};
 use crate::physical_plan::{Accumulator, ColumnStatistics, Statistics};
 use futures::StreamExt;
 
@@ -155,11 +155,14 @@ pub async fn get_statistics_with_limit(
 
 fn create_max_min_accs(
     schema: &Schema,
-) -> (Vec<Option<MaxAccumulator>>, Vec<Option<MinAccumulator>>) {
-    let max_values: Vec<Option<MaxAccumulator>> = schema
+) -> (
+    Vec<Option<SlidingMaxAccumulator>>,
+    Vec<Option<MinAccumulator>>,
+) {
+    let max_values: Vec<Option<SlidingMaxAccumulator>> = schema
         .fields()
         .iter()
-        .map(|field| MaxAccumulator::try_new(field.data_type()).ok())
+        .map(|field| SlidingMaxAccumulator::try_new(field.data_type()).ok())
         .collect::<Vec<_>>();
     let min_values: Vec<Option<MinAccumulator>> = schema
         .fields()
@@ -172,7 +175,7 @@ fn create_max_min_accs(
 fn get_col_stats(
     schema: &Schema,
     null_counts: Vec<usize>,
-    max_values: &mut [Option<MaxAccumulator>],
+    max_values: &mut [Option<SlidingMaxAccumulator>],
     min_values: &mut [Option<MinAccumulator>],
 ) -> Vec<ColumnStatistics> {
     (0..schema.fields().len())

datafusion/core/src/physical_plan/windows/mod.rs

@@ -33,7 +33,9 @@ use datafusion_expr::{
     window_function::{signature_for_built_in, BuiltInWindowFunction, WindowFunction},
     WindowFrame,
 };
-use datafusion_physical_expr::window::BuiltInWindowFunctionExpr;
+use datafusion_physical_expr::window::{
+    BuiltInWindowFunctionExpr, SlidingAggregateWindowExpr,
+};
 use std::convert::TryInto;
 use std::sync::Arc;
 
@@ -55,12 +57,25 @@ pub fn create_window_expr(
     input_schema: &Schema,
 ) -> Result<Arc<dyn WindowExpr>> {
     Ok(match fun {
-        WindowFunction::AggregateFunction(fun) => Arc::new(AggregateWindowExpr::new(
-            aggregates::create_aggregate_expr(fun, false, args, input_schema, name)?,
-            partition_by,
-            order_by,
-            window_frame,
-        )),
+        WindowFunction::AggregateFunction(fun) => {
+            let aggregate =
+                aggregates::create_aggregate_expr(fun, false, args, input_schema, name)?;
+            if !window_frame.start_bound.is_unbounded() {
+                Arc::new(SlidingAggregateWindowExpr::new(
+                    aggregate,
+                    partition_by,
+                    order_by,
+                    window_frame,
+                ))
+            } else {
+                Arc::new(AggregateWindowExpr::new(
+                    aggregate,
+                    partition_by,
+                    order_by,
+                    window_frame,
+                ))
+            }
+        }
         WindowFunction::BuiltInWindowFunction(fun) => Arc::new(BuiltInWindowExpr::new(
             create_built_in_window_expr(fun, args, input_schema, name)?,
             partition_by,

datafusion/core/tests/sql/aggregates.rs

@@ -682,6 +682,60 @@ async fn aggregate_grouped_min() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn aggregate_min_max_w_custom_window_frames() -> Result<()> {
+    let ctx = SessionContext::new();
+    register_aggregate_csv(&ctx).await?;
+    let sql =
+        "SELECT
+        MIN(c12) OVER (ORDER BY C12 RANGE BETWEEN 0.3 PRECEDING AND 0.2 FOLLOWING) as min1,
+        MAX(c12) OVER (ORDER BY C11 RANGE BETWEEN 0.1 PRECEDING AND 0.2 FOLLOWING) as max1
+        FROM aggregate_test_100
+        ORDER BY C9
+        LIMIT 5";
+    let actual = execute_to_batches(&ctx, sql).await;
+    let expected = vec![
+        "+---------------------+--------------------+",
+        "| min1                | max1               |",
+        "+---------------------+--------------------+",
+        "| 0.01479305307777301 | 0.9965400387585364 |",
+        "| 0.01479305307777301 | 0.9800193410444061 |",
+        "| 0.01479305307777301 | 0.9706712283358269 |",
+        "| 0.2667177795079635  | 0.9965400387585364 |",
+        "| 0.3600766362333053  | 0.9706712283358269 |",
+        "+---------------------+--------------------+",
+    ];
+    assert_batches_eq!(expected, &actual);
+    Ok(())
+}
+
+#[tokio::test]
+async fn aggregate_min_max_w_custom_window_frames_unbounded_start() -> Result<()> {
+    let ctx = SessionContext::new();
+    register_aggregate_csv(&ctx).await?;
+    let sql =
+        "SELECT
+        MIN(c12) OVER (ORDER BY C12 RANGE BETWEEN UNBOUNDED PRECEDING AND 0.2 FOLLOWING) as min1,
+        MAX(c12) OVER (ORDER BY C11 RANGE BETWEEN UNBOUNDED PRECEDING AND 0.2 FOLLOWING) as max1
+        FROM aggregate_test_100
+        ORDER BY C9
+        LIMIT 5";
+    let actual = execute_to_batches(&ctx, sql).await;
+    let expected = vec![
+        "+---------------------+--------------------+",
+        "| min1                | max1               |",
+        "+---------------------+--------------------+",
+        "| 0.01479305307777301 | 0.9965400387585364 |",
+        "| 0.01479305307777301 | 0.9800193410444061 |",
+        "| 0.01479305307777301 | 0.9800193410444061 |",
+        "| 0.01479305307777301 | 0.9965400387585364 |",
+        "| 0.01479305307777301 | 0.9800193410444061 |",
+        "+---------------------+--------------------+",
+    ];
+    assert_batches_eq!(expected, &actual);
+    Ok(())
+}
+
 #[tokio::test]
 async fn aggregate_avg_add() -> Result<()> {
     let results = execute_with_partition(

datafusion/expr/src/window_frame.rs

@@ -147,6 +147,16 @@ pub enum WindowFrameBound {
     Following(ScalarValue),
 }
 
+impl WindowFrameBound {
+    pub fn is_unbounded(&self) -> bool {
+        match self {
+            WindowFrameBound::Preceding(elem) => elem.is_null(),
+            WindowFrameBound::CurrentRow => false,
+            WindowFrameBound::Following(elem) => elem.is_null(),
+        }
+    }
+}
+
 impl TryFrom<ast::WindowFrameBound> for WindowFrameBound {
     type Error = DataFusionError;
 

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

@@ -21,7 +21,9 @@ use std::any::Any;
 use std::convert::TryFrom;
 use std::sync::Arc;
 
-use crate::aggregate::row_accumulator::RowAccumulator;
+use crate::aggregate::row_accumulator::{
+    is_row_accumulator_support_dtype, RowAccumulator,
+};
 use crate::aggregate::sum;
 use crate::expressions::format_state_name;
 use crate::{AggregateExpr, PhysicalExpr};
@@ -105,19 +107,7 @@ impl AggregateExpr for Avg {
     }
 
     fn row_accumulator_supported(&self) -> bool {
-        matches!(
-            self.data_type,
-            DataType::UInt8
-                | DataType::UInt16
-                | DataType::UInt32
-                | DataType::UInt64
-                | DataType::Int8
-                | DataType::Int16
-                | DataType::Int32
-                | DataType::Int64
-                | DataType::Float32
-                | DataType::Float64
-        )
+        is_row_accumulator_support_dtype(&self.data_type)
     }
 
     fn create_row_accumulator(

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

@@ -104,6 +104,10 @@ impl AggregateExpr for Count {
     ) -> Result<Box<dyn RowAccumulator>> {
         Ok(Box::new(CountRowAccumulator::new(start_index)))
     }
+
+    fn create_sliding_accumulator(&self) -> Result<Box<dyn Accumulator>> {
+        Ok(Box::new(CountAccumulator::new()))
+    }
 }
 
 #[derive(Debug)]