1. WAL needs in zero-ed content of newly initialized page.

2. Log record for PageRepaireFragmentation now keeps array
   of !LP_USED offnums to redo cleanup properly.

src/backend/access/heap/heapam.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/access/heap/heapam.c,v 1.104 2000/12/30 06:52:33 vadim Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/access/heap/heapam.c,v 1.105 2000/12/30 15:19:54 vadim Exp $
  *
  *
  * INTERFACE ROUTINES
@@ -90,7 +90,8 @@
 
 XLogRecPtr	log_heap_move(Relation reln, Buffer oldbuf, ItemPointerData from, 
 							Buffer newbuf, HeapTuple newtup);
-XLogRecPtr	log_heap_clean(Relation reln, Buffer buffer);
+XLogRecPtr	log_heap_clean(Relation reln, Buffer buffer,
+							char *unused, int unlen);
 
 /* comments are in heap_update */
 static xl_heaptid	_locked_tuple_;
@@ -2002,11 +2003,11 @@ heap_restrpos(HeapScanDesc scan)
 }
 
 XLogRecPtr
-log_heap_clean(Relation reln, Buffer buffer)
+log_heap_clean(Relation reln, Buffer buffer, char *unused, int unlen)
 {
 	xl_heap_clean	xlrec;
 	XLogRecPtr		recptr;
-	XLogRecData		rdata[2];
+	XLogRecData		rdata[3];
 
 	xlrec.node = reln->rd_node;
 	xlrec.block = BufferGetBlockNumber(buffer);
@@ -2015,10 +2016,20 @@ log_heap_clean(Relation reln, Buffer buffer)
 	rdata[0].len = SizeOfHeapClean;
 	rdata[0].next = &(rdata[1]);
 
-	rdata[1].buffer = buffer;
-	rdata[1].data = NULL;
-	rdata[1].len = 0;
-	rdata[1].next = NULL;
+	if (unlen > 0)
+	{
+		rdata[1].buffer = buffer;
+		rdata[1].data = unused;
+		rdata[1].len = unlen;
+		rdata[1].next = &(rdata[2]);
+	}
+	else
+		rdata[0].next = &(rdata[2]);
+
+	rdata[2].buffer = buffer;
+	rdata[2].data = NULL;
+	rdata[2].len = 0;
+	rdata[2].next = NULL;
 
 	recptr = XLogInsert(RM_HEAP_ID, XLOG_HEAP_CLEAN, rdata);
 
@@ -2102,14 +2113,10 @@ log_heap_move(Relation reln, Buffer oldbuf, ItemPointerData from,
 static void
 heap_xlog_clean(bool redo, XLogRecPtr lsn, XLogRecord *record)
 {
-	xl_heap_clean *xlrec = (xl_heap_clean*) XLogRecGetData(record);
+	xl_heap_clean  *xlrec = (xl_heap_clean*) XLogRecGetData(record);
 	Relation		reln;
 	Buffer			buffer;
 	Page			page;
-	OffsetNumber	maxoff;
-	OffsetNumber	offnum;
-	HeapTupleHeader	htup;
-	ItemId			lp;
 
 	if (!redo || (record->xl_info & XLR_BKP_BLOCK_1))
 		return;
@@ -2133,23 +2140,25 @@ heap_xlog_clean(bool redo, XLogRecPtr lsn, XLogRecord *record)
 		return;
 	}
 
-	maxoff = PageGetMaxOffsetNumber(page);
-	for (offnum = FirstOffsetNumber;
-		 offnum <= maxoff;
-		 offnum = OffsetNumberNext(offnum))
+	if (record->xl_len > SizeOfHeapClean)
 	{
-		lp = PageGetItemId(page, offnum);
-
-		if (!ItemIdIsUsed(lp))
-			continue;
+		char			unbuf[BLCKSZ];
+		OffsetNumber   *unused = (OffsetNumber*)unbuf;
+		char		   *unend;
+		ItemId			lp;
 
-		htup = (HeapTupleHeader) PageGetItem(page, lp);
+		memcpy(unbuf, (char*)xlrec + SizeOfHeapClean, record->xl_len - SizeOfHeapClean);
+		unend = unbuf + (record->xl_len - SizeOfHeapClean);
 
-		if (!HeapTupleSatisfiesNow(htup))
+		while((char*)unused < unend)
+		{
+			lp = ((PageHeader) page)->pd_linp + *unused;
 			lp->lp_flags &= ~LP_USED;
+			unused++;
+		}
 	}
 
-	PageRepairFragmentation(page);
+	PageRepairFragmentation(page, NULL);
 	UnlockAndWriteBuffer(buffer);
 }
 
@@ -2247,7 +2256,10 @@ heap_xlog_insert(bool redo, XLogRecPtr lsn, XLogRecord *record)
 		uint32			newlen;
 
 		if (record->xl_info & XLOG_HEAP_INIT_PAGE)
+		{
 			PageInit(page, BufferGetPageSize(buffer), 0);
+			PageZero(page);
+		}
 
 		if (XLByteLE(lsn, PageGetLSN(page)))	/* changes are applied */
 		{
@@ -2401,7 +2413,10 @@ newsame:;
 		uint32			newlen;
 
 		if (record->xl_info & XLOG_HEAP_INIT_PAGE)
+		{
 			PageInit(page, BufferGetPageSize(buffer), 0);
+			PageZero(page);
+		}
 
 		if (XLByteLE(lsn, PageGetLSN(page)))	/* changes are applied */
 		{
@@ -2583,6 +2598,12 @@ heap_desc(char *buf, uint8 xl_info, char* rec)
 			ItemPointerGetBlockNumber(&(xlrec->newtid)), 
 			ItemPointerGetOffsetNumber(&(xlrec->newtid)));
 	}
+	else if (info == XLOG_HEAP_CLEAN)
+	{
+		xl_heap_clean	*xlrec = (xl_heap_clean*) rec;
+		sprintf(buf + strlen(buf), "clean: node %u/%u; blk %u",
+			xlrec->node.tblNode, xlrec->node.relNode, xlrec->block);
+	}
 	else
 		strcat(buf, "UNKNOWN");
 }

src/backend/commands/vacuum.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/commands/vacuum.c,v 1.180 2000/12/28 13:00:18 vadim Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/commands/vacuum.c,v 1.181 2000/12/30 15:19:55 vadim Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -47,7 +47,8 @@
 #include "utils/syscache.h"
 #include "utils/temprel.h"
 
-extern XLogRecPtr	log_heap_clean(Relation reln, Buffer buffer);
+extern XLogRecPtr	log_heap_clean(Relation reln, Buffer buffer,
+									char *unused, int unlen);
 extern XLogRecPtr	log_heap_move(Relation reln, 
 						Buffer oldbuf, ItemPointerData from,
 						Buffer newbuf, HeapTuple newtup);
@@ -878,7 +879,7 @@ scan_heap(VRelStats *vacrelstats, Relation onerel,
 
 		if (tempPage != (Page) NULL)
 		{						/* Some tuples are gone */
-			PageRepairFragmentation(tempPage);
+			PageRepairFragmentation(tempPage, NULL);
 			vacpage->free = ((PageHeader) tempPage)->pd_upper - ((PageHeader) tempPage)->pd_lower;
 			free_size += vacpage->free;
 			reap_page(vacuum_pages, vacpage);
@@ -1898,6 +1899,10 @@ failed to add item with len = %lu to page %u (free space %lu, nusd %u, noff %u)"
 		if (vacpage->blkno == (BlockNumber) (blkno - 1) &&
 			vacpage->offsets_free > 0)
 		{
+			char			unbuf[BLCKSZ];
+			OffsetNumber   *unused = (OffsetNumber*)unbuf;
+			int				uncnt;
+
 			buf = ReadBuffer(onerel, vacpage->blkno);
 			LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
 			START_CRIT_CODE;
@@ -1928,9 +1933,11 @@ failed to add item with len = %lu to page %u (free space %lu, nusd %u, noff %u)"
 
 			}
 			Assert(vacpage->offsets_free == num_tuples);
-			PageRepairFragmentation(page);
+			uncnt = PageRepairFragmentation(page, unused);
 			{
-				XLogRecPtr	recptr = log_heap_clean(onerel, buf);
+				XLogRecPtr	recptr;
+				recptr = log_heap_clean(onerel, buf, (char*)unused,
+					(char*)(&(unused[uncnt])) - (char*)unused);
 				PageSetLSN(page, recptr);
 				PageSetSUI(page, ThisStartUpID);
 			}
@@ -2039,9 +2046,12 @@ vacuum_heap(VRelStats *vacrelstats, Relation onerel, VacPageList vacuum_pages)
 static void
 vacuum_page(Relation onerel, Buffer buffer, VacPage vacpage)
 {
-	Page		page = BufferGetPage(buffer);
-	ItemId		itemid;
-	int			i;
+	char			unbuf[BLCKSZ];
+	OffsetNumber   *unused = (OffsetNumber*)unbuf;
+	int				uncnt;
+	Page			page = BufferGetPage(buffer);
+	ItemId			itemid;
+	int				i;
 
 	/* There shouldn't be any tuples moved onto the page yet! */
 	Assert(vacpage->offsets_used == 0);
@@ -2052,9 +2062,11 @@ vacuum_page(Relation onerel, Buffer buffer, VacPage vacpage)
 		itemid = &(((PageHeader) page)->pd_linp[vacpage->offsets[i] - 1]);
 		itemid->lp_flags &= ~LP_USED;
 	}
-	PageRepairFragmentation(page);
+	uncnt = PageRepairFragmentation(page, unused);
 	{
-		XLogRecPtr	recptr = log_heap_clean(onerel, buffer);
+		XLogRecPtr	recptr;
+		recptr = log_heap_clean(onerel, buffer, (char*)unused,
+					(char*)(&(unused[uncnt])) - (char*)unused);
 		PageSetLSN(page, recptr);
 		PageSetSUI(page, ThisStartUpID);
 	}

src/backend/storage/page/bufpage.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/storage/page/bufpage.c,v 1.33 2000/10/21 15:43:29 vadim Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/storage/page/bufpage.c,v 1.34 2000/12/30 15:19:55 vadim Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -48,6 +48,19 @@ PageInit(Page page, Size pageSize, Size specialSize)
 	p->pd_upper = pageSize - specialSize;
 	p->pd_special = pageSize - specialSize;
 	PageSetPageSize(page, pageSize);
+
+	p->pd_lsn.xlogid = p->pd_lsn.xrecoff = 0;
+	p->pd_sui = 0;
+}
+
+/*
+ * WAL needs in zero-ed page data content
+ */
+void
+PageZero(Page page)
+{
+	MemSet((char*)page + ((PageHeader)page)->pd_lower, 0,
+			((PageHeader)page)->pd_special - ((PageHeader)page)->pd_lower);
 }
 
 /* ----------------
@@ -251,8 +264,8 @@ itemidcompare(const void *itemidp1, const void *itemidp2)
  * This routine is usable for heap pages only.
  *
  */
-void
-PageRepairFragmentation(Page page)
+int
+PageRepairFragmentation(Page page, OffsetNumber *unused)
 {
 	int			i;
 	struct itemIdSortData *itemidbase,
@@ -272,6 +285,8 @@ PageRepairFragmentation(Page page)
 			(*lp).lp_flags &= ~(LP_USED | LP_DELETE);
 		if ((*lp).lp_flags & LP_USED)
 			nused++;
+		else if (unused)
+			unused[i - nused] = (OffsetNumber)i;
 	}
 
 	if (nused == 0)
@@ -328,6 +343,8 @@ PageRepairFragmentation(Page page)
 
 		pfree(itemidbase);
 	}
+
+	return(nline - nused);
 }
 
 /*

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.43 2000/12/28 13:00:25 vadim Exp $
+ * $Id: htup.h,v 1.44 2000/12/30 15:19:56 vadim Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -131,6 +131,7 @@ typedef struct xl_heap_clean
 {
 	RelFileNode			node;
 	BlockNumber			block;
+	/* UNUSED OFFSET NUMBERS FOLLOW AT THE END */
 } xl_heap_clean;
 
 #define	SizeOfHeapClean	(offsetof(xl_heap_clean, block) + sizeof(BlockNumber))

src/include/storage/bufpage.h

@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2000, PostgreSQL, Inc
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $Id: bufpage.h,v 1.37 2000/11/30 08:46:26 vadim Exp $
+ * $Id: bufpage.h,v 1.38 2000/12/30 15:19:57 vadim Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -314,11 +314,12 @@ typedef enum
  */
 
 extern void PageInit(Page page, Size pageSize, Size specialSize);
+extern void PageZero(Page page);
 extern OffsetNumber PageAddItem(Page page, Item item, Size size,
 			OffsetNumber offsetNumber, ItemIdFlags flags);
 extern Page PageGetTempPage(Page page, Size specialSize);
 extern void PageRestoreTempPage(Page tempPage, Page oldPage);
-extern void PageRepairFragmentation(Page page);
+extern int PageRepairFragmentation(Page page, OffsetNumber *unused);
 extern Size PageGetFreeSpace(Page page);
 extern void PageIndexTupleDelete(Page page, OffsetNumber offset);
 extern void IndexPageCleanup(Buffer buffer);