chore(query): tracking query log for mysql handler (#15495)

* chore(query): tracking query log for mysql handler

* feat(query): tracking query log for mysql handler

Author: zhang2014

Cargo.lock

@@ -43,7 +43,7 @@ pub trait TrySpawn {
     ///
     /// It allows to return an error before spawning the task.
     #[track_caller]
-    fn try_spawn<T>(&self, id: impl Into<String>, task: T) -> Result<JoinHandle<T::Output>>
+    fn try_spawn<T>(&self, task: T) -> Result<JoinHandle<T::Output>>
     where
         T: Future + Send + 'static,
         T::Output: Send + 'static;
@@ -52,32 +52,32 @@ pub trait TrySpawn {
     ///
     /// A default impl of this method just calls `try_spawn` and just panics if there is an error.
     #[track_caller]
-    fn spawn<T>(&self, id: impl Into<String>, task: T) -> JoinHandle<T::Output>
+    fn spawn<T>(&self, task: T) -> JoinHandle<T::Output>
     where
         T: Future + Send + 'static,
         T::Output: Send + 'static,
     {
-        self.try_spawn(id, task).unwrap()
+        self.try_spawn(task).unwrap()
     }
 }
 
 impl<S: TrySpawn> TrySpawn for Arc<S> {
     #[track_caller]
-    fn try_spawn<T>(&self, id: impl Into<String>, task: T) -> Result<JoinHandle<T::Output>>
+    fn try_spawn<T>(&self, task: T) -> Result<JoinHandle<T::Output>>
     where
         T: Future + Send + 'static,
         T::Output: Send + 'static,
     {
-        self.as_ref().try_spawn(id, task)
+        self.as_ref().try_spawn(task)
     }
 
     #[track_caller]
-    fn spawn<T>(&self, id: impl Into<String>, task: T) -> JoinHandle<T::Output>
+    fn spawn<T>(&self, task: T) -> JoinHandle<T::Output>
     where
         T: Future + Send + 'static,
         T::Output: Send + 'static,
     {
-        self.as_ref().spawn(id, task)
+        self.as_ref().spawn(task)
     }
 }
 
@@ -302,19 +302,20 @@ impl Runtime {
 
 impl TrySpawn for Runtime {
     #[track_caller]
-    fn try_spawn<T>(&self, id: impl Into<String>, task: T) -> Result<JoinHandle<T::Output>>
+    fn try_spawn<T>(&self, task: T) -> Result<JoinHandle<T::Output>>
     where
         T: Future + Send + 'static,
         T::Output: Send + 'static,
     {
         let task = ThreadTracker::tracking_future(task);
-        let id = id.into();
-        let task = match id == GLOBAL_TASK {
-            true => async_backtrace::location!(String::from(GLOBAL_TASK_DESC)).frame(task),
-            false => {
-                async_backtrace::location!(format!("Running query {} spawn task", id)).frame(task)
+        let task = match ThreadTracker::query_id() {
+            None => async_backtrace::location!(String::from(GLOBAL_TASK_DESC)).frame(task),
+            Some(query_id) => {
+                async_backtrace::location!(format!("Running query {} spawn task", query_id))
+                    .frame(task)
             }
         };
+
         #[expect(clippy::disallowed_methods)]
         Ok(self.handle.spawn(task))
     }

src/common/base/src/runtime/runtime.rs

@@ -20,7 +20,6 @@ use std::time::Instant;
 
 use databend_common_base::runtime::Runtime;
 use databend_common_base::runtime::TrySpawn;
-use databend_common_base::GLOBAL_TASK;
 use databend_common_exception::Result;
 use rand::distributions::Distribution;
 use rand::distributions::Uniform;
@@ -33,16 +32,16 @@ async fn test_runtime() -> Result<()> {
 
     let runtime = Runtime::with_default_worker_threads()?;
     let runtime_counter = Arc::clone(&counter);
-    let runtime_header = runtime.spawn(GLOBAL_TASK, async move {
+    let runtime_header = runtime.spawn(async move {
         let rt1 = Runtime::with_default_worker_threads().unwrap();
         let rt1_counter = Arc::clone(&runtime_counter);
-        let rt1_header = rt1.spawn(GLOBAL_TASK, async move {
+        let rt1_header = rt1.spawn(async move {
             let rt2 = Runtime::with_worker_threads(1, None).unwrap();
             let rt2_counter = Arc::clone(&rt1_counter);
-            let rt2_header = rt2.spawn(GLOBAL_TASK, async move {
+            let rt2_header = rt2.spawn(async move {
                 let rt3 = Runtime::with_default_worker_threads().unwrap();
                 let rt3_counter = Arc::clone(&rt2_counter);
-                let rt3_header = rt3.spawn(GLOBAL_TASK, async move {
+                let rt3_header = rt3.spawn(async move {
                     let mut num = rt3_counter.lock().unwrap();
                     *num += 1;
                 });
@@ -73,7 +72,7 @@ async fn test_runtime() -> Result<()> {
 async fn test_shutdown_long_run_runtime() -> Result<()> {
     let runtime = Runtime::with_default_worker_threads()?;
 
-    runtime.spawn(GLOBAL_TASK, async move {
+    runtime.spawn(async move {
         tokio::time::sleep(Duration::from_secs(6)).await;
     });
 

src/common/base/tests/it/runtime.rs

@@ -24,7 +24,6 @@ use anyhow::anyhow;
 use databend_common_base::base::GlobalInstance;
 use databend_common_base::runtime::GlobalIORuntime;
 use databend_common_base::runtime::TrySpawn;
-use databend_common_base::GLOBAL_TASK;
 use databend_common_exception::ErrorCode;
 use databend_common_meta_app::storage::StorageAzblobConfig;
 use databend_common_meta_app::storage::StorageCosConfig;
@@ -467,7 +466,7 @@ impl DataOperator {
         // IO hang on reuse connection.
         let op = operator.clone();
         if let Err(cause) = GlobalIORuntime::instance()
-            .spawn(GLOBAL_TASK, async move {
+            .spawn(async move {
                 let res = op.stat("/").await;
                 match res {
                     Ok(_) => Ok(()),

src/common/storage/src/operator.rs

@@ -18,7 +18,6 @@ use std::sync::Arc;
 
 use databend_common_base::runtime::Runtime;
 use databend_common_base::runtime::TrySpawn;
-use databend_common_base::GLOBAL_TASK;
 use futures::Future;
 use opendal::raw::oio;
 use opendal::raw::Access;
@@ -106,7 +105,7 @@ impl<A: Access> LayeredAccess for RuntimeAccessor<A> {
         let op = self.inner.clone();
         let path = path.to_string();
         self.runtime
-            .spawn(GLOBAL_TASK, async move { op.create_dir(&path, args).await })
+            .spawn(async move { op.create_dir(&path, args).await })
             .await
             .expect("join must success")
     }
@@ -117,7 +116,7 @@ impl<A: Access> LayeredAccess for RuntimeAccessor<A> {
         let path = path.to_string();
 
         self.runtime
-            .spawn(GLOBAL_TASK, async move { op.read(&path, args).await })
+            .spawn(async move { op.read(&path, args).await })
             .await
             .expect("join must success")
             .map(|(rp, r)| {
@@ -131,7 +130,7 @@ impl<A: Access> LayeredAccess for RuntimeAccessor<A> {
         let op = self.inner.clone();
         let path = path.to_string();
         self.runtime
-            .spawn(GLOBAL_TASK, async move { op.write(&path, args).await })
+            .spawn(async move { op.write(&path, args).await })
             .await
             .expect("join must success")
     }
@@ -141,7 +140,7 @@ impl<A: Access> LayeredAccess for RuntimeAccessor<A> {
         let op = self.inner.clone();
         let path = path.to_string();
         self.runtime
-            .spawn(GLOBAL_TASK, async move { op.stat(&path, args).await })
+            .spawn(async move { op.stat(&path, args).await })
             .await
             .expect("join must success")
     }
@@ -151,7 +150,7 @@ impl<A: Access> LayeredAccess for RuntimeAccessor<A> {
         let op = self.inner.clone();
         let path = path.to_string();
         self.runtime
-            .spawn(GLOBAL_TASK, async move { op.delete(&path, args).await })
+            .spawn(async move { op.delete(&path, args).await })
             .await
             .expect("join must success")
     }
@@ -161,7 +160,7 @@ impl<A: Access> LayeredAccess for RuntimeAccessor<A> {
         let op = self.inner.clone();
         let path = path.to_string();
         self.runtime
-            .spawn(GLOBAL_TASK, async move { op.list(&path, args).await })
+            .spawn(async move { op.list(&path, args).await })
             .await
             .expect("join must success")
     }
@@ -204,7 +203,7 @@ impl<R: oio::Read> oio::Read for RuntimeIO<R> {
         let runtime = self.runtime.clone();
         async move {
             runtime
-                .spawn(GLOBAL_TASK, async move { r.read_at(offset, limit).await })
+                .spawn(async move { r.read_at(offset, limit).await })
                 .await
                 .expect("join must success")
         }

src/common/storage/src/runtime_layer.rs

@@ -39,7 +39,6 @@ use databend_common_base::runtime::ThreadTracker;
 use databend_common_base::runtime::TrackingPayload;
 use databend_common_base::runtime::TrySpawn;
 use databend_common_base::runtime::UnlimitedFuture;
-use databend_common_base::GLOBAL_TASK;
 use databend_common_grpc::ConnectionFactory;
 use databend_common_grpc::GrpcConnectionError;
 use databend_common_grpc::RpcClientConf;
@@ -440,14 +439,13 @@ impl MetaGrpcClient {
             rt: rt.clone(),
         });
 
-        rt.spawn(
-            GLOBAL_TASK,
-            UnlimitedFuture::create(Self::worker_loop(worker.clone(), rx)),
-        );
-        rt.spawn(
-            GLOBAL_TASK,
-            UnlimitedFuture::create(Self::auto_sync_endpoints(worker, one_tx)),
-        );
+        rt.spawn(UnlimitedFuture::create(Self::worker_loop(
+            worker.clone(),
+            rx,
+        )));
+        rt.spawn(UnlimitedFuture::create(Self::auto_sync_endpoints(
+            worker, one_tx,
+        )));
 
         Ok(handle)
     }

src/meta/client/src/grpc_client.rs

@@ -13,7 +13,6 @@ test = true
 
 [dependencies]
 databend-common-base = { path = "../../../common/base" }
-databend-common-catalog = { path = "../../catalog" }
 databend-common-exception = { path = "../../../common/exception" }
 databend-common-expression = { path = "../../expression" }
 databend-common-pipeline-core = { path = "../core" }

src/query/pipeline/sinks/Cargo.toml

@@ -20,7 +20,6 @@ use async_trait::unboxed_simple;
 use databend_common_base::runtime::drop_guard;
 use databend_common_base::runtime::GlobalIORuntime;
 use databend_common_base::runtime::TrySpawn;
-use databend_common_catalog::table_context::TableContext;
 use databend_common_exception::Result;
 use databend_common_expression::DataBlock;
 use databend_common_pipeline_core::processors::Event;
@@ -53,21 +52,15 @@ pub struct AsyncSinker<T: AsyncSink + 'static> {
     inner: Option<T>,
     finished: bool,
     input: Arc<InputPort>,
-    query_id: String,
     input_data: Option<DataBlock>,
     called_on_start: bool,
     called_on_finish: bool,
 }
 
 impl<T: AsyncSink + 'static> AsyncSinker<T> {
-    pub fn create(
-        input: Arc<InputPort>,
-        ctx: Arc<dyn TableContext>,
-        inner: T,
-    ) -> Box<dyn Processor> {
+    pub fn create(input: Arc<InputPort>, inner: T) -> Box<dyn Processor> {
         Box::new(AsyncSinker {
             input,
-            query_id: ctx.get_id(),
             finished: false,
             input_data: None,
             inner: Some(inner),
@@ -82,7 +75,7 @@ impl<T: AsyncSink + 'static> Drop for AsyncSinker<T> {
         drop_guard(move || {
             if !self.called_on_start || !self.called_on_finish {
                 if let Some(mut inner) = self.inner.take() {
-                    GlobalIORuntime::instance().spawn(self.query_id.clone(), {
+                    GlobalIORuntime::instance().spawn({
                         let called_on_start = self.called_on_start;
                         let called_on_finish = self.called_on_finish;
                         async move {

src/query/pipeline/sinks/src/async_sink.rs

@@ -17,7 +17,6 @@ use std::sync::Arc;
 use async_channel::Sender;
 use async_trait::async_trait;
 use async_trait::unboxed_simple;
-use databend_common_catalog::table_context::TableContext;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::DataBlock;
@@ -32,12 +31,8 @@ pub struct UnionReceiveSink {
 }
 
 impl UnionReceiveSink {
-    pub fn create(
-        sender: Option<Sender<DataBlock>>,
-        input: Arc<InputPort>,
-        ctx: Arc<dyn TableContext>,
-    ) -> Box<dyn Processor> {
-        AsyncSinker::create(input, ctx, UnionReceiveSink { sender })
+    pub fn create(tx: Option<Sender<DataBlock>>, input: Arc<InputPort>) -> Box<dyn Processor> {
+        AsyncSinker::create(input, UnionReceiveSink { sender: tx })
     }
 }
 

src/query/pipeline/sinks/src/union_receive_sink.rs

@@ -117,7 +117,7 @@ pub trait InputFormatPipe: Sized + Send + 'static {
         let (split_tx, split_rx) = async_channel::bounded(ctx.num_prefetch_splits()?);
         Self::build_pipeline_with_aligner(&ctx, split_rx, pipeline)?;
 
-        GlobalIORuntime::instance().spawn(ctx.table_context.get_id(), async move {
+        GlobalIORuntime::instance().spawn(async move {
             let mut sender: Option<Sender<Result<Self::ReadBatch>>> = None;
             while let Some(batch_result) = input.recv().await {
                 match batch_result {
@@ -161,7 +161,7 @@ pub trait InputFormatPipe: Sized + Send + 'static {
         Self::build_pipeline_with_aligner(&ctx, split_rx, pipeline)?;
 
         let ctx_clone = ctx.clone();
-        GlobalIORuntime::instance().spawn(ctx.table_context.get_id(), async move {
+        GlobalIORuntime::instance().spawn(async move {
             debug!("start copy splits feeder");
             for s in &ctx_clone.splits {
                 let (data_tx, data_rx) = tokio::sync::mpsc::channel(ctx.num_prefetch_per_split());