From 38e112f68776733f5c159f104f67a181f2fe232d Mon Sep 17 00:00:00 2001 From: Andrey Borodin Date: Sun, 26 Jul 2026 18:53:56 +0500 Subject: [PATCH v12 3/3] WIP: compress WAL records against earlier records in a stream Small WAL records rarely compress well in isolation, but often resemble earlier records. Add optional zstd streams that retain compression history across records. Backends lease a stream while compressing and inserting, so compression order agrees with WAL order. One compressor follows each writer's current lease, resetting when ownership changes. Restart streams at fixed 4MB boundaries. Readers starting partway through WAL rewind to a boundary to reconstruct the decompression state. Adapt pg_waldump, pg_walinspect, logical decoding, walsummarizer and pg_rewind to this starting procedure. This remains experimental: reader memory grows with the stream count, and rewinding needs earlier WAL to remain available. Streams are off by default. The WAL format change requires an XLOG_PAGE_MAGIC bump at commit time. Author: Andrey Borodin Discussion: https://www.postgresql.org/message-id/04119E77-9C0F-415A-A9E0-65E6E4A57940@yandex-team.ru --- contrib/pg_walinspect/pg_walinspect.c | 8 +- doc/src/sgml/config.sgml | 35 ++ src/backend/access/transam/xlog.c | 273 ++++++++++++++- src/backend/access/transam/xloginsert.c | 278 +++++++++++++++ src/backend/access/transam/xlogreader.c | 325 ++++++++++++++++-- src/backend/access/transam/xlogrecovery.c | 8 + src/backend/postmaster/walsummarizer.c | 5 +- src/backend/replication/logical/logical.c | 8 +- .../replication/logical/logicalfuncs.c | 3 +- src/backend/replication/walsender.c | 11 +- .../utils/activity/wait_event_names.txt | 1 + src/backend/utils/misc/guc_parameters.dat | 12 + src/backend/utils/misc/postgresql.conf.sample | 4 + src/bin/pg_rewind/parsexlog.c | 24 +- src/bin/pg_waldump/pg_waldump.c | 10 +- src/include/access/xlog.h | 5 + src/include/access/xlogreader.h | 27 ++ src/include/access/xlogrecord.h | 35 +- src/include/storage/lwlocklist.h | 1 + src/test/perl/PostgreSQL/Test/Cluster.pm | 10 +- .../recovery/t/043_no_contrecord_switch.pl | 5 +- src/test/recovery/t/061_wal_compression.pl | 187 +++++++++- src/tools/pgindent/typedefs.list | 2 + 23 files changed, 1218 insertions(+), 59 deletions(-) diff --git a/contrib/pg_walinspect/pg_walinspect.c b/contrib/pg_walinspect/pg_walinspect.c index 97ffb17d59a..0694acb5861 100644 --- a/contrib/pg_walinspect/pg_walinspect.c +++ b/contrib/pg_walinspect/pg_walinspect.c @@ -127,8 +127,12 @@ InitXLogReaderState(XLogRecPtr lsn) errmsg("out of memory"), errdetail("Failed while allocating a WAL reading processor."))); - /* first find a valid recptr to start from */ - first_valid_record = XLogFindNextRecord(xlogreader, lsn, &errormsg); + /* + * Find a valid recptr to start from. A record can be compressed against + * earlier records of its stream, so this backs up to where the streams + * start over and reads forward from there to rebuild the decompressors. + */ + first_valid_record = XLogBeginReadStreamed(xlogreader, lsn, &errormsg); if (!XLogRecPtrIsValid(first_valid_record)) { diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 1b9b39f5179..714cabea967 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -3703,6 +3703,41 @@ include_dir 'conf.d' + + wal_compression_streams (integer) + + wal_compression_streams configuration parameter + + + + + Sets the number of shared WAL compression streams. The default is + zero, which disables stream compression. When this is nonzero and + is zstd, + records can be compressed using the history of earlier records in + the same stream. This can reduce the size of small records that do + not compress well individually. The maximum is 64. This parameter + can only be set at server start. + + + Compression history restarts at 4MB WAL boundaries. A reader + starting within WAL may need to read from an earlier boundary to + rebuild that history, and replication slots retain WAL back to + that boundary. Each reader needs a separate decompression context + for every stream it encounters. More streams can therefore increase + memory use during recovery, logical decoding and WAL inspection. + + + Each writer retains one stream compression context, resetting it + when its lease changes. Streams ignore + and can make individual + records larger: keeping such records preserves compression history + for subsequent records. Set this parameter to zero for FPI-only + operation, in addition to raising the threshold. + + + + wal_compression_threshold (integer) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 8af539f30b2..73adef2810a 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -143,6 +143,7 @@ int max_slot_wal_keep_size_mb = -1; int wal_decode_buffer_size = 512 * 1024; bool track_wal_io_timing = false; int wal_compression_threshold = 512; +int wal_compression_streams = 0; #ifdef WAL_DEBUG bool XLOG_DEBUG = false; @@ -398,6 +399,22 @@ typedef union WALInsertLockPadded */ static SessionBackupState sessionBackupState = SESSION_BACKUP_NONE; +typedef struct WALCompressionSlot +{ + LWLock lock; /* held across compress + insert */ + int owner; /* ProcNumber holding the lease, or -1 */ + uint64 generation; /* bumped whenever the stream restarts */ + XLogRecPtr redo; /* RedoRecPtr the stream started under */ + XLogRecPtr last_write; /* end of the last record in this stream */ + bool restart; /* next record has to restart the stream */ +} WALCompressionSlot; + +typedef union WALCompressionSlotPadded +{ + WALCompressionSlot s; + char pad[PG_CACHE_LINE_SIZE]; +} WALCompressionSlotPadded; + /* * Shared state data for WAL insertion. */ @@ -450,6 +467,12 @@ typedef struct XLogCtlInsert * WAL insertion locks. */ WALInsertLockPadded *WALInsertLocks; + + /* + * The compression stream slots, kept here so that a process which only + * attaches to shared memory, rather than inheriting it, can find them. + */ + WALCompressionSlotPadded *WALCompressionSlots; } XLogCtlInsert; /* @@ -591,6 +614,28 @@ typedef enum static XLogCtlData *XLogCtl = NULL; +/* + * State of one WAL compression stream. + * + * The compressor itself cannot live here: it is a libzstd object full of + * pointers, and shared memory is not mapped at the same address in every + * process under EXEC_BACKEND. So a backend keeps its own compressor and + * leases the stream: while it owns the slot it may compress against what it + * put there before, and a backend taking the slot over has to start the + * stream afresh, which it announces with XLR_STREAM_RESET. + */ + +static WALCompressionSlotPadded *WALCompressionSlots = NULL; + +/* + * The stream this backend is holding between XLogCompressionStreamAcquire() + * and XLogCompressionStreamRelease(), and whether the record it compressed + * starts that stream over. XLogInsertRecord() needs both to tell whether the + * position it is about to reserve is one the record may be placed at. + */ +static int insert_stream_slot = -1; +static bool insert_stream_restarted = false; + /* a private copy of XLogCtl->Insert.WALInsertLocks, for convenience */ static WALInsertLockPadded *WALInsertLocks = NULL; @@ -789,8 +834,9 @@ static void CopyXLogRecordToWAL(int write_len, bool isLogSwitch, XLogRecData *rdata, XLogRecPtr StartPos, XLogRecPtr EndPos, TimeLineID tli); -static void ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos, - XLogRecPtr *EndPos, XLogRecPtr *PrevPtr); +static bool ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos, + XLogRecPtr *EndPos, XLogRecPtr *PrevPtr, + uint64 startbefore); static bool ReserveXLogSwitch(XLogRecPtr *StartPos, XLogRecPtr *EndPos, XLogRecPtr *PrevPtr); static XLogRecPtr WaitXLogInsertionsToFinish(XLogRecPtr upto); @@ -822,6 +868,114 @@ XLogGetRecordTotalLen(XLogRecord *record) } #endif +/* + * Take a compression stream, or report that none is worth taking. + * + * "preferred" is the slot this backend used last, or -1. Returns the slot to + * use and sets *restart if its stream has to start over, or returns -1, which + * means the caller should compress this record on its own. The lock is held + * until XLogCompressionStreamRelease(), so that records enter a stream in the + * same order they are given their LSNs. + * + * Keeping a stream is what makes it pay, so a backend that keeps writing + * keeps its slot, and we never take one away from a backend that is still + * feeding it. A slot is up for grabs only if nobody owns it, or if its last + * record fell below the latest reset boundary: such a stream has to start over + * anyway, so taking it costs its owner nothing. This also returns the slots + * of backends that went idle or died, without needing to be told that they + * did. + */ +int +XLogCompressionStreamAcquire(int preferred, XLogRecPtr redo, bool *restart) +{ + XLogRecPtr insert_at; + int i; + + /* Fast path: the slot we had, if it is still ours. */ + if (preferred >= 0) + { + WALCompressionSlot *s = &WALCompressionSlots[preferred].s; + + LWLockAcquire(&s->lock, LW_EXCLUSIVE); + if (s->owner == MyProcNumber) + { + *restart = (s->restart || s->redo != redo); + if (*restart) + { + s->generation++; + s->redo = redo; + s->restart = false; + } + insert_stream_slot = preferred; + insert_stream_restarted = *restart; + return preferred; + } + LWLockRelease(&s->lock); + } + + /* + * Slow path. Reading the insert position takes a spinlock, which is why + * it is only done here, when we have no slot of our own. + */ + insert_at = GetXLogInsertRecPtr(); + + for (i = 0; i < wal_compression_streams; i++) + { + WALCompressionSlot *s = &WALCompressionSlots[i].s; + + if (!LWLockConditionalAcquire(&s->lock, LW_EXCLUSIVE)) + continue; /* busy; its owner is clearly active */ + + if (s->owner == -1 || + XLogCompressionStreamCrosses(s->last_write, insert_at)) + { + s->owner = MyProcNumber; + s->generation++; + s->redo = redo; + s->restart = false; + s->last_write = insert_at; + *restart = true; + insert_stream_slot = i; + insert_stream_restarted = true; + return i; + } + LWLockRelease(&s->lock); + } + + return -1; +} + +/* + * Release the lease. "failed" means the record never made it into WAL, so + * whatever we fed the compressor has to be thrown away by everyone. + */ +void +XLogCompressionStreamRelease(int slot, XLogRecPtr end_pos, bool restarted, + bool failed) +{ + WALCompressionSlot *s = &WALCompressionSlots[slot].s; + + insert_stream_slot = -1; + + if (failed) + { + s->owner = -1; + s->restart = true; + } + else if (XLogRecPtrIsValid(end_pos)) + { + /* + * A record that crossed a boundary is of no use to a reader starting + * at that boundary, which skips it as continuation data. So the next + * record has to start over even when this one already did. + */ + if (XLogCompressionStreamCrosses(s->last_write, end_pos - 1)) + s->restart = true; + s->last_write = end_pos - 1; /* where this stream last wrote */ + } + LWLockRelease(&s->lock); +} + /* * Insert an XLOG record represented by an already-constructed chain of data * chunks. This is a low-level routine; to construct the WAL record header @@ -865,8 +1019,9 @@ XLogInsertRecord(XLogRecData *rdata, XLogRecord *rechdr = (XLogRecord *) rdata->data; uint8 info = rechdr->xl_info & ~XLR_INFO_MASK; WalInsertClass class = WALINSERT_NORMAL; - XLogRecPtr StartPos; - XLogRecPtr EndPos; + XLogRecPtr StartPos = InvalidXLogRecPtr; + XLogRecPtr EndPos = InvalidXLogRecPtr; + uint64 startbefore; bool prevDoPageWrites = doPageWrites; TimeLineID insertTLI; @@ -924,6 +1079,25 @@ XLogInsertRecord(XLogRecData *rdata, * *---------- */ + + /* + * A record that continues a stream has to stay below the next reset + * boundary, since every stream starts over there and a reader beginning + * at the boundary keeps nothing from before it. Convert the boundary to + * a byte position here, where no lock is held; the reservation itself + * then costs one comparison. + */ + startbefore = PG_UINT64_MAX; + if (insert_stream_slot >= 0 && !insert_stream_restarted) + { + XLogRecPtr last_write = WALCompressionSlots[insert_stream_slot].s.last_write; + XLogRecPtr boundary; + + boundary = (last_write / WAL_COMPRESSION_STREAM_RESET + 1) * + WAL_COMPRESSION_STREAM_RESET; + startbefore = XLogRecPtrToBytePos(boundary); + } + START_CRIT_SECTION(); if (likely(class == WALINSERT_NORMAL)) @@ -951,6 +1125,21 @@ XLogInsertRecord(XLogRecData *rdata, Assert(RedoRecPtr < Insert->RedoRecPtr); RedoRecPtr = Insert->RedoRecPtr; } + + /* + * A stream must not span a checkpoint's redo point either: recovery + * begins there with nothing earlier to decompress against. Streams + * do start over when RedoRecPtr moves, but the copy this record was + * compressed under was read before the lock was taken and may already + * have been stale, so check it now that it cannot be. + */ + if (insert_stream_slot >= 0 && !insert_stream_restarted && + WALCompressionSlots[insert_stream_slot].s.redo != RedoRecPtr) + { + WALInsertLockRelease(); + END_CRIT_SECTION(); + return InvalidXLogRecPtr; + } doPageWrites = (Insert->fullPageWrites || Insert->runningBackups > 0); if (doPageWrites && @@ -970,8 +1159,20 @@ XLogInsertRecord(XLogRecData *rdata, * Reserve space for the record in the WAL. This also sets the xl_prev * pointer. */ - ReserveXLogInsertLocation(rechdr->xl_tot_len, &StartPos, &EndPos, - &rechdr->xl_prev); + if (!ReserveXLogInsertLocation(rechdr->xl_tot_len, &StartPos, &EndPos, + &rechdr->xl_prev, startbefore)) + { + /* + * The record would land past a stream reset boundary while + * continuing a stream that began below it. A reader starting at + * that boundary would have nothing to decompress it against, so + * throw the record away and let the caller build it again against + * a stream that starts over. + */ + WALInsertLockRelease(); + END_CRIT_SECTION(); + return InvalidXLogRecPtr; + } /* Normal records are always inserted. */ inserted = true; @@ -1006,8 +1207,9 @@ XLogInsertRecord(XLogRecData *rdata, */ Assert(!XLogRecPtrIsValid(fpw_lsn)); WALInsertLockAcquireExclusive(); - ReserveXLogInsertLocation(rechdr->xl_tot_len, &StartPos, &EndPos, - &rechdr->xl_prev); + (void) ReserveXLogInsertLocation(rechdr->xl_tot_len, &StartPos, + &EndPos, &rechdr->xl_prev, + PG_UINT64_MAX); RedoRecPtr = Insert->RedoRecPtr = StartPos; inserted = true; } @@ -1220,9 +1422,9 @@ XLogInsertRecord(XLogRecData *rdata, * however, because there are two call sites, the compiler is reluctant to * inline. We use pg_always_inline here to try to convince it. */ -static pg_always_inline void +static pg_always_inline bool ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos, XLogRecPtr *EndPos, - XLogRecPtr *PrevPtr) + XLogRecPtr *PrevPtr, uint64 startbefore) { XLogCtlInsert *Insert = &XLogCtl->Insert; uint64 startbytepos; @@ -1247,6 +1449,18 @@ ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos, XLogRecPtr *EndPos, SpinLockAcquire(&Insert->insertpos_lck); startbytepos = Insert->CurrBytePos; + + /* + * The caller may only place this record below startbefore. Comparing + * byte positions rather than LSNs is what keeps this to one test: the + * caller converted the boundary once, outside the lock. + */ + if (startbytepos >= startbefore) + { + SpinLockRelease(&Insert->insertpos_lck); + return false; + } + endbytepos = startbytepos + size; prevbytepos = Insert->PrevBytePos; Insert->CurrBytePos = endbytepos; @@ -1265,6 +1479,8 @@ ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos, XLogRecPtr *EndPos, Assert(XLogRecPtrToBytePos(*StartPos) == startbytepos); Assert(XLogRecPtrToBytePos(*EndPos) == endbytepos); Assert(XLogRecPtrToBytePos(*PrevPtr) == prevbytepos); + + return true; } /* @@ -5562,6 +5778,11 @@ XLOGShmemRequest(void *arg) /* WAL insertion locks, plus alignment */ size = add_size(size, mul_size(sizeof(WALInsertLockPadded), NUM_XLOGINSERT_LOCKS + 1)); + + /* WAL compression stream slots, plus alignment slack */ + if (wal_compression_streams > 0) + size = add_size(size, mul_size(sizeof(WALCompressionSlotPadded), + wal_compression_streams + 1)); /* xlblocks array */ size = add_size(size, mul_size(sizeof(pg_atomic_uint64), XLOGbuffers)); /* extra alignment padding for XLOG I/O buffers */ @@ -5629,6 +5850,27 @@ XLOGShmemInit(void *arg) WALInsertLocks[i].l.lastImportantAt = InvalidXLogRecPtr; } + /* WAL compression streams, likewise aligned to their padded size */ + if (wal_compression_streams > 0) + { + allocptr += sizeof(WALCompressionSlotPadded) - + ((uintptr_t) allocptr) % sizeof(WALCompressionSlotPadded); + WALCompressionSlots = XLogCtl->Insert.WALCompressionSlots = + (WALCompressionSlotPadded *) allocptr; + allocptr += sizeof(WALCompressionSlotPadded) * wal_compression_streams; + + for (i = 0; i < wal_compression_streams; i++) + { + LWLockInitialize(&WALCompressionSlots[i].s.lock, + LWTRANCHE_WAL_COMPRESSION_STREAM); + WALCompressionSlots[i].s.owner = -1; + WALCompressionSlots[i].s.generation = 0; + WALCompressionSlots[i].s.redo = InvalidXLogRecPtr; + WALCompressionSlots[i].s.last_write = InvalidXLogRecPtr; + WALCompressionSlots[i].s.restart = true; + } + } + /* * Align the start of the page buffers to a full xlog block size boundary. * This simplifies some calculations in XLOG insertion. It is also @@ -5663,12 +5905,13 @@ XLOGShmemInit(void *arg) } /* - * XLOGShmemAttach - re-establish WALInsertLocks pointer after attaching. + * XLOGShmemAttach - re-establish pointers into shared memory after attaching. */ static void XLOGShmemAttach(void *arg) { WALInsertLocks = XLogCtl->Insert.WALInsertLocks; + WALCompressionSlots = XLogCtl->Insert.WALCompressionSlots; } /* @@ -8949,6 +9192,14 @@ KeepLogSeg(XLogRecPtr recptr, XLogSegNo *logSegNo) keep = XLogGetReplicationSlotMinimumLSN(); if (XLogRecPtrIsValid(keep) && keep < recptr) { + /* + * A reader starting where the slot needs it rewinds to the stream + * reset boundary below that, so keep the WAL from there on. Done + * whatever wal_compression_streams says, since the slot may well + * point into WAL written while it said something else. + */ + keep -= keep % WAL_COMPRESSION_STREAM_RESET; + XLByteToSeg(keep, segno, wal_segment_size); /* diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c index 5d105cb6d3b..cdf76ef201a 100644 --- a/src/backend/access/transam/xloginsert.c +++ b/src/backend/access/transam/xloginsert.c @@ -164,6 +164,29 @@ static int compression_buf_offset; /* fill level for FPI packing */ static char *compressed_data = NULL; #endif +#ifdef USE_ZSTD +/* + * One compressor follows this backend's current lease. Acquiring another + * slot always requires a reset, so retaining contexts for earlier slots is + * unnecessary. The context stays backend-local because it contains pointers. + */ +static ZSTD_CCtx *stream_cctx = NULL; + +/* + * The stream slot this backend is using, or -1. Holding on to it is the + * whole point: a stream only pays once it has seen some records. + * + * When every slot is held by a backend that is still writing, there is + * nothing to take, and we compress records on their own instead. Looking + * for a slot costs a spinlock read, so after a failed search we do not look + * again for a while. + */ +static int my_stream_slot = -1; +static int stream_scan_backoff = 0; + +#define STREAM_SCAN_BACKOFF_RECORDS 1000 +#endif + /* * An array of XLogRecData structs, to hold registered data. */ @@ -509,6 +532,43 @@ XLogSetRecordFlags(uint8 flags) curinsert_flags |= flags; } +#ifdef USE_ZSTD + +/* + * May this record be compressed against earlier ones? + * + * The rule is not about the resource manager but about who reads the record: + * anything that someone looks up by LSN, without replaying what comes before + * it, has to stay readable on its own. Recovery finds the checkpoint record + * from pg_control, and twophase.c reads a PREPARE record from a stored LSN. + * XLOG_SWITCH is looked for by anything scanning WAL, and is tiny anyway. + * XLOG_CHECKPOINT_REDO marks the point recovery starts from, and is read + * before anything that could have preceded it in a stream. + */ +static bool +XLogRecordJoinsStream(RmgrId rmid, uint8 info) +{ + if (rmid == RM_XLOG_ID) + { + uint8 xlinfo = info & ~XLR_INFO_MASK; + + if (xlinfo == XLOG_CHECKPOINT_SHUTDOWN || + xlinfo == XLOG_CHECKPOINT_ONLINE || + xlinfo == XLOG_CHECKPOINT_REDO || + xlinfo == XLOG_END_OF_RECOVERY || + xlinfo == XLOG_SWITCH) + return false; + } + else if (rmid == RM_XACT_ID) + { + if ((info & XLOG_XACT_OPMASK) == XLOG_XACT_PREPARE) + return false; + } + return true; +} + +#endif /* USE_ZSTD */ + /* Compress the assembled record; NULL if that did not pay off */ static XLogRecData * XLogCompressRdt(XLogRecData *rdt) @@ -548,6 +608,7 @@ XLogCompressRdt(XLogRecData *rdt) compressed_header->record_header = *src_header; compressed_header->decompressed_length = flat_len; + compressed_header->stream = XLR_NO_STREAM; orig_len = src_header->xl_tot_len - SizeOfXLogRecord; @@ -626,6 +687,143 @@ XLogCompressRdt(XLogRecData *rdt) #endif } +#ifdef USE_ZSTD +/* + * Compress a record into stream "slot", against everything this stream has + * compressed since it last restarted. + * + * The output has to be complete when we return: the record is about to be + * given an LSN and copied into WAL, so nothing of it may stay inside the + * compressor. That is what ZSTD_e_flush buys, and it is also why the + * compressed length is known before the space is reserved. + * + * Returns NULL if the record did not compress into the space we have. The + * stream is unusable after that, because part of the record was consumed, so + * the caller must restart it. + */ +static XLogRecData * +XLogCompressRdtStream(XLogRecData *rdt, int slot, bool restart, bool *poisoned) +{ + static XLogRecData compressed_rdt_hdr; + XLogCompressionHeader *compressed_header; + XLogRecord *src_header; + uint32 flat_len = 0; + uint32 skipped; + ZSTD_CCtx *cctx; + ZSTD_inBuffer in; + ZSTD_outBuffer out; + size_t rem; + + *poisoned = false; + + if (stream_cctx == NULL) + { + stream_cctx = ZSTD_createCCtx(); + if (stream_cctx == NULL) + { + /* The next acquisition must still start a new stream. */ + *poisoned = true; + return NULL; + } + ZSTD_CCtx_setParameter(stream_cctx, ZSTD_c_compressionLevel, + ZSTD_CLEVEL_DEFAULT); + Assert(restart); + } + cctx = stream_cctx; + + if (restart) + ZSTD_CCtx_reset(cctx, ZSTD_reset_session_only); + + for (const XLogRecData *r = rdt; r != NULL; r = r->next) + flat_len += r->len; + + /* The first chunk is the record header; see XLogInsertRecord(). */ + src_header = (XLogRecord *) rdt->data; + compressed_header = (XLogCompressionHeader *) compressed_data; + + /* Zero it first: the padding in the header reaches disk */ + memset(compressed_header, 0, SizeOfXLogCompressedRecord); + compressed_header->record_header = *src_header; + compressed_header->decompressed_length = flat_len; + compressed_header->method = XLR_COMPRESS_ZSTD; + compressed_header->stream = (uint8) slot; + compressed_header->stream_flags = restart ? XLR_STREAM_RESET : 0; + + out.dst = (char *) &compressed_header[1]; + out.size = WAL_COMPRESSION_BUFSIZE - SizeOfXLogCompressedRecord; + out.pos = 0; + + /* + * Feed the chain as it stands. Flattening it first would have to write + * somewhere, and the only buffer large enough is the one holding the + * compressed page images this very chain points into. + */ + skipped = 0; + for (const XLogRecData *r = rdt; r != NULL; r = r->next) + { + const char *data = r->data; + uint32 len = r->len; + + /* The record header travels in the clear, ahead of the payload. */ + if (skipped < SizeOfXLogRecord) + { + uint32 skip = Min(len, SizeOfXLogRecord - skipped); + + data += skip; + len -= skip; + skipped += skip; + } + if (len == 0) + continue; + + in.src = data; + in.size = len; + in.pos = 0; + while (in.pos < in.size) + { + rem = ZSTD_compressStream2(cctx, &out, &in, ZSTD_e_continue); + if (ZSTD_isError(rem) || out.pos == out.size) + { + /* no room left, and the compressor has eaten part of it */ + *poisoned = true; + return NULL; + } + } + } + + /* End the block, so a reader gets this record back on its own. */ + in.src = NULL; + in.size = 0; + in.pos = 0; + do + { + rem = ZSTD_compressStream2(cctx, &out, &in, ZSTD_e_flush); + if (ZSTD_isError(rem) || (rem != 0 && out.pos == out.size)) + { + *poisoned = true; + return NULL; + } + } while (rem != 0); + + /* + * A record that did not come out smaller is still used. Dropping it + * would mean restarting the stream, since the compressor has already + * consumed it, and a cold stream costs far more than the few bytes this + * record loses. + */ + + compressed_header->record_header.xl_tot_len = + SizeOfXLogCompressedRecord + out.pos; + compressed_header->record_header.xl_info |= XLR_COMPRESSED; + + compressed_rdt_hdr.data = compressed_data; + compressed_rdt_hdr.len = compressed_header->record_header.xl_tot_len; + compressed_rdt_hdr.next = NULL; + + return &compressed_rdt_hdr; +} +#endif /* USE_ZSTD */ + /* Checksum assembled record (which may be compressed). */ static void XLogChecksumRecord(XLogRecData *rdt) @@ -676,6 +874,19 @@ XLogInsert(RmgrId rmid, uint8 info) wal_compression_threshold < WAL_COMPRESSION_BUFSIZE); + /* + * Streams are zstd only for now, and only for records nobody reads out of + * order. One slot per backend, so that a backend usually finds its own + * compressor still in place. + */ +#ifdef USE_ZSTD + bool use_stream; + + use_stream = (wal_compression_streams > 0 && + wal_compression == WAL_COMPRESSION_ZSTD && + XLogRecordJoinsStream(rmid, info)); +#endif + /* XLogBeginInsert() must have been called. */ if (!begininsert_called) elog(ERROR, "XLogBeginInsert was not called"); @@ -733,6 +944,68 @@ XLogInsert(RmgrId rmid, uint8 info) * record is not compressed as a whole after all, reassemble it to get * the per-FPI compression back. */ +#ifdef USE_ZSTD + { + int slot = -1; + bool restart = false; + bool poisoned; + XLogRecData *rdt_compressed; + + /* + * Take a stream, unless a recent look found every one of them + * held by a backend that is still writing. Then hold it while we + * compress and while the record is given its LSN, so the order + * records enter the stream is the order a reader will meet them + * in. + */ + if (use_stream && rec_size <= WAL_COMPRESSION_BUFSIZE && + (my_stream_slot >= 0 || stream_scan_backoff-- <= 0)) + { + slot = XLogCompressionStreamAcquire(my_stream_slot, RedoRecPtr, + &restart); + Assert(slot < 0 || slot == my_stream_slot || restart); + my_stream_slot = slot; + if (slot < 0) + stream_scan_backoff = STREAM_SCAN_BACKOFF_RECORDS; + } + + if (slot < 0) + goto no_stream; + + rdt_compressed = XLogCompressRdtStream(rdt, slot, restart, + &poisoned); + if (rdt_compressed != NULL) + { + XLogChecksumRecord(rdt_compressed); + EndPos = XLogInsertRecord(rdt_compressed, fpw_lsn, + curinsert_flags, num_fpi, + fpi_bytes, topxid_included); + XLogCompressionStreamRelease(slot, EndPos, restart, + !XLogRecPtrIsValid(EndPos)); + if (XLogRecPtrIsValid(EndPos)) + break; + continue; /* retry, with the stream restarted */ + } + + /* Could not compress; the stream may have eaten part of it */ + XLogCompressionStreamRelease(slot, InvalidXLogRecPtr, + restart, poisoned); + if (num_fpi > 0) + { + compression_buf_offset = 0; + rdt = XLogRecordAssemble(rmid, info, RedoRecPtr, doPageWrites, + &fpw_lsn, &num_fpi, &fpi_bytes, + &topxid_included, &rec_size, + false); + } + XLogChecksumRecord(rdt); + EndPos = XLogInsertRecord(rdt, fpw_lsn, curinsert_flags, num_fpi, + fpi_bytes, topxid_included); + continue; + } +no_stream: +#endif + if (try_whole_record) { bool whole_record_compressed = false; @@ -858,6 +1131,11 @@ AllocCompressionBuffers(int method, int threshold) threshold < WAL_COMPRESSION_BUFSIZE) required_size = 2 * WAL_COMPRESSION_BUFSIZE; #endif +#ifdef USE_ZSTD + /* Streams ignore the threshold, including an FPI-only threshold. */ + if (method == WAL_COMPRESSION_ZSTD && wal_compression_streams > 0) + required_size = 2 * WAL_COMPRESSION_BUFSIZE; +#endif if (required_size > compression_buf_size) { diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c index dd139cb2d34..dd5d3b785c8 100644 --- a/src/backend/access/transam/xlogreader.c +++ b/src/backend/access/transam/xlogreader.c @@ -62,11 +62,14 @@ static void xlogreader_memory_context_reset_cb(void *arg); static void XLogReaderFreeZstdContext(void *arg); #endif #endif +static XLogRecPtr XLogFindRecordStart(XLogReaderState *state, XLogRecPtr RecPtr, + char **errormsg); static XLogRecord *XLogDecompressRecordIfNeeded(XLogReaderState *state, XLogRecord *record, XLogRecPtr recptr); #ifdef USE_ZSTD static ZSTD_DCtx *XLogReaderGetZstdContext(XLogReaderState *state); +static void XLogReaderFreeZstdStreams(void *arg); #endif /* size of the buffer allocated for error message. */ @@ -202,6 +205,17 @@ XLogReaderFree(XLogReaderState *state) #endif if (state->decompression_buffer) pfree(state->decompression_buffer); +#ifdef USE_ZSTD + if (state->stream_dctx) + { +#ifndef FRONTEND + MemoryContextUnregisterResetCallback(GetMemoryChunkContext(state), + &state->stream_dctx_cb); +#endif + XLogReaderFreeZstdStreams(state->stream_dctx); + pfree(state->stream_dctx); + } +#endif pfree(state->readBuf); pfree(state); } @@ -324,6 +338,94 @@ XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr) state->NextRecPtr = RecPtr; state->ReadRecPtr = InvalidXLogRecPtr; state->DecodeRecPtr = InvalidXLogRecPtr; + state->warmupEndPtr = InvalidXLogRecPtr; +} + +/* + * Prepare to read records from RecPtr onwards, which unlike XLogBeginRead() + * works even when the records there were compressed against earlier ones. + * + * A record compressed as part of a stream cannot be decompressed on its own, + * so reading starts at the last stream reset boundary at or below RecPtr, + * where every stream is known to begin again. The records in between are + * decoded only to rebuild the decompressors and are not returned by + * XLogReadRecord(). + * + * Returns the position reading actually starts from, or InvalidXLogRecPtr with + * *errormsg set. If the WAL below RecPtr is gone -- a replication slot may + * hold nothing older than what it needs itself -- reading starts at RecPtr, + * which is correct for every record that is not part of a stream. + */ +XLogRecPtr +XLogBeginReadStreamed(XLogReaderState *state, XLogRecPtr RecPtr, + char **errormsg) +{ + XLogRecPtr boundary; + XLogRecPtr found; + char *msg; + + Assert(XLogRecPtrIsValid(RecPtr)); + +#ifdef USE_ZSTD + /* No history from an earlier read position may survive the rewind. */ + if (state->stream_ready != NULL) + memset(state->stream_ready, 0, sizeof(bool) * XLR_MAX_STREAMS); +#endif + + if (errormsg == NULL) + errormsg = &msg; /* the caller is content with the fallback */ + *errormsg = NULL; + boundary = RecPtr - (RecPtr % WAL_COMPRESSION_STREAM_RESET); + + if (boundary == 0) + { + XLogBeginRead(state, RecPtr); + return RecPtr; + } + + found = XLogFindRecordStart(state, boundary, errormsg); + if (XLogRecPtrIsValid(found)) + { + /* + * Read forward from the boundary to bring the decompressors up to + * date, stopping once the next record to read is the one the caller + * asked for. That record must not be read here: a stream record fed + * to its decompressor twice leaves it in a state the next records + * were not compressed against. + */ + XLogBeginRead(state, found); + state->warmupEndPtr = RecPtr; + while (state->NextRecPtr < RecPtr) + { + if (XLogReadRecord(state, errormsg) == NULL) + break; + } + + if (state->NextRecPtr >= RecPtr) + { + /* + * Leave the reader exactly as XLogBeginRead() would have, so that + * callers see no trace of the rewind. The decompressors are not + * part of that state and stay as the warm-up left them. + */ + found = state->NextRecPtr; + XLogBeginRead(state, found); + return found; + } + } + + { + /* + * The WAL below RecPtr is gone; a replication slot need not keep + * anything older than it uses itself. Start where we were asked to, + * which reads every record that is not part of a stream. + */ + *errormsg = NULL; + state->warmupEndPtr = InvalidXLogRecPtr; + return XLogFindNextRecord(state, RecPtr, errormsg); + } + + return found; } /* @@ -1085,7 +1187,26 @@ restart: Assert(decoded != NULL); } - if (DecodeXLogRecord(state, decoded, record, RecPtr, &errormsg)) + if (state->framing_only) + { + /* + * Fill in only what says where this record sits. Nothing reads the + * payload in this mode, and decoding it would run the record through + * a decompressor that has already consumed it. + */ + decoded->header = *record; + decoded->lsn = RecPtr; + decoded->next = NULL; + decoded->record_origin = InvalidReplOriginId; + decoded->toplevel_xid = InvalidTransactionId; + decoded->main_data = NULL; + decoded->main_data_len = 0; + decoded->max_block_id = -1; + decoded->size = MAXALIGN(offsetof(DecodedXLogRecord, blocks)); + } + + if (state->framing_only || + DecodeXLogRecord(state, decoded, record, RecPtr, &errormsg)) { /* Record the location of the next record. */ decoded->next_lsn = state->NextRecPtr; @@ -1114,6 +1235,21 @@ restart: return XLREAD_SUCCESS; } + if (RecPtr < state->warmupEndPtr) + { + /* + * We are only reading this record to feed the decompressors, and its + * stream has not started over yet, so it cannot be decoded and the + * caller was never going to see it. Move on to the next one. + */ + if (decoded->oversized) + pfree(decoded); + state->errormsg_buf[0] = '\0'; + state->errormsg_deferred = false; + RecPtr = state->NextRecPtr; + goto restart; + } + err: if (assembled) { @@ -1598,27 +1734,18 @@ XLogReaderResetError(XLogReaderState *state) } /* - * Find the first record with an lsn >= RecPtr. + * Find where the first record at or after RecPtr starts, without reading it. * - * This is different from XLogBeginRead() in that RecPtr doesn't need to point - * to a valid record boundary. Useful for checking whether RecPtr is a valid - * xlog address for reading, and to find the first valid address after some - * address when dumping records for debugging purposes. + * RecPtr does not need to point to a record boundary: page headers alone tell + * us where to skip continuation data, which is enough to land on the start of + * a record. * - * This positions the reader, like XLogBeginRead(), so that the next call to - * XLogReadRecord() will read the next valid record. - * - * On failure, InvalidXLogRecPtr is returned, and *errormsg is set to a string - * with details of the failure. - * - * When set, *errormsg points to an internal buffer that's valid until the next - * call to XLogReadRecord. + * On failure, InvalidXLogRecPtr is returned and *errormsg is set. */ -XLogRecPtr -XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg) +static XLogRecPtr +XLogFindRecordStart(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg) { XLogRecPtr tmpRecPtr; - XLogRecPtr found = InvalidXLogRecPtr; XLogPageHeader header; *errormsg = NULL; @@ -1705,18 +1832,7 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg) * because either we're at the first record after the beginning of a page * or we just jumped over the remaining data of a continuation. */ - XLogBeginRead(state, tmpRecPtr); - while (XLogReadRecord(state, errormsg) != NULL) - { - /* past the record we've found, break out */ - if (RecPtr <= state->ReadRecPtr) - { - /* Rewind the reader to the beginning of the last record. */ - found = state->ReadRecPtr; - XLogBeginRead(state, found); - return found; - } - } + return tmpRecPtr; err: XLogReaderInvalReadState(state); @@ -1735,6 +1851,49 @@ err: return InvalidXLogRecPtr; } +/* + * Find the first record with an lsn >= RecPtr. + * + * This is different from XLogBeginRead() in that RecPtr doesn't need to point + * to a valid record boundary. Useful for checking whether RecPtr is a valid + * xlog address for reading, and to find the first valid address after some + * address when dumping records for debugging purposes. + * + * This positions the reader, like XLogBeginRead(), so that the next call to + * XLogReadRecord() will read the next valid record. + * + * On failure, InvalidXLogRecPtr is returned, and *errormsg is set to a string + * with details of the failure. + * + * When set, *errormsg points to an internal buffer that's valid until the next + * call to XLogReadRecord. + */ +XLogRecPtr +XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg) +{ + XLogRecPtr start; + + start = XLogFindRecordStart(state, RecPtr, errormsg); + if (!XLogRecPtrIsValid(start)) + return InvalidXLogRecPtr; + + XLogBeginRead(state, start); + while (XLogReadRecord(state, errormsg) != NULL) + { + /* past the record we've found, break out */ + if (RecPtr <= state->ReadRecPtr) + { + /* Rewind the reader to the beginning of the last record. */ + XLogRecPtr found = state->ReadRecPtr; + + XLogBeginRead(state, found); + return found; + } + } + + return InvalidXLogRecPtr; +} + /* * Helper function to ease writing of XLogReaderRoutine->page_read callbacks. * If this function is used, caller must supply a segment_open callback in @@ -1905,6 +2064,17 @@ DecodeXLogRecordRequiredSpace(size_t xl_tot_len) } #ifdef USE_ZSTD +/* Also used as a reset callback in backend builds. */ +static void +XLogReaderFreeZstdStreams(void *arg) +{ + void **contexts = arg; + + for (int i = 0; i < XLR_MAX_STREAMS; i++) + if (contexts[i] != NULL) + ZSTD_freeDCtx(contexts[i]); +} + /* Reuse one context for independent compressed records and page images. */ static ZSTD_DCtx * XLogReaderGetZstdContext(XLogReaderState *state) @@ -1985,7 +2155,104 @@ XLogDecompressRecordIfNeeded(XLogReaderState *state, dst_h->xl_tot_len = src->decompressed_length; dst = (char *) &dst_h[1]; +#ifdef USE_ZSTD + if (src->stream != XLR_NO_STREAM) + { + ZSTD_DCtx *dctx; + ZSTD_inBuffer in; + ZSTD_outBuffer out; + + if (src->method != XLR_COMPRESS_ZSTD) + { + report_invalid_record(state, + "streamed record at %X/%08X uses an unexpected compression method", + LSN_FORMAT_ARGS(recptr)); + return NULL; + } + + if (state->stream_dctx == NULL) + { + size_t ctxsize = sizeof(void *) * XLR_MAX_STREAMS; + size_t total = ctxsize + sizeof(bool) * XLR_MAX_STREAMS; + + /* Allocate both arrays with the reader's lifetime. */ +#ifndef FRONTEND + state->stream_dctx = + MemoryContextAllocZero(GetMemoryChunkContext(state), total); + state->stream_dctx_cb.func = XLogReaderFreeZstdStreams; + state->stream_dctx_cb.arg = state->stream_dctx; + MemoryContextRegisterResetCallback(GetMemoryChunkContext(state), + &state->stream_dctx_cb); +#else + state->stream_dctx = palloc0(total); +#endif + state->stream_ready = (bool *) ((char *) state->stream_dctx + ctxsize); + } + + /* + * A reader that started in the middle of WAL meets records whose + * stream began earlier. Nothing here can decompress them, and + * feeding them to a fresh context would decode to whatever the + * bytes happen to mean, so refuse until the stream starts over. + */ + if (!(src->stream_flags & XLR_STREAM_RESET) && + !state->stream_ready[src->stream]) + { + report_invalid_record(state, + "no decompression state for stream %u at %X/%08X", + src->stream, LSN_FORMAT_ARGS(recptr)); + return NULL; + } + + dctx = (ZSTD_DCtx *) state->stream_dctx[src->stream]; + if (dctx == NULL) + { + dctx = ZSTD_createDCtx(); + if (dctx == NULL) + { + report_invalid_record(state, + "out of memory while decompressing record at %X/%08X", + LSN_FORMAT_ARGS(recptr)); + return NULL; + } + state->stream_dctx[src->stream] = dctx; + } + + if (src->stream_flags & XLR_STREAM_RESET) + { + ZSTD_DCtx_reset(dctx, ZSTD_reset_session_only); + state->stream_ready[src->stream] = true; + } + + in.src = (char *) &src[1]; + in.size = srclen; + in.pos = 0; + out.dst = dst; + out.size = body_len; + out.pos = 0; + + while (in.pos < in.size) + { + size_t ret = ZSTD_decompressStream(dctx, &out, &in); + + if (ZSTD_isError(ret)) + { + decomp_success = false; + break; + } + if (out.pos == out.size && in.pos < in.size) + { + decomp_success = false; + break; + } + } + if (decomp_success && out.pos != body_len) + decomp_success = false; + } + else if (src->method == XLR_COMPRESS_LZ4) +#else if (src->method == XLR_COMPRESS_LZ4) +#endif { #ifdef USE_LZ4 if (LZ4_decompress_safe((char *) &src[1], dst, diff --git a/src/backend/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.c index 20be788da78..394bd8b82af 100644 --- a/src/backend/access/transam/xlogrecovery.c +++ b/src/backend/access/transam/xlogrecovery.c @@ -1493,7 +1493,15 @@ FinishWalRecovery(void) lastRecTLI = XLogRecoveryCtl->lastReplayedTLI; } XLogPrefetcherBeginRead(xlogprefetcher, lastRec); + + /* + * Only the extent of the record and its page are wanted here, and a + * record that was compressed against earlier ones cannot be decoded a + * second time. + */ + xlogreader->framing_only = true; (void) ReadRecord(xlogprefetcher, PANIC, false, lastRecTLI); + xlogreader->framing_only = false; endOfLog = xlogreader->EndRecPtr; /* diff --git a/src/backend/postmaster/walsummarizer.c b/src/backend/postmaster/walsummarizer.c index 8333a4eaaa3..94ff476faf7 100644 --- a/src/backend/postmaster/walsummarizer.c +++ b/src/backend/postmaster/walsummarizer.c @@ -1061,12 +1061,13 @@ SummarizeWAL(TimeLineID tli, XLogRecPtr start_lsn, bool exact, * file is less than the start LSN of the next file. When only a page * header is skipped, nothing has been missed. */ - XLogBeginRead(xlogreader, start_lsn); + XLogBeginReadStreamed(xlogreader, start_lsn, NULL); summary_start_lsn = start_lsn; } else { - summary_start_lsn = XLogFindNextRecord(xlogreader, start_lsn, &errormsg); + summary_start_lsn = XLogBeginReadStreamed(xlogreader, start_lsn, + &errormsg); if (!XLogRecPtrIsValid(summary_start_lsn)) { /* diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 98e5f1dd8f9..30ff38b44a8 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -715,7 +715,7 @@ DecodingContextFindStartpoint(LogicalDecodingContext *ctx) ReplicationSlot *slot = ctx->slot; /* Initialize from where to start reading WAL. */ - XLogBeginRead(ctx->reader, slot->data.restart_lsn); + XLogBeginReadStreamed(ctx->reader, slot->data.restart_lsn, NULL); elog(DEBUG1, "searching for logical decoding starting point, starting at %X/%08X", LSN_FORMAT_ARGS(slot->data.restart_lsn)); @@ -2121,7 +2121,8 @@ LogicalReplicationSlotCheckPendingWal(XLogRecPtr end_of_wal, * Start reading at the slot's restart_lsn, which we know points to a * valid record. */ - XLogBeginRead(ctx->reader, MyReplicationSlot->data.restart_lsn); + XLogBeginReadStreamed(ctx->reader, + MyReplicationSlot->data.restart_lsn, NULL); /* Invalidate non-timetravel entries */ InvalidateSystemCaches(); @@ -2224,7 +2225,8 @@ LogicalSlotAdvanceAndCheckSnapState(XLogRecPtr moveto, * Start reading at the slot's restart_lsn, which we know to point to * a valid record. */ - XLogBeginRead(ctx->reader, MyReplicationSlot->data.restart_lsn); + XLogBeginReadStreamed(ctx->reader, + MyReplicationSlot->data.restart_lsn, NULL); /* invalidate non-timetravel entries */ InvalidateSystemCaches(); diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c index 71fbaf72269..67251c15071 100644 --- a/src/backend/replication/logical/logicalfuncs.c +++ b/src/backend/replication/logical/logicalfuncs.c @@ -243,7 +243,8 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin * xacts that committed after the slot's confirmed_flush can be * accumulated into reorder buffers. */ - XLogBeginRead(ctx->reader, MyReplicationSlot->data.restart_lsn); + XLogBeginReadStreamed(ctx->reader, + MyReplicationSlot->data.restart_lsn, NULL); /* invalidate non-timetravel entries */ InvalidateSystemCaches(); diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 9f21c7c8dfe..24808d25a49 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -1584,9 +1584,6 @@ StartLogicalReplication(StartReplicationCmd *cmd) pq_endmessage(&buf); pq_flush(); - /* Start reading WAL from the oldest required WAL. */ - XLogBeginRead(logical_decoding_ctx->reader, - MyReplicationSlot->data.restart_lsn); /* * Report the location after which we'll send out further commits as the @@ -1603,6 +1600,14 @@ StartLogicalReplication(StartReplicationCmd *cmd) SyncRepInitConfig(); + /* + * Start reading WAL from the oldest required WAL. Done once the + * walsender is otherwise ready, because rebuilding the decompressors + * reads WAL through the same callback the main loop uses. + */ + XLogBeginReadStreamed(logical_decoding_ctx->reader, + MyReplicationSlot->data.restart_lsn, NULL); + /* Main loop of walsender */ WalSndLoop(XLogSendLogical); diff --git a/src/backend/utils/activity/wait_event_names.txt b/src/backend/utils/activity/wait_event_names.txt index 3d366fd1114..c18fd5b9741 100644 --- a/src/backend/utils/activity/wait_event_names.txt +++ b/src/backend/utils/activity/wait_event_names.txt @@ -419,6 +419,7 @@ XactSLRU "Waiting to access the transaction status SLRU cache." ParallelVacuumDSA "Waiting for parallel vacuum dynamic shared memory allocation." AioUringCompletion "Waiting for another process to complete IO via io_uring." ShmemIndex "Waiting to find or allocate space in shared memory." +WALCompressionStream "Waiting to compress a WAL record into a shared compression stream." # No "ABI_compatibility" region here as WaitEventLWLock has its own C code. diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat index 53806cee6cc..46c178c7ecf 100644 --- a/src/backend/utils/misc/guc_parameters.dat +++ b/src/backend/utils/misc/guc_parameters.dat @@ -3470,6 +3470,18 @@ assign_hook => 'assign_wal_compression', }, +{ name => 'wal_compression_streams', type => 'int', context => 'PGC_POSTMASTER', group => 'WAL_SETTINGS', + short_desc => 'Number of WAL compression streams shared by all backends.', + long_desc => 'Zero compresses every record on its own. A positive value lets a record ' + . 'be compressed against the records that preceded it in the same stream, ' + . 'which compresses small records much better, at the cost of one ' + . 'decompression context per stream in every process that reads WAL records.', + variable => 'wal_compression_streams', + boot_val => '0', + min => '0', + max => '64', +}, + { name => 'wal_compression_threshold', type => 'int', context => 'PGC_SUSET', group => 'WAL_SETTINGS', short_desc => 'Minimum WAL record length to engage whole-record compression.', long_desc => 'Records at least this large are compressed as a single unit instead of ' diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 66bd5759298..ae4941b1446 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -267,6 +267,10 @@ # 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_compression_streams = 0 # 0 disables compressing a record against + # earlier ones; a positive value is the + # number of shared compression streams + # (change requires restart) #wal_init_zero = on # zero-fill new WAL files #wal_recycle = on # recycle WAL files #wal_buffers = -1 # min 32kB, -1 sets based on shared_buffers diff --git a/src/bin/pg_rewind/parsexlog.c b/src/bin/pg_rewind/parsexlog.c index 6e87b00f8c2..82419d33585 100644 --- a/src/bin/pg_rewind/parsexlog.c +++ b/src/bin/pg_rewind/parsexlog.c @@ -79,7 +79,7 @@ extractPageMap(const char *datadir, XLogRecPtr startpoint, int tliIndex, if (xlogreader == NULL) pg_fatal("out of memory while allocating a WAL reading processor"); - XLogBeginRead(xlogreader, startpoint); + XLogBeginReadStreamed(xlogreader, startpoint, NULL); do { record = XLogReadRecord(xlogreader, &errormsg); @@ -138,7 +138,12 @@ readOneRecord(const char *datadir, XLogRecPtr ptr, int tliIndex, if (xlogreader == NULL) pg_fatal("out of memory while allocating a WAL reading processor"); + /* + * Only the extent of the record is wanted, so there is no need to decode + * it -- which also means no need to rewind for a compression stream. + */ XLogBeginRead(xlogreader, ptr); + xlogreader->framing_only = true; record = XLogReadRecord(xlogreader, &errormsg); if (record == NULL) { @@ -206,7 +211,13 @@ findLastCheckpoint(const char *datadir, XLogRecPtr forkptr, int tliIndex, { uint8 info; + /* + * Walking back only needs each record's header, so decode nothing: + * rewinding to rebuild a decompressor at every step would make this + * loop cost the whole distance walked, over and over. + */ XLogBeginRead(xlogreader, searchptr); + xlogreader->framing_only = true; record = XLogReadRecord(xlogreader, &errormsg); if (record == NULL) @@ -252,6 +263,17 @@ findLastCheckpoint(const char *datadir, XLogRecPtr forkptr, int tliIndex, { CheckPoint checkPoint; + /* + * This one is wanted whole. A checkpoint record never joins a + * compression stream, so it reads on its own. + */ + XLogBeginRead(xlogreader, searchptr); + xlogreader->framing_only = false; + record = XLogReadRecord(xlogreader, &errormsg); + if (record == NULL) + pg_fatal("could not read checkpoint record at %X/%08X", + LSN_FORMAT_ARGS(searchptr)); + memcpy(&checkPoint, XLogRecGetData(xlogreader), sizeof(CheckPoint)); *lastchkptrec = searchptr; *lastchkpttli = checkPoint.ThisTimeLineID; diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c index 473c692a2af..bb55babcbeb 100644 --- a/src/bin/pg_waldump/pg_waldump.c +++ b/src/bin/pg_waldump/pg_waldump.c @@ -1420,8 +1420,14 @@ main(int argc, char **argv) atexit(cleanup_tmpwal_dir_atexit); xlogreader_state_cleanup = xlogreader_state; - /* first find a valid recptr to start from */ - first_record = XLogFindNextRecord(xlogreader_state, private.startptr, &errormsg); + /* + * Find a valid recptr to start from. A record can be compressed against + * earlier records of its stream, so this backs up to where the streams + * start over and decodes from there; those earlier records rebuild the + * decompressors and are not printed. + */ + first_record = XLogBeginReadStreamed(xlogreader_state, private.startptr, + &errormsg); if (!XLogRecPtrIsValid(first_record)) { diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index 984266b4b8a..1c8deaf0c31 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -59,6 +59,7 @@ extern PGDLLIMPORT bool track_wal_io_timing; extern PGDLLIMPORT int wal_decode_buffer_size; extern PGDLLIMPORT int data_checksums; extern PGDLLIMPORT int wal_compression_threshold; +extern PGDLLIMPORT int wal_compression_streams; extern PGDLLIMPORT int CheckPointSegments; @@ -224,6 +225,10 @@ typedef enum WALAvailability struct XLogRecData; struct XLogReaderState; +extern int XLogCompressionStreamAcquire(int preferred, XLogRecPtr redo, + bool *restart); +extern void XLogCompressionStreamRelease(int slot, XLogRecPtr end_pos, + bool restarted, bool failed); extern XLogRecPtr XLogInsertRecord(struct XLogRecData *rdata, XLogRecPtr fpw_lsn, uint8 flags, diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h index f4034c836d0..5a867d50564 100644 --- a/src/include/access/xlogreader.h +++ b/src/include/access/xlogreader.h @@ -231,6 +231,21 @@ struct XLogReaderState * requested starting position. */ XLogRecPtr DecodeRecPtr; /* start of last record decoded */ + + /* + * Records below this are decoded to rebuild the decompressors and then + * dropped rather than returned; see XLogBeginReadStreamed(). + */ + XLogRecPtr warmupEndPtr; + + /* + * Frame records without decoding them: the caller only wants to know + * where a record ends and to have its page in readBuf. A record that + * belongs to a compression stream cannot be decoded twice anyway, since + * the decompressor has already consumed it. + */ + bool framing_only; + XLogRecPtr NextRecPtr; /* end+1 of last record decoded */ XLogRecPtr PrevRecPtr; /* start of previous record decoded */ @@ -267,6 +282,16 @@ struct XLogReaderState char *decompression_buffer; uint32 decompression_buffer_size; +#ifdef USE_ZSTD + /* One zstd decompression context per stream, indexed by stream ID. */ + void **stream_dctx; + /* Whether a reset record has established each stream's history. */ + bool *stream_ready; +#ifndef FRONTEND + MemoryContextCallback stream_dctx_cb; +#endif +#endif + /* * Queue of records that have been decoded. This is a linked list that * usually consists of consecutive records in decode_buffer, but may also @@ -369,6 +394,8 @@ extern void XLogReaderSetDecodeBuffer(XLogReaderState *state, /* Position the XLogReader to given record */ extern void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr); +extern XLogRecPtr XLogBeginReadStreamed(XLogReaderState *state, + XLogRecPtr RecPtr, char **errormsg); extern XLogRecPtr XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg); diff --git a/src/include/access/xlogrecord.h b/src/include/access/xlogrecord.h index be662e6558f..e9332b5ccb7 100644 --- a/src/include/access/xlogrecord.h +++ b/src/include/access/xlogrecord.h @@ -205,11 +205,44 @@ typedef struct XLogCompressionHeader XLogRecord record_header; uint32 decompressed_length; uint8 method; /* XLR_COMPRESS_* */ - /* 3 bytes of padding here, initialize to zero */ + uint8 stream; /* stream slot, or XLR_NO_STREAM */ + uint8 stream_flags; /* XLR_STREAM_* */ + /* 1 byte of padding here, initialize to zero */ } XLogCompressionHeader; +/* + * A record compressed on its own carries XLR_NO_STREAM. Otherwise "stream" + * names the compression stream it belongs to, and the record can only be + * decompressed after every earlier record of that stream. XLR_STREAM_RESET + * says the stream starts here, so the reader must discard what it had. + */ +#define XLR_MAX_STREAMS 255 +#define XLR_NO_STREAM 0xFF +#define XLR_STREAM_RESET 0x01 + #define SizeOfXLogCompressedRecord sizeof(XLogCompressionHeader) +/* + * Every stream starts over at fixed WAL_COMPRESSION_STREAM_RESET boundaries. + * That is what lets a reader begin in the middle of WAL: it rewinds to the + * last boundary below the record it wants and reads forward from there, since + * every stream begins again at or after such a boundary. The distance is + * therefore what a reader has to re-read, and what a replication slot has to + * keep beyond the WAL it needs for itself. + * + * A fixed distance rather than the WAL segment size, so that how far a reader + * rewinds does not change when a cluster is initialised with a different + * segment size. Compression is insensitive to the value -- pgbench emits the + * same WAL per transaction to within 2% anywhere between 1MB and 64MB -- so it + * is chosen for the readers, not for the ratio. + */ +#define WAL_COMPRESSION_STREAM_RESET (UINT64CONST(4) * 1024 * 1024) + +/* Must a stream that last wrote at "from" start over to write at "to"? */ +#define XLogCompressionStreamCrosses(from, to) \ + ((uint64) (from) / WAL_COMPRESSION_STREAM_RESET != \ + (uint64) (to) / WAL_COMPRESSION_STREAM_RESET) + /* * 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/storage/lwlocklist.h b/src/include/storage/lwlocklist.h index 8d858be9927..43f9c5e4859 100644 --- a/src/include/storage/lwlocklist.h +++ b/src/include/storage/lwlocklist.h @@ -141,3 +141,4 @@ PG_LWLOCKTRANCHE(XACT_SLRU, XactSLRU) PG_LWLOCKTRANCHE(PARALLEL_VACUUM_DSA, ParallelVacuumDSA) PG_LWLOCKTRANCHE(AIO_URING_COMPLETION, AioUringCompletion) PG_LWLOCKTRANCHE(SHMEM_INDEX, ShmemIndex) +PG_LWLOCKTRANCHE(WAL_COMPRESSION_STREAM, WALCompressionStream) diff --git a/src/test/perl/PostgreSQL/Test/Cluster.pm b/src/test/perl/PostgreSQL/Test/Cluster.pm index 1c75f127021..199e0911ada 100644 --- a/src/test/perl/PostgreSQL/Test/Cluster.pm +++ b/src/test/perl/PostgreSQL/Test/Cluster.pm @@ -731,8 +731,14 @@ sub init # 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{no_wal_compression}) + { + print $conf "wal_compression_threshold = " . (1024 * 1024 * 1024) . "\n"; + + # Streams ignore the threshold on purpose -- reaching small records is + # what they are for -- so they have to be turned off separately. + print $conf "wal_compression_streams = 0\n"; + } if ($params{allows_streaming}) { diff --git a/src/test/recovery/t/043_no_contrecord_switch.pl b/src/test/recovery/t/043_no_contrecord_switch.pl index 6fab73a064a..de219160222 100644 --- a/src/test/recovery/t/043_no_contrecord_switch.pl +++ b/src/test/recovery/t/043_no_contrecord_switch.pl @@ -53,7 +53,10 @@ sub start_of_page } my $primary = PostgreSQL::Test::Cluster->new('primary'); -$primary->init(allows_streaming => 1, has_archiving => 1); +# This test computes the size of a record so that it overflows a WAL page, so +# the record has to reach WAL at the size asked for. +$primary->init(allows_streaming => 1, has_archiving => 1, + no_wal_compression => 1); # The configuration is chosen here to minimize the friction with # concurrent WAL activity. checkpoint_timeout avoids noise with diff --git a/src/test/recovery/t/061_wal_compression.pl b/src/test/recovery/t/061_wal_compression.pl index 584d6c06a9e..6a3bba41486 100644 --- a/src/test/recovery/t/061_wal_compression.pl +++ b/src/test/recovery/t/061_wal_compression.pl @@ -25,6 +25,15 @@ push @methods, 'pglz'; # 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. +# Render a string with its non-printable bytes visible, so that a difference +# which a terminal hides can still be read in the test output. +sub _escape +{ + my ($str) = @_; + $str =~ s/([^\x20-\x7e])/sprintf('\\x%02x', ord($1))/ge; + return $str; +} + # Emit one 256kB compressible logical message at the given threshold and # report how many bytes of WAL it consumed. sub wal_used @@ -55,10 +64,14 @@ sub test_wal_compression $primary->init(allows_streaming => 1); # Start with FPI-only compression to check its initial workspace. + # Streams are off here: this part measures what whole-record compression + # does on its own, and streams would compress both sides of that + # comparison. They are covered separately below. $primary->append_conf( 'postgresql.conf', "wal_compression = '$method'\n" - . "wal_compression_threshold = 1GB\n"); + . "wal_compression_threshold = 1GB\n" + . "wal_compression_streams = 0\n"); $primary->start; # FPI-only sessions should start with space for five images, not the @@ -219,4 +232,176 @@ foreach my $method (@methods) test_wal_compression($method); } +# A record compressed against earlier records of its stream cannot be read on +# its own, so a reader that starts partway through WAL has to rewind far enough +# to rebuild the decompressors. Check that what it then reads is what a reader +# that started at the beginning sees at the same place. +SKIP: +{ + skip 'zstd not supported by this build', 9 + unless check_pg_config('#define HAVE_LIBZSTD 1'); + + my $node = PostgreSQL::Test::Cluster->new('streams'); + $node->init(allows_streaming => 1); + $node->append_conf( + 'postgresql.conf', qq( +wal_compression = zstd +wal_compression_streams = 8 +wal_compression_threshold = 1GB +wal_keep_size = 1GB +max_wal_size = 1GB +)); + $node->start; + + $node->backup('before_streams'); + my $standby = PostgreSQL::Test::Cluster->new('stream_standby'); + $standby->init_from_backup($node, 'before_streams', has_streaming => 1); + $standby->start; + + $node->safe_psql('postgres', q{ + CREATE TABLE leases(id int PRIMARY KEY, data text); + SELECT pg_switch_wal(); + }); + my $start_lsn = $node->safe_psql('postgres', + 'SELECT pg_current_wal_insert_lsn()'); + + # Keep three writers alive while changing lease ownership. A and B take + # the first two slots. A segment switch makes those leases reclaimable; + # C takes A's old slot, then A takes B's, resetting the same compressor. + # The high threshold also checks that streams prepare their workspace + # even when independent whole-record compression is disabled. + my $writer_a = $node->background_psql('postgres'); + my $writer_b = $node->background_psql('postgres'); + my $writer_c = $node->background_psql('postgres'); + $writer_a->query_safe("INSERT INTO leases VALUES (1, repeat('alpha', 200))"); + $writer_b->query_safe("INSERT INTO leases VALUES (2, repeat('bravo', 200))"); + $node->safe_psql('postgres', 'SELECT pg_switch_wal()'); + $writer_c->query_safe("INSERT INTO leases VALUES (3, repeat('charlie', 200))"); + $writer_a->query_safe("UPDATE leases SET data = repeat('delta', 200) WHERE id = 1"); + $writer_a->query_safe("UPDATE leases SET data = repeat('echo', 200) WHERE id = 1"); + $writer_b->query_safe("UPDATE leases SET data = repeat('foxtrot', 200) WHERE id = 2"); + $writer_a->quit; + $writer_b->quit; + $writer_c->quit; + + my $bulk_start = $node->safe_psql('postgres', + 'SELECT pg_current_wal_insert_lsn()'); + + # Enough traffic to cross several stream reset boundaries. + $node->safe_psql( + 'postgres', q{ + CREATE TABLE t (id int, pad text); + INSERT INTO t SELECT g, repeat('a', 200) FROM generate_series(1, 200000) g; + UPDATE t SET pad = repeat('b', 200) WHERE id % 3 = 0; + }); + my $end_lsn = + $node->safe_psql('postgres', 'SELECT pg_current_wal_flush_lsn()'); + + my $waldir = $node->data_dir . '/pg_wal'; + my ($full, $full_err) = run_command( + [ 'pg_waldump', '--path' => $waldir, + '--start' => $start_lsn, '--end' => $end_lsn ]); + my @full = split(/\n/, $full); + ok(@full > 1000, 'the workload produced records to read'); + + # Start halfway through the bulk workload, away from the zero-filled + # segment tail left by the explicit WAL switch in the lease test. + my $mid = $node->safe_psql('postgres', + "SELECT ('$bulk_start'::pg_lsn + (('$end_lsn'::pg_lsn - '$bulk_start'::pg_lsn) / 2)::bigint)::text" + ); + my ($part, $part_err) = run_command( + [ 'pg_waldump', '--path' => $waldir, + '--start' => $mid, '--end' => $end_lsn ]); + my @part = split(/\n/, $part); + unlike($part_err, qr/error/, + 'reading from an arbitrary LSN reports no error'); + die "pg_waldump returned no records: $part_err" unless @part; + + # Whatever it starts with must appear in the full dump, and everything + # from there on must match it line for line. Trailing whitespace is + # normalised away: how a line ends is the platform's business, not this + # test's. + my ($first) = $part[0] =~ /lsn: ([0-9A-F]+\/[0-9A-F]+),/; + my ($at) = grep { $full[$_] =~ /lsn: \Q$first\E,/ } 0 .. $#full; + my @tail = defined $at ? @full[ $at .. $#full ] : (); + s/\s+\z// for @part, @tail; + + my $same = is_deeply(\@part, \@tail, + 'records read from an arbitrary LSN match a full read'); + + # On failure, say what actually differs: the strings print the same when + # they differ only in bytes a terminal does not show. + if (!$same) + { + diag(sprintf('read from %s: %d lines, tail of full read: %d lines', + $mid, scalar(@part), scalar(@tail))); + for my $i (0 .. $#part) + { + next if defined $tail[$i] and $part[$i] eq $tail[$i]; + diag(sprintf("first difference at line %d\n got: %s\n exp: %s", + $i, + _escape($part[$i]), + defined $tail[$i] ? _escape($tail[$i]) : '(missing)')); + last; + } + } + + # pg_walinspect reads from an LSN its caller picks, so it has to rewind + # the same way pg_waldump does. + SKIP: + { + skip 'pg_walinspect not installed', 2 + unless $node->check_extension('pg_walinspect'); + + $node->safe_psql('postgres', 'CREATE EXTENSION pg_walinspect'); + my $differing = $node->safe_psql( + 'postgres', qq{ + WITH f AS (SELECT start_lsn, record_type, record_length + FROM pg_get_wal_records_info('$start_lsn', '$end_lsn') + WHERE start_lsn >= '$mid'::pg_lsn), + t AS (SELECT start_lsn, record_type, record_length + FROM pg_get_wal_records_info('$mid', '$end_lsn')) + SELECT (SELECT count(*) FROM (SELECT * FROM f EXCEPT ALL SELECT * FROM t) a) + + (SELECT count(*) FROM (SELECT * FROM t EXCEPT ALL SELECT * FROM f) b) + }); + is($differing, '0', + 'pg_walinspect from an arbitrary LSN matches a full read'); + + # emit_wal() is used to reach exact WAL positions, so its message + # must remain uncompressed even when streams are enabled. + my $helper_start = $node->safe_psql('postgres', + 'SELECT pg_current_wal_insert_lsn()'); + $node->emit_wal(8192); + my $helper_end = $node->safe_psql('postgres', + 'SELECT pg_current_wal_flush_lsn()'); + is($node->safe_psql('postgres', qq{ + SELECT count(*) = 1 AND bool_and(physical_length = record_length) + FROM pg_get_wal_records_info('$helper_start', '$helper_end') + WHERE resource_manager = 'LogicalMessage'; + }), 't', 'emit_wal bypasses stream compression'); + } + + # Compare the complete ordered contents, not just the number of rows. + my $digest_sql = + q{SELECT md5(string_agg(id::text || ':' || pad, ',' ORDER BY id)) FROM t}; + my $expected = $node->safe_psql('postgres', $digest_sql); + my $lease_sql = q{SELECT id, data FROM leases ORDER BY id}; + my $expected_leases = $node->safe_psql('postgres', $lease_sql); + $node->wait_for_replay_catchup($standby); + is($standby->safe_psql('postgres', $digest_sql), $expected, + 'streamed WAL reproduces the table on the standby'); + is($standby->safe_psql('postgres', $lease_sql), $expected_leases, + 'lease changes replay correctly on the standby'); + + $node->stop('immediate'); + $node->start; + is($node->safe_psql('postgres', $digest_sql), $expected, + 'crash recovery reproduces the table from streamed WAL'); + is($node->safe_psql('postgres', $lease_sql), $expected_leases, + 'lease changes replay correctly during crash recovery'); + + $node->stop; + $standby->stop; +} + done_testing(); diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index f3db609f1f3..731e7b23baf 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -3379,6 +3379,8 @@ VolatileFunctionStatus Vsrt WAIT_ORDER WALAvailability +WALCompressionSlot +WALCompressionSlotPadded WALInsertLock WALInsertLockPadded WALOpenSegment -- That's all, folks. May the source be with you.