external
/
citus
mirror of https://github.com/citusdata/citus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Do not send metadata changes during add node if citus.enable_metadata_sync is set to false

pull/5979/head
Onder Kalaci 2022-05-30 12:25:19 +02:00
parent 010a2a408e
commit 7157152f6c
1 changed files with 25 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
src/backend/distributed/metadata/node_metadata.c
View File

@ -1899,12 +1899,15 @@ RemoveNodeFromCluster(char *nodeName, int32 nodePort)
RemoveOldShardPlacementForNodeGroup(workerNode->groupId); RemoveOldShardPlacementForNodeGroup(workerNode->groupId);
char *nodeDeleteCommand = NodeDeleteCommand(workerNode->nodeId);
/* make sure we don't have any lingering session lifespan connections */ /* make sure we don't have any lingering session lifespan connections */
CloseNodeConnectionsAfterTransaction(workerNode->workerName, nodePort); CloseNodeConnectionsAfterTransaction(workerNode->workerName, nodePort);
SendCommandToWorkersWithMetadata(nodeDeleteCommand); if (EnableMetadataSync)
{
char *nodeDeleteCommand = NodeDeleteCommand(workerNode->nodeId);
SendCommandToWorkersWithMetadata(nodeDeleteCommand);
}
} }
@ -2155,18 +2158,21 @@ AddNodeMetadata(char *nodeName, int32 nodePort,
workerNode = FindWorkerNodeAnyCluster(nodeName, nodePort); workerNode = FindWorkerNodeAnyCluster(nodeName, nodePort);
/* send the delete command to all primary nodes with metadata */ if (EnableMetadataSync)
char *nodeDeleteCommand = NodeDeleteCommand(workerNode->nodeId);
SendCommandToWorkersWithMetadata(nodeDeleteCommand);
/* finally prepare the insert command and send it to all primary nodes */
uint32 primariesWithMetadata = CountPrimariesWithMetadata();
if (primariesWithMetadata != 0)
{ {
List *workerNodeList = list_make1(workerNode); /* send the delete command to all primary nodes with metadata */
char *nodeInsertCommand = NodeListInsertCommand(workerNodeList); char *nodeDeleteCommand = NodeDeleteCommand(workerNode->nodeId);
SendCommandToWorkersWithMetadata(nodeDeleteCommand);
SendCommandToWorkersWithMetadata(nodeInsertCommand); /* finally prepare the insert command and send it to all primary nodes */
uint32 primariesWithMetadata = CountPrimariesWithMetadata();
if (primariesWithMetadata != 0)
{
List *workerNodeList = list_make1(workerNode);
char *nodeInsertCommand = NodeListInsertCommand(workerNodeList);
SendCommandToWorkersWithMetadata(nodeInsertCommand);
}
} }
return workerNode->nodeId; return workerNode->nodeId;
@ -2185,11 +2191,13 @@ SetWorkerColumn(WorkerNode *workerNode, int columnIndex, Datum value)
{ {
workerNode = SetWorkerColumnLocalOnly(workerNode, columnIndex, value); workerNode = SetWorkerColumnLocalOnly(workerNode, columnIndex, value);
char *metadataSyncCommand = GetMetadataSyncCommandToSetNodeColumn(workerNode, if (EnableMetadataSync)
columnIndex, {
value); char *metadataSyncCommand =
GetMetadataSyncCommandToSetNodeColumn(workerNode, columnIndex, value);
SendCommandToWorkersWithMetadata(metadataSyncCommand); SendCommandToWorkersWithMetadata(metadataSyncCommand);
}
return workerNode; return workerNode;
} }