Do not set coordinator's metadatasynced column to false

After a disable_node

(cherry picked from commit 5fc7661169)

src/backend/distributed/metadata/node_metadata.c

@@ -116,7 +116,7 @@ static WorkerNode * ModifiableWorkerNode(const char *nodeName, int32 nodePort);
 static bool NodeIsLocal(WorkerNode *worker);
 static void SetLockTimeoutLocally(int32 lock_cooldown);
 static void UpdateNodeLocation(int32 nodeId, char *newNodeName, int32 newNodePort);
-static bool UnsetMetadataSyncedForAll(void);
+static bool UnsetMetadataSyncedForAllWorkers(void);
 static char * GetMetadataSyncCommandToSetNodeColumn(WorkerNode *workerNode,
 													int columnIndex,
 													Datum value);
@@ -535,7 +535,7 @@ citus_disable_node(PG_FUNCTION_ARGS)
 	 * metadata at this point. Instead, we defer that to citus_activate_node()
 	 * where we expect all nodes up and running.
 	 */
-	if (UnsetMetadataSyncedForAll())
+	if (UnsetMetadataSyncedForAllWorkers())
 	{
 		TriggerMetadataSyncOnCommit();
 	}
@@ -1319,7 +1319,7 @@ citus_update_node(PG_FUNCTION_ARGS)
 	 * early, but that's fine, since this will start a retry loop with
 	 * 5 second intervals until sync is complete.
 	 */
-	if (UnsetMetadataSyncedForAll())
+	if (UnsetMetadataSyncedForAllWorkers())
 	{
 		TriggerMetadataSyncOnCommit();
 	}
@@ -2646,15 +2646,15 @@ DatumToString(Datum datum, Oid dataType)
 
 
 /*
- * UnsetMetadataSyncedForAll sets the metadatasynced column of all metadata
+ * UnsetMetadataSyncedForAllWorkers sets the metadatasynced column of all metadata
- * nodes to false. It returns true if it updated at least a node.
+ * worker nodes to false. It returns true if it updated at least a node.
  */
 static bool
-UnsetMetadataSyncedForAll(void)
+UnsetMetadataSyncedForAllWorkers(void)
 {
 	bool updatedAtLeastOne = false;
-	ScanKeyData scanKey[2];
+	ScanKeyData scanKey[3];
-	int scanKeyCount = 2;
+	int scanKeyCount = 3;
 	bool indexOK = false;
 
 	/*
@@ -2669,6 +2669,11 @@ UnsetMetadataSyncedForAll(void)
 	ScanKeyInit(&scanKey[1], Anum_pg_dist_node_metadatasynced,
 				BTEqualStrategyNumber, F_BOOLEQ, BoolGetDatum(true));
 
+	/* coordinator always has the up to date metadata */
+	ScanKeyInit(&scanKey[2], Anum_pg_dist_node_groupid,
+				BTGreaterStrategyNumber, F_INT4GT,
+				Int32GetDatum(COORDINATOR_GROUP_ID));
+
 	CatalogIndexState indstate = CatalogOpenIndexes(relation);
 
 	SysScanDesc scanDescriptor = systable_beginscan(relation,

src/test/regress/expected/multi_remove_node_reference_table.out

@@ -978,6 +978,12 @@ ORDER BY shardid ASC;
 (0 rows)
 
 \c - - - :master_port
+SELECT 1 FROM citus_set_coordinator_host('localhost', :master_port);
+ ?column?
+---------------------------------------------------------------------
+        1
+(1 row)
+
 SELECT citus_disable_node('localhost', :worker_2_port);
  citus_disable_node
 ---------------------------------------------------------------------
@@ -997,6 +1003,19 @@ SELECT COUNT(*) FROM pg_dist_node WHERE nodeport = :worker_2_port;
      1
 (1 row)
 
+-- never mark coordinator metadatasynced = false
+SELECT hasmetadata, metadatasynced FROM pg_dist_node WHERE nodeport = :master_port;
+ hasmetadata | metadatasynced
+---------------------------------------------------------------------
+ t           | t
+(1 row)
+
+SELECT 1 FROM citus_remove_node('localhost', :master_port);
+ ?column?
+---------------------------------------------------------------------
+        1
+(1 row)
+
 SELECT
     shardid, shardstate, shardlength, nodename, nodeport
 FROM

src/test/regress/sql/multi_remove_node_reference_table.sql

@@ -580,13 +580,19 @@ WHERE
 ORDER BY shardid ASC;
 
 \c - - - :master_port
-
+SELECT 1 FROM citus_set_coordinator_host('localhost', :master_port);
 SELECT citus_disable_node('localhost', :worker_2_port);
 SELECT public.wait_until_metadata_sync();
 
 -- status after citus_disable_node_and_wait
 SELECT COUNT(*) FROM pg_dist_node WHERE nodeport = :worker_2_port;
 
+-- never mark coordinator metadatasynced = false
+SELECT hasmetadata, metadatasynced FROM pg_dist_node WHERE nodeport = :master_port;
+
+SELECT 1 FROM citus_remove_node('localhost', :master_port);
+
+
 SELECT
     shardid, shardstate, shardlength, nodename, nodeport
 FROM