DDL support for collations

- collowner field
- CREATE COLLATION
- ALTER COLLATION
- DROP COLLATION
- COMMENT ON COLLATION
- integration with extensions
- pg_dump support for the above
- dependency management
- psql tab completion
- psql \dO command

doc/src/sgml/catalogs.sgml

@@ -2114,11 +2114,30 @@
       </entry>
      </row>
 
+     <row>
+      <entry><structfield>collowner</structfield></entry>
+      <entry><type>oid</type></entry>
+      <entry><literal><link linkend="catalog-pg-authid"><structname>pg_authid</structname></link>.oid</literal></entry>
+      <entry>Owner of the collation</entry>
+     </row>
+
      <row>
       <entry><structfield>collencoding</structfield></entry>
       <entry><type>int4</type></entry>
       <entry></entry>
-      <entry>Encoding to which the collation is applicable</entry>
+      <entry>
+       Encoding to which the collation is applicable.  SQL-level
+       commands such as <command>ALTER COLLATION</command> only
+       operate on the collation belonging to the current database
+       encoding.  But this field is necessary because when this
+       catalog is initialized, the encoding of future databases is not
+       yet known.  For practical purposes, collations that do not
+       match the current database encoding should be considered
+       invalid or invisible.  It could be useful, however, to create
+       collations whose encoding does not match the database encoding
+       in template databases.  This would currently have to be done
+       manually.
+      </entry>
      </row>
 
      <row>

doc/src/sgml/charset.sgml

@@ -459,11 +459,12 @@ SELECT a || ('foo' COLLATE "y") FROM test1;
 
    <para>
     In case a collation is needed that has different values for
-    <symbol>LC_COLLATE</symbol> and <symbol>LC_CTYPE</symbol>, or a
+    <symbol>LC_COLLATE</symbol> and <symbol>LC_CTYPE</symbol>, a new
-    different name is needed for a collation (for example, for
+    collation may be created using
-    compatibility with existing applications), a new collation may be
+    the <xref linkend="sql-createcollation"> command.  That command
-    created.  But there is currently no SQL-level support for creating
+    can also be used to create a new collation from an existing
-    or changing collations.
+    collation, which can be useful to be able to use operating-system
+    independent collation names in applications.
    </para>
   </sect2>
  </sect1>

doc/src/sgml/ref/allfiles.sgml

@@ -7,6 +7,7 @@ Complete list of usable sgml source files in this directory.
 <!-- SQL commands -->
 <!entity abort              system "abort.sgml">
 <!entity alterAggregate     system "alter_aggregate.sgml">
+<!entity alterCollation     system "alter_collation.sgml">
 <!entity alterConversion    system "alter_conversion.sgml">
 <!entity alterDatabase      system "alter_database.sgml">
 <!entity alterDefaultPrivileges system "alter_default_privileges.sgml">
@@ -48,6 +49,7 @@ Complete list of usable sgml source files in this directory.
 <!entity copyTable          system "copy.sgml">
 <!entity createAggregate    system "create_aggregate.sgml">
 <!entity createCast         system "create_cast.sgml">
+<!entity createCollation    system "create_collation.sgml">
 <!entity createConversion   system "create_conversion.sgml">
 <!entity createDatabase     system "create_database.sgml">
 <!entity createDomain       system "create_domain.sgml">
@@ -85,6 +87,7 @@ Complete list of usable sgml source files in this directory.
 <!entity do                 system "do.sgml">
 <!entity dropAggregate      system "drop_aggregate.sgml">
 <!entity dropCast           system "drop_cast.sgml">
+<!entity dropCollation      system "drop_collation.sgml">
 <!entity dropConversion     system "drop_conversion.sgml">
 <!entity dropDatabase       system "drop_database.sgml">
 <!entity dropDomain         system "drop_domain.sgml">

doc/src/sgml/ref/alter_collation.sgml

@@ -0,0 +1,128 @@
+<!--
+doc/src/sgml/ref/alter_collation.sgml
+PostgreSQL documentation
+-->
+
+<refentry id="SQL-ALTERCOLLATION">
+ <refmeta>
+  <refentrytitle>ALTER COLLATION</refentrytitle>
+  <manvolnum>7</manvolnum>
+  <refmiscinfo>SQL - Language Statements</refmiscinfo>
+ </refmeta>
+
+ <refnamediv>
+  <refname>ALTER COLLATION</refname>
+  <refpurpose>change the definition of a collation</refpurpose>
+ </refnamediv>
+
+ <indexterm zone="sql-altercollation">
+  <primary>ALTER COLLATION</primary>
+ </indexterm>
+
+ <refsynopsisdiv>
+<synopsis>
+ALTER COLLATION <replaceable>name</replaceable> RENAME TO <replaceable>new_name</replaceable>
+ALTER COLLATION <replaceable>name</replaceable> OWNER TO <replaceable>new_owner</replaceable>
+ALTER COLLATION <replaceable>name</replaceable> SET SCHEMA <replaceable>new_schema</replaceable>
+</synopsis>
+ </refsynopsisdiv>
+
+ <refsect1>
+  <title>Description</title>
+
+  <para>
+   <command>ALTER COLLATION</command> changes the definition of a
+   collation.
+  </para>
+
+  <para>
+   You must own the collation to use <command>ALTER COLLATION</>.
+   To alter the owner, you must also be a direct or indirect member of the new
+   owning role, and that role must have <literal>CREATE</literal> privilege on
+   the collation's schema.  (These restrictions enforce that altering the
+   owner doesn't do anything you couldn't do by dropping and recreating the
+   collation. However, a superuser can alter ownership of any collation
+   anyway.)
+  </para>
+ </refsect1>
+
+ <refsect1>
+  <title>Parameters</title>
+
+  <variablelist>
+   <varlistentry>
+    <term><replaceable class="parameter">name</replaceable></term>
+    <listitem>
+     <para>
+      The name (optionally schema-qualified) of an existing collation.
+     </para>
+    </listitem>
+   </varlistentry>
+
+   <varlistentry>
+    <term><replaceable class="parameter">new_name</replaceable></term>
+    <listitem>
+     <para>
+      The new name of the collation.
+     </para>
+    </listitem>
+   </varlistentry>
+
+   <varlistentry>
+    <term><replaceable class="parameter">new_owner</replaceable></term>
+    <listitem>
+     <para>
+      The new owner of the collation.
+     </para>
+    </listitem>
+   </varlistentry>
+
+   <varlistentry>
+    <term><replaceable class="parameter">new_schema</replaceable></term>
+    <listitem>
+     <para>
+      The new schema for the collation.
+     </para>
+    </listitem>
+   </varlistentry>
+  </variablelist>
+ </refsect1>
+
+ <refsect1>
+  <title>Examples</title>
+
+  <para>
+   To rename the collation <literal>de_DE</literal> to
+   <literal>german</literal>:
+<programlisting>
+ALTER COLLATION "de_DE" RENAME TO german;
+</programlisting>
+  </para>
+
+  <para>
+   To change the owner of the collation <literal>en_US</literal> to
+   <literal>joe</literal>:
+<programlisting>
+ALTER COLLATION "en_US" OWNER TO joe;
+</programlisting>
+  </para>
+ </refsect1>
+
+ <refsect1>
+  <title>Compatibility</title>
+
+  <para>
+   There is no <command>ALTER COLLATION</command> statement in the SQL
+   standard.
+  </para>
+ </refsect1>
+
+ <refsect1>
+  <title>See Also</title>
+
+  <simplelist type="inline">
+   <member><xref linkend="sql-createcollation"></member>
+   <member><xref linkend="sql-dropcollation"></member>
+  </simplelist>
+ </refsect1>
+</refentry>

doc/src/sgml/ref/alter_extension.sgml

@@ -32,6 +32,7 @@ ALTER EXTENSION <replaceable class="PARAMETER">extension_name</replaceable> DROP
 
   AGGREGATE <replaceable class="PARAMETER">agg_name</replaceable> (<replaceable class="PARAMETER">agg_type</replaceable> [, ...] ) |
   CAST (<replaceable>source_type</replaceable> AS <replaceable>target_type</replaceable>) |
+  COLLATION <replaceable class="PARAMETER">object_name</replaceable> |
   CONVERSION <replaceable class="PARAMETER">object_name</replaceable> |
   DOMAIN <replaceable class="PARAMETER">object_name</replaceable> |
   FOREIGN DATA WRAPPER <replaceable class="PARAMETER">object_name</replaceable> |

doc/src/sgml/ref/comment.sgml

@@ -27,6 +27,7 @@ COMMENT ON
   COLUMN <replaceable class="PARAMETER">table_name</replaceable>.<replaceable class="PARAMETER">column_name</replaceable> |
   AGGREGATE <replaceable class="PARAMETER">agg_name</replaceable> (<replaceable class="PARAMETER">agg_type</replaceable> [, ...] ) |
   CAST (<replaceable>source_type</replaceable> AS <replaceable>target_type</replaceable>) |
+  COLLATION <replaceable class="PARAMETER">object_name</replaceable> |
   CONSTRAINT <replaceable class="PARAMETER">constraint_name</replaceable> ON <replaceable class="PARAMETER">table_name</replaceable> |
   CONVERSION <replaceable class="PARAMETER">object_name</replaceable> |
   DATABASE <replaceable class="PARAMETER">object_name</replaceable> |
@@ -245,6 +246,7 @@ COMMENT ON TABLE mytable IS NULL;
 <programlisting>
 COMMENT ON AGGREGATE my_aggregate (double precision) IS 'Computes sample variance';
 COMMENT ON CAST (text AS int4) IS 'Allow casts from text to int4';
+COMMENT ON COLLATION "fr_CA" IS 'Canadian French';
 COMMENT ON COLUMN my_table.my_column IS 'Employee ID number';
 COMMENT ON CONVERSION my_conv IS 'Conversion to UTF8';
 COMMENT ON DATABASE my_database IS 'Development Database';

doc/src/sgml/ref/create_collation.sgml

@@ -0,0 +1,175 @@
+<!-- doc/src/sgml/ref/create_collation.sgml -->
+
+<refentry id="SQL-CREATECOLLATION">
+ <refmeta>
+  <refentrytitle>CREATE COLLATION</refentrytitle>
+  <manvolnum>7</manvolnum>
+  <refmiscinfo>SQL - Language Statements</refmiscinfo>
+ </refmeta>
+
+ <refnamediv>
+  <refname>CREATE COLLATION</refname>
+  <refpurpose>define a new collation</refpurpose>
+ </refnamediv>
+
+ <indexterm zone="sql-createcollation">
+  <primary>CREATE COLLATION</primary>
+ </indexterm>
+
+ <refsynopsisdiv>
+<synopsis>
+CREATE COLLATION <replaceable>name</replaceable> (
+    [ LOCALE = <replaceable>locale</replaceable>, ]
+    [ LC_COLLATE = <replaceable>lc_collate</replaceable>, ]
+    [ LC_CTYPE = <replaceable>lc_ctype</replaceable>, ]
+)
+CREATE COLLATION <replaceable>name</replaceable> FROM <replaceable>existing_collation</replaceable>
+</synopsis>
+ </refsynopsisdiv>
+
+ <refsect1 id="sql-createcollation-description">
+  <title>Description</title>
+
+  <para>
+   <command>CREATE COLLATION</command> defines a new collation using
+   the specified operating system locales or from an existing collation.
+ </para>
+
+  <para>
+   To be able to create a collation, you must
+   have <literal>CREATE</literal> privilege on the destination schema.
+  </para>
+ </refsect1>
+
+
+ <refsect1>
+  <title>Parameters</title>
+
+   <variablelist>
+    <varlistentry>
+     <term><replaceable>name</replaceable></term>
+
+     <listitem>
+      <para>
+       The name of the collation. The collation name can be
+       schema-qualified. If it is not, the collation is defined in the
+       current schema. The collation name must be unique within a
+       schema.  (The system catalogs can contain collations with the
+       same name for other encodings, but these are not usable if the
+       database encoding does not match.)
+      </para>
+     </listitem>
+    </varlistentry>
+
+    <varlistentry>
+     <term><replaceable>existing_collation</replaceable></term>
+
+     <listitem>
+      <para>
+       The name of an existing collation to copy.  The new collation
+       will have the same properties as the existing one, but they
+       will become independent objects.
+      </para>
+     </listitem>
+    </varlistentry>
+
+    <varlistentry>
+     <term><replaceable>locale</replaceable></term>
+
+     <listitem>
+      <para>
+       This is a shortcut for setting <symbol>LC_COLLATE</symbol>
+       and <symbol>LC_CTYPE</symbol> at once.  If you specify this,
+       you cannot specify either of the other parameters.
+      </para>
+     </listitem>
+    </varlistentry>
+
+    <varlistentry>
+     <term><replaceable>lc_collate</replaceable></term>
+
+     <listitem>
+      <para>
+       Use the specified operating system locale for
+       the <symbol>LC_COLLATE</symbol> locale category.  The locale
+       must be applicable to the current database encoding.
+       (See <xref linkend="sql-createdatabase"> for the precise
+       rules.)
+      </para>
+     </listitem>
+    </varlistentry>
+
+    <varlistentry>
+     <term><replaceable>lc_ctype</replaceable></term>
+
+     <listitem>
+      <para>
+       Use the specified operating system locale for
+       the <symbol>LC_CTYPE</symbol> locale category.  The locale
+       must be applicable to the current database encoding.
+       (See <xref linkend="sql-createdatabase"> for the precise
+       rules.)
+      </para>
+     </listitem>
+    </varlistentry>
+   </variablelist>
+ </refsect1>
+
+
+ <refsect1 id="sql-createcollation-notes">
+  <title>Notes</title>
+
+  <para>
+   Use <command>DROP COLLATION</command> to remove user-defined collations.
+  </para>
+
+  <para>
+   See <xref linkend="collation"> for more information about collation
+   support in PostgreSQL.
+  </para>
+ </refsect1>
+
+ <refsect1 id="sql-createcollation-examples">
+  <title>Examples</title>
+
+  <para>
+   To create a collation from the locale <literal>fr_FR.utf8</literal>
+   (assuming the current database encoding is <literal>UTF8</literal>):
+<programlisting>
+CREATE COLLATION french (LOCALE = 'fr_FR.utf8');
+</programlisting>
+  </para>
+
+  <para>
+   To create a collation from an existing collation:
+<programlisting>
+CREATE COLLATION german FROM "de_DE";
+</programlisting>
+   This can be convenient to be able to use operating-system
+   independent collation names in applications.
+  </para>
+ </refsect1>
+
+
+ <refsect1 id="sql-createcollation-compat">
+  <title>Compatibility</title>
+
+  <para>
+   There is a <command>CREATE COLLATION</command> statement in the SQL
+   standard, but it is limited to copying an existing collation.  The
+   syntax to create a new collation is
+   a <productname>PostgreSQL</productname> extension.
+  </para>
+ </refsect1>
+
+
+ <refsect1 id="sql-createcollation-seealso">
+  <title>See Also</title>
+
+  <simplelist type="inline">
+   <member><xref linkend="sql-altercollation"></member>
+   <member><xref linkend="sql-dropcollation"></member>
+  </simplelist>
+ </refsect1>
+
+</refentry>

doc/src/sgml/ref/drop_collation.sgml

@@ -0,0 +1,110 @@
+<!-- doc/src/sgml/ref/drop_collation.sgml -->
+
+<refentry id="SQL-DROPCOLLATION">
+ <refmeta>
+  <refentrytitle>DROP COLLATION</refentrytitle>
+  <manvolnum>7</manvolnum>
+  <refmiscinfo>SQL - Language Statements</refmiscinfo>
+ </refmeta>
+
+ <refnamediv>
+  <refname>DROP COLLATION</refname>
+  <refpurpose>remove a collation</refpurpose>
+ </refnamediv>
+
+ <indexterm zone="sql-dropcollation">
+  <primary>DROP COLLATION</primary>
+ </indexterm>
+
+ <refsynopsisdiv>
+<synopsis>
+DROP COLLATION [ IF EXISTS ] <replaceable>name</replaceable> [ CASCADE | RESTRICT ]
+</synopsis>
+ </refsynopsisdiv>
+
+ <refsect1 id="sql-dropcollation-description">
+  <title>Description</title>
+
+  <para>
+   <command>DROP COLLATION</command> removes a previously defined collation.
+   To be able to drop a collation, you must own the collation.
+  </para>
+ </refsect1>
+
+ <refsect1>
+  <title>Parameters</title>
+
+   <variablelist>
+    <varlistentry>
+     <term><literal>IF EXISTS</literal></term>
+     <listitem>
+      <para>
+       Do not throw an error if the collation does not exist.
+       A notice is issued in this case.
+      </para>
+     </listitem>
+    </varlistentry>
+
+    <varlistentry>
+     <term><replaceable>name</replaceable></term>
+
+     <listitem>
+      <para>
+       The name of the collation. The collation name can be
+       schema-qualified.
+      </para>
+     </listitem>
+    </varlistentry>
+
+    <varlistentry>
+     <term><literal>CASCADE</literal></term>
+     <listitem>
+      <para>
+       Automatically drop objects that depend on the collation.
+      </para>
+     </listitem>
+    </varlistentry>
+
+    <varlistentry>
+     <term><literal>RESTRICT</literal></term>
+     <listitem>
+      <para>
+       Refuse to drop the collation if any objects depend on it.  This
+       is the default.
+      </para>
+     </listitem>
+    </varlistentry>
+   </variablelist>
+ </refsect1>
+
+ <refsect1 id="sql-dropcollation-examples">
+  <title>Examples</title>
+
+  <para>
+   To drop the collation named <literal>german</>:
+<programlisting>
+DROP COLLATION german;
+</programlisting>
+  </para>
+ </refsect1>
+
+ <refsect1 id="sql-dropcollation-compat">
+  <title>Compatibility</title>
+
+  <para>
+   The <command>DROP COLLATION</command> command conforms to the
+   <acronym>SQL</acronym> standard, apart from the <literal>IF
+   EXISTS</> option, which is a <productname>PostgreSQL</> extension..
+  </para>
+ </refsect1>
+
+ <refsect1>
+  <title>See Also</title>
+
+  <simplelist type="inline">
+   <member><xref linkend="sql-altercollation"></member>
+   <member><xref linkend="sql-createcollation"></member>
+  </simplelist>
+ </refsect1>
+
+</refentry>

doc/src/sgml/ref/psql-ref.sgml

@@ -1265,6 +1265,7 @@ testdb=>
         </listitem>
       </varlistentry>
 
+
       <varlistentry>
         <term><literal>\dn[S+] [ <link linkend="APP-PSQL-patterns"><replaceable class="parameter">pattern</replaceable></link> ]</literal></term>
 
@@ -1297,6 +1298,24 @@ testdb=&gt;
       </varlistentry>
 
 
+      <varlistentry>
+        <term><literal>\dO[S+] [ <link linkend="APP-PSQL-patterns"><replaceable class="parameter">pattern</replaceable></link> ]</literal></term>
+
+        <listitem>
+        <para>
+        Lists collations.
+        If <replaceable class="parameter">pattern</replaceable> is
+        specified, only collations whose names match the pattern are
+        listed.  By default, only user-created objects are shown;
+        supply a pattern or the <literal>S</literal> modifier to
+        include system objects.  If <literal>+</literal> is appended
+        to the command name, each object is listed with its associated
+        description, if any.
+        </para>
+        </listitem>
+      </varlistentry>
+
+
       <varlistentry>
         <term><literal>\dp [ <link linkend="APP-PSQL-patterns"><replaceable class="parameter">pattern</replaceable></link> ]</literal></term>
         <listitem>

doc/src/sgml/reference.sgml

@@ -35,6 +35,7 @@
 
    &abort;
    &alterAggregate;
+   &alterCollation;
    &alterConversion;
    &alterDatabase;
    &alterDefaultPrivileges;
@@ -76,6 +77,7 @@
    &copyTable;
    &createAggregate;
    &createCast;
+   &createCollation;
    &createConversion;
    &createDatabase;
    &createDomain;
@@ -113,6 +115,7 @@
    &do;
    &dropAggregate;
    &dropCast;
+   &dropCollation;
    &dropConversion;
    &dropDatabase;
    &dropDomain;

src/backend/catalog/Makefile

@@ -11,7 +11,7 @@ top_builddir = ../../..
 include $(top_builddir)/src/Makefile.global
 
 OBJS = catalog.o dependency.o heap.o index.o indexing.o namespace.o aclchk.o \
-       objectaddress.o pg_aggregate.o pg_constraint.o pg_conversion.o \
+       objectaddress.o pg_aggregate.o pg_collation.o pg_constraint.o pg_conversion.o \
        pg_depend.o pg_enum.o pg_inherits.o pg_largeobject.o pg_namespace.o \
        pg_operator.o pg_proc.o pg_db_role_setting.o pg_shdepend.o pg_type.o \
        storage.o toasting.o

src/backend/catalog/aclchk.c

@@ -25,6 +25,7 @@
 #include "catalog/dependency.h"
 #include "catalog/indexing.h"
 #include "catalog/pg_authid.h"
+#include "catalog/pg_collation.h"
 #include "catalog/pg_conversion.h"
 #include "catalog/pg_database.h"
 #include "catalog/pg_default_acl.h"
@@ -3131,6 +3132,8 @@ static const char *const no_priv_msg[MAX_ACL_KIND] =
 	gettext_noop("permission denied for operator class %s"),
 	/* ACL_KIND_OPFAMILY */
 	gettext_noop("permission denied for operator family %s"),
+	/* ACL_KIND_COLLATION */
+	gettext_noop("permission denied for collation %s"),
 	/* ACL_KIND_CONVERSION */
 	gettext_noop("permission denied for conversion %s"),
 	/* ACL_KIND_TABLESPACE */
@@ -3173,6 +3176,8 @@ static const char *const not_owner_msg[MAX_ACL_KIND] =
 	gettext_noop("must be owner of operator class %s"),
 	/* ACL_KIND_OPFAMILY */
 	gettext_noop("must be owner of operator family %s"),
+	/* ACL_KIND_COLLATION */
+	gettext_noop("must be owner of collation %s"),
 	/* ACL_KIND_CONVERSION */
 	gettext_noop("must be owner of conversion %s"),
 	/* ACL_KIND_TABLESPACE */
@@ -4631,6 +4636,32 @@ pg_database_ownercheck(Oid db_oid, Oid roleid)
 	return has_privs_of_role(roleid, dba);
 }
 
+/*
+ * Ownership check for a collation (specified by OID).
+ */
+bool
+pg_collation_ownercheck(Oid coll_oid, Oid roleid)
+{
+	HeapTuple	tuple;
+	Oid			ownerId;
+
+	/* Superusers bypass all permission checking. */
+	if (superuser_arg(roleid))
+		return true;
+
+	tuple = SearchSysCache1(COLLOID, ObjectIdGetDatum(coll_oid));
+	if (!HeapTupleIsValid(tuple))
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("collation with OID %u does not exist", coll_oid)));
+
+	ownerId = ((Form_pg_collation) GETSTRUCT(tuple))->collowner;
+
+	ReleaseSysCache(tuple);
+
+	return has_privs_of_role(roleid, ownerId);
+}
+
 /*
  * Ownership check for a conversion (specified by OID).
  */

src/backend/catalog/dependency.c

@@ -28,6 +28,8 @@
 #include "catalog/pg_attrdef.h"
 #include "catalog/pg_authid.h"
 #include "catalog/pg_cast.h"
+#include "catalog/pg_collation.h"
+#include "catalog/pg_collation_fn.h"
 #include "catalog/pg_constraint.h"
 #include "catalog/pg_conversion.h"
 #include "catalog/pg_conversion_fn.h"
@@ -133,6 +135,7 @@ static const Oid object_classes[MAX_OCLASS] = {
 	ProcedureRelationId,		/* OCLASS_PROC */
 	TypeRelationId,				/* OCLASS_TYPE */
 	CastRelationId,				/* OCLASS_CAST */
+	CollationRelationId,		/* OCLASS_COLLATION */
 	ConstraintRelationId,		/* OCLASS_CONSTRAINT */
 	ConversionRelationId,		/* OCLASS_CONVERSION */
 	AttrDefaultRelationId,		/* OCLASS_DEFAULT */
@@ -1075,6 +1078,10 @@ doDeletion(const ObjectAddress *object)
 			DropCastById(object->objectId);
 			break;
 
+		case OCLASS_COLLATION:
+			RemoveCollationById(object->objectId);
+			break;
+
 		case OCLASS_CONSTRAINT:
 			RemoveConstraintById(object->objectId);
 			break;
@@ -1417,6 +1424,9 @@ find_expr_references_walker(Node *node,
 		/* A constant must depend on the constant's datatype */
 		add_object_address(OCLASS_TYPE, con->consttype, 0,
 						   context->addrs);
+		if (OidIsValid(con->constcollid))
+			add_object_address(OCLASS_COLLATION, con->constcollid, 0,
+							   context->addrs);
 
 		/*
 		 * If it's a regclass or similar literal referring to an existing
@@ -1483,6 +1493,9 @@ find_expr_references_walker(Node *node,
 		/* A parameter must depend on the parameter's datatype */
 		add_object_address(OCLASS_TYPE, param->paramtype, 0,
 						   context->addrs);
+		if (OidIsValid(param->paramcollation))
+			add_object_address(OCLASS_COLLATION, param->paramcollation, 0,
+							   context->addrs);
 	}
 	else if (IsA(node, FuncExpr))
 	{
@@ -1553,6 +1566,13 @@ find_expr_references_walker(Node *node,
 		add_object_address(OCLASS_TYPE, relab->resulttype, 0,
 						   context->addrs);
 	}
+	else if (IsA(node, CollateClause))
+	{
+		CollateClause *coll = (CollateClause *) node;
+
+		add_object_address(OCLASS_COLLATION, coll->collOid, 0,
+						   context->addrs);
+	}
 	else if (IsA(node, CoerceViaIO))
 	{
 		CoerceViaIO *iocoerce = (CoerceViaIO *) node;
@@ -1653,6 +1673,14 @@ find_expr_references_walker(Node *node,
 						add_object_address(OCLASS_TYPE, lfirst_oid(ct), 0,
 										   context->addrs);
 					}
+					foreach(ct, rte->funccolcollations)
+					{
+						Oid collid = lfirst_oid(ct);
+
+						if (OidIsValid(collid))
+							add_object_address(OCLASS_COLLATION, collid, 0,
+											   context->addrs);
+					}
 					break;
 				default:
 					break;
@@ -2019,6 +2047,9 @@ getObjectClass(const ObjectAddress *object)
 		case CastRelationId:
 			return OCLASS_CAST;
 
+		case CollationRelationId:
+			return OCLASS_COLLATION;
+
 		case ConstraintRelationId:
 			return OCLASS_CONSTRAINT;
 
@@ -2167,6 +2198,21 @@ getObjectDescription(const ObjectAddress *object)
 				break;
 			}
 
+		case OCLASS_COLLATION:
+			{
+				HeapTuple	collTup;
+
+				collTup = SearchSysCache1(COLLOID,
+										 ObjectIdGetDatum(object->objectId));
+				if (!HeapTupleIsValid(collTup))
+					elog(ERROR, "cache lookup failed for collation %u",
+						 object->objectId);
+				appendStringInfo(&buffer, _("collation %s"),
+				 NameStr(((Form_pg_collation) GETSTRUCT(collTup))->collname));
+				ReleaseSysCache(collTup);
+				break;
+			}
+
 		case OCLASS_CONSTRAINT:
 			{
 				HeapTuple	conTup;

src/backend/catalog/heap.c

@@ -42,6 +42,7 @@
 #include "catalog/namespace.h"
 #include "catalog/objectaccess.h"
 #include "catalog/pg_attrdef.h"
+#include "catalog/pg_collation.h"
 #include "catalog/pg_constraint.h"
 #include "catalog/pg_foreign_table.h"
 #include "catalog/pg_inherits.h"
@@ -613,6 +614,14 @@ AddNewAttributeTuples(Oid new_rel_oid,
 		referenced.objectId = attr->atttypid;
 		referenced.objectSubId = 0;
 		recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
+
+		if (OidIsValid(attr->attcollation))
+		{
+			referenced.classId = CollationRelationId;
+			referenced.objectId = attr->attcollation;
+			referenced.objectSubId = 0;
+			recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
+		}
 	}
 
 	/*

src/backend/catalog/index.c

@@ -36,6 +36,7 @@
 #include "catalog/index.h"
 #include "catalog/indexing.h"
 #include "catalog/namespace.h"
+#include "catalog/pg_collation.h"
 #include "catalog/pg_constraint.h"
 #include "catalog/pg_operator.h"
 #include "catalog/pg_opclass.h"
@@ -960,6 +961,19 @@ index_create(Relation heapRelation,
 			Assert(!initdeferred);
 		}
 
+		/* Store dependency on collations */
+		for (i = 0; i < indexInfo->ii_NumIndexAttrs; i++)
+		{
+			if (OidIsValid(collationObjectId[i]))
+			{
+				referenced.classId = CollationRelationId;
+				referenced.objectId = collationObjectId[i];
+				referenced.objectSubId = 0;
+
+				recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
+			}
+		}
+
 		/* Store dependency on operator classes */
 		for (i = 0; i < indexInfo->ii_NumIndexAttrs; i++)
 		{

src/backend/catalog/objectaddress.c

@@ -25,6 +25,7 @@
 #include "catalog/pg_authid.h"
 #include "catalog/pg_cast.h"
 #include "catalog/pg_class.h"
+#include "catalog/pg_collation.h"
 #include "catalog/pg_constraint.h"
 #include "catalog/pg_conversion.h"
 #include "catalog/pg_database.h"
@@ -165,6 +166,11 @@ get_object_address(ObjectType objtype, List *objname, List *objargs,
 										false, -1);
 			address.objectSubId = 0;
 			break;
+		case OBJECT_COLLATION:
+			address.classId = CollationRelationId;
+			address.objectId = get_collation_oid(objname, false);
+			address.objectSubId = 0;
+			break;
 		case OBJECT_CONVERSION:
 			address.classId = ConversionRelationId;
 			address.objectId = get_conversion_oid(objname, false);
@@ -621,6 +627,9 @@ object_exists(ObjectAddress address)
 		case OperatorRelationId:
 			cache = OPEROID;
 			break;
+		case CollationRelationId:
+			cache = COLLOID;
+			break;
 		case ConversionRelationId:
 			cache = CONVOID;
 			break;

src/backend/catalog/pg_collation.c

@@ -0,0 +1,163 @@
+/*-------------------------------------------------------------------------
+ *
+ * pg_collation.c
+ *	  routines to support manipulation of the pg_collation relation
+ *
+ * Portions Copyright (c) 1996-2011, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *
+ * IDENTIFICATION
+ *	  src/backend/catalog/pg_collation.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/heapam.h"
+#include "access/sysattr.h"
+#include "catalog/dependency.h"
+#include "catalog/indexing.h"
+#include "catalog/objectaccess.h"
+#include "catalog/pg_collation.h"
+#include "catalog/pg_collation_fn.h"
+#include "catalog/pg_namespace.h"
+#include "catalog/pg_proc.h"
+#include "mb/pg_wchar.h"
+#include "miscadmin.h"
+#include "utils/acl.h"
+#include "utils/builtins.h"
+#include "utils/fmgroids.h"
+#include "utils/rel.h"
+#include "utils/syscache.h"
+#include "utils/tqual.h"
+
+/*
+ * CollationCreate
+ *
+ * Add a new tuple to pg_collation.
+ */
+Oid
+CollationCreate(const char *collname, Oid collnamespace,
+				Oid collowner,
+				int32 collencoding,
+				const char *collcollate, const char *collctype)
+{
+	int			i;
+	Relation	rel;
+	TupleDesc	tupDesc;
+	HeapTuple	tup;
+	bool		nulls[Natts_pg_collation];
+	Datum		values[Natts_pg_collation];
+	NameData	name_name, name_collate, name_ctype;
+	Oid			oid;
+	ObjectAddress myself,
+				referenced;
+
+	AssertArg(collname);
+	AssertArg(collnamespace);
+	AssertArg(collowner);
+	AssertArg(collcollate);
+	AssertArg(collctype);
+
+	/* make sure there is no existing collation of same name */
+	if (SearchSysCacheExists3(COLLNAMEENCNSP,
+							  PointerGetDatum(collname),
+							  Int32GetDatum(collencoding),
+							  ObjectIdGetDatum(collnamespace)))
+		ereport(ERROR,
+				(errcode(ERRCODE_DUPLICATE_OBJECT),
+				 errmsg("collation \"%s\" for encoding \"%s\" already exists",
+						collname, pg_encoding_to_char(collencoding))));
+
+	/* open pg_collation */
+	rel = heap_open(CollationRelationId, RowExclusiveLock);
+	tupDesc = rel->rd_att;
+
+	/* initialize nulls and values */
+	for (i = 0; i < Natts_pg_collation; i++)
+	{
+		nulls[i] = false;
+		values[i] = (Datum) NULL;
+	}
+
+	/* form a tuple */
+	namestrcpy(&name_name, collname);
+	values[Anum_pg_collation_collname - 1] = NameGetDatum(&name_name);
+	values[Anum_pg_collation_collnamespace - 1] = ObjectIdGetDatum(collnamespace);
+	values[Anum_pg_collation_collowner - 1] = ObjectIdGetDatum(collowner);
+	values[Anum_pg_collation_collencoding - 1] = Int32GetDatum(collencoding);
+	namestrcpy(&name_collate, collcollate);
+	values[Anum_pg_collation_collcollate - 1] = NameGetDatum(&name_collate);
+	namestrcpy(&name_ctype, collctype);
+	values[Anum_pg_collation_collctype - 1] = NameGetDatum(&name_ctype);
+
+	tup = heap_form_tuple(tupDesc, values, nulls);
+
+	/* insert a new tuple */
+	oid = simple_heap_insert(rel, tup);
+	Assert(OidIsValid(oid));
+
+	/* update the index if any */
+	CatalogUpdateIndexes(rel, tup);
+
+	myself.classId = CollationRelationId;
+	myself.objectId = HeapTupleGetOid(tup);
+	myself.objectSubId = 0;
+
+	/* create dependency on namespace */
+	referenced.classId = NamespaceRelationId;
+	referenced.objectId = collnamespace;
+	referenced.objectSubId = 0;
+	recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
+
+	/* create dependency on owner */
+	recordDependencyOnOwner(CollationRelationId, HeapTupleGetOid(tup),
+							collowner);
+
+	/* dependency on extension */
+	recordDependencyOnCurrentExtension(&myself);
+
+	/* Post creation hook for new collation */
+	InvokeObjectAccessHook(OAT_POST_CREATE,
+						   CollationRelationId, HeapTupleGetOid(tup), 0);
+
+	heap_freetuple(tup);
+	heap_close(rel, RowExclusiveLock);
+
+	return oid;
+}
+
+/*
+ * RemoveCollationById
+ *
+ * Remove a tuple from pg_collation by Oid. This function is solely
+ * called inside catalog/dependency.c
+ */
+void
+RemoveCollationById(Oid collationOid)
+{
+	Relation	rel;
+	HeapTuple	tuple;
+	HeapScanDesc scan;
+	ScanKeyData scanKeyData;
+
+	ScanKeyInit(&scanKeyData,
+				ObjectIdAttributeNumber,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(collationOid));
+
+	/* open pg_collation */
+	rel = heap_open(CollationRelationId, RowExclusiveLock);
+
+	scan = heap_beginscan(rel, SnapshotNow,
+						  1, &scanKeyData);
+
+	/* search for the target tuple */
+	if (HeapTupleIsValid(tuple = heap_getnext(scan, ForwardScanDirection)))
+		simple_heap_delete(rel, &tuple->t_self);
+	else
+		elog(ERROR, "could not find tuple for collation %u", collationOid);
+	heap_endscan(scan);
+	heap_close(rel, RowExclusiveLock);
+}

src/backend/catalog/pg_shdepend.c

@@ -21,6 +21,7 @@
 #include "catalog/dependency.h"
 #include "catalog/indexing.h"
 #include "catalog/pg_authid.h"
+#include "catalog/pg_collation.h"
 #include "catalog/pg_conversion.h"
 #include "catalog/pg_database.h"
 #include "catalog/pg_default_acl.h"
@@ -35,6 +36,7 @@
 #include "catalog/pg_tablespace.h"
 #include "catalog/pg_type.h"
 #include "commands/dbcommands.h"
+#include "commands/collationcmds.h"
 #include "commands/conversioncmds.h"
 #include "commands/defrem.h"
 #include "commands/proclang.h"
@@ -1323,6 +1325,10 @@ shdepReassignOwned(List *roleids, Oid newrole)
 			/* Issue the appropriate ALTER OWNER call */
 			switch (sdepForm->classid)
 			{
+				case CollationRelationId:
+					AlterCollationOwner_oid(sdepForm->objid, newrole);
+					break;
+
 				case ConversionRelationId:
 					AlterConversionOwner_oid(sdepForm->objid, newrole);
 					break;

src/backend/catalog/pg_type.c

@@ -19,6 +19,7 @@
 #include "catalog/dependency.h"
 #include "catalog/indexing.h"
 #include "catalog/objectaccess.h"
+#include "catalog/pg_collation.h"
 #include "catalog/pg_namespace.h"
 #include "catalog/pg_proc.h"
 #include "catalog/pg_type.h"
@@ -156,6 +157,7 @@ TypeShellMake(const char *typeName, Oid typeNamespace, Oid ownerId)
 								 InvalidOid,
 								 false,
 								 InvalidOid,
+								 InvalidOid,
 								 NULL,
 								 false);
 
@@ -460,6 +462,7 @@ TypeCreate(Oid newTypeOid,
 								 elementType,
 								 isImplicitArray,
 								 baseType,
+								 typeCollation,
 								 (defaultTypeBin ?
 								  stringToNode(defaultTypeBin) :
 								  NULL),
@@ -499,6 +502,7 @@ GenerateTypeDependencies(Oid typeNamespace,
 						 Oid elementType,
 						 bool isImplicitArray,
 						 Oid baseType,
+						 Oid typeCollation,
 						 Node *defaultExpr,
 						 bool rebuild)
 {
@@ -639,6 +643,15 @@ GenerateTypeDependencies(Oid typeNamespace,
 		recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
 	}
 
+	/* Normal dependency from a domain to its base type's collation. */
+	if (OidIsValid(typeCollation))
+	{
+		referenced.classId = CollationRelationId;
+		referenced.objectId = typeCollation;
+		referenced.objectSubId = 0;
+		recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
+	}
+
 	/* Normal dependency on the default expression. */
 	if (defaultExpr)
 		recordDependencyOnExpr(&myself, defaultExpr, NIL, DEPENDENCY_NORMAL);

src/backend/commands/Makefile

@@ -13,7 +13,7 @@ top_builddir = ../../..
 include $(top_builddir)/src/Makefile.global
 
 OBJS = aggregatecmds.o alter.o analyze.o async.o cluster.o comment.o  \
-	constraint.o conversioncmds.o copy.o \
+	collationcmds.o constraint.o conversioncmds.o copy.o \
 	dbcommands.o define.o discard.o explain.o extension.o \
 	foreigncmds.o functioncmds.o \
 	indexcmds.o lockcmds.o operatorcmds.o opclasscmds.o \

src/backend/commands/alter.c

@@ -20,6 +20,7 @@
 #include "catalog/pg_largeobject.h"
 #include "catalog/pg_namespace.h"
 #include "commands/alter.h"
+#include "commands/collationcmds.h"
 #include "commands/conversioncmds.h"
 #include "commands/dbcommands.h"
 #include "commands/defrem.h"
@@ -53,6 +54,10 @@ ExecRenameStmt(RenameStmt *stmt)
 			RenameAggregate(stmt->object, stmt->objarg, stmt->newname);
 			break;
 
+		case OBJECT_COLLATION:
+			RenameCollation(stmt->object, stmt->newname);
+			break;
+
 		case OBJECT_CONVERSION:
 			RenameConversion(stmt->object, stmt->newname);
 			break;
@@ -185,6 +190,10 @@ ExecAlterObjectSchemaStmt(AlterObjectSchemaStmt *stmt)
 								   stmt->newschema);
 			break;
 
+		case OBJECT_COLLATION:
+			AlterCollationNamespace(stmt->object, stmt->newschema);
+			break;
+
 		case OBJECT_CONVERSION:
 			AlterConversionNamespace(stmt->object, stmt->newschema);
 			break;
@@ -302,6 +311,10 @@ AlterObjectNamespace_oid(Oid classId, Oid objid, Oid nspOid)
 			oldNspOid = AlterTypeNamespace_oid(objid, nspOid);
 			break;
 
+		case OCLASS_COLLATION:
+			oldNspOid = AlterCollationNamespace_oid(objid, nspOid);
+			break;
+
 		case OCLASS_CONVERSION:
 			oldNspOid = AlterConversionNamespace_oid(objid, nspOid);
 			break;
@@ -478,6 +491,10 @@ ExecAlterOwnerStmt(AlterOwnerStmt *stmt)
 			AlterAggregateOwner(stmt->object, stmt->objarg, newowner);
 			break;
 
+		case OBJECT_COLLATION:
+			AlterCollationOwner(stmt->object, newowner);
+			break;
+
 		case OBJECT_CONVERSION:
 			AlterConversionOwner(stmt->object, newowner);
 			break;

src/backend/commands/collationcmds.c

@@ -0,0 +1,401 @@
+/*-------------------------------------------------------------------------
+ *
+ * collationcmds.c
+ *	  collation creation command support code
+ *
+ * Portions Copyright (c) 1996-2011, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *
+ * IDENTIFICATION
+ *	  src/backend/commands/collationcmds.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/heapam.h"
+#include "catalog/dependency.h"
+#include "catalog/indexing.h"
+#include "catalog/namespace.h"
+#include "catalog/pg_collation.h"
+#include "catalog/pg_collation_fn.h"
+#include "commands/alter.h"
+#include "commands/collationcmds.h"
+#include "commands/dbcommands.h"
+#include "commands/defrem.h"
+#include "mb/pg_wchar.h"
+#include "miscadmin.h"
+#include "parser/parse_type.h"
+#include "utils/acl.h"
+#include "utils/builtins.h"
+#include "utils/lsyscache.h"
+#include "utils/syscache.h"
+
+static void AlterCollationOwner_internal(Relation rel, Oid collationOid,
+							  Oid newOwnerId);
+
+/*
+ * CREATE COLLATION
+ */
+void
+DefineCollation(List *names, List *parameters)
+{
+	char	   *collName;
+	Oid			collNamespace;
+	AclResult	aclresult;
+	ListCell   *pl;
+	DefElem	   *fromEl = NULL;
+	DefElem	   *localeEl = NULL;
+	DefElem	   *lccollateEl = NULL;
+	DefElem	   *lcctypeEl = NULL;
+	char	   *collcollate = NULL;
+	char	   *collctype = NULL;
+
+	collNamespace = QualifiedNameGetCreationNamespace(names, &collName);
+
+	aclresult = pg_namespace_aclcheck(collNamespace, GetUserId(), ACL_CREATE);
+	if (aclresult != ACLCHECK_OK)
+		aclcheck_error(aclresult, ACL_KIND_NAMESPACE,
+					   get_namespace_name(collNamespace));
+
+	foreach(pl, parameters)
+	{
+		DefElem	   *defel = (DefElem *) lfirst(pl);
+		DefElem   **defelp;
+
+		if (pg_strcasecmp(defel->defname, "from") == 0)
+			defelp = &fromEl;
+		else if (pg_strcasecmp(defel->defname, "locale") == 0)
+			defelp = &localeEl;
+		else if (pg_strcasecmp(defel->defname, "lc_collate") == 0)
+			defelp = &lccollateEl;
+		else if (pg_strcasecmp(defel->defname, "lc_ctype") == 0)
+			defelp = &lcctypeEl;
+		else
+		{
+			ereport(ERROR,
+					(errcode(ERRCODE_SYNTAX_ERROR),
+					 errmsg("collation attribute \"%s\" not recognized",
+							defel->defname)));
+			break;
+		}
+
+		*defelp = defel;
+	}
+
+	if ((localeEl && (lccollateEl || lcctypeEl))
+		|| (fromEl && list_length(parameters) != 1))
+		ereport(ERROR,
+				(errcode(ERRCODE_SYNTAX_ERROR),
+				 errmsg("conflicting or redundant options")));
+
+	if (fromEl)
+	{
+		Oid			collid;
+		HeapTuple	tp;
+
+		collid = LookupCollation(NULL, defGetQualifiedName(fromEl), -1);
+		tp = SearchSysCache1(COLLOID, ObjectIdGetDatum(collid));
+		if (!HeapTupleIsValid(tp))
+			elog(ERROR, "cache lookup failed for collation %u", collid);
+
+		collcollate = pstrdup(NameStr(((Form_pg_collation) GETSTRUCT(tp))->collcollate));
+		collctype = pstrdup(NameStr(((Form_pg_collation) GETSTRUCT(tp))->collctype));
+
+		ReleaseSysCache(tp);
+	}
+
+	if (localeEl)
+	{
+		collcollate = defGetString(localeEl);
+		collctype = defGetString(localeEl);
+	}
+
+	if (lccollateEl)
+		collcollate = defGetString(lccollateEl);
+
+	if (lcctypeEl)
+		collctype = defGetString(lcctypeEl);
+
+	if (!collcollate)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+				 errmsg("parameter \"lc_collate\" parameter must be specified")));
+
+	if (!collctype)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+				 errmsg("parameter \"lc_ctype\" must be specified")));
+
+	check_encoding_locale_matches(GetDatabaseEncoding(), collcollate, collctype);
+
+	CollationCreate(collName,
+					collNamespace,
+					GetUserId(),
+					GetDatabaseEncoding(),
+					collcollate,
+					collctype);
+}
+
+/*
+ * DROP COLLATION
+ */
+void
+DropCollationsCommand(DropStmt *drop)
+{
+	ObjectAddresses *objects;
+	ListCell   *cell;
+
+	/*
+	 * First we identify all the collations, then we delete them in a single
+	 * performMultipleDeletions() call.  This is to avoid unwanted DROP
+	 * RESTRICT errors if one of the collations depends on another. (Not that
+	 * that is very likely, but we may as well do this consistently.)
+	 */
+	objects = new_object_addresses();
+
+	foreach(cell, drop->objects)
+	{
+		List	   *name = (List *) lfirst(cell);
+		Oid			collationOid;
+		HeapTuple	tuple;
+		Form_pg_collation coll;
+		ObjectAddress object;
+
+		collationOid = get_collation_oid(name, drop->missing_ok);
+
+		if (!OidIsValid(collationOid))
+		{
+			ereport(NOTICE,
+					(errmsg("collation \"%s\" does not exist, skipping",
+							NameListToString(name))));
+			continue;
+		}
+
+		tuple = SearchSysCache1(COLLOID, ObjectIdGetDatum(collationOid));
+		if (!HeapTupleIsValid(tuple))
+			elog(ERROR, "cache lookup failed for collation %u",
+				 collationOid);
+		coll = (Form_pg_collation) GETSTRUCT(tuple);
+
+		/* Permission check: must own collation or its namespace */
+		if (!pg_collation_ownercheck(collationOid, GetUserId()) &&
+			!pg_namespace_ownercheck(coll->collnamespace, GetUserId()))
+			aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_COLLATION,
+						   NameStr(coll->collname));
+
+		object.classId = CollationRelationId;
+		object.objectId = collationOid;
+		object.objectSubId = 0;
+
+		add_exact_object_address(&object, objects);
+
+		ReleaseSysCache(tuple);
+	}
+
+	performMultipleDeletions(objects, drop->behavior);
+
+	free_object_addresses(objects);
+}
+
+/*
+ * Rename collation
+ */
+void
+RenameCollation(List *name, const char *newname)
+{
+	Oid			collationOid;
+	Oid			namespaceOid;
+	HeapTuple	tup;
+	Relation	rel;
+	AclResult	aclresult;
+
+	rel = heap_open(CollationRelationId, RowExclusiveLock);
+
+	collationOid = get_collation_oid(name, false);
+
+	tup = SearchSysCacheCopy1(COLLOID, ObjectIdGetDatum(collationOid));
+	if (!HeapTupleIsValid(tup)) /* should not happen */
+		elog(ERROR, "cache lookup failed for collation %u", collationOid);
+
+	namespaceOid = ((Form_pg_collation) GETSTRUCT(tup))->collnamespace;
+
+	/* make sure the new name doesn't exist */
+	if (SearchSysCacheExists3(COLLNAMEENCNSP,
+							  CStringGetDatum(newname),
+							  Int32GetDatum(GetDatabaseEncoding()),
+							  ObjectIdGetDatum(namespaceOid)))
+		ereport(ERROR,
+				(errcode(ERRCODE_DUPLICATE_OBJECT),
+				 errmsg("collation \"%s\" for current database encoding \"%s\" already exists in schema \"%s\"",
+						newname,
+						GetDatabaseEncodingName(),
+						get_namespace_name(namespaceOid))));
+
+	/* must be owner */
+	if (!pg_collation_ownercheck(collationOid, GetUserId()))
+		aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_COLLATION,
+					   NameListToString(name));
+
+	/* must have CREATE privilege on namespace */
+	aclresult = pg_namespace_aclcheck(namespaceOid, GetUserId(), ACL_CREATE);
+	if (aclresult != ACLCHECK_OK)
+		aclcheck_error(aclresult, ACL_KIND_NAMESPACE,
+					   get_namespace_name(namespaceOid));
+
+	/* rename */
+	namestrcpy(&(((Form_pg_collation) GETSTRUCT(tup))->collname), newname);
+	simple_heap_update(rel, &tup->t_self, tup);
+	CatalogUpdateIndexes(rel, tup);
+
+	heap_close(rel, NoLock);
+	heap_freetuple(tup);
+}
+
+/*
+ * Change collation owner, by name
+ */
+void
+AlterCollationOwner(List *name, Oid newOwnerId)
+{
+	Oid			collationOid;
+	Relation	rel;
+
+	rel = heap_open(CollationRelationId, RowExclusiveLock);
+
+	collationOid = get_collation_oid(name, false);
+
+	AlterCollationOwner_internal(rel, collationOid, newOwnerId);
+
+	heap_close(rel, NoLock);
+}
+
+/*
+ * Change collation owner, by oid
+ */
+void
+AlterCollationOwner_oid(Oid collationOid, Oid newOwnerId)
+{
+	Relation	rel;
+
+	rel = heap_open(CollationRelationId, RowExclusiveLock);
+
+	AlterCollationOwner_internal(rel, collationOid, newOwnerId);
+
+	heap_close(rel, NoLock);
+}
+
+/*
+ * AlterCollationOwner_internal
+ *
+ * Internal routine for changing the owner.  rel must be pg_collation, already
+ * open and suitably locked; it will not be closed.
+ */
+static void
+AlterCollationOwner_internal(Relation rel, Oid collationOid, Oid newOwnerId)
+{
+	Form_pg_collation collForm;
+	HeapTuple	tup;
+
+	Assert(RelationGetRelid(rel) == CollationRelationId);
+
+	tup = SearchSysCacheCopy1(COLLOID, ObjectIdGetDatum(collationOid));
+	if (!HeapTupleIsValid(tup)) /* should not happen */
+		elog(ERROR, "cache lookup failed for collation %u", collationOid);
+
+	collForm = (Form_pg_collation) GETSTRUCT(tup);
+
+	/*
+	 * If the new owner is the same as the existing owner, consider the
+	 * command to have succeeded.  This is for dump restoration purposes.
+	 */
+	if (collForm->collowner != newOwnerId)
+	{
+		AclResult	aclresult;
+
+		/* Superusers can always do it */
+		if (!superuser())
+		{
+			/* Otherwise, must be owner of the existing object */
+			if (!pg_collation_ownercheck(HeapTupleGetOid(tup), GetUserId()))
+				aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_COLLATION,
+							   NameStr(collForm->collname));
+
+			/* Must be able to become new owner */
+			check_is_member_of_role(GetUserId(), newOwnerId);
+
+			/* New owner must have CREATE privilege on namespace */
+			aclresult = pg_namespace_aclcheck(collForm->collnamespace,
+											  newOwnerId,
+											  ACL_CREATE);
+			if (aclresult != ACLCHECK_OK)
+				aclcheck_error(aclresult, ACL_KIND_NAMESPACE,
+							   get_namespace_name(collForm->collnamespace));
+		}
+
+		/*
+		 * Modify the owner --- okay to scribble on tup because it's a copy
+		 */
+		collForm->collowner = newOwnerId;
+
+		simple_heap_update(rel, &tup->t_self, tup);
+
+		CatalogUpdateIndexes(rel, tup);
+
+		/* Update owner dependency reference */
+		changeDependencyOnOwner(CollationRelationId, collationOid,
+								newOwnerId);
+	}
+
+	heap_freetuple(tup);
+}
+
+/*
+ * Execute ALTER COLLATION SET SCHEMA
+ */
+void
+AlterCollationNamespace(List *name, const char *newschema)
+{
+	Oid			collOid, nspOid;
+	Relation	rel;
+
+	rel = heap_open(CollationRelationId, RowExclusiveLock);
+
+	collOid = get_collation_oid(name, false);
+
+	/* get schema OID */
+	nspOid = LookupCreationNamespace(newschema);
+
+	AlterObjectNamespace(rel, COLLOID, -1,
+						 collOid, nspOid,
+						 Anum_pg_collation_collname,
+						 Anum_pg_collation_collnamespace,
+						 Anum_pg_collation_collowner,
+						 ACL_KIND_COLLATION);
+
+	heap_close(rel, NoLock);
+}
+
+/*
+ * Change collation schema, by oid
+ */
+Oid
+AlterCollationNamespace_oid(Oid collOid, Oid newNspOid)
+{
+	Oid         oldNspOid;
+	Relation	rel;
+
+	rel = heap_open(CollationRelationId, RowExclusiveLock);
+
+	oldNspOid = AlterObjectNamespace(rel, COLLOID, -1,
+									 collOid, newNspOid,
+									 Anum_pg_collation_collname,
+									 Anum_pg_collation_collnamespace,
+									 Anum_pg_collation_collowner,
+									 ACL_KIND_COLLATION);
+
+	heap_close(rel, RowExclusiveLock);
+
+	return oldNspOid;
+}

src/backend/commands/comment.c

@@ -133,6 +133,11 @@ CommentObject(CommentStmt *stmt)
 				aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_NAMESPACE,
 							   strVal(linitial(stmt->objname)));
 			break;
+		case OBJECT_COLLATION:
+			if (!pg_collation_ownercheck(address.objectId, GetUserId()))
+				aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_COLLATION,
+							   NameListToString(stmt->objname));
+			break;
 		case OBJECT_CONVERSION:
 			if (!pg_conversion_ownercheck(address.objectId, GetUserId()))
 				aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CONVERSION,

src/backend/commands/dbcommands.c

@@ -129,8 +129,6 @@ createdb(const CreatedbStmt *stmt)
 	char	   *dbctype = NULL;
 	int			encoding = -1;
 	int			dbconnlimit = -1;
-	int			ctype_encoding;
-	int			collate_encoding;
 	int			notherbackends;
 	int			npreparedxacts;
 	createdb_failure_params fparms;
@@ -334,60 +332,7 @@ createdb(const CreatedbStmt *stmt)
 				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
 				 errmsg("invalid locale name %s", dbctype)));
 
-	/*
+	check_encoding_locale_matches(encoding, dbcollate, dbctype);
-	 * Check whether chosen encoding matches chosen locale settings.  This
-	 * restriction is necessary because libc's locale-specific code usually
-	 * fails when presented with data in an encoding it's not expecting. We
-	 * allow mismatch in four cases:
-	 *
-	 * 1. locale encoding = SQL_ASCII, which means that the locale is C/POSIX
-	 * which works with any encoding.
-	 *
-	 * 2. locale encoding = -1, which means that we couldn't determine the
-	 * locale's encoding and have to trust the user to get it right.
-	 *
-	 * 3. selected encoding is UTF8 and platform is win32. This is because
-	 * UTF8 is a pseudo codepage that is supported in all locales since it's
-	 * converted to UTF16 before being used.
-	 *
-	 * 4. selected encoding is SQL_ASCII, but only if you're a superuser. This
-	 * is risky but we have historically allowed it --- notably, the
-	 * regression tests require it.
-	 *
-	 * Note: if you change this policy, fix initdb to match.
-	 */
-	ctype_encoding = pg_get_encoding_from_locale(dbctype, true);
-	collate_encoding = pg_get_encoding_from_locale(dbcollate, true);
-
-	if (!(ctype_encoding == encoding ||
-		  ctype_encoding == PG_SQL_ASCII ||
-		  ctype_encoding == -1 ||
-#ifdef WIN32
-		  encoding == PG_UTF8 ||
-#endif
-		  (encoding == PG_SQL_ASCII && superuser())))
-		ereport(ERROR,
-				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-				 errmsg("encoding %s does not match locale %s",
-						pg_encoding_to_char(encoding),
-						dbctype),
-			   errdetail("The chosen LC_CTYPE setting requires encoding %s.",
-						 pg_encoding_to_char(ctype_encoding))));
-
-	if (!(collate_encoding == encoding ||
-		  collate_encoding == PG_SQL_ASCII ||
-		  collate_encoding == -1 ||
-#ifdef WIN32
-		  encoding == PG_UTF8 ||
-#endif
-		  (encoding == PG_SQL_ASCII && superuser())))
-		ereport(ERROR,
-				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-				 errmsg("encoding %s does not match locale %s",
-						pg_encoding_to_char(encoding),
-						dbcollate),
-			 errdetail("The chosen LC_COLLATE setting requires encoding %s.",
-					   pg_encoding_to_char(collate_encoding))));
 
 	/*
 	 * Check that the new encoding and locale settings match the source
@@ -710,6 +655,65 @@ createdb(const CreatedbStmt *stmt)
 								PointerGetDatum(&fparms));
 }
 
+/*
+ * Check whether chosen encoding matches chosen locale settings.  This
+ * restriction is necessary because libc's locale-specific code usually
+ * fails when presented with data in an encoding it's not expecting. We
+ * allow mismatch in four cases:
+ *
+ * 1. locale encoding = SQL_ASCII, which means that the locale is C/POSIX
+ * which works with any encoding.
+ *
+ * 2. locale encoding = -1, which means that we couldn't determine the
+ * locale's encoding and have to trust the user to get it right.
+ *
+ * 3. selected encoding is UTF8 and platform is win32. This is because
+ * UTF8 is a pseudo codepage that is supported in all locales since it's
+ * converted to UTF16 before being used.
+ *
+ * 4. selected encoding is SQL_ASCII, but only if you're a superuser. This
+ * is risky but we have historically allowed it --- notably, the
+ * regression tests require it.
+ *
+ * Note: if you change this policy, fix initdb to match.
+ */
+void
+check_encoding_locale_matches(int encoding, const char *collate, const char *ctype)
+{
+	int ctype_encoding = pg_get_encoding_from_locale(ctype, true);
+	int collate_encoding = pg_get_encoding_from_locale(collate, true);
+
+	if (!(ctype_encoding == encoding ||
+		  ctype_encoding == PG_SQL_ASCII ||
+		  ctype_encoding == -1 ||
+#ifdef WIN32
+		  encoding == PG_UTF8 ||
+#endif
+		  (encoding == PG_SQL_ASCII && superuser())))
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("encoding %s does not match locale %s",
+						pg_encoding_to_char(encoding),
+						ctype),
+			   errdetail("The chosen LC_CTYPE setting requires encoding %s.",
+						 pg_encoding_to_char(ctype_encoding))));
+
+	if (!(collate_encoding == encoding ||
+		  collate_encoding == PG_SQL_ASCII ||
+		  collate_encoding == -1 ||
+#ifdef WIN32
+		  encoding == PG_UTF8 ||
+#endif
+		  (encoding == PG_SQL_ASCII && superuser())))
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("encoding %s does not match locale %s",
+						pg_encoding_to_char(encoding),
+						collate),
+			 errdetail("The chosen LC_COLLATE setting requires encoding %s.",
+					   pg_encoding_to_char(collate_encoding))));
+}
+
 /* Error cleanup callback for createdb */
 static void
 createdb_failure_callback(int code, Datum arg)

src/backend/commands/tablecmds.c

@@ -27,6 +27,7 @@
 #include "catalog/indexing.h"
 #include "catalog/namespace.h"
 #include "catalog/objectaccess.h"
+#include "catalog/pg_collation.h"
 #include "catalog/pg_constraint.h"
 #include "catalog/pg_depend.h"
 #include "catalog/pg_foreign_table.h"
@@ -293,7 +294,7 @@ static void ATPrepAddColumn(List **wqueue, Relation rel, bool recurse, bool recu
 				AlterTableCmd *cmd, LOCKMODE lockmode);
 static void ATExecAddColumn(AlteredTableInfo *tab, Relation rel,
 				ColumnDef *colDef, bool isOid, LOCKMODE lockmode);
-static void add_column_datatype_dependency(Oid relid, int32 attnum, Oid typid);
+static void add_column_datatype_dependency(Oid relid, int32 attnum, Oid typid, Oid collid);
 static void ATPrepAddOids(List **wqueue, Relation rel, bool recurse,
 			  AlterTableCmd *cmd, LOCKMODE lockmode);
 static void ATExecDropNotNull(Relation rel, const char *colName, LOCKMODE lockmode);
@@ -4369,14 +4370,14 @@ ATExecAddColumn(AlteredTableInfo *tab, Relation rel,
 	/*
 	 * Add needed dependency entries for the new column.
 	 */
-	add_column_datatype_dependency(myrelid, newattnum, attribute.atttypid);
+	add_column_datatype_dependency(myrelid, newattnum, attribute.atttypid, attribute.attcollation);
 }
 
 /*
  * Install a column's dependency on its datatype.
  */
 static void
-add_column_datatype_dependency(Oid relid, int32 attnum, Oid typid)
+add_column_datatype_dependency(Oid relid, int32 attnum, Oid typid, Oid collid)
 {
 	ObjectAddress myself,
 				referenced;
@@ -4388,6 +4389,14 @@ add_column_datatype_dependency(Oid relid, int32 attnum, Oid typid)
 	referenced.objectId = typid;
 	referenced.objectSubId = 0;
 	recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
+
+	if (collid)
+	{
+		referenced.classId = CollationRelationId;
+		referenced.objectId = collid;
+		referenced.objectSubId = 0;
+		recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
+	}
 }
 
 /*
@@ -6877,6 +6886,7 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
 			case OCLASS_PROC:
 			case OCLASS_TYPE:
 			case OCLASS_CAST:
+			case OCLASS_COLLATION:
 			case OCLASS_CONVERSION:
 			case OCLASS_LANGUAGE:
 			case OCLASS_LARGEOBJECT:
@@ -6918,7 +6928,7 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
 	/*
 	 * Now scan for dependencies of this column on other things.  The only
 	 * thing we should find is the dependency on the column datatype, which we
-	 * want to remove.
+	 * want to remove, and possibly an associated collation.
 	 */
 	ScanKeyInit(&key[0],
 				Anum_pg_depend_classid,
@@ -6943,8 +6953,10 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
 		if (foundDep->deptype != DEPENDENCY_NORMAL)
 			elog(ERROR, "found unexpected dependency type '%c'",
 				 foundDep->deptype);
-		if (foundDep->refclassid != TypeRelationId ||
+		if (!(foundDep->refclassid == TypeRelationId &&
-			foundDep->refobjid != attTup->atttypid)
+			  foundDep->refobjid == attTup->atttypid) &&
+			!(foundDep->refclassid == CollationRelationId &&
+			  foundDep->refobjid == attTup->attcollation))
 			elog(ERROR, "found unexpected dependency for column");
 
 		simple_heap_delete(depRel, &depTup->t_self);
@@ -6977,7 +6989,7 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
 	heap_close(attrelation, RowExclusiveLock);
 
 	/* Install dependency on new datatype */
-	add_column_datatype_dependency(RelationGetRelid(rel), attnum, targettype);
+	add_column_datatype_dependency(RelationGetRelid(rel), attnum, targettype, targetcollid);
 
 	/*
 	 * Drop any pg_statistic entry for the column, since it's now wrong type

src/backend/commands/typecmds.c

@@ -1736,6 +1736,7 @@ AlterDomainDefault(List *names, Node *defaultRaw)
 							 InvalidOid,
 							 false,		/* a domain isn't an implicit array */
 							 typTup->typbasetype,
+							 typTup->typcollation,
 							 defaultExpr,
 							 true);		/* Rebuild is true */
 

src/backend/parser/gram.y

@@ -482,7 +482,7 @@ static RangeVar *makeRangeVarFromAnyName(List *names, int position, core_yyscan_
 
 	CACHE CALLED CASCADE CASCADED CASE CAST CATALOG_P CHAIN CHAR_P
 	CHARACTER CHARACTERISTICS CHECK CHECKPOINT CLASS CLOSE
-	CLUSTER COALESCE COLLATE COLUMN COMMENT COMMENTS COMMIT
+	CLUSTER COALESCE COLLATE COLLATION COLUMN COMMENT COMMENTS COMMIT
 	COMMITTED CONCURRENTLY CONFIGURATION CONNECTION CONSTRAINT CONSTRAINTS
 	CONTENT_P CONTINUE_P CONVERSION_P COPY COST CREATE CREATEDB
 	CREATEROLE CREATEUSER CROSS CSV CURRENT_P
@@ -3316,6 +3316,15 @@ AlterExtensionContentsStmt:
 					n->objargs = list_make1($9);
 					$$ = (Node *) n;
 				}
+			| ALTER EXTENSION name add_drop COLLATION any_name
+				{
+					AlterExtensionContentsStmt *n = makeNode(AlterExtensionContentsStmt);
+					n->extname = $3;
+					n->action = $4;
+					n->objtype = OBJECT_COLLATION;
+					n->objname = $6;
+					$$ = (Node *)n;
+				}
 			| ALTER EXTENSION name add_drop CONVERSION_P any_name
 				{
 					AlterExtensionContentsStmt *n = makeNode(AlterExtensionContentsStmt);
@@ -4248,6 +4257,24 @@ DefineStmt:
 					n->definition = $6;
 					$$ = (Node *)n;
 				}
+			| CREATE COLLATION any_name definition
+				{
+					DefineStmt *n = makeNode(DefineStmt);
+					n->kind = OBJECT_COLLATION;
+					n->args = NIL;
+					n->defnames = $3;
+					n->definition = $4;
+					$$ = (Node *)n;
+				}
+			| CREATE COLLATION any_name FROM any_name
+				{
+					DefineStmt *n = makeNode(DefineStmt);
+					n->kind = OBJECT_COLLATION;
+					n->args = NIL;
+					n->defnames = $3;
+					n->definition = list_make1(makeDefElem("from", (Node *) $5));
+					$$ = (Node *)n;
+				}
 		;
 
 definition: '(' def_list ')'						{ $$ = $2; }
@@ -4621,6 +4648,7 @@ drop_type:	TABLE									{ $$ = OBJECT_TABLE; }
 			| FOREIGN TABLE							{ $$ = OBJECT_FOREIGN_TABLE; }
 			| TYPE_P								{ $$ = OBJECT_TYPE; }
 			| DOMAIN_P								{ $$ = OBJECT_DOMAIN; }
+			| COLLATION								{ $$ = OBJECT_COLLATION; }
 			| CONVERSION_P							{ $$ = OBJECT_CONVERSION; }
 			| SCHEMA								{ $$ = OBJECT_SCHEMA; }
 			| EXTENSION								{ $$ = OBJECT_EXTENSION; }
@@ -4676,7 +4704,7 @@ opt_restart_seqs:
  *	the object associated with the comment. The form of the statement is:
  *
  *	COMMENT ON [ [ DATABASE | DOMAIN | INDEX | SEQUENCE | TABLE | TYPE | VIEW |
- *				   CONVERSION | LANGUAGE | OPERATOR CLASS | LARGE OBJECT |
+ *				   COLLATION | CONVERSION | LANGUAGE | OPERATOR CLASS | LARGE OBJECT |
  *				   CAST | COLUMN | SCHEMA | TABLESPACE | EXTENSION | ROLE |
  *				   TEXT SEARCH PARSER | TEXT SEARCH DICTIONARY |
  *				   TEXT SEARCH TEMPLATE | TEXT SEARCH CONFIGURATION |
@@ -4854,6 +4882,7 @@ comment_type:
 			| DOMAIN_P							{ $$ = OBJECT_DOMAIN; }
 			| TYPE_P							{ $$ = OBJECT_TYPE; }
 			| VIEW								{ $$ = OBJECT_VIEW; }
+			| COLLATION							{ $$ = OBJECT_COLLATION; }
 			| CONVERSION_P						{ $$ = OBJECT_CONVERSION; }
 			| TABLESPACE						{ $$ = OBJECT_TABLESPACE; }
 			| EXTENSION 						{ $$ = OBJECT_EXTENSION; }
@@ -6275,6 +6304,14 @@ RenameStmt: ALTER AGGREGATE func_name aggr_args RENAME TO name
 					n->newname = $7;
 					$$ = (Node *)n;
 				}
+			| ALTER COLLATION any_name RENAME TO name
+				{
+					RenameStmt *n = makeNode(RenameStmt);
+					n->renameType = OBJECT_COLLATION;
+					n->object = $3;
+					n->newname = $6;
+					$$ = (Node *)n;
+				}
 			| ALTER CONVERSION_P any_name RENAME TO name
 				{
 					RenameStmt *n = makeNode(RenameStmt);
@@ -6535,6 +6572,14 @@ AlterObjectSchemaStmt:
 					n->newschema = $7;
 					$$ = (Node *)n;
 				}
+			| ALTER COLLATION any_name SET SCHEMA name
+				{
+					AlterObjectSchemaStmt *n = makeNode(AlterObjectSchemaStmt);
+					n->objectType = OBJECT_COLLATION;
+					n->object = $3;
+					n->newschema = $6;
+					$$ = (Node *)n;
+				}
 			| ALTER CONVERSION_P any_name SET SCHEMA name
 				{
 					AlterObjectSchemaStmt *n = makeNode(AlterObjectSchemaStmt);
@@ -6684,6 +6729,14 @@ AlterOwnerStmt: ALTER AGGREGATE func_name aggr_args OWNER TO RoleId
 					n->newowner = $7;
 					$$ = (Node *)n;
 				}
+			| ALTER COLLATION any_name OWNER TO RoleId
+				{
+					AlterOwnerStmt *n = makeNode(AlterOwnerStmt);
+					n->objectType = OBJECT_COLLATION;
+					n->object = $3;
+					n->newowner = $6;
+					$$ = (Node *)n;
+				}
 			| ALTER CONVERSION_P any_name OWNER TO RoleId
 				{
 					AlterOwnerStmt *n = makeNode(AlterOwnerStmt);

src/backend/tcop/utility.c

@@ -26,6 +26,7 @@
 #include "commands/async.h"
 #include "commands/cluster.h"
 #include "commands/comment.h"
+#include "commands/collationcmds.h"
 #include "commands/conversioncmds.h"
 #include "commands/copy.h"
 #include "commands/dbcommands.h"
@@ -665,6 +666,10 @@ standard_ProcessUtility(Node *parsetree,
 						RemoveTypes(stmt);
 						break;
 
+					case OBJECT_COLLATION:
+						DropCollationsCommand(stmt);
+						break;
+
 					case OBJECT_CONVERSION:
 						DropConversionsCommand(stmt);
 						break;
@@ -884,6 +889,10 @@ standard_ProcessUtility(Node *parsetree,
 						Assert(stmt->args == NIL);
 						DefineTSConfiguration(stmt->defnames, stmt->definition);
 						break;
+					case OBJECT_COLLATION:
+						Assert(stmt->args == NIL);
+						DefineCollation(stmt->defnames, stmt->definition);
+						break;
 					default:
 						elog(ERROR, "unrecognized define stmt type: %d",
 							 (int) stmt->kind);
@@ -1453,6 +1462,9 @@ AlterObjectTypeCommandTag(ObjectType objtype)
 		case OBJECT_CAST:
 			tag = "ALTER CAST";
 			break;
+		case OBJECT_COLLATION:
+			tag = "ALTER COLLATION";
+			break;
 		case OBJECT_COLUMN:
 			tag = "ALTER TABLE";
 			break;
@@ -1754,6 +1766,9 @@ CreateCommandTag(Node *parsetree)
 				case OBJECT_DOMAIN:
 					tag = "DROP DOMAIN";
 					break;
+				case OBJECT_COLLATION:
+					tag = "DROP COLLATION";
+					break;
 				case OBJECT_CONVERSION:
 					tag = "DROP CONVERSION";
 					break;
@@ -1867,6 +1882,9 @@ CreateCommandTag(Node *parsetree)
 				case OBJECT_TSCONFIGURATION:
 					tag = "CREATE TEXT SEARCH CONFIGURATION";
 					break;
+				case OBJECT_COLLATION:
+					tag = "CREATE COLLATION";
+					break;
 				default:
 					tag = "???";
 			}

src/bin/initdb/initdb.c

@@ -1648,10 +1648,11 @@ setup_collation(void)
 	 * matches the OS locale name, else the first name by sort order
 	 * (arbitrary choice to be deterministic).
 	 */
-	PG_CMD_PUTS("INSERT INTO pg_collation (collname, collnamespace, collencoding, collcollate, collctype) "
+	PG_CMD_PUTS("INSERT INTO pg_collation (collname, collnamespace, collowner, collencoding, collcollate, collctype) "
 				" SELECT DISTINCT ON (final_collname, collnamespace, encoding)"
 				"   COALESCE(collname, locale) AS final_collname, "
 				"   (SELECT oid FROM pg_namespace WHERE nspname = 'pg_catalog') AS collnamespace, "
+				"   (SELECT relowner FROM pg_class WHERE relname = 'pg_collation') AS collowner, "
 				"   encoding, "
 				"   locale, locale "
 				"  FROM tmp_pg_collation"

src/bin/pg_dump/common.c

@@ -87,6 +87,7 @@ getSchemaData(int *numTablesPtr)
 	CastInfo   *castinfo;
 	OpclassInfo *opcinfo;
 	OpfamilyInfo *opfinfo;
+	CollInfo   *collinfo;
 	ConvInfo   *convinfo;
 	TSParserInfo *prsinfo;
 	TSTemplateInfo *tmplinfo;
@@ -104,6 +105,7 @@ getSchemaData(int *numTablesPtr)
 	int			numCasts;
 	int			numOpclasses;
 	int			numOpfamilies;
+	int			numCollations;
 	int			numConversions;
 	int			numTSParsers;
 	int			numTSTemplates;
@@ -182,6 +184,10 @@ getSchemaData(int *numTablesPtr)
 		write_msg(NULL, "reading default privileges\n");
 	daclinfo = getDefaultACLs(&numDefaultACLs);
 
+	if (g_verbose)
+		write_msg(NULL, "reading user-defined collations\n");
+	collinfo = getCollations(&numCollations);
+
 	if (g_verbose)
 		write_msg(NULL, "reading user-defined conversions\n");
 	convinfo = getConversions(&numConversions);

src/bin/pg_dump/pg_backup_archiver.c

@@ -2777,7 +2777,8 @@ _getObjectDescription(PQExpBuffer buf, TocEntry *te, ArchiveHandle *AH)
 		type = "TABLE";
 
 	/* objects named by a schema and name */
-	if (strcmp(type, "CONVERSION") == 0 ||
+	if (strcmp(type, "COLLATION") == 0 ||
+		strcmp(type, "CONVERSION") == 0 ||
 		strcmp(type, "DOMAIN") == 0 ||
 		strcmp(type, "TABLE") == 0 ||
 		strcmp(type, "TYPE") == 0 ||
@@ -2961,6 +2962,7 @@ _printTocEntry(ArchiveHandle *AH, TocEntry *te, RestoreOptions *ropt, bool isDat
 	{
 		if (strcmp(te->desc, "AGGREGATE") == 0 ||
 			strcmp(te->desc, "BLOB") == 0 ||
+			strcmp(te->desc, "COLLATION") == 0 ||
 			strcmp(te->desc, "CONVERSION") == 0 ||
 			strcmp(te->desc, "DATABASE") == 0 ||
 			strcmp(te->desc, "DOMAIN") == 0 ||

src/bin/pg_dump/pg_dump.c

@@ -175,6 +175,7 @@ static void dumpCast(Archive *fout, CastInfo *cast);
 static void dumpOpr(Archive *fout, OprInfo *oprinfo);
 static void dumpOpclass(Archive *fout, OpclassInfo *opcinfo);
 static void dumpOpfamily(Archive *fout, OpfamilyInfo *opfinfo);
+static void dumpCollation(Archive *fout, CollInfo *convinfo);
 static void dumpConversion(Archive *fout, ConvInfo *convinfo);
 static void dumpRule(Archive *fout, RuleInfo *rinfo);
 static void dumpAgg(Archive *fout, AggInfo *agginfo);
@@ -3094,6 +3095,84 @@ getOperators(int *numOprs)
 	return oprinfo;
 }
 
+/*
+ * getCollations:
+ *	  read all collations in the system catalogs and return them in the
+ * CollInfo* structure
+ *
+ *	numCollations is set to the number of collations read in
+ */
+CollInfo *
+getCollations(int *numCollations)
+{
+	PGresult   *res;
+	int			ntups;
+	int			i;
+	PQExpBuffer query = createPQExpBuffer();
+	CollInfo   *collinfo;
+	int			i_tableoid;
+	int			i_oid;
+	int			i_collname;
+	int			i_collnamespace;
+	int			i_rolname;
+
+	/* Collations didn't exist pre-9.1 */
+	if (g_fout->remoteVersion < 90100)
+	{
+		*numCollations = 0;
+		return NULL;
+	}
+
+	/*
+	 * find all collations, including builtin collations; we filter out
+	 * system-defined collations at dump-out time.
+	 */
+
+	/* Make sure we are in proper schema */
+	selectSourceSchema("pg_catalog");
+
+	appendPQExpBuffer(query, "SELECT tableoid, oid, collname, "
+					  "collnamespace, "
+					  "(%s collowner) AS rolname "
+					  "FROM pg_collation",
+					  username_subquery);
+
+	res = PQexec(g_conn, query->data);
+	check_sql_result(res, g_conn, query->data, PGRES_TUPLES_OK);
+
+	ntups = PQntuples(res);
+	*numCollations = ntups;
+
+	collinfo = (CollInfo *) malloc(ntups * sizeof(CollInfo));
+
+	i_tableoid = PQfnumber(res, "tableoid");
+	i_oid = PQfnumber(res, "oid");
+	i_collname = PQfnumber(res, "collname");
+	i_collnamespace = PQfnumber(res, "collnamespace");
+	i_rolname = PQfnumber(res, "rolname");
+
+	for (i = 0; i < ntups; i++)
+	{
+		collinfo[i].dobj.objType = DO_COLLATION;
+		collinfo[i].dobj.catId.tableoid = atooid(PQgetvalue(res, i, i_tableoid));
+		collinfo[i].dobj.catId.oid = atooid(PQgetvalue(res, i, i_oid));
+		AssignDumpId(&collinfo[i].dobj);
+		collinfo[i].dobj.name = strdup(PQgetvalue(res, i, i_collname));
+		collinfo[i].dobj.namespace = findNamespace(atooid(PQgetvalue(res, i, i_collnamespace)),
+												 collinfo[i].dobj.catId.oid);
+		collinfo[i].rolname = strdup(PQgetvalue(res, i, i_rolname));
+
+		/* Decide whether we want to dump it */
+		selectDumpableObject(&(collinfo[i].dobj));
+	}
+
+	PQclear(res);
+
+	destroyPQExpBuffer(query);
+
+	return collinfo;
+}
+
 /*
  * getConversions:
  *	  read all conversions in the system catalogs and return them in the
@@ -6763,6 +6842,9 @@ dumpDumpableObject(Archive *fout, DumpableObject *dobj)
 		case DO_OPFAMILY:
 			dumpOpfamily(fout, (OpfamilyInfo *) dobj);
 			break;
+		case DO_COLLATION:
+			dumpCollation(fout, (CollInfo *) dobj);
+			break;
 		case DO_CONVERSION:
 			dumpConversion(fout, (ConvInfo *) dobj);
 			break;
@@ -9926,6 +10008,111 @@ dumpOpfamily(Archive *fout, OpfamilyInfo *opfinfo)
 	destroyPQExpBuffer(labelq);
 }
 
+/*
+ * dumpCollation
+ *	  write out a single collation definition
+ */
+static void
+dumpCollation(Archive *fout, CollInfo *collinfo)
+{
+	PQExpBuffer query;
+	PQExpBuffer q;
+	PQExpBuffer delq;
+	PQExpBuffer labelq;
+	PGresult   *res;
+	int			ntups;
+	int			i_collname;
+	int			i_collcollate;
+	int			i_collctype;
+	const char *collname;
+	const char *collcollate;
+	const char *collctype;
+
+	/* Skip if not to be dumped */
+	if (!collinfo->dobj.dump || dataOnly)
+		return;
+
+	query = createPQExpBuffer();
+	q = createPQExpBuffer();
+	delq = createPQExpBuffer();
+	labelq = createPQExpBuffer();
+
+	/* Make sure we are in proper schema */
+	selectSourceSchema(collinfo->dobj.namespace->dobj.name);
+
+	/* Get conversion-specific details */
+	appendPQExpBuffer(query, "SELECT collname, "
+					  "collcollate, "
+					  "collctype "
+					  "FROM pg_catalog.pg_collation c "
+					  "WHERE c.oid = '%u'::pg_catalog.oid",
+					  collinfo->dobj.catId.oid);
+
+	res = PQexec(g_conn, query->data);
+	check_sql_result(res, g_conn, query->data, PGRES_TUPLES_OK);
+
+	/* Expecting a single result only */
+	ntups = PQntuples(res);
+	if (ntups != 1)
+	{
+		write_msg(NULL, ngettext("query returned %d row instead of one: %s\n",
+							   "query returned %d rows instead of one: %s\n",
+								 ntups),
+				  ntups, query->data);
+		exit_nicely();
+	}
+
+	i_collname = PQfnumber(res, "collname");
+	i_collcollate = PQfnumber(res, "collcollate");
+	i_collctype = PQfnumber(res, "collctype");
+
+	collname = PQgetvalue(res, 0, i_collname);
+	collcollate = PQgetvalue(res, 0, i_collcollate);
+	collctype = PQgetvalue(res, 0, i_collctype);
+
+	/*
+	 * DROP must be fully qualified in case same name appears in pg_catalog
+	 */
+	appendPQExpBuffer(delq, "DROP COLLATION %s",
+					  fmtId(collinfo->dobj.namespace->dobj.name));
+	appendPQExpBuffer(delq, ".%s;\n",
+					  fmtId(collinfo->dobj.name));
+
+	appendPQExpBuffer(q, "CREATE COLLATION %s (lc_collate = ",
+					  fmtId(collinfo->dobj.name));
+	appendStringLiteralAH(q, collcollate, fout);
+	appendPQExpBuffer(q, ", lc_ctype = ");
+	appendStringLiteralAH(q, collctype, fout);
+	appendPQExpBuffer(q, ");\n");
+
+	appendPQExpBuffer(labelq, "COLLATION %s", fmtId(collinfo->dobj.name));
+
+	if (binary_upgrade)
+		binary_upgrade_extension_member(q, &collinfo->dobj, labelq->data);
+
+	ArchiveEntry(fout, collinfo->dobj.catId, collinfo->dobj.dumpId,
+				 collinfo->dobj.name,
+				 collinfo->dobj.namespace->dobj.name,
+				 NULL,
+				 collinfo->rolname,
+				 false, "COLLATION", SECTION_PRE_DATA,
+				 q->data, delq->data, NULL,
+				 collinfo->dobj.dependencies, collinfo->dobj.nDeps,
+				 NULL, NULL);
+
+	/* Dump Collation Comments */
+	dumpComment(fout, labelq->data,
+				collinfo->dobj.namespace->dobj.name, collinfo->rolname,
+				collinfo->dobj.catId, 0, collinfo->dobj.dumpId);
+
+	PQclear(res);
+
+	destroyPQExpBuffer(query);
+	destroyPQExpBuffer(q);
+	destroyPQExpBuffer(delq);
+	destroyPQExpBuffer(labelq);
+}
+
 /*
  * dumpConversion
  *	  write out a single conversion definition

src/bin/pg_dump/pg_dump.h

@@ -117,7 +117,8 @@ typedef enum
 	DO_FOREIGN_SERVER,
 	DO_DEFAULT_ACL,
 	DO_BLOB,
-	DO_BLOB_DATA
+	DO_BLOB_DATA,
+	DO_COLLATION
 } DumpableObjectType;
 
 typedef struct _dumpableObject
@@ -217,6 +218,12 @@ typedef struct _opfamilyInfo
 	char	   *rolname;
 } OpfamilyInfo;
 
+typedef struct _collInfo
+{
+	DumpableObject dobj;
+	char	   *rolname;
+} CollInfo;
+
 typedef struct _convInfo
 {
 	DumpableObject dobj;
@@ -533,6 +540,7 @@ extern AggInfo *getAggregates(int *numAggregates);
 extern OprInfo *getOperators(int *numOperators);
 extern OpclassInfo *getOpclasses(int *numOpclasses);
 extern OpfamilyInfo *getOpfamilies(int *numOpfamilies);
+extern CollInfo *getCollations(int *numCollations);
 extern ConvInfo *getConversions(int *numConversions);
 extern TableInfo *getTables(int *numTables);
 extern InhInfo *getInherits(int *numInherits);

src/bin/pg_dump/pg_dump_sort.c

@@ -22,9 +22,9 @@ static const char *modulename = gettext_noop("sorter");
  * Sort priority for object types when dumping a pre-7.3 database.
  * Objects are sorted by priority levels, and within an equal priority level
  * by OID.	(This is a relatively crude hack to provide semi-reasonable
- * behavior for old databases without full dependency info.)  Note: extensions,
+ * behavior for old databases without full dependency info.)  Note: collations,
- * text search, foreign-data, and default ACL objects can't really happen here,
+ * extensions, text search, foreign-data, and default ACL objects can't really
- * so the rather bogus priorities for them don't matter.
+ * happen here, so the rather bogus priorities for them don't matter.
  */
 static const int oldObjectTypePriority[] =
 {
@@ -57,7 +57,8 @@ static const int oldObjectTypePriority[] =
 	4,							/* DO_FOREIGN_SERVER */
 	17,							/* DO_DEFAULT_ACL */
 	9,							/* DO_BLOB */
-	11							/* DO_BLOB_DATA */
+	11,							/* DO_BLOB_DATA */
+	2							/* DO_COLLATION */
 };
 
 /*
@@ -95,7 +96,8 @@ static const int newObjectTypePriority[] =
 	16,							/* DO_FOREIGN_SERVER */
 	28,							/* DO_DEFAULT_ACL */
 	20,							/* DO_BLOB */
-	22							/* DO_BLOB_DATA */
+	22,							/* DO_BLOB_DATA */
+	3							/* DO_COLLATION */
 };
 
 
@@ -1065,6 +1067,11 @@ describeDumpableObject(DumpableObject *obj, char *buf, int bufsize)
 					 "OPERATOR FAMILY %s  (ID %d OID %u)",
 					 obj->name, obj->dumpId, obj->catId.oid);
 			return;
+		case DO_COLLATION:
+			snprintf(buf, bufsize,
+					 "COLLATION %s  (ID %d OID %u)",
+					 obj->name, obj->dumpId, obj->catId.oid);
+			return;
 		case DO_CONVERSION:
 			snprintf(buf, bufsize,
 					 "CONVERSION %s  (ID %d OID %u)",

src/bin/psql/command.c

@@ -425,6 +425,9 @@ exec_command(const char *cmd,
 			case 'o':
 				success = describeOperators(pattern, show_system);
 				break;
+			case 'O':
+				success = listCollations(pattern, show_verbose, show_system);
+				break;
 			case 'p':
 				success = permissionsList(pattern);
 				break;

src/bin/psql/describe.c

@@ -627,7 +627,7 @@ listAllDbs(bool verbose)
 		appendPQExpBuffer(&buf,
 						  "       d.datcollate as \"%s\",\n"
 						  "       d.datctype as \"%s\",\n",
-						  gettext_noop("Collation"),
+						  gettext_noop("Collate"),
 						  gettext_noop("Ctype"));
 	appendPQExpBuffer(&buf, "       ");
 	printACLColumn(&buf, "d.datacl");
@@ -2856,6 +2856,66 @@ listCasts(const char *pattern)
 	return true;
 }
 
+/*
+ * \dO
+ *
+ * Describes collations
+ */
+bool
+listCollations(const char *pattern, bool verbose, bool showSystem)
+{
+	PQExpBufferData buf;
+	PGresult   *res;
+	printQueryOpt myopt = pset.popt;
+	static const bool translate_columns[] = {false, false, false, false, false};
+
+	initPQExpBuffer(&buf);
+
+	printfPQExpBuffer(&buf,
+					  "SELECT n.nspname AS \"%s\",\n"
+					  "       c.collname AS \"%s\",\n"
+					  "       c.collcollate AS \"%s\",\n"
+					  "       c.collctype AS \"%s\"",
+					  gettext_noop("Schema"),
+					  gettext_noop("Name"),
+					  gettext_noop("Collate"),
+					  gettext_noop("Ctype"));
+
+	if (verbose)
+		appendPQExpBuffer(&buf,
+		  ",\n  pg_catalog.obj_description(c.oid, 'pg_collation') AS \"%s\"",
+						  gettext_noop("Description"));
+
+	appendPQExpBuffer(&buf,
+					  "FROM pg_catalog.pg_collation c, pg_catalog.pg_namespace n\n"
+					  "WHERE n.oid = c.collnamespace\n");
+
+	if (!showSystem && !pattern)
+		appendPQExpBuffer(&buf, "      AND n.nspname <> 'pg_catalog'\n"
+						  "      AND n.nspname <> 'information_schema'\n");
+
+	processSQLNamePattern(pset.db, &buf, pattern, true, false,
+						  "n.nspname", "c.collname", NULL,
+						  "pg_catalog.pg_collation_is_visible(c.oid)");
+
+	appendPQExpBuffer(&buf, "ORDER BY 1, 2;");
+
+	res = PSQLexec(buf.data, false);
+	termPQExpBuffer(&buf);
+	if (!res)
+		return false;
+
+	myopt.nullPrint = NULL;
+	myopt.title = _("List of collations");
+	myopt.translate_header = true;
+	myopt.translate_columns = translate_columns;
+
+	printQuery(res, &myopt, pset.queryFout, pset.logfile);
+
+	PQclear(res);
+	return true;
+}
+
 /*
  * \dn
  *

src/bin/psql/describe.h

@@ -69,6 +69,9 @@ extern bool listConversions(const char *pattern, bool showSystem);
 /* \dC */
 extern bool listCasts(const char *pattern);
 
+/* \dO */
+extern bool listCollations(const char *pattern, bool verbose, bool showSystem);
+
 /* \dn */
 extern bool listSchemas(const char *pattern, bool verbose, bool showSystem);
 

src/bin/psql/help.c

@@ -214,6 +214,7 @@ slashUsage(unsigned short int pager)
 	fprintf(output, _("  \\dL[S+] [PATTERN]      list procedural languages\n"));
 	fprintf(output, _("  \\dn[S+] [PATTERN]      list schemas\n"));
 	fprintf(output, _("  \\do[S]  [PATTERN]      list operators\n"));
+	fprintf(output, _("  \\dO[S+] [PATTERN]      list collations\n"));
 	fprintf(output, _("  \\dp     [PATTERN]      list table, view, and sequence access privileges\n"));
 	fprintf(output, _("  \\drds [PATRN1 [PATRN2]] list per-database role settings\n"));
 	fprintf(output, _("  \\ds[S+] [PATTERN]      list sequences\n"));

src/bin/psql/tab-complete.c

@@ -606,6 +606,7 @@ static const pgsql_thing_t words_after_create[] = {
 	{"AGGREGATE", NULL, &Query_for_list_of_aggregates},
 	{"CAST", NULL, NULL},		/* Casts have complex structures for names, so
 								 * skip it */
+	{"COLLATION", "SELECT pg_catalog.quote_ident(collname) FROM pg_catalog.pg_collation WHERE collencoding = pg_char_to_encoding(getdatabaseencoding()) AND substring(pg_catalog.quote_ident(collname),1,%d)='%s'"},
 
 	/*
 	 * CREATE CONSTRAINT TRIGGER is not supported here because it is designed
@@ -797,7 +798,7 @@ psql_completion(char *text, int start, int end)
 			 pg_strcasecmp(prev3_wd, "TABLE") != 0)
 	{
 		static const char *const list_ALTER[] =
-		{"AGGREGATE", "CONVERSION", "DATABASE", "DEFAULT PRIVILEGES", "DOMAIN",
+		{"AGGREGATE", "COLLATION", "CONVERSION", "DATABASE", "DEFAULT PRIVILEGES", "DOMAIN",
 		 "EXTENSION", "FOREIGN DATA WRAPPER", "FOREIGN TABLE", "FUNCTION",
 		 "GROUP", "INDEX", "LANGUAGE", "LARGE OBJECT", "OPERATOR",
 		 "ROLE", "SCHEMA", "SERVER", "SEQUENCE", "TABLE",
@@ -843,6 +844,16 @@ psql_completion(char *text, int start, int end)
 		COMPLETE_WITH_LIST(list_ALTERGEN);
 	}
 
+	/* ALTER COLLATION <name> */
+	else if (pg_strcasecmp(prev3_wd, "ALTER") == 0 &&
+			 pg_strcasecmp(prev2_wd, "COLLATION") == 0)
+	{
+		static const char *const list_ALTERGEN[] =
+		{"OWNER TO", "RENAME TO", "SET SCHEMA", NULL};
+
+		COMPLETE_WITH_LIST(list_ALTERGEN);
+	}
+
 	/* ALTER CONVERSION <name> */
 	else if (pg_strcasecmp(prev3_wd, "ALTER") == 0 &&
 			 pg_strcasecmp(prev2_wd, "CONVERSION") == 0)
@@ -1521,7 +1532,7 @@ psql_completion(char *text, int start, int end)
 			 pg_strcasecmp(prev_wd, "ON") == 0)
 	{
 		static const char *const list_COMMENT[] =
-		{"CAST", "CONVERSION", "DATABASE", "FOREIGN TABLE", "INDEX", "LANGUAGE", "RULE", "SCHEMA",
+		{"CAST", "COLLATION", "CONVERSION", "DATABASE", "FOREIGN TABLE", "INDEX", "LANGUAGE", "RULE", "SCHEMA",
 			"SEQUENCE", "TABLE", "TYPE", "VIEW", "COLUMN", "AGGREGATE", "FUNCTION",
 			"OPERATOR", "TRIGGER", "CONSTRAINT", "DOMAIN", "LARGE OBJECT",
 		"TABLESPACE", "TEXT SEARCH", "ROLE", NULL};
@@ -1965,7 +1976,8 @@ psql_completion(char *text, int start, int end)
 
 	/* DROP object with CASCADE / RESTRICT */
 	else if ((pg_strcasecmp(prev3_wd, "DROP") == 0 &&
-			  (pg_strcasecmp(prev2_wd, "CONVERSION") == 0 ||
+			  (pg_strcasecmp(prev2_wd, "COLLATION") == 0 ||
+			   pg_strcasecmp(prev2_wd, "CONVERSION") == 0 ||
 			   pg_strcasecmp(prev2_wd, "DOMAIN") == 0 ||
 			   pg_strcasecmp(prev2_wd, "EXTENSION") == 0 ||
 			   pg_strcasecmp(prev2_wd, "FUNCTION") == 0 ||

src/include/catalog/catversion.h

@@ -53,6 +53,6 @@
  */
 
 /*							yyyymmddN */
-#define CATALOG_VERSION_NO	201102101
+#define CATALOG_VERSION_NO	201102121
 
 #endif

src/include/catalog/dependency.h

@@ -121,6 +121,7 @@ typedef enum ObjectClass
 	OCLASS_PROC,				/* pg_proc */
 	OCLASS_TYPE,				/* pg_type */
 	OCLASS_CAST,				/* pg_cast */
+	OCLASS_COLLATION,			/* pg_collation */
 	OCLASS_CONSTRAINT,			/* pg_constraint */
 	OCLASS_CONVERSION,			/* pg_conversion */
 	OCLASS_DEFAULT,				/* pg_attrdef */

src/include/catalog/pg_collation.h

@@ -32,6 +32,7 @@ CATALOG(pg_collation,3456)
 {
 	NameData	collname;		/* collation name */
 	Oid			collnamespace;	/* OID of namespace containing this collation */
+	Oid			collowner;
 	int4		collencoding;	/* encoding that this collation applies to */
 	NameData	collcollate;	/* LC_COLLATE setting */
 	NameData	collctype;		/* LC_CTYPE setting */
@@ -48,14 +49,15 @@ typedef FormData_pg_collation *Form_pg_collation;
  *		compiler constants for pg_collation
  * ----------------
  */
-#define Natts_pg_collation				5
+#define Natts_pg_collation				6
 #define Anum_pg_collation_collname		1
 #define Anum_pg_collation_collnamespace	2
-#define Anum_pg_collation_collencoding	3
+#define Anum_pg_collation_collowner		3
-#define Anum_pg_collation_collcollate	4
+#define Anum_pg_collation_collencoding	4
-#define Anum_pg_collation_collctype		5
+#define Anum_pg_collation_collcollate	5
+#define Anum_pg_collation_collctype		6
 
-DATA(insert OID = 100 ( default PGNSP 0 "" "" ));
+DATA(insert OID = 100 ( default PGNSP PGUID 0 "" "" ));
 DESCR("placeholder for default collation");
 #define DEFAULT_COLLATION_OID			100
 

src/include/catalog/pg_collation_fn.h

@@ -0,0 +1,23 @@
+/*-------------------------------------------------------------------------
+ *
+ * pg_collation_fn.h
+ *	 prototypes for functions in catalog/pg_collation.c
+ *
+ *
+ * Portions Copyright (c) 1996-2011, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/catalog/pg_collation_fn.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef PG_COLLATION_FN_H
+#define PG_COLLATION_FN_H
+
+extern Oid CollationCreate(const char *collname, Oid collnamespace,
+						   Oid collowner,
+						   int32 collencoding,
+						   const char *collcollate, const char *collctype);
+extern void RemoveCollationById(Oid collationOid);
+
+#endif   /* PG_COLLATION_FN_H */

src/include/catalog/pg_type_fn.h

@@ -68,6 +68,7 @@ extern void GenerateTypeDependencies(Oid typeNamespace,
 						 Oid elementType,
 						 bool isImplicitArray,
 						 Oid baseType,
+						 Oid typeCollation,
 						 Node *defaultExpr,
 						 bool rebuild);
 

src/include/commands/collationcmds.h

@@ -0,0 +1,28 @@
+/*-------------------------------------------------------------------------
+ *
+ * collationcmds.h
+ *	  prototypes for collationcmds.c.
+ *
+ *
+ * Portions Copyright (c) 1996-2011, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/commands/collationcmds.h
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#ifndef COLLATIONCMDS_H
+#define COLLATIONCMDS_H
+
+#include "nodes/parsenodes.h"
+
+extern void DefineCollation(List *names, List *parameters);
+extern void DropCollationsCommand(DropStmt *drop);
+extern void RenameCollation(List *name, const char *newname);
+extern void AlterCollationOwner(List *name, Oid newOwnerId);
+extern void AlterCollationOwner_oid(Oid collationOid, Oid newOwnerId);
+extern void AlterCollationNamespace(List *name, const char *newschema);
+extern Oid AlterCollationNamespace_oid(Oid collOid, Oid newNspOid);
+
+#endif   /* COLLATIONCMDS_H */

src/include/commands/dbcommands.h

@@ -65,4 +65,6 @@ extern char *get_database_name(Oid dbid);
 extern void dbase_redo(XLogRecPtr lsn, XLogRecord *rptr);
 extern void dbase_desc(StringInfo buf, uint8 xl_info, char *rec);
 
+extern void check_encoding_locale_matches(int encoding, const char *collate, const char *ctype);
+
 #endif   /* DBCOMMANDS_H */

src/include/nodes/parsenodes.h

@@ -1070,6 +1070,7 @@ typedef enum ObjectType
 	OBJECT_CAST,
 	OBJECT_COLUMN,
 	OBJECT_CONSTRAINT,
+	OBJECT_COLLATION,
 	OBJECT_CONVERSION,
 	OBJECT_DATABASE,
 	OBJECT_DOMAIN,

src/include/parser/kwlist.h

@@ -79,6 +79,7 @@ PG_KEYWORD("close", CLOSE, UNRESERVED_KEYWORD)
 PG_KEYWORD("cluster", CLUSTER, UNRESERVED_KEYWORD)
 PG_KEYWORD("coalesce", COALESCE, COL_NAME_KEYWORD)
 PG_KEYWORD("collate", COLLATE, RESERVED_KEYWORD)
+PG_KEYWORD("collation", COLLATION, UNRESERVED_KEYWORD)
 PG_KEYWORD("column", COLUMN, RESERVED_KEYWORD)
 PG_KEYWORD("comment", COMMENT, UNRESERVED_KEYWORD)
 PG_KEYWORD("comments", COMMENTS, UNRESERVED_KEYWORD)

src/include/utils/acl.h

@@ -188,6 +188,7 @@ typedef enum AclObjectKind
 	ACL_KIND_NAMESPACE,			/* pg_namespace */
 	ACL_KIND_OPCLASS,			/* pg_opclass */
 	ACL_KIND_OPFAMILY,			/* pg_opfamily */
+	ACL_KIND_COLLATION,			/* pg_collation */
 	ACL_KIND_CONVERSION,		/* pg_conversion */
 	ACL_KIND_TABLESPACE,		/* pg_tablespace */
 	ACL_KIND_TSDICTIONARY,		/* pg_ts_dict */
@@ -309,6 +310,7 @@ extern bool pg_tablespace_ownercheck(Oid spc_oid, Oid roleid);
 extern bool pg_opclass_ownercheck(Oid opc_oid, Oid roleid);
 extern bool pg_opfamily_ownercheck(Oid opf_oid, Oid roleid);
 extern bool pg_database_ownercheck(Oid db_oid, Oid roleid);
+extern bool pg_collation_ownercheck(Oid coll_oid, Oid roleid);
 extern bool pg_conversion_ownercheck(Oid conv_oid, Oid roleid);
 extern bool pg_ts_dict_ownercheck(Oid dict_oid, Oid roleid);
 extern bool pg_ts_config_ownercheck(Oid cfg_oid, Oid roleid);

src/test/regress/expected/collate.linux.utf8.out

@@ -732,3 +732,96 @@ SELECT relname, pg_get_indexdef(oid) FROM pg_class WHERE relname LIKE 'collate_t
  collate_test1_idx3 | CREATE INDEX collate_test1_idx3 ON collate_test1 USING btree (((b COLLATE "C")) COLLATE "C")
 (3 rows)
 
+-- schema manipulation commands
+CREATE ROLE regress_test_role;
+CREATE SCHEMA test_schema;
+CREATE COLLATION test0 (locale = 'en_US.utf8');
+CREATE COLLATION test0 (locale = 'en_US.utf8'); -- fail
+ERROR:  collation "test0" for encoding "UTF8" already exists
+CREATE COLLATION test1 (lc_collate = 'en_US.utf8', lc_ctype = 'de_DE.utf8');
+CREATE COLLATION test2 (locale = 'en_US'); -- fail
+ERROR:  encoding UTF8 does not match locale en_US
+DETAIL:  The chosen LC_CTYPE setting requires encoding LATIN1.
+CREATE COLLATION test3 (lc_collate = 'en_US.utf8'); -- fail
+ERROR:  parameter "lc_ctype" must be specified
+CREATE COLLATION test4 FROM nonsense;
+ERROR:  collation "nonsense" for current database encoding "UTF8" does not exist
+CREATE COLLATION test5 FROM test0;
+SELECT collname, collencoding, collcollate, collctype FROM pg_collation WHERE collname LIKE 'test%' ORDER BY 1;
+ collname | collencoding | collcollate | collctype  
+----------+--------------+-------------+------------
+ test0    |            6 | en_US.utf8  | en_US.utf8
+ test1    |            6 | en_US.utf8  | de_DE.utf8
+ test5    |            6 | en_US.utf8  | en_US.utf8
+(3 rows)
+
+ALTER COLLATION test1 RENAME TO test11;
+ALTER COLLATION test0 RENAME TO test11; -- fail
+ERROR:  collation "test11" for current database encoding "UTF8" already exists in schema "public"
+ALTER COLLATION test1 RENAME TO test22; -- fail
+ERROR:  collation "test1" for current database encoding "UTF8" does not exist
+ALTER COLLATION test11 OWNER TO regress_test_role;
+ALTER COLLATION test11 OWNER TO nonsense;
+ERROR:  role "nonsense" does not exist
+ALTER COLLATION test11 SET SCHEMA test_schema;
+COMMENT ON COLLATION test0 IS 'US English';
+SELECT collname, nspname, obj_description(pg_collation.oid, 'pg_collation')
+    FROM pg_collation JOIN pg_namespace ON (collnamespace = pg_namespace.oid)
+    WHERE collname LIKE 'test%'
+    ORDER BY 1;
+ collname |   nspname   | obj_description 
+----------+-------------+-----------------
+ test0    | public      | US English
+ test11   | test_schema | 
+ test5    | public      | 
+(3 rows)
+
+DROP COLLATION test0, test_schema.test11, test5;
+DROP COLLATION test0; -- fail
+ERROR:  collation "test0" for current database encoding "UTF8" does not exist
+DROP COLLATION IF EXISTS test0;
+NOTICE:  collation "test0" does not exist, skipping
+SELECT collname FROM pg_collation WHERE collname LIKE 'test%';
+ collname 
+----------
+(0 rows)
+
+DROP SCHEMA test_schema;
+DROP ROLE regress_test_role;
+-- dependencies
+CREATE COLLATION test0 (locale = 'en_US.utf8');
+CREATE TABLE collate_dep_test1 (a int, b text COLLATE test0);
+CREATE DOMAIN collate_dep_dom1 AS text COLLATE test0;
+CREATE TYPE collate_dep_test2 AS (x int, y text COLLATE test0);
+CREATE VIEW collate_dep_test3 AS SELECT text 'foo' COLLATE test0 AS foo;
+CREATE TABLE collate_dep_test4t (a int, b text);
+CREATE INDEX collate_dep_test4i ON collate_dep_test4t (b COLLATE test0);
+DROP COLLATION test0 RESTRICT; -- fail
+ERROR:  cannot drop collation test0 because other objects depend on it
+DETAIL:  table collate_dep_test1 column b depends on collation test0
+type collate_dep_dom1 depends on collation test0
+composite type collate_dep_test2 column y depends on collation test0
+view collate_dep_test3 depends on collation test0
+index collate_dep_test4i depends on collation test0
+HINT:  Use DROP ... CASCADE to drop the dependent objects too.
+DROP COLLATION test0 CASCADE;
+NOTICE:  drop cascades to 5 other objects
+DETAIL:  drop cascades to table collate_dep_test1 column b
+drop cascades to type collate_dep_dom1
+drop cascades to composite type collate_dep_test2 column y
+drop cascades to view collate_dep_test3
+drop cascades to index collate_dep_test4i
+\d collate_dep_test1
+Table "public.collate_dep_test1"
+ Column |  Type   | Modifiers 
+--------+---------+-----------
+ a      | integer | 
+
+\d collate_dep_test2
+Composite type "public.collate_dep_test2"
+ Column |  Type   
+--------+---------
+ x      | integer
+
+DROP TABLE collate_dep_test1, collate_dep_test4t;
+DROP TYPE collate_dep_test2;

src/test/regress/sql/collate.linux.utf8.sql

@@ -222,3 +222,65 @@ CREATE INDEX collate_test1_idx4 ON collate_test1 (a COLLATE "C"); -- fail
 CREATE INDEX collate_test1_idx5 ON collate_test1 ((a COLLATE "C")); -- fail
 
 SELECT relname, pg_get_indexdef(oid) FROM pg_class WHERE relname LIKE 'collate_test%_idx%';
+
+
+-- schema manipulation commands
+
+CREATE ROLE regress_test_role;
+CREATE SCHEMA test_schema;
+
+CREATE COLLATION test0 (locale = 'en_US.utf8');
+CREATE COLLATION test0 (locale = 'en_US.utf8'); -- fail
+CREATE COLLATION test1 (lc_collate = 'en_US.utf8', lc_ctype = 'de_DE.utf8');
+CREATE COLLATION test2 (locale = 'en_US'); -- fail
+CREATE COLLATION test3 (lc_collate = 'en_US.utf8'); -- fail
+
+CREATE COLLATION test4 FROM nonsense;
+CREATE COLLATION test5 FROM test0;
+
+SELECT collname, collencoding, collcollate, collctype FROM pg_collation WHERE collname LIKE 'test%' ORDER BY 1;
+
+ALTER COLLATION test1 RENAME TO test11;
+ALTER COLLATION test0 RENAME TO test11; -- fail
+ALTER COLLATION test1 RENAME TO test22; -- fail
+
+ALTER COLLATION test11 OWNER TO regress_test_role;
+ALTER COLLATION test11 OWNER TO nonsense;
+ALTER COLLATION test11 SET SCHEMA test_schema;
+
+COMMENT ON COLLATION test0 IS 'US English';
+
+SELECT collname, nspname, obj_description(pg_collation.oid, 'pg_collation')
+    FROM pg_collation JOIN pg_namespace ON (collnamespace = pg_namespace.oid)
+    WHERE collname LIKE 'test%'
+    ORDER BY 1;
+
+DROP COLLATION test0, test_schema.test11, test5;
+DROP COLLATION test0; -- fail
+DROP COLLATION IF EXISTS test0;
+
+SELECT collname FROM pg_collation WHERE collname LIKE 'test%';
+
+DROP SCHEMA test_schema;
+DROP ROLE regress_test_role;
+
+
+-- dependencies
+
+CREATE COLLATION test0 (locale = 'en_US.utf8');
+
+CREATE TABLE collate_dep_test1 (a int, b text COLLATE test0);
+CREATE DOMAIN collate_dep_dom1 AS text COLLATE test0;
+CREATE TYPE collate_dep_test2 AS (x int, y text COLLATE test0);
+CREATE VIEW collate_dep_test3 AS SELECT text 'foo' COLLATE test0 AS foo;
+CREATE TABLE collate_dep_test4t (a int, b text);
+CREATE INDEX collate_dep_test4i ON collate_dep_test4t (b COLLATE test0);
+
+DROP COLLATION test0 RESTRICT; -- fail
+DROP COLLATION test0 CASCADE;
+
+\d collate_dep_test1
+\d collate_dep_test2
+
+DROP TABLE collate_dep_test1, collate_dep_test4t;
+DROP TYPE collate_dep_test2;