Avoid upgrading lock for INSERT statements

Commit 3fe43b8a introduced a lock upgrade in the plan revalidation
for UDFs. This makes the lock acquire in RevalidateCachedPlanWithParams()
match CdbTryOpenRelation() closer in order to avoid distributed deadlock
for UPDATE/DELETE DMLs. It does however also upgrade the lock for INSERT
which is overly aggressive. Fix by only upgrading the lock for the two
specified DML commands. Also includes an isolationtest test that cause
distributed deadlock without this patch.

This solves reported cases of deadlock introduced around INSERTs in UDFs.

src/backend/utils/cache/plancache.c

@@ -757,7 +757,9 @@ AcquireExecutorLocks(List *stmt_list, bool acquire)
 				 * replicated across cluster and don't suffer from the
 				 * deadlock.
 				 */
-				if (rte->relid > FirstNormalObjectId)
+				if (rte->relid > FirstNormalObjectId &&
+					(plannedstmt->commandType == CMD_UPDATE ||
+					 plannedstmt->commandType == CMD_DELETE))
 					lockmode = ExclusiveLock;
 				else
 					lockmode = RowExclusiveLock;
@@ -830,7 +832,9 @@ ScanQueryForLocks(Query *parsetree, bool acquire)
 					 * Catalog tables are replicated across cluster and don't
 					 * suffer from the deadlock.
 					 */
-					if (rte->relid > FirstNormalObjectId)
+					if (rte->relid > FirstNormalObjectId &&
+						(parsetree->commandType == CMD_UPDATE ||
+						 parsetree->commandType == CMD_DELETE))
 						lockmode = ExclusiveLock;
 					else
 						lockmode = RowExclusiveLock;

src/test/isolation/expected/udf-insert-deadlock.out

+Parsed test spec with 2 sessions
+
+starting permutation: s1begin s2begin s1insert_1_stat s2insert_2_stat s1insert_2_stat s2insert_1_stat s1commit s2commit
+step s1begin: BEGIN;
+step s2begin: BEGIN;
+step s1insert_1_stat: SELECT i_one(10, False);
+i_one          
+
+               
+step s2insert_2_stat: SELECT i_two(100, False);
+i_two          
+
+               
+step s1insert_2_stat: SELECT i_two(10, False);
+i_two          
+
+               
+step s2insert_1_stat: SELECT i_one(100, False);
+i_one          
+
+               
+step s1commit: COMMIT;
+step s2commit: COMMIT;
+
+starting permutation: s1begin s2begin s1insert_1_dyn s2insert_2_dyn s1insert_2_dyn s2insert_1_dyn s1commit s2commit
+step s1begin: BEGIN;
+step s2begin: BEGIN;
+step s1insert_1_dyn: SELECT i_one(15, True);
+i_one          
+
+               
+step s2insert_2_dyn: SELECT i_two(150, True);
+i_two          
+
+               
+step s1insert_2_dyn: SELECT i_two(15, True);
+i_two          
+
+               
+step s2insert_1_dyn: SELECT i_one(150, True);
+i_one          
+
+               
+step s1commit: COMMIT;
+step s2commit: COMMIT;
+
+starting permutation: s1begin s2begin s1insert_1_stat s2insert_2_dyn s1insert_2_stat s2insert_1_dyn s1commit s2commit
+step s1begin: BEGIN;
+step s2begin: BEGIN;
+step s1insert_1_stat: SELECT i_one(10, False);
+i_one          
+
+               
+step s2insert_2_dyn: SELECT i_two(150, True);
+i_two          
+
+               
+step s1insert_2_stat: SELECT i_two(10, False);
+i_two          
+
+               
+step s2insert_1_dyn: SELECT i_one(150, True);
+i_one          
+
+               
+step s1commit: COMMIT;
+step s2commit: COMMIT;
+
+starting permutation: s1begin s2begin s1insert_1_dyn s2insert_2_stat s1insert_2_dyn s2insert_1_stat s1commit s2commit
+step s1begin: BEGIN;
+step s2begin: BEGIN;
+step s1insert_1_dyn: SELECT i_one(15, True);
+i_one          
+
+               
+step s2insert_2_stat: SELECT i_two(100, False);
+i_two          
+
+               
+step s1insert_2_dyn: SELECT i_two(15, True);
+i_two          
+
+               
+step s2insert_1_stat: SELECT i_one(100, False);
+i_one          
+
+               
+step s1commit: COMMIT;
+step s2commit: COMMIT;

src/test/isolation/isolation_schedule

@@ -6,3 +6,4 @@ test: ao-serializable-read
 test: ao-serializable-vacuum
 test: ao-insert-eof
 test: create_index_hot
+test: udf-insert-deadlock

src/test/isolation/specs/udf-insert-deadlock.spec

+# Test validating that concurrent UDFs inserting data doesn't cause distributed
+# deadlocks due to ExclusiveLocks.
+#
+# The observed issue was that the cached plan revalidation was elevating the
+# lock for INSERT to ExclusiveLock. This is required for UPDATE/DELETE but is
+# too strict for INSERT where RowExclusiveLock is adequate.
+#
+# Runs two sessions which in turn insert into two tables in a criss-cross
+# pattern to try and induce a deadlock.
+
+setup
+{
+	CREATE TABLE udf_dl_one (a int, b int) DISTRIBUTED BY (a);
+	CREATE TABLE udf_dl_two (a int, b int) DISTRIBUTED BY (a);
+
+	CREATE FUNCTION i_one (val int, dyn bool) RETURNS void AS $$
+	BEGIN
+		IF (dyn) THEN
+			EXECUTE 'INSERT INTO udf_dl_one VALUES ($1, $2)' USING val,val;
+		ELSE
+			INSERT INTO udf_dl_one VALUES (val, val);
+		END IF;
+	END;
+	$$ LANGUAGE plpgsql;
+
+	CREATE FUNCTION i_two (val int, dyn bool) RETURNS void AS $$
+	BEGIN
+		IF (dyn) THEN
+			EXECUTE 'INSERT INTO udf_dl_two VALUES ($1, $2)' USING val,val;
+		ELSE
+			INSERT INTO udf_dl_two VALUES (val, val);
+		END IF;
+	END;
+	$$ LANGUAGE plpgsql;
+}
+
+teardown
+{
+	DROP TABLE udf_dl_one;
+	DROP TABLE udf_dl_two;
+
+	DROP FUNCTION i_one(int, bool);
+	DROP FUNCTION i_two(int, bool);
+}
+
+session "s1"
+step "s1begin"         { BEGIN; }
+step "s1insert_1_stat" { SELECT i_one(10, False); }
+step "s1insert_1_dyn"  { SELECT i_one(15, True); }
+step "s1insert_2_stat" { SELECT i_two(10, False); }
+step "s1insert_2_dyn"  { SELECT i_two(15, True); }
+step "s1commit"        { COMMIT; }
+
+session "s2"
+step "s2begin"         { BEGIN; }
+step "s2insert_2_stat" { SELECT i_two(100, False); }
+step "s2insert_2_dyn"  { SELECT i_two(150, True); }
+step "s2insert_1_stat" { SELECT i_one(100, False); }
+step "s2insert_1_dyn"  { SELECT i_one(150, True); }
+step "s2commit"        { COMMIT; }
+
+permutation "s1begin" "s2begin" "s1insert_1_stat" "s2insert_2_stat" "s1insert_2_stat" "s2insert_1_stat" "s1commit" "s2commit"
+permutation "s1begin" "s2begin" "s1insert_1_dyn" "s2insert_2_dyn" "s1insert_2_dyn" "s2insert_1_dyn" "s1commit" "s2commit"
+permutation "s1begin" "s2begin" "s1insert_1_stat" "s2insert_2_dyn" "s1insert_2_stat" "s2insert_1_dyn" "s1commit" "s2commit"
+permutation "s1begin" "s2begin" "s1insert_1_dyn" "s2insert_2_stat" "s1insert_2_dyn" "s2insert_1_stat" "s1commit" "s2commit"