Fix: should render result set at last of select.

Author: RinChanNOWWW

src/query/service/src/interpreters/interpreter_insert_v2.rs

@@ -204,12 +204,15 @@ impl InsertInterpreterV2 {
         catalog: String,
         table: Arc<dyn Table>,
     ) -> Result<SendableDataBlockStream> {
-        let inner_plan = match plan {
+        let (inner_plan, select_column_bindings) = match plan {
             Plan::Query {
-                s_expr, metadata, ..
+                s_expr,
+                metadata,
+                bind_context,
+                ..
             } => {
                 let builder = PhysicalPlanBuilder::new(metadata.clone(), self.ctx.clone());
-                builder.build(s_expr).await?
+                (builder.build(s_expr).await?, bind_context.columns.clone())
             }
             _ => unreachable!(),
         };
@@ -222,6 +225,7 @@ impl InsertInterpreterV2 {
                 catalog,
                 table_info: table.get_table_info().clone(),
                 select_schema: plan.schema(),
+                select_column_bindings,
                 insert_schema: self.plan.schema(),
                 cast_needed: self.check_schema_cast(plan)?,
             }));

src/query/service/src/sql/executor/physical_plan.rs

@@ -34,6 +34,7 @@ use super::AggregateFunctionDesc;
 use super::SortDesc;
 use crate::sql::optimizer::ColumnSet;
 use crate::sql::plans::JoinType;
+use crate::sql::ColumnBinding;
 use crate::sql::IndexType;
 
 pub type ColumnID = String;
@@ -312,6 +313,7 @@ pub struct DistributedInsertSelect {
     pub table_info: TableInfo,
     pub insert_schema: DataSchemaRef,
     pub select_schema: DataSchemaRef,
+    pub select_column_bindings: Vec<ColumnBinding>,
     pub cast_needed: bool,
 }
 

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

@@ -187,6 +187,7 @@ pub trait PhysicalPlanReplacer {
                 table_info: plan.table_info.clone(),
                 select_schema: plan.select_schema.clone(),
                 insert_schema: plan.insert_schema.clone(),
+                select_column_bindings: plan.select_column_bindings.clone(),
                 cast_needed: plan.cast_needed,
             },
         )))

src/query/service/src/sql/executor/pipeline_builder.rs

@@ -540,10 +540,17 @@ impl PipelineBuilder {
         let insert_schema = &insert_select.insert_schema;
 
         self.build_pipeline(&insert_select.input)?;
+
+        // should render result for select
+        PipelineBuilder::render_result_set(
+            insert_select.input.output_schema()?,
+            &insert_select.select_column_bindings,
+            &mut self.main_pipeline,
+        )?;
+
         if insert_select.cast_needed {
             let mut functions = Vec::with_capacity(insert_schema.fields().len());
-            for (target_field, original_field) in insert_select
-                .insert_schema
+            for (target_field, original_field) in insert_schema
                 .fields()
                 .iter()
                 .zip(select_schema.fields().iter())

src/query/service/src/sql/planner/binder/bind_context.rs

@@ -34,7 +34,7 @@ use crate::sql::optimizer::SExpr;
 use crate::sql::plans::Scalar;
 use crate::sql::NameResolutionContext;
 
-#[derive(Clone, Debug, Eq, PartialEq)]
+#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
 pub enum Visibility {
     // Default for a column
     Visible,
@@ -46,7 +46,7 @@ pub enum Visibility {
     UnqualifiedWildcardInVisible,
 }
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
 pub struct ColumnBinding {
     /// Database name of this `ColumnBinding` in current context
     pub database_name: Option<String>,