From 1856785f8801eb899fc9bdd7f8c7a6670de02c5b Mon Sep 17 00:00:00 2001
From: Andrey Borodin <amborodin@acm.org>
Date: Sun, 26 Jul 2026 18:53:56 +0500
Subject: [PATCH v9 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.  That is what reaches the small records, which per-record
compression cannot touch; pgbench emits 29% less WAL per transaction.

Streams start over at fixed 4MB boundaries, and a record that would
continue one past a boundary is refused its reserved position and built
again, so every stream begins afresh at or after a boundary.
XLogBeginReadStreamed() uses that to read from an arbitrary LSN: it
rewinds to the boundary below and reads forward to rebuild the
decompressors.

Still WIP: only pg_waldump has been converted to it, and a replication
slot has to keep one boundary more WAL than it needs for itself.
---
 contrib/pg_walinspect/pg_walinspect.c         |   8 +-
 src/backend/access/transam/xlog.c             | 272 +++++++++++++++-
 src/backend/access/transam/xloginsert.c       | 275 ++++++++++++++++
 src/backend/access/transam/xlogreader.c       | 293 ++++++++++++++++--
 src/backend/access/transam/xlogrecovery.c     |   8 +
 src/backend/postmaster/walsummarizer.c        |   5 +-
 src/backend/replication/logical/logical.c     |   8 +-
 .../replication/logical/logicalfuncs.c        |   3 +-
 src/backend/replication/walsender.c           |  11 +-
 .../utils/activity/wait_event_names.txt       |   1 +
 src/backend/utils/misc/guc_parameters.dat     |  12 +
 src/backend/utils/misc/postgresql.conf.sample |   4 +
 src/bin/pg_rewind/parsexlog.c                 |  24 +-
 src/bin/pg_waldump/pg_waldump.c               |  10 +-
 src/include/access/xlog.h                     |   5 +
 src/include/access/xlogreader.h               |  30 ++
 src/include/access/xlogrecord.h               |  35 ++-
 src/include/storage/lwlocklist.h              |   1 +
 src/test/perl/PostgreSQL/Test/Cluster.pm      |  10 +-
 src/test/recovery/Makefile                    |   1 +
 src/test/recovery/meson.build                 |   1 +
 .../recovery/t/043_no_contrecord_switch.pl    |   5 +-
 ..._compression.pl => 056_wal_compression.pl} | 121 +++++++-
 23 files changed, 1084 insertions(+), 59 deletions(-)
 rename src/test/recovery/t/{052_wal_compression.pl => 056_wal_compression.pl} (53%)

diff --git a/contrib/pg_walinspect/pg_walinspect.c b/contrib/pg_walinspect/pg_walinspect.c
index a172f9e2b40..e31eac9e620 100644
--- a/contrib/pg_walinspect/pg_walinspect.c
+++ b/contrib/pg_walinspect/pg_walinspect.c
@@ -126,8 +126,12 @@ InitXLogReaderState(XLogRecPtr lsn)
 				 errmsg("out of memory"),
 				 errdetail("Failed while allocating a WAL reading processor.")));
 
-	/* first find a valid recptr to start from */
-	first_valid_record = XLogFindNextRecord(xlogreader, lsn, &errormsg);
+	/*
+	 * Find a valid recptr to start from.  A record can be compressed against
+	 * earlier records of its stream, so this backs up to where the streams
+	 * start over and reads forward from there to rebuild the decompressors.
+	 */
+	first_valid_record = XLogBeginReadStreamed(xlogreader, lsn, &errormsg);
 
 	if (!XLogRecPtrIsValid(first_valid_record))
 	{
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 0824d182160..3a15622bdab 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;
@@ -398,6 +399,22 @@ typedef union WALInsertLockPadded
  */
 static SessionBackupState sessionBackupState = SESSION_BACKUP_NONE;
 
+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 */
+	XLogRecPtr	last_write;		/* end of the last record in this stream */
+	bool		restart;		/* next record has to restart the stream */
+} WALCompressionSlot;
+
+typedef union WALCompressionSlotPadded
+{
+	WALCompressionSlot s;
+	char		pad[PG_CACHE_LINE_SIZE];
+} WALCompressionSlotPadded;
+
 /*
  * Shared state data for WAL insertion.
  */
@@ -450,6 +467,12 @@ typedef struct XLogCtlInsert
 	 * WAL insertion locks.
 	 */
 	WALInsertLockPadded *WALInsertLocks;
+
+	/*
+	 * The compression stream slots, kept here so that a process which only
+	 * attaches to shared memory, rather than inheriting it, can find them.
+	 */
+	WALCompressionSlotPadded *WALCompressionSlots;
 } XLogCtlInsert;
 
 /*
@@ -575,6 +598,28 @@ 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.
+ */
+
+static WALCompressionSlotPadded *WALCompressionSlots = NULL;
+
+/*
+ * The stream this backend is holding between XLogCompressionStreamAcquire()
+ * and XLogCompressionStreamRelease(), and whether the record it compressed
+ * starts that stream over.  XLogInsertRecord() needs both to tell whether the
+ * position it is about to reserve is one the record may be placed at.
+ */
+static int	insert_stream_slot = -1;
+static bool insert_stream_restarted = false;
+
 /* a private copy of XLogCtl->Insert.WALInsertLocks, for convenience */
 static WALInsertLockPadded *WALInsertLocks = NULL;
 
@@ -735,8 +780,9 @@ static void CopyXLogRecordToWAL(int write_len, bool isLogSwitch,
 								XLogRecData *rdata,
 								XLogRecPtr StartPos, XLogRecPtr EndPos,
 								TimeLineID tli);
-static void ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos,
-									  XLogRecPtr *EndPos, XLogRecPtr *PrevPtr);
+static bool ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos,
+									  XLogRecPtr *EndPos, XLogRecPtr *PrevPtr,
+									  uint64 startbefore);
 static bool ReserveXLogSwitch(XLogRecPtr *StartPos, XLogRecPtr *EndPos,
 							  XLogRecPtr *PrevPtr);
 static XLogRecPtr WaitXLogInsertionsToFinish(XLogRecPtr upto);
@@ -768,6 +814,114 @@ XLogGetRecordTotalLen(XLogRecord *record)
 }
 #endif
 
+/*
+ * Take a compression stream, or report that none is worth taking.
+ *
+ * "preferred" is the slot this backend used last, or -1.  Returns the slot to
+ * use and sets *restart if its stream has to start over, or returns -1, which
+ * means the caller should compress this record on its own.  The lock is held
+ * until XLogCompressionStreamRelease(), so that records enter a stream in the
+ * same order they are given their LSNs.
+ *
+ * Keeping a stream is what makes it pay, so a backend that keeps writing
+ * keeps its slot, and we never take one away from a backend that is still
+ * feeding it.  A slot is up for grabs only if nobody owns it, or if its last
+ * record fell below the latest reset boundary: such a stream has to start over
+ * anyway, so taking it costs its owner nothing.  This also returns the slots
+ * of backends that went idle or died, without needing to be told that they
+ * did.
+ */
+int
+XLogCompressionStreamAcquire(int preferred, XLogRecPtr redo, bool *restart)
+{
+	XLogRecPtr	insert_at;
+	int			i;
+
+	/* Fast path: the slot we had, if it is still ours. */
+	if (preferred >= 0)
+	{
+		WALCompressionSlot *s = &WALCompressionSlots[preferred].s;
+
+		LWLockAcquire(&s->lock, LW_EXCLUSIVE);
+		if (s->owner == MyProcNumber)
+		{
+			*restart = (s->restart || s->redo != redo);
+			if (*restart)
+			{
+				s->generation++;
+				s->redo = redo;
+				s->restart = false;
+			}
+			insert_stream_slot = preferred;
+			insert_stream_restarted = *restart;
+			return preferred;
+		}
+		LWLockRelease(&s->lock);
+	}
+
+	/*
+	 * Slow path.  Reading the insert position takes a spinlock, which is why
+	 * it is only done here, when we have no slot of our own.
+	 */
+	insert_at = GetXLogInsertRecPtr();
+
+	for (i = 0; i < wal_compression_streams; i++)
+	{
+		WALCompressionSlot *s = &WALCompressionSlots[i].s;
+
+		if (!LWLockConditionalAcquire(&s->lock, LW_EXCLUSIVE))
+			continue;			/* busy; its owner is clearly active */
+
+		if (s->owner == -1 ||
+			XLogCompressionStreamCrosses(s->last_write, insert_at))
+		{
+			s->owner = MyProcNumber;
+			s->generation++;
+			s->redo = redo;
+			s->restart = false;
+			s->last_write = insert_at;
+			*restart = true;
+			insert_stream_slot = i;
+			insert_stream_restarted = true;
+			return i;
+		}
+		LWLockRelease(&s->lock);
+	}
+
+	return -1;
+}
+
+/*
+ * 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;
+
+	insert_stream_slot = -1;
+
+	if (failed)
+	{
+		s->owner = -1;
+		s->restart = true;
+	}
+	else if (XLogRecPtrIsValid(end_pos))
+	{
+		/*
+		 * A record that crossed a boundary is of no use to a reader starting
+		 * at that boundary, which skips it as continuation data.  So the next
+		 * record has to start over even when this one already did.
+		 */
+		if (XLogCompressionStreamCrosses(s->last_write, end_pos - 1))
+			s->restart = true;
+		s->last_write = end_pos - 1;	/* where this stream last wrote */
+	}
+	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
@@ -811,8 +965,9 @@ XLogInsertRecord(XLogRecData *rdata,
 	XLogRecord *rechdr = (XLogRecord *) rdata->data;
 	uint8		info = rechdr->xl_info & ~XLR_INFO_MASK;
 	WalInsertClass class = WALINSERT_NORMAL;
-	XLogRecPtr	StartPos;
-	XLogRecPtr	EndPos;
+	XLogRecPtr	StartPos = InvalidXLogRecPtr;
+	XLogRecPtr	EndPos = InvalidXLogRecPtr;
+	uint64		startbefore;
 	bool		prevDoPageWrites = doPageWrites;
 	TimeLineID	insertTLI;
 
@@ -870,6 +1025,24 @@ XLogInsertRecord(XLogRecData *rdata,
 	 *
 	 *----------
 	 */
+	/*
+	 * A record that continues a stream has to stay below the next reset
+	 * boundary, since every stream starts over there and a reader beginning
+	 * at the boundary keeps nothing from before it.  Convert the boundary to
+	 * a byte position here, where no lock is held; the reservation itself
+	 * then costs one comparison.
+	 */
+	startbefore = PG_UINT64_MAX;
+	if (insert_stream_slot >= 0 && !insert_stream_restarted)
+	{
+		XLogRecPtr	last_write = WALCompressionSlots[insert_stream_slot].s.last_write;
+		XLogRecPtr	boundary;
+
+		boundary = (last_write / WAL_COMPRESSION_STREAM_RESET + 1) *
+			WAL_COMPRESSION_STREAM_RESET;
+		startbefore = XLogRecPtrToBytePos(boundary);
+	}
+
 	START_CRIT_SECTION();
 
 	if (likely(class == WALINSERT_NORMAL))
@@ -897,6 +1070,21 @@ XLogInsertRecord(XLogRecData *rdata,
 			Assert(RedoRecPtr < Insert->RedoRecPtr);
 			RedoRecPtr = Insert->RedoRecPtr;
 		}
+
+		/*
+		 * A stream must not span a checkpoint's redo point either: recovery
+		 * begins there with nothing earlier to decompress against.  Streams
+		 * do start over when RedoRecPtr moves, but the copy this record was
+		 * compressed under was read before the lock was taken and may already
+		 * have been stale, so check it now that it cannot be.
+		 */
+		if (insert_stream_slot >= 0 && !insert_stream_restarted &&
+			WALCompressionSlots[insert_stream_slot].s.redo != RedoRecPtr)
+		{
+			WALInsertLockRelease();
+			END_CRIT_SECTION();
+			return InvalidXLogRecPtr;
+		}
 		doPageWrites = (Insert->fullPageWrites || Insert->runningBackups > 0);
 
 		if (doPageWrites &&
@@ -916,8 +1104,20 @@ XLogInsertRecord(XLogRecData *rdata,
 		 * Reserve space for the record in the WAL. This also sets the xl_prev
 		 * pointer.
 		 */
-		ReserveXLogInsertLocation(rechdr->xl_tot_len, &StartPos, &EndPos,
-								  &rechdr->xl_prev);
+		if (!ReserveXLogInsertLocation(rechdr->xl_tot_len, &StartPos, &EndPos,
+									   &rechdr->xl_prev, startbefore))
+		{
+			/*
+			 * The record would land past a stream reset boundary while
+			 * continuing a stream that began below it.  A reader starting at
+			 * that boundary would have nothing to decompress it against, so
+			 * throw the record away and let the caller build it again against
+			 * a stream that starts over.
+			 */
+			WALInsertLockRelease();
+			END_CRIT_SECTION();
+			return InvalidXLogRecPtr;
+		}
 
 		/* Normal records are always inserted. */
 		inserted = true;
@@ -952,8 +1152,9 @@ XLogInsertRecord(XLogRecData *rdata,
 		 */
 		Assert(!XLogRecPtrIsValid(fpw_lsn));
 		WALInsertLockAcquireExclusive();
-		ReserveXLogInsertLocation(rechdr->xl_tot_len, &StartPos, &EndPos,
-								  &rechdr->xl_prev);
+		(void) ReserveXLogInsertLocation(rechdr->xl_tot_len, &StartPos,
+										 &EndPos, &rechdr->xl_prev,
+										 PG_UINT64_MAX);
 		RedoRecPtr = Insert->RedoRecPtr = StartPos;
 		inserted = true;
 	}
@@ -1162,9 +1363,9 @@ XLogInsertRecord(XLogRecData *rdata,
  * however, because there are two call sites, the compiler is reluctant to
  * inline. We use pg_always_inline here to try to convince it.
  */
-static pg_always_inline void
+static pg_always_inline bool
 ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos, XLogRecPtr *EndPos,
-						  XLogRecPtr *PrevPtr)
+						  XLogRecPtr *PrevPtr, uint64 startbefore)
 {
 	XLogCtlInsert *Insert = &XLogCtl->Insert;
 	uint64		startbytepos;
@@ -1189,6 +1390,18 @@ ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos, XLogRecPtr *EndPos,
 	SpinLockAcquire(&Insert->insertpos_lck);
 
 	startbytepos = Insert->CurrBytePos;
+
+	/*
+	 * The caller may only place this record below startbefore.  Comparing
+	 * byte positions rather than LSNs is what keeps this to one test: the
+	 * caller converted the boundary once, outside the lock.
+	 */
+	if (startbytepos >= startbefore)
+	{
+		SpinLockRelease(&Insert->insertpos_lck);
+		return false;
+	}
+
 	endbytepos = startbytepos + size;
 	prevbytepos = Insert->PrevBytePos;
 	Insert->CurrBytePos = endbytepos;
@@ -1207,6 +1420,8 @@ ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos, XLogRecPtr *EndPos,
 	Assert(XLogRecPtrToBytePos(*StartPos) == startbytepos);
 	Assert(XLogRecPtrToBytePos(*EndPos) == endbytepos);
 	Assert(XLogRecPtrToBytePos(*PrevPtr) == prevbytepos);
+
+	return true;
 }
 
 /*
@@ -5340,6 +5555,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 +5643,27 @@ 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 = XLogCtl->Insert.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.last_write = InvalidXLogRecPtr;
+			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
@@ -5454,12 +5695,13 @@ XLOGShmemInit(void *arg)
 }
 
 /*
- * XLOGShmemAttach - re-establish WALInsertLocks pointer after attaching.
+ * XLOGShmemAttach - re-establish pointers into shared memory after attaching.
  */
 static void
 XLOGShmemAttach(void *arg)
 {
 	WALInsertLocks = XLogCtl->Insert.WALInsertLocks;
+	WALCompressionSlots = XLogCtl->Insert.WALCompressionSlots;
 }
 
 /*
@@ -8532,6 +8774,14 @@ KeepLogSeg(XLogRecPtr recptr, XLogSegNo *logSegNo)
 	keep = XLogGetReplicationSlotMinimumLSN();
 	if (XLogRecPtrIsValid(keep) && keep < recptr)
 	{
+		/*
+		 * A reader starting where the slot needs it rewinds to the stream
+		 * reset boundary below that, so keep the WAL from there on.  Done
+		 * whatever wal_compression_streams says, since the slot may well point
+		 * into WAL written while it said something else.
+		 */
+		keep -= keep % WAL_COMPRESSION_STREAM_RESET;
+
 		XLByteToSeg(keep, segno, wal_segment_size);
 
 		/*
diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index a5a3a1422cb..0af8f01974e 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -171,6 +171,29 @@ 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;
+
+/*
+ * The stream slot this backend is using, or -1.  Holding on to it is the
+ * whole point: a stream only pays once it has seen some records.
+ *
+ * When every slot is held by a backend that is still writing, there is
+ * nothing to take, and we compress records on their own instead.  Looking
+ * for a slot costs a spinlock read, so after a failed search we do not look
+ * again for a while.
+ */
+static int	my_stream_slot = -1;
+static int	stream_scan_backoff = 0;
+
+#define STREAM_SCAN_BACKOFF_RECORDS 1000
+#endif
+
 /*
  * An array of XLogRecData structs, to hold registered data.
  */
@@ -515,6 +538,43 @@ XLogSetRecordFlags(uint8 flags)
 	curinsert_flags |= flags;
 }
 
+#ifdef USE_ZSTD
+
+/*
+ * 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.
+ * XLOG_CHECKPOINT_REDO marks the point recovery starts from, and is read
+ * before anything that could have preceded it in a stream.
+ */
+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_CHECKPOINT_REDO ||
+			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;
+}
+
+#endif							/* USE_ZSTD */
+
 /* Compress the assembled record; NULL if that did not pay off */
 static XLogRecData *
 XLogCompressRdt(XLogRecData *rdt)
@@ -554,6 +614,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 +689,139 @@ 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		skipped;
+	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)
+		flat_len += r->len;
+
+	/* The first chunk is the record header; see XLogInsertRecord(). */
+	src_header = (XLogRecord *) rdt->data;
+	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;
+
+	out.dst = (char *) &compressed_header[1];
+	out.size = WAL_COMPRESSION_BUFSIZE - SizeOfXLogCompressedRecord;
+	out.pos = 0;
+
+	/*
+	 * Feed the chain as it stands.  Flattening it first would have to write
+	 * somewhere, and the only buffer large enough is the one holding the
+	 * compressed page images this very chain points into.
+	 */
+	skipped = 0;
+	for (const XLogRecData *r = rdt; r != NULL; r = r->next)
+	{
+		const char *data = r->data;
+		uint32		len = r->len;
+
+		/* The record header travels in the clear, ahead of the payload. */
+		if (skipped < SizeOfXLogRecord)
+		{
+			uint32		skip = Min(len, SizeOfXLogRecord - skipped);
+
+			data += skip;
+			len -= skip;
+			skipped += skip;
+		}
+		if (len == 0)
+			continue;
+
+		in.src = data;
+		in.size = len;
+		in.pos = 0;
+		while (in.pos < in.size)
+		{
+			rem = ZSTD_compressStream2(cctx, &out, &in, ZSTD_e_continue);
+			if (ZSTD_isError(rem) || out.pos == out.size)
+			{
+				/* no room left, and the compressor has eaten part of it */
+				*poisoned = true;
+				return NULL;
+			}
+		}
+	}
+
+	/* End the block, so a reader gets this record back on its own. */
+	in.src = NULL;
+	in.size = 0;
+	in.pos = 0;
+	do
+	{
+		rem = ZSTD_compressStream2(cctx, &out, &in, ZSTD_e_flush);
+		if (ZSTD_isError(rem) || (rem != 0 && out.pos == out.size))
+		{
+			*poisoned = true;
+			return NULL;
+		}
+	} while (rem != 0);
+
+	/*
+	 * A record that did not come out smaller is still used.  Dropping it
+	 * would mean restarting the stream, since the compressor has already
+	 * consumed it, and a cold stream costs far more than the few bytes this
+	 * record loses.
+	 */
+
+	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 +872,20 @@ 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.
+	 */
+#ifdef USE_ZSTD
+	bool		use_stream;
+
+	use_stream = (wal_compression_streams > 0 &&
+				  wal_compression == WAL_COMPRESSION_ZSTD &&
+				  stream_cctx != NULL &&
+				  XLogRecordJoinsStream(rmid, info));
+#endif
+
 	/* XLogBeginInsert() must have been called. */
 	if (!begininsert_called)
 		elog(ERROR, "XLogBeginInsert was not called");
@@ -735,6 +943,67 @@ 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
+		{
+			int			slot = -1;
+			bool		restart = false;
+			bool		poisoned;
+			XLogRecData *rdt_compressed;
+
+			/*
+			 * Take a stream, unless a recent look found every one of them
+			 * held by a backend that is still writing.  Then hold it 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.
+			 */
+			if (use_stream && rec_size <= WAL_COMPRESSION_BUFSIZE &&
+				(my_stream_slot >= 0 || stream_scan_backoff-- <= 0))
+			{
+				slot = XLogCompressionStreamAcquire(my_stream_slot, RedoRecPtr,
+													&restart);
+				my_stream_slot = slot;
+				if (slot < 0)
+					stream_scan_backoff = STREAM_SCAN_BACKOFF_RECORDS;
+			}
+
+			if (slot < 0)
+				goto no_stream;
+
+			rdt_compressed = XLogCompressRdtStream(rdt, 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(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(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;
+		}
+no_stream:
+#endif
+
 		if (try_whole_record)
 		{
 			bool		whole_record_compressed = false;
@@ -858,6 +1127,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 a07f258e375..eae2a5ae7d5 100644
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -56,6 +56,8 @@ static bool ValidXLogRecord(XLogReaderState *state, XLogRecord *record,
 static void ResetDecoder(XLogReaderState *state);
 static void WALOpenSegmentInit(WALOpenSegment *seg, WALSegmentContext *segcxt,
 							   int segsize, const char *waldir);
+static XLogRecPtr XLogFindRecordStart(XLogReaderState *state, XLogRecPtr RecPtr,
+									  char **errormsg);
 static XLogRecord *XLogDecompressRecordIfNeeded(XLogReaderState *state,
 												XLogRecord *record,
 												XLogRecPtr recptr);
@@ -181,6 +183,16 @@ XLogReaderFree(XLogReaderState *state)
 #endif
 	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);
+		pfree(state->stream_ready);
+	}
+#endif
 	pfree(state->readBuf);
 	pfree(state);
 }
@@ -256,6 +268,88 @@ XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
 	state->NextRecPtr = RecPtr;
 	state->ReadRecPtr = InvalidXLogRecPtr;
 	state->DecodeRecPtr = InvalidXLogRecPtr;
+	state->warmupEndPtr = InvalidXLogRecPtr;
+}
+
+/*
+ * Prepare to read records from RecPtr onwards, which unlike XLogBeginRead()
+ * works even when the records there were compressed against earlier ones.
+ *
+ * A record compressed as part of a stream cannot be decompressed on its own,
+ * so reading starts at the last stream reset boundary at or below RecPtr,
+ * where every stream is known to begin again.  The records in between are
+ * decoded only to rebuild the decompressors and are not returned by
+ * XLogReadRecord().
+ *
+ * Returns the position reading actually starts from, or InvalidXLogRecPtr with
+ * *errormsg set.  If the WAL below RecPtr is gone -- a replication slot may
+ * hold nothing older than what it needs itself -- reading starts at RecPtr,
+ * which is correct for every record that is not part of a stream.
+ */
+XLogRecPtr
+XLogBeginReadStreamed(XLogReaderState *state, XLogRecPtr RecPtr,
+					  char **errormsg)
+{
+	XLogRecPtr	boundary;
+	XLogRecPtr	found;
+	char	   *msg;
+
+	Assert(XLogRecPtrIsValid(RecPtr));
+
+	if (errormsg == NULL)
+		errormsg = &msg;	/* the caller is content with the fallback */
+	*errormsg = NULL;
+	boundary = RecPtr - (RecPtr % WAL_COMPRESSION_STREAM_RESET);
+
+	if (boundary == 0)
+	{
+		XLogBeginRead(state, RecPtr);
+		return RecPtr;
+	}
+
+	found = XLogFindRecordStart(state, boundary, errormsg);
+	if (XLogRecPtrIsValid(found))
+	{
+		/*
+		 * Read forward from the boundary to bring the decompressors up to
+		 * date, stopping once the next record to read is the one the caller
+		 * asked for.  That record must not be read here: a stream record fed
+		 * to its decompressor twice leaves it in a state the next records
+		 * were not compressed against.
+		 */
+		XLogBeginRead(state, found);
+		state->warmupEndPtr = RecPtr;
+		while (state->NextRecPtr < RecPtr)
+		{
+			if (XLogReadRecord(state, errormsg) == NULL)
+				break;
+		}
+
+		if (state->NextRecPtr >= RecPtr)
+		{
+			/*
+			 * Leave the reader exactly as XLogBeginRead() would have, so that
+			 * callers see no trace of the rewind.  The decompressors are not
+			 * part of that state and stay as the warm-up left them.
+			 */
+			found = state->NextRecPtr;
+			XLogBeginRead(state, found);
+			return found;
+		}
+	}
+
+	{
+		/*
+		 * The WAL below RecPtr is gone; a replication slot need not keep
+		 * anything older than it uses itself.  Start where we were asked to,
+		 * which reads every record that is not part of a stream.
+		 */
+		*errormsg = NULL;
+		state->warmupEndPtr = InvalidXLogRecPtr;
+		return XLogFindNextRecord(state, RecPtr, errormsg);
+	}
+
+	return found;
 }
 
 /*
@@ -1017,7 +1111,26 @@ restart:
 		Assert(decoded != NULL);
 	}
 
-	if (DecodeXLogRecord(state, decoded, record, RecPtr, &errormsg))
+	if (state->framing_only)
+	{
+		/*
+		 * Fill in only what says where this record sits.  Nothing reads the
+		 * payload in this mode, and decoding it would run the record through
+		 * a decompressor that has already consumed it.
+		 */
+		decoded->header = *record;
+		decoded->lsn = RecPtr;
+		decoded->next = NULL;
+		decoded->record_origin = InvalidReplOriginId;
+		decoded->toplevel_xid = InvalidTransactionId;
+		decoded->main_data = NULL;
+		decoded->main_data_len = 0;
+		decoded->max_block_id = -1;
+		decoded->size = MAXALIGN(offsetof(DecodedXLogRecord, blocks));
+	}
+
+	if (state->framing_only ||
+		DecodeXLogRecord(state, decoded, record, RecPtr, &errormsg))
 	{
 		/* Record the location of the next record. */
 		decoded->next_lsn = state->NextRecPtr;
@@ -1046,6 +1159,21 @@ restart:
 		return XLREAD_SUCCESS;
 	}
 
+	if (RecPtr < state->warmupEndPtr)
+	{
+		/*
+		 * We are only reading this record to feed the decompressors, and its
+		 * stream has not started over yet, so it cannot be decoded and the
+		 * caller was never going to see it.  Move on to the next one.
+		 */
+		if (decoded->oversized)
+			pfree(decoded);
+		state->errormsg_buf[0] = '\0';
+		state->errormsg_deferred = false;
+		RecPtr = state->NextRecPtr;
+		goto restart;
+	}
+
 err:
 	if (assembled)
 	{
@@ -1511,27 +1639,18 @@ XLogReaderResetError(XLogReaderState *state)
 }
 
 /*
- * Find the first record with an lsn >= RecPtr.
+ * Find where the first record at or after RecPtr starts, without reading it.
  *
- * This is different from XLogBeginRead() in that RecPtr doesn't need to point
- * to a valid record boundary.  Useful for checking whether RecPtr is a valid
- * xlog address for reading, and to find the first valid address after some
- * address when dumping records for debugging purposes.
+ * RecPtr does not need to point to a record boundary: page headers alone tell
+ * us where to skip continuation data, which is enough to land on the start of
+ * a record.
  *
- * This positions the reader, like XLogBeginRead(), so that the next call to
- * XLogReadRecord() will read the next valid record.
- *
- * On failure, InvalidXLogRecPtr is returned, and *errormsg is set to a string
- * with details of the failure.
- *
- * When set, *errormsg points to an internal buffer that's valid until the next
- * call to XLogReadRecord.
+ * On failure, InvalidXLogRecPtr is returned and *errormsg is set.
  */
-XLogRecPtr
-XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg)
+static XLogRecPtr
+XLogFindRecordStart(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg)
 {
 	XLogRecPtr	tmpRecPtr;
-	XLogRecPtr	found = InvalidXLogRecPtr;
 	XLogPageHeader header;
 
 	*errormsg = NULL;
@@ -1618,18 +1737,7 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg)
 	 * because either we're at the first record after the beginning of a page
 	 * or we just jumped over the remaining data of a continuation.
 	 */
-	XLogBeginRead(state, tmpRecPtr);
-	while (XLogReadRecord(state, errormsg) != NULL)
-	{
-		/* past the record we've found, break out */
-		if (RecPtr <= state->ReadRecPtr)
-		{
-			/* Rewind the reader to the beginning of the last record. */
-			found = state->ReadRecPtr;
-			XLogBeginRead(state, found);
-			return found;
-		}
-	}
+	return tmpRecPtr;
 
 err:
 	XLogReaderInvalReadState(state);
@@ -1648,6 +1756,49 @@ err:
 	return InvalidXLogRecPtr;
 }
 
+/*
+ * Find the first record with an lsn >= RecPtr.
+ *
+ * This is different from XLogBeginRead() in that RecPtr doesn't need to point
+ * to a valid record boundary.  Useful for checking whether RecPtr is a valid
+ * xlog address for reading, and to find the first valid address after some
+ * address when dumping records for debugging purposes.
+ *
+ * This positions the reader, like XLogBeginRead(), so that the next call to
+ * XLogReadRecord() will read the next valid record.
+ *
+ * On failure, InvalidXLogRecPtr is returned, and *errormsg is set to a string
+ * with details of the failure.
+ *
+ * When set, *errormsg points to an internal buffer that's valid until the next
+ * call to XLogReadRecord.
+ */
+XLogRecPtr
+XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg)
+{
+	XLogRecPtr	start;
+
+	start = XLogFindRecordStart(state, RecPtr, errormsg);
+	if (!XLogRecPtrIsValid(start))
+		return InvalidXLogRecPtr;
+
+	XLogBeginRead(state, start);
+	while (XLogReadRecord(state, errormsg) != NULL)
+	{
+		/* past the record we've found, break out */
+		if (RecPtr <= state->ReadRecPtr)
+		{
+			/* Rewind the reader to the beginning of the last record. */
+			XLogRecPtr	found = state->ReadRecPtr;
+
+			XLogBeginRead(state, found);
+			return found;
+		}
+	}
+
+	return InvalidXLogRecPtr;
+}
+
 /*
  * Helper function to ease writing of XLogReaderRoutine->page_read callbacks.
  * If this function is used, caller must supply a segment_open callback in
@@ -1869,7 +2020,91 @@ 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_array(void *, XLR_MAX_STREAMS);
+				state->stream_ready = palloc0_array(bool, XLR_MAX_STREAMS);
+			}
+
+			/*
+			 * A reader that started in the middle of WAL meets records whose
+			 * stream began earlier.  Nothing here can decompress them, and
+			 * feeding them to a fresh context would decode to whatever the
+			 * bytes happen to mean, so refuse until the stream starts over.
+			 */
+			if (!(src->stream_flags & XLR_STREAM_RESET) &&
+				!state->stream_ready[src->stream])
+			{
+				report_invalid_record(state,
+									  "no decompression state for stream %u at %X/%08X",
+									  src->stream, LSN_FORMAT_ARGS(recptr));
+				return NULL;
+			}
+
+			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);
+				state->stream_ready[src->stream] = true;
+			}
+
+			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/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.c
index 6de13b91748..b6edc7c9343 100644
--- a/src/backend/access/transam/xlogrecovery.c
+++ b/src/backend/access/transam/xlogrecovery.c
@@ -1493,7 +1493,15 @@ FinishWalRecovery(void)
 		lastRecTLI = XLogRecoveryCtl->lastReplayedTLI;
 	}
 	XLogPrefetcherBeginRead(xlogprefetcher, lastRec);
+
+	/*
+	 * Only the extent of the record and its page are wanted here, and a
+	 * record that was compressed against earlier ones cannot be decoded a
+	 * second time.
+	 */
+	xlogreader->framing_only = true;
 	(void) ReadRecord(xlogprefetcher, PANIC, false, lastRecTLI);
+	xlogreader->framing_only = false;
 	endOfLog = xlogreader->EndRecPtr;
 
 	/*
diff --git a/src/backend/postmaster/walsummarizer.c b/src/backend/postmaster/walsummarizer.c
index ff246b07a21..9888a0aecd9 100644
--- a/src/backend/postmaster/walsummarizer.c
+++ b/src/backend/postmaster/walsummarizer.c
@@ -1061,12 +1061,13 @@ SummarizeWAL(TimeLineID tli, XLogRecPtr start_lsn, bool exact,
 		 * file is less than the start LSN of the next file. When only a page
 		 * header is skipped, nothing has been missed.
 		 */
-		XLogBeginRead(xlogreader, start_lsn);
+		XLogBeginReadStreamed(xlogreader, start_lsn, NULL);
 		summary_start_lsn = start_lsn;
 	}
 	else
 	{
-		summary_start_lsn = XLogFindNextRecord(xlogreader, start_lsn, &errormsg);
+		summary_start_lsn = XLogBeginReadStreamed(xlogreader, start_lsn,
+												  &errormsg);
 		if (!XLogRecPtrIsValid(summary_start_lsn))
 		{
 			/*
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 98e5f1dd8f9..30ff38b44a8 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -715,7 +715,7 @@ DecodingContextFindStartpoint(LogicalDecodingContext *ctx)
 	ReplicationSlot *slot = ctx->slot;
 
 	/* Initialize from where to start reading WAL. */
-	XLogBeginRead(ctx->reader, slot->data.restart_lsn);
+	XLogBeginReadStreamed(ctx->reader, slot->data.restart_lsn, NULL);
 
 	elog(DEBUG1, "searching for logical decoding starting point, starting at %X/%08X",
 		 LSN_FORMAT_ARGS(slot->data.restart_lsn));
@@ -2121,7 +2121,8 @@ LogicalReplicationSlotCheckPendingWal(XLogRecPtr end_of_wal,
 		 * Start reading at the slot's restart_lsn, which we know points to a
 		 * valid record.
 		 */
-		XLogBeginRead(ctx->reader, MyReplicationSlot->data.restart_lsn);
+		XLogBeginReadStreamed(ctx->reader,
+							  MyReplicationSlot->data.restart_lsn, NULL);
 
 		/* Invalidate non-timetravel entries */
 		InvalidateSystemCaches();
@@ -2224,7 +2225,8 @@ LogicalSlotAdvanceAndCheckSnapState(XLogRecPtr moveto,
 		 * Start reading at the slot's restart_lsn, which we know to point to
 		 * a valid record.
 		 */
-		XLogBeginRead(ctx->reader, MyReplicationSlot->data.restart_lsn);
+		XLogBeginReadStreamed(ctx->reader,
+							  MyReplicationSlot->data.restart_lsn, NULL);
 
 		/* invalidate non-timetravel entries */
 		InvalidateSystemCaches();
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index 71fbaf72269..67251c15071 100644
--- a/src/backend/replication/logical/logicalfuncs.c
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -243,7 +243,8 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
 		 * xacts that committed after the slot's confirmed_flush can be
 		 * accumulated into reorder buffers.
 		 */
-		XLogBeginRead(ctx->reader, MyReplicationSlot->data.restart_lsn);
+		XLogBeginReadStreamed(ctx->reader,
+							  MyReplicationSlot->data.restart_lsn, NULL);
 
 		/* invalidate non-timetravel entries */
 		InvalidateSystemCaches();
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index c65dd324325..31f37ca09d9 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1578,9 +1578,6 @@ StartLogicalReplication(StartReplicationCmd *cmd)
 	pq_endmessage(&buf);
 	pq_flush();
 
-	/* Start reading WAL from the oldest required WAL. */
-	XLogBeginRead(logical_decoding_ctx->reader,
-				  MyReplicationSlot->data.restart_lsn);
 
 	/*
 	 * Report the location after which we'll send out further commits as the
@@ -1597,6 +1594,14 @@ StartLogicalReplication(StartReplicationCmd *cmd)
 
 	SyncRepInitConfig();
 
+	/*
+	 * Start reading WAL from the oldest required WAL.  Done once the walsender
+	 * is otherwise ready, because rebuilding the decompressors reads WAL
+	 * through the same callback the main loop uses.
+	 */
+	XLogBeginReadStreamed(logical_decoding_ctx->reader,
+						  MyReplicationSlot->data.restart_lsn, NULL);
+
 	/* Main loop of walsender */
 	WalSndLoop(XLogSendLogical);
 
diff --git a/src/backend/utils/activity/wait_event_names.txt b/src/backend/utils/activity/wait_event_names.txt
index 256b3a3c02e..8d34545f541 100644
--- a/src/backend/utils/activity/wait_event_names.txt
+++ b/src/backend/utils/activity/wait_event_names.txt
@@ -418,6 +418,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 0a7d8360698..5ebd1d48f4e 100644
--- a/src/backend/utils/misc/guc_parameters.dat
+++ b/src/backend/utils/misc/guc_parameters.dat
@@ -3476,6 +3476,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 1e6caceb07c..c97f02c97a3 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -267,6 +267,10 @@
                                         # of zstd, lz4 and pglz, supported by the build)
 #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/bin/pg_rewind/parsexlog.c b/src/bin/pg_rewind/parsexlog.c
index 023e23b063c..2a59b5f726a 100644
--- a/src/bin/pg_rewind/parsexlog.c
+++ b/src/bin/pg_rewind/parsexlog.c
@@ -79,7 +79,7 @@ extractPageMap(const char *datadir, XLogRecPtr startpoint, int tliIndex,
 	if (xlogreader == NULL)
 		pg_fatal("out of memory while allocating a WAL reading processor");
 
-	XLogBeginRead(xlogreader, startpoint);
+	XLogBeginReadStreamed(xlogreader, startpoint, NULL);
 	do
 	{
 		record = XLogReadRecord(xlogreader, &errormsg);
@@ -138,7 +138,12 @@ readOneRecord(const char *datadir, XLogRecPtr ptr, int tliIndex,
 	if (xlogreader == NULL)
 		pg_fatal("out of memory while allocating a WAL reading processor");
 
+	/*
+	 * Only the extent of the record is wanted, so there is no need to decode
+	 * it -- which also means no need to rewind for a compression stream.
+	 */
 	XLogBeginRead(xlogreader, ptr);
+	xlogreader->framing_only = true;
 	record = XLogReadRecord(xlogreader, &errormsg);
 	if (record == NULL)
 	{
@@ -205,7 +210,13 @@ findLastCheckpoint(const char *datadir, XLogRecPtr forkptr, int tliIndex,
 	{
 		uint8		info;
 
+		/*
+		 * Walking back only needs each record's header, so decode nothing:
+		 * rewinding to rebuild a decompressor at every step would make this
+		 * loop cost the whole distance walked, over and over.
+		 */
 		XLogBeginRead(xlogreader, searchptr);
+		xlogreader->framing_only = true;
 		record = XLogReadRecord(xlogreader, &errormsg);
 
 		if (record == NULL)
@@ -251,6 +262,17 @@ findLastCheckpoint(const char *datadir, XLogRecPtr forkptr, int tliIndex,
 		{
 			CheckPoint	checkPoint;
 
+			/*
+			 * This one is wanted whole.  A checkpoint record never joins a
+			 * compression stream, so it reads on its own.
+			 */
+			XLogBeginRead(xlogreader, searchptr);
+			xlogreader->framing_only = false;
+			record = XLogReadRecord(xlogreader, &errormsg);
+			if (record == NULL)
+				pg_fatal("could not read checkpoint record at %X/%08X",
+						 LSN_FORMAT_ARGS(searchptr));
+
 			memcpy(&checkPoint, XLogRecGetData(xlogreader), sizeof(CheckPoint));
 			*lastchkptrec = searchptr;
 			*lastchkpttli = checkPoint.ThisTimeLineID;
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 6f0574a8764..a66ad0a2644 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -1412,8 +1412,14 @@ main(int argc, char **argv)
 	atexit(cleanup_tmpwal_dir_atexit);
 	xlogreader_state_cleanup = xlogreader_state;
 
-	/* first find a valid recptr to start from */
-	first_record = XLogFindNextRecord(xlogreader_state, private.startptr, &errormsg);
+	/*
+	 * Find a valid recptr to start from.  A record can be compressed against
+	 * earlier records of its stream, so this backs up to where the streams
+	 * start over and decodes from there; those earlier records rebuild the
+	 * decompressors and are not printed.
+	 */
+	first_record = XLogBeginReadStreamed(xlogreader_state, private.startptr,
+										 &errormsg);
 
 	if (!XLogRecPtrIsValid(first_record))
 	{
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 91e3aefed21..a4e9251410c 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;
 
@@ -224,6 +225,10 @@ typedef enum WALAvailability
 struct XLogRecData;
 struct XLogReaderState;
 
+extern int	XLogCompressionStreamAcquire(int preferred, XLogRecPtr redo,
+									 bool *restart);
+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 5c869d78be7..e213e014f09 100644
--- a/src/include/access/xlogreader.h
+++ b/src/include/access/xlogreader.h
@@ -235,6 +235,27 @@ struct XLogReaderState
 	 * this header is included where zstd.h is not.
 	 */
 	void	   *fpi_dctx;
+
+	/*
+	 * Records below this are decoded to rebuild the decompressors and then
+	 * dropped rather than returned; see XLogBeginReadStreamed().
+	 */
+	XLogRecPtr	warmupEndPtr;
+
+	/*
+	 * Frame records without decoding them: the caller only wants to know
+	 * where a record ends and to have its page in readBuf.  A record that
+	 * belongs to a compression stream cannot be decoded twice anyway, since
+	 * the decompressor has already consumed it.
+	 */
+	bool		framing_only;
+
+	/*
+	 * Per stream: has a record that starts the stream over been seen yet?
+	 * Until one has, the stream's earlier records cannot be decompressed,
+	 * because the state they were compressed against is not here.
+	 */
+	bool	   *stream_ready;
 	XLogRecPtr	NextRecPtr;		/* end+1 of last record decoded */
 	XLogRecPtr	PrevRecPtr;		/* start of previous record decoded */
 
@@ -262,6 +283,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
@@ -353,6 +381,8 @@ extern void XLogReaderSetDecodeBuffer(XLogReaderState *state,
 
 /* Position the XLogReader to given record */
 extern void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr);
+extern XLogRecPtr XLogBeginReadStreamed(XLogReaderState *state,
+										XLogRecPtr RecPtr, char **errormsg);
 extern XLogRecPtr XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr,
 									 char **errormsg);
 
diff --git a/src/include/access/xlogrecord.h b/src/include/access/xlogrecord.h
index be662e6558f..e9332b5ccb7 100644
--- a/src/include/access/xlogrecord.h
+++ b/src/include/access/xlogrecord.h
@@ -205,11 +205,44 @@ 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)
 
+/*
+ * Every stream starts over at fixed WAL_COMPRESSION_STREAM_RESET boundaries.
+ * That is what lets a reader begin in the middle of WAL: it rewinds to the
+ * last boundary below the record it wants and reads forward from there, since
+ * every stream begins again at or after such a boundary.  The distance is
+ * therefore what a reader has to re-read, and what a replication slot has to
+ * keep beyond the WAL it needs for itself.
+ *
+ * A fixed distance rather than the WAL segment size, so that how far a reader
+ * rewinds does not change when a cluster is initialised with a different
+ * segment size.  Compression is insensitive to the value -- pgbench emits the
+ * same WAL per transaction to within 2% anywhere between 1MB and 64MB -- so it
+ * is chosen for the readers, not for the ratio.
+ */
+#define WAL_COMPRESSION_STREAM_RESET	(UINT64CONST(4) * 1024 * 1024)
+
+/* Must a stream that last wrote at "from" start over to write at "to"? */
+#define XLogCompressionStreamCrosses(from, to) \
+	((uint64) (from) / WAL_COMPRESSION_STREAM_RESET != \
+	 (uint64) (to) / WAL_COMPRESSION_STREAM_RESET)
+
 /*
  * Maximum size of the header for a block reference. This is used to size a
  * temporary buffer for constructing the header.
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)
diff --git a/src/test/perl/PostgreSQL/Test/Cluster.pm b/src/test/perl/PostgreSQL/Test/Cluster.pm
index 04e127ba924..55d068d62ca 100644
--- a/src/test/perl/PostgreSQL/Test/Cluster.pm
+++ b/src/test/perl/PostgreSQL/Test/Cluster.pm
@@ -731,8 +731,14 @@ sub init
 	# This is printed after TEMP_CONFIG on purpose.  It is a correctness
 	# requirement of the test, not a preference, so it must win over whatever
 	# the buildfarm animal supplies.
-	print $conf "wal_compression_threshold = " . (1024 * 1024 * 1024) . "\n"
-	  if $params{no_wal_compression};
+	if ($params{no_wal_compression})
+	{
+		print $conf "wal_compression_threshold = " . (1024 * 1024 * 1024) . "\n";
+
+		# Streams ignore the threshold on purpose -- reaching small records is
+		# what they are for -- so they have to be turned off separately.
+		print $conf "wal_compression_streams = 0\n";
+	}
 
 	if ($params{allows_streaming})
 	{
diff --git a/src/test/recovery/Makefile b/src/test/recovery/Makefile
index d41aaaf8ae1..b73bdb7fe1d 100644
--- a/src/test/recovery/Makefile
+++ b/src/test/recovery/Makefile
@@ -11,6 +11,7 @@
 
 EXTRA_INSTALL=contrib/pg_prewarm \
 	contrib/pg_stat_statements \
+	contrib/pg_walinspect \
 	contrib/test_decoding \
 	src/test/modules/injection_points
 
diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build
index 39ec8c4946d..b119596c08c 100644
--- a/src/test/recovery/meson.build
+++ b/src/test/recovery/meson.build
@@ -64,6 +64,7 @@ tests += {
       't/053_standby_login_event_trigger.pl',
       't/054_unlogged_sequence_promotion.pl',
       't/055_cascade_reconnect.pl',
+      't/056_wal_compression.pl',
     ],
   },
 }
diff --git a/src/test/recovery/t/043_no_contrecord_switch.pl b/src/test/recovery/t/043_no_contrecord_switch.pl
index 6fab73a064a..de219160222 100644
--- a/src/test/recovery/t/043_no_contrecord_switch.pl
+++ b/src/test/recovery/t/043_no_contrecord_switch.pl
@@ -53,7 +53,10 @@ sub start_of_page
 }
 
 my $primary = PostgreSQL::Test::Cluster->new('primary');
-$primary->init(allows_streaming => 1, has_archiving => 1);
+# This test computes the size of a record so that it overflows a WAL page, so
+# the record has to reach WAL at the size asked for.
+$primary->init(allows_streaming => 1, has_archiving => 1,
+	no_wal_compression => 1);
 
 # The configuration is chosen here to minimize the friction with
 # concurrent WAL activity.  checkpoint_timeout avoids noise with
diff --git a/src/test/recovery/t/052_wal_compression.pl b/src/test/recovery/t/056_wal_compression.pl
similarity index 53%
rename from src/test/recovery/t/052_wal_compression.pl
rename to src/test/recovery/t/056_wal_compression.pl
index 001772daf23..79cada8a39d 100644
--- a/src/test/recovery/t/052_wal_compression.pl
+++ b/src/test/recovery/t/056_wal_compression.pl
@@ -25,6 +25,15 @@ push @methods, 'pglz';
 # used to verify that compressed WAL is transmitted and decoded correctly.
 # Then the primary is stopped immediately (simulating a crash) and
 # restarted to verify that startup recovery can decompress WAL records.
+# Render a string with its non-printable bytes visible, so that a difference
+# which a terminal hides can still be read in the test output.
+sub _escape
+{
+	my ($str) = @_;
+	$str =~ s/([^\x20-\x7e])/sprintf('\\x%02x', ord($1))/ge;
+	return $str;
+}
+
 # Emit one 256kB compressible logical message at the given threshold and
 # report how many bytes of WAL it consumed.
 sub wal_used
@@ -55,10 +64,14 @@ sub test_wal_compression
 	$primary->init(allows_streaming => 1);
 
 	# Use the minimum threshold so virtually every record gets compressed.
+	# Streams are off here: this part measures what whole-record compression
+	# does on its own, and streams would compress both sides of that
+	# comparison.  They are covered separately below.
 	$primary->append_conf(
 		'postgresql.conf',
 		"wal_compression = '$method'\n"
-		  . "wal_compression_threshold = 32\n");
+		  . "wal_compression_threshold = 32\n"
+		  . "wal_compression_streams = 0\n");
 	$primary->start;
 
 	my $backup_name = "backup_$method";
@@ -149,4 +162,110 @@ foreach my $method (@methods)
 	test_wal_compression($method);
 }
 
+# A record compressed against earlier records of its stream cannot be read on
+# its own, so a reader that starts partway through WAL has to rewind far enough
+# to rebuild the decompressors.  Check that what it then reads is what a reader
+# that started at the beginning sees at the same place.
+SKIP:
+{
+	skip 'zstd not supported by this build', 3
+	  unless check_pg_config('#define HAVE_LIBZSTD 1');
+
+	my $node = PostgreSQL::Test::Cluster->new('streams');
+	$node->init;
+	$node->append_conf(
+		'postgresql.conf', qq(
+wal_compression = zstd
+wal_compression_streams = 8
+wal_compression_threshold = 64
+wal_keep_size = 1GB
+max_wal_size = 1GB
+));
+	$node->start;
+
+	my $start_lsn = $node->safe_psql('postgres',
+		'SELECT pg_current_wal_insert_lsn()');
+
+	# Enough traffic to cross several stream reset boundaries.
+	$node->safe_psql(
+		'postgres', q{
+		CREATE TABLE t (id int, pad text);
+		INSERT INTO t SELECT g, repeat('a', 200) FROM generate_series(1, 200000) g;
+		UPDATE t SET pad = repeat('b', 200) WHERE id % 3 = 0;
+	});
+	my $end_lsn =
+	  $node->safe_psql('postgres', 'SELECT pg_current_wal_flush_lsn()');
+
+	my $waldir = $node->data_dir . '/pg_wal';
+	my ($full, $full_err) = run_command(
+		[ 'pg_waldump', '--path' => $waldir,
+		  '--start' => $start_lsn, '--end' => $end_lsn ]);
+	my @full = split(/\n/, $full);
+	ok(@full > 1000, 'the workload produced records to read');
+
+	# Start halfway in, at an LSN that is nobody's record boundary.
+	my $mid = $node->safe_psql('postgres',
+		"SELECT ('$start_lsn'::pg_lsn + (('$end_lsn'::pg_lsn - '$start_lsn'::pg_lsn) / 2)::bigint)::text"
+	);
+	my ($part, $part_err) = run_command(
+		[ 'pg_waldump', '--path' => $waldir,
+		  '--start' => $mid, '--end' => $end_lsn ]);
+	my @part = split(/\n/, $part);
+	unlike($part_err, qr/error/,
+		'reading from an arbitrary LSN reports no error');
+
+	# Whatever it starts with must appear in the full dump, and everything
+	# from there on must match it line for line.  Trailing whitespace is
+	# normalised away: how a line ends is the platform's business, not this
+	# test's.
+	my ($first) = $part[0] =~ /lsn: ([0-9A-F]+\/[0-9A-F]+),/;
+	my ($at) = grep { $full[$_] =~ /lsn: \Q$first\E,/ } 0 .. $#full;
+	my @tail = defined $at ? @full[ $at .. $#full ] : ();
+	s/\s+\z// for @part, @tail;
+
+	my $same = is_deeply(\@part, \@tail,
+		'records read from an arbitrary LSN match a full read');
+
+	# On failure, say what actually differs: the strings print the same when
+	# they differ only in bytes a terminal does not show.
+	if (!$same)
+	{
+		diag(sprintf('read from %s: %d lines, tail of full read: %d lines',
+				$mid, scalar(@part), scalar(@tail)));
+		for my $i (0 .. $#part)
+		{
+			next if defined $tail[$i] and $part[$i] eq $tail[$i];
+			diag(sprintf("first difference at line %d\n  got: %s\n  exp: %s",
+					$i,
+					_escape($part[$i]),
+					defined $tail[$i] ? _escape($tail[$i]) : '(missing)'));
+			last;
+		}
+	}
+
+	# pg_walinspect reads from an LSN its caller picks, so it has to rewind
+	# the same way pg_waldump does.
+  SKIP:
+	{
+		skip 'pg_walinspect not installed', 1
+		  unless $node->check_extension('pg_walinspect');
+
+		$node->safe_psql('postgres', 'CREATE EXTENSION pg_walinspect');
+		my $differing = $node->safe_psql(
+			'postgres', qq{
+			WITH f AS (SELECT start_lsn, record_type, record_length
+			             FROM pg_get_wal_records_info('$start_lsn', '$end_lsn')
+			            WHERE start_lsn >= '$mid'::pg_lsn),
+			     t AS (SELECT start_lsn, record_type, record_length
+			             FROM pg_get_wal_records_info('$mid', '$end_lsn'))
+			SELECT (SELECT count(*) FROM (SELECT * FROM f EXCEPT ALL SELECT * FROM t) a)
+			     + (SELECT count(*) FROM (SELECT * FROM t EXCEPT ALL SELECT * FROM f) b)
+		});
+		is($differing, '0',
+			'pg_walinspect from an arbitrary LSN matches a full read');
+	}
+
+	$node->stop;
+}
+
 done_testing();
-- 
2.50.1 (Apple Git-155)

