*** 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);