fix(executor): fix queries executor cannot kill problem and reactivate kill test assertion (#15443)

* fix(executor): fix queries executor cluster cannot kill problem and reactivate kill test assert

Signed-off-by: Liuqing Yue <dqhl76@gmail.com>

* fix: add finish notify in graph

Signed-off-by: Liuqing Yue <dqhl76@gmail.com>

* disable queries executor, ensure not affect

Signed-off-by: Liuqing Yue <dqhl76@gmail.com>

---------

Signed-off-by: Liuqing Yue <dqhl76@gmail.com>

Author: dqhl76

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

@@ -21,6 +21,7 @@ use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
 
+use databend_common_base::base::WatchNotify;
 use databend_common_base::runtime::error_info::NodeErrorType;
 use databend_common_base::runtime::profile::Profile;
 use databend_common_base::runtime::profile::ProfileStatisticsName;
@@ -191,6 +192,7 @@ struct ExecutingGraph {
     points: AtomicU64,
     query_id: Arc<String>,
     should_finish: AtomicBool,
+    finished_notify: Arc<WatchNotify>,
     finish_condvar_notify: Option<Arc<(Mutex<bool>, Condvar)>>,
     finished_error: Mutex<Option<ErrorCode>>,
 }
@@ -212,6 +214,7 @@ impl ExecutingGraph {
             points: AtomicU64::new((MAX_POINTS << 32) | init_epoch as u64),
             query_id,
             should_finish: AtomicBool::new(false),
+            finished_notify: Arc::new(WatchNotify::new()),
             finish_condvar_notify,
             finished_error: Mutex::new(None),
         })
@@ -235,6 +238,7 @@ impl ExecutingGraph {
             points: AtomicU64::new((MAX_POINTS << 32) | init_epoch as u64),
             query_id,
             should_finish: AtomicBool::new(false),
+            finished_notify: Arc::new(WatchNotify::new()),
             finish_condvar_notify,
             finished_error: Mutex::new(None),
         })
@@ -793,6 +797,7 @@ impl RunningGraph {
             return Ok(());
         }
         self.0.should_finish.store(true, Ordering::SeqCst);
+        self.0.finished_notify.notify_waiters();
         self.interrupt_running_nodes();
         let mut finished_error = self.0.finished_error.lock();
         if finished_error.is_none() {
@@ -836,6 +841,10 @@ impl RunningGraph {
         self.0.points.load(Ordering::SeqCst)
     }
 
+    pub fn get_finished_notify(&self) -> Arc<WatchNotify> {
+        self.0.finished_notify.clone()
+    }
+
     pub fn format_graph_nodes(&self) -> String {
         pub struct NodeDisplay {
             id: usize,

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

@@ -78,6 +78,10 @@ impl ExecutorTasksQueue {
             }
         }
     }
+
+    pub fn is_queries_executor(&self) -> bool {
+        matches!(self, ExecutorTasksQueue::QueriesExecutorTasksQueue(_))
+    }
 }
 
 pub struct ProcessorAsyncTask {
@@ -101,7 +105,11 @@ impl ProcessorAsyncTask {
         graph: Arc<RunningGraph>,
         inner: Inner,
     ) -> ProcessorAsyncTask {
-        let finished_notify = queue.get_finished_notify();
+        let finished_notify = if queue.is_queries_executor() {
+            graph.get_finished_notify()
+        } else {
+            queue.get_finished_notify()
+        };
 
         let inner = async move {
             let left = Box::pin(inner);

tests/suites/1_stateful/02_query/02_0000_kill_query.py

@@ -37,12 +37,11 @@
     res = mycursor.fetchone()
     kill_query = "kill query " + str(res[0]) + ";"
     mycursor.execute(kill_query)
-    time.sleep(0.1)
+    time.sleep(0.5)
     mycursor.execute(
         "SELECT * FROM system.processes WHERE extra_info LIKE '%SELECT max(number)%' AND extra_info NOT LIKE '%system.processes%';"
     )
     res = mycursor.fetchone()
 
-    ## TODO NEW EXPRESSION
-    ## assert res is None
+    assert res is None
     client1.expect(prompt)