add TransformRuntimeFilterWait

Author: SkyFan2002

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

@@ -42,6 +42,7 @@ use crate::operations::read::native_data_transform_reader::ReadNativeDataTransfo
 use crate::operations::read::parquet_data_transform_reader::ReadParquetDataTransform;
 use crate::operations::read::DeserializeDataTransform;
 use crate::operations::read::NativeDeserializeDataTransform;
+use crate::operations::read::TransformRuntimeFilterWait;
 
 #[allow(clippy::too_many_arguments)]
 pub fn build_fuse_native_source_pipeline(
@@ -59,6 +60,8 @@ pub fn build_fuse_native_source_pipeline(
     (max_threads, max_io_requests) =
         adjust_threads_and_request(true, max_threads, max_io_requests, plan);
 
+    let need_runtime_filter_wait = !ctx.get_runtime_filter_ready(plan.scan_id).is_empty();
+
     if topk.is_some() {
         max_threads = max_threads.min(16);
         max_io_requests = max_io_requests.min(16);
@@ -82,6 +85,16 @@ pub fn build_fuse_native_source_pipeline(
                     pipeline.add_pipe(pipe);
                 }
             }
+            if need_runtime_filter_wait {
+                pipeline.add_transform(|input, output| {
+                    Ok(TransformRuntimeFilterWait::create(
+                        ctx.clone(),
+                        plan.scan_id,
+                        input,
+                        output,
+                    ))
+                })?;
+            }
             pipeline.add_transform(|input, output| {
                 ReadNativeDataTransform::<true>::create(
                     plan.scan_id,
@@ -118,6 +131,17 @@ pub fn build_fuse_native_source_pipeline(
                 }
             }
 
+            if need_runtime_filter_wait {
+                pipeline.add_transform(|input, output| {
+                    Ok(TransformRuntimeFilterWait::create(
+                        ctx.clone(),
+                        plan.scan_id,
+                        input,
+                        output,
+                    ))
+                })?;
+            }
+
             pipeline.add_transform(|input, output| {
                 ReadNativeDataTransform::<false>::create(
                     plan.scan_id,
@@ -171,6 +195,7 @@ pub fn build_fuse_parquet_source_pipeline(
         blocks_total: AtomicU64::new(0),
         blocks_pruned: AtomicU64::new(0),
     });
+    let need_runtime_filter_wait = !ctx.get_runtime_filter_ready(plan.scan_id).is_empty();
 
     match block_reader.support_blocking_api() {
         true => {
@@ -190,6 +215,17 @@ pub fn build_fuse_parquet_source_pipeline(
             let unfinished_processors_count =
                 Arc::new(AtomicU64::new(pipeline.output_len() as u64));
 
+            if need_runtime_filter_wait {
+                pipeline.add_transform(|input, output| {
+                    Ok(TransformRuntimeFilterWait::create(
+                        ctx.clone(),
+                        plan.scan_id,
+                        input,
+                        output,
+                    ))
+                })?;
+            }
+
             pipeline.add_transform(|input, output| {
                 ReadParquetDataTransform::<true>::create(
                     plan.scan_id,
@@ -233,6 +269,17 @@ pub fn build_fuse_parquet_source_pipeline(
             let unfinished_processors_count =
                 Arc::new(AtomicU64::new(pipeline.output_len() as u64));
 
+            if need_runtime_filter_wait {
+                pipeline.add_transform(|input, output| {
+                    Ok(TransformRuntimeFilterWait::create(
+                        ctx.clone(),
+                        plan.scan_id,
+                        input,
+                        output,
+                    ))
+                })?;
+            }
+
             pipeline.add_transform(|input, output| {
                 ReadParquetDataTransform::<false>::create(
                     plan.table_index,

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

@@ -21,6 +21,7 @@ mod parquet_data_source;
 mod parquet_data_source_deserializer;
 mod parquet_data_transform_reader;
 mod parquet_rows_fetcher;
+mod runtime_filter_wait;
 
 mod block_partition_meta;
 mod block_partition_receiver_source;
@@ -32,4 +33,5 @@ pub use fuse_rows_fetcher::row_fetch_processor;
 pub use fuse_source::build_fuse_parquet_source_pipeline;
 pub use native_data_source_deserializer::NativeDeserializeDataTransform;
 pub use parquet_data_source_deserializer::DeserializeDataTransform;
+pub use runtime_filter_wait::TransformRuntimeFilterWait;
 pub use util::need_reserve_block_info;

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

@@ -28,10 +28,8 @@ use databend_common_catalog::plan::PartInfoPtr;
 use databend_common_catalog::plan::PushDownInfo;
 use databend_common_catalog::plan::TopK;
 use databend_common_catalog::runtime_filter_info::RuntimeFilterEntry;
-use databend_common_catalog::runtime_filter_info::RuntimeFilterReady;
 use databend_common_catalog::runtime_filter_info::RuntimeFilterStats;
 use databend_common_catalog::table_context::TableContext;
-use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::filter_helper::FilterHelpers;
 use databend_common_expression::types::BooleanType;
@@ -217,8 +215,6 @@ pub struct NativeDeserializeDataTransform {
     // Structures for the bloom runtime filter:
     ctx: Arc<dyn TableContext>,
     bloom_runtime_filter: Option<Vec<BloomRuntimeFilterRef>>,
-    need_wait_runtime_filter: bool,
-    runtime_filter_ready: Vec<Arc<RuntimeFilterReady>>,
 
     // Structures for aggregating index:
     index_reader: Arc<Option<AggIndexReader>>,
@@ -311,9 +307,6 @@ impl NativeDeserializeDataTransform {
         let mut output_schema = plan.schema().as_ref().clone();
         output_schema.remove_internal_fields();
         let output_schema: DataSchema = (&output_schema).into();
-        let need_wait_runtime_filter =
-            !ctx.get_cluster().is_empty() && ctx.get_wait_runtime_filter(plan.scan_id);
-
         Ok(ProcessorPtr::create(Box::new(
             NativeDeserializeDataTransform {
                 ctx,
@@ -339,8 +332,6 @@ impl NativeDeserializeDataTransform {
                 bloom_runtime_filter: None,
                 read_state: ReadPartState::new(),
                 need_reserve_block_info,
-                need_wait_runtime_filter,
-                runtime_filter_ready: Vec::new(),
             },
         )))
     }
@@ -781,20 +772,6 @@ impl NativeDeserializeDataTransform {
         }
     }
 
-    fn prepare_runtime_filter_wait(&mut self) -> bool {
-        if !self.need_wait_runtime_filter {
-            return false;
-        }
-        self.need_wait_runtime_filter = false;
-        let runtime_filter_ready = self.ctx.get_runtime_filter_ready(self.scan_id);
-        if !runtime_filter_ready.is_empty() {
-            self.runtime_filter_ready = runtime_filter_ready;
-            true
-        } else {
-            false
-        }
-    }
-
     /// Pre-process the partition before reading it.
     fn pre_process_partition(&mut self) -> Result<()> {
         debug_assert!(!self.columns.is_empty());
@@ -880,10 +857,6 @@ impl Processor for NativeDeserializeDataTransform {
     }
 
     fn event(&mut self) -> Result<Event> {
-        if self.prepare_runtime_filter_wait() {
-            return Ok(Event::Async);
-        }
-
         if self.output.is_finished() {
             self.input.finish();
             return Ok(Event::Finished);
@@ -931,33 +904,6 @@ impl Processor for NativeDeserializeDataTransform {
 
     #[async_backtrace::framed]
     async fn async_process(&mut self) -> Result<()> {
-        use std::time::Duration;
-
-        use databend_common_base::base::tokio::time::timeout;
-
-        let timeout_duration = Duration::from_secs(30);
-
-        for runtime_filter_ready in &self.runtime_filter_ready {
-            let mut rx = runtime_filter_ready.runtime_filter_watcher.subscribe();
-            if (*rx.borrow()).is_some() {
-                continue;
-            }
-
-            match timeout(timeout_duration, rx.changed()).await {
-                Ok(Ok(())) => {}
-                Ok(Err(_)) => {
-                    return Err(ErrorCode::TokioError("watcher's sender is dropped"));
-                }
-                Err(_) => {
-                    log::warn!(
-                        "Runtime filter wait timeout after {:?} for scan_id: {}",
-                        timeout_duration,
-                        self.scan_id
-                    );
-                }
-            }
-        }
-        self.runtime_filter_ready.clear();
         Ok(())
     }
 

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

@@ -24,10 +24,8 @@ use databend_common_base::runtime::profile::ProfileStatisticsName;
 use databend_common_catalog::plan::DataSourcePlan;
 use databend_common_catalog::plan::PartInfoPtr;
 use databend_common_catalog::runtime_filter_info::RuntimeFilterEntry;
-use databend_common_catalog::runtime_filter_info::RuntimeFilterReady;
 use databend_common_catalog::runtime_filter_info::RuntimeFilterStats;
 use databend_common_catalog::table_context::TableContext;
-use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::types::Bitmap;
 use databend_common_expression::types::DataType;
@@ -44,7 +42,6 @@ use databend_common_pipeline_core::processors::InputPort;
 use databend_common_pipeline_core::processors::OutputPort;
 use databend_common_pipeline_core::processors::Processor;
 use databend_common_pipeline_core::processors::ProcessorPtr;
-use databend_common_sql::IndexType;
 use xorf::BinaryFuse16;
 
 use super::parquet_data_source::ParquetDataSource;
@@ -77,8 +74,6 @@ pub struct DeserializeDataTransform {
     base_block_ids: Option<Scalar>,
     cached_runtime_filter: Option<Vec<BloomRuntimeFilterRef>>,
     need_reserve_block_info: bool,
-    already_waited_runtime_filter: bool,
-    runtime_filter_ready: Vec<Arc<RuntimeFilterReady>>,
 }
 
 #[derive(Clone)]
@@ -136,8 +131,6 @@ impl DeserializeDataTransform {
             base_block_ids: plan.base_block_ids.clone(),
             cached_runtime_filter: None,
             need_reserve_block_info,
-            already_waited_runtime_filter: false,
-            runtime_filter_ready: ctx.get_runtime_filter_ready(plan.scan_id),
         })))
     }
 
@@ -192,14 +185,6 @@ impl DeserializeDataTransform {
             Ok(None)
         }
     }
-
-    fn need_wait_runtime_filter(&mut self) -> bool {
-        if self.already_waited_runtime_filter {
-            return false;
-        }
-        self.already_waited_runtime_filter = true;
-        !self.runtime_filter_ready.is_empty()
-    }
 }
 
 #[async_trait::async_trait]
@@ -213,10 +198,6 @@ impl Processor for DeserializeDataTransform {
     }
 
     fn event(&mut self) -> Result<Event> {
-        if self.need_wait_runtime_filter() {
-            return Ok(Event::Async);
-        }
-
         if self.output.is_finished() {
             self.input.finish();
             return Ok(Event::Finished);
@@ -374,35 +355,4 @@ impl Processor for DeserializeDataTransform {
 
         Ok(())
     }
-
-    #[async_backtrace::framed]
-    async fn async_process(&mut self) -> Result<()> {
-        use std::time::Duration;
-
-        use databend_common_base::base::tokio::time::timeout;
-
-        let timeout_duration = Duration::from_secs(30);
-
-        for runtime_filter_ready in &self.runtime_filter_ready {
-            let mut rx = runtime_filter_ready.runtime_filter_watcher.subscribe();
-            if (*rx.borrow()).is_some() {
-                continue;
-            }
-
-            match timeout(timeout_duration, rx.changed()).await {
-                Ok(Ok(())) => {}
-                Ok(Err(_)) => {
-                    return Err(ErrorCode::TokioError("watcher's sender is dropped"));
-                }
-                Err(_) => {
-                    log::warn!(
-                        "Runtime filter wait timeout after {:?} for scan_id: {}",
-                        timeout_duration,
-                        self.scan_id
-                    );
-                }
-            }
-        }
-        Ok(())
-    }
 }