pg.ddx.io pgsql-hackers@postgresql.org mailing list archive help / color / mirror / Atom feed From: Tomas Vondra <tomas.vondra@2ndquadrant.com>
To: Pengzhou Tang <ptang@pivotal.io>
Cc: Richard Guo <guofenglinux@gmail.com>
Cc: Jesse Zhang <sbjesse@gmail.com>
Cc: Richard Guo <riguo@pivotal.io>
Cc: Amit Kapila <amit.kapila16@gmail.com>
Cc: Michael Paquier <michael@paquier.xyz>
Cc: PostgreSQL-development <pgsql-hackers@postgresql.org>
Cc: Andres Freund <andres@anarazel.de>
Cc: Andrew Gierth <andrew@tao11.riddles.org.uk>
Subject: Re: Parallel grouping sets
Date: Thu, 19 Mar 2020 03:09:14 +0100
Message-ID: <20200319020914.zkcjpm5r4yj7umqx@development> (raw )
In-Reply-To: <CAG4reAQ8rFCc+i0oju3VjaW7xSOJAkvLrqa4F-NYZzAG4SW7iQ@mail.gmail.com >
References: <20191201020257.GJ2355@paquier.xyz >
<CAN_9JTwtTTnxhbr5AHuqVcriz3HxvPpx1JWE--DCSdJYuHrLtA@mail.gmail.com >
<CAN_9JTx3NM12ZDzEYcOVLFiCBvwMHyM0gENvtTpKBoOOgcs=kw@mail.gmail.com >
<CAA4eK1L_T=4FVQ58rOFjoBLkGq2yvGRGG-CemsJX01PHExU4fg@mail.gmail.com >
<CAGf+fX5288z0Gt_UL9FYHvNcySvG1wrKxPQ7+zretbcaUDc5Jw@mail.gmail.com >
<CAN_9JTwtzttEmdXvMbJqXt=51kXiBTCKEPKq6kk2PZ6Xz6m5ig@mail.gmail.com >
<CAGf+fX4SDEqbPwT9_mfJLDXkBKQTT6O-wAHeKvUAViE+56n2Gw@mail.gmail.com >
<CAG4reASx+p3z5W59O57xhPC57Z4MW3mSnE=s-MJGromnmgg2fA@mail.gmail.com >
<CAMbWs4-fjjfnHNJA9YbeDcku5KAYN_LTVEPOaTrKLioLK23d1w@mail.gmail.com >
<CAG4reAQ8rFCc+i0oju3VjaW7xSOJAkvLrqa4F-NYZzAG4SW7iQ@mail.gmail.com >
Hi,
unfortunately this got a bit broken by the disk-based hash aggregation,
committed today, and so it needs a rebase. I've started looking at the
patch before that, and I have it rebased on e00912e11a9e (i.e. the
commit before the one that breaks it).
Attached is the rebased patch series (now broken), with a couple of
commits with some minor cosmetic changes I propose to make (easier than
explaining it on a list, it's mostly about whitespace, comments etc).
Feel free to reject the changes, it's up to you.
I'll continue doing the review, but it'd be good to have a fully rebased
version.
regards
--
Tomas Vondra http://www.2ndQuadrant.com
PostgreSQL Development, 24x7 Support, Remote DBA, Training & Services
Attachments: download all (7 files, .tar.gz)
0001-All-grouping-sets-do-their-own-sorting.patch (text/plain, 34.4K) download
From 7f932c2a2897a92f261b5ccfaea2c2b90823996c Mon Sep 17 00:00:00 2001
From: Pengzhou Tang <ptang@pivotal.io>
Date: Wed, 11 Mar 2020 23:07:29 -0400
Subject: [PATCH 1/7] All grouping sets do their own sorting
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
PG used to add a SORT path explicitly beneath the AGG for sort aggregate,
grouping sets path also add a SORT path for the first sort aggregate phase,
but the following sort aggregate phases do their own sorting using a tuplesort.
This commit unified the way how grouping sets path doing sort, all sort aggregate
phases now do their own sorting using tuplesort.
This commit is mainly a preparing step to support parallel grouping sets, the
main idea of parallel grouping sets is: like parallel aggregate, we separate
grouping sets into two stages:
The initial stage: this stage has almost the same plan and execution routines
with the current implementation of grouping sets, the differenceis are 1) it
only produces partial aggregate results 2) the output is attached with an extra
grouping set id. We know partial aggregate results will be combined in the final
stage and we have multiple grouping sets, so only partial aggregate results
belong to the same grouping set can be combined, that is why grouping set id is
introduced to identify the sets. We keep all the optimizations of multiple
grouping sets in the initial stage, eg, 1) the grouping sets (that can be
grouped by one single sort) are put into the one rollup structure so those sets
arecomputed in one aggregate phase. 2) do hash aggregate concurrently when a
sort aggregate is performed. 3) do all hash transitions in one expression state.
The final stage: this stage combine the partial aggregate results according to
the grouping set id. Obviously, all the optimizations in the initial stage
cannot be used, so all rollups are extracted, each rollup contains only one
grouping set, then each aggregate phase only processes one set. We do a filter
in the final stage, redirect the tuples to each aggregate phase.
Obviously, adding a SORT path underneath the AGG in the final stage is not
right. This commit can avoid it and all non-hashed aggregate phases can do
their own sorting after thetuples are redirected.
---
src/backend/commands/explain.c | 5 +-
src/backend/executor/nodeAgg.c | 79 +++++++++++--
src/backend/nodes/copyfuncs.c | 1 +
src/backend/nodes/outfuncs.c | 1 +
src/backend/nodes/readfuncs.c | 1 +
src/backend/optimizer/plan/createplan.c | 65 ++++++++---
src/backend/optimizer/plan/planner.c | 66 +++++++----
src/backend/optimizer/util/pathnode.c | 30 ++++-
src/include/executor/nodeAgg.h | 2 -
src/include/nodes/execnodes.h | 5 +-
src/include/nodes/pathnodes.h | 1 +
src/include/nodes/plannodes.h | 1 +
src/include/optimizer/pathnode.h | 3 +-
src/include/optimizer/planmain.h | 2 +-
src/test/regress/expected/groupingsets.out | 130 ++++++++++-----------
15 files changed, 260 insertions(+), 132 deletions(-)
diff --git a/src/backend/commands/explain.c b/src/backend/commands/explain.c
index d901dc4a50..b1609b339a 100644
--- a/src/backend/commands/explain.c
+++ b/src/backend/commands/explain.c
@@ -2289,15 +2289,14 @@ show_grouping_sets(PlanState *planstate, Agg *agg,
ExplainOpenGroup("Grouping Sets", "Grouping Sets", false, es);
- show_grouping_set_keys(planstate, agg, NULL,
+ show_grouping_set_keys(planstate, agg, (Sort *) agg->sortnode,
context, useprefix, ancestors, es);
foreach(lc, agg->chain)
{
Agg *aggnode = lfirst(lc);
- Sort *sortnode = (Sort *) aggnode->plan.lefttree;
- show_grouping_set_keys(planstate, aggnode, sortnode,
+ show_grouping_set_keys(planstate, aggnode, (Sort *) aggnode->sortnode,
context, useprefix, ancestors, es);
}
diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c
index 7aebb247d8..b4f53bf77a 100644
--- a/src/backend/executor/nodeAgg.c
+++ b/src/backend/executor/nodeAgg.c
@@ -278,6 +278,7 @@ static void build_hash_table(AggState *aggstate, int setno, long nbuckets);
static AggStatePerGroup lookup_hash_entry(AggState *aggstate, uint32 hash);
static void lookup_hash_entries(AggState *aggstate);
static TupleTableSlot *agg_retrieve_direct(AggState *aggstate);
+static void agg_sort_input(AggState *aggstate);
static void agg_fill_hash_table(AggState *aggstate);
static TupleTableSlot *agg_retrieve_hash_table(AggState *aggstate);
static Datum GetAggInitVal(Datum textInitVal, Oid transtype);
@@ -367,7 +368,7 @@ initialize_phase(AggState *aggstate, int newphase)
*/
if (newphase > 0 && newphase < aggstate->numphases - 1)
{
- Sort *sortnode = aggstate->phases[newphase + 1].sortnode;
+ Sort *sortnode = (Sort *)aggstate->phases[newphase + 1].aggnode->sortnode;
PlanState *outerNode = outerPlanState(aggstate);
TupleDesc tupDesc = ExecGetResultType(outerNode);
@@ -1594,6 +1595,8 @@ ExecAgg(PlanState *pstate)
break;
case AGG_PLAIN:
case AGG_SORTED:
+ if (!node->input_sorted)
+ agg_sort_input(node);
result = agg_retrieve_direct(node);
break;
}
@@ -1945,6 +1948,45 @@ agg_retrieve_direct(AggState *aggstate)
return NULL;
}
+static void
+agg_sort_input(AggState *aggstate)
+{
+ AggStatePerPhase phase = &aggstate->phases[1];
+ TupleDesc tupDesc;
+ Sort *sortnode;
+
+ Assert(!aggstate->input_sorted);
+ Assert(phase->aggnode->sortnode);
+
+ sortnode = (Sort *) phase->aggnode->sortnode;
+ tupDesc = ExecGetResultType(outerPlanState(aggstate));
+
+ aggstate->sort_in = tuplesort_begin_heap(tupDesc,
+ sortnode->numCols,
+ sortnode->sortColIdx,
+ sortnode->sortOperators,
+ sortnode->collations,
+ sortnode->nullsFirst,
+ work_mem,
+ NULL, false);
+ for (;;)
+ {
+ TupleTableSlot *outerslot;
+
+ outerslot = ExecProcNode(outerPlanState(aggstate));
+ if (TupIsNull(outerslot))
+ break;
+
+ tuplesort_puttupleslot(aggstate->sort_in, outerslot);
+ }
+
+ /* Sort the first phase */
+ tuplesort_performsort(aggstate->sort_in);
+
+ /* Mark the input to be sorted */
+ aggstate->input_sorted = true;
+}
+
/*
* ExecAgg for hashed case: read input and build hash table
*/
@@ -2127,6 +2169,7 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
Plan *outerPlan;
ExprContext *econtext;
TupleDesc scanDesc;
+ Agg *firstSortAgg;
int numaggs,
transno,
aggno;
@@ -2171,6 +2214,7 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
aggstate->grp_firstTuple = NULL;
aggstate->sort_in = NULL;
aggstate->sort_out = NULL;
+ aggstate->input_sorted = true;
/*
* phases[0] always exists, but is dummy in sorted/plain mode
@@ -2178,6 +2222,8 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
numPhases = (use_hashing ? 1 : 2);
numHashes = (use_hashing ? 1 : 0);
+ firstSortAgg = node->aggstrategy == AGG_SORTED ? node : NULL;
+
/*
* Calculate the maximum number of grouping sets in any phase; this
* determines the size of some allocations. Also calculate the number of
@@ -2199,7 +2245,13 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
* others add an extra phase.
*/
if (agg->aggstrategy != AGG_HASHED)
+ {
++numPhases;
+
+ if (!firstSortAgg)
+ firstSortAgg = agg;
+
+ }
else
++numHashes;
}
@@ -2208,6 +2260,13 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
aggstate->maxsets = numGroupingSets;
aggstate->numphases = numPhases;
+ /*
+ * The first SORTED phase is not sorted, agg need to do its own sort. See
+ * agg_sort_input(), this can only happen in groupingsets case.
+ */
+ if (firstSortAgg && firstSortAgg->sortnode)
+ aggstate->input_sorted = false;
+
aggstate->aggcontexts = (ExprContext **)
palloc0(sizeof(ExprContext *) * numGroupingSets);
@@ -2269,7 +2328,8 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
* If there are more than two phases (including a potential dummy phase
* 0), input will be resorted using tuplesort. Need a slot for that.
*/
- if (numPhases > 2)
+ if (numPhases > 2 ||
+ !aggstate->input_sorted)
{
aggstate->sort_slot = ExecInitExtraTupleSlot(estate, scanDesc,
&TTSOpsMinimalTuple);
@@ -2340,20 +2400,11 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
for (phaseidx = 0; phaseidx <= list_length(node->chain); ++phaseidx)
{
Agg *aggnode;
- Sort *sortnode;
if (phaseidx > 0)
- {
aggnode = list_nth_node(Agg, node->chain, phaseidx - 1);
- sortnode = castNode(Sort, aggnode->plan.lefttree);
- }
else
- {
aggnode = node;
- sortnode = NULL;
- }
-
- Assert(phase <= 1 || sortnode);
if (aggnode->aggstrategy == AGG_HASHED
|| aggnode->aggstrategy == AGG_MIXED)
@@ -2470,7 +2521,6 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
phasedata->aggnode = aggnode;
phasedata->aggstrategy = aggnode->aggstrategy;
- phasedata->sortnode = sortnode;
}
}
@@ -3559,6 +3609,10 @@ ExecReScanAgg(AggState *node)
sizeof(AggStatePerGroupData) * node->numaggs);
}
+ /* Reset input_sorted */
+ if (aggnode->sortnode)
+ node->input_sorted = false;
+
/* reset to phase 1 */
initialize_phase(node, 1);
@@ -3566,6 +3620,7 @@ ExecReScanAgg(AggState *node)
node->projected_set = -1;
}
+
if (outerPlan->chgParam == NULL)
ExecReScan(outerPlan);
}
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index eaab97f753..04b4c65858 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -992,6 +992,7 @@ _copyAgg(const Agg *from)
COPY_BITMAPSET_FIELD(aggParams);
COPY_NODE_FIELD(groupingSets);
COPY_NODE_FIELD(chain);
+ COPY_NODE_FIELD(sortnode);
return newnode;
}
diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c
index e084c3f069..5816d122c1 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -787,6 +787,7 @@ _outAgg(StringInfo str, const Agg *node)
WRITE_BITMAPSET_FIELD(aggParams);
WRITE_NODE_FIELD(groupingSets);
WRITE_NODE_FIELD(chain);
+ WRITE_NODE_FIELD(sortnode);
}
static void
diff --git a/src/backend/nodes/readfuncs.c b/src/backend/nodes/readfuncs.c
index d5b23a3479..af4fcfe1ee 100644
--- a/src/backend/nodes/readfuncs.c
+++ b/src/backend/nodes/readfuncs.c
@@ -2207,6 +2207,7 @@ _readAgg(void)
READ_BITMAPSET_FIELD(aggParams);
READ_NODE_FIELD(groupingSets);
READ_NODE_FIELD(chain);
+ READ_NODE_FIELD(sortnode);
READ_DONE();
}
diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c
index fc25908dc6..d5b34089aa 100644
--- a/src/backend/optimizer/plan/createplan.c
+++ b/src/backend/optimizer/plan/createplan.c
@@ -1645,6 +1645,7 @@ create_unique_plan(PlannerInfo *root, UniquePath *best_path, int flags)
NIL,
best_path->path.rows,
0,
+ NULL,
subplan);
}
else
@@ -2098,6 +2099,7 @@ create_agg_plan(PlannerInfo *root, AggPath *best_path)
NIL,
best_path->numGroups,
best_path->transitionSpace,
+ NULL,
subplan);
copy_generic_path_info(&plan->plan, (Path *) best_path);
@@ -2159,6 +2161,7 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
List *rollups = best_path->rollups;
AttrNumber *grouping_map;
int maxref;
+ int flags = CP_LABEL_TLIST;
List *chain;
ListCell *lc;
@@ -2168,9 +2171,15 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
/*
* Agg can project, so no need to be terribly picky about child tlist, but
- * we do need grouping columns to be available
+ * we do need grouping columns to be available; If the groupingsets need
+ * to sort the input, the agg will store the input rows in a tuplesort,
+ * it therefore behooves us to request a small tlist to avoid wasting
+ * spaces.
*/
- subplan = create_plan_recurse(root, best_path->subpath, CP_LABEL_TLIST);
+ if (!best_path->is_sorted)
+ flags = flags | CP_SMALL_TLIST;
+
+ subplan = create_plan_recurse(root, best_path->subpath, flags);
/*
* Compute the mapping from tleSortGroupRef to column index in the child's
@@ -2230,12 +2239,22 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
new_grpColIdx = remap_groupColIdx(root, rollup->groupClause);
- if (!rollup->is_hashed && !is_first_sort)
+ if (!rollup->is_hashed)
{
- sort_plan = (Plan *)
- make_sort_from_groupcols(rollup->groupClause,
- new_grpColIdx,
- subplan);
+ if (!is_first_sort ||
+ (is_first_sort && !best_path->is_sorted))
+ {
+ sort_plan = (Plan *)
+ make_sort_from_groupcols(rollup->groupClause,
+ new_grpColIdx,
+ subplan);
+
+ /*
+ * Remove stuff we don't need to avoid bloating debug output.
+ */
+ sort_plan->targetlist = NIL;
+ sort_plan->lefttree = NULL;
+ }
}
if (!rollup->is_hashed)
@@ -2260,16 +2279,8 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
NIL,
rollup->numGroups,
best_path->transitionSpace,
- sort_plan);
-
- /*
- * Remove stuff we don't need to avoid bloating debug output.
- */
- if (sort_plan)
- {
- sort_plan->targetlist = NIL;
- sort_plan->lefttree = NULL;
- }
+ sort_plan,
+ NULL);
chain = lappend(chain, agg_plan);
}
@@ -2281,10 +2292,26 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
{
RollupData *rollup = linitial(rollups);
AttrNumber *top_grpColIdx;
+ Plan *sort_plan = NULL;
int numGroupCols;
top_grpColIdx = remap_groupColIdx(root, rollup->groupClause);
+ /* the input is not sorted yet */
+ if (!rollup->is_hashed &&
+ !best_path->is_sorted)
+ {
+ sort_plan = (Plan *)
+ make_sort_from_groupcols(rollup->groupClause,
+ top_grpColIdx,
+ subplan);
+ /*
+ * Remove stuff we don't need to avoid bloating debug output.
+ */
+ sort_plan->targetlist = NIL;
+ sort_plan->lefttree = NULL;
+ }
+
numGroupCols = list_length((List *) linitial(rollup->gsets));
plan = make_agg(build_path_tlist(root, &best_path->path),
@@ -2299,6 +2326,7 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
chain,
rollup->numGroups,
best_path->transitionSpace,
+ sort_plan,
subplan);
/* Copy cost data from Path to Plan */
@@ -6197,7 +6225,7 @@ make_agg(List *tlist, List *qual,
AggStrategy aggstrategy, AggSplit aggsplit,
int numGroupCols, AttrNumber *grpColIdx, Oid *grpOperators, Oid *grpCollations,
List *groupingSets, List *chain, double dNumGroups,
- Size transitionSpace, Plan *lefttree)
+ Size transitionSpace, Plan *sortnode, Plan *lefttree)
{
Agg *node = makeNode(Agg);
Plan *plan = &node->plan;
@@ -6217,6 +6245,7 @@ make_agg(List *tlist, List *qual,
node->aggParams = NULL; /* SS_finalize_plan() will fill this */
node->groupingSets = groupingSets;
node->chain = chain;
+ node->sortnode = sortnode;
plan->qual = qual;
plan->targetlist = tlist;
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
index b44efd6314..82a15761b4 100644
--- a/src/backend/optimizer/plan/planner.c
+++ b/src/backend/optimizer/plan/planner.c
@@ -175,7 +175,8 @@ static void consider_groupingsets_paths(PlannerInfo *root,
bool can_hash,
grouping_sets_data *gd,
const AggClauseCosts *agg_costs,
- double dNumGroups);
+ double dNumGroups,
+ AggStrategy strat);
static RelOptInfo *create_window_paths(PlannerInfo *root,
RelOptInfo *input_rel,
PathTarget *input_target,
@@ -4186,6 +4187,14 @@ create_ordinary_grouping_paths(PlannerInfo *root, RelOptInfo *input_rel,
* it, by combinations of hashing and sorting. This can be called multiple
* times, so it's important that it not scribble on input. No result is
* returned, but any generated paths are added to grouped_rel.
+ *
+ * - strat:
+ * preferred aggregate strategy to use.
+ *
+ * - is_sorted:
+ * Is the input sorted on the groupCols of the first rollup. Caller
+ * must set it correctly if strat is set to AGG_SORTED, the planner
+ * uses it to generate a sortnode.
*/
static void
consider_groupingsets_paths(PlannerInfo *root,
@@ -4195,13 +4204,15 @@ consider_groupingsets_paths(PlannerInfo *root,
bool can_hash,
grouping_sets_data *gd,
const AggClauseCosts *agg_costs,
- double dNumGroups)
+ double dNumGroups,
+ AggStrategy strat)
{
Query *parse = root->parse;
+ Assert(strat == AGG_HASHED || strat == AGG_SORTED);
/*
- * If we're not being offered sorted input, then only consider plans that
- * can be done entirely by hashing.
+ * If strat is AGG_HASHED, then only consider plans that can be done
+ * entirely by hashing.
*
* We can hash everything if it looks like it'll fit in work_mem. But if
* the input is actually sorted despite not being advertised as such, we
@@ -4210,7 +4221,7 @@ consider_groupingsets_paths(PlannerInfo *root,
* If none of the grouping sets are sortable, then ignore the work_mem
* limit and generate a path anyway, since otherwise we'll just fail.
*/
- if (!is_sorted)
+ if (strat == AGG_HASHED)
{
List *new_rollups = NIL;
RollupData *unhashed_rollup = NULL;
@@ -4251,6 +4262,8 @@ consider_groupingsets_paths(PlannerInfo *root,
unhashed_rollup = lfirst_node(RollupData, l_start);
exclude_groups = unhashed_rollup->numGroups;
l_start = lnext(gd->rollups, l_start);
+ /* update is_sorted to true */
+ is_sorted = true;
}
hashsize = estimate_hashagg_tablesize(path,
@@ -4348,6 +4361,8 @@ consider_groupingsets_paths(PlannerInfo *root,
rollup->hashable = false;
rollup->is_hashed = false;
new_rollups = lappend(new_rollups, rollup);
+ /* update is_sorted to true */
+ is_sorted = true;
strat = AGG_MIXED;
}
@@ -4359,18 +4374,23 @@ consider_groupingsets_paths(PlannerInfo *root,
strat,
new_rollups,
agg_costs,
- dNumGroups));
+ dNumGroups,
+ is_sorted));
return;
}
/*
- * If we have sorted input but nothing we can do with it, bail.
+ * Strategy is AGG_SORTED but nothing we can do with it, bail.
*/
if (list_length(gd->rollups) == 0)
return;
/*
- * Given sorted input, we try and make two paths: one sorted and one mixed
+ * Callers consider AGG_SORTED strategy, the first rollup must
+ * use non-hashed aggregate, 'is_sorted' tells whether the first
+ * rollup need to do its own sort.
+ *
+ * we try and make two paths: one sorted and one mixed
* sort/hash. (We need to try both because hashagg might be disabled, or
* some columns might not be sortable.)
*
@@ -4427,7 +4447,7 @@ consider_groupingsets_paths(PlannerInfo *root,
/*
* We leave the first rollup out of consideration since it's the
- * one that matches the input sort order. We assign indexes "i"
+ * one that need to be sorted. We assign indexes "i"
* to only those entries considered for hashing; the second loop,
* below, must use the same condition.
*/
@@ -4516,7 +4536,8 @@ consider_groupingsets_paths(PlannerInfo *root,
AGG_MIXED,
rollups,
agg_costs,
- dNumGroups));
+ dNumGroups,
+ is_sorted));
}
}
@@ -4532,7 +4553,8 @@ consider_groupingsets_paths(PlannerInfo *root,
AGG_SORTED,
gd->rollups,
agg_costs,
- dNumGroups));
+ dNumGroups,
+ is_sorted));
}
/*
@@ -6399,6 +6421,16 @@ add_paths_to_grouping_rel(PlannerInfo *root, RelOptInfo *input_rel,
path->pathkeys);
if (path == cheapest_path || is_sorted)
{
+ if (parse->groupingSets)
+ {
+ /* consider AGG_SORTED strategy */
+ consider_groupingsets_paths(root, grouped_rel,
+ path, is_sorted, can_hash,
+ gd, agg_costs, dNumGroups,
+ AGG_SORTED);
+ continue;
+ }
+
/* Sort the cheapest-total path if it isn't already sorted */
if (!is_sorted)
path = (Path *) create_sort_path(root,
@@ -6407,14 +6439,7 @@ add_paths_to_grouping_rel(PlannerInfo *root, RelOptInfo *input_rel,
root->group_pathkeys,
-1.0);
- /* Now decide what to stick atop it */
- if (parse->groupingSets)
- {
- consider_groupingsets_paths(root, grouped_rel,
- path, true, can_hash,
- gd, agg_costs, dNumGroups);
- }
- else if (parse->hasAggs)
+ if (parse->hasAggs)
{
/*
* We have aggregation, possibly with plain GROUP BY. Make
@@ -6514,7 +6539,8 @@ add_paths_to_grouping_rel(PlannerInfo *root, RelOptInfo *input_rel,
*/
consider_groupingsets_paths(root, grouped_rel,
cheapest_path, false, true,
- gd, agg_costs, dNumGroups);
+ gd, agg_costs, dNumGroups,
+ AGG_HASHED);
}
else
{
diff --git a/src/backend/optimizer/util/pathnode.c b/src/backend/optimizer/util/pathnode.c
index d9ce516211..0feb3363d3 100644
--- a/src/backend/optimizer/util/pathnode.c
+++ b/src/backend/optimizer/util/pathnode.c
@@ -2983,6 +2983,7 @@ create_agg_path(PlannerInfo *root,
* 'rollups' is a list of RollupData nodes
* 'agg_costs' contains cost info about the aggregate functions to be computed
* 'numGroups' is the estimated total number of groups
+ * 'is_sorted' is the input sorted in the group cols of first rollup
*/
GroupingSetsPath *
create_groupingsets_path(PlannerInfo *root,
@@ -2992,7 +2993,8 @@ create_groupingsets_path(PlannerInfo *root,
AggStrategy aggstrategy,
List *rollups,
const AggClauseCosts *agg_costs,
- double numGroups)
+ double numGroups,
+ bool is_sorted)
{
GroupingSetsPath *pathnode = makeNode(GroupingSetsPath);
PathTarget *target = rel->reltarget;
@@ -3010,6 +3012,7 @@ create_groupingsets_path(PlannerInfo *root,
subpath->parallel_safe;
pathnode->path.parallel_workers = subpath->parallel_workers;
pathnode->subpath = subpath;
+ pathnode->is_sorted = is_sorted;
/*
* Simplify callers by downgrading AGG_SORTED to AGG_PLAIN, and AGG_MIXED
@@ -3061,14 +3064,33 @@ create_groupingsets_path(PlannerInfo *root,
*/
if (is_first)
{
+ Cost input_startup_cost = subpath->startup_cost;
+ Cost input_total_cost = subpath->total_cost;
+
+ if (!rollup->is_hashed && !is_sorted && numGroupCols)
+ {
+ Path sort_path; /* dummy for result of cost_sort */
+
+ cost_sort(&sort_path, root, NIL,
+ input_total_cost,
+ subpath->rows,
+ subpath->pathtarget->width,
+ 0.0,
+ work_mem,
+ -1.0);
+
+ input_startup_cost = sort_path.startup_cost;
+ input_total_cost = sort_path.total_cost;
+ }
+
cost_agg(&pathnode->path, root,
aggstrategy,
agg_costs,
numGroupCols,
rollup->numGroups,
having_qual,
- subpath->startup_cost,
- subpath->total_cost,
+ input_startup_cost,
+ input_total_cost,
subpath->rows);
is_first = false;
if (!rollup->is_hashed)
@@ -3079,7 +3101,7 @@ create_groupingsets_path(PlannerInfo *root,
Path sort_path; /* dummy for result of cost_sort */
Path agg_path; /* dummy for result of cost_agg */
- if (rollup->is_hashed || is_first_sort)
+ if (rollup->is_hashed || (is_first_sort && is_sorted))
{
/*
* Account for cost of aggregation, but don't charge input
diff --git a/src/include/executor/nodeAgg.h b/src/include/executor/nodeAgg.h
index 264916f9a9..66a83b9ac9 100644
--- a/src/include/executor/nodeAgg.h
+++ b/src/include/executor/nodeAgg.h
@@ -277,8 +277,6 @@ typedef struct AggStatePerPhaseData
ExprState **eqfunctions; /* expression returning equality, indexed by
* nr of cols to compare */
Agg *aggnode; /* Agg node for phase data */
- Sort *sortnode; /* Sort node for input ordering for phase */
-
ExprState *evaltrans; /* evaluation of transition functions */
} AggStatePerPhaseData;
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index cd3ddf781f..5e33a368f5 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -2083,8 +2083,11 @@ typedef struct AggState
AggStatePerGroup *hash_pergroup; /* grouping set indexed array of
* per-group pointers */
+ /* these fields are used in AGG_SORTED and AGG_MIXED */
+ bool input_sorted; /* hash table filled yet? */
+
/* support for evaluation of agg input expressions: */
-#define FIELDNO_AGGSTATE_ALL_PERGROUPS 34
+#define FIELDNO_AGGSTATE_ALL_PERGROUPS 35
AggStatePerGroup *all_pergroups; /* array of first ->pergroups, than
* ->hash_pergroup */
ProjectionInfo *combinedproj; /* projection machinery */
diff --git a/src/include/nodes/pathnodes.h b/src/include/nodes/pathnodes.h
index 0ceb809644..c1e69c808f 100644
--- a/src/include/nodes/pathnodes.h
+++ b/src/include/nodes/pathnodes.h
@@ -1702,6 +1702,7 @@ typedef struct GroupingSetsPath
List *rollups; /* list of RollupData */
List *qual; /* quals (HAVING quals), if any */
uint64 transitionSpace; /* for pass-by-ref transition data */
+ bool is_sorted; /* input sorted in groupcols of first rollup */
} GroupingSetsPath;
/*
diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h
index 4869fe7b6d..3cd2537e9e 100644
--- a/src/include/nodes/plannodes.h
+++ b/src/include/nodes/plannodes.h
@@ -818,6 +818,7 @@ typedef struct Agg
/* Note: planner provides numGroups & aggParams only in HASHED/MIXED case */
List *groupingSets; /* grouping sets to use */
List *chain; /* chained Agg/Sort nodes */
+ Plan *sortnode; /* agg does its own sort, only used by grouping sets now */
} Agg;
/* ----------------
diff --git a/src/include/optimizer/pathnode.h b/src/include/optimizer/pathnode.h
index e450fe112a..f9f388ba06 100644
--- a/src/include/optimizer/pathnode.h
+++ b/src/include/optimizer/pathnode.h
@@ -217,7 +217,8 @@ extern GroupingSetsPath *create_groupingsets_path(PlannerInfo *root,
AggStrategy aggstrategy,
List *rollups,
const AggClauseCosts *agg_costs,
- double numGroups);
+ double numGroups,
+ bool is_sorted);
extern MinMaxAggPath *create_minmaxagg_path(PlannerInfo *root,
RelOptInfo *rel,
PathTarget *target,
diff --git a/src/include/optimizer/planmain.h b/src/include/optimizer/planmain.h
index 4781201001..5954ff3997 100644
--- a/src/include/optimizer/planmain.h
+++ b/src/include/optimizer/planmain.h
@@ -55,7 +55,7 @@ extern Agg *make_agg(List *tlist, List *qual,
AggStrategy aggstrategy, AggSplit aggsplit,
int numGroupCols, AttrNumber *grpColIdx, Oid *grpOperators, Oid *grpCollations,
List *groupingSets, List *chain, double dNumGroups,
- Size transitionSpace, Plan *lefttree);
+ Size transitionSpace, Plan *sortnode, Plan *lefttree);
extern Limit *make_limit(Plan *lefttree, Node *limitOffset, Node *limitCount);
/*
diff --git a/src/test/regress/expected/groupingsets.out b/src/test/regress/expected/groupingsets.out
index c1f802c88a..12425f46ca 100644
--- a/src/test/regress/expected/groupingsets.out
+++ b/src/test/regress/expected/groupingsets.out
@@ -366,15 +366,14 @@ explain (costs off)
select g as alias1, g as alias2
from generate_series(1,3) g
group by alias1, rollup(alias2);
- QUERY PLAN
-------------------------------------------------
+ QUERY PLAN
+------------------------------------------
GroupAggregate
- Group Key: g, g
- Group Key: g
- -> Sort
- Sort Key: g
- -> Function Scan on generate_series g
-(6 rows)
+ Sort Key: g, g
+ Group Key: g, g
+ Group Key: g
+ -> Function Scan on generate_series g
+(5 rows)
select g as alias1, g as alias2
from generate_series(1,3) g
@@ -640,15 +639,14 @@ select a, b, sum(v.x)
-- Test reordering of grouping sets
explain (costs off)
select * from gstest1 group by grouping sets((a,b,v),(v)) order by v,b,a;
- QUERY PLAN
-------------------------------------------------------------------------------
+ QUERY PLAN
+---------------------------------------------------------------------------
GroupAggregate
- Group Key: "*VALUES*".column3, "*VALUES*".column2, "*VALUES*".column1
- Group Key: "*VALUES*".column3
- -> Sort
- Sort Key: "*VALUES*".column3, "*VALUES*".column2, "*VALUES*".column1
- -> Values Scan on "*VALUES*"
-(6 rows)
+ Sort Key: "*VALUES*".column3, "*VALUES*".column2, "*VALUES*".column1
+ Group Key: "*VALUES*".column3, "*VALUES*".column2, "*VALUES*".column1
+ Group Key: "*VALUES*".column3
+ -> Values Scan on "*VALUES*"
+(5 rows)
-- Agg level check. This query should error out.
select (select grouping(a,b) from gstest2) from gstest2 group by a,b;
@@ -723,13 +721,12 @@ explain (costs off)
QUERY PLAN
----------------------------------
GroupAggregate
- Group Key: a
- Group Key: ()
+ Sort Key: a
+ Group Key: a
+ Group Key: ()
Filter: (a IS DISTINCT FROM 1)
- -> Sort
- Sort Key: a
- -> Seq Scan on gstest2
-(7 rows)
+ -> Seq Scan on gstest2
+(6 rows)
select v.c, (select count(*) from gstest2 group by () having v.c)
from (values (false),(true)) v(c) order by v.c;
@@ -1018,18 +1015,17 @@ explain (costs off) select a, b, grouping(a,b), sum(v), count(*), max(v)
explain (costs off)
select a, b, grouping(a,b), array_agg(v order by v)
from gstest1 group by cube(a,b);
- QUERY PLAN
-----------------------------------------------------------
+ QUERY PLAN
+-------------------------------------------------------
GroupAggregate
- Group Key: "*VALUES*".column1, "*VALUES*".column2
- Group Key: "*VALUES*".column1
- Group Key: ()
+ Sort Key: "*VALUES*".column1, "*VALUES*".column2
+ Group Key: "*VALUES*".column1, "*VALUES*".column2
+ Group Key: "*VALUES*".column1
+ Group Key: ()
Sort Key: "*VALUES*".column2
Group Key: "*VALUES*".column2
- -> Sort
- Sort Key: "*VALUES*".column1, "*VALUES*".column2
- -> Values Scan on "*VALUES*"
-(9 rows)
+ -> Values Scan on "*VALUES*"
+(8 rows)
-- unsortable cases
select unsortable_col, count(*)
@@ -1071,11 +1067,10 @@ explain (costs off)
Sort Key: (GROUPING(unhashable_col, unsortable_col)), (sum(v))
-> MixedAggregate
Hash Key: unsortable_col
- Group Key: unhashable_col
- -> Sort
- Sort Key: unhashable_col
- -> Seq Scan on gstest4
-(8 rows)
+ Sort Key: unhashable_col
+ Group Key: unhashable_col
+ -> Seq Scan on gstest4
+(7 rows)
select unhashable_col, unsortable_col,
grouping(unhashable_col, unsortable_col),
@@ -1114,11 +1109,10 @@ explain (costs off)
Sort Key: (GROUPING(unhashable_col, unsortable_col)), (sum(v))
-> MixedAggregate
Hash Key: v, unsortable_col
- Group Key: v, unhashable_col
- -> Sort
- Sort Key: v, unhashable_col
- -> Seq Scan on gstest4
-(8 rows)
+ Sort Key: v, unhashable_col
+ Group Key: v, unhashable_col
+ -> Seq Scan on gstest4
+(7 rows)
-- empty input: first is 0 rows, second 1, third 3 etc.
select a, b, sum(v), count(*) from gstest_empty group by grouping sets ((a,b),a);
@@ -1366,19 +1360,18 @@ explain (costs off)
BEGIN;
SET LOCAL enable_hashagg = false;
EXPLAIN (COSTS OFF) SELECT a, b, count(*), max(a), max(b) FROM gstest3 GROUP BY GROUPING SETS(a, b,()) ORDER BY a, b;
- QUERY PLAN
----------------------------------------
+ QUERY PLAN
+---------------------------------
Sort
Sort Key: a, b
-> GroupAggregate
- Group Key: a
- Group Key: ()
+ Sort Key: a
+ Group Key: a
+ Group Key: ()
Sort Key: b
Group Key: b
- -> Sort
- Sort Key: a
- -> Seq Scan on gstest3
-(10 rows)
+ -> Seq Scan on gstest3
+(9 rows)
SELECT a, b, count(*), max(a), max(b) FROM gstest3 GROUP BY GROUPING SETS(a, b,()) ORDER BY a, b;
a | b | count | max | max
@@ -1549,22 +1542,21 @@ explain (costs off)
count(hundred), count(thousand), count(twothousand),
count(*)
from tenk1 group by grouping sets (unique1,twothousand,thousand,hundred,ten,four,two);
- QUERY PLAN
--------------------------------
+ QUERY PLAN
+----------------------------
MixedAggregate
Hash Key: two
Hash Key: four
Hash Key: ten
Hash Key: hundred
- Group Key: unique1
+ Sort Key: unique1
+ Group Key: unique1
Sort Key: twothousand
Group Key: twothousand
Sort Key: thousand
Group Key: thousand
- -> Sort
- Sort Key: unique1
- -> Seq Scan on tenk1
-(13 rows)
+ -> Seq Scan on tenk1
+(12 rows)
explain (costs off)
select unique1,
@@ -1572,18 +1564,17 @@ explain (costs off)
count(hundred), count(thousand), count(twothousand),
count(*)
from tenk1 group by grouping sets (unique1,hundred,ten,four,two);
- QUERY PLAN
--------------------------------
+ QUERY PLAN
+-------------------------
MixedAggregate
Hash Key: two
Hash Key: four
Hash Key: ten
Hash Key: hundred
- Group Key: unique1
- -> Sort
- Sort Key: unique1
- -> Seq Scan on tenk1
-(9 rows)
+ Sort Key: unique1
+ Group Key: unique1
+ -> Seq Scan on tenk1
+(8 rows)
set work_mem = '384kB';
explain (costs off)
@@ -1592,21 +1583,20 @@ explain (costs off)
count(hundred), count(thousand), count(twothousand),
count(*)
from tenk1 group by grouping sets (unique1,twothousand,thousand,hundred,ten,four,two);
- QUERY PLAN
--------------------------------
+ QUERY PLAN
+----------------------------
MixedAggregate
Hash Key: two
Hash Key: four
Hash Key: ten
Hash Key: hundred
Hash Key: thousand
- Group Key: unique1
+ Sort Key: unique1
+ Group Key: unique1
Sort Key: twothousand
Group Key: twothousand
- -> Sort
- Sort Key: unique1
- -> Seq Scan on tenk1
-(12 rows)
+ -> Seq Scan on tenk1
+(11 rows)
-- check collation-sensitive matching between grouping expressions
-- (similar to a check for aggregates, but there are additional code
--
2.21.1
0002-fixes.patch (text/plain, 5.9K) download
From 57f30d60158398bd57f02ef00800d3f63c6ab12f Mon Sep 17 00:00:00 2001
From: Tomas Vondra <tv@fuzzy.cz>
Date: Thu, 19 Mar 2020 01:20:58 +0100
Subject: [PATCH 2/7] fixes
---
src/backend/executor/nodeAgg.c | 3 +--
src/backend/optimizer/plan/createplan.c | 14 ++++++++++----
src/backend/optimizer/plan/planner.c | 23 +++++++++++++----------
src/test/modules/unsafe_tests/Makefile | 2 +-
4 files changed, 25 insertions(+), 17 deletions(-)
diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c
index b4f53bf77a..c3a043e448 100644
--- a/src/backend/executor/nodeAgg.c
+++ b/src/backend/executor/nodeAgg.c
@@ -368,7 +368,7 @@ initialize_phase(AggState *aggstate, int newphase)
*/
if (newphase > 0 && newphase < aggstate->numphases - 1)
{
- Sort *sortnode = (Sort *)aggstate->phases[newphase + 1].aggnode->sortnode;
+ Sort *sortnode = (Sort *) aggstate->phases[newphase + 1].aggnode->sortnode;
PlanState *outerNode = outerPlanState(aggstate);
TupleDesc tupDesc = ExecGetResultType(outerNode);
@@ -3620,7 +3620,6 @@ ExecReScanAgg(AggState *node)
node->projected_set = -1;
}
-
if (outerPlan->chgParam == NULL)
ExecReScan(outerPlan);
}
diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c
index d5b34089aa..044ec92aa8 100644
--- a/src/backend/optimizer/plan/createplan.c
+++ b/src/backend/optimizer/plan/createplan.c
@@ -2171,10 +2171,9 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
/*
* Agg can project, so no need to be terribly picky about child tlist, but
- * we do need grouping columns to be available; If the groupingsets need
+ * we do need grouping columns to be available. If the groupingsets need
* to sort the input, the agg will store the input rows in a tuplesort,
- * it therefore behooves us to request a small tlist to avoid wasting
- * spaces.
+ * so we request a small tlist to avoid wasting space.
*/
if (!best_path->is_sorted)
flags = flags | CP_SMALL_TLIST;
@@ -2239,6 +2238,10 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
new_grpColIdx = remap_groupColIdx(root, rollup->groupClause);
+ /*
+ * FIXME This combination of nested if checks needs some explanation
+ * why we need this particular combination of flags.
+ */
if (!rollup->is_hashed)
{
if (!is_first_sort ||
@@ -2297,7 +2300,10 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
top_grpColIdx = remap_groupColIdx(root, rollup->groupClause);
- /* the input is not sorted yet */
+ /*
+ * When the rollup uses sorted mode, and the input is not already sorted,
+ * add an explicit sort.
+ */
if (!rollup->is_hashed &&
!best_path->is_sorted)
{
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
index 82a15761b4..fc6a1d0044 100644
--- a/src/backend/optimizer/plan/planner.c
+++ b/src/backend/optimizer/plan/planner.c
@@ -4188,13 +4188,9 @@ create_ordinary_grouping_paths(PlannerInfo *root, RelOptInfo *input_rel,
* times, so it's important that it not scribble on input. No result is
* returned, but any generated paths are added to grouped_rel.
*
- * - strat:
- * preferred aggregate strategy to use.
- *
- * - is_sorted:
- * Is the input sorted on the groupCols of the first rollup. Caller
- * must set it correctly if strat is set to AGG_SORTED, the planner
- * uses it to generate a sortnode.
+ * The caller specifies the preferred aggregate strategy (sorted or hashed) using
+ * the strat aprameter. When the requested strategy is AGG_SORTED, the input path
+ * needs to be sorted accordingly (is_sorted needs to be true).
*/
static void
consider_groupingsets_paths(PlannerInfo *root,
@@ -4262,7 +4258,8 @@ consider_groupingsets_paths(PlannerInfo *root,
unhashed_rollup = lfirst_node(RollupData, l_start);
exclude_groups = unhashed_rollup->numGroups;
l_start = lnext(gd->rollups, l_start);
- /* update is_sorted to true */
+ /* update is_sorted to true
+ * XXX why? shouldn't it be already set by the caller? */
is_sorted = true;
}
@@ -4361,7 +4358,9 @@ consider_groupingsets_paths(PlannerInfo *root,
rollup->hashable = false;
rollup->is_hashed = false;
new_rollups = lappend(new_rollups, rollup);
- /* update is_sorted to true */
+ /* update is_sorted to true
+ * XXX why? shouldn't it be already set by the caller?
+ */
is_sorted = true;
strat = AGG_MIXED;
}
@@ -4396,6 +4395,9 @@ consider_groupingsets_paths(PlannerInfo *root,
*
* can_hash is passed in as false if some obstacle elsewhere (such as
* ordered aggs) means that we shouldn't consider hashing at all.
+ *
+ * XXX This comment seems to be broken by the patch, and it's not very
+ * clear to me what it tries to say.
*/
if (can_hash && gd->any_hashable)
{
@@ -4447,7 +4449,7 @@ consider_groupingsets_paths(PlannerInfo *root,
/*
* We leave the first rollup out of consideration since it's the
- * one that need to be sorted. We assign indexes "i"
+ * one that matches the input sort order. We assign indexes "i"
* to only those entries considered for hashing; the second loop,
* below, must use the same condition.
*/
@@ -6421,6 +6423,7 @@ add_paths_to_grouping_rel(PlannerInfo *root, RelOptInfo *input_rel,
path->pathkeys);
if (path == cheapest_path || is_sorted)
{
+ /* XXX Why do we do it before possibly adding an explicit sort on top? */
if (parse->groupingSets)
{
/* consider AGG_SORTED strategy */
diff --git a/src/test/modules/unsafe_tests/Makefile b/src/test/modules/unsafe_tests/Makefile
index 3ecf5fcfc5..2cf710eb2c 100644
--- a/src/test/modules/unsafe_tests/Makefile
+++ b/src/test/modules/unsafe_tests/Makefile
@@ -1,6 +1,6 @@
# src/test/modules/unsafe_tests/Makefile
-REGRESS = rolenames alter_system_table
+REGRESS = alter_system_table
ifdef USE_PGXS
PG_CONFIG = pg_config
--
2.21.1
0003-fix-a-numtrans-bug.patch (text/plain, 1.4K) download
From d765999cd2446fa05ad9b53a5e87dd5480a9b55c Mon Sep 17 00:00:00 2001
From: Pengzhou Tang <ptang@pivotal.io>
Date: Thu, 12 Mar 2020 04:38:36 -0400
Subject: [PATCH 3/7] fix a numtrans bug
aggstate->numtrans is always zero when building the hash table for
hash aggregates, this make the additional size of hash table not
correct.
---
src/backend/executor/nodeAgg.c | 12 ++++++++----
1 file changed, 8 insertions(+), 4 deletions(-)
diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c
index c3a043e448..f7af5eebd0 100644
--- a/src/backend/executor/nodeAgg.c
+++ b/src/backend/executor/nodeAgg.c
@@ -2570,10 +2570,6 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
{
/* this is an array of pointers, not structures */
aggstate->hash_pergroup = pergroups;
-
- find_hash_columns(aggstate);
- build_hash_tables(aggstate);
- aggstate->table_filled = false;
}
/*
@@ -2929,6 +2925,14 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
(errcode(ERRCODE_GROUPING_ERROR),
errmsg("aggregate function calls cannot be nested")));
+ /* Initialize hash tables for hash aggregates */
+ if (use_hashing)
+ {
+ find_hash_columns(aggstate);
+ build_hash_tables(aggstate);
+ aggstate->table_filled = false;
+ }
+
/*
* Build expressions doing all the transition work at once. We build a
* different one for each phase, as the number of transition function
--
2.21.1
0004-Reorganise-the-aggregate-phases.patch (text/plain, 70.3K) download
From 62ea6872e4da65f03dd93f977b62ab483e8b46f7 Mon Sep 17 00:00:00 2001
From: Tomas Vondra <tv@fuzzy.cz>
Date: Thu, 19 Mar 2020 01:35:48 +0100
Subject: [PATCH 4/7] Reorganise the aggregate phases
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
This commit is a preparing step to support parallel grouping sets.
When planning, PG used to organize the grouping sets in [HASHED] -> [SORTED]
order which means HASHED aggregates were always located before SORTED aggregate,
when initializing AGG node, PG also organized the aggregate phases in
[HASHED]->[SORTED] order, all HASHED grouping sets were squeezed to the phase 0,
when executing AGG node, if followed AGG_SORTED or AGG_MIXED strategy, the
executor will start from phase1 -> phases2-> phases3 then phase0 if it is an
AGG_MIXED strategy. This bothers a lot when adding the support for parallel
grouping sets, firstly, we need complicated logic to locate the first sort
rollup/phase and handle the special order for a different strategy in many
places, Secondly, squeezing all hashed grouping sets to phase 0 is not working
for parallel grouping sets, we can not put all hash transition functions to one
expression state in the final stage.
This commit organizes the grouping sets in a more natural order: [SORTED]->[HASHED]
and the HASHED sets are no longer squeezed to a single phase, we use another way
to put all hash transitions to the first phase's expression state, the executor
now starts execution from phase0 for all strategies.
This commit also move 'sort_in' from AggState to AggStatePerPhase* structure,
this helps to handle more complicated cases when parallel groupingsets is
introduced, we might need to add a tuplestore 'store_in' to store partial
aggregates results for PLAIN sets then.
---
src/backend/commands/explain.c | 2 +-
src/backend/executor/execExpr.c | 58 +-
src/backend/executor/execExprInterp.c | 30 +-
src/backend/executor/nodeAgg.c | 718 +++++++++---------
src/backend/jit/llvm/llvmjit_expr.c | 51 +-
src/backend/optimizer/plan/createplan.c | 29 +-
src/backend/optimizer/plan/planner.c | 9 +-
src/backend/optimizer/util/pathnode.c | 71 +-
src/include/executor/execExpr.h | 5 +-
src/include/executor/executor.h | 2 +-
src/include/executor/nodeAgg.h | 26 +-
src/include/nodes/execnodes.h | 18 +-
src/test/regress/expected/groupingsets.out | 38 +-
.../regress/expected/partition_aggregate.out | 2 +-
14 files changed, 529 insertions(+), 530 deletions(-)
diff --git a/src/backend/commands/explain.c b/src/backend/commands/explain.c
index b1609b339a..2c63cdb46c 100644
--- a/src/backend/commands/explain.c
+++ b/src/backend/commands/explain.c
@@ -2317,7 +2317,7 @@ show_grouping_set_keys(PlanState *planstate,
const char *keyname;
const char *keysetname;
- if (aggnode->aggstrategy == AGG_HASHED || aggnode->aggstrategy == AGG_MIXED)
+ if (aggnode->aggstrategy == AGG_HASHED)
{
keyname = "Hash Key";
keysetname = "Hash Keys";
diff --git a/src/backend/executor/execExpr.c b/src/backend/executor/execExpr.c
index 1370ffec50..07789501f7 100644
--- a/src/backend/executor/execExpr.c
+++ b/src/backend/executor/execExpr.c
@@ -80,7 +80,7 @@ static void ExecInitCoerceToDomain(ExprEvalStep *scratch, CoerceToDomain *ctest,
static void ExecBuildAggTransCall(ExprState *state, AggState *aggstate,
ExprEvalStep *scratch,
FunctionCallInfo fcinfo, AggStatePerTrans pertrans,
- int transno, int setno, int setoff, bool ishash,
+ int transno, int setno, AggStatePerPhase perphase,
bool nullcheck);
@@ -2931,13 +2931,13 @@ ExecInitCoerceToDomain(ExprEvalStep *scratch, CoerceToDomain *ctest,
* the array of AggStatePerGroup, and skip evaluation if so.
*/
ExprState *
-ExecBuildAggTrans(AggState *aggstate, AggStatePerPhase phase,
- bool doSort, bool doHash, bool nullcheck)
+ExecBuildAggTrans(AggState *aggstate, AggStatePerPhase phase, bool nullcheck)
{
ExprState *state = makeNode(ExprState);
PlanState *parent = &aggstate->ss.ps;
ExprEvalStep scratch = {0};
bool isCombine = DO_AGGSPLIT_COMBINE(aggstate->aggsplit);
+ ListCell *lc;
LastAttnumInfo deform = {0, 0, 0};
state->expr = (Expr *) aggstate;
@@ -3155,38 +3155,24 @@ ExecBuildAggTrans(AggState *aggstate, AggStatePerPhase phase,
* grouping set). Do so for both sort and hash based computations, as
* applicable.
*/
- if (doSort)
+ for (int setno = 0; setno < phase->numsets; setno++)
{
- int processGroupingSets = Max(phase->numsets, 1);
- int setoff = 0;
-
- for (int setno = 0; setno < processGroupingSets; setno++)
- {
- ExecBuildAggTransCall(state, aggstate, &scratch, trans_fcinfo,
- pertrans, transno, setno, setoff, false,
- nullcheck);
- setoff++;
- }
+ ExecBuildAggTransCall(state, aggstate, &scratch, trans_fcinfo,
+ pertrans, transno, setno, phase, nullcheck);
}
- if (doHash)
+ /*
+ * Call transition function for HASHED aggs that can be
+ * advanced concurrently.
+ */
+ foreach(lc, phase->concurrent_hashes)
{
- int numHashes = aggstate->num_hashes;
- int setoff;
+ AggStatePerPhaseHash perhash = (AggStatePerPhaseHash) lfirst(lc);
- /* in MIXED mode, there'll be preceding transition values */
- if (aggstate->aggstrategy != AGG_HASHED)
- setoff = aggstate->maxsets;
- else
- setoff = 0;
-
- for (int setno = 0; setno < numHashes; setno++)
- {
- ExecBuildAggTransCall(state, aggstate, &scratch, trans_fcinfo,
- pertrans, transno, setno, setoff, true,
- nullcheck);
- setoff++;
- }
+ ExecBuildAggTransCall(state, aggstate, &scratch, trans_fcinfo,
+ pertrans, transno, 0,
+ (AggStatePerPhase) perhash,
+ nullcheck);
}
/* adjust early bail out jump target(s) */
@@ -3234,14 +3220,17 @@ static void
ExecBuildAggTransCall(ExprState *state, AggState *aggstate,
ExprEvalStep *scratch,
FunctionCallInfo fcinfo, AggStatePerTrans pertrans,
- int transno, int setno, int setoff, bool ishash,
+ int transno, int setno, AggStatePerPhase perphase,
bool nullcheck)
{
ExprContext *aggcontext;
int adjust_jumpnull = -1;
- if (ishash)
+ if (perphase->is_hashed)
+ {
+ Assert(setno == 0);
aggcontext = aggstate->hashcontext;
+ }
else
aggcontext = aggstate->aggcontexts[setno];
@@ -3249,9 +3238,10 @@ ExecBuildAggTransCall(ExprState *state, AggState *aggstate,
if (nullcheck)
{
scratch->opcode = EEOP_AGG_PLAIN_PERGROUP_NULLCHECK;
- scratch->d.agg_plain_pergroup_nullcheck.setoff = setoff;
+ scratch->d.agg_plain_pergroup_nullcheck.pergroups = perphase->pergroups;
/* adjust later */
scratch->d.agg_plain_pergroup_nullcheck.jumpnull = -1;
+ scratch->d.agg_plain_pergroup_nullcheck.setno = setno;
ExprEvalPushStep(state, scratch);
adjust_jumpnull = state->steps_len - 1;
}
@@ -3319,7 +3309,7 @@ ExecBuildAggTransCall(ExprState *state, AggState *aggstate,
scratch->d.agg_trans.pertrans = pertrans;
scratch->d.agg_trans.setno = setno;
- scratch->d.agg_trans.setoff = setoff;
+ scratch->d.agg_trans.pergroups = perphase->pergroups;
scratch->d.agg_trans.transno = transno;
scratch->d.agg_trans.aggcontext = aggcontext;
ExprEvalPushStep(state, scratch);
diff --git a/src/backend/executor/execExprInterp.c b/src/backend/executor/execExprInterp.c
index 113ed1547c..b0dbba4e55 100644
--- a/src/backend/executor/execExprInterp.c
+++ b/src/backend/executor/execExprInterp.c
@@ -1610,9 +1610,9 @@ ExecInterpExpr(ExprState *state, ExprContext *econtext, bool *isnull)
EEO_CASE(EEOP_AGG_PLAIN_PERGROUP_NULLCHECK)
{
- AggState *aggstate = castNode(AggState, state->parent);
- AggStatePerGroup pergroup_allaggs = aggstate->all_pergroups
- [op->d.agg_plain_pergroup_nullcheck.setoff];
+ AggStatePerGroup pergroup_allaggs =
+ op->d.agg_plain_pergroup_nullcheck.pergroups
+ [op->d.agg_plain_pergroup_nullcheck.setno];
if (pergroup_allaggs == NULL)
EEO_JUMP(op->d.agg_plain_pergroup_nullcheck.jumpnull);
@@ -1636,8 +1636,8 @@ ExecInterpExpr(ExprState *state, ExprContext *econtext, bool *isnull)
{
AggState *aggstate = castNode(AggState, state->parent);
AggStatePerTrans pertrans = op->d.agg_trans.pertrans;
- AggStatePerGroup pergroup = &aggstate->all_pergroups
- [op->d.agg_trans.setoff]
+ AggStatePerGroup pergroup = &op->d.agg_trans.pergroups
+ [op->d.agg_trans.setno]
[op->d.agg_trans.transno];
Assert(pertrans->transtypeByVal);
@@ -1665,8 +1665,8 @@ ExecInterpExpr(ExprState *state, ExprContext *econtext, bool *isnull)
{
AggState *aggstate = castNode(AggState, state->parent);
AggStatePerTrans pertrans = op->d.agg_trans.pertrans;
- AggStatePerGroup pergroup = &aggstate->all_pergroups
- [op->d.agg_trans.setoff]
+ AggStatePerGroup pergroup = &op->d.agg_trans.pergroups
+ [op->d.agg_trans.setno]
[op->d.agg_trans.transno];
Assert(pertrans->transtypeByVal);
@@ -1684,8 +1684,8 @@ ExecInterpExpr(ExprState *state, ExprContext *econtext, bool *isnull)
{
AggState *aggstate = castNode(AggState, state->parent);
AggStatePerTrans pertrans = op->d.agg_trans.pertrans;
- AggStatePerGroup pergroup = &aggstate->all_pergroups
- [op->d.agg_trans.setoff]
+ AggStatePerGroup pergroup = &op->d.agg_trans.pergroups
+ [op->d.agg_trans.setno]
[op->d.agg_trans.transno];
Assert(pertrans->transtypeByVal);
@@ -1702,8 +1702,8 @@ ExecInterpExpr(ExprState *state, ExprContext *econtext, bool *isnull)
{
AggState *aggstate = castNode(AggState, state->parent);
AggStatePerTrans pertrans = op->d.agg_trans.pertrans;
- AggStatePerGroup pergroup = &aggstate->all_pergroups
- [op->d.agg_trans.setoff]
+ AggStatePerGroup pergroup = &op->d.agg_trans.pergroups
+ [op->d.agg_trans.setno]
[op->d.agg_trans.transno];
Assert(!pertrans->transtypeByVal);
@@ -1724,8 +1724,8 @@ ExecInterpExpr(ExprState *state, ExprContext *econtext, bool *isnull)
{
AggState *aggstate = castNode(AggState, state->parent);
AggStatePerTrans pertrans = op->d.agg_trans.pertrans;
- AggStatePerGroup pergroup = &aggstate->all_pergroups
- [op->d.agg_trans.setoff]
+ AggStatePerGroup pergroup = &op->d.agg_trans.pergroups
+ [op->d.agg_trans.setno]
[op->d.agg_trans.transno];
Assert(!pertrans->transtypeByVal);
@@ -1742,8 +1742,8 @@ ExecInterpExpr(ExprState *state, ExprContext *econtext, bool *isnull)
{
AggState *aggstate = castNode(AggState, state->parent);
AggStatePerTrans pertrans = op->d.agg_trans.pertrans;
- AggStatePerGroup pergroup = &aggstate->all_pergroups
- [op->d.agg_trans.setoff]
+ AggStatePerGroup pergroup = &op->d.agg_trans.pergroups
+ [op->d.agg_trans.setno]
[op->d.agg_trans.transno];
Assert(!pertrans->transtypeByVal);
diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c
index f7af5eebd0..20c5eb98b3 100644
--- a/src/backend/executor/nodeAgg.c
+++ b/src/backend/executor/nodeAgg.c
@@ -227,6 +227,7 @@
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "nodes/nodeFuncs.h"
+#include "nodes/pathnodes.h"
#include "optimizer/optimizer.h"
#include "parser/parse_agg.h"
#include "parser/parse_coerce.h"
@@ -263,7 +264,7 @@ static void finalize_partialaggregate(AggState *aggstate,
AggStatePerAgg peragg,
AggStatePerGroup pergroupstate,
Datum *resultVal, bool *resultIsNull);
-static void prepare_hash_slot(AggState *aggstate);
+static void prepare_hash_slot(AggState *aggstate, AggStatePerPhaseHash perhash);
static void prepare_projection_slot(AggState *aggstate,
TupleTableSlot *slot,
int currentSet);
@@ -274,12 +275,13 @@ static TupleTableSlot *project_aggregates(AggState *aggstate);
static Bitmapset *find_unaggregated_cols(AggState *aggstate);
static bool find_unaggregated_cols_walker(Node *node, Bitmapset **colnos);
static void build_hash_tables(AggState *aggstate);
-static void build_hash_table(AggState *aggstate, int setno, long nbuckets);
-static AggStatePerGroup lookup_hash_entry(AggState *aggstate, uint32 hash);
-static void lookup_hash_entries(AggState *aggstate);
+static void build_hash_table(AggState *aggstate, AggStatePerPhaseHash perhash);
+static void lookup_hash_entries(AggState *aggstate, List *perhashes);
+static void lookup_hash_entry(AggState *aggstate, AggStatePerPhaseHash perhash,
+ uint32 hash);
static TupleTableSlot *agg_retrieve_direct(AggState *aggstate);
-static void agg_sort_input(AggState *aggstate);
static void agg_fill_hash_table(AggState *aggstate);
+static void agg_sort_input(AggState *aggstate);
static TupleTableSlot *agg_retrieve_hash_table(AggState *aggstate);
static Datum GetAggInitVal(Datum textInitVal, Oid transtype);
static void build_pertrans_for_aggref(AggStatePerTrans pertrans,
@@ -310,7 +312,10 @@ select_current_set(AggState *aggstate, int setno, bool is_hash)
* ExecAggPlainTransByRef().
*/
if (is_hash)
+ {
+ Assert(setno == 0);
aggstate->curaggcontext = aggstate->hashcontext;
+ }
else
aggstate->curaggcontext = aggstate->aggcontexts[setno];
@@ -318,72 +323,75 @@ select_current_set(AggState *aggstate, int setno, bool is_hash)
}
/*
- * Switch to phase "newphase", which must either be 0 or 1 (to reset) or
+ * Switch to phase "newphase", which must either be 0 (to reset) or
* current_phase + 1. Juggle the tuplesorts accordingly.
- *
- * Phase 0 is for hashing, which we currently handle last in the AGG_MIXED
- * case, so when entering phase 0, all we need to do is drop open sorts.
*/
static void
initialize_phase(AggState *aggstate, int newphase)
{
- Assert(newphase <= 1 || newphase == aggstate->current_phase + 1);
+ AggStatePerPhase current_phase;
+ AggStatePerPhaseSort persort;
+
+ Assert(newphase == 0 || newphase == aggstate->current_phase + 1);
+
+ /* Don't use aggstate->phase here, it might not be initialized yet*/
+ current_phase = aggstate->phases[aggstate->current_phase];
/*
* Whatever the previous state, we're now done with whatever input
- * tuplesort was in use.
+ * tuplesort was in use, cleanup them.
+ *
+ * Note: we keep the first tuplesort/tuplestore, this will benifit the
+ * rescan in some cases without resorting the input again.
*/
- if (aggstate->sort_in)
- {
- tuplesort_end(aggstate->sort_in);
- aggstate->sort_in = NULL;
- }
-
- if (newphase <= 1)
+ if (!current_phase->is_hashed && aggstate->current_phase > 0)
{
- /*
- * Discard any existing output tuplesort.
- */
- if (aggstate->sort_out)
+ persort = (AggStatePerPhaseSort) current_phase;
+ if (persort->sort_in)
{
- tuplesort_end(aggstate->sort_out);
- aggstate->sort_out = NULL;
+ tuplesort_end(persort->sort_in);
+ persort->sort_in = NULL;
}
}
- else
- {
- /*
- * The old output tuplesort becomes the new input one, and this is the
- * right time to actually sort it.
- */
- aggstate->sort_in = aggstate->sort_out;
- aggstate->sort_out = NULL;
- Assert(aggstate->sort_in);
- tuplesort_performsort(aggstate->sort_in);
- }
+
+ /* advance to next phase */
+ aggstate->current_phase = newphase;
+ aggstate->phase = aggstate->phases[newphase];
+
+ if (aggstate->phase->is_hashed)
+ return;
+
+ /* New phase is not hashed */
+ persort = (AggStatePerPhaseSort) aggstate->phase;
+
+ /* This is the right time to actually sort it. */
+ if (persort->sort_in)
+ tuplesort_performsort(persort->sort_in);
/*
- * If this isn't the last phase, we need to sort appropriately for the
+ * If copy_out is set, we need to sort appropriately for the
* next phase in sequence.
*/
- if (newphase > 0 && newphase < aggstate->numphases - 1)
- {
- Sort *sortnode = (Sort *) aggstate->phases[newphase + 1].aggnode->sortnode;
- PlanState *outerNode = outerPlanState(aggstate);
- TupleDesc tupDesc = ExecGetResultType(outerNode);
-
- aggstate->sort_out = tuplesort_begin_heap(tupDesc,
- sortnode->numCols,
- sortnode->sortColIdx,
- sortnode->sortOperators,
- sortnode->collations,
- sortnode->nullsFirst,
- work_mem,
- NULL, false);
+ if (persort->copy_out)
+ {
+ AggStatePerPhaseSort next =
+ (AggStatePerPhaseSort) aggstate->phases[newphase + 1];
+ Sort *sortnode = (Sort *) next->phasedata.aggnode->sortnode;
+ PlanState *outerNode = outerPlanState(aggstate);
+ TupleDesc tupDesc = ExecGetResultType(outerNode);
+
+ Assert(!next->phasedata.is_hashed);
+
+ if (!next->sort_in)
+ next->sort_in = tuplesort_begin_heap(tupDesc,
+ sortnode->numCols,
+ sortnode->sortColIdx,
+ sortnode->sortOperators,
+ sortnode->collations,
+ sortnode->nullsFirst,
+ work_mem,
+ NULL, false);
}
-
- aggstate->current_phase = newphase;
- aggstate->phase = &aggstate->phases[newphase];
}
/*
@@ -398,12 +406,16 @@ static TupleTableSlot *
fetch_input_tuple(AggState *aggstate)
{
TupleTableSlot *slot;
+ AggStatePerPhaseSort current_phase;
+
+ Assert(!aggstate->phase->is_hashed);
+ current_phase = (AggStatePerPhaseSort) aggstate->phase;
- if (aggstate->sort_in)
+ if (current_phase->sort_in)
{
/* make sure we check for interrupts in either path through here */
CHECK_FOR_INTERRUPTS();
- if (!tuplesort_gettupleslot(aggstate->sort_in, true, false,
+ if (!tuplesort_gettupleslot(current_phase->sort_in, true, false,
aggstate->sort_slot, NULL))
return NULL;
slot = aggstate->sort_slot;
@@ -411,8 +423,13 @@ fetch_input_tuple(AggState *aggstate)
else
slot = ExecProcNode(outerPlanState(aggstate));
- if (!TupIsNull(slot) && aggstate->sort_out)
- tuplesort_puttupleslot(aggstate->sort_out, slot);
+ if (!TupIsNull(slot) && current_phase->copy_out)
+ {
+ AggStatePerPhaseSort next =
+ (AggStatePerPhaseSort) aggstate->phases[aggstate->current_phase + 1];
+ Assert(!next->phasedata.is_hashed);
+ tuplesort_puttupleslot(next->sort_in, slot);
+ }
return slot;
}
@@ -518,7 +535,7 @@ initialize_aggregates(AggState *aggstate,
int numReset)
{
int transno;
- int numGroupingSets = Max(aggstate->phase->numsets, 1);
+ int numGroupingSets = aggstate->phase->numsets;
int setno = 0;
int numTrans = aggstate->numtrans;
AggStatePerTrans transstates = aggstate->pertrans;
@@ -1046,10 +1063,9 @@ finalize_partialaggregate(AggState *aggstate,
* hashslot. This is necessary to compute the hash or perform a lookup.
*/
static void
-prepare_hash_slot(AggState *aggstate)
+prepare_hash_slot(AggState *aggstate, AggStatePerPhaseHash perhash)
{
TupleTableSlot *inputslot = aggstate->tmpcontext->ecxt_outertuple;
- AggStatePerHash perhash = &aggstate->perhash[aggstate->current_set];
TupleTableSlot *hashslot = perhash->hashslot;
int i;
@@ -1283,18 +1299,22 @@ find_unaggregated_cols_walker(Node *node, Bitmapset **colnos)
static void
build_hash_tables(AggState *aggstate)
{
- int setno;
+ int phaseidx;
- for (setno = 0; setno < aggstate->num_hashes; ++setno)
+ for (phaseidx = 0; phaseidx < aggstate->numphases; phaseidx++)
{
- AggStatePerHash perhash = &aggstate->perhash[setno];
+ AggStatePerPhase phase;
+ AggStatePerPhaseHash perhash;
- Assert(perhash->aggnode->numGroups > 0);
+ phase = aggstate->phases[phaseidx];
+ if (!phase->is_hashed)
+ continue;
+ perhash = (AggStatePerPhaseHash) phase;
if (perhash->hashtable)
ResetTupleHashTable(perhash->hashtable);
else
- build_hash_table(aggstate, setno, perhash->aggnode->numGroups);
+ build_hash_table(aggstate, perhash);
}
}
@@ -1302,9 +1322,8 @@ build_hash_tables(AggState *aggstate)
* Build a single hashtable for this grouping set.
*/
static void
-build_hash_table(AggState *aggstate, int setno, long nbuckets)
+build_hash_table(AggState *aggstate, AggStatePerPhaseHash perhash)
{
- AggStatePerHash perhash = &aggstate->perhash[setno];
MemoryContext metacxt = aggstate->ss.ps.state->es_query_cxt;
MemoryContext hashcxt = aggstate->hashcontext->ecxt_per_tuple_memory;
MemoryContext tmpcxt = aggstate->tmpcontext->ecxt_per_tuple_memory;
@@ -1328,8 +1347,8 @@ build_hash_table(AggState *aggstate, int setno, long nbuckets)
perhash->hashGrpColIdxHash,
perhash->eqfuncoids,
perhash->hashfunctions,
- perhash->aggnode->grpCollations,
- nbuckets,
+ perhash->phasedata.aggnode->grpCollations,
+ perhash->phasedata.aggnode->numGroups,
additionalsize,
metacxt,
hashcxt,
@@ -1367,23 +1386,29 @@ find_hash_columns(AggState *aggstate)
{
Bitmapset *base_colnos;
List *outerTlist = outerPlanState(aggstate)->plan->targetlist;
- int numHashes = aggstate->num_hashes;
EState *estate = aggstate->ss.ps.state;
int j;
/* Find Vars that will be needed in tlist and qual */
base_colnos = find_unaggregated_cols(aggstate);
- for (j = 0; j < numHashes; ++j)
+ for (j = 0; j < aggstate->numphases; ++j)
{
- AggStatePerHash perhash = &aggstate->perhash[j];
+ AggStatePerPhase perphase = aggstate->phases[j];
+ AggStatePerPhaseHash perhash;
Bitmapset *colnos = bms_copy(base_colnos);
- AttrNumber *grpColIdx = perhash->aggnode->grpColIdx;
+ Bitmapset *grouped_cols = perphase->grouped_cols[0];
+ AttrNumber *grpColIdx = perphase->aggnode->grpColIdx;
List *hashTlist = NIL;
+ ListCell *lc;
TupleDesc hashDesc;
int maxCols;
int i;
+ if (!perphase->is_hashed)
+ continue;
+
+ perhash = (AggStatePerPhaseHash) perphase;
perhash->largestGrpColIdx = 0;
/*
@@ -1393,18 +1418,12 @@ find_hash_columns(AggState *aggstate)
* there'd be no point storing them. Use prepare_projection_slot's
* logic to determine which.
*/
- if (aggstate->phases[0].grouped_cols)
+ foreach(lc, aggstate->all_grouped_cols)
{
- Bitmapset *grouped_cols = aggstate->phases[0].grouped_cols[j];
- ListCell *lc;
-
- foreach(lc, aggstate->all_grouped_cols)
- {
- int attnum = lfirst_int(lc);
+ int attnum = lfirst_int(lc);
- if (!bms_is_member(attnum, grouped_cols))
- colnos = bms_del_member(colnos, attnum);
- }
+ if (!bms_is_member(attnum, grouped_cols))
+ colnos = bms_del_member(colnos, attnum);
}
/*
@@ -1460,7 +1479,7 @@ find_hash_columns(AggState *aggstate)
hashDesc = ExecTypeFromTL(hashTlist);
execTuplesHashPrepare(perhash->numCols,
- perhash->aggnode->grpOperators,
+ perphase->aggnode->grpOperators,
&perhash->eqfuncoids,
&perhash->hashfunctions);
perhash->hashslot =
@@ -1497,10 +1516,9 @@ hash_agg_entry_size(int numAggs, Size tupleWidth, Size transitionSpace)
* When called, CurrentMemoryContext should be the per-query context. The
* already-calculated hash value for the tuple must be specified.
*/
-static AggStatePerGroup
-lookup_hash_entry(AggState *aggstate, uint32 hash)
+static void
+lookup_hash_entry(AggState *aggstate, AggStatePerPhaseHash perhash, uint32 hash)
{
- AggStatePerHash perhash = &aggstate->perhash[aggstate->current_set];
TupleTableSlot *hashslot = perhash->hashslot;
TupleHashEntryData *entry;
bool isnew;
@@ -1532,7 +1550,7 @@ lookup_hash_entry(AggState *aggstate, uint32 hash)
}
}
- return entry->additional;
+ perhash->phasedata.pergroups[0] = entry->additional;
}
/*
@@ -1542,21 +1560,19 @@ lookup_hash_entry(AggState *aggstate, uint32 hash)
* Be aware that lookup_hash_entry can reset the tmpcontext.
*/
static void
-lookup_hash_entries(AggState *aggstate)
+lookup_hash_entries(AggState *aggstate, List *perhashes)
{
- int numHashes = aggstate->num_hashes;
- AggStatePerGroup *pergroup = aggstate->hash_pergroup;
- int setno;
+ ListCell *lc;
- for (setno = 0; setno < numHashes; setno++)
+ foreach (lc, perhashes)
{
- AggStatePerHash perhash = &aggstate->perhash[setno];
uint32 hash;
+ AggStatePerPhaseHash perhash = (AggStatePerPhaseHash) lfirst(lc);
- select_current_set(aggstate, setno, true);
- prepare_hash_slot(aggstate);
+ select_current_set(aggstate, 0, true);
+ prepare_hash_slot(aggstate, perhash);
hash = TupleHashTableHash(perhash->hashtable, perhash->hashslot);
- pergroup[setno] = lookup_hash_entry(aggstate, hash);
+ lookup_hash_entry(aggstate, perhash, hash);
}
}
@@ -1589,12 +1605,11 @@ ExecAgg(PlanState *pstate)
case AGG_HASHED:
if (!node->table_filled)
agg_fill_hash_table(node);
- /* FALLTHROUGH */
- case AGG_MIXED:
result = agg_retrieve_hash_table(node);
break;
case AGG_PLAIN:
case AGG_SORTED:
+ case AGG_MIXED:
if (!node->input_sorted)
agg_sort_input(node);
result = agg_retrieve_direct(node);
@@ -1622,8 +1637,8 @@ agg_retrieve_direct(AggState *aggstate)
TupleTableSlot *outerslot;
TupleTableSlot *firstSlot;
TupleTableSlot *result;
- bool hasGroupingSets = aggstate->phase->numsets > 0;
- int numGroupingSets = Max(aggstate->phase->numsets, 1);
+ bool hasGroupingSets = aggstate->phase->aggnode->groupingSets != NULL;
+ int numGroupingSets = aggstate->phase->numsets;
int currentSet;
int nextSetSize;
int numReset;
@@ -1640,7 +1655,7 @@ agg_retrieve_direct(AggState *aggstate)
tmpcontext = aggstate->tmpcontext;
peragg = aggstate->peragg;
- pergroups = aggstate->pergroups;
+ pergroups = aggstate->phase->pergroups;
firstSlot = aggstate->ss.ss_ScanTupleSlot;
/*
@@ -1698,25 +1713,32 @@ agg_retrieve_direct(AggState *aggstate)
{
if (aggstate->current_phase < aggstate->numphases - 1)
{
+ /* Advance to the next phase */
initialize_phase(aggstate, aggstate->current_phase + 1);
- aggstate->input_done = false;
- aggstate->projected_set = -1;
- numGroupingSets = Max(aggstate->phase->numsets, 1);
- node = aggstate->phase->aggnode;
- numReset = numGroupingSets;
- }
- else if (aggstate->aggstrategy == AGG_MIXED)
- {
- /*
- * Mixed mode; we've output all the grouped stuff and have
- * full hashtables, so switch to outputting those.
- */
- initialize_phase(aggstate, 0);
- aggstate->table_filled = true;
- ResetTupleHashIterator(aggstate->perhash[0].hashtable,
- &aggstate->perhash[0].hashiter);
- select_current_set(aggstate, 0, true);
- return agg_retrieve_hash_table(aggstate);
+
+ /* Check whether new phase is an AGG_HASHED */
+ if (!aggstate->phase->is_hashed)
+ {
+ aggstate->input_done = false;
+ aggstate->projected_set = -1;
+ numGroupingSets = aggstate->phase->numsets;
+ node = aggstate->phase->aggnode;
+ numReset = numGroupingSets;
+ pergroups = aggstate->phase->pergroups;
+ }
+ else
+ {
+ AggStatePerPhaseHash perhash = (AggStatePerPhaseHash) aggstate->phase;
+
+ /*
+ * Mixed mode; we've output all the grouped stuff and have
+ * full hashtables, so switch to outputting those.
+ */
+ aggstate->table_filled = true;
+ ResetTupleHashIterator(perhash->hashtable, &perhash->hashiter);
+ select_current_set(aggstate, 0, true);
+ return agg_retrieve_hash_table(aggstate);
+ }
}
else
{
@@ -1755,11 +1777,11 @@ agg_retrieve_direct(AggState *aggstate)
*/
tmpcontext->ecxt_innertuple = econtext->ecxt_outertuple;
if (aggstate->input_done ||
- (node->aggstrategy != AGG_PLAIN &&
+ (aggstate->phase->aggnode->numCols > 0 &&
aggstate->projected_set != -1 &&
aggstate->projected_set < (numGroupingSets - 1) &&
nextSetSize > 0 &&
- !ExecQualAndReset(aggstate->phase->eqfunctions[nextSetSize - 1],
+ !ExecQualAndReset(((AggStatePerPhaseSort) aggstate->phase)->eqfunctions[nextSetSize - 1],
tmpcontext)))
{
aggstate->projected_set += 1;
@@ -1862,13 +1884,13 @@ agg_retrieve_direct(AggState *aggstate)
for (;;)
{
/*
- * During phase 1 only of a mixed agg, we need to update
- * hashtables as well in advance_aggregates.
+ * If current phase can do transition concurrently, we need
+ * to update hashtables as well in advance_aggregates.
*/
- if (aggstate->aggstrategy == AGG_MIXED &&
- aggstate->current_phase == 1)
+ if (aggstate->phase->concurrent_hashes)
{
- lookup_hash_entries(aggstate);
+ lookup_hash_entries(aggstate,
+ aggstate->phase->concurrent_hashes);
}
/* Advance the aggregates (or combine functions) */
@@ -1899,10 +1921,10 @@ agg_retrieve_direct(AggState *aggstate)
* If we are grouping, check whether we've crossed a group
* boundary.
*/
- if (node->aggstrategy != AGG_PLAIN)
+ if (aggstate->phase->aggnode->numCols > 0)
{
tmpcontext->ecxt_innertuple = firstSlot;
- if (!ExecQual(aggstate->phase->eqfunctions[node->numCols - 1],
+ if (!ExecQual(((AggStatePerPhaseSort) aggstate->phase)->eqfunctions[node->numCols - 1],
tmpcontext))
{
aggstate->grp_firstTuple = ExecCopySlotHeapTuple(outerslot);
@@ -1951,24 +1973,31 @@ agg_retrieve_direct(AggState *aggstate)
static void
agg_sort_input(AggState *aggstate)
{
- AggStatePerPhase phase = &aggstate->phases[1];
+ AggStatePerPhase phase = aggstate->phases[0];
+ AggStatePerPhaseSort persort = (AggStatePerPhaseSort) phase;
TupleDesc tupDesc;
Sort *sortnode;
+ bool randomAccess;
Assert(!aggstate->input_sorted);
+ Assert(!phase->is_hashed);
Assert(phase->aggnode->sortnode);
sortnode = (Sort *) phase->aggnode->sortnode;
tupDesc = ExecGetResultType(outerPlanState(aggstate));
-
- aggstate->sort_in = tuplesort_begin_heap(tupDesc,
- sortnode->numCols,
- sortnode->sortColIdx,
- sortnode->sortOperators,
- sortnode->collations,
- sortnode->nullsFirst,
- work_mem,
- NULL, false);
+ randomAccess = (aggstate->eflags & (EXEC_FLAG_REWIND |
+ EXEC_FLAG_BACKWARD |
+ EXEC_FLAG_MARK)) != 0;
+
+
+ persort->sort_in = tuplesort_begin_heap(tupDesc,
+ sortnode->numCols,
+ sortnode->sortColIdx,
+ sortnode->sortOperators,
+ sortnode->collations,
+ sortnode->nullsFirst,
+ work_mem,
+ NULL, randomAccess);
for (;;)
{
TupleTableSlot *outerslot;
@@ -1977,11 +2006,11 @@ agg_sort_input(AggState *aggstate)
if (TupIsNull(outerslot))
break;
- tuplesort_puttupleslot(aggstate->sort_in, outerslot);
+ tuplesort_puttupleslot(persort->sort_in, outerslot);
}
/* Sort the first phase */
- tuplesort_performsort(aggstate->sort_in);
+ tuplesort_performsort(persort->sort_in);
/* Mark the input to be sorted */
aggstate->input_sorted = true;
@@ -1993,8 +2022,14 @@ agg_sort_input(AggState *aggstate)
static void
agg_fill_hash_table(AggState *aggstate)
{
+ AggStatePerPhaseHash currentphase;
TupleTableSlot *outerslot;
ExprContext *tmpcontext = aggstate->tmpcontext;
+ List *concurrent_hashes = aggstate->phase->concurrent_hashes;
+
+ /* Current phase must be the first phase */
+ Assert(aggstate->current_phase == 0);
+ currentphase = (AggStatePerPhaseHash) aggstate->phase;
/*
* Process each outer-plan tuple, and then fetch the next one, until we
@@ -2002,15 +2037,25 @@ agg_fill_hash_table(AggState *aggstate)
*/
for (;;)
{
- outerslot = fetch_input_tuple(aggstate);
+ int hash;
+
+ outerslot = ExecProcNode(outerPlanState(aggstate));
if (TupIsNull(outerslot))
break;
/* set up for lookup_hash_entries and advance_aggregates */
tmpcontext->ecxt_outertuple = outerslot;
- /* Find or build hashtable entries */
- lookup_hash_entries(aggstate);
+ /* Find hashtable entry of current phase */
+ select_current_set(aggstate, 0, true);
+ prepare_hash_slot(aggstate, currentphase);
+ hash = TupleHashTableHash(currentphase->hashtable, currentphase->hashslot);
+ lookup_hash_entry(aggstate, currentphase, hash);
+
+
+ /* Find or build hashtable entries of concurrent hash phases */
+ if (concurrent_hashes)
+ lookup_hash_entries(aggstate, concurrent_hashes);
/* Advance the aggregates (or combine functions) */
advance_aggregates(aggstate);
@@ -2025,8 +2070,7 @@ agg_fill_hash_table(AggState *aggstate)
aggstate->table_filled = true;
/* Initialize to walk the first hash table */
select_current_set(aggstate, 0, true);
- ResetTupleHashIterator(aggstate->perhash[0].hashtable,
- &aggstate->perhash[0].hashiter);
+ ResetTupleHashIterator(currentphase->hashtable, ¤tphase->hashiter);
}
/*
@@ -2041,7 +2085,7 @@ agg_retrieve_hash_table(AggState *aggstate)
TupleHashEntryData *entry;
TupleTableSlot *firstSlot;
TupleTableSlot *result;
- AggStatePerHash perhash;
+ AggStatePerPhaseHash perhash;
/*
* get state info from node.
@@ -2052,11 +2096,7 @@ agg_retrieve_hash_table(AggState *aggstate)
peragg = aggstate->peragg;
firstSlot = aggstate->ss.ss_ScanTupleSlot;
- /*
- * Note that perhash (and therefore anything accessed through it) can
- * change inside the loop, as we change between grouping sets.
- */
- perhash = &aggstate->perhash[aggstate->current_set];
+ perhash = (AggStatePerPhaseHash) aggstate->phase;
/*
* We loop retrieving groups until we find one satisfying
@@ -2075,18 +2115,15 @@ agg_retrieve_hash_table(AggState *aggstate)
entry = ScanTupleHashTable(perhash->hashtable, &perhash->hashiter);
if (entry == NULL)
{
- int nextset = aggstate->current_set + 1;
-
- if (nextset < aggstate->num_hashes)
+ if (aggstate->current_phase + 1 < aggstate->numphases)
{
/*
* Switch to next grouping set, reinitialize, and restart the
* loop.
*/
- select_current_set(aggstate, nextset, true);
-
- perhash = &aggstate->perhash[aggstate->current_set];
-
+ select_current_set(aggstate, 0, true);
+ initialize_phase(aggstate, aggstate->current_phase + 1);
+ perhash = (AggStatePerPhaseHash) aggstate->phase;
ResetTupleHashIterator(perhash->hashtable, &perhash->hashiter);
continue;
@@ -2165,23 +2202,19 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
AggState *aggstate;
AggStatePerAgg peraggs;
AggStatePerTrans pertransstates;
- AggStatePerGroup *pergroups;
Plan *outerPlan;
ExprContext *econtext;
TupleDesc scanDesc;
- Agg *firstSortAgg;
int numaggs,
transno,
aggno;
- int phase;
int phaseidx;
ListCell *l;
Bitmapset *all_grouped_cols = NULL;
int numGroupingSets = 1;
- int numPhases;
- int numHashes;
int i = 0;
int j = 0;
+ bool need_extra_slot = false;
bool use_hashing = (node->aggstrategy == AGG_HASHED ||
node->aggstrategy == AGG_MIXED);
@@ -2210,24 +2243,13 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
aggstate->curpertrans = NULL;
aggstate->input_done = false;
aggstate->agg_done = false;
- aggstate->pergroups = NULL;
aggstate->grp_firstTuple = NULL;
- aggstate->sort_in = NULL;
- aggstate->sort_out = NULL;
aggstate->input_sorted = true;
-
- /*
- * phases[0] always exists, but is dummy in sorted/plain mode
- */
- numPhases = (use_hashing ? 1 : 2);
- numHashes = (use_hashing ? 1 : 0);
-
- firstSortAgg = node->aggstrategy == AGG_SORTED ? node : NULL;
+ aggstate->eflags = eflags;
/*
* Calculate the maximum number of grouping sets in any phase; this
- * determines the size of some allocations. Also calculate the number of
- * phases, since all hashed/mixed nodes contribute to only a single phase.
+ * determines the size of some allocations.
*/
if (node->groupingSets)
{
@@ -2240,31 +2262,19 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
numGroupingSets = Max(numGroupingSets,
list_length(agg->groupingSets));
- /*
- * additional AGG_HASHED aggs become part of phase 0, but all
- * others add an extra phase.
- */
if (agg->aggstrategy != AGG_HASHED)
- {
- ++numPhases;
-
- if (!firstSortAgg)
- firstSortAgg = agg;
-
- }
- else
- ++numHashes;
+ need_extra_slot = true;
}
}
aggstate->maxsets = numGroupingSets;
- aggstate->numphases = numPhases;
-
+ aggstate->numphases = 1 + list_length(node->chain);
+
/*
- * The first SORTED phase is not sorted, agg need to do its own sort. See
+ * The first phase is not sorted, agg need to do its own sort. See
* agg_sort_input(), this can only happen in groupingsets case.
*/
- if (firstSortAgg && firstSortAgg->sortnode)
+ if (node->sortnode)
aggstate->input_sorted = false;
aggstate->aggcontexts = (ExprContext **)
@@ -2325,11 +2335,10 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
scanDesc = aggstate->ss.ss_ScanTupleSlot->tts_tupleDescriptor;
/*
- * If there are more than two phases (including a potential dummy phase
- * 0), input will be resorted using tuplesort. Need a slot for that.
+ * An extra slot is needed if 1) agg need to do its own sort 2) agg
+ * has more than one non-hashed phases
*/
- if (numPhases > 2 ||
- !aggstate->input_sorted)
+ if (node->sortnode || need_extra_slot)
{
aggstate->sort_slot = ExecInitExtraTupleSlot(estate, scanDesc,
&TTSOpsMinimalTuple);
@@ -2381,76 +2390,94 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
numaggs = aggstate->numaggs;
Assert(numaggs == list_length(aggstate->aggs));
- /*
+ /*
* For each phase, prepare grouping set data and fmgr lookup data for
* compare functions. Accumulate all_grouped_cols in passing.
*/
- aggstate->phases = palloc0(numPhases * sizeof(AggStatePerPhaseData));
-
- aggstate->num_hashes = numHashes;
- if (numHashes)
- {
- aggstate->perhash = palloc0(sizeof(AggStatePerHashData) * numHashes);
- aggstate->phases[0].numsets = 0;
- aggstate->phases[0].gset_lengths = palloc(numHashes * sizeof(int));
- aggstate->phases[0].grouped_cols = palloc(numHashes * sizeof(Bitmapset *));
- }
+ aggstate->phases = palloc0(aggstate->numphases * sizeof(AggStatePerPhase));
- phase = 0;
- for (phaseidx = 0; phaseidx <= list_length(node->chain); ++phaseidx)
+ for (phaseidx = 0; phaseidx < aggstate->numphases; phaseidx++)
{
Agg *aggnode;
+ AggStatePerPhase phasedata = NULL;
if (phaseidx > 0)
aggnode = list_nth_node(Agg, node->chain, phaseidx - 1);
else
aggnode = node;
- if (aggnode->aggstrategy == AGG_HASHED
- || aggnode->aggstrategy == AGG_MIXED)
+ if (aggnode->aggstrategy == AGG_HASHED)
{
- AggStatePerPhase phasedata = &aggstate->phases[0];
- AggStatePerHash perhash;
- Bitmapset *cols = NULL;
-
- Assert(phase == 0);
- i = phasedata->numsets++;
- perhash = &aggstate->perhash[i];
+ AggStatePerPhaseHash perhash;
+ Bitmapset *cols = NULL;
- /* phase 0 always points to the "real" Agg in the hash case */
- phasedata->aggnode = node;
- phasedata->aggstrategy = node->aggstrategy;
+ perhash = (AggStatePerPhaseHash) palloc0(sizeof(AggStatePerPhaseHashData));
+ phasedata = (AggStatePerPhase) perhash;
+ phasedata->is_hashed = true;
+ phasedata->aggnode = aggnode;
+ phasedata->aggstrategy = aggnode->aggstrategy;
- /* but the actual Agg node representing this hash is saved here */
- perhash->aggnode = aggnode;
+ /* AGG_HASHED always has only one set */
+ phasedata->numsets = 1;
- phasedata->gset_lengths[i] = perhash->numCols = aggnode->numCols;
+ phasedata->gset_lengths = palloc(sizeof(int));
+ phasedata->gset_lengths[0] = perhash->numCols = aggnode->numCols;
+ phasedata->grouped_cols = palloc(sizeof(Bitmapset *));
for (j = 0; j < aggnode->numCols; ++j)
cols = bms_add_member(cols, aggnode->grpColIdx[j]);
-
- phasedata->grouped_cols[i] = cols;
+ phasedata->grouped_cols[0] = cols;
all_grouped_cols = bms_add_members(all_grouped_cols, cols);
- continue;
+
+ /*
+ * Initialize pergroup state. For AGG_HASHED, all groups do transition
+ * on the fly, all pergroup states are kept in hashtable, everytime
+ * a tuple is processed, lookup_hash_entry() choose one group and
+ * set phasedata->pergroups[0], then advance_aggregates can use it
+ * to do transition in this group.
+ * We do not need to allocate a real pergroup and set the pointer
+ * here, there are too many pergroup states, lookup_hash_entry() will
+ * allocate it.
+ */
+ phasedata->pergroups =
+ (AggStatePerGroup *) palloc0(sizeof(AggStatePerGroup));
+
+ /*
+ * Hash aggregate does not require the order of input tuples, so
+ * we can do the transition immediately when a tuple is fetched,
+ * which means we can do the transition concurrently with the
+ * first phase.
+ */
+ if (phaseidx > 0)
+ {
+ aggstate->phases[0]->concurrent_hashes =
+ lappend(aggstate->phases[0]->concurrent_hashes, perhash);
+ /* current phase don't need to build transition functions */
+ phasedata->skip_build_trans = true;
+ }
}
else
{
- AggStatePerPhase phasedata = &aggstate->phases[++phase];
- int num_sets;
+ AggStatePerPhaseSort persort;
- phasedata->numsets = num_sets = list_length(aggnode->groupingSets);
+ persort = (AggStatePerPhaseSort) palloc0(sizeof(AggStatePerPhaseSortData));
+ phasedata = (AggStatePerPhase) persort;
+ phasedata->is_hashed = false;
+ phasedata->aggnode = aggnode;
+ phasedata->aggstrategy = aggnode->aggstrategy;
- if (num_sets)
+ if (aggnode->groupingSets)
{
- phasedata->gset_lengths = palloc(num_sets * sizeof(int));
- phasedata->grouped_cols = palloc(num_sets * sizeof(Bitmapset *));
+ phasedata->numsets = list_length(aggnode->groupingSets);
+ phasedata->gset_lengths = palloc(phasedata->numsets * sizeof(int));
+ phasedata->grouped_cols = palloc(phasedata->numsets * sizeof(Bitmapset *));
i = 0;
foreach(l, aggnode->groupingSets)
{
- int current_length = list_length(lfirst(l));
- Bitmapset *cols = NULL;
+ int current_length = list_length(lfirst(l));
+ Bitmapset *cols = NULL;
/* planner forces this to be correct */
for (j = 0; j < current_length; ++j)
@@ -2467,37 +2494,49 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
}
else
{
- Assert(phaseidx == 0);
-
+ phasedata->numsets = 1;
phasedata->gset_lengths = NULL;
phasedata->grouped_cols = NULL;
}
+ /*
+ * Initialize pergroup states for AGG_SORTED/AGG_PLAIN/AGG_MIXED
+ * phases, each set only have one group on the fly, all groups in
+ * a set can reuse a pergroup state. Unlike AGG_HASHED, we
+ * pre-allocate the pergroup states here.
+ */
+ phasedata->pergroups =
+ (AggStatePerGroup *) palloc0(sizeof(AggStatePerGroup) * phasedata->numsets);
+
+ for (i = 0; i < phasedata->numsets; i++)
+ {
+ phasedata->pergroups[i] =
+ (AggStatePerGroup) palloc0(sizeof(AggStatePerGroupData) * numaggs);
+ }
+
/*
* If we are grouping, precompute fmgr lookup data for inner loop.
*/
- if (aggnode->aggstrategy == AGG_SORTED)
+ if (aggnode->numCols > 0)
{
int i = 0;
- Assert(aggnode->numCols > 0);
-
/*
* Build a separate function for each subset of columns that
* need to be compared.
*/
- phasedata->eqfunctions =
+ persort->eqfunctions =
(ExprState **) palloc0(aggnode->numCols * sizeof(ExprState *));
/* for each grouping set */
- for (i = 0; i < phasedata->numsets; i++)
+ for (i = 0; i < phasedata->numsets && phasedata->gset_lengths; i++)
{
int length = phasedata->gset_lengths[i];
- if (phasedata->eqfunctions[length - 1] != NULL)
+ if (persort->eqfunctions[length - 1] != NULL)
continue;
- phasedata->eqfunctions[length - 1] =
+ persort->eqfunctions[length - 1] =
execTuplesMatchPrepare(scanDesc,
length,
aggnode->grpColIdx,
@@ -2507,9 +2546,9 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
}
/* and for all grouped columns, unless already computed */
- if (phasedata->eqfunctions[aggnode->numCols - 1] == NULL)
+ if (persort->eqfunctions[aggnode->numCols - 1] == NULL)
{
- phasedata->eqfunctions[aggnode->numCols - 1] =
+ persort->eqfunctions[aggnode->numCols - 1] =
execTuplesMatchPrepare(scanDesc,
aggnode->numCols,
aggnode->grpColIdx,
@@ -2519,9 +2558,23 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
}
}
- phasedata->aggnode = aggnode;
- phasedata->aggstrategy = aggnode->aggstrategy;
+ /*
+ * For non-first AGG_SORTED phase, it processes the same input
+ * tuples with previous phase except that it need to resort the
+ * input tuples. Tell the previous phase to copy out the tuples.
+ */
+ if (phaseidx > 0)
+ {
+ AggStatePerPhaseSort prev =
+ (AggStatePerPhaseSort) aggstate->phases[phaseidx - 1];
+
+ Assert(!prev->phasedata.is_hashed);
+ /* Tell the previous phase to copy the tuple to the sort_in */
+ prev->copy_out = true;
+ }
}
+
+ aggstate->phases[phaseidx] = phasedata;
}
/*
@@ -2545,51 +2598,12 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
aggstate->peragg = peraggs;
aggstate->pertrans = pertransstates;
-
- aggstate->all_pergroups =
- (AggStatePerGroup *) palloc0(sizeof(AggStatePerGroup)
- * (numGroupingSets + numHashes));
- pergroups = aggstate->all_pergroups;
-
- if (node->aggstrategy != AGG_HASHED)
- {
- for (i = 0; i < numGroupingSets; i++)
- {
- pergroups[i] = (AggStatePerGroup) palloc0(sizeof(AggStatePerGroupData)
- * numaggs);
- }
-
- aggstate->pergroups = pergroups;
- pergroups += numGroupingSets;
- }
-
/*
- * Hashing can only appear in the initial phase.
+ * Initialize current phase-dependent values to initial phase.
*/
- if (use_hashing)
- {
- /* this is an array of pointers, not structures */
- aggstate->hash_pergroup = pergroups;
- }
-
- /*
- * Initialize current phase-dependent values to initial phase. The initial
- * phase is 1 (first sort pass) for all strategies that use sorting (if
- * hashing is being done too, then phase 0 is processed last); but if only
- * hashing is being done, then phase 0 is all there is.
- */
- if (node->aggstrategy == AGG_HASHED)
- {
- aggstate->current_phase = 0;
- initialize_phase(aggstate, 0);
- select_current_set(aggstate, 0, true);
- }
- else
- {
- aggstate->current_phase = 1;
- initialize_phase(aggstate, 1);
- select_current_set(aggstate, 0, false);
- }
+ aggstate->current_phase = 0;
+ initialize_phase(aggstate, 0);
+ select_current_set(aggstate, 0, aggstate->aggstrategy == AGG_HASHED);
/* -----------------
* Perform lookups of aggregate function info, and initialize the
@@ -2942,49 +2956,12 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
*/
for (phaseidx = 0; phaseidx < aggstate->numphases; phaseidx++)
{
- AggStatePerPhase phase = &aggstate->phases[phaseidx];
- bool dohash = false;
- bool dosort = false;
+ AggStatePerPhase phase = aggstate->phases[phaseidx];
- /* phase 0 doesn't necessarily exist */
- if (!phase->aggnode)
- continue;
-
- if (aggstate->aggstrategy == AGG_MIXED && phaseidx == 1)
- {
- /*
- * Phase one, and only phase one, in a mixed agg performs both
- * sorting and aggregation.
- */
- dohash = true;
- dosort = true;
- }
- else if (aggstate->aggstrategy == AGG_MIXED && phaseidx == 0)
- {
- /*
- * No need to compute a transition function for an AGG_MIXED phase
- * 0 - the contents of the hashtables will have been computed
- * during phase 1.
- */
+ if (phase->skip_build_trans)
continue;
- }
- else if (phase->aggstrategy == AGG_PLAIN ||
- phase->aggstrategy == AGG_SORTED)
- {
- dohash = false;
- dosort = true;
- }
- else if (phase->aggstrategy == AGG_HASHED)
- {
- dohash = true;
- dosort = false;
- }
- else
- Assert(false);
-
- phase->evaltrans = ExecBuildAggTrans(aggstate, phase, dosort, dohash,
- false);
+ phase->evaltrans = ExecBuildAggTrans(aggstate, phase, false);
}
return aggstate;
@@ -3470,13 +3447,21 @@ ExecEndAgg(AggState *node)
int transno;
int numGroupingSets = Max(node->maxsets, 1);
int setno;
+ int phaseidx;
/* Make sure we have closed any open tuplesorts */
+ for (phaseidx = 0; phaseidx < node->numphases; phaseidx++)
+ {
+ AggStatePerPhase phase = node->phases[phaseidx++];
+ AggStatePerPhaseSort persort;
- if (node->sort_in)
- tuplesort_end(node->sort_in);
- if (node->sort_out)
- tuplesort_end(node->sort_out);
+ if (phase->is_hashed)
+ continue;
+
+ persort = (AggStatePerPhaseSort) phase;
+ if (persort->sort_in)
+ tuplesort_end(persort->sort_in);
+ }
for (transno = 0; transno < node->numtrans; transno++)
{
@@ -3518,6 +3503,7 @@ ExecReScanAgg(AggState *node)
int transno;
int numGroupingSets = Max(node->maxsets, 1);
int setno;
+ int phaseidx;
node->agg_done = false;
@@ -3541,8 +3527,12 @@ ExecReScanAgg(AggState *node)
if (outerPlan->chgParam == NULL &&
!bms_overlap(node->ss.ps.chgParam, aggnode->aggParams))
{
- ResetTupleHashIterator(node->perhash[0].hashtable,
- &node->perhash[0].hashiter);
+ AggStatePerPhaseHash perhash = (AggStatePerPhaseHash) node->phases[0];
+ ResetTupleHashIterator(perhash->hashtable,
+ &perhash->hashiter);
+
+ /* reset to phase 0 */
+ initialize_phase(node, 0);
select_current_set(node, 0, true);
return;
}
@@ -3607,18 +3597,54 @@ ExecReScanAgg(AggState *node)
/*
* Reset the per-group state (in particular, mark transvalues null)
*/
- for (setno = 0; setno < numGroupingSets; setno++)
+ for (phaseidx = 0; phaseidx < node->numphases; phaseidx++)
{
- MemSet(node->pergroups[setno], 0,
- sizeof(AggStatePerGroupData) * node->numaggs);
+ AggStatePerPhase phase = node->phases[phaseidx];
+
+ /* hash pergroups is reset by build_hash_tables */
+ if (phase->is_hashed)
+ continue;
+
+ for (setno = 0; setno < phase->numsets; setno++)
+ MemSet(phase->pergroups[setno], 0,
+ sizeof(AggStatePerGroupData) * node->numaggs);
}
- /* Reset input_sorted */
+ /*
+ * the agg did its own first sort using tuplesort and the first
+ * tuplesort is kept (see initialize_phase), if the subplan does
+ * not have any parameter changes, and none of our own parameter
+ * changes affect input expressions of the aggregated functions,
+ * then we can just rescan the first tuplesort, no need to build
+ * it again.
+ *
+ * Note: agg only do its own sort for groupingsets now.
+ */
if (aggnode->sortnode)
- node->input_sorted = false;
+ {
+ AggStatePerPhaseSort firstphase = (AggStatePerPhaseSort) node->phases[0];
+ bool randomAccess = (node->eflags & (EXEC_FLAG_REWIND |
+ EXEC_FLAG_BACKWARD |
+ EXEC_FLAG_MARK)) != 0;
+ if (firstphase->sort_in &&
+ randomAccess &&
+ outerPlan->chgParam == NULL &&
+ !bms_overlap(node->ss.ps.chgParam, aggnode->aggParams))
+ {
+ tuplesort_rescan(firstphase->sort_in);
+ node->input_sorted = true;
+ }
+ else
+ {
+ if (firstphase->sort_in)
+ tuplesort_end(firstphase->sort_in);
+ firstphase->sort_in = NULL;
+ node->input_sorted = false;
+ }
+ }
- /* reset to phase 1 */
- initialize_phase(node, 1);
+ /* reset to phase 0 */
+ initialize_phase(node, 0);
node->input_done = false;
node->projected_set = -1;
diff --git a/src/backend/jit/llvm/llvmjit_expr.c b/src/backend/jit/llvm/llvmjit_expr.c
index b855e73957..066cd59554 100644
--- a/src/backend/jit/llvm/llvmjit_expr.c
+++ b/src/backend/jit/llvm/llvmjit_expr.c
@@ -2049,30 +2049,26 @@ llvm_compile_expr(ExprState *state)
case EEOP_AGG_PLAIN_PERGROUP_NULLCHECK:
{
int jumpnull;
- LLVMValueRef v_aggstatep;
- LLVMValueRef v_allpergroupsp;
+ LLVMValueRef v_pergroupsp;
LLVMValueRef v_pergroup_allaggs;
- LLVMValueRef v_setoff;
+ LLVMValueRef v_setno;
jumpnull = op->d.agg_plain_pergroup_nullcheck.jumpnull;
/*
- * pergroup_allaggs = aggstate->all_pergroups
- * [op->d.agg_plain_pergroup_nullcheck.setoff];
+ * pergroup =
+ * &op->d.agg_plain_pergroup_nullcheck.pergroups
+ * [op->d.agg_plain_pergroup_nullcheck.setno];
*/
- v_aggstatep = LLVMBuildBitCast(
- b, v_parent, l_ptr(StructAggState), "");
+ v_pergroupsp =
+ l_ptr_const(op->d.agg_plain_pergroup_nullcheck.pergroups,
+ l_ptr(l_ptr(StructAggStatePerGroupData)));
- v_allpergroupsp = l_load_struct_gep(
- b, v_aggstatep,
- FIELDNO_AGGSTATE_ALL_PERGROUPS,
- "aggstate.all_pergroups");
+ v_setno =
+ l_int32_const(op->d.agg_plain_pergroup_nullcheck.setno);
- v_setoff = l_int32_const(
- op->d.agg_plain_pergroup_nullcheck.setoff);
-
- v_pergroup_allaggs = l_load_gep1(
- b, v_allpergroupsp, v_setoff, "");
+ v_pergroup_allaggs =
+ l_load_gep1(b, v_pergroupsp, v_setno, "");
LLVMBuildCondBr(
b,
@@ -2094,6 +2090,7 @@ llvm_compile_expr(ExprState *state)
{
AggState *aggstate;
AggStatePerTrans pertrans;
+ AggStatePerGroup *pergroups;
FunctionCallInfo fcinfo;
LLVMValueRef v_aggstatep;
@@ -2103,12 +2100,12 @@ llvm_compile_expr(ExprState *state)
LLVMValueRef v_transvaluep;
LLVMValueRef v_transnullp;
- LLVMValueRef v_setoff;
+ LLVMValueRef v_setno;
LLVMValueRef v_transno;
LLVMValueRef v_aggcontext;
- LLVMValueRef v_allpergroupsp;
+ LLVMValueRef v_pergroupsp;
LLVMValueRef v_current_setp;
LLVMValueRef v_current_pertransp;
LLVMValueRef v_curaggcontext;
@@ -2124,6 +2121,7 @@ llvm_compile_expr(ExprState *state)
aggstate = castNode(AggState, state->parent);
pertrans = op->d.agg_trans.pertrans;
+ pergroups = op->d.agg_trans.pergroups;
fcinfo = pertrans->transfn_fcinfo;
@@ -2133,19 +2131,18 @@ llvm_compile_expr(ExprState *state)
l_ptr(StructAggStatePerTransData));
/*
- * pergroup = &aggstate->all_pergroups
- * [op->d.agg_strict_trans_check.setoff]
- * [op->d.agg_init_trans_check.transno];
+ * pergroup = &op->d.agg_trans.pergroups
+ * [op->d.agg_trans.setno]
+ * [op->d.agg_trans.transno];
*/
- v_allpergroupsp =
- l_load_struct_gep(b, v_aggstatep,
- FIELDNO_AGGSTATE_ALL_PERGROUPS,
- "aggstate.all_pergroups");
- v_setoff = l_int32_const(op->d.agg_trans.setoff);
+ v_pergroupsp =
+ l_ptr_const(pergroups,
+ l_ptr(l_ptr(StructAggStatePerGroupData)));
+ v_setno = l_int32_const(op->d.agg_trans.setno);
v_transno = l_int32_const(op->d.agg_trans.transno);
v_pergroupp =
LLVMBuildGEP(b,
- l_load_gep1(b, v_allpergroupsp, v_setoff, ""),
+ l_load_gep1(b, v_pergroupsp, v_setno, ""),
&v_transno, 1, "");
diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c
index 044ec92aa8..29f88bf0b7 100644
--- a/src/backend/optimizer/plan/createplan.c
+++ b/src/backend/optimizer/plan/createplan.c
@@ -2226,8 +2226,6 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
chain = NIL;
if (list_length(rollups) > 1)
{
- bool is_first_sort = ((RollupData *) linitial(rollups))->is_hashed;
-
for_each_cell(lc, rollups, list_second_cell(rollups))
{
RollupData *rollup = lfirst(lc);
@@ -2244,24 +2242,17 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
*/
if (!rollup->is_hashed)
{
- if (!is_first_sort ||
- (is_first_sort && !best_path->is_sorted))
- {
- sort_plan = (Plan *)
- make_sort_from_groupcols(rollup->groupClause,
- new_grpColIdx,
- subplan);
-
- /*
- * Remove stuff we don't need to avoid bloating debug output.
- */
- sort_plan->targetlist = NIL;
- sort_plan->lefttree = NULL;
- }
- }
+ sort_plan = (Plan *)
+ make_sort_from_groupcols(rollup->groupClause,
+ new_grpColIdx,
+ subplan);
- if (!rollup->is_hashed)
- is_first_sort = false;
+ /*
+ * Remove stuff we don't need to avoid bloating debug output.
+ */
+ sort_plan->targetlist = NIL;
+ sort_plan->lefttree = NULL;
+ }
if (rollup->is_hashed)
strat = AGG_HASHED;
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
index fc6a1d0044..68d9c88a53 100644
--- a/src/backend/optimizer/plan/planner.c
+++ b/src/backend/optimizer/plan/planner.c
@@ -4344,7 +4344,8 @@ consider_groupingsets_paths(PlannerInfo *root,
if (unhashed_rollup)
{
- new_rollups = lappend(new_rollups, unhashed_rollup);
+ /* unhashed rollups always sit before hashed rollups */
+ new_rollups = lcons(unhashed_rollup, new_rollups);
strat = AGG_MIXED;
}
else if (empty_sets)
@@ -4357,7 +4358,8 @@ consider_groupingsets_paths(PlannerInfo *root,
rollup->numGroups = list_length(empty_sets);
rollup->hashable = false;
rollup->is_hashed = false;
- new_rollups = lappend(new_rollups, rollup);
+ /* unhashed rollups always sit before hashed rollups */
+ new_rollups = lcons(rollup, new_rollups);
/* update is_sorted to true
* XXX why? shouldn't it be already set by the caller?
*/
@@ -4525,7 +4527,8 @@ consider_groupingsets_paths(PlannerInfo *root,
rollup->numGroups = gs->numGroups;
rollup->hashable = true;
rollup->is_hashed = true;
- rollups = lcons(rollup, rollups);
+ /* non-hashed rollup always sit before hashed rollup */
+ rollups = lappend(rollups, rollup);
}
if (rollups)
diff --git a/src/backend/optimizer/util/pathnode.c b/src/backend/optimizer/util/pathnode.c
index 0feb3363d3..2dfa3fa17e 100644
--- a/src/backend/optimizer/util/pathnode.c
+++ b/src/backend/optimizer/util/pathnode.c
@@ -2983,7 +2983,7 @@ create_agg_path(PlannerInfo *root,
* 'rollups' is a list of RollupData nodes
* 'agg_costs' contains cost info about the aggregate functions to be computed
* 'numGroups' is the estimated total number of groups
- * 'is_sorted' is the input sorted in the group cols of first rollup
+ * 'is_sorted' is the input sorted in the group cols of first rollup
*/
GroupingSetsPath *
create_groupingsets_path(PlannerInfo *root,
@@ -3000,7 +3000,6 @@ create_groupingsets_path(PlannerInfo *root,
PathTarget *target = rel->reltarget;
ListCell *lc;
bool is_first = true;
- bool is_first_sort = true;
/* The topmost generated Plan node will be an Agg */
pathnode->path.pathtype = T_Agg;
@@ -3053,14 +3052,13 @@ create_groupingsets_path(PlannerInfo *root,
int numGroupCols = list_length(linitial(gsets));
/*
- * In AGG_SORTED or AGG_PLAIN mode, the first rollup takes the
- * (already-sorted) input, and following ones do their own sort.
+ * In AGG_SORTED or AGG_PLAIN mode, the first rollup do its own
+ * sort if is_sorted is false, the following ones do their own sort.
*
* In AGG_HASHED mode, there is one rollup for each grouping set.
*
- * In AGG_MIXED mode, the first rollups are hashed, the first
- * non-hashed one takes the (already-sorted) input, and following ones
- * do their own sort.
+ * In AGG_MIXED mode, the first rollup do its own sort if is_sorted
+ * is false, the following non-hashed ones do their own sort.
*/
if (is_first)
{
@@ -3092,33 +3090,23 @@ create_groupingsets_path(PlannerInfo *root,
input_startup_cost,
input_total_cost,
subpath->rows);
+
is_first = false;
- if (!rollup->is_hashed)
- is_first_sort = false;
}
else
{
- Path sort_path; /* dummy for result of cost_sort */
- Path agg_path; /* dummy for result of cost_agg */
-
- if (rollup->is_hashed || (is_first_sort && is_sorted))
- {
- /*
- * Account for cost of aggregation, but don't charge input
- * cost again
- */
- cost_agg(&agg_path, root,
- rollup->is_hashed ? AGG_HASHED : AGG_SORTED,
- agg_costs,
- numGroupCols,
- rollup->numGroups,
- having_qual,
- 0.0, 0.0,
- subpath->rows);
- if (!rollup->is_hashed)
- is_first_sort = false;
- }
- else
+ AggStrategy rollup_strategy;
+ Path sort_path; /* dummy for result of cost_sort */
+ Path agg_path; /* dummy for result of cost_agg */
+
+ sort_path.startup_cost = 0;
+ sort_path.total_cost = 0;
+ sort_path.rows = subpath->rows;
+
+ rollup_strategy = rollup->is_hashed ?
+ AGG_HASHED : (numGroupCols ? AGG_SORTED : AGG_PLAIN);
+
+ if (!rollup->is_hashed && numGroupCols)
{
/* Account for cost of sort, but don't charge input cost again */
cost_sort(&sort_path, root, NIL,
@@ -3128,20 +3116,19 @@ create_groupingsets_path(PlannerInfo *root,
0.0,
work_mem,
-1.0);
-
- /* Account for cost of aggregation */
-
- cost_agg(&agg_path, root,
- AGG_SORTED,
- agg_costs,
- numGroupCols,
- rollup->numGroups,
- having_qual,
- sort_path.startup_cost,
- sort_path.total_cost,
- sort_path.rows);
}
+ /* Account for cost of aggregation */
+ cost_agg(&agg_path, root,
+ rollup_strategy,
+ agg_costs,
+ numGroupCols,
+ rollup->numGroups,
+ having_qual,
+ sort_path.startup_cost,
+ sort_path.total_cost,
+ sort_path.rows);
+
pathnode->path.total_cost += agg_path.total_cost;
pathnode->path.rows += agg_path.rows;
}
diff --git a/src/include/executor/execExpr.h b/src/include/executor/execExpr.h
index dbe8649a57..4ed5d0a7de 100644
--- a/src/include/executor/execExpr.h
+++ b/src/include/executor/execExpr.h
@@ -626,7 +626,8 @@ typedef struct ExprEvalStep
/* for EEOP_AGG_PLAIN_PERGROUP_NULLCHECK */
struct
{
- int setoff;
+ AggStatePerGroup *pergroups;
+ int setno;
int jumpnull;
} agg_plain_pergroup_nullcheck;
@@ -634,11 +635,11 @@ typedef struct ExprEvalStep
/* for EEOP_AGG_ORDERED_TRANS_{DATUM,TUPLE} */
struct
{
+ AggStatePerGroup *pergroups;
AggStatePerTrans pertrans;
ExprContext *aggcontext;
int setno;
int transno;
- int setoff;
} agg_trans;
} d;
} ExprEvalStep;
diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h
index 94890512dc..1f37f9236b 100644
--- a/src/include/executor/executor.h
+++ b/src/include/executor/executor.h
@@ -255,7 +255,7 @@ extern ExprState *ExecInitQual(List *qual, PlanState *parent);
extern ExprState *ExecInitCheck(List *qual, PlanState *parent);
extern List *ExecInitExprList(List *nodes, PlanState *parent);
extern ExprState *ExecBuildAggTrans(AggState *aggstate, struct AggStatePerPhaseData *phase,
- bool doSort, bool doHash, bool nullcheck);
+ bool nullcheck);
extern ExprState *ExecBuildGroupingEqual(TupleDesc ldesc, TupleDesc rdesc,
const TupleTableSlotOps *lops, const TupleTableSlotOps *rops,
int numCols,
diff --git a/src/include/executor/nodeAgg.h b/src/include/executor/nodeAgg.h
index 66a83b9ac9..c5d4121c37 100644
--- a/src/include/executor/nodeAgg.h
+++ b/src/include/executor/nodeAgg.h
@@ -270,16 +270,29 @@ typedef struct AggStatePerGroupData
*/
typedef struct AggStatePerPhaseData
{
+ bool is_hashed; /* plan to do hash aggregate */
AggStrategy aggstrategy; /* strategy for this phase */
- int numsets; /* number of grouping sets (or 0) */
+ int numsets; /* number of grouping sets */
int *gset_lengths; /* lengths of grouping sets */
Bitmapset **grouped_cols; /* column groupings for rollup */
- ExprState **eqfunctions; /* expression returning equality, indexed by
- * nr of cols to compare */
Agg *aggnode; /* Agg node for phase data */
ExprState *evaltrans; /* evaluation of transition functions */
+
+ List *concurrent_hashes; /* hash phases can do transition concurrently */
+ AggStatePerGroup *pergroups; /* pergroup states for a phase */
+ bool skip_build_trans;
} AggStatePerPhaseData;
+typedef struct AggStatePerPhaseSortData
+{
+ AggStatePerPhaseData phasedata;
+ Tuplesortstate *sort_in; /* sorted input to phases > 1 */
+ Tuplestorestate *store_in; /* sorted input to phases > 1 */
+ ExprState **eqfunctions; /* expression returning equality, indexed by
+ * nr of cols to compare */
+ bool copy_out; /* hint for copy input tuples for next phase */
+} AggStatePerPhaseSortData;
+
/*
* AggStatePerHashData - per-hashtable state
*
@@ -287,8 +300,9 @@ typedef struct AggStatePerPhaseData
* grouping set. (When doing hashing without grouping sets, we have just one of
* them.)
*/
-typedef struct AggStatePerHashData
+typedef struct AggStatePerPhaseHashData
{
+ AggStatePerPhaseData phasedata;
TupleHashTable hashtable; /* hash table with one entry per group */
TupleHashIterator hashiter; /* for iterating through hash table */
TupleTableSlot *hashslot; /* slot for loading hash table */
@@ -299,9 +313,7 @@ typedef struct AggStatePerHashData
int largestGrpColIdx; /* largest col required for hashing */
AttrNumber *hashGrpColIdxInput; /* hash col indices in input slot */
AttrNumber *hashGrpColIdxHash; /* indices in hash table tuples */
- Agg *aggnode; /* original Agg node, for numGroups etc. */
-} AggStatePerHashData;
-
+} AggStatePerPhaseHashData;
extern AggState *ExecInitAgg(Agg *node, EState *estate, int eflags);
extern void ExecEndAgg(AggState *node);
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index 5e33a368f5..4081a0978e 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -2036,7 +2036,8 @@ typedef struct AggStatePerAggData *AggStatePerAgg;
typedef struct AggStatePerTransData *AggStatePerTrans;
typedef struct AggStatePerGroupData *AggStatePerGroup;
typedef struct AggStatePerPhaseData *AggStatePerPhase;
-typedef struct AggStatePerHashData *AggStatePerHash;
+typedef struct AggStatePerPhaseSortData *AggStatePerPhaseSort;
+typedef struct AggStatePerPhaseHashData *AggStatePerPhaseHash;
typedef struct AggState
{
@@ -2068,28 +2069,19 @@ typedef struct AggState
List *all_grouped_cols; /* list of all grouped cols in DESC order */
/* These fields are for grouping set phase data */
int maxsets; /* The max number of sets in any phase */
- AggStatePerPhase phases; /* array of all phases */
+ AggStatePerPhase *phases; /* array of all phases */
Tuplesortstate *sort_in; /* sorted input to phases > 1 */
Tuplesortstate *sort_out; /* input is copied here for next phase */
TupleTableSlot *sort_slot; /* slot for sort results */
/* these fields are used in AGG_PLAIN and AGG_SORTED modes: */
- AggStatePerGroup *pergroups; /* grouping set indexed array of per-group
- * pointers */
HeapTuple grp_firstTuple; /* copy of first tuple of current group */
- /* these fields are used in AGG_HASHED and AGG_MIXED modes: */
+ /* these fields are used in AGG_HASHED */
bool table_filled; /* hash table filled yet? */
- int num_hashes;
- AggStatePerHash perhash; /* array of per-hashtable data */
- AggStatePerGroup *hash_pergroup; /* grouping set indexed array of
- * per-group pointers */
/* these fields are used in AGG_SORTED and AGG_MIXED */
bool input_sorted; /* hash table filled yet? */
+ int eflags; /* eflags for the first sort */
- /* support for evaluation of agg input expressions: */
-#define FIELDNO_AGGSTATE_ALL_PERGROUPS 35
- AggStatePerGroup *all_pergroups; /* array of first ->pergroups, than
- * ->hash_pergroup */
ProjectionInfo *combinedproj; /* projection machinery */
} AggState;
diff --git a/src/test/regress/expected/groupingsets.out b/src/test/regress/expected/groupingsets.out
index 12425f46ca..e7689ebd16 100644
--- a/src/test/regress/expected/groupingsets.out
+++ b/src/test/regress/expected/groupingsets.out
@@ -1004,10 +1004,10 @@ explain (costs off) select a, b, grouping(a,b), sum(v), count(*), max(v)
Sort
Sort Key: (GROUPING("*VALUES*".column1, "*VALUES*".column2)), "*VALUES*".column1, "*VALUES*".column2
-> MixedAggregate
+ Group Key: ()
Hash Key: "*VALUES*".column1, "*VALUES*".column2
Hash Key: "*VALUES*".column1
Hash Key: "*VALUES*".column2
- Group Key: ()
-> Values Scan on "*VALUES*"
(8 rows)
@@ -1066,9 +1066,9 @@ explain (costs off)
Sort
Sort Key: (GROUPING(unhashable_col, unsortable_col)), (sum(v))
-> MixedAggregate
- Hash Key: unsortable_col
Sort Key: unhashable_col
Group Key: unhashable_col
+ Hash Key: unsortable_col
-> Seq Scan on gstest4
(7 rows)
@@ -1108,9 +1108,9 @@ explain (costs off)
Sort
Sort Key: (GROUPING(unhashable_col, unsortable_col)), (sum(v))
-> MixedAggregate
- Hash Key: v, unsortable_col
Sort Key: v, unhashable_col
Group Key: v, unhashable_col
+ Hash Key: v, unsortable_col
-> Seq Scan on gstest4
(7 rows)
@@ -1149,10 +1149,10 @@ explain (costs off)
QUERY PLAN
--------------------------------
MixedAggregate
- Hash Key: a, b
Group Key: ()
Group Key: ()
Group Key: ()
+ Hash Key: a, b
-> Seq Scan on gstest_empty
(6 rows)
@@ -1310,10 +1310,10 @@ explain (costs off)
-> Sort
Sort Key: a, b
-> MixedAggregate
+ Group Key: ()
Hash Key: a, b
Hash Key: a
Hash Key: b
- Group Key: ()
-> Seq Scan on gstest2
(11 rows)
@@ -1345,10 +1345,10 @@ explain (costs off)
Sort
Sort Key: gstest_data.a, gstest_data.b
-> MixedAggregate
+ Group Key: ()
Hash Key: gstest_data.a, gstest_data.b
Hash Key: gstest_data.a
Hash Key: gstest_data.b
- Group Key: ()
-> Nested Loop
-> Values Scan on "*VALUES*"
-> Function Scan on gstest_data
@@ -1545,16 +1545,16 @@ explain (costs off)
QUERY PLAN
----------------------------
MixedAggregate
- Hash Key: two
- Hash Key: four
- Hash Key: ten
- Hash Key: hundred
Sort Key: unique1
Group Key: unique1
Sort Key: twothousand
Group Key: twothousand
Sort Key: thousand
Group Key: thousand
+ Hash Key: hundred
+ Hash Key: ten
+ Hash Key: four
+ Hash Key: two
-> Seq Scan on tenk1
(12 rows)
@@ -1567,12 +1567,12 @@ explain (costs off)
QUERY PLAN
-------------------------
MixedAggregate
- Hash Key: two
- Hash Key: four
- Hash Key: ten
- Hash Key: hundred
Sort Key: unique1
Group Key: unique1
+ Hash Key: hundred
+ Hash Key: ten
+ Hash Key: four
+ Hash Key: two
-> Seq Scan on tenk1
(8 rows)
@@ -1586,15 +1586,15 @@ explain (costs off)
QUERY PLAN
----------------------------
MixedAggregate
- Hash Key: two
- Hash Key: four
- Hash Key: ten
- Hash Key: hundred
- Hash Key: thousand
Sort Key: unique1
Group Key: unique1
Sort Key: twothousand
Group Key: twothousand
+ Hash Key: thousand
+ Hash Key: hundred
+ Hash Key: ten
+ Hash Key: four
+ Hash Key: two
-> Seq Scan on tenk1
(11 rows)
diff --git a/src/test/regress/expected/partition_aggregate.out b/src/test/regress/expected/partition_aggregate.out
index fbc8d3ac6c..7818f02032 100644
--- a/src/test/regress/expected/partition_aggregate.out
+++ b/src/test/regress/expected/partition_aggregate.out
@@ -340,8 +340,8 @@ SELECT c, sum(a) FROM pagg_tab GROUP BY rollup(c) ORDER BY 1, 2;
Sort
Sort Key: pagg_tab.c, (sum(pagg_tab.a))
-> MixedAggregate
- Hash Key: pagg_tab.c
Group Key: ()
+ Hash Key: pagg_tab.c
-> Append
-> Seq Scan on pagg_tab_p1 pagg_tab_1
-> Seq Scan on pagg_tab_p2 pagg_tab_2
--
2.21.1
0005-fix.patch (text/plain, 6.9K) download
From df0f50a9bdf32def4a8768ad03e75e6a6b42b249 Mon Sep 17 00:00:00 2001
From: Tomas Vondra <tv@fuzzy.cz>
Date: Thu, 19 Mar 2020 02:54:39 +0100
Subject: [PATCH 5/7] fix
---
.../postgres_fdw/expected/postgres_fdw.out | 4 ++--
src/backend/executor/execExpr.c | 5 +++--
src/backend/executor/nodeAgg.c | 20 +++++++++----------
src/backend/optimizer/util/pathnode.c | 4 ++--
4 files changed, 17 insertions(+), 16 deletions(-)
diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out
index 62c2697920..fc0ed2f4d5 100644
--- a/contrib/postgres_fdw/expected/postgres_fdw.out
+++ b/contrib/postgres_fdw/expected/postgres_fdw.out
@@ -3448,8 +3448,8 @@ select c2, sum(c1) from ft1 where c2 < 3 group by rollup(c2) order by 1 nulls la
Sort Key: ft1.c2
-> MixedAggregate
Output: c2, sum(c1)
- Hash Key: ft1.c2
Group Key: ()
+ Hash Key: ft1.c2
-> Foreign Scan on public.ft1
Output: c2, c1
Remote SQL: SELECT "C 1", c2 FROM "S 1"."T 1" WHERE ((c2 < 3))
@@ -3473,8 +3473,8 @@ select c2, sum(c1) from ft1 where c2 < 3 group by cube(c2) order by 1 nulls last
Sort Key: ft1.c2
-> MixedAggregate
Output: c2, sum(c1)
- Hash Key: ft1.c2
Group Key: ()
+ Hash Key: ft1.c2
-> Foreign Scan on public.ft1
Output: c2, c1
Remote SQL: SELECT "C 1", c2 FROM "S 1"."T 1" WHERE ((c2 < 3))
diff --git a/src/backend/executor/execExpr.c b/src/backend/executor/execExpr.c
index 07789501f7..669843faf5 100644
--- a/src/backend/executor/execExpr.c
+++ b/src/backend/executor/execExpr.c
@@ -2937,7 +2937,7 @@ ExecBuildAggTrans(AggState *aggstate, AggStatePerPhase phase, bool nullcheck)
PlanState *parent = &aggstate->ss.ps;
ExprEvalStep scratch = {0};
bool isCombine = DO_AGGSPLIT_COMBINE(aggstate->aggsplit);
- ListCell *lc;
+ ListCell *lc;
LastAttnumInfo deform = {0, 0, 0};
state->expr = (Expr *) aggstate;
@@ -2978,6 +2978,7 @@ ExecBuildAggTrans(AggState *aggstate, AggStatePerPhase phase, bool nullcheck)
NullableDatum *strictargs = NULL;
bool *strictnulls = NULL;
int argno;
+ int setno;
ListCell *bail;
/*
@@ -3155,7 +3156,7 @@ ExecBuildAggTrans(AggState *aggstate, AggStatePerPhase phase, bool nullcheck)
* grouping set). Do so for both sort and hash based computations, as
* applicable.
*/
- for (int setno = 0; setno < phase->numsets; setno++)
+ for (setno = 0; setno < phase->numsets; setno++)
{
ExecBuildAggTransCall(state, aggstate, &scratch, trans_fcinfo,
pertrans, transno, setno, phase, nullcheck);
diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c
index 20c5eb98b3..38d0bd5895 100644
--- a/src/backend/executor/nodeAgg.c
+++ b/src/backend/executor/nodeAgg.c
@@ -333,7 +333,7 @@ initialize_phase(AggState *aggstate, int newphase)
AggStatePerPhaseSort persort;
Assert(newphase == 0 || newphase == aggstate->current_phase + 1);
-
+
/* Don't use aggstate->phase here, it might not be initialized yet*/
current_phase = aggstate->phases[aggstate->current_phase];
@@ -1516,7 +1516,7 @@ hash_agg_entry_size(int numAggs, Size tupleWidth, Size transitionSpace)
* When called, CurrentMemoryContext should be the per-query context. The
* already-calculated hash value for the tuple must be specified.
*/
-static void
+static void
lookup_hash_entry(AggState *aggstate, AggStatePerPhaseHash perhash, uint32 hash)
{
TupleTableSlot *hashslot = perhash->hashslot;
@@ -1724,7 +1724,7 @@ agg_retrieve_direct(AggState *aggstate)
numGroupingSets = aggstate->phase->numsets;
node = aggstate->phase->aggnode;
numReset = numGroupingSets;
- pergroups = aggstate->phase->pergroups;
+ pergroups = aggstate->phase->pergroups;
}
else
{
@@ -2123,7 +2123,7 @@ agg_retrieve_hash_table(AggState *aggstate)
*/
select_current_set(aggstate, 0, true);
initialize_phase(aggstate, aggstate->current_phase + 1);
- perhash = (AggStatePerPhaseHash) aggstate->phase;
+ perhash = (AggStatePerPhaseHash) aggstate->phase;
ResetTupleHashIterator(perhash->hashtable, &perhash->hashiter);
continue;
@@ -2269,7 +2269,7 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
aggstate->maxsets = numGroupingSets;
aggstate->numphases = 1 + list_length(node->chain);
-
+
/*
* The first phase is not sorted, agg need to do its own sort. See
* agg_sort_input(), this can only happen in groupingsets case.
@@ -2390,7 +2390,7 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
numaggs = aggstate->numaggs;
Assert(numaggs == list_length(aggstate->aggs));
- /*
+ /*
* For each phase, prepare grouping set data and fmgr lookup data for
* compare functions. Accumulate all_grouped_cols in passing.
*/
@@ -2430,7 +2430,7 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
all_grouped_cols = bms_add_members(all_grouped_cols, cols);
- /*
+ /*
* Initialize pergroup state. For AGG_HASHED, all groups do transition
* on the fly, all pergroup states are kept in hashtable, everytime
* a tuple is processed, lookup_hash_entry() choose one group and
@@ -2499,7 +2499,7 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
phasedata->grouped_cols = NULL;
}
- /*
+ /*
* Initialize pergroup states for AGG_SORTED/AGG_PLAIN/AGG_MIXED
* phases, each set only have one group on the fly, all groups in
* a set can reuse a pergroup state. Unlike AGG_HASHED, we
@@ -3610,8 +3610,8 @@ ExecReScanAgg(AggState *node)
sizeof(AggStatePerGroupData) * node->numaggs);
}
- /*
- * the agg did its own first sort using tuplesort and the first
+ /*
+ * The agg did its own first sort using tuplesort and the first
* tuplesort is kept (see initialize_phase), if the subplan does
* not have any parameter changes, and none of our own parameter
* changes affect input expressions of the aggregated functions,
diff --git a/src/backend/optimizer/util/pathnode.c b/src/backend/optimizer/util/pathnode.c
index 2dfa3fa17e..ff8f676dfb 100644
--- a/src/backend/optimizer/util/pathnode.c
+++ b/src/backend/optimizer/util/pathnode.c
@@ -2983,7 +2983,7 @@ create_agg_path(PlannerInfo *root,
* 'rollups' is a list of RollupData nodes
* 'agg_costs' contains cost info about the aggregate functions to be computed
* 'numGroups' is the estimated total number of groups
- * 'is_sorted' is the input sorted in the group cols of first rollup
+ * 'is_sorted' is the input sorted in the group cols of first rollup
*/
GroupingSetsPath *
create_groupingsets_path(PlannerInfo *root,
@@ -3098,7 +3098,7 @@ create_groupingsets_path(PlannerInfo *root,
AggStrategy rollup_strategy;
Path sort_path; /* dummy for result of cost_sort */
Path agg_path; /* dummy for result of cost_agg */
-
+
sort_path.startup_cost = 0;
sort_path.total_cost = 0;
sort_path.rows = subpath->rows;
--
2.21.1
0006-Parallel-grouping-sets.patch (text/plain, 58.2K) download
From 243965f12c16c8ded348255254a179ec198f6812 Mon Sep 17 00:00:00 2001
From: Tomas Vondra <tv@fuzzy.cz>
Date: Thu, 19 Mar 2020 03:00:50 +0100
Subject: [PATCH 6/7] Parallel grouping sets
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
We used to support grouping sets in one worker only, this PR
want to support parallel grouping sets using multiple workers.
the main idea of parallel grouping sets is: like parallel aggregate, we separate
grouping sets into two stages:
The initial stage: this stage has almost the same plan and execution routines
with the current implementation of grouping sets, the differenceis are 1) it
only produces partial aggregate results 2) the output is attached with an extra
grouping set id. We know partial aggregate results will be combined in the final
stage and we have multiple grouping sets, so only partial aggregate results
belong to the same grouping set can be combined, that is why grouping set id is
introduced to identify the sets. We keep all the optimizations of multiple
grouping sets in the initial stage, eg, 1) the grouping sets (that can be
grouped by one single sort) are put into the one rollup structure so those sets
arecomputed in one aggregate phase. 2) do hash aggregate concurrently when a
sort aggregate is performed. 3) do all hash transitions in one expression state.
The final stage: this stage combine the partial aggregate results according to
the grouping set id. Obviously, all the optimizations in the initial stage
cannot be used, so all rollups are extracted, each rollup contains only one
grouping set, then each aggregate phase only processes one set. We do a filter
in the final stage, redirect the tuples to each aggregate phase.
---
src/backend/commands/explain.c | 10 +-
src/backend/executor/execExpr.c | 10 +-
src/backend/executor/execExprInterp.c | 11 +
src/backend/executor/nodeAgg.c | 261 +++++++++++++++++-
src/backend/jit/llvm/llvmjit_expr.c | 40 +++
src/backend/nodes/copyfuncs.c | 56 +++-
src/backend/nodes/equalfuncs.c | 3 +
src/backend/nodes/nodeFuncs.c | 8 +
src/backend/nodes/outfuncs.c | 14 +-
src/backend/nodes/readfuncs.c | 53 +++-
src/backend/optimizer/path/allpaths.c | 5 +-
src/backend/optimizer/plan/createplan.c | 26 +-
src/backend/optimizer/plan/planner.c | 343 ++++++++++++++++++------
src/backend/optimizer/plan/setrefs.c | 16 ++
src/backend/optimizer/util/pathnode.c | 27 +-
src/backend/utils/adt/ruleutils.c | 6 +
src/include/executor/execExpr.h | 1 +
src/include/executor/nodeAgg.h | 2 +
src/include/nodes/execnodes.h | 8 +-
src/include/nodes/nodes.h | 1 +
src/include/nodes/pathnodes.h | 2 +
src/include/nodes/plannodes.h | 4 +-
src/include/nodes/primnodes.h | 6 +
src/include/optimizer/pathnode.h | 1 +
src/include/optimizer/planmain.h | 2 +-
25 files changed, 791 insertions(+), 125 deletions(-)
diff --git a/src/backend/commands/explain.c b/src/backend/commands/explain.c
index 2c63cdb46c..8b6877c41e 100644
--- a/src/backend/commands/explain.c
+++ b/src/backend/commands/explain.c
@@ -2256,12 +2256,16 @@ show_agg_keys(AggState *astate, List *ancestors,
{
Agg *plan = (Agg *) astate->ss.ps.plan;
- if (plan->numCols > 0 || plan->groupingSets)
+ if (plan->gsetid)
+ show_expression((Node *) plan->gsetid, "Filtered by",
+ (PlanState *) astate, ancestors, true, es);
+
+ if (plan->numCols > 0 || plan->rollup)
{
/* The key columns refer to the tlist of the child plan */
ancestors = lcons(plan, ancestors);
- if (plan->groupingSets)
+ if (plan->rollup)
show_grouping_sets(outerPlanState(astate), plan, ancestors, es);
else
show_sort_group_keys(outerPlanState(astate), "Group Key",
@@ -2312,7 +2316,7 @@ show_grouping_set_keys(PlanState *planstate,
Plan *plan = planstate->plan;
char *exprstr;
ListCell *lc;
- List *gsets = aggnode->groupingSets;
+ List *gsets = aggnode->rollup->gsets;
AttrNumber *keycols = aggnode->grpColIdx;
const char *keyname;
const char *keysetname;
diff --git a/src/backend/executor/execExpr.c b/src/backend/executor/execExpr.c
index 669843faf5..bf69fcfe97 100644
--- a/src/backend/executor/execExpr.c
+++ b/src/backend/executor/execExpr.c
@@ -815,7 +815,7 @@ ExecInitExprRec(Expr *node, ExprState *state,
agg = (Agg *) (state->parent->plan);
- if (agg->groupingSets)
+ if (agg->rollup)
scratch.d.grouping_func.clauses = grp_node->cols;
else
scratch.d.grouping_func.clauses = NIL;
@@ -824,6 +824,14 @@ ExecInitExprRec(Expr *node, ExprState *state,
break;
}
+ case T_GroupingSetId:
+ {
+ scratch.opcode = EEOP_GROUPING_SET_ID;
+
+ ExprEvalPushStep(state, &scratch);
+ break;
+ }
+
case T_WindowFunc:
{
WindowFunc *wfunc = (WindowFunc *) node;
diff --git a/src/backend/executor/execExprInterp.c b/src/backend/executor/execExprInterp.c
index b0dbba4e55..b3537eb8d9 100644
--- a/src/backend/executor/execExprInterp.c
+++ b/src/backend/executor/execExprInterp.c
@@ -428,6 +428,7 @@ ExecInterpExpr(ExprState *state, ExprContext *econtext, bool *isnull)
&&CASE_EEOP_XMLEXPR,
&&CASE_EEOP_AGGREF,
&&CASE_EEOP_GROUPING_FUNC,
+ &&CASE_EEOP_GROUPING_SET_ID,
&&CASE_EEOP_WINDOW_FUNC,
&&CASE_EEOP_SUBPLAN,
&&CASE_EEOP_ALTERNATIVE_SUBPLAN,
@@ -1512,6 +1513,16 @@ ExecInterpExpr(ExprState *state, ExprContext *econtext, bool *isnull)
EEO_NEXT();
}
+ EEO_CASE(EEOP_GROUPING_SET_ID)
+ {
+ AggState *aggstate = castNode(AggState, state->parent);
+
+ *op->resvalue = aggstate->phase->setno_gsetids[aggstate->current_set];
+ *op->resnull = false;
+
+ EEO_NEXT();
+ }
+
EEO_CASE(EEOP_WINDOW_FUNC)
{
/*
diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c
index 38d0bd5895..f7b98dd798 100644
--- a/src/backend/executor/nodeAgg.c
+++ b/src/backend/executor/nodeAgg.c
@@ -282,6 +282,7 @@ static void lookup_hash_entry(AggState *aggstate, AggStatePerPhaseHash perhash,
static TupleTableSlot *agg_retrieve_direct(AggState *aggstate);
static void agg_fill_hash_table(AggState *aggstate);
static void agg_sort_input(AggState *aggstate);
+static void agg_preprocess_groupingsets(AggState *aggstate);
static TupleTableSlot *agg_retrieve_hash_table(AggState *aggstate);
static Datum GetAggInitVal(Datum textInitVal, Oid transtype);
static void build_pertrans_for_aggref(AggStatePerTrans pertrans,
@@ -341,17 +342,26 @@ initialize_phase(AggState *aggstate, int newphase)
* Whatever the previous state, we're now done with whatever input
* tuplesort was in use, cleanup them.
*
- * Note: we keep the first tuplesort/tuplestore, this will benifit the
+ * Note: we keep the first tuplesort/tuplestore when it's not the
+ * final stage of partial groupingsets, this will benifit the
* rescan in some cases without resorting the input again.
*/
- if (!current_phase->is_hashed && aggstate->current_phase > 0)
+ if (!current_phase->is_hashed &&
+ (aggstate->current_phase > 0 || DO_AGGSPLIT_COMBINE(aggstate->aggsplit)))
{
persort = (AggStatePerPhaseSort) current_phase;
+
if (persort->sort_in)
{
tuplesort_end(persort->sort_in);
persort->sort_in = NULL;
}
+
+ if (persort->store_in)
+ {
+ tuplestore_end(persort->store_in);
+ persort->store_in = NULL;
+ }
}
/* advance to next phase */
@@ -420,6 +430,15 @@ fetch_input_tuple(AggState *aggstate)
return NULL;
slot = aggstate->sort_slot;
}
+ else if (current_phase->store_in)
+ {
+ /* make sure we check for interrupts in either path through here */
+ CHECK_FOR_INTERRUPTS();
+ if (!tuplestore_gettupleslot(current_phase->store_in, true, false,
+ aggstate->sort_slot))
+ return NULL;
+ slot = aggstate->sort_slot;
+ }
else
slot = ExecProcNode(outerPlanState(aggstate));
@@ -1597,6 +1616,9 @@ ExecAgg(PlanState *pstate)
CHECK_FOR_INTERRUPTS();
+ if (node->groupingsets_preprocess)
+ agg_preprocess_groupingsets(node);
+
if (!node->agg_done)
{
/* Dispatch based on strategy */
@@ -1637,7 +1659,7 @@ agg_retrieve_direct(AggState *aggstate)
TupleTableSlot *outerslot;
TupleTableSlot *firstSlot;
TupleTableSlot *result;
- bool hasGroupingSets = aggstate->phase->aggnode->groupingSets != NULL;
+ bool hasGroupingSets = aggstate->phase->aggnode->rollup != NULL;
int numGroupingSets = aggstate->phase->numsets;
int currentSet;
int nextSetSize;
@@ -1970,6 +1992,135 @@ agg_retrieve_direct(AggState *aggstate)
return NULL;
}
+/*
+ * Routine for final phase of partial grouping sets:
+ *
+ * Preprocess tuples for final phase of grouping sets. In initial phase,
+ * tuples is decorated with a grouping set ID and in the final phase, all
+ * grouping set are fit into different aggregate phases, so we need to
+ * redirect the tuples to each aggregate phases according to the grouping
+ * set ID.
+ */
+static void
+agg_preprocess_groupingsets(AggState *aggstate)
+{
+ AggStatePerPhaseSort persort;
+ AggStatePerPhaseHash perhash;
+ AggStatePerPhase phase;
+ TupleTableSlot *outerslot;
+ ExprContext *tmpcontext = aggstate->tmpcontext;
+ int phaseidx;
+
+ Assert(DO_AGGSPLIT_COMBINE(aggstate->aggsplit));
+ Assert(aggstate->groupingsets_preprocess);
+
+ /* Initialize tuples storage for each aggregate phases */
+ for (phaseidx = 0; phaseidx < aggstate->numphases; phaseidx++)
+ {
+ phase = aggstate->phases[phaseidx];
+
+ if (!phase->is_hashed)
+ {
+ persort = (AggStatePerPhaseSort) phase;
+ if (phase->aggnode->sortnode)
+ {
+ Sort *sortnode = (Sort *) phase->aggnode->sortnode;
+ PlanState *outerNode = outerPlanState(aggstate);
+ TupleDesc tupDesc = ExecGetResultType(outerNode);
+
+ persort->sort_in = tuplesort_begin_heap(tupDesc,
+ sortnode->numCols,
+ sortnode->sortColIdx,
+ sortnode->sortOperators,
+ sortnode->collations,
+ sortnode->nullsFirst,
+ work_mem,
+ NULL, false);
+ }
+ else
+ {
+ persort->store_in = tuplestore_begin_heap(false, false, work_mem);
+ }
+ }
+ else
+ {
+ /*
+ * If it's a AGG_HASHED, we don't need a storage to store
+ * the tuples for later process, we can do the transition
+ * immediately.
+ */
+ }
+ }
+
+ for (;;)
+ {
+ Datum ret;
+ bool isNull;
+ int setid;
+
+ outerslot = ExecProcNode(outerPlanState(aggstate));
+ if (TupIsNull(outerslot))
+ break;
+
+ tmpcontext->ecxt_outertuple = outerslot;
+
+ /* Finger out which group set the tuple belongs to ?*/
+ ret = ExecEvalExprSwitchContext(aggstate->gsetid, tmpcontext, &isNull);
+
+ setid = DatumGetInt32(ret);
+ phase = aggstate->phases[aggstate->gsetid_phaseidxs[setid]];
+
+ if (!phase->is_hashed)
+ {
+ persort = (AggStatePerPhaseSort) phase;
+
+ Assert(persort->sort_in || persort->store_in);
+
+ if (persort->sort_in)
+ tuplesort_puttupleslot(persort->sort_in, outerslot);
+ else if (persort->store_in)
+ tuplestore_puttupleslot(persort->store_in, outerslot);
+ }
+ else
+ {
+ int hash;
+ bool dummynull;
+
+ perhash = (AggStatePerPhaseHash) phase;
+
+ /* If it is hashed, we can do the transition now. */
+ select_current_set(aggstate, 0, true);
+ prepare_hash_slot(aggstate, perhash);
+ hash = TupleHashTableHash(perhash->hashtable, perhash->hashslot);
+ lookup_hash_entry(aggstate, perhash, hash);
+
+ ExecEvalExprSwitchContext(phase->evaltrans,
+ tmpcontext,
+ &dummynull);
+ }
+
+ ResetExprContext(aggstate->tmpcontext);
+ }
+
+ /* Sort the first phase if needed */
+ if (aggstate->aggstrategy != AGG_HASHED)
+ {
+ persort = (AggStatePerPhaseSort) aggstate->phase;
+
+ if (persort->sort_in)
+ tuplesort_performsort(persort->sort_in);
+ }
+
+ /* Mark the hash table to be filled */
+ aggstate->table_filled = true;
+
+ /* Mark the input table to be sorted */
+ aggstate->input_sorted = true;
+
+ /* Mark the flag to not preprocessing groupingsets again */
+ aggstate->groupingsets_preprocess = false;
+}
+
static void
agg_sort_input(AggState *aggstate)
{
@@ -2246,21 +2397,22 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
aggstate->grp_firstTuple = NULL;
aggstate->input_sorted = true;
aggstate->eflags = eflags;
+ aggstate->groupingsets_preprocess = false;
/*
* Calculate the maximum number of grouping sets in any phase; this
* determines the size of some allocations.
*/
- if (node->groupingSets)
+ if (node->rollup)
{
- numGroupingSets = list_length(node->groupingSets);
+ numGroupingSets = list_length(node->rollup->gsets);
foreach(l, node->chain)
{
Agg *agg = lfirst(l);
numGroupingSets = Max(numGroupingSets,
- list_length(agg->groupingSets));
+ list_length(agg->rollup->gsets));
if (agg->aggstrategy != AGG_HASHED)
need_extra_slot = true;
@@ -2270,6 +2422,28 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
aggstate->maxsets = numGroupingSets;
aggstate->numphases = 1 + list_length(node->chain);
+ /*
+ * We are doing final stage of partial groupingsets, do preprocess
+ * to input tuples first, redirect the tuples to according aggregate
+ * phases. See agg_preprocess_groupingsets().
+ */
+ if (node->rollup && DO_AGGSPLIT_COMBINE(aggstate->aggsplit))
+ {
+ aggstate->groupingsets_preprocess = true;
+
+ /*
+ * Allocate gsetid <-> phases mapping, in final stage of
+ * partial groupingsets, all grouping sets are extracted
+ * to individual phases, so the number of sets is equal
+ * to the number of phases
+ */
+ aggstate->gsetid_phaseidxs =
+ (int *) palloc0(aggstate->numphases * sizeof(int));
+
+ if (aggstate->aggstrategy != AGG_HASHED)
+ need_extra_slot = true;
+ }
+
/*
* The first phase is not sorted, agg need to do its own sort. See
* agg_sort_input(), this can only happen in groupingsets case.
@@ -2384,6 +2558,12 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
aggstate->ss.ps.qual =
ExecInitQual(node->plan.qual, (PlanState *) aggstate);
+ /*
+ * Initialize expression state to fetch grouping set id from
+ * the partial groupingsets aggregate result.
+ */
+ aggstate->gsetid =
+ ExecInitExpr(node->gsetid, (PlanState *)aggstate);
/*
* We should now have found all Aggrefs in the targetlist and quals.
*/
@@ -2431,6 +2611,22 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
all_grouped_cols = bms_add_members(all_grouped_cols, cols);
/*
+ * In the initial stage of partial grouping sets, it provides extra
+ * grouping sets ID in the targetlist, fill the setno <-> gsetid
+ * map, so EEOP_GROUPING_SET_ID can evaluate correct gsetid for
+ * the output.
+ */
+ if (aggnode->rollup &&
+ DO_AGGSPLIT_SERIALIZE(aggnode->aggsplit))
+ {
+ GroupingSetData *gs;
+ phasedata->setno_gsetids = palloc(sizeof(int));
+ gs = linitial_node(GroupingSetData,
+ aggnode->rollup->gsets_data);
+ phasedata->setno_gsetids[0] = gs->setId;
+ }
+
+ /*
* Initialize pergroup state. For AGG_HASHED, all groups do transition
* on the fly, all pergroup states are kept in hashtable, everytime
* a tuple is processed, lookup_hash_entry() choose one group and
@@ -2448,8 +2644,12 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
* we can do the transition immediately when a tuple is fetched,
* which means we can do the transition concurrently with the
* first phase.
+ *
+ * Note: this is not work for final phase of partial groupingsets in
+ * which the partial input tuple has a specified target aggregate
+ * phase.
*/
- if (phaseidx > 0)
+ if (phaseidx > 0 && !aggstate->groupingsets_preprocess)
{
aggstate->phases[0]->concurrent_hashes =
lappend(aggstate->phases[0]->concurrent_hashes, perhash);
@@ -2467,17 +2667,19 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
phasedata->aggnode = aggnode;
phasedata->aggstrategy = aggnode->aggstrategy;
- if (aggnode->groupingSets)
+ if (aggnode->rollup)
{
- phasedata->numsets = list_length(aggnode->groupingSets);
+ phasedata->numsets = list_length(aggnode->rollup->gsets_data);
phasedata->gset_lengths = palloc(phasedata->numsets * sizeof(int));
phasedata->grouped_cols = palloc(phasedata->numsets * sizeof(Bitmapset *));
+ phasedata->setno_gsetids = palloc(phasedata->numsets * sizeof(int));
i = 0;
- foreach(l, aggnode->groupingSets)
+ foreach(l, aggnode->rollup->gsets_data)
{
- int current_length = list_length(lfirst(l));
- Bitmapset *cols = NULL;
+ GroupingSetData *gs = lfirst_node(GroupingSetData, l);
+ int current_length = list_length(gs->set);
+ Bitmapset *cols = NULL;
/* planner forces this to be correct */
for (j = 0; j < current_length; ++j)
@@ -2486,6 +2688,15 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
phasedata->grouped_cols[i] = cols;
phasedata->gset_lengths[i] = current_length;
+ /*
+ * In the initial stage of partial grouping sets, it provides extra
+ * grouping sets ID in the targetlist, fill the setno <-> gsetid
+ * map, so EEOP_GROUPING_SET_ID can evaluate correct gsetid for
+ * the output.
+ */
+ if (DO_AGGSPLIT_SERIALIZE(aggstate->aggsplit))
+ phasedata->setno_gsetids[i] = gs->setId;
+
++i;
}
@@ -2562,8 +2773,11 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
* For non-first AGG_SORTED phase, it processes the same input
* tuples with previous phase except that it need to resort the
* input tuples. Tell the previous phase to copy out the tuples.
+ *
+ * Note: it doesn't work for final stage of partial grouping sets
+ * in which tuple has specified target aggregate phase.
*/
- if (phaseidx > 0)
+ if (phaseidx > 0 && !aggstate->groupingsets_preprocess)
{
AggStatePerPhaseSort prev =
(AggStatePerPhaseSort) aggstate->phases[phaseidx - 1];
@@ -2574,6 +2788,18 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
}
}
+ /*
+ * Fill the gsetid_phaseidxs array, so we can find according phases
+ * using gsetid.
+ */
+ if (aggstate->groupingsets_preprocess)
+ {
+ GroupingSetData *gs =
+ linitial_node(GroupingSetData, aggnode->rollup->gsets_data);
+
+ aggstate->gsetid_phaseidxs[gs->setId] = phaseidx;
+ }
+
aggstate->phases[phaseidx] = phasedata;
}
@@ -3461,6 +3687,8 @@ ExecEndAgg(AggState *node)
persort = (AggStatePerPhaseSort) phase;
if (persort->sort_in)
tuplesort_end(persort->sort_in);
+ if (persort->store_in)
+ tuplestore_end(persort->store_in);
}
for (transno = 0; transno < node->numtrans; transno++)
@@ -3643,6 +3871,13 @@ ExecReScanAgg(AggState *node)
}
}
+ /*
+ * if the agg is doing final stage of partial groupingsets, reset the
+ * flag to do groupingsets preprocess again.
+ */
+ if (aggnode->rollup && DO_AGGSPLIT_COMBINE(node->aggsplit))
+ node->groupingsets_preprocess = true;
+
/* reset to phase 0 */
initialize_phase(node, 0);
diff --git a/src/backend/jit/llvm/llvmjit_expr.c b/src/backend/jit/llvm/llvmjit_expr.c
index 066cd59554..f442442269 100644
--- a/src/backend/jit/llvm/llvmjit_expr.c
+++ b/src/backend/jit/llvm/llvmjit_expr.c
@@ -1882,6 +1882,46 @@ llvm_compile_expr(ExprState *state)
LLVMBuildBr(b, opblocks[opno + 1]);
break;
+ case EEOP_GROUPING_SET_ID:
+ {
+ LLVMValueRef v_resvalue;
+ LLVMValueRef v_aggstatep;
+ LLVMValueRef v_phase;
+ LLVMValueRef v_current_set;
+ LLVMValueRef v_setno_gsetids;
+
+ v_aggstatep =
+ LLVMBuildBitCast(b, v_parent, l_ptr(StructAggState), "");
+
+ /*
+ * op->resvalue =
+ * aggstate->phase->setno_gsetids
+ * [aggstate->current_set]
+ */
+ v_phase =
+ l_load_struct_gep(b, v_aggstatep,
+ FIELDNO_AGGSTATE_PHASE,
+ "aggstate.phase");
+ v_setno_gsetids =
+ l_load_struct_gep(b, v_phase,
+ FIELDNO_AGGSTATEPERPHASE_SETNOGSETIDS,
+ "aggstateperphase.setno_gsetids");
+ v_current_set =
+ l_load_struct_gep(b, v_aggstatep,
+ FIELDNO_AGGSTATE_CURRENT_SET,
+ "aggstate.current_set");
+ v_resvalue =
+ l_load_gep1(b, v_setno_gsetids, v_current_set, "");
+ v_resvalue =
+ LLVMBuildZExt(b, v_resvalue, TypeSizeT, "");
+
+ LLVMBuildStore(b, v_resvalue, v_resvaluep);
+ LLVMBuildStore(b, l_sbool_const(0), v_resnullp);
+
+ LLVMBuildBr(b, opblocks[opno + 1]);
+ break;
+ }
+
case EEOP_WINDOW_FUNC:
{
WindowFuncExprState *wfunc = op->d.window_func.wfstate;
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index 04b4c65858..de4dcfe165 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -990,8 +990,9 @@ _copyAgg(const Agg *from)
COPY_SCALAR_FIELD(numGroups);
COPY_SCALAR_FIELD(transitionSpace);
COPY_BITMAPSET_FIELD(aggParams);
- COPY_NODE_FIELD(groupingSets);
+ COPY_NODE_FIELD(rollup);
COPY_NODE_FIELD(chain);
+ COPY_NODE_FIELD(gsetid);
COPY_NODE_FIELD(sortnode);
return newnode;
@@ -1478,6 +1479,50 @@ _copyGroupingFunc(const GroupingFunc *from)
return newnode;
}
+/*
+ * _copyGroupingSetId
+ */
+static GroupingSetId *
+_copyGroupingSetId(const GroupingSetId *from)
+{
+ GroupingSetId *newnode = makeNode(GroupingSetId);
+
+ return newnode;
+}
+
+/*
+ * _copyRollupData
+ */
+static RollupData*
+_copyRollupData(const RollupData *from)
+{
+ RollupData *newnode = makeNode(RollupData);
+
+ COPY_NODE_FIELD(groupClause);
+ COPY_NODE_FIELD(gsets);
+ COPY_NODE_FIELD(gsets_data);
+ COPY_SCALAR_FIELD(numGroups);
+ COPY_SCALAR_FIELD(hashable);
+ COPY_SCALAR_FIELD(is_hashed);
+
+ return newnode;
+}
+
+/*
+ * _copyGroupingSetData
+ */
+static GroupingSetData *
+_copyGroupingSetData(const GroupingSetData *from)
+{
+ GroupingSetData *newnode = makeNode(GroupingSetData);
+
+ COPY_NODE_FIELD(set);
+ COPY_SCALAR_FIELD(setId);
+ COPY_SCALAR_FIELD(numGroups);
+
+ return newnode;
+}
+
/*
* _copyWindowFunc
*/
@@ -4972,6 +5017,9 @@ copyObjectImpl(const void *from)
case T_GroupingFunc:
retval = _copyGroupingFunc(from);
break;
+ case T_GroupingSetId:
+ retval = _copyGroupingSetId(from);
+ break;
case T_WindowFunc:
retval = _copyWindowFunc(from);
break;
@@ -5608,6 +5656,12 @@ copyObjectImpl(const void *from)
case T_SortGroupClause:
retval = _copySortGroupClause(from);
break;
+ case T_RollupData:
+ retval = _copyRollupData(from);
+ break;
+ case T_GroupingSetData:
+ retval = _copyGroupingSetData(from);
+ break;
case T_GroupingSet:
retval = _copyGroupingSet(from);
break;
diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c
index 88b912977e..6aa71d3723 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -3078,6 +3078,9 @@ equal(const void *a, const void *b)
case T_GroupingFunc:
retval = _equalGroupingFunc(a, b);
break;
+ case T_GroupingSetId:
+ retval = true;
+ break;
case T_WindowFunc:
retval = _equalWindowFunc(a, b);
break;
diff --git a/src/backend/nodes/nodeFuncs.c b/src/backend/nodes/nodeFuncs.c
index d85ca9f7c5..877ea0bc16 100644
--- a/src/backend/nodes/nodeFuncs.c
+++ b/src/backend/nodes/nodeFuncs.c
@@ -62,6 +62,9 @@ exprType(const Node *expr)
case T_GroupingFunc:
type = INT4OID;
break;
+ case T_GroupingSetId:
+ type = INT4OID;
+ break;
case T_WindowFunc:
type = ((const WindowFunc *) expr)->wintype;
break;
@@ -740,6 +743,9 @@ exprCollation(const Node *expr)
case T_GroupingFunc:
coll = InvalidOid;
break;
+ case T_GroupingSetId:
+ coll = InvalidOid;
+ break;
case T_WindowFunc:
coll = ((const WindowFunc *) expr)->wincollid;
break;
@@ -1869,6 +1875,7 @@ expression_tree_walker(Node *node,
case T_NextValueExpr:
case T_RangeTblRef:
case T_SortGroupClause:
+ case T_GroupingSetId:
/* primitive node types with no expression subnodes */
break;
case T_WithCheckOption:
@@ -2575,6 +2582,7 @@ expression_tree_mutator(Node *node,
case T_NextValueExpr:
case T_RangeTblRef:
case T_SortGroupClause:
+ case T_GroupingSetId:
return (Node *) copyObject(node);
case T_WithCheckOption:
{
diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c
index 5816d122c1..efcb1c7d4f 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -785,8 +785,9 @@ _outAgg(StringInfo str, const Agg *node)
WRITE_LONG_FIELD(numGroups);
WRITE_UINT64_FIELD(transitionSpace);
WRITE_BITMAPSET_FIELD(aggParams);
- WRITE_NODE_FIELD(groupingSets);
+ WRITE_NODE_FIELD(rollup);
WRITE_NODE_FIELD(chain);
+ WRITE_NODE_FIELD(gsetid);
WRITE_NODE_FIELD(sortnode);
}
@@ -1150,6 +1151,13 @@ _outGroupingFunc(StringInfo str, const GroupingFunc *node)
WRITE_LOCATION_FIELD(location);
}
+static void
+_outGroupingSetId(StringInfo str,
+ const GroupingSetId *node __attribute__((unused)))
+{
+ WRITE_NODE_TYPE("GROUPINGSETID");
+}
+
static void
_outWindowFunc(StringInfo str, const WindowFunc *node)
{
@@ -2002,6 +2010,7 @@ _outGroupingSetData(StringInfo str, const GroupingSetData *node)
WRITE_NODE_TYPE("GSDATA");
WRITE_NODE_FIELD(set);
+ WRITE_INT_FIELD(setId);
WRITE_FLOAT_FIELD(numGroups, "%.0f");
}
@@ -3847,6 +3856,9 @@ outNode(StringInfo str, const void *obj)
case T_GroupingFunc:
_outGroupingFunc(str, obj);
break;
+ case T_GroupingSetId:
+ _outGroupingSetId(str, obj);
+ break;
case T_WindowFunc:
_outWindowFunc(str, obj);
break;
diff --git a/src/backend/nodes/readfuncs.c b/src/backend/nodes/readfuncs.c
index af4fcfe1ee..c9a3340f58 100644
--- a/src/backend/nodes/readfuncs.c
+++ b/src/backend/nodes/readfuncs.c
@@ -636,6 +636,50 @@ _readGroupingFunc(void)
READ_DONE();
}
+/*
+ * _readGroupingSetId
+ */
+static GroupingSetId *
+_readGroupingSetId(void)
+{
+ READ_LOCALS_NO_FIELDS(GroupingSetId);
+
+ READ_DONE();
+}
+
+/*
+ * _readRollupData
+ */
+static RollupData *
+_readRollupData(void)
+{
+ READ_LOCALS(RollupData);
+
+ READ_NODE_FIELD(groupClause);
+ READ_NODE_FIELD(gsets);
+ READ_NODE_FIELD(gsets_data);
+ READ_FLOAT_FIELD(numGroups);
+ READ_BOOL_FIELD(hashable);
+ READ_BOOL_FIELD(is_hashed);
+
+ READ_DONE();
+}
+
+/*
+ * _readGroupingSetData
+ */
+static GroupingSetData *
+_readGroupingSetData(void)
+{
+ READ_LOCALS(GroupingSetData);
+
+ READ_NODE_FIELD(set);
+ READ_INT_FIELD(setId);
+ READ_FLOAT_FIELD(numGroups);
+
+ READ_DONE();
+}
+
/*
* _readWindowFunc
*/
@@ -2205,8 +2249,9 @@ _readAgg(void)
READ_LONG_FIELD(numGroups);
READ_UINT64_FIELD(transitionSpace);
READ_BITMAPSET_FIELD(aggParams);
- READ_NODE_FIELD(groupingSets);
+ READ_NODE_FIELD(rollup);
READ_NODE_FIELD(chain);
+ READ_NODE_FIELD(gsetid);
READ_NODE_FIELD(sortnode);
READ_DONE();
@@ -2642,6 +2687,12 @@ parseNodeString(void)
return_value = _readAggref();
else if (MATCH("GROUPINGFUNC", 12))
return_value = _readGroupingFunc();
+ else if (MATCH("GROUPINGSETID", 13))
+ return_value = _readGroupingSetId();
+ else if (MATCH("ROLLUP", 6))
+ return_value = _readRollupData();
+ else if (MATCH("GSDATA", 6))
+ return_value = _readGroupingSetData();
else if (MATCH("WINDOWFUNC", 10))
return_value = _readWindowFunc();
else if (MATCH("SUBSCRIPTINGREF", 15))
diff --git a/src/backend/optimizer/path/allpaths.c b/src/backend/optimizer/path/allpaths.c
index 905bbe77d8..e6c7f080e0 100644
--- a/src/backend/optimizer/path/allpaths.c
+++ b/src/backend/optimizer/path/allpaths.c
@@ -2710,8 +2710,11 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
/*
* For each useful ordering, we can consider an order-preserving Gather
- * Merge.
+ * Merge. Don't do this for partial groupingsets.
*/
+ if (root->parse->groupingSets)
+ return;
+
foreach(lc, rel->partial_pathlist)
{
Path *subpath = (Path *) lfirst(lc);
diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c
index 29f88bf0b7..64205893a3 100644
--- a/src/backend/optimizer/plan/createplan.c
+++ b/src/backend/optimizer/plan/createplan.c
@@ -1641,7 +1641,7 @@ create_unique_plan(PlannerInfo *root, UniquePath *best_path, int flags)
groupColIdx,
groupOperators,
groupCollations,
- NIL,
+ NULL,
NIL,
best_path->path.rows,
0,
@@ -2095,7 +2095,7 @@ create_agg_plan(PlannerInfo *root, AggPath *best_path)
extract_grouping_ops(best_path->groupClause),
extract_grouping_collations(best_path->groupClause,
subplan->targetlist),
- NIL,
+ NULL,
NIL,
best_path->numGroups,
best_path->transitionSpace,
@@ -2214,7 +2214,6 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
* never be grouping in an UPDATE/DELETE; but let's Assert that.
*/
Assert(root->inhTargetKind == INHKIND_NONE);
- Assert(root->grouping_map == NULL);
root->grouping_map = grouping_map;
/*
@@ -2237,10 +2236,13 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
new_grpColIdx = remap_groupColIdx(root, rollup->groupClause);
/*
+ * In final stage, rollup may contain empty set here
+ *
* FIXME This combination of nested if checks needs some explanation
* why we need this particular combination of flags.
*/
- if (!rollup->is_hashed)
+ if (!rollup->is_hashed &&
+ list_length(linitial(rollup->gsets)) != 0)
{
sort_plan = (Plan *)
make_sort_from_groupcols(rollup->groupClause,
@@ -2264,12 +2266,12 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
agg_plan = (Plan *) make_agg(NIL,
NIL,
strat,
- AGGSPLIT_SIMPLE,
+ best_path->aggsplit,
list_length((List *) linitial(rollup->gsets)),
new_grpColIdx,
extract_grouping_ops(rollup->groupClause),
extract_grouping_collations(rollup->groupClause, subplan->targetlist),
- rollup->gsets,
+ rollup,
NIL,
rollup->numGroups,
best_path->transitionSpace,
@@ -2281,8 +2283,7 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
}
/*
- * Now make the real Agg node
- */
+ * Now make the real Agg node */
{
RollupData *rollup = linitial(rollups);
AttrNumber *top_grpColIdx;
@@ -2314,12 +2315,12 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path)
plan = make_agg(build_path_tlist(root, &best_path->path),
best_path->qual,
best_path->aggstrategy,
- AGGSPLIT_SIMPLE,
+ best_path->aggsplit,
numGroupCols,
top_grpColIdx,
extract_grouping_ops(rollup->groupClause),
extract_grouping_collations(rollup->groupClause, subplan->targetlist),
- rollup->gsets,
+ rollup,
chain,
rollup->numGroups,
best_path->transitionSpace,
@@ -6221,7 +6222,7 @@ Agg *
make_agg(List *tlist, List *qual,
AggStrategy aggstrategy, AggSplit aggsplit,
int numGroupCols, AttrNumber *grpColIdx, Oid *grpOperators, Oid *grpCollations,
- List *groupingSets, List *chain, double dNumGroups,
+ RollupData *rollup, List *chain, double dNumGroups,
Size transitionSpace, Plan *sortnode, Plan *lefttree)
{
Agg *node = makeNode(Agg);
@@ -6240,8 +6241,9 @@ make_agg(List *tlist, List *qual,
node->numGroups = numGroups;
node->transitionSpace = transitionSpace;
node->aggParams = NULL; /* SS_finalize_plan() will fill this */
- node->groupingSets = groupingSets;
+ node->rollup= rollup;
node->chain = chain;
+ node->gsetid = NULL;
node->sortnode = sortnode;
plan->qual = qual;
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
index 68d9c88a53..cedd3e1c9d 100644
--- a/src/backend/optimizer/plan/planner.c
+++ b/src/backend/optimizer/plan/planner.c
@@ -113,6 +113,7 @@ typedef struct
Bitmapset *unhashable_refs;
List *unsortable_sets;
int *tleref_to_colnum_map;
+ int num_sets;
} grouping_sets_data;
/*
@@ -126,6 +127,8 @@ typedef struct
* clauses per Window */
} WindowClauseSortData;
+typedef void (*AddPathCallback) (RelOptInfo *parent_rel, Path *new_path);
+
/* Local functions */
static Node *preprocess_expression(PlannerInfo *root, Node *expr, int kind);
static void preprocess_qual_conditions(PlannerInfo *root, Node *jtnode);
@@ -142,7 +145,8 @@ static double preprocess_limit(PlannerInfo *root,
static void remove_useless_groupby_columns(PlannerInfo *root);
static List *preprocess_groupclause(PlannerInfo *root, List *force);
static List *extract_rollup_sets(List *groupingSets);
-static List *reorder_grouping_sets(List *groupingSets, List *sortclause);
+static List *reorder_grouping_sets(grouping_sets_data *gd,
+ List *groupingSets, List *sortclause);
static void standard_qp_callback(PlannerInfo *root, void *extra);
static double get_number_of_groups(PlannerInfo *root,
double path_rows,
@@ -176,7 +180,11 @@ static void consider_groupingsets_paths(PlannerInfo *root,
grouping_sets_data *gd,
const AggClauseCosts *agg_costs,
double dNumGroups,
- AggStrategy strat);
+ List *havingQual,
+ AggStrategy strat,
+ AggSplit aggsplit,
+ AddPathCallback add_path_fn);
+
static RelOptInfo *create_window_paths(PlannerInfo *root,
RelOptInfo *input_rel,
PathTarget *input_target,
@@ -250,6 +258,9 @@ static bool group_by_has_partkey(RelOptInfo *input_rel,
List *groupClause);
static int common_prefix_cmp(const void *a, const void *b);
+static List *extract_final_rollups(PlannerInfo *root,
+ grouping_sets_data *gd,
+ List *rollups);
/*****************************************************************************
*
@@ -2494,6 +2505,7 @@ preprocess_grouping_sets(PlannerInfo *root)
GroupingSetData *gs = makeNode(GroupingSetData);
gs->set = gset;
+ gs->setId = gd->num_sets++;
gd->unsortable_sets = lappend(gd->unsortable_sets, gs);
/*
@@ -2538,7 +2550,7 @@ preprocess_grouping_sets(PlannerInfo *root)
* largest-member-first, and applies the GroupingSetData annotations,
* though the data will be filled in later.
*/
- current_sets = reorder_grouping_sets(current_sets,
+ current_sets = reorder_grouping_sets(gd, current_sets,
(list_length(sets) == 1
? parse->sortClause
: NIL));
@@ -3547,7 +3559,7 @@ extract_rollup_sets(List *groupingSets)
* gets implemented in one pass.)
*/
static List *
-reorder_grouping_sets(List *groupingsets, List *sortclause)
+reorder_grouping_sets(grouping_sets_data *gd, List *groupingsets, List *sortclause)
{
ListCell *lc;
List *previous = NIL;
@@ -3581,6 +3593,7 @@ reorder_grouping_sets(List *groupingsets, List *sortclause)
previous = list_concat(previous, new_elems);
gs->set = list_copy(previous);
+ gs->setId = gd->num_sets++;
result = lcons(gs, result);
}
@@ -4191,8 +4204,14 @@ create_ordinary_grouping_paths(PlannerInfo *root, RelOptInfo *input_rel,
* The caller specifies the preferred aggregate strategy (sorted or hashed) using
* the strat aprameter. When the requested strategy is AGG_SORTED, the input path
* needs to be sorted accordingly (is_sorted needs to be true).
+ *
+ * The caller also needs to specify a callback used to add the path to the
+ * appropriate list - we can't simply use add_path, because with partial
+ * aggregation (PARTITIONWISE_AGGREGATE_PARTIAL) the path may need to be
+ * added to grouped_rel->pathlist. And aggsplit value is not sufficient to
+ * make a decision.
*/
-static void
+static void
consider_groupingsets_paths(PlannerInfo *root,
RelOptInfo *grouped_rel,
Path *path,
@@ -4201,9 +4220,11 @@ consider_groupingsets_paths(PlannerInfo *root,
grouping_sets_data *gd,
const AggClauseCosts *agg_costs,
double dNumGroups,
- AggStrategy strat)
+ List *havingQual,
+ AggStrategy strat,
+ AggSplit aggsplit,
+ AddPathCallback add_path_fn)
{
- Query *parse = root->parse;
Assert(strat == AGG_HASHED || strat == AGG_SORTED);
/*
@@ -4367,16 +4388,20 @@ consider_groupingsets_paths(PlannerInfo *root,
strat = AGG_MIXED;
}
- add_path(grouped_rel, (Path *)
- create_groupingsets_path(root,
- grouped_rel,
- path,
- (List *) parse->havingQual,
- strat,
- new_rollups,
- agg_costs,
- dNumGroups,
- is_sorted));
+ if (DO_AGGSPLIT_COMBINE(aggsplit))
+ new_rollups = extract_final_rollups(root, gd, new_rollups);
+
+ add_path_fn(grouped_rel, (Path *)
+ create_groupingsets_path(root,
+ grouped_rel,
+ path,
+ havingQual,
+ strat,
+ new_rollups,
+ agg_costs,
+ dNumGroups,
+ aggsplit,
+ is_sorted));
return;
}
@@ -4388,7 +4413,7 @@ consider_groupingsets_paths(PlannerInfo *root,
/*
* Callers consider AGG_SORTED strategy, the first rollup must
- * use non-hashed aggregate, 'is_sorted' tells whether the first
+ * use non-hashed aggregate, is_sorted tells whether the first
* rollup need to do its own sort.
*
* we try and make two paths: one sorted and one mixed
@@ -4533,16 +4558,20 @@ consider_groupingsets_paths(PlannerInfo *root,
if (rollups)
{
- add_path(grouped_rel, (Path *)
- create_groupingsets_path(root,
- grouped_rel,
- path,
- (List *) parse->havingQual,
- AGG_MIXED,
- rollups,
- agg_costs,
- dNumGroups,
- is_sorted));
+ if (DO_AGGSPLIT_COMBINE(aggsplit))
+ rollups = extract_final_rollups(root, gd, rollups);
+
+ add_path_fn(grouped_rel, (Path *)
+ create_groupingsets_path(root,
+ grouped_rel,
+ path,
+ havingQual,
+ AGG_MIXED,
+ rollups,
+ agg_costs,
+ dNumGroups,
+ aggsplit,
+ is_sorted));
}
}
@@ -4550,16 +4579,82 @@ consider_groupingsets_paths(PlannerInfo *root,
* Now try the simple sorted case.
*/
if (!gd->unsortable_sets)
- add_path(grouped_rel, (Path *)
- create_groupingsets_path(root,
- grouped_rel,
- path,
- (List *) parse->havingQual,
- AGG_SORTED,
- gd->rollups,
- agg_costs,
- dNumGroups,
- is_sorted));
+ {
+ List *rollups;
+
+ if (DO_AGGSPLIT_COMBINE(aggsplit))
+ rollups = extract_final_rollups(root, gd, gd->rollups);
+ else
+ rollups = gd->rollups;
+
+ add_path_fn(grouped_rel, (Path *)
+ create_groupingsets_path(root,
+ grouped_rel,
+ path,
+ havingQual,
+ AGG_SORTED,
+ rollups,
+ agg_costs,
+ dNumGroups,
+ aggsplit,
+ is_sorted));
+ }
+}
+
+/*
+ * If we are combining the partial groupingsets aggregation, the input is
+ * mixed with tuples from different grouping sets, executor dispatch the
+ * tuples to different rollups (phases) according to the grouping set id.
+ *
+ * We cannot use the same rollups with initial stage in which each tuple
+ * is processed by one or more grouping sets in one rollup, because in
+ * combining stage, each tuple only belong to one single grouping set.
+ * In this case, we use final_rollups instead in which each rollup has
+ * only one grouping set.
+ */
+static List *
+extract_final_rollups(PlannerInfo *root,
+ grouping_sets_data *gd,
+ List *rollups)
+{
+ ListCell *lc;
+ List *new_rollups = NIL;
+
+ foreach(lc, rollups)
+ {
+ ListCell *lc1;
+ RollupData *rollup = lfirst_node(RollupData, lc);
+
+ foreach(lc1, rollup->gsets_data)
+ {
+ GroupingSetData *gs = lfirst_node(GroupingSetData, lc1);
+ RollupData *new_rollup = makeNode(RollupData);
+
+ if (gs->set != NIL)
+ {
+ new_rollup->groupClause = preprocess_groupclause(root, gs->set);
+ new_rollup->gsets_data = list_make1(gs);
+ new_rollup->gsets = remap_to_groupclause_idx(new_rollup->groupClause,
+ new_rollup->gsets_data,
+ gd->tleref_to_colnum_map);
+ new_rollup->hashable = rollup->hashable;
+ new_rollup->is_hashed = rollup->is_hashed;
+ }
+ else
+ {
+ new_rollup->groupClause = NIL;
+ new_rollup->gsets_data = list_make1(gs);
+ new_rollup->gsets = list_make1(NIL);
+ new_rollup->hashable = false;
+ new_rollup->is_hashed = false;
+ }
+
+ new_rollup->numGroups = gs->numGroups;
+ new_rollups = lappend(new_rollups, new_rollup);
+ }
+ }
+
+ return new_rollups;
}
/*
@@ -5269,6 +5364,17 @@ make_partial_grouping_target(PlannerInfo *root,
add_new_columns_to_pathtarget(partial_target, non_group_exprs);
+ /*
+ * We are generate partial groupingsets path, add an expression to show
+ * the grouping set ID for a tuple, so in the final stage, executor can
+ * know which set this tuple belongs to and combine them.
+ * */
+ if (parse->groupingSets)
+ {
+ GroupingSetId *expr = makeNode(GroupingSetId);
+ add_new_column_to_pathtarget(partial_target, (Expr *)expr);
+ }
+
/*
* Adjust Aggrefs to put them in partial mode. At this point all Aggrefs
* are at the top level of the target list, so we can just scan the list
@@ -6433,7 +6539,10 @@ add_paths_to_grouping_rel(PlannerInfo *root, RelOptInfo *input_rel,
consider_groupingsets_paths(root, grouped_rel,
path, is_sorted, can_hash,
gd, agg_costs, dNumGroups,
- AGG_SORTED);
+ havingQual,
+ AGG_SORTED,
+ AGGSPLIT_SIMPLE,
+ add_path);
continue;
}
@@ -6494,15 +6603,37 @@ add_paths_to_grouping_rel(PlannerInfo *root, RelOptInfo *input_rel,
foreach(lc, partially_grouped_rel->pathlist)
{
Path *path = (Path *) lfirst(lc);
+ bool is_sorted;
+
+ is_sorted = pathkeys_contained_in(root->group_pathkeys,
+ path->pathkeys);
+
+ /*
+ * Use any available suitably-sorted path as input, and also
+ * consider sorting the cheapest-total path.
+ */
+ if (path != partially_grouped_rel->cheapest_total_path &&
+ !is_sorted)
+ continue;
+
+ if (parse->groupingSets)
+ {
+ consider_groupingsets_paths(root, grouped_rel,
+ path, is_sorted, can_hash,
+ gd, agg_final_costs, dNumGroups,
+ havingQual,
+ AGG_SORTED,
+ AGGSPLIT_FINAL_DESERIAL,
+ add_path);
+ continue;
+ }
/*
* Insert a Sort node, if required. But there's no point in
* sorting anything but the cheapest path.
*/
- if (!pathkeys_contained_in(root->group_pathkeys, path->pathkeys))
+ if (!is_sorted)
{
- if (path != partially_grouped_rel->cheapest_total_path)
- continue;
path = (Path *) create_sort_path(root,
grouped_rel,
path,
@@ -6546,7 +6677,10 @@ add_paths_to_grouping_rel(PlannerInfo *root, RelOptInfo *input_rel,
consider_groupingsets_paths(root, grouped_rel,
cheapest_path, false, true,
gd, agg_costs, dNumGroups,
- AGG_HASHED);
+ havingQual,
+ AGG_HASHED,
+ AGGSPLIT_SIMPLE,
+ add_path);
}
else
{
@@ -6589,22 +6723,39 @@ add_paths_to_grouping_rel(PlannerInfo *root, RelOptInfo *input_rel,
{
Path *path = partially_grouped_rel->cheapest_total_path;
- hashaggtablesize = estimate_hashagg_tablesize(path,
- agg_final_costs,
- dNumGroups);
+ if (parse->groupingSets)
+ {
+ /*
+ * Try for a hash-only groupingsets path over unsorted input.
+ */
+ consider_groupingsets_paths(root, grouped_rel,
+ path, false, true,
+ gd, agg_final_costs, dNumGroups,
+ havingQual,
+ AGG_HASHED,
+ AGGSPLIT_FINAL_DESERIAL,
+ add_path);
+ }
+ else
+ {
- if (hashaggtablesize < work_mem * 1024L)
- add_path(grouped_rel, (Path *)
- create_agg_path(root,
- grouped_rel,
- path,
- grouped_rel->reltarget,
- AGG_HASHED,
- AGGSPLIT_FINAL_DESERIAL,
- parse->groupClause,
- havingQual,
- agg_final_costs,
- dNumGroups));
+ hashaggtablesize = estimate_hashagg_tablesize(path,
+ agg_final_costs,
+ dNumGroups);
+
+ if (hashaggtablesize < work_mem * 1024L)
+ add_path(grouped_rel, (Path *)
+ create_agg_path(root,
+ grouped_rel,
+ path,
+ grouped_rel->reltarget,
+ AGG_HASHED,
+ AGGSPLIT_FINAL_DESERIAL,
+ parse->groupClause,
+ havingQual,
+ agg_final_costs,
+ dNumGroups));
+ }
}
}
@@ -6814,6 +6965,19 @@ create_partial_grouping_paths(PlannerInfo *root,
path->pathkeys);
if (path == cheapest_partial_path || is_sorted)
{
+ if (parse->groupingSets)
+ {
+ consider_groupingsets_paths(root, partially_grouped_rel,
+ path, is_sorted, can_hash,
+ gd, agg_partial_costs,
+ dNumPartialPartialGroups,
+ NIL,
+ AGG_SORTED,
+ AGGSPLIT_INITIAL_SERIAL,
+ add_partial_path);
+ continue;
+ }
+
/* Sort the cheapest partial path, if it isn't already */
if (!is_sorted)
path = (Path *) create_sort_path(root,
@@ -6821,7 +6985,7 @@ create_partial_grouping_paths(PlannerInfo *root,
path,
root->group_pathkeys,
-1.0);
-
+
if (parse->hasAggs)
add_partial_path(partially_grouped_rel, (Path *)
create_agg_path(root,
@@ -6883,26 +7047,41 @@ create_partial_grouping_paths(PlannerInfo *root,
{
double hashaggtablesize;
- hashaggtablesize =
- estimate_hashagg_tablesize(cheapest_partial_path,
- agg_partial_costs,
- dNumPartialPartialGroups);
-
- /* Do the same for partial paths. */
- if (hashaggtablesize < work_mem * 1024L &&
- cheapest_partial_path != NULL)
+ if (parse->groupingSets)
{
- add_partial_path(partially_grouped_rel, (Path *)
- create_agg_path(root,
- partially_grouped_rel,
- cheapest_partial_path,
- partially_grouped_rel->reltarget,
- AGG_HASHED,
- AGGSPLIT_INITIAL_SERIAL,
- parse->groupClause,
- NIL,
- agg_partial_costs,
- dNumPartialPartialGroups));
+ consider_groupingsets_paths(root, partially_grouped_rel,
+ cheapest_partial_path,
+ false, true,
+ gd, agg_partial_costs,
+ dNumPartialPartialGroups,
+ NIL,
+ AGG_HASHED,
+ AGGSPLIT_INITIAL_SERIAL,
+ add_partial_path);
+ }
+ else
+ {
+ hashaggtablesize =
+ estimate_hashagg_tablesize(cheapest_partial_path,
+ agg_partial_costs,
+ dNumPartialPartialGroups);
+
+ /* Do the same for partial paths. */
+ if (hashaggtablesize < work_mem * 1024L &&
+ cheapest_partial_path != NULL)
+ {
+ add_partial_path(partially_grouped_rel, (Path *)
+ create_agg_path(root,
+ partially_grouped_rel,
+ cheapest_partial_path,
+ partially_grouped_rel->reltarget,
+ AGG_HASHED,
+ AGGSPLIT_INITIAL_SERIAL,
+ parse->groupClause,
+ NIL,
+ agg_partial_costs,
+ dNumPartialPartialGroups));
+ }
}
}
@@ -6946,6 +7125,9 @@ gather_grouping_paths(PlannerInfo *root, RelOptInfo *rel)
generate_gather_paths(root, rel, true);
/* Try cheapest partial path + explicit Sort + Gather Merge. */
+ if (root->parse->groupingSets)
+ return;
+
cheapest_partial_path = linitial(rel->partial_pathlist);
if (!pathkeys_contained_in(root->group_pathkeys,
cheapest_partial_path->pathkeys))
@@ -6990,11 +7172,6 @@ can_partial_agg(PlannerInfo *root, const AggClauseCosts *agg_costs)
*/
return false;
}
- else if (parse->groupingSets)
- {
- /* We don't know how to do grouping sets in parallel. */
- return false;
- }
else if (agg_costs->hasNonPartial || agg_costs->hasNonSerial)
{
/* Insufficient support for partial mode. */
diff --git a/src/backend/optimizer/plan/setrefs.c b/src/backend/optimizer/plan/setrefs.c
index 3dcded506b..eae7d15701 100644
--- a/src/backend/optimizer/plan/setrefs.c
+++ b/src/backend/optimizer/plan/setrefs.c
@@ -754,6 +754,22 @@ set_plan_refs(PlannerInfo *root, Plan *plan, int rtoffset)
plan->qual = (List *)
convert_combining_aggrefs((Node *) plan->qual,
NULL);
+
+ /*
+ * If it's groupingsets, we must add expression to evaluate
+ * the grouping set ID and set the reference from the
+ * targetlist of child plan node.
+ */
+ if (agg->rollup)
+ {
+ GroupingSetId *expr = makeNode(GroupingSetId);
+ indexed_tlist *subplan_itlist = build_tlist_index(plan->lefttree->targetlist);
+
+ agg->gsetid = (Expr *) fix_upper_expr(root, (Node *)expr,
+ subplan_itlist,
+ OUTER_VAR,
+ rtoffset);
+ }
}
set_upper_references(root, plan, rtoffset);
diff --git a/src/backend/optimizer/util/pathnode.c b/src/backend/optimizer/util/pathnode.c
index ff8f676dfb..9fe6f6a003 100644
--- a/src/backend/optimizer/util/pathnode.c
+++ b/src/backend/optimizer/util/pathnode.c
@@ -2994,6 +2994,7 @@ create_groupingsets_path(PlannerInfo *root,
List *rollups,
const AggClauseCosts *agg_costs,
double numGroups,
+ AggSplit aggsplit,
bool is_sorted)
{
GroupingSetsPath *pathnode = makeNode(GroupingSetsPath);
@@ -3011,6 +3012,7 @@ create_groupingsets_path(PlannerInfo *root,
subpath->parallel_safe;
pathnode->path.parallel_workers = subpath->parallel_workers;
pathnode->subpath = subpath;
+ pathnode->aggsplit = aggsplit;
pathnode->is_sorted = is_sorted;
/*
@@ -3045,11 +3047,27 @@ create_groupingsets_path(PlannerInfo *root,
Assert(aggstrategy != AGG_PLAIN || list_length(rollups) == 1);
Assert(aggstrategy != AGG_MIXED || list_length(rollups) > 1);
+ /*
+ * Estimate the cost of groupingsets.
+ *
+ * If we are finalizing grouping sets, the subpath->rows
+ * contains rows from all sets, we need to estimate the
+ * number of rows in each rollup. Meanwhile, the cost of
+ * preprocess groupingsets is not estimated, the expression
+ * to redirect tuples is a simple Var expression which is
+ * normally cost zero.
+ */
foreach(lc, rollups)
{
RollupData *rollup = lfirst(lc);
List *gsets = rollup->gsets;
int numGroupCols = list_length(linitial(gsets));
+ int rows = 0.0;
+
+ if (DO_AGGSPLIT_COMBINE(aggsplit))
+ rows = rollup->numGroups * subpath->rows / numGroups;
+ else
+ rows = subpath->rows;
/*
* In AGG_SORTED or AGG_PLAIN mode, the first rollup do its own
@@ -3071,7 +3089,7 @@ create_groupingsets_path(PlannerInfo *root,
cost_sort(&sort_path, root, NIL,
input_total_cost,
- subpath->rows,
+ rows,
subpath->pathtarget->width,
0.0,
work_mem,
@@ -3089,7 +3107,7 @@ create_groupingsets_path(PlannerInfo *root,
having_qual,
input_startup_cost,
input_total_cost,
- subpath->rows);
+ rows);
is_first = false;
}
@@ -3101,7 +3119,6 @@ create_groupingsets_path(PlannerInfo *root,
sort_path.startup_cost = 0;
sort_path.total_cost = 0;
- sort_path.rows = subpath->rows;
rollup_strategy = rollup->is_hashed ?
AGG_HASHED : (numGroupCols ? AGG_SORTED : AGG_PLAIN);
@@ -3111,7 +3128,7 @@ create_groupingsets_path(PlannerInfo *root,
/* Account for cost of sort, but don't charge input cost again */
cost_sort(&sort_path, root, NIL,
0.0,
- subpath->rows,
+ rows,
subpath->pathtarget->width,
0.0,
work_mem,
@@ -3127,7 +3144,7 @@ create_groupingsets_path(PlannerInfo *root,
having_qual,
sort_path.startup_cost,
sort_path.total_cost,
- sort_path.rows);
+ rows);
pathnode->path.total_cost += agg_path.total_cost;
pathnode->path.rows += agg_path.rows;
diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c
index 5e63238f03..5779d158ba 100644
--- a/src/backend/utils/adt/ruleutils.c
+++ b/src/backend/utils/adt/ruleutils.c
@@ -7941,6 +7941,12 @@ get_rule_expr(Node *node, deparse_context *context,
}
break;
+ case T_GroupingSetId:
+ {
+ appendStringInfoString(buf, "GROUPINGSETID()");
+ }
+ break;
+
case T_WindowFunc:
get_windowfunc_expr((WindowFunc *) node, context);
break;
diff --git a/src/include/executor/execExpr.h b/src/include/executor/execExpr.h
index 4ed5d0a7de..4d36c2d77b 100644
--- a/src/include/executor/execExpr.h
+++ b/src/include/executor/execExpr.h
@@ -216,6 +216,7 @@ typedef enum ExprEvalOp
EEOP_XMLEXPR,
EEOP_AGGREF,
EEOP_GROUPING_FUNC,
+ EEOP_GROUPING_SET_ID,
EEOP_WINDOW_FUNC,
EEOP_SUBPLAN,
EEOP_ALTERNATIVE_SUBPLAN,
diff --git a/src/include/executor/nodeAgg.h b/src/include/executor/nodeAgg.h
index c5d4121c37..967af08af7 100644
--- a/src/include/executor/nodeAgg.h
+++ b/src/include/executor/nodeAgg.h
@@ -281,6 +281,8 @@ typedef struct AggStatePerPhaseData
List *concurrent_hashes; /* hash phases can do transition concurrently */
AggStatePerGroup *pergroups; /* pergroup states for a phase */
bool skip_build_trans;
+#define FIELDNO_AGGSTATEPERPHASE_SETNOGSETIDS 10
+ int *setno_gsetids; /* setno <-> gsetid map */
} AggStatePerPhaseData;
typedef struct AggStatePerPhaseSortData
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index 4081a0978e..dea5b10597 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -2047,6 +2047,7 @@ typedef struct AggState
int numtrans; /* number of pertrans items */
AggStrategy aggstrategy; /* strategy mode */
AggSplit aggsplit; /* agg-splitting mode, see nodes.h */
+#define FIELDNO_AGGSTATE_PHASE 6
AggStatePerPhase phase; /* pointer to current phase data */
int numphases; /* number of phases (including phase 0) */
int current_phase; /* current phase number */
@@ -2070,8 +2071,6 @@ typedef struct AggState
/* These fields are for grouping set phase data */
int maxsets; /* The max number of sets in any phase */
AggStatePerPhase *phases; /* array of all phases */
- Tuplesortstate *sort_in; /* sorted input to phases > 1 */
- Tuplesortstate *sort_out; /* input is copied here for next phase */
TupleTableSlot *sort_slot; /* slot for sort results */
/* these fields are used in AGG_PLAIN and AGG_SORTED modes: */
HeapTuple grp_firstTuple; /* copy of first tuple of current group */
@@ -2083,6 +2082,11 @@ typedef struct AggState
int eflags; /* eflags for the first sort */
ProjectionInfo *combinedproj; /* projection machinery */
+
+ /* these field are used in parallel grouping sets */
+ bool groupingsets_preprocess; /* groupingsets preprocessed yet? */
+ ExprState *gsetid; /* expression state to get grpsetid from input */
+ int *gsetid_phaseidxs; /* grpsetid <-> phaseidx mapping */
} AggState;
/* ----------------
diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h
index 8a76afe8cc..a48a7af0e3 100644
--- a/src/include/nodes/nodes.h
+++ b/src/include/nodes/nodes.h
@@ -153,6 +153,7 @@ typedef enum NodeTag
T_Param,
T_Aggref,
T_GroupingFunc,
+ T_GroupingSetId,
T_WindowFunc,
T_SubscriptingRef,
T_FuncExpr,
diff --git a/src/include/nodes/pathnodes.h b/src/include/nodes/pathnodes.h
index c1e69c808f..2761fa6d01 100644
--- a/src/include/nodes/pathnodes.h
+++ b/src/include/nodes/pathnodes.h
@@ -1676,6 +1676,7 @@ typedef struct GroupingSetData
{
NodeTag type;
List *set; /* grouping set as list of sortgrouprefs */
+ int setId; /* unique grouping set identifier */
double numGroups; /* est. number of result groups */
} GroupingSetData;
@@ -1702,6 +1703,7 @@ typedef struct GroupingSetsPath
List *rollups; /* list of RollupData */
List *qual; /* quals (HAVING quals), if any */
uint64 transitionSpace; /* for pass-by-ref transition data */
+ AggSplit aggsplit; /* agg-splitting mode, see nodes.h */
bool is_sorted; /* input sorted in groupcols of first rollup */
} GroupingSetsPath;
diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h
index 3cd2537e9e..5b1239adf2 100644
--- a/src/include/nodes/plannodes.h
+++ b/src/include/nodes/plannodes.h
@@ -20,6 +20,7 @@
#include "nodes/bitmapset.h"
#include "nodes/lockoptions.h"
#include "nodes/primnodes.h"
+#include "nodes/pathnodes.h"
/* ----------------------------------------------------------------
@@ -816,8 +817,9 @@ typedef struct Agg
uint64 transitionSpace; /* for pass-by-ref transition data */
Bitmapset *aggParams; /* IDs of Params used in Aggref inputs */
/* Note: planner provides numGroups & aggParams only in HASHED/MIXED case */
- List *groupingSets; /* grouping sets to use */
+ RollupData *rollup; /* grouping sets to use */
List *chain; /* chained Agg/Sort nodes */
+ Expr *gsetid; /* expression to fetch grouping set id */
Plan *sortnode; /* agg does its own sort, only used by grouping sets now */
} Agg;
diff --git a/src/include/nodes/primnodes.h b/src/include/nodes/primnodes.h
index d73be2ad46..f8f85d431a 100644
--- a/src/include/nodes/primnodes.h
+++ b/src/include/nodes/primnodes.h
@@ -364,6 +364,12 @@ typedef struct GroupingFunc
int location; /* token location */
} GroupingFunc;
+/* GroupingSetId */
+typedef struct GroupingSetId
+{
+ Expr xpr;
+} GroupingSetId;
+
/*
* WindowFunc
*/
diff --git a/src/include/optimizer/pathnode.h b/src/include/optimizer/pathnode.h
index f9f388ba06..4fde8b22bf 100644
--- a/src/include/optimizer/pathnode.h
+++ b/src/include/optimizer/pathnode.h
@@ -218,6 +218,7 @@ extern GroupingSetsPath *create_groupingsets_path(PlannerInfo *root,
List *rollups,
const AggClauseCosts *agg_costs,
double numGroups,
+ AggSplit aggsplit,
bool is_sorted);
extern MinMaxAggPath *create_minmaxagg_path(PlannerInfo *root,
RelOptInfo *rel,
diff --git a/src/include/optimizer/planmain.h b/src/include/optimizer/planmain.h
index 5954ff3997..e987011328 100644
--- a/src/include/optimizer/planmain.h
+++ b/src/include/optimizer/planmain.h
@@ -54,7 +54,7 @@ extern Sort *make_sort_from_sortclauses(List *sortcls, Plan *lefttree);
extern Agg *make_agg(List *tlist, List *qual,
AggStrategy aggstrategy, AggSplit aggsplit,
int numGroupCols, AttrNumber *grpColIdx, Oid *grpOperators, Oid *grpCollations,
- List *groupingSets, List *chain, double dNumGroups,
+ RollupData *rollup, List *chain, double dNumGroups,
Size transitionSpace, Plan *sortnode, Plan *lefttree);
extern Limit *make_limit(Plan *lefttree, Node *limitOffset, Node *limitCount);
--
2.21.1
0007-fix.patch (text/plain, 5.9K) download
From b87116ee21a59501ef9d9decfebf1cf5aa48d734 Mon Sep 17 00:00:00 2001
From: Tomas Vondra <tv@fuzzy.cz>
Date: Thu, 19 Mar 2020 03:02:42 +0100
Subject: [PATCH 7/7] fix
---
src/backend/executor/nodeAgg.c | 20 ++++++++++----------
src/backend/jit/llvm/llvmjit_expr.c | 2 +-
src/backend/optimizer/plan/planner.c | 12 ++++++------
3 files changed, 17 insertions(+), 17 deletions(-)
diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c
index f7b98dd798..51c7f229e2 100644
--- a/src/backend/executor/nodeAgg.c
+++ b/src/backend/executor/nodeAgg.c
@@ -360,7 +360,7 @@ initialize_phase(AggState *aggstate, int newphase)
if (persort->store_in)
{
tuplestore_end(persort->store_in);
- persort->store_in = NULL;
+ persort->store_in = NULL;
}
}
@@ -2017,7 +2017,7 @@ agg_preprocess_groupingsets(AggState *aggstate)
/* Initialize tuples storage for each aggregate phases */
for (phaseidx = 0; phaseidx < aggstate->numphases; phaseidx++)
{
- phase = aggstate->phases[phaseidx];
+ phase = aggstate->phases[phaseidx];
if (!phase->is_hashed)
{
@@ -2039,12 +2039,12 @@ agg_preprocess_groupingsets(AggState *aggstate)
}
else
{
- persort->store_in = tuplestore_begin_heap(false, false, work_mem);
+ persort->store_in = tuplestore_begin_heap(false, false, work_mem);
}
}
else
{
- /*
+ /*
* If it's a AGG_HASHED, we don't need a storage to store
* the tuples for later process, we can do the transition
* immediately.
@@ -2422,7 +2422,7 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
aggstate->maxsets = numGroupingSets;
aggstate->numphases = 1 + list_length(node->chain);
- /*
+ /*
* We are doing final stage of partial groupingsets, do preprocess
* to input tuples first, redirect the tuples to according aggregate
* phases. See agg_preprocess_groupingsets().
@@ -2431,7 +2431,7 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
{
aggstate->groupingsets_preprocess = true;
- /*
+ /*
* Allocate gsetid <-> phases mapping, in final stage of
* partial groupingsets, all grouping sets are extracted
* to individual phases, so the number of sets is equal
@@ -2449,7 +2449,7 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
* agg_sort_input(), this can only happen in groupingsets case.
*/
if (node->sortnode)
- aggstate->input_sorted = false;
+ aggstate->input_sorted = false;
aggstate->aggcontexts = (ExprContext **)
palloc0(sizeof(ExprContext *) * numGroupingSets);
@@ -2626,7 +2626,7 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
phasedata->setno_gsetids[0] = gs->setId;
}
- /*
+ /*
* Initialize pergroup state. For AGG_HASHED, all groups do transition
* on the fly, all pergroup states are kept in hashtable, everytime
* a tuple is processed, lookup_hash_entry() choose one group and
@@ -2688,7 +2688,7 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
phasedata->grouped_cols[i] = cols;
phasedata->gset_lengths[i] = current_length;
- /*
+ /*
* In the initial stage of partial grouping sets, it provides extra
* grouping sets ID in the targetlist, fill the setno <-> gsetid
* map, so EEOP_GROUPING_SET_ID can evaluate correct gsetid for
@@ -3871,7 +3871,7 @@ ExecReScanAgg(AggState *node)
}
}
- /*
+ /*
* if the agg is doing final stage of partial groupingsets, reset the
* flag to do groupingsets preprocess again.
*/
diff --git a/src/backend/jit/llvm/llvmjit_expr.c b/src/backend/jit/llvm/llvmjit_expr.c
index f442442269..f70eaabd0c 100644
--- a/src/backend/jit/llvm/llvmjit_expr.c
+++ b/src/backend/jit/llvm/llvmjit_expr.c
@@ -1893,7 +1893,7 @@ llvm_compile_expr(ExprState *state)
v_aggstatep =
LLVMBuildBitCast(b, v_parent, l_ptr(StructAggState), "");
- /*
+ /*
* op->resvalue =
* aggstate->phase->setno_gsetids
* [aggstate->current_set]
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
index cedd3e1c9d..a0186091a1 100644
--- a/src/backend/optimizer/plan/planner.c
+++ b/src/backend/optimizer/plan/planner.c
@@ -4211,7 +4211,7 @@ create_ordinary_grouping_paths(PlannerInfo *root, RelOptInfo *input_rel,
* added to grouped_rel->pathlist. And aggsplit value is not sufficient to
* make a decision.
*/
-static void
+static void
consider_groupingsets_paths(PlannerInfo *root,
RelOptInfo *grouped_rel,
Path *path,
@@ -4601,7 +4601,7 @@ consider_groupingsets_paths(PlannerInfo *root,
}
}
-/*
+/*
* If we are combining the partial groupingsets aggregation, the input is
* mixed with tuples from different grouping sets, executor dispatch the
* tuples to different rollups (phases) according to the grouping set id.
@@ -4644,7 +4644,7 @@ extract_final_rollups(PlannerInfo *root,
{
new_rollup->groupClause = NIL;
new_rollup->gsets_data = list_make1(gs);
- new_rollup->gsets = list_make1(NIL);
+ new_rollup->gsets = list_make1(NIL);
new_rollup->hashable = false;
new_rollup->is_hashed = false;
}
@@ -5364,7 +5364,7 @@ make_partial_grouping_target(PlannerInfo *root,
add_new_columns_to_pathtarget(partial_target, non_group_exprs);
- /*
+ /*
* We are generate partial groupingsets path, add an expression to show
* the grouping set ID for a tuple, so in the final stage, executor can
* know which set this tuple belongs to and combine them.
@@ -6985,7 +6985,7 @@ create_partial_grouping_paths(PlannerInfo *root,
path,
root->group_pathkeys,
-1.0);
-
+
if (parse->hasAggs)
add_partial_path(partially_grouped_rel, (Path *)
create_agg_path(root,
@@ -7059,7 +7059,7 @@ create_partial_grouping_paths(PlannerInfo *root,
AGGSPLIT_INITIAL_SERIAL,
add_partial_path);
}
- else
+ else
{
hashaggtablesize =
estimate_hashagg_tablesize(cheapest_partial_path,
--
2.21.1
view thread (30+ messages) latest in thread
Message-ID: <20200319020914.zkcjpm5r4yj7umqx@development>
Permalink: ../20200319020914.zkcjpm5r4yj7umqx@development/
Also on: postgresql.org/message-id/20200319020914.zkcjpm5r4yj7umqx@development
copy link · copy postgr.es · copy git am
reply Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: pgsql-hackers@postgresql.org
Cc: tomas.vondra@2ndquadrant.com, ptang@pivotal.io, guofenglinux@gmail.com, sbjesse@gmail.com, riguo@pivotal.io, amit.kapila16@gmail.com, michael@paquier.xyz, andres@anarazel.de, andrew@tao11.riddles.org.uk
Subject: Re: Parallel grouping sets
In-Reply-To: <20200319020914.zkcjpm5r4yj7umqx@development>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by DDX for PostgreSQL; see mirroring instructions
for how to clone and mirror all data and code used for this inbox