笔者最近由于工作需要开始调研 Apache Doris，通过阅读聚合函数代码切入 Apache Doris 内核，同时也秉承着开源的精神，开发了 array_agg 函数并贡献给社区。笔者通过这篇文章记录下对源码的一些理解，同时也方便后面的新人更快速地上手源码开发。
聚合函数，顾名思义，即对一组数据执行聚合计算并返回结果的函数，在统计分析过程中属于最常见的函数之一，最典型的聚合函数包括 count、min、max、sum 等。基于聚合函数可以实现对大量数据的汇总计算，以更简洁的形式呈现数据并支持数据可视化。
相较于单机数据库，由于所有数据都存储在同一台机器上、无需跨节点的网络数据传输，往往单机数据库的聚合函数执行效率更高，而分布式数据库由于数据分散存储于多个节点、并行执行计算时需要从多个节点汇集数据，带来了额外的网络传输和本地磁盘 IO 开销，且多副本机制和分片策略也进一步增加了计算的数据量和管理的复杂性。
为避免单点瓶颈同时减少网络 IO，往往需要使用多阶段的方式进行执行，因此 Apache Doris 实现了灵活的多阶段聚合机制，能够根据查询语句的特点为其选择适当的聚合方式，从而在执行时间和执行开销（如内存，IO 等）之间取得有效的平衡。
在 Apache Doris 中，主要聚合机制有如下几种：
一阶段聚合：Group By 仅包含分桶列，不同 Tablet 的数据在不同的分组中，因此不同 BE 可以独立并行计算；
两阶段聚合：Group By 包含非分桶列，同一个分组中的数据可能分布在多个 BE 上；
三阶段聚合：Count Distinct 包含 Group By（即 2 个两阶段聚合的组合）；
四阶段聚合：Count Distinct 不包含 Group By，通常采用 4 阶段聚合（1 个一阶段聚合和 1 个二阶段聚合的组合）
以如下查询为例，c1 是分桶列：
由于每个 BE 存储了若干个 Tablet ，每台 BE 只需要对当前节点上的 Tablet Set，分别进行 Hash Aggregate 即可，也称为 Final Hash Aggregate，随后对各个 BE 结果进行汇总。
同一个 BE 可以使用多个线程来同时进行 Final Hash Aggregate 以提高效率，这里为了便于更简单理解仅讨论单线程。
以如下查询为例，c2 不是分桶列：
对于上述查询，可以生成如下两阶段查询：
BE 对本节点上的 Tablet Set 进行第一次 Hash Aggregate，也称为 Pre Hash Aggregate；
BE 将 Pre Hash Aggregate 产生的结果按照完全相同的规则进行 Shuffle，其目的是将相同分组中的数据分发到同一台机器上；
BE 收到 Shuffle 数据后，再次进行 Hash Aggregate，也称为 Final Hash Aggregate；
对各个 BE 结果进行汇总
以如下查询为例：
对于上述查询，可以生成如下三阶段查询：
对 scan 分区按照 group by 和 distinct 字段（即 c2, c1）进行分组聚合；
将聚合后的结果按照 group by 和 distinct 字段进行重分区，然后对新的分区按照 group by 和 distinct 字段进行分组聚合；
对新的分区按照 group by 字段（即 c2）进行分组聚合。
以如下查询为例：
对于上述查询，可以生成如下四阶段查询：
对 scan 分区按照 distinct 字段进行分组聚合；
将聚合后的结果按照 distinct 字段进行重分区，然后对新的分区按照 distinct 字段进行分组聚合；
将 count distinct 转换为 count，对新的分区进行聚合；
对各分区的结果进行汇总聚合。
对于上述多阶段聚合中的第一阶段，其主要作用是通过预聚合减少重分区产生的网络 IO。如果在聚合时使用了高基数的维度作为分组维度（如 group by ID），则预聚合的效果可能会大打折扣。为此，Apache Doris 支持为此聚合阶段启用流式预聚合，在此模式下如果 Aggregate Pipeline 发现聚合操作产生的行数减少效果不及预期，则不再对新的 Block 进行聚合而是将其转换后放到队列中。而 Read Pipeline 也无需等待前者聚合完毕才开始执行，而是读取队列中 Block 进行处理，直到 Aggregate Pipeline 执行完毕后才读取 Hash 表中的聚合结果。
简单而言，聚合过程中如果 Hash Table 需要扩容但发现聚合效果不好（比如输入 1w 条数据，经聚合后还有 1w 个分组）就会跳过聚合，直接把每一行输入当作一个分组。即在第一阶段，对不同的数据分布，采用不同的处理方式能够进一步提高效率：
若数据聚合度高，那么在该阶段进行聚合，可以有效减少数据量，降低 Shuffle 时的网络开销；
若数据聚合度低，在该阶段进行聚合无法起到很好的聚合效果，同时伴随着额外的开销，例如哈希计算、额外的 Map、Set 存储空间，此时我们可以将该算子退化成一个简单的流式传输的算子，数据进入该算子后，不做处理直接输出。
由于聚合计算的执行过程和最终结果的生成方式不同，聚合函数可以分为需要 Finalize 的和不需要 Finalize 的这两类。需要 Finalize 的聚合函数（在计算过程中会产生中间结果，这些中间结果可能需要进一步的处理或合并才能得到最终的聚合结果）包括：
不需要 Finalize 的聚合函数（在计算过程中可以直接得到最终结果）包括：
对于非第一阶段的聚合算子来说，由于其读取到的是经过聚合后的数据，因此在执行时需要将聚合状态进行合并。而对于最后阶段的聚合算子，则需要在聚合计算后计算出最终的聚合结果。
在 Apache Doris 之中，定义了一个统一的聚合函数接口 IAggregateFunction。上文笔者提到的聚合函数，则都是作为抽象类 IAggregateFunction 的子类来实现的。该类中所有函数都是纯虚函数，需要子类自己实现，其中该接口最为核心的方法如下：
首先看聚合节点 Aggregetor 是如何调用 add_batch 函数：
这里依次遍历 AggFnEvaluator 并调用 execute_batch_add–>add_batch，而 add_batch 接口就是一行行的遍历列进行聚合计算：
构造函数：
argument_types_ 指的是函数的参数类型，比如函数 select avg(a), avg(b), c from test group by c，这里 a, b 分别是 UInt16 类型与 Decimal 类型，那么这个 avg(a) 与 avg(b) 的参数就不同。
聚合函数结果输出接口 将聚合计算的结果重新组织为列存：
AggregationNode 同样是遍历 Hash 表之中的结果，将 Key 列先组织成列存，然后调用 insert_result_info 函数将聚合计算的结果也转换为列存。以 avg 的实现为例：
这里就是调用 ConstAggregateDataPtr ，即 AggregateFunctionAvgData 的 result() 函数获取 avg 计算的结果添加到内存中。
这个接口是上面提到 IAggregateFunction 的辅助子类接口，主要实现获取 add/serialize/deserialize 函数地址的方法。
聚合函数需要大量使用 Doris 的核心接口 IColumn 类。IColumn 接口是所有数据存储类型的基类，其表达了所有数据的内存结构，其他带有具体数据类型的如：ColumnNullable、ColumnUInt8、ColumnString、ColumnVector、ColumnArray 等，都实现了对应的列接口，并且在子类之中具象实现了不同的内存布局。
在此以 avg 的实现为例（这里简化了对 Decimal 类型的处理)：
这里 columns 是一个二维数组，通过 columns[0] 可以取到第一列。这里只有涉及到一列，为什么 columns 是二维数组呢？因为处理多列的时候，也是通过对应的接口，而 array 就需要应用二维数组了。注意这里有一个强制的类型转换，column 已经转换为 ColVecType 类型了，这是模板派生出 IColumn 的子类。
然后通过 IColumn 子类实现的 get_data() 方法获取对应 row_num 行的数据，进行 add 函数调用就完成了一次聚合函数的计算了。由于这里是计算平均值，我们可以看到不仅仅累加了 value 还计算 count。
在上述 Pipeline 中，Aggregate Pipeline 负责使用 Hash 表（有 group by 的情况下）对输入数据进行聚合，Read Pipeline 负责读取聚合后的数据并发送至父算子，因此两者存在依赖关系，后者需要等待前者执行完成后才能开始执行。
在此仅以 BE 节点收到来自 FE 节点的 Execution Fragment 来分析。Aggregate 逻辑的入口位于 AggregationNode，处理流程根据是否启用流式预聚合而有所不同。但是不论哪种，都依赖于 AggregationNode 的实现。在介绍具体实现之前，我们先介绍下 AggregationNode。
AggregationNode 的一些重要成员如下，其中中文部分是笔者添加的注释：
Apache Doris 在聚合计算过程中使用了一种比较灵活的方式，在 AggregationNode 中事先声明了一个 executor 结构体，其中封装了多个 std::function，分别代表执行阶段可能需要调用到的函数。在 Prepare 阶段会使用 std::bind 将函数绑定到具体的实现上，根据是否开启 streaming pre-agg、是否存在 group by、是否存在 distinct 等条件来确定具体绑定什么函数。
这几个函数的大致调用关系过程可如下所示：
对应的相关绑定过程：
在没有启用流式预聚合的情况下，处理流程如下：
1. 调用 AggregationNode::init 函数进行初始化，包含如下处理逻辑：
调用 VExpr::create_expr_trees 函数创建 group by 相关的信息；
调用 AggFnEvaluator::create 函数创建聚合函数。在代码中，这里是一个 for 循环，即如果 SQL 中包含多个聚合函数，需要创建多次。
2. 调用 AggregationNode::prepare 函数执行运行前的准备，包含如下处理逻辑：
调用 ExecNode::prepare 函数为父类成员执行运行前的准备；
对 group by 表达式调用 VExpr::prepare 函数执行运行前的准备；
计算聚合函数需要的状态空间大小及各聚合函数的偏移，这些偏移量后续取地址的时候会用到
AggregationNode::prepare_profile 根据当前聚合类型及是否涉及 group by 参数 bind 对应的处理函数，分配逻辑如下：
如果当前聚合包含 group by 参数：
如果当前聚合需要 merge 聚合状态（多阶段聚合），则使用 AggregationNode::_merge_with_serialized_key 函数用于处理输入 block（下称 execute 函数），否则使用 AggregationNode::_execute_with_serialized_key 函数。如果是多阶段聚合多个 AggregationNode 会分别绑定_merge_with_serialized_key 和 _execute_with_serialized_key。
如果当前聚合需要对聚合结果执行 finalize，则使用 AggregationNode::_get_with_serialized_key_result 函数用于读取聚合结果（下称 get_result 函数），否则使用AggregationNode::_serialize_with_serialized_key_result 函数。
如果当前聚合不包含 group by 参数：
如果当前聚合需要 merge 聚合状态，则使用 AggregationNode::_merge_without_key 函数用于处理输入 block（下称execute函数），否则使用 AggregationNode::_execute_without_key 函数。
如果当前聚合需要对聚合结果执行 finalize，则使用 AggregationNode::_get_with_serialized_key_result 函数用于读取聚合结果（下称 get_result 函数），否则使用 AggregationNode::_serialize_with_serialized_key_result 函数。
如果当前聚合包含 group by 参数，则需要根据参数类型分配对应的 hash 方法：_init_hash_method
对 block 数据的聚合逻辑较为简单，以包含 group by 参数的情况为例，聚合流程如下：
调用 AggregationNode::_emplace_into_hash_table 函数创建具体的聚合方法类，并获取 Hash 表中对应行的聚合状态。
如果当前聚合处理的是原始的行数据，则调用 AggFnEvaluator::execute_batch_add 函数进行聚合处理。
如果当前聚合需要 merge 聚合状态，则首先需要对聚合状态中的结果进行反序列化，然后调用 IAggregateFunctionHelper::merge_vec 函数对当前聚合状态进行合并。
对于 hash 分组效果不佳的场景，会启用流式预聚合，处理流程如下：
调用 AggregationNode::init 函数进行初始化；
调用AggregationNode::prepare函数执行运行前的准备；
调用 AggregationNode::do_pre_agg 函数对输入 block 进行聚合，该函数会调用 _pre_agg_with_serialized_key 函数进行实际的聚合操作。如果在处理过程中 hash 扩容达到阈值，则跳过聚合，直接把每一行输入当作一个分组，即调用 streaming_agg_serialize_to_column，否则还是使用朴素的方法 AggFnEvaluator::execute_batch_add；
调用 AggregationNode::pull 函数取出聚合结果，发送至父算子进行处理；
调用 AggregationNode::release_resource 函数释放资源。
感兴趣的读者可以自行阅读流式聚合相关的源码，可以给 streaming_agg_serialize_to_column 加断点进行 debug，触发方法如下：
TPC-H 准备 3G 数据，方法见 https://doris.apache.org/zh-CN/docs/1.2/benchmark/tpch/
执行 SQL select count() from (select map_agg(o_orderstatus,o_clerk) from orders group by o_custkey, o_orderdate) a
语法：MAP_AGG(expr1, expr2)
功能：返回一个 map，由 expr1 作为键、expr2 作为对应的值。
Step 1: 维护 FunctionSet.java（https://github.com/apache/doris/blob/master/fe/fe-core/src/main/java/org/apache/doris/catalog/FunctionSet.java）
FE 通过 initAggregateBuiltins 来描述聚合函数，所有的聚合函数都会注册在 FunctionSet 中。初始化阶段在FunctionSet.java （https://github.com/apache/doris/blob/master/fe/fe-core/src/main/java/org/apache/doris/catalog/FunctionSet.java）的 initAggregateBuiltins 中增加对应的 AggregateFunction.createBuiltin 函数即可。
以上代码的理解思路如下：
如果 map_agg 的 key 不是 josn blob 字段( if (!Type.JSONB.equals(t)) )，则先找到 map_agg 相关函数 ( for (Type valueType : Type.getMapSubTypes())) 。
通过 addBuiltin 初始化对应 MAP_AGG 函数，value 类型是传进来的 valueType，中间状态变量是 Type.VARCHAR。
找到 array 相关函数( for (Type v : Type.getArraySubTypes()))，通过 addBuiltin 初始化对应 MAP_AGG 函数， value 类型是 ArrayType，中间状态变量是 MapType。
Step 2：维护 AggregateFunction.java（https://github.com/apache/doris/blob/master/fe/fe-core/src/main/java/org/apache/doris/catalog/AggregateFunction.java）
在 AggregateFunction.java (https://github.com/apache/doris/blob/master/fe/fe-core/src/main/java/org/apache/doris/catalog/AggregateFunction.java)文件中，注册 FunctionSet.MAP_AGG，具体如下：
这一步是为了让 AggregateFunctionSimpleFactory 可以根据函数名找到对应的函数，函数的创建通过 factory.register_function_both 实现，相关的改动可以在 aggregate_function_map.cc (https://github.com/xingyingone/doris/blob/b41fcbb7834bf89f9744d351b1cfb9ac2485008b/be/src/vec/aggregate_functions/aggregate_function_map.cpp) 中 grep register_aggregate_function_map_agg 看到，比较简单，在此不再赘述。
重点是如何描述中间结果以及 AggregateFunctionMapAgg 如何实现 IAggregateFunction的核心接口。
Step 1：转换类型
由于我们最终结果需要返回一系列 map，所以输出类型为 DataTypeMap：
由于默认的中间状态是 string 类型，如果是 string，需要处理比较复杂的序列化/反序列化操作。
所以在 AggregateFunctionMapAgg 重新了序列化/反序列化的中间类型：
Step 2：聚合操作
代码中需要将每行的数据取出来进行对应的聚合计算，具体是通重写 add 函数来实现的：
具体的代码实现如下：
Step 3：序列化/反序列化
由于中间传输的是 ColumnMap 类型，所以只需进行数据拷贝即可
Step 4：输出结果
insert_result_into 表示最终的返回，所以里面转换的类型要跟 return_type 里面的一致，所以可以看到我们将类型转换为 ColumnMap 进行处理。
Step 5：维护测试用例及文档
这块比较简单，可以参考官方文档 https://doris.apache.org/zh-CN/community/developer-guide/regression-testing/
语法：ARRAY_AGG(col)
功能：将一列中的值（包括空值 null）串联成一个数组，可以用于多行转一行（行转列）。
需要注意点：
实验 SQL 如下：
group by + 多阶段聚合
group by + 一阶段聚合
group by + 多阶段聚合
注意点：
在学习和掌握 Apache Doris 的过程中，作为 OLAP 新人的笔者遇到了很多疑惑点。好在 Apache Doris 不仅功能强大，社区更是十分活跃，社区技术大佬们对于新人的问题也特别热心，不厌其烦帮我们新人们答疑解惑，这无疑为笔者在调研过程中增加了不少信心，在此由衷地感谢社区大佬 @yiguolei @mrhhsg。也期待未来有更多的小伙伴可以参与到社区当中来，一同学习与成长。
隐形(邢颖) 网易资深数据库内核工程师，毕业至今一直从事数据库内核开发工作，目前主要参与 MySQL 与 Apache Doris 的开发维护和业务支持工作。
作为 MySQL 内核贡献者，为 MySQL 上报了 50 多个 Bug 及优化项，多个提交被合入 MySQL 8.0 版本。从 2023 年起加入 Apache Doris 社区，Apache Doris Active Contributor，已为社区提交并合入数十个 Commits。
[1]https://zhuanlan.zhihu.com/p/614555403
[2]https://www.slidestalk.com/doris.apache/Doris22141
[3]https://github.com/apache/doris/blob/master/be/src/vec/aggregate_functions/aggregate_function_avg.h
[4]https://doris.apache.org/zh-CN/docs/1.2/benchmark/tpch/
[5]https://github.com/apache/doris/pull/22043
[6]https://github.com/apache/doris/blob/master/fe/fe-core/src/main/java/org/apache/doris/catalog/FunctionSet.java
[7]https://github.com/apache/doris/blob/master/fe/fe-core/src/main/java/org/apache/doris/catalog/AggregateFunction.java
[8]https://github.com/apache/doris/blob/master/fe/fe-core/src/main/java/org/apache/doris/analysis/FunctionCallExpr.java
[9]https://github.com/xingyingone/doris/blob/b41fcbb7834bf89f9744d351b1cfb9ac2485008b/be/src/vec/aggregate_functions/aggregate_function_map.cpp
[10]https://doris.apache.org/zh-CN/community/developer-guide/regression-testing/
[11]https://github.com/apache/doris/pull/22043/files
[12]https://github.com/apache/doris/pull/23474/files
[13]https://bugs.mysql.com/search.php?cmd=display&status=All&severity=all&reporter=15843759
END
AI 领域大佬欢聚一堂 跳起科目三
这里有最新开源资讯、软件更新、技术干货等内容
微信扫码关注该文公众号作者
https://redian.news/wxnews/719868