WIP: Allow router to touch multiple shards in a transaction.

This primarily for now is to be able to test distributed deadlock
detection.

Todo:
- Fix regression tests properly

src/backend/distributed/executor/multi_router_executor.c

@@ -78,8 +78,7 @@ static void AssignInsertTaskShardId(Query *jobQuery, List *taskList);
 static void ExecuteSingleModifyTask(CitusScanState *scanState, Task *task,
 									bool expectResults);
 static void ExecuteSingleSelectTask(CitusScanState *scanState, Task *task);
-static List * GetModifyConnections(List *taskPlacementList, bool markCritical,
+static List * GetModifyConnections(List *taskPlacementList, bool markCritical);
-								   bool startedInTransaction);
 static void ExecuteMultipleTasks(CitusScanState *scanState, List *taskList,
 								 bool isModificationQuery, bool expectResults);
 static int64 ExecuteModifyTasks(List *taskList, bool expectResults,
@@ -666,8 +665,6 @@ ExecuteSingleModifyTask(CitusScanState *scanState, Task *task, bool expectResult
 
 	char *queryString = task->queryString;
 	bool taskRequiresTwoPhaseCommit = (task->replicationModel == REPLICATION_MODEL_2PC);
-	bool startedInTransaction =
-		InCoordinatedTransaction() && XactModificationLevel == XACT_MODIFICATION_DATA;
 
 	if (XactModificationLevel == XACT_MODIFICATION_MULTI_SHARD)
 	{
@@ -706,8 +703,7 @@ ExecuteSingleModifyTask(CitusScanState *scanState, Task *task, bool expectResult
 	 * table) and start a transaction (when in a transaction).
 	 */
 	connectionList = GetModifyConnections(taskPlacementList,
-										  taskRequiresTwoPhaseCommit,
+										  taskRequiresTwoPhaseCommit);
-										  startedInTransaction);
 
 	/* prevent replicas of the same shard from diverging */
 	AcquireExecutorShardLock(task, operation);
@@ -808,7 +804,7 @@ ExecuteSingleModifyTask(CitusScanState *scanState, Task *task, bool expectResult
  * transaction in progress.
  */
 static List *
-GetModifyConnections(List *taskPlacementList, bool markCritical, bool noNewTransactions)
+GetModifyConnections(List *taskPlacementList, bool markCritical)
 {
 	ListCell *taskPlacementCell = NULL;
 	List *multiConnectionList = NIL;
@@ -828,26 +824,6 @@ GetModifyConnections(List *taskPlacementList, bool markCritical, bool noNewTrans
 		 */
 		multiConnection = StartPlacementConnection(connectionFlags, taskPlacement, NULL);
 
-		/*
-		 * If already in a transaction, disallow expanding set of remote
-		 * transactions. That prevents some forms of distributed deadlocks.
-		 */
-		if (noNewTransactions)
-		{
-			RemoteTransaction *transaction = &multiConnection->remoteTransaction;
-
-			if (EnableDeadlockPrevention &&
-				transaction->transactionState == REMOTE_TRANS_INVALID)
-			{
-				ereport(ERROR, (errcode(ERRCODE_CONNECTION_DOES_NOT_EXIST),
-								errmsg("no transaction participant matches %s:%d",
-									   taskPlacement->nodeName, taskPlacement->nodePort),
-								errdetail("Transactions which modify distributed tables "
-										  "may only target nodes affected by the "
-										  "modification command which began the transaction.")));
-			}
-		}
-
 		if (markCritical)
 		{
 			MarkRemoteTransactionCritical(multiConnection);

src/test/regress/expected/multi_modifying_xacts.out

@@ -136,14 +136,15 @@ SELECT * FROM researchers, labs WHERE labs.id = researchers.lab_id;
 BEGIN;
 INSERT INTO labs VALUES (6, 'Bell Labs');
 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');
+ERROR:  duplicate key value violates unique constraint "avoid_name_confusion_idx_1200001"
+DETAIL:  Key (lab_id, name)=(6, Leslie Lamport) already exists.
+CONTEXT:  while executing command on localhost:57638
 ABORT;
 -- SELECTs may occur after a modification: First check that selecting
 -- from the modified node works.
@@ -152,7 +153,7 @@ INSERT INTO labs VALUES (6, 'Bell Labs');
 SELECT count(*) FROM researchers WHERE lab_id = 6;
  count 
 -------
-     0
+     1
 (1 row)
 
 ABORT;
@@ -168,7 +169,7 @@ INSERT INTO labs VALUES (6, 'Bell Labs');
 SELECT count(*) FROM researchers WHERE lab_id = 6;
  count 
 -------
-     0
+     1
 (1 row)
 
 ABORT;
@@ -193,9 +194,10 @@ SELECT "Column", "Type", "Modifiers" FROM table_desc WHERE relid='public.labs'::
 (2 rows)
 
 SELECT * FROM labs WHERE id = 6;
- id | name 
+ id |   name    
-----+------
+----+-----------
-(0 rows)
+  6 | Bell Labs
+(1 row)
 
 -- COPY can happen after single row INSERT
 BEGIN;
@@ -256,9 +258,10 @@ DETAIL:  Key (lab_id, name)=(6,  'Bjarne Stroustrup') already exists.
 COMMIT;
 -- verify rollback
 SELECT * FROM researchers WHERE lab_id = 6;
- id | lab_id | name 
+ id | lab_id |      name      
-----+--------+------
+----+--------+----------------
-(0 rows)
+  9 |      6 | Leslie Lamport
+(1 row)
 
 SELECT count(*) FROM pg_dist_transaction;
  count 
@@ -283,9 +286,10 @@ DETAIL:  Key (lab_id, name)=(6,  'Bjarne Stroustrup') already exists.
 COMMIT;
 -- verify rollback
 SELECT * FROM researchers WHERE lab_id = 6;
- id | lab_id | name 
+ id | lab_id |      name      
-----+--------+------
+----+--------+----------------
-(0 rows)
+  9 |      6 | Leslie Lamport
+(1 row)
 
 SELECT count(*) FROM pg_dist_transaction;
  count 
@@ -301,9 +305,10 @@ COMMIT;
 SELECT * FROM researchers WHERE lab_id = 6;
  id | lab_id |         name         
 ----+--------+----------------------
+  9 |      6 | Leslie Lamport
  17 |      6 |  'Bjarne Stroustrup'
  18 |      6 |  'Dennis Ritchie'
-(2 rows)
+(3 rows)
 
 -- verify 2pc
 SELECT count(*) FROM pg_dist_transaction;
@@ -360,9 +365,10 @@ ERROR:  could not commit transaction on any active node
 SELECT * FROM researchers WHERE lab_id = 6;
  id | lab_id |         name         
 ----+--------+----------------------
+  9 |      6 | Leslie Lamport
  17 |      6 |  'Bjarne Stroustrup'
  18 |      6 |  'Dennis Ritchie'
-(2 rows)
+(3 rows)
 
 -- cleanup triggers and the function
 SELECT * from run_command_on_placements('researchers', 'drop trigger reject_large_researcher_id on %s')
@@ -400,7 +406,7 @@ ALTER TABLE labs ADD COLUMN motto text;
 SELECT master_modify_multiple_shards('DELETE FROM labs');
  master_modify_multiple_shards 
 -------------------------------
-                             7
+                             8
 (1 row)
 
 ALTER TABLE labs ADD COLUMN score float;
@@ -880,7 +886,6 @@ SELECT create_distributed_table('hash_modifying_xacts', 'key');
 BEGIN;
 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;

src/test/regress/expected/multi_mx_modifying_xacts.out

@@ -140,8 +140,6 @@ SELECT * FROM researchers_mx, labs_mx WHERE labs_mx.id = researchers_mx.lab_id;
 BEGIN;
 INSERT INTO labs_mx VALUES (6, 'Bell labs_mx');
 INSERT INTO researchers_mx 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;
 -- have the same test on the other worker node
 \c - - - :worker_2_port
@@ -163,8 +161,6 @@ SELECT * FROM researchers_mx, labs_mx WHERE labs_mx.id = researchers_mx.lab_id;
 BEGIN;
 INSERT INTO labs_mx VALUES (6, 'Bell labs_mx');
 INSERT INTO researchers_mx 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;
 -- switch back to the worker node
 \c - - - :worker_1_port
@@ -175,7 +171,7 @@ INSERT INTO labs_mx VALUES (6, 'Bell labs_mx');
 SELECT count(*) FROM researchers_mx WHERE lab_id = 6;
  count 
 -------
-     0
+     2
 (1 row)
 
 ABORT;