v17-0002-Move-page-reader-out-of-XLogReadRecord.patch

From f6ddd723fd67db5045c454f70786fcda98f36fab Mon Sep 17 00:00:00 2001
From: Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
Date: Tue, 10 Sep 2019 12:58:27 +0900
Subject: [PATCH v17 02/10] Move page-reader out of XLogReadRecord().

This is the second step of removing callbacks from the WAL decoder.
XLogReadRecord() return XLREAD_NEED_DATA to indicate that the caller
should supply new data, and the decoder works as a state machine.

Author: Kyotaro HORIGUCHI <horiguchi.kyotaro@lab.ntt.co.jp>
Author: Heikki Linnakangas <hlinnaka@iki.fi>
Reviewed-by: Antonin Houska <ah@cybertec.at>
Reviewed-by: Alvaro Herrera <alvherre@2ndquadrant.com>
Reviewed-by: Takashi Menjo <takashi.menjo@gmail.com>
Reviewed-by: Thomas Munro <thomas.munro@gmail.com>
Discussion: https://postgr.es/m/20190418.210257.43726183.horiguchi.kyotaro%40lab.ntt.co.jp
---
 src/backend/access/transam/twophase.c         |  14 +-
 src/backend/access/transam/xlog.c             |  58 +-
 src/backend/access/transam/xlogreader.c       | 654 ++++++++++--------
 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                 |  86 ++-
 src/bin/pg_waldump/pg_waldump.c               |  36 +-
 src/include/access/xlogreader.h               | 122 ++--
 src/include/access/xlogutils.h                |   4 +-
 src/include/pg_config_manual.h                |   2 +-
 src/include/replication/logical.h             |  11 +-
 14 files changed, 611 insertions(+), 482 deletions(-)

diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 89335b64a2..3137cb3ecc 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)
 	char	   *errormsg;
 	TimeLineID	save_currtli = ThisTimeLineID;
 
-	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,12 @@ 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;
+	}
 
 	/*
 	 * Restore immediately the timeline where it was previously, as
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 8085ca1117..b7d7e6d31b 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -838,13 +838,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
@@ -920,8 +913,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);
@@ -1234,8 +1227,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)
 		{
@@ -4369,15 +4361,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;
@@ -4385,8 +4372,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)
@@ -6456,7 +6451,6 @@ StartupXLOG(void)
 	bool		backupFromStandby = false;
 	DBState		dbstate_at_startup;
 	XLogReaderState *xlogreader;
-	XLogPageReadPrivate private;
 	bool		promoted = false;
 	struct stat st;
 
@@ -6615,13 +6609,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),
@@ -12107,12 +12097,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;
@@ -12155,8 +12146,8 @@ retry:
 		 flushedUpto < targetPagePtr + reqLen))
 	{
 		if (!WaitForWALToBecomeAvailable(targetPagePtr + reqLen,
-										 private->randAccess,
-										 private->fetching_ckpt,
+										 randAccess,
+										 fetching_ckpt,
 										 targetRecPtr))
 		{
 			if (readFile >= 0)
@@ -12261,6 +12252,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 f2345ab09e..661863e94b 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,12 +253,12 @@ 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.
  *
  * Returns XLREAD_NEED_DATA if more data is needed to finish reading the
  * current record.  In that case, state->readPagePtr and state->readLen inform
@@ -304,307 +303,410 @@ XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
  * 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)
+	switch (state->readRecordState)
 	{
-		/* read the record after the one we just read */
+		case XLREAD_NEXT_RECORD:
+			ResetDecoder(state);
 
-		/*
-		 * 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;
-	}
+			if (state->ReadRecPtr != InvalidXLogRecPtr)
+			{
+				/* read the record after the one we just read */
 
-	state->currRecPtr = RecPtr;
+				/*
+				 * 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;
 
-	targetPagePtr = RecPtr - (RecPtr % XLOG_BLCKSZ);
-	targetRecOff = RecPtr % XLOG_BLCKSZ;
+				/*
+				 * 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 */
+			}
 
-	/*
-	 * 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;
-	}
+			state->record_verified = false;
+			state->readRecordState = XLREAD_TOT_LEN;
+			/* fall through */
 
-	if (!state->page_verified)
-		goto err;
+		case XLREAD_TOT_LEN:
+			{
+				uint32		total_len;
+				uint32		pageHeaderSize;
+				XLogRecPtr	targetPagePtr;
+				uint32		targetRecOff;
+				XLogPageHeader pageHeader;
 
-	/*
-	 * 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",
-							  LSN_FORMAT_ARGS(RecPtr));
-		goto err;
-	}
+				targetPagePtr =
+					state->ReadRecPtr - (state->ReadRecPtr % XLOG_BLCKSZ);
+				targetRecOff = state->ReadRecPtr % XLOG_BLCKSZ;
 
-	if ((((XLogPageHeader) state->readBuf)->xlp_info & XLP_FIRST_IS_CONTRECORD) &&
-		targetRecOff == pageHeaderSize)
-	{
-		report_invalid_record(state, "contrecord is requested by %X/%X",
-							  LSN_FORMAT_ARGS(RecPtr));
-		goto err;
-	}
+				/*
+				 * 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;
 
-	/* XLogNeedData has verified the page header */
-	Assert(pageHeaderSize <= state->readLen);
+				/* error out if caller supplied bogus page */
+				if (!state->page_verified)
+					goto err;
 
-	/*
-	 * 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;
+				/* 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",
+										  LSN_FORMAT_ARGS(state->ReadRecPtr));
+					goto err;
+				}
 
-	/*
-	 * 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",
-								  LSN_FORMAT_ARGS(RecPtr),
-								  (uint32) SizeOfXLogRecord, total_len);
-			goto err;
-		}
-		gotheader = false;
-	}
+				pageHeader = (XLogPageHeader) state->readBuf;
+				if ((pageHeader->xlp_info & XLP_FIRST_IS_CONTRECORD) &&
+					targetRecOff == pageHeaderSize)
+				{
+					report_invalid_record(state, "contrecord is requested by %X/%X",
+										  (uint32) (state->ReadRecPtr >> 32),
+										  (uint32) state->ReadRecPtr);
+					goto err;
+				}
 
-	len = XLOG_BLCKSZ - RecPtr % XLOG_BLCKSZ;
-	if (total_len > len)
-	{
-		/* Need to reassemble record */
-		char	   *contdata;
-		XLogPageHeader pageHeader;
-		char	   *buffer;
-		uint32		gotlen;
+				/* XLogNeedData has verified the page header */
+				Assert(pageHeaderSize <= state->readLen);
 
-		/*
-		 * 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, LSN_FORMAT_ARGS(RecPtr));
-			goto err;
-		}
+				/*
+				 * 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;
 
-		/* 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;
+				/*
+				 * 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;
 
-		do
-		{
-			int			rest_len = total_len - gotlen;
+					state->record_verified = 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",
+											  LSN_FORMAT_ARGS(state->ReadRecPtr),
+											  (uint32) SizeOfXLogRecord, total_len);
+						goto err;
+					}
+				}
 
-			/* Calculate pointer to beginning of next page */
-			targetPagePtr += XLOG_BLCKSZ;
+				/*
+				 * 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 */
 
-			/* Wait for the next page to become available */
-			while (XLogNeedData(state, targetPagePtr,
-								Min(rest_len, XLOG_BLCKSZ),
-								false))
+		case XLREAD_FIRST_FRAGMENT:
 			{
-				if (!state->routine.page_read(state, state->readPagePtr,
-											  state->readLen,
-											  state->ReadRecPtr,
-											  state->readBuf))
-					break;
-			}
+				uint32		total_len = state->recordRemainLen;
+				uint32		request_len;
+				uint32		record_len;
+				XLogRecPtr	targetPagePtr;
+				uint32		targetRecOff;
 
-			if (!state->page_verified)
-				goto err;
+				/*
+				 * 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;
 
-			Assert(SizeOfXLogShortPHD <= state->readLen);
+				request_len = Min(targetRecOff + total_len, XLOG_BLCKSZ);
+				record_len = request_len - targetRecOff;
 
-			/* 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",
-									  LSN_FORMAT_ARGS(RecPtr));
-				goto err;
-			}
+				/* ReadRecPtr contains page header */
+				Assert(targetRecOff != 0);
+				if (XLogNeedData(state, targetPagePtr, request_len, true))
+					return XLREAD_NEED_DATA;
 
-			/*
-			 * 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 (expected %lld) at %X/%X",
-									  pageHeader->xlp_rem_len,
-									  ((long long) total_len) - gotlen,
-									  LSN_FORMAT_ARGS(RecPtr));
-				goto err;
-			}
+				/* error out if caller supplied bogus page */
+				if (!state->page_verified)
+					goto err;
 
-			/* Append the continuation from this page to the buffer */
-			pageHeaderSize = XLogPageHeaderSize(pageHeader);
+				prec = (XLogRecord *) (state->readBuf + targetRecOff);
 
-			Assert(pageHeaderSize <= state->readLen);
+				/* validate record header if not yet */
+				if (!state->record_verified && record_len >= SizeOfXLogRecord)
+				{
+					if (!ValidXLogRecordHeader(state, state->ReadRecPtr,
+											   state->PrevRecPtr, prec))
+						goto err;
 
-			contdata = (char *) state->readBuf + pageHeaderSize;
-			len = XLOG_BLCKSZ - pageHeaderSize;
-			if (pageHeader->xlp_rem_len < len)
-				len = pageHeader->xlp_rem_len;
+					state->record_verified = true;
+				}
 
-			Assert(pageHeaderSize + len <= state->readLen);
-			memcpy(buffer, (char *) contdata, len);
-			buffer += len;
-			gotlen += len;
 
-			/* If we just reassembled the record header, validate it. */
-			if (!gotheader)
-			{
-				record = (XLogRecord *) state->readRecordBuf;
-				if (!ValidXLogRecordHeader(state, RecPtr, state->ReadRecPtr,
-										   record, randAccess))
+				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,
+										  LSN_FORMAT_ARGS(state->ReadRecPtr));
 					goto err;
-				gotheader = true;
-			}
-		} while (gotlen < total_len);
+				}
 
-		Assert(gotheader);
+				/* 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;
 
-		record = (XLogRecord *) state->readRecordBuf;
-		if (!ValidXLogRecord(state, record, RecPtr))
-			goto err;
+				/* Calculate pointer to beginning of next page */
+				state->recordContRecPtr = state->ReadRecPtr + record_len;
+				Assert(state->recordContRecPtr % XLOG_BLCKSZ == 0);
 
-		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))
-		{
-			if (!state->routine.page_read(state, state->readPagePtr,
-										  state->readLen,
-										  state->ReadRecPtr, state->readBuf))
-				break;
-		}
+				state->readRecordState = XLREAD_CONTINUATION;
+			}
+			/* fall through */
 
-		if (!state->page_verified)
-			goto err;
+		case XLREAD_CONTINUATION:
+			{
+				XLogPageHeader pageHeader;
+				uint32		pageHeaderSize;
+				XLogRecPtr	targetPagePtr;
 
-		/* Record does not cross a page boundary */
-		if (!ValidXLogRecord(state, record, RecPtr))
-			goto err;
+				/*
+				 * 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);
+
+					/* 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;
+					}
+
+					/*
+					 * 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;
+					}
+
+					/* Append the continuation from this page to the buffer */
+					pageHeaderSize = XLogPageHeaderSize(pageHeader);
+
+					/*
+					 * 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;
+				}
 
-		state->EndRecPtr = RecPtr + MAXALIGN(total_len);
+				/* 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);
 
-		state->ReadRecPtr = RecPtr;
+				*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);
@@ -612,7 +714,8 @@ err:
 	if (state->errormsg_buf[0] != '\0')
 		*errormsg = state->errormsg_buf;
 
-	return NULL;
+	*record = NULL;
+	return XLREAD_FAIL;
 }
 
 /*
@@ -764,11 +867,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)
 	{
@@ -785,7 +889,7 @@ ValidXLogRecordHeader(XLogReaderState *state, XLogRecPtr RecPtr,
 							  record->xl_rmid, LSN_FORMAT_ARGS(RecPtr));
 		return false;
 	}
-	if (randAccess)
+	if (PrevRecPtr == InvalidXLogRecPtr)
 	{
 		/*
 		 * We can't exactly verify the prev-link, but surely it should be less
@@ -1015,11 +1119,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));
@@ -1052,9 +1159,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;
 		}
 
@@ -1106,8 +1211,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)
 		{
@@ -1127,9 +1240,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'.
@@ -1142,6 +1255,7 @@ err:
  */
 bool
 WALRead(XLogReaderState *state,
+		WALSegmentOpenCB segopenfn, WALSegmentCloseCB segclosefn,
 		char *buf, XLogRecPtr startptr, Size count, TimeLineID tli,
 		WALReadError *errinfo)
 {
@@ -1173,10 +1287,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 46eda33f25..b003990745 100644
--- a/src/backend/access/transam/xlogutils.c
+++ b/src/backend/access/transam/xlogutils.c
@@ -686,8 +686,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);
@@ -702,7 +701,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;
 
@@ -788,6 +787,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;
@@ -825,9 +825,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;
@@ -940,11 +942,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 2f6803637b..4f6e87f18d 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -148,7 +148,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)
@@ -198,11 +199,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 =
@@ -319,7 +321,8 @@ CreateInitDecodingContext(const 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)
@@ -422,7 +425,7 @@ CreateInitDecodingContext(const 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 */
@@ -476,7 +479,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)
@@ -528,8 +532,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);
@@ -585,7 +589,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 01d354829b..8f8c129620 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 d9d36879ed..7ab0b804e4 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -153,9 +153,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);
 
 	/*
@@ -512,9 +511,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);
 
 		/*
@@ -536,7 +534,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 a4d6f30957..b024bbc3cd 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,
@@ -807,9 +804,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;
@@ -837,7 +836,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,
@@ -1011,9 +1010,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);
 
@@ -1171,9 +1169,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;
@@ -2749,7 +2746,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,
@@ -2847,7 +2844,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 cf119848b0..712c85281c 100644
--- a/src/bin/pg_rewind/parsexlog.c
+++ b/src/bin/pg_rewind/parsexlog.c
@@ -41,15 +41,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
@@ -66,20 +60,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)
 		{
@@ -123,19 +119,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)
@@ -170,7 +166,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,
@@ -186,11 +181,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");
 
@@ -200,7 +191,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)
 		{
@@ -247,10 +244,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;
@@ -283,14 +281,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--;
-
-		XLogFileName(xlogfname, targetHistory[private->tliIndex].tli,
+		while (*tliIndex < targetNentries - 1 &&
+			   targetHistory[*tliIndex].end < targetSegEnd)
+			(*tliIndex)++;
+		while (*tliIndex > 0 &&
+			   targetHistory[*tliIndex].begin >= targetSegEnd)
+			(*tliIndex)--;
+
+		XLogFileName(xlogfname, targetHistory[*tliIndex].tli,
 					 xlogreadsegno, WalSegSz);
 
 		snprintf(xlogfpath, MAXPGPATH, "%s/" XLOGDIR "/%s",
@@ -303,7 +301,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;
@@ -317,7 +315,7 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
 			xlogreadfd = RestoreArchivedFile(xlogreader->segcxt.ws_dir,
 											 xlogfname,
 											 WalSegSz,
-											 private->restoreCommand);
+											 restoreCommand);
 
 			if (xlogreadfd < 0)
 			{
@@ -359,7 +357,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 75ece5c658..c4047b92b5 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;
 }
@@ -1044,16 +1050,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",
@@ -1076,7 +1080,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 5d9e0d3292..1492f1992d 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,36 @@ 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 +137,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
@@ -186,7 +157,9 @@ struct XLogReaderState
 								 * read by reader, which must be larger than
 								 * the request, or -1 on error */
 	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? */
+
 
 
 	/* ----------------------------------------
@@ -228,8 +201,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;
 
@@ -256,6 +227,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;
 };
@@ -263,9 +243,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);
@@ -273,12 +251,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,
@@ -298,6 +281,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 364a21c4ea..397fb27fc2 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/pg_config_manual.h b/src/include/pg_config_manual.h
index e28c990382..bd9a5c6c2b 100644
--- a/src/include/pg_config_manual.h
+++ b/src/include/pg_config_manual.h
@@ -384,7 +384,7 @@
  * Enable debugging print statements for WAL-related operations; see
  * also the wal_debug GUC var.
  */
-/* #define WAL_DEBUG */
+#define WAL_DEBUG
 
 /*
  * Enable tracing of resource consumption during sort operations;
diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h
index af551d6f4e..94e278ef81 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;
 
@@ -105,14 +110,16 @@ extern LogicalDecodingContext *CreateInitDecodingContext(const 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.30.1