TLA Line data Source code
1 : /*-------------------------------------------------------------------------
2 : *
3 : * lockcmds.c
4 : * LOCK command support code
5 : *
6 : * Portions Copyright (c) 1996-2023, PostgreSQL Global Development Group
7 : * Portions Copyright (c) 1994, Regents of the University of California
8 : *
9 : *
10 : * IDENTIFICATION
11 : * src/backend/commands/lockcmds.c
12 : *
13 : *-------------------------------------------------------------------------
14 : */
15 : #include "postgres.h"
16 :
17 : #include "access/table.h"
18 : #include "access/xact.h"
19 : #include "catalog/namespace.h"
20 : #include "catalog/pg_inherits.h"
21 : #include "commands/lockcmds.h"
22 : #include "commands/tablecmds.h"
23 : #include "miscadmin.h"
24 : #include "nodes/nodeFuncs.h"
25 : #include "parser/parse_clause.h"
26 : #include "rewrite/rewriteHandler.h"
27 : #include "storage/lmgr.h"
28 : #include "utils/acl.h"
29 : #include "utils/lsyscache.h"
30 : #include "utils/syscache.h"
31 :
32 : static void LockTableRecurse(Oid reloid, LOCKMODE lockmode, bool nowait);
33 : static AclResult LockTableAclCheck(Oid reloid, LOCKMODE lockmode, Oid userid);
34 : static void RangeVarCallbackForLockTable(const RangeVar *rv, Oid relid,
35 : Oid oldrelid, void *arg);
36 : static void LockViewRecurse(Oid reloid, LOCKMODE lockmode, bool nowait,
37 : List *ancestor_views);
38 :
39 : /*
40 : * LOCK TABLE
41 : */
42 : void
43 GIC 522 : LockTableCommand(LockStmt *lockstmt)
44 ECB : {
45 : ListCell *p;
46 :
47 : /*
48 : * Iterate over the list and process the named relations one at a time
49 : */
50 GIC 5144 : foreach(p, lockstmt->relations)
51 ECB : {
52 GIC 4660 : RangeVar *rv = (RangeVar *) lfirst(p);
53 CBC 4660 : bool recurse = rv->inh;
54 ECB : Oid reloid;
55 :
56 GIC 4660 : reloid = RangeVarGetRelidExtended(rv, lockstmt->mode,
57 CBC 4660 : lockstmt->nowait ? RVR_NOWAIT : 0,
58 ECB : RangeVarCallbackForLockTable,
59 GIC 4660 : (void *) &lockstmt->mode);
60 ECB :
61 GIC 4625 : if (get_rel_relkind(reloid) == RELKIND_VIEW)
62 CBC 33 : LockViewRecurse(reloid, lockstmt->mode, lockstmt->nowait, NIL);
63 4592 : else if (recurse)
64 4589 : LockTableRecurse(reloid, lockstmt->mode, lockstmt->nowait);
65 ECB : }
66 GIC 484 : }
67 ECB :
68 : /*
69 : * Before acquiring a table lock on the named table, check whether we have
70 : * permission to do so.
71 : */
72 : static void
73 GIC 4696 : RangeVarCallbackForLockTable(const RangeVar *rv, Oid relid, Oid oldrelid,
74 ECB : void *arg)
75 : {
76 GIC 4696 : LOCKMODE lockmode = *(LOCKMODE *) arg;
77 ECB : char relkind;
78 : char relpersistence;
79 : AclResult aclresult;
80 :
81 GIC 4696 : if (!OidIsValid(relid))
82 LBC 0 : return; /* doesn't exist, so no permissions check */
83 GBC 4696 : relkind = get_rel_relkind(relid);
84 CBC 4696 : if (!relkind)
85 LBC 0 : return; /* woops, concurrently dropped; no permissions
86 EUB : * check */
87 :
88 : /* Currently, we only allow plain tables or views to be locked */
89 GIC 4696 : if (relkind != RELKIND_RELATION && relkind != RELKIND_PARTITIONED_TABLE &&
90 ECB : relkind != RELKIND_VIEW)
91 UIC 0 : ereport(ERROR,
92 EUB : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
93 : errmsg("cannot lock relation \"%s\"",
94 : rv->relname),
95 : errdetail_relkind_not_supported(relkind)));
96 :
97 : /*
98 : * Make note if a temporary relation has been accessed in this
99 : * transaction.
100 : */
101 GIC 4696 : relpersistence = get_rel_persistence(relid);
102 CBC 4696 : if (relpersistence == RELPERSISTENCE_TEMP)
103 5 : MyXactFlags |= XACT_FLAGS_ACCESSEDTEMPNAMESPACE;
104 ECB :
105 : /* Check permissions. */
106 GIC 4696 : aclresult = LockTableAclCheck(relid, lockmode, GetUserId());
107 CBC 4696 : if (aclresult != ACLCHECK_OK)
108 24 : aclcheck_error(aclresult, get_relkind_objtype(get_rel_relkind(relid)), rv->relname);
109 ECB : }
110 :
111 : /*
112 : * Apply LOCK TABLE recursively over an inheritance tree
113 : *
114 : * This doesn't check permission to perform LOCK TABLE on the child tables,
115 : * because getting here means that the user has permission to lock the
116 : * parent which is enough.
117 : */
118 : static void
119 GIC 4625 : LockTableRecurse(Oid reloid, LOCKMODE lockmode, bool nowait)
120 ECB : {
121 : List *children;
122 : ListCell *lc;
123 :
124 GIC 4625 : children = find_all_inheritors(reloid, NoLock, NULL);
125 ECB :
126 GIC 11102 : foreach(lc, children)
127 ECB : {
128 GIC 6477 : Oid childreloid = lfirst_oid(lc);
129 ECB :
130 : /* Parent already locked. */
131 GIC 6477 : if (childreloid == reloid)
132 CBC 4625 : continue;
133 ECB :
134 GIC 1852 : if (!nowait)
135 CBC 1840 : LockRelationOid(childreloid, lockmode);
136 12 : else if (!ConditionalLockRelationOid(childreloid, lockmode))
137 ECB : {
138 : /* try to throw error by name; relation could be deleted... */
139 UIC 0 : char *relname = get_rel_name(childreloid);
140 EUB :
141 UIC 0 : if (!relname)
142 UBC 0 : continue; /* child concurrently dropped, just skip it */
143 0 : ereport(ERROR,
144 EUB : (errcode(ERRCODE_LOCK_NOT_AVAILABLE),
145 : errmsg("could not obtain lock on relation \"%s\"",
146 : relname)));
147 : }
148 :
149 : /*
150 : * Even if we got the lock, child might have been concurrently
151 : * dropped. If so, we can skip it.
152 : */
153 GIC 1852 : if (!SearchSysCacheExists1(RELOID, ObjectIdGetDatum(childreloid)))
154 ECB : {
155 : /* Release useless lock */
156 UIC 0 : UnlockRelationOid(childreloid, lockmode);
157 UBC 0 : continue;
158 EUB : }
159 : }
160 GIC 4625 : }
161 ECB :
162 : /*
163 : * Apply LOCK TABLE recursively over a view
164 : *
165 : * All tables and views appearing in the view definition query are locked
166 : * recursively with the same lock mode.
167 : */
168 :
169 : typedef struct
170 : {
171 : LOCKMODE lockmode; /* lock mode to use */
172 : bool nowait; /* no wait mode */
173 : Oid check_as_user; /* user for checking the privilege */
174 : Oid viewoid; /* OID of the view to be locked */
175 : List *ancestor_views; /* OIDs of ancestor views */
176 : } LockViewRecurse_context;
177 :
178 : static bool
179 GIC 999 : LockViewRecurse_walker(Node *node, LockViewRecurse_context *context)
180 ECB : {
181 GIC 999 : if (node == NULL)
182 CBC 618 : return false;
183 ECB :
184 GIC 381 : if (IsA(node, Query))
185 ECB : {
186 GIC 54 : Query *query = (Query *) node;
187 ECB : ListCell *rtable;
188 :
189 GIC 111 : foreach(rtable, query->rtable)
190 ECB : {
191 GIC 60 : RangeTblEntry *rte = lfirst(rtable);
192 ECB : AclResult aclresult;
193 :
194 GIC 60 : Oid relid = rte->relid;
195 CBC 60 : char relkind = rte->relkind;
196 60 : char *relname = get_rel_name(relid);
197 ECB :
198 : /* Currently, we only allow plain tables or views to be locked. */
199 CBC 60 : if (relkind != RELKIND_RELATION && relkind != RELKIND_PARTITIONED_TABLE &&
200 ECB : relkind != RELKIND_VIEW)
201 GIC 3 : continue;
202 :
203 : /*
204 : * We might be dealing with a self-referential view. If so, we
205 : * can just stop recursing, since we already locked it.
206 ECB : */
207 GIC 57 : if (list_member_oid(context->ancestor_views, relid))
208 CBC 6 : continue;
209 ECB :
210 : /*
211 : * Check permissions as the specified user. This will either be
212 : * the view owner or the current user.
213 : */
214 GBC 51 : aclresult = LockTableAclCheck(relid, context->lockmode,
215 EUB : context->check_as_user);
216 GIC 51 : if (aclresult != ACLCHECK_OK)
217 3 : aclcheck_error(aclresult, get_relkind_objtype(relkind), relname);
218 :
219 : /* We have enough rights to lock the relation; do so. */
220 CBC 48 : if (!context->nowait)
221 48 : LockRelationOid(relid, context->lockmode);
222 UIC 0 : else if (!ConditionalLockRelationOid(relid, context->lockmode))
223 LBC 0 : ereport(ERROR,
224 ECB : (errcode(ERRCODE_LOCK_NOT_AVAILABLE),
225 : errmsg("could not obtain lock on relation \"%s\"",
226 : relname)));
227 :
228 GIC 48 : if (relkind == RELKIND_VIEW)
229 12 : LockViewRecurse(relid, context->lockmode, context->nowait,
230 : context->ancestor_views);
231 36 : else if (rte->inh)
232 36 : LockTableRecurse(relid, context->lockmode, context->nowait);
233 ECB : }
234 :
235 GIC 51 : return query_tree_walker(query,
236 : LockViewRecurse_walker,
237 : context,
238 : QTW_IGNORE_JOINALIASES);
239 ECB : }
240 :
241 GIC 327 : return expression_tree_walker(node,
242 : LockViewRecurse_walker,
243 : context);
244 : }
245 :
246 : static void
247 CBC 45 : LockViewRecurse(Oid reloid, LOCKMODE lockmode, bool nowait,
248 ECB : List *ancestor_views)
249 : {
250 : LockViewRecurse_context context;
251 : Relation view;
252 : Query *viewquery;
253 :
254 : /* caller has already locked the view */
255 CBC 45 : view = table_open(reloid, NoLock);
256 45 : viewquery = get_view_query(view);
257 ECB :
258 : /*
259 : * If the view has the security_invoker property set, check permissions as
260 : * the current user. Otherwise, check permissions as the view owner.
261 : */
262 GIC 45 : context.lockmode = lockmode;
263 CBC 45 : context.nowait = nowait;
264 GIC 45 : if (RelationHasSecurityInvoker(view))
265 CBC 6 : context.check_as_user = GetUserId();
266 : else
267 39 : context.check_as_user = view->rd_rel->relowner;
268 45 : context.viewoid = reloid;
269 GIC 45 : context.ancestor_views = lappend_oid(ancestor_views, reloid);
270 :
271 45 : LockViewRecurse_walker((Node *) viewquery, &context);
272 :
273 42 : context.ancestor_views = list_delete_last(context.ancestor_views);
274 ECB :
275 GIC 42 : table_close(view, NoLock);
276 42 : }
277 :
278 : /*
279 : * Check whether the current user is permitted to lock this relation.
280 ECB : */
281 : static AclResult
282 GIC 4747 : LockTableAclCheck(Oid reloid, LOCKMODE lockmode, Oid userid)
283 ECB : {
284 : AclResult aclresult;
285 : AclMode aclmask;
286 :
287 : /* any of these privileges permit any lock mode */
288 GNC 4747 : aclmask = ACL_MAINTAIN | ACL_UPDATE | ACL_DELETE | ACL_TRUNCATE;
289 :
290 : /* SELECT privileges also permit ACCESS SHARE and below */
291 4747 : if (lockmode <= AccessShareLock)
292 4372 : aclmask |= ACL_SELECT;
293 :
294 : /* INSERT privileges also permit ROW EXCLUSIVE and below */
295 4747 : if (lockmode <= RowExclusiveLock)
296 4426 : aclmask |= ACL_INSERT;
297 :
298 CBC 4747 : aclresult = pg_class_aclcheck(reloid, userid, aclmask);
299 ECB :
300 : /*
301 : * If this is a partition, check permissions of its ancestors if needed.
302 : */
303 GNC 4774 : if (aclresult != ACLCHECK_OK &&
304 27 : has_partition_ancestor_privs(reloid, userid, ACL_MAINTAIN))
305 UNC 0 : aclresult = ACLCHECK_OK;
306 :
307 GBC 4747 : return aclresult;
308 : }
|
