fix send

Author: SkyFan2002

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

@@ -15,6 +15,7 @@
 use databend_common_exception::Result;
 use databend_common_sql::executor::physical_plans::RuntimeFilterSink;
 use databend_common_sql::executor::physical_plans::RuntimeFilterSource;
+use databend_common_storages_fuse::TableContext;
 
 use crate::pipelines::processors::transforms::RuntimeFilterSinkProcessor;
 use crate::pipelines::processors::transforms::RuntimeFilterSourceProcessor;
@@ -37,7 +38,8 @@ impl PipelineBuilder {
     pub(crate) fn build_runtime_filter_sink(&mut self, sink: &RuntimeFilterSink) -> Result<()> {
         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(RuntimeFilterSinkProcessor::create)
+            .add_sink(|input| RuntimeFilterSinkProcessor::create(input, node_num))
     }
 }

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

@@ -79,7 +79,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::RuntimeFilterMeta;
+use crate::pipelines::processors::transforms::RuntimeFiltersMeta;
 use crate::pipelines::processors::HashJoinState;
 use crate::sessions::QueryContext;
 
@@ -119,7 +119,7 @@ pub struct HashJoinBuildState {
 
     /// Spill related states.
     pub(crate) memory_settings: MemorySettings,
-    pub(crate) runtime_filter_sender: Option<Sender<RuntimeFilterMeta>>,
+    pub(crate) runtime_filter_sender: Option<Sender<RuntimeFiltersMeta>>,
 }
 
 impl HashJoinBuildState {
@@ -131,7 +131,7 @@ impl HashJoinBuildState {
         build_projections: &ColumnSet,
         hash_join_state: Arc<HashJoinState>,
         num_threads: usize,
-        rf_src_send: Option<Sender<RuntimeFilterMeta>>,
+        rf_src_send: Option<Sender<RuntimeFiltersMeta>>,
     ) -> Result<Arc<HashJoinBuildState>> {
         let hash_key_types = build_keys
             .iter()
@@ -291,6 +291,7 @@ impl HashJoinBuildState {
                     .build_watcher
                     .send(HashTableType::Empty)
                     .map_err(|_| ErrorCode::TokioError("build_watcher channel is closed"))?;
+                self.send_runtime_filter_meta(Default::default())?;
                 self.set_bloom_filter_ready(false)?;
                 return Ok(());
             }
@@ -310,6 +311,7 @@ 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)?;
             }
 
@@ -842,6 +844,7 @@ 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();
         for rf in self.runtime_filter_desc() {
             let mut runtime_filter = RuntimeFilterInfo::default();
             if rf.enable_inlist_runtime_filter && build_num_rows < INLIST_RUNTIME_FILTER_THRESHOLD {
@@ -870,10 +873,11 @@ impl HashJoinBuildState {
             }
             if !runtime_filter.is_empty() {
                 bloom_filter_ready |= !runtime_filter.is_blooms_empty();
-                self.send_runtime_filter_meta(&runtime_filter)?;
+                runtime_filters_meta.add(rf.scan_id, &runtime_filter);
                 self.ctx.set_runtime_filter((rf.scan_id, runtime_filter));
             }
         }
+        self.send_runtime_filter_meta(runtime_filters_meta)?;
         self.set_bloom_filter_ready(bloom_filter_ready)?;
         Ok(())
     }
@@ -1075,9 +1079,9 @@ impl HashJoinBuildState {
             .any(|rf| rf.enable_min_max_runtime_filter)
     }
 
-    fn send_runtime_filter_meta(&self, runtime_filter_info: &RuntimeFilterInfo) -> Result<()> {
+    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_info.into()).unwrap();
+            sender.send_blocking(runtime_filter).unwrap();
             sender.close();
         }
         Ok(())

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

@@ -42,9 +42,9 @@ mod transform_udf_server;
 mod window;
 
 pub use hash_join::*;
-pub use runtime_filter::RuntimeFilterMeta;
 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;

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

@@ -32,13 +32,13 @@ use databend_common_pipeline_sources::AsyncSource;
 use databend_common_pipeline_sources::AsyncSourcer;
 
 pub struct RuntimeFilterSourceProcessor {
-    pub meta_receiver: Receiver<RuntimeFilterMeta>,
+    pub meta_receiver: Receiver<RuntimeFiltersMeta>,
 }
 
 impl RuntimeFilterSourceProcessor {
     pub fn create(
         ctx: Arc<dyn TableContext>,
-        receiver: Receiver<RuntimeFilterMeta>,
+        receiver: Receiver<RuntimeFiltersMeta>,
         output_port: Arc<OutputPort>,
     ) -> Result<ProcessorPtr> {
         AsyncSourcer::create(ctx, output_port, Self {
@@ -63,12 +63,7 @@ impl AsyncSource for RuntimeFilterSourceProcessor {
             rf.is_ok()
         );
         match rf {
-            Ok(runtime_filter) => Ok(Some(DataBlock::empty_with_meta(Box::new(
-                RuntimeFilterMeta {
-                    inlist: runtime_filter.inlist,
-                    min_max: runtime_filter.min_max,
-                },
-            )))),
+            Ok(runtime_filter) => Ok(Some(DataBlock::empty_with_meta(Box::new(runtime_filter)))),
             Err(_) => {
                 // The channel is closed, we should return None to stop generating
                 Ok(None)
@@ -77,22 +72,29 @@ impl AsyncSource for RuntimeFilterSourceProcessor {
     }
 }
 
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Default)]
+pub struct RuntimeFiltersMeta {
+    runtime_filters: Vec<RuntimeFilterMeta>,
+}
+
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct RuntimeFilterMeta {
+    scan_id: usize,
     inlist: Vec<RemoteExpr<String>>,
     min_max: Vec<RemoteExpr<String>>,
 }
 
-#[typetag::serde(name = "runtime_filter_meta")]
-impl BlockMetaInfo for RuntimeFilterMeta {
+#[typetag::serde(name = "runtime_filters_meta")]
+impl BlockMetaInfo for RuntimeFiltersMeta {
     fn clone_self(&self) -> Box<dyn BlockMetaInfo> {
         Box::new(self.clone())
     }
 }
 
-impl From<&RuntimeFilterInfo> for RuntimeFilterMeta {
-    fn from(runtime_filter_info: &RuntimeFilterInfo) -> Self {
-        Self {
+impl RuntimeFiltersMeta {
+    pub fn add(&mut self, scan_id: usize, runtime_filter_info: &RuntimeFilterInfo) {
+        let rf = RuntimeFilterMeta {
+            scan_id,
             inlist: runtime_filter_info
                 .inlists_ref()
                 .iter()
@@ -103,14 +105,21 @@ impl From<&RuntimeFilterInfo> for RuntimeFilterMeta {
                 .iter()
                 .map(|expr| expr.as_remote_expr())
                 .collect(),
-        }
+        };
+        self.runtime_filters.push(rf);
     }
 }
-pub struct RuntimeFilterSinkProcessor {}
+pub struct RuntimeFilterSinkProcessor {
+    node_num: usize,
+    recv_num: usize,
+}
 
 impl RuntimeFilterSinkProcessor {
-    pub fn create(input: Arc<InputPort>) -> Result<ProcessorPtr> {
-        Ok(ProcessorPtr::create(AsyncSinker::create(input, Self {})))
+    pub fn create(input: Arc<InputPort>, node_num: usize) -> Result<ProcessorPtr> {
+        Ok(ProcessorPtr::create(AsyncSinker::create(input, Self {
+            node_num,
+            recv_num: 0,
+        })))
     }
 }
 
@@ -128,12 +137,13 @@ impl AsyncSink for RuntimeFilterSinkProcessor {
         let ptr = data_block
             .take_meta()
             .ok_or_else(|| ErrorCode::Internal("Cannot downcast meta to RuntimeFilterMeta"))?;
-        let runtime_filter = RuntimeFilterMeta::downcast_from(ptr)
+        let runtime_filter = RuntimeFiltersMeta::downcast_from(ptr)
             .ok_or_else(|| ErrorCode::Internal("Cannot downcast meta to RuntimeFilterMeta"))?;
         log::info!(
             "RuntimeFilterSinkProcessor recv runtime filter: {:?}",
             runtime_filter
         );
-        Ok(true)
+        self.recv_num += 1;
+        Ok(self.node_num == self.recv_num)
     }
 }

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

@@ -137,7 +137,7 @@ use crate::clusters::Cluster;
 use crate::clusters::ClusterHelper;
 use crate::locks::LockManager;
 use crate::pipelines::executor::PipelineExecutor;
-use crate::pipelines::processors::transforms::RuntimeFilterMeta;
+use crate::pipelines::processors::transforms::RuntimeFiltersMeta;
 use crate::servers::flight::v1::exchange::DataExchangeManager;
 use crate::sessions::query_affect::QueryAffect;
 use crate::sessions::query_ctx_shared::MemoryUpdater;
@@ -284,11 +284,11 @@ impl QueryContext {
         }
     }
 
-    pub fn rf_src_recv(&self, join_id: u32) -> Receiver<RuntimeFilterMeta> {
+    pub fn rf_src_recv(&self, join_id: u32) -> Receiver<RuntimeFiltersMeta> {
         self.shared.rf_src_recv(join_id)
     }
 
-    pub fn rf_src_send(&self, join_id: u32) -> Sender<RuntimeFilterMeta> {
+    pub fn rf_src_send(&self, join_id: u32) -> Sender<RuntimeFiltersMeta> {
         self.shared.rf_src_send(join_id)
     }
 

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

@@ -70,7 +70,7 @@ use uuid::Uuid;
 use crate::clusters::Cluster;
 use crate::clusters::ClusterDiscovery;
 use crate::pipelines::executor::PipelineExecutor;
-use crate::pipelines::processors::transforms::RuntimeFilterMeta;
+use crate::pipelines::processors::transforms::RuntimeFiltersMeta;
 use crate::sessions::query_affect::QueryAffect;
 use crate::sessions::Session;
 use crate::storages::Table;
@@ -179,8 +179,8 @@ pub struct QueryContextShared {
 }
 
 type RuntimeFilterChannel = (
-    Option<Sender<RuntimeFilterMeta>>,
-    Option<Receiver<RuntimeFilterMeta>>,
+    Option<Sender<RuntimeFiltersMeta>>,
+    Option<Receiver<RuntimeFiltersMeta>>,
 );
 
 impl QueryContextShared {
@@ -251,7 +251,7 @@ impl QueryContextShared {
         }))
     }
 
-    pub fn rf_src_recv(&self, join_id: u32) -> Receiver<RuntimeFilterMeta> {
+    pub fn rf_src_recv(&self, join_id: u32) -> Receiver<RuntimeFiltersMeta> {
         let mut rf_source = self.rf_source.lock();
         match rf_source.get_mut(&join_id).map(|(_, receiver)| receiver) {
             Some(receiver) => receiver.take().unwrap(),
@@ -262,7 +262,7 @@ impl QueryContextShared {
             }
         }
     }
-    pub fn rf_src_send(&self, join_id: u32) -> Sender<RuntimeFilterMeta> {
+    pub fn rf_src_send(&self, join_id: u32) -> Sender<RuntimeFiltersMeta> {
         let mut rf_source = self.rf_source.lock();
         match rf_source.get_mut(&join_id).map(|(sender, _)| sender) {
             Some(sender) => sender.take().unwrap(),