diff --git a/src/backend/utils/cache/plancache.c b/src/backend/utils/cache/plancache.c
index 02ff37009bae4f499ca4df2c8437d89789b5df0b..72d2883ebb68ccca0bb1c6a6d9aa600df1557f97 100644
--- a/src/backend/utils/cache/plancache.c
+++ b/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;
diff --git a/src/test/isolation/expected/udf-insert-deadlock.out b/src/test/isolation/expected/udf-insert-deadlock.out
new file mode 100644
index 0000000000000000000000000000000000000000..847977439539ca020fcb4d086ce6b49d2165a7e9
--- /dev/null
+++ b/src/test/isolation/expected/udf-insert-deadlock.out
@@ -0,0 +1,89 @@
+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;
diff --git a/src/test/isolation/isolation_schedule b/src/test/isolation/isolation_schedule
index 788d710ed1d876ab5de9ba9972dfc6d8aa0273ac..96bb6a2e91c68f3870d04f3f422baedba6b024d5 100644
--- a/src/test/isolation/isolation_schedule
+++ b/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
diff --git a/src/test/isolation/specs/udf-insert-deadlock.spec b/src/test/isolation/specs/udf-insert-deadlock.spec
new file mode 100644
index 0000000000000000000000000000000000000000..116b6e414940dcea34d9d7504150c0f8332579b7
--- /dev/null
+++ b/src/test/isolation/specs/udf-insert-deadlock.spec
@@ -0,0 +1,65 @@
+# 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"