Set persistentTid and persistentSerialNum in XLog record in log_newpa…

…ge().

The log_newpage() routine allocates an XLog record on the stack with
uninitialized persistentTid and persistentSerialNum fields.  Insertion of these
XLog records causes problems in XLog replay during recovery.

This commit ensures that the XLog records written by log_newpage() always have
the persistentTid and persistentSerial set correctly.

This commit also makes the test
src/test/tinc/tincrepo/mpp/gpdb/tests/storage/crashrecovery/test_reindex_pg_class.py
pass.

Signed off by: Asim Praveen <[email protected]>, Amil Khanzada
<[email protected]>

src/backend/access/bitmap/bitmapattutil.c

@@ -126,12 +126,11 @@ _bitmap_create_lov_heapandindex(Relation rel,
 		{
 			// Fetch gp_persistent_relation_node information that will be added to XLOG record.
 			RelationFetchGpRelationNodeForXLog(lovIndex);
-
-			log_newpage(&lovIndex->rd_node,
-						BufferGetBlockNumber(btree_metabuf),
+
+			log_newpage_rel(lovIndex, BufferGetBlockNumber(btree_metabuf),
 						btree_metapage);
 		}
-		
+
 		/* This cache value is not valid anymore. */
 		if (lovIndex->rd_amcache)
 		{

src/backend/access/heap/heapam.c

@@ -4926,32 +4926,40 @@ log_heap_move(Relation reln, Buffer oldbuf, ItemPointerData from,
 	return log_heap_update(reln, oldbuf, from, newbuf, newtup, true);
 }
 
+
 /*
- * Perform XLogInsert of a HEAP_NEWPAGE record to WAL. Caller is responsible
- * for writing the page to disk after calling this routine.
+ * Insert HEAP_NEWPAGE record into XLOG.  Caller is responsible for providing a
+ * xl_heap_newpage record with valid blkno, persistent TID, persistent serial
+ * number, and relfilenode set.
  *
  * Note: all current callers build pages in private memory and write them
  * directly to smgr, rather than using bufmgr.	Therefore there is no need
  * to pass a buffer ID to XLogInsert, nor to perform MarkBufferDirty within
  * the critical section.
  *
- * Note: the NEWPAGE log record is used for both heaps and indexes, so do
+ * Note: the NEWPAGE log record is used for both heap and indexes, so do
  * not do anything that assumes we are touching a heap.
+ *
  */
-XLogRecPtr
-log_newpage(RelFileNode *rnode, BlockNumber blkno, Page page)
+static XLogRecPtr
+log_newpage_internal(xl_heap_newpage *xlrec, Page page)
+
 {
-	xl_heap_newpage xlrec;
-	XLogRecPtr	recptr;
+	XLogRecPtr  recptr;
 	XLogRecData rdata[2];
 
+	Assert(!RelFileNode_IsEmpty(&xlrec->heapnode.node));
+	Assert(BlockNumberIsValid(xlrec->blkno));
+	if (!IsBootstrapProcessingMode() && !gp_before_persistence_work)
+	{
+		Assert(ItemPointerIsValid(&xlrec->heapnode.persistentTid));
+		Assert(xlrec->heapnode.persistentSerialNum);
+	}
+
 	/* NO ELOG(ERROR) from here till newpage op is logged */
 	START_CRIT_SECTION();
 
-	xlrec.heapnode.node = *rnode;
-	xlrec.blkno = blkno;
-
-	rdata[0].data = (char *) &xlrec;
+	rdata[0].data = (char *) xlrec;
 	rdata[0].len = SizeOfHeapNewpage;
 	rdata[0].buffer = InvalidBuffer;
 	rdata[0].next = &(rdata[1]);
@@ -4978,6 +4986,40 @@ log_newpage(RelFileNode *rnode, BlockNumber blkno, Page page)
 	return recptr;
 }
 
+
+/*
+ * This is a wrapper over log_newpage_internal() to be used when the Relation
+ * object contains the persistentTid and persistentSerialNum.
+ */
+XLogRecPtr
+log_newpage_rel(Relation rel, BlockNumber blkno, Page page)
+{
+	xl_heap_newpage xlrec;
+	xl_heapnode_set(&xlrec.heapnode, rel);
+	xlrec.blkno = blkno;
+	return log_newpage_internal(&xlrec, page);
+}
+
+
+/*
+ * This is a wrapper over log_newpage_internal to be used when we don't have a
+ * Relation object available with persistentTid and persistentSerialNum.  In
+ * which case, the persistentTid and persistentSerialNum are explicitly passed
+ * along with the relFileNode for the relation.
+ */
+XLogRecPtr
+log_newpage_relFileNode(RelFileNode *relFileNode, BlockNumber blkno, Page page,
+						ItemPointer persistentTid, int64 persistentSerialNum)
+{
+	xl_heap_newpage xlrec;
+	xlrec.heapnode.node = *relFileNode;
+	xlrec.heapnode.persistentTid = *persistentTid;
+	xlrec.heapnode.persistentSerialNum = persistentSerialNum;
+	xlrec.blkno = blkno;
+	return log_newpage_internal(&xlrec, page);
+}
+
+
 /*
  * Handles CLEAN and CLEAN_MOVE record types
  */

src/backend/access/heap/rewriteheap.c

@@ -267,9 +267,8 @@ end_heap_rewrite(RewriteState state)
 	if (state->rs_buffer_valid)
 	{
 		if (state->rs_use_wal)
-			log_newpage(&state->rs_new_rel->rd_node,
-						state->rs_blockno,
-						state->rs_buffer);
+			log_newpage_rel(state->rs_new_rel,state->rs_blockno, state->rs_buffer);
+
 		RelationOpenSmgr(state->rs_new_rel);
 		smgrextend(state->rs_new_rel->rd_smgr, state->rs_blockno,
 				   (char *) state->rs_buffer, true);
@@ -604,9 +603,7 @@ raw_heap_insert(RewriteState state, HeapTuple tup)
 
 			/* XLOG stuff */
 			if (state->rs_use_wal)
-				log_newpage(&state->rs_new_rel->rd_node,
-							state->rs_blockno,
-							page);
+				log_newpage_rel(state->rs_new_rel, state->rs_blockno, page);
 
 			/*
 			 * Now write the page. We say isTemp = true even if it's not a

src/backend/access/nbtree/nbtsort.c

@@ -289,7 +289,7 @@ _bt_blwritepage(BTWriteState *wstate, Page page, BlockNumber blkno)
 	if (wstate->btws_use_wal)
 	{
 		/* We use the heap NEWPAGE record type for this */
-		log_newpage(&wstate->index->rd_node, blkno, page);
+		log_newpage_rel(wstate->index, blkno, page);
 	}
 
 	else
@@ -322,7 +322,7 @@ _bt_blwritepage(BTWriteState *wstate, Page page, BlockNumber blkno)
 		// -------- MirroredLock ----------
 	}
 
-	
+
 	// -------- MirroredLock ----------
 	// UNDONE: Unfortunately, I think we write temp relations to the mirror...
 	LWLockAcquire(MirroredLock, LW_SHARED);

src/backend/commands/tablecmds.c

@@ -10593,7 +10593,10 @@ copy_buffer_pool_data(Relation rel, SMgrRelation dst,
 
 		/* XLOG stuff */
 		if (useWal)
-			log_newpage(&dst->smgr_rnode, blkno, page);
+		{
+			log_newpage_relFileNode(&dst->smgr_rnode, blkno, page, persistentTid,
+						persistentSerialNum);
+		}
 
 		// -------- MirroredLock ----------
 		LWLockAcquire(MirroredLock, LW_SHARED);

src/include/access/heapam.h

@@ -308,7 +308,14 @@ extern XLogRecPtr log_heap_clean(Relation reln, Buffer buffer,
 extern XLogRecPtr log_heap_freeze(Relation reln, Buffer buffer,
 				TransactionId cutoff_xid,
 				OffsetNumber *offsets, int offcnt);
-extern XLogRecPtr log_newpage(RelFileNode *rnode, BlockNumber blk, Page page);
+
+extern XLogRecPtr log_newpage_rel(Relation rel, BlockNumber blkno,
+								  Page page);
+
+extern XLogRecPtr log_newpage_relFileNode(RelFileNode *relFileNode,
+										  BlockNumber blkno, Page page,
+										  ItemPointer persistentTid,
+										  int64 persistentSerialNum);
 
 /* in common/heaptuple.c */
 extern Size heap_compute_data_size(TupleDesc tupleDesc,