fix: Test bug2

src/query/service/src/api/rpc/exchange/exchange_manager.rs

@@ -712,12 +712,20 @@ impl QueryCoordinator {
         }
 
         let ctx = query_ctx.clone();
+        let ctx_clone = ctx.clone();
         let (_, request_server_exchange) = request_server_exchanges.into_iter().next().unwrap();
         let mut statistics_sender =
             StatisticsSender::spawn_sender(&query_id, ctx, request_server_exchange);
 
         Thread::named_spawn(Some(String::from("Distributed-Executor")), move || {
-            statistics_sender.shutdown(executor.execute().err());
+            let res = executor.execute();
+            if res.is_err() {
+                println!(
+                    "read two fileds at node:{}",
+                    ctx_clone.get_cluster().local_id
+                );
+            }
+            statistics_sender.shutdown(res.err());
             query_ctx
                 .get_exchange_manager()
                 .on_finished_query(&query_id);

src/query/service/src/api/rpc/exchange/statistics_sender.rs

@@ -111,6 +111,8 @@ impl StatisticsSender {
                     "Cannot send data via flight exchange, cause: {:?}",
                     error_code
                 );
+                tracing::error!("send a error!");
+                println!("send a error")
             }
 
             shutdown_flag_sender.close();

src/query/service/src/interpreters/interpreter.rs

@@ -97,8 +97,13 @@ pub trait Interpreter: Sync + Send {
             let complete_executor = PipelineCompleteExecutor::from_pipelines(pipelines, settings)?;
 
             ctx.set_executor(complete_executor.get_inner())?;
-            complete_executor.execute()?;
-            Ok(Box::pin(DataBlockStream::create(None, vec![])))
+            match complete_executor.execute() {
+                Err(error) => {
+                    println!("read two fileds at node:{}", ctx.get_cluster().local_id);
+                    return Err(error);
+                }
+                Ok(_) => return Ok(Box::pin(DataBlockStream::create(None, vec![]))),
+            }
         } else {
             let pulling_executor = PipelinePullingExecutor::from_pipelines(build_res, settings)?;
 

src/query/service/src/schedulers/scheduler.rs

@@ -17,6 +17,7 @@ use std::sync::Arc;
 use common_exception::ErrorCode;
 use common_exception::Result;
 use common_profile::SharedProcessorProfiles;
+use tracing::error;
 
 use crate::pipelines::PipelineBuildResult;
 use crate::pipelines::PipelineBuilder;
@@ -98,7 +99,10 @@ pub async fn build_distributed_pipeline(
     enable_profiling: bool,
 ) -> Result<PipelineBuildResult> {
     let fragmenter = Fragmenter::try_create(ctx.clone())?;
-
+    error!(
+        "Build distributed pipeline at: {}",
+        ctx.get_cluster().local_id
+    );
     let root_fragment = fragmenter.build_fragment(plan)?;
     let mut fragments_actions = QueryFragmentsActions::create(ctx.clone(), enable_profiling);
     root_fragment.get_actions(ctx.clone(), &mut fragments_actions)?;

src/query/storages/stage/src/stage_table.rs

@@ -45,6 +45,7 @@ use common_storage::StageFileInfo;
 use dashmap::DashMap;
 use opendal::Operator;
 use parking_lot::Mutex;
+use tracing::error;
 
 use crate::parquet_file::append_data_to_parquet_files;
 use crate::row_based_file::append_data_to_row_based_files;
@@ -172,7 +173,16 @@ impl Table for StageTable {
                 splits.push(Arc::new(split.clone()));
             }
         }
-
+        let mut file_names = String::from("");
+        for file in &splits {
+            file_names.push_str(&file.file.path);
+            file_names.push_str(",");
+        }
+        error!(
+            "read files at node id {}: {}",
+            ctx.get_cluster().local_id,
+            file_names
+        );
         //  Build copy pipeline.
         let settings = ctx.get_settings();
         let fields = stage_table_info
@@ -206,7 +216,7 @@ impl Table for StageTable {
             self.table_info.is_select,
             projection,
         )?);
-        tracing::debug!("start copy splits feeder in {}", ctx.get_cluster().local_id);
+        tracing::info!("start copy splits feeder in {}", ctx.get_cluster().local_id);
         input_ctx.format.exec_copy(input_ctx.clone(), pipeline)?;
         Ok(())
     }

tests/sqllogictests/suites/mode/cluster/distributed_copy_into_table2_execption.test

@@ -55,10 +55,10 @@ copy into products from @s1 pattern = '.*[.]csv' purge = true;
 query I
 select count(*) from products;
 ----
-0
+8
 
 statement ok
 select block_count from fuse_snapshot('default','products');
 
 statement ok
-set enable_distributed_copy_into = 0;
+set enable_distributed_copy_into = 0;