rupg_analyze/
searchcycle.rs

1//! The `SEARCH` and `CYCLE` clauses of a recursive query of `WITH`: the checks of `analyzeCTE` and the rewrite of `rewriteSearchCycle.c`. The rewrite changes the two terms of the `UNION` so that each row of the query also gives the column of the order, or the mark and the path of a cycle.
2
3use rupg_common::{Error, Result, SqlState};
4use rupg_pgcatalog::builtin;
5use rupg_sql::nodes::{CTECycleClause, CTESearchClause, List, Node};
6use rupg_types::{Value, oid};
7
8use crate::Analyzer;
9use crate::coerce::AtOpt;
10use crate::expr::{Case, CastForm, Expr, ExprKind, Func, FuncForm, Var};
11use crate::from::{Column, FromItem, Relation};
12use crate::select::{Query, Target};
13use crate::setop::SetTree;
14use crate::sort::set_group;
15use crate::typcache;
16use crate::typename::place;
17use crate::types;
18
19/// `RECORD_EQ_OP`: the `=` operator of two values of `record`.
20const RECORD_EQ_OP: u32 = 2988;
21/// `F_ARRAY_CAT`.
22const ARRAY_CAT: u32 = 383;
23/// `F_INT8INC`.
24const INT8INC: u32 = 1219;
25
26/// `CTESearchClause`: `SEARCH DEPTH FIRST` or `SEARCH BREADTH FIRST`.
27#[derive(Clone, Debug, PartialEq)]
28pub struct SearchClause {
29    /// `search_col_list`: the columns that give the order.
30    pub columns: Vec<String>,
31    /// `search_breadth_first`.
32    pub breadth: bool,
33    /// `search_seq_column`: the name of the column of the order.
34    pub seq: String,
35    location: Option<usize>,
36}
37
38/// `CTECycleClause`: `CYCLE`, with the columns that find a cycle, the column of the mark with its two values, and the column of the path.
39#[derive(Clone, Debug, PartialEq)]
40pub struct CycleClause {
41    /// `cycle_col_list`.
42    pub columns: Vec<String>,
43    /// `cycle_mark_column`.
44    pub mark: String,
45    /// `cycle_mark_value`: the mark of a row that closes a cycle.
46    pub value: Expr,
47    /// `cycle_mark_default`: the mark of the other rows.
48    pub default: Expr,
49    /// `cycle_mark_type`.
50    pub ty: u32,
51    /// `cycle_mark_typmod`.
52    pub typmod: i32,
53    /// `cycle_path_column`.
54    pub path: String,
55    /// `cycle_mark_neop`: the `<>` operator of the type of the mark, and its function.
56    ne: (u32, u32),
57    location: Option<usize>,
58}
59
60/// The strings of a list of `String` nodes.
61fn strings(list: &List) -> Vec<String> {
62    list.iter()
63        .filter_map(|n| match n {
64            Some(Node::String(s)) => Some(s.to_string()),
65            _ => None,
66        })
67        .collect()
68}
69
70/// The clause of the raw tree.
71pub(crate) fn search_clause(raw: &CTESearchClause) -> SearchClause {
72    SearchClause {
73        columns: strings(&raw.search_col_list),
74        breadth: raw.search_breadth_first,
75        seq: raw.search_seq_column.as_deref().unwrap_or_default().to_string(),
76        location: place(raw.location),
77    }
78}
79
80/// The columns that the clauses add after the columns of the query, as `addRangeTableEntryForCTE` adds them to each reference: the column of the order, of type `record` for `BREADTH` or `record[]` for `DEPTH`, then the mark and the path, of type `record[]`.
81pub(crate) fn extra_columns(
82    search: Option<&SearchClause>,
83    cycle: Option<&CycleClause>,
84) -> Vec<Column> {
85    let mut columns = Vec::new();
86    let column =
87        |name: &str, ty, typmod| Column { name: name.to_string(), ty, typmod, not_null: false };
88    if let Some(s) = search {
89        let ty = if s.breadth { oid::RECORD } else { oid::RECORD_ARRAY };
90        columns.push(column(&s.seq, ty, -1));
91    }
92    if let Some(c) = cycle {
93        columns.push(column(&c.mark, c.ty, c.typmod));
94        columns.push(column(&c.path, oid::RECORD_ARRAY, -1));
95    }
96    columns
97}
98
99/// A syntax error at the place of a clause.
100fn syntax(message: String, at: Option<usize>) -> Error {
101    Error::new(SqlState::SYNTAX_ERROR, message).at_opt(at)
102}
103
104/// The checks of the columns of a clause: each column must be a column of the query and must not come twice.
105fn check_columns(what: &str, list: &[String], names: &[&str], at: Option<usize>) -> Result<()> {
106    for (i, name) in list.iter().enumerate() {
107        if !names.contains(&name.as_str()) {
108            return Err(syntax(
109                format!("{what} column \"{name}\" not in WITH query column list"),
110                at,
111            ));
112        }
113        if list[..i].contains(name) {
114            return Err(Error::new(
115                SqlState::DUPLICATE_COLUMN,
116                format!("{what} column \"{name}\" specified more than once"),
117            )
118            .at_opt(at));
119        }
120    }
121    Ok(())
122}
123
124/// The checks of `analyzeCTE` after the analysis of the query: the query must be recursive with a `SELECT` on each side of the `UNION`, and the names of the clauses must be good.
125pub(crate) fn check(
126    columns: &[Column],
127    recursive: bool,
128    location: Option<usize>,
129    search: Option<&SearchClause>,
130    cycle: Option<&CycleClause>,
131    query: &Query,
132) -> Result<()> {
133    if !recursive {
134        return Err(syntax("WITH query is not recursive".to_string(), location));
135    }
136    let Some(op) = &query.set_op else {
137        return Err(Error::internal("a recursive query with no set operation"));
138    };
139    if !matches!(op.left, SetTree::Leaf(_)) {
140        return Err(Error::new(
141            SqlState::FEATURE_NOT_SUPPORTED,
142            "with a SEARCH or CYCLE clause, the left side of the UNION must be a SELECT",
143        ));
144    }
145    if !matches!(op.right, SetTree::Leaf(_)) {
146        return Err(syntax(
147            "with a SEARCH or CYCLE clause, the right side of the UNION must be a SELECT"
148                .to_string(),
149            None,
150        ));
151    }
152    let names: Vec<&str> = columns.iter().map(|c| c.name.as_str()).collect();
153    if let Some(s) = search {
154        check_columns("search", &s.columns, &names, s.location)?;
155        if names.contains(&s.seq.as_str()) {
156            return Err(syntax(
157                format!(
158                    "search sequence column name \"{}\" already used in WITH query column list",
159                    s.seq
160                ),
161                s.location,
162            ));
163        }
164    }
165    if let Some(c) = cycle {
166        check_columns("cycle", &c.columns, &names, c.location)?;
167        if names.contains(&c.mark.as_str()) {
168            return Err(syntax(
169                format!(
170                    "cycle mark column name \"{}\" already used in WITH query column list",
171                    c.mark
172                ),
173                c.location,
174            ));
175        }
176        if names.contains(&c.path.as_str()) {
177            return Err(syntax(
178                format!(
179                    "cycle path column name \"{}\" already used in WITH query column list",
180                    c.path
181                ),
182                c.location,
183            ));
184        }
185        if c.mark == c.path {
186            return Err(syntax(
187                "cycle mark column name and cycle path column name are the same".to_string(),
188                c.location,
189            ));
190        }
191    }
192    if let (Some(s), Some(c)) = (search, cycle) {
193        if s.seq == c.mark {
194            return Err(syntax(
195                "search sequence column name and cycle mark column name are the same".to_string(),
196                s.location,
197            ));
198        }
199        if s.seq == c.path {
200            return Err(syntax(
201                "search sequence column name and cycle path column name are the same".to_string(),
202                s.location,
203            ));
204        }
205    }
206    Ok(())
207}
208
209/// A column of the relation with this index.
210fn var(relation: usize, index: usize, ty: u32, typmod: i32) -> Expr {
211    let attnum = i16::try_from(index + 1).unwrap_or(i16::MAX);
212    let var = Var { relation, attnum, levels_up: 0 };
213    Expr { kind: ExprKind::Var(var), ty, typmod, location: None }
214}
215
216/// A call of the function with this OID.
217fn call(func: u32, args: Vec<Expr>, ty: u32) -> Expr {
218    let func = Func { oid: func, args, form: FuncForm::Call, variadic: false, retset: false };
219    Expr::new(ExprKind::Func(func), ty)
220}
221
222/// `make_path_rowexpr`: `ROW(cols)` of the columns of the list, from the first relation.
223fn row_of(list: &[String], columns: &[Column]) -> Vec<Expr> {
224    list.iter()
225        .filter_map(|name| {
226            let i = columns.iter().position(|c| &c.name == name)?;
227            Some(var(0, i, columns[i].ty, columns[i].typmod))
228        })
229        .collect()
230}
231
232/// `ROW(...)` of type `record`.
233fn row(args: Vec<Expr>) -> Expr {
234    Expr::new(ExprKind::Row(args, CastForm::Implicit), oid::RECORD)
235}
236
237/// `make_path_initial_array`: `ARRAY[ROW(cols)]`.
238fn array(args: Vec<Expr>) -> Expr {
239    let elements = vec![row(args)];
240    Expr::new(
241        ExprKind::Array { element: oid::RECORD, multidims: false, elements },
242        oid::RECORD_ARRAY,
243    )
244}
245
246/// `make_path_cat_expr`: `path || ARRAY[ROW(cols)]`, where `path` is the column with this index.
247fn path_cat(args: Vec<Expr>, index: usize) -> Expr {
248    let path = var(0, index, oid::RECORD_ARRAY, -1);
249    call(ARRAY_CAT, vec![path, array(args)], oid::RECORD_ARRAY)
250}
251
252/// A query that reads one subquery in `FROM`, as `newq1` and `newq2` of `rewriteSearchAndCycle`.
253fn wrap(name: &str, columns: Vec<Column>, sub: Query, targets: Vec<Target>) -> Query {
254    let mut relation = Relation::new(0, columns, Some(Box::new(sub)), None, None);
255    relation.name = name.to_string();
256    Query {
257        relations: vec![relation],
258        from: vec![FromItem::Relation(0)],
259        targets,
260        ..Query::default()
261    }
262}
263
264/// A target that the result shows.
265fn target(name: &str, expr: Expr) -> Target {
266    Target { name: name.to_string(), expr, origin: None, junk: false }
267}
268
269/// `IncrementVarSublevelsUp` by one level for a term that goes into a new query.
270fn raise(query: &mut Query) {
271    query.each_expr_mut(0, &mut |expr, depth| expr.raise_from(1, depth + 1));
272}
273
274/// `rewriteSearchAndCycle`: the query of `WITH` that gives the columns of the clauses. The left term becomes a query that reads the old term and adds the first values of the columns. The right term gives the columns of the reference to the query too, and a query that reads it adds the next values. With `CYCLE`, the right term does not read a row with the mark of a cycle.
275pub(crate) fn rewrite(
276    name: &str,
277    columns: &[Column],
278    work: Option<usize>,
279    search: Option<&SearchClause>,
280    cycle: Option<&CycleClause>,
281    query: &Query,
282) -> Result<Query> {
283    let mut new = query.clone();
284    let Some(op) = new.set_op.as_mut() else {
285        return Err(Error::internal("a recursive query with no set operation"));
286    };
287    let (SetTree::Leaf(left), SetTree::Leaf(right)) = (&op.left, &op.right) else {
288        return Err(Error::internal("a term of SEARCH or CYCLE that is not a SELECT"));
289    };
290    let (left, right) = (*left, *right);
291    let width = columns.len();
292    let extras = extra_columns(search, cycle);
293    let mut all = columns.to_vec();
294    all.extend(extras.iter().cloned());
295    let plain: Vec<Target> = columns
296        .iter()
297        .enumerate()
298        .map(|(i, c)| target(&c.name, var(0, i, c.ty, c.typmod)))
299        .collect();
300
301    // The left term: `SELECT cols, extras FROM (q1) "*TLOCRN*"(cols)`.
302    let Some(q1) = new.relations[left].subquery.as_deref() else {
303        return Err(Error::internal("a SELECT of a set operation with no query"));
304    };
305    let mut q1 = q1.clone();
306    raise(&mut q1);
307    let mut targets = plain.clone();
308    if let Some(s) = search {
309        let args = row_of(&s.columns, columns);
310        let expr = if s.breadth {
311            let zero = Expr::constant(Value::Int8(0), oid::INT8, -1, None);
312            row(std::iter::once(zero).chain(args).collect())
313        } else {
314            array(args)
315        };
316        targets.push(target(&s.seq, expr));
317    }
318    if let Some(c) = cycle {
319        targets.push(target(&c.mark, c.default.clone()));
320        targets.push(target(&c.path, array(row_of(&c.columns, columns))));
321    }
322    let newq1 = wrap("*TLOCRN*", columns.to_vec(), q1, targets);
323
324    // The right term: `SELECT cols, next extras FROM (q2 with the extras of the reference) "*TROCRN*"(cols, extras) WHERE mark <> value`.
325    let Some(q2) = new.relations[right].subquery.as_deref() else {
326        return Err(Error::internal("a SELECT of a set operation with no query"));
327    };
328    let mut q2 = q2.clone();
329    let reference = q2.relations.iter().position(|r| {
330        r.cte.as_ref().is_some_and(|c| c.name == name && c.work.is_some() && c.work == work)
331    });
332    let Some(reference) = reference else {
333        return Err(Error::new(
334            SqlState::FEATURE_NOT_SUPPORTED,
335            format!(
336                "with a SEARCH or CYCLE clause, the recursive reference to WITH query \"{name}\" must be at the top level of its right-hand SELECT"
337            ),
338        ));
339    };
340    if q2.targets.iter().any(|t| t.junk) {
341        return Err(Error::new(
342            SqlState::FEATURE_NOT_SUPPORTED,
343            "with a SEARCH or CYCLE clause, a right-hand SELECT with a junk column is not supported yet",
344        ));
345    }
346    raise(&mut q2);
347    for (k, extra) in extras.iter().enumerate() {
348        q2.targets.push(target(&extra.name, var(reference, width + k, extra.ty, extra.typmod)));
349    }
350    let mut targets = plain;
351    let mut next = width;
352    if let Some(s) = search {
353        let args = row_of(&s.columns, columns);
354        let expr = if s.breadth {
355            let seq = var(0, next, oid::RECORD, -1);
356            let depth = Expr::new(ExprKind::FieldSelect(Box::new(seq), 0), oid::INT8);
357            let depth = call(INT8INC, vec![depth], oid::INT8);
358            row(std::iter::once(depth).chain(args).collect())
359        } else {
360            path_cat(args, next)
361        };
362        targets.push(target(&s.seq, expr));
363        next += 1;
364    }
365    let mut filter = None;
366    if let Some(c) = cycle {
367        let (mark, path) = (next, next + 1);
368        let eq = builtin::operator_by_oid(RECORD_EQ_OP).map_or(0, |o| o.code);
369        let path_var = var(0, path, oid::RECORD_ARRAY, -1);
370        let args = vec![row(row_of(&c.columns, columns)), path_var];
371        let found = Expr::new(ExprKind::ScalarArrayOp { func: eq, any: true, args }, oid::BOOL);
372        let case = Case {
373            arg: None,
374            whens: vec![(found, c.value.clone())],
375            default: Box::new(c.default.clone()),
376        };
377        let case = Expr { typmod: c.typmod, ..Expr::new(ExprKind::Case(case), c.ty) };
378        targets.push(target(&c.mark, case));
379        targets.push(target(&c.path, path_cat(row_of(&c.columns, columns), path)));
380        let (ne, func) = c.ne;
381        let args = vec![var(0, mark, c.ty, c.typmod), c.value.clone()];
382        let ne =
383            Func { oid: func, args, form: FuncForm::Operator(ne), variadic: false, retset: false };
384        filter = Some(Expr::new(ExprKind::Func(ne), oid::BOOL));
385    }
386    let mut newq2 = wrap("*TROCRN*", all, q2, targets);
387    newq2.filter = filter;
388
389    // The set operation and the query of `WITH` give the extras too.
390    for (k, extra) in extras.iter().enumerate() {
391        op.columns.push((extra.ty, extra.typmod));
392        if !op.all {
393            let mut group = set_group(width + k, extra.ty, None)?;
394            if matches!(extra.ty, oid::RECORD | oid::RECORD_ARRAY) {
395                group.hashable = true;
396            }
397            op.groups.push(group);
398        }
399    }
400    for (index, sub) in [(left, newq1), (right, newq2)] {
401        let relation = &mut new.relations[index];
402        relation.subquery = Some(Box::new(sub));
403        relation.columns.extend(extras.iter().cloned());
404    }
405    for (k, extra) in extras.iter().enumerate() {
406        new.targets.push(target(&extra.name, var(left, width + k, extra.ty, extra.typmod)));
407    }
408    Ok(new)
409}
410
411impl Analyzer<'_> {
412    /// The part of `analyzeCTE` before the analysis of the query: the type of the mark from its two values, and the `<>` operator of the type.
413    pub(crate) fn cycle_clause(&mut self, raw: &CTECycleClause) -> Result<CycleClause> {
414        let value = self.transform(raw.cycle_mark_value.as_ref())?;
415        let default = self.transform(raw.cycle_mark_default.as_ref())?;
416        let ty = self.common_type(&[value.clone(), default.clone()], "CYCLE")?;
417        let value = self.coerce_to_common(value, ty, "CYCLE/SET/TO")?;
418        let default = self.coerce_to_common(default, ty, "CYCLE/SET/DEFAULT")?;
419        let typmod = if value.typmod == default.typmod { value.typmod } else { -1 };
420        let eq = typcache::operators(ty).eq;
421        if eq == 0 {
422            return Err(Error::new(
423                SqlState::UNDEFINED_FUNCTION,
424                format!("could not identify an equality operator for type {}", types::name(ty)),
425            ));
426        }
427        let ne = builtin::operator_by_oid(eq).map_or(0, |o| o.negator);
428        let Some(func) = builtin::operator_by_oid(ne).map(|o| o.code).filter(|&f| f != 0) else {
429            return Err(Error::new(
430                SqlState::UNDEFINED_FUNCTION,
431                format!("could not identify an inequality operator for type {}", types::name(ty)),
432            ));
433        };
434        Ok(CycleClause {
435            columns: strings(&raw.cycle_col_list),
436            mark: raw.cycle_mark_column.as_deref().unwrap_or_default().to_string(),
437            value,
438            default,
439            ty,
440            typmod,
441            path: raw.cycle_path_column.as_deref().unwrap_or_default().to_string(),
442            ne: (ne, func),
443            location: place(raw.location),
444        })
445    }
446}
rupg_analyze/searchcycle.rs

rupg_analyze/
searchcycle.rs
