PostgreSQL技术内幕26：PG聚合算子实现分析

0.简介

聚合算子在聚合函数在数据分析、报告生成和统计计算中扮演着重要角色，通过对多行数据进行计算，将多个输入值压缩为单一输出值，如求和、平均值、计数等。本文将通过对PostgreSQL聚合算子的源码进行解析，深入了解其实现机制。

1.概念说明

聚合函数：聚合函数是用于对一组值进行计算并返回单个结果的函数。它们通常与GROUP BY子句结合使用，将数据分组后对每组进行计算。例如，COUNT函数用于计算行数或非NULL值的数量，SUM函数用于计算数值列的总和，AVG函数用于计算数值列的平均值，MAX和MIN函数分别用于计算数值列中的最大值和最小值。
常见实现：常见的聚合算子实现分为两种，一种是基于哈希表的实现（哈希聚集），另外一种是基于排序的实现（分组聚集），这两种聚集方式一般来说都是对于带有group by的查询来说的。因为对于没有group by的查询来说，只需要扫描一遍表并对相应元组进行累计操作即可（也被称为朴素聚集）。
聚合整体实现：PG聚合操作整体可以分为三个步骤：1.扫描数据并生成中间值；2.收集并对中间值进行合并（并行场景下需要）；3.最终结果生成。这几个步骤可以在pg_aggregate系统表对应查看，下面是sum函数int类型对应的信息，主要看以下几个信息，aggtranstype表示中间值是一个数组，初始值agginitval是空，也就是0，三个基本步骤就是aggtransfn、aggcollectfn、aggfinalfn列。

postgres=# select * from pg_aggregate where aggfnoid = 2109;aggfnoid    | aggkind | aggnumdirectargs | aggtransfn | aggfinalfn | aggcombinefn | aggserialfn | aggdeserialfn |  a
ggmtransfn   |   aggminvtransfn   | aggmfinalfn  | aggfinalextra | aggmfinalextra | aggfinalmodify | aggmfinalmodify | a
ggsortop | aggtranstype | aggtransspace | aggmtranstype | aggmtransspace | agginitval | aggminitval
----------------+---------+------------------+------------+------------+--------------+-------------+---------------+---
-------------+--------------------+--------------+---------------+----------------+----------------+-----------------+--
---------+--------------+---------------+---------------+----------------+------------+-------------pg_catalog.sum | n       |                0 | int2_sum   | -          | int8pl       | -           | -             | in
t2_avg_accum | int2_avg_accum_inv | int2int4_sum | f             | f              | r              | r               |0 |           20 |             0 |          1016 |              0 |            | {0,0}
(1 row)

2.朴素聚集

因为sum函数并不需要最后的最终结果生成，只需要转换函数即可，所以下面会以求平均值作为例子，扫描数据并生成中间值就是将所有元组进行累加，最终结果生成就是对累加结果进行sum/count。

上图展示了无并行的朴素聚集的流程，但由于是单进程实现而聚集又是一个cpu密集算子，其不能利用cpu多核的优势，从而导致性能的瓶颈。而解决方案就是并行进行扫描和初始聚集，由leader节点进行收集和后处理。

3.Group by聚集

3.1 哈希聚集

非并行哈希函数整体包含两个阶段：1.执行数据扫描构建哈希表；2.执行后处理然后输出给上层算子。
第一个步骤做的实际就是扫描元组然后计算其对应的哈希值，如果不存在需要插入，如何存在则更新。

static void
agg_fill_hash_table(AggState *aggstate)
{TupleTableSlot *outerslot;ExprContext *tmpcontext = aggstate->tmpcontext;/** Process each outer-plan tuple, and then fetch the next one, until we* exhaust the outer plan.*/for (;;){outerslot = fetch_input_tuple(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);/* Advance the aggregates (or combine functions) */advance_aggregates(aggstate);/** Reset per-input-tuple context after each tuple, but note that the* hash lookups do this too*/ResetExprContext(aggstate->tmpcontext);}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);
}

第二步就是将哈希表进行遍历，依次处理结果并输出给上层算子。

static TupleTableSlot *
ExecAgg(PlanState *pstate)
{AggState   *node = castNode(AggState, pstate);TupleTableSlot *result = NULL;CHECK_FOR_INTERRUPTS();if (!node->agg_done){/* Dispatch based on strategy */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;}if (!TupIsNull(result))return result;}return NULL;
}

3.2 分组聚集

分组聚集使用的是排序的方式，其和朴素聚集的区别主要是增加了排序流程。代码也可以看，其排序和朴素聚集走了一样的代码：

static TupleTableSlot *
ExecAgg(PlanState *pstate)
{AggState   *node = castNode(AggState, pstate);TupleTableSlot *result = NULL;CHECK_FOR_INTERRUPTS();if (!node->agg_done){/* Dispatch based on strategy */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;}if (!TupIsNull(result))return result;}return NULL;
}