add bloom index for merge into runtime filter

Author: JackTan25

Cargo.lock

@@ -10,6 +10,7 @@ test = false
 
 [dependencies]
 databend-common-base = { path = "../../common/base" }
+databend-common-arrow = { path = "../../common/arrow" }
 databend-common-config = { path = "../config" }
 databend-common-exception = { path = "../../common/exception" }
 databend-common-expression = { path = "../expression" }

src/query/catalog/Cargo.toml

@@ -13,12 +13,14 @@
 // limitations under the License.
 
 use std::collections::BTreeMap;
+use std::collections::HashMap;
 
 use databend_common_expression::FieldIndex;
 use databend_common_expression::RemoteExpr;
 use databend_common_expression::Scalar;
 use databend_common_expression::TableSchemaRef;
 use databend_common_meta_app::schema::CatalogInfo;
+use databend_storages_common_table_meta::meta::BlockMeta;
 
 use crate::plan::datasource::datasource_info::DataSourceInfo;
 use crate::plan::PartStatistics;
@@ -49,6 +51,7 @@ pub struct DataSourcePlan {
     pub data_mask_policy: Option<BTreeMap<FieldIndex, RemoteExpr>>,
 
     pub table_index: usize,
+    // pub merge_into_target_table_block_meta: Option<HashMap<BlockMetaIndex, BlockMeta>>,
 }
 
 impl DataSourcePlan {

src/query/catalog/src/plan/datasource/datasource_plan.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use databend_common_arrow::arrow::bitmap::Bitmap;
+use databend_common_arrow::arrow::buffer::Buffer;
 use databend_common_expression::Expr;
 use xorf::BinaryFuse16;
 
@@ -20,6 +22,7 @@ pub struct RuntimeFilterInfo {
     inlist: Vec<Expr<String>>,
     min_max: Vec<Expr<String>>,
     bloom: Vec<(String, BinaryFuse16)>,
+    siphashes: Vec<(String, (Buffer<u64>, Option<Bitmap>))>,
 }
 
 impl RuntimeFilterInfo {
@@ -31,6 +34,13 @@ impl RuntimeFilterInfo {
         self.bloom.push(bloom);
     }
 
+    pub fn add_merge_into_source_build_siphashkeys(
+        &mut self,
+        digests: (String, (Buffer<u64>, Option<Bitmap>)),
+    ) {
+        self.siphashes.push(digests);
+    }
+
     pub fn add_min_max(&mut self, expr: Expr<String>) {
         self.min_max.push(expr);
     }

src/query/catalog/src/runtime_filter_info.rs

@@ -21,7 +21,9 @@ use std::sync::atomic::Ordering;
 use std::sync::Arc;
 
 use databend_common_arrow::arrow::bitmap::Bitmap;
+use databend_common_arrow::arrow::buffer::Buffer;
 use databend_common_base::base::tokio::sync::Barrier;
+use databend_common_catalog::merge_into_join::MergeIntoJoinType;
 use databend_common_catalog::runtime_filter_info::RuntimeFilterInfo;
 use databend_common_catalog::table_context::TableContext;
 use databend_common_exception::ErrorCode;
@@ -55,6 +57,7 @@ use databend_common_hashtable::StringRawEntry;
 use databend_common_hashtable::STRING_EARLY_SIZE;
 use databend_common_sql::plans::JoinType;
 use databend_common_sql::ColumnSet;
+use databend_storages_common_index::BloomIndex;
 use ethnum::U256;
 use itertools::Itertools;
 use log::info;
@@ -845,9 +848,19 @@ impl HashJoinBuildState {
                     probe_key,
                 )?;
             }
+
+            // add BloomIndex hash keys for merge into source build.
+            self.build_merge_into_runtime_filter_siphashes(
+                build_chunks,
+                &mut runtime_filter,
+                build_key,
+                probe_key,
+            )?;
+
             if self.enable_bloom_runtime_filter {
                 self.bloom_runtime_filter(build_chunks, &mut runtime_filter, build_key, probe_key)?;
             }
+
             if self.enable_min_max_runtime_filter {
                 self.min_max_runtime_filter(
                     build_chunks,
@@ -912,6 +925,52 @@ impl HashJoinBuildState {
         Ok(())
     }
 
+    //      for merge into source build cases, like below:
+    // merge into `t1` using `t2` on xxx when matched xx when not matched xxx, if merge_into_optimizer
+    // gives `t2` as source build side, we can build source join keys `siphashes`, that's because we use
+    // siphash to build target table's bloom index block.
+    // in this way, we can avoid current `runtime_filter()` func's performance cost, especially for large
+    // target table case, the `runtime_filter()`'s cost is even higer than disable `runtime_filter()`.
+    //      However, for `build_runtime_filter_siphashes()` usages, we currently just used for merge into,
+    // we doesn't support join query, and it's only for `source build` cases. In fact, source build is the
+    // main case in most time.
+    fn build_merge_into_runtime_filter_siphashes(
+        &self,
+        data_blocks: &[DataBlock],
+        runtime_filter: &mut RuntimeFilterInfo,
+        build_key: &Expr,
+        probe_key: &Expr<String>,
+    ) -> Result<()> {
+        // `calculate_nullable_column_digest`, `apply_bloom_pruning`
+        let merge_type = self.ctx.get_merge_into_join();
+        if matches!(merge_type.merge_into_join_type, MergeIntoJoinType::Right) {
+            if let Expr::ColumnRef { id, .. } = probe_key {
+                let mut columns = Vec::with_capacity(data_blocks.len());
+                for block in data_blocks.iter() {
+                    if block.num_columns() == 0 {
+                        continue;
+                    }
+                    let evaluator = Evaluator::new(block, &self.func_ctx, &BUILTIN_FUNCTIONS);
+                    let column = evaluator
+                        .run(build_key)?
+                        .convert_to_full_column(build_key.data_type(), block.num_rows());
+                    columns.push(column);
+                }
+                if columns.is_empty() {
+                    return Ok(());
+                }
+                let build_key_column = Column::concat_columns(columns.into_iter())?;
+                let digests = BloomIndex::calculate_nullable_column_digest(
+                    &self.func_ctx,
+                    &build_key_column,
+                    &build_key_column.data_type(),
+                )?;
+                runtime_filter.add_merge_into_source_build_siphashkeys((id.to_string(), digests));
+            }
+        }
+        Ok(())
+    }
+
     fn inlist_runtime_filter(
         &self,
         runtime_filter: &mut RuntimeFilterInfo,

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

@@ -485,6 +485,12 @@ impl DefaultSettings {
                     mode: SettingMode::Both,
                     range: Some(SettingRange::Numeric(0..=1)),
                 }),
+                ("enable_merge_into_source_build_bloom", DefaultSettingValue {
+                    value: UserSettingValue::UInt64(0),
+                    desc: "Enable merge into source build bloom.",
+                    mode: SettingMode::Both,
+                    range: Some(SettingRange::Numeric(0..=1)),
+                }),
                 ("enable_distributed_merge_into", DefaultSettingValue {
                     value: UserSettingValue::UInt64(0),
                     desc: "Enable distributed merge into.",

src/query/settings/src/settings_default.rs

@@ -438,6 +438,10 @@ impl Settings {
         Ok(self.try_get_u64("enable_experimental_merge_into")? != 0)
     }
 
+    pub fn get_enable_merge_into_source_build_bloom(&self) -> Result<bool> {
+        Ok(self.try_get_u64("enable_merge_into_source_build_bloom")? != 0)
+    }
+
     pub fn get_enable_distributed_merge_into(&self) -> Result<bool> {
         Ok(self.try_get_u64("enable_distributed_merge_into")? != 0)
     }

src/query/settings/src/settings_getter_setter.rs

@@ -324,7 +324,7 @@ fn format_merge_into(merge_into: &MergeInto) -> Result<String> {
     ));
     // add macthed clauses
     let mut matched_children = Vec::with_capacity(merge_into.matched_evaluators.len());
-    let taregt_schema = table_entry.table().schema();
+    let taregt_schema = table_entry.table().schema_with_stream();
     for evaluator in &merge_into.matched_evaluators {
         let condition_format = evaluator.condition.as_ref().map_or_else(
             || "condition: None".to_string(),

src/query/sql/src/planner/format/display_plan.rs

@@ -15,6 +15,8 @@
 use std::collections::VecDeque;
 use std::sync::Arc;
 
+use databend_common_catalog::merge_into_join::MergeIntoJoinType;
+use databend_common_catalog::plan::DataSourceInfo;
 use databend_common_catalog::plan::DataSourcePlan;
 use databend_common_catalog::plan::PartInfoPtr;
 use databend_common_catalog::plan::StealablePartitions;
@@ -148,8 +150,34 @@ pub fn build_fuse_parquet_source_pipeline(
     (max_threads, max_io_requests) =
         adjust_threads_and_request(false, max_threads, max_io_requests, plan);
 
-    let mut source_builder = SourcePipeBuilder::create();
+    let merge_into_join = ctx.get_merge_into_join();
+    if ctx
+        .get_settings()
+        .get_enable_merge_into_source_build_bloom()?
+        && matches!(
+            merge_into_join.merge_into_join_type,
+            MergeIntoJoinType::Right
+        )
+        && merge_into_join.target_tbl_idx == plan.table_index
+    {
+        // we can add block_metas info for merge into runtime filter, they will
+        // be used for bloom prune for target table block.
+        assert!(matches!(plan.source_info, DataSourceInfo::TableSource(_)));
+        // if let DataSourceInfo::TableSource(table_info) = plan.source_info {
+        // let table = ctx
+        //     .get_table(table_info.catalog(), table_info.name(), table_name)
+        //     .await?;
+        // let fuse_table = table.as_any().downcast_ref::<FuseTable>().ok_or_else(|| {
+        //     ErrorCode::Unimplemented(format!(
+        //         "table {}, engine type {}, does not support MERGE INTO",
+        //         table.name(),
+        //         table.get_table_info().engine(),
+        //     ))
+        // })?;
+        // }
+    }
 
+    let mut source_builder = SourcePipeBuilder::create();
     match block_reader.support_blocking_api() {
         true => {
             let partitions = dispatch_partitions(ctx.clone(), plan, max_threads);