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

Refactor code from review comments

pull/6029/head
Nitish Upreti 2022-07-08 22:23:38 -07:00
parent 7ef8ea948d
commit a30c791eeb
2 changed files with 43 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/backend/distributed/operations/citus_split_shard_by_split_points.c
View File

@ -77,11 +77,11 @@ LookupSplitMode(Oid shardSplitModeOid)
Datum enumLabelDatum = DirectFunctionCall1(enum_out, shardSplitModeOid); Datum enumLabelDatum = DirectFunctionCall1(enum_out, shardSplitModeOid);
char *enumLabel = DatumGetCString(enumLabelDatum); char *enumLabel = DatumGetCString(enumLabelDatum);
/* Extend with other modes as we support them */
if (strncmp(enumLabel, "blocking", NAMEDATALEN) == 0) if (strncmp(enumLabel, "blocking", NAMEDATALEN) == 0)
{ {
shardSplitMode = BLOCKING_SPLIT; shardSplitMode = BLOCKING_SPLIT;
} }
/* Extend with other modes as we support them */
else else
{ {
ereport(ERROR, (errmsg("Invalid split mode: %s. Expected split mode is blocking.", ereport(ERROR, (errmsg("Invalid split mode: %s. Expected split mode is blocking.",

54
src/backend/distributed/operations/shard_split.c
View File

@ -38,10 +38,14 @@ static void ErrorIfCannotSplitShardExtended(SplitOperation splitOperation,
ShardInterval *shardIntervalToSplit, ShardInterval *shardIntervalToSplit,
List *shardSplitPointsList, List *shardSplitPointsList,
List *nodeIdsForPlacementList); List *nodeIdsForPlacementList);
static void CreateSplitShardsForShardGroup(WorkerNode *sourceShardNode, static void CreateAndCopySplitShardsForShardGroup(WorkerNode *sourceShardNode,
List *sourceColocatedShardIntervalList, List *sourceColocatedShardIntervalList,
List *shardGroupSplitIntervalListList, List *shardGroupSplitIntervalListList,
List *workersForPlacementList);
static void CreateSplitShardsForShardGroup(List *shardGroupSplitIntervalListList,
List *workersForPlacementList); List *workersForPlacementList);
static void CreateAuxiliaryStructuresForShardGroup(List *shardGroupSplitIntervalListList,
List *workersForPlacementList);
static void CreateObjectOnPlacement(List *objectCreationCommandList, static void CreateObjectOnPlacement(List *objectCreationCommandList,
WorkerNode *workerNode); WorkerNode *workerNode);
static List * CreateSplitIntervalsForShardGroup(List *sourceColocatedShardList, static List * CreateSplitIntervalsForShardGroup(List *sourceColocatedShardList,
@ -420,11 +424,11 @@ BlockingShardSplit(SplitOperation splitOperation,
PG_TRY(); PG_TRY();
{ {
/* /*
* Physically create split children, perform split copy and create auxillary structures. * Physically create split children, perform split copy and create auxiliary structures.
* This includes: indexes, replicaIdentity. triggers and statistics. * This includes: indexes, replicaIdentity. triggers and statistics.
* Foreign key constraints are created after Metadata changes (see CreateForeignKeyConstraints). * Foreign key constraints are created after Metadata changes (see CreateForeignKeyConstraints).
*/ */
CreateSplitShardsForShardGroup( CreateAndCopySplitShardsForShardGroup(
sourceShardToCopyNode, sourceShardToCopyNode,
sourceColocatedShardIntervalList, sourceColocatedShardIntervalList,
shardGroupSplitIntervalListList, shardGroupSplitIntervalListList,
@ -465,9 +469,7 @@ BlockingShardSplit(SplitOperation splitOperation,
/* Create ShardGroup split children on a list of corresponding workers. */ /* Create ShardGroup split children on a list of corresponding workers. */
static void static void
CreateSplitShardsForShardGroup(WorkerNode *sourceShardNode, CreateSplitShardsForShardGroup(List *shardGroupSplitIntervalListList,
List *sourceColocatedShardIntervalList,
List *shardGroupSplitIntervalListList,
List *workersForPlacementList) List *workersForPlacementList)
{ {
/* Iterate on shard interval list for shard group */ /* Iterate on shard interval list for shard group */
@ -493,14 +495,20 @@ CreateSplitShardsForShardGroup(WorkerNode *sourceShardNode,
CreateObjectOnPlacement(splitShardCreationCommandList, workerPlacementNode); CreateObjectOnPlacement(splitShardCreationCommandList, workerPlacementNode);
} }
} }
}
/* Perform Split Copy */
DoSplitCopy(sourceShardNode, sourceColocatedShardIntervalList,
shardGroupSplitIntervalListList, workersForPlacementList);
/* Create ShardGroup auxiliary structures (indexes, stats, replicaindentities, triggers)
* on a list of corresponding workers.
*/
static void
CreateAuxiliaryStructuresForShardGroup(List *shardGroupSplitIntervalListList,
List *workersForPlacementList)
{
/* /*
* Create auxillary structures post copy. * Create auxiliary structures post copy.
*/ */
List *shardIntervalList = NULL;
foreach_ptr(shardIntervalList, shardGroupSplitIntervalListList) foreach_ptr(shardIntervalList, shardGroupSplitIntervalListList)
{ {
/* Iterate on split shard interval list and corresponding placement worker */ /* Iterate on split shard interval list and corresponding placement worker */
@ -523,6 +531,28 @@ CreateSplitShardsForShardGroup(WorkerNode *sourceShardNode,
} }
/*
* Create ShardGroup split children, perform copy and create auxiliary structures
* on a list of corresponding workers.
*/
static void
CreateAndCopySplitShardsForShardGroup(WorkerNode *sourceShardNode,
List *sourceColocatedShardIntervalList,
List *shardGroupSplitIntervalListList,
List *workersForPlacementList)
{
CreateSplitShardsForShardGroup(shardGroupSplitIntervalListList,
workersForPlacementList);
DoSplitCopy(sourceShardNode, sourceColocatedShardIntervalList,
shardGroupSplitIntervalListList, workersForPlacementList);
/* Create auxiliary structures (indexes, stats, replicaindentities, triggers) */
CreateAuxiliaryStructuresForShardGroup(shardGroupSplitIntervalListList,
workersForPlacementList);
}
/* /*
* Perform Split Copy from source shard(s) to split children. * Perform Split Copy from source shard(s) to split children.
* 'sourceShardNode' : Source shard worker node. * 'sourceShardNode' : Source shard worker node.