Skip to main content

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::ast::{
31    CteDefinition, Expr, Filter, Projection, QueryExpr, QueryWithCte, SelectItem, TableSource,
32};
33use super::super::unified::{ExecutionError, UnifiedRecord, UnifiedResult};
34use crate::storage::schema::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::Boolean(b) => {
319                1u8.hash(hasher);
320                b.hash(hasher);
321            }
322            Value::Integer(i) => {
323                2u8.hash(hasher);
324                i.hash(hasher);
325            }
326            Value::UnsignedInteger(u) => {
327                3u8.hash(hasher);
328                u.hash(hasher);
329            }
330            Value::Float(f) => {
331                4u8.hash(hasher);
332                f.to_bits().hash(hasher);
333            }
334            Value::Text(s) => {
335                5u8.hash(hasher);
336                s.hash(hasher);
337            }
338            Value::Blob(b) => {
339                6u8.hash(hasher);
340                b.hash(hasher);
341            }
342            Value::Timestamp(t) => {
343                7u8.hash(hasher);
344                t.hash(hasher);
345            }
346            Value::Duration(d) => {
347                8u8.hash(hasher);
348                d.hash(hasher);
349            }
350            Value::IpAddr(addr) => {
351                9u8.hash(hasher);
352                match addr {
353                    std::net::IpAddr::V4(v4) => v4.octets().hash(hasher),
354                    std::net::IpAddr::V6(v6) => v6.octets().hash(hasher),
355                }
356            }
357            Value::MacAddr(mac) => {
358                10u8.hash(hasher);
359                mac.hash(hasher);
360            }
361            Value::Vector(v) => {
362                11u8.hash(hasher);
363                v.len().hash(hasher);
364                for f in v {
365                    f.to_bits().hash(hasher);
366                }
367            }
368            Value::Json(j) => {
369                12u8.hash(hasher);
370                j.hash(hasher);
371            }
372            Value::Uuid(u) => {
373                13u8.hash(hasher);
374                u.hash(hasher);
375            }
376            Value::NodeRef(n) => {
377                14u8.hash(hasher);
378                n.hash(hasher);
379            }
380            Value::EdgeRef(e) => {
381                15u8.hash(hasher);
382                e.hash(hasher);
383            }
384            Value::VectorRef(coll, id) => {
385                16u8.hash(hasher);
386                coll.hash(hasher);
387                id.hash(hasher);
388            }
389            Value::RowRef(table, id) => {
390                17u8.hash(hasher);
391                table.hash(hasher);
392                id.hash(hasher);
393            }
394            Value::Color(rgb) => {
395                18u8.hash(hasher);
396                rgb.hash(hasher);
397            }
398            Value::Email(s) => {
399                19u8.hash(hasher);
400                s.hash(hasher);
401            }
402            Value::Url(s) => {
403                20u8.hash(hasher);
404                s.hash(hasher);
405            }
406            Value::Phone(n) => {
407                21u8.hash(hasher);
408                n.hash(hasher);
409            }
410            Value::Semver(v) => {
411                22u8.hash(hasher);
412                v.hash(hasher);
413            }
414            Value::Cidr(ip, prefix) => {
415                23u8.hash(hasher);
416                ip.hash(hasher);
417                prefix.hash(hasher);
418            }
419            Value::Date(d) => {
420                24u8.hash(hasher);
421                d.hash(hasher);
422            }
423            Value::Time(t) => {
424                25u8.hash(hasher);
425                t.hash(hasher);
426            }
427            Value::Decimal(v) => {
428                26u8.hash(hasher);
429                v.hash(hasher);
430            }
431            Value::DecimalText(v) => {
432                52u8.hash(hasher);
433                v.hash(hasher);
434            }
435            Value::EnumValue(i) => {
436                27u8.hash(hasher);
437                i.hash(hasher);
438            }
439            Value::Array(elems) => {
440                28u8.hash(hasher);
441                elems.len().hash(hasher);
442                for elem in elems {
443                    Self::hash_value(elem, hasher);
444                }
445            }
446            Value::TimestampMs(v) => {
447                29u8.hash(hasher);
448                v.hash(hasher);
449            }
450            Value::Ipv4(v) => {
451                30u8.hash(hasher);
452                v.hash(hasher);
453            }
454            Value::Ipv6(bytes) => {
455                31u8.hash(hasher);
456                bytes.hash(hasher);
457            }
458            Value::Subnet(ip, mask) => {
459                32u8.hash(hasher);
460                ip.hash(hasher);
461                mask.hash(hasher);
462            }
463            Value::Port(v) => {
464                33u8.hash(hasher);
465                v.hash(hasher);
466            }
467            Value::Latitude(v) => {
468                34u8.hash(hasher);
469                v.hash(hasher);
470            }
471            Value::Longitude(v) => {
472                35u8.hash(hasher);
473                v.hash(hasher);
474            }
475            Value::GeoPoint(lat, lon) => {
476                36u8.hash(hasher);
477                lat.hash(hasher);
478                lon.hash(hasher);
479            }
480            Value::Country2(c) => {
481                37u8.hash(hasher);
482                c.hash(hasher);
483            }
484            Value::Country3(c) => {
485                38u8.hash(hasher);
486                c.hash(hasher);
487            }
488            Value::Lang2(c) => {
489                39u8.hash(hasher);
490                c.hash(hasher);
491            }
492            Value::Lang5(c) => {
493                40u8.hash(hasher);
494                c.hash(hasher);
495            }
496            Value::Currency(c) => {
497                41u8.hash(hasher);
498                c.hash(hasher);
499            }
500            Value::AssetCode(code) => {
501                50u8.hash(hasher);
502                code.hash(hasher);
503            }
504            Value::Money {
505                asset_code,
506                minor_units,
507                scale,
508            } => {
509                51u8.hash(hasher);
510                asset_code.hash(hasher);
511                minor_units.hash(hasher);
512                scale.hash(hasher);
513            }
514            Value::ColorAlpha(rgba) => {
515                42u8.hash(hasher);
516                rgba.hash(hasher);
517            }
518            Value::BigInt(v) => {
519                43u8.hash(hasher);
520                v.hash(hasher);
521            }
522            Value::KeyRef(col, key) => {
523                44u8.hash(hasher);
524                col.hash(hasher);
525                key.hash(hasher);
526            }
527            Value::DocRef(col, id) => {
528                45u8.hash(hasher);
529                col.hash(hasher);
530                id.hash(hasher);
531            }
532            Value::TableRef(name) => {
533                46u8.hash(hasher);
534                name.hash(hasher);
535            }
536            Value::PageRef(page_id) => {
537                47u8.hash(hasher);
538                page_id.hash(hasher);
539            }
540            Value::Secret(bytes) => {
541                48u8.hash(hasher);
542                bytes.hash(hasher);
543            }
544            Value::Password(hash) => {
545                49u8.hash(hasher);
546                hash.hash(hasher);
547            }
548        }
549    }
550}
551
552/// Helper to parse UNION structure for recursive CTEs
553pub fn split_union_parts(query: &QueryExpr) -> Option<(QueryExpr, QueryExpr)> {
554    // UNION support is not represented in the current AST; recursive queries execute
555    // the full body expression each iteration.
556    let _ = query;
557    None
558}
559
560// ─────────────────────────────────────────────────────────────────────
561// CTE inlining (#41) — non-recursive
562//
563// Rewrites a `QueryWithCte` into a plain `QueryExpr` by walking the
564// AST and substituting every `TableSource::Name(name)` (or legacy
565// `TableQuery.table` field) that matches a CTE name with
566// `TableSource::Subquery(cte.query)`. After this pass the runtime's
567// existing subquery-in-FROM machinery executes the result with no
568// CTE-specific dispatch needed.
569//
570// Recursive CTEs are rejected up-front — the iterative fixpoint
571// strategy is implemented in `CteExecutor` but is not wired into the
572// runtime yet (separate slice).
573// ─────────────────────────────────────────────────────────────────────
574
575/// Inline a `QueryWithCte`'s WITH clause into its inner query. Returns
576/// the rewritten `QueryExpr` ready for dispatch. Recursive CTEs are
577/// rejected with a clear error.
578pub fn inline_ctes(query: QueryWithCte) -> Result<QueryExpr, ExecutionError> {
579    let Some(with_clause) = query.with_clause else {
580        return Ok(query.query);
581    };
582    if with_clause.has_recursive {
583        return Err(ExecutionError::new(
584            "WITH RECURSIVE is not yet supported by the executor; \
585             non-recursive WITH clauses run today, recursive support \
586             is tracked separately"
587                .to_string(),
588        ));
589    }
590
591    // Inline each CTE into its successors first so chained CTEs
592    // (`WITH a AS (...), b AS (... a ...)`) end up with fully resolved
593    // bodies before they're substituted into the outer query.
594    let mut resolved: HashMap<String, QueryExpr> = HashMap::new();
595    for cte in &with_clause.ctes {
596        let mut body = (*cte.query).clone();
597        rewrite(&mut body, &resolved);
598        resolved.insert(cte.name.clone(), body);
599    }
600
601    let mut outer = query.query;
602    rewrite(&mut outer, &resolved);
603    Ok(outer)
604}
605
606/// Walk a `QueryExpr` and replace any table reference whose name
607/// matches a key in `ctes` with the inlined CTE body. Recurses
608/// through `Join` and nested `Subquery` sources so CTE refs inside
609/// JOINs and subqueries resolve too. Mirrors the view-rewrite
610/// convention: when the outer table reference carries filter / limit
611/// / offset constraints we wrap the body in a `Subquery` to preserve
612/// them; otherwise we replace the whole `Table` node verbatim with
613/// the CTE body so dispatchers that key off `QueryExpr::Table` (like
614/// the JOIN executor) see the right shape.
615fn rewrite(expr: &mut QueryExpr, ctes: &HashMap<String, QueryExpr>) {
616    match expr {
617        QueryExpr::Table(tq) => {
618            let lookup_name = match &tq.source {
619                Some(TableSource::Subquery(_)) => None,
620                Some(TableSource::Name(n)) => Some(n.clone()),
621                // Table-valued functions are not CTE references (issue #795);
622                // the inline-graph form likewise references no CTE (issue #799).
623                Some(TableSource::Function { .. } | TableSource::InlineGraphFunction { .. }) => {
624                    None
625                }
626                None => Some(tq.table.clone()),
627            };
628
629            if let Some(name) = lookup_name {
630                if let Some(body) = ctes.get(&name) {
631                    let outer_has_constraints = tq.filter.is_some()
632                        || tq.where_expr.is_some()
633                        || tq.limit.is_some()
634                        || tq.offset.is_some()
635                        || !tq.columns.is_empty()
636                        || !tq.select_items.is_empty()
637                        || !tq.group_by.is_empty()
638                        || !tq.order_by.is_empty();
639
640                    if outer_has_constraints {
641                        // Outer ref carries projections / filters /
642                        // limits — keep those by wrapping the body in
643                        // a subquery source. Sentinel name so legacy
644                        // `table` consumers can't resolve it against
645                        // the real schema.
646                        tq.source = Some(TableSource::Subquery(Box::new(body.clone())));
647                        tq.table = format!("__cte_{name}");
648                    } else {
649                        // Bare `FROM cte` (possibly with alias) —
650                        // replace verbatim so JOIN / dispatch paths
651                        // see the CTE body's natural shape.
652                        *expr = body.clone();
653                    }
654                    return;
655                }
656            }
657
658            match tq.source.as_mut() {
659                Some(TableSource::Subquery(body)) => rewrite(body, ctes),
660                Some(TableSource::InlineGraphFunction { nodes, edges, .. }) => {
661                    rewrite(nodes, ctes);
662                    rewrite(edges, ctes);
663                }
664                _ => {}
665            }
666            rewrite_table_query_parts(tq, ctes);
667        }
668        QueryExpr::Join(jq) => {
669            rewrite(&mut jq.left, ctes);
670            rewrite(&mut jq.right, ctes);
671            if let Some(filter) = jq.filter.as_mut() {
672                rewrite_filter(filter, ctes);
673            }
674            for item in &mut jq.order_by {
675                if let Some(expr) = item.expr.as_mut() {
676                    rewrite_expr(expr, ctes);
677                }
678            }
679            for item in &mut jq.return_items {
680                rewrite_select_item(item, ctes);
681            }
682            for projection in &mut jq.return_ {
683                rewrite_projection(projection, ctes);
684            }
685        }
686        QueryExpr::Graph(gq) => {
687            if let Some(filter) = gq.filter.as_mut() {
688                rewrite_filter(filter, ctes);
689            }
690            for projection in &mut gq.return_ {
691                rewrite_projection(projection, ctes);
692            }
693        }
694        _ => {}
695    }
696}
697
698fn rewrite_table_query_parts(
699    tq: &mut super::super::ast::TableQuery,
700    ctes: &HashMap<String, QueryExpr>,
701) {
702    for item in &mut tq.select_items {
703        rewrite_select_item(item, ctes);
704    }
705    for projection in &mut tq.columns {
706        rewrite_projection(projection, ctes);
707    }
708    if let Some(expr) = tq.where_expr.as_mut() {
709        rewrite_expr(expr, ctes);
710    }
711    if let Some(filter) = tq.filter.as_mut() {
712        rewrite_filter(filter, ctes);
713    }
714    for expr in &mut tq.group_by_exprs {
715        rewrite_expr(expr, ctes);
716    }
717    if let Some(expr) = tq.having_expr.as_mut() {
718        rewrite_expr(expr, ctes);
719    }
720    if let Some(filter) = tq.having.as_mut() {
721        rewrite_filter(filter, ctes);
722    }
723    for item in &mut tq.order_by {
724        if let Some(expr) = item.expr.as_mut() {
725            rewrite_expr(expr, ctes);
726        }
727    }
728}
729
730fn rewrite_select_item(item: &mut SelectItem, ctes: &HashMap<String, QueryExpr>) {
731    if let SelectItem::Expr { expr, .. } = item {
732        rewrite_expr(expr, ctes);
733    }
734}
735
736fn rewrite_projection(projection: &mut Projection, ctes: &HashMap<String, QueryExpr>) {
737    match projection {
738        Projection::Function(_, args) => {
739            for arg in args {
740                rewrite_projection(arg, ctes);
741            }
742        }
743        Projection::Expression(filter, _) => rewrite_filter(filter, ctes),
744        Projection::Window { args, window, .. } => {
745            for arg in args {
746                rewrite_projection(arg, ctes);
747            }
748            for expr in &mut window.partition_by {
749                rewrite_expr(expr, ctes);
750            }
751            for item in &mut window.order_by {
752                rewrite_expr(&mut item.expr, ctes);
753            }
754        }
755        Projection::All
756        | Projection::Column(_)
757        | Projection::Alias(_, _)
758        | Projection::Field(_, _) => {}
759    }
760}
761
762fn rewrite_filter(filter: &mut Filter, ctes: &HashMap<String, QueryExpr>) {
763    match filter {
764        Filter::CompareExpr { lhs, rhs, .. } => {
765            rewrite_expr(lhs, ctes);
766            rewrite_expr(rhs, ctes);
767        }
768        Filter::And(left, right) | Filter::Or(left, right) => {
769            rewrite_filter(left, ctes);
770            rewrite_filter(right, ctes);
771        }
772        Filter::Not(inner) => rewrite_filter(inner, ctes),
773        Filter::Compare { .. }
774        | Filter::CompareFields { .. }
775        | Filter::IsNull(_)
776        | Filter::IsNotNull(_)
777        | Filter::In { .. }
778        | Filter::Between { .. }
779        | Filter::Like { .. }
780        | Filter::StartsWith { .. }
781        | Filter::EndsWith { .. }
782        | Filter::Contains { .. } => {}
783    }
784}
785
786fn rewrite_expr(expr: &mut Expr, ctes: &HashMap<String, QueryExpr>) {
787    match expr {
788        Expr::BinaryOp { lhs, rhs, .. } => {
789            rewrite_expr(lhs, ctes);
790            rewrite_expr(rhs, ctes);
791        }
792        Expr::UnaryOp { operand, .. } => rewrite_expr(operand, ctes),
793        Expr::Cast { inner, .. } => rewrite_expr(inner, ctes),
794        Expr::FunctionCall { args, .. } => {
795            for arg in args {
796                rewrite_expr(arg, ctes);
797            }
798        }
799        Expr::Case {
800            branches, else_, ..
801        } => {
802            for (condition, value) in branches {
803                rewrite_expr(condition, ctes);
804                rewrite_expr(value, ctes);
805            }
806            if let Some(value) = else_ {
807                rewrite_expr(value, ctes);
808            }
809        }
810        Expr::IsNull { operand, .. } => rewrite_expr(operand, ctes),
811        Expr::InList { target, values, .. } => {
812            rewrite_expr(target, ctes);
813            for value in values {
814                rewrite_expr(value, ctes);
815            }
816        }
817        Expr::Between {
818            target, low, high, ..
819        } => {
820            rewrite_expr(target, ctes);
821            rewrite_expr(low, ctes);
822            rewrite_expr(high, ctes);
823        }
824        Expr::Subquery { query, .. } => rewrite(&mut query.query, ctes),
825        Expr::WindowFunctionCall { args, window, .. } => {
826            for arg in args {
827                rewrite_expr(arg, ctes);
828            }
829            for expr in &mut window.partition_by {
830                rewrite_expr(expr, ctes);
831            }
832            for item in &mut window.order_by {
833                rewrite_expr(&mut item.expr, ctes);
834            }
835        }
836        Expr::Literal { .. } | Expr::Column { .. } | Expr::Parameter { .. } => {}
837    }
838}
839
840// ============================================================================
841// Tests
842// ============================================================================
843
844#[cfg(test)]
845mod tests {
846    use super::*;
847    use crate::storage::query::ast::CteQueryBuilder;
848    use crate::storage::query::WithClause;
849
850    fn mock_execute(
851        _query: &QueryExpr,
852        _ctx: &CteContext,
853    ) -> Result<UnifiedResult, ExecutionError> {
854        // Simple mock that returns empty result
855        Ok(UnifiedResult::empty())
856    }
857
858    #[test]
859    fn test_cte_context() {
860        let mut ctx = CteContext::new();
861
862        // Test empty context
863        assert!(ctx.get("test").is_none());
864        assert!(!ctx.is_evaluating("test"));
865
866        // Test storing results
867        let result = UnifiedResult::with_columns(vec!["col1".to_string()]);
868        ctx.store("test".to_string(), result);
869        assert!(ctx.get("test").is_some());
870
871        // Test evaluation tracking
872        ctx.start_evaluating("other");
873        assert!(ctx.is_evaluating("other"));
874        ctx.done_evaluating("other");
875        assert!(!ctx.is_evaluating("other"));
876    }
877
878    #[test]
879    fn test_simple_cte_execution() {
880        let executor = CteExecutor::new(|_query, _ctx| {
881            let mut result = UnifiedResult::with_columns(vec!["id".to_string()]);
882            let mut record = UnifiedRecord::new();
883            record.set("id", Value::Integer(1));
884            result.push(record);
885            Ok(result)
886        });
887
888        // Create a simple CTE query
889        let cte = CteDefinition {
890            name: "test_cte".to_string(),
891            columns: vec!["id".to_string()],
892            query: Box::new(QueryExpr::table("dummy").build()),
893            recursive: false,
894        };
895
896        let with_clause = WithClause::new().add(cte);
897        let query = QueryWithCte::with_ctes(with_clause, QueryExpr::table("test_cte").build());
898
899        let result = executor.execute(&query);
900        assert!(result.is_ok());
901    }
902
903    #[test]
904    fn test_cte_builder() {
905        let query = CteQueryBuilder::new()
906            .cte_with_columns(
907                "nums",
908                vec!["n".to_string()],
909                QueryExpr::table("numbers").build(),
910            )
911            .build(QueryExpr::table("nums").build());
912
913        assert!(query.with_clause.is_some());
914        let with_clause = query.with_clause.unwrap();
915        assert_eq!(with_clause.ctes.len(), 1);
916        assert_eq!(with_clause.ctes[0].name, "nums");
917    }
918
919    #[test]
920    fn test_recursive_cte_builder() {
921        let query = CteQueryBuilder::new()
922            .recursive_cte("paths", QueryExpr::table("connections").build())
923            .build(QueryExpr::table("paths").build());
924
925        assert!(query.with_clause.is_some());
926        let with_clause = query.with_clause.unwrap();
927        assert!(with_clause.has_recursive);
928        assert!(with_clause.ctes[0].recursive);
929    }
930
931    #[test]
932    fn test_circular_reference_detection() {
933        let mut ctx = CteContext::new();
934        ctx.start_evaluating("cte_a");
935
936        // Simulate trying to evaluate cte_a while it's being evaluated
937        assert!(ctx.is_evaluating("cte_a"));
938    }
939
940    #[test]
941    fn test_cte_stats() {
942        let ctx = CteContext::new();
943        let stats = ctx.stats();
944
945        assert_eq!(stats.ctes_executed, 0);
946        assert_eq!(stats.recursive_iterations, 0);
947        assert_eq!(stats.rows_produced, 0);
948    }
949
950    #[test]
951    fn test_hash_record() {
952        let executor = CteExecutor::new(mock_execute);
953
954        let mut record1 = UnifiedRecord::new();
955        record1.set("id", Value::Integer(1));
956        record1.set("name", Value::text("test".to_string()));
957
958        let mut record2 = UnifiedRecord::new();
959        record2.set("id", Value::Integer(1));
960        record2.set("name", Value::text("test".to_string()));
961
962        let mut record3 = UnifiedRecord::new();
963        record3.set("id", Value::Integer(2));
964        record3.set("name", Value::text("test".to_string()));
965
966        // Same content should have same hash
967        assert_eq!(
968            executor.hash_record(&record1),
969            executor.hash_record(&record2)
970        );
971
972        // Different content should have different hash
973        assert_ne!(
974            executor.hash_record(&record1),
975            executor.hash_record(&record3)
976        );
977    }
978
979    #[test]
980    fn test_hash_various_value_types() {
981        let executor = CteExecutor::new(mock_execute);
982
983        // Test hashing different value types
984        let mut record = UnifiedRecord::new();
985        record.set("null_val", Value::Null);
986        record.set("bool_val", Value::Boolean(true));
987        record.set("int_val", Value::Integer(42));
988        record.set("float_val", Value::Float(2.5));
989        record.set("text_val", Value::text("hello".to_string()));
990        record.set("blob_val", Value::Blob(vec![1, 2, 3]));
991        record.set("timestamp_val", Value::Timestamp(1234567890));
992        record.set("duration_val", Value::Duration(5000));
993
994        // Should not panic
995        let hash = executor.hash_record(&record);
996        assert!(hash > 0);
997    }
998
999    #[test]
1000    fn test_project_columns() {
1001        let executor = CteExecutor::new(mock_execute);
1002
1003        let mut original =
1004            UnifiedResult::with_columns(vec!["a".to_string(), "b".to_string(), "c".to_string()]);
1005
1006        let mut record = UnifiedRecord::new();
1007        record.set("a", Value::Integer(1));
1008        record.set("b", Value::Integer(2));
1009        record.set("c", Value::Integer(3));
1010        original.push(record);
1011
1012        // Project to different column names
1013        let projected = executor.project_columns(&original, &["x".to_string(), "y".to_string()]);
1014
1015        assert_eq!(projected.columns, vec!["x", "y"]);
1016        assert_eq!(projected.len(), 1);
1017    }
1018
1019    #[test]
1020    fn test_empty_columns_projection() {
1021        let executor = CteExecutor::new(mock_execute);
1022
1023        let original = UnifiedResult::with_columns(vec!["a".to_string()]);
1024
1025        // Empty columns should return original
1026        let projected = executor.project_columns(&original, &[]);
1027        assert_eq!(projected.columns, original.columns);
1028    }
1029
1030    #[test]
1031    fn test_cte_with_multiple_definitions() {
1032        let executor = CteExecutor::new(|query, ctx| {
1033            // Return different results based on which CTE is being queried
1034            match query {
1035                QueryExpr::Table(t) if t.table == "base" => {
1036                    let mut result = UnifiedResult::with_columns(vec!["id".to_string()]);
1037                    let mut record = UnifiedRecord::new();
1038                    record.set("id", Value::Integer(1));
1039                    result.push(record);
1040                    Ok(result)
1041                }
1042                QueryExpr::Table(t) if t.table == "cte1" => {
1043                    // Should be able to see cte1 in context
1044                    if ctx.get("cte1").is_some() {
1045                        Ok(ctx.get("cte1").unwrap().clone())
1046                    } else {
1047                        Ok(UnifiedResult::empty())
1048                    }
1049                }
1050                _ => Ok(UnifiedResult::empty()),
1051            }
1052        });
1053
1054        let cte1 = CteDefinition {
1055            name: "cte1".to_string(),
1056            columns: vec!["id".to_string()],
1057            query: Box::new(QueryExpr::table("base").build()),
1058            recursive: false,
1059        };
1060
1061        let cte2 = CteDefinition {
1062            name: "cte2".to_string(),
1063            columns: vec!["id".to_string()],
1064            query: Box::new(QueryExpr::table("cte1").build()),
1065            recursive: false,
1066        };
1067
1068        let with_clause = WithClause::new().add(cte1).add(cte2);
1069        let query = QueryWithCte::with_ctes(with_clause, QueryExpr::table("cte2").build());
1070
1071        let result = executor.execute(&query);
1072        assert!(result.is_ok());
1073    }
1074}