Hi.
Tomas Vondra писал 2021-10-15 17:56:
As for the proposed approach, it's probably good enough for the first
version to restrict this to aggregates where the aggregate result is
sufficient, i.e. we don't need any new export/import procedures.
But it's very unlikely we'd want to restrict it the way the patch does
it, i.e. based on aggregate name. That's both fragile (people can
create new aggregates with such name) and against the PostgreSQL
extensibility (people may implement custom aggregates, but won't be
able to benefit from this just because of name).
So for v0 maybe, but I think there neeeds to be a way to relax this in
some way, for example we could add a new flag to pg_aggregate to mark
aggregates supporting this.
Updated patch to mark aggregates as pushdown-safe in pg_aggregates.
So far have no solution for aggregates with internal aggtranstype.
--
Best regards,
Alexander Pyhalov,
Postgres Professional
From 823a389caf003a21dd4c8e758f89d08ba89c5856 Mon Sep 17 00:00:00 2001
From: Alexander Pyhalov <a.pyha...@postgrespro.ru>
Date: Thu, 14 Oct 2021 17:30:34 +0300
Subject: [PATCH] Partial aggregates push down
---
contrib/postgres_fdw/deparse.c | 45 +++-
.../postgres_fdw/expected/postgres_fdw.out | 215 +++++++++++++++++-
contrib/postgres_fdw/postgres_fdw.c | 29 ++-
contrib/postgres_fdw/sql/postgres_fdw.sql | 31 ++-
src/backend/catalog/pg_aggregate.c | 4 +-
src/backend/commands/aggregatecmds.c | 6 +-
src/include/catalog/catversion.h | 2 +-
src/include/catalog/pg_aggregate.dat | 101 ++++----
src/include/catalog/pg_aggregate.h | 6 +-
9 files changed, 362 insertions(+), 77 deletions(-)
diff --git a/contrib/postgres_fdw/deparse.c b/contrib/postgres_fdw/deparse.c
index d98bd666818..cf6b2d9f066 100644
--- a/contrib/postgres_fdw/deparse.c
+++ b/contrib/postgres_fdw/deparse.c
@@ -196,6 +196,7 @@ static bool is_subquery_var(Var *node, RelOptInfo *foreignrel,
static void get_relation_column_alias_ids(Var *node, RelOptInfo *foreignrel,
int *relno, int *colno);
+static bool partial_agg_ok(Aggref *agg);
/*
* Examine each qual clause in input_conds, and classify them into two groups,
@@ -831,8 +832,10 @@ 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))
return false;
/* As usual, it must be shippable. */
@@ -3249,7 +3252,7 @@ deparseAggref(Aggref *node, deparse_expr_cxt *context)
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;
@@ -3719,3 +3722,39 @@ get_relation_column_alias_ids(Var *node, RelOptInfo *foreignrel,
/* Shouldn't get here */
elog(ERROR, "unexpected expression in subquery output");
}
+
+/*
+ * Check that partial aggregate agg is fine to push down
+ */
+static bool
+partial_agg_ok(Aggref *agg)
+{
+ HeapTuple aggtup;
+ Form_pg_aggregate aggform;
+
+ Assert(agg->aggsplit == AGGSPLIT_INITIAL_SERIAL);
+
+ /* We don't support complex partial aggregates */
+ if (agg->aggdistinct || agg->aggvariadic || agg->aggkind != AGGKIND_NORMAL || agg->aggorder != NIL)
+ return false;
+
+ /* Can't process aggregates which require serialization/deserialization */
+ if (agg->aggtranstype == INTERNALOID)
+ return false;
+
+ aggtup = SearchSysCache1(AGGFNOID, ObjectIdGetDatum(agg->aggfnoid));
+ if (!HeapTupleIsValid(aggtup))
+ elog(ERROR, "cache lookup failed for function %u", agg->aggfnoid);
+ aggform = (Form_pg_aggregate) GETSTRUCT(aggtup);
+
+ /* Only aggregates, marked as pushdown safe, are allowed */
+ if (aggform->aggpartialpushdownsafe != true)
+ {
+ ReleaseSysCache(aggtup);
+ return false;
+ }
+
+ ReleaseSysCache(aggtup);
+
+ return true;
+}
diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out
index 44c4367b8f9..89451e208e0 100644
--- a/contrib/postgres_fdw/expected/postgres_fdw.out
+++ b/contrib/postgres_fdw/expected/postgres_fdw.out
@@ -9279,13 +9279,13 @@ RESET enable_partitionwise_join;
-- ===================================================================
-- test partitionwise aggregates
-- ===================================================================
-CREATE TABLE pagg_tab (a int, b int, c text) PARTITION BY RANGE(a);
+CREATE TABLE pagg_tab (a int, b int, c text, d numeric) 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');
@@ -9344,8 +9344,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
@@ -9356,21 +9356,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;
@@ -9406,6 +9406,203 @@ 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, max(a), count(*) FROM pagg_tab GROUP BY b 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
+ -> Append
+ -> Foreign Scan
+ Output: pagg_tab.b, (PARTIAL max(pagg_tab.a)), (PARTIAL count(*))
+ Relations: Aggregate on (public.fpagg_tab_p1 pagg_tab)
+ Remote SQL: SELECT b, max(a), count(*) FROM public.pagg_tab_p1 GROUP BY 1
+ -> Foreign Scan
+ Output: pagg_tab_1.b, (PARTIAL max(pagg_tab_1.a)), (PARTIAL count(*))
+ Relations: Aggregate on (public.fpagg_tab_p2 pagg_tab_1)
+ Remote SQL: SELECT b, max(a), count(*) FROM public.pagg_tab_p2 GROUP BY 1
+ -> Foreign Scan
+ Output: pagg_tab_2.b, (PARTIAL max(pagg_tab_2.a)), (PARTIAL count(*))
+ Relations: Aggregate on (public.fpagg_tab_p3 pagg_tab_2)
+ Remote SQL: SELECT b, max(a), count(*) FROM public.pagg_tab_p3 GROUP BY 1
+(19 rows)
+
+SELECT b, max(a), count(*) FROM pagg_tab GROUP BY b ORDER BY 1;
+ b | max | count
+----+-----+-------
+ 0 | 20 | 60
+ 1 | 21 | 60
+ 2 | 22 | 60
+ 3 | 23 | 60
+ 4 | 24 | 60
+ 5 | 25 | 60
+ 6 | 26 | 60
+ 7 | 27 | 60
+ 8 | 28 | 60
+ 9 | 29 | 60
+ 10 | 20 | 60
+ 11 | 21 | 60
+ 12 | 22 | 60
+ 13 | 23 | 60
+ 14 | 24 | 60
+ 15 | 25 | 60
+ 16 | 26 | 60
+ 17 | 27 | 60
+ 18 | 28 | 60
+ 19 | 29 | 60
+ 20 | 20 | 60
+ 21 | 21 | 60
+ 22 | 22 | 60
+ 23 | 23 | 60
+ 24 | 24 | 60
+ 25 | 25 | 60
+ 26 | 26 | 60
+ 27 | 27 | 60
+ 28 | 28 | 60
+ 29 | 29 | 60
+ 30 | 20 | 60
+ 31 | 21 | 60
+ 32 | 22 | 60
+ 33 | 23 | 60
+ 34 | 24 | 60
+ 35 | 25 | 60
+ 36 | 26 | 60
+ 37 | 27 | 60
+ 38 | 28 | 60
+ 39 | 29 | 60
+ 40 | 20 | 60
+ 41 | 21 | 60
+ 42 | 22 | 60
+ 43 | 23 | 60
+ 44 | 24 | 60
+ 45 | 25 | 60
+ 46 | 26 | 60
+ 47 | 27 | 60
+ 48 | 28 | 60
+ 49 | 29 | 60
+(50 rows)
+
+-- Partial aggregates are fine to push down
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT max(a), count(*) FROM pagg_tab;
+ QUERY PLAN
+---------------------------------------------------------------------------
+ Finalize Aggregate
+ Output: max(pagg_tab.a), count(*)
+ -> Append
+ -> Foreign Scan
+ Output: (PARTIAL max(pagg_tab.a)), (PARTIAL count(*))
+ Relations: Aggregate on (public.fpagg_tab_p1 pagg_tab)
+ Remote SQL: SELECT max(a), count(*) FROM public.pagg_tab_p1
+ -> Foreign Scan
+ Output: (PARTIAL max(pagg_tab_1.a)), (PARTIAL count(*))
+ Relations: Aggregate on (public.fpagg_tab_p2 pagg_tab_1)
+ Remote SQL: SELECT max(a), count(*) FROM public.pagg_tab_p2
+ -> Foreign Scan
+ Output: (PARTIAL max(pagg_tab_2.a)), (PARTIAL count(*))
+ Relations: Aggregate on (public.fpagg_tab_p3 pagg_tab_2)
+ Remote SQL: SELECT max(a), count(*) FROM public.pagg_tab_p3
+(15 rows)
+
+SELECT max(a), count(*) FROM pagg_tab;
+ max | count
+-----+-------
+ 29 | 3000
+(1 row)
+
+-- Shouldn't try to push down
+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)
+ -> Append
+ -> 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
+ -> 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
+ -> 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
+(12 rows)
+
+SELECT max(a), count(distinct b) FROM pagg_tab;
+ max | count
+-----+-------
+ 29 | 50
+(1 row)
+
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT sum(d) FROM pagg_tab;
+ QUERY PLAN
+------------------------------------------------------------------
+ Finalize Aggregate
+ Output: sum(pagg_tab.d)
+ -> Append
+ -> Partial Aggregate
+ Output: PARTIAL sum(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 sum(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 sum(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 sum(d) FROM pagg_tab;
+ sum
+-------
+ 58500
+(1 row)
+
-- ===================================================================
-- access rights and superuser
-- ===================================================================
diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c
index 45a09337d08..90e484bc640 100644
--- a/contrib/postgres_fdw/postgres_fdw.c
+++ b/contrib/postgres_fdw/postgres_fdw.c
@@ -517,7 +517,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, bool partial);
static List *get_useful_pathkeys_for_relation(PlannerInfo *root,
RelOptInfo *rel);
static List *get_useful_ecs_for_relation(PlannerInfo *root, RelOptInfo *rel);
@@ -526,7 +526,8 @@ static void add_paths_with_pathkeys_for_rel(PlannerInfo *root, RelOptInfo *rel,
static void add_foreign_grouping_paths(PlannerInfo *root,
RelOptInfo *input_rel,
RelOptInfo *grouped_rel,
- GroupPathExtraData *extra);
+ GroupPathExtraData *extra,
+ bool partial);
static void add_foreign_ordered_paths(PlannerInfo *root,
RelOptInfo *input_rel,
RelOptInfo *ordered_rel);
@@ -6083,7 +6084,7 @@ postgresGetForeignJoinPaths(PlannerInfo *root,
*/
static bool
foreign_grouping_ok(PlannerInfo *root, RelOptInfo *grouped_rel,
- Node *havingQual)
+ Node *havingQual, bool partial)
{
Query *query = root->parse;
PgFdwRelationInfo *fpinfo = (PgFdwRelationInfo *) grouped_rel->fdw_private;
@@ -6097,6 +6098,11 @@ foreign_grouping_ok(PlannerInfo *root, RelOptInfo *grouped_rel,
if (query->groupingSets)
return false;
+
+ /* It's unsafe to push having statements with partial aggregates */
+ if (partial && havingQual)
+ return false;
+
/* Get the fpinfo of the underlying scan relation. */
ofpinfo = (PgFdwRelationInfo *) fpinfo->outerrel->fdw_private;
@@ -6336,6 +6342,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)
@@ -6350,7 +6357,11 @@ postgresGetForeignUpperPaths(PlannerInfo *root, UpperRelationKind stage,
{
case UPPERREL_GROUP_AGG:
add_foreign_grouping_paths(root, input_rel, output_rel,
- (GroupPathExtraData *) extra);
+ (GroupPathExtraData *) extra, false);
+ break;
+ case UPPERREL_PARTIAL_GROUP_AGG:
+ add_foreign_grouping_paths(root, input_rel, output_rel,
+ (GroupPathExtraData *) extra, true);
break;
case UPPERREL_ORDERED:
add_foreign_ordered_paths(root, input_rel, output_rel);
@@ -6375,7 +6386,8 @@ postgresGetForeignUpperPaths(PlannerInfo *root, UpperRelationKind stage,
static void
add_foreign_grouping_paths(PlannerInfo *root, RelOptInfo *input_rel,
RelOptInfo *grouped_rel,
- GroupPathExtraData *extra)
+ GroupPathExtraData *extra,
+ bool partial)
{
Query *parse = root->parse;
PgFdwRelationInfo *ifpinfo = input_rel->fdw_private;
@@ -6391,8 +6403,9 @@ add_foreign_grouping_paths(PlannerInfo *root, RelOptInfo *input_rel,
!root->hasHavingQual)
return;
- Assert(extra->patype == PARTITIONWISE_AGGREGATE_NONE ||
- extra->patype == PARTITIONWISE_AGGREGATE_FULL);
+ Assert(((extra->patype == PARTITIONWISE_AGGREGATE_NONE ||
+ extra->patype == PARTITIONWISE_AGGREGATE_FULL) && !partial) ||
+ (extra->patype == PARTITIONWISE_AGGREGATE_PARTIAL && partial));
/* save the input_rel as outerrel in fpinfo */
fpinfo->outerrel = input_rel;
@@ -6412,7 +6425,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, partial))
return;
/*
diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql
index e7b869f8cea..63c0c03da37 100644
--- a/contrib/postgres_fdw/sql/postgres_fdw.sql
+++ b/contrib/postgres_fdw/sql/postgres_fdw.sql
@@ -2729,15 +2729,15 @@ RESET enable_partitionwise_join;
-- test partitionwise aggregates
-- ===================================================================
-CREATE TABLE pagg_tab (a int, b int, c text) PARTITION BY RANGE(a);
+CREATE TABLE pagg_tab (a int, b int, c text, d numeric) 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');
@@ -2771,6 +2771,29 @@ 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, max(a), count(*) FROM pagg_tab GROUP BY b ORDER BY 1;
+SELECT b, max(a), count(*) FROM pagg_tab GROUP BY b ORDER BY 1;
+
+-- Partial aggregates are fine to push down
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT max(a), count(*) FROM pagg_tab;
+SELECT max(a), count(*) FROM pagg_tab;
+
+-- Shouldn't try to push down
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT max(a), count(distinct b) FROM pagg_tab;
+SELECT max(a), count(distinct b) FROM pagg_tab;
+
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT sum(d) FROM pagg_tab;
+SELECT sum(d) FROM pagg_tab;
+
-- ===================================================================
-- access rights and superuser
-- ===================================================================
diff --git a/src/backend/catalog/pg_aggregate.c b/src/backend/catalog/pg_aggregate.c
index 1f63d8081b2..8c91c87a85b 100644
--- a/src/backend/catalog/pg_aggregate.c
+++ b/src/backend/catalog/pg_aggregate.c
@@ -74,7 +74,8 @@ AggregateCreate(const char *aggName,
int32 aggmTransSpace,
const char *agginitval,
const char *aggminitval,
- char proparallel)
+ char proparallel,
+ bool partialPushdownSafe)
{
Relation aggdesc;
HeapTuple tup;
@@ -676,6 +677,7 @@ AggregateCreate(const char *aggName,
values[Anum_pg_aggregate_aggtransspace - 1] = Int32GetDatum(aggTransSpace);
values[Anum_pg_aggregate_aggmtranstype - 1] = ObjectIdGetDatum(aggmTransType);
values[Anum_pg_aggregate_aggmtransspace - 1] = Int32GetDatum(aggmTransSpace);
+ values[Anum_pg_aggregate_aggpartialpushdownsafe - 1] = BoolGetDatum(partialPushdownSafe);
if (agginitval)
values[Anum_pg_aggregate_agginitval - 1] = CStringGetTextDatum(agginitval);
else
diff --git a/src/backend/commands/aggregatecmds.c b/src/backend/commands/aggregatecmds.c
index 046cf2df08f..53976487fec 100644
--- a/src/backend/commands/aggregatecmds.c
+++ b/src/backend/commands/aggregatecmds.c
@@ -74,6 +74,7 @@ DefineAggregate(ParseState *pstate,
List *mfinalfuncName = NIL;
bool finalfuncExtraArgs = false;
bool mfinalfuncExtraArgs = false;
+ bool partialPushdownSafe = false;
char finalfuncModify = 0;
char mfinalfuncModify = 0;
List *sortoperatorName = NIL;
@@ -189,6 +190,8 @@ DefineAggregate(ParseState *pstate,
minitval = defGetString(defel);
else if (strcmp(defel->defname, "parallel") == 0)
parallel = defGetString(defel);
+ else if (strcmp(defel->defname, "partial_pushdown_safe") == 0)
+ partialPushdownSafe = defGetBoolean(defel);
else
ereport(WARNING,
(errcode(ERRCODE_SYNTAX_ERROR),
@@ -471,7 +474,8 @@ DefineAggregate(ParseState *pstate,
mtransSpace, /* transition space */
initval, /* initial condition */
minitval, /* initial condition */
- proparallel); /* parallel safe? */
+ proparallel, /* parallel safe? */
+ partialPushdownSafe); /* partial pushdown safe? */
}
/*
diff --git a/src/include/catalog/catversion.h b/src/include/catalog/catversion.h
index 3253b8751b1..187c81b4857 100644
--- a/src/include/catalog/catversion.h
+++ b/src/include/catalog/catversion.h
@@ -53,6 +53,6 @@
*/
/* yyyymmddN */
-#define CATALOG_VERSION_NO 202109101
+#define CATALOG_VERSION_NO 202110191
#endif
diff --git a/src/include/catalog/pg_aggregate.dat b/src/include/catalog/pg_aggregate.dat
index fc6d3bfd945..e8df26e334c 100644
--- a/src/include/catalog/pg_aggregate.dat
+++ b/src/include/catalog/pg_aggregate.dat
@@ -60,22 +60,22 @@
{ aggfnoid => 'sum(int4)', aggtransfn => 'int4_sum', aggcombinefn => 'int8pl',
aggmtransfn => 'int4_avg_accum', aggminvtransfn => 'int4_avg_accum_inv',
aggmfinalfn => 'int2int4_sum', aggtranstype => 'int8',
- aggmtranstype => '_int8', aggminitval => '{0,0}' },
+ aggmtranstype => '_int8', aggminitval => '{0,0}', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'sum(int2)', aggtransfn => 'int2_sum', aggcombinefn => 'int8pl',
aggmtransfn => 'int2_avg_accum', aggminvtransfn => 'int2_avg_accum_inv',
aggmfinalfn => 'int2int4_sum', aggtranstype => 'int8',
- aggmtranstype => '_int8', aggminitval => '{0,0}' },
+ aggmtranstype => '_int8', aggminitval => '{0,0}', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'sum(float4)', aggtransfn => 'float4pl',
aggcombinefn => 'float4pl', aggtranstype => 'float4' },
{ aggfnoid => 'sum(float8)', aggtransfn => 'float8pl',
- aggcombinefn => 'float8pl', aggtranstype => 'float8' },
+ aggcombinefn => 'float8pl', aggtranstype => 'float8', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'sum(money)', aggtransfn => 'cash_pl', aggcombinefn => 'cash_pl',
aggmtransfn => 'cash_pl', aggminvtransfn => 'cash_mi',
- aggtranstype => 'money', aggmtranstype => 'money' },
+ aggtranstype => 'money', aggmtranstype => 'money', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'sum(interval)', aggtransfn => 'interval_pl',
aggcombinefn => 'interval_pl', aggmtransfn => 'interval_pl',
aggminvtransfn => 'interval_mi', aggtranstype => 'interval',
- aggmtranstype => 'interval' },
+ aggmtranstype => 'interval', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'sum(numeric)', aggtransfn => 'numeric_avg_accum',
aggfinalfn => 'numeric_sum', aggcombinefn => 'numeric_avg_combine',
aggserialfn => 'numeric_avg_serialize',
@@ -83,146 +83,149 @@
aggmtransfn => 'numeric_avg_accum', aggminvtransfn => 'numeric_accum_inv',
aggmfinalfn => 'numeric_sum', aggtranstype => 'internal',
aggtransspace => '128', aggmtranstype => 'internal',
- aggmtransspace => '128' },
+ aggmtransspace => '128', aggpartialpushdownsafe => 't' },
# max
{ aggfnoid => 'max(int8)', aggtransfn => 'int8larger',
aggcombinefn => 'int8larger', aggsortop => '>(int8,int8)',
- aggtranstype => 'int8' },
+ aggtranstype => 'int8', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'max(int4)', aggtransfn => 'int4larger',
aggcombinefn => 'int4larger', aggsortop => '>(int4,int4)',
- aggtranstype => 'int4' },
+ aggtranstype => 'int4', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'max(int2)', aggtransfn => 'int2larger',
aggcombinefn => 'int2larger', aggsortop => '>(int2,int2)',
- aggtranstype => 'int2' },
+ aggtranstype => 'int2', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'max(oid)', aggtransfn => 'oidlarger',
aggcombinefn => 'oidlarger', aggsortop => '>(oid,oid)',
- aggtranstype => 'oid' },
+ aggtranstype => 'oid', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'max(float4)', aggtransfn => 'float4larger',
aggcombinefn => 'float4larger', aggsortop => '>(float4,float4)',
- aggtranstype => 'float4' },
+ aggtranstype => 'float4', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'max(float8)', aggtransfn => 'float8larger',
aggcombinefn => 'float8larger', aggsortop => '>(float8,float8)',
- aggtranstype => 'float8' },
+ aggtranstype => 'float8', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'max(date)', aggtransfn => 'date_larger',
aggcombinefn => 'date_larger', aggsortop => '>(date,date)',
- aggtranstype => 'date' },
+ aggtranstype => 'date', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'max(time)', aggtransfn => 'time_larger',
aggcombinefn => 'time_larger', aggsortop => '>(time,time)',
- aggtranstype => 'time' },
+ aggtranstype => 'time', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'max(timetz)', aggtransfn => 'timetz_larger',
aggcombinefn => 'timetz_larger', aggsortop => '>(timetz,timetz)',
- aggtranstype => 'timetz' },
+ aggtranstype => 'timetz', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'max(money)', aggtransfn => 'cashlarger',
aggcombinefn => 'cashlarger', aggsortop => '>(money,money)',
- aggtranstype => 'money' },
+ aggtranstype => 'money', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'max(timestamp)', aggtransfn => 'timestamp_larger',
aggcombinefn => 'timestamp_larger', aggsortop => '>(timestamp,timestamp)',
- aggtranstype => 'timestamp' },
+ aggtranstype => 'timestamp', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'max(timestamptz)', aggtransfn => 'timestamptz_larger',
aggcombinefn => 'timestamptz_larger',
aggsortop => '>(timestamptz,timestamptz)', aggtranstype => 'timestamptz' },
{ aggfnoid => 'max(interval)', aggtransfn => 'interval_larger',
aggcombinefn => 'interval_larger', aggsortop => '>(interval,interval)',
- aggtranstype => 'interval' },
+ aggtranstype => 'interval', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'max(text)', aggtransfn => 'text_larger',
aggcombinefn => 'text_larger', aggsortop => '>(text,text)',
- aggtranstype => 'text' },
+ aggtranstype => 'text', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'max(numeric)', aggtransfn => 'numeric_larger',
aggcombinefn => 'numeric_larger', aggsortop => '>(numeric,numeric)',
- aggtranstype => 'numeric' },
+ aggtranstype => 'numeric', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'max(anyarray)', aggtransfn => 'array_larger',
aggcombinefn => 'array_larger', aggsortop => '>(anyarray,anyarray)',
- aggtranstype => 'anyarray' },
+ aggtranstype => 'anyarray', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'max(bpchar)', aggtransfn => 'bpchar_larger',
aggcombinefn => 'bpchar_larger', aggsortop => '>(bpchar,bpchar)',
- aggtranstype => 'bpchar' },
+ aggtranstype => 'bpchar', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'max(tid)', aggtransfn => 'tidlarger',
aggcombinefn => 'tidlarger', aggsortop => '>(tid,tid)',
- aggtranstype => 'tid' },
+ aggtranstype => 'tid', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'max(anyenum)', aggtransfn => 'enum_larger',
aggcombinefn => 'enum_larger', aggsortop => '>(anyenum,anyenum)',
- aggtranstype => 'anyenum' },
+ aggtranstype => 'anyenum', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'max(inet)', aggtransfn => 'network_larger',
aggcombinefn => 'network_larger', aggsortop => '>(inet,inet)',
- aggtranstype => 'inet' },
+ aggtranstype => 'inet', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'max(pg_lsn)', aggtransfn => 'pg_lsn_larger',
aggcombinefn => 'pg_lsn_larger', aggsortop => '>(pg_lsn,pg_lsn)',
- aggtranstype => 'pg_lsn' },
+ aggtranstype => 'pg_lsn', aggpartialpushdownsafe => 't' },
# min
{ aggfnoid => 'min(int8)', aggtransfn => 'int8smaller',
aggcombinefn => 'int8smaller', aggsortop => '<(int8,int8)',
- aggtranstype => 'int8' },
+ aggtranstype => 'int8', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'min(int4)', aggtransfn => 'int4smaller',
aggcombinefn => 'int4smaller', aggsortop => '<(int4,int4)',
- aggtranstype => 'int4' },
+ aggtranstype => 'int4', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'min(int2)', aggtransfn => 'int2smaller',
aggcombinefn => 'int2smaller', aggsortop => '<(int2,int2)',
- aggtranstype => 'int2' },
+ aggtranstype => 'int2', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'min(oid)', aggtransfn => 'oidsmaller',
aggcombinefn => 'oidsmaller', aggsortop => '<(oid,oid)',
- aggtranstype => 'oid' },
+ aggtranstype => 'oid', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'min(float4)', aggtransfn => 'float4smaller',
aggcombinefn => 'float4smaller', aggsortop => '<(float4,float4)',
- aggtranstype => 'float4' },
+ aggtranstype => 'float4', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'min(float8)', aggtransfn => 'float8smaller',
aggcombinefn => 'float8smaller', aggsortop => '<(float8,float8)',
- aggtranstype => 'float8' },
+ aggtranstype => 'float8', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'min(date)', aggtransfn => 'date_smaller',
aggcombinefn => 'date_smaller', aggsortop => '<(date,date)',
- aggtranstype => 'date' },
+ aggtranstype => 'date', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'min(time)', aggtransfn => 'time_smaller',
aggcombinefn => 'time_smaller', aggsortop => '<(time,time)',
- aggtranstype => 'time' },
+ aggtranstype => 'time', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'min(timetz)', aggtransfn => 'timetz_smaller',
aggcombinefn => 'timetz_smaller', aggsortop => '<(timetz,timetz)',
- aggtranstype => 'timetz' },
+ aggtranstype => 'timetz', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'min(money)', aggtransfn => 'cashsmaller',
aggcombinefn => 'cashsmaller', aggsortop => '<(money,money)',
- aggtranstype => 'money' },
+ aggtranstype => 'money', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'min(timestamp)', aggtransfn => 'timestamp_smaller',
aggcombinefn => 'timestamp_smaller', aggsortop => '<(timestamp,timestamp)',
- aggtranstype => 'timestamp' },
+ aggtranstype => 'timestamp', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'min(timestamptz)', aggtransfn => 'timestamptz_smaller',
aggcombinefn => 'timestamptz_smaller',
- aggsortop => '<(timestamptz,timestamptz)', aggtranstype => 'timestamptz' },
+ aggsortop => '<(timestamptz,timestamptz)', aggtranstype => 'timestamptz',
+ aggpartialpushdownsafe => 't' },
{ aggfnoid => 'min(interval)', aggtransfn => 'interval_smaller',
aggcombinefn => 'interval_smaller', aggsortop => '<(interval,interval)',
- aggtranstype => 'interval' },
+ aggtranstype => 'interval', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'min(text)', aggtransfn => 'text_smaller',
aggcombinefn => 'text_smaller', aggsortop => '<(text,text)',
- aggtranstype => 'text' },
+ aggtranstype => 'text', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'min(numeric)', aggtransfn => 'numeric_smaller',
aggcombinefn => 'numeric_smaller', aggsortop => '<(numeric,numeric)',
- aggtranstype => 'numeric' },
+ aggtranstype => 'numeric', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'min(anyarray)', aggtransfn => 'array_smaller',
aggcombinefn => 'array_smaller', aggsortop => '<(anyarray,anyarray)',
- aggtranstype => 'anyarray' },
+ aggtranstype => 'anyarray', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'min(bpchar)', aggtransfn => 'bpchar_smaller',
aggcombinefn => 'bpchar_smaller', aggsortop => '<(bpchar,bpchar)',
- aggtranstype => 'bpchar' },
+ aggtranstype => 'bpchar', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'min(tid)', aggtransfn => 'tidsmaller',
aggcombinefn => 'tidsmaller', aggsortop => '<(tid,tid)',
- aggtranstype => 'tid' },
+ aggtranstype => 'tid', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'min(anyenum)', aggtransfn => 'enum_smaller',
aggcombinefn => 'enum_smaller', aggsortop => '<(anyenum,anyenum)',
- aggtranstype => 'anyenum' },
+ aggtranstype => 'anyenum', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'min(inet)', aggtransfn => 'network_smaller',
aggcombinefn => 'network_smaller', aggsortop => '<(inet,inet)',
- aggtranstype => 'inet' },
+ aggtranstype => 'inet', aggpartialpushdownsafe => 't' },
{ aggfnoid => 'min(pg_lsn)', aggtransfn => 'pg_lsn_smaller',
aggcombinefn => 'pg_lsn_smaller', aggsortop => '<(pg_lsn,pg_lsn)',
- aggtranstype => 'pg_lsn' },
+ aggtranstype => 'pg_lsn' , aggpartialpushdownsafe => 't'},
# count
{ aggfnoid => 'count(any)', aggtransfn => 'int8inc_any',
aggcombinefn => 'int8pl', aggmtransfn => 'int8inc_any',
aggminvtransfn => 'int8dec_any', aggtranstype => 'int8',
- aggmtranstype => 'int8', agginitval => '0', aggminitval => '0' },
+ aggmtranstype => 'int8', agginitval => '0', aggminitval => '0',
+ aggpartialpushdownsafe => 't' },
{ aggfnoid => 'count()', aggtransfn => 'int8inc', aggcombinefn => 'int8pl',
aggmtransfn => 'int8inc', aggminvtransfn => 'int8dec', aggtranstype => 'int8',
- aggmtranstype => 'int8', agginitval => '0', aggminitval => '0' },
+ aggmtranstype => 'int8', agginitval => '0', aggminitval => '0',
+ aggpartialpushdownsafe => 't' },
# var_pop
{ aggfnoid => 'var_pop(int8)', aggtransfn => 'int8_accum',
diff --git a/src/include/catalog/pg_aggregate.h b/src/include/catalog/pg_aggregate.h
index 08c9379ba77..1138b64d478 100644
--- a/src/include/catalog/pg_aggregate.h
+++ b/src/include/catalog/pg_aggregate.h
@@ -91,6 +91,9 @@ CATALOG(pg_aggregate,2600,AggregateRelationId)
/* estimated size of moving-agg state (0 for default est) */
int32 aggmtransspace BKI_DEFAULT(0);
+ /* true if partial aggregate is fine to push down */
+ bool aggpartialpushdownsafe BKI_DEFAULT(f);
+
#ifdef CATALOG_VARLEN /* variable-length fields start here */
/* initial value for transition state (can be NULL) */
@@ -175,6 +178,7 @@ extern ObjectAddress AggregateCreate(const char *aggName,
int32 aggmTransSpace,
const char *agginitval,
const char *aggminitval,
- char proparallel);
+ char proparallel,
+ bool partialPushdownSafe);
#endif /* PG_AGGREGATE_H */
--
2.25.1
