nodedb_lite/query/query_ops/
recursive_value.rs

1// SPDX-License-Identifier: Apache-2.0
2//! RecursiveValue — iterative expression evaluation for value-generating CTEs.
3//!
4//! Implements `WITH RECURSIVE cte(cols) AS (VALUES(init) UNION [ALL] SELECT step FROM cte
5//! WHERE condition)` where no underlying collection is scanned. Column values are
6//! produced purely by expression evaluation on the previous row.
7
8use std::collections::{HashMap, HashSet};
9
10use nodedb_query::expr_parse::parser::parse_generated_expr;
11use nodedb_query::value_ops::is_truthy;
12use nodedb_types::result::QueryResult;
13use nodedb_types::value::Value;
14
15use crate::error::LiteError;
16use crate::query::query_ops::joins::common::maps_to_result;
17
18/// Iterative expression evaluation for value-generating recursive CTEs.
19///
20/// Evaluates `init_exprs` once to produce the anchor row, then iterates:
21/// bind the previous row as column variables, evaluate `condition` (stop when
22/// false), evaluate `step_exprs` to get the next row.  Stops at `max_depth`
23/// iterations (returns a typed error when exceeded) or when `condition` is
24/// false.  Deduplicates if `distinct` is true.
25pub async fn execute_recursive_value(
26    cte_name: &str,
27    columns: &[String],
28    init_exprs: &[String],
29    step_exprs: &[String],
30    condition: Option<&str>,
31    max_depth: usize,
32    distinct: bool,
33) -> Result<QueryResult, LiteError> {
34    if columns.len() != init_exprs.len() || columns.len() != step_exprs.len() {
35        return Err(LiteError::Storage {
36            detail: format!(
37                "recursive CTE '{cte_name}': columns/init_exprs/step_exprs length mismatch \
38                 (columns={}, init={}, step={})",
39                columns.len(),
40                init_exprs.len(),
41                step_exprs.len(),
42            ),
43        });
44    }
45
46    // Parse anchor expressions.
47    let init_parsed: Vec<_> = init_exprs
48        .iter()
49        .map(|e| parse_expr_text(cte_name, e))
50        .collect::<Result<_, _>>()?;
51
52    // Parse step expressions.
53    let step_parsed: Vec<_> = step_exprs
54        .iter()
55        .map(|e| parse_expr_text(cte_name, e))
56        .collect::<Result<_, _>>()?;
57
58    // Parse condition if present.
59    let cond_parsed = condition
60        .map(|c| parse_expr_text(cte_name, c))
61        .transpose()?;
62
63    let effective_max = if max_depth == 0 { 100 } else { max_depth };
64
65    let mut accumulator: Vec<HashMap<String, Value>> = Vec::new();
66    let mut seen: HashSet<String> = HashSet::new();
67
68    // Evaluate anchor row against an empty document.
69    let empty_doc = Value::Object(HashMap::new());
70    let anchor_row = eval_exprs(&init_parsed, &empty_doc, columns)?;
71    maybe_push(&mut accumulator, &mut seen, anchor_row.clone(), distinct);
72
73    let mut current_row = anchor_row;
74
75    for iter in 0..usize::MAX {
76        if iter >= effective_max {
77            return Err(LiteError::Storage {
78                detail: format!("recursive CTE '{cte_name}' exceeded max_depth={effective_max}"),
79            });
80        }
81
82        let doc = Value::Object(current_row.clone());
83
84        // Evaluate condition; stop when false (or absent after first anchor).
85        if let Some(cond) = &cond_parsed {
86            let result = cond.eval(&doc)?;
87            if !is_truthy(&result) {
88                break;
89            }
90        } else {
91            // No condition: run exactly one step (anchor only) — but that was
92            // already done, so we stop if we've already added at least one row.
93            if !accumulator.is_empty() {
94                break;
95            }
96        }
97
98        let next_row = eval_exprs(&step_parsed, &doc, columns)?;
99        // When distinct, a duplicate step row means fixed point — stop.
100        if distinct {
101            let key = row_dedup_key(&next_row);
102            if seen.contains(&key) {
103                break;
104            }
105        }
106        maybe_push(&mut accumulator, &mut seen, next_row.clone(), distinct);
107        current_row = next_row;
108    }
109
110    Ok(maps_to_result(accumulator))
111}
112
113// ── Helpers ─────────────────────────────────────────────────────────────────
114
115fn parse_expr_text(
116    cte_name: &str,
117    text: &str,
118) -> Result<nodedb_query::expr::types::SqlExpr, LiteError> {
119    parse_generated_expr(text)
120        .map(|(expr, _)| expr)
121        .map_err(|e| LiteError::Storage {
122            detail: format!("recursive CTE '{cte_name}': failed to parse expr '{text}': {e}"),
123        })
124}
125
126fn eval_exprs(
127    exprs: &[nodedb_query::expr::types::SqlExpr],
128    doc: &Value,
129    columns: &[String],
130) -> Result<HashMap<String, Value>, LiteError> {
131    columns
132        .iter()
133        .zip(exprs.iter())
134        .map(|(col, expr)| Ok((col.clone(), expr.eval(doc)?)))
135        .collect()
136}
137
138fn maybe_push(
139    acc: &mut Vec<HashMap<String, Value>>,
140    seen: &mut HashSet<String>,
141    row: HashMap<String, Value>,
142    distinct: bool,
143) {
144    if distinct {
145        let key = row_dedup_key(&row);
146        if seen.contains(&key) {
147            return;
148        }
149        seen.insert(key);
150    }
151    acc.push(row);
152}
153
154fn row_dedup_key(row: &HashMap<String, Value>) -> String {
155    let mut pairs: Vec<(&String, &Value)> = row.iter().collect();
156    pairs.sort_by_key(|(k, _)| k.as_str());
157    pairs
158        .into_iter()
159        .map(|(k, v)| format!("{k}={v:?}"))
160        .collect::<Vec<_>>()
161        .join(";")
162}
163
164#[cfg(test)]
165mod tests {
166    use super::*;
167
168    /// Counter CTE: VALUES(1) UNION ALL SELECT n+1 FROM cte WHERE n < 10.
169    #[tokio::test]
170    async fn test_recursive_value_counter_1_to_10() {
171        let columns = vec!["n".to_string()];
172        let init = vec!["1".to_string()];
173        let step = vec!["n + 1".to_string()];
174        let condition = Some("n < 10");
175
176        let result =
177            execute_recursive_value("counter", &columns, &init, &step, condition, 50, false)
178                .await
179                .unwrap();
180
181        // Rows: n = 1, 2, ..., 10.
182        assert_eq!(result.rows.len(), 10, "expected 10 rows (1 through 10)");
183        let n_col = result.columns.iter().position(|c| c == "n").unwrap();
184        let first = &result.rows[0][n_col];
185        let last = &result.rows[9][n_col];
186        assert_eq!(*first, Value::Integer(1));
187        assert_eq!(*last, Value::Integer(10));
188    }
189
190    /// Distinct deduplication: all step values are the same constant, so only
191    /// the anchor should appear.
192    #[tokio::test]
193    async fn test_recursive_value_distinct_deduplication() {
194        // init = 5, step = 5, condition = n < 10 — step never changes n.
195        // Without distinct: would loop to max_depth producing [5, 5, 5, ...].
196        // With distinct: stops after anchor because the next row is a duplicate.
197        let columns = vec!["n".to_string()];
198        let init = vec!["5".to_string()];
199        let step = vec!["5".to_string()]; // constant
200        let condition = Some("n < 10");
201
202        let result =
203            execute_recursive_value("dup_cte", &columns, &init, &step, condition, 20, true)
204                .await
205                .unwrap();
206
207        // Only the anchor row should appear (dedup eliminates identical step rows).
208        assert_eq!(result.rows.len(), 1);
209    }
210}
nodedb_lite/query/query_ops/recursive_value.rs

nodedb_lite/query/query_ops/
recursive_value.rs
