Subject: [PATCH v15 2/4] Move page-reader out of XLogReadRecord

From: Kyotaro Horiguchi <[email-masked]>
Date: Tue, 10 Sep 2019 12:58:27 +0900

This is the second step of removing callbacks from WAL record reader.
Since it is essential to take in additional data while reading a
record, the function have to ask caller for new data while keeping
working state. Thus the function is turned into a state machine.
---
 src/backend/access/transam/twophase.c         |  15 +-
 src/backend/access/transam/xlog.c             |  58 +-
 src/backend/access/transam/xlogreader.c       | 700 +++++++++++-------
 src/backend/access/transam/xlogutils.c        |  17 +-
 src/backend/replication/logical/logical.c     |  26 +-
 .../replication/logical/logicalfuncs.c        |  13 +-
 src/backend/replication/slotfuncs.c           |  18 +-
 src/backend/replication/walsender.c           |  32 +-
 src/bin/pg_rewind/parsexlog.c                 |  84 +--
 src/bin/pg_waldump/pg_waldump.c               |  36 +-
 src/include/access/xlogreader.h               | 121 ++-
 src/include/access/xlogutils.h                |   4 +-
 src/include/replication/logical.h             |  11 +-
 13 files changed, 648 insertions(+), 487 deletions(-)

diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 9b2e59bf0e..b0d60a0d0f 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -1330,11 +1330,8 @@ XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len)
 	XLogReaderState *xlogreader;
 	char	   *errormsg;
 
-	xlogreader = XLogReaderAllocate(wal_segment_size, NULL,
-									XL_ROUTINE(.page_read = &read_local_xlog_page,
-											   .segment_open = &wal_segment_open,
-											   .segment_close = &wal_segment_close),
-									NULL);
+	xlogreader = XLogReaderAllocate(wal_segment_size, NULL, wal_segment_close);
+
 	if (!xlogreader)
 		ereport(ERROR,
 				(errcode(ERRCODE_OUT_OF_MEMORY),
@@ -1342,7 +1339,13 @@ XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len)
 				 errdetail("Failed while allocating a WAL reading processor.")));
 
 	XLogBeginRead(xlogreader, lsn);
-	record = XLogReadRecord(xlogreader, &errormsg);
+	while (XLogReadRecord(xlogreader, &record, &errormsg) ==
+		   XLREAD_NEED_DATA)
+	{
+		if (!read_local_xlog_page(xlogreader))
+			break;
+	}
+
 	if (record == NULL)
 		ereport(ERROR,
 				(errcode_for_file_access(),
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index e570e56a24..f9b0108602 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -831,13 +831,6 @@ static XLogSource currentSource = XLOG_FROM_ANY;
 static bool lastSourceFailed = false;
 static bool pendingWalRcvRestart = false;
 
-typedef struct XLogPageReadPrivate
-{
-	int			emode;
-	bool		fetching_ckpt;	/* are we fetching a checkpoint record? */
-	bool		randAccess;
-} XLogPageReadPrivate;
-
 /*
  * These variables track when we last obtained some WAL data to process,
  * and where we got it from.  (XLogReceiptSource is initially the same as
@@ -912,8 +905,8 @@ static bool InstallXLogFileSegment(XLogSegNo *segno, char *tmppath,
 static int	XLogFileRead(XLogSegNo segno, int emode, TimeLineID tli,
 						 XLogSource source, bool notfoundOk);
 static int	XLogFileReadAnyTLI(XLogSegNo segno, int emode, XLogSource source);
-static bool	XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
-						 int reqLen, XLogRecPtr targetRecPtr, char *readBuf);
+static bool XLogPageRead(XLogReaderState *xlogreader,
+						 bool fetching_ckpt, int emode, bool randAccess);
 static bool WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
 										bool fetching_ckpt, XLogRecPtr tliRecPtr);
 static int	emode_for_corrupt_record(int emode, XLogRecPtr RecPtr);
@@ -1225,8 +1218,7 @@ XLogInsertRecord(XLogRecData *rdata,
 			appendBinaryStringInfo(&recordBuf, rdata->data, rdata->len);
 
 		if (!debug_reader)
-			debug_reader = XLogReaderAllocate(wal_segment_size, NULL,
-											  XL_ROUTINE(), NULL);
+			debug_reader = XLogReaderAllocate(wal_segment_size, NULL NULL);
 
 		if (!debug_reader)
 		{
@@ -4326,15 +4318,10 @@ CleanupBackupHistory(void)
  * record is available.
  */
 static XLogRecord *
-ReadRecord(XLogReaderState *xlogreader, int emode,
-		   bool fetching_ckpt)
+ReadRecord(XLogReaderState *xlogreader, int emode, bool fetching_ckpt)
 {
 	XLogRecord *record;
-	XLogPageReadPrivate *private = (XLogPageReadPrivate *) xlogreader->private_data;
-
-	private->fetching_ckpt = fetching_ckpt;
-	private->emode = emode;
-	private->randAccess = (xlogreader->ReadRecPtr == InvalidXLogRecPtr);
+	bool		randAccess = (xlogreader->ReadRecPtr == InvalidXLogRecPtr);
 
 	/* This is the first attempt to read this page. */
 	lastSourceFailed = false;
@@ -4342,8 +4329,16 @@ ReadRecord(XLogReaderState *xlogreader, int emode,
 	for (;;)
 	{
 		char	   *errormsg;
+		XLogReadRecordResult result;
+
+		while ((result = XLogReadRecord(xlogreader, &record, &errormsg))
+			   == XLREAD_NEED_DATA)
+		{
+			if (!XLogPageRead(xlogreader, fetching_ckpt, emode, randAccess))
+				break;
+
+		}
 
-		record = XLogReadRecord(xlogreader, &errormsg);
 		ReadRecPtr = xlogreader->ReadRecPtr;
 		EndRecPtr = xlogreader->EndRecPtr;
 		if (record == NULL)
@@ -6320,7 +6315,6 @@ StartupXLOG(void)
 	bool		backupFromStandby = false;
 	DBState		dbstate_at_startup;
 	XLogReaderState *xlogreader;
-	XLogPageReadPrivate private;
 	bool		fast_promoted = false;
 	struct stat st;
 
@@ -6480,13 +6474,9 @@ StartupXLOG(void)
 		OwnLatch(&XLogCtl->recoveryWakeupLatch);
 
 	/* Set up XLOG reader facility */
-	MemSet(&private, 0, sizeof(XLogPageReadPrivate));
 	xlogreader =
-		XLogReaderAllocate(wal_segment_size, NULL,
-						   XL_ROUTINE(.page_read = &XLogPageRead,
-									  .segment_open = NULL,
-									  .segment_close = wal_segment_close),
-						   &private);
+		XLogReaderAllocate(wal_segment_size, NULL, wal_segment_close);
+
 	if (!xlogreader)
 		ereport(ERROR,
 				(errcode(ERRCODE_OUT_OF_MEMORY),
@@ -11860,12 +11850,13 @@ CancelBackup(void)
  * sleep and retry.
  */
 static bool
-XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, int reqLen,
-			 XLogRecPtr targetRecPtr, char *readBuf)
+XLogPageRead(XLogReaderState *xlogreader,
+			 bool fetching_ckpt, int emode, bool randAccess)
 {
-	XLogPageReadPrivate *private =
-	(XLogPageReadPrivate *) xlogreader->private_data;
-	int			emode = private->emode;
+	char *readBuf				= xlogreader->readBuf;
+	XLogRecPtr targetPagePtr	= xlogreader->readPagePtr;
+	int reqLen					= xlogreader->readLen;
+	XLogRecPtr targetRecPtr		= xlogreader->ReadRecPtr;
 	uint32		targetPageOff;
 	XLogSegNo	targetSegNo PG_USED_FOR_ASSERTS_ONLY;
 	int			r;
@@ -11908,8 +11899,8 @@ retry:
 		 flushedUpto < targetPagePtr + reqLen))
 	{
 		if (!WaitForWALToBecomeAvailable(targetPagePtr + reqLen,
-										 private->randAccess,
-										 private->fetching_ckpt,
+										 randAccess,
+										 fetching_ckpt,
 										 targetRecPtr))
 		{
 			if (readFile >= 0)
@@ -12014,6 +12005,7 @@ retry:
 		goto next_record_is_invalid;
 	}
 
+	Assert(xlogreader->readPagePtr == targetPagePtr);
 	xlogreader->readLen = readLen;
 	return true;
 
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
index 3d599325ee..9281b57379 100644
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -40,7 +40,7 @@ static bool XLogNeedData(XLogReaderState *state, XLogRecPtr pageptr,
 						 int reqLen, bool header_inclusive);
 static void XLogReaderInvalReadState(XLogReaderState *state);
 static bool ValidXLogRecordHeader(XLogReaderState *state, XLogRecPtr RecPtr,
-								  XLogRecPtr PrevRecPtr, XLogRecord *record, bool randAccess);
+								  XLogRecPtr PrevRecPtr, XLogRecord *record);
 static bool ValidXLogRecord(XLogReaderState *state, XLogRecord *record,
 							XLogRecPtr recptr);
 static void ResetDecoder(XLogReaderState *state);
@@ -73,7 +73,7 @@ report_invalid_record(XLogReaderState *state, const char *fmt,...)
  */
 XLogReaderState *
 XLogReaderAllocate(int wal_segment_size, const char *waldir,
-				   XLogReaderRoutine *routine, void *private_data)
+				   WALSegmentCleanupCB cleanup_cb)
 {
 	XLogReaderState *state;
 
@@ -84,7 +84,7 @@ XLogReaderAllocate(int wal_segment_size, const char *waldir,
 		return NULL;
 
 	/* initialize caller-provided support functions */
-	state->routine = *routine;
+	state->cleanup_cb = cleanup_cb;
 
 	state->max_block_id = -1;
 
@@ -107,8 +107,6 @@ XLogReaderAllocate(int wal_segment_size, const char *waldir,
 	WALOpenSegmentInit(&state->seg, &state->segcxt, wal_segment_size,
 					   waldir);
 
-	/* system_identifier initialized to zeroes above */
-	state->private_data = private_data;
 	/* ReadRecPtr, EndRecPtr and readLen initialized to zeroes above */
 	state->errormsg_buf = palloc_extended(MAX_ERRORMSG_LEN + 1,
 										  MCXT_ALLOC_NO_OOM);
@@ -140,8 +138,8 @@ XLogReaderFree(XLogReaderState *state)
 {
 	int			block_id;
 
-	if (state->seg.ws_file != -1)
-		state->routine.segment_close(state);
+	if (state->seg.ws_file >= 0)
+		state->cleanup_cb(state);
 
 	for (block_id = 0; block_id <= XLR_MAX_BLOCK_ID; block_id++)
 	{
@@ -246,6 +244,7 @@ XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
 	/* Begin at the passed-in record pointer. */
 	state->EndRecPtr = RecPtr;
 	state->ReadRecPtr = InvalidXLogRecPtr;
+	state->readRecordState = XLREAD_NEXT_RECORD;
 }
 
 /*
@@ -254,317 +253,452 @@ XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
  * XLogBeginRead() or XLogFindNextRecord() must be called before the first call
  * to XLogReadRecord().
  *
- * If the page_read callback fails to read the requested data, NULL is
- * returned.  The callback is expected to have reported the error; errormsg
- * is set to NULL.
+ * This function may return XLREAD_NEED_DATA several times before returning a
+ * result record. The caller shall read in some new data then call this
+ * function again with the same parameters.
  *
- * If the reading fails for some other reason, NULL is also returned, and
- * *errormsg is set to a string with details of the failure.
+ * When a record is successfully read, returns XLREAD_SUCCESS with result
+ * record being stored in *record. Otherwise *record is NULL.
  *
- * The returned pointer (or *errormsg) points to an internal buffer that's
- * valid until the next call to XLogReadRecord.
+ * Returns XLREAD_NEED_DATA if more data is needed to finish reading the
+ * current record.  In that case, state->readPagePtr and state->readLen inform
+ * the desired position and minimum length of data needed. The caller shall
+ * read in the requested data and set state->readBuf to point to a buffer
+ * containing it. The caller must also set state->readPageTLI and
+ * state->readLen to indicate the timeline that it was read from, and the
+ * length of data that is now available (which must be >= given readLen),
+ * respectively.
+ *
+ * If invalid data is encountered, returns XLREAD_FAIL with *record being set to
+ * NULL. *errormsg is set to a string with details of the failure.
+ * The returned pointer (or *errormsg) points to an internal buffer that's valid
+ * until the next call to XLogReadRecord.
+ *
+ *
+ * This function runs a state machine consists of the following states.
+ *
+ * XLREAD_NEXT_RECORD :
+ *    The initial state, if called with valid RecPtr, try to read a record at
+ *    that position.  If invalid RecPtr is given try to read a record just after
+ *    the last one previously read.
+ *    This state ens after setting ReadRecPtr. Then goes to XLREAD_TOT_LEN.
+ *
+ * XLREAD_TOT_LEN:
+ *    Examining record header. Ends after reading record total
+ *    length. recordRemainLen and recordGotLen are initialized.
+ *
+ * XLREAD_FIRST_FRAGMENT:
+ *    Reading the first fragment. Ends with finishing reading a single
+ *    record. Goes to XLREAD_NEXT_RECORD if that's all or
+ *    XLREAD_CONTINUATION if we have continuation.
+
+ * XLREAD_CONTINUATION:
+ *    Reading continuation of record. Ends with finishing the whole record then
+ *    goes to XLREAD_NEXT_RECORD. During this state, recordRemainLen indicates
+ *    how much is left and readRecordBuf holds the partially assert
+ *    record.recordContRecPtr points to the beginning of the next page where to
+ *    continue.
+ *
+ * If wrong data found in any state, the state machine stays at the current
+ * state. This behavior allows to continue reading a reacord switching among
+ * different souces, while streaming replication.
  */
-XLogRecord *
-XLogReadRecord(XLogReaderState *state, char **errormsg)
+XLogReadRecordResult
+XLogReadRecord(XLogReaderState *state, XLogRecord **record, char **errormsg)
 {
-	XLogRecPtr	RecPtr;
-	XLogRecord *record;
-	XLogRecPtr	targetPagePtr;
-	bool		randAccess;
-	uint32		len,
-				total_len;
-	uint32		targetRecOff;
-	uint32		pageHeaderSize;
-	bool		gotheader;
+	XLogRecord *prec;
 
-	/*
-	 * randAccess indicates whether to verify the previous-record pointer of
-	 * the record we're reading.  We only do this if we're reading
-	 * sequentially, which is what we initially assume.
-	 */
-	randAccess = false;
+	*record = NULL;
 
 	/* reset error state */
 	*errormsg = NULL;
 	state->errormsg_buf[0] = '\0';
 
-	ResetDecoder(state);
-
-	RecPtr = state->EndRecPtr;
-
-	if (state->ReadRecPtr != InvalidXLogRecPtr)
-	{
-		/* read the record after the one we just read */
-
-		/*
-		 * EndRecPtr is pointing to end+1 of the previous WAL record.  If
-		 * we're at a page boundary, no more records can fit on the current
-		 * page. We must skip over the page header, but we can't do that until
-		 * we've read in the page, since the header size is variable.
-		 */
-	}
-	else
-	{
-		/*
-		 * Caller supplied a position to start at.
-		 *
-		 * In this case, EndRecPtr should already be pointing to a valid
-		 * record starting position.
-		 */
-		Assert(XRecOffIsValid(RecPtr));
-		randAccess = true;
-	}
-
-	state->currRecPtr = RecPtr;
-
-	targetPagePtr = RecPtr - (RecPtr % XLOG_BLCKSZ);
-	targetRecOff = RecPtr % XLOG_BLCKSZ;
-
-	/*
-	 * Read the page containing the record into state->readBuf. Request enough
-	 * byte to cover the whole record header, or at least the part of it that
-	 * fits on the same page.
-	 */
-	while (XLogNeedData(state, targetPagePtr,
-						Min(targetRecOff + SizeOfXLogRecord, XLOG_BLCKSZ),
-						targetRecOff != 0))
-	{
-		if (!state->routine.page_read(state, state->readPagePtr, state->readLen,
-									  RecPtr, state->readBuf))
-			break;
-	}
-
-	if (!state->page_verified)
-		goto err;
-
-	/*
-	 * We have at least the page header, so we can examine it now.
-	 */
-	pageHeaderSize = XLogPageHeaderSize((XLogPageHeader) state->readBuf);
-	if (targetRecOff == 0)
-	{
-		/*
-		 * At page start, so skip over page header.
-		 */
-		RecPtr += pageHeaderSize;
-		targetRecOff = pageHeaderSize;
-	}
-	else if (targetRecOff < pageHeaderSize)
-	{
-		report_invalid_record(state, "invalid record offset at %X/%X",
-							  (uint32) (RecPtr >> 32), (uint32) RecPtr);
-		goto err;
-	}
-
-	if ((((XLogPageHeader) state->readBuf)->xlp_info & XLP_FIRST_IS_CONTRECORD) &&
-		targetRecOff == pageHeaderSize)
-	{
-		report_invalid_record(state, "contrecord is requested by %X/%X",
-							  (uint32) (RecPtr >> 32), (uint32) RecPtr);
-		goto err;
-	}
-
-	/* XLogNeedData has verified the page header */
-	Assert(pageHeaderSize <= state->readLen);
-
-	/*
-	 * Read the record length.
-	 *
-	 * NB: Even though we use an XLogRecord pointer here, the whole record
-	 * header might not fit on this page. xl_tot_len is the first field of the
-	 * struct, so it must be on this page (the records are MAXALIGNed), but we
-	 * cannot access any other fields until we've verified that we got the
-	 * whole header.
-	 */
-	record = (XLogRecord *) (state->readBuf + RecPtr % XLOG_BLCKSZ);
-	total_len = record->xl_tot_len;
-
-	/*
-	 * If the whole record header is on this page, validate it immediately.
-	 * Otherwise do just a basic sanity check on xl_tot_len, and validate the
-	 * rest of the header after reading it from the next page.  The xl_tot_len
-	 * check is necessary here to ensure that we enter the "Need to reassemble
-	 * record" code path below; otherwise we might fail to apply
-	 * ValidXLogRecordHeader at all.
-	 */
-	if (targetRecOff <= XLOG_BLCKSZ - SizeOfXLogRecord)
-	{
-		if (!ValidXLogRecordHeader(state, RecPtr, state->ReadRecPtr, record,
-								   randAccess))
-			goto err;
-		gotheader = true;
-	}
-	else
-	{
-		/* XXX: more validation should be done here */
-		if (total_len < SizeOfXLogRecord)
-		{
-			report_invalid_record(state,
-								  "invalid record length at %X/%X: wanted %u, got %u",
-								  (uint32) (RecPtr >> 32), (uint32) RecPtr,
-								  (uint32) SizeOfXLogRecord, total_len);
-			goto err;
-		}
-		gotheader = false;
-	}
-
-	len = XLOG_BLCKSZ - RecPtr % XLOG_BLCKSZ;
-	if (total_len > len)
+	switch (state->readRecordState)
 	{
-		/* Need to reassemble record */
-		char	   *contdata;
-		XLogPageHeader pageHeader;
-		char	   *buffer;
-		uint32		gotlen;
-
-		/*
-		 * Enlarge readRecordBuf as needed.
-		 */
-		if (total_len > state->readRecordBufSize &&
-			!allocate_recordbuf(state, total_len))
-		{
-			/* We treat this as a "bogus data" condition */
-			report_invalid_record(state, "record length %u at %X/%X too long",
-								  total_len,
-								  (uint32) (RecPtr >> 32), (uint32) RecPtr);
-			goto err;
-		}
+		case XLREAD_NEXT_RECORD:
+			ResetDecoder(state);
 
-		/* Copy the first fragment of the record from the first page. */
-		memcpy(state->readRecordBuf,
-			   state->readBuf + RecPtr % XLOG_BLCKSZ, len);
-		buffer = state->readRecordBuf + len;
-		gotlen = len;
-
-		do
-		{
-			int rest_len = total_len - gotlen;
-
-			/* Calculate pointer to beginning of next page */
-			targetPagePtr += XLOG_BLCKSZ;
-
-			/* Wait for the next page to become available */
-			while (XLogNeedData(state, targetPagePtr,
-								Min(rest_len, XLOG_BLCKSZ),
-								false))
+			if (state->ReadRecPtr != InvalidXLogRecPtr)
 			{
-				if (!state->routine.page_read(state, state->readPagePtr,
-											  state->readLen,
-											  state->ReadRecPtr,
-											  state->readBuf))
-					break;
+				/* read the record after the one we just read */
+
+				/*
+				 * EndRecPtr is pointing to end+1 of the previous WAL record.
+				 * If we're at a page boundary, no more records can fit on the
+				 * current page. We must skip over the page header, but we
+				 * can't do that until we've read in the page, since the header
+				 * size is variable.
+				 */
+				state->PrevRecPtr = state->ReadRecPtr;
+				state->ReadRecPtr = state->EndRecPtr;
+			}
+			else
+			{
+				/*
+				 * Caller supplied a position to start at.
+				 *
+				 * In this case, EndRecPtr should already be pointing to a
+				 * valid record starting position.
+				 */
+				Assert(XRecOffIsValid(state->EndRecPtr));
+				state->ReadRecPtr = state->EndRecPtr;
+
+				/*
+				 * We cannot verify the previous-record pointer when we're
+				 * seeking to a particular record. Reset PrevRecPtr so that we
+				 * won't try doing that.
+				 */
+				state->PrevRecPtr = InvalidXLogRecPtr;
+				state->EndRecPtr = InvalidXLogRecPtr; 		/* to be tidy */
 			}
 
+			state->record_verified = false;
+			state->readRecordState = XLREAD_TOT_LEN;
+			/* fall through */
+
+		case XLREAD_TOT_LEN:
+		{
+			uint32		total_len;
+			uint32		pageHeaderSize;
+			XLogRecPtr	targetPagePtr;
+			uint32		targetRecOff;
+			XLogPageHeader pageHeader;
+
+			targetPagePtr =
+				state->ReadRecPtr - (state->ReadRecPtr % XLOG_BLCKSZ);
+			targetRecOff = state->ReadRecPtr % XLOG_BLCKSZ;
+
+			/*
+			 * Check if we have enough data. For the first record in the page,
+			 * the requesting length doesn't contain page header.
+			 */
+			if (XLogNeedData(state, targetPagePtr,
+							 Min(targetRecOff + SizeOfXLogRecord, XLOG_BLCKSZ),
+							 targetRecOff != 0))
+				return XLREAD_NEED_DATA;
+
+			/* error out if caller supplied bogus page */
 			if (!state->page_verified)
 				goto err;
 
-			Assert(SizeOfXLogShortPHD <= state->readLen);
+			/* examine page header now. */
+			pageHeaderSize =
+				XLogPageHeaderSize((XLogPageHeader) state->readBuf);
+			if (targetRecOff == 0)
+			{
+				/* At page start, so skip over page header. */
+				state->ReadRecPtr += pageHeaderSize;
+				targetRecOff = pageHeaderSize;
+			}
+			else if (targetRecOff < pageHeaderSize)
+			{
+				report_invalid_record(state, "invalid record offset at %X/%X",
+									  (uint32) (state->ReadRecPtr >> 32),
+									  (uint32) state->ReadRecPtr);
+				goto err;
+			}
 
-			/* Check that the continuation on next page looks valid */
 			pageHeader = (XLogPageHeader) state->readBuf;
-			if (!(pageHeader->xlp_info & XLP_FIRST_IS_CONTRECORD))
+			if ((pageHeader->xlp_info & XLP_FIRST_IS_CONTRECORD) &&
+				targetRecOff == pageHeaderSize)
 			{
-				report_invalid_record(state,
-									  "there is no contrecord flag at %X/%X",
-									  (uint32) (RecPtr >> 32), (uint32) RecPtr);
+				report_invalid_record(state, "contrecord is requested by %X/%X",
+									  (uint32) (state->ReadRecPtr >> 32),
+									  (uint32) state->ReadRecPtr);
 				goto err;
 			}
 
-			/*
-			 * Cross-check that xlp_rem_len agrees with how much of the record
-			 * we expect there to be left.
-			 */
-			if (pageHeader->xlp_rem_len == 0 ||
-				total_len != (pageHeader->xlp_rem_len + gotlen))
-			{
-				report_invalid_record(state,
-									  "invalid contrecord length %u at %X/%X",
-									  pageHeader->xlp_rem_len,
-									  (uint32) (RecPtr >> 32), (uint32) RecPtr);
-				goto err;
-			}
-
-			/* Append the continuation from this page to the buffer */
-			pageHeaderSize = XLogPageHeaderSize(pageHeader);
-
+			/* XLogNeedData has verified the page header */
 			Assert(pageHeaderSize <= state->readLen);
 
-			contdata = (char *) state->readBuf + pageHeaderSize;
-			len = XLOG_BLCKSZ - pageHeaderSize;
-			if (pageHeader->xlp_rem_len < len)
-				len = pageHeader->xlp_rem_len;
+			/*
+			 * Read the record length.
+			 *
+			 * NB: Even though we use an XLogRecord pointer here, the whole
+			 * record header might not fit on this page. xl_tot_len is the first
+			 * field of the struct, so it must be on this page (the records are
+			 * MAXALIGNed), but we cannot access any other fields until we've
+			 * verified that we got the whole header.
+			 */
+			prec = (XLogRecord *) (state->readBuf +
+								   state->ReadRecPtr % XLOG_BLCKSZ);
+			total_len = prec->xl_tot_len;
 
-			Assert (pageHeaderSize + len <= state->readLen);
-			memcpy(buffer, (char *) contdata, len);
-			buffer += len;
-			gotlen += len;
+			/*
+			 * If the whole record header is on this page, validate it
+			 * immediately.  Otherwise do just a basic sanity check on
+			 * xl_tot_len, and validate the rest of the header after reading it
+			 * from the next page.  The xl_tot_len check is necessary here to
+			 * ensure that we enter the XLREAD_CONTINUATION state below;
+			 * otherwise we might fail to apply ValidXLogRecordHeader at all.
+			 */
+			if (targetRecOff <= XLOG_BLCKSZ - SizeOfXLogRecord)
+			{
+				if (!ValidXLogRecordHeader(state, state->ReadRecPtr,
+										   state->PrevRecPtr, prec))
+					goto err;
 
-			/* If we just reassembled the record header, validate it. */
-			if (!gotheader)
+				state->record_verified = true;
+			}
+			else
 			{
-				record = (XLogRecord *) state->readRecordBuf;
-				if (!ValidXLogRecordHeader(state, RecPtr, state->ReadRecPtr,
-										   record, randAccess))
+				/* XXX: more validation should be done here */
+				if (total_len < SizeOfXLogRecord)
+				{
+					report_invalid_record(state,
+										  "invalid record length at %X/%X: wanted %u, got %u",
+										  (uint32) (state->ReadRecPtr >> 32),
+										  (uint32) state->ReadRecPtr,
+										  (uint32) SizeOfXLogRecord, total_len);
 					goto err;
-				gotheader = true;
+				}
 			}
-		} while (gotlen < total_len);
-
-		Assert(gotheader);
 
-		record = (XLogRecord *) state->readRecordBuf;
-		if (!ValidXLogRecord(state, record, RecPtr))
-			goto err;
+			/*
+			 * Wait for the rest of the record, or the part of the record that
+			 * fit on the first page if crossed a page boundary, to become
+			 * available.
+			 */
+			state->recordGotLen = 0;
+			state->recordRemainLen = total_len;
+			state->readRecordState = XLREAD_FIRST_FRAGMENT;
+		}
+		/* fall through */
 
-		pageHeaderSize = XLogPageHeaderSize((XLogPageHeader) state->readBuf);
-		state->ReadRecPtr = RecPtr;
-		state->EndRecPtr = targetPagePtr + pageHeaderSize
-			+ MAXALIGN(pageHeader->xlp_rem_len);
-	}
-	else
-	{
-		/* Wait for the record data to become available */
-		while (XLogNeedData(state, targetPagePtr,
-							Min(targetRecOff + total_len, XLOG_BLCKSZ), true))
+		case XLREAD_FIRST_FRAGMENT:
 		{
-			if (!state->routine.page_read(state, state->readPagePtr,
-										  state->readLen,
-										  state->ReadRecPtr, state->readBuf))
+			uint32		total_len = state->recordRemainLen;
+			uint32		request_len;
+			uint32		record_len;
+			XLogRecPtr	targetPagePtr;
+			uint32		targetRecOff;
+
+			/*
+			 * Wait for the rest of the record on the first page to become
+			 * available
+			 */
+			targetPagePtr =
+				state->ReadRecPtr - (state->ReadRecPtr % XLOG_BLCKSZ);
+			targetRecOff = state->ReadRecPtr % XLOG_BLCKSZ;
+
+			request_len = Min(targetRecOff + total_len, XLOG_BLCKSZ);
+			record_len = request_len - targetRecOff;
+
+			/* ReadRecPtr contains page header */
+			Assert (targetRecOff != 0);
+			if (XLogNeedData(state, targetPagePtr, request_len, true))
+				return XLREAD_NEED_DATA;
+
+			/* error out if caller supplied bogus page */
+			if (!state->page_verified)
+				goto err;
+
+			prec = (XLogRecord *) (state->readBuf + targetRecOff);
+
+			/* validate record header if not yet */
+			if (!state->record_verified && record_len >= SizeOfXLogRecord)
+			{
+				if (!ValidXLogRecordHeader(state, state->ReadRecPtr,
+										   state->PrevRecPtr, prec))
+					goto err;
+
+				state->record_verified = true;
+			}
+
+
+			if (total_len == record_len)
+			{
+				/* Record does not cross a page boundary */
+				Assert(state->record_verified);
+
+				if (!ValidXLogRecord(state, prec, state->ReadRecPtr))
+					goto err;
+
+				state->record_verified = true;  /* to be tidy */
+
+				/* We already checked the header earlier */
+				state->EndRecPtr = state->ReadRecPtr + MAXALIGN(record_len);
+
+				*record = prec;
+				state->readRecordState = XLREAD_NEXT_RECORD;
 				break;
+			}
+
+			/*
+			 * The record continues on the next page. Need to reassemble
+			 * record
+			 */
+			Assert(total_len > record_len);
+
+			/* Enlarge readRecordBuf as needed. */
+			if (total_len > state->readRecordBufSize &&
+				!allocate_recordbuf(state, total_len))
+			{
+				/* We treat this as a "bogus data" condition */
+				report_invalid_record(state,
+									  "record length %u at %X/%X too long",
+									  total_len,
+									  (uint32) (state->ReadRecPtr >> 32),
+									  (uint32) state->ReadRecPtr);
+				goto err;
+			}
+
+			/* Copy the first fragment of the record from the first page. */
+			memcpy(state->readRecordBuf, state->readBuf + targetRecOff,
+				   record_len);
+			state->recordGotLen += record_len;
+			state->recordRemainLen -= record_len;
+
+			/* Calculate pointer to beginning of next page */
+			state->recordContRecPtr = state->ReadRecPtr + record_len;
+			Assert(state->recordContRecPtr % XLOG_BLCKSZ == 0);
+
+			state->readRecordState = XLREAD_CONTINUATION;
 		}
+		/* fall through */
+
+		case XLREAD_CONTINUATION:
+		{
+			XLogPageHeader pageHeader;
+			uint32		pageHeaderSize;
+			XLogRecPtr	targetPagePtr;
+
+			/* we enter this state only if we haven't read the whole record. */
+			Assert (state->recordRemainLen > 0);
+
+			while(state->recordRemainLen > 0)
+			{
+				char	   *contdata;
+				uint32		request_len;
+				uint32		record_len;
+
+				/* Wait for the next page to become available */
+				targetPagePtr = state->recordContRecPtr;
+
+				/* this request contains page header */
+				Assert (targetPagePtr != 0);
+				if (XLogNeedData(state, targetPagePtr,
+								 Min(state->recordRemainLen, XLOG_BLCKSZ),
+								 false))
+					return XLREAD_NEED_DATA;
+
+				if (!state->page_verified)
+					goto err;
+
+				Assert(SizeOfXLogShortPHD <= state->readLen);
 
-		if (!state->page_verified)
-			goto err;
+				/* Check that the continuation on next page looks valid */
+				pageHeader = (XLogPageHeader) state->readBuf;
+				if (!(pageHeader->xlp_info & XLP_FIRST_IS_CONTRECORD))
+				{
+					report_invalid_record(
+						state,
+						"there is no contrecord flag at %X/%X reading %X/%X",
+						(uint32) (state->recordContRecPtr >> 32),
+						(uint32) state->recordContRecPtr,
+						(uint32) (state->ReadRecPtr >> 32),
+						(uint32) state->ReadRecPtr);
+					goto err;
+				}
 
-		/* Record does not cross a page boundary */
-		if (!ValidXLogRecord(state, record, RecPtr))
-			goto err;
+				/*
+				 * Cross-check that xlp_rem_len agrees with how much of the
+				 * record we expect there to be left.
+				 */
+				if (pageHeader->xlp_rem_len == 0 ||
+					pageHeader->xlp_rem_len != state->recordRemainLen)
+				{
+					report_invalid_record(
+						state,
+						"invalid contrecord length %u at %X/%X reading %X/%X, expected %u",
+						pageHeader->xlp_rem_len,
+						(uint32) (state->recordContRecPtr >> 32),
+						(uint32) state->recordContRecPtr,
+						(uint32) (state->ReadRecPtr >> 32),
+						(uint32) state->ReadRecPtr,
+						state->recordRemainLen);
+					goto err;
+				}
 
-		state->EndRecPtr = RecPtr + MAXALIGN(total_len);
+				/* Append the continuation from this page to the buffer */
+				pageHeaderSize = XLogPageHeaderSize(pageHeader);
 
-		state->ReadRecPtr = RecPtr;
+				/*
+				 * XLogNeedData should have ensured that the whole page header
+				 * was read
+				 */
+				Assert(state->readLen >= pageHeaderSize);
+
+				contdata = (char *) state->readBuf + pageHeaderSize;
+				record_len = XLOG_BLCKSZ - pageHeaderSize;
+				if (pageHeader->xlp_rem_len < record_len)
+					record_len = pageHeader->xlp_rem_len;
+
+				request_len = record_len + pageHeaderSize;
+
+				/* XLogNeedData should have ensured all needed data was read */
+				Assert (state->readLen >= request_len);
+
+				memcpy(state->readRecordBuf + state->recordGotLen,
+					   (char *) contdata, record_len);
+				state->recordGotLen += record_len;
+				state->recordRemainLen -= record_len;
+
+				/* If we just reassembled the record header, validate it. */
+				if (!state->record_verified)
+				{
+					Assert(state->recordGotLen >= SizeOfXLogRecord);
+					if (!ValidXLogRecordHeader(state, state->ReadRecPtr,
+											   state->PrevRecPtr,
+											   (XLogRecord *) state->readRecordBuf))
+						goto err;
+
+					state->record_verified = true;
+				}
+
+				/* Calculate pointer to beginning of next page, and continue */
+				state->recordContRecPtr += XLOG_BLCKSZ;
+			}
+
+			/* targetPagePtr is pointing the last-read page here */
+			prec = (XLogRecord *) state->readRecordBuf;
+			if (!ValidXLogRecord(state, prec, state->ReadRecPtr))
+				goto err;
+
+			pageHeaderSize =
+				XLogPageHeaderSize((XLogPageHeader) state->readBuf);
+			state->EndRecPtr = targetPagePtr + pageHeaderSize
+				+ MAXALIGN(pageHeader->xlp_rem_len);
+
+			*record = prec;
+			state->readRecordState = XLREAD_NEXT_RECORD;
+			break;
+		}
 	}
 
 	/*
 	 * Special processing if it's an XLOG SWITCH record
 	 */
-	if (record->xl_rmid == RM_XLOG_ID &&
-		(record->xl_info & ~XLR_INFO_MASK) == XLOG_SWITCH)
+	if ((*record)->xl_rmid == RM_XLOG_ID &&
+		((*record)->xl_info & ~XLR_INFO_MASK) == XLOG_SWITCH)
 	{
 		/* Pretend it extends to end of segment */
 		state->EndRecPtr += state->segcxt.ws_segsize - 1;
 		state->EndRecPtr -= XLogSegmentOffset(state->EndRecPtr, state->segcxt.ws_segsize);
 	}
 
-	if (DecodeXLogRecord(state, record, errormsg))
-		return record;
-	else
-		return NULL;
+	Assert (!*record || state->readLen >= 0);
+	if (DecodeXLogRecord(state, *record, errormsg))
+		return XLREAD_SUCCESS;
+
+	*record = NULL;
+	return XLREAD_FAIL;
 
 err:
 
 	/*
-	 * Invalidate the read state. We might read from a different source after
+	 * Invalidate the read page. We might read from a different source after
 	 * failure.
 	 */
 	XLogReaderInvalReadState(state);
@@ -572,7 +706,8 @@ err:
 	if (state->errormsg_buf[0] != '\0')
 		*errormsg = state->errormsg_buf;
 
-	return NULL;
+	*record = NULL;
+	return XLREAD_FAIL;
 }
 
 /*
@@ -724,11 +859,12 @@ XLogReaderInvalReadState(XLogReaderState *state)
  *
  * This is just a convenience subroutine to avoid duplicated code in
  * XLogReadRecord.  It's not intended for use from anywhere else.
+ *
+ * If PrevRecPtr is valid, the xl_prev is is cross-checked with it.
  */
 static bool
 ValidXLogRecordHeader(XLogReaderState *state, XLogRecPtr RecPtr,
-					  XLogRecPtr PrevRecPtr, XLogRecord *record,
-					  bool randAccess)
+					  XLogRecPtr PrevRecPtr, XLogRecord *record)
 {
 	if (record->xl_tot_len < SizeOfXLogRecord)
 	{
@@ -746,7 +882,7 @@ ValidXLogRecordHeader(XLogReaderState *state, XLogRecPtr RecPtr,
 							  (uint32) RecPtr);
 		return false;
 	}
-	if (randAccess)
+	if (PrevRecPtr == InvalidXLogRecPtr)
 	{
 		/*
 		 * We can't exactly verify the prev-link, but surely it should be less
@@ -978,11 +1114,14 @@ XLogReaderValidatePageHeader(XLogReaderState *state, XLogRecPtr recptr,
  * XLogReadRecord() will read the next valid record.
  */
 XLogRecPtr
-XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
+XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr,
+				   XLogFindNextRecordCB read_page, void *private)
 {
 	XLogRecPtr	tmpRecPtr;
 	XLogRecPtr	found = InvalidXLogRecPtr;
 	XLogPageHeader header;
+	XLogRecord *record;
+	XLogReadRecordResult result;
 	char	   *errormsg;
 
 	Assert(!XLogRecPtrIsInvalid(RecPtr));
@@ -1015,9 +1154,7 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
 		while(XLogNeedData(state, targetPagePtr, targetRecOff,
 						   targetRecOff != 0))
 		{
-			if (!state->routine.page_read(state, state->readPagePtr,
-										  state->readLen,
-										  state->ReadRecPtr, state->readBuf))
+			if (!read_page(state, private))
 				break;
 		}
 
@@ -1069,8 +1206,16 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
 	 * or we just jumped over the remaining data of a continuation.
 	 */
 	XLogBeginRead(state, tmpRecPtr);
-	while (XLogReadRecord(state, &errormsg) != NULL)
+	while ((result = XLogReadRecord(state, &record, &errormsg)) !=
+		   XLREAD_FAIL)
 	{
+		if (result == XLREAD_NEED_DATA)
+		{
+			if (!read_page(state, private))
+				goto err;
+			continue;
+		}
+
 		/* past the record we've found, break out */
 		if (RecPtr <= state->ReadRecPtr)
 		{
@@ -1090,9 +1235,9 @@ err:
 #endif							/* FRONTEND */
 
 /*
- * Helper function to ease writing of XLogRoutine->page_read callbacks.
- * If this function is used, caller must supply a segment_open callback in
- * 'state', as that is used here.
+ * Helper function to ease writing of page_read callback.
+ * If this function is used, caller must supply a segment_open callback and
+ * segment_close callback as that is used here.
  *
  * Read 'count' bytes into 'buf', starting at location 'startptr', from WAL
  * fetched from timeline 'tli'.
@@ -1105,6 +1250,7 @@ err:
  */
 bool
 WALRead(XLogReaderState *state,
+		WALSegmentOpenCB segopenfn, WALSegmentCloseCB segclosefn,
 		char *buf, XLogRecPtr startptr, Size count, TimeLineID tli,
 		WALReadError *errinfo)
 {
@@ -1136,10 +1282,10 @@ WALRead(XLogReaderState *state,
 			XLogSegNo	nextSegNo;
 
 			if (state->seg.ws_file >= 0)
-				state->routine.segment_close(state);
+				segclosefn(state);
 
 			XLByteToSeg(recptr, nextSegNo, state->segcxt.ws_segsize);
-			state->routine.segment_open(state, nextSegNo, &tli);
+			segopenfn(state, nextSegNo, &tli);
 
 			/* This shouldn't happen -- indicates a bug in segment_open */
 			Assert(state->seg.ws_file >= 0);
diff --git a/src/backend/access/transam/xlogutils.c b/src/backend/access/transam/xlogutils.c
index 1b10b6df20..303333571f 100644
--- a/src/backend/access/transam/xlogutils.c
+++ b/src/backend/access/transam/xlogutils.c
@@ -689,8 +689,7 @@ XLogTruncateRelation(RelFileNode rnode, ForkNumber forkNum,
 void
 XLogReadDetermineTimeline(XLogReaderState *state, XLogRecPtr wantPage, uint32 wantLength)
 {
-	const XLogRecPtr lastReadPage = state->seg.ws_segno *
-	state->segcxt.ws_segsize + state->readLen;
+	const XLogRecPtr lastReadPage = state->readPagePtr;
 
 	Assert(wantPage != InvalidXLogRecPtr && wantPage % XLOG_BLCKSZ == 0);
 	Assert(wantLength <= XLOG_BLCKSZ);
@@ -705,7 +704,7 @@ XLogReadDetermineTimeline(XLogReaderState *state, XLogRecPtr wantPage, uint32 wa
 	 * current TLI has since become historical.
 	 */
 	if (lastReadPage == wantPage &&
-		state->readLen != 0 &&
+		state->page_verified &&
 		lastReadPage + state->readLen >= wantPage + Min(wantLength, XLOG_BLCKSZ - 1))
 		return;
 
@@ -792,6 +791,7 @@ wal_segment_open(XLogReaderState *state, XLogSegNo nextSegNo,
 	char		path[MAXPGPATH];
 
 	XLogFilePath(path, tli, nextSegNo, state->segcxt.ws_segsize);
+	elog(LOG, "HOGE: %lu, %d => %s", nextSegNo, tli, path);
 	state->seg.ws_file = BasicOpenFile(path, O_RDONLY | PG_BINARY);
 	if (state->seg.ws_file >= 0)
 		return;
@@ -829,9 +829,11 @@ wal_segment_close(XLogReaderState *state)
  * loop for now.
  */
 bool
-read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
-					 int reqLen, XLogRecPtr targetRecPtr, char *cur_page)
+read_local_xlog_page(XLogReaderState *state)
 {
+	XLogRecPtr	targetPagePtr = state->readPagePtr;
+	int			reqLen		  = state->readLen;
+	char	   *cur_page	  = state->readBuf;
 	XLogRecPtr	read_upto,
 				loc;
 	TimeLineID	tli;
@@ -944,11 +946,12 @@ read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
 	 * as 'count', read the whole page anyway. It's guaranteed to be
 	 * zero-padded up to the page boundary if it's incomplete.
 	 */
-	if (!WALRead(state, cur_page, targetPagePtr, XLOG_BLCKSZ, tli,
-				 &errinfo))
+	if (!WALRead(state, wal_segment_open, wal_segment_close,
+				 cur_page, targetPagePtr, XLOG_BLCKSZ, tli, &errinfo))
 		WALReadRaiseError(&errinfo);
 
 	/* number of valid bytes in the buffer */
+	state->readPagePtr = targetPagePtr;
 	state->readLen = count;
 	return true;
 }
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 61902be3b0..a15b0b3355 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -120,7 +120,8 @@ StartupDecodingContext(List *output_plugin_options,
 					   TransactionId xmin_horizon,
 					   bool need_full_snapshot,
 					   bool fast_forward,
-					   XLogReaderRoutine *xl_routine,
+					   LogicalDecodingXLogPageReadCB page_read,
+					   WALSegmentCleanupCB cleanup_cb,
 					   LogicalOutputPluginWriterPrepareWrite prepare_write,
 					   LogicalOutputPluginWriterWrite do_write,
 					   LogicalOutputPluginWriterUpdateProgress update_progress)
@@ -169,11 +170,12 @@ StartupDecodingContext(List *output_plugin_options,
 
 	ctx->slot = slot;
 
-	ctx->reader = XLogReaderAllocate(wal_segment_size, NULL, xl_routine, ctx);
+	ctx->reader = XLogReaderAllocate(wal_segment_size, NULL, cleanup_cb);
 	if (!ctx->reader)
 		ereport(ERROR,
 				(errcode(ERRCODE_OUT_OF_MEMORY),
 				 errmsg("out of memory")));
+	ctx->page_read = page_read;
 
 	ctx->reorder = ReorderBufferAllocate();
 	ctx->snapshot_builder =
@@ -231,7 +233,8 @@ CreateInitDecodingContext(char *plugin,
 						  List *output_plugin_options,
 						  bool need_full_snapshot,
 						  XLogRecPtr restart_lsn,
-						  XLogReaderRoutine *xl_routine,
+						  LogicalDecodingXLogPageReadCB page_read,
+						  WALSegmentCleanupCB cleanup_cb,
 						  LogicalOutputPluginWriterPrepareWrite prepare_write,
 						  LogicalOutputPluginWriterWrite do_write,
 						  LogicalOutputPluginWriterUpdateProgress update_progress)
@@ -328,7 +331,7 @@ CreateInitDecodingContext(char *plugin,
 
 	ctx = StartupDecodingContext(NIL, restart_lsn, xmin_horizon,
 								 need_full_snapshot, false,
-								 xl_routine, prepare_write, do_write,
+								 page_read, cleanup_cb, prepare_write, do_write,
 								 update_progress);
 
 	/* call output plugin initialization callback */
@@ -376,7 +379,8 @@ LogicalDecodingContext *
 CreateDecodingContext(XLogRecPtr start_lsn,
 					  List *output_plugin_options,
 					  bool fast_forward,
-					  XLogReaderRoutine *xl_routine,
+					  LogicalDecodingXLogPageReadCB page_read,
+					  WALSegmentCleanupCB cleanup_cb,
 					  LogicalOutputPluginWriterPrepareWrite prepare_write,
 					  LogicalOutputPluginWriterWrite do_write,
 					  LogicalOutputPluginWriterUpdateProgress update_progress)
@@ -429,8 +433,8 @@ CreateDecodingContext(XLogRecPtr start_lsn,
 
 	ctx = StartupDecodingContext(output_plugin_options,
 								 start_lsn, InvalidTransactionId, false,
-								 fast_forward, xl_routine, prepare_write,
-								 do_write, update_progress);
+								 fast_forward, page_read, cleanup_cb,
+								 prepare_write, do_write, update_progress);
 
 	/* call output plugin initialization callback */
 	old_context = MemoryContextSwitchTo(ctx->context);
@@ -483,7 +487,13 @@ DecodingContextFindStartpoint(LogicalDecodingContext *ctx)
 		char	   *err = NULL;
 
 		/* the read_page callback waits for new WAL */
-		record = XLogReadRecord(ctx->reader, &err);
+		while (XLogReadRecord(ctx->reader, &record, &err) ==
+			   XLREAD_NEED_DATA)
+		{
+			if (!ctx->page_read(ctx->reader))
+				break;
+		}
+
 		if (err)
 			elog(ERROR, "%s", err);
 		if (!record)
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index b99c94e848..fcc81b358e 100644
--- a/src/backend/replication/logical/logicalfuncs.c
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -233,9 +233,8 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
 		ctx = CreateDecodingContext(InvalidXLogRecPtr,
 									options,
 									false,
-									XL_ROUTINE(.page_read = read_local_xlog_page,
-											   .segment_open = wal_segment_open,
-											   .segment_close = wal_segment_close),
+									read_local_xlog_page,
+									wal_segment_close,
 									LogicalOutputPrepareWrite,
 									LogicalOutputWrite, NULL);
 
@@ -284,7 +283,13 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
 			XLogRecord *record;
 			char	   *errm = NULL;
 
-			record = XLogReadRecord(ctx->reader, &errm);
+			while (XLogReadRecord(ctx->reader, &record, &errm) ==
+				   XLREAD_NEED_DATA)
+			{
+				if (!ctx->page_read(ctx->reader))
+					break;
+			}
+
 			if (errm)
 				elog(ERROR, "%s", errm);
 
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 88033a79b2..8a678074d5 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -152,9 +152,8 @@ create_logical_replication_slot(char *name, char *plugin,
 	ctx = CreateInitDecodingContext(plugin, NIL,
 									false,	/* just catalogs is OK */
 									restart_lsn,
-									XL_ROUTINE(.page_read = read_local_xlog_page,
-											   .segment_open = wal_segment_open,
-											   .segment_close = wal_segment_close),
+									read_local_xlog_page,
+									wal_segment_close,
 									NULL, NULL, NULL);
 
 	/*
@@ -486,9 +485,8 @@ pg_logical_replication_slot_advance(XLogRecPtr moveto)
 		ctx = CreateDecodingContext(InvalidXLogRecPtr,
 									NIL,
 									true,	/* fast_forward */
-									XL_ROUTINE(.page_read = read_local_xlog_page,
-											   .segment_open = wal_segment_open,
-											   .segment_close = wal_segment_close),
+									read_local_xlog_page,
+									wal_segment_close,
 									NULL, NULL, NULL);
 
 		/*
@@ -513,7 +511,13 @@ pg_logical_replication_slot_advance(XLogRecPtr moveto)
 			 * Read records.  No changes are generated in fast_forward mode,
 			 * but snapbuilder/slot statuses are updated properly.
 			 */
-			record = XLogReadRecord(ctx->reader, &errm);
+			while (XLogReadRecord(ctx->reader, &record, &errm) ==
+				   XLREAD_NEED_DATA)
+			{
+				if (!ctx->page_read(ctx->reader))
+					break;
+			}
+
 			if (errm)
 				elog(ERROR, "%s", errm);
 
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index bf2711bddd..2fd24a9b14 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -580,10 +580,7 @@ StartReplication(StartReplicationCmd *cmd)
 
 	/* create xlogreader for physical replication */
 	xlogreader =
-		XLogReaderAllocate(wal_segment_size, NULL,
-						   XL_ROUTINE(.segment_open = WalSndSegmentOpen,
-									  .segment_close = wal_segment_close),
-						   NULL);
+		XLogReaderAllocate(wal_segment_size, NULL, wal_segment_close);
 
 	if (!xlogreader)
 		ereport(ERROR,
@@ -812,9 +809,11 @@ StartReplication(StartReplicationCmd *cmd)
  * set every time WAL is flushed.
  */
 static bool
-logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen,
-					   XLogRecPtr targetRecPtr, char *cur_page)
+logical_read_xlog_page(XLogReaderState *state)
 {
+	XLogRecPtr		targetPagePtr = state->readPagePtr;
+	int				reqLen		  = state->readLen;
+	char		   *cur_page	  = state->readBuf;
 	XLogRecPtr	flushptr;
 	int			count;
 	WALReadError errinfo;
@@ -842,7 +841,7 @@ logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int req
 		count = flushptr - targetPagePtr;	/* part of the page available */
 
 	/* now actually read the data, we know it's there */
-	if (!WALRead(state,
+	if (!WALRead(state, WalSndSegmentOpen, wal_segment_close,
 				 cur_page,
 				 targetPagePtr,
 				 XLOG_BLCKSZ,
@@ -1014,9 +1013,8 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
 
 		ctx = CreateInitDecodingContext(cmd->plugin, NIL, need_full_snapshot,
 										InvalidXLogRecPtr,
-										XL_ROUTINE(.page_read = logical_read_xlog_page,
-												   .segment_open = WalSndSegmentOpen,
-												   .segment_close = wal_segment_close),
+										logical_read_xlog_page,
+										wal_segment_close,
 										WalSndPrepareWrite, WalSndWriteData,
 										WalSndUpdateProgress);
 
@@ -1179,9 +1177,8 @@ StartLogicalReplication(StartReplicationCmd *cmd)
 	 */
 	logical_decoding_ctx =
 		CreateDecodingContext(cmd->startpoint, cmd->options, false,
-							  XL_ROUTINE(.page_read = logical_read_xlog_page,
-										 .segment_open = WalSndSegmentOpen,
-										 .segment_close = wal_segment_close),
+							  logical_read_xlog_page,
+							  wal_segment_close,
 							  WalSndPrepareWrite, WalSndWriteData,
 							  WalSndUpdateProgress);
 	xlogreader = logical_decoding_ctx->reader;
@@ -2761,7 +2758,7 @@ XLogSendPhysical(void)
 	enlargeStringInfo(&output_message, nbytes);
 
 retry:
-	if (!WALRead(xlogreader,
+	if (!WALRead(xlogreader, WalSndSegmentOpen, wal_segment_close,
 				 &output_message.data[output_message.len],
 				 startptr,
 				 nbytes,
@@ -2859,7 +2856,12 @@ XLogSendLogical(void)
 	 */
 	WalSndCaughtUp = false;
 
-	record = XLogReadRecord(logical_decoding_ctx->reader, &errm);
+	while (XLogReadRecord(logical_decoding_ctx->reader, &record, &errm) ==
+		   XLREAD_NEED_DATA)
+	{
+		if (!logical_decoding_ctx->page_read(logical_decoding_ctx->reader))
+			break;
+	}
 
 	/* xlog record was invalid */
 	if (errm != NULL)
diff --git a/src/bin/pg_rewind/parsexlog.c b/src/bin/pg_rewind/parsexlog.c
index 0b7e73ae79..3dd6df4be6 100644
--- a/src/bin/pg_rewind/parsexlog.c
+++ b/src/bin/pg_rewind/parsexlog.c
@@ -40,15 +40,9 @@ static int	xlogreadfd = -1;
 static XLogSegNo xlogreadsegno = -1;
 static char xlogfpath[MAXPGPATH];
 
-typedef struct XLogPageReadPrivate
-{
-	const char *restoreCommand;
-	int			tliIndex;
-} XLogPageReadPrivate;
-
-static bool	SimpleXLogPageRead(XLogReaderState *xlogreader,
-							   XLogRecPtr targetPagePtr,
-							   int reqLen, XLogRecPtr targetRecPtr, char *readBuf);
+static bool SimpleXLogPageRead(XLogReaderState *xlogreader,
+							   const char *datadir, int *tliIndex,
+							   const char *restoreCommand);
 
 /*
  * Read WAL from the datadir/pg_wal, starting from 'startpoint' on timeline
@@ -62,20 +56,22 @@ extractPageMap(const char *datadir, XLogRecPtr startpoint, int tliIndex,
 	XLogRecord *record;
 	XLogReaderState *xlogreader;
 	char	   *errormsg;
-	XLogPageReadPrivate private;
 
-	private.tliIndex = tliIndex;
-	private.restoreCommand = restoreCommand;
-	xlogreader = XLogReaderAllocate(WalSegSz, datadir,
-									XL_ROUTINE(.page_read = &SimpleXLogPageRead),
-									&private);
+	xlogreader = XLogReaderAllocate(WalSegSz, datadir, NULL);
+
 	if (xlogreader == NULL)
 		pg_fatal("out of memory");
 
 	XLogBeginRead(xlogreader, startpoint);
 	do
 	{
-		record = XLogReadRecord(xlogreader, &errormsg);
+		while (XLogReadRecord(xlogreader, &record, &errormsg) ==
+			   XLREAD_NEED_DATA)
+		{
+			if (!SimpleXLogPageRead(xlogreader, datadir,
+									&tliIndex, restoreCommand))
+				break;
+		}
 
 		if (record == NULL)
 		{
@@ -113,19 +109,19 @@ readOneRecord(const char *datadir, XLogRecPtr ptr, int tliIndex,
 	XLogRecord *record;
 	XLogReaderState *xlogreader;
 	char	   *errormsg;
-	XLogPageReadPrivate private;
 	XLogRecPtr	endptr;
 
-	private.tliIndex = tliIndex;
-	private.restoreCommand = restoreCommand;
-	xlogreader = XLogReaderAllocate(WalSegSz, datadir,
-									XL_ROUTINE(.page_read = &SimpleXLogPageRead),
-									&private);
+	xlogreader = XLogReaderAllocate(WalSegSz, datadir, NULL);
 	if (xlogreader == NULL)
 		pg_fatal("out of memory");
 
 	XLogBeginRead(xlogreader, ptr);
-	record = XLogReadRecord(xlogreader, &errormsg);
+	while (XLogReadRecord(xlogreader, &record, &errormsg) ==
+		   XLREAD_NEED_DATA)
+	{
+		if (!SimpleXLogPageRead(xlogreader, datadir, &tliIndex, restoreCommand))
+			break;
+	}
 	if (record == NULL)
 	{
 		if (errormsg)
@@ -160,7 +156,6 @@ findLastCheckpoint(const char *datadir, XLogRecPtr forkptr, int tliIndex,
 	XLogRecPtr	searchptr;
 	XLogReaderState *xlogreader;
 	char	   *errormsg;
-	XLogPageReadPrivate private;
 
 	/*
 	 * The given fork pointer points to the end of the last common record,
@@ -176,11 +171,7 @@ findLastCheckpoint(const char *datadir, XLogRecPtr forkptr, int tliIndex,
 			forkptr += SizeOfXLogShortPHD;
 	}
 
-	private.tliIndex = tliIndex;
-	private.restoreCommand = restoreCommand;
-	xlogreader = XLogReaderAllocate(WalSegSz, datadir,
-									XL_ROUTINE(.page_read = &SimpleXLogPageRead),
-									&private);
+	xlogreader = XLogReaderAllocate(WalSegSz, datadir, NULL);
 	if (xlogreader == NULL)
 		pg_fatal("out of memory");
 
@@ -190,7 +181,13 @@ findLastCheckpoint(const char *datadir, XLogRecPtr forkptr, int tliIndex,
 		uint8		info;
 
 		XLogBeginRead(xlogreader, searchptr);
-		record = XLogReadRecord(xlogreader, &errormsg);
+		while (XLogReadRecord(xlogreader, &record, &errormsg) ==
+			   XLREAD_NEED_DATA)
+		{
+			if (!SimpleXLogPageRead(xlogreader, datadir,
+									&tliIndex, restoreCommand))
+				break;
+		}
 
 		if (record == NULL)
 		{
@@ -237,10 +234,11 @@ findLastCheckpoint(const char *datadir, XLogRecPtr forkptr, int tliIndex,
 
 /* XLogReader callback function, to read a WAL page */
 static bool
-SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
-				   int reqLen, XLogRecPtr targetRecPtr, char *readBuf)
+SimpleXLogPageRead(XLogReaderState *xlogreader, const char *datadir,
+				   int *tliIndex, const char *restoreCommand)
 {
-	XLogPageReadPrivate *private = (XLogPageReadPrivate *) xlogreader->private_data;
+	XLogRecPtr	targetPagePtr = xlogreader->readPagePtr;
+	char	   *readBuf		  = xlogreader->readBuf;
 	uint32		targetPageOff;
 	XLogRecPtr	targetSegEnd;
 	XLogSegNo	targetSegNo;
@@ -273,14 +271,14 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
 		 * be done both forward and backward, consider also switching timeline
 		 * accordingly.
 		 */
-		while (private->tliIndex < targetNentries - 1 &&
-			   targetHistory[private->tliIndex].end < targetSegEnd)
-			private->tliIndex++;
-		while (private->tliIndex > 0 &&
-			   targetHistory[private->tliIndex].begin >= targetSegEnd)
-			private->tliIndex--;
+		while (*tliIndex < targetNentries - 1 &&
+			   targetHistory[*tliIndex].end < targetSegEnd)
+			(*tliIndex)++;
+		while (*tliIndex > 0 &&
+			   targetHistory[*tliIndex].begin >= targetSegEnd)
+			(*tliIndex)--;
 
-		XLogFileName(xlogfname, targetHistory[private->tliIndex].tli,
+		XLogFileName(xlogfname, targetHistory[*tliIndex].tli,
 					 xlogreadsegno, WalSegSz);
 
 		snprintf(xlogfpath, MAXPGPATH, "%s/" XLOGDIR "/%s",
@@ -293,7 +291,7 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
 			/*
 			 * If we have no restore_command to execute, then exit.
 			 */
-			if (private->restoreCommand == NULL)
+			if (restoreCommand == NULL)
 			{
 				pg_log_error("could not open file \"%s\": %m", xlogfpath);
 				xlogreader->readLen = -1;
@@ -307,7 +305,7 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
 			xlogreadfd = RestoreArchivedFile(xlogreader->segcxt.ws_dir,
 											 xlogfname,
 											 WalSegSz,
-											 private->restoreCommand);
+											 restoreCommand);
 
 			if (xlogreadfd < 0)
 			{
@@ -349,7 +347,7 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
 
 	Assert(targetSegNo == xlogreadsegno);
 
-	xlogreader->seg.ws_tli = targetHistory[private->tliIndex].tli;
+	xlogreader->seg.ws_tli = targetHistory[*tliIndex].tli;
 	xlogreader->readLen = XLOG_BLCKSZ;
 	return true;
 }
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index ab0c28d9b7..02e71138cc 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -330,12 +330,17 @@ WALDumpCloseSegment(XLogReaderState *state)
 	state->seg.ws_file = -1;
 }
 
-/* pg_waldump's XLogReaderRoutine->page_read callback */
+/*
+ * pg_waldump's WAL page rader, also used as page_read callback for
+ * XLogFindNextRecord
+ */
 static bool
-WALDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen,
-				XLogRecPtr targetPtr, char *readBuff)
+WALDumpReadPage(XLogReaderState *state, void *priv)
 {
-	XLogDumpPrivate *private = state->private_data;
+	XLogRecPtr	targetPagePtr = state->readPagePtr;
+	int			reqLen		  = state->readLen;
+	char	   *readBuff	  = state->readBuf;
+	XLogDumpPrivate *private  = (XLogDumpPrivate *) priv;
 	int			count = XLOG_BLCKSZ;
 	WALReadError errinfo;
 
@@ -353,8 +358,8 @@ WALDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen,
 		}
 	}
 
-	if (!WALRead(state, readBuff, targetPagePtr, count, private->timeline,
-				 &errinfo))
+	if (!WALRead(state, WALDumpOpenSegment, WALDumpCloseSegment,
+				 readBuff, targetPagePtr, count, private->timeline, &errinfo))
 	{
 		WALOpenSegment *seg = &errinfo.wre_seg;
 		char		fname[MAXPGPATH];
@@ -374,6 +379,7 @@ WALDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen,
 						(Size) errinfo.wre_req);
 	}
 
+	Assert(count >= state->readLen);
 	state->readLen = count;
 	return true;
 }
@@ -1042,16 +1048,14 @@ main(int argc, char **argv)
 
 	/* we have everything we need, start reading */
 	xlogreader_state =
-		XLogReaderAllocate(WalSegSz, waldir,
-						   XL_ROUTINE(.page_read = WALDumpReadPage,
-									  .segment_open = WALDumpOpenSegment,
-									  .segment_close = WALDumpCloseSegment),
-						   &private);
+		XLogReaderAllocate(WalSegSz, waldir, WALDumpCloseSegment);
+
 	if (!xlogreader_state)
 		fatal_error("out of memory");
 
 	/* first find a valid recptr to start from */
-	first_record = XLogFindNextRecord(xlogreader_state, private.startptr);
+	first_record = XLogFindNextRecord(xlogreader_state, private.startptr,
+									  &WALDumpReadPage, (void*) &private);
 
 	if (first_record == InvalidXLogRecPtr)
 		fatal_error("could not find a valid record after %X/%X",
@@ -1075,7 +1079,13 @@ main(int argc, char **argv)
 	for (;;)
 	{
 		/* try to read the next record */
-		record = XLogReadRecord(xlogreader_state, &errormsg);
+		while (XLogReadRecord(xlogreader_state, &record, &errormsg) ==
+			   XLREAD_NEED_DATA)
+		{
+			if (!WALDumpReadPage(xlogreader_state, (void *) &private))
+				break;
+		}
+
 		if (!record)
 		{
 			if (!config.follow || private.endptr_reached)
diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h
index 6c8848e14f..6a7fe140cb 100644
--- a/src/include/access/xlogreader.h
+++ b/src/include/access/xlogreader.h
@@ -57,64 +57,15 @@ typedef struct WALSegmentContext
 
 typedef struct XLogReaderState XLogReaderState;
 
-/* Function type definition for the read_page callback */
-typedef bool (*XLogPageReadCB) (XLogReaderState *xlogreader,
-							   XLogRecPtr targetPagePtr,
-							   int reqLen,
-							   XLogRecPtr targetRecPtr,
-							   char *readBuf);
+/* Function type definition for the segment cleanup callback */
+typedef void (*WALSegmentCleanupCB) (XLogReaderState *xlogreader);
+
+/* Function type definition for the open/close callbacks for WALRead() */
 typedef void (*WALSegmentOpenCB) (XLogReaderState *xlogreader,
 								  XLogSegNo nextSegNo,
 								  TimeLineID *tli_p);
 typedef void (*WALSegmentCloseCB) (XLogReaderState *xlogreader);
 
-typedef struct XLogReaderRoutine
-{
-	/*
-	 * Data input callback
-	 *
-	 * This callback shall read at least reqLen valid bytes of the xlog page
-	 * starting at targetPagePtr, and store them in readBuf.  The callback
-	 * shall return the number of bytes read (never more than XLOG_BLCKSZ), or
-	 * -1 on failure.  The callback shall sleep, if necessary, to wait for the
-	 * requested bytes to become available.  The callback will not be invoked
-	 * again for the same page unless more than the returned number of bytes
-	 * are needed.
-	 *
-	 * targetRecPtr is the position of the WAL record we're reading.  Usually
-	 * it is equal to targetPagePtr + reqLen, but sometimes xlogreader needs
-	 * to read and verify the page or segment header, before it reads the
-	 * actual WAL record it's interested in.  In that case, targetRecPtr can
-	 * be used to determine which timeline to read the page from.
-	 *
-	 * The callback shall set ->seg.ws_tli to the TLI of the file the page was
-	 * read from.
-	 */
-	XLogPageReadCB page_read;
-
-	/*
-	 * Callback to open the specified WAL segment for reading.  ->seg.ws_file
-	 * shall be set to the file descriptor of the opened segment.  In case of
-	 * failure, an error shall be raised by the callback and it shall not
-	 * return.
-	 *
-	 * "nextSegNo" is the number of the segment to be opened.
-	 *
-	 * "tli_p" is an input/output argument. WALRead() uses it to pass the
-	 * timeline in which the new segment should be found, but the callback can
-	 * use it to return the TLI that it actually opened.
-	 */
-	WALSegmentOpenCB segment_open;
-
-	/*
-	 * WAL segment close callback.  ->seg.ws_file shall be set to a negative
-	 * number.
-	 */
-	WALSegmentCloseCB segment_close;
-} XLogReaderRoutine;
-
-#define XL_ROUTINE(...) &(XLogReaderRoutine){__VA_ARGS__}
-
 typedef struct
 {
 	/* Is this block ref in use? */
@@ -144,12 +95,35 @@ typedef struct
 	uint16		data_bufsz;
 } DecodedBkpBlock;
 
+/* Return code from XLogReadRecord */
+typedef enum XLogReadRecordResult
+{
+	XLREAD_SUCCESS,		/* record is successfully read */
+	XLREAD_NEED_DATA,	/* need more data. see XLogReadRecord. */
+	XLREAD_FAIL			/* failed during reading a record */
+} XLogReadRecordResult;
+
+/*
+ * internal state of XLogReadRecord
+ *
+ * XLogReadState runs a state machine while reading a record. Theses states
+ * are not seen outside the function. Each state may repeat several times
+ * exiting requesting caller for new data. See the comment of XLogReadRecrod
+ * for details.
+ */
+typedef enum XLogReadRecordState {
+	XLREAD_NEXT_RECORD,
+	XLREAD_TOT_LEN,
+	XLREAD_FIRST_FRAGMENT,
+	XLREAD_CONTINUATION
+} XLogReadRecordState;
+
 struct XLogReaderState
 {
 	/*
 	 * Operational callbacks
 	 */
-	XLogReaderRoutine routine;
+	WALSegmentCleanupCB cleanup_cb;
 
 	/* ----------------------------------------
 	 * Public parameters
@@ -162,18 +136,14 @@ struct XLogReaderState
 	 */
 	uint64		system_identifier;
 
-	/*
-	 * Opaque data for callbacks to use.  Not used by XLogReader.
-	 */
-	void	   *private_data;
-
 	/*
 	 * Start and end point of last record read.  EndRecPtr is also used as the
 	 * position to read next.  Calling XLogBeginRead() sets EndRecPtr to the
 	 * starting position and ReadRecPtr to invalid.
 	 */
-	XLogRecPtr	ReadRecPtr;		/* start of last record read */
+	XLogRecPtr	ReadRecPtr;		/* start of last record read or being read */
 	XLogRecPtr	EndRecPtr;		/* end+1 of last record read */
+	XLogRecPtr	PrevRecPtr;		/* start of previous record read */
 
 	/* ----------------------------------------
 	 * Communication with page reader
@@ -187,7 +157,9 @@ struct XLogReaderState
 								 * the request, or -1 on error */
 	TimeLineID	readPageTLI;	/* TLI for data currently in readBuf */
 	char	   *readBuf;		/* buffer to store data */
-	bool		page_verified;  /* is the page on the buffer verified? */
+	bool		page_verified;  /* is the page header on the buffer verified? */
+	bool		record_verified;/* is the current record header verified? */
+

 	/* ----------------------------------------
@@ -227,8 +199,6 @@ struct XLogReaderState
 	XLogRecPtr	latestPagePtr;
 	TimeLineID	latestPageTLI;
 
-	/* beginning of the WAL record being read. */
-	XLogRecPtr	currRecPtr;
 	/* timeline to read it from, 0 if a lookup is required */
 	TimeLineID	currTLI;
 
@@ -255,6 +225,15 @@ struct XLogReaderState
 	char	   *readRecordBuf;
 	uint32		readRecordBufSize;
 
+	/*
+	 * XLogReadRecord() state
+	 */
+	XLogReadRecordState	readRecordState;/* state machine state */
+	int			recordGotLen;		/* amount of current record that has
+									 * already been read */
+	int			recordRemainLen;	/* length of current record that remains */
+	XLogRecPtr	recordContRecPtr;	/* where the current record continues */
+
 	/* Buffer to hold error message */
 	char	   *errormsg_buf;
 };
@@ -262,9 +241,7 @@ struct XLogReaderState
 /* Get a new XLogReader */
 extern XLogReaderState *XLogReaderAllocate(int wal_segment_size,
 										   const char *waldir,
-										   XLogReaderRoutine *routine,
-										   void *private_data);
-extern XLogReaderRoutine *LocalXLogReaderRoutine(void);
+										   WALSegmentCleanupCB cleanup_cb);
 
 /* Free an XLogReader */
 extern void XLogReaderFree(XLogReaderState *state);
@@ -272,12 +249,17 @@ extern void XLogReaderFree(XLogReaderState *state);
 /* Position the XLogReader to given record */
 extern void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr);
 #ifdef FRONTEND
-extern XLogRecPtr XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr);
+/* Function type definition for the read_page callback */
+typedef bool (*XLogFindNextRecordCB) (XLogReaderState *xlogreader,
+									  void *private);
+extern XLogRecPtr XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr,
+									 XLogFindNextRecordCB read_page, void *private);
 #endif							/* FRONTEND */
 
 /* Read the next XLog record. Returns NULL on end-of-WAL or failure */
-extern struct XLogRecord *XLogReadRecord(XLogReaderState *state,
-										 char **errormsg);
+extern XLogReadRecordResult XLogReadRecord(XLogReaderState *state,
+										   XLogRecord **record,
+										   char **errormsg);
 
 /* Validate a page */
 extern bool XLogReaderValidatePageHeader(XLogReaderState *state,
@@ -297,6 +279,7 @@ typedef struct WALReadError
 } WALReadError;
 
 extern bool WALRead(XLogReaderState *state,
+					WALSegmentOpenCB segopenfn, WALSegmentCloseCB sgclosefn,
 					char *buf, XLogRecPtr startptr, Size count,
 					TimeLineID tli, WALReadError *errinfo);
 
diff --git a/src/include/access/xlogutils.h b/src/include/access/xlogutils.h
index c287edf206..482e4ced69 100644
--- a/src/include/access/xlogutils.h
+++ b/src/include/access/xlogutils.h
@@ -47,9 +47,7 @@ extern Buffer XLogReadBufferExtended(RelFileNode rnode, ForkNumber forknum,
 extern Relation CreateFakeRelcacheEntry(RelFileNode rnode);
 extern void FreeFakeRelcacheEntry(Relation fakerel);
 
-extern bool	read_local_xlog_page(XLogReaderState *state,
-								 XLogRecPtr targetPagePtr, int reqLen,
-								 XLogRecPtr targetRecPtr, char *cur_page);
+extern bool read_local_xlog_page(XLogReaderState *state);
 extern void wal_segment_open(XLogReaderState *state,
 							 XLogSegNo nextSegNo,
 							 TimeLineID *tli_p);
diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h
index c2f2475e5d..a743db49cf 100644
--- a/src/include/replication/logical.h
+++ b/src/include/replication/logical.h
@@ -29,6 +29,10 @@ typedef void (*LogicalOutputPluginWriterUpdateProgress) (struct LogicalDecodingC
 														 TransactionId xid
 );
 
+typedef struct LogicalDecodingContext LogicalDecodingContext;
+
+typedef bool (*LogicalDecodingXLogPageReadCB)(XLogReaderState *ctx);
+
 typedef struct LogicalDecodingContext
 {
 	/* memory context this is all allocated in */
@@ -39,6 +43,7 @@ typedef struct LogicalDecodingContext
 
 	/* infrastructure pieces for decoding */
 	XLogReaderState *reader;
+	LogicalDecodingXLogPageReadCB page_read;
 	struct ReorderBuffer *reorder;
 	struct SnapBuild *snapshot_builder;
 
@@ -95,14 +100,16 @@ extern LogicalDecodingContext *CreateInitDecodingContext(char *plugin,
 														 List *output_plugin_options,
 														 bool need_full_snapshot,
 														 XLogRecPtr restart_lsn,
-														 XLogReaderRoutine *xl_routine,
+														 LogicalDecodingXLogPageReadCB page_read,
+														 WALSegmentCleanupCB cleanup_cb,
 														 LogicalOutputPluginWriterPrepareWrite prepare_write,
 														 LogicalOutputPluginWriterWrite do_write,
 														 LogicalOutputPluginWriterUpdateProgress update_progress);
 extern LogicalDecodingContext *CreateDecodingContext(XLogRecPtr start_lsn,
 													 List *output_plugin_options,
 													 bool fast_forward,
-													 XLogReaderRoutine *xl_routine,
+													 LogicalDecodingXLogPageReadCB page_read,
+													 WALSegmentCleanupCB cleanup_cb,
 													 LogicalOutputPluginWriterPrepareWrite prepare_write,
 													 LogicalOutputPluginWriterWrite do_write,
 													 LogicalOutputPluginWriterUpdateProgress update_progress);
-- 
2.18.4

----Next_Part(Thu_Jul__2_13_53_30_2020_072)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="v15-0003-Remove-globals-readOff-readLen-and-readSegNo.patch"