atomic-openrv-poc.patch

*** a/src/backend/access/heap/heapam.c
--- b/src/backend/access/heap/heapam.c
***************
*** 970,995 **** relation_openrv(const RangeVar *relation, LOCKMODE lockmode)
  {
  	Oid			relOid;
  
! 	/*
! 	 * Check for shared-cache-inval messages before trying to open the
! 	 * relation.  This is needed to cover the case where the name identifies a
! 	 * rel that has been dropped and recreated since the start of our
! 	 * transaction: if we don't flush the old syscache entry then we'll latch
! 	 * onto that entry and suffer an error when we do RelationIdGetRelation.
! 	 * Note that relation_open does not need to do this, since a relation's
! 	 * OID never changes.
! 	 *
! 	 * We skip this if asked for NoLock, on the assumption that the caller has
! 	 * already ensured some appropriate lock is held.
! 	 */
! 	if (lockmode != NoLock)
! 		AcceptInvalidationMessages();
! 
! 	/* Look up the appropriate relation using namespace search */
! 	relOid = RangeVarGetRelid(relation, false);
  
  	/* Let relation_open do the rest */
! 	return relation_open(relOid, lockmode);
  }
  
  /* ----------------
--- 970,980 ----
  {
  	Oid			relOid;
  
! 	/* Look up and lock the appropriate relation using namespace search */
! 	relOid = RangeVarLockRelid(relation, lockmode, false);
  
  	/* Let relation_open do the rest */
! 	return relation_open(relOid, NoLock);
  }
  
  /* ----------------
***************
*** 1005,1034 **** try_relation_openrv(const RangeVar *relation, LOCKMODE lockmode)
  {
  	Oid			relOid;
  
! 	/*
! 	 * Check for shared-cache-inval messages before trying to open the
! 	 * relation.  This is needed to cover the case where the name identifies a
! 	 * rel that has been dropped and recreated since the start of our
! 	 * transaction: if we don't flush the old syscache entry then we'll latch
! 	 * onto that entry and suffer an error when we do RelationIdGetRelation.
! 	 * Note that relation_open does not need to do this, since a relation's
! 	 * OID never changes.
! 	 *
! 	 * We skip this if asked for NoLock, on the assumption that the caller has
! 	 * already ensured some appropriate lock is held.
! 	 */
! 	if (lockmode != NoLock)
! 		AcceptInvalidationMessages();
! 
! 	/* Look up the appropriate relation using namespace search */
! 	relOid = RangeVarGetRelid(relation, true);
  
  	/* Return NULL on not-found */
  	if (!OidIsValid(relOid))
  		return NULL;
  
  	/* Let relation_open do the rest */
! 	return relation_open(relOid, lockmode);
  }
  
  /* ----------------
--- 990,1004 ----
  {
  	Oid			relOid;
  
! 	/* Look up and lock the appropriate relation using namespace search */
! 	relOid = RangeVarLockRelid(relation, lockmode, true);
  
  	/* Return NULL on not-found */
  	if (!OidIsValid(relOid))
  		return NULL;
  
  	/* Let relation_open do the rest */
! 	return relation_open(relOid, NoLock);
  }
  
  /* ----------------
*** a/src/backend/catalog/namespace.c
--- b/src/backend/catalog/namespace.c
***************
*** 42,47 ****
--- 42,48 ----
  #include "parser/parse_func.h"
  #include "storage/backendid.h"
  #include "storage/ipc.h"
+ #include "storage/lmgr.h"
  #include "utils/acl.h"
  #include "utils/builtins.h"
  #include "utils/guc.h"
***************
*** 282,287 **** RangeVarGetRelid(const RangeVar *relation, bool failOK)
--- 283,340 ----
  }
  
  /*
+  * RangeVarLockRelid
+  *		Like RangeVarGetRelid, but simulatenously acquire the specified lock on
+  *		the relation.  This works properly in the face of concurrent DDL that
+  *		may drop, create or rename relations.
+  *
+  * If the relation is not found and failOK = true, take no lock and return
+  * InvalidOid.  Otherwise, raise an error.
+  */
+ Oid
+ RangeVarLockRelid(const RangeVar *relation, LOCKMODE lockmode,
+ 				  bool failOK)
+ {
+ 	Oid			relOid1,
+ 				relOid2;
+ 
+ 	/*
+ 	 * First attempt.  If the caller requested NoLock, it already acquired an
+ 	 * appropriate lock and has called AcceptInvalidationMessages() since doing
+ 	 * so.  In this case, our first search is always correct, and we degenerate
+ 	 * to behave exactly like RangeVarGetRelid().
+ 	 */
+ 	relOid1 = RangeVarGetRelid(relation, failOK);
+ 	if (lockmode == NoLock)
+ 		return relOid1;
+ 
+ 	/*
+ 	 * By the time we acquire the lock, our RangeVar may denote a different
+ 	 * relation or no relation at all.  In particular, this can happen when the
+ 	 * lock acquisition blocks on a transaction performing DROP or ALTER TABLE
+ 	 * RENAME.  However, once and while we do hold a lock of any level, we can
+ 	 * count on the name correspondence remaining stable.
+ 	 */
+ 	do
+ 	{
+ 		/* Not-found is always final. */
+ 		if (!OidIsValid(relOid1))
+ 			return relOid1;
+ 
+ 		LockRelationOid(relOid1, lockmode);
+ 
+ 		/* Make recent DDL effects visible.  Names are stable; search again. */
+ 		AcceptInvalidationMessages();
+ 		relOid2 = relOid1;
+ 		relOid1 = RangeVarGetRelid(relation, failOK);
+ 
+ 		/* Done when our RangeVar denotes the same relation we locked. */
+ 	} while (relOid1 != relOid2);
+ 
+ 	return relOid1;
+ }
+ 
+ /*
   * RangeVarGetCreationNamespace
   *		Given a RangeVar describing a to-be-created relation,
   *		choose which namespace to create it in.
*** a/src/include/catalog/namespace.h
--- b/src/include/catalog/namespace.h
***************
*** 48,53 **** typedef struct OverrideSearchPath
--- 48,55 ----
  
  
  extern Oid	RangeVarGetRelid(const RangeVar *relation, bool failOK);
+ extern Oid	RangeVarLockRelid(const RangeVar *relation, LOCKMODE lockmode,
+ 				  bool failOK);
  extern Oid	RangeVarGetCreationNamespace(const RangeVar *newRelation);
  extern Oid	RelnameGetRelid(const char *relname);
  extern bool RelationIsVisible(Oid relid);