chore(query): dump slow async running graph separately (#19926)

* chore(query): dump slow async running graph separately

* chore(query): add exchange sender details status

Author: dqhl76

src/common/tracing/src/init.rs

@@ -128,6 +128,7 @@ fn env_filter(level: &str) -> EnvFilter {
             .filter(Some("databend::log::structlog"), LevelFilter::Off)
             .filter(Some("databend::log::time_series"), LevelFilter::Off)
             .filter(Some("databend::log::access"), LevelFilter::Off)
+            .filter(Some("databend::log::dump_graph"), LevelFilter::Off)
             .parse(level),
     )
 }
@@ -462,3 +463,35 @@ pub fn init_logging(
 
     _drop_guards
 }
+
+#[cfg(test)]
+mod tests {
+    use log::Level;
+    use log::Metadata;
+    use logforth::Filter;
+    use logforth::filter::FilterResult;
+
+    use super::env_filter;
+
+    #[test]
+    fn test_dump_graph_filter_is_off_by_default() {
+        let filter = env_filter("TRACE");
+        let metadata = Metadata::builder()
+            .target("databend::log::dump_graph")
+            .level(Level::Trace)
+            .build();
+
+        assert_eq!(filter.enabled(&metadata), FilterResult::Reject);
+    }
+
+    #[test]
+    fn test_dump_graph_filter_can_be_enabled_by_level() {
+        let filter = env_filter("WARN,databend::log::dump_graph=TRACE");
+        let metadata = Metadata::builder()
+            .target("databend::log::dump_graph")
+            .level(Level::Trace)
+            .build();
+
+        assert_eq!(filter.enabled(&metadata), FilterResult::Neutral);
+    }
+}

src/query/service/src/pipelines/executor/processor_async_task.rs

@@ -31,6 +31,7 @@ use databend_common_pipeline::core::ProcessorPtr;
 use futures_util::FutureExt;
 use futures_util::future::BoxFuture;
 use futures_util::future::Either;
+use log::error;
 use log::warn;
 use petgraph::prelude::NodeIndex;
 use tokio::time::sleep;
@@ -41,6 +42,8 @@ use crate::pipelines::executor::QueryExecutorTasksQueue;
 use crate::pipelines::executor::RunningGraph;
 use crate::pipelines::executor::WorkersCondvar;
 
+const TARGET_DUMP_GRAPH: &str = "databend::log::dump_graph";
+
 pub enum ExecutorTasksQueue {
     QueryExecutorTasksQueue(Arc<QueryExecutorTasksQueue>),
     QueriesExecutorTasksQueue(Arc<QueriesExecutorTasksQueue>),
@@ -124,9 +127,11 @@ impl ProcessorAsyncTask {
         let processor_id = unsafe { processor.id() };
         let processor_name = unsafe { processor.name() };
         let queue_clone = queue.clone();
+        let graph_clone = graph.clone();
         let inner = async move {
             let start = Instant::now();
             let mut inner = inner.boxed();
+            let mut log_graph = false;
 
             loop {
                 let interval = Box::pin(sleep(Duration::from_secs(5)));
@@ -139,6 +144,20 @@ impl ProcessorAsyncTask {
                             "Slow async task detected - query: {:?}, processor: {:?} ({}), elapsed: {:?}, active workers: {:?}",
                             query_id, processor_id, processor_name, elapsed, active_workers
                         );
+                        if elapsed >= Duration::from_secs(200) && active_workers == 0 && !log_graph
+                        {
+                            log_graph = true;
+                            error!(
+                                target: TARGET_DUMP_GRAPH,
+                                "Slow async task running graph dump - query: {:?}, processor: {:?} ({}), elapsed: {:?}, active workers: {:?}, {}",
+                                query_id,
+                                processor_id,
+                                processor_name,
+                                elapsed,
+                                active_workers,
+                                graph_clone.format_graph_nodes(false)
+                            );
+                        }
                     }
                     Either::Right((res, _)) => {
                         return res;

src/query/service/src/servers/flight/v1/exchange/broadcast_send_transform.rs

@@ -170,12 +170,12 @@ impl Processor for BroadcastSendTransform {
 
     fn details_status(&self) -> Option<String> {
         Some(format!(
-            "BroadcastSendTransform {} {:?}",
+            "handle_pending={}, local_pos={}, closed_channels={}/{}, closed={:?}",
             self.handle.is_some(),
-            self.channels
-                .iter()
-                .map(|(_, x)| x.is_closed())
-                .collect::<Vec<_>>()
+            self.local_pos,
+            self.channels.closed_count(),
+            self.channels.len(),
+            self.channels.closed_status(),
         ))
     }
 

src/query/service/src/servers/flight/v1/exchange/hash_send_sink.rs

@@ -187,6 +187,17 @@ impl Processor for HashSendSink {
         Ok(Event::NeedData)
     }
 
+    fn details_status(&self) -> Option<String> {
+        Some(format!(
+            "handle_pending={}, closed_channels={}/{}, closed={:?}, buffered_partitions={:?}",
+            self.handle.is_some(),
+            self.channels.closed_count(),
+            self.channels.len(),
+            self.channels.closed_status(),
+            self.partition_stream.partition_ids(),
+        ))
+    }
+
     fn set_id(&mut self, id: NodeIndex) {
         self.id = id;
     }

src/query/service/src/servers/flight/v1/exchange/hash_send_transform.rs

@@ -223,6 +223,18 @@ impl Processor for HashSendTransform {
         Ok(Event::NeedData)
     }
 
+    fn details_status(&self) -> Option<String> {
+        Some(format!(
+            "handle_pending={}, local_pos={}, closed_channels={}/{}, closed={:?}, buffered_partitions={:?}",
+            self.handle.is_some(),
+            self.local_pos,
+            self.channels.closed_count(),
+            self.channels.len(),
+            self.channels.closed_status(),
+            self.partition_stream.partition_ids(),
+        ))
+    }
+
     fn set_id(&mut self, id: NodeIndex) {
         self.id = id;
     }

src/query/service/src/servers/flight/v1/exchange/outbound_send_channels.rs

@@ -61,6 +61,14 @@ impl OutboundSendChannels {
             .all(|(idx, ch)| idx == except_idx || ch.is_closed())
     }
 
+    pub(super) fn closed_status(&self) -> Vec<bool> {
+        self.channels.iter().map(|ch| ch.is_closed()).collect()
+    }
+
+    pub(super) fn closed_count(&self) -> usize {
+        self.channels.iter().filter(|ch| ch.is_closed()).count()
+    }
+
     pub(super) fn close(&mut self, idx: usize) {
         if !self.channels[idx].is_closed() {
             let mut closed = DummyOutboundChannel::create();