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

wip

metadata_sync_imp
Onder Kalaci 2022-11-28 19:25:00 +01:00
parent 93a446254c
commit 23775a861b
8 changed files with 232 additions and 213 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
src/backend/distributed/commands/dependencies.c
View File

@ -532,15 +532,8 @@ GetAllDependencyCreateDDLCommands(const List *dependencies)
* clusterHasDistributedFunction if there are any distributed functions. * clusterHasDistributedFunction if there are any distributed functions.
*/ */
void void
ReplicateAllObjectsToNodeCommandList(List *nodeToSyncMetadataConnections, ReplicateAllObjectsToNodeCommandList(MetadataSyncContext syncContext)
List **ddlCommands)
{ {
/* since we are executing ddl commands disable propagation first, primarily for mx */
if (ddlCommands != NULL)
{
*ddlCommands = list_make1(DISABLE_DDL_PROPAGATION);
}
/* /*
* collect all dependencies in creation order and get their ddl commands * collect all dependencies in creation order and get their ddl commands
*/ */
@ -582,30 +575,23 @@ ReplicateAllObjectsToNodeCommandList(List *nodeToSyncMetadataConnections,
continue; continue;
} }
List *perObjCommands = GetDependencyCreateDDLCommands(dependency); List *perObjCommands = GetDependencyCreateDDLCommands(dependency);
if (list_length(nodeToSyncMetadataConnections) != 0) if (syncContext.syncImmediately)
{ {
SendCommandListToWorkerOutsideTransactionWithConnection(linitial( SendCommandListToWorkerOutsideTransactionWithConnection(linitial(
nodeToSyncMetadataConnections), syncContext.
nodeConnectionList),
perObjCommands); perObjCommands);
} }
if (ddlCommands != NULL)
{
*ddlCommands = list_concat(*ddlCommands,
perObjCommands);
}
else else
{ {
list_free_deep(perObjCommands); syncContext.ddlCommandList =
list_concat(syncContext.ddlCommandList, perObjCommands);
} }
} }
if (ddlCommands != NULL)
{
*ddlCommands = lappend(*ddlCommands, ENABLE_DDL_PROPAGATION);
}
} }

80
src/backend/distributed/metadata/metadata_sync.c
View File

@ -226,10 +226,8 @@ start_metadata_sync_to_all_nodes(PG_FUNCTION_ARGS)
* start_metadata_sync_to_node(). * start_metadata_sync_to_node().
*/ */
void void
SyncNodeMetadataToNode(const char *nodeNameString, int32 nodePort) SyncNodeMetadataToNode(MetadataSyncContext syncContext)
{ {
char *escapedNodeName = quote_literal_cstr(nodeNameString);
CheckCitusVersion(ERROR); CheckCitusVersion(ERROR);
EnsureCoordinator(); EnsureCoordinator();
EnsureModificationsCanRun(); EnsureModificationsCanRun();
@ -663,44 +661,44 @@ DropMetadataSnapshotOnNode(WorkerNode *workerNode)
EnsureSequentialModeMetadataOperations(); EnsureSequentialModeMetadataOperations();
char *userName = CurrentUserName(); char *userName = CurrentUserName();
List *dropMetadataCommandList = NIL;
/* /*
* Detach partitions, break dependencies between sequences and table then * Detach partitions, break dependencies between sequences and table then
* remove shell tables first. * remove shell tables first.
*/ */
List *detachPartitionCommandList = NIL; MetadataSyncContext syncContext;
syncContext.syncImmediately = false;
DetachPartitionCommandList(NIL, &detachPartitionCommandList); DetachPartitionCommandList(syncContext);
dropMetadataCommandList = list_concat(dropMetadataCommandList,
detachPartitionCommandList);
dropMetadataCommandList = lappend(dropMetadataCommandList, syncContext.ddlCommandList = lappend(syncContext.ddlCommandList,
BREAK_CITUS_TABLE_SEQUENCE_DEPENDENCY_COMMAND); BREAK_CITUS_TABLE_SEQUENCE_DEPENDENCY_COMMAND);
dropMetadataCommandList = lappend(dropMetadataCommandList, syncContext.ddlCommandList = lappend(syncContext.ddlCommandList,
REMOVE_ALL_SHELL_TABLES_COMMAND); REMOVE_ALL_SHELL_TABLES_COMMAND);
dropMetadataCommandList = list_concat(dropMetadataCommandList, syncContext.ddlCommandList = list_concat(syncContext.ddlCommandList,
NodeMetadataDropCommands()); NodeMetadataDropCommands());
dropMetadataCommandList = lappend(dropMetadataCommandList, syncContext.ddlCommandList = lappend(syncContext.ddlCommandList,
LocalGroupIdUpdateCommand(0)); LocalGroupIdUpdateCommand(0));
/* remove all dist table and object/table related metadata afterwards */ /* remove all dist table and object/table related metadata afterwards */
dropMetadataCommandList = lappend(dropMetadataCommandList, DELETE_ALL_PARTITIONS); syncContext.ddlCommandList = lappend(syncContext.ddlCommandList,
dropMetadataCommandList = lappend(dropMetadataCommandList, DELETE_ALL_SHARDS); DELETE_ALL_PARTITIONS);
dropMetadataCommandList = lappend(dropMetadataCommandList, DELETE_ALL_PLACEMENTS); syncContext.ddlCommandList = lappend(syncContext.ddlCommandList, DELETE_ALL_SHARDS);
dropMetadataCommandList = lappend(dropMetadataCommandList, syncContext.ddlCommandList = lappend(syncContext.ddlCommandList,
DELETE_ALL_DISTRIBUTED_OBJECTS); DELETE_ALL_PLACEMENTS);
dropMetadataCommandList = lappend(dropMetadataCommandList, DELETE_ALL_COLOCATION); syncContext.ddlCommandList = lappend(syncContext.ddlCommandList,
DELETE_ALL_DISTRIBUTED_OBJECTS);
syncContext.ddlCommandList = lappend(syncContext.ddlCommandList,
DELETE_ALL_COLOCATION);
Assert(superuser()); Assert(superuser());
SendOptionalMetadataCommandListToWorkerInCoordinatedTransaction( SendOptionalMetadataCommandListToWorkerInCoordinatedTransaction(
workerNode->workerName, workerNode->workerName,
workerNode->workerPort, workerNode->workerPort,
userName, userName,
dropMetadataCommandList); syncContext.ddlCommandList);
} }
@ -2667,8 +2665,7 @@ CreateTableMetadataOnWorkers(Oid relationId)
* empty list to not disable/enable DDL propagation for nothing. * empty list to not disable/enable DDL propagation for nothing.
*/ */
void void
DetachPartitionCommandList(List *nodeToSyncMetadataConnections, DetachPartitionCommandList(MetadataSyncContext syncContext)
List **detachPartitionCommandList)
{ {
List *citusTableIdList = CitusTableTypeIdList(ANY_CITUS_TABLE_TYPE); List *citusTableIdList = CitusTableTypeIdList(ANY_CITUS_TABLE_TYPE);
@ -2693,42 +2690,31 @@ DetachPartitionCommandList(List *nodeToSyncMetadataConnections,
Assert(PartitionTable(partitionRelOid)); Assert(PartitionTable(partitionRelOid));
char *detachCommand = GenerateDetachPartitionCommand(partitionRelOid); char *detachCommand = GenerateDetachPartitionCommand(partitionRelOid);
if (list_length(nodeToSyncMetadataConnections) != 0) if (syncContext.syncImmediately)
{ {
SendCommandListToWorkerOutsideTransactionWithConnection(linitial( SyncMetadataCommands(syncContext, list_make1(detachCommand), false);
nodeToSyncMetadataConnections),
list_make1(
detachCommand));
}
if (detachPartitionCommandList != NULL) pfree(detachCommand);
{
*detachPartitionCommandList =
lappend(*detachPartitionCommandList, detachCommand);
} }
else else
{ {
pfree(detachCommand); syncContext.ddlCommandList =
lappend(syncContext.ddlCommandList, detachCommand);
} }
} }
} }
if (!foundAnyPartitionedTable) if (foundAnyPartitionedTable && !syncContext.syncImmediately)
{ {
return; syncContext.ddlCommandList =
} lcons(DISABLE_DDL_PROPAGATION, syncContext.ddlCommandList);
if (detachPartitionCommandList != NULL)
{
*detachPartitionCommandList =
lcons(DISABLE_DDL_PROPAGATION, *detachPartitionCommandList);
/* /*
* We probably do not need this but as an extra precaution, we are enabling * We probably do not need this but as an extra precaution, we are enabling
* DDL propagation to switch back to original state. * DDL propagation to switch back to original state.
*/ */
*detachPartitionCommandList = lappend(*detachPartitionCommandList, syncContext.ddlCommandList = lappend(syncContext.ddlCommandList,
ENABLE_DDL_PROPAGATION); ENABLE_DDL_PROPAGATION);
} }
} }

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

@ -33,6 +33,7 @@
#include "distributed/metadata/distobject.h" #include "distributed/metadata/distobject.h"
#include "distributed/metadata_cache.h" #include "distributed/metadata_cache.h"
#include "distributed/metadata_sync.h" #include "distributed/metadata_sync.h"
#include "distributed/metadata_sync_context.h"
#include "distributed/multi_join_order.h" #include "distributed/multi_join_order.h"
#include "distributed/multi_router_planner.h" #include "distributed/multi_router_planner.h"
#include "distributed/pg_dist_node.h" #include "distributed/pg_dist_node.h"
@ -100,15 +101,13 @@ static void InsertPlaceholderCoordinatorRecord(void);
static void InsertNodeRow(int nodeid, char *nodename, int32 nodeport, NodeMetadata static void InsertNodeRow(int nodeid, char *nodename, int32 nodeport, NodeMetadata
*nodeMetadata); *nodeMetadata);
static void DeleteNodeRow(char *nodename, int32 nodeport); static void DeleteNodeRow(char *nodename, int32 nodeport);
static void SyncDistributedObjectsToNodeList(List *workerNodeList); static void SyncDistributedObjectsToNodeList(MetadataSyncContext syncContext);
static void UpdateLocalGroupIdOnNode(WorkerNode *workerNode); static void UpdateLocalGroupIdOnNode(MultiConnection *connection, WorkerNode *workerNode);
static void SyncPgDistTableMetadataToNodeList(List *nodeToSyncMetadataConnections); static void SyncPgDistTableMetadataToNodeList(MetadataSyncContext syncContext);
static void InterTableRelationshipCommandList(List *nodeToSyncMetadataConnections, static void InterTableRelationshipCommandList(MetadataSyncContext syncContext);
List **ddlCommandList);
static void BlockDistributedQueriesOnMetadataNodes(void); static void BlockDistributedQueriesOnMetadataNodes(void);
static WorkerNode * TupleToWorkerNode(TupleDesc tupleDescriptor, HeapTuple heapTuple); static WorkerNode * TupleToWorkerNode(TupleDesc tupleDescriptor, HeapTuple heapTuple);
static void PropagateNodeWideObjectsCommandList(List *nodeToSyncMetadataConnections, static void PropagateNodeWideObjectsCommandList(MetadataSyncContext syncContext);
List **nodeWideObjectCommandList);
static WorkerNode * ModifiableWorkerNode(const char *nodeName, int32 nodePort); static WorkerNode * ModifiableWorkerNode(const char *nodeName, int32 nodePort);
static bool NodeIsLocal(WorkerNode *worker); static bool NodeIsLocal(WorkerNode *worker);
static void SetLockTimeoutLocally(int32 lock_cooldown); static void SetLockTimeoutLocally(int32 lock_cooldown);
@ -655,8 +654,7 @@ master_set_node_property(PG_FUNCTION_ARGS)
* for each citus table. * for each citus table.
*/ */
static void static void
InterTableRelationshipCommandList(List *nodeToSyncMetadataConnections, InterTableRelationshipCommandList(MetadataSyncContext syncContext)
List **multipleTableIntegrationCommandList)
{ {
List *citusTableIdList = CitusTableTypeIdList(ANY_CITUS_TABLE_TYPE); List *citusTableIdList = CitusTableTypeIdList(ANY_CITUS_TABLE_TYPE);
List *propagatedTableList = NIL; List *propagatedTableList = NIL;
@ -680,32 +678,27 @@ InterTableRelationshipCommandList(List *nodeToSyncMetadataConnections,
List *commandListForRelation = List *commandListForRelation =
InterTableRelationshipOfRelationCommandList(relationId); InterTableRelationshipOfRelationCommandList(relationId);
if (syncContext.syncImmediately)
if (list_length(nodeToSyncMetadataConnections) != 0)
{ {
SendCommandListToWorkerOutsideTransactionWithConnection(linitial( SendCommandListToWorkerOutsideTransactionWithConnection(linitial(
nodeToSyncMetadataConnections), syncContext.
nodeConnectionList),
commandListForRelation); commandListForRelation);
} }
if (multipleTableIntegrationCommandList != NULL)
{
*multipleTableIntegrationCommandList = list_concat(
*multipleTableIntegrationCommandList,
commandListForRelation);
}
else else
{ {
list_free_deep(commandListForRelation); syncContext.ddlCommandList = list_concat(
syncContext.ddlCommandList,
commandListForRelation);
} }
} }
if (multipleTableIntegrationCommandList != NULL) if (!syncContext.syncImmediately)
{ {
*multipleTableIntegrationCommandList = lcons(DISABLE_DDL_PROPAGATION, syncContext.ddlCommandList = lcons(DISABLE_DDL_PROPAGATION,
*multipleTableIntegrationCommandList); syncContext.ddlCommandList);
*multipleTableIntegrationCommandList = lappend( syncContext.ddlCommandList = lappend(
*multipleTableIntegrationCommandList, syncContext.ddlCommandList,
ENABLE_DDL_PROPAGATION); ENABLE_DDL_PROPAGATION);
} }
} }
@ -716,8 +709,7 @@ InterTableRelationshipCommandList(List *nodeToSyncMetadataConnections,
* (except pg_dist_node) metadata. We call them as table metadata. * (except pg_dist_node) metadata. We call them as table metadata.
*/ */
void void
PgDistTableMetadataSyncCommandList(List *nodeToSyncMetadataConnections, PgDistTableMetadataSyncCommandList(MetadataSyncContext syncContext)
List **metadataSnapshotCommandList)
{ {
List *distributedTableList = CitusTableList(); List *distributedTableList = CitusTableList();
List *propagatedTableList = NIL; List *propagatedTableList = NIL;
@ -732,24 +724,27 @@ PgDistTableMetadataSyncCommandList(List *nodeToSyncMetadataConnections,
} }
} }
/* remove all dist table and object related metadata first */
*metadataSnapshotCommandList = lappend(*metadataSnapshotCommandList,
DELETE_ALL_PARTITIONS);
*metadataSnapshotCommandList = lappend(*metadataSnapshotCommandList,
DELETE_ALL_SHARDS);
*metadataSnapshotCommandList = lappend(*metadataSnapshotCommandList,
DELETE_ALL_PLACEMENTS);
*metadataSnapshotCommandList = lappend(*metadataSnapshotCommandList,
DELETE_ALL_DISTRIBUTED_OBJECTS);
*metadataSnapshotCommandList = lappend(*metadataSnapshotCommandList,
DELETE_ALL_COLOCATION);
if (list_length(nodeToSyncMetadataConnections) != 0) List *pgDistTableMetadata = NIL;
/* remove all dist table and object related metadata first */
pgDistTableMetadata = lappend(pgDistTableMetadata,
DELETE_ALL_PARTITIONS);
pgDistTableMetadata = lappend(pgDistTableMetadata,
DELETE_ALL_SHARDS);
pgDistTableMetadata = lappend(pgDistTableMetadata,
DELETE_ALL_PLACEMENTS);
pgDistTableMetadata = lappend(pgDistTableMetadata,
DELETE_ALL_DISTRIBUTED_OBJECTS);
pgDistTableMetadata = lappend(pgDistTableMetadata,
DELETE_ALL_COLOCATION);
if (syncContext.syncImmediately)
{ {
SendCommandListToWorkerOutsideTransactionWithConnection(linitial( SendCommandListToWorkerOutsideTransactionWithConnection(linitial(
nodeToSyncMetadataConnections), syncContext.
* nodeConnectionList),
metadataSnapshotCommandList); pgDistTableMetadata);
} }
/* create pg_dist_partition, pg_dist_shard and pg_dist_placement entries */ /* create pg_dist_partition, pg_dist_shard and pg_dist_placement entries */
@ -758,38 +753,105 @@ PgDistTableMetadataSyncCommandList(List *nodeToSyncMetadataConnections,
List *tableMetadataCreateCommandList = List *tableMetadataCreateCommandList =
CitusTableMetadataCreateCommandList(cacheEntry->relationId); CitusTableMetadataCreateCommandList(cacheEntry->relationId);
SendCommandListToWorkerOutsideTransactionWithConnection(linitial( if (syncContext.syncImmediately)
nodeToSyncMetadataConnections), {
tableMetadataCreateCommandList); SendCommandListToWorkerOutsideTransactionWithConnection(linitial(
syncContext.
nodeConnectionList),
tableMetadataCreateCommandList);
*metadataSnapshotCommandList = list_concat(*metadataSnapshotCommandList, list_free_deep(tableMetadataCreateCommandList);
tableMetadataCreateCommandList); }
else
{
pgDistTableMetadata = list_concat(pgDistTableMetadata,
tableMetadataCreateCommandList);
}
} }
/* commands to insert pg_dist_colocation entries */ /* commands to insert pg_dist_colocation entries */
List *colocationGroupSyncCommandList = ColocationGroupCreateCommandList(); List *colocationGroupSyncCommandList = ColocationGroupCreateCommandList();
*metadataSnapshotCommandList = list_concat(*metadataSnapshotCommandList, if (syncContext.syncImmediately)
colocationGroupSyncCommandList); {
SendCommandListToWorkerOutsideTransactionWithConnection(linitial(
syncContext.
nodeConnectionList),
colocationGroupSyncCommandList);
list_free_deep(colocationGroupSyncCommandList);
}
else
{
pgDistTableMetadata = list_concat(pgDistTableMetadata,
colocationGroupSyncCommandList);
}
List *distributedObjectSyncCommandList = DistributedObjectMetadataSyncCommandList(); List *distributedObjectSyncCommandList = DistributedObjectMetadataSyncCommandList();
*metadataSnapshotCommandList = list_concat(*metadataSnapshotCommandList, if (syncContext.syncImmediately)
distributedObjectSyncCommandList); {
SendCommandListToWorkerOutsideTransactionWithConnection(linitial(
syncContext.
nodeConnectionList),
distributedObjectSyncCommandList);
*metadataSnapshotCommandList = lcons(DISABLE_DDL_PROPAGATION, list_free_deep(distributedObjectSyncCommandList);
*metadataSnapshotCommandList); }
*metadataSnapshotCommandList = lappend(*metadataSnapshotCommandList, else
ENABLE_DDL_PROPAGATION); {
pgDistTableMetadata = list_concat(pgDistTableMetadata,
distributedObjectSyncCommandList);
}
if (!syncContext.syncImmediately)
{
pgDistTableMetadata = lcons(DISABLE_DDL_PROPAGATION,
pgDistTableMetadata);
pgDistTableMetadata = lappend(pgDistTableMetadata,
ENABLE_DDL_PROPAGATION);
syncContext.ddlCommandList = list_concat(syncContext.ddlCommandList,
pgDistTableMetadata);
}
} }
void
SyncMetadataCommands(MetadataSyncContext syncContext, List *commandList,
bool raiseErrors)
{
/* iterate over the commands and execute them in the same connection */
const char *commandString = NULL;
foreach_ptr(commandString, commandList)
{
MultiConnection *connection = NULL;
foreach_ptr(connection, syncContext.nodeConnectionList)
{
bool success = SendRemoteCommand(connection, commandString);
if (!success)
{
HandleRemoteTransactionConnectionError(connection, raiseErrors);
}
}
WaitForAllConnections(syncContext.nodeConnectionList, raiseErrors);
}
MultiConnection *connection = NULL;
foreach_ptr(connection, syncContext.nodeConnectionList)
{
ForgetResults(connection);
}
}
/* /*
* PropagateNodeWideObjectsCommandList is called during node activation to * PropagateNodeWideObjectsCommandList is called during node activation to
* propagate any object that should be propagated for every node. These are * propagate any object that should be propagated for every node. These are
* generally not linked to any distributed object but change system wide behaviour. * generally not linked to any distributed object but change system wide behaviour.
*/ */
static void static void
PropagateNodeWideObjectsCommandList(List *nodeToSyncMetadataConnections, PropagateNodeWideObjectsCommandList(MetadataSyncContext syncContext)
List **nodeWideObjectCommandList)
{ {
if (EnableAlterRoleSetPropagation) if (EnableAlterRoleSetPropagation)
{ {
@ -799,23 +861,23 @@ PropagateNodeWideObjectsCommandList(List *nodeToSyncMetadataConnections,
*/ */
List *alterRoleSetCommands = GenerateAlterRoleSetCommandForRole(InvalidOid); List *alterRoleSetCommands = GenerateAlterRoleSetCommandForRole(InvalidOid);
if (nodeWideObjectCommandList != NULL) if (!syncContext.syncImmediately)
{ {
*nodeWideObjectCommandList = list_concat(*nodeWideObjectCommandList,
alterRoleSetCommands);
/* if there are command wrap them in enable_ddl_propagation off */ /* if there are command wrap them in enable_ddl_propagation off */
*nodeWideObjectCommandList = lcons(DISABLE_DDL_PROPAGATION, syncContext.ddlCommandList =
*nodeWideObjectCommandList); lappend(syncContext.ddlCommandList, DISABLE_DDL_PROPAGATION);
*nodeWideObjectCommandList = lappend(*nodeWideObjectCommandList,
ENABLE_DDL_PROPAGATION);
}
if (list_length(nodeToSyncMetadataConnections) != 0) syncContext.ddlCommandList =
list_concat(syncContext.ddlCommandList, alterRoleSetCommands);
syncContext.ddlCommandList =
lappend(syncContext.ddlCommandList, ENABLE_DDL_PROPAGATION);
}
else
{ {
SendCommandListToWorkerOutsideTransactionWithConnection(linitial( bool raiseErrors = true;
nodeToSyncMetadataConnections), SyncMetadataCommands(syncContext, alterRoleSetCommands, raiseErrors);
alterRoleSetCommands);
} }
} }
} }
@ -837,63 +899,51 @@ PropagateNodeWideObjectsCommandList(List *nodeToSyncMetadataConnections,
* requires it. * requires it.
*/ */
void void
SyncDistributedObjectsCommandList(List *nodeToSyncMetadataConnections, List **commandList) SyncDistributedObjectsCommandList(MetadataSyncContext syncContext)
{ {
/* /*
* Propagate node wide objects. It includes only roles for now. * Propagate node wide objects. It includes only roles for now.
*/ */
PropagateNodeWideObjectsCommandList(nodeToSyncMetadataConnections, commandList); PropagateNodeWideObjectsCommandList(syncContext);
/* /*
* Detach partitions, break dependencies between sequences and table then * Detach partitions, break dependencies between sequences and table then
* remove shell tables first. * remove shell tables first.
*/ */
DetachPartitionCommandList(nodeToSyncMetadataConnections, commandList); DetachPartitionCommandList(syncContext);
if (list_length(nodeToSyncMetadataConnections) != 0) if (syncContext.syncImmediately)
{ {
SendCommandListToWorkerOutsideTransactionWithConnection(linitial( SendCommandListToWorkerOutsideTransactionWithConnection(linitial(
nodeToSyncMetadataConnections), syncContext.
nodeConnectionList),
list_make1( list_make1(
BREAK_CITUS_TABLE_SEQUENCE_DEPENDENCY_COMMAND)); BREAK_CITUS_TABLE_SEQUENCE_DEPENDENCY_COMMAND));
SendCommandListToWorkerOutsideTransactionWithConnection(linitial( SendCommandListToWorkerOutsideTransactionWithConnection(linitial(
nodeToSyncMetadataConnections), syncContext.
nodeConnectionList),
list_make1( list_make1(
REMOVE_ALL_SHELL_TABLES_COMMAND)); REMOVE_ALL_SHELL_TABLES_COMMAND));
} }
else
if (commandList != NULL)
{ {
*commandList = lappend(*commandList, syncContext.ddlCommandList = lappend(syncContext.ddlCommandList,
BREAK_CITUS_TABLE_SEQUENCE_DEPENDENCY_COMMAND); BREAK_CITUS_TABLE_SEQUENCE_DEPENDENCY_COMMAND);
*commandList = lappend(*commandList, REMOVE_ALL_SHELL_TABLES_COMMAND); syncContext.ddlCommandList = lappend(syncContext.ddlCommandList,
REMOVE_ALL_SHELL_TABLES_COMMAND);
} }
/* /*
* Replicate all objects of the pg_dist_object to the remote node. * Replicate all objects of the pg_dist_object to the remote node.
*/ */
List *replicateAllObjectsToNodeCommandList = NIL; ReplicateAllObjectsToNodeCommandList(syncContext);
ReplicateAllObjectsToNodeCommandList(nodeToSyncMetadataConnections,
&replicateAllObjectsToNodeCommandList);
if (commandList != NULL)
{
*commandList = list_concat(*commandList, replicateAllObjectsToNodeCommandList);
}
/* /*
* After creating each table, handle the inter table relationship between * After creating each table, handle the inter table relationship between
* those tables. * those tables.
*/ */
List *interTableRelationshipCommandList = NIL; InterTableRelationshipCommandList(syncContext);
InterTableRelationshipCommandList(nodeToSyncMetadataConnections,
&interTableRelationshipCommandList);
if (commandList != NULL)
{
*commandList = list_concat(*commandList, interTableRelationshipCommandList);
}
} }
@ -907,13 +957,8 @@ SyncDistributedObjectsCommandList(List *nodeToSyncMetadataConnections, List **co
* since all the dependencies should be present in the coordinator already. * since all the dependencies should be present in the coordinator already.
*/ */
static void static void
SyncDistributedObjectsToNodeList(List *nodeToSyncMetadataConnections) SyncDistributedObjectsToNodeList(MetadataSyncContext syncContext)
{ {
if (nodeToSyncMetadataConnections == NIL)
{
return;
}
EnsureSequentialModeMetadataOperations(); EnsureSequentialModeMetadataOperations();
Assert(ShouldPropagate()); Assert(ShouldPropagate());
@ -921,7 +966,7 @@ SyncDistributedObjectsToNodeList(List *nodeToSyncMetadataConnections)
/* send commands to new workers, the current user should be a superuser */ /* send commands to new workers, the current user should be a superuser */
Assert(superuser()); Assert(superuser());
SyncDistributedObjectsCommandList(nodeToSyncMetadataConnections, NULL); SyncDistributedObjectsCommandList(syncContext);
} }
@ -929,18 +974,14 @@ SyncDistributedObjectsToNodeList(List *nodeToSyncMetadataConnections)
* UpdateLocalGroupIdOnNode updates local group id on node. * UpdateLocalGroupIdOnNode updates local group id on node.
*/ */
static void static void
UpdateLocalGroupIdOnNode(WorkerNode *workerNode) UpdateLocalGroupIdOnNode(MultiConnection *connection, WorkerNode *workerNode)
{ {
if (NodeIsPrimary(workerNode) && !NodeIsCoordinator(workerNode))
{ {
List *commandList = list_make1(LocalGroupIdUpdateCommand(workerNode->groupId)); List *commandList = list_make1(LocalGroupIdUpdateCommand(workerNode->groupId));
/* send commands to new workers, the current user should be a superuser */ /* send commands to new workers, the current user should be a superuser */
Assert(superuser()); Assert(superuser());
SendMetadataCommandListToWorkerListInCoordinatedTransaction( SendCommandListToWorkerOutsideTransactionWithConnection(connection, commandList);
list_make1(workerNode),
CurrentUserName(),
commandList);
} }
} }
@ -951,14 +992,12 @@ UpdateLocalGroupIdOnNode(WorkerNode *workerNode)
* *
*/ */
static void static void
SyncPgDistTableMetadataToNodeList(List *nodeToSyncMetadataConnections) SyncPgDistTableMetadataToNodeList(MetadataSyncContext syncContext)
{ {
/* send commands to new workers, the current user should be a superuser */ /* send commands to new workers, the current user should be a superuser */
Assert(superuser()); Assert(superuser());
List *syncPgDistMetadataCommandList = NIL; PgDistTableMetadataSyncCommandList(syncContext);
PgDistTableMetadataSyncCommandList(nodeToSyncMetadataConnections,
&syncPgDistMetadataCommandList);
} }
@ -1180,7 +1219,11 @@ ActivateNodeList(List *nodeList)
List *nodeToSyncMetadata = NIL; List *nodeToSyncMetadata = NIL;
List *nodeToSyncMetadataConnections = NIL; MetadataSyncContext syncContext;
/* we don't need to collect any the ddl commands */
syncContext.syncImmediately = true;
syncContext.nodeConnectionList = NIL;
WorkerNode *node = NULL; WorkerNode *node = NULL;
foreach_ptr(node, nodeList) foreach_ptr(node, nodeList)
@ -1232,11 +1275,6 @@ ActivateNodeList(List *nodeList)
SetWorkerColumn(workerNode, Anum_pg_dist_node_metadatasynced, SetWorkerColumn(workerNode, Anum_pg_dist_node_metadatasynced,
BoolGetDatum(true)); BoolGetDatum(true));
/*
* Update local group id first, as object dependency logic requires to have
* updated local group id.
*/
UpdateLocalGroupIdOnNode(workerNode);
nodeToSyncMetadata = lappend(nodeToSyncMetadata, workerNode); nodeToSyncMetadata = lappend(nodeToSyncMetadata, workerNode);
@ -1247,13 +1285,19 @@ ActivateNodeList(List *nodeList)
ClaimConnectionExclusively(connection); ClaimConnectionExclusively(connection);
nodeToSyncMetadataConnections = syncContext.nodeConnectionList =
lappend(nodeToSyncMetadataConnections, connection); lappend(syncContext.nodeConnectionList, connection);
SendCommandListToWorkerOutsideTransactionWithConnection(linitial( SendCommandListToWorkerOutsideTransactionWithConnection(connection,
nodeToSyncMetadataConnections),
list_make1( list_make1(
DISABLE_DDL_PROPAGATION)); DISABLE_DDL_PROPAGATION));
/*
* Update local group id first, as object dependency logic requires to have
* updated local group id.
*/
UpdateLocalGroupIdOnNode(connection, workerNode);
} }
} }
@ -1262,24 +1306,22 @@ ActivateNodeList(List *nodeList)
* replicating reference tables to the remote node, as reference tables may * replicating reference tables to the remote node, as reference tables may
* need such objects. * need such objects.
*/ */
SyncDistributedObjectsToNodeList(nodeToSyncMetadataConnections); SyncDistributedObjectsToNodeList(syncContext);
/* /*
* Sync node metadata. We must sync node metadata before syncing table * Sync node metadata. We must sync node metadata before syncing table
* related pg_dist_xxx metadata. Since table related metadata requires * related pg_dist_xxx metadata. Since table related metadata requires
* to have right pg_dist_node entries. * to have right pg_dist_node entries.
*/ */
foreach_ptr(node, nodeToSyncMetadata) SyncNodeMetadataToNode(node->workerName, node->workerPort);
{
SyncNodeMetadataToNode(node->workerName, node->workerPort);
}
/* /*
* As the last step, sync the table related metadata to the remote node. * As the last step, sync the table related metadata to the remote node.
* We must handle it as the last step because of limitations shared with * We must handle it as the last step because of limitations shared with
* above comments. * above comments.
*/ */
SyncPgDistTableMetadataToNodeList(nodeToSyncMetadataConnections); SyncPgDistTableMetadataToNodeList(syncContext);
foreach_ptr(node, nodeList) foreach_ptr(node, nodeList)
{ {

12
src/backend/distributed/test/metadata_sync.c
View File

@ -49,17 +49,21 @@ activate_node_snapshot(PG_FUNCTION_ARGS)
*/ */
WorkerNode *dummyWorkerNode = GetFirstPrimaryWorkerNode(); WorkerNode *dummyWorkerNode = GetFirstPrimaryWorkerNode();
List *updateLocalGroupCommand = List *updateLocalGroupCommand =
list_make1(LocalGroupIdUpdateCommand(dummyWorkerNode->groupId)); list_make1(LocalGroupIdUpdateCommand(dummyWorkerNode->groupId));
List *syncDistObjCommands = NIL; MetadataSyncContext syncContext = { false, NIL, NIL };
SyncDistributedObjectsCommandList(NIL, &syncDistObjCommands);
SyncDistributedObjectsCommandList(syncContext);
List *syncDistObjCommands = syncContext.ddlCommandList;
List *dropSnapshotCommands = NodeMetadataDropCommands(); List *dropSnapshotCommands = NodeMetadataDropCommands();
List *createSnapshotCommands = NodeMetadataCreateCommands(); List *createSnapshotCommands = NodeMetadataCreateCommands();
List *pgDistTableMetadataSyncCommands = NIL; syncContext.ddlCommandList = NIL;
PgDistTableMetadataSyncCommandList(NIL, &pgDistTableMetadataSyncCommands); PgDistTableMetadataSyncCommandList(syncContext);
List *pgDistTableMetadataSyncCommands = syncContext.ddlCommandList;
List *activateNodeCommandList = NIL; List *activateNodeCommandList = NIL;
int activateNodeCommandIndex = 0; int activateNodeCommandIndex = 0;

1
src/include/distributed/commands/utility_hook.h
View File

@ -76,6 +76,7 @@ typedef struct DDLJob
List *taskList; /* worker DDL tasks to execute */ List *taskList; /* worker DDL tasks to execute */
} DDLJob; } DDLJob;
extern ProcessUtility_hook_type PrevProcessUtility; extern ProcessUtility_hook_type PrevProcessUtility;
extern void multi_ProcessUtility(PlannedStmt *pstmt, const char *queryString, extern void multi_ProcessUtility(PlannedStmt *pstmt, const char *queryString,

4
src/include/distributed/metadata_sync.h
View File

@ -48,7 +48,6 @@ typedef struct SequenceInfo
/* Functions declarations for metadata syncing */ /* Functions declarations for metadata syncing */
extern void SyncNodeMetadataToNode(const char *nodeNameString, int32 nodePort);
extern void SyncCitusTableMetadata(Oid relationId); extern void SyncCitusTableMetadata(Oid relationId);
extern void EnsureSequentialModeMetadataOperations(void); extern void EnsureSequentialModeMetadataOperations(void);
extern bool ClusterHasKnownMetadataWorkers(void); extern bool ClusterHasKnownMetadataWorkers(void);
@ -90,8 +89,7 @@ extern char * PlacementUpsertCommand(uint64 shardId, uint64 placementId, int sha
extern TableDDLCommand * TruncateTriggerCreateCommand(Oid relationId); extern TableDDLCommand * TruncateTriggerCreateCommand(Oid relationId);
extern void CreateInterTableRelationshipOfRelationOnWorkers(Oid relationId); extern void CreateInterTableRelationshipOfRelationOnWorkers(Oid relationId);
extern List * InterTableRelationshipOfRelationCommandList(Oid relationId); extern List * InterTableRelationshipOfRelationCommandList(Oid relationId);
extern void DetachPartitionCommandList(List *nodeToSyncMetadataConnections, extern void DetachPartitionCommandList(MetadataSyncContext syncContext);
List **detachPartitionCommandList);
extern void SyncNodeMetadataToNodes(void); extern void SyncNodeMetadataToNodes(void);
extern BackgroundWorkerHandle * SpawnSyncNodeMetadataToNodes(Oid database, Oid owner); extern BackgroundWorkerHandle * SpawnSyncNodeMetadataToNodes(Oid database, Oid owner);
extern void SyncNodeMetadataToNodesMain(Datum main_arg); extern void SyncNodeMetadataToNodesMain(Datum main_arg);

4
src/include/distributed/metadata_utility.h
View File

@ -264,6 +264,7 @@ typedef struct BackgroundTask
} __nullable_storage; } __nullable_storage;
} BackgroundTask; } BackgroundTask;
#define SET_NULLABLE_FIELD(ptr, field, value) \ #define SET_NULLABLE_FIELD(ptr, field, value) \
(ptr)->__nullable_storage.field = (value); \ (ptr)->__nullable_storage.field = (value); \
(ptr)->field = &((ptr)->__nullable_storage.field) (ptr)->field = &((ptr)->__nullable_storage.field)
@ -343,8 +344,7 @@ extern List * GetAllDependencyCreateDDLCommands(const List *dependencies);
extern bool ShouldPropagate(void); extern bool ShouldPropagate(void);
extern bool ShouldPropagateCreateInCoordinatedTransction(void); extern bool ShouldPropagateCreateInCoordinatedTransction(void);
extern bool ShouldPropagateAnyObject(List *addresses); extern bool ShouldPropagateAnyObject(List *addresses);
extern void ReplicateAllObjectsToNodeCommandList(List *nodeToSyncMetadataConnections, extern void ReplicateAllObjectsToNodeCommandList(MetadataSyncContext syncContext);
List **ddlCommands);
/* Remaining metadata utility functions */ /* Remaining metadata utility functions */
extern Oid TableOwnerOid(Oid relationId); extern Oid TableOwnerOid(Oid relationId);

12
src/include/distributed/worker_manager.h
View File

@ -16,6 +16,8 @@
#include "postgres.h" #include "postgres.h"
#include "distributed/metadata_utility.h"
#include "distributed/metadata_sync_context.h"
#include "storage/lmgr.h" #include "storage/lmgr.h"
#include "storage/lockdefs.h" #include "storage/lockdefs.h"
#include "nodes/pg_list.h" #include "nodes/pg_list.h"
@ -105,11 +107,11 @@ extern WorkerNode * SetWorkerColumnLocalOnly(WorkerNode *workerNode, int columnI
Datum value); Datum value);
extern uint32 CountPrimariesWithMetadata(void); extern uint32 CountPrimariesWithMetadata(void);
extern WorkerNode * GetFirstPrimaryWorkerNode(void); extern WorkerNode * GetFirstPrimaryWorkerNode(void);
extern void SyncDistributedObjectsCommandList(List *nodeToSyncMetadataConnections, extern void SyncDistributedObjectsCommandList(MetadataSyncContext syncContext);
List **commandList); extern void PgDistTableMetadataSyncCommandList(MetadataSyncContext syncContext);
extern void PgDistTableMetadataSyncCommandList(List *nodeToSyncMetadataConnections, extern void
List **metadataSnapshotCommandList); SyncMetadataCommands(MetadataSyncContext syncContext, List *commandList,
bool raiseErrors);
/* Function declarations for worker node utilities */ /* Function declarations for worker node utilities */
extern int CompareWorkerNodes(const void *leftElement, const void *rightElement); extern int CompareWorkerNodes(const void *leftElement, const void *rightElement);
extern uint32 WorkerNodeHashCode(const void *key, Size keySize); extern uint32 WorkerNodeHashCode(const void *key, Size keySize);