merge rf

Author: SkyFan2002

Cargo.lock

@@ -12,62 +12,32 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashMap;
+
 use databend_common_base::base::tokio::sync::watch;
 use databend_common_base::base::tokio::sync::watch::Receiver;
 use databend_common_base::base::tokio::sync::watch::Sender;
 use databend_common_expression::Expr;
 use xorf::BinaryFuse16;
 
 #[derive(Clone, Debug, Default)]
-pub struct RuntimeFilterInfo {
-    inlist: Vec<Expr<String>>,
-    min_max: Vec<Expr<String>>,
-    bloom: Vec<(String, BinaryFuse16)>,
+pub struct RuntimeFiltersForScan {
+    pub inlist: HashMap<usize, Expr<String>>,
+    pub min_max: HashMap<usize, Expr<String>>,
+    pub bloom: HashMap<usize, (String, BinaryFuse16)>,
 }
 
-impl RuntimeFilterInfo {
-    pub fn add_inlist(&mut self, expr: Expr<String>) {
-        self.inlist.push(expr);
-    }
-
-    pub fn add_bloom(&mut self, bloom: (String, BinaryFuse16)) {
-        self.bloom.push(bloom);
-    }
-
-    pub fn add_min_max(&mut self, expr: Expr<String>) {
-        self.min_max.push(expr);
-    }
-
-    pub fn get_inlist(&self) -> &Vec<Expr<String>> {
-        &self.inlist
-    }
-
-    pub fn get_bloom(&self) -> &Vec<(String, BinaryFuse16)> {
-        &self.bloom
-    }
-
-    pub fn get_min_max(&self) -> &Vec<Expr<String>> {
-        &self.min_max
-    }
-
-    pub fn blooms(self) -> Vec<(String, BinaryFuse16)> {
-        self.bloom
-    }
-
-    pub fn inlists(self) -> Vec<Expr<String>> {
-        self.inlist
-    }
-
-    pub fn inlists_ref(&self) -> &Vec<Expr<String>> {
-        &self.inlist
+impl RuntimeFiltersForScan {
+    pub fn add_inlist(&mut self, rf_id: usize, expr: Expr<String>) {
+        self.inlist.insert(rf_id, expr);
     }
 
-    pub fn min_maxs(self) -> Vec<Expr<String>> {
-        self.min_max
+    pub fn add_bloom(&mut self, rf_id: usize, bloom: (String, BinaryFuse16)) {
+        self.bloom.insert(rf_id, bloom);
     }
 
-    pub fn min_maxs_ref(&self) -> &Vec<Expr<String>> {
-        &self.min_max
+    pub fn add_min_max(&mut self, rf_id: usize, expr: Expr<String>) {
+        self.min_max.insert(rf_id, expr);
     }
 
     pub fn is_empty(&self) -> bool {

src/query/catalog/src/runtime_filter_info.rs

@@ -74,8 +74,8 @@ use crate::plan::PartInfoPtr;
 use crate::plan::PartStatistics;
 use crate::plan::Partitions;
 use crate::query_kind::QueryKind;
-use crate::runtime_filter_info::RuntimeFilterInfo;
 use crate::runtime_filter_info::RuntimeFilterReady;
+use crate::runtime_filter_info::RuntimeFiltersForScan;
 use crate::statistics::data_cache_statistics::DataCacheMetrics;
 use crate::table::Table;
 
@@ -324,7 +324,7 @@ pub trait TableContext: Send + Sync {
 
     fn get_query_profiles(&self) -> Vec<PlanProfile>;
 
-    fn set_runtime_filter(&self, filters: (usize, RuntimeFilterInfo));
+    fn set_runtime_filter(&self, filters: (usize, RuntimeFiltersForScan));
 
     fn set_runtime_filter_ready(&self, table_index: usize, ready: Arc<RuntimeFilterReady>);
 

src/query/catalog/src/table_context.rs

@@ -15,7 +15,6 @@ databend-common-exception = { workspace = true }
 databend-common-expression = { workspace = true }
 databend-common-pipeline-core = { workspace = true }
 futures = { workspace = true }
-log = { workspace = true }
 parking_lot = { workspace = true }
 
 [lints]

src/query/pipeline/sources/Cargo.toml

@@ -27,6 +27,7 @@ use crate::pipelines::processors::transforms::range_join::TransformRangeJoinLeft
 use crate::pipelines::processors::transforms::range_join::TransformRangeJoinRight;
 use crate::pipelines::processors::transforms::HashJoinBuildState;
 use crate::pipelines::processors::transforms::HashJoinProbeState;
+use crate::pipelines::processors::transforms::RuntimeFilterChannels;
 use crate::pipelines::processors::transforms::TransformHashJoinBuild;
 use crate::pipelines::processors::transforms::TransformHashJoinProbe;
 use crate::pipelines::processors::HashJoinDesc;
@@ -157,7 +158,10 @@ impl PipelineBuilder {
             hash_join_plan
                 .runtime_filter_plan
                 .as_ref()
-                .map(|_| self.ctx.rf_src_send(hash_join_plan.join_id)),
+                .map(|_| RuntimeFilterChannels {
+                    rf_src_send: self.ctx.rf_src_send(hash_join_plan.join_id),
+                    rf_sink_recv: self.ctx.rf_sink_recv(hash_join_plan.join_id),
+                }),
         )?;
         build_state.add_runtime_filter_ready();
 

src/query/service/src/pipelines/builders/builder_join.rs

@@ -39,7 +39,8 @@ impl PipelineBuilder {
         self.build_pipeline(&sink.input)?;
         self.main_pipeline.resize(1, true)?;
         let node_num = self.ctx.get_cluster().nodes.len();
-        self.main_pipeline
-            .add_sink(|input| RuntimeFilterSinkProcessor::create(input, node_num))
+        self.main_pipeline.add_sink(|input| {
+            RuntimeFilterSinkProcessor::create(input, node_num, self.ctx.rf_sink_send(sink.join_id))
+        })
     }
 }

src/query/service/src/pipelines/builders/builder_runtime_filter.rs

@@ -54,7 +54,7 @@ pub struct HashJoinDesc {
 }
 
 pub struct RuntimeFilterDesc {
-    pub _id: usize,
+    pub id: usize,
     pub build_key: Expr,
     pub probe_key: Expr<String>,
     pub scan_id: usize,
@@ -78,7 +78,7 @@ impl From<&RemoteRuntimeFiltersDesc> for RuntimeFiltersDesc {
 impl From<&RemoteRuntimeFilterDesc> for RuntimeFilterDesc {
     fn from(runtime_filter: &RemoteRuntimeFilterDesc) -> Self {
         Self {
-            _id: runtime_filter.id,
+            id: runtime_filter.id,
             build_key: runtime_filter.build_key.as_expr(&BUILTIN_FUNCTIONS),
             probe_key: runtime_filter.probe_key.as_expr(&BUILTIN_FUNCTIONS),
             scan_id: runtime_filter.scan_id,

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

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashMap;
 use std::collections::HashSet;
 use std::collections::VecDeque;
 use std::ops::ControlFlow;
@@ -20,10 +21,11 @@ use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
 
+use async_channel::Receiver;
 use async_channel::Sender;
 use databend_common_base::base::tokio::sync::Barrier;
-use databend_common_catalog::runtime_filter_info::RuntimeFilterInfo;
 use databend_common_catalog::runtime_filter_info::RuntimeFilterReady;
+use databend_common_catalog::runtime_filter_info::RuntimeFiltersForScan;
 use databend_common_catalog::table_context::TableContext;
 use databend_common_column::bitmap::Bitmap;
 use databend_common_exception::ErrorCode;
@@ -79,7 +81,7 @@ use crate::pipelines::processors::transforms::hash_join::FixedKeyHashJoinHashTab
 use crate::pipelines::processors::transforms::hash_join::HashJoinHashTable;
 use crate::pipelines::processors::transforms::hash_join::SerializerHashJoinHashTable;
 use crate::pipelines::processors::transforms::hash_join::SingleBinaryHashJoinHashTable;
-use crate::pipelines::processors::transforms::RuntimeFiltersMeta;
+use crate::pipelines::processors::transforms::RemoteRuntimeFilters;
 use crate::pipelines::processors::HashJoinState;
 use crate::sessions::QueryContext;
 
@@ -119,7 +121,14 @@ pub struct HashJoinBuildState {
 
     /// Spill related states.
     pub(crate) memory_settings: MemorySettings,
-    pub(crate) runtime_filter_sender: Option<Sender<RuntimeFiltersMeta>>,
+    pub(crate) rf_channels: Option<RuntimeFilterChannels>,
+}
+
+pub struct RuntimeFilterChannels {
+    /// send runtime filter to `RuntimeFilterSourceProcessor`
+    pub(crate) rf_src_send: Sender<RemoteRuntimeFilters>,
+    /// receive runtime filter from `RuntimeFilterSinkProcessor`
+    pub(crate) rf_sink_recv: Receiver<RemoteRuntimeFilters>,
 }
 
 impl HashJoinBuildState {
@@ -131,7 +140,7 @@ impl HashJoinBuildState {
         build_projections: &ColumnSet,
         hash_join_state: Arc<HashJoinState>,
         num_threads: usize,
-        rf_src_send: Option<Sender<RuntimeFiltersMeta>>,
+        rf_channels: Option<RuntimeFilterChannels>,
     ) -> Result<Arc<HashJoinBuildState>> {
         let hash_key_types = build_keys
             .iter()
@@ -168,7 +177,7 @@ impl HashJoinBuildState {
             build_hash_table_tasks: Default::default(),
             mutex: Default::default(),
             memory_settings,
-            runtime_filter_sender: rf_src_send,
+            rf_channels,
         }))
     }
 
@@ -311,7 +320,6 @@ impl HashJoinBuildState {
             if self.hash_join_state.spilled_partitions.read().is_empty() {
                 self.add_runtime_filter(&build_chunks, build_num_rows)?;
             } else {
-                self.send_runtime_filter_meta(Default::default())?;
                 self.set_bloom_filter_ready(false)?;
             }
 
@@ -844,15 +852,16 @@ impl HashJoinBuildState {
 
     fn add_runtime_filter(&self, build_chunks: &[DataBlock], build_num_rows: usize) -> Result<()> {
         let mut bloom_filter_ready = false;
-        let mut runtime_filters_meta = RuntimeFiltersMeta::default();
+        let mut runtime_filters = HashMap::new();
         for rf in self.runtime_filter_desc() {
-            let mut runtime_filter = RuntimeFilterInfo::default();
+            let mut runtime_filter = RuntimeFiltersForScan::default();
             if rf.enable_inlist_runtime_filter && build_num_rows < INLIST_RUNTIME_FILTER_THRESHOLD {
                 self.inlist_runtime_filter(
                     &mut runtime_filter,
                     build_chunks,
                     &rf.build_key,
                     &rf.probe_key,
+                    rf.id,
                 )?;
             }
             if rf.enable_bloom_runtime_filter {
@@ -861,6 +870,7 @@ impl HashJoinBuildState {
                     &mut runtime_filter,
                     &rf.build_key,
                     &rf.probe_key,
+                    rf.id,
                 )?;
             }
             if rf.enable_min_max_runtime_filter {
@@ -869,25 +879,26 @@ impl HashJoinBuildState {
                     &mut runtime_filter,
                     &rf.build_key,
                     &rf.probe_key,
+                    rf.id,
                 )?;
             }
             if !runtime_filter.is_empty() {
                 bloom_filter_ready |= !runtime_filter.is_blooms_empty();
-                runtime_filters_meta.add(rf.scan_id, &runtime_filter);
-                self.ctx.set_runtime_filter((rf.scan_id, runtime_filter));
+                runtime_filters.insert(rf.scan_id, runtime_filter);
             }
         }
-        self.send_runtime_filter_meta(runtime_filters_meta)?;
+        self.send_runtime_filter_meta(runtime_filters)?;
         self.set_bloom_filter_ready(bloom_filter_ready)?;
         Ok(())
     }
 
     fn bloom_runtime_filter(
         &self,
         data_blocks: &[DataBlock],
-        runtime_filter: &mut RuntimeFilterInfo,
+        runtime_filter: &mut RuntimeFiltersForScan,
         build_key: &Expr,
         probe_key: &Expr<String>,
+        rf_id: usize,
     ) -> Result<()> {
         if !build_key.data_type().remove_nullable().is_number()
             && !build_key.data_type().remove_nullable().is_string()
@@ -925,23 +936,24 @@ impl HashJoinBuildState {
             hashes_vec.push(hash);
         });
         let filter = BinaryFuse16::try_from(&hashes_vec)?;
-        runtime_filter.add_bloom((id.to_string(), filter));
+        runtime_filter.add_bloom(rf_id, (id.to_string(), filter));
         Ok(())
     }
 
     fn inlist_runtime_filter(
         &self,
-        runtime_filter: &mut RuntimeFilterInfo,
+        runtime_filter: &mut RuntimeFiltersForScan,
         data_blocks: &[DataBlock],
         build_key: &Expr,
         probe_key: &Expr<String>,
+        rf_id: usize,
     ) -> Result<()> {
         if let Some(distinct_build_column) =
             dedup_build_key_column(&self.func_ctx, data_blocks, build_key)?
         {
             if let Some(filter) = inlist_filter(probe_key, distinct_build_column.clone())? {
                 info!("inlist_filter: {:?}", filter.sql_display());
-                runtime_filter.add_inlist(filter);
+                runtime_filter.add_inlist(rf_id, filter);
             }
         }
         Ok(())
@@ -950,9 +962,10 @@ impl HashJoinBuildState {
     fn min_max_runtime_filter(
         &self,
         data_blocks: &[DataBlock],
-        runtime_filter: &mut RuntimeFilterInfo,
+        runtime_filter: &mut RuntimeFiltersForScan,
         build_key: &Expr,
         probe_key: &Expr<String>,
+        rf_id: usize,
     ) -> Result<()> {
         if !build_key.runtime_filter_supported_types() {
             return Ok(());
@@ -1045,7 +1058,7 @@ impl HashJoinBuildState {
             };
             if let Some(min_max_filter) = min_max_filter {
                 info!("min_max_filter: {:?}", min_max_filter.sql_display());
-                runtime_filter.add_min_max(min_max_filter);
+                runtime_filter.add_min_max(rf_id, min_max_filter);
             }
         }
         Ok(())
@@ -1079,10 +1092,24 @@ impl HashJoinBuildState {
             .any(|rf| rf.enable_min_max_runtime_filter)
     }
 
-    fn send_runtime_filter_meta(&self, runtime_filter: RuntimeFiltersMeta) -> Result<()> {
-        if let Some(sender) = self.runtime_filter_sender.as_ref() {
-            sender.send_blocking(runtime_filter).unwrap();
-            sender.close();
+    fn send_runtime_filter_meta(
+        &self,
+        mut rf: HashMap<usize, RuntimeFiltersForScan>,
+    ) -> Result<()> {
+        if let Some(channels) = self.rf_channels.as_ref() {
+            channels
+                .rf_src_send
+                .send_blocking(rf.into())
+                .map_err(|_| ErrorCode::TokioError("send runtime filter meta failed"))?;
+            channels.rf_src_send.close();
+            let merged_rf = channels
+                .rf_sink_recv
+                .recv_blocking()
+                .map_err(|_| ErrorCode::TokioError("receive runtime filter meta failed"))?;
+            rf = merged_rf.into();
+        }
+        for (scan_id, runtime_filter) in rf.into_iter() {
+            self.ctx.set_runtime_filter((scan_id, runtime_filter));
         }
         Ok(())
     }

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

@@ -31,6 +31,7 @@ mod util;
 
 pub use desc::HashJoinDesc;
 pub use hash_join_build_state::HashJoinBuildState;
+pub use hash_join_build_state::RuntimeFilterChannels;
 pub use hash_join_probe_state::HashJoinProbeState;
 pub use hash_join_spiller::HashJoinSpiller;
 pub use hash_join_state::*;

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

@@ -42,9 +42,9 @@ mod transform_udf_server;
 mod window;
 
 pub use hash_join::*;
+pub use runtime_filter::RemoteRuntimeFilters;
 pub use runtime_filter::RuntimeFilterSinkProcessor;
 pub use runtime_filter::RuntimeFilterSourceProcessor;
-pub use runtime_filter::RuntimeFiltersMeta;
 pub use transform_add_computed_columns::TransformAddComputedColumns;
 pub use transform_add_const_columns::TransformAddConstColumns;
 pub use transform_add_internal_columns::TransformAddInternalColumns;