0003-Move-pg_get_wal_records_info-functionality-from-pg_w.patch

From df324a756b175ad9014d8dfe439fea1fa8137029 Mon Sep 17 00:00:00 2001
From: Vignesh C <vignesh21@gmail.com>
Date: Fri, 28 Jul 2023 16:50:52 +0530
Subject: [PATCH 3/3] Move pg_get_wal_records_info functionality from
 pg_walinspect to backend.

Upgrade of publications required pg_get_wal_records_info to check that
there are no WAL record other thatn CHECKPOINT_SHUTDOWN WAL record to be
consumed. Hence moved pg_get_wal_records_info functionality as
pg_get_wal_records_content to the backend so that it can be called from
pg_upgrade.
---
 contrib/pg_walinspect/Makefile                |   2 +-
 contrib/pg_walinspect/meson.build             |   1 +
 .../pg_walinspect/pg_walinspect--1.1--1.2.sql |  32 ++++
 contrib/pg_walinspect/pg_walinspect.c         | 148 +-----------------
 contrib/pg_walinspect/pg_walinspect.control   |   2 +-
 doc/src/sgml/func.sgml                        |  52 ++++++
 src/backend/access/transam/xlog.c             |  27 ++++
 src/backend/access/transam/xlogfuncs.c        | 125 +++++++++++++++
 src/backend/access/transam/xlogutils.c        | 117 ++++++++++++++
 src/backend/catalog/system_functions.sql      |   4 +
 src/include/access/xlog.h                     |   2 +
 src/include/access/xlogutils.h                |   4 +
 src/include/catalog/pg_proc.dat               |   9 ++
 src/test/modules/brin/t/02_wal_consistency.pl |   2 +-
 .../modules/test_custom_rmgrs/t/001_basic.pl  |   2 +-
 src/test/regress/expected/misc_functions.out  |  79 ++++++++++
 src/test/regress/sql/misc_functions.sql       |  57 +++++++
 17 files changed, 515 insertions(+), 150 deletions(-)
 create mode 100644 contrib/pg_walinspect/pg_walinspect--1.1--1.2.sql

diff --git a/contrib/pg_walinspect/Makefile b/contrib/pg_walinspect/Makefile
index 22090f7716..5cc7d81b42 100644
--- a/contrib/pg_walinspect/Makefile
+++ b/contrib/pg_walinspect/Makefile
@@ -7,7 +7,7 @@ OBJS = \
 PGFILEDESC = "pg_walinspect - functions to inspect contents of PostgreSQL Write-Ahead Log"
 
 EXTENSION = pg_walinspect
-DATA = pg_walinspect--1.0.sql pg_walinspect--1.0--1.1.sql
+DATA = pg_walinspect--1.0.sql pg_walinspect--1.0--1.1.sql pg_walinspect--1.1--1.2.sql
 
 REGRESS = pg_walinspect oldextversions
 
diff --git a/contrib/pg_walinspect/meson.build b/contrib/pg_walinspect/meson.build
index 80059f6119..8f7a99a493 100644
--- a/contrib/pg_walinspect/meson.build
+++ b/contrib/pg_walinspect/meson.build
@@ -20,6 +20,7 @@ install_data(
   'pg_walinspect.control',
   'pg_walinspect--1.0.sql',
   'pg_walinspect--1.0--1.1.sql',
+  'pg_walinspect--1.1--1.2.sql',
   kwargs: contrib_data_args,
 )
 
diff --git a/contrib/pg_walinspect/pg_walinspect--1.1--1.2.sql b/contrib/pg_walinspect/pg_walinspect--1.1--1.2.sql
new file mode 100644
index 0000000000..41ec538623
--- /dev/null
+++ b/contrib/pg_walinspect/pg_walinspect--1.1--1.2.sql
@@ -0,0 +1,32 @@
+/* contrib/pg_walinspect/pg_walinspect--1.1--1.2.sql */
+
+-- complain if script is sourced in psql, rather than via CREATE EXTENSION
+\echo Use "ALTER EXTENSION pg_walinspect UPDATE TO 1.2" to load this file. \quit
+
+/* These functions are now in the backend and callers should update to use those */
+ALTER EXTENSION pg_walinspect DROP FUNCTION pg_get_wal_records_info;
+DROP FUNCTION pg_get_wal_records_info(pg_lsn, pg_lsn);
+
+--
+-- pg_get_wal_records_info()
+--
+CREATE FUNCTION pg_get_wal_records_info(IN start_lsn pg_lsn,
+    IN end_lsn pg_lsn,
+    OUT start_lsn pg_lsn,
+    OUT end_lsn pg_lsn,
+    OUT prev_lsn pg_lsn,
+    OUT xid xid,
+    OUT resource_manager text,
+    OUT record_type text,
+    OUT record_length int4,
+    OUT main_data_length int4,
+    OUT fpi_length int4,
+    OUT description text,
+    OUT block_ref text
+)
+RETURNS SETOF record
+AS 'pg_get_wal_records_content'
+LANGUAGE INTERNAL STRICT PARALLEL SAFE;
+
+REVOKE EXECUTE ON FUNCTION pg_get_wal_records_info(pg_lsn, pg_lsn) FROM PUBLIC;
+GRANT EXECUTE ON FUNCTION pg_get_wal_records_info(pg_lsn, pg_lsn) TO pg_read_server_files;
diff --git a/contrib/pg_walinspect/pg_walinspect.c b/contrib/pg_walinspect/pg_walinspect.c
index 796a74f322..6e572d3436 100644
--- a/contrib/pg_walinspect/pg_walinspect.c
+++ b/contrib/pg_walinspect/pg_walinspect.c
@@ -38,10 +38,6 @@ PG_FUNCTION_INFO_V1(pg_get_wal_records_info_till_end_of_wal);
 PG_FUNCTION_INFO_V1(pg_get_wal_stats);
 PG_FUNCTION_INFO_V1(pg_get_wal_stats_till_end_of_wal);
 
-static void ValidateInputLSNs(XLogRecPtr start_lsn, XLogRecPtr *end_lsn);
-static XLogRecPtr GetCurrentLSN(void);
-static XLogReaderState *InitXLogReaderState(XLogRecPtr lsn);
-static XLogRecord *ReadNextXLogRecord(XLogReaderState *xlogreader);
 static void GetWALRecordInfo(XLogReaderState *record, Datum *values,
 							 bool *nulls, uint32 ncols);
 static void GetWALRecordsInfo(FunctionCallInfo fcinfo,
@@ -62,120 +58,6 @@ static void GetWalStats(FunctionCallInfo fcinfo,
 static void GetWALBlockInfo(FunctionCallInfo fcinfo, XLogReaderState *record,
 							bool show_data);
 
-/*
- * Return the LSN up to which the server has WAL.
- */
-static XLogRecPtr
-GetCurrentLSN(void)
-{
-	XLogRecPtr	curr_lsn;
-
-	/*
-	 * We determine the current LSN of the server similar to how page_read
-	 * callback read_local_xlog_page_no_wait does.
-	 */
-	if (!RecoveryInProgress())
-		curr_lsn = GetFlushRecPtr(NULL);
-	else
-		curr_lsn = GetXLogReplayRecPtr(NULL);
-
-	Assert(!XLogRecPtrIsInvalid(curr_lsn));
-
-	return curr_lsn;
-}
-
-/*
- * Initialize WAL reader and identify first valid LSN.
- */
-static XLogReaderState *
-InitXLogReaderState(XLogRecPtr lsn)
-{
-	XLogReaderState *xlogreader;
-	ReadLocalXLogPageNoWaitPrivate *private_data;
-	XLogRecPtr	first_valid_record;
-
-	/*
-	 * Reading WAL below the first page of the first segments isn't allowed.
-	 * This is a bootstrap WAL page and the page_read callback fails to read
-	 * it.
-	 */
-	if (lsn < XLOG_BLCKSZ)
-		ereport(ERROR,
-				(errmsg("could not read WAL at LSN %X/%X",
-						LSN_FORMAT_ARGS(lsn))));
-
-	private_data = (ReadLocalXLogPageNoWaitPrivate *)
-		palloc0(sizeof(ReadLocalXLogPageNoWaitPrivate));
-
-	xlogreader = XLogReaderAllocate(wal_segment_size, NULL,
-									XL_ROUTINE(.page_read = &read_local_xlog_page_no_wait,
-											   .segment_open = &wal_segment_open,
-											   .segment_close = &wal_segment_close),
-									private_data);
-
-	if (xlogreader == NULL)
-		ereport(ERROR,
-				(errcode(ERRCODE_OUT_OF_MEMORY),
-				 errmsg("out of memory"),
-				 errdetail("Failed while allocating a WAL reading processor.")));
-
-	/* first find a valid recptr to start from */
-	first_valid_record = XLogFindNextRecord(xlogreader, lsn);
-
-	if (XLogRecPtrIsInvalid(first_valid_record))
-		ereport(ERROR,
-				(errmsg("could not find a valid record after %X/%X",
-						LSN_FORMAT_ARGS(lsn))));
-
-	return xlogreader;
-}
-
-/*
- * Read next WAL record.
- *
- * By design, to be less intrusive in a running system, no slot is allocated
- * to reserve the WAL we're about to read. Therefore this function can
- * encounter read errors for historical WAL.
- *
- * We guard against ordinary errors trying to read WAL that hasn't been
- * written yet by limiting end_lsn to the flushed WAL, but that can also
- * encounter errors if the flush pointer falls in the middle of a record. In
- * that case we'll return NULL.
- */
-static XLogRecord *
-ReadNextXLogRecord(XLogReaderState *xlogreader)
-{
-	XLogRecord *record;
-	char	   *errormsg;
-
-	record = XLogReadRecord(xlogreader, &errormsg);
-
-	if (record == NULL)
-	{
-		ReadLocalXLogPageNoWaitPrivate *private_data;
-
-		/* return NULL, if end of WAL is reached */
-		private_data = (ReadLocalXLogPageNoWaitPrivate *)
-			xlogreader->private_data;
-
-		if (private_data->end_of_wal)
-			return NULL;
-
-		if (errormsg)
-			ereport(ERROR,
-					(errcode_for_file_access(),
-					 errmsg("could not read WAL at %X/%X: %s",
-							LSN_FORMAT_ARGS(xlogreader->EndRecPtr), errormsg)));
-		else
-			ereport(ERROR,
-					(errcode_for_file_access(),
-					 errmsg("could not read WAL at %X/%X",
-							LSN_FORMAT_ARGS(xlogreader->EndRecPtr))));
-	}
-
-	return record;
-}
-
 /*
  * Output values that make up a row describing caller's WAL record.
  *
@@ -502,33 +384,6 @@ pg_get_wal_record_info(PG_FUNCTION_ARGS)
 #undef PG_GET_WAL_RECORD_INFO_COLS
 }
 
-/*
- * Validate start and end LSNs coming from the function inputs.
- *
- * If end_lsn is found to be higher than the current LSN reported by the
- * cluster, use the current LSN as the upper bound.
- */
-static void
-ValidateInputLSNs(XLogRecPtr start_lsn, XLogRecPtr *end_lsn)
-{
-	XLogRecPtr	curr_lsn = GetCurrentLSN();
-
-	if (start_lsn > curr_lsn)
-		ereport(ERROR,
-				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-				 errmsg("WAL start LSN must be less than current LSN"),
-				 errdetail("Current WAL LSN on the database system is at %X/%X.",
-						   LSN_FORMAT_ARGS(curr_lsn))));
-
-	if (start_lsn > *end_lsn)
-		ereport(ERROR,
-				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-				 errmsg("WAL start LSN must be less than end LSN")));
-
-	if (*end_lsn > curr_lsn)
-		*end_lsn = curr_lsn;
-}
-
 /*
  * Get info of all WAL records between start LSN and end LSN.
  */
@@ -582,7 +437,8 @@ GetWALRecordsInfo(FunctionCallInfo fcinfo, XLogRecPtr start_lsn,
 }
 
 /*
- * Get info of all WAL records between start LSN and end LSN.
+ * The following functions have been removed in newer versions in 1.2, but
+ * they are kept around for compatibility.
  */
 Datum
 pg_get_wal_records_info(PG_FUNCTION_ARGS)
diff --git a/contrib/pg_walinspect/pg_walinspect.control b/contrib/pg_walinspect/pg_walinspect.control
index efa3cb2cfe..5f574b865b 100644
--- a/contrib/pg_walinspect/pg_walinspect.control
+++ b/contrib/pg_walinspect/pg_walinspect.control
@@ -1,5 +1,5 @@
 # pg_walinspect extension
 comment = 'functions to inspect contents of PostgreSQL Write-Ahead Log'
-default_version = '1.1'
+default_version = '1.2'
 module_pathname = '$libdir/pg_walinspect'
 relocatable = true
diff --git a/doc/src/sgml/func.sgml b/doc/src/sgml/func.sgml
index b94827674c..a8866feaa3 100644
--- a/doc/src/sgml/func.sgml
+++ b/doc/src/sgml/func.sgml
@@ -26670,6 +26670,37 @@ LOG:  Grand total: 1651920 bytes in 201 blocks; 622360 free (88 chunks); 1029560
         get the replication lag.
        </para></entry>
       </row>
+
+      <row>
+       <entry role="func_table_entry"><para role="func_signature">
+        <indexterm>
+         <primary>pg_get_wal_records_content</primary>
+        </indexterm>
+        <function>pg_get_wal_records_content</function> ( <parameter>lsn1</parameter> <type>pg_lsn</type>, <parameter>lsn2</parameter> <type>pg_lsn</type> )
+        <returnvalue>record</returnvalue>
+        ( <parameter>start_lsn</parameter> <type>pg_lsn</type>,
+        <parameter>end_lsn</parameter> <type>pg_lsn</type>,
+        <parameter>prev_lsn</parameter> <type>pg_lsn</type>,
+        <parameter>xid</parameter> <type>xid</type>,
+        <parameter>resource_manager</parameter> <type>text</type>,
+        <parameter>record_type</parameter> <type>text</type>,
+        <parameter>record_length</parameter> <type>integer</type>,
+        <parameter>main_data_length</parameter> <type>integer</type>,
+        <parameter>fpi_length</parameter> <type>integer</type>,
+        <parameter>description</parameter> <type>text</type>,
+        <parameter>block_ref</parameter> <type>text</type>
+         )
+       </para>
+       <para>
+        Gets information of all the valid WAL records between
+        <replaceable>start_lsn</replaceable> and <replaceable>end_lsn</replaceable>.
+        Returns one row per WAL record. This can be used
+        with <structname>pg_stat_replication</structname> or some of the
+        functions shown in <xref linkend="functions-admin-backup-table"/> to
+        get the replication lag WAL records content. The function raises an
+        error if <replaceable>start_lsn</replaceable> is not available.
+       </para></entry>
+      </row>
      </tbody>
     </tgroup>
    </table>
@@ -26728,6 +26759,27 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
 </programlisting>
    </para>
 
+   <para>
+    <function>pg_get_wal_records_content</function> gets information of all the valid WAL records between
+     <replaceable>start_lsn</replaceable> and <replaceable>end_lsn</replaceable>.
+     Returns one row per WAL record.  For example:
+<screen>
+postgres=# SELECT * FROM pg_get_wal_records_content('0/1E913618', '0/1E913740') LIMIT 1;
+-[ RECORD 1 ]----+--------------------------------------------------------------
+start_lsn        | 0/1E913618
+end_lsn          | 0/1E913650
+prev_lsn         | 0/1E9135A0
+xid              | 0
+resource_manager | Standby
+record_type      | RUNNING_XACTS
+record_length    | 50
+main_data_length | 24
+fpi_length       | 0
+description      | nextXid 33775 latestCompletedXid 33774 oldestRunningXid 33775
+block_ref        |
+</screen>
+   </para>
+
   </sect2>
 
   <sect2 id="functions-recovery-control">
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 60c0b7ec3a..6115df57fd 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -8988,3 +8988,30 @@ SetWalWriterSleeping(bool sleeping)
 	XLogCtl->WalWriterSleeping = sleeping;
 	SpinLockRelease(&XLogCtl->info_lck);
 }
+
+/*
+ * Validate start and end LSNs coming from the function inputs.
+ *
+ * If end_lsn is found to be higher than the current LSN reported by the
+ * cluster, use the current LSN as the upper bound.
+ */
+void
+ValidateInputLSNs(XLogRecPtr start_lsn, XLogRecPtr *end_lsn)
+{
+	XLogRecPtr	curr_lsn = GetCurrentLSN();
+
+	if (start_lsn > curr_lsn)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("WAL start LSN must be less than current LSN"),
+				 errdetail("Current WAL LSN on the database system is at %X/%X.",
+						   LSN_FORMAT_ARGS(curr_lsn))));
+
+	if (start_lsn > *end_lsn)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("WAL start LSN must be less than end LSN")));
+
+	if (*end_lsn > curr_lsn)
+		*end_lsn = curr_lsn;
+}
diff --git a/src/backend/access/transam/xlogfuncs.c b/src/backend/access/transam/xlogfuncs.c
index 5044ff0643..bb66eb26b6 100644
--- a/src/backend/access/transam/xlogfuncs.c
+++ b/src/backend/access/transam/xlogfuncs.c
@@ -754,3 +754,128 @@ pg_promote(PG_FUNCTION_ARGS)
 						   wait_seconds)));
 	PG_RETURN_BOOL(false);
 }
+
+/*
+ * Get content of all WAL records between start LSN and end LSN.
+ */
+Datum
+pg_get_wal_records_content(PG_FUNCTION_ARGS)
+{
+#define PG_GET_WAL_RECORDS_INFO_COLS 11
+	FuncCallContext *funcctx;
+	XLogReaderState *xlogreader;
+
+	XLogRecPtr	start_lsn = PG_GETARG_LSN(0);
+	XLogRecPtr	end_lsn = PG_GETARG_LSN(1);
+
+	if (SRF_IS_FIRSTCALL())
+	{
+		MemoryContext oldcontext;
+		TupleDesc	tupdesc;
+
+		ValidateInputLSNs(start_lsn, &end_lsn);
+
+		/* create a function context for cross-call persistence */
+		funcctx = SRF_FIRSTCALL_INIT();
+
+		/*
+		 * Switch to memory context appropriate for multiple function calls
+		 */
+		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
+
+		/* build tupdesc for result tuples */
+		tupdesc = CreateTemplateTupleDesc(11);
+		TupleDescInitEntry(tupdesc, (AttrNumber) 1, "start_lsn",
+						   PG_LSNOID, -1, 0);
+		TupleDescInitEntry(tupdesc, (AttrNumber) 2, "end_lsn",
+						   PG_LSNOID, -1, 0);
+		TupleDescInitEntry(tupdesc, (AttrNumber) 3, "prev_lsn",
+						   PG_LSNOID, -1, 0);
+		TupleDescInitEntry(tupdesc, (AttrNumber) 4, "xid",
+						   XIDOID, -1, 0);
+		TupleDescInitEntry(tupdesc, (AttrNumber) 5, "resource_manager",
+						   TEXTOID, -1, 0);
+		TupleDescInitEntry(tupdesc, (AttrNumber) 6, "record_type",
+						   TEXTOID, -1, 0);
+		TupleDescInitEntry(tupdesc, (AttrNumber) 7, "record_length",
+						   INT4OID, -1, 0);
+		TupleDescInitEntry(tupdesc, (AttrNumber) 8, "main_data_length",
+						   INT4OID, -1, 0);
+		TupleDescInitEntry(tupdesc, (AttrNumber) 9, "fpi_length",
+						   INT4OID, -1, 0);
+		TupleDescInitEntry(tupdesc, (AttrNumber) 10, "description",
+						   TEXTOID, -1, 0);
+		TupleDescInitEntry(tupdesc, (AttrNumber) 11, "block_ref",
+						   TEXTOID, -1, 0);
+
+		funcctx->tuple_desc = BlessTupleDesc(tupdesc);
+
+		if (start_lsn < end_lsn)
+			funcctx->user_fctx = InitXLogReaderState(start_lsn);
+		else
+			funcctx->user_fctx = NULL;
+
+		MemoryContextSwitchTo(oldcontext);
+	}
+
+	funcctx = SRF_PERCALL_SETUP();
+	xlogreader = (XLogReaderState *) funcctx->user_fctx;
+
+	while (xlogreader && ReadNextXLogRecord(xlogreader) &&
+		   xlogreader->EndRecPtr <= end_lsn)
+	{
+		Datum		values[11] = {0};
+		bool		nulls[11] = {0};
+		HeapTuple	tuple;
+		Datum		result;
+		XLogReaderState *record = xlogreader;
+		const char *record_type;
+		StringInfoData rec_desc;
+		StringInfoData rec_blk_ref;
+		int			i = 0;
+		uint32		fpi_len = 0;
+		RmgrData	desc;
+
+		desc = GetRmgr(XLogRecGetRmid(record));
+		record_type = desc.rm_identify(XLogRecGetInfo(record));
+
+		if (record_type == NULL)
+			record_type = psprintf("UNKNOWN (%x)", XLogRecGetInfo(record) & ~XLR_INFO_MASK);
+
+		initStringInfo(&rec_desc);
+		desc.rm_desc(&rec_desc, record);
+
+		if (XLogRecHasAnyBlockRefs(record))
+		{
+			initStringInfo(&rec_blk_ref);
+			XLogRecGetBlockRefInfo(record, false, true, &rec_blk_ref, &fpi_len);
+		}
+
+		values[i++] = LSNGetDatum(record->ReadRecPtr);
+		values[i++] = LSNGetDatum(record->EndRecPtr);
+		values[i++] = LSNGetDatum(XLogRecGetPrev(record));
+		values[i++] = TransactionIdGetDatum(XLogRecGetXid(record));
+		values[i++] = CStringGetTextDatum(desc.rm_name);
+		values[i++] = CStringGetTextDatum(record_type);
+		values[i++] = UInt32GetDatum(XLogRecGetTotalLen(record));
+		values[i++] = UInt32GetDatum(XLogRecGetDataLen(record));
+		values[i++] = UInt32GetDatum(fpi_len);
+
+		if (rec_desc.len > 0)
+			values[i++] = CStringGetTextDatum(rec_desc.data);
+		else
+			nulls[i++] = true;
+
+		if (XLogRecHasAnyBlockRefs(record))
+			values[i++] = CStringGetTextDatum(rec_blk_ref.data);
+		else
+			nulls[i++] = true;
+
+		tuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);
+		result = HeapTupleGetDatum(tuple);
+		SRF_RETURN_NEXT(funcctx, result);
+	}
+
+	SRF_RETURN_DONE(funcctx);
+#undef PG_GET_WAL_RECORDS_INFO_COLS
+}
diff --git a/src/backend/access/transam/xlogutils.c b/src/backend/access/transam/xlogutils.c
index e174a2a891..90eab3e2f3 100644
--- a/src/backend/access/transam/xlogutils.c
+++ b/src/backend/access/transam/xlogutils.c
@@ -1048,3 +1048,120 @@ WALReadRaiseError(WALReadError *errinfo)
 						errinfo->wre_req)));
 	}
 }
+
+/*
+ * Return the LSN up to which the server has WAL.
+ */
+XLogRecPtr
+GetCurrentLSN(void)
+{
+	XLogRecPtr	curr_lsn;
+
+	/*
+	 * We determine the current LSN of the server similar to how page_read
+	 * callback read_local_xlog_page_no_wait does.
+	 */
+	if (!RecoveryInProgress())
+		curr_lsn = GetFlushRecPtr(NULL);
+	else
+		curr_lsn = GetXLogReplayRecPtr(NULL);
+
+	Assert(!XLogRecPtrIsInvalid(curr_lsn));
+
+	return curr_lsn;
+}
+
+/*
+ * Initialize WAL reader and identify first valid LSN.
+ */
+XLogReaderState *
+InitXLogReaderState(XLogRecPtr lsn)
+{
+	XLogReaderState *xlogreader;
+	ReadLocalXLogPageNoWaitPrivate *private_data;
+	XLogRecPtr	first_valid_record;
+
+	/*
+	 * Reading WAL below the first page of the first segments isn't allowed.
+	 * This is a bootstrap WAL page and the page_read callback fails to read
+	 * it.
+	 */
+	if (lsn < XLOG_BLCKSZ)
+		ereport(ERROR,
+				(errmsg("could not read WAL at LSN %X/%X",
+						LSN_FORMAT_ARGS(lsn))));
+
+	private_data = (ReadLocalXLogPageNoWaitPrivate *)
+		palloc0(sizeof(ReadLocalXLogPageNoWaitPrivate));
+
+	xlogreader = XLogReaderAllocate(wal_segment_size, NULL,
+									XL_ROUTINE(.page_read = &read_local_xlog_page_no_wait,
+											   .segment_open = &wal_segment_open,
+											   .segment_close = &wal_segment_close),
+									private_data);
+
+	if (xlogreader == NULL)
+		ereport(ERROR,
+				(errcode(ERRCODE_OUT_OF_MEMORY),
+				 errmsg("out of memory"),
+				 errdetail("Failed while allocating a WAL reading processor.")));
+
+	/* first find a valid recptr to start from */
+	first_valid_record = XLogFindNextRecord(xlogreader, lsn);
+
+	if (XLogRecPtrIsInvalid(first_valid_record))
+	{
+		ereport(ERROR,
+				(errmsg("could not find a valid record after %X/%X",
+						LSN_FORMAT_ARGS(lsn))));
+	}
+
+	return xlogreader;
+}
+
+
+/*
+ * Read next WAL record.
+ *
+ * By design, to be less intrusive in a running system, no slot is allocated
+ * to reserve the WAL we're about to read. Therefore this function can
+ * encounter read errors for historical WAL.
+ *
+ * We guard against ordinary errors trying to read WAL that hasn't been
+ * written yet by limiting end_lsn to the flushed WAL, but that can also
+ * encounter errors if the flush pointer falls in the middle of a record. In
+ * that case we'll return NULL.
+ */
+XLogRecord *
+ReadNextXLogRecord(XLogReaderState *xlogreader)
+{
+	XLogRecord *record;
+	char	   *errormsg;
+
+	record = XLogReadRecord(xlogreader, &errormsg);
+
+	if (record == NULL)
+	{
+		ReadLocalXLogPageNoWaitPrivate *private_data;
+
+		/* return NULL, if end of WAL is reached */
+		private_data = (ReadLocalXLogPageNoWaitPrivate *)
+			xlogreader->private_data;
+
+		if (private_data->end_of_wal)
+			return NULL;
+
+		if (errormsg)
+			ereport(ERROR,
+					(errcode_for_file_access(),
+					 errmsg("could not read WAL at %X/%X: %s",
+							LSN_FORMAT_ARGS(xlogreader->EndRecPtr), errormsg)));
+		else
+			ereport(ERROR,
+					(errcode_for_file_access(),
+					 errmsg("could not read WAL at %X/%X",
+							LSN_FORMAT_ARGS(xlogreader->EndRecPtr))));
+	}
+
+	return record;
+}
diff --git a/src/backend/catalog/system_functions.sql b/src/backend/catalog/system_functions.sql
index 07c0d89c4f..1fed0def9b 100644
--- a/src/backend/catalog/system_functions.sql
+++ b/src/backend/catalog/system_functions.sql
@@ -616,6 +616,8 @@ REVOKE EXECUTE ON FUNCTION pg_backup_stop(boolean) FROM public;
 
 REVOKE EXECUTE ON FUNCTION pg_create_restore_point(text) FROM public;
 
+REVOKE EXECUTE ON FUNCTION pg_get_wal_records_content(pg_lsn, pg_lsn) FROM public;
+
 REVOKE EXECUTE ON FUNCTION pg_switch_wal() FROM public;
 
 REVOKE EXECUTE ON FUNCTION pg_log_standby_snapshot() FROM public;
@@ -726,6 +728,8 @@ REVOKE EXECUTE ON FUNCTION pg_ls_replslotdir(text) FROM PUBLIC;
 -- We also set up some things as accessible to standard roles.
 --
 
+GRANT EXECUTE ON FUNCTION pg_get_wal_records_content(pg_lsn, pg_lsn) TO pg_read_server_files;
+
 GRANT EXECUTE ON FUNCTION pg_ls_logdir() TO pg_monitor;
 
 GRANT EXECUTE ON FUNCTION pg_ls_waldir() TO pg_monitor;
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 48ca852381..4ec1fc9d82 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -259,6 +259,8 @@ extern void SetInstallXLogFileSegmentActive(void);
 extern bool IsInstallXLogFileSegmentActive(void);
 extern void XLogShutdownWalRcv(void);
 
+extern void ValidateInputLSNs(XLogRecPtr start_lsn, XLogRecPtr *end_lsn);
+
 /*
  * Routines to start, stop, and get status of a base backup.
  */
diff --git a/src/include/access/xlogutils.h b/src/include/access/xlogutils.h
index 5b77b11f50..77e2974ab3 100644
--- a/src/include/access/xlogutils.h
+++ b/src/include/access/xlogutils.h
@@ -115,4 +115,8 @@ extern void XLogReadDetermineTimeline(XLogReaderState *state,
 
 extern void WALReadRaiseError(WALReadError *errinfo);
 
+extern XLogRecPtr GetCurrentLSN(void);
+extern XLogReaderState *InitXLogReaderState(XLogRecPtr lsn);
+extern XLogRecord *ReadNextXLogRecord(XLogReaderState *xlogreader);
+
 #endif
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 6996073989..d45696fca0 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -6489,6 +6489,15 @@
   proargnames => '{rm_id, rm_name, rm_builtin}',
   prosrc => 'pg_get_wal_resource_managers' },
 
+{ oid => '3813',
+  descr => 'Info of WAL conents between start LSN and end LSN',
+  proname => 'pg_get_wal_records_content', prorows => '10', proretset => 't',
+  provolatile => 's', prorettype => 'record', proargtypes => 'pg_lsn pg_lsn',
+  proallargtypes => '{pg_lsn,pg_lsn,pg_lsn,pg_lsn,pg_lsn,xid,text,text,int4,int4,int4,text,text}',
+  proargmodes => '{i,i,o,o,o,o,o,o,o,o,o,o,o}',
+  proargnames => '{start_lsn,end_lsn,start_lsn,end_lsn,prev_lsn,xid,resource_manager,record_type,record_length,main_data_length,fpi_length,description,block_ref}',
+  prosrc => 'pg_get_wal_records_content' },
+
 { oid => '2621', descr => 'reload configuration files',
   proname => 'pg_reload_conf', provolatile => 'v', prorettype => 'bool',
   proargtypes => '', prosrc => 'pg_reload_conf' },
diff --git a/src/test/modules/brin/t/02_wal_consistency.pl b/src/test/modules/brin/t/02_wal_consistency.pl
index 8b2b244feb..288bbd7a60 100644
--- a/src/test/modules/brin/t/02_wal_consistency.pl
+++ b/src/test/modules/brin/t/02_wal_consistency.pl
@@ -64,7 +64,7 @@ my $end_lsn = $whiskey->lsn('flush');
 
 my ($ret, $out, $err) = $whiskey->psql(
 	'postgres', qq{
-	select count(*) from pg_get_wal_records_info('$start_lsn', '$end_lsn')
+	select count(*) from pg_get_wal_records_content('$start_lsn', '$end_lsn')
 	where resource_manager = 'BRIN' AND
 	record_type ILIKE '%revmap%'
 	});
diff --git a/src/test/modules/test_custom_rmgrs/t/001_basic.pl b/src/test/modules/test_custom_rmgrs/t/001_basic.pl
index 50655d3788..4000de0560 100644
--- a/src/test/modules/test_custom_rmgrs/t/001_basic.pl
+++ b/src/test/modules/test_custom_rmgrs/t/001_basic.pl
@@ -54,7 +54,7 @@ my $expected =
   qq($record_end_lsn|test_custom_rmgrs|TEST_CUSTOM_RMGRS_MESSAGE|0|payload (10 bytes): payload123);
 my $result = $node->safe_psql(
 	'postgres',
-	qq[SELECT end_lsn, resource_manager, record_type, fpi_length, description FROM pg_get_wal_records_info('$start_lsn', '$end_lsn')
+	qq[SELECT end_lsn, resource_manager, record_type, fpi_length, description FROM pg_get_wal_records_content('$start_lsn', '$end_lsn')
 		WHERE resource_manager = 'test_custom_rmgrs';]);
 is($result, $expected,
 	'custom WAL resource manager has successfully written a WAL record');
diff --git a/src/test/regress/expected/misc_functions.out b/src/test/regress/expected/misc_functions.out
index c669948370..e5e10cddf9 100644
--- a/src/test/regress/expected/misc_functions.out
+++ b/src/test/regress/expected/misc_functions.out
@@ -642,3 +642,82 @@ SELECT segment_number > 0 AS ok_segment_number, timeline_id
  t                 |  4294967295
 (1 row)
 
+-- pg_get_wal_records_content
+CREATE TABLE sample_tbl(col1 int, col2 int);
+SELECT pg_current_wal_lsn() AS wal_lsn1 \gset
+INSERT INTO sample_tbl SELECT * FROM generate_series(1, 2);
+SELECT pg_current_wal_lsn() AS wal_lsn2 \gset
+-- Mask DETAIL messages as these could refer to current LSN positions.
+\set VERBOSITY terse
+-- Invalid start LSN.
+SELECT * FROM pg_get_wal_records_content('0/0', :'wal_lsn1');
+ERROR:  could not read WAL at LSN 0/0
+-- Start LSN > End LSN.
+SELECT * FROM pg_get_wal_records_content(:'wal_lsn2', :'wal_lsn1');
+ERROR:  WAL start LSN must be less than end LSN
+-- Success with end LSNs.
+SELECT COUNT(*) >= 1 AS ok FROM pg_get_wal_records_content(:'wal_lsn1', 'FFFFFFFF/FFFFFFFF');
+ ok 
+----
+ t
+(1 row)
+
+-- Failures with start LSNs.
+SELECT * FROM pg_get_wal_records_content('FFFFFFFF/FFFFFFFE', 'FFFFFFFF/FFFFFFFF');
+ERROR:  WAL start LSN must be less than current LSN
+SELECT COUNT(*) >= 1 AS ok FROM pg_get_wal_records_content(:'wal_lsn1', :'wal_lsn2');
+ ok 
+----
+ t
+(1 row)
+
+-- Test for filtering out WAL records of a particular table
+SELECT oid AS sample_tbl_oid FROM pg_class WHERE relname = 'sample_tbl' \gset
+SELECT COUNT(*) >= 1 AS ok FROM pg_get_wal_records_content(:'wal_lsn1', :'wal_lsn2')
+			WHERE block_ref LIKE concat('%', :'sample_tbl_oid', '%') AND resource_manager = 'Heap';
+ ok 
+----
+ t
+(1 row)
+
+-- Test for filtering out WAL records based on resource_manager and
+-- record_type
+SELECT COUNT(*) >= 1 AS ok FROM pg_get_wal_records_content(:'wal_lsn1', :'wal_lsn2')
+			WHERE resource_manager = 'Heap' AND record_type = 'INSERT';
+ ok 
+----
+ t
+(1 row)
+
+\set VERBOSITY default
+-- Tests for permissions
+CREATE ROLE regress_pg_get_wal;
+SELECT has_function_privilege('regress_pg_get_wal',
+  'pg_get_wal_records_content(pg_lsn, pg_lsn) ', 'EXECUTE'); -- no
+ has_function_privilege 
+------------------------
+ f
+(1 row)
+
+-- Functions accessible by users with role pg_read_server_files.
+GRANT pg_read_server_files TO regress_pg_get_wal;
+SELECT has_function_privilege('regress_pg_get_wal',
+  'pg_get_wal_records_content(pg_lsn, pg_lsn) ', 'EXECUTE'); -- yes
+ has_function_privilege 
+------------------------
+ t
+(1 row)
+
+-- Superuser can grant execute to other users.
+GRANT EXECUTE ON FUNCTION pg_get_wal_records_content(pg_lsn, pg_lsn)
+  TO regress_pg_get_wal;
+SELECT has_function_privilege('regress_pg_get_wal',
+  'pg_get_wal_records_content(pg_lsn, pg_lsn) ', 'EXECUTE'); -- yes
+ has_function_privilege 
+------------------------
+ t
+(1 row)
+
+REVOKE EXECUTE ON FUNCTION pg_get_wal_records_content(pg_lsn, pg_lsn)
+  FROM regress_pg_get_wal;
+DROP ROLE regress_pg_get_wal;
diff --git a/src/test/regress/sql/misc_functions.sql b/src/test/regress/sql/misc_functions.sql
index b57f01f3e9..40cee85dbc 100644
--- a/src/test/regress/sql/misc_functions.sql
+++ b/src/test/regress/sql/misc_functions.sql
@@ -237,3 +237,60 @@ SELECT segment_number > 0 AS ok_segment_number, timeline_id
   FROM pg_split_walfile_name('000000010000000100000000');
 SELECT segment_number > 0 AS ok_segment_number, timeline_id
   FROM pg_split_walfile_name('ffffffFF00000001000000af');
+
+-- pg_get_wal_records_content
+CREATE TABLE sample_tbl(col1 int, col2 int);
+SELECT pg_current_wal_lsn() AS wal_lsn1 \gset
+INSERT INTO sample_tbl SELECT * FROM generate_series(1, 2);
+SELECT pg_current_wal_lsn() AS wal_lsn2 \gset
+
+-- Mask DETAIL messages as these could refer to current LSN positions.
+\set VERBOSITY terse
+
+-- Invalid start LSN.
+SELECT * FROM pg_get_wal_records_content('0/0', :'wal_lsn1');
+-- Start LSN > End LSN.
+SELECT * FROM pg_get_wal_records_content(:'wal_lsn2', :'wal_lsn1');
+-- Success with end LSNs.
+SELECT COUNT(*) >= 1 AS ok FROM pg_get_wal_records_content(:'wal_lsn1', 'FFFFFFFF/FFFFFFFF');
+-- Failures with start LSNs.
+SELECT * FROM pg_get_wal_records_content('FFFFFFFF/FFFFFFFE', 'FFFFFFFF/FFFFFFFF');
+
+SELECT COUNT(*) >= 1 AS ok FROM pg_get_wal_records_content(:'wal_lsn1', :'wal_lsn2');
+
+-- Test for filtering out WAL records of a particular table
+SELECT oid AS sample_tbl_oid FROM pg_class WHERE relname = 'sample_tbl' \gset
+
+SELECT COUNT(*) >= 1 AS ok FROM pg_get_wal_records_content(:'wal_lsn1', :'wal_lsn2')
+			WHERE block_ref LIKE concat('%', :'sample_tbl_oid', '%') AND resource_manager = 'Heap';
+
+-- Test for filtering out WAL records based on resource_manager and
+-- record_type
+
+SELECT COUNT(*) >= 1 AS ok FROM pg_get_wal_records_content(:'wal_lsn1', :'wal_lsn2')
+			WHERE resource_manager = 'Heap' AND record_type = 'INSERT';
+
+\set VERBOSITY default
+
+-- Tests for permissions
+CREATE ROLE regress_pg_get_wal;
+SELECT has_function_privilege('regress_pg_get_wal',
+  'pg_get_wal_records_content(pg_lsn, pg_lsn) ', 'EXECUTE'); -- no
+
+-- Functions accessible by users with role pg_read_server_files.
+GRANT pg_read_server_files TO regress_pg_get_wal;
+
+SELECT has_function_privilege('regress_pg_get_wal',
+  'pg_get_wal_records_content(pg_lsn, pg_lsn) ', 'EXECUTE'); -- yes
+
+-- Superuser can grant execute to other users.
+GRANT EXECUTE ON FUNCTION pg_get_wal_records_content(pg_lsn, pg_lsn)
+  TO regress_pg_get_wal;
+
+SELECT has_function_privilege('regress_pg_get_wal',
+  'pg_get_wal_records_content(pg_lsn, pg_lsn) ', 'EXECUTE'); -- yes
+
+REVOKE EXECUTE ON FUNCTION pg_get_wal_records_content(pg_lsn, pg_lsn)
+  FROM regress_pg_get_wal;
+
+DROP ROLE regress_pg_get_wal;
-- 
2.34.1