From 6852cd48eb6fdac3599a7c26e3ad60d55fc91017 Mon Sep 17 00:00:00 2001
From: "houzj.fnst" <houzj.fnst@cn.fujitsu.com>
Date: Wed, 10 Aug 2022 20:37:22 +0800
Subject: [PATCH v40 3/6] Add some checks before using parallel apply worker to
 apply changes

streaming=parallel mode has two requirements:
1) The unique column in the relation on the subscriber-side should also be the
unique column on the publisher-side;
2) There cannot be any non-immutable functions used by the subscriber-side
replicated table. Look for functions in the following places:
* a. Trigger functions
* b. Column default value expressions and domain constraints
* c. Constraint expressions
It is better to only check the foreign key when it is detected that the
publisher does not have the foreign key and the subscriber does. But this
requires the publisher to send more information. In addition, since foreign key
does not take effect in the subscriber's apply worker by default, it seems that
foreign key does not hit this ERROR frequently. So, only perform a simple check
based on the subscriber schema when checking non-immutable function uesd by
trigger.

Without these safety checks, the following scenario may occur:
The parallel apply worker locks a row when processing a streaming transaction,
after that the leader apply worker tries to lock the same row when processing
another non-streamed transaction. At this time, the leader apply worker waits
for the streaming transaction to complete and the lock to be released, it won't
send subsequent data of the streaming transaction to the parallel apply worker;
the parallel apply worker waits to receive the rest of streaming transaction
and can't finish this transaction. Now a deadlock has occurred, so both workers
will wait indefinitely.
---
 doc/src/sgml/ref/create_subscription.sgml     |   5 +
 .../replication/logical/applyparallelworker.c |  40 ++
 src/backend/replication/logical/proto.c       |  86 ++-
 src/backend/replication/logical/relation.c    | 199 ++++++
 src/backend/replication/logical/tablesync.c   |   1 +
 src/backend/replication/logical/worker.c      |  24 +-
 src/backend/utils/cache/typcache.c            |  17 +
 src/include/replication/logicalproto.h        |   1 +
 src/include/replication/logicalrelation.h     |   7 +
 src/include/replication/worker_internal.h     |   2 +
 src/include/utils/typcache.h                  |   2 +
 src/test/subscription/meson.build             |   1 +
 src/test/subscription/t/015_stream.pl         |   9 +-
 src/test/subscription/t/016_stream_subxact.pl |   9 +-
 .../subscription/t/022_twophase_cascade.pl    |   8 +
 .../subscription/t/023_twophase_stream.pl     |   9 +-
 .../t/032_streaming_parallel_safety.pl        | 616 ++++++++++++++++++
 17 files changed, 1021 insertions(+), 15 deletions(-)
 create mode 100644 src/test/subscription/t/032_streaming_parallel_safety.pl

diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 03fc047af4..79250acaf5 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -245,6 +245,11 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
           transaction is committed. Note that if an error happens in a
           parallel apply worker, the finish LSN of the remote transaction
           might not be reported in the server log.
+          There are two prerequisites for using <literal>parallel</literal>
+          mode: 1) the unique column in the table on the subscriber-side must
+          also be the unique column on the publisher-side; 2) there cannot be
+          any non-immutable functions used by the subscriber-side replicated
+          table.
          </para>
         </listitem>
        </varlistentry>
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 328672e3f6..f787a50411 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -1097,3 +1097,43 @@ parallel_apply_replorigin_reset(void)
 	replorigin_session_origin_lsn = InvalidXLogRecPtr;
 	replorigin_session_origin_timestamp = 0;
 }
+
+/*
+ * Check if changes on this relation can be applied using a parallel apply
+ * worker.
+ *
+ * Although the commit order is maintained by only allowing one process to
+ * commit at a time, the access order to the relation has changed. This could
+ * cause unexpected problems when applying transactions using a parallel
+ * apply worker if the unique column on the replicated table is
+ * inconsistent with the publisher-side, or if the relation contains
+ * non-immutable functions.
+ */
+void
+parallel_apply_relation_check(LogicalRepRelMapEntry *rel)
+{
+	/* Skip if not in parallel mode */
+	if (!MyLogicalRepWorker->parallel_apply)
+		return;
+
+	/*
+	 * Partition table checks are done later in function
+	 * apply_handle_tuple_routing.
+	 */
+	if (rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
+		return;
+
+	if (!rel->parallel_apply_valid)
+		logicalrep_rel_mark_parallel_apply(rel);
+
+	if (rel->parallel_apply_details != NULL)
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("cannot replicate target relation \"%s.%s\" using "
+						"subscription parameter %s",
+						rel->remoterel.nspname, rel->remoterel.relname,
+						"streaming = parallel"),
+				 errdetail_internal("%s", rel->parallel_apply_details),
+				 errhint("Please change to use subscription parameter %s.",
+						 "streaming = on")));
+}
diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c
index d6160e65bc..bd406a075f 100644
--- a/src/backend/replication/logical/proto.c
+++ b/src/backend/replication/logical/proto.c
@@ -23,7 +23,8 @@
 /*
  * Protocol message flags.
  */
-#define LOGICALREP_IS_REPLICA_IDENTITY 1
+#define ATTR_IS_REPLICA_IDENTITY	(1 << 0)
+#define ATTR_IS_UNIQUE				(1 << 1)
 
 #define MESSAGE_TRANSACTIONAL (1<<0)
 #define TRUNCATE_CASCADE		(1<<0)
@@ -40,6 +41,66 @@ static void logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple);
 static void logicalrep_write_namespace(StringInfo out, Oid nspid);
 static const char *logicalrep_read_namespace(StringInfo in);
 
+/*
+ * RelationGetUniqueKeyBitmap -- get a bitmap of unique attribute numbers
+ *
+ * This is similar to RelationGetIdentityKeyBitmap(), but returns a bitmap of
+ * index attribute numbers for all unique indexes.
+ */
+static Bitmapset *
+RelationGetUniqueKeyBitmap(Relation rel)
+{
+	List	   *indexoidlist = NIL;
+	ListCell   *indexoidscan;
+	Bitmapset  *attunique = NULL;
+
+	if (!rel->rd_rel->relhasindex)
+		return NULL;
+
+	indexoidlist = RelationGetIndexList(rel);
+
+	foreach(indexoidscan, indexoidlist)
+	{
+		Oid			indexoid = lfirst_oid(indexoidscan);
+		Relation	indexRel;
+		int			i;
+
+		/* Look up the description for index */
+		indexRel = RelationIdGetRelation(indexoid);
+
+		if (!RelationIsValid(indexRel))
+			elog(ERROR, "could not open relation with OID %u", indexoid);
+
+		if (!indexRel->rd_index->indisunique)
+		{
+			RelationClose(indexRel);
+			continue;
+		}
+
+		/* Add referenced attributes to attunique */
+		for (i = 0; i < indexRel->rd_index->indnatts; i++)
+		{
+			int			attrnum = indexRel->rd_index->indkey.values[i];
+
+			/*
+			 * We don't include non-key columns into attunique bitmaps. See
+			 * RelationGetIndexAttrBitmap.
+			 */
+			if (attrnum != 0)
+			{
+				if (i < indexRel->rd_index->indnkeyatts &&
+					!bms_is_member(attrnum - FirstLowInvalidHeapAttributeNumber, attunique))
+					attunique = bms_add_member(attunique,
+											   attrnum - FirstLowInvalidHeapAttributeNumber);
+			}
+		}
+		RelationClose(indexRel);
+	}
+	list_free(indexoidlist);
+
+	return attunique;
+}
+
 /*
  * Check if a column is covered by a column list.
  *
@@ -933,7 +994,8 @@ logicalrep_write_attrs(StringInfo out, Relation rel, Bitmapset *columns)
 	TupleDesc	desc;
 	int			i;
 	uint16		nliveatts = 0;
-	Bitmapset  *idattrs = NULL;
+	Bitmapset  *idattrs = NULL,
+			   *attunique = NULL;
 	bool		replidentfull;
 
 	desc = RelationGetDescr(rel);
@@ -958,6 +1020,9 @@ logicalrep_write_attrs(StringInfo out, Relation rel, Bitmapset *columns)
 	if (!replidentfull)
 		idattrs = RelationGetIdentityKeyBitmap(rel);
 
+	/* fetch bitmap of UNIQUE attributes */
+	attunique = RelationGetUniqueKeyBitmap(rel);
+
 	/* send the attributes */
 	for (i = 0; i < desc->natts; i++)
 	{
@@ -974,7 +1039,11 @@ logicalrep_write_attrs(StringInfo out, Relation rel, Bitmapset *columns)
 		if (replidentfull ||
 			bms_is_member(att->attnum - FirstLowInvalidHeapAttributeNumber,
 						  idattrs))
-			flags |= LOGICALREP_IS_REPLICA_IDENTITY;
+			flags |= ATTR_IS_REPLICA_IDENTITY;
+
+		if (bms_is_member(att->attnum - FirstLowInvalidHeapAttributeNumber,
+						  attunique))
+			flags |= ATTR_IS_UNIQUE;
 
 		pq_sendbyte(out, flags);
 
@@ -989,6 +1058,7 @@ logicalrep_write_attrs(StringInfo out, Relation rel, Bitmapset *columns)
 	}
 
 	bms_free(idattrs);
+	bms_free(attunique);
 }
 
 /*
@@ -1001,7 +1071,8 @@ logicalrep_read_attrs(StringInfo in, LogicalRepRelation *rel)
 	int			natts;
 	char	  **attnames;
 	Oid		   *atttyps;
-	Bitmapset  *attkeys = NULL;
+	Bitmapset  *attkeys = NULL,
+			   *attunique = NULL;
 
 	natts = pq_getmsgint(in, 2);
 	attnames = palloc(natts * sizeof(char *));
@@ -1014,9 +1085,13 @@ logicalrep_read_attrs(StringInfo in, LogicalRepRelation *rel)
 
 		/* Check for replica identity column */
 		flags = pq_getmsgbyte(in);
-		if (flags & LOGICALREP_IS_REPLICA_IDENTITY)
+		if (flags & ATTR_IS_REPLICA_IDENTITY)
 			attkeys = bms_add_member(attkeys, i);
 
+		/* Check for unique column */
+		if (flags & ATTR_IS_UNIQUE)
+			attunique = bms_add_member(attunique, i);
+
 		/* attribute name */
 		attnames[i] = pstrdup(pq_getmsgstring(in));
 
@@ -1030,6 +1105,7 @@ logicalrep_read_attrs(StringInfo in, LogicalRepRelation *rel)
 	rel->attnames = attnames;
 	rel->atttyps = atttyps;
 	rel->attkeys = attkeys;
+	rel->attunique = attunique;
 	rel->natts = natts;
 }
 
diff --git a/src/backend/replication/logical/relation.c b/src/backend/replication/logical/relation.c
index e989047681..695244110b 100644
--- a/src/backend/replication/logical/relation.c
+++ b/src/backend/replication/logical/relation.c
@@ -19,12 +19,19 @@
 
 #include "access/table.h"
 #include "catalog/namespace.h"
+#include "catalog/pg_proc.h"
 #include "catalog/pg_subscription_rel.h"
+#include "commands/trigger.h"
 #include "executor/executor.h"
 #include "nodes/makefuncs.h"
+#include "optimizer/optimizer.h"
 #include "replication/logicalrelation.h"
 #include "replication/worker_internal.h"
+#include "rewrite/rewriteHandler.h"
 #include "utils/inval.h"
+#include "utils/lsyscache.h"
+#include "utils/syscache.h"
+#include "utils/typcache.h"
 
 
 static MemoryContext LogicalRepRelMapContext = NULL;
@@ -91,6 +98,23 @@ logicalrep_relmap_invalidate_cb(Datum arg, Oid reloid)
 	}
 }
 
+/*
+ * Syscache invalidation callback to reset parallel_apply_valid flag.
+ */
+static void
+logicalrep_relmap_reset_parallel_cb(Datum arg, int cacheid, uint32 hashvalue)
+{
+	HASH_SEQ_STATUS hash_seq;
+	LogicalRepRelMapEntry *entry;
+
+	if (LogicalRepRelMap == NULL)
+		return;
+
+	hash_seq_init(&hash_seq, LogicalRepRelMap);
+	while ((entry = hash_seq_search(&hash_seq)) != NULL)
+		entry->parallel_apply_valid = false;
+}
+
 /*
  * Initialize the relation map cache.
  */
@@ -116,6 +140,16 @@ logicalrep_relmap_init(void)
 	/* Watch for invalidation events. */
 	CacheRegisterRelcacheCallback(logicalrep_relmap_invalidate_cb,
 								  (Datum) 0);
+
+	if (MyLogicalRepWorker->parallel_apply)
+	{
+		CacheRegisterSyscacheCallback(PROCOID,
+									  logicalrep_relmap_reset_parallel_cb,
+									  (Datum) 0);
+		CacheRegisterSyscacheCallback(TYPEOID,
+									  logicalrep_relmap_reset_parallel_cb,
+									  (Datum) 0);
+	}
 }
 
 /*
@@ -142,6 +176,7 @@ logicalrep_relmap_free_entry(LogicalRepRelMapEntry *entry)
 		pfree(remoterel->atttyps);
 	}
 	bms_free(remoterel->attkeys);
+	bms_free(remoterel->attunique);
 
 	if (entry->attrmap)
 		free_attrmap(entry->attrmap);
@@ -190,6 +225,7 @@ logicalrep_relmap_update(LogicalRepRelation *remoterel)
 	}
 	entry->remoterel.replident = remoterel->replident;
 	entry->remoterel.attkeys = bms_copy(remoterel->attkeys);
+	entry->remoterel.attunique = bms_copy(remoterel->attunique);
 	MemoryContextSwitchTo(oldctx);
 }
 
@@ -310,6 +346,162 @@ logicalrep_rel_mark_updatable(LogicalRepRelMapEntry *entry)
 	}
 }
 
+/*
+ * Check if changes on one relation can be applied using a parallel apply
+ * worker and assign the 'parallel_apply_valid' flag and
+ * 'parallel_apply_details' field.
+ *
+ * There are two requirements for applying changes using a parallel apply
+ * worker: 1) the unique column in the table on the subscriber-side should also
+ * be the unique column on the publisher-side; 2) there cannot be any
+ * non-immutable functions used by the subscriber-side replicated table.
+ *
+ * Without these safety checks, the following scenario may occur: The parallel
+ * apply worker locks a row when processing a streaming transaction, after
+ * that the leader apply worker tries to lock the same row when processing
+ * another non-streamed transaction. At this time, the leader apply worker
+ * waits for the streaming transaction to complete and the lock to be released,
+ * it won't send subsequent data of the streaming transaction to the parallel
+ * apply worker; the parallel apply worker waits to receive the rest of
+ * streaming transaction and can't finish this transaction. Now a deadlock has
+ * occurred, so both workers will wait indefinitely.
+ *
+ * We just mark the relation entry as valid here. If changes on one relation
+ * can not be applied using a parallel apply worker, save the related details
+ * in 'parallel_apply_details' and leave it to parallel_apply_relation_check()
+ * to throw the actual error if needed.
+ */
+void
+logicalrep_rel_mark_parallel_apply(LogicalRepRelMapEntry *entry)
+{
+	Bitmapset  *ukey;
+	int			i;
+	TupleDesc	tupdesc;
+	int			attnum;
+
+	/* Skip if not in parallel mode */
+	if (!MyLogicalRepWorker->parallel_apply)
+		return;
+
+	entry->parallel_apply_valid = true;
+
+	/*
+	 * First, check if the unique column in the relation on the
+	 * subscriber-side is also the unique column on the publisher-side.
+	 */
+	ukey = RelationGetIndexAttrBitmap(entry->localrel,
+									  INDEX_ATTR_BITMAP_KEY);
+
+	if (ukey)
+	{
+		i = -1;
+		while ((i = bms_next_member(ukey, i)) >= 0)
+		{
+			attnum = AttrNumberGetAttrOffset(i + FirstLowInvalidHeapAttributeNumber);
+
+			if (entry->attrmap->attnums[attnum] < 0 ||
+				!bms_is_member(entry->attrmap->attnums[attnum], entry->remoterel.attunique))
+			{
+				entry->parallel_apply_details = _("Unique column on the subscriber-side is not included in the unique columns on the publisher-side.");
+				bms_free(ukey);
+				return;
+			}
+		}
+
+		bms_free(ukey);
+	}
+
+	/*
+	 * Then, check if there is any non-immutable function used by the
+	 * subscriber-side relation. Look for functions in the following places:
+	 * a. trigger functions; b. Column default value expressions and domain
+	 * constraints; c. Constraint expressions;
+	 */
+	/* Check the trigger functions. */
+	if (entry->localrel->trigdesc != NULL)
+	{
+		for (i = 0; i < entry->localrel->trigdesc->numtriggers; i++)
+		{
+			Trigger    *trig = entry->localrel->trigdesc->triggers + i;
+
+			if (trig->tgenabled != TRIGGER_FIRES_ALWAYS &&
+				trig->tgenabled != TRIGGER_FIRES_ON_REPLICA)
+				continue;
+
+			if (func_volatile(trig->tgfoid) != PROVOLATILE_IMMUTABLE)
+			{
+				entry->parallel_apply_details = _("Trigger uses non-immutable function.");
+				return;
+			}
+		}
+	}
+
+	/* Check the columns. */
+	tupdesc = RelationGetDescr(entry->localrel);
+	for (attnum = 0; attnum < tupdesc->natts; attnum++)
+	{
+		Form_pg_attribute att = TupleDescAttr(tupdesc, attnum);
+		Node	   *defaultexpr = NULL;
+
+		/* We don't check dropped or generated attributes */
+		if (att->attisdropped || att->attgenerated)
+			continue;
+
+		defaultexpr = build_column_default(entry->localrel, attnum + 1);
+		if (defaultexpr && contain_mutable_functions(defaultexpr))
+		{
+			entry->parallel_apply_details = _("The default expression of column uses non-immutable function.");
+			return;
+		}
+
+		/*
+		 * If the column is of a DOMAIN type, determine whether that domain
+		 * has any CHECK expressions that are not immutable.
+		 */
+		if (get_typtype(att->atttypid) == TYPTYPE_DOMAIN)
+		{
+			List	   *domain_constraints;
+			ListCell   *lc;
+
+			domain_constraints = GetDomainConstraints(att->atttypid);
+
+			foreach(lc, domain_constraints)
+			{
+				DomainConstraintState *con = (DomainConstraintState *) lfirst(lc);
+
+				if (con->check_expr && contain_mutable_functions((Node *) con->check_expr))
+				{
+					entry->parallel_apply_details = _("The default expression of column uses non-immutable function.");
+					return;
+				}
+			}
+		}
+	}
+
+	/* Check the constraints. */
+	if (tupdesc->constr)
+	{
+		ConstrCheck *check = tupdesc->constr->check;
+
+		/*
+		 * Determine if there are any CHECK constraints which contains
+		 * non-immutable function.
+		 */
+		for (i = 0; i < tupdesc->constr->num_check; i++)
+		{
+			Expr	   *check_expr = stringToNode(check[i].ccbin);
+
+			if (contain_mutable_functions((Node *) check_expr))
+			{
+				entry->parallel_apply_details = _("Constraints use non-immutable function.");
+				return;
+			}
+		}
+	}
+
+	entry->parallel_apply_details = NULL;
+}
+
 /*
  * Open the local relation associated with the remote one.
  *
@@ -438,6 +630,9 @@ logicalrep_rel_open(LogicalRepRelId remoteid, LOCKMODE lockmode)
 		 */
 		logicalrep_rel_mark_updatable(entry);
 
+		/* Set if changes could be applied using a parallel apply worker */
+		logicalrep_rel_mark_parallel_apply(entry);
+
 		entry->localrelvalid = true;
 	}
 
@@ -653,6 +848,7 @@ logicalrep_partition_open(LogicalRepRelMapEntry *root,
 		}
 		entry->remoterel.replident = remoterel->replident;
 		entry->remoterel.attkeys = bms_copy(remoterel->attkeys);
+		entry->remoterel.attunique = bms_copy(remoterel->attunique);
 	}
 
 	entry->localrel = partrel;
@@ -696,6 +892,9 @@ logicalrep_partition_open(LogicalRepRelMapEntry *root,
 	/* Set if the table's replica identity is enough to apply update/delete. */
 	logicalrep_rel_mark_updatable(entry);
 
+	/* Set if changes could be applied using a parallel apply worker */
+	logicalrep_rel_mark_parallel_apply(entry);
+
 	entry->localrelvalid = true;
 
 	/* state and statelsn are left set to 0. */
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 1c1a8e7919..87594eacf0 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -925,6 +925,7 @@ fetch_remote_table_info(char *nspname, char *relname,
 	lrel->attnames = palloc0(MaxTupleAttributeNumber * sizeof(char *));
 	lrel->atttyps = palloc0(MaxTupleAttributeNumber * sizeof(Oid));
 	lrel->attkeys = NULL;
+	lrel->attunique = NULL;
 
 	/*
 	 * Store the columns as a list of names.  Ignore those that are not
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index f804560caa..85363652ac 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -1524,6 +1524,15 @@ apply_handle_stream_stop(StringInfo s)
 		case TRANS_LEADER_SEND_TO_PARALLEL:
 			Assert(winfo);
 
+			/*
+			 * Unlike stream_commit, we don't need to wait here for
+			 * stream_stop to finish. Allowing the other transaction to be
+			 * applied before stream_stop is finished can lead to failures if
+			 * the unique index/constraint is different between publisher and
+			 * subscriber. But for such cases, we don't allow streamed
+			 * transactions to be applied in parallel. See
+			 * parallel_apply_relation_check.
+			 */
 			parallel_apply_send_data(winfo, s->len, s->data);
 
 			elog(DEBUG1, "applied %u changes in the streaming chunk",
@@ -2131,6 +2140,8 @@ apply_handle_insert(StringInfo s)
 	/* Set relation for error callback */
 	apply_error_callback_arg.rel = rel;
 
+	parallel_apply_relation_check(rel);
+
 	/* Initialize the executor state. */
 	edata = create_edata_for_relation(rel);
 	estate = edata->estate;
@@ -2274,6 +2285,8 @@ apply_handle_update(StringInfo s)
 	/* Check if we can do the update. */
 	check_relation_updatable(rel);
 
+	parallel_apply_relation_check(rel);
+
 	/* Initialize the executor state. */
 	edata = create_edata_for_relation(rel);
 	estate = edata->estate;
@@ -2442,6 +2455,8 @@ apply_handle_delete(StringInfo s)
 	/* Check if we can do the delete. */
 	check_relation_updatable(rel);
 
+	parallel_apply_relation_check(rel);
+
 	/* Initialize the executor state. */
 	edata = create_edata_for_relation(rel);
 	estate = edata->estate;
@@ -2627,13 +2642,14 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
 	}
 	MemoryContextSwitchTo(oldctx);
 
+	part_entry = logicalrep_partition_open(relmapentry, partrel,
+										   attrmap);
+
 	/* Check if we can do the update or delete on the leaf partition. */
 	if (operation == CMD_UPDATE || operation == CMD_DELETE)
-	{
-		part_entry = logicalrep_partition_open(relmapentry, partrel,
-											   attrmap);
 		check_relation_updatable(part_entry);
-	}
+
+	parallel_apply_relation_check(part_entry);
 
 	switch (operation)
 	{
diff --git a/src/backend/utils/cache/typcache.c b/src/backend/utils/cache/typcache.c
index 808f9ebd0d..511b23d263 100644
--- a/src/backend/utils/cache/typcache.c
+++ b/src/backend/utils/cache/typcache.c
@@ -2540,6 +2540,23 @@ compare_values_of_enum(TypeCacheEntry *tcache, Oid arg1, Oid arg2)
 		return 0;
 }
 
+/*
+ * GetDomainConstraints --- get DomainConstraintState list of specified domain type
+ */
+List *
+GetDomainConstraints(Oid type_id)
+{
+	TypeCacheEntry *typentry;
+	List	   *constraints = NIL;
+
+	typentry = lookup_type_cache(type_id, TYPECACHE_DOMAIN_CONSTR_INFO);
+
+	if (typentry->domainData != NULL)
+		constraints = typentry->domainData->constraints;
+
+	return constraints;
+}
+
 /*
  * Load (or re-load) the enumData member of the typcache entry.
  */
diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h
index 1da7b01ca6..d4bc92c5e2 100644
--- a/src/include/replication/logicalproto.h
+++ b/src/include/replication/logicalproto.h
@@ -113,6 +113,7 @@ typedef struct LogicalRepRelation
 	char		replident;		/* replica identity */
 	char		relkind;		/* remote relation kind */
 	Bitmapset  *attkeys;		/* Bitmap of key columns */
+	Bitmapset  *attunique;		/* Bitmap of unique columns */
 } LogicalRepRelation;
 
 /* Type mapping info */
diff --git a/src/include/replication/logicalrelation.h b/src/include/replication/logicalrelation.h
index 78cd7e77f5..400387b797 100644
--- a/src/include/replication/logicalrelation.h
+++ b/src/include/replication/logicalrelation.h
@@ -31,6 +31,11 @@ typedef struct LogicalRepRelMapEntry
 	Relation	localrel;		/* relcache entry (NULL when closed) */
 	AttrMap    *attrmap;		/* map of local attributes to remote ones */
 	bool		updatable;		/* Can apply updates/deletes? */
+	bool		parallel_apply_valid;	/* Can apply changes in a
+										 * parallel apply worker? */
+	const char *parallel_apply_details;	/* Details about not being able to
+										 * apply changes in a parallel apply
+										 * worker. */
 
 	/* Sync state. */
 	char		state;
@@ -47,4 +52,6 @@ extern LogicalRepRelMapEntry *logicalrep_partition_open(LogicalRepRelMapEntry *r
 extern void logicalrep_rel_close(LogicalRepRelMapEntry *rel,
 								 LOCKMODE lockmode);
 
+extern void logicalrep_rel_mark_parallel_apply(LogicalRepRelMapEntry *entry);
+
 #endif							/* LOGICALRELATION_H */
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 87f2ec9f72..2f51521a8f 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -218,6 +218,8 @@ extern void parallel_apply_stream_abort(LogicalRepStreamAbortData *abort_data);
 extern void parallel_apply_replorigin_setup(void);
 extern void parallel_apply_replorigin_reset(void);
 
+extern void parallel_apply_relation_check(LogicalRepRelMapEntry *rel);
+
 #define isParallelApplyWorker(worker) ((worker)->apply_leader_pid != InvalidPid)
 
 static inline bool
diff --git a/src/include/utils/typcache.h b/src/include/utils/typcache.h
index 431ad7f1b3..ed7c2e7f48 100644
--- a/src/include/utils/typcache.h
+++ b/src/include/utils/typcache.h
@@ -199,6 +199,8 @@ extern uint64 assign_record_type_identifier(Oid type_id, int32 typmod);
 
 extern int	compare_values_of_enum(TypeCacheEntry *tcache, Oid arg1, Oid arg2);
 
+extern List *GetDomainConstraints(Oid type_id);
+
 extern size_t SharedRecordTypmodRegistryEstimate(void);
 
 extern void SharedRecordTypmodRegistryInit(SharedRecordTypmodRegistry *,
diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build
index 85d1dd9295..baabfb4a6a 100644
--- a/src/test/subscription/meson.build
+++ b/src/test/subscription/meson.build
@@ -36,6 +36,7 @@ tests += {
       't/029_on_error.pl',
       't/030_origin.pl',
       't/031_column_list.pl',
+      't/032_streaming_parallel_safety.pl',
       't/100_bugs.pl',
     ],
   },
diff --git a/src/test/subscription/t/015_stream.pl b/src/test/subscription/t/015_stream.pl
index 65f43f0881..6d1ff8d5e9 100644
--- a/src/test/subscription/t/015_stream.pl
+++ b/src/test/subscription/t/015_stream.pl
@@ -205,8 +205,13 @@ my $oldpid = $node_publisher->safe_psql('postgres',
 	"SELECT pid FROM pg_stat_replication WHERE application_name = '$appname' AND state = 'streaming';"
 );
 
-$node_subscriber->safe_psql('postgres',
-	"ALTER SUBSCRIPTION tap_sub SET(streaming = parallel, binary = off)");
+# "streaming = parallel" does not support non-immutable functions, so change
+# the function in the default expression of column "c".
+$node_subscriber->safe_psql(
+	'postgres', qq{
+ALTER TABLE test_tab ALTER COLUMN c SET DEFAULT to_timestamp(0);
+ALTER SUBSCRIPTION tap_sub SET(streaming = parallel, binary = off);
+});
 
 $node_publisher->poll_query_until('postgres',
 	"SELECT pid != $oldpid FROM pg_stat_replication WHERE application_name = '$appname' AND state = 'streaming';"
diff --git a/src/test/subscription/t/016_stream_subxact.pl b/src/test/subscription/t/016_stream_subxact.pl
index 020ad96d1f..0efa41d930 100644
--- a/src/test/subscription/t/016_stream_subxact.pl
+++ b/src/test/subscription/t/016_stream_subxact.pl
@@ -150,8 +150,13 @@ my $oldpid = $node_publisher->safe_psql('postgres',
 	"SELECT pid FROM pg_stat_replication WHERE application_name = '$appname' AND state = 'streaming';"
 );
 
-$node_subscriber->safe_psql('postgres',
-	"ALTER SUBSCRIPTION tap_sub SET(streaming = parallel)");
+# "streaming = parallel" does not support non-immutable functions, so change
+# the function in the default expression of column "c".
+$node_subscriber->safe_psql(
+	'postgres', qq{
+ALTER TABLE test_tab ALTER COLUMN c SET DEFAULT to_timestamp(0);
+ALTER SUBSCRIPTION tap_sub SET(streaming = parallel);
+});
 
 $node_publisher->poll_query_until('postgres',
 	"SELECT pid != $oldpid FROM pg_stat_replication WHERE application_name = '$appname' AND state = 'streaming';"
diff --git a/src/test/subscription/t/022_twophase_cascade.pl b/src/test/subscription/t/022_twophase_cascade.pl
index 4265d3b3f5..893faaa0f5 100644
--- a/src/test/subscription/t/022_twophase_cascade.pl
+++ b/src/test/subscription/t/022_twophase_cascade.pl
@@ -501,6 +501,14 @@ test_streaming($node_A, $node_B, $node_C, $appname_B, $appname_C, 'on');
 ######################################
 # Test using streaming mode 'parallel'
 ######################################
+
+# "streaming = parallel" does not support non-immutable functions, so change
+# the function in the default expression of column "c".
+$node_B->safe_psql('postgres',
+	"ALTER TABLE test_tab ALTER COLUMN c SET DEFAULT to_timestamp(0);");
+$node_C->safe_psql('postgres',
+	"ALTER TABLE test_tab ALTER COLUMN c SET DEFAULT to_timestamp(0);");
+
 test_streaming($node_A, $node_B, $node_C, $appname_B, $appname_C, 'parallel');
 
 ###############################
diff --git a/src/test/subscription/t/023_twophase_stream.pl b/src/test/subscription/t/023_twophase_stream.pl
index 82b5e6c12e..2cd453c03f 100644
--- a/src/test/subscription/t/023_twophase_stream.pl
+++ b/src/test/subscription/t/023_twophase_stream.pl
@@ -384,8 +384,13 @@ my $oldpid = $node_publisher->safe_psql('postgres',
 	"SELECT pid FROM pg_stat_replication WHERE application_name = '$appname' AND state = 'streaming';"
 );
 
-$node_subscriber->safe_psql('postgres',
-	"ALTER SUBSCRIPTION tap_sub SET(streaming = parallel)");
+# "streaming = parallel" does not support non-immutable functions, so change
+# the function in the default expression of column "c".
+$node_subscriber->safe_psql(
+	'postgres', qq{
+ALTER TABLE test_tab ALTER COLUMN c SET DEFAULT to_timestamp(0);
+ALTER SUBSCRIPTION tap_sub SET(streaming = parallel);
+});
 
 $node_publisher->poll_query_until('postgres',
 	"SELECT pid != $oldpid FROM pg_stat_replication WHERE application_name = '$appname' AND state = 'streaming';"
diff --git a/src/test/subscription/t/032_streaming_parallel_safety.pl b/src/test/subscription/t/032_streaming_parallel_safety.pl
new file mode 100644
index 0000000000..3aaa699c07
--- /dev/null
+++ b/src/test/subscription/t/032_streaming_parallel_safety.pl
@@ -0,0 +1,616 @@
+# Copyright (c) 2022, PostgreSQL Global Development Group
+
+# Test the safety checks of streaming mode "parallel" in logical replication.
+# Without these safety checks, the subscriber's apply worker may fall into an
+# infinite wait without the user knowing.
+#
+# For normal tables, we use deadlock-producing test cases to ensure that future
+# modifications do not invalidate constraint checks.
+#
+# For partitioned tables, we just use test cases to confirm that the constraint
+# checks are as expected.
+
+use strict;
+use warnings;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+my $offset = 0;
+
+# Create publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->append_conf('postgresql.conf',
+	'logical_decoding_work_mem = 64kB');
+$node_publisher->start;
+
+# Create subscriber node
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init;
+$node_subscriber->start;
+
+# Setup structure on publisher
+$node_publisher->safe_psql('postgres', "CREATE TABLE test_tab1 (a int)");
+$node_publisher->safe_psql('postgres', "CREATE TABLE test_tab2 (a int)");
+$node_publisher->safe_psql('postgres',
+	"CREATE TABLE test_tab_partitioned (a int primary key, b varchar)");
+
+# Setup structure on subscriber
+# We need to test normal table and partition table.
+$node_subscriber->safe_psql('postgres', "CREATE TABLE test_tab1 (a int)");
+$node_subscriber->safe_psql('postgres', "CREATE TABLE test_tab2 (a int)");
+$node_subscriber->safe_psql('postgres',
+	"CREATE TABLE test_tab_partitioned (a int primary key, b varchar) PARTITION BY RANGE(a)"
+);
+$node_subscriber->safe_psql('postgres',
+	"CREATE TABLE test_tab_partition (LIKE test_tab_partitioned)");
+$node_subscriber->safe_psql('postgres',
+	"ALTER TABLE test_tab_partitioned ATTACH PARTITION test_tab_partition DEFAULT"
+);
+
+# Setup logical replication
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+$node_publisher->safe_psql('postgres',
+	"CREATE PUBLICATION tap_pub_normal FOR TABLE test_tab1, test_tab2");
+$node_publisher->safe_psql('postgres',
+	"CREATE PUBLICATION tap_pub_partitioned FOR TABLE test_tab_partitioned");
+
+my $appname = 'tap_sub';
+$node_subscriber->safe_psql(
+	'postgres', "
+	CREATE SUBSCRIPTION tap_sub
+	CONNECTION '$publisher_connstr application_name=$appname'
+	PUBLICATION tap_pub_normal, tap_pub_partitioned
+	WITH (streaming = parallel, copy_data = false)");
+
+$node_publisher->wait_for_catchup($appname);
+
+# Interleave a pair of transactions, each exceeding the 64kB limit.
+my $in  = '';
+my $out = '';
+
+my $timer = IPC::Run::timeout($PostgreSQL::Test::Utils::timeout_default);
+
+my $h = $node_publisher->background_psql('postgres', \$in, \$out, $timer,
+	on_error_stop => 0);
+
+# ============================================================================
+# It is not allowed that the unique column in the relation on the
+# subscriber-side is not the unique column on the publisher-side. Check the
+# error reported by parallel worker in this case.
+# ============================================================================
+
+# First we check the unique index on normal table.
+$node_subscriber->safe_psql('postgres',
+	"CREATE UNIQUE INDEX idx_tab1 on test_tab1(a)");
+
+$in .= q{
+BEGIN;
+INSERT INTO test_tab1 SELECT i FROM generate_series(1, 5000) s(i);
+};
+$h->pump_nb;
+
+$node_publisher->safe_psql('postgres', "INSERT INTO test_tab1 values(1)");
+
+$in .= q{
+COMMIT;
+\q
+};
+$h->finish;
+
+$node_subscriber->wait_for_log(
+	qr/ERROR: ( [A-Z0-9]+:)? cannot replicate target relation "public.test_tab1" using subscription parameter streaming = parallel/,
+	$offset);
+
+# Drop the unique index on the subscriber, now it works.
+$node_subscriber->safe_psql('postgres', "DROP INDEX idx_tab1");
+
+# Wait for this streaming transaction to be applied in the apply worker.
+$node_publisher->wait_for_catchup($appname);
+
+my $result =
+  $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab1");
+is($result, qq(5001), 'data replicated to subscriber after dropping index');
+
+# Clean up test data from the environment.
+$node_publisher->safe_psql('postgres', "TRUNCATE TABLE test_tab1");
+$node_publisher->wait_for_catchup($appname);
+
+# Then we check the unique index on partition table.
+$node_subscriber->safe_psql('postgres',
+	"CREATE UNIQUE INDEX test_tab_b_partition_idx ON test_tab_partition (b)");
+
+# Check the subscriber log from now on.
+$offset = -s $node_subscriber->logfile;
+
+$node_publisher->safe_psql('postgres',
+	"INSERT INTO test_tab_partitioned SELECT i, md5(i::text) FROM generate_series(1, 5000) s(i)"
+);
+
+$node_subscriber->wait_for_log(
+	qr/ERROR: ( [A-Z0-9]+:)? cannot replicate target relation "public.test_tab_partitioned" using subscription parameter streaming = parallel/,
+	$offset);
+
+# Drop the unique index on the subscriber, now it works.
+$node_subscriber->safe_psql('postgres',
+	"DROP INDEX test_tab_b_partition_idx");
+
+# Wait for this streaming transaction to be applied in the apply worker.
+$node_publisher->wait_for_catchup($appname);
+
+$result =
+  $node_subscriber->safe_psql('postgres',
+	"SELECT count(*) FROM test_tab_partitioned");
+is($result, qq(5000), 'data replicated to subscriber after dropping index');
+
+# ============================================================================
+# Triggers which execute non-immutable function are not allowed on the
+# subscriber side. Check the error reported by parallel worker in this case.
+# ============================================================================
+
+# First we check the trigger function on normal table.
+$node_publisher->safe_psql('postgres',
+	"CREATE UNIQUE INDEX idx_tab2 on test_tab2(a)");
+$node_subscriber->safe_psql('postgres',
+	"CREATE UNIQUE INDEX idx_tab2 on test_tab2(a)");
+
+$node_subscriber->safe_psql(
+	'postgres', qq{
+CREATE FUNCTION trigger_func_tab1_unsafe() RETURNS TRIGGER AS \$\$
+  BEGIN
+    INSERT INTO public.test_tab2 VALUES (NEW.*);
+    RETURN NEW;
+  END
+\$\$ language plpgsql;
+CREATE TRIGGER tri_tab1_unsafe
+BEFORE INSERT ON public.test_tab1
+FOR EACH ROW EXECUTE PROCEDURE trigger_func_tab1_unsafe();
+ALTER TABLE test_tab1 ENABLE REPLICA TRIGGER tri_tab1_unsafe;
+
+CREATE FUNCTION trigger_func_tab1_safe() RETURNS TRIGGER AS \$\$
+  BEGIN
+    RAISE NOTICE 'test for safe trigger function';
+	RETURN NEW;
+  END
+\$\$ language plpgsql;
+ALTER FUNCTION trigger_func_tab1_safe IMMUTABLE;
+CREATE TRIGGER tri_tab1_safe
+BEFORE INSERT ON public.test_tab1
+FOR EACH ROW EXECUTE PROCEDURE trigger_func_tab1_safe();
+});
+
+# Check the subscriber log from now on.
+$offset = -s $node_subscriber->logfile;
+
+$in .= q{
+BEGIN;
+INSERT INTO test_tab1 VALUES(5001);
+INSERT INTO test_tab2 SELECT i FROM generate_series(1, 5000) s(i);
+};
+$h->pump_nb;
+
+$node_publisher->safe_psql('postgres', "INSERT INTO test_tab1 VALUES(5001)");
+
+$in .= q{
+COMMIT;
+\q
+};
+$h->finish;
+
+$node_subscriber->wait_for_log(
+	qr/ERROR: ( [A-Z0-9]+:)? cannot replicate target relation "public.test_tab1" using subscription parameter streaming = parallel/,
+	$offset);
+
+# Using trigger with immutable function, now it works.
+$node_subscriber->safe_psql(
+	'postgres', qq{
+ALTER TABLE test_tab1 ENABLE REPLICA TRIGGER tri_tab1_safe;
+DROP TRIGGER tri_tab1_unsafe ON public.test_tab1;
+});
+
+# Wait for this streaming transaction to be applied in the apply worker.
+$node_publisher->wait_for_catchup($appname);
+
+$result =
+  $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab1");
+is($result, qq(2),
+	'data replicated to subscriber after using immutable expression');
+
+# Clean up test data from the environment.
+$node_subscriber->safe_psql(
+	'postgres', qq{
+DROP INDEX idx_tab2;
+DROP TRIGGER tri_tab1_safe ON public.test_tab1;
+DROP function trigger_func_tab1_unsafe;
+DROP function trigger_func_tab1_safe;
+});
+$node_publisher->safe_psql(
+	'postgres', qq{
+DROP INDEX idx_tab2;
+TRUNCATE TABLE test_tab1;
+TRUNCATE TABLE test_tab2;
+});
+$node_publisher->wait_for_catchup($appname);
+
+# Then we check the trigger function on partition table.
+$node_subscriber->safe_psql(
+	'postgres', qq{
+CREATE FUNCTION trigger_func() RETURNS TRIGGER AS \$\$
+  BEGIN
+    RETURN NULL;
+  END
+\$\$ language plpgsql;
+CREATE TRIGGER insert_trig
+BEFORE INSERT ON test_tab_partition
+FOR EACH ROW EXECUTE PROCEDURE trigger_func();
+ALTER TABLE test_tab_partition ENABLE REPLICA TRIGGER insert_trig;
+});
+
+# Check the subscriber log from now on.
+$offset = -s $node_subscriber->logfile;
+
+$node_publisher->safe_psql('postgres', "DELETE FROM test_tab_partitioned");
+
+$node_subscriber->wait_for_log(
+	qr/ERROR: ( [A-Z0-9]+:)? cannot replicate target relation "public.test_tab_partitioned" using subscription parameter streaming = parallel/,
+	$offset);
+
+# Drop the trigger on the subscriber, now it works.
+$node_subscriber->safe_psql('postgres',
+	"DROP TRIGGER insert_trig ON test_tab_partition");
+
+# Wait for this streaming transaction to be applied in the apply worker.
+$node_publisher->wait_for_catchup($appname);
+
+$result =
+  $node_subscriber->safe_psql('postgres',
+	"SELECT count(*) FROM test_tab_partitioned");
+is($result, qq(0), 'data replicated to subscriber after dropping trigger');
+
+# ============================================================================
+# It is not allowed that column default value expression contains a
+# non-immutable function on the subscriber side. Check the error reported by
+# parallel worker in this case.
+# ============================================================================
+
+# First we check the column default value expression on normal table.
+$node_publisher->safe_psql('postgres', "INSERT INTO test_tab2 VALUES(1)");
+
+$node_subscriber->safe_psql(
+	'postgres', qq{
+CREATE FUNCTION func_count_tab2() RETURNS INT AS \$\$
+  BEGIN
+    RETURN (SELECT count(*) FROM public.test_tab2);
+  END
+\$\$ language plpgsql;
+ALTER TABLE test_tab1 ADD COLUMN b int DEFAULT func_count_tab2();
+});
+
+# Check the subscriber log from now on.
+$offset = -s $node_subscriber->logfile;
+
+$in .= q{
+BEGIN;
+TRUNCATE test_tab2;
+INSERT INTO test_tab1 SELECT i FROM generate_series(1, 5000) s(i);
+};
+$h->pump_nb;
+
+$node_publisher->safe_psql('postgres', "INSERT INTO test_tab1(a) VALUES(1)");
+
+$in .= q{
+COMMIT;
+\q
+};
+$h->finish;
+
+$node_subscriber->wait_for_log(
+	qr/ERROR: ( [A-Z0-9]+:)? cannot replicate target relation "public.test_tab1" using subscription parameter streaming = parallel/,
+	$offset);
+
+# Alter default values to immutable expression, now it works.
+$node_subscriber->safe_psql('postgres',
+	"ALTER TABLE test_tab1 ALTER COLUMN b SET DEFAULT 1");
+
+# Wait for this streaming transaction to be applied in the apply worker.
+$node_publisher->wait_for_catchup($appname);
+
+$result =
+  $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab1");
+is($result, qq(5001),
+	'data replicated to subscriber after using immutable expression');
+
+# Clean up test data from the environment.
+$node_subscriber->safe_psql('postgres',
+	"ALTER TABLE test_tab1 DROP COLUMN b");
+$node_publisher->safe_psql(
+	'postgres', qq{
+TRUNCATE TABLE test_tab1;
+TRUNCATE TABLE test_tab2;
+});
+$node_publisher->wait_for_catchup($appname);
+
+# Then we check the column default value expression on partition table.
+$node_subscriber->safe_psql('postgres',
+	"ALTER TABLE test_tab_partition ALTER COLUMN b SET DEFAULT random()");
+
+# Check the subscriber log from now on.
+$offset = -s $node_subscriber->logfile;
+
+$node_publisher->safe_psql('postgres',
+	"INSERT INTO test_tab_partitioned SELECT i, md5(i::text) FROM generate_series(1, 5000) s(i)"
+);
+
+$node_subscriber->wait_for_log(
+	qr/ERROR: ( [A-Z0-9]+:)? cannot replicate target relation "public.test_tab_partitioned" using subscription parameter streaming = parallel/,
+	$offset);
+
+# Drop default value on the subscriber, now it works.
+$node_subscriber->safe_psql('postgres',
+	"ALTER TABLE test_tab_partition ALTER COLUMN b DROP DEFAULT");
+
+# Wait for this streaming transaction to be applied in the apply worker.
+$node_publisher->wait_for_catchup($appname);
+
+$result =
+  $node_subscriber->safe_psql('postgres',
+	"SELECT count(*) FROM test_tab_partitioned");
+is($result, qq(5000),
+	'data replicated to subscriber after dropping default value expression');
+
+# ============================================================================
+# It is not allowed that domain constraint expression contains a non-immutable
+# function on the subscriber side. Check the error reported by parallel
+# worker in this case.
+# ============================================================================
+
+# Because the column type of the partition table must be the same as its parent
+# table, only test normal table here.
+$node_publisher->safe_psql('postgres', "INSERT INTO test_tab2 VALUES(1)");
+
+$node_publisher->safe_psql(
+	'postgres', qq{
+CREATE DOMAIN tmp_domain AS int CHECK (VALUE > -1);
+ALTER TABLE test_tab1 ALTER COLUMN a TYPE tmp_domain;
+});
+
+$node_subscriber->safe_psql(
+	'postgres', qq{
+CREATE DOMAIN tmp_domain AS INT CONSTRAINT domain_check CHECK (VALUE >= func_count_tab2());
+ALTER TABLE test_tab1 ALTER COLUMN a TYPE tmp_domain;
+});
+
+# Check the subscriber log from now on.
+$offset = -s $node_subscriber->logfile;
+
+$in .= q{
+BEGIN;
+TRUNCATE test_tab2;
+INSERT INTO test_tab1 SELECT i FROM generate_series(1, 5000) s(i);
+};
+$h->pump_nb;
+
+$node_publisher->safe_psql('postgres', "INSERT INTO test_tab1(a) VALUES(1)");
+
+$in .= q{
+COMMIT;
+\q
+};
+$h->finish;
+
+$node_subscriber->wait_for_log(
+	qr/ERROR: ( [A-Z0-9]+:)? cannot replicate target relation "public.test_tab1" using subscription parameter streaming = parallel/,
+	$offset);
+
+# Drop domain constraint expression, now it works.
+$node_subscriber->safe_psql('postgres',
+	"ALTER DOMAIN tmp_domain DROP CONSTRAINT domain_check");
+
+# Wait for this streaming transaction to be applied in the apply worker.
+$node_publisher->wait_for_catchup($appname);
+
+$result =
+  $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab1");
+is($result, qq(5001),
+	'data replicated to subscriber after using immutable expression');
+
+# Clean up test data from the environment.
+$node_subscriber->safe_psql('postgres',
+	"ALTER TABLE test_tab1 ALTER COLUMN a TYPE int");
+$node_publisher->safe_psql(
+	'postgres', qq{
+TRUNCATE TABLE test_tab1;
+TRUNCATE TABLE test_tab2;
+});
+$node_publisher->wait_for_catchup($appname);
+
+# ============================================================================
+# It is not allowed that constraint expression contains a non-immutable function
+# on the subscriber side. Check the error reported by parallel worker in this
+# case.
+# ============================================================================
+
+# First we check the constraint expression on normal table.
+$node_publisher->safe_psql('postgres', "INSERT INTO test_tab2 VALUES(1)");
+
+$node_subscriber->safe_psql(
+	'postgres', qq{
+ALTER TABLE test_tab1 ADD CONSTRAINT const_tab1_unsafe CHECK(a >= func_count_tab2());
+});
+
+# Check the subscriber log from now on.
+$offset = -s $node_subscriber->logfile;
+
+$in .= q{
+BEGIN;
+TRUNCATE test_tab2;
+INSERT INTO test_tab1 SELECT i FROM generate_series(1, 5000) s(i);
+};
+$h->pump_nb;
+
+$node_publisher->safe_psql('postgres', "INSERT INTO test_tab1(a) VALUES(1)");
+
+$in .= q{
+COMMIT;
+\q
+};
+$h->finish;
+
+$node_subscriber->wait_for_log(
+	qr/ERROR: ( [A-Z0-9]+:)? cannot replicate target relation "public.test_tab1" using subscription parameter streaming = parallel/,
+	$offset);
+
+# Alter constraint expression to immutable expression, now it works.
+$node_subscriber->safe_psql(
+	'postgres', qq{
+ALTER TABLE test_tab1 DROP CONSTRAINT const_tab1_unsafe;
+ALTER TABLE test_tab1 ADD CONSTRAINT const_tab1_safe CHECK(a >= 0);
+});
+
+# Wait for this streaming transaction to be applied in the apply worker.
+$node_publisher->wait_for_catchup($appname);
+
+$result =
+  $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab1");
+is($result, qq(5001),
+	'data replicated to subscriber after using immutable expression');
+
+# Clean up test data from the environment.
+$node_subscriber->safe_psql('postgres',
+	"ALTER TABLE test_tab1 DROP CONSTRAINT const_tab1_safe");
+$node_publisher->safe_psql(
+	'postgres', qq{
+TRUNCATE TABLE test_tab1;
+TRUNCATE TABLE test_tab2;
+});
+$node_publisher->wait_for_catchup($appname);
+
+# Then we check the constraint expression on partition table.
+$node_subscriber->safe_psql(
+	'postgres', qq{
+ALTER TABLE test_tab_partition ADD CONSTRAINT test_tab_con check (a > random());
+});
+
+# Check the subscriber log from now on.
+$offset = -s $node_subscriber->logfile;
+
+$node_publisher->safe_psql('postgres', "DELETE FROM test_tab_partitioned");
+
+$node_subscriber->wait_for_log(
+	qr/ERROR: ( [A-Z0-9]+:)? cannot replicate target relation "public.test_tab_partitioned" using subscription parameter streaming = parallel/,
+	$offset);
+
+# Drop constraint on the subscriber, now it works.
+$node_subscriber->safe_psql('postgres',
+	"ALTER TABLE test_tab_partition DROP CONSTRAINT test_tab_con");
+
+# Wait for this streaming transaction to be applied in the apply worker.
+$node_publisher->wait_for_catchup($appname);
+
+$result =
+  $node_subscriber->safe_psql('postgres',
+	"SELECT count(*) FROM test_tab_partitioned");
+is($result, qq(0),
+	'data replicated to subscriber after dropping constraint expression');
+
+# ============================================================================
+# It is not allowed that foreign key on the subscriber side. Check the error
+# reported by parallel worker in this case.
+# ============================================================================
+
+# First we check the foreign key on normal table.
+$node_publisher->safe_psql(
+	'postgres', qq{
+CREATE TABLE tab_nopublic(a int);
+ALTER TABLE test_tab2 ADD PRIMARY KEY (a);
+ALTER TABLE test_tab2 REPLICA IDENTITY FULL;
+INSERT INTO test_tab2 VALUES(1);
+});
+$node_subscriber->safe_psql('postgres',
+	"ALTER TABLE test_tab2 ADD PRIMARY KEY (a);");
+
+$node_subscriber->safe_psql(
+	'postgres', qq{
+ALTER TABLE test_tab1 ADD CONSTRAINT test_tab1fk FOREIGN KEY(a) REFERENCES test_tab2(a);
+SELECT 'ALTER TABLE test_tab1 ENABLE REPLICA TRIGGER "' || tgname || '"' FROM pg_trigger WHERE tgrelid = 'test_tab1'::regclass::oid \\gexec
+});
+
+# Check the subscriber log from now on.
+$offset = -s $node_subscriber->logfile;
+
+$in .= q{
+BEGIN;
+INSERT INTO test_tab1(a) VALUES(1);
+INSERT INTO tab_nopublic SELECT i FROM generate_series(1, 5000) s(i);
+};
+$h->pump_nb;
+
+$node_publisher->safe_psql('postgres', "DELETE FROM test_tab2");
+
+$in .= q{
+COMMIT;
+\q
+};
+$h->finish;
+
+$node_subscriber->wait_for_log(
+	qr/ERROR: ( [A-Z0-9]+:)? cannot replicate target relation "public.test_tab1" using subscription parameter streaming = parallel/,
+	$offset);
+
+# Drop the foreign key constraint on the subscriber, now it works.
+$node_subscriber->safe_psql('postgres',
+	"ALTER TABLE test_tab1 DROP CONSTRAINT test_tab1fk");
+
+# Wait for this streaming transaction to be applied in the apply worker.
+$node_publisher->wait_for_catchup($appname);
+
+$result =
+  $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab1");
+is($result, qq(1),
+	'data replicated to subscriber after dropping the foreign key');
+
+# Clean up test data from the environment.
+$node_publisher->safe_psql(
+	'postgres', qq{
+TRUNCATE TABLE test_tab1;
+TRUNCATE TABLE test_tab2;
+});
+$node_publisher->wait_for_catchup($appname);
+
+# Then we check the foreign key on partition table.
+$node_subscriber->safe_psql(
+	'postgres', qq{
+CREATE TABLE test_tab_partition_f (a int primary key);
+ALTER TABLE test_tab_partition ADD CONSTRAINT test_tab_patition_fk FOREIGN KEY(a) REFERENCES test_tab_partition_f(a);
+SELECT 'ALTER TABLE test_tab_partition ENABLE REPLICA TRIGGER "' || tgname || '"' FROM pg_trigger WHERE tgrelid = 'test_tab_partition'::regclass::oid \\gexec
+});
+
+# Check the subscriber log from now on.
+$offset = -s $node_subscriber->logfile;
+
+$node_publisher->safe_psql('postgres',
+	"INSERT INTO test_tab_partitioned SELECT i, md5(i::text) FROM generate_series(1, 5000) s(i)"
+);
+
+$node_subscriber->wait_for_log(
+	qr/ERROR: ( [A-Z0-9]+:)? cannot replicate target relation "public.test_tab_partitioned" using subscription parameter streaming = parallel/,
+	$offset);
+
+# Drop the foreign key constraint on the subscriber, now it works.
+$node_subscriber->safe_psql('postgres',
+	"ALTER TABLE test_tab_partition DROP CONSTRAINT test_tab_patition_fk");
+
+# Wait for this streaming transaction to be applied in the apply worker.
+$node_publisher->wait_for_catchup($appname);
+
+$result =
+  $node_subscriber->safe_psql('postgres',
+	"SELECT count(*) FROM test_tab_partitioned");
+is($result, qq(5000),
+	'data replicated to subscriber after dropping the foreign key');
+
+$node_subscriber->stop;
+$node_publisher->stop;
+
+done_testing();
-- 
2.23.0.windows.1

