refactor: enables the bloom runtime filter to be turned on adaptively (#14686)

* feat: enables the bloom runtime filter to be turned on adaptively

* fix

* refactor

Author: xudong963

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

@@ -47,6 +47,8 @@ pub struct HashJoinDesc {
     pub(crate) probe_keys_rt: Vec<Option<(Expr<String>, IndexType)>>,
     // Under cluster, mark if the join is broadcast join.
     pub broadcast: bool,
+    // If enable bloom runtime filter
+    pub enable_bloom_runtime_filter: bool,
 }
 
 impl HashJoinDesc {
@@ -87,6 +89,7 @@ impl HashJoinDesc {
             probe_keys_rt,
             broadcast: join.broadcast,
             original_join_type: join.original_join_type.clone(),
+            enable_bloom_runtime_filter: join.enable_bloom_runtime_filter,
         })
     }
 

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

@@ -152,7 +152,9 @@ impl HashJoinBuildState {
             if !is_cluster || is_broadcast_join {
                 enable_inlist_runtime_filter = true;
                 enable_min_max_runtime_filter = true;
-                if ctx.get_settings().get_runtime_filter()? {
+                if ctx.get_settings().get_bloom_runtime_filter()?
+                    && hash_join_state.hash_join_desc.enable_bloom_runtime_filter
+                {
                     enable_bloom_runtime_filter = true;
                 }
             }

src/query/service/src/schedulers/fragments/fragmenter.rs

@@ -251,6 +251,7 @@ impl PhysicalPlanReplacer for Fragmenter {
             need_hold_hash_table: plan.need_hold_hash_table,
             stat_info: plan.stat_info.clone(),
             probe_keys_rt: plan.probe_keys_rt.clone(),
+            enable_bloom_runtime_filter: plan.enable_bloom_runtime_filter,
             broadcast: plan.broadcast,
             original_join_type: plan.original_join_type.clone(),
         }))

src/query/settings/src/settings_default.rs

@@ -265,8 +265,8 @@ impl DefaultSettings {
                     mode: SettingMode::Both,
                     range: None,
                 }),
-                ("enable_runtime_filter", DefaultSettingValue {
-                    value: UserSettingValue::UInt64(0),
+                ("enable_bloom_runtime_filter", DefaultSettingValue {
+                    value: UserSettingValue::UInt64(1),
                     desc: "Enables runtime filter optimization for JOIN.",
                     mode: SettingMode::Both,
                     range: Some(SettingRange::Numeric(0..=1)),

src/query/settings/src/settings_getter_setter.rs

@@ -269,8 +269,8 @@ impl Settings {
         Ok(self.try_get_u64("join_spilling_threshold")? as usize)
     }
 
-    pub fn get_runtime_filter(&self) -> Result<bool> {
-        Ok(self.try_get_u64("enable_runtime_filter")? != 0)
+    pub fn get_bloom_runtime_filter(&self) -> Result<bool> {
+        Ok(self.try_get_u64("enable_bloom_runtime_filter")? != 0)
     }
 
     pub fn get_prefer_broadcast_join(&self) -> Result<bool> {

src/query/sql/src/executor/physical_plan_visitor.rs

@@ -237,6 +237,7 @@ pub trait PhysicalPlanReplacer {
             need_hold_hash_table: plan.need_hold_hash_table,
             stat_info: plan.stat_info.clone(),
             probe_keys_rt: plan.probe_keys_rt.clone(),
+            enable_bloom_runtime_filter: plan.enable_bloom_runtime_filter,
             broadcast: plan.broadcast,
             original_join_type: plan.original_join_type.clone(),
         }))

src/query/sql/src/executor/physical_plans/physical_hash_join.rs

@@ -12,6 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::sync::Arc;
+
+use databend_common_catalog::table_context::TableContext;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::type_check::check_cast;
@@ -31,11 +34,13 @@ use crate::executor::physical_plans::FragmentKind;
 use crate::executor::PhysicalPlan;
 use crate::executor::PhysicalPlanBuilder;
 use crate::optimizer::ColumnSet;
+use crate::optimizer::RelExpr;
 use crate::optimizer::SExpr;
 use crate::plans::Join;
 use crate::plans::JoinType;
 use crate::ColumnEntry;
 use crate::IndexType;
+use crate::MetadataRef;
 use crate::ScalarExpr;
 use crate::TypeCheck;
 
@@ -72,6 +77,8 @@ pub struct HashJoin {
 
     // probe keys for runtime filter, and record the index of table that used in probe keys.
     pub probe_keys_rt: Vec<Option<(RemoteExpr<String>, IndexType)>>,
+    // If enable bloom runtime filter
+    pub enable_bloom_runtime_filter: bool,
     // Under cluster, mark if the join is broadcast join.
     pub broadcast: bool,
     // Original join type. Left/Right single join may be convert to inner join
@@ -191,6 +198,7 @@ impl PhysicalPlanBuilder {
         let mut right_join_conditions = Vec::new();
         let mut left_join_conditions_rt = Vec::new();
         let mut probe_to_build_index = Vec::new();
+        let mut table_index = None;
         for (left_condition, right_condition) in join
             .left_conditions
             .iter()
@@ -213,18 +221,21 @@ impl PhysicalPlanBuilder {
             }) {
                 if let Some(column_idx) = left_condition.used_columns().iter().next() {
                     // Safe to unwrap because we have checked the column is a base table column.
-                    let table_index = self
-                        .metadata
-                        .read()
-                        .column(*column_idx)
-                        .table_index()
-                        .unwrap();
+                    if table_index.is_none() {
+                        table_index = Some(
+                            self.metadata
+                                .read()
+                                .column(*column_idx)
+                                .table_index()
+                                .unwrap(),
+                        );
+                    }
                     Some((
                         left_condition
                             .as_raw_expr()
                             .type_check(&*self.metadata.read())?
                             .project_column_ref(|col| col.column_name.clone()),
-                        table_index,
+                        table_index.unwrap(),
                     ))
                 } else {
                     None
@@ -499,6 +510,35 @@ impl PhysicalPlanBuilder {
             stat_info: Some(stat_info),
             broadcast: is_broadcast,
             original_join_type: join.original_join_type.clone(),
+            enable_bloom_runtime_filter: adjust_bloom_runtime_filter(
+                self.ctx.clone(),
+                &self.metadata,
+                table_index,
+                s_expr,
+            )
+            .await?,
         }))
     }
 }
+
+// Check if enable bloom runtime filter
+async fn adjust_bloom_runtime_filter(
+    ctx: Arc<dyn TableContext>,
+    metadata: &MetadataRef,
+    table_index: Option<IndexType>,
+    s_expr: &SExpr,
+) -> Result<bool> {
+    if let Some(table_index) = table_index {
+        let table = metadata.read().table(table_index).table();
+        if let Some(stats) = table.table_statistics(ctx.clone()).await? {
+            if let Some(num_rows) = stats.num_rows {
+                let join_cardinality = RelExpr::with_s_expr(s_expr)
+                    .derive_cardinality()?
+                    .cardinality;
+                // If the filtered data reduces to less than 1/1000 of the original dataset, we will enable bloom runtime filter.
+                return Ok(join_cardinality <= (num_rows / 1000) as f64);
+            }
+        }
+    }
+    Ok(false)
+}

tests/sqllogictests/suites/base/09_fuse_engine/09_0026_merge_into.test

@@ -1,13 +1,7 @@
 statement ok
 set enable_distributed_merge_into = 1;
 
-statement ok
-set enable_runtime_filter = 1;
-
 include ./09_0036_merge_into_without_distributed_enable.test
 
 statement ok
 set enable_distributed_merge_into = 0;
-
-statement ok
-set enable_runtime_filter = 0;

tests/sqllogictests/suites/query/join/runtime_filter.test

@@ -1,6 +1,3 @@
-statement ok
-set enable_runtime_filter = 1;
-
 statement ok
 CREATE TABLE table1 (
     key1 String,
@@ -89,9 +86,6 @@ NULL NULL
 NULL NULL
 NULL NULL
 
-statement ok
-set enable_runtime_filter = 0;
-
 statement ok
 drop table table1;
 

tests/sqllogictests/suites/tpcds/queries.test

@@ -1,9 +1,6 @@
 statement ok
 set sandbox_tenant = 'test_tenant';
 
-statement ok
-set enable_runtime_filter = 1;
-
 statement ok
 use tpcds;
 
@@ -7979,6 +7976,3 @@ Conventional childr NEXT DAY ny metro 256 251 235 0 0
 Conventional childr OVERNIGHT ny metro 188 181 188 0 0
 Conventional childr REGULAR ny metro 179 150 215 0 0
 Conventional childr TWO DAY ny metro 185 183 158 0 0
-
-statement ok
-set enable_runtime_filter = 0