Adjust tuple data lookup logic in multi-insert logical decoding

As of now, logical decoding of a multi-insert has been scanning all
xl_multi_insert_tuple entries only if XLH_INSERT_CONTAINS_NEW_TUPLE was
getting set in the record.  This is not an issue on HEAD as multi-insert
records are not used for system catalogs, but the logical decoding logic
includes all the code necessary to handle that properly, except that the
code missed to iterate correctly over all xl_multi_insert_tuple entries
when the flag is not set.  Hence, when trying to use multi-insert for
system catalogs, an assertion would be triggered.

An upcoming patch is going to make use of multi-insert for system
catalogs, and this fixes the logic to make sure that all entries are
scanned correctly without softening the existing assertions.

Reported-by: Daniel Gustafsson
Author: Michael Paquier
Reviewed-by: Daniel Gustafsson
Discussion: https://postgr.es/m/CBFFD532-C033-49EB-9A5A-F67EAEE9EB0B@yesql.se

src/backend/replication/logical/decode.c

@@ -900,7 +900,12 @@ DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 	if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
 		return;
 
+	/*
+	 * As multi_insert is not used for catalogs yet, the block should always
+	 * have data even if a full-page write of it is taken.
+	 */
 	tupledata = XLogRecGetBlockData(r, 0, &tuplelen);
+	Assert(tupledata != NULL);
 
 	data = tupledata;
 	for (i = 0; i < xlrec->ntuples; i++)
@@ -916,6 +921,10 @@ DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
 		memcpy(&change->data.tp.relnode, &rnode, sizeof(RelFileNode));
 
+		xlhdr = (xl_multi_insert_tuple *) SHORTALIGN(data);
+		data = ((char *) xlhdr) + SizeOfMultiInsertTuple;
+		datalen = xlhdr->datalen;
+
 		/*
 		 * CONTAINS_NEW_TUPLE will always be set currently as multi_insert
 		 * isn't used for catalogs, but better be future proof.
@@ -927,10 +936,6 @@ DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 		{
 			HeapTupleHeader header;
 
-			xlhdr = (xl_multi_insert_tuple *) SHORTALIGN(data);
-			data = ((char *) xlhdr) + SizeOfMultiInsertTuple;
-			datalen = xlhdr->datalen;
-
 			change->data.tp.newtuple =
 				ReorderBufferGetTupleBuf(ctx->reorder, datalen);
 
@@ -953,8 +958,6 @@ DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 			memcpy((char *) tuple->tuple.t_data + SizeofHeapTupleHeader,
 				   (char *) data,
 				   datalen);
-			data += datalen;
-
 			header->t_infomask = xlhdr->t_infomask;
 			header->t_infomask2 = xlhdr->t_infomask2;
 			header->t_hoff = xlhdr->t_hoff;
@@ -973,6 +976,9 @@ DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
 		ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r),
 								 buf->origptr, change);
+
+		/* move to the next xl_multi_insert_tuple entry */
+		data += datalen;
 	}
 	Assert(data == tupledata + tuplelen);
 }