fix

Author: SkyFan2002

src/query/service/src/pipelines/processors/transforms/hash_join/hash_join_build_state.rs

@@ -840,24 +840,19 @@ impl HashJoinBuildState {
     }
 
     pub fn add_runtime_filter_ready(&self) {
-        if self.ctx.get_cluster().is_empty() {
-            return;
-        }
-
-        let mut wait_runtime_filter_table_indexes = HashSet::new();
+        let mut scan_ids = HashSet::new();
         for rf in self.runtime_filter_desc() {
-            // Collect scan_id from all probe targets (since one filter can be pushed to multiple scans)
             for (_probe_key, scan_id) in &rf.probe_targets {
-                wait_runtime_filter_table_indexes.insert(*scan_id);
+                scan_ids.insert(*scan_id);
             }
         }
 
         let build_state = unsafe { &mut *self.hash_join_state.build_state.get() };
         let runtime_filter_ready = &mut build_state.runtime_filter_ready;
-        for table_index in wait_runtime_filter_table_indexes.into_iter() {
+        for scan_id in scan_ids.into_iter() {
             let ready = Arc::new(RuntimeFilterReady::default());
             runtime_filter_ready.push(ready.clone());
-            self.ctx.set_runtime_filter_ready(table_index, ready);
+            self.ctx.set_runtime_filter_ready(scan_id, ready);
         }
     }
 

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

@@ -1548,9 +1548,9 @@ impl TableContext for QueryContext {
         }
     }
 
-    fn get_runtime_filter_ready(&self, table_index: usize) -> Vec<Arc<RuntimeFilterReady>> {
+    fn get_runtime_filter_ready(&self, scan_id: usize) -> Vec<Arc<RuntimeFilterReady>> {
         let runtime_filter_ready = self.shared.runtime_filter_ready.read();
-        match runtime_filter_ready.get(&table_index) {
+        match runtime_filter_ready.get(&scan_id) {
             Some(v) => v.to_vec(),
             None => vec![],
         }