From dc0abcf52584a10713e8568e53d73e46c697cecc Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy Date: Mon, 17 Aug 2026 18:55:23 +0000 Subject: [PATCH v5] Use WALReadFromBuffers() in more places. Commit 91f2cae7a4 introduced WALReadFromBuffers() but used it only for physical replication walsenders. This commit extends it to all callers that go through read_local_xlog_page(), including logical replication walsenders. This helps when logical replication consumers are keeping up with WAL generation: the walsender finds the requested WAL still in the buffers, avoiding a file read. The gain is largest with WAL direct IO, where a file read is a physical disk read. Without direct IO it still saves the syscall and does not regress. The benefit depends on the workload and how closely clients follow the insertion point. A read fully satisfied from WAL buffers does not call WALRead(), which is where the reader closes and reopens its segment file as it crosses segments. So a buffer-only read never notices a segment change: the open file stays on the old segment while the reader's segment number advances to the new one. For example, when the first page of segment 2 comes from buffers, the segment number becomes 2 but the file is still open on segment 1. A later read of segment 2 that falls back to the file reuses that stale descriptor and returns segment 1's data, seen during decoding as an "unexpected pageaddr" error. Fix this by closing the open segment after a buffer-only read so the next file read reopens the correct one. Author: Bharath Rupireddy Reviewed-by: Jingtang Zhang Reviewed-by: Nitin Jadhav Discussion: https://www.postgresql.org/message-id/CALj2ACVfF2Uj9NoFy-5m98HNtjHpuD17EDE9twVeJng-jTAe7A%40mail.gmail.com --- src/backend/access/transam/xlogutils.c | 32 ++++++++- src/backend/replication/walsender.c | 95 ++++++++++++++++++-------- 2 files changed, 97 insertions(+), 30 deletions(-) diff --git a/src/backend/access/transam/xlogutils.c b/src/backend/access/transam/xlogutils.c index 58b9dab6a90..723e480d394 100644 --- a/src/backend/access/transam/xlogutils.c +++ b/src/backend/access/transam/xlogutils.c @@ -900,6 +900,7 @@ read_local_xlog_page_guts(XLogReaderState *state, XLogRecPtr targetPagePtr, int count; WALReadError errinfo; TimeLineID currTLI; + Size bytesRead; loc = targetPagePtr + reqLen; @@ -1031,9 +1032,34 @@ read_local_xlog_page_guts(XLogReaderState *state, XLogRecPtr targetPagePtr, count = read_upto - targetPagePtr; } - if (!WALRead(state, cur_page, targetPagePtr, count, tli, - &errinfo)) - WALReadRaiseError(&errinfo); + /* Try reading from WAL buffers first */ + bytesRead = WALReadFromBuffers(cur_page, targetPagePtr, count, currTLI); + + /* Read whatever is left from the WAL file */ + if (bytesRead < count) + { + if (!WALRead(state, + cur_page + bytesRead, + targetPagePtr + bytesRead, + count - bytesRead, + tli, + &errinfo)) + { + WALReadRaiseError(&errinfo); + } + bytesRead = count; /* All requested bytes read */ + } + else if (state->seg.ws_file >= 0) + { + /* + * Close the segment after a read fully satisfied from WAL buffers, so + * the next file read reopens the correct one. See + * logical_read_xlog_page() for why this is needed. + */ + state->routine.segment_close(state); + } + + Assert(bytesRead == count); /* number of valid bytes in the buffer */ return count; diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index c65dd324325..87e3f08a63f 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -1098,6 +1098,7 @@ logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int req WALReadError errinfo; XLogSegNo segno; TimeLineID currTLI; + Size bytesRead; /* * Make sure we have enough WAL available before retrieving the current @@ -1157,16 +1158,47 @@ logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int req else count = flushptr - targetPagePtr; /* part of the page available */ - /* now actually read the data, we know it's there */ - if (!WALRead(state, - cur_page, - targetPagePtr, - count, - currTLI, /* Pass the current TLI because only + /* Try reading from WAL buffers first */ + bytesRead = WALReadFromBuffers(cur_page, targetPagePtr, count, currTLI); + + targetPagePtr += bytesRead; + + /* Read whatever is left from the WAL file */ + if (bytesRead < count) + { + if (!WALRead(state, + cur_page + bytesRead, + targetPagePtr, + count - bytesRead, + currTLI, /* Pass the current TLI because only * WalSndSegmentOpen controls whether new TLI * is needed. */ - &errinfo)) - WALReadRaiseError(&errinfo); + &errinfo)) + { + WALReadRaiseError(&errinfo); + } + bytesRead = count; /* All requested bytes read */ + } + else if (state->seg.ws_file >= 0) + { + /* + * A read fully satisfied from WAL buffers skips WALRead(), which is + * where ws_file is closed and reopened as the reader crosses + * segments. So a buffer-only read never notices the segment change. + * ws_file stays open on the old segment while ReadPageInternal() + * advances ws_segno. For example, when the first page of segment 2 + * comes from buffers, ws_segno becomes 2 but ws_file is still open on + * segment 1. A later read of segment 2 that falls back to the file + * reuses the stale descriptor, since WALRead() decides whether to + * reopen from ws_segno (already 2) rather than the open file. It + * reads segment 1 and returns the wrong segment's WAL, seen during + * decoding as an "unexpected pageaddr" error. Close the segment after + * a buffer-only read so the next file read reopens the correct one. + */ + state->routine.segment_close(state); + } + + Assert(bytesRead == count); /* * After reading into the buffer, check that what we read was valid. We do @@ -3367,7 +3399,7 @@ XLogSendPhysical(void) Size nbytes; XLogSegNo segno; WALReadError errinfo; - Size rbytes; + Size bytesRead; /* If requested switch the WAL sender to the stopping state. */ if (got_STOPPING) @@ -3583,24 +3615,33 @@ XLogSendPhysical(void) enlargeStringInfo(&output_message, nbytes); retry: - /* attempt to read WAL from WAL buffers first */ - rbytes = WALReadFromBuffers(&output_message.data[output_message.len], - startptr, nbytes, xlogreader->seg.ws_tli); - output_message.len += rbytes; - startptr += rbytes; - nbytes -= rbytes; - - /* now read the remaining WAL from WAL file */ - if (nbytes > 0 && - !WALRead(xlogreader, - &output_message.data[output_message.len], - startptr, - nbytes, - xlogreader->seg.ws_tli, /* Pass the current TLI because - * only WalSndSegmentOpen controls - * whether new TLI is needed. */ - &errinfo)) - WALReadRaiseError(&errinfo); + /* Try reading from WAL buffers first */ + bytesRead = WALReadFromBuffers(&output_message.data[output_message.len], + startptr, + nbytes, + xlogreader->seg.ws_tli); + + startptr += bytesRead; + + /* Read whatever is left from the WAL file */ + if (bytesRead < nbytes) + { + if (!WALRead(xlogreader, + &output_message.data[output_message.len + bytesRead], + startptr, + nbytes - bytesRead, + xlogreader->seg.ws_tli, /* Pass the current TLI + * because only + * WalSndSegmentOpen controls + * whether new TLI is needed. */ + &errinfo)) + { + WALReadRaiseError(&errinfo); + } + bytesRead = nbytes; /* All requested bytes read */ + } + + Assert(bytesRead == nbytes); /* See logical_read_xlog_page(). */ XLByteToSeg(startptr, segno, xlogreader->segcxt.ws_segsize); -- 2.47.3