From 1d427b7b61f69e0402d093b6dd35096638cfd033 Mon Sep 17 00:00:00 2001
From: Julien Tachoires <julmon@gmail.com>
Date: Sun, 22 Sep 2024 16:59:10 +0200
Subject: [PATCH 1/6] Compress ReorderBuffer spill files using PGLZ

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, decoding
of large transactions by multiple replication slots can lead to disk
space saturation and high I/O utilization.

This patch enables data compression/decompression of these temporary
files. Each transaction change that must be written on disk is now
compressed using the in-house compression method PGLZ.
---
 src/backend/replication/logical/Makefile      |   1 +
 src/backend/replication/logical/meson.build   |   1 +
 .../replication/logical/reorderbuffer.c       | 356 ++++++++++++++----
 .../replication/logical/reorderbuffer_pglz.c  |  56 +++
 src/include/replication/reorderbuffer.h       |   6 +
 .../replication/reorderbuffer_compression.h   |  42 +++
 6 files changed, 392 insertions(+), 70 deletions(-)
 create mode 100644 src/backend/replication/logical/reorderbuffer_pglz.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 1e08bbbd4e..96c733d009 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_pglz.o \
 	slotsync.o \
 	snapbuild.o \
 	tablesync.o \
diff --git a/src/backend/replication/logical/meson.build b/src/backend/replication/logical/meson.build
index 3d36249d8a..48df907b35 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_pglz.c',
   'slotsync.c',
   'snapbuild.c',
   'tablesync.c',
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index e3a5c7b660..649b196d5f 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -97,11 +97,13 @@
 #include "access/xlog_internal.h"
 #include "catalog/catalog.h"
 #include "common/int.h"
+#include "common/pg_lzcompress.h"
 #include "lib/binaryheap.h"
 #include "miscadmin.h"
 #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"
@@ -176,9 +178,10 @@ typedef struct ReorderBufferToastEnt
 /* Disk serialization support datastructures */
 typedef struct ReorderBufferDiskChange
 {
-	Size		size;
-	ReorderBufferChange change;
-	/* data follows */
+	ReorderBufferCompressionStrategy comp_strat;	/* Compression strategy */
+	Size		size;			/* Ondisk size */
+	Size		raw_size;		/* Raw/uncompressed data size */
+	/* ReorderBufferChange + data follow */
 } ReorderBufferDiskChange;
 
 #define IsSpecInsert(action) \
@@ -215,6 +218,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_PGLZ_COMPRESSION;
+
 /* ---------------------------------------
  * primary reorderbuffer support routines
  * ---------------------------------------
@@ -255,6 +261,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);
@@ -301,6 +309,24 @@ static void ReorderBufferChangeMemoryUpdate(ReorderBuffer *rb,
 											ReorderBufferTXN *txn,
 											bool addition, Size sz);
 
+/*
+ * ---------------------------------------
+ * spill files compression
+ * ---------------------------------------
+ */
+static void *ReorderBufferNewCompressorState(MemoryContext context,
+											 int compression_method);
+static void ReorderBufferFreeCompressorState(MemoryContext context,
+											 int compression_method,
+											 void *compressor_state);
+static void ReorderBufferCompress(ReorderBuffer *rb,
+								  ReorderBufferDiskChange **ondisk,
+								  int compression_method, Size data_size,
+								  void *compressor_state);
+static void ReorderBufferDecompress(ReorderBuffer *rb, char *data,
+									ReorderBufferDiskChange *ondisk,
+									void *compressor_state);
+
 /*
  * Allocate a new ReorderBuffer and clean out any old serialized state from
  * prior ReorderBuffer instances for the same slot.
@@ -432,6 +458,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;
 }
@@ -469,6 +497,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);
 
@@ -3806,12 +3838,12 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 							 int fd, ReorderBufferChange *change)
 {
 	ReorderBufferDiskChange *ondisk;
-	Size		sz = sizeof(ReorderBufferDiskChange);
+	Size		sz = sizeof(ReorderBufferDiskChange) + sizeof(ReorderBufferChange);
 
 	ReorderBufferSerializeReserve(rb, sz);
 
 	ondisk = (ReorderBufferDiskChange *) rb->outbuf;
-	memcpy(&ondisk->change, change, sizeof(ReorderBufferChange));
+	memcpy((char *) rb->outbuf + sizeof(ReorderBufferDiskChange), change, sizeof(ReorderBufferChange));
 
 	switch (change->action)
 	{
@@ -3847,7 +3879,7 @@ 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(ReorderBufferDiskChange) + sizeof(ReorderBufferChange);
 				/* might have been reallocated above */
 				ondisk = (ReorderBufferDiskChange *) rb->outbuf;
 
@@ -3879,7 +3911,7 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 					sizeof(Size) + sizeof(Size);
 				ReorderBufferSerializeReserve(rb, sz);
 
-				data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange);
+				data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange) + sizeof(ReorderBufferChange);
 
 				/* might have been reallocated above */
 				ondisk = (ReorderBufferDiskChange *) rb->outbuf;
@@ -3909,7 +3941,7 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 				sz += inval_size;
 
 				ReorderBufferSerializeReserve(rb, sz);
-				data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange);
+				data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange) + sizeof(ReorderBufferChange);
 
 				/* might have been reallocated above */
 				ondisk = (ReorderBufferDiskChange *) rb->outbuf;
@@ -3931,7 +3963,7 @@ 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(ReorderBufferDiskChange) + sizeof(ReorderBufferChange);
 				/* might have been reallocated above */
 				ondisk = (ReorderBufferDiskChange *) rb->outbuf;
 
@@ -3965,7 +3997,7 @@ 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(ReorderBufferDiskChange) + sizeof(ReorderBufferChange);
 				/* might have been reallocated above */
 				ondisk = (ReorderBufferDiskChange *) rb->outbuf;
 
@@ -3982,7 +4014,9 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 			break;
 	}
 
-	ondisk->size = sz;
+	/* Inplace ReorderBuffer content compression before writing it on disk */
+	ReorderBufferCompress(rb, &ondisk, logical_decoding_spill_compression,
+						  sz, txn->compressor_state);
 
 	errno = 0;
 	pgstat_report_wait_start(WAIT_EVENT_REORDER_BUFFER_WRITE);
@@ -4011,8 +4045,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. */
@@ -4281,9 +4313,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)
@@ -4322,60 +4351,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 to 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
@@ -4388,6 +4372,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;
+	ReorderBufferDiskChange *ondisk;
+	char	   *header;			/* 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(ReorderBufferDiskChange));
+	readBytes = FileRead(file->vfd, header,
+						 sizeof(ReorderBufferDiskChange),
+						 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(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 *) header;
+
+	/* Read ondisk data */
+	data = (char *) palloc0(ondisk->size - sizeof(ReorderBufferDiskChange));
+	readBytes = FileRead(file->vfd,
+						 data,
+						 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)))));
+
+	/* Decompress data */
+	ReorderBufferDecompress(rb, data, ondisk, 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.
@@ -4400,17 +4461,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(ReorderBufferDiskChange), sizeof(ReorderBufferChange));
 
-	data += sizeof(ReorderBufferDiskChange);
+	data += sizeof(ReorderBufferDiskChange) + sizeof(ReorderBufferChange);
 
 	/* restore individual stuff */
 	switch (change->action)
@@ -5338,3 +5396,161 @@ restart:
 		*cmax = ent->cmax;
 	return true;
 }
+
+/*
+ * Allocate a new Compressor State, depending on the compression method.
+ */
+static void *
+ReorderBufferNewCompressorState(MemoryContext context, int compression_method)
+{
+	switch (compression_method)
+	{
+		case REORDER_BUFFER_PGLZ_COMPRESSION:
+			return pglz_NewCompressorState(context);
+			break;
+		case REORDER_BUFFER_NO_COMPRESSION:
+		default:
+			return NULL;
+			break;
+	}
+}
+
+/*
+ * Free memory allocated to a Compressor State, depending on the compression
+ * method.
+ */
+static void
+ReorderBufferFreeCompressorState(MemoryContext context, int compression_method,
+								 void *compressor_state)
+{
+	switch (compression_method)
+	{
+		case REORDER_BUFFER_PGLZ_COMPRESSION:
+			return pglz_FreeCompressorState(context, compressor_state);
+			break;
+		case REORDER_BUFFER_NO_COMPRESSION:
+		default:
+			break;
+	}
+}
+
+/*
+ * Compress ReorderBuffer content. This function is called in order to compress
+ * data before spilling on disk.
+ */
+static void
+ReorderBufferCompress(ReorderBuffer *rb, ReorderBufferDiskChange **ondisk,
+					  int compression_method, Size data_size, void *compressor_state)
+{
+	ReorderBufferDiskChange *hdr = *ondisk;
+
+	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(ReorderBufferDiskChange);
+
+				break;
+			}
+			/* PGLZ compression */
+		case REORDER_BUFFER_PGLZ_COMPRESSION:
+			{
+				int32		dst_size = 0;
+				char	   *src = (char *) rb->outbuf + sizeof(ReorderBufferDiskChange);
+				int32		src_size = data_size - sizeof(ReorderBufferDiskChange);
+				int32		max_size = PGLZ_MAX_OUTPUT(src_size);
+				PGLZCompressorState *cstate = (PGLZCompressorState *) compressor_state;
+
+				/*
+				 * Make sure the buffer used for data compression has enough
+				 * space.
+				 */
+				enlargeStringInfo(cstate->buf, max_size);
+
+				dst_size = pglz_compress(src, src_size, cstate->buf->data, PGLZ_strategy_always);
+
+				cstate->buf->len = dst_size;
+
+				if (dst_size < 0)
+					ereport(ERROR,
+							(errcode(ERRCODE_DATA_CORRUPTED),
+							 errmsg_internal("PGLZ compression failed")));
+
+				ReorderBufferSerializeReserve(rb, (Size) (dst_size + sizeof(ReorderBufferDiskChange)));
+
+				hdr = (ReorderBufferDiskChange *) rb->outbuf;
+				hdr->comp_strat = REORDER_BUFFER_STRAT_PGLZ;
+				hdr->size = (Size) dst_size + sizeof(ReorderBufferDiskChange);
+				hdr->raw_size = (Size) src_size;
+
+				*ondisk = hdr;
+
+				/* Copy back compressed data into the ReorderBuffer */
+				memcpy((char *) rb->outbuf + sizeof(ReorderBufferDiskChange),
+					   cstate->buf->data, cstate->buf->len);
+
+				break;
+			}
+		default:
+			/* Other compression methods not yet supported */
+			break;
+	}
+}
+
+/*
+ * Decompress data read from disk and copy it into the ReorderBuffer.
+ */
+static void
+ReorderBufferDecompress(ReorderBuffer *rb, char *data,
+						ReorderBufferDiskChange *ondisk, void *compressor_state)
+{
+	/*
+	 * Make sure the output reorder buffer has enough space to store
+	 * decompressed/raw data.
+	 */
+	ReorderBufferSerializeReserve(rb, (Size) (ondisk->raw_size + sizeof(ReorderBufferDiskChange)));
+
+	/* Make a copy of the header read on disk into the ReorderBuffer */
+	memcpy(rb->outbuf, (char *) ondisk, sizeof(ReorderBufferDiskChange));
+
+	switch (ondisk->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(ReorderBufferDiskChange),
+					   data, ondisk->raw_size);
+				break;
+			}
+			/* PGLZ decompression */
+		case REORDER_BUFFER_STRAT_PGLZ:
+			{
+				char	   *buf = NULL;
+				int32		src_size = (int32) ondisk->size - sizeof(ReorderBufferDiskChange);
+				int32		buf_size = (int32) ondisk->raw_size;
+				int32		decBytes;
+
+				/* Decompress data directly into the ReorderBuffer */
+				buf = (char *) rb->outbuf;
+				buf += sizeof(ReorderBufferDiskChange);
+
+				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/backend/replication/logical/reorderbuffer_pglz.c b/src/backend/replication/logical/reorderbuffer_pglz.c
new file mode 100644
index 0000000000..1dea81c9fe
--- /dev/null
+++ b/src/backend/replication/logical/reorderbuffer_pglz.c
@@ -0,0 +1,56 @@
+/*-------------------------------------------------------------------------
+ *
+ * reorderbuffer_pglz.c
+ *	  Functions used for ReorderBuffer compression using PGLZ.
+ *
+ * Copyright (c) 2024-2024, PostgreSQL Global Development Group
+ *
+ *
+ * IDENTIFICATION
+ *	  src/backend/access/common/reorderbuffer_pglz.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+#include "replication/reorderbuffer_compression.h"
+
+/*
+ * Allocate a new PGLZCompressorState.
+ */
+void *
+pglz_NewCompressorState(MemoryContext context)
+{
+	PGLZCompressorState *cstate;
+	MemoryContext oldcontext = MemoryContextSwitchTo(context);
+
+	cstate = (PGLZCompressorState *)
+		MemoryContextAlloc(context, sizeof(PGLZCompressorState));
+
+	cstate->buf = makeStringInfo();
+
+	MemoryContextSwitchTo(oldcontext);
+
+	return (void *) cstate;
+}
+
+/*
+ * Free PGLZ memory resources and compressor state.
+ */
+void
+pglz_FreeCompressorState(MemoryContext context, void *compressor_state)
+{
+	PGLZCompressorState *cstate;
+	MemoryContext oldcontext;
+
+	if (compressor_state == NULL)
+		return;
+
+	oldcontext = MemoryContextSwitchTo(context);
+
+	cstate = (PGLZCompressorState *) compressor_state;
+
+	destroyStringInfo(cstate->buf);
+	pfree(compressor_state);
+
+	MemoryContextSwitchTo(oldcontext);
+}
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index 6ad5a8cb9c..5f231b5f90 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -26,6 +26,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
@@ -426,6 +427,11 @@ typedef struct ReorderBufferTXN
 	 * Private data pointer of the output plugin.
 	 */
 	void	   *output_plugin_private;
+
+	/*
+	 * Streaming compression state used for spill files compression.
+	 */
+	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 0000000000..9e9565ca7f
--- /dev/null
+++ b/src/include/replication/reorderbuffer_compression.h
@@ -0,0 +1,42 @@
+/*-------------------------------------------------------------------------
+ *
+ * reorderbuffer_compression.h
+ *	  ReorderBuffer spill files compression.
+ *
+ * Copyright (c) 2024-2024, PostgreSQL Global Development Group
+ *
+ * src/include/access/reorderbuffer_compression.h
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#ifndef REORDERBUFFER_COMPRESSION_H
+#define REORDERBUFFER_COMPRESSION_H
+
+/* ReorderBuffer on disk compression algorithms */
+typedef enum ReorderBufferCompressionMethod
+{
+	REORDER_BUFFER_NO_COMPRESSION,
+	REORDER_BUFFER_PGLZ_COMPRESSION,
+}			ReorderBufferCompressionMethod;
+
+/*
+ * Compression strategy applied to ReorderBuffer records spilled on disk
+ */
+typedef enum ReorderBufferCompressionStrategy
+{
+	REORDER_BUFFER_STRAT_UNCOMPRESSED,
+	REORDER_BUFFER_STRAT_PGLZ,
+}			ReorderBufferCompressionStrategy;
+
+typedef struct PGLZCompressorState
+{
+	/* Buffer used to store compressed data */
+	StringInfo	buf;
+}			PGLZCompressorState;
+
+extern void *pglz_NewCompressorState(MemoryContext context);
+extern void pglz_FreeCompressorState(MemoryContext context,
+									 void *compressor_state);
+
+#endif							/* REORDERBUFFER_COMPRESSION_H */
-- 
2.43.0

