v45-0002-Add-rm_is_record_decodable-callback-for-WAL-rmgr.patch

From 39ee4c84288bba30183a986aed458b16f0b12a58 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <bharath.rupireddyforpostgres@gmail.com>
Date: Tue, 3 Oct 2023 04:01:05 +0000
Subject: [PATCH v45] Add rm_is_record_decodable callback for WAL rmgrs

This commit lets WAL resource managers (rmgrs) define if WAL
record types that they add are logically decodable. Every rmgr
including custom rmgrs must define the new callback
rm_is_record_decodable returning true if given WAL record type of
theirs is logically decodable. In other words, return true for
record types that have something to do with logical decoding in
their rm_decode functions.

An immediate use of this new callback is in pg_upgrade. During
pg_upgrade, one can know if there are any logically decodable WAL
records after a certain point in WAL.

Bump WAL version indicator XLOG_PAGE_MAGIC.
---
 doc/src/sgml/custom-rmgr.sgml              |   1 +
 src/backend/access/transam/rmgr.c          |   4 +-
 src/backend/replication/logical/decode.c   | 151 +++++++++++++++++++++
 src/backend/utils/adt/pg_upgrade_support.c |  31 +++--
 src/bin/pg_rewind/parsexlog.c              |   2 +-
 src/bin/pg_waldump/rmgrdesc.c              |   2 +-
 src/include/access/rmgr.h                  |   2 +-
 src/include/access/rmgrlist.h              |  46 +++----
 src/include/access/xlog_internal.h         |   3 +-
 src/include/replication/decode.h           |   7 +
 10 files changed, 207 insertions(+), 42 deletions(-)

diff --git a/doc/src/sgml/custom-rmgr.sgml b/doc/src/sgml/custom-rmgr.sgml
index baf86b1c07..e804ca0689 100644
--- a/doc/src/sgml/custom-rmgr.sgml
+++ b/doc/src/sgml/custom-rmgr.sgml
@@ -54,6 +54,7 @@ typedef struct RmgrData
     void        (*rm_mask) (char *pagedata, BlockNumber blkno);
     void        (*rm_decode) (struct LogicalDecodingContext *ctx,
                               struct XLogRecordBuffer *buf);
+    bool        (*rm_is_record_decodable) (uint8 type);
 } RmgrData;
 </programlisting>
  </para>
diff --git a/src/backend/access/transam/rmgr.c b/src/backend/access/transam/rmgr.c
index 7d67eda5f7..001bdf3535 100644
--- a/src/backend/access/transam/rmgr.c
+++ b/src/backend/access/transam/rmgr.c
@@ -35,8 +35,8 @@
 #include "utils/relmapper.h"
 
 /* must be kept in sync with RmgrData definition in xlog_internal.h */
-#define PG_RMGR(symname,name,redo,desc,identify,startup,cleanup,mask,decode) \
-	{ name, redo, desc, identify, startup, cleanup, mask, decode },
+#define PG_RMGR(symname,name,redo,desc,identify,startup,cleanup,mask,decode,is_record_decodable) \
+	{ name, redo, desc, identify, startup, cleanup, mask, decode, is_record_decodable },
 
 RmgrData	RmgrTable[RM_MAX_ID + 1] = {
 #include "access/rmgrlist.h"
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index 730061c9da..60d26ae015 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -116,7 +116,17 @@ LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *recor
 	rmgr = GetRmgr(XLogRecGetRmid(record));
 
 	if (rmgr.rm_decode != NULL)
+	{
+#ifdef USE_ASSERT_CHECKING
+		if (rmgr.rm_is_record_decodable == NULL)
+			ereport(ERROR,
+					errmsg("cannot check logical decodability for resource manager \"%s\" with ID %d",
+						   rmgr.rm_name, XLogRecGetRmid(record)),
+					errdetail("Logical decodability callback is not defined for the resource manager."));
+#endif
+
 		rmgr.rm_decode(ctx, &buf);
+	}
 	else
 	{
 		/* just deal with xid, and done */
@@ -196,6 +206,36 @@ xlog_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 	}
 }
 
+/*
+ * Return true if given XLOG_ID record type is logically decodable. In other
+ * words, return true for record types that have something to do with logical
+ * decoding in xlog_decode.
+ */
+bool
+xlog_is_record_decodable(uint8 info)
+{
+	switch (info)
+	{
+		case XLOG_CHECKPOINT_SHUTDOWN:
+		case XLOG_END_OF_RECOVERY:
+			return true;
+		case XLOG_CHECKPOINT_ONLINE:
+		case XLOG_PARAMETER_CHANGE:
+		case XLOG_NOOP:
+		case XLOG_NEXTOID:
+		case XLOG_SWITCH:
+		case XLOG_BACKUP_END:
+		case XLOG_RESTORE_POINT:
+		case XLOG_FPW_CHANGE:
+		case XLOG_FPI_FOR_HINT:
+		case XLOG_FPI:
+		case XLOG_OVERWRITE_CONTRECORD:
+			return false;
+		default:
+			elog(ERROR, "unexpected RM_XLOG_ID record type: %u", info);
+	}
+}
+
 /*
  * Handle rmgr XACT_ID records for LogicalDecodingProcessRecord().
  */
@@ -353,6 +393,30 @@ xact_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 	}
 }
 
+/*
+ * Return true if given XACT_ID record type is logically decodable. In other
+ * words, return true for record types that have something to do with logical
+ * decoding in xact_decode.
+ */
+bool
+xact_is_record_decodable(uint8 info)
+{
+	switch (info)
+	{
+		case XLOG_XACT_COMMIT:
+		case XLOG_XACT_COMMIT_PREPARED:
+		case XLOG_XACT_ABORT:
+		case XLOG_XACT_ABORT_PREPARED:
+		case XLOG_XACT_INVALIDATIONS:
+		case XLOG_XACT_PREPARE:
+			return true;
+		case XLOG_XACT_ASSIGNMENT:
+			return false;
+		default:
+			elog(ERROR, "unexpected RM_XACT_ID record type: %u", info);
+	}
+}
+
 /*
  * Handle rmgr STANDBY_ID records for LogicalDecodingProcessRecord().
  */
@@ -399,6 +463,26 @@ standby_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 	}
 }
 
+/*
+ * Return true if given STANDBY_ID record type is logically decodable. In other
+ * words, return true for record types that have something to do with logical
+ * decoding in standby_decode.
+ */
+bool
+standy_is_record_decodable(uint8 info)
+{
+	switch (info)
+	{
+		case XLOG_RUNNING_XACTS:
+			return true;
+		case XLOG_STANDBY_LOCK:
+		case XLOG_INVALIDATIONS:
+			return false;
+		default:
+			elog(ERROR, "unexpected RM_STANDBY_ID record type: %u", info);
+	}
+}
+
 /*
  * Handle rmgr HEAP2_ID records for LogicalDecodingProcessRecord().
  */
@@ -458,6 +542,31 @@ heap2_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 	}
 }
 
+/*
+ * Return true if given HEAP2_ID record type is logically decodable. In other
+ * words, return true for record types that have something to do with logical
+ * decoding in heap2_decode.
+ */
+bool
+heap2_is_record_decodable(uint8 info)
+{
+	switch (info)
+	{
+		case XLOG_HEAP2_MULTI_INSERT:
+		case XLOG_HEAP2_NEW_CID:
+			return true;
+		case XLOG_HEAP2_REWRITE:
+		case XLOG_HEAP2_FREEZE_PAGE:
+		case XLOG_HEAP2_PRUNE:
+		case XLOG_HEAP2_VACUUM:
+		case XLOG_HEAP2_VISIBLE:
+		case XLOG_HEAP2_LOCK_UPDATED:
+			return false;
+		default:
+			elog(ERROR, "unexpected RM_HEAP2_ID record type: %u", info);
+	}
+}
+
 /*
  * Handle rmgr HEAP_ID records for LogicalDecodingProcessRecord().
  */
@@ -544,6 +653,31 @@ heap_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 	}
 }
 
+/*
+ * Return true if given HEAP_ID record type is logically decodable. In other
+ * words, return true for record types that have something to do with logical
+ * decoding in heap_decode.
+ */
+bool
+heap_is_record_decodable(uint8 info)
+{
+	switch (info)
+	{
+		case XLOG_HEAP_INSERT:
+		case XLOG_HEAP_HOT_UPDATE:
+		case XLOG_HEAP_UPDATE:
+		case XLOG_HEAP_DELETE:
+		case XLOG_HEAP_TRUNCATE:
+		case XLOG_HEAP_INPLACE:
+		case XLOG_HEAP_CONFIRM:
+			return true;
+		case XLOG_HEAP_LOCK:
+			return false;
+		default:
+			elog(ERROR, "unexpected RM_HEAP_ID record type: %u", info);
+	}
+}
+
 /*
  * Ask output plugin whether we want to skip this PREPARE and send
  * this transaction as a regular commit later.
@@ -640,6 +774,23 @@ logicalmsg_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 							  message->message + message->prefix_size);
 }
 
+/*
+ * Return true if given LOGICALMSG_ID record type is logically decodable. In
+ * other words, return true for record types that have something to do with
+ * logical decoding in logicalmsg_decode.
+ */
+bool
+logicalmsg_is_record_decodable(uint8 info)
+{
+	switch (info)
+	{
+		case XLOG_LOGICAL_MESSAGE:
+			return true;
+		default:
+			elog(ERROR, "unexpected RM_LOGICALMSG_ID record type: %u", info);
+	}
+}
+
 /*
  * Consolidated commit record handling between the different form of commit
  * records.
diff --git a/src/backend/utils/adt/pg_upgrade_support.c b/src/backend/utils/adt/pg_upgrade_support.c
index bd64da7205..cfd3e448b1 100644
--- a/src/backend/utils/adt/pg_upgrade_support.c
+++ b/src/backend/utils/adt/pg_upgrade_support.c
@@ -13,6 +13,7 @@
 
 #include "access/heapam_xlog.h"
 #include "access/xlog.h"
+#include "access/xlog_internal.h"
 #include "access/xlogutils.h"
 #include "catalog/binary_upgrade.h"
 #include "catalog/heap.h"
@@ -291,10 +292,7 @@ is_xlog_record_type(RmgrId rmgrid, uint8 info,
  * generated that don't need to be decoded. Such records are ignored.
  *
  * XLOG_CHECKPOINT_SHUTDOWN and XLOG_SWITCH are ignored because they would be
- * inserted after the walsender exits. Moreover, the following types of records
- * could be generated during the pg_upgrade --check, so they are ignored too:
- * XLOG_CHECKPOINT_ONLINE, XLOG_RUNNING_XACTS, XLOG_FPI_FOR_HINT,
- * XLOG_HEAP2_PRUNE, XLOG_PARAMETER_CHANGE.
+ * inserted after the walsender exits.
  */
 Datum
 binary_upgrade_validate_wal_logical_end(PG_FUNCTION_ARGS)
@@ -321,6 +319,7 @@ binary_upgrade_validate_wal_logical_end(PG_FUNCTION_ARGS)
 	/* Loop until all WALs are read, or unexpected record is found */
 	while (is_valid && ReadNextXLogRecord(xlogreader))
 	{
+		RmgrData	rmgr;
 		RmgrIds		rmid;
 		uint8		info;
 
@@ -342,16 +341,22 @@ binary_upgrade_validate_wal_logical_end(PG_FUNCTION_ARGS)
 		}
 
 		/*
-		 * There is a possibility that following records may be generated
-		 * during the upgrade.
+		 * Check if the WAL record is logically decodable. We do this because
+		 * there is a possibility that some of the WAL records may be
+		 * generated during the upgrade.
 		 */
-		is_valid = is_xlog_record_type(rmid, info, RM_XLOG_ID, XLOG_CHECKPOINT_SHUTDOWN) ||
-			is_xlog_record_type(rmid, info, RM_XLOG_ID, XLOG_CHECKPOINT_ONLINE) ||
-			is_xlog_record_type(rmid, info, RM_XLOG_ID, XLOG_SWITCH) ||
-			is_xlog_record_type(rmid, info, RM_XLOG_ID, XLOG_FPI_FOR_HINT) ||
-			is_xlog_record_type(rmid, info, RM_XLOG_ID, XLOG_PARAMETER_CHANGE) ||
-			is_xlog_record_type(rmid, info, RM_STANDBY_ID, XLOG_RUNNING_XACTS) ||
-			is_xlog_record_type(rmid, info, RM_HEAP2_ID, XLOG_HEAP2_PRUNE);
+		rmgr = GetRmgr(XLogRecGetRmid(xlogreader));
+
+		if (rmgr.rm_decode != NULL)
+		{
+			if (rmgr.rm_is_record_decodable != NULL)
+				is_valid = rmgr.rm_is_record_decodable(info);
+			else
+				ereport(ERROR,
+						errmsg("cannot check logical decodability for resource manager \"%s\" with ID %d",
+							   rmgr.rm_name, rmid),
+						errdetail("Logical decodability callback is not defined for the resource manager."));
+		}
 
 		CHECK_FOR_INTERRUPTS();
 	}
diff --git a/src/bin/pg_rewind/parsexlog.c b/src/bin/pg_rewind/parsexlog.c
index 0233ece88b..0a4f8a3bcd 100644
--- a/src/bin/pg_rewind/parsexlog.c
+++ b/src/bin/pg_rewind/parsexlog.c
@@ -28,7 +28,7 @@
  * RmgrNames is an array of the built-in resource manager names, to make error
  * messages a bit nicer.
  */
-#define PG_RMGR(symname,name,redo,desc,identify,startup,cleanup,mask,decode) \
+#define PG_RMGR(symname,name,redo,desc,identify,startup,cleanup,mask,decode,is_record_decodable) \
   name,
 
 static const char *const RmgrNames[RM_MAX_ID + 1] = {
diff --git a/src/bin/pg_waldump/rmgrdesc.c b/src/bin/pg_waldump/rmgrdesc.c
index 6b8c17bb4c..0b2110351c 100644
--- a/src/bin/pg_waldump/rmgrdesc.c
+++ b/src/bin/pg_waldump/rmgrdesc.c
@@ -32,7 +32,7 @@
 #include "storage/standbydefs.h"
 #include "utils/relmapper.h"
 
-#define PG_RMGR(symname,name,redo,desc,identify,startup,cleanup,mask,decode) \
+#define PG_RMGR(symname,name,redo,desc,identify,startup,cleanup,mask,decode,is_record_decodable) \
 	{ name, desc, identify},
 
 static const RmgrDescData RmgrDescTable[RM_N_BUILTIN_IDS] = {
diff --git a/src/include/access/rmgr.h b/src/include/access/rmgr.h
index 3b6a497e1b..8c9aed1bc6 100644
--- a/src/include/access/rmgr.h
+++ b/src/include/access/rmgr.h
@@ -19,7 +19,7 @@ typedef uint8 RmgrId;
  * Note: RM_MAX_ID must fit in RmgrId; widening that type will affect the XLOG
  * file format.
  */
-#define PG_RMGR(symname,name,redo,desc,identify,startup,cleanup,mask,decode) \
+#define PG_RMGR(symname,name,redo,desc,identify,startup,cleanup,mask,decode,is_record_decodable) \
 	symname,
 
 typedef enum RmgrIds
diff --git a/src/include/access/rmgrlist.h b/src/include/access/rmgrlist.h
index 463bcb67c5..a471e77a7c 100644
--- a/src/include/access/rmgrlist.h
+++ b/src/include/access/rmgrlist.h
@@ -24,26 +24,26 @@
  * Changes to this list possibly need an XLOG_PAGE_MAGIC bump.
  */
 
-/* symbol name, textual name, redo, desc, identify, startup, cleanup, mask, decode */
-PG_RMGR(RM_XLOG_ID, "XLOG", xlog_redo, xlog_desc, xlog_identify, NULL, NULL, NULL, xlog_decode)
-PG_RMGR(RM_XACT_ID, "Transaction", xact_redo, xact_desc, xact_identify, NULL, NULL, NULL, xact_decode)
-PG_RMGR(RM_SMGR_ID, "Storage", smgr_redo, smgr_desc, smgr_identify, NULL, NULL, NULL, NULL)
-PG_RMGR(RM_CLOG_ID, "CLOG", clog_redo, clog_desc, clog_identify, NULL, NULL, NULL, NULL)
-PG_RMGR(RM_DBASE_ID, "Database", dbase_redo, dbase_desc, dbase_identify, NULL, NULL, NULL, NULL)
-PG_RMGR(RM_TBLSPC_ID, "Tablespace", tblspc_redo, tblspc_desc, tblspc_identify, NULL, NULL, NULL, NULL)
-PG_RMGR(RM_MULTIXACT_ID, "MultiXact", multixact_redo, multixact_desc, multixact_identify, NULL, NULL, NULL, NULL)
-PG_RMGR(RM_RELMAP_ID, "RelMap", relmap_redo, relmap_desc, relmap_identify, NULL, NULL, NULL, NULL)
-PG_RMGR(RM_STANDBY_ID, "Standby", standby_redo, standby_desc, standby_identify, NULL, NULL, NULL, standby_decode)
-PG_RMGR(RM_HEAP2_ID, "Heap2", heap2_redo, heap2_desc, heap2_identify, NULL, NULL, heap_mask, heap2_decode)
-PG_RMGR(RM_HEAP_ID, "Heap", heap_redo, heap_desc, heap_identify, NULL, NULL, heap_mask, heap_decode)
-PG_RMGR(RM_BTREE_ID, "Btree", btree_redo, btree_desc, btree_identify, btree_xlog_startup, btree_xlog_cleanup, btree_mask, NULL)
-PG_RMGR(RM_HASH_ID, "Hash", hash_redo, hash_desc, hash_identify, NULL, NULL, hash_mask, NULL)
-PG_RMGR(RM_GIN_ID, "Gin", gin_redo, gin_desc, gin_identify, gin_xlog_startup, gin_xlog_cleanup, gin_mask, NULL)
-PG_RMGR(RM_GIST_ID, "Gist", gist_redo, gist_desc, gist_identify, gist_xlog_startup, gist_xlog_cleanup, gist_mask, NULL)
-PG_RMGR(RM_SEQ_ID, "Sequence", seq_redo, seq_desc, seq_identify, NULL, NULL, seq_mask, NULL)
-PG_RMGR(RM_SPGIST_ID, "SPGist", spg_redo, spg_desc, spg_identify, spg_xlog_startup, spg_xlog_cleanup, spg_mask, NULL)
-PG_RMGR(RM_BRIN_ID, "BRIN", brin_redo, brin_desc, brin_identify, NULL, NULL, brin_mask, NULL)
-PG_RMGR(RM_COMMIT_TS_ID, "CommitTs", commit_ts_redo, commit_ts_desc, commit_ts_identify, NULL, NULL, NULL, NULL)
-PG_RMGR(RM_REPLORIGIN_ID, "ReplicationOrigin", replorigin_redo, replorigin_desc, replorigin_identify, NULL, NULL, NULL, NULL)
-PG_RMGR(RM_GENERIC_ID, "Generic", generic_redo, generic_desc, generic_identify, NULL, NULL, generic_mask, NULL)
-PG_RMGR(RM_LOGICALMSG_ID, "LogicalMessage", logicalmsg_redo, logicalmsg_desc, logicalmsg_identify, NULL, NULL, NULL, logicalmsg_decode)
+/* symbol name, textual name, redo, desc, identify, startup, cleanup, mask, decode, is_record_decodable */
+PG_RMGR(RM_XLOG_ID, "XLOG", xlog_redo, xlog_desc, xlog_identify, NULL, NULL, NULL, xlog_decode, xlog_is_record_decodable)
+PG_RMGR(RM_XACT_ID, "Transaction", xact_redo, xact_desc, xact_identify, NULL, NULL, NULL, xact_decode, xact_is_record_decodable)
+PG_RMGR(RM_SMGR_ID, "Storage", smgr_redo, smgr_desc, smgr_identify, NULL, NULL, NULL, NULL, NULL)
+PG_RMGR(RM_CLOG_ID, "CLOG", clog_redo, clog_desc, clog_identify, NULL, NULL, NULL, NULL, NULL)
+PG_RMGR(RM_DBASE_ID, "Database", dbase_redo, dbase_desc, dbase_identify, NULL, NULL, NULL, NULL, NULL)
+PG_RMGR(RM_TBLSPC_ID, "Tablespace", tblspc_redo, tblspc_desc, tblspc_identify, NULL, NULL, NULL, NULL, NULL)
+PG_RMGR(RM_MULTIXACT_ID, "MultiXact", multixact_redo, multixact_desc, multixact_identify, NULL, NULL, NULL, NULL, NULL)
+PG_RMGR(RM_RELMAP_ID, "RelMap", relmap_redo, relmap_desc, relmap_identify, NULL, NULL, NULL, NULL, NULL)
+PG_RMGR(RM_STANDBY_ID, "Standby", standby_redo, standby_desc, standby_identify, NULL, NULL, NULL, standby_decode, standy_is_record_decodable)
+PG_RMGR(RM_HEAP2_ID, "Heap2", heap2_redo, heap2_desc, heap2_identify, NULL, NULL, heap_mask, heap2_decode, heap2_is_record_decodable)
+PG_RMGR(RM_HEAP_ID, "Heap", heap_redo, heap_desc, heap_identify, NULL, NULL, heap_mask, heap_decode, heap_is_record_decodable)
+PG_RMGR(RM_BTREE_ID, "Btree", btree_redo, btree_desc, btree_identify, btree_xlog_startup, btree_xlog_cleanup, btree_mask, NULL, NULL)
+PG_RMGR(RM_HASH_ID, "Hash", hash_redo, hash_desc, hash_identify, NULL, NULL, hash_mask, NULL, NULL)
+PG_RMGR(RM_GIN_ID, "Gin", gin_redo, gin_desc, gin_identify, gin_xlog_startup, gin_xlog_cleanup, gin_mask, NULL, NULL)
+PG_RMGR(RM_GIST_ID, "Gist", gist_redo, gist_desc, gist_identify, gist_xlog_startup, gist_xlog_cleanup, gist_mask, NULL, NULL)
+PG_RMGR(RM_SEQ_ID, "Sequence", seq_redo, seq_desc, seq_identify, NULL, NULL, seq_mask, NULL, NULL)
+PG_RMGR(RM_SPGIST_ID, "SPGist", spg_redo, spg_desc, spg_identify, spg_xlog_startup, spg_xlog_cleanup, spg_mask, NULL, NULL)
+PG_RMGR(RM_BRIN_ID, "BRIN", brin_redo, brin_desc, brin_identify, NULL, NULL, brin_mask, NULL, NULL)
+PG_RMGR(RM_COMMIT_TS_ID, "CommitTs", commit_ts_redo, commit_ts_desc, commit_ts_identify, NULL, NULL, NULL, NULL, NULL)
+PG_RMGR(RM_REPLORIGIN_ID, "ReplicationOrigin", replorigin_redo, replorigin_desc, replorigin_identify, NULL, NULL, NULL, NULL, NULL)
+PG_RMGR(RM_GENERIC_ID, "Generic", generic_redo, generic_desc, generic_identify, NULL, NULL, generic_mask, NULL, NULL)
+PG_RMGR(RM_LOGICALMSG_ID, "LogicalMessage", logicalmsg_redo, logicalmsg_desc, logicalmsg_identify, NULL, NULL, NULL, logicalmsg_decode, logicalmsg_is_record_decodable)
diff --git a/src/include/access/xlog_internal.h b/src/include/access/xlog_internal.h
index b0fd338a00..6e113ef53d 100644
--- a/src/include/access/xlog_internal.h
+++ b/src/include/access/xlog_internal.h
@@ -31,7 +31,7 @@
 /*
  * Each page of XLOG file has a header like this:
  */
-#define XLOG_PAGE_MAGIC 0xD113	/* can be used as WAL version indicator */
+#define XLOG_PAGE_MAGIC 0xD114	/* can be used as WAL version indicator */
 
 typedef struct XLogPageHeaderData
 {
@@ -356,6 +356,7 @@ typedef struct RmgrData
 	void		(*rm_mask) (char *pagedata, BlockNumber blkno);
 	void		(*rm_decode) (struct LogicalDecodingContext *ctx,
 							  struct XLogRecordBuffer *buf);
+	bool		(*rm_is_record_decodable) (uint8 type);
 } RmgrData;
 
 extern PGDLLIMPORT RmgrData RmgrTable[];
diff --git a/src/include/replication/decode.h b/src/include/replication/decode.h
index 14fa921ab4..3885ce671d 100644
--- a/src/include/replication/decode.h
+++ b/src/include/replication/decode.h
@@ -28,6 +28,13 @@ extern void xact_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
 extern void standby_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
 extern void logicalmsg_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
 
+extern bool xlog_is_record_decodable(uint8 info);
+extern bool xact_is_record_decodable(uint8 info);
+extern bool standy_is_record_decodable(uint8 info);
+extern bool heap2_is_record_decodable(uint8 info);
+extern bool heap_is_record_decodable(uint8 info);
+extern bool logicalmsg_is_record_decodable(uint8 info);
+
 extern void LogicalDecodingProcessRecord(LogicalDecodingContext *ctx,
 										 XLogReaderState *record);
 
-- 
2.34.1