Fix LOCK TABLE to eliminate the race condition that could make it give weird

errors when tables are concurrently dropped.  To do this we must take lock
on each relation before we check its privileges.  The old code was trying
to do that the other way around, which is a bit pointless when there are lots
of other commands that lock relations before checking privileges.  I did keep
it checking each relation's privilege before locking the next relation, which
is a detail that ALTER TABLE isn't too picky about.

diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 52115cf64e1832679b4762ed4e025ba3be29d1ea..ad11770aec9d22987bccf4fc51a917a9b8a21c8c 100644 (file)
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -947,56 +947,6 @@ try_relation_open(Oid relationId, LOCKMODE lockmode)
        return r;
 }
 
-/* ----------------
- *             relation_open_nowait - open but don't wait for lock
- *
- *             Same as relation_open, except throw an error instead of waiting
- *             when the requested lock is not immediately obtainable.
- * ----------------
- */
-Relation
-relation_open_nowait(Oid relationId, LOCKMODE lockmode)
-{
-       Relation        r;
-
-       Assert(lockmode >= NoLock && lockmode < MAX_LOCKMODES);
-
-       /* Get the lock before trying to open the relcache entry */
-       if (lockmode != NoLock)
-       {
-               if (!ConditionalLockRelationOid(relationId, lockmode))
-               {
-                       /* try to throw error by name; relation could be deleted... */
-                       char       *relname = get_rel_name(relationId);
-
-                       if (relname)
-                               ereport(ERROR,
-                                               (errcode(ERRCODE_LOCK_NOT_AVAILABLE),
-                                                errmsg("could not obtain lock on relation \"%s\"",
-                                                               relname)));
-                       else
-                               ereport(ERROR,
-                                               (errcode(ERRCODE_LOCK_NOT_AVAILABLE),
-                                         errmsg("could not obtain lock on relation with OID %u",
-                                                        relationId)));
-               }
-       }
-
-       /* The relcache does all the real work... */
-       r = RelationIdGetRelation(relationId);
-
-       if (!RelationIsValid(r))
-               elog(ERROR, "could not open relation with OID %u", relationId);
-
-       /* Make note that we've accessed a temporary relation */
-       if (r->rd_istemp)
-               MyXactAccessedTempRel = true;
-
-       pgstat_initstats(r);
-
-       return r;
-}
-
 /* ----------------
  *             relation_openrv - open any relation specified by a RangeVar
  *

diff --git a/src/backend/commands/lockcmds.c b/src/backend/commands/lockcmds.c
index ba493bdf40434ac46982d8984dfa331b0412c5c7..9d85690b6ddf388c0b7db37992dc9d99c3438b82 100644 (file)
--- a/src/backend/commands/lockcmds.c
+++ b/src/backend/commands/lockcmds.c
@@ -1,7 +1,7 @@
 /*-------------------------------------------------------------------------
  *
  * lockcmds.c
- *       Lock command support code
+ *       LOCK command support code
  *
  * Portions Copyright (c) 1996-2009, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
@@ -20,9 +20,12 @@
 #include "commands/lockcmds.h"
 #include "miscadmin.h"
 #include "parser/parse_clause.h"
+#include "storage/lmgr.h"
 #include "utils/acl.h"
 #include "utils/lsyscache.h"
-#include "utils/rel.h"
+
+static void LockTableRecurse(Oid reloid, RangeVar *rv,
+                                                        LOCKMODE lockmode, bool nowait, bool recurse);
 
 
 /*
@@ -34,57 +37,127 @@ LockTableCommand(LockStmt *lockstmt)
        ListCell   *p;
 
        /*
-        * Iterate over the list and open, lock, and close the relations one at a
-        * time
+        * Iterate over the list and process the named relations one at a time
         */
-
        foreach(p, lockstmt->relations)
        {
-               RangeVar   *relation = lfirst(p);
-               Oid                     reloid;
+               RangeVar   *relation = (RangeVar *) lfirst(p);
                bool            recurse = interpretInhOption(relation->inhOpt);
-               List       *children_and_self;
-               ListCell   *child;
+               Oid                     reloid;
 
                reloid = RangeVarGetRelid(relation, false);
 
-               /* XXX NoLock here is not really a good idea */
-               if (recurse)
-                       children_and_self = find_all_inheritors(reloid, NoLock);
-               else
-                       children_and_self = list_make1_oid(reloid);
+               LockTableRecurse(reloid, relation,
+                                                lockstmt->mode, lockstmt->nowait, recurse);
+       }
+}
 
-               foreach(child, children_and_self)
+/*
+ * Apply LOCK TABLE recursively over an inheritance tree
+ *
+ * At top level, "rv" is the original command argument; we use it to throw
+ * an appropriate error message if the relation isn't there.  Below top level,
+ * "rv" is NULL and we should just silently ignore any dropped child rel.
+ */
+static void
+LockTableRecurse(Oid reloid, RangeVar *rv,
+                                LOCKMODE lockmode, bool nowait, bool recurse)
+{
+       Relation        rel;
+       AclResult       aclresult;
+
+       /*
+        * Acquire the lock.  We must do this first to protect against
+        * concurrent drops.  Note that a lock against an already-dropped
+        * relation's OID won't fail.
+        */
+       if (nowait)
+       {
+               if (!ConditionalLockRelationOid(reloid, lockmode))
                {
-                       Oid                     childreloid = lfirst_oid(child);
-                       Relation        rel;
-                       AclResult       aclresult;
-
-                       /* We don't want to open the relation until we've checked privilege. */
-                       if (lockstmt->mode == AccessShareLock)
-                               aclresult = pg_class_aclcheck(childreloid, GetUserId(),
-                                                                                         ACL_SELECT);
+                       /* try to throw error by name; relation could be deleted... */
+                       char       *relname = rv ? rv->relname : get_rel_name(reloid);
+
+                       if (relname)
+                               ereport(ERROR,
+                                               (errcode(ERRCODE_LOCK_NOT_AVAILABLE),
+                                                errmsg("could not obtain lock on relation \"%s\"",
+                                                               relname)));
                        else
-                               aclresult = pg_class_aclcheck(childreloid, GetUserId(),
-                                                                                         ACL_UPDATE | ACL_DELETE | ACL_TRUNCATE);
+                               ereport(ERROR,
+                                               (errcode(ERRCODE_LOCK_NOT_AVAILABLE),
+                                         errmsg("could not obtain lock on relation with OID %u",
+                                                        reloid)));
+               }
+       }
+       else
+               LockRelationOid(reloid, lockmode);
 
-                       if (aclresult != ACLCHECK_OK)
-                               aclcheck_error(aclresult, ACL_KIND_CLASS,
-                                                          get_rel_name(childreloid));
+       /*
+        * Now that we have the lock, check to see if the relation really exists
+        * or not.
+        */
+       rel = try_relation_open(reloid, NoLock);
 
-                       if (lockstmt->nowait)
-                               rel = relation_open_nowait(childreloid, lockstmt->mode);
-                       else
-                               rel = relation_open(childreloid, lockstmt->mode);
+       if (!rel)
+       {
+               /* Release useless lock */
+               UnlockRelationOid(reloid, lockmode);
 
-                       /* Currently, we only allow plain tables to be locked */
-                       if (rel->rd_rel->relkind != RELKIND_RELATION)
+               /* At top level, throw error; otherwise, ignore this child rel */
+               if (rv)
+               {
+                       if (rv->schemaname)
                                ereport(ERROR,
-                                               (errcode(ERRCODE_WRONG_OBJECT_TYPE),
-                                                errmsg("\"%s\" is not a table",
-                                                               get_rel_name(childreloid))));
+                                               (errcode(ERRCODE_UNDEFINED_TABLE),
+                                                errmsg("relation \"%s.%s\" does not exist",
+                                                               rv->schemaname, rv->relname)));
+                       else
+                               ereport(ERROR,
+                                               (errcode(ERRCODE_UNDEFINED_TABLE),
+                                                errmsg("relation \"%s\" does not exist",
+                                                               rv->relname)));
+               }
+
+               return;
+       }
 
-                       relation_close(rel, NoLock);    /* close rel, keep lock */
+       /* Verify adequate privilege */
+       if (lockmode == AccessShareLock)
+               aclresult = pg_class_aclcheck(reloid, GetUserId(),
+                                                                         ACL_SELECT);
+       else
+               aclresult = pg_class_aclcheck(reloid, GetUserId(),
+                                                                         ACL_UPDATE | ACL_DELETE | ACL_TRUNCATE);
+       if (aclresult != ACLCHECK_OK)
+               aclcheck_error(aclresult, ACL_KIND_CLASS,
+                                          RelationGetRelationName(rel));
+
+       /* Currently, we only allow plain tables to be locked */
+       if (rel->rd_rel->relkind != RELKIND_RELATION)
+               ereport(ERROR,
+                               (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+                                errmsg("\"%s\" is not a table",
+                                               RelationGetRelationName(rel))));
+
+       /*
+        * If requested, recurse to children.  We use find_inheritance_children
+        * not find_all_inheritors to avoid taking locks far in advance of
+        * checking privileges.  This means we'll visit multiply-inheriting
+        * children more than once, but that's no problem.
+        */
+       if (recurse)
+       {
+               List   *children = find_inheritance_children(reloid, NoLock);
+               ListCell *lc;
+
+               foreach(lc, children)
+               {
+                       Oid                     childreloid = lfirst_oid(lc);
+
+                       LockTableRecurse(childreloid, NULL, lockmode, nowait, recurse);
                }
        }
+
+       relation_close(rel, NoLock);    /* close rel, keep lock */
 }

diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h
index a5d97697944a7d86ad438765e6544d7e0828e54a..609aabbeea6ece88b7930600f510c4e862fb49d5 100644 (file)
--- a/src/include/access/heapam.h
+++ b/src/include/access/heapam.h
@@ -49,7 +49,6 @@ typedef enum
 /* in heap/heapam.c */
 extern Relation relation_open(Oid relationId, LOCKMODE lockmode);
 extern Relation try_relation_open(Oid relationId, LOCKMODE lockmode);
-extern Relation relation_open_nowait(Oid relationId, LOCKMODE lockmode);
 extern Relation relation_openrv(const RangeVar *relation, LOCKMODE lockmode);
 extern Relation try_relation_openrv(const RangeVar *relation, LOCKMODE lockmode);
 extern void relation_close(Relation relation, LOCKMODE lockmode);