mirror of https://github.com/citusdata/citus.git
Move DistributedObjectMetadataSyncCommandList back
parent
4ae87e8057
commit
e107fb7519
|
@ -546,6 +546,95 @@ MetadataCreateCommands(void)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
/*
|
||||||
|
* DistributedObjectMetadataSyncCommandList returns the necessary commands to create
|
||||||
|
* pg_dist_object entries on the new node.
|
||||||
|
*/
|
||||||
|
List *
|
||||||
|
DistributedObjectMetadataSyncCommandList(void)
|
||||||
|
{
|
||||||
|
HeapTuple pgDistObjectTup = NULL;
|
||||||
|
Relation pgDistObjectRel = table_open(DistObjectRelationId(), AccessShareLock);
|
||||||
|
Relation pgDistObjectIndexRel = index_open(DistObjectPrimaryKeyIndexId(),
|
||||||
|
AccessShareLock);
|
||||||
|
TupleDesc pgDistObjectDesc = RelationGetDescr(pgDistObjectRel);
|
||||||
|
|
||||||
|
List *objectAddressList = NIL;
|
||||||
|
List *distArgumentIndexList = NIL;
|
||||||
|
List *colocationIdList = NIL;
|
||||||
|
|
||||||
|
/* It is not strictly necessary to read the tuples in order.
|
||||||
|
* However, it is useful to get consistent behavior, both for regression
|
||||||
|
* tests and also in production systems.
|
||||||
|
*/
|
||||||
|
SysScanDesc pgDistObjectScan = systable_beginscan_ordered(pgDistObjectRel,
|
||||||
|
pgDistObjectIndexRel, NULL,
|
||||||
|
0, NULL);
|
||||||
|
while (HeapTupleIsValid(pgDistObjectTup = systable_getnext_ordered(pgDistObjectScan,
|
||||||
|
ForwardScanDirection)))
|
||||||
|
{
|
||||||
|
Form_pg_dist_object pg_dist_object = (Form_pg_dist_object) GETSTRUCT(
|
||||||
|
pgDistObjectTup);
|
||||||
|
|
||||||
|
ObjectAddress *address = palloc(sizeof(ObjectAddress));
|
||||||
|
|
||||||
|
ObjectAddressSubSet(*address, pg_dist_object->classid, pg_dist_object->objid,
|
||||||
|
pg_dist_object->objsubid);
|
||||||
|
|
||||||
|
bool distributionArgumentIndexIsNull = false;
|
||||||
|
Datum distributionArgumentIndexDatum =
|
||||||
|
heap_getattr(pgDistObjectTup,
|
||||||
|
Anum_pg_dist_object_distribution_argument_index,
|
||||||
|
pgDistObjectDesc,
|
||||||
|
&distributionArgumentIndexIsNull);
|
||||||
|
int32 distributionArgumentIndex = DatumGetInt32(distributionArgumentIndexDatum);
|
||||||
|
|
||||||
|
bool colocationIdIsNull = false;
|
||||||
|
Datum colocationIdDatum =
|
||||||
|
heap_getattr(pgDistObjectTup,
|
||||||
|
Anum_pg_dist_object_colocationid,
|
||||||
|
pgDistObjectDesc,
|
||||||
|
&colocationIdIsNull);
|
||||||
|
int32 colocationId = DatumGetInt32(colocationIdDatum);
|
||||||
|
|
||||||
|
objectAddressList = lappend(objectAddressList, address);
|
||||||
|
|
||||||
|
if (distributionArgumentIndexIsNull)
|
||||||
|
{
|
||||||
|
distArgumentIndexList = lappend_int(distArgumentIndexList,
|
||||||
|
INVALID_DISTRIBUTION_ARGUMENT_INDEX);
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
distArgumentIndexList = lappend_int(distArgumentIndexList,
|
||||||
|
distributionArgumentIndex);
|
||||||
|
}
|
||||||
|
|
||||||
|
if (colocationIdIsNull)
|
||||||
|
{
|
||||||
|
colocationIdList = lappend_int(colocationIdList,
|
||||||
|
INVALID_COLOCATION_ID);
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
colocationIdList = lappend_int(colocationIdList, colocationId);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
systable_endscan_ordered(pgDistObjectScan);
|
||||||
|
index_close(pgDistObjectIndexRel, AccessShareLock);
|
||||||
|
relation_close(pgDistObjectRel, NoLock);
|
||||||
|
|
||||||
|
char *workerMetadataUpdateCommand =
|
||||||
|
MarkObjectsDistributedCreateCommand(objectAddressList,
|
||||||
|
distArgumentIndexList,
|
||||||
|
colocationIdList);
|
||||||
|
List *commandList = list_make1(workerMetadataUpdateCommand);
|
||||||
|
|
||||||
|
return commandList;
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* GetDistributedTableMetadataEvents returns the full set of DDL commands necessary to
|
* GetDistributedTableMetadataEvents returns the full set of DDL commands necessary to
|
||||||
* create the given distributed table metadata on a worker. The list includes setting up
|
* create the given distributed table metadata on a worker. The list includes setting up
|
||||||
|
|
|
@ -92,7 +92,6 @@ typedef struct NodeMetadata
|
||||||
} NodeMetadata;
|
} NodeMetadata;
|
||||||
|
|
||||||
/* local function forward declarations */
|
/* local function forward declarations */
|
||||||
static List * DistributedObjectMetadataSyncCommandList(void);
|
|
||||||
static List * DetachPartitionCommandList(void);
|
static List * DetachPartitionCommandList(void);
|
||||||
static int ActivateNode(char *nodeName, int nodePort);
|
static int ActivateNode(char *nodeName, int nodePort);
|
||||||
static void RemoveNodeFromCluster(char *nodeName, int32 nodePort);
|
static void RemoveNodeFromCluster(char *nodeName, int32 nodePort);
|
||||||
|
@ -732,95 +731,6 @@ SetUpObjectMetadata(WorkerNode *workerNode)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
/*
|
|
||||||
* DistributedObjectMetadataSyncCommandList returns the necessary commands to create
|
|
||||||
* pg_dist_object entries on the new node.
|
|
||||||
*/
|
|
||||||
static List *
|
|
||||||
DistributedObjectMetadataSyncCommandList(void)
|
|
||||||
{
|
|
||||||
HeapTuple pgDistObjectTup = NULL;
|
|
||||||
Relation pgDistObjectRel = table_open(DistObjectRelationId(), AccessShareLock);
|
|
||||||
Relation pgDistObjectIndexRel = index_open(DistObjectPrimaryKeyIndexId(),
|
|
||||||
AccessShareLock);
|
|
||||||
TupleDesc pgDistObjectDesc = RelationGetDescr(pgDistObjectRel);
|
|
||||||
|
|
||||||
List *objectAddressList = NIL;
|
|
||||||
List *distArgumentIndexList = NIL;
|
|
||||||
List *colocationIdList = NIL;
|
|
||||||
|
|
||||||
/* It is not strictly necessary to read the tuples in order.
|
|
||||||
* However, it is useful to get consistent behavior, both for regression
|
|
||||||
* tests and also in production systems.
|
|
||||||
*/
|
|
||||||
SysScanDesc pgDistObjectScan = systable_beginscan_ordered(pgDistObjectRel,
|
|
||||||
pgDistObjectIndexRel, NULL,
|
|
||||||
0, NULL);
|
|
||||||
while (HeapTupleIsValid(pgDistObjectTup = systable_getnext_ordered(pgDistObjectScan,
|
|
||||||
ForwardScanDirection)))
|
|
||||||
{
|
|
||||||
Form_pg_dist_object pg_dist_object = (Form_pg_dist_object) GETSTRUCT(
|
|
||||||
pgDistObjectTup);
|
|
||||||
|
|
||||||
ObjectAddress *address = palloc(sizeof(ObjectAddress));
|
|
||||||
|
|
||||||
ObjectAddressSubSet(*address, pg_dist_object->classid, pg_dist_object->objid,
|
|
||||||
pg_dist_object->objsubid);
|
|
||||||
|
|
||||||
bool distributionArgumentIndexIsNull = false;
|
|
||||||
Datum distributionArgumentIndexDatum =
|
|
||||||
heap_getattr(pgDistObjectTup,
|
|
||||||
Anum_pg_dist_object_distribution_argument_index,
|
|
||||||
pgDistObjectDesc,
|
|
||||||
&distributionArgumentIndexIsNull);
|
|
||||||
int32 distributionArgumentIndex = DatumGetInt32(distributionArgumentIndexDatum);
|
|
||||||
|
|
||||||
bool colocationIdIsNull = false;
|
|
||||||
Datum colocationIdDatum =
|
|
||||||
heap_getattr(pgDistObjectTup,
|
|
||||||
Anum_pg_dist_object_colocationid,
|
|
||||||
pgDistObjectDesc,
|
|
||||||
&colocationIdIsNull);
|
|
||||||
int32 colocationId = DatumGetInt32(colocationIdDatum);
|
|
||||||
|
|
||||||
objectAddressList = lappend(objectAddressList, address);
|
|
||||||
|
|
||||||
if (distributionArgumentIndexIsNull)
|
|
||||||
{
|
|
||||||
distArgumentIndexList = lappend_int(distArgumentIndexList,
|
|
||||||
INVALID_DISTRIBUTION_ARGUMENT_INDEX);
|
|
||||||
}
|
|
||||||
else
|
|
||||||
{
|
|
||||||
distArgumentIndexList = lappend_int(distArgumentIndexList,
|
|
||||||
distributionArgumentIndex);
|
|
||||||
}
|
|
||||||
|
|
||||||
if (colocationIdIsNull)
|
|
||||||
{
|
|
||||||
colocationIdList = lappend_int(colocationIdList,
|
|
||||||
INVALID_COLOCATION_ID);
|
|
||||||
}
|
|
||||||
else
|
|
||||||
{
|
|
||||||
colocationIdList = lappend_int(colocationIdList, colocationId);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
systable_endscan_ordered(pgDistObjectScan);
|
|
||||||
index_close(pgDistObjectIndexRel, AccessShareLock);
|
|
||||||
relation_close(pgDistObjectRel, NoLock);
|
|
||||||
|
|
||||||
char *workerMetadataUpdateCommand =
|
|
||||||
MarkObjectsDistributedCreateCommand(objectAddressList,
|
|
||||||
distArgumentIndexList,
|
|
||||||
colocationIdList);
|
|
||||||
List *commandList = list_make1(workerMetadataUpdateCommand);
|
|
||||||
|
|
||||||
return commandList;
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* UpdatePgDistLocalGroupOnNode updates the pg_dist_local_group on the given node
|
* UpdatePgDistLocalGroupOnNode updates the pg_dist_local_group on the given node
|
||||||
*/
|
*/
|
||||||
|
|
|
@ -167,7 +167,6 @@ worker_drop_distributed_table_only(PG_FUNCTION_ARGS)
|
||||||
|
|
||||||
/* first check the relation type */
|
/* first check the relation type */
|
||||||
Relation distributedRelation = relation_open(relationId, AccessShareLock);
|
Relation distributedRelation = relation_open(relationId, AccessShareLock);
|
||||||
char relationKind = distributedRelation->rd_rel->relkind;
|
|
||||||
EnsureRelationKindSupported(relationId);
|
EnsureRelationKindSupported(relationId);
|
||||||
|
|
||||||
/* close the relation since we do not need anymore */
|
/* close the relation since we do not need anymore */
|
||||||
|
@ -194,28 +193,7 @@ worker_drop_distributed_table_only(PG_FUNCTION_ARGS)
|
||||||
UnmarkObjectDistributed(&ownedSequenceAddress);
|
UnmarkObjectDistributed(&ownedSequenceAddress);
|
||||||
}
|
}
|
||||||
|
|
||||||
/* drop the server for the foreign relations */
|
if (!IsObjectAddressOwnedByExtension(&distributedTableObject, NULL))
|
||||||
if (relationKind == RELKIND_FOREIGN_TABLE)
|
|
||||||
{
|
|
||||||
ObjectAddresses *objects = new_object_addresses();
|
|
||||||
ObjectAddress foreignServerObject = { InvalidOid, InvalidOid, 0 };
|
|
||||||
ForeignTable *foreignTable = GetForeignTable(relationId);
|
|
||||||
Oid serverId = foreignTable->serverid;
|
|
||||||
|
|
||||||
/* prepare foreignServerObject for dropping the server */
|
|
||||||
foreignServerObject.classId = ForeignServerRelationId;
|
|
||||||
foreignServerObject.objectId = serverId;
|
|
||||||
foreignServerObject.objectSubId = 0;
|
|
||||||
|
|
||||||
/* add the addresses that are going to be dropped */
|
|
||||||
add_exact_object_address(&distributedTableObject, objects);
|
|
||||||
add_exact_object_address(&foreignServerObject, objects);
|
|
||||||
|
|
||||||
/* drop both the table and the server */
|
|
||||||
performMultipleDeletions(objects, DROP_RESTRICT,
|
|
||||||
PERFORM_DELETION_INTERNAL);
|
|
||||||
}
|
|
||||||
else if (!IsObjectAddressOwnedByExtension(&distributedTableObject, NULL))
|
|
||||||
{
|
{
|
||||||
/*
|
/*
|
||||||
* If the table is owned by an extension, we cannot drop it, nor should we
|
* If the table is owned by an extension, we cannot drop it, nor should we
|
||||||
|
|
|
@ -34,6 +34,7 @@ extern char * LocalGroupIdUpdateCommand(int32 groupId);
|
||||||
extern bool ShouldSyncTableMetadata(Oid relationId);
|
extern bool ShouldSyncTableMetadata(Oid relationId);
|
||||||
extern bool ShouldSyncTableMetadataViaCatalog(Oid relationId);
|
extern bool ShouldSyncTableMetadataViaCatalog(Oid relationId);
|
||||||
extern List * MetadataCreateCommands(void);
|
extern List * MetadataCreateCommands(void);
|
||||||
|
extern List * DistributedObjectMetadataSyncCommandList(void);
|
||||||
extern List * MetadataDropCommands(void);
|
extern List * MetadataDropCommands(void);
|
||||||
extern char * MarkObjectsDistributedCreateCommand(List *addresses,
|
extern char * MarkObjectsDistributedCreateCommand(List *addresses,
|
||||||
List *distributionArgumentIndexes,
|
List *distributionArgumentIndexes,
|
||||||
|
|
Loading…
Reference in New Issue