Replace logical plan from Arc<T> to Box<T>

datafusion-examples/examples/dataframe_subquery.rs

@@ -16,7 +16,6 @@
 // under the License.
 
 use arrow_schema::DataType;
-use std::sync::Arc;
 
 use datafusion::error::Result;
 use datafusion::prelude::*;
@@ -44,7 +43,7 @@ async fn where_scalar_subquery(ctx: &SessionContext) -> Result<()> {
     ctx.table("t1")
         .await?
         .filter(
-            scalar_subquery(Arc::new(
+            scalar_subquery(Box::new(
                 ctx.table("t2")
                     .await?
                     .filter(out_ref_col(DataType::Utf8, "t1.c1").eq(col("t2.c1")))?
@@ -67,7 +66,7 @@ async fn where_in_subquery(ctx: &SessionContext) -> Result<()> {
         .await?
         .filter(in_subquery(
             col("t1.c2"),
-            Arc::new(
+            Box::new(
                 ctx.table("t2")
                     .await?
                     .filter(col("t2.c1").gt(lit(ScalarValue::UInt8(Some(0)))))?
@@ -87,7 +86,7 @@ async fn where_in_subquery(ctx: &SessionContext) -> Result<()> {
 async fn where_exist_subquery(ctx: &SessionContext) -> Result<()> {
     ctx.table("t1")
         .await?
-        .filter(exists(Arc::new(
+        .filter(exists(Box::new(
             ctx.table("t2")
                 .await?
                 .filter(out_ref_col(DataType::Utf8, "t1.c1").eq(col("t2.c1")))?

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

@@ -529,8 +529,7 @@ impl SessionContext {
             column_defaults,
         } = cmd;
 
-        let input = Arc::try_unwrap(input).unwrap_or_else(|e| e.as_ref().clone());
-        let input = self.state().optimize(&input)?;
+        let input = self.state().optimize(input.as_ref())?;
         let table = self.table(&name).await;
         match (if_not_exists, or_replace, table) {
             (true, false, Ok(_)) => self.return_empty_dataframe(),
@@ -1886,7 +1885,7 @@ impl SessionState {
                 },
             );
             let (plan, logical_optimization_succeeded) = match optimized_plan {
-                Ok(plan) => (Arc::new(plan), true),
+                Ok(plan) => (Box::new(plan), true),
                 Err(DataFusionError::Context(optimizer_name, err)) => {
                     let plan_type = PlanType::OptimizedLogicalPlan { optimizer_name };
                     stringified_plans

datafusion/core/src/physical_planner.rs

@@ -919,7 +919,7 @@ impl DefaultPhysicalPlanner {
                     Ok(Arc::new(filter.with_default_selectivity(selectivity)?))
                 }
                 LogicalPlan::Union(Union { inputs, schema: _ }) => {
-                    let physical_plans = self.create_initial_plan_multi(inputs.iter().map(|lp| lp.as_ref()), session_state).await?;
+                    let physical_plans = self.create_initial_plan_multi(inputs.iter(), session_state).await?;
 
                     Ok(Arc::new(UnionExec::new(physical_plans)))
                 }
@@ -1020,8 +1020,8 @@ impl DefaultPhysicalPlanner {
                         let join_plan =
                             LogicalPlan::Join(Join::try_new_with_project_input(
                                 logical_plan,
-                                Arc::new(left),
-                                Arc::new(right),
+                                Box::new(left),
+                                Box::new(right),
                                 column_on,
                             )?);
 
@@ -1037,7 +1037,7 @@ impl DefaultPhysicalPlanner {
                             let projection =
                                 Projection::try_new(
                                     final_join_result,
-                                    Arc::new(join_plan),
+                                    Box::new(join_plan),
                                 )?;
                             LogicalPlan::Projection(projection)
                         } else {

datafusion/core/tests/dataframe/mod.rs

@@ -102,7 +102,7 @@ async fn test_count_wildcard_on_where_in() -> Result<()> {
         .await?
         .filter(in_subquery(
             col("a"),
-            Arc::new(
+            Box::new(
                 ctx.table("t2")
                     .await?
                     .aggregate(vec![], vec![count(wildcard())])?
@@ -139,7 +139,7 @@ async fn test_count_wildcard_on_where_exist() -> Result<()> {
     let df_results = ctx
         .table("t1")
         .await?
-        .filter(exists(Arc::new(
+        .filter(exists(Box::new(
             ctx.table("t2")
                 .await?
                 .aggregate(vec![], vec![count(wildcard())])?
@@ -251,7 +251,7 @@ async fn test_count_wildcard_on_where_scalar_subquery() -> Result<()> {
         .table("t1")
         .await?
         .filter(
-            scalar_subquery(Arc::new(
+            scalar_subquery(Box::new(
                 ctx.table("t2")
                     .await?
                     .filter(out_ref_col(DataType::UInt32, "t1.a").eq(col("t2.a")))?

datafusion/core/tests/user_defined/user_defined_table_functions.rs

@@ -158,7 +158,7 @@ impl SimpleCsvTable {
                 normalize_col(self.exprs[0].clone(), &plan)?,
                 plan.schema(),
             )],
-            Arc::new(plan),
+            Box::new(plan),
         )
         .map(LogicalPlan::Projection)?;
         let rbs = collect(

datafusion/expr/src/expr_fn.rs

@@ -377,7 +377,7 @@ pub fn approx_percentile_cont_with_weight(
 }
 
 /// Create an EXISTS subquery expression
-pub fn exists(subquery: Arc<LogicalPlan>) -> Expr {
+pub fn exists(subquery: Box<LogicalPlan>) -> Expr {
     let outer_ref_columns = subquery.all_out_ref_exprs();
     Expr::Exists(Exists {
         subquery: Subquery {
@@ -389,7 +389,7 @@ pub fn exists(subquery: Arc<LogicalPlan>) -> Expr {
 }
 
 /// Create a NOT EXISTS subquery expression
-pub fn not_exists(subquery: Arc<LogicalPlan>) -> Expr {
+pub fn not_exists(subquery: Box<LogicalPlan>) -> Expr {
     let outer_ref_columns = subquery.all_out_ref_exprs();
     Expr::Exists(Exists {
         subquery: Subquery {
@@ -401,7 +401,7 @@ pub fn not_exists(subquery: Arc<LogicalPlan>) -> Expr {
 }
 
 /// Create an IN subquery expression
-pub fn in_subquery(expr: Expr, subquery: Arc<LogicalPlan>) -> Expr {
+pub fn in_subquery(expr: Expr, subquery: Box<LogicalPlan>) -> Expr {
     let outer_ref_columns = subquery.all_out_ref_exprs();
     Expr::InSubquery(InSubquery::new(
         Box::new(expr),
@@ -414,7 +414,7 @@ pub fn in_subquery(expr: Expr, subquery: Arc<LogicalPlan>) -> Expr {
 }
 
 /// Create a NOT IN subquery expression
-pub fn not_in_subquery(expr: Expr, subquery: Arc<LogicalPlan>) -> Expr {
+pub fn not_in_subquery(expr: Expr, subquery: Box<LogicalPlan>) -> Expr {
     let outer_ref_columns = subquery.all_out_ref_exprs();
     Expr::InSubquery(InSubquery::new(
         Box::new(expr),
@@ -427,7 +427,7 @@ pub fn not_in_subquery(expr: Expr, subquery: Arc<LogicalPlan>) -> Expr {
 }
 
 /// Create a scalar subquery expression
-pub fn scalar_subquery(subquery: Arc<LogicalPlan>) -> Expr {
+pub fn scalar_subquery(subquery: Box<LogicalPlan>) -> Expr {
     let outer_ref_columns = subquery.all_out_ref_exprs();
     Expr::ScalarSubquery(Subquery {
         subquery,

datafusion/expr/src/expr_rewriter/mod.rs

@@ -19,7 +19,6 @@
 
 use std::collections::HashMap;
 use std::collections::HashSet;
-use std::sync::Arc;
 
 use crate::expr::{Alias, Unnest};
 use crate::logical_plan::Projection;
@@ -220,7 +219,7 @@ pub fn coerce_plan_expr_for_schema(
             let new_exprs = coerce_exprs_for_schema(exprs, plan.schema(), schema)?;
             let add_project = new_exprs.iter().any(|expr| expr.try_into_col().is_err());
             if add_project {
-                let projection = Projection::try_new(new_exprs, Arc::new(plan.clone()))?;
+                let projection = Projection::try_new(new_exprs, Box::new(plan.clone()))?;
                 Ok(LogicalPlan::Projection(projection))
             } else {
                 Ok(plan.clone())

datafusion/expr/src/expr_schema.rs

@@ -32,7 +32,6 @@ use datafusion_common::{
     ExprSchema, Result,
 };
 use std::collections::HashMap;
-use std::sync::Arc;
 
 /// trait to allow expr to typable with respect to a schema
 pub trait ExprSchemable {
@@ -544,7 +543,7 @@ pub fn cast_subquery(subquery: Subquery, cast_to_type: &DataType) -> Result<Subq
         }
     };
     Ok(Subquery {
-        subquery: Arc::new(new_plan),
+        subquery: Box::new(new_plan),
         outer_ref_columns: subquery.outer_ref_columns,
     })
 }