Nested transactions. There is still much left to do, especially on the

performance front, but with feature freeze upon us I think it's time to
drive a stake in the ground and say that this will be in 7.5.

Alvaro Herrera, with some help from Tom Lane.

10 files changed, 1037 insertions, 300 deletions

diff --git a/src/backend/utils/cache/catcache.c b/src/backend/utils/cache/catcache.c
index 5e91a7283ec..8bfa3610bdb 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.112 2004/05/26 04:41:40 neilc Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/cache/catcache.c,v 1.113 2004/07/01 00:51:17 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -360,6 +360,8 @@ 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;
@@ -394,6 +396,8 @@ 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);
 }
 
@@ -518,9 +522,9 @@ CreateCacheMemoryContext(void)
 	if (!CacheMemoryContext)
 		CacheMemoryContext = AllocSetContextCreate(TopMemoryContext,
 												   "CacheMemoryContext",
-												ALLOCSET_DEFAULT_MINSIZE,
-											   ALLOCSET_DEFAULT_INITSIZE,
-											   ALLOCSET_DEFAULT_MAXSIZE);
+												   ALLOCSET_DEFAULT_MINSIZE,
+												   ALLOCSET_DEFAULT_INITSIZE,
+												   ALLOCSET_DEFAULT_MAXSIZE);
 }
 
 
@@ -560,6 +564,13 @@ 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);
@@ -585,6 +596,13 @@ 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);
@@ -592,6 +610,161 @@ 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.
@@ -1505,6 +1678,9 @@ SearchCatCacheList(CatCache *cache,
 	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->dead = false;
 	cl->ordered = ordered;
 	cl->nkeys = nkeys;
@@ -1603,6 +1779,9 @@ CatalogCacheCreateEntry(CatCache *cache, HeapTuple ntp,
 	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);
diff --git a/src/backend/utils/cache/inval.c b/src/backend/utils/cache/inval.c
index ea958a27b46..e54a74fae4b 100644
--- a/src/backend/utils/cache/inval.c
+++ b/src/backend/utils/cache/inval.c
@@ -33,6 +33,10 @@
  *	to record the transaction commit before sending SI messages, otherwise
  *	the other backends won't see our updated tuples as good.
  *
+ *	When a subtransaction aborts, we can process and discard any events
+ *	it has queued.  When a subtransaction commits, we just add its events
+ *	to the pending lists of the parent transaction.
+ *
  *	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
@@ -66,15 +70,17 @@
  *	manipulating the init file is in relcache.c, but we keep track of the
  *	need for it here.
  *
- *	All the request lists are kept in TopTransactionContext memory, since
- *	they need not live beyond the end of the current transaction.
+ *	The request lists proper are kept in CurTransactionContext of their
+ *	creating (sub)transaction, since they can be forgotten on abort of that
+ *	transaction but must be kept till top-level commit otherwise.  For
+ *	simplicity we keep the controlling list-of-lists in TopTransactionContext.
  *
  *
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/utils/cache/inval.c,v 1.62 2004/06/18 06:13:52 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/cache/inval.c,v 1.63 2004/07/01 00:51:17 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -95,7 +101,7 @@
  * To minimize palloc traffic, we keep pending requests in successively-
  * larger chunks (a slightly more sophisticated version of an expansible
  * array).	All request types can be stored as SharedInvalidationMessage
- * records.
+ * records.  The ordering of requests within a list is never significant.
  */
 typedef struct InvalidationChunk
 {
@@ -112,12 +118,15 @@ typedef struct InvalidationListHeader
 } InvalidationListHeader;
 
 /*----------------
- *	Invalidation info is divided into two lists:
+ * 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.
+ * Actually, we need two such lists for each level of nested transaction,
+ * so that we can discard events from an aborted subtransaction.  When
+ * a subtransaction commits, we append its lists to the parent's lists.
  *
  * 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
@@ -125,13 +134,22 @@ typedef struct InvalidationListHeader
  *----------------
  */
 
-/* head of current-command event list */
-static InvalidationListHeader CurrentCmdInvalidMsgs;
+typedef struct TransInvalidationInfo
+{
+	/* Back link to parent transaction's info */
+	struct TransInvalidationInfo *parent;
+
+	/* head of current-command event list */
+	InvalidationListHeader CurrentCmdInvalidMsgs;
 
-/* head of previous-commands event list */
-static InvalidationListHeader PriorCmdInvalidMsgs;
+	/* head of previous-commands event list */
+	InvalidationListHeader PriorCmdInvalidMsgs;
 
-static bool RelcacheInitFileInval;		/* init file must be invalidated? */
+	/* init file must be invalidated? */
+	bool RelcacheInitFileInval;
+} TransInvalidationInfo;
+
+static TransInvalidationInfo *transInvalInfo = NULL;
 
 /*
  * Dynamically-registered callback functions.  Current implementation
@@ -176,7 +194,7 @@ AddInvalidationMessage(InvalidationChunk **listHdr,
 		/* First time through; create initial chunk */
 #define FIRSTCHUNKSIZE 16
 		chunk = (InvalidationChunk *)
-			MemoryContextAlloc(TopTransactionContext,
+			MemoryContextAlloc(CurTransactionContext,
 							   sizeof(InvalidationChunk) +
 				(FIRSTCHUNKSIZE - 1) *sizeof(SharedInvalidationMessage));
 		chunk->nitems = 0;
@@ -190,7 +208,7 @@ AddInvalidationMessage(InvalidationChunk **listHdr,
 		int			chunksize = 2 * chunk->maxitems;
 
 		chunk = (InvalidationChunk *)
-			MemoryContextAlloc(TopTransactionContext,
+			MemoryContextAlloc(CurTransactionContext,
 							   sizeof(InvalidationChunk) +
 					 (chunksize - 1) *sizeof(SharedInvalidationMessage));
 		chunk->nitems = 0;
@@ -204,29 +222,6 @@ AddInvalidationMessage(InvalidationChunk **listHdr,
 }
 
 /*
- * Free a list of inval message chunks.
- *
- * NOTE: when we are about to commit or abort a transaction, it's
- * not really necessary to pfree the lists explicitly, since they will
- * go away anyway when TopTransactionContext is destroyed.
- */
-static void
-FreeInvalidationMessageList(InvalidationChunk **listHdr)
-{
-	InvalidationChunk *chunk = *listHdr;
-
-	*listHdr = NULL;
-
-	while (chunk != NULL)
-	{
-		InvalidationChunk *nextchunk = chunk->next;
-
-		pfree(chunk);
-		chunk = nextchunk;
-	}
-}
-
-/*
  * Append one list of invalidation message chunks to another, resetting
  * the source chunk-list pointer to NULL.
  */
@@ -332,31 +327,6 @@ AppendInvalidationMessages(InvalidationListHeader *dest,
 }
 
 /*
- * Reset an invalidation list to empty
- *
- * physicalFree may be set false if caller knows transaction is ending
- */
-static void
-DiscardInvalidationMessages(InvalidationListHeader *hdr, bool physicalFree)
-{
-	if (physicalFree)
-	{
-		/* Physically pfree the list data */
-		FreeInvalidationMessageList(&hdr->cclist);
-		FreeInvalidationMessageList(&hdr->rclist);
-	}
-	else
-	{
-		/*
-		 * Assume the storage will go away at xact end, just reset
-		 * pointers
-		 */
-		hdr->cclist = NULL;
-		hdr->rclist = NULL;
-	}
-}
-
-/*
  * Execute the given function for all the messages in an invalidation list.
  * The list is not altered.
  *
@@ -386,7 +356,7 @@ RegisterCatcacheInvalidation(int cacheId,
 							 ItemPointer tuplePtr,
 							 Oid dbId)
 {
-	AddCatcacheInvalidationMessage(&CurrentCmdInvalidMsgs,
+	AddCatcacheInvalidationMessage(&transInvalInfo->CurrentCmdInvalidMsgs,
 								   cacheId, hashValue, tuplePtr, dbId);
 }
 
@@ -398,7 +368,7 @@ RegisterCatcacheInvalidation(int cacheId,
 static void
 RegisterRelcacheInvalidation(Oid dbId, Oid relId, RelFileNode physId)
 {
-	AddRelcacheInvalidationMessage(&CurrentCmdInvalidMsgs,
+	AddRelcacheInvalidationMessage(&transInvalInfo->CurrentCmdInvalidMsgs,
 								   dbId, relId, physId);
 
 	/*
@@ -406,7 +376,7 @@ RegisterRelcacheInvalidation(Oid dbId, Oid relId, RelFileNode physId)
 	 * relcache init file, mark that we need to zap that file at commit.
 	 */
 	if (RelationIdIsInInitFile(relId))
-		RelcacheInitFileInval = true;
+		transInvalInfo->RelcacheInitFileInval = true;
 }
 
 /*
@@ -619,8 +589,38 @@ AcceptInvalidationMessages(void)
 }
 
 /*
- * AtEOXactInvalidationMessages
- *		Process queued-up invalidation messages at end of transaction.
+ * AtStart_Inval
+ *		Initialize inval lists at start of a main transaction.
+ */
+void
+AtStart_Inval(void)
+{
+	Assert(transInvalInfo == NULL);
+	transInvalInfo = (TransInvalidationInfo *)
+		MemoryContextAllocZero(TopTransactionContext,
+							   sizeof(TransInvalidationInfo));
+}
+
+/*
+ * AtSubStart_Inval
+ *		Initialize inval lists at start of a subtransaction.
+ */
+void
+AtSubStart_Inval(void)
+{
+	TransInvalidationInfo *myInfo;
+
+	Assert(transInvalInfo != NULL);
+	myInfo = (TransInvalidationInfo *)
+		MemoryContextAllocZero(TopTransactionContext,
+							   sizeof(TransInvalidationInfo));
+	myInfo->parent = transInvalInfo;
+	transInvalInfo = myInfo;
+}
+
+/*
+ * AtEOXact_Inval
+ *		Process queued-up invalidation messages at end of main transaction.
  *
  * If isCommit, we must send out the messages in our PriorCmdInvalidMsgs list
  * to the shared invalidation message queue.  Note that these will be read
@@ -643,8 +643,11 @@ AcceptInvalidationMessages(void)
  *		This should be called as the last step in processing a transaction.
  */
 void
-AtEOXactInvalidationMessages(bool isCommit)
+AtEOXact_Inval(bool isCommit)
 {
+	/* Must be at top of stack */
+	Assert(transInvalInfo != NULL && transInvalInfo->parent == NULL);
+
 	if (isCommit)
 	{
 		/*
@@ -652,28 +655,77 @@ AtEOXactInvalidationMessages(bool isCommit)
 		 * and after we send the SI messages.  However, we need not do
 		 * anything unless we committed.
 		 */
-		if (RelcacheInitFileInval)
+		if (transInvalInfo->RelcacheInitFileInval)
 			RelationCacheInitFileInvalidate(true);
 
-		AppendInvalidationMessages(&PriorCmdInvalidMsgs,
-								   &CurrentCmdInvalidMsgs);
+		AppendInvalidationMessages(&transInvalInfo->PriorCmdInvalidMsgs,
+								   &transInvalInfo->CurrentCmdInvalidMsgs);
 
-		ProcessInvalidationMessages(&PriorCmdInvalidMsgs,
+		ProcessInvalidationMessages(&transInvalInfo->PriorCmdInvalidMsgs,
 									SendSharedInvalidMessage);
 
-		if (RelcacheInitFileInval)
+		if (transInvalInfo->RelcacheInitFileInval)
 			RelationCacheInitFileInvalidate(false);
 	}
 	else
 	{
-		ProcessInvalidationMessages(&PriorCmdInvalidMsgs,
+		ProcessInvalidationMessages(&transInvalInfo->PriorCmdInvalidMsgs,
 									LocalExecuteInvalidationMessage);
 	}
 
-	RelcacheInitFileInval = false;
+	/* Need not free anything explicitly */
+	transInvalInfo = NULL;
+}
+
+/*
+ * AtSubEOXact_Inval
+ *		Process queued-up invalidation messages at end of subtransaction.
+ *
+ * If isCommit, process CurrentCmdInvalidMsgs if any (there probably aren't),
+ * and then attach both CurrentCmdInvalidMsgs and PriorCmdInvalidMsgs to the
+ * parent's PriorCmdInvalidMsgs list.
+ *
+ * If not isCommit, we are aborting, and must locally process the messages
+ * in PriorCmdInvalidMsgs.	No messages need be sent to other backends.
+ * We can forget about CurrentCmdInvalidMsgs too, since those changes haven't
+ * touched the caches yet.
+ *
+ * In any case, pop the transaction stack.  We need not physically free memory
+ * here, since CurTransactionContext is about to be emptied anyway
+ * (if aborting).
+ */
+void
+AtSubEOXact_Inval(bool isCommit)
+{
+	TransInvalidationInfo *myInfo = transInvalInfo;
+
+	/* Must be at non-top of stack */
+	Assert(myInfo != NULL && myInfo->parent != NULL);
+
+	if (isCommit)
+	{
+		/* If CurrentCmdInvalidMsgs still has anything, fix it */
+		CommandEndInvalidationMessages();
+
+		/* Pass up my inval messages to parent */
+		AppendInvalidationMessages(&myInfo->parent->PriorCmdInvalidMsgs,
+								   &myInfo->PriorCmdInvalidMsgs);
 
-	DiscardInvalidationMessages(&PriorCmdInvalidMsgs, false);
-	DiscardInvalidationMessages(&CurrentCmdInvalidMsgs, false);
+		/* Pending relcache inval becomes parent's problem too */
+		if (myInfo->RelcacheInitFileInval)
+			myInfo->parent->RelcacheInitFileInval = true;
+	}
+	else
+	{
+		ProcessInvalidationMessages(&myInfo->PriorCmdInvalidMsgs,
+									LocalExecuteInvalidationMessage);
+	}
+
+	/* Pop the transaction state stack */
+	transInvalInfo = myInfo->parent;
+
+	/* Need not free anything else explicitly */
+	pfree(myInfo);
 }
 
 /*
@@ -687,27 +739,25 @@ AtEOXactInvalidationMessages(bool isCommit)
  * 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.
- *
  * Note:
  *		This should be called during CommandCounterIncrement(),
  *		after we have advanced the command ID.
  */
 void
-CommandEndInvalidationMessages(bool isCommit)
+CommandEndInvalidationMessages(void)
 {
-	if (isCommit)
-	{
-		ProcessInvalidationMessages(&CurrentCmdInvalidMsgs,
-									LocalExecuteInvalidationMessage);
-		AppendInvalidationMessages(&PriorCmdInvalidMsgs,
-								   &CurrentCmdInvalidMsgs);
-	}
-	else
-	{
-		/* XXX what needs to be done here? */
-	}
+	/*
+	 * You might think this shouldn't be called outside any transaction,
+	 * but bootstrap does it, and also ABORT issued when not in a transaction.
+	 * So just quietly return if no state to work on.
+	 */
+	if (transInvalInfo == NULL)
+		return;
+
+	ProcessInvalidationMessages(&transInvalInfo->CurrentCmdInvalidMsgs,
+								LocalExecuteInvalidationMessage);
+	AppendInvalidationMessages(&transInvalInfo->PriorCmdInvalidMsgs,
+							   &transInvalInfo->CurrentCmdInvalidMsgs);
 }
 
 /*
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index ee8b46407e1..23428992724 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.205 2004/06/18 06:13:52 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/cache/relcache.c,v 1.206 2004/07/01 00:51:17 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -273,6 +273,8 @@ 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);
 
 
 /*
@@ -1678,6 +1680,8 @@ 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
@@ -1968,7 +1972,7 @@ RelationCacheInvalidate(void)
  * we must reset refcnts before handling pending invalidations.
  */
 void
-AtEOXact_RelationCache(bool commit)
+AtEOXact_RelationCache(bool isCommit)
 {
 	HASH_SEQ_STATUS status;
 	RelIdCacheEnt *idhentry;
@@ -1993,7 +1997,7 @@ AtEOXact_RelationCache(bool commit)
 		 */
 		if (relation->rd_isnew)
 		{
-			if (commit)
+			if (isCommit)
 				relation->rd_isnew = false;
 			else
 			{
@@ -2019,7 +2023,7 @@ AtEOXact_RelationCache(bool commit)
 		 */
 		expected_refcnt = relation->rd_isnailed ? 1 : 0;
 
-		if (commit)
+		if (isCommit)
 		{
 			if (relation->rd_refcnt != expected_refcnt &&
 				!IsBootstrapProcessingMode())
@@ -2037,6 +2041,12 @@ AtEOXact_RelationCache(bool commit)
 		}
 
 		/*
+		 * 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)
@@ -2049,6 +2059,131 @@ AtEOXact_RelationCache(bool commit)
 }
 
 /*
+ * 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.
diff --git a/src/backend/utils/init/postinit.c b/src/backend/utils/init/postinit.c
index 48d28d429f2..3caf18c5f33 100644
--- a/src/backend/utils/init/postinit.c
+++ b/src/backend/utils/init/postinit.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/utils/init/postinit.c,v 1.134 2004/06/18 06:13:54 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/init/postinit.c,v 1.135 2004/07/01 00:51:20 tgl Exp $
  *
  *
  *-------------------------------------------------------------------------
@@ -27,7 +27,6 @@
 #include "catalog/pg_database.h"
 #include "catalog/pg_shadow.h"
 #include "catalog/pg_tablespace.h"
-#include "commands/trigger.h"
 #include "mb/pg_wchar.h"
 #include "miscadmin.h"
 #include "postmaster/postmaster.h"
@@ -350,12 +349,6 @@ InitPostgres(const char *dbname, const char *username)
 	/* Initialize portal manager */
 	EnablePortalManager();
 
-	/*
-	 * Initialize the deferred trigger manager --- must happen before
-	 * first transaction start.
-	 */
-	DeferredTriggerInit();
-
 	/* start a new transaction here before access to db */
 	if (!bootstrap)
 		StartTransactionCommand();
diff --git a/src/backend/utils/misc/README b/src/backend/utils/misc/README
index 12a2cdef036..3ea838b1f53 100644
--- a/src/backend/utils/misc/README
+++ b/src/backend/utils/misc/README
@@ -1,4 +1,4 @@
-$PostgreSQL: pgsql/src/backend/utils/misc/README,v 1.4 2004/01/19 19:04:40 tgl Exp $
+$PostgreSQL: pgsql/src/backend/utils/misc/README,v 1.5 2004/07/01 00:51:24 tgl Exp $
 
 
 GUC IMPLEMENTATION NOTES
@@ -68,49 +68,66 @@ SET on transaction abort, and rollback of SET LOCAL at transaction end
 would be effective had there never been any SET commands in the current
 session.
 
-To handle these cases we must keep track of as many as four distinct
-values for each variable.  They are:
+To handle these cases we must keep track of many distinct values for each
+variable.  The primary values are:
 
 * actual variable contents	always the current effective value
 
 * reset_value			the value to use for RESET
 
-* session_value			the "committed" setting for the session
-
 * tentative_value		the uncommitted result of SET
 
-During initialization we set the first three of these (actual, reset_value,
-and session_value) based on whichever non-interactive source has the
-highest priority.  All three will have the same value.
+The reason we need a tentative_value separate from the actual value is
+that when a transaction does SET followed by SET LOCAL, the actual value
+will now be the LOCAL value, but we want to remember the prior SET so that
+that value is restored at transaction commit.
+
+In addition, for each level of transaction (possibly nested) we have to
+remember the transaction-entry-time actual and tentative values, in case
+we need to restore them at transaction end.  (The RESET value is essentially
+non-transactional, so it doesn't have to be stacked.)  For efficiency these
+stack entries are not constructed until/unless the variable is actually SET
+within a particular transaction.
+
+During initialization we set the actual value and reset_value based on
+whichever non-interactive source has the highest priority.  They will
+have the same value.  The tentative_value is not meaningful at this point.
+
+A SET command starts by stacking the existing actual and tentative values
+if this hasn't already been done within the current transaction.  Then:
 
 A SET LOCAL command sets the actual variable (and nothing else).  At
-transaction end, the session_value is used to restore the actual variable
-to its pre-transaction value.
+transaction end, the stacked values are used to restore the GUC entry
+to its pre-transaction state.
 
 A SET (or SET SESSION) command sets the actual variable, and if no error,
 then sets the tentative_value.  If the transaction commits, the
-tentative_value is assigned to the session_value and the actual variable
-(which could by now be different, if the SET was followed by SET LOCAL).
-If the transaction aborts, the tentative_value is discarded and the
-actual variable is restored from the session_value.
+tentative_value is assigned again to the actual variable (which could by
+now be different, if the SET was followed by SET LOCAL).  If the
+transaction aborts, the stacked values are used to restore the GUC entry
+to its pre-transaction state.
+
+In the case of SET within nested subtransactions, at each commit the
+tentative_value propagates out to the next transaction level.  It will
+be thrown away at abort of any level, or after exiting the top transaction.
 
 RESET is executed like a SET, but using the reset_value as the desired new
 value.  (We do not provide a RESET LOCAL command, but SET LOCAL TO DEFAULT
 has the same behavior that RESET LOCAL would.)  The source associated with
-the reset_value also becomes associated with the actual and session values.
+the reset_value also becomes associated with the actual and tentative values.
 
 If SIGHUP is received, the GUC code rereads the postgresql.conf
 configuration file (this does not happen in the signal handler, but at
 next return to main loop; note that it can be executed while within a
 transaction).  New values from postgresql.conf are assigned to actual
-variable, reset_value, and session_value, but only if each of these has a
-current source priority <= PGC_S_FILE.  (It is thus possible for
-reset_value to track the config-file setting even if there is currently
-a different interactive value of the actual variable.)
+variable, reset_value, and stacked actual values, but only if each of
+these has a current source priority <= PGC_S_FILE.  (It is thus possible
+for reset_value to track the config-file setting even if there is
+currently a different interactive value of the actual variable.)
 
 Note that tentative_value is unused and undefined except between a SET
 command and the end of the transaction.  Also notice that we must track
-the source associated with each of the four values.
+the source associated with each one of the values.
 
 The assign_hook and show_hook routines work only with the actual variable,
 and are not directly aware of the additional values maintained by GUC.
@@ -129,9 +146,9 @@ pstrdup/palloc mechanisms.  We would need to keep them in a permanent
 context anyway, and strdup gives us more control over handling
 out-of-memory failures.
 
-We allow a variable's actual value, reset_val, session_val, and
-tentative_val to point at the same storage.  This makes it slightly harder
-to free space (must test that the value to be freed isn't equal to any of
-the other three pointers).  The main advantage is that we never need to
-strdup during transaction commit/abort, so cannot cause an out-of-memory
-failure there.
+We allow a string variable's actual value, reset_val, tentative_val, and
+stacked copies of same to point at the same storage.  This makes it
+slightly harder to free space (must test whether a value to be freed isn't
+equal to any of the other pointers in the GUC entry or associated stack
+items).  The main advantage is that we never need to strdup during
+transaction commit/abort, so cannot cause an out-of-memory failure there.
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index f050e201e2a..f5c16de83ba 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -10,17 +10,16 @@
  * Written by Peter Eisentraut <peter_e@gmx.net>.
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/utils/misc/guc.c,v 1.211 2004/06/11 03:54:54 momjian Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/misc/guc.c,v 1.212 2004/07/01 00:51:24 tgl Exp $
  *
  *--------------------------------------------------------------------
  */
 #include "postgres.h"
 
-#include <errno.h>
+#include <ctype.h>
 #include <float.h>
 #include <limits.h>
 #include <unistd.h>
-#include <ctype.h>
 
 #include "utils/guc.h"
 #include "utils/guc_tables.h"
@@ -54,6 +53,7 @@
 #include "tcop/tcopprot.h"
 #include "utils/array.h"
 #include "utils/builtins.h"
+#include "utils/memutils.h"
 #include "utils/pg_locale.h"
 #include "pgstat.h"
 
@@ -105,6 +105,7 @@ static const char *assign_custom_variable_classes(const char *newval, bool doit,
 						   GucSource source);
 static bool assign_stage_log_stats(bool newval, bool doit, GucSource source);
 static bool assign_log_stats(bool newval, bool doit, GucSource source);
+static bool assign_transaction_read_only(bool newval, bool doit, GucSource source);
 
 
 /*
@@ -174,45 +175,6 @@ static int	max_identifier_length;
 static int	block_size;
 static bool integer_datetimes;
 
-/* Macros for freeing malloc'd pointers only if appropriate to do so */
-/* Some of these tests are probably redundant, but be safe ... */
-#define SET_STRING_VARIABLE(rec, newval) \
-	do { \
-		if (*(rec)->variable && \
-			*(rec)->variable != (rec)->reset_val && \
-			*(rec)->variable != (rec)->session_val && \
-			*(rec)->variable != (rec)->tentative_val) \
-			free(*(rec)->variable); \
-		*(rec)->variable = (newval); \
-	} while (0)
-#define SET_STRING_RESET_VAL(rec, newval) \
-	do { \
-		if ((rec)->reset_val && \
-			(rec)->reset_val != *(rec)->variable && \
-			(rec)->reset_val != (rec)->session_val && \
-			(rec)->reset_val != (rec)->tentative_val) \
-			free((rec)->reset_val); \
-		(rec)->reset_val = (newval); \
-	} while (0)
-#define SET_STRING_SESSION_VAL(rec, newval) \
-	do { \
-		if ((rec)->session_val && \
-			(rec)->session_val != *(rec)->variable && \
-			(rec)->session_val != (rec)->reset_val && \
-			(rec)->session_val != (rec)->tentative_val) \
-			free((rec)->session_val); \
-		(rec)->session_val = (newval); \
-	} while (0)
-#define SET_STRING_TENTATIVE_VAL(rec, newval) \
-	do { \
-		if ((rec)->tentative_val && \
-			(rec)->tentative_val != *(rec)->variable && \
-			(rec)->tentative_val != (rec)->reset_val && \
-			(rec)->tentative_val != (rec)->session_val) \
-			free((rec)->tentative_val); \
-		(rec)->tentative_val = (newval); \
-	} while (0)
-
 
 /*
  * Displayable names for context types (enum GucContext)
@@ -801,7 +763,7 @@ static struct config_bool ConfigureNamesBool[] =
 			GUC_NO_RESET_ALL | GUC_NOT_IN_SAMPLE | GUC_DISALLOW_IN_FILE
 		},
 		&XactReadOnly,
-		false, NULL, NULL
+		false, assign_transaction_read_only, NULL
 	},
 	{
 		{"add_missing_from", PGC_USERSET, COMPAT_OPTIONS_PREVIOUS,
@@ -1766,14 +1728,13 @@ static const char * const map_old_guc_names[] = {
  */
 static struct config_generic **guc_variables;
 
-/* Current number of variables contained in the vector
- */
+/* Current number of variables contained in the vector */
 static int num_guc_variables;
 
-/* Vector capacity
- */
+/* Vector capacity */
 static int size_guc_variables;
 
+
 static bool guc_dirty;			/* TRUE if need to do commit/abort work */
 
 static bool reporting_enabled;	/* TRUE to enable GUC_REPORT */
@@ -1783,14 +1744,71 @@ static char *guc_string_workspace;		/* for avoiding memory leaks */
 
 static int	guc_var_compare(const void *a, const void *b);
 static int	guc_name_compare(const char *namea, const char *nameb);
+static void push_old_value(struct config_generic *gconf);
 static void ReportGUCOption(struct config_generic * record);
 static char *_ShowOption(struct config_generic * record);
 
-struct config_generic** get_guc_variables()
+
+/*
+ * Support for assigning to a field of a string GUC item.  Free the prior
+ * value if it's not referenced anywhere else in the item (including stacked
+ * states).
+ */
+static void
+set_string_field(struct config_string *conf, char **field, char *newval)
+{
+	char	*oldval = *field;
+	GucStack *stack;
+
+	/* Do the assignment */
+	*field = newval;
+
+	/* Exit if any duplicate references, or if old value was NULL anyway */
+	if (oldval == NULL ||
+		oldval == *(conf->variable) ||
+		oldval == conf->reset_val ||
+		oldval == conf->tentative_val)
+		return;
+	for (stack = conf->gen.stack; stack; stack = stack->prev)
+	{
+		if (oldval == stack->tentative_val.stringval ||
+			oldval == stack->value.stringval)
+			return;
+	}
+
+	/* Not used anymore, so free it */
+	free(oldval);
+}
+
+/*
+ * Detect whether strval is referenced anywhere in a GUC string item
+ */
+static bool
+string_field_used(struct config_string *conf, char *strval)
+{
+	GucStack *stack;
+
+	if (strval == *(conf->variable) ||
+		strval == conf->reset_val ||
+		strval == conf->tentative_val)
+		return true;
+	for (stack = conf->gen.stack; stack; stack = stack->prev)
+	{
+		if (strval == stack->tentative_val.stringval ||
+			strval == stack->value.stringval)
+			return true;
+	}
+	return false;
+}
+
+
+struct config_generic **
+get_guc_variables(void)
 {
 	return guc_variables;
 }
 
+
 /*
  * Build the sorted array.	This is split out so that it could be
  * re-executed after startup (eg, we could allow loadable modules to
@@ -2001,14 +2019,13 @@ find_option(const char *name)
 			return find_option(map_old_guc_names[i+1]);
 	}
 
-	/* Check if the name is qualified, and if so, check if the qualifier
+	/*
+	 * Check if the name is qualified, and if so, check if the qualifier
 	 * maps to a custom variable class.
 	 */
 	dot = strchr(name, GUC_QUALIFIER_SEPARATOR);
 	if(dot != NULL && is_custom_class(name, dot - name))
-		/*
-		 * Add a placeholder variable for this name
-		 */
+		/* Add a placeholder variable for this name */
 		return (struct config_generic*)add_placeholder_variable(name);
 
 	/* Unknown name */
@@ -2081,9 +2098,9 @@ InitializeGUCOptions(void)
 
 		gconf->status = 0;
 		gconf->reset_source = PGC_S_DEFAULT;
-		gconf->session_source = PGC_S_DEFAULT;
 		gconf->tentative_source = PGC_S_DEFAULT;
 		gconf->source = PGC_S_DEFAULT;
+		gconf->stack = NULL;
 
 		switch (gconf->vartype)
 		{
@@ -2097,7 +2114,6 @@ InitializeGUCOptions(void)
 							elog(FATAL, "failed to initialize %s to %d",
 								 conf->gen.name, (int) conf->reset_val);
 					*conf->variable = conf->reset_val;
-					conf->session_val = conf->reset_val;
 					break;
 				}
 			case PGC_INT:
@@ -2119,7 +2135,6 @@ InitializeGUCOptions(void)
 							elog(FATAL, "failed to initialize %s to %d",
 								 conf->gen.name, conf->reset_val);
 					*conf->variable = conf->reset_val;
-					conf->session_val = conf->reset_val;
 					break;
 				}
 			case PGC_REAL:
@@ -2135,7 +2150,6 @@ InitializeGUCOptions(void)
 							elog(FATAL, "failed to initialize %s to %g",
 								 conf->gen.name, conf->reset_val);
 					*conf->variable = conf->reset_val;
-					conf->session_val = conf->reset_val;
 					break;
 				}
 			case PGC_STRING:
@@ -2150,7 +2164,6 @@ InitializeGUCOptions(void)
 						   conf->assign_hook == assign_log_statement);
 					*conf->variable = NULL;
 					conf->reset_val = NULL;
-					conf->session_val = NULL;
 					conf->tentative_val = NULL;
 
 					if (conf->boot_val == NULL)
@@ -2190,7 +2203,6 @@ InitializeGUCOptions(void)
 						}
 					}
 					*conf->variable = str;
-					conf->session_val = str;
 					break;
 				}
 		}
@@ -2254,6 +2266,9 @@ ResetAllOptions(void)
 		if (gconf->source <= PGC_S_OVERRIDE)
 			continue;
 
+		/* Save old value to support transaction abort */
+		push_old_value(gconf);
+
 		switch (gconf->vartype)
 		{
 			case PGC_BOOL:
@@ -2336,8 +2351,8 @@ ResetAllOptions(void)
 						}
 					}
 
-					SET_STRING_VARIABLE(conf, str);
-					SET_STRING_TENTATIVE_VAL(conf, str);
+					set_string_field(conf, conf->variable, str);
+					set_string_field(conf, &conf->tentative_val, str);
 					conf->gen.source = conf->gen.reset_source;
 					conf->gen.tentative_source = conf->gen.reset_source;
 					conf->gen.status |= GUC_HAVE_TENTATIVE;
@@ -2353,11 +2368,93 @@ ResetAllOptions(void)
 
 
 /*
- * Do GUC processing at transaction commit or abort.
+ * push_old_value
+ *		Push previous state during first assignment to a GUC variable
+ *		within a particular transaction.
+ *
+ * We have to be willing to "back-fill" the state stack if the first
+ * assignment occurs within a subtransaction nested several levels deep.
+ * This ensures that if an intermediate transaction aborts, it will have
+ * the proper value available to restore the setting to.
+ */
+static void
+push_old_value(struct config_generic *gconf)
+{
+	int			my_level = GetCurrentTransactionNestLevel();
+	GucStack   *stack;
+
+	/* If we're not inside a transaction, do nothing */
+	if (my_level == 0)
+		return;
+
+	for (;;)
+	{
+		/* Done if we already pushed it at this nesting depth */
+		if (gconf->stack && gconf->stack->nest_level >= my_level)
+			return;
+
+		/*
+		 * We keep all the stack entries in TopTransactionContext so as to
+		 * avoid allocation problems when a subtransaction back-fills stack
+		 * entries for upper transaction levels.
+		 */
+		stack = (GucStack *) MemoryContextAlloc(TopTransactionContext,
+												sizeof(GucStack));
+
+		stack->prev = gconf->stack;
+		stack->nest_level = stack->prev ? stack->prev->nest_level + 1 : 1;
+		stack->status = gconf->status;
+		stack->tentative_source = gconf->tentative_source;
+		stack->source = gconf->source;
+
+		switch (gconf->vartype)
+		{
+			case PGC_BOOL:
+				stack->tentative_val.boolval =
+					((struct config_bool *) gconf)->tentative_val;
+				stack->value.boolval =
+					*((struct config_bool *) gconf)->variable;
+				break;
+
+			case PGC_INT:
+				stack->tentative_val.intval =
+					((struct config_int *) gconf)->tentative_val;
+				stack->value.intval =
+					*((struct config_int *) gconf)->variable;
+				break;
+
+			case PGC_REAL:
+				stack->tentative_val.realval =
+					((struct config_real *) gconf)->tentative_val;
+				stack->value.realval =
+					*((struct config_real *) gconf)->variable;
+				break;
+
+			case PGC_STRING:
+				stack->tentative_val.stringval =
+					((struct config_string *) gconf)->tentative_val;
+				stack->value.stringval =
+					*((struct config_string *) gconf)->variable;
+				break;
+		}
+
+		gconf->stack = stack;
+
+		/* Set state to indicate nothing happened yet within this level */
+		gconf->status = GUC_HAVE_STACK;
+
+		/* Ensure we remember to pop at end of xact */
+		guc_dirty = true;
+	}
+}
+
+/*
+ * Do GUC processing at transaction or subtransaction commit or abort.
  */
 void
-AtEOXact_GUC(bool isCommit)
+AtEOXact_GUC(bool isCommit, bool isSubXact)
 {
+	int			my_level;
 	int			i;
 
 	/* Quick exit if nothing's changed in this transaction */
@@ -2371,15 +2468,56 @@ AtEOXact_GUC(bool isCommit)
 		guc_string_workspace = NULL;
 	}
 
+	my_level = GetCurrentTransactionNestLevel();
+	Assert(isSubXact ? (my_level > 1) : (my_level == 1));
+
 	for (i = 0; i < num_guc_variables; i++)
 	{
 		struct config_generic *gconf = guc_variables[i];
+		int			my_status = gconf->status;
+		GucStack   *stack = gconf->stack;
+		bool		useTentative;
 		bool		changed;
 
-		/* Skip if nothing's happened to this var in this transaction */
-		if (gconf->status == 0)
+		/*
+		 * Skip if nothing's happened to this var in this transaction
+		 */
+		if (my_status == 0)
+		{
+			Assert(stack == NULL);
+			continue;
+		}
+		/* Assert that we stacked old value before changing it */
+		Assert(stack != NULL && (my_status & GUC_HAVE_STACK));
+		/* However, the last change may have been at an outer xact level */
+		if (stack->nest_level < my_level)
 			continue;
+		Assert(stack->nest_level == my_level);
+
+		/*
+		 * We will pop the stack entry.  Start by restoring outer xact status
+		 * (since we may want to modify it below).  Be careful to use
+		 * my_status to reference the inner xact status below this point...
+		 */
+		gconf->status = stack->status;
+
+		/*
+		 * We have two cases:
+		 *
+		 * If commit and HAVE_TENTATIVE, set actual value to tentative
+		 * (this is to override a SET LOCAL if one occurred later than SET).
+		 * We keep the tentative value and propagate HAVE_TENTATIVE to
+		 * the parent status, allowing the SET's effect to percolate up.
+		 * (But if we're exiting the outermost transaction, we'll drop the
+		 * HAVE_TENTATIVE bit below.)
+		 *
+		 * Otherwise, we have a transaction that aborted or executed only
+		 * SET LOCAL (or no SET at all).  In either case it should have no
+		 * further effect, so restore both tentative and actual values from
+		 * the stack entry.
+		 */
 
+		useTentative = isCommit && (my_status & GUC_HAVE_TENTATIVE) != 0;
 		changed = false;
 
 		switch (gconf->vartype)
@@ -2387,126 +2525,190 @@ AtEOXact_GUC(bool isCommit)
 			case PGC_BOOL:
 				{
 					struct config_bool *conf = (struct config_bool *) gconf;
+					bool		newval;
+					GucSource	newsource;
 
-					if (isCommit && (conf->gen.status & GUC_HAVE_TENTATIVE))
+					if (useTentative)
+					{
+						newval = conf->tentative_val;
+						newsource = conf->gen.tentative_source;
+						conf->gen.status |= GUC_HAVE_TENTATIVE;
+					}
+					else
 					{
-						conf->session_val = conf->tentative_val;
-						conf->gen.session_source = conf->gen.tentative_source;
+						newval = stack->value.boolval;
+						newsource = stack->source;
+						conf->tentative_val = stack->tentative_val.boolval;
+						conf->gen.tentative_source = stack->tentative_source;
 					}
 
-					if (*conf->variable != conf->session_val)
+					if (*conf->variable != newval)
 					{
 						if (conf->assign_hook)
-							if (!(*conf->assign_hook) (conf->session_val,
+							if (!(*conf->assign_hook) (newval,
 													   true, PGC_S_OVERRIDE))
 								elog(LOG, "failed to commit %s",
 									 conf->gen.name);
-						*conf->variable = conf->session_val;
+						*conf->variable = newval;
 						changed = true;
 					}
-					conf->gen.source = conf->gen.session_source;
-					conf->gen.status = 0;
+					conf->gen.source = newsource;
 					break;
 				}
 			case PGC_INT:
 				{
 					struct config_int *conf = (struct config_int *) gconf;
+					int			newval;
+					GucSource	newsource;
 
-					if (isCommit && (conf->gen.status & GUC_HAVE_TENTATIVE))
+					if (useTentative)
+					{
+						newval = conf->tentative_val;
+						newsource = conf->gen.tentative_source;
+						conf->gen.status |= GUC_HAVE_TENTATIVE;
+					}
+					else
 					{
-						conf->session_val = conf->tentative_val;
-						conf->gen.session_source = conf->gen.tentative_source;
+						newval = stack->value.intval;
+						newsource = stack->source;
+						conf->tentative_val = stack->tentative_val.intval;
+						conf->gen.tentative_source = stack->tentative_source;
 					}
 
-					if (*conf->variable != conf->session_val)
+					if (*conf->variable != newval)
 					{
 						if (conf->assign_hook)
-							if (!(*conf->assign_hook) (conf->session_val,
+							if (!(*conf->assign_hook) (newval,
 													   true, PGC_S_OVERRIDE))
 								elog(LOG, "failed to commit %s",
 									 conf->gen.name);
-						*conf->variable = conf->session_val;
+						*conf->variable = newval;
 						changed = true;
 					}
-					conf->gen.source = conf->gen.session_source;
-					conf->gen.status = 0;
+					conf->gen.source = newsource;
 					break;
 				}
 			case PGC_REAL:
 				{
 					struct config_real *conf = (struct config_real *) gconf;
+					double		newval;
+					GucSource	newsource;
 
-					if (isCommit && (conf->gen.status & GUC_HAVE_TENTATIVE))
+					if (useTentative)
 					{
-						conf->session_val = conf->tentative_val;
-						conf->gen.session_source = conf->gen.tentative_source;
+						newval = conf->tentative_val;
+						newsource = conf->gen.tentative_source;
+						conf->gen.status |= GUC_HAVE_TENTATIVE;
+					}
+					else
+					{
+						newval = stack->value.realval;
+						newsource = stack->source;
+						conf->tentative_val = stack->tentative_val.realval;
+						conf->gen.tentative_source = stack->tentative_source;
 					}
 
-					if (*conf->variable != conf->session_val)
+					if (*conf->variable != newval)
 					{
 						if (conf->assign_hook)
-							if (!(*conf->assign_hook) (conf->session_val,
+							if (!(*conf->assign_hook) (newval,
 													   true, PGC_S_OVERRIDE))
 								elog(LOG, "failed to commit %s",
 									 conf->gen.name);
-						*conf->variable = conf->session_val;
+						*conf->variable = newval;
 						changed = true;
 					}
-					conf->gen.source = conf->gen.session_source;
-					conf->gen.status = 0;
+					conf->gen.source = newsource;
 					break;
 				}
 			case PGC_STRING:
 				{
 					struct config_string *conf = (struct config_string *) gconf;
+					char	   *newval;
+					GucSource	newsource;
 
-					if (isCommit && (conf->gen.status & GUC_HAVE_TENTATIVE))
+					if (useTentative)
 					{
-						SET_STRING_SESSION_VAL(conf, conf->tentative_val);
-						conf->gen.session_source = conf->gen.tentative_source;
-						conf->tentative_val = NULL;		/* transfer ownership */
+						newval = conf->tentative_val;
+						newsource = conf->gen.tentative_source;
+						conf->gen.status |= GUC_HAVE_TENTATIVE;
 					}
 					else
-						SET_STRING_TENTATIVE_VAL(conf, NULL);
-
-					if (*conf->variable != conf->session_val)
 					{
-						char	   *str = conf->session_val;
+						newval = stack->value.stringval;
+						newsource = stack->source;
+						set_string_field(conf, &conf->tentative_val,
+										 stack->tentative_val.stringval);
+						conf->gen.tentative_source = stack->tentative_source;
+					}
 
+					if (*conf->variable != newval)
+					{
 						if (conf->assign_hook)
 						{
 							const char *newstr;
 
-							newstr = (*conf->assign_hook) (str, true,
+							newstr = (*conf->assign_hook) (newval, true,
 														   PGC_S_OVERRIDE);
 							if (newstr == NULL)
 								elog(LOG, "failed to commit %s",
 									 conf->gen.name);
-							else if (newstr != str)
+							else if (newstr != newval)
 							{
 								/*
+								 * If newval should now be freed, it'll be
+								 * taken care of below.
+								 *
 								 * See notes in set_config_option about
 								 * casting
 								 */
-								str = (char *) newstr;
-								SET_STRING_SESSION_VAL(conf, str);
+								newval = (char *) newstr;
 							}
 						}
 
-						SET_STRING_VARIABLE(conf, str);
+						set_string_field(conf, conf->variable, newval);
 						changed = true;
 					}
-					conf->gen.source = conf->gen.session_source;
-					conf->gen.status = 0;
+					conf->gen.source = newsource;
+					/* Release stacked values if not used anymore */
+					set_string_field(conf, &stack->value.stringval,
+									 NULL);
+					set_string_field(conf, &stack->tentative_val.stringval,
+									 NULL);
+					/* Don't store tentative value separately after commit */
+					if (!isSubXact)
+						set_string_field(conf, &conf->tentative_val, NULL);
 					break;
 				}
 		}
 
+		/* Finish popping the state stack */
+		gconf->stack = stack->prev;
+		pfree(stack);
+
+		/*
+		 * If we're now out of all xact levels, forget TENTATIVE status bit;
+		 * there's nothing tentative about the value anymore.
+		 */
+		if (!isSubXact)
+		{
+			Assert(gconf->stack == NULL);
+			gconf->status = 0;
+		}
+
+		/* Report new value if we changed it */
 		if (changed && (gconf->flags & GUC_REPORT))
 			ReportGUCOption(gconf);
 	}
 
-	guc_dirty = false;
+	/*
+	 * If we're now out of all xact levels, we can clear guc_dirty.
+	 * (Note: we cannot reset guc_dirty when exiting a subtransaction,
+	 * because we know that all outer transaction levels will have stacked
+	 * values to deal with.)
+	 */
+	if (!isSubXact)
+		guc_dirty = false;
 }
 
 
@@ -2810,7 +3012,7 @@ set_config_option(const char *name, const char *value,
 	}
 
 	/*
-	 * Should we set reset/session values?	(If so, the behavior is not
+	 * Should we set reset/stacked values?	(If so, the behavior is not
 	 * transactional.)
 	 */
 	makeDefault = changeVal && (source <= PGC_S_OVERRIDE) && (value != NULL);
@@ -2820,7 +3022,7 @@ set_config_option(const char *name, const char *value,
 	 * However, if changeVal is false then plow ahead anyway since we are
 	 * trying to find out if the value is potentially good, not actually
 	 * use it. Also keep going if makeDefault is true, since we may want
-	 * to set the reset/session values even if we can't set the variable
+	 * to set the reset/stacked values even if we can't set the variable
 	 * itself.
 	 */
 	if (record->source > source)
@@ -2901,6 +3103,9 @@ set_config_option(const char *name, const char *value,
 
 				if (changeVal || makeDefault)
 				{
+					/* Save old value to support transaction abort */
+					if (!makeDefault)
+						push_old_value(&conf->gen);
 					if (changeVal)
 					{
 						*conf->variable = newval;
@@ -2908,15 +3113,20 @@ set_config_option(const char *name, const char *value,
 					}
 					if (makeDefault)
 					{
+						GucStack *stack;
+
 						if (conf->gen.reset_source <= source)
 						{
 							conf->reset_val = newval;
 							conf->gen.reset_source = source;
 						}
-						if (conf->gen.session_source <= source)
+						for (stack = conf->gen.stack; stack; stack = stack->prev)
 						{
-							conf->session_val = newval;
-							conf->gen.session_source = source;
+							if (stack->source <= source)
+							{
+								stack->value.boolval = newval;
+								stack->source = source;
+							}
 						}
 					}
 					else if (isLocal)
@@ -3006,6 +3216,9 @@ set_config_option(const char *name, const char *value,
 
 				if (changeVal || makeDefault)
 				{
+					/* Save old value to support transaction abort */
+					if (!makeDefault)
+						push_old_value(&conf->gen);
 					if (changeVal)
 					{
 						*conf->variable = newval;
@@ -3013,15 +3226,20 @@ set_config_option(const char *name, const char *value,
 					}
 					if (makeDefault)
 					{
+						GucStack *stack;
+
 						if (conf->gen.reset_source <= source)
 						{
 							conf->reset_val = newval;
 							conf->gen.reset_source = source;
 						}
-						if (conf->gen.session_source <= source)
+						for (stack = conf->gen.stack; stack; stack = stack->prev)
 						{
-							conf->session_val = newval;
-							conf->gen.session_source = source;
+							if (stack->source <= source)
+							{
+								stack->value.intval = newval;
+								stack->source = source;
+							}
 						}
 					}
 					else if (isLocal)
@@ -3101,6 +3319,9 @@ set_config_option(const char *name, const char *value,
 
 				if (changeVal || makeDefault)
 				{
+					/* Save old value to support transaction abort */
+					if (!makeDefault)
+						push_old_value(&conf->gen);
 					if (changeVal)
 					{
 						*conf->variable = newval;
@@ -3108,15 +3329,20 @@ set_config_option(const char *name, const char *value,
 					}
 					if (makeDefault)
 					{
+						GucStack *stack;
+
 						if (conf->gen.reset_source <= source)
 						{
 							conf->reset_val = newval;
 							conf->gen.reset_source = source;
 						}
-						if (conf->gen.session_source <= source)
+						for (stack = conf->gen.stack; stack; stack = stack->prev)
 						{
-							conf->session_val = newval;
-							conf->gen.session_source = source;
+							if (stack->source <= source)
+							{
+								stack->value.realval = newval;
+								stack->source = source;
+							}
 						}
 					}
 					else if (isLocal)
@@ -3261,27 +3487,34 @@ set_config_option(const char *name, const char *value,
 
 				if (changeVal || makeDefault)
 				{
+					/* Save old value to support transaction abort */
+					if (!makeDefault)
+						push_old_value(&conf->gen);
 					if (changeVal)
 					{
-						SET_STRING_VARIABLE(conf, newval);
+						set_string_field(conf, conf->variable, newval);
 						conf->gen.source = source;
 					}
 					if (makeDefault)
 					{
+						GucStack *stack;
+
 						if (conf->gen.reset_source <= source)
 						{
-							SET_STRING_RESET_VAL(conf, newval);
+							set_string_field(conf, &conf->reset_val, newval);
 							conf->gen.reset_source = source;
 						}
-						if (conf->gen.session_source <= source)
+						for (stack = conf->gen.stack; stack; stack = stack->prev)
 						{
-							SET_STRING_SESSION_VAL(conf, newval);
-							conf->gen.session_source = source;
+							if (stack->source <= source)
+							{
+								set_string_field(conf, &stack->value.stringval,
+												 newval);
+								stack->source = source;
+							}
 						}
 						/* Perhaps we didn't install newval anywhere */
-						if (newval != *conf->variable &&
-							newval != conf->session_val &&
-							newval != conf->reset_val)
+						if (!string_field_used(conf, newval))
 							free(newval);
 					}
 					else if (isLocal)
@@ -3291,7 +3524,7 @@ set_config_option(const char *name, const char *value,
 					}
 					else
 					{
-						SET_STRING_TENTATIVE_VAL(conf, newval);
+						set_string_field(conf, &conf->tentative_val, newval);
 						conf->gen.tentative_source = source;
 						conf->gen.status |= GUC_HAVE_TENTATIVE;
 						guc_dirty = true;
@@ -3608,44 +3841,36 @@ define_custom_variable(struct config_generic* variable)
 	/* This better be a placeholder
 	 */
 	if(((*res)->flags & GUC_CUSTOM_PLACEHOLDER) == 0)
-	{
 		ereport(ERROR,
 				(errcode(ERRCODE_INTERNAL_ERROR),
 				 errmsg("attempt to redefine parameter \"%s\"", name)));
-	}
-	pHolder = (struct config_string*)*res;
+
+	Assert((*res)->vartype == PGC_STRING);
+	pHolder = (struct config_string*) *res;
 	
-	/* We have the same name, no sorting is necessary.
-	 */
+	/* We have the same name, no sorting is necessary */
 	*res = variable;
 
 	value = *pHolder->variable;
 
-	/* Assign the variable stored in the placeholder to the real
-	 * variable.
+	/*
+	 * Assign the string value stored in the placeholder to the real variable.
+	 *
+	 * XXX this is not really good enough --- it should be a nontransactional
+	 * assignment, since we don't want it to roll back if the current xact
+	 * fails later.
 	 */
 	set_config_option(name, value,
 				  pHolder->gen.context, pHolder->gen.source,
 				  false, true);
 
-	/* Free up stuff occupied by the placeholder variable
+	/*
+	 * Free up as much as we conveniently can of the placeholder structure
+	 * (this neglects any stack items...)
 	 */
-	if(value != NULL)
-		free((void*)value);
-
-	if(pHolder->reset_val != NULL && pHolder->reset_val != value)
-		free(pHolder->reset_val);
-
-	if(pHolder->session_val != NULL
-	&& pHolder->session_val != value
-	&& pHolder->session_val != pHolder->reset_val)
-		free(pHolder->session_val);
-
-	if(pHolder->tentative_val != NULL
-	&& pHolder->tentative_val != value
-	&& pHolder->tentative_val != pHolder->reset_val
-	&& pHolder->tentative_val != pHolder->session_val)
-		free(pHolder->tentative_val);
+	set_string_field(pHolder, pHolder->variable, NULL);
+	set_string_field(pHolder, &pHolder->reset_val, NULL);
+	set_string_field(pHolder, &pHolder->tentative_val, NULL);
 
 	free(pHolder);
 }
@@ -3754,7 +3979,7 @@ void DefineCustomStringVariable(
 	define_custom_variable(&var->gen);
 }
 
-extern void EmittWarningsOnPlaceholders(const char* className)
+extern void EmitWarningsOnPlaceholders(const char* className)
 {
 	struct config_generic** vars = guc_variables;
 	struct config_generic** last = vars + num_guc_variables;
@@ -5133,5 +5358,14 @@ assign_log_stats(bool newval, bool doit, GucSource source)
 	return true;
 }
 
+static bool
+assign_transaction_read_only(bool newval, bool doit, GucSource source)
+{
+	if (doit && source >= PGC_S_INTERACTIVE && IsSubTransaction())
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("cannot set transaction read only mode inside a subtransaction")));
+	return true;
+}
 
 #include "guc-file.c"
diff --git a/src/backend/utils/mmgr/README b/src/backend/utils/mmgr/README
index f705827c1b6..490b781cc91 100644
--- a/src/backend/utils/mmgr/README
+++ b/src/backend/utils/mmgr/README
@@ -1,4 +1,4 @@
-$PostgreSQL: pgsql/src/backend/utils/mmgr/README,v 1.6 2004/06/05 19:48:09 tgl Exp $
+$PostgreSQL: pgsql/src/backend/utils/mmgr/README,v 1.7 2004/07/01 00:51:29 tgl Exp $
 
 Notes about memory allocation redesign
 --------------------------------------
@@ -90,7 +90,7 @@ context managers as discussed below.
 We could even consider getting rid of CurrentMemoryContext entirely,
 instead requiring the target memory context for allocation to be specified
 explicitly.  But I think that would be too much notational overhead ---
-we'd have to pass an apppropriate memory context to called routines in
+we'd have to pass an appropriate memory context to called routines in
 many places.  For example, the copyObject routines would need to be passed
 a context, as would function execution routines that return a
 pass-by-reference datatype.  And what of routines that temporarily
@@ -176,15 +176,30 @@ is kept separate from per-transaction and per-portal contexts because a
 query string might need to live either a longer or shorter time than any
 single transaction or portal.
 
-TopTransactionContext --- this holds everything that lives until end of
-transaction (longer than one statement within a transaction!).  An example
-of what has to be here is the list of pending NOTIFY messages to be sent
-at xact commit.  This context will be reset, and all its children deleted,
-at conclusion of each transaction cycle.  Note: this context is NOT
-cleared immediately upon error; its contents will survive until the
-transaction block is exited by COMMIT/ROLLBACK.
-(If we ever implement nested transactions, TopTransactionContext may need
-to be split into a true "top" pointer and a "current transaction" pointer.)
+TopTransactionContext --- this holds everything that lives until end of the
+top-level transaction.  This context will be reset, and all its children
+deleted, at conclusion of each top-level transaction cycle.  In most cases
+you don't want to allocate stuff directly here, but in CurTransactionContext;
+what does belong here is control information that exists explicitly to manage
+status across multiple subtransactions.  Note: this context is NOT cleared
+immediately upon error; its contents will survive until the transaction block
+is exited by COMMIT/ROLLBACK.
+
+CurTransactionContext --- this holds data that has to survive until the end
+of the current transaction, and in particular will be needed at top-level
+transaction commit.  When we are in a top-level transaction this is the same
+as TopTransactionContext, but in subtransactions it points to a child context.
+It is important to understand that if a subtransaction aborts, its
+CurTransactionContext is thrown away after finishing the abort processing;
+but a committed subtransaction's CurTransactionContext is kept until top-level
+commit (unless of course one of the intermediate levels of subtransaction
+aborts).  This ensures that we do not keep data from a failed subtransaction
+longer than necessary.  Because of this behavior, you must be careful to clean
+up properly during subtransaction abort --- the subtransaction's state must be
+delinked from any pointers or lists kept in upper transactions, or you will
+have dangling pointers leading to a crash at top-level commit.  An example of
+data kept here is pending NOTIFY messages, which are sent at top-level commit,
+but only if the generating subtransaction did not abort.
 
 QueryContext --- this is not actually a separate context, but a global
 variable pointing to the context that holds the current command's parse
diff --git a/src/backend/utils/mmgr/mcxt.c b/src/backend/utils/mmgr/mcxt.c
index c444886e140..96ffb1a8e1c 100644
--- a/src/backend/utils/mmgr/mcxt.c
+++ b/src/backend/utils/mmgr/mcxt.c
@@ -14,7 +14,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/utils/mmgr/mcxt.c,v 1.45 2004/06/05 19:48:09 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/mmgr/mcxt.c,v 1.46 2004/07/01 00:51:29 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -45,6 +45,7 @@ MemoryContext PostmasterContext = NULL;
 MemoryContext CacheMemoryContext = NULL;
 MemoryContext MessageContext = NULL;
 MemoryContext TopTransactionContext = NULL;
+MemoryContext CurTransactionContext = NULL;
 
 /* These two are transient links to contexts owned by other objects: */
 MemoryContext QueryContext = NULL;
diff --git a/src/backend/utils/mmgr/portalmem.c b/src/backend/utils/mmgr/portalmem.c
index f77125cebf5..466b2fc97bf 100644
--- a/src/backend/utils/mmgr/portalmem.c
+++ b/src/backend/utils/mmgr/portalmem.c
@@ -12,7 +12,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/utils/mmgr/portalmem.c,v 1.65 2004/05/30 23:40:39 neilc Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/mmgr/portalmem.c,v 1.66 2004/07/01 00:51:29 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -511,3 +511,94 @@ AtCleanup_Portals(void)
 		PortalDrop(portal, true);
 	}
 }
+
+/*
+ * Pre-subcommit processing for portals.
+ *
+ * Reassign the portals created in the current subtransaction to the parent
+ * transaction.  (XXX perhaps we should reassign only holdable cursors,
+ * and drop the rest?)
+ */
+void
+AtSubCommit_Portals(TransactionId parentXid)
+{
+	HASH_SEQ_STATUS status;
+	PortalHashEnt *hentry;
+	TransactionId curXid = GetCurrentTransactionId();
+
+	hash_seq_init(&status, PortalHashTable);
+
+	while ((hentry = (PortalHashEnt *) hash_seq_search(&status)) != NULL)
+	{
+		Portal	portal = hentry->portal;
+
+		if (portal->createXact == curXid)
+			portal->createXact = parentXid;
+	}
+}
+
+/*
+ * Subtransaction abort handling for portals.
+ *
+ * Deactivate all portals created during the failed subtransaction.
+ * Note that per AtSubCommit_Portals, this will catch portals created
+ * in descendants of the subtransaction too.
+ */
+void
+AtSubAbort_Portals(void)
+{
+	HASH_SEQ_STATUS status;
+	PortalHashEnt *hentry;
+	TransactionId curXid = GetCurrentTransactionId();
+
+	hash_seq_init(&status, PortalHashTable);
+
+	while ((hentry = (PortalHashEnt *) hash_seq_search(&status)) != NULL)
+	{
+		Portal	portal = hentry->portal;
+
+		if (portal->createXact != curXid)
+			continue;
+
+		portal->portalActive = false;
+
+		/* let portalcmds.c clean up the state it knows about */
+		if (PointerIsValid(portal->cleanup))
+		{
+			(*portal->cleanup) (portal, true);
+			portal->cleanup = NULL;
+		}
+	}
+}
+
+/*
+ * Post-subabort cleanup for portals.
+ *
+ * Drop all portals created in the finishing subtransaction and all
+ * its descendants.
+ */
+void
+AtSubCleanup_Portals(void)
+{
+	HASH_SEQ_STATUS status;
+	PortalHashEnt *hentry;
+	TransactionId curXid = GetCurrentTransactionId();
+
+	hash_seq_init(&status, PortalHashTable);
+
+	while ((hentry = (PortalHashEnt *) hash_seq_search(&status)) != NULL)
+	{
+		Portal		portal = hentry->portal;
+
+		if (portal->createXact != curXid)
+			continue;
+
+		/*
+		 * Let's just make sure no one's active...
+		 */
+		portal->portalActive = false;
+
+		/* Zap it with prejudice. */
+		PortalDrop(portal, true);
+	}
+}
diff --git a/src/backend/utils/time/tqual.c b/src/backend/utils/time/tqual.c
index a56e59a3d65..446ee4b72c5 100644
--- a/src/backend/utils/time/tqual.c
+++ b/src/backend/utils/time/tqual.c
@@ -16,13 +16,14 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/utils/time/tqual.c,v 1.72 2003/11/29 19:52:04 pgsql Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/time/tqual.c,v 1.73 2004/07/01 00:51:33 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
 
 #include "postgres.h"
 
+#include "access/subtrans.h"
 #include "storage/sinval.h"
 #include "utils/tqual.h"
 
@@ -115,6 +116,10 @@ HeapTupleSatisfiesItself(HeapTupleHeader tuple)
 			if (tuple->t_infomask & HEAP_XMAX_INVALID)	/* xid invalid */
 				return true;
 
+			/* deleting subtransaction aborted */
+			if (TransactionIdDidAbort(HeapTupleHeaderGetXmax(tuple)))
+				return true;
+
 			Assert(TransactionIdIsCurrentTransactionId(HeapTupleHeaderGetXmax(tuple)));
 
 			if (tuple->t_infomask & HEAP_MARKED_FOR_UPDATE)
@@ -261,6 +266,10 @@ HeapTupleSatisfiesNow(HeapTupleHeader tuple)
 			if (tuple->t_infomask & HEAP_XMAX_INVALID)	/* xid invalid */
 				return true;
 
+			/* deleting subtransaction aborted */
+			if (TransactionIdDidAbort(HeapTupleHeaderGetXmax(tuple)))
+				return true;
+
 			Assert(TransactionIdIsCurrentTransactionId(HeapTupleHeaderGetXmax(tuple)));
 
 			if (tuple->t_infomask & HEAP_MARKED_FOR_UPDATE)
@@ -441,6 +450,10 @@ HeapTupleSatisfiesUpdate(HeapTupleHeader tuple, CommandId curcid)
 			if (tuple->t_infomask & HEAP_XMAX_INVALID)	/* xid invalid */
 				return HeapTupleMayBeUpdated;
 
+			/* deleting subtransaction aborted */
+			if (TransactionIdDidAbort(HeapTupleHeaderGetXmax(tuple)))
+				return HeapTupleMayBeUpdated;
+
 			Assert(TransactionIdIsCurrentTransactionId(HeapTupleHeaderGetXmax(tuple)));
 
 			if (tuple->t_infomask & HEAP_MARKED_FOR_UPDATE)
@@ -575,6 +588,10 @@ HeapTupleSatisfiesDirty(HeapTupleHeader tuple)
 			if (tuple->t_infomask & HEAP_XMAX_INVALID)	/* xid invalid */
 				return true;
 
+			/* deleting subtransaction aborted */
+			if (TransactionIdDidAbort(HeapTupleHeaderGetXmax(tuple)))
+				return true;
+
 			Assert(TransactionIdIsCurrentTransactionId(HeapTupleHeaderGetXmax(tuple)));
 
 			if (tuple->t_infomask & HEAP_MARKED_FOR_UPDATE)
@@ -712,6 +729,11 @@ HeapTupleSatisfiesSnapshot(HeapTupleHeader tuple, Snapshot snapshot)
 			if (tuple->t_infomask & HEAP_XMAX_INVALID)	/* xid invalid */
 				return true;
 
+			/* deleting subtransaction aborted */
+			/* FIXME -- is this correct w.r.t. the cmax of the tuple? */
+			if (TransactionIdDidAbort(HeapTupleHeaderGetXmax(tuple)))
+				return true;
+
 			Assert(TransactionIdIsCurrentTransactionId(HeapTupleHeaderGetXmax(tuple)));
 
 			if (tuple->t_infomask & HEAP_MARKED_FOR_UPDATE)
@@ -747,7 +769,7 @@ HeapTupleSatisfiesSnapshot(HeapTupleHeader tuple, Snapshot snapshot)
 
 		for (i = 0; i < snapshot->xcnt; i++)
 		{
-			if (TransactionIdEquals(HeapTupleHeaderGetXmin(tuple),
+			if (SubTransXidsHaveCommonAncestor(HeapTupleHeaderGetXmin(tuple),
 									snapshot->xip[i]))
 				return false;
 		}
@@ -792,7 +814,7 @@ HeapTupleSatisfiesSnapshot(HeapTupleHeader tuple, Snapshot snapshot)
 			return true;
 		for (i = 0; i < snapshot->xcnt; i++)
 		{
-			if (TransactionIdEquals(HeapTupleHeaderGetXmax(tuple), snapshot->xip[i]))
+			if (SubTransXidsHaveCommonAncestor(HeapTupleHeaderGetXmax(tuple), snapshot->xip[i]))
 				return true;
 		}
 	}
@@ -868,8 +890,8 @@ HeapTupleSatisfiesVacuum(HeapTupleHeader tuple, TransactionId OldestXmin)
 		{
 			if (tuple->t_infomask & HEAP_XMAX_INVALID)	/* xid invalid */
 				return HEAPTUPLE_INSERT_IN_PROGRESS;
-			Assert(HeapTupleHeaderGetXmin(tuple) ==
-				   HeapTupleHeaderGetXmax(tuple));
+			Assert(SubTransXidsHaveCommonAncestor(HeapTupleHeaderGetXmin(tuple),
+						HeapTupleHeaderGetXmax(tuple)));
 			if (tuple->t_infomask & HEAP_MARKED_FOR_UPDATE)
 				return HEAPTUPLE_INSERT_IN_PROGRESS;
 			/* inserted and then deleted by same xact */
@@ -943,7 +965,7 @@ HeapTupleSatisfiesVacuum(HeapTupleHeader tuple, TransactionId OldestXmin)
 	 * Deleter committed, but check special cases.
 	 */
 
-	if (TransactionIdEquals(HeapTupleHeaderGetXmin(tuple),
+	if (SubTransXidsHaveCommonAncestor(HeapTupleHeaderGetXmin(tuple),
 							HeapTupleHeaderGetXmax(tuple)))
 	{
 		/*