refactor(query): combine short-query and page-aware execution

src/query/service/src/interpreters/access/privilege_access.rs

@@ -756,6 +756,16 @@ impl PrivilegeAccess {
     ) -> Result<()> {
         let session = self.ctx.get_current_session();
 
+        // Direct user grants do not require resolving an ownership object.
+        if self
+            .ctx
+            .get_current_user()?
+            .grants
+            .verify_privilege(grant_object, privilege)
+        {
+            return Ok(());
+        }
+
         let verify_ownership = match grant_object {
             GrantObject::Database(_, _)
             | GrantObject::Table(_, _, _)

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

@@ -16,11 +16,14 @@ use std::collections::BTreeMap;
 use std::sync::Arc;
 use std::time::SystemTime;
 
+use databend_common_base::runtime::ThreadTracker;
 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_config::GlobalConfig;
 use databend_common_exception::ErrorCode;
 use databend_common_pipeline::core::PlanProfile;
+use databend_common_tracing::HistoryConfig;
 use log::error;
 use log::info;
 
@@ -33,6 +36,43 @@ use crate::sessions::TableContextQueryIdentity;
 use crate::sessions::TableContextQueryProfile;
 use crate::sessions::TableContextSpillProgress;
 
+const QUERY_HISTORY_TABLE: &str = "query_history";
+const PROFILE_HISTORY_TABLE: &str = "profile_history";
+const ACCESS_HISTORY_TABLE: &str = "access_history";
+
+fn history_table_log_enabled(history: &HistoryConfig, history_table_name: &str) -> bool {
+    history.on
+        && history
+            .tables
+            .iter()
+            .any(|table| table.table_name.eq_ignore_ascii_case(history_table_name))
+}
+
+fn captured_info_log_enabled() -> bool {
+    ThreadTracker::capture_log_settings().is_some_and(|settings| {
+        settings.queue.is_some() && settings.level >= log::LevelFilter::Info
+    })
+}
+
+pub(crate) fn query_log_enabled() -> bool {
+    let log = &GlobalConfig::instance().log;
+    log.query.on
+        || history_table_log_enabled(&log.history, QUERY_HISTORY_TABLE)
+        || captured_info_log_enabled()
+}
+
+pub(crate) fn profile_log_enabled() -> bool {
+    let log = &GlobalConfig::instance().log;
+    log.profile.on
+        || history_table_log_enabled(&log.history, PROFILE_HISTORY_TABLE)
+        || captured_info_log_enabled()
+}
+
+pub(crate) fn access_log_enabled() -> bool {
+    let log = &GlobalConfig::instance().log;
+    history_table_log_enabled(&log.history, ACCESS_HISTORY_TABLE) || captured_info_log_enabled()
+}
+
 pub fn log_query_start(ctx: &QueryContext) {
     InterpreterMetrics::record_query_start(ctx);
     let now = SystemTime::now();
@@ -42,7 +82,9 @@ pub fn log_query_start(ctx: &QueryContext) {
         SessionManager::instance().status.write().query_start(now);
     }
 
-    if let Err(error) = InterpreterQueryLog::log_start(ctx, now, None) {
+    if query_log_enabled()
+        && let Err(error) = InterpreterQueryLog::log_start(ctx, now, None)
+    {
         error!("Failed to log query start: {:?}", error)
     }
 }
@@ -52,6 +94,7 @@ pub fn log_query_finished(ctx: &QueryContext, error: Option<ErrorCode>) {
     InterpreterMetrics::record_query_finished(ctx, error.clone());
 
     let now = SystemTime::now();
+    ctx.set_finish_time(now);
     let session = ctx.get_current_session();
 
     session.get_status().write().query_finish();
@@ -72,11 +115,16 @@ pub fn log_query_finished(ctx: &QueryContext, error: Option<ErrorCode>) {
 
     info!(memory:? = ctx.get_node_peek_memory_usage(); "total memory usage");
 
-    // databend::log::profile
-    let query_profiles = ctx.get_query_profiles();
+    let query_log_is_enabled = query_log_enabled();
+    let profile_log_is_enabled = profile_log_enabled();
+    let query_profiles = if query_log_is_enabled || profile_log_is_enabled {
+        ctx.get_query_profiles()
+    } else {
+        Vec::new()
+    };
     let has_profiles = !query_profiles.is_empty();
 
-    if has_profiles {
+    if has_profiles && profile_log_is_enabled {
         #[derive(serde::Serialize)]
         struct QueryProfiles {
             query_id: String,
@@ -98,10 +146,62 @@ pub fn log_query_finished(ctx: &QueryContext, error: Option<ErrorCode>) {
         }
     }
 
-    // databend::log::query
-    if let Err(error) =
-        InterpreterQueryLog::log_finish(ctx, now, error, has_profiles, &query_profiles)
+    if query_log_is_enabled
+        && let Err(error) =
+            InterpreterQueryLog::log_finish(ctx, now, error, has_profiles, &query_profiles)
     {
         error!("Failed to log query finish: {:?}", error)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use concurrent_queue::ConcurrentQueue;
+    use databend_common_base::runtime::CaptureLogSettings;
+    use databend_common_base::runtime::ThreadTracker;
+    use databend_common_tracing::HistoryConfig;
+    use databend_common_tracing::HistoryTableConfig;
+    use log::LevelFilter;
+
+    use super::captured_info_log_enabled;
+    use super::history_table_log_enabled;
+
+    #[test]
+    fn test_history_table_log_enabled() {
+        let mut history = HistoryConfig::default();
+        history.tables.push(HistoryTableConfig {
+            table_name: "QUERY_HISTORY".to_string(),
+            ..Default::default()
+        });
+
+        assert!(!history_table_log_enabled(&history, "query_history"));
+
+        history.on = true;
+        assert!(history_table_log_enabled(&history, "query_history"));
+        assert!(!history_table_log_enabled(&history, "profile_history"));
+    }
+
+    #[test]
+    fn test_captured_info_log_enabled() {
+        assert!(!captured_info_log_enabled());
+
+        {
+            let mut payload = ThreadTracker::new_tracking_payload();
+            payload.capture_log_settings = Some(CaptureLogSettings::capture_off());
+            let _guard = ThreadTracker::tracking(payload);
+            assert!(!captured_info_log_enabled());
+        }
+
+        {
+            let mut payload = ThreadTracker::new_tracking_payload();
+            payload.capture_log_settings = Some(CaptureLogSettings::capture_query(
+                LevelFilter::Info,
+                Arc::new(ConcurrentQueue::unbounded()),
+            ));
+            let _guard = ThreadTracker::tracking(payload);
+            assert!(captured_info_log_enabled());
+        }
+    }
+}

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

@@ -34,6 +34,7 @@ use log::info;
 use serde_json;
 use serde_json::Value;
 
+use super::query_log_enabled;
 use crate::sessions::QueryContext;
 use crate::sessions::TableContextAuthorization;
 use crate::sessions::TableContextCluster;
@@ -123,25 +124,31 @@ fn resource_usage_query_log(stats: IoStatsSnapshot, profiles: &[PlanProfile]) ->
 }
 
 impl InterpreterQueryLog {
+    pub fn enabled() -> bool {
+        query_log_enabled()
+    }
+
     fn write_log(mut event: QueryLogElement) -> Result<()> {
-        // log the query event in the system_history.query_history table
+        // Normal callers are gated by `enabled()`. The log facade's target check is
+        // not specific enough to tell whether this query sink is configured.
         let event_str = serde_json::to_string(&event)?;
         info!(target: "databend::log::query", "{}", event_str);
 
-        // log the query event in `query-details` log file
-        // remove some fields to keep tidy in the log file
+        // Remove verbose fields from the query-details file.
         event.session_settings.clear();
         event.sql_user_quota.clear();
         event.sql_user_privileges.clear();
         let event_str = serde_json::to_string(&event)?;
         info!(target: "databend::log::query::file", "{}", event_str);
 
-        // log the query event in the system log
         info!("query: {} becomes {:?}", event.query_id, event.log_type);
         Ok(())
     }
 
     pub fn fail_to_start(ctx: Arc<QueryContext>, err: ErrorCode) {
+        if !Self::enabled() {
+            return;
+        }
         InterpreterQueryLog::log_start(&ctx, SystemTime::now(), Some(err))
             .unwrap_or_else(|e| error!("fail to write query_log {:?}", e));
     }

src/query/service/src/interpreters/interpreter_factory.rs

@@ -52,6 +52,7 @@ use crate::interpreters::ShowPublicKeysInterpreter;
 use crate::interpreters::UnsetObjectTagsInterpreter;
 use crate::interpreters::access::Accessor;
 use crate::interpreters::access_log::AccessLogger;
+use crate::interpreters::common::access_log_enabled;
 use crate::interpreters::interpreter_add_warehouse_cluster::AddWarehouseClusterInterpreter;
 use crate::interpreters::interpreter_assign_warehouse_nodes::AssignWarehouseNodesInterpreter;
 use crate::interpreters::interpreter_catalog_drop::DropCatalogInterpreter;
@@ -153,9 +154,11 @@ impl InterpreterFactory {
                 error!("Access.denied(v2): {:?}", e);
             }
         })?;
-        let mut access_logger = AccessLogger::create(ctx.clone());
-        access_logger.log(plan);
-        access_logger.output();
+        if access_log_enabled() {
+            let mut access_logger = AccessLogger::create(ctx.clone());
+            access_logger.log(plan);
+            access_logger.output();
+        }
         Self::get_warehouses_interpreter(ctx, plan, Self::get_inner)
     }
 

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

@@ -286,40 +286,60 @@ impl QueryPipelineExecutor {
 
         self.start_executor_daemon()?;
 
-        let mut thread_join_handles = self.execute_threads(self.threads_num);
+        // Pulling and complete pipelines call this from a dedicated OS thread, which can also act
+        // as one worker. Generic callers may invoke the executor from a Tokio runtime, where a
+        // processor using Tokio's blocking APIs would panic if it ran inline.
+        let execute_inline = tokio::runtime::Handle::try_current().is_err();
+        let spawned_threads = self.threads_num - usize::from(execute_inline);
+        let mut thread_join_handles = self.execute_threads(spawned_threads);
+
+        if execute_inline {
+            let inline_thread_num = spawned_threads;
+            let inline_thread_name = format!("PipelineExecutor-{}", inline_thread_num);
+            let inline_span =
+                Span::enter_with_local_parent("QueryPipelineExecutor::execute_threads")
+                    .with_property(|| ("thread_name", inline_thread_name));
+            let thread_res = self.execute_thread(inline_thread_num, inline_span);
+            self.check_thread_result(thread_res)?;
+        }
 
         while let Some(join_handle) = thread_join_handles.pop() {
             let thread_res = join_handle.join().flatten();
+            self.check_thread_result(thread_res)?;
+        }
 
-            {
-                let finished_error_guard = self.finished_error.lock();
-                if let Some(error) = finished_error_guard.as_ref() {
-                    let may_error = error.clone();
-                    drop(finished_error_guard);
-
-                    let profiling = self.fetch_plans_profile(true);
-                    self.on_finished(ExecutionInfo::create(Err(may_error.clone()), profiling))?;
-                    return Err(may_error);
-                }
-            }
+        if let Err(error) = self.graph.assert_finished_graph() {
+            let profiling = self.fetch_plans_profile(true);
+            self.on_finished(ExecutionInfo::create(Err(error.clone()), profiling))?;
+            return Err(error);
+        }
+
+        let profiling = self.fetch_plans_profile(true);
+        self.on_finished(ExecutionInfo::create(Ok(()), profiling))?;
+        Ok(())
+    }
+
+    fn check_thread_result(&self, thread_res: Result<()>) -> Result<()> {
+        {
+            let finished_error_guard = self.finished_error.lock();
+            if let Some(error) = finished_error_guard.as_ref() {
+                let may_error = error.clone();
+                drop(finished_error_guard);
 
-            // We will ignore the abort query error, because returned by finished_error if abort query.
-            if matches!(&thread_res, Err(error) if error.code() != ErrorCode::ABORTED_QUERY) {
-                let may_error = thread_res.unwrap_err();
                 let profiling = self.fetch_plans_profile(true);
                 self.on_finished(ExecutionInfo::create(Err(may_error.clone()), profiling))?;
                 return Err(may_error);
             }
         }
 
-        if let Err(error) = self.graph.assert_finished_graph() {
+        // We will ignore the abort query error, because returned by finished_error if abort query.
+        if matches!(&thread_res, Err(error) if error.code() != ErrorCode::ABORTED_QUERY) {
+            let may_error = thread_res.unwrap_err();
             let profiling = self.fetch_plans_profile(true);
-            self.on_finished(ExecutionInfo::create(Err(error.clone()), profiling))?;
-            return Err(error);
+            self.on_finished(ExecutionInfo::create(Err(may_error.clone()), profiling))?;
+            return Err(may_error);
         }
 
-        let profiling = self.fetch_plans_profile(true);
-        self.on_finished(ExecutionInfo::create(Ok(()), profiling))?;
         Ok(())
     }
 
@@ -417,28 +437,32 @@ impl QueryPipelineExecutor {
 
             let span = Span::enter_with_local_parent("QueryPipelineExecutor::execute_threads")
                 .with_property(|| ("thread_name", name.clone()));
-            thread_join_handles.push(Thread::named_spawn(Some(name), move || unsafe {
-                let _g = span.set_local_parent();
-                let this_clone = this.clone();
-                let try_result = catch_unwind(|| this_clone.execute_single_thread(thread_num));
-
-                // finish the pipeline executor when has error or panic
-                if let Err(cause) = try_result.flatten() {
-                    span.with_property(|| ("error", "true")).add_properties(|| {
-                        [
-                            ("error.type", cause.code().to_string()),
-                            ("error.message", cause.display_text()),
-                        ]
-                    });
-                    this.finish(Some(cause));
-                }
-
-                Ok(())
+            thread_join_handles.push(Thread::named_spawn(Some(name), move || {
+                this.execute_thread(thread_num, span)
             }));
         }
         thread_join_handles
     }
 
+    fn execute_thread(self: &Arc<Self>, thread_num: usize, span: Span) -> Result<()> {
+        let _g = span.set_local_parent();
+        let this = self.clone();
+        let try_result = catch_unwind(|| unsafe { this.execute_single_thread(thread_num) });
+
+        // Finish the pipeline executor when a worker returns an error or panics.
+        if let Err(cause) = try_result.flatten() {
+            span.with_property(|| ("error", "true")).add_properties(|| {
+                [
+                    ("error.type", cause.code().to_string()),
+                    ("error.message", cause.display_text()),
+                ]
+            });
+            self.finish(Some(cause));
+        }
+
+        Ok(())
+    }
+
     /// # Safety
     ///
     /// Method is thread unsafe and require thread safe call