update

Author: SkyFan2002

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

@@ -33,6 +33,7 @@ use databend_common_expression::RawExpr;
 use databend_common_expression::Scalar;
 use databend_common_functions::BUILTIN_FUNCTIONS;
 use fastbloom::BloomFilter;
+use fastbloom::DefaultHasher;
 
 use super::builder::should_enable_runtime_filter;
 use super::packet::JoinRuntimeFilterPacket;
@@ -252,9 +253,12 @@ async fn build_bloom_filter(
     let probe_key = probe_key.as_column_ref().unwrap();
     let column_name = probe_key.id.to_string();
     let total_items = bloom.len();
+    let hasher = DefaultHasher::default();
 
     if total_items < 50000 {
-        let filter = BloomFilter::with_false_pos(0.01).items(bloom);
+        let filter = BloomFilter::with_false_pos(0.01)
+            .hasher(hasher)
+            .items(bloom);
         return Ok(RuntimeFilterBloom {
             column_name,
             filter,
@@ -271,11 +275,12 @@ async fn build_bloom_filter(
     let tasks: Vec<_> = chunks
         .into_iter()
         .map(|chunk| {
+            let hasher = hasher.clone();
             databend_common_base::runtime::spawn(async move {
-                let mut filter = BloomFilter::with_false_pos(0.01).expected_items(total_items);
-                for hash in chunk {
-                    filter.insert_hash(hash);
-                }
+                let mut filter = BloomFilter::with_false_pos(0.01)
+                    .hasher(hasher)
+                    .expected_items(total_items);
+                filter.extend(chunk);
                 Ok::<BloomFilter, ErrorCode>(filter)
             })
         })