From fcf2fd819b175d4b23640911c052dcbb49e077a1 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Philip=20Dub=C3=A9?= <gunkmute@gmail.com>
Date: Fri, 23 Aug 2019 17:32:56 +0000
Subject: [PATCH] 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
---
 .../commands/create_distributed_table.c       | 21 ++++-
 src/backend/distributed/commands/function.c   |  3 +-
 .../distributed/master/master_split_shards.c  |  4 +-
 .../distributed/metadata/metadata_cache.c     | 41 +++-----
 .../planner/multi_physical_planner.c          | 64 +++++++++----
 .../planner/multi_router_planner.c            | 15 +--
 .../distributed/planner/shard_pruning.c       |  6 +-
 .../distributed/sql/citus--9.1-1--9.2-1.sql   | 11 +++
 .../distributed/utils/colocation_utils.c      | 64 ++++++++-----
 .../distributed/utils/reference_table_utils.c | 10 +-
 .../distributed/utils/shardinterval_utils.c   | 50 ++++++++--
 .../worker/worker_partition_protocol.c        | 53 +++++------
 src/include/distributed/colocation_utils.h    |  6 +-
 src/include/distributed/pg_dist_colocation.h  |  4 +-
 src/include/distributed/shardinterval_utils.h | 17 +++-
 src/include/distributed/worker_protocol.h     |  2 -
 .../expected/distributed_collations.out       | 28 ++++++
 .../expected/multi_colocation_utils.out       | 88 ++++++++---------
 .../regress/expected/multi_metadata_sync.out  |  4 +-
 .../multi_remove_node_reference_table.out     | 94 +++++++++----------
 .../multi_replicate_reference_table.out       | 82 ++++++++--------
 .../multi_upgrade_reference_table.out         | 94 +++++++++----------
 src/test/regress/expected/pg12.out            | 34 ++++++-
 .../regress/sql/distributed_collations.sql    | 20 ++++
 src/test/regress/sql/pg12.sql                 | 25 +++++
 25 files changed, 527 insertions(+), 313 deletions(-)

diff --git a/src/backend/distributed/commands/create_distributed_table.c b/src/backend/distributed/commands/create_distributed_table.c
index fa9620acf..e81f3ef3d 100644
--- a/src/backend/distributed/commands/create_distributed_table.c
+++ b/src/backend/distributed/commands/create_distributed_table.c
@@ -1,6 +1,6 @@
 /*-------------------------------------------------------------------------
  *
- * create_distributed_relation.c
+ * create_distributed_table.c
  *	  Routines relation to the creation of distributed relations.
  *
  * 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
 		 * until this transaction is committed.
 		 */
+		Assert(distributionMethod == DISTRIBUTE_BY_HASH);
+
 		Relation pgDistColocation = heap_open(DistColocationRelationId(), ExclusiveLock);
 
 		Oid distributionColumnType = distributionColumn->vartype;
+		Oid distributionColumnCollation = get_typcollation(distributionColumnType);
 		bool createdColocationGroup = false;
 
 		if (pg_strncasecmp(colocateWithTableName, "default", NAMEDATALEN) == 0)
 		{
 			/* check for default colocation group */
 			colocationId = ColocationId(ShardCount, ShardReplicationFactor,
-										distributionColumnType);
+										distributionColumnType,
+										distributionColumnCollation);
 
 			if (colocationId == INVALID_COLOCATION_ID)
 			{
 				colocationId = CreateColocationGroup(ShardCount, ShardReplicationFactor,
-													 distributionColumnType);
+													 distributionColumnType,
+													 distributionColumnCollation);
 				createdColocationGroup = true;
 			}
 		}
@@ -685,6 +690,16 @@ EnsureRelationCanBeDistributed(Oid relationId, Var *distributionColumn,
 							errdetail("Partition column types must have a hash function "
 									  "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)
 	{
diff --git a/src/backend/distributed/commands/function.c b/src/backend/distributed/commands/function.c
index 62ddbd5f2..f8114444e 100644
--- a/src/backend/distributed/commands/function.c
+++ b/src/backend/distributed/commands/function.c
@@ -346,7 +346,8 @@ GetFunctionColocationId(Oid functionOid, char *colocateWithTableName,
 	{
 		/* check for default colocation group */
 		colocationId = ColocationId(ShardCount, ShardReplicationFactor,
-									distributionArgumentOid);
+									distributionArgumentOid, get_typcollation(
+										distributionArgumentOid));
 
 		if (colocationId == INVALID_COLOCATION_ID)
 		{
diff --git a/src/backend/distributed/master/master_split_shards.c b/src/backend/distributed/master/master_split_shards.c
index 4ce9e8abd..f1d74a137 100644
--- a/src/backend/distributed/master/master_split_shards.c
+++ b/src/backend/distributed/master/master_split_shards.c
@@ -81,8 +81,8 @@ worker_hash(PG_FUNCTION_ARGS)
 	fmgr_info_copy(hashFunction, &(typeEntry->hash_proc_finfo), CurrentMemoryContext);
 
 	/* calculate hash value */
-	Datum hashedValueDatum = FunctionCall1Coll(hashFunction, PG_GET_COLLATION(),
-											   valueDatum);
+	Datum hashedValueDatum =
+		FunctionCall1Coll(hashFunction, PG_GET_COLLATION(), valueDatum);
 
 	PG_RETURN_INT32(hashedValueDatum);
 }
diff --git a/src/backend/distributed/metadata/metadata_cache.c b/src/backend/distributed/metadata/metadata_cache.c
index 9519159a0..ceb7d0748 100644
--- a/src/backend/distributed/metadata/metadata_cache.c
+++ b/src/backend/distributed/metadata/metadata_cache.c
@@ -20,6 +20,7 @@
 #include "access/sysattr.h"
 #include "catalog/indexing.h"
 #include "catalog/pg_am.h"
+#include "catalog/pg_collation.h"
 #include "catalog/pg_enum.h"
 #include "catalog/pg_extension.h"
 #include "catalog/pg_namespace.h"
@@ -75,6 +76,7 @@
 /* user configuration */
 int ReadFromSecondaries = USE_SECONDARY_NODES_NEVER;
 
+
 /*
  * ShardCacheEntry represents an entry in the shardId -> ShardInterval cache.
  * 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 void BuildDistTableCacheEntry(DistTableCacheEntry *cacheEntry);
 static void BuildCachedShardList(DistTableCacheEntry *cacheEntry);
-static ShardInterval ** SortShardIntervalArray(ShardInterval **shardIntervalArray,
-											   int shardCount,
-											   FmgrInfo *
-											   shardIntervalSortCompareFunction);
 static void PrepareWorkerNodeCache(void);
 static bool HasUninitializedShardInterval(ShardInterval **sortedShardIntervalArray,
 										  int shardCount);
@@ -201,6 +199,7 @@ static char * AvailableExtensionVersion(void);
 static char * InstalledExtensionVersion(void);
 static bool HasOverlappingShardInterval(ShardInterval **shardIntervalArray,
 										int shardIntervalArrayLength,
+										Oid shardIntervalCollation,
 										FmgrInfo *shardIntervalSortCompareFunction);
 static bool CitusHasBeenLoadedInternal(void);
 static void InitializeCaches(void);
@@ -1222,6 +1221,8 @@ BuildCachedShardList(DistTableCacheEntry *cacheEntry)
 		/* sort the interval array */
 		sortedShardIntervalArray = SortShardIntervalArray(shardIntervalArray,
 														  shardIntervalArrayLength,
+														  cacheEntry->partitionColumn->
+														  varcollid,
 														  shardIntervalCompareFunction);
 
 		/* check if there exists any shard intervals with no min/max values */
@@ -1234,6 +1235,7 @@ BuildCachedShardList(DistTableCacheEntry *cacheEntry)
 			cacheEntry->hasOverlappingShardInterval =
 				HasOverlappingShardInterval(sortedShardIntervalArray,
 											shardIntervalArrayLength,
+											cacheEntry->partitionColumn->varcollid,
 											shardIntervalCompareFunction);
 		}
 		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
  * has a uniform hash distribution, as produced by master_create_worker_shards for
@@ -1428,6 +1407,7 @@ HasUninitializedShardInterval(ShardInterval **sortedShardIntervalArray, int shar
 static bool
 HasOverlappingShardInterval(ShardInterval **shardIntervalArray,
 							int shardIntervalArrayLength,
+							Oid shardIntervalCollation,
 							FmgrInfo *shardIntervalSortCompareFunction)
 {
 	Datum comparisonDatum = 0;
@@ -1448,9 +1428,10 @@ HasOverlappingShardInterval(ShardInterval **shardIntervalArray,
 		Assert(lastShardInterval->minValueExists && lastShardInterval->maxValueExists);
 		Assert(curShardInterval->minValueExists && curShardInterval->maxValueExists);
 
-		comparisonDatum = CompareCall2(shardIntervalSortCompareFunction,
-									   lastShardInterval->maxValue,
-									   curShardInterval->minValue);
+		comparisonDatum = FunctionCall2Coll(shardIntervalSortCompareFunction,
+											shardIntervalCollation,
+											lastShardInterval->maxValue,
+											curShardInterval->minValue);
 		comparisonResult = DatumGetInt32(comparisonDatum);
 
 		if (comparisonResult >= 0)
diff --git a/src/backend/distributed/planner/multi_physical_planner.c b/src/backend/distributed/planner/multi_physical_planner.c
index d546a17e2..bd0283b0c 100644
--- a/src/backend/distributed/planner/multi_physical_planner.c
+++ b/src/backend/distributed/planner/multi_physical_planner.c
@@ -144,6 +144,7 @@ static Task * QueryPushdownTaskCreate(Query *originalQuery, int shardIndex,
 									  TaskType taskType,
 									  bool modifyRequiresMasterEvaluation);
 static bool ShardIntervalsEqual(FmgrInfo *comparisonFunction,
+								Oid collation,
 								ShardInterval *firstInterval,
 								ShardInterval *secondInterval);
 static List * SqlTaskList(Job *job);
@@ -2469,9 +2470,9 @@ QueryPushdownTaskCreate(Query *originalQuery, int shardIndex,
 bool
 CoPartitionedTables(Oid firstRelationId, Oid secondRelationId)
 {
-	bool coPartitionedTables = true;
 	DistTableCacheEntry *firstTableCache = DistributedTableCacheEntry(firstRelationId);
 	DistTableCacheEntry *secondTableCache = DistributedTableCacheEntry(secondRelationId);
+
 	ShardInterval **sortedFirstIntervalArray = firstTableCache->sortedShardIntervalArray;
 	ShardInterval **sortedSecondIntervalArray =
 		secondTableCache->sortedShardIntervalArray;
@@ -2479,6 +2480,18 @@ CoPartitionedTables(Oid firstRelationId, Oid secondRelationId)
 	uint32 secondListShardCount = secondTableCache->shardIntervalArrayLength;
 	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)
 	{
 		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
 	 * 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];
 
 		bool shardIntervalsEqual = ShardIntervalsEqual(comparisonFunction,
+													   collation,
 													   firstInterval,
 													   secondInterval);
 		if (!shardIntervalsEqual || !CoPlacedShardIntervals(firstInterval,
 															secondInterval))
 		{
-			coPartitionedTables = false;
-			break;
+			return false;
 		}
 	}
 
-	return coPartitionedTables;
+	return true;
 }
 
 
@@ -2585,8 +2610,8 @@ CoPlacedShardIntervals(ShardInterval *firstInterval, ShardInterval *secondInterv
  * ShardIntervalsEqual checks if given shard intervals have equal min/max values.
  */
 static bool
-ShardIntervalsEqual(FmgrInfo *comparisonFunction, ShardInterval *firstInterval,
-					ShardInterval *secondInterval)
+ShardIntervalsEqual(FmgrInfo *comparisonFunction, Oid collation,
+					ShardInterval *firstInterval, ShardInterval *secondInterval)
 {
 	bool shardIntervalsEqual = false;
 
@@ -2598,8 +2623,10 @@ ShardIntervalsEqual(FmgrInfo *comparisonFunction, ShardInterval *firstInterval,
 	if (firstInterval->minValueExists && firstInterval->maxValueExists &&
 		secondInterval->minValueExists && secondInterval->maxValueExists)
 	{
-		Datum minDatum = CompareCall2(comparisonFunction, firstMin, secondMin);
-		Datum maxDatum = CompareCall2(comparisonFunction, firstMax, secondMax);
+		Datum minDatum = FunctionCall2Coll(comparisonFunction, collation, firstMin,
+										   secondMin);
+		Datum maxDatum = FunctionCall2Coll(comparisonFunction, collation, firstMax,
+										   secondMax);
 		int firstComparison = DatumGetInt32(minDatum);
 		int secondComparison = DatumGetInt32(maxDatum);
 
@@ -3691,8 +3718,6 @@ FindRangeTableFragmentsList(List *rangeTableFragmentsList, int tableId)
 static bool
 JoinPrunable(RangeTableFragment *leftFragment, RangeTableFragment *rightFragment)
 {
-	bool joinPrunable = false;
-
 	/*
 	 * 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
@@ -3738,10 +3763,10 @@ JoinPrunable(RangeTableFragment *leftFragment, RangeTableFragment *rightFragment
 									leftString->data, rightString->data)));
 		}
 
-		joinPrunable = true;
+		return true;
 	}
 
-	return joinPrunable;
+	return false;
 }
 
 
@@ -3774,10 +3799,13 @@ FragmentInterval(RangeTableFragment *fragment)
 bool
 ShardIntervalsOverlap(ShardInterval *firstInterval, ShardInterval *secondInterval)
 {
-	bool nonOverlap = false;
 	DistTableCacheEntry *intervalRelation =
 		DistributedTableCacheEntry(firstInterval->relationId);
+
+	Assert(intervalRelation->partitionMethod != DISTRIBUTE_BY_NONE);
+
 	FmgrInfo *comparisonFunction = intervalRelation->shardIntervalCompareFunction;
+	Oid collation = intervalRelation->partitionColumn->varcollid;
 
 
 	Datum firstMin = firstInterval->minValue;
@@ -3794,18 +3822,20 @@ ShardIntervalsOverlap(ShardInterval *firstInterval, ShardInterval *secondInterva
 	if (firstInterval->minValueExists && firstInterval->maxValueExists &&
 		secondInterval->minValueExists && secondInterval->maxValueExists)
 	{
-		Datum firstDatum = CompareCall2(comparisonFunction, firstMax, secondMin);
-		Datum secondDatum = CompareCall2(comparisonFunction, secondMax, firstMin);
+		Datum firstDatum = FunctionCall2Coll(comparisonFunction, collation, firstMax,
+											 secondMin);
+		Datum secondDatum = FunctionCall2Coll(comparisonFunction, collation, secondMax,
+											  firstMin);
 		int firstComparison = DatumGetInt32(firstDatum);
 		int secondComparison = DatumGetInt32(secondDatum);
 
 		if (firstComparison < 0 || secondComparison < 0)
 		{
-			nonOverlap = true;
+			return false;
 		}
 	}
 
-	return (!nonOverlap);
+	return true;
 }
 
 
diff --git a/src/backend/distributed/planner/multi_router_planner.c b/src/backend/distributed/planner/multi_router_planner.c
index 0700d11f7..fe472a895 100644
--- a/src/backend/distributed/planner/multi_router_planner.c
+++ b/src/backend/distributed/planner/multi_router_planner.c
@@ -17,9 +17,9 @@
 #include "access/stratnum.h"
 #include "access/xact.h"
 #include "catalog/pg_opfamily.h"
-#include "distributed/citus_clauses.h"
 #include "catalog/pg_type.h"
 #include "distributed/colocation_utils.h"
+#include "distributed/citus_clauses.h"
 #include "distributed/citus_nodes.h"
 #include "distributed/citus_nodefuncs.h"
 #include "distributed/deparse_shard_query.h"
@@ -379,7 +379,7 @@ AddShardIntervalRestrictionToSelect(Query *subqery, ShardInterval *shardInterval
 	TypeCacheEntry *typeEntry = lookup_type_cache(targetPartitionColumnVar->vartype,
 												  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))
 	{
 		ereport(ERROR, (errcode(ERRCODE_UNDEFINED_FUNCTION),
@@ -387,7 +387,10 @@ AddShardIntervalRestrictionToSelect(Query *subqery, ShardInterval *shardInterval
 							   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);
 	hashFunctionExpr->funcid = CitusWorkerHashFunctionId();
 	hashFunctionExpr->args = list_make1(targetPartitionColumnVar);
@@ -401,8 +404,7 @@ AddShardIntervalRestrictionToSelect(Query *subqery, ShardInterval *shardInterval
 								 InvalidOid, false,
 								 (Expr *) hashFunctionExpr,
 								 (Expr *) MakeInt4Constant(shardInterval->minValue),
-								 targetPartitionColumnVar->varcollid,
-								 targetPartitionColumnVar->varcollid);
+								 InvalidOid, InvalidOid);
 
 	/* update the operators with correct operator numbers and function ids */
 	greaterThanAndEqualsBoundExpr->opfuncid =
@@ -416,8 +418,7 @@ AddShardIntervalRestrictionToSelect(Query *subqery, ShardInterval *shardInterval
 								 InvalidOid, false,
 								 (Expr *) hashFunctionExpr,
 								 (Expr *) MakeInt4Constant(shardInterval->maxValue),
-								 targetPartitionColumnVar->varcollid,
-								 targetPartitionColumnVar->varcollid);
+								 InvalidOid, InvalidOid);
 
 	/* update the operators with correct operator numbers and function ids */
 	lessThanAndEqualsBoundExpr->opfuncid = get_opcode(lessThanAndEqualsBoundExpr->opno);
diff --git a/src/backend/distributed/planner/shard_pruning.c b/src/backend/distributed/planner/shard_pruning.c
index dec1338ce..08b8a5978 100644
--- a/src/backend/distributed/planner/shard_pruning.c
+++ b/src/backend/distributed/planner/shard_pruning.c
@@ -276,7 +276,7 @@ PruneShards(Oid relationId, Index rangeTableId, List *whereClauseList,
 		InitFunctionCallInfoData(*(FunctionCallInfo) &
 								 context.compareIntervalFunctionCall,
 								 cacheEntry->shardIntervalCompareFunction,
-								 2, DEFAULT_COLLATION_OID, NULL, NULL);
+								 2, cacheEntry->partitionColumn->varcollid, NULL, NULL);
 	}
 	else
 	{
@@ -290,7 +290,7 @@ PruneShards(Oid relationId, Index rangeTableId, List *whereClauseList,
 		InitFunctionCallInfoData(*(FunctionCallInfo) &
 								 context.compareValueFunctionCall,
 								 cacheEntry->shardColumnCompareFunction,
-								 2, DEFAULT_COLLATION_OID, NULL, NULL);
+								 2, cacheEntry->partitionColumn->varcollid, NULL, NULL);
 	}
 	else
 	{
@@ -679,7 +679,7 @@ AddSAOPartitionKeyRestrictionToInstance(ClauseWalkerContext *context,
 			arrayEqualityOp->inputcollid = arrayOperatorExpression->inputcollid;
 			arrayEqualityOp->opresulttype = get_func_rettype(
 				arrayOperatorExpression->opfuncid);
-			arrayEqualityOp->opcollid = DEFAULT_COLLATION_OID;
+			arrayEqualityOp->opcollid = context->partitionColumn->varcollid;
 			arrayEqualityOp->location = -1;
 			arrayEqualityOp->args = list_make2(strippedLeftOpExpression, constElement);
 
diff --git a/src/backend/distributed/sql/citus--9.1-1--9.2-1.sql b/src/backend/distributed/sql/citus--9.1-1--9.2-1.sql
index e69de29bb..70dfb74b5 100644
--- a/src/backend/distributed/sql/citus--9.1-1--9.2-1.sql
+++ b/src/backend/distributed/sql/citus--9.1-1--9.2-1.sql
@@ -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);
+
diff --git a/src/backend/distributed/utils/colocation_utils.c b/src/backend/distributed/utils/colocation_utils.c
index c1ed7ae05..3892d3364 100644
--- a/src/backend/distributed/utils/colocation_utils.c
+++ b/src/backend/distributed/utils/colocation_utils.c
@@ -167,15 +167,18 @@ MarkTablesColocated(Oid sourceRelationId, Oid targetRelationId)
 
 		Var *sourceDistributionColumn = DistPartitionKey(sourceRelationId);
 		Oid sourceDistributionColumnType = InvalidOid;
+		Oid sourceDistributionColumnCollation = InvalidOid;
 
 		/* reference tables has NULL distribution column */
 		if (sourceDistributionColumn != NULL)
 		{
 			sourceDistributionColumnType = sourceDistributionColumn->vartype;
+			sourceDistributionColumnCollation = sourceDistributionColumn->varcollid;
 		}
 
 		sourceColocationId = CreateColocationGroup(shardCount, shardReplicationFactor,
-												   sourceDistributionColumnType);
+												   sourceDistributionColumnType,
+												   sourceDistributionColumnCollation);
 		UpdateRelationColocationGroup(sourceRelationId, sourceColocationId);
 	}
 
@@ -417,27 +420,31 @@ CompareShardPlacementsByNode(const void *leftElement, const void *rightElement)
 
 
 /*
- * ColocationId searches pg_dist_colocation for shard count, replication factor
- * and distribution column type. If a matching entry is found, it returns the
- * colocation id, otherwise it returns INVALID_COLOCATION_ID.
+ * ColocationId searches pg_dist_colocation for shard count, replication factor,
+ * distribution column type, and distribution column collation. If a matching entry
+ * is found, it returns the colocation id, otherwise returns INVALID_COLOCATION_ID.
  */
 uint32
-ColocationId(int shardCount, int replicationFactor, Oid distributionColumnType)
+ColocationId(int shardCount, int replicationFactor, Oid distributionColumnType, Oid
+			 distributionColumnCollation)
 {
 	uint32 colocationId = INVALID_COLOCATION_ID;
-	const int scanKeyCount = 3;
-	ScanKeyData scanKey[3];
+	const int scanKeyCount = 4;
+	ScanKeyData scanKey[4];
 	bool indexOK = true;
 
 	Relation pgDistColocation = heap_open(DistColocationRelationId(), AccessShareLock);
 
 	/* set scan arguments */
-	ScanKeyInit(&scanKey[0], Anum_pg_dist_colocation_shardcount,
-				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,
+	ScanKeyInit(&scanKey[0], Anum_pg_dist_colocation_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,
 													DistColocationConfigurationIndexId(),
@@ -465,7 +472,8 @@ ColocationId(int shardCount, int replicationFactor, Oid distributionColumnType)
  * colocation id.
  */
 uint32
-CreateColocationGroup(int shardCount, int replicationFactor, Oid distributionColumnType)
+CreateColocationGroup(int shardCount, int replicationFactor, Oid distributionColumnType,
+					  Oid distributionColumnCollation)
 {
 	uint32 colocationId = GetNextColocationId();
 	Datum values[Natts_pg_dist_colocation];
@@ -481,6 +489,8 @@ CreateColocationGroup(int shardCount, int replicationFactor, Oid distributionCol
 		UInt32GetDatum(replicationFactor);
 	values[Anum_pg_dist_colocation_distributioncolumntype - 1] =
 		ObjectIdGetDatum(distributionColumnType);
+	values[Anum_pg_dist_colocation_distributioncolumncollation - 1] =
+		ObjectIdGetDatum(distributionColumnCollation);
 
 	/* open colocation relation and insert the new tuple */
 	Relation pgDistColocation = heap_open(DistColocationRelationId(), RowExclusiveLock);
@@ -566,27 +576,23 @@ CheckDistributionColumnType(Oid sourceRelationId, Oid targetRelationId)
 {
 	Oid sourceDistributionColumnType = InvalidOid;
 	Oid targetDistributionColumnType = InvalidOid;
+	Oid sourceDistributionColumnCollation = InvalidOid;
+	Oid targetDistributionColumnCollation = InvalidOid;
 
 	/* reference tables have NULL distribution column */
 	Var *sourceDistributionColumn = DistPartitionKey(sourceRelationId);
-	if (sourceDistributionColumn == NULL)
-	{
-		sourceDistributionColumnType = InvalidOid;
-	}
-	else
+	if (sourceDistributionColumn != NULL)
 	{
 		sourceDistributionColumnType = sourceDistributionColumn->vartype;
+		sourceDistributionColumnCollation = sourceDistributionColumn->varcollid;
 	}
 
 	/* reference tables have NULL distribution column */
 	Var *targetDistributionColumn = DistPartitionKey(targetRelationId);
-	if (targetDistributionColumn == NULL)
-	{
-		targetDistributionColumnType = InvalidOid;
-	}
-	else
+	if (targetDistributionColumn != NULL)
 	{
 		targetDistributionColumnType = targetDistributionColumn->vartype;
+		targetDistributionColumnCollation = targetDistributionColumn->varcollid;
 	}
 
 	if (sourceDistributionColumnType != targetDistributionColumnType)
@@ -600,6 +606,18 @@ CheckDistributionColumnType(Oid sourceRelationId, Oid targetRelationId)
 								  "%s and %s.", sourceRelationName,
 								  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)));
+	}
 }
 
 
diff --git a/src/backend/distributed/utils/reference_table_utils.c b/src/backend/distributed/utils/reference_table_utils.c
index 8a1255ef0..c0546565b 100644
--- a/src/backend/distributed/utils/reference_table_utils.c
+++ b/src/backend/distributed/utils/reference_table_utils.c
@@ -375,6 +375,7 @@ CreateReferenceTableColocationId()
 {
 	int shardCount = 1;
 	Oid distributionColumnType = InvalidOid;
+	Oid distributionColumnCollation = InvalidOid;
 
 	/*
 	 * We don't maintain replication factor of reference tables anymore and
@@ -383,12 +384,15 @@ CreateReferenceTableColocationId()
 	int replicationFactor = -1;
 
 	/* check for existing colocations */
-	uint32 colocationId = ColocationId(shardCount, replicationFactor,
-									   distributionColumnType);
+	uint32 colocationId =
+		ColocationId(shardCount, replicationFactor, distributionColumnType,
+					 distributionColumnCollation);
+
 	if (colocationId == INVALID_COLOCATION_ID)
 	{
 		colocationId = CreateColocationGroup(shardCount, replicationFactor,
-											 distributionColumnType);
+											 distributionColumnType,
+											 distributionColumnCollation);
 	}
 
 	return colocationId;
diff --git a/src/backend/distributed/utils/shardinterval_utils.c b/src/backend/distributed/utils/shardinterval_utils.c
index 35b5f4a41..11f9a60ba 100644
--- a/src/backend/distributed/utils/shardinterval_utils.c
+++ b/src/backend/distributed/utils/shardinterval_utils.c
@@ -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
  * by their minimum values, using the value's type comparison function.
@@ -60,7 +87,7 @@ LowestShardIntervalById(List *shardIntervalList)
  */
 int
 CompareShardIntervals(const void *leftElement, const void *rightElement,
-					  FmgrInfo *typeCompareFunction)
+					  SortShardIntervalContext *sortContext)
 {
 	ShardInterval *leftShardInterval = *((ShardInterval **) leftElement);
 	ShardInterval *rightShardInterval = *((ShardInterval **) rightElement);
@@ -70,7 +97,7 @@ CompareShardIntervals(const void *leftElement, const void *rightElement,
 	bool rightHasNull = (!rightShardInterval->minValueExists ||
 						 !rightShardInterval->maxValueExists);
 
-	Assert(typeCompareFunction != NULL);
+	Assert(sortContext->comparisonFunction != NULL);
 
 	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 */
 		Datum leftDatum = leftShardInterval->minValue;
 		Datum rightDatum = rightShardInterval->minValue;
-		Datum comparisonDatum = CompareCall2(typeCompareFunction, leftDatum, rightDatum);
+		Datum comparisonDatum = FunctionCall2Coll(sortContext->comparisonFunction,
+												  sortContext->collation, leftDatum,
+												  rightDatum);
 		comparisonResult = DatumGetInt32(comparisonDatum);
 	}
 
@@ -303,8 +332,10 @@ FindShardIntervalIndex(Datum searchedValue, DistTableCacheEntry *cacheEntry)
 		{
 			Assert(compareFunction != NULL);
 
+			Oid shardIntervalCollation = cacheEntry->partitionColumn->varcollid;
 			shardIndex = SearchCachedShardInterval(searchedValue, shardIntervalCache,
-												   shardCount, compareFunction);
+												   shardCount, shardIntervalCollation,
+												   compareFunction);
 
 			/* we should always return a valid shard index for hash partitioned tables */
 			if (shardIndex == INVALID_SHARD_INDEX)
@@ -345,8 +376,10 @@ FindShardIntervalIndex(Datum searchedValue, DistTableCacheEntry *cacheEntry)
 	{
 		Assert(compareFunction != NULL);
 
+		Oid shardIntervalCollation = cacheEntry->partitionColumn->varcollid;
 		shardIndex = SearchCachedShardInterval(searchedValue, shardIntervalCache,
-											   shardCount, compareFunction);
+											   shardCount, shardIntervalCollation,
+											   compareFunction);
 	}
 
 	return shardIndex;
@@ -370,7 +403,8 @@ FindShardIntervalIndex(Datum searchedValue, DistTableCacheEntry *cacheEntry)
  */
 int
 SearchCachedShardInterval(Datum partitionColumnValue, ShardInterval **shardIntervalCache,
-						  int shardCount, FmgrInfo *compareFunction)
+						  int shardCount, Oid shardIntervalCollation,
+						  FmgrInfo *compareFunction)
 {
 	int lowerBoundIndex = 0;
 	int upperBoundIndex = shardCount;
@@ -380,7 +414,7 @@ SearchCachedShardInterval(Datum partitionColumnValue, ShardInterval **shardInter
 		int middleIndex = (lowerBoundIndex + upperBoundIndex) / 2;
 
 		int minValueComparison = FunctionCall2Coll(compareFunction,
-												   DEFAULT_COLLATION_OID,
+												   shardIntervalCollation,
 												   partitionColumnValue,
 												   shardIntervalCache[middleIndex]->
 												   minValue);
@@ -392,7 +426,7 @@ SearchCachedShardInterval(Datum partitionColumnValue, ShardInterval **shardInter
 		}
 
 		int maxValueComparison = FunctionCall2Coll(compareFunction,
-												   DEFAULT_COLLATION_OID,
+												   shardIntervalCollation,
 												   partitionColumnValue,
 												   shardIntervalCache[middleIndex]->
 												   maxValue);
diff --git a/src/backend/distributed/worker/worker_partition_protocol.c b/src/backend/distributed/worker/worker_partition_protocol.c
index da452ed42..f643fac6e 100644
--- a/src/backend/distributed/worker/worker_partition_protocol.c
+++ b/src/backend/distributed/worker/worker_partition_protocol.c
@@ -69,7 +69,8 @@ static void FileOutputStreamWrite(FileOutputStream *file, StringInfo dataToWrite
 static void FileOutputStreamFlush(FileOutputStream *file);
 static void FilterAndPartitionTable(const char *filterQuery,
 									const char *columnName, Oid columnType,
-									uint32 (*PartitionIdFunction)(Datum, const void *),
+									uint32 (*PartitionIdFunction)(Datum, Oid, const
+																  void *),
 									const void *partitionIdContext,
 									FileOutputStream *partitionFileArray,
 									uint32 fileCount);
@@ -78,8 +79,10 @@ static CopyOutState InitRowOutputState(void);
 static void ClearRowOutputState(CopyOutState copyState);
 static void OutputBinaryHeaders(FileOutputStream *partitionFileArray, uint32 fileCount);
 static void OutputBinaryFooters(FileOutputStream *partitionFileArray, uint32 fileCount);
-static uint32 RangePartitionId(Datum partitionValue, const void *context);
-static uint32 HashPartitionId(Datum partitionValue, const void *context);
+static uint32 RangePartitionId(Datum partitionValue, Oid partitionCollation, const
+							   void *context);
+static uint32 HashPartitionId(Datum partitionValue, Oid partitionCollation, const
+							  void *context);
 static StringInfo UserPartitionFilename(StringInfo directoryName, uint32 partitionId);
 static bool FileIsLink(char *filename, struct stat filestat);
 
@@ -187,8 +190,6 @@ worker_hash_partition_table(PG_FUNCTION_ARGS)
 	Datum *hashRangeArray = DeconstructArrayObject(hashRangeObject);
 	int32 partitionCount = ArrayObjectCount(hashRangeObject);
 
-	uint32 (*hashPartitionIdFunction)(Datum, const void *);
-
 	CheckCitusVersion(ERROR);
 
 	HashPartitionContext *partitionContext = palloc0(sizeof(HashPartitionContext));
@@ -198,8 +199,6 @@ worker_hash_partition_table(PG_FUNCTION_ARGS)
 		HasUniformHashDistribution(partitionContext->syntheticShardIntervalArray,
 								   partitionCount);
 
-	hashPartitionIdFunction = &HashPartitionId;
-
 	/* use column's type information to get the hashing function */
 	FmgrInfo *hashFunction = GetFunctionInfo(partitionColumnType, HASH_AM_OID,
 											 HASHSTANDARD_PROC);
@@ -209,7 +208,6 @@ worker_hash_partition_table(PG_FUNCTION_ARGS)
 
 	partitionContext->hashFunction = hashFunction;
 	partitionContext->partitionCount = partitionCount;
-	partitionContext->collation = PG_GET_COLLATION();
 
 	/* we'll use binary search, we need the comparison function */
 	if (!partitionContext->hasUniformHashDistribution)
@@ -228,7 +226,7 @@ worker_hash_partition_table(PG_FUNCTION_ARGS)
 
 	/* call the partitioning function that does the actual work */
 	FilterAndPartitionTable(filterQuery, partitionColumn, partitionColumnType,
-							hashPartitionIdFunction, (const void *) partitionContext,
+							&HashPartitionId, (const void *) partitionContext,
 							partitionFileArray, fileCount);
 
 	/* close partition files and atomically rename (commit) them */
@@ -856,7 +854,7 @@ FileOutputStreamFlush(FileOutputStream *file)
 static void
 FilterAndPartitionTable(const char *filterQuery,
 						const char *partitionColumnName, Oid partitionColumnType,
-						uint32 (*PartitionIdFunction)(Datum, const void *),
+						uint32 (*PartitionIdFunction)(Datum, Oid, const void *),
 						const void *partitionIdContext,
 						FileOutputStream *partitionFileArray,
 						uint32 fileCount)
@@ -864,6 +862,7 @@ FilterAndPartitionTable(const char *filterQuery,
 	FmgrInfo *columnOutputFunctions = NULL;
 	int partitionColumnIndex = 0;
 	Oid partitionColumnTypeId = InvalidOid;
+	Oid partitionColumnCollation = InvalidOid;
 
 	const char *noPortalName = NULL;
 	const bool readOnly = true;
@@ -900,6 +899,9 @@ FilterAndPartitionTable(const char *filterQuery,
 
 		partitionColumnIndex = ColumnIndex(rowDescriptor, partitionColumnName);
 		partitionColumnTypeId = SPI_gettypeid(rowDescriptor, partitionColumnIndex);
+		partitionColumnCollation = TupleDescAttr(rowDescriptor, partitionColumnIndex -
+												 1)->attcollation;
+
 		if (partitionColumnType != partitionColumnTypeId)
 		{
 			ereport(ERROR, (errmsg("partition column types %u and %u do not match",
@@ -939,7 +941,9 @@ FilterAndPartitionTable(const char *filterQuery,
 			 */
 			if (!partitionKeyNull)
 			{
-				partitionId = (*PartitionIdFunction)(partitionKey, partitionIdContext);
+				partitionId = (*PartitionIdFunction)(partitionKey,
+													 partitionColumnCollation,
+													 partitionIdContext);
 				if (partitionId == INVALID_SHARD_INDEX)
 				{
 					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
  * 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.
  */
 static uint32
-RangePartitionId(Datum partitionValue, const void *context)
+RangePartitionId(Datum partitionValue, Oid partitionCollation, const void *context)
 {
 	RangePartitionContext *rangePartitionContext = (RangePartitionContext *) context;
 	FmgrInfo *comparisonFunction = rangePartitionContext->comparisonFunction;
@@ -1191,8 +1185,9 @@ RangePartitionId(Datum partitionValue, const void *context)
 
 		Datum middlePoint = pointArray[middleIndex];
 
-		Datum comparisonDatum = CompareCall2(comparisonFunction, partitionValue,
-											 middlePoint);
+		Datum comparisonDatum =
+			FunctionCall2Coll(comparisonFunction, partitionCollation,
+							  partitionValue, middlePoint);
 		int comparisonResult = DatumGetInt32(comparisonDatum);
 
 		/* 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
  * given data value is null. If not, the function follows the exact same approach
  * as Citus distributed planner uses.
+ *
+ * partitionCollation is unused, as we do not support non deterministic collations
+ * for hash distributed tables.
  */
 static uint32
-HashPartitionId(Datum partitionValue, const void *context)
+HashPartitionId(Datum partitionValue, Oid partitionCollation, const void *context)
 {
 	HashPartitionContext *hashPartitionContext = (HashPartitionContext *) context;
 	FmgrInfo *hashFunction = hashPartitionContext->hashFunction;
@@ -1227,7 +1225,7 @@ HashPartitionId(Datum partitionValue, const void *context)
 	ShardInterval **syntheticShardIntervalArray =
 		hashPartitionContext->syntheticShardIntervalArray;
 	FmgrInfo *comparisonFunction = hashPartitionContext->comparisonFunction;
-	Datum hashDatum = FunctionCall1Coll(hashFunction, hashPartitionContext->collation,
+	Datum hashDatum = FunctionCall1Coll(hashFunction, DEFAULT_COLLATION_OID,
 										partitionValue);
 	int32 hashResult = 0;
 	uint32 hashPartitionId = 0;
@@ -1248,7 +1246,8 @@ HashPartitionId(Datum partitionValue, const void *context)
 	{
 		hashPartitionId =
 			SearchCachedShardInterval(hashDatum, syntheticShardIntervalArray,
-									  partitionCount, comparisonFunction);
+									  partitionCount, InvalidOid,
+									  comparisonFunction);
 	}
 
 
diff --git a/src/include/distributed/colocation_utils.h b/src/include/distributed/colocation_utils.h
index 900bf509a..6853a526a 100644
--- a/src/include/distributed/colocation_utils.h
+++ b/src/include/distributed/colocation_utils.h
@@ -25,9 +25,11 @@ extern List * ColocatedTableList(Oid distributedTableId);
 extern List * ColocatedShardIntervalList(ShardInterval *shardInterval);
 extern Oid ColocatedTableId(Oid colocationId);
 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,
-									Oid distributionColumnType);
+									Oid distributionColumnType,
+									Oid distributionColumnCollation);
 extern uint32 GetNextColocationId(void);
 extern void CheckReplicationModel(Oid sourceRelationId, Oid targetRelationId);
 extern void CheckDistributionColumnType(Oid sourceRelationId, Oid targetRelationId);
diff --git a/src/include/distributed/pg_dist_colocation.h b/src/include/distributed/pg_dist_colocation.h
index 02a3e3d9a..e539bd0b5 100644
--- a/src/include/distributed/pg_dist_colocation.h
+++ b/src/include/distributed/pg_dist_colocation.h
@@ -22,6 +22,7 @@ typedef struct FormData_pg_dist_colocation
 	uint32 shardcount;
 	uint32 replicationfactor;
 	Oid distributioncolumntype;
+	Oid distributioncolumncollation;
 } FormData_pg_dist_colocation;
 
 /* ----------------
@@ -35,11 +36,12 @@ typedef FormData_pg_dist_colocation *Form_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_shardcount 2
 #define Anum_pg_dist_colocation_replicationfactor 3
 #define Anum_pg_dist_colocation_distributioncolumntype 4
+#define Anum_pg_dist_colocation_distributioncolumncollation 5
 
 #define COLOCATIONID_SEQUENCE_NAME "pg_dist_colocationid_seq"
 
diff --git a/src/include/distributed/shardinterval_utils.h b/src/include/distributed/shardinterval_utils.h
index baded1971..5074dde49 100644
--- a/src/include/distributed/shardinterval_utils.h
+++ b/src/include/distributed/shardinterval_utils.h
@@ -26,9 +26,21 @@ typedef struct ShardIntervalCompareFunctionCacheEntry
 	FmgrInfo *functionInfo;
 } 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 int CompareShardIntervals(const void *leftElement, const void *rightElement,
-								 FmgrInfo *typeCompareFunction);
+								 SortShardIntervalContext *sortContext);
 extern int CompareShardIntervalsById(const void *leftElement, const void *rightElement);
 extern int CompareShardPlacementsByShardId(const void *leftElement, const
 										   void *rightElement);
@@ -40,7 +52,8 @@ extern ShardInterval * FindShardInterval(Datum partitionColumnValue,
 extern int FindShardIntervalIndex(Datum searchedValue, DistTableCacheEntry *cacheEntry);
 extern int SearchCachedShardInterval(Datum partitionColumnValue,
 									 ShardInterval **shardIntervalCache,
-									 int shardCount, FmgrInfo *compareFunction);
+									 int shardCount, Oid shardIntervalCollation,
+									 FmgrInfo *compareFunction);
 extern bool SingleReplicatedTable(Oid relationId);
 
 
diff --git a/src/include/distributed/worker_protocol.h b/src/include/distributed/worker_protocol.h
index ddb31c84b..e05a9a330 100644
--- a/src/include/distributed/worker_protocol.h
+++ b/src/include/distributed/worker_protocol.h
@@ -82,7 +82,6 @@ typedef struct HashPartitionContext
 	FmgrInfo *comparisonFunction;
 	ShardInterval **syntheticShardIntervalArray;
 	uint32 partitionCount;
-	Oid collation;
 	bool hasUniformHashDistribution;
 } HashPartitionContext;
 
@@ -137,7 +136,6 @@ extern StringInfo UserTaskFilename(StringInfo directoryName, uint32 taskId);
 extern List * ColumnDefinitionList(List *columnNameList, List *columnTypeList);
 extern CreateStmt * CreateStatement(RangeVar *relation, List *columnDefinitionList);
 extern CopyStmt * CopyStatement(RangeVar *relation, char *sourceFilename);
-extern Datum CompareCall2(FmgrInfo *funcInfo, Datum leftArgument, Datum rightArgument);
 
 /* Function declaration for parsing tree node */
 extern Node * ParseTreeNode(const char *ddlCommand);
diff --git a/src/test/regress/expected/distributed_collations.out b/src/test/regress/expected/distributed_collations.out
index 561bcabee..1562f8232 100644
--- a/src/test/regress/expected/distributed_collations.out
+++ b/src/test/regress/expected/distributed_collations.out
@@ -53,6 +53,34 @@ SELECT * FROM collation_tests.test_propagate WHERE t2 < 'b' COLLATE "C";
   2 | Voẞr | Vossr
 (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
 SELECT c.collname, nsp.nspname, a.rolname
 FROM pg_collation c
diff --git a/src/test/regress/expected/multi_colocation_utils.out b/src/test/regress/expected/multi_colocation_utils.out
index f690f436b..e4c03ca7b 100644
--- a/src/test/regress/expected/multi_colocation_utils.out
+++ b/src/test/regress/expected/multi_colocation_utils.out
@@ -431,13 +431,13 @@ NOTICE:  foreign-data wrapper "fake_fdw" does not have an extension defined
 SELECT * FROM pg_dist_colocation 
     WHERE colocationid >= 1 AND colocationid < 1000 
     ORDER BY colocationid;
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-            3 |          4 |                 2 |                     23
-            4 |          2 |                 2 |                     23
-            5 |          2 |                 1 |                     23
-            6 |          2 |                 2 |                     25
-            7 |          8 |                 2 |                     23
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+            3 |          4 |                 2 |                     23 |                           0
+            4 |          2 |                 2 |                     23 |                           0
+            5 |          2 |                 1 |                     23 |                           0
+            6 |          2 |                 2 |                     25 |                         100
+            7 |          8 |                 2 |                     23 |                           0
 (5 rows)
 
 SELECT logicalrelid, colocationid FROM pg_dist_partition
@@ -459,17 +459,17 @@ SELECT logicalrelid, colocationid FROM pg_dist_partition
 -- check effects of dropping tables
 DROP TABLE table1_groupA;
 SELECT * FROM pg_dist_colocation WHERE colocationid = 4;
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-            4 |          2 |                 2 |                     23
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+            4 |          2 |                 2 |                     23 |                           0
 (1 row)
 
 -- dropping all tables in a colocation group also deletes the colocation group
 DROP TABLE table2_groupA;
 SELECT * FROM pg_dist_colocation WHERE colocationid = 4;
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-            4 |          2 |                 2 |                     23
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+            4 |          2 |                 2 |                     23 |                           0
 (1 row)
 
 -- create dropped colocation group again
@@ -555,14 +555,14 @@ SELECT create_distributed_table('table1_group_default', 'id', colocate_with => '
 SELECT * FROM pg_dist_colocation
     WHERE colocationid >= 1 AND colocationid < 1000
     ORDER BY colocationid;
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-            3 |          4 |                 2 |                     23
-            4 |          2 |                 2 |                     23
-            5 |          2 |                 1 |                     23
-            6 |          2 |                 2 |                     25
-            7 |          8 |                 2 |                     23
-           11 |          3 |                 2 |                     23
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+            3 |          4 |                 2 |                     23 |                           0
+            4 |          2 |                 2 |                     23 |                           0
+            5 |          2 |                 1 |                     23 |                           0
+            6 |          2 |                 2 |                     25 |                         100
+            7 |          8 |                 2 |                     23 |                           0
+           11 |          3 |                 2 |                     23 |                           0
 (6 rows)
 
 SELECT logicalrelid, colocationid FROM pg_dist_partition
@@ -650,14 +650,14 @@ SELECT create_reference_table('table2_groupF');
 SELECT * FROM pg_dist_colocation
     WHERE colocationid >= 1 AND colocationid < 1000
     ORDER BY colocationid;
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-            3 |          4 |                 2 |                     23
-            4 |          2 |                 2 |                     23
-            5 |          2 |                 1 |                     23
-            6 |          2 |                 2 |                     25
-            7 |          8 |                 2 |                     23
-           11 |          3 |                 2 |                     23
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+            3 |          4 |                 2 |                     23 |                           0
+            4 |          2 |                 2 |                     23 |                           0
+            5 |          2 |                 1 |                     23 |                           0
+            6 |          2 |                 2 |                     25 |                         100
+            7 |          8 |                 2 |                     23 |                           0
+           11 |          3 |                 2 |                     23 |                           0
 (6 rows)
 
 -- cross check with internal colocation API
@@ -839,8 +839,8 @@ UPDATE pg_dist_partition SET colocationid = 0
 SELECT * FROM pg_dist_colocation
     WHERE colocationid >= 1 AND colocationid < 1000
     ORDER BY colocationid;
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
 (0 rows)
 
 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
     WHERE colocationid >= 1 AND colocationid < 1000
     ORDER BY colocationid;
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
 (0 rows)
 
 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 
     WHERE colocationid >= 1 AND colocationid < 1000 
     ORDER BY colocationid;
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-            2 |          2 |                 1 |                     23
-            3 |          2 |                 2 |                     25
-            4 |          8 |                 2 |                     23
-            5 |          2 |                 2 |                     23
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+            2 |          2 |                 1 |                     23 |                           0
+            3 |          2 |                 2 |                     25 |                         100
+            4 |          8 |                 2 |                     23 |                           0
+            5 |          2 |                 2 |                     23 |                           0
 (4 rows)
 
 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
     WHERE colocationid >= 1 AND colocationid < 1000
     ORDER BY colocationid;
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-            2 |          2 |                 1 |                     23
-            3 |          2 |                 2 |                     25
-            4 |          8 |                 2 |                     23
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+            2 |          2 |                 1 |                     23 |                           0
+            3 |          2 |                 2 |                     25 |                         100
+            4 |          8 |                 2 |                     23 |                           0
 (3 rows)
 
 SELECT logicalrelid, colocationid FROM pg_dist_partition
diff --git a/src/test/regress/expected/multi_metadata_sync.out b/src/test/regress/expected/multi_metadata_sync.out
index 3c446fd03..72cc80e1d 100644
--- a/src/test/regress/expected/multi_metadata_sync.out
+++ b/src/test/regress/expected/multi_metadata_sync.out
@@ -295,8 +295,8 @@ SELECT "Column", "Type", "Definition" FROM index_attrs WHERE
 
 -- Check that pg_dist_colocation is not synced
 SELECT * FROM pg_dist_colocation ORDER BY colocationid;
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
 (0 rows)
 
 -- Make sure that truncate trigger has been set for the MX table on worker
diff --git a/src/test/regress/expected/multi_remove_node_reference_table.out b/src/test/regress/expected/multi_remove_node_reference_table.out
index 7f991f614..662463386 100644
--- a/src/test/regress/expected/multi_remove_node_reference_table.out
+++ b/src/test/regress/expected/multi_remove_node_reference_table.out
@@ -140,9 +140,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'remove_node_reference_table'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 \c - - - :worker_1_port
@@ -193,9 +193,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'remove_node_reference_table'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 \c - - - :worker_1_port
@@ -273,9 +273,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'remove_node_reference_table'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 \c - - - :worker_1_port
@@ -329,9 +329,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'remove_node_reference_table'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 \c - - - :worker_1_port
@@ -378,9 +378,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'remove_node_reference_table'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 \c - - - :worker_1_port
@@ -433,9 +433,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'remove_node_reference_table'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 \c - - - :worker_1_port
@@ -489,9 +489,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'remove_node_reference_table'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 \c - - - :worker_1_port
@@ -545,9 +545,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'remove_node_reference_table'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 --verify the data is inserted
@@ -614,9 +614,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'remove_node_reference_table'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 \c - - - :worker_1_port
@@ -670,9 +670,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'remove_node_reference_table'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 \c - - - :worker_1_port
@@ -735,9 +735,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'remove_node_reference_table'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 BEGIN;
@@ -767,8 +767,8 @@ WHERE
 (0 rows)
 
 SELECT * FROM pg_dist_colocation WHERE colocationid = 1380000;
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
 (0 rows)
 
 -- re-add the node for next tests
@@ -822,9 +822,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'remove_node_reference_table_schema.table1'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 \c - - - :worker_1_port
@@ -878,9 +878,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'remove_node_reference_table_schema.table1'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 \c - - - :worker_1_port
@@ -938,9 +938,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'remove_node_reference_table'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 \c - - - :worker_1_port
@@ -993,9 +993,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'remove_node_reference_table'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 \c - - - :worker_1_port
diff --git a/src/test/regress/expected/multi_replicate_reference_table.out b/src/test/regress/expected/multi_replicate_reference_table.out
index ab54c801b..5a724afa5 100644
--- a/src/test/regress/expected/multi_replicate_reference_table.out
+++ b/src/test/regress/expected/multi_replicate_reference_table.out
@@ -116,9 +116,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'replicate_reference_table_valid'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 SELECT 1 FROM master_add_node('localhost', :worker_2_port);
@@ -147,9 +147,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'replicate_reference_table_valid'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 -- test add same node twice
@@ -172,9 +172,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'replicate_reference_table_valid'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 SELECT 1 FROM master_add_node('localhost', :worker_2_port);
@@ -202,9 +202,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'replicate_reference_table_valid'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 DROP TABLE replicate_reference_table_valid;
@@ -240,9 +240,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'replicate_reference_table_rollback'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 BEGIN;
@@ -272,9 +272,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'replicate_reference_table_rollback'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 DROP TABLE replicate_reference_table_rollback;
@@ -304,9 +304,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'replicate_reference_table_commit'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 BEGIN;
@@ -337,9 +337,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'replicate_reference_table_commit'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 DROP TABLE replicate_reference_table_commit;
@@ -387,9 +387,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'replicate_reference_table_reference_one'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 SELECT colocationid AS reference_table_colocationid FROM pg_dist_colocation WHERE distributioncolumntype = 0 \gset
@@ -448,9 +448,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'replicate_reference_table_reference_one'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 SELECT
@@ -545,9 +545,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'replicate_reference_table_drop'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 BEGIN;
@@ -573,8 +573,8 @@ ORDER BY shardid, nodeport;
 (0 rows)
 
 SELECT * FROM pg_dist_colocation WHERE colocationid = 1370009;
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
 (0 rows)
 
 -- test adding a node while there is a reference table at another schema
@@ -610,9 +610,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'replicate_reference_table_schema.table1'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 SELECT 1 FROM master_add_node('localhost', :worker_2_port);
@@ -641,9 +641,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'replicate_reference_table_schema.table1'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 DROP TABLE replicate_reference_table_schema.table1;
diff --git a/src/test/regress/expected/multi_upgrade_reference_table.out b/src/test/regress/expected/multi_upgrade_reference_table.out
index 232f5824b..8101d3ec1 100644
--- a/src/test/regress/expected/multi_upgrade_reference_table.out
+++ b/src/test/regress/expected/multi_upgrade_reference_table.out
@@ -155,8 +155,8 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'upgrade_reference_table_append'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
 (0 rows)
 
 SELECT count(*) active_primaries FROM pg_dist_node WHERE isactive AND noderole='primary' \gset
@@ -209,9 +209,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'upgrade_reference_table_append'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 SELECT
@@ -267,9 +267,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'upgrade_reference_table_one_worker'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-      1360001 |          1 |                 1 |                     23
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+      1360001 |          1 |                 1 |                     23 |                           0
 (1 row)
 
 SELECT
@@ -321,9 +321,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'upgrade_reference_table_one_worker'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 SELECT
@@ -381,9 +381,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'upgrade_reference_table_one_unhealthy'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-      1360002 |          1 |                 2 |                     23
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+      1360002 |          1 |                 2 |                     23 |                           0
 (1 row)
 
 SELECT
@@ -436,9 +436,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'upgrade_reference_table_one_unhealthy'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 SELECT
@@ -494,9 +494,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'upgrade_reference_table_both_healthy'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-      1360003 |          1 |                 2 |                     23
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+      1360003 |          1 |                 2 |                     23 |                           0
 (1 row)
 
 SELECT
@@ -548,9 +548,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'upgrade_reference_table_both_healthy'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 SELECT
@@ -607,9 +607,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'upgrade_reference_table_transaction_rollback'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-      1360004 |          1 |                 1 |                     23
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+      1360004 |          1 |                 1 |                     23 |                           0
 (1 row)
 
 SELECT
@@ -663,9 +663,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'upgrade_reference_table_transaction_rollback'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-      1360004 |          1 |                 1 |                     23
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+      1360004 |          1 |                 1 |                     23 |                           0
 (1 row)
 
 SELECT
@@ -722,9 +722,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'upgrade_reference_table_transaction_commit'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-      1360004 |          1 |                 1 |                     23
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+      1360004 |          1 |                 1 |                     23 |                           0
 (1 row)
 
 SELECT
@@ -778,9 +778,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'upgrade_reference_table_transaction_commit'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 SELECT
@@ -848,9 +848,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'upgrade_reference_table_mx'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-      1360005 |          1 |                 1 |                     23
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+      1360005 |          1 |                 1 |                     23 |                           0
 (1 row)
 
 SELECT
@@ -899,9 +899,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'upgrade_reference_table_mx'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-      1360005 |          1 |                 1 |                     23
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+      1360005 |          1 |                 1 |                     23 |                           0
 (1 row)
 
 SELECT
@@ -968,9 +968,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'upgrade_reference_table_mx'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-      1360006 |          1 |                 2 |                     23
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+      1360006 |          1 |                 2 |                     23 |                           0
 (1 row)
 
 SELECT
@@ -1023,9 +1023,9 @@ WHERE colocationid IN
     (SELECT colocationid
      FROM pg_dist_partition
      WHERE logicalrelid = 'upgrade_reference_table_mx'::regclass);
- colocationid | shardcount | replicationfactor | distributioncolumntype 
---------------+------------+-------------------+------------------------
-        10004 |          1 |                -1 |                      0
+ colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation 
+--------------+------------+-------------------+------------------------+-----------------------------
+        10004 |          1 |                -1 |                      0 |                           0
 (1 row)
 
 SELECT
diff --git a/src/test/regress/expected/pg12.out b/src/test/regress/expected/pg12.out
index b946f99c7..8e5350b42 100644
--- a/src/test/regress/expected/pg12.out
+++ b/src/test/regress/expected/pg12.out
@@ -365,8 +365,40 @@ SELECT DISTINCT y FROM test;
  25
 (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
 drop schema test_pg12 cascade;
-NOTICE:  drop cascades to 11 other objects
+NOTICE:  drop cascades to 13 other objects
 \set VERBOSITY default
 SET citus.shard_replication_factor to 2;
diff --git a/src/test/regress/sql/distributed_collations.sql b/src/test/regress/sql/distributed_collations.sql
index 85fc37814..c32100e5c 100644
--- a/src/test/regress/sql/distributed_collations.sql
+++ b/src/test/regress/sql/distributed_collations.sql
@@ -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' 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
 SELECT c.collname, nsp.nspname, a.rolname
 FROM pg_collation c
diff --git a/src/test/regress/sql/pg12.sql b/src/test/regress/sql/pg12.sql
index 9b3848cca..1658c511d 100644
--- a/src/test/regress/sql/pg12.sql
+++ b/src/test/regress/sql/pg12.sql
@@ -253,6 +253,31 @@ UPDATE test SET y = 40;
 COMMIT;
 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
 drop schema test_pg12 cascade;