fix: r_cte wrong/flaky results under concurrency

src/query/service/src/pipelines/processors/transforms/transform_recursive_cte_source.rs

@@ -84,6 +84,16 @@ impl TransformRecursiveCteSource {
         union_plan: UnionAll,
     ) -> Result<ProcessorPtr> {
         let mut union_plan = union_plan;
+
+        // Recursive CTE uses internal MEMORY tables addressed by name in the current database.
+        // If we keep using the stable scan name (cte name/alias), concurrent queries can interfere
+        // by creating/dropping/recreating the same table name, leading to wrong or flaky results.
+        //
+        // Make the internal table names query-unique by prefixing them with the query id.
+        // This is purely internal and does not change user-visible semantics.
+        let rcte_prefix = make_rcte_prefix(&ctx.get_id());
+        rewrite_rcte_internal_table_names(&mut union_plan, &rcte_prefix);
+
         let mut exec_ids: HashMap<String, Vec<u64>> = HashMap::new();
         assign_exec_ids(&mut union_plan.left, &mut exec_ids);
         assign_exec_ids(&mut union_plan.right, &mut exec_ids);
@@ -134,6 +144,8 @@ impl TransformRecursiveCteSource {
         if ctx.get_settings().get_max_cte_recursive_depth()? < recursive_step {
             return Err(ErrorCode::Internal("Recursive depth is reached"));
         }
+        #[cfg(test)]
+        crate::test_kits::rcte_hooks::maybe_pause_before_step(&ctx.get_id(), recursive_step).await;
         let mut cte_scan_tables = vec![];
         let plan = if recursive_step == 0 {
             // Find all cte scan in the union right child plan, then create memory table for them.
@@ -172,6 +184,42 @@ impl TransformRecursiveCteSource {
     }
 }
 
+fn make_rcte_prefix(query_id: &str) -> String {
+    // Keep it readable and safe as an identifier.
+    // Use enough entropy to be effectively unique for concurrent queries.
+    let mut short = String::with_capacity(32);
+    for ch in query_id.chars() {
+        if ch.is_ascii_alphanumeric() {
+            short.push(ch);
+        }
+        if short.len() >= 32 {
+            break;
+        }
+    }
+    if short.is_empty() {
+        short.push_str("unknown");
+    }
+    format!("__rcte_{short}_")
+}
+
+fn rewrite_rcte_internal_table_names(union_plan: &mut UnionAll, prefix: &str) {
+    rewrite_recursive_cte_scan_names(&mut union_plan.left, prefix);
+    rewrite_recursive_cte_scan_names(&mut union_plan.right, prefix);
+    for name in union_plan.cte_scan_names.iter_mut() {
+        *name = format!("{prefix}{name}");
+    }
+}
+
+fn rewrite_recursive_cte_scan_names(plan: &mut PhysicalPlan, prefix: &str) {
+    if let Some(recursive_cte_scan) = RecursiveCteScan::from_mut_physical_plan(plan) {
+        recursive_cte_scan.table_name = format!("{prefix}{}", recursive_cte_scan.table_name);
+    }
+
+    for child in plan.children_mut() {
+        rewrite_recursive_cte_scan_names(child, prefix);
+    }
+}
+
 #[async_trait::async_trait]
 impl AsyncSource for TransformRecursiveCteSource {
     const NAME: &'static str = "TransformRecursiveCteSource";
@@ -311,7 +359,6 @@ async fn create_memory_table_for_cte_scan(
 
                 let mut options = BTreeMap::new();
                 options.insert(OPT_KEY_RECURSIVE_CTE.to_string(), "1".to_string());
-
                 self.plans.push(CreateTablePlan {
                     schema,
                     create_option: CreateOption::CreateIfNotExists,

src/query/service/src/test_kits/mod.rs

@@ -31,3 +31,4 @@ pub use config::config_with_spill;
 pub use context::*;
 pub use fixture::*;
 pub use fuse::*;
+pub mod rcte_hooks;

src/query/service/src/test_kits/rcte_hooks.rs

@@ -0,0 +1,157 @@
+// 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.
+
+//! Test-only hooks for recursive CTE execution.
+//!
+//! This module is intended to make race conditions reproducible by providing
+//! deterministic pause/resume points in the recursive CTE executor.
+//!
+//! By default no hooks are installed and the hook checks are no-ops.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::sync::OnceLock;
+use std::sync::atomic::AtomicUsize;
+use std::sync::atomic::Ordering;
+
+use tokio::sync::Notify;
+
+static HOOKS: OnceLock<Arc<RcteHookRegistry>> = OnceLock::new();
+
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+struct GateKey {
+    query_id: String,
+    step: usize,
+}
+
+impl GateKey {
+    fn new(query_id: &str, step: usize) -> Self {
+        Self {
+            query_id: query_id.to_string(),
+            step,
+        }
+    }
+}
+
+#[derive(Default)]
+pub struct RcteHookRegistry {
+    gates: Mutex<HashMap<GateKey, Arc<PauseGate>>>,
+}
+
+impl RcteHookRegistry {
+    pub fn global() -> Arc<RcteHookRegistry> {
+        HOOKS
+            .get_or_init(|| Arc::new(RcteHookRegistry::default()))
+            .clone()
+    }
+
+    pub fn install_pause_before_step(&self, query_id: &str, step: usize) -> Arc<PauseGate> {
+        let mut gates = self.gates.lock().unwrap();
+        let key = GateKey::new(query_id, step);
+        gates
+            .entry(key)
+            .or_insert_with(|| Arc::new(PauseGate::new(step)))
+            .clone()
+    }
+
+    fn get_gate(&self, query_id: &str, step: usize) -> Option<Arc<PauseGate>> {
+        let key = GateKey::new(query_id, step);
+        self.gates.lock().unwrap().get(&key).cloned()
+    }
+}
+
+/// A reusable pause gate for a single step number.
+///
+/// When the code hits the hook point, it increments `arrived` and blocks until
+/// the test releases the same hit index via `release(hit_no)`.
+pub struct PauseGate {
+    step: usize,
+    arrived: AtomicUsize,
+    released: AtomicUsize,
+    arrived_notify: Notify,
+    released_notify: Notify,
+}
+
+impl PauseGate {
+    fn new(step: usize) -> Self {
+        Self {
+            step,
+            arrived: AtomicUsize::new(0),
+            released: AtomicUsize::new(0),
+            arrived_notify: Notify::new(),
+            released_notify: Notify::new(),
+        }
+    }
+
+    pub fn step(&self) -> usize {
+        self.step
+    }
+
+    pub fn arrived(&self) -> usize {
+        self.arrived.load(Ordering::Acquire)
+    }
+
+    pub async fn wait_arrived_at_least(&self, n: usize) {
+        loop {
+            if self.arrived() >= n {
+                return;
+            }
+            self.arrived_notify.notified().await;
+        }
+    }
+
+    /// Release the `hit_no`-th arrival (1-based).
+    pub fn release(&self, hit_no: usize) {
+        // Monotonic release.
+        let mut cur = self.released.load(Ordering::Acquire);
+        while cur < hit_no {
+            match self
+                .released
+                .compare_exchange(cur, hit_no, Ordering::AcqRel, Ordering::Acquire)
+            {
+                Ok(_) => break,
+                Err(v) => cur = v,
+            }
+        }
+        self.released_notify.notify_waiters();
+    }
+
+    async fn hit(&self) {
+        let hit_no = self.arrived.fetch_add(1, Ordering::AcqRel) + 1;
+        self.arrived_notify.notify_waiters();
+
+        loop {
+            let released = self.released.load(Ordering::Acquire);
+            if released >= hit_no {
+                return;
+            }
+            self.released_notify.notified().await;
+        }
+    }
+}
+
+/// Called from the recursive CTE executor.
+///
+/// If a pause gate is installed for `step`, this call will block until released.
+#[async_backtrace::framed]
+pub async fn maybe_pause_before_step(query_id: &str, step: usize) {
+    let Some(registry) = HOOKS.get() else {
+        return;
+    };
+    let Some(gate) = registry.get_gate(query_id, step) else {
+        return;
+    };
+    gate.hit().await;
+}

src/query/service/tests/it/sql/mod.rs

@@ -15,3 +15,4 @@
 mod exec;
 mod expr;
 mod planner;
+mod recursive_cte;