diff --git a/contrib/test_decoding/t/001_twophase.pl b/contrib/test_decoding/t/001_twophase.pl
new file mode 100644
index 0000000000..6722317c9f
--- /dev/null
+++ b/contrib/test_decoding/t/001_twophase.pl
@@ -0,0 +1,102 @@
+# logical replication of 2PC test
+use strict;
+use warnings;
+use PostgresNode;
+use TestLib;
+use Test::More tests => 3;
+
+# Initialize node
+my $node_logical = get_new_node('logical');
+$node_logical->init(allows_streaming => 'logical');
+$node_logical->append_conf(
+        'postgresql.conf', qq(
+        max_prepared_transactions = 10
+));
+$node_logical->start;
+
+# Create some pre-existing content on logical
+$node_logical->safe_psql('postgres', "CREATE TABLE tab (a int PRIMARY KEY)");
+$node_logical->safe_psql('postgres',
+	"INSERT INTO tab SELECT generate_series(1,10)");
+$node_logical->safe_psql('postgres',
+	"SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_decoding');");
+$node_logical->safe_psql('postgres',
+	"SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot2', 'test_decoding');");
+
+# This test is specifically for testing concurrent abort while logical decode is
+# ongoing. The decode-delay value will allow for each change decode to sleep for
+# those many seconds. We also hold the LogicalLockTransaction while we sleep.
+# We will fire off a ROLLBACK from another session when this delayed decode is
+# ongoing. Since we are holding the lock from the call above, this ROLLBACK
+# will wait for the logical backends to do a LogicalUnlockTransaction. We will
+# stop decoding immediately post this and the next pg_logical_slot_get_changes call
+# should show only a few records decoded from the entire two phase transaction
+#
+# We use two slots to test multiple decoding backends here
+
+$node_logical->safe_psql('postgres', "
+    BEGIN;
+    INSERT INTO tab VALUES (11);
+    INSERT INTO tab VALUES (12);
+    ALTER TABLE tab ADD COLUMN b INT;
+    PREPARE TRANSACTION 'test_prepared_tab';");
+
+# start decoding the above with decode-delay in the background.
+my $logical_connstr = $node_logical->connstr . ' dbname=postgres';
+
+# decode now, it should only decode 1 INSERT record and should include
+# an ABORT entry because of the ROLLBACK below
+system_log("psql -d \"$logical_connstr\" -c \"SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'twophase-decoding', '1', 'twophase-decode-with-catalog-changes', '1', 'decode-delay', '3');\" \&");
+
+system_log("psql -d \"$logical_connstr\" -c \"SELECT data FROM pg_logical_slot_get_changes('regression_slot2', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'twophase-decoding', '1', 'twophase-decode-with-catalog-changes', '1', 'decode-delay', '3');\" \&");
+
+# sleep for a little while (shorter than decode-delay)
+$node_logical->safe_psql('postgres', "select pg_sleep(1)");
+
+# rollback the prepared transaction whose first record is being decoded
+# after sleeping for decode-delay time
+$node_logical->safe_psql('postgres', "ROLLBACK PREPARED 'test_prepared_tab';");
+
+# wait for decoding to stop
+$node_logical->psql('postgres', "select pg_sleep(4)");
+
+# consume any remaining changes
+$node_logical->safe_psql('postgres', "SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'twophase-decoding', '1', 'twophase-decode-with-catalog-changes', '1');");
+
+$node_logical->safe_psql('postgres', "SELECT data FROM pg_logical_slot_get_changes('regression_slot2', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'twophase-decoding', '1', 'twophase-decode-with-catalog-changes', '1');");
+
+# check for occurrence of log about waiting backends
+my $output_file = slurp_file($node_logical->logfile());
+my $waiting_str = "Waiting for backends to abort";
+like($output_file, qr/$waiting_str/, "Waiting log found in server log");
+
+# check for occurrence of log about stopping decoding
+my $abort_str = "stopping decoding of test_prepared_tab ";
+like($output_file, qr/$abort_str/, "ABORT found in server log");
+
+# Check that commit prepared is decoded properly on immediate restart
+$node_logical->safe_psql('postgres', "
+    BEGIN;
+    INSERT INTO tab VALUES (11);
+    INSERT INTO tab VALUES (12);
+    ALTER TABLE tab ADD COLUMN b INT;
+    INSERT INTO tab VALUES (13, 11);
+    PREPARE TRANSACTION 'test_prepared_tab';");
+# consume changes
+$node_logical->safe_psql('postgres', "SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'twophase-decoding', '1', 'twophase-decode-with-catalog-changes', '1');");
+$node_logical->safe_psql('postgres', "SELECT data FROM pg_logical_slot_get_changes('regression_slot2', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'twophase-decoding', '1', 'twophase-decode-with-catalog-changes', '1');");
+$node_logical->stop('immediate');
+$node_logical->start;
+
+# commit post the restart
+$node_logical->safe_psql('postgres', "COMMIT PREPARED 'test_prepared_tab';");
+$node_logical->safe_psql('postgres', "SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'twophase-decoding', '1', 'twophase-decode-with-catalog-changes', '1');");
+$node_logical->safe_psql('postgres', "SELECT data FROM pg_logical_slot_get_changes('regression_slot2', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'twophase-decoding', '1', 'twophase-decode-with-catalog-changes', '1');");
+
+# check inserts are visible
+my $result = $node_logical->safe_psql('postgres', "SELECT count(*) FROM tab where a IN (11,12) OR b IN (11);");
+is($result, qq(3), 'Rows inserted via 2PC are visible on restart');
+
+$node_logical->safe_psql('postgres', "SELECT pg_drop_replication_slot('regression_slot');");
+$node_logical->safe_psql('postgres', "SELECT pg_drop_replication_slot('regression_slot2');");
+$node_logical->stop('fast');
diff --git a/contrib/test_decoding/test_decoding.c b/contrib/test_decoding/test_decoding.c
index 0f18afa852..477c950b8d 100644
--- a/contrib/test_decoding/test_decoding.c
+++ b/contrib/test_decoding/test_decoding.c
@@ -24,6 +24,8 @@
 #include "replication/message.h"
 #include "replication/origin.h"
 
+#include "storage/procarray.h"
+
 #include "utils/builtins.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
@@ -46,6 +48,9 @@ typedef struct
 	bool		skip_empty_xacts;
 	bool		xact_wrote_changes;
 	bool		only_local;
+	int			decode_delay; /* seconds to sleep after every change record */
 } TestDecodingData;
 
 static void pg_decode_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
@@ -107,6 +134,9 @@ pg_decode_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
 	data->include_timestamp = false;
 	data->skip_empty_xacts = false;
 	data->only_local = false;
+	data->decode_delay = 0;
 
 	ctx->output_plugin_private = data;
 
@@ -176,6 +204,41 @@ pg_decode_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
 						 errmsg("could not parse value \"%s\" for parameter \"%s\"",
 								strVal(elem->arg), elem->defname)));
 		}
+		else if (strcmp(elem->defname, "decode-delay") == 0)
+		{
+			if (elem->arg == NULL)
+				data->decode_delay = 2; /* default to 2 seconds */
+			else
+				data->decode_delay = pg_atoi(strVal(elem->arg),
+											 sizeof(int), 0);
+
+			if (data->decode_delay <= 0)
+				ereport(ERROR,
+						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				  errmsg("Specify positive value for parameter \"%s\","
+						 " you specified \"%s\"",
+						 elem->defname, strVal(elem->arg))));
+		}
 		else
 		{
 			ereport(ERROR,
@@ -409,8 +622,18 @@ pg_decode_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 	}
 	data->xact_wrote_changes = true;
 
+	if (!LogicalLockTransaction(txn))
+		return;
+	/* if decode_delay is specified, sleep with above lock held */
+	if (data->decode_delay > 0)
+	{
+		elog(LOG, "sleeping for %d seconds", data->decode_delay);
+		pg_usleep(data->decode_delay * 1000000L);
+	}
 	class_form = RelationGetForm(relation);
 	tupdesc = RelationGetDescr(relation);
+	LogicalUnlockTransaction(txn);
+
 
 	/* Avoid leaking memory by using and resetting our own context */
 	old = MemoryContextSwitchTo(data->context);
diff --git a/doc/src/sgml/logicaldecoding.sgml b/doc/src/sgml/logicaldecoding.sgml
index 5501eed108..7edda72e5e 100644
--- a/doc/src/sgml/logicaldecoding.sgml
+++ b/doc/src/sgml/logicaldecoding.sgml
@@ -559,12 +638,30 @@ typedef void (*LogicalDecodeCommitCB) (struct LogicalDecodingContext *ctx,
       an <command>INSERT</command>, <command>UPDATE</command>,
       or <command>DELETE</command>. Even if the original command modified
       several rows at once the callback will be called individually for each
-      row.
+      row. The <function>change_cb</function> callback may access system or
+      user catalog tables to aid in the process of outputting the row
+      modification details. The <function>change_cb</function> call should invoke
+      <function>LogicalLockTransaction</function> function before such access of
+      system or user catalog tables. In case of decoding a prepared (but yet
+      uncommitted) transaction or decoding of an uncommitted transaction, this
+      function interlocks the decoding activity with simultaneous rollback by
+      another backend of this very same transaction. The
+      <function>change_cb</function> should invoke
+      <function>LogicalUnlockTransaction</function> function immediately after
+      the catalog tables access.
 <programlisting>
 typedef void (*LogicalDecodeChangeCB) (struct LogicalDecodingContext *ctx,
                                        ReorderBufferTXN *txn,
                                        Relation relation,
                                        ReorderBufferChange *change);
+</programlisting>
+      Here's an example of the use of <function>LogicalLockTransaction</function>
+      and <function>LogicalUnlockTransaction</function> in an output plugin:
+<programlisting>
+    if (!LogicalLockTransaction(txn))
+        return;
+    relation = RelationIdGetRelation(reloid);
+    LogicalUnlockTransaction(txn);
 </programlisting>
       The <parameter>ctx</parameter> and <parameter>txn</parameter> parameters
       have the same contents as for the <function>begin_cb</function>
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index c479c4881b..97499707f7 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -1435,13 +1510,19 @@ FinishPreparedTransaction(const char *gid, bool isCommit)
 	ProcArrayRemove(proc, latestXid);
+	/*
+	 * Tell logical decoding backends interested in this XID
+	 * that this is going away
+	 */
+	LogicalDecodeRemoveTransaction(proc, isCommit);
 
 	/*
 	 * In case we fail while running the callbacks, mark the gxact invalid so
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 7637efc32e..50c08acc34 100644
@@ -1013,3 +1217,164 @@ LogicalConfirmReceivedLocation(XLogRecPtr lsn)
 		SpinLockRelease(&MyReplicationSlot->mutex);
 	}
 }
+
+bool
+LogicalLockTransaction(ReorderBufferTXN *txn)
+{
+	bool	ok = false;
+
+	/*
+	 * Prepared transactions and uncommitted transactions
+	 * that have modified catalogs need to interlock with
+	 * concurrent rollback to ensure that there are no
+	 * issues while decoding
+	 */
+
+	if (!txn_has_catalog_changes(txn))
+		return true;
+
+	/*
+	 * Is it a prepared txn? Similar checks for uncommitted
+	 * transactions when we start supporting them
+	 */
+	if (!txn_prepared(txn))
+		return true;
+
+	/* check cached status */
+	if (txn_commit(txn))
+		return true;
+	if (txn_rollback(txn))
+		return false;
+
+	/*
+	 * Find the PROC that is handling this XID and add ourself as a
+	 * decodeGroupMember
+	 */
+	if (MyProc->decodeGroupLeader == NULL)
+	{
+		PGPROC *proc = BecomeDecodeGroupLeader(txn->xid, txn_prepared(txn));
+
+		/*
+		 * If decodeGroupLeader is NULL, then the only possibility
+		 * is that the transaction completed and went away
+		 */
+		if (proc == NULL)
+		{
+			Assert(!TransactionIdIsInProgress(txn->xid));
+			if (TransactionIdDidCommit(txn->xid))
+			{
+				txn->txn_flags |= TXN_COMMIT;
+				return true;
+			}
+			else
+			{
+				txn->txn_flags |= TXN_ROLLBACK;
+				return false;
+			}
+		}
+
+		/* Add ourself as a decodeGroupMember */
+		if (!BecomeDecodeGroupMember(proc, proc->pid, txn_prepared(txn)))
+		{
+			Assert(!TransactionIdIsInProgress(txn->xid));
+			if (TransactionIdDidCommit(txn->xid))
+			{
+				txn->txn_flags |= TXN_COMMIT;
+				return true;
+			}
+			else
+			{
+				txn->txn_flags |= TXN_ROLLBACK;
+				return false;
+			}
+		}
+	}
+
+	/*
+	 * If we were able to add ourself, then Abort processing will
+	 * interlock with us. Check if the transaction is still around
+	 */
+	Assert(MyProc->decodeGroupLeader);
+
+	if (MyProc->decodeGroupLeader)
+	{
+		LWLock	   *leader_lwlock;
+
+		leader_lwlock = LockHashPartitionLockByProc(MyProc->decodeGroupLeader);
+		LWLockAcquire(leader_lwlock, LW_SHARED);
+		if (MyProc->decodeAbortPending)
+		{
+			/*
+			 * Remove ourself from the decodeGroupMembership and return
+			 * false so that the decoding plugin also initiates abort
+			 * processing
+			 */
+			LWLockRelease(leader_lwlock);
+			LWLockAcquire(leader_lwlock, LW_EXCLUSIVE);
+			RemoveDecodeGroupMemberLocked(MyProc->decodeGroupLeader);
+			/* reset the bool to let the leader know that we are going away */
+			MyProc->decodeAbortPending = false;
+			MyProc->decodeLocked = false;
+			txn->txn_flags |= TXN_ROLLBACK;
+			ok = false;
+		}
+		else
+		{
+			MyProc->decodeLocked = true;
+			ok = true;
+		}
+		LWLockRelease(leader_lwlock);
+	}
+	else
+		return false;
+
+	return ok;
+}
+
+void
+LogicalUnlockTransaction(ReorderBufferTXN *txn)
+{
+	LWLock	   *leader_lwlock;
+
+	/*
+	 * Prepared transactions and uncommitted transactions
+	 * that have modified catalogs need to interlock with
+	 * concurrent rollback to ensure that there are no
+	 * issues while decoding
+	 */
+
+	if (!txn_has_catalog_changes(txn))
+		return;
+
+	/*
+	 * Is it a prepared txn? Similar checks for uncommitted
+	 * transactions when we start supporting them
+	 */
+	if (!txn_prepared(txn))
+		return;
+
+	/* check cached status */
+	if (txn_commit(txn))
+		return;
+	if (txn_rollback(txn))
+		return;
+
+	Assert(MyProc->decodeGroupLeader);
+	leader_lwlock = LockHashPartitionLockByProc(MyProc->decodeGroupLeader);
+	LWLockAcquire(leader_lwlock, LW_SHARED);
+	if (MyProc->decodeAbortPending)
+	{
+		/*
+		 * Remove ourself from the decodeGroupMembership
+		 */
+		LWLockRelease(leader_lwlock);
+		LWLockAcquire(leader_lwlock, LW_EXCLUSIVE);
+		RemoveDecodeGroupMemberLocked(MyProc->decodeGroupLeader);
+		/* reset the bool to let the leader know that we are going away */
+		MyProc->decodeAbortPending = false;
+		txn->txn_flags |= TXN_ROLLBACK;
+	}
+	MyProc->decodeLocked = false;
+	LWLockRelease(leader_lwlock);
+	return;
+}
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index c72a611a39..0b45f4f20f 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -1375,7 +1410,17 @@ ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
 							 relpathperm(change->data.tp.relnode,
 										 MAIN_FORKNUM));
 
+					/* Lock transaction before catalog access */
+					if (!LogicalLockTransaction(txn))
+					{
+						elog(LOG, "stopping decoding of %s (%u)",
+							 is_prepared? txn->gid:"",
+							 txn->xid);
+						change_cleanup = true;
+						goto change_cleanuptxn;
+					}
 					relation = RelationIdGetRelation(reloid);
+					LogicalUnlockTransaction(txn);
 
 					if (relation == NULL)
 						elog(ERROR, "could not open relation with OID %u (for filenode \"%s\")",
@@ -1546,6 +1591,7 @@ ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
 			}
 		}
 
+change_cleanuptxn:
 		/*
 		 * There's a speculative insertion remaining, just clean in up, it
 		 * can't have been successful, otherwise we'd gotten a confirmation
@@ -1561,8 +1607,24 @@ ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
 		ReorderBufferIterTXNFinish(rb, iterstate);
 		iterstate = NULL;
 
-		/* call commit callback */
-		rb->commit(rb, txn, commit_lsn);
+		if (change_cleanup)
+		{
+			/* call abort if we have sent any changes */
+			if (apply_started)
+				rb->abort(rb, txn, commit_lsn);
+		}
+		else
+		{
+			/* call commit or prepare callback */
+			if (txn_prepared(txn))
+				rb->prepare(rb, txn, commit_lsn);
+			else
+				rb->commit(rb, txn, commit_lsn);
+		}
+
+		/* remove ourself from the decodeGroupLeader */
+		if (MyProc->decodeGroupLeader)
+			RemoveDecodeGroupMember(MyProc->decodeGroupLeader);
 
 		/* this is just a sanity check against bad output plugin behaviour */
 		if (GetCurrentTransactionIdIfAny() != InvalidTransactionId)
diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index 1a00011adc..f6bb4e509f 100644
--- a/src/backend/storage/ipc/procarray.c
+++ b/src/backend/storage/ipc/procarray.c
@@ -2432,6 +2432,45 @@ BackendXidGetPid(TransactionId xid)
 	return result;
 }
 
+/*
+ * BackendXidGetProc -- get a backend's PGPROC given its XID
+ *
+ * Note that it is up to the caller to be sure that the question
+ * remains meaningful for long enough for the answer to be used ...
+ *
+ * Only main transaction Ids are considered.
+ *
+ */
+PGPROC *
+BackendXidGetProc(TransactionId xid)
+{
+	PGPROC			*result = NULL;
+	ProcArrayStruct *arrayP = procArray;
+	int			index;
+
+	if (xid == InvalidTransactionId)	/* never match invalid xid */
+		return 0;
+
+	LWLockAcquire(ProcArrayLock, LW_SHARED);
+
+	for (index = 0; index < arrayP->numProcs; index++)
+	{
+		int			pgprocno = arrayP->pgprocnos[index];
+		PGPROC *proc = &allProcs[pgprocno];
+		volatile PGXACT *pgxact = &allPgXact[pgprocno];
+
+		if (pgxact->xid == xid)
+		{
+			result = proc;
+			break;
+		}
+	}
+
+	LWLockRelease(ProcArrayLock);
+
+	return result;
+}
+
 /*
  * IsBackendPid -- is a given pid a running backend
  *
diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c
index 6f30e082b2..26d35c7807 100644
--- a/src/backend/storage/lmgr/proc.c
+++ b/src/backend/storage/lmgr/proc.c
@@ -267,6 +267,11 @@ InitProcGlobal(void)
 
 		/* Initialize lockGroupMembers list. */
 		dlist_init(&procs[i].lockGroupMembers);
+
+		/* Initialize decodeGroupMembers list. */
+		dlist_init(&procs[i].decodeGroupMembers);
+		procs[i].decodeAbortPending = false;
+		procs[i].decodeLocked = false;
 	}
 
 	/*
@@ -406,6 +411,12 @@ InitProcess(void)
 	Assert(MyProc->lockGroupLeader == NULL);
 	Assert(dlist_is_empty(&MyProc->lockGroupMembers));
 
+	/* Check that group decode fields are in a proper initial state. */
+	Assert(MyProc->decodeGroupLeader == NULL);
+	Assert(dlist_is_empty(&MyProc->decodeGroupMembers));
+	MyProc->decodeAbortPending = false;
+	MyProc->decodeLocked = false;
+
 	/* Initialize wait event information. */
 	MyProc->wait_event_info = 0;
 
@@ -581,6 +592,12 @@ InitAuxiliaryProcess(void)
 	Assert(MyProc->lockGroupLeader == NULL);
 	Assert(dlist_is_empty(&MyProc->lockGroupMembers));
 
+	/* Check that group decode fields are in a proper initial state. */
+	Assert(MyProc->decodeGroupLeader == NULL);
+	Assert(dlist_is_empty(&MyProc->decodeGroupMembers));
+	MyProc->decodeAbortPending = false;
+	MyProc->decodeLocked = false;
+
 	/*
 	 * We might be reusing a semaphore that belonged to a failed process. So
 	 * be careful and reinitialize its value here.  (This is not strictly
@@ -1887,3 +1904,268 @@ BecomeLockGroupMember(PGPROC *leader, int pid)
 
 	return ok;
 }
+
+/*
+ * BecomeDecodeGroupLeader - designate process as decode group leader
+ *
+ * Once this function has returned, other processes can join the decode group
+ * by calling BecomeDecodeGroupMember.
+ */
+PGPROC *
+BecomeDecodeGroupLeader(TransactionId xid, bool is_prepared)
+{
+	PGPROC *proc = NULL;
+	int			pid;
+	LWLock	   *leader_lwlock;
+
+	Assert(xid != InvalidTransactionId);
+
+
+	proc = BackendXidGetProc(xid);
+	if (proc)
+		pid = proc->pid;
+
+	/*
+	 * This proc will become decodeGroupLeader if it's
+	 * not already
+	 */
+	if (proc && proc->decodeGroupLeader != proc)
+	{
+		volatile PGXACT *pgxact;
+		/* Create single-member group, containing this proc. */
+		leader_lwlock = LockHashPartitionLockByProc(proc);
+		LWLockAcquire(leader_lwlock, LW_EXCLUSIVE);
+		/* recheck we are still the same */
+		pgxact = &ProcGlobal->allPgXact[proc->pgprocno];
+		if (proc->pid == pid && pgxact->xid == xid)
+		{
+			if (is_prepared)
+				Assert(pid == 0);
+			/* recheck if someone else did not already assign us */
+			if (proc->decodeGroupLeader != proc)
+			{
+				/* We had better not be a follower. */
+				Assert(proc->decodeGroupLeader == NULL);
+				proc->decodeGroupLeader = proc;
+				dlist_push_head(&proc->decodeGroupMembers,
+								&proc->decodeGroupLink);
+			}
+		}
+		else
+		{
+			/* proc entry is gone */
+			proc = NULL;
+		}
+		LWLockRelease(leader_lwlock);
+	}
+
+	elog(DEBUG1, "became group leader (%p)", proc);
+	return proc;
+}
+
+/*
+ * BecomeDecodeGroupMember - designate process as decode group member
+ *
+ * This is pretty straightforward except for the possibility that the leader
+ * whose group we're trying to join might exit before we manage to do so;
+ * and the PGPROC might get recycled for an unrelated process.  To avoid
+ * that, we require the caller to pass the PID of the intended PGPROC as
+ * an interlock.  Returns true if we successfully join the intended lock
+ * group, and false if not.
+ */
+bool
+BecomeDecodeGroupMember(PGPROC *leader, int pid, bool is_prepared)
+{
+	LWLock	   *leader_lwlock;
+	bool		ok = false;
+
+	/* Group leader can't become member of group */
+	Assert(MyProc != leader);
+
+	/* Can't already be a member of a group */
+	Assert(MyProc->decodeGroupLeader == NULL);
+
+	/* PID must be valid OR this is a prepared transaction. */
+	Assert(pid != 0 || is_prepared);
+
+	/*
+	 * Get lock protecting the group fields.  Note LockHashPartitionLockByProc
+	 * accesses leader->pgprocno in a PGPROC that might be free.  This is safe
+	 * because all PGPROCs' pgprocno fields are set during shared memory
+	 * initialization and never change thereafter; so we will acquire the
+	 * correct lock even if the leader PGPROC is in process of being recycled.
+	 */
+	leader_lwlock = LockHashPartitionLockByProc(leader);
+	LWLockAcquire(leader_lwlock, LW_EXCLUSIVE);
+
+	/* Is this the leader we're looking for? */
+	if (leader->pid == pid && leader->decodeGroupLeader == leader)
+	{
+		if (is_prepared)
+			Assert(pid == 0);
+		/* is the leader going away? */
+		if (leader->decodeAbortPending)
+			ok = false;
+		else
+		{
+			/* OK, join the group */
+			ok = true;
+			MyProc->decodeGroupLeader = leader;
+			dlist_push_tail(&leader->decodeGroupMembers, &MyProc->decodeGroupLink);
+		}
+	}
+	else
+		MyProc->decodeGroupLeader = NULL;
+	LWLockRelease(leader_lwlock);
+
+	elog(DEBUG1, "became group member (%p) to (%p)", MyProc, leader);
+	return ok;
+}
+
+/*
+ * Remove a decodeGroupMember from the decodeGroupMembership of
+ * decodeGroupLeader
+ * Acquire lock
+ */
+void
+RemoveDecodeGroupMember(PGPROC *leader)
+{
+	LWLock	   *leader_lwlock;
+
+	leader_lwlock = LockHashPartitionLockByProc(leader);
+	LWLockAcquire(leader_lwlock, LW_EXCLUSIVE);
+	RemoveDecodeGroupMemberLocked(leader);
+	LWLockRelease(leader_lwlock);
+
+	return;
+}
+
+/*
+ * Remove a decodeGroupMember from the decodeGroupMembership of
+ * decodeGroupLeader
+ * Assumes that the caller is holding appropriate lock
+ */
+void
+RemoveDecodeGroupMemberLocked(PGPROC *leader)
+{
+	Assert(!dlist_is_empty(&leader->decodeGroupMembers));
+	dlist_delete(&MyProc->decodeGroupLink);
+	/* leader links to itself, so never empty */
+	Assert(!dlist_is_empty(&leader->decodeGroupMembers));
+	MyProc->decodeGroupLeader = NULL;
+	elog(DEBUG1, "removed group member (%p) from (%p)", MyProc, leader);
+
+	return;
+}
+
+/*
+ * Indicate to all decodeGroupMembers that this transaction is
+ * going away.
+ *
+ * Wait for all decodeGroupMembers to ack back before returning
+ * from here but only in case of aborts.
+ *
+ * This function should be called *after* the proc has been
+ * removed from the procArray.
+ *
+ * If the transaction is committing, it's ok for the
+ * decoders to continue merrily. When it tries to lock this
+ * proc, it won't find it and check for transaction status
+ * and cache the commit status for future calls in
+ * LogicalLockTransaction
+ */
+void
+LogicalDecodeRemoveTransaction(PGPROC *leader, bool isCommit)
+{
+	LWLock	   *leader_lwlock;
+	dlist_mutable_iter change_i;
+	dlist_iter  iter;
+	PGPROC	   *proc;
+	bool		do_wait;
+
+	leader_lwlock = LockHashPartitionLockByProc(leader);
+	LWLockAcquire(leader_lwlock, LW_EXCLUSIVE);
+	/* mark ourself as aborting */
+	if (!isCommit)
+		leader->decodeAbortPending = true;
+
+	if (leader->decodeGroupLeader == NULL)
+	{
+		Assert(dlist_is_empty(&leader->decodeGroupMembers));
+		LWLockRelease(leader_lwlock);
+		return;
+	}
+
+recheck:
+	do_wait = false;
+	Assert(leader->decodeGroupLeader == leader);
+	Assert(!dlist_is_empty(&leader->decodeGroupMembers));
+	if (!isCommit)
+	{
+		dlist_foreach(iter, &leader->decodeGroupMembers)
+		{
+			proc = dlist_container(PGPROC, decodeGroupLink, iter.cur);
+			/* mark the proc to indicate abort is pending */
+			if (proc == leader)
+				continue;
+			if (!proc->decodeAbortPending)
+			{
+				proc->decodeAbortPending = true;
+				elog(DEBUG1, "marking group member (%p) from (%p) for abort",
+					 proc, leader);
+			}
+			/* if the proc is currently locked, wait */
+			if (proc->decodeLocked)
+				do_wait = true;
+		}
+
+		if (do_wait)
+		{
+			int			rc;
+			LWLockRelease(leader_lwlock);
+
+			elog(LOG, "Waiting for backends to abort decoding");
+			/*
+			 * Wait on our latch to allow decodeGroupMembers to
+			 * go away soon
+			 */
+			rc = WaitLatch(MyLatch,
+						   WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
+						   100L,
+						   WAIT_EVENT_PG_SLEEP);
+			ResetLatch(MyLatch);
+
+			/* emergency bailout if postmaster has died */
+			if (rc & WL_POSTMASTER_DEATH)
+				proc_exit(1);
+
+			CHECK_FOR_INTERRUPTS();
+
+			/* Recheck decodeGroupMembers */
+			LWLockAcquire(leader_lwlock, LW_EXCLUSIVE);
+			goto recheck;
+		}
+	}
+
+	/*
+	 * All backends exited cleanly in case of aborts above,
+	 * remove decodeGroupMembers now for both commit/abort cases
+	 */
+	Assert(leader->decodeGroupLeader == leader);
+	Assert(!dlist_is_empty(&leader->decodeGroupMembers));
+	dlist_foreach_modify(change_i, &leader->decodeGroupMembers)
+	{
+		proc = dlist_container(PGPROC, decodeGroupLink, change_i.cur);
+		Assert(!proc->decodeLocked);
+		dlist_delete(&proc->decodeGroupLink);
+		elog(DEBUG1, "deleting group member (%p) from (%p)",
+			 proc, leader);
+		proc->decodeGroupLeader = NULL;
+	}
+	Assert(dlist_is_empty(&leader->decodeGroupMembers));
+	leader->decodeGroupLeader = NULL;
+	leader->decodeAbortPending = false;
+	LWLockRelease(leader_lwlock);
+
+	return;
+}
diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h
index 619c5f4d73..9dad4c997f 100644
--- a/src/include/replication/logical.h
+++ b/src/include/replication/logical.h
@@ -117,6 +122,8 @@ extern void LogicalIncreaseXminForSlot(XLogRecPtr lsn, TransactionId xmin);
 extern void LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn,
 									  XLogRecPtr restart_lsn);
 extern void LogicalConfirmReceivedLocation(XLogRecPtr lsn);
+extern bool LogicalLockTransaction(ReorderBufferTXN *txn);
+extern void LogicalUnlockTransaction(ReorderBufferTXN *txn);
 
 extern bool filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id);
 
diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h
index 5c19a61dcf..fdfc582874 100644
--- a/src/include/storage/proc.h
+++ b/src/include/storage/proc.h
@@ -200,6 +200,26 @@ struct PGPROC
 	PGPROC	   *lockGroupLeader;	/* lock group leader, if I'm a member */
 	dlist_head	lockGroupMembers;	/* list of members, if I'm a leader */
 	dlist_node	lockGroupLink;	/* my member link, if I'm a member */
+
+	/*
+	 * Support for decoding groups.  Use LockHashPartitionLockByProc on the group
+	 * leader to get the LWLock protecting these fields.
+	 *
+	 * For prepared and uncommitted transactions, decoding backends working on
+	 * the same XID will link themselves up to the corresponding PGPROC
+	 * entry (decodeGroupLeader).
+	 *
+	 * They will remove themselves when they are done decoding.
+	 *
+	 * If the prepared or uncommitted transaction decides to abort, then
+	 * the decodeGroupLeader will set the decodeAbortPending flag allowing
+	 * the decodeGroupMembers to abort their decoding appropriately
+	 */
+	PGPROC	   *decodeGroupLeader;	/* decode group leader, if I'm a member */
+	dlist_head	decodeGroupMembers;	/* list of members, if I'm a leader */
+	dlist_node	decodeGroupLink;	/* my member link, if I'm a member */
+	bool		decodeLocked;		/* is it currently locked by this proc? */
+	bool		decodeAbortPending; /* is the decode group leader aborting? */
 };
 
 /* NOTE: "typedef struct PGPROC PGPROC" appears in storage/lock.h. */
@@ -326,5 +346,10 @@ extern PGPROC *AuxiliaryPidGetProc(int pid);
 
 extern void BecomeLockGroupLeader(void);
 extern bool BecomeLockGroupMember(PGPROC *leader, int pid);
+extern PGPROC *BecomeDecodeGroupLeader(TransactionId xid, bool is_prepared);
+extern bool BecomeDecodeGroupMember(PGPROC *leader, int pid, bool is_prepared);
+extern void RemoveDecodeGroupMember(PGPROC *leader);
+extern void RemoveDecodeGroupMemberLocked(PGPROC *leader);
+extern void LogicalDecodeRemoveTransaction(PGPROC *leader, bool isCommit);
 
 #endif							/* PROC_H */
diff --git a/src/include/storage/procarray.h b/src/include/storage/procarray.h
index 75bab2985f..68173743ae 100644
--- a/src/include/storage/procarray.h
+++ b/src/include/storage/procarray.h
@@ -98,6 +98,7 @@ extern bool HaveVirtualXIDsDelayingChkpt(VirtualTransactionId *vxids, int nvxids
 extern PGPROC *BackendPidGetProc(int pid);
 extern PGPROC *BackendPidGetProcWithLock(int pid);
 extern int	BackendXidGetPid(TransactionId xid);
+extern PGPROC *BackendXidGetProc(TransactionId xid);
 extern bool IsBackendPid(int pid);
 
 extern VirtualTransactionId *GetCurrentVirtualXIDs(TransactionId limitXmin,
