rupg_analyze/ddl/
lock.rs

1//! `LOCK TABLE`, as `LockTableCommand` in `lockcmds.c` runs it: the checks of each relation of the statement and of the relations that a view reads, and the lock of each of these relations and of their children. The caller takes the locks. Only `LOCK TABLE` takes a lock on a relation in rupg, so a lock of `LOCK TABLE` waits only for the locks of `LOCK TABLE` in other sessions. The other statements take their locks in milestone 5.
2
3use rupg_catalog::RelKind;
4use rupg_common::{Error, Result, SqlState};
5use rupg_pgcatalog::builtin::{self, Named};
6use rupg_sql::nodes::{LockStmt, Node, RangeVar};
7
8use super::relkind_detail;
9use super::view::parse_bool;
10use crate::expr::{Expr, ExprKind};
11use crate::{Analyzer, Env, Params, Query};
12
13/// The privilege check of `LOCK TABLE`. `allowed(relid, role)` is `LockTableAclCheck` for the mode of the statement: true when the role can lock the relation.
14pub type Allowed<'a> = dyn FnMut(u32, u32) -> Result<bool> + 'a;
15
16/// The lock of `LOCK TABLE` on one relation. `take(relid)` is `LockRelationOid` in the mode of the statement, or `ConditionalLockRelationOid` with `NOWAIT`. It gives false when the lock is not free and the statement has `NOWAIT`.
17pub type Take<'a> = dyn FnMut(u32) -> Result<bool> + 'a;
18
19/// `LockTableCommand` for the role `user`. The caller does `RequireTransactionBlock` first. The statement locks each relation after its checks, then the relations that the query of a view reads, or the children of a table without `ONLY`.
20///
21/// # Errors
22///
23/// `55P03` for a lock that is not free with `NOWAIT`, the errors of `take`, the errors of `RangeVarGetRelid` for a name, `42809` for a relation that is not a table or a view, and `42501` for a relation that the role cannot lock. For a view, the owner of the view, or the current user when the view has `security_invoker`, must be able to lock each table and view that the query of the view reads.
24pub fn lock_table(
25    stmt: &LockStmt,
26    env: &dyn Env,
27    user: u32,
28    allowed: &mut Allowed<'_>,
29    take: &mut Take<'_>,
30) -> Result<()> {
31    let params = Params::default();
32    let an = Analyzer::new(env, &params);
33    let mut locker = Locker { an, allowed, take, ancestors: Vec::new() };
34    for node in stmt.relations.iter().flatten() {
35        let Node::RangeVar(rv) = node else { continue };
36        let relid = lookup(&locker.an, rv)?;
37        let name = rv.relname.as_deref().unwrap_or_default();
38        let kind = relkind(env, relid);
39        if !matches!(kind, b'r' | b'p' | b'v') {
40            return Err(Error::new(
41                SqlState::WRONG_OBJECT_TYPE,
42                format!("cannot lock relation \"{name}\""),
43            )
44            .with_detail(relkind_detail(kind)));
45        }
46        if !(locker.allowed)(relid, user)? {
47            return Err(denied(kind, name));
48        }
49        if !(locker.take)(relid)? {
50            let name = match rv.schemaname.as_deref() {
51                Some(schema) => format!("{schema}.{name}"),
52                None => name.to_owned(),
53            };
54            return Err(not_free(&name));
55        }
56        if kind == b'v' {
57            locker.view(relid, user)?;
58        } else if rv.inh {
59            locker.children(relid)?;
60        }
61    }
62    Ok(())
63}
64
65/// `RangeVarGetRelid` with no `missing_ok`. The errors have no position, as in PostgreSQL.
66pub(super) fn lookup(an: &Analyzer<'_>, rv: &RangeVar) -> Result<u32> {
67    let found = an.lookup_relation(rv).map_err(|error| error.with_position(None))?;
68    if let Some(relid) = found {
69        return Ok(relid);
70    }
71    let name = rv.relname.as_deref().unwrap_or_default();
72    Err(match rv.schemaname.as_deref() {
73        Some(schema) if an.schema(schema).is_none() => {
74            Error::new(SqlState::UNDEFINED_SCHEMA, format!("schema \"{schema}\" does not exist"))
75        }
76        Some(schema) => Error::new(
77            SqlState::UNDEFINED_TABLE,
78            format!("relation \"{schema}.{name}\" does not exist"),
79        ),
80        None => {
81            Error::new(SqlState::UNDEFINED_TABLE, format!("relation \"{name}\" does not exist"))
82        }
83    })
84}
85
86/// The state of one `LOCK TABLE`.
87struct Locker<'a, 'f> {
88    an: Analyzer<'a>,
89    allowed: &'f mut Allowed<'f>,
90    take: &'f mut Take<'f>,
91    /// `ancestor_views`: the views that the walk is in.
92    ancestors: Vec<u32>,
93}
94
95impl Locker<'_, '_> {
96    /// `LockTableRecurse`: the lock of each table that inherits from `relid` and of each partition, at each level. The children need no privileges.
97    fn children(&mut self, relid: u32) -> Result<()> {
98        let children = self.an.env.catalog().map(|c| c.all_inheritors(relid)).unwrap_or_default();
99        for child in children {
100            if !(self.take)(child)? {
101                return Err(not_free(&class_name(self.an.env, child)));
102            }
103        }
104        Ok(())
105    }
106
107    /// `LockViewRecurse` for a view of the catalog. A built-in view has the bootstrap superuser as its owner and no `security_invoker`, so each check in its query passes, and the walk skips it.
108    fn view(&mut self, relid: u32, user: u32) -> Result<()> {
109        let Some(query) = self.an.view_query(relid)? else { return Ok(()) };
110        self.view_query(relid, &query, user)
111    }
112
113    /// The walk of the query of the view `relid`, with the checks as the owner of the view or as `user`.
114    fn view_query(&mut self, relid: u32, query: &Query, user: u32) -> Result<()> {
115        let rel = self.an.env.catalog().and_then(|c| c.relation(relid));
116        let invoker = rel.is_some_and(|r| {
117            r.options
118                .iter()
119                .filter_map(|o| o.strip_prefix("security_invoker="))
120                .next_back()
121                .and_then(parse_bool)
122                .unwrap_or(false)
123        });
124        let check_as = match rel {
125            Some(rel) if !invoker => rel.owner,
126            _ => user,
127        };
128        self.ancestors.push(relid);
129        let result = self.walk(query, check_as, user);
130        self.ancestors.pop();
131        result
132    }
133
134    /// `LockViewRecurse_walker` for a query: the tables and the views of the query, then the subqueries in the expressions, in `WITH` and in `FROM`.
135    fn walk(&mut self, query: &Query, check_as: u32, user: u32) -> Result<()> {
136        for relation in &query.relations {
137            if relation.oid == 0 || relation.cte.is_some() {
138                continue;
139            }
140            let kind = relkind(self.an.env, relation.oid);
141            if !matches!(kind, b'r' | b'p' | b'v') || self.ancestors.contains(&relation.oid) {
142                continue;
143            }
144            if !(self.allowed)(relation.oid, check_as)? {
145                return Err(denied(kind, &class_name(self.an.env, relation.oid)));
146            }
147            if !(self.take)(relation.oid)? {
148                return Err(not_free(&class_name(self.an.env, relation.oid)));
149            }
150            if kind == b'v' {
151                if let Some(sub) = relation.subquery.as_deref() {
152                    self.view_query(relation.oid, sub, user)?;
153                }
154            } else if !relation.only {
155                self.children(relation.oid)?;
156            }
157        }
158        for expr in query.exprs() {
159            self.walk_expr(expr, check_as, user)?;
160        }
161        for cte in &query.ctes {
162            self.walk(&cte.query, check_as, user)?;
163        }
164        for relation in &query.relations {
165            if relation.oid == 0
166                && relation.cte.is_none()
167                && let Some(sub) = relation.subquery.as_deref()
168            {
169                self.walk(sub, check_as, user)?;
170            }
171        }
172        Ok(())
173    }
174
175    /// The subqueries of `SubLink` in an expression.
176    fn walk_expr(&mut self, expr: &Expr, check_as: u32, user: u32) -> Result<()> {
177        if let ExprKind::SubLink(sub) = &expr.kind {
178            self.walk(&sub.query, check_as, user)?;
179        }
180        for child in expr.children() {
181            self.walk_expr(child, check_as, user)?;
182        }
183        Ok(())
184    }
185}
186
187/// The `relkind` of a relation of the catalog or of the built-in catalog.
188pub(crate) fn relkind(env: &dyn Env, relid: u32) -> u8 {
189    match env.catalog().and_then(|c| c.relation(relid)) {
190        Some(rel) => match rel.kind {
191            RelKind::Table => b'r',
192            RelKind::Index => b'i',
193            RelKind::Sequence => b'S',
194            RelKind::Toast => b't',
195            RelKind::View => b'v',
196            RelKind::Composite => b'c',
197            RelKind::Partitioned => b'p',
198            RelKind::PartitionedIndex => b'I',
199            RelKind::MatView => b'm',
200        },
201        None => builtin::class_by_oid(relid).map_or(0, |row| row.kind),
202    }
203}
204
205/// The name of a relation of the catalog or of the built-in catalog.
206pub(super) fn class_name(env: &dyn Env, relid: u32) -> String {
207    match env.catalog().and_then(|c| c.relation(relid)) {
208        Some(rel) => rel.name.clone(),
209        None => builtin::named_by_oid(Named::Class, relid)
210            .map_or_else(String::new, |n| n.name.to_string()),
211    }
212}
213
214/// `aclcheck_error` with `ACLCHECK_NO_PRIV` for a table or a view.
215fn denied(kind: u8, name: &str) -> Error {
216    let object = if kind == b'v' { "view" } else { "table" };
217    Error::new(SqlState::INSUFFICIENT_PRIVILEGE, format!("permission denied for {object} {name}"))
218}
219
220/// The error of a lock that is not free with `NOWAIT`.
221fn not_free(name: &str) -> Error {
222    Error::new(
223        SqlState::LOCK_NOT_AVAILABLE,
224        format!("could not obtain lock on relation \"{name}\""),
225    )
226}
rupg_analyze/ddl/lock.rs

rupg_analyze/ddl/
lock.rs
