Merge pull request #9244 from sandflee/log2

chore(executor): add query id in slow log

Author: zhang2014

src/query/service/src/api/rpc/exchange/exchange_manager.rs

@@ -552,7 +552,9 @@ impl QueryCoordinator {
             }
         }
 
-        let executor_settings = ExecutorSettings::try_create(&info.query_ctx.get_settings())?;
+        let query_id = info.query_ctx.get_id();
+        let executor_settings =
+            ExecutorSettings::try_create(&info.query_ctx.get_settings(), query_id)?;
 
         let executor = PipelineCompleteExecutor::from_pipelines(pipelines, executor_settings)?;
 

src/query/service/src/interpreters/interpreter.rs

@@ -85,8 +85,9 @@ pub trait Interpreter: Sync + Send {
         });
 
         let settings = ctx.get_settings();
+        let query_id = ctx.get_id();
         build_res.set_max_threads(settings.get_max_threads()? as usize);
-        let settings = ExecutorSettings::try_create(&settings)?;
+        let settings = ExecutorSettings::try_create(&settings, query_id)?;
 
         if build_res.main_pipeline.is_complete_pipeline()? {
             let mut pipelines = build_res.sources_pipelines;

src/query/service/src/interpreters/interpreter_delete.rs

@@ -76,7 +76,8 @@ impl Interpreter for DeleteInterpreter {
         if !pipeline.pipes.is_empty() {
             let settings = self.ctx.get_settings();
             pipeline.set_max_threads(settings.get_max_threads()? as usize);
-            let executor_settings = ExecutorSettings::try_create(&settings)?;
+            let query_id = self.ctx.get_id();
+            let executor_settings = ExecutorSettings::try_create(&settings, query_id)?;
             let executor = PipelineCompleteExecutor::try_create(pipeline, executor_settings)?;
 
             self.ctx.set_executor(Arc::downgrade(&executor.get_inner()));

src/query/service/src/interpreters/interpreter_table_optimize.rs

@@ -93,7 +93,8 @@ impl Interpreter for OptimizeTableInterpreter {
             {
                 let settings = ctx.get_settings();
                 pipeline.set_max_threads(settings.get_max_threads()? as usize);
-                let executor_settings = ExecutorSettings::try_create(&settings)?;
+                let query_id = ctx.get_id();
+                let executor_settings = ExecutorSettings::try_create(&settings, query_id)?;
                 let executor = PipelineCompleteExecutor::try_create(pipeline, executor_settings)?;
 
                 ctx.set_executor(Arc::downgrade(&executor.get_inner()));

src/query/service/src/interpreters/interpreter_table_recluster.rs

@@ -85,7 +85,8 @@ impl Interpreter for ReclusterTableInterpreter {
 
             pipeline.set_max_threads(settings.get_max_threads()? as usize);
 
-            let executor_settings = ExecutorSettings::try_create(&settings)?;
+            let query_id = ctx.get_id();
+            let executor_settings = ExecutorSettings::try_create(&settings, query_id)?;
             let executor = PipelineCompleteExecutor::try_create(pipeline, executor_settings)?;
 
             ctx.set_executor(Arc::downgrade(&executor.get_inner()));

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

@@ -12,19 +12,22 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::sync::Arc;
 use std::time::Duration;
 
 use common_exception::Result;
 use common_settings::Settings;
 
 pub struct ExecutorSettings {
+    pub query_id: Arc<String>,
     pub max_execute_time: Duration,
 }
 
 impl ExecutorSettings {
-    pub fn try_create(settings: &Settings) -> Result<ExecutorSettings> {
+    pub fn try_create(settings: &Settings, query_id: String) -> Result<ExecutorSettings> {
         let max_execute_time = settings.get_max_execute_time()?;
         Ok(ExecutorSettings {
+            query_id: Arc::new(query_id),
             max_execute_time: Duration::from_millis(max_execute_time),
         })
     }

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

@@ -38,14 +38,20 @@ pub enum ExecutorTask {
 }
 
 pub struct ExecutorWorkerContext {
+    query_id: Arc<String>,
     worker_num: usize,
     task: ExecutorTask,
     workers_condvar: Arc<WorkersCondvar>,
 }
 
 impl ExecutorWorkerContext {
-    pub fn create(worker_num: usize, workers_condvar: Arc<WorkersCondvar>) -> Self {
+    pub fn create(
+        worker_num: usize,
+        workers_condvar: Arc<WorkersCondvar>,
+        query_id: Arc<String>,
+    ) -> Self {
         ExecutorWorkerContext {
+            query_id,
             worker_num,
             workers_condvar,
             task: ExecutorTask::None,
@@ -97,6 +103,7 @@ impl ExecutorWorkerContext {
         executor.async_runtime.spawn(TrackedFuture::create(
             ThreadTracker::fork(),
             ProcessorAsyncTask::create(
+                self.query_id.clone(),
                 worker_id,
                 processor.clone(),
                 tasks_queue,

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

@@ -284,7 +284,11 @@ impl PipelineExecutor {
     /// Method is thread unsafe and require thread safe call
     pub unsafe fn execute_single_thread(&self, thread_num: usize) -> Result<()> {
         let workers_condvar = self.workers_condvar.clone();
-        let mut context = ExecutorWorkerContext::create(thread_num, workers_condvar);
+        let mut context = ExecutorWorkerContext::create(
+            thread_num,
+            workers_condvar,
+            self.settings.query_id.clone(),
+        );
 
         while !self.global_tasks_queue.is_finished() {
             // When there are not enough tasks, the thread will be blocked, so we need loop check.

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

@@ -42,6 +42,7 @@ pub struct ProcessorAsyncTask {
 
 impl ProcessorAsyncTask {
     pub fn create<Inner: Future<Output = Result<()>> + Send + 'static>(
+        query_id: Arc<String>,
         worker_id: usize,
         processor: ProcessorPtr,
         queue: Arc<ExecutorTasksQueue>,
@@ -73,7 +74,8 @@ impl ProcessorAsyncTask {
                         Either::Left((_, right)) => {
                             inner = right;
                             tracing::warn!(
-                                "Very slow processor async task, processor id: {:?}, name: {:?}, elapsed: {:?}",
+                                "Very slow processor async task, query_id:{:?}, processor id: {:?}, name: {:?}, elapsed: {:?}",
+                                query_id,
                                 wraning_processor.id(),
                                 wraning_processor.name(),
                                 start.elapsed()

src/query/service/src/servers/http/v1/query/execute_state.rs

@@ -342,7 +342,8 @@ impl HttpQueryHandle {
         });
 
         let query_ctx = ctx.clone();
-        let executor_settings = ExecutorSettings::try_create(&ctx.get_settings())?;
+        let query_id = ctx.get_id();
+        let executor_settings = ExecutorSettings::try_create(&ctx.get_settings(), query_id)?;
 
         let run = move || -> Result<()> {
             let mut pipelines = build_res.sources_pipelines;