Subject: RE: Partial aggregates pushdown

From: "[email-masked]" <[email-masked]>
Date: Fri, 31 Mar 2023 05:49:21 +0000

Hi Mr.Momjian

> First, am I correct?
Yes, you are correct. This patch uses new special aggregate functions for partial aggregate
(then we call this partialaggfunc).

> Second, how far away is this from being committable
> and/or what work needs to be done to get it committable, either for PG 16 or 17?
I believe there are three: 1. and 2. are not clear if they are necessary or not; 3. are clearly necessary.
I would like to hear the opinions of the development community on whether or not 1. and 2. need to be addressed.

1. Making partialaggfunc user-defined function
In v17, I make partialaggfuncs as built-in functions.
Because of this approach, v17 changes specification of BKI file and pg_aggregate.
For now, partialaggfuncs are needed by only postgres_fdw which is just an extension of PostgreSQL.
In the future, when revising the specifications for BKI files and pg_aggregate when modifying existing PostgreSQL functions,
It is necessary to align them with this patch's changes.
I am concerned that this may be undesirable.
So I am thinking that v17 should be modified to making partialaggfunc as user defined function.

2. Automation of creating definition of partialaggfuncs
In development of v17, I manually create definition of partialaggfuncs for avg, min, max, sum, count.
I am concerned that this may be undesirable.
So I am thinking that v17 should be modified to automate creating definition of partialaggfuncs
for all built-in aggregate functions.

3. Documentation
I need add explanation of partialaggfunc to documents on postgres_fdw and other places.

Sincerely yours,
Yuuki Fujii

--
Yuuki Fujii
Information Technology R&D Center Mitsubishi Electric Corporation

---

From: Bruce Momjian <[email-masked]>
Date: Thu, 6 Apr 2023 21:59:39 -0400

> (then we call this partialaggfunc).

First, my apologies for not addressing this sooner.  I was so focused on
my own tasks that I didn't realize this very important patch was not
getting attention.  I will try my best to get it into PG 17.

What amazes me is that you didn't need to create _any_ actual aggregate
functions.  Rather, you just needed to hook existing functions into the
aggregate tables for partial FDW execution.

> So I am thinking that v17 should be modified to making partialaggfunc as user defined function.

I think we have three possible cases for aggregates pushdown to FDWs:

1)  Postgres built-in aggregate functions
2)  Postgres user-defined & extension aggregate functions
3)  aggregate functions calls to non-PG FDWs

Your patch handles #1 by checking that the FDW Postgres version is the
same as the calling Postgres version.  However, it doesn't check for
extension versions, and frankly, I don't see how we could implement that
cleanly without significant complexity.

I suggest we remove the version check requirement --- instead just
document that the FDW Postgres version should be the same or newer than
the calling Postgres server --- that way, we can assume that whatever is
in the system catalogs of the caller is in the receiving side.  We
should add a GUC to turn off this optimization for cases where the FDW
Postgres version is older than the caller.  This handles case 1-2.

For case 3, I don't even know how much pushdown those do of _any_
aggregates to non-PG servers, let along parallel FDW ones.  Does anyone
know the details?

> for all built-in aggregate functions.

Are there any other builtin functions that need this?  I think we can
just provide documention for extensions on how to do this.

> I need add explanation of partialaggfunc to documents on postgres_fdw and other places.

I can help with that once we decide on the above.

I think 'partialaggfn' should be named 'aggpartialfn' to match other
columns in pg_aggregate.

I am confused by these changes to pg_aggegate:

+{ aggfnoid => 'sum_p_int8', aggtransfn => 'int8_avg_accum',
+  aggfinalfn => 'int8_avg_serialize', aggcombinefn => 'int8_avg_combine',
+  aggserialfn => 'int8_avg_serialize', aggdeserialfn => 'int8_avg_deserialize',
+  aggtranstype => 'internal', aggtransspace => '48' },

...

+{ aggfnoid => 'sum_p_numeric', aggtransfn => 'numeric_avg_accum',
+  aggfinalfn => 'numeric_avg_serialize', aggcombinefn => 'numeric_avg_combine',
+  aggserialfn => 'numeric_avg_serialize',
+  aggdeserialfn => 'numeric_avg_deserialize',
+  aggtranstype => 'internal', aggtransspace => '128' },

Why are these marked as 'sum' but use 'avg' functions?

It would be good to explain exactly how this is diffent from background
worker parallelism.

-- 
  Bruce Momjian  <[email-masked]>        https://momjian.us
  EDB                                      https://enterprisedb.com

  Embrace your flaws.  They make you human, rather than perfect,
  which you will never be.
diff --git a/contrib/postgres_fdw/deparse.c b/contrib/postgres_fdw/deparse.c
index 09d6dd60dd..fe219b8d2f 100644
--- a/contrib/postgres_fdw/deparse.c
+++ b/contrib/postgres_fdw/deparse.c
@@ -202,7 +202,7 @@ static bool is_subquery_var(Var *node, RelOptInfo *foreignrel,
 							int *relno, int *colno);
 static void get_relation_column_alias_ids(Var *node, RelOptInfo *foreignrel,
 										  int *relno, int *colno);
-
+static bool partial_agg_ok(Aggref* agg, PgFdwRelationInfo* fpinfo);
 
 /*
  * Examine each qual clause in input_conds, and classify them into two groups,
@@ -907,8 +907,9 @@ foreign_expr_walker(Node *node,
 				if (!IS_UPPER_REL(glob_cxt->foreignrel))
 					return false;
 
-				/* Only non-split aggregates are pushable. */
-				if (agg->aggsplit != AGGSPLIT_SIMPLE)
+				if ((agg->aggsplit != AGGSPLIT_SIMPLE) && (agg->aggsplit != AGGSPLIT_INITIAL_SERIAL))
+					return false;
+				if (agg->aggsplit == AGGSPLIT_INITIAL_SERIAL && !partial_agg_ok(agg, fpinfo))
 					return false;
 
 				/* As usual, it must be shippable. */
@@ -3517,14 +3518,37 @@ deparseAggref(Aggref *node, deparse_expr_cxt *context)
 	StringInfo	buf = context->buf;
 	bool		use_variadic;
 
-	/* Only basic, non-split aggregation accepted. */
-	Assert(node->aggsplit == AGGSPLIT_SIMPLE);
+
+	Assert((node->aggsplit == AGGSPLIT_SIMPLE) ||
+		(node->aggsplit == AGGSPLIT_INITIAL_SERIAL));
 
 	/* Check if need to print VARIADIC (cf. ruleutils.c) */
 	use_variadic = node->aggvariadic;
 
-	/* Find aggregate name from aggfnoid which is a pg_proc entry */
-	appendFunctionName(node->aggfnoid, context);
+	if (node->aggsplit == AGGSPLIT_SIMPLE) {
+		/* Find aggregate name from aggfnoid which is a pg_proc entry */
+		appendFunctionName(node->aggfnoid, context);
+	}
+	else {
+		HeapTuple			aggtup;
+		Form_pg_aggregate	aggform;
+
+		/* Find aggregate name from aggfnoid or partialaggfn which is a pg_proc entry */
+		aggtup = SearchSysCache1(AGGFNOID, ObjectIdGetDatum(node->aggfnoid));
+		if (!HeapTupleIsValid(aggtup))
+			elog(ERROR, "cache lookup failed for aggregate %u", node->aggfnoid);
+		aggform = (Form_pg_aggregate)GETSTRUCT(aggtup);
+
+		if ((aggform->aggtranstype != INTERNALOID) && (aggform->aggfinalfn == InvalidOid)) {
+			appendFunctionName(node->aggfnoid, context);
+		} else if(aggform->partialaggfn) {
+			appendFunctionName(aggform->partialaggfn, context);
+		} else {
+			elog(ERROR, "there is no partialaggfn %u", node->aggfnoid);
+		}
+		ReleaseSysCache(aggtup);
+	}
+
 	appendStringInfoChar(buf, '(');
 
 	/* Add DISTINCT */
@@ -4047,3 +4071,62 @@ get_relation_column_alias_ids(Var *node, RelOptInfo *foreignrel,
 	/* Shouldn't get here */
 	elog(ERROR, "unexpected expression in subquery output");
 }
+
+/*
+ * Check that partial aggregate function, described by aggform,
+ * exists on remote server, described by fpinfo.
+ */
+static bool
+partial_agg_compatible(Form_pg_aggregate aggform, PgFdwRelationInfo *fpinfo)
+{
+	int32  partialagg_minversion = PG_VERSION_NUM;
+	if (aggform->partialagg_minversion != PARTIALAGG_MINVERSION_DEFAULT) {
+		partialagg_minversion = aggform->partialagg_minversion;
+	}
+	return (fpinfo->server_version >= partialagg_minversion);
+}
+
+/*
+ * Check that partial aggregate agg is fine to push down.
+ *
+ * It is fine when all of the following conditions are true.
+ * condition1) agg is AGGKIND_NORMAL aggregate which contains no distinct
+ *   or order by clauses
+ * condition2) there is an aggregate function for partial aggregation
+ *   (then we call this partialaggfunc) corresponding to agg.
+ *   condition2 is true when either of the following cases is true.
+ *   case2-1) return type of agg is not internal and agg has no finalfunc.
+ *     In this case, partialaggfunc is agg itself.
+ *   case2-2) agg has valid partialaggfn and partialagg_minversion <= server_version.
+ *     In this case, partialaggfunc is a aggregate function whose oid is partialaggfn.
+ */
+static bool
+partial_agg_ok(Aggref* agg, PgFdwRelationInfo *fpinfo)
+{
+	HeapTuple	aggtup;
+	Form_pg_aggregate aggform;
+	bool        partial_agg_ok = true;
+
+	Assert(agg->aggsplit == AGGSPLIT_INITIAL_SERIAL);
+
+	/* We don't support complex partial aggregates */
+	if (agg->aggdistinct || agg->aggkind != AGGKIND_NORMAL || agg->aggorder != NIL)
+		return false;
+
+	aggtup = SearchSysCache1(AGGFNOID, ObjectIdGetDatum(agg->aggfnoid));
+	if (!HeapTupleIsValid(aggtup))
+		elog(ERROR, "cache lookup failed for aggregate %u", agg->aggfnoid);
+	aggform = (Form_pg_aggregate)GETSTRUCT(aggtup);
+
+	if ((aggform->aggtranstype == INTERNALOID) || (aggform->aggfinalfn != InvalidOid)) {
+		/* Only aggregates which has partialaggfn, are allowed */
+		if (!aggform->partialaggfn) {
+			partial_agg_ok = false;
+		} else if (!partial_agg_compatible(aggform, fpinfo)) {
+			partial_agg_ok = false;
+		}
+	}
+
+	ReleaseSysCache(aggtup);
+	return partial_agg_ok;
+}
diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out
index 8f6a04f71b..ebdb7cff3b 100644
--- a/contrib/postgres_fdw/expected/postgres_fdw.out
+++ b/contrib/postgres_fdw/expected/postgres_fdw.out
@@ -9677,13 +9677,34 @@ RESET enable_partitionwise_join;
 -- ===================================================================
 -- test partitionwise aggregates
 -- ===================================================================
-CREATE TABLE pagg_tab (a int, b int, c text) PARTITION BY RANGE(a);
+CREATE OR REPLACE FUNCTION f_server_version_lag(version_offset int4) RETURNS text AS $$
+    SELECT (setting::int4 + version_offset)::text FROM pg_settings WHERE name = 'server_version_num';
+$$ LANGUAGE sql;
+CREATE OR REPLACE FUNCTION f_alter_server_version(server_name text, operation text, version_offset int4) RETURNS void AS $$
+    DECLARE
+        version_num text;
+    BEGIN
+        EXECUTE 'SELECT f_server_version_lag($1) '
+            INTO version_num
+            USING version_offset;
+        EXECUTE format('ALTER SERVER %I OPTIONS (%I server_version %L)',
+            server_name, operation, version_num);
+        RETURN;
+    END;
+$$ LANGUAGE plpgsql;
+SELECT f_alter_server_version('loopback', 'add', 0);
+ f_alter_server_version 
+------------------------
+ 
+(1 row)
+
+CREATE TABLE pagg_tab (a int, b int, c text, d int4) PARTITION BY RANGE(a);
 CREATE TABLE pagg_tab_p1 (LIKE pagg_tab);
 CREATE TABLE pagg_tab_p2 (LIKE pagg_tab);
 CREATE TABLE pagg_tab_p3 (LIKE pagg_tab);
-INSERT INTO pagg_tab_p1 SELECT i % 30, i % 50, to_char(i/30, 'FM0000') FROM generate_series(1, 3000) i WHERE (i % 30) < 10;
-INSERT INTO pagg_tab_p2 SELECT i % 30, i % 50, to_char(i/30, 'FM0000') FROM generate_series(1, 3000) i WHERE (i % 30) < 20 and (i % 30) >= 10;
-INSERT INTO pagg_tab_p3 SELECT i % 30, i % 50, to_char(i/30, 'FM0000') FROM generate_series(1, 3000) i WHERE (i % 30) < 30 and (i % 30) >= 20;
+INSERT INTO pagg_tab_p1 SELECT i % 30, i % 50, to_char(i/30, 'FM0000'), i % 40 FROM generate_series(1, 3000) i WHERE (i % 30) < 10;
+INSERT INTO pagg_tab_p2 SELECT i % 30, i % 50, to_char(i/30, 'FM0000'), i % 40 FROM generate_series(1, 3000) i WHERE (i % 30) < 20 and (i % 30) >= 10;
+INSERT INTO pagg_tab_p3 SELECT i % 30, i % 50, to_char(i/30, 'FM0000'), i % 40 FROM generate_series(1, 3000) i WHERE (i % 30) < 30 and (i % 30) >= 20;
 -- Create foreign partitions
 CREATE FOREIGN TABLE fpagg_tab_p1 PARTITION OF pagg_tab FOR VALUES FROM (0) TO (10) SERVER loopback OPTIONS (table_name 'pagg_tab_p1');
 CREATE FOREIGN TABLE fpagg_tab_p2 PARTITION OF pagg_tab FOR VALUES FROM (10) TO (20) SERVER loopback OPTIONS (table_name 'pagg_tab_p2');
@@ -9742,8 +9763,8 @@ SELECT a, sum(b), min(b), count(*) FROM pagg_tab GROUP BY a HAVING avg(b) < 22 O
 -- Should have all the columns in the target list for the given relation
 EXPLAIN (VERBOSE, COSTS OFF)
 SELECT a, count(t1) FROM pagg_tab t1 GROUP BY a HAVING avg(b) < 22 ORDER BY 1;
-                               QUERY PLAN                               
-------------------------------------------------------------------------
+                                QUERY PLAN                                 
+---------------------------------------------------------------------------
  Sort
    Output: t1.a, (count(((t1.*)::pagg_tab)))
    Sort Key: t1.a
@@ -9754,21 +9775,21 @@ SELECT a, count(t1) FROM pagg_tab t1 GROUP BY a HAVING avg(b) < 22 ORDER BY 1;
                Filter: (avg(t1.b) < '22'::numeric)
                ->  Foreign Scan on public.fpagg_tab_p1 t1
                      Output: t1.a, t1.*, t1.b
-                     Remote SQL: SELECT a, b, c FROM public.pagg_tab_p1
+                     Remote SQL: SELECT a, b, c, d FROM public.pagg_tab_p1
          ->  HashAggregate
                Output: t1_1.a, count(((t1_1.*)::pagg_tab))
                Group Key: t1_1.a
                Filter: (avg(t1_1.b) < '22'::numeric)
                ->  Foreign Scan on public.fpagg_tab_p2 t1_1
                      Output: t1_1.a, t1_1.*, t1_1.b
-                     Remote SQL: SELECT a, b, c FROM public.pagg_tab_p2
+                     Remote SQL: SELECT a, b, c, d FROM public.pagg_tab_p2
          ->  HashAggregate
                Output: t1_2.a, count(((t1_2.*)::pagg_tab))
                Group Key: t1_2.a
                Filter: (avg(t1_2.b) < '22'::numeric)
                ->  Foreign Scan on public.fpagg_tab_p3 t1_2
                      Output: t1_2.a, t1_2.*, t1_2.b
-                     Remote SQL: SELECT a, b, c FROM public.pagg_tab_p3
+                     Remote SQL: SELECT a, b, c, d FROM public.pagg_tab_p3
 (25 rows)
 
 SELECT a, count(t1) FROM pagg_tab t1 GROUP BY a HAVING avg(b) < 22 ORDER BY 1;
@@ -9804,6 +9825,263 @@ SELECT b, avg(a), max(a), count(*) FROM pagg_tab GROUP BY b HAVING sum(a) < 700
                      ->  Foreign Scan on fpagg_tab_p3 pagg_tab_2
 (15 rows)
 
+-- It's unsafe to push down having clause when there are partial aggregates
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT b, max(a), count(*) FROM pagg_tab GROUP BY b HAVING sum(a) < 700 ORDER BY 1;
+                                                    QUERY PLAN                                                    
+------------------------------------------------------------------------------------------------------------------
+ Sort
+   Output: pagg_tab.b, (max(pagg_tab.a)), (count(*))
+   Sort Key: pagg_tab.b
+   ->  Finalize HashAggregate
+         Output: pagg_tab.b, max(pagg_tab.a), count(*)
+         Group Key: pagg_tab.b
+         Filter: (sum(pagg_tab.a) < 700)
+         ->  Append
+               ->  Partial HashAggregate
+                     Output: pagg_tab.b, PARTIAL max(pagg_tab.a), PARTIAL count(*), PARTIAL sum(pagg_tab.a)
+                     Group Key: pagg_tab.b
+                     ->  Foreign Scan on public.fpagg_tab_p1 pagg_tab
+                           Output: pagg_tab.b, pagg_tab.a
+                           Remote SQL: SELECT a, b FROM public.pagg_tab_p1
+               ->  Partial HashAggregate
+                     Output: pagg_tab_1.b, PARTIAL max(pagg_tab_1.a), PARTIAL count(*), PARTIAL sum(pagg_tab_1.a)
+                     Group Key: pagg_tab_1.b
+                     ->  Foreign Scan on public.fpagg_tab_p2 pagg_tab_1
+                           Output: pagg_tab_1.b, pagg_tab_1.a
+                           Remote SQL: SELECT a, b FROM public.pagg_tab_p2
+               ->  Partial HashAggregate
+                     Output: pagg_tab_2.b, PARTIAL max(pagg_tab_2.a), PARTIAL count(*), PARTIAL sum(pagg_tab_2.a)
+                     Group Key: pagg_tab_2.b
+                     ->  Foreign Scan on public.fpagg_tab_p3 pagg_tab_2
+                           Output: pagg_tab_2.b, pagg_tab_2.a
+                           Remote SQL: SELECT a, b FROM public.pagg_tab_p3
+(26 rows)
+
+-- Partial aggregates are fine to push down without having clause
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT b, min(a), max(a), count(*), sum(d), sum(d::int8), avg(d), avg(d::int8) FROM pagg_tab GROUP BY b ORDER BY 1;
+                                                                                                                           QUERY PLAN                                                                                                                           
+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ Sort
+   Output: pagg_tab.b, (min(pagg_tab.a)), (max(pagg_tab.a)), (count(*)), (sum(pagg_tab.d)), (sum((pagg_tab.d)::bigint)), (avg(pagg_tab.d)), (avg((pagg_tab.d)::bigint))
+   Sort Key: pagg_tab.b
+   ->  Finalize HashAggregate
+         Output: pagg_tab.b, min(pagg_tab.a), max(pagg_tab.a), count(*), sum(pagg_tab.d), sum((pagg_tab.d)::bigint), avg(pagg_tab.d), avg((pagg_tab.d)::bigint)
+         Group Key: pagg_tab.b
+         ->  Append
+               ->  Foreign Scan
+                     Output: pagg_tab.b, (PARTIAL min(pagg_tab.a)), (PARTIAL max(pagg_tab.a)), (PARTIAL count(*)), (PARTIAL sum(pagg_tab.d)), (PARTIAL sum((pagg_tab.d)::bigint)), (PARTIAL avg(pagg_tab.d)), (PARTIAL avg((pagg_tab.d)::bigint))
+                     Relations: Aggregate on (public.fpagg_tab_p1 pagg_tab)
+                     Remote SQL: SELECT b, min(a), max(a), count(*), sum(d), sum_p_int8(d::bigint), avg_p_int4(d), avg_p_int8(d::bigint) FROM public.pagg_tab_p1 GROUP BY 1
+               ->  Foreign Scan
+                     Output: pagg_tab_1.b, (PARTIAL min(pagg_tab_1.a)), (PARTIAL max(pagg_tab_1.a)), (PARTIAL count(*)), (PARTIAL sum(pagg_tab_1.d)), (PARTIAL sum((pagg_tab_1.d)::bigint)), (PARTIAL avg(pagg_tab_1.d)), (PARTIAL avg((pagg_tab_1.d)::bigint))
+                     Relations: Aggregate on (public.fpagg_tab_p2 pagg_tab_1)
+                     Remote SQL: SELECT b, min(a), max(a), count(*), sum(d), sum_p_int8(d::bigint), avg_p_int4(d), avg_p_int8(d::bigint) FROM public.pagg_tab_p2 GROUP BY 1
+               ->  Foreign Scan
+                     Output: pagg_tab_2.b, (PARTIAL min(pagg_tab_2.a)), (PARTIAL max(pagg_tab_2.a)), (PARTIAL count(*)), (PARTIAL sum(pagg_tab_2.d)), (PARTIAL sum((pagg_tab_2.d)::bigint)), (PARTIAL avg(pagg_tab_2.d)), (PARTIAL avg((pagg_tab_2.d)::bigint))
+                     Relations: Aggregate on (public.fpagg_tab_p3 pagg_tab_2)
+                     Remote SQL: SELECT b, min(a), max(a), count(*), sum(d), sum_p_int8(d::bigint), avg_p_int4(d), avg_p_int8(d::bigint) FROM public.pagg_tab_p3 GROUP BY 1
+(19 rows)
+
+SELECT b, min(a), max(a), count(*), sum(d), sum(d::int8), avg(d), avg(d::int8) FROM pagg_tab GROUP BY b ORDER BY 1;
+ b  | min | max | count | sum  | sum  |         avg         |         avg         
+----+-----+-----+-------+------+------+---------------------+---------------------
+  0 |   0 |  20 |    60 |  900 |  900 | 15.0000000000000000 | 15.0000000000000000
+  1 |   1 |  21 |    60 |  960 |  960 | 16.0000000000000000 | 16.0000000000000000
+  2 |   2 |  22 |    60 | 1020 | 1020 | 17.0000000000000000 | 17.0000000000000000
+  3 |   3 |  23 |    60 | 1080 | 1080 | 18.0000000000000000 | 18.0000000000000000
+  4 |   4 |  24 |    60 | 1140 | 1140 | 19.0000000000000000 | 19.0000000000000000
+  5 |   5 |  25 |    60 | 1200 | 1200 | 20.0000000000000000 | 20.0000000000000000
+  6 |   6 |  26 |    60 | 1260 | 1260 | 21.0000000000000000 | 21.0000000000000000
+  7 |   7 |  27 |    60 | 1320 | 1320 | 22.0000000000000000 | 22.0000000000000000
+  8 |   8 |  28 |    60 | 1380 | 1380 | 23.0000000000000000 | 23.0000000000000000
+  9 |   9 |  29 |    60 | 1440 | 1440 | 24.0000000000000000 | 24.0000000000000000
+ 10 |   0 |  20 |    60 |  900 |  900 | 15.0000000000000000 | 15.0000000000000000
+ 11 |   1 |  21 |    60 |  960 |  960 | 16.0000000000000000 | 16.0000000000000000
+ 12 |   2 |  22 |    60 | 1020 | 1020 | 17.0000000000000000 | 17.0000000000000000
+ 13 |   3 |  23 |    60 | 1080 | 1080 | 18.0000000000000000 | 18.0000000000000000
+ 14 |   4 |  24 |    60 | 1140 | 1140 | 19.0000000000000000 | 19.0000000000000000
+ 15 |   5 |  25 |    60 | 1200 | 1200 | 20.0000000000000000 | 20.0000000000000000
+ 16 |   6 |  26 |    60 | 1260 | 1260 | 21.0000000000000000 | 21.0000000000000000
+ 17 |   7 |  27 |    60 | 1320 | 1320 | 22.0000000000000000 | 22.0000000000000000
+ 18 |   8 |  28 |    60 | 1380 | 1380 | 23.0000000000000000 | 23.0000000000000000
+ 19 |   9 |  29 |    60 | 1440 | 1440 | 24.0000000000000000 | 24.0000000000000000
+ 20 |   0 |  20 |    60 |  900 |  900 | 15.0000000000000000 | 15.0000000000000000
+ 21 |   1 |  21 |    60 |  960 |  960 | 16.0000000000000000 | 16.0000000000000000
+ 22 |   2 |  22 |    60 | 1020 | 1020 | 17.0000000000000000 | 17.0000000000000000
+ 23 |   3 |  23 |    60 | 1080 | 1080 | 18.0000000000000000 | 18.0000000000000000
+ 24 |   4 |  24 |    60 | 1140 | 1140 | 19.0000000000000000 | 19.0000000000000000
+ 25 |   5 |  25 |    60 | 1200 | 1200 | 20.0000000000000000 | 20.0000000000000000
+ 26 |   6 |  26 |    60 | 1260 | 1260 | 21.0000000000000000 | 21.0000000000000000
+ 27 |   7 |  27 |    60 | 1320 | 1320 | 22.0000000000000000 | 22.0000000000000000
+ 28 |   8 |  28 |    60 | 1380 | 1380 | 23.0000000000000000 | 23.0000000000000000
+ 29 |   9 |  29 |    60 | 1440 | 1440 | 24.0000000000000000 | 24.0000000000000000
+ 30 |   0 |  20 |    60 |  900 |  900 | 15.0000000000000000 | 15.0000000000000000
+ 31 |   1 |  21 |    60 |  960 |  960 | 16.0000000000000000 | 16.0000000000000000
+ 32 |   2 |  22 |    60 | 1020 | 1020 | 17.0000000000000000 | 17.0000000000000000
+ 33 |   3 |  23 |    60 | 1080 | 1080 | 18.0000000000000000 | 18.0000000000000000
+ 34 |   4 |  24 |    60 | 1140 | 1140 | 19.0000000000000000 | 19.0000000000000000
+ 35 |   5 |  25 |    60 | 1200 | 1200 | 20.0000000000000000 | 20.0000000000000000
+ 36 |   6 |  26 |    60 | 1260 | 1260 | 21.0000000000000000 | 21.0000000000000000
+ 37 |   7 |  27 |    60 | 1320 | 1320 | 22.0000000000000000 | 22.0000000000000000
+ 38 |   8 |  28 |    60 | 1380 | 1380 | 23.0000000000000000 | 23.0000000000000000
+ 39 |   9 |  29 |    60 | 1440 | 1440 | 24.0000000000000000 | 24.0000000000000000
+ 40 |   0 |  20 |    60 |  900 |  900 | 15.0000000000000000 | 15.0000000000000000
+ 41 |   1 |  21 |    60 |  960 |  960 | 16.0000000000000000 | 16.0000000000000000
+ 42 |   2 |  22 |    60 | 1020 | 1020 | 17.0000000000000000 | 17.0000000000000000
+ 43 |   3 |  23 |    60 | 1080 | 1080 | 18.0000000000000000 | 18.0000000000000000
+ 44 |   4 |  24 |    60 | 1140 | 1140 | 19.0000000000000000 | 19.0000000000000000
+ 45 |   5 |  25 |    60 | 1200 | 1200 | 20.0000000000000000 | 20.0000000000000000
+ 46 |   6 |  26 |    60 | 1260 | 1260 | 21.0000000000000000 | 21.0000000000000000
+ 47 |   7 |  27 |    60 | 1320 | 1320 | 22.0000000000000000 | 22.0000000000000000
+ 48 |   8 |  28 |    60 | 1380 | 1380 | 23.0000000000000000 | 23.0000000000000000
+ 49 |   9 |  29 |    60 | 1440 | 1440 | 24.0000000000000000 | 24.0000000000000000
+(50 rows)
+
+-- Partial aggregates are fine to push down
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT min(a), max(a), count(*), sum(d), sum(d::int8), avg(d), avg(d::int8) FROM pagg_tab;
+                                                                                                                 QUERY PLAN                                                                                                                 
+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ Finalize Aggregate
+   Output: min(pagg_tab.a), max(pagg_tab.a), count(*), sum(pagg_tab.d), sum((pagg_tab.d)::bigint), avg(pagg_tab.d), avg((pagg_tab.d)::bigint)
+   ->  Append
+         ->  Foreign Scan
+               Output: (PARTIAL min(pagg_tab.a)), (PARTIAL max(pagg_tab.a)), (PARTIAL count(*)), (PARTIAL sum(pagg_tab.d)), (PARTIAL sum((pagg_tab.d)::bigint)), (PARTIAL avg(pagg_tab.d)), (PARTIAL avg((pagg_tab.d)::bigint))
+               Relations: Aggregate on (public.fpagg_tab_p1 pagg_tab)
+               Remote SQL: SELECT min(a), max(a), count(*), sum(d), sum_p_int8(d::bigint), avg_p_int4(d), avg_p_int8(d::bigint) FROM public.pagg_tab_p1
+         ->  Foreign Scan
+               Output: (PARTIAL min(pagg_tab_1.a)), (PARTIAL max(pagg_tab_1.a)), (PARTIAL count(*)), (PARTIAL sum(pagg_tab_1.d)), (PARTIAL sum((pagg_tab_1.d)::bigint)), (PARTIAL avg(pagg_tab_1.d)), (PARTIAL avg((pagg_tab_1.d)::bigint))
+               Relations: Aggregate on (public.fpagg_tab_p2 pagg_tab_1)
+               Remote SQL: SELECT min(a), max(a), count(*), sum(d), sum_p_int8(d::bigint), avg_p_int4(d), avg_p_int8(d::bigint) FROM public.pagg_tab_p2
+         ->  Foreign Scan
+               Output: (PARTIAL min(pagg_tab_2.a)), (PARTIAL max(pagg_tab_2.a)), (PARTIAL count(*)), (PARTIAL sum(pagg_tab_2.d)), (PARTIAL sum((pagg_tab_2.d)::bigint)), (PARTIAL avg(pagg_tab_2.d)), (PARTIAL avg((pagg_tab_2.d)::bigint))
+               Relations: Aggregate on (public.fpagg_tab_p3 pagg_tab_2)
+               Remote SQL: SELECT min(a), max(a), count(*), sum(d), sum_p_int8(d::bigint), avg_p_int4(d), avg_p_int8(d::bigint) FROM public.pagg_tab_p3
+(15 rows)
+
+SELECT min(a), max(a), count(*), sum(d), sum(d::int8), avg(d), avg(d::int8) FROM pagg_tab;
+ min | max | count |  sum  |  sum  |         avg         |         avg         
+-----+-----+-------+-------+-------+---------------------+---------------------
+   0 |  29 |  3000 | 58500 | 58500 | 19.5000000000000000 | 19.5000000000000000
+(1 row)
+
+-- It's unsafe to push down partial aggregates which contains distinct clause
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT max(a), count(distinct b) FROM pagg_tab;
+                                       QUERY PLAN                                        
+-----------------------------------------------------------------------------------------
+ Aggregate
+   Output: max(pagg_tab.a), count(DISTINCT pagg_tab.b)
+   ->  Merge Append
+         Sort Key: pagg_tab.b
+         ->  Foreign Scan on public.fpagg_tab_p1 pagg_tab_1
+               Output: pagg_tab_1.a, pagg_tab_1.b
+               Remote SQL: SELECT a, b FROM public.pagg_tab_p1 ORDER BY b ASC NULLS LAST
+         ->  Foreign Scan on public.fpagg_tab_p2 pagg_tab_2
+               Output: pagg_tab_2.a, pagg_tab_2.b
+               Remote SQL: SELECT a, b FROM public.pagg_tab_p2 ORDER BY b ASC NULLS LAST
+         ->  Foreign Scan on public.fpagg_tab_p3 pagg_tab_3
+               Output: pagg_tab_3.a, pagg_tab_3.b
+               Remote SQL: SELECT a, b FROM public.pagg_tab_p3 ORDER BY b ASC NULLS LAST
+(13 rows)
+
+SELECT max(a), count(distinct b) FROM pagg_tab;
+ max | count 
+-----+-------
+  29 |    50
+(1 row)
+
+-- It's unsafe to push down partial aggregates when
+-- server_version is older than partialagg_minversion, described by pg_aggregate
+SELECT f_alter_server_version('loopback', 'set', -1);
+ f_alter_server_version 
+------------------------
+ 
+(1 row)
+
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT avg(d) FROM pagg_tab;
+                            QUERY PLAN                            
+------------------------------------------------------------------
+ Finalize Aggregate
+   Output: avg(pagg_tab.d)
+   ->  Append
+         ->  Partial Aggregate
+               Output: PARTIAL avg(pagg_tab.d)
+               ->  Foreign Scan on public.fpagg_tab_p1 pagg_tab
+                     Output: pagg_tab.d
+                     Remote SQL: SELECT d FROM public.pagg_tab_p1
+         ->  Partial Aggregate
+               Output: PARTIAL avg(pagg_tab_1.d)
+               ->  Foreign Scan on public.fpagg_tab_p2 pagg_tab_1
+                     Output: pagg_tab_1.d
+                     Remote SQL: SELECT d FROM public.pagg_tab_p2
+         ->  Partial Aggregate
+               Output: PARTIAL avg(pagg_tab_2.d)
+               ->  Foreign Scan on public.fpagg_tab_p3 pagg_tab_2
+                     Output: pagg_tab_2.d
+                     Remote SQL: SELECT d FROM public.pagg_tab_p3
+(18 rows)
+
+SELECT avg(d) FROM pagg_tab;
+         avg         
+---------------------
+ 19.5000000000000000
+(1 row)
+
+-- It's safe to push down partial aggregate for user defined aggregate
+-- which has valid options.
+CREATE AGGREGATE postgres_fdw_sum_p_int8(int8) (
+	SFUNC = int8_avg_accum,
+	STYPE = internal,
+	FINALFUNC = int8_avg_serialize
+);
+CREATE AGGREGATE postgres_fdw_sum(int8) (
+	SFUNC = int8_avg_accum,
+	STYPE = internal,
+	COMBINEFUNC = int8_avg_combine,
+	FINALFUNC = numeric_poly_sum,
+	SERIALFUNC = int8_avg_serialize,
+	DESERIALFUNC = int8_avg_deserialize,
+	PARTIALAGGFUNC = postgres_fdw_sum_p_int8,
+	PARTIALAGG_MINVERSION = 150000
+);
+ALTER EXTENSION postgres_fdw ADD FUNCTION postgres_fdw_sum(int8);
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT postgres_fdw_sum(d::int8) FROM pagg_tab;
+                                             QUERY PLAN                                             
+----------------------------------------------------------------------------------------------------
+ Finalize Aggregate
+   Output: postgres_fdw_sum((pagg_tab.d)::bigint)
+   ->  Append
+         ->  Foreign Scan
+               Output: (PARTIAL postgres_fdw_sum((pagg_tab.d)::bigint))
+               Relations: Aggregate on (public.fpagg_tab_p1 pagg_tab)
+               Remote SQL: SELECT public.postgres_fdw_sum_p_int8(d::bigint) FROM public.pagg_tab_p1
+         ->  Foreign Scan
+               Output: (PARTIAL postgres_fdw_sum((pagg_tab_1.d)::bigint))
+               Relations: Aggregate on (public.fpagg_tab_p2 pagg_tab_1)
+               Remote SQL: SELECT public.postgres_fdw_sum_p_int8(d::bigint) FROM public.pagg_tab_p2
+         ->  Foreign Scan
+               Output: (PARTIAL postgres_fdw_sum((pagg_tab_2.d)::bigint))
+               Relations: Aggregate on (public.fpagg_tab_p3 pagg_tab_2)
+               Remote SQL: SELECT public.postgres_fdw_sum_p_int8(d::bigint) FROM public.pagg_tab_p3
+(15 rows)
+
+SELECT postgres_fdw_sum(d::int8) FROM pagg_tab;
+ postgres_fdw_sum 
+------------------
+            58500
+(1 row)
+
+ALTER EXTENSION postgres_fdw DROP FUNCTION postgres_fdw_sum(int8);
+DROP AGGREGATE postgres_fdw_sum(int8);
+DROP AGGREGATE postgres_fdw_sum_p_int8(int8);
+DROP FUNCTION f_server_version_lag(int4);
+DROP FUNCTION f_alter_server_version(text, text, int4);
+ALTER SERVER loopback OPTIONS (DROP server_version);
 -- ===================================================================
 -- access rights and superuser
 -- ===================================================================
diff --git a/contrib/postgres_fdw/option.c b/contrib/postgres_fdw/option.c
index 4229d2048c..75283fa2d8 100644
--- a/contrib/postgres_fdw/option.c
+++ b/contrib/postgres_fdw/option.c
@@ -211,6 +211,27 @@ postgres_fdw_validator(PG_FUNCTION_ARGS)
 						 errmsg("sslcert and sslkey are superuser-only"),
 						 errhint("User mappings with the sslcert or sslkey options set may only be created or modified by the superuser.")));
 		}
+		else if (strcmp(def->defname, "server_version") == 0)
+		{
+			char* value;
+			int			int_val;
+			bool		is_parsed;
+
+			value = defGetString(def);
+			is_parsed = parse_int(value, &int_val, 0, NULL);
+
+			if (!is_parsed)
+				ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+						errmsg("invalid value for integer option \"%s\": %s",
+							def->defname, value)));
+
+			if (int_val <= 0)
+				ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+						errmsg("\"%s\" must be an integer value greater than zero",
+							def->defname)));
+		}
 		else if (strcmp(def->defname, "analyze_sampling") == 0)
 		{
 			char	   *value;
@@ -288,6 +309,9 @@ InitPgFdwOptions(void)
 		{"sslcert", UserMappingRelationId, true},
 		{"sslkey", UserMappingRelationId, true},
 
+		/* options for partial aggregation pushdown */
+		{"server_version", ForeignServerRelationId, false},
+
 		{NULL, InvalidOid, false}
 	};
 
diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c
index d728bd70b3..7b51acadf3 100644
--- a/contrib/postgres_fdw/postgres_fdw.c
+++ b/contrib/postgres_fdw/postgres_fdw.c
@@ -519,7 +519,7 @@ static bool foreign_join_ok(PlannerInfo *root, RelOptInfo *joinrel,
 							JoinType jointype, RelOptInfo *outerrel, RelOptInfo *innerrel,
 							JoinPathExtraData *extra);
 static bool foreign_grouping_ok(PlannerInfo *root, RelOptInfo *grouped_rel,
-								Node *havingQual);
+								Node *havingQual, PartitionwiseAggregateType patype);
 static List *get_useful_pathkeys_for_relation(PlannerInfo *root,
 											  RelOptInfo *rel);
 static List *get_useful_ecs_for_relation(PlannerInfo *root, RelOptInfo *rel);
@@ -651,6 +651,7 @@ postgresGetForeignRelSize(PlannerInfo *root,
 	fpinfo->shippable_extensions = NIL;
 	fpinfo->fetch_size = 100;
 	fpinfo->async_capable = false;
+	fpinfo->server_version = 0;
 
 	apply_server_options(fpinfo);
 	apply_table_options(fpinfo);
@@ -6130,6 +6131,8 @@ apply_server_options(PgFdwRelationInfo *fpinfo)
 			(void) parse_int(defGetString(def), &fpinfo->fetch_size, 0, NULL);
 		else if (strcmp(def->defname, "async_capable") == 0)
 			fpinfo->async_capable = defGetBoolean(def);
+		else if (strcmp(def->defname, "server_version") == 0)
+			(void)parse_int(defGetString(def), &fpinfo->server_version, 0, NULL);
 	}
 }
 
@@ -6188,6 +6191,7 @@ merge_fdw_options(PgFdwRelationInfo *fpinfo,
 	fpinfo->use_remote_estimate = fpinfo_o->use_remote_estimate;
 	fpinfo->fetch_size = fpinfo_o->fetch_size;
 	fpinfo->async_capable = fpinfo_o->async_capable;
+	fpinfo->server_version = fpinfo_o->server_version;
 
 	/* Merge the table level options from either side of the join. */
 	if (fpinfo_i)
@@ -6367,7 +6371,7 @@ postgresGetForeignJoinPaths(PlannerInfo *root,
  */
 static bool
 foreign_grouping_ok(PlannerInfo *root, RelOptInfo *grouped_rel,
-					Node *havingQual)
+					Node *havingQual, PartitionwiseAggregateType patype)
 {
 	Query	   *query = root->parse;
 	PgFdwRelationInfo *fpinfo = (PgFdwRelationInfo *) grouped_rel->fdw_private;
@@ -6381,6 +6385,10 @@ foreign_grouping_ok(PlannerInfo *root, RelOptInfo *grouped_rel,
 	if (query->groupingSets)
 		return false;
 
+	/* It's unsafe to push having statements with partial aggregates */
+	if ((patype == PARTITIONWISE_AGGREGATE_PARTIAL) && havingQual)
+		return false;
+
 	/* Get the fpinfo of the underlying scan relation. */
 	ofpinfo = (PgFdwRelationInfo *) fpinfo->outerrel->fdw_private;
 
@@ -6619,6 +6627,7 @@ postgresGetForeignUpperPaths(PlannerInfo *root, UpperRelationKind stage,
 
 	/* Ignore stages we don't support; and skip any duplicate calls. */
 	if ((stage != UPPERREL_GROUP_AGG &&
+		stage != UPPERREL_PARTIAL_GROUP_AGG &&
 		 stage != UPPERREL_ORDERED &&
 		 stage != UPPERREL_FINAL) ||
 		output_rel->fdw_private)
@@ -6635,6 +6644,10 @@ postgresGetForeignUpperPaths(PlannerInfo *root, UpperRelationKind stage,
 			add_foreign_grouping_paths(root, input_rel, output_rel,
 									   (GroupPathExtraData *) extra);
 			break;
+		case UPPERREL_PARTIAL_GROUP_AGG:
+			add_foreign_grouping_paths(root, input_rel, output_rel,
+									   (GroupPathExtraData*)extra);
+			break;
 		case UPPERREL_ORDERED:
 			add_foreign_ordered_paths(root, input_rel, output_rel);
 			break;
@@ -6675,7 +6688,8 @@ add_foreign_grouping_paths(PlannerInfo *root, RelOptInfo *input_rel,
 		return;
 
 	Assert(extra->patype == PARTITIONWISE_AGGREGATE_NONE ||
-		   extra->patype == PARTITIONWISE_AGGREGATE_FULL);
+		   extra->patype == PARTITIONWISE_AGGREGATE_FULL ||
+		   extra->patype == PARTITIONWISE_AGGREGATE_PARTIAL);
 
 	/* save the input_rel as outerrel in fpinfo */
 	fpinfo->outerrel = input_rel;
@@ -6695,7 +6709,7 @@ add_foreign_grouping_paths(PlannerInfo *root, RelOptInfo *input_rel,
 	 * Use HAVING qual from extra. In case of child partition, it will have
 	 * translated Vars.
 	 */
-	if (!foreign_grouping_ok(root, grouped_rel, extra->havingQual))
+	if (!foreign_grouping_ok(root, grouped_rel, extra->havingQual, extra->patype))
 		return;
 
 	/*
diff --git a/contrib/postgres_fdw/postgres_fdw.h b/contrib/postgres_fdw/postgres_fdw.h
index 02c1152319..616f559080 100644
--- a/contrib/postgres_fdw/postgres_fdw.h
+++ b/contrib/postgres_fdw/postgres_fdw.h
@@ -88,6 +88,9 @@ typedef struct PgFdwRelationInfo
 
 	int			fetch_size;		/* fetch size for this remote table */
 
+	/* Options for checking compatibility of partial aggregation */
+	int			server_version;
+
 	/*
 	 * Name of the relation, for use while EXPLAINing ForeignScan.  It is used
 	 * for join and upper relations but is set for all relations.  For a base
diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql
index 5bd69339df..dcbf60d919 100644
--- a/contrib/postgres_fdw/sql/postgres_fdw.sql
+++ b/contrib/postgres_fdw/sql/postgres_fdw.sql
@@ -2944,16 +2944,34 @@ RESET enable_partitionwise_join;
 -- ===================================================================
 -- test partitionwise aggregates
 -- ===================================================================
+CREATE OR REPLACE FUNCTION f_server_version_lag(version_offset int4) RETURNS text AS $$
+    SELECT (setting::int4 + version_offset)::text FROM pg_settings WHERE name = 'server_version_num';
+$$ LANGUAGE sql;
 
-CREATE TABLE pagg_tab (a int, b int, c text) PARTITION BY RANGE(a);
+CREATE OR REPLACE FUNCTION f_alter_server_version(server_name text, operation text, version_offset int4) RETURNS void AS $$
+    DECLARE
+        version_num text;
+    BEGIN
+        EXECUTE 'SELECT f_server_version_lag($1) '
+            INTO version_num
+            USING version_offset;
+        EXECUTE format('ALTER SERVER %I OPTIONS (%I server_version %L)',
+            server_name, operation, version_num);
+        RETURN;
+    END;
+$$ LANGUAGE plpgsql;
+
+SELECT f_alter_server_version('loopback', 'add', 0);
+
+CREATE TABLE pagg_tab (a int, b int, c text, d int4) PARTITION BY RANGE(a);
 
 CREATE TABLE pagg_tab_p1 (LIKE pagg_tab);
 CREATE TABLE pagg_tab_p2 (LIKE pagg_tab);
 CREATE TABLE pagg_tab_p3 (LIKE pagg_tab);
 
-INSERT INTO pagg_tab_p1 SELECT i % 30, i % 50, to_char(i/30, 'FM0000') FROM generate_series(1, 3000) i WHERE (i % 30) < 10;
-INSERT INTO pagg_tab_p2 SELECT i % 30, i % 50, to_char(i/30, 'FM0000') FROM generate_series(1, 3000) i WHERE (i % 30) < 20 and (i % 30) >= 10;
-INSERT INTO pagg_tab_p3 SELECT i % 30, i % 50, to_char(i/30, 'FM0000') FROM generate_series(1, 3000) i WHERE (i % 30) < 30 and (i % 30) >= 20;
+INSERT INTO pagg_tab_p1 SELECT i % 30, i % 50, to_char(i/30, 'FM0000'), i % 40 FROM generate_series(1, 3000) i WHERE (i % 30) < 10;
+INSERT INTO pagg_tab_p2 SELECT i % 30, i % 50, to_char(i/30, 'FM0000'), i % 40 FROM generate_series(1, 3000) i WHERE (i % 30) < 20 and (i % 30) >= 10;
+INSERT INTO pagg_tab_p3 SELECT i % 30, i % 50, to_char(i/30, 'FM0000'), i % 40 FROM generate_series(1, 3000) i WHERE (i % 30) < 30 and (i % 30) >= 20;
 
 -- Create foreign partitions
 CREATE FOREIGN TABLE fpagg_tab_p1 PARTITION OF pagg_tab FOR VALUES FROM (0) TO (10) SERVER loopback OPTIONS (table_name 'pagg_tab_p1');
@@ -2987,6 +3005,65 @@ SELECT a, count(t1) FROM pagg_tab t1 GROUP BY a HAVING avg(b) < 22 ORDER BY 1;
 EXPLAIN (COSTS OFF)
 SELECT b, avg(a), max(a), count(*) FROM pagg_tab GROUP BY b HAVING sum(a) < 700 ORDER BY 1;
 
+-- It's unsafe to push down having clause when there are partial aggregates
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT b, max(a), count(*) FROM pagg_tab GROUP BY b HAVING sum(a) < 700 ORDER BY 1;
+
+-- Partial aggregates are fine to push down without having clause
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT b, min(a), max(a), count(*), sum(d), sum(d::int8), avg(d), avg(d::int8) FROM pagg_tab GROUP BY b ORDER BY 1;
+SELECT b, min(a), max(a), count(*), sum(d), sum(d::int8), avg(d), avg(d::int8) FROM pagg_tab GROUP BY b ORDER BY 1;
+
+-- Partial aggregates are fine to push down
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT min(a), max(a), count(*), sum(d), sum(d::int8), avg(d), avg(d::int8) FROM pagg_tab;
+SELECT min(a), max(a), count(*), sum(d), sum(d::int8), avg(d), avg(d::int8) FROM pagg_tab;
+
+-- It's unsafe to push down partial aggregates which contains distinct clause
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT max(a), count(distinct b) FROM pagg_tab;
+SELECT max(a), count(distinct b) FROM pagg_tab;
+
+-- It's unsafe to push down partial aggregates when
+-- server_version is older than partialagg_minversion, described by pg_aggregate
+SELECT f_alter_server_version('loopback', 'set', -1);
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT avg(d) FROM pagg_tab;
+SELECT avg(d) FROM pagg_tab;
+
+-- It's safe to push down partial aggregate for user defined aggregate
+-- which has valid options.
+CREATE AGGREGATE postgres_fdw_sum_p_int8(int8) (
+	SFUNC = int8_avg_accum,
+	STYPE = internal,
+	FINALFUNC = int8_avg_serialize
+);
+
+CREATE AGGREGATE postgres_fdw_sum(int8) (
+	SFUNC = int8_avg_accum,
+	STYPE = internal,
+	COMBINEFUNC = int8_avg_combine,
+	FINALFUNC = numeric_poly_sum,
+	SERIALFUNC = int8_avg_serialize,
+	DESERIALFUNC = int8_avg_deserialize,
+	PARTIALAGGFUNC = postgres_fdw_sum_p_int8,
+	PARTIALAGG_MINVERSION = 150000
+);
+
+ALTER EXTENSION postgres_fdw ADD FUNCTION postgres_fdw_sum(int8);
+
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT postgres_fdw_sum(d::int8) FROM pagg_tab;
+SELECT postgres_fdw_sum(d::int8) FROM pagg_tab;
+
+ALTER EXTENSION postgres_fdw DROP FUNCTION postgres_fdw_sum(int8);
+DROP AGGREGATE postgres_fdw_sum(int8);
+DROP AGGREGATE postgres_fdw_sum_p_int8(int8);
+
+DROP FUNCTION f_server_version_lag(int4);
+DROP FUNCTION f_alter_server_version(text, text, int4);
+ALTER SERVER loopback OPTIONS (DROP server_version);
+
 -- ===================================================================
 -- access rights and superuser
 -- ===================================================================
diff --git a/src/backend/catalog/pg_aggregate.c b/src/backend/catalog/pg_aggregate.c
index ebc4454743..037eb06f4d 100644
--- a/src/backend/catalog/pg_aggregate.c
+++ b/src/backend/catalog/pg_aggregate.c
@@ -63,6 +63,7 @@ AggregateCreate(const char *aggName,
 				List *aggmtransfnName,
 				List *aggminvtransfnName,
 				List *aggmfinalfnName,
+				List *partialaggfnName,
 				bool finalfnExtraArgs,
 				bool mfinalfnExtraArgs,
 				char finalfnModify,
@@ -72,6 +73,7 @@ AggregateCreate(const char *aggName,
 				int32 aggTransSpace,
 				Oid aggmTransType,
 				int32 aggmTransSpace,
+				int32 partialaggMinversion,
 				const char *agginitval,
 				const char *aggminitval,
 				char proparallel)
@@ -91,6 +93,7 @@ AggregateCreate(const char *aggName,
 	Oid			mtransfn = InvalidOid;	/* can be omitted */
 	Oid			minvtransfn = InvalidOid;	/* can be omitted */
 	Oid			mfinalfn = InvalidOid;	/* can be omitted */
+	Oid			partialaggfn = InvalidOid;	/* can be omitted */
 	Oid			sortop = InvalidOid;	/* can be omitted */
 	Oid		   *aggArgTypes = parameterTypes->values;
 	bool		mtransIsStrict = false;
@@ -569,6 +572,33 @@ AggregateCreate(const char *aggName,
 							format_type_be(finaltype))));
 	}
 
+	/*
+	 * Validate the partial aggregate function, if present.
+	 */
+	if (partialaggfnName)
+	{
+		HeapTuple aggtup;
+		partialaggfn = LookupFuncName(partialaggfnName, numArgs, aggArgTypes, false);
+
+		/* Check aggregate creator has permission to call the function */
+		aclresult = object_aclcheck(ProcedureRelationId, partialaggfn, GetUserId(), ACL_EXECUTE);
+		if (aclresult != ACLCHECK_OK)
+			aclcheck_error(aclresult, OBJECT_FUNCTION, get_func_name(partialaggfn));
+
+		rettype = get_func_rettype(partialaggfn);
+
+		if (((aggTransType == INTERNALOID) && (rettype != BYTEAOID))
+			|| ((aggTransType != INTERNALOID) && (rettype != aggTransType)))
+			ereport(ERROR,
+				(errcode(ERRCODE_DATATYPE_MISMATCH),
+					errmsg("return type of partialaggfunc of %s is not stype",
+						NameListToString(partialaggfnName))));
+		aggtup = SearchSysCache1(AGGFNOID, ObjectIdGetDatum(partialaggfn));
+		if (!HeapTupleIsValid(aggtup))
+			elog(ERROR, "cache lookup failed for partialaggfunc %u", partialaggfn);
+		ReleaseSysCache(aggtup);
+	}
+
 	/* handle sortop, if supplied */
 	if (aggsortopName)
 	{
@@ -684,6 +714,8 @@ AggregateCreate(const char *aggName,
 		values[Anum_pg_aggregate_aggminitval - 1] = CStringGetTextDatum(aggminitval);
 	else
 		nulls[Anum_pg_aggregate_aggminitval - 1] = true;
+	values[Anum_pg_aggregate_partialaggfn - 1] = ObjectIdGetDatum(partialaggfn);
+	values[Anum_pg_aggregate_partialagg_minversion - 1] = Int32GetDatum(partialaggMinversion);
 
 	if (replace)
 		oldtup = SearchSysCache1(AGGFNOID, ObjectIdGetDatum(procOid));
diff --git a/src/backend/commands/aggregatecmds.c b/src/backend/commands/aggregatecmds.c
index fda9d1aa77..cb49a79cfa 100644
--- a/src/backend/commands/aggregatecmds.c
+++ b/src/backend/commands/aggregatecmds.c
@@ -70,6 +70,7 @@ DefineAggregate(ParseState *pstate,
 	List	   *combinefuncName = NIL;
 	List	   *serialfuncName = NIL;
 	List	   *deserialfuncName = NIL;
+	List	   *partialaggfuncName = NIL;
 	List	   *mtransfuncName = NIL;
 	List	   *minvtransfuncName = NIL;
 	List	   *mfinalfuncName = NIL;
@@ -83,6 +84,7 @@ DefineAggregate(ParseState *pstate,
 	TypeName   *mtransType = NULL;
 	int32		transSpace = 0;
 	int32		mtransSpace = 0;
+	int32		partialaggMinversion = 0;
 	char	   *initval = NULL;
 	char	   *minitval = NULL;
 	char	   *parallel = NULL;
@@ -143,6 +145,8 @@ DefineAggregate(ParseState *pstate,
 			serialfuncName = defGetQualifiedName(defel);
 		else if (strcmp(defel->defname, "deserialfunc") == 0)
 			deserialfuncName = defGetQualifiedName(defel);
+		else if (strcmp(defel->defname, "partialaggfunc") == 0)
+			partialaggfuncName = defGetQualifiedName(defel);
 		else if (strcmp(defel->defname, "msfunc") == 0)
 			mtransfuncName = defGetQualifiedName(defel);
 		else if (strcmp(defel->defname, "minvfunc") == 0)
@@ -182,6 +186,8 @@ DefineAggregate(ParseState *pstate,
 			mtransType = defGetTypeName(defel);
 		else if (strcmp(defel->defname, "msspace") == 0)
 			mtransSpace = defGetInt32(defel);
+		else if (strcmp(defel->defname, "partialagg_minversion") == 0)
+			partialaggMinversion = defGetInt32(defel);
 		else if (strcmp(defel->defname, "initcond") == 0)
 			initval = defGetString(defel);
 		else if (strcmp(defel->defname, "initcond1") == 0)
@@ -461,6 +467,7 @@ DefineAggregate(ParseState *pstate,
 						   mtransfuncName,	/* fwd trans function name */
 						   minvtransfuncName,	/* inv trans function name */
 						   mfinalfuncName,	/* final function name */
+						   partialaggfuncName,
 						   finalfuncExtraArgs,
 						   mfinalfuncExtraArgs,
 						   finalfuncModify,
@@ -470,6 +477,7 @@ DefineAggregate(ParseState *pstate,
 						   transSpace,	/* transition space */
 						   mtransTypeId,	/* transition data type */
 						   mtransSpace, /* transition space */
+						   partialaggMinversion,
 						   initval, /* initial condition */
 						   minitval,	/* initial condition */
 						   proparallel);	/* parallel safe? */
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 7a504dfe25..80165ada03 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -13693,6 +13693,8 @@ dumpAgg(Archive *fout, const AggInfo *agginfo)
 	const char *aggmtransfn;
 	const char *aggminvtransfn;
 	const char *aggmfinalfn;
+	const char *partialaggfn;
+	const char *partialagg_minversion;
 	bool		aggfinalextra;
 	bool		aggmfinalextra;
 	char		aggfinalmodify;
@@ -13775,11 +13777,19 @@ dumpAgg(Archive *fout, const AggInfo *agginfo)
 		if (fout->remoteVersion >= 110000)
 			appendPQExpBufferStr(query,
 								 "aggfinalmodify,\n"
-								 "aggmfinalmodify\n");
+								 "aggmfinalmodify,\n");
 		else
 			appendPQExpBufferStr(query,
 								 "'0' AS aggfinalmodify,\n"
-								 "'0' AS aggmfinalmodify\n");
+								 "'0' AS aggmfinalmodify,\n");
+		if (fout->remoteVersion >= 160000)
+			appendPQExpBufferStr(query,
+								"partialaggfn,\n"
+								"partialagg_minversion\n");
+		else
+			appendPQExpBufferStr(query,
+								"'-' AS partialaggfn,\n"
+								"0 AS partialagg_minversion\n");
 
 		appendPQExpBufferStr(query,
 							 "FROM pg_catalog.pg_aggregate a, pg_catalog.pg_proc p "
@@ -13805,6 +13815,8 @@ dumpAgg(Archive *fout, const AggInfo *agginfo)
 	aggcombinefn = PQgetvalue(res, 0, PQfnumber(res, "aggcombinefn"));
 	aggserialfn = PQgetvalue(res, 0, PQfnumber(res, "aggserialfn"));
 	aggdeserialfn = PQgetvalue(res, 0, PQfnumber(res, "aggdeserialfn"));
+	partialaggfn = PQgetvalue(res, 0, PQfnumber(res, "partialaggfn"));
+	partialagg_minversion = PQgetvalue(res, 0, PQfnumber(res, "partialagg_minversion"));
 	aggmtransfn = PQgetvalue(res, 0, PQfnumber(res, "aggmtransfn"));
 	aggminvtransfn = PQgetvalue(res, 0, PQfnumber(res, "aggminvtransfn"));
 	aggmfinalfn = PQgetvalue(res, 0, PQfnumber(res, "aggmfinalfn"));
@@ -13894,6 +13906,15 @@ dumpAgg(Archive *fout, const AggInfo *agginfo)
 	if (strcmp(aggdeserialfn, "-") != 0)
 		appendPQExpBuffer(details, ",\n    DESERIALFUNC = %s", aggdeserialfn);
 
+	if (strcmp(partialaggfn, "-") != 0)
+		appendPQExpBuffer(details, ",\n    PARTIALAGGFUNC = %s", partialaggfn);
+
+	if (strcmp(partialagg_minversion, "0") != 0)
+	{
+		appendPQExpBuffer(details, ",\n    PARTIALAGG_MINVERSION = %s",
+						  partialagg_minversion);
+	}
+
 	if (strcmp(aggmtransfn, "-") != 0)
 	{
 		appendPQExpBuffer(details, ",\n    MSFUNC = %s,\n    MINVFUNC = %s,\n    MSTYPE = %s",
diff --git a/src/include/catalog/pg_aggregate.dat b/src/include/catalog/pg_aggregate.dat
index 283f494bf5..689ed3546a 100644
--- a/src/include/catalog/pg_aggregate.dat
+++ b/src/include/catalog/pg_aggregate.dat
@@ -13,22 +13,44 @@
 [
 
 # avg
+{ aggfnoid => 'avg_p_int8', aggtransfn => 'int8_avg_accum',
+  aggfinalfn => 'int8_avg_serialize', aggcombinefn => 'int8_avg_combine',
+  aggserialfn => 'int8_avg_serialize', aggdeserialfn => 'int8_avg_deserialize',
+  aggtranstype => 'internal',
+  aggtransspace => '48' },
 { aggfnoid => 'avg(int8)', aggtransfn => 'int8_avg_accum',
   aggfinalfn => 'numeric_poly_avg', aggcombinefn => 'int8_avg_combine',
   aggserialfn => 'int8_avg_serialize', aggdeserialfn => 'int8_avg_deserialize',
   aggmtransfn => 'int8_avg_accum', aggminvtransfn => 'int8_avg_accum_inv',
   aggmfinalfn => 'numeric_poly_avg', aggtranstype => 'internal',
-  aggtransspace => '48', aggmtranstype => 'internal', aggmtransspace => '48' },
+  aggtransspace => '48', aggmtranstype => 'internal', aggmtransspace => '48',
+  partialaggfn => 'avg_p_int8', partialagg_minversion => '160000' },
+{ aggfnoid => 'avg_p_int4', aggtransfn => 'int4_avg_accum',
+  aggcombinefn => 'int4_avg_combine',
+  aggtranstype => '_int8',
+  agginitval => '{0,0}' },
 { aggfnoid => 'avg(int4)', aggtransfn => 'int4_avg_accum',
   aggfinalfn => 'int8_avg', aggcombinefn => 'int4_avg_combine',
   aggmtransfn => 'int4_avg_accum', aggminvtransfn => 'int4_avg_accum_inv',
   aggmfinalfn => 'int8_avg', aggtranstype => '_int8', aggmtranstype => '_int8',
-  agginitval => '{0,0}', aggminitval => '{0,0}' },
+  agginitval => '{0,0}', aggminitval => '{0,0}',
+  partialaggfn => 'avg_p_int4', partialagg_minversion => '160000' },
+{ aggfnoid => 'avg_p_int2', aggtransfn => 'int2_avg_accum',
+  aggcombinefn => 'int4_avg_combine',
+  aggtranstype => '_int8',
+  agginitval => '{0,0}'  },
 { aggfnoid => 'avg(int2)', aggtransfn => 'int2_avg_accum',
   aggfinalfn => 'int8_avg', aggcombinefn => 'int4_avg_combine',
   aggmtransfn => 'int2_avg_accum', aggminvtransfn => 'int2_avg_accum_inv',
   aggmfinalfn => 'int8_avg', aggtranstype => '_int8', aggmtranstype => '_int8',
-  agginitval => '{0,0}', aggminitval => '{0,0}' },
+  agginitval => '{0,0}', aggminitval => '{0,0}',
+  partialaggfn => 'avg_p_int2', partialagg_minversion => '160000'  },
+{ aggfnoid => 'avg_p_numeric', aggtransfn => 'numeric_avg_accum',
+  aggfinalfn => 'numeric_avg_serialize', aggcombinefn => 'numeric_avg_combine',
+  aggserialfn => 'numeric_avg_serialize',
+  aggdeserialfn => 'numeric_avg_deserialize',
+  aggtranstype => 'internal', aggtransspace => '128',
+  partialaggfn => 'avg_p_numeric', partialagg_minversion => '160000' },
 { aggfnoid => 'avg(numeric)', aggtransfn => 'numeric_avg_accum',
   aggfinalfn => 'numeric_avg', aggcombinefn => 'numeric_avg_combine',
   aggserialfn => 'numeric_avg_serialize',
@@ -36,27 +58,45 @@
   aggmtransfn => 'numeric_avg_accum', aggminvtransfn => 'numeric_accum_inv',
   aggmfinalfn => 'numeric_avg', aggtranstype => 'internal',
   aggtransspace => '128', aggmtranstype => 'internal',
-  aggmtransspace => '128' },
+  aggmtransspace => '128',
+  partialaggfn => 'avg_p_numeric', partialagg_minversion => '160000' },
+{ aggfnoid => 'avg_p_float4', aggtransfn => 'float4_accum',
+  aggcombinefn => 'float8_combine',
+  aggtranstype => '_float8', agginitval => '{0,0,0}' },
 { aggfnoid => 'avg(float4)', aggtransfn => 'float4_accum',
   aggfinalfn => 'float8_avg', aggcombinefn => 'float8_combine',
+  aggtranstype => '_float8', agginitval => '{0,0,0}',
+  partialaggfn => 'avg_p_float4', partialagg_minversion => '160000' },
+{ aggfnoid => 'avg_p_float8', aggtransfn => 'float8_accum',
+  aggcombinefn => 'float8_combine',
   aggtranstype => '_float8', agginitval => '{0,0,0}' },
 { aggfnoid => 'avg(float8)', aggtransfn => 'float8_accum',
   aggfinalfn => 'float8_avg', aggcombinefn => 'float8_combine',
-  aggtranstype => '_float8', agginitval => '{0,0,0}' },
+  aggtranstype => '_float8', agginitval => '{0,0,0}',
+  partialaggfn => 'avg_p_float8', partialagg_minversion => '160000'  },
+{ aggfnoid => 'avg_p_interval', aggtransfn => 'interval_accum',
+  aggcombinefn => 'interval_combine',
+  aggtranstype => '_interval', agginitval => '{0 second,0 second}'  },
 { aggfnoid => 'avg(interval)', aggtransfn => 'interval_accum',
   aggfinalfn => 'interval_avg', aggcombinefn => 'interval_combine',
   aggmtransfn => 'interval_accum', aggminvtransfn => 'interval_accum_inv',
   aggmfinalfn => 'interval_avg', aggtranstype => '_interval',
   aggmtranstype => '_interval', agginitval => '{0 second,0 second}',
-  aggminitval => '{0 second,0 second}' },
+  aggminitval => '{0 second,0 second}',
+  partialaggfn => 'avg_p_interval', partialagg_minversion => '160000'  },
 
 # sum
+{ aggfnoid => 'sum_p_int8', aggtransfn => 'int8_avg_accum',
+  aggfinalfn => 'int8_avg_serialize', aggcombinefn => 'int8_avg_combine',
+  aggserialfn => 'int8_avg_serialize', aggdeserialfn => 'int8_avg_deserialize',
+  aggtranstype => 'internal', aggtransspace => '48' },
 { aggfnoid => 'sum(int8)', aggtransfn => 'int8_avg_accum',
   aggfinalfn => 'numeric_poly_sum', aggcombinefn => 'int8_avg_combine',
   aggserialfn => 'int8_avg_serialize', aggdeserialfn => 'int8_avg_deserialize',
   aggmtransfn => 'int8_avg_accum', aggminvtransfn => 'int8_avg_accum_inv',
   aggmfinalfn => 'numeric_poly_sum', aggtranstype => 'internal',
-  aggtransspace => '48', aggmtranstype => 'internal', aggmtransspace => '48' },
+  aggtransspace => '48', aggmtranstype => 'internal', aggmtransspace => '48',
+  partialaggfn => 'sum_p_int8', partialagg_minversion => '160000' },
 { aggfnoid => 'sum(int4)', aggtransfn => 'int4_sum', aggcombinefn => 'int8pl',
   aggmtransfn => 'int4_avg_accum', aggminvtransfn => 'int4_avg_accum_inv',
   aggmfinalfn => 'int2int4_sum', aggtranstype => 'int8',
@@ -76,6 +116,11 @@
   aggcombinefn => 'interval_pl', aggmtransfn => 'interval_pl',
   aggminvtransfn => 'interval_mi', aggtranstype => 'interval',
   aggmtranstype => 'interval' },
+{ aggfnoid => 'sum_p_numeric', aggtransfn => 'numeric_avg_accum',
+  aggfinalfn => 'numeric_avg_serialize', aggcombinefn => 'numeric_avg_combine',
+  aggserialfn => 'numeric_avg_serialize',
+  aggdeserialfn => 'numeric_avg_deserialize',
+  aggtranstype => 'internal', aggtransspace => '128' },
 { aggfnoid => 'sum(numeric)', aggtransfn => 'numeric_avg_accum',
   aggfinalfn => 'numeric_sum', aggcombinefn => 'numeric_avg_combine',
   aggserialfn => 'numeric_avg_serialize',
@@ -83,7 +128,8 @@
   aggmtransfn => 'numeric_avg_accum', aggminvtransfn => 'numeric_accum_inv',
   aggmfinalfn => 'numeric_sum', aggtranstype => 'internal',
   aggtransspace => '128', aggmtranstype => 'internal',
-  aggmtransspace => '128' },
+  aggmtransspace => '128',
+  partialaggfn => 'sum_p_numeric', partialagg_minversion => '160000' },
 
 # max
 { aggfnoid => 'max(int8)', aggtransfn => 'int8larger',
diff --git a/src/include/catalog/pg_aggregate.h b/src/include/catalog/pg_aggregate.h
index 3112881193..1cefa1bfba 100644
--- a/src/include/catalog/pg_aggregate.h
+++ b/src/include/catalog/pg_aggregate.h
@@ -55,6 +55,12 @@ CATALOG(pg_aggregate,2600,AggregateRelationId)
 	/* function to convert bytea to transtype (0 if none) */
 	regproc		aggdeserialfn BKI_DEFAULT(-) BKI_LOOKUP_OPT(pg_proc);
 
+	/* special aggregate function for partial aggregation (0 if none) */
+	regproc		partialaggfn BKI_DEFAULT(-) BKI_LOOKUP_OPT(pg_proc);
+
+	/* minimum PostgreSQL's version which has compatible partialaggfn */
+	int32		partialagg_minversion BKI_DEFAULT(0);
+
 	/* forward function for moving-aggregate mode (0 if none) */
 	regproc		aggmtransfn BKI_DEFAULT(-) BKI_LOOKUP_OPT(pg_proc);
 
@@ -141,6 +147,9 @@ DECLARE_UNIQUE_INDEX_PKEY(pg_aggregate_fnoid_index, 2650, AggregateFnoidIndexId,
 #define AGGMODIFY_SHAREABLE			's'
 #define AGGMODIFY_READ_WRITE		'w'
 
+/* Symbolic value for default partialagg_minversion */
+#define	PARTIALAGG_MINVERSION_DEFAULT	0
+
 #endif							/* EXPOSE_TO_CLIENT_CODE */

@@ -164,6 +173,7 @@ extern ObjectAddress AggregateCreate(const char *aggName,
 									 List *aggmtransfnName,
 									 List *aggminvtransfnName,
 									 List *aggmfinalfnName,
+									 List *aggpartialfnName,
 									 bool finalfnExtraArgs,
 									 bool mfinalfnExtraArgs,
 									 char finalfnModify,
@@ -173,6 +183,7 @@ extern ObjectAddress AggregateCreate(const char *aggName,
 									 int32 aggTransSpace,
 									 Oid aggmTransType,
 									 int32 aggmTransSpace,
+									 int32 partialaggMinversion,
 									 const char *agginitval,
 									 const char *aggminitval,
 									 char proparallel);
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index f9f2642201..2f27b54133 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -6594,35 +6594,67 @@
   descr => 'the average (arithmetic mean) as numeric of all bigint values',
   proname => 'avg', prokind => 'a', proisstrict => 'f', prorettype => 'numeric',
   proargtypes => 'int8', prosrc => 'aggregate_dummy' },
+{ oid => '13500',
+  descr => 'the partial average (arithmetic mean) as numeric of all bigint values',
+  proname => 'avg_p_int8', prokind => 'a', proisstrict => 'f', prorettype => 'bytea',
+  proargtypes => 'int8', prosrc => 'aggregate_dummy' },
 { oid => '2101',
   descr => 'the average (arithmetic mean) as numeric of all integer values',
   proname => 'avg', prokind => 'a', proisstrict => 'f', prorettype => 'numeric',
   proargtypes => 'int4', prosrc => 'aggregate_dummy' },
+{ oid => '13501',
+  descr => 'the partial average (arithmetic mean) as numeric of all integer values',
+  proname => 'avg_p_int4', prokind => 'a', proisstrict => 'f', prorettype => '_int8',
+  proargtypes => 'int4', prosrc => 'aggregate_dummy' },
 { oid => '2102',
   descr => 'the average (arithmetic mean) as numeric of all smallint values',
   proname => 'avg', prokind => 'a', proisstrict => 'f', prorettype => 'numeric',
   proargtypes => 'int2', prosrc => 'aggregate_dummy' },
+{ oid => '13502',
+  descr => 'the partial average (arithmetic mean) as numeric of all smallint values',
+  proname => 'avg_p_int2', prokind => 'a', proisstrict => 'f', prorettype => '_int8',
+  proargtypes => 'int2', prosrc => 'aggregate_dummy' },
 { oid => '2103',
   descr => 'the average (arithmetic mean) as numeric of all numeric values',
   proname => 'avg', prokind => 'a', proisstrict => 'f', prorettype => 'numeric',
   proargtypes => 'numeric', prosrc => 'aggregate_dummy' },
+{ oid => '13503',
+  descr => 'the partial average (arithmetic mean) as numeric of all numeric values',
+  proname => 'avg_p_numeric', prokind => 'a', proisstrict => 'f', prorettype => 'bytea',
+  proargtypes => 'numeric', prosrc => 'aggregate_dummy' },
 { oid => '2104',
   descr => 'the average (arithmetic mean) as float8 of all float4 values',
   proname => 'avg', prokind => 'a', proisstrict => 'f', prorettype => 'float8',
   proargtypes => 'float4', prosrc => 'aggregate_dummy' },
+{ oid => '13504',
+  descr => 'the partial average (arithmetic mean) as float8 of all float4 values',
+  proname => 'avg_p_float4', prokind => 'a', proisstrict => 'f', prorettype => '_float8',
+  proargtypes => 'float4', prosrc => 'aggregate_dummy' },
 { oid => '2105',
   descr => 'the average (arithmetic mean) as float8 of all float8 values',
   proname => 'avg', prokind => 'a', proisstrict => 'f', prorettype => 'float8',
   proargtypes => 'float8', prosrc => 'aggregate_dummy' },
+{ oid => '13505',
+  descr => 'the partial average (arithmetic mean) as float8 of all float8 values',
+  proname => 'avg_p_float8', prokind => 'a', proisstrict => 'f', prorettype => '_float8',
+  proargtypes => 'float8', prosrc => 'aggregate_dummy' },
 { oid => '2106',
   descr => 'the average (arithmetic mean) as interval of all interval values',
   proname => 'avg', prokind => 'a', proisstrict => 'f',
   prorettype => 'interval', proargtypes => 'interval',
   prosrc => 'aggregate_dummy' },
+{ oid => '13506',
+  descr => 'the partial average (arithmetic mean) as interval of all interval values',
+  proname => 'avg_p_interval', prokind => 'a', proisstrict => 'f',
+  prorettype => '_interval', proargtypes => 'interval',
+  prosrc => 'aggregate_dummy' },
 
 { oid => '2107', descr => 'sum as numeric across all bigint input values',
   proname => 'sum', prokind => 'a', proisstrict => 'f', prorettype => 'numeric',
   proargtypes => 'int8', prosrc => 'aggregate_dummy' },
+{ oid => '13507', descr => 'partial sum as numeric across all bigint input values',
+  proname => 'sum_p_int8', prokind => 'a', proisstrict => 'f', prorettype => 'bytea',
+  proargtypes => 'int8', prosrc => 'aggregate_dummy' },
 { oid => '2108', descr => 'sum as bigint across all integer input values',
   proname => 'sum', prokind => 'a', proisstrict => 'f', prorettype => 'int8',
   proargtypes => 'int4', prosrc => 'aggregate_dummy' },
@@ -6645,6 +6677,9 @@
 { oid => '2114', descr => 'sum as numeric across all numeric input values',
   proname => 'sum', prokind => 'a', proisstrict => 'f', prorettype => 'numeric',
   proargtypes => 'numeric', prosrc => 'aggregate_dummy' },
+{ oid => '13508', descr => 'partial sum as numeric across all numeric input values',
+  proname => 'sum_p_numeric', prokind => 'a', proisstrict => 'f', prorettype => 'bytea',
+  proargtypes => 'numeric', prosrc => 'aggregate_dummy' },
 
 { oid => '2115', descr => 'maximum value of all bigint input values',
   proname => 'max', prokind => 'a', proisstrict => 'f', prorettype => 'int8',
diff --git a/src/test/regress/expected/create_aggregate.out b/src/test/regress/expected/create_aggregate.out
index dcf6909423..3197c8abba 100644
--- a/src/test/regress/expected/create_aggregate.out
+++ b/src/test/regress/expected/create_aggregate.out
@@ -322,3 +322,27 @@ WARNING:  aggregate attribute "Finalfunc_extra" not recognized
 WARNING:  aggregate attribute "Finalfunc_modify" not recognized
 WARNING:  aggregate attribute "Parallel" not recognized
 ERROR:  aggregate stype must be specified
+-- invalid: partialaggfunc which doesn't exist
+CREATE AGGREGATE haspartialagg_agg(int8) (
+	sfunc = int8_avg_accum,
+	stype = internal,
+	combinefunc = int8_avg_combine,
+	finalfunc = numeric_poly_sum,
+	serialfunc = int8_avg_serialize,
+	deserialfunc = int8_avg_deserialize,
+	partialaggfunc = partialagg_foo,
+	partialagg_minversion = 160000
+);
+ERROR:  function partialagg_foo(bigint) does not exist
+-- invalid: partialagg_minversion is not integer
+CREATE AGGREGATE haspartialagg_agg(int8) (
+	sfunc = int8_avg_accum,
+	stype = internal,
+	combinefunc = int8_avg_combine,
+	finalfunc = numeric_poly_sum,
+	serialfunc = int8_avg_serialize,
+	deserialfunc = int8_avg_deserialize,
+	partialaggfunc = partialagg_foo,
+	partialagg_minversion = aaa
+);
+ERROR:  partialagg_minversion requires an integer value
diff --git a/src/test/regress/expected/oidjoins.out b/src/test/regress/expected/oidjoins.out
index 215eb899be..49f2c9bc35 100644
--- a/src/test/regress/expected/oidjoins.out
+++ b/src/test/regress/expected/oidjoins.out
@@ -145,6 +145,7 @@ NOTICE:  checking pg_aggregate {aggfinalfn} => pg_proc {oid}
 NOTICE:  checking pg_aggregate {aggcombinefn} => pg_proc {oid}
 NOTICE:  checking pg_aggregate {aggserialfn} => pg_proc {oid}
 NOTICE:  checking pg_aggregate {aggdeserialfn} => pg_proc {oid}
+NOTICE:  checking pg_aggregate {partialaggfn} => pg_proc {oid}
 NOTICE:  checking pg_aggregate {aggmtransfn} => pg_proc {oid}
 NOTICE:  checking pg_aggregate {aggminvtransfn} => pg_proc {oid}
 NOTICE:  checking pg_aggregate {aggmfinalfn} => pg_proc {oid}
diff --git a/src/test/regress/sql/create_aggregate.sql b/src/test/regress/sql/create_aggregate.sql
index d4b4036fd7..499c120ca5 100644
--- a/src/test/regress/sql/create_aggregate.sql
+++ b/src/test/regress/sql/create_aggregate.sql
@@ -328,3 +328,27 @@ CREATE AGGREGATE case_agg(float8)
 	"Finalfunc_modify" = read_write,
 	"Parallel" = safe
 );
+
+-- invalid: partialaggfunc which doesn't exist
+CREATE AGGREGATE haspartialagg_agg(int8) (
+	sfunc = int8_avg_accum,
+	stype = internal,
+	combinefunc = int8_avg_combine,
+	finalfunc = numeric_poly_sum,
+	serialfunc = int8_avg_serialize,
+	deserialfunc = int8_avg_deserialize,
+	partialaggfunc = partialagg_foo,
+	partialagg_minversion = 160000
+);
+
+-- invalid: partialagg_minversion is not integer
+CREATE AGGREGATE haspartialagg_agg(int8) (
+	sfunc = int8_avg_accum,
+	stype = internal,
+	combinefunc = int8_avg_combine,
+	finalfunc = numeric_poly_sum,
+	serialfunc = int8_avg_serialize,
+	deserialfunc = int8_avg_deserialize,
+	partialaggfunc = partialagg_foo,
+	partialagg_minversion = aaa
+);
\ No newline at end of file

---

From: "[email-masked]" <[email-masked]>
Date: Fri, 7 Apr 2023 09:25:52 +0000

Hi Mr.Momjian

> First, my apologies for not addressing this sooner.  I was so focused on my
> own tasks that I didn't realize this very important patch was not getting
> attention.  I will try my best to get it into PG 17.
Thank you very much for your comments. 
I will improve this patch for PG17.
I believe that this patch will help us use PostgreSQL's built-in sharding for OLAP.

> aggregate tables for partial FDW execution.
Yes. This patch enables partial aggregate pushdown using 
only existing functions which belong to existing aggregate function
and are needed by parallel query(such as state transition function and serialization function).
This patch does not need new types of function belonging to aggregate functions
and does not need new functions belonging to existing aggregate functions.

> I suggest we remove the version check requirement --- instead just document
> that the FDW Postgres version should be the same or newer than the calling
> Postgres server --- that way, we can assume that whatever is in the system
> catalogs of the caller is in the receiving side.  
Thanks for the comment. I will modify this patch according to your comment.

> We should add a GUC to turn off
> this optimization for cases where the FDW Postgres version is older than the
> caller.  This handles case 1-2.
Thanks for the advice here too.
I thought it would be more appropriate to add a foregin server option of 
postgres_fdw rather than adding GUC. 
Would you mind if I ask you what you think about it?

> > 2. Automation of creating definition of partialaggfuncs In development
> > of v17, I manually create definition of partialaggfuncs for avg, min, max, sum,
> count.
> > So I am thinking that v17 should be modified to automate creating
> > definition of partialaggfuncs for all built-in aggregate functions.
> 
> Are there any other builtin functions that need this?  I think we can just
> provide documention for extensions on how to do this.
For practical purposes, it is sufficient 
if partial aggregate for the above functions can be pushed down.
I think you are right, it would be sufficient to document how to achieve
 partial aggregate pushdown for other built-in functions.

> I can help with that once we decide on the above.
Thank you. In the next verion of this patch, I will add documents on postgres_fdw
and other places. 

> I think 'partialaggfn' should be named 'aggpartialfn' to match other columns in
> pg_aggregate.
Thanks for the comment. I will modify this patch according to your comment.

> aggregates to non-PG servers, let along parallel FDW ones.  Does anyone
To allow partial aggregate pushdown for non-PG FDWs,
I think we need to add pushdown logic to their FDWs for each function.
For example, we need to add logic avg() -> sum()/count() to their FDWs for avg.
To allow parallel partial aggregate by non-PG FDWs,
I think we need to add FDW Routines for Asynchronous Execution to their FDWs[1].

> +  aggfinalfn => 'int8_avg_serialize', aggcombinefn =>
> +'int8_avg_combine',
> +  aggserialfn => 'int8_avg_serialize', aggdeserialfn =>
> +'int8_avg_deserialize',
> +  aggfinalfn => 'numeric_avg_serialize', aggcombinefn =>
> +'numeric_avg_combine',
This reason is that sum(int8)/sum(numeric) shares some functions with avg(int8)/avg(numeric),
and sum_p_int8 is aggpartialfn of sum(int8) and sum_p_numeric is aggpartialfn of sum(numeric).

--Part of avg(int8) in BKI file in PostgreSQL15.0[2].
{ aggfnoid => 'avg(int8)', aggtransfn => 'int8_avg_accum',
  aggfinalfn => 'numeric_poly_avg', aggcombinefn => 'int8_avg_combine',
  aggserialfn => 'int8_avg_serialize', aggdeserialfn => 'int8_avg_deserialize',
  aggmtransfn => 'int8_avg_accum', aggminvtransfn => 'int8_avg_accum_inv',
  aggmfinalfn => 'numeric_poly_avg', aggtranstype => 'internal',
  aggtransspace => '48', aggmtranstype => 'internal', aggmtransspace => '48' },
--

--Part of sum(int8) in BKI file in PostgreSQL15.0[2].
{ aggfnoid => 'sum(int8)', aggtransfn => 'int8_avg_accum',
  aggfinalfn => 'numeric_poly_sum', aggcombinefn => 'int8_avg_combine',
  aggserialfn => 'int8_avg_serialize', aggdeserialfn => 'int8_avg_deserialize',
  aggmtransfn => 'int8_avg_accum', aggminvtransfn => 'int8_avg_accum_inv',
  aggmfinalfn => 'numeric_poly_sum', aggtranstype => 'internal',
  aggtransspace => '48', aggmtranstype => 'internal', aggmtransspace => '48' },
--

[1] https://www.postgresql.org/docs/15/fdw-callbacks.html#FDW-CALLBACKS-ASYNC
[2] https://github.com/postgres/postgres/blob/REL_15_0/src/include/catalog/pg_aggregate.dat

Sincerely yours,
Yuuki Fujii

---

From: Bruce Momjian <[email-masked]>
Date: Fri, 7 Apr 2023 21:50:11 -0400

> I believe that this patch will help us use PostgreSQL's built-in sharding for OLAP.

Agreed!  Again, my apologies for not helping with this _much_ sooner. 
You have done amazing work here.

> and does not need new functions belonging to existing aggregate functions.

Very nice.

> Would you mind if I ask you what you think about it?

I like the GUC idea because it gives administrators a single place to
check if the feature is enabled.  However, I can imagine cases where you
might have multiple remote FDW servers and some might be older than the
sending server.

What I don't want is an error-prone setup where administrators have to
remember what the per-server settings are.  Based on your suggestion,
let's allow CREATE SERVER to have an option 'enable_async_aggregate' (is
that the right name?), which defaults to 'true' for _all_ servers, even
those that don't support async aggregates.  With that, all FDW servers
are enabled by default, and if the FDW extension supports async
aggregates, they will automatically be pushed down and will report an
error only if the remote FDW is too old to support it.

>  partial aggregate pushdown for other built-in functions.

Uh, we actually want the patch to implement partial aggregate pushdown
for all builtin data types that can support it.  Is that done?  I think
it is only extension aggregates, which we do not control, that need this
documentation.

> Thank you. In the next verion of this patch, I will add documents on postgres_fdw

Good.

> I think we need to add FDW Routines for Asynchronous Execution to their FDWs[1].

Okay, I think we can just implement this for 1-2 and let extensions
worry about 3.

> and sum_p_int8 is aggpartialfn of sum(int8) and sum_p_numeric is aggpartialfn of sum(numeric).

Ah, I see this now, thanks.

---

From: Tom Lane <[email-masked]>
Date: Fri, 07 Apr 2023 21:55:00 -0400

Bruce Momjian <[email-masked]> writes:
> those that don't support async aggregates.

Uh, what?  Why would we not be able to tell from the remote server's
version number whether it has this ability?

			regards, tom lane

---

From: Bruce Momjian <[email-masked]>
Date: Fri, 7 Apr 2023 22:34:08 -0400

> version number whether it has this ability?

That was covered here:

	https://www.postgresql.org/message-id/ZC95C0%2BPVhVP3iax%40momjian.us

	I think we have three possible cases for aggregate pushdown to FDWs:

	Your patch handles #1 by checking that the FDW Postgres version is the
-->	same as the calling Postgres version.  However, it doesn't check for
-->	extension versions, and frankly, I don't see how we could implement that
-->	cleanly without significant complexity.

The issue is not a mismatch of postgres_fdw versions but the extension
versions and whether the partial aggregate functions exist on the remote
side, e.g., something like a PostGIS upgrade.

---

From: Tom Lane <[email-masked]>
Date: Fri, 07 Apr 2023 22:44:09 -0400

> side, e.g., something like a PostGIS upgrade.

postgres_fdw has no business pushing down calls to non-builtin functions
unless the user has explicitly authorized that via the existing
whitelisting mechanism.  I think you're reinventing the wheel,
and not very well.

			regards, tom lane

---

From: Bruce Momjian <[email-masked]>
Date: Fri, 7 Apr 2023 22:53:53 -0400

> whitelisting mechanism.  I think you're reinventing the wheel,

The patch has you assign an option at CREATE AGGREGATE time if it can do
push down, and postgres_fdw checks that.  What whitelisting mechanism
are you talking about?  async_capable?

---

From: Bruce Momjian <[email-masked]>
Date: Fri, 7 Apr 2023 23:04:23 -0400

> are you talking about?  async_capable?

FYI, in the patch the CREATE AGGREGATE option is called PARTIALAGGFUNC.

---

From: Andres Freund <[email-masked]>
Date: Fri, 7 Apr 2023 21:16:14 -0700

> are you talking about?  async_capable?

extensions (string)

    This option is a comma-separated list of names of PostgreSQL extensions that are installed, in compatible versions, on both the local and remote servers. Functions and operators that are immutable and belong to a listed extension will be considered shippable to the remote server. This option can only be specified for foreign servers, not per-table.

    When using the extensions option, it is the user's responsibility that the listed extensions exist and behave identically on both the local and remote servers. Otherwise, remote queries may fail or behave unexpectedly.

---

From: Bruce Momjian <[email-masked]>
Date: Sat, 8 Apr 2023 10:15:40 -0400

>     When using the extensions option, it is the user's responsibility that the listed extensions exist and behave identically on both the local and remote servers. Otherwise, remote queries may fai…

Okay, this is very helpful --- it is exactly the issue we are dealing
with --- how can we know if partial aggregate functions exists on the
remote server.  (I knew I was going to need API help on this.)

So, let's remove the PARTIALAGG_MINVERSION option from the patch and
just make it automatic --- we push down builtin partial aggregates if
the remote server is the same or newer _major_ version than the sending
server.  For extensions, if people have older extensions on the same or
newer foreign servers, they can adjust 'extensions' above.

---

From: Bruce Momjian <[email-masked]>
Date: Sat, 8 Apr 2023 12:18:21 -0400

> newer foreign servers, they can adjust 'extensions' above.

Looking further, I don't see any cases where we check if a builtin
function added in a major release also exists on the foreign server, so
maybe we don't need any checks but just need a mention in the release
notes.

---

From: "[email-masked]" <[email-masked]>
Date: Mon, 10 Apr 2023 01:18:37 +0000

Hi Mr.Momjian, Mr.Lane, Mr.Freund.

Thank you for advices.

> From: Bruce Momjian <[email-masked]>
> > > > 2. Automation of creating definition of partialaggfuncs In
> > > > development of v17, I manually create definition of
> > > > partialaggfuncs for avg, min, max, sum,
> > > count.
> > > Are there any other builtin functions that need this?  I think we
> > > can just provide documention for extensions on how to do this.
> > For practical purposes, it is sufficient if partial aggregate for the
> > above functions can be pushed down.
> > I think you are right, it would be sufficient to document how to
> > achieve  partial aggregate pushdown for other built-in functions.
> 
> Uh, we actually want the patch to implement partial aggregate pushdown for all
> builtin data types that can support it.  Is that done?  I think it is only extension
> aggregates, which we do not control, that need this documentation.
The last version of this patch can't pushdown partial aggregate for all builtin aggregate functions that can support it.
I will improve this patch to pushdown partial aggregate for all builtin aggregate functions
that can support it.

There is one more thing I would like your opinion on.
As the major version of PostgreSQL increase, it is possible that
the new builtin aggregate functions are added to the newer PostgreSQL.
This patch assume that aggpartialfns definitions exist in BKI files.
Due to this assumption, someone should add aggpartialfns definitions of new builtin aggregate functions to BKI files.
There are two possible ways to address this issue. Would the way1 be sufficient?
Or would way2 be preferable?
  way1) Adding documentaion for how to add these definitions to BKI files
  way2) Improving this patch to automatically add these definitions to BKI files by some tool such as initdb.

> From: Bruce Momjian <[email-masked]>
> > > > postgres_fdw has no business pushing down calls to non-builtin
> > > > functions unless the user has explicitly authorized that via the
> > > > existing whitelisting mechanism.  I think you're reinventing the
> > > > wheel, and not very well.
> > >
> > > The patch has you assign an option at CREATE AGGREGATE time if it
> > > can do push down, and postgres_fdw checks that.  What whitelisting
> > > mechanism are you talking about?  async_capable?
> >
> > extensions (string)
> >
> >     This option is a comma-separated list of names of PostgreSQL
> extensions that are installed, in compatible versions, on both the local and
> remote servers. Functions and operators that are immutable and belong to a
> listed extension will be considered shippable to the remote server. This option
> can only be specified for foreign servers, not per-table.
> >
> >     When using the extensions option, it is the user's responsibility that the
> listed extensions exist and behave identically on both the local and remote
> servers. Otherwise, remote queries may fail or behave unexpectedly.
> 
> Okay, this is very helpful --- it is exactly the issue we are dealing with --- how
> can we know if partial aggregate functions exists on the remote server.  (I
> knew I was going to need API help on this.)
> 
> So, let's remove the PARTIALAGG_MINVERSION option from the patch and just
> make it automatic --- we push down builtin partial aggregates if the remote
> server is the same or newer _major_ version than the sending server.  For
> extensions, if people have older extensions on the same or newer foreign
> servers, they can adjust 'extensions' above.
Okay, I understand. I will remove PARTIALAGG_MINVERSION option from the patch
and I will add check whether aggpartialfn depends on some extension which
is containd in extensions list of the postgres_fdw's foreign server.
In the next version of this patch,
we can pushdown partial aggregate for an user-defined aggregate function only 
when the function pass through this check.

Sincerely yours,
Yuuki Fujii

---

From: Bruce Momjian <[email-masked]>
Date: Thu, 13 Apr 2023 02:12:44 -0400

>   way2) Improving this patch to automatically add these definitions to BKI files by some tool such as initdb.

I think documentation is sufficient.  You already showed that someone
can do this with CREATE AGGREGATE for non-builtin aggregates.

> is containd in extensions list of the postgres_fdw's foreign server.

Yes, good.  Did we never push down aggregates before?  I thought we
pushed down partitionwise aggregates already, and such a check should
already be done.  If the check isn't there, it should be.

> when the function pass through this check.

Understood.

---

From: Bruce Momjian <[email-masked]>
Date: Thu, 13 Apr 2023 02:50:36 -0400

> > when the function pass through this check.

In summary, we don't do any version check for built-in function
pushdown, so we don't need it for aggregates either.  We check extension
functions against the extension pushdown list, so we should be checking
this for partial aggregate pushdown, and for partition-wise aggregate
pushdown.  If we don't do that last check already, it is a bug.

---

From: "[email-masked]" <[email-masked]>
Date: Thu, 13 Apr 2023 10:56:26 +0000

Hi Mr.Momjian.

> > As the major version of PostgreSQL increase, it is possible that the
> > new builtin aggregate functions are added to the newer PostgreSQL.
> > Due to this assumption, someone should add aggpartialfns definitions of
> new builtin aggregate functions to BKI files.
> I think documentation is sufficient.  You already showed that someone can do
> this with CREATE AGGREGATE for non-builtin aggregates.
Thank you for your opinion. I will modify this patch according to the way1.

> > > just make it automatic --- we push down builtin partial aggregates
> > > if the remote server is the same or newer _major_ version than the
> > > sending server.  For extensions, if people have older extensions on
> > > the same or newer foreign servers, they can adjust 'extensions' above.
> > Okay, I understand. I will remove PARTIALAGG_MINVERSION option from
> > the patch and I will add check whether aggpartialfn depends on some
> > extension which is containd in extensions list of the postgres_fdw's foreign
> server.
> 
Yes. The last version of this patch(and original postgres_fdw) checks if
user-defined aggregate depends some extension which is contained in 'extensions'.
But, in the last version of this patch, there is no such check for 
aggpartialfn of user-defined aggregate. So, I will add such check to this patch. 
I think that this modification is easy to do . 

> In summary, we don't do any version check for built-in function pushdown, so
> we don't need it for aggregates either.  We check extension functions against
> the extension pushdown list, so we should be checking this for partial
> aggregate pushdown, and for partition-wise aggregate pushdown.  If we don't
> do that last check already, it is a bug.
I understood.

Sincerely yours,
Yuuki Fujii

---

From: Bruce Momjian <[email-masked]>
Date: Fri, 14 Apr 2023 02:54:23 -0400

> I think that this modification is easy to do . 

Good, so our existing code is correct and the patch just needs
adjustment.

---

From: "[email-masked]" <[email-masked]>
Date: Fri, 2 Jun 2023 03:54:06 +0000

Hi Mr.Bruce, hackers.

I updated the patch.
The following is a list of comments received on the previous version of the patch
and my update to them in this version of the patch.

[comment1]
> Sent: Saturday, April 8, 2023 10:50 AM
> builtin data types that can support it.

I improved the patch to push partial aggregate down for all builtin aggregates that 
support it.

[comment2]
> Sent: Thursday, April 13, 2023 3:51 PM
> aggregate pushdown, and for partition-wise aggregate pushdown.

I removed partialagg_minversion from pg_aggregate and removed the version 
check for partial aggregate pushdown by it. postgres_fdw assumes that every 
built-in aggregate function has its aggpartialfunc on remote server.
postgres_fdw assumes that a user-defined aggregate function has its 
aggpartialfunc on remote server only when the user-defined aggregate function 
and the aggpartialfunc of it belong to some extension that's listed in the 
foreign server's extensions option. 

[comment3]
> Sent: Saturday, April 8, 2023 10:50 AM
> > > > 3. Documentation
> > > > I need add explanation of partialaggfunc to documents on
> > > > postgres_fdw and
> > > other places.
> > >
> > Thank you. In the next verion of this patch, I will add documents on
> > postgres_fdw and other places.
> 
> Good.

I appended description for partial aggregate pushdown feature by postgres_fdw
to existing documents.
The following is a list of sgml files that have been appended and the
contents of the additions.
  postgres-fdw.sgml    : Description about this partial aggregate pushdown feature
                         and definition of aggpartialfunc.
      * Unlike existing aggregate pushdown feature, this partial aggregate
        pushdown feature is a one of the built-in sharding features in PostgreSQL.
        So I added a section about built-in sharding feature in PostgreSQL, and in that
        section I added a description of this partial aggregate pushdown feature.
        In this document, a description of the built-in sharding feature in PostgreSQL
        is based on [1].
  xaggr.sgml           :Partial aggregate pushdown feature for user-defined
                        aggregate functions.
  create_aggregate.sgml:Description about additional parameters for
                        partial aggregate pushdown feature.
  catalogs.sgml        :Description about aggpartialfn column.

[comment4]
> Sent: Thursday, April 13, 2023 3:13 PM

The update addressesing to comment3 also addresses this comment.
If a new aggregate function is added in future, 
definition of aggpartialfunc in postgres-fdw.sgml
helps postgres_fdw developer add a new aggpartialfunc corresponding to the aggregate
function to existing BKI files.
For user-defined aggregate functions, description in xaggr.sgml and
create_aggregate.sgml help a user create an aggpartialfunc corresponding to
a user-defined aggregate function.

In addition, I added a validation mechanism to determine whether an
aggpartialfunc corresponding to an aggregate function is correctly created.
The aggpartialfunc information for the built-in aggregate functions is registered
in the system catalog pg_aggregate at the time the database cluster is created
from BKI files.
So I added this validation process to the regression test of postgres_fdw.

However, I am concerned that it might be more appropriate to add this validation
process to the build process or initdb, rather than to the regression test.
I would appreciate comments from the PostgreSQL community on this point.
For aggpartialfunc for user-defined functions,
I added this validation process to pg_aggregate.c.

[comment5]
> Sent: Friday, April 7, 2023 11:00 AM

Fixed.

[1] PostgreSQL wiki, WIP PostgreSQL Sharding
https://wiki.postgresql.org/wiki/WIP_PostgreSQL_Sharding

Sincerely yours,
Yuuki Fujii

---

From: Alexander Pyhalov <[email-masked]>
Date: Mon, 05 Jun 2023 12:00:27 +0300

> and my update to them in this version of the patch.

Hi.

I've looked through the last version of the patch.

Have found one issue -

src/backend/catalog/pg_aggregate.c

585                 if(strcmp(strVal(linitial(aggpartialfnName)), 
aggName) == 0){
586                         if(((aggTransType != INTERNALOID) && 
(finalfn != InvalidOid))
587                                         || ((aggTransType == 
INTERNALOID) && (finalfn != serialfn)))
588                                 elog(ERROR, "%s is not its own 
aggpartialfunc", aggName);
589                 } else {

Here string comparison of aggName and aggpartialfnName looks very 
suspicios - it seems you should compare oids, not names (in this case,
likely oids of transition function and partial aggregation function). 
The fact that aggregate name matches partial aggregation function name
is not a enough to make any decisions.

In documentation

doc/src/sgml/postgres-fdw.sgml:

  930    <filename>postgres_fdw</filename> attempts to optimize remote 
queries to reduce
  931    the amount of data transferred from foreign servers.  This is 
done by
  932    sending query <literal>WHERE</literal> clauses and ggregate 
expressions
  933    to the remote server for execution, and by not retrieving table 
columns that
  934    are not needed for the current query.
  935    To reduce the risk of misexecution of queries,
  936    <literal>WHERE</literal> clauses and ggregate expressions are 
not sent to
  937    the remote server unless they use only data types, operators, 
and functions
  938    that are built-in or belong to an extension that's listed in the 
foreign
  939    server's <literal>extensions</literal> option.
  940    Operators and functions in such clauses must
  941    be <literal>IMMUTABLE</literal> as well.

there are misprints in lines 932 and 936 - missing "a" in "aggregate" 
expressions.

Note that after these changes "select sum()" will fail for certain 
cases, when remote server version is not the latest. In other cases we 
tried
to preserve compatibility. Should we have a switch for a foreign server 
to turn this optimization off? Or do we just state that users
should use different workarounds if remote server version doesn't match 
local one?

-- 
Best regards,
Alexander Pyhalov,
Postgres Professional

---

From: Bruce Momjian <[email-masked]>
Date: Mon, 5 Jun 2023 20:10:18 -0400

> and my update to them in this version of the patch.

This thread started in October 2021 so I would like to explain what this
feature adds.

Basically for partitions made up of postgres_fdw tables, there are four
possible optimizations:

1.  Pruning, 3 stages, see slide 30 here:

	https://momjian.us/main/writings/pgsql/partitioning.pdf#page=30

2.  Parallelism across partitions, see slide 38 here:

	https://momjian.us/main/writings/pgsql/beyond.pdf#page=38

3.  Pushdown of partition-wise joins and aggregates, see slide 43 here:

	https://momjian.us/main/writings/pgsql/partitioning.pdf#page=43

4.  Pushdown of aggregates that aren't partition-wise

As far as I know, over the years we have accomplished all of these
items, except for #4.  #3 involves aggregates where the GROUP BY or
JOINed tables match the partition keys.

Number 4 involves things like a SUM our COUNT that does not match the
partition key, or has no groupings at all.

#3 is easier than #4 since we just need to pass _rows_ back from the
foreign servers.  #4 is more complex because _partial_ count/sum, or
even average values must be passed from the foreign servers to the
requesting server.

The good news is that we already have partial aggregate support as part
of our parallel aggregate feature, see:

	https://momjian.us/main/writings/pgsql/beyond.pdf#page=38

What the patch does is to expand the existing partial aggregate code to
allow partial aggregate results to pass from the foreign servers to the
requesting server.  This feature will be very useful for data warehouse
queries that need to compute aggregate across partitions.

  Only you can decide what is important to you.

---

From: "[email-masked]" <[email-masked]>
Date: Tue, 6 Jun 2023 03:08:50 +0000

Hi Mr.Pyhalov.

Thank you for your always thoughtful review.

> From: Alexander Pyhalov <[email-masked]>
> Sent: Monday, June 5, 2023 6:00 PM
> Have found one issue -
> 

I see no problem with this string comparison. Here is the reason.
The intent of this code is, to determine whether the user specifies 
the new aggregate function whose aggpartialfunc is itself.
For two aggregate functions,
If the argument list and function name match, then the two aggregate functions must match.
By definition of aggpartialfunc,
every aggregate function and its aggpartialfn must have the same argument list.
Thus, if aggpartialfnName and aggName are equal as strings,
I think it is correct to determine that the user is specifying 
the new aggregate function whose aggpartialfunc is itself.

However, since the document does not state these intentions
I think your suspicions are valid.
Therefore, I have added a specification to the document reflecting the above intentions.

> From: Alexander Pyhalov <[email-masked]>
> Sent: Monday, June 5, 2023 6:00 PM
> In documentation
> 

Fixed.

> From: Alexander Pyhalov <[email-masked]>
> Sent: Monday, June 5, 2023 6:00 PM

It is the latter.
I added description about the above limitation to F.38.6. Built-in sharding in PostgreSQL and
F.38.8 Cross-Version Compatibility of doc/src/sgml/postgres-fdw.sgml.

> From: Alexander Pyhalov <[email-masked]>
> Sent: Tuesday, June 6, 2023 3:15 AM
> Bruce Momjian писал 2023-06-05 19:26:
> > On Mon, Jun  5, 2023 at 12:00:27PM +0300, Alexander Pyhalov wrote:
> >> cases, when remote server version is not the latest. In other cases
> >> we tried to preserve compatibility. Should we have a switch for a
> >> foreign server to turn this optimization off? Or do we just state
> >> that users should use different workarounds if remote server version
> >> doesn't match local one?
> >
> > We covered this in April in this and previous emails:
> >
> >
> 	https://www.postgresql.org/message-id/ZDGTza4rovCa%2BN3d%40
> momjian.us
> >
> > We don't check the version number for _any_ builtin functions so why
> > would we need to check for aggregate pushdown?  Yes, these will be new
> > functions in PG 17, we have added functions regularly in major
> > releases and have never heard reports of problems about that.
> >
> Hi.
> 
> I've seen this message. But introduction of new built-in function will break
> requests to old servers only if this new function is used in the request (this
> means that query changes). However, this patch changes the behavior of old
> queries, which worked prior to update. This seems to be different to me.

You are right.
However, for now, partial aggregates pushdown is mainly used when using built-in sharding in PostgreSQL.
I believe when using built-in sharding in PostgreSQL, the version of the primary node server and
the version of the remote server will usually be the same.
So I think it would be sufficient to include in the documentation a note about such problem
and a workaround for them.

Sincerely yours,
Yuuki Fujii

---

From: Alexander Pyhalov <[email-masked]>
Date: Tue, 06 Jun 2023 07:19:01 +0300

> Therefore, I have added a specification to the document reflecting the

Hi. Let me explain.

Look at this example, taken from test.

CREATE AGGREGATE udf_avg_p_int4(int4) (
        sfunc = int4_avg_accum,
        stype = _int8,
        combinefunc = int4_avg_combine,
        initcond = '{0,0}'
);
CREATE AGGREGATE udf_sum(int4) (
        sfunc = int4_avg_accum,
        stype = _int8,
        finalfunc = int8_avg,
        combinefunc = int4_avg_combine,
        initcond = '{0,0}',
        aggpartialfunc = udf_avg_p_int4
);

Now, let's create another aggregate.

# create schema test ;
create aggregate test.udf_avg_p_int4(int4) (
        sfunc = int4_avg_accum,
        stype = _int8,
        finalfunc = int8_avg,
        combinefunc = int4_avg_combine,
        initcond = '{0,0}',
        aggpartialfunc = udf_avg_p_int4
);
ERROR:  udf_avg_p_int4 is not its own aggpartialfunc

What's the difference between test.udf_avg_p_int4(int4) aggregate and 
udf_sum(int4)? They are essentially the same, but second one can't be 
defined.

Also note difference:

# CREATE AGGREGATE udf_sum(int4) (
        sfunc = int4_avg_accum,
        stype = _int8,
        finalfunc = int8_avg,
        combinefunc = pg_catalog.int4_avg_combine,
        initcond = '{0,0}',
        aggpartialfunc = udf_avg_p_int4
);
CREATE AGGREGATE

# CREATE AGGREGATE udf_sum(int4) (
        sfunc = int4_avg_accum,
        stype = _int8,
        finalfunc = int8_avg,
        combinefunc = pg_catalog.int4_avg_combine,
        initcond = '{0,0}',
        aggpartialfunc = public.udf_avg_p_int4
);
ERROR:  aggpartialfnName is invalid

It seems that assumption about aggpartialfnName - that it's a 
non-qualified name is incorrect. And if we use qualified names, we can't 
compare just names, likely we should compare oids.

-- 
Best regards,
Alexander Pyhalov,
Postgres Professional

---

From: "[email-masked]" <[email-masked]>
Date: Tue, 6 Jun 2023 12:31:20 +0000

Hi Mr.Pyhalov.

Thank you for comments.

> Sent: Tuesday, June 6, 2023 1:19 PM
> >> Have found one issue -
> >>
> >> suspicios - it seems you should compare oids, not names (in this
> >> case, likely oids of transition function and partial aggregation function).
> >> The fact that aggregate name matches partial aggregation function
> >> name is not a enough to make any decisions.
> >
> > Thus, if aggpartialfnName and aggName are equal as strings, I think it
> > is correct to determine that the user is specifying the new aggregate
> > function whose aggpartialfunc is itself.
> >
> > However, since the document does not state these intentions I think
> > your suspicions are valid.
>         aggpartialfunc = public.udf_avg_p_int4 );
> It seems that assumption about aggpartialfnName - that it's a non-qualified
> name is incorrect. And if we use qualified names, we can't compare just names,
> likely we should compare oids.

Thanks for the explanation.
I understand that the method of comparing two function name strings is incorrect.
Instead, I added the parameter isaggpartialfunc indicating whether the aggregate
function and its aggpartialfunc are the same or different.

Sincerely yours,
Yuuki Fujii

---

From: Alexander Pyhalov <[email-masked]>
Date: Wed, 07 Jun 2023 12:47:01 +0300

Hi.

This seems to be more robust, but the interface became more strange.
I'm not sure what to do with it. Some ideas I had to avoid introducing 
this parameter. Not sure I like any of them.

1) You can use QualifiedNameGetCreationNamespace() for aggpartialfnName 
and still compare namespace and function name  for it and  aggName, 
aggNamespace.
Seems to be not ideal, but avoids introducing new parameters.

2) You can lookup for partial aggregate function after ProcedureCreate() 
in AggregateCreate(), if it wasn't found at earlier stages. If it is the 
aggregate itself - check it. If it's still not found, error out. Also 
seems to be a bit ugly - you leave uncommitted garbage for vacuum in 
catalogue.

Another issue - the patch misses recording dependency between 
aggpartialfn and aggregate procedure.

-- 
Best regards,
Alexander Pyhalov,
Postgres Professional

---

From: "[email-masked]" <[email-masked]>
Date: Wed, 7 Jun 2023 23:08:54 +0000

Hi Mr.Pyhalov.

Thank you for comments.

> Sent: Wednesday, June 7, 2023 6:47 PM
> I'm not sure what to do with it. Some ideas I had to avoid introducing this
> parameter. Not sure I like any of them.
> 
> 2) You can lookup for partial aggregate function after ProcedureCreate() in
> AggregateCreate(), if it wasn't found at earlier stages. If it is the aggregate itself
> - check it. If it's still not found, error out. Also seems to be a bit ugly - you leave
> uncommitted garbage for vacuum in catalogue.
Thank you for suggesting alternatives.
The disadvantages of alternative 2) appear to be undesirable, 
I have modified it according to alternative 1)

> Another issue - the patch misses recording dependency between aggpartialfn
> and aggregate procedure.
I added code to record dependencys between aggpartialfn
and aggregate procedure, similar to the code for functions such as combinefunc.

Sincerely yours,
Yuuki Fujii

---

From: Alexander Pyhalov <[email-masked]>
Date: Thu, 08 Jun 2023 10:39:30 +0300

Hi.

Looks better. The only question I have is should we record dependency 
between procOid and aggpartialfn if aggpartialfn == procOid.

Also it seems new code likely should be run through pgindent.

+   For <literal>WHERE</literal> clauses,
+   <literal>JOIN</literal> clauses, this sending is active if
+   conditions in <xref 
linkend="postgres-fdw-remote-query-optimization"/>
+   hold and <varname>enable_partitionwise_join</varname> is true(this 
condition
+   is need for only <literal>JOIN</literal> clauses).
+   For aggregate expressions, this sending is active if conditions in

No space between "true" and "(" in "is true(this condition".

Some sentences in documentation, like one starting with
"For aggregate expressions, this sending is active if conditions in..."
seem to be too long, but I'm not the best man to read out documentation.

In "Built-in sharding in PostgreSQL" term "shard" doesn't have a 
definition.

By the way, I'm not sure that "sharding" documentation belongs to this 
patch,
at least it needs a review from native speaker.

-- 
Best regards,
Alexander Pyhalov,
Postgres Professional

---

From: "[email-masked]" <[email-masked]>
Date: Fri, 9 Jun 2023 11:38:37 +0000

Hi Mr.Pyhalov.

Thank you for comments.

> Sent: Thursday, June 8, 2023 4:40 PM
Fixed.

> Sent: Thursday, June 8, 2023 4:40 PM
Done.

> Sent: Thursday, June 8, 2023 4:40 PM
Fixed.

> Sent: Thursday, June 8, 2023 4:40 PM
> Some sentences in documentation, like one starting with "For aggregate
> expressions, this sending is active if conditions in..."
Fixed.

> Sent: Thursday, June 8, 2023 4:40 PM
> In "Built-in sharding in PostgreSQL" term "shard" doesn't have a definition.
I have removed the sentence you pointed out.

> Sent: Thursday, June 8, 2023 4:40 PM
> By the way, I'm not sure that "sharding" documentation belongs to this patch, at
> least it needs a review from native speaker.
I removed general description of sharding.

Sincerely yours,
Yuuki Fujii

---

From: Alexander Pyhalov <[email-masked]>
Date: Fri, 09 Jun 2023 17:09:42 +0300

Hi.

+       An aggregate function, called the partial aggregate function for 
partial aggregate
+       that corresponding to the aggregate function, is defined on the 
primary node and
+       the <filename>postgres_fdw</filename> node.

Something is clearly wrong here.

+   When using built-in sharding feature in PostgreSQL is used,

And here.

Overall the code looks good to me, but I suppose that documentation 
needs further review from some native speaker.

-- 
Best regards,
Alexander Pyhalov,
Postgres Professional

---

From: Bruce Momjian <[email-masked]>
Date: Fri, 9 Jun 2023 12:44:03 -0400

> and a workaround for them.

I agree that this feature is designed for built-in sharding, but it is
possible people could be using aggregates on partitions backed by
foreign tables without sharding.  Adding a requirement for non-sharding
setups to need PG 17+ servers might be unreasonable.

Looking at previous release note incompatibilities, we don't normally
change non-administrative functions in a way that causes errors if run
on older servers.  Based on Alexander's observations, I wonder if we
need to re-add the postgres_fdw option to control partial aggregate
pushdown, and default it to enabled.

If we ever add more function breakage we might need more postgres_fdw
options.  Fortunately, such changes are rare.

Yuki, thank you for writing and updating this patch, and Alexander,
thank you for helping with this patch.