The commit fixes two issues

FIRST
If for an aggregate function, collection function does not exist, we need to
collect the raw data from the data nodes and apply the transition and
finalization phases of such an aggregate on coordinator itself, i.e. such
an aggregate can not be pushed to the datanode. In PGXC, such aggregates are
indicated by invalid collection function oid in pg_aggregate.
In case of aggregates, array_agg and string_agg, the transition result type is
'internal'. These aggregates use internals structures such as ArrayBuildState
and StringInfo resp. as transition results. The clients (in this case
coordinator) can not handle transition results of 'internal' type. Hence setting
invalid collection function oid for these aggregates.
SECOND
For aggregates which use polymorphic transition types, those polymorphic types
need to be resolved before creating tuple descriptors for the remote query being
pushed to the datanode with these types of aggregates. The polymorphic
transition types are resolved during the execution phase where as the tuple
descriptor is created at planning time. Hence for now, aggregates with
polymorphic transition result types are not pushed to the datanodes.

5 files changed, 28 insertions, 8 deletions

diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c
index 215ed5fcae..bfcda98c53 100644
--- a/src/backend/executor/nodeAgg.c
+++ b/src/backend/executor/nodeAgg.c
@@ -532,6 +532,8 @@ advance_collection_function(AggState *aggstate,
 	Datum		newVal;
 	MemoryContext oldContext;
 
+	Assert(OidIsValid(peraggstate->collectfn.fn_oid));
+
 	/*
 	 * numArgument has to be one, since each datanode is going to send a single
 	 * transition value
diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c
index a96619b112..19c8bd2c00 100644
--- a/src/backend/optimizer/plan/createplan.c
+++ b/src/backend/optimizer/plan/createplan.c
@@ -5452,9 +5452,9 @@ pgxc_process_grouping_targetlist(PlannerInfo *root, List **local_tlist)
 
 	/*
 	 * Walk through the target list and find out whether we can push the
-	 * aggregates and grouping to datanodes. We can do so if the target list
-	 * contains plain aggregates (without any expression involving those) and
-	 * expressions in group by clauses only (last one to make the query legit.
+	 * aggregates and grouping to datanodes. Also while doing so, create the
+	 * targetlist for the query to be shipped to the datanode. Adjust the local
+	 * targetlist accordingly.
 	 */
 	foreach(temp, *local_tlist)
 	{
@@ -5465,7 +5465,24 @@ pgxc_process_grouping_targetlist(PlannerInfo *root, List **local_tlist)
 		if (IsA(expr, Aggref))
 		{
 			Aggref	*aggref = (Aggref *)expr;
-			if (aggref->aggorder || aggref->aggdistinct || aggref->agglevelsup)
+			/*
+			 * If the aggregation needs tuples ordered specifically, or only
+			 * accepts distinct values, we can not aggregate unless we have all
+			 * the qualifying rows. Hence partial aggregation at data nodes can
+			 * give wrong results. Hence we can not such aggregates to the
+			 * datanodes.
+			 * If there is no collection function, we can not combine the
+			 * partial aggregation results from the data nodes, hence can not
+			 * push such aggregate to the data nodes.
+			 * PGXCTODO: If the transition type of the collection is polymorphic we
+			 * need to resolve it first. That tells us the partial aggregation type
+			 * expected from data node.
+			 */
+			if (aggref->aggorder ||
+				aggref->aggdistinct ||
+				aggref->agglevelsup ||
+				!aggref->has_collectfn ||
+				IsPolymorphicType(aggref->aggtrantype))
 			{
 				shippable_remote_tlist = false;
 				break;
diff --git a/src/backend/parser/parse_agg.c b/src/backend/parser/parse_agg.c
index 43a7974cee..4846b09b49 100644
--- a/src/backend/parser/parse_agg.c
+++ b/src/backend/parser/parse_agg.c
@@ -204,6 +204,7 @@ transformAggregateCall(ParseState *pstate, Aggref *agg,
 			 agg->aggfnoid);
 	aggform = (Form_pg_aggregate) GETSTRUCT(aggTuple);
 	agg->aggtrantype = aggform->aggtranstype;
+	agg->has_collectfn = OidIsValid(aggform->aggcollectfn);
 	if (IS_PGXC_DATANODE)
 		agg->aggtype = agg->aggtrantype;
 
diff --git a/src/include/catalog/pg_aggregate.h b/src/include/catalog/pg_aggregate.h
index 5c2103ccc0..f5ad8a9793 100644
--- a/src/include/catalog/pg_aggregate.h
+++ b/src/include/catalog/pg_aggregate.h
@@ -428,8 +428,7 @@ DATA(insert ( 2901 xmlconcat2	  xmlconcat2	  -					0	142		_null_ _null_ ));
 
 /* array */
 #ifdef PGXC
-/* PGXCTODO */
-//DATA(insert ( 2335	array_agg_transfn	array_agg_finalfn		0	2281	_null_ ));
+DATA(insert ( 2335	array_agg_transfn	-	array_agg_finalfn		0	2281	_null_ _null_ ));
 #endif
 #ifdef PGXC
 //DATA(insert ( 2335	array_agg_transfn	array_agg_finalfn		0	2281	_null_ ));
@@ -437,8 +436,8 @@ DATA(insert ( 2901 xmlconcat2	  xmlconcat2	  -					0	142		_null_ _null_ ));
 
 /* text */
 #ifdef PGXC
-//DATA(insert (3537	string_agg_transfn			string_agg_finalfn	0	2281	_null_ ));
-//DATA(insert (3538	string_agg_delim_transfn	string_agg_finalfn	0	2281	_null_ ));
+DATA(insert (3537	string_agg_transfn			-	string_agg_finalfn	0	2281	_null_ _null_ ));
+DATA(insert (3538	string_agg_delim_transfn	-	string_agg_finalfn	0	2281	_null_ _null_ ));
 #endif
 
 /*
diff --git a/src/include/nodes/primnodes.h b/src/include/nodes/primnodes.h
index 8d38c24958..d0b628211a 100644
--- a/src/include/nodes/primnodes.h
+++ b/src/include/nodes/primnodes.h
@@ -230,6 +230,7 @@ typedef struct Aggref
 	Oid			aggtype;		/* type Oid of result of the aggregate */
 #ifdef PGXC
 	Oid			aggtrantype;	/* type Oid of transition results */
+	bool		has_collectfn;	/* is collection function available */
 #endif /* PGXC */
 	List	   *args;			/* arguments and sort expressions */
 	List	   *aggorder;		/* ORDER BY (list of SortGroupClause) */