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

refactor code where GetObjectAddressFromParseTree is called because it returns list of addresses now

pull/6063/head
aykutbozkurt 2022-07-14 16:33:08 +03:00
parent 9d232d7b00
commit ebb6d1c8c0
32 changed files with 525 additions and 253 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
src/backend/distributed/commands/alter_table.c
View File

@ -1302,7 +1302,7 @@ ErrorIfUnsupportedCascadeObjects(Oid relationId)
*
* Extension dependency is different than the rest. If an object depends on an extension
* dropping the object would drop the extension too.
* So we check with IsObjectAddressOwnedByExtension function.
* So we check with IsAnyObjectAddressOwnedByExtension function.
*/
static bool
DoesCascadeDropUnsupportedObject(Oid classId, Oid objectId, HTAB *nodeMap)
@ -1315,10 +1315,9 @@ DoesCascadeDropUnsupportedObject(Oid classId, Oid objectId, HTAB *nodeMap)
return false;
}
ObjectAddress objectAddress = { 0 };
ObjectAddressSet(objectAddress, classId, objectId);
if (IsObjectAddressOwnedByExtension(&objectAddress, NULL))
ObjectAddress *objectAddress = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*objectAddress, classId, objectId);
if (IsAnyObjectAddressOwnedByExtension(list_make1(objectAddress), NULL))
{
return true;
}

6
src/backend/distributed/commands/citus_add_local_table_to_metadata.c
View File

@ -307,8 +307,8 @@ CreateCitusLocalTable(Oid relationId, bool cascadeViaForeignKeys, bool autoConve
}
}
ObjectAddress tableAddress = { 0 };
ObjectAddressSet(tableAddress, RelationRelationId, relationId);
ObjectAddress *tableAddress = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*tableAddress, RelationRelationId, relationId);
/*
* Ensure that the sequences used in column defaults of the table
@ -320,7 +320,7 @@ CreateCitusLocalTable(Oid relationId, bool cascadeViaForeignKeys, bool autoConve
* Ensure dependencies exist as we will create shell table on the other nodes
* in the MX case.
*/
EnsureDependenciesExistOnAllNodes(&tableAddress);
EnsureAllObjectDependenciesExistOnAllNodes(list_make1(tableAddress));
/*
* Make sure that existing reference tables have been replicated to all

37
src/backend/distributed/commands/common.c
View File

@ -61,22 +61,26 @@ PostprocessCreateDistributedObjectFromCatalogStmt(Node *stmt, const char *queryS
return NIL;
}
ObjectAddress address = GetObjectAddressFromParseTree(stmt, false);
List *addresses = GetObjectAddressListFromParseTree(stmt, false);
/* the code-path only supports a single object */
Assert(list_length(addresses) == 1);
EnsureCoordinator();
EnsureSequentialMode(ops->objectType);
/* If the object has any unsupported dependency warn, and only create locally */
DeferredErrorMessage *depError = DeferErrorIfHasUnsupportedDependency(&address);
DeferredErrorMessage *depError = DeferErrorIfAnyObjectHasUnsupportedDependency(
addresses);
if (depError != NULL)
{
RaiseDeferredError(depError, WARNING);
return NIL;
}
EnsureDependenciesExistOnAllNodes(&address);
EnsureAllObjectDependenciesExistOnAllNodes(addresses);
List *commands = GetDependencyCreateDDLCommands(&address);
List *commands = GetAllDependencyCreateDDLCommands(addresses);
commands = lcons(DISABLE_DDL_PROPAGATION, commands);
commands = lappend(commands, ENABLE_DDL_PROPAGATION);
@ -111,8 +115,12 @@ PreprocessAlterDistributedObjectStmt(Node *stmt, const char *queryString,
const DistributeObjectOps *ops = GetDistributeObjectOps(stmt);
Assert(ops != NULL);
ObjectAddress address = GetObjectAddressFromParseTree(stmt, false);
if (!ShouldPropagateObject(&address))
List *addresses = GetObjectAddressListFromParseTree(stmt, false);
/* the code-path only supports a single object */
Assert(list_length(addresses) == 1);
if (!ShouldPropagateAnyObject(addresses))
{
return NIL;
}
@ -156,8 +164,12 @@ PostprocessAlterDistributedObjectStmt(Node *stmt, const char *queryString)
const DistributeObjectOps *ops = GetDistributeObjectOps(stmt);
Assert(ops != NULL);
ObjectAddress address = GetObjectAddressFromParseTree(stmt, false);
if (!ShouldPropagateObject(&address))
List *addresses = GetObjectAddressListFromParseTree(stmt, false);
/* the code-path only supports a single object */
Assert(list_length(addresses) == 1);
if (!ShouldPropagateAnyObject(addresses))
{
return NIL;
}
@ -168,7 +180,7 @@ PostprocessAlterDistributedObjectStmt(Node *stmt, const char *queryString)
return NIL;
}
EnsureDependenciesExistOnAllNodes(&address);
EnsureAllObjectDependenciesExistOnAllNodes(addresses);
return NIL;
}
@ -223,11 +235,10 @@ PreprocessDropDistributedObjectStmt(Node *node, const char *queryString,
Relation rel = NULL; /* not used, but required to pass to get_object_address */
ObjectAddress address = get_object_address(stmt->removeType, object, &rel,
AccessShareLock, stmt->missing_ok);
if (IsObjectDistributed(&address))
ObjectAddress *addressPtr = palloc0(sizeof(ObjectAddress));
*addressPtr = address;
if (IsAnyObjectDistributed(list_make1(addressPtr)))
{
ObjectAddress *addressPtr = palloc0(sizeof(ObjectAddress));
*addressPtr = address;
distributedObjects = lappend(distributedObjects, object);
distributedObjectAddresses = lappend(distributedObjectAddresses, addressPtr);
}

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

@ -442,10 +442,9 @@ CreateDistributedTable(Oid relationId, char *distributionColumnName,
* via their own connection and committed immediately so they become visible to all
* sessions creating shards.
*/
ObjectAddress tableAddress = { 0 };
ObjectAddressSet(tableAddress, RelationRelationId, relationId);
EnsureDependenciesExistOnAllNodes(&tableAddress);
ObjectAddress *tableAddress = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*tableAddress, RelationRelationId, relationId);
EnsureAllObjectDependenciesExistOnAllNodes(list_make1(tableAddress));
char replicationModel = DecideReplicationModel(distributionMethod,
colocateWithTableName,

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

@ -36,6 +36,9 @@ static void ErrorIfCircularDependencyExists(const ObjectAddress *objectAddress);
static int ObjectAddressComparator(const void *a, const void *b);
static List * FilterObjectAddressListByPredicate(List *objectAddressList,
AddressPredicate predicate);
static void EnsureDependenciesExistOnAllNodes(const ObjectAddress *target);
static List * GetDependencyCreateDDLCommands(const ObjectAddress *dependency);
static bool ShouldPropagateObject(const ObjectAddress *address);
/*
* EnsureDependenciesExistOnAllNodes finds all the dependencies that we support and makes
@ -51,7 +54,7 @@ static List * FilterObjectAddressListByPredicate(List *objectAddressList,
* This is solved by creating the dependencies in an idempotent manner, either via
* postgres native CREATE IF NOT EXISTS, or citus helper functions.
*/
void
static void
EnsureDependenciesExistOnAllNodes(const ObjectAddress *target)
{
List *dependenciesWithCommands = NIL;
@ -142,6 +145,21 @@ EnsureDependenciesExistOnAllNodes(const ObjectAddress *target)
}
/*
* EnsureAllObjectDependenciesExistOnAllNodes iteratively calls EnsureDependenciesExistOnAllNodes
* for given targets.
*/
void
EnsureAllObjectDependenciesExistOnAllNodes(const List *targets)
{
ObjectAddress *target = NULL;
foreach_ptr(target, targets)
{
EnsureDependenciesExistOnAllNodes(target);
}
}
/*
* EnsureDependenciesCanBeDistributed ensures all dependencies of the given object
* can be distributed.
@ -153,7 +171,8 @@ EnsureDependenciesCanBeDistributed(const ObjectAddress *objectAddress)
ErrorIfCircularDependencyExists(objectAddress);
/* If the object has any unsupported dependency, error out */
DeferredErrorMessage *depError = DeferErrorIfHasUnsupportedDependency(objectAddress);
DeferredErrorMessage *depError = DeferErrorIfAnyObjectHasUnsupportedDependency(
list_make1((ObjectAddress *) objectAddress));
if (depError != NULL)
{
@ -310,7 +329,7 @@ GetDistributableDependenciesForObject(const ObjectAddress *target)
* GetDependencyCreateDDLCommands returns a list (potentially empty or NIL) of ddl
* commands to execute on a worker to create the object.
*/
List *
static List *
GetDependencyCreateDDLCommands(const ObjectAddress *dependency)
{
switch (getObjectClass(dependency))
@ -488,6 +507,25 @@ GetDependencyCreateDDLCommands(const ObjectAddress *dependency)
}
/*
* GetAllDependencyCreateDDLCommands iteratively calls GetDependencyCreateDDLCommands
* for given dependencies.
*/
List *
GetAllDependencyCreateDDLCommands(const List *dependencies)
{
List *commands = NIL;
ObjectAddress *dependency = NULL;
foreach_ptr(dependency, dependencies)
{
commands = list_concat(commands, GetDependencyCreateDDLCommands(dependency));
}
return commands;
}
/*
* ReplicateAllObjectsToNodeCommandList returns commands to replicate all
* previously marked objects to a worker node. The function also sets
@ -531,7 +569,7 @@ ReplicateAllObjectsToNodeCommandList(const char *nodeName, int nodePort)
ObjectAddress *dependency = NULL;
foreach_ptr(dependency, dependencies)
{
if (IsObjectAddressOwnedByExtension(dependency, NULL))
if (IsAnyObjectAddressOwnedByExtension(list_make1(dependency), NULL))
{
/*
* we expect extension-owned objects to be created as a result
@ -663,7 +701,7 @@ ShouldPropagateCreateInCoordinatedTransction()
* ShouldPropagateObject determines if we should be propagating DDLs based
* on their object address.
*/
bool
static bool
ShouldPropagateObject(const ObjectAddress *address)
{
if (!ShouldPropagate())
@ -671,7 +709,7 @@ ShouldPropagateObject(const ObjectAddress *address)
return false;
}
if (!IsObjectDistributed(address))
if (!IsAnyObjectDistributed(list_make1((ObjectAddress *) address)))
{
/* do not propagate for non-distributed types */
return false;
@ -681,6 +719,26 @@ ShouldPropagateObject(const ObjectAddress *address)
}
/*
* ShouldPropagateAnyObject determines if we should be propagating DDLs based
* on their object addresses.
*/
bool
ShouldPropagateAnyObject(List *addresses)
{
ObjectAddress *address = NULL;
foreach_ptr(address, addresses)
{
if (ShouldPropagateObject(address))
{
return true;
}
}
return false;
}
/*
* FilterObjectAddressListByPredicate takes a list of ObjectAddress *'s and returns a list
* only containing the ObjectAddress *'s for which the predicate returned true.

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

@ -181,9 +181,12 @@ PostprocessCreateExtensionStmt(Node *node, const char *queryString)
(void *) createExtensionStmtSql,
ENABLE_DDL_PROPAGATION);
ObjectAddress extensionAddress = GetObjectAddressFromParseTree(node, false);
List *extensionAddresses = GetObjectAddressListFromParseTree(node, false);
EnsureDependenciesExistOnAllNodes(&extensionAddress);
/* the code-path only supports a single object */
Assert(list_length(extensionAddresses) == 1);
EnsureAllObjectDependenciesExistOnAllNodes(extensionAddresses);
return NodeDDLTaskList(NON_COORDINATOR_NODES, commands);
}
@ -319,10 +322,9 @@ FilterDistributedExtensions(List *extensionObjectList)
continue;
}
ObjectAddress address = { 0 };
ObjectAddressSet(address, ExtensionRelationId, extensionOid);
if (!IsObjectDistributed(&address))
ObjectAddress *address = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*address, ExtensionRelationId, extensionOid);
if (!IsAnyObjectDistributed(list_make1(address)))
{
continue;
}
@ -411,7 +413,10 @@ PreprocessAlterExtensionSchemaStmt(Node *node, const char *queryString,
List *
PostprocessAlterExtensionSchemaStmt(Node *node, const char *queryString)
{
ObjectAddress extensionAddress = GetObjectAddressFromParseTree(node, false);
List *extensionAddresses = GetObjectAddressListFromParseTree(node, false);
/* the code-path only supports a single object */
Assert(list_length(extensionAddresses) == 1);
if (!ShouldPropagateExtensionCommand(node))
{
@ -419,7 +424,7 @@ PostprocessAlterExtensionSchemaStmt(Node *node, const char *queryString)
}
/* dependencies (schema) have changed let's ensure they exist */
EnsureDependenciesExistOnAllNodes(&extensionAddress);
EnsureAllObjectDependenciesExistOnAllNodes(extensionAddresses);
return NIL;
}
@ -504,7 +509,7 @@ PostprocessAlterExtensionCitusUpdateStmt(Node *node)
*
* Note that this function is not responsible for ensuring if dependencies exist on
* nodes and satisfying these dependendencies if not exists, which is already done by
* EnsureDependenciesExistOnAllNodes on demand. Hence, this function is just designed
* EnsureAllObjectDependenciesExistOnAllNodes on demand. Hence, this function is just designed
* to be used when "ALTER EXTENSION citus UPDATE" is executed.
* This is because we want to add existing objects that would have already been in
* pg_dist_object if we had created them in new version of Citus to pg_dist_object.

10
src/backend/distributed/commands/foreign_data_wrapper.c
View File

@ -64,6 +64,7 @@ PreprocessGrantOnFDWStmt(Node *node, const char *queryString,
EnsureCoordinator();
/* the code-path only supports a single object */
Assert(list_length(stmt->objects) == 1);
char *sql = DeparseTreeNode((Node *) stmt);
@ -87,12 +88,15 @@ NameListHasFDWOwnedByDistributedExtension(List *FDWNames)
foreach_ptr(FDWValue, FDWNames)
{
/* captures the extension address during lookup */
ObjectAddress extensionAddress = { 0 };
ObjectAddress *extensionAddress = palloc0(sizeof(ObjectAddress));
ObjectAddress FDWAddress = GetObjectAddressByFDWName(strVal(FDWValue), false);
if (IsObjectAddressOwnedByExtension(&FDWAddress, &extensionAddress))
ObjectAddress *copyFDWAddress = palloc0(sizeof(ObjectAddress));
*copyFDWAddress = FDWAddress;
if (IsAnyObjectAddressOwnedByExtension(list_make1(copyFDWAddress),
extensionAddress))
{
if (IsObjectDistributed(&extensionAddress))
if (IsAnyObjectDistributed(list_make1(extensionAddress)))
{
return true;
}

12
src/backend/distributed/commands/foreign_server.c
View File

@ -102,6 +102,7 @@ PreprocessGrantOnForeignServerStmt(Node *node, const char *queryString,
EnsureCoordinator();
/* the code-path only supports a single object */
Assert(list_length(stmt->objects) == 1);
char *sql = DeparseTreeNode((Node *) stmt);
@ -247,15 +248,14 @@ NameListHasDistributedServer(List *serverNames)
foreach_ptr(serverValue, serverNames)
{
List *addresses = GetObjectAddressByServerName(strVal(serverValue), false);
if (list_length(addresses) > 1)
{
ereport(ERROR, errmsg(
"citus does not support multiple object addresses in NameListHasDistributedServer"));
}
/* the code-path only supports a single object */
Assert(list_length(addresses) == 1);
/* We have already asserted that we have exactly 1 address in the addresses. */
ObjectAddress *address = linitial(addresses);
if (IsObjectDistributed(address))
if (IsAnyObjectDistributed(list_make1(address)))
{
return true;
}

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

@ -128,7 +128,7 @@ create_distributed_function(PG_FUNCTION_ARGS)
text *colocateWithText = NULL; /* optional */
StringInfoData ddlCommand = { 0 };
ObjectAddress functionAddress = { 0 };
ObjectAddress *functionAddress = palloc0(sizeof(ObjectAddress));
Oid distributionArgumentOid = InvalidOid;
bool colocatedWithReferenceTable = false;
@ -203,9 +203,9 @@ create_distributed_function(PG_FUNCTION_ARGS)
EnsureCoordinator();
EnsureFunctionOwner(funcOid);
ObjectAddressSet(functionAddress, ProcedureRelationId, funcOid);
ObjectAddressSet(*functionAddress, ProcedureRelationId, funcOid);
if (RecreateSameNonColocatedFunction(functionAddress,
if (RecreateSameNonColocatedFunction(*functionAddress,
distributionArgumentName,
colocateWithTableNameDefault,
forceDelegationAddress))
@ -224,9 +224,10 @@ create_distributed_function(PG_FUNCTION_ARGS)
* pg_dist_object, and not propagate the CREATE FUNCTION. Function
* will be created by the virtue of the extension creation.
*/
if (IsObjectAddressOwnedByExtension(&functionAddress, &extensionAddress))
if (IsAnyObjectAddressOwnedByExtension(list_make1(functionAddress),
&extensionAddress))
{
EnsureExtensionFunctionCanBeDistributed(functionAddress, extensionAddress,
EnsureExtensionFunctionCanBeDistributed(*functionAddress, extensionAddress,
distributionArgumentName);
}
else
@ -237,7 +238,7 @@ create_distributed_function(PG_FUNCTION_ARGS)
*/
EnsureSequentialMode(OBJECT_FUNCTION);
EnsureDependenciesExistOnAllNodes(&functionAddress);
EnsureAllObjectDependenciesExistOnAllNodes(list_make1(functionAddress));
const char *createFunctionSQL = GetFunctionDDLCommand(funcOid, true);
const char *alterFunctionOwnerSQL = GetFunctionAlterOwnerCommand(funcOid);
@ -257,7 +258,7 @@ create_distributed_function(PG_FUNCTION_ARGS)
ddlCommand.data);
}
MarkObjectDistributed(&functionAddress);
MarkObjectDistributed(functionAddress);
if (distributionArgumentName != NULL)
{
@ -272,12 +273,12 @@ create_distributed_function(PG_FUNCTION_ARGS)
distributionArgumentOid,
colocateWithTableName,
forceDelegationAddress,
&functionAddress);
functionAddress);
}
else if (!colocatedWithReferenceTable)
{
DistributeFunctionColocatedWithDistributedTable(funcOid, colocateWithTableName,
&functionAddress);
functionAddress);
}
else if (colocatedWithReferenceTable)
{
@ -288,7 +289,7 @@ create_distributed_function(PG_FUNCTION_ARGS)
*/
ErrorIfAnyNodeDoesNotHaveMetadata();
DistributeFunctionColocatedWithReferenceTable(&functionAddress);
DistributeFunctionColocatedWithReferenceTable(functionAddress);
}
PG_RETURN_VOID();
@ -1308,7 +1309,7 @@ ShouldPropagateAlterFunction(const ObjectAddress *address)
return false;
}
if (!IsObjectDistributed(address))
if (!IsAnyObjectDistributed(list_make1((ObjectAddress *) address)))
{
/* do not propagate alter function for non-distributed functions */
return false;
@ -1373,15 +1374,19 @@ PostprocessCreateFunctionStmt(Node *node, const char *queryString)
return NIL;
}
ObjectAddress functionAddress = GetObjectAddressFromParseTree((Node *) stmt, false);
List *functionAddresses = GetObjectAddressListFromParseTree((Node *) stmt, false);
if (IsObjectAddressOwnedByExtension(&functionAddress, NULL))
/* the code-path only supports a single object */
Assert(list_length(functionAddresses) == 1);
if (IsAnyObjectAddressOwnedByExtension(functionAddresses, NULL))
{
return NIL;
}
/* If the function has any unsupported dependency, create it locally */
DeferredErrorMessage *errMsg = DeferErrorIfHasUnsupportedDependency(&functionAddress);
DeferredErrorMessage *errMsg = DeferErrorIfAnyObjectHasUnsupportedDependency(
functionAddresses);
if (errMsg != NULL)
{
@ -1389,11 +1394,14 @@ PostprocessCreateFunctionStmt(Node *node, const char *queryString)
return NIL;
}
EnsureDependenciesExistOnAllNodes(&functionAddress);
EnsureAllObjectDependenciesExistOnAllNodes(functionAddresses);
/* We have already asserted that we have exactly 1 address in the addresses. */
ObjectAddress *functionAddress = linitial(functionAddresses);
List *commands = list_make1(DISABLE_DDL_PROPAGATION);
commands = list_concat(commands, CreateFunctionDDLCommandsIdempotent(
&functionAddress));
functionAddress));
commands = list_concat(commands, list_make1(ENABLE_DDL_PROPAGATION));
return NodeDDLTaskList(NON_COORDINATOR_NODES, commands);
@ -1494,8 +1502,15 @@ PreprocessAlterFunctionStmt(Node *node, const char *queryString,
AlterFunctionStmt *stmt = castNode(AlterFunctionStmt, node);
AssertObjectTypeIsFunctional(stmt->objtype);
ObjectAddress address = GetObjectAddressFromParseTree((Node *) stmt, false);
if (!ShouldPropagateAlterFunction(&address))
List *addresses = GetObjectAddressListFromParseTree((Node *) stmt, false);
/* the code-path only supports a single object */
Assert(list_length(addresses) == 1);
/* We have already asserted that we have exactly 1 address in the addresses. */
ObjectAddress *address = linitial(addresses);
if (!ShouldPropagateAlterFunction(address))
{
return NIL;
}
@ -1549,20 +1564,26 @@ PreprocessAlterFunctionDependsStmt(Node *node, const char *queryString,
return NIL;
}
ObjectAddress address = GetObjectAddressFromParseTree((Node *) stmt, true);
if (!IsObjectDistributed(&address))
List *addresses = GetObjectAddressListFromParseTree((Node *) stmt, true);
/* the code-path only supports a single object */
Assert(list_length(addresses) == 1);
if (!IsAnyObjectDistributed(addresses))
{
return NIL;
}
/* We have already asserted that we have exactly 1 address in the addresses. */
ObjectAddress *address = linitial(addresses);
/*
* Distributed objects should not start depending on an extension, this will break
* the dependency resolving mechanism we use to replicate distributed objects to new
* workers
*/
const char *functionName =
getObjectIdentity_compat(&address, /* missingOk: */ false);
getObjectIdentity_compat(address, /* missingOk: */ false);
ereport(ERROR, (errmsg("distrtibuted functions are not allowed to depend on an "
"extension"),
errdetail("Function \"%s\" is already distributed. Functions from "
@ -1920,7 +1941,7 @@ EnsureExtensionFunctionCanBeDistributed(const ObjectAddress functionAddress,
/*
* Ensure corresponding extension is in pg_dist_object.
* Functions owned by an extension are depending internally on that extension,
* hence EnsureDependenciesExistOnAllNodes() creates the extension, which in
* hence EnsureAllObjectDependenciesExistOnAllNodes() creates the extension, which in
* turn creates the function, and thus we don't have to create it ourself like
* we do for non-extension functions.
*/
@ -1930,7 +1951,9 @@ EnsureExtensionFunctionCanBeDistributed(const ObjectAddress functionAddress,
get_extension_name(extensionAddress.objectId),
get_func_name(functionAddress.objectId))));
EnsureDependenciesExistOnAllNodes(&functionAddress);
ObjectAddress *copyFunctionAddress = palloc0(sizeof(ObjectAddress));
*copyFunctionAddress = functionAddress;
EnsureAllObjectDependenciesExistOnAllNodes(list_make1(copyFunctionAddress));
}
@ -2004,7 +2027,7 @@ PostprocessGrantOnFunctionStmt(Node *node, const char *queryString)
ObjectAddress *functionAddress = NULL;
foreach_ptr(functionAddress, distributedFunctions)
{
EnsureDependenciesExistOnAllNodes(functionAddress);
EnsureAllObjectDependenciesExistOnAllNodes(list_make1(functionAddress));
}
return NIL;
}
@ -2083,7 +2106,7 @@ FilterDistributedFunctions(GrantStmt *grantStmt)
* if this function from GRANT .. ON FUNCTION .. is a distributed
* function, add it to the list
*/
if (IsObjectDistributed(functionAddress))
if (IsAnyObjectDistributed(list_make1(functionAddress)))
{
grantFunctionList = lappend(grantFunctionList, functionAddress);
}

6
src/backend/distributed/commands/grant.c
View File

@ -238,9 +238,9 @@ CollectGrantTableIdList(GrantStmt *grantStmt)
}
/* check for distributed sequences included in GRANT ON TABLE statement */
ObjectAddress sequenceAddress = { 0 };
ObjectAddressSet(sequenceAddress, RelationRelationId, relationId);
if (IsObjectDistributed(&sequenceAddress))
ObjectAddress *sequenceAddress = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*sequenceAddress, RelationRelationId, relationId);
if (IsAnyObjectDistributed(list_make1(sequenceAddress)))
{
grantTableList = lappend_oid(grantTableList, relationId);
}

8
src/backend/distributed/commands/index.c
View File

@ -761,9 +761,9 @@ PostprocessIndexStmt(Node *node, const char *queryString)
Oid indexRelationId = get_relname_relid(indexStmt->idxname, schemaId);
/* ensure dependencies of index exist on all nodes */
ObjectAddress address = { 0 };
ObjectAddressSet(address, RelationRelationId, indexRelationId);
EnsureDependenciesExistOnAllNodes(&address);
ObjectAddress *address = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*address, RelationRelationId, indexRelationId);
EnsureAllObjectDependenciesExistOnAllNodes(list_make1(address));
/* furtheron we are only processing CONCURRENT index statements */
if (!indexStmt->concurrent)
@ -772,7 +772,7 @@ PostprocessIndexStmt(Node *node, const char *queryString)
}
/*
* EnsureDependenciesExistOnAllNodes could have distributed objects that are required
* EnsureAllObjectDependenciesExistOnAllNodes could have distributed objects that are required
* by this index. During the propagation process an active snapshout might be left as
* a side effect of inserting the local tuples via SPI. To not leak a snapshot like
* that we will pop any snapshot if we have any right before we commit.

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

@ -137,8 +137,12 @@ RoleSpecToObjectAddress(RoleSpec *role, bool missing_ok)
List *
PostprocessAlterRoleStmt(Node *node, const char *queryString)
{
ObjectAddress address = GetObjectAddressFromParseTree(node, false);
if (!ShouldPropagateObject(&address))
List *addresses = GetObjectAddressListFromParseTree(node, false);
/* the code-path only supports a single object */
Assert(list_length(addresses) == 1);
if (!ShouldPropagateAnyObject(addresses))
{
return NIL;
}
@ -208,14 +212,17 @@ PreprocessAlterRoleSetStmt(Node *node, const char *queryString,
return NIL;
}
ObjectAddress address = GetObjectAddressFromParseTree(node, false);
List *addresses = GetObjectAddressListFromParseTree(node, false);
/* the code-path only supports a single object */
Assert(list_length(addresses) == 1);
/*
* stmt->role could be NULL when the statement is on 'ALL' roles, we do propagate for
* ALL roles. If it is not NULL the role is for a specific role. If that role is not
* distributed we will not propagate the statement
*/
if (stmt->role != NULL && !IsObjectDistributed(&address))
if (stmt->role != NULL && !IsAnyObjectDistributed(addresses))
{
return NIL;
}
@ -1056,7 +1063,6 @@ FilterDistributedRoles(List *roles)
foreach_ptr(roleNode, roles)
{
RoleSpec *role = castNode(RoleSpec, roleNode);
ObjectAddress roleAddress = { 0 };
Oid roleOid = get_rolespec_oid(role, true);
if (roleOid == InvalidOid)
{
@ -1066,8 +1072,9 @@ FilterDistributedRoles(List *roles)
*/
continue;
}
ObjectAddressSet(roleAddress, AuthIdRelationId, roleOid);
if (IsObjectDistributed(&roleAddress))
ObjectAddress *roleAddress = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*roleAddress, AuthIdRelationId, roleOid);
if (IsAnyObjectDistributed(list_make1(roleAddress)))
{
distributedRoles = lappend(distributedRoles, role);
}
@ -1137,12 +1144,13 @@ PostprocessGrantRoleStmt(Node *node, const char *queryString)
RoleSpec *role = NULL;
foreach_ptr(role, stmt->grantee_roles)
{
ObjectAddress roleAddress = { 0 };
Oid roleOid = get_rolespec_oid(role, false);
ObjectAddressSet(roleAddress, AuthIdRelationId, roleOid);
if (IsObjectDistributed(&roleAddress))
ObjectAddress *roleAddress = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*roleAddress, AuthIdRelationId, roleOid);
if (IsAnyObjectDistributed(list_make1(roleAddress)))
{
EnsureDependenciesExistOnAllNodes(&roleAddress);
EnsureAllObjectDependenciesExistOnAllNodes(list_make1(roleAddress));
}
}
return NIL;

7
src/backend/distributed/commands/schema.c
View File

@ -259,10 +259,9 @@ FilterDistributedSchemas(List *schemas)
continue;
}
ObjectAddress address = { 0 };
ObjectAddressSet(address, NamespaceRelationId, schemaOid);
if (!IsObjectDistributed(&address))
ObjectAddress *address = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*address, NamespaceRelationId, schemaOid);
if (!IsAnyObjectDistributed(list_make1(address)))
{
continue;
}

87
src/backend/distributed/commands/sequence.c
View File

@ -268,18 +268,16 @@ PreprocessDropSequenceStmt(Node *node, const char *queryString,
Oid seqOid = RangeVarGetRelid(seq, NoLock, stmt->missing_ok);
ObjectAddress sequenceAddress = { 0 };
ObjectAddressSet(sequenceAddress, RelationRelationId, seqOid);
if (!IsObjectDistributed(&sequenceAddress))
ObjectAddress *sequenceAddress = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*sequenceAddress, RelationRelationId, seqOid);
if (!IsAnyObjectDistributed(list_make1(sequenceAddress)))
{
continue;
}
/* collect information for all distributed sequences */
ObjectAddress *addressp = palloc(sizeof(ObjectAddress));
*addressp = sequenceAddress;
distributedSequenceAddresses = lappend(distributedSequenceAddresses, addressp);
distributedSequenceAddresses = lappend(distributedSequenceAddresses,
sequenceAddress);
distributedSequencesList = lappend(distributedSequencesList, objectNameList);
}
@ -334,10 +332,13 @@ PreprocessRenameSequenceStmt(Node *node, const char *queryString, ProcessUtility
RenameStmt *stmt = castNode(RenameStmt, node);
Assert(stmt->renameType == OBJECT_SEQUENCE);
ObjectAddress address = GetObjectAddressFromParseTree((Node *) stmt,
stmt->missing_ok);
List *addresses = GetObjectAddressListFromParseTree((Node *) stmt,
stmt->missing_ok);
if (!ShouldPropagateObject(&address))
/* the code-path only supports a single object */
Assert(list_length(addresses) == 1);
if (!ShouldPropagateAnyObject(addresses))
{
return NIL;
}
@ -395,21 +396,27 @@ PreprocessAlterSequenceStmt(Node *node, const char *queryString,
{
AlterSeqStmt *stmt = castNode(AlterSeqStmt, node);
ObjectAddress address = GetObjectAddressFromParseTree((Node *) stmt,
stmt->missing_ok);
List *addresses = GetObjectAddressListFromParseTree((Node *) stmt,
stmt->missing_ok);
/* the code-path only supports a single object */
Assert(list_length(addresses) == 1);
/* error out if the sequence is distributed */
if (IsObjectDistributed(&address))
if (IsAnyObjectDistributed(addresses))
{
ereport(ERROR, (errmsg(
"Altering a distributed sequence is currently not supported.")));
}
/* We have already asserted that we have exactly 1 address in the addresses. */
ObjectAddress *address = linitial(addresses);
/*
* error out if the sequence is used in a distributed table
* and this is an ALTER SEQUENCE .. AS .. statement
*/
Oid citusTableId = SequenceUsedInDistributedTable(&address);
Oid citusTableId = SequenceUsedInDistributedTable(address);
if (citusTableId != InvalidOid)
{
List *options = stmt->options;
@ -463,6 +470,7 @@ SequenceUsedInDistributedTable(const ObjectAddress *sequenceAddress)
}
}
}
return InvalidOid;
}
@ -498,9 +506,13 @@ PreprocessAlterSequenceSchemaStmt(Node *node, const char *queryString,
AlterObjectSchemaStmt *stmt = castNode(AlterObjectSchemaStmt, node);
Assert(stmt->objectType == OBJECT_SEQUENCE);
ObjectAddress address = GetObjectAddressFromParseTree((Node *) stmt,
stmt->missing_ok);
if (!ShouldPropagateObject(&address))
List *addresses = GetObjectAddressListFromParseTree((Node *) stmt,
stmt->missing_ok);
/* the code-path only supports a single object */
Assert(list_length(addresses) == 1);
if (!ShouldPropagateAnyObject(addresses))
{
return NIL;
}
@ -572,16 +584,19 @@ PostprocessAlterSequenceSchemaStmt(Node *node, const char *queryString)
{
AlterObjectSchemaStmt *stmt = castNode(AlterObjectSchemaStmt, node);
Assert(stmt->objectType == OBJECT_SEQUENCE);
ObjectAddress address = GetObjectAddressFromParseTree((Node *) stmt,
stmt->missing_ok);
List *addresses = GetObjectAddressListFromParseTree((Node *) stmt,
stmt->missing_ok);
if (!ShouldPropagateObject(&address))
/* the code-path only supports a single object */
Assert(list_length(addresses) == 1);
if (!ShouldPropagateAnyObject(addresses))
{
return NIL;
}
/* dependencies have changed (schema) let's ensure they exist */
EnsureDependenciesExistOnAllNodes(&address);
EnsureAllObjectDependenciesExistOnAllNodes(addresses);
return NIL;
}
@ -601,8 +616,12 @@ PreprocessAlterSequenceOwnerStmt(Node *node, const char *queryString,
AlterTableStmt *stmt = castNode(AlterTableStmt, node);
Assert(AlterTableStmtObjType_compat(stmt) == OBJECT_SEQUENCE);
ObjectAddress sequenceAddress = GetObjectAddressFromParseTree((Node *) stmt, false);
if (!ShouldPropagateObject(&sequenceAddress))
List *sequenceAddresses = GetObjectAddressListFromParseTree((Node *) stmt, false);
/* the code-path only supports a single object */
Assert(list_length(sequenceAddresses) == 1);
if (!ShouldPropagateAnyObject(sequenceAddresses))
{
return NIL;
}
@ -649,14 +668,18 @@ PostprocessAlterSequenceOwnerStmt(Node *node, const char *queryString)
AlterTableStmt *stmt = castNode(AlterTableStmt, node);
Assert(AlterTableStmtObjType_compat(stmt) == OBJECT_SEQUENCE);
ObjectAddress sequenceAddress = GetObjectAddressFromParseTree((Node *) stmt, false);
if (!ShouldPropagateObject(&sequenceAddress))
List *sequenceAddresses = GetObjectAddressListFromParseTree((Node *) stmt, false);
/* the code-path only supports a single object */
Assert(list_length(sequenceAddresses) == 1);
if (!ShouldPropagateAnyObject(sequenceAddresses))
{
return NIL;
}
/* dependencies have changed (owner) let's ensure they exist */
EnsureDependenciesExistOnAllNodes(&sequenceAddress);
EnsureAllObjectDependenciesExistOnAllNodes(sequenceAddresses);
return NIL;
}
@ -744,10 +767,10 @@ PostprocessGrantOnSequenceStmt(Node *node, const char *queryString)
RangeVar *sequence = NULL;
foreach_ptr(sequence, distributedSequences)
{
ObjectAddress sequenceAddress = { 0 };
ObjectAddress *sequenceAddress = palloc0(sizeof(ObjectAddress));
Oid sequenceOid = RangeVarGetRelid(sequence, NoLock, false);
ObjectAddressSet(sequenceAddress, RelationRelationId, sequenceOid);
EnsureDependenciesExistOnAllNodes(&sequenceAddress);
ObjectAddressSet(*sequenceAddress, RelationRelationId, sequenceOid);
EnsureAllObjectDependenciesExistOnAllNodes(list_make1(sequenceAddress));
}
return NIL;
}
@ -866,15 +889,15 @@ FilterDistributedSequences(GrantStmt *stmt)
RangeVar *sequenceRangeVar = NULL;
foreach_ptr(sequenceRangeVar, stmt->objects)
{
ObjectAddress sequenceAddress = { 0 };
Oid sequenceOid = RangeVarGetRelid(sequenceRangeVar, NoLock, missing_ok);
ObjectAddressSet(sequenceAddress, RelationRelationId, sequenceOid);
ObjectAddress *sequenceAddress = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*sequenceAddress, RelationRelationId, sequenceOid);
/*
* if this sequence from GRANT .. ON SEQUENCE .. is a distributed
* sequence, add it to the list
*/
if (IsObjectDistributed(&sequenceAddress))
if (IsAnyObjectDistributed(list_make1(sequenceAddress)))
{
grantSequenceList = lappend(grantSequenceList, sequenceRangeVar);
}

21
src/backend/distributed/commands/statistics.c
View File

@ -122,9 +122,12 @@ PostprocessCreateStatisticsStmt(Node *node, const char *queryString)
}
bool missingOk = false;
ObjectAddress objectAddress = GetObjectAddressFromParseTree((Node *) stmt, missingOk);
List *objectAddresses = GetObjectAddressListFromParseTree((Node *) stmt, missingOk);
EnsureDependenciesExistOnAllNodes(&objectAddress);
/* the code-path only supports a single object */
Assert(list_length(objectAddresses) == 1);
EnsureAllObjectDependenciesExistOnAllNodes(objectAddresses);
return NIL;
}
@ -306,9 +309,12 @@ PostprocessAlterStatisticsSchemaStmt(Node *node, const char *queryString)
}
bool missingOk = false;
ObjectAddress objectAddress = GetObjectAddressFromParseTree((Node *) stmt, missingOk);
List *objectAddresses = GetObjectAddressListFromParseTree((Node *) stmt, missingOk);
EnsureDependenciesExistOnAllNodes(&objectAddress);
/* the code-path only supports a single object */
Assert(list_length(objectAddresses) == 1);
EnsureAllObjectDependenciesExistOnAllNodes(objectAddresses);
return NIL;
}
@ -449,10 +455,9 @@ PostprocessAlterStatisticsOwnerStmt(Node *node, const char *queryString)
return NIL;
}
ObjectAddress statisticsAddress = { 0 };
ObjectAddressSet(statisticsAddress, StatisticExtRelationId, statsOid);
EnsureDependenciesExistOnAllNodes(&statisticsAddress);
ObjectAddress *statisticsAddress = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*statisticsAddress, StatisticExtRelationId, statsOid);
EnsureAllObjectDependenciesExistOnAllNodes(list_make1(statisticsAddress));
return NIL;
}

32
src/backend/distributed/commands/table.c
View File

@ -649,13 +649,19 @@ PostprocessAlterTableSchemaStmt(Node *node, const char *queryString)
/*
* We will let Postgres deal with missing_ok
*/
ObjectAddress tableAddress = GetObjectAddressFromParseTree((Node *) stmt, true);
List *tableAddresses = GetObjectAddressListFromParseTree((Node *) stmt, true);
/* the code-path only supports a single object */
Assert(list_length(tableAddress) == 1);
/* We have already asserted that we have exactly 1 address in the addresses. */
ObjectAddress *tableAddress = linitial(tableAddresses);
/*
* Check whether we are dealing with a sequence or view here and route queries
* accordingly to the right processor function.
*/
char relKind = get_rel_relkind(tableAddress.objectId);
char relKind = get_rel_relkind(tableAddress->objectId);
if (relKind == RELKIND_SEQUENCE)
{
stmt->objectType = OBJECT_SEQUENCE;
@ -667,12 +673,12 @@ PostprocessAlterTableSchemaStmt(Node *node, const char *queryString)
return PostprocessAlterViewSchemaStmt((Node *) stmt, queryString);
}
if (!ShouldPropagate() || !IsCitusTable(tableAddress.objectId))
if (!ShouldPropagate() || !IsCitusTable(tableAddress->objectId))
{
return NIL;
}
EnsureDependenciesExistOnAllNodes(&tableAddress);
EnsureAllObjectDependenciesExistOnAllNodes(tableAddresses);
return NIL;
}
@ -1776,9 +1782,15 @@ PreprocessAlterTableSchemaStmt(Node *node, const char *queryString,
return NIL;
}
ObjectAddress address = GetObjectAddressFromParseTree((Node *) stmt,
stmt->missing_ok);
Oid relationId = address.objectId;
List *addresses = GetObjectAddressListFromParseTree((Node *) stmt,
stmt->missing_ok);
/* the code-path only supports a single object */
Assert(list_length(addresses) == 1);
/* We have already asserted that we have exactly 1 address in the addresses. */
ObjectAddress *address = linitial(addresses);
Oid relationId = address->objectId;
/*
* Check whether we are dealing with a sequence or view here and route queries
@ -1990,9 +2002,9 @@ PostprocessAlterTableStmt(AlterTableStmt *alterTableStatement)
EnsureRelationHasCompatibleSequenceTypes(relationId);
/* changing a relation could introduce new dependencies */
ObjectAddress tableAddress = { 0 };
ObjectAddressSet(tableAddress, RelationRelationId, relationId);
EnsureDependenciesExistOnAllNodes(&tableAddress);
ObjectAddress *tableAddress = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*tableAddress, RelationRelationId, relationId);
EnsureAllObjectDependenciesExistOnAllNodes(list_make1(tableAddress));
}
/* for the new sequences coming with this ALTER TABLE statement */

8
src/backend/distributed/commands/trigger.c
View File

@ -224,8 +224,12 @@ PostprocessCreateTriggerStmt(Node *node, const char *queryString)
EnsureCoordinator();
ErrorOutForTriggerIfNotSupported(relationId);
ObjectAddress objectAddress = GetObjectAddressFromParseTree(node, missingOk);
EnsureDependenciesExistOnAllNodes(&objectAddress);
List *objectAddresses = GetObjectAddressListFromParseTree(node, missingOk);
/* the code-path only supports a single object */
Assert(list_length(objectAddresses) == 1);
EnsureAllObjectDependenciesExistOnAllNodes(objectAddresses);
char *triggerName = createTriggerStmt->trigname;
return CitusCreateTriggerCommandDDLJob(relationId, triggerName,

8
src/backend/distributed/commands/type.c
View File

@ -117,8 +117,12 @@ PreprocessRenameTypeAttributeStmt(Node *node, const char *queryString,
Assert(stmt->renameType == OBJECT_ATTRIBUTE);
Assert(stmt->relationType == OBJECT_TYPE);
ObjectAddress typeAddress = GetObjectAddressFromParseTree((Node *) stmt, false);
if (!ShouldPropagateObject(&typeAddress))
List *typeAddresses = GetObjectAddressListFromParseTree((Node *) stmt, false);
/* the code-path only supports a single object */
Assert(list_length(objectAddresses) == 1);
if (!ShouldPropagateAnyObject(typeAddresses))
{
return NIL;
}

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

@ -853,8 +853,12 @@ ProcessUtilityInternal(PlannedStmt *pstmt,
*/
if (ops && ops->markDistributed)
{
ObjectAddress address = GetObjectAddressFromParseTree(parsetree, false);
MarkObjectDistributed(&address);
List *addresses = GetObjectAddressListFromParseTree(parsetree, false);
ObjectAddress *address = NULL;
foreach_ptr(address, addresses)
{
MarkObjectDistributed(address);
}
}
}

87
src/backend/distributed/commands/view.c
View File

@ -94,22 +94,27 @@ PostprocessViewStmt(Node *node, const char *queryString)
return NIL;
}
ObjectAddress viewAddress = GetObjectAddressFromParseTree((Node *) stmt, false);
List *viewAddresses = GetObjectAddressListFromParseTree((Node *) stmt, false);
if (IsObjectAddressOwnedByExtension(&viewAddress, NULL))
/* the code-path only supports a single object */
Assert(list_length(viewAddresses) == 1);
if (IsAnyObjectAddressOwnedByExtension(viewAddresses, NULL))
{
return NIL;
}
/* If the view has any unsupported dependency, create it locally */
if (ErrorOrWarnIfObjectHasUnsupportedDependency(&viewAddress))
if (ErrorOrWarnIfAnyObjectHasUnsupportedDependency(viewAddresses))
{
return NIL;
}
EnsureDependenciesExistOnAllNodes(&viewAddress);
EnsureAllObjectDependenciesExistOnAllNodes(viewAddresses);
char *command = CreateViewDDLCommand(viewAddress.objectId);
/* We have already asserted that we have exactly 1 address in the addresses. */
ObjectAddress *viewAddress = linitial(viewAddresses);
char *command = CreateViewDDLCommand(viewAddress->objectId);
/*
* We'd typically use NodeDDLTaskList() for generating node-level DDL commands,
@ -140,7 +145,7 @@ PostprocessViewStmt(Node *node, const char *queryString)
*
*/
DDLJob *ddlJob = palloc0(sizeof(DDLJob));
ddlJob->targetObjectAddress = viewAddress;
ddlJob->targetObjectAddress = *viewAddress;
ddlJob->metadataSyncCommand = command;
ddlJob->taskList = NIL;
@ -442,10 +447,9 @@ IsViewDistributed(Oid viewOid)
Assert(get_rel_relkind(viewOid) == RELKIND_VIEW ||
get_rel_relkind(viewOid) == RELKIND_MATVIEW);
ObjectAddress viewAddress = { 0 };
ObjectAddressSet(viewAddress, RelationRelationId, viewOid);
return IsObjectDistributed(&viewAddress);
ObjectAddress *viewAddress = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*viewAddress, RelationRelationId, viewOid);
return IsAnyObjectDistributed(list_make1(viewAddress));
}
@ -458,8 +462,12 @@ PreprocessAlterViewStmt(Node *node, const char *queryString, ProcessUtilityConte
{
AlterTableStmt *stmt = castNode(AlterTableStmt, node);
ObjectAddress viewAddress = GetObjectAddressFromParseTree((Node *) stmt, true);
if (!ShouldPropagateObject(&viewAddress))
List *viewAddresses = GetObjectAddressListFromParseTree((Node *) stmt, true);
/* the code-path only supports a single object */
Assert(list_length(viewAddresses) == 1);
if (!ShouldPropagateAnyObject(viewAddresses))
{
return NIL;
}
@ -471,12 +479,15 @@ PreprocessAlterViewStmt(Node *node, const char *queryString, ProcessUtilityConte
/* reconstruct alter statement in a portable fashion */
const char *alterViewStmtSql = DeparseTreeNode((Node *) stmt);
/* We have already asserted that we have exactly 1 address in the addresses. */
ObjectAddress *viewAddress = linitial(viewAddresses);
/*
* To avoid sequential mode, we are using metadata connection. For the
* detailed explanation, please check the comment on PostprocessViewStmt.
*/
DDLJob *ddlJob = palloc0(sizeof(DDLJob));
ddlJob->targetObjectAddress = viewAddress;
ddlJob->targetObjectAddress = *viewAddress;
ddlJob->metadataSyncCommand = alterViewStmtSql;
ddlJob->taskList = NIL;
@ -493,24 +504,28 @@ PostprocessAlterViewStmt(Node *node, const char *queryString)
AlterTableStmt *stmt = castNode(AlterTableStmt, node);
Assert(AlterTableStmtObjType_compat(stmt) == OBJECT_VIEW);
ObjectAddress viewAddress = GetObjectAddressFromParseTree((Node *) stmt, true);
if (!ShouldPropagateObject(&viewAddress))
List *viewAddresses = GetObjectAddressListFromParseTree((Node *) stmt, true);
/* the code-path only supports a single object */
Assert(list_length(viewAddresses) == 1);
if (!ShouldPropagateAnyObject(viewAddresses))
{
return NIL;
}
if (IsObjectAddressOwnedByExtension(&viewAddress, NULL))
if (IsAnyObjectAddressOwnedByExtension(viewAddresses, NULL))
{
return NIL;
}
/* If the view has any unsupported dependency, create it locally */
if (ErrorOrWarnIfObjectHasUnsupportedDependency(&viewAddress))
if (ErrorOrWarnIfAnyObjectHasUnsupportedDependency(viewAddresses))
{
return NIL;
}
EnsureDependenciesExistOnAllNodes(&viewAddress);
EnsureAllObjectDependenciesExistOnAllNodes(viewAddresses);
return NIL;
}
@ -541,8 +556,12 @@ List *
PreprocessRenameViewStmt(Node *node, const char *queryString,
ProcessUtilityContext processUtilityContext)
{
ObjectAddress viewAddress = GetObjectAddressFromParseTree(node, true);
if (!ShouldPropagateObject(&viewAddress))
List *viewAddresses = GetObjectAddressListFromParseTree(node, true);
/* the code-path only supports a single object */
Assert(list_length(viewAddresses) == 1);
if (!ShouldPropagateAnyObject(viewAddresses))
{
return NIL;
}
@ -555,12 +574,15 @@ PreprocessRenameViewStmt(Node *node, const char *queryString,
/* deparse sql*/
const char *renameStmtSql = DeparseTreeNode(node);
/* We have already asserted that we have exactly 1 address in the addresses. */
ObjectAddress *viewAddress = linitial(viewAddresses);
/*
* To avoid sequential mode, we are using metadata connection. For the
* detailed explanation, please check the comment on PostprocessViewStmt.
*/
DDLJob *ddlJob = palloc0(sizeof(DDLJob));
ddlJob->targetObjectAddress = viewAddress;
ddlJob->targetObjectAddress = *viewAddress;
ddlJob->metadataSyncCommand = renameStmtSql;
ddlJob->taskList = NIL;
@ -596,8 +618,12 @@ PreprocessAlterViewSchemaStmt(Node *node, const char *queryString,
{
AlterObjectSchemaStmt *stmt = castNode(AlterObjectSchemaStmt, node);
ObjectAddress viewAddress = GetObjectAddressFromParseTree((Node *) stmt, true);
if (!ShouldPropagateObject(&viewAddress))
List *viewAddresses = GetObjectAddressListFromParseTree((Node *) stmt, true);
/* the code-path only supports a single object */
Assert(list_length(viewAddresses) == 1);
if (!ShouldPropagateAnyObject(viewAddresses))
{
return NIL;
}
@ -608,12 +634,15 @@ PreprocessAlterViewSchemaStmt(Node *node, const char *queryString,
const char *sql = DeparseTreeNode((Node *) stmt);
/* We have already asserted that we have exactly 1 address in the addresses. */
ObjectAddress *viewAddress = linitial(viewAddresses);
/*
* To avoid sequential mode, we are using metadata connection. For the
* detailed explanation, please check the comment on PostprocessViewStmt.
*/
DDLJob *ddlJob = palloc0(sizeof(DDLJob));
ddlJob->targetObjectAddress = viewAddress;
ddlJob->targetObjectAddress = *viewAddress;
ddlJob->metadataSyncCommand = sql;
ddlJob->taskList = NIL;
@ -631,14 +660,18 @@ PostprocessAlterViewSchemaStmt(Node *node, const char *queryString)
{
AlterObjectSchemaStmt *stmt = castNode(AlterObjectSchemaStmt, node);
ObjectAddress viewAddress = GetObjectAddressFromParseTree((Node *) stmt, true);
if (!ShouldPropagateObject(&viewAddress))
List *viewAddresses = GetObjectAddressListFromParseTree((Node *) stmt, true);
/* the code-path only supports a single object */
Assert(list_length(viewAddresses) == 1);
if (!ShouldPropagateAnyObject(viewAddresses))
{
return NIL;
}
/* dependencies have changed (schema) let's ensure they exist */
EnsureDependenciesExistOnAllNodes(&viewAddress);
EnsureAllObjectDependenciesExistOnAllNodes(viewAddresses);
return NIL;
}

20
src/backend/distributed/deparser/objectaddress.c
View File

@ -20,11 +20,11 @@
/*
* GetObjectAddressFromParseTree returns the ObjectAddress of the main target of the parse
* GetObjectAddressListFromParseTree returns the list of ObjectAddress of the main target of the parse
* tree.
*/
ObjectAddress
GetObjectAddressFromParseTree(Node *parseTree, bool missing_ok)
List *
GetObjectAddressListFromParseTree(Node *parseTree, bool missing_ok)
{
const DistributeObjectOps *ops = GetDistributeObjectOps(parseTree);
@ -33,19 +33,7 @@ GetObjectAddressFromParseTree(Node *parseTree, bool missing_ok)
ereport(ERROR, (errmsg("unsupported statement to get object address for")));
}
List *objectAddresses = ops->address(parseTree, missing_ok);
if (list_length(objectAddresses) > 1)
{
ereport(ERROR, (errmsg(
"citus does not support multiple object addresses in GetObjectAddressFromParseTree")));
}
Assert(list_length(objectAddresses) == 1);
ObjectAddress *objectAddress = linitial(objectAddresses);
return *objectAddress;
return ops->address(parseTree, missing_ok);
}

90
src/backend/distributed/metadata/dependency.c
View File

@ -137,6 +137,8 @@ static DependencyDefinition * CreateObjectAddressDependencyDef(Oid classId, Oid
static List * GetTypeConstraintDependencyDefinition(Oid typeId);
static List * CreateObjectAddressDependencyDefList(Oid classId, List *objectIdList);
static ObjectAddress DependencyDefinitionObjectAddress(DependencyDefinition *definition);
static DeferredErrorMessage * DeferErrorIfHasUnsupportedDependency(const ObjectAddress *
objectAddress);
/* forward declarations for functions to interact with the ObjectAddressCollector */
static void InitObjectAddressCollector(ObjectAddressCollector *collector);
@ -176,7 +178,10 @@ static List * ExpandCitusSupportedTypes(ObjectAddressCollector *collector,
static List * GetDependentRoleIdsFDW(Oid FDWOid);
static List * ExpandRolesToGroups(Oid roleid);
static ViewDependencyNode * BuildViewDependencyGraph(Oid relationId, HTAB *nodeMap);
static bool IsObjectAddressOwnedByExtension(const ObjectAddress *target,
ObjectAddress *extensionAddress);
static bool ErrorOrWarnIfObjectHasUnsupportedDependency(const
ObjectAddress *objectAddress);
/*
* GetUniqueDependenciesList takes a list of object addresses and returns a new list
@ -774,8 +779,8 @@ SupportedDependencyByCitus(const ObjectAddress *address)
* object doesn't have any unsupported dependency, else throws a message with proper level
* (except the cluster doesn't have any node) and return true.
*/
bool
ErrorOrWarnIfObjectHasUnsupportedDependency(ObjectAddress *objectAddress)
static bool
ErrorOrWarnIfObjectHasUnsupportedDependency(const ObjectAddress *objectAddress)
{
DeferredErrorMessage *errMsg = DeferErrorIfHasUnsupportedDependency(objectAddress);
if (errMsg != NULL)
@ -805,7 +810,7 @@ ErrorOrWarnIfObjectHasUnsupportedDependency(ObjectAddress *objectAddress)
* is not distributed yet, we can create it locally to not affect user's local
* usage experience.
*/
else if (IsObjectDistributed(objectAddress))
else if (IsAnyObjectDistributed(list_make1((ObjectAddress *) objectAddress)))
{
RaiseDeferredError(errMsg, ERROR);
}
@ -821,11 +826,31 @@ ErrorOrWarnIfObjectHasUnsupportedDependency(ObjectAddress *objectAddress)
}
/*
* ErrorOrWarnIfAnyObjectHasUnsupportedDependency iteratively calls
* ErrorOrWarnIfObjectHasUnsupportedDependency for given addresses.
*/
bool
ErrorOrWarnIfAnyObjectHasUnsupportedDependency(List *objectAddresses)
{
ObjectAddress *objectAddress = NULL;
foreach_ptr(objectAddress, objectAddresses)
{
if (ErrorOrWarnIfObjectHasUnsupportedDependency(objectAddress))
{
return true;
}
}
return false;
}
/*
* DeferErrorIfHasUnsupportedDependency returns deferred error message if the given
* object has any undistributable dependency.
*/
DeferredErrorMessage *
static DeferredErrorMessage *
DeferErrorIfHasUnsupportedDependency(const ObjectAddress *objectAddress)
{
ObjectAddress *undistributableDependency = GetUndistributableDependency(
@ -858,7 +883,7 @@ DeferErrorIfHasUnsupportedDependency(const ObjectAddress *objectAddress)
* Otherwise, callers are expected to throw the error returned from this
* function as a hard one by ignoring the detail part.
*/
if (!IsObjectDistributed(objectAddress))
if (!IsAnyObjectDistributed(list_make1((ObjectAddress *) objectAddress)))
{
appendStringInfo(detailInfo, "\"%s\" will be created only locally",
objectDescription);
@ -873,7 +898,7 @@ DeferErrorIfHasUnsupportedDependency(const ObjectAddress *objectAddress)
objectDescription,
dependencyDescription);
if (IsObjectDistributed(objectAddress))
if (IsAnyObjectDistributed(list_make1((ObjectAddress *) objectAddress)))
{
appendStringInfo(hintInfo,
"Distribute \"%s\" first to modify \"%s\" on worker nodes",
@ -900,6 +925,28 @@ DeferErrorIfHasUnsupportedDependency(const ObjectAddress *objectAddress)
}
/*
* DeferErrorIfAnyObjectHasUnsupportedDependency iteratively calls
* DeferErrorIfHasUnsupportedDependency for given addresses.
*/
DeferredErrorMessage *
DeferErrorIfAnyObjectHasUnsupportedDependency(const List *objectAddresses)
{
DeferredErrorMessage *deferredErrorMessage = NULL;
ObjectAddress *objectAddress = NULL;
foreach_ptr(objectAddress, objectAddresses)
{
deferredErrorMessage = DeferErrorIfHasUnsupportedDependency(objectAddress);
if (deferredErrorMessage)
{
return deferredErrorMessage;
}
}
return NULL;
}
/*
* GetUndistributableDependency checks whether object has any non-distributable
* dependency. If any one found, it will be returned.
@ -936,7 +983,7 @@ GetUndistributableDependency(const ObjectAddress *objectAddress)
/*
* If object is distributed already, ignore it.
*/
if (IsObjectDistributed(dependency))
if (IsAnyObjectDistributed(list_make1(dependency)))
{
continue;
}
@ -1015,7 +1062,7 @@ IsTableOwnedByExtension(Oid relationId)
* If extensionAddress is not set to a NULL pointer the function will write the extension
* address this function depends on into this location.
*/
bool
static bool
IsObjectAddressOwnedByExtension(const ObjectAddress *target,
ObjectAddress *extensionAddress)
{
@ -1055,6 +1102,27 @@ IsObjectAddressOwnedByExtension(const ObjectAddress *target,
}
/*
* IsAnyObjectAddressOwnedByExtension iteratively calls IsObjectAddressOwnedByExtension
* for given addresses to determine if any address is owned by an extension.
*/
bool
IsAnyObjectAddressOwnedByExtension(const List *targets,
ObjectAddress *extensionAddress)
{
ObjectAddress *target = NULL;
foreach_ptr(target, targets)
{
if (IsObjectAddressOwnedByExtension(target, extensionAddress))
{
return true;
}
}
return false;
}
/*
* FollowNewSupportedDependencies applies filters on pg_depend entries to follow all
* objects which should be distributed before the root object can safely be created.
@ -1097,7 +1165,9 @@ FollowNewSupportedDependencies(ObjectAddressCollector *collector,
* If the object is already distributed it is not a `new` object that needs to be
* distributed before we create a dependent object
*/
if (IsObjectDistributed(&address))
ObjectAddress *copyAddress = palloc0(sizeof(ObjectAddress));
*copyAddress = address;
if (IsAnyObjectDistributed(list_make1(copyAddress)))
{
return false;
}

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

@ -53,6 +53,7 @@
static char * CreatePgDistObjectEntryCommand(const ObjectAddress *objectAddress);
static int ExecuteCommandAsSuperuser(char *query, int paramCount, Oid *paramTypes,
Datum *paramValues);
static bool IsObjectDistributed(const ObjectAddress *address);
PG_FUNCTION_INFO_V1(citus_unmark_object_distributed);
PG_FUNCTION_INFO_V1(master_unmark_object_distributed);
@ -240,17 +241,18 @@ ShouldMarkRelationDistributed(Oid relationId)
return false;
}
ObjectAddress relationAddress = { 0 };
ObjectAddressSet(relationAddress, RelationRelationId, relationId);
ObjectAddress *relationAddress = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*relationAddress, RelationRelationId, relationId);
bool pgObject = (relationId < FirstNormalObjectId);
bool isObjectSupported = SupportedDependencyByCitus(&relationAddress);
bool isObjectSupported = SupportedDependencyByCitus(relationAddress);
bool ownedByExtension = IsTableOwnedByExtension(relationId);
bool alreadyDistributed = IsObjectDistributed(&relationAddress);
bool alreadyDistributed = IsObjectDistributed(relationAddress);
bool hasUnsupportedDependency =
DeferErrorIfHasUnsupportedDependency(&relationAddress) != NULL;
DeferErrorIfAnyObjectHasUnsupportedDependency(list_make1(relationAddress)) !=
NULL;
bool hasCircularDependency =
DeferErrorIfCircularDependencyExists(&relationAddress) != NULL;
DeferErrorIfCircularDependencyExists(relationAddress) != NULL;
/*
* pgObject: Citus never marks pg objects as distributed
@ -390,7 +392,7 @@ UnmarkObjectDistributed(const ObjectAddress *address)
* IsObjectDistributed returns if the object addressed is already distributed in the
* cluster. This performs a local indexed lookup in pg_dist_object.
*/
bool
static bool
IsObjectDistributed(const ObjectAddress *address)
{
ScanKeyData key[3];
@ -422,6 +424,26 @@ IsObjectDistributed(const ObjectAddress *address)
}
/*
* IsAnyObjectDistributed iteratively calls IsObjectDistributed for given addresses to
* determine if any object is distributed.
*/
bool
IsAnyObjectDistributed(const List *addresses)
{
ObjectAddress *address = NULL;
foreach_ptr(address, addresses)
{
if (IsObjectDistributed(address))
{
return true;
}
}
return false;
}
/*
* GetDistributedObjectAddressList returns a list of ObjectAddresses that contains all
* distributed objects as marked in pg_dist_object

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

@ -356,10 +356,9 @@ CreateDependingViewsOnWorkers(Oid relationId)
continue;
}
ObjectAddress viewAddress = { 0 };
ObjectAddressSet(viewAddress, RelationRelationId, viewOid);
EnsureDependenciesExistOnAllNodes(&viewAddress);
ObjectAddress *viewAddress = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*viewAddress, RelationRelationId, viewOid);
EnsureAllObjectDependenciesExistOnAllNodes(list_make1(viewAddress));
char *createViewCommand = CreateViewDDLCommand(viewOid);
char *alterViewOwnerCommand = AlterViewOwnerCommand(viewOid);
@ -367,7 +366,7 @@ CreateDependingViewsOnWorkers(Oid relationId)
SendCommandToWorkersWithMetadata(createViewCommand);
SendCommandToWorkersWithMetadata(alterViewOwnerCommand);
MarkObjectDistributed(&viewAddress);
MarkObjectDistributed(viewAddress);
}
SendCommandToWorkersWithMetadata(ENABLE_DDL_PROPAGATION);
@ -603,10 +602,10 @@ ShouldSyncSequenceMetadata(Oid relationId)
return false;
}
ObjectAddress sequenceAddress = { 0 };
ObjectAddressSet(sequenceAddress, RelationRelationId, relationId);
ObjectAddress *sequenceAddress = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*sequenceAddress, RelationRelationId, relationId);
return IsObjectDistributed(&sequenceAddress);
return IsAnyObjectDistributed(list_make1(sequenceAddress));
}

6
src/backend/distributed/operations/create_shards.c
View File

@ -70,7 +70,6 @@ master_create_worker_shards(PG_FUNCTION_ARGS)
text *tableNameText = PG_GETARG_TEXT_P(0);
int32 shardCount = PG_GETARG_INT32(1);
int32 replicationFactor = PG_GETARG_INT32(2);
ObjectAddress tableAddress = { 0 };
Oid distributedTableId = ResolveRelationId(tableNameText, false);
@ -83,8 +82,9 @@ master_create_worker_shards(PG_FUNCTION_ARGS)
* via their own connection and committed immediately so they become visible to all
* sessions creating shards.
*/
ObjectAddressSet(tableAddress, RelationRelationId, distributedTableId);
EnsureDependenciesExistOnAllNodes(&tableAddress);
ObjectAddress *tableAddress = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*tableAddress, RelationRelationId, distributedTableId);
EnsureAllObjectDependenciesExistOnAllNodes(list_make1(tableAddress));
EnsureReferenceTablesExistOnAllNodes();

6
src/backend/distributed/operations/stage_protocol.c
View File

@ -96,7 +96,7 @@ master_create_empty_shard(PG_FUNCTION_ARGS)
text *relationNameText = PG_GETARG_TEXT_P(0);
char *relationName = text_to_cstring(relationNameText);
uint32 attemptableNodeCount = 0;
ObjectAddress tableAddress = { 0 };
ObjectAddress *tableAddress = palloc0(sizeof(ObjectAddress));
uint32 candidateNodeIndex = 0;
List *candidateNodeList = NIL;
@ -115,8 +115,8 @@ master_create_empty_shard(PG_FUNCTION_ARGS)
* via their own connection and committed immediately so they become visible to all
* sessions creating shards.
*/
ObjectAddressSet(tableAddress, RelationRelationId, relationId);
EnsureDependenciesExistOnAllNodes(&tableAddress);
ObjectAddressSet(*tableAddress, RelationRelationId, relationId);
EnsureAllObjectDependenciesExistOnAllNodes(list_make1(tableAddress));
EnsureReferenceTablesExistOnAllNodes();
/* don't allow the table to be dropped */

15
src/backend/distributed/worker/worker_create_or_replace.c
View File

@ -181,8 +181,13 @@ WorkerCreateOrReplaceObject(List *sqlStatements)
* same subject.
*/
Node *parseTree = ParseTreeNode(linitial(sqlStatements));
ObjectAddress address = GetObjectAddressFromParseTree(parseTree, true);
if (ObjectExists(&address))
List *addresses = GetObjectAddressListFromParseTree(parseTree, true);
Assert(list_length(viewAddresses) == 1);
/* We have already asserted that we have exactly 1 address in the addresses. */
ObjectAddress *address = linitial(addresses);
if (ObjectExists(address))
{
/*
* Object with name from statement is already found locally, check if states are
@ -195,7 +200,7 @@ WorkerCreateOrReplaceObject(List *sqlStatements)
* recreate our version of the object. This we can compare to what the coordinator
* sent us. If they match we don't do anything.
*/
List *localSqlStatements = CreateStmtListByObjectAddress(&address);
List *localSqlStatements = CreateStmtListByObjectAddress(address);
if (CompareStringList(sqlStatements, localSqlStatements))
{
/*
@ -208,9 +213,9 @@ WorkerCreateOrReplaceObject(List *sqlStatements)
return false;
}
char *newName = GenerateBackupNameForCollision(&address);
char *newName = GenerateBackupNameForCollision(address);
RenameStmt *renameStmt = CreateRenameStatement(&address, newName);
RenameStmt *renameStmt = CreateRenameStatement(address, newName);
const char *sqlRenameStmt = DeparseTreeNode((Node *) renameStmt);
ProcessUtilityParseTree((Node *) renameStmt, sqlRenameStmt,
PROCESS_UTILITY_QUERY,

18
src/backend/distributed/worker/worker_drop_protocol.c
View File

@ -127,7 +127,8 @@ WorkerDropDistributedTable(Oid relationId)
relation_close(distributedRelation, AccessShareLock);
/* prepare distributedTableObject for dropping the table */
ObjectAddress distributedTableObject = { RelationRelationId, relationId, 0 };
ObjectAddress *distributedTableObject = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*distributedTableObject, RelationRelationId, relationId);
/* Drop dependent sequences from pg_dist_object */
#if PG_VERSION_NUM >= PG_VERSION_13
@ -144,7 +145,7 @@ WorkerDropDistributedTable(Oid relationId)
UnmarkObjectDistributed(&ownedSequenceAddress);
}
UnmarkObjectDistributed(&distributedTableObject);
UnmarkObjectDistributed(distributedTableObject);
/*
* Remove metadata before object's itself to make functions no-op within
@ -177,7 +178,7 @@ WorkerDropDistributedTable(Oid relationId)
* until the user runs DROP EXTENSION. Therefore, we skip dropping the
* table.
*/
if (!IsObjectAddressOwnedByExtension(&distributedTableObject, NULL))
if (!IsAnyObjectAddressOwnedByExtension(list_make1(distributedTableObject), NULL))
{
char *relName = get_rel_name(relationId);
Oid schemaId = get_rel_namespace(relationId);
@ -238,12 +239,9 @@ worker_drop_shell_table(PG_FUNCTION_ARGS)
relation_close(distributedRelation, AccessShareLock);
/* prepare distributedTableObject for dropping the table */
ObjectAddress distributedTableObject = { InvalidOid, InvalidOid, 0 };
distributedTableObject.classId = RelationRelationId;
distributedTableObject.objectId = relationId;
distributedTableObject.objectSubId = 0;
if (IsObjectAddressOwnedByExtension(&distributedTableObject, NULL))
ObjectAddress *distributedTableObject = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*distributedTableObject, RelationRelationId, relationId);
if (IsAnyObjectAddressOwnedByExtension(list_make1(distributedTableObject), NULL))
{
PG_RETURN_VOID();
}
@ -270,7 +268,7 @@ worker_drop_shell_table(PG_FUNCTION_ARGS)
*
* We drop the table with cascade since other tables may be referring to it.
*/
performDeletion(&distributedTableObject, DROP_CASCADE,
performDeletion(distributedTableObject, DROP_CASCADE,
PERFORM_DELETION_INTERNAL);
PG_RETURN_VOID();

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

@ -148,7 +148,7 @@ extern void QualifyAlterTypeOwnerStmt(Node *stmt);
extern char * GetTypeNamespaceNameByNameList(List *names);
extern Oid TypeOidGetNamespaceOid(Oid typeOid);
extern ObjectAddress GetObjectAddressFromParseTree(Node *parseTree, bool missing_ok);
extern List * GetObjectAddressListFromParseTree(Node *parseTree, bool missing_ok);
extern List * RenameAttributeStmtObjectAddress(Node *stmt, bool missing_ok);
/* forward declarations for deparse_view_stmts.c */

7
src/include/distributed/metadata/dependency.h
View File

@ -23,10 +23,9 @@ extern List * GetUniqueDependenciesList(List *objectAddressesList);
extern List * GetDependenciesForObject(const ObjectAddress *target);
extern List * GetAllSupportedDependenciesForObject(const ObjectAddress *target);
extern List * GetAllDependenciesForObject(const ObjectAddress *target);
extern bool ErrorOrWarnIfObjectHasUnsupportedDependency(ObjectAddress *objectAddress);
extern DeferredErrorMessage * DeferErrorIfHasUnsupportedDependency(const
ObjectAddress *
objectAddress);
extern bool ErrorOrWarnIfAnyObjectHasUnsupportedDependency(List *objectAddresses);
extern DeferredErrorMessage * DeferErrorIfAnyObjectHasUnsupportedDependency(const List *
objectAddresses);
extern List * OrderObjectAddressListInDependencyOrder(List *objectAddressList);
extern bool SupportedDependencyByCitus(const ObjectAddress *address);
extern List * GetPgDependTuplesForDependingObjects(Oid targetObjectClassId,

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

@ -20,15 +20,15 @@
extern bool ObjectExists(const ObjectAddress *address);
extern bool CitusExtensionObject(const ObjectAddress *objectAddress);
extern bool IsObjectDistributed(const ObjectAddress *address);
extern bool IsAnyObjectDistributed(const List *addresses);
extern bool ClusterHasDistributedFunctionWithDistArgument(void);
extern void MarkObjectDistributed(const ObjectAddress *distAddress);
extern void MarkObjectDistributedViaSuperUser(const ObjectAddress *distAddress);
extern void MarkObjectDistributedLocally(const ObjectAddress *distAddress);
extern void UnmarkObjectDistributed(const ObjectAddress *address);
extern bool IsTableOwnedByExtension(Oid relationId);
extern bool IsObjectAddressOwnedByExtension(const ObjectAddress *target,
ObjectAddress *extensionAddress);
extern bool IsAnyObjectAddressOwnedByExtension(const List *targets,
ObjectAddress *extensionAddress);
extern ObjectAddress PgGetObjectAddress(char *ttype, ArrayType *namearr,
ArrayType *argsarr);
extern List * GetDistributedObjectAddressList(void);

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

@ -258,15 +258,15 @@ extern void CreateDistributedTable(Oid relationId, char *distributionColumnName,
extern void CreateTruncateTrigger(Oid relationId);
extern TableConversionReturn * UndistributeTable(TableConversionParameters *params);
extern void EnsureDependenciesExistOnAllNodes(const ObjectAddress *target);
extern void EnsureAllObjectDependenciesExistOnAllNodes(const List *targets);
extern DeferredErrorMessage * DeferErrorIfCircularDependencyExists(const
ObjectAddress *
objectAddress);
extern List * GetDistributableDependenciesForObject(const ObjectAddress *target);
extern List * GetDependencyCreateDDLCommands(const ObjectAddress *dependency);
extern List * GetAllDependencyCreateDDLCommands(const List *dependencies);
extern bool ShouldPropagate(void);
extern bool ShouldPropagateCreateInCoordinatedTransction(void);
extern bool ShouldPropagateObject(const ObjectAddress *address);
extern bool ShouldPropagateAnyObject(List *addresses);
extern List * ReplicateAllObjectsToNodeCommandList(const char *nodeName, int nodePort);
/* Remaining metadata utility functions */