Fix plpgsql to re-look-up composite type names at need.

Commit 4b93f579 rearranged things in plpgsql to make it cope better with
composite types changing underneath it intra-session.  However, I failed to
consider the case of a composite type being dropped and recreated entirely.
In my defense, the previous coding didn't consider that possibility at all
either --- but it would accidentally work so long as you didn't change the
type's field list, because the built-at-compile-time list of component
variables would then still match the type's new definition.  The new
coding, however, occasionally tries to re-look-up the type by OID, and
then fails to find the dropped type.

To fix this, we need to save the TypeName struct, and then redo the type
OID lookup from that.  Of course that's expensive, so we don't want to do
it every time we need the type OID.  This can be fixed in the same way that
4b93f579 dealt with changes to composite types' definitions: keep an eye
on the type's typcache entry to see if its tupledesc has been invalidated.
(Perhaps, at some point, this mechanism should be generalized so it can
work for non-composite types too; but for now, plpgsql only tries to
cope with intra-session redefinitions of composites.)

I'm slightly hesitant to back-patch this into v11, because it changes
the contents of struct PLpgSQL_type as well as the signature of
plpgsql_build_datatype(), so in principle it could break code that is
poking into the innards of plpgsql.  However, the only popular extension
of that ilk is pldebugger, and it doesn't seem to be affected.  Since
this is a regression for people who were relying on the old behavior,
it seems worth taking the small risk of causing compatibility issues.

Per bug #15913 from Daniel Fiori.  Back-patch to v11 where 4b93f579
came in.

Discussion: https://postgr.es/m/15913-a7e112e16dedcffc@postgresql.org

src/backend/utils/cache/typcache.c

@@ -2116,6 +2116,16 @@ TypeCacheRelCallback(Datum arg, Oid relid)
 				if (--typentry->tupDesc->tdrefcount == 0)
 					FreeTupleDesc(typentry->tupDesc);
 				typentry->tupDesc = NULL;
+
+				/*
+				 * Also clear tupDesc_identifier, so that anything watching
+				 * that will realize that the tupdesc has possibly changed.
+				 * (Alternatively, we could specify that to detect possible
+				 * tupdesc change, one must check for tupDesc != NULL as well
+				 * as tupDesc_identifier being the same as what was previously
+				 * seen.  That seems error-prone.)
+				 */
+				typentry->tupDesc_identifier = 0;
 			}
 
 			/* Reset equality/comparison/hashing validity information */

src/pl/plpgsql/src/expected/plpgsql_record.out

@@ -447,6 +447,35 @@ alter table mutable drop column f3;
 select getf3(null::mutable);  -- fails again
 ERROR:  record "x" has no field "f3"
 \set SHOW_CONTEXT errors
+-- check behavior with creating/dropping a named rowtype
+set check_function_bodies = off;  -- else reference to nonexistent type fails
+create function sillyaddtwo(int) returns int language plpgsql as
+$$ declare r mutable2; begin r.f1 := $1; return r.f1 + 2; end $$;
+reset check_function_bodies;
+select sillyaddtwo(42);  -- fail
+ERROR:  type "mutable2" does not exist
+LINE 1:  declare r mutable2; begin r.f1 := $1; return r.f1 + 2; end 
+                   ^
+QUERY:   declare r mutable2; begin r.f1 := $1; return r.f1 + 2; end 
+CONTEXT:  compilation of PL/pgSQL function "sillyaddtwo" near line 1
+create table mutable2(f1 int, f2 text);
+select sillyaddtwo(42);
+ sillyaddtwo 
+-------------
+          44
+(1 row)
+
+drop table mutable2;
+select sillyaddtwo(42);  -- fail
+ERROR:  type "mutable2" does not exist
+CONTEXT:  PL/pgSQL function sillyaddtwo(integer) line 1 at assignment
+create table mutable2(f0 text, f1 int, f2 text);
+select sillyaddtwo(42);
+ sillyaddtwo 
+-------------
+          44
+(1 row)
+
 -- check access to system columns in a record variable
 create function sillytrig() returns trigger language plpgsql as
 $$begin

src/pl/plpgsql/src/pl_exec.c

@@ -32,6 +32,7 @@
 #include "nodes/nodeFuncs.h"
 #include "optimizer/optimizer.h"
 #include "parser/parse_coerce.h"
+#include "parser/parse_type.h"
 #include "parser/scansup.h"
 #include "storage/proc.h"
 #include "tcop/tcopprot.h"
@@ -382,6 +383,7 @@ static void plpgsql_param_eval_generic_ro(ExprState *state, ExprEvalStep *op,
 static void exec_move_row(PLpgSQL_execstate *estate,
 						  PLpgSQL_variable *target,
 						  HeapTuple tup, TupleDesc tupdesc);
+static void revalidate_rectypeid(PLpgSQL_rec *rec);
 static ExpandedRecordHeader *make_expanded_record_for_rec(PLpgSQL_execstate *estate,
 														  PLpgSQL_rec *rec,
 														  TupleDesc srctupdesc,
@@ -2520,7 +2522,8 @@ exec_stmt_case(PLpgSQL_execstate *estate, PLpgSQL_stmt_case *stmt)
 			t_var->datatype->atttypmod != t_typmod)
 			t_var->datatype = plpgsql_build_datatype(t_typoid,
 													 t_typmod,
-													 estate->func->fn_input_collation);
+													 estate->func->fn_input_collation,
+													 NULL);
 
 		/* now we can assign to the variable */
 		exec_assign_value(estate,
@@ -6836,6 +6839,67 @@ exec_move_row(PLpgSQL_execstate *estate,
 	}
 }
 
+/*
+ * Verify that a PLpgSQL_rec's rectypeid is up-to-date.
+ */
+static void
+revalidate_rectypeid(PLpgSQL_rec *rec)
+{
+	PLpgSQL_type *typ = rec->datatype;
+	TypeCacheEntry *typentry;
+
+	if (rec->rectypeid == RECORDOID)
+		return;					/* it's RECORD, so nothing to do */
+	Assert(typ != NULL);
+	if (typ->tcache &&
+		typ->tcache->tupDesc_identifier == typ->tupdesc_id)
+		return;					/* known up-to-date */
+
+	/*
+	 * typcache entry has suffered invalidation, so re-look-up the type name
+	 * if possible, and then recheck the type OID.  If we don't have a
+	 * TypeName, then we just have to soldier on with the OID we've got.
+	 */
+	if (typ->origtypname != NULL)
+	{
+		/* this bit should match parse_datatype() in pl_gram.y */
+		typenameTypeIdAndMod(NULL, typ->origtypname,
+							 &typ->typoid,
+							 &typ->atttypmod);
+	}
+
+	/* this bit should match build_datatype() in pl_comp.c */
+	typentry = lookup_type_cache(typ->typoid,
+								 TYPECACHE_TUPDESC |
+								 TYPECACHE_DOMAIN_BASE_INFO);
+	if (typentry->typtype == TYPTYPE_DOMAIN)
+		typentry = lookup_type_cache(typentry->domainBaseType,
+									 TYPECACHE_TUPDESC);
+	if (typentry->tupDesc == NULL)
+	{
+		/*
+		 * If we get here, user tried to replace a composite type with a
+		 * non-composite one.  We're not gonna support that.
+		 */
+		ereport(ERROR,
+				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
+				 errmsg("type %s is not composite",
+						format_type_be(typ->typoid))));
+	}
+
+	/*
+	 * Update tcache and tupdesc_id.  Since we don't support changing to a
+	 * non-composite type, none of the rest of *typ needs to change.
+	 */
+	typ->tcache = typentry;
+	typ->tupdesc_id = typentry->tupDesc_identifier;
+
+	/*
+	 * Update *rec, too.  (We'll deal with subsidiary RECFIELDs as needed.)
+	 */
+	rec->rectypeid = typ->typoid;
+}
+
 /*
  * Build an expanded record object suitable for assignment to "rec".
  *
@@ -6860,6 +6924,11 @@ make_expanded_record_for_rec(PLpgSQL_execstate *estate,
 
 	if (rec->rectypeid != RECORDOID)
 	{
+		/*
+		 * Make sure rec->rectypeid is up-to-date before using it.
+		 */
+		revalidate_rectypeid(rec);
+
 		/*
 		 * New record must be of desired type, but maybe srcerh has already
 		 * done all the same lookups.
@@ -7331,6 +7400,11 @@ exec_move_row_from_datum(PLpgSQL_execstate *estate,
 			if (erh == rec->erh)
 				return;
 
+			/*
+			 * Make sure rec->rectypeid is up-to-date before using it.
+			 */
+			revalidate_rectypeid(rec);
+
 			/*
 			 * If we have a R/W pointer, we're allowed to just commandeer
 			 * ownership of the expanded record.  If it's of the right type to
@@ -7543,6 +7617,9 @@ instantiate_empty_record_variable(PLpgSQL_execstate *estate, PLpgSQL_rec *rec)
 				 errmsg("record \"%s\" is not assigned yet", rec->refname),
 				 errdetail("The tuple structure of a not-yet-assigned record is indeterminate.")));
 
+	/* Make sure rec->rectypeid is up-to-date before using it */
+	revalidate_rectypeid(rec);
+
 	/* OK, do it */
 	rec->erh = make_expanded_record_from_typeid(rec->rectypeid, -1,
 												estate->datum_context);

src/pl/plpgsql/src/pl_gram.y

@@ -551,7 +551,8 @@ decl_statement	: decl_varname decl_const decl_datatype decl_collate decl_notnull
 							plpgsql_build_variable($1.name, $1.lineno,
 												   plpgsql_build_datatype(REFCURSOROID,
 																		  -1,
-																		  InvalidOid),
+																		  InvalidOid,
+																		  NULL),
 												   true);
 
 						curname_def = palloc0(sizeof(PLpgSQL_expr));
@@ -1504,7 +1505,8 @@ for_control		: for_variable K_IN
 														   $1.lineno,
 														   plpgsql_build_datatype(INT4OID,
 																				  -1,
-																				  InvalidOid),
+																				  InvalidOid,
+																				  NULL),
 														   true);
 
 								new = palloc0(sizeof(PLpgSQL_stmt_fori));
@@ -2317,7 +2319,8 @@ exception_sect	:
 						var = plpgsql_build_variable("sqlstate", lineno,
 													 plpgsql_build_datatype(TEXTOID,
 																			-1,
-																			plpgsql_curr_compile->fn_input_collation),
+																			plpgsql_curr_compile->fn_input_collation,
+																			NULL),
 													 true);
 						var->isconst = true;
 						new->sqlstate_varno = var->dno;
@@ -2325,7 +2328,8 @@ exception_sect	:
 						var = plpgsql_build_variable("sqlerrm", lineno,
 													 plpgsql_build_datatype(TEXTOID,
 																			-1,
-																			plpgsql_curr_compile->fn_input_collation),
+																			plpgsql_curr_compile->fn_input_collation,
+																			NULL),
 													 true);
 						var->isconst = true;
 						new->sqlerrm_varno = var->dno;
@@ -3707,6 +3711,7 @@ plpgsql_sql_error_callback(void *arg)
 static PLpgSQL_type *
 parse_datatype(const char *string, int location)
 {
+	TypeName   *typeName;
 	Oid			type_id;
 	int32		typmod;
 	sql_error_callback_arg cbarg;
@@ -3721,14 +3726,16 @@ parse_datatype(const char *string, int location)
 	error_context_stack = &syntax_errcontext;
 
 	/* Let the main parser try to parse it under standard SQL rules */
-	parseTypeString(string, &type_id, &typmod, false);
+	typeName = typeStringToTypeName(string);
+	typenameTypeIdAndMod(NULL, typeName, &type_id, &typmod);
 
 	/* Restore former ereport callback */
 	error_context_stack = syntax_errcontext.previous;
 
 	/* Okay, build a PLpgSQL_type data structure for it */
 	return plpgsql_build_datatype(type_id, typmod,
-								  plpgsql_curr_compile->fn_input_collation);
+								  plpgsql_curr_compile->fn_input_collation,
+								  typeName);
 }
 
 /*
@@ -4080,7 +4087,8 @@ make_case(int location, PLpgSQL_expr *t_expr,
 			plpgsql_build_variable(varname, new->lineno,
 								   plpgsql_build_datatype(INT4OID,
 														  -1,
-														  InvalidOid),
+														  InvalidOid,
+														  NULL),
 								   true);
 		new->t_varno = t_var->dno;
 

src/pl/plpgsql/src/plpgsql.h

@@ -21,6 +21,7 @@
 #include "commands/trigger.h"
 #include "executor/spi.h"
 #include "utils/expandedrecord.h"
+#include "utils/typcache.h"
 
 
 /**********************************************************************
@@ -206,6 +207,10 @@ typedef struct PLpgSQL_type
 	Oid			collation;		/* from pg_type, but can be overridden */
 	bool		typisarray;		/* is "true" array, or domain over one */
 	int32		atttypmod;		/* typmod (taken from someplace else) */
+	/* Remaining fields are used only for named composite types (not RECORD) */
+	TypeName   *origtypname;	/* type name as written by user */
+	TypeCacheEntry *tcache;		/* typcache entry for composite type */
+	uint64		tupdesc_id;		/* last-seen tupdesc identifier */
 } PLpgSQL_type;
 
 /*
@@ -372,6 +377,12 @@ typedef struct PLpgSQL_rec
 	PLpgSQL_expr *default_val;
 	/* end of PLpgSQL_variable fields */
 
+	/*
+	 * Note: for non-RECORD cases, we may from time to time re-look-up the
+	 * composite type, using datatype->origtypname.  That can result in
+	 * changing rectypeid.
+	 */
+
 	PLpgSQL_type *datatype;		/* can be NULL, if rectypeid is RECORDOID */
 	Oid			rectypeid;		/* declared type of variable */
 	/* RECFIELDs for this record are chained together for easy access */
@@ -1227,7 +1238,8 @@ 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,
-											Oid collation);
+											Oid collation,
+											TypeName *origtypname);
 extern PLpgSQL_variable *plpgsql_build_variable(const char *refname, int lineno,
 												PLpgSQL_type *dtype,
 												bool add2namespace);

src/pl/plpgsql/src/sql/plpgsql_record.sql

@@ -288,6 +288,22 @@ alter table mutable drop column f3;
 select getf3(null::mutable);  -- fails again
 \set SHOW_CONTEXT errors
 
+-- check behavior with creating/dropping a named rowtype
+set check_function_bodies = off;  -- else reference to nonexistent type fails
+
+create function sillyaddtwo(int) returns int language plpgsql as
+$$ declare r mutable2; begin r.f1 := $1; return r.f1 + 2; end $$;
+
+reset check_function_bodies;
+
+select sillyaddtwo(42);  -- fail
+create table mutable2(f1 int, f2 text);
+select sillyaddtwo(42);
+drop table mutable2;
+select sillyaddtwo(42);  -- fail
+create table mutable2(f0 text, f1 int, f2 text);
+select sillyaddtwo(42);
+
 -- check access to system columns in a record variable
 
 create function sillytrig() returns trigger language plpgsql as