fix

Author: SkyFan2002

src/query/catalog/src/runtime_filter_info.rs

@@ -74,7 +74,7 @@ pub struct RuntimeFilterBloom {
 
 #[derive(Default)]
 pub struct RuntimeFilterStats {
-    _filter_id: usize,
+    pub filter_id: usize,
     bloom_time_ns: AtomicU64,
     bloom_rows_filtered: AtomicU64,
     inlist_min_max_time_ns: AtomicU64,
@@ -83,8 +83,15 @@ pub struct RuntimeFilterStats {
 }
 
 impl RuntimeFilterStats {
-    pub fn new() -> Self {
-        Self::default()
+    pub fn new(filter_id: usize) -> Self {
+        Self {
+            filter_id,
+            bloom_time_ns: AtomicU64::new(0),
+            bloom_rows_filtered: AtomicU64::new(0),
+            inlist_min_max_time_ns: AtomicU64::new(0),
+            min_max_rows_filtered: AtomicU64::new(0),
+            min_max_partitions_pruned: AtomicU64::new(0),
+        }
     }
 
     pub fn record_bloom(&self, time_ns: u64, rows_filtered: u64) {

src/query/service/src/physical_plans/format/format_table_scan.rs

@@ -156,9 +156,9 @@ impl<'a> PhysicalFormat for TableScanFormatter<'a> {
                             vec![FormatTreeNode::new(format!("type: [{}]", type_text))];
 
                         if let Some(ref filter_stats) = stats {
-                            // Find the matching stat by filter id
-                            // For now, we just use the first one as stats are grouped by scan_id
-                            if let Some(stat) = filter_stats.first() {
+                            if let Some(stat) =
+                                filter_stats.iter().find(|stat| stat.filter_id == filter.id)
+                            {
                                 detail_children.push(FormatTreeNode::new(format!(
                                     "bloom rows filtered: {}",
                                     stat.bloom_rows_filtered()

src/query/service/src/physical_plans/physical_hash_join.rs

@@ -59,7 +59,6 @@ use crate::pipelines::processors::transforms::memory::outer_left_join::OuterLeft
 use crate::pipelines::processors::transforms::BasicHashJoinState;
 use crate::pipelines::processors::transforms::HashJoinProbeState;
 use crate::pipelines::processors::transforms::InnerHashJoin;
-use crate::pipelines::processors::transforms::RuntimeFilterContext;
 use crate::pipelines::processors::transforms::TransformHashJoin;
 use crate::pipelines::processors::transforms::TransformHashJoinBuild;
 use crate::pipelines::processors::transforms::TransformHashJoinProbe;
@@ -284,7 +283,7 @@ impl IPhysicalPlan for HashJoin {
         }
 
         // Create the join state with optimization flags
-        let state = self.build_state(builder)?;
+        let state = self.build_state(builder, desc)?;
 
         if let Some((build_cache_index, _)) = self.build_side_cache_info {
             builder.hash_join_states.insert(
@@ -306,13 +305,17 @@ impl IPhysicalPlan for HashJoin {
 }
 
 impl HashJoin {
-    fn build_state(&self, builder: &mut PipelineBuilder) -> Result<Arc<HashJoinState>> {
+    fn build_state(
+        &self,
+        builder: &mut PipelineBuilder,
+        desc: Arc<HashJoinDesc>,
+    ) -> Result<Arc<HashJoinState>> {
         let (enable_optimization, is_distributed) = builder.merge_into_get_optimization_flag(self);
         HashJoinState::try_create(
             builder.ctx.clone(),
             self.build.output_schema()?,
             &self.build_projections,
-            HashJoinDesc::create(&builder.ctx, builder.func_ctx.clone(), self)?,
+            desc.clone(),
             &self.probe_to_build,
             is_distributed,
             enable_optimization,
@@ -408,14 +411,6 @@ impl HashJoin {
         desc: Arc<HashJoinDesc>,
     ) -> Result<()> {
         let state = Arc::new(BasicHashJoinState::create());
-        // We must build the runtime filter before constructing the child nodes,
-        // as we will inject some runtime filter information into the context for the child nodes to use.
-        let rf_desc = RuntimeFilterContext::create(
-            &builder.ctx,
-            builder.func_ctx.clone(),
-            &self.runtime_filter,
-            self.broadcast_id,
-        )?;
 
         if let Some((build_cache_index, _)) = self.build_side_cache_info {
             builder.hash_join_states.insert(
@@ -461,7 +456,7 @@ impl HashJoin {
                 self.create_join(&self.join_type, builder, desc.clone(), state.clone())?,
                 stage_sync_barrier.clone(),
                 self.projections.clone(),
-                Arc::new(rf_desc.clone()),
+                Arc::new(desc.runtime_filter.clone()),
             );
 
             join_pipe_items.push(PipeItem::create(

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

@@ -111,7 +111,7 @@ pub struct HashJoinState {
     /// A dummy receiver to make build done watcher channel open
     pub(crate) _build_done_dummy_receiver: Receiver<HashTableType>,
     /// Some description of hash join. Such as join type, join keys, etc.
-    pub(crate) hash_join_desc: HashJoinDesc,
+    pub(crate) hash_join_desc: Arc<HashJoinDesc>,
     /// Interrupt the build phase or probe phase.
     pub(crate) interrupt: AtomicBool,
     /// If there is no data in build side, maybe we can fast return.
@@ -156,7 +156,7 @@ impl HashJoinState {
         ctx: Arc<QueryContext>,
         mut build_schema: DataSchemaRef,
         build_projections: &ColumnSet,
-        hash_join_desc: HashJoinDesc,
+        hash_join_desc: Arc<HashJoinDesc>,
         probe_to_build: &[(usize, (bool, bool))],
         merge_into_is_distributed: bool,
         enable_merge_into_optimization: bool,

src/query/service/src/pipelines/processors/transforms/runtime_filter/context.rs

@@ -141,6 +141,8 @@ impl RuntimeFilterContext {
         let runtime_filter_infos =
             super::convert::build_runtime_filter_infos(packet, runtime_filter_descs)?;
 
+        self.ctx
+            .register_runtime_filter_stats(&runtime_filter_infos);
         self.channels.publish_all(&runtime_filter_infos)?;
 
         let merge_time = merge_start.elapsed();

src/query/service/src/pipelines/processors/transforms/runtime_filter/convert.rs

@@ -87,7 +87,7 @@ pub fn build_runtime_filter_infos(
                 } else {
                     None
                 },
-                stats: Arc::new(RuntimeFilterStats::new()),
+                stats: Arc::new(RuntimeFilterStats::new(desc.id)),
             };
 
             if let Some(existing) = entry