add merge_into_source_build_bloom_info

Author: JackTan25

src/query/catalog/src/merge_into_join.rs

@@ -12,6 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::sync::Arc;
+
+use databend_common_meta_app::schema::CatalogInfo;
+use databend_common_meta_app::schema::TableInfo;
+use databend_storages_common_table_meta::meta::Location;
+
 #[derive(Clone)]
 pub enum MergeIntoJoinType {
     Left,
@@ -23,14 +29,38 @@ pub enum MergeIntoJoinType {
     NormalJoin,
 }
 
-// for now, we just support MergeIntoJoinType::Left to use MergeIntoBlockInfoHashTable in two situations:
-// 1. distributed broadcast join and target table as build side.
-// 2. in standalone mode and target table as build side.
-// we will support Inner next, so the merge_into_join_type is only Left for current implementation in fact.
+pub type MergeIntoSourceBuildSegments = Arc<Vec<(usize, Location)>>;
+
+// MergeIntoJoin is used in two cases:
+// I. target build optimization:
+// we should support MergeIntoJoinType::Left(need to support LeftInner,LeftAnti) to use MergeIntoBlockInfoHashTable in two situations:
+// 1. distributed broadcast join and target table as build side (don't support this now).
+// 2. in standalone mode and target table as build side (supported).
+// 3. native(not-supported) and parquet(supported) format
+// for the `target build optimization`, merge_into_join_type is only Left for current implementation in fact.And we
+// don't support distributed mode and parquet. So only if it is a LeftJoin and non-dirtributed-merge-into(even if in distributed environment
+// but the merge_into_optimizer can also give a non-dirtributed-merge-into) and uses parquet, the `target build optimization` can be enabled.
+// II. source build runtime bloom filter.
+// only if it's a RightJoin((need to support RightInner,RightAnti)),target_tbl_idx is not invalid. We can make sure it enables
+// `source build runtime bloom filter`. And in this case the table_info and catalog_info must be some. We give an `assert` for this judge.
+//
+// So let's do a summary:
+// In most cases, the MergeIntoJoin's `merge_into_join_type` is NormalJoin (even this is a real merge into join),because
+// MergeIntoJoin is used to judge whether we enable target_build_optimizations or source build runtime bloom filter.
+// Both of these optimizations all will check  `merge_into_join_type` to judge if the correlated optimization is enabled.
+// So we only `set_merge_into_join` when we use these two optimziations (of course there is an exception, the `merge_into_optimzier` can't distinct
+// parquet and native, so it will think it's using parquet format in default, so if it's a native format in fact, the `interpreter_merge_into` will rollback it.
+// But the modification is a little tricky. Because the `MergeIntoJoinType` doesn't contain any info about parquet or native,we set the target_table_index as
+// DUMMY_TABLE_INDEX, but in fact it has a target_table_index. However it's ok to do this. Firstly we can do this to make sure the `target build optimization` is
+// closed. And it won't cause a conflict with `source build runtime bloom filter`. Because for the `target build optimization`, it's target build, and for
+// `source build runtime bloom filter`, it's source build).
 pub struct MergeIntoJoin {
     pub merge_into_join_type: MergeIntoJoinType,
     pub is_distributed: bool,
     pub target_tbl_idx: usize,
+    pub table_info: Option<TableInfo>,
+    pub catalog_info: Option<CatalogInfo>,
+    pub database_name: String,
 }
 
 impl Default for MergeIntoJoin {
@@ -40,6 +70,9 @@ impl Default for MergeIntoJoin {
             is_distributed: false,
             // Invalid Index
             target_tbl_idx: usize::MAX,
+            table_info: None,
+            catalog_info: None,
+            database_name: Default::default(),
         }
     }
 }

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

@@ -13,14 +13,12 @@
 // 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;
@@ -51,7 +49,6 @@ 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/runtime_filter_info.rs

@@ -34,6 +34,12 @@ impl RuntimeFilterInfo {
         self.bloom.push(bloom);
     }
 
+    pub fn get_merge_into_source_build_siphashkeys(
+        &mut self,
+    ) -> Vec<(String, (Buffer<u64>, Option<Bitmap>))> {
+        self.siphashes.clone()
+    }
+
     pub fn add_merge_into_source_build_siphashkeys(
         &mut self,
         digests: (String, (Buffer<u64>, Option<Bitmap>)),

src/query/catalog/src/table_context.rs

@@ -22,6 +22,8 @@ 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;
@@ -55,6 +57,7 @@ use xorf::BinaryFuse16;
 use crate::catalog::Catalog;
 use crate::cluster_info::Cluster;
 use crate::merge_into_join::MergeIntoJoin;
+use crate::merge_into_join::MergeIntoSourceBuildSegments;
 use crate::plan::DataSourcePlan;
 use crate::plan::PartInfoPtr;
 use crate::plan::Partitions;
@@ -254,18 +257,28 @@ pub trait TableContext: Send + Sync {
 
     fn get_query_profiles(&self) -> Vec<PlanProfile>;
 
+    // <ProbeSide Index, RuntimeFilterInfo>
     fn set_runtime_filter(&self, filters: (usize, RuntimeFilterInfo));
 
     fn set_merge_into_join(&self, join: MergeIntoJoin);
 
     fn get_merge_into_join(&self) -> MergeIntoJoin;
+    // set the target table's segments
+    fn set_merge_into_source_build_segments(&self, segments: MergeIntoSourceBuildSegments);
+    // get the target table's segments
+    fn get_merge_into_source_build_segments(&self) -> MergeIntoSourceBuildSegments;
 
     fn get_bloom_runtime_filter_with_id(&self, id: usize) -> Vec<(String, BinaryFuse16)>;
 
     fn get_inlist_runtime_filter_with_id(&self, id: usize) -> Vec<Expr<String>>;
 
     fn get_min_max_runtime_filter_with_id(&self, id: usize) -> Vec<Expr<String>>;
 
+    fn get_merge_into_source_build_siphashkeys_with_id(
+        &self,
+        id: usize,
+    ) -> Vec<(String, (Buffer<u64>, Option<Bitmap>))>;
+
     fn has_bloom_runtime_filters(&self, id: usize) -> bool;
     fn txn_mgr(&self) -> TxnManagerRef;
 }

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

@@ -17,6 +17,7 @@ use std::sync::Arc;
 use std::u64::MAX;
 
 use databend_common_catalog::merge_into_join::MergeIntoJoin;
+use databend_common_catalog::merge_into_join::MergeIntoJoinType;
 use databend_common_catalog::table::TableExt;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
@@ -181,7 +182,7 @@ impl MergeIntoInterpreter {
 
         // for `target_build_optimization` we don't need to read rowId column. for now, there are two cases we don't read rowid:
         // I. InsertOnly, the MergeIntoType is InsertOnly
-        // II. target build optimization for this pr. the MergeIntoType is MergeIntoType
+        // II. target build optimization for this pr. the MergeIntoType is FullOperation
         let mut target_build_optimization =
             matches!(self.plan.merge_type, MergeIntoType::FullOperation)
                 && !self.plan.columns_set.contains(&self.plan.row_id_index);
@@ -199,6 +200,9 @@ impl MergeIntoInterpreter {
                     target_tbl_idx: DUMMY_TABLE_INDEX,
                     is_distributed: merge_into_join.is_distributed,
                     merge_into_join_type: merge_into_join.merge_into_join_type,
+                    table_info: merge_into_join.table_info.clone(),
+                    catalog_info: merge_into_join.catalog_info.clone(),
+                    database_name: merge_into_join.database_name.clone(),
                 });
             }
         }
@@ -287,7 +291,22 @@ impl MergeIntoInterpreter {
 
         let table_info = fuse_table.get_table_info().clone();
         let catalog_ = self.ctx.get_catalog(catalog).await?;
-
+        // try add catalog_info and table_info for `source_build_bloom_filter`
+        let merge_into_join = self.ctx.get_merge_into_join();
+        let source_build_bloom_filter = matches!(
+            merge_into_join.merge_into_join_type,
+            MergeIntoJoinType::Right
+        ) && merge_into_join.target_tbl_idx != DUMMY_TABLE_INDEX;
+        if source_build_bloom_filter {
+            self.ctx.set_merge_into_join(MergeIntoJoin {
+                target_tbl_idx: merge_into_join.target_tbl_idx,
+                is_distributed: merge_into_join.is_distributed,
+                merge_into_join_type: merge_into_join.merge_into_join_type,
+                table_info: Some(table_info.clone()),
+                catalog_info: Some(catalog_.info()),
+                database_name: merge_into_join.database_name.clone(),
+            })
+        }
         // merge_into_source is used to recv join's datablocks and split them into macthed and not matched
         // datablocks.
         let merge_into_source = if !*distributed && extract_exchange {

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

@@ -32,13 +32,16 @@ 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;
 use databend_common_base::runtime::profile::Profile;
 use databend_common_base::runtime::profile::ProfileStatisticsName;
 use databend_common_base::runtime::TrySpawn;
 use databend_common_catalog::merge_into_join::MergeIntoJoin;
+use databend_common_catalog::merge_into_join::MergeIntoSourceBuildSegments;
 use databend_common_catalog::plan::DataSourceInfo;
 use databend_common_catalog::plan::DataSourcePlan;
 use databend_common_catalog::plan::PartInfoPtr;
@@ -1008,6 +1011,9 @@ impl TableContext for QueryContext {
             merge_into_join_type: merge_into_join.merge_into_join_type.clone(),
             is_distributed: merge_into_join.is_distributed,
             target_tbl_idx: merge_into_join.target_tbl_idx,
+            catalog_info: merge_into_join.catalog_info.clone(),
+            table_info: merge_into_join.table_info.clone(),
+            database_name: merge_into_join.database_name.clone(),
         }
     }
 
@@ -1019,6 +1025,17 @@ impl TableContext for QueryContext {
         }
     }
 
+    fn get_merge_into_source_build_siphashkeys_with_id(
+        &self,
+        id: IndexType,
+    ) -> Vec<(String, (Buffer<u64>, Option<Bitmap>))> {
+        let runtime_filters = self.shared.runtime_filters.read();
+        match runtime_filters.get(&id) {
+            Some(v) => v.get_merge_into_source_build_siphashkeys(),
+            None => vec![],
+        }
+    }
+
     fn get_inlist_runtime_filter_with_id(&self, id: IndexType) -> Vec<Expr<String>> {
         let runtime_filters = self.shared.runtime_filters.read();
         match runtime_filters.get(&id) {
@@ -1045,6 +1062,16 @@ impl TableContext for QueryContext {
     fn txn_mgr(&self) -> TxnManagerRef {
         self.shared.session.session_ctx.txn_mgr()
     }
+
+    fn set_merge_into_source_build_segments(&self, segments: MergeIntoSourceBuildSegments) {
+        let mut merge_into_source_build_segments =
+            self.shared.merge_into_source_build_segments.write();
+        *merge_into_source_build_segments = segments;
+    }
+
+    fn get_merge_into_source_build_segments(&self) -> MergeIntoSourceBuildSegments {
+        self.shared.merge_into_source_build_segments.read().clone()
+    }
 }
 
 impl TrySpawn for QueryContext {

src/query/service/src/sessions/query_ctx_shared.rs

@@ -26,6 +26,7 @@ use databend_common_base::runtime::drop_guard;
 use databend_common_base::runtime::Runtime;
 use databend_common_catalog::catalog::CatalogManager;
 use databend_common_catalog::merge_into_join::MergeIntoJoin;
+use databend_common_catalog::merge_into_join::MergeIntoSourceBuildSegments;
 use databend_common_catalog::query_kind::QueryKind;
 use databend_common_catalog::runtime_filter_info::RuntimeFilterInfo;
 use databend_common_catalog::statistics::data_cache_statistics::DataCacheMetrics;
@@ -121,6 +122,9 @@ pub struct QueryContextShared {
 
     pub(in crate::sessions) merge_into_join: Arc<RwLock<MergeIntoJoin>>,
 
+    pub(in crate::sessions) merge_into_source_build_segments:
+        Arc<RwLock<MergeIntoSourceBuildSegments>>,
+
     // Records query level data cache metrics
     pub(in crate::sessions) query_cache_metrics: DataCacheMetrics,
 }
@@ -170,6 +174,7 @@ impl QueryContextShared {
             query_profiles: Arc::new(RwLock::new(HashMap::new())),
             runtime_filters: Default::default(),
             merge_into_join: Default::default(),
+            merge_into_source_build_segments: Default::default(),
         }))
     }