Per-column collation support

This adds collation support for columns and domains, a COLLATE clause
to override it per expression, and B-tree index support.

Peter Eisentraut
reviewed by Pavel Stehule, Itagaki Takahiro, Robert Haas, Noah Misch

src/backend/access/common/scankey.c

@@ -103,3 +103,16 @@ ScanKeyEntryInitializeWithInfo(ScanKey entry,
 	entry->sk_argument = argument;
 	fmgr_info_copy(&entry->sk_func, finfo, CurrentMemoryContext);
 }
+
+/*
+ * ScanKeyEntryInitializeCollation
+ *
+ * Initialize the collation of a scan key.  This is just a notational
+ * convenience and small abstraction.
+ */
+void
+ScanKeyEntryInitializeCollation(ScanKey entry,
+								Oid collation)
+{
+	entry->sk_func.fn_collation = collation;
+}

src/backend/access/common/tupdesc.c

@@ -488,10 +488,32 @@ TupleDescInitEntry(TupleDesc desc,
 	att->attbyval = typeForm->typbyval;
 	att->attalign = typeForm->typalign;
 	att->attstorage = typeForm->typstorage;
+	att->attcollation = typeForm->typcollation;
 
 	ReleaseSysCache(tuple);
 }
 
+/*
+ * TupleDescInitEntryCollation
+ *
+ * Fill in the collation for an attribute in a previously initialized
+ * tuple descriptor.
+ */
+void
+TupleDescInitEntryCollation(TupleDesc desc,
+							AttrNumber attributeNumber,
+							Oid collationid)
+{
+	/*
+	 * sanity checks
+	 */
+	AssertArg(PointerIsValid(desc));
+	AssertArg(attributeNumber >= 1);
+	AssertArg(attributeNumber <= desc->natts);
+
+	desc->attrs[attributeNumber - 1]->attcollation = collationid;
+}
+
 
 /*
  * BuildDescForRelation
@@ -513,6 +535,7 @@ BuildDescForRelation(List *schema)
 	char	   *attname;
 	Oid			atttypid;
 	int32		atttypmod;
+	Oid			attcollation;
 	int			attdim;
 
 	/*
@@ -536,7 +559,7 @@ BuildDescForRelation(List *schema)
 		attnum++;
 
 		attname = entry->colname;
-		typenameTypeIdAndMod(NULL, entry->typeName, &atttypid, &atttypmod);
+		typenameTypeIdModColl(NULL, entry->typeName, &atttypid, &atttypmod, &attcollation);
 		attdim = list_length(entry->typeName->arrayBounds);
 
 		if (entry->typeName->setof)
@@ -547,6 +570,7 @@ BuildDescForRelation(List *schema)
 
 		TupleDescInitEntry(desc, attnum, attname,
 						   atttypid, atttypmod, attdim);
+		TupleDescInitEntryCollation(desc, attnum, attcollation);
 
 		/* Override TupleDescInitEntry's settings as requested */
 		if (entry->storage)
@@ -588,18 +612,20 @@ BuildDescForRelation(List *schema)
  * with functions returning RECORD.
  */
 TupleDesc
-BuildDescFromLists(List *names, List *types, List *typmods)
+BuildDescFromLists(List *names, List *types, List *typmods, List *collations)
 {
 	int			natts;
 	AttrNumber	attnum;
 	ListCell   *l1;
 	ListCell   *l2;
 	ListCell   *l3;
+	ListCell   *l4;
 	TupleDesc	desc;
 
 	natts = list_length(names);
 	Assert(natts == list_length(types));
 	Assert(natts == list_length(typmods));
+	Assert(natts == list_length(collations));
 
 	/*
 	 * allocate a new tuple descriptor
@@ -610,20 +636,25 @@ BuildDescFromLists(List *names, List *types, List *typmods)
 
 	l2 = list_head(types);
 	l3 = list_head(typmods);
+	l4 = list_head(collations);
 	foreach(l1, names)
 	{
 		char	   *attname = strVal(lfirst(l1));
 		Oid			atttypid;
 		int32		atttypmod;
+		Oid			attcollation;
 
 		atttypid = lfirst_oid(l2);
 		l2 = lnext(l2);
 		atttypmod = lfirst_int(l3);
 		l3 = lnext(l3);
+		attcollation = lfirst_oid(l4);
+		l4 = lnext(l4);
 
 		attnum++;
 
 		TupleDescInitEntry(desc, attnum, attname, atttypid, atttypmod, 0);
+		TupleDescInitEntryCollation(desc, attnum, attcollation);
 	}
 
 	return desc;

src/backend/access/gin/ginutil.c

@@ -22,6 +22,7 @@
 #include "storage/freespace.h"
 #include "storage/indexfsm.h"
 #include "storage/lmgr.h"
+#include "utils/lsyscache.h"
 
 
 /*
@@ -60,6 +61,8 @@ initGinState(GinState *state, Relation index)
 		fmgr_info_copy(&(state->compareFn[i]),
 					   index_getprocinfo(index, i + 1, GIN_COMPARE_PROC),
 					   CurrentMemoryContext);
+		fmgr_info_collation(get_typcollation(index->rd_att->attrs[i]->atttypid),
+							&(state->compareFn[i]));
 		fmgr_info_copy(&(state->extractValueFn[i]),
 					   index_getprocinfo(index, i + 1, GIN_EXTRACTVALUE_PROC),
 					   CurrentMemoryContext);

src/backend/access/index/indexam.c

@@ -872,6 +872,8 @@ index_getprocinfo(Relation irel,
 				 procnum, attnum, RelationGetRelationName(irel));
 
 		fmgr_info_cxt(procId, locinfo, irel->rd_indexcxt);
+		fmgr_info_collation(irel->rd_index->indcollation.values[attnum-1],
+							locinfo);
 	}
 
 	return locinfo;

src/backend/access/nbtree/nbtsearch.c

@@ -723,6 +723,8 @@ _bt_first(IndexScanDesc scan, ScanDirection dir)
 											   cur->sk_subtype,
 											   procinfo,
 											   cur->sk_argument);
+				ScanKeyEntryInitializeCollation(scankeys + i,
+												cur->sk_func.fn_collation);
 			}
 			else
 			{
@@ -743,6 +745,8 @@ _bt_first(IndexScanDesc scan, ScanDirection dir)
 									   cur->sk_subtype,
 									   cmp_proc,
 									   cur->sk_argument);
+				ScanKeyEntryInitializeCollation(scankeys + i,
+												cur->sk_func.fn_collation);
 			}
 		}
 	}