From dc0abcf52584a10713e8568e53d73e46c697cecc Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <bharath.rupireddyforpostgres@gmail.com>
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 <bharath.rupireddyforpostgres@gmail.com>
Reviewed-by: Jingtang Zhang <mrdrivingduck@gmail.com>
Reviewed-by: Nitin Jadhav <nitinjadhavpostgres@gmail.com>
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

