fix schema

Author: SkyFan2002

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

@@ -65,15 +65,15 @@ impl Interpreter for AppendInterpreter {
     #[fastrace::trace]
     #[async_backtrace::framed]
     async fn execute2(&self) -> Result<PipelineBuildResult> {
-        debug!("ctx.id" = self.ctx.get_id().as_str(); "copy_into_table_interpreter_execute_v2");
+        debug!("ctx.id" = self.ctx.get_id().as_str(); "append_interpreter_execute");
         if check_deduplicate_label(self.ctx.clone()).await? {
             return Ok(PipelineBuildResult::create());
         }
 
-        let copy_into_table: Append = self.s_expr.plan().clone().try_into()?;
+        let append: Append = self.s_expr.plan().clone().try_into()?;
         let (target_table, catalog, database, table) = {
             let metadata = self.metadata.read();
-            let t = metadata.table(copy_into_table.table_index);
+            let t = metadata.table(append.table_index);
             (
                 t.table(),
                 t.catalog().to_string(),
@@ -162,8 +162,8 @@ impl Interpreter for AppendInterpreter {
     }
 
     fn inject_result(&self) -> Result<SendableDataBlockStream> {
-        let copy_into_table: Append = self.s_expr.plan().clone().try_into()?;
-        match &copy_into_table.append_type {
+        let append: Append = self.s_expr.plan().clone().try_into()?;
+        match &append.append_type {
             AppendType::CopyInto => {
                 let blocks = self.get_copy_into_table_result()?;
                 Ok(Box::pin(DataBlockStream::create(None, blocks)))

src/query/service/src/pipelines/builders/builder_append.rs

@@ -41,7 +41,7 @@ use crate::pipelines::PipelineBuilder;
 use crate::sessions::QueryContext;
 
 impl PipelineBuilder {
-    pub(crate) fn build_copy_into_table(&mut self, copy: &PhysicalAppend) -> Result<()> {
+    pub(crate) fn build_append(&mut self, copy: &PhysicalAppend) -> Result<()> {
         let to_table = self.ctx.build_table_by_table_info(&copy.table_info, None)?;
         self.ctx
             .set_read_block_thresholds(to_table.get_block_thresholds());

src/query/service/src/pipelines/pipeline_builder.rs

@@ -195,7 +195,7 @@ impl PipelineBuilder {
             }
 
             // Copy into.
-            PhysicalPlan::Append(copy) => self.build_copy_into_table(copy),
+            PhysicalPlan::Append(append) => self.build_append(append),
             PhysicalPlan::CopyIntoLocation(copy) => self.build_copy_into_location(copy),
 
             // Replace.

src/query/service/src/schedulers/fragments/fragmenter.rs

@@ -178,7 +178,7 @@ impl PhysicalPlanReplacer for Fragmenter {
         })))
     }
 
-    fn replace_copy_into_table(&mut self, plan: &PhysicalAppend) -> Result<PhysicalPlan> {
+    fn replace_append(&mut self, plan: &PhysicalAppend) -> Result<PhysicalPlan> {
         let input = self.replace(&plan.input)?;
         Ok(PhysicalPlan::Append(Box::new(PhysicalAppend {
             plan_id: plan.plan_id,

src/query/service/src/schedulers/fragments/plan_fragment.rs

@@ -535,7 +535,7 @@ impl PhysicalPlanReplacer for ReplaceReadSource {
         }))
     }
 
-    fn replace_copy_into_table(&mut self, plan: &PhysicalAppend) -> Result<PhysicalPlan> {
+    fn replace_append(&mut self, plan: &PhysicalAppend) -> Result<PhysicalPlan> {
         let input = self.replace(&plan.input)?;
         Ok(PhysicalPlan::Append(Box::new(PhysicalAppend {
             plan_id: plan.plan_id,

src/query/sql/src/executor/physical_plan_builder.rs

@@ -142,9 +142,7 @@ impl PhysicalPlanBuilder {
             }
             RelOperator::Recluster(recluster) => self.build_recluster(recluster).await,
             RelOperator::CompactBlock(compact) => self.build_compact_block(compact).await,
-            RelOperator::Append(copy_into_table) => {
-                self.build_copy_into_table(s_expr, copy_into_table).await
-            }
+            RelOperator::Append(append) => self.build_append(s_expr, append).await,
             RelOperator::ValueScan(value_scan) => self.build_value_scan(value_scan).await,
         }
     }

src/query/sql/src/executor/physical_plan_visitor.rs

@@ -92,7 +92,7 @@ pub trait PhysicalPlanReplacer {
             PhysicalPlan::CompactSource(plan) => self.replace_compact_source(plan),
             PhysicalPlan::CommitSink(plan) => self.replace_commit_sink(plan),
             PhysicalPlan::RangeJoin(plan) => self.replace_range_join(plan),
-            PhysicalPlan::Append(plan) => self.replace_copy_into_table(plan),
+            PhysicalPlan::Append(plan) => self.replace_append(plan),
             PhysicalPlan::CopyIntoLocation(plan) => self.replace_copy_into_location(plan),
             PhysicalPlan::ReplaceAsyncSourcer(plan) => self.replace_async_sourcer(plan),
             PhysicalPlan::ReplaceDeduplicate(plan) => self.replace_deduplicate(plan),
@@ -402,7 +402,7 @@ pub trait PhysicalPlanReplacer {
         }))
     }
 
-    fn replace_copy_into_table(&mut self, plan: &PhysicalAppend) -> Result<PhysicalPlan> {
+    fn replace_append(&mut self, plan: &PhysicalAppend) -> Result<PhysicalPlan> {
         let input = self.replace(&plan.input)?;
 
         Ok(PhysicalPlan::Append(Box::new(PhysicalAppend {

src/query/sql/src/planner/binder/copy_into_table.rs

@@ -257,7 +257,7 @@ impl<'a> Binder {
     async fn bind_copy_into_table_from_location(
         &mut self,
         bind_ctx: &BindContext,
-        copy_into_table_plan: Append,
+        mut copy_into_table_plan: Append,
         stage_table_info: StageTableInfo,
     ) -> Result<Plan> {
         let use_query = matches!(&stage_table_info.stage_info.file_format_params,
@@ -304,7 +304,7 @@ impl<'a> Binder {
             )
             .await
         } else {
-            let (scan, _) = self
+            let (scan, bind_context) = self
                 .bind_stage_table(
                     self.ctx.clone(),
                     bind_ctx,
@@ -314,6 +314,7 @@ impl<'a> Binder {
                     stage_table_info.files_to_copy.clone(),
                 )
                 .await?;
+            copy_into_table_plan.project_columns = Some(bind_context.columns.clone());
 
             let copy_into =
                 SExpr::create_unary(Arc::new(copy_into_table_plan.into()), Arc::new(scan));

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

@@ -301,14 +301,19 @@ impl Operator for Append {
 }
 
 impl PhysicalPlanBuilder {
-    pub async fn build_copy_into_table(
+    pub async fn build_append(
         &mut self,
         s_expr: &SExpr,
         plan: &crate::plans::Append,
     ) -> Result<PhysicalPlan> {
         let target_table = self.metadata.read().table(plan.table_index).table();
 
-        let source = self.build(s_expr.child(0)?, Default::default()).await?;
+        let column_set = plan
+            .project_columns
+            .as_ref()
+            .map(|project_columns| project_columns.iter().map(|c| c.index).collect())
+            .unwrap_or_default();
+        let source = self.build(s_expr.child(0)?, column_set).await?;
 
         Ok(PhysicalPlan::Append(Box::new(PhysicalAppend {
             plan_id: 0,
@@ -317,7 +322,7 @@ impl PhysicalPlanBuilder {
             values_consts: plan.values_consts.clone(),
             required_source_schema: plan.required_source_schema.clone(),
             table_info: target_table.get_table_info().clone(),
-            project_columns: None,
+            project_columns: plan.project_columns.clone(),
         })))
     }
 }