Handle no-op visibility map set during redo

add323da40 consolidated visibility map updates into XLOG_HEAP2_PRUNE
records. It assumed that a record setting the visibility map would
always modify the VM page during replay. That is not true when the
requested bits are already set on the standby.

Primary and standby visibility maps can diverge for several reasons.
This issue was exposed by a CREATE DATABASE ... STRATEGY WAL_LOG bug.
After a standby with an out-of-date VM is promoted, VACUUM may set bits
that were already present on the former primary. When the former primary
later replays that record as a standby, visibilitymap_set() is a no-op.

Restore visibilitymap_set()'s former API, which returns the state of the
VM bits before setting the requested flags. Use that result to set the
VM page LSN only when the visibilitymap_set() updates the page.

Reported-by: Rogers Wang <rogers.ww@qq.com>
Discussion: https://postgr.es/m/tencent_2E870046716FD94285045E96505A2D4E2908%40qq.com
Backpatch-through: 19

src/backend/access/heap/heapam.c

@@ -2484,11 +2484,11 @@ heap_multi_insert(Relation relation, TupleTableSlot **slots, int ntuples,
 		{
 			PageSetAllVisible(page);
 			PageClearPrunable(page);
-			visibilitymap_set(BufferGetBlockNumber(buffer),
-							  vmbuffer,
-							  VISIBILITYMAP_ALL_VISIBLE |
-							  VISIBILITYMAP_ALL_FROZEN,
-							  relation->rd_locator);
+			(void) visibilitymap_set(BufferGetBlockNumber(buffer),
+									 vmbuffer,
+									 VISIBILITYMAP_ALL_VISIBLE |
+									 VISIBILITYMAP_ALL_FROZEN,
+									 relation->rd_locator);
 		}
 
 		/*

src/backend/access/heap/heapam_xlog.c

@@ -286,10 +286,8 @@ heap_xlog_prune_freeze(XLogReaderState *record)
 		if (PageIsNew(vmpage))
 			PageInit(vmpage, BLCKSZ, 0);
 
-		visibilitymap_set(blkno, vmbuffer, vmflags, rlocator);
-
-		Assert(BufferIsDirty(vmbuffer));
-		PageSetLSN(vmpage, lsn);
+		if (visibilitymap_set(blkno, vmbuffer, vmflags, rlocator) != vmflags)
+			PageSetLSN(vmpage, lsn);
 	}
 
 	if (BufferIsValid(vmbuffer))
@@ -706,19 +704,15 @@ heap_xlog_multi_insert(XLogReaderState *record)
 									  &vmbuffer) == BLK_NEEDS_REDO)
 	{
 		Page		vmpage = BufferGetPage(vmbuffer);
+		uint8		vmflags = VISIBILITYMAP_ALL_VISIBLE |
+			VISIBILITYMAP_ALL_FROZEN;
 
 		/* initialize the page if it was read as zeros */
 		if (PageIsNew(vmpage))
 			PageInit(vmpage, BLCKSZ, 0);
 
-		visibilitymap_set(blkno,
-						  vmbuffer,
-						  VISIBILITYMAP_ALL_VISIBLE |
-						  VISIBILITYMAP_ALL_FROZEN,
-						  rlocator);
-
-		Assert(BufferIsDirty(vmbuffer));
-		PageSetLSN(vmpage, lsn);
+		if (visibilitymap_set(blkno, vmbuffer, vmflags, rlocator) != vmflags)
+			PageSetLSN(vmpage, lsn);
 	}
 
 	if (BufferIsValid(vmbuffer))

src/backend/access/heap/pruneheap.c

@@ -1315,8 +1315,9 @@ heap_page_prune_and_freeze(PruneFreezeParams *params,
 			 */
 			PageSetAllVisible(prstate.page);
 			PageClearPrunable(prstate.page);
-			visibilitymap_set(prstate.block, prstate.vmbuffer, prstate.new_vmbits,
-							  prstate.relation->rd_locator);
+			(void) visibilitymap_set(prstate.block, prstate.vmbuffer,
+									 prstate.new_vmbits,
+									 prstate.relation->rd_locator);
 		}
 
 		MarkBufferDirty(prstate.buffer);

src/backend/access/heap/vacuumlazy.c

@@ -1965,11 +1965,11 @@ lazy_scan_new_or_empty(LVRelState *vacrel, Buffer buf, BlockNumber blkno,
 
 			PageSetAllVisible(page);
 			PageClearPrunable(page);
-			visibilitymap_set(blkno,
-							  vmbuffer,
-							  VISIBILITYMAP_ALL_VISIBLE |
-							  VISIBILITYMAP_ALL_FROZEN,
-							  vacrel->rel->rd_locator);
+			(void) visibilitymap_set(blkno,
+									 vmbuffer,
+									 VISIBILITYMAP_ALL_VISIBLE |
+									 VISIBILITYMAP_ALL_FROZEN,
+									 vacrel->rel->rd_locator);
 
 			/*
 			 * Emit WAL for setting PD_ALL_VISIBLE on the heap page and
@@ -2847,9 +2847,9 @@ lazy_vacuum_heap_page(LVRelState *vacrel, BlockNumber blkno, Buffer buffer,
 		 */
 		PageSetAllVisible(page);
 		PageClearPrunable(page);
-		visibilitymap_set(blkno,
-						  vmbuffer, vmflags,
-						  vacrel->rel->rd_locator);
+		(void) visibilitymap_set(blkno,
+								 vmbuffer, vmflags,
+								 vacrel->rel->rd_locator);
 		conflict_xid = newest_live_xid;
 	}
 

src/backend/access/heap/visibilitymap.c

@@ -252,8 +252,10 @@ visibilitymap_pin_ok(BlockNumber heapBlk, Buffer vmbuf)
  * containing heapBlk.
  *
  * rlocator is used only for debugging messages.
+ *
+ * Returns the state of the page's VM bits before setting flags.
  */
-void
+uint8
 visibilitymap_set(BlockNumber heapBlk,
 				  Buffer vmBuf, uint8 flags,
 				  const RelFileLocator rlocator)
@@ -296,6 +298,8 @@ visibilitymap_set(BlockNumber heapBlk,
 		map[mapByte] |= (flags << mapOffset);
 		MarkBufferDirty(vmBuf);
 	}
+
+	return status;
 }
 
 /*

src/include/access/visibilitymap.h

@@ -31,9 +31,9 @@ extern bool visibilitymap_clear(Relation rel, BlockNumber heapBlk,
 extern void visibilitymap_pin(Relation rel, BlockNumber heapBlk,
 							  Buffer *vmbuf);
 extern bool visibilitymap_pin_ok(BlockNumber heapBlk, Buffer vmbuf);
-extern void visibilitymap_set(BlockNumber heapBlk,
-							  Buffer vmBuf, uint8 flags,
-							  const RelFileLocator rlocator);
+extern uint8 visibilitymap_set(BlockNumber heapBlk,
+							   Buffer vmBuf, uint8 flags,
+							   const RelFileLocator rlocator);
 extern uint8 visibilitymap_get_status(Relation rel, BlockNumber heapBlk, Buffer *vmbuf);
 extern void visibilitymap_count(Relation rel, BlockNumber *all_visible, BlockNumber *all_frozen);
 extern BlockNumber visibilitymap_prepare_truncate(Relation rel,