From d49b82edd7d5ed84b3186b62423db6ffca499b14 Mon Sep 17 00:00:00 2001
From: Andrey Borodin <amborodin@acm.org>
Date: Wed, 8 Jan 2025 16:39:20 +0500
Subject: [PATCH v8 2/3] Add whole-record WAL compression alongside FPI
 compression

When a record is larger than wal_compression_threshold, compress it as a
single unit instead of compressing each full-page image separately.  This
wins whenever the images in one record share content, which is typical of
B-tree index builds: CREATE INDEX over 10M random doubles drops from
189.5MB of WAL to 161.0MB with lz4, and from 144.8MB to 116.9MB with
zstd.

Only lz4 and zstd take part.  pglz refuses input it cannot shrink by 25%,
which a record made of full-page images rarely clears, so it saved
nothing while costing an extra compression attempt.  It still compresses
full-page images as before.

Assembly skips per-FPI compression while whole-record compression is
attempted, and reassembles the record if that does not pay off.  Setting
wal_compression_threshold above the largest possible record restores the
previous behaviour.

The per-block compressed_page arrays make way for two buffers of the same
maximum size, one staging the record and one taking the compressor
output.  XLR_COMPRESSED in xl_info marks a compressed record, which
begins with an XLogCompressionHeader giving the method and the
decompressed length.

Author: Andrey Borodin
Discussion: https://postgr.es/m/4DC38068-976E-4A84-8EE6-4EFACBBD927A@yandex-team.ru
---
 doc/src/sgml/config.sgml                      |  38 +-
 src/backend/access/transam/xlog.c             |  19 +-
 src/backend/access/transam/xloginsert.c       | 330 ++++++++++++++++--
 src/backend/access/transam/xlogreader.c       | 251 +++++++++++--
 src/backend/utils/misc/guc_parameters.dat     |  18 +-
 src/backend/utils/misc/postgresql.conf.sample |   5 +-
 src/include/access/xlog.h                     |   1 +
 src/include/access/xlogreader.h               |   4 +
 src/include/access/xlogrecord.h               |  48 ++-
 src/include/utils/guc_hooks.h                 |   1 +
 src/test/perl/PostgreSQL/Test/Cluster.pm      |  20 +-
 .../recovery/t/026_overwrite_contrecord.pl    |   4 +-
 src/test/recovery/t/052_wal_compression.pl    | 152 ++++++++
 src/tools/pgindent/typedefs.list              |   1 +
 14 files changed, 822 insertions(+), 70 deletions(-)
 create mode 100644 src/test/recovery/t/052_wal_compression.pl

diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index aa7b1bd75d2..ff79ed0a2ec 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -3665,8 +3665,11 @@ include_dir 'conf.d'
         When enabled, the <productname>PostgreSQL</productname>
         server compresses full page images written to WAL (e.g. when
         <xref linkend="guc-full-page-writes"/> is on, during a base backup,
-        etc.).
-        A compressed page image will be decompressed during WAL replay.
+        etc.).  With <literal>lz4</literal> or <literal>zstd</literal>, a
+        record larger than <xref linkend="guc-wal-compression-threshold"/>
+        is instead compressed as a whole, which usually does better when
+        one record carries several page images.
+        Compressed data will be decompressed during WAL replay.
         The supported methods are <literal>pglz</literal>,
         <literal>lz4</literal> (if <productname>PostgreSQL</productname>
         was compiled with <option>--with-lz4</option>) and
@@ -3687,6 +3690,37 @@ include_dir 'conf.d'
       </listitem>
      </varlistentry>
 
+     <varlistentry id="guc-wal-compression-threshold" xreflabel="wal_compression_threshold">
+      <term><varname>wal_compression_threshold</varname> (<type>integer</type>)
+      <indexterm>
+       <primary><varname>wal_compression_threshold</varname> configuration parameter</primary>
+      </indexterm>
+      </term>
+      <listitem>
+       <para>
+        Smallest WAL record that <xref linkend="guc-wal-compression"/>
+        compresses as a whole rather than compressing each of its full page
+        images separately.  If this value is specified without units, it is
+        taken as bytes.  The default is 512 bytes.  Only records compressed
+        with <literal>lz4</literal> or <literal>zstd</literal> are affected;
+        <literal>pglz</literal> always compresses page images individually.
+        Only superusers and users with the appropriate <literal>SET</literal>
+        privilege can change this setting.
+       </para>
+
+       <para>
+        Lowering this value does not simply compress more.  A record taken by
+        whole-record compression no longer has its page images compressed
+        individually, and carries a slightly larger header, so a threshold
+        below the default can produce <emphasis>more</emphasis> WAL rather
+        than less.  Setting it above the size of the largest possible WAL
+        record disables whole-record compression entirely, leaving the
+        page-image compression that <xref linkend="guc-wal-compression"/>
+        performed before this setting existed.
+       </para>
+      </listitem>
+     </varlistentry>
+
      <varlistentry id="guc-wal-init-zero" xreflabel="wal_init_zero">
       <term><varname>wal_init_zero</varname> (<type>boolean</type>)
       <indexterm>
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index f8b939853e9..eb1d1fddb77 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -142,6 +142,7 @@ int			wal_retrieve_retry_interval = 5000;
 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;
 
 #ifdef WAL_DEBUG
 bool		XLOG_DEBUG = false;
@@ -751,6 +752,22 @@ static void WALInsertLockUpdateInsertingAt(XLogRecPtr insertingAt);
 
 static void XLogChecksums(uint32 new_type);
 
+#ifdef WAL_DEBUG
+/* Read length of a record, accounting for possible compression */
+static uint32
+XLogGetRecordTotalLen(XLogRecord *record)
+{
+	if (record->xl_info & XLR_COMPRESSED)
+	{
+		XLogCompressionHeader *c = (XLogCompressionHeader *) record;
+
+		Assert(c->decompressed_length > 0);
+		return c->decompressed_length;
+	}
+	return record->xl_tot_len;
+}
+#endif
+
 /*
  * 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
@@ -1068,7 +1085,7 @@ XLogInsertRecord(XLogRecData *rdata,
 		/* We also need temporary space to decode the record. */
 		record = (XLogRecord *) recordBuf.data;
 		decoded = (DecodedXLogRecord *)
-			palloc(DecodeXLogRecordRequiredSpace(record->xl_tot_len));
+			palloc(DecodeXLogRecordRequiredSpace(XLogGetRecordTotalLen(record)));
 
 		if (!debug_reader)
 			debug_reader = XLogReaderAllocate(wal_segment_size, NULL,
diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index d752eb8a762..a5a3a1422cb 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -39,6 +39,7 @@
 #include "replication/origin.h"
 #include "storage/bufmgr.h"
 #include "storage/proc.h"
+#include "utils/guc_hooks.h"
 #include "utils/memutils.h"
 #include "utils/pgstat_internal.h"
 #include "utils/rel.h"
@@ -85,8 +86,8 @@ typedef struct
 	XLogRecData bkp_rdatas[2];	/* temporary rdatas used to hold references to
 								 * backup block data in XLogRecordAssemble() */
 
-	/* buffer to store a compressed version of backup block image */
-	char		compressed_page[COMPRESS_BUFSIZE];
+	/* pointer into compression_buf for compressed page image */
+	char	   *compressed_page;
 } registered_buffer;
 
 static registered_buffer *registered_buffers;
@@ -128,6 +129,15 @@ static char *hdr_scratch = NULL;
  * from here.
  */
 static ZSTD_CCtx *zstd_cctx = NULL;
+
+/* Get this backend's zstd context, creating it on first use.  NULL on OOM. */
+static ZSTD_CCtx *
+GetZstdCCtx(void)
+{
+	if (zstd_cctx == NULL)
+		zstd_cctx = ZSTD_createCCtx();
+	return zstd_cctx;
+}
 #endif
 
 #define SizeOfXlogOrigin	(sizeof(ReplOriginId) + sizeof(char))
@@ -139,6 +149,28 @@ static ZSTD_CCtx *zstd_cctx = NULL;
 	 SizeOfXLogRecordDataHeaderLong + SizeOfXlogOrigin + \
 	 SizeOfXLogTransactionId)
 
+/*
+ * Size of the compression buffers: enough for a compressed image of every
+ * block a record may reference, plus the record headers.  That is the largest
+ * record we can be asked to build, so a record that does not fit is not
+ * compressed as a whole.
+ */
+#define WAL_COMPRESSION_BUFSIZE \
+	((XLR_MAX_BLOCK_ID + 1) * COMPRESS_BUFSIZE + HEADER_SCRATCH_SIZE)
+
+/*
+ * Compression buffers, allocated together when wal_compression is first
+ * enabled.  compression_buf serves two uses that never overlap: per-FPI
+ * compression packs block images into it, tracked by compression_buf_offset,
+ * while whole-record compression flattens the rdt chain into it and compresses
+ * into compressed_data.
+ */
+static char *compression_buf = NULL;
+static int	compression_buf_offset; /* fill level for FPI packing */
+#ifdef WAL_WHOLE_RECORD_COMPRESSION
+static char *compressed_data = NULL;
+#endif
+
 /*
  * An array of XLogRecData structs, to hold registered data.
  */
@@ -151,11 +183,13 @@ static bool begininsert_called = false;
 /* Memory context to hold the registered buffer and data references. */
 static MemoryContext xloginsert_cxt;
 
+static void AllocCompressionBuffers(void);
 static XLogRecData *XLogRecordAssemble(RmgrId rmid, uint8 info,
 									   XLogRecPtr RedoRecPtr, bool doPageWrites,
 									   XLogRecPtr *fpw_lsn, int *num_fpi,
 									   uint64 *fpi_bytes,
-									   bool *topxid_included);
+									   bool *topxid_included, uint64 *rec_size,
+									   bool skip_fpi_compression);
 static bool XLogCompressBackupBlock(const PageData *page, uint16 hole_offset,
 									uint16 hole_length, void *dest, uint16 *dlen);
 
@@ -481,6 +515,142 @@ XLogSetRecordFlags(uint8 flags)
 	curinsert_flags |= flags;
 }
 
+/* Compress the assembled record; NULL if that did not pay off */
+static XLogRecData *
+XLogCompressRdt(XLogRecData *rdt)
+{
+#ifndef WAL_WHOLE_RECORD_COMPRESSION
+	/* Without a supported method the caller never gets here */
+	Assert(false);
+	return NULL;
+#else
+	static XLogRecData compressed_rdt_hdr;
+	XLogCompressionHeader *compressed_header;
+	XLogRecord *src_header;
+	uint32		flat_len = 0;
+	uint32		orig_len;
+	int32		compr_len = -1;
+	int32		dest_size;
+
+	Assert(wal_compression != WAL_COMPRESSION_NONE);
+	Assert(compression_buf != NULL);
+
+	/*
+	 * Flatten the rdt chain into compression_buf.  The caller skipped per-FPI
+	 * compression, so the buffer is free, and checked that the record fits.
+	 */
+	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;
+
+	orig_len = src_header->xl_tot_len - SizeOfXLogRecord;
+
+	/*
+	 * compressed_data is not sized to the worst-case expansion bound, so hand
+	 * the compressor what is really left and let it fail if that is too
+	 * little.  A record that does not fit was not worth compressing anyway.
+	 */
+	dest_size = WAL_COMPRESSION_BUFSIZE - SizeOfXLogCompressedRecord;
+
+	switch ((WalCompression) wal_compression)
+	{
+		case WAL_COMPRESSION_NONE:
+		case WAL_COMPRESSION_PGLZ:
+			/* caller does not use whole-record compression for these */
+			Assert(false);
+			return NULL;
+
+		case WAL_COMPRESSION_LZ4:
+#ifdef USE_LZ4
+			compressed_header->method = XLR_COMPRESS_LZ4;
+			compr_len = LZ4_compress_default((char *) &src_header[1],
+											 (char *) &compressed_header[1],
+											 orig_len, dest_size);
+			if (compr_len <= 0)
+				return NULL;
+#else
+			elog(ERROR, "LZ4 is not supported by this build");
+#endif
+			break;
+
+		case WAL_COMPRESSION_ZSTD:
+#ifdef USE_ZSTD
+			{
+				size_t		zstd_len;
+
+				ZSTD_CCtx  *cctx = GetZstdCCtx();
+
+				if (cctx == NULL)
+					return NULL;
+
+				compressed_header->method = XLR_COMPRESS_ZSTD;
+				zstd_len = ZSTD_compressCCtx(cctx,
+											 (char *) &compressed_header[1],
+											 dest_size,
+											 (char *) &src_header[1], orig_len,
+											 ZSTD_CLEVEL_DEFAULT);
+				if (ZSTD_isError(zstd_len))
+					return NULL;
+				compr_len = (int32) zstd_len;
+			}
+#else
+			elog(ERROR, "zstd is not supported by this build");
+#endif
+			break;
+
+			/* no default case, so that compiler will warn */
+	}
+
+	Assert(compr_len > 0 && compr_len <= dest_size);
+
+	compressed_header->record_header.xl_tot_len =
+		SizeOfXLogCompressedRecord + compr_len;
+
+	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
+}
+
+/* Checksum assembled record (which may be compressed). */
+static void
+XLogChecksumRecord(XLogRecData *rdt)
+{
+	pg_crc32c	rdata_crc;
+	XLogRecord *rechdr = (XLogRecord *) rdt->data;
+
+	/*
+	 * Calculate CRC of the data
+	 *
+	 * Note that the record header isn't added into the CRC initially since we
+	 * don't know the prev-link yet.  Thus, the CRC will represent the CRC of
+	 * the whole record in the order: rdata, then backup blocks, then record
+	 * header.
+	 */
+	INIT_CRC32C(rdata_crc);
+	COMP_CRC32C(rdata_crc, ((char *) rdt->data) + SizeOfXLogRecord,
+				rdt->len - SizeOfXLogRecord);
+	for (rdt = rdt->next; rdt != NULL; rdt = rdt->next)
+		COMP_CRC32C(rdata_crc, rdt->data, rdt->len);
+	rechdr->xl_crc = rdata_crc;
+}
+
 /*
  * Insert an XLOG record having the specified RMID and info bytes, with the
  * body of the record being the data and buffer references registered earlier
@@ -497,6 +667,17 @@ XLogInsert(RmgrId rmid, uint8 info)
 {
 	XLogRecPtr	EndPos;
 
+	/*
+	 * Only lz4 and zstd compress whole records; pglz rarely beats its own
+	 * minimum compression rate on a record made of full-page images.  A
+	 * threshold at or above the buffer size disables this entirely, which is
+	 * how a user asks for full-page-image-only compression.
+	 */
+	bool		try_whole_record = ((wal_compression == WAL_COMPRESSION_LZ4 ||
+									 wal_compression == WAL_COMPRESSION_ZSTD) &&
+									wal_compression_threshold <
+									WAL_COMPRESSION_BUFSIZE);
+
 	/* XLogBeginInsert() must have been called. */
 	if (!begininsert_called)
 		elog(ERROR, "XLogBeginInsert was not called");
@@ -532,6 +713,10 @@ XLogInsert(RmgrId rmid, uint8 info)
 		XLogRecData *rdt;
 		int			num_fpi = 0;
 		uint64		fpi_bytes = 0;
+		uint64		rec_size;
+
+		/* Do not accumulate FPI data across retries of this loop */
+		compression_buf_offset = 0;
 
 		/*
 		 * Get values needed to decide whether to do full-page writes. Since
@@ -542,7 +727,41 @@ XLogInsert(RmgrId rmid, uint8 info)
 
 		rdt = XLogRecordAssemble(rmid, info, RedoRecPtr, doPageWrites,
 								 &fpw_lsn, &num_fpi, &fpi_bytes,
-								 &topxid_included);
+								 &topxid_included, &rec_size,
+								 try_whole_record);
+
+		/*
+		 * Assembly skipped per-FPI compression on our behalf, so if the
+		 * record is not compressed as a whole after all, reassemble it to get
+		 * the per-FPI compression back.
+		 */
+		if (try_whole_record)
+		{
+			bool		whole_record_compressed = false;
+
+			if (rec_size > wal_compression_threshold &&
+				rec_size <= WAL_COMPRESSION_BUFSIZE)
+			{
+				XLogRecData *rdt_compressed = XLogCompressRdt(rdt);
+
+				if (rdt_compressed != NULL)
+				{
+					rdt = rdt_compressed;
+					whole_record_compressed = true;
+				}
+			}
+
+			if (!whole_record_compressed && 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);
@@ -616,6 +835,44 @@ XLogGetFakeLSN(Relation rel)
 	}
 }
 
+/*
+ * Allocate the compression buffers, once per backend.
+ *
+ * Callers must be outside a critical section, which is why this happens at
+ * backend start or in the wal_compression assign hook, never in XLogInsert().
+ * The buffers are kept when compression is switched back off, so that turning
+ * it on again cannot need an allocation.
+ */
+static void
+AllocCompressionBuffers(void)
+{
+	Assert(CritSectionCount == 0);
+	Assert(xloginsert_cxt != NULL);
+
+	if (compression_buf != NULL)
+		return;					/* already done */
+
+	compression_buf = MemoryContextAlloc(xloginsert_cxt,
+										 WAL_COMPRESSION_BUFSIZE);
+#ifdef WAL_WHOLE_RECORD_COMPRESSION
+	compressed_data = MemoryContextAlloc(xloginsert_cxt,
+										 WAL_COMPRESSION_BUFSIZE);
+#endif
+}
+
+/*
+ * GUC assign hook for wal_compression.
+ *
+ * Runs before wal_compression itself is updated, hence the test on newval.  If
+ * xloginsert_cxt does not exist yet, InitXLogInsert() will do the allocation.
+ */
+void
+assign_wal_compression(int newval, void *extra)
+{
+	if (newval != WAL_COMPRESSION_NONE && xloginsert_cxt != NULL)
+		AllocCompressionBuffers();
+}
+
 /*
  * Assemble a WAL record from the registered data and buffers into an
  * XLogRecData chain, ready for insertion with XLogInsertRecord().
@@ -635,12 +892,11 @@ static XLogRecData *
 XLogRecordAssemble(RmgrId rmid, uint8 info,
 				   XLogRecPtr RedoRecPtr, bool doPageWrites,
 				   XLogRecPtr *fpw_lsn, int *num_fpi, uint64 *fpi_bytes,
-				   bool *topxid_included)
+				   bool *topxid_included, uint64 *rec_size,
+				   bool skip_fpi_compression)
 {
-	XLogRecData *rdt;
 	uint64		total_len = 0;
 	int			block_id;
-	pg_crc32c	rdata_crc;
 	registered_buffer *prev_regbuf = NULL;
 	XLogRecData *rdt_datas_last;
 	XLogRecord *rechdr;
@@ -668,6 +924,9 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 	if (wal_consistency_checking[rmid])
 		info |= XLR_CHECK_CONSISTENCY;
 
+	/* We are often in a critical section here, so we cannot allocate */
+	Assert(wal_compression == WAL_COMPRESSION_NONE || compression_buf != NULL);
+
 	/*
 	 * Make an rdata chain containing all the data portions of all block
 	 * references. This includes the data for full-page images. Also append
@@ -771,15 +1030,26 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 			}
 
 			/*
-			 * Try to compress a block image if wal_compression is enabled
+			 * Try to compress a block image if wal_compression is enabled and
+			 * the caller has not reserved the buffer for whole-record
+			 * compression.  Each image gets its own slice of the buffer,
+			 * which is large enough for all of them.
 			 */
-			if (wal_compression != WAL_COMPRESSION_NONE)
+			if (!skip_fpi_compression &&
+				wal_compression != WAL_COMPRESSION_NONE)
 			{
+				Assert(compression_buf_offset + COMPRESS_BUFSIZE <=
+					   WAL_COMPRESSION_BUFSIZE);
+				regbuf->compressed_page = compression_buf + compression_buf_offset;
+
 				is_compressed =
 					XLogCompressBackupBlock(page, bimg.hole_offset,
 											cbimg.hole_length,
 											regbuf->compressed_page,
 											&compressed_len);
+
+				if (is_compressed)
+					compression_buf_offset += compressed_len;
 			}
 
 			/*
@@ -983,19 +1253,6 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 	hdr_rdt.len = (scratch - hdr_scratch);
 	total_len += hdr_rdt.len;
 
-	/*
-	 * Calculate CRC of the data
-	 *
-	 * Note that the record header isn't added into the CRC initially since we
-	 * don't know the prev-link yet.  Thus, the CRC will represent the CRC of
-	 * the whole record in the order: rdata, then backup blocks, then record
-	 * header.
-	 */
-	INIT_CRC32C(rdata_crc);
-	COMP_CRC32C(rdata_crc, hdr_scratch + SizeOfXLogRecord, hdr_rdt.len - SizeOfXLogRecord);
-	for (rdt = hdr_rdt.next; rdt != NULL; rdt = rdt->next)
-		COMP_CRC32C(rdata_crc, rdt->data, rdt->len);
-
 	/*
 	 * Ensure that the XLogRecord is not too large.
 	 *
@@ -1019,7 +1276,9 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 	rechdr->xl_info = info;
 	rechdr->xl_rmid = rmid;
 	rechdr->xl_prev = InvalidXLogRecPtr;
-	rechdr->xl_crc = rdata_crc;
+	rechdr->xl_crc = 0;
+
+	*rec_size = rechdr->xl_tot_len;
 
 	return &hdr_rdt;
 }
@@ -1078,17 +1337,19 @@ XLogCompressBackupBlock(const PageData *page, uint16 hole_offset, uint16 hole_le
 
 		case WAL_COMPRESSION_ZSTD:
 #ifdef USE_ZSTD
-			if (zstd_cctx == NULL)
-				zstd_cctx = ZSTD_createCCtx();
-
-			if (zstd_cctx == NULL)
-				len = -1;		/* out of memory; store the image as is */
-			else
 			{
-				len = ZSTD_compressCCtx(zstd_cctx, dest, COMPRESS_BUFSIZE,
-										source, orig_len, ZSTD_CLEVEL_DEFAULT);
-				if (ZSTD_isError(len))
-					len = -1;	/* failure */
+				ZSTD_CCtx  *cctx = GetZstdCCtx();
+
+				if (cctx == NULL)
+					len = -1;	/* out of memory; store the image as is */
+				else
+				{
+					len = ZSTD_compressCCtx(cctx, dest, COMPRESS_BUFSIZE,
+											source, orig_len,
+											ZSTD_CLEVEL_DEFAULT);
+					if (ZSTD_isError(len))
+						len = -1;	/* failure */
+				}
 			}
 #else
 			elog(ERROR, "zstd is not supported by this build");
@@ -1460,4 +1721,7 @@ InitXLogInsert(void)
 	if (hdr_scratch == NULL)
 		hdr_scratch = MemoryContextAllocZero(xloginsert_cxt,
 											 HEADER_SCRATCH_SIZE);
+
+	if (wal_compression != WAL_COMPRESSION_NONE)
+		AllocCompressionBuffers();
 }
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
index 06479ba9c4e..78227d9547a 100644
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -32,6 +32,7 @@
 #include "catalog/pg_control.h"
 #include "common/pg_lzcompress.h"
 #include "replication/origin.h"
+#include "utils/memutils.h"
 
 #ifndef FRONTEND
 #include "pgstat.h"
@@ -55,6 +56,9 @@ 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 XLogRecord *XLogDecompressRecordIfNeeded(XLogReaderState *state,
+												XLogRecord *record,
+												XLogRecPtr recptr);
 
 /* size of the buffer allocated for error message. */
 #define MAX_ERRORMSG_LEN 1000
@@ -175,6 +179,8 @@ XLogReaderFree(XLogReaderState *state)
 	if (state->fpi_dctx)
 		ZSTD_freeDCtx((ZSTD_DCtx *) state->fpi_dctx);
 #endif
+	if (state->decompression_buffer)
+		pfree(state->decompression_buffer);
 	pfree(state->readBuf);
 	pfree(state);
 }
@@ -538,10 +544,12 @@ XLogDecodeNextRecord(XLogReaderState *state, bool nonblocking)
 	XLogRecPtr	targetPagePtr;
 	bool		randAccess;
 	uint32		len,
-				total_len;
+				total_len_decomp,
+				total_len_physical;
 	uint32		targetRecOff;
 	uint32		pageHeaderSize;
 	bool		assembled;
+	bool		have_decomp_len;
 	bool		gotheader;
 	int			readOff;
 	DecodedXLogRecord *decoded;
@@ -592,6 +600,14 @@ restart:
 	state->currRecPtr = RecPtr;
 	assembled = false;
 
+	/*
+	 * Start each pass without a decode slot.  We do not always take one below
+	 * - a compressed record whose header straddles a page does not tell us
+	 * how much space it needs yet - and carrying one over from a previous pass
+	 * would hand out a slot that is already in the decode queue.
+	 */
+	decoded = NULL;
+
 	targetPagePtr = RecPtr - (RecPtr % XLOG_BLCKSZ);
 	targetRecOff = RecPtr % XLOG_BLCKSZ;
 
@@ -649,7 +665,70 @@ restart:
 	 * whole header.
 	 */
 	record = (XLogRecord *) (state->readBuf + RecPtr % XLOG_BLCKSZ);
-	total_len = record->xl_tot_len;
+	total_len_physical = record->xl_tot_len;
+
+	/*
+	 * Size the record will occupy once decoded, which for a compressed record
+	 * is its decompressed length rather than xl_tot_len.
+	 *
+	 * Only xl_tot_len is guaranteed to be on this page; xl_info may live on
+	 * the next one when the record starts near the end of a page.  If we
+	 * cannot tell yet, have_decomp_len stays false and we allocate after
+	 * assembly instead.
+	 */
+	total_len_decomp = 0;
+	have_decomp_len = false;
+
+	if (targetRecOff <= XLOG_BLCKSZ - SizeOfXLogRecord)
+	{
+		/* Full header is on this page; safe to read xl_info. */
+		if (record->xl_info & XLR_COMPRESSED)
+		{
+			/*
+			 * Skip if the compression header spans pages, or if the record is
+			 * too short to be a valid compressed one.
+			 */
+			if (targetRecOff <= XLOG_BLCKSZ - SizeOfXLogCompressedRecord &&
+				total_len_physical >= SizeOfXLogCompressedRecord)
+			{
+				XLogCompressionHeader *c = (XLogCompressionHeader *) record;
+				uint32		dlen = c->decompressed_length;
+				bool		valid_method;
+
+				valid_method = false
+#ifdef USE_LZ4
+					|| (c->method == XLR_COMPRESS_LZ4)
+#endif
+#ifdef USE_ZSTD
+					|| (c->method == XLR_COMPRESS_ZSTD)
+#endif
+					;
+
+				/* Sanity-check before using for allocation */
+				if (dlen > SizeOfXLogRecord && dlen <= MaxAllocSize &&
+					valid_method)
+				{
+					total_len_decomp = dlen;
+					have_decomp_len = true;
+				}
+			}
+		}
+		else
+		{
+			total_len_decomp = record->xl_tot_len;
+			have_decomp_len = true;
+		}
+	}
+	else
+	{
+		/*
+		 * Header spans pages, so we cannot read xl_info and do not know
+		 * whether this is a compressed record, let alone how big it decodes
+		 * to.  xl_tot_len would be the compressed size, and reserving that
+		 * much would leave the decoder writing past its slot, so leave the
+		 * allocation to the pass after assembly.
+		 */
+	}
 
 	/*
 	 * If the whole record header is on this page, validate it immediately.
@@ -665,16 +744,18 @@ restart:
 								   randAccess))
 			goto err;
 		gotheader = true;
+		if (record->xl_info & XLR_COMPRESSED)
+			gotheader = targetRecOff <= XLOG_BLCKSZ - SizeOfXLogCompressedRecord;
 	}
 	else
 	{
 		/* There may be no next page if it's too small. */
-		if (total_len < SizeOfXLogRecord)
+		if (total_len_physical < SizeOfXLogRecord)
 		{
 			report_invalid_record(state,
 								  "invalid record length at %X/%08X: expected at least %u, got %u",
 								  LSN_FORMAT_ARGS(RecPtr),
-								  (uint32) SizeOfXLogRecord, total_len);
+								  (uint32) SizeOfXLogRecord, total_len_physical);
 			goto err;
 		}
 		/* We'll validate the header once we have the next page. */
@@ -686,21 +767,24 @@ restart:
 	 * calling palloc.  If we can't, we'll try again below after we've
 	 * validated that total_len isn't garbage bytes from a recycled WAL page.
 	 */
-	decoded = XLogReadRecordAlloc(state,
-								  total_len,
-								  false /* allow_oversized */ );
+	if (have_decomp_len)
+		decoded = XLogReadRecordAlloc(state,
+									  total_len_decomp,
+									  false /* allow_oversized */ );
+
 	if (decoded == NULL && nonblocking)
 	{
 		/*
-		 * There is no space in the circular decode buffer, and the caller is
-		 * only reading ahead.  The caller should consume existing records to
-		 * make space.
+		 * Either there is no space in the circular decode buffer, or we could
+		 * not tell how much this record needs.  Either way the caller is only
+		 * reading ahead, so let it replay what it has; it will come back for
+		 * this record with nonblocking off, where we may allocate.
 		 */
 		return XLREAD_WOULDBLOCK;
 	}
 
 	len = XLOG_BLCKSZ - RecPtr % XLOG_BLCKSZ;
-	if (total_len > len)
+	if (total_len_physical > len)
 	{
 		/* Need to reassemble record */
 		char	   *contdata;
@@ -772,19 +856,19 @@ restart:
 			 * we expect there to be left.
 			 */
 			if (pageHeader->xlp_rem_len == 0 ||
-				total_len != (pageHeader->xlp_rem_len + gotlen))
+				total_len_physical != (pageHeader->xlp_rem_len + gotlen))
 			{
 				report_invalid_record(state,
 									  "invalid contrecord length %u (expected %lld) at %X/%08X",
 									  pageHeader->xlp_rem_len,
-									  ((long long) total_len) - gotlen,
+									  ((long long) total_len_physical) - gotlen,
 									  LSN_FORMAT_ARGS(RecPtr));
 				goto err;
 			}
 
 			/* Wait for the next page to become available */
 			readOff = ReadPageInternal(state, targetPagePtr,
-									   Min(total_len - gotlen + SizeOfXLogShortPHD,
+									   Min(total_len_physical - gotlen + SizeOfXLogShortPHD,
 										   XLOG_BLCKSZ));
 			if (readOff == XLREAD_WOULDBLOCK)
 				return XLREAD_WOULDBLOCK;
@@ -829,7 +913,7 @@ restart:
 			 * also cross-checked total_len against xlp_rem_len on the second
 			 * page, and verified xlp_pageaddr on both.
 			 */
-			if (total_len > state->readRecordBufSize)
+			if (total_len_physical > state->readRecordBufSize)
 			{
 				char		save_copy[XLOG_BLCKSZ * 2];
 
@@ -840,11 +924,11 @@ restart:
 				Assert(gotlen <= lengthof(save_copy));
 				Assert(gotlen <= state->readRecordBufSize);
 				memcpy(save_copy, state->readRecordBuf, gotlen);
-				allocate_recordbuf(state, total_len);
+				allocate_recordbuf(state, total_len_physical);
 				memcpy(state->readRecordBuf, save_copy, gotlen);
 				buffer = state->readRecordBuf + gotlen;
 			}
-		} while (gotlen < total_len);
+		} while (gotlen < total_len_physical);
 		Assert(gotheader);
 
 		record = (XLogRecord *) state->readRecordBuf;
@@ -859,8 +943,9 @@ restart:
 	else
 	{
 		/* Wait for the record data to become available */
+		Assert(targetRecOff + total_len_physical <= XLOG_BLCKSZ);
 		readOff = ReadPageInternal(state, targetPagePtr,
-								   Min(targetRecOff + total_len, XLOG_BLCKSZ));
+								   targetRecOff + total_len_physical);
 		if (readOff == XLREAD_WOULDBLOCK)
 			return XLREAD_WOULDBLOCK;
 		else if (readOff < 0)
@@ -870,7 +955,7 @@ restart:
 		if (!ValidXLogRecord(state, record, RecPtr))
 			goto err;
 
-		state->NextRecPtr = RecPtr + MAXALIGN(total_len);
+		state->NextRecPtr = RecPtr + MAXALIGN(total_len_physical);
 
 		state->DecodeRecPtr = RecPtr;
 	}
@@ -893,8 +978,20 @@ restart:
 	if (decoded == NULL)
 	{
 		Assert(!nonblocking);
+
+		/* total_len_decomp may not yet reflect the actual decompressed size */
+		if (record->xl_info & XLR_COMPRESSED)
+		{
+			XLogCompressionHeader *c = (XLogCompressionHeader *) record;
+
+			Assert(c->decompressed_length > 0);
+			Assert(c->decompressed_length < MaxAllocSize);
+			total_len_decomp = c->decompressed_length;
+		}
+		else
+			total_len_decomp = record->xl_tot_len;
 		decoded = XLogReadRecordAlloc(state,
-									  total_len,
+									  total_len_decomp,
 									  true /* allow_oversized */ );
 		/* allocation should always happen under allow_oversized */
 		Assert(decoded != NULL);
@@ -1691,6 +1788,108 @@ DecodeXLogRecordRequiredSpace(size_t xl_tot_len)
 	return size;
 }
 
+static XLogRecord *
+XLogDecompressRecordIfNeeded(XLogReaderState *state,
+							 XLogRecord *record,
+							 XLogRecPtr recptr)
+{
+	if (record->xl_info & XLR_COMPRESSED)
+	{
+#ifndef WAL_WHOLE_RECORD_COMPRESSION
+		report_invalid_record(state,
+							  "could not decompress record at %X/%08X compressed with method %u not supported by build",
+							  LSN_FORMAT_ARGS(recptr),
+							  ((XLogCompressionHeader *) record)->method);
+		return NULL;
+#else
+		XLogCompressionHeader *src = (XLogCompressionHeader *) record;
+
+		/* decompressed_length covers the XLogRecord header + body */
+		uint32		body_len = src->decompressed_length - SizeOfXLogRecord;
+		uint32		srclen = src->record_header.xl_tot_len - SizeOfXLogCompressedRecord;
+		bool		decomp_success = true;
+		char	   *dst;
+		XLogRecord *dst_h;
+
+		/*
+		 * Grow the decompression buffer if needed, rounding up to BLCKSZ to
+		 * avoid frequent small reallocations.  Since the buffer content is
+		 * always fully overwritten, we simply pfree and reallocate.
+		 */
+		if (state->decompression_buffer_size < src->decompressed_length)
+		{
+			uint32		new_size = (uint32) TYPEALIGN(BLCKSZ, src->decompressed_length);
+
+			if (state->decompression_buffer)
+				pfree(state->decompression_buffer);
+			state->decompression_buffer =
+				palloc_extended(new_size, MCXT_ALLOC_NO_OOM);
+			if (!state->decompression_buffer)
+			{
+				state->decompression_buffer_size = 0;
+				report_invalid_record(state,
+									  "out of memory while decompressing record at %X/%08X",
+									  LSN_FORMAT_ARGS(recptr));
+				return NULL;
+			}
+			state->decompression_buffer_size = new_size;
+		}
+
+		dst_h = (XLogRecord *) state->decompression_buffer;
+		*dst_h = src->record_header;
+		dst_h->xl_tot_len = src->decompressed_length;
+		dst = (char *) &dst_h[1];
+
+		if (src->method == XLR_COMPRESS_LZ4)
+		{
+#ifdef USE_LZ4
+			if (LZ4_decompress_safe((char *) &src[1], dst,
+									srclen, body_len) <= 0)
+				decomp_success = false;
+#else
+			report_invalid_record(state,
+								  "could not decompress record at %X/%08X compressed with %s not supported by build",
+								  LSN_FORMAT_ARGS((XLogRecPtr) recptr), "lz4");
+			return NULL;
+#endif
+		}
+		else if (src->method == XLR_COMPRESS_ZSTD)
+		{
+#ifdef USE_ZSTD
+			size_t		decomp_result = ZSTD_decompress(dst, body_len,
+														(char *) &src[1], srclen);
+
+			if (ZSTD_isError(decomp_result))
+				decomp_success = false;
+#else
+			report_invalid_record(state,
+								  "could not decompress record at %X/%08X compressed with %s not supported by build",
+								  LSN_FORMAT_ARGS((XLogRecPtr) recptr), "zstd");
+			return NULL;
+#endif
+		}
+		else
+		{
+			report_invalid_record(state,
+								  "could not decompress record at %X/%08X compressed with unknown method",
+								  LSN_FORMAT_ARGS((XLogRecPtr) recptr));
+			return NULL;
+		}
+
+		if (!decomp_success)
+		{
+			report_invalid_record(state,
+								  "could not decompress record at %X/%08X",
+								  LSN_FORMAT_ARGS(recptr));
+			return NULL;
+		}
+
+		return (XLogRecord *) state->decompression_buffer;
+#endif
+	}
+	return record;
+}
+
 /*
  * Decode a record.  "decoded" must point to a MAXALIGNed memory area that has
  * space for at least DecodeXLogRecordRequiredSpace(record) bytes.  On
@@ -1729,6 +1928,14 @@ DecodeXLogRecord(XLogReaderState *state,
 	RelFileLocator *rlocator = NULL;
 	uint8		block_id;
 
+	record = XLogDecompressRecordIfNeeded(state, record, lsn);
+
+	if (!record)
+	{
+		/* Decompression failed, error must be reported already */
+		return false;
+	}
+
 	decoded->header = *record;
 	decoded->lsn = lsn;
 	decoded->next = NULL;
@@ -1903,8 +2110,8 @@ DecodeXLogRecord(XLogReaderState *state,
 					blk->bimg_len != BLCKSZ)
 				{
 					report_invalid_record(state,
-										  "neither BKPIMAGE_HAS_HOLE nor BKPIMAGE_COMPRESSED set, but block image length is %d at %X/%08X",
-										  blk->data_len,
+										  "neither BKPIMAGE_HAS_HOLE nor BKPIMAGE_COMPRESSED set, but block image length is %u at %X/%08X",
+										  (unsigned int) blk->bimg_len,
 										  LSN_FORMAT_ARGS(state->ReadRecPtr));
 					goto err;
 				}
diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat
index adb72361ce0..28efd8f6e95 100644
--- a/src/backend/utils/misc/guc_parameters.dat
+++ b/src/backend/utils/misc/guc_parameters.dat
@@ -3458,10 +3458,26 @@
 },
 
 { name => 'wal_compression', type => 'enum', context => 'PGC_SUSET', group => 'WAL_SETTINGS',
-  short_desc => 'Compresses full-page writes written in WAL file with specified method.',
+  short_desc => 'Compresses WAL written with the specified method.',
+  long_desc => 'Full-page images are compressed individually. With lz4 or zstd, '
+             . 'a record larger than wal_compression_threshold is compressed as a whole instead.',
   variable => 'wal_compression',
   boot_val => 'WAL_COMPRESSION_NONE',
   options => 'wal_compression_options',
+  assign_hook => 'assign_wal_compression',
+},
+
+{ 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 '
+             . 'compressing their full-page images individually. This applies only when '
+             . 'wal_compression selects lz4 or zstd. Set above the size of the largest '
+             . 'possible WAL record to keep the full-page-image-only behavior.',
+  flags => 'GUC_UNIT_BYTE',
+  variable => 'wal_compression_threshold',
+  boot_val => '512',
+  min => '32',
+  max => 'INT_MAX',
 },
 
 { name => 'wal_consistency_checking', type => 'string', context => 'PGC_SUSET', group => 'DEVELOPER_OPTIONS',
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index 7958653077b..406b8fc52ca 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -261,8 +261,11 @@
 #full_page_writes = on                  # recover from partial page writes
 #wal_log_hints = off                    # also do full page writes of non-critical updates
                                         # (change requires restart)
-#wal_compression = off                  # enables compression of full-page writes;
+#wal_compression = off                  # enables compression of full-page writes
+                                        # and of whole WAL records;
                                         # 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_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 4dd98624204..7fda7eb1f23 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -58,6 +58,7 @@ extern PGDLLIMPORT int CommitSiblings;
 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 CheckPointSegments;
 
diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h
index 5916e4cf124..5c869d78be7 100644
--- a/src/include/access/xlogreader.h
+++ b/src/include/access/xlogreader.h
@@ -258,6 +258,10 @@ struct XLogReaderState
 	char	   *decode_buffer_head; /* data is read from the head */
 	char	   *decode_buffer_tail; /* new data is written at the tail */
 
+	/* Buffer for decompressing whole-record compressed WAL records */
+	char	   *decompression_buffer;
+	uint32		decompression_buffer_size;
+
 	/*
 	 * 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 e8999d3fe91..be662e6558f 100644
--- a/src/include/access/xlogrecord.h
+++ b/src/include/access/xlogrecord.h
@@ -48,7 +48,10 @@ typedef struct XLogRecord
 	/* 2 bytes of padding here, initialize to zero */
 	pg_crc32c	xl_crc;			/* CRC for this record */
 
-	/* XLogRecordBlockHeaders and XLogRecordDataHeader follow, no padding */
+	/*
+	 * XLogRecordBlockHeaders, XLogRecordDataHeader or compression header
+	 * follow, no padding
+	 */
 
 } XLogRecord;
 
@@ -90,6 +93,9 @@ typedef struct XLogRecord
  */
 #define XLR_CHECK_CONSISTENCY	0x02
 
+/* This bit in xl_info means the record is compressed */
+#define XLR_COMPRESSED	0x04
+
 /*
  * Header info for block data appended to an XLOG record.
  *
@@ -143,11 +149,6 @@ typedef struct XLogRecordBlockImageHeader
 	uint16		length;			/* number of page image bytes */
 	uint16		hole_offset;	/* number of bytes before "hole" */
 	uint8		bimg_info;		/* flag bits, see below */
-
-	/*
-	 * If BKPIMAGE_HAS_HOLE and BKPIMAGE_COMPRESSED(), an
-	 * XLogRecordBlockCompressHeader struct follows.
-	 */
 } XLogRecordBlockImageHeader;
 
 #define SizeOfXLogRecordBlockImageHeader	\
@@ -157,13 +158,13 @@ typedef struct XLogRecordBlockImageHeader
 #define BKPIMAGE_HAS_HOLE		0x01	/* page image has "hole" */
 #define BKPIMAGE_APPLY			0x02	/* page image should be restored
 										 * during replay */
-/* compression methods supported */
+/* Compression methods supported for FPI (stored in bimg_info) */
 #define BKPIMAGE_COMPRESS_PGLZ	0x04
 #define BKPIMAGE_COMPRESS_LZ4	0x08
 #define BKPIMAGE_COMPRESS_ZSTD	0x10
 
 #define	BKPIMAGE_COMPRESSED(info) \
-	((info & (BKPIMAGE_COMPRESS_PGLZ | BKPIMAGE_COMPRESS_LZ4 | \
+	(((info) & (BKPIMAGE_COMPRESS_PGLZ | BKPIMAGE_COMPRESS_LZ4 | \
 			  BKPIMAGE_COMPRESS_ZSTD)) != 0)
 
 /*
@@ -178,6 +179,37 @@ typedef struct XLogRecordBlockCompressHeader
 #define SizeOfXLogRecordBlockCompressHeader \
 	sizeof(XLogRecordBlockCompressHeader)
 
+/* compression methods supported for whole-record compression */
+#define XLR_COMPRESS_LZ4	0x08
+#define XLR_COMPRESS_ZSTD	0x10
+
+/*
+ * Whole-record compression is only offered for methods that earn it, so a
+ * build without any of them carries neither side of the machinery.  It must
+ * still recognize such a record when reading, to report it properly.
+ */
+#if defined(USE_LZ4) || defined(USE_ZSTD)
+#define WAL_WHOLE_RECORD_COMPRESSION 1
+#endif
+
+/*
+ * Header prepended to a whole-record compressed WAL record.
+ *
+ * The compressed payload follows immediately, at SizeOfXLogCompressedRecord.
+ * record_header.xl_tot_len covers this header plus that payload, so it is the
+ * on-disk size; decompressed_length is what the record expands back to,
+ * including its XLogRecord header.
+ */
+typedef struct XLogCompressionHeader
+{
+	XLogRecord	record_header;
+	uint32		decompressed_length;
+	uint8		method;			/* XLR_COMPRESS_* */
+	/* 3 bytes of padding here, initialize to zero */
+} XLogCompressionHeader;
+
+#define SizeOfXLogCompressedRecord	sizeof(XLogCompressionHeader)
+
 /*
  * 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/utils/guc_hooks.h b/src/include/utils/guc_hooks.h
index 6a76f8d5ed6..15825fa126c 100644
--- a/src/include/utils/guc_hooks.h
+++ b/src/include/utils/guc_hooks.h
@@ -172,6 +172,7 @@ extern const char *show_unix_socket_permissions(void);
 extern bool check_wal_buffers(int *newval, void **extra, GucSource source);
 extern bool check_wal_consistency_checking(char **newval, void **extra,
 										   GucSource source);
+extern void assign_wal_compression(int newval, void *extra);
 extern void assign_wal_consistency_checking(const char *newval, void *extra);
 extern bool check_wal_segment_size(int *newval, void **extra, GucSource source);
 extern void assign_wal_sync_method(int new_wal_sync_method, void *extra);
diff --git a/src/test/perl/PostgreSQL/Test/Cluster.pm b/src/test/perl/PostgreSQL/Test/Cluster.pm
index 3eae4cf6281..96aaa88f1ce 100644
--- a/src/test/perl/PostgreSQL/Test/Cluster.pm
+++ b/src/test/perl/PostgreSQL/Test/Cluster.pm
@@ -719,9 +719,21 @@ sub init
 	# TEMP_CONFIG.  Otherwise, print it before TEMP_CONFIG, thereby permitting
 	# overrides.  Settings that merely improve performance or ease debugging
 	# belong before TEMP_CONFIG.
+
 	print $conf PostgreSQL::Test::Utils::slurp_file($ENV{TEMP_CONFIG})
 	  if defined $ENV{TEMP_CONFIG};
 
+	# A test that reasons about exact WAL positions cannot tolerate
+	# whole-record compression: the record it emits would shrink by an amount
+	# it has no way to predict.  Such a test passes no_wal_compression => 1 to
+	# init(), and we put the record back beyond any possible record size.
+	#
+	# 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{allows_streaming})
 	{
 		if ($params{allows_streaming} eq "logical")
@@ -3191,10 +3203,16 @@ sub emit_wal
 {
 	my ($self, $size) = @_;
 
+	# Disable whole-record WAL compression for this emission.  Tests call
+	# emit_wal() to place an exact number of bytes in WAL (e.g. to reach a
+	# specific page offset); whole-record compression would shrink the record
+	# and break those position-sensitive calculations.
 	return int(
 		$self->safe_psql(
 			'postgres',
-			"SELECT pg_logical_emit_message(true, '', repeat('a', $size)) - '0/0'"
+			"SET wal_compression_threshold = 2147483647;
+			 SELECT pg_logical_emit_message(true, '', repeat('a', $size)) - '0/0';
+			 RESET wal_compression_threshold"
 		));
 }
 
diff --git a/src/test/recovery/t/026_overwrite_contrecord.pl b/src/test/recovery/t/026_overwrite_contrecord.pl
index 82567ca6bfc..47342bb8c98 100644
--- a/src/test/recovery/t/026_overwrite_contrecord.pl
+++ b/src/test/recovery/t/026_overwrite_contrecord.pl
@@ -16,7 +16,9 @@ use Test::More;
 # file and the standby promotes successfully.
 
 my $node = PostgreSQL::Test::Cluster->new('primary');
-$node->init(allows_streaming => 1);
+# This test emits a message of a chosen size and requires the resulting record
+# to cross a segment boundary, so the record must reach WAL at that size.
+$node->init(allows_streaming => 1, no_wal_compression => 1);
 # We need these settings for stability of WAL behavior.
 $node->append_conf(
 	'postgresql.conf', qq(
diff --git a/src/test/recovery/t/052_wal_compression.pl b/src/test/recovery/t/052_wal_compression.pl
new file mode 100644
index 00000000000..001772daf23
--- /dev/null
+++ b/src/test/recovery/t/052_wal_compression.pl
@@ -0,0 +1,152 @@
+# Copyright (c) 2025-2026, PostgreSQL Global Development Group
+#
+# Test whole-record WAL compression via wal_compression and
+# wal_compression_threshold.  Exercises compression during replication
+# (walsender path) and decompression during crash recovery (startup path).
+#
+
+use strict;
+use warnings FATAL => 'all';
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# Determine which compression methods are compiled in, preferring the
+# faster hardware-accelerated codecs first for quicker failure feedback.
+my @methods = ();
+push @methods, 'zstd' if check_pg_config('#define HAVE_LIBZSTD 1');
+push @methods, 'lz4'  if check_pg_config('#define HAVE_LIBLZ4 1');
+push @methods, 'pglz';
+
+# Test whole-record WAL compression with a specific method.
+#
+# Creates a primary with the method enabled and a low threshold so that
+# every non-trivial WAL record is compressed.  A streaming standby is
+# 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.
+# Emit one 256kB compressible logical message at the given threshold and
+# report how many bytes of WAL it consumed.
+sub wal_used
+{
+	my ($node, $threshold) = @_;
+
+	my $lsns = $node->safe_psql(
+		'postgres', qq{
+		SET wal_compression_threshold = $threshold;
+		SELECT pg_current_wal_insert_lsn();
+		SELECT pg_logical_emit_message(true, 'test 052', repeat('abcd', 65536));
+		SELECT pg_current_wal_insert_lsn();
+	});
+
+	# Three result rows: the LSN before, the message's own LSN, the LSN after.
+	my ($before, $after) = (split /\n/, $lsns)[ 0, 2 ];
+	return $node->safe_psql('postgres',
+		"SELECT '$after'::pg_lsn - '$before'::pg_lsn");
+}
+
+sub test_wal_compression
+{
+	my ($method) = @_;
+
+	note "testing wal_compression = $method";
+
+	my $primary = PostgreSQL::Test::Cluster->new("primary_$method");
+	$primary->init(allows_streaming => 1);
+
+	# Use the minimum threshold so virtually every record gets compressed.
+	$primary->append_conf(
+		'postgresql.conf',
+		"wal_compression = '$method'\n"
+		  . "wal_compression_threshold = 32\n");
+	$primary->start;
+
+	my $backup_name = "backup_$method";
+	$primary->backup($backup_name);
+
+	my $standby = PostgreSQL::Test::Cluster->new("standby_$method");
+	$standby->init_from_backup($primary, $backup_name, has_streaming => 1);
+	$standby->start;
+
+	# Prove the feature actually engages, rather than silently doing nothing.
+	# A large, highly compressible logical message is one record with no
+	# full-page images, so the WAL it occupies is decided purely by whole-record
+	# compression.  Measure it against the same message with the threshold
+	# raised out of reach.
+	my $compressed_bytes = wal_used($primary, 32);
+	my $plain_bytes = wal_used($primary, 1024 * 1024 * 1024);
+
+	if ($method eq 'pglz')
+	{
+		# pglz is excluded from whole-record compression on purpose, so the
+		# threshold must make no difference at all.
+		is($compressed_bytes, $plain_bytes,
+			"whole-record compression does not engage for pglz");
+	}
+	else
+	{
+		cmp_ok($compressed_bytes, '<', $plain_bytes / 2,
+			"whole-record compression shrinks the record ($method): "
+			  . "$compressed_bytes vs $plain_bytes bytes");
+	}
+
+	my $start_lsn =
+	  $primary->safe_psql('postgres', 'SELECT pg_current_wal_insert_lsn()');
+
+	# Generate WAL with records that exceed the compression threshold.
+	# Each row's WAL record includes 200-byte payload well above 32 bytes.
+	$primary->safe_psql(
+		'postgres',
+		"CREATE TABLE t AS
+		 SELECT g, repeat('x', 200) AS d
+		 FROM generate_series(1, 100) AS g");
+
+	$primary->wait_for_replay_catchup($standby);
+
+	is( $standby->safe_psql('postgres', 'SELECT count(*) FROM t'),
+		'100',
+		"compressed WAL replicated via streaming ($method)");
+
+	# Insert another batch that will be recovered after the simulated crash.
+	$primary->safe_psql(
+		'postgres',
+		"INSERT INTO t
+		 SELECT g, repeat('y', 200)
+		 FROM generate_series(101, 200) AS g");
+
+	# pg_waldump has to walk these records too.  That is the frontend decoding
+	# path, which reaches XLogDecompressRecordIfNeeded() without any of the
+	# server's state, so it is worth covering separately from recovery.
+	my $end_lsn =
+	  $primary->safe_psql('postgres', 'SELECT pg_current_wal_flush_lsn()');
+
+	$primary->command_ok(
+		[
+			'pg_waldump',
+			'--quiet',
+			'--path' => $primary->data_dir . '/pg_wal',
+			'--start' => $start_lsn,
+			'--end' => $end_lsn
+		],
+		"pg_waldump decodes compressed WAL ($method)");
+
+	# Stop without a clean shutdown.  On restart PostgreSQL will replay WAL
+	# from the last checkpoint, exercising XLogDecompressRecordIfNeeded for
+	# every compressed record generated since that checkpoint.
+	$primary->stop('immediate');
+	$primary->start;
+
+	is( $primary->safe_psql('postgres', 'SELECT count(*) FROM t'),
+		'200',
+		"crash recovery replays compressed WAL ($method)");
+
+	$primary->stop;
+	$standby->stop;
+}
+
+foreach my $method (@methods)
+{
+	test_wal_compression($method);
+}
+
+done_testing();
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 56c1f997f88..be79df27aee 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -3507,6 +3507,7 @@ X509_STORE
 X509_STORE_CTX
 X86FeatureId
 XLTW_Oper
+XLogCompressionHeader
 XLogCtlData
 XLogCtlInsert
 XLogDumpConfig
-- 
2.50.1 (Apple Git-155)

