rupg_analyze/ddl/
rule.rs

1//! `DefineRule`: `CREATE RULE` and `DROP RULE`.
2//!
3//! The analyzer reads the condition and the actions of a rule as `transformRuleStmt` does, with the relation of the rule as `old` and as `new`, and runs the checks of `DefineQueryRewrite`. The catalog keeps the text of the statement and the schemas of `search_path`, and [`rule_tree`] reads the rule again for `pg_get_ruledef`. [`crate::query_rewrite`] applies the rules of a write.
4
5use rupg_catalog::{NewRule, ObjRef, PG_CLASS, PG_REWRITE};
6use rupg_common::{Error, Result, SqlState};
7use rupg_sql::nodes::{CmdType, DropBehavior, DropStmt, Node, RuleStmt};
8
9use super::comment::{name_list, not_owner, range_var};
10use super::drop::owner_word;
11use super::view::{expr_references, query_references};
12use super::{Definer, Found, not_yet, relkind_detail, relname};
13use crate::agg::Kind;
14use crate::expr::{Expr, ExprKind};
15use crate::select::Query;
16use crate::{Analyzer, Env, Params, types};
17
18/// The name of the `ON SELECT` rule of a view.
19const RETURN: &str = "_RETURN";
20
21/// The query trees of a rule, as `transformRuleStmt` makes them. Each query has the relation of the rule as its first two relations, `old` and `new`.
22#[derive(Clone, Debug, Default)]
23pub struct RuleTree {
24    /// The range table of the condition: only `old` and `new`. `DO INSTEAD NOTHING` has this query as its action.
25    pub base: Query,
26    /// The condition of `WHERE`, or `None`.
27    pub qual: Option<Expr>,
28    /// The actions in order. `NOTHING` has no actions.
29    pub actions: Vec<Query>,
30}
31
32/// The `ev_type` code of an event.
33fn event_code(event: CmdType) -> u8 {
34    match event {
35        CmdType::CMD_SELECT => b'1',
36        CmdType::CMD_UPDATE => b'2',
37        CmdType::CMD_INSERT => b'3',
38        _ => b'4',
39    }
40}
41
42/// `rangeTableEntry_used` for the relation at the index of the query `depth` levels up: true when a column of the relation is in the query, in its subqueries or in its `WITH` queries.
43fn query_uses(query: &Query, relation: usize, depth: usize) -> bool {
44    query.all_exprs(depth).into_iter().any(|(e, d)| expr_uses(e, relation, d))
45        || query.ctes.iter().any(|c| query_uses(&c.query, relation, depth + 1))
46}
47
48/// `rangeTableEntry_used` for an expression.
49fn expr_uses(expr: &Expr, relation: usize, depth: usize) -> bool {
50    expr.find(depth, &mut |e, d| match e.kind {
51        ExprKind::Var(var) if var.relation == relation && var.levels_up == d => Some(()),
52        _ => None,
53    })
54    .is_some()
55}
56
57/// An error of `DefineQueryRewrite` with `ERRCODE_INVALID_OBJECT_DEFINITION`.
58fn invalid(message: impl Into<String>) -> Error {
59    Error::new(SqlState::INVALID_OBJECT_DEFINITION, message)
60}
61
62impl Analyzer<'_> {
63    /// `transformRuleStmt` for the relation `oid` with the name `name`.
64    pub(crate) fn rule_tree(&mut self, stmt: &RuleStmt, oid: u32, name: &str) -> Result<RuleTree> {
65        let visible = match stmt.event {
66            CmdType::CMD_UPDATE => [true, true],
67            CmdType::CMD_INSERT => [false, true],
68            _ => [true, false],
69        };
70        self.rule_scope(oid, name, visible, true)?;
71        let qual = match &stmt.whereClause {
72            Some(node) => {
73                let expr = self.with_kind(Kind::Where, |a| a.transform(Some(node)))?;
74                Some(self.coerce_to_boolean(expr, "WHERE")?)
75            }
76            None => None,
77        };
78        let base =
79            Query { relations: std::mem::take(&mut self.scope.relations), ..Query::default() };
80        let mut actions = Vec::with_capacity(stmt.actions.len());
81        for action in stmt.actions.iter().flatten() {
82            let select = match action {
83                Node::SelectStmt(select) => select,
84                Node::InsertStmt(_) | Node::UpdateStmt(_) | Node::DeleteStmt(_) => {
85                    actions.push(self.rule_write(stmt, oid, name, action, qual.as_ref())?);
86                    continue;
87                }
88                Node::NotifyStmt(_) if qual.is_some() => {
89                    return Err(invalid(
90                        "rules with WHERE conditions can only have SELECT, INSERT, UPDATE, or DELETE actions",
91                    ));
92                }
93                Node::NotifyStmt(_) => return Err(not_yet("NOTIFY in a rule", None)),
94                _ => return Err(Error::internal("a rule action that is not a statement")),
95            };
96            self.rule_scope(oid, name, [true, true], false)?;
97            let query = self.select(select)?;
98            if query.set_op.is_some() && qual.is_some() {
99                return Err(Error::new(
100                    SqlState::FEATURE_NOT_SUPPORTED,
101                    "conditional UNION/INTERSECT/EXCEPT statements are not implemented",
102                ));
103            }
104            check_rows(stmt, &query, qual.as_ref(), 0)?;
105            actions.push(query);
106        }
107        Ok(RuleTree { base, qual, actions })
108    }
109
110    /// `transformRuleStmt` for an action `INSERT`, `UPDATE` or `DELETE`. The action is a query below the range table of `old` and `new`, so that the target table of the action is its relation 0, and the action reads `old` and `new` one level up.
111    fn rule_write(
112        &mut self,
113        stmt: &RuleStmt,
114        oid: u32,
115        name: &str,
116        action: &Node,
117        qual: Option<&Expr>,
118    ) -> Result<Query> {
119        let returning = match action {
120            Node::InsertStmt(s) => s.returningClause.is_some(),
121            Node::UpdateStmt(s) => s.returningClause.is_some(),
122            Node::DeleteStmt(s) => s.returningClause.is_some(),
123            _ => false,
124        };
125        if returning {
126            return Err(not_yet("RETURNING in a rule", None));
127        }
128        self.rule_scope(oid, name, [true, true], false)?;
129        let defining = std::mem::replace(&mut self.rule.defining, true);
130        let query = self.sub_query(|analyzer| match action {
131            Node::InsertStmt(s) => analyzer.insert(s),
132            Node::UpdateStmt(s) => analyzer.update(s),
133            Node::DeleteStmt(s) => analyzer.delete(s),
134            _ => Err(Error::internal("a rule action that is not a write")),
135        });
136        self.rule.defining = defining;
137        let query = query?;
138        check_rows(stmt, &query, qual, 1)?;
139        Ok(query)
140    }
141}
142
143/// The checks of `transformRuleStmt` of the names `old` and `new` in an action, whose range table of `old` and `new` is `depth` levels up.
144fn check_rows(stmt: &RuleStmt, query: &Query, qual: Option<&Expr>, depth: usize) -> Result<()> {
145    let in_qual = |relation| qual.is_some_and(|q| expr_uses(q, relation, 0));
146    let has_old = query_uses(query, 0, depth) || in_qual(0);
147    let has_new = query_uses(query, 1, depth) || in_qual(1);
148    match stmt.event {
149        CmdType::CMD_SELECT if has_old => {
150            return Err(invalid("ON SELECT rule cannot use OLD"));
151        }
152        CmdType::CMD_SELECT if has_new => {
153            return Err(invalid("ON SELECT rule cannot use NEW"));
154        }
155        CmdType::CMD_INSERT if has_old => {
156            return Err(invalid("ON INSERT rule cannot use OLD"));
157        }
158        CmdType::CMD_DELETE if has_new => {
159            return Err(invalid("ON DELETE rule cannot use NEW"));
160        }
161        _ => {}
162    }
163    for (relation, word) in [(0, "OLD"), (1, "NEW")] {
164        if query.ctes.iter().any(|c| query_uses(&c.query, relation, depth + 1)) {
165            return Err(Error::new(
166                SqlState::FEATURE_NOT_SUPPORTED,
167                format!("cannot refer to {word} within WITH query"),
168            ));
169        }
170    }
171    Ok(())
172}
173
174/// The query trees of the rule of the catalog with the OID, from the text of the rule and the `search_path` that the rule keeps, or `None` when the catalog has no rule with the OID. `pg_get_ruledef` shows these queries.
175pub fn rule_tree(env: &dyn Env, oid: u32) -> Result<Option<RuleTree>> {
176    let Some(catalog) = env.catalog() else { return Ok(None) };
177    let Some(rule) = catalog.rule(oid) else { return Ok(None) };
178    let name = catalog.relation(rule.relation).map_or("", |r| r.name.as_str());
179    let (list, _) = rupg_sql::parse(&rule.text).map_err(Error::from)?;
180    let stmt = list.iter().flatten().find_map(|node| match node {
181        Node::RawStmt(raw) => match raw.stmt.as_ref() {
182            Some(Node::RuleStmt(stmt)) => Some(stmt),
183            _ => None,
184        },
185        _ => None,
186    });
187    let Some(stmt) = stmt else { return Err(Error::internal("the text of a rule is not a rule")) };
188    let mut an = Analyzer::new(env, &Params::default());
189    an.set_path(&rule.path);
190    an.rule_tree(stmt, rule.relation, name).map(Some).map_err(|error| error.with_position(None))
191}
192
193impl Definer<'_, '_> {
194    /// `DefineRule`: `transformRuleStmt`, then the checks of `DefineQueryRewrite`, then `InsertRule`. `text` is the text of the statement, which the catalog keeps.
195    pub(super) fn create_rule(&mut self, stmt: &RuleStmt, text: &str) -> Result<()> {
196        let rv = stmt.relation.as_deref().cloned().unwrap_or_default();
197        let found = self.lookup_relation(&rv)?;
198        let name = relname(&rv).to_string();
199        let code = self.kind_of(found);
200        let oid = match found {
201            Found::User(oid) => oid,
202            Found::Builtin(row) => row.oid,
203        };
204        if code == b'm' {
205            return Err(Error::new(
206                SqlState::FEATURE_NOT_SUPPORTED,
207                "rules on materialized views are not supported",
208            ));
209        }
210        let tree = self.an.rule_tree(stmt, oid, &name)?;
211        if !matches!(code, b'r' | b'm' | b'v' | b'p') {
212            return Err(Error::new(
213                SqlState::WRONG_OBJECT_TYPE,
214                format!("relation \"{name}\" cannot have rules"),
215            )
216            .with_detail(relkind_detail(code)));
217        }
218        if self.is_system(found) && !self.an.env.allow_system_table_mods() {
219            return Err(Error::new(
220                SqlState::INSUFFICIENT_PRIVILEGE,
221                format!("permission denied: \"{name}\" is a system catalog"),
222            ));
223        }
224        let owner = match found {
225            Found::User(oid) => self.catalog.relation(oid).map_or(0, |r| r.owner),
226            Found::Builtin(row) => row.owner,
227        };
228        if !self.is_superuser() && owner != self.user {
229            return Err(not_owner(owner_word(code), &name));
230        }
231        let rulename = stmt.rulename.as_deref().unwrap_or_default();
232        if stmt.event == CmdType::CMD_SELECT {
233            self.check_select_rule(stmt, &tree, code, oid, &name)?;
234        } else if rulename == RETURN {
235            return Err(invalid(format!(
236                "non-view rule for \"{name}\" must not be named \"{RETURN}\""
237            )));
238        }
239        let Found::User(oid) = found else {
240            return Err(not_yet("a rule on a system catalog", None));
241        };
242        let mut actions = Vec::new();
243        if tree.actions.is_empty() {
244            query_references(self.catalog, &tree.base, &[], &mut actions);
245        }
246        for action in &tree.actions {
247            match &action.write {
248                // An action that writes is below the range table of `old` and `new`.
249                Some(write) => {
250                    actions.push(ObjRef::new(PG_CLASS, write.table));
251                    query_references(self.catalog, action, &[&tree.base], &mut actions);
252                }
253                None => query_references(self.catalog, action, &[], &mut actions),
254            }
255        }
256        let mut qual = Vec::new();
257        if let Some(expr) = &tree.qual {
258            let context = tree.actions.first().filter(|a| a.write.is_none()).unwrap_or(&tree.base);
259            expr_references(self.catalog, expr, &[context], &mut qual);
260        }
261        let new = NewRule {
262            name: rulename.to_string(),
263            relation: oid,
264            event: event_code(stmt.event),
265            instead: stmt.instead,
266            qualified: tree.qual.is_some(),
267            text: text.to_string(),
268            path: self.an.path.clone(),
269        };
270        self.catalog.create_rule(new, stmt.replace, &actions, &qual)?;
271        Ok(())
272    }
273
274    /// The checks of `DefineQueryRewrite` for an `ON SELECT` rule. Only a view can have one, and it has its `_RETURN` rule, so a new rule is an error. rupg does not replace the `_RETURN` rule yet.
275    fn check_select_rule(
276        &self,
277        stmt: &RuleStmt,
278        tree: &RuleTree,
279        code: u8,
280        oid: u32,
281        name: &str,
282    ) -> Result<()> {
283        if !matches!(code, b'v' | b'm') {
284            return Err(Error::new(
285                SqlState::WRONG_OBJECT_TYPE,
286                format!("relation \"{name}\" cannot have ON SELECT rules"),
287            )
288            .with_detail(relkind_detail(code)));
289        }
290        if tree.actions.len() > 1 {
291            return Err(Error::new(
292                SqlState::FEATURE_NOT_SUPPORTED,
293                "multiple actions for rules on SELECT are not implemented",
294            ));
295        }
296        let Some(query) = tree.actions.first().filter(|_| stmt.instead) else {
297            return Err(Error::new(
298                SqlState::FEATURE_NOT_SUPPORTED,
299                "rules on SELECT must have action INSTEAD SELECT",
300            ));
301        };
302        if tree.qual.is_some() {
303            return Err(Error::new(
304                SqlState::FEATURE_NOT_SUPPORTED,
305                "event qualifications are not implemented for rules on SELECT",
306            ));
307        }
308        let columns = self.catalog.relation(oid).map(|r| r.columns.clone()).unwrap_or_default();
309        let mut count = 0;
310        for target in query.targets.iter().filter(|t| !t.junk) {
311            count += 1;
312            let Some(column) = columns.get(count - 1) else {
313                return Err(invalid("SELECT rule's target list has too many entries"));
314            };
315            let (expr, attname) = (&target.expr, &column.name);
316            if target.name != *attname {
317                return Err(invalid(format!(
318                    "SELECT rule's target entry {count} has different column name from column \"{attname}\""
319                ))
320                .with_detail(format!("SELECT target entry is named \"{}\".", target.name)));
321            }
322            if expr.ty != column.ty {
323                return Err(invalid(format!(
324                    "SELECT rule's target entry {count} has different type from column \"{attname}\""
325                ))
326                .with_detail(format!(
327                    "SELECT target entry has type {}, but column has type {}.",
328                    types::name(expr.ty),
329                    types::name(column.ty)
330                )));
331            }
332            if expr.typmod != column.typmod && expr.typmod != -1 && column.typmod != -1 {
333                return Err(invalid(format!(
334                    "SELECT rule's target entry {count} has different size from column \"{attname}\""
335                ))
336                .with_detail(format!(
337                    "SELECT target entry has type {}, but column has type {}.",
338                    types::with_typmod(expr.ty, expr.typmod),
339                    types::with_typmod(column.ty, column.typmod)
340                )));
341            }
342        }
343        if count != columns.len() {
344            return Err(invalid("SELECT rule's target list has too few entries"));
345        }
346        if !stmt.replace {
347            return Err(Error::new(
348                SqlState::OBJECT_NOT_IN_PREREQUISITE_STATE,
349                format!("\"{name}\" is already a view"),
350            ));
351        }
352        if stmt.rulename.as_deref() != Some(RETURN) {
353            return Err(invalid(format!("view rule for \"{name}\" must be named \"{RETURN}\"")));
354        }
355        Err(not_yet("CREATE OR REPLACE RULE of the _RETURN rule of a view", None))
356    }
357
358    /// `RemoveObjects` for `DROP RULE`: the rule of the relation with the name, with the check of `check_object_ownership`.
359    pub(super) fn drop_rule(&mut self, stmt: &DropStmt) -> Result<()> {
360        let mut objects = Vec::new();
361        for item in stmt.objects.iter().flatten() {
362            let names = name_list(Some(item));
363            let Some((&rulename, parent)) = names.split_last() else { continue };
364            let rv = range_var(parent)?;
365            let found = match self.lookup_relation(&rv) {
366                Ok(found) => found,
367                Err(error)
368                    if stmt.missing_ok
369                        && matches!(
370                            error.state(),
371                            SqlState::UNDEFINED_TABLE | SqlState::UNDEFINED_SCHEMA
372                        ) =>
373                {
374                    let message = error.message().to_string();
375                    self.notice(SqlState::SUCCESSFUL_COMPLETION, format!("{message}, skipping"));
376                    continue;
377                }
378                Err(error) => return Err(error),
379            };
380            let Found::User(relation) = found else {
381                return Err(not_yet("DROP RULE of a system catalog", None));
382            };
383            let rel = self.catalog.relation(relation);
384            let relation_name = rel.map_or(String::new(), |r| r.name.clone());
385            let owner = rel.map_or(0, |r| r.owner);
386            let rule = match rel.and_then(|r| r.view.as_ref()) {
387                Some(view) if rulename == RETURN => Some(view.rule),
388                _ => self.catalog.rule_by_name(relation, rulename).map(|r| r.oid),
389            };
390            let Some(rule) = rule else {
391                if stmt.missing_ok {
392                    self.notice(
393                        SqlState::SUCCESSFUL_COMPLETION,
394                        format!(
395                            "rule \"{rulename}\" for relation \"{}\" does not exist, skipping",
396                            parent.join(".")
397                        ),
398                    );
399                    continue;
400                }
401                return Err(Error::new(
402                    SqlState::UNDEFINED_OBJECT,
403                    format!("rule \"{rulename}\" for relation \"{relation_name}\" does not exist"),
404                ));
405            };
406            if !self.is_superuser() && owner != self.user {
407                return Err(not_owner("relation", &relation_name));
408            }
409            objects.push(ObjRef::new(PG_REWRITE, rule));
410        }
411        self.perform(&objects, stmt.behavior == DropBehavior::DROP_CASCADE)
412    }
413}
rupg_analyze/ddl/rule.rs

rupg_analyze/ddl/
rule.rs
