rupg_analyze/
merge.rs

1//! `transformMergeStmt` of `parse_merge.c`, with `transformMergeSupportFunc` of `parse_expr.c` and the join of `transform_MERGE_to_join` of `prepjointree.c`.
2//!
3//! The query of a `MERGE` joins the source and the target table, which is the relation 0. Each row of the query gives the columns of the target table, the OID of the leaf partition of the row when the target table is a partitioned table, the ID of the version of the row of the target table, a marker that is null when the row has no source row, and then the carried values: the columns of the source that the `WHEN` clauses and `RETURNING` read. The ID of the version is null when the row has no target row.
4//!
5//! Each `WHEN` clause has a small query that reads two relations: the row of the target table with the ID of its version, and the carried values. It gives the new row of an `INSERT` or an `UPDATE`, then the conditions of the check constraints, or one `true` value for a `DELETE` and `DO NOTHING`. It gives no row when its condition is not true.
6
7use rupg_common::{Error, Result, SqlState};
8use rupg_sql::nodes::{
9    CmdType, List, MergeMatchKind, MergeStmt, MergeSupportFunc, MergeWhenClause, Node, RangeVar,
10};
11use rupg_types::{Value, oid};
12
13use crate::Analyzer;
14use crate::agg::Kind;
15use crate::coerce::AtOpt;
16use crate::expr::{Expr, ExprKind, Var};
17use crate::from::{Column, FromItem, Join, JoinKind, Relation};
18use crate::insert::{Foreign, Write, WriteKind, targets, with_checks};
19use crate::modify::{carried_targets, old_row, set_columns, table_oid_target, tid_target};
20use crate::perm::{ACL_DELETE, ACL_INSERT, ACL_SELECT, ACL_UPDATE};
21use crate::returning::rewrite;
22use crate::select::{Query, Target};
23use crate::typename::place;
24
25/// `MergeMatchKind`: the rows that a `WHEN` clause takes.
26#[derive(Clone, Copy, Debug, PartialEq, Eq)]
27pub enum MatchKind {
28    /// `WHEN MATCHED`: a row of the target table with a source row.
29    Matched,
30    /// `WHEN NOT MATCHED BY SOURCE`: a row of the target table with no source row.
31    NotMatchedBySource,
32    /// `WHEN NOT MATCHED [BY TARGET]`: a source row with no row of the target table.
33    NotMatchedByTarget,
34}
35
36/// The action of a `WHEN` clause.
37#[derive(Clone, Copy, Debug, PartialEq, Eq)]
38pub enum MergeCommand {
39    Insert,
40    Update,
41    Delete,
42    /// `DO NOTHING`.
43    Nothing,
44}
45
46/// `MergeAction`: a `WHEN` clause of a `MERGE`.
47#[derive(Clone, Debug, PartialEq)]
48pub struct MergeClause {
49    pub kind: MatchKind,
50    pub command: MergeCommand,
51    /// The query of the clause, which reads the row of the target table with the ID of its version and the carried values. It gives the new row of an `INSERT` or an `UPDATE` with the conditions of the check constraints after the columns, or one `true` value for a `DELETE` and `DO NOTHING`. It gives no row when the condition of the clause is not true.
52    pub query: Box<Query>,
53}
54
55/// The clauses of a `MERGE` and the foreign keys that its actions check.
56#[derive(Clone, Debug, PartialEq)]
57pub struct Merge {
58    /// The `WHEN` clauses in the order of the statement.
59    pub clauses: Vec<MergeClause>,
60    /// The number of the carried values at the end of each row of the query of the statement.
61    pub carried: usize,
62    /// The foreign keys that an inserted row checks.
63    pub insert_foreign: Vec<Foreign>,
64    /// The foreign keys that an updated row checks.
65    pub update_foreign: Vec<Foreign>,
66    /// The foreign keys that a deleted row checks.
67    pub delete_foreign: Vec<Foreign>,
68}
69
70/// A `WHEN` clause after the analysis of its parts, before the analyzer knows all the carried values.
71struct Parts {
72    kind: MatchKind,
73    command: MergeCommand,
74    condition: Option<Expr>,
75    row: Vec<Expr>,
76}
77
78/// The action of a `WHEN` clause.
79fn command(kind: CmdType) -> MergeCommand {
80    match kind {
81        CmdType::CMD_INSERT => MergeCommand::Insert,
82        CmdType::CMD_UPDATE => MergeCommand::Update,
83        CmdType::CMD_DELETE => MergeCommand::Delete,
84        _ => MergeCommand::Nothing,
85    }
86}
87
88/// The kind of the rows of a `WHEN` clause.
89fn match_kind(clause: &MergeWhenClause) -> MatchKind {
90    match clause.matchKind {
91        MergeMatchKind::MERGE_WHEN_MATCHED => MatchKind::Matched,
92        MergeMatchKind::MERGE_WHEN_NOT_MATCHED_BY_SOURCE => MatchKind::NotMatchedBySource,
93        _ => MatchKind::NotMatchedByTarget,
94    }
95}
96
97/// The place of a carried value that reads the column `attnum` of the relation `relation`, which the list gets when it does not have it.
98pub(crate) fn carry(carried: &mut Vec<Expr>, relation: usize, attnum: i16, expr: &Expr) -> usize {
99    let found = carried.iter().position(
100        |c| matches!(c.kind, ExprKind::Var(v) if v.relation == relation && v.attnum == attnum),
101    );
102    found.unwrap_or_else(|| {
103        let mut value = Expr::new(ExprKind::Var(Var { relation, attnum, levels_up: 0 }), expr.ty);
104        value.typmod = expr.typmod;
105        carried.push(value);
106        carried.len() - 1
107    })
108}
109
110impl Analyzer<'_> {
111    /// `transformMergeStmt`: the query tree of a `MERGE`.
112    pub(crate) fn merge(&mut self, s: &MergeStmt) -> Result<Query> {
113        self.write_with(s.withClause.as_deref())?;
114        let mut clauses = Vec::with_capacity(s.mergeWhenClauses.len());
115        for node in &s.mergeWhenClauses {
116            let Some(Node::MergeWhenClause(clause)) = node else {
117                return Err(Error::internal("a WHEN clause that is not MergeWhenClause"));
118            };
119            clauses.push(&**clause);
120        }
121        // A clause after a clause of the same kind with no condition cannot take a row.
122        let mut unconditional: Vec<MatchKind> = Vec::new();
123        for clause in &clauses {
124            let kind = match_kind(clause);
125            if unconditional.contains(&kind) {
126                return Err(Error::new(
127                    SqlState::SYNTAX_ERROR,
128                    "unreachable WHEN clause specified after unconditional WHEN clause",
129                ));
130            }
131            if clause.condition.is_none() {
132                unconditional.push(kind);
133            }
134        }
135        let rv = s.relation.as_deref().ok_or_else(|| Error::internal("a MERGE with no table"))?;
136        self.merge_actions = clauses.iter().map(|c| command(c.commandType)).collect();
137        let (table, name, columns) = self.target_table(rv, WriteKind::Merge)?;
138        // `DO NOTHING` needs `SELECT`, so that the target table needs some privilege.
139        let required = clauses.iter().fold(0, |bits, clause| {
140            bits | match clause.commandType {
141                CmdType::CMD_INSERT => ACL_INSERT,
142                CmdType::CMD_UPDATE => ACL_UPDATE,
143                CmdType::CMD_DELETE => ACL_DELETE,
144                _ => ACL_SELECT,
145            }
146        });
147        if let Some(perm) = self.target_perm_mut() {
148            perm.required |= required;
149        }
150        let source =
151            s.sourceRelation.as_ref().ok_or_else(|| Error::internal("a MERGE with no source"))?;
152        let source = self.merge_from(rv, source)?;
153        let on = self.with_kind(Kind::JoinOn, |a| a.transform(s.joinCondition.as_ref()))?;
154        let on = self.coerce_to_boolean(on, "JOIN/ON")?;
155        let (returning, mut carried) = match s.returningClause.as_deref() {
156            None => (None, Vec::new()),
157            Some(clause) => {
158                let (query, carried) = self.returning(clause, WriteKind::Merge)?;
159                (Some(Box::new(query)), carried)
160            }
161        };
162        let mut parts = Vec::with_capacity(clauses.len());
163        let mut updated = Vec::new();
164        for clause in &clauses {
165            let kind = match_kind(clause);
166            let update = matches!(clause.commandType, CmdType::CMD_UPDATE);
167            // The items of `SET` of an updatable view name the columns of its base relation.
168            let set =
169                if update { self.auto_view_set(rv, &clause.targetList, &columns) } else { None };
170            let (renamed, error) = match set {
171                Some((list, unknown, fixed)) => (Some(list), unknown.or(fixed)),
172                None => (None, None),
173            };
174            let list = renamed.as_ref().unwrap_or(&clause.targetList);
175            if update {
176                for column in set_columns(list, &columns) {
177                    if !updated.contains(&column) {
178                        updated.push(column);
179                    }
180                }
181            }
182            let full = self.merge_namespace(
183                kind != MatchKind::NotMatchedByTarget,
184                kind != MatchKind::NotMatchedBySource,
185            );
186            let part = match error {
187                Some(error) => self.set_items(list).and(Err(error)),
188                None => self.merge_clause(clause, list, kind, rv, (table, &name, &columns)),
189            };
190            self.set_namespace(full);
191            let mut part = part?;
192            // A column of the source becomes a carried value, which is the relation 1 of the query of the clause.
193            let mut change = |expr: &mut Expr, depth: usize| {
194                let ExprKind::Var(var) = expr.kind else { return };
195                if var.levels_up != depth || var.relation == 0 {
196                    return;
197                }
198                let at = carry(&mut carried, var.relation, var.attnum, expr);
199                let attnum = i16::try_from(at + 1).unwrap_or(i16::MAX);
200                expr.kind = ExprKind::Var(Var { relation: 1, attnum, levels_up: depth });
201            };
202            for expr in part.row.iter_mut().chain(part.condition.as_mut()) {
203                rewrite(expr, 0, &mut change);
204            }
205            parts.push(part);
206        }
207        // The leaf partition of each row of a partitioned table checks the row after the routing, and so does the table of each row of a table with children of `INHERITS`.
208        let inherit = self.inherits_target(WriteKind::Merge);
209        let partitioned = self.is_partitioned(table) || inherit;
210        let checks = if partitioned { Vec::new() } else { self.checks(table)? };
211        let names = checks.iter().map(|(n, _)| n.clone()).collect();
212        let generated = if partitioned { Vec::new() } else { self.generated_columns(table)? };
213        let carried_columns: Vec<Column> = carried
214            .iter()
215            .map(|expr| Column {
216                name: String::new(),
217                ty: expr.ty,
218                typmod: expr.typmod,
219                not_null: false,
220            })
221            .collect();
222        let mut target = self
223            .scope
224            .relations
225            .first()
226            .cloned()
227            .ok_or_else(|| Error::internal("a MERGE with no table"))?;
228        // The query of a clause reads the row of the join, and not the query of a view.
229        target.subquery = None;
230        let mut merge_clauses = Vec::with_capacity(parts.len());
231        for part in parts {
232            let writes = matches!(part.command, MergeCommand::Insert | MergeCommand::Update);
233            let row = if writes {
234                targets(&columns, part.row)
235            } else {
236                let expr = Expr::new(ExprKind::Const(Value::Bool(true)), oid::BOOL);
237                vec![Target { name: String::new(), expr, origin: None, junk: false }]
238            };
239            let mut query = Query {
240                relations: vec![
241                    target.clone(),
242                    Relation::new(0, carried_columns.clone(), None, None, None),
243                ],
244                from: vec![FromItem::Relation(0), FromItem::Relation(1)],
245                targets: row,
246                filter: part.condition,
247                ..Query::default()
248            };
249            if writes && (!checks.is_empty() || !generated.is_empty()) {
250                query = with_checks(&name, &columns, query, checks.clone(), &generated, &[]);
251            }
252            merge_clauses.push(MergeClause {
253                kind: part.kind,
254                command: part.command,
255                query: Box::new(query),
256            });
257        }
258        let has = |kind: MatchKind| merge_clauses.iter().any(|c| c.kind == kind);
259        let (by_source, by_target) =
260            (has(MatchKind::NotMatchedBySource), has(MatchKind::NotMatchedByTarget));
261        // With `NOT MATCHED BY SOURCE`, a marker with each source row tells a row of the target table with no source row.
262        let (source, marker) = if by_source {
263            let index = self.scope.relations.len();
264            let column =
265                Column { name: "marker".to_string(), ty: oid::BOOL, typmod: -1, not_null: false };
266            let value = Expr::new(ExprKind::Const(Value::Bool(true)), oid::BOOL);
267            let mut relation = Relation::new(0, vec![column], None, None, Some(vec![vec![value]]));
268            relation.name = "*VALUES*".to_string();
269            self.scope.relations.push(relation);
270            let var = Var { relation: index, attnum: 1, levels_up: 0 };
271            (join(JoinKind::Inner, source, FromItem::Relation(index), None), ExprKind::Var(var))
272        } else {
273            (source, ExprKind::Const(Value::Bool(true)))
274        };
275        let kind = match (by_source, by_target) {
276            (true, true) => JoinKind::Full,
277            (true, false) => JoinKind::Right,
278            (false, true) => JoinKind::Left,
279            (false, false) => JoinKind::Inner,
280        };
281        // The conditions of an updatable view limit the rows of the target table before the join, as a join with one row does.
282        let target_item = match self.auto_view_quals() {
283            None => FromItem::Relation(0),
284            Some(quals) => {
285                let index = self.scope.relations.len();
286                let column =
287                    Column { name: "one".to_string(), ty: oid::BOOL, typmod: -1, not_null: false };
288                let value = Expr::new(ExprKind::Const(Value::Bool(true)), oid::BOOL);
289                let mut relation =
290                    Relation::new(0, vec![column], None, None, Some(vec![vec![value]]));
291                relation.name = "*VALUES*".to_string();
292                self.scope.relations.push(relation);
293                join(JoinKind::Inner, FromItem::Relation(0), FromItem::Relation(index), Some(quals))
294            }
295        };
296        // `view_has_instead_trigger`: the triggers `INSTEAD OF` of a view do the actions, and the rows of the view have no ID of their versions. A column of a join with one row tells a row of the view from a source row with no match.
297        let instead = self.is_view(table);
298        let mut tid = tid_target();
299        let target_item = if instead {
300            let index = self.scope.relations.len();
301            let column =
302                Column { name: "ctid".to_string(), ty: oid::INT8, typmod: -1, not_null: false };
303            let value = Expr::new(ExprKind::Const(Value::Int8(0)), oid::INT8);
304            let mut relation = Relation::new(0, vec![column], None, None, Some(vec![vec![value]]));
305            relation.name = "*VALUES*".to_string();
306            self.scope.relations.push(relation);
307            let var = Var { relation: index, attnum: 1, levels_up: 0 };
308            tid.expr = Expr::new(ExprKind::Var(var), oid::INT8);
309            join(JoinKind::Inner, target_item, FromItem::Relation(index), None)
310        } else {
311            target_item
312        };
313        let from = join(kind, source, target_item, Some(on));
314        let mut query_targets = old_row(&columns);
315        if partitioned {
316            query_targets.push(table_oid_target());
317        }
318        query_targets.push(tid);
319        query_targets.push(Target {
320            name: String::new(),
321            expr: Expr::new(marker, oid::BOOL),
322            origin: None,
323            junk: false,
324        });
325        let count = carried.len();
326        query_targets.extend(carried_targets(carried));
327        let uses = |command: MergeCommand| merge_clauses.iter().any(|c| c.command == command);
328        if !updated.is_empty() {
329            self.mark_updated(&updated);
330        }
331        let foreign = |a: &Self, command: MergeCommand, kind: WriteKind| {
332            if uses(command) { a.foreign_keys(table, kind, &columns) } else { Ok(Vec::new()) }
333        };
334        let insert_foreign = foreign(self, MergeCommand::Insert, WriteKind::Insert)?;
335        let update_foreign = foreign(self, MergeCommand::Update, WriteKind::Update)?;
336        let delete_foreign = foreign(self, MergeCommand::Delete, WriteKind::Delete)?;
337        let unique = self.unique_keys(table, &columns)?;
338        // The triggers `BEFORE ... FOR EACH ROW` of the inserts and of the updates come before the checks of the constraints.
339        let recheck =
340            [(MergeCommand::Insert, WriteKind::Insert), (MergeCommand::Update, WriteKind::Update)]
341                .into_iter()
342                .filter(|&(command, _)| uses(command))
343                .find_map(|(_, kind)| {
344                    self.recheck(table, kind, &name, &columns, &checks, &generated)
345                });
346        let mut query = Query {
347            relations: std::mem::take(&mut self.scope.relations),
348            from: vec![from],
349            targets: query_targets,
350            write: Some(Write {
351                kind: WriteKind::Merge,
352                table,
353                name,
354                alias: None,
355                width: columns.len(),
356                checks: names,
357                unique,
358                returning,
359                foreign: Vec::new(),
360                on_conflict: None,
361                merge: Some(Merge {
362                    clauses: merge_clauses,
363                    carried: count,
364                    insert_foreign,
365                    update_foreign,
366                    delete_foreign,
367                }),
368                recheck,
369                updated: crate::modify::with_extra_updated(updated, &generated),
370                instead,
371                inherit,
372                view_check: None,
373                views: 0,
374            }),
375            ..Query::default()
376        };
377        self.end_write_with(&mut query, s.withClause.as_deref())?;
378        Ok(query)
379    }
380
381    /// `transformMergeStmt` for one `WHEN` clause: the condition, and the new row of an `INSERT` or an `UPDATE`.
382    fn merge_clause(
383        &mut self,
384        clause: &MergeWhenClause,
385        list: &List,
386        kind: MatchKind,
387        rv: &RangeVar,
388        (table, name, columns): (u32, &str, &[Column]),
389    ) -> Result<Parts> {
390        let condition = match &clause.condition {
391            None => None,
392            Some(node) => {
393                let expr = self.with_kind(Kind::MergeWhen, |a| a.transform(Some(node)))?;
394                Some(self.coerce_to_boolean(expr, "WHEN")?)
395            }
396        };
397        let (command, row) = match clause.commandType {
398            CmdType::CMD_INSERT => {
399                self.swap_auto_view_defaults();
400                let row = self.merge_insert(clause, table, name, columns);
401                self.swap_auto_view_defaults();
402                (MergeCommand::Insert, row?)
403            }
404            CmdType::CMD_UPDATE => {
405                (MergeCommand::Update, self.update_row(rv, table, name, columns, list)?)
406            }
407            CmdType::CMD_DELETE => (MergeCommand::Delete, Vec::new()),
408            _ => (MergeCommand::Nothing, Vec::new()),
409        };
410        Ok(Parts { kind, command, condition, row })
411    }
412
413    /// The new row of an `INSERT` action of a `MERGE`: the expressions of `VALUES` for the columns that the action gives, and the defaults of the other columns.
414    fn merge_insert(
415        &mut self,
416        clause: &MergeWhenClause,
417        table: u32,
418        name: &str,
419        columns: &[Column],
420    ) -> Result<Vec<Expr>> {
421        let given = match self.auto_view_targets(&clause.targetList, columns.len())? {
422            Some(given) => given,
423            None => self.insert_targets(&clause.targetList, name, columns)?,
424        };
425        let exprs = if clause.values.is_empty() {
426            // `INSERT DEFAULT VALUES`.
427            Vec::new()
428        } else {
429            self.mark_inserted(&given);
430            let items = self.value_items(&clause.values, Kind::ValuesSingle)?;
431            let explicit = !clause.targetList.is_empty();
432            self.insert_row(items, table, columns, &given, explicit, clause.r#override)?
433        };
434        self.full_row(table, columns, &given, exprs)
435    }
436
437    /// `transformMergeSupportFunc`: `merge_action()`, which is only in the list of `RETURNING` of a `MERGE` or in a subquery of it. It reads the column of the relation of the action.
438    ///
439    /// # Errors
440    ///
441    /// `42601` in another place.
442    pub(crate) fn merge_support_func(&mut self, f: &MergeSupportFunc) -> Result<Expr> {
443        let count = self.outer_kinds.len();
444        for (level, scope) in self.levels() {
445            let kind = match level {
446                0 => Some(self.kind),
447                _ => count.checked_sub(level).and_then(|i| self.outer_kinds.get(i)?.0),
448            };
449            if kind == Some(Kind::Returning)
450                && let Some(relation) = scope.merge_action
451            {
452                let var = Var { relation, attnum: 1, levels_up: level };
453                return Ok(Expr::new(ExprKind::Var(var), oid::TEXT).at(place(f.location)));
454            }
455        }
456        Err(Error::new(
457            SqlState::SYNTAX_ERROR,
458            "MERGE_ACTION() can only be used in the RETURNING list of a MERGE command",
459        )
460        .at_opt(place(f.location)))
461    }
462}
463
464/// A join with no names of its own.
465fn join(kind: JoinKind, left: FromItem, right: FromItem, on: Option<Expr>) -> FromItem {
466    FromItem::Join(Box::new(Join {
467        kind,
468        left,
469        right,
470        on,
471        using: Vec::new(),
472        plain_using: true,
473        alias: None,
474        using_alias: None,
475        merged: Vec::new(),
476        columns: Vec::new(),
477        aliases: Vec::new(),
478        sources: Vec::new(),
479    }))
480}
rupg_analyze/merge.rs

rupg_analyze/
merge.rs
