Subject: Re: Deduplicate aggregates and transition functions in planner
To: Andres Freund <andres@anarazel.de>
Cc: pgsql-hackers <pgsql-hackers@postgresql.org>
References: <b2e3536b-1dbc-8303-c97e-89cb0b4a9a48@iki.fi>
 <20201028195901.t55yiwmxe3quguot@alap3.anarazel.de>
From: Heikki Linnakangas <hlinnaka@iki.fi>
Message-ID: <16a1e564-1fa5-b4ea-c512-9ca133ea4bc0@iki.fi>
Date: Thu, 29 Oct 2020 10:17:20 +0200
User-Agent: Mozilla/5.0 (X11; Linux x86_64; rv:78.0) Gecko/20100101
 Thunderbird/78.4.0
MIME-Version: 1.0
In-Reply-To: <20201028195901.t55yiwmxe3quguot@alap3.anarazel.de>
Content-Type: multipart/mixed;
 boundary="------------778645A61BB63B2AE9F12F88"
Content-Language: en-US
Precedence: bulk

This is a multi-part message in MIME format.
--------------778645A61BB63B2AE9F12F88
Content-Type: text/plain; charset=utf-8; format=flowed
Content-Transfer-Encoding: 7bit

On 28/10/2020 21:59, Andres Freund wrote:
> On 2020-10-28 21:10:41 +0200, Heikki Linnakangas wrote:
>> Currently, ExecInitAgg() performs quite a lot of work, to deduplicate
>> identical Aggrefs, as well as Aggrefs that can share the same transition
>> state. That doesn't really belong in the executor, we should perform that
>> work in the planner. It doesn't change from one invocation of the plan to
>> another, and it would be nice to reflect the state-sharing in the plan
>> costs.
> 
> Woo! Very glad to see this tackled.
> 
> It wouldn't surprise me to see a small execution time speedup here -
> I've seen the load of the aggno show up in profiles.

I think you'd be hard-pressed to find a real-life query where it 
matters. But if you don't care about real life:

regression=# do $$
begin
   for i in 1..100000 loop
     perform sum(g), sum(g+0), sum(g+1), sum(g+2), sum(g+3), sum(g+4), 
sum(g+5), sum(g+6), sum(g+7), sum(g+8), sum(g+9), sum(g+10) from 
generate_series(1,1) g;
   end loop;
end;
$$;
DO
Time: 1282.701 ms (00:01.283)

vs.

Time: 860.323 ms

with the patch.

>> @@ -783,14 +783,13 @@ ExecInitExprRec(Expr *node, ExprState *state,
>>   
>>   				scratch.opcode = EEOP_AGGREF;
>>   				scratch.d.aggref.astate = astate;
>> -				astate->aggref = aggref;
>> +				astate->aggno = aggref->aggno;
>>   
>>   				if (state->parent && IsA(state->parent, AggState))
>>   				{
>>   					AggState   *aggstate = (AggState *) state->parent;
>>   
>> -					aggstate->aggs = lappend(aggstate->aggs, astate);
>> -					aggstate->numaggs++;
>> +					aggstate->aggs = lappend(aggstate->aggs, aggref);
> 
> Hm. Why is aggstate->aggs still built during expression initialization?
> Imo that's a pretty huge wart that also introduces more
> order-of-operation brittleness to executor startup.

The Agg node itself doesn't include any information about the aggregates 
and transition functions. Because of that, ExecInitAgg needs a 
"representive" Aggref for each transition state and agg, to initialize 
the per-trans and per-agg structs. The expression initialization makes 
those Aggrefs available for ExecInitAgg.

Instead of collecting all the Aggrefs in a list, ExecInitExprRec() could 
set each representative Aggref directly in the right per-trans and 
per-agg struct, based on the 'aggno' and 'aggtransno' fields. That 
requires initializing the per-trans and per-agg arrays earlier, and for 
that, we would need to store the # of aggs and transition states in the 
Agg node, like you also suggested. Certainly doable, but on the whole, 
it didn't really seem better to me. Attached is a patch, to demonstrate 
what that looks like, on top of the main patch. It's not complete, 
there's at least one case with hash-DISTINCT for queries like "SELECT 
DISTINCT aggregate(x) ..." where the planner creates an Agg for the 
DISTINCT without aggregates, but the code currently passes numAggs=1 to 
the executor. Some further changes would be needed in the planner, to 
mark the AggPath generated for deduplication differently from the 
AggPaths created for aggregation. Again that's doable, but on the whole 
I prefer the approach to scan the Aggrefs in executor startup, for now.

I'd like to get rid of the "representative Aggrefs" altogether, and pass 
information about the transition and final functions from planner to 
executor in some other form. But that's exactly what got me into the 
refactoring that was ballooning out of hand that I mentioned.

- Heikki

--------------778645A61BB63B2AE9F12F88
Content-Type: text/x-patch; charset=UTF-8;
 name="pass-numaggs-from-planner.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
 filename="pass-numaggs-from-planner.patch"

diff --git a/src/backend/executor/execExpr.c b/src/backend/executor/execExpr.c
index 2a4dea2b052..6a03fa730e5 100644
--- a/src/backend/executor/execExpr.c
+++ b/src/backend/executor/execExpr.c
@@ -785,11 +785,23 @@ ExecInitExprRec(Expr *node, ExprState *state,
 				scratch.d.aggref.astate = astate;
 				astate->aggno = aggref->aggno;
 
+				/*
+				 * Remember this Aggref as the representative for the
+				 * aggregate. ExecInitAgg needs a representative for
+				 * initializing the states, and it can also be accessed
+				 * by the user-defined functions by AggGetAggref().
+				 */
 				if (state->parent && IsA(state->parent, AggState))
 				{
 					AggState   *aggstate = (AggState *) state->parent;
+					AggStatePerAgg peragg;
+
+					if (aggref->aggno >= aggstate->numaggs)
+						elog(ERROR, "invalid aggno %d", aggref->aggno);
+					peragg = &aggstate->peragg[aggref->aggno];
 
-					aggstate->aggs = lappend(aggstate->aggs, aggref);
+					if (peragg->aggref == NULL)
+						peragg->aggref = aggref;
 				}
 				else
 				{
diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c
index 7585689b94d..1949729a494 100644
--- a/src/backend/executor/nodeAgg.c
+++ b/src/backend/executor/nodeAgg.c
@@ -3236,13 +3236,11 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 	Plan	   *outerPlan;
 	ExprContext *econtext;
 	TupleDesc	scanDesc;
-	int			max_aggno;
-	int			max_transno;
-	int			numaggrefs;
 	int			numaggs;
 	int			numtrans;
 	int			phase;
 	int			phaseidx;
+	int			aggno;
 	ListCell   *l;
 	Bitmapset  *all_grouped_cols = NULL;
 	int			numGroupingSets = 1;
@@ -3264,7 +3262,6 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 	aggstate->ss.ps.state = estate;
 	aggstate->ss.ps.ExecProcNode = ExecAgg;
 
-	aggstate->aggs = NIL;
 	aggstate->numaggs = 0;
 	aggstate->numtrans = 0;
 	aggstate->aggstrategy = node->aggstrategy;
@@ -3401,6 +3398,17 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 			aggstate->ss.ps.outeropsfixed = false;
 	}
 
+	/* Initialize the per-agg and per-trans arrays before calling ExecInitExpr() */
+	numaggs = node->numAggs;
+	peraggs = (AggStatePerAgg) palloc0(sizeof(AggStatePerAggData) * numaggs);
+	aggstate->peragg = peraggs;
+	aggstate->numaggs = numaggs;
+
+	numtrans = node->numTrans;
+	pertransstates = (AggStatePerTrans) palloc0(sizeof(AggStatePerTransData) * numtrans);
+	aggstate->pertrans = pertransstates;
+	aggstate->numtrans = numtrans;
+
 	/*
 	 * Initialize result type, slot and projection.
 	 */
@@ -3423,22 +3431,6 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 	aggstate->ss.ps.qual =
 		ExecInitQual(node->plan.qual, (PlanState *) aggstate);
 
-	/*
-	 * We should now have found all Aggrefs in the targetlist and quals.
-	 */
-	numaggrefs = list_length(aggstate->aggs);
-	max_aggno = -1;
-	max_transno = -1;
-	foreach(l, aggstate->aggs)
-	{
-		Aggref	   *aggref = (Aggref *) lfirst(l);
-
-		max_aggno = Max(max_aggno, aggref->aggno);
-		max_transno = Max(max_transno, aggref->aggtransno);
-	}
-	numaggs = max_aggno + 1;
-	numtrans = max_transno + 1;
-
 	/*
 	 * For each phase, prepare grouping set data and fmgr lookup data for
 	 * compare functions.  Accumulate all_grouped_cols in passing.
@@ -3607,13 +3599,6 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 	econtext->ecxt_aggvalues = (Datum *) palloc0(sizeof(Datum) * numaggs);
 	econtext->ecxt_aggnulls = (bool *) palloc0(sizeof(bool) * numaggs);
 
-	peraggs = (AggStatePerAgg) palloc0(sizeof(AggStatePerAggData) * numaggs);
-	pertransstates = (AggStatePerTrans) palloc0(sizeof(AggStatePerTransData) * numtrans);
-
-	aggstate->peragg = peraggs;
-	aggstate->pertrans = pertransstates;
-
-
 	aggstate->all_pergroups =
 		(AggStatePerGroup *) palloc0(sizeof(AggStatePerGroup)
 									 * (numGroupingSets + numHashes));
@@ -3699,11 +3684,11 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 	 * Perform lookups of aggregate function info, and initialize the
 	 * unchanging fields of the per-agg and per-trans data.
 	 */
-	foreach(l, aggstate->aggs)
+	for (aggno = 0; aggno < numaggs; aggno++)
 	{
-		Aggref	   *aggref = lfirst(l);
-		AggStatePerAgg peragg;
+		AggStatePerAgg peragg = &peraggs[aggno];
 		AggStatePerTrans pertrans;
+		Aggref	   *aggref;
 		Oid			inputTypes[FUNC_MAX_ARGS];
 		int			numArguments;
 		int			numDirectArgs;
@@ -3717,18 +3702,18 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 		Expr	   *finalfnexpr;
 		Oid			aggtranstype;
 
+		if (peragg->aggref == NULL)
+		{
+			elog(ERROR, "aggregate %d of %d was not found in expression initialization",
+				 aggno, numaggs);
+		}
+		aggref = peragg->aggref;
+
 		/* Planner should have assigned aggregate to correct level */
 		Assert(aggref->agglevelsup == 0);
 		/* ... and the split mode should match */
 		Assert(aggref->aggsplit == aggstate->aggsplit);
 
-		peragg = &peraggs[aggref->aggno];
-
-		/* Check if we initialized the state for this aggregate already. */
-		if (peragg->aggref != NULL)
-			continue;
-
-		peragg->aggref = aggref;
 		peragg->transno = aggref->aggtransno;
 
 		/* Fetch the pg_aggregate row */
@@ -3934,13 +3919,6 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 		ReleaseSysCache(aggTuple);
 	}
 
-	/*
-	 * Update aggstate->numaggs to be the number of unique aggregates found.
-	 * Also set numstates to the number of unique transition states found.
-	 */
-	aggstate->numaggs = numaggs;
-	aggstate->numtrans = numtrans;
-
 	/*
 	 * Last, check whether any more aggregates got added onto the node while
 	 * we processed the expressions for the aggregate arguments (including not
@@ -3951,10 +3929,13 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 	 * need to work hard on a helpful error message; but we defend against it
 	 * here anyway, just to be sure.)
 	 */
+	/* FIXME: need another way to sanity check this now. Not hard. Or just remove this */
+#if 0
 	if (numaggrefs != list_length(aggstate->aggs))
 		ereport(ERROR,
 				(errcode(ERRCODE_GROUPING_ERROR),
 				 errmsg("aggregate function calls cannot be nested")));
+#endif
 
 	/*
 	 * Build expressions doing all the transition work at once. We build a
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index d15866de89f..812735576f2 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -1024,6 +1024,8 @@ _copyAgg(const Agg *from)
 	COPY_BITMAPSET_FIELD(aggParams);
 	COPY_NODE_FIELD(groupingSets);
 	COPY_NODE_FIELD(chain);
+	COPY_SCALAR_FIELD(numAggs);
+	COPY_SCALAR_FIELD(numTrans);
 
 	return newnode;
 }
diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c
index 3c740b805f3..e3c168531f9 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -785,6 +785,8 @@ _outAgg(StringInfo str, const Agg *node)
 	WRITE_BITMAPSET_FIELD(aggParams);
 	WRITE_NODE_FIELD(groupingSets);
 	WRITE_NODE_FIELD(chain);
+	WRITE_INT_FIELD(numAggs);
+	WRITE_INT_FIELD(numTrans);
 }
 
 static void
diff --git a/src/backend/nodes/readfuncs.c b/src/backend/nodes/readfuncs.c
index 169d5581b91..43533e13daa 100644
--- a/src/backend/nodes/readfuncs.c
+++ b/src/backend/nodes/readfuncs.c
@@ -2235,6 +2235,8 @@ _readAgg(void)
 	READ_BITMAPSET_FIELD(aggParams);
 	READ_NODE_FIELD(groupingSets);
 	READ_NODE_FIELD(chain);
+	READ_INT_FIELD(numAggs);
+	READ_INT_FIELD(numTrans);
 
 	READ_DONE();
 }
diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c
index 94280a730c4..b5e88acf055 100644
--- a/src/backend/optimizer/plan/createplan.c
+++ b/src/backend/optimizer/plan/createplan.c
@@ -1657,7 +1657,7 @@ create_unique_plan(PlannerInfo *root, UniquePath *best_path, int flags)
 								 NIL,
 								 NIL,
 								 best_path->path.rows,
-								 0,
+								 0, 0, 0,
 								 subplan);
 	}
 	else
@@ -2139,6 +2139,7 @@ create_agg_plan(PlannerInfo *root, AggPath *best_path)
 					NIL,
 					best_path->numGroups,
 					best_path->transitionSpace,
+					list_length(root->agginfos), list_length(root->aggtransinfos),
 					subplan);
 
 	copy_generic_path_info(&plan->plan, (Path *) best_path);
@@ -2301,6 +2302,8 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
 										 NIL,
 										 rollup->numGroups,
 										 best_path->transitionSpace,
+										 list_length(root->agginfos),
+										 list_length(root->aggtransinfos),
 										 sort_plan);
 
 			/*
@@ -2340,6 +2343,8 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
 						chain,
 						rollup->numGroups,
 						best_path->transitionSpace,
+						list_length(root->agginfos),
+						list_length(root->aggtransinfos),
 						subplan);
 
 		/* Copy cost data from Path to Plan */
@@ -6350,7 +6355,7 @@ make_agg(List *tlist, List *qual,
 		 AggStrategy aggstrategy, AggSplit aggsplit,
 		 int numGroupCols, AttrNumber *grpColIdx, Oid *grpOperators, Oid *grpCollations,
 		 List *groupingSets, List *chain, double dNumGroups,
-		 Size transitionSpace, Plan *lefttree)
+		 Size transitionSpace, int numAggs, int numTrans, Plan *lefttree)
 {
 	Agg		   *node = makeNode(Agg);
 	Plan	   *plan = &node->plan;
@@ -6370,6 +6375,8 @@ make_agg(List *tlist, List *qual,
 	node->aggParams = NULL;		/* SS_finalize_plan() will fill this */
 	node->groupingSets = groupingSets;
 	node->chain = chain;
+	node->numAggs = numAggs;
+	node->numTrans = numTrans;
 
 	plan->qual = qual;
 	plan->targetlist = tlist;
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index fc5698cff20..ce47d7869dc 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -2140,7 +2140,6 @@ typedef struct AggStatePerHashData *AggStatePerHash;
 typedef struct AggState
 {
 	ScanState	ss;				/* its first field is NodeTag */
-	List	   *aggs;			/* all Aggref nodes in targetlist & quals */
 	int			numaggs;		/* length of list (could be zero!) */
 	int			numtrans;		/* number of pertrans items */
 	AggStrategy aggstrategy;	/* strategy mode */
@@ -2153,15 +2152,15 @@ typedef struct AggState
 	ExprContext *hashcontext;	/* econtexts for long-lived data (hashtable) */
 	ExprContext **aggcontexts;	/* econtexts for long-lived data (per GS) */
 	ExprContext *tmpcontext;	/* econtext for input expressions */
-#define FIELDNO_AGGSTATE_CURAGGCONTEXT 14
+#define FIELDNO_AGGSTATE_CURAGGCONTEXT 13
 	ExprContext *curaggcontext; /* currently active aggcontext */
 	AggStatePerAgg curperagg;	/* currently active aggregate, if any */
-#define FIELDNO_AGGSTATE_CURPERTRANS 16
+#define FIELDNO_AGGSTATE_CURPERTRANS 15
 	AggStatePerTrans curpertrans;	/* currently active trans state, if any */
 	bool		input_done;		/* indicates end of input */
 	bool		agg_done;		/* indicates completion of Agg scan */
 	int			projected_set;	/* The last projected grouping set */
-#define FIELDNO_AGGSTATE_CURRENT_SET 20
+#define FIELDNO_AGGSTATE_CURRENT_SET 19
 	int			current_set;	/* The current grouping set being evaluated */
 	Bitmapset  *grouped_cols;	/* grouped cols in current projection */
 	List	   *all_grouped_cols;	/* list of all grouped cols in DESC order */
@@ -2207,7 +2206,7 @@ typedef struct AggState
 										 * per-group pointers */
 
 	/* support for evaluation of agg input expressions: */
-#define FIELDNO_AGGSTATE_ALL_PERGROUPS 53
+#define FIELDNO_AGGSTATE_ALL_PERGROUPS 52
 	AggStatePerGroup *all_pergroups;	/* array of first ->pergroups, than
 										 * ->hash_pergroup */
 	ProjectionInfo *combinedproj;	/* projection machinery */
diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h
index 7e6b10f86b9..f60f301a2b1 100644
--- a/src/include/nodes/plannodes.h
+++ b/src/include/nodes/plannodes.h
@@ -828,6 +828,9 @@ typedef struct Agg
 	/* Note: planner provides numGroups & aggParams only in HASHED/MIXED case */
 	List	   *groupingSets;	/* grouping sets to use */
 	List	   *chain;			/* chained Agg/Sort nodes */
+
+	int			numAggs;
+	int			numTrans;
 } Agg;
 
 /* ----------------
diff --git a/src/include/optimizer/planmain.h b/src/include/optimizer/planmain.h
index f3cefe67b8d..d64f7c1344b 100644
--- a/src/include/optimizer/planmain.h
+++ b/src/include/optimizer/planmain.h
@@ -55,7 +55,7 @@ extern Agg *make_agg(List *tlist, List *qual,
 					 AggStrategy aggstrategy, AggSplit aggsplit,
 					 int numGroupCols, AttrNumber *grpColIdx, Oid *grpOperators, Oid *grpCollations,
 					 List *groupingSets, List *chain, double dNumGroups,
-					 Size transitionSpace, Plan *lefttree);
+					 Size transitionSpace, int numAggs, int numTrans, Plan *lefttree);
 extern Limit *make_limit(Plan *lefttree, Node *limitOffset, Node *limitCount,
 						 LimitOption limitOption, int uniqNumCols,
 						 AttrNumber *uniqColIdx, Oid *uniqOperators,

--------------778645A61BB63B2AE9F12F88--