From 72cb0dfcf9a128783b148c5fbaa0f7f347af04e2 Mon Sep 17 00:00:00 2001
From: Zhijie Hou <houzj.fnst@fujitsu.com>
Date: Tue, 4 Aug 2026 10:23:00 +0800
Subject: [PATCH v2 3/3] Support parallel SELECT for INSERT ... SELECT

Allow INSERT ... SELECT to use a parallel plan for the underlying
SELECT, when it is safe to modify the target table while in parallel
mode.  All writes are still performed by the leader backend; the
parallel workers only run the SELECT.

The planner's parallelModeOK gate now admits a table-modification
statement when is_parallel_allowed_for_modify() says so: currently only
INSERT with an underlying SELECT, excluding INSERT ... ON CONFLICT ...
DO UPDATE (UPDATE in parallel mode would need combo CIDs to be shared
between leader and workers).  In addition, max_parallel_hazard() folds
in the parallel DML safety hazard level of the target relation, which
is cached in its relcache entry by RelationGetParallelDmlSafety(): a
parallel-unsafe trigger, index expression or predicate, check
constraint, column default or partition key on the target table (or on
any partition of a partitioned target table) disqualifies parallelism,
while a parallel-restricted object merely means the construct must be
evaluated in the leader, which is where the INSERT happens anyway.

The executor rechecks each partition's cached hazard level (in
ExecInitPartitionInfo, once per partition per execution rather than per row or
per function call) before inserting into it while in parallel mode, and errors
out if it has become unsafe.  This is needed because adding a parallel-unsafe
object to a partition does not lock the partition's ancestors, so the
invalidation of the ancestors' cached hazard level could not be processed in
time by a backend running a parallel INSERT SELECT; it also covers the narrow
window where invalidations are processed in the middle of execution, e.g. while
waiting for a lock.

A follow-on commit will invalidate cached plans for the target table when a
parallel-unsafe object is added to it, or when any function's parallel safety
is changed. This will be done using the parallel DML safety invalidation
message, ensuring that subsequent parallel INSERT ... SELECT statements see
the updated hazard level.

For anything else that slips through (e.g. a volatile function
attempting a write from a trigger), the pre-existing defenses in the
executor (read-only workers, XID/CID assignment, nextval, ...) raise an
error.

On the executor side, PrepareParallelModePlanExec() assigns a
TransactionId before entering parallel mode, since heap_insert() cannot
assign one once in parallel mode.  This wastes an XID when the SELECT
returns no rows, which should be rare enough not to matter.  The
command ID used to mark inserted tuples is already assigned in
ExecutorStart, before parallel mode begins.

The new insert_parallel regression test verifies that a Gather appears
exactly when expected (including plan flips after dropping an unsafe
object or altering a function's parallel safety), that INSERT ... ON
CONFLICT DO UPDATE and INSERT ... VALUES are not parallelized, the
cached-plan invalidation behavior described above, and that the
inserted rows all carry the same xmin/cmin.  The new
parallel-insert-unsafe-partition isolation test verifies the
per-partition runtime check with a deterministic two-session scenario.
---
 doc/src/sgml/parallel.sgml                    |  20 +
 src/backend/access/transam/xact.c             |  26 ++
 src/backend/executor/execMain.c               |   9 +-
 src/backend/executor/execPartition.c          |  26 ++
 src/backend/optimizer/plan/planner.c          |  29 +-
 src/backend/optimizer/util/clauses.c          |  94 ++++-
 src/include/access/xact.h                     |  16 +
 src/include/optimizer/clauses.h               |   1 +
 .../parallel-insert-unsafe-partition.out      |  20 +
 src/test/isolation/isolation_schedule         |   1 +
 .../parallel-insert-unsafe-partition.spec     |  50 +++
 src/test/regress/expected/insert_parallel.out | 355 ++++++++++++++++++
 src/test/regress/parallel_schedule            |   1 +
 src/test/regress/sql/insert_parallel.sql      | 206 ++++++++++
 14 files changed, 839 insertions(+), 15 deletions(-)
 create mode 100644 src/test/isolation/expected/parallel-insert-unsafe-partition.out
 create mode 100644 src/test/isolation/specs/parallel-insert-unsafe-partition.spec
 create mode 100644 src/test/regress/expected/insert_parallel.out
 create mode 100644 src/test/regress/sql/insert_parallel.sql

diff --git a/doc/src/sgml/parallel.sgml b/doc/src/sgml/parallel.sgml
index af43484703e..72c87d3a72a 100644
--- a/doc/src/sgml/parallel.sgml
+++ b/doc/src/sgml/parallel.sgml
@@ -155,6 +155,9 @@ EXPLAIN SELECT * FROM pgbench_accounts WHERE filler LIKE '%x%';
           <listitem>
             <para><command>SELECT INTO</command></para>
           </listitem>
+          <listitem>
+            <para><command>INSERT INTO ... SELECT</command></para>
+          </listitem>
           <listitem>
             <para><command>CREATE MATERIALIZED VIEW</command></para>
           </listitem>
@@ -162,6 +165,23 @@ EXPLAIN SELECT * FROM pgbench_accounts WHERE filler LIKE '%x%';
             <para><command>REFRESH MATERIALIZED VIEW</command></para>
           </listitem>
         </itemizedlist>
+
+        Note that <command>INSERT INTO ... SELECT</command> can use a
+        parallel plan for the underlying <literal>SELECT</literal> part of
+        the query only if the query has no <literal>ON CONFLICT ... DO
+        UPDATE</literal> clause and the target table is parallel-safe. All
+        writes are performed by the leader backend, but it must be possible
+        to run in parallel mode while modifying the target table, so the
+        table must not have any parallel-unsafe triggers, index expressions
+        or predicates, check constraints, column default expressions, stored
+        generated expressions, or partition key expressions (a
+        parallel-restricted construct does not prevent this, since it is only
+        evaluated in the leader).  For an <command>INSERT</command> into a
+        partitioned table, every partition must be parallel-safe as well. If an
+        unsafe object is added to a partition while such a query is executing,
+        for example, by a concurrent <command>CREATE TRIGGER</command>, the
+        query detects this before inserting into that partition and fails with
+        an error.
       </para>
     </listitem>
 
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index 3a89149016f..a485499f606 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -1123,6 +1123,32 @@ IsInParallelMode(void)
 	return s->parallelModeLevel != 0 || s->parallelChildXact;
 }
 
+/*
+ *	PrepareParallelModePlanExec
+ *
+ * Prepare for entering parallel mode plan execution, based on command-type.
+ */
+void
+PrepareParallelModePlanExec(CmdType commandType)
+{
+	if (IsModifySupportedInParallelMode(commandType))
+	{
+		Assert(!IsInParallelMode());
+
+		/*
+		 * Prepare for entering parallel mode by assigning a TransactionId.
+		 * Failure to do this now would result in heap_insert() subsequently
+		 * attempting to assign a TransactionId whilst in parallel mode,
+		 * which is not allowed.
+		 *
+		 * This approach has a disadvantage in that if the underlying SELECT
+		 * does not return any rows, then the TransactionId is not used,
+		 * however that shouldn't happen in practice in many cases.
+		 */
+		(void) GetCurrentTransactionId();
+	}
+}
+
 /*
  *	CommandCounterIncrement
  */
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index fde502efd38..624c775478c 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -160,8 +160,10 @@ standard_ExecutorStart(QueryDesc *queryDesc, int eflags)
 	 * would require (a) storing the combo CID hash in shared memory, rather
 	 * than synchronizing it just once at the start of parallelism, and (b) an
 	 * alternative to heap_update()'s reliance on xmax for mutual exclusion.
-	 * INSERT may have no such troubles, but we forbid it to simplify the
-	 * checks.
+	 * INSERT has no such troubles, but it is only allowed for a top-level
+	 * INSERT ... SELECT that enters parallel mode after this check; a
+	 * command that is started while already in parallel mode (e.g. from a
+	 * function running in a worker) must not write.
 	 *
 	 * We have lower-level defenses in CommandCounterIncrement and elsewhere
 	 * against performing unsafe operations in parallel mode, but this gives a
@@ -1757,7 +1759,10 @@ ExecutePlan(QueryDesc *queryDesc,
 
 	estate->es_use_parallel_mode = use_parallel_mode;
 	if (use_parallel_mode)
+	{
+		PrepareParallelModePlanExec(queryDesc->plannedstmt->commandType);
 		EnterParallelMode();
+	}
 
 	/*
 	 * Loop until we've processed the proper number of tuples from the plan.
diff --git a/src/backend/executor/execPartition.c b/src/backend/executor/execPartition.c
index 9b4738884af..f6df143a7a6 100644
--- a/src/backend/executor/execPartition.c
+++ b/src/backend/executor/execPartition.c
@@ -18,12 +18,14 @@
 #include "access/tupconvert.h"
 #include "catalog/index.h"
 #include "catalog/partition.h"
+#include "catalog/pg_proc.h"
 #include "executor/execPartition.h"
 #include "executor/executor.h"
 #include "executor/nodeModifyTable.h"
 #include "foreign/fdwapi.h"
 #include "mb/pg_wchar.h"
 #include "miscadmin.h"
+#include "optimizer/clauses.h"
 #include "partitioning/partbounds.h"
 #include "partitioning/partdesc.h"
 #include "partitioning/partprune.h"
@@ -588,6 +590,30 @@ ExecInitPartitionInfo(ModifyTableState *mtstate, EState *estate,
 					  rootResultRelInfo,
 					  estate->es_instrument);
 
+	/*
+	 * When in parallel mode, the partition must be safe to modify in
+	 * parallel mode.  The planner checked the target relation as a whole
+	 * using the hazard level cached in its relcache entry, but that check
+	 * can be out of date: adding a parallel-unsafe object to a partition
+	 * does not lock the partition's ancestors, so neither the invalidation
+	 * of the ancestors' cached hazard level nor the invalidation of cached
+	 * plans is ensured to be processed in time by a backend running a
+	 * parallel INSERT.  A transaction committing such a change
+	 * concurrently can slip it in after the plan was made (or even after
+	 * execution started, with the invalidation being processed
+	 * mid-execution, e.g. while this statement was waiting for a lock).
+	 * Recheck the partition's own cached hazard level now, once per
+	 * partition per execution rather than per row, and error out rather
+	 * than risk modifying an unsafe partition in parallel mode.
+	 */
+	if (IsInParallelMode() &&
+		RelationGetParallelDmlSafety(partrel) == PROPARALLEL_UNSAFE)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_TRANSACTION_STATE),
+				 errmsg("cannot execute INSERT with a parallel plan on relation \"%s\"",
+						RelationGetRelationName(partrel)),
+				 errdetail("The relation has parallel-unsafe objects, which may have appeared after the plan was made.")));
+
 	/*
 	 * Verify result relation is a valid target for an INSERT.  An UPDATE of a
 	 * partition-key becomes a DELETE+INSERT operation, so this check is still
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
index a0ff9159ae0..0439375ba93 100644
--- a/src/backend/optimizer/plan/planner.c
+++ b/src/backend/optimizer/plan/planner.c
@@ -392,18 +392,22 @@ standard_planner(Query *parse, const char *query_string, int cursorOptions,
 	/*
 	 * Assess whether it's feasible to use parallel mode for this query. We
 	 * can't do this in a standalone backend, or if the command will try to
-	 * modify any data, or if this is a cursor operation, or if GUCs are set
-	 * to values that don't permit parallelism, or if parallel-unsafe
-	 * functions are present in the query tree.
+	 * modify any data (except for INSERT ... SELECT), or if this is a
+	 * cursor operation, or if GUCs are set to values that don't permit
+	 * parallelism, or if parallel-unsafe functions are present in the query
+	 * tree.
 	 *
-	 * (Note that we do allow CREATE TABLE AS, SELECT INTO, and CREATE
-	 * MATERIALIZED VIEW to use parallel plans, but this is safe only because
-	 * the command is writing into a completely new table which workers won't
-	 * be able to see.  If the workers could see the table, the fact that
-	 * group locking would cause them to ignore the leader's heavyweight GIN
-	 * page locks would make this unsafe.  We'll have to fix that somehow if
-	 * we want to allow parallel inserts in general; updates and deletes have
-	 * additional problems especially around combo CIDs.)
+	 * (Note that we do allow CREATE TABLE AS, INSERT INTO ... SELECT,
+	 * SELECT INTO, and CREATE MATERIALIZED VIEW to use parallel plans for
+	 * the underlying SELECT.  For INSERT INTO ... SELECT, this is safe only
+	 * because the leader backend performs all the writes, and only when the
+	 * target table has no parallel-unsafe objects attached to it (this is
+	 * checked in max_parallel_hazard, using the hazard level cached in the
+	 * target table's relcache entry).  If the workers could see the table,
+	 * the fact that group locking would cause them to ignore the leader's
+	 * heavyweight GIN page locks would make this unsafe.  We'll have to fix
+	 * that somehow if we want to allow parallel inserts in general; updates
+	 * and deletes have additional problems especially around combo CIDs.)
 	 *
 	 * For now, we don't try to use parallel mode if we're running inside a
 	 * parallel worker.  We might eventually be able to relax this
@@ -412,7 +416,8 @@ standard_planner(Query *parse, const char *query_string, int cursorOptions,
 	 */
 	if ((cursorOptions & CURSOR_OPT_PARALLEL_OK) != 0 &&
 		IsUnderPostmaster &&
-		parse->commandType == CMD_SELECT &&
+		(parse->commandType == CMD_SELECT ||
+		 is_parallel_allowed_for_modify(parse)) &&
 		!parse->hasModifyingCTE &&
 		max_parallel_workers_per_gather > 0 &&
 		!IsParallelWorker())
diff --git a/src/backend/optimizer/util/clauses.c b/src/backend/optimizer/util/clauses.c
index 2e981ae27b9..c8483cec727 100644
--- a/src/backend/optimizer/util/clauses.c
+++ b/src/backend/optimizer/util/clauses.c
@@ -22,6 +22,7 @@
 #include "access/genam.h"
 #include "access/htup_details.h"
 #include "access/table.h"
+#include "access/xact.h"
 #include "catalog/pg_class.h"
 #include "catalog/pg_inherits.h"
 #include "catalog/pg_language.h"
@@ -124,6 +125,8 @@ static bool contain_volatile_functions_walker(Node *node, void *context);
 static bool contain_volatile_functions_not_nextval_walker(Node *node, void *context);
 static bool max_parallel_hazard_walker(Node *node,
 									   max_parallel_hazard_context *context);
+static bool max_parallel_hazard_test(char proparallel,
+									 max_parallel_hazard_context *context);
 static bool parallel_dml_hazard_walker(Node *node,
 									   max_parallel_hazard_context *context);
 static bool table_parallel_dml_hazard_recurse(Relation rel,
@@ -783,13 +786,41 @@ contain_volatile_functions_not_nextval_walker(Node *node, void *context)
 char
 max_parallel_hazard(Query *parse)
 {
+	bool		max_hazard_found;
 	max_parallel_hazard_context context;
 
 	context.max_hazard = PROPARALLEL_SAFE;
 	context.max_interesting = PROPARALLEL_UNSAFE;
 	context.safe_param_ids = NIL;
 	context.partition_directory = NULL;
-	(void) max_parallel_hazard_walker((Node *) parse, &context);
+
+	max_hazard_found = max_parallel_hazard_walker((Node *) parse, &context);
+
+	/*
+	 * For a table-modification statement that can be executed in parallel
+	 * mode, the worst hazard found among the objects attached to the target
+	 * relation counts too; the hazard level is cached in the relation's
+	 * relcache entry.
+	 */
+	if (!max_hazard_found &&
+		IsModifySupportedInParallelMode(parse->commandType))
+	{
+		RangeTblEntry *rte = rt_fetch(parse->resultRelation, parse->rtable);
+		Relation	target_rel;
+
+		/*
+		 * The target table is already locked by the caller (this is done in
+		 * the parse/analyze phase), and remains locked until end of
+		 * transaction.
+		 */
+		target_rel = table_open(rte->relid, NoLock);
+
+		(void) max_parallel_hazard_test(RelationGetParallelDmlSafety(target_rel),
+										&context);
+
+		table_close(target_rel, NoLock);
+	}
+
 	return context.max_hazard;
 }
 
@@ -1024,6 +1055,67 @@ max_parallel_hazard_walker(Node *node, max_parallel_hazard_context *context)
 								  context);
 }
 
+/*
+ * is_parallel_allowed_for_modify
+ *
+ * Check at a high-level if parallel mode is able to be used for the
+ * specified table-modification statement.  Currently, we support only
+ * INSERT.
+ *
+ * It's not possible in the following cases:
+ *
+ *  1) INSERT ... ON CONFLICT ... DO UPDATE
+ *  2) INSERT without SELECT
+ *
+ * (Note: we don't do in-depth parallel-safety checks here; we do only the
+ * cheaper tests that can quickly exclude obvious cases for which
+ * parallelism isn't supported, to avoid having to do further
+ * parallel-safety checks for these.  The parallel safety of the target
+ * relation is checked separately in max_parallel_hazard.)
+ */
+bool
+is_parallel_allowed_for_modify(Query *parse)
+{
+	bool		hasSubQuery;
+	RangeTblEntry *rte;
+	ListCell   *lc;
+
+	if (!IsModifySupportedInParallelMode(parse->commandType))
+		return false;
+
+	/*
+	 * UPDATE is not currently supported in parallel mode, so prohibit
+	 * INSERT ... ON CONFLICT ... DO UPDATE.
+	 *
+	 * In order to support UPDATE, even if only in the leader, some further
+	 * work would need to be done.  A mechanism would be needed for sharing
+	 * combo CIDs between leader and workers during parallel mode, since for
+	 * example, the leader might generate a combo CID and it would need to
+	 * be propagated to the workers.
+	 */
+	if (parse->commandType == CMD_INSERT &&
+		parse->onConflict != NULL &&
+		parse->onConflict->action == ONCONFLICT_UPDATE)
+		return false;
+
+	/*
+	 * If there is no underlying SELECT, a parallel insert operation is not
+	 * desirable.
+	 */
+	hasSubQuery = false;
+	foreach(lc, parse->rtable)
+	{
+		rte = lfirst_node(RangeTblEntry, lc);
+		if (rte->rtekind == RTE_SUBQUERY)
+		{
+			hasSubQuery = true;
+			break;
+		}
+	}
+
+	return hasSubQuery;
+}
+
 /*
  * The workhorse for RelationGetParallelDmlSafety(). Recursively examine a
  * relation and all of the objects attached to it for PARALLEL UNSAFE/RESTRICTED
diff --git a/src/include/access/xact.h b/src/include/access/xact.h
index a8cbdf247c8..426633db82f 100644
--- a/src/include/access/xact.h
+++ b/src/include/access/xact.h
@@ -18,6 +18,7 @@
 #include "access/xlogreader.h"
 #include "datatype/timestamp.h"
 #include "lib/stringinfo.h"
+#include "nodes/nodes.h"
 #include "nodes/pg_list.h"
 #include "storage/relfilelocator.h"
 #include "storage/sinval.h"
@@ -534,5 +535,20 @@ extern void ParsePrepareRecord(uint8 info, xl_xact_prepare *xlrec, xl_xact_parse
 extern void EnterParallelMode(void);
 extern void ExitParallelMode(void);
 extern bool IsInParallelMode(void);
+extern void PrepareParallelModePlanExec(CmdType commandType);
+
+/*
+ * IsModifySupportedInParallelMode
+ *
+ * Indicates whether execution of the specified table-modification command
+ * (INSERT/UPDATE/DELETE) in parallel mode is supported, subject to certain
+ * parallel-safety conditions.
+ */
+static inline bool
+IsModifySupportedInParallelMode(CmdType commandType)
+{
+	/* Currently only INSERT is supported */
+	return (commandType == CMD_INSERT);
+}
 
 #endif							/* XACT_H */
diff --git a/src/include/optimizer/clauses.h b/src/include/optimizer/clauses.h
index d43f45d0f23..71bd460d0ba 100644
--- a/src/include/optimizer/clauses.h
+++ b/src/include/optimizer/clauses.h
@@ -45,6 +45,7 @@ extern bool contain_subplans(Node *clause);
 
 extern char max_parallel_hazard(Query *parse);
 extern bool is_parallel_safe(PlannerInfo *root, Node *node);
+extern bool is_parallel_allowed_for_modify(Query *parse);
 extern char max_parallel_dml_hazard(Relation rel);
 extern bool contain_nonstrict_functions(Node *clause);
 extern bool contain_exec_param(Node *clause, List *param_ids);
diff --git a/src/test/isolation/expected/parallel-insert-unsafe-partition.out b/src/test/isolation/expected/parallel-insert-unsafe-partition.out
new file mode 100644
index 00000000000..7209dd72257
--- /dev/null
+++ b/src/test/isolation/expected/parallel-insert-unsafe-partition.out
@@ -0,0 +1,20 @@
+Parsed test spec with 2 sessions
+
+starting permutation: lb llock iset iins ltrg lc
+step lb: BEGIN;
+step llock: SELECT pg_advisory_xact_lock(1);
+pg_advisory_xact_lock
+---------------------
+                     
+(1 row)
+
+step iset: SET debug_parallel_query = on;
+step iins: INSERT INTO pi_root
+			  SELECT CASE WHEN i = 1 THEN 50 ELSE 1 END
+			  FROM generate_series(1, 2) i
+			  WHERE (i <> 1 OR pg_advisory_lock(1) IS NOT NULL); <waiting ...>
+step ltrg: CREATE TRIGGER trg_bad BEFORE INSERT ON pi_p1
+			  FOR EACH ROW EXECUTE FUNCTION pi_unsafe_fn();
+step lc: COMMIT;
+step iins: <... completed>
+ERROR:  cannot execute INSERT with a parallel plan on relation "pi_p1"
diff --git a/src/test/isolation/isolation_schedule b/src/test/isolation/isolation_schedule
index a27480a86a2..8fd3f7545b9 100644
--- a/src/test/isolation/isolation_schedule
+++ b/src/test/isolation/isolation_schedule
@@ -131,3 +131,4 @@ test: for-portion-of
 test: ddl-dependency-locking
 test: pub-concurrent-drop
 test: drop-owned-grant
+test: parallel-insert-unsafe-partition
diff --git a/src/test/isolation/specs/parallel-insert-unsafe-partition.spec b/src/test/isolation/specs/parallel-insert-unsafe-partition.spec
new file mode 100644
index 00000000000..c4b4363f41f
--- /dev/null
+++ b/src/test/isolation/specs/parallel-insert-unsafe-partition.spec
@@ -0,0 +1,50 @@
+# Test that inserting into a partition that has become parallel-unsafe after
+# the statement started execution fails, rather than running with a stale
+# safety assumption.
+#
+# The inserter runs an INSERT into a partitioned table in parallel mode
+# (forced with debug_parallel_query; the plan has no Gather, so everything
+# runs in the leader, in a deterministic row order).  Evaluation of the
+# first row waits for an advisory lock held by the other session.  While
+# the inserter is blocked, the other session adds a parallel-unsafe trigger
+# to partition pi_p1 and commits.  When the inserter wakes up, it processes
+# the invalidation messages (during the lock wait), so when the next row is
+# routed to pi_p1, the execution-time check in ExecInitPartitionInfo
+# detects that pi_p1 is no longer parallel-safe and errors out.
+
+setup
+{
+  CREATE FUNCTION pi_unsafe_fn() RETURNS trigger
+    LANGUAGE plpgsql PARALLEL UNSAFE AS $$
+  BEGIN
+    RETURN NEW;
+  END $$;
+
+  CREATE TABLE pi_root (a int) PARTITION BY RANGE (a);
+  CREATE TABLE pi_p1 PARTITION OF pi_root FOR VALUES FROM (0) TO (50);
+  CREATE TABLE pi_p2 PARTITION OF pi_root FOR VALUES FROM (50) TO (100);
+}
+
+teardown
+{
+  DROP TABLE pi_root;
+  DROP FUNCTION pi_unsafe_fn();
+}
+
+session inserter
+step iset	{ SET debug_parallel_query = on; }
+# the first row (50, routed to pi_p2) waits for the advisory lock; the
+# second row (1, routed to pi_p1) is evaluated afterwards
+step iins	{ INSERT INTO pi_root
+			  SELECT CASE WHEN i = 1 THEN 50 ELSE 1 END
+			  FROM generate_series(1, 2) i
+			  WHERE (i <> 1 OR pg_advisory_lock(1) IS NOT NULL); }
+
+session locker
+step lb		{ BEGIN; }
+step llock	{ SELECT pg_advisory_xact_lock(1); }
+step ltrg	{ CREATE TRIGGER trg_bad BEFORE INSERT ON pi_p1
+			  FOR EACH ROW EXECUTE FUNCTION pi_unsafe_fn(); }
+step lc		{ COMMIT; }
+
+permutation lb llock iset iins ltrg lc
diff --git a/src/test/regress/expected/insert_parallel.out b/src/test/regress/expected/insert_parallel.out
new file mode 100644
index 00000000000..0fc70ce4c34
--- /dev/null
+++ b/src/test/regress/expected/insert_parallel.out
@@ -0,0 +1,355 @@
+--
+-- PARALLEL INSERT ... SELECT
+--
+--
+-- START: setup some functions needed by the tests.
+--
+-- For testing purposes, we'll mark these functions as parallel-unsafe or
+-- parallel-restricted; plpgsql so that they are not inlined into the
+-- stored expressions
+create function fullname_parallel_unsafe(f text, l text) returns text
+language plpgsql immutable parallel unsafe as $$
+begin
+	return f || l;
+end; $$;
+create function fullname_parallel_restricted(f text, l text) returns text
+language plpgsql immutable parallel restricted as $$
+begin
+	return f || l;
+end; $$;
+create function bdefault_unsafe() returns int
+language plpgsql parallel unsafe as $$
+begin
+	return 5;
+end; $$;
+create function cdefault_restricted() returns int
+language plpgsql parallel restricted as $$
+begin
+	return 10;
+end; $$;
+create function trg_unsafe_fn() returns trigger
+language plpgsql parallel unsafe as $$
+begin
+	return new;
+end; $$;
+create function trg_restricted_fn() returns trigger
+language plpgsql parallel restricted as $$
+begin
+	return new;
+end; $$;
+--
+-- END: setup.
+--
+begin;
+-- encourage use of parallel plans
+set parallel_setup_cost = 0;
+set parallel_tuple_cost = 0;
+set min_parallel_table_scan_size = 0;
+set max_parallel_workers_per_gather = 4;
+-- force seq scans, for deterministic plans
+set enable_indexonlyscan = off;
+set enable_indexscan = off;
+set enable_bitmapscan = off;
+--
+-- A plain target table has no parallel-unsafe objects, so the SELECT part
+-- can be parallelized.
+--
+create table para_insert (a int, b name);
+explain (costs off) insert into para_insert select unique1, stringu1 from tenk1;
+               QUERY PLAN               
+----------------------------------------
+ Insert on para_insert
+   ->  Gather
+         Workers Planned: 4
+         ->  Parallel Seq Scan on tenk1
+(4 rows)
+
+insert into para_insert select unique1, stringu1 from tenk1;
+-- select some values to verify that the parallel insert worked
+select count(*), sum(a) from para_insert;
+ count |   sum    
+-------+----------
+ 10000 | 49995000
+(1 row)
+
+-- verify that the same transaction and command has been used for all rows
+select count(*) from (select distinct cmin, xmin from para_insert) as dt;
+ count 
+-------
+     1
+(1 row)
+
+-- INSERT ... VALUES (no underlying SELECT) can't be parallelized
+explain (costs off) insert into para_insert values (1, 'one');
+      QUERY PLAN       
+-----------------------
+ Insert on para_insert
+   ->  Result
+(2 rows)
+
+-- a parallel-unsafe function in the SELECT part can't be parallelized
+explain (costs off)
+insert into para_insert select unique1, fullname_parallel_unsafe(stringu1, 'x') from tenk1;
+       QUERY PLAN        
+-------------------------
+ Insert on para_insert
+   ->  Seq Scan on tenk1
+(2 rows)
+
+--
+-- A parallel-unsafe trigger on the target table prevents parallel SELECT.
+--
+create trigger trg before insert on para_insert
+	for each row execute function trg_unsafe_fn();
+explain (costs off) insert into para_insert select unique1, stringu1 from tenk1;
+       QUERY PLAN        
+-------------------------
+ Insert on para_insert
+   ->  Seq Scan on tenk1
+(2 rows)
+
+-- dropping it makes the target parallel-safe again
+drop trigger trg on para_insert;
+explain (costs off) insert into para_insert select unique1, stringu1 from tenk1;
+               QUERY PLAN               
+----------------------------------------
+ Insert on para_insert
+   ->  Gather
+         Workers Planned: 4
+         ->  Parallel Seq Scan on tenk1
+(4 rows)
+
+--
+-- A parallel-restricted trigger on the target table still allows parallel
+-- SELECT; the trigger fires in the leader.
+--
+create trigger trg before insert on para_insert
+	for each row execute function trg_restricted_fn();
+explain (costs off) insert into para_insert select unique1, stringu1 from tenk1;
+               QUERY PLAN               
+----------------------------------------
+ Insert on para_insert
+   ->  Gather
+         Workers Planned: 4
+         ->  Parallel Seq Scan on tenk1
+(4 rows)
+
+drop trigger trg on para_insert;
+--
+-- A parallel-unsafe check constraint on the target table prevents
+-- parallel SELECT.
+--
+alter table para_insert add constraint chk_u check (fullname_parallel_unsafe(b, 'x') <> '');
+explain (costs off) insert into para_insert select unique1, stringu1 from tenk1;
+       QUERY PLAN        
+-------------------------
+ Insert on para_insert
+   ->  Seq Scan on tenk1
+(2 rows)
+
+alter table para_insert drop constraint chk_u;
+explain (costs off) insert into para_insert select unique1, stringu1 from tenk1;
+               QUERY PLAN               
+----------------------------------------
+ Insert on para_insert
+   ->  Gather
+         Workers Planned: 4
+         ->  Parallel Seq Scan on tenk1
+(4 rows)
+
+--
+-- A parallel-unsafe index expression on the target table prevents
+-- parallel SELECT.
+--
+create index para_insert_unsafe_idx on para_insert (fullname_parallel_unsafe(b, 'x'));
+explain (costs off) insert into para_insert select unique1, stringu1 from tenk1;
+       QUERY PLAN        
+-------------------------
+ Insert on para_insert
+   ->  Seq Scan on tenk1
+(2 rows)
+
+drop index para_insert_unsafe_idx;
+explain (costs off) insert into para_insert select unique1, stringu1 from tenk1;
+               QUERY PLAN               
+----------------------------------------
+ Insert on para_insert
+   ->  Gather
+         Workers Planned: 4
+         ->  Parallel Seq Scan on tenk1
+(4 rows)
+
+--
+-- Column defaults are checked: an unsafe default prevents parallel SELECT,
+-- while a restricted one is allowed (defaults are evaluated in the leader).
+--
+create table testdef (a int,
+					  b int default bdefault_unsafe(),
+					  c int default cdefault_restricted());
+explain (costs off) insert into testdef select unique1 from tenk1;
+       QUERY PLAN        
+-------------------------
+ Insert on testdef
+   ->  Seq Scan on tenk1
+(2 rows)
+
+alter table testdef alter column b drop default;
+explain (costs off) insert into testdef select unique1 from tenk1;
+               QUERY PLAN               
+----------------------------------------
+ Insert on testdef
+   ->  Gather
+         Workers Planned: 4
+         ->  Parallel Seq Scan on tenk1
+(4 rows)
+
+--
+-- For an INSERT into a partitioned table, every partition must be
+-- parallel-safe.
+--
+create table part_para_insert (a int) partition by range (a);
+create table part_para_insert_p1 partition of part_para_insert
+	for values from (0) to (5000);
+create table part_para_insert_p2 partition of part_para_insert
+	for values from (5000) to (10000);
+explain (costs off) insert into part_para_insert select unique1 from tenk1;
+               QUERY PLAN               
+----------------------------------------
+ Insert on part_para_insert
+   ->  Gather
+         Workers Planned: 4
+         ->  Parallel Seq Scan on tenk1
+(4 rows)
+
+insert into part_para_insert select unique1 from tenk1;
+select count(*) from part_para_insert;
+ count 
+-------
+ 10000
+(1 row)
+
+-- inserting into a partition directly is fine too
+explain (costs off) insert into part_para_insert_p1 select unique1 from tenk1;
+               QUERY PLAN               
+----------------------------------------
+ Insert on part_para_insert_p1
+   ->  Gather
+         Workers Planned: 4
+         ->  Parallel Seq Scan on tenk1
+(4 rows)
+
+-- a pre-existing unsafe trigger on one partition prevents parallel SELECT
+-- into the parent
+create table part_para_unsafe (a int) partition by range (a);
+create table part_para_unsafe_p1 partition of part_para_unsafe
+	for values from (0) to (10000);
+create trigger trg before insert on part_para_unsafe_p1
+	for each row execute function trg_unsafe_fn();
+explain (costs off) insert into part_para_unsafe select unique1 from tenk1;
+         QUERY PLAN         
+----------------------------
+ Insert on part_para_unsafe
+   ->  Seq Scan on tenk1
+(2 rows)
+
+--
+-- Changing a function's parallel safety invalidates the cached safety of
+-- the target table.
+--
+create function alterable_fn(int) returns bool
+language plpgsql immutable parallel safe as $$
+begin
+	return $1 > 0;
+end; $$;
+create table fn_para_insert (a int check (alterable_fn(a)));
+explain (costs off) insert into fn_para_insert select unique1 from tenk1;
+               QUERY PLAN               
+----------------------------------------
+ Insert on fn_para_insert
+   ->  Gather
+         Workers Planned: 4
+         ->  Parallel Seq Scan on tenk1
+(4 rows)
+
+alter function alterable_fn(int) parallel unsafe;
+explain (costs off) insert into fn_para_insert select unique1 from tenk1;
+        QUERY PLAN        
+--------------------------
+ Insert on fn_para_insert
+   ->  Seq Scan on tenk1
+(2 rows)
+
+alter function alterable_fn(int) parallel safe;
+explain (costs off) insert into fn_para_insert select unique1 from tenk1;
+               QUERY PLAN               
+----------------------------------------
+ Insert on fn_para_insert
+   ->  Gather
+         Workers Planned: 4
+         ->  Parallel Seq Scan on tenk1
+(4 rows)
+
+--
+-- INSERT ... ON CONFLICT DO UPDATE can't be parallelized, because UPDATE is
+-- not supported in parallel mode; ON CONFLICT DO NOTHING can.
+--
+create table para_insert_pk (a int primary key, b name);
+explain (costs off)
+insert into para_insert_pk select unique1, stringu1 from tenk1
+	on conflict (a) do update set b = excluded.b;
+                   QUERY PLAN                    
+-------------------------------------------------
+ Insert on para_insert_pk
+   Conflict Resolution: UPDATE
+   Conflict Arbiter Indexes: para_insert_pk_pkey
+   ->  Seq Scan on tenk1
+(4 rows)
+
+explain (costs off)
+insert into para_insert_pk select unique1, stringu1 from tenk1
+	on conflict (a) do nothing;
+                   QUERY PLAN                    
+-------------------------------------------------
+ Insert on para_insert_pk
+   Conflict Resolution: NOTHING
+   Conflict Arbiter Indexes: para_insert_pk_pkey
+   ->  Gather
+         Workers Planned: 4
+         ->  Parallel Seq Scan on tenk1
+(6 rows)
+
+--
+-- A temporary or foreign target table is parallel-restricted: only the
+-- leader can modify it, but the SELECT part can still be parallelized.
+--
+create temp table temp_para_insert (a int);
+explain (costs off) insert into temp_para_insert select unique1 from tenk1;
+               QUERY PLAN               
+----------------------------------------
+ Insert on temp_para_insert
+   ->  Gather
+         Workers Planned: 4
+         ->  Parallel Seq Scan on tenk1
+(4 rows)
+
+--
+-- Forcing parallel mode when the plan has no Gather must work too; the
+-- transaction ID has to be assigned before entering parallel mode.
+--
+set debug_parallel_query = on;
+explain (costs off) insert into para_insert select i, 'x'::name from generate_series(1, 10) i;
+                QUERY PLAN                
+------------------------------------------
+ Insert on para_insert
+   ->  Function Scan on generate_series i
+(2 rows)
+
+insert into para_insert select i, 'x'::name from generate_series(1, 10) i;
+select count(*) from para_insert;
+ count 
+-------
+ 10010
+(1 row)
+
+set debug_parallel_query = off;
+rollback;
diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule
index 8fa0a6c47fb..0c8108a62d7 100644
--- a/src/test/regress/parallel_schedule
+++ b/src/test/regress/parallel_schedule
@@ -88,6 +88,7 @@ test: rules psql psql_crosstab psql_pipeline amutils stats_ext collate.linux.utf
 # select_parallel depends on create_misc
 # ----------
 test: select_parallel
+test: insert_parallel
 test: write_parallel
 test: vacuum_parallel
 
diff --git a/src/test/regress/sql/insert_parallel.sql b/src/test/regress/sql/insert_parallel.sql
new file mode 100644
index 00000000000..0c01780a5e3
--- /dev/null
+++ b/src/test/regress/sql/insert_parallel.sql
@@ -0,0 +1,206 @@
+--
+-- PARALLEL INSERT ... SELECT
+--
+
+--
+-- START: setup some functions needed by the tests.
+--
+
+-- For testing purposes, we'll mark these functions as parallel-unsafe or
+-- parallel-restricted; plpgsql so that they are not inlined into the
+-- stored expressions
+create function fullname_parallel_unsafe(f text, l text) returns text
+language plpgsql immutable parallel unsafe as $$
+begin
+	return f || l;
+end; $$;
+
+create function fullname_parallel_restricted(f text, l text) returns text
+language plpgsql immutable parallel restricted as $$
+begin
+	return f || l;
+end; $$;
+
+create function bdefault_unsafe() returns int
+language plpgsql parallel unsafe as $$
+begin
+	return 5;
+end; $$;
+
+create function cdefault_restricted() returns int
+language plpgsql parallel restricted as $$
+begin
+	return 10;
+end; $$;
+
+create function trg_unsafe_fn() returns trigger
+language plpgsql parallel unsafe as $$
+begin
+	return new;
+end; $$;
+
+create function trg_restricted_fn() returns trigger
+language plpgsql parallel restricted as $$
+begin
+	return new;
+end; $$;
+
+--
+-- END: setup.
+--
+
+begin;
+
+-- encourage use of parallel plans
+set parallel_setup_cost = 0;
+set parallel_tuple_cost = 0;
+set min_parallel_table_scan_size = 0;
+set max_parallel_workers_per_gather = 4;
+
+-- force seq scans, for deterministic plans
+set enable_indexonlyscan = off;
+set enable_indexscan = off;
+set enable_bitmapscan = off;
+
+--
+-- A plain target table has no parallel-unsafe objects, so the SELECT part
+-- can be parallelized.
+--
+create table para_insert (a int, b name);
+
+explain (costs off) insert into para_insert select unique1, stringu1 from tenk1;
+insert into para_insert select unique1, stringu1 from tenk1;
+
+-- select some values to verify that the parallel insert worked
+select count(*), sum(a) from para_insert;
+-- verify that the same transaction and command has been used for all rows
+select count(*) from (select distinct cmin, xmin from para_insert) as dt;
+
+-- INSERT ... VALUES (no underlying SELECT) can't be parallelized
+explain (costs off) insert into para_insert values (1, 'one');
+
+-- a parallel-unsafe function in the SELECT part can't be parallelized
+explain (costs off)
+insert into para_insert select unique1, fullname_parallel_unsafe(stringu1, 'x') from tenk1;
+
+--
+-- A parallel-unsafe trigger on the target table prevents parallel SELECT.
+--
+create trigger trg before insert on para_insert
+	for each row execute function trg_unsafe_fn();
+explain (costs off) insert into para_insert select unique1, stringu1 from tenk1;
+-- dropping it makes the target parallel-safe again
+drop trigger trg on para_insert;
+explain (costs off) insert into para_insert select unique1, stringu1 from tenk1;
+
+--
+-- A parallel-restricted trigger on the target table still allows parallel
+-- SELECT; the trigger fires in the leader.
+--
+create trigger trg before insert on para_insert
+	for each row execute function trg_restricted_fn();
+explain (costs off) insert into para_insert select unique1, stringu1 from tenk1;
+drop trigger trg on para_insert;
+
+--
+-- A parallel-unsafe check constraint on the target table prevents
+-- parallel SELECT.
+--
+alter table para_insert add constraint chk_u check (fullname_parallel_unsafe(b, 'x') <> '');
+explain (costs off) insert into para_insert select unique1, stringu1 from tenk1;
+alter table para_insert drop constraint chk_u;
+explain (costs off) insert into para_insert select unique1, stringu1 from tenk1;
+
+--
+-- A parallel-unsafe index expression on the target table prevents
+-- parallel SELECT.
+--
+create index para_insert_unsafe_idx on para_insert (fullname_parallel_unsafe(b, 'x'));
+explain (costs off) insert into para_insert select unique1, stringu1 from tenk1;
+drop index para_insert_unsafe_idx;
+explain (costs off) insert into para_insert select unique1, stringu1 from tenk1;
+
+--
+-- Column defaults are checked: an unsafe default prevents parallel SELECT,
+-- while a restricted one is allowed (defaults are evaluated in the leader).
+--
+create table testdef (a int,
+					  b int default bdefault_unsafe(),
+					  c int default cdefault_restricted());
+explain (costs off) insert into testdef select unique1 from tenk1;
+alter table testdef alter column b drop default;
+explain (costs off) insert into testdef select unique1 from tenk1;
+
+--
+-- For an INSERT into a partitioned table, every partition must be
+-- parallel-safe.
+--
+create table part_para_insert (a int) partition by range (a);
+create table part_para_insert_p1 partition of part_para_insert
+	for values from (0) to (5000);
+create table part_para_insert_p2 partition of part_para_insert
+	for values from (5000) to (10000);
+
+explain (costs off) insert into part_para_insert select unique1 from tenk1;
+insert into part_para_insert select unique1 from tenk1;
+select count(*) from part_para_insert;
+
+-- inserting into a partition directly is fine too
+explain (costs off) insert into part_para_insert_p1 select unique1 from tenk1;
+
+-- a pre-existing unsafe trigger on one partition prevents parallel SELECT
+-- into the parent
+create table part_para_unsafe (a int) partition by range (a);
+create table part_para_unsafe_p1 partition of part_para_unsafe
+	for values from (0) to (10000);
+create trigger trg before insert on part_para_unsafe_p1
+	for each row execute function trg_unsafe_fn();
+explain (costs off) insert into part_para_unsafe select unique1 from tenk1;
+
+--
+-- Changing a function's parallel safety invalidates the cached safety of
+-- the target table.
+--
+create function alterable_fn(int) returns bool
+language plpgsql immutable parallel safe as $$
+begin
+	return $1 > 0;
+end; $$;
+
+create table fn_para_insert (a int check (alterable_fn(a)));
+explain (costs off) insert into fn_para_insert select unique1 from tenk1;
+alter function alterable_fn(int) parallel unsafe;
+explain (costs off) insert into fn_para_insert select unique1 from tenk1;
+alter function alterable_fn(int) parallel safe;
+explain (costs off) insert into fn_para_insert select unique1 from tenk1;
+
+--
+-- INSERT ... ON CONFLICT DO UPDATE can't be parallelized, because UPDATE is
+-- not supported in parallel mode; ON CONFLICT DO NOTHING can.
+--
+create table para_insert_pk (a int primary key, b name);
+explain (costs off)
+insert into para_insert_pk select unique1, stringu1 from tenk1
+	on conflict (a) do update set b = excluded.b;
+explain (costs off)
+insert into para_insert_pk select unique1, stringu1 from tenk1
+	on conflict (a) do nothing;
+
+--
+-- A temporary or foreign target table is parallel-restricted: only the
+-- leader can modify it, but the SELECT part can still be parallelized.
+--
+create temp table temp_para_insert (a int);
+explain (costs off) insert into temp_para_insert select unique1 from tenk1;
+
+--
+-- Forcing parallel mode when the plan has no Gather must work too; the
+-- transaction ID has to be assigned before entering parallel mode.
+--
+set debug_parallel_query = on;
+explain (costs off) insert into para_insert select i, 'x'::name from generate_series(1, 10) i;
+insert into para_insert select i, 'x'::name from generate_series(1, 10) i;
+select count(*) from para_insert;
+set debug_parallel_query = off;
+
+rollback;
-- 
2.43.0

