Add an enable_deadlock_prevention flag to allow router transactions to expand to multiple nodes

src/backend/distributed/executor/multi_router_executor.c

@@ -70,6 +70,7 @@
 
 /* controls use of locks to enforce safe commutativity */
 bool AllModificationsCommutative = false;
+bool EnableDeadlockPrevention = true;
 
 /* functions needed during run phase */
 static void ReacquireMetadataLocks(List *taskList);
@@ -838,7 +839,8 @@ GetModifyConnections(List *taskPlacementList, bool markCritical, bool noNewTrans
 		{
 			RemoteTransaction *transaction = &multiConnection->remoteTransaction;
 
-			if (transaction->transactionState == REMOTE_TRANS_INVALID)
+			if (EnableDeadlockPrevention &&
+				transaction->transactionState == REMOTE_TRANS_INVALID)
 			{
 				ereport(ERROR, (errcode(ERRCODE_CONNECTION_DOES_NOT_EXIST),
 								errmsg("no transaction participant matches %s:%d",

src/backend/distributed/shared_library_init.c

@@ -329,6 +329,19 @@ RegisterCitusConfigVariables(void)
 		0,
 		NULL, NULL, NULL);
 
+	DefineCustomBoolVariable(
+		"citus.enable_deadlock_prevention",
+		gettext_noop("Prevents transactions from expanding to multiple nodes"),
+		gettext_noop("When enabled, consecutive DML statements that write to "
+					 "shards on different nodes are prevented to avoid creating "
+					 "undetectable distributed deadlocks when performed "
+					 "concurrently."),
+		&EnableDeadlockPrevention,
+		true,
+		PGC_USERSET,
+		GUC_NO_SHOW_ALL,
+		NULL, NULL, NULL);
+
 	DefineCustomBoolVariable(
 		"citus.enable_ddl_propagation",
 		gettext_noop("Enables propagating DDL statements to worker shards"),

src/include/distributed/multi_router_executor.h

@@ -31,6 +31,7 @@ typedef struct XactShardConnSet
 
 /* Config variables managed via guc.c */
 extern bool AllModificationsCommutative;
+extern bool EnableDeadlockPrevention;
 
 
 extern void RouterExecutorStart(QueryDesc *queryDesc, int eflags, List *taskList);

src/test/regress/expected/multi_modifying_xacts.out

@@ -140,6 +140,12 @@ INSERT INTO researchers VALUES (9, 6, 'Leslie Lamport');
 ERROR:  no transaction participant matches localhost:57638
 DETAIL:  Transactions which modify distributed tables may only target nodes affected by the modification command which began the transaction.
 COMMIT;
+-- unless we disable deadlock prevention
+BEGIN;
+SET citus.enable_deadlock_prevention TO off;
+INSERT INTO labs VALUES (6, 'Bell Labs');
+INSERT INTO researchers VALUES (9, 6, 'Leslie Lamport');
+ABORT;
 -- SELECTs may occur after a modification: First check that selecting
 -- from the modified node works.
 BEGIN;
@@ -888,6 +894,17 @@ INSERT INTO hash_modifying_xacts VALUES (1, 1);
 INSERT INTO reference_modifying_xacts VALUES (10, 10);
 ERROR:  no transaction participant matches localhost:57638
 COMMIT;
+-- it is allowed when turning off deadlock prevention
+BEGIN;
+SET citus.enable_deadlock_prevention TO off;
+INSERT INTO hash_modifying_xacts VALUES (1, 1);
+INSERT INTO reference_modifying_xacts VALUES (10, 10);
+ABORT;
+BEGIN;
+SET citus.enable_deadlock_prevention TO off;
+INSERT INTO hash_modifying_xacts VALUES (1, 1);
+INSERT INTO hash_modifying_xacts VALUES (2, 2);
+ABORT;
 -- lets fail one of the workers before COMMIT time for the hash table
 \c - - - :worker_1_port
 CREATE FUNCTION reject_bad_hash() RETURNS trigger AS $rb$

src/test/regress/sql/multi_modifying_xacts.sql

@@ -111,6 +111,13 @@ INSERT INTO labs VALUES (6, 'Bell Labs');
 INSERT INTO researchers VALUES (9, 6, 'Leslie Lamport');
 COMMIT;
 
+-- unless we disable deadlock prevention
+BEGIN;
+SET citus.enable_deadlock_prevention TO off;
+INSERT INTO labs VALUES (6, 'Bell Labs');
+INSERT INTO researchers VALUES (9, 6, 'Leslie Lamport');
+ABORT;
+
 -- SELECTs may occur after a modification: First check that selecting
 -- from the modified node works.
 BEGIN;
@@ -658,6 +665,19 @@ INSERT INTO hash_modifying_xacts VALUES (1, 1);
 INSERT INTO reference_modifying_xacts VALUES (10, 10);
 COMMIT;
 
+-- it is allowed when turning off deadlock prevention
+BEGIN;
+SET citus.enable_deadlock_prevention TO off;
+INSERT INTO hash_modifying_xacts VALUES (1, 1);
+INSERT INTO reference_modifying_xacts VALUES (10, 10);
+ABORT;
+
+BEGIN;
+SET citus.enable_deadlock_prevention TO off;
+INSERT INTO hash_modifying_xacts VALUES (1, 1);
+INSERT INTO hash_modifying_xacts VALUES (2, 2);
+ABORT;
+
 -- lets fail one of the workers before COMMIT time for the hash table
 \c - - - :worker_1_port