From 67bf139a9355adc93801c5b35b912154504075c0 Mon Sep 17 00:00:00 2001
From: Kyotaro Horiguchi <horikyoga.ntt@gmail.com>
Date: Wed, 20 Jan 2021 23:59:39 +0900
Subject: [PATCH v5 3/3] Poc: Keep page-LSN updated while WAL-skipping.

WAL-skipping optimization omits bumping page LSN when ALTER TABLE SET
TABLESPACE. Also heap_page_prune does the same.  However,
old_snapshot_threshold feature needs page LSN to be kept updated on
these operations so that TestForOldSnapshot properly find identify
whether a snapshot is invalidated.
---
 src/backend/access/heap/pruneheap.c | 14 ++++++++++++++
 src/backend/catalog/storage.c       |  8 ++++++++
 src/include/storage/bufmgr.h        |  3 ++-
 src/include/utils/rel.h             |  4 ++--
 4 files changed, 26 insertions(+), 3 deletions(-)

diff --git a/src/backend/access/heap/pruneheap.c b/src/backend/access/heap/pruneheap.c
index e3a716a2a2..33c0841b36 100644
--- a/src/backend/access/heap/pruneheap.c
+++ b/src/backend/access/heap/pruneheap.c
@@ -70,6 +70,8 @@ static void heap_prune_record_redirect(PruneState *prstate,
 static void heap_prune_record_dead(PruneState *prstate, OffsetNumber offnum);
 static void heap_prune_record_unused(PruneState *prstate, OffsetNumber offnum);
 
+/* XXXXXXXX tmporary modification */
+extern XLogRecPtr gistXLogAssignLSN(void);
 
 /*
  * Optionally prune and repair fragmentation in the specified page.
@@ -331,6 +333,18 @@ heap_page_prune(Relation relation, Buffer buffer,
 
 			PageSetLSN(BufferGetPage(buffer), recptr);
 		}
+		else if (relation->rd_rel->relpersistence == RELPERSISTENCE_PERMANENT)
+		{
+			XLogRecPtr	currlsn = GetXLogInsertRecPtr();
+			Page		page = BufferGetPage(buffer);
+
+			Assert(wal_level < WAL_LEVEL_REPLICA);
+
+			/* we need to update page LSN to notify reader of pruning */
+			if (PageGetLSN(page) == currlsn)
+				currlsn = gistXLogAssignLSN();
+			PageSetLSN(page, currlsn);
+		}
 	}
 	else
 	{
diff --git a/src/backend/catalog/storage.c b/src/backend/catalog/storage.c
index cba7a9ada0..488ac6f760 100644
--- a/src/backend/catalog/storage.c
+++ b/src/backend/catalog/storage.c
@@ -75,6 +75,8 @@ typedef struct PendingRelSync
 static PendingRelDelete *pendingDeletes = NULL; /* head of linked list */
 HTAB	   *pendingSyncHash = NULL;
 
+/* XXXXXXXX tmporary modification */
+extern XLogRecPtr gistXLogAssignLSN(void);
 
 /*
  * AddPendingSync
@@ -414,6 +416,7 @@ RelationCopyStorage(SMgrRelation src, SMgrRelation dst,
 	bool		copying_initfork;
 	BlockNumber nblocks;
 	BlockNumber blkno;
+	XLogRecPtr	fakepagelsn;
 
 	page = (Page) buf.data;
 
@@ -434,6 +437,9 @@ RelationCopyStorage(SMgrRelation src, SMgrRelation dst,
 	use_wal = XLogIsNeeded() &&
 		(relpersistence == RELPERSISTENCE_PERMANENT || copying_initfork);
 
+	if (!use_wal)
+		fakepagelsn = gistXLogAssignLSN();
+
 	nblocks = smgrnblocks(src, forkNum);
 
 	for (blkno = 0; blkno < nblocks; blkno++)
@@ -460,6 +466,8 @@ RelationCopyStorage(SMgrRelation src, SMgrRelation dst,
 		 */
 		if (use_wal)
 			log_newpage(&dst->smgr_rnode.node, forkNum, blkno, page, false);
+		else
+			PageSetLSN(page, fakepagelsn);
 
 		PageSetChecksumInplace(page, blkno);
 
diff --git a/src/include/storage/bufmgr.h b/src/include/storage/bufmgr.h
index e641174798..9ab9e7394c 100644
--- a/src/include/storage/bufmgr.h
+++ b/src/include/storage/bufmgr.h
@@ -19,6 +19,7 @@
 #include "storage/buf.h"
 #include "storage/bufpage.h"
 #include "storage/relfilenode.h"
+#include "utils/rel.h"
 #include "utils/relcache.h"
 #include "utils/snapmgr.h"
 
@@ -289,7 +290,7 @@ TestForOldSnapshot(Snapshot snapshot, Relation relation, Page page)
 		&& ((snapshot)->snapshot_type == SNAPSHOT_MVCC
 			|| (snapshot)->snapshot_type == SNAPSHOT_TOAST)
 		&& !XLogRecPtrIsInvalid((snapshot)->lsn)
-		&& (!XLogIsNeeded() || PageGetLSN(page) > (snapshot)->lsn))
+		&& PageGetLSN(page) > (snapshot)->lsn)
 		TestForOldSnapshot_impl(snapshot, relation);
 }
 
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index f58d65cf28..35e53c8d74 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -618,8 +618,8 @@ typedef struct ViewOptions
  *		decoding snapshot.
  */
 #define RelationIsAccessibleInLogicalDecoding(relation) \
-	(XLogLogicalInfoActive() && \
-	 RelationNeedsWAL(relation) && \
+	(XLogLogicalInfoActive() &&											\
+	 (relation)->rd_rel->relpersistence == RELPERSISTENCE_PERMANENT &&	\
 	 (IsCatalogRelation(relation) || RelationIsUsedAsCatalogTable(relation)))
 
 /*
-- 
2.27.0