Logical insert/update/delete WAL records for custom table AMs

diff --git a/src/backend/access/rmgrdesc/Makefile b/src/backend/access/rmgrdesc/Makefile
index f88d72fd862..ed6dff179be 100644
--- a/src/backend/access/rmgrdesc/Makefile
+++ b/src/backend/access/rmgrdesc/Makefile
@@ -18,7 +18,7 @@ OBJS = \
 	gistdesc.o \
 	hashdesc.o \
 	heapdesc.o \
-	logicalmsgdesc.o \
+	logicaldesc.o \
 	mxactdesc.o \
 	nbtdesc.o \
 	relmapdesc.o \
diff --git a/src/backend/access/rmgrdesc/logicalmsgdesc.c b/src/backend/access/rmgrdesc/logicaldesc.c
similarity index 59%
rename from src/backend/access/rmgrdesc/logicalmsgdesc.c
rename to src/backend/access/rmgrdesc/logicaldesc.c
index 099e11a84e7..c2b0434a606 100644
--- a/src/backend/access/rmgrdesc/logicalmsgdesc.c
+++ b/src/backend/access/rmgrdesc/logicaldesc.c
@@ -1,22 +1,22 @@
 /*-------------------------------------------------------------------------
  *
- * logicalmsgdesc.c
- *	  rmgr descriptor routines for replication/logical/message.c
+ * logicaldesc.c
+ *	  rmgr descriptor routines for replication/logical/logical_xlog.c
  *
  * Portions Copyright (c) 2015-2022, PostgreSQL Global Development Group
  *
  *
  * IDENTIFICATION
- *	  src/backend/access/rmgrdesc/logicalmsgdesc.c
+ *	  src/backend/access/rmgrdesc/logicaldesc.c
  *
  *-------------------------------------------------------------------------
  */
 #include "postgres.h"
 
-#include "replication/message.h"
+#include "replication/logical_xlog.h"
 
 void
-logicalmsg_desc(StringInfo buf, XLogReaderState *record)
+logical_desc(StringInfo buf, XLogReaderState *record)
 {
 	char	   *rec = XLogRecGetData(record);
 	uint8		info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
@@ -40,13 +40,42 @@ logicalmsg_desc(StringInfo buf, XLogReaderState *record)
 			sep = " ";
 		}
 	}
+	else if (info == XLOG_LOGICAL_INSERT)
+	{
+
+	}
+	else if (info == XLOG_LOGICAL_UPDATE)
+	{
+
+	}
+	else if (info == XLOG_LOGICAL_DELETE)
+	{
+
+	}
+	else if (info == XLOG_LOGICAL_TRUNCATE)
+	{
+
+	}
 }
 
 const char *
-logicalmsg_identify(uint8 info)
+logical_identify(uint8 info)
 {
-	if ((info & ~XLR_INFO_MASK) == XLOG_LOGICAL_MESSAGE)
-		return "MESSAGE";
+	switch (info)
+	{
+		case XLOG_LOGICAL_MESSAGE:
+			return "MESSAGE";
+		case XLOG_LOGICAL_INSERT:
+			return "INSERT";
+		case XLOG_LOGICAL_UPDATE:
+			return "UPDATE";
+		case XLOG_LOGICAL_DELETE:
+			return "DELETE";
+		case XLOG_LOGICAL_TRUNCATE:
+			return "TRUNCATE";
+		default:
+			return NULL;
+	}
 
 	return NULL;
 }
diff --git a/src/backend/access/transam/rmgr.c b/src/backend/access/transam/rmgr.c
index 58091f6b520..f31f7187ac4 100644
--- a/src/backend/access/transam/rmgr.c
+++ b/src/backend/access/transam/rmgr.c
@@ -24,7 +24,7 @@
 #include "commands/dbcommands_xlog.h"
 #include "commands/sequence.h"
 #include "commands/tablespace.h"
-#include "replication/message.h"
+#include "replication/logical_xlog.h"
 #include "replication/origin.h"
 #include "storage/standby.h"
 #include "utils/relmapper.h"
diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile
index c4e2fdeb719..9fa281a1dca 100644
--- a/src/backend/replication/logical/Makefile
+++ b/src/backend/replication/logical/Makefile
@@ -19,7 +19,7 @@ OBJS = \
 	launcher.o \
 	logical.o \
 	logicalfuncs.o \
-	message.o \
+	logical_xlog.o \
 	origin.o \
 	proto.o \
 	relation.o \
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index 1d22208c1ad..1abccd7190b 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -37,7 +37,7 @@
 #include "catalog/pg_control.h"
 #include "replication/decode.h"
 #include "replication/logical.h"
-#include "replication/message.h"
+#include "replication/logical_xlog.h"
 #include "replication/origin.h"
 #include "replication/reorderbuffer.h"
 #include "replication/snapbuild.h"
@@ -56,16 +56,19 @@ static void DecodeHeapOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
 static void DecodeHeap2Op(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
 static void DecodeXactOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
 static void DecodeStandbyOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
-static void DecodeLogicalMsgOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
+static void DecodeLogicalOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
 
 /* individual record(group)'s handlers */
-static void DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
+static void DecodeHeapInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
 static void DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
 static void DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
 static void DecodeTruncate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
 static void DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
 static void DecodeSpecConfirm(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
 
+static void DecodeLogicalMsg(LogicalDecodingContext *cxt, XLogRecordBuffer *buf);
+static void DecodeLogicalInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
+
 static void DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 						 xl_xact_parsed_commit *parsed, TransactionId xid,
 						 bool two_phase);
@@ -154,8 +157,8 @@ LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *recor
 			DecodeHeapOp(ctx, &buf);
 			break;
 
-		case RM_LOGICALMSG_ID:
-			DecodeLogicalMsgOp(ctx, &buf);
+		case RM_LOGICAL_ID:
+			DecodeLogicalOp(ctx, &buf);
 			break;
 
 			/*
@@ -518,7 +521,7 @@ DecodeHeapOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 	{
 		case XLOG_HEAP_INSERT:
 			if (SnapBuildProcessChange(builder, xid, buf->origptr))
-				DecodeInsert(ctx, buf);
+				DecodeHeapInsert(ctx, buf);
 			break;
 
 			/*
@@ -616,35 +619,22 @@ FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
 	return filter_by_origin_cb_wrapper(ctx, origin_id);
 }
 
-/*
- * Handle rmgr LOGICALMSG_ID records for DecodeRecordIntoReorderBuffer().
- */
 static void
-DecodeLogicalMsgOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
+DecodeLogicalMsg(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 {
-	SnapBuild  *builder = ctx->snapshot_builder;
 	XLogReaderState *r = buf->record;
+	SnapBuild  *builder = ctx->snapshot_builder;
 	TransactionId xid = XLogRecGetXid(r);
-	uint8		info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
 	RepOriginId origin_id = XLogRecGetOrigin(r);
 	Snapshot	snapshot;
 	xl_logical_message *message;
 
-	if (info != XLOG_LOGICAL_MESSAGE)
-		elog(ERROR, "unexpected RM_LOGICALMSG_ID record type: %u", info);
-
-	ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(r), buf->origptr);
+	message = (xl_logical_message *) XLogRecGetData(r);
 
-	/*
-	 * If we don't have snapshot or we are just fast-forwarding, there is no
-	 * point in decoding messages.
-	 */
-	if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT ||
-		ctx->fast_forward)
+	if (message->transactional &&
+		!SnapBuildProcessChange(builder, xid, buf->origptr))
 		return;
 
-	message = (xl_logical_message *) XLogRecGetData(r);
-
 	if (message->dbId != ctx->slot->data.database ||
 		FilterByOrigin(ctx, origin_id))
 		return;
@@ -664,6 +654,115 @@ DecodeLogicalMsgOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 												 * prefix */
 							  message->message_size,
 							  message->message + message->prefix_size);
+
+}
+
+/*
+ * Parse XLOG_HEAP_INSERT (not MULTI_INSERT!) records into tuplebufs.
+ *
+ * Deletes can contain the new tuple.
+ */
+static void
+DecodeLogicalInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
+{
+	char	   *tupledata;
+	Size		tuplelen;
+	XLogReaderState *r = buf->record;
+	xl_logical_insert *xlrec;
+	ReorderBufferChange *change;
+
+	xlrec = (xl_logical_insert *) XLogRecGetData(r);
+	tupledata = XLogRecGetData(r) + SizeOfLogicalInsert;
+
+	/*
+	 * Ignore insert records without new tuples (this does happen when
+	 * raw_heap_insert marks the TOAST record as HEAP_INSERT_NO_LOGICAL).
+	 */
+	if (!(xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE))
+		return;
+
+	/* only interested in our database */
+	if (xlrec->node.dbNode != ctx->slot->data.database)
+		return;
+
+	/* output plugin doesn't look for this origin, no need to queue */
+	if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
+		return;
+
+	change = ReorderBufferGetChange(ctx->reorder);
+	if (!(xlrec->flags & XLH_INSERT_IS_SPECULATIVE))
+		change->action = REORDER_BUFFER_CHANGE_INSERT;
+	else
+		change->action = REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT;
+	change->origin_id = XLogRecGetOrigin(r);
+
+	memcpy(&change->data.tp.relnode, &xlrec->node, sizeof(RelFileNode));
+
+	tuplelen = xlrec->datalen - SizeOfHeapHeader;
+
+	change->data.tp.newtuple =
+		ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
+
+	DecodeXLogTuple(tupledata, xlrec->datalen, change->data.tp.newtuple);
+
+	change->data.tp.clear_toast_afterwards = true;
+
+	ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr,
+							 change,
+							 xlrec->flags & XLH_INSERT_ON_TOAST_RELATION);
+}
+
+/*
+ * Handle rmgr LOGICAL_ID records for DecodeRecordIntoReorderBuffer().
+ */
+static void
+DecodeLogicalOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
+{
+	XLogReaderState *r = buf->record;
+	SnapBuild  *builder = ctx->snapshot_builder;
+	TransactionId xid = XLogRecGetXid(r);
+	uint8		info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
+
+	ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
+
+	/*
+	 * If we don't have snapshot or we are just fast-forwarding, there is no
+	 * point in decoding data changes.
+	 */
+	if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT ||
+		ctx->fast_forward)
+		return;
+
+	switch (info)
+	{
+		case XLOG_LOGICAL_MESSAGE:
+			DecodeLogicalMsg(ctx, buf);
+			break;
+
+		case XLOG_LOGICAL_INSERT:
+			if (SnapBuildProcessChange(builder, xid, buf->origptr))
+				DecodeLogicalInsert(ctx, buf);
+			break;
+
+		case XLOG_LOGICAL_UPDATE:
+			if (SnapBuildProcessChange(builder, xid, buf->origptr))
+				DecodeUpdate(ctx, buf);
+			break;
+
+		case XLOG_LOGICAL_DELETE:
+			if (SnapBuildProcessChange(builder, xid, buf->origptr))
+				DecodeDelete(ctx, buf);
+			break;
+
+		case XLOG_LOGICAL_TRUNCATE:
+			if (SnapBuildProcessChange(builder, xid, buf->origptr))
+				DecodeTruncate(ctx, buf);
+			break;
+
+		default:
+			elog(ERROR, "unexpected RM_LOGICAL_ID record type: %u", info);
+			break;
+	}
 }
 
 /*
@@ -900,7 +999,7 @@ DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
  * Deletes can contain the new tuple.
  */
 static void
-DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
+DecodeHeapInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 {
 	Size		datalen;
 	char	   *tupledata;
diff --git a/src/backend/replication/logical/message.c b/src/backend/replication/logical/logical_xlog.c
similarity index 50%
rename from src/backend/replication/logical/message.c
rename to src/backend/replication/logical/logical_xlog.c
index b02363f0bda..efec90b3619 100644
--- a/src/backend/replication/logical/message.c
+++ b/src/backend/replication/logical/logical_xlog.c
@@ -1,14 +1,14 @@
 /*-------------------------------------------------------------------------
  *
- * message.c
- *	  Generic logical messages.
+ * logical_xlog.c
+ *	  Logical xlog records.
  *
  * Copyright (c) 2013-2022, PostgreSQL Global Development Group
  *
  * IDENTIFICATION
- *	  src/backend/replication/logical/message.c
+ *	  src/backend/replication/logical/logical_xlog.c
  *
- * NOTES
+ * Logical Messages
  *
  * Generic logical messages allow XLOG logging of arbitrary binary blobs that
  * get passed to the logical decoding plugin. In normal XLOG processing they
@@ -26,16 +26,27 @@
  * plugins. The plugin authors must take extra care to use unique prefix,
  * good options seems to be for example to use the name of the extension.
  *
+ * Logical Insert/Update/Delete/Truncate
+ *
+ * These records are intended to be used by non-heap table access methods that
+ * wish to support logical decoding and replication. They are treated
+ * similarly to the analogous heap records, but are not tied to physical pages
+ * or other details of the heap. These records are not processed during redo,
+ * so do not contribute to durability or physical replication; use generic WAL
+ * records for that. Note that using both logical WAL records and generic WAL
+ * records is redundant compared with the heap.
+ *
  * ---------------------------------------------------------------------------
  */
 
 #include "postgres.h"
 
+#include "access/heapam_xlog.h"
 #include "access/xact.h"
 #include "miscadmin.h"
 #include "nodes/execnodes.h"
 #include "replication/logical.h"
-#include "replication/message.h"
+#include "replication/logical_xlog.h"
 #include "utils/memutils.h"
 
 /*
@@ -70,19 +81,75 @@ LogLogicalMessage(const char *prefix, const char *message, size_t size,
 	/* allow origin filtering */
 	XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);
 
-	return XLogInsert(RM_LOGICALMSG_ID, XLOG_LOGICAL_MESSAGE);
+	return XLogInsert(RM_LOGICAL_ID, XLOG_LOGICAL_MESSAGE);
 }
 
+XLogRecPtr
+LogLogicalInsert(Relation relation, TupleTableSlot *slot)
+{
+	bool		shouldFree;
+	HeapTuple	tuple = ExecFetchSlotHeapTuple(slot, true, &shouldFree);
+	xl_logical_insert xlrec;
+	xl_heap_header xlhdr;
+	XLogRecPtr recptr;
+
+	/* force xid to be allocated */
+	Assert(IsTransactionState());
+	GetCurrentTransactionId();
+
+	tuple->t_tableOid = slot->tts_tableOid;
+	ItemPointerCopy(&tuple->t_self, &slot->tts_tid);
+
+	xlrec.node = relation->rd_node;
+	xlrec.datalen = tuple->t_len;
+	xlrec.offnum = ItemPointerGetOffsetNumber(&tuple->t_self);
+	xlrec.flags = 0;
+	xlrec.flags |= XLH_INSERT_CONTAINS_NEW_TUPLE;
+
+	XLogBeginInsert();
+	XLogRegisterData((char *) &xlrec, SizeOfLogicalInsert);
+
+	xlhdr.t_infomask2 = tuple->t_data->t_infomask2;
+	xlhdr.t_infomask = tuple->t_data->t_infomask;
+	xlhdr.t_hoff = tuple->t_data->t_hoff;
+
+	XLogRegisterData((char *) &xlhdr, SizeOfHeapHeader);
+	/* PG73FORMAT: write bitmap [+ padding] [+ oid] + data */
+	XLogRegisterData((char *) tuple->t_data + SizeofHeapTupleHeader,
+					 tuple->t_len - SizeofHeapTupleHeader);
+
+
+	/* allow origin filtering */
+	XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);
+
+	recptr = XLogInsert(RM_LOGICAL_ID, XLOG_LOGICAL_INSERT);
+
+	if (shouldFree)
+		pfree(tuple);
+
+	return recptr;
+}
+
+
 /*
  * Redo is basically just noop for logical decoding messages.
  */
 void
-logicalmsg_redo(XLogReaderState *record)
+logical_redo(XLogReaderState *record)
 {
 	uint8		info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
 
-	if (info != XLOG_LOGICAL_MESSAGE)
-		elog(PANIC, "logicalmsg_redo: unknown op code %u", info);
+	switch (info)
+	{
+		case XLOG_LOGICAL_MESSAGE:
+		case XLOG_LOGICAL_INSERT:
+		case XLOG_LOGICAL_UPDATE:
+		case XLOG_LOGICAL_DELETE:
+		case XLOG_LOGICAL_TRUNCATE:
+			break;
+		default:
+			elog(PANIC, "logical_redo: unknown op code %u", info);
+	}
 
 	/* This is only interesting for logical decoding, see decode.c. */
 }
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index 4f633888b4f..edb06b86d2d 100644
--- a/src/backend/replication/logical/logicalfuncs.c
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -28,7 +28,7 @@
 #include "nodes/makefuncs.h"
 #include "replication/decode.h"
 #include "replication/logical.h"
-#include "replication/message.h"
+#include "replication/logical_xlog.h"
 #include "storage/fd.h"
 #include "utils/array.h"
 #include "utils/builtins.h"
diff --git a/src/bin/pg_waldump/.gitignore b/src/bin/pg_waldump/.gitignore
index 3be00a8b61f..567655fa626 100644
--- a/src/bin/pg_waldump/.gitignore
+++ b/src/bin/pg_waldump/.gitignore
@@ -10,7 +10,7 @@
 /gistdesc.c
 /hashdesc.c
 /heapdesc.c
-/logicalmsgdesc.c
+/logicaldesc.c
 /mxactdesc.c
 /nbtdesc.c
 /relmapdesc.c
diff --git a/src/bin/pg_waldump/rmgrdesc.c b/src/bin/pg_waldump/rmgrdesc.c
index 852d8ca4b1c..c7db22e70f1 100644
--- a/src/bin/pg_waldump/rmgrdesc.c
+++ b/src/bin/pg_waldump/rmgrdesc.c
@@ -26,7 +26,7 @@
 #include "commands/dbcommands_xlog.h"
 #include "commands/sequence.h"
 #include "commands/tablespace.h"
-#include "replication/message.h"
+#include "replication/logical_xlog.h"
 #include "replication/origin.h"
 #include "rmgrdesc.h"
 #include "storage/standbydefs.h"
diff --git a/src/include/access/rmgrlist.h b/src/include/access/rmgrlist.h
index ed751aaf039..5a55b18810e 100644
--- a/src/include/access/rmgrlist.h
+++ b/src/include/access/rmgrlist.h
@@ -46,4 +46,4 @@ PG_RMGR(RM_BRIN_ID, "BRIN", brin_redo, brin_desc, brin_identify, NULL, NULL, bri
 PG_RMGR(RM_COMMIT_TS_ID, "CommitTs", commit_ts_redo, commit_ts_desc, commit_ts_identify, NULL, NULL, NULL)
 PG_RMGR(RM_REPLORIGIN_ID, "ReplicationOrigin", replorigin_redo, replorigin_desc, replorigin_identify, NULL, NULL, NULL)
 PG_RMGR(RM_GENERIC_ID, "Generic", generic_redo, generic_desc, generic_identify, NULL, NULL, generic_mask)
-PG_RMGR(RM_LOGICALMSG_ID, "LogicalMessage", logicalmsg_redo, logicalmsg_desc, logicalmsg_identify, NULL, NULL, NULL)
+PG_RMGR(RM_LOGICAL_ID, "Logical", logical_redo, logical_desc, logical_identify, NULL, NULL, NULL)
diff --git a/src/include/replication/message.h b/src/include/replication/logical_xlog.h
similarity index 54%
rename from src/include/replication/message.h
rename to src/include/replication/logical_xlog.h
index 7d7785292f1..bf20a740326 100644
--- a/src/include/replication/message.h
+++ b/src/include/replication/logical_xlog.h
@@ -1,10 +1,10 @@
 /*-------------------------------------------------------------------------
- * message.h
- *	   Exports from replication/logical/message.c
+ * logical_xlog.h
+ *	   Exports from replication/logical/logical_xlog.c
  *
  * Copyright (c) 2013-2022, PostgreSQL Global Development Group
  *
- * src/include/replication/message.h
+ * src/include/replication/logical_xlog.h
  *-------------------------------------------------------------------------
  */
 #ifndef PG_LOGICAL_MESSAGE_H
@@ -13,6 +13,7 @@
 #include "access/xlog.h"
 #include "access/xlogdefs.h"
 #include "access/xlogreader.h"
+#include "storage/off.h"
 
 /*
  * Generic logical decoding message wal record.
@@ -29,13 +30,34 @@ typedef struct xl_logical_message
 
 #define SizeOfLogicalMessage	(offsetof(xl_logical_message, message))
 
+/* This is what we need to know about insert */
+typedef struct xl_logical_insert
+{
+	RelFileNode		node;
+	OffsetNumber	offnum;		/* inserted tuple's offset */
+	Size			datalen;
+	uint8			flags;
+
+	/* xl_heap_header & TUPLE DATA in backup block 0 */
+} xl_logical_insert;
+
+#define SizeOfLogicalInsert	(offsetof(xl_logical_insert, flags) + sizeof(uint8))
+
+struct TupleTableSlot;
+
 extern XLogRecPtr LogLogicalMessage(const char *prefix, const char *message,
 									size_t size, bool transactional);
+extern XLogRecPtr LogLogicalInsert(Relation relation, struct TupleTableSlot *slot);
 
 /* RMGR API*/
 #define XLOG_LOGICAL_MESSAGE	0x00
-void		logicalmsg_redo(XLogReaderState *record);
-void		logicalmsg_desc(StringInfo buf, XLogReaderState *record);
-const char *logicalmsg_identify(uint8 info);
+#define XLOG_LOGICAL_INSERT		0x10
+#define XLOG_LOGICAL_UPDATE		0x20
+#define XLOG_LOGICAL_DELETE		0x30
+#define XLOG_LOGICAL_TRUNCATE	0x40
+
+void		logical_redo(XLogReaderState *record);
+void		logical_desc(StringInfo buf, XLogReaderState *record);
+const char *logical_identify(uint8 info);
 
 #endif							/* PG_LOGICAL_MESSAGE_H */

From f7b85aea60b06eb7019befec38566b5e014bea12 Mon Sep 17 00:00:00 2001
From: Jeff Davis <jeff@j-davis.com>
Date: Sat, 30 Oct 2021 12:07:35 -0700
Subject: [PATCH] Logical wal.

---
 contrib/test_decoding/expected/messages.out   | 148 +++++++
 contrib/test_decoding/sql/messages.sql        |  58 +++
 src/backend/access/heap/heapam.c              |   4 +-
 src/backend/access/rmgrdesc/Makefile          |   2 +-
 .../{logicalmsgdesc.c => logicaldesc.c}       |  45 +-
 src/backend/access/transam/rmgr.c             |   2 +-
 src/backend/replication/logical/Makefile      |   2 +-
 src/backend/replication/logical/decode.c      | 275 ++++++++++--
 .../replication/logical/logical_xlog.c        | 399 ++++++++++++++++++
 .../replication/logical/logicalfuncs.c        | 165 +++++++-
 src/backend/replication/logical/message.c     |  89 ----
 src/bin/pg_waldump/.gitignore                 |   2 +-
 src/bin/pg_waldump/rmgrdesc.c                 |   2 +-
 src/include/access/heapam.h                   |   2 +
 src/include/access/heapam_xlog.h              |   3 +
 src/include/access/rmgrlist.h                 |   2 +-
 src/include/catalog/pg_proc.dat               |  17 +
 src/include/replication/decode.h              |   2 +-
 src/include/replication/logical_xlog.h        | 124 ++++++
 src/include/replication/message.h             |  41 --
 20 files changed, 1211 insertions(+), 173 deletions(-)
 rename src/backend/access/rmgrdesc/{logicalmsgdesc.c => logicaldesc.c} (59%)
 create mode 100644 src/backend/replication/logical/logical_xlog.c
 delete mode 100644 src/backend/replication/logical/message.c
 create mode 100644 src/include/replication/logical_xlog.h
 delete mode 100644 src/include/replication/message.h

diff --git a/contrib/test_decoding/expected/messages.out b/contrib/test_decoding/expected/messages.out
index c75d40190b6..aa284bc37c2 100644
--- a/contrib/test_decoding/expected/messages.out
+++ b/contrib/test_decoding/expected/messages.out
@@ -91,6 +91,154 @@ SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'for
 ------
 (0 rows)
 
+-- no data in this table, but emit logical INSERT/UPDATE/DELETE for it
+CREATE TABLE dummy(i int, t text, n numeric, primary key(t));
+SELECT pg_logical_emit_insert('dummy', row(1, 'one', 0.1)::dummy) <> '0/0'::pg_lsn;
+ ?column? 
+----------
+ t
+(1 row)
+
+SELECT pg_logical_emit_insert('dummy', row(2, 'two', 0.2)::dummy) <> '0/0'::pg_lsn;
+ ?column? 
+----------
+ t
+(1 row)
+
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'force-binary', '0', 'skip-empty-xacts', '1', 'include-xids', '0');
+                                 data                                  
+-----------------------------------------------------------------------
+ BEGIN
+ table public.dummy: INSERT: i[integer]:1 t[text]:'one' n[numeric]:0.1
+ COMMIT
+ BEGIN
+ table public.dummy: INSERT: i[integer]:2 t[text]:'two' n[numeric]:0.2
+ COMMIT
+(6 rows)
+
+SELECT * FROM dummy;
+ i | t | n 
+---+---+---
+(0 rows)
+
+SELECT pg_logical_emit_delete('dummy', row(12, 'twelve', 0.12)::dummy) <> '0/0'::pg_lsn;
+ ?column? 
+----------
+ t
+(1 row)
+
+SELECT pg_logical_emit_update('dummy', row(15, 'fifteen', 0.15)::dummy,
+       row(16, 'sixteen', 0.16)::dummy) <> '0/0'::pg_lsn;
+ ?column? 
+----------
+ t
+(1 row)
+
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'force-binary', '0', 'skip-empty-xacts', '1', 'include-xids', '0');
+                                                       data                                                        
+-------------------------------------------------------------------------------------------------------------------
+ BEGIN
+ table public.dummy: DELETE: t[text]:'twelve'
+ COMMIT
+ BEGIN
+ table public.dummy: UPDATE: old-key: t[text]:'fifteen' new-tuple: i[integer]:16 t[text]:'sixteen' n[numeric]:0.16
+ COMMIT
+(6 rows)
+
+ALTER TABLE dummy REPLICA IDENTITY NOTHING;
+SELECT pg_logical_emit_delete('dummy', row(12, 'twelve', 0.12)::dummy) <> '0/0'::pg_lsn;
+ ?column? 
+----------
+ t
+(1 row)
+
+SELECT pg_logical_emit_update('dummy', row(15, 'fifteen', 0.15)::dummy,
+       row(16, 'sixteen', 0.16)::dummy) <> '0/0'::pg_lsn;
+ ?column? 
+----------
+ t
+(1 row)
+
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'force-binary', '0', 'skip-empty-xacts', '1', 'include-xids', '0');
+                                    data                                     
+-----------------------------------------------------------------------------
+ BEGIN
+ table public.dummy: DELETE: (no-tuple-data)
+ COMMIT
+ BEGIN
+ table public.dummy: UPDATE: i[integer]:16 t[text]:'sixteen' n[numeric]:0.16
+ COMMIT
+(6 rows)
+
+SELECT pg_logical_emit_truncate(ARRAY['dummy'], true, false) <> '0/0'::pg_lsn;
+ ?column? 
+----------
+ t
+(1 row)
+
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'force-binary', '0', 'skip-empty-xacts', '1', 'include-xids', '0');
+                 data                  
+---------------------------------------
+ BEGIN
+ table public.dummy: TRUNCATE: cascade
+ COMMIT
+(3 rows)
+
+CREATE UNLOGGED TABLE dummy_u(i int, t text, n numeric, primary key (t));
+-- return invalid
+SELECT pg_logical_emit_insert('dummy_u', row(7, 'seven', 0.7)::dummy_u);
+ pg_logical_emit_insert 
+------------------------
+ 0/0
+(1 row)
+
+-- return invalid
+SELECT pg_logical_emit_update('dummy_u', row(7, 'seven', 0.7)::dummy_u,
+       row(11, 'eleven', 0.11)::dummy_u);
+ pg_logical_emit_update 
+------------------------
+ 0/0
+(1 row)
+
+-- return invalid
+SELECT pg_logical_emit_update('dummy_u', row(NULL, 'seven', 0.7)::dummy_u,
+       row(11, 'eleven', 0.11)::dummy_u);
+ pg_logical_emit_update 
+------------------------
+ 0/0
+(1 row)
+
+-- error
+SELECT pg_logical_emit_update('dummy_u', row(7, NULL, 0.7)::dummy_u,
+       row(11, 'eleven', 0.11)::dummy_u);
+ERROR:  replica identity column is NULL
+-- return invalid
+SELECT pg_logical_emit_delete('dummy_u', row(7, 'seven', 0.7)::dummy_u);
+ pg_logical_emit_delete 
+------------------------
+ 0/0
+(1 row)
+
+-- error
+SELECT pg_logical_emit_delete('dummy_u', row(7, NULL, 0.7)::dummy_u);
+ERROR:  replica identity column is NULL
+-- return invalid
+SELECT pg_logical_emit_truncate(ARRAY['dummy_u'], false, false);
+ pg_logical_emit_truncate 
+--------------------------
+ 0/0
+(1 row)
+
+SELECT pg_logical_emit_truncate(ARRAY['dummy','dummy_u'], false, true) <> '0/0'::pg_lsn;
+ ?column? 
+----------
+ t
+(1 row)
+
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'force-binary', '0', 'skip-empty-xacts', '1', 'include-xids', '0');
+ERROR:  could not open relation with OID 1663
+DROP TABLE dummy;
+DROP TABLE dummy_u;
 SELECT 'cleanup' FROM pg_drop_replication_slot('regression_slot');
  ?column? 
 ----------
diff --git a/contrib/test_decoding/sql/messages.sql b/contrib/test_decoding/sql/messages.sql
index cf3f7738e57..c04a2e1a0c8 100644
--- a/contrib/test_decoding/sql/messages.sql
+++ b/contrib/test_decoding/sql/messages.sql
@@ -31,4 +31,62 @@ SELECT 'otherdb2' FROM pg_logical_emit_message(true, 'test', 'otherdb2');
 \c :prevdb
 SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'force-binary', '0', 'skip-empty-xacts', '1');
 
+-- no data in this table, but emit logical INSERT/UPDATE/DELETE for it
+CREATE TABLE dummy(i int, t text, n numeric, primary key(t));
+
+SELECT pg_logical_emit_insert('dummy', row(1, 'one', 0.1)::dummy) <> '0/0'::pg_lsn;
+SELECT pg_logical_emit_insert('dummy', row(2, 'two', 0.2)::dummy) <> '0/0'::pg_lsn;
+
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'force-binary', '0', 'skip-empty-xacts', '1', 'include-xids', '0');
+
+SELECT * FROM dummy;
+
+SELECT pg_logical_emit_delete('dummy', row(12, 'twelve', 0.12)::dummy) <> '0/0'::pg_lsn;
+
+SELECT pg_logical_emit_update('dummy', row(15, 'fifteen', 0.15)::dummy,
+       row(16, 'sixteen', 0.16)::dummy) <> '0/0'::pg_lsn;
+
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'force-binary', '0', 'skip-empty-xacts', '1', 'include-xids', '0');
+
+ALTER TABLE dummy REPLICA IDENTITY NOTHING;
+
+SELECT pg_logical_emit_delete('dummy', row(12, 'twelve', 0.12)::dummy) <> '0/0'::pg_lsn;
+
+SELECT pg_logical_emit_update('dummy', row(15, 'fifteen', 0.15)::dummy,
+       row(16, 'sixteen', 0.16)::dummy) <> '0/0'::pg_lsn;
+
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'force-binary', '0', 'skip-empty-xacts', '1', 'include-xids', '0');
+
+SELECT pg_logical_emit_truncate(ARRAY['dummy'], true, false) <> '0/0'::pg_lsn;
+
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'force-binary', '0', 'skip-empty-xacts', '1', 'include-xids', '0');
+
+CREATE UNLOGGED TABLE dummy_u(i int, t text, n numeric, primary key (t));
+-- return invalid
+SELECT pg_logical_emit_insert('dummy_u', row(7, 'seven', 0.7)::dummy_u);
+-- return invalid
+SELECT pg_logical_emit_update('dummy_u', row(7, 'seven', 0.7)::dummy_u,
+       row(11, 'eleven', 0.11)::dummy_u);
+-- return invalid
+SELECT pg_logical_emit_update('dummy_u', row(NULL, 'seven', 0.7)::dummy_u,
+       row(11, 'eleven', 0.11)::dummy_u);
+-- error
+SELECT pg_logical_emit_update('dummy_u', row(7, NULL, 0.7)::dummy_u,
+       row(11, 'eleven', 0.11)::dummy_u);
+-- return invalid
+SELECT pg_logical_emit_delete('dummy_u', row(7, 'seven', 0.7)::dummy_u);
+-- error
+SELECT pg_logical_emit_delete('dummy_u', row(7, NULL, 0.7)::dummy_u);
+
+-- return invalid
+SELECT pg_logical_emit_truncate(ARRAY['dummy_u'], false, false);
+
+SELECT pg_logical_emit_truncate(ARRAY['dummy','dummy_u'], false, true) <> '0/0'::pg_lsn;
+
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'force-binary', '0', 'skip-empty-xacts', '1', 'include-xids', '0');
+
+DROP TABLE dummy;
+
+DROP TABLE dummy_u;
+
 SELECT 'cleanup' FROM pg_drop_replication_slot('regression_slot');
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 74ad445e59b..9d70598dbde 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -108,8 +108,6 @@ static bool ConditionalMultiXactIdWait(MultiXactId multi, MultiXactStatus status
 static void index_delete_sort(TM_IndexDeleteOp *delstate);
 static int	bottomup_sort_and_shrink(TM_IndexDeleteOp *delstate);
 static XLogRecPtr log_heap_new_cid(Relation relation, HeapTuple tup);
-static HeapTuple ExtractReplicaIdentity(Relation rel, HeapTuple tup, bool key_required,
-										bool *copy);
 
 
 /*
@@ -8372,7 +8370,7 @@ log_heap_new_cid(Relation relation, HeapTuple tup)
  * *copy is set to true if the returned tuple is a modified copy rather than
  * the same tuple that was passed in.
  */
-static HeapTuple
+HeapTuple
 ExtractReplicaIdentity(Relation relation, HeapTuple tp, bool key_required,
 					   bool *copy)
 {
diff --git a/src/backend/access/rmgrdesc/Makefile b/src/backend/access/rmgrdesc/Makefile
index f88d72fd862..ed6dff179be 100644
--- a/src/backend/access/rmgrdesc/Makefile
+++ b/src/backend/access/rmgrdesc/Makefile
@@ -18,7 +18,7 @@ OBJS = \
 	gistdesc.o \
 	hashdesc.o \
 	heapdesc.o \
-	logicalmsgdesc.o \
+	logicaldesc.o \
 	mxactdesc.o \
 	nbtdesc.o \
 	relmapdesc.o \
diff --git a/src/backend/access/rmgrdesc/logicalmsgdesc.c b/src/backend/access/rmgrdesc/logicaldesc.c
similarity index 59%
rename from src/backend/access/rmgrdesc/logicalmsgdesc.c
rename to src/backend/access/rmgrdesc/logicaldesc.c
index 099e11a84e7..c2b0434a606 100644
--- a/src/backend/access/rmgrdesc/logicalmsgdesc.c
+++ b/src/backend/access/rmgrdesc/logicaldesc.c
@@ -1,22 +1,22 @@
 /*-------------------------------------------------------------------------
  *
- * logicalmsgdesc.c
- *	  rmgr descriptor routines for replication/logical/message.c
+ * logicaldesc.c
+ *	  rmgr descriptor routines for replication/logical/logical_xlog.c
  *
  * Portions Copyright (c) 2015-2022, PostgreSQL Global Development Group
  *
  *
  * IDENTIFICATION
- *	  src/backend/access/rmgrdesc/logicalmsgdesc.c
+ *	  src/backend/access/rmgrdesc/logicaldesc.c
  *
  *-------------------------------------------------------------------------
  */
 #include "postgres.h"
 
-#include "replication/message.h"
+#include "replication/logical_xlog.h"
 
 void
-logicalmsg_desc(StringInfo buf, XLogReaderState *record)
+logical_desc(StringInfo buf, XLogReaderState *record)
 {
 	char	   *rec = XLogRecGetData(record);
 	uint8		info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
@@ -40,13 +40,42 @@ logicalmsg_desc(StringInfo buf, XLogReaderState *record)
 			sep = " ";
 		}
 	}
+	else if (info == XLOG_LOGICAL_INSERT)
+	{
+
+	}
+	else if (info == XLOG_LOGICAL_UPDATE)
+	{
+
+	}
+	else if (info == XLOG_LOGICAL_DELETE)
+	{
+
+	}
+	else if (info == XLOG_LOGICAL_TRUNCATE)
+	{
+
+	}
 }
 
 const char *
-logicalmsg_identify(uint8 info)
+logical_identify(uint8 info)
 {
-	if ((info & ~XLR_INFO_MASK) == XLOG_LOGICAL_MESSAGE)
-		return "MESSAGE";
+	switch (info)
+	{
+		case XLOG_LOGICAL_MESSAGE:
+			return "MESSAGE";
+		case XLOG_LOGICAL_INSERT:
+			return "INSERT";
+		case XLOG_LOGICAL_UPDATE:
+			return "UPDATE";
+		case XLOG_LOGICAL_DELETE:
+			return "DELETE";
+		case XLOG_LOGICAL_TRUNCATE:
+			return "TRUNCATE";
+		default:
+			return NULL;
+	}
 
 	return NULL;
 }
diff --git a/src/backend/access/transam/rmgr.c b/src/backend/access/transam/rmgr.c
index f8847d5aebf..ca0bd614fcf 100644
--- a/src/backend/access/transam/rmgr.c
+++ b/src/backend/access/transam/rmgr.c
@@ -25,7 +25,7 @@
 #include "commands/sequence.h"
 #include "commands/tablespace.h"
 #include "replication/decode.h"
-#include "replication/message.h"
+#include "replication/logical_xlog.h"
 #include "replication/origin.h"
 #include "storage/standby.h"
 #include "utils/relmapper.h"
diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile
index c4e2fdeb719..9fa281a1dca 100644
--- a/src/backend/replication/logical/Makefile
+++ b/src/backend/replication/logical/Makefile
@@ -19,7 +19,7 @@ OBJS = \
 	launcher.o \
 	logical.o \
 	logicalfuncs.o \
-	message.o \
+	logical_xlog.o \
 	origin.o \
 	proto.o \
 	relation.o \
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index 77bc7aea7a0..81feca97eb0 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -35,23 +35,29 @@
 #include "access/xlogrecord.h"
 #include "access/xlogutils.h"
 #include "catalog/pg_control.h"
+#include "commands/sequence.h"
 #include "replication/decode.h"
 #include "replication/logical.h"
-#include "replication/message.h"
+#include "replication/logical_xlog.h"
 #include "replication/origin.h"
 #include "replication/reorderbuffer.h"
 #include "replication/snapbuild.h"
 #include "storage/standby.h"
-#include "commands/sequence.h"
 
 /* individual record(group)'s handlers */
-static void DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
-static void DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
-static void DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
-static void DecodeTruncate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
+static void DecodeHeapInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
+static void DecodeHeapUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
+static void DecodeHeapDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
+static void DecodeHeapTruncate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
 static void DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
 static void DecodeSpecConfirm(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
 
+static void DecodeLogicalMsg(LogicalDecodingContext *cxt, XLogRecordBuffer *buf);
+static void DecodeLogicalInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
+static void DecodeLogicalUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
+static void DecodeLogicalDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
+static void DecodeLogicalTruncate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
+
 static void DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 						 xl_xact_parsed_commit *parsed, TransactionId xid,
 						 bool two_phase);
@@ -457,7 +463,7 @@ heap_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 	{
 		case XLOG_HEAP_INSERT:
 			if (SnapBuildProcessChange(builder, xid, buf->origptr))
-				DecodeInsert(ctx, buf);
+				DecodeHeapInsert(ctx, buf);
 			break;
 
 			/*
@@ -468,17 +474,17 @@ heap_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 		case XLOG_HEAP_HOT_UPDATE:
 		case XLOG_HEAP_UPDATE:
 			if (SnapBuildProcessChange(builder, xid, buf->origptr))
-				DecodeUpdate(ctx, buf);
+				DecodeHeapUpdate(ctx, buf);
 			break;
 
 		case XLOG_HEAP_DELETE:
 			if (SnapBuildProcessChange(builder, xid, buf->origptr))
-				DecodeDelete(ctx, buf);
+				DecodeHeapDelete(ctx, buf);
 			break;
 
 		case XLOG_HEAP_TRUNCATE:
 			if (SnapBuildProcessChange(builder, xid, buf->origptr))
-				DecodeTruncate(ctx, buf);
+				DecodeHeapTruncate(ctx, buf);
 			break;
 
 		case XLOG_HEAP_INPLACE:
@@ -559,31 +565,71 @@ FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
  * Handle rmgr LOGICALMSG_ID records for DecodeRecordIntoReorderBuffer().
  */
 void
-logicalmsg_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
+logical_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 {
-	SnapBuild  *builder = ctx->snapshot_builder;
 	XLogReaderState *r = buf->record;
+	SnapBuild  *builder = ctx->snapshot_builder;
 	TransactionId xid = XLogRecGetXid(r);
 	uint8		info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
-	RepOriginId origin_id = XLogRecGetOrigin(r);
-	Snapshot	snapshot;
-	xl_logical_message *message;
-
-	if (info != XLOG_LOGICAL_MESSAGE)
-		elog(ERROR, "unexpected RM_LOGICALMSG_ID record type: %u", info);
 
-	ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(r), buf->origptr);
+	ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
 
 	/*
 	 * If we don't have snapshot or we are just fast-forwarding, there is no
-	 * point in decoding messages.
+	 * point in decoding data changes.
 	 */
 	if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT ||
 		ctx->fast_forward)
 		return;
 
+	switch (info)
+	{
+		case XLOG_LOGICAL_MESSAGE:
+			DecodeLogicalMsg(ctx, buf);
+			break;
+
+		case XLOG_LOGICAL_INSERT:
+			if (SnapBuildProcessChange(builder, xid, buf->origptr))
+				DecodeLogicalInsert(ctx, buf);
+			break;
+
+		case XLOG_LOGICAL_UPDATE:
+			if (SnapBuildProcessChange(builder, xid, buf->origptr))
+				DecodeLogicalUpdate(ctx, buf);
+			break;
+
+		case XLOG_LOGICAL_DELETE:
+			if (SnapBuildProcessChange(builder, xid, buf->origptr))
+				DecodeLogicalDelete(ctx, buf);
+			break;
+
+		case XLOG_LOGICAL_TRUNCATE:
+			if (SnapBuildProcessChange(builder, xid, buf->origptr))
+				DecodeLogicalTruncate(ctx, buf);
+			break;
+
+		default:
+			elog(ERROR, "unexpected RM_LOGICAL_ID record type: %u", info);
+			break;
+	}
+}
+
+static void
+DecodeLogicalMsg(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
+{
+	XLogReaderState *r = buf->record;
+	SnapBuild  *builder = ctx->snapshot_builder;
+	TransactionId xid = XLogRecGetXid(r);
+	RepOriginId origin_id = XLogRecGetOrigin(r);
+	Snapshot	snapshot;
+	xl_logical_message *message;
+
 	message = (xl_logical_message *) XLogRecGetData(r);
 
+	if (message->transactional &&
+		!SnapBuildProcessChange(builder, xid, buf->origptr))
+		return;
+
 	if (message->dbId != ctx->slot->data.database ||
 		FilterByOrigin(ctx, origin_id))
 		return;
@@ -603,6 +649,187 @@ logicalmsg_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 												 * prefix */
 							  message->message_size,
 							  message->message + message->prefix_size);
+
+}
+
+/*
+ * Parse XLOG_HEAP_INSERT (not MULTI_INSERT!) records into tuplebufs.
+ *
+ * Deletes can contain the new tuple.
+ */
+static void
+DecodeLogicalInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
+{
+	XLogReaderState *r = buf->record;
+	Size		datalen = XLogRecGetDataLen(r) - SizeOfLogicalInsert;
+	char	   *tupledata = XLogRecGetData(r) + SizeOfLogicalInsert;
+	Size		tuplelen = datalen - SizeOfHeapHeader;
+	xl_logical_insert *xlrec;
+	ReorderBufferChange *change;
+
+	xlrec = (xl_logical_insert *) XLogRecGetData(r);
+
+	/* only interested in our database */
+	if (xlrec->node.dbNode != ctx->slot->data.database)
+		return;
+
+	/* output plugin doesn't look for this origin, no need to queue */
+	if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
+		return;
+
+	change = ReorderBufferGetChange(ctx->reorder);
+	change->action = REORDER_BUFFER_CHANGE_INSERT;
+	change->origin_id = XLogRecGetOrigin(r);
+
+	memcpy(&change->data.tp.relnode, &xlrec->node, sizeof(RelFileNode));
+
+	change->data.tp.newtuple =
+		ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
+
+	DecodeXLogTuple(tupledata, datalen, change->data.tp.newtuple);
+
+	change->data.tp.clear_toast_afterwards = true;
+
+	ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr,
+							 change, false);
+}
+
+/*
+ * Parse XLOG_LOGICAL_UPDATE from wal into proper tuplebufs.
+ *
+ * Updates can possibly contain a new tuple and the old primary key.
+ */
+static void
+DecodeLogicalUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
+{
+	XLogReaderState *r = buf->record;
+	xl_logical_update *xlrec = (xl_logical_update *) XLogRecGetData(r);
+	char	   *new_tupledata = XLogRecGetData(r) + SizeOfLogicalUpdate;
+	Size		new_datalen = xlrec->new_datalen;
+	Size		new_tuplelen = new_datalen - SizeOfHeapHeader;
+	ReorderBufferChange *change;
+
+	/* only interested in our database */
+	if (xlrec->node.dbNode != ctx->slot->data.database)
+		return;
+
+	/* output plugin doesn't look for this origin, no need to queue */
+	if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
+		return;
+
+	change = ReorderBufferGetChange(ctx->reorder);
+	change->action = REORDER_BUFFER_CHANGE_UPDATE;
+	change->origin_id = XLogRecGetOrigin(r);
+
+	memcpy(&change->data.tp.relnode, &xlrec->node, sizeof(RelFileNode));
+
+	change->data.tp.newtuple =
+		ReorderBufferGetTupleBuf(ctx->reorder, new_tuplelen);
+
+	DecodeXLogTuple(new_tupledata, new_datalen, change->data.tp.newtuple);
+
+	if (xlrec->flags & XLL_UPDATE_CONTAINS_OLD)
+	{
+		char	*old_tupledata = new_tupledata + new_datalen;
+		Size	 old_datalen;
+		Size	 old_tuplelen;
+
+		/* remaining data is the old tuple */
+		old_datalen = XLogRecGetDataLen(r) - new_datalen - SizeOfLogicalUpdate;
+		old_tuplelen = old_datalen - SizeOfHeapHeader;
+
+		change->data.tp.oldtuple =
+			ReorderBufferGetTupleBuf(ctx->reorder, old_tuplelen);
+
+		DecodeXLogTuple(old_tupledata, old_datalen, change->data.tp.oldtuple);
+	}
+
+	change->data.tp.clear_toast_afterwards = true;
+
+	ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr,
+							 change, false);
+}
+
+/*
+ * Parse XLOG_LOGICAL_DELETE from wal into proper tuplebufs.
+ *
+ * Deletes can possibly contain the old primary key.
+ */
+static void
+DecodeLogicalDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
+{
+	XLogReaderState *r = buf->record;
+	xl_logical_delete *xlrec;
+	ReorderBufferChange *change;
+
+	xlrec = (xl_logical_delete *) XLogRecGetData(r);
+
+	/* only interested in our database */
+	if (xlrec->node.dbNode != ctx->slot->data.database)
+		return;
+
+	/* output plugin doesn't look for this origin, no need to queue */
+	if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
+		return;
+
+	change = ReorderBufferGetChange(ctx->reorder);
+	change->action = REORDER_BUFFER_CHANGE_DELETE;
+	change->origin_id = XLogRecGetOrigin(r);
+
+	memcpy(&change->data.tp.relnode, &xlrec->node, sizeof(RelFileNode));
+
+	if (xlrec->flags & XLL_DELETE_CONTAINS_OLD)
+	{
+		Size	 old_datalen   = XLogRecGetDataLen(r) - SizeOfLogicalDelete;
+		char	*old_tupledata = XLogRecGetData(r) + SizeOfLogicalDelete;
+		Size	 old_tuplelen  = old_datalen - SizeOfHeapHeader;
+
+		change->data.tp.oldtuple =
+			ReorderBufferGetTupleBuf(ctx->reorder, old_tuplelen);
+
+		DecodeXLogTuple(old_tupledata, old_datalen, change->data.tp.oldtuple);
+	}
+
+	change->data.tp.clear_toast_afterwards = true;
+
+	ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr,
+							 change, false);
+}
+
+/*
+ * Parse XLOG_LOGICAL_TRUNCATE from wal
+ */
+static void
+DecodeLogicalTruncate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
+{
+	XLogReaderState *r = buf->record;
+	xl_logical_truncate *xlrec;
+	ReorderBufferChange *change;
+
+	xlrec = (xl_logical_truncate *) XLogRecGetData(r);
+
+	/* only interested in our database */
+	if (xlrec->dbId != ctx->slot->data.database)
+		return;
+
+	/* output plugin doesn't look for this origin, no need to queue */
+	if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
+		return;
+
+	change = ReorderBufferGetChange(ctx->reorder);
+	change->action = REORDER_BUFFER_CHANGE_TRUNCATE;
+	change->origin_id = XLogRecGetOrigin(r);
+	if (xlrec->flags & XLL_TRUNCATE_CASCADE)
+		change->data.truncate.cascade = true;
+	if (xlrec->flags & XLL_TRUNCATE_RESTART_SEQS)
+		change->data.truncate.restart_seqs = true;
+	change->data.truncate.nrelids = xlrec->nrelids;
+	change->data.truncate.relids = ReorderBufferGetRelids(ctx->reorder,
+														  xlrec->nrelids);
+	memcpy(change->data.truncate.relids, xlrec->relids,
+		   xlrec->nrelids * sizeof(Oid));
+	ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r),
+							 buf->origptr, change, false);
 }
 
 /*
@@ -839,7 +1066,7 @@ DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
  * Deletes can contain the new tuple.
  */
 static void
-DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
+DecodeHeapInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 {
 	Size		datalen;
 	char	   *tupledata;
@@ -898,7 +1125,7 @@ DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
  * Updates can possibly contain a new tuple and the old primary key.
  */
 static void
-DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
+DecodeHeapUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 {
 	XLogReaderState *r = buf->record;
 	xl_heap_update *xlrec;
@@ -965,7 +1192,7 @@ DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
  * Deletes can possibly contain the old primary key.
  */
 static void
-DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
+DecodeHeapDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 {
 	XLogReaderState *r = buf->record;
 	xl_heap_delete *xlrec;
@@ -1019,7 +1246,7 @@ DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
  * Parse XLOG_HEAP_TRUNCATE from wal
  */
 static void
-DecodeTruncate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
+DecodeHeapTruncate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 {
 	XLogReaderState *r = buf->record;
 	xl_heap_truncate *xlrec;
diff --git a/src/backend/replication/logical/logical_xlog.c b/src/backend/replication/logical/logical_xlog.c
new file mode 100644
index 00000000000..30ae3e73be6
--- /dev/null
+++ b/src/backend/replication/logical/logical_xlog.c
@@ -0,0 +1,399 @@
+/*-------------------------------------------------------------------------
+ *
+ * logical_xlog.c
+ *	  Logical xlog records.
+ *
+ * Copyright (c) 2013-2022, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *	  src/backend/replication/logical/logical_xlog.c
+ *
+ * Logical Messages
+ *
+ * Generic logical messages allow XLOG logging of arbitrary binary blobs that
+ * get passed to the logical decoding plugin. In normal XLOG processing they
+ * are same as NOOP.
+ *
+ * These messages can be either transactional or non-transactional.
+ * Transactional messages are part of current transaction and will be sent to
+ * decoding plugin using in a same way as DML operations.
+ * Non-transactional messages are sent to the plugin at the time when the
+ * logical decoding reads them from XLOG. This also means that transactional
+ * messages won't be delivered if the transaction was rolled back but the
+ * non-transactional one will always be delivered.
+ *
+ * Every message carries prefix to avoid conflicts between different decoding
+ * plugins. The plugin authors must take extra care to use unique prefix,
+ * good options seems to be for example to use the name of the extension.
+ *
+ * Logical Insert/Update/Delete/Truncate
+ *
+ * These records are intended to be used by non-heap table access methods that
+ * wish to support logical decoding and replication. They are treated
+ * similarly to the analogous heap records, but are not tied to physical pages
+ * or other details of the heap. These records are not processed during redo,
+ * so do not contribute to durability or physical replication; use generic WAL
+ * records for that. Note that using both logical WAL records and generic WAL
+ * records is redundant compared with the heap.
+ *
+ * ---------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "access/heapam.h"
+#include "access/heapam_xlog.h"
+#include "access/xact.h"
+#include "access/xloginsert.h"
+#include "catalog/catalog.h"
+#include "miscadmin.h"
+#include "nodes/execnodes.h"
+#include "replication/logical.h"
+#include "replication/logical_xlog.h"
+#include "utils/memutils.h"
+#include "utils/relcache.h"
+
+static void CheckReplicaIdentity(Relation relation, TupleTableSlot *slot);
+
+/*
+ * Write logical decoding message into XLog.
+ */
+XLogRecPtr
+LogLogicalMessage(const char *prefix, const char *message, size_t size,
+				  bool transactional)
+{
+	xl_logical_message xlrec;
+
+	/*
+	 * Force xid to be allocated if we're emitting a transactional message.
+	 */
+	if (transactional)
+	{
+		Assert(IsTransactionState());
+		GetCurrentTransactionId();
+	}
+
+	xlrec.dbId = MyDatabaseId;
+	xlrec.transactional = transactional;
+	/* trailing zero is critical; see logicalmsg_desc */
+	xlrec.prefix_size = strlen(prefix) + 1;
+	xlrec.message_size = size;
+
+	XLogBeginInsert();
+	XLogRegisterData((char *) &xlrec, SizeOfLogicalMessage);
+	XLogRegisterData(unconstify(char *, prefix), xlrec.prefix_size);
+	XLogRegisterData(unconstify(char *, message), size);
+
+	/* allow origin filtering */
+	XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);
+
+	return XLogInsert(RM_LOGICAL_ID, XLOG_LOGICAL_MESSAGE);
+}
+
+/*
+ * Write logical insert into log.
+ */
+XLogRecPtr
+LogLogicalInsert(Relation relation, TupleTableSlot *new_slot)
+{
+	bool		free_new_tuple;
+	HeapTuple	new_tuple;
+	xl_logical_insert xlrec;
+	xl_heap_header xlhdr;
+	XLogRecPtr recptr;
+
+	if (!equalTupleDescs(relation->rd_att, new_slot->tts_tupleDescriptor))
+		ereport(ERROR, (errmsg("record type must match relation type")));
+
+	if (!RelationIsLogicallyLogged(relation))
+		return InvalidXLogRecPtr;
+
+	new_tuple = ExecFetchSlotHeapTuple(new_slot, true, &free_new_tuple);
+
+	/* force xid to be allocated */
+	Assert(IsTransactionState());
+	GetCurrentTransactionId();
+
+	new_tuple->t_tableOid = new_slot->tts_tableOid;
+	ItemPointerCopy(&new_tuple->t_self, &new_slot->tts_tid);
+
+	xlrec.node = relation->rd_node;
+
+	XLogBeginInsert();
+	XLogRegisterData((char *) &xlrec, SizeOfLogicalInsert);
+
+	xlhdr.t_infomask2 = new_tuple->t_data->t_infomask2;
+	xlhdr.t_infomask = new_tuple->t_data->t_infomask;
+	xlhdr.t_hoff = new_tuple->t_data->t_hoff;
+
+	XLogRegisterData((char *) &xlhdr, SizeOfHeapHeader);
+	XLogRegisterData((char *) new_tuple->t_data + SizeofHeapTupleHeader,
+					 new_tuple->t_len - SizeofHeapTupleHeader);
+
+	/* allow origin filtering */
+	XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);
+
+	recptr = XLogInsert(RM_LOGICAL_ID, XLOG_LOGICAL_INSERT);
+
+	if (free_new_tuple)
+		pfree(new_tuple);
+
+	return recptr;
+}
+
+/*
+ * Write logical update into log.
+ */
+XLogRecPtr
+LogLogicalUpdate(Relation relation, TupleTableSlot *old_slot,
+				 TupleTableSlot *new_slot)
+{
+	HeapTuple	old_whole_tuple;
+	HeapTuple	old_id_tuple;
+	HeapTuple	new_tuple;
+	bool		free_old_whole_tuple;
+	bool		free_old_id_tuple;
+	bool		free_new_tuple;
+	xl_heap_header new_xlhdr;
+	xl_logical_update xlrec;
+	XLogRecPtr recptr;
+
+	if (!equalTupleDescs(relation->rd_att, old_slot->tts_tupleDescriptor) ||
+		!equalTupleDescs(relation->rd_att, new_slot->tts_tupleDescriptor))
+		ereport(ERROR, (errmsg("record types must match relation type")));
+
+	CheckReplicaIdentity(relation, old_slot);
+
+	if (!RelationIsLogicallyLogged(relation))
+		return InvalidXLogRecPtr;
+
+	/* force xid to be allocated */
+	Assert(IsTransactionState());
+	GetCurrentTransactionId();
+
+	old_whole_tuple = ExecFetchSlotHeapTuple(old_slot, true,
+											 &free_old_whole_tuple);
+	new_tuple = ExecFetchSlotHeapTuple(new_slot, true, &free_new_tuple);
+
+	xlrec.node = relation->rd_node;
+	xlrec.new_datalen = new_tuple->t_len - SizeofHeapTupleHeader +
+		SizeOfHeapHeader;
+	xlrec.flags = 0;
+
+	old_id_tuple = ExtractReplicaIdentity(relation, old_whole_tuple, true,
+										  &free_old_id_tuple);
+
+	if (old_id_tuple != NULL)
+	{
+		xlrec.flags |= XLL_UPDATE_CONTAINS_OLD;
+		old_id_tuple->t_tableOid = old_slot->tts_tableOid;
+		ItemPointerCopy(&old_id_tuple->t_self, &old_slot->tts_tid);
+	}
+
+	XLogBeginInsert();
+	XLogRegisterData((char *) &xlrec, SizeOfLogicalUpdate);
+
+	new_tuple->t_tableOid = new_slot->tts_tableOid;
+	ItemPointerCopy(&new_tuple->t_self, &new_slot->tts_tid);
+
+	new_xlhdr.t_infomask2 = new_tuple->t_data->t_infomask2;
+	new_xlhdr.t_infomask = new_tuple->t_data->t_infomask;
+	new_xlhdr.t_hoff = new_tuple->t_data->t_hoff;
+
+	/* write new tuple first, then old */
+	XLogRegisterData((char *) &new_xlhdr, SizeOfHeapHeader);
+	XLogRegisterData((char *) new_tuple->t_data + SizeofHeapTupleHeader,
+					 new_tuple->t_len - SizeofHeapTupleHeader);
+
+	if (old_id_tuple != NULL)
+	{
+		xl_heap_header old_xlhdr;
+
+		old_xlhdr.t_infomask2 = old_id_tuple->t_data->t_infomask2;
+		old_xlhdr.t_infomask = old_id_tuple->t_data->t_infomask;
+		old_xlhdr.t_hoff = old_id_tuple->t_data->t_hoff;
+
+		XLogRegisterData((char *) &old_xlhdr, SizeOfHeapHeader);
+		XLogRegisterData((char *) old_id_tuple->t_data + SizeofHeapTupleHeader,
+					 old_id_tuple->t_len - SizeofHeapTupleHeader);
+	}
+
+	/* allow origin filtering */
+	XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);
+
+	recptr = XLogInsert(RM_LOGICAL_ID, XLOG_LOGICAL_UPDATE);
+
+	if (free_old_whole_tuple)
+		pfree(old_whole_tuple);
+	if (free_old_id_tuple)
+		pfree(old_id_tuple);
+	if (free_new_tuple)
+		pfree(new_tuple);
+
+	return recptr;
+}
+
+/*
+ * Write logical update into log.
+ */
+XLogRecPtr
+LogLogicalDelete(Relation relation, TupleTableSlot *old_slot)
+{
+	HeapTuple		 old_whole_tuple;
+	HeapTuple		 old_id_tuple;
+	bool			 free_old_whole_tuple;
+	bool			 free_old_id_tuple;
+	xl_logical_delete xlrec;
+	XLogRecPtr		 recptr;
+
+	if (!equalTupleDescs(relation->rd_att, old_slot->tts_tupleDescriptor))
+		ereport(ERROR, (errmsg("record type must match relation type")));
+
+	CheckReplicaIdentity(relation, old_slot);
+
+	if (!RelationIsLogicallyLogged(relation))
+		return InvalidXLogRecPtr;
+
+	/* force xid to be allocated */
+	Assert(IsTransactionState());
+	GetCurrentTransactionId();
+
+	old_whole_tuple = ExecFetchSlotHeapTuple(old_slot, true,
+											 &free_old_whole_tuple);
+
+	xlrec.node = relation->rd_node;
+	xlrec.flags = 0;
+
+	old_id_tuple = ExtractReplicaIdentity(relation, old_whole_tuple, true,
+										  &free_old_id_tuple);
+
+	if (old_id_tuple != NULL)
+	{
+		xlrec.flags |= XLL_UPDATE_CONTAINS_OLD;
+		old_id_tuple->t_tableOid = old_slot->tts_tableOid;
+		ItemPointerCopy(&old_id_tuple->t_self, &old_slot->tts_tid);
+	}
+
+	XLogBeginInsert();
+	XLogRegisterData((char *) &xlrec, SizeOfLogicalDelete);
+
+	if (old_id_tuple != NULL)
+	{
+		xl_heap_header old_xlhdr;
+
+		old_xlhdr.t_infomask2 = old_id_tuple->t_data->t_infomask2;
+		old_xlhdr.t_infomask = old_id_tuple->t_data->t_infomask;
+		old_xlhdr.t_hoff = old_id_tuple->t_data->t_hoff;
+
+		XLogRegisterData((char *) &old_xlhdr, SizeOfHeapHeader);
+		XLogRegisterData((char *) old_id_tuple->t_data + SizeofHeapTupleHeader,
+					 old_id_tuple->t_len - SizeofHeapTupleHeader);
+	}
+
+	/* allow origin filtering */
+	XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);
+
+	recptr = XLogInsert(RM_LOGICAL_ID, XLOG_LOGICAL_DELETE);
+
+	if (free_old_whole_tuple)
+		pfree(old_whole_tuple);
+	if (free_old_id_tuple)
+		pfree(old_id_tuple);
+
+	return recptr;
+}
+
+/*
+ * Write logical truncate into log.
+ */
+XLogRecPtr
+LogLogicalTruncate(List *relids, bool cascade, bool restart_seqs)
+{
+	xl_logical_truncate	 xlrec;
+	XLogRecPtr			 recptr;
+	Oid					*logrelids;
+	ListCell			*lc;
+	int					 nrelids = 0;
+
+	/* force xid to be allocated */
+	Assert(IsTransactionState());
+	GetCurrentTransactionId();
+
+	xlrec.dbId = MyDatabaseId;
+	xlrec.nrelids = list_length(relids);
+	xlrec.flags = 0;
+	if (cascade)
+		xlrec.flags |= XLL_TRUNCATE_CASCADE;
+	if (restart_seqs)
+		xlrec.flags |= XLL_TRUNCATE_RESTART_SEQS;
+
+	logrelids = palloc(list_length(relids) * sizeof(Oid));
+	foreach(lc, relids)
+	{
+		Oid			relid = lfirst_oid(lc);
+		Relation	rel	  = relation_open(relid, AccessShareLock);
+
+		if (RelationIsLogicallyLogged(rel))
+		{
+			logrelids[nrelids++] = lfirst_oid(lc);
+		}
+
+		relation_close(rel, NoLock);
+	}
+
+	if (nrelids == 0)
+		return InvalidXLogRecPtr;
+
+	XLogBeginInsert();
+	XLogRegisterData((char *) &xlrec, SizeOfLogicalTruncate);
+	XLogRegisterData((char *) logrelids, nrelids * sizeof(Oid));
+
+	/* allow origin filtering */
+	XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);
+
+	recptr = XLogInsert(RM_LOGICAL_ID, XLOG_LOGICAL_TRUNCATE);
+
+	return recptr;
+}
+
+/*
+ * Redo is basically just noop for logical decoding messages.
+ */
+void
+logical_redo(XLogReaderState *record)
+{
+	uint8		info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
+
+	switch (info)
+	{
+		case XLOG_LOGICAL_MESSAGE:
+		case XLOG_LOGICAL_INSERT:
+		case XLOG_LOGICAL_UPDATE:
+		case XLOG_LOGICAL_DELETE:
+		case XLOG_LOGICAL_TRUNCATE:
+			break;
+		default:
+			elog(PANIC, "logical_redo: unknown op code %u", info);
+	}
+
+	/* This is only interesting for logical decoding, see decode.c. */
+}
+
+/*
+ * Check that replica identity columns are non-NULL.
+ */
+static void
+CheckReplicaIdentity(Relation relation, TupleTableSlot *slot)
+{
+	/* check for NULL attributes in the replica identity */
+	Bitmapset *id_attrs = RelationGetIndexAttrBitmap(
+		relation, INDEX_ATTR_BITMAP_IDENTITY_KEY);
+	int id_attr = (-1) * FirstLowInvalidHeapAttributeNumber;
+
+	while ((id_attr = bms_next_member(id_attrs, id_attr)) >= 0)
+	{
+		AttrNumber attno = id_attr + FirstLowInvalidHeapAttributeNumber;
+		if (slot_attisnull(slot, attno))
+			ereport(ERROR, (errmsg("replica identity column is NULL")));
+	}
+}
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index 6058d36e0d5..2776861ab8d 100644
--- a/src/backend/replication/logical/logicalfuncs.c
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -17,6 +17,7 @@
 
 #include <unistd.h>
 
+#include "access/relation.h"
 #include "access/xact.h"
 #include "access/xlog_internal.h"
 #include "access/xlogrecovery.h"
@@ -29,7 +30,7 @@
 #include "nodes/makefuncs.h"
 #include "replication/decode.h"
 #include "replication/logical.h"
-#include "replication/message.h"
+#include "replication/logical_xlog.h"
 #include "storage/fd.h"
 #include "utils/array.h"
 #include "utils/builtins.h"
@@ -389,3 +390,165 @@ pg_logical_emit_message_text(PG_FUNCTION_ARGS)
 	/* bytea and text are compatible */
 	return pg_logical_emit_message_bytea(fcinfo);
 }
+
+/*
+ * SQL function for writing logical insert into WAL.
+ */
+Datum
+pg_logical_emit_insert(PG_FUNCTION_ARGS)
+{
+	Oid				 relid = PG_GETARG_OID(0);
+	HeapTupleHeader	 htup  = PG_GETARG_HEAPTUPLEHEADER(1);
+	Relation		 rel   = relation_open(relid, AccessShareLock);
+	HeapTupleData	 tuple;
+	TupleTableSlot	*slot;
+	Oid				 rel_type;
+	Oid				 tup_type;
+	XLogRecPtr		 lsn;
+
+	/* check that tuple matches the type of the relation */
+	rel_type = get_rel_type_id(relid);
+	tup_type = HeapTupleHeaderGetTypeId(htup);
+	if (rel_type != tup_type)
+		ereport(ERROR, (errmsg("record type must match table type")));
+
+	tuple.t_len = HeapTupleHeaderGetDatumLength(htup);
+	ItemPointerSetInvalid(&(tuple.t_self));
+	tuple.t_tableOid = relid;
+	tuple.t_data = htup;
+
+	slot = MakeTupleTableSlot(rel->rd_att, &TTSOpsHeapTuple);
+	ExecClearTuple(slot);
+	ExecStoreHeapTuple(&tuple, slot, false);
+
+	lsn = LogLogicalInsert(rel, slot);
+
+	ExecDropSingleTupleTableSlot(slot);
+
+	relation_close(rel, NoLock);
+
+	PG_RETURN_LSN(lsn);
+}
+
+/*
+ * SQL function for writing logical update into WAL.
+ */
+Datum
+pg_logical_emit_update(PG_FUNCTION_ARGS)
+{
+	Oid				 relid = PG_GETARG_OID(0);
+	HeapTupleHeader	 old_htup  = PG_GETARG_HEAPTUPLEHEADER(1);
+	HeapTupleHeader	 new_htup  = PG_GETARG_HEAPTUPLEHEADER(2);
+	Relation		 rel   = relation_open(relid, AccessShareLock);
+	HeapTupleData	 old_tuple;
+	HeapTupleData	 new_tuple;
+	TupleTableSlot	*old_slot;
+	TupleTableSlot	*new_slot;
+	Oid				 rel_type;
+	Oid				 old_tup_type;
+	Oid				 new_tup_type;
+	XLogRecPtr		 lsn;
+
+	/* check that tuple matches the type of the relation */
+	rel_type = get_rel_type_id(relid);
+	old_tup_type = HeapTupleHeaderGetTypeId(old_htup);
+	new_tup_type = HeapTupleHeaderGetTypeId(new_htup);
+
+	if (rel_type != old_tup_type || rel_type != new_tup_type)
+		ereport(ERROR, (errmsg("record type must match table type")));
+
+	old_tuple.t_len = HeapTupleHeaderGetDatumLength(old_htup);
+	ItemPointerSetInvalid(&(old_tuple.t_self));
+	old_tuple.t_tableOid = relid;
+	old_tuple.t_data = old_htup;
+
+	new_tuple.t_len = HeapTupleHeaderGetDatumLength(new_htup);
+	ItemPointerSetInvalid(&(new_tuple.t_self));
+	new_tuple.t_tableOid = relid;
+	new_tuple.t_data = new_htup;
+
+	old_slot = MakeTupleTableSlot(rel->rd_att, &TTSOpsHeapTuple);
+	new_slot = MakeTupleTableSlot(rel->rd_att, &TTSOpsHeapTuple);
+	ExecClearTuple(old_slot);
+	ExecClearTuple(new_slot);
+	ExecStoreHeapTuple(&old_tuple, old_slot, false);
+	ExecStoreHeapTuple(&new_tuple, new_slot, false);
+
+	lsn = LogLogicalUpdate(rel, old_slot, new_slot);
+
+	ExecDropSingleTupleTableSlot(old_slot);
+	ExecDropSingleTupleTableSlot(new_slot);
+
+	relation_close(rel, NoLock);
+
+	PG_RETURN_LSN(lsn);
+}
+
+/*
+ * SQL function for writing logical delete into WAL.
+ */
+Datum
+pg_logical_emit_delete(PG_FUNCTION_ARGS)
+{
+	Oid				 relid = PG_GETARG_OID(0);
+	HeapTupleHeader	 htup  = PG_GETARG_HEAPTUPLEHEADER(1);
+	Relation		 rel   = relation_open(relid, AccessShareLock);
+	HeapTupleData	 tuple;
+	TupleTableSlot	*slot;
+	Oid				 rel_type;
+	Oid				 tup_type;
+	XLogRecPtr		 lsn;
+
+	/* check that tuple matches the type of the relation */
+	rel_type = get_rel_type_id(relid);
+	tup_type = HeapTupleHeaderGetTypeId(htup);
+	if (rel_type != tup_type)
+		ereport(ERROR, (errmsg("record type must match table type")));
+
+	tuple.t_len = HeapTupleHeaderGetDatumLength(htup);
+	ItemPointerSetInvalid(&(tuple.t_self));
+	tuple.t_tableOid = relid;
+	tuple.t_data = htup;
+
+	slot = MakeTupleTableSlot(rel->rd_att, &TTSOpsHeapTuple);
+	ExecClearTuple(slot);
+	ExecStoreHeapTuple(&tuple, slot, false);
+
+	lsn = LogLogicalDelete(rel, slot);
+
+	ExecDropSingleTupleTableSlot(slot);
+
+	relation_close(rel, NoLock);
+
+	PG_RETURN_LSN(lsn);
+}
+
+/*
+ * SQL function for writing logical truncate into WAL.
+ */
+Datum
+pg_logical_emit_truncate(PG_FUNCTION_ARGS)
+{
+	ArrayType	*arr		  = PG_GETARG_ARRAYTYPE_P(0);
+	bool		 cascade	  = PG_GETARG_BOOL(1);
+	bool		 restart_seqs = PG_GETARG_BOOL(2);
+	Datum		*values;
+	bool		*nulls;
+	int			 nrelids;
+	List		*relids		  = NIL;
+	XLogRecPtr	 lsn;
+
+	deconstruct_array(arr, REGCLASSOID, sizeof(Oid), true, TYPALIGN_INT,
+					  &values, &nulls, &nrelids);
+
+	for (int i = 0; i < nrelids; i++)
+	{
+		if (nulls[i])
+			ereport(ERROR, (errmsg("unexpected NULL element")));
+		relids = lappend_oid(relids, DatumGetObjectId(values[i]));
+	}
+
+	lsn = LogLogicalTruncate(relids, cascade, restart_seqs);
+
+	PG_RETURN_LSN(lsn);
+}
diff --git a/src/backend/replication/logical/message.c b/src/backend/replication/logical/message.c
deleted file mode 100644
index 1c34912610e..00000000000
--- a/src/backend/replication/logical/message.c
+++ /dev/null
@@ -1,89 +0,0 @@
-/*-------------------------------------------------------------------------
- *
- * message.c
- *	  Generic logical messages.
- *
- * Copyright (c) 2013-2022, PostgreSQL Global Development Group
- *
- * IDENTIFICATION
- *	  src/backend/replication/logical/message.c
- *
- * NOTES
- *
- * Generic logical messages allow XLOG logging of arbitrary binary blobs that
- * get passed to the logical decoding plugin. In normal XLOG processing they
- * are same as NOOP.
- *
- * These messages can be either transactional or non-transactional.
- * Transactional messages are part of current transaction and will be sent to
- * decoding plugin using in a same way as DML operations.
- * Non-transactional messages are sent to the plugin at the time when the
- * logical decoding reads them from XLOG. This also means that transactional
- * messages won't be delivered if the transaction was rolled back but the
- * non-transactional one will always be delivered.
- *
- * Every message carries prefix to avoid conflicts between different decoding
- * plugins. The plugin authors must take extra care to use unique prefix,
- * good options seems to be for example to use the name of the extension.
- *
- * ---------------------------------------------------------------------------
- */
-
-#include "postgres.h"
-
-#include "access/xact.h"
-#include "access/xloginsert.h"
-#include "miscadmin.h"
-#include "nodes/execnodes.h"
-#include "replication/logical.h"
-#include "replication/message.h"
-#include "utils/memutils.h"
-
-/*
- * Write logical decoding message into XLog.
- */
-XLogRecPtr
-LogLogicalMessage(const char *prefix, const char *message, size_t size,
-				  bool transactional)
-{
-	xl_logical_message xlrec;
-
-	/*
-	 * Force xid to be allocated if we're emitting a transactional message.
-	 */
-	if (transactional)
-	{
-		Assert(IsTransactionState());
-		GetCurrentTransactionId();
-	}
-
-	xlrec.dbId = MyDatabaseId;
-	xlrec.transactional = transactional;
-	/* trailing zero is critical; see logicalmsg_desc */
-	xlrec.prefix_size = strlen(prefix) + 1;
-	xlrec.message_size = size;
-
-	XLogBeginInsert();
-	XLogRegisterData((char *) &xlrec, SizeOfLogicalMessage);
-	XLogRegisterData(unconstify(char *, prefix), xlrec.prefix_size);
-	XLogRegisterData(unconstify(char *, message), size);
-
-	/* allow origin filtering */
-	XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);
-
-	return XLogInsert(RM_LOGICALMSG_ID, XLOG_LOGICAL_MESSAGE);
-}
-
-/*
- * Redo is basically just noop for logical decoding messages.
- */
-void
-logicalmsg_redo(XLogReaderState *record)
-{
-	uint8		info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
-
-	if (info != XLOG_LOGICAL_MESSAGE)
-		elog(PANIC, "logicalmsg_redo: unknown op code %u", info);
-
-	/* This is only interesting for logical decoding, see decode.c. */
-}
diff --git a/src/bin/pg_waldump/.gitignore b/src/bin/pg_waldump/.gitignore
index 3be00a8b61f..567655fa626 100644
--- a/src/bin/pg_waldump/.gitignore
+++ b/src/bin/pg_waldump/.gitignore
@@ -10,7 +10,7 @@
 /gistdesc.c
 /hashdesc.c
 /heapdesc.c
-/logicalmsgdesc.c
+/logicaldesc.c
 /mxactdesc.c
 /nbtdesc.c
 /relmapdesc.c
diff --git a/src/bin/pg_waldump/rmgrdesc.c b/src/bin/pg_waldump/rmgrdesc.c
index 6a4ebd1310b..86804a243bc 100644
--- a/src/bin/pg_waldump/rmgrdesc.c
+++ b/src/bin/pg_waldump/rmgrdesc.c
@@ -26,7 +26,7 @@
 #include "commands/dbcommands_xlog.h"
 #include "commands/sequence.h"
 #include "commands/tablespace.h"
-#include "replication/message.h"
+#include "replication/logical_xlog.h"
 #include "replication/origin.h"
 #include "rmgrdesc.h"
 #include "storage/standbydefs.h"
diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h
index b46ab7d7390..77fa0337f6e 100644
--- a/src/include/access/heapam.h
+++ b/src/include/access/heapam.h
@@ -175,6 +175,8 @@ extern void simple_heap_insert(Relation relation, HeapTuple tup);
 extern void simple_heap_delete(Relation relation, ItemPointer tid);
 extern void simple_heap_update(Relation relation, ItemPointer otid,
 							   HeapTuple tup);
+extern HeapTuple ExtractReplicaIdentity(Relation rel, HeapTuple tup,
+										bool key_required, bool *copy);
 
 extern TransactionId heap_index_delete_tuples(Relation rel,
 											  TM_IndexDeleteOp *delstate);
diff --git a/src/include/access/heapam_xlog.h b/src/include/access/heapam_xlog.h
index 5c47fdcec80..5460f7836f9 100644
--- a/src/include/access/heapam_xlog.h
+++ b/src/include/access/heapam_xlog.h
@@ -124,6 +124,9 @@ typedef struct xl_heap_delete
  * For truncate we list all truncated relids in an array, followed by all
  * sequence relids that need to be restarted, if any.
  * All rels are always within the same database, so we just list dbid once.
+ *
+ * Note: identical to xl_logical_truncate, except that for
+ * xl_logical_truncate, no redo is performed.
  */
 typedef struct xl_heap_truncate
 {
diff --git a/src/include/access/rmgrlist.h b/src/include/access/rmgrlist.h
index cf8b6d48193..d43175088fb 100644
--- a/src/include/access/rmgrlist.h
+++ b/src/include/access/rmgrlist.h
@@ -46,4 +46,4 @@ PG_RMGR(RM_BRIN_ID, "BRIN", brin_redo, brin_desc, brin_identify, NULL, NULL, bri
 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)
+PG_RMGR(RM_LOGICAL_ID, "Logical", logical_redo, logical_desc, logical_identify, NULL, NULL, NULL, logical_decode)
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 01e1dd4d6d1..cd57c2b3aa2 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -10937,6 +10937,23 @@
   proname => 'pg_logical_emit_message', provolatile => 'v', proparallel => 'u',
   prorettype => 'pg_lsn', proargtypes => 'bool text bytea',
   prosrc => 'pg_logical_emit_message_bytea' },
+{ oid => '9297', descr => 'emit a logical insert',
+  proname => 'pg_logical_emit_insert', provolatile => 'v', proparallel => 'u',
+  prorettype => 'pg_lsn', proargtypes => 'regclass record',
+  prosrc => 'pg_logical_emit_insert' },
+{ oid => '9298', descr => 'emit a logical update',
+  proname => 'pg_logical_emit_update', provolatile => 'v', proparallel => 'u',
+  prorettype => 'pg_lsn', proargtypes => 'regclass record record',
+  prosrc => 'pg_logical_emit_update' },
+{ oid => '9299', descr => 'emit a logical delete',
+  proname => 'pg_logical_emit_delete', provolatile => 'v', proparallel => 'u',
+  prorettype => 'pg_lsn', proargtypes => 'regclass record',
+  prosrc => 'pg_logical_emit_delete' },
+{ oid => '9300', descr => 'emit a logical truncate',
+  proname => 'pg_logical_emit_truncate', provolatile => 'v', proparallel => 'u',
+  prorettype => 'pg_lsn', proargtypes => '_regclass bool bool',
+  prosrc => 'pg_logical_emit_truncate' },
+
 
 # event triggers
 { oid => '3566', descr => 'list objects dropped by the current command',
diff --git a/src/include/replication/decode.h b/src/include/replication/decode.h
index 8e07bb7409a..118590ad61e 100644
--- a/src/include/replication/decode.h
+++ b/src/include/replication/decode.h
@@ -26,7 +26,7 @@ extern void heap_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
 extern void heap2_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
 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 void logical_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
 extern void sequence_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
 
 extern void	LogicalDecodingProcessRecord(LogicalDecodingContext *ctx,
diff --git a/src/include/replication/logical_xlog.h b/src/include/replication/logical_xlog.h
new file mode 100644
index 00000000000..dce74f71111
--- /dev/null
+++ b/src/include/replication/logical_xlog.h
@@ -0,0 +1,124 @@
+/*-------------------------------------------------------------------------
+ * logical_xlog.h
+ *	   Exports from replication/logical/logical_xlog.c
+ *
+ * Copyright (c) 2013-2022, PostgreSQL Global Development Group
+ *
+ * src/include/replication/logical_xlog.h
+ *-------------------------------------------------------------------------
+ */
+#ifndef PG_LOGICAL_MESSAGE_H
+#define PG_LOGICAL_MESSAGE_H
+
+#include "access/xlog.h"
+#include "access/xlogdefs.h"
+#include "access/xlogreader.h"
+#include "storage/off.h"
+#include "utils/rel.h"
+
+/*
+ * xl_heap_update flag values, 8 bits are available.
+ */
+#define XLL_UPDATE_CONTAINS_OLD		(1<<0)
+
+/*
+ * xl_heap_delete flag values, 8 bits are available.
+ */
+#define XLL_DELETE_CONTAINS_OLD		(1<<0)
+
+/*
+ * Generic logical decoding message wal record.
+ */
+typedef struct xl_logical_message
+{
+	Oid			dbId;			/* database Oid emitted from */
+	bool		transactional;	/* is message transactional? */
+	Size		prefix_size;	/* length of prefix */
+	Size		message_size;	/* size of the message */
+	/* payload, including null-terminated prefix of length prefix_size */
+	char		message[FLEXIBLE_ARRAY_MEMBER];
+} xl_logical_message;
+
+#define SizeOfLogicalMessage	(offsetof(xl_logical_message, message))
+
+/* This is what we need to know about insert */
+typedef struct xl_logical_insert
+{
+	RelFileNode		node;
+
+	/* tuple data follows */
+} xl_logical_insert;
+
+#define SizeOfLogicalInsert	(offsetof(xl_logical_insert, node) + sizeof(RelFileNode))
+
+/*
+ * This is what we need to know about update.
+ */
+typedef struct xl_logical_update
+{
+	RelFileNode	node;
+	Size		new_datalen;
+	uint8		flags;
+
+	/* tuple data follows */
+} xl_logical_update;
+
+#define SizeOfLogicalUpdate	(offsetof(xl_logical_update, flags) + sizeof(uint8))
+
+/* This is what we need to know about delete */
+typedef struct xl_logical_delete
+{
+	RelFileNode	node;
+	uint8		flags;
+
+	/* tuple data follows */
+} xl_logical_delete;
+
+#define SizeOfLogicalDelete	(offsetof(xl_logical_delete, flags) + sizeof(uint8))
+
+/*
+ * For truncate we list all truncated relids in an array, followed by all
+ * sequence relids that need to be restarted, if any.
+ * All rels are always within the same database, so we just list dbid once.
+ *
+ * Note: identical to xl_logical_truncate, except that no redo is performed, only
+ * decoding.
+ */
+typedef struct xl_logical_truncate
+{
+	Oid			dbId;
+	uint32		nrelids;
+	uint8		flags;
+	Oid			relids[FLEXIBLE_ARRAY_MEMBER];
+} xl_logical_truncate;
+
+#define SizeOfLogicalTruncate	(offsetof(xl_logical_truncate, relids))
+
+struct TupleTableSlot;
+
+extern XLogRecPtr LogLogicalMessage(const char *prefix, const char *message,
+									size_t size, bool transactional);
+extern XLogRecPtr LogLogicalInsert(Relation relation, struct TupleTableSlot *slot);
+extern XLogRecPtr LogLogicalUpdate(Relation relation, struct TupleTableSlot *old_slot,
+								   struct TupleTableSlot *new_slot);
+extern XLogRecPtr LogLogicalDelete(Relation relation, struct TupleTableSlot *slot);
+extern XLogRecPtr LogLogicalTruncate(List *relids, bool cascade, bool restart_seqs);
+
+/* RMGR API*/
+#define XLOG_LOGICAL_MESSAGE	0x00
+#define XLOG_LOGICAL_INSERT		0x10
+#define XLOG_LOGICAL_UPDATE		0x20
+#define XLOG_LOGICAL_DELETE		0x30
+#define XLOG_LOGICAL_TRUNCATE	0x40
+
+/*
+ * xl_logical_truncate flag values, 8 bits are available.
+ */
+#define XLL_TRUNCATE_CASCADE					(1<<0)
+#define XLL_TRUNCATE_RESTART_SEQS				(1<<1)
+
+void		logical_redo(XLogReaderState *record);
+void		logical_desc(StringInfo buf, XLogReaderState *record);
+const char *logical_identify(uint8 info);
+
+#endif							/* PG_LOGICAL_MESSAGE_H */
diff --git a/src/include/replication/message.h b/src/include/replication/message.h
deleted file mode 100644
index 7d7785292f1..00000000000
--- a/src/include/replication/message.h
+++ /dev/null
@@ -1,41 +0,0 @@
-/*-------------------------------------------------------------------------
- * message.h
- *	   Exports from replication/logical/message.c
- *
- * Copyright (c) 2013-2022, PostgreSQL Global Development Group
- *
- * src/include/replication/message.h
- *-------------------------------------------------------------------------
- */
-#ifndef PG_LOGICAL_MESSAGE_H
-#define PG_LOGICAL_MESSAGE_H
-
-#include "access/xlog.h"
-#include "access/xlogdefs.h"
-#include "access/xlogreader.h"
-
-/*
- * Generic logical decoding message wal record.
- */
-typedef struct xl_logical_message
-{
-	Oid			dbId;			/* database Oid emitted from */
-	bool		transactional;	/* is message transactional? */
-	Size		prefix_size;	/* length of prefix */
-	Size		message_size;	/* size of the message */
-	/* payload, including null-terminated prefix of length prefix_size */
-	char		message[FLEXIBLE_ARRAY_MEMBER];
-} xl_logical_message;
-
-#define SizeOfLogicalMessage	(offsetof(xl_logical_message, message))
-
-extern XLogRecPtr LogLogicalMessage(const char *prefix, const char *message,
-									size_t size, bool transactional);
-
-/* RMGR API*/
-#define XLOG_LOGICAL_MESSAGE	0x00
-void		logicalmsg_redo(XLogReaderState *record);
-void		logicalmsg_desc(StringInfo buf, XLogReaderState *record);
-const char *logicalmsg_identify(uint8 info);
-
-#endif							/* PG_LOGICAL_MESSAGE_H */
-- 
2.17.1