vacuum-secondphase-setvm-v1.patch

diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 74c41fa..cd76158 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -1945,6 +1945,16 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
 							ItemPointerGetBlockNumber(&(heaptup->t_self)),
 							vmbuffer);
 	}
+	else if (PageIsAllVisibleOrDead(BufferGetPage(buffer)))
+	{
+		/*
+		 * We don't need to worry about WAL-logging this action. The only place
+		 * where this flag is consulted is in the second phase of vacuum and
+		 * the first phase of vacuum will take care of clearing/setting the
+		 * flag appropriately
+		 */
+		PageClearAllVisibleOrDead(BufferGetPage(buffer));
+	}
 
 	/*
 	 * XXX Should we set PageSetPrunable on this page ?
@@ -2204,6 +2214,8 @@ heap_multi_insert(Relation relation, HeapTuple *tuples, int ntuples,
 								BufferGetBlockNumber(buffer),
 								vmbuffer);
 		}
+		else if (PageIsAllVisibleOrDead(page))
+			PageClearAllVisibleOrDead(page);
 
 		/*
 		 * XXX Should we set PageSetPrunable on this page ? See heap_insert()
@@ -2585,6 +2597,8 @@ l1:
 		visibilitymap_clear(relation, BufferGetBlockNumber(buffer),
 							vmbuffer);
 	}
+	else if (PageIsAllVisibleOrDead(page))
+		PageClearAllVisibleOrDead(page);
 
 	/* store transaction information of xact deleting the tuple */
 	tp.t_data->t_infomask &= ~(HEAP_XMAX_COMMITTED |
@@ -3197,6 +3211,9 @@ l2:
 		visibilitymap_clear(relation, BufferGetBlockNumber(buffer),
 							vmbuffer);
 	}
+	else if (PageIsAllVisibleOrDead(BufferGetPage(buffer)))
+		PageClearAllVisibleOrDead(BufferGetPage(buffer));
+
 	if (newbuf != buffer && PageIsAllVisible(BufferGetPage(newbuf)))
 	{
 		all_visible_cleared_new = true;
@@ -3204,6 +3221,8 @@ l2:
 		visibilitymap_clear(relation, BufferGetBlockNumber(newbuf),
 							vmbuffer_new);
 	}
+	else if (PageIsAllVisibleOrDead(BufferGetPage(newbuf)))
+		PageClearAllVisibleOrDead(BufferGetPage(newbuf));
 
 	if (newbuf != buffer)
 		MarkBufferDirty(newbuf);
diff --git a/src/backend/access/heap/pruneheap.c b/src/backend/access/heap/pruneheap.c
index 97a2868..221ddf0 100644
--- a/src/backend/access/heap/pruneheap.c
+++ b/src/backend/access/heap/pruneheap.c
@@ -222,6 +222,13 @@ heap_page_prune(Relation relation, Buffer buffer, TransactionId OldestXmin,
 		((PageHeader) page)->pd_prune_xid = prstate.new_prune_xid;
 
 		/*
+		 * If we don't find new prune xid, we assume that there is nothing more
+		 * do on the page and so clear the prunable flag
+		 */
+		if (!TransactionIdIsValid(prstate.new_prune_xid))
+			PageClearPrunable(page);
+
+		/*
 		 * Also clear the "page is full" flag, since there's no point in
 		 * repeating the prune/defrag process until something else happens to
 		 * the page.
diff --git a/src/backend/commands/vacuumlazy.c b/src/backend/commands/vacuumlazy.c
index 5036849..ea9cfa5 100644
--- a/src/backend/commands/vacuumlazy.c
+++ b/src/backend/commands/vacuumlazy.c
@@ -458,6 +458,7 @@ lazy_scan_heap(Relation onerel, LVRelStats *vacrelstats,
 		Size		freespace;
 		bool		all_visible_according_to_vm;
 		bool		all_visible;
+		bool		all_visible_or_dead;
 		bool		has_dead_tuples;
 		TransactionId visibility_cutoff_xid = InvalidTransactionId;
 
@@ -661,6 +662,7 @@ lazy_scan_heap(Relation onerel, LVRelStats *vacrelstats,
 		 * requiring freezing.
 		 */
 		all_visible = true;
+		all_visible_or_dead = true;
 		has_dead_tuples = false;
 		nfrozen = 0;
 		hastup = false;
@@ -759,6 +761,7 @@ lazy_scan_heap(Relation onerel, LVRelStats *vacrelstats,
 						if (!(tuple.t_data->t_infomask & HEAP_XMIN_COMMITTED))
 						{
 							all_visible = false;
+							all_visible_or_dead = false;
 							break;
 						}
 
@@ -770,6 +773,7 @@ lazy_scan_heap(Relation onerel, LVRelStats *vacrelstats,
 						if (!TransactionIdPrecedes(xmin, OldestXmin))
 						{
 							all_visible = false;
+							all_visible_or_dead = false;
 							break;
 						}
 
@@ -786,14 +790,17 @@ lazy_scan_heap(Relation onerel, LVRelStats *vacrelstats,
 					 */
 					nkeep += 1;
 					all_visible = false;
+					all_visible_or_dead = false;
 					break;
 				case HEAPTUPLE_INSERT_IN_PROGRESS:
 					/* This is an expected case during concurrent vacuum */
 					all_visible = false;
+					all_visible_or_dead = false;
 					break;
 				case HEAPTUPLE_DELETE_IN_PROGRESS:
 					/* This is an expected case during concurrent vacuum */
 					all_visible = false;
+					all_visible_or_dead = false;
 					break;
 				default:
 					elog(ERROR, "unexpected HeapTupleSatisfiesVacuum result");
@@ -858,6 +865,24 @@ lazy_scan_heap(Relation onerel, LVRelStats *vacrelstats,
 			 */
 			vacrelstats->num_dead_tuples = 0;
 			vacuumed_pages++;
+
+			/*
+			 * If the page only contains either all-visible or dead tuples, the
+			 * dead tuples themselves would have been removed by the
+			 * lazy_vacuum_page(). So we can now mark the page all-visible
+			 */
+			if (all_visible_or_dead)
+				all_visible = true;
+		}
+
+		/* 
+		 * Clear the PD_ALL_VISIBLE_OR_DEAD flag which might be left over from
+		 * the previous failed vacuum. We may set it again below
+		 */
+		if (PageIsAllVisibleOrDead(page))
+		{
+			PageClearAllVisibleOrDead(page);
+			MarkBufferDirty(buf);
 		}
 
 		freespace = PageGetHeapFreeSpace(page);
@@ -921,6 +946,23 @@ lazy_scan_heap(Relation onerel, LVRelStats *vacrelstats,
 			MarkBufferDirty(buf);
 			visibilitymap_clear(onerel, blkno, vmbuffer);
 		}
+		/*
+		 * If the page contains only all-visible and dead tuples, remember that
+		 * information in the page header. The page will get marked all-visible
+		 * if no new actions happens between now and the second phase of
+		 * vacuum. But the visibilitymap_set() function also needs the
+		 * cutoff-xid for conflict resolution at the Hot Standby. We reuse the
+		 * pd_prune_xid field in the page header to store this cutoff-xid.
+		 *
+		 * Note: We don't override the pd_prune_xid if its already set. This is
+		 * probably not required since the page does not contain any
+		 * prunable items so the pd_prune_xid is of little use.
+		 */
+		else if (all_visible_or_dead && !PageXidIsPruneXid(page))
+		{
+			PageSetAllVisibleOrDead(page, visibility_cutoff_xid);
+			MarkBufferDirty(buf);
+		}
 
 		UnlockReleaseBuffer(buf);
 
@@ -1096,6 +1138,35 @@ lazy_vacuum_page(Relation onerel, BlockNumber blkno, Buffer buffer,
 		unused[uncnt++] = toff;
 	}
 
+	/*
+	 * The page is marked as PD_ALL_VISIBLE_OR_DEAD. The dead tuples we
+	 * just removed, so what remains must be all-visible. Mark the page
+	 * all-visible and also update the visibility map.
+	 */ 
+	if (PageIsAllVisibleOrDead(page))
+	{
+		Buffer vmbuffer = InvalidBuffer;
+
+		/*
+		 * We had at least one dead tuple on this page. So we must not have
+		 * marked this page as all-visible in the visibility map. But if we
+		 * had, then at least show a WARNING here
+		 */
+		if (visibilitymap_test(onerel, blkno, &vmbuffer))
+			elog(WARNING, "page is not marked all-visible but visibility map bit is set in relation \"%s\" page %u",
+				 RelationGetRelationName(onerel), blkno);
+		else
+			visibilitymap_set(onerel, blkno, InvalidXLogRecPtr, vmbuffer,
+					((PageHeader) (page))->pd_prune_xid); 
+
+		PageClearAllVisibleOrDead(page);
+		PageSetAllVisible(page);
+
+		if (BufferIsValid(vmbuffer))
+			ReleaseBuffer(vmbuffer);
+		vmbuffer = InvalidBuffer;
+	}
+
 	PageRepairFragmentation(page);
 
 	MarkBufferDirty(buffer);
diff --git a/src/include/storage/bufpage.h b/src/include/storage/bufpage.h
index fc3a69b..c2ab194 100644
--- a/src/include/storage/bufpage.h
+++ b/src/include/storage/bufpage.h
@@ -156,14 +156,22 @@ typedef PageHeaderData *PageHeader;
  * PD_PAGE_FULL is set if an UPDATE doesn't find enough free space in the
  * page for its new tuple version; this suggests that a prune is needed.
  * Again, this is just a hint.
+ *
+ * PD_ALL_VISIBLE_OR_DEAD is set if the first phase of VACUUM encounters a page
+ * which is all-visible except one or more dead tuples which will be removed by
+ * the second phase of VACUUM.
  */
 #define PD_HAS_FREE_LINES	0x0001		/* are there any unused line pointers? */
 #define PD_PAGE_FULL		0x0002		/* not enough free space for new
 										 * tuple? */
 #define PD_ALL_VISIBLE		0x0004		/* all tuples on page are visible to
 										 * everyone */
+#define PD_ALL_VISIBLE_OR_DEAD	0x0008	/* tuples on the page are either
+										   visible to everyone or dead for
+										   everyone */
+#define PD_XID_IS_PRUNEXID	0x0010		/* pd_prune_xid is prune xid */
 
-#define PD_VALID_FLAG_BITS	0x0007		/* OR of all valid pd_flags bits */
+#define PD_VALID_FLAG_BITS	0x001F		/* OR of all valid pd_flags bits */
 
 /*
  * Page layout version number 0 is for pre-7.3 Postgres releases.
@@ -354,9 +362,31 @@ typedef PageHeaderData *PageHeader;
 #define PageClearAllVisible(page) \
 	(((PageHeader) (page))->pd_flags &= ~PD_ALL_VISIBLE)
 
+#define PageXidIsPruneXid(page) \
+	(((PageHeader) (page))->pd_flags & PD_XID_IS_PRUNEXID)
+#define PageSetXidIsPruneXid(page) \
+	(((PageHeader) (page))->pd_flags |= PD_XID_IS_PRUNEXID)
+#define PageClearXidIsPruneXid(page) \
+	(((PageHeader) (page))->pd_flags &= ~PD_XID_IS_PRUNEXID)
+
+#define PageIsAllVisibleOrDead(page) \
+	(((PageHeader) (page))->pd_flags & PD_ALL_VISIBLE_OR_DEAD)
+#define PageSetAllVisibleOrDead(page, xid) \
+do { \
+	AssertMacro(!PageXidIsPruneXid(page)); \
+	(((PageHeader) (page))->pd_flags |= PD_ALL_VISIBLE_OR_DEAD); \
+	((PageHeader) (page))->pd_prune_xid = (xid); \
+} while (0)
+#define PageClearAllVisibleOrDead(page) \
+do { \
+	(((PageHeader) (page))->pd_flags &= ~PD_ALL_VISIBLE_OR_DEAD); \
+	((PageHeader) (page))->pd_prune_xid = InvalidTransactionId; \
+} while (0)
+
 #define PageIsPrunable(page, oldestxmin) \
 ( \
 	AssertMacro(TransactionIdIsNormal(oldestxmin)), \
+  	PageXidIsPruneXid(page) && \
 	TransactionIdIsValid(((PageHeader) (page))->pd_prune_xid) && \
 	TransactionIdPrecedes(((PageHeader) (page))->pd_prune_xid, oldestxmin) \
 )
@@ -364,11 +394,18 @@ typedef PageHeaderData *PageHeader;
 do { \
 	Assert(TransactionIdIsNormal(xid)); \
 	if (!TransactionIdIsValid(((PageHeader) (page))->pd_prune_xid) || \
+		!PageXidIsPruneXid(page) || \
 		TransactionIdPrecedes(xid, ((PageHeader) (page))->pd_prune_xid)) \
+	{ \
 		((PageHeader) (page))->pd_prune_xid = (xid); \
+		PageSetXidIsPruneXid(page); \
+	} \
 } while (0)
 #define PageClearPrunable(page) \
-	(((PageHeader) (page))->pd_prune_xid = InvalidTransactionId)
+do { \
+	(((PageHeader) (page))->pd_prune_xid = InvalidTransactionId); \
+	PageClearXidIsPruneXid(page); \
+} while (0)
 
 
 /* ----------------------------------------------------------------