refactor(interpreter): move commit into pipeline

Author: zhang2014

src/query/service/src/interpreters/interpreter_common.rs

@@ -61,13 +61,23 @@ pub fn append2table(
     }
 
     table.append2(ctx.clone(), &mut build_res.main_pipeline)?;
+
     let query_need_abort = ctx.query_need_abort();
     let executor_settings = ExecutorSettings::try_create(&ctx.get_settings())?;
     build_res.set_max_threads(ctx.get_settings().get_max_threads()? as usize);
     let mut pipelines = build_res.sources_pipelines;
     pipelines.push(build_res.main_pipeline);
-    let executor =
-        PipelineCompleteExecutor::from_pipelines(query_need_abort, pipelines, executor_settings)?;
+    let executor = PipelineCompleteExecutor::from_pipelines(query_need_abort, pipelines, executor_settings)?;
+    executor.execute()
+}
+
+pub fn execute_pipeline(ctx: Arc<QueryContext>, mut res: PipelineBuildResult) -> Result<()> {
+    let query_need_abort = ctx.query_need_abort();
+    let executor_settings = ExecutorSettings::try_create(&ctx.get_settings())?;
+    res.set_max_threads(ctx.get_settings().get_max_threads()? as usize);
+    let mut pipelines = res.sources_pipelines;
+    pipelines.push(res.main_pipeline);
+    let executor = PipelineCompleteExecutor::from_pipelines(query_need_abort, pipelines, executor_settings)?;
     executor.execute()
 }
 

src/query/service/src/interpreters/interpreter_insert.rs

@@ -26,13 +26,15 @@ use common_planners::SelectPlan;
 use common_streams::DataBlockStream;
 use common_streams::SendableDataBlockStream;
 use parking_lot::Mutex;
+use common_base::base::{GlobalIORuntime, TrySpawn};
 
 use super::commit2table;
 use super::interpreter_common::append2table;
 use crate::interpreters::Interpreter;
+use crate::interpreters::interpreter_common::execute_pipeline;
 use crate::interpreters::SelectInterpreter;
 use crate::pipelines::processors::port::OutputPort;
-use crate::pipelines::processors::BlocksSource;
+use crate::pipelines::processors::{BlocksSource, TransformAddOn};
 use crate::pipelines::processors::TransformCastSchema;
 use crate::pipelines::Pipeline;
 use crate::pipelines::PipelineBuildResult;
@@ -85,13 +87,21 @@ impl Interpreter for InsertInterpreter {
     }
 
     async fn execute(&self) -> Result<SendableDataBlockStream> {
+        let build_res = self.create_new_pipeline().await?;
+
+        execute_pipeline(self.ctx.clone(), build_res)?;
+        Ok(Box::pin(DataBlockStream::create(self.plan.schema(), None, vec![])))
+    }
+
+    async fn create_new_pipeline(&self) -> Result<PipelineBuildResult> {
         let plan = &self.plan;
         let settings = self.ctx.get_settings();
         let table = self
             .ctx
             .get_table(&plan.catalog, &plan.database, &plan.table)
             .await?;
-        let mut build_res = self.create_new_pipeline().await?;
+
+        let mut build_res = PipelineBuildResult::create();
         let mut builder = SourcePipeBuilder::create();
         if self.async_insert {
             build_res.main_pipeline.add_pipe(
@@ -160,21 +170,54 @@ impl Interpreter for InsertInterpreter {
             };
         }
 
-        append2table(self.ctx.clone(), table.clone(), plan.schema(), build_res)?;
-        commit2table(self.ctx.clone(), table.clone(), plan.overwrite).await?;
-        Ok(Box::pin(DataBlockStream::create(
-            self.plan.schema(),
-            None,
-            vec![],
-        )))
-    }
+        if table.schema() != plan.schema() {
+            build_res.main_pipeline
+                .add_transform(|transform_input_port, transform_output_port| {
+                    TransformAddOn::try_create(
+                        transform_input_port,
+                        transform_output_port,
+                        plan.schema().clone(),
+                        table.schema(),
+                        self.ctx.clone(),
+                    )
+                })?;
+        }
 
-    async fn create_new_pipeline(&self) -> Result<PipelineBuildResult> {
-        let insert_pipeline = Pipeline::create();
-        Ok(PipelineBuildResult {
-            main_pipeline: insert_pipeline,
-            sources_pipelines: vec![],
-        })
+        table.append2(self.ctx.clone(), &mut build_res.main_pipeline)?;
+
+        let ctx = self.ctx.clone();
+        let overwrite = self.plan.overwrite;
+
+        build_res.main_pipeline.set_on_finished(move |may_error| {
+            // capture out variable
+            let overwrite = overwrite;
+            let ctx = ctx.clone();
+            let table = table.clone();
+
+            if may_error.is_none() {
+                let append_entries = ctx.consume_precommit_blocks();
+                // We must put the commit operation to global runtime, which will avoid the "dispatch dropped without returning error" in tower
+                let catalog_name = ctx.get_current_catalog();
+                let commit_handle = GlobalIORuntime::instance().spawn(async move {
+                    table
+                        .commit_insertion(ctx, &catalog_name, append_entries, overwrite)
+                        .await
+                });
+
+                return match futures::executor::block_on(commit_handle) {
+                    Ok(Ok(_)) => Ok(()),
+                    Ok(Err(error)) => Err(error),
+                    Err(cause) => Err(ErrorCode::PanicError(format!(
+                        "Maybe panic while in commit insert. {}",
+                        cause
+                    )))
+                };
+            }
+
+            Err(may_error.as_ref().unwrap().clone())
+        });
+
+        Ok(build_res)
     }
 
     fn set_source_pipe_builder(&self, builder: Option<SourcePipeBuilder>) -> Result<()> {

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

@@ -61,7 +61,30 @@ impl InsertInterpreterV2 {
         }))
     }
 
-    async fn execute_new(&self) -> Result<SendableDataBlockStream> {
+    fn check_schema_cast(&self, plan: &Plan) -> common_exception::Result<bool> {
+        let output_schema = &self.plan.schema;
+        let select_schema = plan.schema();
+
+        // validate schema
+        if select_schema.fields().len() < output_schema.fields().len() {
+            return Err(ErrorCode::BadArguments(
+                "Fields in select statement is less than expected",
+            ));
+        }
+
+        // check if cast needed
+        let cast_needed = select_schema != *output_schema;
+        Ok(cast_needed)
+    }
+}
+
+#[async_trait::async_trait]
+impl Interpreter for InsertInterpreterV2 {
+    fn name(&self) -> &str {
+        "InsertIntoInterpreter"
+    }
+
+    async fn execute(&self) -> Result<SendableDataBlockStream> {
         let plan = &self.plan;
         let settings = self.ctx.get_settings();
         let table = self
@@ -162,33 +185,6 @@ impl InsertInterpreterV2 {
         )))
     }
 
-    fn check_schema_cast(&self, plan: &Plan) -> common_exception::Result<bool> {
-        let output_schema = &self.plan.schema;
-        let select_schema = plan.schema();
-
-        // validate schema
-        if select_schema.fields().len() < output_schema.fields().len() {
-            return Err(ErrorCode::BadArguments(
-                "Fields in select statement is less than expected",
-            ));
-        }
-
-        // check if cast needed
-        let cast_needed = select_schema != *output_schema;
-        Ok(cast_needed)
-    }
-}
-
-#[async_trait::async_trait]
-impl Interpreter for InsertInterpreterV2 {
-    fn name(&self) -> &str {
-        "InsertIntoInterpreter"
-    }
-
-    async fn execute(&self) -> Result<SendableDataBlockStream> {
-        self.execute_new().await
-    }
-
     async fn create_new_pipeline(&self) -> Result<PipelineBuildResult> {
         let insert_pipeline = Pipeline::create();
         Ok(PipelineBuildResult {

src/query/service/src/pipelines/pipeline_build_res.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::sync::Arc;
+use common_catalog::table::Table;
 use common_pipeline_core::Pipeline;
 
 pub struct PipelineBuildResult {
@@ -21,6 +23,13 @@ pub struct PipelineBuildResult {
 }
 
 impl PipelineBuildResult {
+    pub fn create() -> PipelineBuildResult {
+        PipelineBuildResult {
+            main_pipeline: Pipeline::create(),
+            sources_pipelines: vec![],
+        }
+    }
+
     pub fn set_max_threads(&mut self, max_threads: usize) {
         self.main_pipeline.set_max_threads(max_threads);