feat: broadcast the runtime filter to the global equivalence classes.

src/query/catalog/src/runtime_filter_info.rs

@@ -60,7 +60,7 @@ impl RuntimeFilterInfo {
 pub struct RuntimeFilterEntry {
     pub id: usize,
     pub probe_expr: Expr<String>,
-    pub bloom: Option<RuntimeFilterBloom>,
+    pub bloom: Option<Arc<RuntimeFilterBloom>>,
     pub inlist: Option<Expr<String>>,
     pub min_max: Option<Expr<String>>,
     pub stats: Arc<RuntimeFilterStats>,
@@ -69,7 +69,6 @@ pub struct RuntimeFilterEntry {
     pub enabled: bool,
 }
 
-#[derive(Clone)]
 pub struct RuntimeFilterBloom {
     pub column_name: String,
     pub filter: RuntimeBloomFilter,

src/query/catalog/src/table_context.rs

@@ -359,6 +359,18 @@ pub trait TableContext: Send + Sync {
 
     fn get_runtime_filter_ready(&self, table_index: usize) -> Vec<Arc<RuntimeFilterReady>>;
 
+    /// Set the pushed runtime filter statistics for a scan_id
+    /// Parameters: scan_id, selectivity (0.0-1.0), row count
+    fn set_pushed_runtime_filter_stats(&self, _scan_id: usize, _selectivity: f64, _rows: u64) {
+        unimplemented!()
+    }
+
+    /// Get the pushed runtime filter statistics for a scan_id
+    /// Returns: Option<(selectivity, rows)>
+    fn get_pushed_runtime_filter_stats(&self, _scan_id: usize) -> Option<(f64, u64)> {
+        unimplemented!()
+    }
+
     fn clear_runtime_filter(&self);
     fn assert_no_runtime_filter_state(&self) -> Result<()> {
         unimplemented!()

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

@@ -44,6 +44,7 @@ use databend_common_sql::ScalarExpr;
 use databend_common_sql::TypeCheck;
 use tokio::sync::Barrier;
 
+use super::runtime_filter::supported_join_type_for_runtime_filter;
 use super::runtime_filter::PhysicalRuntimeFilters;
 use super::PhysicalPlanCast;
 use crate::physical_plans::explain::PlanStatsInfo;
@@ -75,6 +76,7 @@ type JoinConditionsResult = (
     Vec<Option<(RemoteExpr<String>, usize, usize, IndexType)>>,
     Vec<((usize, bool), usize)>,
     Vec<Option<IndexType>>,
+    Vec<ScalarExpr>, // Build key ScalarExprs for equivalence class checking
 );
 
 type ProjectionsResult = (
@@ -508,16 +510,29 @@ impl HashJoin {
 }
 
 impl PhysicalPlanBuilder {
-    /// Builds the physical plans for both sides of the join
     pub async fn build_join_sides(
         &mut self,
         s_expr: &SExpr,
+        join: Option<&Join>,
         left_required: ColumnSet,
         right_required: ColumnSet,
     ) -> Result<(PhysicalPlan, PhysicalPlan)> {
         let probe_side = self.build(s_expr.left_child(), left_required).await?;
+
+        let should_track = join
+            .map(|j| supported_join_type_for_runtime_filter(&j.join_type))
+            .unwrap_or(false);
+
+        if should_track {
+            self.runtime_filter_anchors.push(Arc::new(s_expr.clone()));
+        }
+
         let build_side = self.build(s_expr.right_child(), right_required).await?;
 
+        if should_track {
+            self.runtime_filter_anchors.pop();
+        }
+
         Ok((probe_side, build_side))
     }
 
@@ -788,6 +803,7 @@ impl PhysicalPlanBuilder {
     ) -> Result<JoinConditionsResult> {
         let mut left_join_conditions = Vec::new();
         let mut right_join_conditions = Vec::new();
+        let mut right_join_conditions_scalar = Vec::new();
         let mut is_null_equal = Vec::new();
         let mut left_join_conditions_rt = Vec::new();
         let mut probe_to_build_index = Vec::new();
@@ -890,6 +906,7 @@ impl PhysicalPlanBuilder {
             // Add to result collections
             left_join_conditions.push(left_expr.as_remote_expr());
             right_join_conditions.push(right_expr.as_remote_expr());
+            right_join_conditions_scalar.push(right_condition.clone());
             is_null_equal.push(condition.is_null_equal);
             left_join_conditions_rt.push(left_expr_for_runtime_filter.map(
                 |(expr, scan_id, table_index, column_idx)| {
@@ -906,6 +923,7 @@ impl PhysicalPlanBuilder {
             left_join_conditions_rt,
             probe_to_build_index,
             build_table_indexes,
+            right_join_conditions_scalar,
         ))
     }
 
@@ -1270,7 +1288,7 @@ impl PhysicalPlanBuilder {
     ) -> Result<PhysicalPlan> {
         // Step 1: Build probe and build sides
         let (mut probe_side, mut build_side) = self
-            .build_join_sides(s_expr, left_required, right_required)
+            .build_join_sides(s_expr, Some(join), left_required, right_required)
             .await?;
 
         // Step 2: Prepare column projections
@@ -1292,6 +1310,7 @@ impl PhysicalPlanBuilder {
             left_join_conditions_rt,
             mut probe_to_build_index,
             build_table_indexes,
+            right_join_conditions_scalar,
         ) = self.process_equi_conditions(
             join,
             &probe_schema,
@@ -1337,8 +1356,11 @@ impl PhysicalPlanBuilder {
             join,
             s_expr,
             &right_join_conditions,
+            &right_join_conditions_scalar,
             left_join_conditions_rt,
             build_table_indexes,
+            &self.runtime_filter_anchors,
+            &mut self.join_equivalence_classes,
         )
         .await?;
 

src/query/service/src/physical_plans/physical_plan_builder.rs

@@ -31,6 +31,7 @@ use databend_storages_common_table_meta::meta::TableSnapshot;
 
 use crate::physical_plans::explain::PlanStatsInfo;
 use crate::physical_plans::physical_plan::PhysicalPlan;
+use crate::physical_plans::runtime_filter::JoinEquivalenceClasses;
 
 pub struct PhysicalPlanBuilder {
     pub metadata: MetadataRef,
@@ -39,6 +40,9 @@ pub struct PhysicalPlanBuilder {
     pub dry_run: bool,
     // DataMutation info, used to build MergeInto physical plan
     pub mutation_build_info: Option<MutationBuildInfo>,
+    pub runtime_filter_anchors: Vec<Arc<SExpr>>,
+    pub join_equivalence_classes: JoinEquivalenceClasses,
+    pub inited: bool,
 }
 
 impl PhysicalPlanBuilder {
@@ -50,6 +54,9 @@ impl PhysicalPlanBuilder {
             func_ctx,
             dry_run,
             mutation_build_info: None,
+            runtime_filter_anchors: Vec::new(),
+            join_equivalence_classes: JoinEquivalenceClasses::default(),
+            inited: false,
         }
     }
 
@@ -63,6 +70,10 @@ impl PhysicalPlanBuilder {
     }
 
     pub async fn build(&mut self, s_expr: &SExpr, required: ColumnSet) -> Result<PhysicalPlan> {
+        if !self.inited {
+            self.join_equivalence_classes = JoinEquivalenceClasses::build_from_sexpr(s_expr);
+            self.inited = true;
+        }
         let mut plan = self.build_physical_plan(s_expr, required).await?;
         plan.adjust_plan_id(&mut 0);
 

src/query/service/src/physical_plans/physical_range_join.rs

@@ -226,7 +226,7 @@ impl PhysicalPlanBuilder {
 
         // Construct IEJoin
         let (right_side, left_side) = self
-            .build_join_sides(s_expr, left_required, right_required)
+            .build_join_sides(s_expr, None, left_required, right_required)
             .await?;
 
         let left_schema = self.prepare_probe_schema(join_type, &left_side)?;