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