fix: some queries print query log and profile twice (#19455)

* refactor(logging): replace on_execution_finished with QueryFinishHooks
fix(logging): prevent duplicate query logs by fixing root causes
fix(logging): deduplicate query logs and improve log tests

* fix test

clean code

* apply suggestions

Author: dqhl76

src/query/service/src/interpreters/common/finish_hook.rs

@@ -0,0 +1,102 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+
+use databend_common_catalog::table_context::TableContext;
+use databend_common_exception::Result;
+use databend_common_pipeline::core::ExecutionInfo;
+
+use crate::interpreters::common::log_query_finished;
+use crate::interpreters::hook::vacuum_hook::hook_clear_m_cte_temp_table;
+use crate::interpreters::hook::vacuum_hook::hook_disk_temp_dir;
+use crate::interpreters::hook::vacuum_hook::hook_vacuum_temp_files;
+use crate::sessions::QueryContext;
+
+fn run_hooks(query_ctx: Arc<QueryContext>) -> Result<()> {
+    hook_clear_m_cte_temp_table(&query_ctx)?;
+    hook_vacuum_temp_files(&query_ctx)?;
+    hook_disk_temp_dir(&query_ctx)
+}
+
+/// Controls which post-execution actions are performed when a pipeline finishes.
+///
+/// Use [`QueryFinishHooks::top_level`] for normal user-facing queries,
+/// [`QueryFinishHooks::nested_with_hooks`] for internal sub-executions that may
+/// create temporary artifacts (e.g. EXPLAIN ANALYZE, EXPLAIN PERF inner pipelines),
+/// and [`QueryFinishHooks::nested`] for lightweight internal pipelines that don't
+/// need cleanup (e.g. recursive CTE inner pipeline).
+pub struct QueryFinishHooks {
+    /// Collect pipeline execution profiles into the query context.
+    pub collect_profiles: bool,
+    /// Run post-query cleanup hooks (CTE temp tables, spill files, disk temp dirs).
+    pub run_hooks: bool,
+    /// Emit the query-finish log, metrics, and profile JSON.
+    pub log_finished: bool,
+}
+
+impl QueryFinishHooks {
+    /// All three actions enabled. Use for top-level user queries.
+    pub fn top_level() -> Self {
+        Self {
+            collect_profiles: true,
+            run_hooks: true,
+            log_finished: true,
+        }
+    }
+
+    /// Profiles only — no hooks, no logging. Use for nested/internal pipeline
+    /// executions where the outer query owns the lifecycle (e.g. recursive CTE
+    /// inner pipeline).
+    pub fn nested() -> Self {
+        Self {
+            collect_profiles: true,
+            run_hooks: false,
+            log_finished: false,
+        }
+    }
+
+    /// Profiles and cleanup hooks, but no logging. Use for nested pipelines
+    /// that may create temporary artifacts (spill files, CTE temp tables) and
+    /// need cleanup even on failure, while the outer query owns the log
+    /// lifecycle (e.g. EXPLAIN ANALYZE, EXPLAIN PERF inner pipelines).
+    pub fn nested_with_hooks() -> Self {
+        Self {
+            collect_profiles: true,
+            run_hooks: true,
+            log_finished: false,
+        }
+    }
+
+    /// Convert into a closure suitable for [`Pipeline::set_on_finished`].
+    pub fn into_callback(
+        self,
+        ctx: Arc<QueryContext>,
+    ) -> impl Fn(&ExecutionInfo) -> Result<()> + Send + Sync + 'static {
+        move |info: &ExecutionInfo| {
+            if self.collect_profiles {
+                ctx.add_query_profiles(&info.profiling);
+            }
+            let hooks_res = if self.run_hooks {
+                run_hooks(ctx.clone())
+            } else {
+                Ok(())
+            };
+            if self.log_finished {
+                log_query_finished(&ctx, info.res.clone().err());
+            }
+            info.res.clone().and(hooks_res)
+        }
+    }
+}

src/query/service/src/interpreters/common/log.rs

@@ -0,0 +1,102 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use std::time::SystemTime;
+
+use databend_common_base::runtime::profile::ProfileDesc;
+use databend_common_base::runtime::profile::ProfileStatisticsName;
+use databend_common_base::runtime::profile::get_statistics_desc;
+use databend_common_catalog::table_context::TableContext;
+use databend_common_exception::ErrorCode;
+use databend_common_pipeline::core::PlanProfile;
+use log::error;
+use log::info;
+
+use crate::interpreters::InterpreterMetrics;
+use crate::interpreters::InterpreterQueryLog;
+use crate::sessions::QueryContext;
+use crate::sessions::SessionManager;
+
+pub fn log_query_start(ctx: &QueryContext) {
+    InterpreterMetrics::record_query_start(ctx);
+    let now = SystemTime::now();
+    let session = ctx.get_current_session();
+    let typ = session.get_type();
+    if typ.is_user_session() {
+        SessionManager::instance().status.write().query_start(now);
+    }
+
+    if let Err(error) = InterpreterQueryLog::log_start(ctx, now, None) {
+        error!("Failed to log query start: {:?}", error)
+    }
+}
+
+pub fn log_query_finished(ctx: &QueryContext, error: Option<ErrorCode>) {
+    // metrics
+    InterpreterMetrics::record_query_finished(ctx, error.clone());
+
+    let now = SystemTime::now();
+    let session = ctx.get_current_session();
+
+    session.get_status().write().query_finish();
+    let typ = session.get_type();
+    if typ.is_user_session() {
+        SessionManager::instance().status.write().query_finish(now);
+        SessionManager::instance()
+            .metrics_collector
+            .track_finished_query(
+                ctx.get_scan_progress_value(),
+                ctx.get_write_progress_value(),
+                ctx.get_join_spill_progress_value(),
+                ctx.get_aggregate_spill_progress_value(),
+                ctx.get_group_by_spill_progress_value(),
+                ctx.get_window_partition_spill_progress_value(),
+            );
+    }
+
+    info!(memory:? = ctx.get_node_peek_memory_usage(); "total memory usage");
+
+    // databend::log::profile
+    let query_profiles = ctx.get_query_profiles();
+    let has_profiles = !query_profiles.is_empty();
+
+    if has_profiles {
+        #[derive(serde::Serialize)]
+        struct QueryProfiles {
+            query_id: String,
+            profiles: Vec<PlanProfile>,
+            statistics_desc: Arc<BTreeMap<ProfileStatisticsName, ProfileDesc>>,
+        }
+
+        match serde_json::to_string(&QueryProfiles {
+            query_id: ctx.get_id(),
+            profiles: query_profiles.clone(),
+            statistics_desc: get_statistics_desc(),
+        }) {
+            Ok(profile_json) => {
+                info!(target: "databend::log::profile", "{}", profile_json);
+            }
+            Err(err) => {
+                error!("Failed to serialize query profiles: {:?}", err);
+            }
+        }
+    }
+
+    // databend::log::query
+    if let Err(error) = InterpreterQueryLog::log_finish(ctx, now, error, has_profiles) {
+        error!("Failed to log query finish: {:?}", error)
+    }
+}

src/query/service/src/interpreters/common/mod.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 mod column;
+mod finish_hook;
 mod grant;
 mod metrics;
 mod notification;
@@ -22,10 +23,13 @@ mod table;
 mod task;
 mod util;
 
+mod log;
 pub mod table_option_validation;
 
 pub use column::*;
+pub use finish_hook::QueryFinishHooks;
 pub use grant::validate_grant_object_exists;
+pub use log::*;
 pub use notification::get_notification_client_config;
 pub use query_log::InterpreterQueryLog;
 pub use stream::dml_build_update_stream_req;