From e78a94b35938961e39db10ad3ef75003a624aef1 Mon Sep 17 00:00:00 2001
From: Andrey Borodin <amborodin@acm.org>
Date: Sun, 26 Jul 2026 18:53:56 +0500
Subject: [PATCH v7 3/3] WIP: compress WAL records against earlier records in a
 stream

Adds wal_compression_streams: a pool of leased compression streams, so a
record can be compressed against the records that preceded it in the same
stream.  This is what reaches the small records, which per-record
compression cannot touch.

Recovery, whole-segment reads and the data itself are correct, but a
reader starting at an arbitrary LSN inside a segment cannot decode, so
this is not proposable yet.
---
 src/backend/access/transam/xlog.c             | 112 ++++++++++
 src/backend/access/transam/xloginsert.c       | 210 ++++++++++++++++++
 src/backend/access/transam/xlogreader.c       |  72 ++++++
 .../utils/activity/wait_event_names.txt       |   1 +
 src/backend/utils/misc/guc_parameters.dat     |  12 +
 src/backend/utils/misc/postgresql.conf.sample |   4 +
 src/include/access/xlog.h                     |   4 +
 src/include/access/xlogreader.h               |   7 +
 src/include/access/xlogrecord.h               |  14 +-
 src/include/storage/lwlocklist.h              |   1 +
 10 files changed, 436 insertions(+), 1 deletion(-)

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index eb1d1fddb77..ce94684749a 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -143,6 +143,7 @@ int			max_slot_wal_keep_size_mb = -1;
 int			wal_decode_buffer_size = 512 * 1024;
 bool		track_wal_io_timing = false;
 int			wal_compression_threshold = 512;
+int			wal_compression_streams = 0;
 
 #ifdef WAL_DEBUG
 bool		XLOG_DEBUG = false;
@@ -575,6 +576,34 @@ typedef enum
 
 static XLogCtlData *XLogCtl = NULL;
 
+/*
+ * State of one WAL compression stream.
+ *
+ * The compressor itself cannot live here: it is a libzstd object full of
+ * pointers, and shared memory is not mapped at the same address in every
+ * process under EXEC_BACKEND.  So a backend keeps its own compressor and
+ * leases the stream: while it owns the slot it may compress against what it
+ * put there before, and a backend taking the slot over has to start the
+ * stream afresh, which it announces with XLR_STREAM_RESET.
+ */
+typedef struct WALCompressionSlot
+{
+	LWLock		lock;			/* held across compress + insert */
+	int			owner;			/* ProcNumber holding the lease, or -1 */
+	uint64		generation;		/* bumped whenever the stream restarts */
+	XLogRecPtr	redo;			/* RedoRecPtr the stream started under */
+	XLogSegNo	seg;			/* segment the stream started in */
+	bool		restart;		/* next record has to restart the stream */
+} WALCompressionSlot;
+
+typedef union WALCompressionSlotPadded
+{
+	WALCompressionSlot s;
+	char		pad[PG_CACHE_LINE_SIZE];
+} WALCompressionSlotPadded;
+
+static WALCompressionSlotPadded *WALCompressionSlots = NULL;
+
 /* a private copy of XLogCtl->Insert.WALInsertLocks, for convenience */
 static WALInsertLockPadded *WALInsertLocks = NULL;
 
@@ -768,6 +797,64 @@ XLogGetRecordTotalLen(XLogRecord *record)
 }
 #endif
 
+/*
+ * Take the lease on a compression stream, and say whether the stream has to
+ * start over.  The lock is held until XLogCompressionStreamRelease(), so that
+ * records enter the stream in the same order they are given their LSNs.
+ *
+ * A stream restarts when another backend used the slot in the meantime (its
+ * compressor state is in that other process and is gone for us), when a
+ * checkpoint moved the redo point out from under it, or when the previous
+ * record ran into the next WAL segment, which keeps a segment decodable on
+ * its own.
+ */
+bool
+XLogCompressionStreamAcquire(int slot, XLogRecPtr redo)
+{
+	WALCompressionSlot *s = &WALCompressionSlots[slot].s;
+	bool		restart;
+
+	LWLockAcquire(&s->lock, LW_EXCLUSIVE);
+
+	restart = (s->restart || s->owner != MyProcNumber || s->redo != redo);
+	if (restart)
+	{
+		s->owner = MyProcNumber;
+		s->generation++;
+		s->redo = redo;
+		s->restart = false;
+	}
+	return restart;
+}
+
+/*
+ * Release the lease.  "failed" means the record never made it into WAL, so
+ * whatever we fed the compressor has to be thrown away by everyone.
+ */
+void
+XLogCompressionStreamRelease(int slot, XLogRecPtr end_pos, bool restarted,
+							 bool failed)
+{
+	WALCompressionSlot *s = &WALCompressionSlots[slot].s;
+
+	if (failed)
+	{
+		s->owner = -1;
+		s->restart = true;
+	}
+	else if (XLogRecPtrIsValid(end_pos))
+	{
+		XLogSegNo	seg;
+
+		XLByteToSeg(end_pos - 1, seg, wal_segment_size);
+		if (restarted)
+			s->seg = seg;
+		else if (seg != s->seg)
+			s->restart = true;	/* we have left the segment we started in */
+	}
+	LWLockRelease(&s->lock);
+}
+
 /*
  * Insert an XLOG record represented by an already-constructed chain of data
  * chunks.  This is a low-level routine; to construct the WAL record header
@@ -5340,6 +5427,11 @@ XLOGShmemRequest(void *arg)
 
 	/* WAL insertion locks, plus alignment */
 	size = add_size(size, mul_size(sizeof(WALInsertLockPadded), NUM_XLOGINSERT_LOCKS + 1));
+
+	/* WAL compression stream slots, plus alignment slack */
+	if (wal_compression_streams > 0)
+		size = add_size(size, mul_size(sizeof(WALCompressionSlotPadded),
+									   wal_compression_streams + 1));
 	/* xlblocks array */
 	size = add_size(size, mul_size(sizeof(pg_atomic_uint64), XLOGbuffers));
 	/* extra alignment padding for XLOG I/O buffers */
@@ -5423,6 +5515,26 @@ XLOGShmemInit(void *arg)
 		WALInsertLocks[i].l.lastImportantAt = InvalidXLogRecPtr;
 	}
 
+	/* WAL compression streams, likewise aligned to their padded size */
+	if (wal_compression_streams > 0)
+	{
+		allocptr += sizeof(WALCompressionSlotPadded) -
+			((uintptr_t) allocptr) % sizeof(WALCompressionSlotPadded);
+		WALCompressionSlots = (WALCompressionSlotPadded *) allocptr;
+		allocptr += sizeof(WALCompressionSlotPadded) * wal_compression_streams;
+
+		for (i = 0; i < wal_compression_streams; i++)
+		{
+			LWLockInitialize(&WALCompressionSlots[i].s.lock,
+							 LWTRANCHE_WAL_COMPRESSION_STREAM);
+			WALCompressionSlots[i].s.owner = -1;
+			WALCompressionSlots[i].s.generation = 0;
+			WALCompressionSlots[i].s.redo = InvalidXLogRecPtr;
+			WALCompressionSlots[i].s.seg = 0;
+			WALCompressionSlots[i].s.restart = true;
+		}
+	}
+
 	/*
 	 * Align the start of the page buffers to a full xlog block size boundary.
 	 * This simplifies some calculations in XLOG insertion. It is also
diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index a5a3a1422cb..bc418e11466 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -171,6 +171,15 @@ static int	compression_buf_offset; /* fill level for FPI packing */
 static char *compressed_data = NULL;
 #endif
 
+#ifdef USE_ZSTD
+/*
+ * One compressor per stream slot we currently hold a lease on.  These stay in
+ * the backend because a zstd context cannot live in shared memory; a backend
+ * taking a slot over therefore has to restart the stream.
+ */
+static ZSTD_CCtx **stream_cctx = NULL;
+#endif
+
 /*
  * An array of XLogRecData structs, to hold registered data.
  */
@@ -515,6 +524,36 @@ XLogSetRecordFlags(uint8 flags)
 	curinsert_flags |= flags;
 }
 
+/*
+ * May this record be compressed against earlier ones?
+ *
+ * The rule is not about the resource manager but about who reads the record:
+ * anything that someone looks up by LSN, without replaying what comes before
+ * it, has to stay readable on its own.  Recovery finds the checkpoint record
+ * from pg_control, and twophase.c reads a PREPARE record from a stored LSN.
+ * XLOG_SWITCH is looked for by anything scanning WAL, and is tiny anyway.
+ */
+static bool
+XLogRecordJoinsStream(RmgrId rmid, uint8 info)
+{
+	if (rmid == RM_XLOG_ID)
+	{
+		uint8		xlinfo = info & ~XLR_INFO_MASK;
+
+		if (xlinfo == XLOG_CHECKPOINT_SHUTDOWN ||
+			xlinfo == XLOG_CHECKPOINT_ONLINE ||
+			xlinfo == XLOG_END_OF_RECOVERY ||
+			xlinfo == XLOG_SWITCH)
+			return false;
+	}
+	else if (rmid == RM_XACT_ID)
+	{
+		if ((info & XLOG_XACT_OPMASK) == XLOG_XACT_PREPARE)
+			return false;
+	}
+	return true;
+}
+
 /* Compress the assembled record; NULL if that did not pay off */
 static XLogRecData *
 XLogCompressRdt(XLogRecData *rdt)
@@ -554,6 +593,7 @@ XLogCompressRdt(XLogRecData *rdt)
 
 	compressed_header->record_header = *src_header;
 	compressed_header->decompressed_length = flat_len;
+	compressed_header->stream = XLR_NO_STREAM;
 
 	orig_len = src_header->xl_tot_len - SizeOfXLogRecord;
 
@@ -628,6 +668,111 @@ XLogCompressRdt(XLogRecData *rdt)
 #endif
 }
 
+#ifdef USE_ZSTD
+/*
+ * Compress a record into stream "slot", against everything this stream has
+ * compressed since it last restarted.
+ *
+ * The output has to be complete when we return: the record is about to be
+ * given an LSN and copied into WAL, so nothing of it may stay inside the
+ * compressor.  That is what ZSTD_e_flush buys, and it is also why the
+ * compressed length is known before the space is reserved.
+ *
+ * Returns NULL if the record did not compress into the space we have.  The
+ * stream is unusable after that, because part of the record was consumed, so
+ * the caller must restart it.
+ */
+static XLogRecData *
+XLogCompressRdtStream(XLogRecData *rdt, int slot, bool restart, bool *poisoned)
+{
+	static XLogRecData compressed_rdt_hdr;
+	XLogCompressionHeader *compressed_header;
+	XLogRecord *src_header;
+	uint32		flat_len = 0;
+	uint32		orig_len;
+	ZSTD_CCtx  *cctx;
+	ZSTD_inBuffer in;
+	ZSTD_outBuffer out;
+	size_t		rem;
+
+	*poisoned = false;
+
+	if (stream_cctx[slot] == NULL)
+	{
+		stream_cctx[slot] = ZSTD_createCCtx();
+		if (stream_cctx[slot] == NULL)
+			return NULL;
+		ZSTD_CCtx_setParameter(stream_cctx[slot], ZSTD_c_compressionLevel,
+							   ZSTD_CLEVEL_DEFAULT);
+		restart = true;
+	}
+	cctx = stream_cctx[slot];
+
+	if (restart)
+		ZSTD_CCtx_reset(cctx, ZSTD_reset_session_only);
+
+	for (const XLogRecData *r = rdt; r != NULL; r = r->next)
+	{
+		memcpy(compression_buf + flat_len, r->data, r->len);
+		flat_len += r->len;
+	}
+	Assert(flat_len <= WAL_COMPRESSION_BUFSIZE);
+
+	src_header = (XLogRecord *) compression_buf;
+	compressed_header = (XLogCompressionHeader *) compressed_data;
+
+	/* Zero it first: the padding in the header reaches disk */
+	memset(compressed_header, 0, SizeOfXLogCompressedRecord);
+	compressed_header->record_header = *src_header;
+	compressed_header->decompressed_length = flat_len;
+	compressed_header->method = XLR_COMPRESS_ZSTD;
+	compressed_header->stream = (uint8) slot;
+	compressed_header->stream_flags = restart ? XLR_STREAM_RESET : 0;
+
+	orig_len = src_header->xl_tot_len - SizeOfXLogRecord;
+
+	in.src = (char *) &src_header[1];
+	in.size = orig_len;
+	in.pos = 0;
+	out.dst = (char *) &compressed_header[1];
+	out.size = WAL_COMPRESSION_BUFSIZE - SizeOfXLogCompressedRecord;
+	out.pos = 0;
+
+	do
+	{
+		rem = ZSTD_compressStream2(cctx, &out, &in, ZSTD_e_flush);
+		if (ZSTD_isError(rem))
+		{
+			*poisoned = true;
+			return NULL;
+		}
+		if (out.pos == out.size && (rem != 0 || in.pos < in.size))
+		{
+			/* no room left, and the compressor has eaten part of the record */
+			*poisoned = true;
+			return NULL;
+		}
+	} while (rem != 0 || in.pos < in.size);
+
+	if (out.pos + SizeOfXLogCompressedRecord >= flat_len)
+	{
+		/* bigger than the plain record; the stream ate it either way */
+		*poisoned = true;
+		return NULL;
+	}
+
+	compressed_header->record_header.xl_tot_len =
+		SizeOfXLogCompressedRecord + out.pos;
+	compressed_header->record_header.xl_info |= XLR_COMPRESSED;
+
+	compressed_rdt_hdr.data = compressed_data;
+	compressed_rdt_hdr.len = compressed_header->record_header.xl_tot_len;
+	compressed_rdt_hdr.next = NULL;
+
+	return &compressed_rdt_hdr;
+}
+#endif							/* USE_ZSTD */
+
 /* Checksum assembled record (which may be compressed). */
 static void
 XLogChecksumRecord(XLogRecData *rdt)
@@ -678,6 +823,19 @@ XLogInsert(RmgrId rmid, uint8 info)
 									wal_compression_threshold <
 									WAL_COMPRESSION_BUFSIZE);
 
+	/*
+	 * Streams are zstd only for now, and only for records nobody reads out of
+	 * order.  One slot per backend, so that a backend usually finds its own
+	 * compressor still in place.
+	 */
+	int			stream_slot = -1;
+
+#ifdef USE_ZSTD
+	if (wal_compression_streams > 0 && wal_compression == WAL_COMPRESSION_ZSTD &&
+		stream_cctx != NULL && XLogRecordJoinsStream(rmid, info))
+		stream_slot = MyProcNumber % wal_compression_streams;
+#endif
+
 	/* XLogBeginInsert() must have been called. */
 	if (!begininsert_called)
 		elog(ERROR, "XLogBeginInsert was not called");
@@ -735,6 +893,52 @@ XLogInsert(RmgrId rmid, uint8 info)
 		 * record is not compressed as a whole after all, reassemble it to get
 		 * the per-FPI compression back.
 		 */
+#ifdef USE_ZSTD
+		if (stream_slot >= 0 && rec_size <= WAL_COMPRESSION_BUFSIZE)
+		{
+			bool		restart;
+			bool		poisoned;
+			XLogRecData *rdt_compressed;
+
+			/*
+			 * Hold the stream while we compress and while the record is given
+			 * its LSN, so the order records enter the stream is the order a
+			 * reader will meet them in.
+			 */
+			restart = XLogCompressionStreamAcquire(stream_slot, RedoRecPtr);
+			rdt_compressed = XLogCompressRdtStream(rdt, stream_slot, restart,
+												   &poisoned);
+			if (rdt_compressed != NULL)
+			{
+				XLogChecksumRecord(rdt_compressed);
+				EndPos = XLogInsertRecord(rdt_compressed, fpw_lsn,
+										  curinsert_flags, num_fpi,
+										  fpi_bytes, topxid_included);
+				XLogCompressionStreamRelease(stream_slot, EndPos, restart,
+											 !XLogRecPtrIsValid(EndPos));
+				if (XLogRecPtrIsValid(EndPos))
+					break;
+				continue;		/* retry, with the stream restarted */
+			}
+
+			/* Could not compress; the stream may have eaten part of it */
+			XLogCompressionStreamRelease(stream_slot, InvalidXLogRecPtr,
+										 restart, poisoned);
+			if (num_fpi > 0)
+			{
+				compression_buf_offset = 0;
+				rdt = XLogRecordAssemble(rmid, info, RedoRecPtr, doPageWrites,
+										 &fpw_lsn, &num_fpi, &fpi_bytes,
+										 &topxid_included, &rec_size,
+										 false);
+			}
+			XLogChecksumRecord(rdt);
+			EndPos = XLogInsertRecord(rdt, fpw_lsn, curinsert_flags, num_fpi,
+									  fpi_bytes, topxid_included);
+			continue;
+		}
+#endif
+
 		if (try_whole_record)
 		{
 			bool		whole_record_compressed = false;
@@ -858,6 +1062,12 @@ AllocCompressionBuffers(void)
 	compressed_data = MemoryContextAlloc(xloginsert_cxt,
 										 WAL_COMPRESSION_BUFSIZE);
 #endif
+#ifdef USE_ZSTD
+	if (wal_compression_streams > 0)
+		stream_cctx = MemoryContextAllocZero(xloginsert_cxt,
+											 sizeof(ZSTD_CCtx *) *
+											 wal_compression_streams);
+#endif
 }
 
 /*
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
index 23fe90c7f11..54936bf02bb 100644
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -177,6 +177,15 @@ XLogReaderFree(XLogReaderState *state)
 		pfree(state->readRecordBuf);
 	if (state->decompression_buffer)
 		pfree(state->decompression_buffer);
+#ifdef USE_ZSTD
+	if (state->stream_dctx)
+	{
+		for (int i = 0; i < XLR_MAX_STREAMS; i++)
+			if (state->stream_dctx[i])
+				ZSTD_freeDCtx((ZSTD_DCtx *) state->stream_dctx[i]);
+		pfree(state->stream_dctx);
+	}
+#endif
 	pfree(state->readBuf);
 	pfree(state);
 }
@@ -1838,7 +1847,70 @@ XLogDecompressRecordIfNeeded(XLogReaderState *state,
 		dst_h->xl_tot_len = src->decompressed_length;
 		dst = (char *) &dst_h[1];
 
+#ifdef USE_ZSTD
+		if (src->stream != XLR_NO_STREAM)
+		{
+			ZSTD_DCtx  *dctx;
+			ZSTD_inBuffer in;
+			ZSTD_outBuffer out;
+
+			if (src->method != XLR_COMPRESS_ZSTD)
+			{
+				report_invalid_record(state,
+									  "streamed record at %X/%08X uses an unexpected compression method",
+									  LSN_FORMAT_ARGS(recptr));
+				return NULL;
+			}
+
+			if (state->stream_dctx == NULL)
+				state->stream_dctx = palloc0(sizeof(void *) * XLR_MAX_STREAMS);
+
+			dctx = (ZSTD_DCtx *) state->stream_dctx[src->stream];
+			if (dctx == NULL)
+			{
+				dctx = ZSTD_createDCtx();
+				if (dctx == NULL)
+				{
+					report_invalid_record(state,
+										  "out of memory while decompressing record at %X/%08X",
+										  LSN_FORMAT_ARGS(recptr));
+					return NULL;
+				}
+				state->stream_dctx[src->stream] = dctx;
+			}
+
+			if (src->stream_flags & XLR_STREAM_RESET)
+				ZSTD_DCtx_reset(dctx, ZSTD_reset_session_only);
+
+			in.src = (char *) &src[1];
+			in.size = srclen;
+			in.pos = 0;
+			out.dst = dst;
+			out.size = body_len;
+			out.pos = 0;
+
+			while (in.pos < in.size)
+			{
+				size_t		ret = ZSTD_decompressStream(dctx, &out, &in);
+
+				if (ZSTD_isError(ret))
+				{
+					decomp_success = false;
+					break;
+				}
+				if (out.pos == out.size && in.pos < in.size)
+				{
+					decomp_success = false;
+					break;
+				}
+			}
+			if (decomp_success && out.pos != body_len)
+				decomp_success = false;
+		}
+		else if (src->method == XLR_COMPRESS_LZ4)
+#else
 		if (src->method == XLR_COMPRESS_LZ4)
+#endif
 		{
 #ifdef USE_LZ4
 			if (LZ4_decompress_safe((char *) &src[1], dst,
diff --git a/src/backend/utils/activity/wait_event_names.txt b/src/backend/utils/activity/wait_event_names.txt
index 1016502d042..088c9bac3ba 100644
--- a/src/backend/utils/activity/wait_event_names.txt
+++ b/src/backend/utils/activity/wait_event_names.txt
@@ -417,6 +417,7 @@ XactSLRU	"Waiting to access the transaction status SLRU cache."
 ParallelVacuumDSA	"Waiting for parallel vacuum dynamic shared memory allocation."
 AioUringCompletion	"Waiting for another process to complete IO via io_uring."
 ShmemIndex	"Waiting to find or allocate space in shared memory."
+WALCompressionStream	"Waiting to compress a WAL record into a shared compression stream."
 
 # No "ABI_compatibility" region here as WaitEventLWLock has its own C code.
 
diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat
index bddd3a26a51..de3293f3c3b 100644
--- a/src/backend/utils/misc/guc_parameters.dat
+++ b/src/backend/utils/misc/guc_parameters.dat
@@ -3465,6 +3465,18 @@
   assign_hook => 'assign_wal_compression',
 },
 
+{ name => 'wal_compression_streams', type => 'int', context => 'PGC_POSTMASTER', group => 'WAL_SETTINGS',
+  short_desc => 'Number of WAL compression streams shared by all backends.',
+  long_desc => 'Zero compresses every record on its own. A positive value lets a record '
+             . 'be compressed against the records that preceded it in the same stream, '
+             . 'which compresses small records much better, at the cost of one '
+             . 'decompression context per stream in every process that reads WAL records.',
+  variable => 'wal_compression_streams',
+  boot_val => '0',
+  min => '0',
+  max => '64',
+},
+
 { name => 'wal_compression_threshold', type => 'int', context => 'PGC_SUSET', group => 'WAL_SETTINGS',
   short_desc => 'Minimum WAL record length to engage whole-record compression.',
   long_desc => 'Records at least this large are compressed as a single unit instead of '
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index 406b8fc52ca..701d7e2ca7b 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -266,6 +266,10 @@
                                         # off, pglz (or "on"), lz4, or zstd
 #wal_compression_threshold = 512        # min 32, smallest record compressed as a whole;
                                         # whole-record compression needs lz4 or zstd
+#wal_compression_streams = 0            # 0 disables compressing a record against
+                                        # earlier ones; a positive value is the
+                                        # number of shared compression streams
+                                        # (change requires restart)
 #wal_init_zero = on                     # zero-fill new WAL files
 #wal_recycle = on                       # recycle WAL files
 #wal_buffers = -1                       # min 32kB, -1 sets based on shared_buffers
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 7fda7eb1f23..fb8e6e8afaa 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -59,6 +59,7 @@ extern PGDLLIMPORT bool track_wal_io_timing;
 extern PGDLLIMPORT int wal_decode_buffer_size;
 extern PGDLLIMPORT int data_checksums;
 extern PGDLLIMPORT int wal_compression_threshold;
+extern PGDLLIMPORT int wal_compression_streams;
 
 extern PGDLLIMPORT int CheckPointSegments;
 
@@ -211,6 +212,9 @@ typedef enum WALAvailability
 struct XLogRecData;
 struct XLogReaderState;
 
+extern bool XLogCompressionStreamAcquire(int slot, XLogRecPtr redo);
+extern void XLogCompressionStreamRelease(int slot, XLogRecPtr end_pos,
+										 bool restarted, bool failed);
 extern XLogRecPtr XLogInsertRecord(struct XLogRecData *rdata,
 								   XLogRecPtr fpw_lsn,
 								   uint8 flags,
diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h
index 988734f8411..3c226fddd04 100644
--- a/src/include/access/xlogreader.h
+++ b/src/include/access/xlogreader.h
@@ -255,6 +255,13 @@ struct XLogReaderState
 	char	   *decompression_buffer;
 	uint32		decompression_buffer_size;
 
+	/*
+	 * One decompression context per compression stream met so far, indexed by
+	 * the stream id in the record.  Void because the type belongs to whichever
+	 * compression library the build has.
+	 */
+	void	  **stream_dctx;
+
 	/*
 	 * Queue of records that have been decoded.  This is a linked list that
 	 * usually consists of consecutive records in decode_buffer, but may also
diff --git a/src/include/access/xlogrecord.h b/src/include/access/xlogrecord.h
index be662e6558f..0b80df7d626 100644
--- a/src/include/access/xlogrecord.h
+++ b/src/include/access/xlogrecord.h
@@ -205,9 +205,21 @@ typedef struct XLogCompressionHeader
 	XLogRecord	record_header;
 	uint32		decompressed_length;
 	uint8		method;			/* XLR_COMPRESS_* */
-	/* 3 bytes of padding here, initialize to zero */
+	uint8		stream;			/* stream slot, or XLR_NO_STREAM */
+	uint8		stream_flags;	/* XLR_STREAM_* */
+	/* 1 byte of padding here, initialize to zero */
 } XLogCompressionHeader;
 
+/*
+ * A record compressed on its own carries XLR_NO_STREAM.  Otherwise "stream"
+ * names the compression stream it belongs to, and the record can only be
+ * decompressed after every earlier record of that stream.  XLR_STREAM_RESET
+ * says the stream starts here, so the reader must discard what it had.
+ */
+#define XLR_MAX_STREAMS		255
+#define XLR_NO_STREAM		0xFF
+#define XLR_STREAM_RESET	0x01
+
 #define SizeOfXLogCompressedRecord	sizeof(XLogCompressionHeader)
 
 /*
diff --git a/src/include/storage/lwlocklist.h b/src/include/storage/lwlocklist.h
index d7eb648bd27..af1a8c2f932 100644
--- a/src/include/storage/lwlocklist.h
+++ b/src/include/storage/lwlocklist.h
@@ -140,3 +140,4 @@ PG_LWLOCKTRANCHE(XACT_SLRU, XactSLRU)
 PG_LWLOCKTRANCHE(PARALLEL_VACUUM_DSA, ParallelVacuumDSA)
 PG_LWLOCKTRANCHE(AIO_URING_COMPLETION, AioUringCompletion)
 PG_LWLOCKTRANCHE(SHMEM_INDEX, ShmemIndex)
+PG_LWLOCKTRANCHE(WAL_COMPRESSION_STREAM, WALCompressionStream)
-- 
2.50.1 (Apple Git-155)

