Support creating collations as part of dependency resolution. Propagate ALTER/DROP on distributed collations

Propagate CREATE COLLATION when outside transaction

src/backend/distributed/commands/collation.c

@@ -0,0 +1,601 @@
+/*-------------------------------------------------------------------------
+ * collation.c
+ *
+ * This file contains functions to create, alter and drop policies on
+ * distributed tables.
+ *
+ * Copyright (c) 2019, Citus Data, Inc.
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/htup_details.h"
+#include "access/xact.h"
+#include "catalog/pg_collation.h"
+#include "distributed/commands/utility_hook.h"
+#include "distributed/commands.h"
+#include "distributed/deparser.h"
+#include "distributed/master_metadata_utility.h"
+#include "distributed/metadata/distobject.h"
+#include "distributed/metadata_sync.h"
+#include "distributed/multi_executor.h"
+#include "distributed/relation_access_tracking.h"
+#include "distributed/worker_create_or_replace.h"
+#include "distributed/worker_manager.h"
+#include "parser/parse_type.h"
+#include "utils/builtins.h"
+#include "utils/lsyscache.h"
+#include "utils/syscache.h"
+#include "miscadmin.h"
+
+
+static char * CreateCollationDDLInternal(Oid collationId, Oid *collowner,
+										 char **quotedCollationName);
+static List * FilterNameListForDistributedCollations(List *objects, bool missing_ok,
+													 List **addresses);
+static void EnsureSequentialModeForCollationDDL(void);
+
+
+/*
+ * GetCreateCollationDDLInternal returns a CREATE COLLATE sql string for the
+ * given collationId.
+ *
+ * It includes 2 out parameters to assist creation of ALTER COLLATION OWNER.
+ * quotedCollationName must not be NULL.
+ */
+static char *
+CreateCollationDDLInternal(Oid collationId, Oid *collowner, char **quotedCollationName)
+{
+	char *schemaName = NULL;
+	StringInfoData collationNameDef;
+	const char *providerString = NULL;
+	HeapTuple heapTuple = NULL;
+	Form_pg_collation collationForm = NULL;
+	char collprovider;
+	const char *collcollate;
+	const char *collctype;
+	const char *collname;
+	Oid collnamespace;
+#if PG_VERSION_NUM >= 120000
+	bool collisdeterministic;
+#endif
+
+	heapTuple = SearchSysCache1(COLLOID, ObjectIdGetDatum(collationId));
+	if (!HeapTupleIsValid(heapTuple))
+	{
+		elog(ERROR, "citus cache lookup failed for collation %u", collationId);
+	}
+
+	collationForm = (Form_pg_collation) GETSTRUCT(heapTuple);
+	collprovider = collationForm->collprovider;
+	collcollate = NameStr(collationForm->collcollate);
+	collctype = NameStr(collationForm->collctype);
+	collnamespace = collationForm->collnamespace;
+	collname = NameStr(collationForm->collname);
+#if PG_VERSION_NUM >= 120000
+	collisdeterministic = collationForm->collisdeterministic;
+#endif
+
+	if (collowner != NULL)
+	{
+		*collowner = collationForm->collowner;
+	}
+
+	ReleaseSysCache(heapTuple);
+	schemaName = get_namespace_name(collnamespace);
+	*quotedCollationName = quote_qualified_identifier(schemaName, collname);
+	providerString =
+		collprovider == COLLPROVIDER_DEFAULT ? "default" :
+		collprovider == COLLPROVIDER_ICU ? "icu" :
+		collprovider == COLLPROVIDER_LIBC ? "libc" : NULL;
+
+	if (providerString == NULL)
+	{
+		elog(ERROR, "unknown collation provider: %c", collprovider);
+	}
+
+	initStringInfo(&collationNameDef);
+	appendStringInfo(&collationNameDef,
+					 "CREATE COLLATION %s (provider = '%s'",
+					 *quotedCollationName, providerString);
+
+	if (strcmp(collcollate, collctype))
+	{
+		appendStringInfo(&collationNameDef,
+						 ", locale = %s",
+						 quote_literal_cstr(collcollate));
+	}
+	else
+	{
+		appendStringInfo(&collationNameDef,
+						 ", lc_collate = %s, lc_ctype = %s",
+						 quote_literal_cstr(collcollate),
+						 quote_literal_cstr(collctype));
+	}
+
+#if PG_VERSION_NUM >= 120000
+	if (!collisdeterministic)
+	{
+		appendStringInfoString(&collationNameDef, ", deterministic = false");
+	}
+#endif
+
+
+	appendStringInfoChar(&collationNameDef, ')');
+	return collationNameDef.data;
+}
+
+
+/*
+ * CreateCollationDDL wrap CreateCollationDDLInternal to hide the out parameters.
+ */
+char *
+CreateCollationDDL(Oid collationId)
+{
+	char *quotedCollationName = NULL;
+	return CreateCollationDDLInternal(collationId, NULL, &quotedCollationName);
+}
+
+
+/*
+ * CreateCollationDDLsIdempotent returns a List of cstrings for creating the collation
+ * using create_or_replace_object & includes ALTER COLLATION ROLE.
+ */
+List *
+CreateCollationDDLsIdempotent(Oid collationId)
+{
+	StringInfoData collationAlterOwnerCommand;
+	Oid collowner = InvalidOid;
+	char *quotedCollationName = NULL;
+	char *createCollationCommand = CreateCollationDDLInternal(collationId, &collowner,
+															  &quotedCollationName);
+
+	initStringInfo(&collationAlterOwnerCommand);
+	appendStringInfo(&collationAlterOwnerCommand,
+					 "ALTER COLLATION %s OWNER TO %s",
+					 quotedCollationName,
+					 quote_identifier(GetUserNameFromId(collowner, false)));
+
+	return list_make2(WrapCreateOrReplace(createCollationCommand),
+					  collationAlterOwnerCommand.data);
+}
+
+
+ObjectAddress
+AlterCollationOwnerObjectAddress(AlterOwnerStmt *stmt)
+{
+	Relation relation;
+
+	Assert(stmt->objectType == OBJECT_COLLATION);
+
+	return get_object_address(stmt->objectType, stmt->object, &relation,
+							  AccessExclusiveLock, false);
+}
+
+
+/*
+ * FilterNameListForDistributedCollations takes a list of objects to delete.
+ * This list is filtered against the collations that are distributed.
+ *
+ * The original list will not be touched, a new list will be created with only the objects
+ * in there.
+ *
+ * objectAddresses is replaced with a list of object addresses for the filtered objects.
+ */
+static List *
+FilterNameListForDistributedCollations(List *objects, bool missing_ok,
+									   List **objectAddresses)
+{
+	ListCell *objectCell = NULL;
+	List *result = NIL;
+
+	*objectAddresses = NIL;
+
+	foreach(objectCell, objects)
+	{
+		List *collName = lfirst(objectCell);
+		Oid collOid = get_collation_oid(collName, true);
+		ObjectAddress collAddress = { 0 };
+
+		if (!OidIsValid(collOid))
+		{
+			continue;
+		}
+
+		ObjectAddressSet(collAddress, CollationRelationId, collOid);
+		if (IsObjectDistributed(&collAddress))
+		{
+			ObjectAddress *address = palloc0(sizeof(ObjectAddress));
+			*address = collAddress;
+			*objectAddresses = lappend(*objectAddresses, address);
+			result = lappend(result, collName);
+		}
+	}
+	return result;
+}
+
+
+List *
+PlanDropCollationStmt(DropStmt *stmt)
+{
+	/*
+	 * We swap the list of objects to remove during deparse so we need a reference back to
+	 * the old list to put back
+	 */
+	ListCell *addressCell = NULL;
+	List *distributedTypeAddresses = NIL;
+
+	if (!ShouldPropagate())
+	{
+		return NIL;
+	}
+
+	QualifyTreeNode((Node *) stmt);
+
+	List *oldCollations = stmt->objects;
+	List *distributedCollations =
+		FilterNameListForDistributedCollations(oldCollations, stmt->missing_ok,
+											   &distributedTypeAddresses);
+	if (list_length(distributedCollations) <= 0)
+	{
+		/* no distributed types to drop */
+		return NIL;
+	}
+
+	/*
+	 * managing collations can only be done on the coordinator if ddl propagation is on. when
+	 * it is off we will never get here. MX workers don't have a notion of distributed
+	 * collations, so we block the call.
+	 */
+	EnsureCoordinator();
+
+	/*
+	 * remove the entries for the distributed objects on dropping
+	 */
+	foreach(addressCell, distributedTypeAddresses)
+	{
+		ObjectAddress *address = (ObjectAddress *) lfirst(addressCell);
+		UnmarkObjectDistributed(address);
+	}
+
+	/*
+	 * temporary swap the lists of objects to delete with the distributed objects and
+	 * deparse to an executable sql statement for the workers
+	 */
+	stmt->objects = distributedCollations;
+	char *dropStmtSql = DeparseTreeNode((Node *) stmt);
+	stmt->objects = oldCollations;
+
+	EnsureSequentialModeForCollationDDL();
+
+	/* to prevent recursion with mx we disable ddl propagation */
+	List *commands = list_make3(DISABLE_DDL_PROPAGATION,
+								(void *) dropStmtSql,
+								ENABLE_DDL_PROPAGATION);
+
+	return NodeDDLTaskList(ALL_WORKERS, commands);
+}
+
+
+/*
+ * PlanAlterCollationOwnerStmt is called for change of ownership of collations
+ * before the ownership is changed on the local instance.
+ *
+ * If the type for which the owner is changed is distributed we execute the change on all
+ * the workers to keep the type in sync across the cluster.
+ */
+List *
+PlanAlterCollationOwnerStmt(AlterOwnerStmt *stmt, const char *queryString)
+{
+	Assert(stmt->objectType == OBJECT_COLLATION);
+
+	ObjectAddress *collationAddress = GetObjectAddressFromParseTree((Node *) stmt, false);
+	if (!ShouldPropagateObject(collationAddress))
+	{
+		return NIL;
+	}
+
+	EnsureCoordinator();
+
+	QualifyTreeNode((Node *) stmt);
+	char *sql = DeparseTreeNode((Node *) stmt);
+
+	EnsureSequentialModeForCollationDDL();
+	List *commands = list_make3(DISABLE_DDL_PROPAGATION,
+								(void *) sql,
+								ENABLE_DDL_PROPAGATION);
+
+	return NodeDDLTaskList(ALL_WORKERS, commands);
+}
+
+
+/*
+ * PlanRenameCollationStmt is called when the user is renaming the collation. The invocation happens
+ * before the statement is applied locally.
+ *
+ * As the collation already exists we have access to the ObjectAddress for the collation, this is
+ * used to check if the collation is distributed. If the collation is distributed the rename is
+ * executed on all the workers to keep the collation in sync across the cluster.
+ */
+List *
+PlanRenameCollationStmt(RenameStmt *stmt, const char *queryString)
+{
+	ObjectAddress *collationAddress = GetObjectAddressFromParseTree((Node *) stmt, false);
+	if (!ShouldPropagateObject(collationAddress))
+	{
+		return NIL;
+	}
+
+	/* fully qualify */
+	QualifyTreeNode((Node *) stmt);
+
+	/* deparse sql*/
+	char *renameStmtSql = DeparseTreeNode((Node *) stmt);
+
+	EnsureSequentialModeForCollationDDL();
+
+	/* to prevent recursion with mx we disable ddl propagation */
+	List *commands = list_make3(DISABLE_DDL_PROPAGATION,
+								(void *) renameStmtSql,
+								ENABLE_DDL_PROPAGATION);
+
+	return NodeDDLTaskList(ALL_WORKERS, commands);
+}
+
+
+/*
+ * PlanAlterCollationSchemaStmt is executed before the statement is applied to the local
+ * postgres instance.
+ *
+ * In this stage we can prepare the commands that need to be run on all workers.
+ */
+List *
+PlanAlterCollationSchemaStmt(AlterObjectSchemaStmt *stmt, const char *queryString)
+{
+	Assert(stmt->objectType == OBJECT_COLLATION);
+
+	ObjectAddress *collationAddress = GetObjectAddressFromParseTree((Node *) stmt, false);
+	if (!ShouldPropagateObject(collationAddress))
+	{
+		return NIL;
+	}
+
+	EnsureCoordinator();
+
+	QualifyTreeNode((Node *) stmt);
+	char *sql = DeparseTreeNode((Node *) stmt);
+
+	EnsureSequentialModeForCollationDDL();
+
+	List *commands = list_make3(DISABLE_DDL_PROPAGATION,
+								(void *) sql,
+								ENABLE_DDL_PROPAGATION);
+
+	return NodeDDLTaskList(ALL_WORKERS, commands);
+}
+
+
+/*
+ * ProcessAlterCollationSchemaStmt is executed after the change has been applied locally, we
+ * can now use the new dependencies of the type to ensure all its dependencies exist on
+ * the workers before we apply the commands remotely.
+ */
+void
+ProcessAlterCollationSchemaStmt(AlterObjectSchemaStmt *stmt, const char *queryString)
+{
+	Assert(stmt->objectType == OBJECT_COLLATION);
+
+	ObjectAddress *collationAddress = GetObjectAddressFromParseTree((Node *) stmt, false);
+	if (!ShouldPropagateObject(collationAddress))
+	{
+		return;
+	}
+
+	/* dependencies have changed (schema) let's ensure they exist */
+	EnsureDependenciesExistsOnAllNodes(collationAddress);
+}
+
+
+/*
+ * RenameCollationStmtObjectAddress returns the ObjectAddress of the type that is the object
+ * of the RenameStmt. Errors if missing_ok is false.
+ */
+ObjectAddress *
+RenameCollationStmtObjectAddress(RenameStmt *stmt, bool missing_ok)
+{
+	Assert(stmt->renameType == OBJECT_COLLATION);
+
+	ObjectAddress *address = palloc0(sizeof(ObjectAddress));
+	Oid collationOid = get_collation_oid((List *) stmt->object, missing_ok);
+
+	ObjectAddressSet(*address, CollationRelationId, collationOid);
+	return address;
+}
+
+
+/*
+ * AlterCollationSchemaStmtObjectAddress returns the ObjectAddress of the type that is the
+ * subject of the AlterObjectSchemaStmt. Errors if missing_ok is false.
+ *
+ * This could be called both before or after it has been applied locally. It will look in
+ * the old schema first, if the type cannot be found in that schema it will look in the
+ * new schema. Errors if missing_ok is false and the type cannot be found in either of the
+ * schemas.
+ */
+ObjectAddress *
+AlterCollationSchemaStmtObjectAddress(AlterObjectSchemaStmt *stmt, bool missing_ok)
+{
+	Assert(stmt->objectType == OBJECT_COLLATION);
+
+	List *name = (List *) stmt->object;
+	Oid collationOid = get_collation_oid(name, true);
+
+	if (collationOid == InvalidOid)
+	{
+		List *newName = list_make2(makeString(stmt->newschema), lfirst(list_tail(name)));
+
+		collationOid = get_collation_oid(newName, true);
+
+		if (!missing_ok && collationOid == InvalidOid)
+		{
+			ereport(ERROR, (errcode(ERRCODE_UNDEFINED_OBJECT),
+							errmsg("type \"%s\" does not exist",
+								   NameListToString(name))));
+		}
+	}
+
+	ObjectAddress *address = palloc0(sizeof(ObjectAddress));
+	ObjectAddressSet(*address, CollationRelationId, collationOid);
+	return address;
+}
+
+
+/*
+ * EnsureSequentialModeForCollationDDL makes sure that the current transaction is already in
+ * sequential mode, or can still safely be put in sequential mode, it errors if that is
+ * not possible. The error contains information for the user to retry the transaction with
+ * sequential mode set from the begining.
+ *
+ * As collations are node scoped objects there exists only 1 instance of the collation used by
+ * potentially multiple shards. To make sure all shards in the transaction can interact
+ * with the type the type needs to be visible on all connections used by the transaction,
+ * meaning we can only use 1 connection per node.
+ */
+static void
+EnsureSequentialModeForCollationDDL(void)
+{
+	if (!IsTransactionBlock())
+	{
+		/* we do not need to switch to sequential mode if we are not in a transaction */
+		return;
+	}
+
+	if (ParallelQueryExecutedInTransaction())
+	{
+		ereport(ERROR, (errmsg("cannot create or modify collation because there was a "
+							   "parallel operation on a distributed table in the "
+							   "transaction"),
+						errdetail("When creating or altering a collation, Citus needs to "
+								  "perform all operations over a single connection per "
+								  "node to ensure consistency."),
+						errhint("Try re-running the transaction with "
+								"\"SET LOCAL citus.multi_shard_modify_mode TO "
+								"\'sequential\';\"")));
+	}
+
+	ereport(DEBUG1, (errmsg("switching to sequential query execution mode"),
+					 errdetail("Collation is created or altered. To make sure subsequent "
+							   "commands see the collation correctly we need to make sure to "
+							   "use only one connection for all future commands")));
+	SetLocalMultiShardModifyModeToSequential();
+}
+
+
+/*
+ * GenerateBackupNameForCollationCollision generates a new collation name for an existing collation.
+ * The name is generated in such a way that the new name doesn't overlap with an existing collation
+ * by adding a suffix with incrementing number after the new name.
+ */
+char *
+GenerateBackupNameForCollationCollision(const ObjectAddress *address)
+{
+	char *newName = palloc0(NAMEDATALEN);
+	char suffix[NAMEDATALEN] = { 0 };
+	int count = 0;
+	char *baseName = get_collation_name(address->objectId);
+	int baseLength = strlen(baseName);
+	HeapTuple collationTuple = SearchSysCache1(COLLOID, address->objectId);
+
+	if (!HeapTupleIsValid(collationTuple))
+	{
+		elog(ERROR, "citus cache lookup failed");
+		return NULL;
+	}
+	Form_pg_collation collationForm = (Form_pg_collation) GETSTRUCT(collationTuple);
+	Value *namespace = makeString(get_namespace_name(collationForm->collnamespace));
+	ReleaseSysCache(collationTuple);
+
+	while (true)
+	{
+		int suffixLength = snprintf(suffix, NAMEDATALEN - 1, "(citus_backup_%d)",
+									count);
+
+		/* trim the base name at the end to leave space for the suffix and trailing \0 */
+		baseLength = Min(baseLength, NAMEDATALEN - suffixLength - 1);
+
+		/* clear newName before copying the potentially trimmed baseName and suffix */
+		memset(newName, 0, NAMEDATALEN);
+		strncpy(newName, baseName, baseLength);
+		strncpy(newName + baseLength, suffix, suffixLength);
+
+		List *newCollationName = list_make2(namespace, makeString(newName));
+
+		/* don't need to rename if the input arguments don't match */
+		Oid existingCollationId = get_collation_oid(newCollationName, true);
+
+		if (existingCollationId == InvalidOid)
+		{
+			return newName;
+		}
+
+		count++;
+	}
+}
+
+
+ObjectAddress *
+DefineCollationStmtObjectAddress(DefineStmt *stmt, bool missing_ok)
+{
+	Assert(stmt->kind == OBJECT_COLLATION);
+
+	Oid collOid = get_collation_oid(stmt->defnames, missing_ok);
+	ObjectAddress *address = palloc0(sizeof(ObjectAddress));
+
+	ObjectAddressSet(*address, CollationRelationId, collOid);
+
+	return address;
+}
+
+
+/*
+ * ProcessCollationDefineStmt executed after the extension has been
+ * created locally and before we create it on the worker nodes.
+ * As we now have access to ObjectAddress of the extension that is just
+ * created, we can mark it as distributed to make sure that its
+ * dependencies exist on all nodes.
+ */
+List *
+ProcessCollationDefineStmt(DefineStmt *stmt, const char *queryString)
+{
+	Assert(stmt->kind == OBJECT_COLLATION);
+
+	if (!ShouldPropagate())
+	{
+		return NIL;
+	}
+
+	/*
+	 * If the create collation command is a part of a multi-statement transaction,
+	 * do not propagate it
+	 */
+	if (IsMultiStatementTransaction())
+	{
+		return NIL;
+	}
+
+	ObjectAddress *collationAddress =
+		DefineCollationStmtObjectAddress(stmt, false);
+
+	if (collationAddress->objectId == InvalidOid)
+	{
+		return NIL;
+	}
+
+	EnsureDependenciesExistsOnAllNodes(collationAddress);
+
+	MarkObjectDistributed(collationAddress);
+
+	return NodeDDLTaskList(ALL_WORKERS, CreateCollationDDLsIdempotent(
+							   collationAddress->objectId));
+}

src/backend/distributed/commands/dependencies.c

@@ -158,6 +158,11 @@ GetDependencyCreateDDLCommands(const ObjectAddress *dependency)
 			break;
 		}
 
+		case OCLASS_COLLATION:
+		{
+			return CreateCollationDDLsIdempotent(dependency->objectId);
+		}
+
 		case OCLASS_PROC:
 		{
 			return CreateFunctionDDLCommandsIdempotent(dependency);

src/backend/distributed/commands/extension.c

@@ -130,7 +130,7 @@ PlanCreateExtensionStmt(CreateExtensionStmt *createExtensionStmt, const char *qu
 	}
 
 	/*
-	 * If the extension command is a part of a bigger multi-statement transaction,
+	 * If the extension command is a part of a multi-statement transaction,
 	 * do not propagate it
 	 */
 	if (IsMultiStatementTransaction())

src/backend/distributed/commands/function.c

@@ -1567,7 +1567,7 @@ ProcessAlterFunctionSchemaStmt(AlterObjectSchemaStmt *stmt, const char *queryStr
 		return;
 	}
 
-	/* dependencies have changed (schema) lets ensure they exist */
+	/* dependencies have changed (schema) let's ensure they exist */
 	EnsureDependenciesExistsOnAllNodes(address);
 }
 

src/backend/distributed/commands/schema.c

@@ -121,6 +121,11 @@ PlanAlterObjectSchemaStmt(AlterObjectSchemaStmt *stmt, const char *queryString)
 			return PlanAlterTypeSchemaStmt(stmt, queryString);
 		}
 
+		case OBJECT_COLLATION:
+		{
+			return PlanAlterCollationSchemaStmt(stmt, queryString);
+		}
+
 		case OBJECT_PROCEDURE:
 		case OBJECT_AGGREGATE:
 		case OBJECT_FUNCTION:
@@ -200,6 +205,12 @@ ProcessAlterObjectSchemaStmt(AlterObjectSchemaStmt *stmt, const char *queryStrin
 			return;
 		}
 
+		case OBJECT_COLLATION:
+		{
+			ProcessAlterCollationSchemaStmt(stmt, queryString);
+			return;
+		}
+
 		case OBJECT_PROCEDURE:
 		case OBJECT_AGGREGATE:
 		case OBJECT_FUNCTION:

src/backend/distributed/commands/type.c

@@ -489,9 +489,9 @@ PlanDropTypeStmt(DropStmt *stmt, const char *queryString)
 	const char *dropStmtSql = DeparseTreeNode((Node *) stmt);
 	stmt->objects = oldTypes;
 
-	/* to prevent recursion with mx we disable ddl propagation */
 	EnsureSequentialModeForTypeDDL();
 
+	/* to prevent recursion with mx we disable ddl propagation */
 	List *commands = list_make3(DISABLE_DDL_PROPAGATION,
 								(void *) dropStmtSql,
 								ENABLE_DDL_PROPAGATION);
@@ -524,9 +524,9 @@ PlanRenameTypeStmt(RenameStmt *stmt, const char *queryString)
 	/* deparse sql*/
 	const char *renameStmtSql = DeparseTreeNode((Node *) stmt);
 
-	/* to prevent recursion with mx we disable ddl propagation */
 	EnsureSequentialModeForTypeDDL();
 
+	/* to prevent recursion with mx we disable ddl propagation */
 	List *commands = list_make3(DISABLE_DDL_PROPAGATION,
 								(void *) renameStmtSql,
 								ENABLE_DDL_PROPAGATION);
@@ -618,7 +618,7 @@ ProcessAlterTypeSchemaStmt(AlterObjectSchemaStmt *stmt, const char *queryString)
 		return;
 	}
 
-	/* dependencies have changed (schema) lets ensure they exist */
+	/* dependencies have changed (schema) let's ensure they exist */
 	EnsureDependenciesExistsOnAllNodes(typeAddress);
 }
 

src/backend/distributed/commands/utility_hook.c

@@ -393,6 +393,12 @@ multi_ProcessUtility(PlannedStmt *pstmt,
 			DropStmt *dropStatement = (DropStmt *) parsetree;
 			switch (dropStatement->removeType)
 			{
+				case OBJECT_COLLATION:
+				{
+					ddlJobs = PlanDropCollationStmt(dropStatement);
+					break;
+				}
+
 				case OBJECT_INDEX:
 				{
 					ddlJobs = PlanDropIndexStmt(dropStatement, queryString);
@@ -475,6 +481,12 @@ multi_ProcessUtility(PlannedStmt *pstmt,
 
 			switch (renameStmt->renameType)
 			{
+				case OBJECT_COLLATION:
+				{
+					ddlJobs = PlanRenameCollationStmt(renameStmt, queryString);
+					break;
+				}
+
 				case OBJECT_TYPE:
 				{
 					ddlJobs = PlanRenameTypeStmt(renameStmt, queryString);
@@ -741,6 +753,16 @@ multi_ProcessUtility(PlannedStmt *pstmt,
 			ProcessCreateEnumStmt(castNode(CreateEnumStmt, parsetree), queryString);
 		}
 
+		if (IsA(parsetree, DefineStmt))
+		{
+			DefineStmt *defineStmt = castNode(DefineStmt, parsetree);
+
+			if (defineStmt->kind == OBJECT_COLLATION)
+			{
+				ddlJobs = ProcessCollationDefineStmt(defineStmt, queryString);
+			}
+		}
+
 		if (IsA(parsetree, AlterObjectSchemaStmt))
 		{
 			ProcessAlterObjectSchemaStmt(castNode(AlterObjectSchemaStmt, parsetree),
@@ -933,6 +955,11 @@ PlanAlterOwnerStmt(AlterOwnerStmt *stmt, const char *queryString)
 {
 	switch (stmt->objectType)
 	{
+		case OBJECT_COLLATION:
+		{
+			return PlanAlterCollationOwnerStmt(stmt, queryString);
+		}
+
 		case OBJECT_TYPE:
 		{
 			return PlanAlterTypeOwnerStmt(stmt, queryString);

src/backend/distributed/deparser/deparse.c

@@ -34,10 +34,8 @@ static char * DeparseAlterObjectDependsStmt(AlterObjectDependsStmt *stmt);
  *  - CREATE TYPE, ALTER TYPE, DROP TYPE
  *  - ALTER FUNCTION, ALTER PROCEDURE, ALTER AGGREGATE
  *  - DROP FUNCTION, DROP PROCEDURE, DROP AGGREGATE
- *
- *  - CREATE EXTENSION
- *  - ALTER EXTENSION
- *  - DROP EXTENSION
+ *  - CREATE EXTENSION, ALTER EXTENSION, DROP EXTENSION
+ *  - ALTER COLLATION, DROP COLLATION
  */
 char *
 DeparseTreeNode(Node *stmt)
@@ -133,6 +131,11 @@ DeparseDropStmt(DropStmt *stmt)
 			return DeparseDropTypeStmt(stmt);
 		}
 
+		case OBJECT_COLLATION:
+		{
+			return DeparseDropCollationStmt(stmt);
+		}
+
 		case OBJECT_PROCEDURE:
 		case OBJECT_AGGREGATE:
 		case OBJECT_FUNCTION:
@@ -205,6 +208,11 @@ DeparseRenameStmt(RenameStmt *stmt)
 			return DeparseRenameAttributeStmt(stmt);
 		}
 
+		case OBJECT_COLLATION:
+		{
+			return DeparseRenameCollationStmt(stmt);
+		}
+
 		case OBJECT_PROCEDURE:
 		case OBJECT_AGGREGATE:
 		case OBJECT_FUNCTION:
@@ -259,6 +267,11 @@ DeparseAlterObjectSchemaStmt(AlterObjectSchemaStmt *stmt)
 			return DeparseAlterTypeSchemaStmt(stmt);
 		}
 
+		case OBJECT_COLLATION:
+		{
+			return DeparseAlterCollationSchemaStmt(stmt);
+		}
+
 		case OBJECT_PROCEDURE:
 		case OBJECT_AGGREGATE:
 		case OBJECT_FUNCTION:
@@ -297,6 +310,11 @@ DeparseAlterOwnerStmt(AlterOwnerStmt *stmt)
 			return DeparseAlterTypeOwnerStmt(stmt);
 		}
 
+		case OBJECT_COLLATION:
+		{
+			return DeparseAlterCollationOwnerStmt(stmt);
+		}
+
 		case OBJECT_PROCEDURE:
 		case OBJECT_AGGREGATE:
 		case OBJECT_FUNCTION:

src/backend/distributed/deparser/deparse_collation_stmts.c

@@ -0,0 +1,177 @@
+/*-------------------------------------------------------------------------
+ *
+ * deparse_collation_stmts.c
+ *	  All routines to deparse collation statements.
+ *	  This file contains all entry points specific for type statement deparsing as well as
+ *	  functions that are currently only used for deparsing of the type statements.
+ *
+ * Copyright (c) Citus Data, Inc.
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "catalog/namespace.h"
+#include "lib/stringinfo.h"
+#include "nodes/value.h"
+#include "utils/builtins.h"
+
+#include "distributed/deparser.h"
+#include "distributed/citus_ruleutils.h"
+
+static void AppendDropCollationStmt(StringInfo buf, DropStmt *stmt);
+static void AppendRenameCollationStmt(StringInfo buf, RenameStmt *stmt);
+static void AppendAlterCollationSchemaStmt(StringInfo buf, AlterObjectSchemaStmt *stmt);
+static void AppendAlterCollationOwnerStmt(StringInfo buf, AlterOwnerStmt *stmt);
+static void AppendNameList(StringInfo buf, List *objects);
+
+
+/*
+ * DeparseDropCollationStmt builds and returns a string representing the DropStmt
+ */
+char *
+DeparseDropCollationStmt(DropStmt *stmt)
+{
+	StringInfoData str = { 0 };
+	initStringInfo(&str);
+
+	Assert(stmt->removeType == OBJECT_COLLATION);
+
+	AppendDropCollationStmt(&str, stmt);
+
+	return str.data;
+}
+
+
+/*
+ * AppendDropCollationStmt appends a string representing the DropStmt to a buffer
+ */
+static void
+AppendDropCollationStmt(StringInfo buf, DropStmt *stmt)
+{
+	appendStringInfoString(buf, "DROP COLLATION ");
+	if (stmt->missing_ok)
+	{
+		appendStringInfoString(buf, "IF EXISTS ");
+	}
+	AppendNameList(buf, stmt->objects);
+	if (stmt->behavior == DROP_CASCADE)
+	{
+		appendStringInfoString(buf, " CASCADE");
+	}
+}
+
+
+/*
+ * DeparseRenameCollationStmt builds and returns a string representing the RenameStmt
+ */
+char *
+DeparseRenameCollationStmt(RenameStmt *stmt)
+{
+	StringInfoData str = { 0 };
+	initStringInfo(&str);
+
+	Assert(stmt->renameType == OBJECT_COLLATION);
+
+	AppendRenameCollationStmt(&str, stmt);
+
+	return str.data;
+}
+
+
+/*
+ * AppendRenameCollationStmt appends a string representing the RenameStmt to a buffer
+ */
+static void
+AppendRenameCollationStmt(StringInfo buf, RenameStmt *stmt)
+{
+	List *names = (List *) stmt->object;
+
+	appendStringInfo(buf, "ALTER COLLATION %s RENAME TO %s;", NameListToQuotedString(
+						 names),
+					 quote_identifier(stmt->newname));
+}
+
+
+/*
+ * DeparseAlterCollationSchemaStmt builds and returns a string representing the AlterObjectSchemaStmt
+ */
+char *
+DeparseAlterCollationSchemaStmt(AlterObjectSchemaStmt *stmt)
+{
+	StringInfoData str = { 0 };
+	initStringInfo(&str);
+
+	Assert(stmt->objectType == OBJECT_COLLATION);
+
+	AppendAlterCollationSchemaStmt(&str, stmt);
+
+	return str.data;
+}
+
+
+/*
+ * AppendAlterCollationSchemaStmt appends a string representing the AlterObjectSchemaStmt to a buffer
+ */
+static void
+AppendAlterCollationSchemaStmt(StringInfo buf, AlterObjectSchemaStmt *stmt)
+{
+	Assert(stmt->objectType == OBJECT_COLLATION);
+
+	List *names = (List *) stmt->object;
+	appendStringInfo(buf, "ALTER COLLATION %s SET SCHEMA %s;", NameListToQuotedString(
+						 names),
+					 quote_identifier(stmt->newschema));
+}
+
+
+/*
+ * DeparseAlterCollationOwnerStmt builds and returns a string representing the AlterOwnerStmt
+ */
+char *
+DeparseAlterCollationOwnerStmt(AlterOwnerStmt *stmt)
+{
+	StringInfoData str = { 0 };
+	initStringInfo(&str);
+
+	Assert(stmt->objectType == OBJECT_COLLATION);
+
+	AppendAlterCollationOwnerStmt(&str, stmt);
+
+	return str.data;
+}
+
+
+/*
+ * AppendAlterCollationOwnerStmt appends a string representing the AlterOwnerStmt to a buffer
+ */
+static void
+AppendAlterCollationOwnerStmt(StringInfo buf, AlterOwnerStmt *stmt)
+{
+	Assert(stmt->objectType == OBJECT_COLLATION);
+
+	List *names = (List *) stmt->object;
+	appendStringInfo(buf, "ALTER COLLATION %s OWNER TO %s;", NameListToQuotedString(
+						 names),
+					 RoleSpecString(stmt->newowner, true));
+}
+
+
+static void
+AppendNameList(StringInfo buf, List *objects)
+{
+	ListCell *objectCell = NULL;
+
+	foreach(objectCell, objects)
+	{
+		List *name = castNode(List, lfirst(objectCell));
+
+		if (objectCell != list_head(objects))
+		{
+			appendStringInfo(buf, ", ");
+		}
+
+		appendStringInfoString(buf, NameListToQuotedString(name));
+	}
+}

src/backend/distributed/deparser/objectaddress.c

@@ -103,9 +103,23 @@ GetObjectAddressFromParseTree(Node *parseTree, bool missing_ok)
 		case T_DefineStmt:
 		{
 			DefineStmt *stmt = castNode(DefineStmt, parseTree);
-			if (stmt->kind == OBJECT_AGGREGATE)
+
+			switch (stmt->kind)
 			{
-				return DefineAggregateStmtObjectAddress(stmt, missing_ok);
+				case OBJECT_AGGREGATE:
+				{
+					return DefineAggregateStmtObjectAddress(stmt, missing_ok);
+				}
+
+				case OBJECT_COLLATION:
+				{
+					return DefineCollationStmtObjectAddress(stmt, missing_ok);
+				}
+
+				default:
+				{
+					break;
+				}
 			}
 
 			ereport(ERROR, (errmsg(
@@ -172,6 +186,11 @@ RenameStmtObjectAddress(RenameStmt *stmt, bool missing_ok)
 			return RenameAttributeStmtObjectAddress(stmt, missing_ok);
 		}
 
+		case OBJECT_COLLATION:
+		{
+			return RenameCollationStmtObjectAddress(stmt, missing_ok);
+		}
+
 		case OBJECT_PROCEDURE:
 		case OBJECT_AGGREGATE:
 		case OBJECT_FUNCTION:
@@ -198,6 +217,11 @@ AlterObjectSchemaStmtObjectAddress(AlterObjectSchemaStmt *stmt, bool missing_ok)
 			return AlterTypeSchemaStmtObjectAddress(stmt, missing_ok);
 		}
 
+		case OBJECT_COLLATION:
+		{
+			return AlterCollationSchemaStmtObjectAddress(stmt, missing_ok);
+		}
+
 		case OBJECT_PROCEDURE:
 		case OBJECT_AGGREGATE:
 		case OBJECT_FUNCTION:
@@ -245,6 +269,13 @@ AlterOwnerStmtObjectAddress(AlterOwnerStmt *stmt, bool missing_ok)
 {
 	switch (stmt->objectType)
 	{
+		case OBJECT_COLLATION:
+		{
+			ObjectAddress *address = palloc(sizeof(ObjectAddress));
+			*address = AlterCollationOwnerObjectAddress(stmt);
+			return address;
+		}
+
 		case OBJECT_TYPE:
 		{
 			return AlterTypeOwnerObjectAddress(stmt, missing_ok);

src/backend/distributed/deparser/qualify.c

@@ -30,6 +30,7 @@ static void QualifyAlterTableStmt(AlterTableStmt *stmt);
 static void QualifyAlterObjectSchemaStmt(AlterObjectSchemaStmt *stmt);
 static void QualifyAlterOwnerStmt(AlterOwnerStmt *stmt);
 static void QualifyAlterObjectDependsStmt(AlterObjectDependsStmt *stmt);
+static void QualifyDropObjectStmt(DropStmt *stmt);
 
 /*
  * QualifyTreeNode transforms the statement in place and makes all (supported) statements
@@ -95,6 +96,12 @@ QualifyTreeNode(Node *stmt)
 			return;
 		}
 
+		case T_DropStmt:
+		{
+			QualifyDropObjectStmt(castNode(DropStmt, stmt));
+			break;
+		}
+
 		default:
 		{
 			/* skip unsupported statements */
@@ -125,6 +132,12 @@ QualifyRenameStmt(RenameStmt *stmt)
 			return;
 		}
 
+		case OBJECT_COLLATION:
+		{
+			QualifyRenameCollationStmt(stmt);
+			return;
+		}
+
 		case OBJECT_AGGREGATE:
 		case OBJECT_FUNCTION:
 		case OBJECT_PROCEDURE:
@@ -198,6 +211,12 @@ QualifyAlterObjectSchemaStmt(AlterObjectSchemaStmt *stmt)
 			return;
 		}
 
+		case OBJECT_COLLATION:
+		{
+			QualifyAlterCollationSchemaStmt(stmt);
+			return;
+		}
+
 		case OBJECT_AGGREGATE:
 		case OBJECT_FUNCTION:
 		case OBJECT_PROCEDURE:
@@ -226,6 +245,12 @@ QualifyAlterOwnerStmt(AlterOwnerStmt *stmt)
 			return;
 		}
 
+		case OBJECT_COLLATION:
+		{
+			QualifyAlterCollationOwnerStmt(stmt);
+			return;
+		}
+
 		case OBJECT_AGGREGATE:
 		case OBJECT_FUNCTION:
 		case OBJECT_PROCEDURE:
@@ -260,3 +285,22 @@ QualifyAlterObjectDependsStmt(AlterObjectDependsStmt *stmt)
 		}
 	}
 }
+
+
+static void
+QualifyDropObjectStmt(DropStmt *stmt)
+{
+	switch (stmt->removeType)
+	{
+		case OBJECT_COLLATION:
+		{
+			QualifyDropCollationStmt(stmt);
+			return;
+		}
+
+		default:
+		{
+			return;
+		}
+	}
+}

src/backend/distributed/deparser/qualify_collation_stmt.c

@@ -0,0 +1,129 @@
+/*-------------------------------------------------------------------------
+ *
+ * qualify_collation_stmt.c
+ *	  Functions specialized in fully qualifying all collation statements. These
+ *	  functions are dispatched from qualify.c
+ *
+ *	  Goal would be that the deparser functions for these statements can
+ *	  serialize the statement without any external lookups.
+ *
+ * Copyright (c), Citus Data, Inc.
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "access/htup_details.h"
+#include "catalog/namespace.h"
+#include "catalog/pg_collation.h"
+#include "utils/lsyscache.h"
+#include "utils/syscache.h"
+
+#include "distributed/deparser.h"
+#include "distributed/listutils.h"
+
+static Node * QualifyCollationName(List *func);
+
+
+/*
+ * QualifyRenameCollationStmt transforms a
+ * ALTER COLLATION .. RENAME TO ..
+ * statement in place and makes the collation name fully qualified.
+ */
+void
+QualifyRenameCollationStmt(RenameStmt *stmt)
+{
+	Assert(stmt->renameType == OBJECT_COLLATION);
+
+	stmt->object = QualifyCollationName(castNode(List, stmt->object));
+}
+
+
+/*
+ * QualifyAlterCollationSchemaStmt transforms a
+ * ALTER COLLATION .. SET SCHEMA ..
+ * statement in place and makes the collation name fully qualified.
+ */
+void
+QualifyAlterCollationSchemaStmt(AlterObjectSchemaStmt *stmt)
+{
+	Assert(stmt->objectType == OBJECT_COLLATION);
+
+	stmt->object = QualifyCollationName(castNode(List, stmt->object));
+}
+
+
+/*
+ * QualifyAlterCollationOwnerStmt transforms a
+ * ALTER COLLATION .. OWNER TO ..
+ * statement in place and makes the collation name fully qualified.
+ */
+void
+QualifyAlterCollationOwnerStmt(AlterOwnerStmt *stmt)
+{
+	Assert(stmt->objectType == OBJECT_COLLATION);
+
+	stmt->object = QualifyCollationName(castNode(List, stmt->object));
+}
+
+
+/*
+ * QualifyDropCollationStmt transforms a
+ * DROP COLLATION ..
+ * statement in place and makes the collation name fully qualified.
+ */
+void
+QualifyDropCollationStmt(DropStmt *stmt)
+{
+	List *names = NIL;
+	List *name = NIL;
+
+	foreach_ptr(name, stmt->objects)
+	{
+		names = lappend(names, QualifyCollationName(name));
+	}
+
+	stmt->objects = names;
+}
+
+
+/*
+ * QualifyCollation transforms a collation in place and makes its name fully qualified.
+ */
+Node *
+QualifyCollationName(List *name)
+{
+	char *collationName = NULL;
+	char *schemaName = NULL;
+
+	/* check if the collation name is already qualified */
+	DeconstructQualifiedName(name, &schemaName, &collationName);
+
+	/* do a lookup for the schema name if the statement does not include one */
+	if (schemaName == NULL)
+	{
+		Oid collid = get_collation_oid(name, true);
+
+		if (collid == InvalidOid)
+		{
+			return (Node *) name;
+		}
+
+		HeapTuple colltup = SearchSysCache1(COLLOID, collid);
+
+		if (!HeapTupleIsValid(colltup))
+		{
+			return (Node *) name;
+		}
+		Form_pg_collation collationForm =
+			(Form_pg_collation) GETSTRUCT(colltup);
+
+		schemaName = get_namespace_name(collationForm->collnamespace);
+		collationName = NameStr(collationForm->collname);
+		name = list_make2(makeString(schemaName), makeString(collationName));
+		ReleaseSysCache(colltup);
+	}
+
+	return (Node *) name;
+}

src/backend/distributed/metadata/dependency.c

@@ -210,7 +210,7 @@ recurse_pg_depend(const ObjectAddress *target,
 	relation_close(depRel, AccessShareLock);
 
 	/*
-	 * concat expended entries if applicable
+	 * concat expanded entries if applicable
 	 */
 	if (expand != NULL)
 	{
@@ -389,6 +389,7 @@ SupportedDependencyByCitus(const ObjectAddress *address)
 	 */
 	switch (getObjectClass(address))
 	{
+		case OCLASS_COLLATION:
 		case OCLASS_SCHEMA:
 		{
 			return true;

src/backend/distributed/worker/worker_create_or_replace.c

@@ -9,7 +9,9 @@
 
 #include "postgres.h"
 
+#include "access/htup_details.h"
 #include "catalog/dependency.h"
+#include "catalog/pg_collation.h"
 #include "catalog/pg_proc.h"
 #include "catalog/pg_type.h"
 #include "fmgr.h"
@@ -19,6 +21,7 @@
 #include "tcop/dest.h"
 #include "tcop/utility.h"
 #include "utils/builtins.h"
+#include "utils/syscache.h"
 #include "utils/lsyscache.h"
 #include "utils/regproc.h"
 
@@ -134,6 +137,11 @@ CreateStmtByObjectAddress(const ObjectAddress *address)
 {
 	switch (getObjectClass(address))
 	{
+		case OCLASS_COLLATION:
+		{
+			return CreateCollationDDL(address->objectId);
+		}
+
 		case OCLASS_PROC:
 		{
 			return GetFunctionDDLCommand(address->objectId, false);
@@ -163,6 +171,11 @@ GenerateBackupNameForCollision(const ObjectAddress *address)
 {
 	switch (getObjectClass(address))
 	{
+		case OCLASS_COLLATION:
+		{
+			return GenerateBackupNameForCollationCollision(address);
+		}
+
 		case OCLASS_PROC:
 		{
 			return GenerateBackupNameForProcCollision(address);
@@ -183,6 +196,39 @@ GenerateBackupNameForCollision(const ObjectAddress *address)
 }
 
 
+/*
+ * CreateRenameTypeStmt creates a rename statement for a type based on its ObjectAddress.
+ * The rename statement will rename the existing object on its address to the value
+ * provided in newName.
+ */
+static RenameStmt *
+CreateRenameCollationStmt(const ObjectAddress *address, char *newName)
+{
+	RenameStmt *stmt = makeNode(RenameStmt);
+	Oid collid = address->objectId;
+
+	HeapTuple colltup = SearchSysCache1(COLLOID, collid);
+	if (!HeapTupleIsValid(colltup))
+	{
+		elog(ERROR, "citus cache lookup error");
+		return NULL;
+	}
+	Form_pg_collation collationForm =
+		(Form_pg_collation) GETSTRUCT(colltup);
+
+	char *schemaName = get_namespace_name(collationForm->collnamespace);
+	char *collationName = NameStr(collationForm->collname);
+	List *name = list_make2(makeString(schemaName), makeString(collationName));
+	ReleaseSysCache(colltup);
+
+	stmt->renameType = OBJECT_COLLATION;
+	stmt->object = (Node *) name;
+	stmt->newname = newName;
+
+	return stmt;
+}
+
+
 /*
  * CreateRenameTypeStmt creates a rename statement for a type based on its ObjectAddress.
  * The rename statement will rename the existing object on its address to the value
@@ -253,6 +299,11 @@ CreateRenameStatement(const ObjectAddress *address, char *newName)
 {
 	switch (getObjectClass(address))
 	{
+		case OCLASS_COLLATION:
+		{
+			return CreateRenameCollationStmt(address, newName);
+		}
+
 		case OCLASS_PROC:
 		{
 			return CreateRenameProcStmt(address, newName);

src/include/distributed/commands.h

@@ -27,6 +27,25 @@ extern List * PlanClusterStmt(ClusterStmt *clusterStmt, const char *clusterComma
 /* call.c */
 extern bool CallDistributedProcedureRemotely(CallStmt *callStmt, DestReceiver *dest);
 
+/* collation.c - forward declarations */
+extern char * CreateCollationDDL(Oid collationId);
+extern List * CreateCollationDDLsIdempotent(Oid collationId);
+extern ObjectAddress AlterCollationOwnerObjectAddress(AlterOwnerStmt *stmt);
+extern List * PlanDropCollationStmt(DropStmt *stmt);
+extern List * PlanAlterCollationOwnerStmt(AlterOwnerStmt *stmt, const char *queryString);
+extern List * PlanAlterCollationSchemaStmt(AlterObjectSchemaStmt *stmt,
+										   const char *queryString);
+extern List * PlanRenameCollationStmt(RenameStmt *stmt, const char *queryString);
+extern ObjectAddress * RenameCollationStmtObjectAddress(RenameStmt *stmt,
+														bool missing_ok);
+extern ObjectAddress * AlterCollationSchemaStmtObjectAddress(AlterObjectSchemaStmt *stmt,
+															 bool missing_ok);
+extern void ProcessAlterCollationSchemaStmt(AlterObjectSchemaStmt *stmt,
+											const char *queryString);
+extern char * GenerateBackupNameForCollationCollision(const ObjectAddress *address);
+extern ObjectAddress * DefineCollationStmtObjectAddress(DefineStmt *stmt, bool
+														missing_ok);
+extern List * ProcessCollationDefineStmt(DefineStmt *stmt, const char *queryString);
 
 /* extension.c - forward declarations */
 extern bool IsCreateAlterExtensionUpdateCitusStmt(Node *parsetree);

src/include/distributed/deparser.h

@@ -32,6 +32,17 @@ extern void AssertObjectTypeIsFunctional(ObjectType type);
 extern void QualifyTreeNode(Node *stmt);
 extern char * DeparseTreeNode(Node *stmt);
 
+/* forward declarations for deparse_collation_stmts.c */
+extern char * DeparseDropCollationStmt(DropStmt *stmt);
+extern char * DeparseRenameCollationStmt(RenameStmt *stmt);
+extern char * DeparseAlterCollationSchemaStmt(AlterObjectSchemaStmt *stmt);
+extern char * DeparseAlterCollationOwnerStmt(AlterOwnerStmt *stmt);
+
+extern void QualifyDropCollationStmt(DropStmt *stmt);
+extern void QualifyRenameCollationStmt(RenameStmt *stmt);
+extern void QualifyAlterCollationSchemaStmt(AlterObjectSchemaStmt *stmt);
+extern void QualifyAlterCollationOwnerStmt(AlterOwnerStmt *stmt);
+
 /* forward declarations for deparse_type_stmts.c */
 extern char * DeparseCompositeTypeStmt(CompositeTypeStmt *stmt);
 extern char * DeparseCreateEnumStmt(CreateEnumStmt *stmt);
@@ -52,8 +63,7 @@ extern void QualifyCreateEnumStmt(CreateEnumStmt *stmt);
 extern void QualifyAlterTypeSchemaStmt(AlterObjectSchemaStmt *stmt);
 extern void QualifyAlterTypeOwnerStmt(AlterOwnerStmt *stmt);
 
-extern ObjectAddress * GetObjectAddressFromParseTree(Node *parseTree, bool
-													 missing_ok);
+extern ObjectAddress * GetObjectAddressFromParseTree(Node *parseTree, bool missing_ok);
 
 /* forward declarations for deparse_function_stmts.c */
 extern char * DeparseDropFunctionStmt(DropStmt *stmt);

src/test/regress/expected/distributed_collations.out

@@ -0,0 +1,107 @@
+SET citus.next_shard_id TO 20050000;
+CREATE USER collationuser;
+NOTICE:  not propagating CREATE ROLE/USER commands to worker nodes
+HINT:  Connect to worker nodes directly to manually create all necessary users and roles.
+SELECT run_command_on_workers($$CREATE USER collationuser;$$);
+      run_command_on_workers       
+-----------------------------------
+ (localhost,57637,t,"CREATE ROLE")
+ (localhost,57638,t,"CREATE ROLE")
+(2 rows)
+
+CREATE SCHEMA collation_tests AUTHORIZATION collationuser;
+CREATE SCHEMA collation_tests2 AUTHORIZATION collationuser;
+SET search_path to collation_tests;
+CREATE COLLATION german_phonebook (provider = icu, locale = 'de-u-co-phonebk');
+SET citus.enable_ddl_propagation TO off;
+CREATE COLLATION german_phonebook_unpropagated (provider = icu, locale = 'de-u-co-phonebk');
+SET citus.enable_ddl_propagation TO on;
+\c - - - :worker_1_port
+SELECT c.collname, nsp.nspname, a.rolname
+FROM pg_collation c
+JOIN pg_namespace nsp ON nsp.oid = c.collnamespace
+JOIN pg_authid a ON a.oid = c.collowner
+WHERE collname like 'german_phonebook%'
+ORDER BY 1,2,3;
+     collname     |     nspname     | rolname  
+------------------+-----------------+----------
+ german_phonebook | collation_tests | postgres
+(1 row)
+
+\c - - - :master_port
+SET search_path to collation_tests;
+CREATE TABLE test_propagate(id int, t1 text COLLATE german_phonebook,
+    t2 text COLLATE german_phonebook_unpropagated);
+INSERT INTO test_propagate VALUES (1, 'aesop', U&'\00E4sop'), (2, U&'Vo\1E9Er', 'Vossr');
+SELECT create_distributed_table('test_propagate', 'id');
+NOTICE:  Copying data from local table...
+ create_distributed_table 
+--------------------------
+ 
+(1 row)
+
+-- Test COLLATE is pushed down
+SELECT * FROM collation_tests.test_propagate WHERE t2 < 'b';
+ id |  t1   |  t2  
+----+-------+------
+  1 | aesop | äsop
+(1 row)
+
+SELECT * FROM collation_tests.test_propagate WHERE t2 < 'b' COLLATE "C";
+ id |  t1  |  t2   
+----+------+-------
+  2 | Voẞr | Vossr
+(1 row)
+
+\c - - - :worker_1_port
+SELECT c.collname, nsp.nspname, a.rolname
+FROM pg_collation c
+JOIN pg_namespace nsp ON nsp.oid = c.collnamespace
+JOIN pg_authid a ON a.oid = c.collowner
+WHERE collname like 'german_phonebook%'
+ORDER BY 1,2,3;
+           collname            |     nspname     | rolname  
+-------------------------------+-----------------+----------
+ german_phonebook              | collation_tests | postgres
+ german_phonebook_unpropagated | collation_tests | postgres
+(2 rows)
+
+\c - - - :master_port
+ALTER COLLATION collation_tests.german_phonebook RENAME TO german_phonebook2;
+ALTER COLLATION collation_tests.german_phonebook2 SET SCHEMA collation_tests2;
+ALTER COLLATION collation_tests2.german_phonebook2 OWNER TO collationuser;
+\c - - - :worker_1_port
+SELECT c.collname, nsp.nspname, a.rolname
+FROM pg_collation c
+JOIN pg_namespace nsp ON nsp.oid = c.collnamespace
+JOIN pg_authid a ON a.oid = c.collowner
+WHERE collname like 'german_phonebook%'
+ORDER BY 1,2,3;
+           collname            |     nspname      |    rolname    
+-------------------------------+------------------+---------------
+ german_phonebook2             | collation_tests2 | collationuser
+ german_phonebook_unpropagated | collation_tests  | postgres
+(2 rows)
+
+\c - - - :master_port
+SET client_min_messages TO error; -- suppress cascading objects dropping
+DROP SCHEMA collation_tests CASCADE;
+DROP SCHEMA collation_tests2 CASCADE;
+-- This is hacky, but we should clean-up the resources as below
+\c - - - :worker_1_port
+SET client_min_messages TO error; -- suppress cascading objects dropping
+DROP SCHEMA collation_tests CASCADE;
+DROP SCHEMA collation_tests2 CASCADE;
+\c - - - :worker_2_port
+SET client_min_messages TO error; -- suppress cascading objects dropping
+DROP SCHEMA collation_tests CASCADE;
+DROP SCHEMA collation_tests2 CASCADE;
+\c - - - :master_port
+DROP USER collationuser;
+SELECT run_command_on_workers($$DROP USER collationuser;$$);
+     run_command_on_workers      
+---------------------------------
+ (localhost,57637,t,"DROP ROLE")
+ (localhost,57638,t,"DROP ROLE")
+(2 rows)
+

src/test/regress/expected/distributed_collations_conflict.out

@@ -0,0 +1,96 @@
+CREATE SCHEMA collation_conflict;
+SELECT run_command_on_workers($$CREATE SCHEMA collation_conflict;$$);
+       run_command_on_workers        
+-------------------------------------
+ (localhost,57637,t,"CREATE SCHEMA")
+ (localhost,57638,t,"CREATE SCHEMA")
+(2 rows)
+
+\c - - - :worker_1_port
+SET search_path TO collation_conflict;
+CREATE COLLATION caseinsensitive (
+	provider = icu,
+	locale = 'und-u-ks-level2'
+);
+\c - - - :master_port
+SET search_path TO collation_conflict;
+CREATE COLLATION caseinsensitive (
+	provider = icu,
+	locale = 'und-u-ks-level2'
+);
+CREATE TABLE tblcoll(val text COLLATE caseinsensitive);
+SELECT create_reference_table('tblcoll');
+ create_reference_table 
+------------------------
+ 
+(1 row)
+
+\c - - - :worker_1_port
+SELECT c.collname, nsp.nspname, a.rolname
+FROM pg_collation c
+JOIN pg_namespace nsp ON nsp.oid = c.collnamespace
+JOIN pg_authid a ON a.oid = c.collowner
+WHERE collname like 'caseinsensitive%'
+ORDER BY 1,2,3;
+    collname     |      nspname       | rolname  
+-----------------+--------------------+----------
+ caseinsensitive | collation_conflict | postgres
+(1 row)
+
+\c - - - :master_port
+SET search_path TO collation_conflict;
+-- Now drop & recreate in order to make sure rename detects the existing renamed objects
+-- hide cascades
+--SET client_min_messages TO error;
+DROP TABLE tblcoll;
+DROP COLLATION caseinsensitive;
+\c - - - :worker_1_port
+SET search_path TO collation_conflict;
+CREATE COLLATION caseinsensitive (
+	provider = icu,
+	locale = 'und-u-ks-level1'
+);
+\c - - - :master_port
+SET search_path TO collation_conflict;
+CREATE COLLATION caseinsensitive (
+	provider = icu,
+	locale = 'und-u-ks-level2'
+);
+CREATE TABLE tblcoll(val text COLLATE caseinsensitive);
+SELECT create_reference_table('tblcoll');
+ create_reference_table 
+------------------------
+ 
+(1 row)
+
+\c - - - :worker_1_port
+SELECT c.collname, nsp.nspname, a.rolname
+FROM pg_collation c
+JOIN pg_namespace nsp ON nsp.oid = c.collnamespace
+JOIN pg_authid a ON a.oid = c.collowner
+WHERE collname like 'caseinsensitive%'
+ORDER BY 1,2,3;
+            collname             |      nspname       | rolname  
+---------------------------------+--------------------+----------
+ caseinsensitive                 | collation_conflict | postgres
+ caseinsensitive(citus_backup_0) | collation_conflict | postgres
+(2 rows)
+
+\c - - - :master_port
+SET search_path TO collation_conflict;
+-- now test worker_create_or_replace_object directly
+SELECT worker_create_or_replace_object($$CREATE COLLATION collation_conflict.caseinsensitive (provider = 'icu', lc_collate = 'und-u-ks-level2', lc_ctype = 'und-u-ks-level2')$$);
+ worker_create_or_replace_object 
+---------------------------------
+ f
+(1 row)
+
+SELECT worker_create_or_replace_object($$CREATE COLLATION collation_conflict.caseinsensitive (provider = 'icu', lc_collate = 'und-u-ks-level2', lc_ctype = 'und-u-ks-level2')$$);
+ worker_create_or_replace_object 
+---------------------------------
+ f
+(1 row)
+
+-- hide cascades
+SET client_min_messages TO error;
+DROP SCHEMA collation_conflict CASCADE;

src/test/regress/expected/multi_mx_create_table.out

@@ -94,8 +94,6 @@ CREATE OPERATOR citus_mx_test_schema.=== (
     NEGATOR = !==,
     HASHES, MERGES
 );
-SET search_path TO public;
-CREATE COLLATION citus_mx_test_schema.english (LOCALE=:current_locale);
 -- now create required stuff in the worker 2
 \c - - - :worker_2_port
 -- create schema to test schema support
@@ -134,8 +132,6 @@ CREATE OPERATOR citus_mx_test_schema.=== (
     NEGATOR = !==,
     HASHES, MERGES
 );
-SET search_path TO public;
-CREATE COLLATION citus_mx_test_schema.english (LOCALE=:current_locale);
 -- connect back to the master, and do some more tests
 \c - - - :master_port
 SET citus.shard_replication_factor TO 1;

src/test/regress/expected/multi_schema_support.out

@@ -402,8 +402,8 @@ SET search_path TO public;
 UPDATE test_schema_support.nation_hash SET n_regionkey = n_regionkey + 1;
 --verify modification
 SELECT * FROM test_schema_support.nation_hash ORDER BY 1,2,3,4;
- n_nationkey |          n_name           | n_regionkey |                                                  n_comment                                                  
--------------+---------------------------+-------------+-------------------------------------------------------------------------------------------------------------
+ n_nationkey |          n_name           | n_regionkey |                                                 n_comment                                                  
+-------------+---------------------------+-------------+------------------------------------------------------------------------------------------------------------
            0 | ALGERIA                   |           1 | haggle. carefully final deposits detect slyly agai
            1 | ARGENTINA                 |           2 | al foxes promise slyly according to the regular accounts. bold requests alon
            2 | BRAZIL                    |           2 | y alongside of the pending deposits. carefully special packages are about the ironic forges. slyly special
@@ -419,8 +419,8 @@ SET search_path TO test_schema_support;
 UPDATE nation_hash SET n_regionkey = n_regionkey + 1;
 --verify modification
 SELECT * FROM nation_hash ORDER BY 1,2,3,4;
- n_nationkey |          n_name           | n_regionkey |                                                  n_comment                                                  
--------------+---------------------------+-------------+-------------------------------------------------------------------------------------------------------------
+ n_nationkey |          n_name           | n_regionkey |                                                 n_comment                                                  
+-------------+---------------------------+-------------+------------------------------------------------------------------------------------------------------------
            0 | ALGERIA                   |           2 | haggle. carefully final deposits detect slyly agai
            1 | ARGENTINA                 |           3 | al foxes promise slyly according to the regular accounts. bold requests alon
            2 | BRAZIL                    |           3 | y alongside of the pending deposits. carefully special packages are about the ironic forges. slyly special
@@ -435,12 +435,6 @@ SELECT * FROM nation_hash ORDER BY 1,2,3,4;
 SET search_path TO public;
 SELECT quote_ident(current_setting('lc_collate')) as current_locale \gset
 CREATE COLLATION test_schema_support.english (LOCALE = :current_locale);
--- create COLLATION in worker node 1 in schema
-\c - - - :worker_1_port
-CREATE COLLATION test_schema_support.english (LOCALE = :current_locale);
--- create COLLATION in worker node 2 in schema
-\c - - - :worker_2_port
-CREATE COLLATION test_schema_support.english (LOCALE = :current_locale);
 \c - - - :master_port
 CREATE TABLE test_schema_support.nation_hash_collation(
     n_nationkey integer not null,
@@ -469,8 +463,8 @@ SELECT master_create_worker_shards('test_schema_support.nation_hash_collation',
 
 \copy test_schema_support.nation_hash_collation FROM STDIN with delimiter '|';
 SELECT * FROM test_schema_support.nation_hash_collation ORDER BY 1,2,3,4;
- n_nationkey |          n_name           | n_regionkey |                                                  n_comment                                                  
--------------+---------------------------+-------------+-------------------------------------------------------------------------------------------------------------
+ n_nationkey |          n_name           | n_regionkey |                                                 n_comment                                                  
+-------------+---------------------------+-------------+------------------------------------------------------------------------------------------------------------
            0 | ALGERIA                   |           0 | haggle. carefully final deposits detect slyly agai
            1 | ARGENTINA                 |           1 | al foxes promise slyly according to the regular accounts. bold requests alon
            2 | BRAZIL                    |           1 | y alongside of the pending deposits. carefully special packages are about the ironic forges. slyly special
@@ -480,8 +474,8 @@ SELECT * FROM test_schema_support.nation_hash_collation ORDER BY 1,2,3,4;
 (6 rows)
 
 SELECT n_comment FROM test_schema_support.nation_hash_collation ORDER BY n_comment COLLATE test_schema_support.english;
-                                                  n_comment                                                  
--------------------------------------------------------------------------------------------------------------
+                                                 n_comment                                                  
+------------------------------------------------------------------------------------------------------------
  al foxes promise slyly according to the regular accounts. bold requests alon
  eas hang ironic, silent packages. slyly regular packages are furiously over the tithes. fluffily bold
  haggle. carefully final deposits detect slyly agai
@@ -512,8 +506,8 @@ SELECT master_create_worker_shards('nation_hash_collation_search_path', 4, 2);
 
 \copy nation_hash_collation_search_path FROM STDIN with delimiter '|';
 SELECT * FROM nation_hash_collation_search_path ORDER BY 1 DESC, 2 DESC, 3 DESC, 4 DESC;
- n_nationkey |          n_name           | n_regionkey |                                                  n_comment                                                  
--------------+---------------------------+-------------+-------------------------------------------------------------------------------------------------------------
+ n_nationkey |          n_name           | n_regionkey |                                                 n_comment                                                  
+-------------+---------------------------+-------------+------------------------------------------------------------------------------------------------------------
            5 | ETHIOPIA                  |           0 | ven packages wake quickly. regu
            4 | EGYPT                     |           4 | y above the carefully unusual theodolites. final dugouts are quickly across the furiously regular d
            3 | CANADA                    |           1 | eas hang ironic, silent packages. slyly regular packages are furiously over the tithes. fluffily bold
@@ -523,8 +517,8 @@ SELECT * FROM nation_hash_collation_search_path ORDER BY 1 DESC, 2 DESC, 3 DESC,
 (6 rows)
 
 SELECT n_comment FROM nation_hash_collation_search_path ORDER BY n_comment COLLATE english;
-                                                  n_comment                                                  
--------------------------------------------------------------------------------------------------------------
+                                                 n_comment                                                  
+------------------------------------------------------------------------------------------------------------
  al foxes promise slyly according to the regular accounts. bold requests alon
  eas hang ironic, silent packages. slyly regular packages are furiously over the tithes. fluffily bold
  haggle. carefully final deposits detect slyly agai

src/test/regress/multi_schedule

@@ -302,6 +302,7 @@ test: ssl_by_default
 # ---------
 test: distributed_types distributed_types_conflict disable_object_propagation distributed_types_xact_add_enum_value
 test: distributed_functions distributed_functions_conflict
+test: distributed_collations distributed_collations_conflict
 test: distributed_procedure
 
 # ---------

src/test/regress/sql/distributed_collations.sql

@@ -0,0 +1,79 @@
+SET citus.next_shard_id TO 20050000;
+
+CREATE USER collationuser;
+SELECT run_command_on_workers($$CREATE USER collationuser;$$);
+
+CREATE SCHEMA collation_tests AUTHORIZATION collationuser;
+CREATE SCHEMA collation_tests2 AUTHORIZATION collationuser;
+
+SET search_path to collation_tests;
+
+CREATE COLLATION german_phonebook (provider = icu, locale = 'de-u-co-phonebk');
+
+SET citus.enable_ddl_propagation TO off;
+
+CREATE COLLATION german_phonebook_unpropagated (provider = icu, locale = 'de-u-co-phonebk');
+
+SET citus.enable_ddl_propagation TO on;
+
+\c - - - :worker_1_port
+SELECT c.collname, nsp.nspname, a.rolname
+FROM pg_collation c
+JOIN pg_namespace nsp ON nsp.oid = c.collnamespace
+JOIN pg_authid a ON a.oid = c.collowner
+WHERE collname like 'german_phonebook%'
+ORDER BY 1,2,3;
+\c - - - :master_port
+SET search_path to collation_tests;
+
+CREATE TABLE test_propagate(id int, t1 text COLLATE german_phonebook,
+    t2 text COLLATE german_phonebook_unpropagated);
+INSERT INTO test_propagate VALUES (1, 'aesop', U&'\00E4sop'), (2, U&'Vo\1E9Er', 'Vossr');
+SELECT create_distributed_table('test_propagate', 'id');
+
+-- Test COLLATE is pushed down
+SELECT * FROM collation_tests.test_propagate WHERE t2 < 'b';
+SELECT * FROM collation_tests.test_propagate WHERE t2 < 'b' COLLATE "C";
+
+\c - - - :worker_1_port
+SELECT c.collname, nsp.nspname, a.rolname
+FROM pg_collation c
+JOIN pg_namespace nsp ON nsp.oid = c.collnamespace
+JOIN pg_authid a ON a.oid = c.collowner
+WHERE collname like 'german_phonebook%'
+ORDER BY 1,2,3;
+\c - - - :master_port
+
+ALTER COLLATION collation_tests.german_phonebook RENAME TO german_phonebook2;
+ALTER COLLATION collation_tests.german_phonebook2 SET SCHEMA collation_tests2;
+ALTER COLLATION collation_tests2.german_phonebook2 OWNER TO collationuser;
+
+\c - - - :worker_1_port
+SELECT c.collname, nsp.nspname, a.rolname
+FROM pg_collation c
+JOIN pg_namespace nsp ON nsp.oid = c.collnamespace
+JOIN pg_authid a ON a.oid = c.collowner
+WHERE collname like 'german_phonebook%'
+ORDER BY 1,2,3;
+\c - - - :master_port
+
+SET client_min_messages TO error; -- suppress cascading objects dropping
+DROP SCHEMA collation_tests CASCADE;
+DROP SCHEMA collation_tests2 CASCADE;
+
+-- This is hacky, but we should clean-up the resources as below
+
+\c - - - :worker_1_port
+SET client_min_messages TO error; -- suppress cascading objects dropping
+DROP SCHEMA collation_tests CASCADE;
+DROP SCHEMA collation_tests2 CASCADE;
+
+\c - - - :worker_2_port
+SET client_min_messages TO error; -- suppress cascading objects dropping
+DROP SCHEMA collation_tests CASCADE;
+DROP SCHEMA collation_tests2 CASCADE;
+
+\c - - - :master_port
+
+DROP USER collationuser;
+SELECT run_command_on_workers($$DROP USER collationuser;$$);

src/test/regress/sql/distributed_collations_conflict.sql

@@ -0,0 +1,73 @@
+CREATE SCHEMA collation_conflict;
+SELECT run_command_on_workers($$CREATE SCHEMA collation_conflict;$$);
+
+\c - - - :worker_1_port
+SET search_path TO collation_conflict;
+
+CREATE COLLATION caseinsensitive (
+	provider = icu,
+	locale = 'und-u-ks-level2'
+);
+
+\c - - - :master_port
+SET search_path TO collation_conflict;
+
+CREATE COLLATION caseinsensitive (
+	provider = icu,
+	locale = 'und-u-ks-level2'
+);
+CREATE TABLE tblcoll(val text COLLATE caseinsensitive);
+SELECT create_reference_table('tblcoll');
+
+\c - - - :worker_1_port
+SELECT c.collname, nsp.nspname, a.rolname
+FROM pg_collation c
+JOIN pg_namespace nsp ON nsp.oid = c.collnamespace
+JOIN pg_authid a ON a.oid = c.collowner
+WHERE collname like 'caseinsensitive%'
+ORDER BY 1,2,3;
+\c - - - :master_port
+SET search_path TO collation_conflict;
+
+-- Now drop & recreate in order to make sure rename detects the existing renamed objects
+-- hide cascades
+--SET client_min_messages TO error;
+DROP TABLE tblcoll;
+DROP COLLATION caseinsensitive;
+
+\c - - - :worker_1_port
+SET search_path TO collation_conflict;
+
+CREATE COLLATION caseinsensitive (
+	provider = icu,
+	locale = 'und-u-ks-level1'
+);
+
+\c - - - :master_port
+SET search_path TO collation_conflict;
+
+CREATE COLLATION caseinsensitive (
+	provider = icu,
+	locale = 'und-u-ks-level2'
+);
+CREATE TABLE tblcoll(val text COLLATE caseinsensitive);
+SELECT create_reference_table('tblcoll');
+
+\c - - - :worker_1_port
+SELECT c.collname, nsp.nspname, a.rolname
+FROM pg_collation c
+JOIN pg_namespace nsp ON nsp.oid = c.collnamespace
+JOIN pg_authid a ON a.oid = c.collowner
+WHERE collname like 'caseinsensitive%'
+ORDER BY 1,2,3;
+\c - - - :master_port
+SET search_path TO collation_conflict;
+
+-- now test worker_create_or_replace_object directly
+SELECT worker_create_or_replace_object($$CREATE COLLATION collation_conflict.caseinsensitive (provider = 'icu', lc_collate = 'und-u-ks-level2', lc_ctype = 'und-u-ks-level2')$$);
+SELECT worker_create_or_replace_object($$CREATE COLLATION collation_conflict.caseinsensitive (provider = 'icu', lc_collate = 'und-u-ks-level2', lc_ctype = 'und-u-ks-level2')$$);
+
+-- hide cascades
+SET client_min_messages TO error;
+DROP SCHEMA collation_conflict CASCADE;
+

src/test/regress/sql/multi_mx_create_table.sql

@@ -100,9 +100,6 @@ CREATE OPERATOR citus_mx_test_schema.=== (
     HASHES, MERGES
 );
 
-SET search_path TO public;
-CREATE COLLATION citus_mx_test_schema.english (LOCALE=:current_locale);
-
 -- now create required stuff in the worker 2
 \c - - - :worker_2_port
 
@@ -148,10 +145,6 @@ CREATE OPERATOR citus_mx_test_schema.=== (
     HASHES, MERGES
 );
 
-
-SET search_path TO public;
-CREATE COLLATION citus_mx_test_schema.english (LOCALE=:current_locale);
-
 -- connect back to the master, and do some more tests
 \c - - - :master_port
 

src/test/regress/sql/multi_schema_support.sql

@@ -322,14 +322,6 @@ SET search_path TO public;
 SELECT quote_ident(current_setting('lc_collate')) as current_locale \gset
 CREATE COLLATION test_schema_support.english (LOCALE = :current_locale);
 
--- create COLLATION in worker node 1 in schema
-\c - - - :worker_1_port
-CREATE COLLATION test_schema_support.english (LOCALE = :current_locale);
-
--- create COLLATION in worker node 2 in schema
-\c - - - :worker_2_port
-CREATE COLLATION test_schema_support.english (LOCALE = :current_locale);
-
 \c - - - :master_port
 
 CREATE TABLE test_schema_support.nation_hash_collation(