Subject: Re: Compress ReorderBuffer spill files using LZ4

From: Julien Tachoires <[email-masked]>
Date: Fri, 7 Jun 2024 06:18:23 -0700

Le ven. 7 juin 2024 à 05:59, Tomas Vondra
<[email-masked]> a écrit :
>
> On 6/6/24 12:58, Julien Tachoires wrote:
> > ...
> >
> > When compiled with LZ4 support (--with-lz4), this patch enables data
> > compression/decompression of these temporary files. Each transaction
> > change that must be written on disk (ReorderBufferDiskChange) is now
> > compressed and encapsulated in a new structure.
> >
>
> I'm a bit confused, but why tie this to having lz4? Why shouldn't this
> be supported even for pglz, or whatever algorithms we add in the future?

That's right, reworking this patch in that sense.

Regards,

JT

---

From: Julien Tachoires <[email-masked]>
Date: Mon, 15 Jul 2024 11:50:56 -0700

Hi,

> That's right, reworking this patch in that sense.

Please find a new version of this patch adding support for LZ4, pglz
and ZSTD. It introduces the new GUC logical_decoding_spill_compression
which is used to set the compression method. In order to stay aligned
with the other server side GUCs related to compression methods
(wal_compression, default_toast_compression), the compression level is
not exposed to users.

The last patch of this set is still in WIP, it adds the machinery
required for setting the compression methods as a subscription option:
CREATE SUBSCRIPTION ... WITH (spill_compression = ...);
I think there is a major problem with this approach: the logical
decoding context is tied to one replication slot, but multiple
subscriptions can use the same replication slot. How should this work
if 2 subscriptions want to use the same replication slot but different
compression methods?

At this point, compression is only available for the changes spilled
on disk. It is still not clear to me if the compression of data
transiting through the streaming protocol should be addressed by this
patch set or by another one. Thought ?

Regards,

JT

---

From: Tomas Vondra <[email-masked]>
Date: Mon, 15 Jul 2024 21:28:29 +0200

> (wal_compression, default_toast_compression), the compression level is

Sounds reasonable. I wonder if it might be useful to allow specifying
the compression level in those places, but that's clearly not something
this patch needs to do.

> if 2 subscriptions want to use the same replication slot but different

Do we really support multiple subscriptions sharing the same slot? I
don't think we do, but maybe I'm missing something.

> patch set or by another one. Thought ?

I'd stick to only compressing the data spilled to disk. It might be
useful to compress the streamed data too, but why shouldn't we compress
the regular (non-streamed) transactions too? Yeah, it's more efficient
to compress larger chunks, but we can fit quite large transactions into
logical_decoding_work_mem without spilling.

FWIW I'd expect that to be handled at the libpq level - there's already
a patch for that, but I haven't checked if it would handle this. But
maybe more importantly, I think compressing streamed data might need to
handle some sort of negotiation of the compression algorithm, which
seems fairly complex.

To conclude, I'd leave this out of scope for this patch.

regards

-- 
Tomas Vondra
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

---

From: Julien Tachoires <[email-masked]>
Date: Tue, 16 Jul 2024 01:08:30 -0700

> don't think we do, but maybe I'm missing something.

You are right, it's not supported, the following error is raised in this case:
ERROR:  replication slot "sub1" is active for PID 51735

I was distracted by the fact that nothing prevents the configuration
of multiple subscriptions sharing the same replication slot.

Thanks,

JT

---

From: Amit Kapila <[email-masked]>
Date: Tue, 16 Jul 2024 18:22:51 +0530

> To conclude, I'd leave this out of scope for this patch.

Your point sounds reasonable to me. OTOH, if we want to support
compression for spill case then shouldn't there be a question how
frequent such an option would be required? Users currently have an
option to stream large transactions for parallel apply or otherwise in
which case no spilling is required. I feel sooner or later we will
make such behavior (streaming=parallel) as default, and then spilling
should happen in very few cases. Is it worth adding this new option
and GUC if that is true?

-- 
With Regards,
Amit Kapila.

---

From: Tomas Vondra <[email-masked]>
Date: Tue, 16 Jul 2024 16:01:46 +0200

> should happen in very few cases. Is it worth adding this new option

I don't know, but streaming is 'off' by default, and I'm not aware of
any proposals to change this, so when you suggest "sooner or later"
we'll change this, I'd probably bet on "later or never".

I haven't been following the discussions about parallel apply very
closely, but my impression from dealing with similar stuff in other
tools is that it's rather easy to run into issues with some workloads,
which just makes me more skeptical about "streamin=parallel" by default.
But as I said, I'm out of the loop so I may be wrong ...

As for whether the GUC is needed, I don't know. I guess we might do the
same thing we do for streaming - we don't have a GUC to enable this, but
we default to 'off' and the client has to request that when opening the
replication connection. So it'd be specified at the subscription level,
more or less.

But then how would we specify compression for cases that invoke decoding
directly by pg_logical_slot_get_changes()? Through options?

BTW if we specify this at subscription level, will it be possible to
change the compression method?

regards

---

From: Amit Kapila <[email-masked]>
Date: Wed, 17 Jul 2024 14:42:36 +0530

> But as I said, I'm out of the loop so I may be wrong ...

It is difficult to say whether enabling it by default will have issues
or not but till now we haven't seen many reports for the streaming =
'parallel' option. It could be due to the reason that not many people
enable it in their workloads. We can probably find out by enabling it
by default.

> directly by pg_logical_slot_get_changes()? Through options?

If we decide to go with this then yeah that is one way, another
possibility is to make it a slot's property, so we can allow to take a
new parameter in pg_create_logical_replication_slot(). We can even
think of inventing a new API to alter the slot's properties if we
decide to go this route.

> change the compression method?

This needs analysis but offhand I can't see the problems with it.

-- 
With Regards,
Amit Kapila.

---

From: Julien Tachoires <[email-masked]>
Date: Fri, 19 Jul 2024 15:05:07 -0700

> think of inventing a new API to alter the slot's properties if we

Please find a new version of this patch set. The compression method is
now set on subscriber level via CREATE SUBSCRIPTION or ALTER
SUBSCRIPTION and can be passed to
pg_logical_slot_get_changes()/pg_logical_slot_get_binary_changes()
through the option spill_compression.

> This needs analysis but offhand I can't see the problems with it.

I didn't notice any issue, the compression method can be changed even
when a decoding is in progress, in this case, the replication worker
restart due to parameter change.

JT

---

From: Tomas Vondra <[email-masked]>
Date: Tue, 17 Sep 2024 20:48:10 +0200

Hi Julien,

Thanks for the last patch version and sorry for the delay. Here's a
quick review, I plan to do a bit of additional testing, hopefully
sometime this week.

Attached is v4, which is your v3 rebased to current master, with a
couple "review" commits, adding comments to relevant places, which I
find easier to follow. There's also a "pgindent" commit at the end,
showing some formatting issues, adding structs to typedefs, etc. This
needs to be applied to the earlier patches, I didn't want to have too
many pgindent commits.

v4-0002-review.patch
--------------------

1) I don't think we need to rename ReorderBufferDiskChange to
ReorderBufferDiskHeader. It seems we could easily just add a field to
ReorderBufferDiskChange, and keep using that, no? It's still just a
wrapper for ReorderBufferChange, even if it's compressed.

2) Likewise, I don't see the point in moving ReorderBufferDiskChange to
a different file. It seems like something that should remain private to
reorderbuffer.c. IMHO the correct thing would be to move the various
ReorderBuffer* funct from reorderbuffer_compress.c to reorderbuffer.c.
The compression is something the ReorderBuffer is responsible for, so
reorderbuffer.c is the right place for that.

3) That means the reorderbuffer_compress.c would have only the actual
compression code. But wouldn't it be better to have one file for each
compression algorithm, similar to src/fe_utils/astreamer_{pglz,lz4,...}?
Just a suggestion, though. Not sure.

4) logical_decoding_spill_compression moved a bit, next to the variables
for other logical_decoding GUCs. It's also missing the definition in a
header file, so I get a compiler warning, so add it to reorderbuffer.h.

5) Won't the code in ReorderBufferCompress() do palloc/pfree for each
piece of data we compress? That seems it might be pretty expensive, at
least for records that happen to be large (>8kB), because oversized
chunks are not cached in the context. IMHO this should use a single
buffer, similarly to what astreamer_lz4_compressor_content does.

6) I'm really confused by having "streaming" and "regular" compression.
I know lz4 supports different modes, but I'd expect only one of them
being really suitable for this, so why support both? But even more
importantly, ReorderBufferCompress() sets the strategy using

  #define lz4_CanDoStreamingCompression(s) \
    (s < (LZ4_RING_BUFFER_SIZE / 2))

But it does that with "s" being the "change size", and that can vary
wildly. Doesn't that means the strategy will change withing a single
file? Does that even make sense?

v4-0004-review.patch
--------------------

7) I'm not sure the "Fix spill_bytes counter" is actually a fix. Even
now (i.e. without compression) it tracks the size of transactions, not
the bytes written to disk exactly, because it doesn't include the bytes
for the size field, so maybe we should not change that ...

v4-0006-review.patch
--------------------

8) It seems a bit strange that we add lz4 first, and only later pglz.
IMHO it should be the other way around, as pglz is the default algorithm
we can rely to have, while lz4 etc. are optional.

One question I have is whether it might be better to compress stuff at a
lower layer - not in reorderbuffer.c, but for the whole file. But maybe
there are reasons why that would be difficult, I haven't tried that.

regards

-- 
Tomas Vondra

From c677ca6271b3b433a65f089a19c371b568bf7a0b Mon Sep 17 00:00:00 2001
From: Julien Tachoires <[email-masked]>
Date: Thu, 6 Jun 2024 00:57:38 -0700
Subject: [PATCH v4 01/11] Compress ReorderBuffer spill files using LZ4

When the content of a large transaction (size exceeding
logical_decoding_work_mem) and its sub-transactions has to be
reordered during logical decoding, then, all the changes are written
on disk in temporary files located in pg_replslot/<slot_name>.

This behavior happens only when the subscriber's option "streaming"
is set to "off", which is the default value.

In this case, large transactions decoding by multiple replication
slots can lead to disk space saturation and high I/O utilization.

When compiled with LZ4 support (--with-lz4), this patch enables data
compression/decompression of these temporary files. Each transaction
change that must be written on disk is now compressed and wrapped in
a new structure named ReorderBufferDiskHeader.

3 different compression strategies are currently implemented:

1. LZ4 streaming compression is the preferred one and works
   efficiently for small individual changes.
2. LZ4 regular compression when the changes are too large for using
   LZ4 streaming API.
3. No compression.
---
 src/backend/replication/logical/Makefile      |   1 +
 src/backend/replication/logical/meson.build   |   1 +
 .../replication/logical/reorderbuffer.c       | 196 ++++---
 .../logical/reorderbuffer_compression.c       | 502 ++++++++++++++++++
 src/include/replication/reorderbuffer.h       |   6 +
 .../replication/reorderbuffer_compression.h   |  95 ++++
 6 files changed, 718 insertions(+), 83 deletions(-)
 create mode 100644 src/backend/replication/logical/reorderbuffer_compression.c
 create mode 100644 src/include/replication/reorderbuffer_compression.h

diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile
index 1e08bbbd4eb..6f9b0c6f8a1 100644
--- a/src/backend/replication/logical/Makefile
+++ b/src/backend/replication/logical/Makefile
@@ -26,6 +26,7 @@ OBJS = \
 	proto.o \
 	relation.o \
 	reorderbuffer.o \
+	reorderbuffer_compression.o \
 	slotsync.o \
 	snapbuild.o \
 	tablesync.o \
diff --git a/src/backend/replication/logical/meson.build b/src/backend/replication/logical/meson.build
index 3d36249d8ad..3f8ff4d798f 100644
--- a/src/backend/replication/logical/meson.build
+++ b/src/backend/replication/logical/meson.build
@@ -12,6 +12,7 @@ backend_sources += files(
   'proto.c',
   'relation.c',
   'reorderbuffer.c',
+  'reorderbuffer_compression.c',
   'slotsync.c',
   'snapbuild.c',
   'tablesync.c',
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 22bcf171ff0..c36179d44b5 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -102,6 +102,7 @@
 #include "pgstat.h"
 #include "replication/logical.h"
 #include "replication/reorderbuffer.h"
+#include "replication/reorderbuffer_compression.h"
 #include "replication/slot.h"
 #include "replication/snapbuild.h"	/* just for SnapBuildSnapDecRefcount */
 #include "storage/bufmgr.h"
@@ -112,6 +113,8 @@
 #include "utils/rel.h"
 #include "utils/relfilenumbermap.h"
 
+int			logical_decoding_spill_compression = REORDER_BUFFER_NO_COMPRESSION;
+
 /* entry for a hash table we use to map from xid to our transaction state */
 typedef struct ReorderBufferTXNByIdEnt
 {
@@ -173,14 +176,6 @@ typedef struct ReorderBufferToastEnt
 									 * main tup */
 } ReorderBufferToastEnt;
 
-/* Disk serialization support datastructures */
-typedef struct ReorderBufferDiskChange
-{
-	Size		size;
-	ReorderBufferChange change;
-	/* data follows */
-} ReorderBufferDiskChange;
-
 #define IsSpecInsert(action) \
 ( \
 	((action) == REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT) \
@@ -255,6 +250,8 @@ static void ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *tx
 										 int fd, ReorderBufferChange *change);
 static Size ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn,
 										TXNEntryFile *file, XLogSegNo *segno);
+static bool ReorderBufferReadOndiskChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
+								   TXNEntryFile *file, XLogSegNo *segno);
 static void ReorderBufferRestoreChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 									   char *data);
 static void ReorderBufferRestoreCleanup(ReorderBuffer *rb, ReorderBufferTXN *txn);
@@ -427,6 +424,8 @@ ReorderBufferGetTXN(ReorderBuffer *rb)
 	/* InvalidCommandId is not zero, so set it explicitly */
 	txn->command_id = InvalidCommandId;
 	txn->output_plugin_private = NULL;
+	txn->compressor_state = ReorderBufferNewCompressorState(rb->context,
+															logical_decoding_spill_compression);
 
 	return txn;
 }
@@ -464,6 +463,10 @@ ReorderBufferReturnTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 		txn->invalidations = NULL;
 	}
 
+	ReorderBufferFreeCompressorState(rb->context,
+									 logical_decoding_spill_compression,
+									 txn->compressor_state);
+
 	/* Reset the toast hash */
 	ReorderBufferToastReset(rb, txn);
 
@@ -3800,13 +3803,13 @@ static void
 ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 							 int fd, ReorderBufferChange *change)
 {
-	ReorderBufferDiskChange *ondisk;
-	Size		sz = sizeof(ReorderBufferDiskChange);
+	ReorderBufferDiskHeader *disk_hdr;
+	Size		sz = sizeof(ReorderBufferDiskHeader) + sizeof(ReorderBufferChange);
 
 	ReorderBufferSerializeReserve(rb, sz);
 
-	ondisk = (ReorderBufferDiskChange *) rb->outbuf;
-	memcpy(&ondisk->change, change, sizeof(ReorderBufferChange));
+	disk_hdr = (ReorderBufferDiskHeader *) rb->outbuf;
+	memcpy((char *)rb->outbuf + sizeof(ReorderBufferDiskHeader), change, sizeof(ReorderBufferChange));
 
 	switch (change->action)
 	{
@@ -3842,9 +3845,9 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 				/* make sure we have enough space */
 				ReorderBufferSerializeReserve(rb, sz);
 
-				data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange);
+				data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskHeader) + sizeof(ReorderBufferChange);
 				/* might have been reallocated above */
-				ondisk = (ReorderBufferDiskChange *) rb->outbuf;
+				disk_hdr = (ReorderBufferDiskHeader *) rb->outbuf;
 
 				if (oldlen)
 				{
@@ -3874,10 +3877,10 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 					sizeof(Size) + sizeof(Size);
 				ReorderBufferSerializeReserve(rb, sz);
 
-				data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange);
+				data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskHeader) + sizeof(ReorderBufferChange);
 
 				/* might have been reallocated above */
-				ondisk = (ReorderBufferDiskChange *) rb->outbuf;
+				disk_hdr = (ReorderBufferDiskHeader *) rb->outbuf;
 
 				/* write the prefix including the size */
 				memcpy(data, &prefix_size, sizeof(Size));
@@ -3904,10 +3907,10 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 				sz += inval_size;
 
 				ReorderBufferSerializeReserve(rb, sz);
-				data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange);
+				data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskHeader) + sizeof(ReorderBufferChange);
 
 				/* might have been reallocated above */
-				ondisk = (ReorderBufferDiskChange *) rb->outbuf;
+				disk_hdr = (ReorderBufferDiskHeader *) rb->outbuf;
 				memcpy(data, change->data.inval.invalidations, inval_size);
 				data += inval_size;
 
@@ -3926,9 +3929,9 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 
 				/* make sure we have enough space */
 				ReorderBufferSerializeReserve(rb, sz);
-				data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange);
+				data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskHeader) + sizeof(ReorderBufferChange);
 				/* might have been reallocated above */
-				ondisk = (ReorderBufferDiskChange *) rb->outbuf;
+				disk_hdr = (ReorderBufferDiskHeader *) rb->outbuf;
 
 				memcpy(data, snap, sizeof(SnapshotData));
 				data += sizeof(SnapshotData);
@@ -3960,9 +3963,9 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 				/* make sure we have enough space */
 				ReorderBufferSerializeReserve(rb, sz);
 
-				data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange);
+				data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskHeader) + sizeof(ReorderBufferChange);
 				/* might have been reallocated above */
-				ondisk = (ReorderBufferDiskChange *) rb->outbuf;
+				disk_hdr = (ReorderBufferDiskHeader *) rb->outbuf;
 
 				memcpy(data, change->data.truncate.relids, size);
 				data += size;
@@ -3977,11 +3980,14 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 			break;
 	}
 
-	ondisk->size = sz;
+	/* Inplace ReorderBuffer content compression before writing it on disk */
+	ReorderBufferCompress(rb, &disk_hdr, logical_decoding_spill_compression,
+						  sz, txn->compressor_state);
 
 	errno = 0;
 	pgstat_report_wait_start(WAIT_EVENT_REORDER_BUFFER_WRITE);
-	if (write(fd, rb->outbuf, ondisk->size) != ondisk->size)
+
+	if (write(fd, rb->outbuf, disk_hdr->size) != disk_hdr->size)
 	{
 		int			save_errno = errno;
 
@@ -4006,8 +4012,6 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 	 */
 	if (txn->final_lsn < change->lsn)
 		txn->final_lsn = change->lsn;
-
-	Assert(ondisk->change.action == change->action);
 }
 
 /* Returns true, if the output plugin supports streaming, false, otherwise. */
@@ -4276,9 +4280,6 @@ ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn,
 
 	while (restored < max_changes_in_memory && *segno <= last_segno)
 	{
-		int			readBytes;
-		ReorderBufferDiskChange *ondisk;
-
 		CHECK_FOR_INTERRUPTS();
 
 		if (*fd == -1)
@@ -4317,60 +4318,15 @@ ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn,
 		}
 
 		/*
-		 * Read the statically sized part of a change which has information
-		 * about the total size. If we couldn't read a record, we're at the
-		 * end of this file.
+		 * Read the full change from disk.
+		 * If ReorderBufferReadOndiskChange returns false, then we are at the
+		 * eof, so, move the next segment.
 		 */
-		ReorderBufferSerializeReserve(rb, sizeof(ReorderBufferDiskChange));
-		readBytes = FileRead(file->vfd, rb->outbuf,
-							 sizeof(ReorderBufferDiskChange),
-							 file->curOffset, WAIT_EVENT_REORDER_BUFFER_READ);
-
-		/* eof */
-		if (readBytes == 0)
+		if (!ReorderBufferReadOndiskChange(rb, txn, file, segno))
 		{
-			FileClose(*fd);
 			*fd = -1;
-			(*segno)++;
 			continue;
 		}
-		else if (readBytes < 0)
-			ereport(ERROR,
-					(errcode_for_file_access(),
-					 errmsg("could not read from reorderbuffer spill file: %m")));
-		else if (readBytes != sizeof(ReorderBufferDiskChange))
-			ereport(ERROR,
-					(errcode_for_file_access(),
-					 errmsg("could not read from reorderbuffer spill file: read %d instead of %u bytes",
-							readBytes,
-							(uint32) sizeof(ReorderBufferDiskChange))));
-
-		file->curOffset += readBytes;
-
-		ondisk = (ReorderBufferDiskChange *) rb->outbuf;
-
-		ReorderBufferSerializeReserve(rb,
-									  sizeof(ReorderBufferDiskChange) + ondisk->size);
-		ondisk = (ReorderBufferDiskChange *) rb->outbuf;
-
-		readBytes = FileRead(file->vfd,
-							 rb->outbuf + sizeof(ReorderBufferDiskChange),
-							 ondisk->size - sizeof(ReorderBufferDiskChange),
-							 file->curOffset,
-							 WAIT_EVENT_REORDER_BUFFER_READ);
-
-		if (readBytes < 0)
-			ereport(ERROR,
-					(errcode_for_file_access(),
-					 errmsg("could not read from reorderbuffer spill file: %m")));
-		else if (readBytes != ondisk->size - sizeof(ReorderBufferDiskChange))
-			ereport(ERROR,
-					(errcode_for_file_access(),
-					 errmsg("could not read from reorderbuffer spill file: read %d instead of %u bytes",
-							readBytes,
-							(uint32) (ondisk->size - sizeof(ReorderBufferDiskChange)))));
-
-		file->curOffset += readBytes;
 
 		/*
 		 * ok, read a full change from disk, now restore it into proper
@@ -4383,6 +4339,83 @@ ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn,
 	return restored;
 }
 
+/*
+ * Read a change spilled to disk and decompress it if compressed.
+ */
+static bool
+ReorderBufferReadOndiskChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
+							  TXNEntryFile *file, XLogSegNo *segno)
+{
+	int			readBytes;
+	ReorderBufferDiskHeader *disk_hdr;
+	char	   *header;			/* disk header buffer*/
+	char	   *data;			/* data buffer */
+
+	/*
+	 * Read the statically sized part of a change which has information about
+	 * the total size and compression method. If we couldn't read a record,
+	 * we're at the end of this file.
+	 */
+	header = (char *) palloc0(sizeof(ReorderBufferDiskHeader));
+	readBytes = FileRead(file->vfd, header,
+						 sizeof(ReorderBufferDiskHeader),
+						 file->curOffset, WAIT_EVENT_REORDER_BUFFER_READ);
+
+	/* eof */
+	if (readBytes == 0)
+	{
+
+		FileClose(file->vfd);
+		(*segno)++;
+		pfree(header);
+
+		return false;
+	}
+	else if (readBytes < 0)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not read from reorderbuffer spill file: %m")));
+	else if (readBytes != sizeof(ReorderBufferDiskHeader))
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not read from reorderbuffer spill file: read %d instead of %u bytes",
+						readBytes,
+						(uint32) sizeof(ReorderBufferDiskHeader))));
+
+	file->curOffset += readBytes;
+
+	disk_hdr = (ReorderBufferDiskHeader *) header;
+
+	/* Read ondisk data */
+	data = (char *) palloc0(disk_hdr->size - sizeof(ReorderBufferDiskHeader));
+	readBytes = FileRead(file->vfd,
+						 data,
+						 disk_hdr->size - sizeof(ReorderBufferDiskHeader),
+						 file->curOffset,
+						 WAIT_EVENT_REORDER_BUFFER_READ);
+
+	if (readBytes < 0)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not read from reorderbuffer spill file: %m")));
+	else if (readBytes != (disk_hdr->size - sizeof(ReorderBufferDiskHeader)))
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not read from reorderbuffer spill file: read %d instead of %u bytes",
+						readBytes,
+						(uint32) (disk_hdr->size - sizeof(ReorderBufferDiskHeader)))));
+
+	/* Decompress data */
+	ReorderBufferDecompress(rb, data, disk_hdr, txn->compressor_state);
+
+	pfree(data);
+	pfree(header);
+
+	file->curOffset += readBytes;
+
+	return true;
+}
+
 /*
  * Convert change from its on-disk format to in-memory format and queue it onto
  * the TXN's ->changes list.
@@ -4395,17 +4428,14 @@ static void
 ReorderBufferRestoreChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 						   char *data)
 {
-	ReorderBufferDiskChange *ondisk;
 	ReorderBufferChange *change;
 
-	ondisk = (ReorderBufferDiskChange *) data;
-
 	change = ReorderBufferGetChange(rb);
 
 	/* copy static part */
-	memcpy(change, &ondisk->change, sizeof(ReorderBufferChange));
+	memcpy(change, data + sizeof(ReorderBufferDiskHeader), sizeof(ReorderBufferChange));
 
-	data += sizeof(ReorderBufferDiskChange);
+	data += sizeof(ReorderBufferDiskHeader) + sizeof(ReorderBufferChange);
 
 	/* restore individual stuff */
 	switch (change->action)
diff --git a/src/backend/replication/logical/reorderbuffer_compression.c b/src/backend/replication/logical/reorderbuffer_compression.c
new file mode 100644
index 00000000000..77f5c76929b
--- /dev/null
+++ b/src/backend/replication/logical/reorderbuffer_compression.c
@@ -0,0 +1,502 @@
+/*-------------------------------------------------------------------------
+ *
+ * reorderbuffer_compression.c
+ *	  Functions for ReorderBuffer compression.
+ *
+ * Copyright (c) 2024-2024, PostgreSQL Global Development Group
+ *
+ *
+ * IDENTIFICATION
+ *	  src/backend/access/common/reorderbuffer_compression.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#ifdef USE_LZ4
+#include <lz4.h>
+#endif
+
+#include "replication/reorderbuffer_compression.h"
+
+#define NO_LZ4_SUPPORT() \
+	ereport(ERROR, \
+			(errcode(ERRCODE_FEATURE_NOT_SUPPORTED), \
+			 errmsg("compression method lz4 not supported"), \
+			 errdetail("This functionality requires the server to be built with lz4 support.")))
+
+/*
+ * Allocate a new LZ4StreamingCompressorState.
+ */
+static void *
+lz4_NewCompressorState(MemoryContext context)
+{
+#ifndef USE_LZ4
+	NO_LZ4_SUPPORT();
+	return NULL;				/* keep compiler quiet */
+#else
+	LZ4StreamingCompressorState *cstate;
+
+	cstate = (LZ4StreamingCompressorState *)
+		MemoryContextAlloc(context, sizeof(LZ4StreamingCompressorState));
+
+	/*
+	 * We do not allocate LZ4 ring buffers and streaming handlers at this
+	 * point because we have no guarantee that we will need them later. Let's
+	 * allocate only when we are about to use them.
+	 */
+	cstate->lz4_in_buf = NULL;
+	cstate->lz4_out_buf = NULL;
+	cstate->lz4_in_buf_offset = 0;
+	cstate->lz4_out_buf_offset = 0;
+	cstate->lz4_stream = NULL;
+	cstate->lz4_stream_decode = NULL;
+
+	return (void *) cstate;
+#endif
+}
+
+/*
+ * Free LZ4 memory resources and the compressor state.
+ */
+static void
+lz4_FreeCompressorState(MemoryContext context, void *compressor_state)
+{
+#ifndef USE_LZ4
+	NO_LZ4_SUPPORT();
+#else
+	LZ4StreamingCompressorState *cstate;
+	MemoryContext oldcontext;
+
+	if (compressor_state == NULL)
+		return;
+
+	oldcontext = MemoryContextSwitchTo(context);
+
+	cstate = (LZ4StreamingCompressorState *) compressor_state;
+
+	if (cstate->lz4_in_buf != NULL)
+	{
+		pfree(cstate->lz4_in_buf);
+		LZ4_freeStream(cstate->lz4_stream);
+	}
+	if (cstate->lz4_out_buf != NULL)
+	{
+		pfree(cstate->lz4_out_buf);
+		LZ4_freeStreamDecode(cstate->lz4_stream_decode);
+	}
+
+	pfree(compressor_state);
+
+	MemoryContextSwitchTo(oldcontext);
+#endif
+}
+
+#ifdef USE_LZ4
+/*
+ * Allocate LZ4 input ring buffer and create the streaming compression handler.
+ */
+static void
+lz4_CreateStreamCompressorState(MemoryContext context, void *compressor_state)
+{
+	LZ4StreamingCompressorState *cstate;
+	MemoryContext oldcontext = MemoryContextSwitchTo(context);
+
+	cstate = (LZ4StreamingCompressorState *) compressor_state;
+	cstate->lz4_in_buf = (char *) palloc0(LZ4_RING_BUFFER_SIZE);
+	cstate->lz4_stream = LZ4_createStream();
+
+	MemoryContextSwitchTo(oldcontext);
+}
+#endif
+
+#ifdef USE_LZ4
+/*
+ * Allocate LZ4 output ring buffer and create the streaming decompression
+ * handler.
+ */
+static void
+lz4_CreateStreamDecodeCompressorState(MemoryContext context,
+									  void *compressor_state)
+{
+	LZ4StreamingCompressorState *cstate;
+	MemoryContext oldcontext = MemoryContextSwitchTo(context);
+
+	cstate = (LZ4StreamingCompressorState *) compressor_state;
+	cstate->lz4_out_buf = (char *) palloc0(LZ4_RING_BUFFER_SIZE);
+	cstate->lz4_stream_decode = LZ4_createStreamDecode();
+
+	MemoryContextSwitchTo(oldcontext);
+}
+#endif
+
+/*
+ * Data compression using LZ4 streaming API.
+ * Caller must ensure that the source data can fit in LZ4 input ring buffer,
+ * this checking must be done by lz4_CanDoStreamingCompression().
+ */
+static void
+lz4_StreamingCompressData(MemoryContext context, char *src, Size src_size,
+						  char **dst, Size *dst_size, void *compressor_state)
+{
+#ifndef USE_LZ4
+	NO_LZ4_SUPPORT();
+#else
+	LZ4StreamingCompressorState *cstate;
+	int			lz4_cmp_size = 0;	/* compressed size */
+	char	   *buf;				/* buffer used for compression */
+	Size		buf_size;			/* buffer size */
+	char	   *lz4_in_bufPtr;		/* input ring buffer pointer */
+
+	cstate = (LZ4StreamingCompressorState *) compressor_state;
+
+	/* Allocate LZ4 input ring buffer and streaming compression handler */
+	if (cstate->lz4_in_buf == NULL)
+		lz4_CreateStreamCompressorState(context, compressor_state);
+
+	/* Ring buffer offset wraparound */
+	if ((cstate->lz4_in_buf_offset + src_size) > LZ4_RING_BUFFER_SIZE)
+		cstate->lz4_in_buf_offset = 0;
+
+	/* Get the pointer of the next entry in the ring buffer */
+	lz4_in_bufPtr = cstate->lz4_in_buf + cstate->lz4_in_buf_offset;
+
+	/* Copy data that should be compressed into LZ4 input ring buffer */
+	memcpy(lz4_in_bufPtr, src, src_size);
+
+	/* Allocate space for compressed data */
+	buf_size = LZ4_COMPRESSBOUND(src_size);
+	buf = (char *) palloc0(buf_size);
+
+	/* Use LZ4 streaming compression API */
+	lz4_cmp_size = LZ4_compress_fast_continue(cstate->lz4_stream,
+											  lz4_in_bufPtr, buf, src_size,
+											  buf_size, 1);
+
+	if (lz4_cmp_size <= 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_DATA_CORRUPTED),
+				 errmsg_internal("LZ4 compression failed")));
+
+	/* Move the input ring buffer offset */
+	cstate->lz4_in_buf_offset += src_size;
+
+	*dst_size = lz4_cmp_size;
+	*dst = buf;
+#endif
+}
+
+/*
+ * Data compression using LZ4 API.
+ */
+static void
+lz4_CompressData(char *src, Size src_size, char **dst,  Size *dst_size)
+{
+#ifndef USE_LZ4
+	NO_LZ4_SUPPORT();
+#else
+	int			lz4_cmp_size = 0;	/* compressed size */
+	char	   *buf;				/* buffer used for compression */
+	Size		buf_size;			/* buffer size */
+
+	buf_size = LZ4_COMPRESSBOUND(src_size);
+	buf = (char *) palloc0(buf_size);
+
+	/* Use LZ4 regular compression API */
+	lz4_cmp_size = LZ4_compress_default(src, buf, src_size, buf_size);
+
+	if (lz4_cmp_size <= 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_DATA_CORRUPTED),
+				 errmsg_internal("LZ4 compression failed")));
+
+	*dst_size = lz4_cmp_size;
+	*dst = buf;
+#endif
+}
+
+/*
+ * Data decompression using LZ4 streaming API.
+ * LZ4 decompression uses the output ring buffer to store decompressed data,
+ * thus, we don't need to create a new buffer. We return the pointer to data
+ * location.
+ */
+static void
+lz4_StreamingDecompressData(MemoryContext context, char *src, Size src_size,
+							char **dst, Size dst_size, void *compressor_state)
+{
+#ifndef USE_LZ4
+	NO_LZ4_SUPPORT();
+#else
+	LZ4StreamingCompressorState *cstate;
+	char	   *lz4_out_bufPtr;		/* output ring buffer pointer */
+	int			lz4_dec_size;		/* decompressed data size */
+
+	cstate = (LZ4StreamingCompressorState *) compressor_state;
+
+	/* Allocate LZ4 output ring buffer and streaming decompression handler */
+	if (cstate->lz4_out_buf == NULL)
+		lz4_CreateStreamDecodeCompressorState(context, compressor_state);
+
+	/* Ring buffer offset wraparound */
+	if ((cstate->lz4_out_buf_offset + dst_size) > LZ4_RING_BUFFER_SIZE)
+		cstate->lz4_out_buf_offset = 0;
+
+	/* Get current entry pointer in the ring buffer */
+	lz4_out_bufPtr = cstate->lz4_out_buf + cstate->lz4_out_buf_offset;
+
+	lz4_dec_size = LZ4_decompress_safe_continue(cstate->lz4_stream_decode,
+												src,
+												lz4_out_bufPtr,
+												src_size,
+												dst_size);
+
+	Assert(lz4_dec_size == dst_size);
+
+	if (lz4_dec_size < 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_DATA_CORRUPTED),
+				 errmsg_internal("compressed LZ4 data is corrupted")));
+	else if (lz4_dec_size != dst_size)
+		ereport(ERROR,
+			(errcode(ERRCODE_DATA_CORRUPTED),
+			 errmsg_internal("decompressed LZ4 data size differs from original size")));
+
+	/* Move the output ring buffer offset */
+	cstate->lz4_out_buf_offset += lz4_dec_size;
+
+	/* Point to the decompressed data location */
+	*dst = lz4_out_bufPtr;
+#endif
+}
+
+/*
+ * Data decompression using LZ4 API.
+ */
+static void
+lz4_DecompressData(char *src, Size src_size, char **dst, Size dst_size)
+{
+#ifndef USE_LZ4
+	NO_LZ4_SUPPORT();
+#else
+	int			lz4_dec_bytes;
+	char	   *buf;
+
+	buf = (char *) palloc0(dst_size);
+
+	lz4_dec_bytes = LZ4_decompress_safe(src, buf, src_size, dst_size);
+
+	Assert(lz4_dec_bytes == dst_size);
+
+	if (lz4_dec_bytes < 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_DATA_CORRUPTED),
+				 errmsg_internal("compressed LZ4 data is corrupted")));
+	else if (lz4_dec_bytes != dst_size)
+		ereport(ERROR,
+				(errcode(ERRCODE_DATA_CORRUPTED),
+				 errmsg_internal("decompressed LZ4 data size differs from original size")));
+
+	*dst = buf;
+#endif
+}
+
+/*
+ * Allocate a new Compressor State, depending on the compression method.
+ */
+void *
+ReorderBufferNewCompressorState(MemoryContext context, int compression_method)
+{
+	switch (compression_method)
+	{
+		case REORDER_BUFFER_LZ4_COMPRESSION:
+			return lz4_NewCompressorState(context);
+			break;
+		case REORDER_BUFFER_NO_COMPRESSION:
+		default:
+			return NULL;
+			break;
+	}
+}
+
+/*
+ * Free memory allocated to a Compressor State, depending on the compression
+ * method.
+ */
+void
+ReorderBufferFreeCompressorState(MemoryContext context, int compression_method,
+								 void *compressor_state)
+{
+	switch (compression_method)
+	{
+		case REORDER_BUFFER_LZ4_COMPRESSION:
+			return lz4_FreeCompressorState(context, compressor_state);
+			break;
+		case REORDER_BUFFER_NO_COMPRESSION:
+		default:
+			break;
+	}
+}
+
+/*
+ * Ensure the IO buffer is >= sz.
+ */
+static void
+ReorderBufferReserve(ReorderBuffer *rb, Size sz)
+{
+	if (rb->outbufsize < sz)
+	{
+		rb->outbuf = repalloc(rb->outbuf, sz);
+		rb->outbufsize = sz;
+	}
+}
+
+/*
+ * Compress ReorderBuffer content. This function is called in order to compress
+ * data before spilling on disk.
+ */
+void
+ReorderBufferCompress(ReorderBuffer *rb, ReorderBufferDiskHeader **header,
+					  int compression_method, Size data_size,
+					  void *compressor_state)
+{
+	ReorderBufferDiskHeader *hdr = *header;
+
+	switch (compression_method)
+	{
+		/* No compression */
+		case REORDER_BUFFER_NO_COMPRESSION:
+		{
+			hdr->comp_strat = REORDER_BUFFER_STRAT_UNCOMPRESSED;
+			hdr->size = data_size;
+			hdr->raw_size = data_size - sizeof(ReorderBufferDiskHeader);
+
+			break;
+		}
+		/* LZ4 Compression */
+		case REORDER_BUFFER_LZ4_COMPRESSION:
+		{
+			char	   *dst = NULL;
+			Size		dst_size = 0;
+			char	   *src = (char *) rb->outbuf + sizeof(ReorderBufferDiskHeader);
+			Size		src_size = data_size - sizeof(ReorderBufferDiskHeader);
+			ReorderBufferCompressionStrategy strat;
+
+			if (lz4_CanDoStreamingCompression(src_size))
+			{
+				/* Use LZ4 streaming compression if possible */
+				lz4_StreamingCompressData(rb->context, src, src_size, &dst,
+										  &dst_size, compressor_state);
+				strat = REORDER_BUFFER_STRAT_LZ4_STREAMING;
+			}
+			else
+			{
+				/* Fallback to LZ4 regular compression */
+				lz4_CompressData(src, src_size, &dst, &dst_size);
+				strat = REORDER_BUFFER_STRAT_LZ4_REGULAR;
+			}
+
+			/*
+			 * Make sure the ReorderBuffer has enough space to store compressed
+			 * data. Compressed data must be smaller than raw data, so, the
+			 * ReorderBuffer should already have room for compressed data, but
+			 * we do this to avoid buffer overflow risks.
+			 */
+			ReorderBufferReserve(rb, (dst_size + sizeof(ReorderBufferDiskHeader)));
+
+			hdr = (ReorderBufferDiskHeader *) rb->outbuf;
+			hdr->comp_strat = strat;
+			hdr->size = dst_size + sizeof(ReorderBufferDiskHeader);
+			hdr->raw_size = src_size;
+
+			/*
+			 * Update header: hdr pointer has potentially changed due to
+			 * ReorderBufferReserve()
+			 */
+			*header = hdr;
+
+			/* Copy back compressed data into the ReorderBuffer */
+			memcpy((char *) rb->outbuf + sizeof(ReorderBufferDiskHeader), dst,
+				   dst_size);
+
+			pfree(dst);
+
+			break;
+		}
+	}
+}
+
+/*
+ * Decompress data read from disk and copy it into the ReorderBuffer.
+ */
+void
+ReorderBufferDecompress(ReorderBuffer *rb, char *data,
+						ReorderBufferDiskHeader *header, void *compressor_state)
+{
+	Size		raw_outbufsize = header->raw_size + sizeof(ReorderBufferDiskHeader);
+	/*
+	 * Make sure the output reorder buffer has enough space to store
+	 * decompressed/raw data.
+	 */
+	if (rb->outbufsize < raw_outbufsize)
+	{
+		rb->outbuf = repalloc(rb->outbuf, raw_outbufsize);
+		rb->outbufsize = raw_outbufsize;
+	}
+
+	/* Make a copy of the header read on disk into the ReorderBuffer */
+	memcpy(rb->outbuf, (char *) header, sizeof(ReorderBufferDiskHeader));
+
+	switch (header->comp_strat)
+	{
+		/* No decompression */
+		case REORDER_BUFFER_STRAT_UNCOMPRESSED:
+			{
+				/*
+				 * Make a copy of what was read on disk into the reorder
+				 * buffer.
+				 */
+				memcpy((char *) rb->outbuf + sizeof(ReorderBufferDiskHeader),
+					   data, header->raw_size);
+				break;
+			}
+		/* LZ4 regular decompression */
+		case REORDER_BUFFER_STRAT_LZ4_REGULAR:
+			{
+				char	   *buf;
+				Size		src_size = header->size - sizeof(ReorderBufferDiskHeader);
+				Size		buf_size = header->raw_size;
+
+				lz4_DecompressData(data, src_size, &buf, buf_size);
+
+				/* Copy decompressed data into the ReorderBuffer */
+				memcpy((char *) rb->outbuf + sizeof(ReorderBufferDiskHeader),
+					   buf, buf_size);
+
+				pfree(buf);
+				break;
+			}
+		/* LZ4 streaming decompression */
+		case REORDER_BUFFER_STRAT_LZ4_STREAMING:
+			{
+				char	   *buf;
+				Size		src_size = header->size - sizeof(ReorderBufferDiskHeader);
+				Size		buf_size = header->raw_size;
+
+				lz4_StreamingDecompressData(rb->context, data, src_size, &buf,
+										   buf_size, compressor_state);
+
+				/* Copy decompressed data into the ReorderBuffer */
+				memcpy((char *) rb->outbuf + sizeof(ReorderBufferDiskHeader),
+					   buf, buf_size);
+				/*
+				 * Not necessary to free buf in this case: it points to the
+				 * decompressed data stored in LZ4 output ring buffer.
+				 */
+				break;
+			}
+		default:
+			/* Other compression methods not yet supported */
+			break;
+	}
+}
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index e332635f70b..f1562a77719 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -9,6 +9,10 @@
 #ifndef REORDERBUFFER_H
 #define REORDERBUFFER_H
 
+#ifdef USE_LZ4
+#include <lz4.h>
+#endif
+
 #include "access/htup_details.h"
 #include "lib/ilist.h"
 #include "lib/pairingheap.h"
@@ -427,6 +431,8 @@ typedef struct ReorderBufferTXN
 	 * Private data pointer of the output plugin.
 	 */
 	void	   *output_plugin_private;
+
+	void	   *compressor_state;
 } ReorderBufferTXN;
 
 /* so we can define the callbacks used inside struct ReorderBuffer itself */
diff --git a/src/include/replication/reorderbuffer_compression.h b/src/include/replication/reorderbuffer_compression.h
new file mode 100644
index 00000000000..9aa8aea56f4
--- /dev/null
+++ b/src/include/replication/reorderbuffer_compression.h
@@ -0,0 +1,95 @@
+/*-------------------------------------------------------------------------
+ *
+ * reorderbuffer_compression.h
+ *	  Functions for ReorderBuffer compression.
+ *
+ * Copyright (c) 2024-2024, PostgreSQL Global Development Group
+ *
+ * src/include/access/reorderbuffer_compression.h
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#ifndef REORDERBUFFER_COMPRESSION_H
+#define REORDERBUFFER_COMPRESSION_H
+
+#include "replication/reorderbuffer.h"
+
+#ifdef USE_LZ4
+#include <lz4.h>
+#endif
+
+/* ReorderBuffer on disk compression algorithms */
+typedef enum ReorderBufferCompressionMethod
+{
+	REORDER_BUFFER_NO_COMPRESSION,
+	REORDER_BUFFER_LZ4_COMPRESSION,
+} ReorderBufferCompressionMethod;
+
+/*
+ * Compression strategy applied to ReorderBuffer records spilled on disk
+ */
+typedef enum ReorderBufferCompressionStrategy
+{
+	REORDER_BUFFER_STRAT_UNCOMPRESSED,
+	REORDER_BUFFER_STRAT_LZ4_STREAMING,
+	REORDER_BUFFER_STRAT_LZ4_REGULAR,
+} ReorderBufferCompressionStrategy;
+
+/* Disk serialization support datastructures */
+typedef struct ReorderBufferDiskHeader
+{
+	ReorderBufferCompressionStrategy comp_strat; /* Compression strategy */
+	Size		size;					/* Ondisk size */
+	Size		raw_size;				/* Raw/uncompressed data size */
+	/* ReorderBufferChange + data follows */
+} ReorderBufferDiskHeader;
+
+#ifdef USE_LZ4
+/*
+ * We use a fairly small LZ4 ring buffer size (64kB). Using a larger buffer
+ * size provide better compression ratio, but as long as we have to allocate
+ * two LZ4 ring buffers per ReorderBufferTXN, we should keep it small.
+ */
+#define LZ4_RING_BUFFER_SIZE (64 * 1024)
+
+/*
+ * Use LZ4 streaming compression iff we can keep at least 2 uncompressed
+ * records into the LZ4 input ring buffer. If raw data size is too large, let's
+ * use regular LZ4 compression.
+ */
+#define lz4_CanDoStreamingCompression(s) (s < (LZ4_RING_BUFFER_SIZE / 2))
+
+/*
+ * LZ4 streaming compression/decompression handlers and ring
+ * buffers.
+ */
+typedef struct LZ4StreamingCompressorState {
+	/* Streaming compression handler */
+	LZ4_stream_t *lz4_stream;
+	/* Streaming decompression handler */
+	LZ4_streamDecode_t *lz4_stream_decode;
+	/* LZ4 in/out ring buffers used for streaming compression */
+	char	   *lz4_in_buf;
+	int			lz4_in_buf_offset;
+	char	   *lz4_out_buf;
+	int			lz4_out_buf_offset;
+} LZ4StreamingCompressorState;
+#else
+#define lz4_CanDoStreamingCompression(s) (false)
+#endif
+
+extern void *ReorderBufferNewCompressorState(MemoryContext context,
+											 int compression_method);
+extern void ReorderBufferFreeCompressorState(MemoryContext context,
+											 int compression_method,
+											 void *compressor_state);
+extern void ReorderBufferCompress(ReorderBuffer *rb,
+								  ReorderBufferDiskHeader **header,
+								  int compression_method, Size data_size,
+								  void *compressor_state);
+extern void ReorderBufferDecompress(ReorderBuffer *rb, char *data,
+									ReorderBufferDiskHeader *header,
+									void *compressor_state);
+
+#endif							/* REORDERBUFFER_COMPRESSION_H */
-- 
2.46.0

From 17aed41dfdd452727643a40e5dba6c0c4705c64a Mon Sep 17 00:00:00 2001
From: Tomas Vondra <[email-masked]>
Date: Tue, 17 Sep 2024 16:37:49 +0200
Subject: [PATCH v4 02/11] review

---
 .../replication/logical/reorderbuffer.c        | 10 ++++++++--
 .../logical/reorderbuffer_compression.c        | 18 +++++++++++++++++-
 src/include/replication/reorderbuffer.h        |  2 ++
 .../replication/reorderbuffer_compression.h    |  8 ++++++++
 4 files changed, 35 insertions(+), 3 deletions(-)

diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index c36179d44b5..f5c51316d9d 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -113,8 +113,6 @@
 #include "utils/rel.h"
 #include "utils/relfilenumbermap.h"
 
-int			logical_decoding_spill_compression = REORDER_BUFFER_NO_COMPRESSION;
-
 /* entry for a hash table we use to map from xid to our transaction state */
 typedef struct ReorderBufferTXNByIdEnt
 {
@@ -176,6 +174,11 @@ typedef struct ReorderBufferToastEnt
 									 * main tup */
 } ReorderBufferToastEnt;
 
+/*
+ * XXX seems wrong to move ReorderBufferDiskChange typedef, it should be private
+ * in this file, even with compression IMHO
+ */
+
 #define IsSpecInsert(action) \
 ( \
 	((action) == REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT) \
@@ -210,6 +213,9 @@ static const Size max_changes_in_memory = 4096; /* XXX for restore only */
 /* GUC variable */
 int			debug_logical_replication_streaming = DEBUG_LOGICAL_REP_STREAMING_BUFFERED;
 
+/* Compression strategy for spilled data. */
+int			logical_decoding_spill_compression = REORDER_BUFFER_NO_COMPRESSION;
+
 /* ---------------------------------------
  * primary reorderbuffer support routines
  * ---------------------------------------
diff --git a/src/backend/replication/logical/reorderbuffer_compression.c b/src/backend/replication/logical/reorderbuffer_compression.c
index 77f5c76929b..b4fea15ebd9 100644
--- a/src/backend/replication/logical/reorderbuffer_compression.c
+++ b/src/backend/replication/logical/reorderbuffer_compression.c
@@ -9,6 +9,8 @@
  * IDENTIFICATION
  *	  src/backend/access/common/reorderbuffer_compression.c
  *
+ *
+ * XXX I'd rename this to reorderbuffer_lz4 or something like that.
  *-------------------------------------------------------------------------
  */
 #include "postgres.h"
@@ -301,6 +303,8 @@ lz4_DecompressData(char *src, Size src_size, char **dst, Size dst_size)
 #endif
 }
 
+/* XXX The rest should be in reorderbuffer.c, with the rest of ReorderBuffer functions */
+
 /*
  * Allocate a new Compressor State, depending on the compression method.
  */
@@ -376,6 +380,12 @@ ReorderBufferCompress(ReorderBuffer *rb, ReorderBufferDiskHeader **header,
 		/* LZ4 Compression */
 		case REORDER_BUFFER_LZ4_COMPRESSION:
 		{
+			/*
+			 * XXX Won't this cause a lot of palloc/pfree traffic? We allocate a new
+			 * buffer for every compression, only to immediately throw it away. Look
+			 * at what astreamer_lz4_compressor_content does to reuse a buffer, maybe
+			 * we should do that here too?
+			 */
 			char	   *dst = NULL;
 			Size		dst_size = 0;
 			char	   *src = (char *) rb->outbuf + sizeof(ReorderBufferDiskHeader);
@@ -428,6 +438,12 @@ ReorderBufferCompress(ReorderBuffer *rb, ReorderBufferDiskHeader **header,
 
 /*
  * Decompress data read from disk and copy it into the ReorderBuffer.
+ *
+ * XXX Does it make sense to support both "regular" and "streaming" for lz4?
+ * Isn't one of those clearly better for this use case?
+ *
+ * XXX Also, we only ever create LZ4StreamingCompressorState, so does that
+ * work for non-streaming state?
  */
 void
 ReorderBufferDecompress(ReorderBuffer *rb, char *data,
@@ -479,7 +495,7 @@ ReorderBufferDecompress(ReorderBuffer *rb, char *data,
 		/* LZ4 streaming decompression */
 		case REORDER_BUFFER_STRAT_LZ4_STREAMING:
 			{
-				char	   *buf;
+				char	   *buf;	/* XXX shouldn't this be set to NULL explicitly? */
 				Size		src_size = header->size - sizeof(ReorderBufferDiskHeader);
 				Size		buf_size = header->raw_size;
 
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index f1562a77719..1a2ccda2eca 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -9,6 +9,7 @@
 #ifndef REORDERBUFFER_H
 #define REORDERBUFFER_H
 
+/* XXX we don't need any lz4 stuff in this header, right? */
 #ifdef USE_LZ4
 #include <lz4.h>
 #endif
@@ -30,6 +31,7 @@
 /* GUC variables */
 extern PGDLLIMPORT int logical_decoding_work_mem;
 extern PGDLLIMPORT int debug_logical_replication_streaming;
+extern PGDLLIMPORT int logical_decoding_spill_compression;
 
 /* possible values for debug_logical_replication_streaming */
 typedef enum
diff --git a/src/include/replication/reorderbuffer_compression.h b/src/include/replication/reorderbuffer_compression.h
index 9aa8aea56f4..be739d9ebeb 100644
--- a/src/include/replication/reorderbuffer_compression.h
+++ b/src/include/replication/reorderbuffer_compression.h
@@ -37,6 +37,14 @@ typedef enum ReorderBufferCompressionStrategy
 } ReorderBufferCompressionStrategy;
 
 /* Disk serialization support datastructures */
+
+/* XXX but what does ReorderBufferDiskHeader represent? what is it for?
+ * XXX Also, shouldn't it be ReorderBufferDiskChangeHeader? It's a header
+ * for a change, not for a disk.
+ * XXX I don't quite understand why we renamed this, DiskChange seems quite
+ * fine to me, even if it gets new fields for compression. Still, I'd keep
+ * it as private in reorderbuffer.c.
+ */
 typedef struct ReorderBufferDiskHeader
 {
 	ReorderBufferCompressionStrategy comp_strat; /* Compression strategy */
-- 
2.46.0

From 487eea83f36dc8731bb486dac1336b4598048471 Mon Sep 17 00:00:00 2001
From: Julien Tachoires <[email-masked]>
Date: Sun, 23 Jun 2024 14:42:04 -0700
Subject: [PATCH v4 03/11] Fix spill_bytes counter

The spill_bytes counter considers now the fact that decoded changes
are spilled on disk compressed.
---
 src/backend/replication/logical/reorderbuffer.c | 12 ++++++++----
 1 file changed, 8 insertions(+), 4 deletions(-)

diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index f5c51316d9d..19b72e90af8 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -252,7 +252,7 @@ static void ReorderBufferExecuteInvalidations(uint32 nmsgs, SharedInvalidationMe
  */
 static void ReorderBufferCheckMemoryLimit(ReorderBuffer *rb);
 static void ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn);
-static void ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
+static Size ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 										 int fd, ReorderBufferChange *change);
 static Size ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn,
 										TXNEntryFile *file, XLogSegNo *segno);
@@ -3719,6 +3719,7 @@ ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 	XLogSegNo	curOpenSegNo = 0;
 	Size		spilled = 0;
 	Size		size = txn->size;
+	Size		spillBytes = 0;
 
 	elog(DEBUG2, "spill %u changes in XID %u to disk",
 		 (uint32) txn->nentries_mem, txn->xid);
@@ -3770,7 +3771,7 @@ ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 						 errmsg("could not open file \"%s\": %m", path)));
 		}
 
-		ReorderBufferSerializeChange(rb, txn, fd, change);
+		spillBytes += ReorderBufferSerializeChange(rb, txn, fd, change);
 		dlist_delete(&change->node);
 		ReorderBufferReturnChange(rb, change, false);
 
@@ -3784,7 +3785,7 @@ ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 	if (spilled)
 	{
 		rb->spillCount += 1;
-		rb->spillBytes += size;
+		rb->spillBytes += spillBytes;
 
 		/* don't consider already serialized transactions */
 		rb->spillTxns += (rbtxn_is_serialized(txn) || rbtxn_is_serialized_clear(txn)) ? 0 : 1;
@@ -3805,7 +3806,7 @@ ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 /*
  * Serialize individual change to disk.
  */
-static void
+static Size
 ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 							 int fd, ReorderBufferChange *change)
 {
@@ -4018,6 +4019,9 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 	 */
 	if (txn->final_lsn < change->lsn)
 		txn->final_lsn = change->lsn;
+
+	/* Return data size written to disk */
+	return disk_hdr->size;
 }
 
 /* Returns true, if the output plugin supports streaming, false, otherwise. */
-- 
2.46.0

From 2e120c01baa97e82a62fd8c6936612f23dd4ac3c Mon Sep 17 00:00:00 2001
From: Tomas Vondra <[email-masked]>
Date: Tue, 17 Sep 2024 16:47:55 +0200
Subject: [PATCH v4 04/11] review

---
 src/backend/replication/logical/reorderbuffer.c | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 19b72e90af8..168956d944a 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -3805,6 +3805,14 @@ ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 
 /*
  * Serialize individual change to disk.
+ *
+ * Returns the amount of data written to disk (with compression, this is the
+ * size of the compressed representation).
+ *
+ * XXX But is this actually the right thing to do? Even without compression we
+ * don't really count the bytes written to the disk (we don't account for the
+ * DiskChange header), but rather the memory representation. So why should we
+ * do that with compression? Seems a bit strange.
  */
 static Size
 ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
-- 
2.46.0

From fbbc3693210041e3840e362e2892cdf94fbec709 Mon Sep 17 00:00:00 2001
From: Julien Tachoires <[email-masked]>
Date: Thu, 18 Jul 2024 07:36:04 -0700
Subject: [PATCH v4 05/11] Compress ReorderBuffer spill files using PGLZ

---
 .../logical/reorderbuffer_compression.c       | 58 +++++++++++++++++++
 .../replication/reorderbuffer_compression.h   |  2 +
 2 files changed, 60 insertions(+)

diff --git a/src/backend/replication/logical/reorderbuffer_compression.c b/src/backend/replication/logical/reorderbuffer_compression.c
index b4fea15ebd9..8d158aa51da 100644
--- a/src/backend/replication/logical/reorderbuffer_compression.c
+++ b/src/backend/replication/logical/reorderbuffer_compression.c
@@ -15,6 +15,8 @@
  */
 #include "postgres.h"
 
+#include "common/pg_lzcompress.h"
+
 #ifdef USE_LZ4
 #include <lz4.h>
 #endif
@@ -317,6 +319,7 @@ ReorderBufferNewCompressorState(MemoryContext context, int compression_method)
 			return lz4_NewCompressorState(context);
 			break;
 		case REORDER_BUFFER_NO_COMPRESSION:
+		case REORDER_BUFFER_PGLZ_COMPRESSION:
 		default:
 			return NULL;
 			break;
@@ -337,6 +340,7 @@ ReorderBufferFreeCompressorState(MemoryContext context, int compression_method,
 			return lz4_FreeCompressorState(context, compressor_state);
 			break;
 		case REORDER_BUFFER_NO_COMPRESSION:
+		case REORDER_BUFFER_PGLZ_COMPRESSION:
 		default:
 			break;
 	}
@@ -431,6 +435,40 @@ ReorderBufferCompress(ReorderBuffer *rb, ReorderBufferDiskHeader **header,
 
 			pfree(dst);
 
+			break;
+		}
+		/* PGLZ compression */
+		case REORDER_BUFFER_PGLZ_COMPRESSION:
+		{
+			int32		dst_size = 0;
+			char	   *dst = NULL;
+			char	   *src = (char *) rb->outbuf + sizeof(ReorderBufferDiskHeader);
+			int32		src_size = data_size - sizeof(ReorderBufferDiskHeader);
+			int32		max_size = PGLZ_MAX_OUTPUT(src_size);
+
+			dst = (char *) palloc0(max_size);
+			dst_size = pglz_compress(src, src_size, dst, PGLZ_strategy_always);
+
+			if (dst_size < 0)
+				ereport(ERROR,
+						(errcode(ERRCODE_DATA_CORRUPTED),
+						 errmsg_internal("PGLZ compression failed")));
+
+			ReorderBufferReserve(rb, (Size) (dst_size + sizeof(ReorderBufferDiskHeader)));
+
+			hdr = (ReorderBufferDiskHeader *) rb->outbuf;
+			hdr->comp_strat = REORDER_BUFFER_STRAT_PGLZ;
+			hdr->size = (Size) dst_size + sizeof(ReorderBufferDiskHeader);
+			hdr->raw_size = (Size) src_size;
+
+			*header = hdr;
+
+			/* Copy back compressed data into the ReorderBuffer */
+			memcpy((char *) rb->outbuf + sizeof(ReorderBufferDiskHeader), dst,
+				   dst_size);
+
+			pfree(dst);
+
 			break;
 		}
 	}
@@ -511,6 +549,26 @@ ReorderBufferDecompress(ReorderBuffer *rb, char *data,
 				 */
 				break;
 			}
+		/* PGLZ decompression */
+		case REORDER_BUFFER_STRAT_PGLZ:
+			{
+				char	   *buf;
+				int32		src_size = (int32) header->size - sizeof(ReorderBufferDiskHeader);
+				int32		buf_size = (int32) header->raw_size;
+				int32		decBytes;
+
+				/* Decompress data directly into the ReorderBuffer */
+				buf = (char *) rb->outbuf;
+				buf += sizeof(ReorderBufferDiskHeader);
+
+				decBytes = pglz_decompress(data, src_size, buf, buf_size, false);
+
+				if (decBytes < 0)
+					ereport(ERROR,
+							(errcode(ERRCODE_DATA_CORRUPTED),
+							 errmsg_internal("compressed PGLZ data is corrupted")));
+				break;
+			}
 		default:
 			/* Other compression methods not yet supported */
 			break;
diff --git a/src/include/replication/reorderbuffer_compression.h b/src/include/replication/reorderbuffer_compression.h
index be739d9ebeb..133513880e6 100644
--- a/src/include/replication/reorderbuffer_compression.h
+++ b/src/include/replication/reorderbuffer_compression.h
@@ -24,6 +24,7 @@ typedef enum ReorderBufferCompressionMethod
 {
 	REORDER_BUFFER_NO_COMPRESSION,
 	REORDER_BUFFER_LZ4_COMPRESSION,
+	REORDER_BUFFER_PGLZ_COMPRESSION,
 } ReorderBufferCompressionMethod;
 
 /*
@@ -34,6 +35,7 @@ typedef enum ReorderBufferCompressionStrategy
 	REORDER_BUFFER_STRAT_UNCOMPRESSED,
 	REORDER_BUFFER_STRAT_LZ4_STREAMING,
 	REORDER_BUFFER_STRAT_LZ4_REGULAR,
+	REORDER_BUFFER_STRAT_PGLZ,
 } ReorderBufferCompressionStrategy;
 
 /* Disk serialization support datastructures */
-- 
2.46.0

From 8e243868ce9d6254480a239012bae730f4db77a9 Mon Sep 17 00:00:00 2001
From: Tomas Vondra <[email-masked]>
Date: Tue, 17 Sep 2024 17:05:10 +0200
Subject: [PATCH v4 06/11] review

---
 .../replication/logical/reorderbuffer_compression.c       | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/src/backend/replication/logical/reorderbuffer_compression.c b/src/backend/replication/logical/reorderbuffer_compression.c
index 8d158aa51da..e8785d27e49 100644
--- a/src/backend/replication/logical/reorderbuffer_compression.c
+++ b/src/backend/replication/logical/reorderbuffer_compression.c
@@ -438,8 +438,16 @@ ReorderBufferCompress(ReorderBuffer *rb, ReorderBufferDiskHeader **header,
 			break;
 		}
 		/* PGLZ compression */
+		/*
+		 * XXX I'd probably start by adding pglz first, and only then add lz4 as a
+		 * separate patch, simply because pglz is the default and always supported,
+		 * while lz4 requires extra flags.
+		 */
 		case REORDER_BUFFER_PGLZ_COMPRESSION:
 		{
+			/*
+			 * XXX Same comment about palloc/pfree traffic as for lz4 ...
+			 */
 			int32		dst_size = 0;
 			char	   *dst = NULL;
 			char	   *src = (char *) rb->outbuf + sizeof(ReorderBufferDiskHeader);
-- 
2.46.0

From 8e55c5bf35de74c9a9bf09e0cd3c16c3599dabd5 Mon Sep 17 00:00:00 2001
From: Julien Tachoires <[email-masked]>
Date: Thu, 18 Jul 2024 07:39:25 -0700
Subject: [PATCH v4 07/11] Compress ReorderBuffer spill files using ZSTD

---
 .../logical/reorderbuffer_compression.c       | 364 ++++++++++++++++++
 .../replication/reorderbuffer_compression.h   |  39 ++
 2 files changed, 403 insertions(+)

diff --git a/src/backend/replication/logical/reorderbuffer_compression.c b/src/backend/replication/logical/reorderbuffer_compression.c
index e8785d27e49..96751852ef8 100644
--- a/src/backend/replication/logical/reorderbuffer_compression.c
+++ b/src/backend/replication/logical/reorderbuffer_compression.c
@@ -21,6 +21,10 @@
 #include <lz4.h>
 #endif
 
+#ifdef USE_ZSTD
+#include <zstd.h>
+#endif
+
 #include "replication/reorderbuffer_compression.h"
 
 #define NO_LZ4_SUPPORT() \
@@ -29,6 +33,12 @@
 			 errmsg("compression method lz4 not supported"), \
 			 errdetail("This functionality requires the server to be built with lz4 support.")))
 
+#define NO_ZSTD_SUPPORT() \
+	ereport(ERROR, \
+			(errcode(ERRCODE_FEATURE_NOT_SUPPORTED), \
+			 errmsg("compression method zstd not supported"), \
+			 errdetail("This functionality requires the server to be built with zstd support.")))
+
 /*
  * Allocate a new LZ4StreamingCompressorState.
  */
@@ -307,6 +317,309 @@ lz4_DecompressData(char *src, Size src_size, char **dst, Size dst_size)
 
 /* XXX The rest should be in reorderbuffer.c, with the rest of ReorderBuffer functions */
 
+/*
+ * Allocate a new ZSTDStreamingCompressorState.
+ */
+static void *
+zstd_NewCompressorState(MemoryContext context)
+{
+#ifndef USE_ZSTD
+	NO_ZSTD_SUPPORT();
+	return NULL;				/* keep compiler quiet */
+#else
+	ZSTDStreamingCompressorState *cstate;
+
+	cstate = (ZSTDStreamingCompressorState *)
+		MemoryContextAlloc(context, sizeof(ZSTDStreamingCompressorState));
+
+	/*
+	 * We do not allocate ZSTD buffers and contexts at this point because we
+	 * have no guarantee that we will need them later. Let's allocate only when
+	 * we are about to use them.
+	 */
+	cstate->zstd_c_ctx = NULL;
+	cstate->zstd_c_in_buf = NULL;
+	cstate->zstd_c_in_buf_size = 0;
+	cstate->zstd_c_out_buf = NULL;
+	cstate->zstd_c_out_buf_size = 0;
+	cstate->zstd_frame_size = 0;
+	cstate->zstd_d_ctx = NULL;
+	cstate->zstd_d_in_buf = NULL;
+	cstate->zstd_d_in_buf_size = 0;
+	cstate->zstd_d_out_buf = NULL;
+	cstate->zstd_d_out_buf_size = 0;
+
+	return (void *) cstate;
+#endif
+}
+
+/*
+ * Free ZSTD memory resources and the compressor state.
+ */
+static void
+zstd_FreeCompressorState(MemoryContext context, void *compressor_state)
+{
+#ifndef USE_ZSTD
+	NO_ZSTD_SUPPORT();
+#else
+	ZSTDStreamingCompressorState *cstate;
+	MemoryContext oldcontext;
+
+	if (compressor_state == NULL)
+		return;
+
+	oldcontext = MemoryContextSwitchTo(context);
+
+	cstate = (ZSTDStreamingCompressorState *) compressor_state;
+
+	if (cstate->zstd_c_ctx != NULL)
+	{
+		/* Compressor state was used for compression */
+		pfree(cstate->zstd_c_in_buf);
+		pfree(cstate->zstd_c_out_buf);
+		ZSTD_freeCCtx(cstate->zstd_c_ctx);
+	}
+	if (cstate->zstd_d_ctx != NULL)
+	{
+		/* Compressor state was used for decompression */
+		pfree(cstate->zstd_d_in_buf);
+		pfree(cstate->zstd_d_out_buf);
+		ZSTD_freeDCtx(cstate->zstd_d_ctx);
+	}
+
+	pfree(compressor_state);
+
+	MemoryContextSwitchTo(oldcontext);
+#endif
+}
+
+#ifdef USE_ZSTD
+/*
+ * Allocate ZSTD compression buffers and create the ZSTD compression context.
+ */
+static void
+zstd_CreateStreamCompressorState(MemoryContext context, void *compressor_state)
+{
+	ZSTDStreamingCompressorState *cstate;
+	MemoryContext oldcontext = MemoryContextSwitchTo(context);
+
+	cstate = (ZSTDStreamingCompressorState *) compressor_state;
+	cstate->zstd_c_in_buf_size = ZSTD_CStreamInSize();
+	cstate->zstd_c_in_buf = (char *) palloc0(cstate->zstd_c_in_buf_size);
+	cstate->zstd_c_out_buf_size = ZSTD_CStreamOutSize();
+	cstate->zstd_c_out_buf = (char *) palloc0(cstate->zstd_c_out_buf_size);
+	cstate->zstd_c_ctx = ZSTD_createCCtx();
+
+	if (cstate->zstd_c_ctx == NULL)
+		ereport(ERROR,
+				(errcode(ERRCODE_OUT_OF_MEMORY),
+				 errmsg("could not create ZSTD compression context")));
+
+	/* Set compression level */
+	ZSTD_CCtx_setParameter(cstate->zstd_c_ctx, ZSTD_c_compressionLevel,
+						   ZSTD_COMPRESSION_LEVEL);
+
+	MemoryContextSwitchTo(oldcontext);
+}
+#endif
+
+#ifdef USE_ZSTD
+/*
+ * Allocate ZSTD decompression buffers and create the ZSTD decompression
+ * context.
+ */
+static void
+zstd_CreateStreamDecodeCompressorState(MemoryContext context, void *compressor_state)
+{
+	ZSTDStreamingCompressorState *cstate;
+	MemoryContext oldcontext = MemoryContextSwitchTo(context);
+
+	cstate = (ZSTDStreamingCompressorState *) compressor_state;
+	cstate->zstd_d_in_buf_size = ZSTD_DStreamInSize();
+	cstate->zstd_d_in_buf = (char *) palloc0(cstate->zstd_d_in_buf_size);
+	cstate->zstd_d_out_buf_size = ZSTD_DStreamOutSize();
+	cstate->zstd_d_out_buf = (char *) palloc0(cstate->zstd_d_out_buf_size);
+	cstate->zstd_d_ctx = ZSTD_createDCtx();
+
+	if (cstate->zstd_d_ctx == NULL)
+		ereport(ERROR,
+				(errcode(ERRCODE_OUT_OF_MEMORY),
+				 errmsg("could not create ZSTD decompression context")));
+
+	MemoryContextSwitchTo(oldcontext);
+}
+#endif
+
+/*
+ * Data compression using ZSTD streaming API.
+ */
+static void
+zstd_StreamingCompressData(MemoryContext context, char *src, Size src_size,
+						   char **dst, Size *dst_size, void *compressor_state)
+{
+#ifndef USE_ZSTD
+	NO_ZSTD_SUPPORT();
+#else
+	ZSTDStreamingCompressorState *cstate;
+	/* Size of remaining data to be copied from src into ZSTD input buffer */
+	Size		toCpy = src_size;
+	char	   *dst_data;
+
+	cstate = (ZSTDStreamingCompressorState *) compressor_state;
+	/* Allocate ZSTD buffers and context */
+	if (cstate->zstd_c_ctx == NULL)
+		zstd_CreateStreamCompressorState(context, compressor_state);
+
+	/* Allocate memory that will be used to store compressed data */
+	*dst = (char *) palloc0(ZSTD_compressBound(src_size));
+
+	dst_data = *dst;
+	*dst_size = 0;
+
+	/*
+	 * ZSTD streaming compression works with chunks: the source data needs to
+	 * be splitted out in chunks, each of them is then copied into ZSTD input
+	 * buffer.
+	 * For each chunk, we proceed with compression. Streaming compression is
+	 * not intended to compress the whole input chunk, so we have the call
+	 * ZSTD_compressStream2() multiple times until the entire chunk is
+	 * consumed.
+	 */
+	while (toCpy > 0)
+	{
+		/* Are we on the last chunk? */
+		bool		last_chunk = (toCpy < cstate->zstd_c_in_buf_size);
+		/* Size of the data copied into ZSTD input buffer */
+		Size		cpySize = last_chunk ? toCpy : cstate->zstd_c_in_buf_size;
+		bool		finished = false;
+		ZSTD_inBuffer input;
+		ZSTD_EndDirective mode = last_chunk ? ZSTD_e_flush : ZSTD_e_continue;
+
+		/* Copy data from src into ZSTD input buffer */
+		memcpy(cstate->zstd_c_in_buf, src, cpySize);
+
+		/*
+		 * Close the frame when we are on the last chunk and we've reached max
+		 * frame size.
+		 */
+		if (last_chunk && (cstate->zstd_frame_size > ZSTD_MAX_FRAME_SIZE))
+		{
+			mode = ZSTD_e_end;
+			cstate->zstd_frame_size = 0;
+		}
+
+		cstate->zstd_frame_size += cpySize;
+
+		input.src = cstate->zstd_c_in_buf;
+		input.size = cpySize;
+		input.pos = 0;
+
+		do
+		{
+			Size		remaining;
+			ZSTD_outBuffer output;
+
+			output.dst = cstate->zstd_c_out_buf;
+			output.size = cstate->zstd_c_out_buf_size;
+			output.pos = 0;
+
+			remaining = ZSTD_compressStream2(cstate->zstd_c_ctx, &output,
+											 &input, mode);
+
+			if (ZSTD_isError(remaining))
+				ereport(ERROR,
+						(errcode(ERRCODE_DATA_CORRUPTED),
+						 errmsg_internal("ZSTD compression failed")));
+
+			/* Copy back compressed data from ZSTD output buffer */
+			memcpy(dst_data, (char *) cstate->zstd_c_out_buf, output.pos);
+
+			dst_data += output.pos;
+			*dst_size += output.pos;
+
+			/*
+			 * Compression is done when we are working on the last chunk and
+			 * there is nothing left to compress, or, when we reach the end of
+			 * the chunk.
+			 */
+			finished = last_chunk ? (remaining == 0) : (input.pos == input.size);
+		} while (!finished);
+
+		src += cpySize;
+		toCpy -= cpySize;
+	}
+#endif
+}
+
+/*
+ * Data decompression using ZSTD streaming API.
+ */
+static void
+zstd_StreamingDecompressData(MemoryContext context, char *src, Size src_size,
+							char **dst, Size dst_size, void *compressor_state)
+{
+#ifndef USE_ZSTD
+	NO_ZSTD_SUPPORT();
+#else
+	ZSTDStreamingCompressorState *cstate;
+	/* Size of remaining data to be copied from src into ZSTD input buffer */
+	Size		toCpy = src_size;
+	char	   *dst_data;
+	Size		decBytes = 0;	/* Size of decompressed data */
+
+	cstate = (ZSTDStreamingCompressorState *) compressor_state;
+	/* Allocate ZSTD buffers and context */
+	if (cstate->zstd_d_ctx == NULL)
+		zstd_CreateStreamDecodeCompressorState(context, compressor_state);
+
+	/* Allocate memory that will be used to store decompressed data */
+	*dst = (char *) palloc0(dst_size);
+
+	dst_data = *dst;
+
+	while (toCpy > 0)
+	{
+		ZSTD_inBuffer input;
+		Size		cpySize = (toCpy > cstate->zstd_d_in_buf_size) ? cstate->zstd_d_in_buf_size : toCpy;
+
+		/* Copy data from src into ZSTD input buffer */
+		memcpy(cstate->zstd_d_in_buf, src, cpySize);
+
+		input.src = cstate->zstd_d_in_buf;
+		input.size = cpySize;
+		input.pos = 0;
+
+		while (input.pos < input.size)
+		{
+			ZSTD_outBuffer output;
+			Size		ret;
+
+			output.dst = cstate->zstd_d_out_buf;
+			output.size = cstate->zstd_d_out_buf_size;
+			output.pos = 0;
+
+			ret = ZSTD_decompressStream(cstate->zstd_d_ctx, &output , &input);
+
+			if (ZSTD_isError(ret))
+				ereport(ERROR,
+						(errcode(ERRCODE_DATA_CORRUPTED),
+						 errmsg_internal("ZSTD decompression failed")));
+
+			/* Copy back compressed data from ZSTD output buffer */
+			memcpy(dst_data, (char *) cstate->zstd_d_out_buf, output.pos);
+
+			dst_data += output.pos;
+			decBytes += output.pos;
+		}
+
+		src += cpySize;
+		toCpy -= cpySize;
+	}
+
+	Assert(dst_size == decBytes);
+#endif
+}
+
 /*
  * Allocate a new Compressor State, depending on the compression method.
  */
@@ -318,6 +631,9 @@ ReorderBufferNewCompressorState(MemoryContext context, int compression_method)
 		case REORDER_BUFFER_LZ4_COMPRESSION:
 			return lz4_NewCompressorState(context);
 			break;
+		case REORDER_BUFFER_ZSTD_COMPRESSION:
+			return zstd_NewCompressorState(context);
+			break;
 		case REORDER_BUFFER_NO_COMPRESSION:
 		case REORDER_BUFFER_PGLZ_COMPRESSION:
 		default:
@@ -339,6 +655,9 @@ ReorderBufferFreeCompressorState(MemoryContext context, int compression_method,
 		case REORDER_BUFFER_LZ4_COMPRESSION:
 			return lz4_FreeCompressorState(context, compressor_state);
 			break;
+		case REORDER_BUFFER_ZSTD_COMPRESSION:
+			return zstd_FreeCompressorState(context, compressor_state);
+			break;
 		case REORDER_BUFFER_NO_COMPRESSION:
 		case REORDER_BUFFER_PGLZ_COMPRESSION:
 		default:
@@ -477,6 +796,35 @@ ReorderBufferCompress(ReorderBuffer *rb, ReorderBufferDiskHeader **header,
 
 			pfree(dst);
 
+			break;
+		}
+		/* ZSTD Compression */
+		case REORDER_BUFFER_ZSTD_COMPRESSION:
+		{
+			char	   *dst = NULL;
+			Size		dst_size = 0;
+			char	   *src = (char *) rb->outbuf + sizeof(ReorderBufferDiskHeader);
+			Size		src_size = data_size - sizeof(ReorderBufferDiskHeader);
+
+			/* Use ZSTD streaming compression */
+			zstd_StreamingCompressData(rb->context, src, src_size, &dst,
+									   &dst_size, compressor_state);
+
+			ReorderBufferReserve(rb, (dst_size + sizeof(ReorderBufferDiskHeader)));
+
+			hdr = (ReorderBufferDiskHeader *) rb->outbuf;
+			hdr->comp_strat = REORDER_BUFFER_STRAT_ZSTD_STREAMING;
+			hdr->size = dst_size + sizeof(ReorderBufferDiskHeader);
+			hdr->raw_size = src_size;
+
+			*header = hdr;
+
+			/* Copy back compressed data into the ReorderBuffer */
+			memcpy((char *) rb->outbuf + sizeof(ReorderBufferDiskHeader), dst,
+				   dst_size);
+
+			pfree(dst);
+
 			break;
 		}
 	}
@@ -577,6 +925,22 @@ ReorderBufferDecompress(ReorderBuffer *rb, char *data,
 							 errmsg_internal("compressed PGLZ data is corrupted")));
 				break;
 			}
+		/* ZSTD streaming decompression */
+		case REORDER_BUFFER_STRAT_ZSTD_STREAMING:
+			{
+				char	   *buf;
+				Size		src_size = header->size - sizeof(ReorderBufferDiskHeader);
+				Size		buf_size = header->raw_size;
+
+				zstd_StreamingDecompressData(rb->context, data, src_size, &buf,
+											 buf_size, compressor_state);
+
+				/* Copy decompressed data into the ReorderBuffer */
+				memcpy((char *) rb->outbuf + sizeof(ReorderBufferDiskHeader),
+					   buf, buf_size);
+				pfree(buf);
+				break;
+			}
 		default:
 			/* Other compression methods not yet supported */
 			break;
diff --git a/src/include/replication/reorderbuffer_compression.h b/src/include/replication/reorderbuffer_compression.h
index 133513880e6..960064d571e 100644
--- a/src/include/replication/reorderbuffer_compression.h
+++ b/src/include/replication/reorderbuffer_compression.h
@@ -19,12 +19,17 @@
 #include <lz4.h>
 #endif
 
+#ifdef USE_ZSTD
+#include <zstd.h>
+#endif
+
 /* ReorderBuffer on disk compression algorithms */
 typedef enum ReorderBufferCompressionMethod
 {
 	REORDER_BUFFER_NO_COMPRESSION,
 	REORDER_BUFFER_LZ4_COMPRESSION,
 	REORDER_BUFFER_PGLZ_COMPRESSION,
+	REORDER_BUFFER_ZSTD_COMPRESSION,
 } ReorderBufferCompressionMethod;
 
 /*
@@ -36,6 +41,7 @@ typedef enum ReorderBufferCompressionStrategy
 	REORDER_BUFFER_STRAT_LZ4_STREAMING,
 	REORDER_BUFFER_STRAT_LZ4_REGULAR,
 	REORDER_BUFFER_STRAT_PGLZ,
+	REORDER_BUFFER_STRAT_ZSTD_STREAMING,
 } ReorderBufferCompressionStrategy;
 
 /* Disk serialization support datastructures */
@@ -89,6 +95,39 @@ typedef struct LZ4StreamingCompressorState {
 #define lz4_CanDoStreamingCompression(s) (false)
 #endif
 
+#ifdef USE_ZSTD
+/*
+ * Low compression level provides high compression speed and decent compression
+ * rate. Minimum level is 1, maximum is 22.
+ */
+#define ZSTD_COMPRESSION_LEVEL 1
+
+/*
+ * Maximum volume of data encoded in the current ZSTD frame. When this
+ * threshold is reached then we close the current frame and start a new one.
+ */
+#define ZSTD_MAX_FRAME_SIZE (64 * 1024)
+
+/*
+ * ZSTD streaming compression/decompression handlers and buffers.
+ */
+typedef struct ZSTDStreamingCompressorState {
+	/* Compression */
+	ZSTD_CCtx  *zstd_c_ctx;
+	Size		zstd_c_in_buf_size;
+	char	   *zstd_c_in_buf;
+	Size		zstd_c_out_buf_size;
+	char	   *zstd_c_out_buf;
+	Size		zstd_frame_size;
+	/* Decompression */
+	ZSTD_DCtx  *zstd_d_ctx;
+	Size		zstd_d_in_buf_size;
+	char	   *zstd_d_in_buf;
+	Size		zstd_d_out_buf_size;
+	char	   *zstd_d_out_buf;
+} ZSTDStreamingCompressorState;
+#endif
+
 extern void *ReorderBufferNewCompressorState(MemoryContext context,
 											 int compression_method);
 extern void ReorderBufferFreeCompressorState(MemoryContext context,
-- 
2.46.0

From 7cbf94f473454987410373b6ca78e44c06b5fe08 Mon Sep 17 00:00:00 2001
From: Tomas Vondra <[email-masked]>
Date: Tue, 17 Sep 2024 17:08:35 +0200
Subject: [PATCH v4 08/11] review

---
 src/backend/replication/logical/reorderbuffer_compression.c | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/src/backend/replication/logical/reorderbuffer_compression.c b/src/backend/replication/logical/reorderbuffer_compression.c
index 96751852ef8..67068db9d2c 100644
--- a/src/backend/replication/logical/reorderbuffer_compression.c
+++ b/src/backend/replication/logical/reorderbuffer_compression.c
@@ -801,6 +801,9 @@ ReorderBufferCompress(ReorderBuffer *rb, ReorderBufferDiskHeader **header,
 		/* ZSTD Compression */
 		case REORDER_BUFFER_ZSTD_COMPRESSION:
 		{
+			/*
+			 * XXX Same comment about palloc/pfree traffic as for lz4 ...
+			 */
 			char	   *dst = NULL;
 			Size		dst_size = 0;
 			char	   *src = (char *) rb->outbuf + sizeof(ReorderBufferDiskHeader);
@@ -928,7 +931,7 @@ ReorderBufferDecompress(ReorderBuffer *rb, char *data,
 		/* ZSTD streaming decompression */
 		case REORDER_BUFFER_STRAT_ZSTD_STREAMING:
 			{
-				char	   *buf;
+				char	   *buf;	/* XXX shouldn't this be set to NULL explicitly? */
 				Size		src_size = header->size - sizeof(ReorderBufferDiskHeader);
 				Size		buf_size = header->raw_size;
 
-- 
2.46.0

From 406e38977b0dd868b20ac31107c8bfbd89e5ad0d Mon Sep 17 00:00:00 2001
From: Tomas Vondra <[email-masked]>
Date: Tue, 17 Sep 2024 15:30:53 +0200
Subject: [PATCH v4 09/11] Add the subscription option spill_compression

---
 doc/src/sgml/ref/create_subscription.sgml     |  24 +++
 src/backend/catalog/pg_subscription.c         |   6 +
 src/backend/catalog/system_views.sql          |   3 +-
 src/backend/commands/subscriptioncmds.c       |  31 +++-
 .../libpqwalreceiver/libpqwalreceiver.c       |   5 +
 src/backend/replication/logical/logical.c     |   4 +
 .../replication/logical/reorderbuffer.c       |  15 +-
 .../logical/reorderbuffer_compression.c       |  57 +++++++
 src/backend/replication/logical/worker.c      |  13 +-
 src/backend/replication/pgoutput/pgoutput.c   |  28 ++++
 src/bin/pg_dump/pg_dump.c                     |  18 +-
 src/bin/pg_dump/pg_dump.h                     |   1 +
 src/bin/pg_dump/t/002_pg_dump.pl              |   8 +-
 src/bin/psql/describe.c                       |   7 +-
 src/bin/psql/tab-complete.c                   |   5 +-
 src/include/catalog/pg_subscription.h         |   4 +
 src/include/replication/logical.h             |   2 +
 src/include/replication/pgoutput.h            |   1 +
 .../replication/reorderbuffer_compression.h   |   4 +
 src/include/replication/walreceiver.h         |   1 +
 src/test/regress/expected/subscription.out    | 156 +++++++++---------
 src/test/regress/sql/subscription.sql         |   4 +
 22 files changed, 305 insertions(+), 92 deletions(-)

diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 740b7d94210..56f733eaf8d 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -428,6 +428,30 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
          </para>
         </listitem>
        </varlistentry>
+
+       <varlistentry id="sql-createsubscription-params-with-spill-compression">
+        <term><literal>spill_compression</literal> (<type>enum</type>)</term>
+        <listitem>
+         <para>
+          Specifies whether the decoded changes that eventually need to be
+          temporarily written on disk by the publisher are compressed or not.
+          Default value is <literal>off</literal> meaning no data compression
+          involved. Setting <literal>spill_compression</literal> to
+          <literal>on</literal> or <literal>pglz</literal> means that the
+          decoded changes are compressed using the internal
+          <literal>PGLZ</literal> compression algorithm.
+         </para>
+
+         <para>
+          If the <productname>PostgreSQL</productname> server running the
+          publisher node supports the external compression libraries
+          <productname>LZ4</productname> or
+          <productname>Zstandard</productname>,
+          <literal>spill_compression</literal> can be set respectively to
+          <literal>lz4</literal> or <literal>zstd</literal>.
+         </para>
+        </listitem>
+       </varlistentry>
       </variablelist></para>
 
     </listitem>
diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index 9efc9159f2c..a3329043dca 100644
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -110,6 +110,12 @@ GetSubscription(Oid subid, bool missing_ok)
 	/* Is the subscription owner a superuser? */
 	sub->ownersuperuser = superuser_arg(sub->owner);
 
+	/* Get spill_compression */
+	datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
+								   tup,
+								   Anum_pg_subscription_subspillcompression);
+	sub->spill_compression = TextDatumGetCString(datum);
+
 	ReleaseSysCache(tup);
 
 	return sub;
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 7fd5d256a18..40768a9ae9d 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1356,7 +1356,8 @@ REVOKE ALL ON pg_subscription FROM public;
 GRANT SELECT (oid, subdbid, subskiplsn, subname, subowner, subenabled,
               subbinary, substream, subtwophasestate, subdisableonerr,
 			  subpasswordrequired, subrunasowner, subfailover,
-              subslotname, subsynccommit, subpublications, suborigin)
+              subslotname, subsynccommit, subpublications, suborigin,
+              subspillcompression)
     ON pg_subscription TO public;
 
 CREATE VIEW pg_stat_subscription_stats AS
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 02ccc636b80..018f42494d0 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -40,6 +40,7 @@
 #include "replication/logicallauncher.h"
 #include "replication/logicalworker.h"
 #include "replication/origin.h"
+#include "replication/reorderbuffer_compression.h"
 #include "replication/slot.h"
 #include "replication/walreceiver.h"
 #include "replication/walsender.h"
@@ -73,6 +74,7 @@
 #define SUBOPT_FAILOVER				0x00002000
 #define SUBOPT_LSN					0x00004000
 #define SUBOPT_ORIGIN				0x00008000
+#define SUBOPT_SPILL_COMPRESSION	0x00010000
 
 /* check if the 'val' has 'bits' set */
 #define IsSet(val, bits)  (((val) & (bits)) == (bits))
@@ -100,6 +102,7 @@ typedef struct SubOpts
 	bool		failover;
 	char	   *origin;
 	XLogRecPtr	lsn;
+	char	   *spill_compression;
 } SubOpts;
 
 static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
@@ -164,6 +167,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
 		opts->failover = false;
 	if (IsSet(supported_opts, SUBOPT_ORIGIN))
 		opts->origin = pstrdup(LOGICALREP_ORIGIN_ANY);
+	if (IsSet(supported_opts, SUBOPT_SPILL_COMPRESSION))
+		opts->spill_compression = "off";
 
 	/* Parse options */
 	foreach(lc, stmt_options)
@@ -357,6 +362,18 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
 			opts->specified_opts |= SUBOPT_LSN;
 			opts->lsn = lsn;
 		}
+		else if (IsSet(supported_opts, SUBOPT_SPILL_COMPRESSION) &&
+				 strcmp(defel->defname, "spill_compression") == 0)
+		{
+			if (IsSet(opts->specified_opts, SUBOPT_SPILL_COMPRESSION))
+				errorConflictingDefElem(defel, pstate);
+
+			opts->specified_opts |= SUBOPT_SPILL_COMPRESSION;
+			opts->spill_compression = defGetString(defel);
+
+			ReorderBufferValidateCompressionMethod(opts->spill_compression,
+												   ERROR);
+		}
 		else
 			ereport(ERROR,
 					(errcode(ERRCODE_SYNTAX_ERROR),
@@ -594,7 +611,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 					  SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
 					  SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
 					  SUBOPT_DISABLE_ON_ERR | SUBOPT_PASSWORD_REQUIRED |
-					  SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER | SUBOPT_ORIGIN);
+					  SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER | SUBOPT_ORIGIN |
+					  SUBOPT_SPILL_COMPRESSION);
 	parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
 
 	/*
@@ -714,6 +732,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 		publicationListToArray(publications);
 	values[Anum_pg_subscription_suborigin - 1] =
 		CStringGetTextDatum(opts.origin);
+	values[Anum_pg_subscription_subspillcompression - 1] =
+		CStringGetTextDatum(opts.spill_compression);
 
 	tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
 
@@ -1196,7 +1216,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 								  SUBOPT_DISABLE_ON_ERR |
 								  SUBOPT_PASSWORD_REQUIRED |
 								  SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
-								  SUBOPT_ORIGIN);
+								  SUBOPT_ORIGIN | SUBOPT_SPILL_COMPRESSION);
 
 				parse_subscription_options(pstate, stmt->options,
 										   supported_opts, &opts);
@@ -1363,6 +1383,13 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 					replaces[Anum_pg_subscription_suborigin - 1] = true;
 				}
 
+				if (IsSet(opts.specified_opts, SUBOPT_SPILL_COMPRESSION))
+				{
+					values[Anum_pg_subscription_subspillcompression - 1] =
+						CStringGetTextDatum(opts.spill_compression);
+					replaces[Anum_pg_subscription_subspillcompression - 1] = true;
+				}
+
 				update_tuple = true;
 				break;
 			}
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 97f957cd87b..20f5e4a83e8 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -620,6 +620,11 @@ libpqrcv_startstreaming(WalReceiverConn *conn,
 			PQserverVersion(conn->streamConn) >= 140000)
 			appendStringInfoString(&cmd, ", binary 'true'");
 
+		if (options->proto.logical.spill_compression &&
+			PQserverVersion(conn->streamConn) >= 180000)
+			appendStringInfo(&cmd, ", spill_compression '%s'",
+							 options->proto.logical.spill_compression);
+
 		appendStringInfoChar(&cmd, ')');
 	}
 	else
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 3fe1774a1e9..54fbbe6fea6 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -36,6 +36,7 @@
 #include "replication/decode.h"
 #include "replication/logical.h"
 #include "replication/reorderbuffer.h"
+#include "replication/reorderbuffer_compression.h"
 #include "replication/slotsync.h"
 #include "replication/snapbuild.h"
 #include "storage/proc.h"
@@ -298,6 +299,9 @@ StartupDecodingContext(List *output_plugin_options,
 
 	ctx->fast_forward = fast_forward;
 
+	/* No spill files compression by default */
+	ctx->spill_compression_method = REORDER_BUFFER_NO_COMPRESSION;
+
 	MemoryContextSwitchTo(old_context);
 
 	return ctx;
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 168956d944a..6545c310f7f 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -410,6 +410,15 @@ ReorderBufferFree(ReorderBuffer *rb)
 	ReorderBufferCleanupSerializedTXNs(NameStr(MyReplicationSlot->data.name));
 }
 
+/* Returns spill files compression method */
+static inline uint8
+ReorderBufferSpillCompressionMethod(ReorderBuffer *rb)
+{
+	LogicalDecodingContext *ctx = rb->private_data;
+
+	return ctx->spill_compression_method;
+}
+
 /*
  * Get an unused, possibly preallocated, ReorderBufferTXN.
  */
@@ -431,7 +440,7 @@ ReorderBufferGetTXN(ReorderBuffer *rb)
 	txn->command_id = InvalidCommandId;
 	txn->output_plugin_private = NULL;
 	txn->compressor_state = ReorderBufferNewCompressorState(rb->context,
-															logical_decoding_spill_compression);
+															ReorderBufferSpillCompressionMethod(rb));
 
 	return txn;
 }
@@ -470,7 +479,7 @@ ReorderBufferReturnTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 	}
 
 	ReorderBufferFreeCompressorState(rb->context,
-									 logical_decoding_spill_compression,
+									 ReorderBufferSpillCompressionMethod(rb),
 									 txn->compressor_state);
 
 	/* Reset the toast hash */
@@ -3996,7 +4005,7 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 	}
 
 	/* Inplace ReorderBuffer content compression before writing it on disk */
-	ReorderBufferCompress(rb, &disk_hdr, logical_decoding_spill_compression,
+	ReorderBufferCompress(rb, &disk_hdr, ReorderBufferSpillCompressionMethod(rb),
 						  sz, txn->compressor_state);
 
 	errno = 0;
diff --git a/src/backend/replication/logical/reorderbuffer_compression.c b/src/backend/replication/logical/reorderbuffer_compression.c
index 67068db9d2c..707f35cff7c 100644
--- a/src/backend/replication/logical/reorderbuffer_compression.c
+++ b/src/backend/replication/logical/reorderbuffer_compression.c
@@ -949,3 +949,60 @@ ReorderBufferDecompress(ReorderBuffer *rb, char *data,
 			break;
 	}
 }
+
+/*
+ * According to a given compression method (as string representation), returns
+ * the corresponding ReorderBufferCompressionMethod
+ */
+ReorderBufferCompressionMethod
+ReorderBufferParseCompressionMethod(const char *method)
+{
+	if (pg_strcasecmp(method, "on") == 0)
+		return REORDER_BUFFER_PGLZ_COMPRESSION;
+	else if (pg_strcasecmp(method, "pglz") == 0)
+		return REORDER_BUFFER_PGLZ_COMPRESSION;
+	else if (pg_strcasecmp(method, "off") == 0)
+		return REORDER_BUFFER_NO_COMPRESSION;
+#ifdef USE_LZ4
+	else if (pg_strcasecmp(method, "lz4") == 0)
+		return REORDER_BUFFER_LZ4_COMPRESSION;
+#endif
+#ifdef USE_ZSTD
+	else if (pg_strcasecmp(method, "zstd") == 0)
+		return REORDER_BUFFER_ZSTD_COMPRESSION;
+#endif
+	else
+		return REORDER_BUFFER_INVALID_COMPRESSION;
+}
+
+/*
+ * Check whether the passed compression method is valid and report errors at
+ * elevel.
+ *
+ * As this validation is intended to be executed on subscriber side, then we
+ * actually don't know if the server running the publisher supports external
+ * compression libraries. We only check if the compression method is
+ * potentially supported. The real validation is done by the publisher when
+ * the replication starts, an error is then triggered if the compression method
+ * is not supported.
+ */
+void
+ReorderBufferValidateCompressionMethod(const char *method, int elevel)
+{
+	bool		valid = false;
+	char		methods[5][5] = {"on", "off", "pglz", "lz4", "zstd"};
+
+	for (int i = 0; i < 5; i++)
+	{
+		if (pg_strcasecmp(method, methods[i]) == 0)
+		{
+			valid = true;
+			break;
+		}
+	}
+
+	if (!valid)
+		ereport(elevel,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("compression method \"%s\" not valid", method)));
+}
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 925dff9cc44..eb42e844f99 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -4021,7 +4021,8 @@ maybe_reread_subscription(void)
 		newsub->passwordrequired != MySubscription->passwordrequired ||
 		strcmp(newsub->origin, MySubscription->origin) != 0 ||
 		newsub->owner != MySubscription->owner ||
-		!equal(newsub->publications, MySubscription->publications))
+		!equal(newsub->publications, MySubscription->publications) ||
+		strcmp(newsub->spill_compression, MySubscription->spill_compression) != 0)
 	{
 		if (am_parallel_apply_worker())
 			ereport(LOG,
@@ -4469,6 +4470,16 @@ set_stream_options(WalRcvStreamOptions *options,
 		MyLogicalRepWorker->parallel_apply = false;
 	}
 
+	if (server_version >= 180000 &&
+			 MySubscription->stream == LOGICALREP_STREAM_OFF &&
+			 MySubscription->spill_compression != NULL)
+	{
+		options->proto.logical.spill_compression =
+			pstrdup(MySubscription->spill_compression);
+	}
+	else
+		options->proto.logical.spill_compression = NULL;
+
 	options->proto.logical.twophase = false;
 	options->proto.logical.origin = pstrdup(MySubscription->origin);
 }
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 00e7024563e..521b646bb60 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -27,6 +27,7 @@
 #include "replication/logicalproto.h"
 #include "replication/origin.h"
 #include "replication/pgoutput.h"
+#include "replication/reorderbuffer_compression.h"
 #include "utils/builtins.h"
 #include "utils/inval.h"
 #include "utils/lsyscache.h"
@@ -283,11 +284,13 @@ parse_output_parameters(List *options, PGOutputData *data)
 	bool		streaming_given = false;
 	bool		two_phase_option_given = false;
 	bool		origin_option_given = false;
+	bool		spill_compression_option_given = false;
 
 	data->binary = false;
 	data->streaming = LOGICALREP_STREAM_OFF;
 	data->messages = false;
 	data->two_phase = false;
+	data->spill_compression_method = REORDER_BUFFER_NO_COMPRESSION;
 
 	foreach(lc, options)
 	{
@@ -396,6 +399,28 @@ parse_output_parameters(List *options, PGOutputData *data)
 						errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 						errmsg("unrecognized origin value: \"%s\"", origin));
 		}
+		else if (strcmp(defel->defname, "spill_compression") == 0)
+		{
+			uint8		method;
+			char	   *method_str;
+
+			if (spill_compression_option_given)
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("conflicting or redundant options")));
+			spill_compression_option_given = true;
+
+			method_str = defGetString(defel);
+			method = ReorderBufferParseCompressionMethod(method_str);
+
+			if (method == REORDER_BUFFER_INVALID_COMPRESSION)
+				ereport(ERROR,
+						errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+						errmsg("invalid spill files compression method: \"%s\"",
+							   method_str));
+
+			data->spill_compression_method = method;
+		}
 		else
 			elog(ERROR, "unrecognized pgoutput option: %s", defel->defname);
 	}
@@ -508,6 +533,9 @@ pgoutput_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
 		data->publications = NIL;
 		publications_valid = false;
 
+		/* Init spill files compression method */
+		ctx->spill_compression_method = data->spill_compression_method;
+
 		/*
 		 * Register callback for pg_publication if we didn't already do that
 		 * during some previous call in this process.
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 6e07984e8d5..e660b3bd444 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -4847,6 +4847,7 @@ getSubscriptions(Archive *fout)
 	int			i_suboriginremotelsn;
 	int			i_subenabled;
 	int			i_subfailover;
+	int			i_subspillcompression;
 	int			i,
 				ntups;
 
@@ -4919,10 +4920,17 @@ getSubscriptions(Archive *fout)
 
 	if (fout->remoteVersion >= 170000)
 		appendPQExpBufferStr(query,
-							 " s.subfailover\n");
+							 " s.subfailover,\n");
 	else
 		appendPQExpBuffer(query,
-						  " false AS subfailover\n");
+						  " false AS subfailover,\n");
+
+	if (fout->remoteVersion >= 180000)
+		appendPQExpBufferStr(query,
+							 " s.subspillcompression\n");
+	else
+		appendPQExpBuffer(query,
+						  " 'off' AS subspillcompression\n");
 
 	appendPQExpBufferStr(query,
 						 "FROM pg_subscription s\n");
@@ -4962,6 +4970,7 @@ getSubscriptions(Archive *fout)
 	i_suboriginremotelsn = PQfnumber(res, "suboriginremotelsn");
 	i_subenabled = PQfnumber(res, "subenabled");
 	i_subfailover = PQfnumber(res, "subfailover");
+	i_subspillcompression = PQfnumber(res, "subspillcompression");
 
 	subinfo = pg_malloc(ntups * sizeof(SubscriptionInfo));
 
@@ -5008,6 +5017,8 @@ getSubscriptions(Archive *fout)
 			pg_strdup(PQgetvalue(res, i, i_subenabled));
 		subinfo[i].subfailover =
 			pg_strdup(PQgetvalue(res, i, i_subfailover));
+		subinfo[i].subspillcompression =
+			pg_strdup(PQgetvalue(res, i, i_subspillcompression));
 
 		/* Decide whether we want to dump it */
 		selectDumpableObject(&(subinfo[i].dobj), fout);
@@ -5254,6 +5265,9 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
 	if (pg_strcasecmp(subinfo->suborigin, LOGICALREP_ORIGIN_ANY) != 0)
 		appendPQExpBuffer(query, ", origin = %s", subinfo->suborigin);
 
+	if (strcmp(subinfo->subspillcompression, "off") != 0)
+		appendPQExpBuffer(query, ", spill_compression = %s", subinfo->subspillcompression);
+
 	appendPQExpBufferStr(query, ");\n");
 
 	/*
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index 9f907ed5ad4..ecbf2c2e276 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -673,6 +673,7 @@ typedef struct _SubscriptionInfo
 	char	   *suborigin;
 	char	   *suboriginremotelsn;
 	char	   *subfailover;
+	char	   *subspillcompression;
 } SubscriptionInfo;
 
 /*
diff --git a/src/bin/pg_dump/t/002_pg_dump.pl b/src/bin/pg_dump/t/002_pg_dump.pl
index ab6c8304913..dea2fc40eac 100644
--- a/src/bin/pg_dump/t/002_pg_dump.pl
+++ b/src/bin/pg_dump/t/002_pg_dump.pl
@@ -3001,9 +3001,9 @@ my %tests = (
 		create_order => 50,
 		create_sql => 'CREATE SUBSCRIPTION sub2
 						 CONNECTION \'dbname=doesnotexist\' PUBLICATION pub1
-						 WITH (connect = false, origin = none);',
+						 WITH (connect = false, origin = none, spill_compression = on);',
 		regexp => qr/^
-			\QCREATE SUBSCRIPTION sub2 CONNECTION 'dbname=doesnotexist' PUBLICATION pub1 WITH (connect = false, slot_name = 'sub2', origin = none);\E
+			\QCREATE SUBSCRIPTION sub2 CONNECTION 'dbname=doesnotexist' PUBLICATION pub1 WITH (connect = false, slot_name = 'sub2', origin = none, spill_compression = on);\E
 			/xm,
 		like => { %full_runs, section_post_data => 1, },
 	},
@@ -3012,9 +3012,9 @@ my %tests = (
 		create_order => 50,
 		create_sql => 'CREATE SUBSCRIPTION sub3
 						 CONNECTION \'dbname=doesnotexist\' PUBLICATION pub1
-						 WITH (connect = false, origin = any);',
+						 WITH (connect = false, origin = any, spill_compression = pglz);',
 		regexp => qr/^
-			\QCREATE SUBSCRIPTION sub3 CONNECTION 'dbname=doesnotexist' PUBLICATION pub1 WITH (connect = false, slot_name = 'sub3');\E
+			\QCREATE SUBSCRIPTION sub3 CONNECTION 'dbname=doesnotexist' PUBLICATION pub1 WITH (connect = false, slot_name = 'sub3', spill_compression = pglz);\E
 			/xm,
 		like => { %full_runs, section_post_data => 1, },
 	},
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index faabecbc76f..9378a03bf23 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -6547,7 +6547,7 @@ describeSubscriptions(const char *pattern, bool verbose)
 	printQueryOpt myopt = pset.popt;
 	static const bool translate_columns[] = {false, false, false, false,
 		false, false, false, false, false, false, false, false, false, false,
-	false};
+	false, false};
 
 	if (pset.sversion < 100000)
 	{
@@ -6627,6 +6627,11 @@ describeSubscriptions(const char *pattern, bool verbose)
 			appendPQExpBuffer(&buf,
 							  ", subskiplsn AS \"%s\"\n",
 							  gettext_noop("Skip LSN"));
+
+		if (pset.sversion >= 180000)
+			appendPQExpBuffer(&buf,
+							  ", subspillcompression AS \"%s\"\n",
+							  gettext_noop("Spill files compression"));
 	}
 
 	/* Only display subscriptions in current database. */
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index a7ccde6d7df..728f2bd771e 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1948,7 +1948,7 @@ psql_completion(const char *text, int start, int end)
 	else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SET", "("))
 		COMPLETE_WITH("binary", "disable_on_error", "failover", "origin",
 					  "password_required", "run_as_owner", "slot_name",
-					  "streaming", "synchronous_commit", "two_phase");
+					  "spill_compression", "streaming", "synchronous_commit", "two_phase");
 	/* ALTER SUBSCRIPTION <name> SKIP ( */
 	else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SKIP", "("))
 		COMPLETE_WITH("lsn");
@@ -3345,7 +3345,8 @@ psql_completion(const char *text, int start, int end)
 		COMPLETE_WITH("binary", "connect", "copy_data", "create_slot",
 					  "disable_on_error", "enabled", "failover", "origin",
 					  "password_required", "run_as_owner", "slot_name",
-					  "streaming", "synchronous_commit", "two_phase");
+					  "spill_compression", "streaming", "synchronous_commit",
+					  "two_phase");
 
 /* CREATE TRIGGER --- is allowed inside CREATE SCHEMA, so use TailMatches */
 
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index 0aa14ec4a27..61c284349ca 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -113,6 +113,9 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
 
 	/* Only publish data originating from the specified origin */
 	text		suborigin BKI_DEFAULT(LOGICALREP_ORIGIN_ANY);
+
+	/* Spill files compression algorithm */
+	text		subspillcompression BKI_FORCE_NOT_NULL;
 #endif
 } FormData_pg_subscription;
 
@@ -157,6 +160,7 @@ typedef struct Subscription
 	List	   *publications;	/* List of publication names to subscribe to */
 	char	   *origin;			/* Only publish data originating from the
 								 * specified origin */
+	char	   *spill_compression;	/* Spill files compression algorithm */
 } Subscription;
 
 /* Disallow streaming in-progress transactions. */
diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h
index aff38e8d049..75c17866c38 100644
--- a/src/include/replication/logical.h
+++ b/src/include/replication/logical.h
@@ -112,6 +112,8 @@ typedef struct LogicalDecodingContext
 
 	/* Do we need to process any change in fast_forward mode? */
 	bool		processing_required;
+	/* Compression method used to compress spill files */
+	uint8		spill_compression_method;
 } LogicalDecodingContext;

diff --git a/src/include/replication/pgoutput.h b/src/include/replication/pgoutput.h
index 89f94e11472..eabcca62af9 100644
--- a/src/include/replication/pgoutput.h
+++ b/src/include/replication/pgoutput.h
@@ -33,6 +33,7 @@ typedef struct PGOutputData
 	bool		messages;
 	bool		two_phase;
 	bool		publish_no_origin;
+	uint8		spill_compression_method;
 } PGOutputData;
 
 #endif							/* PGOUTPUT_H */
diff --git a/src/include/replication/reorderbuffer_compression.h b/src/include/replication/reorderbuffer_compression.h
index 960064d571e..f72f0e0fbd7 100644
--- a/src/include/replication/reorderbuffer_compression.h
+++ b/src/include/replication/reorderbuffer_compression.h
@@ -26,6 +26,7 @@
 /* ReorderBuffer on disk compression algorithms */
 typedef enum ReorderBufferCompressionMethod
 {
+	REORDER_BUFFER_INVALID_COMPRESSION,
 	REORDER_BUFFER_NO_COMPRESSION,
 	REORDER_BUFFER_LZ4_COMPRESSION,
 	REORDER_BUFFER_PGLZ_COMPRESSION,
@@ -140,5 +141,8 @@ extern void ReorderBufferCompress(ReorderBuffer *rb,
 extern void ReorderBufferDecompress(ReorderBuffer *rb, char *data,
 									ReorderBufferDiskHeader *header,
 									void *compressor_state);
+extern ReorderBufferCompressionMethod ReorderBufferParseCompressionMethod(const char *method);
+extern void ReorderBufferValidateCompressionMethod(const char *method,
+												   int elevel);
 
 #endif							/* REORDERBUFFER_COMPRESSION_H */
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 132e789948b..b759b5807d8 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -186,6 +186,7 @@ typedef struct
 									 * prepare time */
 			char	   *origin; /* Only publish data originating from the
 								 * specified origin */
+			char	   *spill_compression;	/* Spill files compression algo */
 		}			logical;
 	}			proto;
 } WalRcvStreamOptions;
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 17d48b16857..80ea7776513 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -116,18 +116,18 @@ CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PU
 WARNING:  subscription was created, but is not connected
 HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
 \dRs+ regress_testsub4
-                                                                                                                 List of subscriptions
-       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | none   | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                              List of subscriptions
+       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN | Spill files compression 
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+-------------------------
+ regress_testsub4 | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | none   | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | off
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub4 SET (origin = any);
 \dRs+ regress_testsub4
-                                                                                                                 List of subscriptions
-       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                              List of subscriptions
+       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN | Spill files compression 
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+-------------------------
+ regress_testsub4 | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | off
 (1 row)
 
 DROP SUBSCRIPTION regress_testsub3;
@@ -145,10 +145,10 @@ ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar';
 ERROR:  invalid connection string syntax: missing "=" after "foobar" in connection info string
 
 \dRs+
-                                                                                                                List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                             List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN | Spill files compression 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+-------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | off
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false);
@@ -157,10 +157,10 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = 'newname');
 ALTER SUBSCRIPTION regress_testsub SET (password_required = false);
 ALTER SUBSCRIPTION regress_testsub SET (run_as_owner = true);
 \dRs+
-                                                                                                                     List of subscriptions
-      Name       |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |           Conninfo           | Skip LSN 
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub2,testpub3} | f      | off       | d                | f                | any    | f                 | t             | f        | off                | dbname=regress_doesnotexist2 | 0/0
+                                                                                                                                  List of subscriptions
+      Name       |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |           Conninfo           | Skip LSN | Spill files compression 
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------+-------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub2,testpub3} | f      | off       | d                | f                | any    | f                 | t             | f        | off                | dbname=regress_doesnotexist2 | 0/0      | off
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (password_required = true);
@@ -176,10 +176,10 @@ ERROR:  unrecognized subscription parameter: "create_slot"
 -- ok
 ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/12345');
 \dRs+
-                                                                                                                     List of subscriptions
-      Name       |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |           Conninfo           | Skip LSN 
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub2,testpub3} | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist2 | 0/12345
+                                                                                                                                  List of subscriptions
+      Name       |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |           Conninfo           | Skip LSN | Spill files compression 
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------+-------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub2,testpub3} | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist2 | 0/12345  | off
 (1 row)
 
 -- ok - with lsn = NONE
@@ -188,10 +188,10 @@ ALTER SUBSCRIPTION regress_testsub SKIP (lsn = NONE);
 ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/0');
 ERROR:  invalid WAL location (LSN): 0/0
 \dRs+
-                                                                                                                     List of subscriptions
-      Name       |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |           Conninfo           | Skip LSN 
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub2,testpub3} | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist2 | 0/0
+                                                                                                                                  List of subscriptions
+      Name       |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |           Conninfo           | Skip LSN | Spill files compression 
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------+-------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub2,testpub3} | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist2 | 0/0      | off
 (1 row)
 
 BEGIN;
@@ -222,11 +222,15 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = local);
 ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = foobar);
 ERROR:  invalid value for parameter "synchronous_commit": "foobar"
 HINT:  Available values: local, remote_write, remote_apply, on, off.
+ALTER SUBSCRIPTION regress_testsub_foo SET (spill_compression = pglz);
+ALTER SUBSCRIPTION regress_testsub_foo SET (spill_compression = off);
+ALTER SUBSCRIPTION regress_testsub_foo SET (spill_compression = foobar);
+ERROR:  compression method "foobar" not valid
 \dRs+
-                                                                                                                       List of subscriptions
-        Name         |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |           Conninfo           | Skip LSN 
----------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub_foo | regress_subscription_user | f       | {testpub2,testpub3} | f      | off       | d                | f                | any    | t                 | f             | f        | local              | dbname=regress_doesnotexist2 | 0/0
+                                                                                                                                    List of subscriptions
+        Name         |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |           Conninfo           | Skip LSN | Spill files compression 
+---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------+-------------------------
+ regress_testsub_foo | regress_subscription_user | f       | {testpub2,testpub3} | f      | off       | d                | f                | any    | t                 | f             | f        | local              | dbname=regress_doesnotexist2 | 0/0      | off
 (1 row)
 
 -- rename back to keep the rest simple
@@ -255,19 +259,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 WARNING:  subscription was created, but is not connected
 HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
 \dRs+
-                                                                                                                List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | t      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                             List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN | Spill files compression 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+-------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | t      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | off
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (binary = false);
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 \dRs+
-                                                                                                                List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                             List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN | Spill files compression 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+-------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | off
 (1 row)
 
 DROP SUBSCRIPTION regress_testsub;
@@ -279,27 +283,27 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 WARNING:  subscription was created, but is not connected
 HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
 \dRs+
-                                                                                                                List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                             List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN | Spill files compression 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+-------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | off
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel);
 \dRs+
-                                                                                                                List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                             List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN | Spill files compression 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+-------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | off
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (streaming = false);
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 \dRs+
-                                                                                                                List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                             List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN | Spill files compression 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+-------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | off
 (1 row)
 
 -- fail - publication already exists
@@ -314,10 +318,10 @@ ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refr
 ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false);
 ERROR:  publication "testpub1" is already in subscription "regress_testsub"
 \dRs+
-                                                                                                                        List of subscriptions
-      Name       |           Owner           | Enabled |         Publication         | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub,testpub1,testpub2} | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                                     List of subscriptions
+      Name       |           Owner           | Enabled |         Publication         | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN | Spill files compression 
+-----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+-------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub,testpub1,testpub2} | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | off
 (1 row)
 
 -- fail - publication used more than once
@@ -332,10 +336,10 @@ ERROR:  publication "testpub3" is not in subscription "regress_testsub"
 -- ok - delete publications
 ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false);
 \dRs+
-                                                                                                                List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                             List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN | Spill files compression 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+-------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | off
 (1 row)
 
 DROP SUBSCRIPTION regress_testsub;
@@ -371,19 +375,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 WARNING:  subscription was created, but is not connected
 HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
 \dRs+
-                                                                                                                List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | p                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                             List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN | Spill files compression 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+-------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | p                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | off
 (1 row)
 
 -- we can alter streaming when two_phase enabled
 ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
 \dRs+
-                                                                                                                List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | p                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                             List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN | Spill files compression 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+-------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | p                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | off
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -393,10 +397,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 WARNING:  subscription was created, but is not connected
 HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
 \dRs+
-                                                                                                                List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | p                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                             List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN | Spill files compression 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+-------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | p                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | off
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -409,18 +413,18 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 WARNING:  subscription was created, but is not connected
 HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
 \dRs+
-                                                                                                                List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                             List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN | Spill files compression 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+-------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | off
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
 \dRs+
-                                                                                                                List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | t                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                             List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN | Spill files compression 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+-------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | t                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | off
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index 007c9e70374..368696f430c 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -140,6 +140,10 @@ ALTER SUBSCRIPTION regress_testsub RENAME TO regress_testsub_foo;
 ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = local);
 ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = foobar);
 
+ALTER SUBSCRIPTION regress_testsub_foo SET (spill_compression = pglz);
+ALTER SUBSCRIPTION regress_testsub_foo SET (spill_compression = off);
+ALTER SUBSCRIPTION regress_testsub_foo SET (spill_compression = foobar);
+
 \dRs+
 
 -- rename back to keep the rest simple
-- 
2.46.0

From 238d391dfee9a1dc0020c17fe158f2a5d1bbf9da Mon Sep 17 00:00:00 2001
From: Julien Tachoires <[email-masked]>
Date: Thu, 18 Jul 2024 07:51:29 -0700
Subject: [PATCH v4 10/11] Add ReorderBuffer ondisk compression tests

---
 src/test/subscription/Makefile                |  2 +
 src/test/subscription/meson.build             |  7 +-
 .../t/034_reorderbuffer_compression.pl        | 99 +++++++++++++++++++
 3 files changed, 107 insertions(+), 1 deletion(-)
 create mode 100644 src/test/subscription/t/034_reorderbuffer_compression.pl

diff --git a/src/test/subscription/Makefile b/src/test/subscription/Makefile
index ce1ca430095..9341f1493c5 100644
--- a/src/test/subscription/Makefile
+++ b/src/test/subscription/Makefile
@@ -16,6 +16,8 @@ include $(top_builddir)/src/Makefile.global
 EXTRA_INSTALL = contrib/hstore
 
 export with_icu
+export with_lz4
+export with_zstd
 
 check:
 	$(prove_check)
diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build
index c591cd7d619..772eeb817f6 100644
--- a/src/test/subscription/meson.build
+++ b/src/test/subscription/meson.build
@@ -5,7 +5,11 @@ tests += {
   'sd': meson.current_source_dir(),
   'bd': meson.current_build_dir(),
   'tap': {
-    'env': {'with_icu': icu.found() ? 'yes' : 'no'},
+    'env': {
+      'with_icu': icu.found() ? 'yes' : 'no',
+      'with_lz4': lz4.found() ? 'yes' : 'no',
+      'with_zstd': zstd.found() ? 'yes' : 'no',
+    },
     'tests': [
       't/001_rep_changes.pl',
       't/002_types.pl',
@@ -40,6 +44,7 @@ tests += {
       't/031_column_list.pl',
       't/032_subscribe_use_index.pl',
       't/033_run_as_table_owner.pl',
+      't/034_reorderbuffer_compression.pl',
       't/100_bugs.pl',
     ],
   },
diff --git a/src/test/subscription/t/034_reorderbuffer_compression.pl b/src/test/subscription/t/034_reorderbuffer_compression.pl
new file mode 100644
index 00000000000..65c9be14a22
--- /dev/null
+++ b/src/test/subscription/t/034_reorderbuffer_compression.pl
@@ -0,0 +1,99 @@
+
+# Copyright (c) 2024, PostgreSQL Global Development Group
+
+# Test ReorderBuffer compression
+use strict;
+use warnings FATAL => 'all';
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+sub test_reorderbuffer_compression
+{
+	my ($node_publisher, $node_subscriber, $appname, $compression) = @_;
+
+	# Set subscriber's spill_compression option
+	$node_subscriber->safe_psql('postgres',
+		"ALTER SUBSCRIPTION tap_sub SET (spill_compression = $compression)");
+
+	# Make sure the table is empty
+	$node_publisher->safe_psql('postgres', 'TRUNCATE test_tab');
+
+	# Reset replication slot stats
+	$node_publisher->safe_psql('postgres',
+		"SELECT pg_stat_reset_replication_slot('tap_sub')");
+
+	# Insert 1 million rows in the table
+	$node_publisher->safe_psql('postgres',
+		"INSERT INTO test_tab SELECT i, 'Message number #'||i::TEXT FROM generate_series(1, 1000000) as i"
+	);
+
+	$node_publisher->wait_for_catchup($appname);
+
+	# Check if table content is replicated
+	my $result =
+	  $node_subscriber->safe_psql('postgres',
+		"SELECT count(*) FROM test_tab");
+	is($result, qq(1000000), 'check data was copied to subscriber');
+
+	# Check if the transaction was spilled on disk
+	my $res_stats =
+	  $node_publisher->safe_psql('postgres',
+		"SELECT spill_txns FROM pg_catalog.pg_stat_get_replication_slot('tap_sub');");
+	is($res_stats, qq(1), 'check if the transaction was spilled on disk');
+}
+
+# Create publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->append_conf('postgresql.conf',
+	'logical_decoding_work_mem = 64');
+$node_publisher->start;
+
+# Create subscriber node
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init;
+$node_subscriber->start;
+
+# Setup structure on publisher
+$node_publisher->safe_psql('postgres',
+	"CREATE TABLE test_tab (a int primary key, b text)");
+
+# Setup structure on subscriber
+$node_subscriber->safe_psql('postgres',
+	"CREATE TABLE test_tab (a int primary key, b text)");
+
+# Setup logical replication
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+$node_publisher->safe_psql('postgres',
+	"CREATE PUBLICATION tap_pub FOR TABLE test_tab");
+
+my $appname = 'tap_sub';
+
+$node_subscriber->safe_psql('postgres',
+	"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming = off)"
+);
+
+test_reorderbuffer_compression($node_publisher, $node_subscriber, $appname,
+	'off');
+test_reorderbuffer_compression($node_publisher, $node_subscriber, $appname,
+	'pglz');
+
+SKIP:
+{
+	skip "LZ4 not supported by this build", 2 if ($ENV{with_lz4} ne 'yes');
+	test_reorderbuffer_compression($node_publisher, $node_subscriber, $appname,
+		'lz4');
+}
+
+SKIP:
+{
+	skip "ZSTD not supported by this build", 2 if ($ENV{with_zstd} ne 'yes');
+	test_reorderbuffer_compression($node_publisher, $node_subscriber, $appname,
+		'zstd');
+}
+
+$node_subscriber->stop;
+$node_publisher->stop;
+
+done_testing();
-- 
2.46.0

From db7367f9fe2bf1072793fedbb749422e4977b5e0 Mon Sep 17 00:00:00 2001
From: Tomas Vondra <[email-masked]>
Date: Tue, 17 Sep 2024 17:23:26 +0200
Subject: [PATCH v4 11/11] pgindent

---
 .../replication/logical/reorderbuffer.c       |  11 +-
 .../logical/reorderbuffer_compression.c       | 285 +++++++++---------
 src/backend/replication/logical/worker.c      |   4 +-
 .../replication/reorderbuffer_compression.h   |  12 +-
 src/tools/pgindent/typedefs.list              |   5 +
 5 files changed, 166 insertions(+), 151 deletions(-)

diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 6545c310f7f..1b6ddecb83e 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -257,7 +257,7 @@ static Size ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *tx
 static Size ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn,
 										TXNEntryFile *file, XLogSegNo *segno);
 static bool ReorderBufferReadOndiskChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
-								   TXNEntryFile *file, XLogSegNo *segno);
+										  TXNEntryFile *file, XLogSegNo *segno);
 static void ReorderBufferRestoreChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 									   char *data);
 static void ReorderBufferRestoreCleanup(ReorderBuffer *rb, ReorderBufferTXN *txn);
@@ -3833,7 +3833,7 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 	ReorderBufferSerializeReserve(rb, sz);
 
 	disk_hdr = (ReorderBufferDiskHeader *) rb->outbuf;
-	memcpy((char *)rb->outbuf + sizeof(ReorderBufferDiskHeader), change, sizeof(ReorderBufferChange));
+	memcpy((char *) rb->outbuf + sizeof(ReorderBufferDiskHeader), change, sizeof(ReorderBufferChange));
 
 	switch (change->action)
 	{
@@ -4345,9 +4345,8 @@ ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn,
 		}
 
 		/*
-		 * Read the full change from disk.
-		 * If ReorderBufferReadOndiskChange returns false, then we are at the
-		 * eof, so, move the next segment.
+		 * Read the full change from disk. If ReorderBufferReadOndiskChange
+		 * returns false, then we are at the eof, so, move the next segment.
 		 */
 		if (!ReorderBufferReadOndiskChange(rb, txn, file, segno))
 		{
@@ -4375,7 +4374,7 @@ ReorderBufferReadOndiskChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 {
 	int			readBytes;
 	ReorderBufferDiskHeader *disk_hdr;
-	char	   *header;			/* disk header buffer*/
+	char	   *header;			/* disk header buffer */
 	char	   *data;			/* data buffer */
 
 	/*
diff --git a/src/backend/replication/logical/reorderbuffer_compression.c b/src/backend/replication/logical/reorderbuffer_compression.c
index 707f35cff7c..6d19c60b601 100644
--- a/src/backend/replication/logical/reorderbuffer_compression.c
+++ b/src/backend/replication/logical/reorderbuffer_compression.c
@@ -158,9 +158,9 @@ lz4_StreamingCompressData(MemoryContext context, char *src, Size src_size,
 #else
 	LZ4StreamingCompressorState *cstate;
 	int			lz4_cmp_size = 0;	/* compressed size */
-	char	   *buf;				/* buffer used for compression */
-	Size		buf_size;			/* buffer size */
-	char	   *lz4_in_bufPtr;		/* input ring buffer pointer */
+	char	   *buf;			/* buffer used for compression */
+	Size		buf_size;		/* buffer size */
+	char	   *lz4_in_bufPtr;	/* input ring buffer pointer */
 
 	cstate = (LZ4StreamingCompressorState *) compressor_state;
 
@@ -204,14 +204,14 @@ lz4_StreamingCompressData(MemoryContext context, char *src, Size src_size,
  * Data compression using LZ4 API.
  */
 static void
-lz4_CompressData(char *src, Size src_size, char **dst,  Size *dst_size)
+lz4_CompressData(char *src, Size src_size, char **dst, Size *dst_size)
 {
 #ifndef USE_LZ4
 	NO_LZ4_SUPPORT();
 #else
 	int			lz4_cmp_size = 0;	/* compressed size */
-	char	   *buf;				/* buffer used for compression */
-	Size		buf_size;			/* buffer size */
+	char	   *buf;			/* buffer used for compression */
+	Size		buf_size;		/* buffer size */
 
 	buf_size = LZ4_COMPRESSBOUND(src_size);
 	buf = (char *) palloc0(buf_size);
@@ -243,8 +243,8 @@ lz4_StreamingDecompressData(MemoryContext context, char *src, Size src_size,
 	NO_LZ4_SUPPORT();
 #else
 	LZ4StreamingCompressorState *cstate;
-	char	   *lz4_out_bufPtr;		/* output ring buffer pointer */
-	int			lz4_dec_size;		/* decompressed data size */
+	char	   *lz4_out_bufPtr; /* output ring buffer pointer */
+	int			lz4_dec_size;	/* decompressed data size */
 
 	cstate = (LZ4StreamingCompressorState *) compressor_state;
 
@@ -273,8 +273,8 @@ lz4_StreamingDecompressData(MemoryContext context, char *src, Size src_size,
 				 errmsg_internal("compressed LZ4 data is corrupted")));
 	else if (lz4_dec_size != dst_size)
 		ereport(ERROR,
-			(errcode(ERRCODE_DATA_CORRUPTED),
-			 errmsg_internal("decompressed LZ4 data size differs from original size")));
+				(errcode(ERRCODE_DATA_CORRUPTED),
+				 errmsg_internal("decompressed LZ4 data size differs from original size")));
 
 	/* Move the output ring buffer offset */
 	cstate->lz4_out_buf_offset += lz4_dec_size;
@@ -334,8 +334,8 @@ zstd_NewCompressorState(MemoryContext context)
 
 	/*
 	 * We do not allocate ZSTD buffers and contexts at this point because we
-	 * have no guarantee that we will need them later. Let's allocate only when
-	 * we are about to use them.
+	 * have no guarantee that we will need them later. Let's allocate only
+	 * when we are about to use them.
 	 */
 	cstate->zstd_c_ctx = NULL;
 	cstate->zstd_c_in_buf = NULL;
@@ -461,6 +461,7 @@ zstd_StreamingCompressData(MemoryContext context, char *src, Size src_size,
 	NO_ZSTD_SUPPORT();
 #else
 	ZSTDStreamingCompressorState *cstate;
+
 	/* Size of remaining data to be copied from src into ZSTD input buffer */
 	Size		toCpy = src_size;
 	char	   *dst_data;
@@ -479,16 +480,16 @@ zstd_StreamingCompressData(MemoryContext context, char *src, Size src_size,
 	/*
 	 * ZSTD streaming compression works with chunks: the source data needs to
 	 * be splitted out in chunks, each of them is then copied into ZSTD input
-	 * buffer.
-	 * For each chunk, we proceed with compression. Streaming compression is
-	 * not intended to compress the whole input chunk, so we have the call
-	 * ZSTD_compressStream2() multiple times until the entire chunk is
-	 * consumed.
+	 * buffer. For each chunk, we proceed with compression. Streaming
+	 * compression is not intended to compress the whole input chunk, so we
+	 * have the call ZSTD_compressStream2() multiple times until the entire
+	 * chunk is consumed.
 	 */
 	while (toCpy > 0)
 	{
 		/* Are we on the last chunk? */
 		bool		last_chunk = (toCpy < cstate->zstd_c_in_buf_size);
+
 		/* Size of the data copied into ZSTD input buffer */
 		Size		cpySize = last_chunk ? toCpy : cstate->zstd_c_in_buf_size;
 		bool		finished = false;
@@ -556,12 +557,13 @@ zstd_StreamingCompressData(MemoryContext context, char *src, Size src_size,
  */
 static void
 zstd_StreamingDecompressData(MemoryContext context, char *src, Size src_size,
-							char **dst, Size dst_size, void *compressor_state)
+							 char **dst, Size dst_size, void *compressor_state)
 {
 #ifndef USE_ZSTD
 	NO_ZSTD_SUPPORT();
 #else
 	ZSTDStreamingCompressorState *cstate;
+
 	/* Size of remaining data to be copied from src into ZSTD input buffer */
 	Size		toCpy = src_size;
 	char	   *dst_data;
@@ -598,7 +600,7 @@ zstd_StreamingDecompressData(MemoryContext context, char *src, Size src_size,
 			output.size = cstate->zstd_d_out_buf_size;
 			output.pos = 0;
 
-			ret = ZSTD_decompressStream(cstate->zstd_d_ctx, &output , &input);
+			ret = ZSTD_decompressStream(cstate->zstd_d_ctx, &output, &input);
 
 			if (ZSTD_isError(ret))
 				ereport(ERROR,
@@ -691,145 +693,148 @@ ReorderBufferCompress(ReorderBuffer *rb, ReorderBufferDiskHeader **header,
 
 	switch (compression_method)
 	{
-		/* No compression */
+			/* No compression */
 		case REORDER_BUFFER_NO_COMPRESSION:
-		{
-			hdr->comp_strat = REORDER_BUFFER_STRAT_UNCOMPRESSED;
-			hdr->size = data_size;
-			hdr->raw_size = data_size - sizeof(ReorderBufferDiskHeader);
-
-			break;
-		}
-		/* LZ4 Compression */
-		case REORDER_BUFFER_LZ4_COMPRESSION:
-		{
-			/*
-			 * XXX Won't this cause a lot of palloc/pfree traffic? We allocate a new
-			 * buffer for every compression, only to immediately throw it away. Look
-			 * at what astreamer_lz4_compressor_content does to reuse a buffer, maybe
-			 * we should do that here too?
-			 */
-			char	   *dst = NULL;
-			Size		dst_size = 0;
-			char	   *src = (char *) rb->outbuf + sizeof(ReorderBufferDiskHeader);
-			Size		src_size = data_size - sizeof(ReorderBufferDiskHeader);
-			ReorderBufferCompressionStrategy strat;
-
-			if (lz4_CanDoStreamingCompression(src_size))
 			{
-				/* Use LZ4 streaming compression if possible */
-				lz4_StreamingCompressData(rb->context, src, src_size, &dst,
-										  &dst_size, compressor_state);
-				strat = REORDER_BUFFER_STRAT_LZ4_STREAMING;
+				hdr->comp_strat = REORDER_BUFFER_STRAT_UNCOMPRESSED;
+				hdr->size = data_size;
+				hdr->raw_size = data_size - sizeof(ReorderBufferDiskHeader);
+
+				break;
 			}
-			else
+			/* LZ4 Compression */
+		case REORDER_BUFFER_LZ4_COMPRESSION:
 			{
-				/* Fallback to LZ4 regular compression */
-				lz4_CompressData(src, src_size, &dst, &dst_size);
-				strat = REORDER_BUFFER_STRAT_LZ4_REGULAR;
-			}
+				/*
+				 * XXX Won't this cause a lot of palloc/pfree traffic? We
+				 * allocate a new buffer for every compression, only to
+				 * immediately throw it away. Look at what
+				 * astreamer_lz4_compressor_content does to reuse a buffer,
+				 * maybe we should do that here too?
+				 */
+				char	   *dst = NULL;
+				Size		dst_size = 0;
+				char	   *src = (char *) rb->outbuf + sizeof(ReorderBufferDiskHeader);
+				Size		src_size = data_size - sizeof(ReorderBufferDiskHeader);
+				ReorderBufferCompressionStrategy strat;
+
+				if (lz4_CanDoStreamingCompression(src_size))
+				{
+					/* Use LZ4 streaming compression if possible */
+					lz4_StreamingCompressData(rb->context, src, src_size, &dst,
+											  &dst_size, compressor_state);
+					strat = REORDER_BUFFER_STRAT_LZ4_STREAMING;
+				}
+				else
+				{
+					/* Fallback to LZ4 regular compression */
+					lz4_CompressData(src, src_size, &dst, &dst_size);
+					strat = REORDER_BUFFER_STRAT_LZ4_REGULAR;
+				}
 
-			/*
-			 * Make sure the ReorderBuffer has enough space to store compressed
-			 * data. Compressed data must be smaller than raw data, so, the
-			 * ReorderBuffer should already have room for compressed data, but
-			 * we do this to avoid buffer overflow risks.
-			 */
-			ReorderBufferReserve(rb, (dst_size + sizeof(ReorderBufferDiskHeader)));
+				/*
+				 * Make sure the ReorderBuffer has enough space to store
+				 * compressed data. Compressed data must be smaller than raw
+				 * data, so, the ReorderBuffer should already have room for
+				 * compressed data, but we do this to avoid buffer overflow
+				 * risks.
+				 */
+				ReorderBufferReserve(rb, (dst_size + sizeof(ReorderBufferDiskHeader)));
 
-			hdr = (ReorderBufferDiskHeader *) rb->outbuf;
-			hdr->comp_strat = strat;
-			hdr->size = dst_size + sizeof(ReorderBufferDiskHeader);
-			hdr->raw_size = src_size;
+				hdr = (ReorderBufferDiskHeader *) rb->outbuf;
+				hdr->comp_strat = strat;
+				hdr->size = dst_size + sizeof(ReorderBufferDiskHeader);
+				hdr->raw_size = src_size;
 
-			/*
-			 * Update header: hdr pointer has potentially changed due to
-			 * ReorderBufferReserve()
-			 */
-			*header = hdr;
+				/*
+				 * Update header: hdr pointer has potentially changed due to
+				 * ReorderBufferReserve()
+				 */
+				*header = hdr;
 
-			/* Copy back compressed data into the ReorderBuffer */
-			memcpy((char *) rb->outbuf + sizeof(ReorderBufferDiskHeader), dst,
-				   dst_size);
+				/* Copy back compressed data into the ReorderBuffer */
+				memcpy((char *) rb->outbuf + sizeof(ReorderBufferDiskHeader), dst,
+					   dst_size);
 
-			pfree(dst);
+				pfree(dst);
+
+				break;
+			}
+			/* PGLZ compression */
 
-			break;
-		}
-		/* PGLZ compression */
-		/*
-		 * XXX I'd probably start by adding pglz first, and only then add lz4 as a
-		 * separate patch, simply because pglz is the default and always supported,
-		 * while lz4 requires extra flags.
-		 */
-		case REORDER_BUFFER_PGLZ_COMPRESSION:
-		{
 			/*
-			 * XXX Same comment about palloc/pfree traffic as for lz4 ...
+			 * XXX I'd probably start by adding pglz first, and only then add
+			 * lz4 as a separate patch, simply because pglz is the default and
+			 * always supported, while lz4 requires extra flags.
 			 */
-			int32		dst_size = 0;
-			char	   *dst = NULL;
-			char	   *src = (char *) rb->outbuf + sizeof(ReorderBufferDiskHeader);
-			int32		src_size = data_size - sizeof(ReorderBufferDiskHeader);
-			int32		max_size = PGLZ_MAX_OUTPUT(src_size);
+		case REORDER_BUFFER_PGLZ_COMPRESSION:
+			{
+				/*
+				 * XXX Same comment about palloc/pfree traffic as for lz4 ...
+				 */
+				int32		dst_size = 0;
+				char	   *dst = NULL;
+				char	   *src = (char *) rb->outbuf + sizeof(ReorderBufferDiskHeader);
+				int32		src_size = data_size - sizeof(ReorderBufferDiskHeader);
+				int32		max_size = PGLZ_MAX_OUTPUT(src_size);
 
-			dst = (char *) palloc0(max_size);
-			dst_size = pglz_compress(src, src_size, dst, PGLZ_strategy_always);
+				dst = (char *) palloc0(max_size);
+				dst_size = pglz_compress(src, src_size, dst, PGLZ_strategy_always);
 
-			if (dst_size < 0)
-				ereport(ERROR,
-						(errcode(ERRCODE_DATA_CORRUPTED),
-						 errmsg_internal("PGLZ compression failed")));
+				if (dst_size < 0)
+					ereport(ERROR,
+							(errcode(ERRCODE_DATA_CORRUPTED),
+							 errmsg_internal("PGLZ compression failed")));
 
-			ReorderBufferReserve(rb, (Size) (dst_size + sizeof(ReorderBufferDiskHeader)));
+				ReorderBufferReserve(rb, (Size) (dst_size + sizeof(ReorderBufferDiskHeader)));
 
-			hdr = (ReorderBufferDiskHeader *) rb->outbuf;
-			hdr->comp_strat = REORDER_BUFFER_STRAT_PGLZ;
-			hdr->size = (Size) dst_size + sizeof(ReorderBufferDiskHeader);
-			hdr->raw_size = (Size) src_size;
+				hdr = (ReorderBufferDiskHeader *) rb->outbuf;
+				hdr->comp_strat = REORDER_BUFFER_STRAT_PGLZ;
+				hdr->size = (Size) dst_size + sizeof(ReorderBufferDiskHeader);
+				hdr->raw_size = (Size) src_size;
 
-			*header = hdr;
+				*header = hdr;
 
-			/* Copy back compressed data into the ReorderBuffer */
-			memcpy((char *) rb->outbuf + sizeof(ReorderBufferDiskHeader), dst,
-				   dst_size);
+				/* Copy back compressed data into the ReorderBuffer */
+				memcpy((char *) rb->outbuf + sizeof(ReorderBufferDiskHeader), dst,
+					   dst_size);
 
-			pfree(dst);
+				pfree(dst);
 
-			break;
-		}
-		/* ZSTD Compression */
+				break;
+			}
+			/* ZSTD Compression */
 		case REORDER_BUFFER_ZSTD_COMPRESSION:
-		{
-			/*
-			 * XXX Same comment about palloc/pfree traffic as for lz4 ...
-			 */
-			char	   *dst = NULL;
-			Size		dst_size = 0;
-			char	   *src = (char *) rb->outbuf + sizeof(ReorderBufferDiskHeader);
-			Size		src_size = data_size - sizeof(ReorderBufferDiskHeader);
+			{
+				/*
+				 * XXX Same comment about palloc/pfree traffic as for lz4 ...
+				 */
+				char	   *dst = NULL;
+				Size		dst_size = 0;
+				char	   *src = (char *) rb->outbuf + sizeof(ReorderBufferDiskHeader);
+				Size		src_size = data_size - sizeof(ReorderBufferDiskHeader);
 
-			/* Use ZSTD streaming compression */
-			zstd_StreamingCompressData(rb->context, src, src_size, &dst,
-									   &dst_size, compressor_state);
+				/* Use ZSTD streaming compression */
+				zstd_StreamingCompressData(rb->context, src, src_size, &dst,
+										   &dst_size, compressor_state);
 
-			ReorderBufferReserve(rb, (dst_size + sizeof(ReorderBufferDiskHeader)));
+				ReorderBufferReserve(rb, (dst_size + sizeof(ReorderBufferDiskHeader)));
 
-			hdr = (ReorderBufferDiskHeader *) rb->outbuf;
-			hdr->comp_strat = REORDER_BUFFER_STRAT_ZSTD_STREAMING;
-			hdr->size = dst_size + sizeof(ReorderBufferDiskHeader);
-			hdr->raw_size = src_size;
+				hdr = (ReorderBufferDiskHeader *) rb->outbuf;
+				hdr->comp_strat = REORDER_BUFFER_STRAT_ZSTD_STREAMING;
+				hdr->size = dst_size + sizeof(ReorderBufferDiskHeader);
+				hdr->raw_size = src_size;
 
-			*header = hdr;
+				*header = hdr;
 
-			/* Copy back compressed data into the ReorderBuffer */
-			memcpy((char *) rb->outbuf + sizeof(ReorderBufferDiskHeader), dst,
-				   dst_size);
+				/* Copy back compressed data into the ReorderBuffer */
+				memcpy((char *) rb->outbuf + sizeof(ReorderBufferDiskHeader), dst,
+					   dst_size);
 
-			pfree(dst);
+				pfree(dst);
 
-			break;
-		}
+				break;
+			}
 	}
 }
 
@@ -847,6 +852,7 @@ ReorderBufferDecompress(ReorderBuffer *rb, char *data,
 						ReorderBufferDiskHeader *header, void *compressor_state)
 {
 	Size		raw_outbufsize = header->raw_size + sizeof(ReorderBufferDiskHeader);
+
 	/*
 	 * Make sure the output reorder buffer has enough space to store
 	 * decompressed/raw data.
@@ -862,7 +868,7 @@ ReorderBufferDecompress(ReorderBuffer *rb, char *data,
 
 	switch (header->comp_strat)
 	{
-		/* No decompression */
+			/* No decompression */
 		case REORDER_BUFFER_STRAT_UNCOMPRESSED:
 			{
 				/*
@@ -873,7 +879,7 @@ ReorderBufferDecompress(ReorderBuffer *rb, char *data,
 					   data, header->raw_size);
 				break;
 			}
-		/* LZ4 regular decompression */
+			/* LZ4 regular decompression */
 		case REORDER_BUFFER_STRAT_LZ4_REGULAR:
 			{
 				char	   *buf;
@@ -889,26 +895,28 @@ ReorderBufferDecompress(ReorderBuffer *rb, char *data,
 				pfree(buf);
 				break;
 			}
-		/* LZ4 streaming decompression */
+			/* LZ4 streaming decompression */
 		case REORDER_BUFFER_STRAT_LZ4_STREAMING:
 			{
-				char	   *buf;	/* XXX shouldn't this be set to NULL explicitly? */
+				char	   *buf;	/* XXX shouldn't this be set to NULL
+									 * explicitly? */
 				Size		src_size = header->size - sizeof(ReorderBufferDiskHeader);
 				Size		buf_size = header->raw_size;
 
 				lz4_StreamingDecompressData(rb->context, data, src_size, &buf,
-										   buf_size, compressor_state);
+											buf_size, compressor_state);
 
 				/* Copy decompressed data into the ReorderBuffer */
 				memcpy((char *) rb->outbuf + sizeof(ReorderBufferDiskHeader),
 					   buf, buf_size);
+
 				/*
 				 * Not necessary to free buf in this case: it points to the
 				 * decompressed data stored in LZ4 output ring buffer.
 				 */
 				break;
 			}
-		/* PGLZ decompression */
+			/* PGLZ decompression */
 		case REORDER_BUFFER_STRAT_PGLZ:
 			{
 				char	   *buf;
@@ -928,10 +936,11 @@ ReorderBufferDecompress(ReorderBuffer *rb, char *data,
 							 errmsg_internal("compressed PGLZ data is corrupted")));
 				break;
 			}
-		/* ZSTD streaming decompression */
+			/* ZSTD streaming decompression */
 		case REORDER_BUFFER_STRAT_ZSTD_STREAMING:
 			{
-				char	   *buf;	/* XXX shouldn't this be set to NULL explicitly? */
+				char	   *buf;	/* XXX shouldn't this be set to NULL
+									 * explicitly? */
 				Size		src_size = header->size - sizeof(ReorderBufferDiskHeader);
 				Size		buf_size = header->raw_size;
 
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index eb42e844f99..32b38b94dd7 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -4471,8 +4471,8 @@ set_stream_options(WalRcvStreamOptions *options,
 	}
 
 	if (server_version >= 180000 &&
-			 MySubscription->stream == LOGICALREP_STREAM_OFF &&
-			 MySubscription->spill_compression != NULL)
+		MySubscription->stream == LOGICALREP_STREAM_OFF &&
+		MySubscription->spill_compression != NULL)
 	{
 		options->proto.logical.spill_compression =
 			pstrdup(MySubscription->spill_compression);
diff --git a/src/include/replication/reorderbuffer_compression.h b/src/include/replication/reorderbuffer_compression.h
index f72f0e0fbd7..3f69d2c2610 100644
--- a/src/include/replication/reorderbuffer_compression.h
+++ b/src/include/replication/reorderbuffer_compression.h
@@ -56,9 +56,9 @@ typedef enum ReorderBufferCompressionStrategy
  */
 typedef struct ReorderBufferDiskHeader
 {
-	ReorderBufferCompressionStrategy comp_strat; /* Compression strategy */
-	Size		size;					/* Ondisk size */
-	Size		raw_size;				/* Raw/uncompressed data size */
+	ReorderBufferCompressionStrategy comp_strat;	/* Compression strategy */
+	Size		size;			/* Ondisk size */
+	Size		raw_size;		/* Raw/uncompressed data size */
 	/* ReorderBufferChange + data follows */
 } ReorderBufferDiskHeader;
 
@@ -81,7 +81,8 @@ typedef struct ReorderBufferDiskHeader
  * LZ4 streaming compression/decompression handlers and ring
  * buffers.
  */
-typedef struct LZ4StreamingCompressorState {
+typedef struct LZ4StreamingCompressorState
+{
 	/* Streaming compression handler */
 	LZ4_stream_t *lz4_stream;
 	/* Streaming decompression handler */
@@ -112,7 +113,8 @@ typedef struct LZ4StreamingCompressorState {
 /*
  * ZSTD streaming compression/decompression handlers and buffers.
  */
-typedef struct ZSTDStreamingCompressorState {
+typedef struct ZSTDStreamingCompressorState
+{
 	/* Compression */
 	ZSTD_CCtx  *zstd_c_ctx;
 	Size		zstd_c_in_buf_size;
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index ace5414fa5b..ea13d621f6c 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1487,6 +1487,7 @@ LZ4F_decompressionContext_t
 LZ4F_errorCode_t
 LZ4F_preferences_t
 LZ4State
+LZ4StreamingCompressorState
 LabelProvider
 LagTracker
 LargeObjectDesc
@@ -2412,7 +2413,10 @@ ReorderBufferChange
 ReorderBufferChangeType
 ReorderBufferCommitCB
 ReorderBufferCommitPreparedCB
+ReorderBufferCompressionMethod
+ReorderBufferCompressionStrategy
 ReorderBufferDiskChange
+ReorderBufferDiskHeader
 ReorderBufferIterTXNEntry
 ReorderBufferIterTXNState
 ReorderBufferMessageCB
@@ -3247,6 +3251,7 @@ XmlTableBuilderData
 YYLTYPE
 YYSTYPE
 YY_BUFFER_STATE
+ZSTDStreamingCompressorState
 ZSTD_CCtx
 ZSTD_CStream
 ZSTD_DCtx
-- 
2.46.0