fix(processor): try fix incorrect error message

src/query/service/src/pipelines/executor/pipeline_executor.rs

@@ -25,6 +25,7 @@ use common_base::base::ThreadJoinHandle;
 use common_base::base::TrySpawn;
 use common_exception::ErrorCode;
 use common_exception::Result;
+use common_exception::ABORT_QUERY;
 use futures::future::select;
 use futures_util::future::Either;
 use parking_lot::Mutex;
@@ -144,7 +145,15 @@ impl PipelineExecutor {
     }
 
     pub fn finish(&self, cause: Option<ErrorCode>) {
-        *self.finished_error.lock() = cause;
+        if let Some(cause) = cause {
+            let mut finished_error = self.finished_error.lock();
+
+            // We only save the cause of the first error.
+            if finished_error.is_none() {
+                *finished_error = Some(cause);
+            }
+        }
+
         self.global_tasks_queue.finish(self.workers_condvar.clone());
         self.graph.interrupt_running_nodes();
         self.finished_notify.notify_waiters();
@@ -162,18 +171,21 @@ impl PipelineExecutor {
         let mut thread_join_handles = self.execute_threads(self.threads_num);
 
         while let Some(join_handle) = thread_join_handles.pop() {
-            if let Err(error_code) = join_handle.join().flatten() {
-                let may_error = Some(error_code);
-                (self.on_finished_callback)(&may_error)?;
-                return Err(may_error.unwrap());
+            let thread_res = join_handle.join().flatten();
+
+            {
+                let finished_error_guard = self.finished_error.lock();
+                if let Some(error) = finished_error_guard.as_ref() {
+                    let may_error = Some(error.clone());
+                    drop(finished_error_guard);
+                    (self.on_finished_callback)(&may_error)?;
+                    return Err(may_error.unwrap());
+                }
             }
-        }
 
-        {
-            let finished_error_guard = self.finished_error.lock();
-            if let Some(error) = finished_error_guard.as_ref() {
-                let may_error = Some(error.clone());
-                drop(finished_error_guard);
+            // We will ignore the abort query error, because returned by finished_error if abort query.
+            if matches!(&thread_res, Err(error) if error.code() != ABORT_QUERY) {
+                let may_error = Some(thread_res.unwrap_err());
                 (self.on_finished_callback)(&may_error)?;
                 return Err(may_error.unwrap());
             }

src/query/service/src/pipelines/executor/pipeline_pulling_executor.rs

@@ -24,8 +24,6 @@ use common_base::base::Thread;
 use common_datablocks::DataBlock;
 use common_exception::ErrorCode;
 use common_exception::Result;
-use common_exception::ABORT_QUERY;
-use common_exception::ABORT_SESSION;
 use parking_lot::Condvar;
 use parking_lot::Mutex;
 
@@ -186,29 +184,16 @@ impl PipelinePullingExecutor {
                     continue;
                 }
                 Err(_disconnected) => {
-                    let mut killed = false;
-
                     if !self.executor.is_finished() {
-                        killed = true;
                         self.executor.finish(None);
                     }
 
                     self.state.wait_finish();
 
-                    if self.state.is_catch_error() {
-                        let error_code = self.state.get_catch_error();
-
-                        // If the query is killed here, we should ignore the abort error.
-                        // when executing `select * from xx limit xx`, if enough rows have been returned, we try to kill the query in here for finish query as soon as possible.
-                        if killed
-                            && error_code.code() != ABORT_QUERY
-                            && error_code.code() != ABORT_SESSION
-                        {
-                            return Err(error_code);
-                        }
+                    match !self.state.is_catch_error() {
+                        true => Ok(None),
+                        false => Err(self.state.get_catch_error()),
                     }
-
-                    Ok(None)
                 }
             };
         }