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

Add distributioncolumncollation to to pg_dist_colocation 

Use partition column's collation for range distributed tables
Don't allow non deterministic collations for hash distributed tables
CoPartitionedTables: don't compare unequal types
pull/2907/head
Philip Dubé 2019-08-23 17:32:56 +00:00
parent 91f8be76e1
commit fcf2fd819b
25 changed files with 527 additions and 313 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,6 +1,6 @@
/*------------------------------------------------------------------------- /*-------------------------------------------------------------------------
* *
* create_distributed_relation.c * create_distributed_table.c
* Routines relation to the creation of distributed relations. * Routines relation to the creation of distributed relations.
* *
* Copyright (c) Citus Data, Inc. * Copyright (c) Citus Data, Inc.
@ -551,21 +551,26 @@ ColocationIdForNewTable(Oid relationId, Var *distributionColumn,
* can be sure that there will no modifications on the colocation table * can be sure that there will no modifications on the colocation table
* until this transaction is committed. * until this transaction is committed.
*/ */
Assert(distributionMethod == DISTRIBUTE_BY_HASH);
Relation pgDistColocation = heap_open(DistColocationRelationId(), ExclusiveLock); Relation pgDistColocation = heap_open(DistColocationRelationId(), ExclusiveLock);
Oid distributionColumnType = distributionColumn->vartype; Oid distributionColumnType = distributionColumn->vartype;
Oid distributionColumnCollation = get_typcollation(distributionColumnType);
bool createdColocationGroup = false; bool createdColocationGroup = false;
if (pg_strncasecmp(colocateWithTableName, "default", NAMEDATALEN) == 0) if (pg_strncasecmp(colocateWithTableName, "default", NAMEDATALEN) == 0)
{ {
/* check for default colocation group */ /* check for default colocation group */
colocationId = ColocationId(ShardCount, ShardReplicationFactor, colocationId = ColocationId(ShardCount, ShardReplicationFactor,
distributionColumnType); distributionColumnType,
distributionColumnCollation);
if (colocationId == INVALID_COLOCATION_ID) if (colocationId == INVALID_COLOCATION_ID)
{ {
colocationId = CreateColocationGroup(ShardCount, ShardReplicationFactor, colocationId = CreateColocationGroup(ShardCount, ShardReplicationFactor,
distributionColumnType); distributionColumnType,
distributionColumnCollation);
createdColocationGroup = true; createdColocationGroup = true;
} }
} }
@ -685,6 +690,16 @@ EnsureRelationCanBeDistributed(Oid relationId, Var *distributionColumn,
errdetail("Partition column types must have a hash function " errdetail("Partition column types must have a hash function "
"defined to use hash partitioning."))); "defined to use hash partitioning.")));
} }
#if PG_VERSION_NUM >= 120000
if (distributionColumn->varcollid != InvalidOid &&
!get_collation_isdeterministic(distributionColumn->varcollid))
{
ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("Hash distributed partition columns may not use "
"a non deterministic collation")));
}
#endif
} }
else if (distributionMethod == DISTRIBUTE_BY_RANGE) else if (distributionMethod == DISTRIBUTE_BY_RANGE)
{ {

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

@ -346,7 +346,8 @@ GetFunctionColocationId(Oid functionOid, char *colocateWithTableName,
{ {
/* check for default colocation group */ /* check for default colocation group */
colocationId = ColocationId(ShardCount, ShardReplicationFactor, colocationId = ColocationId(ShardCount, ShardReplicationFactor,
distributionArgumentOid); distributionArgumentOid, get_typcollation(
distributionArgumentOid));
if (colocationId == INVALID_COLOCATION_ID) if (colocationId == INVALID_COLOCATION_ID)
{ {

4
src/backend/distributed/master/master_split_shards.c
View File

@ -81,8 +81,8 @@ worker_hash(PG_FUNCTION_ARGS)
fmgr_info_copy(hashFunction, &(typeEntry->hash_proc_finfo), CurrentMemoryContext); fmgr_info_copy(hashFunction, &(typeEntry->hash_proc_finfo), CurrentMemoryContext);
/* calculate hash value */ /* calculate hash value */
Datum hashedValueDatum = FunctionCall1Coll(hashFunction, PG_GET_COLLATION(), Datum hashedValueDatum =
valueDatum); FunctionCall1Coll(hashFunction, PG_GET_COLLATION(), valueDatum);
PG_RETURN_INT32(hashedValueDatum); PG_RETURN_INT32(hashedValueDatum);
} }

41
src/backend/distributed/metadata/metadata_cache.c
View File

@ -20,6 +20,7 @@
#include "access/sysattr.h" #include "access/sysattr.h"
#include "catalog/indexing.h" #include "catalog/indexing.h"
#include "catalog/pg_am.h" #include "catalog/pg_am.h"
#include "catalog/pg_collation.h"
#include "catalog/pg_enum.h" #include "catalog/pg_enum.h"
#include "catalog/pg_extension.h" #include "catalog/pg_extension.h"
#include "catalog/pg_namespace.h" #include "catalog/pg_namespace.h"
@ -75,6 +76,7 @@
/* user configuration */ /* user configuration */
int ReadFromSecondaries = USE_SECONDARY_NODES_NEVER; int ReadFromSecondaries = USE_SECONDARY_NODES_NEVER;
/* /*
* ShardCacheEntry represents an entry in the shardId -> ShardInterval cache. * ShardCacheEntry represents an entry in the shardId -> ShardInterval cache.
* To avoid duplicating data and invalidation logic between this cache and the * To avoid duplicating data and invalidation logic between this cache and the
@ -189,10 +191,6 @@ static ShardCacheEntry * LookupShardCacheEntry(int64 shardId);
static DistTableCacheEntry * LookupDistTableCacheEntry(Oid relationId); static DistTableCacheEntry * LookupDistTableCacheEntry(Oid relationId);
static void BuildDistTableCacheEntry(DistTableCacheEntry *cacheEntry); static void BuildDistTableCacheEntry(DistTableCacheEntry *cacheEntry);
static void BuildCachedShardList(DistTableCacheEntry *cacheEntry); static void BuildCachedShardList(DistTableCacheEntry *cacheEntry);
static ShardInterval ** SortShardIntervalArray(ShardInterval **shardIntervalArray,
int shardCount,
FmgrInfo *
shardIntervalSortCompareFunction);
static void PrepareWorkerNodeCache(void); static void PrepareWorkerNodeCache(void);
static bool HasUninitializedShardInterval(ShardInterval **sortedShardIntervalArray, static bool HasUninitializedShardInterval(ShardInterval **sortedShardIntervalArray,
int shardCount); int shardCount);
@ -201,6 +199,7 @@ static char * AvailableExtensionVersion(void);
static char * InstalledExtensionVersion(void); static char * InstalledExtensionVersion(void);
static bool HasOverlappingShardInterval(ShardInterval **shardIntervalArray, static bool HasOverlappingShardInterval(ShardInterval **shardIntervalArray,
int shardIntervalArrayLength, int shardIntervalArrayLength,
Oid shardIntervalCollation,
FmgrInfo *shardIntervalSortCompareFunction); FmgrInfo *shardIntervalSortCompareFunction);
static bool CitusHasBeenLoadedInternal(void); static bool CitusHasBeenLoadedInternal(void);
static void InitializeCaches(void); static void InitializeCaches(void);
@ -1222,6 +1221,8 @@ BuildCachedShardList(DistTableCacheEntry *cacheEntry)
/* sort the interval array */ /* sort the interval array */
sortedShardIntervalArray = SortShardIntervalArray(shardIntervalArray, sortedShardIntervalArray = SortShardIntervalArray(shardIntervalArray,
shardIntervalArrayLength, shardIntervalArrayLength,
cacheEntry->partitionColumn->
varcollid,
shardIntervalCompareFunction); shardIntervalCompareFunction);
/* check if there exists any shard intervals with no min/max values */ /* check if there exists any shard intervals with no min/max values */
@ -1234,6 +1235,7 @@ BuildCachedShardList(DistTableCacheEntry *cacheEntry)
cacheEntry->hasOverlappingShardInterval = cacheEntry->hasOverlappingShardInterval =
HasOverlappingShardInterval(sortedShardIntervalArray, HasOverlappingShardInterval(sortedShardIntervalArray,
shardIntervalArrayLength, shardIntervalArrayLength,
cacheEntry->partitionColumn->varcollid,
shardIntervalCompareFunction); shardIntervalCompareFunction);
} }
else else
@ -1326,29 +1328,6 @@ BuildCachedShardList(DistTableCacheEntry *cacheEntry)
} }
/*
* SortedShardIntervalArray sorts the input shardIntervalArray. Shard intervals with
* no min/max values are placed at the end of the array.
*/
static ShardInterval **
SortShardIntervalArray(ShardInterval **shardIntervalArray, int shardCount,
FmgrInfo *shardIntervalSortCompareFunction)
{
/* short cut if there are no shard intervals in the array */
if (shardCount == 0)
{
return shardIntervalArray;
}
/* if a shard doesn't have min/max values, it's placed in the end of the array */
qsort_arg(shardIntervalArray, shardCount, sizeof(ShardInterval *),
(qsort_arg_comparator) CompareShardIntervals,
(void *) shardIntervalSortCompareFunction);
return shardIntervalArray;
}
/* /*
* HasUniformHashDistribution determines whether the given list of sorted shards * HasUniformHashDistribution determines whether the given list of sorted shards
* has a uniform hash distribution, as produced by master_create_worker_shards for * has a uniform hash distribution, as produced by master_create_worker_shards for
@ -1428,6 +1407,7 @@ HasUninitializedShardInterval(ShardInterval **sortedShardIntervalArray, int shar
static bool static bool
HasOverlappingShardInterval(ShardInterval **shardIntervalArray, HasOverlappingShardInterval(ShardInterval **shardIntervalArray,
int shardIntervalArrayLength, int shardIntervalArrayLength,
Oid shardIntervalCollation,
FmgrInfo *shardIntervalSortCompareFunction) FmgrInfo *shardIntervalSortCompareFunction)
{ {
Datum comparisonDatum = 0; Datum comparisonDatum = 0;
@ -1448,9 +1428,10 @@ HasOverlappingShardInterval(ShardInterval **shardIntervalArray,
Assert(lastShardInterval->minValueExists && lastShardInterval->maxValueExists); Assert(lastShardInterval->minValueExists && lastShardInterval->maxValueExists);
Assert(curShardInterval->minValueExists && curShardInterval->maxValueExists); Assert(curShardInterval->minValueExists && curShardInterval->maxValueExists);
comparisonDatum = CompareCall2(shardIntervalSortCompareFunction, comparisonDatum = FunctionCall2Coll(shardIntervalSortCompareFunction,
lastShardInterval->maxValue, shardIntervalCollation,
curShardInterval->minValue); lastShardInterval->maxValue,
curShardInterval->minValue);
comparisonResult = DatumGetInt32(comparisonDatum); comparisonResult = DatumGetInt32(comparisonDatum);
if (comparisonResult >= 0) if (comparisonResult >= 0)

64
src/backend/distributed/planner/multi_physical_planner.c
View File

@ -144,6 +144,7 @@ static Task * QueryPushdownTaskCreate(Query *originalQuery, int shardIndex,
TaskType taskType, TaskType taskType,
bool modifyRequiresMasterEvaluation); bool modifyRequiresMasterEvaluation);
static bool ShardIntervalsEqual(FmgrInfo *comparisonFunction, static bool ShardIntervalsEqual(FmgrInfo *comparisonFunction,
Oid collation,
ShardInterval *firstInterval, ShardInterval *firstInterval,
ShardInterval *secondInterval); ShardInterval *secondInterval);
static List * SqlTaskList(Job *job); static List * SqlTaskList(Job *job);
@ -2469,9 +2470,9 @@ QueryPushdownTaskCreate(Query *originalQuery, int shardIndex,
bool bool
CoPartitionedTables(Oid firstRelationId, Oid secondRelationId) CoPartitionedTables(Oid firstRelationId, Oid secondRelationId)
{ {
bool coPartitionedTables = true;
DistTableCacheEntry *firstTableCache = DistributedTableCacheEntry(firstRelationId); DistTableCacheEntry *firstTableCache = DistributedTableCacheEntry(firstRelationId);
DistTableCacheEntry *secondTableCache = DistributedTableCacheEntry(secondRelationId); DistTableCacheEntry *secondTableCache = DistributedTableCacheEntry(secondRelationId);
ShardInterval **sortedFirstIntervalArray = firstTableCache->sortedShardIntervalArray; ShardInterval **sortedFirstIntervalArray = firstTableCache->sortedShardIntervalArray;
ShardInterval **sortedSecondIntervalArray = ShardInterval **sortedSecondIntervalArray =
secondTableCache->sortedShardIntervalArray; secondTableCache->sortedShardIntervalArray;
@ -2479,6 +2480,18 @@ CoPartitionedTables(Oid firstRelationId, Oid secondRelationId)
uint32 secondListShardCount = secondTableCache->shardIntervalArrayLength; uint32 secondListShardCount = secondTableCache->shardIntervalArrayLength;
FmgrInfo *comparisonFunction = firstTableCache->shardIntervalCompareFunction; FmgrInfo *comparisonFunction = firstTableCache->shardIntervalCompareFunction;
/* reference tables are always & only copartitioned with reference tables */
if (firstTableCache->partitionMethod == DISTRIBUTE_BY_NONE &&
secondTableCache->partitionMethod == DISTRIBUTE_BY_NONE)
{
return true;
}
else if (firstTableCache->partitionMethod == DISTRIBUTE_BY_NONE ||
secondTableCache->partitionMethod == DISTRIBUTE_BY_NONE)
{
return false;
}
if (firstListShardCount != secondListShardCount) if (firstListShardCount != secondListShardCount)
{ {
return false; return false;
@ -2516,6 +2529,18 @@ CoPartitionedTables(Oid firstRelationId, Oid secondRelationId)
} }
/*
* Don't compare unequal types
*/
Oid collation = firstTableCache->partitionColumn->varcollid;
if (firstTableCache->partitionColumn->vartype !=
secondTableCache->partitionColumn->vartype ||
collation != secondTableCache->partitionColumn->varcollid)
{
return false;
}
/* /*
* If not known to be colocated check if the remaining shards are * If not known to be colocated check if the remaining shards are
* anyway. Do so by comparing the shard interval arrays that are sorted on * anyway. Do so by comparing the shard interval arrays that are sorted on
@ -2529,17 +2554,17 @@ CoPartitionedTables(Oid firstRelationId, Oid secondRelationId)
ShardInterval *secondInterval = sortedSecondIntervalArray[intervalIndex]; ShardInterval *secondInterval = sortedSecondIntervalArray[intervalIndex];
bool shardIntervalsEqual = ShardIntervalsEqual(comparisonFunction, bool shardIntervalsEqual = ShardIntervalsEqual(comparisonFunction,
collation,
firstInterval, firstInterval,
secondInterval); secondInterval);
if (!shardIntervalsEqual || !CoPlacedShardIntervals(firstInterval, if (!shardIntervalsEqual || !CoPlacedShardIntervals(firstInterval,
secondInterval)) secondInterval))
{ {
coPartitionedTables = false; return false;
break;
} }
} }
return coPartitionedTables; return true;
} }
@ -2585,8 +2610,8 @@ CoPlacedShardIntervals(ShardInterval *firstInterval, ShardInterval *secondInterv
* ShardIntervalsEqual checks if given shard intervals have equal min/max values. * ShardIntervalsEqual checks if given shard intervals have equal min/max values.
*/ */
static bool static bool
ShardIntervalsEqual(FmgrInfo *comparisonFunction, ShardInterval *firstInterval, ShardIntervalsEqual(FmgrInfo *comparisonFunction, Oid collation,
ShardInterval *secondInterval) ShardInterval *firstInterval, ShardInterval *secondInterval)
{ {
bool shardIntervalsEqual = false; bool shardIntervalsEqual = false;
@ -2598,8 +2623,10 @@ ShardIntervalsEqual(FmgrInfo *comparisonFunction, ShardInterval *firstInterval,
if (firstInterval->minValueExists && firstInterval->maxValueExists && if (firstInterval->minValueExists && firstInterval->maxValueExists &&
secondInterval->minValueExists && secondInterval->maxValueExists) secondInterval->minValueExists && secondInterval->maxValueExists)
{ {
Datum minDatum = CompareCall2(comparisonFunction, firstMin, secondMin); Datum minDatum = FunctionCall2Coll(comparisonFunction, collation, firstMin,
Datum maxDatum = CompareCall2(comparisonFunction, firstMax, secondMax); secondMin);
Datum maxDatum = FunctionCall2Coll(comparisonFunction, collation, firstMax,
secondMax);
int firstComparison = DatumGetInt32(minDatum); int firstComparison = DatumGetInt32(minDatum);
int secondComparison = DatumGetInt32(maxDatum); int secondComparison = DatumGetInt32(maxDatum);
@ -3691,8 +3718,6 @@ FindRangeTableFragmentsList(List *rangeTableFragmentsList, int tableId)
static bool static bool
JoinPrunable(RangeTableFragment *leftFragment, RangeTableFragment *rightFragment) JoinPrunable(RangeTableFragment *leftFragment, RangeTableFragment *rightFragment)
{ {
bool joinPrunable = false;
/* /*
* If both range tables are remote queries, we then have a hash repartition * If both range tables are remote queries, we then have a hash repartition
* join. In that case, we can just prune away this join if left and right * join. In that case, we can just prune away this join if left and right
@ -3738,10 +3763,10 @@ JoinPrunable(RangeTableFragment *leftFragment, RangeTableFragment *rightFragment
leftString->data, rightString->data))); leftString->data, rightString->data)));
} }
joinPrunable = true; return true;
} }
return joinPrunable; return false;
} }
@ -3774,10 +3799,13 @@ FragmentInterval(RangeTableFragment *fragment)
bool bool
ShardIntervalsOverlap(ShardInterval *firstInterval, ShardInterval *secondInterval) ShardIntervalsOverlap(ShardInterval *firstInterval, ShardInterval *secondInterval)
{ {
bool nonOverlap = false;
DistTableCacheEntry *intervalRelation = DistTableCacheEntry *intervalRelation =
DistributedTableCacheEntry(firstInterval->relationId); DistributedTableCacheEntry(firstInterval->relationId);
Assert(intervalRelation->partitionMethod != DISTRIBUTE_BY_NONE);
FmgrInfo *comparisonFunction = intervalRelation->shardIntervalCompareFunction; FmgrInfo *comparisonFunction = intervalRelation->shardIntervalCompareFunction;
Oid collation = intervalRelation->partitionColumn->varcollid;
Datum firstMin = firstInterval->minValue; Datum firstMin = firstInterval->minValue;
@ -3794,18 +3822,20 @@ ShardIntervalsOverlap(ShardInterval *firstInterval, ShardInterval *secondInterva
if (firstInterval->minValueExists && firstInterval->maxValueExists && if (firstInterval->minValueExists && firstInterval->maxValueExists &&
secondInterval->minValueExists && secondInterval->maxValueExists) secondInterval->minValueExists && secondInterval->maxValueExists)
{ {
Datum firstDatum = CompareCall2(comparisonFunction, firstMax, secondMin); Datum firstDatum = FunctionCall2Coll(comparisonFunction, collation, firstMax,
Datum secondDatum = CompareCall2(comparisonFunction, secondMax, firstMin); secondMin);
Datum secondDatum = FunctionCall2Coll(comparisonFunction, collation, secondMax,
firstMin);
int firstComparison = DatumGetInt32(firstDatum); int firstComparison = DatumGetInt32(firstDatum);
int secondComparison = DatumGetInt32(secondDatum); int secondComparison = DatumGetInt32(secondDatum);
if (firstComparison < 0 || secondComparison < 0) if (firstComparison < 0 || secondComparison < 0)
{ {
nonOverlap = true; return false;
} }
} }
return (!nonOverlap); return true;
} }

15
src/backend/distributed/planner/multi_router_planner.c
View File

@ -17,9 +17,9 @@
#include "access/stratnum.h" #include "access/stratnum.h"
#include "access/xact.h" #include "access/xact.h"
#include "catalog/pg_opfamily.h" #include "catalog/pg_opfamily.h"
#include "distributed/citus_clauses.h"
#include "catalog/pg_type.h" #include "catalog/pg_type.h"
#include "distributed/colocation_utils.h" #include "distributed/colocation_utils.h"
#include "distributed/citus_clauses.h"
#include "distributed/citus_nodes.h" #include "distributed/citus_nodes.h"
#include "distributed/citus_nodefuncs.h" #include "distributed/citus_nodefuncs.h"
#include "distributed/deparse_shard_query.h" #include "distributed/deparse_shard_query.h"
@ -379,7 +379,7 @@ AddShardIntervalRestrictionToSelect(Query *subqery, ShardInterval *shardInterval
TypeCacheEntry *typeEntry = lookup_type_cache(targetPartitionColumnVar->vartype, TypeCacheEntry *typeEntry = lookup_type_cache(targetPartitionColumnVar->vartype,
TYPECACHE_HASH_PROC_FINFO); TYPECACHE_HASH_PROC_FINFO);
/* probable never possible given that the tables are already hash partitioned */ /* probably never possible given that the tables are already hash partitioned */
if (!OidIsValid(typeEntry->hash_proc_finfo.fn_oid)) if (!OidIsValid(typeEntry->hash_proc_finfo.fn_oid))
{ {
ereport(ERROR, (errcode(ERRCODE_UNDEFINED_FUNCTION), ereport(ERROR, (errcode(ERRCODE_UNDEFINED_FUNCTION),
@ -387,7 +387,10 @@ AddShardIntervalRestrictionToSelect(Query *subqery, ShardInterval *shardInterval
format_type_be(targetPartitionColumnVar->vartype)))); format_type_be(targetPartitionColumnVar->vartype))));
} }
/* generate hashfunc(partCol) expression */ /*
* Generate hashfunc(partCol) expression.
* Don't set inputcollid as we don't support non deterministic collations.
*/
FuncExpr *hashFunctionExpr = makeNode(FuncExpr); FuncExpr *hashFunctionExpr = makeNode(FuncExpr);
hashFunctionExpr->funcid = CitusWorkerHashFunctionId(); hashFunctionExpr->funcid = CitusWorkerHashFunctionId();
hashFunctionExpr->args = list_make1(targetPartitionColumnVar); hashFunctionExpr->args = list_make1(targetPartitionColumnVar);
@ -401,8 +404,7 @@ AddShardIntervalRestrictionToSelect(Query *subqery, ShardInterval *shardInterval
InvalidOid, false, InvalidOid, false,
(Expr *) hashFunctionExpr, (Expr *) hashFunctionExpr,
(Expr *) MakeInt4Constant(shardInterval->minValue), (Expr *) MakeInt4Constant(shardInterval->minValue),
targetPartitionColumnVar->varcollid, InvalidOid, InvalidOid);
targetPartitionColumnVar->varcollid);
/* update the operators with correct operator numbers and function ids */ /* update the operators with correct operator numbers and function ids */
greaterThanAndEqualsBoundExpr->opfuncid = greaterThanAndEqualsBoundExpr->opfuncid =
@ -416,8 +418,7 @@ AddShardIntervalRestrictionToSelect(Query *subqery, ShardInterval *shardInterval
InvalidOid, false, InvalidOid, false,
(Expr *) hashFunctionExpr, (Expr *) hashFunctionExpr,
(Expr *) MakeInt4Constant(shardInterval->maxValue), (Expr *) MakeInt4Constant(shardInterval->maxValue),
targetPartitionColumnVar->varcollid, InvalidOid, InvalidOid);
targetPartitionColumnVar->varcollid);
/* update the operators with correct operator numbers and function ids */ /* update the operators with correct operator numbers and function ids */
lessThanAndEqualsBoundExpr->opfuncid = get_opcode(lessThanAndEqualsBoundExpr->opno); lessThanAndEqualsBoundExpr->opfuncid = get_opcode(lessThanAndEqualsBoundExpr->opno);

6
src/backend/distributed/planner/shard_pruning.c
View File

@ -276,7 +276,7 @@ PruneShards(Oid relationId, Index rangeTableId, List *whereClauseList,
InitFunctionCallInfoData(*(FunctionCallInfo) & InitFunctionCallInfoData(*(FunctionCallInfo) &
context.compareIntervalFunctionCall, context.compareIntervalFunctionCall,
cacheEntry->shardIntervalCompareFunction, cacheEntry->shardIntervalCompareFunction,
2, DEFAULT_COLLATION_OID, NULL, NULL); 2, cacheEntry->partitionColumn->varcollid, NULL, NULL);
} }
else else
{ {
@ -290,7 +290,7 @@ PruneShards(Oid relationId, Index rangeTableId, List *whereClauseList,
InitFunctionCallInfoData(*(FunctionCallInfo) & InitFunctionCallInfoData(*(FunctionCallInfo) &
context.compareValueFunctionCall, context.compareValueFunctionCall,
cacheEntry->shardColumnCompareFunction, cacheEntry->shardColumnCompareFunction,
2, DEFAULT_COLLATION_OID, NULL, NULL); 2, cacheEntry->partitionColumn->varcollid, NULL, NULL);
} }
else else
{ {
@ -679,7 +679,7 @@ AddSAOPartitionKeyRestrictionToInstance(ClauseWalkerContext *context,
arrayEqualityOp->inputcollid = arrayOperatorExpression->inputcollid; arrayEqualityOp->inputcollid = arrayOperatorExpression->inputcollid;
arrayEqualityOp->opresulttype = get_func_rettype( arrayEqualityOp->opresulttype = get_func_rettype(
arrayOperatorExpression->opfuncid); arrayOperatorExpression->opfuncid);
arrayEqualityOp->opcollid = DEFAULT_COLLATION_OID; arrayEqualityOp->opcollid = context->partitionColumn->varcollid;
arrayEqualityOp->location = -1; arrayEqualityOp->location = -1;
arrayEqualityOp->args = list_make2(strippedLeftOpExpression, constElement); arrayEqualityOp->args = list_make2(strippedLeftOpExpression, constElement);

11
src/backend/distributed/sql/citus--9.1-1--9.2-1.sql
View File

@ -0,0 +1,11 @@
ALTER TABLE pg_catalog.pg_dist_colocation ADD distributioncolumncollation oid;
UPDATE pg_catalog.pg_dist_colocation dc SET distributioncolumncollation = t.typcollation
FROM pg_catalog.pg_type t WHERE t.oid = dc.distributioncolumntype;
UPDATE pg_catalog.pg_dist_colocation dc SET distributioncolumncollation = 0 WHERE distributioncolumncollation IS NULL;
ALTER TABLE pg_catalog.pg_dist_colocation ALTER COLUMN distributioncolumncollation SET NOT NULL;
DROP INDEX pg_dist_colocation_configuration_index;
-- distributioncolumntype should be listed first so that this index can be used for looking up reference tables' colocation id
CREATE INDEX pg_dist_colocation_configuration_index
ON pg_dist_colocation USING btree(distributioncolumntype, shardcount, replicationfactor, distributioncolumncollation);

64
src/backend/distributed/utils/colocation_utils.c
View File

@ -167,15 +167,18 @@ MarkTablesColocated(Oid sourceRelationId, Oid targetRelationId)
Var *sourceDistributionColumn = DistPartitionKey(sourceRelationId); Var *sourceDistributionColumn = DistPartitionKey(sourceRelationId);
Oid sourceDistributionColumnType = InvalidOid; Oid sourceDistributionColumnType = InvalidOid;
Oid sourceDistributionColumnCollation = InvalidOid;
/* reference tables has NULL distribution column */ /* reference tables has NULL distribution column */
if (sourceDistributionColumn != NULL) if (sourceDistributionColumn != NULL)
{ {
sourceDistributionColumnType = sourceDistributionColumn->vartype; sourceDistributionColumnType = sourceDistributionColumn->vartype;
sourceDistributionColumnCollation = sourceDistributionColumn->varcollid;
} }
sourceColocationId = CreateColocationGroup(shardCount, shardReplicationFactor, sourceColocationId = CreateColocationGroup(shardCount, shardReplicationFactor,
sourceDistributionColumnType); sourceDistributionColumnType,
sourceDistributionColumnCollation);
UpdateRelationColocationGroup(sourceRelationId, sourceColocationId); UpdateRelationColocationGroup(sourceRelationId, sourceColocationId);
} }
@ -417,27 +420,31 @@ CompareShardPlacementsByNode(const void *leftElement, const void *rightElement)
/* /*
* ColocationId searches pg_dist_colocation for shard count, replication factor * ColocationId searches pg_dist_colocation for shard count, replication factor,
* and distribution column type. If a matching entry is found, it returns the * distribution column type, and distribution column collation. If a matching entry
* colocation id, otherwise it returns INVALID_COLOCATION_ID. * is found, it returns the colocation id, otherwise returns INVALID_COLOCATION_ID.
*/ */
uint32 uint32
ColocationId(int shardCount, int replicationFactor, Oid distributionColumnType) ColocationId(int shardCount, int replicationFactor, Oid distributionColumnType, Oid
distributionColumnCollation)
{ {
uint32 colocationId = INVALID_COLOCATION_ID; uint32 colocationId = INVALID_COLOCATION_ID;
const int scanKeyCount = 3; const int scanKeyCount = 4;
ScanKeyData scanKey[3]; ScanKeyData scanKey[4];
bool indexOK = true; bool indexOK = true;
Relation pgDistColocation = heap_open(DistColocationRelationId(), AccessShareLock); Relation pgDistColocation = heap_open(DistColocationRelationId(), AccessShareLock);
/* set scan arguments */ /* set scan arguments */
ScanKeyInit(&scanKey[0], Anum_pg_dist_colocation_shardcount, ScanKeyInit(&scanKey[0], Anum_pg_dist_colocation_distributioncolumntype,
BTEqualStrategyNumber, F_INT4EQ, UInt32GetDatum(shardCount));
ScanKeyInit(&scanKey[1], Anum_pg_dist_colocation_replicationfactor,
BTEqualStrategyNumber, F_INT4EQ, Int32GetDatum(replicationFactor));
ScanKeyInit(&scanKey[2], Anum_pg_dist_colocation_distributioncolumntype,
BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(distributionColumnType)); BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(distributionColumnType));
ScanKeyInit(&scanKey[1], Anum_pg_dist_colocation_shardcount,
BTEqualStrategyNumber, F_INT4EQ, UInt32GetDatum(shardCount));
ScanKeyInit(&scanKey[2], Anum_pg_dist_colocation_replicationfactor,
BTEqualStrategyNumber, F_INT4EQ, Int32GetDatum(replicationFactor));
ScanKeyInit(&scanKey[3], Anum_pg_dist_colocation_distributioncolumncollation,
BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(
distributionColumnCollation));
SysScanDesc scanDescriptor = systable_beginscan(pgDistColocation, SysScanDesc scanDescriptor = systable_beginscan(pgDistColocation,
DistColocationConfigurationIndexId(), DistColocationConfigurationIndexId(),
@ -465,7 +472,8 @@ ColocationId(int shardCount, int replicationFactor, Oid distributionColumnType)
* colocation id. * colocation id.
*/ */
uint32 uint32
CreateColocationGroup(int shardCount, int replicationFactor, Oid distributionColumnType) CreateColocationGroup(int shardCount, int replicationFactor, Oid distributionColumnType,
Oid distributionColumnCollation)
{ {
uint32 colocationId = GetNextColocationId(); uint32 colocationId = GetNextColocationId();
Datum values[Natts_pg_dist_colocation]; Datum values[Natts_pg_dist_colocation];
@ -481,6 +489,8 @@ CreateColocationGroup(int shardCount, int replicationFactor, Oid distributionCol
UInt32GetDatum(replicationFactor); UInt32GetDatum(replicationFactor);
values[Anum_pg_dist_colocation_distributioncolumntype - 1] = values[Anum_pg_dist_colocation_distributioncolumntype - 1] =
ObjectIdGetDatum(distributionColumnType); ObjectIdGetDatum(distributionColumnType);
values[Anum_pg_dist_colocation_distributioncolumncollation - 1] =
ObjectIdGetDatum(distributionColumnCollation);
/* open colocation relation and insert the new tuple */ /* open colocation relation and insert the new tuple */
Relation pgDistColocation = heap_open(DistColocationRelationId(), RowExclusiveLock); Relation pgDistColocation = heap_open(DistColocationRelationId(), RowExclusiveLock);
@ -566,27 +576,23 @@ CheckDistributionColumnType(Oid sourceRelationId, Oid targetRelationId)
{ {
Oid sourceDistributionColumnType = InvalidOid; Oid sourceDistributionColumnType = InvalidOid;
Oid targetDistributionColumnType = InvalidOid; Oid targetDistributionColumnType = InvalidOid;
Oid sourceDistributionColumnCollation = InvalidOid;
Oid targetDistributionColumnCollation = InvalidOid;
/* reference tables have NULL distribution column */ /* reference tables have NULL distribution column */
Var *sourceDistributionColumn = DistPartitionKey(sourceRelationId); Var *sourceDistributionColumn = DistPartitionKey(sourceRelationId);
if (sourceDistributionColumn == NULL) if (sourceDistributionColumn != NULL)
{
sourceDistributionColumnType = InvalidOid;
}
else
{ {
sourceDistributionColumnType = sourceDistributionColumn->vartype; sourceDistributionColumnType = sourceDistributionColumn->vartype;
sourceDistributionColumnCollation = sourceDistributionColumn->varcollid;
} }
/* reference tables have NULL distribution column */ /* reference tables have NULL distribution column */
Var *targetDistributionColumn = DistPartitionKey(targetRelationId); Var *targetDistributionColumn = DistPartitionKey(targetRelationId);
if (targetDistributionColumn == NULL) if (targetDistributionColumn != NULL)
{
targetDistributionColumnType = InvalidOid;
}
else
{ {
targetDistributionColumnType = targetDistributionColumn->vartype; targetDistributionColumnType = targetDistributionColumn->vartype;
targetDistributionColumnCollation = targetDistributionColumn->varcollid;
} }
if (sourceDistributionColumnType != targetDistributionColumnType) if (sourceDistributionColumnType != targetDistributionColumnType)
@ -600,6 +606,18 @@ CheckDistributionColumnType(Oid sourceRelationId, Oid targetRelationId)
"%s and %s.", sourceRelationName, "%s and %s.", sourceRelationName,
targetRelationName))); targetRelationName)));
} }
if (sourceDistributionColumnCollation != targetDistributionColumnCollation)
{
char *sourceRelationName = get_rel_name(sourceRelationId);
char *targetRelationName = get_rel_name(targetRelationId);
ereport(ERROR, (errmsg("cannot colocate tables %s and %s",
sourceRelationName, targetRelationName),
errdetail("Distribution column collations don't match for "
"%s and %s.", sourceRelationName,
targetRelationName)));
}
} }

10
src/backend/distributed/utils/reference_table_utils.c
View File

@ -375,6 +375,7 @@ CreateReferenceTableColocationId()
{ {
int shardCount = 1; int shardCount = 1;
Oid distributionColumnType = InvalidOid; Oid distributionColumnType = InvalidOid;
Oid distributionColumnCollation = InvalidOid;
/* /*
* We don't maintain replication factor of reference tables anymore and * We don't maintain replication factor of reference tables anymore and
@ -383,12 +384,15 @@ CreateReferenceTableColocationId()
int replicationFactor = -1; int replicationFactor = -1;
/* check for existing colocations */ /* check for existing colocations */
uint32 colocationId = ColocationId(shardCount, replicationFactor, uint32 colocationId =
distributionColumnType); ColocationId(shardCount, replicationFactor, distributionColumnType,
distributionColumnCollation);
if (colocationId == INVALID_COLOCATION_ID) if (colocationId == INVALID_COLOCATION_ID)
{ {
colocationId = CreateColocationGroup(shardCount, replicationFactor, colocationId = CreateColocationGroup(shardCount, replicationFactor,
distributionColumnType); distributionColumnType,
distributionColumnCollation);
} }
return colocationId; return colocationId;

50
src/backend/distributed/utils/shardinterval_utils.c
View File

@ -51,6 +51,33 @@ LowestShardIntervalById(List *shardIntervalList)
} }
/*
* SortedShardIntervalArray sorts the input shardIntervalArray. Shard intervals with
* no min/max values are placed at the end of the array.
*/
ShardInterval **
SortShardIntervalArray(ShardInterval **shardIntervalArray, int shardCount,
Oid collation, FmgrInfo *shardIntervalSortCompareFunction)
{
SortShardIntervalContext sortContext = {
.comparisonFunction = shardIntervalSortCompareFunction,
.collation = collation
};
/* short cut if there are no shard intervals in the array */
if (shardCount == 0)
{
return shardIntervalArray;
}
/* if a shard doesn't have min/max values, it's placed in the end of the array */
qsort_arg(shardIntervalArray, shardCount, sizeof(ShardInterval *),
(qsort_arg_comparator) CompareShardIntervals, (void *) &sortContext);
return shardIntervalArray;
}
/* /*
* CompareShardIntervals acts as a helper function to compare two shard intervals * CompareShardIntervals acts as a helper function to compare two shard intervals
* by their minimum values, using the value's type comparison function. * by their minimum values, using the value's type comparison function.
@ -60,7 +87,7 @@ LowestShardIntervalById(List *shardIntervalList)
*/ */
int int
CompareShardIntervals(const void *leftElement, const void *rightElement, CompareShardIntervals(const void *leftElement, const void *rightElement,
FmgrInfo *typeCompareFunction) SortShardIntervalContext *sortContext)
{ {
ShardInterval *leftShardInterval = *((ShardInterval **) leftElement); ShardInterval *leftShardInterval = *((ShardInterval **) leftElement);
ShardInterval *rightShardInterval = *((ShardInterval **) rightElement); ShardInterval *rightShardInterval = *((ShardInterval **) rightElement);
@ -70,7 +97,7 @@ CompareShardIntervals(const void *leftElement, const void *rightElement,
bool rightHasNull = (!rightShardInterval->minValueExists || bool rightHasNull = (!rightShardInterval->minValueExists ||
!rightShardInterval->maxValueExists); !rightShardInterval->maxValueExists);
Assert(typeCompareFunction != NULL); Assert(sortContext->comparisonFunction != NULL);
if (leftHasNull && rightHasNull) if (leftHasNull && rightHasNull)
{ {
@ -89,7 +116,9 @@ CompareShardIntervals(const void *leftElement, const void *rightElement,
/* if both shard interval have min/max values, calculate comparison result */ /* if both shard interval have min/max values, calculate comparison result */
Datum leftDatum = leftShardInterval->minValue; Datum leftDatum = leftShardInterval->minValue;
Datum rightDatum = rightShardInterval->minValue; Datum rightDatum = rightShardInterval->minValue;
Datum comparisonDatum = CompareCall2(typeCompareFunction, leftDatum, rightDatum); Datum comparisonDatum = FunctionCall2Coll(sortContext->comparisonFunction,
sortContext->collation, leftDatum,
rightDatum);
comparisonResult = DatumGetInt32(comparisonDatum); comparisonResult = DatumGetInt32(comparisonDatum);
} }
@ -303,8 +332,10 @@ FindShardIntervalIndex(Datum searchedValue, DistTableCacheEntry *cacheEntry)
{ {
Assert(compareFunction != NULL); Assert(compareFunction != NULL);
Oid shardIntervalCollation = cacheEntry->partitionColumn->varcollid;
shardIndex = SearchCachedShardInterval(searchedValue, shardIntervalCache, shardIndex = SearchCachedShardInterval(searchedValue, shardIntervalCache,
shardCount, compareFunction); shardCount, shardIntervalCollation,
compareFunction);
/* we should always return a valid shard index for hash partitioned tables */ /* we should always return a valid shard index for hash partitioned tables */
if (shardIndex == INVALID_SHARD_INDEX) if (shardIndex == INVALID_SHARD_INDEX)
@ -345,8 +376,10 @@ FindShardIntervalIndex(Datum searchedValue, DistTableCacheEntry *cacheEntry)
{ {
Assert(compareFunction != NULL); Assert(compareFunction != NULL);
Oid shardIntervalCollation = cacheEntry->partitionColumn->varcollid;
shardIndex = SearchCachedShardInterval(searchedValue, shardIntervalCache, shardIndex = SearchCachedShardInterval(searchedValue, shardIntervalCache,
shardCount, compareFunction); shardCount, shardIntervalCollation,
compareFunction);
} }
return shardIndex; return shardIndex;
@ -370,7 +403,8 @@ FindShardIntervalIndex(Datum searchedValue, DistTableCacheEntry *cacheEntry)
*/ */
int int
SearchCachedShardInterval(Datum partitionColumnValue, ShardInterval **shardIntervalCache, SearchCachedShardInterval(Datum partitionColumnValue, ShardInterval **shardIntervalCache,
int shardCount, FmgrInfo *compareFunction) int shardCount, Oid shardIntervalCollation,
FmgrInfo *compareFunction)
{ {
int lowerBoundIndex = 0; int lowerBoundIndex = 0;
int upperBoundIndex = shardCount; int upperBoundIndex = shardCount;
@ -380,7 +414,7 @@ SearchCachedShardInterval(Datum partitionColumnValue, ShardInterval **shardInter
int middleIndex = (lowerBoundIndex + upperBoundIndex) / 2; int middleIndex = (lowerBoundIndex + upperBoundIndex) / 2;
int minValueComparison = FunctionCall2Coll(compareFunction, int minValueComparison = FunctionCall2Coll(compareFunction,
DEFAULT_COLLATION_OID, shardIntervalCollation,
partitionColumnValue, partitionColumnValue,
shardIntervalCache[middleIndex]-> shardIntervalCache[middleIndex]->
minValue); minValue);
@ -392,7 +426,7 @@ SearchCachedShardInterval(Datum partitionColumnValue, ShardInterval **shardInter
} }
int maxValueComparison = FunctionCall2Coll(compareFunction, int maxValueComparison = FunctionCall2Coll(compareFunction,
DEFAULT_COLLATION_OID, shardIntervalCollation,
partitionColumnValue, partitionColumnValue,
shardIntervalCache[middleIndex]-> shardIntervalCache[middleIndex]->
maxValue); maxValue);

53
src/backend/distributed/worker/worker_partition_protocol.c
View File

@ -69,7 +69,8 @@ static void FileOutputStreamWrite(FileOutputStream *file, StringInfo dataToWrite
static void FileOutputStreamFlush(FileOutputStream *file); static void FileOutputStreamFlush(FileOutputStream *file);
static void FilterAndPartitionTable(const char *filterQuery, static void FilterAndPartitionTable(const char *filterQuery,
const char *columnName, Oid columnType, const char *columnName, Oid columnType,
uint32 (*PartitionIdFunction)(Datum, const void *), uint32 (*PartitionIdFunction)(Datum, Oid, const
void *),
const void *partitionIdContext, const void *partitionIdContext,
FileOutputStream *partitionFileArray, FileOutputStream *partitionFileArray,
uint32 fileCount); uint32 fileCount);
@ -78,8 +79,10 @@ static CopyOutState InitRowOutputState(void);
static void ClearRowOutputState(CopyOutState copyState); static void ClearRowOutputState(CopyOutState copyState);
static void OutputBinaryHeaders(FileOutputStream *partitionFileArray, uint32 fileCount); static void OutputBinaryHeaders(FileOutputStream *partitionFileArray, uint32 fileCount);
static void OutputBinaryFooters(FileOutputStream *partitionFileArray, uint32 fileCount); static void OutputBinaryFooters(FileOutputStream *partitionFileArray, uint32 fileCount);
static uint32 RangePartitionId(Datum partitionValue, const void *context); static uint32 RangePartitionId(Datum partitionValue, Oid partitionCollation, const
static uint32 HashPartitionId(Datum partitionValue, const void *context); void *context);
static uint32 HashPartitionId(Datum partitionValue, Oid partitionCollation, const
void *context);
static StringInfo UserPartitionFilename(StringInfo directoryName, uint32 partitionId); static StringInfo UserPartitionFilename(StringInfo directoryName, uint32 partitionId);
static bool FileIsLink(char *filename, struct stat filestat); static bool FileIsLink(char *filename, struct stat filestat);
@ -187,8 +190,6 @@ worker_hash_partition_table(PG_FUNCTION_ARGS)
Datum *hashRangeArray = DeconstructArrayObject(hashRangeObject); Datum *hashRangeArray = DeconstructArrayObject(hashRangeObject);
int32 partitionCount = ArrayObjectCount(hashRangeObject); int32 partitionCount = ArrayObjectCount(hashRangeObject);
uint32 (*hashPartitionIdFunction)(Datum, const void *);
CheckCitusVersion(ERROR); CheckCitusVersion(ERROR);
HashPartitionContext *partitionContext = palloc0(sizeof(HashPartitionContext)); HashPartitionContext *partitionContext = palloc0(sizeof(HashPartitionContext));
@ -198,8 +199,6 @@ worker_hash_partition_table(PG_FUNCTION_ARGS)
HasUniformHashDistribution(partitionContext->syntheticShardIntervalArray, HasUniformHashDistribution(partitionContext->syntheticShardIntervalArray,
partitionCount); partitionCount);
hashPartitionIdFunction = &HashPartitionId;
/* use column's type information to get the hashing function */ /* use column's type information to get the hashing function */
FmgrInfo *hashFunction = GetFunctionInfo(partitionColumnType, HASH_AM_OID, FmgrInfo *hashFunction = GetFunctionInfo(partitionColumnType, HASH_AM_OID,
HASHSTANDARD_PROC); HASHSTANDARD_PROC);
@ -209,7 +208,6 @@ worker_hash_partition_table(PG_FUNCTION_ARGS)
partitionContext->hashFunction = hashFunction; partitionContext->hashFunction = hashFunction;
partitionContext->partitionCount = partitionCount; partitionContext->partitionCount = partitionCount;
partitionContext->collation = PG_GET_COLLATION();
/* we'll use binary search, we need the comparison function */ /* we'll use binary search, we need the comparison function */
if (!partitionContext->hasUniformHashDistribution) if (!partitionContext->hasUniformHashDistribution)
@ -228,7 +226,7 @@ worker_hash_partition_table(PG_FUNCTION_ARGS)
/* call the partitioning function that does the actual work */ /* call the partitioning function that does the actual work */
FilterAndPartitionTable(filterQuery, partitionColumn, partitionColumnType, FilterAndPartitionTable(filterQuery, partitionColumn, partitionColumnType,
hashPartitionIdFunction, (const void *) partitionContext, &HashPartitionId, (const void *) partitionContext,
partitionFileArray, fileCount); partitionFileArray, fileCount);
/* close partition files and atomically rename (commit) them */ /* close partition files and atomically rename (commit) them */
@ -856,7 +854,7 @@ FileOutputStreamFlush(FileOutputStream *file)
static void static void
FilterAndPartitionTable(const char *filterQuery, FilterAndPartitionTable(const char *filterQuery,
const char *partitionColumnName, Oid partitionColumnType, const char *partitionColumnName, Oid partitionColumnType,
uint32 (*PartitionIdFunction)(Datum, const void *), uint32 (*PartitionIdFunction)(Datum, Oid, const void *),
const void *partitionIdContext, const void *partitionIdContext,
FileOutputStream *partitionFileArray, FileOutputStream *partitionFileArray,
uint32 fileCount) uint32 fileCount)
@ -864,6 +862,7 @@ FilterAndPartitionTable(const char *filterQuery,
FmgrInfo *columnOutputFunctions = NULL; FmgrInfo *columnOutputFunctions = NULL;
int partitionColumnIndex = 0; int partitionColumnIndex = 0;
Oid partitionColumnTypeId = InvalidOid; Oid partitionColumnTypeId = InvalidOid;
Oid partitionColumnCollation = InvalidOid;
const char *noPortalName = NULL; const char *noPortalName = NULL;
const bool readOnly = true; const bool readOnly = true;
@ -900,6 +899,9 @@ FilterAndPartitionTable(const char *filterQuery,
partitionColumnIndex = ColumnIndex(rowDescriptor, partitionColumnName); partitionColumnIndex = ColumnIndex(rowDescriptor, partitionColumnName);
partitionColumnTypeId = SPI_gettypeid(rowDescriptor, partitionColumnIndex); partitionColumnTypeId = SPI_gettypeid(rowDescriptor, partitionColumnIndex);
partitionColumnCollation = TupleDescAttr(rowDescriptor, partitionColumnIndex -
1)->attcollation;
if (partitionColumnType != partitionColumnTypeId) if (partitionColumnType != partitionColumnTypeId)
{ {
ereport(ERROR, (errmsg("partition column types %u and %u do not match", ereport(ERROR, (errmsg("partition column types %u and %u do not match",
@ -939,7 +941,9 @@ FilterAndPartitionTable(const char *filterQuery,
*/ */
if (!partitionKeyNull) if (!partitionKeyNull)
{ {
partitionId = (*PartitionIdFunction)(partitionKey, partitionIdContext); partitionId = (*PartitionIdFunction)(partitionKey,
partitionColumnCollation,
partitionIdContext);
if (partitionId == INVALID_SHARD_INDEX) if (partitionId == INVALID_SHARD_INDEX)
{ {
ereport(ERROR, (errmsg("invalid distribution column value"))); ereport(ERROR, (errmsg("invalid distribution column value")));
@ -1145,16 +1149,6 @@ OutputBinaryFooters(FileOutputStream *partitionFileArray, uint32 fileCount)
} }
/* Helper function that invokes a function with the default collation oid. */
Datum
CompareCall2(FmgrInfo *functionInfo, Datum leftArgument, Datum rightArgument)
{
Datum result = FunctionCall2Coll(functionInfo, DEFAULT_COLLATION_OID,
leftArgument, rightArgument);
return result;
}
/* /*
* RangePartitionId determines the partition number for the given data value * RangePartitionId determines the partition number for the given data value
* by applying range partitioning. More specifically, the function takes in a * by applying range partitioning. More specifically, the function takes in a
@ -1167,7 +1161,7 @@ CompareCall2(FmgrInfo *functionInfo, Datum leftArgument, Datum rightArgument)
* full compatibility with the semantics of Hadoop's TotalOrderPartitioner. * full compatibility with the semantics of Hadoop's TotalOrderPartitioner.
*/ */
static uint32 static uint32
RangePartitionId(Datum partitionValue, const void *context) RangePartitionId(Datum partitionValue, Oid partitionCollation, const void *context)
{ {
RangePartitionContext *rangePartitionContext = (RangePartitionContext *) context; RangePartitionContext *rangePartitionContext = (RangePartitionContext *) context;
FmgrInfo *comparisonFunction = rangePartitionContext->comparisonFunction; FmgrInfo *comparisonFunction = rangePartitionContext->comparisonFunction;
@ -1191,8 +1185,9 @@ RangePartitionId(Datum partitionValue, const void *context)
Datum middlePoint = pointArray[middleIndex]; Datum middlePoint = pointArray[middleIndex];
Datum comparisonDatum = CompareCall2(comparisonFunction, partitionValue, Datum comparisonDatum =
middlePoint); FunctionCall2Coll(comparisonFunction, partitionCollation,
partitionValue, middlePoint);
int comparisonResult = DatumGetInt32(comparisonDatum); int comparisonResult = DatumGetInt32(comparisonDatum);
/* if partition value is less than middle point */ /* if partition value is less than middle point */
@ -1217,9 +1212,12 @@ RangePartitionId(Datum partitionValue, const void *context)
* using hash partitioning. More specifically, the function returns zero if the * using hash partitioning. More specifically, the function returns zero if the
* given data value is null. If not, the function follows the exact same approach * given data value is null. If not, the function follows the exact same approach
* as Citus distributed planner uses. * as Citus distributed planner uses.
*
* partitionCollation is unused, as we do not support non deterministic collations
* for hash distributed tables.
*/ */
static uint32 static uint32
HashPartitionId(Datum partitionValue, const void *context) HashPartitionId(Datum partitionValue, Oid partitionCollation, const void *context)
{ {
HashPartitionContext *hashPartitionContext = (HashPartitionContext *) context; HashPartitionContext *hashPartitionContext = (HashPartitionContext *) context;
FmgrInfo *hashFunction = hashPartitionContext->hashFunction; FmgrInfo *hashFunction = hashPartitionContext->hashFunction;
@ -1227,7 +1225,7 @@ HashPartitionId(Datum partitionValue, const void *context)
ShardInterval **syntheticShardIntervalArray = ShardInterval **syntheticShardIntervalArray =
hashPartitionContext->syntheticShardIntervalArray; hashPartitionContext->syntheticShardIntervalArray;
FmgrInfo *comparisonFunction = hashPartitionContext->comparisonFunction; FmgrInfo *comparisonFunction = hashPartitionContext->comparisonFunction;
Datum hashDatum = FunctionCall1Coll(hashFunction, hashPartitionContext->collation, Datum hashDatum = FunctionCall1Coll(hashFunction, DEFAULT_COLLATION_OID,
partitionValue); partitionValue);
int32 hashResult = 0; int32 hashResult = 0;
uint32 hashPartitionId = 0; uint32 hashPartitionId = 0;
@ -1248,7 +1246,8 @@ HashPartitionId(Datum partitionValue, const void *context)
{ {
hashPartitionId = hashPartitionId =
SearchCachedShardInterval(hashDatum, syntheticShardIntervalArray, SearchCachedShardInterval(hashDatum, syntheticShardIntervalArray,
partitionCount, comparisonFunction); partitionCount, InvalidOid,
comparisonFunction);
} }

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

@ -25,9 +25,11 @@ extern List * ColocatedTableList(Oid distributedTableId);
extern List * ColocatedShardIntervalList(ShardInterval *shardInterval); extern List * ColocatedShardIntervalList(ShardInterval *shardInterval);
extern Oid ColocatedTableId(Oid colocationId); extern Oid ColocatedTableId(Oid colocationId);
extern uint64 ColocatedShardIdInRelation(Oid relationId, int shardIndex); extern uint64 ColocatedShardIdInRelation(Oid relationId, int shardIndex);
uint32 ColocationId(int shardCount, int replicationFactor, Oid distributionColumnType); uint32 ColocationId(int shardCount, int replicationFactor, Oid distributionColumnType,
Oid distributionColumnCollation);
extern uint32 CreateColocationGroup(int shardCount, int replicationFactor, extern uint32 CreateColocationGroup(int shardCount, int replicationFactor,
Oid distributionColumnType); Oid distributionColumnType,
Oid distributionColumnCollation);
extern uint32 GetNextColocationId(void); extern uint32 GetNextColocationId(void);
extern void CheckReplicationModel(Oid sourceRelationId, Oid targetRelationId); extern void CheckReplicationModel(Oid sourceRelationId, Oid targetRelationId);
extern void CheckDistributionColumnType(Oid sourceRelationId, Oid targetRelationId); extern void CheckDistributionColumnType(Oid sourceRelationId, Oid targetRelationId);

4
src/include/distributed/pg_dist_colocation.h
View File

@ -22,6 +22,7 @@ typedef struct FormData_pg_dist_colocation
uint32 shardcount; uint32 shardcount;
uint32 replicationfactor; uint32 replicationfactor;
Oid distributioncolumntype; Oid distributioncolumntype;
Oid distributioncolumncollation;
} FormData_pg_dist_colocation; } FormData_pg_dist_colocation;
/* ---------------- /* ----------------
@ -35,11 +36,12 @@ typedef FormData_pg_dist_colocation *Form_pg_dist_colocation;
* compiler constants for pg_dist_colocation * compiler constants for pg_dist_colocation
* ---------------- * ----------------
*/ */
#define Natts_pg_dist_colocation 4 #define Natts_pg_dist_colocation 5
#define Anum_pg_dist_colocation_colocationid 1 #define Anum_pg_dist_colocation_colocationid 1
#define Anum_pg_dist_colocation_shardcount 2 #define Anum_pg_dist_colocation_shardcount 2
#define Anum_pg_dist_colocation_replicationfactor 3 #define Anum_pg_dist_colocation_replicationfactor 3
#define Anum_pg_dist_colocation_distributioncolumntype 4 #define Anum_pg_dist_colocation_distributioncolumntype 4
#define Anum_pg_dist_colocation_distributioncolumncollation 5
#define COLOCATIONID_SEQUENCE_NAME "pg_dist_colocationid_seq" #define COLOCATIONID_SEQUENCE_NAME "pg_dist_colocationid_seq"

17
src/include/distributed/shardinterval_utils.h
View File

@ -26,9 +26,21 @@ typedef struct ShardIntervalCompareFunctionCacheEntry
FmgrInfo *functionInfo; FmgrInfo *functionInfo;
} ShardIntervalCompareFunctionCacheEntry; } ShardIntervalCompareFunctionCacheEntry;
/*
* SortShardIntervalContext is the context parameter in SortShardIntervalArray
*/
typedef struct SortShardIntervalContext
{
FmgrInfo *comparisonFunction;
Oid collation;
} SortShardIntervalContext;
extern ShardInterval ** SortShardIntervalArray(ShardInterval **shardIntervalArray, int
shardCount, Oid collation,
FmgrInfo *shardIntervalSortCompareFunction);
extern ShardInterval * LowestShardIntervalById(List *shardIntervalList); extern ShardInterval * LowestShardIntervalById(List *shardIntervalList);
extern int CompareShardIntervals(const void *leftElement, const void *rightElement, extern int CompareShardIntervals(const void *leftElement, const void *rightElement,
FmgrInfo *typeCompareFunction); SortShardIntervalContext *sortContext);
extern int CompareShardIntervalsById(const void *leftElement, const void *rightElement); extern int CompareShardIntervalsById(const void *leftElement, const void *rightElement);
extern int CompareShardPlacementsByShardId(const void *leftElement, const extern int CompareShardPlacementsByShardId(const void *leftElement, const
void *rightElement); void *rightElement);
@ -40,7 +52,8 @@ extern ShardInterval * FindShardInterval(Datum partitionColumnValue,
extern int FindShardIntervalIndex(Datum searchedValue, DistTableCacheEntry *cacheEntry); extern int FindShardIntervalIndex(Datum searchedValue, DistTableCacheEntry *cacheEntry);
extern int SearchCachedShardInterval(Datum partitionColumnValue, extern int SearchCachedShardInterval(Datum partitionColumnValue,
ShardInterval **shardIntervalCache, ShardInterval **shardIntervalCache,
int shardCount, FmgrInfo *compareFunction); int shardCount, Oid shardIntervalCollation,
FmgrInfo *compareFunction);
extern bool SingleReplicatedTable(Oid relationId); extern bool SingleReplicatedTable(Oid relationId);

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

@ -82,7 +82,6 @@ typedef struct HashPartitionContext
FmgrInfo *comparisonFunction; FmgrInfo *comparisonFunction;
ShardInterval **syntheticShardIntervalArray; ShardInterval **syntheticShardIntervalArray;
uint32 partitionCount; uint32 partitionCount;
Oid collation;
bool hasUniformHashDistribution; bool hasUniformHashDistribution;
} HashPartitionContext; } HashPartitionContext;
@ -137,7 +136,6 @@ extern StringInfo UserTaskFilename(StringInfo directoryName, uint32 taskId);
extern List * ColumnDefinitionList(List *columnNameList, List *columnTypeList); extern List * ColumnDefinitionList(List *columnNameList, List *columnTypeList);
extern CreateStmt * CreateStatement(RangeVar *relation, List *columnDefinitionList); extern CreateStmt * CreateStatement(RangeVar *relation, List *columnDefinitionList);
extern CopyStmt * CopyStatement(RangeVar *relation, char *sourceFilename); extern CopyStmt * CopyStatement(RangeVar *relation, char *sourceFilename);
extern Datum CompareCall2(FmgrInfo *funcInfo, Datum leftArgument, Datum rightArgument);
/* Function declaration for parsing tree node */ /* Function declaration for parsing tree node */
extern Node * ParseTreeNode(const char *ddlCommand); extern Node * ParseTreeNode(const char *ddlCommand);

28
src/test/regress/expected/distributed_collations.out
View File

@ -53,6 +53,34 @@ SELECT * FROM collation_tests.test_propagate WHERE t2 < 'b' COLLATE "C";
2 | Voẞr | Vossr 2 | Voẞr | Vossr
(1 row) (1 row)
-- Test range table with collated distribution column
CREATE TABLE test_range(key text COLLATE german_phonebook, val int);
SELECT create_distributed_table('test_range', 'key', 'range');
create_distributed_table
--------------------------
(1 row)
SELECT master_create_empty_shard('test_range') AS new_shard_id
\gset
UPDATE pg_dist_shard SET shardminvalue = 'a', shardmaxvalue = 'f'
WHERE shardid = :new_shard_id;
SELECT master_create_empty_shard('test_range') AS new_shard_id
\gset
UPDATE pg_dist_shard SET shardminvalue = 'G', shardmaxvalue = 'Z'
WHERE shardid = :new_shard_id;
-- without german_phonebook collation, this would fail
INSERT INTO test_range VALUES (U&'\00E4sop', 1), (U&'Vo\1E9Er', 2);
-- without german_phonebook collation, this would not be router executable
SET client_min_messages TO debug;
SELECT * FROM test_range WHERE key > 'Ab' AND key < U&'\00E4z';
DEBUG: Creating router plan
DEBUG: Plan is router executable
key | val
------+-----
äsop | 1
(1 row)
\c - - - :worker_1_port \c - - - :worker_1_port
SELECT c.collname, nsp.nspname, a.rolname SELECT c.collname, nsp.nspname, a.rolname
FROM pg_collation c FROM pg_collation c

88
src/test/regress/expected/multi_colocation_utils.out
View File

@ -431,13 +431,13 @@ NOTICE: foreign-data wrapper "fake_fdw" does not have an extension defined
SELECT * FROM pg_dist_colocation SELECT * FROM pg_dist_colocation
WHERE colocationid >= 1 AND colocationid < 1000 WHERE colocationid >= 1 AND colocationid < 1000
ORDER BY colocationid; ORDER BY colocationid;
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
3 | 4 | 2 | 23 3 | 4 | 2 | 23 | 0
4 | 2 | 2 | 23 4 | 2 | 2 | 23 | 0
5 | 2 | 1 | 23 5 | 2 | 1 | 23 | 0
6 | 2 | 2 | 25 6 | 2 | 2 | 25 | 100
7 | 8 | 2 | 23 7 | 8 | 2 | 23 | 0
(5 rows) (5 rows)
SELECT logicalrelid, colocationid FROM pg_dist_partition SELECT logicalrelid, colocationid FROM pg_dist_partition
@ -459,17 +459,17 @@ SELECT logicalrelid, colocationid FROM pg_dist_partition
-- check effects of dropping tables -- check effects of dropping tables
DROP TABLE table1_groupA; DROP TABLE table1_groupA;
SELECT * FROM pg_dist_colocation WHERE colocationid = 4; SELECT * FROM pg_dist_colocation WHERE colocationid = 4;
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
4 | 2 | 2 | 23 4 | 2 | 2 | 23 | 0
(1 row) (1 row)
-- dropping all tables in a colocation group also deletes the colocation group -- dropping all tables in a colocation group also deletes the colocation group
DROP TABLE table2_groupA; DROP TABLE table2_groupA;
SELECT * FROM pg_dist_colocation WHERE colocationid = 4; SELECT * FROM pg_dist_colocation WHERE colocationid = 4;
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
4 | 2 | 2 | 23 4 | 2 | 2 | 23 | 0
(1 row) (1 row)
-- create dropped colocation group again -- create dropped colocation group again
@ -555,14 +555,14 @@ SELECT create_distributed_table('table1_group_default', 'id', colocate_with => '
SELECT * FROM pg_dist_colocation SELECT * FROM pg_dist_colocation
WHERE colocationid >= 1 AND colocationid < 1000 WHERE colocationid >= 1 AND colocationid < 1000
ORDER BY colocationid; ORDER BY colocationid;
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
3 | 4 | 2 | 23 3 | 4 | 2 | 23 | 0
4 | 2 | 2 | 23 4 | 2 | 2 | 23 | 0
5 | 2 | 1 | 23 5 | 2 | 1 | 23 | 0
6 | 2 | 2 | 25 6 | 2 | 2 | 25 | 100
7 | 8 | 2 | 23 7 | 8 | 2 | 23 | 0
11 | 3 | 2 | 23 11 | 3 | 2 | 23 | 0
(6 rows) (6 rows)
SELECT logicalrelid, colocationid FROM pg_dist_partition SELECT logicalrelid, colocationid FROM pg_dist_partition
@ -650,14 +650,14 @@ SELECT create_reference_table('table2_groupF');
SELECT * FROM pg_dist_colocation SELECT * FROM pg_dist_colocation
WHERE colocationid >= 1 AND colocationid < 1000 WHERE colocationid >= 1 AND colocationid < 1000
ORDER BY colocationid; ORDER BY colocationid;
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
3 | 4 | 2 | 23 3 | 4 | 2 | 23 | 0
4 | 2 | 2 | 23 4 | 2 | 2 | 23 | 0
5 | 2 | 1 | 23 5 | 2 | 1 | 23 | 0
6 | 2 | 2 | 25 6 | 2 | 2 | 25 | 100
7 | 8 | 2 | 23 7 | 8 | 2 | 23 | 0
11 | 3 | 2 | 23 11 | 3 | 2 | 23 | 0
(6 rows) (6 rows)
-- cross check with internal colocation API -- cross check with internal colocation API
@ -839,8 +839,8 @@ UPDATE pg_dist_partition SET colocationid = 0
SELECT * FROM pg_dist_colocation SELECT * FROM pg_dist_colocation
WHERE colocationid >= 1 AND colocationid < 1000 WHERE colocationid >= 1 AND colocationid < 1000
ORDER BY colocationid; ORDER BY colocationid;
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
(0 rows) (0 rows)
SELECT logicalrelid, colocationid FROM pg_dist_partition SELECT logicalrelid, colocationid FROM pg_dist_partition
@ -870,8 +870,8 @@ DETAIL: Shard counts don't match for table1_groupb and table1_groupd.
SELECT * FROM pg_dist_colocation SELECT * FROM pg_dist_colocation
WHERE colocationid >= 1 AND colocationid < 1000 WHERE colocationid >= 1 AND colocationid < 1000
ORDER BY colocationid; ORDER BY colocationid;
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
(0 rows) (0 rows)
SELECT logicalrelid, colocationid FROM pg_dist_partition SELECT logicalrelid, colocationid FROM pg_dist_partition
@ -938,12 +938,12 @@ SELECT create_distributed_table('table2_group_none', 'id', colocate_with => 'NON
SELECT * FROM pg_dist_colocation SELECT * FROM pg_dist_colocation
WHERE colocationid >= 1 AND colocationid < 1000 WHERE colocationid >= 1 AND colocationid < 1000
ORDER BY colocationid; ORDER BY colocationid;
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
2 | 2 | 1 | 23 2 | 2 | 1 | 23 | 0
3 | 2 | 2 | 25 3 | 2 | 2 | 25 | 100
4 | 8 | 2 | 23 4 | 8 | 2 | 23 | 0
5 | 2 | 2 | 23 5 | 2 | 2 | 23 | 0
(4 rows) (4 rows)
SELECT logicalrelid, colocationid FROM pg_dist_partition SELECT logicalrelid, colocationid FROM pg_dist_partition
@ -982,11 +982,11 @@ SELECT mark_tables_colocated('table1_group_none', ARRAY['table2_group_none']);
SELECT * FROM pg_dist_colocation SELECT * FROM pg_dist_colocation
WHERE colocationid >= 1 AND colocationid < 1000 WHERE colocationid >= 1 AND colocationid < 1000
ORDER BY colocationid; ORDER BY colocationid;
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
2 | 2 | 1 | 23 2 | 2 | 1 | 23 | 0
3 | 2 | 2 | 25 3 | 2 | 2 | 25 | 100
4 | 8 | 2 | 23 4 | 8 | 2 | 23 | 0
(3 rows) (3 rows)
SELECT logicalrelid, colocationid FROM pg_dist_partition SELECT logicalrelid, colocationid FROM pg_dist_partition

4
src/test/regress/expected/multi_metadata_sync.out
View File

@ -295,8 +295,8 @@ SELECT "Column", "Type", "Definition" FROM index_attrs WHERE
-- Check that pg_dist_colocation is not synced -- Check that pg_dist_colocation is not synced
SELECT * FROM pg_dist_colocation ORDER BY colocationid; SELECT * FROM pg_dist_colocation ORDER BY colocationid;
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
(0 rows) (0 rows)
-- Make sure that truncate trigger has been set for the MX table on worker -- Make sure that truncate trigger has been set for the MX table on worker

94
src/test/regress/expected/multi_remove_node_reference_table.out
View File

@ -140,9 +140,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'remove_node_reference_table'::regclass); WHERE logicalrelid = 'remove_node_reference_table'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
\c - - - :worker_1_port \c - - - :worker_1_port
@ -193,9 +193,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'remove_node_reference_table'::regclass); WHERE logicalrelid = 'remove_node_reference_table'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
\c - - - :worker_1_port \c - - - :worker_1_port
@ -273,9 +273,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'remove_node_reference_table'::regclass); WHERE logicalrelid = 'remove_node_reference_table'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
\c - - - :worker_1_port \c - - - :worker_1_port
@ -329,9 +329,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'remove_node_reference_table'::regclass); WHERE logicalrelid = 'remove_node_reference_table'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
\c - - - :worker_1_port \c - - - :worker_1_port
@ -378,9 +378,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'remove_node_reference_table'::regclass); WHERE logicalrelid = 'remove_node_reference_table'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
\c - - - :worker_1_port \c - - - :worker_1_port
@ -433,9 +433,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'remove_node_reference_table'::regclass); WHERE logicalrelid = 'remove_node_reference_table'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
\c - - - :worker_1_port \c - - - :worker_1_port
@ -489,9 +489,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'remove_node_reference_table'::regclass); WHERE logicalrelid = 'remove_node_reference_table'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
\c - - - :worker_1_port \c - - - :worker_1_port
@ -545,9 +545,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'remove_node_reference_table'::regclass); WHERE logicalrelid = 'remove_node_reference_table'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
--verify the data is inserted --verify the data is inserted
@ -614,9 +614,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'remove_node_reference_table'::regclass); WHERE logicalrelid = 'remove_node_reference_table'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
\c - - - :worker_1_port \c - - - :worker_1_port
@ -670,9 +670,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'remove_node_reference_table'::regclass); WHERE logicalrelid = 'remove_node_reference_table'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
\c - - - :worker_1_port \c - - - :worker_1_port
@ -735,9 +735,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'remove_node_reference_table'::regclass); WHERE logicalrelid = 'remove_node_reference_table'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
BEGIN; BEGIN;
@ -767,8 +767,8 @@ WHERE
(0 rows) (0 rows)
SELECT * FROM pg_dist_colocation WHERE colocationid = 1380000; SELECT * FROM pg_dist_colocation WHERE colocationid = 1380000;
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
(0 rows) (0 rows)
-- re-add the node for next tests -- re-add the node for next tests
@ -822,9 +822,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'remove_node_reference_table_schema.table1'::regclass); WHERE logicalrelid = 'remove_node_reference_table_schema.table1'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
\c - - - :worker_1_port \c - - - :worker_1_port
@ -878,9 +878,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'remove_node_reference_table_schema.table1'::regclass); WHERE logicalrelid = 'remove_node_reference_table_schema.table1'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
\c - - - :worker_1_port \c - - - :worker_1_port
@ -938,9 +938,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'remove_node_reference_table'::regclass); WHERE logicalrelid = 'remove_node_reference_table'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
\c - - - :worker_1_port \c - - - :worker_1_port
@ -993,9 +993,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'remove_node_reference_table'::regclass); WHERE logicalrelid = 'remove_node_reference_table'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
\c - - - :worker_1_port \c - - - :worker_1_port

82
src/test/regress/expected/multi_replicate_reference_table.out
View File

@ -116,9 +116,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'replicate_reference_table_valid'::regclass); WHERE logicalrelid = 'replicate_reference_table_valid'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
SELECT 1 FROM master_add_node('localhost', :worker_2_port); SELECT 1 FROM master_add_node('localhost', :worker_2_port);
@ -147,9 +147,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'replicate_reference_table_valid'::regclass); WHERE logicalrelid = 'replicate_reference_table_valid'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
-- test add same node twice -- test add same node twice
@ -172,9 +172,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'replicate_reference_table_valid'::regclass); WHERE logicalrelid = 'replicate_reference_table_valid'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
SELECT 1 FROM master_add_node('localhost', :worker_2_port); SELECT 1 FROM master_add_node('localhost', :worker_2_port);
@ -202,9 +202,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'replicate_reference_table_valid'::regclass); WHERE logicalrelid = 'replicate_reference_table_valid'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
DROP TABLE replicate_reference_table_valid; DROP TABLE replicate_reference_table_valid;
@ -240,9 +240,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'replicate_reference_table_rollback'::regclass); WHERE logicalrelid = 'replicate_reference_table_rollback'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
BEGIN; BEGIN;
@ -272,9 +272,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'replicate_reference_table_rollback'::regclass); WHERE logicalrelid = 'replicate_reference_table_rollback'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
DROP TABLE replicate_reference_table_rollback; DROP TABLE replicate_reference_table_rollback;
@ -304,9 +304,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'replicate_reference_table_commit'::regclass); WHERE logicalrelid = 'replicate_reference_table_commit'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
BEGIN; BEGIN;
@ -337,9 +337,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'replicate_reference_table_commit'::regclass); WHERE logicalrelid = 'replicate_reference_table_commit'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
DROP TABLE replicate_reference_table_commit; DROP TABLE replicate_reference_table_commit;
@ -387,9 +387,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'replicate_reference_table_reference_one'::regclass); WHERE logicalrelid = 'replicate_reference_table_reference_one'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
SELECT colocationid AS reference_table_colocationid FROM pg_dist_colocation WHERE distributioncolumntype = 0 \gset SELECT colocationid AS reference_table_colocationid FROM pg_dist_colocation WHERE distributioncolumntype = 0 \gset
@ -448,9 +448,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'replicate_reference_table_reference_one'::regclass); WHERE logicalrelid = 'replicate_reference_table_reference_one'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
SELECT SELECT
@ -545,9 +545,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'replicate_reference_table_drop'::regclass); WHERE logicalrelid = 'replicate_reference_table_drop'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
BEGIN; BEGIN;
@ -573,8 +573,8 @@ ORDER BY shardid, nodeport;
(0 rows) (0 rows)
SELECT * FROM pg_dist_colocation WHERE colocationid = 1370009; SELECT * FROM pg_dist_colocation WHERE colocationid = 1370009;
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
(0 rows) (0 rows)
-- test adding a node while there is a reference table at another schema -- test adding a node while there is a reference table at another schema
@ -610,9 +610,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'replicate_reference_table_schema.table1'::regclass); WHERE logicalrelid = 'replicate_reference_table_schema.table1'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
SELECT 1 FROM master_add_node('localhost', :worker_2_port); SELECT 1 FROM master_add_node('localhost', :worker_2_port);
@ -641,9 +641,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'replicate_reference_table_schema.table1'::regclass); WHERE logicalrelid = 'replicate_reference_table_schema.table1'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
DROP TABLE replicate_reference_table_schema.table1; DROP TABLE replicate_reference_table_schema.table1;

94
src/test/regress/expected/multi_upgrade_reference_table.out
View File

@ -155,8 +155,8 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'upgrade_reference_table_append'::regclass); WHERE logicalrelid = 'upgrade_reference_table_append'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
(0 rows) (0 rows)
SELECT count(*) active_primaries FROM pg_dist_node WHERE isactive AND noderole='primary' \gset SELECT count(*) active_primaries FROM pg_dist_node WHERE isactive AND noderole='primary' \gset
@ -209,9 +209,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'upgrade_reference_table_append'::regclass); WHERE logicalrelid = 'upgrade_reference_table_append'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
SELECT SELECT
@ -267,9 +267,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'upgrade_reference_table_one_worker'::regclass); WHERE logicalrelid = 'upgrade_reference_table_one_worker'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
1360001 | 1 | 1 | 23 1360001 | 1 | 1 | 23 | 0
(1 row) (1 row)
SELECT SELECT
@ -321,9 +321,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'upgrade_reference_table_one_worker'::regclass); WHERE logicalrelid = 'upgrade_reference_table_one_worker'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
SELECT SELECT
@ -381,9 +381,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'upgrade_reference_table_one_unhealthy'::regclass); WHERE logicalrelid = 'upgrade_reference_table_one_unhealthy'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
1360002 | 1 | 2 | 23 1360002 | 1 | 2 | 23 | 0
(1 row) (1 row)
SELECT SELECT
@ -436,9 +436,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'upgrade_reference_table_one_unhealthy'::regclass); WHERE logicalrelid = 'upgrade_reference_table_one_unhealthy'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
SELECT SELECT
@ -494,9 +494,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'upgrade_reference_table_both_healthy'::regclass); WHERE logicalrelid = 'upgrade_reference_table_both_healthy'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
1360003 | 1 | 2 | 23 1360003 | 1 | 2 | 23 | 0
(1 row) (1 row)
SELECT SELECT
@ -548,9 +548,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'upgrade_reference_table_both_healthy'::regclass); WHERE logicalrelid = 'upgrade_reference_table_both_healthy'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
SELECT SELECT
@ -607,9 +607,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'upgrade_reference_table_transaction_rollback'::regclass); WHERE logicalrelid = 'upgrade_reference_table_transaction_rollback'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
1360004 | 1 | 1 | 23 1360004 | 1 | 1 | 23 | 0
(1 row) (1 row)
SELECT SELECT
@ -663,9 +663,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'upgrade_reference_table_transaction_rollback'::regclass); WHERE logicalrelid = 'upgrade_reference_table_transaction_rollback'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
1360004 | 1 | 1 | 23 1360004 | 1 | 1 | 23 | 0
(1 row) (1 row)
SELECT SELECT
@ -722,9 +722,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'upgrade_reference_table_transaction_commit'::regclass); WHERE logicalrelid = 'upgrade_reference_table_transaction_commit'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
1360004 | 1 | 1 | 23 1360004 | 1 | 1 | 23 | 0
(1 row) (1 row)
SELECT SELECT
@ -778,9 +778,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'upgrade_reference_table_transaction_commit'::regclass); WHERE logicalrelid = 'upgrade_reference_table_transaction_commit'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
SELECT SELECT
@ -848,9 +848,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'upgrade_reference_table_mx'::regclass); WHERE logicalrelid = 'upgrade_reference_table_mx'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
1360005 | 1 | 1 | 23 1360005 | 1 | 1 | 23 | 0
(1 row) (1 row)
SELECT SELECT
@ -899,9 +899,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'upgrade_reference_table_mx'::regclass); WHERE logicalrelid = 'upgrade_reference_table_mx'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
1360005 | 1 | 1 | 23 1360005 | 1 | 1 | 23 | 0
(1 row) (1 row)
SELECT SELECT
@ -968,9 +968,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'upgrade_reference_table_mx'::regclass); WHERE logicalrelid = 'upgrade_reference_table_mx'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
1360006 | 1 | 2 | 23 1360006 | 1 | 2 | 23 | 0
(1 row) (1 row)
SELECT SELECT
@ -1023,9 +1023,9 @@ WHERE colocationid IN
(SELECT colocationid (SELECT colocationid
FROM pg_dist_partition FROM pg_dist_partition
WHERE logicalrelid = 'upgrade_reference_table_mx'::regclass); WHERE logicalrelid = 'upgrade_reference_table_mx'::regclass);
colocationid | shardcount | replicationfactor | distributioncolumntype colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
--------------+------------+-------------------+------------------------ --------------+------------+-------------------+------------------------+-----------------------------
10004 | 1 | -1 | 0 10004 | 1 | -1 | 0 | 0
(1 row) (1 row)
SELECT SELECT

34
src/test/regress/expected/pg12.out
View File

@ -365,8 +365,40 @@ SELECT DISTINCT y FROM test;
25 25
(1 row) (1 row)
-- non deterministic collations
CREATE COLLATION test_pg12.case_insensitive (
provider = icu,
locale = 'und-u-ks-level2',
deterministic = false
);
CREATE TABLE col_test (
id int,
val text collate case_insensitive
);
insert into col_test values
(1, 'asdF'), (2, 'vAlue'), (3, 'asDF');
-- Hash distribution of non deterministic collations are unsupported
select create_distributed_table('col_test', 'val');
ERROR: Hash distributed partition columns may not use a non deterministic collation
select create_distributed_table('col_test', 'id');
NOTICE: Copying data from local table...
create_distributed_table
--------------------------
(1 row)
insert into col_test values
(4, 'vALue'), (5, 'AsDf'), (6, 'value');
select count(*)
from col_test
where val = 'asdf';
count
-------
3
(1 row)
\set VERBOSITY terse \set VERBOSITY terse
drop schema test_pg12 cascade; drop schema test_pg12 cascade;
NOTICE: drop cascades to 11 other objects NOTICE: drop cascades to 13 other objects
\set VERBOSITY default \set VERBOSITY default
SET citus.shard_replication_factor to 2; SET citus.shard_replication_factor to 2;

20
src/test/regress/sql/distributed_collations.sql
View File

@ -35,6 +35,26 @@ SELECT create_distributed_table('test_propagate', 'id');
SELECT * FROM collation_tests.test_propagate WHERE t2 < 'b'; SELECT * FROM collation_tests.test_propagate WHERE t2 < 'b';
SELECT * FROM collation_tests.test_propagate WHERE t2 < 'b' COLLATE "C"; SELECT * FROM collation_tests.test_propagate WHERE t2 < 'b' COLLATE "C";
-- Test range table with collated distribution column
CREATE TABLE test_range(key text COLLATE german_phonebook, val int);
SELECT create_distributed_table('test_range', 'key', 'range');
SELECT master_create_empty_shard('test_range') AS new_shard_id
\gset
UPDATE pg_dist_shard SET shardminvalue = 'a', shardmaxvalue = 'f'
WHERE shardid = :new_shard_id;
SELECT master_create_empty_shard('test_range') AS new_shard_id
\gset
UPDATE pg_dist_shard SET shardminvalue = 'G', shardmaxvalue = 'Z'
WHERE shardid = :new_shard_id;
-- without german_phonebook collation, this would fail
INSERT INTO test_range VALUES (U&'\00E4sop', 1), (U&'Vo\1E9Er', 2);
-- without german_phonebook collation, this would not be router executable
SET client_min_messages TO debug;
SELECT * FROM test_range WHERE key > 'Ab' AND key < U&'\00E4z';
\c - - - :worker_1_port \c - - - :worker_1_port
SELECT c.collname, nsp.nspname, a.rolname SELECT c.collname, nsp.nspname, a.rolname
FROM pg_collation c FROM pg_collation c

25
src/test/regress/sql/pg12.sql
View File

@ -253,6 +253,31 @@ UPDATE test SET y = 40;
COMMIT; COMMIT;
SELECT DISTINCT y FROM test; SELECT DISTINCT y FROM test;
-- non deterministic collations
CREATE COLLATION test_pg12.case_insensitive (
provider = icu,
locale = 'und-u-ks-level2',
deterministic = false
);
CREATE TABLE col_test (
id int,
val text collate case_insensitive
);
insert into col_test values
(1, 'asdF'), (2, 'vAlue'), (3, 'asDF');
-- Hash distribution of non deterministic collations are unsupported
select create_distributed_table('col_test', 'val');
select create_distributed_table('col_test', 'id');
insert into col_test values
(4, 'vALue'), (5, 'AsDf'), (6, 'value');
select count(*)
from col_test
where val = 'asdf';
\set VERBOSITY terse \set VERBOSITY terse
drop schema test_pg12 cascade; drop schema test_pg12 cascade;