Reduce cloning in LogicalPlanBuilder

 - Migrate function arguments from `LogicalPlan` to `impl Into<Arc<LogicalPlan>>`
 - Migrage function arguments from `&Schema` to `impl Into<SchemaRef>`
 - Update usages (mostly in tests)

Author: joroKr21

datafusion/core/src/physical_planner.rs

@@ -974,17 +974,15 @@ impl DefaultPhysicalPlanner {
                     let (left, left_col_keys, left_projected) =
                         wrap_projection_for_join_if_necessary(
                             &left_keys,
-                            original_left.as_ref().clone(),
+                            Arc::clone(original_left),
                         )?;
                     let (right, right_col_keys, right_projected) =
                         wrap_projection_for_join_if_necessary(
                             &right_keys,
-                            original_right.as_ref().clone(),
+                            Arc::clone(original_right),
                         )?;
-                    let column_on = (left_col_keys, right_col_keys);
 
-                    let left = Arc::new(left);
-                    let right = Arc::new(right);
+                    let column_on = (left_col_keys, right_col_keys);
                     let (new_join, requalified) = Join::try_new_with_project_input(
                         node,
                         Arc::clone(&left),

datafusion/core/tests/dataframe/dataframe_functions.rs

@@ -1294,7 +1294,7 @@ async fn test_count_wildcard() -> Result<()> {
         Field::new("c", DataType::UInt32, false),
     ]);
 
-    let table_scan = table_scan(Some("test"), &schema, None)?.build()?;
+    let table_scan = table_scan(Some("test"), schema, None)?.build()?;
     let plan = LogicalPlanBuilder::from(table_scan)
         .aggregate(vec![col("b")], vec![count_all()])
         .unwrap()

datafusion/core/tests/expr_api/simplification.rs

@@ -107,7 +107,7 @@ fn test_table_scan() -> LogicalPlan {
         Field::new("d", DataType::UInt32, false),
         Field::new("e", DataType::UInt32, true),
     ]);
-    table_scan(Some("test"), &schema, None)
+    table_scan(Some("test"), schema, None)
         .expect("creating scan")
         .build()
         .expect("building plan")
@@ -324,7 +324,7 @@ fn select_date_plus_interval() -> Result<()> {
 fn simplify_project_scalar_fn() -> Result<()> {
     // Issue https://github.com/apache/datafusion/issues/5996
     let schema = Schema::new(vec![Field::new("f", DataType::Float64, false)]);
-    let plan = table_scan(Some("test"), &schema, None)?
+    let plan = table_scan(Some("test"), schema, None)?
         .project(vec![power(col("f"), lit(1.0))])?
         .build()?;
 
@@ -350,7 +350,7 @@ fn simplify_scan_predicate() -> Result<()> {
     ]);
     let plan = table_scan_with_filters(
         Some("test"),
-        &schema,
+        schema,
         None,
         vec![col("g").eq(power(col("f"), lit(1.0)))],
     )?

datafusion/expr/src/expr_rewriter/mod.rs

@@ -36,15 +36,15 @@ pub use order_by::rewrite_sort_cols_by_aggs;
 
 /// Trait for rewriting [`Expr`]s into function calls.
 ///
-/// This trait is used with `FunctionRegistry::register_function_rewrite` to
+/// This trait is used with `FunctionRegistry::register_function_rewrite`
 /// to evaluating `Expr`s using functions that may not be built in to DataFusion
 ///
 /// For example, concatenating arrays `a || b` is represented as
 /// `Operator::ArrowAt`, but can be implemented by calling a function
 /// `array_concat` from the `functions-nested` crate.
 // This is not used in datafusion internally, but it is still helpful for downstream project so don't remove it.
 pub trait FunctionRewrite: Debug {
-    /// Return a human readable name for this rewrite
+    /// Return a human-readable name for this rewrite
     fn name(&self) -> &str;
 
     /// Potentially rewrite `expr` to some other expression
@@ -214,26 +214,29 @@ pub fn strip_outer_reference(expr: Expr) -> Expr {
 /// Returns plan with expressions coerced to types compatible with
 /// schema types
 pub fn coerce_plan_expr_for_schema(
-    plan: LogicalPlan,
+    plan: Arc<LogicalPlan>,
     schema: &DFSchema,
-) -> Result<LogicalPlan> {
-    match plan {
+) -> Result<Arc<LogicalPlan>> {
+    if matches!(plan.as_ref(), LogicalPlan::Projection(_)) {
         // special case Projection to avoid adding multiple projections
-        LogicalPlan::Projection(Projection { expr, input, .. }) => {
-            let new_exprs = coerce_exprs_for_schema(expr, input.schema(), schema)?;
-            let projection = Projection::try_new(new_exprs, input)?;
-            Ok(LogicalPlan::Projection(projection))
-        }
-        _ => {
-            let exprs: Vec<Expr> = plan.schema().iter().map(Expr::from).collect();
-            let new_exprs = coerce_exprs_for_schema(exprs, plan.schema(), schema)?;
-            let add_project = new_exprs.iter().any(|expr| expr.try_as_col().is_none());
-            if add_project {
-                let projection = Projection::try_new(new_exprs, Arc::new(plan))?;
-                Ok(LogicalPlan::Projection(projection))
-            } else {
-                Ok(plan)
-            }
+        let LogicalPlan::Projection(Projection { expr, input, .. }) =
+            Arc::unwrap_or_clone(plan)
+        else {
+            unreachable!()
+        };
+
+        let new_exprs = coerce_exprs_for_schema(expr, input.schema(), schema)?;
+        let projection = Projection::try_new(new_exprs, input)?;
+        Ok(Arc::new(LogicalPlan::Projection(projection)))
+    } else {
+        let exprs: Vec<Expr> = plan.schema().iter().map(Expr::from).collect();
+        let new_exprs = coerce_exprs_for_schema(exprs, plan.schema(), schema)?;
+        let add_project = new_exprs.iter().any(|expr| expr.try_as_col().is_none());
+        if add_project {
+            let projection = Projection::try_new(new_exprs, plan)?;
+            Ok(Arc::new(LogicalPlan::Projection(projection)))
+        } else {
+            Ok(plan)
         }
     }
 }
@@ -418,7 +421,7 @@ mod test {
     fn normalize_cols() {
         let expr = col("a") + col("b") + col("c");
 
-        // Schemas with some matching and some non matching cols
+        // Schemas with some matching and some non-matching cols
         let schema_a = make_schema_with_empty_metadata(
             vec![Some("tableA".into()), Some("tableA".into())],
             vec!["a", "aa"],