Add citus.cluster_name GUC

- Nodes with a nodecluster which does not match citus.cluster_name
  are excluded from the metadata cache and never seen by another part of
  Citus.

src/backend/distributed/shared_library_init.c

@@ -670,6 +670,16 @@ RegisterCitusConfigVariables(void)
 		0,
 		NULL, NULL, NULL);
 
+	DefineCustomStringVariable(
+		"citus.cluster_name",
+		gettext_noop("Which cluster this node is a part of"),
+		NULL,
+		&CurrentCluster,
+		"default",
+		PGC_SU_BACKEND,
+		0,
+		NULL, NULL, NULL);
+
 	DefineCustomBoolVariable(
 		"citus.enable_version_checks",
 		gettext_noop("Enables version checks during CREATE/ALTER EXTENSION commands"),

src/backend/distributed/utils/node_metadata.c

@@ -49,6 +49,8 @@
 /* default group size */
 int GroupSize = 1;
 
+/* config variable managed via guc.c */
+char *CurrentCluster = "default";
 
 /* local function forward declarations */
 static Datum ActivateNode(char *nodeName, int nodePort);
@@ -395,7 +397,7 @@ master_initialize_node_metadata(PG_FUNCTION_ARGS)
 
 	/* nodeRole and nodeCluster don't exist when this function is caled */
 	Oid nodeRole = InvalidOid;
-	char *nodeCluster = "default";
+	char *nodeCluster = WORKER_DEFAULT_CLUSTER;
 
 	CheckCitusVersion(ERROR);
 
@@ -556,7 +558,12 @@ ReadWorkerNodes()
 	while (HeapTupleIsValid(heapTuple))
 	{
 		WorkerNode *workerNode = TupleToWorkerNode(tupleDescriptor, heapTuple);
-		workerNodeList = lappend(workerNodeList, workerNode);
+
+		if (strncmp(workerNode->nodeCluster, CurrentCluster, WORKER_LENGTH) == 0)
+		{
+			/* the coordinator acts as if it never sees nodes not in it's cluster */
+			workerNodeList = lappend(workerNodeList, workerNode);
+		}
 
 		heapTuple = systable_getnext(scanDescriptor);
 	}

src/include/distributed/worker_manager.h

@@ -53,6 +53,7 @@ typedef struct WorkerNode
 /* Config variables managed via guc.c */
 extern int MaxWorkerNodesTracked;
 extern char *WorkerListFileName;
+extern char *CurrentCluster;
 
 
 /* Function declarations for finding worker nodes to place shards on */