Add support for ON EMPTY clause in aggregate and window functions

This commit introduces the ON EMPTY clause, allowing users to specify
a default value to be returned when an aggregate or window function
receives an empty input set (zero processed rows).  The syntax follows
the pattern:

	agg_function(args, default_value ON EMPTY)

The ON EMPTY clause is distinct from standard NULL handling.  It is
only invoked when the aggregate processes no rows at all.  If the
aggregate processes rows that happen to be NULL, the normal transition
logic applies and the ON EMPTY clause is ignored.  Likewise, when all
rows are removed by a FILTER clause the input set is empty and the
ON EMPTY value is returned.  For a window aggregate, the default is
returned for any row whose frame contains no rows.

Implementation adds an 'inputReceived' flag to each per-group
aggregate state.  A dedicated EEOP_AGG_INPUT_RECEIVED expression step
sets the flag once per row that reaches the aggregate -- after any
FILTER but before the strict-input NULL check -- so that a row whose
input is NULL (and is thus skipped by a strict transition function)
still counts as input.  The step is emitted only for aggregates that
carry an ON EMPTY default.  At finalization, if the flag is false,
the default expression is evaluated and returned in place of the
normal result (and final function, if any).  Because a partial
aggregate's leader only sees workers' combined transition states
rather than the original rows, this flag cannot distinguish the
empty-input case under partial aggregation; such aggregates are
therefore marked non-partial so they are not parallelized.

This patch implements the new step in the expression interpreter only;
the feature is fully functional without LLVM.  LLVM/JIT support for
the EEOP_AGG_INPUT_RECEIVED step is added by a separate follow-on
patch.

The default expression must be a constant-like expression coercible to
the aggregate's result type.  It may not reference columns (at any
query level) or contain aggregates, window functions, subqueries, or
volatile functions.  ON EMPTY cannot be combined with DISTINCT, and is
rejected for non-aggregate window functions; it may be used with an
ordered-set (WITHIN GROUP) aggregate, written before the WITHIN GROUP
clause.

Patch also includes documentation and regression tests for aggregates
and window functions, covering empty tables, filtered results, type
coercion, all-NULL inputs to strict aggregates, and view deparsing.

This commit adds fields to Aggref and WindowFunc (and FuncCall).
These node types are serialized into pg_rewrite (views/rules) and
other stored expression trees, which requires a CATALOG_VERSION_NO
bump.  That is omitted here to avoid conflicts with concurrent
commits; the committer should bump it at commit time.

Proposed-by: Peter Eisentraut

Jeevan Chalke

Author: jeevanchalke

doc/src/sgml/syntax.sgml

@@ -1555,6 +1555,10 @@ sqrt(2)
     <primary>FILTER</primary>
    </indexterm>
 
+   <indexterm zone="syntax-aggregates">
+    <primary>ON EMPTY</primary>
+   </indexterm>
+
    <para>
     An <firstterm>aggregate expression</firstterm> represents the
     application of an aggregate function across the rows selected by a
@@ -1564,6 +1568,7 @@ sqrt(2)
 
 <synopsis>
 <replaceable>aggregate_name</replaceable> (<replaceable>expression</replaceable> [ , ... ] [ <replaceable>order_by_clause</replaceable> ] ) [ FILTER ( WHERE <replaceable>filter_clause</replaceable> ) ]
+<replaceable>aggregate_name</replaceable> (<replaceable>expression</replaceable> [ , ... ] , <replaceable>default_expression</replaceable> ON EMPTY [ <replaceable>order_by_clause</replaceable> ] ) [ FILTER ( WHERE <replaceable>filter_clause</replaceable> ) ]
 <replaceable>aggregate_name</replaceable> (ALL <replaceable>expression</replaceable> [ , ... ] [ <replaceable>order_by_clause</replaceable> ] ) [ FILTER ( WHERE <replaceable>filter_clause</replaceable> ) ]
 <replaceable>aggregate_name</replaceable> (DISTINCT <replaceable>expression</replaceable> [ , ... ] [ <replaceable>order_by_clause</replaceable> ] ) [ FILTER ( WHERE <replaceable>filter_clause</replaceable> ) ]
 <replaceable>aggregate_name</replaceable> ( * ) [ FILTER ( WHERE <replaceable>filter_clause</replaceable> ) ]
@@ -1744,6 +1749,36 @@ FROM generate_series(1,10) AS s(i);
 </programlisting>
    </para>
 
+   <para>
+    If <literal>ON EMPTY</literal> is specified, then
+    <replaceable>default_expression</replaceable> supplies the value of the
+    aggregate when it processes no input rows at all; in that case the
+    aggregate's normal final result (and its final function, if any) is not
+    computed.  This is distinct from null handling: it is triggered only by
+    an empty input set, not by null inputs that are merely ignored during
+    aggregation.  Note that when rows exist but are all removed by a
+    <replaceable>filter_clause</replaceable>, the input set is empty and the
+    <literal>ON EMPTY</literal> value is returned.  For example:
+<programlisting>
+SELECT sum(i) AS plain, sum(i, -1 ON EMPTY) AS defaulted
+FROM generate_series(1,10) AS s(i) WHERE i &gt; 100;
+ plain | defaulted
+-------+-----------
+       |        -1
+(1 row)
+</programlisting>
+    <replaceable>default_expression</replaceable> must be a constant-like
+    expression of a type coercible to the aggregate's result type: it may not
+    refer to columns or contain aggregates, window functions, subqueries, or
+    volatile functions.  <literal>ON EMPTY</literal> cannot be combined with
+    <literal>DISTINCT</literal>.  It may be used with an ordered-set
+    aggregate, written before the <literal>WITHIN GROUP</literal> clause.
+    Because a grouped query never produces empty groups,
+    <literal>ON EMPTY</literal> only takes effect for an ungrouped aggregate
+    over zero rows (or a group whose rows are all removed by
+    <replaceable>filter_clause</replaceable>).
+   </para>
+
    <para>
     The predefined aggregate functions are described in <xref
     linkend="functions-aggregate"/>.  Other aggregate functions can be added
@@ -1843,6 +1878,17 @@ EXCLUDE NO OTHERS
     described in <xref linkend="functions-window"/>.
    </para>
 
+   <para>
+    When an aggregate function is used as a window function, an
+    <literal>ON EMPTY</literal> clause may be given after its arguments
+    (using the same syntax as for a plain aggregate call, described in
+    <xref linkend="syntax-aggregates"/>).  The
+    <replaceable>default_expression</replaceable> is then returned for any
+    row whose window frame contains no rows, including a frame that becomes
+    empty as the frame moves.  <literal>ON EMPTY</literal> is not accepted for
+    non-aggregate window functions.
+   </para>
+
    <para>
     <replaceable>window_name</replaceable> is a reference to a named window
     specification defined in the query's <literal>WINDOW</literal> clause.

src/backend/executor/execExpr.c

@@ -3736,6 +3736,70 @@ ExecBuildAggTrans(AggState *aggstate, AggStatePerPhase phase,
 										 state->steps_len - 1);
 		}
 
+		/*
+		 * If aggonempty present, emit step(s) to record that the group has
+		 * received an input row.  This is placed after any FILTER (so that
+		 * filtered-out rows do not count as input) but before the
+		 * strict-input NULL check emitted below (so that a row whose input is
+		 * NULL still counts as input).
+		 *
+		 * The step is emitted once per concurrently-evaluated grouping set,
+		 * mirroring the transition-function calls below.  We skip it entirely
+		 * for aggregates without ON EMPTY so the common case pays no
+		 * overhead.
+		 */
+		{
+			bool		has_aggonempty = false;
+
+			for (int aggno = 0; aggno < aggstate->numaggs; aggno++)
+			{
+				if (aggstate->peragg[aggno].transno == transno &&
+					aggstate->peragg[aggno].aggref->aggonempty != NULL)
+				{
+					has_aggonempty = true;
+					break;
+				}
+			}
+
+			if (has_aggonempty)
+			{
+				scratch.opcode = EEOP_AGG_INPUT_RECEIVED;
+				scratch.d.agg_input_received.transno = transno;
+
+				if (doSort)
+				{
+					int			processGroupingSets = Max(phase->numsets, 1);
+					int			setoff = 0;
+
+					for (int setno = 0; setno < processGroupingSets; setno++)
+					{
+						scratch.d.agg_input_received.setoff = setoff;
+						ExprEvalPushStep(state, &scratch);
+						setoff++;
+					}
+				}
+
+				if (doHash)
+				{
+					int			numHashes = aggstate->num_hashes;
+					int			setoff;
+
+					/* in MIXED mode, there'll be preceding transition values */
+					if (aggstate->aggstrategy != AGG_HASHED)
+						setoff = aggstate->maxsets;
+					else
+						setoff = 0;
+
+					for (int setno = 0; setno < numHashes; setno++)
+					{
+						scratch.d.agg_input_received.setoff = setoff;
+						ExprEvalPushStep(state, &scratch);
+						setoff++;
+					}
+				}
+			}
+		}
+
 		/*
 		 * Evaluate arguments to aggregate/combine function.
 		 */

src/backend/executor/execExprInterp.c

@@ -592,6 +592,7 @@ ExecInterpExpr(ExprState *state, ExprContext *econtext, bool *isnull)
 		&&CASE_EEOP_AGG_STRICT_INPUT_CHECK_ARGS_1,
 		&&CASE_EEOP_AGG_STRICT_INPUT_CHECK_NULLS,
 		&&CASE_EEOP_AGG_PLAIN_PERGROUP_NULLCHECK,
+		&&CASE_EEOP_AGG_INPUT_RECEIVED,
 		&&CASE_EEOP_AGG_PLAIN_TRANS_INIT_STRICT_BYVAL,
 		&&CASE_EEOP_AGG_PLAIN_TRANS_STRICT_BYVAL,
 		&&CASE_EEOP_AGG_PLAIN_TRANS_BYVAL,
@@ -2106,6 +2107,27 @@ ExecInterpExpr(ExprState *state, ExprContext *econtext, bool *isnull)
 			EEO_NEXT();
 		}
 
+		/*
+		 * Mark that an aggregate's per-group state has received an input row
+		 * (used by the ON EMPTY clause).  This is emitted once per row that
+		 * reaches the aggregate, after any FILTER but before the strict-input
+		 * NULL check, so that a row whose input is NULL still counts as
+		 * input. Only emitted for aggregates that actually carry an ON EMPTY
+		 * default.
+		 */
+		EEO_CASE(EEOP_AGG_INPUT_RECEIVED)
+		{
+			AggState   *aggstate = castNode(AggState, state->parent);
+			AggStatePerGroup pergroup_allaggs =
+				aggstate->all_pergroups[op->d.agg_input_received.setoff];
+
+			/* pergroup may be NULL in the hashed/spilled case; just skip */
+			if (pergroup_allaggs != NULL)
+				pergroup_allaggs[op->d.agg_input_received.transno].inputReceived = true;
+
+			EEO_NEXT();
+		}
+
 		/*
 		 * Different types of aggregate transition functions are implemented
 		 * as different types of steps, to avoid incurring unnecessary

src/backend/executor/nodeAgg.c

@@ -649,6 +649,12 @@ initialize_aggregate(AggState *aggstate, AggStatePerTrans pertrans,
 	 * still need to do this.
 	 */
 	pergroupstate->noTransValue = pertrans->initValueIsNull;
+
+	/*
+	 * Initialize the flag used for ON EMPTY to track whether any input rows
+	 * were received.
+	 */
+	pergroupstate->inputReceived = false;
 }
 
 /*
@@ -1057,6 +1063,25 @@ finalize_aggregate(AggState *aggstate,
 
 	oldContext = MemoryContextSwitchTo(aggstate->ss.ps.ps_ExprContext->ecxt_per_tuple_memory);
 
+	/*
+	 * If the ON EMPTY clause is specified with a default value, evaluate and
+	 * return it in cases where no input rows were received.
+	 *
+	 * The inputReceived flag is used to detect the empty set case (zero rows
+	 * processed).
+	 */
+	if (peragg->aggonemptystate != NULL && !pergroupstate->inputReceived)
+	{
+		*resultVal = ExecEvalExpr(peragg->aggonemptystate,
+								  aggstate->ss.ps.ps_ExprContext,
+								  resultIsNull);
+
+		/* Switch back to the caller's context before returning */
+		MemoryContextSwitchTo(oldContext);
+
+		return;
+	}
+
 	/*
 	 * Evaluate any direct arguments.  We do this even if there's no finalfn
 	 * (which is unlikely anyway), so that side-effects happen as expected.
@@ -3959,6 +3984,13 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 							   get_func_name(transfn_oid));
 			InvokeFunctionExecuteHook(transfn_oid);
 
+			/* Build expression state for ON EMPTY default expression */
+			if (aggref->aggonempty)
+				peragg->aggonemptystate = ExecInitExpr(aggref->aggonempty,
+													   (PlanState *) aggstate);
+			else
+				peragg->aggonemptystate = NULL;
+
 			/*
 			 * initval is potentially null, so don't try to access it as a
 			 * struct field. Must do it the hard way with SysCacheGetAttr.

src/backend/executor/nodeWindowAgg.c

@@ -171,6 +171,12 @@ typedef struct WindowStatePerAggData
 
 	/* Data local to eval_windowaggregates() */
 	bool		restart;		/* need to restart this agg in this cycle? */
+
+	/* ON EMPTY support */
+	bool		inputReceived;	/* true if any input row was received in
+								 * current frame */
+	/* ExprState for evaluating ON EMPTY default value, or NULL */
+	ExprState  *aggonemptystate;
 } WindowStatePerAggData;
 
 static void initialize_windowaggregate(WindowAggState *winstate,
@@ -278,6 +284,12 @@ initialize_windowaggregate(WindowAggState *winstate,
 	peraggstate->transValueCount = 0;
 	peraggstate->resultValue = (Datum) 0;
 	peraggstate->resultValueIsNull = true;
+
+	/*
+	 * Initialize the flag used for ON EMPTY to track whether any input rows
+	 * were received in this frame.
+	 */
+	peraggstate->inputReceived = false;
 }
 
 /*
@@ -325,6 +337,9 @@ advance_windowaggregate(WindowAggState *winstate,
 		i++;
 	}
 
+	/* Mark that this aggregate received input (used for ON EMPTY) */
+	peraggstate->inputReceived = true;
+
 	if (peraggstate->transfn.fn_strict)
 	{
 		/*
@@ -634,6 +649,25 @@ finalize_windowaggregate(WindowAggState *winstate,
 
 	oldContext = MemoryContextSwitchTo(winstate->ss.ps.ps_ExprContext->ecxt_per_tuple_memory);
 
+	/*
+	 * If the ON EMPTY clause is specified with a default value, evaluate and
+	 * return it in cases where no input rows were received in the current
+	 * frame.
+	 *
+	 * The inputReceived flag is used to detect the empty frame case (zero
+	 * rows processed in this frame).
+	 */
+	if (peraggstate->aggonemptystate != NULL && !peraggstate->inputReceived)
+	{
+		*result = ExecEvalExpr(peraggstate->aggonemptystate,
+							   winstate->ss.ps.ps_ExprContext,
+							   isnull);
+
+		MemoryContextSwitchTo(oldContext);
+
+		return;
+	}
+
 	/*
 	 * Apply the agg's finalfn if one is provided, else return transValue.
 	 */
@@ -3131,6 +3165,13 @@ initialize_peragg(WindowAggState *winstate, WindowFunc *wfunc,
 					&peraggstate->transtypeLen,
 					&peraggstate->transtypeByVal);
 
+	/* Build expression state for ON EMPTY default expression */
+	if (wfunc->aggonempty)
+		peraggstate->aggonemptystate = ExecInitExpr(wfunc->aggonempty,
+													(PlanState *) winstate);
+	else
+		peraggstate->aggonemptystate = NULL;
+
 	/*
 	 * initval is potentially null, so don't try to access it as a struct
 	 * field. Must do it the hard way with SysCacheGetAttr.

src/backend/nodes/makefuncs.c

@@ -681,6 +681,7 @@ makeFuncCall(List *name, List *args, CoercionForm funcformat, int location)
 	n->args = args;
 	n->agg_order = NIL;
 	n->agg_filter = NULL;
+	n->agg_on_empty = NULL;
 	n->over = NULL;
 	n->agg_within_group = false;
 	n->agg_star = false;

src/backend/nodes/nodeFuncs.c

@@ -2168,6 +2168,8 @@ expression_tree_walker_impl(Node *node,
 					return true;
 				if (WALK(expr->aggfilter))
 					return true;
+				if (WALK(expr->aggonempty))
+					return true;
 			}
 			break;
 		case T_GroupingFunc:
@@ -2187,6 +2189,8 @@ expression_tree_walker_impl(Node *node,
 					return true;
 				if (WALK(expr->aggfilter))
 					return true;
+				if (WALK(expr->aggonempty))
+					return true;
 				if (WALK(expr->runCondition))
 					return true;
 			}
@@ -3101,6 +3105,7 @@ expression_tree_mutator_impl(Node *node,
 				MUTATE(newnode->aggorder, aggref->aggorder, List *);
 				MUTATE(newnode->aggdistinct, aggref->aggdistinct, List *);
 				MUTATE(newnode->aggfilter, aggref->aggfilter, Expr *);
+				MUTATE(newnode->aggonempty, aggref->aggonempty, Expr *);
 				return (Node *) newnode;
 			}
 			break;
@@ -3135,6 +3140,7 @@ expression_tree_mutator_impl(Node *node,
 				FLATCOPY(newnode, wfunc, WindowFunc);
 				MUTATE(newnode->args, wfunc->args, List *);
 				MUTATE(newnode->aggfilter, wfunc->aggfilter, Expr *);
+				MUTATE(newnode->aggonempty, wfunc->aggonempty, Expr *);
 				return (Node *) newnode;
 			}
 			break;
@@ -4532,6 +4538,8 @@ raw_expression_tree_walker_impl(Node *node,
 					return true;
 				if (WALK(fcall->agg_filter))
 					return true;
+				if (WALK(fcall->agg_on_empty))
+					return true;
 				if (WALK(fcall->over))
 					return true;
 				/* function name is deemed uninteresting */

src/backend/optimizer/prep/prepagg.c

@@ -215,6 +215,18 @@ preprocess_aggref(Aggref *aggref, PlannerInfo *root)
 
 	ReleaseSysCache(aggTuple);
 
+	/*
+	 * An ON EMPTY default is returned only when the aggregate processes zero
+	 * input rows, which is detected via a per-group "input received" flag set
+	 * as the aggregate's input rows are processed.  Under partial aggregation
+	 * the leader sees only the workers' combined partial states (via the
+	 * combine function), not the original rows, so that flag cannot reliably
+	 * distinguish the empty-input case.  Disable partial aggregation for such
+	 * aggregates.
+	 */
+	if (aggref->aggonempty != NULL)
+		root->hasNonPartialAggs = true;
+
 	/*
 	 * 1. See if this is identical to another aggregate function call that
 	 * we've seen already.