PostgreSQL 技术内幕 (三) 聚集算子

聚集算子 (Aggregation) 是数据库最常用的功能之一。在上次的直播中，我们主要介绍了 Parallel HashAggregation 的基本实现细节。

本文根据直播文字整理，作者现任 HashData 内核研发工程师。

1.聚集算子概述

常见的聚集算子实现有两种，一种是基于哈希表的实现（哈希聚集），另一种则是基于排序的实现（分组聚集）。不过这两种方法通常来说都是针对有 group by 的查询而言的。

对于没有 group by 的查询来说，我们只需要扫描一遍表，并对每一个元组进行 count\sum\min 等累积计算操作即可，这种情况在 PG 中被称为 plain （朴素）聚集。而 count\sum\min 这样的函数，我们称为聚集函数。

在 Postgres 中，聚集函数的计算最多可能被分为 3 个步骤，分别是转换函数、合并函数和后处理函数；其中,合并函数在并行聚集中才会用到。

2.转换函数、后处理函数

我们先通过最简单的朴素聚集来介绍一下聚集算子的基本思想和转换函数、后处理函数的作用；我们用下面这个求 tenk1 表中所有数据 ten 列均值的 SQL 查询为例进行讨论。

select avg(ten) from tenk1;
explain select avg(ten) from tenk1;                           QUERY PLAN----------------------------------------------------------------- Aggregate  (cost=470.00..470.01 rows=1 width=32)   ->  Seq Scan on tenk1  (cost=0.00..445.00 rows=10000 width=4)(2 rows)

如果手动来写一段程序来实现这个功能，很容易写出类似于下面的代码：

for (auto tuple: tuples) {    aggstate->sum += tuple.ten;    aggstate->cnt ++;}
avg = aggstate->sum / aggstate->cnt;

事实上，Postgres 对聚集算子的实现方式是类似的。由于 Postgres 采用的是火山模型，我们将会从下层算子中一条条读出元组，随后在聚集算子中将每个元组聚集所需要的信息积累到一个状态变量 AggState 中。而累积的过程，本质上就是将新元组的值和累积的状态进行某种运算，运算的具体实现就是转换函数。

比如，在我们的例子中想要求的是表中 ten 列的均值，所需要累积的状态就包括此前扫描元组的总数和扫描元组 ten 列之和，其中对每个元组 ten 列进行累加和计数的工作就是转换函数所实现的。而最后，扫描完整张表，我们有了所有元组数量和 ten 列之和，要求均值还需要进行一次平均操作，即用总和除以总数，这一步就是用后处理函数实现的。

显然，不是每种聚集都需要后处理函数，比如 min\max 只需要转换函数就足够完成整个聚集过程。

完成这两步之后，整个朴素聚集过程也就完成了，最终执行器会将聚集的结果返回给上层算子；整个流程大致如上图所示：右侧的 int4_avg_accum 和 int8_avg 就是 postgres 在执行例子中的查询时真正调用的 transfn 和 finalfn 。

3.合并函数与并行聚集

既然有了转换函数和后处理函数我们已经可以完成对聚集算子的正确实现，那合并函数是用来做什么的呢？

首先，我们先要了解 PG 并行聚集的思想和其底层需要依赖的并行扫描。

聚集算子本身是一个 CPU 密集型的算子，如果我们的聚集是基于单进程实现的，那就无法利用现代 CPU 多核架构的能力，可能导致执行过程阻塞在 CPU 上。即使还没有达到 IO 设备的性能瓶颈，也无法提升整个执行过程速度。

对此，解决的方案就是实现并行聚集。其实在 PG14 中，并行聚集已经默认打开；如果表足够大，或者像下面这样调整一些 GUC，同样的查询就会生成并行聚集的查询计划了。

SET parallel_setup_cost = 0;SET parallel_tuple_cost = 0;SET min_parallel_table_scan_size = 0;explain select avg(ten) from tenk1;                                     QUERY PLAN------------------------------------------------------------------------------------- Finalize Aggregate  (cost=397.10..397.11 rows=1 width=32)   ->  Gather  (cost=397.08..397.09 rows=2 width=32)         Workers Planned: 3         ->  Partial Aggregate  (cost=397.08..397.09 rows=1 width=32)               ->  Parallel Seq Scan on tenk1  (cost=0.00..386.67 rows=4167 width=4)(5 rows)

例子中的并行查询计划可以认为也只包括两个部分，分别是下层的顺序扫描算子和上层的聚集算子。但这次，上层的聚集算子则不再是简单的一阶段聚集即可完成的，而是分成了部分聚集(Partial Aggregate)和最终聚集(Finalize Aggregate)两个阶段完成。示意图如下：

在 PG 的实现中，并行算子底层都需要依赖并行扫描，因此我们会根据查询计划，由 Leader 进程先启动若干个工作进程进行并行扫描。通常 Leader 也会和进程一起工作。

随后，为了更高效地利用 CPU 资源，我们不会直接像简单的 select * from tenk; 语句这样，直接对并行扫描出来的元组直接进行收集，然后用唯一的 Leader 进程进行后续的聚集计算。而更好的方式是，我们可以让每个工作进程直接对自己收集的一部分元组先进行一次聚集计算，得到一个部分聚集的结果；再将这个部分聚集的结果汇总到 Leader 进程，由 Leader 进程进行最终聚集。这样，每个工作进程就可以最大化地利用系统 CPU 资源了。

这里要强调一下，仅仅汇聚(Gather)部分聚集的结果是不正确的，因此最终聚集是必须存在的。在朴素聚集的例子中这件事其实是很好理解的，因为每个工作进程只扫描到了一部分数据，不管是做 SUM\MIN\MAX 这样不需要后处理函数的聚集还是 AVG 这样需要后处理函数的聚集，每个工作进程聚集出来的结果也只会是全表聚集结果的一部分。我们除了将每个工作进程的聚集结果汇聚起来，当然还需要对他们进行合并，这就是所谓合并函数的用武之地了。

不同的聚集函数对应的合并函数显然也是不一样的。比如在我们的例子中求的是均值，则合并函数的工作就应该是收集每个 Worker 所部分聚集的 count 和 sum，并分别对这两者进行累加；最终再交由后处理函数进行计算。对于本例， PG 最终会采用 int4_avg_combine 进行合并。

Datumint4_avg_combine(PG_FUNCTION_ARGS){    state1 = (Int8TransTypeData *) ARR_DATA_PTR(transarray1);    state2 = (Int8TransTypeData *) ARR_DATA_PTR(transarray2);
    state1->count += state2->count;    state1->sum += state2->sum;
    PG_RETURN_ARRAYTYPE_P(transarray1);}

所以整个执行流程大致如下：在 gather 之前，PG 共有 N 个 Worker 并行扫描和部分聚集（Leader 进程也可参与工作）；在 gather 之后，所有算子都在 Leader 进程中执行，我们也通常称这种做法为两阶段聚集。这里还有一点需要注意，就是在部分聚集的时候，我们是不能执行后处理函数的。我们可以通过 aggstate->aggsplit 区分是否需要在聚集算子中执行后处理函数。

4.Group By

不过，朴素聚集只是聚集算子中非常特殊的一种情况。在大部分情况下，我们的查询都是带有 group by 关键字的。以实验所用的查询为例，我们希望返回的不再是全表的均值，而是按照 stringu1 字段的第一个字母分类之后，每个类别 ten 列的均值。由于在我们的数据集中，每一个首字母都出现了，所以最终会返回 26 行数据；并且数据整体分布比较均衡，每个首字母的均值都在 4.5 左右。

select left(stringu1, 1) as FirstLetter, avg(ten) from tenk1 group by left(stringu1, 1); firstletter |        avg-------------+-------------------- N           | 5.0000000000000000 O           | 4.0000000000000000 V           | 5.0000000000000000 L           | 5.0000000000000000 Z           | 4.9895833333333333 M           | 4.0000000000000000 D           | 5.0000000000000000 Q           | 4.0104166666666667 G           | 4.0000000000000000 X           | 4.9947916666666667 J           | 5.0000000000000000 P           | 5.0000000000000000 I           | 4.0000000000000000 U           | 4.0000000000000000 K           | 4.0000000000000000 A           | 4.0000000000000000 Y           | 3.9895833333333333 R           | 5.0104166666666667 W           | 3.9947916666666667 E           | 4.0000000000000000 B           | 5.0000000000000000 C           | 4.0000000000000000 T           | 5.0052083333333333 H           | 5.0000000000000000 F           | 5.0000000000000000 S           | 4.0052083333333333(26 rows)

那对于带有 group by 的查询，我们又应该如何实现聚集算子呢？其本质就是要把列值相同的行放在一起，这样我们就可以分门别类的进行状态累积了；类似的需求在哪里出现过呢？

目前，各个语言基础库中都会实现的 Map<Key, Value>字典类，经常有两种实现方式，一种是基于哈希思想的，一种是基于平衡树或者跳表这样的有序集合；其本质可以认为就是实现了分门别类的功能。因此，对于 group by 聚集这样的要求，也有两种主流的实现方式；一种是基于哈希表，我们称为哈希聚集；另一种则要求先对元组进行排序，我们称为分组聚集。

5.哈希聚集

哈希聚集的实现和朴素聚集的主要区别就在于在算子中建立一张哈希表。

非并行的哈希聚集整体上也可以分为两个步骤：第一个步骤执行转移函数，第二个步骤执行后处理函数并输出元组到上层算子。

在第一个步骤中，我们依旧逐行扫描元组。与朴素聚集直接累积状态不同，我们首先要计算元组的哈希值并探查哈希表。如果发现记录不存在，则需要先在哈希表中插入一条新的记录。反之，则基于当前元组和哈希表中找到的记录执行转换函数，更新累积状态至哈希表中。

第二个步骤，则是直接遍历哈希表，依次对每个状态进行后处理，并逐行输出最后的聚集结果给上层算子使用。

上图为 CMU-15445 课件中的一张图，很好地阐述了这个过程。在这个例子中，我们要求的是选修 CMU 各个课程的平均分，就需要将每个同学的考试成绩按照课程编号进行分类。我们以课程编号为键，为每门课累积的状态为值建立一张哈希表。然后遍历所有的考试成绩，累积 count 和 sum 到哈希表；这个过程就是在不断地执行转换函数。当全表扫描完毕，遍历所建的哈希表，调用后处理函数也就是求解平均值，再输出给上层算子即完成了整个哈希聚集的过程。

6.分组聚集

分组聚集基于排序思想进行分类，其思路其实和朴素聚集更加接近，从 PG 的代码实现中也可见一斑。两者共享着大部分的代码，唯一的区别在于分组聚集会先对元组进行排序（通常是加一个 Sort 算子作为下层节点）。

        switch (node->phase->aggstrategy)        {            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:                result = agg_retrieve_direct(node);                break;        }

排序之后，所有键相同的元素自然会排列在一起；我们只需从头到尾遍历一遍，用和朴素聚集一样的方式进行累积状态，然后在每次扫描到和上一个元素不同的键时截断并进行后再发给上层节点即可；两者的代码区别很有限，因此复用一样的代码逻辑也就很好理解了。

7.小结

就哈希聚集和分组聚集这两种算法来说，通常在表比较大的时候，排序的代价是比较高的，因此代价模型常常会选出哈希聚集作为聚集方式。不过，这也并非绝对，比如在下层算子本身已经保证有序性的前提下，我们不需要付出额外的代价进行排序；而分组聚集算子本身只需要依次遍历输出即可，不需要像哈希表这样占据大量的内存空间，也更不会有 spill to disk 的额外成本，在这种时候就常常成为更优的选择。

而关于并行，事实上，哈希聚集和分组聚集都可以成为 partial agg 或者 finalize agg 的选择；他们和并行本身是正交的，可以组合使用。

感兴趣的同学可以思考一下：相比于朴素聚集，FinalizeGroupAggregate 在实现时有哪些需要注意的地方？与朴素聚集的实现又有什么区别？欢迎大家留言讨论。