From c0c5518364e1bf0ef1ad9a28f479a860c7244cfe Mon Sep 17 00:00:00 2001 From: Andrey Borodin Date: Sun, 26 Jul 2026 18:53:56 +0500 Subject: [PATCH v8 3/3] WIP: compress WAL records against earlier records in a stream Adds wal_compression_streams: a pool of leased compression streams, so a record can be compressed against the records that preceded it in the same stream. That is what reaches the small records, which per-record compression cannot touch; pgbench emits 29% less WAL per transaction. Streams start over at fixed 4MB boundaries, and a record that would continue one past a boundary is refused its reserved position and built again, so every stream begins afresh at or after a boundary. XLogBeginReadStreamed() uses that to read from an arbitrary LSN: it rewinds to the boundary below and reads forward to rebuild the decompressors. Still WIP: only pg_waldump has been converted to it, and a replication slot has to keep one boundary more WAL than it needs for itself. --- contrib/pg_walinspect/pg_walinspect.c | 8 +- src/backend/access/transam/xlog.c | 261 +++++++++++++++- src/backend/access/transam/xloginsert.c | 275 ++++++++++++++++ src/backend/access/transam/xlogreader.c | 293 ++++++++++++++++-- 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 | 30 ++ src/include/access/xlogrecord.h | 35 ++- src/include/storage/lwlocklist.h | 1 + src/test/perl/PostgreSQL/Test/Cluster.pm | 10 +- src/test/recovery/Makefile | 1 + src/test/recovery/meson.build | 1 + .../recovery/t/043_no_contrecord_switch.pl | 5 +- ..._compression.pl => 055_wal_compression.pl} | 91 +++++- 23 files changed, 1044 insertions(+), 58 deletions(-) rename src/test/recovery/t/{052_wal_compression.pl => 055_wal_compression.pl} (59%) diff --git a/contrib/pg_walinspect/pg_walinspect.c b/contrib/pg_walinspect/pg_walinspect.c index a172f9e2b40..e31eac9e620 100644 --- a/contrib/pg_walinspect/pg_walinspect.c +++ b/contrib/pg_walinspect/pg_walinspect.c @@ -126,8 +126,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/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index eb1d1fddb77..b33e1bf6431 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -143,6 +143,7 @@ int max_slot_wal_keep_size_mb = -1; int wal_decode_buffer_size = 512 * 1024; bool track_wal_io_timing = false; int wal_compression_threshold = 512; +int wal_compression_streams = 0; #ifdef WAL_DEBUG bool XLOG_DEBUG = false; @@ -575,6 +576,43 @@ typedef enum static XLogCtlData *XLogCtl = NULL; +/* + * State of one WAL compression stream. + * + * The compressor itself cannot live here: it is a libzstd object full of + * pointers, and shared memory is not mapped at the same address in every + * process under EXEC_BACKEND. So a backend keeps its own compressor and + * leases the stream: while it owns the slot it may compress against what it + * put there before, and a backend taking the slot over has to start the + * stream afresh, which it announces with XLR_STREAM_RESET. + */ +typedef struct WALCompressionSlot +{ + LWLock lock; /* held across compress + insert */ + int owner; /* ProcNumber holding the lease, or -1 */ + uint64 generation; /* bumped whenever the stream restarts */ + XLogRecPtr redo; /* RedoRecPtr the stream started under */ + 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; + +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; @@ -735,8 +773,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); @@ -768,6 +807,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 @@ -811,8 +958,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; @@ -870,6 +1018,24 @@ 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)) @@ -897,6 +1063,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 && @@ -916,8 +1097,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; @@ -952,8 +1145,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; } @@ -1162,9 +1356,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; @@ -1189,6 +1383,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; @@ -1207,6 +1413,8 @@ ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos, XLogRecPtr *EndPos, Assert(XLogRecPtrToBytePos(*StartPos) == startbytepos); Assert(XLogRecPtrToBytePos(*EndPos) == endbytepos); Assert(XLogRecPtrToBytePos(*PrevPtr) == prevbytepos); + + return true; } /* @@ -5340,6 +5548,11 @@ XLOGShmemRequest(void *arg) /* WAL insertion locks, plus alignment */ size = add_size(size, mul_size(sizeof(WALInsertLockPadded), NUM_XLOGINSERT_LOCKS + 1)); + + /* WAL compression stream slots, plus alignment slack */ + if (wal_compression_streams > 0) + size = add_size(size, mul_size(sizeof(WALCompressionSlotPadded), + wal_compression_streams + 1)); /* xlblocks array */ size = add_size(size, mul_size(sizeof(pg_atomic_uint64), XLOGbuffers)); /* extra alignment padding for XLOG I/O buffers */ @@ -5423,6 +5636,26 @@ XLOGShmemInit(void *arg) WALInsertLocks[i].l.lastImportantAt = InvalidXLogRecPtr; } + /* WAL compression streams, likewise aligned to their padded size */ + if (wal_compression_streams > 0) + { + allocptr += sizeof(WALCompressionSlotPadded) - + ((uintptr_t) allocptr) % sizeof(WALCompressionSlotPadded); + WALCompressionSlots = (WALCompressionSlotPadded *) allocptr; + allocptr += sizeof(WALCompressionSlotPadded) * wal_compression_streams; + + for (i = 0; i < wal_compression_streams; i++) + { + LWLockInitialize(&WALCompressionSlots[i].s.lock, + LWTRANCHE_WAL_COMPRESSION_STREAM); + WALCompressionSlots[i].s.owner = -1; + WALCompressionSlots[i].s.generation = 0; + WALCompressionSlots[i].s.redo = InvalidXLogRecPtr; + WALCompressionSlots[i].s.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 @@ -8525,6 +8758,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 a5a3a1422cb..0af8f01974e 100644 --- a/src/backend/access/transam/xloginsert.c +++ b/src/backend/access/transam/xloginsert.c @@ -171,6 +171,29 @@ static int compression_buf_offset; /* fill level for FPI packing */ static char *compressed_data = NULL; #endif +#ifdef USE_ZSTD +/* + * One compressor per stream slot we currently hold a lease on. These stay in + * the backend because a zstd context cannot live in shared memory; a backend + * taking a slot over therefore has to restart the stream. + */ +static ZSTD_CCtx **stream_cctx = NULL; + +/* + * 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. */ @@ -515,6 +538,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) @@ -554,6 +614,7 @@ XLogCompressRdt(XLogRecData *rdt) compressed_header->record_header = *src_header; compressed_header->decompressed_length = flat_len; + compressed_header->stream = XLR_NO_STREAM; orig_len = src_header->xl_tot_len - SizeOfXLogRecord; @@ -628,6 +689,139 @@ 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[slot] == NULL) + { + stream_cctx[slot] = ZSTD_createCCtx(); + if (stream_cctx[slot] == NULL) + return NULL; + ZSTD_CCtx_setParameter(stream_cctx[slot], ZSTD_c_compressionLevel, + ZSTD_CLEVEL_DEFAULT); + restart = true; + } + cctx = stream_cctx[slot]; + + if (restart) + ZSTD_CCtx_reset(cctx, ZSTD_reset_session_only); + + for (const XLogRecData *r = rdt; r != NULL; r = r->next) + 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) @@ -678,6 +872,20 @@ 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 && + stream_cctx != NULL && + XLogRecordJoinsStream(rmid, info)); +#endif + /* XLogBeginInsert() must have been called. */ if (!begininsert_called) elog(ERROR, "XLogBeginInsert was not called"); @@ -735,6 +943,67 @@ 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); + 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 +1127,12 @@ AllocCompressionBuffers(void) compressed_data = MemoryContextAlloc(xloginsert_cxt, WAL_COMPRESSION_BUFSIZE); #endif +#ifdef USE_ZSTD + if (wal_compression_streams > 0) + stream_cctx = MemoryContextAllocZero(xloginsert_cxt, + sizeof(ZSTD_CCtx *) * + wal_compression_streams); +#endif } /* diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c index 78227d9547a..b895d6ca439 100644 --- a/src/backend/access/transam/xlogreader.c +++ b/src/backend/access/transam/xlogreader.c @@ -56,6 +56,8 @@ 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 XLogRecPtr XLogFindRecordStart(XLogReaderState *state, XLogRecPtr RecPtr, + char **errormsg); static XLogRecord *XLogDecompressRecordIfNeeded(XLogReaderState *state, XLogRecord *record, XLogRecPtr recptr); @@ -181,6 +183,16 @@ XLogReaderFree(XLogReaderState *state) #endif if (state->decompression_buffer) pfree(state->decompression_buffer); +#ifdef USE_ZSTD + if (state->stream_dctx) + { + for (int i = 0; i < XLR_MAX_STREAMS; i++) + if (state->stream_dctx[i]) + ZSTD_freeDCtx((ZSTD_DCtx *) state->stream_dctx[i]); + pfree(state->stream_dctx); + pfree(state->stream_ready); + } +#endif pfree(state->readBuf); pfree(state); } @@ -251,6 +263,88 @@ 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)); + + 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; } /* @@ -997,7 +1091,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; @@ -1026,6 +1139,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) { @@ -1482,27 +1610,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; @@ -1589,18 +1708,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); @@ -1619,6 +1727,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 @@ -1840,7 +1991,91 @@ XLogDecompressRecordIfNeeded(XLogReaderState *state, dst_h->xl_tot_len = src->decompressed_length; dst = (char *) &dst_h[1]; +#ifdef USE_ZSTD + if (src->stream != XLR_NO_STREAM) + { + ZSTD_DCtx *dctx; + ZSTD_inBuffer in; + ZSTD_outBuffer out; + + if (src->method != XLR_COMPRESS_ZSTD) + { + report_invalid_record(state, + "streamed record at %X/%08X uses an unexpected compression method", + LSN_FORMAT_ARGS(recptr)); + return NULL; + } + + if (state->stream_dctx == NULL) + { + state->stream_dctx = palloc0(sizeof(void *) * XLR_MAX_STREAMS); + state->stream_ready = palloc0(sizeof(bool) * XLR_MAX_STREAMS); + } + + /* + * 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 5f3b065b894..3220f0c393d 100644 --- a/src/backend/access/transam/xlogrecovery.c +++ b/src/backend/access/transam/xlogrecovery.c @@ -1484,7 +1484,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 8b429cb51d7..7b4dbd6addb 100644 --- a/src/backend/postmaster/walsummarizer.c +++ b/src/backend/postmaster/walsummarizer.c @@ -964,12 +964,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 c30d40a8641..d0cc8a3d6ec 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -628,7 +628,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)); @@ -2032,7 +2032,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(); @@ -2135,7 +2136,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 35ebc7e61c8..a7e014069e2 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -1576,9 +1576,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 @@ -1595,6 +1592,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 1016502d042..088c9bac3ba 100644 --- a/src/backend/utils/activity/wait_event_names.txt +++ b/src/backend/utils/activity/wait_event_names.txt @@ -417,6 +417,7 @@ XactSLRU "Waiting to access the transaction status SLRU cache." ParallelVacuumDSA "Waiting for parallel vacuum dynamic shared memory allocation." AioUringCompletion "Waiting for another process to complete IO via io_uring." ShmemIndex "Waiting to find or allocate space in shared memory." +WALCompressionStream "Waiting to compress a WAL record into a shared compression stream." # No "ABI_compatibility" region here as WaitEventLWLock has its own C code. diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat index 28efd8f6e95..6c1e21254b3 100644 --- a/src/backend/utils/misc/guc_parameters.dat +++ b/src/backend/utils/misc/guc_parameters.dat @@ -3467,6 +3467,18 @@ assign_hook => 'assign_wal_compression', }, +{ name => 'wal_compression_streams', type => 'int', context => 'PGC_POSTMASTER', group => 'WAL_SETTINGS', + short_desc => 'Number of WAL compression streams shared by all backends.', + long_desc => 'Zero compresses every record on its own. A positive value lets a record ' + . 'be compressed against the records that preceded it in the same stream, ' + . 'which compresses small records much better, at the cost of one ' + . 'decompression context per stream in every process that reads WAL records.', + variable => 'wal_compression_streams', + boot_val => '0', + min => '0', + max => '64', +}, + { name => 'wal_compression_threshold', type => 'int', context => 'PGC_SUSET', group => 'WAL_SETTINGS', short_desc => 'Minimum WAL record length to engage whole-record compression.', long_desc => 'Records at least this large are compressed as a single unit instead of ' diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 406b8fc52ca..701d7e2ca7b 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -266,6 +266,10 @@ # off, pglz (or "on"), lz4, or zstd #wal_compression_threshold = 512 # min 32, smallest record compressed as a whole; # whole-record compression needs lz4 or zstd +#wal_compression_streams = 0 # 0 disables compressing a record against + # earlier ones; a positive value is the + # number of shared compression streams + # (change requires restart) #wal_init_zero = on # zero-fill new WAL files #wal_recycle = on # recycle WAL files #wal_buffers = -1 # min 32kB, -1 sets based on shared_buffers diff --git a/src/bin/pg_rewind/parsexlog.c b/src/bin/pg_rewind/parsexlog.c index 023e23b063c..2a59b5f726a 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) { @@ -205,7 +210,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) @@ -251,6 +262,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 cf760d8b236..5275c88594f 100644 --- a/src/bin/pg_waldump/pg_waldump.c +++ b/src/bin/pg_waldump/pg_waldump.c @@ -1415,8 +1415,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 7fda7eb1f23..26dbb51d184 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -59,6 +59,7 @@ extern PGDLLIMPORT bool track_wal_io_timing; extern PGDLLIMPORT int wal_decode_buffer_size; extern PGDLLIMPORT int data_checksums; extern PGDLLIMPORT int wal_compression_threshold; +extern PGDLLIMPORT int wal_compression_streams; extern PGDLLIMPORT int CheckPointSegments; @@ -211,6 +212,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 5c869d78be7..e213e014f09 100644 --- a/src/include/access/xlogreader.h +++ b/src/include/access/xlogreader.h @@ -235,6 +235,27 @@ struct XLogReaderState * this header is included where zstd.h is not. */ void *fpi_dctx; + + /* + * 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; + + /* + * Per stream: has a record that starts the stream over been seen yet? + * Until one has, the stream's earlier records cannot be decompressed, + * because the state they were compressed against is not here. + */ + bool *stream_ready; XLogRecPtr NextRecPtr; /* end+1 of last record decoded */ XLogRecPtr PrevRecPtr; /* start of previous record decoded */ @@ -262,6 +283,13 @@ struct XLogReaderState char *decompression_buffer; uint32 decompression_buffer_size; + /* + * One decompression context per compression stream met so far, indexed by + * the stream id in the record. Void because the type belongs to whichever + * compression library the build has. + */ + void **stream_dctx; + /* * Queue of records that have been decoded. This is a linked list that * usually consists of consecutive records in decode_buffer, but may also @@ -353,6 +381,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 d7eb648bd27..af1a8c2f932 100644 --- a/src/include/storage/lwlocklist.h +++ b/src/include/storage/lwlocklist.h @@ -140,3 +140,4 @@ PG_LWLOCKTRANCHE(XACT_SLRU, XactSLRU) PG_LWLOCKTRANCHE(PARALLEL_VACUUM_DSA, ParallelVacuumDSA) PG_LWLOCKTRANCHE(AIO_URING_COMPLETION, AioUringCompletion) PG_LWLOCKTRANCHE(SHMEM_INDEX, ShmemIndex) +PG_LWLOCKTRANCHE(WAL_COMPRESSION_STREAM, WALCompressionStream) diff --git a/src/test/perl/PostgreSQL/Test/Cluster.pm b/src/test/perl/PostgreSQL/Test/Cluster.pm index 96aaa88f1ce..f9880f5c96c 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/Makefile b/src/test/recovery/Makefile index d41aaaf8ae1..b73bdb7fe1d 100644 --- a/src/test/recovery/Makefile +++ b/src/test/recovery/Makefile @@ -11,6 +11,7 @@ EXTRA_INSTALL=contrib/pg_prewarm \ contrib/pg_stat_statements \ + contrib/pg_walinspect \ contrib/test_decoding \ src/test/modules/injection_points diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build index ad0d85f4189..9c1200d0d65 100644 --- a/src/test/recovery/meson.build +++ b/src/test/recovery/meson.build @@ -63,6 +63,7 @@ tests += { 't/052_checkpoint_segment_missing.pl', 't/053_standby_login_event_trigger.pl', 't/054_unlogged_sequence_promotion.pl', + 't/055_wal_compression.pl', ], }, } 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/052_wal_compression.pl b/src/test/recovery/t/055_wal_compression.pl similarity index 59% rename from src/test/recovery/t/052_wal_compression.pl rename to src/test/recovery/t/055_wal_compression.pl index 001772daf23..6c52562082c 100644 --- a/src/test/recovery/t/052_wal_compression.pl +++ b/src/test/recovery/t/055_wal_compression.pl @@ -55,10 +55,14 @@ sub test_wal_compression $primary->init(allows_streaming => 1); # Use the minimum threshold so virtually every record gets compressed. + # 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 = 32\n"); + . "wal_compression_threshold = 32\n" + . "wal_compression_streams = 0\n"); $primary->start; my $backup_name = "backup_$method"; @@ -149,4 +153,89 @@ 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', 3 + unless check_pg_config('#define HAVE_LIBZSTD 1'); + + my $node = PostgreSQL::Test::Cluster->new('streams'); + $node->init; + $node->append_conf( + 'postgresql.conf', qq( +wal_compression = zstd +wal_compression_streams = 8 +wal_compression_threshold = 64 +wal_keep_size = 1GB +max_wal_size = 1GB +)); + $node->start; + + my $start_lsn = $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 in, at an LSN that is nobody's record boundary. + my $mid = $node->safe_psql('postgres', + "SELECT ('$start_lsn'::pg_lsn + (('$end_lsn'::pg_lsn - '$start_lsn'::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'); + + # Whatever it starts with must appear in the full dump, and everything + # from there on must match it line for line. + 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 ] : (); + is_deeply(\@part, \@tail, + 'records read from an arbitrary LSN match a full read'); + + # 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', 1 + 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'); + } + + $node->stop; +} + done_testing(); -- 2.50.1 (Apple Git-155)