From a00b2193edb4b62cdd5bfae80c87327884f75143 Mon Sep 17 00:00:00 2001 From: Andrey Borodin Date: Wed, 8 Jan 2025 16:39:20 +0500 Subject: [PATCH v7 2/3] Add whole-record WAL compression alongside FPI compression When a record is larger than wal_compression_threshold, compress it as a single unit instead of compressing each full-page image separately. This wins whenever the images in one record share content, which is typical of B-tree index builds: CREATE INDEX over 10M random floats drops from 160MB of WAL to 132MB with lz4, and from 125MB to 97MB with zstd. Only lz4 and zstd take part. pglz refuses input it cannot shrink by 25%, which a record made of full-page images rarely clears, so it saved nothing while costing an extra compression attempt. It still compresses full-page images as before. Assembly skips per-FPI compression while whole-record compression is attempted, and reassembles the record if that does not pay off. Setting wal_compression_threshold above the largest possible record restores the previous behaviour. The per-block compressed_page arrays make way for two buffers of the same maximum size, one staging the record and one taking the compressor output. XLR_COMPRESSED in xl_info marks a compressed record, which begins with an XLogCompressionHeader giving the method and the decompressed length. Author: Andrey Borodin Discussion: https://postgr.es/m/4DC38068-976E-4A84-8EE6-4EFACBBD927A@yandex-team.ru --- src/backend/access/transam/xlog.c | 19 +- src/backend/access/transam/xloginsert.c | 330 ++++++++++++++++-- src/backend/access/transam/xlogreader.c | 255 ++++++++++++-- src/backend/utils/misc/guc_parameters.dat | 14 + src/backend/utils/misc/postgresql.conf.sample | 5 +- src/include/access/xlog.h | 1 + src/include/access/xlogreader.h | 4 + src/include/access/xlogrecord.h | 48 ++- src/include/utils/guc_hooks.h | 1 + src/test/perl/PostgreSQL/Test/Cluster.pm | 20 +- src/test/recovery/Makefile | 10 + .../recovery/t/026_overwrite_contrecord.pl | 4 +- .../recovery/t/046_checkpoint_logical_slot.pl | 5 +- src/test/recovery/t/052_wal_compression.pl | 152 ++++++++ src/test/recovery/wal_compression.conf | 6 + src/tools/pgindent/typedefs.list | 1 + 16 files changed, 804 insertions(+), 71 deletions(-) create mode 100644 src/test/recovery/t/052_wal_compression.pl create mode 100644 src/test/recovery/wal_compression.conf diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f8b939853e9..eb1d1fddb77 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -142,6 +142,7 @@ int wal_retrieve_retry_interval = 5000; int max_slot_wal_keep_size_mb = -1; int wal_decode_buffer_size = 512 * 1024; bool track_wal_io_timing = false; +int wal_compression_threshold = 512; #ifdef WAL_DEBUG bool XLOG_DEBUG = false; @@ -751,6 +752,22 @@ static void WALInsertLockUpdateInsertingAt(XLogRecPtr insertingAt); static void XLogChecksums(uint32 new_type); +#ifdef WAL_DEBUG +/* Read length of a record, accounting for possible compression */ +static uint32 +XLogGetRecordTotalLen(XLogRecord *record) +{ + if (record->xl_info & XLR_COMPRESSED) + { + XLogCompressionHeader *c = (XLogCompressionHeader *) record; + + Assert(c->decompressed_length > 0); + return c->decompressed_length; + } + return record->xl_tot_len; +} +#endif + /* * Insert an XLOG record represented by an already-constructed chain of data * chunks. This is a low-level routine; to construct the WAL record header @@ -1068,7 +1085,7 @@ XLogInsertRecord(XLogRecData *rdata, /* We also need temporary space to decode the record. */ record = (XLogRecord *) recordBuf.data; decoded = (DecodedXLogRecord *) - palloc(DecodeXLogRecordRequiredSpace(record->xl_tot_len)); + palloc(DecodeXLogRecordRequiredSpace(XLogGetRecordTotalLen(record))); if (!debug_reader) debug_reader = XLogReaderAllocate(wal_segment_size, NULL, diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c index d752eb8a762..a5a3a1422cb 100644 --- a/src/backend/access/transam/xloginsert.c +++ b/src/backend/access/transam/xloginsert.c @@ -39,6 +39,7 @@ #include "replication/origin.h" #include "storage/bufmgr.h" #include "storage/proc.h" +#include "utils/guc_hooks.h" #include "utils/memutils.h" #include "utils/pgstat_internal.h" #include "utils/rel.h" @@ -85,8 +86,8 @@ typedef struct XLogRecData bkp_rdatas[2]; /* temporary rdatas used to hold references to * backup block data in XLogRecordAssemble() */ - /* buffer to store a compressed version of backup block image */ - char compressed_page[COMPRESS_BUFSIZE]; + /* pointer into compression_buf for compressed page image */ + char *compressed_page; } registered_buffer; static registered_buffer *registered_buffers; @@ -128,6 +129,15 @@ static char *hdr_scratch = NULL; * from here. */ static ZSTD_CCtx *zstd_cctx = NULL; + +/* Get this backend's zstd context, creating it on first use. NULL on OOM. */ +static ZSTD_CCtx * +GetZstdCCtx(void) +{ + if (zstd_cctx == NULL) + zstd_cctx = ZSTD_createCCtx(); + return zstd_cctx; +} #endif #define SizeOfXlogOrigin (sizeof(ReplOriginId) + sizeof(char)) @@ -139,6 +149,28 @@ static ZSTD_CCtx *zstd_cctx = NULL; SizeOfXLogRecordDataHeaderLong + SizeOfXlogOrigin + \ SizeOfXLogTransactionId) +/* + * Size of the compression buffers: enough for a compressed image of every + * block a record may reference, plus the record headers. That is the largest + * record we can be asked to build, so a record that does not fit is not + * compressed as a whole. + */ +#define WAL_COMPRESSION_BUFSIZE \ + ((XLR_MAX_BLOCK_ID + 1) * COMPRESS_BUFSIZE + HEADER_SCRATCH_SIZE) + +/* + * Compression buffers, allocated together when wal_compression is first + * enabled. compression_buf serves two uses that never overlap: per-FPI + * compression packs block images into it, tracked by compression_buf_offset, + * while whole-record compression flattens the rdt chain into it and compresses + * into compressed_data. + */ +static char *compression_buf = NULL; +static int compression_buf_offset; /* fill level for FPI packing */ +#ifdef WAL_WHOLE_RECORD_COMPRESSION +static char *compressed_data = NULL; +#endif + /* * An array of XLogRecData structs, to hold registered data. */ @@ -151,11 +183,13 @@ static bool begininsert_called = false; /* Memory context to hold the registered buffer and data references. */ static MemoryContext xloginsert_cxt; +static void AllocCompressionBuffers(void); static XLogRecData *XLogRecordAssemble(RmgrId rmid, uint8 info, XLogRecPtr RedoRecPtr, bool doPageWrites, XLogRecPtr *fpw_lsn, int *num_fpi, uint64 *fpi_bytes, - bool *topxid_included); + bool *topxid_included, uint64 *rec_size, + bool skip_fpi_compression); static bool XLogCompressBackupBlock(const PageData *page, uint16 hole_offset, uint16 hole_length, void *dest, uint16 *dlen); @@ -481,6 +515,142 @@ XLogSetRecordFlags(uint8 flags) curinsert_flags |= flags; } +/* Compress the assembled record; NULL if that did not pay off */ +static XLogRecData * +XLogCompressRdt(XLogRecData *rdt) +{ +#ifndef WAL_WHOLE_RECORD_COMPRESSION + /* Without a supported method the caller never gets here */ + Assert(false); + return NULL; +#else + static XLogRecData compressed_rdt_hdr; + XLogCompressionHeader *compressed_header; + XLogRecord *src_header; + uint32 flat_len = 0; + uint32 orig_len; + int32 compr_len = -1; + int32 dest_size; + + Assert(wal_compression != WAL_COMPRESSION_NONE); + Assert(compression_buf != NULL); + + /* + * Flatten the rdt chain into compression_buf. The caller skipped per-FPI + * compression, so the buffer is free, and checked that the record fits. + */ + for (const XLogRecData *r = rdt; r != NULL; r = r->next) + { + memcpy(compression_buf + flat_len, r->data, r->len); + flat_len += r->len; + } + Assert(flat_len <= WAL_COMPRESSION_BUFSIZE); + + src_header = (XLogRecord *) compression_buf; + compressed_header = (XLogCompressionHeader *) compressed_data; + + /* Zero it first: the padding in the header reaches disk */ + memset(compressed_header, 0, SizeOfXLogCompressedRecord); + + compressed_header->record_header = *src_header; + compressed_header->decompressed_length = flat_len; + + orig_len = src_header->xl_tot_len - SizeOfXLogRecord; + + /* + * compressed_data is not sized to the worst-case expansion bound, so hand + * the compressor what is really left and let it fail if that is too + * little. A record that does not fit was not worth compressing anyway. + */ + dest_size = WAL_COMPRESSION_BUFSIZE - SizeOfXLogCompressedRecord; + + switch ((WalCompression) wal_compression) + { + case WAL_COMPRESSION_NONE: + case WAL_COMPRESSION_PGLZ: + /* caller does not use whole-record compression for these */ + Assert(false); + return NULL; + + case WAL_COMPRESSION_LZ4: +#ifdef USE_LZ4 + compressed_header->method = XLR_COMPRESS_LZ4; + compr_len = LZ4_compress_default((char *) &src_header[1], + (char *) &compressed_header[1], + orig_len, dest_size); + if (compr_len <= 0) + return NULL; +#else + elog(ERROR, "LZ4 is not supported by this build"); +#endif + break; + + case WAL_COMPRESSION_ZSTD: +#ifdef USE_ZSTD + { + size_t zstd_len; + + ZSTD_CCtx *cctx = GetZstdCCtx(); + + if (cctx == NULL) + return NULL; + + compressed_header->method = XLR_COMPRESS_ZSTD; + zstd_len = ZSTD_compressCCtx(cctx, + (char *) &compressed_header[1], + dest_size, + (char *) &src_header[1], orig_len, + ZSTD_CLEVEL_DEFAULT); + if (ZSTD_isError(zstd_len)) + return NULL; + compr_len = (int32) zstd_len; + } +#else + elog(ERROR, "zstd is not supported by this build"); +#endif + break; + + /* no default case, so that compiler will warn */ + } + + Assert(compr_len > 0 && compr_len <= dest_size); + + compressed_header->record_header.xl_tot_len = + SizeOfXLogCompressedRecord + compr_len; + + compressed_header->record_header.xl_info |= XLR_COMPRESSED; + + compressed_rdt_hdr.data = compressed_data; + compressed_rdt_hdr.len = compressed_header->record_header.xl_tot_len; + compressed_rdt_hdr.next = NULL; + + return &compressed_rdt_hdr; +#endif +} + +/* Checksum assembled record (which may be compressed). */ +static void +XLogChecksumRecord(XLogRecData *rdt) +{ + pg_crc32c rdata_crc; + XLogRecord *rechdr = (XLogRecord *) rdt->data; + + /* + * Calculate CRC of the data + * + * Note that the record header isn't added into the CRC initially since we + * don't know the prev-link yet. Thus, the CRC will represent the CRC of + * the whole record in the order: rdata, then backup blocks, then record + * header. + */ + INIT_CRC32C(rdata_crc); + COMP_CRC32C(rdata_crc, ((char *) rdt->data) + SizeOfXLogRecord, + rdt->len - SizeOfXLogRecord); + for (rdt = rdt->next; rdt != NULL; rdt = rdt->next) + COMP_CRC32C(rdata_crc, rdt->data, rdt->len); + rechdr->xl_crc = rdata_crc; +} + /* * Insert an XLOG record having the specified RMID and info bytes, with the * body of the record being the data and buffer references registered earlier @@ -497,6 +667,17 @@ XLogInsert(RmgrId rmid, uint8 info) { XLogRecPtr EndPos; + /* + * Only lz4 and zstd compress whole records; pglz rarely beats its own + * minimum compression rate on a record made of full-page images. A + * threshold at or above the buffer size disables this entirely, which is + * how a user asks for full-page-image-only compression. + */ + bool try_whole_record = ((wal_compression == WAL_COMPRESSION_LZ4 || + wal_compression == WAL_COMPRESSION_ZSTD) && + wal_compression_threshold < + WAL_COMPRESSION_BUFSIZE); + /* XLogBeginInsert() must have been called. */ if (!begininsert_called) elog(ERROR, "XLogBeginInsert was not called"); @@ -532,6 +713,10 @@ XLogInsert(RmgrId rmid, uint8 info) XLogRecData *rdt; int num_fpi = 0; uint64 fpi_bytes = 0; + uint64 rec_size; + + /* Do not accumulate FPI data across retries of this loop */ + compression_buf_offset = 0; /* * Get values needed to decide whether to do full-page writes. Since @@ -542,7 +727,41 @@ XLogInsert(RmgrId rmid, uint8 info) rdt = XLogRecordAssemble(rmid, info, RedoRecPtr, doPageWrites, &fpw_lsn, &num_fpi, &fpi_bytes, - &topxid_included); + &topxid_included, &rec_size, + try_whole_record); + + /* + * Assembly skipped per-FPI compression on our behalf, so if the + * record is not compressed as a whole after all, reassemble it to get + * the per-FPI compression back. + */ + if (try_whole_record) + { + bool whole_record_compressed = false; + + if (rec_size > wal_compression_threshold && + rec_size <= WAL_COMPRESSION_BUFSIZE) + { + XLogRecData *rdt_compressed = XLogCompressRdt(rdt); + + if (rdt_compressed != NULL) + { + rdt = rdt_compressed; + whole_record_compressed = true; + } + } + + if (!whole_record_compressed && num_fpi > 0) + { + compression_buf_offset = 0; + rdt = XLogRecordAssemble(rmid, info, RedoRecPtr, doPageWrites, + &fpw_lsn, &num_fpi, &fpi_bytes, + &topxid_included, &rec_size, + false); + } + } + + XLogChecksumRecord(rdt); EndPos = XLogInsertRecord(rdt, fpw_lsn, curinsert_flags, num_fpi, fpi_bytes, topxid_included); @@ -616,6 +835,44 @@ XLogGetFakeLSN(Relation rel) } } +/* + * Allocate the compression buffers, once per backend. + * + * Callers must be outside a critical section, which is why this happens at + * backend start or in the wal_compression assign hook, never in XLogInsert(). + * The buffers are kept when compression is switched back off, so that turning + * it on again cannot need an allocation. + */ +static void +AllocCompressionBuffers(void) +{ + Assert(CritSectionCount == 0); + Assert(xloginsert_cxt != NULL); + + if (compression_buf != NULL) + return; /* already done */ + + compression_buf = MemoryContextAlloc(xloginsert_cxt, + WAL_COMPRESSION_BUFSIZE); +#ifdef WAL_WHOLE_RECORD_COMPRESSION + compressed_data = MemoryContextAlloc(xloginsert_cxt, + WAL_COMPRESSION_BUFSIZE); +#endif +} + +/* + * GUC assign hook for wal_compression. + * + * Runs before wal_compression itself is updated, hence the test on newval. If + * xloginsert_cxt does not exist yet, InitXLogInsert() will do the allocation. + */ +void +assign_wal_compression(int newval, void *extra) +{ + if (newval != WAL_COMPRESSION_NONE && xloginsert_cxt != NULL) + AllocCompressionBuffers(); +} + /* * Assemble a WAL record from the registered data and buffers into an * XLogRecData chain, ready for insertion with XLogInsertRecord(). @@ -635,12 +892,11 @@ static XLogRecData * XLogRecordAssemble(RmgrId rmid, uint8 info, XLogRecPtr RedoRecPtr, bool doPageWrites, XLogRecPtr *fpw_lsn, int *num_fpi, uint64 *fpi_bytes, - bool *topxid_included) + bool *topxid_included, uint64 *rec_size, + bool skip_fpi_compression) { - XLogRecData *rdt; uint64 total_len = 0; int block_id; - pg_crc32c rdata_crc; registered_buffer *prev_regbuf = NULL; XLogRecData *rdt_datas_last; XLogRecord *rechdr; @@ -668,6 +924,9 @@ XLogRecordAssemble(RmgrId rmid, uint8 info, if (wal_consistency_checking[rmid]) info |= XLR_CHECK_CONSISTENCY; + /* We are often in a critical section here, so we cannot allocate */ + Assert(wal_compression == WAL_COMPRESSION_NONE || compression_buf != NULL); + /* * Make an rdata chain containing all the data portions of all block * references. This includes the data for full-page images. Also append @@ -771,15 +1030,26 @@ XLogRecordAssemble(RmgrId rmid, uint8 info, } /* - * Try to compress a block image if wal_compression is enabled + * Try to compress a block image if wal_compression is enabled and + * the caller has not reserved the buffer for whole-record + * compression. Each image gets its own slice of the buffer, + * which is large enough for all of them. */ - if (wal_compression != WAL_COMPRESSION_NONE) + if (!skip_fpi_compression && + wal_compression != WAL_COMPRESSION_NONE) { + Assert(compression_buf_offset + COMPRESS_BUFSIZE <= + WAL_COMPRESSION_BUFSIZE); + regbuf->compressed_page = compression_buf + compression_buf_offset; + is_compressed = XLogCompressBackupBlock(page, bimg.hole_offset, cbimg.hole_length, regbuf->compressed_page, &compressed_len); + + if (is_compressed) + compression_buf_offset += compressed_len; } /* @@ -983,19 +1253,6 @@ XLogRecordAssemble(RmgrId rmid, uint8 info, hdr_rdt.len = (scratch - hdr_scratch); total_len += hdr_rdt.len; - /* - * Calculate CRC of the data - * - * Note that the record header isn't added into the CRC initially since we - * don't know the prev-link yet. Thus, the CRC will represent the CRC of - * the whole record in the order: rdata, then backup blocks, then record - * header. - */ - INIT_CRC32C(rdata_crc); - COMP_CRC32C(rdata_crc, hdr_scratch + SizeOfXLogRecord, hdr_rdt.len - SizeOfXLogRecord); - for (rdt = hdr_rdt.next; rdt != NULL; rdt = rdt->next) - COMP_CRC32C(rdata_crc, rdt->data, rdt->len); - /* * Ensure that the XLogRecord is not too large. * @@ -1019,7 +1276,9 @@ XLogRecordAssemble(RmgrId rmid, uint8 info, rechdr->xl_info = info; rechdr->xl_rmid = rmid; rechdr->xl_prev = InvalidXLogRecPtr; - rechdr->xl_crc = rdata_crc; + rechdr->xl_crc = 0; + + *rec_size = rechdr->xl_tot_len; return &hdr_rdt; } @@ -1078,17 +1337,19 @@ XLogCompressBackupBlock(const PageData *page, uint16 hole_offset, uint16 hole_le case WAL_COMPRESSION_ZSTD: #ifdef USE_ZSTD - if (zstd_cctx == NULL) - zstd_cctx = ZSTD_createCCtx(); - - if (zstd_cctx == NULL) - len = -1; /* out of memory; store the image as is */ - else { - len = ZSTD_compressCCtx(zstd_cctx, dest, COMPRESS_BUFSIZE, - source, orig_len, ZSTD_CLEVEL_DEFAULT); - if (ZSTD_isError(len)) - len = -1; /* failure */ + ZSTD_CCtx *cctx = GetZstdCCtx(); + + if (cctx == NULL) + len = -1; /* out of memory; store the image as is */ + else + { + len = ZSTD_compressCCtx(cctx, dest, COMPRESS_BUFSIZE, + source, orig_len, + ZSTD_CLEVEL_DEFAULT); + if (ZSTD_isError(len)) + len = -1; /* failure */ + } } #else elog(ERROR, "zstd is not supported by this build"); @@ -1460,4 +1721,7 @@ InitXLogInsert(void) if (hdr_scratch == NULL) hdr_scratch = MemoryContextAllocZero(xloginsert_cxt, HEADER_SCRATCH_SIZE); + + if (wal_compression != WAL_COMPRESSION_NONE) + AllocCompressionBuffers(); } diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c index 946907a2507..23fe90c7f11 100644 --- a/src/backend/access/transam/xlogreader.c +++ b/src/backend/access/transam/xlogreader.c @@ -32,6 +32,7 @@ #include "catalog/pg_control.h" #include "common/pg_lzcompress.h" #include "replication/origin.h" +#include "utils/memutils.h" #ifndef FRONTEND #include "pgstat.h" @@ -55,6 +56,9 @@ static bool ValidXLogRecord(XLogReaderState *state, XLogRecord *record, static void ResetDecoder(XLogReaderState *state); static void WALOpenSegmentInit(WALOpenSegment *seg, WALSegmentContext *segcxt, int segsize, const char *waldir); +static XLogRecord *XLogDecompressRecordIfNeeded(XLogReaderState *state, + XLogRecord *record, + XLogRecPtr recptr); /* size of the buffer allocated for error message. */ #define MAX_ERRORMSG_LEN 1000 @@ -171,6 +175,8 @@ XLogReaderFree(XLogReaderState *state) pfree(state->errormsg_buf); if (state->readRecordBuf) pfree(state->readRecordBuf); + if (state->decompression_buffer) + pfree(state->decompression_buffer); pfree(state->readBuf); pfree(state); } @@ -534,10 +540,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; @@ -588,6 +596,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; @@ -645,7 +661,70 @@ restart: * whole header. */ record = (XLogRecord *) (state->readBuf + RecPtr % XLOG_BLCKSZ); - total_len = record->xl_tot_len; + total_len_physical = record->xl_tot_len; + + /* + * Size the record will occupy once decoded, which for a compressed record + * is its decompressed length rather than xl_tot_len. + * + * Only xl_tot_len is guaranteed to be on this page; xl_info may live on + * the next one when the record starts near the end of a page. If we + * cannot tell yet, have_decomp_len stays false and we allocate after + * assembly instead. + */ + total_len_decomp = 0; + have_decomp_len = false; + + if (targetRecOff <= XLOG_BLCKSZ - SizeOfXLogRecord) + { + /* Full header is on this page; safe to read xl_info. */ + if (record->xl_info & XLR_COMPRESSED) + { + /* + * Skip if the compression header spans pages, or if the record is + * too short to be a valid compressed one. + */ + if (targetRecOff <= XLOG_BLCKSZ - SizeOfXLogCompressedRecord && + total_len_physical >= SizeOfXLogCompressedRecord) + { + XLogCompressionHeader *c = (XLogCompressionHeader *) record; + uint32 dlen = c->decompressed_length; + bool valid_method; + + valid_method = false +#ifdef USE_LZ4 + || (c->method == XLR_COMPRESS_LZ4) +#endif +#ifdef USE_ZSTD + || (c->method == XLR_COMPRESS_ZSTD) +#endif + ; + + /* Sanity-check before using for allocation */ + if (dlen > SizeOfXLogRecord && dlen <= MaxAllocSize && + valid_method) + { + total_len_decomp = dlen; + have_decomp_len = true; + } + } + } + else + { + total_len_decomp = record->xl_tot_len; + have_decomp_len = true; + } + } + else + { + /* + * Header spans pages, so we cannot read xl_info and do not know + * whether this is a compressed record, let alone how big it decodes + * to. xl_tot_len would be the compressed size, and reserving that + * much would leave the decoder writing past its slot, so leave the + * allocation to the pass after assembly. + */ + } /* * If the whole record header is on this page, validate it immediately. @@ -661,16 +740,18 @@ restart: randAccess)) goto err; gotheader = true; + if (record->xl_info & XLR_COMPRESSED) + gotheader = targetRecOff <= XLOG_BLCKSZ - SizeOfXLogCompressedRecord; } else { /* There may be no next page if it's too small. */ - if (total_len < SizeOfXLogRecord) + if (total_len_physical < SizeOfXLogRecord) { report_invalid_record(state, "invalid record length at %X/%08X: expected at least %u, got %u", LSN_FORMAT_ARGS(RecPtr), - (uint32) SizeOfXLogRecord, total_len); + (uint32) SizeOfXLogRecord, total_len_physical); goto err; } /* We'll validate the header once we have the next page. */ @@ -682,21 +763,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; @@ -729,7 +813,9 @@ restart: * can handle the case where the previous record ended as being a * partial one. */ - readOff = ReadPageInternal(state, targetPagePtr, SizeOfXLogShortPHD); + readOff = ReadPageInternal(state, targetPagePtr, + Min(total_len_physical - gotlen + SizeOfXLogShortPHD, + XLOG_BLCKSZ)); if (readOff == XLREAD_WOULDBLOCK) return XLREAD_WOULDBLOCK; else if (readOff < 0) @@ -768,19 +854,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; @@ -825,7 +911,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]; @@ -836,11 +922,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; @@ -855,8 +941,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) @@ -866,7 +953,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; } @@ -889,8 +976,20 @@ restart: if (decoded == NULL) { Assert(!nonblocking); + + /* total_len_decomp may not yet reflect the actual decompressed size */ + if (record->xl_info & XLR_COMPRESSED) + { + XLogCompressionHeader *c = (XLogCompressionHeader *) record; + + Assert(c->decompressed_length > 0); + Assert(c->decompressed_length < MaxAllocSize); + total_len_decomp = c->decompressed_length; + } + else + total_len_decomp = record->xl_tot_len; decoded = XLogReadRecordAlloc(state, - total_len, + total_len_decomp, true /* allow_oversized */ ); /* allocation should always happen under allow_oversized */ Assert(decoded != NULL); @@ -1687,6 +1786,108 @@ DecodeXLogRecordRequiredSpace(size_t xl_tot_len) return size; } +static XLogRecord * +XLogDecompressRecordIfNeeded(XLogReaderState *state, + XLogRecord *record, + XLogRecPtr recptr) +{ + if (record->xl_info & XLR_COMPRESSED) + { +#ifndef WAL_WHOLE_RECORD_COMPRESSION + report_invalid_record(state, + "could not decompress record at %X/%08X compressed with method %u not supported by build", + LSN_FORMAT_ARGS(recptr), + ((XLogCompressionHeader *) record)->method); + return NULL; +#else + XLogCompressionHeader *src = (XLogCompressionHeader *) record; + + /* decompressed_length covers the XLogRecord header + body */ + uint32 body_len = src->decompressed_length - SizeOfXLogRecord; + uint32 srclen = src->record_header.xl_tot_len - SizeOfXLogCompressedRecord; + bool decomp_success = true; + char *dst; + XLogRecord *dst_h; + + /* + * Grow the decompression buffer if needed, rounding up to BLCKSZ to + * avoid frequent small reallocations. Since the buffer content is + * always fully overwritten, we simply pfree and reallocate. + */ + if (state->decompression_buffer_size < src->decompressed_length) + { + uint32 new_size = (uint32) TYPEALIGN(BLCKSZ, src->decompressed_length); + + if (state->decompression_buffer) + pfree(state->decompression_buffer); + state->decompression_buffer = + palloc_extended(new_size, MCXT_ALLOC_NO_OOM); + if (!state->decompression_buffer) + { + state->decompression_buffer_size = 0; + report_invalid_record(state, + "out of memory while decompressing record at %X/%08X", + LSN_FORMAT_ARGS(recptr)); + return NULL; + } + state->decompression_buffer_size = new_size; + } + + dst_h = (XLogRecord *) state->decompression_buffer; + *dst_h = src->record_header; + dst_h->xl_tot_len = src->decompressed_length; + dst = (char *) &dst_h[1]; + + if (src->method == XLR_COMPRESS_LZ4) + { +#ifdef USE_LZ4 + if (LZ4_decompress_safe((char *) &src[1], dst, + srclen, body_len) <= 0) + decomp_success = false; +#else + report_invalid_record(state, + "could not decompress record at %X/%08X compressed with %s not supported by build", + LSN_FORMAT_ARGS((XLogRecPtr) recptr), "lz4"); + return NULL; +#endif + } + else if (src->method == XLR_COMPRESS_ZSTD) + { +#ifdef USE_ZSTD + size_t decomp_result = ZSTD_decompress(dst, body_len, + (char *) &src[1], srclen); + + if (ZSTD_isError(decomp_result)) + decomp_success = false; +#else + report_invalid_record(state, + "could not decompress record at %X/%08X compressed with %s not supported by build", + LSN_FORMAT_ARGS((XLogRecPtr) recptr), "zstd"); + return NULL; +#endif + } + else + { + report_invalid_record(state, + "could not decompress record at %X/%08X compressed with unknown method", + LSN_FORMAT_ARGS((XLogRecPtr) recptr)); + return NULL; + } + + if (!decomp_success) + { + report_invalid_record(state, + "could not decompress record at %X/%08X", + LSN_FORMAT_ARGS(recptr)); + return NULL; + } + + return (XLogRecord *) state->decompression_buffer; +#endif + } + return record; +} + /* * Decode a record. "decoded" must point to a MAXALIGNed memory area that has * space for at least DecodeXLogRecordRequiredSpace(record) bytes. On @@ -1725,6 +1926,14 @@ DecodeXLogRecord(XLogReaderState *state, RelFileLocator *rlocator = NULL; uint8 block_id; + record = XLogDecompressRecordIfNeeded(state, record, lsn); + + if (!record) + { + /* Decompression failed, error must be reported already */ + return false; + } + decoded->header = *record; decoded->lsn = lsn; decoded->next = NULL; @@ -1899,8 +2108,8 @@ DecodeXLogRecord(XLogReaderState *state, blk->bimg_len != BLCKSZ) { report_invalid_record(state, - "neither BKPIMAGE_HAS_HOLE nor BKPIMAGE_COMPRESSED set, but block image length is %d at %X/%08X", - blk->data_len, + "neither BKPIMAGE_HAS_HOLE nor BKPIMAGE_COMPRESSED set, but block image length is %u at %X/%08X", + (unsigned int) blk->bimg_len, LSN_FORMAT_ARGS(state->ReadRecPtr)); goto err; } diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat index adb72361ce0..bddd3a26a51 100644 --- a/src/backend/utils/misc/guc_parameters.dat +++ b/src/backend/utils/misc/guc_parameters.dat @@ -3462,6 +3462,20 @@ variable => 'wal_compression', boot_val => 'WAL_COMPRESSION_NONE', options => 'wal_compression_options', + assign_hook => 'assign_wal_compression', +}, + +{ name => 'wal_compression_threshold', type => 'int', context => 'PGC_SUSET', group => 'WAL_SETTINGS', + short_desc => 'Minimum WAL record length to engage whole-record compression.', + long_desc => 'Records at least this large are compressed as a single unit instead of ' + . 'compressing their full-page images individually. This applies only when ' + . 'wal_compression selects lz4 or zstd. Set above the size of the largest ' + . 'possible WAL record to keep the full-page-image-only behavior.', + flags => 'GUC_UNIT_BYTE', + variable => 'wal_compression_threshold', + boot_val => '512', + min => '32', + max => 'INT_MAX', }, { name => 'wal_consistency_checking', type => 'string', context => 'PGC_SUSET', group => 'DEVELOPER_OPTIONS', diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 7958653077b..406b8fc52ca 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -261,8 +261,11 @@ #full_page_writes = on # recover from partial page writes #wal_log_hints = off # also do full page writes of non-critical updates # (change requires restart) -#wal_compression = off # enables compression of full-page writes; +#wal_compression = off # enables compression of full-page writes + # and of whole WAL records; # off, pglz (or "on"), lz4, or zstd +#wal_compression_threshold = 512 # min 32, smallest record compressed as a whole; + # whole-record compression needs lz4 or zstd #wal_init_zero = on # zero-fill new WAL files #wal_recycle = on # recycle WAL files #wal_buffers = -1 # min 32kB, -1 sets based on shared_buffers diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index 4dd98624204..7fda7eb1f23 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -58,6 +58,7 @@ extern PGDLLIMPORT int CommitSiblings; extern PGDLLIMPORT bool track_wal_io_timing; extern PGDLLIMPORT int wal_decode_buffer_size; extern PGDLLIMPORT int data_checksums; +extern PGDLLIMPORT int wal_compression_threshold; extern PGDLLIMPORT int CheckPointSegments; diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h index 4a9a687e879..988734f8411 100644 --- a/src/include/access/xlogreader.h +++ b/src/include/access/xlogreader.h @@ -251,6 +251,10 @@ struct XLogReaderState char *decode_buffer_head; /* data is read from the head */ char *decode_buffer_tail; /* new data is written at the tail */ + /* Buffer for decompressing whole-record compressed WAL records */ + char *decompression_buffer; + uint32 decompression_buffer_size; + /* * Queue of records that have been decoded. This is a linked list that * usually consists of consecutive records in decode_buffer, but may also diff --git a/src/include/access/xlogrecord.h b/src/include/access/xlogrecord.h index e8999d3fe91..be662e6558f 100644 --- a/src/include/access/xlogrecord.h +++ b/src/include/access/xlogrecord.h @@ -48,7 +48,10 @@ typedef struct XLogRecord /* 2 bytes of padding here, initialize to zero */ pg_crc32c xl_crc; /* CRC for this record */ - /* XLogRecordBlockHeaders and XLogRecordDataHeader follow, no padding */ + /* + * XLogRecordBlockHeaders, XLogRecordDataHeader or compression header + * follow, no padding + */ } XLogRecord; @@ -90,6 +93,9 @@ typedef struct XLogRecord */ #define XLR_CHECK_CONSISTENCY 0x02 +/* This bit in xl_info means the record is compressed */ +#define XLR_COMPRESSED 0x04 + /* * Header info for block data appended to an XLOG record. * @@ -143,11 +149,6 @@ typedef struct XLogRecordBlockImageHeader uint16 length; /* number of page image bytes */ uint16 hole_offset; /* number of bytes before "hole" */ uint8 bimg_info; /* flag bits, see below */ - - /* - * If BKPIMAGE_HAS_HOLE and BKPIMAGE_COMPRESSED(), an - * XLogRecordBlockCompressHeader struct follows. - */ } XLogRecordBlockImageHeader; #define SizeOfXLogRecordBlockImageHeader \ @@ -157,13 +158,13 @@ typedef struct XLogRecordBlockImageHeader #define BKPIMAGE_HAS_HOLE 0x01 /* page image has "hole" */ #define BKPIMAGE_APPLY 0x02 /* page image should be restored * during replay */ -/* compression methods supported */ +/* Compression methods supported for FPI (stored in bimg_info) */ #define BKPIMAGE_COMPRESS_PGLZ 0x04 #define BKPIMAGE_COMPRESS_LZ4 0x08 #define BKPIMAGE_COMPRESS_ZSTD 0x10 #define BKPIMAGE_COMPRESSED(info) \ - ((info & (BKPIMAGE_COMPRESS_PGLZ | BKPIMAGE_COMPRESS_LZ4 | \ + (((info) & (BKPIMAGE_COMPRESS_PGLZ | BKPIMAGE_COMPRESS_LZ4 | \ BKPIMAGE_COMPRESS_ZSTD)) != 0) /* @@ -178,6 +179,37 @@ typedef struct XLogRecordBlockCompressHeader #define SizeOfXLogRecordBlockCompressHeader \ sizeof(XLogRecordBlockCompressHeader) +/* compression methods supported for whole-record compression */ +#define XLR_COMPRESS_LZ4 0x08 +#define XLR_COMPRESS_ZSTD 0x10 + +/* + * Whole-record compression is only offered for methods that earn it, so a + * build without any of them carries neither side of the machinery. It must + * still recognize such a record when reading, to report it properly. + */ +#if defined(USE_LZ4) || defined(USE_ZSTD) +#define WAL_WHOLE_RECORD_COMPRESSION 1 +#endif + +/* + * Header prepended to a whole-record compressed WAL record. + * + * The compressed payload follows immediately, at SizeOfXLogCompressedRecord. + * record_header.xl_tot_len covers this header plus that payload, so it is the + * on-disk size; decompressed_length is what the record expands back to, + * including its XLogRecord header. + */ +typedef struct XLogCompressionHeader +{ + XLogRecord record_header; + uint32 decompressed_length; + uint8 method; /* XLR_COMPRESS_* */ + /* 3 bytes of padding here, initialize to zero */ +} XLogCompressionHeader; + +#define SizeOfXLogCompressedRecord sizeof(XLogCompressionHeader) + /* * Maximum size of the header for a block reference. This is used to size a * temporary buffer for constructing the header. diff --git a/src/include/utils/guc_hooks.h b/src/include/utils/guc_hooks.h index 6a76f8d5ed6..15825fa126c 100644 --- a/src/include/utils/guc_hooks.h +++ b/src/include/utils/guc_hooks.h @@ -172,6 +172,7 @@ extern const char *show_unix_socket_permissions(void); extern bool check_wal_buffers(int *newval, void **extra, GucSource source); extern bool check_wal_consistency_checking(char **newval, void **extra, GucSource source); +extern void assign_wal_compression(int newval, void *extra); extern void assign_wal_consistency_checking(const char *newval, void *extra); extern bool check_wal_segment_size(int *newval, void **extra, GucSource source); extern void assign_wal_sync_method(int new_wal_sync_method, void *extra); diff --git a/src/test/perl/PostgreSQL/Test/Cluster.pm b/src/test/perl/PostgreSQL/Test/Cluster.pm index 3eae4cf6281..96aaa88f1ce 100644 --- a/src/test/perl/PostgreSQL/Test/Cluster.pm +++ b/src/test/perl/PostgreSQL/Test/Cluster.pm @@ -719,9 +719,21 @@ sub init # TEMP_CONFIG. Otherwise, print it before TEMP_CONFIG, thereby permitting # overrides. Settings that merely improve performance or ease debugging # belong before TEMP_CONFIG. + print $conf PostgreSQL::Test::Utils::slurp_file($ENV{TEMP_CONFIG}) if defined $ENV{TEMP_CONFIG}; + # A test that reasons about exact WAL positions cannot tolerate + # whole-record compression: the record it emits would shrink by an amount + # it has no way to predict. Such a test passes no_wal_compression => 1 to + # init(), and we put the record back beyond any possible record size. + # + # This is printed after TEMP_CONFIG on purpose. It is a correctness + # requirement of the test, not a preference, so it must win over whatever + # the buildfarm animal supplies. + print $conf "wal_compression_threshold = " . (1024 * 1024 * 1024) . "\n" + if $params{no_wal_compression}; + if ($params{allows_streaming}) { if ($params{allows_streaming} eq "logical") @@ -3191,10 +3203,16 @@ sub emit_wal { my ($self, $size) = @_; + # Disable whole-record WAL compression for this emission. Tests call + # emit_wal() to place an exact number of bytes in WAL (e.g. to reach a + # specific page offset); whole-record compression would shrink the record + # and break those position-sensitive calculations. return int( $self->safe_psql( 'postgres', - "SELECT pg_logical_emit_message(true, '', repeat('a', $size)) - '0/0'" + "SET wal_compression_threshold = 2147483647; + SELECT pg_logical_emit_message(true, '', repeat('a', $size)) - '0/0'; + RESET wal_compression_threshold" )); } diff --git a/src/test/recovery/Makefile b/src/test/recovery/Makefile index d41aaaf8ae1..f4c9cd89811 100644 --- a/src/test/recovery/Makefile +++ b/src/test/recovery/Makefile @@ -24,6 +24,16 @@ export enable_injection_points REGRESS_SHLIB=$(abs_top_builddir)/src/test/regress/regress$(DLSUFFIX) export REGRESS_SHLIB +# Exercise WAL compression in recovery tests. Set USE_WAL_COMPRESSION_CONFIG= +# to disable, or TEMP_CONFIG=/path to use a different config. +USE_WAL_COMPRESSION_CONFIG ?= 1 +ifneq ($(USE_WAL_COMPRESSION_CONFIG),) +TEMP_CONFIG ?= $(srcdir)/wal_compression.conf +export TEMP_CONFIG +else +unexport TEMP_CONFIG +endif + check: $(prove_check) diff --git a/src/test/recovery/t/026_overwrite_contrecord.pl b/src/test/recovery/t/026_overwrite_contrecord.pl index 82567ca6bfc..47342bb8c98 100644 --- a/src/test/recovery/t/026_overwrite_contrecord.pl +++ b/src/test/recovery/t/026_overwrite_contrecord.pl @@ -16,7 +16,9 @@ use Test::More; # file and the standby promotes successfully. my $node = PostgreSQL::Test::Cluster->new('primary'); -$node->init(allows_streaming => 1); +# This test emits a message of a chosen size and requires the resulting record +# to cross a segment boundary, so the record must reach WAL at that size. +$node->init(allows_streaming => 1, no_wal_compression => 1); # We need these settings for stability of WAL behavior. $node->append_conf( 'postgresql.conf', qq( diff --git a/src/test/recovery/t/046_checkpoint_logical_slot.pl b/src/test/recovery/t/046_checkpoint_logical_slot.pl index 66761bf56c1..5c56cb1265c 100644 --- a/src/test/recovery/t/046_checkpoint_logical_slot.pl +++ b/src/test/recovery/t/046_checkpoint_logical_slot.pl @@ -115,10 +115,9 @@ $node->safe_psql('postgres', q{select pg_replication_slot_advance('slot_physical', pg_current_wal_lsn())} ); -# Generate a long WAL record, spawning at least two pages for the follow-up +# Generate a long WAL record, spanning at least two pages for the follow-up # post-recovery check. -$node->safe_psql('postgres', - q{select pg_logical_emit_message(false, '', repeat('123456789', 1000))}); +$node->emit_wal(9000); # Continue the checkpoint and wait for its completion. my $log_offset = -s $node->logfile; diff --git a/src/test/recovery/t/052_wal_compression.pl b/src/test/recovery/t/052_wal_compression.pl new file mode 100644 index 00000000000..001772daf23 --- /dev/null +++ b/src/test/recovery/t/052_wal_compression.pl @@ -0,0 +1,152 @@ +# Copyright (c) 2025-2026, PostgreSQL Global Development Group +# +# Test whole-record WAL compression via wal_compression and +# wal_compression_threshold. Exercises compression during replication +# (walsender path) and decompression during crash recovery (startup path). +# + +use strict; +use warnings FATAL => 'all'; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +# Determine which compression methods are compiled in, preferring the +# faster hardware-accelerated codecs first for quicker failure feedback. +my @methods = (); +push @methods, 'zstd' if check_pg_config('#define HAVE_LIBZSTD 1'); +push @methods, 'lz4' if check_pg_config('#define HAVE_LIBLZ4 1'); +push @methods, 'pglz'; + +# Test whole-record WAL compression with a specific method. +# +# Creates a primary with the method enabled and a low threshold so that +# every non-trivial WAL record is compressed. A streaming standby is +# used to verify that compressed WAL is transmitted and decoded correctly. +# Then the primary is stopped immediately (simulating a crash) and +# restarted to verify that startup recovery can decompress WAL records. +# Emit one 256kB compressible logical message at the given threshold and +# report how many bytes of WAL it consumed. +sub wal_used +{ + my ($node, $threshold) = @_; + + my $lsns = $node->safe_psql( + 'postgres', qq{ + SET wal_compression_threshold = $threshold; + SELECT pg_current_wal_insert_lsn(); + SELECT pg_logical_emit_message(true, 'test 052', repeat('abcd', 65536)); + SELECT pg_current_wal_insert_lsn(); + }); + + # Three result rows: the LSN before, the message's own LSN, the LSN after. + my ($before, $after) = (split /\n/, $lsns)[ 0, 2 ]; + return $node->safe_psql('postgres', + "SELECT '$after'::pg_lsn - '$before'::pg_lsn"); +} + +sub test_wal_compression +{ + my ($method) = @_; + + note "testing wal_compression = $method"; + + my $primary = PostgreSQL::Test::Cluster->new("primary_$method"); + $primary->init(allows_streaming => 1); + + # Use the minimum threshold so virtually every record gets compressed. + $primary->append_conf( + 'postgresql.conf', + "wal_compression = '$method'\n" + . "wal_compression_threshold = 32\n"); + $primary->start; + + my $backup_name = "backup_$method"; + $primary->backup($backup_name); + + my $standby = PostgreSQL::Test::Cluster->new("standby_$method"); + $standby->init_from_backup($primary, $backup_name, has_streaming => 1); + $standby->start; + + # Prove the feature actually engages, rather than silently doing nothing. + # A large, highly compressible logical message is one record with no + # full-page images, so the WAL it occupies is decided purely by whole-record + # compression. Measure it against the same message with the threshold + # raised out of reach. + my $compressed_bytes = wal_used($primary, 32); + my $plain_bytes = wal_used($primary, 1024 * 1024 * 1024); + + if ($method eq 'pglz') + { + # pglz is excluded from whole-record compression on purpose, so the + # threshold must make no difference at all. + is($compressed_bytes, $plain_bytes, + "whole-record compression does not engage for pglz"); + } + else + { + cmp_ok($compressed_bytes, '<', $plain_bytes / 2, + "whole-record compression shrinks the record ($method): " + . "$compressed_bytes vs $plain_bytes bytes"); + } + + my $start_lsn = + $primary->safe_psql('postgres', 'SELECT pg_current_wal_insert_lsn()'); + + # Generate WAL with records that exceed the compression threshold. + # Each row's WAL record includes 200-byte payload well above 32 bytes. + $primary->safe_psql( + 'postgres', + "CREATE TABLE t AS + SELECT g, repeat('x', 200) AS d + FROM generate_series(1, 100) AS g"); + + $primary->wait_for_replay_catchup($standby); + + is( $standby->safe_psql('postgres', 'SELECT count(*) FROM t'), + '100', + "compressed WAL replicated via streaming ($method)"); + + # Insert another batch that will be recovered after the simulated crash. + $primary->safe_psql( + 'postgres', + "INSERT INTO t + SELECT g, repeat('y', 200) + FROM generate_series(101, 200) AS g"); + + # pg_waldump has to walk these records too. That is the frontend decoding + # path, which reaches XLogDecompressRecordIfNeeded() without any of the + # server's state, so it is worth covering separately from recovery. + my $end_lsn = + $primary->safe_psql('postgres', 'SELECT pg_current_wal_flush_lsn()'); + + $primary->command_ok( + [ + 'pg_waldump', + '--quiet', + '--path' => $primary->data_dir . '/pg_wal', + '--start' => $start_lsn, + '--end' => $end_lsn + ], + "pg_waldump decodes compressed WAL ($method)"); + + # Stop without a clean shutdown. On restart PostgreSQL will replay WAL + # from the last checkpoint, exercising XLogDecompressRecordIfNeeded for + # every compressed record generated since that checkpoint. + $primary->stop('immediate'); + $primary->start; + + is( $primary->safe_psql('postgres', 'SELECT count(*) FROM t'), + '200', + "crash recovery replays compressed WAL ($method)"); + + $primary->stop; + $standby->stop; +} + +foreach my $method (@methods) +{ + test_wal_compression($method); +} + +done_testing(); diff --git a/src/test/recovery/wal_compression.conf b/src/test/recovery/wal_compression.conf new file mode 100644 index 00000000000..7cb690d1665 --- /dev/null +++ b/src/test/recovery/wal_compression.conf @@ -0,0 +1,6 @@ +# Exercise WAL compression throughout the recovery tests. +# +# pglz is used deliberately: it is always available, so this needs no guard for +# builds configured without --with-lz4 or --with-zstd. 052_wal_compression.pl +# covers lz4 and zstd separately, skipping whichever a build lacks. +wal_compression = 'pglz' diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 56c1f997f88..be79df27aee 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -3507,6 +3507,7 @@ X509_STORE X509_STORE_CTX X86FeatureId XLTW_Oper +XLogCompressionHeader XLogCtlData XLogCtlInsert XLogDumpConfig -- 2.50.1 (Apple Git-155)