Invent ResourceOwner mechanism as per my recent proposal, and use it to

keep track of portal-related resources separately from transaction-related
resources.  This allows cursors to work in a somewhat sane fashion with
nested transactions.  For now, cursor behavior is non-subtransactional,
that is a cursor's state does not roll back if you abort a subtransaction
that fetched from the cursor.  We might want to change that later.

2 files changed, 95 insertions, 344 deletions

diff --git a/src/backend/utils/cache/catcache.c b/src/backend/utils/cache/catcache.c
index 8bfa3610bd..c382d74974 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.113 2004/07/01 00:51:17 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/cache/catcache.c,v 1.114 2004/07/17 03:29:25 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -31,6 +31,7 @@
 #include "utils/fmgroids.h"
 #include "utils/catcache.h"
 #include "utils/relcache.h"
+#include "utils/resowner.h"
 #include "utils/syscache.h"
 
 
@@ -360,8 +361,6 @@ CatCacheRemoveCTup(CatCache *cache, CatCTup *ct)
 	/* free associated tuple data */
 	if (ct->tuple.t_data != NULL)
 		pfree(ct->tuple.t_data);
-	if (ct->prev_refcount != NULL)
-		pfree(ct->prev_refcount);
 	pfree(ct);
 
 	--cache->cc_ntup;
@@ -396,8 +395,6 @@ CatCacheRemoveCList(CatCache *cache, CatCList *cl)
 	/* free associated tuple data */
 	if (cl->tuple.t_data != NULL)
 		pfree(cl->tuple.t_data);
-	if (cl->prev_refcount != NULL)
-		pfree(cl->prev_refcount);
 	pfree(cl);
 }
 
@@ -531,7 +528,7 @@ CreateCacheMemoryContext(void)
 /*
  *		AtEOXact_CatCache
  *
- * Clean up catcaches at end of transaction (either commit or abort)
+ * Clean up catcaches at end of main transaction (either commit or abort)
  *
  * We scan the caches to reset refcounts to zero.  This is of course
  * necessary in the abort case, since elog() may have interrupted routines.
@@ -564,13 +561,6 @@ AtEOXact_CatCache(bool isCommit)
 				cl->refcount = 0;
 			}
 
-			/*
-			 * Reset the refcount stack.  Drop the item count to zero,
-			 * but don't deallocate the stack itself, so it can be used by
-			 * future subtransactions.
-			 */
-			cl->numpushes = 0;
-
 			/* Clean up any now-deletable dead entries */
 			if (cl->dead)
 				CatCacheRemoveCList(ccp, cl);
@@ -596,13 +586,6 @@ AtEOXact_CatCache(bool isCommit)
 			ct->refcount = 0;
 		}
 
-		/*
-		 * Reset the refcount stack.  Drop the item count to zero,
-		 * but don't deallocate the stack itself, so it can be used by
-		 * future subtransactions.
-		 */
-		ct->numpushes = 0;
-
 		/* Clean up any now-deletable dead entries */
 		if (ct->dead)
 			CatCacheRemoveCTup(ct->my_cache, ct);
@@ -610,161 +593,6 @@ AtEOXact_CatCache(bool isCommit)
 }
 
 /*
- * AtSubStart_CatCache
- *
- * Saves reference counts of each entry at subtransaction start so they
- * can be restored if the subtransaction later aborts.
- */
-void
-AtSubStart_CatCache(void)
-{
-	CatCache   *ccp;
-	Dlelem	   *elt,
-			   *nextelt;
-	MemoryContext old_cxt;
-   
-
-	old_cxt = MemoryContextSwitchTo(CacheMemoryContext);
-
-	/*
-	 * Prepare CLists
-	 */
-	for (ccp = CacheHdr->ch_caches; ccp; ccp = ccp->cc_next)
-	{
-		for (elt = DLGetHead(&ccp->cc_lists); elt; elt = nextelt)
-		{
-			CatCList   *cl = (CatCList *) DLE_VAL(elt);
-
-			nextelt = DLGetSucc(elt);
-
-			if (cl->numpushes == cl->numalloc)
-			{
-				if (cl->numalloc == 0)
-				{
-					cl->numalloc = 8;
-					cl->prev_refcount = palloc(sizeof(int) * cl->numalloc);
-				}
-				else
-				{
-					cl->numalloc *= 2;
-					cl->prev_refcount = repalloc(cl->prev_refcount, cl->numalloc * sizeof(int));
-				}
-			}
-
-			cl->prev_refcount[cl->numpushes++] = cl->refcount;
-		}
-	}
-
-	/*
-	 * Prepare CTuples
-	 */
-	for (elt = DLGetHead(&CacheHdr->ch_lrulist); elt; elt = nextelt)
-	{
-		CatCTup    *ct = (CatCTup *) DLE_VAL(elt);
-
-		nextelt = DLGetSucc(elt);
-
-		if (ct->numpushes == ct->numalloc)
-		{
-			if (ct->numalloc == 0)
-			{
-				ct->numalloc = 8;
-				ct->prev_refcount = palloc(sizeof(int) * ct->numalloc);
-			}
-			else
-			{
-				ct->numalloc *= 2;
-				ct->prev_refcount = repalloc(ct->prev_refcount, sizeof(int) * ct->numalloc);
-			}
-		}
-
-		ct->prev_refcount[ct->numpushes++] = ct->refcount;
-	}
-
-	MemoryContextSwitchTo(old_cxt);
-}
-
-void
-AtEOSubXact_CatCache(bool isCommit)
-{
-	CatCache   *ccp;
-	Dlelem	   *elt,
-			   *nextelt;
-	
-	/*
-	 * Restore CLists
-	 */
-	for (ccp = CacheHdr->ch_caches; ccp; ccp = ccp->cc_next)
-	{
-		for (elt = DLGetHead(&ccp->cc_lists); elt; elt = nextelt)
-		{
-			CatCList   *cl = (CatCList *) DLE_VAL(elt);
-
-			nextelt = DLGetSucc(elt);
-
-			/*
-			 * During commit, check whether the count is what
-			 * we expect.
-			 */
-			if (isCommit)
-			{
-				int expected_refcount;
-				if (cl->numpushes > 0)
-					expected_refcount = cl->prev_refcount[cl->numpushes - 1];
-				else
-					expected_refcount = 0;
-
-				if (cl->refcount != expected_refcount)
-					elog(WARNING, "catcache reference leak");
-			}
-
-			/*
-			 * During abort we have to restore the original count;
-			 * during commit, we have to restore in case of a leak,
-			 * and it won't harm if this is the expected count.
-			 */
-			if (cl->numpushes > 0)
-				cl->refcount = cl->prev_refcount[--cl->numpushes];
-			else
-				cl->refcount = 0;
-		}
-	}
-
-	/*
-	 * Prepare CTuples
-	 */
-	for (elt = DLGetHead(&CacheHdr->ch_lrulist); elt; elt = nextelt)
-	{
-		CatCTup    *ct = (CatCTup *) DLE_VAL(elt);
-
-		nextelt = DLGetSucc(elt);
-
-		if (isCommit)
-		{
-			int expected_refcount;
-
-			if (ct->numpushes > 0)
-				expected_refcount = ct->prev_refcount[ct->numpushes - 1];
-			else
-				expected_refcount = 0;
-
-			if (ct->refcount != expected_refcount)
-				elog(WARNING, "catcache reference leak");
-		}
-
-		/*
-		 * During abort we have to restore the original count;
-		 * during commit, we have to restore in case of a leak,
-		 * and it won't harm if this is the expected count.
-		 */
-		if (ct->numpushes > 0)
-			ct->refcount = ct->prev_refcount[--ct->numpushes];
-		else
-			ct->refcount = 0;
-	}
-}
-
-/*
  *		ResetCatalogCache
  *
  * Reset one catalog cache to empty.
@@ -1334,7 +1162,9 @@ SearchCatCache(CatCache *cache,
 		 */
 		if (!ct->negative)
 		{
+			ResourceOwnerEnlargeCatCacheRefs(CurrentResourceOwner);
 			ct->refcount++;
+			ResourceOwnerRememberCatCacheRef(CurrentResourceOwner, &ct->tuple);
 
 			CACHE3_elog(DEBUG2, "SearchCatCache(%s): found in bucket %d",
 						cache->cc_relname, hashIndex);
@@ -1389,6 +1219,10 @@ SearchCatCache(CatCache *cache,
 		ct = CatalogCacheCreateEntry(cache, ntp,
 									 hashValue, hashIndex,
 									 false);
+		/* immediately set the refcount to 1 */
+		ResourceOwnerEnlargeCatCacheRefs(CurrentResourceOwner);
+		ct->refcount++;
+		ResourceOwnerRememberCatCacheRef(CurrentResourceOwner, &ct->tuple);
 		break;					/* assume only one match */
 	}
 
@@ -1415,10 +1249,9 @@ SearchCatCache(CatCache *cache,
 					cache->cc_relname, hashIndex);
 
 		/*
-		 * We are not returning the new entry to the caller, so reset its
-		 * refcount.
+		 * We are not returning the negative entry to the caller, so leave
+		 * its refcount zero.
 		 */
-		ct->refcount = 0;		/* negative entries never have refs */
 
 		return NULL;
 	}
@@ -1457,6 +1290,7 @@ ReleaseCatCache(HeapTuple tuple)
 	Assert(ct->refcount > 0);
 
 	ct->refcount--;
+	ResourceOwnerForgetCatCacheRef(CurrentResourceOwner, &ct->tuple);
 
 	if (ct->refcount == 0
 #ifndef CATCACHE_FORCE_RELEASE
@@ -1564,7 +1398,10 @@ SearchCatCacheList(CatCache *cache,
 		 * do not move the members to the fronts of their hashbucket
 		 * lists, however, since there's no point in that unless they are
 		 * searched for individually.)	Also bump the members' refcounts.
+		 * (member refcounts are NOT registered separately with the
+		 * resource owner.)
 		 */
+		ResourceOwnerEnlargeCatCacheListRefs(CurrentResourceOwner);
 		for (i = 0; i < cl->n_members; i++)
 		{
 			cl->members[i]->refcount++;
@@ -1574,6 +1411,7 @@ SearchCatCacheList(CatCache *cache,
 
 		/* Bump the list's refcount and return it */
 		cl->refcount++;
+		ResourceOwnerRememberCatCacheListRef(CurrentResourceOwner, cl);
 
 		CACHE2_elog(DEBUG2, "SearchCatCacheList(%s): found list",
 					cache->cc_relname);
@@ -1639,9 +1477,7 @@ SearchCatCacheList(CatCache *cache,
 			if (ct->c_list)
 				continue;
 
-			/* Found a match, so bump its refcount and move to front */
-			ct->refcount++;
-
+			/* Found a match, so move it to front */
 			DLMoveToFront(&ct->lrulist_elem);
 
 			break;
@@ -1655,6 +1491,16 @@ SearchCatCacheList(CatCache *cache,
 										 false);
 		}
 
+		/*
+		 * We have to bump the member refcounts immediately to ensure they
+		 * won't get dropped from the cache while loading other members.
+		 * If we get an error before we finish constructing the CatCList
+		 * then we will leak those reference counts.  This is annoying but
+		 * it has no real consequence beyond possibly generating some
+		 * warning messages at the next transaction commit, so it's not
+		 * worth fixing.
+		 */
+		ct->refcount++;
 		ctlist = lcons(ct, ctlist);
 		nmembers++;
 	}
@@ -1677,10 +1523,7 @@ SearchCatCacheList(CatCache *cache,
 	cl->cl_magic = CL_MAGIC;
 	cl->my_cache = cache;
 	DLInitElem(&cl->cache_elem, (void *) cl);
-	cl->refcount = 1;			/* count this first reference */
-	cl->prev_refcount = NULL;
-	cl->numpushes = 0;
-	cl->numalloc = 0;
+	cl->refcount = 0;			/* for the moment */
 	cl->dead = false;
 	cl->ordered = ordered;
 	cl->nkeys = nkeys;
@@ -1704,6 +1547,11 @@ SearchCatCacheList(CatCache *cache,
 	CACHE3_elog(DEBUG2, "SearchCatCacheList(%s): made list of %d members",
 				cache->cc_relname, nmembers);
 
+	/* Finally, bump the list's refcount and return it */
+	ResourceOwnerEnlargeCatCacheListRefs(CurrentResourceOwner);
+	cl->refcount++;
+	ResourceOwnerRememberCatCacheListRef(CurrentResourceOwner, cl);
+
 	return cl;
 }
 
@@ -1735,6 +1583,7 @@ ReleaseCatCacheList(CatCList *list)
 	}
 
 	list->refcount--;
+	ResourceOwnerForgetCatCacheListRef(CurrentResourceOwner, list);
 
 	if (list->refcount == 0
 #ifndef CATCACHE_FORCE_RELEASE
@@ -1748,7 +1597,7 @@ ReleaseCatCacheList(CatCList *list)
 /*
  * CatalogCacheCreateEntry
  *		Create a new CatCTup entry, copying the given HeapTuple and other
- *		supplied data into it.	The new entry is given refcount 1.
+ *		supplied data into it.	The new entry initially has refcount 0.
  */
 static CatCTup *
 CatalogCacheCreateEntry(CatCache *cache, HeapTuple ntp,
@@ -1775,13 +1624,10 @@ CatalogCacheCreateEntry(CatCache *cache, HeapTuple ntp,
 	DLInitElem(&ct->lrulist_elem, (void *) ct);
 	DLInitElem(&ct->cache_elem, (void *) ct);
 	ct->c_list = NULL;
-	ct->refcount = 1;			/* count this first reference */
+	ct->refcount = 0;			/* for the moment */
 	ct->dead = false;
 	ct->negative = negative;
 	ct->hash_value = hashValue;
-	ct->prev_refcount = NULL;
-	ct->numpushes = 0;
-	ct->numalloc = 0;
 
 	DLAddHead(&CacheHdr->ch_lrulist, &ct->lrulist_elem);
 	DLAddHead(&cache->cc_bucket[hashIndex], &ct->cache_elem);
@@ -1791,8 +1637,8 @@ CatalogCacheCreateEntry(CatCache *cache, HeapTuple ntp,
 
 	/*
 	 * If we've exceeded the desired size of the caches, try to throw away
-	 * the least recently used entry.  NB: the newly-built entry cannot
-	 * get thrown away here, because it has positive refcount.
+	 * the least recently used entry.  NB: be careful not to throw away
+	 * the newly-built entry...
 	 */
 	if (CacheHdr->ch_ntup > CacheHdr->ch_maxtup)
 	{
@@ -1805,7 +1651,7 @@ CatalogCacheCreateEntry(CatCache *cache, HeapTuple ntp,
 
 			prevelt = DLGetPred(elt);
 
-			if (oldct->refcount == 0)
+			if (oldct->refcount == 0 && oldct != ct)
 			{
 				CACHE2_elog(DEBUG2, "CatCacheCreateEntry(%s): Overflow, LRU removal",
 							cache->cc_relname);
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 2342899272..c4787042c0 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.206 2004/07/01 00:51:17 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/cache/relcache.c,v 1.207 2004/07/17 03:29:25 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -62,6 +62,7 @@
 #include "utils/inval.h"
 #include "utils/lsyscache.h"
 #include "utils/relcache.h"
+#include "utils/resowner.h"
 #include "utils/syscache.h"
 #include "utils/typcache.h"
 
@@ -273,8 +274,6 @@ static void IndexSupportInitialize(Form_pg_index iform,
 static OpClassCacheEnt *LookupOpclassInfo(Oid operatorClassOid,
 				  StrategyNumber numStrats,
 				  StrategyNumber numSupport);
-static inline void RelationPushReferenceCount(Relation rel);
-static inline void RelationPopReferenceCount(Relation rel);
 
 
 /*
@@ -830,16 +829,12 @@ RelationBuildDesc(RelationBuildDescInfo buildinfo,
 	RelationGetRelid(relation) = relid;
 
 	/*
-	 * initialize relation->rd_refcnt
-	 */
-	RelationSetReferenceCount(relation, 1);
-
-	/*
 	 * normal relations are not nailed into the cache; nor can a
 	 * pre-existing relation be new.  It could be temp though.	(Actually,
 	 * it could be new too, but it's okay to forget that fact if forced to
 	 * flush the entry.)
 	 */
+	relation->rd_refcnt = 0;
 	relation->rd_isnailed = 0;
 	relation->rd_isnew = false;
 	relation->rd_istemp = isTempNamespace(relation->rd_rel->relnamespace);
@@ -1280,9 +1275,9 @@ formrdesc(const char *relationName,
 	relation->rd_smgr = NULL;
 
 	/*
-	 * initialize reference count
+	 * initialize reference count: 1 because it is nailed in cache
 	 */
-	RelationSetReferenceCount(relation, 1);
+	relation->rd_refcnt = 1;
 
 	/*
 	 * all entries built with this routine are nailed-in-cache; none are
@@ -1487,6 +1482,8 @@ RelationIdGetRelation(Oid relationId)
 	buildinfo.i.info_id = relationId;
 
 	rd = RelationBuildDesc(buildinfo, NULL);
+	if (RelationIsValid(rd))
+		RelationIncrementReferenceCount(rd);
 	return rd;
 }
 
@@ -1516,6 +1513,8 @@ RelationSysNameGetRelation(const char *relationName)
 	buildinfo.i.info_name = (char *) relationName;
 
 	rd = RelationBuildDesc(buildinfo, NULL);
+	if (RelationIsValid(rd))
+		RelationIncrementReferenceCount(rd);
 	return rd;
 }
 
@@ -1525,6 +1524,36 @@ RelationSysNameGetRelation(const char *relationName)
  */
 
 /*
+ * RelationIncrementReferenceCount
+ *		Increments relation reference count.
+ *
+ * Note: bootstrap mode has its own weird ideas about relation refcount
+ * behavior; we ought to fix it someday, but for now, just disable
+ * reference count ownership tracking in bootstrap mode.
+ */
+void
+RelationIncrementReferenceCount(Relation rel)
+{
+	ResourceOwnerEnlargeRelationRefs(CurrentResourceOwner);
+	rel->rd_refcnt += 1;
+	if (!IsBootstrapProcessingMode())
+		ResourceOwnerRememberRelationRef(CurrentResourceOwner, rel);
+}
+
+/*
+ * RelationDecrementReferenceCount
+ *		Decrements relation reference count.
+ */
+void
+RelationDecrementReferenceCount(Relation rel)
+{
+	Assert(rel->rd_refcnt > 0);
+	rel->rd_refcnt -= 1;
+	if (!IsBootstrapProcessingMode())
+		ResourceOwnerForgetRelationRef(CurrentResourceOwner, rel);
+}
+
+/*
  * RelationClose - close an open relation
  *
  *	Actually, we just decrement the refcount.
@@ -1680,8 +1709,6 @@ RelationClearRelation(Relation relation, bool rebuild)
 	list_free(relation->rd_indexlist);
 	if (relation->rd_indexcxt)
 		MemoryContextDelete(relation->rd_indexcxt);
-	if (relation->rd_prevrefcnt)
-		pfree(relation->rd_prevrefcnt);
 
 	/*
 	 * If we're really done with the relcache entry, blow it away. But if
@@ -1704,6 +1731,10 @@ RelationClearRelation(Relation relation, bool rebuild)
 		 * When rebuilding an open relcache entry, must preserve ref count
 		 * and rd_isnew flag.  Also attempt to preserve the tupledesc and
 		 * rewrite-rule substructures in place.
+		 *
+		 * Note that this process does not touch CurrentResourceOwner;
+		 * which is good because whatever ref counts the entry may have
+		 * do not necessarily belong to that resource owner.
 		 */
 		int			old_refcnt = relation->rd_refcnt;
 		bool		old_isnew = relation->rd_isnew;
@@ -1726,7 +1757,7 @@ RelationClearRelation(Relation relation, bool rebuild)
 			elog(ERROR, "relation %u deleted while still in use",
 				 buildinfo.i.info_id);
 		}
-		RelationSetReferenceCount(relation, old_refcnt);
+		relation->rd_refcnt = old_refcnt;
 		relation->rd_isnew = old_isnew;
 		if (equalTupleDescs(old_att, relation->rd_att))
 		{
@@ -1964,7 +1995,7 @@ RelationCacheInvalidate(void)
 /*
  * AtEOXact_RelationCache
  *
- *	Clean up the relcache at transaction commit or abort.
+ *	Clean up the relcache at main-transaction commit or abort.
  *
  * Note: this must be called *before* processing invalidation messages.
  * In the case of abort, we don't want to try to rebuild any invalidated
@@ -2031,22 +2062,16 @@ AtEOXact_RelationCache(bool isCommit)
 				elog(WARNING, "relcache reference leak: relation \"%s\" has refcnt %d instead of %d",
 					 RelationGetRelationName(relation),
 					 relation->rd_refcnt, expected_refcnt);
-				RelationSetReferenceCount(relation, expected_refcnt);
+				relation->rd_refcnt = expected_refcnt;
 			}
 		}
 		else
 		{
 			/* abort case, just reset it quietly */
-			RelationSetReferenceCount(relation, expected_refcnt);
+			relation->rd_refcnt = expected_refcnt;
 		}
 
 		/*
-		 * Reset the refcount stack.  Just drop the item count; don't deallocate
-		 * the stack itself so it can be reused by future subtransactions.
-		 */
-		relation->rd_numpushed = 0;
-
-		/*
 		 * Flush any temporary index list.
 		 */
 		if (relation->rd_indexvalid == 2)
@@ -2059,131 +2084,6 @@ AtEOXact_RelationCache(bool isCommit)
 }
 
 /*
- * RelationPushReferenceCount
- *
- * Push the current reference count into the stack.  Don't modify the
- * reference count itself.
- */
-static inline void
-RelationPushReferenceCount(Relation rel)
-{
-	/* Enlarge the stack if we run out of space. */
-	if (rel->rd_numpushed == rel->rd_numalloc)
-	{
-		MemoryContext	old_cxt = MemoryContextSwitchTo(CacheMemoryContext);
-
-		if (rel->rd_numalloc == 0)
-		{
-			rel->rd_numalloc = 8;
-			rel->rd_prevrefcnt = palloc(rel->rd_numalloc * sizeof(int));
-		}
-		else
-		{
-			rel->rd_numalloc *= 2;
-			rel->rd_prevrefcnt = repalloc(rel->rd_prevrefcnt, rel->rd_numalloc * sizeof(int));
-		}
-
-		MemoryContextSwitchTo(old_cxt);
-	}
-
-	rel->rd_prevrefcnt[rel->rd_numpushed++] = rel->rd_refcnt;
-}
-
-/*
- * RelationPopReferenceCount
- *
- * Pop the latest stored reference count.  If there is none, drop it
- * to zero; the entry was created in the current subtransaction.
- */
-static inline void
-RelationPopReferenceCount(Relation rel)
-{
-	if (rel->rd_numpushed == 0)
-	{
-		rel->rd_refcnt = rel->rd_isnailed ? 1 : 0;
-		return;
-	}
-
-	rel->rd_refcnt = rel->rd_prevrefcnt[--rel->rd_numpushed];
-}
-
-/*
- * AtEOSubXact_RelationCache
- */
-void
-AtEOSubXact_RelationCache(bool isCommit)
-{
-	HASH_SEQ_STATUS status;
-	RelIdCacheEnt *idhentry;
-
-	/* We'd better not be bootstrapping. */
-	Assert(!IsBootstrapProcessingMode());
-
-	hash_seq_init(&status, RelationIdCache);
-
-	while ((idhentry = (RelIdCacheEnt *) hash_seq_search(&status)) != NULL)
-	{
-		Relation	relation = idhentry->reldesc;
-
-		/*
-		 * During subtransaction commit, we first check whether the
-		 * current refcount is correct: if there is no item in the stack,
-		 * the relcache entry was created during this subtransaction, it should
-		 * be 0 (or 1 for nailed relations).  If the stack has at least one
-		 * item, the expected count is whatever that item is.
-		 */
-		if (isCommit)
-		{
-			int expected_refcnt;
-
-			if (relation->rd_numpushed == 0)
-				expected_refcnt = relation->rd_isnailed ? 1 : 0;
-			else
-				expected_refcnt = relation->rd_prevrefcnt[relation->rd_numpushed - 1];
-
-			if (relation->rd_refcnt != expected_refcnt)
-			{
-				elog(WARNING, "relcache reference leak: relation \"%s\" has refcnt %d instead of %d",
-						RelationGetRelationName(relation),
-						relation->rd_refcnt, expected_refcnt);
-			}
-		}
-
-		/*
-		 * On commit, the expected count is stored so there's no harm in
-		 * popping it (and we may need to fix if there was a leak); and during
-		 * abort, the correct refcount has to be restored.
-		 */
-		RelationPopReferenceCount(relation);
-	}
-}
-
-/*
- * AtSubStart_RelationCache
- *
- * At subtransaction start, we push the current reference count into
- * the refcount stack, so it can be restored if the subtransaction aborts.
- */
-void
-AtSubStart_RelationCache(void)
-{
-	HASH_SEQ_STATUS status;
-	RelIdCacheEnt *idhentry;
-
-	/* We'd better not be bootstrapping. */
-	Assert(!IsBootstrapProcessingMode());
-
-	hash_seq_init(&status, RelationIdCache);
-
-	while ((idhentry = (RelIdCacheEnt *) hash_seq_search(&status)) != NULL)
-	{
-		Relation	relation = idhentry->reldesc;
-
-		RelationPushReferenceCount(relation);
-	}
-}
-
-/*
  *		RelationBuildLocalRelation
  *			Build a relcache entry for an about-to-be-created relation,
  *			and enter it into the relcache.
@@ -2223,7 +2123,7 @@ RelationBuildLocalRelation(const char *relname,
 	/* make sure relation is marked as having no open file yet */
 	rel->rd_smgr = NULL;
 
-	RelationSetReferenceCount(rel, 1);
+	rel->rd_refcnt = nailit ? 1 : 0;
 
 	/* it's being created in this transaction */
 	rel->rd_isnew = true;
@@ -2305,6 +2205,11 @@ RelationBuildLocalRelation(const char *relname,
 	 */
 	MemoryContextSwitchTo(oldcxt);
 
+	/*
+	 * Caller expects us to pin the returned entry.
+	 */
+	RelationIncrementReferenceCount(rel);
+
 	return rel;
 }
 
@@ -2422,7 +2327,7 @@ RelationCacheInitializePhase2(void)
 			buildinfo.i.info_name = (indname); \
 			ird = RelationBuildDesc(buildinfo, NULL); \
 			ird->rd_isnailed = 1; \
-			RelationSetReferenceCount(ird, 1); \
+			ird->rd_refcnt = 1; \
 		} while (0)
 
 		LOAD_CRIT_INDEX(ClassNameNspIndex);
@@ -3201,9 +3106,9 @@ load_relcache_init_file(void)
 		rel->rd_smgr = NULL;
 		rel->rd_targblock = InvalidBlockNumber;
 		if (rel->rd_isnailed)
-			RelationSetReferenceCount(rel, 1);
+			rel->rd_refcnt = 1;
 		else
-			RelationSetReferenceCount(rel, 0);
+			rel->rd_refcnt = 0;
 		rel->rd_indexvalid = 0;
 		rel->rd_indexlist = NIL;
 		MemSet(&rel->pgstat_info, 0, sizeof(rel->pgstat_info));