Subject: [PATCH 3/7] Move XLOG stuff from heap_insert and heap_delete

From: Kyotaro Horiguchi <[email-masked]>
Date: Mon, 25 Mar 2019 13:29:50 +0900

Succeeding commit makes heap_update emit insert and delete WAL
records. Move out XLOG stuff for insert and delete so that heap_update
can use the stuff.
---
 src/backend/access/heap/heapam.c | 277 ++++++++++++++++++++++-----------------
 1 file changed, 157 insertions(+), 120 deletions(-)

diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 65536c7214..fe5d939c45 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -71,6 +71,11 @@
 
 static HeapTuple heap_prepare_insert(Relation relation, HeapTuple tup,
 					TransactionId xid, CommandId cid, int options);
+static XLogRecPtr log_heap_insert(Relation relation, Buffer buffer,
+				HeapTuple heaptup, int options, bool all_visible_cleared);
+static XLogRecPtr log_heap_delete(Relation relation, Buffer buffer,
+				HeapTuple tp, HeapTuple old_key_tuple, TransactionId new_xmax,
+				bool changingPart, bool all_visible_cleared);
 static XLogRecPtr log_heap_update(Relation reln, Buffer oldbuf,
 				Buffer newbuf, HeapTuple oldtup,
 				HeapTuple newtup, HeapTuple old_key_tup,
@@ -1889,6 +1894,7 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
 	TransactionId xid = GetCurrentTransactionId();
 	HeapTuple	heaptup;
 	Buffer		buffer;
+	Page		page;
 	Buffer		vmbuffer = InvalidBuffer;
 	bool		all_visible_cleared = false;
 
@@ -1925,16 +1931,18 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
 	 */
 	CheckForSerializableConflictIn(relation, NULL, InvalidBuffer);
 
+	page = BufferGetPage(buffer);
+
 	/* NO EREPORT(ERROR) from here till changes are logged */
 	START_CRIT_SECTION();
 
 	RelationPutHeapTuple(relation, buffer, heaptup,
 						 (options & HEAP_INSERT_SPECULATIVE) != 0);
 
-	if (PageIsAllVisible(BufferGetPage(buffer)))
+	if (PageIsAllVisible(page))
 	{
 		all_visible_cleared = true;
-		PageClearAllVisible(BufferGetPage(buffer));
+		PageClearAllVisible(page);
 		visibilitymap_clear(relation,
 							ItemPointerGetBlockNumber(&(heaptup->t_self)),
 							vmbuffer, VISIBILITYMAP_VALID_BITS);
@@ -1956,76 +1964,11 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
 	/* XLOG stuff */
 	if (!(options & HEAP_INSERT_SKIP_WAL) && RelationNeedsWAL(relation))
 	{
-		xl_heap_insert xlrec;
-		xl_heap_header xlhdr;
 		XLogRecPtr	recptr;
-		Page		page = BufferGetPage(buffer);
-		uint8		info = XLOG_HEAP_INSERT;
-		int			bufflags = 0;
-
-		/*
-		 * If this is a catalog, we need to transmit combocids to properly
-		 * decode, so log that as well.
-		 */
-		if (RelationIsAccessibleInLogicalDecoding(relation))
-			log_heap_new_cid(relation, heaptup);
-
-		/*
-		 * If this is the single and first tuple on page, we can reinit the
-		 * page instead of restoring the whole thing.  Set flag, and hide
-		 * buffer references from XLogInsert.
-		 */
-		if (ItemPointerGetOffsetNumber(&(heaptup->t_self)) == FirstOffsetNumber &&
-			PageGetMaxOffsetNumber(page) == FirstOffsetNumber)
-		{
-			info |= XLOG_HEAP_INIT_PAGE;
-			bufflags |= REGBUF_WILL_INIT;
-		}
-
-		xlrec.offnum = ItemPointerGetOffsetNumber(&heaptup->t_self);
-		xlrec.flags = 0;
-		if (all_visible_cleared)
-			xlrec.flags |= XLH_INSERT_ALL_VISIBLE_CLEARED;
-		if (options & HEAP_INSERT_SPECULATIVE)
-			xlrec.flags |= XLH_INSERT_IS_SPECULATIVE;
-		Assert(ItemPointerGetBlockNumber(&heaptup->t_self) == BufferGetBlockNumber(buffer));
-
-		/*
-		 * For logical decoding, we need the tuple even if we're doing a full
-		 * page write, so make sure it's included even if we take a full-page
-		 * image. (XXX We could alternatively store a pointer into the FPW).
-		 */
-		if (RelationIsLogicallyLogged(relation) &&
-			!(options & HEAP_INSERT_NO_LOGICAL))
-		{
-			xlrec.flags |= XLH_INSERT_CONTAINS_NEW_TUPLE;
-			bufflags |= REGBUF_KEEP_DATA;
-		}
-
-		XLogBeginInsert();
-		XLogRegisterData((char *) &xlrec, SizeOfHeapInsert);
-
-		xlhdr.t_infomask2 = heaptup->t_data->t_infomask2;
-		xlhdr.t_infomask = heaptup->t_data->t_infomask;
-		xlhdr.t_hoff = heaptup->t_data->t_hoff;
-
-		/*
-		 * note we mark xlhdr as belonging to buffer; if XLogInsert decides to
-		 * write the whole page to the xlog, we don't need to store
-		 * xl_heap_header in the xlog.
-		 */
-		XLogRegisterBuffer(0, buffer, REGBUF_STANDARD | bufflags);
-		XLogRegisterBufData(0, (char *) &xlhdr, SizeOfHeapHeader);
-		/* PG73FORMAT: write bitmap [+ padding] [+ oid] + data */
-		XLogRegisterBufData(0,
-							(char *) heaptup->t_data + SizeofHeapTupleHeader,
-							heaptup->t_len - SizeofHeapTupleHeader);
-
-		/* filtering by origin on a row level is much more efficient */
-		XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);
-
-		recptr = XLogInsert(RM_HEAP_ID, info);
 
+		recptr = log_heap_insert(relation, buffer, heaptup,
+								 options, all_visible_cleared);
+			
 		PageSetLSN(page, recptr);
 	}
 
@@ -2744,58 +2687,10 @@ l1:
 	 */
 	if (RelationNeedsWAL(relation))
 	{
-		xl_heap_delete xlrec;
-		xl_heap_header xlhdr;
 		XLogRecPtr	recptr;
 
-		/* For logical decode we need combocids to properly decode the catalog */
-		if (RelationIsAccessibleInLogicalDecoding(relation))
-			log_heap_new_cid(relation, &tp);
-
-		xlrec.flags = 0;
-		if (all_visible_cleared)
-			xlrec.flags |= XLH_DELETE_ALL_VISIBLE_CLEARED;
-		if (changingPart)
-			xlrec.flags |= XLH_DELETE_IS_PARTITION_MOVE;
-		xlrec.infobits_set = compute_infobits(tp.t_data->t_infomask,
-											  tp.t_data->t_infomask2);
-		xlrec.offnum = ItemPointerGetOffsetNumber(&tp.t_self);
-		xlrec.xmax = new_xmax;
-
-		if (old_key_tuple != NULL)
-		{
-			if (relation->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
-				xlrec.flags |= XLH_DELETE_CONTAINS_OLD_TUPLE;
-			else
-				xlrec.flags |= XLH_DELETE_CONTAINS_OLD_KEY;
-		}
-
-		XLogBeginInsert();
-		XLogRegisterData((char *) &xlrec, SizeOfHeapDelete);
-
-		XLogRegisterBuffer(0, buffer, REGBUF_STANDARD);
-
-		/*
-		 * Log replica identity of the deleted tuple if there is one
-		 */
-		if (old_key_tuple != NULL)
-		{
-			xlhdr.t_infomask2 = old_key_tuple->t_data->t_infomask2;
-			xlhdr.t_infomask = old_key_tuple->t_data->t_infomask;
-			xlhdr.t_hoff = old_key_tuple->t_data->t_hoff;
-
-			XLogRegisterData((char *) &xlhdr, SizeOfHeapHeader);
-			XLogRegisterData((char *) old_key_tuple->t_data
-							 + SizeofHeapTupleHeader,
-							 old_key_tuple->t_len
-							 - SizeofHeapTupleHeader);
-		}
-
-		/* filtering by origin on a row level is much more efficient */
-		XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);
-
-		recptr = XLogInsert(RM_HEAP_ID, XLOG_HEAP_DELETE);
-
+		recptr = log_heap_delete(relation, buffer, &tp, old_key_tuple, new_xmax,
+								 changingPart, all_visible_cleared);
 		PageSetLSN(page, recptr);
 	}
 
@@ -7045,6 +6940,148 @@ log_heap_visible(RelFileNode rnode, Buffer heap_buffer, Buffer vm_buffer,
 	return recptr;
 }
 
+/*
+ * Perform XLogInsert for a heap-insert operation.  Caller must already
+ * have modified the buffer and marked it dirty.
+ */
+XLogRecPtr
+log_heap_insert(Relation relation, Buffer buffer,
+				HeapTuple heaptup, int options, bool all_visible_cleared)
+{
+	xl_heap_insert xlrec;
+	xl_heap_header xlhdr;
+	uint8		info = XLOG_HEAP_INSERT;
+	int			bufflags = 0;
+	Page		page = BufferGetPage(buffer);
+
+	/*
+	 * If this is a catalog, we need to transmit combocids to properly
+	 * decode, so log that as well.
+	 */
+	if (RelationIsAccessibleInLogicalDecoding(relation))
+		log_heap_new_cid(relation, heaptup);
+
+	/*
+	 * If this is the single and first tuple on page, we can reinit the
+	 * page instead of restoring the whole thing.  Set flag, and hide
+	 * buffer references from XLogInsert.
+	 */
+	if (ItemPointerGetOffsetNumber(&(heaptup->t_self)) == FirstOffsetNumber &&
+		PageGetMaxOffsetNumber(page) == FirstOffsetNumber)
+	{
+		info |= XLOG_HEAP_INIT_PAGE;
+		bufflags |= REGBUF_WILL_INIT;
+	}
+
+	xlrec.offnum = ItemPointerGetOffsetNumber(&heaptup->t_self);
+	xlrec.flags = 0;
+	if (all_visible_cleared)
+		xlrec.flags |= XLH_INSERT_ALL_VISIBLE_CLEARED;
+	if (options & HEAP_INSERT_SPECULATIVE)
+		xlrec.flags |= XLH_INSERT_IS_SPECULATIVE;
+	Assert(ItemPointerGetBlockNumber(&heaptup->t_self) == BufferGetBlockNumber(buffer));
+
+	/*
+	 * For logical decoding, we need the tuple even if we're doing a full
+	 * page write, so make sure it's included even if we take a full-page
+	 * image. (XXX We could alternatively store a pointer into the FPW).
+	 */
+	if (RelationIsLogicallyLogged(relation) &&
+		!(options & HEAP_INSERT_NO_LOGICAL))
+	{
+		xlrec.flags |= XLH_INSERT_CONTAINS_NEW_TUPLE;
+		bufflags |= REGBUF_KEEP_DATA;
+	}
+
+	XLogBeginInsert();
+	XLogRegisterData((char *) &xlrec, SizeOfHeapInsert);
+
+	xlhdr.t_infomask2 = heaptup->t_data->t_infomask2;
+	xlhdr.t_infomask = heaptup->t_data->t_infomask;
+	xlhdr.t_hoff = heaptup->t_data->t_hoff;
+
+	/*
+	 * note we mark xlhdr as belonging to buffer; if XLogInsert decides to
+	 * write the whole page to the xlog, we don't need to store
+	 * xl_heap_header in the xlog.
+	 */
+	XLogRegisterBuffer(0, buffer, REGBUF_STANDARD | bufflags);
+	XLogRegisterBufData(0, (char *) &xlhdr, SizeOfHeapHeader);
+	/* PG73FORMAT: write bitmap [+ padding] [+ oid] + data */
+	XLogRegisterBufData(0,
+						(char *) heaptup->t_data + SizeofHeapTupleHeader,
+						heaptup->t_len - SizeofHeapTupleHeader);
+
+	/* filtering by origin on a row level is much more efficient */
+	XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);
+
+	return XLogInsert(RM_HEAP_ID, info);
+}
+
+/*
+ * Perform XLogInsert for a heap-insert operation.  Caller must already
+ * have modified the buffer and marked it dirty.
+ *
+ * NB: heap_abort_speculative() uses the same xlog record and replay
+ * routines.
+ */
+static XLogRecPtr
+log_heap_delete(Relation relation, Buffer buffer,
+				HeapTuple tp, HeapTuple old_key_tuple, TransactionId new_xmax,
+				bool changingPart, bool all_visible_cleared)
+{
+	xl_heap_delete xlrec;
+	xl_heap_header xlhdr;
+
+	/* For logical decode we need combocids to properly decode the catalog */
+	if (RelationIsAccessibleInLogicalDecoding(relation))
+		log_heap_new_cid(relation, tp);
+
+	xlrec.flags = 0;
+	if (all_visible_cleared)
+		xlrec.flags |= XLH_DELETE_ALL_VISIBLE_CLEARED;
+	if (changingPart)
+		xlrec.flags |= XLH_DELETE_IS_PARTITION_MOVE;
+	xlrec.infobits_set = compute_infobits(tp->t_data->t_infomask,
+										  tp->t_data->t_infomask2);
+	xlrec.offnum = ItemPointerGetOffsetNumber(&tp->t_self);
+	xlrec.xmax = new_xmax;
+
+	if (old_key_tuple != NULL)
+	{
+		if (relation->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
+			xlrec.flags |= XLH_DELETE_CONTAINS_OLD_TUPLE;
+		else
+			xlrec.flags |= XLH_DELETE_CONTAINS_OLD_KEY;
+	}
+
+	XLogBeginInsert();
+	XLogRegisterData((char *) &xlrec, SizeOfHeapDelete);
+
+	XLogRegisterBuffer(0, buffer, REGBUF_STANDARD);
+
+	/*
+	 * Log replica identity of the deleted tuple if there is one
+	 */
+	if (old_key_tuple != NULL)
+	{
+		xlhdr.t_infomask2 = old_key_tuple->t_data->t_infomask2;
+		xlhdr.t_infomask = old_key_tuple->t_data->t_infomask;
+		xlhdr.t_hoff = old_key_tuple->t_data->t_hoff;
+
+		XLogRegisterData((char *) &xlhdr, SizeOfHeapHeader);
+		XLogRegisterData((char *) old_key_tuple->t_data
+						 + SizeofHeapTupleHeader,
+						 old_key_tuple->t_len
+						 - SizeofHeapTupleHeader);
+	}
+
+	/* filtering by origin on a row level is much more efficient */
+	XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);
+
+	return XLogInsert(RM_HEAP_ID, XLOG_HEAP_DELETE);
+}
+
 /*
  * Perform XLogInsert for a heap-update operation.  Caller must already
  * have modified the buffer(s) and marked them dirty.
-- 
2.16.3

----Next_Part(Mon_Mar_25_21_32_04_2019_838)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="v8-0004-Add-infrastructure-to-WAL-logging-skip-feature.patch"