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

Reindent

pull/6029/head
Nitish Upreti 2022-06-27 22:24:23 -07:00
parent 928fa56558
commit 529f1bc8e3
2 changed files with 37 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/backend/distributed/operations/worker_shard_copy.c
View File

@ -231,7 +231,8 @@ ShardCopyDestReceiverStartup(DestReceiver *dest, int operation, TupleDesc
/* define how tuples will be serialised */ /* define how tuples will be serialised */
CopyOutState copyOutState = (CopyOutState) palloc0(sizeof(CopyOutStateData)); CopyOutState copyOutState = (CopyOutState) palloc0(sizeof(CopyOutStateData));
copyOutState->binary = EnableBinaryProtocol && CanUseBinaryCopyFormat(inputTupleDescriptor); copyOutState->binary = EnableBinaryProtocol && CanUseBinaryCopyFormat(
inputTupleDescriptor);
copyOutState->null_print = (char *) nullPrintCharacter; copyOutState->null_print = (char *) nullPrintCharacter;
copyOutState->null_print_client = (char *) nullPrintCharacter; copyOutState->null_print_client = (char *) nullPrintCharacter;
copyOutState->fe_msgbuf = makeStringInfo(); copyOutState->fe_msgbuf = makeStringInfo();

56
src/backend/distributed/operations/worker_split_copy_udf.c
View File

@ -31,12 +31,16 @@ typedef struct SplitCopyInfo
static void ParseSplitCopyInfoDatum(Datum splitCopyInfoDatum, static void ParseSplitCopyInfoDatum(Datum splitCopyInfoDatum,
SplitCopyInfo **splitCopyInfo); SplitCopyInfo **splitCopyInfo);
static DestReceiver** CreateShardCopyDestReceivers(EState *estate, ShardInterval *shardIntervalToSplitCopy, List *splitCopyInfoList); static DestReceiver ** CreateShardCopyDestReceivers(EState *estate,
static DestReceiver* CreatePartitionedSplitCopyDestReceiver( ShardInterval *
EState *executor, shardIntervalToSplitCopy,
ShardInterval *shardIntervalToSplitCopy, List *splitCopyInfoList);
List *splitCopyInfoList); static DestReceiver * CreatePartitionedSplitCopyDestReceiver(EState *executor,
static void BuildMinMaxRangeArrays(List *splitCopyInfoList, ArrayType **minValueArray, ArrayType **maxValueArray); ShardInterval *
shardIntervalToSplitCopy,
List *splitCopyInfoList);
static void BuildMinMaxRangeArrays(List *splitCopyInfoList, ArrayType **minValueArray,
ArrayType **maxValueArray);
/* /*
* worker_split_copy(source_shard_id bigint, splitCopyInfo citus.split_copy_info[]) * worker_split_copy(source_shard_id bigint, splitCopyInfo citus.split_copy_info[])
@ -73,8 +77,8 @@ worker_split_copy(PG_FUNCTION_ARGS)
EState *executor = CreateExecutorState(); EState *executor = CreateExecutorState();
DestReceiver *splitCopyDestReceiver = CreatePartitionedSplitCopyDestReceiver(executor, DestReceiver *splitCopyDestReceiver = CreatePartitionedSplitCopyDestReceiver(executor,
shardIntervalToSplitCopy, shardIntervalToSplitCopy,
splitCopyInfoList); splitCopyInfoList);
char *sourceShardToCopyName = generate_qualified_relation_name( char *sourceShardToCopyName = generate_qualified_relation_name(
shardIntervalToSplitCopy->relationId); shardIntervalToSplitCopy->relationId);
@ -145,7 +149,8 @@ ParseSplitCopyInfoDatum(Datum splitCopyInfoDatum, SplitCopyInfo **splitCopyInfo)
/* Build 'min/max' hash range arrays for PartitionedResultDestReceiver */ /* Build 'min/max' hash range arrays for PartitionedResultDestReceiver */
static void static void
BuildMinMaxRangeArrays(List *splitCopyInfoList, ArrayType **minValueArray, ArrayType **maxValueArray) BuildMinMaxRangeArrays(List *splitCopyInfoList, ArrayType **minValueArray,
ArrayType **maxValueArray)
{ {
int partitionCount = list_length(splitCopyInfoList); int partitionCount = list_length(splitCopyInfoList);
@ -167,8 +172,10 @@ BuildMinMaxRangeArrays(List *splitCopyInfoList, ArrayType **minValueArray, Array
index++; index++;
} }
*minValueArray = CreateArrayFromDatums(minValues, minValueNulls, partitionCount, TEXTOID); *minValueArray = CreateArrayFromDatums(minValues, minValueNulls, partitionCount,
*maxValueArray = CreateArrayFromDatums(maxValues, maxValueNulls, partitionCount, TEXTOID); TEXTOID);
*maxValueArray = CreateArrayFromDatums(maxValues, maxValueNulls, partitionCount,
TEXTOID);
} }
@ -177,10 +184,12 @@ BuildMinMaxRangeArrays(List *splitCopyInfoList, ArrayType **minValueArray, Array
* Each ShardCopyDestReceivers will be responsible for copying tuples from source shard, * Each ShardCopyDestReceivers will be responsible for copying tuples from source shard,
* that fall under its min/max range, to specified destination shard. * that fall under its min/max range, to specified destination shard.
*/ */
static DestReceiver** static DestReceiver **
CreateShardCopyDestReceivers(EState *estate, ShardInterval *shardIntervalToSplitCopy, List *splitCopyInfoList) CreateShardCopyDestReceivers(EState *estate, ShardInterval *shardIntervalToSplitCopy,
List *splitCopyInfoList)
{ {
DestReceiver **shardCopyDests = palloc0(splitCopyInfoList->length * sizeof(DestReceiver *)); DestReceiver **shardCopyDests = palloc0(splitCopyInfoList->length *
sizeof(DestReceiver *));
SplitCopyInfo *splitCopyInfo = NULL; SplitCopyInfo *splitCopyInfo = NULL;
int index = 0; int index = 0;
@ -188,7 +197,8 @@ CreateShardCopyDestReceivers(EState *estate, ShardInterval *shardIntervalToSplit
foreach_ptr(splitCopyInfo, splitCopyInfoList) foreach_ptr(splitCopyInfo, splitCopyInfoList)
{ {
char *destinationShardSchemaName = get_namespace_name(get_rel_namespace( char *destinationShardSchemaName = get_namespace_name(get_rel_namespace(
shardIntervalToSplitCopy->relationId)); shardIntervalToSplitCopy
->relationId));
char *destinationShardNameCopy = pstrdup(sourceShardNamePrefix); char *destinationShardNameCopy = pstrdup(sourceShardNamePrefix);
AppendShardIdToName(&destinationShardNameCopy, splitCopyInfo->destinationShardId); AppendShardIdToName(&destinationShardNameCopy, splitCopyInfo->destinationShardId);
@ -206,11 +216,13 @@ CreateShardCopyDestReceivers(EState *estate, ShardInterval *shardIntervalToSplit
/* Create PartitionedSplitCopyDestReceiver along with underlying ShardCopyDestReceivers */ /* Create PartitionedSplitCopyDestReceiver along with underlying ShardCopyDestReceivers */
static DestReceiver* static DestReceiver *
CreatePartitionedSplitCopyDestReceiver(EState *estate, ShardInterval *shardIntervalToSplitCopy, List *splitCopyInfoList) CreatePartitionedSplitCopyDestReceiver(EState *estate,
ShardInterval *shardIntervalToSplitCopy,
List *splitCopyInfoList)
{ {
/* Create underlying ShardCopyDestReceivers */ /* Create underlying ShardCopyDestReceivers */
DestReceiver** shardCopyDestReceivers = CreateShardCopyDestReceivers( DestReceiver **shardCopyDestReceivers = CreateShardCopyDestReceivers(
estate, estate,
shardIntervalToSplitCopy, shardIntervalToSplitCopy,
splitCopyInfoList); splitCopyInfoList);
@ -219,9 +231,11 @@ CreatePartitionedSplitCopyDestReceiver(EState *estate, ShardInterval *shardInter
ArrayType *minValuesArray = NULL; ArrayType *minValuesArray = NULL;
ArrayType *maxValuesArray = NULL; ArrayType *maxValuesArray = NULL;
BuildMinMaxRangeArrays(splitCopyInfoList, &minValuesArray, &maxValuesArray); BuildMinMaxRangeArrays(splitCopyInfoList, &minValuesArray, &maxValuesArray);
char partitionMethod = PartitionMethodViaCatalog(shardIntervalToSplitCopy->relationId); char partitionMethod = PartitionMethodViaCatalog(
CitusTableCacheEntry *cacheEntry = GetCitusTableCacheEntry(shardIntervalToSplitCopy->relationId); shardIntervalToSplitCopy->relationId);
Var* partitionColumn = cacheEntry->partitionColumn; CitusTableCacheEntry *cacheEntry = GetCitusTableCacheEntry(
shardIntervalToSplitCopy->relationId);
Var *partitionColumn = cacheEntry->partitionColumn;
CitusTableCacheEntry *shardSearchInfo = CitusTableCacheEntry *shardSearchInfo =
QueryTupleShardSearchInfo(minValuesArray, maxValuesArray, QueryTupleShardSearchInfo(minValuesArray, maxValuesArray,