Fix collation handling in plpgsql functions.

Make plpgsql treat the input collation as a polymorphism variable, so
that we cache separate plans for each input collation that's used in a
particular session, as per recent discussion.  Propagate the input
collation to all collatable input parameters.

I chose to also propagate the input collation to all declared variables of
collatable types, which is a bit more debatable but seems to be necessary
for non-astonishing behavior.  (Copying a parameter into a separate local
variable shouldn't result in a change of behavior, for example.)  There is
enough infrastructure here to support declaring a collation for each local
variable to override that default, but I thought we should wait to see what
the field demand is before adding such a feature.

In passing, remove exec_get_rec_fieldtype(), which wasn't used anywhere.

Documentation patch to follow.

src/pl/plpgsql/src/gram.y

@@ -487,7 +487,8 @@ decl_statement	: decl_varname decl_const decl_datatype decl_notnull decl_defval
 						new = (PLpgSQL_var *)
 							plpgsql_build_variable($1.name, $1.lineno,
 												   plpgsql_build_datatype(REFCURSOROID,
-																		  -1),
+																		  -1,
+																		  InvalidOid),
 												   true);
 
 						curname_def = palloc0(sizeof(PLpgSQL_expr));
@@ -1248,7 +1249,8 @@ for_control		: for_variable K_IN
 									plpgsql_build_variable($1.name,
 														   $1.lineno,
 														   plpgsql_build_datatype(INT4OID,
-																				  -1),
+																				  -1,
+																				  InvalidOid),
 														   true);
 
 								new = palloc0(sizeof(PLpgSQL_stmt_fori));
@@ -1932,13 +1934,17 @@ exception_sect	:
 						PLpgSQL_variable *var;
 
 						var = plpgsql_build_variable("sqlstate", lineno,
-													 plpgsql_build_datatype(TEXTOID, -1),
+													 plpgsql_build_datatype(TEXTOID,
+																			-1,
+																			plpgsql_curr_compile->fn_input_collation),
 													 true);
 						((PLpgSQL_var *) var)->isconst = true;
 						new->sqlstate_varno = var->dno;
 
 						var = plpgsql_build_variable("sqlerrm", lineno,
-													 plpgsql_build_datatype(TEXTOID, -1),
+													 plpgsql_build_datatype(TEXTOID,
+																			-1,
+																			plpgsql_curr_compile->fn_input_collation),
 													 true);
 						((PLpgSQL_var *) var)->isconst = true;
 						new->sqlerrm_varno = var->dno;
@@ -3227,7 +3233,8 @@ parse_datatype(const char *string, int location)
 	error_context_stack = syntax_errcontext.previous;
 
 	/* Okay, build a PLpgSQL_type data structure for it */
-	return plpgsql_build_datatype(type_id, typmod);
+	return plpgsql_build_datatype(type_id, typmod,
+								  plpgsql_curr_compile->fn_input_collation);
 }
 
 /*
@@ -3400,7 +3407,9 @@ make_case(int location, PLpgSQL_expr *t_expr,
 		 */
 		t_var = (PLpgSQL_var *)
 			plpgsql_build_variable(varname, new->lineno,
-								   plpgsql_build_datatype(INT4OID, -1),
+								   plpgsql_build_datatype(INT4OID,
+														  -1,
+														  InvalidOid),
 								   true);
 		new->t_varno = t_var->dno;
 

src/pl/plpgsql/src/pl_exec.c

@@ -1516,7 +1516,9 @@ exec_stmt_case(PLpgSQL_execstate *estate, PLpgSQL_stmt_case *stmt)
 		 * this doesn't affect the originally stored function parse tree.
 		 */
 		if (t_var->datatype->typoid != t_oid)
-			t_var->datatype = plpgsql_build_datatype(t_oid, -1);
+			t_var->datatype = plpgsql_build_datatype(t_oid,
+													 -1,
+													 estate->func->fn_input_collation);
 
 		/* now we can assign to the variable */
 		exec_assign_value(estate,
@@ -4307,33 +4309,64 @@ exec_get_datum_type(PLpgSQL_execstate *estate,
 }
 
 /*
- * exec_get_rec_fieldtype				Get datatype of a PLpgSQL record field
- *
- * Also returns the field number to *fieldno.
+ * exec_get_datum_collation				Get collation of a PLpgSQL_datum
  */
 Oid
-exec_get_rec_fieldtype(PLpgSQL_rec *rec, const char *fieldname,
-					   int *fieldno)
+exec_get_datum_collation(PLpgSQL_execstate *estate,
+						 PLpgSQL_datum *datum)
 {
-	Oid			typeid;
-	int			fno;
+	Oid			collid;
 
-	if (rec->tupdesc == NULL)
-		ereport(ERROR,
-				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-				 errmsg("record \"%s\" is not assigned yet",
-						rec->refname),
-				 errdetail("The tuple structure of a not-yet-assigned record is indeterminate.")));
-	fno = SPI_fnumber(rec->tupdesc, fieldname);
-	if (fno == SPI_ERROR_NOATTRIBUTE)
-		ereport(ERROR,
-				(errcode(ERRCODE_UNDEFINED_COLUMN),
-				 errmsg("record \"%s\" has no field \"%s\"",
-						rec->refname, fieldname)));
-	typeid = SPI_gettypeid(rec->tupdesc, fno);
+	switch (datum->dtype)
+	{
+		case PLPGSQL_DTYPE_VAR:
+			{
+				PLpgSQL_var *var = (PLpgSQL_var *) datum;
 
-	*fieldno = fno;
-	return typeid;
+				collid = var->datatype->collation;
+				break;
+			}
+
+		case PLPGSQL_DTYPE_ROW:
+		case PLPGSQL_DTYPE_REC:
+			/* composite types are never collatable */
+			collid = InvalidOid;
+			break;
+
+		case PLPGSQL_DTYPE_RECFIELD:
+			{
+				PLpgSQL_recfield *recfield = (PLpgSQL_recfield *) datum;
+				PLpgSQL_rec *rec;
+				int			fno;
+
+				rec = (PLpgSQL_rec *) (estate->datums[recfield->recparentno]);
+				if (rec->tupdesc == NULL)
+					ereport(ERROR,
+						  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+						   errmsg("record \"%s\" is not assigned yet",
+								  rec->refname),
+						   errdetail("The tuple structure of a not-yet-assigned record is indeterminate.")));
+				fno = SPI_fnumber(rec->tupdesc, recfield->fieldname);
+				if (fno == SPI_ERROR_NOATTRIBUTE)
+					ereport(ERROR,
+							(errcode(ERRCODE_UNDEFINED_COLUMN),
+							 errmsg("record \"%s\" has no field \"%s\"",
+									rec->refname, recfield->fieldname)));
+				/* XXX there's no SPI_getcollid, as yet */
+				if (fno > 0)
+					collid = rec->tupdesc->attrs[fno - 1]->attcollation;
+				else			/* no system column types have collation */
+					collid = InvalidOid;
+				break;
+			}
+
+		default:
+			elog(ERROR, "unrecognized dtype: %d", datum->dtype);
+			collid = InvalidOid;	/* keep compiler quiet */
+			break;
+	}
+
+	return collid;
 }
 
 /* ----------

src/pl/plpgsql/src/plpgsql.h

@@ -167,6 +167,7 @@ typedef struct
 	bool		typbyval;
 	Oid			typrelid;
 	Oid			typioparam;
+	Oid			collation;		/* from pg_type, but can be overridden */
 	FmgrInfo	typinput;		/* lookup info for typinput function */
 	int32		atttypmod;		/* typmod (taken from someplace else) */
 } PLpgSQL_type;
@@ -634,12 +635,19 @@ typedef struct PLpgSQL_func_hashkey
 
 	/*
 	 * For a trigger function, the OID of the relation triggered on is part of
-	 * the hashkey --- we want to compile the trigger separately for each
+	 * the hash key --- we want to compile the trigger separately for each
 	 * relation it is used with, in case the rowtype is different.	Zero if
 	 * not called as a trigger.
 	 */
 	Oid			trigrelOid;
 
+	/*
+	 * We must include the input collation as part of the hash key too,
+	 * because we have to generate different plans (with different Param
+	 * collations) for different collation settings.
+	 */
+	Oid			inputCollation;
+
 	/*
 	 * We include actual argument types in the hash key to support polymorphic
 	 * PLpgSQL functions.  Be careful that extra positions are zeroed!
@@ -655,6 +663,7 @@ typedef struct PLpgSQL_function
 	TransactionId fn_xmin;
 	ItemPointerData fn_tid;
 	bool		fn_is_trigger;
+	Oid			fn_input_collation;
 	PLpgSQL_func_hashkey *fn_hashkey;	/* back-link to hashtable key */
 	MemoryContext fn_cxt;
 
@@ -860,7 +869,8 @@ extern PLpgSQL_type *plpgsql_parse_wordtype(char *ident);
 extern PLpgSQL_type *plpgsql_parse_cwordtype(List *idents);
 extern PLpgSQL_type *plpgsql_parse_wordrowtype(char *ident);
 extern PLpgSQL_type *plpgsql_parse_cwordrowtype(List *idents);
-extern PLpgSQL_type *plpgsql_build_datatype(Oid typeOid, int32 typmod);
+extern PLpgSQL_type *plpgsql_build_datatype(Oid typeOid, int32 typmod,
+											Oid collation);
 extern PLpgSQL_variable *plpgsql_build_variable(const char *refname, int lineno,
 					   PLpgSQL_type *dtype,
 					   bool add2namespace);
@@ -895,8 +905,8 @@ extern void plpgsql_subxact_cb(SubXactEvent event, SubTransactionId mySubid,
 				   SubTransactionId parentSubid, void *arg);
 extern Oid exec_get_datum_type(PLpgSQL_execstate *estate,
 					PLpgSQL_datum *datum);
-extern Oid exec_get_rec_fieldtype(PLpgSQL_rec *rec, const char *fieldname,
-					   int *fieldno);
+extern Oid exec_get_datum_collation(PLpgSQL_execstate *estate,
+					PLpgSQL_datum *datum);
 
 /* ----------
  * Functions for namespace handling in pl_funcs.c

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

@@ -686,57 +686,61 @@ SELECT a, CAST(b AS varchar) FROM collate_test3 ORDER BY 2;
  2 | äbc
 (4 rows)
 
--- propagation of collation in inlined and non-inlined cases
+-- propagation of collation in SQL functions (inlined and non-inlined cases)
+-- and plpgsql functions too
 CREATE FUNCTION mylt (text, text) RETURNS boolean LANGUAGE sql
     AS $$ select $1 < $2 $$;
 CREATE FUNCTION mylt_noninline (text, text) RETURNS boolean LANGUAGE sql
     AS $$ select $1 < $2 limit 1 $$;
+CREATE FUNCTION mylt_plpgsql (text, text) RETURNS boolean LANGUAGE plpgsql
+    AS $$ begin return $1 < $2; end $$;
 SELECT a.b AS a, b.b AS b, a.b < b.b AS lt,
-       mylt(a.b, b.b), mylt_noninline(a.b, b.b)
+       mylt(a.b, b.b), mylt_noninline(a.b, b.b), mylt_plpgsql(a.b, b.b)
 FROM collate_test1 a, collate_test1 b
 ORDER BY a.b, b.b;
-  a  |  b  | lt | mylt | mylt_noninline 
------+-----+----+------+----------------
- abc | abc | f  | f    | f
- abc | ABC | t  | t    | t
- abc | äbc | t  | t    | t
- abc | bbc | t  | t    | t
- ABC | abc | f  | f    | f
- ABC | ABC | f  | f    | f
- ABC | äbc | t  | t    | t
- ABC | bbc | t  | t    | t
- äbc | abc | f  | f    | f
- äbc | ABC | f  | f    | f
- äbc | äbc | f  | f    | f
- äbc | bbc | t  | t    | t
- bbc | abc | f  | f    | f
- bbc | ABC | f  | f    | f
- bbc | äbc | f  | f    | f
- bbc | bbc | f  | f    | f
+  a  |  b  | lt | mylt | mylt_noninline | mylt_plpgsql 
+-----+-----+----+------+----------------+--------------
+ abc | abc | f  | f    | f              | f
+ abc | ABC | t  | t    | t              | t
+ abc | äbc | t  | t    | t              | t
+ abc | bbc | t  | t    | t              | t
+ ABC | abc | f  | f    | f              | f
+ ABC | ABC | f  | f    | f              | f
+ ABC | äbc | t  | t    | t              | t
+ ABC | bbc | t  | t    | t              | t
+ äbc | abc | f  | f    | f              | f
+ äbc | ABC | f  | f    | f              | f
+ äbc | äbc | f  | f    | f              | f
+ äbc | bbc | t  | t    | t              | t
+ bbc | abc | f  | f    | f              | f
+ bbc | ABC | f  | f    | f              | f
+ bbc | äbc | f  | f    | f              | f
+ bbc | bbc | f  | f    | f              | f
 (16 rows)
 
 SELECT a.b AS a, b.b AS b, a.b < b.b COLLATE "C" AS lt,
-       mylt(a.b, b.b COLLATE "C"), mylt_noninline(a.b, b.b COLLATE "C")
+       mylt(a.b, b.b COLLATE "C"), mylt_noninline(a.b, b.b COLLATE "C"),
+       mylt_plpgsql(a.b, b.b COLLATE "C")
 FROM collate_test1 a, collate_test1 b
 ORDER BY a.b, b.b;
-  a  |  b  | lt | mylt | mylt_noninline 
------+-----+----+------+----------------
- abc | abc | f  | f    | f
- abc | ABC | f  | f    | f
- abc | äbc | t  | t    | t
- abc | bbc | t  | t    | t
- ABC | abc | t  | t    | t
- ABC | ABC | f  | f    | f
- ABC | äbc | t  | t    | t
- ABC | bbc | t  | t    | t
- äbc | abc | f  | f    | f
- äbc | ABC | f  | f    | f
- äbc | äbc | f  | f    | f
- äbc | bbc | f  | f    | f
- bbc | abc | f  | f    | f
- bbc | ABC | f  | f    | f
- bbc | äbc | t  | t    | t
- bbc | bbc | f  | f    | f
+  a  |  b  | lt | mylt | mylt_noninline | mylt_plpgsql 
+-----+-----+----+------+----------------+--------------
+ abc | abc | f  | f    | f              | f
+ abc | ABC | f  | f    | f              | f
+ abc | äbc | t  | t    | t              | t
+ abc | bbc | t  | t    | t              | t
+ ABC | abc | t  | t    | t              | t
+ ABC | ABC | f  | f    | f              | f
+ ABC | äbc | t  | t    | t              | t
+ ABC | bbc | t  | t    | t              | t
+ äbc | abc | f  | f    | f              | f
+ äbc | ABC | f  | f    | f              | f
+ äbc | äbc | f  | f    | f              | f
+ äbc | bbc | f  | f    | f              | f
+ bbc | abc | f  | f    | f              | f
+ bbc | ABC | f  | f    | f              | f
+ bbc | äbc | t  | t    | t              | t
+ bbc | bbc | f  | f    | f              | f
 (16 rows)
 
 -- polymorphism

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

@@ -212,7 +212,8 @@ SELECT a, CAST(b AS varchar) FROM collate_test2 ORDER BY 2;
 SELECT a, CAST(b AS varchar) FROM collate_test3 ORDER BY 2;
 
 
--- propagation of collation in inlined and non-inlined cases
+-- propagation of collation in SQL functions (inlined and non-inlined cases)
+-- and plpgsql functions too
 
 CREATE FUNCTION mylt (text, text) RETURNS boolean LANGUAGE sql
     AS $$ select $1 < $2 $$;
@@ -220,13 +221,17 @@ CREATE FUNCTION mylt (text, text) RETURNS boolean LANGUAGE sql
 CREATE FUNCTION mylt_noninline (text, text) RETURNS boolean LANGUAGE sql
     AS $$ select $1 < $2 limit 1 $$;
 
+CREATE FUNCTION mylt_plpgsql (text, text) RETURNS boolean LANGUAGE plpgsql
+    AS $$ begin return $1 < $2; end $$;
+
 SELECT a.b AS a, b.b AS b, a.b < b.b AS lt,
-       mylt(a.b, b.b), mylt_noninline(a.b, b.b)
+       mylt(a.b, b.b), mylt_noninline(a.b, b.b), mylt_plpgsql(a.b, b.b)
 FROM collate_test1 a, collate_test1 b
 ORDER BY a.b, b.b;
 
 SELECT a.b AS a, b.b AS b, a.b < b.b COLLATE "C" AS lt,
-       mylt(a.b, b.b COLLATE "C"), mylt_noninline(a.b, b.b COLLATE "C")
+       mylt(a.b, b.b COLLATE "C"), mylt_noninline(a.b, b.b COLLATE "C"),
+       mylt_plpgsql(a.b, b.b COLLATE "C")
 FROM collate_test1 a, collate_test1 b
 ORDER BY a.b, b.b;