Skip to content

Commit

Permalink
[#19839] YSQL: Pushdown partial aggregate
Browse files Browse the repository at this point in the history 
Summary:
Allow to pushdown aggregates where split is AGGSPLIT_INITIAL_SERIAL.

Backgrownd: there are 3 types of aggregate split:
- AGGSPLIT_SIMPLE - no split, aggregation evaluates both phase 1 and
  phase 2;
- AGGSPLIT_INITIAL_SERIAL - inner part of the split, aggregation
  evaluates phase 1 only;
- AGGSPLIT_FINAL_DESERIAL - outer part f the split, aggregation
  evaluates phase 2 only.

We previously only allowed to pushdown aggregate functions with no split,
however we can also safely pushdown AGGSPLIT_INITIAL_SERIAL functions,
since the phase 1 is the part of the aggregate function which is actually
being pushed down.

If a Partial Aggregate is pushed down, the plan node actually does
nothing, but forward phase 1 results from the YB scan to upper level,
so indicate that fact in the explain output by labeling the aggregate
node with pushdown as "Noop Aggregate" instead of "Partial Aggregate".
Jira: DB-8771

Test Plan: ybd --java-test 'org.yb.pgsql.TestPgRegressParallel#testPgRegressParallel'

Reviewers: jason, tnayak

Reviewed By: tnayak

Subscribers: yql

Tags: #jenkins-ready

Differential Revision: https://phorge.dev.yugabyte.com/D30579
  • Loading branch information
@andrei-mart
andrei-mart committed Nov 29, 2023
1 parent 51e0dae commit fd5136c
Show file tree
Hide file tree
Showing 4 changed files with 57 additions and 35 deletions.
22 changes: 15 additions & 7 deletions src/postgres/src/backend/commands/explain.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2003,7 +2003,15 @@ ExplainNode(PlanState *planstate, List *ancestors,

if (DO_AGGSPLIT_SKIPFINAL(agg->aggsplit))
{
partialmode = "Partial";
if (((AggState*) planstate)->yb_pushdown_supported)
/*
* If partial aggregate is pushed down, it does not
* really do anything, since entire operation is
* delegated to DocDB.
*/
partialmode = "Noop";
else
partialmode = "Partial";
pname = psprintf("%s %s", partialmode, pname);
}
else if (DO_AGGSPLIT_COMBINE(agg->aggsplit) ||
Expand Down Expand Up @@ -2384,10 +2392,10 @@ ExplainNode(PlanState *planstate, List *ancestors,
if (es->debug && yb_enable_base_scans_cost_model)
{
ExplainPropertyFloat(
"Estimated Seeks", NULL,
"Estimated Seeks", NULL,
((IndexScan *) plan)->yb_estimated_num_seeks, 0, es);
ExplainPropertyFloat(
"Estimated Nexts", NULL,
"Estimated Nexts", NULL,
((IndexScan *) plan)->yb_estimated_num_nexts, 0, es);
}
break;
Expand Down Expand Up @@ -2423,10 +2431,10 @@ ExplainNode(PlanState *planstate, List *ancestors,
if (es->debug && yb_enable_base_scans_cost_model)
{
ExplainPropertyFloat(
"Estimated Seeks", NULL,
"Estimated Seeks", NULL,
((IndexOnlyScan *) plan)->yb_estimated_num_seeks, 0, es);
ExplainPropertyFloat(
"Estimated Nexts", NULL,
"Estimated Nexts", NULL,
((IndexOnlyScan *) plan)->yb_estimated_num_nexts, 0, es);
}
break;
Expand Down Expand Up @@ -2480,10 +2488,10 @@ ExplainNode(PlanState *planstate, List *ancestors,
if (es->debug && yb_enable_base_scans_cost_model)
{
ExplainPropertyFloat(
"Estimated Seeks", NULL,
"Estimated Seeks", NULL,
((YbSeqScan *) plan)->yb_estimated_num_seeks, 0, es);
ExplainPropertyFloat(
"Estimated Nexts", NULL,
"Estimated Nexts", NULL,
((YbSeqScan *) plan)->yb_estimated_num_nexts, 0, es);
}
break;
Expand Down
2 changes: 1 addition & 1 deletion src/postgres/src/backend/executor/nodeAgg.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1627,7 +1627,7 @@ yb_agg_pushdown_supported(AggState *aggstate)
return;

/* Simple split. */
if (aggref->aggsplit != AGGSPLIT_SIMPLE)
if (aggref->aggsplit == AGGSPLIT_FINAL_DESERIAL)
return;


Expand Down
25 changes: 15 additions & 10 deletions src/postgres/src/test/regress/expected/yb_parallel_colocated.out
View file
Original file line number Diff line number Diff line change
Expand Up @@ -54,10 +54,11 @@ SELECT count(*) FROM pctest1 WHERE d LIKE 'Value_9';
Finalize Aggregate
-> Gather
Workers Planned: 2
-> Partial Aggregate
-> Noop Aggregate
-> Parallel Seq Scan on pctest1
Remote Filter: (d ~~ 'Value_9'::text)
(6 rows)
Partial Aggregate: true
(7 rows)

SELECT count(*) FROM pctest1 WHERE d LIKE 'Value_9';
count
Expand Down Expand Up @@ -272,10 +273,11 @@ SELECT count(*) FROM pctest1 WHERE k > 123;
Finalize Aggregate
-> Gather
Workers Planned: 2
-> Partial Aggregate
-> Noop Aggregate
-> Parallel Index Scan using pctest1_pkey on pctest1
Index Cond: (k > 123)
(6 rows)
Partial Aggregate: true
(7 rows)

SELECT count(*) FROM pctest1 WHERE k > 123;
count
Expand Down Expand Up @@ -616,17 +618,18 @@ SELECT * from pctest2
-> Finalize Aggregate
-> Gather
Workers Planned: 2
-> Partial Aggregate
-> Noop Aggregate
-> Parallel Seq Scan on pctest1 pctest1_1
Remote Filter: (d ~~ 'Value_9'::text)
Partial Aggregate: true
-> Parallel Hash Semi Join
Hash Cond: (pctest2.c = pctest1.b)
-> Parallel Seq Scan on pctest2
Remote Filter: ((b)::numeric < $1)
-> Parallel Hash
-> Parallel Seq Scan on pctest1
Remote Filter: (d ~~ 'Value_9'::text)
(17 rows)
(18 rows)

SELECT * from pctest2
WHERE c IN (SELECT b FROM pctest1 WHERE d LIKE 'Value_9')
Expand Down Expand Up @@ -668,10 +671,11 @@ select * from
-> Finalize Aggregate
-> Gather
Workers Planned: 2
-> Partial Aggregate
-> Noop Aggregate
-> Parallel Seq Scan on pctest1
Remote Filter: (b > 10)
(8 rows)
Partial Aggregate: true
(9 rows)

select * from
(SELECT count(*) FROM pctest1 WHERE b > 10) ss
Expand All @@ -694,10 +698,11 @@ select * from
-> Finalize Aggregate
-> Gather
Workers Planned: 2
-> Partial Aggregate
-> Noop Aggregate
-> Parallel Index Only Scan using pctest1_c_idx on pctest1
Index Cond: (c > 10)
(8 rows)
Partial Aggregate: true
(9 rows)

select * from
(SELECT count(*) FROM pctest1 WHERE c > 10) ss
Expand Down
43 changes: 26 additions & 17 deletions src/postgres/src/test/regress/expected/yb_select_parallel.out
View file
Original file line number Diff line number Diff line change
Expand Up @@ -46,10 +46,11 @@ explain (costs off)
Finalize Aggregate
-> Gather
Workers Planned: 4
-> Partial Aggregate
-> Noop Aggregate
-> Parallel Seq Scan on tenk1
Remote Filter: (stringu1 = 'GRAAAA'::name)
(6 rows)
Partial Aggregate: true
(7 rows)

select count(*) from tenk1 where stringu1 = 'GRAAAA';
count
Expand All @@ -67,10 +68,11 @@ explain (costs off)
Finalize Aggregate
-> Gather
Workers Planned: 4
-> Partial Aggregate
-> Noop Aggregate
-> Parallel Seq Scan on tenk1
Remote Filter: (stringu1 = 'GRAAAA'::name)
(6 rows)
Partial Aggregate: true
(7 rows)

select count(*) from tenk1 where stringu1 = 'GRAAAA';
count
Expand Down Expand Up @@ -154,10 +156,11 @@ explain (costs off) execute tenk1_count(1);
Finalize Aggregate
-> Gather
Workers Planned: 4
-> Partial Aggregate
-> Noop Aggregate
-> Parallel Seq Scan on tenk1
Remote Filter: (hundred > 1)
(6 rows)
Partial Aggregate: true
(7 rows)

execute tenk1_count(1);
count
Expand Down Expand Up @@ -217,15 +220,17 @@ explain (costs off)
-> Finalize Aggregate
-> Gather
Workers Planned: 4
-> Partial Aggregate
-> Noop Aggregate
-> Parallel Seq Scan on tenk2
Partial Aggregate: true
-> Gather
Workers Planned: 4
Params Evaluated: $2
-> Partial Aggregate
-> Noop Aggregate
-> Parallel Seq Scan on tenk1
Remote Filter: (unique1 = $2)
(13 rows)
Partial Aggregate: true
(15 rows)

select count(*) from tenk1
where tenk1.unique1 = (Select max(tenk2.unique1) from tenk2);
Expand All @@ -247,10 +252,11 @@ explain (costs off)
Finalize Aggregate
-> Gather
Workers Planned: 4
-> Partial Aggregate
-> Noop Aggregate
-> Parallel Index Scan using tenk1_hundred on tenk1
Index Cond: (hundred > 1)
(6 rows)
Partial Aggregate: true
(7 rows)

select count((unique1)) from tenk1 where hundred > 1;
count
Expand All @@ -266,10 +272,11 @@ explain (costs off)
Finalize Aggregate
-> Gather
Workers Planned: 4
-> Partial Aggregate
-> Noop Aggregate
-> Parallel Index Only Scan using tenk1_thous_tenthous on tenk1
Index Cond: (thousand > 95)
(6 rows)
Partial Aggregate: true
(7 rows)

select count(*) from tenk1 where thousand > 95;
count
Expand All @@ -290,10 +297,11 @@ select * from
-> Finalize Aggregate
-> Gather
Workers Planned: 4
-> Partial Aggregate
-> Noop Aggregate
-> Parallel Index Scan using tenk1_hundred on tenk1
Index Cond: (hundred > 10)
(8 rows)
Partial Aggregate: true
(9 rows)

select * from
(select count(unique1) from tenk1 where hundred > 10) ss
Expand All @@ -316,10 +324,11 @@ select * from
-> Finalize Aggregate
-> Gather
Workers Planned: 4
-> Partial Aggregate
-> Noop Aggregate
-> Parallel Index Only Scan using tenk1_thous_tenthous on tenk1
Index Cond: (thousand > 99)
(8 rows)
Partial Aggregate: true
(9 rows)

select * from
(select count(*) from tenk1 where thousand > 99) ss
Expand Down

0 comments on commit fd5136c

Please sign in to comment.