add runtime filter,need tests

Author: JackTan25

src/query/catalog/src/runtime_filter_info.rs

@@ -15,14 +15,17 @@
 use std::sync::Arc;
 
 use databend_common_expression::Expr;
+use parking_lot::RwLock;
 use xorf::BinaryFuse16;
 
+pub type MergeIntoSourceBuildSiphashkeys = (Vec<String>, Arc<RwLock<Vec<Vec<u64>>>>);
+
 #[derive(Clone, Debug, Default)]
 pub struct RuntimeFilterInfo {
     inlist: Vec<Expr<String>>,
     min_max: Vec<Expr<String>>,
     bloom: Vec<(String, BinaryFuse16)>,
-    siphashes: Vec<(String, Arc<Vec<u64>>)>,
+    siphashes: MergeIntoSourceBuildSiphashkeys,
 }
 
 impl RuntimeFilterInfo {
@@ -34,12 +37,14 @@ impl RuntimeFilterInfo {
         self.bloom.push(bloom);
     }
 
-    pub fn get_merge_into_source_build_siphashkeys(&self) -> Vec<(String, Arc<Vec<u64>>)> {
+    pub fn get_merge_into_source_build_siphashkeys(&self) -> MergeIntoSourceBuildSiphashkeys {
         self.siphashes.clone()
     }
 
-    pub fn add_merge_into_source_build_siphashkeys(&mut self, digests: (String, Arc<Vec<u64>>)) {
-        self.siphashes.push(digests);
+    pub fn add_merge_into_source_build_siphashkeys(&mut self, digests: (String, Vec<u64>)) {
+        self.siphashes.0.push(digests.0);
+        let mut borrow_hash_keys = self.siphashes.1.write();
+        borrow_hash_keys.push(digests.1)
     }
 
     pub fn add_min_max(&mut self, expr: Expr<String>) {

src/query/catalog/src/table_context.rs

@@ -22,8 +22,6 @@ use std::sync::Arc;
 use std::time::SystemTime;
 
 use dashmap::DashMap;
-use databend_common_arrow::arrow::bitmap::Bitmap;
-use databend_common_arrow::arrow::buffer::Buffer;
 use databend_common_base::base::Progress;
 use databend_common_base::base::ProgressValues;
 use databend_common_base::runtime::profile::Profile;
@@ -63,6 +61,7 @@ use crate::plan::DataSourcePlan;
 use crate::plan::PartInfoPtr;
 use crate::plan::Partitions;
 use crate::query_kind::QueryKind;
+use crate::runtime_filter_info::MergeIntoSourceBuildSiphashkeys;
 use crate::runtime_filter_info::RuntimeFilterInfo;
 use crate::statistics::data_cache_statistics::DataCacheMetrics;
 use crate::table::Table;
@@ -278,7 +277,7 @@ pub trait TableContext: Send + Sync {
     fn get_merge_into_source_build_siphashkeys_with_id(
         &self,
         id: usize,
-    ) -> Vec<(String, Arc<Vec<u64>>)>;
+    ) -> Option<MergeIntoSourceBuildSiphashkeys>;
 
     fn get_merge_into_source_build_bloom_probe_keys(&self, id: usize) -> Vec<String>;
 

src/query/service/src/interpreters/interpreter_merge_into.rs

@@ -203,6 +203,7 @@ impl MergeIntoInterpreter {
                     table_info: merge_into_join.table_info.clone(),
                     catalog_info: merge_into_join.catalog_info.clone(),
                     database_name: merge_into_join.database_name.clone(),
+                    table_schema: merge_into_join.table_schema.clone(),
                 });
             }
         }
@@ -305,6 +306,7 @@ impl MergeIntoInterpreter {
                 table_info: Some(table_info.clone()),
                 catalog_info: Some(catalog_.info()),
                 database_name: merge_into_join.database_name.clone(),
+                table_schema: merge_into_join.table_schema.clone(),
             })
         }
         // merge_into_source is used to recv join's datablocks and split them into macthed and not matched

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

@@ -21,7 +21,6 @@ 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;
@@ -964,7 +963,7 @@ impl HashJoinBuildState {
                     return Ok(());
                 }
                 let build_key_column = Column::concat_columns(columns.into_iter())?;
-                // mabye there will be null values here, so we use nullable column, the null value will be treat as default
+                // maybe 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,
@@ -976,7 +975,7 @@ impl HashJoinBuildState {
                     let digests = if bitmap.unset_bits() == 0 {
                         hashes.to_vec()
                     } else {
-                        let new_hashes = Vec::with_capacity(bitmap.len());
+                        let mut 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) {
@@ -986,16 +985,12 @@ impl HashJoinBuildState {
                         new_hashes.to_vec()
                     };
                     // id is probe key name
-                    runtime_filter.add_merge_into_source_build_siphashkeys((
-                        id.to_string(),
-                        Arc::new(digests),
-                    ));
+                    runtime_filter
+                        .add_merge_into_source_build_siphashkeys((id.to_string(), digests));
                 } else {
                     // id is probe key name
-                    runtime_filter.add_merge_into_source_build_siphashkeys((
-                        id.to_string(),
-                        Arc::new(hashes),
-                    ));
+                    runtime_filter
+                        .add_merge_into_source_build_siphashkeys((id.to_string(), hashes.to_vec()));
                 }
             }
         }

src/query/service/src/sessions/query_ctx.rs

@@ -32,8 +32,6 @@ use std::time::UNIX_EPOCH;
 use chrono_tz::Tz;
 use dashmap::mapref::multiple::RefMulti;
 use dashmap::DashMap;
-use databend_common_arrow::arrow::bitmap::Bitmap;
-use databend_common_arrow::arrow::buffer::Buffer;
 use databend_common_base::base::tokio::task::JoinHandle;
 use databend_common_base::base::Progress;
 use databend_common_base::base::ProgressValues;
@@ -48,6 +46,7 @@ use databend_common_catalog::plan::PartInfoPtr;
 use databend_common_catalog::plan::Partitions;
 use databend_common_catalog::plan::StageTableInfo;
 use databend_common_catalog::query_kind::QueryKind;
+use databend_common_catalog::runtime_filter_info::MergeIntoSourceBuildSiphashkeys;
 use databend_common_catalog::runtime_filter_info::RuntimeFilterInfo;
 use databend_common_catalog::statistics::data_cache_statistics::DataCacheMetrics;
 use databend_common_catalog::table_args::TableArgs;
@@ -1031,23 +1030,18 @@ impl TableContext for QueryContext {
 
     fn get_merge_into_source_build_siphashkeys_with_id(
         &self,
-        id: IndexType,
-    ) -> Vec<(String, Arc<Vec<u64>>)> {
+        id: usize,
+    ) -> Option<MergeIntoSourceBuildSiphashkeys> {
         let runtime_filters = self.shared.runtime_filters.read();
-        match runtime_filters.get(&id) {
-            Some(v) => v.get_merge_into_source_build_siphashkeys(),
-            None => vec![],
-        }
+        runtime_filters
+            .get(&id)
+            .map(|v| v.get_merge_into_source_build_siphashkeys())
     }
 
     fn get_merge_into_source_build_bloom_probe_keys(&self, id: usize) -> Vec<String> {
         let runtime_filters = self.shared.runtime_filters.read();
         match runtime_filters.get(&id) {
-            Some(v) => v
-                .get_merge_into_source_build_siphashkeys()
-                .iter()
-                .map(|key| key.0)
-                .collect(),
+            Some(v) => v.get_merge_into_source_build_siphashkeys().0.clone(),
             None => vec![],
         }
     }

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

@@ -18,8 +18,6 @@ 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;
@@ -33,6 +31,7 @@ use databend_common_catalog::plan::DataSourcePlan;
 use databend_common_catalog::plan::PartInfoPtr;
 use databend_common_catalog::plan::Partitions;
 use databend_common_catalog::query_kind::QueryKind;
+use databend_common_catalog::runtime_filter_info::MergeIntoSourceBuildSiphashkeys;
 use databend_common_catalog::runtime_filter_info::RuntimeFilterInfo;
 use databend_common_catalog::statistics::data_cache_statistics::DataCacheMetrics;
 use databend_common_catalog::table::Table;
@@ -832,16 +831,16 @@ impl TableContext for CtxDelegation {
 
     fn get_merge_into_source_build_siphashkeys_with_id(
         &self,
-        id: usize,
-    ) -> Vec<(String, Arc<Vec<u64>>)> {
+        _id: usize,
+    ) -> Option<MergeIntoSourceBuildSiphashkeys> {
         todo!()
     }
 
-    fn get_merge_into_source_build_bloom_probe_keys(&self, id: usize) -> Vec<String> {
+    fn get_merge_into_source_build_bloom_probe_keys(&self, _id: usize) -> Vec<String> {
         todo!()
     }
 
-    fn set_merge_into_source_build_segments(&self, segments: MergeIntoSourceBuildSegments) {
+    fn set_merge_into_source_build_segments(&self, _segments: MergeIntoSourceBuildSegments) {
         todo!()
     }
 

src/query/service/tests/it/storages/fuse/operations/commit.rs

@@ -375,6 +375,33 @@ impl TableContext for CtxDelegation {
         self
     }
 
+    fn get_merge_into_source_build_bloom_probe_keys(&self, _: usize) -> Vec<std::string::String> {
+        todo!()
+    }
+
+    fn set_merge_into_source_build_segments(
+        &self,
+        _: Arc<Vec<(usize, (std::string::String, u64))>>,
+    ) {
+        todo!()
+    }
+
+    fn get_merge_into_source_build_siphashkeys_with_id(
+        &self,
+        _: usize,
+    ) -> std::option::Option<(
+        Vec<std::string::String>,
+        Arc<parking_lot::lock_api::RwLock<parking_lot::RawRwLock, Vec<Vec<u64>>>>,
+    )> {
+        todo!()
+    }
+
+    fn get_merge_into_source_build_segments(
+        &self,
+    ) -> Arc<Vec<(usize, (std::string::String, u64))>> {
+        todo!()
+    }
+
     fn build_table_from_source_plan(&self, _plan: &DataSourcePlan) -> Result<Arc<dyn Table>> {
         todo!()
     }

src/query/sql/src/planner/optimizer/optimizer.rs

@@ -416,8 +416,8 @@ fn optimize_merge_into(opt_ctx: OptimizerContext, plan: Box<MergeInto>) -> Resul
     {
         let merge_into_join = opt_ctx.table_ctx.get_merge_into_join();
         // this is the first time set, so it must be none, and we will set it in `interpreter_merge_into`
-        assert!(matches!(merge_into_join.catalog_info, None));
-        assert!(matches!(merge_into_join.table_info, None));
+        assert!(merge_into_join.catalog_info.is_none());
+        assert!(merge_into_join.table_info.is_none());
         assert!(merge_into_join.database_name.as_str() == "");
         opt_ctx.table_ctx.set_merge_into_join(MergeIntoJoin {
             // we will set catalog_info and table_info in `interpreter_merge_into`
@@ -573,7 +573,7 @@ fn try_to_change_as_broadcast_join(
     if let RelOperator::Exchange(Exchange::Merge) = merge_into_join_sexpr.plan.as_ref() {
         let right_exchange = merge_into_join_sexpr.child(0)?.child(1)?;
         if let RelOperator::Exchange(Exchange::Broadcast) = right_exchange.plan.as_ref() {
-            let mut join: Join = merge_into_join_sexpr.child(0)?.plan().clone().try_into()?;
+            let join: Join = merge_into_join_sexpr.child(0)?.plan().clone().try_into()?;
 
             let join_s_expr = merge_into_join_sexpr
                 .child(0)?

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

@@ -15,7 +15,6 @@
 use std::any::Any;
 use std::sync::Arc;
 
-use databend_common_catalog::merge_into_join;
 use databend_common_catalog::plan::PartInfoPtr;
 use databend_common_catalog::plan::StealablePartitions;
 use databend_common_catalog::table_context::TableContext;