Reduce size of Expr struct (#16207)

* First pass at incorporating boxed WindowFunction

* Second pass

* Adjust tests

* cargo fmt

* fmt

* Add test

* clippy

* clippy

* fmt

* Adjust comments

Author: hendrikmakait

datafusion-examples/examples/advanced_udwf.rs

@@ -191,7 +191,7 @@ impl WindowUDFImpl for SimplifySmoothItUdf {
     /// default implementation will not be called (left as `todo!()`)
     fn simplify(&self) -> Option<WindowFunctionSimplification> {
         let simplify = |window_function: WindowFunction, _: &dyn SimplifyInfo| {
-            Ok(Expr::WindowFunction(WindowFunction {
+            Ok(Expr::from(WindowFunction {
                 fun: WindowFunctionDefinition::AggregateUDF(avg_udaf()),
                 params: WindowFunctionParams {
                     args: window_function.params.args,

datafusion/core/src/physical_planner.rs

@@ -574,27 +574,25 @@ impl DefaultPhysicalPlanner {
                 let input_exec = children.one()?;
 
                 let get_sort_keys = |expr: &Expr| match expr {
-                    Expr::WindowFunction(WindowFunction {
-                        params:
-                            WindowFunctionParams {
-                                ref partition_by,
-                                ref order_by,
-                                ..
-                            },
-                        ..
-                    }) => generate_sort_key(partition_by, order_by),
+                    Expr::WindowFunction(window_fun) => {
+                        let WindowFunctionParams {
+                            ref partition_by,
+                            ref order_by,
+                            ..
+                        } = &window_fun.as_ref().params;
+                        generate_sort_key(partition_by, order_by)
+                    }
                     Expr::Alias(Alias { expr, .. }) => {
                         // Convert &Box<T> to &T
                         match &**expr {
-                            Expr::WindowFunction(WindowFunction {
-                                params:
-                                    WindowFunctionParams {
-                                        ref partition_by,
-                                        ref order_by,
-                                        ..
-                                    },
-                                ..
-                            }) => generate_sort_key(partition_by, order_by),
+                            Expr::WindowFunction(window_fun) => {
+                                let WindowFunctionParams {
+                                    ref partition_by,
+                                    ref order_by,
+                                    ..
+                                } = &window_fun.as_ref().params;
+                                generate_sort_key(partition_by, order_by)
+                            }
                             _ => unreachable!(),
                         }
                     }
@@ -1506,17 +1504,18 @@ pub fn create_window_expr_with_name(
     let name = name.into();
     let physical_schema: &Schema = &logical_schema.into();
     match e {
-        Expr::WindowFunction(WindowFunction {
-            fun,
-            params:
-                WindowFunctionParams {
-                    args,
-                    partition_by,
-                    order_by,
-                    window_frame,
-                    null_treatment,
-                },
-        }) => {
+        Expr::WindowFunction(window_fun) => {
+            let WindowFunction {
+                fun,
+                params:
+                    WindowFunctionParams {
+                        args,
+                        partition_by,
+                        order_by,
+                        window_frame,
+                        null_treatment,
+                    },
+            } = window_fun.as_ref();
             let physical_args =
                 create_physical_exprs(args, logical_schema, execution_props)?;
             let partition_by =

datafusion/core/tests/dataframe/mod.rs

@@ -907,7 +907,7 @@ async fn window_using_aggregates() -> Result<()> {
             vec![col("c3")],
         );
 
-        Expr::WindowFunction(w)
+        Expr::from(w)
             .null_treatment(NullTreatment::IgnoreNulls)
             .order_by(vec![col("c2").sort(true, true), col("c3").sort(true, true)])
             .window_frame(WindowFrame::new_bounds(

datafusion/core/tests/user_defined/user_defined_aggregates.rs

@@ -1106,22 +1106,22 @@ async fn test_metadata_based_aggregate_as_window() -> Result<()> {
         )));
 
     let df = df.select(vec![
-        Expr::WindowFunction(WindowFunction::new(
+        Expr::from(WindowFunction::new(
             WindowFunctionDefinition::AggregateUDF(Arc::clone(&no_output_meta_udf)),
             vec![col("no_metadata")],
         ))
         .alias("meta_no_in_no_out"),
-        Expr::WindowFunction(WindowFunction::new(
+        Expr::from(WindowFunction::new(
             WindowFunctionDefinition::AggregateUDF(no_output_meta_udf),
             vec![col("with_metadata")],
         ))
         .alias("meta_with_in_no_out"),
-        Expr::WindowFunction(WindowFunction::new(
+        Expr::from(WindowFunction::new(
             WindowFunctionDefinition::AggregateUDF(Arc::clone(&with_output_meta_udf)),
             vec![col("no_metadata")],
         ))
         .alias("meta_no_in_with_out"),
-        Expr::WindowFunction(WindowFunction::new(
+        Expr::from(WindowFunction::new(
             WindowFunctionDefinition::AggregateUDF(with_output_meta_udf),
             vec![col("with_metadata")],
         ))

datafusion/datasource/src/file_scan_config.rs

@@ -2391,7 +2391,7 @@ mod tests {
         // Setup sort expression
         let exec_props = ExecutionProps::new();
         let df_schema = DFSchema::try_from_qualified_schema("test", schema.as_ref())?;
-        let sort_expr = vec![col("value").sort(true, false)];
+        let sort_expr = [col("value").sort(true, false)];
 
         let physical_sort_exprs: Vec<_> = sort_expr
             .iter()