ensure we have more active nodes than replication factor. (#6341)

DESCRIPTION: Fixes floating exception during
create_distributed_table_concurrently.

Fixes #6332.
During create_distributed_table_concurrently, when there is no active
primary node, it fails with floating exception. We added similar check
with create_distributed_table. It will fail with proper message if
current active node is less than replication factor.

src/backend/distributed/commands/create_distributed_table.c

@@ -528,6 +528,14 @@ CreateDistributedTableConcurrently(Oid relationId, char *distributionColumnName,
 		colocatedTableId = ColocatedTableId(colocationId);
 	}
 
+	List *workerNodeList = DistributedTablePlacementNodeList(NoLock);
+	if (workerNodeList == NIL)
+	{
+		ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+						errmsg("no worker nodes are available for placing shards"),
+						errhint("Add more worker nodes.")));
+	}
+
 	List *workersForPlacementList;
 	List *shardSplitPointsList;
 
@@ -555,7 +563,6 @@ CreateDistributedTableConcurrently(Oid relationId, char *distributionColumnName,
 		/*
 		 * Place shards in a round-robin fashion across all data nodes.
 		 */
-		List *workerNodeList = DistributedTablePlacementNodeList(NoLock);
 		workersForPlacementList = RoundRobinWorkerNodeList(workerNodeList, shardCount);
 	}
 
@@ -856,6 +863,8 @@ WorkerNodesForShardList(List *shardList)
 static List *
 RoundRobinWorkerNodeList(List *workerNodeList, int listLength)
 {
+	Assert(workerNodeList != NIL);
+
 	List *nodeIdList = NIL;
 
 	for (int idx = 0; idx < listLength; idx++)

src/test/regress/expected/create_distributed_table_concurrently.out

@@ -57,6 +57,35 @@ ERROR:  cannot colocate tables nocolo and test
 DETAIL:  Distribution column types don't match for nocolo and test.
 select create_distributed_table_concurrently('test','key', colocate_with := 'noexists');
 ERROR:  relation "noexists" does not exist
+select citus_set_node_property('localhost', :worker_1_port, 'shouldhaveshards', false);
+ citus_set_node_property
+---------------------------------------------------------------------
+
+(1 row)
+
+select citus_set_node_property('localhost', :worker_2_port, 'shouldhaveshards', false);
+ citus_set_node_property
+---------------------------------------------------------------------
+
+(1 row)
+
+select create_distributed_table_concurrently('test','key');
+NOTICE:  relation test does not have a REPLICA IDENTITY or PRIMARY KEY
+DETAIL:  UPDATE and DELETE commands on the relation will error out during create_distributed_table_concurrently unless there is a REPLICA IDENTITY or PRIMARY KEY. INSERT commands will still work.
+ERROR:  no worker nodes are available for placing shards
+HINT:  Add more worker nodes.
+select citus_set_node_property('localhost', :worker_1_port, 'shouldhaveshards', true);
+ citus_set_node_property
+---------------------------------------------------------------------
+
+(1 row)
+
+select citus_set_node_property('localhost', :worker_2_port, 'shouldhaveshards', true);
+ citus_set_node_property
+---------------------------------------------------------------------
+
+(1 row)
+
 -- use colocate_with "default"
 select create_distributed_table_concurrently('test','key', shard_count := 11);
 NOTICE:  relation test does not have a REPLICA IDENTITY or PRIMARY KEY

src/test/regress/sql/create_distributed_table_concurrently.sql

@@ -38,6 +38,12 @@ select create_distributed_table_concurrently('nocolo','x');
 select create_distributed_table_concurrently('test','key', colocate_with := 'nocolo');
 select create_distributed_table_concurrently('test','key', colocate_with := 'noexists');
 
+select citus_set_node_property('localhost', :worker_1_port, 'shouldhaveshards', false);
+select citus_set_node_property('localhost', :worker_2_port, 'shouldhaveshards', false);
+select create_distributed_table_concurrently('test','key');
+select citus_set_node_property('localhost', :worker_1_port, 'shouldhaveshards', true);
+select citus_set_node_property('localhost', :worker_2_port, 'shouldhaveshards', true);
+
 -- use colocate_with "default"
 select create_distributed_table_concurrently('test','key', shard_count := 11);