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.

Author: tglsfdc

contrib/userlock/user_locks.c

@@ -75,8 +75,7 @@ user_write_unlock_oid(Oid oid)
 int
 user_unlock_all(void)
 {
-	return LockReleaseAll(USER_LOCKMETHOD, MyProc, false,
-						  InvalidTransactionId);
+	return LockReleaseAll(USER_LOCKMETHOD, MyProc, ReleaseAll, 0, NULL);
 }
 
 /* end of file */

src/backend/access/gist/gistscan.c

@@ -8,7 +8,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/access/gist/gistscan.c,v 1.51 2004/01/07 18:56:23 neilc Exp $
+ *	  $PostgreSQL: pgsql/src/backend/access/gist/gistscan.c,v 1.52 2004/07/01 00:49:27 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -41,6 +41,7 @@ static void adjustiptr(IndexScanDesc s, ItemPointer iptr,
 typedef struct GISTScanListData
 {
 	IndexScanDesc gsl_scan;
+	TransactionId gsl_creatingXid;
 	struct GISTScanListData *gsl_next;
 } GISTScanListData;
 
@@ -223,6 +224,7 @@ gistregscan(IndexScanDesc s)
 
 	l = (GISTScanList) palloc(sizeof(GISTScanListData));
 	l->gsl_scan = s;
+	l->gsl_creatingXid = GetCurrentTransactionId();
 	l->gsl_next = GISTScans;
 	GISTScans = l;
 }
@@ -271,6 +273,46 @@ AtEOXact_gist(void)
 	GISTScans = NULL;
 }
 
+/*
+ * AtEOSubXact_gist() --- clean up gist subsystem at subxact abort or commit.
+ *
+ * This is here because it needs to touch this module's static var GISTScans.
+ */
+void
+AtEOSubXact_gist(TransactionId childXid)
+{
+	GISTScanList l;
+	GISTScanList prev;
+	GISTScanList next;
+
+	/*
+	 * Note: these actions should only be necessary during xact abort; but
+	 * they can't hurt during a commit.
+	 */
+
+	/*
+	 * Forget active scans that were started in this subtransaction.
+	 */
+	prev = NULL;
+
+	for (l = GISTScans; l != NULL; l = next)
+	{
+		next = l->gsl_next;
+		if (l->gsl_creatingXid == childXid)
+		{
+			if (prev == NULL)
+				GISTScans = next;
+			else
+				prev->gsl_next = next;
+
+			pfree(l);
+			/* prev does not change */
+		}
+		else
+			prev = l;
+	}
+}
+
 void
 gistadjscans(Relation rel, int op, BlockNumber blkno, OffsetNumber offnum)
 {

src/backend/access/hash/hashscan.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/access/hash/hashscan.c,v 1.33 2004/01/07 18:56:23 neilc Exp $
+ *	  $PostgreSQL: pgsql/src/backend/access/hash/hashscan.c,v 1.34 2004/07/01 00:49:29 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -21,6 +21,7 @@
 typedef struct HashScanListData
 {
 	IndexScanDesc hashsl_scan;
+	TransactionId hashsl_creatingXid;
 	struct HashScanListData *hashsl_next;
 } HashScanListData;
 
@@ -50,6 +51,46 @@ AtEOXact_hash(void)
 	HashScans = NULL;
 }
 
+/*
+ * AtEOSubXact_hash() --- clean up hash subsystem at subxact abort or commit.
+ *
+ * This is here because it needs to touch this module's static var HashScans.
+ */
+void
+AtEOSubXact_hash(TransactionId childXid)
+{
+	HashScanList l;
+	HashScanList prev;
+	HashScanList next;
+
+	/*
+	 * Note: these actions should only be necessary during xact abort; but
+	 * they can't hurt during a commit.
+	 */
+
+	/*
+	 * Forget active scans that were started in this subtransaction.
+	 */
+	prev = NULL;
+
+	for (l = HashScans; l != NULL; l = next)
+	{
+		next = l->hashsl_next;
+		if (l->hashsl_creatingXid == childXid)
+		{
+			if (prev == NULL)
+				HashScans = next;
+			else
+				prev->hashsl_next = next;
+
+			pfree(l);
+			/* prev does not change */
+		}
+		else
+			prev = l;
+	}
+}
+
 /*
  *	_Hash_regscan() -- register a new scan.
  */
@@ -60,6 +101,7 @@ _hash_regscan(IndexScanDesc scan)
 
 	new_el = (HashScanList) palloc(sizeof(HashScanListData));
 	new_el->hashsl_scan = scan;
+	new_el->hashsl_creatingXid = GetCurrentTransactionId();
 	new_el->hashsl_next = HashScans;
 	HashScans = new_el;
 }

src/backend/access/rtree/rtscan.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/access/rtree/rtscan.c,v 1.51 2004/01/07 18:56:24 neilc Exp $
+ *	  $PostgreSQL: pgsql/src/backend/access/rtree/rtscan.c,v 1.52 2004/07/01 00:49:31 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -42,6 +42,7 @@ static void adjustiptr(IndexScanDesc s, ItemPointer iptr,
 typedef struct RTScanListData
 {
 	IndexScanDesc rtsl_scan;
+	TransactionId rtsl_creatingXid;
 	struct RTScanListData *rtsl_next;
 } RTScanListData;
 
@@ -240,6 +241,7 @@ rtregscan(IndexScanDesc s)
 
 	l = (RTScanList) palloc(sizeof(RTScanListData));
 	l->rtsl_scan = s;
+	l->rtsl_creatingXid = GetCurrentTransactionId();
 	l->rtsl_next = RTScans;
 	RTScans = l;
 }
@@ -290,6 +292,46 @@ AtEOXact_rtree(void)
 	RTScans = NULL;
 }
 
+/*
+ * AtEOSubXact_rtree() --- clean up rtree subsystem at subxact abort or commit.
+ *
+ * This is here because it needs to touch this module's static var RTScans.
+ */
+void
+AtEOSubXact_rtree(TransactionId childXid)
+{
+	RTScanList l;
+	RTScanList prev;
+	RTScanList next;
+
+	/*
+	 * Note: these actions should only be necessary during xact abort; but
+	 * they can't hurt during a commit.
+	 */
+
+	/*
+	 * Forget active scans that were started in this subtransaction.
+	 */
+	prev = NULL;
+
+	for (l = RTScans; l != NULL; l = next)
+	{
+		next = l->rtsl_next;
+		if (l->rtsl_creatingXid == childXid)
+		{
+			if (prev == NULL)
+				RTScans = next;
+			else
+				prev->rtsl_next = next;
+
+			pfree(l);
+			/* prev does not change */
+		}
+		else
+			prev = l;
+	}
+}
+
 void
 rtadjscans(Relation r, int op, BlockNumber blkno, OffsetNumber offnum)
 {

src/backend/access/transam/Makefile

@@ -4,15 +4,15 @@
 #    Makefile for access/transam
 #
 # IDENTIFICATION
-#    $PostgreSQL: pgsql/src/backend/access/transam/Makefile,v 1.18 2003/11/29 19:51:40 pgsql Exp $
+#    $PostgreSQL: pgsql/src/backend/access/transam/Makefile,v 1.19 2004/07/01 00:49:42 tgl Exp $
 #
 #-------------------------------------------------------------------------
 
 subdir = src/backend/access/transam
 top_builddir = ../../../..
 include $(top_builddir)/src/Makefile.global
 
-OBJS = clog.o transam.o varsup.o xact.o xlog.o xlogutils.o rmgr.o slru.o
+OBJS = clog.o transam.o varsup.o xact.o xlog.o xlogutils.o rmgr.o slru.o subtrans.o
 
 all: SUBSYS.o
 

src/backend/access/transam/clog.c

@@ -13,22 +13,21 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/backend/access/transam/clog.c,v 1.20 2004/05/31 03:47:54 tgl Exp $
+ * $PostgreSQL: pgsql/src/backend/access/transam/clog.c,v 1.21 2004/07/01 00:49:42 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
 #include "postgres.h"
 
 #include <fcntl.h>
 #include <dirent.h>
-#include <errno.h>
 #include <sys/stat.h>
 #include <unistd.h>
 
 #include "access/clog.h"
 #include "access/slru.h"
-#include "storage/lwlock.h"
 #include "miscadmin.h"
+#include "storage/lwlock.h"
 
 
 /*
@@ -65,7 +64,7 @@
  * is guaranteed flushed through the XLOG commit record before we are called
  * to log a commit, so the WAL rule "write xlog before data" is satisfied
  * automatically for commits, and we don't really care for aborts.  Therefore,
- * we don't need to mark XLOG pages with LSN information; we have enough
+ * we don't need to mark CLOG pages with LSN information; we have enough
  * synchronization already.
  *----------
  */
@@ -95,20 +94,22 @@ TransactionIdSetStatus(TransactionId xid, XidStatus status)
 	char	   *byteptr;
 
 	Assert(status == TRANSACTION_STATUS_COMMITTED ||
-		   status == TRANSACTION_STATUS_ABORTED);
+		   status == TRANSACTION_STATUS_ABORTED ||
+		   status == TRANSACTION_STATUS_SUB_COMMITTED);
 
 	LWLockAcquire(ClogCtl->ControlLock, LW_EXCLUSIVE);
 
 	byteptr = SimpleLruReadPage(ClogCtl, pageno, xid, true);
 	byteptr += byteno;
 
-	/* Current state should be 0 or target state */
+	/* Current state should be 0, subcommitted or target state */
 	Assert(((*byteptr >> bshift) & CLOG_XACT_BITMASK) == 0 ||
+		   ((*byteptr >> bshift) & CLOG_XACT_BITMASK) == TRANSACTION_STATUS_SUB_COMMITTED ||
 		   ((*byteptr >> bshift) & CLOG_XACT_BITMASK) == status);
 
 	*byteptr |= (status << bshift);
 
-	/* ...->page_status[slotno] = CLOG_PAGE_DIRTY; already done */
+	/* ...->page_status[slotno] = SLRU_PAGE_DIRTY; already done */
 
 	LWLockRelease(ClogCtl->ControlLock);
 }
@@ -117,7 +118,7 @@ TransactionIdSetStatus(TransactionId xid, XidStatus status)
  * Interrogate the state of a transaction in the commit log.
  *
  * NB: this is a low-level routine and is NOT the preferred entry point
- * for most uses; TransactionLogTest() in transam.c is the intended caller.
+ * for most uses; TransactionLogFetch() in transam.c is the intended caller.
  */
 XidStatus
 TransactionIdGetStatus(TransactionId xid)
@@ -176,7 +177,7 @@ BootStrapCLOG(void)
 
 	/* Make sure it's written out */
 	SimpleLruWritePage(ClogCtl, slotno, NULL);
-	/* Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN); */
+	/* Assert(ClogCtl->page_status[slotno] == SLRU_PAGE_CLEAN); */
 
 	LWLockRelease(ClogCtl->ControlLock);
 }
@@ -211,7 +212,8 @@ StartupCLOG(void)
 	/*
 	 * Initialize our idea of the latest page number.
 	 */
-	SimpleLruSetLatestPage(ClogCtl, TransactionIdToPage(ShmemVariableCache->nextXid));
+	SimpleLruSetLatestPage(ClogCtl,
+						   TransactionIdToPage(ShmemVariableCache->nextXid));
 }
 
 /*
@@ -333,51 +335,20 @@ WriteZeroPageXlogRec(int pageno)
 	rdata.data = (char *) (&pageno);
 	rdata.len = sizeof(int);
 	rdata.next = NULL;
-	(void) XLogInsert(RM_CLOG_ID, CLOG_ZEROPAGE | XLOG_NO_TRAN, &rdata);
-}
-
-/*
- * CLOG resource manager's routines
- */
-void
-clog_redo(XLogRecPtr lsn, XLogRecord *record)
-{
-	uint8		info = record->xl_info & ~XLR_INFO_MASK;
-
-	if (info == CLOG_ZEROPAGE)
-	{
-		int			pageno;
-		int			slotno;
-
-		memcpy(&pageno, XLogRecGetData(record), sizeof(int));
-
-		LWLockAcquire(ClogCtl->ControlLock, LW_EXCLUSIVE);
-
-		slotno = ZeroCLOGPage(pageno, false);
-		SimpleLruWritePage(ClogCtl, slotno, NULL);
-		/* Assert(ClogCtl->page_status[slotno] == SLRU_PAGE_CLEAN); */
-
-		LWLockRelease(ClogCtl->ControlLock);
-	}
+	(void) XLogInsert(RM_SLRU_ID, CLOG_ZEROPAGE | XLOG_NO_TRAN, &rdata);
 }
 
+/* Redo a ZEROPAGE action during WAL replay */
 void
-clog_undo(XLogRecPtr lsn, XLogRecord *record)
+clog_zeropage_redo(int pageno)
 {
-}
+	int			slotno;
 
-void
-clog_desc(char *buf, uint8 xl_info, char *rec)
-{
-	uint8		info = xl_info & ~XLR_INFO_MASK;
+	LWLockAcquire(ClogCtl->ControlLock, LW_EXCLUSIVE);
 
-	if (info == CLOG_ZEROPAGE)
-	{
-		int			pageno;
+	slotno = ZeroCLOGPage(pageno, false);
+	SimpleLruWritePage(ClogCtl, slotno, NULL);
+	/* Assert(ClogCtl->page_status[slotno] == SLRU_PAGE_CLEAN); */
 
-		memcpy(&pageno, rec, sizeof(int));
-		sprintf(buf + strlen(buf), "zeropage: %d", pageno);
-	}
-	else
-		strcat(buf, "UNKNOWN");
+	LWLockRelease(ClogCtl->ControlLock);
 }