Parallel grouping sets

Hi all,

Paul and I have been hacking recently to implement parallel grouping
sets, and here we have two implementations.

Implementation 1
================

Attached is the patch and also there is a github branch [1] for this
work.

Parallel aggregation has already been supported in PostgreSQL and it is
implemented by aggregating in two stages. First, each worker performs an
aggregation step, producing a partial result for each group of which
that process is aware. Second, the partial results are transferred to
the leader via the Gather node. Finally, the leader merges the partial
results and produces the final result for each group.

We are implementing parallel grouping sets in the same way. The only
difference is that in the final stage, the leader performs a grouping
sets aggregation, rather than a normal aggregation.

The plan looks like:

# explain (costs off, verbose) select c1, c2, avg(c3) from t2 group by grouping sets((c1,c2), (c1), (c2,c3));
                       QUERY PLAN
---------------------------------------------------------
 Finalize MixedAggregate
   Output: c1, c2, avg(c3), c3
   Hash Key: t2.c2, t2.c3
   Group Key: t2.c1, t2.c2
   Group Key: t2.c1
   ->  Gather Merge
         Output: c1, c2, c3, (PARTIAL avg(c3))
         Workers Planned: 2
         ->  Sort
               Output: c1, c2, c3, (PARTIAL avg(c3))
               Sort Key: t2.c1, t2.c2
               ->  Partial HashAggregate
                     Output: c1, c2, c3, PARTIAL avg(c3)
                     Group Key: t2.c1, t2.c2, t2.c3
                     ->  Parallel Seq Scan on public.t2
                           Output: c1, c2, c3
(16 rows)

As the partial aggregation can be performed in parallel, we can expect a
speedup if the number of groups seen by the Finalize Aggregate node is
some less than the number of input rows.

For example, for the table provided in the test case within the patch,
running the above query in my Linux box:

# explain analyze select c1, c2, avg(c3) from t2 group by grouping sets((c1,c2), (c1), (c2,c3)); -- without patch
 Planning Time: 0.123 ms
 Execution Time: 9459.362 ms

# explain analyze select c1, c2, avg(c3) from t2 group by grouping sets((c1,c2), (c1), (c2,c3)); -- with patch
 Planning Time: 0.204 ms
 Execution Time: 1077.654 ms

But sometimes we may not benefit from this patch. For example, in the
worst-case scenario the number of groups seen by the Finalize Aggregate
node could be as many as the number of input rows which were seen by all
worker processes in the Partial Aggregate stage. This is prone to
happening with this patch, because the group key for Partial Aggregate
is all the columns involved in the grouping sets, such as in the above
query, it is (c1, c2, c3).

So, we have been working on another way to implement parallel grouping
sets.