chore(query): refactor metrics (#15421)

* chore(query): refactor metrics

* chore(query): refactor metrics

* chore(query): fix VACUUM TEMPORARY FILES

Author: sundy-li

src/common/metrics/src/metrics/interpreter.rs

@@ -22,7 +22,6 @@ use databend_common_base::runtime::metrics::FamilyHistogram;
 use crate::VecLabels;
 
 const METRIC_QUERY_START: &str = "query_start";
-const METRIC_QUERY_ERROR: &str = "query_error";
 const METRIC_QUERY_SUCCESS: &str = "query_success";
 const METRIC_QUERY_FAILED: &str = "query_failed";
 
@@ -42,8 +41,6 @@ const METRIC_QUERY_RESULT_BYTES: &str = "query_result_bytes";
 
 pub static QUERY_START: LazyLock<FamilyCounter<VecLabels>> =
     LazyLock::new(|| register_counter_family(METRIC_QUERY_START));
-pub static QUERY_ERROR: LazyLock<FamilyCounter<VecLabels>> =
-    LazyLock::new(|| register_counter_family(METRIC_QUERY_ERROR));
 pub static QUERY_SUCCESS: LazyLock<FamilyCounter<VecLabels>> =
     LazyLock::new(|| register_counter_family(METRIC_QUERY_SUCCESS));
 pub static QUERY_FAILED: LazyLock<FamilyCounter<VecLabels>> =

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

@@ -63,40 +63,40 @@ pub trait Interpreter: Sync + Send {
     #[async_backtrace::framed]
     #[minitrace::trace]
     async fn execute(&self, ctx: Arc<QueryContext>) -> Result<SendableDataBlockStream> {
-        ctx.set_status_info("building pipeline");
-        InterpreterMetrics::record_query_start(&ctx);
         log_query_start(&ctx);
-
-        if let Err(err) = ctx.check_aborting() {
-            log_query_finished(&ctx, Some(err.clone()), false);
-            return Err(err);
+        match self.execute_inner(ctx.clone()).await {
+            Ok(stream) => Ok(stream),
+            Err(err) => {
+                log_query_finished(&ctx, Some(err.clone()), false);
+                Err(err)
+            }
         }
+    }
+
+    async fn execute_inner(&self, ctx: Arc<QueryContext>) -> Result<SendableDataBlockStream> {
+        ctx.set_status_info("building pipeline");
+        ctx.check_aborting()?;
         if self.is_ddl() {
             CommitInterpreter::try_create(ctx.clone())?
                 .execute2()
                 .await?;
             ctx.clear_tables_cache();
         }
         if !self.is_txn_command() && ctx.txn_mgr().lock().is_fail() {
-            let error = ErrorCode::CurrentTransactionIsAborted(
+            let err = ErrorCode::CurrentTransactionIsAborted(
                 "current transaction is aborted, commands ignored until end of transaction block",
             );
-            log_query_finished(&ctx, Some(error.clone()), false);
-            return Err(error);
+            return Err(err);
         }
         let mut build_res = match self.execute2().await {
             Ok(build_res) => build_res,
-            Err(build_error) => {
-                InterpreterMetrics::record_query_error(&ctx);
-                log_query_finished(&ctx, Some(build_error.clone()), false);
-                return Err(build_error);
+            Err(err) => {
+                return Err(err);
             }
         };
 
         if build_res.main_pipeline.is_empty() {
-            InterpreterMetrics::record_query_finished(&ctx, None);
             log_query_finished(&ctx, None, false);
-
             return Ok(Box::pin(DataBlockStream::create(None, vec![])));
         }
 
@@ -137,9 +137,7 @@ pub trait Interpreter: Sync + Send {
                     Err(e) => Some(e.clone()),
                 };
 
-                InterpreterMetrics::record_query_finished(&query_ctx, err_opt.clone());
                 log_query_finished(&query_ctx, err_opt, has_profiles);
-
                 match may_error {
                     Ok(_) => Ok(()),
                     Err(error) => Err(error.clone()),
@@ -190,6 +188,7 @@ pub trait Interpreter: Sync + Send {
 pub type InterpreterPtr = Arc<dyn Interpreter>;
 
 fn log_query_start(ctx: &QueryContext) {
+    InterpreterMetrics::record_query_start(ctx);
     let now = SystemTime::now();
     let session = ctx.get_current_session();
 
@@ -203,6 +202,8 @@ fn log_query_start(ctx: &QueryContext) {
 }
 
 fn log_query_finished(ctx: &QueryContext, error: Option<ErrorCode>, has_profiles: bool) {
+    InterpreterMetrics::record_query_finished(ctx, error.clone());
+
     let now = SystemTime::now();
     let session = ctx.get_current_session();
 

src/query/service/src/interpreters/interpreter_metrics.rs

@@ -128,11 +128,6 @@ impl InterpreterMetrics {
             }
         };
     }
-
-    pub fn record_query_error(ctx: &QueryContext) {
-        let labels = Self::common_labels(ctx);
-        QUERY_ERROR.get_or_create(&labels).inc();
-    }
 }
 
 fn convert_query_timestamp(time: SystemTime) -> u128 {

src/query/sql/src/planner/plans/plan.rs

@@ -509,6 +509,7 @@ impl Plan {
                 | Plan::Presign(_)
                 | Plan::VacuumTable(_)
                 | Plan::VacuumDropTable(_)
+                | Plan::VacuumTemporaryFiles(_)
                 | Plan::DescDatamaskPolicy(_)
                 | Plan::DescNetworkPolicy(_)
                 | Plan::ShowNetworkPolicies(_)

tests/sqllogictests/suites/ee/03_ee_vacuum/03_0000_vacuum_temporary_files.test

@@ -24,7 +24,6 @@ SELECT sleep(2) from numbers(1);
 ----
 0
 
-onlyif mysql
 statement ok
 VACUUM TEMPORARY FILES RETAIN 2 SECONDS;