fix bounded but not fixed roundtrip

Author: buraksenn

datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs

@@ -1163,7 +1163,9 @@ mod tests {
     use crate::expressions::PhysicalSortExpr;
     use crate::projection::ProjectionExec;
     use crate::streaming::{PartitionStream, StreamingTableExec};
-    use crate::windows::{create_window_expr, BoundedWindowAggExec, InputOrderMode};
+    use crate::windows::{
+        create_udwf_window_expr, create_window_expr, BoundedWindowAggExec, InputOrderMode,
+    };
     use crate::{execute_stream, get_plan_string, ExecutionPlan};
 
     use arrow_array::builder::{Int64Builder, UInt64Builder};
@@ -1180,13 +1182,14 @@ mod tests {
         WindowFrame, WindowFrameBound, WindowFrameUnits, WindowFunctionDefinition,
     };
     use datafusion_functions_aggregate::count::count_udaf;
-    use datafusion_functions_window::nth_value::first_value_udwf;
     use datafusion_functions_window::nth_value::last_value_udwf;
     use datafusion_functions_window::nth_value::nth_value_udwf;
-    use datafusion_physical_expr::expressions::{col, lit, Column};
+    use datafusion_physical_expr::expressions::{col, Column, Literal};
     use datafusion_physical_expr::{LexOrdering, PhysicalExpr};
 
+    use crate::common::collect;
     use crate::memory::MemoryExec;
+    use datafusion_physical_expr::window::BuiltInWindowExpr;
     use futures::future::Shared;
     use futures::{pin_mut, ready, FutureExt, Stream, StreamExt};
     use itertools::Itertools;
@@ -1524,30 +1527,39 @@ mod tests {
         )
         .map(|e| Arc::new(e) as Arc<dyn ExecutionPlan>)?;
         let col_a = col("a", &schema)?;
-        let nth_value_func1 = NthValue::nth(
-            "nth_value(-1)",
-            Arc::clone(&col_a),
-            DataType::Int32,
-            1,
+        let nth_value_func1 = create_udwf_window_expr(
+            &nth_value_udwf(),
+            &[
+                Arc::clone(&col_a),
+                Arc::new(Literal::new(ScalarValue::Int32(Some(1)))),
+            ],
+            &schema,
+            "nth_value(-1)".to_string(),
             false,
         )?
         .reverse_expr()
         .unwrap();
-        let nth_value_func2 = NthValue::nth(
-            "nth_value(-2)",
-            Arc::clone(&col_a),
-            DataType::Int32,
-            2,
+        let nth_value_func2 = create_udwf_window_expr(
+            &nth_value_udwf(),
+            &[
+                Arc::clone(&col_a),
+                Arc::new(Literal::new(ScalarValue::Int32(Some(2)))),
+            ],
+            &schema,
+            "nth_value(-2)".to_string(),
             false,
         )?
         .reverse_expr()
         .unwrap();
-        let last_value_func = Arc::new(NthValue::last(
-            "last",
-            Arc::clone(&col_a),
-            DataType::Int32,
+
+        let last_value_func = create_udwf_window_expr(
+            &last_value_udwf(),
+            &[Arc::clone(&col_a)],
+            &schema,
+            "last".to_string(),
             false,
-        )) as _;
+        )?;
+
         let window_exprs = vec![
             // LAST_VALUE(a)
             Arc::new(BuiltInWindowExpr::new(

datafusion/proto/tests/cases/roundtrip_physical_plan.rs

@@ -24,7 +24,7 @@ use std::vec;
 use arrow::array::RecordBatch;
 use arrow::csv::WriterBuilder;
 use arrow::datatypes::{Fields, TimeUnit};
-use datafusion::physical_expr::aggregate::{AggregateExprBuilder, AggregateFunctionExpr};
+use datafusion::physical_expr::aggregate::AggregateExprBuilder;
 use datafusion::physical_plan::coalesce_batches::CoalesceBatchesExec;
 use datafusion_expr::dml::InsertOp;
 use datafusion_functions_aggregate::approx_percentile_cont::approx_percentile_cont_udaf;
@@ -47,7 +47,7 @@ use datafusion::datasource::physical_plan::{
 };
 use datafusion::execution::FunctionRegistry;
 use datafusion::functions_aggregate::sum::sum_udaf;
-use datafusion::functions_window::nth_value::nth_value_udwf;
+use datafusion::functions_window::nth_value::first_value_udwf;
 use datafusion::logical_expr::{create_udf, JoinType, Operator, Volatility};
 use datafusion::physical_expr::expressions::Literal;
 use datafusion::physical_expr::window::{BuiltInWindowExpr, SlidingAggregateWindowExpr};
@@ -281,10 +281,16 @@ fn roundtrip_window() -> Result<()> {
         WindowFrameBound::CurrentRow,
     );
 
-    let nth_value_window =
-        create_udwf_window_expr(&nth_value_udwf(), &[col("a", &schema)?], schema.as_ref(),  "FIRST_VALUE(a) PARTITION BY [b] ORDER BY [a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW".to_string(), false)?;
+    let first_value_window = create_udwf_window_expr(
+        &first_value_udwf(),
+        &[col("a", &schema)?],
+        schema.as_ref(),
+        "FIRST_VALUE(a) PARTITION BY [b] ORDER BY [a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW".to_string(),
+        false,
+    )?;
+    //  "FIRST_VALUE(a) PARTITION BY [b] ORDER BY [a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW",
     let builtin_window_expr = Arc::new(BuiltInWindowExpr::new(
-        nth_value_window,
+        first_value_window,
         &[col("b", &schema)?],
         &LexOrdering {
             inner: vec![PhysicalSortExpr {