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

Rename argument to localOnly

pg_dist_object-metadatasync
Jelte Fennema 2021-07-22 11:33:05 +02:00
parent f3b7e53734
commit 428b819516
9 changed files with 27 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/backend/distributed/commands/create_distributed_table.c
View File

@ -700,8 +700,8 @@ MarkSequenceListDistributed(List *sequenceList)
ObjectAddress address; ObjectAddress address;
ObjectAddressSet(address, RelationRelationId, sequenceOid); ObjectAddressSet(address, RelationRelationId, sequenceOid);
bool shouldSyncMetadata = true; bool localOnly = false;
MarkObjectDistributed(&address, shouldSyncMetadata); MarkObjectDistributed(&address, localOnly);
} }
} }

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

@ -121,8 +121,8 @@ EnsureDependenciesExistOnAllNodes(const ObjectAddress *target)
ObjectAddress *dependency = NULL; ObjectAddress *dependency = NULL;
foreach_ptr(dependency, dependenciesWithCommands) foreach_ptr(dependency, dependenciesWithCommands)
{ {
bool shouldSyncMetadata = true; bool localOnly = false;
MarkObjectDistributed(dependency, shouldSyncMetadata); MarkObjectDistributed(dependency, localOnly);
} }
} }

4
src/backend/distributed/commands/extension.c
View File

@ -591,8 +591,8 @@ MarkExistingObjectDependenciesDistributedIfSupported()
ObjectAddress *objectAddress = NULL; ObjectAddress *objectAddress = NULL;
foreach_ptr(objectAddress, uniqueObjectAddresses) foreach_ptr(objectAddress, uniqueObjectAddresses)
{ {
bool shouldSyncMetadata = true; bool localOnly = false;
MarkObjectDistributed(objectAddress, shouldSyncMetadata); MarkObjectDistributed(objectAddress, localOnly);
} }
} }

20
src/backend/distributed/commands/function.c
View File

@ -191,8 +191,8 @@ create_distributed_function(PG_FUNCTION_ARGS)
ENABLE_DDL_PROPAGATION); ENABLE_DDL_PROPAGATION);
SendCommandToWorkersAsUser(NON_COORDINATOR_NODES, CurrentUserName(), ddlCommand.data); SendCommandToWorkersAsUser(NON_COORDINATOR_NODES, CurrentUserName(), ddlCommand.data);
bool shouldSyncMetadata = true; bool localOnly = false;
MarkObjectDistributed(&functionAddress, shouldSyncMetadata); MarkObjectDistributed(&functionAddress, localOnly);
if (distributionArgumentName != NULL) if (distributionArgumentName != NULL)
{ {
@ -238,9 +238,9 @@ DistributeFunctionWithDistributionArgument(RegProcedure funcOid,
distributionArgumentOid); distributionArgumentOid);
/* record the distribution argument and colocationId */ /* record the distribution argument and colocationId */
bool shouldSyncMetadata = true; bool localOnly = false;
UpdateFunctionDistributionInfo(functionAddress, &distributionArgumentIndex, UpdateFunctionDistributionInfo(functionAddress, &distributionArgumentIndex,
&colocationId, shouldSyncMetadata); &colocationId, localOnly);
/* /*
* Once we have at least one distributed function/procedure with distribution * Once we have at least one distributed function/procedure with distribution
@ -277,10 +277,10 @@ DistributeFunctionColocatedWithDistributedTable(RegProcedure funcOid,
" parameter should also be provided"))); " parameter should also be provided")));
} }
bool shouldSyncMetadata = true; bool localOnly = false;
/* set distribution argument and colocationId to NULL */ /* set distribution argument and colocationId to NULL */
UpdateFunctionDistributionInfo(functionAddress, NULL, NULL, shouldSyncMetadata); UpdateFunctionDistributionInfo(functionAddress, NULL, NULL, localOnly);
} }
@ -296,9 +296,9 @@ DistributeFunctionColocatedWithReferenceTable(const ObjectAddress *functionAddre
/* set distribution argument to NULL and colocationId to the reference table colocation id */ /* set distribution argument to NULL and colocationId to the reference table colocation id */
int *distributionArgumentIndex = NULL; int *distributionArgumentIndex = NULL;
bool shouldSyncMetadata = true; bool localOnly = false;
UpdateFunctionDistributionInfo(functionAddress, distributionArgumentIndex, UpdateFunctionDistributionInfo(functionAddress, distributionArgumentIndex,
&colocationId, shouldSyncMetadata); &colocationId, localOnly);
/* /*
* Once we have at least one distributed function/procedure that reads * Once we have at least one distributed function/procedure that reads
@ -573,7 +573,7 @@ EnsureFunctionCanBeColocatedWithTable(Oid functionOid, Oid distributionColumnTyp
void void
UpdateFunctionDistributionInfo(const ObjectAddress *distAddress, UpdateFunctionDistributionInfo(const ObjectAddress *distAddress,
int *distribution_argument_index, int *distribution_argument_index,
int *colocationId, bool shouldSyncMetadata) int *colocationId, bool localOnly)
{ {
const bool indexOK = true; const bool indexOK = true;
@ -644,7 +644,7 @@ UpdateFunctionDistributionInfo(const ObjectAddress *distAddress,
table_close(pgDistObjectRel, NoLock); table_close(pgDistObjectRel, NoLock);
if (shouldSyncMetadata) if (!localOnly)
{ {
char *workerMetadataUpdateCommand = DistributedObjectCreateCommand( char *workerMetadataUpdateCommand = DistributedObjectCreateCommand(
distAddress, distribution_argument_index, colocationId); distAddress, distribution_argument_index, colocationId);

4
src/backend/distributed/commands/utility_hook.c
View File

@ -654,8 +654,8 @@ ProcessUtilityInternal(PlannedStmt *pstmt,
if (ops && ops->markDistributed) if (ops && ops->markDistributed)
{ {
ObjectAddress address = GetObjectAddressFromParseTree(parsetree, false); ObjectAddress address = GetObjectAddressFromParseTree(parsetree, false);
bool shouldSyncMetadata = true; bool localOnly = false;
MarkObjectDistributed(&address, shouldSyncMetadata); MarkObjectDistributed(&address, localOnly);
} }
} }

4
src/backend/distributed/metadata/distobject.c
View File

@ -141,7 +141,7 @@ ObjectExists(const ObjectAddress *address)
* by adding appropriate entries to citus.pg_dist_object. * by adding appropriate entries to citus.pg_dist_object.
*/ */
void void
MarkObjectDistributed(const ObjectAddress *distAddress, bool shouldSyncMetadata) MarkObjectDistributed(const ObjectAddress *distAddress, bool localOnly)
{ {
int paramCount = 3; int paramCount = 3;
Oid paramTypes[3] = { Oid paramTypes[3] = {
@ -165,7 +165,7 @@ MarkObjectDistributed(const ObjectAddress *distAddress, bool shouldSyncMetadata)
ereport(ERROR, (errmsg("failed to insert object into citus.pg_dist_object"))); ereport(ERROR, (errmsg("failed to insert object into citus.pg_dist_object")));
} }
if (shouldSyncMetadata) if (!localOnly)
{ {
char *workerMetadataUpdateCommand = DistributedObjectCreateCommand( char *workerMetadataUpdateCommand = DistributedObjectCreateCommand(
distAddress, NULL, NULL); distAddress, NULL, NULL);

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

@ -399,8 +399,8 @@ SyncMetadataSnapshotToNode(WorkerNode *workerNode, bool raiseOnError)
ObjectAddress *address; ObjectAddress *address;
foreach_ptr(address, newDistributedObjects) foreach_ptr(address, newDistributedObjects)
{ {
bool shouldSyncMetadata = true; bool localOnly = false;
MarkObjectDistributed(address, shouldSyncMetadata); MarkObjectDistributed(address, localOnly);
} }
return true; return true;
} }
@ -2661,10 +2661,10 @@ citus_internal_add_object_metadata(PG_FUNCTION_ARGS)
ObjectAddress address = { 0 }; ObjectAddress address = { 0 };
ObjectAddressSubSet(address, classid, objid, objsubid); ObjectAddressSubSet(address, classid, objid, objsubid);
bool shouldSyncMetadata = false; bool localOnly = true;
MarkObjectDistributed(&address, shouldSyncMetadata); MarkObjectDistributed(&address, localOnly);
UpdateFunctionDistributionInfo(&address, distributionArgumentIndex, colocationId, UpdateFunctionDistributionInfo(&address, distributionArgumentIndex, colocationId,
shouldSyncMetadata); localOnly);
PG_RETURN_VOID(); PG_RETURN_VOID();
} }

2
src/include/distributed/commands.h
View File

@ -474,7 +474,7 @@ extern ObjectWithArgs * ObjectWithArgsFromOid(Oid funcOid);
extern void UpdateFunctionDistributionInfo(const ObjectAddress *distAddress, extern void UpdateFunctionDistributionInfo(const ObjectAddress *distAddress,
int *distribution_argument_index, int *distribution_argument_index,
int *colocationId, int *colocationId,
bool shouldSyncMetadata); bool localOnly);
/* vacuum.c - forward declarations */ /* vacuum.c - forward declarations */
extern void PostprocessVacuumStmt(VacuumStmt *vacuumStmt, const char *vacuumCommand); extern void PostprocessVacuumStmt(VacuumStmt *vacuumStmt, const char *vacuumCommand);

2
src/include/distributed/metadata/distobject.h
View File

@ -21,7 +21,7 @@ extern bool CitusExtensionObject(const ObjectAddress *objectAddress);
extern bool IsObjectDistributed(const ObjectAddress *address); extern bool IsObjectDistributed(const ObjectAddress *address);
extern bool ClusterHasDistributedFunctionWithDistArgument(void); extern bool ClusterHasDistributedFunctionWithDistArgument(void);
extern void MarkObjectDistributed(const ObjectAddress *distAddress, extern void MarkObjectDistributed(const ObjectAddress *distAddress,
bool shouldSyncMetadata); bool localOnly);
extern void UnmarkObjectDistributed(const ObjectAddress *address); extern void UnmarkObjectDistributed(const ObjectAddress *address);
extern bool IsTableOwnedByExtension(Oid relationId); extern bool IsTableOwnedByExtension(Oid relationId);
extern bool IsObjectAddressOwnedByExtension(const ObjectAddress *target, extern bool IsObjectAddressOwnedByExtension(const ObjectAddress *target,