From fec4c04695ad7e383c46d5ad066764f6eaaa7441 Mon Sep 17 00:00:00 2001 From: Andrey Borodin Date: Thu, 6 Aug 2026 12:03:11 +0500 Subject: [PATCH v2] 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/0E59ED14-DE1B-41D9-886D-FE409BF0A056@yandex-team.ru --- 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;" - START_REPLICATION [ SLOT slot_name ] [ PHYSICAL ] XXX/XXX [ TIMELINE tli ] + START_REPLICATION [ SLOT slot_name ] [ PHYSICAL ] XXX/XXX [ TIMELINE tli ] [ ( option [, ...] ) ] START_REPLICATION @@ -2621,6 +2621,24 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;" message, and then starts to stream WAL to the frontend. + + The following option is supported: + + + + + SKIP_WAL_PADDING [ boolean ] + + + Requests that zero-filled padding after a WAL switch be represented + by a Zero WAL data message instead of being sent + as ordinary WAL data. A client must not request this option unless + it supports that message type. + + + + + If a slot's name is provided via slot_name, it will be updated @@ -2725,6 +2743,63 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;" + + Zero WAL data (B) + + + + Byte1('z') + + + 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 SKIP_WAL_PADDING option of + START_REPLICATION. + + + + + + Int64 + + + The starting point of the zero-filled WAL data. + + + + + + Int64 + + + The current end of WAL on the server. + + + + + + Int64 + + + The server's system clock at the time of transmission, as + microseconds since midnight on 2000-01-01. + + + + + + Int64 + + + The number of zero bytes in this section. + + + + + + + Primary keepalive message (B) 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 generic_option_list +%type generic_option_list opt_physical_options %type generic_option %type opt_timeline %type 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(); -- That's all, folks. May the source be with you.