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

Reindent

users/saawasek/non_blocking_split_integrated
Sameer Awasekar 2022-07-24 15:16:44 +05:30
parent 9073564900
commit f6c11bf09c
2 changed files with 16 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -107,19 +107,20 @@ static Task * CreateTaskForDDLCommandList(List *ddlCommandList, WorkerNode *work
static StringInfo CreateSplitShardReplicationSetupUDF(
List *sourceColocatedShardIntervalList, List *shardGroupSplitIntervalListList,
List *destinationWorkerNodesList);
static void AddDummyShardEntryInMap(uint32 targetNodeId, ShardInterval *shardInterval);
static void DropDummyShards(void);
static void TryDroppingShard(MultiConnection *connection, ShardInterval *shardInterval);
char * CreateTemplateReplicationSlotAndReturnSnapshot(ShardInterval *shardInterval,
WorkerNode *sourceWorkerNode,
MultiConnection **
templateSlotConnection);
static char * CreateTemplateReplicationSlotAndReturnSnapshot(ShardInterval *shardInterval,
WorkerNode *sourceWorkerNode,
MultiConnection **
templateSlotConnection);
static List * ParseReplicationSlotInfoFromResult(PGresult *result);
static List * ExecuteSplitShardReplicationSetupUDF(WorkerNode *sourceWorkerNode,
List *sourceColocatedShardIntervalList,
List *shardGroupSplitIntervalListList,
List *destinationWorkerNodesList);
static List * ParseReplicationSlotInfoFromResult(PGresult *result);
static void AddDummyShardEntryInMap(uint32 targetNodeId, ShardInterval *shardInterval);
static void DropDummyShards(void);
static void TryDroppingShard(MultiConnection *connection, ShardInterval *shardInterval);
/* Customize error message strings based on operation type */
static const char *const SplitOperationName[] =
@ -760,6 +761,10 @@ DoSplitCopy(WorkerNode *sourceShardNode, List *sourceColocatedShardIntervalList,
splitShardIntervalList,
destinationWorkerNodesList);
/*
* TODO(saawasek):1)Potentially refactor query list into a different method.
* 2) Assign Distributed Txn(confirm)?
*/
List *ddlCommandList = NIL;
StringInfo beginTransaction = makeStringInfo();
appendStringInfo(beginTransaction,
@ -1419,7 +1424,8 @@ NonBlockingShardSplit(SplitOperation splitOperation,
TryDropSplitShardsOnFailure(mapOfShardToPlacementCreatedByWorkflow);
/*TODO(saawasek): Add checks to open new connection if sourceConnection is not valid anymore.*/
DropAllShardSplitLeftOvers(sourceConnection, shardSplitHashMapForPublication);
DropAllShardSplitLeftOvers(sourceShardToCopyNode,
shardSplitHashMapForPublication);
DropDummyShards();
@ -1746,7 +1752,6 @@ static List *
ParseReplicationSlotInfoFromResult(PGresult *result)
{
int64 rowCount = PQntuples(result);
int64 colCount = PQnfields(result);
List *replicationSlotInfoList = NIL;
for (int64 rowIndex = 0; rowIndex < rowCount; rowIndex++)

1
src/backend/distributed/shardsplit/shardsplit_logical_replication.c
View File

@ -560,6 +560,7 @@ DropAllShardSplitLeftOvers(WorkerNode *sourceNode, HTAB *shardSplitHashMapForPub
ClaimConnectionExclusively(cleanupConnection);
DropAllShardSplitSubscriptions(cleanupConnection);
DropAllShardSplitUsers(cleanupConnection);
/* Close connection after cleanup */