Improve table locking behavior in the face of current DDL.

In the previous coding, callers were faced with an awkward choice:
look up the name, do permissions checks, and then lock the table; or
look up the name, lock the table, and then do permissions checks.
The first choice was wrong because the results of the name lookup
and permissions checks might be out-of-date by the time the table
lock was acquired, while the second allowed a user with no privileges
to interfere with access to a table by users who do have privileges
(e.g. if a malicious backend queues up for an AccessExclusiveLock on
a table on which AccessShareLock is already held, further attempts
to access the table will be blocked until the AccessExclusiveLock
is obtained and the malicious backend's transaction rolls back).

To fix, allow callers of RangeVarGetRelid() to pass a callback which
gets executed after performing the name lookup but before acquiring
the relation lock.  If the name lookup is retried (because
invalidation messages are received), the callback will be re-executed
as well, so we get the best of both worlds.  RangeVarGetRelid() is
renamed to RangeVarGetRelidExtended(); callers not wishing to supply
a callback can continue to invoke it as RangeVarGetRelid(), which is
now a macro.  Since the only one caller that uses nowait = true now
passes a callback anyway, the RangeVarGetRelid() macro defaults nowait
as well.  The callback can also be used for supplemental locking - for
example, REINDEX INDEX needs to acquire the table lock before the index
lock to reduce deadlock possibilities.

There's a lot more work to be done here to fix all the cases where this
can be a problem, but this commit provides the general infrastructure
and fixes the following specific cases: REINDEX INDEX, REINDEX TABLE,
LOCK TABLE, and and DROP TABLE/INDEX/SEQUENCE/VIEW/FOREIGN TABLE.

Per discussion with Noah Misch and Alvaro Herrera.

1 files changed, 97 insertions, 39 deletions

diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c
index 69aa5bf6466..386b95bca20 100644
--- a/src/backend/commands/indexcmds.c
+++ b/src/backend/commands/indexcmds.c
@@ -63,7 +63,10 @@ static void ComputeIndexAttrs(IndexInfo *indexInfo,
 static Oid GetIndexOpClass(List *opclass, Oid attrType,
 				char *accessMethodName, Oid accessMethodId);
 static char *ChooseIndexNameAddition(List *colnames);
-
+static void RangeVarCallbackForReindexTable(const RangeVar *relation,
+								Oid relId, Oid oldRelId, void *arg);
+static void RangeVarCallbackForReindexIndex(const RangeVar *relation,
+								Oid relId, Oid oldRelId, void *arg);
 
 /*
  * CheckIndexCompatible
@@ -129,7 +132,7 @@ CheckIndexCompatible(Oid oldId,
 	Datum		d;
 
 	/* Caller should already have the relation locked in some way. */
-	relationId = RangeVarGetRelid(heapRelation, NoLock, false, false);
+	relationId = RangeVarGetRelid(heapRelation, NoLock, false);
 	/*
 	 * We can pretend isconstraint = false unconditionally.  It only serves to
 	 * decide the text of an error message that should never happen for us.
@@ -1725,33 +1728,74 @@ void
 ReindexIndex(RangeVar *indexRelation)
 {
 	Oid			indOid;
-	HeapTuple	tuple;
+	Oid			heapOid = InvalidOid;
+
+	/* lock level used here should match index lock reindex_index() */
+	indOid = RangeVarGetRelidExtended(indexRelation, AccessExclusiveLock,
+									  false, false,
+									  RangeVarCallbackForReindexIndex,
+									  (void *) &heapOid);
+
+	reindex_index(indOid, false);
+}
+
+/*
+ * Check permissions on table before acquiring relation lock; also lock
+ * the heap before the RangeVarGetRelidExtended takes the index lock, to avoid
+ * deadlocks.
+ */
+static void
+RangeVarCallbackForReindexIndex(const RangeVar *relation,
+								Oid relId, Oid oldRelId, void *arg)
+{
+	char		relkind;
+	Oid		   *heapOid = (Oid *) arg;
 
 	/*
-	 * XXX: This is not safe in the presence of concurrent DDL. We should
-	 * take AccessExclusiveLock here, but that would violate the rule that
-	 * indexes should only be locked after their parent tables.  For now,
-	 * we live with it.
+	 * If we previously locked some other index's heap, and the name we're
+	 * looking up no longer refers to that relation, release the now-useless
+	 * lock.
 	 */
-	indOid = RangeVarGetRelid(indexRelation, NoLock, false, false);
-	tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(indOid));
-	if (!HeapTupleIsValid(tuple))
-		elog(ERROR, "cache lookup failed for relation %u", indOid);
+	if (relId != oldRelId && OidIsValid(oldRelId))
+	{
+		/* lock level here should match reindex_index() heap lock */
+		UnlockRelationOid(*heapOid, ShareLock);
+		*heapOid = InvalidOid;
+	}
+
+	/* If the relation does not exist, there's nothing more to do. */
+	if (!OidIsValid(relId))
+		return;
 
-	if (((Form_pg_class) GETSTRUCT(tuple))->relkind != RELKIND_INDEX)
+	/*
+	 * If the relation does exist, check whether it's an index.  But note
+	 * that the relation might have been dropped between the time we did the
+	 * name lookup and now.  In that case, there's nothing to do.
+	 */
+	relkind = get_rel_relkind(relId);
+	if (!relkind)
+		return;
+	if (relkind != RELKIND_INDEX)
 		ereport(ERROR,
 				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
-				 errmsg("\"%s\" is not an index",
-						indexRelation->relname)));
+				 errmsg("\"%s\" is not an index", relation->relname)));
 
 	/* Check permissions */
-	if (!pg_class_ownercheck(indOid, GetUserId()))
-		aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CLASS,
-					   indexRelation->relname);
-
-	ReleaseSysCache(tuple);
+	if (!pg_class_ownercheck(relId, GetUserId()))
+		aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CLASS, relation->relname);
 
-	reindex_index(indOid, false);
+	/* Lock heap before index to avoid deadlock. */
+	if (relId != oldRelId)
+	{
+		/*
+		 * Lock level here should match reindex_index() heap lock.
+		 * If the OID isn't valid, it means the index as concurrently dropped,
+		 * which is not a problem for us; just return normally.
+		 */
+		*heapOid = IndexGetRelation(relId, true);
+		if (OidIsValid(*heapOid))
+			LockRelationOid(*heapOid, ShareLock);
+	}
 }
 
 /*
@@ -1762,27 +1806,10 @@ void
 ReindexTable(RangeVar *relation)
 {
 	Oid			heapOid;
-	HeapTuple	tuple;
 
 	/* The lock level used here should match reindex_relation(). */
-	heapOid = RangeVarGetRelid(relation, ShareLock, false, false);
-	tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(heapOid));
-	if (!HeapTupleIsValid(tuple))		/* shouldn't happen */
-		elog(ERROR, "cache lookup failed for relation %u", heapOid);
-
-	if (((Form_pg_class) GETSTRUCT(tuple))->relkind != RELKIND_RELATION &&
-		((Form_pg_class) GETSTRUCT(tuple))->relkind != RELKIND_TOASTVALUE)
-		ereport(ERROR,
-				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
-				 errmsg("\"%s\" is not a table",
-						relation->relname)));
-
-	/* Check permissions */
-	if (!pg_class_ownercheck(heapOid, GetUserId()))
-		aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CLASS,
-					   relation->relname);
-
-	ReleaseSysCache(tuple);
+	heapOid = RangeVarGetRelidExtended(relation, ShareLock, false, false,
+									   RangeVarCallbackForReindexTable, NULL);
 
 	if (!reindex_relation(heapOid, REINDEX_REL_PROCESS_TOAST))
 		ereport(NOTICE,
@@ -1791,6 +1818,37 @@ ReindexTable(RangeVar *relation)
 }
 
 /*
+ * Check permissions on table before acquiring relation lock.
+ */
+static void
+RangeVarCallbackForReindexTable(const RangeVar *relation,
+								Oid relId, Oid oldRelId, void *arg)
+{
+	char		relkind;
+
+	/* Nothing to do if the relation was not found. */
+	if (!OidIsValid(relId))
+		return;
+
+	/*
+	 * If the relation does exist, check whether it's an index.  But note
+	 * that the relation might have been dropped between the time we did the
+	 * name lookup and now.  In that case, there's nothing to do.
+	 */
+	relkind = get_rel_relkind(relId);
+	if (!relkind)
+		return;
+	if (relkind != RELKIND_RELATION && relkind != RELKIND_TOASTVALUE)
+		ereport(ERROR,
+				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
+				 errmsg("\"%s\" is not a table", relation->relname)));
+
+	/* Check permissions */
+	if (!pg_class_ownercheck(relId, GetUserId()))
+		aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CLASS, relation->relname);
+}
+
+/*
  * ReindexDatabase
  *		Recreate indexes of a database.
  *