Add Isolation test, fix foreign constraint deadlock issue

src/backend/distributed/operations/shard_split.c

@@ -791,15 +791,18 @@ CreateForeignKeyConstraints(List *shardGroupSplitIntervalListList,
 														 &
 														 referenceTableForeignConstraintList);
 
-			List *commandList = NIL;
+			List *constraintCommandList = NIL;
-			commandList = list_concat(commandList, shardForeignConstraintCommandList);
+			constraintCommandList = list_concat(constraintCommandList, shardForeignConstraintCommandList);
-			commandList = list_concat(commandList, referenceTableForeignConstraintList);
+			constraintCommandList = list_concat(constraintCommandList, referenceTableForeignConstraintList);
 
-			SendCommandListToWorkerOutsideTransaction(
+			char *constraintCommand = NULL;
-				workerPlacementNode->workerName,
+			foreach_ptr(constraintCommand, constraintCommandList)
-				workerPlacementNode->workerPort,
+			{
-				TableOwner(shardInterval->relationId),
+				SendCommandToWorker(
-				commandList);
+					workerPlacementNode->workerName,
+					workerPlacementNode->workerPort,
+					constraintCommand);
+			}
 		}
 	}
 }

src/test/regress/enterprise_isolation_schedule

@@ -12,3 +12,4 @@ test: isolation_pg_send_cancellation
 test: isolation_shard_move_vs_start_metadata_sync
 test: isolation_tenant_isolation
 test: isolation_blocking_shard_split
+test: isolation_blocking_shard_split_with_fkey_to_reference

src/test/regress/expected/isolation_blocking_shard_split_with_fkey_to_reference.out

@@ -0,0 +1,301 @@
+Parsed test spec with 2 sessions
+
+starting permutation: s2-add-fkey s1-begin s2-begin s2-blocking-shard-split s1-delete s2-commit s1-commit s2-print-cluster
+create_distributed_table
+---------------------------------------------------------------------
+
+(1 row)
+
+step s2-add-fkey:
+ ALTER TABLE table_to_split ADD CONSTRAINT fkey_const FOREIGN KEY(value) REFERENCES reference_table(id);
+
+step s1-begin:
+    BEGIN;
+
+step s2-begin:
+ BEGIN;
+
+step s2-blocking-shard-split:
+ SELECT pg_catalog.citus_split_shard_by_split_points(
+  1500002,
+  ARRAY['-1073741824'],
+  ARRAY[1, 2],
+  'blocking');
+
+citus_split_shard_by_split_points
+---------------------------------------------------------------------
+
+(1 row)
+
+step s1-delete:
+ DELETE FROM reference_table WHERE id = 5;
+ <waiting ...>
+step s2-commit: 
+ COMMIT;
+
+step s1-delete: <... completed>
+step s1-commit:
+ COMMIT;
+
+step s2-print-cluster:
+ -- row count per shard
+ SELECT
+  nodeport, shardid, success, result
+ FROM
+  run_command_on_placements('table_to_split', 'select count(*) from %s')
+ ORDER BY
+  nodeport, shardid;
+ -- rows
+ SELECT id, value FROM table_to_split ORDER BY id, value;
+
+nodeport|shardid|success|result
+---------------------------------------------------------------------
+   57637|1500004|t      |     0
+   57638|1500003|t      |     0
+   57638|1500005|t      |     0
+(3 rows)
+
+id|value
+---------------------------------------------------------------------
+(0 rows)
+
+
+starting permutation: s2-add-fkey s1-begin s2-begin s2-blocking-shard-split s1-update s2-commit s1-commit s2-print-cluster
+create_distributed_table
+---------------------------------------------------------------------
+
+(1 row)
+
+step s2-add-fkey:
+ ALTER TABLE table_to_split ADD CONSTRAINT fkey_const FOREIGN KEY(value) REFERENCES reference_table(id);
+
+step s1-begin:
+    BEGIN;
+
+step s2-begin:
+ BEGIN;
+
+step s2-blocking-shard-split:
+ SELECT pg_catalog.citus_split_shard_by_split_points(
+  1500002,
+  ARRAY['-1073741824'],
+  ARRAY[1, 2],
+  'blocking');
+
+citus_split_shard_by_split_points
+---------------------------------------------------------------------
+
+(1 row)
+
+step s1-update:
+ UPDATE reference_table SET value = 5 WHERE id = 5;
+ <waiting ...>
+step s2-commit: 
+ COMMIT;
+
+step s1-update: <... completed>
+step s1-commit:
+ COMMIT;
+
+step s2-print-cluster:
+ -- row count per shard
+ SELECT
+  nodeport, shardid, success, result
+ FROM
+  run_command_on_placements('table_to_split', 'select count(*) from %s')
+ ORDER BY
+  nodeport, shardid;
+ -- rows
+ SELECT id, value FROM table_to_split ORDER BY id, value;
+
+nodeport|shardid|success|result
+---------------------------------------------------------------------
+   57637|1500004|t      |     0
+   57638|1500003|t      |     0
+   57638|1500005|t      |     0
+(3 rows)
+
+id|value
+---------------------------------------------------------------------
+(0 rows)
+
+
+starting permutation: s2-add-fkey s1-begin s2-begin s2-blocking-shard-split s1-insert s2-commit s1-commit s2-print-cluster
+create_distributed_table
+---------------------------------------------------------------------
+
+(1 row)
+
+step s2-add-fkey:
+ ALTER TABLE table_to_split ADD CONSTRAINT fkey_const FOREIGN KEY(value) REFERENCES reference_table(id);
+
+step s1-begin:
+    BEGIN;
+
+step s2-begin:
+ BEGIN;
+
+step s2-blocking-shard-split:
+ SELECT pg_catalog.citus_split_shard_by_split_points(
+  1500002,
+  ARRAY['-1073741824'],
+  ARRAY[1, 2],
+  'blocking');
+
+citus_split_shard_by_split_points
+---------------------------------------------------------------------
+
+(1 row)
+
+step s1-insert:
+ INSERT INTO reference_table VALUES (5, 10);
+ <waiting ...>
+step s2-commit: 
+ COMMIT;
+
+step s1-insert: <... completed>
+step s1-commit:
+ COMMIT;
+
+step s2-print-cluster:
+ -- row count per shard
+ SELECT
+  nodeport, shardid, success, result
+ FROM
+  run_command_on_placements('table_to_split', 'select count(*) from %s')
+ ORDER BY
+  nodeport, shardid;
+ -- rows
+ SELECT id, value FROM table_to_split ORDER BY id, value;
+
+nodeport|shardid|success|result
+---------------------------------------------------------------------
+   57637|1500004|t      |     0
+   57638|1500003|t      |     0
+   57638|1500005|t      |     0
+(3 rows)
+
+id|value
+---------------------------------------------------------------------
+(0 rows)
+
+
+starting permutation: s2-add-fkey s1-begin s2-begin s2-blocking-shard-split s1-copy s2-commit s1-commit s2-print-cluster
+create_distributed_table
+---------------------------------------------------------------------
+
+(1 row)
+
+step s2-add-fkey:
+ ALTER TABLE table_to_split ADD CONSTRAINT fkey_const FOREIGN KEY(value) REFERENCES reference_table(id);
+
+step s1-begin:
+    BEGIN;
+
+step s2-begin:
+ BEGIN;
+
+step s2-blocking-shard-split:
+ SELECT pg_catalog.citus_split_shard_by_split_points(
+  1500002,
+  ARRAY['-1073741824'],
+  ARRAY[1, 2],
+  'blocking');
+
+citus_split_shard_by_split_points
+---------------------------------------------------------------------
+
+(1 row)
+
+step s1-copy:
+ COPY reference_table FROM PROGRAM 'echo "1,1\n2,2\n3,3\n4,4\n5,5"' WITH CSV;
+ <waiting ...>
+step s2-commit: 
+ COMMIT;
+
+step s1-copy: <... completed>
+step s1-commit:
+ COMMIT;
+
+step s2-print-cluster:
+ -- row count per shard
+ SELECT
+  nodeport, shardid, success, result
+ FROM
+  run_command_on_placements('table_to_split', 'select count(*) from %s')
+ ORDER BY
+  nodeport, shardid;
+ -- rows
+ SELECT id, value FROM table_to_split ORDER BY id, value;
+
+nodeport|shardid|success|result
+---------------------------------------------------------------------
+   57637|1500004|t      |     0
+   57638|1500003|t      |     0
+   57638|1500005|t      |     0
+(3 rows)
+
+id|value
+---------------------------------------------------------------------
+(0 rows)
+
+
+starting permutation: s2-add-fkey s1-begin s2-begin s2-blocking-shard-split s1-ddl s2-commit s1-commit s2-print-cluster
+create_distributed_table
+---------------------------------------------------------------------
+
+(1 row)
+
+step s2-add-fkey:
+ ALTER TABLE table_to_split ADD CONSTRAINT fkey_const FOREIGN KEY(value) REFERENCES reference_table(id);
+
+step s1-begin:
+    BEGIN;
+
+step s2-begin:
+ BEGIN;
+
+step s2-blocking-shard-split:
+ SELECT pg_catalog.citus_split_shard_by_split_points(
+  1500002,
+  ARRAY['-1073741824'],
+  ARRAY[1, 2],
+  'blocking');
+
+citus_split_shard_by_split_points
+---------------------------------------------------------------------
+
+(1 row)
+
+step s1-ddl:
+ CREATE INDEX reference_table_index ON reference_table(id);
+ <waiting ...>
+step s2-commit: 
+ COMMIT;
+
+step s1-ddl: <... completed>
+step s1-commit:
+ COMMIT;
+
+step s2-print-cluster:
+ -- row count per shard
+ SELECT
+  nodeport, shardid, success, result
+ FROM
+  run_command_on_placements('table_to_split', 'select count(*) from %s')
+ ORDER BY
+  nodeport, shardid;
+ -- rows
+ SELECT id, value FROM table_to_split ORDER BY id, value;
+
+nodeport|shardid|success|result
+---------------------------------------------------------------------
+   57637|1500004|t      |     0
+   57638|1500003|t      |     0
+   57638|1500005|t      |     0
+(3 rows)
+
+id|value
+---------------------------------------------------------------------
+(0 rows)
+

src/test/regress/spec/isolation_blocking_shard_split.spec

@@ -2,7 +2,7 @@ setup
 {
 	SET citus.shard_count to 2;
 	SET citus.shard_replication_factor to 1;
-    select setval('pg_dist_shardid_seq', 1500000);
+    SELECT setval('pg_dist_shardid_seq', 1500000);
 
 	CREATE TABLE to_split_table (id int, value int);
 	SELECT create_distributed_table('to_split_table', 'id');

src/test/regress/spec/isolation_blocking_shard_split_with_fkey_to_reference.spec

@@ -0,0 +1,104 @@
+setup
+{
+    SELECT setval('pg_dist_shardid_seq', 1500000);
+    SET citus.shard_count to 2;
+	SET citus.shard_replication_factor to 1;
+
+	CREATE TABLE reference_table (id int PRIMARY KEY, value int);
+	SELECT create_reference_table('reference_table');
+
+	CREATE TABLE table_to_split (id int, value int);
+	SELECT create_distributed_table('table_to_split', 'id');
+}
+
+teardown
+{
+	DROP TABLE table_to_split CASCADE;
+	DROP TABLE reference_table CASCADE;
+}
+
+session "s1"
+
+step "s1-begin"
+{
+    BEGIN;
+}
+
+step "s1-insert"
+{
+	INSERT INTO reference_table VALUES (5, 10);
+}
+
+step "s1-update"
+{
+	UPDATE reference_table SET value = 5 WHERE id = 5;
+}
+
+step "s1-delete"
+{
+	DELETE FROM reference_table WHERE id = 5;
+}
+
+step "s1-ddl"
+{
+	CREATE INDEX reference_table_index ON reference_table(id);
+}
+
+step "s1-copy"
+{
+	COPY reference_table FROM PROGRAM 'echo "1,1\n2,2\n3,3\n4,4\n5,5"' WITH CSV;
+}
+
+step "s1-commit"
+{
+	COMMIT;
+}
+
+session "s2"
+
+step "s2-begin"
+{
+	BEGIN;
+}
+
+step "s2-blocking-shard-split"
+{
+	SELECT pg_catalog.citus_split_shard_by_split_points(
+		1500002,
+		ARRAY['-1073741824'],
+		ARRAY[1, 2],
+		'blocking');
+}
+
+step "s2-add-fkey"
+{
+	ALTER TABLE table_to_split ADD CONSTRAINT fkey_const FOREIGN KEY(value) REFERENCES reference_table(id);
+}
+
+step "s2-commit"
+{
+	COMMIT;
+}
+
+step "s2-print-cluster"
+{
+	-- row count per shard
+	SELECT
+		nodeport, shardid, success, result
+	FROM
+		run_command_on_placements('table_to_split', 'select count(*) from %s')
+	ORDER BY
+		nodeport, shardid;
+
+	-- rows
+	SELECT id, value FROM table_to_split ORDER BY id, value;
+}
+
+// Run shard split while concurrently performing an DML and index creation on the
+// reference table which the distributed table have a foreign key to.
+// All modifications should block on shard split.
+permutation "s2-add-fkey" "s1-begin" "s2-begin" "s2-blocking-shard-split" "s1-delete" "s2-commit" "s1-commit" "s2-print-cluster"
+permutation "s2-add-fkey" "s1-begin" "s2-begin" "s2-blocking-shard-split" "s1-update" "s2-commit" "s1-commit" "s2-print-cluster"
+permutation "s2-add-fkey" "s1-begin" "s2-begin" "s2-blocking-shard-split" "s1-insert" "s2-commit" "s1-commit" "s2-print-cluster"
+permutation "s2-add-fkey" "s1-begin" "s2-begin" "s2-blocking-shard-split" "s1-copy" "s2-commit" "s1-commit" "s2-print-cluster"
+permutation "s2-add-fkey" "s1-begin" "s2-begin" "s2-blocking-shard-split" "s1-ddl" "s2-commit" "s1-commit" "s2-print-cluster"