Catcaches can now store negative entries as well as positive ones, to

speed up repetitive failed searches; per pghackers discussion in late
January.  inval.c logic substantially simplified, since we can now treat
inserts and deletes alike as far as inval events are concerned.  Some
repair work needed in heap_create_with_catalog, which turns out to have
been doing CommandCounterIncrement at a point where the new relation has
non-self-consistent catalog entries.  With the new inval code, that
resulted in assert failures during a relcache entry rebuild.

3 files changed, 625 insertions, 435 deletions

diff --git a/src/backend/utils/cache/catcache.c b/src/backend/utils/cache/catcache.c
index b79d671255..e04792d71f 100644
--- a/src/backend/utils/cache/catcache.c
+++ b/src/backend/utils/cache/catcache.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/utils/cache/catcache.c,v 1.89 2002/03/02 21:39:32 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/utils/cache/catcache.c,v 1.90 2002/03/03 17:47:55 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -91,15 +91,15 @@ static const Oid eqproc[] = {
 #define EQPROC(SYSTEMTYPEOID)	eqproc[(SYSTEMTYPEOID)-BOOLOID]
 
 
-static void CatCacheRemoveCTup(CatCache *cache, CatCTup *ct);
-static Index CatalogCacheComputeHashIndex(CatCache *cache,
+static uint32 CatalogCacheComputeHashValue(CatCache *cache,
 							 ScanKey cur_skey);
-static Index CatalogCacheComputeTupleHashIndex(CatCache *cache,
+static uint32 CatalogCacheComputeTupleHashValue(CatCache *cache,
 								  HeapTuple tuple);
-static void CatalogCacheInitializeCache(CatCache *cache);
 #ifdef CATCACHE_STATS
 static void CatCachePrintStats(void);
 #endif
+static void CatCacheRemoveCTup(CatCache *cache, CatCTup *ct);
+static void CatalogCacheInitializeCache(CatCache *cache);
 
 
 /*
@@ -136,221 +136,17 @@ GetCCHashFunc(Oid keytype)
 	}
 }
 
-#ifdef CATCACHE_STATS
-
-static void
-CatCachePrintStats(void)
-{
-	CatCache   *cache;
-	long		cc_searches = 0;
-	long		cc_hits = 0;
-	long		cc_newloads = 0;
-
-	elog(LOG, "Catcache stats dump: %d/%d tuples in catcaches",
-		 CacheHdr->ch_ntup, CacheHdr->ch_maxtup);
-
-	for (cache = CacheHdr->ch_caches; cache; cache = cache->cc_next)
-	{
-		if (cache->cc_ntup == 0 && cache->cc_searches == 0)
-			continue;			/* don't print unused caches */
-		elog(LOG, "Catcache %s/%s: %d tup, %ld srch, %ld hits, %ld loads, %ld not found",
-			 cache->cc_relname,
-			 cache->cc_indname,
-			 cache->cc_ntup,
-			 cache->cc_searches,
-			 cache->cc_hits,
-			 cache->cc_newloads,
-			 cache->cc_searches - cache->cc_hits - cache->cc_newloads);
-		cc_searches += cache->cc_searches;
-		cc_hits += cache->cc_hits;
-		cc_newloads += cache->cc_newloads;
-	}
-	elog(LOG, "Catcache totals: %d tup, %ld srch, %ld hits, %ld loads, %ld not found",
-		 CacheHdr->ch_ntup,
-		 cc_searches,
-		 cc_hits,
-		 cc_newloads,
-		 cc_searches - cc_hits - cc_newloads);
-}
-
-#endif /* CATCACHE_STATS */
-
-
-/*
- * Standard routine for creating cache context if it doesn't exist yet
- *
- * There are a lot of places (probably far more than necessary) that check
- * whether CacheMemoryContext exists yet and want to create it if not.
- * We centralize knowledge of exactly how to create it here.
- */
-void
-CreateCacheMemoryContext(void)
-{
-	/*
-	 * Purely for paranoia, check that context doesn't exist; caller
-	 * probably did so already.
-	 */
-	if (!CacheMemoryContext)
-		CacheMemoryContext = AllocSetContextCreate(TopMemoryContext,
-												   "CacheMemoryContext",
-												ALLOCSET_DEFAULT_MINSIZE,
-											   ALLOCSET_DEFAULT_INITSIZE,
-											   ALLOCSET_DEFAULT_MAXSIZE);
-}
-
-
 /*
- *		CatalogCacheInitializeCache
+ *		CatalogCacheComputeHashValue
  *
- * This function does final initialization of a catcache: obtain the tuple
- * descriptor and set up the hash and equality function links.	We assume
- * that the relcache entry can be opened at this point!
+ * Compute the hash value associated with a given set of lookup keys
  */
-#ifdef CACHEDEBUG
-#define CatalogCacheInitializeCache_DEBUG1 \
-	elog(LOG, "CatalogCacheInitializeCache: cache @%p %s", cache, \
-		 cache->cc_relname)
-
-#define CatalogCacheInitializeCache_DEBUG2 \
-do { \
-		if (cache->cc_key[i] > 0) { \
-			elog(LOG, "CatalogCacheInitializeCache: load %d/%d w/%d, %u", \
-				i+1, cache->cc_nkeys, cache->cc_key[i], \
-				 tupdesc->attrs[cache->cc_key[i] - 1]->atttypid); \
-		} else { \
-			elog(LOG, "CatalogCacheInitializeCache: load %d/%d w/%d", \
-				i+1, cache->cc_nkeys, cache->cc_key[i]); \
-		} \
-} while(0)
-
-#else
-#define CatalogCacheInitializeCache_DEBUG1
-#define CatalogCacheInitializeCache_DEBUG2
-#endif
-
-static void
-CatalogCacheInitializeCache(CatCache *cache)
+static uint32
+CatalogCacheComputeHashValue(CatCache *cache, ScanKey cur_skey)
 {
-	Relation	relation;
-	MemoryContext oldcxt;
-	TupleDesc	tupdesc;
-	int			i;
-
-	CatalogCacheInitializeCache_DEBUG1;
-
-	/*
-	 * Open the relation without locking --- we only need the tupdesc,
-	 * which we assume will never change ...
-	 */
-	relation = heap_openr(cache->cc_relname, NoLock);
-	Assert(RelationIsValid(relation));
-
-	/*
-	 * switch to the cache context so our allocations do not vanish at the
-	 * end of a transaction
-	 */
-	Assert(CacheMemoryContext != NULL);
-
-	oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
-
-	/*
-	 * copy the relcache's tuple descriptor to permanent cache storage
-	 */
-	tupdesc = CreateTupleDescCopyConstr(RelationGetDescr(relation));
-
-	/*
-	 * get the relation's relisshared flag, too
-	 */
-	cache->cc_relisshared = RelationGetForm(relation)->relisshared;
-
-	/*
-	 * return to the caller's memory context and close the rel
-	 */
-	MemoryContextSwitchTo(oldcxt);
-
-	heap_close(relation, NoLock);
+	uint32		hashValue = 0;
 
-	CACHE3_elog(LOG, "CatalogCacheInitializeCache: %s, %d keys",
-				cache->cc_relname, cache->cc_nkeys);
-
-	/*
-	 * initialize cache's key information
-	 */
-	for (i = 0; i < cache->cc_nkeys; ++i)
-	{
-		Oid			keytype;
-
-		CatalogCacheInitializeCache_DEBUG2;
-
-		if (cache->cc_key[i] > 0)
-			keytype = tupdesc->attrs[cache->cc_key[i] - 1]->atttypid;
-		else
-		{
-			if (cache->cc_key[i] != ObjectIdAttributeNumber)
-				elog(FATAL, "CatalogCacheInit: only sys attr supported is OID");
-			keytype = OIDOID;
-		}
-
-		cache->cc_hashfunc[i] = GetCCHashFunc(keytype);
-
-		/*
-		 * If GetCCHashFunc liked the type, safe to index into eqproc[]
-		 */
-		cache->cc_skey[i].sk_procedure = EQPROC(keytype);
-
-		/* Do function lookup */
-		fmgr_info_cxt(cache->cc_skey[i].sk_procedure,
-					  &cache->cc_skey[i].sk_func,
-					  CacheMemoryContext);
-
-		/* Initialize sk_attno suitably for HeapKeyTest() and heap scans */
-		cache->cc_skey[i].sk_attno = cache->cc_key[i];
-
-		CACHE4_elog(LOG, "CatalogCacheInit %s %d %p",
-					cache->cc_relname,
-					i,
-					cache);
-	}
-
-	/*
-	 * mark this cache fully initialized
-	 */
-	cache->cc_tupdesc = tupdesc;
-}
-
-/*
- * InitCatCachePhase2 -- external interface for CatalogCacheInitializeCache
- *
- * The only reason to call this routine is to ensure that the relcache
- * has created entries for all the catalogs and indexes referenced by
- * catcaches.  Therefore, open the index too.  An exception is the indexes
- * on pg_am, which we don't use (cf. IndexScanOK).
- */
-void
-InitCatCachePhase2(CatCache *cache)
-{
-	if (cache->cc_tupdesc == NULL)
-		CatalogCacheInitializeCache(cache);
-
-	if (cache->id != AMOID &&
-		cache->id != AMNAME)
-	{
-		Relation	idesc;
-
-		idesc = index_openr(cache->cc_indname);
-		index_close(idesc);
-	}
-}
-
-/*
- *		CatalogCacheComputeHashIndex
- */
-static Index
-CatalogCacheComputeHashIndex(CatCache *cache, ScanKey cur_skey)
-{
-	uint32		hashIndex = 0;
-
-	CACHE4_elog(LOG, "CatalogCacheComputeHashIndex %s %d %p",
+	CACHE4_elog(DEBUG1, "CatalogCacheComputeHashValue %s %d %p",
 				cache->cc_relname,
 				cache->cc_nkeys,
 				cache);
@@ -358,39 +154,40 @@ CatalogCacheComputeHashIndex(CatCache *cache, ScanKey cur_skey)
 	switch (cache->cc_nkeys)
 	{
 		case 4:
-			hashIndex ^=
+			hashValue ^=
 				DatumGetUInt32(DirectFunctionCall1(cache->cc_hashfunc[3],
 										  cur_skey[3].sk_argument)) << 9;
 			/* FALLTHROUGH */
 		case 3:
-			hashIndex ^=
+			hashValue ^=
 				DatumGetUInt32(DirectFunctionCall1(cache->cc_hashfunc[2],
 										  cur_skey[2].sk_argument)) << 6;
 			/* FALLTHROUGH */
 		case 2:
-			hashIndex ^=
+			hashValue ^=
 				DatumGetUInt32(DirectFunctionCall1(cache->cc_hashfunc[1],
 										  cur_skey[1].sk_argument)) << 3;
 			/* FALLTHROUGH */
 		case 1:
-			hashIndex ^=
+			hashValue ^=
 				DatumGetUInt32(DirectFunctionCall1(cache->cc_hashfunc[0],
 											   cur_skey[0].sk_argument));
 			break;
 		default:
-			elog(FATAL, "CCComputeHashIndex: %d cc_nkeys", cache->cc_nkeys);
+			elog(FATAL, "CCComputeHashValue: %d cc_nkeys", cache->cc_nkeys);
 			break;
 	}
-	hashIndex %= (uint32) cache->cc_size;
-	return (Index) hashIndex;
+
+	return hashValue;
 }
 
 /*
- *		CatalogCacheComputeTupleHashIndex
+ *		CatalogCacheComputeTupleHashValue
+ *
+ * Compute the hash value associated with a given tuple to be cached
  */
-static Index
-CatalogCacheComputeTupleHashIndex(CatCache *cache,
-								  HeapTuple tuple)
+static uint32
+CatalogCacheComputeTupleHashValue(CatCache *cache, HeapTuple tuple)
 {
 	ScanKeyData cur_skey[4];
 	bool		isNull = false;
@@ -442,16 +239,75 @@ CatalogCacheComputeTupleHashIndex(CatCache *cache,
 			Assert(!isNull);
 			break;
 		default:
-			elog(FATAL, "CCComputeTupleHashIndex: %d cc_nkeys",
+			elog(FATAL, "CCComputeTupleHashValue: %d cc_nkeys",
 				 cache->cc_nkeys);
 			break;
 	}
 
-	return CatalogCacheComputeHashIndex(cache, cur_skey);
+	return CatalogCacheComputeHashValue(cache, cur_skey);
 }
 
+
+#ifdef CATCACHE_STATS
+
+static void
+CatCachePrintStats(void)
+{
+	CatCache   *cache;
+	long		cc_searches = 0;
+	long		cc_hits = 0;
+	long		cc_neg_hits = 0;
+	long		cc_newloads = 0;
+	long		cc_invals = 0;
+	long		cc_discards = 0;
+
+	elog(DEBUG1, "Catcache stats dump: %d/%d tuples in catcaches",
+		 CacheHdr->ch_ntup, CacheHdr->ch_maxtup);
+
+	for (cache = CacheHdr->ch_caches; cache; cache = cache->cc_next)
+	{
+		if (cache->cc_ntup == 0 && cache->cc_searches == 0)
+			continue;			/* don't print unused caches */
+		elog(DEBUG1, "Catcache %s/%s: %d tup, %ld srch, %ld+%ld=%ld hits, %ld+%ld=%ld loads, %ld invals, %ld discards",
+			 cache->cc_relname,
+			 cache->cc_indname,
+			 cache->cc_ntup,
+			 cache->cc_searches,
+			 cache->cc_hits,
+			 cache->cc_neg_hits,
+			 cache->cc_hits + cache->cc_neg_hits,
+			 cache->cc_newloads,
+			 cache->cc_searches - cache->cc_hits - cache->cc_neg_hits - cache->cc_newloads,
+			 cache->cc_searches - cache->cc_hits - cache->cc_neg_hits,
+			 cache->cc_invals,
+			 cache->cc_discards);
+		cc_searches += cache->cc_searches;
+		cc_hits += cache->cc_hits;
+		cc_neg_hits += cache->cc_neg_hits;
+		cc_newloads += cache->cc_newloads;
+		cc_invals += cache->cc_invals;
+		cc_discards += cache->cc_discards;
+	}
+	elog(DEBUG1, "Catcache totals: %d tup, %ld srch, %ld+%ld=%ld hits, %ld+%ld=%ld loads, %ld invals, %ld discards",
+		 CacheHdr->ch_ntup,
+		 cc_searches,
+		 cc_hits,
+		 cc_neg_hits,
+		 cc_hits + cc_neg_hits,
+		 cc_newloads,
+		 cc_searches - cc_hits - cc_neg_hits - cc_newloads,
+		 cc_searches - cc_hits - cc_neg_hits,
+		 cc_invals,
+		 cc_discards);
+}
+
+#endif /* CATCACHE_STATS */
+
+
 /*
  *		CatCacheRemoveCTup
+ *
+ * Unlink and delete the given cache entry
  */
 static void
 CatCacheRemoveCTup(CatCache *cache, CatCTup *ct)
@@ -473,16 +329,26 @@ CatCacheRemoveCTup(CatCache *cache, CatCTup *ct)
 }
 
 /*
- *	CatalogCacheIdInvalidate()
+ *	CatalogCacheIdInvalidate
+ *
+ *	Invalidate entries in the specified cache, given a hash value and
+ *	item pointer.  Positive entries are deleted if they match the item
+ *	pointer.  Negative entries must be deleted if they match the hash
+ *	value (since we do not have the exact key of the tuple that's being
+ *	inserted).  But this should only rarely result in loss of a cache
+ *	entry that could have been kept.
  *
- *	Invalidate a tuple given a cache id.  In this case the id should always
- *	be found (whether the cache has opened its relation or not).  Of course,
- *	if the cache has yet to open its relation, there will be no tuples so
- *	no problem.
+ *	Note that it's not very relevant whether the tuple identified by
+ *	the item pointer is being inserted or deleted.  We don't expect to
+ *	find matching positive entries in the one case, and we don't expect
+ *	to find matching negative entries in the other; but we will do the
+ *	right things in any case.
+ *
+ *	This routine is only quasi-public: it should only be used by inval.c.
  */
 void
 CatalogCacheIdInvalidate(int cacheId,
-						 Index hashIndex,
+						 uint32 hashValue,
 						 ItemPointer pointer)
 {
 	CatCache   *ccp;
@@ -491,37 +357,51 @@ CatalogCacheIdInvalidate(int cacheId,
 	 * sanity checks
 	 */
 	Assert(ItemPointerIsValid(pointer));
-	CACHE1_elog(LOG, "CatalogCacheIdInvalidate: called");
+	CACHE1_elog(DEBUG1, "CatalogCacheIdInvalidate: called");
 
 	/*
 	 * inspect caches to find the proper cache
 	 */
 	for (ccp = CacheHdr->ch_caches; ccp; ccp = ccp->cc_next)
 	{
+		Index		hashIndex;
 		Dlelem	   *elt,
 				   *nextelt;
 
 		if (cacheId != ccp->id)
 			continue;
 
-		Assert(hashIndex < ccp->cc_size);
+		/*
+		 * We don't bother to check whether the cache has finished
+		 * initialization yet; if not, there will be no entries in it
+		 * so no problem.
+		 */
 
 		/*
-		 * inspect the hash bucket until we find a match or exhaust
+		 * inspect the proper hash bucket for matches
 		 */
+		hashIndex = (Index) (hashValue % (uint32) ccp->cc_size);
+
 		for (elt = DLGetHead(&ccp->cc_bucket[hashIndex]); elt; elt = nextelt)
 		{
 			CatCTup    *ct = (CatCTup *) DLE_VAL(elt);
 
 			nextelt = DLGetSucc(elt);
 
-			if (ItemPointerEquals(pointer, &ct->tuple.t_self))
+			if (hashValue != ct->hash_value)
+				continue;		/* ignore non-matching hash values */
+
+			if (ct->negative ||
+				ItemPointerEquals(pointer, &ct->tuple.t_self))
 			{
 				if (ct->refcount > 0)
 					ct->dead = true;
 				else
 					CatCacheRemoveCTup(ccp, ct);
-				CACHE1_elog(LOG, "CatalogCacheIdInvalidate: invalidated");
+				CACHE1_elog(DEBUG1, "CatalogCacheIdInvalidate: invalidated");
+#ifdef CATCACHE_STATS
+				ccp->cc_invals++;
+#endif
 				/* could be multiple matches, so keep looking! */
 			}
 		}
@@ -531,18 +411,34 @@ CatalogCacheIdInvalidate(int cacheId,
 
 /* ----------------------------------------------------------------
  *					   public functions
- *
- *		AtEOXact_CatCache
- *		ResetCatalogCaches
- *		InitCatCache
- *		SearchCatCache
- *		ReleaseCatCache
- *		RelationInvalidateCatalogCacheTuple
  * ----------------------------------------------------------------
  */
 
 
 /*
+ * Standard routine for creating cache context if it doesn't exist yet
+ *
+ * There are a lot of places (probably far more than necessary) that check
+ * whether CacheMemoryContext exists yet and want to create it if not.
+ * We centralize knowledge of exactly how to create it here.
+ */
+void
+CreateCacheMemoryContext(void)
+{
+	/*
+	 * Purely for paranoia, check that context doesn't exist; caller
+	 * probably did so already.
+	 */
+	if (!CacheMemoryContext)
+		CacheMemoryContext = AllocSetContextCreate(TopMemoryContext,
+												   "CacheMemoryContext",
+												ALLOCSET_DEFAULT_MINSIZE,
+											   ALLOCSET_DEFAULT_INITSIZE,
+											   ALLOCSET_DEFAULT_MAXSIZE);
+}
+
+
+/*
  *		AtEOXact_CatCache
  *
  * Clean up catcaches at end of transaction (either commit or abort)
@@ -609,6 +505,9 @@ ResetCatalogCache(CatCache *cache)
 				ct->dead = true;
 			else
 				CatCacheRemoveCTup(cache, ct);
+#ifdef CATCACHE_STATS
+			cache->cc_invals++;
+#endif
 		}
 	}
 }
@@ -623,12 +522,12 @@ ResetCatalogCaches(void)
 {
 	CatCache   *cache;
 
-	CACHE1_elog(LOG, "ResetCatalogCaches called");
+	CACHE1_elog(DEBUG1, "ResetCatalogCaches called");
 
 	for (cache = CacheHdr->ch_caches; cache; cache = cache->cc_next)
 		ResetCatalogCache(cache);
 
-	CACHE1_elog(LOG, "end of ResetCatalogCaches call");
+	CACHE1_elog(DEBUG1, "end of ResetCatalogCaches call");
 }
 
 /*
@@ -656,7 +555,7 @@ CatalogCacheFlushRelation(Oid relId)
 {
 	CatCache   *cache;
 
-	CACHE2_elog(LOG, "CatalogCacheFlushRelation called for %u", relId);
+	CACHE2_elog(DEBUG1, "CatalogCacheFlushRelation called for %u", relId);
 
 	for (cache = CacheHdr->ch_caches; cache; cache = cache->cc_next)
 	{
@@ -691,6 +590,13 @@ CatalogCacheFlushRelation(Oid relId)
 
 				nextelt = DLGetSucc(elt);
 
+				/*
+				 * Negative entries are never considered related to a rel,
+				 * even if the rel is part of their lookup key.
+				 */
+				if (ct->negative)
+					continue;
+
 				if (cache->cc_reloidattr == ObjectIdAttributeNumber)
 					tupRelid = ct->tuple.t_data->t_oid;
 				else
@@ -711,12 +617,15 @@ CatalogCacheFlushRelation(Oid relId)
 						ct->dead = true;
 					else
 						CatCacheRemoveCTup(cache, ct);
+#ifdef CATCACHE_STATS
+					cache->cc_invals++;
+#endif
 				}
 			}
 		}
 	}
 
-	CACHE1_elog(LOG, "end of CatalogCacheFlushRelation call");
+	CACHE1_elog(DEBUG1, "end of CatalogCacheFlushRelation call");
 }
 
 /*
@@ -730,7 +639,7 @@ CatalogCacheFlushRelation(Oid relId)
 #ifdef CACHEDEBUG
 #define InitCatCache_DEBUG1 \
 do { \
-	elog(LOG, "InitCatCache: rel=%s id=%d nkeys=%d size=%d\n", \
+	elog(DEBUG1, "InitCatCache: rel=%s id=%d nkeys=%d size=%d\n", \
 		cp->cc_relname, cp->id, cp->cc_nkeys, cp->cc_size); \
 } while(0)
 
@@ -791,9 +700,10 @@ InitCatCache(int id,
 	cp->id = id;
 	cp->cc_relname = relname;
 	cp->cc_indname = indname;
-	cp->cc_reloidattr = reloidattr;
+	cp->cc_reloid = InvalidOid;	/* temporary */
 	cp->cc_relisshared = false; /* temporary */
 	cp->cc_tupdesc = (TupleDesc) NULL;
+	cp->cc_reloidattr = reloidattr;
 	cp->cc_ntup = 0;
 	cp->cc_size = NCCBUCKETS;
 	cp->cc_nkeys = nkeys;
@@ -820,6 +730,152 @@ InitCatCache(int id,
 	return cp;
 }
 
+/*
+ *		CatalogCacheInitializeCache
+ *
+ * This function does final initialization of a catcache: obtain the tuple
+ * descriptor and set up the hash and equality function links.	We assume
+ * that the relcache entry can be opened at this point!
+ */
+#ifdef CACHEDEBUG
+#define CatalogCacheInitializeCache_DEBUG1 \
+	elog(DEBUG1, "CatalogCacheInitializeCache: cache @%p %s", cache, \
+		 cache->cc_relname)
+
+#define CatalogCacheInitializeCache_DEBUG2 \
+do { \
+		if (cache->cc_key[i] > 0) { \
+			elog(DEBUG1, "CatalogCacheInitializeCache: load %d/%d w/%d, %u", \
+				i+1, cache->cc_nkeys, cache->cc_key[i], \
+				 tupdesc->attrs[cache->cc_key[i] - 1]->atttypid); \
+		} else { \
+			elog(DEBUG1, "CatalogCacheInitializeCache: load %d/%d w/%d", \
+				i+1, cache->cc_nkeys, cache->cc_key[i]); \
+		} \
+} while(0)
+
+#else
+#define CatalogCacheInitializeCache_DEBUG1
+#define CatalogCacheInitializeCache_DEBUG2
+#endif
+
+static void
+CatalogCacheInitializeCache(CatCache *cache)
+{
+	Relation	relation;
+	MemoryContext oldcxt;
+	TupleDesc	tupdesc;
+	int			i;
+
+	CatalogCacheInitializeCache_DEBUG1;
+
+	/*
+	 * Open the relation without locking --- we only need the tupdesc,
+	 * which we assume will never change ...
+	 */
+	relation = heap_openr(cache->cc_relname, NoLock);
+	Assert(RelationIsValid(relation));
+
+	/*
+	 * switch to the cache context so our allocations do not vanish at the
+	 * end of a transaction
+	 */
+	Assert(CacheMemoryContext != NULL);
+
+	oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
+
+	/*
+	 * copy the relcache's tuple descriptor to permanent cache storage
+	 */
+	tupdesc = CreateTupleDescCopyConstr(RelationGetDescr(relation));
+
+	/*
+	 * get the relation's OID and relisshared flag, too
+	 */
+	cache->cc_reloid = RelationGetRelid(relation);
+	cache->cc_relisshared = RelationGetForm(relation)->relisshared;
+
+	/*
+	 * return to the caller's memory context and close the rel
+	 */
+	MemoryContextSwitchTo(oldcxt);
+
+	heap_close(relation, NoLock);
+
+	CACHE3_elog(DEBUG1, "CatalogCacheInitializeCache: %s, %d keys",
+				cache->cc_relname, cache->cc_nkeys);
+
+	/*
+	 * initialize cache's key information
+	 */
+	for (i = 0; i < cache->cc_nkeys; ++i)
+	{
+		Oid			keytype;
+
+		CatalogCacheInitializeCache_DEBUG2;
+
+		if (cache->cc_key[i] > 0)
+			keytype = tupdesc->attrs[cache->cc_key[i] - 1]->atttypid;
+		else
+		{
+			if (cache->cc_key[i] != ObjectIdAttributeNumber)
+				elog(FATAL, "CatalogCacheInit: only sys attr supported is OID");
+			keytype = OIDOID;
+		}
+
+		cache->cc_hashfunc[i] = GetCCHashFunc(keytype);
+
+		cache->cc_isname[i] = (keytype == NAMEOID);
+
+		/*
+		 * If GetCCHashFunc liked the type, safe to index into eqproc[]
+		 */
+		cache->cc_skey[i].sk_procedure = EQPROC(keytype);
+
+		/* Do function lookup */
+		fmgr_info_cxt(cache->cc_skey[i].sk_procedure,
+					  &cache->cc_skey[i].sk_func,
+					  CacheMemoryContext);
+
+		/* Initialize sk_attno suitably for HeapKeyTest() and heap scans */
+		cache->cc_skey[i].sk_attno = cache->cc_key[i];
+
+		CACHE4_elog(DEBUG1, "CatalogCacheInit %s %d %p",
+					cache->cc_relname,
+					i,
+					cache);
+	}
+
+	/*
+	 * mark this cache fully initialized
+	 */
+	cache->cc_tupdesc = tupdesc;
+}
+
+/*
+ * InitCatCachePhase2 -- external interface for CatalogCacheInitializeCache
+ *
+ * The only reason to call this routine is to ensure that the relcache
+ * has created entries for all the catalogs and indexes referenced by
+ * catcaches.  Therefore, open the index too.  An exception is the indexes
+ * on pg_am, which we don't use (cf. IndexScanOK).
+ */
+void
+InitCatCachePhase2(CatCache *cache)
+{
+	if (cache->cc_tupdesc == NULL)
+		CatalogCacheInitializeCache(cache);
+
+	if (cache->id != AMOID &&
+		cache->id != AMNAME)
+	{
+		Relation	idesc;
+
+		idesc = index_openr(cache->cc_indname);
+		index_close(idesc);
+	}
+}
+
 
 /*
  *		IndexScanOK
@@ -874,10 +930,20 @@ IndexScanOK(CatCache *cache, ScanKey cur_skey)
 }
 
 /*
- *		SearchCatCache
+ *	SearchCatCache
  *
  *		This call searches a system cache for a tuple, opening the relation
- *		if necessary (the first access to a particular cache).
+ *		if necessary (on the first access to a particular cache).
+ *
+ *		The result is NULL if not found, or a pointer to a HeapTuple in
+ *		the cache.  The caller must not modify the tuple, and must call
+ *		ReleaseCatCache() when done with it.
+ *
+ * The search key values should be expressed as Datums of the key columns'
+ * datatype(s).  (Pass zeroes for any unused parameters.)  As a special
+ * exception, the passed-in key for a NAME column can be just a C string;
+ * the caller need not go to the trouble of converting it to a fully
+ * null-padded NAME.
  */
 HeapTuple
 SearchCatCache(CatCache *cache,
@@ -887,11 +953,13 @@ SearchCatCache(CatCache *cache,
 			   Datum v4)
 {
 	ScanKeyData cur_skey[4];
-	Index		hash;
+	uint32		hashValue;
+	Index		hashIndex;
 	Dlelem	   *elt;
 	CatCTup    *ct;
-	HeapTuple	ntp;
 	Relation	relation;
+	HeapTuple	ntp;
+	int			i;
 	MemoryContext oldcxt;
 
 	/*
@@ -916,12 +984,13 @@ SearchCatCache(CatCache *cache,
 	/*
 	 * find the hash bucket in which to look for the tuple
 	 */
-	hash = CatalogCacheComputeHashIndex(cache, cur_skey);
+	hashValue = CatalogCacheComputeHashValue(cache, cur_skey);
+	hashIndex = (Index) (hashValue % (uint32) cache->cc_size);
 
 	/*
 	 * scan the hash bucket until we find a match or exhaust our tuples
 	 */
-	for (elt = DLGetHead(&cache->cc_bucket[hash]);
+	for (elt = DLGetHead(&cache->cc_bucket[hashIndex]);
 		 elt;
 		 elt = DLGetSucc(elt))
 	{
@@ -932,9 +1001,11 @@ SearchCatCache(CatCache *cache,
 		if (ct->dead)
 			continue;			/* ignore dead entries */
 
+		if (ct->hash_value != hashValue)
+			continue;			/* quickly skip entry if wrong hash val */
+
 		/*
-		 * see if the cached tuple matches our key. (should we be worried
-		 * about time ranges? -cim 10/2/90)
+		 * see if the cached tuple matches our key.
 		 */
 		HeapKeyTest(&ct->tuple,
 					cache->cc_tupdesc,
@@ -945,33 +1016,53 @@ SearchCatCache(CatCache *cache,
 			continue;
 
 		/*
-		 * we found a tuple in the cache: bump its refcount, move it to
-		 * the front of the LRU list, and return it.  We also move it to
-		 * the front of the list for its hashbucket, in order to speed
-		 * subsequent searches.  (The most frequently accessed elements in
-		 * any hashbucket will tend to be near the front of the
-		 * hashbucket's list.)
+		 * we found a match in the cache: move it to the front of the global
+		 * LRU list.  We also move it to the front of the list for its
+		 * hashbucket, in order to speed subsequent searches.  (The most
+		 * frequently accessed elements in any hashbucket will tend to be
+		 * near the front of the hashbucket's list.)
 		 */
-		ct->refcount++;
-
 		DLMoveToFront(&ct->lrulist_elem);
 		DLMoveToFront(&ct->cache_elem);
 
+		/*
+		 * If it's a positive entry, bump its refcount and return it.
+		 * If it's negative, we can report failure to the caller.
+		 */
+		if (!ct->negative)
+		{
+			ct->refcount++;
+
 #ifdef CACHEDEBUG
-		CACHE3_elog(LOG, "SearchCatCache(%s): found in bucket %d",
-					cache->cc_relname, hash);
+			CACHE3_elog(DEBUG1, "SearchCatCache(%s): found in bucket %d",
+						cache->cc_relname, hashIndex);
 #endif   /* CACHEDEBUG */
 
 #ifdef CATCACHE_STATS
-		cache->cc_hits++;
+			cache->cc_hits++;
 #endif
 
-		return &ct->tuple;
+			return &ct->tuple;
+		}
+		else
+		{
+#ifdef CACHEDEBUG
+			CACHE3_elog(DEBUG1, "SearchCatCache(%s): found neg entry in bucket %d",
+						cache->cc_relname, hashIndex);
+#endif   /* CACHEDEBUG */
+
+#ifdef CATCACHE_STATS
+			cache->cc_neg_hits++;
+#endif
+
+			return NULL;
+		}
 	}
 
 	/*
-	 * Tuple was not found in cache, so we have to try and retrieve it
-	 * directly from the relation.	If it's found, we add it to the cache.
+	 * Tuple was not found in cache, so we have to try to retrieve it
+	 * directly from the relation.  If found, we will add it to the
+	 * cache; if not found, we will add a negative cache entry instead.
 	 *
 	 * NOTE: it is possible for recursive cache lookups to occur while
 	 * reading the relation --- for example, due to shared-cache-inval
@@ -987,14 +1078,18 @@ SearchCatCache(CatCache *cache,
 	/*
 	 * open the relation associated with the cache
 	 */
-	relation = heap_openr(cache->cc_relname, AccessShareLock);
+	relation = heap_open(cache->cc_reloid, AccessShareLock);
+
+	/*
+	 * Pre-create cache entry header, and mark no tuple found.
+	 */
+	ct = (CatCTup *) MemoryContextAlloc(CacheMemoryContext, sizeof(CatCTup));
+	ct->negative = true;
 
 	/*
 	 * Scan the relation to find the tuple.  If there's an index, and if
 	 * it's safe to do so, use the index.  Else do a heap scan.
 	 */
-	ct = NULL;
-
 	if ((RelationGetForm(relation))->relhasindex &&
 		!IsIgnoringSystemIndexes() &&
 		IndexScanOK(cache, cur_skey))
@@ -1004,9 +1099,8 @@ SearchCatCache(CatCache *cache,
 		RetrieveIndexResult indexRes;
 		HeapTupleData tuple;
 		Buffer		buffer;
-		int			i;
 
-		CACHE2_elog(LOG, "SearchCatCache(%s): performing index scan",
+		CACHE2_elog(DEBUG1, "SearchCatCache(%s): performing index scan",
 					cache->cc_relname);
 
 		/*
@@ -1031,8 +1125,8 @@ SearchCatCache(CatCache *cache,
 			{
 				/* Copy tuple into our context */
 				oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
-				ct = (CatCTup *) palloc(sizeof(CatCTup));
 				heap_copytuple_with_tuple(&tuple, &ct->tuple);
+				ct->negative = false;
 				MemoryContextSwitchTo(oldcxt);
 				ReleaseBuffer(buffer);
 				break;
@@ -1045,7 +1139,7 @@ SearchCatCache(CatCache *cache,
 	{
 		HeapScanDesc sd;
 
-		CACHE2_elog(LOG, "SearchCatCache(%s): performing heap scan",
+		CACHE2_elog(DEBUG1, "SearchCatCache(%s): performing heap scan",
 					cache->cc_relname);
 
 		sd = heap_beginscan(relation, 0, SnapshotNow,
@@ -1057,8 +1151,8 @@ SearchCatCache(CatCache *cache,
 		{
 			/* Copy tuple into our context */
 			oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
-			ct = (CatCTup *) palloc(sizeof(CatCTup));
 			heap_copytuple_with_tuple(ntp, &ct->tuple);
+			ct->negative = false;
 			MemoryContextSwitchTo(oldcxt);
 			/* We should not free the result of heap_getnext... */
 		}
@@ -1072,58 +1166,140 @@ SearchCatCache(CatCache *cache,
 	heap_close(relation, AccessShareLock);
 
 	/*
-	 * scan is complete.  if tup was found, we can add it to the cache.
+	 * scan is complete.  If tuple was not found, we need to build
+	 * a fake tuple for the negative cache entry.  The fake tuple has
+	 * the correct key columns, but nulls everywhere else.
 	 */
-	if (ct == NULL)
-		return NULL;
+	if (ct->negative)
+	{
+		TupleDesc	tupDesc = cache->cc_tupdesc;
+		Datum	   *values;
+		char	   *nulls;
+		Oid			negOid = InvalidOid;
+
+		values = (Datum *) palloc(tupDesc->natts * sizeof(Datum));
+		nulls = (char *) palloc(tupDesc->natts * sizeof(char));
+
+		memset(values, 0, tupDesc->natts * sizeof(Datum));
+		memset(nulls, 'n', tupDesc->natts * sizeof(char));
+
+		for (i = 0; i < cache->cc_nkeys; i++)
+		{
+			int		attindex = cache->cc_key[i];
+			Datum	keyval = cur_skey[i].sk_argument;
+
+			if (attindex > 0)
+			{
+				/*
+				 * Here we must be careful in case the caller passed a
+				 * C string where a NAME is wanted: convert the given
+				 * argument to a correctly padded NAME.  Otherwise the
+				 * memcpy() done in heap_formtuple could fall off the
+				 * end of memory.
+				 */
+				if (cache->cc_isname[i])
+				{
+					Name	newval = (Name) palloc(NAMEDATALEN);
+
+					namestrcpy(newval, DatumGetCString(keyval));
+					keyval = NameGetDatum(newval);
+				}
+				values[attindex-1] = keyval;
+				nulls[attindex-1] = ' ';
+			}
+			else
+			{
+				Assert(attindex == ObjectIdAttributeNumber);
+				negOid = DatumGetObjectId(keyval);
+			}
+		}
+
+		ntp = heap_formtuple(tupDesc, values, nulls);
+
+		oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
+		heap_copytuple_with_tuple(ntp, &ct->tuple);
+		ct->tuple.t_data->t_oid = negOid;
+		MemoryContextSwitchTo(oldcxt);
+
+		heap_freetuple(ntp);
+		for (i = 0; i < cache->cc_nkeys; i++)
+		{
+			if (cache->cc_isname[i])
+				pfree(DatumGetName(values[cache->cc_key[i]-1]));
+		}
+		pfree(values);
+		pfree(nulls);
+	}
 
 	/*
 	 * Finish initializing the CatCTup header, and add it to the linked
 	 * lists.
 	 */
-	CACHE1_elog(LOG, "SearchCatCache: found tuple");
-
 	ct->ct_magic = CT_MAGIC;
 	ct->my_cache = cache;
 	DLInitElem(&ct->lrulist_elem, (void *) ct);
 	DLInitElem(&ct->cache_elem, (void *) ct);
 	ct->refcount = 1;			/* count this first reference */
 	ct->dead = false;
+	ct->hash_value = hashValue;
 
 	DLAddHead(&CacheHdr->ch_lrulist, &ct->lrulist_elem);
-	DLAddHead(&cache->cc_bucket[hash], &ct->cache_elem);
-
-#ifdef CATCACHE_STATS
-	cache->cc_newloads++;
-#endif
+	DLAddHead(&cache->cc_bucket[hashIndex], &ct->cache_elem);
 
 	/*
 	 * If we've exceeded the desired size of the caches, try to throw away
-	 * the least recently used entry.
+	 * the least recently used entry.  NB: the newly-built entry cannot
+	 * get thrown away here, because it has positive refcount.
 	 */
 	++cache->cc_ntup;
 	if (++CacheHdr->ch_ntup > CacheHdr->ch_maxtup)
 	{
-		for (elt = DLGetTail(&CacheHdr->ch_lrulist);
-			 elt;
-			 elt = DLGetPred(elt))
+		Dlelem	   *prevelt;
+
+		for (elt = DLGetTail(&CacheHdr->ch_lrulist); elt; elt = prevelt)
 		{
 			CatCTup    *oldct = (CatCTup *) DLE_VAL(elt);
 
+			prevelt = DLGetPred(elt);
+
 			if (oldct->refcount == 0)
 			{
-				CACHE2_elog(LOG, "SearchCatCache(%s): Overflow, LRU removal",
+				CACHE2_elog(DEBUG1, "SearchCatCache(%s): Overflow, LRU removal",
 							cache->cc_relname);
+#ifdef CATCACHE_STATS
+				oldct->my_cache->cc_discards++;
+#endif
 				CatCacheRemoveCTup(oldct->my_cache, oldct);
-				break;
+				if (CacheHdr->ch_ntup <= CacheHdr->ch_maxtup)
+					break;
 			}
 		}
 	}
 
-	CACHE4_elog(LOG, "SearchCatCache(%s): Contains %d/%d tuples",
+	CACHE4_elog(DEBUG1, "SearchCatCache(%s): Contains %d/%d tuples",
 				cache->cc_relname, cache->cc_ntup, CacheHdr->ch_ntup);
-	CACHE3_elog(LOG, "SearchCatCache(%s): put in bucket %d",
-				cache->cc_relname, hash);
+
+	if (ct->negative)
+	{
+		CACHE3_elog(DEBUG1, "SearchCatCache(%s): put neg entry in bucket %d",
+					cache->cc_relname, hashIndex);
+
+		/*
+		 * We are not returning the new entry to the caller, so reset its
+		 * refcount.  Note it would be uncool to set the refcount to 0
+		 * before doing the extra-entry removal step above.
+		 */
+		ct->refcount = 0;		/* negative entries never have refs */
+
+		return NULL;
+	}
+
+	CACHE3_elog(DEBUG1, "SearchCatCache(%s): put in bucket %d",
+				cache->cc_relname, hashIndex);
+
+#ifdef CATCACHE_STATS
+	cache->cc_newloads++;
+#endif
 
 	return &ct->tuple;
 }
@@ -1164,7 +1340,7 @@ ReleaseCatCache(HeapTuple tuple)
  *
  *	This is part of a rather subtle chain of events, so pay attention:
  *
- *	When a tuple is updated or deleted, it cannot be flushed from the
+ *	When a tuple is inserted or deleted, it cannot be flushed from the
  *	catcaches immediately, for reasons explained at the top of cache/inval.c.
  *	Instead we have to add entry(s) for the tuple to a list of pending tuple
  *	invalidations that will be done at the end of the command or transaction.
@@ -1172,15 +1348,16 @@ ReleaseCatCache(HeapTuple tuple)
  *	The lists of tuples that need to be flushed are kept by inval.c.  This
  *	routine is a helper routine for inval.c.  Given a tuple belonging to
  *	the specified relation, find all catcaches it could be in, compute the
- *	correct hashindex for each such catcache, and call the specified function
- *	to record the cache id, hashindex, and tuple ItemPointer in inval.c's
+ *	correct hash value for each such catcache, and call the specified function
+ *	to record the cache id, hash value, and tuple ItemPointer in inval.c's
  *	lists.	CatalogCacheIdInvalidate will be called later, if appropriate,
  *	using the recorded information.
  *
  *	Note that it is irrelevant whether the given tuple is actually loaded
  *	into the catcache at the moment.  Even if it's not there now, it might
- *	be by the end of the command --- or might be in other backends' caches
- * --- so we have to be prepared to flush it.
+ *	be by the end of the command, or there might be a matching negative entry
+ *	to flush --- or other backends' caches might have such entries --- so
+ *	we have to make list entries to flush it later.
  *
  *	Also note that it's not an error if there are no catcaches for the
  *	specified relation.  inval.c doesn't know exactly which rels have
@@ -1190,11 +1367,12 @@ ReleaseCatCache(HeapTuple tuple)
 void
 PrepareToInvalidateCacheTuple(Relation relation,
 							  HeapTuple tuple,
-						 void (*function) (int, Index, ItemPointer, Oid))
+						 void (*function) (int, uint32, ItemPointer, Oid))
 {
 	CatCache   *ccp;
+	Oid			reloid;
 
-	CACHE1_elog(LOG, "PrepareToInvalidateCacheTuple: called");
+	CACHE1_elog(DEBUG1, "PrepareToInvalidateCacheTuple: called");
 
 	/*
 	 * sanity checks
@@ -1204,25 +1382,27 @@ PrepareToInvalidateCacheTuple(Relation relation,
 	Assert(PointerIsValid(function));
 	Assert(CacheHdr != NULL);
 
+	reloid = RelationGetRelid(relation);
+
 	/* ----------------
 	 *	for each cache
 	 *	   if the cache contains tuples from the specified relation
-	 *		   compute the tuple's hash index in this cache,
+	 *		   compute the tuple's hash value in this cache,
 	 *		   and call the passed function to register the information.
 	 * ----------------
 	 */
 
 	for (ccp = CacheHdr->ch_caches; ccp; ccp = ccp->cc_next)
 	{
-		if (strcmp(ccp->cc_relname, RelationGetRelationName(relation)) != 0)
-			continue;
-
 		/* Just in case cache hasn't finished initialization yet... */
 		if (ccp->cc_tupdesc == NULL)
 			CatalogCacheInitializeCache(ccp);
 
+		if (ccp->cc_reloid != reloid)
+			continue;
+
 		(*function) (ccp->id,
-					 CatalogCacheComputeTupleHashIndex(ccp, tuple),
+					 CatalogCacheComputeTupleHashValue(ccp, tuple),
 					 &tuple->t_self,
 					 ccp->cc_relisshared ? (Oid) 0 : MyDatabaseId);
 	}
diff --git a/src/backend/utils/cache/inval.c b/src/backend/utils/cache/inval.c
index 0e383b32c8..6d397a7ba9 100644
--- a/src/backend/utils/cache/inval.c
+++ b/src/backend/utils/cache/inval.c
@@ -5,26 +5,38 @@
  *
  *	This is subtle stuff, so pay attention:
  *
- *	When a tuple is updated or deleted, our time qualification rules consider
- *	that it is *still valid* so long as we are in the same command, ie,
- *	until the next CommandCounterIncrement() or transaction commit.
- *	(See utils/time/tqual.c.)  At the command boundary, the old tuple stops
+ *	When a tuple is updated or deleted, our standard time qualification rules
+ *	consider that it is *still valid* so long as we are in the same command,
+ *	ie, until the next CommandCounterIncrement() or transaction commit.
+ *	(See utils/time/tqual.c, and note that system catalogs are generally
+ *	scanned under SnapshotNow rules by the system, or plain user snapshots
+ *	for user queries.)  At the command boundary, the old tuple stops
  *	being valid and the new version, if any, becomes valid.  Therefore,
  *	we cannot simply flush a tuple from the system caches during heap_update()
  *	or heap_delete().  The tuple is still good at that point; what's more,
  *	even if we did flush it, it might be reloaded into the caches by a later
  *	request in the same command.  So the correct behavior is to keep a list
  *	of outdated (updated/deleted) tuples and then do the required cache
- *	flushes at the next command boundary.  Similarly, we need a list of
- *	inserted tuples (including new versions of updated tuples), which we will
- *	use to flush those tuples out of the caches if we abort the transaction.
- *	Notice that the first list lives only till command boundary, whereas the
- *	second lives till end of transaction.  Finally, we need a third list of
- *	all tuples outdated in the current transaction; if we commit, we send
- *	those invalidation events to all other backends (via the SI message queue)
- *	so that they can flush obsolete entries from their caches.	This list
- *	definitely can't be processed until after we commit, otherwise the other
- *	backends won't see our updated tuples as good.
+ *	flushes at the next command boundary.  We must also keep track of
+ *	inserted tuples so that we can flush "negative" cache entries that match
+ *	the new tuples; again, that mustn't happen until end of command.
+ *
+ *	Once we have finished the command, we still need to remember inserted
+ *	tuples (including new versions of updated tuples), so that we can flush
+ *	them from the caches if we abort the transaction.  Similarly, we'd better
+ *	be able to flush "negative" cache entries that may have been loaded in
+ *	place of deleted tuples, so we still need the deleted ones too.
+ *
+ *	If we successfully complete the transaction, we have to broadcast all
+ *	these invalidation events to other backends (via the SI message queue)
+ *	so that they can flush obsolete entries from their caches.  Note we have
+ *	to record the transaction commit before sending SI messages, otherwise
+ *	the other backends won't see our updated tuples as good.
+ *
+ *	In short, we need to remember until xact end every insert or delete
+ *	of a tuple that might be in the system caches.  Updates are treated as
+ *	two events, delete + insert, for simplicity.  (There are cases where
+ *	it'd be possible to record just one event, but we don't currently try.)
  *
  *	We do not need to register EVERY tuple operation in this way, just those
  *	on tuples in relations that have associated catcaches.	We do, however,
@@ -62,7 +74,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/utils/cache/inval.c,v 1.48 2002/02/19 20:11:17 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/utils/cache/inval.c,v 1.49 2002/03/03 17:47:55 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -99,33 +111,26 @@ typedef struct InvalidationListHeader
 
 /*
  * ----------------
- *	Invalidation info is divided into three parts.
- *	1) shared invalidation to be sent to all backends at commit
- *	2) local invalidation for the transaction itself (actually, just
- *	   for the current command within the transaction)
- *	3) rollback information for the transaction itself (in case we abort)
+ *	Invalidation info is divided into two lists:
+ *	1) events so far in current command, not yet reflected to caches.
+ *	2) events in previous commands of current transaction; these have
+ *	   been reflected to local caches, and must be either broadcast to
+ *	   other backends or rolled back from local cache when we commit
+ *	   or abort the transaction.
+ *
+ * The relcache-file-invalidated flag can just be a simple boolean,
+ * since we only act on it at transaction commit; we don't care which
+ * command of the transaction set it.
  * ----------------
  */
 
-/*
- * head of invalidation message list for all backends
- * eaten by AtCommit_Cache() in CommitTransaction()
- */
-static InvalidationListHeader GlobalInvalidMsgs;
+/* head of current-command event list */
+static InvalidationListHeader CurrentCmdInvalidMsgs;
 
-static bool RelcacheInitFileInval; /* init file must be invalidated? */
+/* head of previous-commands event list */
+static InvalidationListHeader PriorCmdInvalidMsgs;
 
-/*
- * head of invalidation message list for the current command
- * eaten by AtCommit_LocalCache() in CommandCounterIncrement()
- */
-static InvalidationListHeader LocalInvalidMsgs;
-
-/*
- * head of rollback message list for abort-time processing
- * eaten by AtAbort_Cache() in AbortTransaction()
- */
-static InvalidationListHeader RollbackMsgs;
+static bool RelcacheInitFileInval; /* init file must be invalidated? */
 
 
 /* ----------------------------------------------------------------
@@ -205,6 +210,29 @@ FreeInvalidationMessageList(InvalidationChunk **listHdr)
 }
 
 /*
+ * Append one list of invalidation message chunks to another, resetting
+ * the source chunk-list pointer to NULL.
+ */
+static void
+AppendInvalidationMessageList(InvalidationChunk **destHdr,
+							  InvalidationChunk **srcHdr)
+{
+	InvalidationChunk *chunk = *srcHdr;
+
+	if (chunk == NULL)
+		return;					/* nothing to do */
+
+	while (chunk->next != NULL)
+		chunk = chunk->next;
+
+	chunk->next = *destHdr;
+
+	*destHdr = *srcHdr;
+
+	*srcHdr = NULL;
+}
+
+/*
  * Process a list of invalidation messages.
  *
  * This is a macro that executes the given code fragment for each message in
@@ -238,15 +266,15 @@ FreeInvalidationMessageList(InvalidationChunk **listHdr)
  */
 static void
 AddCatcacheInvalidationMessage(InvalidationListHeader *hdr,
-							   int id, Index hashIndex,
+							   int id, uint32 hashValue,
 							   ItemPointer tuplePtr, Oid dbId)
 {
 	SharedInvalidationMessage msg;
 
 	msg.cc.id = (int16) id;
-	msg.cc.hashIndex = (uint16) hashIndex;
-	msg.cc.dbId = dbId;
 	msg.cc.tuplePtr = *tuplePtr;
+	msg.cc.dbId = dbId;
+	msg.cc.hashValue = hashValue;
 	AddInvalidationMessage(&hdr->cclist, &msg);
 }
 
@@ -272,6 +300,18 @@ AddRelcacheInvalidationMessage(InvalidationListHeader *hdr,
 }
 
 /*
+ * Append one list of invalidation messages to another, resetting
+ * the source list to empty.
+ */
+static void
+AppendInvalidationMessages(InvalidationListHeader *dest,
+						   InvalidationListHeader *src)
+{
+	AppendInvalidationMessageList(&dest->cclist, &src->cclist);
+	AppendInvalidationMessageList(&dest->rclist, &src->rclist);
+}
+
+/*
  * Reset an invalidation list to empty
  *
  * physicalFree may be set false if caller knows transaction is ending
@@ -318,21 +358,16 @@ ProcessInvalidationMessages(InvalidationListHeader *hdr,
 /*
  * RegisterCatcacheInvalidation
  *
- * Register an invalidation event for an updated/deleted catcache entry.
- * We insert the event into both GlobalInvalidMsgs (for transmission
- * to other backends at transaction commit) and LocalInvalidMsgs (for
- * my local invalidation at end of command within xact).
+ * Register an invalidation event for a catcache tuple entry.
  */
 static void
 RegisterCatcacheInvalidation(int cacheId,
-							 Index hashIndex,
+							 uint32 hashValue,
 							 ItemPointer tuplePtr,
 							 Oid dbId)
 {
-	AddCatcacheInvalidationMessage(&GlobalInvalidMsgs,
-								   cacheId, hashIndex, tuplePtr, dbId);
-	AddCatcacheInvalidationMessage(&LocalInvalidMsgs,
-								   cacheId, hashIndex, tuplePtr, dbId);
+	AddCatcacheInvalidationMessage(&CurrentCmdInvalidMsgs,
+								   cacheId, hashValue, tuplePtr, dbId);
 }
 
 /*
@@ -343,11 +378,8 @@ RegisterCatcacheInvalidation(int cacheId,
 static void
 RegisterRelcacheInvalidation(Oid dbId, Oid relId)
 {
-	AddRelcacheInvalidationMessage(&GlobalInvalidMsgs,
+	AddRelcacheInvalidationMessage(&CurrentCmdInvalidMsgs,
 								   dbId, relId);
-	AddRelcacheInvalidationMessage(&LocalInvalidMsgs,
-								   dbId, relId);
-
 	/*
 	 * If the relation being invalidated is one of those cached in the
 	 * relcache init file, mark that we need to zap that file at commit.
@@ -357,34 +389,6 @@ RegisterRelcacheInvalidation(Oid dbId, Oid relId)
 }
 
 /*
- * RegisterCatcacheRollback
- *
- * Register an invalidation event for an inserted catcache entry.
- * This only needs to be flushed out of my local catcache, if I abort.
- */
-static void
-RegisterCatcacheRollback(int cacheId,
-						 Index hashIndex,
-						 ItemPointer tuplePtr,
-						 Oid dbId)
-{
-	AddCatcacheInvalidationMessage(&RollbackMsgs,
-								   cacheId, hashIndex, tuplePtr, dbId);
-}
-
-/*
- * RegisterRelcacheRollback
- *
- * As above, but register a relcache invalidation event.
- */
-static void
-RegisterRelcacheRollback(Oid dbId, Oid relId)
-{
-	AddRelcacheInvalidationMessage(&RollbackMsgs,
-								   dbId, relId);
-}
-
-/*
  * LocalExecuteInvalidationMessage
  *
  * Process a single invalidation message (which could be either type).
@@ -398,7 +402,7 @@ LocalExecuteInvalidationMessage(SharedInvalidationMessage *msg)
 	{
 		if (msg->cc.dbId == MyDatabaseId || msg->cc.dbId == 0)
 			CatalogCacheIdInvalidate(msg->cc.id,
-									 msg->cc.hashIndex,
+									 msg->cc.hashValue,
 									 &msg->cc.tuplePtr);
 	}
 	else if (msg->id == SHAREDINVALRELCACHE_ID)
@@ -438,7 +442,7 @@ InvalidateSystemCaches(void)
  */
 static void
 PrepareForTupleInvalidation(Relation relation, HeapTuple tuple,
-							void (*CacheIdRegisterFunc) (int, Index,
+							void (*CacheIdRegisterFunc) (int, uint32,
 													   ItemPointer, Oid),
 							void (*RelationIdRegisterFunc) (Oid, Oid))
 {
@@ -517,16 +521,18 @@ AcceptInvalidationMessages(void)
  * AtEOXactInvalidationMessages
  *		Process queued-up invalidation messages at end of transaction.
  *
- * If isCommit, we must send out the messages in our GlobalInvalidMsgs list
+ * If isCommit, we must send out the messages in our PriorCmdInvalidMsgs list
  * to the shared invalidation message queue.  Note that these will be read
  * not only by other backends, but also by our own backend at the next
- * transaction start (via AcceptInvalidationMessages).	Therefore, it's okay
- * to discard any pending LocalInvalidMsgs, since these will be redundant
- * with the global list.
+ * transaction start (via AcceptInvalidationMessages).  This means that
+ * we can skip immediate local processing of anything that's still in
+ * CurrentCmdInvalidMsgs, and just send that list out too.
  *
  * If not isCommit, we are aborting, and must locally process the messages
- * in our RollbackMsgs list.  No messages need be sent to other backends,
- * since they'll not have seen our changed tuples anyway.
+ * in PriorCmdInvalidMsgs.  No messages need be sent to other backends,
+ * since they'll not have seen our changed tuples anyway.  We can forget
+ * about CurrentCmdInvalidMsgs too, since those changes haven't touched
+ * the caches yet.
  *
  * In any case, reset the various lists to empty.  We need not physically
  * free memory here, since TopTransactionContext is about to be emptied
@@ -548,7 +554,10 @@ AtEOXactInvalidationMessages(bool isCommit)
 		if (RelcacheInitFileInval)
 			RelationCacheInitFileInvalidate(true);
 
-		ProcessInvalidationMessages(&GlobalInvalidMsgs,
+		AppendInvalidationMessages(&PriorCmdInvalidMsgs,
+								   &CurrentCmdInvalidMsgs);
+
+		ProcessInvalidationMessages(&PriorCmdInvalidMsgs,
 									SendSharedInvalidMessage);
 
 		if (RelcacheInitFileInval)
@@ -556,15 +565,14 @@ AtEOXactInvalidationMessages(bool isCommit)
 	}
 	else
 	{
-		ProcessInvalidationMessages(&RollbackMsgs,
+		ProcessInvalidationMessages(&PriorCmdInvalidMsgs,
 									LocalExecuteInvalidationMessage);
 	}
 
 	RelcacheInitFileInval = false;
 
-	DiscardInvalidationMessages(&GlobalInvalidMsgs, false);
-	DiscardInvalidationMessages(&LocalInvalidMsgs, false);
-	DiscardInvalidationMessages(&RollbackMsgs, false);
+	DiscardInvalidationMessages(&PriorCmdInvalidMsgs, false);
+	DiscardInvalidationMessages(&CurrentCmdInvalidMsgs, false);
 }
 
 /*
@@ -573,13 +581,13 @@ AtEOXactInvalidationMessages(bool isCommit)
  *		in a transaction.
  *
  * Here, we send no messages to the shared queue, since we don't know yet if
- * we will commit.	But we do need to locally process the LocalInvalidMsgs
- * list, so as to flush our caches of any tuples we have outdated in the
- * current command.
+ * we will commit.	We do need to locally process the CurrentCmdInvalidMsgs
+ * list, so as to flush our caches of any entries we have outdated in the
+ * current command.  We then move the current-cmd list over to become part
+ * of the prior-cmds list.
  *
  * The isCommit = false case is not currently used, but may someday be
  * needed to support rollback to a savepoint within a transaction.
- * (I suspect it needs more work first --- tgl.)
  *
  * Note:
  *		This should be called during CommandCounterIncrement(),
@@ -590,29 +598,24 @@ CommandEndInvalidationMessages(bool isCommit)
 {
 	if (isCommit)
 	{
-		ProcessInvalidationMessages(&LocalInvalidMsgs,
+		ProcessInvalidationMessages(&CurrentCmdInvalidMsgs,
 									LocalExecuteInvalidationMessage);
+		AppendInvalidationMessages(&PriorCmdInvalidMsgs,
+								   &CurrentCmdInvalidMsgs);
 	}
 	else
 	{
-		ProcessInvalidationMessages(&RollbackMsgs,
-									LocalExecuteInvalidationMessage);
+		/* XXX what needs to be done here? */
 	}
-
-	/*
-	 * LocalInvalidMsgs list is not interesting anymore, so flush it (for
-	 * real).  Do *not* clear GlobalInvalidMsgs or RollbackMsgs.
-	 */
-	DiscardInvalidationMessages(&LocalInvalidMsgs, true);
 }
 
 /*
- * RelationInvalidateHeapTuple
+ * CacheInvalidateHeapTuple
  *		Register the given tuple for invalidation at end of command
  *		(ie, current command is outdating this tuple).
  */
 void
-RelationInvalidateHeapTuple(Relation relation, HeapTuple tuple)
+CacheInvalidateHeapTuple(Relation relation, HeapTuple tuple)
 {
 	PrepareForTupleInvalidation(relation, tuple,
 								RegisterCatcacheInvalidation,
@@ -620,14 +623,17 @@ RelationInvalidateHeapTuple(Relation relation, HeapTuple tuple)
 }
 
 /*
- * RelationMark4RollbackHeapTuple
- *		Register the given tuple for invalidation in case of abort
- *		(ie, current command is creating this tuple).
+ * CacheInvalidateRelcache
+ *		Register invalidation of the specified relation's relcache entry
+ *		at end of command.
+ *
+ * This is used in places that need to force relcache rebuild but aren't
+ * changing any of the tuples recognized as contributors to the relcache
+ * entry by PrepareForTupleInvalidation.  (An example is dropping an index.)
  */
 void
-RelationMark4RollbackHeapTuple(Relation relation, HeapTuple tuple)
+CacheInvalidateRelcache(Oid relationId)
 {
-	PrepareForTupleInvalidation(relation, tuple,
-								RegisterCatcacheRollback,
-								RegisterRelcacheRollback);
+	/* See KLUGE ALERT in PrepareForTupleInvalidation */
+	RegisterRelcacheInvalidation(MyDatabaseId, relationId);
 }
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index cd5162ac81..e3b0c69372 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/utils/cache/relcache.c,v 1.152 2002/02/19 20:11:17 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/utils/cache/relcache.c,v 1.153 2002/03/03 17:47:55 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -2065,10 +2065,16 @@ RelationBuildLocalRelation(const char *relname,
 		rel->rd_isnailed = true;
 
 	/*
-	 * create a new tuple descriptor from the one passed in (we do this to
-	 * copy it into the cache context)
+	 * create a new tuple descriptor from the one passed in.  We do this
+	 * partly to copy it into the cache context, and partly because the
+	 * new relation can't have any defaults or constraints yet; they
+	 * have to be added in later steps, because they require additions
+	 * to multiple system catalogs.  We can copy attnotnull constraints
+	 * here, however.
 	 */
-	rel->rd_att = CreateTupleDescCopyConstr(tupDesc);
+	rel->rd_att = CreateTupleDescCopy(tupDesc);
+	for (i = 0; i < natts; i++)
+		rel->rd_att->attrs[i]->attnotnull = tupDesc->attrs[i]->attnotnull;
 
 	/*
 	 * initialize relation tuple form (caller may add/override data later)
@@ -2082,8 +2088,6 @@ RelationBuildLocalRelation(const char *relname,
 	rel->rd_rel->relhasoids = true;
 	rel->rd_rel->relnatts = natts;
 	rel->rd_rel->reltype = InvalidOid;
-	if (tupDesc->constr)
-		rel->rd_rel->relchecks = tupDesc->constr->num_check;
 
 	/*
 	 * Insert relation OID and database/tablespace ID into the right