Enable parallel SELECT for "INSERT INTO ... SELECT ...".

Parallel SELECT can't be utilized for INSERT in the following cases:
- INSERT statement uses the ON CONFLICT DO UPDATE clause
- Target table has a parallel-unsafe: trigger, index expression or
  predicate, column default expression or check constraint
- Target table has a parallel-unsafe domain constraint on any column
- Target table is a partitioned table with a parallel-unsafe partition key
  expression or support function

The planner is updated to perform additional parallel-safety checks for
the cases listed above, for determining whether it is safe to run INSERT
in parallel-mode with an underlying parallel SELECT. The planner will
consider using parallel SELECT for "INSERT INTO ... SELECT ...", provided
nothing unsafe is found from the additional parallel-safety checks, or
from the existing parallel-safety checks for SELECT.

While checking parallel-safety, we need to check it for all the partitions
on the table which can be costly especially when we decide not to use a
parallel plan. So, in a separate patch, we will introduce a GUC and or a
reloption to enable/disable parallelism for Insert statements.

Prior to entering parallel-mode for the execution of INSERT with parallel
SELECT, a TransactionId is acquired and assigned to the current
transaction state. This is necessary to prevent the INSERT from attempting
to assign the 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.

Author: Greg Nancarrow, Amit Langote, Amit Kapila
Reviewed-by: Amit Langote, Hou Zhijie, Takayuki Tsunakawa, Antonin Houska, Bharath Rupireddy, Dilip Kumar, Vignesh C, Zhihong Yu, Amit Kapila
Tested-by: Tang, Haiying
Discussion: https://postgr.es/m/CAJcOf-cXnB5cnMKqWEp2E2z7Mvcd04iLVmV=qpFJrR3AcrTS3g@mail.gmail.com
Discussion: https://postgr.es/m/CAJcOf-fAdj=nDKMsRhQzndm-O13NY4dL6xGcEvdX5Xvbbi0V7g@mail.gmail.com

1 files changed, 32 insertions, 1 deletions

diff --git a/src/backend/utils/cache/plancache.c b/src/backend/utils/cache/plancache.c
index 1a0950489d7..c1f4128445b 100644
--- a/src/backend/utils/cache/plancache.c
+++ b/src/backend/utils/cache/plancache.c
@@ -1735,6 +1735,23 @@ QueryListGetPrimaryStmt(List *stmts)
 	return NULL;
 }
 
+static void
+AcquireExecutorLocksOnPartitions(List *partitionOids, int lockmode,
+								 bool acquire)
+{
+	ListCell   *lc;
+
+	foreach(lc, partitionOids)
+	{
+		Oid			partOid = lfirst_oid(lc);
+
+		if (acquire)
+			LockRelationOid(partOid, lockmode);
+		else
+			UnlockRelationOid(partOid, lockmode);
+	}
+}
+
 /*
  * AcquireExecutorLocks: acquire locks needed for execution of a cached plan;
  * or release them if acquire is false.
@@ -1748,6 +1765,8 @@ AcquireExecutorLocks(List *stmt_list, bool acquire)
 	{
 		PlannedStmt *plannedstmt = lfirst_node(PlannedStmt, lc1);
 		ListCell   *lc2;
+		Index		rti,
+					resultRelation = 0;
 
 		if (plannedstmt->commandType == CMD_UTILITY)
 		{
@@ -1765,6 +1784,9 @@ AcquireExecutorLocks(List *stmt_list, bool acquire)
 			continue;
 		}
 
+		rti = 1;
+		if (plannedstmt->resultRelations)
+			resultRelation = linitial_int(plannedstmt->resultRelations);
 		foreach(lc2, plannedstmt->rtable)
 		{
 			RangeTblEntry *rte = (RangeTblEntry *) lfirst(lc2);
@@ -1782,6 +1804,14 @@ AcquireExecutorLocks(List *stmt_list, bool acquire)
 				LockRelationOid(rte->relid, rte->rellockmode);
 			else
 				UnlockRelationOid(rte->relid, rte->rellockmode);
+
+			/* Lock partitions ahead of modifying them in parallel mode. */
+			if (rti == resultRelation &&
+				plannedstmt->partitionOids != NIL)
+				AcquireExecutorLocksOnPartitions(plannedstmt->partitionOids,
+												 rte->rellockmode, acquire);
+
+			rti++;
 		}
 	}
 }
@@ -1990,7 +2020,8 @@ PlanCacheRelCallback(Datum arg, Oid relid)
 				if (plannedstmt->commandType == CMD_UTILITY)
 					continue;	/* Ignore utility statements */
 				if ((relid == InvalidOid) ? plannedstmt->relationOids != NIL :
-					list_member_oid(plannedstmt->relationOids, relid))
+					(list_member_oid(plannedstmt->relationOids, relid) ||
+					 list_member_oid(plannedstmt->partitionOids, relid)))
 				{
 					/* Invalidate the generic plan only */
 					plansource->gplan->is_valid = false;