From b42ffc126fe714d3a25bb198fb45ef483c9e5795 Mon Sep 17 00:00:00 2001
From: Matthias van de Meent <boekewurm+postgres@gmail.com>
Date: Wed, 12 Oct 2022 21:16:53 +0200
Subject: [PATCH 3/3] Compactify XLogRecordBlockHeader format, where possible.

- Add 'sequential block IDs' compression
  Usually, blocks are registered with sequential IDs starting with 0.
  Utilize that pattern to deduplicate those fields in that common case.
- Add 'mini block data' field size
  If blocks don't need 2 bytes to store length; we don't those 2 bytes.
- Omit data_length if !BKPBLOCK_HAS_DATA
  FPIs have their own data length bookkeeping, so don't bother with
  those 2 bytes if the field is 0.
---
 src/backend/access/transam/xloginsert.c | 115 +++++++++++++++++++++++-
 src/backend/access/transam/xlogreader.c |  39 +++++++-
 src/include/access/xlogrecord.h         |  25 ++++--
 3 files changed, 167 insertions(+), 12 deletions(-)

diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index 2ac1d0a870..c9d54c0a1e 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -559,6 +559,7 @@ XLogRecordAssemble(RmgrId rmid, uint8 info, uint8 rminfo,
 	XLogRecData *rdt_datas_last;
 	XLogRecord *rechdr = (XLogRecord *) hdr_scratch;
 	char	   *scratch = hdr_scratch;
+	uint8	   *first_blockid = NULL;
 
 	Assert((info & ~(XLR_INFO_USERFLAGS)) == 0);
 
@@ -626,7 +627,10 @@ XLogRecordAssemble(RmgrId rmid, uint8 info, uint8 rminfo,
 		bool		include_image;
 
 		if (!regbuf->in_use)
+		{
+			first_blockid = NULL;
 			continue;
+		}
 
 		only_hdr = false;
 		/* Determine if this block needs to be backed up */
@@ -843,8 +847,115 @@ XLogRecordAssemble(RmgrId rmid, uint8 info, uint8 rminfo,
 		prev_regbuf = regbuf;
 
 		/* Ok, copy the header to the scratch buffer */
-		memcpy(scratch, &bkpb, SizeOfXLogRecordBlockHeader);
-		scratch += SizeOfXLogRecordBlockHeader;
+		/* First block needs to initialize optional batching */
+		if (block_id == 0)
+		{
+			first_blockid = (uint8 *) scratch;
+
+			if (bkpb.data_length <= UINT8_MAX)
+			{
+				/*
+				 * Optimistic guess: if the first registered block has no data,
+				 * it is likely that subsequent blocks have little data too.
+				 * If we're wrong, a subsequent block will have 1 more byte
+				 * than when we would have guessed correctly. If we're right,
+				 * it saves a byte for each correctly guessed block.
+				 */
+				bkpb.id |= XLR_BLOCK_DATA_SMALL;
+				memcpy(scratch, &bkpb, offsetof(XLogRecordBlockHeader, data_length));
+				scratch += offsetof(XLogRecordBlockHeader, data_length);
+
+				if (bkpb.data_length == 0)
+					Assert((bkpb.fork_flags & BKPBLOCK_HAS_DATA) == 0);
+				else
+				{
+					uint8	data_length = bkpb.data_length;
+					memcpy(scratch++, &data_length, sizeof(uint8));
+				}
+			}
+			else
+			{
+				memcpy(scratch, &bkpb, SizeOfXLogRecordBlockHeader);
+				scratch += SizeOfXLogRecordBlockHeader;
+			}
+		}
+		/* in a sequence of block ids; with sequence data at first_blockid */
+		else if (first_blockid != NULL)
+		{
+			if (bkpb.data_length == 0)
+			{
+				/*
+				 * Empty blocks don't care about data_length field size, as
+				 * it is not emitted.
+				 */
+				*first_blockid += 1;
+				*first_blockid |= XLR_BLOCK_FIRST_NP1_SEQ;
+
+				memcpy(scratch++, &bkpb.fork_flags, sizeof(bkpb.fork_flags));
+			}
+			else if (*first_blockid & XLR_BLOCK_DATA_SMALL)
+			{
+				if (bkpb.data_length <= UINT8_MAX)
+				{
+					uint8	fork_flags = bkpb.fork_flags;
+					uint8	data_length = bkpb.data_length;
+
+					*first_blockid += 1;
+					*first_blockid |= XLR_BLOCK_FIRST_NP1_SEQ;
+
+					memcpy(scratch++, &fork_flags, sizeof(uint8));
+					memcpy(scratch++, &data_length, sizeof(uint8));
+				}
+				else
+				{
+					first_blockid = NULL;
+					memcpy(scratch, &bkpb, SizeOfXLogRecordBlockHeader);
+					scratch += SizeOfXLogRecordBlockHeader;
+				}
+			}
+			else
+			{
+				/*
+				 * It would save 1 byte in the length field to break the chain,
+				 * but that would cost 1 byte as well, and completely disable
+				 * any chances of saving 1 byte in the block header of a
+				 * later block.
+				 * So as long as the chain continues, we won't have a large
+				 * record header.
+				 * We do still ignore the block_id field in the block header.
+				 */
+				*first_blockid += 1;
+				*first_blockid |= XLR_BLOCK_FIRST_NP1_SEQ;
+
+				memcpy(scratch,
+					   ((char *) &bkpb) + offsetof(XLogRecordBlockHeader, fork_flags),
+					   SizeOfXLogRecordBlockHeader - offsetof(XLogRecordBlockHeader, fork_flags));
+				scratch += SizeOfXLogRecordBlockHeader - offsetof(XLogRecordBlockHeader, fork_flags);
+			}
+		}
+		/* not in a sequence of block ids */
+		else
+		{
+			if (bkpb.data_length <= UINT8_MAX)
+			{
+				memcpy(scratch, &bkpb, offsetof(XLogRecordBlockHeader, data_length));
+				scratch += offsetof(XLogRecordBlockHeader, data_length);
+
+				if (bkpb.data_length == 0)
+					Assert((bkpb.fork_flags & BKPBLOCK_HAS_DATA) == 0);
+				else
+				{
+					uint8	data_length = bkpb.data_length;
+					memcpy(scratch++, &data_length, sizeof(uint8));
+				}
+			}
+			else
+			{
+				memcpy(scratch, &bkpb, SizeOfXLogRecordBlockHeader);
+				scratch += SizeOfXLogRecordBlockHeader;
+			}
+		}
+
 		if (include_image)
 		{
 			memcpy(scratch, &bimg, SizeOfXLogRecordBlockImageHeader);
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
index 56bf86cb27..cb930c9e82 100644
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -1792,6 +1792,9 @@ DecodeXLogRecord(XLogReaderState *state,
 	uint32		datatotal;
 	RelFileLocator *rlocator = NULL;
 	uint8		block_id;
+	bool		shorthdr = false;
+	bool		in_bid_seq = false;
+	uint8		bid_seq_end = 0;
 
 	decoded->header = (XLRHeaderData) {0};
 	decoded->lsn = lsn;
@@ -1851,7 +1854,10 @@ DecodeXLogRecord(XLogReaderState *state,
 	datatotal = 0;
 	while (remaining > datatotal)
 	{
-		COPY_HEADER_FIELD(&block_id, sizeof(uint8));
+		if (in_bid_seq)
+			block_id++;
+		else
+			COPY_HEADER_FIELD(&block_id, sizeof(uint8));
 
 		if (block_id == XLR_BLOCK_ID_DATA_SHORT)
 		{
@@ -1884,12 +1890,24 @@ DecodeXLogRecord(XLogReaderState *state,
 		{
 			COPY_HEADER_FIELD(&decoded->toplevel_xid, sizeof(TransactionId));
 		}
-		else if (block_id <= XLR_MAX_BLOCK_ID)
+		else if ((block_id & XLR_BLOCK_ID_MASK) <= XLR_MAX_BLOCK_ID)
 		{
 			/* XLogRecordBlockHeader */
 			DecodedBkpBlock *blk;
 			uint8		fork_flags;
 
+			if (!in_bid_seq)
+			{
+				shorthdr = (block_id & XLR_BLOCK_DATA_SMALL) != 0;
+
+				if (block_id & XLR_BLOCK_FIRST_NP1_SEQ)
+				{
+					bid_seq_end = (block_id & XLR_BLOCK_ID_MASK) + 1;
+					block_id = decoded->max_block_id + 1;
+					in_bid_seq = true;
+				}
+			}
+
 			/* mark any intervening block IDs as not in use */
 			for (int i = decoded->max_block_id + 1; i < block_id; ++i)
 				decoded->blocks[i].in_use = false;
@@ -1915,8 +1933,19 @@ DecodeXLogRecord(XLogReaderState *state,
 			blk->has_data = ((fork_flags & BKPBLOCK_HAS_DATA) != 0);
 
 			blk->prefetch_buffer = InvalidBuffer;
+			if (!blk->has_data)
+			{
+				/* no data == no data_len, in all situations */
+			}
+			else if (shorthdr)
+			{
+				uint8 len = 0;
+				COPY_HEADER_FIELD(&len, sizeof(uint8));
+				blk->data_len = len;
+			}
+			else
+				COPY_HEADER_FIELD(&blk->data_len, sizeof(uint16));
 
-			COPY_HEADER_FIELD(&blk->data_len, sizeof(uint16));
 			/* cross-check that the HAS_DATA flag is set iff data_length > 0 */
 			if (blk->has_data && blk->data_len == 0)
 			{
@@ -2033,6 +2062,10 @@ DecodeXLogRecord(XLogReaderState *state,
 				blk->rlocator = *rlocator;
 			}
 			COPY_HEADER_FIELD(&blk->blkno, sizeof(BlockNumber));
+
+			/* finish the sequence we're in */
+			if (block_id == bid_seq_end)
+				in_bid_seq = false;
 		}
 		else
 		{
diff --git a/src/include/access/xlogrecord.h b/src/include/access/xlogrecord.h
index 94125f9878..862c922032 100644
--- a/src/include/access/xlogrecord.h
+++ b/src/include/access/xlogrecord.h
@@ -220,13 +220,18 @@ XLogRecordGetRMInfo(XLogRecord *record)
  *
  * Note that we don't attempt to align the XLogRecordBlockHeader struct!
  * So, the struct must be copied to aligned local storage before use.
+ * 
+ * Note that if .id & 0x80 is set; the block header is small instead.
+ * If .id & 0x40 is set, there will be id + 1 following block headers
+ * of this type, having incremental ids, but written to disk without
+ * the id field.
  */
 typedef struct XLogRecordBlockHeader
 {
 	uint8		id;				/* block reference ID */
 	uint8		fork_flags;		/* fork within the relation, and flags */
 	uint16		data_length;	/* number of payload bytes (not including page
-								 * image) */
+								 * image). Emitted iff BKPBLOCK_HAS_DATA */
 
 	/* If BKPBLOCK_HAS_IMAGE, an XLogRecordBlockImageHeader struct follows */
 	/* If BKPBLOCK_SAME_REL is not set, a RelFileLocator follows */
@@ -359,11 +364,17 @@ typedef struct XLogRecordDataHeaderLong
  * need a handful of block references, but there are a few exceptions that
  * need more.
  */
-#define XLR_MAX_BLOCK_ID			32
-
-#define XLR_BLOCK_ID_DATA_SHORT		255
-#define XLR_BLOCK_ID_DATA_LONG		254
-#define XLR_BLOCK_ID_ORIGIN			253
-#define XLR_BLOCK_ID_TOPLEVEL_XID	252
+#define XLR_MAX_BLOCK_ID			0x20
+
+#define XLR_BLOCK_ID_DATA_SHORT		0x3F
+#define XLR_BLOCK_ID_DATA_LONG		0x3E
+#define XLR_BLOCK_ID_ORIGIN			0x3D
+#define XLR_BLOCK_ID_TOPLEVEL_XID	0x3C
+
+#define XLR_BLOCK_ID_MASK			0x3F
+#define XLR_BLOCK_FIRST_NP1_SEQ		0x40 /* the following blocks are 
+										  * 0..block_id + 1, and have omitted
+										  * their block ID */
+#define XLR_BLOCK_DATA_SMALL		0x80 /* data_length field is uint8 */
 
 #endif							/* XLOGRECORD_H */
-- 
2.30.2

