lock-namespace.patch

diff --git a/src/backend/catalog/namespace.c b/src/backend/catalog/namespace.c
index 6d4d4b1..519794a 100644
--- a/src/backend/catalog/namespace.c
+++ b/src/backend/catalog/namespace.c
@@ -463,31 +463,63 @@ RangeVarGetCreationNamespace(const RangeVar *newRelation)
 
 /*
  * RangeVarGetAndCheckCreationNamespace
- *		As RangeVarGetCreationNamespace, but with a permissions check.
+ *		As RangeVarGetCreationNamespace, but check permissions and lock the
+ *		target namespace.
  */
 Oid
 RangeVarGetAndCheckCreationNamespace(const RangeVar *newRelation)
 {
-	Oid			namespaceId;
-
-	namespaceId = RangeVarGetCreationNamespace(newRelation);
+	Oid			namespace_oid;
+	Oid			old_namespace_oid = InvalidOid;
+	bool		retry = false;
+	uint64		inval_count = SharedInvalidMessageCounter;
 
-	/*
-	 * Check we have permission to create there. Skip check if bootstrapping,
-	 * since permissions machinery may not be working yet.
-	 */
-	if (!IsBootstrapProcessingMode())
+	while (1)
 	{
 		AclResult	aclresult;
 
-		aclresult = pg_namespace_aclcheck(namespaceId, GetUserId(),
+		namespace_oid = RangeVarGetCreationNamespace(newRelation);
+
+		/*
+		 * In bootstrap mode, permissions machinery may not be working yet;
+		 * and there's no risk of concurrent activity, so we're done.
+		 */
+		if (IsBootstrapProcessingMode())
+			break;
+
+		/*
+		 * If the meaning of the name doesn't change on retry, we're done.
+		 * If it does, we'll have to release the lock we took previously
+		 * and lock the new referrent of the name instead.
+		 */
+		if (retry)
+		{
+			if (namespace_oid == old_namespace_oid)
+				break;
+			else
+				UnlockDatabaseObject(NamespaceRelationId, old_namespace_oid,
+									 0, AccessShareLock);
+		}
+
+		/* Check permissions. */
+		aclresult = pg_namespace_aclcheck(namespace_oid, GetUserId(),
 										  ACL_CREATE);
 		if (aclresult != ACLCHECK_OK)
 			aclcheck_error(aclresult, ACL_KIND_NAMESPACE,
-						   get_namespace_name(namespaceId));
+						   get_namespace_name(namespace_oid));
+
+		/* Lock target namespace against concurrent drop. */
+		LockDatabaseObject(NamespaceRelationId, namespace_oid, 0,
+						   AccessShareLock);
+
+		/* If nothing's changed under us, we are done.  Else, retry. */
+		if (inval_count == SharedInvalidMessageCounter)
+			break;
+		retry = true;
+		old_namespace_oid = namespace_oid;
 	}
 
-	return namespaceId;
+	return namespace_oid;
 }
 
 /*