PostgreSQL技术内幕：PG聚合算子实现分析
PostgreSQL技术内幕：PG聚合算子实现分析
PostgreSQL的聚合算子在数据分析、报告生成和统计计算中扮演着重要角色。本文将通过对PostgreSQL聚合算子的源码进行解析，深入了解其实现机制。
概念说明
聚合函数是用于对一组值进行计算并返回单个结果的函数。它们通常与GROUP BY子句结合使用，将数据分组后对每组进行计算。例如，COUNT函数用于计算行数或非NULL值的数量，SUM函数用于计算数值列的总和，AVG函数用于计算数值列的平均值，MAX和MIN函数分别用于计算数值列中的最大值和最小值。
常见的聚合算子实现分为两种，一种是基于哈希表的实现（哈希聚集），另外一种是基于排序的实现（分组聚集），这两种聚集方式一般来说都是对于带有group by的查询来说的。因为对于没有group by的查询来说，只需要扫描一遍表并对相应元组进行累计操作即可（也被称为朴素聚集）。
聚合操作整体可以分为三个步骤：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
aggmtransfn | 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)
朴素聚集
因为sum函数并不需要最后的最终结果生成，只需要转换函数即可，所以下面会以求平均值作为例子，扫描数据并生成中间值就是将所有元组进行累加，最终结果生成就是对累加结果进行sum/count。
上图展示了无并行的朴素聚集的流程，但由于是单进程实现而聚集又是一个cpu密集算子，其不能利用cpu多核的优势，从而导致性能的瓶颈。而解决方案就是并行进行扫描和初始聚集，由leader节点进行收集和后处理。
Group by聚集
哈希聚集
非并行哈希函数整体包含两个阶段：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;
}
分组聚集
分组聚集使用的是排序的方式，其和朴素聚集的区别主要是增加了排序流程。代码也可以看，其排序和朴素聚集走了一样的代码：
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;
}