refactor(interpreter): try fix test failure

Author: zhang2014

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

@@ -68,7 +68,6 @@ pub fn append2table(
     build_res: &mut PipelineBuildResult,
     overwrite: bool,
     need_commit: bool,
-    need_output: bool,
 ) -> Result<()> {
     fill_missing_columns(
         ctx.clone(),
@@ -77,7 +76,7 @@ pub fn append2table(
         &mut build_res.main_pipeline,
     )?;
 
-    table.append2(ctx.clone(), &mut build_res.main_pipeline, need_output)?;
+    table.append2(ctx.clone(), &mut build_res.main_pipeline, false)?;
 
     if need_commit {
         build_res.main_pipeline.set_on_finished(move |may_error| {

src/query/service/src/interpreters/interpreter_copy_v2.rs

@@ -275,7 +275,6 @@ impl CopyInterpreterV2 {
             &mut build_res,
             false,
             true,
-            false,
         )?;
         Ok(build_res)
     }

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

@@ -164,7 +164,6 @@ impl Interpreter for InsertInterpreter {
             &mut build_res,
             self.plan.overwrite,
             true,
-            false,
         )?;
 
         Ok(build_res)

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

@@ -15,6 +15,8 @@
 use std::collections::VecDeque;
 use std::sync::Arc;
 
+use common_base::base::GlobalIORuntime;
+use common_base::base::TrySpawn;
 use common_exception::ErrorCode;
 use common_exception::Result;
 use parking_lot::Mutex;
@@ -173,27 +175,47 @@ impl Interpreter for InsertInterpreterV2 {
                         }
                     };
 
-                    build_res = match is_distributed_plan {
+                    let mut build_res = match is_distributed_plan {
                         true => {
-                            build_schedule_pipepline(self.ctx.clone(), &insert_select_plan).await?
+                            build_schedule_pipepline(self.ctx.clone(), &insert_select_plan).await
+                        }
+                        false => {
+                            PipelineBuilder::create(self.ctx.clone()).finalize(&insert_select_plan)
+                        }
+                    }?;
+
+                    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
+                                ))),
+                            };
                         }
-                        false => PipelineBuilder::create(self.ctx.clone())
-                            .finalize(&insert_select_plan)?,
-                    };
 
-                    if is_distributed_plan {
-                        append2table(
-                            self.ctx.clone(),
-                            table.clone(),
-                            plan.schema(),
-                            &mut build_res,
-                            self.plan.overwrite,
-                            true,
-                            true,
-                        )?;
-
-                        return Ok(build_res);
-                    }
+                        Err(may_error.as_ref().unwrap().clone())
+                    });
+
+                    return Ok(build_res);
                 }
             };
         }
@@ -205,7 +227,6 @@ impl Interpreter for InsertInterpreterV2 {
             &mut build_res,
             self.plan.overwrite,
             true,
-            false,
         )?;
 
         Ok(build_res)

src/query/service/src/servers/mysql/writers/query_result_writer.rs

@@ -106,6 +106,18 @@ impl<'a, W: AsyncWrite + Send + Unpin> DFQueryResultWriter<'a, W> {
         // XXX: num_columns == 0 may is error?
 
         if !query_result.has_result_set {
+            // For statements without result sets, we still need to pull the stream because errors may occur in the stream.
+            let blocks = &mut query_result.blocks;
+            while let Some(block) = blocks.next().await {
+                if let Err(e) = block {
+                    dataset_writer
+                        .error(ErrorKind::ER_UNKNOWN_ERROR, &e.to_string().as_bytes())
+                        .await?;
+
+                    return Ok(());
+                }
+            }
+
             dataset_writer.completed(OkResponse::default()).await?;
             return Ok(());
         }

src/query/service/tests/it/storages/fuse/table_test_fixture.rs

@@ -296,7 +296,6 @@ impl TestFixture {
             &mut build_res,
             overwrite,
             commit,
-            false,
         )?;
 
         execute_pipeline(self.ctx.clone(), build_res)