update

Author: SkyFan2002

src/query/service/src/interpreters/interpreter_append.rs

@@ -75,7 +75,12 @@ impl Interpreter for AppendInterpreter {
         let append: Append = match &self.s_expr.plan() {
             RelOperator::Append(append) => append.clone(),
             RelOperator::Exchange(_) => self.s_expr.child(0).unwrap().plan().clone().try_into()?,
-            _ => unreachable!(),
+            plan => {
+                return Err(ErrorCode::Internal(format!(
+                    "AppendInterpreter: unexpected plan type: {:?}",
+                    plan
+                )));
+            }
         };
         let (target_table, catalog, database, table) = {
             let metadata = self.metadata.read();

src/query/sql/src/planner/optimizer/mod.rs

@@ -43,6 +43,7 @@ pub use hyper_dp::DPhpy;
 pub use m_expr::MExpr;
 pub use memo::Memo;
 pub use optimizer::optimize;
+pub use optimizer::optimize_append;
 pub use optimizer::optimize_query;
 pub use optimizer::OptimizerContext;
 pub use optimizer::RecursiveOptimizer;

src/query/sql/src/planner/optimizer/optimizer.rs

@@ -295,44 +295,17 @@ pub async fn optimize(mut opt_ctx: OptimizerContext, plan: Plan) -> Result<Plan>
             overwrite,
             forbid_occ_retry,
         } => {
-            let support_distributed_insert = {
-                let append: Append = s_expr.plan().clone().try_into()?;
-                let metadata = metadata.read();
-                metadata
-                    .table(append.table_index)
-                    .table()
-                    .support_distributed_insert()
-            };
-            let enable_distributed = opt_ctx.enable_distributed_optimization
-                && opt_ctx
-                    .table_ctx
-                    .get_settings()
-                    .get_enable_distributed_copy()?
-                && support_distributed_insert;
-            info!(
-                "after optimization enable_distributed_copy? : {}",
-                enable_distributed
-            );
-            let mut optimized_source =
-                optimize_query(&mut opt_ctx, s_expr.child(0)?.clone()).await?;
-            let optimized = match enable_distributed {
-                true => {
-                    if let RelOperator::Exchange(Exchange::Merge) = optimized_source.plan.as_ref() {
-                        optimized_source = optimized_source.child(0).unwrap().clone();
-                    }
-                    let copy_into = SExpr::create_unary(
-                        Arc::new(s_expr.plan().clone()),
-                        Arc::new(optimized_source),
-                    );
-                    let exchange = Arc::new(RelOperator::Exchange(Exchange::Merge));
-                    SExpr::create_unary(exchange, Arc::new(copy_into))
-                }
-                false => {
-                    SExpr::create_unary(Arc::new(s_expr.plan().clone()), Arc::new(optimized_source))
-                }
-            };
+            let append: Append = s_expr.plan().clone().try_into()?;
+            let source = s_expr.child(0)?.clone();
+            let optimized_source = optimize_query(&mut opt_ctx, source).await?;
+            let optimized_append = optimize_append(
+                append,
+                optimized_source,
+                metadata.clone(),
+                opt_ctx.table_ctx.as_ref(),
+            )?;
             Ok(Plan::Append {
-                s_expr: Box::new(optimized),
+                s_expr: Box::new(optimized_append),
                 metadata,
                 stage_table_info,
                 overwrite,
@@ -610,3 +583,34 @@ async fn optimize_mutation(mut opt_ctx: OptimizerContext, s_expr: SExpr) -> Resu
         metadata: opt_ctx.metadata.clone(),
     })
 }
+
+pub fn optimize_append(
+    append: Append,
+    source: SExpr,
+    metadata: MetadataRef,
+    table_ctx: &dyn TableContext,
+) -> Result<SExpr> {
+    let support_distributed_insert = {
+        let metadata = metadata.read();
+        metadata
+            .table(append.table_index)
+            .table()
+            .support_distributed_insert()
+    };
+    let enable_distributed = table_ctx.get_settings().get_enable_distributed_copy()?
+        && support_distributed_insert
+        && matches!(source.plan(), RelOperator::Exchange(Exchange::Merge));
+    info!(
+        "after optimization enable_distributed_copy? : {}",
+        enable_distributed
+    );
+    match enable_distributed {
+        true => {
+            let source = source.child(0).unwrap().clone();
+            let copy_into = SExpr::create_unary(Arc::new(append.into()), Arc::new(source));
+            let exchange = Arc::new(RelOperator::Exchange(Exchange::Merge));
+            Ok(SExpr::create_unary(exchange, Arc::new(copy_into)))
+        }
+        false => Ok(SExpr::create_unary(Arc::new(append.into()), Arc::new(source))),
+    }
+}

src/query/sql/src/planner/plans/append.rs

@@ -44,6 +44,7 @@ use crate::executor::physical_plans::PhysicalAppend;
 use crate::executor::PhysicalPlan;
 use crate::executor::PhysicalPlanBuilder;
 use crate::optimizer::optimize;
+use crate::optimizer::optimize_append;
 use crate::optimizer::OptimizerContext;
 use crate::optimizer::SExpr;
 use crate::ColumnBinding;
@@ -111,9 +112,10 @@ pub async fn create_append_plan_from_subquery(
         append_type: AppendType::Insert,
     };
 
-    let s_expr = SExpr::create_unary(Arc::new(insert_plan.into()), Arc::new(source));
+    let optimized_append = optimize_append(insert_plan, source, metadata.clone(), ctx.as_ref())?;
+
     let plan = Plan::Append {
-        s_expr: Box::new(s_expr),
+        s_expr: Box::new(optimized_append),
         metadata: metadata.clone(),
         stage_table_info: None,
         overwrite,