chore: use Arc::unwrap_or_clone in more places

Author: Dandandan

datafusion/core/src/execution/context/mod.rs

@@ -950,13 +950,13 @@ impl SessionContext {
         match (or_replace, view) {
             (true, Ok(_)) => {
                 self.deregister_table(name.clone())?;
-                let input = Self::apply_type_coercion(input.as_ref().clone())?;
+                let input = Self::apply_type_coercion(Arc::unwrap_or_clone(input))?;
                 let table = Arc::new(ViewTable::new(input, definition));
                 self.register_table(name, table)?;
                 self.return_empty_dataframe()
             }
             (_, Err(_)) => {
-                let input = Self::apply_type_coercion(input.as_ref().clone())?;
+                let input = Self::apply_type_coercion(Arc::unwrap_or_clone(input))?;
                 let table = Arc::new(ViewTable::new(input, definition));
                 self.register_table(name, table)?;
                 self.return_empty_dataframe()

datafusion/datasource-arrow/src/file_format.rs

@@ -165,7 +165,7 @@ impl FileFormat for ArrowFormat {
                 }
                 GetResultPayload::Stream(stream) => infer_stream_schema(stream).await?,
             };
-            schemas.push(schema.as_ref().clone());
+            schemas.push(Arc::unwrap_or_clone(schema));
         }
         let merged_schema = Schema::try_merge(schemas)?;
         Ok(Arc::new(merged_schema))

datafusion/expr/src/logical_plan/plan.rs

@@ -3625,7 +3625,6 @@ impl Aggregate {
     ///
     /// This method should only be called when you are absolutely sure that the schema being
     /// provided is correct for the aggregate. If in doubt, call [try_new](Self::try_new) instead.
-    #[expect(clippy::needless_pass_by_value)]
     pub fn try_new_with_schema(
         input: Arc<LogicalPlan>,
         group_expr: Vec<Expr>,
@@ -3651,7 +3650,7 @@ impl Aggregate {
 
         let aggregate_func_dependencies =
             calc_func_dependencies_for_aggregate(&group_expr, &input, &schema)?;
-        let new_schema = schema.as_ref().clone();
+        let new_schema = Arc::unwrap_or_clone(schema);
         let schema = Arc::new(
             new_schema.with_functional_dependencies(aggregate_func_dependencies)?,
         );

datafusion/ffi/src/udf/mod.rs

@@ -391,7 +391,7 @@ impl ScalarUDFImpl for ForeignScalarUDF {
             .map(WrappedSchema)
             .collect::<RVec<_>>();
 
-        let return_field = return_field.as_ref().clone();
+        let return_field = Arc::unwrap_or_clone(return_field);
         let return_field = WrappedSchema(FFI_ArrowSchema::try_from(return_field)?);
         let config_options = config_options.as_ref().into();
 

datafusion/optimizer/src/extract_leaf_expressions.rs

@@ -247,9 +247,7 @@ fn extract_from_plan(
                 Some(plan) => Ok(plan),
                 // No extractions for this input — recover the LogicalPlan
                 // without cloning (refcount is 1 since build returned None).
-                None => {
-                    Ok(Arc::try_unwrap(input_arc).unwrap_or_else(|arc| (*arc).clone()))
-                }
+                None => Ok(Arc::unwrap_or_clone(input_arc)),
             }
         })
         .collect::<Result<Vec<_>>>()?;

datafusion/optimizer/src/replace_distinct_aggregate.rs

@@ -109,7 +109,7 @@ impl OptimizerRule for ReplaceDistinctWithAggregate {
                             .enumerate()
                             .all(|(idx, f_idx)| idx == *f_idx)
                     {
-                        return Ok(Transformed::yes(input.as_ref().clone()));
+                        return Ok(Transformed::yes(Arc::unwrap_or_clone(input)));
                     }
                 }
 

datafusion/physical-plan/src/joins/hash_join/exec.rs

@@ -1450,8 +1450,8 @@ impl ExecutionPlan for HashJoinExec {
                 let right_stats = self.right.partition_statistics(Some(partition))?;
 
                 estimate_join_statistics(
-                    (*left_stats).clone(),
-                    (*right_stats).clone(),
+                    Arc::unwrap_or_clone(left_stats),
+                    Arc::unwrap_or_clone(right_stats),
                     &self.on,
                     &self.join_type,
                     &self.join_schema,
@@ -1465,8 +1465,8 @@ impl ExecutionPlan for HashJoinExec {
                 let right_stats = self.right.partition_statistics(Some(partition))?;
 
                 estimate_join_statistics(
-                    (*left_stats).clone(),
-                    (*right_stats).clone(),
+                    Arc::unwrap_or_clone(left_stats),
+                    Arc::unwrap_or_clone(right_stats),
                     &self.on,
                     &self.join_type,
                     &self.join_schema,
@@ -1482,8 +1482,8 @@ impl ExecutionPlan for HashJoinExec {
                 let left_stats = self.left.partition_statistics(None)?;
                 let right_stats = self.right.partition_statistics(None)?;
                 estimate_join_statistics(
-                    (*left_stats).clone(),
-                    (*right_stats).clone(),
+                    Arc::unwrap_or_clone(left_stats),
+                    Arc::unwrap_or_clone(right_stats),
                     &self.on,
                     &self.join_type,
                     &self.join_schema,

datafusion/sql/src/relation/mod.rs

@@ -389,7 +389,7 @@ fn optimize_subquery_sort(
             LogicalPlan::Sort(s) => {
                 if !has_limit {
                     has_limit = false;
-                    return Ok(Transformed::yes(s.input.as_ref().clone()));
+                    return Ok(Transformed::yes(Arc::unwrap_or_clone(s.input)));
                 }
                 Ok(Transformed::no(LogicalPlan::Sort(s)))
             }

datafusion/substrait/src/logical_plan/consumer/rel/aggregate_rel.rs

@@ -106,7 +106,7 @@ pub async fn from_aggregate_rel(
                     not_impl_err!("Aggregate without aggregate function is not supported")
                 }
             };
-            aggr_exprs.push(agg_func?.as_ref().clone());
+            aggr_exprs.push(std::sync::Arc::unwrap_or_clone(agg_func?));
         }
 
         // Ensure that all expressions have a unique name