WAL misc

7 files changed, 281 insertions, 168 deletions

diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 076fa115f04..3a43f3f772d 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/access/heap/heapam.c,v 1.90 2000/10/21 15:43:14 vadim Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/access/heap/heapam.c,v 1.91 2000/10/24 09:56:07 vadim Exp $
  *
  *
  * INTERFACE ROUTINES
@@ -93,9 +93,13 @@ void heap_redo(XLogRecPtr lsn, XLogRecord *record);
 void heap_undo(XLogRecPtr lsn, XLogRecord *record);
 void heap_desc(char *buf, uint8 xl_info, char* rec);
 
+XLogRecPtr	log_heap_move(Relation reln, ItemPointerData from, HeapTuple newtup);
+
 /* comments are in heap_update */
 static xl_heaptid	_locked_tuple_;
 static void _heap_unlock_tuple(void *data);
+static XLogRecPtr log_heap_update(Relation reln, ItemPointerData from, 
+									HeapTuple newtup, bool move);
 
 static void HeapPageCleanup(Buffer buffer);
 
@@ -1706,22 +1710,8 @@ l2:
 #ifdef XLOG
 	/* XLOG stuff */
 	{
-		xl_heap_update	xlrec;
-		XLogRecPtr		recptr;
-
-		xlrec.target.node = relation->rd_node;
-		xlrec.target.cid = GetCurrentCommandId();
-		xlrec.target.tid = oldtup.t_self;
-		xlrec.newtid = newtup->t_self;
-		xlrec.t_natts = newtup->t_data->t_natts;
-		xlrec.t_oid = newtup->t_data->t_oid;
-		xlrec.t_hoff = newtup->t_data->t_hoff;
-		xlrec.mask = newtup->t_data->t_infomask;
-		
-		recptr = XLogInsert(RM_HEAP_ID, XLOG_HEAP_UPDATE,
-			(char*) &xlrec, SizeOfHeapUpdate, 
-			(char*) newtup->t_data + offsetof(HeapTupleHeaderData, t_bits), 
-			newtup->t_len - offsetof(HeapTupleHeaderData, t_bits));
+		XLogRecPtr	recptr = log_heap_update(relation, 
+								oldtup.t_self, newtup, false);
 
 		if (newbuf != buffer)
 		{
@@ -2019,6 +2009,54 @@ heap_restrpos(HeapScanDesc scan)
 
 #ifdef XLOG
 
+static XLogRecPtr
+log_heap_update(Relation reln, ItemPointerData from, 
+	HeapTuple newtup, bool move)
+{
+	char			tbuf[sizeof(xl_heap_update) + 2 * sizeof(TransactionId)];
+	xl_heap_update *xlrec = (xl_heap_update*) tbuf;
+	int				hsize = SizeOfHeapUpdate;
+	XLogRecPtr		recptr;
+
+	xlrec->target.node = reln->rd_node;
+	xlrec->target.tid = from;
+	xlrec->newtid = newtup->t_self;
+	xlrec->t_natts = newtup->t_data->t_natts;
+	xlrec->t_oid = newtup->t_data->t_oid;
+	xlrec->t_hoff = newtup->t_data->t_hoff;
+	xlrec->mask = newtup->t_data->t_infomask;
+
+	if (move)	/* remember xmin & xmax */
+	{
+		TransactionId	xmax;
+
+		xlrec->target.cid = (CommandId) newtup->t_data->t_xmin;
+		if (newtup->t_data->t_infomask & HEAP_XMAX_INVALID ||
+			newtup->t_data->t_infomask & HEAP_MARKED_FOR_UPDATE)
+			xmax = InvalidTransactionId;
+		else
+			xmax = newtup->t_data->t_xmax;
+		memcpy(tbuf + hsize, &xmax, sizeof(TransactionId));
+		hsize += sizeof(TransactionId);
+	}
+	else
+		xlrec->target.cid = GetCurrentCommandId();
+		
+	recptr = XLogInsert(RM_HEAP_ID, 
+			(move) ? XLOG_HEAP_MOVE : XLOG_HEAP_UPDATE,
+			tbuf, hsize, 
+			(char*) newtup->t_data + offsetof(HeapTupleHeaderData, t_bits), 
+			newtup->t_len - offsetof(HeapTupleHeaderData, t_bits));
+
+	return(recptr);
+}
+
+XLogRecPtr
+log_heap_move(Relation reln, ItemPointerData from, HeapTuple newtup)
+{
+	return(log_heap_update(reln, from, newtup, true));
+}
+
 static void
 heap_xlog_delete(bool redo, XLogRecPtr lsn, XLogRecord *record)
 {
@@ -2159,6 +2197,7 @@ heap_xlog_insert(bool redo, XLogRecPtr lsn, XLogRecord *record)
 		htup->t_hoff = xlrec->t_hoff;
 		htup->t_xmin = record->xl_xid;
 		htup->t_cmin = xlrec->target.cid;
+		htup->t_xmax = htup->t_cmax = 0;
 		htup->t_infomask = HEAP_XMAX_INVALID | HEAP_XMIN_COMMITTED | xlrec->mask;
 		
 		offnum = PageAddItem(page, (Item)htup, newlen, 
@@ -2210,8 +2249,11 @@ heap_xlog_insert(bool redo, XLogRecPtr lsn, XLogRecord *record)
 
 }
 
+/*
+ * Handles UPDATE & MOVE
+ */
 static void
-heap_xlog_update(bool redo, XLogRecPtr lsn, XLogRecord *record)
+heap_xlog_update(bool redo, XLogRecPtr lsn, XLogRecord *record, bool move)
 {
 	xl_heap_update *xlrec = (xl_heap_update*) XLogRecGetData(record);
 	Relation		reln = XLogOpenRelation(redo, RM_HEAP_ID, xlrec->target.node);
@@ -2282,10 +2324,20 @@ heap_xlog_update(bool redo, XLogRecPtr lsn, XLogRecord *record)
 
 	if (redo)
 	{
-		htup->t_xmax = record->xl_xid;
-		htup->t_cmax = xlrec->target.cid;
-		htup->t_infomask &= ~(HEAP_XMAX_INVALID | HEAP_MARKED_FOR_UPDATE);
-		htup->t_infomask |= HEAP_XMAX_COMMITTED;
+		if (move)
+		{
+			TransactionIdStore(record->xl_xid, (TransactionId *) &(htup->t_cmin));
+			htup->t_infomask &= 
+				~(HEAP_XMIN_COMMITTED | HEAP_XMIN_INVALID | HEAP_MOVED_IN);
+			htup->t_infomask |= HEAP_MOVED_OFF;
+		}
+		else
+		{
+			htup->t_xmax = record->xl_xid;
+			htup->t_cmax = xlrec->target.cid;
+			htup->t_infomask &= ~(HEAP_XMAX_COMMITTED |
+								HEAP_XMAX_INVALID | HEAP_MARKED_FOR_UPDATE);
+		}
 		PageSetLSN(page, lsn);
 		PageSetSUI(page, ThisStartUpID);
 		UnlockAndWriteBuffer(buffer);
@@ -2293,7 +2345,10 @@ heap_xlog_update(bool redo, XLogRecPtr lsn, XLogRecord *record)
 	}
 
 	/* undo... is it our tuple ? */
-	if (htup->t_xmax != record->xl_xid || htup->t_cmax != xlrec->target.cid)
+	if ((! move && (htup->t_xmax != record->xl_xid || 
+					htup->t_cmax != xlrec->target.cid)) ||
+		xlrec->target.cid != (CommandId) htup->t_xmin ||
+		htup->t_cmin != (CommandId) record->xl_xid)
 	{
 		if (!InRecovery)
 			elog(STOP, "heap_update_undo: invalid old tuple in rollback");
@@ -2301,7 +2356,14 @@ heap_xlog_update(bool redo, XLogRecPtr lsn, XLogRecord *record)
 	}
 	else	/* undo */
 	{
-		htup->t_infomask |= HEAP_XMAX_INVALID;
+		if (move)
+		{
+			htup->t_infomask &= ~(HEAP_XMIN_INVALID | 
+									HEAP_MOVED_IN | HEAP_MOVED_OFF);
+			htup->t_infomask |= HEAP_XMIN_COMMITTED;
+		}
+		else
+			htup->t_infomask |= HEAP_XMAX_INVALID;
 		UnlockAndWriteBuffer(buffer);
 	}
 
@@ -2329,8 +2391,9 @@ newt:;
 
 	if (redo)
 	{
-		char			tbuf[MaxTupleSize];
-		uint32			newlen = record->xl_len - SizeOfHeapUpdate;
+		char	tbuf[MaxTupleSize];
+		int		hsize;
+		uint32	newlen;
 
 		if (XLByteLE(lsn, PageGetLSN(page)))	/* changes are applied */
 		{
@@ -2338,16 +2401,35 @@ newt:;
 			return;
 		}
 
+		hsize = SizeOfHeapUpdate;
+		if (move)
+			hsize += sizeof(TransactionId);
+		newlen = record->xl_len - hsize;
+
 		htup = (HeapTupleHeader) tbuf;
 		memcpy(tbuf + offsetof(HeapTupleHeaderData, t_bits), 
-			(char*)xlrec + SizeOfHeapUpdate, newlen);
+			(char*)xlrec + hsize, newlen);
 		newlen += offsetof(HeapTupleHeaderData, t_bits);
 		htup->t_oid = xlrec->t_oid;
 		htup->t_natts = xlrec->t_natts;
 		htup->t_hoff = xlrec->t_hoff;
-		htup->t_xmin = record->xl_xid;
-		htup->t_cmin = xlrec->target.cid;
-		htup->t_infomask = HEAP_XMAX_INVALID | HEAP_XMIN_COMMITTED | xlrec->mask;
+		if (move)
+		{
+			htup->t_xmin = (TransactionId) xlrec->target.cid;
+			memcpy(&(htup->t_xmax), 
+					(char*)xlrec + SizeOfHeapUpdate, sizeof(TransactionId));
+			htup->t_infomask = xlrec->mask;
+			htup->t_infomask &= ~(HEAP_XMIN_COMMITTED | 
+						HEAP_XMIN_INVALID | HEAP_MOVED_OFF);
+			htup->t_infomask |= HEAP_MOVED_IN;
+		}
+		else
+		{
+			htup->t_xmin = record->xl_xid;
+			htup->t_cmin = xlrec->target.cid;
+			htup->t_xmax = htup->t_cmax = 0;
+			htup->t_infomask = HEAP_XMAX_INVALID | xlrec->mask;
+		}
 		
 		offnum = PageAddItem(page, (Item)htup, newlen, 
 			ItemPointerGetOffsetNumber(&(xlrec->newtid)), 
@@ -2385,7 +2467,10 @@ newt:;
 
 	/* is it our tuple ? */
 	Assert(PageGetSUI(page) == ThisStartUpID);
-	if (htup->t_xmin != record->xl_xid || htup->t_cmin != xlrec->target.cid)
+	if ((! move && (htup->t_xmin != record->xl_xid || 
+					htup->t_cmin != xlrec->target.cid)) ||
+		xlrec->target.cid != (CommandId) htup->t_xmin ||
+		htup->t_cmin != (CommandId) record->xl_xid)
 	{
 		if (!InRecovery)
 			elog(STOP, "heap_update_undo: invalid new tuple in rollback");
@@ -2448,11 +2533,9 @@ void heap_redo(XLogRecPtr lsn, XLogRecord *record)
 	else if (info == XLOG_HEAP_DELETE)
 		heap_xlog_delete(true, lsn, record);
 	else if (info == XLOG_HEAP_UPDATE)
-		heap_xlog_update(true, lsn, record);
-#ifdef NOT_USED
+		heap_xlog_update(true, lsn, record, false);
 	else if (info == XLOG_HEAP_MOVE)
-		heap_xlog_move(true, lsn, record);
-#endif
+		heap_xlog_update(true, lsn, record, true);
 	else
 		elog(STOP, "heap_redo: unknown op code %u", info);
 }
@@ -2466,11 +2549,9 @@ void heap_undo(XLogRecPtr lsn, XLogRecord *record)
 	else if (info == XLOG_HEAP_DELETE)
 		heap_xlog_delete(false, lsn, record);
 	else if (info == XLOG_HEAP_UPDATE)
-		heap_xlog_update(false, lsn, record);
-#ifdef NOT_USED
+		heap_xlog_update(false, lsn, record, false);
 	else if (info == XLOG_HEAP_MOVE)
-		heap_xlog_move(false, lsn, record);
-#endif
+		heap_xlog_update(false, lsn, record, true);
 	else
 		elog(STOP, "heap_undo: unknown op code %u", info);
 }
@@ -2509,19 +2590,13 @@ heap_desc(char *buf, uint8 xl_info, char* rec)
 		strcat(buf, "delete: ");
 		out_target(buf, &(xlrec->target));
 	}
-	else if (info == XLOG_HEAP_UPDATE)
+	else if (info == XLOG_HEAP_UPDATE || info == XLOG_HEAP_MOVE)
 	{
 		xl_heap_update	*xlrec = (xl_heap_update*) rec;
-		strcat(buf, "update: ");
-		out_target(buf, &(xlrec->target));
-		sprintf(buf + strlen(buf), "; new %u/%u",
-			ItemPointerGetBlockNumber(&(xlrec->newtid)), 
-			ItemPointerGetOffsetNumber(&(xlrec->newtid)));
-	}
-	else if (info == XLOG_HEAP_MOVE)
-	{
-		xl_heap_move	*xlrec = (xl_heap_move*) rec;
-		strcat(buf, "move: ");
+		if (info == XLOG_HEAP_UPDATE)
+			strcat(buf, "update: ");
+		else
+			strcat(buf, "move: ");
 		out_target(buf, &(xlrec->target));
 		sprintf(buf + strlen(buf), "; new %u/%u",
 			ItemPointerGetBlockNumber(&(xlrec->newtid)), 
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index 5533fa0509c..9fe1bbeb820 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/access/transam/xact.c,v 1.75 2000/10/23 04:10:05 vadim Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/access/transam/xact.c,v 1.76 2000/10/24 09:56:08 vadim Exp $
  *
  * NOTES
  *		Transaction aborts can now occur two ways:
@@ -220,7 +220,7 @@ int			XactIsoLevel;
 #ifdef XLOG
 #include "access/xlogutils.h"
 
-int			CommitDelay;
+int			CommitDelay = 100;
 
 void		xact_redo(XLogRecPtr lsn, XLogRecord *record);
 void		xact_undo(XLogRecPtr lsn, XLogRecord *record);
@@ -676,7 +676,36 @@ RecordTransactionCommit()
 	 */
 	leak = BufferPoolCheckLeak();
 
-#ifndef XLOG
+#ifdef XLOG
+	if (MyLastRecPtr.xlogid != 0 || MyLastRecPtr.xrecoff != 0)
+	{
+		xl_xact_commit	xlrec;
+		struct timeval	delay;
+		XLogRecPtr		recptr;
+
+		xlrec.xtime = time(NULL);
+		/*
+		 * MUST SAVE ARRAY OF RELFILENODE-s TO DROP
+		 */
+		recptr = XLogInsert(RM_XACT_ID, XLOG_XACT_COMMIT,
+			(char*) &xlrec, SizeOfXactCommit, NULL, 0);
+
+		/* 
+		 * Sleep before commit! So we can flush more than one
+		 * commit records per single fsync.
+		 */
+		delay.tv_sec = 0;
+		delay.tv_usec = CommitDelay;
+		(void) select(0, NULL, NULL, NULL, &delay);
+		XLogFlush(recptr);
+		MyLastRecPtr.xlogid = 0;
+		MyLastRecPtr.xrecoff = 0;
+
+		TransactionIdCommit(xid);
+
+		MyProc->logRec.xrecoff = 0;
+	}
+#else
 	/*
 	 * If no one shared buffer was changed by this transaction then we
 	 * don't flush shared buffers and don't record commit status.
@@ -687,39 +716,12 @@ RecordTransactionCommit()
 		if (leak)
 			ResetBufferPool(true);
 
-#endif
 		/*
 		 * have the transaction access methods record the status of this
 		 * transaction id in the pg_log relation.
 		 */
 		TransactionIdCommit(xid);
 
-#ifdef XLOG
-		if (MyLastRecPtr.xlogid != 0 || MyLastRecPtr.xrecoff != 0)
-		{
-			xl_xact_commit	xlrec;
-			struct timeval	delay;
-			XLogRecPtr		recptr;
-
-			xlrec.xtime = time(NULL);
-			/*
-			 * MUST SAVE ARRAY OF RELFILENODE-s TO DROP
-			 */
-			recptr = XLogInsert(RM_XACT_ID, XLOG_XACT_COMMIT,
-				(char*) &xlrec, SizeOfXactCommit, NULL, 0);
-
-			/* 
-			 * Sleep before commit! So we can flush more than one
-			 * commit records per single fsync.
-			 */
-			delay.tv_sec = 0;
-			delay.tv_usec = CommitDelay;
-			(void) select(0, NULL, NULL, NULL, &delay);
-			XLogFlush(recptr);
-			MyLastRecPtr.xlogid = 0;
-			MyLastRecPtr.xrecoff = 0;
-		}
-#else
 		/*
 		 * Now write the log info to the disk too.
 		 */
@@ -1737,12 +1739,12 @@ xact_redo(XLogRecPtr lsn, XLogRecord *record)
 	{
 		xl_xact_commit	*xlrec = (xl_xact_commit*) XLogRecGetData(record);
 
-		XLogMarkCommitted(record->xl_xid);
+		TransactionIdCommit(record->xl_xid);
 		/* MUST REMOVE FILES OF ALL DROPPED RELATIONS */
 	}
 	else if (info == XLOG_XACT_ABORT)
 	{
-		XLogMarkAborted(record->xl_xid);
+		TransactionIdAbort(record->xl_xid);
 	}
 	else
 		elog(STOP, "xact_redo: unknown op code %u", info);
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 19d7f38c1ef..1343743c096 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -6,7 +6,7 @@
  * Portions Copyright (c) 1996-2000, PostgreSQL, Inc
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $Header: /cvsroot/pgsql/src/backend/access/transam/xlog.c,v 1.20 2000/10/23 04:10:05 vadim Exp $
+ * $Header: /cvsroot/pgsql/src/backend/access/transam/xlog.c,v 1.21 2000/10/24 09:56:09 vadim Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -19,13 +19,14 @@
 
 #include "postgres.h"
 
-#include "access/xlog.h"
 #include "access/xact.h"
 #include "catalog/catversion.h"
 #include "storage/sinval.h"
 #include "storage/proc.h"
 #include "storage/spin.h"
 #include "storage/s_lock.h"
+#include "access/xlog.h"
+#include "access/xlogutils.h"
 
 #include "miscadmin.h"
 
@@ -1343,9 +1344,10 @@ StartupXLOG()
 		elog(STOP, "Invalid length of checkPoint record");
 	checkPoint = *((CheckPoint *) ((char *) record + SizeOfXLogRecord));
 
-	elog(LOG, "Redo record at (%u, %u); Undo record at (%u, %u)",
+	elog(LOG, "Redo record at (%u, %u); Undo record at (%u, %u); Shutdown %s",
 		 checkPoint.redo.xlogid, checkPoint.redo.xrecoff,
-		 checkPoint.undo.xlogid, checkPoint.undo.xrecoff);
+		 checkPoint.undo.xlogid, checkPoint.undo.xrecoff,
+		 (checkPoint.Shutdown) ? "TRUE" : "FALSE");
 	elog(LOG, "NextTransactionId: %u; NextOid: %u",
 		 checkPoint.nextXid, checkPoint.nextOid);
 	if (checkPoint.nextXid < FirstTransactionId ||
@@ -1371,16 +1373,23 @@ StartupXLOG()
 	if (XLByteLT(RecPtr, checkPoint.undo))
 		elog(STOP, "Invalid undo in checkPoint record");
 
-	if (XLByteLT(checkPoint.undo, RecPtr) || XLByteLT(checkPoint.redo, RecPtr))
+	if (XLByteLT(checkPoint.undo, RecPtr) || 
+		XLByteLT(checkPoint.redo, RecPtr))
 	{
+		if (checkPoint.Shutdown)
+			elog(STOP, "Invalid Redo/Undo record in shutdown checkpoint");
 		if (ControlFile->state == DB_SHUTDOWNED)
 			elog(STOP, "Invalid Redo/Undo record in Shutdowned state");
-		recovery = 2;
+		recovery = 1;
 	}
 	else if (ControlFile->state != DB_SHUTDOWNED)
-		recovery = 2;
+	{
+		if (checkPoint.Shutdown)
+			elog(STOP, "Invalid state in control file");
+		recovery = 1;
+	}
 
-	if (recovery > 0)
+	if (recovery)
 	{
 		elog(LOG, "The DataBase system was not properly shut down\n"
 			 "\tAutomatic recovery is in progress...");
@@ -1391,6 +1400,8 @@ StartupXLOG()
 		sie_saved = StopIfError;
 		StopIfError = true;
 
+		XLogOpenLogRelation();	/* open pg_log */
+
 		/* Is REDO required ? */
 		if (XLByteLT(checkPoint.redo, RecPtr))
 			record = ReadRecord(&(checkPoint.redo), buffer);
@@ -1432,11 +1443,9 @@ StartupXLOG()
 			LastRec = ReadRecPtr;
 		}
 		else
-		{
 			elog(LOG, "Redo is not required");
-			recovery--;
-		}
 
+#ifdef NOT_USED
 		/* UNDO */
 		RecPtr = ReadRecPtr;
 		if (XLByteLT(checkPoint.undo, RecPtr))
@@ -1455,10 +1464,8 @@ StartupXLOG()
 				 ReadRecPtr.xlogid, ReadRecPtr.xrecoff);
 		}
 		else
-		{
 			elog(LOG, "Undo is not required");
-			recovery--;
-		}
+#endif
 	}
 
 	/* Init xlog buffer cache */
@@ -1476,17 +1483,8 @@ StartupXLOG()
 		(EndRecPtr.xrecoff + BLCKSZ - XLogCtl->xlblocks[0].xrecoff);
 	Insert->PrevRecord = ControlFile->checkPoint;
 
-	if (recovery > 0)
+	if (recovery)
 	{
-#ifdef NOT_USED
-		int			i;
-
-		/*
-		 * Let resource managers know that recovery is done
-		 */
-		for (i = 0; i <= RM_MAX_ID; i++)
-			RmgrTable[record->xl_rmid].rm_redo(ReadRecPtr, NULL);
-#endif
 		CreateCheckPoint(true);
 		StopIfError = sie_saved;
 	}
@@ -1586,7 +1584,7 @@ CreateCheckPoint(bool shutdown)
 
 	FlushBufferPool();
 
-	/* Get UNDO record ptr */
+	/* Get UNDO record ptr - should use oldest of PROC->logRec */
 	checkPoint.undo.xrecoff = 0;
 
 	if (shutdown && checkPoint.undo.xrecoff != 0)
diff --git a/src/backend/access/transam/xlogutils.c b/src/backend/access/transam/xlogutils.c
index 5140cfaf34a..2800ff0316f 100644
--- a/src/backend/access/transam/xlogutils.c
+++ b/src/backend/access/transam/xlogutils.c
@@ -14,6 +14,7 @@
 #include "postgres.h"
 
 #include "access/xlog.h"
+#include "access/transam.h"
 #include "access/xact.h"
 #include "storage/bufpage.h"
 #include "storage/bufmgr.h"
@@ -134,10 +135,16 @@ XLogIsValidTuple(RelFileNode hnode, ItemPointer iptr)
 
 	/* MUST CHECK WASN'T TUPLE INSERTED IN PREV STARTUP */
 
-	if (XLogIsAborted(htup->t_xmin))
+	if (!(htup->t_infomask & HEAP_XMIN_COMMITTED))
 	{
-		UnlockAndReleaseBuffer(buffer);
-		return(false);
+		if (htup->t_infomask & HEAP_XMIN_INVALID ||
+			(htup->t_infomask & HEAP_MOVED_IN &&
+			TransactionIdDidAbort((TransactionId)htup->t_cmin)) ||
+			TransactionIdDidAbort(htup->t_xmin))
+		{
+			UnlockAndReleaseBuffer(buffer);
+			return(false);
+		}
 	}
 
 	UnlockAndReleaseBuffer(buffer);
@@ -145,41 +152,32 @@ XLogIsValidTuple(RelFileNode hnode, ItemPointer iptr)
 }
 
 /*
- * ---------------------------------------------------------------
- *
- * Transaction support functions for recovery
- *
- * On recovery we create tmp file to know what xactions were
- * committed/aborted (2 bits per xaction).
- *
- *----------------------------------------------------------------
+ * Open pg_log in recovery
  */
-
-bool
-XLogIsAborted(TransactionId xid)
-{
-	return(false);
-}
-
-bool
-XLogIsCommitted(TransactionId xid)
-{
-	return(true);
-}
+extern Relation	LogRelation;	/* pg_log relation */
 
 void
-XLogMarkAborted(TransactionId xid)
+XLogOpenLogRelation(void)
 {
-	return;
+	Relation	logRelation;
+
+	Assert(!LogRelation);
+	logRelation = (Relation) malloc(sizeof(RelationData));
+	memset(logRelation, 0, sizeof(RelationData));
+	logRelation->rd_rel = (Form_pg_class) malloc(sizeof(FormData_pg_class));
+	memset(logRelation->rd_rel, 0, sizeof(FormData_pg_class));
+
+	sprintf(RelationGetPhysicalRelationName(logRelation), "pg_log");
+	logRelation->rd_node.tblNode = InvalidOid;
+	logRelation->rd_node.relNode = RelOid_pg_log;
+	logRelation->rd_unlinked = false;	/* must exists */
+	logRelation->rd_fd = -1;
+	logRelation->rd_fd = smgropen(DEFAULT_SMGR, logRelation);
+	if (logRelation->rd_fd < 0)
+		elog(STOP, "XLogOpenLogRelation: failed to open pg_log");
+	LogRelation = logRelation;
 }
 
-void
-XLogMarkCommitted(TransactionId xid)
-{
-	return;
-}
-
-
 /*
  * ---------------------------------------------------------------
  *
@@ -274,7 +272,7 @@ _xl_init_rel_cache(void)
 	_xlrelarr = (XLogRelDesc*) malloc(sizeof(XLogRelDesc) * _xlcnt);
 	memset(_xlrelarr, 0, sizeof(XLogRelDesc) * _xlcnt);
 	_xlpgcarr = (Form_pg_class) malloc(sizeof(FormData_pg_class) * _xlcnt);
-	memset(_xlpgcarr, 0, sizeof(XLogRelDesc) * _xlcnt);
+	memset(_xlpgcarr, 0, sizeof(FormData_pg_class) * _xlcnt);
 
 	_xlrelarr[0].moreRecently = &(_xlrelarr[0]);
 	_xlrelarr[0].lessRecently = &(_xlrelarr[0]);
diff --git a/src/backend/bootstrap/bootstrap.c b/src/backend/bootstrap/bootstrap.c
index 275433dda39..aab0c21230e 100644
--- a/src/backend/bootstrap/bootstrap.c
+++ b/src/backend/bootstrap/bootstrap.c
@@ -8,7 +8,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/bootstrap/bootstrap.c,v 1.94 2000/10/21 15:43:24 vadim Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/bootstrap/bootstrap.c,v 1.95 2000/10/24 09:56:09 vadim Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -399,6 +399,10 @@ BootstrapMain(int argc, char *argv[])
 	 */
 	Int_yyparse();
 
+#ifdef XLOG
+	FlushBufferPool();
+#endif
+
 	/* clean up processing */
 	StartTransactionCommand();
 	cleanup();
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c
index 9a790aae2e3..0905f60b807 100644
--- a/src/backend/commands/vacuum.c
+++ b/src/backend/commands/vacuum.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/commands/vacuum.c,v 1.169 2000/10/22 19:49:43 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/commands/vacuum.c,v 1.170 2000/10/24 09:56:15 vadim Exp $
  *
 
  *-------------------------------------------------------------------------
@@ -47,6 +47,11 @@
 #include <sys/resource.h>
 #endif
 
+#ifdef XLOG
+#include "access/xlog.h"
+XLogRecPtr	log_heap_move(Relation reln, 
+				ItemPointerData from, HeapTuple newtup);
+#endif
 
 static MemoryContext vac_context = NULL;
 
@@ -1401,11 +1406,23 @@ repair_frag(VRelStats *vacrelstats, Relation onerel,
 				{
 					VacPage	destvacpage = vtmove[ti].vacpage;
 
-					/* Get tuple from chain */
+					/* Get page to move from */
 					tuple.t_self = vtmove[ti].tid;
 					Cbuf = ReadBuffer(onerel,
 							 ItemPointerGetBlockNumber(&(tuple.t_self)));
+
+					/* Get page to move to */
+					cur_buffer = ReadBuffer(onerel, destvacpage->blkno);
+
+					LockBuffer(cur_buffer, BUFFER_LOCK_EXCLUSIVE);
+					if (cur_buffer != Cbuf)
+						LockBuffer(Cbuf, BUFFER_LOCK_EXCLUSIVE);
+
+					ToPage = BufferGetPage(cur_buffer);
 					Cpage = BufferGetPage(Cbuf);
+
+					/* NO ELOG(ERROR) TILL CHANGES ARE LOGGED */
+
 					Citemid = PageGetItemId(Cpage,
 							ItemPointerGetOffsetNumber(&(tuple.t_self)));
 					tuple.t_datamcxt = NULL;
@@ -1425,19 +1442,6 @@ repair_frag(VRelStats *vacrelstats, Relation onerel,
 						~(HEAP_XMIN_COMMITTED | HEAP_XMIN_INVALID | HEAP_MOVED_IN);
 					tuple.t_data->t_infomask |= HEAP_MOVED_OFF;
 
-					/* Get page to move to */
-					cur_buffer = ReadBuffer(onerel, destvacpage->blkno);
-
-					/*
-					 * We should LockBuffer(cur_buffer) but don't, at the
-					 * moment.  This should be safe enough, since we have
-					 * exclusive lock on the whole relation.
-					 * If you'll do LockBuffer then UNLOCK it before
-					 * index_insert: unique btree-s call heap_fetch to get
-					 * t_infomask of inserted heap tuple !!!
-					 */
-					ToPage = BufferGetPage(cur_buffer);
-
 					/*
 					 * If this page was not used before - clean it.
 					 *
@@ -1480,7 +1484,7 @@ repair_frag(VRelStats *vacrelstats, Relation onerel,
 										 InvalidOffsetNumber, LP_USED);
 					if (newoff == InvalidOffsetNumber)
 					{
-						elog(ERROR, "moving chain: failed to add item with len = %u to page %u",
+						elog(STOP, "moving chain: failed to add item with len = %u to page %u",
 							 tuple_len, destvacpage->blkno);
 					}
 					newitemid = PageGetItemId(ToPage, newoff);
@@ -1488,6 +1492,22 @@ repair_frag(VRelStats *vacrelstats, Relation onerel,
 					newtup.t_datamcxt = NULL;
 					newtup.t_data = (HeapTupleHeader) PageGetItem(ToPage, newitemid);
 					ItemPointerSet(&(newtup.t_self), destvacpage->blkno, newoff);
+
+#ifdef XLOG
+					{
+						XLogRecPtr	recptr = 
+							log_heap_move(onerel, tuple.t_self, &newtup);
+
+						if (Cbuf != cur_buffer)
+						{
+							PageSetLSN(Cpage, recptr);
+							PageSetSUI(Cpage, ThisStartUpID);
+						}
+						PageSetLSN(ToPage, recptr);
+						PageSetSUI(ToPage, ThisStartUpID);
+					}
+#endif
+
 					if (((int) destvacpage->blkno) > last_move_dest_block)
 						last_move_dest_block = destvacpage->blkno;
 
@@ -1513,6 +1533,10 @@ repair_frag(VRelStats *vacrelstats, Relation onerel,
 					else
 						keep_tuples++;
 
+					LockBuffer(cur_buffer, BUFFER_LOCK_UNLOCK);
+					if (cur_buffer != Cbuf)
+						LockBuffer(Cbuf, BUFFER_LOCK_UNLOCK);
+
 					if (Irel != (Relation *) NULL)
 					{
 						/*
@@ -1581,11 +1605,16 @@ repair_frag(VRelStats *vacrelstats, Relation onerel,
 				cur_item = i;
 				cur_page = fraged_pages->pagedesc[cur_item];
 				cur_buffer = ReadBuffer(onerel, cur_page->blkno);
+				LockBuffer(cur_buffer, BUFFER_LOCK_EXCLUSIVE);
 				ToPage = BufferGetPage(cur_buffer);
 				/* if this page was not used before - clean it */
 				if (!PageIsEmpty(ToPage) && cur_page->offsets_used == 0)
 					vacuum_page(ToPage, cur_page);
 			}
+			else
+				LockBuffer(cur_buffer, BUFFER_LOCK_EXCLUSIVE);
+
+			LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
 
 			/* copy tuple */
 			heap_copytuple_with_tuple(&tuple, &newtup);
@@ -1627,6 +1656,18 @@ failed to add item with len = %u to page %u (free space %u, nusd %u, noff %u)",
 				~(HEAP_XMIN_COMMITTED | HEAP_XMIN_INVALID | HEAP_MOVED_IN);
 			tuple.t_data->t_infomask |= HEAP_MOVED_OFF;
 
+#ifdef XLOG
+			{
+				XLogRecPtr	recptr = 
+					log_heap_move(onerel, tuple.t_self, &newtup);
+
+				PageSetLSN(page, recptr);
+				PageSetSUI(page, ThisStartUpID);
+				PageSetLSN(ToPage, recptr);
+				PageSetSUI(ToPage, ThisStartUpID);
+			}
+#endif
+
 			cur_page->offsets_used++;
 			num_moved++;
 			cur_page->free = ((PageHeader) ToPage)->pd_upper - ((PageHeader) ToPage)->pd_lower;
@@ -1635,6 +1676,9 @@ failed to add item with len = %u to page %u (free space %u, nusd %u, noff %u)",
 
 			vacpage->offsets[vacpage->offsets_free++] = offnum;
 
+			LockBuffer(cur_buffer, BUFFER_LOCK_UNLOCK);
+			LockBuffer(buf, BUFFER_LOCK_UNLOCK);
+
 			/* insert index' tuples if needed */
 			if (Irel != (Relation *) NULL)
 			{
diff --git a/src/include/access/htup.h b/src/include/access/htup.h
index 6f2b085b5c7..6b3d76ec889 100644
--- a/src/include/access/htup.h
+++ b/src/include/access/htup.h
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2000, PostgreSQL, Inc
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $Id: htup.h,v 1.36 2000/10/20 11:01:14 vadim Exp $
+ * $Id: htup.h,v 1.37 2000/10/24 09:56:23 vadim Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -111,7 +111,7 @@ typedef struct xl_heap_insert
 
 #define SizeOfHeapInsert	(offsetof(xl_heap_insert, mask) + sizeof(uint8))
 
-/* This is what we need to know about update - 32 + data */
+/* This is what we need to know about update|move - 32|36 + data */
 typedef struct xl_heap_update
 {
 	xl_heaptid			target;		/* deleted tuple id */
@@ -122,19 +122,11 @@ typedef struct xl_heap_update
 	uint8				t_hoff;
 	uint8				mask;		/* low 8 bits of t_infomask */
 	/* NEW TUPLE DATA FOLLOWS AT END OF STRUCT */
+	/* (AFTER XMAX FOR MOVE OP) */
 } xl_heap_update;
 
 #define SizeOfHeapUpdate	(offsetof(xl_heap_update, mask) + sizeof(uint8))
 
-/* This is what we need to know about tuple move - 24 bytes */
-typedef struct xl_heap_move
-{
-	xl_heaptid			target;		/* moved from */
-	ItemPointerData		newtid;		/* moved to */
-} xl_heap_move;
-
-#define	SizeOfHeapMove	(offsetof(xl_heap_move, ttid) + SizeOfIptrData))
-
 /* end of XLOG stuff */
 
 #endif	/* XLOG */