Update last written LSN for gin/gist index metadata (#182)

* Update last written LSN for gin/gist index metadata

* Replace SetLastWrittenLSN with family of SetLastWrittenLSNFFor* functions

Author: knizhnik

contrib/neon/pagestore_smgr.c

@@ -84,11 +84,6 @@ static char *hexdump_page(char *page);
 
 const int	SmgrTrace = DEBUG5;
 
-/*
- * Pseudo block number used to associate LSN with relation metadata (relation size)
- */
-#define REL_METADATA_PSEUDO_BLOCKNO InvalidBlockNumber
-
 page_server_api *page_server;
 
 /* GUCs */
@@ -563,7 +558,7 @@ zenith_wallog_page(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 	 * Remember the LSN on this page. When we read the page again, we must
 	 * read the same or newer version of it.
 	 */
-	SetLastWrittenLSN(lsn, reln->smgr_rnode.node.relNode, blocknum, blocknum);
+	SetLastWrittenLSNForBlock(lsn, reln->smgr_rnode.node.relNode, blocknum);
 }
 
 
@@ -910,7 +905,7 @@ zenith_extend(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 		mdextend(reln, forkNum, blkno, buffer, skipFsync);
 #endif
 
-	SetLastWrittenLSN(lsn, reln->smgr_rnode.node.relNode, REL_METADATA_PSEUDO_BLOCKNO, REL_METADATA_PSEUDO_BLOCKNO);
+	SetLastWrittenLSNForRelation(lsn, reln->smgr_rnode.node.relNode);
 }
 
 /*
@@ -1442,7 +1437,7 @@ zenith_truncate(SMgrRelation reln, ForkNumber forknum, BlockNumber nblocks)
 	 * Truncate may affect several chunks of relations. So we should either update last written LSN for all of them,
 	 * either update LSN for "dummy" metadata block. Second approach seems to be more efficient.
 	 */
-	SetLastWrittenLSN(lsn, reln->smgr_rnode.node.relNode, REL_METADATA_PSEUDO_BLOCKNO, REL_METADATA_PSEUDO_BLOCKNO);
+	SetLastWrittenLSNForRelation(lsn, reln->smgr_rnode.node.relNode);
 
 #ifdef DEBUG_COMPARE_LOCAL
 	if (IS_LOCAL_REL(reln))

src/backend/access/gin/gininsert.c

@@ -421,7 +421,8 @@ ginbuild(Relation heap, Relation index, IndexInfo *indexInfo)
 		log_newpage_range(index, MAIN_FORKNUM,
 						  0, RelationGetNumberOfBlocks(index),
 						  true);
-		SetLastWrittenLSN(XactLastRecEnd, index->rd_smgr->smgr_rnode.node.relNode, 0, RelationGetNumberOfBlocks(index));
+		SetLastWrittenLSNForBlockRange(XactLastRecEnd, index->rd_smgr->smgr_rnode.node.relNode, 0, RelationGetNumberOfBlocks(index));
+		SetLastWrittenLSNForRelation(XactLastRecEnd, index->rd_smgr->smgr_rnode.node.relNode);
 	}
 
 	smgr_end_unlogged_build(index->rd_smgr);

src/backend/access/gist/gistbuild.c

@@ -335,9 +335,10 @@ gistbuild(Relation heap, Relation index, IndexInfo *indexInfo)
 			log_newpage_range(index, MAIN_FORKNUM,
 							  0, RelationGetNumberOfBlocks(index),
 							  true);
-			SetLastWrittenLSN(XactLastRecEnd,
+			SetLastWrittenLSNForBlockRange(XactLastRecEnd,
 							  index->rd_smgr->smgr_rnode.node.relNode,
 							  0, RelationGetNumberOfBlocks(index));
+			SetLastWrittenLSNForRelation(XactLastRecEnd, index->rd_smgr->smgr_rnode.node.relNode);
 		}
 		smgr_end_unlogged_build(index->rd_smgr);
 	}
@@ -470,8 +471,9 @@ gist_indexsortbuild(GISTBuildState *state)
 
 		lsn = log_newpage(&state->indexrel->rd_node, MAIN_FORKNUM, GIST_ROOT_BLKNO,
 					pagestate->page, true);
-		SetLastWrittenLSN(lsn, state->indexrel->rd_smgr->smgr_rnode.node.relNode,
-							  GIST_ROOT_BLKNO, GIST_ROOT_BLKNO);
+		SetLastWrittenLSNForBlock(lsn, state->indexrel->rd_smgr->smgr_rnode.node.relNode,
+								  GIST_ROOT_BLKNO);
+		SetLastWrittenLSNForRelation(lsn, state->indexrel->rd_smgr->smgr_rnode.node.relNode);
 	}
 
 	pfree(pagestate->page);

src/backend/access/spgist/spginsert.c

@@ -143,8 +143,9 @@ spgbuild(Relation heap, Relation index, IndexInfo *indexInfo)
 		log_newpage_range(index, MAIN_FORKNUM,
 						  0, RelationGetNumberOfBlocks(index),
 						  true);
-		SetLastWrittenLSN(XactLastRecEnd, index->rd_smgr->smgr_rnode.node.relNode,
+		SetLastWrittenLSNForBlockRange(XactLastRecEnd, index->rd_smgr->smgr_rnode.node.relNode,
 						  0, RelationGetNumberOfBlocks(index));
+		SetLastWrittenLSNForRelation(XactLastRecEnd, index->rd_smgr->smgr_rnode.node.relNode);
 	}
 
 	smgr_end_unlogged_build(index->rd_smgr);

src/backend/access/transam/xlog.c

@@ -8987,17 +8987,18 @@ GetLastWrittenLSN(Oid rnode, BlockNumber blkno)
 }
 
 /*
- * SetLastWrittenLSN -- Set maximal LSN of written page.
+ * SetLastWrittenLSNForBlockRange -- Set maximal LSN of written page range.
  * We maintain cache of last written LSNs with limited size and LRU replacement
  * policy. To reduce cache size we store max LSN not for each page, but for
  * bucket (1024 blocks). This cache allows to use old LSN when
  * requesting pages of unchanged or appended relations.
  *
- * rnode can be InvalidOid, in this case maxLastWrittenLsn is updated. SetLastWrittensn with InvalidOid
+ * rnode can be InvalidOid, in this case maxLastWrittenLsn is updated. 
+ * SetLastWrittenLsn with InvalidOid
  * is used by createdb and dbase_redo functions.
  */
 void
-SetLastWrittenLSN(XLogRecPtr lsn, Oid rnode, BlockNumber from, BlockNumber till)
+SetLastWrittenLSNForBlockRange(XLogRecPtr lsn, Oid rnode, BlockNumber from, BlockNumber till)
 {
 	if (lsn == InvalidXLogRecPtr)
 		return;
@@ -9055,6 +9056,33 @@ SetLastWrittenLSN(XLogRecPtr lsn, Oid rnode, BlockNumber from, BlockNumber till)
 	LWLockRelease(LastWrittenLsnLock);
 }
 
+/*
+ * SetLastWrittenLSNForBlock -- Set maximal LSN for block
+ */
+void
+SetLastWrittenLSNForBlock(XLogRecPtr lsn, Oid rnode, BlockNumber blkno)
+{
+	SetLastWrittenLSNForBlockRange(lsn, rnode, blkno, blkno);
+}
+
+/*
+ * SetLastWrittenLSNForRelation -- Set maximal LSN for relation metadata
+ */
+void
+SetLastWrittenLSNForRelation(XLogRecPtr lsn, Oid rnode)
+{
+	SetLastWrittenLSNForBlock(lsn, rnode, REL_METADATA_PSEUDO_BLOCKNO);
+}
+
+/*
+ * SetLastWrittenLSNForDatabase -- Set maximal LSN for the whole database
+ */
+void
+SetLastWrittenLSNForDatabase(XLogRecPtr lsn)
+{
+	SetLastWrittenLSNForBlock(lsn, InvalidOid, 0);
+}
+
 /*
  * RedoStartLsn is set only once by startup process, locking is not required
  * after its exit.

src/backend/commands/dbcommands.c

@@ -675,7 +675,7 @@ createdb(ParseState *pstate, const CreatedbStmt *stmt)
 
 				lsn = XLogInsert(RM_DBASE_ID,
 								 XLOG_DBASE_CREATE | XLR_SPECIAL_REL_UPDATE);
-				SetLastWrittenLSN(lsn, InvalidOid, 0, 0);
+				SetLastWrittenLSNForDatabase(lsn);
 			}
 		}
 		table_endscan(scan);
@@ -2293,8 +2293,7 @@ dbase_redo(XLogReaderState *record)
 		 */
 		{
 			XLogRecPtr	lsn = record->EndRecPtr;
-
-			SetLastWrittenLSN(lsn, InvalidOid, 0, 0);
+			SetLastWrittenLSNForDatabase(lsn);
 		}
 	}
 	else if (info == XLOG_DBASE_DROP)

src/include/access/xlog.h

@@ -31,6 +31,11 @@ extern int	sync_method;
 
 extern PGDLLIMPORT TimeLineID ThisTimeLineID;	/* current TLI */
 
+/*
+ * Pseudo block number used to associate LSN with relation metadata (relation size)
+ */
+#define REL_METADATA_PSEUDO_BLOCKNO InvalidBlockNumber
+
 /*
  * Prior to 8.4, all activity during recovery was carried out by the startup
  * process. This local variable continues to be used in many parts of the
@@ -352,7 +357,10 @@ extern XLogRecPtr GetFlushRecPtr(void);
 extern XLogRecPtr GetLastImportantRecPtr(void);
 extern void RemovePromoteSignalFiles(void);
 
-extern void SetLastWrittenLSN(XLogRecPtr lsn, Oid relfilenode, BlockNumber from, BlockNumber till);
+extern void SetLastWrittenLSNForBlock(XLogRecPtr lsn, Oid relfilenode, BlockNumber blkno);
+extern void SetLastWrittenLSNForBlockRange(XLogRecPtr lsn, Oid relfilenode, BlockNumber from, BlockNumber till);
+extern void SetLastWrittenLSNForDatabase(XLogRecPtr lsn);
+extern void SetLastWrittenLSNForRelation(XLogRecPtr lsn, Oid relfilenode);
 extern XLogRecPtr GetLastWrittenLSN(Oid relfilenode, BlockNumber blkno);
 
 extern XLogRecPtr GetRedoStartLsn(void);