update

Author: SkyFan2002

src/query/service/src/interpreters/interpreter_copy_into_table.rs

@@ -156,11 +156,18 @@ impl InterpreterFactory {
                 *graphical,
             )?)),
 
-            Plan::CopyIntoTable { .. } => todo!(),
-            // Ok(Arc::new(CopyIntoTableInterpreter::try_create(
-            //     ctx,
-            //     *copy_plan.clone(),
-            // )?)),
+            Plan::CopyIntoTable {
+                s_expr,
+                metadata,
+                stage_context,
+                overwrite,
+            } => Ok(Arc::new(CopyIntoTableInterpreter::try_create(
+                ctx,
+                *s_expr.clone(),
+                metadata.clone(),
+                stage_context.clone(),
+                *overwrite,
+            )?)),
             Plan::CopyIntoLocation(copy_plan) => Ok(Arc::new(
                 CopyIntoLocationInterpreter::try_create(ctx, copy_plan.clone())?,
             )),

src/query/service/src/interpreters/interpreter_factory.rs

@@ -50,7 +50,6 @@ use parking_lot::RwLock;
 
 use crate::interpreters::common::check_deduplicate_label;
 use crate::interpreters::common::dml_build_update_stream_req;
-use crate::interpreters::interpreter_copy_into_table::CopyIntoTableInterpreter;
 use crate::interpreters::HookOperator;
 use crate::interpreters::Interpreter;
 use crate::interpreters::InterpreterPtr;
@@ -375,7 +374,7 @@ impl ReplaceInterpreter {
         ctx: Arc<QueryContext>,
         source: &'a InsertInputSource,
         schema: DataSchemaRef,
-        purge_info: &mut Option<(Vec<StageFileInfo>, StageInfo)>,
+        _purge_info: &mut Option<(Vec<StageFileInfo>, StageInfo)>,
     ) -> Result<ReplaceSourceCtx> {
         match source {
             InsertInputSource::Values(source) => self
@@ -390,27 +389,29 @@ impl ReplaceInterpreter {
             InsertInputSource::SelectPlan(plan) => {
                 self.connect_query_plan_source(ctx.clone(), plan).await
             }
-            InsertInputSource::Stage(plan) => match *plan.clone() {
-                // Plan::CopyIntoTable(copy_plan) => {
-                //     let interpreter =
-                //         CopyIntoTableInterpreter::try_create(ctx.clone(), *copy_plan.clone())?;
-                //     let (physical_plan, _) = interpreter.build_physical_plan(&copy_plan).await?;
-
-                //     // TODO optimization: if copy_plan.stage_table_info.files_to_copy is None, there should be a short-cut plan
-
-                //     *purge_info = Some((
-                //         copy_plan.stage_table_info.files_to_copy.unwrap_or_default(),
-                //         copy_plan.stage_table_info.stage_info.clone(),
-                //     ));
-                //     Ok(ReplaceSourceCtx {
-                //         root: Box::new(physical_plan),
-                //         select_ctx: None,
-                //         update_stream_meta: vec![],
-                //         bind_context: None,
-                //     })
-                // }
-                _ => unreachable!("plan in InsertInputSource::Stag must be CopyIntoTable"),
-            },
+            _ => todo!(),
+            // InsertInputSource::Stage(plan) => match *plan.clone() {
+            //     Plan::CopyIntoTable(_copy_plan) => {
+            //         // let interpreter =
+            //         //     CopyIntoTableInterpreter::try_create(ctx.clone(), *copy_plan.clone())?;
+            //         // let (physical_plan, _) = interpreter.build_physical_plan(&copy_plan).await?;
+
+            //         // // TODO optimization: if copy_plan.stage_table_info.files_to_copy is None, there should be a short-cut plan
+
+            //         // *purge_info = Some((
+            //         //     copy_plan.stage_table_info.files_to_copy.unwrap_or_default(),
+            //         //     copy_plan.stage_table_info.stage_info.clone(),
+            //         // ));
+            //         // Ok(ReplaceSourceCtx {
+            //         //     root: Box::new(physical_plan),
+            //         //     select_ctx: None,
+            //         //     update_stream_meta: vec![],
+            //         //     bind_context: None,
+            //         // })
+            //         todo!()
+            //     }
+            //     _ => unreachable!("plan in InsertInputSource::Stag must be CopyIntoTable"),
+            // },
         }
     }