Support ordered-set (WITHIN GROUP) aggregates.

This patch introduces generic support for ordered-set and hypothetical-set
aggregate functions, as well as implementations of the instances defined in
SQL:2008 (percentile_cont(), percentile_disc(), rank(), dense_rank(),
percent_rank(), cume_dist()).  We also added mode() though it is not in the
spec, as well as versions of percentile_cont() and percentile_disc() that
can compute multiple percentile values in one pass over the data.

Unlike the original submission, this patch puts full control of the sorting
process in the hands of the aggregate's support functions.  To allow the
support functions to find out how they're supposed to sort, a new API
function AggGetAggref() is added to nodeAgg.c.  This allows retrieval of
the aggregate call's Aggref node, which may have other uses beyond the
immediate need.  There is also support for ordered-set aggregates to
install cleanup callback functions, so that they can be sure that
infrastructure such as tuplesort objects gets cleaned up.

In passing, make some fixes in the recently-added support for variadic
aggregates, and make some editorial adjustments in the recent FILTER
additions for aggregates.  Also, simplify use of IsBinaryCoercible() by
allowing it to succeed whenever the target type is ANY or ANYELEMENT.
It was inconsistent that it dealt with other polymorphic target types
but not these.

Atri Sharma and Andrew Gierth; reviewed by Pavel Stehule and Vik Fearing,
and rather heavily editorialized upon by Tom Lane

4 files changed, 261 insertions, 101 deletions

diff --git a/src/backend/executor/execQual.c b/src/backend/executor/execQual.c
index 90c27530e9c..67dca78b771 100644
--- a/src/backend/executor/execQual.c
+++ b/src/backend/executor/execQual.c
@@ -4408,6 +4408,8 @@ ExecInitExpr(Expr *node, PlanState *parent)
 					aggstate->aggs = lcons(astate, aggstate->aggs);
 					naggs = ++aggstate->numaggs;
 
+					astate->aggdirectargs = (List *) ExecInitExpr((Expr *) aggref->aggdirectargs,
+																  parent);
 					astate->args = (List *) ExecInitExpr((Expr *) aggref->args,
 														 parent);
 					astate->aggfilter = ExecInitExpr(aggref->aggfilter,
diff --git a/src/backend/executor/functions.c b/src/backend/executor/functions.c
index ff6a123bc40..893a54b21bb 100644
--- a/src/backend/executor/functions.c
+++ b/src/backend/executor/functions.c
@@ -380,8 +380,8 @@ sql_fn_post_column_ref(ParseState *pstate, ColumnRef *cref, Node *var)
 		param = ParseFuncOrColumn(pstate,
 								  list_make1(subfield),
 								  list_make1(param),
-								  NIL, NULL, false, false, false,
-								  NULL, true, cref->location);
+								  NULL,
+								  cref->location);
 	}
 
 	return param;
diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c
index e02a6ffa8c3..0e2160d2f1c 100644
--- a/src/backend/executor/nodeAgg.c
+++ b/src/backend/executor/nodeAgg.c
@@ -8,14 +8,16 @@
  *		 transvalue = initcond
  *		 foreach input_tuple do
  *			transvalue = transfunc(transvalue, input_value(s))
- *		 result = finalfunc(transvalue)
+ *		 result = finalfunc(transvalue, direct_argument(s))
  *
  *	  If a finalfunc is not supplied then the result is just the ending
  *	  value of transvalue.
  *
- *	  If an aggregate call specifies DISTINCT or ORDER BY, we sort the input
- *	  tuples and eliminate duplicates (if required) before performing the
- *	  above-depicted process.
+ *	  If a normal aggregate call specifies DISTINCT or ORDER BY, we sort the
+ *	  input tuples and eliminate duplicates (if required) before performing
+ *	  the above-depicted process.  (However, we don't do that for ordered-set
+ *	  aggregates; their "ORDER BY" inputs are ordinary aggregate arguments
+ *	  so far as this module is concerned.)
  *
  *	  If transfunc is marked "strict" in pg_proc and initcond is NULL,
  *	  then the first non-NULL input_value is assigned directly to transvalue,
@@ -33,6 +35,14 @@
  *	  of course).  A non-strict finalfunc can make its own choice of
  *	  what to return for a NULL ending transvalue.
  *
+ *	  Ordered-set aggregates are treated specially in one other way: we
+ *	  evaluate any "direct" arguments and pass them to the finalfunc along
+ *	  with the transition value.  In addition, NULL placeholders are
+ *	  provided to match the remaining finalfunc arguments, which correspond
+ *	  to the aggregated expressions.  (These arguments have no use at
+ *	  runtime, but may be needed to allow resolution of a polymorphic
+ *	  aggregate's result type.)
+ *
  *	  We compute aggregate input expressions and run the transition functions
  *	  in a temporary econtext (aggstate->tmpcontext).  This is reset at
  *	  least once per input tuple, so when the transvalue datatype is
@@ -40,7 +50,7 @@
  *	  memory context, and free the prior value to avoid memory leakage.
  *	  We store transvalues in the memory context aggstate->aggcontext,
  *	  which is also used for the hashtable structures in AGG_HASHED mode.
- *	  The node's regular econtext (aggstate->csstate.cstate.cs_ExprContext)
+ *	  The node's regular econtext (aggstate->ss.ps.ps_ExprContext)
  *	  is used to run finalize functions and compute the output tuple;
  *	  this context can be reset once per output tuple.
  *
@@ -66,6 +76,13 @@
  *	  AggState is available as context in earlier releases (back to 8.1),
  *	  but direct examination of the node is needed to use it before 9.0.
  *
+ *	  As of 9.4, aggregate transition functions can also use AggGetAggref()
+ *	  to get hold of the Aggref expression node for their aggregate call.
+ *	  This is mainly intended for ordered-set aggregates, which are not
+ *	  supported as window functions.  (A regular aggregate function would
+ *	  need some fallback logic to use this, since there's no Aggref node
+ *	  for a window function.)
+ *
  *
  * Portions Copyright (c) 1996-2013, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
@@ -82,7 +99,6 @@
 #include "catalog/objectaccess.h"
 #include "catalog/pg_aggregate.h"
 #include "catalog/pg_proc.h"
-#include "catalog/pg_type.h"
 #include "executor/executor.h"
 #include "executor/nodeAgg.h"
 #include "miscadmin.h"
@@ -114,12 +130,27 @@ typedef struct AggStatePerAggData
 	AggrefExprState *aggrefstate;
 	Aggref	   *aggref;
 
-	/* number of input arguments for aggregate function proper */
+	/*
+	 * Nominal number of arguments for aggregate function.	For plain aggs,
+	 * this excludes any ORDER BY expressions.	For ordered-set aggs, this
+	 * counts both the direct and aggregated (ORDER BY) arguments.
+	 */
 	int			numArguments;
 
-	/* number of inputs including ORDER BY expressions */
+	/*
+	 * Number of aggregated input columns.	This includes ORDER BY expressions
+	 * in both the plain-agg and ordered-set cases.  Ordered-set direct args
+	 * are not counted, though.
+	 */
 	int			numInputs;
 
+	/*
+	 * Number of aggregated input columns to pass to the transfn.  This
+	 * includes the ORDER BY columns for ordered-set aggs, but not for plain
+	 * aggs.  (This doesn't count the transition state value!)
+	 */
+	int			numTransInputs;
+
 	/* Oids of transfer functions */
 	Oid			transfn_oid;
 	Oid			finalfn_oid;	/* may be InvalidOid */
@@ -379,7 +410,7 @@ advance_transition_function(AggState *aggstate,
 							AggStatePerGroup pergroupstate,
 							FunctionCallInfoData *fcinfo)
 {
-	int			numArguments = peraggstate->numArguments;
+	int			numTransInputs = peraggstate->numTransInputs;
 	MemoryContext oldContext;
 	Datum		newVal;
 	int			i;
@@ -390,7 +421,7 @@ advance_transition_function(AggState *aggstate,
 		 * For a strict transfn, nothing happens when there's a NULL input; we
 		 * just keep the prior transValue.
 		 */
-		for (i = 1; i <= numArguments; i++)
+		for (i = 1; i <= numTransInputs; i++)
 		{
 			if (fcinfo->argnull[i])
 				return;
@@ -430,11 +461,14 @@ advance_transition_function(AggState *aggstate,
 	/* We run the transition functions in per-input-tuple memory context */
 	oldContext = MemoryContextSwitchTo(aggstate->tmpcontext->ecxt_per_tuple_memory);
 
+	/* set up aggstate->curperagg for AggGetAggref() */
+	aggstate->curperagg = peraggstate;
+
 	/*
 	 * OK to call the transition function
 	 */
 	InitFunctionCallInfoData(*fcinfo, &(peraggstate->transfn),
-							 numArguments + 1,
+							 numTransInputs + 1,
 							 peraggstate->aggCollation,
 							 (void *) aggstate, NULL);
 	fcinfo->arg[0] = pergroupstate->transValue;
@@ -442,6 +476,8 @@ advance_transition_function(AggState *aggstate,
 
 	newVal = FunctionCallInvoke(fcinfo);
 
+	aggstate->curperagg = NULL;
+
 	/*
 	 * If pass-by-ref datatype, must copy the new value into aggcontext and
 	 * pfree the prior transValue.	But if transfn returned a pointer to its
@@ -485,15 +521,15 @@ advance_aggregates(AggState *aggstate, AggStatePerGroup pergroup)
 		AggStatePerAgg peraggstate = &aggstate->peragg[aggno];
 		AggStatePerGroup pergroupstate = &pergroup[aggno];
 		ExprState  *filter = peraggstate->aggrefstate->aggfilter;
-		int			nargs = peraggstate->numArguments;
+		int			numTransInputs = peraggstate->numTransInputs;
 		int			i;
 		TupleTableSlot *slot;
 
 		/* Skip anything FILTERed out */
 		if (filter)
 		{
-			bool		isnull;
 			Datum		res;
+			bool		isnull;
 
 			res = ExecEvalExprSwitchContext(filter, aggstate->tmpcontext,
 											&isnull, NULL);
@@ -512,18 +548,18 @@ advance_aggregates(AggState *aggstate, AggStatePerGroup pergroup)
 			/*
 			 * If the transfn is strict, we want to check for nullity before
 			 * storing the row in the sorter, to save space if there are a lot
-			 * of nulls.  Note that we must only check numArguments columns,
+			 * of nulls.  Note that we must only check numTransInputs columns,
 			 * not numInputs, since nullity in columns used only for sorting
 			 * is not relevant here.
 			 */
 			if (peraggstate->transfn.fn_strict)
 			{
-				for (i = 0; i < nargs; i++)
+				for (i = 0; i < numTransInputs; i++)
 				{
 					if (slot->tts_isnull[i])
 						break;
 				}
-				if (i < nargs)
+				if (i < numTransInputs)
 					continue;
 			}
 
@@ -542,8 +578,8 @@ advance_aggregates(AggState *aggstate, AggStatePerGroup pergroup)
 
 			/* Load values into fcinfo */
 			/* Start from 1, since the 0th arg will be the transition value */
-			Assert(slot->tts_nvalid >= nargs);
-			for (i = 0; i < nargs; i++)
+			Assert(slot->tts_nvalid >= numTransInputs);
+			for (i = 0; i < numTransInputs; i++)
 			{
 				fcinfo.arg[i + 1] = slot->tts_values[i];
 				fcinfo.argnull[i + 1] = slot->tts_isnull[i];
@@ -671,7 +707,7 @@ process_ordered_aggregate_multi(AggState *aggstate,
 	FunctionCallInfoData fcinfo;
 	TupleTableSlot *slot1 = peraggstate->evalslot;
 	TupleTableSlot *slot2 = peraggstate->uniqslot;
-	int			numArguments = peraggstate->numArguments;
+	int			numTransInputs = peraggstate->numTransInputs;
 	int			numDistinctCols = peraggstate->numDistinctCols;
 	bool		haveOldValue = false;
 	int			i;
@@ -685,10 +721,11 @@ process_ordered_aggregate_multi(AggState *aggstate,
 	while (tuplesort_gettupleslot(peraggstate->sortstate, true, slot1))
 	{
 		/*
-		 * Extract the first numArguments as datums to pass to the transfn.
-		 * (This will help execTuplesMatch too, so do it immediately.)
+		 * Extract the first numTransInputs columns as datums to pass to the
+		 * transfn.  (This will help execTuplesMatch too, so we do it
+		 * immediately.)
 		 */
-		slot_getsomeattrs(slot1, numArguments);
+		slot_getsomeattrs(slot1, numTransInputs);
 
 		if (numDistinctCols == 0 ||
 			!haveOldValue ||
@@ -700,7 +737,7 @@ process_ordered_aggregate_multi(AggState *aggstate,
 		{
 			/* Load values into fcinfo */
 			/* Start from 1, since the 0th arg will be the transition value */
-			for (i = 0; i < numArguments; i++)
+			for (i = 0; i < numTransInputs; i++)
 			{
 				fcinfo.arg[i + 1] = slot1->tts_values[i];
 				fcinfo.argnull[i + 1] = slot1->tts_isnull[i];
@@ -746,23 +783,73 @@ finalize_aggregate(AggState *aggstate,
 				   AggStatePerGroup pergroupstate,
 				   Datum *resultVal, bool *resultIsNull)
 {
+	FunctionCallInfoData fcinfo;
+	bool		anynull = false;
 	MemoryContext oldContext;
+	int			i;
+	ListCell   *lc;
 
 	oldContext = MemoryContextSwitchTo(aggstate->ss.ps.ps_ExprContext->ecxt_per_tuple_memory);
 
 	/*
+	 * Evaluate any direct arguments.  We do this even if there's no finalfn
+	 * (which is unlikely anyway), so that side-effects happen as expected.
+	 */
+	i = 1;
+	foreach(lc, peraggstate->aggrefstate->aggdirectargs)
+	{
+		ExprState  *expr = (ExprState *) lfirst(lc);
+
+		fcinfo.arg[i] = ExecEvalExpr(expr,
+									 aggstate->ss.ps.ps_ExprContext,
+									 &fcinfo.argnull[i],
+									 NULL);
+		anynull |= fcinfo.argnull[i];
+		i++;
+	}
+
+	/*
 	 * Apply the agg's finalfn if one is provided, else return transValue.
 	 */
 	if (OidIsValid(peraggstate->finalfn_oid))
 	{
-		FunctionCallInfoData fcinfo;
+		int			numFinalArgs;
 
-		InitFunctionCallInfoData(fcinfo, &(peraggstate->finalfn), 1,
+		/*
+		 * Identify number of arguments being passed to the finalfn.  For a
+		 * plain agg it's just one (the transition state value).  For
+		 * ordered-set aggs we also pass the direct argument(s), plus nulls
+		 * corresponding to the aggregate-input columns.
+		 */
+		if (AGGKIND_IS_ORDERED_SET(peraggstate->aggref->aggkind))
+			numFinalArgs = peraggstate->numArguments + 1;
+		else
+			numFinalArgs = 1;
+		Assert(i <= numFinalArgs);
+
+		/* set up aggstate->curperagg for AggGetAggref() */
+		aggstate->curperagg = peraggstate;
+
+		InitFunctionCallInfoData(fcinfo, &(peraggstate->finalfn),
+								 numFinalArgs,
 								 peraggstate->aggCollation,
 								 (void *) aggstate, NULL);
+
+		/* Fill in the transition state value */
 		fcinfo.arg[0] = pergroupstate->transValue;
 		fcinfo.argnull[0] = pergroupstate->transValueIsNull;
-		if (fcinfo.flinfo->fn_strict && pergroupstate->transValueIsNull)
+		anynull |= pergroupstate->transValueIsNull;
+
+		/* Fill any remaining argument positions with nulls */
+		while (i < numFinalArgs)
+		{
+			fcinfo.arg[i] = (Datum) 0;
+			fcinfo.argnull[i] = true;
+			anynull = true;
+			i++;
+		}
+
+		if (fcinfo.flinfo->fn_strict && anynull)
 		{
 			/* don't call a strict function with NULL inputs */
 			*resultVal = (Datum) 0;
@@ -773,6 +860,7 @@ finalize_aggregate(AggState *aggstate,
 			*resultVal = FunctionCallInvoke(&fcinfo);
 			*resultIsNull = fcinfo.isnull;
 		}
+		aggstate->curperagg = NULL;
 	}
 	else
 	{
@@ -1094,8 +1182,13 @@ agg_retrieve_direct(AggState *aggstate)
 		 * aggcontext (which contains any pass-by-ref transvalues of the old
 		 * group).	We also clear any child contexts of the aggcontext; some
 		 * aggregate functions store working state in such contexts.
+		 *
+		 * We use ReScanExprContext not just ResetExprContext because we want
+		 * any registered shutdown callbacks to be called.	That allows
+		 * aggregate functions to ensure they've cleaned up any non-memory
+		 * resources.
 		 */
-		ResetExprContext(econtext);
+		ReScanExprContext(econtext);
 
 		MemoryContextResetAndDeleteChildren(aggstate->aggcontext);
 
@@ -1164,6 +1257,16 @@ agg_retrieve_direct(AggState *aggstate)
 		}
 
 		/*
+		 * Use the representative input tuple for any references to
+		 * non-aggregated input columns in aggregate direct args, the node
+		 * qual, and the tlist.  (If we are not grouping, and there are no
+		 * input rows at all, we will come here with an empty firstSlot ...
+		 * but if not grouping, there can't be any references to
+		 * non-aggregated input columns, so no problem.)
+		 */
+		econtext->ecxt_outertuple = firstSlot;
+
+		/*
 		 * Done scanning input tuple group. Finalize each aggregate
 		 * calculation, and stash results in the per-output-tuple context.
 		 */
@@ -1189,15 +1292,6 @@ agg_retrieve_direct(AggState *aggstate)
 		}
 
 		/*
-		 * Use the representative input tuple for any references to
-		 * non-aggregated input columns in the qual and tlist.	(If we are not
-		 * grouping, and there are no input rows at all, we will come here
-		 * with an empty firstSlot ... but if not grouping, there can't be any
-		 * references to non-aggregated input columns, so no problem.)
-		 */
-		econtext->ecxt_outertuple = firstSlot;
-
-		/*
 		 * Check the qual (HAVING clause); if the group does not match, ignore
 		 * it and loop back to try to process another group.
 		 */
@@ -1316,6 +1410,10 @@ agg_retrieve_hash_table(AggState *aggstate)
 
 		/*
 		 * Clear the per-output-tuple context for each group
+		 *
+		 * We intentionally don't use ReScanExprContext here; if any aggs have
+		 * registered shutdown callbacks, they mustn't be called yet, since we
+		 * might not be done with that agg.
 		 */
 		ResetExprContext(econtext);
 
@@ -1412,6 +1510,7 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 	aggstate->eqfunctions = NULL;
 	aggstate->hashfunctions = NULL;
 	aggstate->peragg = NULL;
+	aggstate->curperagg = NULL;
 	aggstate->agg_done = false;
 	aggstate->pergroup = NULL;
 	aggstate->grp_firstTuple = NULL;
@@ -1565,6 +1664,7 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 		AggStatePerAgg peraggstate;
 		Oid			inputTypes[FUNC_MAX_ARGS];
 		int			numArguments;
+		int			numDirectArgs;
 		int			numInputs;
 		int			numSortCols;
 		int			numDistinctCols;
@@ -1604,28 +1704,12 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 		/* Mark Aggref state node with assigned index in the result array */
 		aggrefstate->aggno = aggno;
 
-		/* Fill in the peraggstate data */
+		/* Begin filling in the peraggstate data */
 		peraggstate->aggrefstate = aggrefstate;
 		peraggstate->aggref = aggref;
-		numInputs = list_length(aggref->args);
-		peraggstate->numInputs = numInputs;
 		peraggstate->sortstate = NULL;
 
-		/*
-		 * Get actual datatypes of the inputs.	These could be different from
-		 * the agg's declared input types, when the agg accepts ANY or a
-		 * polymorphic type.
-		 */
-		numArguments = 0;
-		foreach(lc, aggref->args)
-		{
-			TargetEntry *tle = (TargetEntry *) lfirst(lc);
-
-			if (!tle->resjunk)
-				inputTypes[numArguments++] = exprType((Node *) tle->expr);
-		}
-		peraggstate->numArguments = numArguments;
-
+		/* Fetch the pg_aggregate row */
 		aggTuple = SearchSysCache1(AGGFNOID,
 								   ObjectIdGetDatum(aggref->aggfnoid));
 		if (!HeapTupleIsValid(aggTuple))
@@ -1674,28 +1758,38 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 			}
 		}
 
+		/*
+		 * Get actual datatypes of the (nominal) aggregate inputs.	These
+		 * could be different from the agg's declared input types, when the
+		 * agg accepts ANY or a polymorphic type.
+		 */
+		numArguments = get_aggregate_argtypes(aggref, inputTypes);
+		peraggstate->numArguments = numArguments;
+
+		/* Count the "direct" arguments, if any */
+		numDirectArgs = list_length(aggref->aggdirectargs);
+
+		/* Count the number of aggregated input columns */
+		numInputs = list_length(aggref->args);
+		peraggstate->numInputs = numInputs;
+
+		/* Detect how many columns to pass to the transfn */
+		if (AGGKIND_IS_ORDERED_SET(aggref->aggkind))
+			peraggstate->numTransInputs = numInputs;
+		else
+			peraggstate->numTransInputs = numArguments;
+
 		/* resolve actual type of transition state, if polymorphic */
-		aggtranstype = aggform->aggtranstype;
-		if (IsPolymorphicType(aggtranstype))
-		{
-			/* have to fetch the agg's declared input types... */
-			Oid		   *declaredArgTypes;
-			int			agg_nargs;
-
-			(void) get_func_signature(aggref->aggfnoid,
-									  &declaredArgTypes, &agg_nargs);
-			Assert(agg_nargs == numArguments);
-			aggtranstype = enforce_generic_type_consistency(inputTypes,
-															declaredArgTypes,
-															agg_nargs,
-															aggtranstype,
-															false);
-			pfree(declaredArgTypes);
-		}
+		aggtranstype = resolve_aggregate_transtype(aggref->aggfnoid,
+												   aggform->aggtranstype,
+												   inputTypes,
+												   numArguments);
 
 		/* build expression trees using actual argument & result types */
 		build_aggregate_fnexprs(inputTypes,
 								numArguments,
+								numDirectArgs,
+								AGGKIND_IS_ORDERED_SET(aggref->aggkind),
 								aggref->aggvariadic,
 								aggtranstype,
 								aggref->aggtype,
@@ -1740,14 +1834,14 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 		/*
 		 * If the transfn is strict and the initval is NULL, make sure input
 		 * type and transtype are the same (or at least binary-compatible), so
-		 * that it's OK to use the first input value as the initial
+		 * that it's OK to use the first aggregated input value as the initial
 		 * transValue.	This should have been checked at agg definition time,
 		 * but just in case...
 		 */
 		if (peraggstate->transfn.fn_strict && peraggstate->initValueIsNull)
 		{
-			if (numArguments < 1 ||
-				!IsBinaryCoercible(inputTypes[0], aggtranstype))
+			if (numArguments <= numDirectArgs ||
+				!IsBinaryCoercible(inputTypes[numDirectArgs], aggtranstype))
 				ereport(ERROR,
 						(errcode(ERRCODE_INVALID_FUNCTION_DEFINITION),
 						 errmsg("aggregate %u needs to have compatible input type and transition type",
@@ -1755,8 +1849,8 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 		}
 
 		/*
-		 * Get a tupledesc corresponding to the inputs (including sort
-		 * expressions) of the agg.
+		 * Get a tupledesc corresponding to the aggregated inputs (including
+		 * sort expressions) of the agg.
 		 */
 		peraggstate->evaldesc = ExecTypeFromTL(aggref->args, false);
 
@@ -1771,14 +1865,20 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 														NULL);
 
 		/*
-		 * If we're doing either DISTINCT or ORDER BY, then we have a list of
-		 * SortGroupClause nodes; fish out the data in them and stick them
-		 * into arrays.
+		 * If we're doing either DISTINCT or ORDER BY for a plain agg, then we
+		 * have a list of SortGroupClause nodes; fish out the data in them and
+		 * stick them into arrays.	We ignore ORDER BY for an ordered-set agg,
+		 * however; the agg's transfn and finalfn are responsible for that.
 		 *
 		 * Note that by construction, if there is a DISTINCT clause then the
 		 * ORDER BY clause is a prefix of it (see transformDistinctClause).
 		 */
-		if (aggref->aggdistinct)
+		if (AGGKIND_IS_ORDERED_SET(aggref->aggkind))
+		{
+			sortlist = NIL;
+			numSortCols = numDistinctCols = 0;
+		}
+		else if (aggref->aggdistinct)
 		{
 			sortlist = aggref->aggdistinct;
 			numSortCols = numDistinctCols = list_length(sortlist);
@@ -1805,7 +1905,7 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 			/* If we have only one input, we need its len/byval info. */
 			if (numInputs == 1)
 			{
-				get_typlenbyval(inputTypes[0],
+				get_typlenbyval(inputTypes[numDirectArgs],
 								&peraggstate->inputtypeLen,
 								&peraggstate->inputtypeByVal);
 			}
@@ -1908,6 +2008,9 @@ ExecEndAgg(AggState *node)
 			tuplesort_end(peraggstate->sortstate);
 	}
 
+	/* And ensure any agg shutdown callbacks have been called */
+	ReScanExprContext(node->ss.ps.ps_ExprContext);
+
 	/*
 	 * Free both the expr contexts.
 	 */
@@ -1967,6 +2070,8 @@ ExecReScanAgg(AggState *node)
 		peraggstate->sortstate = NULL;
 	}
 
+	/* We don't need to ReScanExprContext here; ExecReScan already did it */
+
 	/* Release first tuple of group, if we have made a copy */
 	if (node->grp_firstTuple != NULL)
 	{
@@ -2047,6 +2152,71 @@ AggCheckCallContext(FunctionCallInfo fcinfo, MemoryContext *aggcontext)
 }
 
 /*
+ * AggGetAggref - allow an aggregate support function to get its Aggref
+ *
+ * If the function is being called as an aggregate support function,
+ * return the Aggref node for the aggregate call.  Otherwise, return NULL.
+ *
+ * Note that if an aggregate is being used as a window function, this will
+ * return NULL.  We could provide a similar function to return the relevant
+ * WindowFunc node in such cases, but it's not needed yet.
+ */
+Aggref *
+AggGetAggref(FunctionCallInfo fcinfo)
+{
+	if (fcinfo->context && IsA(fcinfo->context, AggState))
+	{
+		AggStatePerAgg curperagg = ((AggState *) fcinfo->context)->curperagg;
+
+		if (curperagg)
+			return curperagg->aggref;
+	}
+	return NULL;
+}
+
+/*
+ * AggGetPerTupleEContext - fetch per-input-tuple ExprContext
+ *
+ * This is useful in agg final functions; the econtext returned is the
+ * same per-tuple context that the transfn was called in (which can
+ * safely get reset during the final function).
+ *
+ * As above, this is currently not useful for aggs called as window functions.
+ */
+ExprContext *
+AggGetPerTupleEContext(FunctionCallInfo fcinfo)
+{
+	if (fcinfo->context && IsA(fcinfo->context, AggState))
+	{
+		AggState   *aggstate = (AggState *) fcinfo->context;
+
+		return aggstate->tmpcontext;
+	}
+	return NULL;
+}
+
+/*
+ * AggGetPerAggEContext - fetch per-output-tuple ExprContext
+ *
+ * This is useful for aggs to register shutdown callbacks, which will ensure
+ * that non-memory resources are freed.
+ *
+ * As above, this is currently not useful for aggs called as window functions.
+ */
+ExprContext *
+AggGetPerAggEContext(FunctionCallInfo fcinfo)
+{
+	if (fcinfo->context && IsA(fcinfo->context, AggState))
+	{
+		AggState   *aggstate = (AggState *) fcinfo->context;
+
+		return aggstate->ss.ps.ps_ExprContext;
+	}
+	return NULL;
+}
+
+
+/*
  * aggregate_dummy - dummy execution routine for aggregate functions
  *
  * This function is listed as the implementation (prosrc field) of pg_proc
diff --git a/src/backend/executor/nodeWindowAgg.c b/src/backend/executor/nodeWindowAgg.c
index 544ba989de1..81ad9337dbd 100644
--- a/src/backend/executor/nodeWindowAgg.c
+++ b/src/backend/executor/nodeWindowAgg.c
@@ -37,7 +37,6 @@
 #include "catalog/objectaccess.h"
 #include "catalog/pg_aggregate.h"
 #include "catalog/pg_proc.h"
-#include "catalog/pg_type.h"
 #include "executor/executor.h"
 #include "executor/nodeWindowAgg.h"
 #include "miscadmin.h"
@@ -1796,27 +1795,16 @@ initialize_peragg(WindowAggState *winstate, WindowFunc *wfunc,
 	}
 
 	/* resolve actual type of transition state, if polymorphic */
-	aggtranstype = aggform->aggtranstype;
-	if (IsPolymorphicType(aggtranstype))
-	{
-		/* have to fetch the agg's declared input types... */
-		Oid		   *declaredArgTypes;
-		int			agg_nargs;
-
-		get_func_signature(wfunc->winfnoid,
-						   &declaredArgTypes, &agg_nargs);
-		Assert(agg_nargs == numArguments);
-		aggtranstype = enforce_generic_type_consistency(inputTypes,
-														declaredArgTypes,
-														agg_nargs,
-														aggtranstype,
-														false);
-		pfree(declaredArgTypes);
-	}
+	aggtranstype = resolve_aggregate_transtype(wfunc->winfnoid,
+											   aggform->aggtranstype,
+											   inputTypes,
+											   numArguments);
 
 	/* build expression trees using actual argument & result types */
 	build_aggregate_fnexprs(inputTypes,
 							numArguments,
+							0,	/* no ordered-set window functions yet */
+							false,
 							false,		/* no variadic window functions yet */
 							aggtranstype,
 							wfunc->wintype,