refactor(executor): control queries executor via configuration (#18283)

* refactor: make queries executor lazy init

* test

* fix: queries executor hang

* fix: executor of r-cte run in others threads

* fix: use config(enable) + settings(fallback) to control

* clean

* save

* fixut

* Revert "save"

This reverts commit 8fe0c86.

* follow review suggestion

* follow review suggestion

---------

Co-authored-by: Winter Zhang <coswde@gmail.com>

Author: dqhl76

src/query/config/src/config.rs

@@ -1946,6 +1946,9 @@ pub struct QueryConfig {
 
     #[clap(skip)]
     pub resources_management: Option<ResourcesManagementConfig>,
+
+    #[clap(long, value_name = "VALUE", default_value = "false")]
+    pub enable_queries_executor: bool,
 }
 
 impl Default for QueryConfig {
@@ -2047,6 +2050,7 @@ impl TryInto<InnerQueryConfig> for QueryConfig {
                 .map(|(k, v)| (k, v.into()))
                 .collect(),
             resources_management: self.resources_management,
+            enable_queries_executor: self.enable_queries_executor,
         })
     }
 }
@@ -2155,6 +2159,7 @@ impl From<InnerQueryConfig> for QueryConfig {
             network_policy_whitelist: inner.network_policy_whitelist,
             settings: HashMap::new(),
             resources_management: None,
+            enable_queries_executor: inner.enable_queries_executor,
         }
     }
 }

src/query/config/src/inner.rs

@@ -256,6 +256,8 @@ pub struct QueryConfig {
 
     pub settings: HashMap<String, UserSettingValue>,
     pub resources_management: Option<ResourcesManagementConfig>,
+
+    pub enable_queries_executor: bool,
 }
 
 impl Default for QueryConfig {
@@ -343,6 +345,7 @@ impl Default for QueryConfig {
             network_policy_whitelist: Vec::new(),
             settings: HashMap::new(),
             resources_management: None,
+            enable_queries_executor: false,
         }
     }
 }

src/query/service/Cargo.toml

@@ -19,8 +19,6 @@ io-uring = [
     "databend-common-meta-store/io-uring",
 ]
 
-enable_queries_executor = []
-
 [dependencies]
 anyhow = { workspace = true }
 arrow-array = { workspace = true }

src/query/service/src/global_services.rs

@@ -50,7 +50,6 @@ use crate::catalogs::IcebergCreator;
 use crate::clusters::ClusterDiscovery;
 use crate::history_tables::GlobalHistoryLog;
 use crate::locks::LockManager;
-#[cfg(feature = "enable_queries_executor")]
 use crate::pipelines::executor::GlobalQueriesExecutor;
 use crate::servers::flight::v1::exchange::DataExchangeManager;
 use crate::servers::http::v1::ClientSessionManager;
@@ -163,15 +162,14 @@ impl GlobalServices {
             CloudControlApiProvider::init(addr, config.query.cloud_control_grpc_timeout).await?;
         }
 
-        #[cfg(feature = "enable_queries_executor")]
-        {
-            GlobalQueriesExecutor::init()?;
-        }
-
         if !ee_mode {
             DummyResourcesManagement::init()?;
         }
 
+        if config.query.enable_queries_executor {
+            GlobalQueriesExecutor::init()?;
+        }
+
         Self::init_workload_mgr(config).await?;
 
         if config.log.history.on {

src/query/service/src/interpreters/interpreter_set.rs

@@ -113,6 +113,25 @@ impl SetInterpreter {
                         .await?;
                     true
                 }
+                "use_legacy_query_executor" => {
+                    // This is a fallback setting, allowing user to fallback from **queries** executor
+                    // to the **query** executor. So, if queries executor not enable in the config
+                    // we will return an error.
+                    // TODO: we will remove this setting when queries executor is stable.
+                    let config = GlobalConfig::instance();
+                    if !config.query.enable_queries_executor {
+                        return Err(
+                            ErrorCode::InvalidArgument("This setting is not allowed when queries executor is not enabled in the configuration"));
+                    }
+                    if scalar.as_str() == "0" {
+                        return Err(ErrorCode::InvalidArgument(
+                            "This setting is not allowed set to 0, if already enable in the configuration, please use unset to revert this",
+                        ));
+                    }
+                    self.set_settings(var.to_string(), scalar.clone(), is_global)
+                        .await?;
+                    true
+                }
                 _ => {
                     self.set_settings(var.to_string(), scalar.clone(), is_global)
                         .await?;

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

@@ -445,6 +445,7 @@ impl ExecutingGraph {
         let max_points = self.max_points.load(Ordering::SeqCst);
         let mut expected_value = 0;
         let mut desired_value = 0;
+
         loop {
             match self.points.compare_exchange_weak(
                 expected_value,
@@ -460,6 +461,7 @@ impl ExecutingGraph {
                     let epoch = new_expected & EPOCH_MASK;
 
                     expected_value = new_expected;
+
                     if epoch > global_epoch as u64 {
                         desired_value = new_expected;
                     } else if epoch < global_epoch as u64 {

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

@@ -16,6 +16,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use databend_common_catalog::table_context::TableContext;
+use databend_common_config::GlobalConfig;
 use databend_common_exception::Result;
 
 #[derive(Clone)]
@@ -34,8 +35,18 @@ impl ExecutorSettings {
         let max_threads = settings.get_max_threads()?;
         let max_execute_time_in_seconds = settings.get_max_execute_time_in_seconds()?;
 
+        let config_enable_queries_executor = GlobalConfig::instance().query.enable_queries_executor;
+        let setting_use_legacy_query_executor = settings.get_use_legacy_query_executor()?;
+        // If `use_legacy_query_executor` is set to 1, we disable the queries executor
+        // Otherwise, we all follow configuration
+        let enable_queries_executor = if setting_use_legacy_query_executor {
+            false
+        } else {
+            config_enable_queries_executor
+        };
+
         Ok(ExecutorSettings {
-            enable_queries_executor: settings.get_enable_experimental_queries_executor()?,
+            enable_queries_executor,
             query_id: Arc::new(query_id),
             max_execute_time_in_seconds: Duration::from_secs(max_execute_time_in_seconds),
             max_threads,

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

@@ -27,7 +27,7 @@ impl GlobalQueriesExecutor {
     pub fn init() -> Result<()> {
         let num_cpus = num_cpus::get();
         GlobalInstance::set(QueriesPipelineExecutor::create(num_cpus)?);
-        Thread::spawn(|| {
+        Thread::named_spawn(Some("GlobalQueriesExecutor".to_string()), || {
             if let Err(e) = Self::instance().execute() {
                 info!("Executor finished with error: {:?}", e);
             }

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

@@ -49,6 +49,15 @@ impl WorkersTasks {
     pub fn swap_tasks(&mut self) {
         mem::swap(&mut self.current_tasks, &mut self.next_tasks);
     }
+
+    pub fn wakeup_all_waiting_workers(&mut self, condvar: &WorkersCondvar) {
+        for worker_id in 0..self.workers_waiting_status.total_size() {
+            if self.workers_waiting_status.is_waiting(worker_id) {
+                self.workers_waiting_status.wakeup_worker(worker_id);
+                condvar.wakeup(worker_id);
+            }
+        }
+    }
 }
 
 pub struct QueriesExecutorTasksQueue {
@@ -163,19 +172,18 @@ impl QueriesExecutorTasksQueue {
             workers_tasks
                 .next_tasks
                 .push_task(worker_id, ExecutorTask::Sync(proc));
-
             worker_id += 1;
+
             if worker_id == workers_tasks.next_tasks.workers_sync_tasks.len() {
                 worker_id = 0;
             }
-
-            if workers_tasks.workers_waiting_status.is_waiting(worker_id) {
-                workers_tasks
-                    .workers_waiting_status
-                    .wakeup_worker(worker_id);
-                condvar.wakeup(worker_id);
-            }
         }
+
+        // Wake up all workers that is waiting
+        // Only check one worker and wake it up is not enough,
+        // in queries executor, this worker may be blocked and wait for a result
+        // e.g. TransformRecursiveCteSource
+        workers_tasks.wakeup_all_waiting_workers(&condvar);
     }
 
     pub fn init_async_tasks(
@@ -190,19 +198,13 @@ impl QueriesExecutorTasksQueue {
             workers_tasks
                 .next_tasks
                 .push_task(worker_id, ExecutorTask::Async(proc));
-
             worker_id += 1;
             if worker_id == workers_tasks.next_tasks.workers_sync_tasks.len() {
                 worker_id = 0;
             }
-
-            if workers_tasks.workers_waiting_status.is_waiting(worker_id) {
-                workers_tasks
-                    .workers_waiting_status
-                    .wakeup_worker(worker_id);
-                condvar.wakeup(worker_id);
-            }
         }
+
+        workers_tasks.wakeup_all_waiting_workers(&condvar);
     }
 
     pub fn completed_async_task(&self, condvar: Arc<WorkersCondvar>, task: CompletedAsyncTask) {

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

@@ -109,7 +109,7 @@ impl QueriesPipelineExecutor {
         for thread_num in 0..threads {
             let this = self.clone();
             #[allow(unused_mut)]
-            let mut name = format!("PipelineExecutor-{}", thread_num);
+            let mut name = format!("QueriesPipelineExecutor-{}", thread_num);
 
             #[cfg(debug_assertions)]
             {