From b0530ad3660d5571b56478100c94b702307cd09e Mon Sep 17 00:00:00 2001
From: "houzj.fnst" <houzj.fnst@cn.fujitsu.com>
Date: Wed, 20 Apr 2022 16:45:07 +0800
Subject: [PATCH v18 1/4] Perform streaming logical transactions by background
 workers

Currently, for large transactions, the publisher sends the data in multiple
streams (changes divided into chunks depending upon logical_decoding_work_mem),
and then on the subscriber-side, the apply worker writes the changes into
temporary files and once it receives the commit, it reads from the file and
applies the entire transaction. To improve the performance of such
transactions, we can instead allow them to be applied via background workers.

In this approach, we assign a new apply background worker (if available) as
soon as the xact's first stream is received and the main apply worker will send
changes to this new worker via shared memory. The apply background worker will
directly apply the change instead of writing it to temporary files. We keep
this worker assigned till the transaction commit is received and also wait for
the worker to finish at commit. This preserves commit ordering and avoids
writing to and reading from file in most cases. We still need to spill if there
is no worker available.

This patch also extends the SUBSCRIPTION 'streaming' parameter so that the user
can control whether to apply the streaming transaction in an apply background
worker or spill the change to disk. The user can set the streaming parameter to
'on/off', 'parallel'. The parameter value 'parallel' means the streaming will
be applied via an apply background worker, if available. The parameter value
'on' means the streaming transaction will be spilled to disk. The default value
is 'off' (same as current behaviour).
---
 doc/src/sgml/catalogs.sgml                    |  10 +-
 doc/src/sgml/config.sgml                      |  25 +
 doc/src/sgml/logical-replication.sgml         |  10 +
 doc/src/sgml/protocol.sgml                    |  19 +
 doc/src/sgml/ref/create_subscription.sgml     |  24 +-
 src/backend/access/transam/xact.c             |  13 +
 src/backend/commands/subscriptioncmds.c       |  66 +-
 src/backend/postmaster/bgworker.c             |   3 +
 src/backend/replication/logical/Makefile      |   1 +
 .../replication/logical/applybgworker.c       | 802 ++++++++++++++++++
 src/backend/replication/logical/decode.c      |  10 +-
 src/backend/replication/logical/launcher.c    | 130 ++-
 src/backend/replication/logical/origin.c      |  26 +-
 src/backend/replication/logical/proto.c       |  41 +-
 .../replication/logical/reorderbuffer.c       |  10 +-
 src/backend/replication/logical/tablesync.c   |  10 +-
 src/backend/replication/logical/worker.c      | 692 +++++++++++----
 src/backend/replication/pgoutput/pgoutput.c   |   6 +-
 src/backend/utils/activity/wait_event.c       |   3 +
 src/backend/utils/misc/guc.c                  |  12 +
 src/backend/utils/misc/postgresql.conf.sample |   1 +
 src/bin/pg_dump/pg_dump.c                     |   6 +-
 src/include/catalog/pg_subscription.h         |  21 +-
 src/include/replication/logicallauncher.h     |   1 +
 src/include/replication/logicalproto.h        |  27 +-
 src/include/replication/logicalworker.h       |   1 +
 src/include/replication/origin.h              |   2 +-
 src/include/replication/reorderbuffer.h       |   7 +-
 src/include/replication/worker_internal.h     | 102 ++-
 src/include/utils/wait_event.h                |   1 +
 src/test/regress/expected/subscription.out    |   2 +-
 src/tools/pgindent/typedefs.list              |   5 +
 32 files changed, 1869 insertions(+), 220 deletions(-)
 create mode 100644 src/backend/replication/logical/applybgworker.c

diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index 670a5406d6..13c9e8eca1 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -7873,11 +7873,15 @@ SCRAM-SHA-256$<replaceable>&lt;iteration count&gt;</replaceable>:<replaceable>&l
 
      <row>
       <entry role="catalog_table_entry"><para role="column_definition">
-       <structfield>substream</structfield> <type>bool</type>
+       <structfield>substream</structfield> <type>char</type>
       </para>
       <para>
-       If true, the subscription will allow streaming of in-progress
-       transactions
+       Controls how to handle the streaming of in-progress transactions:
+       <literal>f</literal> = disallow streaming of in-progress transactions,
+       <literal>t</literal> = spill the changes of in-progress transactions to
+       disk and apply at once after the transaction is committed on the
+       publisher,
+       <literal>p</literal> = apply changes directly using a background worker
       </para></entry>
      </row>
 
diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 37fd80388c..6bbd986195 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -4970,6 +4970,31 @@ ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="
       </listitem>
      </varlistentry>
 
+     <varlistentry id="guc-max-apply-bgworkers-per-subscription" xreflabel="max_apply_bgworkers_per_subscription">
+      <term><varname>max_apply_bgworkers_per_subscription</varname> (<type>integer</type>)
+      <indexterm>
+       <primary><varname>max_apply_bgworkers_per_subscription</varname> configuration parameter</primary>
+      </indexterm>
+      </term>
+      <listitem>
+       <para>
+        Maximum number of apply background workers per subscription. This
+        parameter controls the amount of parallelism of the streaming of
+        in-progress transactions when subscription parameter
+        <literal>streaming = parallel</literal>.
+       </para>
+       <para>
+        The apply background workers are taken from the pool defined by
+        <varname>max_logical_replication_workers</varname>.
+       </para>
+       <para>
+        The default value is 2. This parameter can only be set in the
+        <filename>postgresql.conf</filename> file or on the server command
+        line.
+       </para>
+      </listitem>
+     </varlistentry>
+
      </variablelist>
     </sect2>
 
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index bdf1e7b727..92997f9299 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -1153,6 +1153,16 @@ CONTEXT:  processing remote data for replication origin "pg_16395" during "INSER
    might not violate any constraint.  This can easily make the subscriber
    inconsistent.
   </para>
+
+  <para>
+   When the streaming mode is <literal>parallel</literal>, the finish LSN of
+   failed transactions may not be logged. In that case, it may be necessary to
+   change the streaming mode to <literal>on</literal> and cause the same
+   conflicts again so the finish LSN of the failed transaction will be written
+   to the server log. For the usage of finish LSN, please refer to <link
+   linkend="sql-altersubscription"><command>ALTER SUBSCRIPTION ...
+   SKIP</command></link>.
+  </para>
  </sect1>
 
  <sect1 id="logical-replication-restrictions">
diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index c0b89a3c01..7e88ba9631 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -6809,6 +6809,25 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;"
        </listitem>
       </varlistentry>
 
+      <varlistentry>
+       <term>Int64 (XLogRecPtr)</term>
+       <listitem>
+        <para>
+         The LSN of the abort.
+        </para>
+       </listitem>
+      </varlistentry>
+
+      <varlistentry>
+       <term>Int64 (TimestampTz)</term>
+       <listitem>
+        <para>
+         Abort timestamp of the transaction. The value is in number
+         of microseconds since PostgreSQL epoch (2000-01-01).
+        </para>
+       </listitem>
+      </varlistentry>
+
       <varlistentry>
        <term>Int32 (TransactionId)</term>
        <listitem>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 34b3264b26..71dd4aca81 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -217,13 +217,29 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
        </varlistentry>
 
        <varlistentry>
-        <term><literal>streaming</literal> (<type>boolean</type>)</term>
+        <term><literal>streaming</literal> (<type>enum</type>)</term>
         <listitem>
          <para>
           Specifies whether to enable streaming of in-progress transactions
-          for this subscription.  By default, all transactions
-          are fully decoded on the publisher and only then sent to the
-          subscriber as a whole.
+          for this subscription.  The default value is <literal>off</literal>,
+          meaning all transactions are fully decoded on the publisher and only
+          then sent to the subscriber as a whole.
+         </para>
+
+         <para>
+          If set to <literal>on</literal>, the incoming changes are written to
+          temporary files and then applied only after the transaction is
+          committed on the publisher.
+         </para>
+
+         <para>
+          If set to <literal>parallel</literal>, incoming changes are directly
+          applied via one of the apply background workers, if available. If no
+          background worker is free to handle streaming transaction then the
+          changes are written to temporary files and applied after the
+          transaction is committed. Note that if an error happens when
+          applying changes in a background worker, the finish LSN of the
+          remote transaction might not be reported in the server log.
          </para>
         </listitem>
        </varlistentry>
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index 116de1175b..3e61a57b50 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -1711,6 +1711,7 @@ RecordTransactionAbort(bool isSubXact)
 	int			nchildren;
 	TransactionId *children;
 	TimestampTz xact_time;
+	bool		replorigin;
 
 	/*
 	 * If we haven't been assigned an XID, nobody will care whether we aborted
@@ -1741,6 +1742,13 @@ RecordTransactionAbort(bool isSubXact)
 		elog(PANIC, "cannot abort transaction %u, it was already committed",
 			 xid);
 
+	/*
+	 * Are we using the replication origins feature?  Or, in other words,
+	 * are we replaying remote actions?
+	 */
+	replorigin = (replorigin_session_origin != InvalidRepOriginId &&
+				  replorigin_session_origin != DoNotReplicateId);
+
 	/* Fetch the data we need for the abort record */
 	nrels = smgrGetPendingDeletes(false, &rels);
 	nchildren = xactGetCommittedChildren(&children);
@@ -1765,6 +1773,11 @@ RecordTransactionAbort(bool isSubXact)
 					   MyXactFlags, InvalidTransactionId,
 					   NULL);
 
+	if (replorigin)
+		/* Move LSNs forward for this replication origin */
+		replorigin_session_advance(replorigin_session_origin_lsn,
+								   XactLastRecEnd);
+
 	/*
 	 * Report the latest async abort LSN, so that the WAL writer knows to
 	 * flush this abort. There's nothing to be gained by delaying this, since
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index bdc1208724..d4b2616ee6 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -83,7 +83,7 @@ typedef struct SubOpts
 	bool		copy_data;
 	bool		refresh;
 	bool		binary;
-	bool		streaming;
+	char		streaming;
 	bool		twophase;
 	bool		disableonerr;
 	XLogRecPtr	lsn;
@@ -95,6 +95,62 @@ static List *merge_publications(List *oldpublist, List *newpublist, bool addpub,
 static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err);
 
 
+/*
+ * Extract the streaming mode value from a DefElem.  This is like
+ * defGetBoolean() but also accepts the special value of "parallel".
+ */
+static char
+defGetStreamingMode(DefElem *def)
+{
+	/*
+	 * If no parameter value given, assume "true" is meant.
+	 */
+	if (def->arg == NULL)
+		return SUBSTREAM_ON;
+
+	/*
+	 * Allow 0, 1, "false", "true", "off", "on" or "parallel".
+	 */
+	switch (nodeTag(def->arg))
+	{
+		case T_Integer:
+			switch (intVal(def->arg))
+			{
+				case 0:
+					return SUBSTREAM_OFF;
+				case 1:
+					return SUBSTREAM_ON;
+				default:
+					/* otherwise, error out below */
+					break;
+			}
+			break;
+		default:
+			{
+				char	   *sval = defGetString(def);
+
+				/*
+				 * The set of strings accepted here should match up with the
+				 * grammar's opt_boolean_or_string production.
+				 */
+				if (pg_strcasecmp(sval, "false") == 0 ||
+					pg_strcasecmp(sval, "off") == 0)
+					return SUBSTREAM_OFF;
+				if (pg_strcasecmp(sval, "true") == 0 ||
+					pg_strcasecmp(sval, "on") == 0)
+					return SUBSTREAM_ON;
+				if (pg_strcasecmp(sval, "parallel") == 0)
+					return SUBSTREAM_PARALLEL;
+			}
+			break;
+	}
+	ereport(ERROR,
+			(errcode(ERRCODE_SYNTAX_ERROR),
+			 errmsg("%s requires a Boolean value or \"parallel\"",
+					def->defname)));
+	return SUBSTREAM_OFF;		/* keep compiler quiet */
+}
+
 /*
  * Common option parsing function for CREATE and ALTER SUBSCRIPTION commands.
  *
@@ -132,7 +188,7 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
 	if (IsSet(supported_opts, SUBOPT_BINARY))
 		opts->binary = false;
 	if (IsSet(supported_opts, SUBOPT_STREAMING))
-		opts->streaming = false;
+		opts->streaming = SUBSTREAM_OFF;
 	if (IsSet(supported_opts, SUBOPT_TWOPHASE_COMMIT))
 		opts->twophase = false;
 	if (IsSet(supported_opts, SUBOPT_DISABLE_ON_ERR))
@@ -233,7 +289,7 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
 				errorConflictingDefElem(defel, pstate);
 
 			opts->specified_opts |= SUBOPT_STREAMING;
-			opts->streaming = defGetBoolean(defel);
+			opts->streaming = defGetStreamingMode(defel);
 		}
 		else if (strcmp(defel->defname, "two_phase") == 0)
 		{
@@ -600,7 +656,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 	values[Anum_pg_subscription_subowner - 1] = ObjectIdGetDatum(owner);
 	values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(opts.enabled);
 	values[Anum_pg_subscription_subbinary - 1] = BoolGetDatum(opts.binary);
-	values[Anum_pg_subscription_substream - 1] = BoolGetDatum(opts.streaming);
+	values[Anum_pg_subscription_substream - 1] = CharGetDatum(opts.streaming);
 	values[Anum_pg_subscription_subtwophasestate - 1] =
 		CharGetDatum(opts.twophase ?
 					 LOGICALREP_TWOPHASE_STATE_PENDING :
@@ -1059,7 +1115,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 				if (IsSet(opts.specified_opts, SUBOPT_STREAMING))
 				{
 					values[Anum_pg_subscription_substream - 1] =
-						BoolGetDatum(opts.streaming);
+						CharGetDatum(opts.streaming);
 					replaces[Anum_pg_subscription_substream - 1] = true;
 				}
 
diff --git a/src/backend/postmaster/bgworker.c b/src/backend/postmaster/bgworker.c
index 40601aefd9..40ccb8993c 100644
--- a/src/backend/postmaster/bgworker.c
+++ b/src/backend/postmaster/bgworker.c
@@ -128,6 +128,9 @@ static const struct
 	},
 	{
 		"ApplyWorkerMain", ApplyWorkerMain
+	},
+	{
+		"ApplyBgworkerMain", ApplyBgworkerMain
 	}
 };
 
diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile
index c4e2fdeb71..cbfb5d794e 100644
--- a/src/backend/replication/logical/Makefile
+++ b/src/backend/replication/logical/Makefile
@@ -15,6 +15,7 @@ include $(top_builddir)/src/Makefile.global
 override CPPFLAGS := -I$(srcdir) $(CPPFLAGS)
 
 OBJS = \
+	applybgworker.o \
 	decode.o \
 	launcher.o \
 	logical.o \
diff --git a/src/backend/replication/logical/applybgworker.c b/src/backend/replication/logical/applybgworker.c
new file mode 100644
index 0000000000..aa222490a0
--- /dev/null
+++ b/src/backend/replication/logical/applybgworker.c
@@ -0,0 +1,802 @@
+/*-------------------------------------------------------------------------
+ * applybgworker.c
+ *     Support routines for applying xact by apply background worker
+ *
+ * Copyright (c) 2016-2022, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *	  src/backend/replication/logical/applybgworker.c
+ *
+ * This file contains routines that are intended to support setting up, using,
+ * and tearing down a ApplyBgworkerState.
+ *
+ * Refer to the comments in file header of logical/worker.c to see more
+ * information about apply background worker.
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "libpq/pqformat.h"
+#include "mb/pg_wchar.h"
+#include "pgstat.h"
+#include "postmaster/interrupt.h"
+#include "replication/logicallauncher.h"
+#include "replication/logicalworker.h"
+#include "replication/origin.h"
+#include "replication/walreceiver.h"
+#include "replication/worker_internal.h"
+#include "storage/ipc.h"
+#include "storage/procarray.h"
+#include "tcop/tcopprot.h"
+#include "utils/inval.h"
+#include "utils/memutils.h"
+#include "utils/resowner.h"
+#include "utils/syscache.h"
+
+#define PG_LOGICAL_APPLY_SHM_MAGIC 0x79fb2447 // TODO Consider change
+
+/*
+ * DSM keys for apply background worker.  Unlike other parallel execution code,
+ * since we don't need to worry about DSM keys conflicting with plan_node_id we
+ * can use small integers.
+ */
+#define APPLY_BGWORKER_KEY_SHARED	1
+#define APPLY_BGWORKER_KEY_MQ		2
+
+/* Queue size of DSM, 16 MB for now. */
+#define DSM_QUEUE_SIZE	160000000
+
+/*
+ * There are three fields in message: start_lsn, end_lsn and send_time. Because
+ * we have updated these statistics in apply worker, we could ignore these
+ * fields in apply background worker. (see function LogicalRepApplyLoop)
+ */
+#define IGNORE_SIZE_IN_MESSAGE (3 * sizeof(uint64))
+
+/*
+ * Entry for a hash table we use to map from xid to our apply background worker
+ * state.
+ */
+typedef struct ApplyBgworkerEntry
+{
+	TransactionId xid;
+	ApplyBgworkerState *wstate;
+} ApplyBgworkerEntry;
+
+/* Apply background workers hash table (initialized on first use) */
+static HTAB *ApplyWorkersHash = NULL;
+static List *ApplyWorkersFreeList = NIL;
+static List *ApplyWorkersList = NIL;
+
+/*
+ * Information shared between main apply worker and apply background worker.
+ */
+volatile ApplyBgworkerShared *MyParallelShared = NULL;
+
+List	   *subxactlist = NIL;
+
+static bool apply_bgworker_can_start(TransactionId xid);
+static ApplyBgworkerState *apply_bgworker_setup(void);
+static void apply_bgworker_setup_dsm(ApplyBgworkerState *wstate);
+
+/*
+ * Check if starting a new apply background worker is allowed.
+ */
+static bool
+apply_bgworker_can_start(TransactionId xid)
+{
+	if (!TransactionIdIsValid(xid))
+		return false;
+
+	/*
+	 * Don't start a new background worker if not in streaming parallel mode.
+	 */
+	if (MySubscription->stream != SUBSTREAM_PARALLEL)
+		return false;
+
+	/*
+	 * Don't start a new background worker if user has set skiplsn as it's
+	 * possible that user want to skip the streaming transaction. For
+	 * streaming transaction, we need to spill the transaction to disk so that
+	 * we can get the last LSN of the transaction to judge whether to skip
+	 * before starting to apply the change.
+	 */
+	if (!XLogRecPtrIsInvalid(MySubscription->skiplsn))
+		return false;
+
+	/*
+	 * For streaming transactions that are being applied in apply background
+	 * worker, we cannot decide whether to apply the change for a relation
+	 * that is not in the READY state (see should_apply_changes_for_rel) as we
+	 * won't know remote_final_lsn by that time. So, we don't start new apply
+	 * background worker in this case.
+	 */
+	if (!AllTablesyncsReady())
+		return false;
+
+	return true;
+}
+
+/*
+ * Try to start an apply background worker and, if successful, cache it in
+ * ApplyWorkersHash keyed by the specified xid.
+ */
+ApplyBgworkerState *
+apply_bgworker_start(TransactionId xid)
+{
+	bool		found;
+	int			server_version;
+	ApplyBgworkerState *wstate;
+	ApplyBgworkerEntry *entry = NULL;
+
+	if (!apply_bgworker_can_start(xid))
+		return NULL;
+
+	/* First time through, initialize apply workers hashtable */
+	if (ApplyWorkersHash == NULL)
+	{
+		HASHCTL		ctl;
+
+		MemSet(&ctl, 0, sizeof(ctl));
+		ctl.keysize = sizeof(TransactionId);
+		ctl.entrysize = sizeof(ApplyBgworkerEntry);
+		ctl.hcxt = ApplyContext;
+
+		ApplyWorkersHash = hash_create("logical apply workers hash", 8, &ctl,
+									   HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
+	}
+
+	/*
+	 * Now, we try to get an apply background worker. If there is at least one
+	 * worker in the free list, then take one. Otherwise, we try to start a
+	 * new apply background worker.
+	 */
+	if (list_length(ApplyWorkersFreeList) > 0)
+	{
+		wstate = (ApplyBgworkerState *) llast(ApplyWorkersFreeList);
+		ApplyWorkersFreeList = list_delete_last(ApplyWorkersFreeList);
+		Assert(wstate->shared->status == APPLY_BGWORKER_FINISHED);
+	}
+	else
+	{
+		wstate = apply_bgworker_setup();
+
+		if (wstate == NULL)
+			return NULL;
+	}
+
+	/*
+	 * Create entry for requested transaction.
+	 */
+	entry = hash_search(ApplyWorkersHash, &xid, HASH_ENTER, &found);
+	if (found)
+		elog(ERROR, "hash table corrupted");
+
+	/* Fill up the hash entry */
+	wstate->shared->status = APPLY_BGWORKER_BUSY;
+
+	server_version = walrcv_server_version(LogRepWorkerWalRcvConn);
+	wstate->shared->server_version =
+		server_version >= 160000 ? LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM :
+		server_version >= 150000 ? LOGICALREP_PROTO_TWOPHASE_VERSION_NUM :
+		server_version >= 140000 ? LOGICALREP_PROTO_STREAM_VERSION_NUM :
+		LOGICALREP_PROTO_VERSION_NUM;
+
+	wstate->shared->stream_xid = xid;
+	entry->wstate = wstate;
+	entry->xid = xid;
+
+	return wstate;
+}
+
+/*
+ * Try to look up worker inside ApplyWorkersHash for requested xid.
+ */
+ApplyBgworkerState *
+apply_bgworker_find(TransactionId xid)
+{
+	bool		found;
+	ApplyBgworkerEntry *entry = NULL;
+
+	if (!TransactionIdIsValid(xid))
+		return NULL;
+
+	if (ApplyWorkersHash == NULL)
+		return NULL;
+
+	/*
+	 * Find entry for requested transaction.
+	 */
+	entry = hash_search(ApplyWorkersHash, &xid, HASH_FIND, &found);
+	if (found)
+	{
+		char status = entry->wstate->shared->status;
+
+		/* If any workers (or the postmaster) have died, we have failed. */
+		if (status == APPLY_BGWORKER_EXIT)
+			ereport(ERROR,
+					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					 errmsg("background worker %u failed to apply transaction %u",
+							entry->wstate->shared->n,
+							entry->wstate->shared->stream_xid)));
+
+		Assert(status == APPLY_BGWORKER_BUSY);
+
+		return entry->wstate;
+	}
+	else
+		return NULL;
+}
+
+/*
+ * Add the worker to the free list and remove the entry from the hash table.
+ */
+void
+apply_bgworker_free(ApplyBgworkerState *wstate)
+{
+	MemoryContext oldctx;
+	TransactionId xid = wstate->shared->stream_xid;
+
+	Assert(wstate->shared->status == APPLY_BGWORKER_FINISHED);
+
+	oldctx = MemoryContextSwitchTo(ApplyContext);
+
+	hash_search(ApplyWorkersHash, &xid, HASH_REMOVE, NULL);
+
+	elog(DEBUG1, "adding finished apply worker #%u for xid %u to the free list",
+		 wstate->shared->n, wstate->shared->stream_xid);
+
+	ApplyWorkersFreeList = lappend(ApplyWorkersFreeList, wstate);
+
+	MemoryContextSwitchTo(oldctx);
+}
+
+/* Apply Background Worker main loop */
+static void
+LogicalApplyBgwLoop(shm_mq_handle *mqh, volatile ApplyBgworkerShared *shared)
+{
+	shm_mq_result shmq_res;
+	PGPROC	   *registrant;
+	ErrorContextCallback errcallback;
+
+	registrant = BackendPidGetProc(MyBgworkerEntry->bgw_notify_pid);
+	SetLatch(&registrant->procLatch);
+
+	/*
+	 * Push apply error context callback. Fields will be filled applying a
+	 * change.
+	 */
+	errcallback.callback = apply_error_callback;
+	errcallback.previous = error_context_stack;
+	error_context_stack = &errcallback;
+
+	for (;;)
+	{
+		void	   *data;
+		Size		len;
+		int			c;
+		StringInfoData s;
+		MemoryContext oldctx;
+
+		CHECK_FOR_INTERRUPTS();
+
+		/* Ensure we are reading the data into our memory context. */
+		oldctx = MemoryContextSwitchTo(ApplyMessageContext);
+
+		shmq_res = shm_mq_receive(mqh, &len, &data, false);
+
+		if (shmq_res != SHM_MQ_SUCCESS)
+			ereport(ERROR,
+					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					 errmsg("lost connection to the main apply worker")));
+
+		if (len == 0)
+			break;
+
+		s.cursor = 0;
+		s.maxlen = -1;
+		s.data = (char *) data;
+		s.len = len;
+
+		/*
+		 * We use first byte of message for additional communication between
+		 * main Logical replication worker and apply background workers, so if
+		 * it differs from 'w', then process it first.
+		 */
+		c = pq_getmsgbyte(&s);
+		switch (c)
+		{
+			/* End message of streaming chunk */
+			case LOGICAL_REP_MSG_STREAM_STOP:
+				elog(DEBUG1, "[Apply BGW #%u] ended processing streaming chunk,"
+					 "waiting on shm_mq_receive", shared->n);
+
+				in_streamed_transaction = false;
+				pgstat_report_activity(STATE_IDLEINTRANSACTION, NULL);
+				continue;
+			case 'w':
+				break;
+			default:
+				elog(ERROR, "[Apply BGW #%u] unexpected message \"%c\"",
+					 shared->n, c);
+				break;
+		}
+
+		/* Ignore statistics fields that have been updated. */
+		s.cursor += IGNORE_SIZE_IN_MESSAGE;
+
+		apply_dispatch(&s);
+
+		if (ConfigReloadPending)
+		{
+			ConfigReloadPending = false;
+			ProcessConfigFile(PGC_SIGHUP);
+		}
+
+		MemoryContextSwitchTo(oldctx);
+		MemoryContextReset(ApplyMessageContext);
+	}
+
+	MemoryContextSwitchTo(TopMemoryContext);
+	MemoryContextReset(ApplyContext);
+
+	/* Pop the error context stack */
+	error_context_stack = errcallback.previous;
+
+	elog(DEBUG1, "[Apply BGW #%u] exiting", shared->n);
+
+	/* Signal main process that we are done. */
+	SetLatch(&registrant->procLatch);
+}
+
+/*
+ * Set the exit status so that the main apply worker can realize we have
+ * shutdown.
+ */
+static void
+apply_bgworker_shutdown(int code, Datum arg)
+{
+	SpinLockAcquire(&MyParallelShared->mutex);
+	MyParallelShared->status = APPLY_BGWORKER_EXIT;
+	SpinLockRelease(&MyParallelShared->mutex);
+
+	dsm_detach((dsm_segment *) DatumGetPointer(arg));
+}
+
+/*
+ * Apply Background Worker entry point
+ */
+void
+ApplyBgworkerMain(Datum main_arg)
+{
+	volatile ApplyBgworkerShared *shared;
+
+	dsm_handle	handle;
+	dsm_segment *seg;
+	shm_toc    *toc;
+	shm_mq	   *mq;
+	shm_mq_handle *mqh;
+	MemoryContext oldcontext;
+	RepOriginId originid;
+	int			worker_slot = DatumGetInt32(main_arg);
+	char		originname[NAMEDATALEN];
+
+	MemoryContextSwitchTo(TopMemoryContext);
+
+	/* Init the memory context for the apply background worker to work in. */
+	ApplyContext = AllocSetContextCreate(TopMemoryContext,
+										 "ApplyContext",
+										 ALLOCSET_DEFAULT_SIZES);
+
+	/*
+	 * Init the ApplyMessageContext which we clean up after each replication
+	 * protocol message.
+	 */
+	ApplyMessageContext = AllocSetContextCreate(ApplyContext,
+												"ApplyMessageContext",
+												ALLOCSET_DEFAULT_SIZES);
+
+	/* Setup signal handling */
+	pqsignal(SIGHUP, SignalHandlerForConfigReload);
+	pqsignal(SIGTERM, die);
+	BackgroundWorkerUnblockSignals();
+
+	/*
+	 * Connect to the dynamic shared memory segment.
+	 *
+	 * The backend that registered this worker passed us the ID of a shared
+	 * memory segment to which we must attach for further instructions.  In
+	 * order to attach to dynamic shared memory, we need a resource owner.
+	 * Once we've mapped the segment in our address space, attach to the table
+	 * of contents so we can locate the various data structures we'll need to
+	 * find within the segment.
+	 */
+	CurrentResourceOwner = ResourceOwnerCreate(NULL, "Logical apply worker");
+	memcpy(&handle, MyBgworkerEntry->bgw_extra, sizeof(dsm_handle));
+	seg = dsm_attach(handle);
+	if (seg == NULL)
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("unable to map dynamic shared memory segment")));
+	toc = shm_toc_attach(PG_LOGICAL_APPLY_SHM_MAGIC, dsm_segment_address(seg));
+	if (toc == NULL)
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("bad magic number in dynamic shared memory segment")));
+
+	before_shmem_exit(apply_bgworker_shutdown, PointerGetDatum(seg));
+
+	/* Look up the shared information. */
+	shared = shm_toc_lookup(toc, APPLY_BGWORKER_KEY_SHARED, false);
+	MyParallelShared = shared;
+
+	/*
+	 * Attach to the message queue.
+	 */
+	mq = shm_toc_lookup(toc, APPLY_BGWORKER_KEY_MQ, false);
+	shm_mq_set_receiver(mq, MyProc);
+	mqh = shm_mq_attach(mq, seg, NULL);
+
+	/* Run as replica session replication role. */
+	SetConfigOption("session_replication_role", "replica",
+					PGC_SUSET, PGC_S_OVERRIDE);
+
+	/*
+	 * Now, we have initialized DSM. Attach to slot.
+	 */
+	logicalrep_worker_attach(worker_slot);
+
+	/* Connect to our database. */
+	BackgroundWorkerInitializeConnectionByOid(MyLogicalRepWorker->dbid,
+											  MyLogicalRepWorker->userid,
+											  0);
+
+	/*
+	 * Set always-secure search path, so malicious users can't redirect user
+	 * code (e.g. pg_index.indexprs).
+	 */
+	SetConfigOption("search_path", "", PGC_SUSET, PGC_S_OVERRIDE);
+
+	/*
+	 * Set the client encoding to the database encoding, since that is what
+	 * the leader will expect.
+	 */
+	SetClientEncoding(GetDatabaseEncoding());
+
+	stream_xid = shared->stream_xid;
+
+	StartTransactionCommand();
+	oldcontext = MemoryContextSwitchTo(ApplyContext);
+
+	MySubscription = GetSubscription(MyLogicalRepWorker->subid, true);
+	if (!MySubscription)
+	{
+		ereport(LOG,
+				(errmsg("logical replication apply worker for subscription %u will not "
+						"start because the subscription was removed during startup",
+						MyLogicalRepWorker->subid)));
+		proc_exit(0);
+	}
+
+	MySubscriptionValid = true;
+	MemoryContextSwitchTo(oldcontext);
+
+	/* Setup synchronous commit according to the user's wishes */
+	SetConfigOption("synchronous_commit", MySubscription->synccommit,
+					PGC_BACKEND, PGC_S_OVERRIDE);
+
+	/* Keep us informed about subscription changes. */
+	CacheRegisterSyscacheCallback(SUBSCRIPTIONOID,
+								  subscription_change_cb,
+								  (Datum) 0);
+
+	CommitTransactionCommand();
+
+	/* Setup replication origin tracking. */
+	StartTransactionCommand();
+	snprintf(originname, sizeof(originname), "pg_%u", MySubscription->oid);
+	originid = replorigin_by_name(originname, true);
+	if (!OidIsValid(originid))
+		originid = replorigin_create(originname);
+
+	/*
+	 * The apply background worker doesn't need to monopolize this replication
+	 * origin which was already acquired by its leader process.
+	 */
+	replorigin_session_setup(originid, false);
+	replorigin_session_origin = originid;
+	CommitTransactionCommand();
+
+	/*
+	 * Allocate the origin name in long-lived context for error context
+	 * message.
+	 */
+	apply_error_callback_arg.origin_name = MemoryContextStrdup(ApplyContext,
+															   originname);
+
+	elog(DEBUG1, "[Apply BGW #%u] started", shared->n);
+
+	LogicalApplyBgwLoop(mqh, shared);
+
+	/*
+	 * We're done.  Explicitly detach the shared memory segment so that we
+	 * don't get a resource leak warning at commit time.  This will fire any
+	 * on_dsm_detach callbacks we've registered, as well.  Once that's done,
+	 * we can go ahead and exit.
+	 */
+	dsm_detach(seg);
+	proc_exit(0);
+}
+
+/*
+ * Set up a dynamic shared memory segment.
+ *
+ * We set up a control region that contains a ApplyBgworkerShared,
+ * plus one region per message queue. There are as many message queues as
+ * the number of workers.
+ */
+static void
+apply_bgworker_setup_dsm(ApplyBgworkerState *wstate)
+{
+	shm_toc_estimator e;
+	Size		segsize;
+	dsm_segment *seg;
+	shm_toc    *toc;
+	ApplyBgworkerShared *shared;
+	shm_mq	   *mq;
+	int64		queue_size = DSM_QUEUE_SIZE;
+	int			server_version;
+
+	/*
+	 * Estimate how much shared memory we need.
+	 *
+	 * Because the TOC machinery may choose to insert padding of oddly-sized
+	 * requests, we must estimate each chunk separately.
+	 *
+	 * We need one key to register the location of the header, and we need
+	 * another key to track the location of the message queue.
+	 */
+	shm_toc_initialize_estimator(&e);
+	shm_toc_estimate_chunk(&e, sizeof(ApplyBgworkerShared));
+	shm_toc_estimate_chunk(&e, (Size) queue_size);
+
+	shm_toc_estimate_keys(&e, 1 + 1);
+	segsize = shm_toc_estimate(&e);
+
+	/* Create the shared memory segment and establish a table of contents. */
+	seg = dsm_create(shm_toc_estimate(&e), 0);
+	toc = shm_toc_create(PG_LOGICAL_APPLY_SHM_MAGIC, dsm_segment_address(seg),
+						 segsize);
+
+	/* Set up the header region. */
+	shared = shm_toc_allocate(toc, sizeof(ApplyBgworkerShared));
+	SpinLockInit(&shared->mutex);
+	shared->status = APPLY_BGWORKER_BUSY;
+
+	server_version = walrcv_server_version(LogRepWorkerWalRcvConn);
+	shared->server_version =
+		server_version >= 160000 ? LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM :
+		server_version >= 150000 ? LOGICALREP_PROTO_TWOPHASE_VERSION_NUM :
+		server_version >= 140000 ? LOGICALREP_PROTO_STREAM_VERSION_NUM :
+		LOGICALREP_PROTO_VERSION_NUM;
+
+	shared->stream_xid = stream_xid;
+	shared->n = list_length(ApplyWorkersList) + 1;
+
+	shm_toc_insert(toc, APPLY_BGWORKER_KEY_SHARED, shared);
+
+	/* Set up message queue for the worker. */
+	mq = shm_mq_create(shm_toc_allocate(toc, (Size) queue_size),
+					   (Size) queue_size);
+	shm_toc_insert(toc, APPLY_BGWORKER_KEY_MQ, mq);
+	shm_mq_set_sender(mq, MyProc);
+
+	/* Attach the queue. */
+	wstate->mq_handle = shm_mq_attach(mq, seg, NULL);
+
+	/* Return results to caller. */
+	wstate->dsm_seg = seg;
+	wstate->shared = shared;
+}
+
+/*
+ * Start apply background worker process and allocate shared memory for it.
+ */
+static ApplyBgworkerState *
+apply_bgworker_setup(void)
+{
+	MemoryContext oldcontext;
+	bool		launched;
+	ApplyBgworkerState *wstate;
+	int			napplyworkers;
+
+	elog(DEBUG1, "setting up apply worker #%u", list_length(ApplyWorkersList) + 1);
+
+	/* Check if there are free worker slot(s) */
+	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+	napplyworkers = logicalrep_apply_bgworker_count(MyLogicalRepWorker->subid);
+	LWLockRelease(LogicalRepWorkerLock);
+	if (napplyworkers >= max_apply_bgworkers_per_subscription)
+		return NULL;
+
+	oldcontext = MemoryContextSwitchTo(ApplyContext);
+
+	wstate = (ApplyBgworkerState *) palloc0(sizeof(ApplyBgworkerState));
+
+	/* Setup shared memory */
+	apply_bgworker_setup_dsm(wstate);
+
+	launched = logicalrep_worker_launch(MyLogicalRepWorker->dbid,
+										MySubscription->oid,
+										MySubscription->name,
+										MyLogicalRepWorker->userid,
+										InvalidOid,
+										dsm_segment_handle(wstate->dsm_seg));
+
+	if (launched)
+		ApplyWorkersList = lappend(ApplyWorkersList, wstate);
+	else
+	{
+		dsm_detach(wstate->dsm_seg);
+		wstate->dsm_seg = NULL;
+
+		pfree(wstate);
+		wstate = NULL;
+	}
+
+	MemoryContextSwitchTo(oldcontext);
+
+	return wstate;
+}
+
+/*
+ * Send the data to the specified apply background worker via shared-memory queue.
+ */
+void
+apply_bgworker_send_data(ApplyBgworkerState *wstate, Size nbytes, const void *data)
+{
+	shm_mq_result result;
+
+	result = shm_mq_send(wstate->mq_handle, nbytes, data, false, true);
+
+	if (result != SHM_MQ_SUCCESS)
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("could not send tuples to shared-memory queue")));
+}
+
+/*
+ * Wait until the status of apply background worker reaches the
+ * 'wait_for_status'
+ */
+void
+apply_bgworker_wait_for(ApplyBgworkerState *wstate,
+						ApplyBgworkerStatus wait_for_status)
+{
+	for (;;)
+	{
+		char		status;
+
+		SpinLockAcquire(&wstate->shared->mutex);
+		status = wstate->shared->status;
+		SpinLockRelease(&wstate->shared->mutex);
+
+		/* Done if already in correct status. */
+		if (status == wait_for_status)
+			break;
+
+		/* If any workers (or the postmaster) have died, we have failed. */
+		if (status == APPLY_BGWORKER_EXIT)
+			ereport(ERROR,
+					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					 errmsg("background worker %u failed to apply transaction %u",
+							wstate->shared->n, wstate->shared->stream_xid)));
+
+		/* Wait to be signalled. */
+		WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
+				  WAIT_EVENT_LOGICAL_APPLY_WORKER_STATE_CHANGE);
+
+		/* Reset the latch so we don't spin. */
+		ResetLatch(MyLatch);
+
+		/* An interrupt may have occurred while we were waiting. */
+		CHECK_FOR_INTERRUPTS();
+	}
+}
+
+/*
+ * Check the status of workers and report an error if any apply background
+ * worker has exited unexpectedly.
+ */
+void
+apply_bgworker_check_status(void)
+{
+	ListCell   *lc;
+
+	if (am_apply_bgworker() || MySubscription->stream != SUBSTREAM_PARALLEL)
+		return;
+
+	foreach(lc, ApplyWorkersList)
+	{
+		ApplyBgworkerState *wstate = (ApplyBgworkerState *) lfirst(lc);
+
+		/*
+		 * We don't lock here as in the worst case we will just detect the
+		 * failure of worker a bit later.
+		 */
+		if (wstate->shared->status == APPLY_BGWORKER_EXIT)
+			ereport(ERROR,
+					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					 errmsg("background worker %u exited unexpectedly",
+							wstate->shared->n)));
+	}
+
+	/*
+	 * Exit if any relation is not in the READY state and if any worker is
+	 * handling the streaming transaction at the same time. Because for
+	 * streaming transactions that is being applied in apply background
+	 * worker, we cannot decide whether to apply the change for a relation
+	 * that is not in the READY state (see should_apply_changes_for_rel) as we
+	 * won't know remote_final_lsn by that time.
+	 */
+	if (list_length(ApplyWorkersFreeList) != list_length(ApplyWorkersList) &&
+		!AllTablesyncsReady())
+	{
+		ereport(LOG,
+				(errmsg("logical replication apply workers for subscription \"%s\" will restart",
+						MySubscription->name),
+				 errdetail("Cannot handle streamed replication transaction by apply "
+						   "background workers until all tables are synchronized")));
+
+		proc_exit(0);
+	}
+}
+
+/* Set the apply background worker status */
+void
+apply_bgworker_set_status(ApplyBgworkerStatus status)
+{
+	if (!am_apply_bgworker())
+		return;
+
+	elog(DEBUG1, "[Apply BGW #%u] set status to %d", MyParallelShared->n, status);
+
+	SpinLockAcquire(&MyParallelShared->mutex);
+	MyParallelShared->status = status;
+	SpinLockRelease(&MyParallelShared->mutex);
+}
+
+/*
+ * Define a savepoint for a subxact in apply background worker if needed.
+ *
+ * Inside apply background worker we can figure out that new subtransaction was
+ * started if new change arrived with different xid. In that case we can define
+ * named savepoint, so that we were able to commit/rollback it separately
+ * later.
+ * Special case is if the first change comes from subtransaction, then
+ * we check that current_xid differs from stream_xid.
+ */
+void
+apply_bgworker_subxact_info_add(TransactionId current_xid)
+{
+	if (current_xid != stream_xid &&
+		!list_member_int(subxactlist, (int) current_xid))
+	{
+		MemoryContext oldctx;
+		char		spname[MAXPGPATH];
+
+		snprintf(spname, MAXPGPATH, "savepoint_for_xid_%u", current_xid);
+
+		elog(DEBUG1, "[Apply BGW #%u] defining savepoint %s",
+			 MyParallelShared->n, spname);
+
+		DefineSavepoint(spname);
+		CommitTransactionCommand();
+
+		oldctx = MemoryContextSwitchTo(ApplyContext);
+		subxactlist = lappend_int(subxactlist, (int) current_xid);
+		MemoryContextSwitchTo(oldctx);
+	}
+}
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index c5c6a2ba68..d4d5093a0b 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -651,9 +651,10 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 	{
 		for (i = 0; i < parsed->nsubxacts; i++)
 		{
-			ReorderBufferForget(ctx->reorder, parsed->subxacts[i], buf->origptr);
+			ReorderBufferForget(ctx->reorder, parsed->subxacts[i], buf->origptr,
+								commit_time);
 		}
-		ReorderBufferForget(ctx->reorder, xid, buf->origptr);
+		ReorderBufferForget(ctx->reorder, xid, buf->origptr, commit_time);
 
 		return;
 	}
@@ -821,10 +822,11 @@ DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 		for (i = 0; i < parsed->nsubxacts; i++)
 		{
 			ReorderBufferAbort(ctx->reorder, parsed->subxacts[i],
-							   buf->record->EndRecPtr);
+							   buf->record->EndRecPtr, abort_time);
 		}
 
-		ReorderBufferAbort(ctx->reorder, xid, buf->record->EndRecPtr);
+		ReorderBufferAbort(ctx->reorder, xid, buf->record->EndRecPtr,
+						   abort_time);
 	}
 
 	/* update the decoding stats */
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 3bbd522724..d92bfaf6d6 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -54,6 +54,7 @@
 
 int			max_logical_replication_workers = 4;
 int			max_sync_workers_per_subscription = 2;
+int			max_apply_bgworkers_per_subscription = 2;
 
 LogicalRepWorker *MyLogicalRepWorker = NULL;
 
@@ -73,6 +74,7 @@ static void logicalrep_launcher_onexit(int code, Datum arg);
 static void logicalrep_worker_onexit(int code, Datum arg);
 static void logicalrep_worker_detach(void);
 static void logicalrep_worker_cleanup(LogicalRepWorker *worker);
+static void logicalrep_worker_stop_internal(LogicalRepWorker *worker);
 
 static bool on_commit_launcher_wakeup = false;
 
@@ -151,8 +153,10 @@ get_subscription_list(void)
  *
  * This is only needed for cleaning up the shared memory in case the worker
  * fails to attach.
+ *
+ * Return false if the attach fails. Otherwise return true.
  */
-static void
+static bool
 WaitForReplicationWorkerAttach(LogicalRepWorker *worker,
 							   uint16 generation,
 							   BackgroundWorkerHandle *handle)
@@ -168,11 +172,11 @@ WaitForReplicationWorkerAttach(LogicalRepWorker *worker,
 
 		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
-		/* Worker either died or has started; no need to do anything. */
+		/* Worker either died or has started. Return false if died. */
 		if (!worker->in_use || worker->proc)
 		{
 			LWLockRelease(LogicalRepWorkerLock);
-			return;
+			return worker->in_use;
 		}
 
 		LWLockRelease(LogicalRepWorkerLock);
@@ -187,7 +191,7 @@ WaitForReplicationWorkerAttach(LogicalRepWorker *worker,
 			if (generation == worker->generation)
 				logicalrep_worker_cleanup(worker);
 			LWLockRelease(LogicalRepWorkerLock);
-			return;
+			return false;
 		}
 
 		/*
@@ -223,6 +227,13 @@ logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
 	{
 		LogicalRepWorker *w = &LogicalRepCtx->workers[i];
 
+		/*
+		 * We are only interested in the main apply worker or table sync worker
+		 * here.
+		 */
+		if (w->subworker)
+			continue;
+
 		if (w->in_use && w->subid == subid && w->relid == relid &&
 			(!only_running || w->proc))
 		{
@@ -259,11 +270,11 @@ logicalrep_workers_find(Oid subid, bool only_running)
 }
 
 /*
- * Start new apply background worker, if possible.
+ * Start new background worker, if possible.
  */
-void
+bool
 logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid,
-						 Oid relid)
+						 Oid relid, dsm_handle subworker_dsm)
 {
 	BackgroundWorker bgw;
 	BackgroundWorkerHandle *bgw_handle;
@@ -273,6 +284,10 @@ logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid,
 	LogicalRepWorker *worker = NULL;
 	int			nsyncworkers;
 	TimestampTz now;
+	bool		is_subworker = (subworker_dsm != DSM_HANDLE_INVALID);
+
+	/* Sanity check : we don't support table sync in subworker. */
+	Assert(!(is_subworker && OidIsValid(relid)));
 
 	ereport(DEBUG1,
 			(errmsg_internal("starting logical replication worker for subscription \"%s\"",
@@ -350,7 +365,7 @@ retry:
 	if (OidIsValid(relid) && nsyncworkers >= max_sync_workers_per_subscription)
 	{
 		LWLockRelease(LogicalRepWorkerLock);
-		return;
+		return false;
 	}
 
 	/*
@@ -364,7 +379,7 @@ retry:
 				(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
 				 errmsg("out of logical replication worker slots"),
 				 errhint("You might need to increase max_logical_replication_workers.")));
-		return;
+		return false;
 	}
 
 	/* Prepare the worker slot. */
@@ -379,6 +394,7 @@ retry:
 	worker->relstate = SUBREL_STATE_UNKNOWN;
 	worker->relstate_lsn = InvalidXLogRecPtr;
 	worker->stream_fileset = NULL;
+	worker->subworker = is_subworker;
 	worker->last_lsn = InvalidXLogRecPtr;
 	TIMESTAMP_NOBEGIN(worker->last_send_time);
 	TIMESTAMP_NOBEGIN(worker->last_recv_time);
@@ -396,19 +412,31 @@ retry:
 		BGWORKER_BACKEND_DATABASE_CONNECTION;
 	bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
 	snprintf(bgw.bgw_library_name, BGW_MAXLEN, "postgres");
-	snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
+
+	if (is_subworker)
+		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyBgworkerMain");
+	else
+		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
+
 	if (OidIsValid(relid))
 		snprintf(bgw.bgw_name, BGW_MAXLEN,
 				 "logical replication worker for subscription %u sync %u", subid, relid);
+	else if (is_subworker)
+		snprintf(bgw.bgw_name, BGW_MAXLEN,
+				 "logical replication apply background worker for subscription %u", subid);
 	else
 		snprintf(bgw.bgw_name, BGW_MAXLEN,
 				 "logical replication worker for subscription %u", subid);
+
 	snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication worker");
 
 	bgw.bgw_restart_time = BGW_NEVER_RESTART;
 	bgw.bgw_notify_pid = MyProcPid;
 	bgw.bgw_main_arg = Int32GetDatum(slot);
 
+	if (is_subworker)
+		memcpy(bgw.bgw_extra, &subworker_dsm, sizeof(dsm_handle));
+
 	if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle))
 	{
 		/* Failed to start worker, so clean up the worker slot. */
@@ -421,11 +449,11 @@ retry:
 				(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
 				 errmsg("out of background worker slots"),
 				 errhint("You might need to increase max_worker_processes.")));
-		return;
+		return false;
 	}
 
 	/* Now wait until it attaches. */
-	WaitForReplicationWorkerAttach(worker, generation, bgw_handle);
+	return WaitForReplicationWorkerAttach(worker, generation, bgw_handle);
 }
 
 /*
@@ -436,18 +464,27 @@ void
 logicalrep_worker_stop(Oid subid, Oid relid)
 {
 	LogicalRepWorker *worker;
-	uint16		generation;
 
 	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
 	worker = logicalrep_worker_find(subid, relid, false);
 
-	/* No worker, nothing to do. */
-	if (!worker)
-	{
-		LWLockRelease(LogicalRepWorkerLock);
-		return;
-	}
+	if (worker)
+		logicalrep_worker_stop_internal(worker);
+
+	LWLockRelease(LogicalRepWorkerLock);
+}
+
+/*
+ * Workhorse for logicalrep_worker_stop() and logicalrep_worker_detach(). Stop
+ * the worker and wait for it to die.
+ */
+static void
+logicalrep_worker_stop_internal(LogicalRepWorker *worker)
+{
+	uint16		generation;
+
+	Assert(LWLockHeldByMe(LogicalRepWorkerLock));
 
 	/*
 	 * Remember which generation was our worker so we can check if what we see
@@ -485,10 +522,7 @@ logicalrep_worker_stop(Oid subid, Oid relid)
 		 * different, meaning that a different worker has taken the slot.
 		 */
 		if (!worker->in_use || worker->generation != generation)
-		{
-			LWLockRelease(LogicalRepWorkerLock);
 			return;
-		}
 
 		/* Worker has assigned proc, so it has started. */
 		if (worker->proc)
@@ -522,8 +556,6 @@ logicalrep_worker_stop(Oid subid, Oid relid)
 
 		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 	}
-
-	LWLockRelease(LogicalRepWorkerLock);
 }
 
 /*
@@ -599,6 +631,29 @@ logicalrep_worker_attach(int slot)
 static void
 logicalrep_worker_detach(void)
 {
+	/*
+	 * If we are the main apply worker, stop all the apply background workers
+	 * we started before.
+	 */
+	if (!MyLogicalRepWorker->subworker)
+	{
+		List	   *workers;
+		ListCell   *lc;
+
+		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+
+		workers = logicalrep_workers_find(MyLogicalRepWorker->subid, true);
+		foreach(lc, workers)
+		{
+			LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
+
+			if (w->subworker)
+				logicalrep_worker_stop_internal(w);
+		}
+
+		LWLockRelease(LogicalRepWorkerLock);
+	}
+
 	/* Block concurrent access. */
 	LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
 
@@ -621,6 +676,7 @@ logicalrep_worker_cleanup(LogicalRepWorker *worker)
 	worker->userid = InvalidOid;
 	worker->subid = InvalidOid;
 	worker->relid = InvalidOid;
+	worker->subworker = false;
 }
 
 /*
@@ -679,6 +735,30 @@ logicalrep_sync_worker_count(Oid subid)
 	return res;
 }
 
+/*
+ * Count the number of registered (not necessarily running) apply background
+ * workers for a subscription.
+ */
+int
+logicalrep_apply_bgworker_count(Oid subid)
+{
+	int			i;
+	int			res = 0;
+
+	Assert(LWLockHeldByMe(LogicalRepWorkerLock));
+
+	/* Search for attached worker for a given subscription id. */
+	for (i = 0; i < max_logical_replication_workers; i++)
+	{
+		LogicalRepWorker *w = &LogicalRepCtx->workers[i];
+
+		if (w->subid == subid && w->subworker)
+			res++;
+	}
+
+	return res;
+}
+
 /*
  * ApplyLauncherShmemSize
  *		Compute space needed for replication launcher shared memory
@@ -868,7 +948,7 @@ ApplyLauncherMain(Datum main_arg)
 					wait_time = wal_retrieve_retry_interval;
 
 					logicalrep_worker_launch(sub->dbid, sub->oid, sub->name,
-											 sub->owner, InvalidOid);
+											 sub->owner, InvalidOid, DSM_HANDLE_INVALID);
 				}
 			}
 
diff --git a/src/backend/replication/logical/origin.c b/src/backend/replication/logical/origin.c
index 21937ab2d3..50c567fb6e 100644
--- a/src/backend/replication/logical/origin.c
+++ b/src/backend/replication/logical/origin.c
@@ -1063,12 +1063,21 @@ ReplicationOriginExitCleanup(int code, Datum arg)
  * array doesn't have to be searched when calling
  * replorigin_session_advance().
  *
- * Obviously only one such cached origin can exist per process and the current
+ * Normally only one such cached origin can exist per process and the current
  * cached value can only be set again after the previous value is torn down
  * with replorigin_session_reset().
+ *
+ * However, if the function parameter 'must_acquire' is false, we allow the
+ * process to use the same slot already acquired by another process. It's safe
+ * because 1) The only caller (apply background workers) will maintain the
+ * commit order by allowing only one process to commit at a time, so no two
+ * workers will be operating on the same origin at the same time (see comments
+ * in logical/worker.c). 2) Even though we try to advance the session's origin
+ * concurrently, it's safe to do so as we change/advance the session_origin
+ * LSNs under replicate_state LWLock.
  */
 void
-replorigin_session_setup(RepOriginId node)
+replorigin_session_setup(RepOriginId node, bool must_acquire)
 {
 	static bool registered_cleanup;
 	int			i;
@@ -1110,7 +1119,7 @@ replorigin_session_setup(RepOriginId node)
 		if (curstate->roident != node)
 			continue;
 
-		else if (curstate->acquired_by != 0)
+		else if (curstate->acquired_by != 0 && must_acquire)
 		{
 			ereport(ERROR,
 					(errcode(ERRCODE_OBJECT_IN_USE),
@@ -1141,7 +1150,14 @@ replorigin_session_setup(RepOriginId node)
 
 	Assert(session_replication_state->roident != InvalidRepOriginId);
 
-	session_replication_state->acquired_by = MyProcPid;
+	if (must_acquire)
+		session_replication_state->acquired_by = MyProcPid;
+	else if (session_replication_state->acquired_by == 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
+				 errmsg("apply background worker could not find replication state slot for replication origin with OID %u",
+						node),
+				 errdetail("There is no replication state slot set by its main apply worker.")));
 
 	LWLockRelease(ReplicationOriginLock);
 
@@ -1321,7 +1337,7 @@ pg_replication_origin_session_setup(PG_FUNCTION_ARGS)
 
 	name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
 	origin = replorigin_by_name(name, false);
-	replorigin_session_setup(origin);
+	replorigin_session_setup(origin, true);
 
 	replorigin_session_origin = origin;
 
diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c
index ff8513e2d2..47bd811fb7 100644
--- a/src/backend/replication/logical/proto.c
+++ b/src/backend/replication/logical/proto.c
@@ -1163,31 +1163,56 @@ logicalrep_read_stream_commit(StringInfo in, LogicalRepCommitData *commit_data)
 /*
  * Write STREAM ABORT to the output stream. Note that xid and subxid will be
  * same for the top-level transaction abort.
+ *
+ * If write_abort_lsn is true, send the abort_lsn and abort_time fields,
+ * otherwise don't.
  */
 void
 logicalrep_write_stream_abort(StringInfo out, TransactionId xid,
-							  TransactionId subxid)
+							  ReorderBufferTXN *txn, XLogRecPtr abort_lsn,
+							  bool write_abort_lsn)
 {
 	pq_sendbyte(out, LOGICAL_REP_MSG_STREAM_ABORT);
 
-	Assert(TransactionIdIsValid(xid) && TransactionIdIsValid(subxid));
+	Assert(TransactionIdIsValid(xid) && TransactionIdIsValid(txn->xid));
 
 	/* transaction ID */
 	pq_sendint32(out, xid);
-	pq_sendint32(out, subxid);
+	pq_sendint32(out, txn->xid);
+
+	if (write_abort_lsn)
+	{
+		pq_sendint64(out, abort_lsn);
+		pq_sendint64(out, txn->xact_time.abort_time);
+	}
 }
 
 /*
  * Read STREAM ABORT from the output stream.
+ *
+ * If read_abort_lsn is true, try to read the abort_lsn and abort_time fields,
+ * otherwise don't.
  */
 void
-logicalrep_read_stream_abort(StringInfo in, TransactionId *xid,
-							 TransactionId *subxid)
+logicalrep_read_stream_abort(StringInfo in,
+							 LogicalRepStreamAbortData *abort_data,
+							 bool read_abort_lsn)
 {
-	Assert(xid && subxid);
+	Assert(abort_data);
 
-	*xid = pq_getmsgint(in, 4);
-	*subxid = pq_getmsgint(in, 4);
+	abort_data->xid = pq_getmsgint(in, 4);
+	abort_data->subxid = pq_getmsgint(in, 4);
+
+	if (read_abort_lsn)
+	{
+		abort_data->abort_lsn = pq_getmsgint64(in);
+		abort_data->abort_time = pq_getmsgint64(in);
+	}
+	else
+	{
+		abort_data->abort_lsn = InvalidXLogRecPtr;
+		abort_data->abort_time = 0;
+	}
 }
 
 /*
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 88a37fde72..8989328046 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -2826,7 +2826,8 @@ ReorderBufferFinishPrepared(ReorderBuffer *rb, TransactionId xid,
  * disk.
  */
 void
-ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
+ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn,
+				   TimestampTz abort_time)
 {
 	ReorderBufferTXN *txn;
 
@@ -2837,6 +2838,8 @@ ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
 	if (txn == NULL)
 		return;
 
+	txn->xact_time.abort_time = abort_time;
+
 	/* For streamed transactions notify the remote node about the abort. */
 	if (rbtxn_is_streamed(txn))
 	{
@@ -2911,7 +2914,8 @@ ReorderBufferAbortOld(ReorderBuffer *rb, TransactionId oldestRunningXid)
  * to this xid might re-create the transaction incompletely.
  */
 void
-ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
+ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn,
+					TimestampTz abort_time)
 {
 	ReorderBufferTXN *txn;
 
@@ -2922,6 +2926,8 @@ ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
 	if (txn == NULL)
 		return;
 
+	txn->xact_time.abort_time = abort_time;
+
 	/* For streamed transactions notify the remote node about the abort. */
 	if (rbtxn_is_streamed(txn))
 		rb->stream_abort(rb, txn, lsn);
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 670c6fcada..8ffba7e2e5 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -568,7 +568,8 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 												 MySubscription->oid,
 												 MySubscription->name,
 												 MyLogicalRepWorker->userid,
-												 rstate->relid);
+												 rstate->relid,
+												 DSM_HANDLE_INVALID);
 						hentry->last_start_time = now;
 					}
 				}
@@ -589,6 +590,9 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 void
 process_syncing_tables(XLogRecPtr current_lsn)
 {
+	if (MyLogicalRepWorker->subworker)
+		return;
+
 	if (am_tablesync_worker())
 		process_syncing_tables_for_sync(current_lsn);
 	else
@@ -1273,7 +1277,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
 		 * time this tablesync was launched.
 		 */
 		originid = replorigin_by_name(originname, false);
-		replorigin_session_setup(originid);
+		replorigin_session_setup(originid, true);
 		replorigin_session_origin = originid;
 		*origin_startpos = replorigin_session_get_progress(false);
 
@@ -1384,7 +1388,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
 						   true /* go backward */ , true /* WAL log */ );
 		UnlockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
 
-		replorigin_session_setup(originid);
+		replorigin_session_setup(originid, true);
 		replorigin_session_origin = originid;
 	}
 	else
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 38e3b1c1b3..b5aae0e19a 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -22,8 +22,28 @@
  * STREAMED TRANSACTIONS
  * ---------------------
  * Streamed transactions (large transactions exceeding a memory limit on the
- * upstream) are not applied immediately, but instead, the data is written
- * to temporary files and then applied at once when the final commit arrives.
+ * upstream) are applied using one of two approaches.
+ *
+ * 1) Separate background workers
+ *
+ * If streaming = parallel, we assign a new apply background worker (if
+ * available) as soon as the xact's first stream is received. The main apply
+ * worker will send changes to this new worker via shared memory. We keep this
+ * worker assigned till the transaction commit is received and also wait for
+ * the worker to finish at commit. This preserves commit ordering and avoids
+ * file I/O in most cases. We still need to spill to a file if there is no
+ * worker available. It is important to maintain commit order to avoid failures
+ * due to (a) transaction dependencies, say if we insert a row in the first
+ * transaction and update it in the second transaction then allowing to apply
+ * both in parallel can lead to failure in the update. (b) deadlocks, allowing
+ * transactions that update the same set of rows/tables in opposite order to be
+ * applied in parallel can lead to deadlocks.
+ *
+ * 2) Write to temporary files and apply when the final commit arrives
+ *
+ * If no worker is available to handle streamed transaction, the data is
+ * written to temporary files and then applied at once when the final commit
+ * arrives.
  *
  * Unlike the regular (non-streamed) case, handling streamed transactions has
  * to handle aborts of both the toplevel transaction and subtransactions. This
@@ -219,20 +239,8 @@ typedef struct ApplyExecutionData
 	PartitionTupleRouting *proute;	/* partition routing info */
 } ApplyExecutionData;
 
-/* Struct for saving and restoring apply errcontext information */
-typedef struct ApplyErrorCallbackArg
-{
-	LogicalRepMsgType command;	/* 0 if invalid */
-	LogicalRepRelMapEntry *rel;
-
-	/* Remote node information */
-	int			remote_attnum;	/* -1 if invalid */
-	TransactionId remote_xid;
-	XLogRecPtr	finish_lsn;
-	char	   *origin_name;
-} ApplyErrorCallbackArg;
-
-static ApplyErrorCallbackArg apply_error_callback_arg =
+/* errcontext tracker */
+ApplyErrorCallbackArg apply_error_callback_arg =
 {
 	.command = 0,
 	.rel = NULL,
@@ -242,7 +250,7 @@ static ApplyErrorCallbackArg apply_error_callback_arg =
 	.origin_name = NULL,
 };
 
-static MemoryContext ApplyMessageContext = NULL;
+MemoryContext ApplyMessageContext = NULL;
 MemoryContext ApplyContext = NULL;
 
 /* per stream context for streaming transactions */
@@ -251,27 +259,39 @@ static MemoryContext LogicalStreamingContext = NULL;
 WalReceiverConn *LogRepWorkerWalRcvConn = NULL;
 
 Subscription *MySubscription = NULL;
-static bool MySubscriptionValid = false;
+bool MySubscriptionValid = false;
 
 bool		in_remote_transaction = false;
 static XLogRecPtr remote_final_lsn = InvalidXLogRecPtr;
 
 /* fields valid only when processing streamed transaction */
-static bool in_streamed_transaction = false;
+bool in_streamed_transaction = false;
+
+TransactionId stream_xid = InvalidTransactionId;
+static ApplyBgworkerState *stream_apply_worker = NULL;
 
-static TransactionId stream_xid = InvalidTransactionId;
+/* Check if we are applying the transaction in an apply background worker */
+#define apply_bgworker_active() (in_streamed_transaction && stream_apply_worker != NULL)
+
+/*
+ * The number of changes during one streaming block (only for apply background
+ * workers)
+ */
+static uint32 nchanges = 0;
 
 /*
  * We enable skipping all data modification changes (INSERT, UPDATE, etc.) for
  * the subscription if the remote transaction's finish LSN matches the subskiplsn.
  * Once we start skipping changes, we don't stop it until we skip all changes of
  * the transaction even if pg_subscription is updated and MySubscription->skiplsn
- * gets changed or reset during that. Also, in streaming transaction cases, we
- * don't skip receiving and spooling the changes since we decide whether or not
+ * gets changed or reset during that. Also, in streaming transaction cases (streaming = on),
+ * we don't skip receiving and spooling the changes since we decide whether or not
  * to skip applying the changes when starting to apply changes. The subskiplsn is
  * cleared after successfully skipping the transaction or applying non-empty
  * transaction. The latter prevents the mistakenly specified subskiplsn from
- * being left.
+ * being left. Note that we cannot skip the streaming transactions when using
+ * apply background workers because we cannot get the finish LSN before
+ * applying the changes.
  */
 static XLogRecPtr skip_xact_finish_lsn = InvalidXLogRecPtr;
 #define is_skipping_changes() (unlikely(!XLogRecPtrIsInvalid(skip_xact_finish_lsn)))
@@ -324,9 +344,6 @@ static void maybe_reread_subscription(void);
 
 static void DisableSubscriptionAndExit(void);
 
-/* prototype needed because of stream_commit */
-static void apply_dispatch(StringInfo s);
-
 static void apply_handle_commit_internal(LogicalRepCommitData *commit_data);
 static void apply_handle_insert_internal(ApplyExecutionData *edata,
 										 ResultRelInfo *relinfo,
@@ -359,7 +376,6 @@ static void stop_skipping_changes(void);
 static void clear_subscription_skip_lsn(XLogRecPtr finish_lsn);
 
 /* Functions for apply error callback */
-static void apply_error_callback(void *arg);
 static inline void set_apply_error_context_xact(TransactionId xid, XLogRecPtr lsn);
 static inline void reset_apply_error_context_info(void);
 
@@ -426,40 +442,85 @@ end_replication_step(void)
 }
 
 /*
- * Handle streamed transactions.
+ * Handle streamed transactions for both the main apply worker and the apply
+ * background workers.
+ *
+ * In streaming case (receiving a block of streamed transaction), for
+ * SUBSTREAM_ON mode, simply redirect it to a file for the proper toplevel
+ * transaction, and for SUBSTREAM_PARALLEL mode, send the changes to apply
+ * background workers (LOGICAL_REP_MSG_RELATION or LOGICAL_REP_MSG_TYPE changes
+ * will also be applied in main apply worker).
  *
- * If in streaming mode (receiving a block of streamed transaction), we
- * simply redirect it to a file for the proper toplevel transaction.
+ * For non-streamed transactions, returns false;
+ * For streamed transactions, returns true if in main apply worker, false
+ * otherwise.
  *
- * Returns true for streamed transactions, false otherwise (regular mode).
+ * Exception: When the main apply worker is applying streaming transactions in
+ * parallel mode (e.g. when addressing LOGICAL_REP_MSG_RELATION or
+ * LOGICAL_REP_MSG_TYPE changes), then return false.
  */
 static bool
 handle_streamed_transaction(LogicalRepMsgType action, StringInfo s)
 {
-	TransactionId xid;
+	TransactionId current_xid = InvalidTransactionId;
 
-	/* not in streaming mode */
-	if (!in_streamed_transaction)
+	/* Not in streaming mode */
+	if (!(in_streamed_transaction || am_apply_bgworker()))
 		return false;
 
-	Assert(stream_fd != NULL);
 	Assert(TransactionIdIsValid(stream_xid));
 
 	/*
 	 * We should have received XID of the subxact as the first part of the
 	 * message, so extract it.
 	 */
-	xid = pq_getmsgint(s, 4);
+	current_xid = pq_getmsgint(s, 4);
 
-	if (!TransactionIdIsValid(xid))
+	if (!TransactionIdIsValid(current_xid))
 		ereport(ERROR,
 				(errcode(ERRCODE_PROTOCOL_VIOLATION),
 				 errmsg_internal("invalid transaction ID in streamed replication transaction")));
 
-	/* Add the new subxact to the array (unless already there). */
-	subxact_info_add(xid);
+	if (am_apply_bgworker())
+	{
+		/* Define a savepoint for a subxact if needed. */
+		apply_bgworker_subxact_info_add(current_xid);
+
+		return false;
+	}
+
+	if (apply_bgworker_active())
+	{
+		/*
+		 * This is the main apply worker, but there is an apply background
+		 * worker, so apply the changes of this transaction in that background
+		 * worker. Pass the data to the worker.
+		 */
+		apply_bgworker_send_data(stream_apply_worker, s->len, s->data);
+		nchanges += 1;
+
+		/*
+		 * XXX The publisher side doesn't always send relation/type update
+		 * messages after the streaming transaction, so also update the
+		 * relation/type in main apply worker here. See function
+		 * cleanup_rel_sync_cache.
+		 */
+		if (action == LOGICAL_REP_MSG_RELATION ||
+			action == LOGICAL_REP_MSG_TYPE)
+			return false;
+
+		return true;
+	}
+
+	/*
+	 * This is the main apply worker, but there is no apply background worker,
+	 * so write to temporary files and apply when the final commit arrives.
+	 *
+	 * Add the new subxact to the array (unless already there).
+	 */
+	subxact_info_add(current_xid);
 
-	/* write the change to the current file */
+	/* Write the change to the current file */
 	stream_write_change(action, s);
 
 	return true;
@@ -844,6 +905,9 @@ apply_handle_commit(StringInfo s)
 
 	apply_handle_commit_internal(&commit_data);
 
+	/* Check the status of apply background worker if any. */
+	apply_bgworker_check_status();
+
 	/* Process any tables that are being synchronized in parallel. */
 	process_syncing_tables(commit_data.end_lsn);
 
@@ -898,7 +962,9 @@ apply_handle_prepare_internal(LogicalRepPreparedTxnData *prepare_data)
 	 * BeginTransactionBlock is necessary to balance the EndTransactionBlock
 	 * called within the PrepareTransactionBlock below.
 	 */
-	BeginTransactionBlock();
+	if (!IsTransactionBlock())
+		BeginTransactionBlock();
+
 	CommitTransactionCommand(); /* Completes the preceding Begin command. */
 
 	/*
@@ -950,6 +1016,9 @@ apply_handle_prepare(StringInfo s)
 
 	in_remote_transaction = false;
 
+	/* Check the status of apply background worker if any. */
+	apply_bgworker_check_status();
+
 	/* Process any tables that are being synchronized in parallel. */
 	process_syncing_tables(prepare_data.end_lsn);
 
@@ -1064,10 +1133,6 @@ apply_handle_rollback_prepared(StringInfo s)
 
 /*
  * Handle STREAM PREPARE.
- *
- * Logic is in two parts:
- * 1. Replay all the spooled operations
- * 2. Mark the transaction as prepared
  */
 static void
 apply_handle_stream_prepare(StringInfo s)
@@ -1088,24 +1153,78 @@ apply_handle_stream_prepare(StringInfo s)
 	logicalrep_read_stream_prepare(s, &prepare_data);
 	set_apply_error_context_xact(prepare_data.xid, prepare_data.prepare_lsn);
 
-	elog(DEBUG1, "received prepare for streamed transaction %u", prepare_data.xid);
+	if (am_apply_bgworker())
+	{
+		/* Mark the transaction as prepared. */
+		apply_handle_prepare_internal(&prepare_data);
 
-	/* Replay all the spooled operations. */
-	apply_spooled_messages(prepare_data.xid, prepare_data.prepare_lsn);
+		CommitTransactionCommand();
 
-	/* Mark the transaction as prepared. */
-	apply_handle_prepare_internal(&prepare_data);
+		pgstat_report_stat(false);
 
-	CommitTransactionCommand();
+		list_free(subxactlist);
+		subxactlist = NIL;
 
-	pgstat_report_stat(false);
+		apply_bgworker_set_status(APPLY_BGWORKER_FINISHED);
+	}
+	else
+	{
+		/* This is the main apply worker. */
+		ApplyBgworkerState *wstate = apply_bgworker_find(prepare_data.xid);
 
-	store_flush_position(prepare_data.end_lsn);
+		elog(DEBUG1, "received prepare for streamed transaction %u",
+			 prepare_data.xid);
+
+		/*
+		 * Check if we are processing this transaction in an apply background
+		 * worker and if so, send the changes to that worker.
+		 */
+		if (wstate)
+		{
+			/* Send STREAM PREPARE message to the apply background worker. */
+			apply_bgworker_send_data(wstate, s->len, s->data);
+
+			/*
+			 * After sending the data to the apply background worker, wait for
+			 * that worker to finish. This is necessary to maintain commit
+			 * order which avoids failures due to transaction dependencies and
+			 * deadlocks.
+			 */
+			apply_bgworker_wait_for(wstate, APPLY_BGWORKER_FINISHED);
+			apply_bgworker_free(wstate);
+
+			pgstat_report_stat(false);
+			store_flush_position(prepare_data.end_lsn);
+		}
+		else
+		{
+			/*
+			 * The transaction has been serialized to file, so replay all the
+			 * spooled operations.
+			 */
+			apply_spooled_messages(prepare_data.xid, prepare_data.prepare_lsn);
+
+			/* Mark the transaction as prepared. */
+			apply_handle_prepare_internal(&prepare_data);
+
+			CommitTransactionCommand();
+
+			pgstat_report_stat(false);
+
+			store_flush_position(prepare_data.end_lsn);
+
+			in_remote_transaction = false;
+
+			/* Unlink the files with serialized changes and subxact info. */
+			stream_cleanup_files(MyLogicalRepWorker->subid, prepare_data.xid);
+		}
+	}
 
 	in_remote_transaction = false;
+	stream_apply_worker = NULL;
 
-	/* unlink the files with serialized changes and subxact info. */
-	stream_cleanup_files(MyLogicalRepWorker->subid, prepare_data.xid);
+	/* Check the status of apply background worker if any. */
+	apply_bgworker_check_status();
 
 	/* Process any tables that are being synchronized in parallel. */
 	process_syncing_tables(prepare_data.end_lsn);
@@ -1155,15 +1274,6 @@ apply_handle_stream_start(StringInfo s)
 				(errcode(ERRCODE_PROTOCOL_VIOLATION),
 				 errmsg_internal("duplicate STREAM START message")));
 
-	/*
-	 * Start a transaction on stream start, this transaction will be committed
-	 * on the stream stop unless it is a tablesync worker in which case it
-	 * will be committed after processing all the messages. We need the
-	 * transaction for handling the buffile, used for serializing the
-	 * streaming data and subxact info.
-	 */
-	begin_replication_step();
-
 	/* notify handle methods we're processing a remote transaction */
 	in_streamed_transaction = true;
 
@@ -1177,36 +1287,93 @@ apply_handle_stream_start(StringInfo s)
 
 	set_apply_error_context_xact(stream_xid, InvalidXLogRecPtr);
 
-	/*
-	 * Initialize the worker's stream_fileset if we haven't yet. This will be
-	 * used for the entire duration of the worker so create it in a permanent
-	 * context. We create this on the very first streaming message from any
-	 * transaction and then use it for this and other streaming transactions.
-	 * Now, we could create a fileset at the start of the worker as well but
-	 * then we won't be sure that it will ever be used.
-	 */
-	if (MyLogicalRepWorker->stream_fileset == NULL)
+	if (am_apply_bgworker())
 	{
-		MemoryContext oldctx;
-
-		oldctx = MemoryContextSwitchTo(ApplyContext);
+		/*
+		 * Make sure the handle apply_dispatch methods are aware we're in a
+		 * remote transaction.
+		 */
+		in_remote_transaction = true;
 
-		MyLogicalRepWorker->stream_fileset = palloc(sizeof(FileSet));
-		FileSetInit(MyLogicalRepWorker->stream_fileset);
+		/* Begin the transaction. */
+		AcceptInvalidationMessages();
+		maybe_reread_subscription();
 
-		MemoryContextSwitchTo(oldctx);
+		StartTransactionCommand();
+		BeginTransactionBlock();
+		CommitTransactionCommand();
 	}
+	else
+	{
+		/*
+		 * This is the main apply worker. Check if there is any free apply
+		 * background worker we can use to process this transaction.
+		 */
+		if (first_segment)
+			stream_apply_worker = apply_bgworker_start(stream_xid);
+		else
+			stream_apply_worker = apply_bgworker_find(stream_xid);
 
-	/* open the spool file for this transaction */
-	stream_open_file(MyLogicalRepWorker->subid, stream_xid, first_segment);
+		if (stream_apply_worker)
+		{
+			/*
+			 * If we have found a free worker or if we are already applying this
+			 * transaction in an apply background worker, then we pass the data to
+			 * that worker.
+			 */
+			if (first_segment)
+				apply_bgworker_send_data(stream_apply_worker, s->len, s->data);
 
-	/* if this is not the first segment, open existing subxact file */
-	if (!first_segment)
-		subxact_info_read(MyLogicalRepWorker->subid, stream_xid);
+			nchanges = 0;
+			elog(DEBUG1, "starting streaming of xid %u", stream_xid);
+		}
+		else
+		{
+			/*
+			 * Since no apply background worker is available for the first
+			 * stream start, serialize all the changes of the transaction.
+			 *
+			 * Start a transaction on stream start, this transaction will be
+			 * committed on the stream stop unless it is a tablesync worker in
+			 * which case it will be committed after processing all the
+			 * messages. We need the transaction for handling the buffile,
+			 * used for serializing the streaming data and subxact info.
+			 */
+			begin_replication_step();
 
-	pgstat_report_activity(STATE_RUNNING, NULL);
+			/*
+			 * Initialize the worker's stream_fileset if we haven't yet. This will
+			 * be used for the entire duration of the worker so create it in a
+			 * permanent context. We create this on the very first streaming
+			 * message from any transaction and then use it for this and other
+			 * streaming transactions. Now, we could create a fileset at the start
+			 * of the worker as well but then we won't be sure that it will ever
+			 * be used.
+			 */
+			if (MyLogicalRepWorker->stream_fileset == NULL)
+			{
+				MemoryContext oldctx;
 
-	end_replication_step();
+				oldctx = MemoryContextSwitchTo(ApplyContext);
+
+				MyLogicalRepWorker->stream_fileset = palloc(sizeof(FileSet));
+				FileSetInit(MyLogicalRepWorker->stream_fileset);
+
+				MemoryContextSwitchTo(oldctx);
+			}
+
+			/* Open the spool file for this transaction. */
+			stream_open_file(MyLogicalRepWorker->subid, stream_xid, first_segment);
+
+			/* If this is not the first segment, open existing subxact file. */
+			if (!first_segment)
+				subxact_info_read(MyLogicalRepWorker->subid, stream_xid);
+
+			end_replication_step();
+		}
+	}
+
+	pgstat_report_activity(STATE_RUNNING, NULL);
 }
 
 /*
@@ -1220,53 +1387,52 @@ apply_handle_stream_stop(StringInfo s)
 				(errcode(ERRCODE_PROTOCOL_VIOLATION),
 				 errmsg_internal("STREAM STOP message without STREAM START")));
 
-	/*
-	 * Close the file with serialized changes, and serialize information about
-	 * subxacts for the toplevel transaction.
-	 */
-	subxact_info_write(MyLogicalRepWorker->subid, stream_xid);
-	stream_close_file();
+	if (apply_bgworker_active())
+	{
+		char action = LOGICAL_REP_MSG_STREAM_STOP;
 
-	/* We must be in a valid transaction state */
-	Assert(IsTransactionState());
+		apply_bgworker_send_data(stream_apply_worker, 1, &action);
 
-	/* Commit the per-stream transaction */
-	CommitTransactionCommand();
+		elog(DEBUG1, "stopped streaming of xid %u, %u changes streamed", stream_xid, nchanges);
+	}
+	else
+	{
+		/*
+		 * Close the file with serialized changes, and serialize information
+		 * about subxacts for the toplevel transaction.
+		 */
+		subxact_info_write(MyLogicalRepWorker->subid, stream_xid);
+		stream_close_file();
 
-	in_streamed_transaction = false;
+		/* We must be in a valid transaction state */
+		Assert(IsTransactionState());
 
-	/* Reset per-stream context */
-	MemoryContextReset(LogicalStreamingContext);
+		/* Commit the per-stream transaction */
+		CommitTransactionCommand();
+
+		/* Reset per-stream context */
+		MemoryContextReset(LogicalStreamingContext);
+	}
+
+	in_streamed_transaction = false;
+	stream_apply_worker = NULL;
 
 	pgstat_report_activity(STATE_IDLE, NULL);
 	reset_apply_error_context_info();
 }
 
 /*
- * Handle STREAM abort message.
+ * Handle STREAM ABORT message when the transaction was spilled to disk.
  */
 static void
-apply_handle_stream_abort(StringInfo s)
+serialize_stream_abort(TransactionId xid, TransactionId subxid)
 {
-	TransactionId xid;
-	TransactionId subxid;
-
-	if (in_streamed_transaction)
-		ereport(ERROR,
-				(errcode(ERRCODE_PROTOCOL_VIOLATION),
-				 errmsg_internal("STREAM ABORT message without STREAM STOP")));
-
-	logicalrep_read_stream_abort(s, &xid, &subxid);
-
 	/*
 	 * If the two XIDs are the same, it's in fact abort of toplevel xact, so
 	 * just delete the files with serialized info.
 	 */
 	if (xid == subxid)
-	{
-		set_apply_error_context_xact(xid, InvalidXLogRecPtr);
 		stream_cleanup_files(MyLogicalRepWorker->subid, xid);
-	}
 	else
 	{
 		/*
@@ -1290,8 +1456,6 @@ apply_handle_stream_abort(StringInfo s)
 		bool		found = false;
 		char		path[MAXPGPATH];
 
-		set_apply_error_context_xact(subxid, InvalidXLogRecPtr);
-
 		subidx = -1;
 		begin_replication_step();
 		subxact_info_read(MyLogicalRepWorker->subid, xid);
@@ -1316,7 +1480,6 @@ apply_handle_stream_abort(StringInfo s)
 			cleanup_subxact_info();
 			end_replication_step();
 			CommitTransactionCommand();
-			reset_apply_error_context_info();
 			return;
 		}
 
@@ -1339,6 +1502,143 @@ apply_handle_stream_abort(StringInfo s)
 		end_replication_step();
 		CommitTransactionCommand();
 	}
+}
+
+/*
+ * Handle STREAM ABORT message.
+ */
+static void
+apply_handle_stream_abort(StringInfo s)
+{
+	TransactionId xid;
+	TransactionId subxid;
+	LogicalRepStreamAbortData abort_data;
+	bool read_abort_lsn = false;
+
+	if (in_streamed_transaction)
+		ereport(ERROR,
+				(errcode(ERRCODE_PROTOCOL_VIOLATION),
+				 errmsg_internal("STREAM ABORT message without STREAM STOP")));
+
+	/* Check whether the publisher sends abort_lsn and abort_time. */
+	if (am_apply_bgworker())
+		read_abort_lsn = MyParallelShared->server_version >=
+						 LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM;
+
+	logicalrep_read_stream_abort(s, &abort_data, read_abort_lsn);
+
+	xid = abort_data.xid;
+	subxid = abort_data.subxid;
+
+	set_apply_error_context_xact(subxid, abort_data.abort_lsn);
+
+	if (am_apply_bgworker())
+	{
+		elog(DEBUG1, "[Apply BGW #%u] aborting current transaction xid=%u, subxid=%u",
+			 MyParallelShared->n, GetCurrentTransactionIdIfAny(),
+			 GetCurrentSubTransactionId());
+
+		/*
+		 * Update origin state so we can restart streaming from correct
+		 * position in case of crash.
+		 */
+		if (read_abort_lsn)
+		{
+			replorigin_session_origin_lsn = abort_data.abort_lsn;
+			replorigin_session_origin_timestamp = abort_data.abort_time;
+		}
+
+		/*
+		 * If the two XIDs are the same, it's in fact abort of toplevel xact,
+		 * so just free the subxactlist.
+		 */
+		if (subxid == xid)
+		{
+			AbortCurrentTransaction();
+
+			EndTransactionBlock(false);
+			CommitTransactionCommand();
+
+			in_remote_transaction = false;
+			pgstat_report_activity(STATE_IDLE, NULL);
+
+			list_free(subxactlist);
+			subxactlist = NIL;
+
+			apply_bgworker_set_status(APPLY_BGWORKER_FINISHED);
+		}
+		else
+		{
+			/*
+			 * OK, so it's a subxact. Rollback to the savepoint.
+			 *
+			 * We also need to read the subxactlist, determine the offset
+			 * tracked for the subxact, and truncate the list.
+			 */
+			int			i;
+			bool		found = false;
+			char		spname[MAXPGPATH];
+
+			snprintf(spname, MAXPGPATH, "savepoint_for_xid_%u", subxid);
+
+			elog(DEBUG1, "[Apply BGW #%u] rolling back to savepoint %s",
+				 MyParallelShared->n, spname);
+
+			for (i = list_length(subxactlist) - 1; i >= 0; i--)
+			{
+				xid = (TransactionId) list_nth_int(subxactlist, i);
+				if (xid == subxid)
+				{
+					found = true;
+					break;
+				}
+			}
+
+			if (found)
+			{
+				RollbackToSavepoint(spname);
+				CommitTransactionCommand();
+				subxactlist = list_truncate(subxactlist, i + 1);
+			}
+
+			pgstat_report_activity(STATE_IDLEINTRANSACTION, NULL);
+		}
+	}
+	else
+	{
+		/* This is the main apply worker. */
+		ApplyBgworkerState *wstate = apply_bgworker_find(xid);
+
+		/*
+		 * Check if we are processing this transaction in an apply background
+		 * worker and if so, send the changes to that worker.
+		 */
+		if (wstate)
+		{
+			/* Send STREAM ABORT message to the apply background worker. */
+			apply_bgworker_send_data(wstate, s->len, s->data);
+
+			/*
+			 * After sending the data to the apply background worker, wait for
+			 * that worker to finish. This is necessary to maintain commit
+			 * order which avoids failures due to transaction dependencies and
+			 * deadlocks.
+			 */
+			if (subxid == xid)
+			{
+				apply_bgworker_wait_for(wstate, APPLY_BGWORKER_FINISHED);
+				apply_bgworker_free(wstate);
+			}
+		}
+		else
+		{
+			/*
+			 * We are in main apply worker and the transaction has been
+			 * serialized to file.
+			 */
+			serialize_stream_abort(xid, subxid);
+		}
+	}
 
 	reset_apply_error_context_info();
 }
@@ -1468,8 +1768,8 @@ apply_spooled_messages(TransactionId xid, XLogRecPtr lsn)
 static void
 apply_handle_stream_commit(StringInfo s)
 {
-	TransactionId xid;
 	LogicalRepCommitData commit_data;
+	TransactionId xid;
 
 	if (in_streamed_transaction)
 		ereport(ERROR,
@@ -1479,14 +1779,81 @@ apply_handle_stream_commit(StringInfo s)
 	xid = logicalrep_read_stream_commit(s, &commit_data);
 	set_apply_error_context_xact(xid, commit_data.commit_lsn);
 
-	elog(DEBUG1, "received commit for streamed transaction %u", xid);
+	if (am_apply_bgworker())
+	{
+		/*
+		 * Update origin state so we can restart streaming from correct
+		 * position in case of crash.
+		 */
+		replorigin_session_origin_lsn = commit_data.end_lsn;
+		replorigin_session_origin_timestamp = commit_data.committime;
 
-	apply_spooled_messages(xid, commit_data.commit_lsn);
+		CommitTransactionCommand();
+		EndTransactionBlock(false);
+		CommitTransactionCommand();
 
-	apply_handle_commit_internal(&commit_data);
+		in_remote_transaction = false;
+
+		pgstat_report_stat(false);
+
+		list_free(subxactlist);
+		subxactlist = NIL;
+
+		apply_bgworker_set_status(APPLY_BGWORKER_FINISHED);
+	}
+	else
+	{
+		/* This is the main apply worker. */
+		ApplyBgworkerState *wstate = apply_bgworker_find(xid);
+
+		elog(DEBUG1, "received commit for streamed transaction %u", xid);
+
+		/*
+		 * Check if we are processing this transaction in an apply background
+		 * worker and if so, send the changes to that worker.
+		 */
+		if (wstate)
+		{
+			/* Send STREAM COMMIT message to the apply background worker. */
+			apply_bgworker_send_data(wstate, s->len, s->data);
+
+			/*
+			 * After sending the data to the apply background worker, wait for
+			 * that worker to finish. This is necessary to maintain commit
+			 * order which avoids failures due to transaction dependencies and
+			 * deadlocks.
+			 */
+			apply_bgworker_wait_for(wstate, APPLY_BGWORKER_FINISHED);
 
-	/* unlink the files with serialized changes and subxact info */
-	stream_cleanup_files(MyLogicalRepWorker->subid, xid);
+			pgstat_report_stat(false);
+			store_flush_position(commit_data.end_lsn);
+			stop_skipping_changes();
+
+			apply_bgworker_free(wstate);
+
+			/*
+			 * The transaction is either non-empty or skipped, so we clear the
+			 * subskiplsn.
+			 */
+			clear_subscription_skip_lsn(commit_data.commit_lsn);
+		}
+		else
+		{
+			/*
+			 * The transaction has been serialized to file, so replay all the
+			 * spooled operations.
+			 */
+			apply_spooled_messages(xid, commit_data.commit_lsn);
+
+			apply_handle_commit_internal(&commit_data);
+
+			/* Unlink the files with serialized changes and subxact info. */
+			stream_cleanup_files(MyLogicalRepWorker->subid, xid);
+		}
+	}
+
+	/* Check the status of apply background worker if any. */
+	apply_bgworker_check_status();
 
 	/* Process any tables that are being synchronized in parallel. */
 	process_syncing_tables(commit_data.end_lsn);
@@ -2467,7 +2834,7 @@ apply_handle_truncate(StringInfo s)
 /*
  * Logical replication protocol message dispatcher.
  */
-static void
+void
 apply_dispatch(StringInfo s)
 {
 	LogicalRepMsgType action = pq_getmsgbyte(s);
@@ -2636,6 +3003,10 @@ store_flush_position(XLogRecPtr remote_lsn)
 {
 	FlushPosition *flushpos;
 
+	/* Skip if not the main apply worker */
+	if (am_apply_bgworker())
+		return;
+
 	/* Need to do this in permanent context */
 	MemoryContextSwitchTo(ApplyContext);
 
@@ -2650,7 +3021,7 @@ store_flush_position(XLogRecPtr remote_lsn)
 
 
 /* Update statistics of the worker. */
-static void
+void
 UpdateWorkerStats(XLogRecPtr last_lsn, TimestampTz send_time, bool reply)
 {
 	MyLogicalRepWorker->last_lsn = last_lsn;
@@ -2812,6 +3183,9 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 			AcceptInvalidationMessages();
 			maybe_reread_subscription();
 
+			/* Check the status of apply background worker if any. */
+			apply_bgworker_check_status();
+
 			/* Process any table synchronization changes. */
 			process_syncing_tables(last_received);
 		}
@@ -3113,7 +3487,7 @@ maybe_reread_subscription(void)
 /*
  * Callback from subscription syscache invalidation.
  */
-static void
+void
 subscription_change_cb(Datum arg, int cacheid, uint32 hashvalue)
 {
 	MySubscriptionValid = false;
@@ -3709,7 +4083,7 @@ ApplyWorkerMain(Datum main_arg)
 		originid = replorigin_by_name(originname, true);
 		if (!OidIsValid(originid))
 			originid = replorigin_create(originname);
-		replorigin_session_setup(originid);
+		replorigin_session_setup(originid, true);
 		replorigin_session_origin = originid;
 		origin_startpos = replorigin_session_get_progress(false);
 		CommitTransactionCommand();
@@ -3750,13 +4124,14 @@ ApplyWorkerMain(Datum main_arg)
 
 	server_version = walrcv_server_version(LogRepWorkerWalRcvConn);
 	options.proto.logical.proto_version =
+		server_version >= 160000 ? LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM :
 		server_version >= 150000 ? LOGICALREP_PROTO_TWOPHASE_VERSION_NUM :
 		server_version >= 140000 ? LOGICALREP_PROTO_STREAM_VERSION_NUM :
 		LOGICALREP_PROTO_VERSION_NUM;
 
 	options.proto.logical.publication_names = MySubscription->publications;
 	options.proto.logical.binary = MySubscription->binary;
-	options.proto.logical.streaming = MySubscription->stream;
+	options.proto.logical.streaming = (MySubscription->stream != SUBSTREAM_OFF);
 	options.proto.logical.twophase = false;
 
 	if (!am_tablesync_worker())
@@ -3914,7 +4289,8 @@ clear_subscription_skip_lsn(XLogRecPtr finish_lsn)
 	XLogRecPtr	myskiplsn = MySubscription->skiplsn;
 	bool		started_tx = false;
 
-	if (likely(XLogRecPtrIsInvalid(myskiplsn)))
+	if (likely(XLogRecPtrIsInvalid(myskiplsn)) ||
+		am_apply_bgworker())
 		return;
 
 	if (!IsTransactionState())
@@ -3986,7 +4362,7 @@ clear_subscription_skip_lsn(XLogRecPtr finish_lsn)
 }
 
 /* Error callback to give more context info about the change being applied */
-static void
+void
 apply_error_callback(void *arg)
 {
 	ApplyErrorCallbackArg *errarg = &apply_error_callback_arg;
@@ -4014,23 +4390,47 @@ apply_error_callback(void *arg)
 					   errarg->remote_xid,
 					   LSN_FORMAT_ARGS(errarg->finish_lsn));
 	}
-	else if (errarg->remote_attnum < 0)
-		errcontext("processing remote data for replication origin \"%s\" during \"%s\" for replication target relation \"%s.%s\" in transaction %u finished at %X/%X",
-				   errarg->origin_name,
-				   logicalrep_message_type(errarg->command),
-				   errarg->rel->remoterel.nspname,
-				   errarg->rel->remoterel.relname,
-				   errarg->remote_xid,
-				   LSN_FORMAT_ARGS(errarg->finish_lsn));
 	else
-		errcontext("processing remote data for replication origin \"%s\" during \"%s\" for replication target relation \"%s.%s\" column \"%s\" in transaction %u finished at %X/%X",
-				   errarg->origin_name,
-				   logicalrep_message_type(errarg->command),
-				   errarg->rel->remoterel.nspname,
-				   errarg->rel->remoterel.relname,
-				   errarg->rel->remoterel.attnames[errarg->remote_attnum],
-				   errarg->remote_xid,
-				   LSN_FORMAT_ARGS(errarg->finish_lsn));
+	{
+		if (errarg->remote_attnum < 0)
+		{
+			if (XLogRecPtrIsInvalid(errarg->finish_lsn))
+				errcontext("processing remote data for replication origin \"%s\" during \"%s\" for replication target relation \"%s.%s\" in transaction %u",
+						   errarg->origin_name,
+						   logicalrep_message_type(errarg->command),
+						   errarg->rel->remoterel.nspname,
+						   errarg->rel->remoterel.relname,
+						   errarg->remote_xid);
+			else
+				errcontext("processing remote data for replication origin \"%s\" during \"%s\" for replication target relation \"%s.%s\" in transaction %u finished at %X/%X",
+						   errarg->origin_name,
+						   logicalrep_message_type(errarg->command),
+						   errarg->rel->remoterel.nspname,
+						   errarg->rel->remoterel.relname,
+						   errarg->remote_xid,
+						   LSN_FORMAT_ARGS(errarg->finish_lsn));
+		}
+		else
+		{
+			if (XLogRecPtrIsInvalid(errarg->finish_lsn))
+				errcontext("processing remote data for replication origin \"%s\" during \"%s\" for replication target relation \"%s.%s\" column \"%s\" in transaction %u",
+						   errarg->origin_name,
+						   logicalrep_message_type(errarg->command),
+						   errarg->rel->remoterel.nspname,
+						   errarg->rel->remoterel.relname,
+						   errarg->rel->remoterel.attnames[errarg->remote_attnum],
+						   errarg->remote_xid);
+			else
+				errcontext("processing remote data for replication origin \"%s\" during \"%s\" for replication target relation \"%s.%s\" column \"%s\" in transaction %u finished at %X/%X",
+						   errarg->origin_name,
+						   logicalrep_message_type(errarg->command),
+						   errarg->rel->remoterel.nspname,
+						   errarg->rel->remoterel.relname,
+						   errarg->rel->remoterel.attnames[errarg->remote_attnum],
+						   errarg->remote_xid,
+						   LSN_FORMAT_ARGS(errarg->finish_lsn));
+		}
+	}
 }
 
 /* Set transaction information of apply error callback */
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index ba8a24d099..de29dc6da9 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -1818,6 +1818,9 @@ pgoutput_stream_abort(struct LogicalDecodingContext *ctx,
 					  XLogRecPtr abort_lsn)
 {
 	ReorderBufferTXN *toptxn;
+	PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
+	bool write_abort_lsn = (data->protocol_version >=
+							LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM);
 
 	/*
 	 * The abort should happen outside streaming block, even for streamed
@@ -1831,7 +1834,8 @@ pgoutput_stream_abort(struct LogicalDecodingContext *ctx,
 	Assert(rbtxn_is_streamed(toptxn));
 
 	OutputPluginPrepareWrite(ctx, true);
-	logicalrep_write_stream_abort(ctx->out, toptxn->xid, txn->xid);
+	logicalrep_write_stream_abort(ctx->out, toptxn->xid, txn, abort_lsn,
+								  write_abort_lsn);
 	OutputPluginWrite(ctx, true);
 
 	cleanup_rel_sync_cache(toptxn->xid, false);
diff --git a/src/backend/utils/activity/wait_event.c b/src/backend/utils/activity/wait_event.c
index da57a93034..2e146fe087 100644
--- a/src/backend/utils/activity/wait_event.c
+++ b/src/backend/utils/activity/wait_event.c
@@ -388,6 +388,9 @@ pgstat_get_wait_ipc(WaitEventIPC w)
 		case WAIT_EVENT_HASH_GROW_BUCKETS_REINSERT:
 			event_name = "HashGrowBucketsReinsert";
 			break;
+		case WAIT_EVENT_LOGICAL_APPLY_WORKER_STATE_CHANGE:
+			event_name = "LogicalApplyWorkerStateChange";
+			break;
 		case WAIT_EVENT_LOGICAL_SYNC_DATA:
 			event_name = "LogicalSyncData";
 			break;
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 0328029d43..4284bcbcd1 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -3220,6 +3220,18 @@ static struct config_int ConfigureNamesInt[] =
 		NULL, NULL, NULL
 	},
 
+	{
+		{"max_apply_bgworkers_per_subscription",
+			PGC_SIGHUP,
+			REPLICATION_SUBSCRIBERS,
+			gettext_noop("Maximum number of apply background workers per subscription."),
+			NULL,
+		},
+		&max_apply_bgworkers_per_subscription,
+		2, 0, MAX_BACKENDS,
+		NULL, NULL, NULL
+	},
+
 	{
 		{"log_rotation_age", PGC_SIGHUP, LOGGING_WHERE,
 			gettext_noop("Sets the amount of time to wait before forcing "
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index b4bc06e5f5..ad18710af4 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -360,6 +360,7 @@
 #max_logical_replication_workers = 4	# taken from max_worker_processes
 					# (change requires restart)
 #max_sync_workers_per_subscription = 2	# taken from max_logical_replication_workers
+#max_apply_bgworkers_per_subscription = 2	# taken from max_logical_replication_workers
 
 
 #------------------------------------------------------------------------------
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index e4fdb6b75b..9099fe5f74 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -4456,7 +4456,7 @@ getSubscriptions(Archive *fout)
 	if (fout->remoteVersion >= 140000)
 		appendPQExpBufferStr(query, " s.substream,\n");
 	else
-		appendPQExpBufferStr(query, " false AS substream,\n");
+		appendPQExpBufferStr(query, " 'f' AS substream,\n");
 
 	if (fout->remoteVersion >= 150000)
 		appendPQExpBufferStr(query,
@@ -4586,8 +4586,10 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
 	if (strcmp(subinfo->subbinary, "t") == 0)
 		appendPQExpBufferStr(query, ", binary = true");
 
-	if (strcmp(subinfo->substream, "f") != 0)
+	if (strcmp(subinfo->substream, "t") == 0)
 		appendPQExpBufferStr(query, ", streaming = on");
+	else if (strcmp(subinfo->substream, "p") == 0)
+		appendPQExpBufferStr(query, ", streaming = parallel");
 
 	if (strcmp(subinfo->subtwophasestate, two_phase_disabled) != 0)
 		appendPQExpBufferStr(query, ", two_phase = on");
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index d1260f590c..d54540f5f5 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -68,7 +68,8 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
 	bool		subbinary;		/* True if the subscription wants the
 								 * publisher to send data in binary */
 
-	bool		substream;		/* Stream in-progress transactions. */
+	char		substream;		/* Stream in-progress transactions.
+								 * See SUBSTREAM_xxx constants. */
 
 	char		subtwophasestate;	/* Stream two-phase transactions */
 
@@ -109,7 +110,8 @@ typedef struct Subscription
 	bool		enabled;		/* Indicates if the subscription is enabled */
 	bool		binary;			/* Indicates if the subscription wants data in
 								 * binary format */
-	bool		stream;			/* Allow streaming in-progress transactions. */
+	char		stream;			/* Allow streaming in-progress transactions.
+								 * See SUBSTREAM_xxx constants. */
 	char		twophasestate;	/* Allow streaming two-phase transactions */
 	bool		disableonerr;	/* Indicates if the subscription should be
 								 * automatically disabled if a worker error
@@ -120,6 +122,21 @@ typedef struct Subscription
 	List	   *publications;	/* List of publication names to subscribe to */
 } Subscription;
 
+/* Disallow streaming in-progress transactions */
+#define SUBSTREAM_OFF 'f'
+
+/*
+ * Streaming in-progress transactions are written to a temporary file and
+ * applied only after the transaction is committed on upstream.
+ */
+#define SUBSTREAM_ON 't'
+
+/*
+ * Streaming in-progress transactions are applied immediately via a background
+ * worker
+ */
+#define SUBSTREAM_PARALLEL 'p'
+
 extern Subscription *GetSubscription(Oid subid, bool missing_ok);
 extern void FreeSubscription(Subscription *sub);
 extern void DisableSubscription(Oid subid);
diff --git a/src/include/replication/logicallauncher.h b/src/include/replication/logicallauncher.h
index f1e2821e25..ac8ef94381 100644
--- a/src/include/replication/logicallauncher.h
+++ b/src/include/replication/logicallauncher.h
@@ -14,6 +14,7 @@
 
 extern PGDLLIMPORT int max_logical_replication_workers;
 extern PGDLLIMPORT int max_sync_workers_per_subscription;
+extern PGDLLIMPORT int max_apply_bgworkers_per_subscription;
 
 extern void ApplyLauncherRegister(void);
 extern void ApplyLauncherMain(Datum main_arg);
diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h
index a771ab8ff3..eb0fd24fd8 100644
--- a/src/include/replication/logicalproto.h
+++ b/src/include/replication/logicalproto.h
@@ -32,12 +32,17 @@
  *
  * LOGICALREP_PROTO_TWOPHASE_VERSION_NUM is the minimum protocol version with
  * support for two-phase commit decoding (at prepare time). Introduced in PG15.
+ *
+ * LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM is the minimum protocol version
+ * with support for streaming large transactions using apply background
+ * workers. Introduced in PG16.
  */
 #define LOGICALREP_PROTO_MIN_VERSION_NUM 1
 #define LOGICALREP_PROTO_VERSION_NUM 1
 #define LOGICALREP_PROTO_STREAM_VERSION_NUM 2
 #define LOGICALREP_PROTO_TWOPHASE_VERSION_NUM 3
-#define LOGICALREP_PROTO_MAX_VERSION_NUM LOGICALREP_PROTO_TWOPHASE_VERSION_NUM
+#define LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM 4
+#define LOGICALREP_PROTO_MAX_VERSION_NUM LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM
 
 /*
  * Logical message types
@@ -175,6 +180,17 @@ typedef struct LogicalRepRollbackPreparedTxnData
 	char		gid[GIDSIZE];
 } LogicalRepRollbackPreparedTxnData;
 
+/*
+ * Transaction protocol information for stream abort.
+ */
+typedef struct LogicalRepStreamAbortData
+{
+	TransactionId xid;
+	TransactionId subxid;
+	XLogRecPtr	abort_lsn;
+	TimestampTz abort_time;
+} LogicalRepStreamAbortData;
+
 extern void logicalrep_write_begin(StringInfo out, ReorderBufferTXN *txn);
 extern void logicalrep_read_begin(StringInfo in,
 								  LogicalRepBeginData *begin_data);
@@ -246,9 +262,12 @@ extern void logicalrep_write_stream_commit(StringInfo out, ReorderBufferTXN *txn
 extern TransactionId logicalrep_read_stream_commit(StringInfo out,
 												   LogicalRepCommitData *commit_data);
 extern void logicalrep_write_stream_abort(StringInfo out, TransactionId xid,
-										  TransactionId subxid);
-extern void logicalrep_read_stream_abort(StringInfo in, TransactionId *xid,
-										 TransactionId *subxid);
+										  ReorderBufferTXN *txn,
+										  XLogRecPtr abort_lsn,
+										  bool write_abort_lsn);
+extern void logicalrep_read_stream_abort(StringInfo in,
+										 LogicalRepStreamAbortData *abort_data,
+										 bool read_abort_lsn);
 extern char *logicalrep_message_type(LogicalRepMsgType action);
 
 #endif							/* LOGICAL_PROTO_H */
diff --git a/src/include/replication/logicalworker.h b/src/include/replication/logicalworker.h
index cd1b6e8afc..6a1af7f13c 100644
--- a/src/include/replication/logicalworker.h
+++ b/src/include/replication/logicalworker.h
@@ -13,6 +13,7 @@
 #define LOGICALWORKER_H
 
 extern void ApplyWorkerMain(Datum main_arg);
+extern void ApplyBgworkerMain(Datum main_arg);
 
 extern bool IsLogicalWorker(void);
 
diff --git a/src/include/replication/origin.h b/src/include/replication/origin.h
index 2d1b5e5c28..c7389b40a7 100644
--- a/src/include/replication/origin.h
+++ b/src/include/replication/origin.h
@@ -53,7 +53,7 @@ extern XLogRecPtr replorigin_get_progress(RepOriginId node, bool flush);
 
 extern void replorigin_session_advance(XLogRecPtr remote_commit,
 									   XLogRecPtr local_commit);
-extern void replorigin_session_setup(RepOriginId node);
+extern void replorigin_session_setup(RepOriginId node, bool must_acquire);
 extern void replorigin_session_reset(void);
 extern XLogRecPtr replorigin_session_get_progress(bool flush);
 
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index d109d0baed..d2a80d79e5 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -301,6 +301,7 @@ typedef struct ReorderBufferTXN
 	{
 		TimestampTz commit_time;
 		TimestampTz prepare_time;
+		TimestampTz abort_time;
 	}			xact_time;
 
 	/*
@@ -647,9 +648,11 @@ extern void ReorderBufferFinishPrepared(ReorderBuffer *rb, TransactionId xid,
 extern void ReorderBufferAssignChild(ReorderBuffer *, TransactionId, TransactionId, XLogRecPtr commit_lsn);
 extern void ReorderBufferCommitChild(ReorderBuffer *, TransactionId, TransactionId,
 									 XLogRecPtr commit_lsn, XLogRecPtr end_lsn);
-extern void ReorderBufferAbort(ReorderBuffer *, TransactionId, XLogRecPtr lsn);
+extern void ReorderBufferAbort(ReorderBuffer *, TransactionId, XLogRecPtr lsn,
+							   TimestampTz abort_time);
 extern void ReorderBufferAbortOld(ReorderBuffer *, TransactionId xid);
-extern void ReorderBufferForget(ReorderBuffer *, TransactionId, XLogRecPtr lsn);
+extern void ReorderBufferForget(ReorderBuffer *, TransactionId, XLogRecPtr lsn,
+								TimestampTz abort_time);
 extern void ReorderBufferInvalidate(ReorderBuffer *, TransactionId, XLogRecPtr lsn);
 
 extern void ReorderBufferSetBaseSnapshot(ReorderBuffer *, TransactionId, XLogRecPtr lsn, struct SnapshotData *snap);
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 901845abc2..a3560d4904 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -17,8 +17,11 @@
 #include "access/xlogdefs.h"
 #include "catalog/pg_subscription.h"
 #include "datatype/timestamp.h"
+#include "replication/logicalrelation.h"
 #include "storage/fileset.h"
 #include "storage/lock.h"
+#include "storage/shm_mq.h"
+#include "storage/shm_toc.h"
 #include "storage/spin.h"
 
 
@@ -60,6 +63,9 @@ typedef struct LogicalRepWorker
 	 */
 	FileSet    *stream_fileset;
 
+	/* Indicates if this slot is used for an apply background worker. */
+	bool		subworker;
+
 	/* Stats. */
 	XLogRecPtr	last_lsn;
 	TimestampTz last_send_time;
@@ -68,8 +74,68 @@ typedef struct LogicalRepWorker
 	TimestampTz reply_time;
 } LogicalRepWorker;
 
+/* Struct for saving and restoring apply errcontext information */
+typedef struct ApplyErrorCallbackArg
+{
+	LogicalRepMsgType command;	/* 0 if invalid */
+	LogicalRepRelMapEntry *rel;
+
+	/* Remote node information */
+	int			remote_attnum;	/* -1 if invalid */
+	TransactionId remote_xid;
+	XLogRecPtr	finish_lsn;
+	char	   *origin_name;
+} ApplyErrorCallbackArg;
+
+/*
+ * Status of apply background worker.
+ */
+typedef enum ApplyBgworkerStatus
+{
+	APPLY_BGWORKER_BUSY = 0,		/* assigned to a transaction */
+	APPLY_BGWORKER_FINISHED,		/* transaction is completed */
+	APPLY_BGWORKER_EXIT				/* exit */
+} ApplyBgworkerStatus;
+
+/*
+ * Struct for sharing information between apply main and apply background
+ * workers.
+ */
+typedef struct ApplyBgworkerShared
+{
+	slock_t	mutex;
+
+	/* Status of apply background worker. */
+	ApplyBgworkerStatus	status;
+
+	/* server version of publisher. */
+	uint32	server_version;
+
+	TransactionId	stream_xid;
+	uint32	n;	/* id of apply background worker */
+} ApplyBgworkerShared;
+
+/*
+ * Struct for maintaining an apply background worker.
+ */
+typedef struct ApplyBgworkerState
+{
+	shm_mq_handle			*mq_handle;
+	dsm_segment				*dsm_seg;
+	ApplyBgworkerShared volatile	*shared;
+} ApplyBgworkerState;
+
 /* Main memory context for apply worker. Permanent during worker lifetime. */
 extern PGDLLIMPORT MemoryContext ApplyContext;
+extern PGDLLIMPORT MemoryContext ApplyMessageContext;
+
+extern PGDLLIMPORT ApplyErrorCallbackArg apply_error_callback_arg;
+
+extern PGDLLIMPORT bool MySubscriptionValid;
+
+extern PGDLLIMPORT volatile ApplyBgworkerShared *MyParallelShared;
+
+extern PGDLLIMPORT List *subxactlist;
 
 /* libpqreceiver connection */
 extern PGDLLIMPORT struct WalReceiverConn *LogRepWorkerWalRcvConn;
@@ -79,18 +145,22 @@ extern PGDLLIMPORT Subscription *MySubscription;
 extern PGDLLIMPORT LogicalRepWorker *MyLogicalRepWorker;
 
 extern PGDLLIMPORT bool in_remote_transaction;
+extern PGDLLIMPORT bool in_streamed_transaction;
+extern PGDLLIMPORT TransactionId stream_xid;
 
 extern void logicalrep_worker_attach(int slot);
 extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid,
 												bool only_running);
 extern List *logicalrep_workers_find(Oid subid, bool only_running);
-extern void logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname,
-									 Oid userid, Oid relid);
+extern bool logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname,
+									 Oid userid, Oid relid,
+									 dsm_handle subworker_dsm);
 extern void logicalrep_worker_stop(Oid subid, Oid relid);
 extern void logicalrep_worker_wakeup(Oid subid, Oid relid);
 extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);
 
 extern int	logicalrep_sync_worker_count(Oid subid);
+extern int	logicalrep_apply_bgworker_count(Oid subid);
 
 extern void ReplicationOriginNameForTablesync(Oid suboid, Oid relid,
 											  char *originname, int szorgname);
@@ -103,10 +173,38 @@ extern void process_syncing_tables(XLogRecPtr current_lsn);
 extern void invalidate_syncing_table_states(Datum arg, int cacheid,
 											uint32 hashvalue);
 
+extern void UpdateWorkerStats(XLogRecPtr last_lsn, TimestampTz send_time,
+							  bool reply);
+
+extern void apply_dispatch(StringInfo s);
+
+/* Function for apply error callback */
+extern void apply_error_callback(void *arg);
+
+extern void subscription_change_cb(Datum arg, int cacheid, uint32 hashvalue);
+
+/* Apply background worker setup and interactions */
+extern ApplyBgworkerState *apply_bgworker_start(TransactionId xid);
+extern ApplyBgworkerState *apply_bgworker_find(TransactionId xid);
+extern void apply_bgworker_wait_for(ApplyBgworkerState *wstate,
+									ApplyBgworkerStatus wait_for_status);
+extern void apply_bgworker_send_data(ApplyBgworkerState *wstate, Size nbytes,
+									 const void *data);
+extern void apply_bgworker_free(ApplyBgworkerState *wstate);
+extern void apply_bgworker_check_status(void);
+extern void apply_bgworker_set_status(ApplyBgworkerStatus status);
+extern void apply_bgworker_subxact_info_add(TransactionId current_xid);
+
 static inline bool
 am_tablesync_worker(void)
 {
 	return OidIsValid(MyLogicalRepWorker->relid);
 }
 
+static inline bool
+am_apply_bgworker(void)
+{
+	return MyLogicalRepWorker->subworker;
+}
+
 #endif							/* WORKER_INTERNAL_H */
diff --git a/src/include/utils/wait_event.h b/src/include/utils/wait_event.h
index c3ade01120..e35f199fd4 100644
--- a/src/include/utils/wait_event.h
+++ b/src/include/utils/wait_event.h
@@ -105,6 +105,7 @@ typedef enum
 	WAIT_EVENT_HASH_GROW_BUCKETS_ALLOCATE,
 	WAIT_EVENT_HASH_GROW_BUCKETS_ELECT,
 	WAIT_EVENT_HASH_GROW_BUCKETS_REINSERT,
+	WAIT_EVENT_LOGICAL_APPLY_WORKER_STATE_CHANGE,
 	WAIT_EVENT_LOGICAL_SYNC_DATA,
 	WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE,
 	WAIT_EVENT_MQ_INTERNAL,
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 5db7146e06..919266ae06 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -197,7 +197,7 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 DROP SUBSCRIPTION regress_testsub;
 -- fail - streaming must be boolean
 CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, streaming = foo);
-ERROR:  streaming requires a Boolean value
+ERROR:  streaming requires a Boolean value or "parallel"
 -- now it works
 CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, streaming = true);
 WARNING:  tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 34a76ceb60..4137dc77b4 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -116,6 +116,10 @@ Append
 AppendPath
 AppendRelInfo
 AppendState
+ApplyBgworkerEntry
+ApplyBgworkerShared
+ApplyBgworkerState
+ApplyBgworkerStatus
 ApplyErrorCallbackArg
 ApplyExecutionData
 ApplySubXactData
@@ -1485,6 +1489,7 @@ LogicalRepRelId
 LogicalRepRelMapEntry
 LogicalRepRelation
 LogicalRepRollbackPreparedTxnData
+LogicalRepStreamAbortData
 LogicalRepTupleData
 LogicalRepTyp
 LogicalRepWorker
-- 
2.23.0.windows.1

