Further reduce overhead for passing plpgsql variables to the executor.

This builds on commit 21dcda27 by keeping
a plpgsql function's shared ParamListInfo's entries for simple variables
(PLPGSQL_DTYPE_VARs) valid at all times.  That adds a few cycles to each
assignment to such variables, but saves significantly more cycles each time
they are used; so except in the pathological case of many dead stores, this
should always be a win.  Initial testing says it's good for about a 10%
speedup of simple calculations; more in large functions with many datums.

We can't use this method for row/record references unfortunately, so what
we do for those is reset those ParamListInfo slots after use; which we
can skip doing unless some of them were actually evaluated during the
previous evaluation call.  So this should frequently be a win as well,
while worst case is that it's similar cost to the previous approach.

Also, closer study suggests that the previous method of instantiating a
new ParamListInfo array per evaluation is actually probably optimal for
cursor-opening executor calls.  The reason is that whatever is visible in
the array is going to get copied into the cursor portal via copyParamList.
So if we used the function's main ParamListInfo for those calls, we'd end
up with all of its DTYPE_VAR vars getting copied, which might well include
large pass-by-reference values that the cursor actually has no need for.
To avoid a possible net degradation in cursor cases, go back to creating
and filling a private ParamListInfo in those cases (which therefore will be
exactly the same speed as before 21dcda27).  We still get some benefit
out of this though, because this approach means that we only have to defend
against copyParamList's try-to-fetch-every-slot behavior in the case of an
unshared ParamListInfo; so plpgsql_param_fetch() can skip testing
expr->paramnos in the common case.

To ensure that the main ParamListInfo's image of a DTYPE_VAR datum is
always valid, all assignments to such variables are now funneled through
assign_simple_var().  But this makes for cleaner and shorter code anyway.

src/pl/plpgsql/src/pl_comp.c

@@ -42,7 +42,7 @@ PLpgSQL_stmt_block *plpgsql_parse_result;
 static int	datums_alloc;
 int			plpgsql_nDatums;
 PLpgSQL_datum **plpgsql_Datums;
-static int	datums_last = 0;
+static int	datums_last;
 
 char	   *plpgsql_error_funcname;
 bool		plpgsql_DumpExecTree = false;
@@ -104,6 +104,8 @@ static Node *make_datum_param(PLpgSQL_expr *expr, int dno, int location);
 static PLpgSQL_row *build_row_from_class(Oid classOid);
 static PLpgSQL_row *build_row_from_vars(PLpgSQL_variable **vars, int numvars);
 static PLpgSQL_type *build_datatype(HeapTuple typeTup, int32 typmod, Oid collation);
+static void plpgsql_start_datums(void);
+static void plpgsql_finish_datums(PLpgSQL_function *function);
 static void compute_function_hashkey(FunctionCallInfo fcinfo,
 						 Form_pg_proc procStruct,
 						 PLpgSQL_func_hashkey *hashkey,
@@ -371,13 +373,7 @@ do_compile(FunctionCallInfo fcinfo,
 	plpgsql_ns_init();
 	plpgsql_ns_push(NameStr(procStruct->proname));
 	plpgsql_DumpExecTree = false;
-
-	datums_alloc = 128;
-	plpgsql_nDatums = 0;
-	/* This is short-lived, so needn't allocate in function's cxt */
-	plpgsql_Datums = MemoryContextAlloc(compile_tmp_cxt,
-									 sizeof(PLpgSQL_datum *) * datums_alloc);
-	datums_last = 0;
+	plpgsql_start_datums();
 
 	switch (function->fn_is_trigger)
 	{
@@ -758,10 +754,8 @@ do_compile(FunctionCallInfo fcinfo,
 	function->fn_nargs = procStruct->pronargs;
 	for (i = 0; i < function->fn_nargs; i++)
 		function->fn_argvarnos[i] = in_arg_varnos[i];
-	function->ndatums = plpgsql_nDatums;
-	function->datums = palloc(sizeof(PLpgSQL_datum *) * plpgsql_nDatums);
-	for (i = 0; i < plpgsql_nDatums; i++)
-		function->datums[i] = plpgsql_Datums[i];
+
+	plpgsql_finish_datums(function);
 
 	/* Debug dump for completed functions */
 	if (plpgsql_DumpExecTree)
@@ -804,7 +798,6 @@ plpgsql_compile_inline(char *proc_source)
 	PLpgSQL_variable *var;
 	int			parse_rc;
 	MemoryContext func_cxt;
-	int			i;
 
 	/*
 	 * Setup the scanner input and error info.  We assume that this function
@@ -860,11 +853,7 @@ plpgsql_compile_inline(char *proc_source)
 	plpgsql_ns_init();
 	plpgsql_ns_push(func_name);
 	plpgsql_DumpExecTree = false;
-
-	datums_alloc = 128;
-	plpgsql_nDatums = 0;
-	plpgsql_Datums = palloc(sizeof(PLpgSQL_datum *) * datums_alloc);
-	datums_last = 0;
+	plpgsql_start_datums();
 
 	/* Set up as though in a function returning VOID */
 	function->fn_rettype = VOIDOID;
@@ -911,10 +900,8 @@ plpgsql_compile_inline(char *proc_source)
 	 * Complete the function's info
 	 */
 	function->fn_nargs = 0;
-	function->ndatums = plpgsql_nDatums;
-	function->datums = palloc(sizeof(PLpgSQL_datum *) * plpgsql_nDatums);
-	for (i = 0; i < plpgsql_nDatums; i++)
-		function->datums[i] = plpgsql_Datums[i];
+
+	plpgsql_finish_datums(function);
 
 	/*
 	 * Pop the error context stack
@@ -1965,6 +1952,7 @@ plpgsql_build_record(const char *refname, int lineno, bool add2namespace)
 	rec->tup = NULL;
 	rec->tupdesc = NULL;
 	rec->freetup = false;
+	rec->freetupdesc = false;
 	plpgsql_adddatum((PLpgSQL_datum *) rec);
 	if (add2namespace)
 		plpgsql_ns_additem(PLPGSQL_NSTYPE_REC, rec->dno, rec->refname);
@@ -2311,6 +2299,22 @@ plpgsql_parse_err_condition(char *condname)
 	return prev;
 }
 
+/* ----------
+ * plpgsql_start_datums			Initialize datum list at compile startup.
+ * ----------
+ */
+static void
+plpgsql_start_datums(void)
+{
+	datums_alloc = 128;
+	plpgsql_nDatums = 0;
+	/* This is short-lived, so needn't allocate in function's cxt */
+	plpgsql_Datums = MemoryContextAlloc(compile_tmp_cxt,
+									 sizeof(PLpgSQL_datum *) * datums_alloc);
+	/* datums_last tracks what's been seen by plpgsql_add_initdatums() */
+	datums_last = 0;
+}
+
 /* ----------
  * plpgsql_adddatum			Add a variable, record or row
  *					to the compiler's datum list.
@@ -2329,6 +2333,39 @@ plpgsql_adddatum(PLpgSQL_datum *new)
 	plpgsql_Datums[plpgsql_nDatums++] = new;
 }
 
+/* ----------
+ * plpgsql_finish_datums	Copy completed datum info into function struct.
+ *
+ * This is also responsible for building resettable_datums, a bitmapset
+ * of the dnos of all ROW, REC, and RECFIELD datums in the function.
+ * ----------
+ */
+static void
+plpgsql_finish_datums(PLpgSQL_function *function)
+{
+	Bitmapset  *resettable_datums = NULL;
+	int			i;
+
+	function->ndatums = plpgsql_nDatums;
+	function->datums = palloc(sizeof(PLpgSQL_datum *) * plpgsql_nDatums);
+	for (i = 0; i < plpgsql_nDatums; i++)
+	{
+		function->datums[i] = plpgsql_Datums[i];
+		switch (function->datums[i]->dtype)
+		{
+			case PLPGSQL_DTYPE_ROW:
+			case PLPGSQL_DTYPE_REC:
+			case PLPGSQL_DTYPE_RECFIELD:
+				resettable_datums = bms_add_member(resettable_datums, i);
+				break;
+
+			default:
+				break;
+		}
+	}
+	function->resettable_datums = resettable_datums;
+}
+
 
 /* ----------
  * plpgsql_add_initdatums		Make an array of the datum numbers of

src/pl/plpgsql/src/plpgsql.h

@@ -752,8 +752,12 @@ typedef struct PLpgSQL_function
 	int			extra_warnings;
 	int			extra_errors;
 
+	/* the datums representing the function's local variables */
 	int			ndatums;
 	PLpgSQL_datum **datums;
+	Bitmapset  *resettable_datums;		/* dnos of non-simple vars */
+
+	/* function body parsetree */
 	PLpgSQL_stmt_block *action;
 
 	/* table for performing casts needed in this function */
@@ -796,6 +800,7 @@ typedef struct PLpgSQL_execstate
 
 	/* we pass datums[i] to the executor, when needed, in paramLI->params[i] */
 	ParamListInfo paramLI;
+	bool		params_dirty;	/* T if any resettable datum has been passed */
 
 	/* EState to use for "simple" expression evaluation */
 	EState	   *simple_eval_estate;