reddb_server/storage/query/executors/
cte.rs

1//! CTE (Common Table Expression) executor.
2//!
3//! Implements the `WITH … AS (…) SELECT …` SQL standard form. CTEs
4//! are materialised as temporary result sets that the main query can
5//! reference by name. Recursive CTEs (`WITH RECURSIVE`) use the
6//! classic iterative-fixpoint shape:
7//!
8//! 1. Execute the non-recursive (base) part once.
9//! 2. Repeat: execute the recursive part with the previous iteration's
10//!    rows visible as the CTE.
11//! 3. Stop when no new rows are produced (or guard limits trip).
12//!
13//! Today only non-recursive CTEs are wired through the runtime —
14//! `inline_ctes` rejects `WITH RECURSIVE` with a clear error. The
15//! iterative-fixpoint code in `CteExecutor` is reachable only via
16//! direct unit tests and is the basis for the future recursive wire-
17//! up (#41 follow-up).
18//!
19//! # Example
20//!
21//! ```ignore
22//! WITH active AS (
23//!     SELECT id, name FROM users WHERE status = 'active'
24//! )
25//! SELECT * FROM active
26//! ```
27
28use std::collections::{HashMap, HashSet};
29
30use super::super::unified::{ExecutionError, UnifiedRecord, UnifiedResult};
31use reddb_rql::ast::{
32    CteDefinition, Expr, Filter, Projection, QueryExpr, QueryWithCte, SelectItem, TableSource,
33};
34use reddb_types::Value;
35
36/// Maximum recursion depth to prevent infinite loops
37const MAX_RECURSION_DEPTH: usize = 1000;
38
39/// Maximum total rows across all iterations
40const MAX_RECURSIVE_ROWS: usize = 100_000;
41
42/// CTE execution context holding materialized CTE results
43#[derive(Debug, Clone, Default)]
44pub struct CteContext {
45    /// Materialized CTE results by name
46    tables: HashMap<String, UnifiedResult>,
47    /// Track which CTEs are currently being evaluated (for cycle detection)
48    evaluating: HashSet<String>,
49    /// Statistics
50    stats: CteStats,
51}
52
53impl CteContext {
54    /// Create a new CTE context
55    pub fn new() -> Self {
56        Self::default()
57    }
58
59    /// Get a materialized CTE result by name
60    pub fn get(&self, name: &str) -> Option<&UnifiedResult> {
61        self.tables.get(name)
62    }
63
64    /// Store a materialized CTE result
65    pub fn store(&mut self, name: String, result: UnifiedResult) {
66        self.tables.insert(name, result);
67    }
68
69    /// Check if a CTE is being evaluated (for recursion detection)
70    pub fn is_evaluating(&self, name: &str) -> bool {
71        self.evaluating.contains(name)
72    }
73
74    /// Mark a CTE as being evaluated
75    pub fn start_evaluating(&mut self, name: &str) {
76        self.evaluating.insert(name.to_string());
77    }
78
79    /// Mark a CTE as done evaluating
80    pub fn done_evaluating(&mut self, name: &str) {
81        self.evaluating.remove(name);
82    }
83
84    /// Get execution statistics
85    pub fn stats(&self) -> &CteStats {
86        &self.stats
87    }
88}
89
90/// Statistics about CTE execution
91#[derive(Debug, Clone, Default)]
92pub struct CteStats {
93    /// Number of CTEs executed
94    pub ctes_executed: usize,
95    /// Number of recursive iterations
96    pub recursive_iterations: usize,
97    /// Total rows produced by CTEs
98    pub rows_produced: usize,
99    /// Execution time in microseconds
100    pub exec_time_us: u64,
101}
102
103/// CTE Executor
104pub struct CteExecutor<F>
105where
106    F: Fn(&QueryExpr, &CteContext) -> Result<UnifiedResult, ExecutionError>,
107{
108    /// Function to execute a query with CTE context
109    execute_fn: F,
110}
111
112impl<F> CteExecutor<F>
113where
114    F: Fn(&QueryExpr, &CteContext) -> Result<UnifiedResult, ExecutionError>,
115{
116    /// Create a new CTE executor
117    pub fn new(execute_fn: F) -> Self {
118        Self { execute_fn }
119    }
120
121    /// Execute a query with CTEs
122    pub fn execute(&self, query: &QueryWithCte) -> Result<UnifiedResult, ExecutionError> {
123        let start = std::time::Instant::now();
124        let mut ctx = CteContext::new();
125
126        // Materialize all CTEs in order
127        if let Some(ref with_clause) = query.with_clause {
128            for cte in &with_clause.ctes {
129                self.materialize_cte(cte, &mut ctx)?;
130            }
131        }
132
133        // Execute the main query with CTE context
134        let result = (self.execute_fn)(&query.query, &ctx)?;
135
136        ctx.stats.exec_time_us = start.elapsed().as_micros() as u64;
137        Ok(result)
138    }
139
140    /// Materialize a single CTE
141    fn materialize_cte(
142        &self,
143        cte: &CteDefinition,
144        ctx: &mut CteContext,
145    ) -> Result<(), ExecutionError> {
146        if ctx.is_evaluating(&cte.name) {
147            return Err(ExecutionError::new(format!(
148                "Circular CTE reference: {}",
149                cte.name
150            )));
151        }
152
153        // Check if already materialized
154        if ctx.get(&cte.name).is_some() {
155            return Ok(());
156        }
157
158        ctx.start_evaluating(&cte.name);
159
160        let result = if cte.recursive {
161            self.execute_recursive_cte(cte, ctx)?
162        } else {
163            // Simple CTE: execute once
164            let result = (self.execute_fn)(&cte.query, ctx)?;
165            self.project_columns(&result, &cte.columns)
166        };
167
168        ctx.stats.ctes_executed += 1;
169        ctx.stats.rows_produced += result.len();
170        ctx.store(cte.name.clone(), result);
171        ctx.done_evaluating(&cte.name);
172
173        Ok(())
174    }
175
176    /// Execute a recursive CTE using iterative fixpoint
177    fn execute_recursive_cte(
178        &self,
179        cte: &CteDefinition,
180        ctx: &mut CteContext,
181    ) -> Result<UnifiedResult, ExecutionError> {
182        // For recursive CTEs, we need to handle UNION ALL structure
183        // The query should be: base_query UNION ALL recursive_query
184        //
185        // Algorithm:
186        // 1. Execute base query -> working_table
187        // 2. result_table = working_table
188        // 3. While working_table not empty:
189        //    a. Execute recursive query with CTE = working_table
190        //    b. new_rows = result - already_seen
191        //    c. working_table = new_rows
192        //    d. result_table += new_rows
193        // 4. Return result_table
194
195        // For simplicity in first implementation, we execute the full query
196        // iteratively, building up the working table
197
198        let mut all_results = UnifiedResult::with_columns(cte.columns.clone());
199        let mut working_table = UnifiedResult::with_columns(cte.columns.clone());
200        let mut seen_rows: HashSet<u64> = HashSet::new();
201        let mut iteration = 0;
202
203        // First iteration: execute the full query (base case)
204        let initial = (self.execute_fn)(&cte.query, ctx)?;
205        let initial = self.project_columns(&initial, &cte.columns);
206
207        for record in &initial.records {
208            let hash = self.hash_record(record);
209            if seen_rows.insert(hash) {
210                working_table.push(record.clone());
211                all_results.push(record.clone());
212            }
213        }
214
215        // Store initial results so recursive references can see them
216        ctx.store(cte.name.clone(), working_table.clone());
217
218        // Iterate until fixpoint
219        while !working_table.is_empty() && iteration < MAX_RECURSION_DEPTH {
220            iteration += 1;
221            ctx.stats.recursive_iterations += 1;
222
223            if all_results.len() > MAX_RECURSIVE_ROWS {
224                return Err(ExecutionError::new(format!(
225                    "Recursive CTE '{}' exceeded maximum rows ({})",
226                    cte.name, MAX_RECURSIVE_ROWS
227                )));
228            }
229
230            // Execute query with current CTE contents
231            let new_results = (self.execute_fn)(&cte.query, ctx)?;
232            let new_results = self.project_columns(&new_results, &cte.columns);
233
234            // Find genuinely new rows
235            let mut new_working_table = UnifiedResult::with_columns(cte.columns.clone());
236            for record in &new_results.records {
237                let hash = self.hash_record(record);
238                if seen_rows.insert(hash) {
239                    new_working_table.push(record.clone());
240                    all_results.push(record.clone());
241                }
242            }
243
244            working_table = new_working_table;
245
246            // Update CTE table for next iteration
247            ctx.store(cte.name.clone(), all_results.clone());
248        }
249
250        if iteration >= MAX_RECURSION_DEPTH && !working_table.is_empty() {
251            return Err(ExecutionError::new(format!(
252                "Recursive CTE '{}' exceeded maximum recursion depth ({})",
253                cte.name, MAX_RECURSION_DEPTH
254            )));
255        }
256
257        Ok(all_results)
258    }
259
260    /// Project result columns according to CTE column list
261    fn project_columns(&self, result: &UnifiedResult, columns: &[String]) -> UnifiedResult {
262        if columns.is_empty() {
263            return result.clone();
264        }
265
266        let mut projected = UnifiedResult::with_columns(columns.to_vec());
267
268        for record in &result.records {
269            let mut new_record = UnifiedRecord::new();
270
271            // Map result columns to CTE columns
272            for (i, col) in columns.iter().enumerate() {
273                // Try to find value by position first, then by name
274                let value = result
275                    .columns
276                    .get(i)
277                    .and_then(|orig_col| record.get(orig_col))
278                    .cloned()
279                    .or_else(|| record.get(col).cloned())
280                    .unwrap_or(Value::Null);
281
282                new_record.set(col, value);
283            }
284
285            projected.push(new_record);
286        }
287
288        projected
289    }
290
291    /// Hash a record for deduplication
292    fn hash_record(&self, record: &UnifiedRecord) -> u64 {
293        use std::collections::hash_map::DefaultHasher;
294        use std::hash::{Hash, Hasher};
295
296        let mut hasher = DefaultHasher::new();
297
298        // Hash all values in deterministic order
299        let mut keys = record.column_names();
300        keys.sort();
301
302        for key in &keys {
303            (**key).hash(&mut hasher);
304            if let Some(value) = record.get(key) {
305                Self::hash_value(value, &mut hasher);
306            }
307        }
308
309        hasher.finish()
310    }
311
312    /// Hash a Value for deduplication
313    fn hash_value(value: &Value, hasher: &mut impl std::hash::Hasher) {
314        use std::hash::Hash;
315
316        match value {
317            Value::Null => 0u8.hash(hasher),
318            Value::Temperature(temperature) => {
319                56u8.hash(hasher);
320                temperature.hash(hasher);
321            }
322            Value::Boolean(b) => {
323                1u8.hash(hasher);
324                b.hash(hasher);
325            }
326            Value::Integer(i) => {
327                2u8.hash(hasher);
328                i.hash(hasher);
329            }
330            Value::UnsignedInteger(u) => {
331                3u8.hash(hasher);
332                u.hash(hasher);
333            }
334            Value::Float(f) => {
335                4u8.hash(hasher);
336                f.to_bits().hash(hasher);
337            }
338            Value::Text(s) => {
339                5u8.hash(hasher);
340                s.hash(hasher);
341            }
342            Value::Blob(b) => {
343                6u8.hash(hasher);
344                b.hash(hasher);
345            }
346            Value::Timestamp(t) => {
347                7u8.hash(hasher);
348                t.hash(hasher);
349            }
350            Value::Duration(d) => {
351                8u8.hash(hasher);
352                d.hash(hasher);
353            }
354            Value::IpAddr(addr) => {
355                9u8.hash(hasher);
356                match addr {
357                    std::net::IpAddr::V4(v4) => v4.octets().hash(hasher),
358                    std::net::IpAddr::V6(v6) => v6.octets().hash(hasher),
359                }
360            }
361            Value::MacAddr(mac) => {
362                10u8.hash(hasher);
363                mac.hash(hasher);
364            }
365            Value::Vector(v) => {
366                11u8.hash(hasher);
367                v.len().hash(hasher);
368                for f in v {
369                    f.to_bits().hash(hasher);
370                }
371            }
372            Value::Json(j) => {
373                12u8.hash(hasher);
374                j.hash(hasher);
375            }
376            Value::Uuid(u) => {
377                13u8.hash(hasher);
378                u.hash(hasher);
379            }
380            Value::NodeRef(n) => {
381                14u8.hash(hasher);
382                n.hash(hasher);
383            }
384            Value::EdgeRef(e) => {
385                15u8.hash(hasher);
386                e.hash(hasher);
387            }
388            Value::VectorRef(coll, id) => {
389                16u8.hash(hasher);
390                coll.hash(hasher);
391                id.hash(hasher);
392            }
393            Value::RowRef(table, id) => {
394                17u8.hash(hasher);
395                table.hash(hasher);
396                id.hash(hasher);
397            }
398            Value::Color(rgb) => {
399                18u8.hash(hasher);
400                rgb.hash(hasher);
401            }
402            Value::Email(s) => {
403                19u8.hash(hasher);
404                s.hash(hasher);
405            }
406            Value::Url(s) => {
407                20u8.hash(hasher);
408                s.hash(hasher);
409            }
410            Value::Phone(n) => {
411                21u8.hash(hasher);
412                n.hash(hasher);
413            }
414            Value::Semver(v) => {
415                22u8.hash(hasher);
416                v.hash(hasher);
417            }
418            Value::Cidr(ip, prefix) => {
419                23u8.hash(hasher);
420                ip.hash(hasher);
421                prefix.hash(hasher);
422            }
423            Value::Date(d) => {
424                24u8.hash(hasher);
425                d.hash(hasher);
426            }
427            Value::Time(t) => {
428                25u8.hash(hasher);
429                t.hash(hasher);
430            }
431            Value::Decimal(v) => {
432                26u8.hash(hasher);
433                v.hash(hasher);
434            }
435            Value::DecimalText(v) => {
436                52u8.hash(hasher);
437                v.hash(hasher);
438            }
439            Value::EnumValue(i) => {
440                27u8.hash(hasher);
441                i.hash(hasher);
442            }
443            Value::Array(elems) => {
444                28u8.hash(hasher);
445                elems.len().hash(hasher);
446                for elem in elems {
447                    Self::hash_value(elem, hasher);
448                }
449            }
450            Value::TimestampMs(v) => {
451                29u8.hash(hasher);
452                v.hash(hasher);
453            }
454            Value::Ipv4(v) => {
455                30u8.hash(hasher);
456                v.hash(hasher);
457            }
458            Value::Ipv6(bytes) => {
459                31u8.hash(hasher);
460                bytes.hash(hasher);
461            }
462            Value::Subnet(ip, mask) => {
463                32u8.hash(hasher);
464                ip.hash(hasher);
465                mask.hash(hasher);
466            }
467            Value::Port(v) => {
468                33u8.hash(hasher);
469                v.hash(hasher);
470            }
471            Value::Latitude(v) => {
472                34u8.hash(hasher);
473                v.hash(hasher);
474            }
475            Value::Longitude(v) => {
476                35u8.hash(hasher);
477                v.hash(hasher);
478            }
479            Value::GeoPoint(lat, lon) => {
480                36u8.hash(hasher);
481                lat.hash(hasher);
482                lon.hash(hasher);
483            }
484            Value::Country2(c) => {
485                37u8.hash(hasher);
486                c.hash(hasher);
487            }
488            Value::Country3(c) => {
489                38u8.hash(hasher);
490                c.hash(hasher);
491            }
492            Value::Lang2(c) => {
493                39u8.hash(hasher);
494                c.hash(hasher);
495            }
496            Value::Lang5(c) => {
497                40u8.hash(hasher);
498                c.hash(hasher);
499            }
500            Value::Currency(c) => {
501                41u8.hash(hasher);
502                c.hash(hasher);
503            }
504            Value::AssetCode(code) => {
505                50u8.hash(hasher);
506                code.hash(hasher);
507            }
508            Value::Money {
509                asset_code,
510                minor_units,
511                scale,
512            } => {
513                51u8.hash(hasher);
514                asset_code.hash(hasher);
515                minor_units.hash(hasher);
516                scale.hash(hasher);
517            }
518            Value::ColorAlpha(rgba) => {
519                42u8.hash(hasher);
520                rgba.hash(hasher);
521            }
522            Value::BigInt(v) => {
523                43u8.hash(hasher);
524                v.hash(hasher);
525            }
526            Value::KeyRef(col, key) => {
527                44u8.hash(hasher);
528                col.hash(hasher);
529                key.hash(hasher);
530            }
531            Value::DocRef(col, id) => {
532                45u8.hash(hasher);
533                col.hash(hasher);
534                id.hash(hasher);
535            }
536            Value::TableRef(name) => {
537                46u8.hash(hasher);
538                name.hash(hasher);
539            }
540            Value::PageRef(page_id) => {
541                47u8.hash(hasher);
542                page_id.hash(hasher);
543            }
544            Value::Secret(bytes) => {
545                48u8.hash(hasher);
546                bytes.hash(hasher);
547            }
548            Value::Password(hash) => {
549                49u8.hash(hasher);
550                hash.hash(hasher);
551            }
552        }
553    }
554}
555
556/// Helper to parse UNION structure for recursive CTEs
557pub fn split_union_parts(query: &QueryExpr) -> Option<(QueryExpr, QueryExpr)> {
558    // UNION support is not represented in the current AST; recursive queries execute
559    // the full body expression each iteration.
560    let _ = query;
561    None
562}
563
564// ─────────────────────────────────────────────────────────────────────
565// CTE inlining (#41) — non-recursive
566//
567// Rewrites a `QueryWithCte` into a plain `QueryExpr` by walking the
568// AST and substituting every `TableSource::Name(name)` (or legacy
569// `TableQuery.table` field) that matches a CTE name with
570// `TableSource::Subquery(cte.query)`. After this pass the runtime's
571// existing subquery-in-FROM machinery executes the result with no
572// CTE-specific dispatch needed.
573//
574// Recursive CTEs are rejected up-front — the iterative fixpoint
575// strategy is implemented in `CteExecutor` but is not wired into the
576// runtime yet (separate slice).
577// ─────────────────────────────────────────────────────────────────────
578
579/// Inline a `QueryWithCte`'s WITH clause into its inner query. Returns
580/// the rewritten `QueryExpr` ready for dispatch. Recursive CTEs are
581/// rejected with a clear error.
582pub fn inline_ctes(query: QueryWithCte) -> Result<QueryExpr, ExecutionError> {
583    let Some(with_clause) = query.with_clause else {
584        return Ok(query.query);
585    };
586    if with_clause.has_recursive {
587        return Err(ExecutionError::new(
588            "WITH RECURSIVE is not yet supported by the executor; \
589             non-recursive WITH clauses run today, recursive support \
590             is tracked separately"
591                .to_string(),
592        ));
593    }
594
595    // Inline each CTE into its successors first so chained CTEs
596    // (`WITH a AS (...), b AS (... a ...)`) end up with fully resolved
597    // bodies before they're substituted into the outer query.
598    let mut resolved: HashMap<String, QueryExpr> = HashMap::new();
599    for cte in &with_clause.ctes {
600        let mut body = (*cte.query).clone();
601        rewrite(&mut body, &resolved);
602        resolved.insert(cte.name.clone(), body);
603    }
604
605    let mut outer = query.query;
606    rewrite(&mut outer, &resolved);
607    Ok(outer)
608}
609
610/// Walk a `QueryExpr` and replace any table reference whose name
611/// matches a key in `ctes` with the inlined CTE body. Recurses
612/// through `Join` and nested `Subquery` sources so CTE refs inside
613/// JOINs and subqueries resolve too. Mirrors the view-rewrite
614/// convention: when the outer table reference carries filter / limit
615/// / offset constraints we wrap the body in a `Subquery` to preserve
616/// them; otherwise we replace the whole `Table` node verbatim with
617/// the CTE body so dispatchers that key off `QueryExpr::Table` (like
618/// the JOIN executor) see the right shape.
619fn rewrite(expr: &mut QueryExpr, ctes: &HashMap<String, QueryExpr>) {
620    match expr {
621        QueryExpr::Table(tq) => {
622            let lookup_name = match &tq.source {
623                Some(TableSource::Subquery(_)) => None,
624                Some(TableSource::Name(n)) => Some(n.clone()),
625                // Table-valued functions are not CTE references (issue #795);
626                // the inline-graph form likewise references no CTE (issue #799).
627                Some(TableSource::Function { .. } | TableSource::InlineGraphFunction { .. }) => {
628                    None
629                }
630                None => Some(tq.table.clone()),
631            };
632
633            if let Some(name) = lookup_name {
634                if let Some(body) = ctes.get(&name) {
635                    let outer_has_constraints = tq.filter.is_some()
636                        || tq.where_expr.is_some()
637                        || tq.limit.is_some()
638                        || tq.offset.is_some()
639                        || !tq.columns.is_empty()
640                        || !tq.select_items.is_empty()
641                        || !tq.group_by.is_empty()
642                        || !tq.order_by.is_empty();
643
644                    if outer_has_constraints {
645                        // Outer ref carries projections / filters /
646                        // limits — keep those by wrapping the body in
647                        // a subquery source. Sentinel name so legacy
648                        // `table` consumers can't resolve it against
649                        // the real schema.
650                        tq.source = Some(TableSource::Subquery(Box::new(body.clone())));
651                        tq.table = format!("__cte_{name}");
652                    } else {
653                        // Bare `FROM cte` (possibly with alias) —
654                        // replace verbatim so JOIN / dispatch paths
655                        // see the CTE body's natural shape.
656                        *expr = body.clone();
657                    }
658                    return;
659                }
660            }
661
662            match tq.source.as_mut() {
663                Some(TableSource::Subquery(body)) => rewrite(body, ctes),
664                Some(TableSource::InlineGraphFunction { nodes, edges, .. }) => {
665                    rewrite(nodes, ctes);
666                    rewrite(edges, ctes);
667                }
668                _ => {}
669            }
670            rewrite_table_query_parts(tq, ctes);
671        }
672        QueryExpr::Join(jq) => {
673            rewrite(&mut jq.left, ctes);
674            rewrite(&mut jq.right, ctes);
675            if let Some(filter) = jq.filter.as_mut() {
676                rewrite_filter(filter, ctes);
677            }
678            for item in &mut jq.order_by {
679                if let Some(expr) = item.expr.as_mut() {
680                    rewrite_expr(expr, ctes);
681                }
682            }
683            for item in &mut jq.return_items {
684                rewrite_select_item(item, ctes);
685            }
686            for projection in &mut jq.return_ {
687                rewrite_projection(projection, ctes);
688            }
689        }
690        QueryExpr::Graph(gq) => {
691            if let Some(filter) = gq.filter.as_mut() {
692                rewrite_filter(filter, ctes);
693            }
694            for projection in &mut gq.return_ {
695                rewrite_projection(projection, ctes);
696            }
697        }
698        _ => {}
699    }
700}
701
702fn rewrite_table_query_parts(
703    tq: &mut reddb_rql::ast::TableQuery,
704    ctes: &HashMap<String, QueryExpr>,
705) {
706    for item in &mut tq.select_items {
707        rewrite_select_item(item, ctes);
708    }
709    for projection in &mut tq.columns {
710        rewrite_projection(projection, ctes);
711    }
712    if let Some(expr) = tq.where_expr.as_mut() {
713        rewrite_expr(expr, ctes);
714    }
715    if let Some(filter) = tq.filter.as_mut() {
716        rewrite_filter(filter, ctes);
717    }
718    for expr in &mut tq.group_by_exprs {
719        rewrite_expr(expr, ctes);
720    }
721    if let Some(expr) = tq.having_expr.as_mut() {
722        rewrite_expr(expr, ctes);
723    }
724    if let Some(filter) = tq.having.as_mut() {
725        rewrite_filter(filter, ctes);
726    }
727    for item in &mut tq.order_by {
728        if let Some(expr) = item.expr.as_mut() {
729            rewrite_expr(expr, ctes);
730        }
731    }
732}
733
734fn rewrite_select_item(item: &mut SelectItem, ctes: &HashMap<String, QueryExpr>) {
735    if let SelectItem::Expr { expr, .. } = item {
736        rewrite_expr(expr, ctes);
737    }
738}
739
740fn rewrite_projection(projection: &mut Projection, ctes: &HashMap<String, QueryExpr>) {
741    match projection {
742        Projection::Function(_, args) => {
743            for arg in args {
744                rewrite_projection(arg, ctes);
745            }
746        }
747        Projection::Expression(filter, _) => rewrite_filter(filter, ctes),
748        Projection::Window { args, window, .. } => {
749            for arg in args {
750                rewrite_projection(arg, ctes);
751            }
752            for expr in &mut window.partition_by {
753                rewrite_expr(expr, ctes);
754            }
755            for item in &mut window.order_by {
756                rewrite_expr(&mut item.expr, ctes);
757            }
758        }
759        Projection::All
760        | Projection::Column(_)
761        | Projection::Alias(_, _)
762        | Projection::Field(_, _) => {}
763    }
764}
765
766fn rewrite_filter(filter: &mut Filter, ctes: &HashMap<String, QueryExpr>) {
767    match filter {
768        Filter::CompareExpr { lhs, rhs, .. } => {
769            rewrite_expr(lhs, ctes);
770            rewrite_expr(rhs, ctes);
771        }
772        Filter::And(left, right) | Filter::Or(left, right) => {
773            rewrite_filter(left, ctes);
774            rewrite_filter(right, ctes);
775        }
776        Filter::Not(inner) => rewrite_filter(inner, ctes),
777        Filter::Compare { .. }
778        | Filter::CompareFields { .. }
779        | Filter::IsNull(_)
780        | Filter::IsNotNull(_)
781        | Filter::In { .. }
782        | Filter::Between { .. }
783        | Filter::Like { .. }
784        | Filter::StartsWith { .. }
785        | Filter::EndsWith { .. }
786        | Filter::Contains { .. } => {}
787    }
788}
789
790fn rewrite_expr(expr: &mut Expr, ctes: &HashMap<String, QueryExpr>) {
791    match expr {
792        Expr::BinaryOp { lhs, rhs, .. } => {
793            rewrite_expr(lhs, ctes);
794            rewrite_expr(rhs, ctes);
795        }
796        Expr::UnaryOp { operand, .. } => rewrite_expr(operand, ctes),
797        Expr::Cast { inner, .. } => rewrite_expr(inner, ctes),
798        Expr::FunctionCall { args, .. } => {
799            for arg in args {
800                rewrite_expr(arg, ctes);
801            }
802        }
803        Expr::Case {
804            branches, else_, ..
805        } => {
806            for (condition, value) in branches {
807                rewrite_expr(condition, ctes);
808                rewrite_expr(value, ctes);
809            }
810            if let Some(value) = else_ {
811                rewrite_expr(value, ctes);
812            }
813        }
814        Expr::IsNull { operand, .. } => rewrite_expr(operand, ctes),
815        Expr::InList { target, values, .. } => {
816            rewrite_expr(target, ctes);
817            for value in values {
818                rewrite_expr(value, ctes);
819            }
820        }
821        Expr::Between {
822            target, low, high, ..
823        } => {
824            rewrite_expr(target, ctes);
825            rewrite_expr(low, ctes);
826            rewrite_expr(high, ctes);
827        }
828        Expr::Subquery { query, .. } => rewrite(&mut query.query, ctes),
829        Expr::WindowFunctionCall { args, window, .. } => {
830            for arg in args {
831                rewrite_expr(arg, ctes);
832            }
833            for expr in &mut window.partition_by {
834                rewrite_expr(expr, ctes);
835            }
836            for item in &mut window.order_by {
837                rewrite_expr(&mut item.expr, ctes);
838            }
839        }
840        Expr::Literal { .. } | Expr::Column { .. } | Expr::Parameter { .. } => {}
841    }
842}
843
844// ============================================================================
845// Tests
846// ============================================================================
847
848#[cfg(test)]
849mod tests {
850    use super::*;
851    use reddb_rql::ast::CteQueryBuilder;
852    use reddb_rql::ast::WithClause;
853
854    fn mock_execute(
855        _query: &QueryExpr,
856        _ctx: &CteContext,
857    ) -> Result<UnifiedResult, ExecutionError> {
858        // Simple mock that returns empty result
859        Ok(UnifiedResult::empty())
860    }
861
862    #[test]
863    fn test_cte_context() {
864        let mut ctx = CteContext::new();
865
866        // Test empty context
867        assert!(ctx.get("test").is_none());
868        assert!(!ctx.is_evaluating("test"));
869
870        // Test storing results
871        let result = UnifiedResult::with_columns(vec!["col1".to_string()]);
872        ctx.store("test".to_string(), result);
873        assert!(ctx.get("test").is_some());
874
875        // Test evaluation tracking
876        ctx.start_evaluating("other");
877        assert!(ctx.is_evaluating("other"));
878        ctx.done_evaluating("other");
879        assert!(!ctx.is_evaluating("other"));
880    }
881
882    #[test]
883    fn test_simple_cte_execution() {
884        let executor = CteExecutor::new(|_query, _ctx| {
885            let mut result = UnifiedResult::with_columns(vec!["id".to_string()]);
886            let mut record = UnifiedRecord::new();
887            record.set("id", Value::Integer(1));
888            result.push(record);
889            Ok(result)
890        });
891
892        // Create a simple CTE query
893        let cte = CteDefinition {
894            name: "test_cte".to_string(),
895            columns: vec!["id".to_string()],
896            query: Box::new(QueryExpr::table("dummy").build()),
897            recursive: false,
898        };
899
900        let with_clause = WithClause::new().add(cte);
901        let query = QueryWithCte::with_ctes(with_clause, QueryExpr::table("test_cte").build());
902
903        let result = executor.execute(&query);
904        assert!(result.is_ok());
905    }
906
907    #[test]
908    fn test_cte_builder() {
909        let query = CteQueryBuilder::new()
910            .cte_with_columns(
911                "nums",
912                vec!["n".to_string()],
913                QueryExpr::table("numbers").build(),
914            )
915            .build(QueryExpr::table("nums").build());
916
917        assert!(query.with_clause.is_some());
918        let with_clause = query.with_clause.unwrap();
919        assert_eq!(with_clause.ctes.len(), 1);
920        assert_eq!(with_clause.ctes[0].name, "nums");
921    }
922
923    #[test]
924    fn test_recursive_cte_builder() {
925        let query = CteQueryBuilder::new()
926            .recursive_cte("paths", QueryExpr::table("connections").build())
927            .build(QueryExpr::table("paths").build());
928
929        assert!(query.with_clause.is_some());
930        let with_clause = query.with_clause.unwrap();
931        assert!(with_clause.has_recursive);
932        assert!(with_clause.ctes[0].recursive);
933    }
934
935    #[test]
936    fn test_circular_reference_detection() {
937        let mut ctx = CteContext::new();
938        ctx.start_evaluating("cte_a");
939
940        // Simulate trying to evaluate cte_a while it's being evaluated
941        assert!(ctx.is_evaluating("cte_a"));
942    }
943
944    #[test]
945    fn test_cte_stats() {
946        let ctx = CteContext::new();
947        let stats = ctx.stats();
948
949        assert_eq!(stats.ctes_executed, 0);
950        assert_eq!(stats.recursive_iterations, 0);
951        assert_eq!(stats.rows_produced, 0);
952    }
953
954    #[test]
955    fn test_hash_record() {
956        let executor = CteExecutor::new(mock_execute);
957
958        let mut record1 = UnifiedRecord::new();
959        record1.set("id", Value::Integer(1));
960        record1.set("name", Value::text("test".to_string()));
961
962        let mut record2 = UnifiedRecord::new();
963        record2.set("id", Value::Integer(1));
964        record2.set("name", Value::text("test".to_string()));
965
966        let mut record3 = UnifiedRecord::new();
967        record3.set("id", Value::Integer(2));
968        record3.set("name", Value::text("test".to_string()));
969
970        // Same content should have same hash
971        assert_eq!(
972            executor.hash_record(&record1),
973            executor.hash_record(&record2)
974        );
975
976        // Different content should have different hash
977        assert_ne!(
978            executor.hash_record(&record1),
979            executor.hash_record(&record3)
980        );
981    }
982
983    #[test]
984    fn test_hash_various_value_types() {
985        let executor = CteExecutor::new(mock_execute);
986
987        // Test hashing different value types
988        let mut record = UnifiedRecord::new();
989        record.set("null_val", Value::Null);
990        record.set("bool_val", Value::Boolean(true));
991        record.set("int_val", Value::Integer(42));
992        record.set("float_val", Value::Float(2.5));
993        record.set("text_val", Value::text("hello".to_string()));
994        record.set("blob_val", Value::Blob(vec![1, 2, 3]));
995        record.set("timestamp_val", Value::Timestamp(1234567890));
996        record.set("duration_val", Value::Duration(5000));
997
998        // Should not panic
999        let hash = executor.hash_record(&record);
1000        assert!(hash > 0);
1001    }
1002
1003    #[test]
1004    fn test_project_columns() {
1005        let executor = CteExecutor::new(mock_execute);
1006
1007        let mut original =
1008            UnifiedResult::with_columns(vec!["a".to_string(), "b".to_string(), "c".to_string()]);
1009
1010        let mut record = UnifiedRecord::new();
1011        record.set("a", Value::Integer(1));
1012        record.set("b", Value::Integer(2));
1013        record.set("c", Value::Integer(3));
1014        original.push(record);
1015
1016        // Project to different column names
1017        let projected = executor.project_columns(&original, &["x".to_string(), "y".to_string()]);
1018
1019        assert_eq!(projected.columns, vec!["x", "y"]);
1020        assert_eq!(projected.len(), 1);
1021    }
1022
1023    #[test]
1024    fn test_empty_columns_projection() {
1025        let executor = CteExecutor::new(mock_execute);
1026
1027        let original = UnifiedResult::with_columns(vec!["a".to_string()]);
1028
1029        // Empty columns should return original
1030        let projected = executor.project_columns(&original, &[]);
1031        assert_eq!(projected.columns, original.columns);
1032    }
1033
1034    #[test]
1035    fn test_cte_with_multiple_definitions() {
1036        let executor = CteExecutor::new(|query, ctx| {
1037            // Return different results based on which CTE is being queried
1038            match query {
1039                QueryExpr::Table(t) if t.table == "base" => {
1040                    let mut result = UnifiedResult::with_columns(vec!["id".to_string()]);
1041                    let mut record = UnifiedRecord::new();
1042                    record.set("id", Value::Integer(1));
1043                    result.push(record);
1044                    Ok(result)
1045                }
1046                QueryExpr::Table(t) if t.table == "cte1" => {
1047                    // Should be able to see cte1 in context
1048                    if ctx.get("cte1").is_some() {
1049                        Ok(ctx.get("cte1").unwrap().clone())
1050                    } else {
1051                        Ok(UnifiedResult::empty())
1052                    }
1053                }
1054                _ => Ok(UnifiedResult::empty()),
1055            }
1056        });
1057
1058        let cte1 = CteDefinition {
1059            name: "cte1".to_string(),
1060            columns: vec!["id".to_string()],
1061            query: Box::new(QueryExpr::table("base").build()),
1062            recursive: false,
1063        };
1064
1065        let cte2 = CteDefinition {
1066            name: "cte2".to_string(),
1067            columns: vec!["id".to_string()],
1068            query: Box::new(QueryExpr::table("cte1").build()),
1069            recursive: false,
1070        };
1071
1072        let with_clause = WithClause::new().add(cte1).add(cte2);
1073        let query = QueryWithCte::with_ctes(with_clause, QueryExpr::table("cte2").build());
1074
1075        let result = executor.execute(&query);
1076        assert!(result.is_ok());
1077    }
1078}
reddb_server/storage/query/executors/cte.rs

reddb_server/storage/query/executors/
cte.rs
