Change the relation_open protocol so that we obtain lock on a relation

(table or index) before trying to open its relcache entry.  This fixes
race conditions in which someone else commits a change to the relation's
catalog entries while we are in process of doing relcache load.  Problems
of that ilk have been reported sporadically for years, but it was not
really practical to fix until recently --- for instance, the recent
addition of WAL-log support for in-place updates helped.

Along the way, remove pg_am.amconcurrent: all AMs are now expected to support
concurrent update.

2 files changed, 35 insertions, 42 deletions

diff --git a/src/backend/utils/cache/catcache.c b/src/backend/utils/cache/catcache.c
index f78db61968e..20b83e196a3 100644
--- a/src/backend/utils/cache/catcache.c
+++ b/src/backend/utils/cache/catcache.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/utils/cache/catcache.c,v 1.131 2006/07/14 14:52:25 momjian Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/cache/catcache.c,v 1.132 2006/07/31 20:09:05 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -904,12 +904,7 @@ CatalogCacheInitializeCache(CatCache *cache)
 
 	CatalogCacheInitializeCache_DEBUG1;
 
-	/*
-	 * Open the relation without locking --- we only need the tupdesc, which
-	 * we assume will never change ...
-	 */
-	relation = heap_open(cache->cc_reloid, NoLock);
-	Assert(RelationIsValid(relation));
+	relation = heap_open(cache->cc_reloid, AccessShareLock);
 
 	/*
 	 * switch to the cache context so our allocations do not vanish at the end
@@ -936,7 +931,7 @@ CatalogCacheInitializeCache(CatCache *cache)
 	 */
 	MemoryContextSwitchTo(oldcxt);
 
-	heap_close(relation, NoLock);
+	heap_close(relation, AccessShareLock);
 
 	CACHE3_elog(DEBUG2, "CatalogCacheInitializeCache: %s, %d keys",
 				cache->cc_relname, cache->cc_nkeys);
@@ -1012,8 +1007,8 @@ InitCatCachePhase2(CatCache *cache)
 	{
 		Relation	idesc;
 
-		idesc = index_open(cache->cc_indexoid);
-		index_close(idesc);
+		idesc = index_open(cache->cc_indexoid, AccessShareLock);
+		index_close(idesc, AccessShareLock);
 	}
 }
 
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 0fe1f29b251..08697d50366 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/utils/cache/relcache.c,v 1.246 2006/07/14 14:52:25 momjian Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/cache/relcache.c,v 1.247 2006/07/31 20:09:05 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -17,7 +17,6 @@
  *		RelationCacheInitialize			- initialize relcache (to empty)
  *		RelationCacheInitializePhase2	- finish initializing relcache
  *		RelationIdGetRelation			- get a reldesc by relation id
- *		RelationIdCacheGetRelation		- get a cached reldesc by relid
  *		RelationClose					- close an open relation
  *
  * NOTES
@@ -34,6 +33,7 @@
 #include "access/heapam.h"
 #include "access/reloptions.h"
 #include "access/xact.h"
+#include "catalog/catalog.h"
 #include "catalog/indexing.h"
 #include "catalog/namespace.h"
 #include "catalog/pg_amop.h"
@@ -763,6 +763,10 @@ equalRuleLocks(RuleLock *rlock1, RuleLock *rlock2)
  *		recycling the given old relation object.  The latter case
  *		supports rebuilding a relcache entry without invalidating
  *		pointers to it.
+ *
+ *		Returns NULL if no pg_class row could be found for the given relid
+ *		(suggesting we are trying to access a just-deleted relation).
+ *		Any other error is reported via elog.
  * --------------------------------
  */
 static Relation
@@ -1388,22 +1392,29 @@ formrdesc(const char *relationName, Oid relationReltype,
  */
 
 /*
- *		RelationIdCacheGetRelation
+ *		RelationIdGetRelation
  *
- *		Lookup an existing reldesc by OID.
+ *		Lookup a reldesc by OID; make one if not already in cache.
  *
- *		Only try to get the reldesc by looking in the cache,
- *		do not go to the disk if it's not present.
+ *		Returns NULL if no pg_class row could be found for the given relid
+ *		(suggesting we are trying to access a just-deleted relation).
+ *		Any other error is reported via elog.
  *
- *		NB: relation ref count is incremented if successful.
+ *		NB: caller should already have at least AccessShareLock on the
+ *		relation ID, else there are nasty race conditions.
+ *
+ *		NB: relation ref count is incremented, or set to 1 if new entry.
  *		Caller should eventually decrement count.  (Usually,
  *		that happens by calling RelationClose().)
  */
 Relation
-RelationIdCacheGetRelation(Oid relationId)
+RelationIdGetRelation(Oid relationId)
 {
 	Relation	rd;
 
+	/*
+	 * first try to find reldesc in the cache
+	 */
 	RelationIdCacheLookup(relationId, rd);
 
 	if (RelationIsValid(rd))
@@ -1412,31 +1423,8 @@ RelationIdCacheGetRelation(Oid relationId)
 		/* revalidate nailed index if necessary */
 		if (!rd->rd_isvalid)
 			RelationReloadClassinfo(rd);
-	}
-
-	return rd;
-}
-
-/*
- *		RelationIdGetRelation
- *
- *		Lookup a reldesc by OID; make one if not already in cache.
- *
- *		NB: relation ref count is incremented, or set to 1 if new entry.
- *		Caller should eventually decrement count.  (Usually,
- *		that happens by calling RelationClose().)
- */
-Relation
-RelationIdGetRelation(Oid relationId)
-{
-	Relation	rd;
-
-	/*
-	 * first try and get a reldesc from the cache
-	 */
-	rd = RelationIdCacheGetRelation(relationId);
-	if (RelationIsValid(rd))
 		return rd;
+	}
 
 	/*
 	 * no reldesc in the cache, so have RelationBuildDesc() build one and add
@@ -2134,6 +2122,16 @@ RelationBuildLocalRelation(const char *relname,
 	}
 
 	/*
+	 * check that hardwired list of shared rels matches what's in the
+	 * bootstrap .bki file.  If you get a failure here during initdb,
+	 * you probably need to fix IsSharedRelation() to match whatever
+	 * you've done to the set of shared relations.
+	 */
+	if (shared_relation != IsSharedRelation(relid))
+		elog(ERROR, "shared_relation flag for \"%s\" does not match IsSharedRelation(%u)",
+			 relname, relid);
+
+	/*
 	 * switch to the cache context to create the relcache entry.
 	 */
 	if (!CacheMemoryContext)