v25-0001-Perform-streaming-logical-transactions-by-parall.patch

From 708e64490228ef170544ad4c21a76a63ec3f1ab4 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 v25 1/5] Perform streaming logical transactions by parallel
 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 parallel workers.

In this approach, we assign a new apply parallel worker (if available) as
soon as the xact's first stream is received and the leader apply worker will
send changes to this new worker via shared memory. The apply parallel 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 parallel
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 parallel 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).

In addition, the patch extends the logical replication STREAM_ABORT message so
that abort_time and abort_lsn can also be sent which can be used to update the
replication origin in apply parallel worker when the streaming transaction is
aborted.
---
 doc/src/sgml/catalogs.sgml                    |   11 +-
 doc/src/sgml/config.sgml                      |   28 +-
 doc/src/sgml/logical-replication.sgml         |   13 +-
 doc/src/sgml/protocol.sgml                    |   29 +-
 doc/src/sgml/ref/create_subscription.sgml     |   24 +-
 src/backend/access/transam/xact.c             |   13 +
 src/backend/commands/subscriptioncmds.c       |   66 +-
 src/backend/libpq/pqmq.c                      |   18 +-
 src/backend/postmaster/bgworker.c             |    3 +
 src/backend/replication/logical/Makefile      |    1 +
 .../replication/logical/applyparallelworker.c | 1093 +++++++++++++++++
 src/backend/replication/logical/decode.c      |   10 +-
 src/backend/replication/logical/launcher.c    |  166 ++-
 src/backend/replication/logical/proto.c       |   41 +-
 .../replication/logical/reorderbuffer.c       |   10 +-
 src/backend/replication/logical/tablesync.c   |    6 +-
 src/backend/replication/logical/worker.c      |  812 +++++++++---
 src/backend/replication/pgoutput/pgoutput.c   |    6 +-
 src/backend/storage/ipc/procsignal.c          |    4 +
 src/backend/tcop/postgres.c                   |    3 +
 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/nodes/pg_list.h                   |   11 +
 src/include/replication/logicallauncher.h     |    1 +
 src/include/replication/logicalproto.h        |   27 +-
 src/include/replication/logicalworker.h       |    6 +
 src/include/replication/reorderbuffer.h       |    7 +-
 src/include/replication/worker_internal.h     |  123 +-
 src/include/storage/procsignal.h              |    1 +
 src/include/utils/wait_event.h                |    1 +
 src/test/regress/expected/subscription.out    |   12 +-
 src/test/regress/sql/subscription.sql         |    6 +-
 src/tools/pgindent/typedefs.list              |    4 +
 36 files changed, 2345 insertions(+), 254 deletions(-)
 create mode 100644 src/backend/replication/logical/applyparallelworker.c

diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index 00f833d210..0d12498859 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -7892,11 +7892,16 @@ 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 an apply parallel
+       worker if available (same as 't' if no worker is available)
       </para></entry>
      </row>
 
diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index a5cd4e44c7..86020e1086 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -4940,7 +4940,8 @@ ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="
       <listitem>
        <para>
         Specifies maximum number of logical replication workers. This includes
-        both apply workers and table synchronization workers.
+        apply leader workers, apply parallel workers, and table synchronization
+        workers.
        </para>
        <para>
         Logical replication workers are taken from the pool defined by
@@ -4980,6 +4981,31 @@ ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="
       </listitem>
      </varlistentry>
 
+     <varlistentry id="guc-max-apply-parallel-workers-per-subscription" xreflabel="max_apply_parallel_workers_per_subscription">
+      <term><varname>max_apply_parallel_workers_per_subscription</varname> (<type>integer</type>)
+      <indexterm>
+       <primary><varname>max_apply_parallel_workers_per_subscription</varname> configuration parameter</primary>
+      </indexterm>
+      </term>
+      <listitem>
+       <para>
+        Maximum number of apply parallel workers per subscription. This
+        parameter controls the amount of parallelism for streaming of
+        in-progress transactions with subscription parameter
+        <literal>streaming = parallel</literal>.
+       </para>
+       <para>
+        The apply parallel 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..a24ca10d9a 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> or <literal>off</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">
@@ -1334,7 +1344,8 @@ CONTEXT:  processing remote data for replication origin "pg_16395" during "INSER
    subscription.  A disabled subscription or a crashed subscription will have
    zero rows in this view.  If the initial data synchronization of any
    table is in progress, there will be additional workers for the tables
-   being synchronized.
+   being synchronized. Moreover, if the streaming transaction is applied in
+   parallel, there will be additional workers.
   </para>
  </sect1>
 
diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index 87870c5b10..d107dd2f80 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -3096,7 +3096,7 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;"
      <listitem>
       <para>
        Protocol version. Currently versions <literal>1</literal>, <literal>2</literal>,
-       and <literal>3</literal> are supported.
+       <literal>3</literal>, and <literal>4</literal> are supported.
       </para>
       <para>
        Version <literal>2</literal> is supported only for server version 14
@@ -3106,6 +3106,11 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;"
        Version <literal>3</literal> is supported only for server version 15
        and above, and it allows streaming of two-phase commits.
       </para>
+      <para>
+       Version <literal>4</literal> is supported only for server version 16
+       and above, and it allows applying streams of large in-progress
+       transactions in parallel.
+      </para>
      </listitem>
     </varlistentry>
 
@@ -6876,6 +6881,28 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;"
         </para>
        </listitem>
       </varlistentry>
+
+      <varlistentry>
+       <term>Int64 (XLogRecPtr)</term>
+       <listitem>
+        <para>
+         The LSN of the abort. This field is available since protocol version
+         4.
+        </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). This field is
+         available since protocol version 4.
+        </para>
+       </listitem>
+      </varlistentry>
+
      </variablelist>
     </listitem>
    </varlistentry>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 7390c715bc..689b379d93 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 and received by the subscriber.
+         </para>
+
+         <para>
+          If set to <literal>parallel</literal>, incoming changes are directly
+          applied via one of the apply parallel workers, if available. If no
+          parallel worker is free to handle streaming transactions 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 parallel 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 50f092d7eb..5a76d9c1a1 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 670b219c8d..3dc03dc220 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -84,7 +84,7 @@ typedef struct SubOpts
 	bool		copy_data;
 	bool		refresh;
 	bool		binary;
-	bool		streaming;
+	char		streaming;
 	bool		twophase;
 	bool		disableonerr;
 	char	   *origin;
@@ -97,6 +97,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.
  *
@@ -134,7 +190,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))
@@ -237,7 +293,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)
 		{
@@ -627,7 +683,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 :
@@ -1089,7 +1145,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/libpq/pqmq.c b/src/backend/libpq/pqmq.c
index 4d0415e379..4b2a732a57 100644
--- a/src/backend/libpq/pqmq.c
+++ b/src/backend/libpq/pqmq.c
@@ -13,11 +13,13 @@
 
 #include "postgres.h"
 
+#include "access/parallel.h"
 #include "libpq/libpq.h"
 #include "libpq/pqformat.h"
 #include "libpq/pqmq.h"
 #include "miscadmin.h"
 #include "pgstat.h"
+#include "replication/logicalworker.h"
 #include "tcop/tcopprot.h"
 #include "utils/builtins.h"
 
@@ -162,9 +164,19 @@ mq_putmessage(char msgtype, const char *s, size_t len)
 		result = shm_mq_sendv(pq_mq_handle, iov, 2, true, true);
 
 		if (pq_mq_parallel_leader_pid != 0)
-			SendProcSignal(pq_mq_parallel_leader_pid,
-						   PROCSIG_PARALLEL_MESSAGE,
-						   pq_mq_parallel_leader_backend_id);
+		{
+			if (IsParallelWorker())
+				SendProcSignal(pq_mq_parallel_leader_pid,
+							   PROCSIG_PARALLEL_MESSAGE,
+							   pq_mq_parallel_leader_backend_id);
+			else
+			{
+				Assert(IsLogicalApplyParallelWorker());
+				SendProcSignal(pq_mq_parallel_leader_pid,
+							   PROCSIG_PARALLEL_APPLY_MESSAGE,
+							   pq_mq_parallel_leader_backend_id);
+			}
+		}
 
 		if (result != SHM_MQ_WOULD_BLOCK)
 			break;
diff --git a/src/backend/postmaster/bgworker.c b/src/backend/postmaster/bgworker.c
index 40601aefd9..29beb23e0b 100644
--- a/src/backend/postmaster/bgworker.c
+++ b/src/backend/postmaster/bgworker.c
@@ -128,6 +128,9 @@ static const struct
 	},
 	{
 		"ApplyWorkerMain", ApplyWorkerMain
+	},
+	{
+		"ApplyParallelWorkerMain", ApplyParallelWorkerMain
 	}
 };
 
diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile
index c4e2fdeb71..2dc25e37bb 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 = \
+	applyparallelworker.o \
 	decode.o \
 	launcher.o \
 	logical.o \
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
new file mode 100644
index 0000000000..832e99cd48
--- /dev/null
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -0,0 +1,1093 @@
+/*-------------------------------------------------------------------------
+ * applyparallelworker.c
+ *     Support routines for applying xact by apply parallel worker
+ *
+ * Copyright (c) 2016-2022, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *	  src/backend/replication/logical/applyparallelworker.c
+ *
+ * This file contains routines that are intended to support setting up, using,
+ * and tearing down a ApplyParallelWorkerInfo.
+ *
+ * Refer to the comments in file header of logical/worker.c to see more
+ * information about apply parallel worker.
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "libpq/pqformat.h"
+#include "libpq/pqmq.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 0x787ca067
+
+/*
+ * DSM keys for apply parallel 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 PARALLEL_APPLY_KEY_SHARED		1
+#define PARALLEL_APPLY_KEY_MQ			2
+#define PARALLEL_APPLY_KEY_ERROR_QUEUE	3
+
+/* Queue size of DSM, 16 MB for now. */
+#define DSM_QUEUE_SIZE	(16 * 1024 * 1024)
+
+/*
+ * Error queue size of DSM. It is desirable to make it large enough that a
+ * typical ErrorResponse can be sent without blocking.  That way, a worker that
+ * errors out can write the whole message into the queue and terminate without
+ * waiting for the user backend.
+ */
+#define DSM_ERROR_QUEUE_SIZE			16384
+
+/*
+ * There are three fields in each message received by apply worker: start_lsn,
+ * end_lsn and send_time. Because we have updated these statistics in apply
+ * worker, we could ignore these fields in apply parallel worker (see function
+ * LogicalRepApplyLoop).
+ */
+#define SIZE_STATS_MESSAGE (2 * sizeof(XLogRecPtr) + sizeof(TimestampTz))
+
+/*
+ * Entry for a hash table we use to map from xid to the apply parallel worker
+ * state.
+ */
+typedef struct ApplyParallelWorkerEntry
+{
+	TransactionId xid;	/* Hash key -- must be first */
+	ApplyParallelWorkerInfo *winfo;
+} ApplyParallelWorkerEntry;
+
+/* Apply parallel workers hash table (initialized on first use). */
+static HTAB *ApplyParallelWorkersHash = NULL;
+static List *ApplyParallelWorkersFreeList = NIL;
+static List *ApplyParallelWorkersList = NIL;
+
+/*
+ * Information shared between leader apply worker and apply parallel worker.
+ */
+volatile ApplyParallelWorkerShared *MyParallelShared = NULL;
+
+/*
+ * Is there a message pending in apply parallel worker which we need to
+ * receive?
+ */
+volatile bool ParallelApplyMessagePending = false;
+
+/*
+ * Cache a pointer to the apply parallel worker which is responsible for
+ * applying the current streaming transactions. It is used to save the
+ * cost of searching the hash table when applying the changes between
+ * STREAM_START and STREAM_STOP.
+ */
+ApplyParallelWorkerInfo *stream_apply_worker = NULL;
+
+List	   *subxactlist = NIL;
+
+static bool parallel_apply_can_start(TransactionId xid);
+static ApplyParallelWorkerInfo *parallel_apply_setup_worker(void);
+static bool parallel_apply_setup_dsm(ApplyParallelWorkerInfo *winfo);
+
+/*
+ * Check if starting a new apply parallel worker is allowed.
+ */
+static bool
+parallel_apply_can_start(TransactionId xid)
+{
+	if (!TransactionIdIsValid(xid))
+		return false;
+
+	/*
+	 * Don't start a new parallel worker if not in streaming parallel mode.
+	 */
+	if (MySubscription->stream != SUBSTREAM_PARALLEL)
+		return false;
+
+	/*
+	 * Don't start a new parallel 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 using apply parallel
+	 * 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 the new apply
+	 * parallel worker in this case.
+	 */
+	if (!AllTablesyncsReady())
+		return false;
+
+	return true;
+}
+
+/*
+ * Return the apply parallel worker information that will be used for the
+ * specified xid.
+ *
+ * If an apply parallel worker is found in the free list then re-use it,
+ * otherwise start a fresh one. Cache the worker information in
+ * ApplyParallelWorkersHash keyed by the specified xid.
+ */
+ApplyParallelWorkerInfo *
+parallel_apply_start_worker(TransactionId xid)
+{
+	bool		found;
+	int			server_version;
+	ApplyParallelWorkerInfo *winfo = NULL;
+	ApplyParallelWorkerEntry *entry = NULL;
+
+	if (!parallel_apply_can_start(xid))
+		return NULL;
+
+	/* First time through, initialize apply workers hashtable. */
+	if (ApplyParallelWorkersHash == NULL)
+	{
+		HASHCTL		ctl;
+
+		MemSet(&ctl, 0, sizeof(ctl));
+		ctl.keysize = sizeof(TransactionId);
+		ctl.entrysize = sizeof(ApplyParallelWorkerEntry);
+		ctl.hcxt = ApplyContext;
+
+		ApplyParallelWorkersHash = hash_create("logical apply workers hash", 16, &ctl,
+									   HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
+	}
+
+	/* Try to get a free apply parallel worker. */
+	while (ApplyParallelWorkersFreeList != NIL)
+	{
+		ApplyParallelWorkerInfo *tmp_winfo;
+
+		tmp_winfo = (ApplyParallelWorkerInfo *) llast(ApplyParallelWorkersFreeList);
+		ApplyParallelWorkersFreeList = list_delete_last(ApplyParallelWorkersFreeList);
+
+		if (tmp_winfo->error_mq_handle != NULL)
+		{
+			winfo = tmp_winfo;
+			break;
+		}
+
+		/*
+		 * Release the worker information and try next one if the apply
+		 * parallel worker exited cleanly.
+		 */
+		ApplyParallelWorkersList = list_delete(ApplyParallelWorkersList, tmp_winfo);
+		shm_mq_detach(tmp_winfo->mq_handle);
+		dsm_detach(tmp_winfo->dsm_seg);
+		pfree(tmp_winfo);
+	}
+
+	/* Try to start a new apply parallel worker. */
+	if (winfo == NULL)
+		winfo = parallel_apply_setup_worker();
+
+	/* Failed to start a new apply parallel worker. */
+	if (winfo == NULL)
+		return NULL;
+
+	/* Create entry for requested transaction. */
+	entry = hash_search(ApplyParallelWorkersHash, &xid, HASH_ENTER, &found);
+	if (found)
+		elog(ERROR, "hash table corrupted");
+
+	server_version = walrcv_server_version(LogRepWorkerWalRcvConn);
+	winfo->shared->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;
+
+	winfo->shared->in_use = true;
+	winfo->shared->stream_xid = xid;
+	entry->winfo = winfo;
+	entry->xid = xid;
+
+	return winfo;
+}
+
+/*
+ * Find the previously assigned worker for the given transaction, if any.
+ */
+ApplyParallelWorkerInfo *
+parallel_apply_find_worker(TransactionId xid)
+{
+	bool		found;
+	ApplyParallelWorkerEntry *entry = NULL;
+
+	if (!TransactionIdIsValid(xid))
+		return NULL;
+
+	if (ApplyParallelWorkersHash == NULL)
+		return NULL;
+
+	/* Return the cached apply parallel worker if valid. */
+	if (stream_apply_worker != NULL)
+		return stream_apply_worker;
+
+	/*
+	 * Find entry for requested transaction.
+	 */
+	entry = hash_search(ApplyParallelWorkersHash, &xid, HASH_FIND, &found);
+	if (found)
+	{
+		/* If any workers (or the postmaster) have died, we have failed. */
+		if (entry->winfo->error_mq_handle == NULL)
+			ereport(ERROR,
+					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					 errmsg("apply parallel worker %u failed to apply transaction %u",
+							entry->winfo->shared->worker_id,
+							entry->winfo->shared->stream_xid)));
+
+		Assert(entry->winfo->shared->in_use);
+
+		return entry->winfo;
+	}
+
+	return NULL;
+}
+
+/*
+ * Remove the apply parallel worker entry from the hash table. And stop the
+ * worker if there are enough workers in the pool, otherwise add it to the free
+ * list.
+ */
+void
+parallel_apply_free_worker(ApplyParallelWorkerInfo *winfo)
+{
+	TransactionId xid = winfo->shared->stream_xid;
+	int			napplyworkers;
+
+	Assert(!winfo->shared->in_use);
+
+	hash_search(ApplyParallelWorkersHash, &xid, HASH_REMOVE, NULL);
+
+	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+	napplyworkers = logicalrep_apply_parallel_worker_count(MyLogicalRepWorker->subid);
+	LWLockRelease(LogicalRepWorkerLock);
+
+	if (napplyworkers > (max_apply_parallel_workers_per_subscription / 2))
+	{
+		logicalrep_worker_stop_by_slot(winfo->shared->logicalrep_worker_slot_no,
+									   winfo->shared->logicalrep_worker_generation);
+
+		ApplyParallelWorkersList = list_delete_ptr(ApplyParallelWorkersList, winfo);
+
+		shm_mq_detach(winfo->mq_handle);
+		shm_mq_detach(winfo->error_mq_handle);
+		dsm_detach(winfo->dsm_seg);
+		pfree(winfo);
+	}
+	else
+	{
+		MemoryContext	oldctx;
+
+		elog(DEBUG1, "adding finished apply worker %u for xid %u to the free list",
+			 winfo->shared->worker_id, winfo->shared->stream_xid);
+
+		oldctx = MemoryContextSwitchTo(ApplyContext);
+		ApplyParallelWorkersFreeList = lappend(ApplyParallelWorkersFreeList, winfo);
+		MemoryContextSwitchTo(oldctx);
+	}
+}
+
+/* Apply parallel worker main loop. */
+static void
+LogicalParallelApplyLoop(shm_mq_handle *mqh, volatile ApplyParallelWorkerShared *shared)
+{
+	shm_mq_result shmq_res;
+	PGPROC	   *registrant;
+	ErrorContextCallback errcallback;
+
+	registrant = BackendPidGetProc(MyBgworkerEntry->bgw_notify_pid);
+	SetLatch(&registrant->procLatch);
+
+	/*
+	 * Init the ApplyMessageContext which we clean up after each replication
+	 * protocol message.
+	 */
+	ApplyMessageContext = AllocSetContextCreate(ApplyContext,
+												"ApplyMessageContext",
+												ALLOCSET_DEFAULT_SIZES);
+
+	/*
+	 * Push apply error context callback. Fields will be filled while 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 leader apply worker")));
+
+		if (len == 0)
+			break;
+
+		s.cursor = 0;
+		s.maxlen = -1;
+		s.data = (char *) data;
+		s.len = len;
+
+		/*
+		 * The first byte of message for additional communication between
+		 * leader apply worker and apply parallel workers can only be 'w'.
+		 */
+		c = pq_getmsgbyte(&s);
+		if (c != 'w')
+			elog(ERROR, "[Apply Parallel Worker #%u] unexpected message \"%c\"",
+				 shared->worker_id, c);
+
+		/*
+		 * Ignore statistics fields that have been updated by the leader apply
+		 * worker.
+		 */
+		s.cursor += SIZE_STATS_MESSAGE;
+
+		apply_dispatch(&s);
+
+		MemoryContextSwitchTo(oldctx);
+		MemoryContextReset(ApplyMessageContext);
+
+		if (ConfigReloadPending)
+		{
+			ConfigReloadPending = false;
+			ProcessConfigFile(PGC_SIGHUP);
+		}
+	}
+
+	/* Pop the error context stack. */
+	error_context_stack = errcallback.previous;
+
+	elog(DEBUG1, "[Apply Parallel Worker #%u] exiting", shared->worker_id);
+
+	/* Signal main process that we are done. */
+	SetLatch(&registrant->procLatch);
+}
+
+/*
+ * Make sure the leader apply worker tries to read from our error queue one more
+ * time. This guards against the case where we exit uncleanly without sending
+ * an ErrorResponse, for example because some code calls proc_exit directly.
+ */
+static void
+parallel_apply_shutdown(int code, Datum arg)
+{
+	SendProcSignal(MyLogicalRepWorker->apply_leader_pid,
+				   PROCSIG_PARALLEL_APPLY_MESSAGE,
+				   InvalidBackendId);
+
+	dsm_detach((dsm_segment *) DatumGetPointer(arg));
+}
+
+/*
+ * Apply parallel worker entry point.
+ */
+void
+ApplyParallelWorkerMain(Datum main_arg)
+{
+	volatile ApplyParallelWorkerShared *shared;
+	dsm_handle	handle;
+	dsm_segment *seg;
+	shm_toc    *toc;
+	shm_mq	   *mq;
+	shm_mq_handle *mqh;
+	shm_mq_handle *error_mqh;
+	int			worker_slot = DatumGetInt32(main_arg);
+	char		originname[NAMEDATALEN];
+
+	/* Setup signal handling. */
+	pqsignal(SIGHUP, SignalHandlerForConfigReload);
+	pqsignal(SIGTERM, die);
+	BackgroundWorkerUnblockSignals();
+
+	/*
+	 * Attach to the dynamic shared memory segment for the parallel apply, and
+	 * find its table of contents.
+	 *
+	 * Like parallel query, we don't need resource owner by this time. See
+	 * ParallelWorkerMain.
+	 */
+	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(parallel_apply_shutdown, PointerGetDatum(seg));
+
+	/* Look up the shared information. */
+	shared = shm_toc_lookup(toc, PARALLEL_APPLY_KEY_SHARED, false);
+	MyParallelShared = shared;
+
+	/*
+	 * Attach to the message queue.
+	 */
+	mq = shm_toc_lookup(toc, PARALLEL_APPLY_KEY_MQ, false);
+	shm_mq_set_receiver(mq, MyProc);
+	mqh = shm_mq_attach(mq, seg, NULL);
+
+	/*
+	 * Attach to the error queue.
+	 */
+	mq = shm_toc_lookup(toc, PARALLEL_APPLY_KEY_ERROR_QUEUE, false);
+	shm_mq_set_sender(mq, MyProc);
+	error_mqh = shm_mq_attach(mq, seg, NULL);
+
+	/*
+	 * Primary initialization is complete. Now, we can attach to our slot. This
+	 * is used to ensure that the leader apply worker does not write data to the
+	 * uninitialized memory queue.
+	 */
+	logicalrep_worker_attach(worker_slot);
+	MyParallelShared->logicalrep_worker_generation = MyLogicalRepWorker->generation;
+	MyParallelShared->logicalrep_worker_slot_no = worker_slot;
+
+	pq_redirect_to_shm_mq(seg, error_mqh);
+	pq_set_parallel_leader(MyLogicalRepWorker->apply_leader_pid,
+						   InvalidBackendId);
+
+	MyLogicalRepWorker->last_send_time = MyLogicalRepWorker->last_recv_time =
+		MyLogicalRepWorker->reply_time = 0;
+
+	InitializeApplyWorker();
+
+	/*
+	 * Setup callback for syscache so that we know when something changes in
+	 * the subscription relation state.
+	 */
+	CacheRegisterSyscacheCallback(SUBSCRIPTIONRELMAP,
+								  invalidate_syncing_table_states,
+								  (Datum) 0);
+
+	/*
+	 * Allocate the origin name in long-lived context for error context
+	 * message.
+	 */
+	apply_error_callback_arg.origin_name = MemoryContextStrdup(ApplyContext,
+															   originname);
+
+	elog(DEBUG1, "[Apply Parallel Worker #%u] started", shared->worker_id);
+
+	LogicalParallelApplyLoop(mqh, shared);
+
+	proc_exit(0);
+}
+
+/*
+ * Handle receipt of an interrupt indicating a apply parallel worker message.
+ *
+ * Note: this is called within a signal handler!  All we can do is set a flag
+ * that will cause the next CHECK_FOR_INTERRUPTS() to invoke
+ * HandleParallelApplyMessages().
+ */
+void
+HandleParallelApplyMessageInterrupt(void)
+{
+	InterruptPending = true;
+	ParallelApplyMessagePending = true;
+	SetLatch(MyLatch);
+}
+
+/*
+ * Handle a single protocol message received from a single apply parallel
+ * worker.
+ */
+static void
+HandleParallelApplyMessage(ApplyParallelWorkerInfo *winfo, StringInfo msg)
+{
+	char		msgtype;
+
+	msgtype = pq_getmsgbyte(msg);
+
+	switch (msgtype)
+	{
+		case 'E':				/* ErrorResponse */
+			{
+				ErrorData	edata;
+				ErrorContextCallback *save_error_context_stack;
+
+				/* Parse ErrorResponse. */
+				pq_parse_errornotice(msg, &edata);
+
+				/* Death of a worker isn't enough justification for suicide. */
+				edata.elevel = Min(edata.elevel, ERROR);
+
+				/*
+				 * If desired, add a context line to show that this is a
+				 * message propagated from a apply parallel worker.
+				 * Otherwise, it can sometimes be confusing to understand what
+				 * actually happened.
+				 */
+				if (edata.context)
+					edata.context = psprintf("%s\n%s", edata.context,
+											 _("apply parallel worker"));
+				else
+					edata.context = pstrdup(_("apply parallel worker"));
+
+				/*
+				 * Context beyond that should use the error context callbacks
+				 * that were in effect in LogicalRepApplyLoop().
+				 */
+				save_error_context_stack = error_context_stack;
+				error_context_stack = apply_error_context_stack;
+
+				ThrowErrorData(&edata);
+
+				/* Should not reach here after rethrowing an error. */
+				error_context_stack = save_error_context_stack;
+
+				break;
+			}
+
+		case 'X':				/* Terminate, indicating clean exit */
+			{
+				shm_mq_detach(winfo->error_mq_handle);
+				winfo->error_mq_handle = NULL;
+				break;
+			}
+
+			/*
+			 * Don't need to do anything about NoticeResponse and
+			 * NotifyResponse as the logical replication worker doesn't need to
+			 * send messages to the client.
+			 */
+		case 'N':
+		case 'A':
+				break;
+		default:
+			{
+				elog(ERROR, "unrecognized message type received from apply parallel worker: %c (message length %d bytes)",
+					 msgtype, msg->len);
+			}
+	}
+}
+
+/*
+ * Handle any queued protocol messages received from apply parallel workers.
+ */
+void
+HandleParallelApplyMessages(void)
+{
+	ListCell	 *lc;
+	MemoryContext oldcontext;
+
+	static MemoryContext hpm_context = NULL;
+
+	/*
+	 * This is invoked from ProcessInterrupts(), and since some of the
+	 * functions it calls contain CHECK_FOR_INTERRUPTS(), there is a potential
+	 * for recursive calls if more signals are received while this runs.  It's
+	 * unclear that recursive entry would be safe, and it doesn't seem useful
+	 * even if it is safe, so let's block interrupts until done.
+	 */
+	HOLD_INTERRUPTS();
+
+	/*
+	 * Moreover, CurrentMemoryContext might be pointing almost anywhere.  We
+	 * don't want to risk leaking data into long-lived contexts, so let's do
+	 * our work here in a private context that we can reset on each use.
+	 */
+	if (hpm_context == NULL)	/* first time through? */
+		hpm_context = AllocSetContextCreate(TopMemoryContext,
+											"HandleParallelApplyMessages",
+											ALLOCSET_DEFAULT_SIZES);
+	else
+		MemoryContextReset(hpm_context);
+
+	oldcontext = MemoryContextSwitchTo(hpm_context);
+
+	/* OK to process messages.  Reset the flag saying there are more to do. */
+	ParallelApplyMessagePending = false;
+
+	foreach(lc, ApplyParallelWorkersList)
+	{
+		shm_mq_result res;
+		Size		nbytes;
+		void	   *data;
+		ApplyParallelWorkerInfo *winfo = (ApplyParallelWorkerInfo *) lfirst(lc);
+
+		if (winfo->error_mq_handle == NULL)
+			continue;
+
+		res = shm_mq_receive(winfo->error_mq_handle, &nbytes,
+							 &data, true);
+
+		if (res == SHM_MQ_WOULD_BLOCK)
+			break;
+		else if (res == SHM_MQ_SUCCESS)
+		{
+			StringInfoData msg;
+
+			initStringInfo(&msg);
+			appendBinaryStringInfo(&msg, data, nbytes);
+			HandleParallelApplyMessage(winfo, &msg);
+			pfree(msg.data);
+		}
+		else
+			ereport(ERROR,
+					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					 errmsg("lost connection to the leader apply worker")));
+	}
+
+	MemoryContextSwitchTo(oldcontext);
+
+	/* Might as well clear the context on our way out */
+	MemoryContextReset(hpm_context);
+
+	RESUME_INTERRUPTS();
+}
+
+/*
+ * Set up a dynamic shared memory segment.
+ *
+ * We set up a control region that contains a ApplyParallelWorkerShared, plus
+ * one region to each of message queue and error queue.
+ *
+ * Returns true on success, false on failure.
+ */
+static bool
+parallel_apply_setup_dsm(ApplyParallelWorkerInfo *winfo)
+{
+	shm_toc_estimator e;
+	Size		segsize;
+	dsm_segment *seg;
+	shm_toc    *toc;
+	ApplyParallelWorkerShared *shared;
+	shm_mq	   *mq;
+	Size		queue_size = DSM_QUEUE_SIZE;
+	Size		error_queue_size = DSM_ERROR_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(ApplyParallelWorkerShared));
+	shm_toc_estimate_chunk(&e, queue_size);
+	shm_toc_estimate_chunk(&e, error_queue_size);
+
+	shm_toc_estimate_keys(&e, 3);
+	segsize = shm_toc_estimate(&e);
+
+	/* Create the shared memory segment and establish a table of contents. */
+	seg = dsm_create(shm_toc_estimate(&e), 0);
+	if (seg == NULL)
+		return false;
+
+	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(ApplyParallelWorkerShared));
+	SpinLockInit(&shared->mutex);
+
+	server_version = walrcv_server_version(LogRepWorkerWalRcvConn);
+	shared->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;
+
+	shared->in_use = false;
+	shared->stream_xid = stream_xid;
+	shared->worker_id = list_length(ApplyParallelWorkersList) + 1;
+
+	shm_toc_insert(toc, PARALLEL_APPLY_KEY_SHARED, shared);
+
+	/* Set up message queue for the worker. */
+	mq = shm_mq_create(shm_toc_allocate(toc, queue_size), queue_size);
+	shm_toc_insert(toc, PARALLEL_APPLY_KEY_MQ, mq);
+	shm_mq_set_sender(mq, MyProc);
+
+	/* Attach the queue. */
+	winfo->mq_handle = shm_mq_attach(mq, seg, NULL);
+
+	/* Set up error queue for the worker. */
+	mq = shm_mq_create(shm_toc_allocate(toc, error_queue_size),
+					   error_queue_size);
+	shm_toc_insert(toc, PARALLEL_APPLY_KEY_ERROR_QUEUE, mq);
+	shm_mq_set_receiver(mq, MyProc);
+
+	/* Attach the queue. */
+	winfo->error_mq_handle = shm_mq_attach(mq, seg, NULL);
+
+	/* Return results to caller. */
+	winfo->dsm_seg = seg;
+	winfo->shared = shared;
+
+	return true;
+}
+
+/*
+ * Start apply parallel worker process and allocate shared memory for it.
+ */
+static ApplyParallelWorkerInfo *
+parallel_apply_setup_worker(void)
+{
+	MemoryContext oldcontext;
+	bool		launched;
+	ApplyParallelWorkerInfo *winfo;
+	int			napplyworkers;
+
+	elog(DEBUG1, "setting up apply parallel worker %u",
+		 list_length(ApplyParallelWorkersList) + 1);
+
+	/* Check if there are free worker slot(s). */
+	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+	napplyworkers = logicalrep_apply_parallel_worker_count(MyLogicalRepWorker->subid);
+	LWLockRelease(LogicalRepWorkerLock);
+
+	if (napplyworkers >= max_apply_parallel_workers_per_subscription)
+		return NULL;
+
+	oldcontext = MemoryContextSwitchTo(ApplyContext);
+
+	winfo = (ApplyParallelWorkerInfo *) palloc0(sizeof(ApplyParallelWorkerInfo));
+
+	/* Setup shared memory. */
+	if (!parallel_apply_setup_dsm(winfo))
+	{
+		MemoryContextSwitchTo(oldcontext);
+		pfree(winfo);
+
+		return NULL;
+	}
+
+	launched = logicalrep_worker_launch(MyLogicalRepWorker->dbid,
+										MySubscription->oid,
+										MySubscription->name,
+										MyLogicalRepWorker->userid,
+										InvalidOid,
+										dsm_segment_handle(winfo->dsm_seg));
+
+	if (launched)
+		ApplyParallelWorkersList = lappend(ApplyParallelWorkersList, winfo);
+	else
+	{
+		shm_mq_detach(winfo->mq_handle);
+		shm_mq_detach(winfo->error_mq_handle);
+		dsm_detach(winfo->dsm_seg);
+		pfree(winfo);
+
+		winfo = NULL;
+	}
+
+	MemoryContextSwitchTo(oldcontext);
+
+	return winfo;
+}
+
+/*
+ * Send the data to the specified apply parallel worker via shared-memory queue.
+ */
+void
+parallel_apply_send_data(ApplyParallelWorkerInfo *winfo, Size nbytes, const void *data)
+{
+	shm_mq_result result;
+
+	result = shm_mq_send(winfo->mq_handle, nbytes, data, false, true);
+
+	if (result != SHM_MQ_SUCCESS)
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("could not send data to shared-memory queue")));
+}
+
+/*
+ * Wait until the apply parallel worker is free (i.e. in_use flag is false).
+ */
+void
+parallel_apply_wait_for_free(ApplyParallelWorkerInfo *winfo)
+{
+	for (;;)
+	{
+		bool		in_use;
+
+		SpinLockAcquire(&winfo->shared->mutex);
+		in_use = winfo->shared->in_use;
+		SpinLockRelease(&winfo->shared->mutex);
+
+		if (!in_use)
+			break;
+
+		/* If any workers have died, we have failed. */
+		if (winfo->error_mq_handle == NULL)
+			ereport(ERROR,
+					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					 errmsg("apply parallel worker %u failed to apply transaction %u",
+							winfo->shared->worker_id, winfo->shared->stream_xid)));
+
+		/* Wait to be signalled. */
+		(void) WaitLatch(MyLatch,
+						 WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+						 1000L,
+						 WAIT_EVENT_LOGICAL_PARALLEL_APPLY_STATE_CHANGE);
+
+		/* Reset the latch so we don't spin. */
+		ResetLatch(MyLatch);
+
+		/* An interrupt may have occurred while we were waiting. */
+		CHECK_FOR_INTERRUPTS();
+	}
+}
+
+/*
+ * Mark the current apply parallel worker as idle.
+ */
+void
+parallel_apply_set_idle(void)
+{
+	Assert(am_apply_parallel_worker());
+
+	SpinLockAcquire(&MyParallelShared->mutex);
+	MyParallelShared->in_use = false;
+	SpinLockRelease(&MyParallelShared->mutex);
+}
+
+/*
+ * Define a savepoint for a subxact in apply parallel worker if needed.
+ *
+ * The apply parallel worker can figure out if a new subtransaction was
+ * started by checking if the new change arrived with different xid. In that
+ * case define a named savepoint, so that we are able to commit/rollback it
+ * separately later.
+ */
+void
+parallel_apply_subxact_info_add(TransactionId current_xid)
+{
+	if (current_xid != stream_xid &&
+		!list_member_xid(subxactlist, (int) current_xid))
+	{
+		MemoryContext oldctx;
+		char		spname[MAXPGPATH];
+
+		parallel_apply_savepoint_name(MySubscription->oid, current_xid,
+									  spname, sizeof(spname));
+
+		elog(DEBUG1, "[Apply Parallel Worker #%u] defining savepoint %s",
+			 MyParallelShared->worker_id, spname);
+
+		/* We must be in transaction block to define the SAVEPOINT. */
+		if (!IsTransactionBlock())
+		{
+			BeginTransactionBlock();
+			CommitTransactionCommand();
+		}
+
+		DefineSavepoint(spname);
+
+		/*
+		 * CommitTransactionCommand is needed to start a subtransaction after
+		 * issuing a SAVEPOINT inside a transaction block (see
+		 * StartSubTransaction()).
+		 */
+		CommitTransactionCommand();
+
+		oldctx = MemoryContextSwitchTo(ApplyContext);
+		subxactlist = lappend_xid(subxactlist, current_xid);
+		MemoryContextSwitchTo(oldctx);
+	}
+}
+
+/*
+ * Handle STREAM ABORT message when the transaction was applied in a apply
+ * parallel worker.
+ */
+void
+parallel_apply_stream_abort(LogicalRepStreamAbortData *abort_data,
+							bool read_abort_lsn)
+{
+	TransactionId xid = abort_data->xid;
+	TransactionId subxid = abort_data->subxid;
+
+	elog(DEBUG1, "[Apply Parallel Worker #%u] aborting current transaction xid=%u, subxid=%u",
+		 MyParallelShared->worker_id , 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)
+	{
+		parallel_apply_replorigin_setup();
+
+		AbortCurrentTransaction();
+
+		if (IsTransactionBlock())
+		{
+			EndTransactionBlock(false);
+			CommitTransactionCommand();
+		}
+
+		parallel_apply_replorigin_reset();
+
+		in_remote_transaction = false;
+		pgstat_report_activity(STATE_IDLE, NULL);
+
+		list_free(subxactlist);
+		subxactlist = NIL;
+
+		parallel_apply_set_idle();
+	}
+	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];
+
+		parallel_apply_savepoint_name(MySubscription->oid, subxid, spname,
+									  sizeof(spname));
+
+		elog(DEBUG1, "[Apply Parallel Worker #%u] rolling back to savepoint %s",
+			 MyParallelShared->worker_id, spname);
+
+		for (i = list_length(subxactlist) - 1; i >= 0; i--)
+		{
+			xid = (TransactionId) list_nth_xid(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);
+	}
+}
+
+/*
+ * Form the savepoint name for the streaming transaction.
+ *
+ * Return the name in the supplied buffer.
+ */
+void
+parallel_apply_savepoint_name(Oid suboid, TransactionId xid,
+							  char *spname, int szsp)
+{
+	snprintf(spname, szsp, "pg_sp_%u_%u", suboid, xid);
+}
+
+/* Setup replication origin tracking. */
+void
+parallel_apply_replorigin_setup(void)
+{
+	RepOriginId originid;
+	char		originname[NAMEDATALEN];
+	bool		started_tx = false;
+
+	/* This function might be called inside or outside of transaction. */
+	if (!IsTransactionState())
+	{
+		StartTransactionCommand();
+		started_tx = true;
+	}
+
+	snprintf(originname, sizeof(originname), "pg_%u", MySubscription->oid);
+	originid = replorigin_by_name(originname, false);
+	replorigin_session_setup(originid);
+	replorigin_session_origin = originid;
+
+	if (started_tx)
+		CommitTransactionCommand();
+}
+
+/* Reset replication origin tracking. */
+void
+parallel_apply_replorigin_reset(void)
+{
+	bool		started_tx = false;
+
+	/* This function might be called inside or outside of transaction. */
+	if (!IsTransactionState())
+	{
+		StartTransactionCommand();
+		started_tx = true;
+	}
+
+	replorigin_session_reset();
+
+	replorigin_session_origin = InvalidRepOriginId;
+	replorigin_session_origin_lsn = InvalidXLogRecPtr;
+	replorigin_session_origin_timestamp = 0;
+
+	if (started_tx)
+		CommitTransactionCommand();
+}
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index 1667d720b1..a4376837e8 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -652,9 +652,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;
 	}
@@ -822,10 +823,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..509161d4cd 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_parallel_workers_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 whether the attach was successful.
  */
-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;
 		}
 
 		/*
@@ -209,6 +213,8 @@ WaitForReplicationWorkerAttach(LogicalRepWorker *worker,
 /*
  * Walks the workers array and searches for one that matches given
  * subscription id and relid.
+ *
+ * We are only interested in the leader apply worker or table sync worker.
  */
 LogicalRepWorker *
 logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
@@ -223,6 +229,10 @@ logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
 	{
 		LogicalRepWorker *w = &LogicalRepCtx->workers[i];
 
+		/* Skip apply parallel workers. */
+		if (isApplyParallelWorker(w))
+			continue;
+
 		if (w->in_use && w->subid == subid && w->relid == relid &&
 			(!only_running || w->proc))
 		{
@@ -259,11 +269,13 @@ logicalrep_workers_find(Oid subid, bool only_running)
 }
 
 /*
- * Start new apply background worker, if possible.
+ * Start new background worker, if possible.
+ *
+ * Returns true on success, false on failure.
  */
-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;
@@ -272,7 +284,12 @@ logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid,
 	int			slot = 0;
 	LogicalRepWorker *worker = NULL;
 	int			nsyncworkers;
+	int			napplyparallelworkers;
 	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 +367,19 @@ retry:
 	if (OidIsValid(relid) && nsyncworkers >= max_sync_workers_per_subscription)
 	{
 		LWLockRelease(LogicalRepWorkerLock);
-		return;
+		return false;
+	}
+
+	napplyparallelworkers = logicalrep_apply_parallel_worker_count(subid);
+
+	/*
+	 * Return silently if the number of apply parallel workers reached the
+	 * limit per subscription.
+	 */
+	if (is_subworker && napplyparallelworkers >= max_apply_parallel_workers_per_subscription)
+	{
+		LWLockRelease(LogicalRepWorkerLock);
+		return false;
 	}
 
 	/*
@@ -364,7 +393,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 +408,7 @@ retry:
 	worker->relstate = SUBREL_STATE_UNKNOWN;
 	worker->relstate_lsn = InvalidXLogRecPtr;
 	worker->stream_fileset = NULL;
+	worker->apply_leader_pid = is_subworker ? MyProcPid : InvalidPid;
 	worker->last_lsn = InvalidXLogRecPtr;
 	TIMESTAMP_NOBEGIN(worker->last_send_time);
 	TIMESTAMP_NOBEGIN(worker->last_recv_time);
@@ -396,10 +426,18 @@ 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, "ApplyParallelWorkerMain");
+	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 parallel worker for subscription %u", subid);
 	else
 		snprintf(bgw.bgw_name, BGW_MAXLEN,
 				 "logical replication worker for subscription %u", subid);
@@ -409,6 +447,9 @@ retry:
 	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 +462,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,19 +477,52 @@ 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)
+	if (worker)
 	{
-		LWLockRelease(LogicalRepWorkerLock);
-		return;
+		Assert(!isApplyParallelWorker(worker));
+		logicalrep_worker_stop_internal(worker);
 	}
 
+	LWLockRelease(LogicalRepWorkerLock);
+}
+
+/*
+ * Stop the logical replication worker corresponding to the input slot number,
+ * and wait until it detaches from the slot.
+ */
+void
+logicalrep_worker_stop_by_slot(int slot_no, uint16 generation)
+{
+	LogicalRepWorker *worker = &LogicalRepCtx->workers[slot_no];
+
+	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+
+	/* Return if the generation doesn't match or the worker is not alive. */
+	if (worker->generation != generation ||
+		worker->proc == NULL)
+		return;
+
+	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
 	 * is still the same one.
@@ -485,10 +559,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 +593,6 @@ logicalrep_worker_stop(Oid subid, Oid relid)
 
 		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 	}
-
-	LWLockRelease(LogicalRepWorkerLock);
 }
 
 /*
@@ -599,6 +668,29 @@ logicalrep_worker_attach(int slot)
 static void
 logicalrep_worker_detach(void)
 {
+	/*
+	 * This is the leader apply worker; stop all the apply parallel workers
+	 * previously started from here.
+	 */
+	if (!isApplyParallelWorker(MyLogicalRepWorker))
+	{
+		List	   *workers;
+		ListCell   *lc;
+
+		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+
+		workers = logicalrep_workers_find(MyLogicalRepWorker->subid, true);
+		foreach(lc, workers)
+		{
+			LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
+
+			if (isApplyParallelWorker(w))
+				logicalrep_worker_stop_internal(w);
+		}
+
+		LWLockRelease(LogicalRepWorkerLock);
+	}
+
 	/* Block concurrent access. */
 	LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
 
@@ -621,6 +713,7 @@ logicalrep_worker_cleanup(LogicalRepWorker *worker)
 	worker->userid = InvalidOid;
 	worker->subid = InvalidOid;
 	worker->relid = InvalidOid;
+	worker->apply_leader_pid = InvalidPid;
 }
 
 /*
@@ -679,6 +772,33 @@ logicalrep_sync_worker_count(Oid subid)
 	return res;
 }
 
+/*
+ * Count the number of registered (but not necessarily running) apply
+ * parallel workers for a subscription.
+ */
+int
+logicalrep_apply_parallel_worker_count(Oid subid)
+{
+	int			i;
+	int			res = 0;
+
+	Assert(LWLockHeldByMe(LogicalRepWorkerLock));
+
+	/*
+	 * Scan all attached apply parallel workers, only counting those which
+	 * have the given subscription id.
+	 */
+	for (i = 0; i < max_logical_replication_workers; i++)
+	{
+		LogicalRepWorker *w = &LogicalRepCtx->workers[i];
+
+		if (w->subid == subid && isApplyParallelWorker(w))
+			res++;
+	}
+
+	return res;
+}
+
 /*
  * ApplyLauncherShmemSize
  *		Compute space needed for replication launcher shared memory
@@ -868,7 +988,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/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 89cf9f9389..5a20c74eec 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -2839,7 +2839,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;
 
@@ -2850,6 +2851,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))
 	{
@@ -2924,7 +2927,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;
 
@@ -2935,6 +2939,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 d37d8a0d74..f38b9ecb6a 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 (am_apply_parallel_worker())
+		return;
+
 	if (am_tablesync_worker())
 		process_syncing_tables_for_sync(current_lsn);
 	else
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 5f8c541763..4526cced62 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -22,8 +22,49 @@
  * 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) Apply parallel workers
+ *
+ * If streaming = parallel, we assign a new apply parallel worker (if
+ * available) as soon as the xact's first stream is received. The leader 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.
+ *
+ * If there are enough apply parallel workers(reache half of the
+ * max_apply_parallel_workers_per_subscription) in the pool, the new apply parallel
+ * worker will be shut down at transaction end. Otherwise, it will be added to
+ * the free worker list.
+ *
+ * The leader apply worker will create separate dynamic shared memory segment
+ * when each apply parallel worker starts. The reason for this design is that
+ * we cannot count how many workers will be started. It may be possible to
+ * allocate enough shared memory in one segment based on the maximum number of
+ * apply parallel workers (max_apply_parallel_workers_per_subscription), but this
+ * may waste some memory if no process is actually started.
+ *
+ * The dynamic shared memory segment will contain (1) a shm_mq that can be used
+ * to transport errors (and other messages reported via elog/ereport) from the
+ * apply parallel worker to leader apply worker (2) another shm_mq that can
+ * be used to transport changes in the transaction from leader apply worker to
+ * apply parallel worker (3) necessary information to be shared among apply
+ * parallel workers to leader apply worker(i.e. protocol version, streaming
+ * transaction ID, in_use flag and the corresponding LogicalRepWorker slot
+ * information).
+ *
+ * 2) Write to temporary files and apply when the final commit arrives
+ *
+ * If no worker is available to handle the 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 +260,20 @@ typedef struct ApplyExecutionData
 	PartitionTupleRouting *proute;	/* partition routing info */
 } ApplyExecutionData;
 
-/* Struct for saving and restoring apply errcontext information */
-typedef struct ApplyErrorCallbackArg
+/* What action to take for the transaction. */
+typedef enum
 {
-	LogicalRepMsgType command;	/* 0 if invalid */
-	LogicalRepRelMapEntry *rel;
+	/* The action for non-streaming transactions. */
+	TA_APPLY_IN_LEADER_WORKER,
 
-	/* Remote node information */
-	int			remote_attnum;	/* -1 if invalid */
-	TransactionId remote_xid;
-	XLogRecPtr	finish_lsn;
-	char	   *origin_name;
-} ApplyErrorCallbackArg;
+	/* Actions for streaming transactions. */
+	TA_SERIALIZE_TO_FILE,
+	TA_APPLY_IN_PARALLEL_WORKER,
+	TA_SEND_TO_PARALLEL_WORKER
+} TransactionApplyAction;
 
-static ApplyErrorCallbackArg apply_error_callback_arg =
+/* errcontext tracker */
+ApplyErrorCallbackArg apply_error_callback_arg =
 {
 	.command = 0,
 	.rel = NULL,
@@ -242,7 +283,9 @@ static ApplyErrorCallbackArg apply_error_callback_arg =
 	.origin_name = NULL,
 };
 
-static MemoryContext ApplyMessageContext = NULL;
+ErrorContextCallback	*apply_error_context_stack = NULL;
+
+MemoryContext ApplyMessageContext = NULL;
 MemoryContext ApplyContext = NULL;
 
 /* per stream context for streaming transactions */
@@ -251,27 +294,35 @@ 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;
 
-static TransactionId stream_xid = InvalidTransactionId;
+TransactionId stream_xid = InvalidTransactionId;
+
+/*
+ * The number of changes sent to apply parallel workers during one streaming
+ * block.
+ */
+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 parallel 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 +375,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,28 +407,52 @@ 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);
 
+static TransactionApplyAction get_transaction_apply_action(TransactionId xid,
+														   ApplyParallelWorkerInfo **winfo);
+
 /*
  * Should this worker apply changes for given relation.
  *
  * This is mainly needed for initial relation data sync as that runs in
  * separate worker process running in parallel and we need some way to skip
- * changes coming to the main apply worker during the sync of a table.
+ * changes coming to the leader apply worker during the sync of a table.
  *
  * Note we need to do smaller or equals comparison for SYNCDONE state because
  * it might hold position of end of initial slot consistent point WAL
  * record + 1 (ie start of next record) and next record can be COMMIT of
  * transaction we are now processing (which is what we set remote_final_lsn
  * to in apply_handle_begin).
+ *
+ * Note that for streaming transactions that are being applied in apply
+ * parallel worker, we disallow applying changes on a table that is not in
+ * the READY state, because we cannot decide whether to apply the change as we
+ * won't know remote_final_lsn by that time.
+ *
+ * We already checked this in parallel_apply_can_start() before assigning the
+ * streaming transaction to the parallel worker, but it also needs to be
+ * checked here because if the user executes ALTER SUBSCRIPTION ... REFRESH
+ * PUBLICATION in parallel, the new table can be added to pg_subscription_rel
+ * in parallel while applying this transaction.
  */
 static bool
 should_apply_changes_for_rel(LogicalRepRelMapEntry *rel)
 {
 	if (am_tablesync_worker())
 		return MyLogicalRepWorker->relid == rel->localreloid;
+	else if (am_apply_parallel_worker())
+	{
+		if (rel->state != SUBREL_STATE_READY)
+			ereport(ERROR,
+					(errmsg("logical replication apply workers for subscription \"%s\" will restart",
+							MySubscription->name),
+					 errdetail("Cannot handle streamed replication transaction by apply "
+							   "parallel workers until all tables are synchronized.")));
+
+		return true;
+	}
 	else
 		return (rel->state == SUBREL_STATE_READY ||
 				(rel->state == SUBREL_STATE_SYNCDONE &&
@@ -426,43 +498,90 @@ end_replication_step(void)
 }
 
 /*
- * Handle streamed transactions.
+ * Handle streamed transactions for both the leader apply worker and the apply
+ * parallel 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
+ * parallel workers (LOGICAL_REP_MSG_RELATION or LOGICAL_REP_MSG_TYPE changes
+ * will be applied by both leader apply worker and apply parallel workers).
  *
- * 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 leader apply worker, false
+ * otherwise.
  *
- * Returns true for streamed transactions, false otherwise (regular mode).
+ * Exception: When the leader 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;
+	ApplyParallelWorkerInfo *winfo;
+	TransactionApplyAction	apply_action;
+
+	apply_action = get_transaction_apply_action(stream_xid, &winfo);
 
 	/* not in streaming mode */
-	if (!in_streamed_transaction)
+	if (apply_action == TA_APPLY_IN_LEADER_WORKER)
 		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 (apply_action == TA_APPLY_IN_PARALLEL_WORKER)
+	{
+		/* Define a savepoint for a subxact if needed. */
+		parallel_apply_subxact_info_add(current_xid);
+
+		return false;
+	}
+	else if (apply_action == TA_SEND_TO_PARALLEL_WORKER)
+	{
+		parallel_apply_send_data(winfo, 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 leader apply worker here. See function
+		 * cleanup_rel_sync_cache.
+		 */
+		if (action == LOGICAL_REP_MSG_RELATION ||
+			action == LOGICAL_REP_MSG_TYPE)
+			return false;
+
+		return true;
+	}
+	else if (apply_action == TA_SERIALIZE_TO_FILE)
+	{
+		Assert(stream_fd != NULL);
 
-	/* write the change to the current file */
-	stream_write_change(action, s);
+		/* Add the new subxact to the array (unless already there). */
+		subxact_info_add(current_xid);
 
-	return true;
+		/* write the change to the current file */
+		stream_write_change(action, s);
+
+		return true;
+	}
+	else
+	{
+		Assert(false);
+		return false;	/* silence compiler warning */
+	}
 }
 
 /*
@@ -898,8 +1017,11 @@ apply_handle_prepare_internal(LogicalRepPreparedTxnData *prepare_data)
 	 * BeginTransactionBlock is necessary to balance the EndTransactionBlock
 	 * called within the PrepareTransactionBlock below.
 	 */
-	BeginTransactionBlock();
-	CommitTransactionCommand(); /* Completes the preceding Begin command. */
+	if (!IsTransactionBlock())
+	{
+		BeginTransactionBlock();
+		CommitTransactionCommand(); /* Completes the preceding Begin command. */
+	}
 
 	/*
 	 * Update origin state so we can restart streaming from correct position
@@ -1064,15 +1186,13 @@ 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)
 {
 	LogicalRepPreparedTxnData prepare_data;
+	ApplyParallelWorkerInfo *winfo;
+	TransactionApplyAction	apply_action;
 
 	if (in_streamed_transaction)
 		ereport(ERROR,
@@ -1088,24 +1208,70 @@ 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);
+	apply_action = get_transaction_apply_action(prepare_data.xid, &winfo);
 
-	/* Replay all the spooled operations. */
-	apply_spooled_messages(prepare_data.xid, prepare_data.prepare_lsn);
+	if (apply_action == TA_APPLY_IN_PARALLEL_WORKER)
+	{
+		parallel_apply_replorigin_setup();
 
-	/* Mark the transaction as prepared. */
-	apply_handle_prepare_internal(&prepare_data);
+		/* Mark the transaction as prepared. */
+		apply_handle_prepare_internal(&prepare_data);
 
-	CommitTransactionCommand();
+		CommitTransactionCommand();
 
-	pgstat_report_stat(false);
+		parallel_apply_replorigin_reset();
 
-	store_flush_position(prepare_data.end_lsn);
+		list_free(subxactlist);
+		subxactlist = NIL;
 
-	in_remote_transaction = false;
+		parallel_apply_set_idle();
+	}
+	else if (apply_action == TA_SEND_TO_PARALLEL_WORKER)
+	{
+		parallel_apply_replorigin_reset();
+
+		/* Send STREAM PREPARE message to the apply parallel worker. */
+		parallel_apply_send_data(winfo, s->len, s->data);
+
+		/*
+		 * After sending the data to the apply parallel worker, wait for
+		 * that worker to finish. This is necessary to maintain commit
+		 * order which avoids failures due to transaction dependencies and
+		 * deadlocks.
+		 */
+		parallel_apply_wait_for_free(winfo);
+		parallel_apply_replorigin_setup();
+		parallel_apply_free_worker(winfo);
 
-	/* unlink the files with serialized changes and subxact info. */
-	stream_cleanup_files(MyLogicalRepWorker->subid, prepare_data.xid);
+		elog(DEBUG1, "streamed transaction %u was prepared using an apply parallel worker",
+			 prepare_data.xid);
+
+		store_flush_position(prepare_data.end_lsn);
+	}
+	else if (apply_action == TA_SERIALIZE_TO_FILE)
+	{
+		/*
+		 * 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();
+
+		store_flush_position(prepare_data.end_lsn);
+
+		/* Unlink the files with serialized changes and subxact info. */
+		stream_cleanup_files(MyLogicalRepWorker->subid, prepare_data.xid);
+	}
+	else
+		Assert(false);
+
+	pgstat_report_stat(false);
+
+	in_remote_transaction = false;
 
 	/* Process any tables that are being synchronized in parallel. */
 	process_syncing_tables(prepare_data.end_lsn);
@@ -1134,7 +1300,7 @@ apply_handle_origin(StringInfo s)
 	 * ORIGIN message can only come inside streaming transaction or inside
 	 * remote transaction and before any actual writes.
 	 */
-	if (!in_streamed_transaction &&
+	if (!in_streamed_transaction && stream_apply_worker == NULL &&
 		(!in_remote_transaction ||
 		 (IsTransactionState() && !am_tablesync_worker())))
 		ereport(ERROR,
@@ -1149,24 +1315,14 @@ static void
 apply_handle_stream_start(StringInfo s)
 {
 	bool		first_segment;
+	ApplyParallelWorkerInfo *winfo;
+	TransactionApplyAction apply_action;
 
 	if (in_streamed_transaction)
 		ereport(ERROR,
 				(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;
-
 	/* extract XID of the top-level transaction */
 	stream_xid = logicalrep_read_stream_start(s, &first_segment);
 
@@ -1177,36 +1333,105 @@ 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)
+	apply_action = get_transaction_apply_action(stream_xid, &winfo);
+
+	if (apply_action == TA_APPLY_IN_PARALLEL_WORKER)
 	{
-		MemoryContext oldctx;
+		/*
+		 * Make sure the handle apply_dispatch methods are aware we're in a
+		 * remote transaction.
+		 */
+		in_remote_transaction = true;
+	}
+	else if (apply_action == TA_SEND_TO_PARALLEL_WORKER)
+	{
+		Assert(!first_segment);
 
-		oldctx = MemoryContextSwitchTo(ApplyContext);
+		parallel_apply_send_data(winfo, s->len, s->data);
 
-		MyLogicalRepWorker->stream_fileset = palloc(sizeof(FileSet));
-		FileSetInit(MyLogicalRepWorker->stream_fileset);
+		nchanges = 0;
+		elog(DEBUG1, "starting streaming of xid %u", stream_xid);
 
-		MemoryContextSwitchTo(oldctx);
+		/* Cache the apply parallel worker for this transaction. */
+		stream_apply_worker = winfo;
 	}
+	else if (apply_action == TA_SERIALIZE_TO_FILE)
+	{
+		/*
+		 * For the first stream start, check if there is any free apply
+		 * parallel worker we can use to process this transaction.
+		 */
+		if (first_segment)
+			winfo = parallel_apply_start_worker(stream_xid);
 
-	/* open the spool file for this transaction */
-	stream_open_file(MyLogicalRepWorker->subid, stream_xid, first_segment);
+		if (winfo)
+		{
+			/*
+			 * If we have found a free worker, then we pass the data to that
+			 * worker.
+			 */
+			parallel_apply_send_data(winfo, 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;
 
-	pgstat_report_activity(STATE_RUNNING, NULL);
+			/* Cache the apply parallel worker for this transaction. */
+			stream_apply_worker = winfo;
+		}
+		else
+		{
+			/*
+			 * Notify handle methods we're processing a remote in-progress
+			 * transaction.
+			 */
+			in_streamed_transaction = true;
 
-	end_replication_step();
+			/*
+			 * Since no apply parallel 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();
+
+			/*
+			 * 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;
+
+				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();
+		}
+	}
+	else
+		Assert(false);
+
+	pgstat_report_activity(STATE_RUNNING, NULL);
 }
 
 /*
@@ -1215,58 +1440,74 @@ apply_handle_stream_start(StringInfo s)
 static void
 apply_handle_stream_stop(StringInfo s)
 {
-	if (!in_streamed_transaction)
-		ereport(ERROR,
-				(errcode(ERRCODE_PROTOCOL_VIOLATION),
-				 errmsg_internal("STREAM STOP message without STREAM START")));
+	ApplyParallelWorkerInfo *winfo;
+	TransactionApplyAction apply_action;
 
-	/*
-	 * 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();
+	apply_action = get_transaction_apply_action(stream_xid, &winfo);
 
-	/* We must be in a valid transaction state */
-	Assert(IsTransactionState());
+	if (apply_action == TA_APPLY_IN_PARALLEL_WORKER)
+	{
+		elog(DEBUG1, "[Apply Parallel Worker #%u] ended processing streaming chunk, "
+			 "waiting on shm_mq_receive", MyParallelShared->worker_id);
 
-	/* Commit the per-stream transaction */
-	CommitTransactionCommand();
+		pgstat_report_activity(STATE_IDLEINTRANSACTION, NULL);
+	}
+	else if (apply_action == TA_SEND_TO_PARALLEL_WORKER)
+	{
+		parallel_apply_send_data(winfo, s->len, s->data);
 
-	in_streamed_transaction = false;
+		elog(DEBUG1, "stopped streaming of xid %u, %u changes streamed",
+			 stream_xid, nchanges);
 
-	/* Reset per-stream context */
-	MemoryContextReset(LogicalStreamingContext);
+		stream_apply_worker = NULL;
+
+		pgstat_report_activity(STATE_IDLE, NULL);
+	}
+	else if (apply_action == TA_SERIALIZE_TO_FILE)
+	{
+		if (!in_streamed_transaction)
+			ereport(ERROR,
+					(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();
+
+		/* We must be in a valid transaction state */
+		Assert(IsTransactionState());
+
+		/* Commit the per-stream transaction */
+		CommitTransactionCommand();
+
+		/* Reset per-stream context */
+		MemoryContextReset(LogicalStreamingContext);
+
+		pgstat_report_activity(STATE_IDLE, NULL);
+	}
+	else
+		Assert(false);
+
+	in_streamed_transaction = false;
 
-	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 +1531,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 +1555,6 @@ apply_handle_stream_abort(StringInfo s)
 			cleanup_subxact_info();
 			end_replication_step();
 			CommitTransactionCommand();
-			reset_apply_error_context_info();
 			return;
 		}
 
@@ -1339,6 +1577,76 @@ 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;
+	ApplyParallelWorkerInfo *winfo;
+	TransactionApplyAction apply_action;
+
+	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_parallel_worker())
+		read_abort_lsn = MyParallelShared->proto_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);
+
+	apply_action = get_transaction_apply_action(xid, &winfo);
+
+	if (apply_action == TA_APPLY_IN_PARALLEL_WORKER)
+		parallel_apply_stream_abort(&abort_data, read_abort_lsn);
+	else if (apply_action == TA_SEND_TO_PARALLEL_WORKER)
+	{
+		if (subxid == xid)
+			parallel_apply_replorigin_reset();
+
+		/* Send STREAM ABORT message to the apply parallel worker. */
+		parallel_apply_send_data(winfo, s->len, s->data);
+
+		/*
+		 * After sending the data to the apply parallel 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)
+		{
+			parallel_apply_wait_for_free(winfo);
+			parallel_apply_replorigin_setup();
+			parallel_apply_free_worker(winfo);
+
+			elog(DEBUG1, "streamed transaction %u was aborted using an apply parallel worker",
+				 xid);
+		}
+	}
+	else if (apply_action == TA_SERIALIZE_TO_FILE)
+	{
+		/*
+		 * We are in leader apply worker and the transaction has been
+		 * serialized to file.
+		 */
+		serialize_stream_abort(xid, subxid);
+	}
+	else
+		Assert(false);
 
 	reset_apply_error_context_info();
 }
@@ -1470,6 +1778,8 @@ apply_handle_stream_commit(StringInfo s)
 {
 	TransactionId xid;
 	LogicalRepCommitData commit_data;
+	ApplyParallelWorkerInfo *winfo;
+	TransactionApplyAction apply_action;
 
 	if (in_streamed_transaction)
 		ereport(ERROR,
@@ -1479,14 +1789,67 @@ 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);
+	apply_action = get_transaction_apply_action(xid, &winfo);
+
+	if (apply_action == TA_APPLY_IN_PARALLEL_WORKER)
+	{
+		parallel_apply_replorigin_setup();
 
-	apply_spooled_messages(xid, commit_data.commit_lsn);
+		apply_handle_commit_internal(&commit_data);
 
-	apply_handle_commit_internal(&commit_data);
+		parallel_apply_replorigin_reset();
+
+		list_free(subxactlist);
+		subxactlist = NIL;
+
+		parallel_apply_set_idle();
+	}
+	else if (apply_action == TA_SEND_TO_PARALLEL_WORKER)
+	{
+		parallel_apply_replorigin_reset();
+
+		/* Send STREAM COMMIT message to the apply parallel worker. */
+		parallel_apply_send_data(winfo, s->len, s->data);
+
+		/*
+		 * After sending the data to the apply parallel worker, wait for
+		 * that worker to finish. This is necessary to maintain commit
+		 * order which avoids failures due to transaction dependencies and
+		 * deadlocks.
+		 */
+		parallel_apply_wait_for_free(winfo);
+		parallel_apply_replorigin_setup();
 
-	/* 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();
+
+		parallel_apply_free_worker(winfo);
+
+		elog(DEBUG1, "streamed transaction %u was committed using an apply parallel worker",
+			 xid);
+
+		/*
+		 * The transaction is either non-empty or skipped, so we clear the
+		 * subskiplsn.
+		 */
+		clear_subscription_skip_lsn(commit_data.commit_lsn);
+	}
+	else if (apply_action == TA_SERIALIZE_TO_FILE)
+	{
+		/*
+		 * 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);
+	}
+	else
+		Assert(false);
 
 	/* Process any tables that are being synchronized in parallel. */
 	process_syncing_tables(commit_data.end_lsn);
@@ -1530,6 +1893,13 @@ apply_handle_commit_internal(LogicalRepCommitData *commit_data)
 		replorigin_session_origin_timestamp = commit_data->committime;
 
 		CommitTransactionCommand();
+
+		if (IsTransactionBlock())
+		{
+			EndTransactionBlock(false);
+			CommitTransactionCommand();
+		}
+
 		pgstat_report_stat(false);
 
 		store_flush_position(commit_data->end_lsn);
@@ -2467,7 +2837,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 +3006,10 @@ store_flush_position(XLogRecPtr remote_lsn)
 {
 	FlushPosition *flushpos;
 
+	/* Skip if not the leader apply worker */
+	if (am_apply_parallel_worker())
+		return;
+
 	/* Need to do this in permanent context */
 	MemoryContextSwitchTo(ApplyContext);
 
@@ -2650,7 +3024,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;
@@ -2700,6 +3074,7 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 	errcallback.callback = apply_error_callback;
 	errcallback.previous = error_context_stack;
 	error_context_stack = &errcallback;
+	apply_error_context_stack = error_context_stack;
 
 	/* This outer loop iterates once per wait. */
 	for (;;)
@@ -2914,6 +3289,7 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 
 	/* Pop the error context stack */
 	error_context_stack = errcallback.previous;
+	apply_error_context_stack = error_context_stack;
 
 	/* All done */
 	walrcv_endstreaming(LogRepWorkerWalRcvConn, &tli);
@@ -3114,7 +3490,7 @@ maybe_reread_subscription(void)
 /*
  * Callback from subscription syscache invalidation.
  */
-static void
+void
 subscription_change_cb(Datum arg, int cacheid, uint32 hashvalue)
 {
 	MySubscriptionValid = false;
@@ -3569,37 +3945,14 @@ start_apply(XLogRecPtr origin_startpos)
 	PG_END_TRY();
 }
 
-/* Logical Replication Apply worker entry point */
+/*
+ * Initialize the database connection, in-memory subscription and necessary
+ * config options.
+ */
 void
-ApplyWorkerMain(Datum main_arg)
+InitializeApplyWorker(void)
 {
-	int			worker_slot = DatumGetInt32(main_arg);
 	MemoryContext oldctx;
-	char		originname[NAMEDATALEN];
-	XLogRecPtr	origin_startpos = InvalidXLogRecPtr;
-	char	   *myslotname = NULL;
-	WalRcvStreamOptions options;
-	int			server_version;
-
-	/* Attach to slot */
-	logicalrep_worker_attach(worker_slot);
-
-	/* Setup signal handling */
-	pqsignal(SIGHUP, SignalHandlerForConfigReload);
-	pqsignal(SIGTERM, die);
-	BackgroundWorkerUnblockSignals();
-
-	/*
-	 * We don't currently need any ResourceOwner in a walreceiver process, but
-	 * if we did, we could call CreateAuxProcessResourceOwner here.
-	 */
-
-	/* Initialise stats to a sanish value */
-	MyLogicalRepWorker->last_send_time = MyLogicalRepWorker->last_recv_time =
-		MyLogicalRepWorker->reply_time = GetCurrentTimestamp();
-
-	/* Load the libpq-specific functions */
-	load_file("libpqwalreceiver", false);
 
 	/* Run as replica session replication role. */
 	SetConfigOption("session_replication_role", "replica",
@@ -3659,12 +4012,50 @@ ApplyWorkerMain(Datum main_arg)
 		ereport(LOG,
 				(errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has started",
 						MySubscription->name, get_rel_name(MyLogicalRepWorker->relid))));
+	else if (am_apply_parallel_worker())
+		ereport(LOG,
+				(errmsg("logical replication apply parallel worker for subscription \"%s\" has started",
+						MySubscription->name)));
 	else
 		ereport(LOG,
 				(errmsg("logical replication apply worker for subscription \"%s\" has started",
 						MySubscription->name)));
 
 	CommitTransactionCommand();
+}
+
+/* Logical Replication Apply worker entry point */
+void
+ApplyWorkerMain(Datum main_arg)
+{
+	int			worker_slot = DatumGetInt32(main_arg);
+	char		originname[NAMEDATALEN];
+	XLogRecPtr	origin_startpos = InvalidXLogRecPtr;
+	char	   *myslotname = NULL;
+	WalRcvStreamOptions options;
+	int			server_version;
+
+	/* Attach to slot */
+	logicalrep_worker_attach(worker_slot);
+
+	/* Setup signal handling */
+	pqsignal(SIGHUP, SignalHandlerForConfigReload);
+	pqsignal(SIGTERM, die);
+	BackgroundWorkerUnblockSignals();
+
+	/*
+	 * We don't currently need any ResourceOwner in a walreceiver process, but
+	 * if we did, we could call CreateAuxProcessResourceOwner here.
+	 */
+
+	/* Initialise stats to a sanish value */
+	MyLogicalRepWorker->last_send_time = MyLogicalRepWorker->last_recv_time =
+		MyLogicalRepWorker->reply_time = GetCurrentTimestamp();
+
+	/* Load the libpq-specific functions */
+	load_file("libpqwalreceiver", false);
+
+	InitializeApplyWorker();
 
 	/* Connect to the origin and start the replication. */
 	elog(DEBUG1, "connecting to publisher using connection string \"%s\"",
@@ -3687,7 +4078,7 @@ ApplyWorkerMain(Datum main_arg)
 	}
 	else
 	{
-		/* This is main apply worker */
+		/* This is leader apply worker */
 		RepOriginId originid;
 		TimeLineID	startpointTLI;
 		char	   *err;
@@ -3751,13 +4142,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;
 	options.proto.logical.origin = pstrdup(MySubscription->origin);
 
@@ -3854,6 +4246,15 @@ IsLogicalWorker(void)
 	return MyLogicalRepWorker != NULL;
 }
 
+/*
+ * Is current process a logical replication apply parallel worker?
+ */
+bool
+IsLogicalApplyParallelWorker(void)
+{
+	return am_apply_parallel_worker();
+}
+
 /*
  * Start skipping changes of the transaction if the given LSN matches the
  * LSN specified by subscription's skiplsn.
@@ -3916,7 +4317,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_parallel_worker())
 		return;
 
 	if (!IsTransactionState())
@@ -3988,7 +4390,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;
@@ -4016,23 +4418,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 */
@@ -4052,3 +4478,25 @@ reset_apply_error_context_info(void)
 	apply_error_callback_arg.remote_attnum = -1;
 	set_apply_error_context_xact(InvalidTransactionId, InvalidXLogRecPtr);
 }
+
+/*
+ * Return the action to take for the given transaction. Also return the apply
+ * parallel worker information if the action is TA_SEND_TO_PARALLEL_WORKER.
+ */
+static TransactionApplyAction
+get_transaction_apply_action(TransactionId xid, ApplyParallelWorkerInfo **winfo)
+{
+	if (am_apply_parallel_worker())
+		return TA_APPLY_IN_PARALLEL_WORKER;
+	else if (in_remote_transaction)
+		return TA_APPLY_IN_LEADER_WORKER;
+
+	/*
+	 * Check if we are processing this transaction using an apply parallel
+	 * worker and if so, send the changes to that worker.
+	 */
+	else if ((*winfo = parallel_apply_find_worker(xid)))
+		return TA_SEND_TO_PARALLEL_WORKER;
+	else
+		return TA_SERIALIZE_TO_FILE;
+}
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 62e0ffecd8..d3f19b805f 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -1843,6 +1843,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
@@ -1856,7 +1859,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/storage/ipc/procsignal.c b/src/backend/storage/ipc/procsignal.c
index 21a9fc0fdd..9cd54287b6 100644
--- a/src/backend/storage/ipc/procsignal.c
+++ b/src/backend/storage/ipc/procsignal.c
@@ -22,6 +22,7 @@
 #include "commands/async.h"
 #include "miscadmin.h"
 #include "pgstat.h"
+#include "replication/logicalworker.h"
 #include "replication/walsender.h"
 #include "storage/condition_variable.h"
 #include "storage/ipc.h"
@@ -657,6 +658,9 @@ procsignal_sigusr1_handler(SIGNAL_ARGS)
 	if (CheckProcSignal(PROCSIG_LOG_MEMORY_CONTEXT))
 		HandleLogMemoryContextInterrupt();
 
+	if (CheckProcSignal(PROCSIG_PARALLEL_APPLY_MESSAGE))
+		HandleParallelApplyMessageInterrupt();
+
 	if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_DATABASE))
 		RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_DATABASE);
 
diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c
index 7bec4e4ff5..fb7bd98091 100644
--- a/src/backend/tcop/postgres.c
+++ b/src/backend/tcop/postgres.c
@@ -3377,6 +3377,9 @@ ProcessInterrupts(void)
 
 	if (LogMemoryContextPending)
 		ProcessLogMemoryContextInterrupt();
+
+	if (ParallelApplyMessagePending)
+		HandleParallelApplyMessages();
 }
 
 /*
diff --git a/src/backend/utils/activity/wait_event.c b/src/backend/utils/activity/wait_event.c
index 92f24a6c9b..c96647c4ed 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_PARALLEL_APPLY_STATE_CHANGE:
+			event_name = "LogicalParallelApplyStateChange";
+			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 9fbbfb1be5..0ff21c57c8 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -3222,6 +3222,18 @@ static struct config_int ConfigureNamesInt[] =
 		NULL, NULL, NULL
 	},
 
+	{
+		{"max_apply_parallel_workers_per_subscription",
+			PGC_SIGHUP,
+			REPLICATION_SUBSCRIBERS,
+			gettext_noop("Maximum number of apply parallel workers per subscription."),
+			NULL,
+		},
+		&max_apply_parallel_workers_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 90bec0502c..7fd548435f 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_parallel_workers_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 d25709ad5f..13c8e10b71 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -4480,7 +4480,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,
@@ -4617,8 +4617,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 7b98714f30..232e66b998 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -80,7 +80,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 */
 
@@ -124,7 +125,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
@@ -137,6 +139,21 @@ typedef struct Subscription
 								 * specified origin */
 } 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 an apply
+ * parallel 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/nodes/pg_list.h b/src/include/nodes/pg_list.h
index a6e04d04a3..c4253a772f 100644
--- a/src/include/nodes/pg_list.h
+++ b/src/include/nodes/pg_list.h
@@ -322,6 +322,17 @@ list_nth_oid(const List *list, int n)
 	return lfirst_oid(list_nth_cell(list, n));
 }
 
+/*
+ * Return the TransactionId value contained in the n'th element of the
+ * specified list.
+ */
+static inline TransactionId
+list_nth_xid(const List *list, int n)
+{
+	Assert(IsA(list, XidList));
+	return lfirst_xid(list_nth_cell(list, n));
+}
+
 #define list_nth_node(type,list,n)	castNode(type, list_nth(list, n))
 
 /*
diff --git a/src/include/replication/logicallauncher.h b/src/include/replication/logicallauncher.h
index f1e2821e25..cbea3de371 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_parallel_workers_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..20b9abbaa3 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 parallel
+ * 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..a4e16fa460 100644
--- a/src/include/replication/logicalworker.h
+++ b/src/include/replication/logicalworker.h
@@ -12,8 +12,14 @@
 #ifndef LOGICALWORKER_H
 #define LOGICALWORKER_H
 
+extern PGDLLIMPORT volatile bool ParallelApplyMessagePending;
+
 extern void ApplyWorkerMain(Datum main_arg);
+extern void ApplyParallelWorkerMain(Datum main_arg);
 
 extern bool IsLogicalWorker(void);
+extern bool IsLogicalApplyParallelWorker(void);
+extern void HandleParallelApplyMessageInterrupt(void);
+extern void HandleParallelApplyMessages(void);
 
 #endif							/* LOGICALWORKER_H */
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index 8695901ba7..a74b616eff 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;
 
 	/*
@@ -658,9 +659,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..09ef4919ef 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -17,8 +17,12 @@
 #include "access/xlogdefs.h"
 #include "catalog/pg_subscription.h"
 #include "datatype/timestamp.h"
+#include "miscadmin.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 +64,12 @@ typedef struct LogicalRepWorker
 	 */
 	FileSet    *stream_fileset;
 
+	/*
+	 * PID of leader apply worker if this slot is used for an apply parallel
+	 * worker, InvalidPid otherwise.
+	 */
+	pid_t			apply_leader_pid;
+
 	/* Stats. */
 	XLogRecPtr	last_lsn;
 	TimestampTz last_send_time;
@@ -68,8 +78,74 @@ 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;
+
+/*
+ * Struct for sharing information between apply leader apply worker and apply
+ * parallel workers.
+ */
+typedef struct ApplyParallelWorkerShared
+{
+	slock_t	mutex;
+
+	bool	in_use;
+
+	/* Logical protocol version. */
+	uint32	proto_version;
+
+	TransactionId	stream_xid;
+
+	/* ID of apply parallel worker */
+	uint32	worker_id;
+
+	/* Information from the corresponding LogicalRepWorker slot */
+	uint16					logicalrep_worker_generation;
+
+	int						logicalrep_worker_slot_no;
+} ApplyParallelWorkerShared;
+
+/*
+ * Information which is used to manage the apply parallel worker.
+ */
+typedef struct ApplyParallelWorkerInfo
+{
+	shm_mq_handle			*mq_handle;
+
+	/*
+	 * The queue used to transfer messages from the apply parallel worker to
+	 * the leader apply worker. NULL if the apply parallel worker exited
+	 * cleanly.
+	 */
+	shm_mq_handle			*error_mq_handle;
+
+	dsm_segment				*dsm_seg;
+
+	ApplyParallelWorkerShared volatile	*shared;
+} ApplyParallelWorkerInfo;
+
 /* Main memory context for apply worker. Permanent during worker lifetime. */
 extern PGDLLIMPORT MemoryContext ApplyContext;
+extern PGDLLIMPORT MemoryContext ApplyMessageContext;
+
+extern PGDLLIMPORT ErrorContextCallback	*apply_error_context_stack;
+extern PGDLLIMPORT ApplyErrorCallbackArg apply_error_callback_arg;
+
+extern PGDLLIMPORT bool MySubscriptionValid;
+
+extern PGDLLIMPORT volatile ApplyParallelWorkerShared *MyParallelShared;
+
+extern PGDLLIMPORT List *subxactlist;
 
 /* libpqreceiver connection */
 extern PGDLLIMPORT struct WalReceiverConn *LogRepWorkerWalRcvConn;
@@ -79,18 +155,24 @@ 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 PGDLLIMPORT ApplyParallelWorkerInfo *stream_apply_worker;
 
 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_stop_by_slot(int slot_no, uint16 generation);
 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_parallel_worker_count(Oid subid);
 
 extern void ReplicationOriginNameForTablesync(Oid suboid, Oid relid,
 											  char *originname, int szorgname);
@@ -103,10 +185,47 @@ 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);
+
+extern void InitializeApplyWorker(void);
+
+/* Function for apply error callback */
+extern void apply_error_callback(void *arg);
+
+extern void subscription_change_cb(Datum arg, int cacheid, uint32 hashvalue);
+
+/* Apply parallel worker setup and interactions */
+extern ApplyParallelWorkerInfo *parallel_apply_start_worker(TransactionId xid);
+extern ApplyParallelWorkerInfo *parallel_apply_find_worker(TransactionId xid);
+extern void parallel_apply_set_idle(void);
+extern void parallel_apply_free_worker(ApplyParallelWorkerInfo *winfo);
+extern void parallel_apply_wait_for_free(ApplyParallelWorkerInfo *winfo);
+extern void parallel_apply_send_data(ApplyParallelWorkerInfo *winfo, Size nbytes,
+									 const void *data);
+
+extern void parallel_apply_subxact_info_add(TransactionId current_xid);
+extern void parallel_apply_stream_abort(LogicalRepStreamAbortData *abort_data,
+										bool read_abort_lsn);
+extern void parallel_apply_savepoint_name(Oid suboid, Oid relid,
+										  char *spname, int szsp);
+extern void parallel_apply_replorigin_setup(void);
+extern void parallel_apply_replorigin_reset(void);
+
+#define isApplyParallelWorker(worker) (worker->apply_leader_pid != InvalidPid)
+
 static inline bool
 am_tablesync_worker(void)
 {
 	return OidIsValid(MyLogicalRepWorker->relid);
 }
 
+static inline bool
+am_apply_parallel_worker(void)
+{
+	return isApplyParallelWorker(MyLogicalRepWorker);
+}
+
 #endif							/* WORKER_INTERNAL_H */
diff --git a/src/include/storage/procsignal.h b/src/include/storage/procsignal.h
index ee636900f3..f7562d7430 100644
--- a/src/include/storage/procsignal.h
+++ b/src/include/storage/procsignal.h
@@ -35,6 +35,7 @@ typedef enum
 	PROCSIG_WALSND_INIT_STOPPING,	/* ask walsenders to prepare for shutdown  */
 	PROCSIG_BARRIER,			/* global barrier interrupt  */
 	PROCSIG_LOG_MEMORY_CONTEXT, /* ask backend to log the memory contexts */
+	PROCSIG_PARALLEL_APPLY_MESSAGE,	/* Message from apply parallel workers */
 
 	/* Recovery conflict reasons */
 	PROCSIG_RECOVERY_CONFLICT_DATABASE,
diff --git a/src/include/utils/wait_event.h b/src/include/utils/wait_event.h
index 6f2d5612e0..b2cdd554e1 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_PARALLEL_APPLY_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 ef0ebf96b9..99b9e867e4 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -217,9 +217,9 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 (1 row)
 
 DROP SUBSCRIPTION regress_testsub;
--- fail - streaming must be boolean
+-- fail - streaming must be boolean or 'parallel'
 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
@@ -230,6 +230,14 @@ WARNING:  tables were not subscribed, you will have to run ALTER SUBSCRIPTION ..
  regress_testsub | regress_subscription_user | f       | {testpub}   | f      | t         | d                | f                | any    | off                | dbname=regress_doesnotexist | 0/0
 (1 row)
 
+ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel);
+\dRs+
+                                                                                         List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Synchronous commit |          Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | p         | d                | f                | any    | off                | dbname=regress_doesnotexist | 0/0
+(1 row)
+
 ALTER SUBSCRIPTION regress_testsub SET (streaming = false);
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 \dRs+
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index 4425fafc46..ed148e4bd4 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -156,7 +156,7 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 
 DROP SUBSCRIPTION regress_testsub;
 
--- fail - streaming must be boolean
+-- fail - streaming must be boolean or 'parallel'
 CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, streaming = foo);
 
 -- now it works
@@ -164,6 +164,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 
 \dRs+
 
+ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel);
+
+\dRs+
+
 ALTER SUBSCRIPTION regress_testsub SET (streaming = false);
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 8ad112c44d..4fa3f74766 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -118,6 +118,9 @@ AppendRelInfo
 AppendState
 ApplyErrorCallbackArg
 ApplyExecutionData
+ApplyParallelWorkerEntry
+ApplyParallelWorkerInfo
+ApplyParallelWorkerShared
 ApplySubXactData
 Archive
 ArchiveCheckConfiguredCB
@@ -1485,6 +1488,7 @@ LogicalRepRelId
 LogicalRepRelMapEntry
 LogicalRepRelation
 LogicalRepRollbackPreparedTxnData
+LogicalRepStreamAbortData
 LogicalRepTupleData
 LogicalRepTyp
 LogicalRepWorker
-- 
2.23.0.windows.1