refactor

Author: SkyFan2002

src/query/catalog/src/runtime_filter_info.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::fmt::Debug;
 use std::fmt::Formatter;
 use std::sync::atomic::AtomicU64;
@@ -21,6 +22,8 @@ use std::sync::Arc;
 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_exception::ErrorCode;
+use databend_common_exception::Result;
 use databend_common_expression::Expr;
 use xorf::BinaryFuse16;
 
@@ -71,6 +74,7 @@ pub struct RuntimeFilterBloom {
 
 #[derive(Default)]
 pub struct RuntimeFilterStats {
+    _filter_id: usize,
     bloom_time_ns: AtomicU64,
     bloom_rows_filtered: AtomicU64,
     inlist_min_max_time_ns: AtomicU64,
@@ -98,47 +102,87 @@ impl RuntimeFilterStats {
             .fetch_add(partitions_pruned, Ordering::Relaxed);
     }
 
-    pub fn snapshot(&self) -> RuntimeFilterStatsSnapshot {
-        RuntimeFilterStatsSnapshot {
-            bloom_time_ns: self.bloom_time_ns.load(Ordering::Relaxed),
-            bloom_rows_filtered: self.bloom_rows_filtered.load(Ordering::Relaxed),
-            inlist_min_max_time_ns: self.inlist_min_max_time_ns.load(Ordering::Relaxed),
-            min_max_rows_filtered: self.min_max_rows_filtered.load(Ordering::Relaxed),
-            min_max_partitions_pruned: self.min_max_partitions_pruned.load(Ordering::Relaxed),
-        }
+    pub fn bloom_time_ns(&self) -> u64 {
+        self.bloom_time_ns.load(Ordering::Relaxed)
     }
-}
 
-#[derive(Default, Clone, Debug)]
-pub struct RuntimeFilterStatsSnapshot {
-    pub bloom_time_ns: u64,
-    pub bloom_rows_filtered: u64,
-    pub inlist_min_max_time_ns: u64,
-    pub min_max_rows_filtered: u64,
-    pub min_max_partitions_pruned: u64,
-}
+    pub fn bloom_rows_filtered(&self) -> u64 {
+        self.bloom_rows_filtered.load(Ordering::Relaxed)
+    }
+
+    pub fn inlist_min_max_time_ns(&self) -> u64 {
+        self.inlist_min_max_time_ns.load(Ordering::Relaxed)
+    }
 
-#[derive(Clone, Debug)]
-pub struct RuntimeFilterReport {
-    pub filter_id: usize,
-    pub has_bloom: bool,
-    pub has_inlist: bool,
-    pub has_min_max: bool,
-    pub stats: RuntimeFilterStatsSnapshot,
+    pub fn min_max_rows_filtered(&self) -> u64 {
+        self.min_max_rows_filtered.load(Ordering::Relaxed)
+    }
+
+    pub fn min_max_partitions_pruned(&self) -> u64 {
+        self.min_max_partitions_pruned.load(Ordering::Relaxed)
+    }
 }
 
-pub struct RuntimeFilterReady {
-    pub runtime_filter_watcher: Sender<Option<()>>,
-    /// A dummy receiver to make runtime_filter_watcher channel open.
-    pub _runtime_filter_dummy_receiver: Receiver<Option<()>>,
+pub struct RuntimeFilterChannel {
+    pub sender: Sender<Option<Vec<RuntimeFilterEntry>>>,
+    /// A dummy receiver to make sender channel open.
+    pub _guard: Receiver<Option<Vec<RuntimeFilterEntry>>>,
 }
 
-impl Default for RuntimeFilterReady {
+impl Default for RuntimeFilterChannel {
     fn default() -> Self {
         let (watcher, dummy_receiver) = watch::channel(None);
         Self {
-            runtime_filter_watcher: watcher,
-            _runtime_filter_dummy_receiver: dummy_receiver,
+            sender: watcher,
+            _guard: dummy_receiver,
+        }
+    }
+}
+
+impl RuntimeFilterChannel {
+    pub async fn wait_ready(&self) -> Result<Vec<RuntimeFilterEntry>> {
+        let mut rx = self.sender.subscribe();
+        if let Some(entries) = (*rx.borrow()).clone() {
+            return Ok(entries);
+        }
+
+        match rx.changed().await {
+            Ok(()) => {}
+            Err(e) => {
+                return Err(ErrorCode::TokioError(format!(
+                    "runtime filter channel changed error: {}",
+                    e
+                )));
+            }
+        }
+        let entries = (*rx.borrow()).clone().unwrap_or_default();
+        Ok(entries)
+    }
+}
+
+#[derive(Default, Clone)]
+pub struct RuntimeFilterChannels {
+    pub channels: HashMap<usize, Arc<RuntimeFilterChannel>>,
+}
+
+impl RuntimeFilterChannels {
+    pub fn register(&mut self, scan_id: usize, channel: Arc<RuntimeFilterChannel>) {
+        self.channels.insert(scan_id, channel);
+    }
+
+    pub fn publish_all(&self, filters: &HashMap<usize, RuntimeFilterInfo>) -> Result<()> {
+        for (scan_id, channel) in &self.channels {
+            let entries = filters
+                .get(scan_id)
+                .map(|info| info.filters.clone())
+                .unwrap_or_default();
+            channel.sender.send(Some(entries.clone())).map_err(|_| {
+                ErrorCode::TokioError(format!(
+                    "runtime filter channel closed for scan_id {}",
+                    scan_id
+                ))
+            })?;
         }
+        Ok(())
     }
 }

src/query/catalog/src/table_context.rs

@@ -31,7 +31,6 @@ use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_exception::ResultExt;
 use databend_common_expression::BlockThresholds;
-use databend_common_expression::Expr;
 use databend_common_expression::FunctionContext;
 use databend_common_expression::Scalar;
 use databend_common_expression::TableSchema;
@@ -67,7 +66,6 @@ use databend_storages_common_table_meta::meta::TableMetaTimestamps;
 use databend_storages_common_table_meta::meta::TableSnapshot;
 use parking_lot::Mutex;
 use parking_lot::RwLock;
-use xorf::BinaryFuse16;
 
 use crate::catalog::Catalog;
 use crate::cluster_info::Cluster;
@@ -78,10 +76,9 @@ use crate::plan::PartInfoPtr;
 use crate::plan::PartStatistics;
 use crate::plan::Partitions;
 use crate::query_kind::QueryKind;
-use crate::runtime_filter_info::RuntimeFilterEntry;
+use crate::runtime_filter_info::RuntimeFilterChannel;
 use crate::runtime_filter_info::RuntimeFilterInfo;
-use crate::runtime_filter_info::RuntimeFilterReady;
-use crate::runtime_filter_info::RuntimeFilterReport;
+use crate::runtime_filter_info::RuntimeFilterStats;
 use crate::session_type::SessionType;
 use crate::statistics::data_cache_statistics::DataCacheMetrics;
 use crate::table::Table;
@@ -351,35 +348,10 @@ pub trait TableContext: Send + Sync {
 
     fn get_query_profiles(&self) -> Vec<PlanProfile>;
 
-    fn set_runtime_filter(&self, _filters: HashMap<usize, RuntimeFilterInfo>) {
-        unimplemented!()
-    }
-
-    fn set_runtime_filter_ready(&self, table_index: usize, ready: Arc<RuntimeFilterReady>);
-
-    fn get_runtime_filter_ready(&self, table_index: usize) -> Vec<Arc<RuntimeFilterReady>>;
-
-    fn set_wait_runtime_filter(&self, table_index: usize, need_to_wait: bool);
-
-    fn get_wait_runtime_filter(&self, table_index: usize) -> bool;
-
-    fn clear_runtime_filter(&self);
-
     fn set_merge_into_join(&self, join: MergeIntoJoin);
 
     fn get_merge_into_join(&self) -> MergeIntoJoin;
 
-    fn get_runtime_filters(&self, id: usize) -> Vec<RuntimeFilterEntry>;
-
-    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 runtime_filter_reports(&self) -> HashMap<usize, Vec<RuntimeFilterReport>>;
-
-    fn has_bloom_runtime_filters(&self, id: usize) -> bool;
     fn txn_mgr(&self) -> TxnManagerRef;
     fn get_table_meta_timestamps(
         &self,
@@ -481,6 +453,26 @@ pub trait TableContext: Send + Sync {
     fn merge_pruned_partitions_stats(&self, _other: &HashMap<u32, PartStatistics>) {
         unimplemented!()
     }
+
+    fn register_runtime_filter_stats(&self, _runtime_filters: &HashMap<usize, RuntimeFilterInfo>) {
+        unimplemented!()
+    }
+
+    fn register_runtime_filter_channel(
+        &self,
+        _scan_id: usize,
+        _channel: Arc<RuntimeFilterChannel>,
+    ) {
+        unimplemented!()
+    }
+
+    fn get_runtime_filter_channels(&self, _scan_id: usize) -> Vec<Arc<RuntimeFilterChannel>> {
+        unimplemented!()
+    }
+
+    fn get_runtime_filter_stats(&self) -> Arc<DashMap<usize, Vec<Arc<RuntimeFilterStats>>>> {
+        unimplemented!()
+    }
 }
 
 pub type AbortChecker = Arc<dyn CheckAbort + Send + Sync>;

src/query/service/src/interpreters/interpreter_explain.rs

@@ -180,7 +180,7 @@ impl Interpreter for ExplainInterpreter {
                         profs: HashMap::new(),
                         metadata: &metadata,
                         scan_id_to_runtime_filters: HashMap::new(),
-                        runtime_filter_reports: HashMap::new(),
+                        runtime_filter_stats: Default::default(),
                     };
 
                     let formatter = plan.formatter()?;
@@ -484,7 +484,7 @@ impl ExplainInterpreter {
             plan.set_pruning_stats(&mut pruned_partitions_stats);
         }
 
-        let runtime_filter_reports = self.ctx.runtime_filter_reports();
+        let runtime_filter_stats = self.ctx.get_runtime_filter_stats();
 
         let result = match self.partial {
             true => {
@@ -493,7 +493,7 @@ impl ExplainInterpreter {
                     profs: query_profiles.clone(),
                     metadata: &metadata,
                     scan_id_to_runtime_filters: HashMap::new(),
-                    runtime_filter_reports: runtime_filter_reports.clone(),
+                    runtime_filter_stats: runtime_filter_stats.clone(),
                 };
 
                 let formatter = plan.formatter()?;
@@ -506,7 +506,7 @@ impl ExplainInterpreter {
                     profs: query_profiles.clone(),
                     metadata: &metadata,
                     scan_id_to_runtime_filters: HashMap::new(),
-                    runtime_filter_reports: runtime_filter_reports.clone(),
+                    runtime_filter_stats: runtime_filter_stats.clone(),
                 };
                 let formatter = plan.formatter()?;
                 let format_node = formatter.format(&mut context)?;

src/query/service/src/physical_plans/format/common.rs

@@ -13,12 +13,14 @@
 // limitations under the License.
 
 use std::collections::HashMap;
+use std::sync::Arc;
 
+use dashmap::DashMap;
 use databend_common_ast::ast::FormatTreeNode;
 use databend_common_base::base::format_byte_size;
 use databend_common_base::runtime::profile::get_statistics_desc;
 use databend_common_catalog::plan::PartStatistics;
-use databend_common_catalog::runtime_filter_info::RuntimeFilterReport;
+use databend_common_catalog::runtime_filter_info::RuntimeFilterStats;
 use databend_common_expression::DataSchemaRef;
 use databend_common_sql::executor::physical_plans::AggregateFunctionDesc;
 use databend_common_sql::IndexType;
@@ -33,7 +35,7 @@ pub struct FormatContext<'a> {
     pub metadata: &'a Metadata,
     pub profs: HashMap<u32, PlanProfile>,
     pub scan_id_to_runtime_filters: HashMap<IndexType, Vec<PhysicalRuntimeFilter>>,
-    pub runtime_filter_reports: HashMap<IndexType, Vec<RuntimeFilterReport>>,
+    pub runtime_filter_stats: Arc<DashMap<IndexType, Vec<Arc<RuntimeFilterStats>>>>,
 }
 
 pub fn pretty_display_agg_desc(desc: &AggregateFunctionDesc, metadata: &Metadata) -> String {

src/query/service/src/physical_plans/format/format_table_scan.rs

@@ -125,7 +125,7 @@ impl<'a> PhysicalFormat for TableScanFormatter<'a> {
         // runtime filters
         if let Some(filters) = ctx.scan_id_to_runtime_filters.get(&self.inner.scan_id) {
             if !filters.is_empty() {
-                let reports = ctx.runtime_filter_reports.get(&self.inner.scan_id);
+                let stats = ctx.runtime_filter_stats.get(&self.inner.scan_id);
                 let mut filter_nodes = Vec::new();
                 for filter in filters.iter() {
                     if let Some((probe_key, _)) = filter
@@ -155,34 +155,30 @@ impl<'a> PhysicalFormat for TableScanFormatter<'a> {
                         let mut detail_children =
                             vec![FormatTreeNode::new(format!("type: [{}]", type_text))];
 
-                        if let Some(reports) = reports {
-                            if let Some(report) =
-                                reports.iter().find(|report| report.filter_id == filter.id)
-                            {
-                                if report.has_bloom {
-                                    detail_children.push(FormatTreeNode::new(format!(
-                                        "bloom rows filtered: {}",
-                                        report.stats.bloom_rows_filtered
-                                    )));
-                                    detail_children.push(FormatTreeNode::new(format!(
-                                        "bloom time: {:?}",
-                                        Duration::from_nanos(report.stats.bloom_time_ns)
-                                    )));
-                                }
-                                if report.has_inlist || report.has_min_max {
-                                    detail_children.push(FormatTreeNode::new(format!(
-                                        "inlist/min-max time: {:?}",
-                                        Duration::from_nanos(report.stats.inlist_min_max_time_ns)
-                                    )));
-                                    detail_children.push(FormatTreeNode::new(format!(
-                                        "min-max rows filtered: {}",
-                                        report.stats.min_max_rows_filtered
-                                    )));
-                                    detail_children.push(FormatTreeNode::new(format!(
-                                        "min-max partitions pruned: {}",
-                                        report.stats.min_max_partitions_pruned
-                                    )));
-                                }
+                        if let Some(ref filter_stats) = stats {
+                            // Find the matching stat by filter id
+                            // For now, we just use the first one as stats are grouped by scan_id
+                            if let Some(stat) = filter_stats.first() {
+                                detail_children.push(FormatTreeNode::new(format!(
+                                    "bloom rows filtered: {}",
+                                    stat.bloom_rows_filtered()
+                                )));
+                                detail_children.push(FormatTreeNode::new(format!(
+                                    "bloom time: {:?}",
+                                    Duration::from_nanos(stat.bloom_time_ns())
+                                )));
+                                detail_children.push(FormatTreeNode::new(format!(
+                                    "inlist/min-max time: {:?}",
+                                    Duration::from_nanos(stat.inlist_min_max_time_ns())
+                                )));
+                                detail_children.push(FormatTreeNode::new(format!(
+                                    "min-max rows filtered: {}",
+                                    stat.min_max_rows_filtered()
+                                )));
+                                detail_children.push(FormatTreeNode::new(format!(
+                                    "min-max partitions pruned: {}",
+                                    stat.min_max_partitions_pruned()
+                                )));
                             }
                         }