From ecf8641e588fd94b748e8d53573e8c959d7b1d14 Mon Sep 17 00:00:00 2001 From: Andrey Borodin Date: Sat, 24 Oct 2026 20:18:40 +0500 Subject: [PATCH v12 1/3] Add whole-record WAL compression alongside FPI compression Compressing full-page images separately cannot exploit similarities between images in one WAL record, or compress records without images. Allow lz4 and zstd to compress the assembled record instead when it exceeds wal_compression_threshold. Keep per-image compression as the fallback, and leave pglz unchanged. Prepare the workspaces outside critical sections, keeping the smaller initial allocation for FPI-only settings. Reject compressed records that do not save space including their headers. Preserve physical and decoded lengths separately for pg_waldump statistics. This changes the WAL format and requires an XLOG_PAGE_MAGIC bump at commit time. Add coverage for compression, replay and crash recovery. Author: Andrey Borodin Discussion: https://www.postgresql.org/message-id/4DC38068-976E-4A84-8EE6-4EFACBBD927A@yandex-team.ru --- doc/src/sgml/config.sgml | 40 +- doc/src/sgml/monitoring.sgml | 6 +- doc/src/sgml/ref/pg_waldump.sgml | 9 + src/backend/access/transam/xlog.c | 20 +- src/backend/access/transam/xloginsert.c | 370 ++++++++++++++++-- src/backend/access/transam/xlogreader.c | 334 ++++++++++++++-- src/backend/access/transam/xlogstats.c | 5 +- src/backend/utils/misc/guc_parameters.dat | 19 +- src/backend/utils/misc/postgresql.conf.sample | 5 +- src/bin/pg_waldump/pg_waldump.c | 30 +- src/include/access/xlog.h | 1 + src/include/access/xlogreader.h | 9 +- src/include/access/xlogrecord.h | 48 ++- src/include/access/xlogstats.h | 1 + src/include/utils/guc_hooks.h | 2 + src/test/perl/PostgreSQL/Test/Cluster.pm | 20 +- src/test/recovery/Makefile | 1 + src/test/recovery/meson.build | 1 + .../recovery/t/026_overwrite_contrecord.pl | 4 +- src/test/recovery/t/061_wal_compression.pl | 204 ++++++++++ src/tools/pgindent/typedefs.list | 1 + 21 files changed, 1024 insertions(+), 106 deletions(-) create mode 100644 src/test/recovery/t/061_wal_compression.pl diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index e537b3b0043..1b9b39f5179 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -3678,8 +3678,11 @@ include_dir 'conf.d' When enabled, the PostgreSQL server compresses full page images written to WAL (e.g. when is on, during a base backup, - etc.). - A compressed page image will be decompressed during WAL replay. + etc.). With lz4 or zstd, a + record larger than + is instead compressed as a whole, which usually does better when + one record carries several page images. + Compressed data will be decompressed during WAL replay. The supported methods are off, on, zstd (if PostgreSQL was compiled with ), lz4 (if PostgreSQL was compiled with ), and @@ -3700,6 +3703,39 @@ include_dir 'conf.d' + + wal_compression_threshold (integer) + + wal_compression_threshold configuration parameter + + + + + Smallest WAL record that + compresses as a whole rather than compressing each of its full page + images separately. If this value is specified without units, it is + taken as bytes. The default is 512 bytes. Only records compressed + with lz4 or zstd are affected; + pglz always compresses page images individually. + Only superusers and users with the appropriate SET + privilege can change this setting. + + + + Lowering this value does not simply compress more. A record taken by + whole-record compression no longer has its page images compressed + individually. The compressed record, including its header, is used + only if smaller than the uncompressed record, but it is not compared + with per-image compression. A lower threshold can therefore produce + more WAL than per-image compression. + Setting it above the size of the largest possible WAL + record disables whole-record compression entirely, leaving the + page-image compression that + performed before this setting existed. + + + + wal_init_zero (boolean) diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index 52d4241a2ff..d67093ac8ce 100644 --- a/doc/src/sgml/monitoring.sgml +++ b/doc/src/sgml/monitoring.sgml @@ -3977,7 +3977,11 @@ description | Waiting for a newly initialized WAL file to reach durable storage wal_fpi_bytes numeric - Total amount of WAL full page images in bytes + Total amount of WAL full page images in bytes, after per-image + compression but before whole-record compression. This can exceed + wal_bytes when whole records are compressed: + their compressed payload cannot be split into image and non-image + portions. diff --git a/doc/src/sgml/ref/pg_waldump.sgml b/doc/src/sgml/ref/pg_waldump.sgml index 9bbb4bd5772..14fd7d2fea1 100644 --- a/doc/src/sgml/ref/pg_waldump.sgml +++ b/doc/src/sgml/ref/pg_waldump.sgml @@ -270,6 +270,15 @@ PostgreSQL documentation generate statistics per-record instead of per-rmgr. + + Record, FPI, and combined sizes are measured before whole-record + compression, but after any per-image compression. Physical size + includes compression headers and compressed payloads, but excludes + WAL page headers and alignment padding. Individual compressed + records likewise show their physical length separately from the + decoded len (rec/tot) values. + + If pg_waldump is terminated by signal SIGINT diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 63d214782da..8af539f30b2 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -142,6 +142,7 @@ int wal_retrieve_retry_interval = 5000; int max_slot_wal_keep_size_mb = -1; int wal_decode_buffer_size = 512 * 1024; bool track_wal_io_timing = false; +int wal_compression_threshold = 512; #ifdef WAL_DEBUG bool XLOG_DEBUG = false; @@ -805,6 +806,22 @@ static void WALInsertLockUpdateInsertingAt(XLogRecPtr insertingAt); static XLogRecPtr XLogChecksums(uint32 new_type); +#ifdef WAL_DEBUG +/* Read length of a record, accounting for possible compression */ +static uint32 +XLogGetRecordTotalLen(XLogRecord *record) +{ + if (record->xl_info & XLR_COMPRESSED) + { + XLogCompressionHeader *c = (XLogCompressionHeader *) record; + + Assert(c->decompressed_length > 0); + return c->decompressed_length; + } + return record->xl_tot_len; +} +#endif + /* * Insert an XLOG record represented by an already-constructed chain of data * chunks. This is a low-level routine; to construct the WAL record header @@ -1125,7 +1142,7 @@ XLogInsertRecord(XLogRecData *rdata, /* We also need temporary space to decode the record. */ record = (XLogRecord *) recordBuf.data; decoded = (DecodedXLogRecord *) - palloc(DecodeXLogRecordRequiredSpace(record->xl_tot_len)); + palloc(DecodeXLogRecordRequiredSpace(XLogGetRecordTotalLen(record))); if (!debug_reader) debug_reader = XLogReaderAllocate(wal_segment_size, NULL, @@ -1175,6 +1192,7 @@ XLogInsertRecord(XLogRecData *rdata, pgWalUsage.wal_bytes += rechdr->xl_tot_len; pgWalUsage.wal_records++; pgWalUsage.wal_fpi += num_fpi; + /* Image bytes cannot be isolated after whole-record compression. */ pgWalUsage.wal_fpi_bytes += fpi_bytes; /* Required for the flush of pending stats WAL data */ diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c index 59b0dfe436d..5d105cb6d3b 100644 --- a/src/backend/access/transam/xloginsert.c +++ b/src/backend/access/transam/xloginsert.c @@ -39,6 +39,7 @@ #include "replication/origin.h" #include "storage/bufmgr.h" #include "storage/proc.h" +#include "utils/guc_hooks.h" #include "utils/memutils.h" #include "utils/pgstat_internal.h" #include "utils/rel.h" @@ -85,8 +86,8 @@ typedef struct XLogRecData bkp_rdatas[2]; /* temporary rdatas used to hold references to * backup block data in XLogRecordAssemble() */ - /* buffer to store a compressed version of backup block image */ - char compressed_page[COMPRESS_BUFSIZE]; + /* pointer into compression_buf for compressed page image */ + char *compressed_page; } registered_buffer; static registered_buffer *registered_buffers; @@ -119,6 +120,15 @@ static char *hdr_scratch = NULL; #ifdef USE_ZSTD /* Compression context reused for zstd-compressed full-page images. */ static ZSTD_CCtx *zstd_cctx = NULL; + +/* Get this backend's zstd context, creating it on first use. NULL on OOM. */ +static ZSTD_CCtx * +GetZstdCCtx(void) +{ + if (zstd_cctx == NULL) + zstd_cctx = ZSTD_createCCtx(); + return zstd_cctx; +} #endif #define SizeOfXlogOrigin (sizeof(ReplOriginId) + sizeof(char)) @@ -130,6 +140,30 @@ static ZSTD_CCtx *zstd_cctx = NULL; SizeOfXLogRecordDataHeaderLong + SizeOfXlogOrigin + \ SizeOfXLogTransactionId) +/* + * Size of the compression buffers: enough for a compressed image of every + * block a record may reference, plus the record headers. Larger records can + * contain arbitrary amounts of main data; do not compress those as a whole. + * Keep this MAXALIGN'd so that the second workspace can hold a record header. + */ +#define WAL_COMPRESSION_BUFSIZE \ + MAXALIGN((XLR_MAX_BLOCK_ID + 1) * COMPRESS_BUFSIZE + HEADER_SCRATCH_SIZE) + +/* + * FPI-only compression starts with space for max_registered_buffers images. + * Whole-record compression needs two maximum-sized workspaces, allocated + * together when enabled. compression_buf serves two uses that never overlap: per-FPI + * compression packs block images into it, tracked by compression_buf_offset, + * while whole-record compression flattens the rdt chain into it and compresses + * into compressed_data. + */ +static char *compression_buf = NULL; +static size_t compression_buf_size; +static int compression_buf_offset; /* fill level for FPI packing */ +#ifdef WAL_WHOLE_RECORD_COMPRESSION +static char *compressed_data = NULL; +#endif + /* * An array of XLogRecData structs, to hold registered data. */ @@ -142,11 +176,13 @@ static bool begininsert_called = false; /* Memory context to hold the registered buffer and data references. */ static MemoryContext xloginsert_cxt; +static void AllocCompressionBuffers(int method, int threshold); static XLogRecData *XLogRecordAssemble(RmgrId rmid, uint8 info, XLogRecPtr RedoRecPtr, bool doPageWrites, XLogRecPtr *fpw_lsn, int *num_fpi, uint64 *fpi_bytes, - bool *topxid_included); + bool *topxid_included, uint64 *rec_size, + bool skip_fpi_compression); static bool XLogCompressBackupBlock(const PageData *page, uint16 hole_offset, uint16 hole_length, void *dest, uint16 *dlen); @@ -221,6 +257,8 @@ XLogEnsureRecordSpace(int max_block_id, int ndatas) rdatas = repalloc_array(rdatas, XLogRecData, ndatas); max_rdatas = ndatas; } + + AllocCompressionBuffers(wal_compression, wal_compression_threshold); } /* @@ -471,6 +509,146 @@ XLogSetRecordFlags(uint8 flags) curinsert_flags |= flags; } +/* Compress the assembled record; NULL if that did not pay off */ +static XLogRecData * +XLogCompressRdt(XLogRecData *rdt) +{ +#ifndef WAL_WHOLE_RECORD_COMPRESSION + /* Without a supported method the caller never gets here */ + Assert(false); + return NULL; +#else + static XLogRecData compressed_rdt_hdr; + XLogCompressionHeader *compressed_header; + XLogRecord *src_header; + uint32 flat_len = 0; + uint32 orig_len; + int32 compr_len = -1; + int32 dest_size; + + Assert(wal_compression != WAL_COMPRESSION_NONE); + Assert(compression_buf != NULL); + + /* + * Flatten the rdt chain into compression_buf. The caller skipped per-FPI + * compression, so the buffer is free, and checked that the record fits. + */ + for (const XLogRecData *r = rdt; r != NULL; r = r->next) + { + memcpy(compression_buf + flat_len, r->data, r->len); + flat_len += r->len; + } + Assert(flat_len <= WAL_COMPRESSION_BUFSIZE); + + src_header = (XLogRecord *) compression_buf; + compressed_header = (XLogCompressionHeader *) compressed_data; + + /* Zero it first: the padding in the header reaches disk */ + memset(compressed_header, 0, SizeOfXLogCompressedRecord); + + compressed_header->record_header = *src_header; + compressed_header->decompressed_length = flat_len; + + orig_len = src_header->xl_tot_len - SizeOfXLogRecord; + + /* + * compressed_data is not sized to the worst-case expansion bound, so hand + * the compressor what is really left and let it fail if that is too + * little. A record that does not fit was not worth compressing anyway. + */ + dest_size = WAL_COMPRESSION_BUFSIZE - SizeOfXLogCompressedRecord; + + switch ((WalCompression) wal_compression) + { + case WAL_COMPRESSION_NONE: + case WAL_COMPRESSION_PGLZ: + /* caller does not use whole-record compression for these */ + Assert(false); + return NULL; + + case WAL_COMPRESSION_LZ4: +#ifdef USE_LZ4 + compressed_header->method = XLR_COMPRESS_LZ4; + compr_len = LZ4_compress_default((char *) &src_header[1], + (char *) &compressed_header[1], + orig_len, dest_size); + if (compr_len <= 0) + return NULL; +#else + elog(ERROR, "LZ4 is not supported by this build"); +#endif + break; + + case WAL_COMPRESSION_ZSTD: +#ifdef USE_ZSTD + { + size_t zstd_len; + + ZSTD_CCtx *cctx = GetZstdCCtx(); + + if (cctx == NULL) + return NULL; + + compressed_header->method = XLR_COMPRESS_ZSTD; + zstd_len = ZSTD_compressCCtx(cctx, + (char *) &compressed_header[1], + dest_size, + (char *) &src_header[1], orig_len, + ZSTD_CLEVEL_DEFAULT); + if (ZSTD_isError(zstd_len)) + return NULL; + compr_len = (int32) zstd_len; + } +#else + elog(ERROR, "zstd is not supported by this build"); +#endif + break; + + /* no default case, so that compiler will warn */ + } + + Assert(compr_len > 0 && compr_len <= dest_size); + + /* Include the compression header when deciding whether this saves space. */ + if (SizeOfXLogCompressedRecord + compr_len >= flat_len) + return NULL; + + compressed_header->record_header.xl_tot_len = + SizeOfXLogCompressedRecord + compr_len; + + compressed_header->record_header.xl_info |= XLR_COMPRESSED; + + compressed_rdt_hdr.data = compressed_data; + compressed_rdt_hdr.len = compressed_header->record_header.xl_tot_len; + compressed_rdt_hdr.next = NULL; + + return &compressed_rdt_hdr; +#endif +} + +/* Checksum assembled record (which may be compressed). */ +static void +XLogChecksumRecord(XLogRecData *rdt) +{ + pg_crc32c rdata_crc; + XLogRecord *rechdr = (XLogRecord *) rdt->data; + + /* + * Calculate CRC of the data + * + * Note that the record header isn't added into the CRC initially since we + * don't know the prev-link yet. Thus, the CRC will represent the CRC of + * the whole record in the order: rdata, then backup blocks, then record + * header. + */ + INIT_CRC32C(rdata_crc); + COMP_CRC32C(rdata_crc, ((char *) rdt->data) + SizeOfXLogRecord, + rdt->len - SizeOfXLogRecord); + for (rdt = rdt->next; rdt != NULL; rdt = rdt->next) + COMP_CRC32C(rdata_crc, rdt->data, rdt->len); + rechdr->xl_crc = rdata_crc; +} + /* * Insert an XLOG record having the specified RMID and info bytes, with the * body of the record being the data and buffer references registered earlier @@ -487,6 +665,17 @@ XLogInsert(RmgrId rmid, uint8 info) { XLogRecPtr EndPos; + /* + * Only lz4 and zstd compress whole records; pglz rarely beats its own + * minimum compression rate on a record made of full-page images. A + * threshold at or above the buffer size disables this entirely, which is + * how a user asks for full-page-image-only compression. + */ + bool try_whole_record = ((wal_compression == WAL_COMPRESSION_LZ4 || + wal_compression == WAL_COMPRESSION_ZSTD) && + wal_compression_threshold < + WAL_COMPRESSION_BUFSIZE); + /* XLogBeginInsert() must have been called. */ if (!begininsert_called) elog(ERROR, "XLogBeginInsert was not called"); @@ -522,6 +711,10 @@ XLogInsert(RmgrId rmid, uint8 info) XLogRecData *rdt; int num_fpi = 0; uint64 fpi_bytes = 0; + uint64 rec_size; + + /* Do not accumulate FPI data across retries of this loop */ + compression_buf_offset = 0; /* * Get values needed to decide whether to do full-page writes. Since @@ -532,7 +725,41 @@ XLogInsert(RmgrId rmid, uint8 info) rdt = XLogRecordAssemble(rmid, info, RedoRecPtr, doPageWrites, &fpw_lsn, &num_fpi, &fpi_bytes, - &topxid_included); + &topxid_included, &rec_size, + try_whole_record); + + /* + * Assembly skipped per-FPI compression on our behalf, so if the + * record is not compressed as a whole after all, reassemble it to get + * the per-FPI compression back. + */ + if (try_whole_record) + { + bool whole_record_compressed = false; + + if (rec_size > wal_compression_threshold && + rec_size <= WAL_COMPRESSION_BUFSIZE) + { + XLogRecData *rdt_compressed = XLogCompressRdt(rdt); + + if (rdt_compressed != NULL) + { + rdt = rdt_compressed; + whole_record_compressed = true; + } + } + + if (!whole_record_compressed && num_fpi > 0) + { + compression_buf_offset = 0; + rdt = XLogRecordAssemble(rmid, info, RedoRecPtr, doPageWrites, + &fpw_lsn, &num_fpi, &fpi_bytes, + &topxid_included, &rec_size, + false); + } + } + + XLogChecksumRecord(rdt); EndPos = XLogInsertRecord(rdt, fpw_lsn, curinsert_flags, num_fpi, fpi_bytes, topxid_included); @@ -606,6 +833,67 @@ XLogGetFakeLSN(Relation rel) } } +/* + * Prepare compression storage without allocating inside a critical section. + * + * Callers must be outside a critical section, which is why this happens at + * backend start, in GUC assign hooks, or in XLogEnsureRecordSpace(), never in + * XLogInsert(). Keep any existing allocation when the setting is reduced. + */ +static void +AllocCompressionBuffers(int method, int threshold) +{ + size_t required_size; + + Assert(CritSectionCount == 0); + Assert(xloginsert_cxt != NULL); + + if (method == WAL_COMPRESSION_NONE) + return; + + required_size = max_registered_buffers * COMPRESS_BUFSIZE; + +#ifdef WAL_WHOLE_RECORD_COMPRESSION + if ((method == WAL_COMPRESSION_LZ4 || method == WAL_COMPRESSION_ZSTD) && + threshold < WAL_COMPRESSION_BUFSIZE) + required_size = 2 * WAL_COMPRESSION_BUFSIZE; +#endif + + if (required_size > compression_buf_size) + { + if (compression_buf == NULL) + compression_buf = MemoryContextAlloc(xloginsert_cxt, required_size); + else + compression_buf = repalloc(compression_buf, required_size); + compression_buf_size = required_size; + } + +#ifdef WAL_WHOLE_RECORD_COMPRESSION + if (compression_buf_size >= 2 * WAL_COMPRESSION_BUFSIZE) + compressed_data = compression_buf + WAL_COMPRESSION_BUFSIZE; +#endif +} + +/* + * GUC assign hook for wal_compression. + * + * Runs before wal_compression itself is updated, hence the test on newval. If + * xloginsert_cxt does not exist yet, InitXLogInsert() will do the allocation. + */ +void +assign_wal_compression(int newval, void *extra) +{ + if (newval != WAL_COMPRESSION_NONE && xloginsert_cxt != NULL) + AllocCompressionBuffers(newval, wal_compression_threshold); +} + +void +assign_wal_compression_threshold(int newval, void *extra) +{ + if (xloginsert_cxt != NULL) + AllocCompressionBuffers(wal_compression, newval); +} + /* * Assemble a WAL record from the registered data and buffers into an * XLogRecData chain, ready for insertion with XLogInsertRecord(). @@ -625,12 +913,11 @@ static XLogRecData * XLogRecordAssemble(RmgrId rmid, uint8 info, XLogRecPtr RedoRecPtr, bool doPageWrites, XLogRecPtr *fpw_lsn, int *num_fpi, uint64 *fpi_bytes, - bool *topxid_included) + bool *topxid_included, uint64 *rec_size, + bool skip_fpi_compression) { - XLogRecData *rdt; uint64 total_len = 0; int block_id; - pg_crc32c rdata_crc; registered_buffer *prev_regbuf = NULL; XLogRecData *rdt_datas_last; XLogRecord *rechdr; @@ -658,12 +945,18 @@ XLogRecordAssemble(RmgrId rmid, uint8 info, if (wal_consistency_checking[rmid]) info |= XLR_CHECK_CONSISTENCY; + /* We are often in a critical section here, so we cannot allocate */ + Assert(wal_compression == WAL_COMPRESSION_NONE || compression_buf != NULL); + /* * Make an rdata chain containing all the data portions of all block * references. This includes the data for full-page images. Also append * the headers for the block references in the scratch buffer. */ *fpw_lsn = InvalidXLogRecPtr; + *num_fpi = 0; + *fpi_bytes = 0; + *topxid_included = false; for (block_id = 0; block_id < max_registered_block_id; block_id++) { registered_buffer *regbuf = ®istered_buffers[block_id]; @@ -761,15 +1054,26 @@ XLogRecordAssemble(RmgrId rmid, uint8 info, } /* - * Try to compress a block image if wal_compression is enabled + * Try to compress a block image if wal_compression is enabled and + * the caller has not reserved the buffer for whole-record + * compression. Each image gets its own slice of the buffer, + * which is large enough for all of them. */ - if (wal_compression != WAL_COMPRESSION_NONE) + if (!skip_fpi_compression && + wal_compression != WAL_COMPRESSION_NONE) { + Assert(compression_buf_offset + COMPRESS_BUFSIZE <= + compression_buf_size); + regbuf->compressed_page = compression_buf + compression_buf_offset; + is_compressed = XLogCompressBackupBlock(page, bimg.hole_offset, cbimg.hole_length, regbuf->compressed_page, &compressed_len); + + if (is_compressed) + compression_buf_offset += compressed_len; } /* @@ -973,19 +1277,6 @@ XLogRecordAssemble(RmgrId rmid, uint8 info, hdr_rdt.len = (scratch - hdr_scratch); total_len += hdr_rdt.len; - /* - * Calculate CRC of the data - * - * Note that the record header isn't added into the CRC initially since we - * don't know the prev-link yet. Thus, the CRC will represent the CRC of - * the whole record in the order: rdata, then backup blocks, then record - * header. - */ - INIT_CRC32C(rdata_crc); - COMP_CRC32C(rdata_crc, hdr_scratch + SizeOfXLogRecord, hdr_rdt.len - SizeOfXLogRecord); - for (rdt = hdr_rdt.next; rdt != NULL; rdt = rdt->next) - COMP_CRC32C(rdata_crc, rdt->data, rdt->len); - /* * Ensure that the XLogRecord is not too large. * @@ -1009,7 +1300,9 @@ XLogRecordAssemble(RmgrId rmid, uint8 info, rechdr->xl_info = info; rechdr->xl_rmid = rmid; rechdr->xl_prev = InvalidXLogRecPtr; - rechdr->xl_crc = rdata_crc; + rechdr->xl_crc = 0; + + *rec_size = rechdr->xl_tot_len; return &hdr_rdt; } @@ -1068,23 +1361,21 @@ XLogCompressBackupBlock(const PageData *page, uint16 hole_offset, uint16 hole_le case WAL_COMPRESSION_ZSTD: #ifdef USE_ZSTD - if (zstd_cctx == NULL) - zstd_cctx = ZSTD_createCCtx(); - - /* - * We may be in a critical section, so skip if the context - * allocation failed. - */ - if (zstd_cctx == NULL) { - len = -1; - break; - } + ZSTD_CCtx *cctx = GetZstdCCtx(); - len = ZSTD_compressCCtx(zstd_cctx, dest, COMPRESS_BUFSIZE, - source, orig_len, ZSTD_CLEVEL_DEFAULT); - if (ZSTD_isError(len)) - len = -1; /* failure */ + /* Compression is optional, even in a critical section. */ + if (cctx == NULL) + len = -1; /* out of memory; store the image as is */ + else + { + len = ZSTD_compressCCtx(cctx, dest, COMPRESS_BUFSIZE, + source, orig_len, + ZSTD_CLEVEL_DEFAULT); + if (ZSTD_isError(len)) + len = -1; /* failure */ + } + } #else elog(ERROR, "zstd is not supported by this build"); #endif @@ -1460,4 +1751,7 @@ InitXLogInsert(void) #ifdef WAL_DEBUG InitWalDebug(); #endif + + if (wal_compression != WAL_COMPRESSION_NONE) + AllocCompressionBuffers(wal_compression, wal_compression_threshold); } diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c index e05a831b6e6..dd139cb2d34 100644 --- a/src/backend/access/transam/xlogreader.c +++ b/src/backend/access/transam/xlogreader.c @@ -62,6 +62,12 @@ static void xlogreader_memory_context_reset_cb(void *arg); static void XLogReaderFreeZstdContext(void *arg); #endif #endif +static XLogRecord *XLogDecompressRecordIfNeeded(XLogReaderState *state, + XLogRecord *record, + XLogRecPtr recptr); +#ifdef USE_ZSTD +static ZSTD_DCtx *XLogReaderGetZstdContext(XLogReaderState *state); +#endif /* size of the buffer allocated for error message. */ #define MAX_ERRORMSG_LEN 1000 @@ -194,6 +200,8 @@ XLogReaderFree(XLogReaderState *state) ZSTD_freeDCtx(state->zstd_dctx); } #endif + if (state->decompression_buffer) + pfree(state->decompression_buffer); pfree(state->readBuf); pfree(state); } @@ -609,10 +617,12 @@ XLogDecodeNextRecord(XLogReaderState *state, bool nonblocking) XLogRecPtr targetPagePtr; bool randAccess; uint32 len, - total_len; + total_len_decomp, + total_len_physical; uint32 targetRecOff; uint32 pageHeaderSize; bool assembled; + bool have_decomp_len; bool gotheader; int readOff; DecodedXLogRecord *decoded; @@ -663,6 +673,14 @@ restart: state->currRecPtr = RecPtr; assembled = false; + /* + * Start each pass without a decode slot. We do not always take one below + * - a compressed record whose header straddles a page does not tell us + * how much space it needs yet - and carrying one over from a previous + * pass would hand out a slot that is already in the decode queue. + */ + decoded = NULL; + targetPagePtr = RecPtr - (RecPtr % XLOG_BLCKSZ); targetRecOff = RecPtr % XLOG_BLCKSZ; @@ -720,7 +738,70 @@ restart: * whole header. */ record = (XLogRecord *) (state->readBuf + RecPtr % XLOG_BLCKSZ); - total_len = record->xl_tot_len; + total_len_physical = record->xl_tot_len; + + /* + * Size the record will occupy once decoded, which for a compressed record + * is its decompressed length rather than xl_tot_len. + * + * Only xl_tot_len is guaranteed to be on this page; xl_info may live on + * the next one when the record starts near the end of a page. If we + * cannot tell yet, have_decomp_len stays false and we allocate after + * assembly instead. + */ + total_len_decomp = 0; + have_decomp_len = false; + + if (targetRecOff <= XLOG_BLCKSZ - SizeOfXLogRecord) + { + /* Full header is on this page; safe to read xl_info. */ + if (record->xl_info & XLR_COMPRESSED) + { + /* + * Skip if the compression header spans pages, or if the record is + * too short to be a valid compressed one. + */ + if (targetRecOff <= XLOG_BLCKSZ - SizeOfXLogCompressedRecord && + total_len_physical >= SizeOfXLogCompressedRecord) + { + XLogCompressionHeader *c = (XLogCompressionHeader *) record; + uint32 dlen = c->decompressed_length; + bool valid_method; + + valid_method = false +#ifdef USE_LZ4 + || (c->method == XLR_COMPRESS_LZ4) +#endif +#ifdef USE_ZSTD + || (c->method == XLR_COMPRESS_ZSTD) +#endif + ; + + /* Sanity-check before using for allocation */ + if (dlen > SizeOfXLogRecord && dlen <= XLogRecordMaxSize && + valid_method) + { + total_len_decomp = dlen; + have_decomp_len = true; + } + } + } + else + { + total_len_decomp = record->xl_tot_len; + have_decomp_len = true; + } + } + else + { + /* + * Header spans pages, so we cannot read xl_info and do not know + * whether this is a compressed record, let alone how big it decodes + * to. xl_tot_len would be the compressed size, and reserving that + * much would leave the decoder writing past its slot, so leave the + * allocation to the pass after assembly. + */ + } /* * If the whole record header is on this page, validate it immediately. @@ -736,16 +817,18 @@ restart: randAccess)) goto err; gotheader = true; + if (record->xl_info & XLR_COMPRESSED) + gotheader = targetRecOff <= XLOG_BLCKSZ - SizeOfXLogCompressedRecord; } else { /* There may be no next page if it's too small. */ - if (total_len < SizeOfXLogRecord) + if (total_len_physical < SizeOfXLogRecord) { report_invalid_record(state, "invalid record length at %X/%08X: expected at least %u, got %u", LSN_FORMAT_ARGS(RecPtr), - (uint32) SizeOfXLogRecord, total_len); + (uint32) SizeOfXLogRecord, total_len_physical); goto err; } @@ -754,12 +837,12 @@ restart: * reconstruct it. The backend enforces the same limit in * XLogRecordAssemble(). */ - if (total_len > XLogRecordMaxSize) + if (total_len_physical > XLogRecordMaxSize) { report_invalid_record(state, "invalid record length at %X/%08X: expected at most %u, got %u", LSN_FORMAT_ARGS(RecPtr), - XLogRecordMaxSize, total_len); + XLogRecordMaxSize, total_len_physical); goto err; } @@ -772,21 +855,24 @@ restart: * calling palloc. If we can't, we'll try again below after we've * validated that total_len isn't garbage bytes from a recycled WAL page. */ - decoded = XLogReadRecordAlloc(state, - total_len, - false /* allow_oversized */ ); + if (have_decomp_len) + decoded = XLogReadRecordAlloc(state, + total_len_decomp, + false /* allow_oversized */ ); + if (decoded == NULL && nonblocking) { /* - * There is no space in the circular decode buffer, and the caller is - * only reading ahead. The caller should consume existing records to - * make space. + * Either there is no space in the circular decode buffer, or we could + * not tell how much this record needs. Either way the caller is only + * reading ahead, so let it replay what it has; it will come back for + * this record with nonblocking off, where we may allocate. */ return XLREAD_WOULDBLOCK; } len = XLOG_BLCKSZ - RecPtr % XLOG_BLCKSZ; - if (total_len > len) + if (total_len_physical > len) { /* Need to reassemble record */ char *contdata; @@ -858,19 +944,19 @@ restart: * we expect there to be left. */ if (pageHeader->xlp_rem_len == 0 || - total_len != (pageHeader->xlp_rem_len + gotlen)) + total_len_physical != (pageHeader->xlp_rem_len + gotlen)) { report_invalid_record(state, "invalid contrecord length %u (expected %lld) at %X/%08X", pageHeader->xlp_rem_len, - ((long long) total_len) - gotlen, + ((long long) total_len_physical) - gotlen, LSN_FORMAT_ARGS(RecPtr)); goto err; } /* Wait for the next page to become available */ readOff = ReadPageInternal(state, targetPagePtr, - Min(total_len - gotlen + SizeOfXLogShortPHD, + Min(total_len_physical - gotlen + SizeOfXLogShortPHD, XLOG_BLCKSZ)); if (readOff == XLREAD_WOULDBLOCK) return XLREAD_WOULDBLOCK; @@ -915,7 +1001,7 @@ restart: * also cross-checked total_len against xlp_rem_len on the second * page, and verified xlp_pageaddr on both. */ - if (total_len > state->readRecordBufSize) + if (total_len_physical > state->readRecordBufSize) { char save_copy[XLOG_BLCKSZ * 2]; @@ -926,11 +1012,11 @@ restart: Assert(gotlen <= lengthof(save_copy)); Assert(gotlen <= state->readRecordBufSize); memcpy(save_copy, state->readRecordBuf, gotlen); - allocate_recordbuf(state, total_len); + allocate_recordbuf(state, total_len_physical); memcpy(state->readRecordBuf, save_copy, gotlen); buffer = state->readRecordBuf + gotlen; } - } while (gotlen < total_len); + } while (gotlen < total_len_physical); Assert(gotheader); record = (XLogRecord *) state->readRecordBuf; @@ -945,8 +1031,9 @@ restart: else { /* Wait for the record data to become available */ + Assert(targetRecOff + total_len_physical <= XLOG_BLCKSZ); readOff = ReadPageInternal(state, targetPagePtr, - Min(targetRecOff + total_len, XLOG_BLCKSZ)); + targetRecOff + total_len_physical); if (readOff == XLREAD_WOULDBLOCK) return XLREAD_WOULDBLOCK; else if (readOff < 0) @@ -956,7 +1043,7 @@ restart: if (!ValidXLogRecord(state, record, RecPtr)) goto err; - state->NextRecPtr = RecPtr + MAXALIGN(total_len); + state->NextRecPtr = RecPtr + MAXALIGN(total_len_physical); state->DecodeRecPtr = RecPtr; } @@ -979,8 +1066,20 @@ restart: if (decoded == NULL) { Assert(!nonblocking); + + /* total_len_decomp may not yet reflect the actual decompressed size */ + if (record->xl_info & XLR_COMPRESSED) + { + XLogCompressionHeader *c = (XLogCompressionHeader *) record; + + Assert(c->decompressed_length > SizeOfXLogRecord); + Assert(c->decompressed_length <= XLogRecordMaxSize); + total_len_decomp = c->decompressed_length; + } + else + total_len_decomp = record->xl_tot_len; decoded = XLogReadRecordAlloc(state, - total_len, + total_len_decomp, true /* allow_oversized */ ); /* allocation should always happen under allow_oversized */ Assert(decoded != NULL); @@ -1322,6 +1421,25 @@ ValidXLogRecord(XLogReaderState *state, XLogRecord *record, XLogRecPtr recptr) return false; } + if (record->xl_info & XLR_COMPRESSED) + { + XLogCompressionHeader *header = (XLogCompressionHeader *) record; + + /* + * Validate before using the decompressed length to size a decode + * slot. + */ + if (record->xl_tot_len <= SizeOfXLogCompressedRecord || + header->decompressed_length <= SizeOfXLogRecord || + header->decompressed_length > XLogRecordMaxSize) + { + report_invalid_record(state, + "invalid compressed record length at %X/%08X", + LSN_FORMAT_ARGS(recptr)); + return false; + } + } + return true; } @@ -1786,6 +1904,147 @@ DecodeXLogRecordRequiredSpace(size_t xl_tot_len) return size; } +#ifdef USE_ZSTD +/* Reuse one context for independent compressed records and page images. */ +static ZSTD_DCtx * +XLogReaderGetZstdContext(XLogReaderState *state) +{ + if (state->zstd_dctx == NULL) + { + state->zstd_dctx = ZSTD_createDCtx(); + if (state->zstd_dctx == NULL) + return NULL; + +#ifndef FRONTEND + /* The reader may outlive the current memory context. */ + state->zstd_dctx_cb.func = XLogReaderFreeZstdContext; + state->zstd_dctx_cb.arg = state->zstd_dctx; + MemoryContextRegisterResetCallback(GetMemoryChunkContext(state), + &state->zstd_dctx_cb); +#endif + } + return state->zstd_dctx; +} +#endif + +static XLogRecord * +XLogDecompressRecordIfNeeded(XLogReaderState *state, + XLogRecord *record, + XLogRecPtr recptr) +{ + if (record->xl_info & XLR_COMPRESSED) + { +#ifndef WAL_WHOLE_RECORD_COMPRESSION + report_invalid_record(state, + "could not decompress record at %X/%08X compressed with method %u not supported by build", + LSN_FORMAT_ARGS(recptr), + ((XLogCompressionHeader *) record)->method); + return NULL; +#else + XLogCompressionHeader *src = (XLogCompressionHeader *) record; + + /* decompressed_length covers the XLogRecord header + body */ + uint32 body_len = src->decompressed_length - SizeOfXLogRecord; + uint32 srclen = src->record_header.xl_tot_len - SizeOfXLogCompressedRecord; + bool decomp_success = true; + char *dst; + XLogRecord *dst_h; + + /* + * Grow the decompression buffer if needed, rounding up to BLCKSZ to + * avoid frequent small reallocations. Since the buffer content is + * always fully overwritten, we simply pfree and reallocate. + */ + if (state->decompression_buffer_size < src->decompressed_length) + { + uint32 new_size = (uint32) TYPEALIGN(BLCKSZ, src->decompressed_length); + + if (state->decompression_buffer) + pfree(state->decompression_buffer); +#ifndef FRONTEND + state->decompression_buffer = + MemoryContextAllocExtended(GetMemoryChunkContext(state), + new_size, MCXT_ALLOC_NO_OOM); +#else + state->decompression_buffer = + palloc_extended(new_size, MCXT_ALLOC_NO_OOM); +#endif + if (!state->decompression_buffer) + { + state->decompression_buffer_size = 0; + report_invalid_record(state, + "out of memory while decompressing record at %X/%08X", + LSN_FORMAT_ARGS(recptr)); + return NULL; + } + state->decompression_buffer_size = new_size; + } + + dst_h = (XLogRecord *) state->decompression_buffer; + *dst_h = src->record_header; + dst_h->xl_tot_len = src->decompressed_length; + dst = (char *) &dst_h[1]; + + if (src->method == XLR_COMPRESS_LZ4) + { +#ifdef USE_LZ4 + if (LZ4_decompress_safe((char *) &src[1], dst, + srclen, body_len) != body_len) + decomp_success = false; +#else + report_invalid_record(state, + "could not decompress record at %X/%08X compressed with %s not supported by build", + LSN_FORMAT_ARGS(recptr), "lz4"); + return NULL; +#endif + } + else if (src->method == XLR_COMPRESS_ZSTD) + { +#ifdef USE_ZSTD + size_t decomp_result; + ZSTD_DCtx *dctx = XLogReaderGetZstdContext(state); + + if (dctx == NULL) + { + report_invalid_record(state, + "out of memory while decompressing record at %X/%08X", + LSN_FORMAT_ARGS(recptr)); + return NULL; + } + decomp_result = ZSTD_decompressDCtx(dctx, dst, body_len, + (char *) &src[1], srclen); + + if (ZSTD_isError(decomp_result) || decomp_result != body_len) + decomp_success = false; +#else + report_invalid_record(state, + "could not decompress record at %X/%08X compressed with %s not supported by build", + LSN_FORMAT_ARGS(recptr), "zstd"); + return NULL; +#endif + } + else + { + report_invalid_record(state, + "could not decompress record at %X/%08X compressed with unknown method", + LSN_FORMAT_ARGS(recptr)); + return NULL; + } + + if (!decomp_success) + { + report_invalid_record(state, + "could not decompress record at %X/%08X", + LSN_FORMAT_ARGS(recptr)); + return NULL; + } + + return (XLogRecord *) state->decompression_buffer; +#endif + } + return record; +} + /* * Decode a record. "decoded" must point to a MAXALIGNed memory area that has * space for at least DecodeXLogRecordRequiredSpace(record) bytes. On @@ -1824,6 +2083,15 @@ DecodeXLogRecord(XLogReaderState *state, RelFileLocator *rlocator = NULL; uint8 block_id; + decoded->physical_length = record->xl_tot_len; + record = XLogDecompressRecordIfNeeded(state, record, lsn); + + if (!record) + { + /* Decompression failed, error must be reported already */ + return false; + } + decoded->header = *record; decoded->lsn = lsn; decoded->next = NULL; @@ -2249,24 +2517,12 @@ RestoreBlockImage(XLogReaderState *record, uint8 block_id, char *page) #ifdef USE_ZSTD size_t decomp_result; - if (record->zstd_dctx == NULL) + if (XLogReaderGetZstdContext(record) == NULL) { - record->zstd_dctx = ZSTD_createDCtx(); - if (record->zstd_dctx == NULL) - { - report_invalid_record(record, "out of memory while restoring image at %X/%08X, block %d", - LSN_FORMAT_ARGS(record->ReadRecPtr), - block_id); - return false; - } - -#ifndef FRONTEND - /* The reader may outlive the current memory context. */ - record->zstd_dctx_cb.func = XLogReaderFreeZstdContext; - record->zstd_dctx_cb.arg = record->zstd_dctx; - MemoryContextRegisterResetCallback(GetMemoryChunkContext(record), - &record->zstd_dctx_cb); -#endif + report_invalid_record(record, "out of memory while restoring image at %X/%08X, block %d", + LSN_FORMAT_ARGS(record->ReadRecPtr), + block_id); + return false; } decomp_result = ZSTD_decompressDCtx(record->zstd_dctx, tmp.data, diff --git a/src/backend/access/transam/xlogstats.c b/src/backend/access/transam/xlogstats.c index f0cec2075f8..8b6a87b1f69 100644 --- a/src/backend/access/transam/xlogstats.c +++ b/src/backend/access/transam/xlogstats.c @@ -16,7 +16,8 @@ #include "access/xlogstats.h" /* - * Calculate the size of a record, split into !FPI and FPI parts. + * Calculate the size of a record, split into !FPI and FPI parts, before + * whole-record compression. Per-image compression is reflected in fpi_len. */ void XLogRecGetLen(XLogReaderState *record, uint32 *rec_len, @@ -71,6 +72,7 @@ XLogRecStoreStats(XLogStats *stats, XLogReaderState *record) stats->rmgr_stats[rmid].count++; stats->rmgr_stats[rmid].rec_len += rec_len; stats->rmgr_stats[rmid].fpi_len += fpi_len; + stats->rmgr_stats[rmid].physical_len += XLogRecGetPhysicalLen(record); /* * Update per-record statistics, where the record is identified by a @@ -93,4 +95,5 @@ XLogRecStoreStats(XLogStats *stats, XLogReaderState *record) stats->record_stats[rmid][recid].count++; stats->record_stats[rmid][recid].rec_len += rec_len; stats->record_stats[rmid][recid].fpi_len += fpi_len; + stats->record_stats[rmid][recid].physical_len += XLogRecGetPhysicalLen(record); } diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat index 380679a76c0..53806cee6cc 100644 --- a/src/backend/utils/misc/guc_parameters.dat +++ b/src/backend/utils/misc/guc_parameters.dat @@ -3461,10 +3461,27 @@ }, { name => 'wal_compression', type => 'enum', context => 'PGC_SUSET', group => 'WAL_SETTINGS', - short_desc => 'Compresses full-page writes written in WAL file with specified method.', + short_desc => 'Compresses WAL written with the specified method.', + long_desc => 'Full-page images are compressed individually. With lz4 or zstd, ' + . 'a record larger than wal_compression_threshold is compressed as a whole instead.', variable => 'wal_compression', boot_val => 'WAL_COMPRESSION_NONE', options => 'wal_compression_options', + assign_hook => 'assign_wal_compression', +}, + +{ name => 'wal_compression_threshold', type => 'int', context => 'PGC_SUSET', group => 'WAL_SETTINGS', + short_desc => 'Minimum WAL record length to engage whole-record compression.', + long_desc => 'Records at least this large are compressed as a single unit instead of ' + . 'compressing their full-page images individually. This applies only when ' + . 'wal_compression selects lz4 or zstd. Set above the size of the largest ' + . 'possible WAL record to keep the full-page-image-only behavior.', + flags => 'GUC_UNIT_BYTE', + variable => 'wal_compression_threshold', + boot_val => '512', + min => '32', + max => 'INT_MAX', + assign_hook => 'assign_wal_compression_threshold', }, { name => 'wal_consistency_checking', type => 'string', context => 'PGC_SUSET', group => 'DEVELOPER_OPTIONS', diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 01f98de8a9f..66bd5759298 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -261,9 +261,12 @@ #full_page_writes = on # recover from partial page writes #wal_log_hints = off # also do full page writes of non-critical updates # (change requires restart) -#wal_compression = off # enables compression of full-page writes; +#wal_compression = off # enables compression of full-page writes + # and of whole WAL records; # off, on, zstd, lz4, or pglz (on means the first # of zstd, lz4 and pglz, supported by the build) +#wal_compression_threshold = 512 # min 32, smallest record compressed as a whole; + # whole-record compression needs lz4 or zstd #wal_init_zero = on # zero-fill new WAL files #wal_recycle = on # recycle WAL files #wal_buffers = -1 # min 32kB, -1 sets based on shared_buffers diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c index 6f0574a8764..473c692a2af 100644 --- a/src/bin/pg_waldump/pg_waldump.c +++ b/src/bin/pg_waldump/pg_waldump.c @@ -679,6 +679,9 @@ XLogDumpDisplayRecord(XLogDumpConfig *config, XLogReaderState *record) LSN_FORMAT_ARGS(record->ReadRecPtr), LSN_FORMAT_ARGS(xl_prev)); + if (info & XLR_COMPRESSED) + printf("physical len: %u, ", XLogRecGetPhysicalLen(record)); + id = desc->rm_identify(info); if (id == NULL) printf("desc: UNKNOWN (%x) ", info & ~XLR_INFO_MASK); @@ -703,7 +706,7 @@ XLogDumpStatsRow(const char *name, uint64 n, uint64 total_count, uint64 rec_len, uint64 total_rec_len, uint64 fpi_len, uint64 total_fpi_len, - uint64 tot_len, uint64 total_len) + uint64 tot_len, uint64 total_len, uint64 physical_len) { double n_pct, rec_len_pct, @@ -730,9 +733,9 @@ XLogDumpStatsRow(const char *name, "%20" PRIu64 " (%6.02f) " "%20" PRIu64 " (%6.02f) " "%20" PRIu64 " (%6.02f) " - "%20" PRIu64 " (%6.02f)\n", + "%20" PRIu64 " (%6.02f) %20" PRIu64 "\n", name, n, n_pct, rec_len, rec_len_pct, fpi_len, fpi_len_pct, - tot_len, tot_len_pct); + tot_len, tot_len_pct, physical_len); } @@ -748,6 +751,7 @@ XLogDumpDisplayStats(XLogDumpConfig *config, XLogStats *stats) uint64 total_rec_len = 0; uint64 total_fpi_len = 0; uint64 total_len = 0; + uint64 total_physical_len = 0; double rec_len_pct, fpi_len_pct; @@ -770,21 +774,23 @@ XLogDumpDisplayStats(XLogDumpConfig *config, XLogStats *stats) total_count += stats->rmgr_stats[ri].count; total_rec_len += stats->rmgr_stats[ri].rec_len; total_fpi_len += stats->rmgr_stats[ri].fpi_len; + total_physical_len += stats->rmgr_stats[ri].physical_len; } total_len = total_rec_len + total_fpi_len; printf("WAL statistics between %X/%08X and %X/%08X:\n", LSN_FORMAT_ARGS(stats->startptr), LSN_FORMAT_ARGS(stats->endptr)); + printf("Record, FPI and combined sizes are before whole-record compression.\n"); /* * 27 is strlen("Transaction/COMMIT_PREPARED"), 20 is strlen(2^64), 8 is * strlen("(100.00%)") */ - printf("%-27s %20s %8s %20s %8s %20s %8s %20s %8s\n" - "%-27s %20s %8s %20s %8s %20s %8s %20s %8s\n", - "Type", "N", "(%)", "Record size", "(%)", "FPI size", "(%)", "Combined size", "(%)", - "----", "-", "---", "-----------", "---", "--------", "---", "-------------", "---"); + printf("%-27s %20s %8s %20s %8s %20s %8s %20s %8s %20s\n" + "%-27s %20s %8s %20s %8s %20s %8s %20s %8s %20s\n", + "Type", "N", "(%)", "Record size", "(%)", "FPI size", "(%)", "Combined size", "(%)", "Physical size", + "----", "-", "---", "-----------", "---", "--------", "---", "-------------", "---", "-------------"); for (ri = 0; ri <= RM_MAX_ID; ri++) { @@ -811,7 +817,8 @@ XLogDumpDisplayStats(XLogDumpConfig *config, XLogStats *stats) XLogDumpStatsRow(desc->rm_name, count, total_count, rec_len, total_rec_len, - fpi_len, total_fpi_len, tot_len, total_len); + fpi_len, total_fpi_len, tot_len, total_len, + stats->rmgr_stats[ri].physical_len); } else { @@ -835,7 +842,8 @@ XLogDumpDisplayStats(XLogDumpConfig *config, XLogStats *stats) XLogDumpStatsRow(psprintf("%s/%s", desc->rm_name, id), count, total_count, rec_len, total_rec_len, - fpi_len, total_fpi_len, tot_len, total_len); + fpi_len, total_fpi_len, tot_len, total_len, + stats->record_stats[ri][rj].physical_len); } } } @@ -862,11 +870,11 @@ XLogDumpDisplayStats(XLogDumpConfig *config, XLogStats *stats) "%20" PRIu64 " %-9s" "%20" PRIu64 " %-9s" "%20" PRIu64 " %-9s" - "%20" PRIu64 " %-6s\n", + "%20" PRIu64 " %-8s %20" PRIu64 "\n", "Total", stats->count, "", total_rec_len, psprintf("[%.02f%%]", rec_len_pct), total_fpi_len, psprintf("[%.02f%%]", fpi_len_pct), - total_len, "[100%]"); + total_len, "[100%]", total_physical_len); } /* diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index 2bf5714b51e..984266b4b8a 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -58,6 +58,7 @@ extern PGDLLIMPORT int CommitSiblings; extern PGDLLIMPORT bool track_wal_io_timing; extern PGDLLIMPORT int wal_decode_buffer_size; extern PGDLLIMPORT int data_checksums; +extern PGDLLIMPORT int wal_compression_threshold; extern PGDLLIMPORT int CheckPointSegments; diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h index e48a0db9ce9..f4034c836d0 100644 --- a/src/include/access/xlogreader.h +++ b/src/include/access/xlogreader.h @@ -165,6 +165,7 @@ typedef struct DecodedXLogRecord XLogRecPtr lsn; /* location */ XLogRecPtr next_lsn; /* location of next record */ XLogRecord header; /* header */ + uint32 physical_length; /* size in WAL, before decompression */ ReplOriginId record_origin; TransactionId toplevel_xid; /* XID of top-level transaction */ char *main_data; /* record's main data portion */ @@ -242,7 +243,7 @@ struct XLogReaderState */ #ifdef USE_ZSTD - /* Decompression context reused for zstd-compressed full-page images. */ + /* Decompression context for independent zstd records and page images. */ void *zstd_dctx; #ifndef FRONTEND /* Reset callback for zstd_dctx */ @@ -262,6 +263,10 @@ struct XLogReaderState char *decode_buffer_head; /* data is read from the head */ char *decode_buffer_tail; /* new data is written at the tail */ + /* Buffer for decompressing whole-record compressed WAL records */ + char *decompression_buffer; + uint32 decompression_buffer_size; + /* * Queue of records that have been decoded. This is a linked list that * usually consists of consecutive records in decode_buffer, but may also @@ -427,7 +432,9 @@ extern bool DecodeXLogRecord(XLogReaderState *state, * Macros that provide access to parts of the record most recently returned by * XLogReadRecord() or XLogNextRecord(). */ +/* TotalLen is before whole-record compression; PhysicalLen is stored in WAL. */ #define XLogRecGetTotalLen(decoder) ((decoder)->record->header.xl_tot_len) +#define XLogRecGetPhysicalLen(decoder) ((decoder)->record->physical_length) #define XLogRecGetPrev(decoder) ((decoder)->record->header.xl_prev) #define XLogRecGetInfo(decoder) ((decoder)->record->header.xl_info) #define XLogRecGetRmid(decoder) ((decoder)->record->header.xl_rmid) diff --git a/src/include/access/xlogrecord.h b/src/include/access/xlogrecord.h index e8999d3fe91..be662e6558f 100644 --- a/src/include/access/xlogrecord.h +++ b/src/include/access/xlogrecord.h @@ -48,7 +48,10 @@ typedef struct XLogRecord /* 2 bytes of padding here, initialize to zero */ pg_crc32c xl_crc; /* CRC for this record */ - /* XLogRecordBlockHeaders and XLogRecordDataHeader follow, no padding */ + /* + * XLogRecordBlockHeaders, XLogRecordDataHeader or compression header + * follow, no padding + */ } XLogRecord; @@ -90,6 +93,9 @@ typedef struct XLogRecord */ #define XLR_CHECK_CONSISTENCY 0x02 +/* This bit in xl_info means the record is compressed */ +#define XLR_COMPRESSED 0x04 + /* * Header info for block data appended to an XLOG record. * @@ -143,11 +149,6 @@ typedef struct XLogRecordBlockImageHeader uint16 length; /* number of page image bytes */ uint16 hole_offset; /* number of bytes before "hole" */ uint8 bimg_info; /* flag bits, see below */ - - /* - * If BKPIMAGE_HAS_HOLE and BKPIMAGE_COMPRESSED(), an - * XLogRecordBlockCompressHeader struct follows. - */ } XLogRecordBlockImageHeader; #define SizeOfXLogRecordBlockImageHeader \ @@ -157,13 +158,13 @@ typedef struct XLogRecordBlockImageHeader #define BKPIMAGE_HAS_HOLE 0x01 /* page image has "hole" */ #define BKPIMAGE_APPLY 0x02 /* page image should be restored * during replay */ -/* compression methods supported */ +/* Compression methods supported for FPI (stored in bimg_info) */ #define BKPIMAGE_COMPRESS_PGLZ 0x04 #define BKPIMAGE_COMPRESS_LZ4 0x08 #define BKPIMAGE_COMPRESS_ZSTD 0x10 #define BKPIMAGE_COMPRESSED(info) \ - ((info & (BKPIMAGE_COMPRESS_PGLZ | BKPIMAGE_COMPRESS_LZ4 | \ + (((info) & (BKPIMAGE_COMPRESS_PGLZ | BKPIMAGE_COMPRESS_LZ4 | \ BKPIMAGE_COMPRESS_ZSTD)) != 0) /* @@ -178,6 +179,37 @@ typedef struct XLogRecordBlockCompressHeader #define SizeOfXLogRecordBlockCompressHeader \ sizeof(XLogRecordBlockCompressHeader) +/* compression methods supported for whole-record compression */ +#define XLR_COMPRESS_LZ4 0x08 +#define XLR_COMPRESS_ZSTD 0x10 + +/* + * Whole-record compression is only offered for methods that earn it, so a + * build without any of them carries neither side of the machinery. It must + * still recognize such a record when reading, to report it properly. + */ +#if defined(USE_LZ4) || defined(USE_ZSTD) +#define WAL_WHOLE_RECORD_COMPRESSION 1 +#endif + +/* + * Header prepended to a whole-record compressed WAL record. + * + * The compressed payload follows immediately, at SizeOfXLogCompressedRecord. + * record_header.xl_tot_len covers this header plus that payload, so it is the + * on-disk size; decompressed_length is what the record expands back to, + * including its XLogRecord header. + */ +typedef struct XLogCompressionHeader +{ + XLogRecord record_header; + uint32 decompressed_length; + uint8 method; /* XLR_COMPRESS_* */ + /* 3 bytes of padding here, initialize to zero */ +} XLogCompressionHeader; + +#define SizeOfXLogCompressedRecord sizeof(XLogCompressionHeader) + /* * Maximum size of the header for a block reference. This is used to size a * temporary buffer for constructing the header. diff --git a/src/include/access/xlogstats.h b/src/include/access/xlogstats.h index 6d6842379e5..e29b8498d6e 100644 --- a/src/include/access/xlogstats.h +++ b/src/include/access/xlogstats.h @@ -23,6 +23,7 @@ typedef struct XLogRecStats uint64 count; uint64 rec_len; uint64 fpi_len; + uint64 physical_len; } XLogRecStats; typedef struct XLogStats diff --git a/src/include/utils/guc_hooks.h b/src/include/utils/guc_hooks.h index 06453a18c03..c291da9ea0c 100644 --- a/src/include/utils/guc_hooks.h +++ b/src/include/utils/guc_hooks.h @@ -169,6 +169,8 @@ extern void assign_transaction_timeout(int newval, void *extra); extern bool check_wal_buffers(int *newval, void **extra, GucSource source); extern bool check_wal_consistency_checking(char **newval, void **extra, GucSource source); +extern void assign_wal_compression(int newval, void *extra); +extern void assign_wal_compression_threshold(int newval, void *extra); extern void assign_wal_consistency_checking(const char *newval, void *extra); extern bool check_wal_segment_size(int *newval, void **extra, GucSource source); extern void assign_wal_sync_method(int new_wal_sync_method, void *extra); diff --git a/src/test/perl/PostgreSQL/Test/Cluster.pm b/src/test/perl/PostgreSQL/Test/Cluster.pm index 920d831be9e..1c75f127021 100644 --- a/src/test/perl/PostgreSQL/Test/Cluster.pm +++ b/src/test/perl/PostgreSQL/Test/Cluster.pm @@ -719,9 +719,21 @@ sub init # TEMP_CONFIG. Otherwise, print it before TEMP_CONFIG, thereby permitting # overrides. Settings that merely improve performance or ease debugging # belong before TEMP_CONFIG. + print $conf PostgreSQL::Test::Utils::slurp_file($ENV{TEMP_CONFIG}) if defined $ENV{TEMP_CONFIG}; + # A test that reasons about exact WAL positions cannot tolerate + # whole-record compression: the record it emits would shrink by an amount + # it has no way to predict. Such a test passes no_wal_compression => 1 to + # init(), and we put the record back beyond any possible record size. + # + # This is printed after TEMP_CONFIG on purpose. It is a correctness + # requirement of the test, not a preference, so it must win over whatever + # the buildfarm animal supplies. + print $conf "wal_compression_threshold = " . (1024 * 1024 * 1024) . "\n" + if $params{no_wal_compression}; + if ($params{allows_streaming}) { if ($params{allows_streaming} eq "logical") @@ -3193,10 +3205,16 @@ sub emit_wal { my ($self, $size) = @_; + # Disable WAL compression for this emission. Tests call + # emit_wal() to place an exact number of bytes in WAL (e.g. to reach a + # specific page offset); whole-record compression would shrink the record + # and break those position-sensitive calculations. return int( $self->safe_psql( 'postgres', - "SELECT pg_logical_emit_message(true, '', repeat('a', $size)) - '0/0'" + "SET wal_compression = off; + SELECT pg_logical_emit_message(true, '', repeat('a', $size)) - '0/0'; + RESET wal_compression" )); } diff --git a/src/test/recovery/Makefile b/src/test/recovery/Makefile index 9c4102b6b2c..ea0001ec9b6 100644 --- a/src/test/recovery/Makefile +++ b/src/test/recovery/Makefile @@ -10,6 +10,7 @@ #------------------------------------------------------------------------- EXTRA_INSTALL=contrib/pg_prewarm \ + contrib/pg_walinspect \ contrib/pg_stat_statements \ contrib/test_decoding \ src/test/modules/injection_points \ diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build index b6ac9dd07a7..0336ce68e79 100644 --- a/src/test/recovery/meson.build +++ b/src/test/recovery/meson.build @@ -69,6 +69,7 @@ tests += { 't/058_shutdown_crash_restart.pl', 't/059_remote_apply_status_interval.pl', 't/060_multixact_crashed_updater.pl', + 't/061_wal_compression.pl', ], }, } diff --git a/src/test/recovery/t/026_overwrite_contrecord.pl b/src/test/recovery/t/026_overwrite_contrecord.pl index 8db06739902..cc6ab17cd71 100644 --- a/src/test/recovery/t/026_overwrite_contrecord.pl +++ b/src/test/recovery/t/026_overwrite_contrecord.pl @@ -16,7 +16,9 @@ use Test::More; # file and the standby promotes successfully. my $node = PostgreSQL::Test::Cluster->new('primary'); -$node->init(allows_streaming => 1); +# This test emits a message of a chosen size and requires the resulting record +# to cross a segment boundary, so the record must reach WAL at that size. +$node->init(allows_streaming => 1, no_wal_compression => 1); # We need these settings for stability of WAL behavior. checkpoint_timeout # avoids noise from checkpoint activity, and autovacuum is disabled to avoid # any WAL activity generated by it. diff --git a/src/test/recovery/t/061_wal_compression.pl b/src/test/recovery/t/061_wal_compression.pl new file mode 100644 index 00000000000..c7f7cf5302d --- /dev/null +++ b/src/test/recovery/t/061_wal_compression.pl @@ -0,0 +1,204 @@ +# Copyright (c) 2025-2026, PostgreSQL Global Development Group +# +# Test whole-record WAL compression via wal_compression and +# wal_compression_threshold. Exercises compression during replication +# (walsender path) and decompression during crash recovery (startup path). +# + +use strict; +use warnings FATAL => 'all'; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +# Determine which compression methods are compiled in, preferring the +# faster hardware-accelerated codecs first for quicker failure feedback. +my @methods = (); +push @methods, 'zstd' if check_pg_config('#define HAVE_LIBZSTD 1'); +push @methods, 'lz4' if check_pg_config('#define HAVE_LIBLZ4 1'); +push @methods, 'pglz'; + +# Test whole-record WAL compression with a specific method. +# +# Creates a primary with the method enabled and a low threshold so that +# every non-trivial WAL record is compressed. A streaming standby is +# used to verify that compressed WAL is transmitted and decoded correctly. +# Then the primary is stopped immediately (simulating a crash) and +# restarted to verify that startup recovery can decompress WAL records. +# Emit one 256kB compressible logical message at the given threshold and +# report how many bytes of WAL it consumed. +sub wal_used +{ + my ($node, $threshold) = @_; + + my $lsns = $node->safe_psql( + 'postgres', qq{ + SET wal_compression_threshold = $threshold; + SELECT pg_current_wal_insert_lsn(); + SELECT pg_logical_emit_message(true, 'wal compression', repeat('abcd', 65536)); + SELECT pg_current_wal_insert_lsn(); + }); + + # Three result rows: the LSN before, the message's own LSN, the LSN after. + my ($before, $after) = (split /\n/, $lsns)[ 0, 2 ]; + return $node->safe_psql('postgres', + "SELECT '$after'::pg_lsn - '$before'::pg_lsn"); +} + +sub test_wal_compression +{ + my ($method) = @_; + + note "testing wal_compression = $method"; + + my $primary = PostgreSQL::Test::Cluster->new("primary_$method"); + $primary->init(allows_streaming => 1); + + # Start with FPI-only compression to check its initial workspace. + $primary->append_conf( + 'postgresql.conf', + "wal_compression = '$method'\n" + . "wal_compression_threshold = 1GB\n"); + $primary->start; + + # FPI-only sessions should start with space for five images, not the + # maximum-sized input and output workspaces for whole-record compression. + # Allow for allocator overhead rather than depending on its exact layout. + is($primary->safe_psql('postgres', q{ + SELECT sum(total_bytes) < 20 * current_setting('block_size')::int + FROM pg_backend_memory_contexts + WHERE name = 'WAL record construction'; + }), 't', "small initial FPI-only workspace ($method)"); + + $primary->safe_psql('postgres', 'CREATE EXTENSION pg_walinspect'); + my $probe_start = $primary->safe_psql('postgres', + 'SELECT pg_current_wal_insert_lsn()'); + $primary->safe_psql('postgres', q{ + SET wal_compression_threshold = 32; + SELECT pg_logical_emit_message(true, '', + decode(substr(repeat(md5(g::text), 8), 1, 2 * g), 'hex')) + FROM generate_series(1, 128) AS g; + }); + my $probe_end = $primary->safe_psql('postgres', + 'SELECT pg_current_wal_flush_lsn()'); + is($primary->safe_psql('postgres', qq{ + SELECT count(*) FROM pg_get_wal_records_info('$probe_start', '$probe_end') + WHERE resource_manager = 'LogicalMessage' + AND (start_lsn - '0/0'::pg_lsn) % + current_setting('wal_block_size')::int + record_length < + current_setting('wal_block_size')::int + AND end_lsn - start_lsn > ((record_length + 7) / 8) * 8; + }), '0', "independent compression does not enlarge records ($method)"); + + # Whole-record storage must also be prepared when the codec changes after + # lowering the threshold, and when SET LOCAL rolls back to another mode. + $primary->safe_psql('postgres', qq{ + SET wal_compression = off; + SET wal_compression_threshold = 32; + SET wal_compression = '$method'; + BEGIN; + SET LOCAL wal_compression_threshold = '1GB'; + SELECT pg_logical_emit_message(true, '', repeat('x', 1000)); + ROLLBACK; + SELECT pg_logical_emit_message(true, '', repeat('x', 1000)); + }); + + # Exercise growth past the initial five image buffers in FPI-only mode. + # VACUUM FULL writes multi-page FPI records through log_newpages(). + $primary->safe_psql('postgres', q{ + CREATE TABLE many_images AS + SELECT g, repeat('x', 1000) AS d FROM generate_series(1, 1000) AS g; + VACUUM FULL many_images; + }); + $primary->append_conf('postgresql.conf', 'wal_compression_threshold = 32'); + $primary->restart; + + my $backup_name = "backup_$method"; + $primary->backup($backup_name); + + my $standby = PostgreSQL::Test::Cluster->new("standby_$method"); + $standby->init_from_backup($primary, $backup_name, has_streaming => 1); + $standby->start; + + # Prove the feature actually engages, rather than silently doing nothing. + # A large, highly compressible logical message is one record with no + # full-page images, so the WAL it occupies is decided purely by whole-record + # compression. Measure it against the same message with the threshold + # raised out of reach. + my $compressed_bytes = wal_used($primary, 32); + my $plain_bytes = wal_used($primary, 1024 * 1024 * 1024); + + if ($method eq 'pglz') + { + # pglz is excluded from whole-record compression on purpose, so the + # threshold must make no difference at all. + is($compressed_bytes, $plain_bytes, + "whole-record compression does not engage for pglz"); + } + else + { + cmp_ok($compressed_bytes, '<', $plain_bytes / 2, + "whole-record compression shrinks the record ($method): " + . "$compressed_bytes vs $plain_bytes bytes"); + } + + my $start_lsn = + $primary->safe_psql('postgres', 'SELECT pg_current_wal_insert_lsn()'); + + # Generate WAL with records that exceed the compression threshold. + # Each row's WAL record includes 200-byte payload well above 32 bytes. + $primary->safe_psql( + 'postgres', + "CREATE TABLE t AS + SELECT g, repeat('x', 200) AS d + FROM generate_series(1, 100) AS g"); + + $primary->wait_for_replay_catchup($standby); + + is( $standby->safe_psql('postgres', 'SELECT count(*) FROM t'), + '100', + "compressed WAL replicated via streaming ($method)"); + + # Insert another batch that will be recovered after the simulated crash. + $primary->safe_psql( + 'postgres', + "INSERT INTO t + SELECT g, repeat('y', 200) + FROM generate_series(101, 200) AS g"); + + # pg_waldump has to walk these records too. That is the frontend decoding + # path, which reaches XLogDecompressRecordIfNeeded() without any of the + # server's state, so it is worth covering separately from recovery. + my $end_lsn = + $primary->safe_psql('postgres', 'SELECT pg_current_wal_flush_lsn()'); + + $primary->command_ok( + [ + 'pg_waldump', + '--quiet', + '--path' => $primary->data_dir . '/pg_wal', + '--start' => $start_lsn, + '--end' => $end_lsn + ], + "pg_waldump decodes compressed WAL ($method)"); + + # Stop without a clean shutdown. On restart PostgreSQL will replay WAL + # from the last checkpoint, exercising XLogDecompressRecordIfNeeded for + # every compressed record generated since that checkpoint. + $primary->stop('immediate'); + $primary->start; + + is( $primary->safe_psql('postgres', 'SELECT count(*) FROM t'), + '200', + "crash recovery replays compressed WAL ($method)"); + + $primary->stop; + $standby->stop; +} + +foreach my $method (@methods) +{ + test_wal_compression($method); +} + +done_testing(); diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index a7a8dbdd23f..f3db609f1f3 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -3486,6 +3486,7 @@ X509_STORE X509_STORE_CTX X86FeatureId XLTW_Oper +XLogCompressionHeader XLogCtlData XLogCtlInsert XLogDumpConfig base-commit: aa7b9cf132b013a4061e61c4393164f31d4f6930 -- That's all, folks. May the source be with you.