On Sun, Sep 6, 2026 at 8:49 AM Andrey Borodin <[email protected]> wrote: > On Sat, Sep 5, 2026, Sehrope Sarkuni wrote: > > I think this needs an opt-in from the client. Something like a > > START_REPLICATION option or at least a protocol guard. > > I agree. v2 adds a SKIP_WAL_PADDING option to physical > START_REPLICATION. Existing clients continue to receive ordinary WALData > messages, and built-in clients request the new format only from PG20 servers. > > The receiver no longer truncates and extends the segment. A segment newly > created by walreceiver is already zero-filled according to wal_init_zero. For > an existing or recycled segment, walreceiver writes the zeros locally with > the normal wait-event and pg_stat_io accounting. This also removes the > short-file crash window. New uncompressed pg_receivewal output is already > pre-padded. Resumed, compressed, and tar output generate the zeros locally. > > I also added the missing Meson registration and used > pg_memory_is_all_zeros(). As a compatibility check, I connected a pre-patch > PG20 pg_receivewal to the patched server and compared the resulting switched > segment byte for byte.
Looks good. v2 addresses everything from my first pass and it all looks right to me. Tested on 798bdcae89d. Builds clean, recovery suite passes, and a pre-patch pg_receivewal against the patched server gets plain WALData and byte-identical segments. With the patch on a standby the wire traffic drops from 16 MiB to ~125 kB. I'm not particularly familiar with the replication command grammar, so other eyes on the START_REPLICATION option parsing would be good, but it looks fine and behaves correctly in all the tests. A few small things: 1. The test no longer proves the 'z' path was taken. An old client produces the same bytes, so the compare passes either way. The attached patch resets pg_stat_io on the standby before the switch and checks the walreceiver wrote less than wal_segment_size: 248 kB here vs exactly 16 MB unpatched. It restarts the standby first to flush the counters, since the walreceiver only reports them on its status timer. Byte counts also avoid the Windows skip a sparse-file check would need. 2. The pg_stat_activity check runs right after $standby->start, before the walreceiver has necessarily connected. That should use poll_query_until(...). 3. recvFileIsNew is reset in XLogWalRcvClose but not at the end-of-timeline close in WalReceiverMain. See attached for a patch atop your v2 to address 1-3. The 0001 is your v2 patch rebased and the additions are in 0002. Nothing special for the rebase either (the patch is identical). CI run for the modified tests shows it ran on Windows and passed there as well: https://github.com/sehrope/postgres/actions/runs/34376945257/job/102552798366#step:14:94 Regards, -- Sehrope Sarkuni Founder & CEO | JackDB, Inc. | https://www.jackdb.com/
From 57c56577f7050b48373ed5e2491278c8b991dca4 Mon Sep 17 00:00:00 2001 From: Andrey Borodin <[email protected]> Date: Thu, 6 Aug 2026 12:03:11 +0500 Subject: [PATCH v3 1/2] Avoid streaming zero-filled WAL switch padding Forced WAL switches can leave almost a whole segment as zero padding. Sending those bytes wastes network bandwidth and can delay synchronous replication, particularly on low-traffic servers using archive_timeout. Represent such padding with a new physical replication message. Require clients to opt in with the SKIP_WAL_PADDING option, so existing physical WAL consumers continue to receive ordinary WALData messages. Built-in clients request the option only from servers that support it. On a receiving server, reuse the zeros established while initializing a new segment. Write the omitted range locally for an existing or recycled segment, with the usual wait-event and I/O accounting. New uncompressed frontend WAL files are already pre-padded; resumed, compressed, and tar output generate the zeros locally. Test that a standby reconstructs a switched WAL segment byte for byte. Discussion: https://postgr.es/m/[email protected] --- doc/src/sgml/protocol.sgml | 77 ++++++++++- src/backend/access/transam/xlog.c | 25 +++- .../libpqwalreceiver/libpqwalreceiver.c | 5 + src/backend/replication/repl_gram.y | 11 +- src/backend/replication/walreceiver.c | 120 ++++++++++++++++-- src/backend/replication/walsender.c | 51 ++++++++ src/bin/pg_basebackup/receivelog.c | 94 ++++++++++++++ src/bin/pg_basebackup/walmethods.c | 73 +++++++++++ src/bin/pg_basebackup/walmethods.h | 3 + src/include/access/xlog.h | 2 +- src/include/libpq/protocol.h | 1 + src/test/recovery/meson.build | 1 + src/test/recovery/t/057_stream_wal_zeros.pl | 65 ++++++++++ 13 files changed, 508 insertions(+), 20 deletions(-) create mode 100644 src/test/recovery/t/057_stream_wal_zeros.pl diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml index 89ac680efd5..80bef9f54ac 100644 --- a/doc/src/sgml/protocol.sgml +++ b/doc/src/sgml/protocol.sgml @@ -2606,7 +2606,7 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;" </varlistentry> <varlistentry id="protocol-replication-start-replication"> - <term><literal>START_REPLICATION</literal> [ <literal>SLOT</literal> <replaceable class="parameter">slot_name</replaceable> ] [ <literal>PHYSICAL</literal> ] <replaceable class="parameter">XXX/XXX</replaceable> [ <literal>TIMELINE</literal> <replaceable class="parameter">tli</replaceable> ] + <term><literal>START_REPLICATION</literal> [ <literal>SLOT</literal> <replaceable class="parameter">slot_name</replaceable> ] [ <literal>PHYSICAL</literal> ] <replaceable class="parameter">XXX/XXX</replaceable> [ <literal>TIMELINE</literal> <replaceable class="parameter">tli</replaceable> ] [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] <indexterm><primary>START_REPLICATION</primary></indexterm> </term> <listitem> @@ -2621,6 +2621,24 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;" message, and then starts to stream WAL to the frontend. </para> + <para> + The following option is supported: + </para> + + <variablelist> + <varlistentry> + <term><literal>SKIP_WAL_PADDING</literal> [ <replaceable class="parameter">boolean</replaceable> ]</term> + <listitem> + <para> + Requests that zero-filled padding after a WAL switch be represented + by a <literal>Zero WAL data</literal> message instead of being sent + as ordinary WAL data. A client must not request this option unless + it supports that message type. + </para> + </listitem> + </varlistentry> + </variablelist> + <para> If a slot's name is provided via <replaceable class="parameter">slot_name</replaceable>, it will be updated @@ -2725,6 +2743,63 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;" </listitem> </varlistentry> + <varlistentry id="protocol-replication-zero-wal-data"> + <term>Zero WAL data (B)</term> + <listitem> + <variablelist> + <varlistentry> + <term>Byte1('z')</term> + <listitem> + <para> + Identifies zero-filled padding after a WAL switch record. The + client may assume that the remainder of this WAL segment is also + zero-filled. This message is sent only when the client requested + the <literal>SKIP_WAL_PADDING</literal> option of + <literal>START_REPLICATION</literal>. + </para> + </listitem> + </varlistentry> + + <varlistentry> + <term>Int64</term> + <listitem> + <para> + The starting point of the zero-filled WAL data. + </para> + </listitem> + </varlistentry> + + <varlistentry> + <term>Int64</term> + <listitem> + <para> + The current end of WAL on the server. + </para> + </listitem> + </varlistentry> + + <varlistentry> + <term>Int64</term> + <listitem> + <para> + The server's system clock at the time of transmission, as + microseconds since midnight on 2000-01-01. + </para> + </listitem> + </varlistentry> + + <varlistentry> + <term>Int64</term> + <listitem> + <para> + The number of zero bytes in this section. + </para> + </listitem> + </varlistentry> + </variablelist> + </listitem> + </varlistentry> + <varlistentry id="protocol-replication-primary-keepalive-message"> <term>Primary keepalive message (B)</term> <listitem> diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 2e3f177100b..33519b4817e 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -2401,7 +2401,7 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible) openLogTLI = tli; /* create/use new log file */ - openLogFile = XLogFileInit(openLogSegNo, tli); + openLogFile = XLogFileInit(openLogSegNo, tli, NULL); ReserveExternalFD(); } @@ -3240,6 +3240,10 @@ XLogNeedsFlush(XLogRecPtr record) * * *added: on return, true if this call raised the number of extant segments. * + * *target_created: if not NULL, set to true if this call created the requested + * segment rather than opening one that already existed. In a race, the + * initialized file might be installed as a later segment instead. + * * path: on return, this char[MAXPGPATH] has the path to the logsegno file. * * Returns -1 or FD of opened file. A -1 here is not an error; a caller @@ -3248,7 +3252,7 @@ XLogNeedsFlush(XLogRecPtr record) */ static int XLogFileInitInternal(XLogSegNo logsegno, TimeLineID logtli, - bool *added, char *path) + bool *added, bool *target_created, char *path) { char tmppath[MAXPGPATH]; XLogSegNo installed_segno; @@ -3266,6 +3270,8 @@ XLogFileInitInternal(XLogSegNo logsegno, TimeLineID logtli, * Try to use existent file (checkpoint maker may have created it already) */ *added = false; + if (target_created) + *target_created = false; fd = BasicOpenFile(path, O_RDWR | PG_BINARY | O_CLOEXEC | get_sync_bit(wal_sync_method)); if (fd < 0) @@ -3406,6 +3412,8 @@ XLogFileInitInternal(XLogSegNo logsegno, TimeLineID logtli, logtli)) { *added = true; + if (target_created && installed_segno == logsegno) + *target_created = true; elog(DEBUG2, "done creating and filling new WAL file"); } else @@ -3426,6 +3434,7 @@ XLogFileInitInternal(XLogSegNo logsegno, TimeLineID logtli, * Create a new XLOG file segment, or open a pre-existing one. * * logsegno: identify segment to be created/opened. + * created: if not NULL, set to true if this created the target segment. * * Returns FD of opened file. * @@ -3435,15 +3444,17 @@ XLogFileInitInternal(XLogSegNo logsegno, TimeLineID logtli, * in a critical section. */ int -XLogFileInit(XLogSegNo logsegno, TimeLineID logtli) +XLogFileInit(XLogSegNo logsegno, TimeLineID logtli, bool *created) { bool ignore_added; + bool ignore_created; char path[MAXPGPATH]; int fd; Assert(logtli != 0); - fd = XLogFileInitInternal(logsegno, logtli, &ignore_added, path); + fd = XLogFileInitInternal(logsegno, logtli, &ignore_added, + created ? created : &ignore_created, path); if (fd >= 0) return fd; @@ -3761,7 +3772,7 @@ PreallocXlogFiles(XLogRecPtr endptr, TimeLineID tli) if (offset >= (uint32) (0.75 * wal_segment_size)) { _logSegNo++; - lf = XLogFileInitInternal(_logSegNo, tli, &added, path); + lf = XLogFileInitInternal(_logSegNo, tli, &added, NULL, path); if (lf >= 0) close(lf); if (added) @@ -5585,7 +5596,7 @@ BootStrapXLOG(uint32 data_checksum_version) /* Create first XLOG segment file */ openLogTLI = BootstrapTimeLineID; - openLogFile = XLogFileInit(1, BootstrapTimeLineID); + openLogFile = XLogFileInit(1, BootstrapTimeLineID, NULL); /* * We needn't bother with Reserve/ReleaseExternalFD here, since we'll @@ -5705,7 +5716,7 @@ XLogInitNewTimeline(TimeLineID endTLI, XLogRecPtr endOfLog, TimeLineID newTLI) */ int fd; - fd = XLogFileInit(startLogSegNo, newTLI); + fd = XLogFileInit(startLogSegNo, newTLI, NULL); if (close(fd) != 0) { diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c index 029990d9fce..58e2e08a8f8 100644 --- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c +++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c @@ -644,9 +644,14 @@ libpqrcv_startstreaming(WalReceiverConn *conn, appendStringInfoChar(&cmd, ')'); } else + { appendStringInfo(&cmd, " TIMELINE %u", options->proto.physical.startpointTLI); + if (PQserverVersion(conn->streamConn) >= 200000) + appendStringInfoString(&cmd, " (SKIP_WAL_PADDING)"); + } + /* Start streaming. */ res = libpqsrv_exec(conn->streamConn, cmd.data, diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y index aa8a96a3a60..e48c963d71f 100644 --- a/src/backend/replication/repl_gram.y +++ b/src/backend/replication/repl_gram.y @@ -86,7 +86,7 @@ create_replication_slot drop_replication_slot alter_replication_slot identify_system read_replication_slot timeline_history show upload_manifest -%type <list> generic_option_list +%type <list> generic_option_list opt_physical_options %type <defelt> generic_option %type <uintval> opt_timeline %type <list> plugin_options plugin_opt_list @@ -280,9 +280,10 @@ alter_replication_slot: /* * START_REPLICATION [SLOT slot] [PHYSICAL] %X/%08X [TIMELINE %u] + * [( option [, ...] )] */ start_replication: - K_START_REPLICATION opt_slot opt_physical RECPTR opt_timeline + K_START_REPLICATION opt_slot opt_physical RECPTR opt_timeline opt_physical_options { StartReplicationCmd *cmd; @@ -291,10 +292,16 @@ start_replication: cmd->slotname = $2; cmd->startpoint = $4; cmd->timeline = $5; + cmd->options = $6; $$ = (Node *) cmd; } ; +opt_physical_options: + '(' generic_option_list ')' { $$ = $2; } + | /* EMPTY */ { $$ = NIL; } + ; + /* START_REPLICATION SLOT slot LOGICAL %X/%08X options */ start_logical_replication: K_START_REPLICATION K_SLOT IDENT K_LOGICAL RECPTR plugin_options diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c index b93e699ba4b..f96180c9251 100644 --- a/src/backend/replication/walreceiver.c +++ b/src/backend/replication/walreceiver.c @@ -60,6 +60,7 @@ #include "access/xlogrecovery.h" #include "access/xlogwait.h" #include "catalog/pg_authid.h" +#include "common/file_utils.h" #include "funcapi.h" #include "libpq/pqformat.h" #include "libpq/pqsignal.h" @@ -104,6 +105,7 @@ WalReceiverFunctionsType *WalReceiverFunctions = NULL; static int recvFile = -1; static TimeLineID recvFileTLI = 0; static XLogSegNo recvSegNo = 0; +static bool recvFileIsNew = false; /* * LogstreamResult indicates the byte positions that we have already @@ -142,6 +144,9 @@ static void XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len, TimeLineID tli); static void XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr, TimeLineID tli); +static void XLogWalRcvWriteZeros(Size nbytes, XLogRecPtr recptr, + TimeLineID tli); +static void XLogWalRcvAdvanceWrite(XLogRecPtr recptr); static void XLogWalRcvFlush(bool dying, TimeLineID tli); static void XLogWalRcvClose(XLogRecPtr recptr, TimeLineID tli); static void XLogWalRcvSendReply(bool force, bool requestReply, bool checkApply); @@ -952,6 +957,34 @@ XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len, TimeLineID tli) XLogWalRcvWrite(buf, len, dataStart, tli); break; } + case PqReplMsg_WALDataZeros: + { + StringInfoData incoming_message; + uint64 nbytes; + + hdrlen = sizeof(int64) + sizeof(int64) + sizeof(int64) + + sizeof(int64); + if (len != hdrlen) + ereport(ERROR, + (errcode(ERRCODE_PROTOCOL_VIOLATION), + errmsg_internal("invalid zero WAL message received from primary"))); + + initReadOnlyStringInfo(&incoming_message, buf, hdrlen); + dataStart = pq_getmsgint64(&incoming_message); + walEnd = pq_getmsgint64(&incoming_message); + sendTime = pq_getmsgint64(&incoming_message); + nbytes = pq_getmsgint64(&incoming_message); + + if (nbytes == 0 || nbytes > wal_segment_size || + dataStart != LogstreamResult.Write) + ereport(ERROR, + (errcode(ERRCODE_PROTOCOL_VIOLATION), + errmsg_internal("invalid zero WAL range received from primary"))); + + ProcessWalSndrMessage(walEnd, sendTime); + XLogWalRcvWriteZeros(nbytes, dataStart, tli); + break; + } case PqReplMsg_Keepalive: { StringInfoData incoming_message; @@ -985,6 +1018,81 @@ XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len, TimeLineID tli) } } +/* + * Reconstruct a run of zeros omitted from the replication stream. + * + * These messages represent the padding after XLOG_SWITCH, so the remainder of + * the segment is known to contain zeros. A segment created by this receiver + * is already zero-filled according to wal_init_zero, but a pre-existing or + * recycled segment might still contain old data that must be overwritten. + */ +static void +XLogWalRcvWriteZeros(Size nbytes, XLogRecPtr recptr, TimeLineID tli) +{ + int startoff; + XLogRecPtr endptr = recptr + nbytes; + + if (recvFile >= 0 && !XLByteInSeg(recptr, recvSegNo, wal_segment_size)) + XLogWalRcvClose(recptr, tli); + + if (recvFile < 0) + { + XLByteToSeg(recptr, recvSegNo, wal_segment_size); + recvFile = XLogFileInit(recvSegNo, tli, &recvFileIsNew); + recvFileTLI = tli; + } + + startoff = XLogSegmentOffset(recptr, wal_segment_size); + if (startoff + nbytes > wal_segment_size) + ereport(ERROR, + (errcode(ERRCODE_PROTOCOL_VIOLATION), + errmsg_internal("zero WAL range crosses a segment boundary"))); + + if (!recvFileIsNew) + { + ssize_t byteswritten; + instr_time start; + + start = pgstat_prepare_io_time(track_wal_io_timing); + + pgstat_report_wait_start(WAIT_EVENT_WAL_WRITE); + byteswritten = pg_pwrite_zeros(recvFile, nbytes, + (pgoff_t) startoff); + pgstat_report_wait_end(); + + if (byteswritten < 0) + { + char xlogfname[MAXFNAMELEN]; + int save_errno = errno; + + XLogFileName(xlogfname, recvFileTLI, recvSegNo, + wal_segment_size); + errno = save_errno; + ereport(PANIC, + (errcode_for_file_access(), + errmsg("could not write to WAL segment %s at offset %d, length %zu: %m", + xlogfname, startoff, nbytes))); + } + + pgstat_count_io_op_time(IOOBJECT_WAL, IOCONTEXT_NORMAL, + IOOP_WRITE, start, 1, byteswritten); + } + + XLogWalRcvAdvanceWrite(endptr); + + if (!XLByteInSeg(endptr, recvSegNo, wal_segment_size)) + XLogWalRcvClose(endptr, tli); +} + +static void +XLogWalRcvAdvanceWrite(XLogRecPtr recptr) +{ + LogstreamResult.Write = recptr; + + pg_atomic_write_membarrier_u64(&WalRcv->writtenUpto, recptr); + WaitLSNWakeup(WAIT_LSN_TYPE_STANDBY_WRITE, recptr); +} + /* * Write XLOG data to disk. */ @@ -1009,7 +1117,7 @@ XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr, TimeLineID tli) { /* Create/use new log file */ XLByteToSeg(recptr, recvSegNo, wal_segment_size); - recvFile = XLogFileInit(recvSegNo, tli); + recvFile = XLogFileInit(recvSegNo, tli, &recvFileIsNew); recvFileTLI = tli; } @@ -1064,14 +1172,7 @@ XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr, TimeLineID tli) LogstreamResult.Write = recptr; } - /* Update shared-memory status */ - pg_atomic_write_membarrier_u64(&WalRcv->writtenUpto, LogstreamResult.Write); - - /* - * Wake up processes waiting for standby write LSN to reach current write - * position. - */ - WaitLSNWakeup(WAIT_LSN_TYPE_STANDBY_WRITE, LogstreamResult.Write); + XLogWalRcvAdvanceWrite(LogstreamResult.Write); /* * Close the current segment if it's fully written up in the last cycle of @@ -1187,6 +1288,7 @@ XLogWalRcvClose(XLogRecPtr recptr, TimeLineID tli) XLogArchiveNotify(xlogfname); recvFile = -1; + recvFileIsNew = false; } /* diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index e9331de3df5..e13bf58b2a6 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -189,6 +189,9 @@ static XLogRecPtr sendTimeLineValidUpto = InvalidXLogRecPtr; */ static XLogRecPtr sentPtr = InvalidXLogRecPtr; +/* Can the physical replication client process Zero WAL data messages? */ +static bool sendZeroWALData = false; + /* Buffers for constructing outgoing messages and processing reply messages. */ static StringInfoData output_message; static StringInfoData reply_message; @@ -862,6 +865,26 @@ StartReplication(StartReplicationCmd *cmd) StringInfoData buf; XLogRecPtr FlushPtr; TimeLineID FlushTLI; + bool o_skip_wal_padding = false; + + sendZeroWALData = false; + foreach_ptr(DefElem, defel, cmd->options) + { + if (strcmp(defel->defname, "skip_wal_padding") == 0) + { + if (o_skip_wal_padding) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("duplicate option \"%s\"", defel->defname))); + sendZeroWALData = defGetBoolean(defel); + o_skip_wal_padding = true; + } + else + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("unrecognized START_REPLICATION option: \"%s\"", + defel->defname))); + } /* create xlogreader for physical replication */ xlogreader = @@ -3391,6 +3414,7 @@ XLogSendPhysical(void) XLogSegNo segno; WALReadError errinfo; Size rbytes; + Size hdrlen; /* If requested switch the WAL sender to the stopping state. */ if (got_STOPPING) @@ -3598,6 +3622,7 @@ XLogSendPhysical(void) pq_sendint64(&output_message, startptr); /* dataStart */ pq_sendint64(&output_message, SendRqstPtr); /* walEnd */ pq_sendint64(&output_message, 0); /* sendtime, filled in last */ + hdrlen = output_message.len; /* * Read the log directly into the output buffer to avoid extra memcpy @@ -3656,6 +3681,32 @@ retry: output_message.len += nbytes; output_message.data[output_message.len] = '\0'; + /* + * Avoid sending zero-filled WAL chunks. XLOG_SWITCH commonly leaves + * almost a whole segment of zeros. Messages are cut only at WAL record + * or page boundaries, where valid WAL pages have nonzero headers, so an + * entirely zero message can only contain end-of-segment padding. + * Detecting the bytes rather than remembering the switch point also + * handles a sender that starts in the middle of the padding. + */ + if (sendZeroWALData && + pg_memory_is_all_zeros(output_message.data + hdrlen, + output_message.len - hdrlen)) + { + XLogSegNo zero_segno; + XLogRecPtr segment_end; + + /* The first zero page proves that this is switch padding. */ + XLByteToSeg(sentPtr, zero_segno, wal_segment_size); + segment_end = (zero_segno + 1) * wal_segment_size; + endptr = Min(segment_end, SendRqstPtr); + WalSndCaughtUp = !sendTimeLineIsHistoric && endptr == SendRqstPtr; + + output_message.data[0] = PqReplMsg_WALDataZeros; + output_message.len = hdrlen; + pq_sendint64(&output_message, endptr - sentPtr); + } + /* * Fill the send timestamp last, so that it is taken as late as possible. */ diff --git a/src/bin/pg_basebackup/receivelog.c b/src/bin/pg_basebackup/receivelog.c index 77a2b4458b3..8c6303d3ef6 100644 --- a/src/bin/pg_basebackup/receivelog.c +++ b/src/bin/pg_basebackup/receivelog.c @@ -42,6 +42,9 @@ static bool ProcessKeepaliveMsg(PGconn *conn, StreamCtl *stream, char *copybuf, int len, XLogRecPtr blockpos, TimestampTz *last_status); static bool ProcessWALDataMsg(PGconn *conn, StreamCtl *stream, char *copybuf, int len, XLogRecPtr *blockpos); +static bool ProcessWALDataZerosMsg(PGconn *conn, StreamCtl *stream, + char *copybuf, int len, + XLogRecPtr *blockpos); static PGresult *HandleEndOfCopyStream(PGconn *conn, StreamCtl *stream, char *copybuf, XLogRecPtr blockpos, XLogRecPtr *stoppos); static bool CheckCopyStreamStop(PGconn *conn, StreamCtl *stream, XLogRecPtr blockpos); @@ -582,6 +585,8 @@ ReceiveXlogStream(PGconn *conn, StreamCtl *stream) appendPQExpBuffer(query, " %X/%08X TIMELINE %u", LSN_FORMAT_ARGS(stream->startpos), stream->timeline); + if (PQserverVersion(conn) >= 200000) + appendPQExpBufferStr(query, " (SKIP_WAL_PADDING)"); res = PQexec(conn, query->data); destroyPQExpBuffer(query); if (PQresultStatus(res) != PGRES_COPY_BOTH) @@ -844,6 +849,14 @@ HandleCopyStream(PGconn *conn, StreamCtl *stream, if (!CheckCopyStreamStop(conn, stream, blockpos)) goto error; } + else if (copybuf[0] == PqReplMsg_WALDataZeros) + { + if (!ProcessWALDataZerosMsg(conn, stream, copybuf, r, &blockpos)) + goto error; + + if (!CheckCopyStreamStop(conn, stream, blockpos)) + goto error; + } else { pg_log_error("unrecognized streaming header: \"%c\"", @@ -1174,6 +1187,87 @@ ProcessWALDataMsg(PGconn *conn, StreamCtl *stream, char *copybuf, int len, return true; } +/* Process a compact representation of a zero-filled WAL range. */ +static bool +ProcessWALDataZerosMsg(PGconn *conn, StreamCtl *stream, char *copybuf, + int len, XLogRecPtr *blockpos) +{ + uint64 bytes_left; + uint64 nbytes; + int xlogoff; + int hdr_len = 1 + 8 + 8 + 8 + 8; + + if (!still_sending) + return true; + + if (len != hdr_len) + { + pg_log_error("invalid zero WAL message size: %d", len); + return false; + } + + *blockpos = fe_recvint64(©buf[1]); + nbytes = fe_recvint64(©buf[1 + 8 + 8 + 8]); + xlogoff = XLogSegmentOffset(*blockpos, WalSegSz); + if (nbytes == 0 || nbytes > WalSegSz - xlogoff) + { + pg_log_error("invalid zero WAL range length: " UINT64_FORMAT, nbytes); + return false; + } + + if ((walfile == NULL && xlogoff != 0) || + (walfile != NULL && walfile->currpos != xlogoff)) + { + pg_log_error("got zero WAL data offset %08x, expected %08x", + xlogoff, walfile == NULL ? 0 : (int) walfile->currpos); + return false; + } + + bytes_left = nbytes; + while (bytes_left > 0) + { + size_t bytes_to_write = Min(bytes_left, WalSegSz - xlogoff); + + if (walfile == NULL && !open_walfile(stream, *blockpos)) + return false; + + if (stream->walmethod->ops->write_zeros(walfile, bytes_to_write) != + bytes_to_write) + { + pg_log_error("could not write %zu zero bytes to WAL file \"%s\": %s", + bytes_to_write, walfile->pathname, + GetLastWalMethodError(stream->walmethod)); + return false; + } + + bytes_left -= bytes_to_write; + *blockpos += bytes_to_write; + xlogoff += bytes_to_write; + + if (XLogSegmentOffset(*blockpos, WalSegSz) == 0) + { + if (!close_walfile(stream, *blockpos)) + return false; + + xlogoff = 0; + if (still_sending && + stream->stream_stop(*blockpos, stream->timeline, true)) + { + if (PQputCopyEnd(conn, NULL) <= 0 || PQflush(conn)) + { + pg_log_error("could not send copy-end packet: %s", + PQerrorMessage(conn)); + return false; + } + still_sending = false; + return true; + } + } + } + + return true; +} + /* * Handle end of the copy stream. */ diff --git a/src/bin/pg_basebackup/walmethods.c b/src/bin/pg_basebackup/walmethods.c index 3a6b3b5f45b..34cd2660d4d 100644 --- a/src/bin/pg_basebackup/walmethods.c +++ b/src/bin/pg_basebackup/walmethods.c @@ -51,6 +51,7 @@ static ssize_t dir_get_file_size(WalWriteMethod *wwmethod, static char *dir_get_file_name(WalWriteMethod *wwmethod, const char *pathname, const char *temp_suffix); static ssize_t dir_write(Walfile *f, const void *buf, size_t count); +static ssize_t dir_write_zeros(Walfile *f, size_t count); static int dir_sync(Walfile *f); static bool dir_finish(WalWriteMethod *wwmethod); static void dir_free(WalWriteMethod *wwmethod); @@ -62,6 +63,7 @@ static const WalWriteMethodOps WalDirectoryMethodOps = { .get_file_size = dir_get_file_size, .get_file_name = dir_get_file_name, .write = dir_write, + .write_zeros = dir_write_zeros, .sync = dir_sync, .finish = dir_finish, .free = dir_free @@ -84,6 +86,7 @@ typedef struct DirectoryMethodFile Walfile base; int fd; char *fullpath; + bool remainder_is_zero; char *temp_suffix; #ifdef HAVE_LIBZ gzFile gzfp; @@ -294,6 +297,7 @@ dir_open_for_write(WalWriteMethod *wwmethod, const char *pathname, f->base.pathname = pg_strdup(pathname); f->fd = fd; f->fullpath = pg_strdup(tmppath); + f->remainder_is_zero = pad_to_size != 0; if (temp_suffix) f->temp_suffix = pg_strdup(temp_suffix); @@ -381,6 +385,56 @@ dir_write(Walfile *f, const void *buf, size_t count) return r; } +static ssize_t +dir_write_zeros(Walfile *f, size_t count) +{ + DirectoryMethodFile *df = (DirectoryMethodFile *) f; + + if (f->wwmethod->compression_algorithm == PG_COMPRESSION_NONE) + { + pgoff_t newpos = f->currpos + count; + ssize_t rc; + + clear_error(f->wwmethod); + + /* A file opened after an earlier run can contain stale data here. */ + if (!df->remainder_is_zero) + { + rc = pg_pwrite_zeros(df->fd, count, f->currpos); + if (rc < 0) + { + f->wwmethod->lasterrno = errno; + return -1; + } + } + + /* On Windows, pg_pwrite_zeros() may have moved the file position. */ + if (lseek(df->fd, newpos, SEEK_SET) != newpos) + { + f->wwmethod->lasterrno = errno; + return -1; + } + + f->currpos = newpos; + return count; + } + else + { + PGAlignedXLogBlock zerobuf = {0}; + size_t remaining = count; + + while (remaining > 0) + { + size_t chunk = Min(remaining, sizeof(zerobuf.data)); + + if (dir_write(f, zerobuf.data, chunk) != chunk) + return -1; + remaining -= chunk; + } + return count; + } +} + static int dir_close(Walfile *f, WalCloseMethod method) { @@ -672,6 +726,7 @@ static ssize_t tar_get_file_size(WalWriteMethod *wwmethod, static char *tar_get_file_name(WalWriteMethod *wwmethod, const char *pathname, const char *temp_suffix); static ssize_t tar_write(Walfile *f, const void *buf, size_t count); +static ssize_t tar_write_zeros(Walfile *f, size_t count); static int tar_sync(Walfile *f); static bool tar_finish(WalWriteMethod *wwmethod); static void tar_free(WalWriteMethod *wwmethod); @@ -683,6 +738,7 @@ static const WalWriteMethodOps WalTarMethodOps = { .get_file_size = tar_get_file_size, .get_file_name = tar_get_file_name, .write = tar_write, + .write_zeros = tar_write_zeros, .sync = tar_sync, .finish = tar_finish, .free = tar_free @@ -801,6 +857,23 @@ tar_write(Walfile *f, const void *buf, size_t count) } } +static ssize_t +tar_write_zeros(Walfile *f, size_t count) +{ + PGAlignedXLogBlock zerobuf = {0}; + size_t remaining = count; + + while (remaining > 0) + { + size_t chunk = Min(remaining, sizeof(zerobuf.data)); + + if (tar_write(f, zerobuf.data, chunk) != chunk) + return -1; + remaining -= chunk; + } + return count; +} + static bool tar_write_padding_data(TarMethodFile *f, size_t bytes) { diff --git a/src/bin/pg_basebackup/walmethods.h b/src/bin/pg_basebackup/walmethods.h index f296a4e43ab..92e3f687659 100644 --- a/src/bin/pg_basebackup/walmethods.h +++ b/src/bin/pg_basebackup/walmethods.h @@ -72,6 +72,9 @@ typedef struct WalWriteMethodOps */ ssize_t (*write) (Walfile *f, const void *buf, size_t count); + /* Advance the output by count zero bytes. */ + ssize_t (*write_zeros) (Walfile *f, size_t count); + /* * fsync the contents of the specified file. Returns 0 on success. */ diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index 130ba929109..3faf9b7b220 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -232,7 +232,7 @@ extern XLogRecPtr XLogInsertRecord(struct XLogRecData *rdata, extern void XLogFlush(XLogRecPtr record); extern bool XLogBackgroundFlush(void); extern bool XLogNeedsFlush(XLogRecPtr record); -extern int XLogFileInit(XLogSegNo logsegno, TimeLineID logtli); +extern int XLogFileInit(XLogSegNo logsegno, TimeLineID logtli, bool *created); extern int XLogFileOpen(XLogSegNo segno, TimeLineID tli); extern void CheckXLogRemoved(XLogSegNo segno, TimeLineID tli); diff --git a/src/include/libpq/protocol.h b/src/include/libpq/protocol.h index eae8f0e7238..0d002bb06eb 100644 --- a/src/include/libpq/protocol.h +++ b/src/include/libpq/protocol.h @@ -75,6 +75,7 @@ #define PqReplMsg_Keepalive 'k' #define PqReplMsg_PrimaryStatusUpdate 's' #define PqReplMsg_WALData 'w' +#define PqReplMsg_WALDataZeros 'z' /* Replication codes sent by the standby (wrapped in CopyData messages). */ diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build index 72113c5ac6e..30eda3573ac 100644 --- a/src/test/recovery/meson.build +++ b/src/test/recovery/meson.build @@ -65,6 +65,7 @@ tests += { 't/054_unlogged_sequence_promotion.pl', 't/055_cascade_reconnect.pl', 't/056_standby_snapshot_export.pl', + 't/057_stream_wal_zeros.pl', ], }, } diff --git a/src/test/recovery/t/057_stream_wal_zeros.pl b/src/test/recovery/t/057_stream_wal_zeros.pl new file mode 100644 index 00000000000..0c8d601c2e6 --- /dev/null +++ b/src/test/recovery/t/057_stream_wal_zeros.pl @@ -0,0 +1,65 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group + +use strict; +use warnings FATAL => 'all'; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +sub files_are_equal +{ + my ($left, $right) = @_; + open(my $left_fh, '<:raw', $left) or die "could not open $left: $!"; + open(my $right_fh, '<:raw', $right) or die "could not open $right: $!"; + + while (1) + { + my ($left_buf, $right_buf); + my $left_len = read($left_fh, $left_buf, 64 * 1024); + my $right_len = read($right_fh, $right_buf, 64 * 1024); + die "could not read WAL files: $!" + if !defined($left_len) || !defined($right_len); + return 0 if $left_len != $right_len || $left_buf ne $right_buf; + last if $left_len == 0; + } + + close($left_fh) or die "could not close $left: $!"; + close($right_fh) or die "could not close $right: $!"; + return 1; +} + +my $primary = PostgreSQL::Test::Cluster->new('primary'); +$primary->init(allows_streaming => 1); +$primary->start; + +$primary->backup('backup'); +my $standby = PostgreSQL::Test::Cluster->new('standby'); +$standby->init_from_backup($primary, 'backup', has_streaming => 1); +$standby->start; + +like( + $primary->safe_psql( + 'postgres', + q[SELECT query FROM pg_stat_activity WHERE backend_type = 'walsender']), + qr/\(SKIP_WAL_PADDING\)$/, + 'standby requests compact WAL padding messages'); + +# Start near the beginning of a segment, then generate a small amount of WAL +# so that the next switch leaves a large zero-filled tail. +$primary->safe_psql('postgres', 'SELECT pg_switch_wal()'); +$primary->wait_for_replay_catchup($standby); +$primary->safe_psql('postgres', + 'CREATE TABLE stream_wal_zeros AS SELECT generate_series(1, 10) AS i'); + +my $walfile = $primary->safe_psql('postgres', + 'SELECT pg_walfile_name(pg_switch_wal())'); +my $flush_lsn = $primary->lsn('flush'); +$primary->wait_for_catchup($standby, 'flush', $flush_lsn); + +my $primary_path = $primary->data_dir . "/pg_wal/$walfile"; +my $standby_path = $standby->data_dir . "/pg_wal/$walfile"; + +ok(files_are_equal($standby_path, $primary_path), + 'streamed WAL segment is reconstructed byte for byte'); + +done_testing(); -- 2.17.1
From 05b870f483794d41e3ba91791fb70ae36eeb619f Mon Sep 17 00:00:00 2001 From: Sehrope Sarkuni <[email protected]> Date: Wed, 9 Sep 2026 16:20:09 +0000 Subject: [PATCH v3 2/2] Fix up test races and reset recvFileIsNew on every close Poll for the walsender's START_REPLICATION instead of reading it once, verify via pg_stat_io that the walreceiver skipped the padding, and reset recvFileIsNew at the end-of-timeline close too. --- src/backend/replication/walreceiver.c | 1 + src/test/recovery/t/057_stream_wal_zeros.pl | 31 ++++++++++++++++++--- 2 files changed, 28 insertions(+), 4 deletions(-) diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c index f96180c9251..365678f5ac0 100644 --- a/src/backend/replication/walreceiver.c +++ b/src/backend/replication/walreceiver.c @@ -737,6 +737,7 @@ WalReceiverMain(const void *startup_data, size_t startup_data_len) XLogArchiveNotify(xlogfname); } recvFile = -1; + recvFileIsNew = false; elog(DEBUG1, "walreceiver ended streaming and awaits new instructions"); WalRcvWaitForStartPosition(&startpoint, &startpointTLI); diff --git a/src/test/recovery/t/057_stream_wal_zeros.pl b/src/test/recovery/t/057_stream_wal_zeros.pl index 0c8d601c2e6..a879761dc53 100644 --- a/src/test/recovery/t/057_stream_wal_zeros.pl +++ b/src/test/recovery/t/057_stream_wal_zeros.pl @@ -37,17 +37,25 @@ my $standby = PostgreSQL::Test::Cluster->new('standby'); $standby->init_from_backup($primary, 'backup', has_streaming => 1); $standby->start; -like( - $primary->safe_psql( +# The walreceiver connects some time after the standby starts accepting +# connections, so wait for its START_REPLICATION to show up rather than +# reading pg_stat_activity once. +ok( $primary->poll_query_until( 'postgres', - q[SELECT query FROM pg_stat_activity WHERE backend_type = 'walsender']), - qr/\(SKIP_WAL_PADDING\)$/, + q[SELECT EXISTS (SELECT 1 FROM pg_stat_activity + WHERE backend_type = 'walsender' + AND query LIKE '%(SKIP_WAL_PADDING)')]), 'standby requests compact WAL padding messages'); # Start near the beginning of a segment, then generate a small amount of WAL # so that the next switch leaves a large zero-filled tail. $primary->safe_psql('postgres', 'SELECT pg_switch_wal()'); $primary->wait_for_replay_catchup($standby); + +# Count only the walreceiver's writes for the segment under test. Counters +# still pending in the walreceiver from before the reset can only make the +# total larger. +$standby->safe_psql('postgres', "SELECT pg_stat_reset_shared('io')"); $primary->safe_psql('postgres', 'CREATE TABLE stream_wal_zeros AS SELECT generate_series(1, 10) AS i'); @@ -62,4 +70,19 @@ my $standby_path = $standby->data_dir . "/pg_wal/$walfile"; ok(files_are_equal($standby_path, $primary_path), 'streamed WAL segment is reconstructed byte for byte'); +# The bytes match whether the padding was streamed or skipped, so also check +# that the walreceiver did not write a whole segment. Restart the standby +# to flush its I/O statistics first. +$standby->restart; + +my $written = $standby->safe_psql('postgres', + q[SELECT write_bytes FROM pg_stat_io + WHERE backend_type = 'walreceiver' AND object = 'wal' + AND context = 'normal']); +my $segsize = $standby->safe_psql('postgres', + q[SELECT setting FROM pg_settings WHERE name = 'wal_segment_size']); +cmp_ok($written, '>', 0, 'walreceiver reported its WAL writes'); +cmp_ok($written, '<', $segsize, + 'walreceiver did not write the zero-filled padding'); + done_testing(); -- 2.17.1
