refactor hashed logic

Author: JackTan25

src/query/catalog/src/runtime_filter_info.rs

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

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

@@ -57,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_common_storages_fuse::operations::can_merge_into_target_build_bloom_filter;
 use databend_storages_common_index::BloomIndex;
 use ethnum::U256;
 use itertools::Itertools;
@@ -851,12 +852,14 @@ impl HashJoinBuildState {
             }
 
             // 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 can_merge_into_target_build_bloom_filter(self.ctx.clone(), *table_index)? {
+                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)?;
@@ -961,12 +964,39 @@ impl HashJoinBuildState {
                     return Ok(());
                 }
                 let build_key_column = Column::concat_columns(columns.into_iter())?;
-                let digests = BloomIndex::calculate_nullable_column_digest(
+                // mabye there will be null values here, so we use nullable column, the null value will be treat as default
+                // value for the sepcified type, like String -> "", int -> 0. so we need to remove the null hash values here.
+                let (hashes, bitmap_op) = 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));
+                if let Some(bitmap) = bitmap_op {
+                    // no null values
+                    let digests = if bitmap.unset_bits() == 0 {
+                        hashes.to_vec()
+                    } else {
+                        let new_hashes = Vec::with_capacity(bitmap.len());
+                        assert_eq!(hashes.len(), bitmap.len());
+                        for row_idx in 0..bitmap.len() {
+                            if bitmap.get_bit(row_idx) {
+                                new_hashes.push(hashes[row_idx])
+                            }
+                        }
+                        new_hashes.to_vec()
+                    };
+                    // id is probe key name
+                    runtime_filter.add_merge_into_source_build_siphashkeys((
+                        id.to_string(),
+                        Arc::new(digests),
+                    ));
+                } else {
+                    // id is probe key name
+                    runtime_filter.add_merge_into_source_build_siphashkeys((
+                        id.to_string(),
+                        Arc::new(hashes),
+                    ));
+                }
             }
         }
         Ok(())

src/query/service/tests/it/sql/exec/get_table_bind_test.rs

@@ -18,6 +18,8 @@ use std::sync::atomic::AtomicUsize;
 use std::sync::Arc;
 
 use dashmap::DashMap;
+use databend_common_arrow::arrow::bitmap::Bitmap;
+use databend_common_arrow::arrow::buffer::Buffer;
 use databend_common_base::base::tokio;
 use databend_common_base::base::Progress;
 use databend_common_base::base::ProgressValues;
@@ -26,6 +28,7 @@ use databend_common_catalog::catalog::Catalog;
 use databend_common_catalog::cluster_info::Cluster;
 use databend_common_catalog::database::Database;
 use databend_common_catalog::merge_into_join::MergeIntoJoin;
+use databend_common_catalog::merge_into_join::MergeIntoSourceBuildSegments;
 use databend_common_catalog::plan::DataSourcePlan;
 use databend_common_catalog::plan::PartInfoPtr;
 use databend_common_catalog::plan::Partitions;
@@ -826,6 +829,21 @@ impl TableContext for CtxDelegation {
     fn set_read_block_thresholds(&self, _thresholds: BlockThresholds) {
         todo!()
     }
+
+    fn get_merge_into_source_build_siphashkeys_with_id(
+        &self,
+        id: usize,
+    ) -> Vec<(String, (Buffer<u64>, Option<Bitmap>))> {
+        todo!()
+    }
+
+    fn set_merge_into_source_build_segments(&self, segments: MergeIntoSourceBuildSegments) {
+        todo!()
+    }
+
+    fn get_merge_into_source_build_segments(&self) -> MergeIntoSourceBuildSegments {
+        todo!()
+    }
 }
 
 #[tokio::test(flavor = "multi_thread")]

src/query/storages/fuse/src/operations/mod.rs

@@ -42,6 +42,7 @@ pub use delete::MutationBlockPruningContext;
 pub use merge_into::*;
 pub use mutation::*;
 pub use read::build_row_fetcher_pipeline;
+pub use read::can_merge_into_target_build_bloom_filter;
 pub use read::need_reserve_block_info;
 pub use replace_into::*;
 pub use util::acquire_task_permit;

src/query/storages/fuse/src/operations/read/runtime_filter_prunner.rs

@@ -181,7 +181,6 @@ pub fn runtime_filter_pruner(
             "merge into source build runtime bloom filter: segment_idx:{},blk_idx:{}",
             segment_idx, block_idx
         );
-        // the row_id is generated by block_id, not block_idx,reference to fill_internal_column_meta()
         let block_meta = segment_info.blocks[block_idx].clone();
     }