Do not place new shards with shards in TO_DELETE state (#3408)

When creating a new distributed table. The shards would colocate with shards
with SHARD_STATE_TO_DELETE (shardstate = 4). This means if that state was
because of a shard move the new shard would be created on two nodes and it
would not get deleted since it's shard state would be 1.

src/backend/distributed/master/master_create_shards.c

@@ -281,7 +281,6 @@ CreateColocatedShards(Oid targetRelationId, Oid sourceRelationId, bool
 		ShardInterval *sourceShardInterval = (ShardInterval *) lfirst(sourceShardCell);
 		uint64 sourceShardId = sourceShardInterval->shardId;
 		uint64 newShardId = GetNextShardId();
-		ListCell *sourceShardPlacementCell = NULL;
 
 		int32 shardMinValue = DatumGetInt32(sourceShardInterval->minValue);
 		int32 shardMaxValue = DatumGetInt32(sourceShardInterval->maxValue);
@@ -292,10 +291,14 @@ CreateColocatedShards(Oid targetRelationId, Oid sourceRelationId, bool
 		InsertShardRow(targetRelationId, newShardId, targetShardStorageType,
 					   shardMinValueText, shardMaxValueText);
 
-		foreach(sourceShardPlacementCell, sourceShardPlacementList)
+		ShardPlacement *sourcePlacement = NULL;
+		foreach_ptr(sourcePlacement, sourceShardPlacementList)
 		{
-			ShardPlacement *sourcePlacement =
+			if (sourcePlacement->shardState == SHARD_STATE_TO_DELETE)
-				(ShardPlacement *) lfirst(sourceShardPlacementCell);
+			{
+				continue;
+			}
+
 			int32 groupId = sourcePlacement->groupId;
 			const ShardState shardState = SHARD_STATE_ACTIVE;
 			const uint64 shardSize = 0;

src/test/regress/expected/multi_cluster_management.out

@@ -221,6 +221,7 @@ SELECT nodename, nodeport, noderole FROM pg_dist_node ORDER BY nodeport;
 
 ABORT;
 \c - postgres - :master_port
+SET citus.next_shard_id TO 1220016;
 SELECT master_get_active_worker_nodes();
  master_get_active_worker_nodes
 ---------------------------------------------------------------------
@@ -263,6 +264,68 @@ SELECT master_get_active_worker_nodes();
  (localhost,57637)
 (2 rows)
 
+-- mark all placements in the candidate node as to be deleted
+UPDATE pg_dist_placement SET shardstate=4 WHERE groupid=:worker_2_group;
+SELECT shardid, shardstate, nodename, nodeport FROM pg_dist_shard_placement WHERE nodeport=:worker_2_port;
+ shardid | shardstate | nodename  | nodeport
+---------------------------------------------------------------------
+ 1220001 |          4 | localhost |    57638
+ 1220003 |          4 | localhost |    57638
+ 1220005 |          4 | localhost |    57638
+ 1220007 |          4 | localhost |    57638
+ 1220009 |          4 | localhost |    57638
+ 1220011 |          4 | localhost |    57638
+ 1220013 |          4 | localhost |    57638
+ 1220015 |          4 | localhost |    57638
+(8 rows)
+
+CREATE TABLE cluster_management_test_colocated (col_1 text, col_2 int);
+SELECT create_distributed_table('cluster_management_test_colocated', 'col_1', 'hash', colocate_with=>'cluster_management_test');
+ create_distributed_table
+---------------------------------------------------------------------
+
+(1 row)
+
+-- Check that colocated shards don't get created for shards that are to be deleted
+SELECT logicalrelid, shardid, shardstate, nodename, nodeport FROM pg_dist_shard_placement NATURAL JOIN pg_dist_shard;
+           logicalrelid            | shardid | shardstate | nodename  | nodeport
+---------------------------------------------------------------------
+ cluster_management_test           | 1220000 |          1 | localhost |    57637
+ cluster_management_test           | 1220002 |          1 | localhost |    57637
+ cluster_management_test           | 1220004 |          1 | localhost |    57637
+ cluster_management_test           | 1220006 |          1 | localhost |    57637
+ cluster_management_test           | 1220008 |          1 | localhost |    57637
+ cluster_management_test           | 1220010 |          1 | localhost |    57637
+ cluster_management_test           | 1220012 |          1 | localhost |    57637
+ cluster_management_test           | 1220014 |          1 | localhost |    57637
+ cluster_management_test_colocated | 1220016 |          1 | localhost |    57637
+ cluster_management_test_colocated | 1220018 |          1 | localhost |    57637
+ cluster_management_test_colocated | 1220020 |          1 | localhost |    57637
+ cluster_management_test_colocated | 1220022 |          1 | localhost |    57637
+ cluster_management_test_colocated | 1220024 |          1 | localhost |    57637
+ cluster_management_test_colocated | 1220026 |          1 | localhost |    57637
+ cluster_management_test_colocated | 1220028 |          1 | localhost |    57637
+ cluster_management_test_colocated | 1220030 |          1 | localhost |    57637
+ cluster_management_test           | 1220001 |          4 | localhost |    57638
+ cluster_management_test           | 1220003 |          4 | localhost |    57638
+ cluster_management_test           | 1220005 |          4 | localhost |    57638
+ cluster_management_test           | 1220007 |          4 | localhost |    57638
+ cluster_management_test           | 1220009 |          4 | localhost |    57638
+ cluster_management_test           | 1220011 |          4 | localhost |    57638
+ cluster_management_test           | 1220013 |          4 | localhost |    57638
+ cluster_management_test           | 1220015 |          4 | localhost |    57638
+(24 rows)
+
+-- try to remove a node with only to be deleted placements and see that removal still fails
+SELECT master_remove_node('localhost', :worker_2_port);
+ERROR:  you cannot remove the primary node of a node group which has shard placements
+SELECT master_get_active_worker_nodes();
+ master_get_active_worker_nodes
+---------------------------------------------------------------------
+ (localhost,57638)
+ (localhost,57637)
+(2 rows)
+
 -- clean-up
 SELECT 1 FROM master_add_node('localhost', :worker_2_port);
  ?column?
@@ -271,6 +334,9 @@ SELECT 1 FROM master_add_node('localhost', :worker_2_port);
 (1 row)
 
 UPDATE pg_dist_placement SET shardstate=1 WHERE groupid=:worker_2_group;
+SET client_min_messages TO ERROR;
+DROP TABLE cluster_management_test_colocated;
+RESET client_min_messages;
 -- when there is no primary we should get a pretty error
 UPDATE pg_dist_node SET noderole = 'secondary' WHERE nodeport=:worker_2_port;
 SELECT * FROM cluster_management_test;

src/test/regress/sql/multi_cluster_management.sql

@@ -92,6 +92,7 @@ SELECT nodename, nodeport, noderole FROM pg_dist_node ORDER BY nodeport;
 ABORT;
 
 \c - postgres - :master_port
+SET citus.next_shard_id TO 1220016;
 SELECT master_get_active_worker_nodes();
 
 -- restore the node for next tests
@@ -109,9 +110,25 @@ SELECT shardid, shardstate, nodename, nodeport FROM pg_dist_shard_placement WHER
 SELECT master_remove_node('localhost', :worker_2_port);
 SELECT master_get_active_worker_nodes();
 
+-- mark all placements in the candidate node as to be deleted
+UPDATE pg_dist_placement SET shardstate=4 WHERE groupid=:worker_2_group;
+SELECT shardid, shardstate, nodename, nodeport FROM pg_dist_shard_placement WHERE nodeport=:worker_2_port;
+CREATE TABLE cluster_management_test_colocated (col_1 text, col_2 int);
+SELECT create_distributed_table('cluster_management_test_colocated', 'col_1', 'hash', colocate_with=>'cluster_management_test');
+
+-- Check that colocated shards don't get created for shards that are to be deleted
+SELECT logicalrelid, shardid, shardstate, nodename, nodeport FROM pg_dist_shard_placement NATURAL JOIN pg_dist_shard;
+
+-- try to remove a node with only to be deleted placements and see that removal still fails
+SELECT master_remove_node('localhost', :worker_2_port);
+SELECT master_get_active_worker_nodes();
+
 -- clean-up
 SELECT 1 FROM master_add_node('localhost', :worker_2_port);
 UPDATE pg_dist_placement SET shardstate=1 WHERE groupid=:worker_2_group;
+SET client_min_messages TO ERROR;
+DROP TABLE cluster_management_test_colocated;
+RESET client_min_messages;
 
 -- when there is no primary we should get a pretty error
 UPDATE pg_dist_node SET noderole = 'secondary' WHERE nodeport=:worker_2_port;