Make plpgsql use its DTYPE_REC code paths for composite-type variables.

Formerly, DTYPE_REC was used only for variables declared as "record";
variables of named composite types used DTYPE_ROW, which is faster for
some purposes but much less flexible.  In particular, the ROW code paths
are entirely incapable of dealing with DDL-caused changes to the number
or data types of the columns of a row variable, once a particular plpgsql
function has been parsed for the first time in a session.  And, since the
stored representation of a ROW isn't a tuple, there wasn't any easy way
to deal with variables of domain-over-composite types, since the domain
constraint checking code would expect the value to be checked to be a
tuple.  A lesser, but still real, annoyance is that ROW format cannot
represent a true NULL composite value, only a row of per-field NULL
values, which is not exactly the same thing.

Hence, switch to using DTYPE_REC for all composite-typed variables,
whether "record", named composite type, or domain over named composite
type.  DTYPE_ROW remains but is used only for its native purpose, to
represent a fixed-at-compile-time list of variables, for instance the
targets of an INTO clause.

To accomplish this without taking significant performance losses, introduce
infrastructure that allows storing composite-type variables as "expanded
objects", similar to the "expanded array" infrastructure introduced in
commit 1dc5ebc9.  A composite variable's value is thereby kept (most of
the time) in the form of separate Datums, so that field accesses and
updates are not much more expensive than they were in the ROW format.
This holds the line, more or less, on performance of variables of named
composite types in field-access-intensive microbenchmarks, and makes
variables declared "record" perform much better than before in similar
tests.  In addition, the logic involved with enforcing composite-domain
constraints against updates of individual fields is in the expanded
record infrastructure not plpgsql proper, so that it might be reusable
for other purposes.

In further support of this, introduce a typcache feature for assigning a
unique-within-process identifier to each distinct tuple descriptor of
interest; in particular, DDL alterations on composite types result in a new
identifier for that type.  This allows very cheap detection of the need to
refresh tupdesc-dependent data.  This improves on the "tupDescSeqNo" idea
I had in commit 687f096e: that assigned identifying sequence numbers to
successive versions of individual composite types, but the numbers were not
unique across different types, nor was there support for assigning numbers
to registered record types.

In passing, allow plpgsql functions to accept as well as return type
"record".  There was no good reason for the old restriction, and it
was out of step with most of the other PLs.

Tom Lane, reviewed by Pavel Stehule

Discussion: https://postgr.es/m/8962.1514399547@sss.pgh.pa.us

doc/src/sgml/plpgsql.sgml

@@ -123,7 +123,9 @@
      and they can return a result of any of these types.  They can also
      accept or return any composite type (row type) specified by name.
      It is also possible to declare a <application>PL/pgSQL</application>
-     function as returning <type>record</type>, which means that the result
+     function as accepting <type>record</type>, which means that any
+     composite type will do as input, or
+     as returning <type>record</type>, which means that the result
      is a row type whose columns are determined by specification in the
      calling query, as discussed in <xref linkend="queries-tablefunctions"/>.
     </para>
@@ -671,14 +673,6 @@ user_id users.user_id%TYPE;
     be selected from it, for example <literal>$1.user_id</literal>.
    </para>
 
-   <para>
-    Only the user-defined columns of a table row are accessible in a
-    row-type variable, not the OID or other system columns (because the
-    row could be from a view).  The fields of the row type inherit the
-    table's field size or precision for data types such as
-    <type>char(<replaceable>n</replaceable>)</type>.
-   </para>
-
    <para>
     Here is an example of using composite types.  <structname>table1</structname>
     and <structname>table2</structname> are existing tables having at least the

src/backend/executor/execExprInterp.c

@@ -70,6 +70,7 @@
 #include "utils/builtins.h"
 #include "utils/date.h"
 #include "utils/datum.h"
+#include "utils/expandedrecord.h"
 #include "utils/lsyscache.h"
 #include "utils/timestamp.h"
 #include "utils/typcache.h"
@@ -2820,57 +2821,105 @@ ExecEvalFieldSelect(ExprState *state, ExprEvalStep *op, ExprContext *econtext)
 	if (*op->resnull)
 		return;
 
-	/* Get the composite datum and extract its type fields */
 	tupDatum = *op->resvalue;
-	tuple = DatumGetHeapTupleHeader(tupDatum);
 
-	tupType = HeapTupleHeaderGetTypeId(tuple);
-	tupTypmod = HeapTupleHeaderGetTypMod(tuple);
+	/* We can special-case expanded records for speed */
+	if (VARATT_IS_EXTERNAL_EXPANDED(DatumGetPointer(tupDatum)))
+	{
+		ExpandedRecordHeader *erh = (ExpandedRecordHeader *) DatumGetEOHP(tupDatum);
 
-	/* Lookup tupdesc if first time through or if type changes */
-	tupDesc = get_cached_rowtype(tupType, tupTypmod,
-								 &op->d.fieldselect.argdesc,
-								 econtext);
+		Assert(erh->er_magic == ER_MAGIC);
 
-	/*
-	 * Find field's attr record.  Note we don't support system columns here: a
-	 * datum tuple doesn't have valid values for most of the interesting
-	 * system columns anyway.
-	 */
-	if (fieldnum <= 0)			/* should never happen */
-		elog(ERROR, "unsupported reference to system column %d in FieldSelect",
-			 fieldnum);
-	if (fieldnum > tupDesc->natts)	/* should never happen */
-		elog(ERROR, "attribute number %d exceeds number of columns %d",
-			 fieldnum, tupDesc->natts);
-	attr = TupleDescAttr(tupDesc, fieldnum - 1);
-
-	/* Check for dropped column, and force a NULL result if so */
-	if (attr->attisdropped)
-	{
-		*op->resnull = true;
-		return;
+		/* Extract record's TupleDesc */
+		tupDesc = expanded_record_get_tupdesc(erh);
+
+		/*
+		 * Find field's attr record.  Note we don't support system columns
+		 * here: a datum tuple doesn't have valid values for most of the
+		 * interesting system columns anyway.
+		 */
+		if (fieldnum <= 0)		/* should never happen */
+			elog(ERROR, "unsupported reference to system column %d in FieldSelect",
+				 fieldnum);
+		if (fieldnum > tupDesc->natts)	/* should never happen */
+			elog(ERROR, "attribute number %d exceeds number of columns %d",
+				 fieldnum, tupDesc->natts);
+		attr = TupleDescAttr(tupDesc, fieldnum - 1);
+
+		/* Check for dropped column, and force a NULL result if so */
+		if (attr->attisdropped)
+		{
+			*op->resnull = true;
+			return;
+		}
+
+		/* Check for type mismatch --- possible after ALTER COLUMN TYPE? */
+		/* As in CheckVarSlotCompatibility, we should but can't check typmod */
+		if (op->d.fieldselect.resulttype != attr->atttypid)
+			ereport(ERROR,
+					(errcode(ERRCODE_DATATYPE_MISMATCH),
+					 errmsg("attribute %d has wrong type", fieldnum),
+					 errdetail("Table has type %s, but query expects %s.",
+							   format_type_be(attr->atttypid),
+							   format_type_be(op->d.fieldselect.resulttype))));
+
+		/* extract the field */
+		*op->resvalue = expanded_record_get_field(erh, fieldnum,
+												  op->resnull);
 	}
+	else
+	{
+		/* Get the composite datum and extract its type fields */
+		tuple = DatumGetHeapTupleHeader(tupDatum);
 
-	/* Check for type mismatch --- possible after ALTER COLUMN TYPE? */
-	/* As in CheckVarSlotCompatibility, we should but can't check typmod */
-	if (op->d.fieldselect.resulttype != attr->atttypid)
-		ereport(ERROR,
-				(errcode(ERRCODE_DATATYPE_MISMATCH),
-				 errmsg("attribute %d has wrong type", fieldnum),
-				 errdetail("Table has type %s, but query expects %s.",
-						   format_type_be(attr->atttypid),
-						   format_type_be(op->d.fieldselect.resulttype))));
+		tupType = HeapTupleHeaderGetTypeId(tuple);
+		tupTypmod = HeapTupleHeaderGetTypMod(tuple);
 
-	/* heap_getattr needs a HeapTuple not a bare HeapTupleHeader */
-	tmptup.t_len = HeapTupleHeaderGetDatumLength(tuple);
-	tmptup.t_data = tuple;
+		/* Lookup tupdesc if first time through or if type changes */
+		tupDesc = get_cached_rowtype(tupType, tupTypmod,
+									 &op->d.fieldselect.argdesc,
+									 econtext);
+
+		/*
+		 * Find field's attr record.  Note we don't support system columns
+		 * here: a datum tuple doesn't have valid values for most of the
+		 * interesting system columns anyway.
+		 */
+		if (fieldnum <= 0)		/* should never happen */
+			elog(ERROR, "unsupported reference to system column %d in FieldSelect",
+				 fieldnum);
+		if (fieldnum > tupDesc->natts)	/* should never happen */
+			elog(ERROR, "attribute number %d exceeds number of columns %d",
+				 fieldnum, tupDesc->natts);
+		attr = TupleDescAttr(tupDesc, fieldnum - 1);
 
-	/* extract the field */
-	*op->resvalue = heap_getattr(&tmptup,
-								 fieldnum,
-								 tupDesc,
-								 op->resnull);
+		/* Check for dropped column, and force a NULL result if so */
+		if (attr->attisdropped)
+		{
+			*op->resnull = true;
+			return;
+		}
+
+		/* Check for type mismatch --- possible after ALTER COLUMN TYPE? */
+		/* As in CheckVarSlotCompatibility, we should but can't check typmod */
+		if (op->d.fieldselect.resulttype != attr->atttypid)
+			ereport(ERROR,
+					(errcode(ERRCODE_DATATYPE_MISMATCH),
+					 errmsg("attribute %d has wrong type", fieldnum),
+					 errdetail("Table has type %s, but query expects %s.",
+							   format_type_be(attr->atttypid),
+							   format_type_be(op->d.fieldselect.resulttype))));
+
+		/* heap_getattr needs a HeapTuple not a bare HeapTupleHeader */
+		tmptup.t_len = HeapTupleHeaderGetDatumLength(tuple);
+		tmptup.t_data = tuple;
+
+		/* extract the field */
+		*op->resvalue = heap_getattr(&tmptup,
+									 fieldnum,
+									 tupDesc,
+									 op->resnull);
+	}
 }
 
 /*

src/backend/utils/adt/Makefile

@@ -12,7 +12,7 @@ include $(top_builddir)/src/Makefile.global
 OBJS = acl.o amutils.o arrayfuncs.o array_expanded.o array_selfuncs.o \
 	array_typanalyze.o array_userfuncs.o arrayutils.o ascii.o \
 	bool.o cash.o char.o date.o datetime.o datum.o dbsize.o domains.o \
-	encode.o enum.o expandeddatum.o \
+	encode.o enum.o expandeddatum.o expandedrecord.o \
 	float.o format_type.o formatting.o genfile.o \
 	geo_ops.o geo_selfuncs.o geo_spgist.o inet_cidr_ntop.o inet_net_pton.o \
 	int.o int8.o json.o jsonb.o jsonb_gin.o jsonb_op.o jsonb_util.o \

src/backend/utils/cache/typcache.c

@@ -259,12 +259,22 @@ static const dshash_parameters srtr_typmod_table_params = {
 	LWTRANCHE_SESSION_TYPMOD_TABLE
 };
 
+/* hashtable for recognizing registered record types */
 static HTAB *RecordCacheHash = NULL;
 
+/* arrays of info about registered record types, indexed by assigned typmod */
 static TupleDesc *RecordCacheArray = NULL;
-static int32 RecordCacheArrayLen = 0;	/* allocated length of array */
+static uint64 *RecordIdentifierArray = NULL;
+static int32 RecordCacheArrayLen = 0;	/* allocated length of above arrays */
 static int32 NextRecordTypmod = 0;	/* number of entries used */
 
+/*
+ * Process-wide counter for generating unique tupledesc identifiers.
+ * Zero and one (INVALID_TUPLEDESC_IDENTIFIER) aren't allowed to be chosen
+ * as identifiers, so we start the counter at INVALID_TUPLEDESC_IDENTIFIER.
+ */
+static uint64 tupledesc_id_counter = INVALID_TUPLEDESC_IDENTIFIER;
+
 static void load_typcache_tupdesc(TypeCacheEntry *typentry);
 static void load_rangetype_info(TypeCacheEntry *typentry);
 static void load_domaintype_info(TypeCacheEntry *typentry);
@@ -793,10 +803,10 @@ load_typcache_tupdesc(TypeCacheEntry *typentry)
 	typentry->tupDesc->tdrefcount++;
 
 	/*
-	 * In future, we could take some pains to not increment the seqno if the
-	 * tupdesc didn't really change; but for now it's not worth it.
+	 * In future, we could take some pains to not change tupDesc_identifier if
+	 * the tupdesc didn't really change; but for now it's not worth it.
 	 */
-	typentry->tupDescSeqNo++;
+	typentry->tupDesc_identifier = ++tupledesc_id_counter;
 
 	relation_close(rel, AccessShareLock);
 }
@@ -1496,7 +1506,8 @@ cache_range_element_properties(TypeCacheEntry *typentry)
 }
 
 /*
- * Make sure that RecordCacheArray is large enough to store 'typmod'.
+ * Make sure that RecordCacheArray and RecordIdentifierArray are large enough
+ * to store 'typmod'.
  */
 static void
 ensure_record_cache_typmod_slot_exists(int32 typmod)
@@ -1505,6 +1516,8 @@ ensure_record_cache_typmod_slot_exists(int32 typmod)
 	{
 		RecordCacheArray = (TupleDesc *)
 			MemoryContextAllocZero(CacheMemoryContext, 64 * sizeof(TupleDesc));
+		RecordIdentifierArray = (uint64 *)
+			MemoryContextAllocZero(CacheMemoryContext, 64 * sizeof(uint64));
 		RecordCacheArrayLen = 64;
 	}
 
@@ -1519,6 +1532,10 @@ ensure_record_cache_typmod_slot_exists(int32 typmod)
 												  newlen * sizeof(TupleDesc));
 		memset(RecordCacheArray + RecordCacheArrayLen, 0,
 			   (newlen - RecordCacheArrayLen) * sizeof(TupleDesc));
+		RecordIdentifierArray = (uint64 *) repalloc(RecordIdentifierArray,
+													newlen * sizeof(uint64));
+		memset(RecordIdentifierArray + RecordCacheArrayLen, 0,
+			   (newlen - RecordCacheArrayLen) * sizeof(uint64));
 		RecordCacheArrayLen = newlen;
 	}
 }
@@ -1581,11 +1598,17 @@ lookup_rowtype_tupdesc_internal(Oid type_id, int32 typmod, bool noError)
 
 					/*
 					 * Our local array can now point directly to the TupleDesc
-					 * in shared memory.
+					 * in shared memory, which is non-reference-counted.
 					 */
 					RecordCacheArray[typmod] = tupdesc;
 					Assert(tupdesc->tdrefcount == -1);
 
+					/*
+					 * We don't share tupdesc identifiers across processes, so
+					 * assign one locally.
+					 */
+					RecordIdentifierArray[typmod] = ++tupledesc_id_counter;
+
 					dshash_release_lock(CurrentSession->shared_typmod_table,
 										entry);
 
@@ -1790,12 +1813,61 @@ assign_record_type_typmod(TupleDesc tupDesc)
 	RecordCacheArray[entDesc->tdtypmod] = entDesc;
 	recentry->tupdesc = entDesc;
 
+	/* Assign a unique tupdesc identifier, too. */
+	RecordIdentifierArray[entDesc->tdtypmod] = ++tupledesc_id_counter;
+
 	/* Update the caller's tuple descriptor. */
 	tupDesc->tdtypmod = entDesc->tdtypmod;
 
 	MemoryContextSwitchTo(oldcxt);
 }
 
+/*
+ * assign_record_type_identifier
+ *
+ * Get an identifier, which will be unique over the lifespan of this backend
+ * process, for the current tuple descriptor of the specified composite type.
+ * For named composite types, the value is guaranteed to change if the type's
+ * definition does.  For registered RECORD types, the value will not change
+ * once assigned, since the registered type won't either.  If an anonymous
+ * RECORD type is specified, we return a new identifier on each call.
+ */
+uint64
+assign_record_type_identifier(Oid type_id, int32 typmod)
+{
+	if (type_id != RECORDOID)
+	{
+		/*
+		 * It's a named composite type, so use the regular typcache.
+		 */
+		TypeCacheEntry *typentry;
+
+		typentry = lookup_type_cache(type_id, TYPECACHE_TUPDESC);
+		if (typentry->tupDesc == NULL)
+			ereport(ERROR,
+					(errcode(ERRCODE_WRONG_OBJECT_TYPE),
+					 errmsg("type %s is not composite",
+							format_type_be(type_id))));
+		Assert(typentry->tupDesc_identifier != 0);
+		return typentry->tupDesc_identifier;
+	}
+	else
+	{
+		/*
+		 * It's a transient record type, so look in our record-type table.
+		 */
+		if (typmod >= 0 && typmod < RecordCacheArrayLen &&
+			RecordCacheArray[typmod] != NULL)
+		{
+			Assert(RecordIdentifierArray[typmod] != 0);
+			return RecordIdentifierArray[typmod];
+		}
+
+		/* For anonymous or unrecognized record type, generate a new ID */
+		return ++tupledesc_id_counter;
+	}
+}
+
 /*
  * Return the amout of shmem required to hold a SharedRecordTypmodRegistry.
  * This exists only to avoid exposing private innards of

src/include/utils/expandedrecord.h

+/*-------------------------------------------------------------------------
+ *
+ * expandedrecord.h
+ *	  Declarations for composite expanded objects.
+ *
+ * Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/utils/expandedrecord.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef EXPANDEDRECORD_H
+#define EXPANDEDRECORD_H
+
+#include "access/htup.h"
+#include "access/tupdesc.h"
+#include "fmgr.h"
+#include "utils/expandeddatum.h"
+
+
+/*
+ * An expanded record is contained within a private memory context (as
+ * all expanded objects must be) and has a control structure as below.
+ *
+ * The expanded record might contain a regular "flat" tuple if that was the
+ * original input and we've not modified it.  Otherwise, the contents are
+ * represented by Datum/isnull arrays plus type information.  We could also
+ * have both forms, if we've deconstructed the original tuple for access
+ * purposes but not yet changed it.  For pass-by-reference field types, the
+ * Datums would point into the flat tuple in this situation.  Once we start
+ * modifying tuple fields, new pass-by-ref fields are separately palloc'd
+ * within the memory context.
+ *
+ * It's possible to build an expanded record that references a "flat" tuple
+ * stored externally, if the caller can guarantee that that tuple will not
+ * change for the lifetime of the expanded record.  (This frammish is mainly
+ * meant to avoid unnecessary data copying in trigger functions.)
+ */
+#define ER_MAGIC 1384727874		/* ID for debugging crosschecks */
+
+typedef struct ExpandedRecordHeader
+{
+	/* Standard header for expanded objects */
+	ExpandedObjectHeader hdr;
+
+	/* Magic value identifying an expanded record (for debugging only) */
+	int			er_magic;
+
+	/* Assorted flag bits */
+	int			flags;
+#define ER_FLAG_FVALUE_VALID	0x0001	/* fvalue is up to date? */
+#define ER_FLAG_FVALUE_ALLOCED	0x0002	/* fvalue is local storage? */
+#define ER_FLAG_DVALUES_VALID	0x0004	/* dvalues/dnulls are up to date? */
+#define ER_FLAG_DVALUES_ALLOCED	0x0008	/* any field values local storage? */
+#define ER_FLAG_HAVE_EXTERNAL	0x0010	/* any field values are external? */
+#define ER_FLAG_TUPDESC_ALLOCED	0x0020	/* tupdesc is local storage? */
+#define ER_FLAG_IS_DOMAIN		0x0040	/* er_decltypeid is domain? */
+#define ER_FLAG_IS_DUMMY		0x0080	/* this header is dummy (see below) */
+/* flag bits that are not to be cleared when replacing tuple data: */
+#define ER_FLAGS_NON_DATA \
+	(ER_FLAG_TUPDESC_ALLOCED | ER_FLAG_IS_DOMAIN | ER_FLAG_IS_DUMMY)
+
+	/* Declared type of the record variable (could be a domain type) */
+	Oid			er_decltypeid;
+
+	/*
+	 * Actual composite type/typmod; never a domain (if ER_FLAG_IS_DOMAIN,
+	 * these identify the composite base type).  These will match
+	 * er_tupdesc->tdtypeid/tdtypmod, as well as the header fields of
+	 * composite datums made from or stored in this expanded record.
+	 */
+	Oid			er_typeid;		/* type OID of the composite type */
+	int32		er_typmod;		/* typmod of the composite type */
+
+	/*
+	 * Tuple descriptor, if we have one, else NULL.  This may point to a
+	 * reference-counted tupdesc originally belonging to the typcache, in
+	 * which case we use a memory context reset callback to release the
+	 * refcount.  It can also be locally allocated in this object's private
+	 * context (in which case ER_FLAG_TUPDESC_ALLOCED is set).
+	 */
+	TupleDesc	er_tupdesc;
+
+	/*
+	 * Unique-within-process identifier for the tupdesc (see typcache.h). This
+	 * field will never be equal to INVALID_TUPLEDESC_IDENTIFIER.
+	 */
+	uint64		er_tupdesc_id;
+
+	/*
+	 * If we have a Datum-array representation of the record, it's kept here;
+	 * else ER_FLAG_DVALUES_VALID is not set, and dvalues/dnulls may be NULL
+	 * if they've not yet been allocated.  If allocated, the dvalues and
+	 * dnulls arrays are palloc'd within the object private context, and are
+	 * of length matching er_tupdesc->natts.  For pass-by-ref field types,
+	 * dvalues entries might point either into the fstartptr..fendptr area, or
+	 * to separately palloc'd chunks.
+	 */
+	Datum	   *dvalues;		/* array of Datums */
+	bool	   *dnulls;			/* array of is-null flags for Datums */
+	int			nfields;		/* length of above arrays */
+
+	/*
+	 * flat_size is the current space requirement for the flat equivalent of
+	 * the expanded record, if known; otherwise it's 0.  We store this to make
+	 * consecutive calls of get_flat_size cheap.  If flat_size is not 0, the
+	 * component values data_len, hoff, and hasnull must be valid too.
+	 */
+	Size		flat_size;
+
+	Size		data_len;		/* data len within flat_size */
+	int			hoff;			/* header offset */
+	bool		hasnull;		/* null bitmap needed? */
+
+	/*
+	 * fvalue points to the flat representation if we have one, else it is
+	 * NULL.  If the flat representation is valid (up to date) then
+	 * ER_FLAG_FVALUE_VALID is set.  Even if we've outdated the flat
+	 * representation due to changes of user fields, it can still be used to
+	 * fetch system column values.  If we have a flat representation then
+	 * fstartptr/fendptr point to the start and end+1 of its data area; this
+	 * is so that we can tell which Datum pointers point into the flat
+	 * representation rather than being pointers to separately palloc'd data.
+	 */
+	HeapTuple	fvalue;			/* might or might not be private storage */
+	char	   *fstartptr;		/* start of its data area */
+	char	   *fendptr;		/* end+1 of its data area */
+
+	/* Working state for domain checking, used if ER_FLAG_IS_DOMAIN is set */
+	MemoryContext er_domain_check_cxt;	/* short-term memory context */
+	struct ExpandedRecordHeader *er_dummy_header;	/* dummy record header */
+	void	   *er_domaininfo;	/* cache space for domain_check() */
+
+	/* Callback info (it's active if er_mcb.arg is not NULL) */
+	MemoryContextCallback er_mcb;
+} ExpandedRecordHeader;
+
+/* fmgr macros for expanded record objects */
+#define PG_GETARG_EXPANDED_RECORD(n)  DatumGetExpandedRecord(PG_GETARG_DATUM(n))
+#define ExpandedRecordGetDatum(erh)   EOHPGetRWDatum(&(erh)->hdr)
+#define ExpandedRecordGetRODatum(erh) EOHPGetRODatum(&(erh)->hdr)
+#define PG_RETURN_EXPANDED_RECORD(x)  PG_RETURN_DATUM(ExpandedRecordGetDatum(x))
+
+/* assorted other macros */
+#define ExpandedRecordIsEmpty(erh) \
+	(((erh)->flags & (ER_FLAG_DVALUES_VALID | ER_FLAG_FVALUE_VALID)) == 0)
+#define ExpandedRecordIsDomain(erh) \
+	(((erh)->flags & ER_FLAG_IS_DOMAIN) != 0)
+
+/* this can substitute for TransferExpandedObject() when we already have erh */
+#define TransferExpandedRecord(erh, cxt) \
+	MemoryContextSetParent((erh)->hdr.eoh_context, cxt)
+
+/* information returned by expanded_record_lookup_field() */
+typedef struct ExpandedRecordFieldInfo
+{
+	int			fnumber;		/* field's attr number in record */
+	Oid			ftypeid;		/* field's type/typmod info */
+	int32		ftypmod;
+	Oid			fcollation;		/* field's collation if any */
+} ExpandedRecordFieldInfo;
+
+/*
+ * prototypes for functions defined in expandedrecord.c
+ */
+extern ExpandedRecordHeader *make_expanded_record_from_typeid(Oid type_id, int32 typmod,
+								 MemoryContext parentcontext);
+extern ExpandedRecordHeader *make_expanded_record_from_tupdesc(TupleDesc tupdesc,
+								  MemoryContext parentcontext);
+extern ExpandedRecordHeader *make_expanded_record_from_exprecord(ExpandedRecordHeader *olderh,
+									MemoryContext parentcontext);
+extern void expanded_record_set_tuple(ExpandedRecordHeader *erh,
+						  HeapTuple tuple, bool copy);
+extern Datum make_expanded_record_from_datum(Datum recorddatum,
+								MemoryContext parentcontext);
+extern TupleDesc expanded_record_fetch_tupdesc(ExpandedRecordHeader *erh);
+extern HeapTuple expanded_record_get_tuple(ExpandedRecordHeader *erh);
+extern ExpandedRecordHeader *DatumGetExpandedRecord(Datum d);
+extern void deconstruct_expanded_record(ExpandedRecordHeader *erh);
+extern bool expanded_record_lookup_field(ExpandedRecordHeader *erh,
+							 const char *fieldname,
+							 ExpandedRecordFieldInfo *finfo);
+extern Datum expanded_record_fetch_field(ExpandedRecordHeader *erh, int fnumber,
+							bool *isnull);
+extern void expanded_record_set_field_internal(ExpandedRecordHeader *erh,
+								   int fnumber,
+								   Datum newValue, bool isnull,
+								   bool check_constraints);
+extern void expanded_record_set_fields(ExpandedRecordHeader *erh,
+						   const Datum *newValues, const bool *isnulls);
+
+/* outside code should never call expanded_record_set_field_internal as such */
+#define expanded_record_set_field(erh, fnumber, newValue, isnull) \
+	expanded_record_set_field_internal(erh, fnumber, newValue, isnull, true)
+
+/*
+ * Inline-able fast cases.  The expanded_record_fetch_xxx functions above
+ * handle the general cases.
+ */
+
+/* Get the tupdesc for the expanded record's actual type */
+static inline TupleDesc
+expanded_record_get_tupdesc(ExpandedRecordHeader *erh)
+{
+	if (likely(erh->er_tupdesc != NULL))
+		return erh->er_tupdesc;
+	else
+		return expanded_record_fetch_tupdesc(erh);
+}
+
+/* Get value of record field */
+static inline Datum
+expanded_record_get_field(ExpandedRecordHeader *erh, int fnumber,
+						  bool *isnull)
+{
+	if ((erh->flags & ER_FLAG_DVALUES_VALID) &&
+		likely(fnumber > 0 && fnumber <= erh->nfields))
+	{
+		*isnull = erh->dnulls[fnumber - 1];
+		return erh->dvalues[fnumber - 1];
+	}
+	else
+		return expanded_record_fetch_field(erh, fnumber, isnull);
+}
+
+#endif							/* EXPANDEDRECORD_H */

src/include/utils/typcache.h

@@ -76,11 +76,14 @@ typedef struct TypeCacheEntry
 	/*
 	 * Tuple descriptor if it's a composite type (row type).  NULL if not
 	 * composite or information hasn't yet been requested.  (NOTE: this is a
-	 * reference-counted tupledesc.)  To simplify caching dependent info,
-	 * tupDescSeqNo is incremented each time tupDesc is rebuilt in a session.
+	 * reference-counted tupledesc.)
+	 *
+	 * To simplify caching dependent info, tupDesc_identifier is an identifier
+	 * for this tupledesc that is unique for the life of the process, and
+	 * changes anytime the tupledesc does.  Zero if not yet determined.
 	 */
 	TupleDesc	tupDesc;
-	int64		tupDescSeqNo;
+	uint64		tupDesc_identifier;
 
 	/*
 	 * Fields computed when TYPECACHE_RANGE_INFO is requested.  Zeroes if not
@@ -138,6 +141,9 @@ typedef struct TypeCacheEntry
 #define TYPECACHE_HASH_EXTENDED_PROC		0x4000
 #define TYPECACHE_HASH_EXTENDED_PROC_FINFO	0x8000
 
+/* This value will not equal any valid tupledesc identifier, nor 0 */
+#define INVALID_TUPLEDESC_IDENTIFIER ((uint64) 1)
+
 /*
  * Callers wishing to maintain a long-lived reference to a domain's constraint
  * set must store it in one of these.  Use InitDomainConstraintRef() and
@@ -179,6 +185,8 @@ extern TupleDesc lookup_rowtype_tupdesc_domain(Oid type_id, int32 typmod,
 
 extern void assign_record_type_typmod(TupleDesc tupDesc);
 
+extern uint64 assign_record_type_identifier(Oid type_id, int32 typmod);
+
 extern int	compare_values_of_enum(TypeCacheEntry *tcache, Oid arg1, Oid arg2);
 
 extern size_t SharedRecordTypmodRegistryEstimate(void);

src/pl/plpgsql/src/Makefile

@@ -26,7 +26,7 @@ DATA = plpgsql.control plpgsql--1.0.sql plpgsql--unpackaged--1.0.sql
 
 REGRESS_OPTS = --dbname=$(PL_TESTDB)
 
-REGRESS = plpgsql_call plpgsql_control plpgsql_transaction
+REGRESS = plpgsql_call plpgsql_control plpgsql_record plpgsql_transaction
 
 all: all-lib
 

src/pl/plpgsql/src/pl_funcs.c

@@ -1618,15 +1618,16 @@ plpgsql_dumptree(PLpgSQL_function *func)
 					printf("ROW %-16s fields", row->refname);
 					for (i = 0; i < row->nfields; i++)
 					{
-						if (row->fieldnames[i])
-							printf(" %s=var %d", row->fieldnames[i],
-								   row->varnos[i]);
+						printf(" %s=var %d", row->fieldnames[i],
+							   row->varnos[i]);
 					}
 					printf("\n");
 				}
 				break;
 			case PLPGSQL_DTYPE_REC:
-				printf("REC %s\n", ((PLpgSQL_rec *) d)->refname);
+				printf("REC %-16s typoid %u\n",
+					   ((PLpgSQL_rec *) d)->refname,
+					   ((PLpgSQL_rec *) d)->rectypeid);
 				break;
 			case PLPGSQL_DTYPE_RECFIELD:
 				printf("RECFIELD %-16s of REC %d\n",

src/pl/plpgsql/src/pl_gram.y

@@ -512,7 +512,7 @@ decl_statement	: decl_varname decl_const decl_datatype decl_collate decl_notnull
 							else
 								ereport(ERROR,
 										(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
-										 errmsg("row or record variable cannot be CONSTANT"),
+										 errmsg("record variable cannot be CONSTANT"),
 										 parser_errposition(@2)));
 						}
 						if ($5)
@@ -522,7 +522,7 @@ decl_statement	: decl_varname decl_const decl_datatype decl_collate decl_notnull
 							else
 								ereport(ERROR,
 										(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
-										 errmsg("row or record variable cannot be NOT NULL"),
+										 errmsg("record variable cannot be NOT NULL"),
 										 parser_errposition(@4)));
 
 						}
@@ -533,7 +533,7 @@ decl_statement	: decl_varname decl_const decl_datatype decl_collate decl_notnull
 							else
 								ereport(ERROR,
 										(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
-										 errmsg("default value for row or record variable is not supported"),
+										 errmsg("default value for record variable is not supported"),
 										 parser_errposition(@5)));
 						}
 					}
@@ -1333,7 +1333,7 @@ for_control		: for_variable K_IN
 							{
 								ereport(ERROR,
 										(errcode(ERRCODE_DATATYPE_MISMATCH),
-										 errmsg("loop variable of loop over rows must be a record or row variable or list of scalar variables"),
+										 errmsg("loop variable of loop over rows must be a record variable or list of scalar variables"),
 										 parser_errposition(@1)));
 							}
 							new->query = expr;
@@ -1386,6 +1386,7 @@ for_control		: for_variable K_IN
 							new->var = (PLpgSQL_variable *)
 								plpgsql_build_record($1.name,
 													 $1.lineno,
+													 RECORDOID,
 													 true);
 
 							$$ = (PLpgSQL_stmt *) new;
@@ -1524,7 +1525,7 @@ for_control		: for_variable K_IN
 								{
 									ereport(ERROR,
 											(errcode(ERRCODE_SYNTAX_ERROR),
-											 errmsg("loop variable of loop over rows must be a record or row variable or list of scalar variables"),
+											 errmsg("loop variable of loop over rows must be a record variable or list of scalar variables"),
 											 parser_errposition(@1)));
 								}
 
@@ -3328,7 +3329,7 @@ check_assignable(PLpgSQL_datum *datum, int location)
 						 parser_errposition(location)));
 			break;
 		case PLPGSQL_DTYPE_ROW:
-			/* always assignable? */
+			/* always assignable?  Shouldn't we check member vars? */
 			break;
 		case PLPGSQL_DTYPE_REC:
 			/* always assignable?  What about NEW/OLD? */
@@ -3385,7 +3386,7 @@ read_into_target(PLpgSQL_variable **target, bool *strict)
 				if ((tok = yylex()) == ',')
 					ereport(ERROR,
 							(errcode(ERRCODE_SYNTAX_ERROR),
-							 errmsg("record or row variable cannot be part of multiple-item INTO list"),
+							 errmsg("record variable cannot be part of multiple-item INTO list"),
 							 parser_errposition(yylloc)));
 				plpgsql_push_back_token(tok);
 			}

src/pl/plpgsql/src/pl_handler.c

@@ -443,14 +443,15 @@ plpgsql_validator(PG_FUNCTION_ARGS)
 	}
 
 	/* Disallow pseudotypes in arguments (either IN or OUT) */
-	/* except for polymorphic */
+	/* except for RECORD and polymorphic */
 	numargs = get_func_arg_info(tuple,
 								&argtypes, &argnames, &argmodes);
 	for (i = 0; i < numargs; i++)
 	{
 		if (get_typtype(argtypes[i]) == TYPTYPE_PSEUDO)
 		{
-			if (!IsPolymorphicType(argtypes[i]))
+			if (argtypes[i] != RECORDOID &&
+				!IsPolymorphicType(argtypes[i]))
 				ereport(ERROR,
 						(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 						 errmsg("PL/pgSQL functions cannot accept type %s",

src/pl/plpgsql/src/plpgsql.h

@@ -20,6 +20,8 @@
 #include "commands/event_trigger.h"
 #include "commands/trigger.h"
 #include "executor/spi.h"
+#include "utils/expandedrecord.h"
+
 
 /**********************************************************************
  * Definitions
@@ -37,10 +39,9 @@
  */
 typedef enum PLpgSQL_nsitem_type
 {
-	PLPGSQL_NSTYPE_LABEL,
-	PLPGSQL_NSTYPE_VAR,
-	PLPGSQL_NSTYPE_ROW,
-	PLPGSQL_NSTYPE_REC
+	PLPGSQL_NSTYPE_LABEL,		/* block label */
+	PLPGSQL_NSTYPE_VAR,			/* scalar variable */
+	PLPGSQL_NSTYPE_REC			/* composite variable */
 } PLpgSQL_nsitem_type;
 
 /*
@@ -72,9 +73,8 @@ typedef enum PLpgSQL_datum_type
 typedef enum PLpgSQL_type_type
 {
 	PLPGSQL_TTYPE_SCALAR,		/* scalar types and domains */
-	PLPGSQL_TTYPE_ROW,			/* composite types */
-	PLPGSQL_TTYPE_REC,			/* RECORD pseudotype */
-	PLPGSQL_TTYPE_PSEUDO		/* other pseudotypes */
+	PLPGSQL_TTYPE_REC,			/* composite types, including RECORD */
+	PLPGSQL_TTYPE_PSEUDO		/* pseudotypes */
 } PLpgSQL_type_type;
 
 /*
@@ -183,7 +183,6 @@ typedef struct PLpgSQL_type
 	int16		typlen;			/* stuff copied from its pg_type entry */
 	bool		typbyval;
 	char		typtype;
-	Oid			typrelid;
 	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) */
@@ -274,7 +273,12 @@ typedef struct PLpgSQL_var
 } PLpgSQL_var;
 
 /*
- * Row variable
+ * Row variable - this represents one or more variables that are listed in an
+ * INTO clause, FOR-loop targetlist, cursor argument list, etc.  We also use
+ * a row to represent a function's OUT parameters when there's more than one.
+ *
+ * Note that there's no way to name the row as such from PL/pgSQL code,
+ * so many functions don't need to support these.
  */
 typedef struct PLpgSQL_row
 {
@@ -283,21 +287,20 @@ typedef struct PLpgSQL_row
 	char	   *refname;
 	int			lineno;
 
-	/* Note: TupleDesc is only set up for named rowtypes, else it is NULL. */
-	TupleDesc	rowtupdesc;
-
 	/*
-	 * Note: if the underlying rowtype contains a dropped column, the
-	 * corresponding fieldnames[] entry will be NULL, and there is no
-	 * corresponding var (varnos[] will be -1).
+	 * rowtupdesc is only set up if we might need to convert the row into a
+	 * composite datum, which currently only happens for OUT parameters.
+	 * Otherwise it is NULL.
 	 */
+	TupleDesc	rowtupdesc;
+
 	int			nfields;
 	char	  **fieldnames;
 	int		   *varnos;
 } PLpgSQL_row;
 
 /*
- * Record variable (non-fixed structure)
+ * Record variable (any composite type, including RECORD)
  */
 typedef struct PLpgSQL_rec
 {
@@ -305,11 +308,11 @@ typedef struct PLpgSQL_rec
 	int			dno;
 	char	   *refname;
 	int			lineno;
-
-	HeapTuple	tup;
-	TupleDesc	tupdesc;
-	bool		freetup;
-	bool		freetupdesc;
+	Oid			rectypeid;		/* declared type of variable */
+	/* RECFIELDs for this record are chained together for easy access */
+	int			firstfield;		/* dno of first RECFIELD, or -1 if none */
+	/* We always store record variables as "expanded" records */
+	ExpandedRecordHeader *erh;
 } PLpgSQL_rec;
 
 /*
@@ -319,8 +322,12 @@ typedef struct PLpgSQL_recfield
 {
 	PLpgSQL_datum_type dtype;
 	int			dno;
-	char	   *fieldname;
+	char	   *fieldname;		/* name of field */
 	int			recparentno;	/* dno of parent record */
+	int			nextfield;		/* dno of next child, or -1 if none */
+	uint64		rectupledescid; /* record's tupledesc ID as of last lookup */
+	ExpandedRecordFieldInfo finfo;	/* field's attnum and type info */
+	/* if rectupledescid == INVALID_TUPLEDESC_IDENTIFIER, finfo isn't valid */
 } PLpgSQL_recfield;
 
 /*
@@ -903,12 +910,12 @@ typedef struct PLpgSQL_execstate
 
 	bool		readonly_func;
 
-	TupleDesc	rettupdesc;
 	char	   *exitlabel;		/* the "target" label of the current EXIT or
 								 * CONTINUE stmt, if any */
 	ErrorData  *cur_error;		/* current exception handler's error */
 
 	Tuplestorestate *tuple_store;	/* SRFs accumulate results here */
+	TupleDesc	tuple_store_desc;	/* descriptor for tuples in tuple_store */
 	MemoryContext tuple_store_cxt;
 	ResourceOwner tuple_store_owner;
 	ReturnSetInfo *rsi;
@@ -917,6 +924,8 @@ typedef struct PLpgSQL_execstate
 	int			found_varno;
 	int			ndatums;
 	PLpgSQL_datum **datums;
+	/* context containing variable values (same as func's SPI_proc context) */
+	MemoryContext datum_context;
 
 	/*
 	 * paramLI is what we use to pass local variable values to the executor.
@@ -1088,7 +1097,9 @@ extern PLpgSQL_variable *plpgsql_build_variable(const char *refname, int lineno,
 					   PLpgSQL_type *dtype,
 					   bool add2namespace);
 extern PLpgSQL_rec *plpgsql_build_record(const char *refname, int lineno,
-					 bool add2namespace);
+					 Oid rectypeid, bool add2namespace);
+extern PLpgSQL_recfield *plpgsql_build_recfield(PLpgSQL_rec *rec,
+					   const char *fldname);
 extern int plpgsql_recognize_err_condition(const char *condname,
 								bool allow_sqlstate);
 extern PLpgSQL_condition *plpgsql_parse_err_condition(char *condname);

src/pl/plpython/plpy_typeio.c

@@ -384,7 +384,7 @@ PLy_output_setup_func(PLyObToDatum *arg, MemoryContext arg_mcxt,
 		/* We'll set up the per-field data later */
 		arg->u.tuple.recdesc = NULL;
 		arg->u.tuple.typentry = typentry;
-		arg->u.tuple.tupdescseq = typentry ? typentry->tupDescSeqNo - 1 : 0;
+		arg->u.tuple.tupdescid = INVALID_TUPLEDESC_IDENTIFIER;
 		arg->u.tuple.atts = NULL;
 		arg->u.tuple.natts = 0;
 		/* Mark this invalid till needed, too */
@@ -499,7 +499,7 @@ PLy_input_setup_func(PLyDatumToOb *arg, MemoryContext arg_mcxt,
 		/* We'll set up the per-field data later */
 		arg->u.tuple.recdesc = NULL;
 		arg->u.tuple.typentry = typentry;
-		arg->u.tuple.tupdescseq = typentry ? typentry->tupDescSeqNo - 1 : 0;
+		arg->u.tuple.tupdescid = INVALID_TUPLEDESC_IDENTIFIER;
 		arg->u.tuple.atts = NULL;
 		arg->u.tuple.natts = 0;
 	}
@@ -969,11 +969,11 @@ PLyObject_ToComposite(PLyObToDatum *arg, PyObject *plrv,
 		/* We should have the descriptor of the type's typcache entry */
 		Assert(desc == arg->u.tuple.typentry->tupDesc);
 		/* Detect change of descriptor, update cache if needed */
-		if (arg->u.tuple.tupdescseq != arg->u.tuple.typentry->tupDescSeqNo)
+		if (arg->u.tuple.tupdescid != arg->u.tuple.typentry->tupDesc_identifier)
 		{
 			PLy_output_setup_tuple(arg, desc,
 								   PLy_current_execution_context()->curr_proc);
-			arg->u.tuple.tupdescseq = arg->u.tuple.typentry->tupDescSeqNo;
+			arg->u.tuple.tupdescid = arg->u.tuple.typentry->tupDesc_identifier;
 		}
 	}
 	else

src/pl/plpython/plpy_typeio.h

@@ -42,7 +42,7 @@ typedef struct PLyTupleToOb
 	TupleDesc	recdesc;
 	/* If we're dealing with a named composite type, these fields are set: */
 	TypeCacheEntry *typentry;	/* typcache entry for type */
-	int64		tupdescseq;		/* last tupdesc seqno seen in typcache */
+	uint64		tupdescid;		/* last tupdesc identifier seen in typcache */
 	/* These fields are NULL/0 if not yet set: */
 	PLyDatumToOb *atts;			/* array of per-column conversion info */
 	int			natts;			/* length of array */
@@ -107,7 +107,7 @@ typedef struct PLyObToTuple
 	TupleDesc	recdesc;
 	/* If we're dealing with a named composite type, these fields are set: */
 	TypeCacheEntry *typentry;	/* typcache entry for type */
-	int64		tupdescseq;		/* last tupdesc seqno seen in typcache */
+	uint64		tupdescid;		/* last tupdesc identifier seen in typcache */
 	/* These fields are NULL/0 if not yet set: */
 	PLyObToDatum *atts;			/* array of per-column conversion info */
 	int			natts;			/* length of array */

src/test/regress/expected/plpgsql.out

@@ -4595,23 +4595,32 @@ begin
       x int;
       y int := i;
       r record;
+      c int8_tbl;
     begin
       if i = 1 then
         x := 42;
         r := row(i, i+1);
+        c := row(i, i+1);
       end if;
       raise notice 'x = %', x;
       raise notice 'y = %', y;
       raise notice 'r = %', r;
+      raise notice 'c = %', c;
     end;
   end loop;
 end$$;
 NOTICE:  x = 42
 NOTICE:  y = 1
 NOTICE:  r = (1,2)
+NOTICE:  c = (1,2)
 NOTICE:  x = <NULL>
 NOTICE:  y = 2
-ERROR:  record "r" is not assigned yet
+NOTICE:  r = <NULL>
+NOTICE:  c = <NULL>
+NOTICE:  x = <NULL>
+NOTICE:  y = 3
+NOTICE:  r = <NULL>
+NOTICE:  c = <NULL>
 \set VERBOSITY default
 -- Check handling of conflicts between plpgsql vars and table columns.
 set plpgsql.variable_conflict = error;

src/test/regress/sql/plpgsql.sql

@@ -3745,14 +3745,17 @@ begin
       x int;
       y int := i;
       r record;
+      c int8_tbl;
     begin
       if i = 1 then
         x := 42;
         r := row(i, i+1);
+        c := row(i, i+1);
       end if;
       raise notice 'x = %', x;
       raise notice 'y = %', y;
       raise notice 'r = %', r;
+      raise notice 'c = %', c;
     end;
   end loop;
 end$$;