chore(query): refine the log and error message (#17993)

* chore(query): refine the log and error message

* fix: 00_0004_copy_with_max_files.result

* fix:tests/it/pipelines/executor/pipeline_executor.rs

* refine the query_ctx.rs error message

Author: BohuTANG

src/query/ee/src/storages/fuse/operations/vacuum_table_v2.rs

@@ -98,7 +98,10 @@ pub async fn do_vacuum2(
 
     let snapshots_before_lvt = match retention_policy {
         RetentionPolicy::ByTimePeriod(delta_duration) => {
-            info!("using by ByTimePeriod policy {:?}", delta_duration);
+            info!(
+                "[FUSE-VACUUM2] Using ByTimePeriod policy {:?}",
+                delta_duration
+            );
             let retention_period = if fuse_table.is_transient() {
                 // For transient table, keep no history data
                 TimeDelta::zero()
@@ -118,7 +121,7 @@ pub async fn do_vacuum2(
             }
 
             ctx.set_status_info(&format!(
-                "set lvt for table {} takes {:?}, lvt: {:?}",
+                "[FUSE-VACUUM2] Set LVT for table {}, elapsed: {:?}, LVT: {:?}",
                 fuse_table.get_table_info().desc,
                 start.elapsed(),
                 lvt
@@ -130,7 +133,7 @@ pub async fn do_vacuum2(
         }
         RetentionPolicy::ByNumOfSnapshotsToKeep(num_snapshots_to_keep) => {
             info!(
-                "using by ByNumOfSnapshotsToKeep policy {:?}",
+                "[FUSE-VACUUM2] Using ByNumOfSnapshotsToKeep policy {:?}",
                 num_snapshots_to_keep
             );
             // List the snapshot order by timestamp asc, till the current snapshot(inclusively).
@@ -171,7 +174,7 @@ pub async fn do_vacuum2(
 
     let elapsed = start.elapsed();
     ctx.set_status_info(&format!(
-        "list snapshots for table {} takes {:?}, snapshots_dir: {:?}, snapshots: {:?}",
+        "[FUSE-VACUUM2] Listed snapshots for table {}, elapsed: {:?}, snapshots_dir: {:?}, snapshots: {:?}",
         fuse_table.get_table_info().desc,
         elapsed,
         fuse_table
@@ -193,7 +196,7 @@ pub async fn do_vacuum2(
         return Ok(vec![]);
     };
     ctx.set_status_info(&format!(
-        "select gc_root for table {} takes {:?}, gc_root: {:?}, snapshots_to_gc: {:?}",
+        "[FUSE-VACUUM2] Selected gc_root for table {}, elapsed: {:?}, gc_root: {:?}, snapshots_to_gc: {:?}",
         fuse_table.get_table_info().desc,
         start.elapsed(),
         gc_root,
@@ -222,7 +225,7 @@ pub async fn do_vacuum2(
     .collect::<Vec<_>>();
 
     ctx.set_status_info(&format!(
-        "list segments before gc_root for table {} takes {:?}, segment_dir: {:?}, gc_root_timestamp: {:?}, segments: {:?}",
+        "[FUSE-VACUUM2] Listed segments before gc_root for table {}, elapsed: {:?}, segment_dir: {:?}, gc_root_timestamp: {:?}, segments: {:?}",
         fuse_table.get_table_info().desc,
         start.elapsed(),
         fuse_table.meta_location_generator().segment_location_prefix(),
@@ -236,7 +239,7 @@ pub async fn do_vacuum2(
         .filter(|s| !gc_root_segments.contains(s))
         .collect();
     ctx.set_status_info(&format!(
-        "Filter segments to gc for table {} takes {:?}, segments_to_gc: {:?}",
+        "[FUSE-VACUUM2] Filtered segments_to_gc for table {}, elapsed: {:?}, segments_to_gc: {:?}",
         fuse_table.get_table_info().desc,
         start.elapsed(),
         slice_summary(&segments_to_gc)
@@ -253,7 +256,7 @@ pub async fn do_vacuum2(
         gc_root_blocks.extend(segment?.block_metas()?.iter().map(|b| b.location.0.clone()));
     }
     ctx.set_status_info(&format!(
-        "read segments for table {} takes {:?}",
+        "[FUSE-VACUUM2] Read segments for table {}, elapsed: {:?}",
         fuse_table.get_table_info().desc,
         start.elapsed(),
     ));
@@ -272,7 +275,7 @@ pub async fn do_vacuum2(
     .collect::<Vec<_>>();
 
     ctx.set_status_info(&format!(
-        "list blocks before gc_root for table {} takes {:?}, block_dir: {:?}, least_visible_timestamp: {:?}, blocks: {:?}",
+        "[FUSE-VACUUM2] Listed blocks before gc_root for table {}, elapsed: {:?}, block_dir: {:?}, gc_root_timestamp: {:?}, blocks: {:?}",
         fuse_table.get_table_info().desc,
         start.elapsed(),
         fuse_table.meta_location_generator().block_location_prefix(),
@@ -286,7 +289,7 @@ pub async fn do_vacuum2(
         .filter(|b| !gc_root_blocks.contains(b))
         .collect();
     ctx.set_status_info(&format!(
-        "Filter blocks to gc for table {} takes {:?}, blocks_to_gc: {:?}",
+        "[FUSE-VACUUM2] Filtered blocks_to_gc for table {}, elapsed: {:?}, blocks_to_gc: {:?}",
         fuse_table.get_table_info().desc,
         start.elapsed(),
         slice_summary(&blocks_to_gc)
@@ -326,7 +329,7 @@ pub async fn do_vacuum2(
     }
 
     ctx.set_status_info(&format!(
-        "collect indexes to gc for table {} takes {:?}, indexes_to_gc: {:?}",
+        "[FUSE-VACUUM2] Collected indexes_to_gc for table {}, elapsed: {:?}, indexes_to_gc: {:?}",
         fuse_table.get_table_info().desc,
         start.elapsed(),
         slice_summary(&indexes_to_gc)
@@ -369,7 +372,7 @@ pub async fn do_vacuum2(
         .chain(indexes_to_gc.into_iter())
         .collect();
     ctx.set_status_info(&format!(
-        "remove files for table {} takes {:?}, files_to_gc: {:?}",
+        "[FUSE-VACUUM2] Removed files for table {}, elapsed: {:?}, files_to_gc: {:?}",
         fuse_table.get_table_info().desc,
         start.elapsed(),
         slice_summary(&files_to_gc)
@@ -419,15 +422,15 @@ async fn set_lvt(
 ) -> Result<Option<DateTime<Utc>>> {
     let Some(latest_snapshot) = fuse_table.read_table_snapshot().await? else {
         info!(
-            "Table {} has no snapshot, stop vacuuming",
+            "[FUSE-VACUUM2] Table {} has no snapshot, stopping vacuum",
             fuse_table.get_table_info().desc
         );
         return Ok(None);
     };
     if !is_uuid_v7(&latest_snapshot.snapshot_id) {
         info!(
-            "latest snapshot {:?} is not v7, stop vacuuming",
-            latest_snapshot
+            "[FUSE-VACUUM2] Latest snapshot is not v7, stopping vacuum: {:?}",
+            latest_snapshot.snapshot_id
         );
         return Ok(None);
     }
@@ -458,7 +461,7 @@ async fn list_until_prefix(
     need_one_more: bool,
     gc_root_meta_ts: Option<DateTime<Utc>>,
 ) -> Result<Vec<Entry>> {
-    info!("list until prefix: {}", until);
+    info!("[FUSE-VACUUM2] Listing files until prefix: {}", until);
     let dal = fuse_table.get_operator_ref();
 
     match dal.info().scheme() {

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

@@ -101,7 +101,7 @@ pub trait Interpreter: Sync + Send {
     async fn execute_inner(&self, ctx: Arc<QueryContext>) -> Result<SendableDataBlockStream> {
         let make_error = || "failed to execute interpreter";
 
-        ctx.set_status_info("building pipeline");
+        ctx.set_status_info("[INTERPRETER] Building execution pipeline");
         ctx.check_aborting().with_context(make_error)?;
 
         let enable_disk_cache = match LicenseManagerSwitch::instance()
@@ -110,7 +110,7 @@ pub trait Interpreter: Sync + Send {
             Ok(_) => true,
             Err(e) => {
                 log::error!(
-                        "[Interpreter] CRITICAL ALERT: License validation FAILED - enterprise features DISABLED, System may operate in DEGRADED MODE with LIMITED CAPABILITIES and REDUCED PERFORMANCE. Please contact us at https://www.databend.com/contact-us/ or email [email protected] to restore full functionality: {}",
+                        "[INTERPRETER] CRITICAL ALERT: License validation FAILED - enterprise features DISABLED, System may operate in DEGRADED MODE with LIMITED CAPABILITIES and REDUCED PERFORMANCE. Please contact us at https://www.databend.com/contact-us/ or email [email protected] to restore full functionality: {}",
                         e
                     );
                 false
@@ -138,7 +138,7 @@ pub trait Interpreter: Sync + Send {
                 on_execution_finished(info, query_ctx)
             }));
 
-        ctx.set_status_info("executing pipeline");
+        ctx.set_status_info("[INTERPRETER] Executing pipeline");
 
         let settings = ctx.get_settings();
         build_res.set_max_threads(settings.get_max_threads()? as usize);
@@ -191,7 +191,7 @@ fn log_query_start(ctx: &QueryContext) {
     }
 
     if let Err(error) = InterpreterQueryLog::log_start(ctx, now, None) {
-        error!("[Interpreter] Query start logging failed: {:?}", error)
+        error!("[INTERPRETER] Failed to log query start: {:?}", error)
     }
 }
 
@@ -218,7 +218,7 @@ fn log_query_finished(ctx: &QueryContext, error: Option<ErrorCode>, has_profiles
     }
 
     if let Err(error) = InterpreterQueryLog::log_finish(ctx, now, error, has_profiles) {
-        error!("[Interpreter] Query finish logging failed: {:?}", error)
+        error!("[INTERPRETER] Failed to log query finish: {:?}", error)
     }
 }
 
@@ -261,7 +261,7 @@ async fn auto_commit_if_not_allowed_in_transaction(
     }
     if !stmt.is_transaction_command() && ctx.txn_mgr().lock().is_fail() {
         let err = ErrorCode::CurrentTransactionIsAborted(
-            "Current transaction is aborted, commands ignored until end of transaction block",
+            "[INTERPRETER] Current transaction is aborted, commands ignored until end of transaction block",
         );
         return Err(err);
     }

src/query/service/src/interpreters/interpreter_select.rs

@@ -112,7 +112,8 @@ impl SelectInterpreter {
     #[async_backtrace::framed]
     pub async fn build_physical_plan(&self) -> Result<PhysicalPlan> {
         let mut builder = PhysicalPlanBuilder::new(self.metadata.clone(), self.ctx.clone(), false);
-        self.ctx.set_status_info("building physical plan");
+        self.ctx
+            .set_status_info("[SELECT-INTERP] Building physical plan");
         builder
             .build(&self.s_expr, self.bind_context.column_set())
             .await
@@ -151,7 +152,7 @@ impl SelectInterpreter {
                                 update_table_metas: streams.update_table_metas,
                                 ..Default::default()
                             };
-                            info!("Updating the stream meta to consume data");
+                            info!("[SELECT-INTERP] Updating stream metadata to consume data");
                             catalog.update_multi_table_meta(r).await.map(|_| ())
                         }
                         None => Ok(()),
@@ -239,7 +240,7 @@ impl SelectInterpreter {
             if t.name().eq_ignore_ascii_case("result_scan") {
                 return if tables.len() > 1 {
                     Err(ErrorCode::Unimplemented(
-                        "The current `RESULT_SCAN` only supports single table queries",
+                        "[SELECT-INTERP] RESULT_SCAN currently supports only single table queries",
                     ))
                 } else {
                     Ok(Some(t.table()))
@@ -279,15 +280,16 @@ impl Interpreter for SelectInterpreter {
     async fn execute2(&self) -> Result<PipelineBuildResult> {
         self.attach_tables_to_ctx();
 
-        self.ctx.set_status_info("preparing plan");
+        self.ctx
+            .set_status_info("[SELECT-INTERP] Preparing execution plan");
 
         // 0. Need to build physical plan first to get the partitions.
         let physical_plan = self.build_physical_plan().await?;
         let query_plan = physical_plan
             .format(self.metadata.clone(), Default::default())?
             .format_pretty()?;
 
-        info!("Query physical plan: \n{}", query_plan);
+        info!("[SELECT-INTERP] Query physical plan:\n{}", query_plan);
 
         if self.ctx.get_settings().get_enable_query_result_cache()?
             && self.ctx.get_cacheable()
@@ -344,7 +346,7 @@ impl Interpreter for SelectInterpreter {
                 }
                 Err(e) => {
                     // 2.3 If an error occurs, turn back to the normal pipeline.
-                    error!("Failed to read query result cache. {}", e);
+                    error!("[SELECT-INTERP] Failed to read query result cache: {}", e);
                 }
             }
         }

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

@@ -159,7 +159,7 @@ impl PipelineExecutor {
             }
 
             info!(
-                "Init pipeline successfully, query_id: {:?}, elapsed: {:?}",
+                "[PIPELINE-EXECUTOR] Pipeline initialized successfully for query {}, elapsed: {:?}",
                 query_id,
                 instant.elapsed()
             );
@@ -171,7 +171,7 @@ impl PipelineExecutor {
         let instants = Instant::now();
         let _guard = defer(move || {
             info!(
-                "Pipeline executor finished, elapsed: {:?}",
+                "[PIPELINE-EXECUTOR] Execution completed, elapsed: {:?}",
                 instants.elapsed()
             );
         });
@@ -226,8 +226,8 @@ impl PipelineExecutor {
                 if let Either::Left(_) = select(max_execute_future, finished_future).await {
                     if let Some(graph) = this_graph.upgrade() {
                         graph.should_finish(Err(ErrorCode::AbortedQuery(
-                            "Aborted query, because the execution time exceeds the maximum execution time limit",
-                        ))).expect("exceed max execute time, but cannot send error message");
+                            "[PIPELINE-EXECUTOR] Query aborted due to execution time exceeding maximum limit",
+                        ))).expect("[PIPELINE-EXECUTOR] Failed to send timeout error message");
                     }
                 }
             });
@@ -244,13 +244,13 @@ impl PipelineExecutor {
                     query_wrapper
                         .graph
                         .should_finish(Err(may_error))
-                        .expect("executor cannot send error message");
+                        .expect("[PIPELINE-EXECUTOR] Failed to send error message");
                 }
                 None => {
                     query_wrapper
                         .graph
                         .should_finish::<()>(Ok(()))
-                        .expect("executor cannot send error message");
+                        .expect("[PIPELINE-EXECUTOR] Failed to send completion message");
                 }
             },
         }
@@ -289,7 +289,7 @@ impl PipelineExecutor {
     pub fn change_priority(&self, priority: u8) {
         match self {
             PipelineExecutor::QueryPipelineExecutor(_) => {
-                unreachable!("Logic error, cannot change priority for QueryPipelineExecutor")
+                unreachable!("[PIPELINE-EXECUTOR] Logic error: cannot change priority for QueryPipelineExecutor")
             }
             PipelineExecutor::QueriesPipelineExecutor(query_wrapper) => {
                 query_wrapper.graph.change_priority(priority as u64);

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

@@ -116,7 +116,7 @@ impl ProcessorAsyncTask {
             match futures::future::select(left, right).await {
                 Either::Left((res, _)) => res,
                 Either::Right((_, _)) => Err(ErrorCode::AbortedQuery(
-                    "Aborted query, because the server is shutting down or the query was killed.",
+                    "[PROCESSOR-ASYNC-TASK] Query aborted due to server shutdown or query termination",
                 )),
             }
         };
@@ -136,7 +136,7 @@ impl ProcessorAsyncTask {
                         let elapsed = start.elapsed();
                         let active_workers = queue_clone.active_workers();
                         warn!(
-                            "Very slow processor async task, query_id:{:?}, processor id: {:?}, name: {:?}, elapsed: {:?}, active sync workers: {:?}",
+                            "[PROCESSOR-ASYNC-TASK] Slow async task detected - query: {:?}, processor: {:?} ({}), elapsed: {:?}, active workers: {:?}",
                             query_id, processor_id, processor_name, elapsed, active_workers
                         );
                     }

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

@@ -76,7 +76,7 @@ impl QueryPipelineExecutor {
 
         if threads_num.is_zero() {
             return Err(ErrorCode::Internal(
-                "Pipeline max threads cannot equals zero.",
+                "[PIPELINE-EXECUTOR] Pipeline max threads cannot be zero",
             ));
         }
 
@@ -108,7 +108,9 @@ impl QueryPipelineExecutor {
         settings: ExecutorSettings,
     ) -> Result<Arc<QueryPipelineExecutor>> {
         if pipelines.is_empty() {
-            return Err(ErrorCode::Internal("Executor Pipelines is empty."));
+            return Err(ErrorCode::Internal(
+                "[PIPELINE-EXECUTOR] Executor pipelines cannot be empty",
+            ));
         }
 
         let threads_num = pipelines
@@ -119,7 +121,7 @@ impl QueryPipelineExecutor {
 
         if threads_num.is_zero() {
             return Err(ErrorCode::Internal(
-                "Pipeline max threads cannot equals zero.",
+                "[PIPELINE-EXECUTOR] Pipeline max threads cannot be zero",
             ));
         }
 
@@ -200,7 +202,8 @@ impl QueryPipelineExecutor {
     }
 
     pub fn finish<C>(&self, cause: Option<ErrorCode<C>>) {
-        let cause = cause.map(|err| err.with_context("pipeline executor finished"));
+        let cause =
+            cause.map(|err| err.with_context("[PIPELINE-EXECUTOR] Pipeline executor finished"));
 
         let mut finished_error = self.finished_error.lock();
         if let Some(cause) = cause {
@@ -284,7 +287,7 @@ impl QueryPipelineExecutor {
                 }
 
                 info!(
-                    "Init pipeline successfully, query_id: {:?}, elapsed: {:?}",
+                    "[PIPELINE-EXECUTOR] Pipeline initialized successfully for query {}, elapsed: {:?}",
                     self.settings.query_id,
                     instant.elapsed()
                 );
@@ -327,7 +330,7 @@ impl QueryPipelineExecutor {
                 if let Either::Left(_) = select(max_execute_future, finished_future).await {
                     if let Some(executor) = this.upgrade() {
                         executor.finish(Some(ErrorCode::AbortedQuery(
-                            "Aborted query, because the execution time exceeds the maximum execution time limit",
+                            "[PIPELINE-EXECUTOR] Query aborted due to execution time exceeding maximum limit",
                         )));
                     }
                 }
@@ -452,7 +455,9 @@ impl Drop for QueryPipelineExecutor {
 
         let cause = match self.finished_error.lock().as_ref() {
             Some(cause) => cause.clone(),
-            None => ErrorCode::Internal("Pipeline illegal state: not successfully shutdown."),
+            None => ErrorCode::Internal(
+                "[PIPELINE-EXECUTOR] Pipeline illegal state: not successfully shutdown",
+            ),
         };
 
         let mut on_finished_chain = self.on_finished_chain.lock();
@@ -463,7 +468,7 @@ impl Drop for QueryPipelineExecutor {
         let profiling = self.fetch_plans_profile(true);
         let info = ExecutionInfo::create(Err(cause), profiling);
         if let Err(cause) = on_finished_chain.apply(info) {
-            warn!("Pipeline executor shutdown failure, {:?}", cause);
+            warn!("[PIPELINE-EXECUTOR] Shutdown failure: {:?}", cause);
         }
     }
 }