chore(query): continue the vacuum query hook even aborted (#17497)

Author: sundy-li

src/query/ee/src/storages/fuse/operations/vacuum_temporary_files.rs

@@ -173,8 +173,10 @@ async fn vacuum_by_duration(
     Ok(removed_total)
 }
 
+// Vacuum temporary files by query hook
+// If query was killed, we still need to clean up the temporary files
 async fn vacuum_query_hook(
-    abort_checker: AbortChecker,
+    _abort_checker: AbortChecker,
     temporary_dir: &str,
     nodes: &[usize],
     query_id: &str,
@@ -198,7 +200,6 @@ async fn vacuum_query_hook(
         .filter_map(|x| x.is_ok().then(|| x.unwrap()));
 
     for (meta_file_path, buffer) in metas {
-        abort_checker.try_check_aborting()?;
         let removed = vacuum_by_meta_buffer(
             &meta_file_path,
             temporary_dir,

src/query/service/src/interpreters/hook/vacuum_hook.rs

@@ -45,11 +45,12 @@ pub fn hook_vacuum_temp_files(query_ctx: &Arc<QueryContext>) -> Result<()> {
 
         let cluster = query_ctx.get_cluster();
         let query_id = query_ctx.get_id();
-
+        let abort_checker = query_ctx.clone().get_abort_checker();
         let mut node_files = HashMap::new();
         for node in cluster.nodes.iter() {
             let stats = query_ctx.get_spill_file_stats(Some(node.id.clone()));
-            if stats.file_nums != 0 {
+            // if query was aborted, we can't trust the stats
+            if stats.file_nums != 0 || abort_checker.try_check_aborting().is_err() {
                 if let Some(index) = cluster.index_of_nodeid(&node.id) {
                     node_files.insert(index, stats.file_nums);
                 }
@@ -65,7 +66,6 @@ pub fn hook_vacuum_temp_files(query_ctx: &Arc<QueryContext>) -> Result<()> {
         );
 
         let nodes = node_files.keys().cloned().collect::<Vec<usize>>();
-        let abort_checker = query_ctx.clone().get_abort_checker();
         let _ = GlobalIORuntime::instance().block_on::<(), ErrorCode, _>(async move {
             let removed_files = handler
                 .do_vacuum_temporary_files(

src/query/service/src/sessions/query_ctx.rs

@@ -401,6 +401,7 @@ impl QueryContext {
     pub fn get_spill_file_stats(&self, node_id: Option<String>) -> SpillProgress {
         let r = self.shared.cluster_spill_progress.read();
         let node_id = node_id.unwrap_or(self.get_cluster().local_id());
+
         r.get(&node_id).cloned().unwrap_or(SpillProgress::default())
     }