add log

Author: SkyFan2002

Cargo.lock

@@ -15,6 +15,7 @@ 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

@@ -131,7 +131,9 @@ impl<T: 'static + AsyncSource> Processor for AsyncSourcer<T> {
             return Ok(());
         }
         match self.inner.generate().await? {
-            None => self.is_finish = true,
+            None => {
+                self.is_finish = true;
+            }
             Some(data_block) => {
                 if !data_block.is_empty() {
                     let progress_values = ProgressValues {

src/query/pipeline/sources/src/async_source.rs

@@ -54,7 +54,15 @@ impl AsyncSource for RuntimeFilterSourceProcessor {
 
     #[async_backtrace::framed]
     async fn generate(&mut self) -> Result<Option<DataBlock>> {
-        match self.meta_receiver.recv().await {
+        let start = std::time::Instant::now();
+        log::info!("RuntimeFilterSource recv() start");
+        let rf = self.meta_receiver.recv().await;
+        log::info!(
+            "RuntimeFilterSource recv() take {:?},get {}",
+            start.elapsed(),
+            rf.is_ok()
+        );
+        match rf {
             Ok(runtime_filter) => Ok(Some(DataBlock::empty_with_meta(Box::new(
                 RuntimeFilterMeta {
                     inlist: runtime_filter.inlist,