dorisdbphp技巧_Apache Doris 聚合函数源码阅读与解析

文章目录 [+]

聚合函数，顾名思义，即对一组数据实行聚合打算并返回结果的函数，在统计剖析过程中属于最常见的函数之一，最范例的聚合函数包括 count、min、max、sum 等。
基于聚合函数可以实现对大量数据的汇总打算，以更简洁的形式呈现数据并支持数据可视化。

相较于单机数据库，由于所有数据都存储在同一台机器上、无需跨节点的网络数据传输，每每单机数据库的聚合函数实行效率更高，而分布式数据库由于数据分散存储于多个节点、并行实行打算时须要从多个节点搜集数据，带来了额外的网络传输和本地磁盘 IO 开销，且多副本机制和分片策略也进一步增加了打算的数据量和管理的繁芜性。

dorisdbphp技巧_Apache Doris 聚合函数源码阅读与解析｜源码解读系列

为避免单点瓶颈同时减少网络 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 是分桶列：

SELECT count(c1) FROM t1 GROUP BY c1

由于每个 BE 存储了多少个 Tablet ，每台 BE 只须要对当前节点上的 Tablet Set，分别进行 Hash Aggregate 即可，也称为 Final Hash Aggregate，随后对各个 BE 结果进行汇总。

同一个 BE 可以利用多个线程来同时进行 Final Hash Aggregate 以提高效率，这里为了便于更大略理解仅谈论单线程。

两阶段聚合

以如下查询为例，c2 不是分桶列：

SELECT c2, count(c1) FROM t1 GROUP BY c2

对付上述查询，可以天生如下两阶段查询：

对 scan 分区按照 group by 字段（即 c2）进行分组聚合；将聚合后的结果按照 group by 字段进行重分区，然后对新的分区按照 group by 字段进行分组聚合。

详细流程如下：

BE 对本节点上的 Tablet Set 进行第一次 Hash Aggregate，也称为 Pre Hash Aggregate；BE 将 Pre Hash Aggregate 产生的结果按照完备相同的规则进行 Shuffle，其目的是将相同分组中的数据分发到同一台机器上；BE 收到 Shuffle 数据后，再次进行 Hash Aggregate，也称为 Final Hash Aggregate；对各个 BE 结果进行汇总

两阶段聚合

三阶段聚合

以如下查询为例：

SELECT count(distinct c1) FROM t1 GROUP BY c2

对付上述查询，可以天生如下三阶段查询：

对 scan 分区按照 group by 和 distinct 字段（即 c2, c1）进行分组聚合；将聚合后的结果按照 group by 和 distinct 字段进行重分区，然后对新的分区按照 group by 和 distinct 字段进行分组聚合；对新的分区按照 group by 字段（即 c2）进行分组聚合。

三阶段聚合

四阶段聚合

以如下查询为例：

SELECT count(distinct c1), sum(c2) FROM t1

对付上述查询，可以天生如下四阶段查询：

对 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 存储空间，此时我们可以将该算子退化成一个大略的流式传输的算子，数据进入该算子后，不做处理直接输出。
Merge & Finalize

由于聚合打算的实行过程和终极结果的天生办法不同，聚合函数可以分为须要 Finalize 的和不须要 Finalize 的这两类。
须要 Finalize 的聚合函数（在打算过程中会产生中间结果，这些中间结果可能须要进一步的处理或合并才能得到终极的聚合结果）包括：

AVG：打算均匀值时须要将所有值相加再除以总数，因此须要 Finalize 操作来完成这个过程；STDDEV：打算标准差时须要先打算方差再开方得到标准差，这个过程须要多次遍历数据集，因此须要 Finalize 操作来完成；VAR_POP、VAR_SAMP：打算方差时须要用到所有数据的平方和，这个过程须要多次遍历数据集，因此须要 Finalize 操作来完成。

不须要 Finalize 的聚合函数（在打算过程中可以直接得到终极结果）包括：

COUNT：只须要统计数据集中的行数，不须要进行其他操作；SUM、MIN、MAX：对数据集进行聚合时，这些函数只须要遍历一次数据集，因此不须要进行 Finalize 操作。

对付非第一阶段的聚合算子来说，由于其读取到的是经由聚合后的数据，因此在实行时须要将聚合状态进行合并。
而对付末了阶段的聚合算子，则须要在聚合打算后打算出终极的聚合结果。

聚合函数核心接口IAggregateFunction接口

在 Apache Doris 之中，定义了一个统一的聚合函数接口 IAggregateFunction。
上文笔者提到的聚合函数，则都是作为抽象类 IAggregateFunction 的子类来实现的。
该类中所有函数都是纯虚函数，须要子类自己实现，个中该接口最为核心的方法如下：

add函数：最为核心的调用接口，将对应 AggregateDataPtr 指针之中数据取出，与列 columns 中的第 row_num 的数据进行对应的聚合打算。
(这里可以看到Doris是一个纯粹的列式存储数据库，所有的操作都是基于列的数据构造。
）merge函数：将两个聚合结果进行合并的函数，常日用在并发实行聚合函数的过程之中，须要将对应的聚合结果进行合并。
serialize 函数与 deserialize 函数：序列化与反序列化的函数，常日用于 Spill-to-Disk 或 BE 节点之间传输中间结果的。
add_batch 函数：虽然它仅仅实现了一个 for 循环调用 add 函数，但通过这样的办法来减少虚函数的调用次数，并且增加了编译器内联的概率。
（虚函数的调用须要一次访存指令，一次查表，终极才能定位到须要调用的函数上，这在传统的火山模型的实现上会带来极大的CPU开销。
）

首先看聚合节点 Aggregetor 是如何调用 add_batch 函数：

for (int i = 0; i < _aggregate_evaluators.size(); ++i) { RETURN_IF_ERROR(_aggregate_evaluators[i]->execute_batch_add( block, _offsets_of_aggregate_states[i], _places.data(), _agg_arena_pool.get()));}

这里依次遍历 AggFnEvaluator 并调用 execute_batch_add-->add_batch，而 add_batch 接口便是一行行的遍历列进行聚合打算：

void add_batch(size_t batch_size, AggregateDataPtr places, size_t place_offset, const IColumn columns, Arena arena, bool agg_many) const override{ for (size_t i = 0; i < batch_size; ++i) { assert_cast<const Derived>(this)->add(places[i] + place_offset, columns, i, arena); } }

布局函数：

IAggregateFunction(const DataTypes& argument_types_) : argument_types(argument_types_) {}

argument_types_ 指的是函数的参数类型，比如函数 select avg(a), avg(b), c from test group by c, 这里 a, b 分别是 UInt16 类型与 Decimal 类型，那么这个 avg(a) 与 avg(b) 的参数就不同。

聚合函数结果输出接口将聚合打算的结果重新组织为列存：

/// Inserts results into a column.virtual void insert_result_into(ConstAggregateDataPtr __restrict place, IColumn& to) const = 0;首先看聚合节点 AggregationNode 是如何调用 insert_result_into 函数的：for (size_t i = 0; i < _aggregate_evaluators.size(); ++i) { _aggregate_evaluators[i]->insert_result_info( mapped + _offsets_of_aggregate_states[i], value_columns[i].get());}void AggFnEvaluator::insert_result_info(AggregateDataPtr place, IColumn column) { _function->insert_result_into(place, column);}

AggregationNode 同样是遍历 Hash表之中的结果，将 Key 列先组织成列存，然后调用 insert_result_info 函数将聚合打算的结果也转换为列存。
以 avg 的实现为例：

void insert_result_into(ConstAggregateDataPtr __restrict place, IColumn& to) const override { auto& column = assert_cast<ColVecResult&>(to); column.get_data().push_back(this->data(place).template result<ResultType>()); } template <typename ResultT> AggregateFunctionAvgData::ResultT result() const { if constexpr (std::is_floating_point_v<ResultT>) { if constexpr (std::numeric_limits<ResultT>::is_iec559) { return static_cast<ResultT>(sum) / count; /// allow division by zero } }// https://github.com/apache/doris/blob/master/be/src/vec/aggregate_functions/aggregate_function_avg.h

这里便是调用 ConstAggregateDataPtr ，即 AggregateFunctionAvgData 的 result() 函数获取 avg 打算的结果添加到内存中。

IAggregateFunctionDataHelper 接口

这个接口是上面提到 IAggregateFunction 的赞助子类接口，紧张实现获取 add/serialize/deserialize 函数地址的方法。

抽象类 IColumn

聚合函数须要大量利用 Doris 的核心接口 IColumn 类。
IColumn 接口是所有数据存储类型的基类，其表达了所有数据的内存构造，其他带有详细数据类型的如：ColumnNullable、ColumnUInt8、ColumnString、ColumnVector、ColumnArray 等，都实现了对应的列接口，并且在子类之中具象实现了不同的内存布局。

在此以 avg 的实现为例（这里简化了对 Decimal 类型的处理)：

void add(AggregateDataPtr __restrict place, const IColumn columns, size_t row_num, Arena) const override { const auto& column = assert_cast<const ColVecType&>(columns[0]); this->data(place).sum += column.get_data()[row_num].value; ++this->data(place).count;}// https://github.com/apache/doris/blob/master/be/src/vec/aggregate_functions/aggregate_function_avg.h

这里 columns 是一个二维数组，通过 columns[0] 可以取到第一列。
这里只有涉及到一列，为什么 columns 是二维数组呢？由于处理多列的时候，也是通过对应的接口，而 array 就须要运用二维数组了。
把稳这里有一个逼迫的类型转换，column 已经转换为 ColVecType 类型了，这是模板派生出 IColumn 的子类。

然后通过 IColumn 子类实现的 get_data() 方法获取对应 row_num 行的数据，进行 add 函数调用就完成了一次聚合函数的打算了。
由于这里是打算均匀值，我们可以看到不仅仅累加了 value 还打算 count。

聚合函数主体流程

在实行时，对应的 Fragment 会被转换为如下 Pipeline：

聚合函数主体流程

在上述 Pipeline 中，Aggregate Pipeline 卖力利用 Hash 表（有 group by 的情形下）对输入数据进行聚合，Read Pipeline 卖力读取聚合后的数据并发送至父算子，因此两者存在依赖关系，后者须要等待前者实行完成后才能开始实行。

在此仅以 BE 节点收到来自 FE 节点的 Execution Fragment 来剖析。
Aggregate 逻辑的入口位于 AggregationNode，处理流程根据是否启用流式预聚合而有所不同。
但是不论哪种，都依赖于 AggregationNode 的实现。
在先容详细实现之前，我们先先容下 AggregationNode。

构造体先容

AggregationNode 的一些主要成员如下，个中中文部分是笔者添加的注释：

class AggregationNode : public ::doris::ExecNode{ Status init(const TPlanNode& tnode, RuntimeState state = nullptr) override; Status prepare_profile(RuntimeState state); Status prepare(RuntimeState state) override; //SQL中包含的聚合函数的数组 std::vector<AggFnEvaluator> _aggregate_evaluators; //是否须要finalize，前文有提到判断准则 bool _needs_finalize; //是否须要merge bool _is_merge; //是否是第一阶段聚合 bool _is_first_phase; //用来bind实行阶段须要用到的函数 executor _executor; //存放聚合过程中的数据 AggregatedDataVariantsUPtr _agg_data; //取出聚合结果，发送至父算子进行处理 //进行读取操作，会利用get_result函数进行处理 Status pull(doris::RuntimeState state, vectorized::Block output_block, bool eos) override; //对输入block进行聚合，该步骤会利用前面分配的execute函数进行处理。 Status sink(doris::RuntimeState state, vectorized::Block input_block, bool eos) override; //读取聚合结果，该函数终极会调用AggregationNode::pull函数进行读取操作 Status get_next(RuntimeState state, Block block, bool eos) override; //实行阶段须要用到的函数 Status _get_without_key_result(RuntimeState state, Block block, bool eos); Status _serialize_without_key(RuntimeState state, Block block, bool eos); Status _execute_without_key(Block block); Status _merge_without_key(Block block); Status _get_with_serialized_key_result(RuntimeState state, Block block, bool eos); Status _get_result_with_serialized_key_non_spill(RuntimeState state, Block block, bool eos); Status _execute_with_serialized_key(Block block); Status _merge_with_serialized_key(Block block);}

Apache Doris 在聚合打算过程中利用了一种比较灵巧的办法，在 AggregationNode 中事先声明了一个executor 构造体，个中封装了多个 std::function，分别代表实行阶段可能须要调用到的函数。
在 Prepare 阶段会利用 std::bind 将函数绑定到详细的实现上，根据是否开启 streaming pre-agg、是否存在 group by、是否存在 distinct 等条件来确定详细绑定什么函数。

struct AggregationNode::executor { vectorized_execute execute; vectorized_pre_agg pre_agg; vectorized_get_result get_result; vectorized_closer close; vectorized_update_memusage update_memusage;};

这几个函数的大致调用关系过程可如下所示：

函数调用关系

对应的干系绑定过程：

干系绑定过程

普通聚合

在没有启用流式预聚合的情形下，处理流程如下：

调用 AggregationNode::init 函数进行初始化，包含如下处理逻辑：调用 VExpr::create_expr_trees 函数创建 group by 干系的信息；调用 AggFnEvaluator::create 函数创建聚合函数。
在代码中，这里是一个 for 循环，即如果 SQL 中包含多个聚合函数，须要创建多次。
调用 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

调用 AggregationNode::sink 函数对输入 Block 进行聚合，该步骤会利用前面分配的 execute 函数进行处理。
调用 AggregationNode::get_next 函数读取聚合结果，该函数终极会调用 AggregationNode::pull 函数进行读取操作，后者会利用前面分配的 get_result 函数进行处理。
调用 AggregationNode::release_resource 函数开释资源，该函数会调用 _executor.close()。

对 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 为例先容添加聚合函数的流程。
以下内容仅为笔者个人的思考，感兴趣的读者可以详细参考 https://github.com/apache/doris/pull/22043。

map_agg 利用先容

语法：MAP_AGG(expr1, expr2)

功能：返回一个 map，由 expr1 作为键、expr2 作为对应的值。

在 FE 创建函数署名

Step 1: 掩护 FunctionSet.java FE 通过 initAggregateBuiltins 来描述聚合函数，所有的聚合函数都会注册在 FunctionSet 中。
初始化阶段在 FunctionSet.java的 initAggregateBuiltins 中增加对应的 AggregateFunction.createBuiltin 函数即可。

if (!Type.JSONB.equals(t)) { for (Type valueType : Type.getMapSubTypes()) { addBuiltin(AggregateFunction.createBuiltin(MAP_AGG, Lists.newArrayList(t, valueType), new MapType(t, valueType), Type.VARCHAR, "", "", "", "", "", null, "", true, true, false, true)); } for (Type v : Type.getArraySubTypes()) { addBuiltin(AggregateFunction.createBuiltin(MAP_AGG, Lists.newArrayList(t, new ArrayType(v)), new MapType(t, new ArrayType(v)), new MapType(t, new ArrayType(v)), "", "", "", "", "", null, "", true, true, false, true)); } }

以上代码的理解思路如下：

如果 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

在 AggregateFunction.java 文件中，注册 FunctionSet.MAP_AGG，详细如下：

public static ImmutableSet<String> NOT_NULLABLE_AGGREGATE_FUNCTION_NAME_SET = ImmutableSet.of("row_number", "rank", "dense_rank", "multi_distinct_count", "multi_distinct_sum", FunctionSet.HLL_UNION_AGG, FunctionSet.HLL_UNION, FunctionSet.HLL_RAW_AGG, FunctionSet.BITMAP_UNION, FunctionSet.BITMAP_INTERSECT, FunctionSet.ORTHOGONAL_BITMAP_INTERSECT, FunctionSet.ORTHOGONAL_BITMAP_INTERSECT_COUNT, FunctionSet.ORTHOGONAL_BITMAP_EXPR_CALCULATE_COUNT, FunctionSet.ORTHOGONAL_BITMAP_EXPR_CALCULATE, FunctionSet.INTERSECT_COUNT, FunctionSet.ORTHOGONAL_BITMAP_UNION_COUNT, FunctionSet.COUNT, "approx_count_distinct", "ndv", FunctionSet.BITMAP_UNION_INT, FunctionSet.BITMAP_UNION_COUNT, "ndv_no_finalize", FunctionSet.WINDOW_FUNNEL, FunctionSet.RETENTION, FunctionSet.SEQUENCE_MATCH, FunctionSet.SEQUENCE_COUNT, FunctionSet.MAP_AGG);

Step 3: 掩护 FunctionCallExpr.java 在 FunctionCallExpr.java 中根据 argumentt 逼迫设置类型，防止丢失 decimal 类型的 scale。

if (fnName.getFunction().equalsIgnoreCase("map_agg")) { fn.setReturnType(new MapType(getChild(0).type, getChild(1).type)); } 在 BE 中注册函数

这一步是为了让 AggregateFunctionSimpleFactory 可以根据函数名找到对应的函数，函数的创建通过 factory.register_function_both 实现，干系的改动可以在 aggregate_function_map.cc 中 grep register_aggregate_function_map_agg 看到，比较大略，在此不再赘述。

在 BE 实现函数的打算逻辑

重点是如何描述中间结果以及 AggregateFunctionMapAgg 如何实现 IAggregateFunction的核心接口。

Step 1：转换类型

由于我们终极结果须要返回一系列 map，以是输出类型为 DataTypeMap：

DataTypePtr get_return_type() const override { /// keys and values column of `ColumnMap` are always nullable. return std::make_shared<DataTypeMap>(make_nullable(argument_types[0]), make_nullable(argument_types[1]));}

由于默认的中间状态是 string 类型，如果是 string，须要处理比较繁芜的序列化/反序列化操作。

IAggregateFunction::get_serialized_type(){return std::make_shared<DataTypeString>();}

以是在 AggregateFunctionMapAgg 重新了序列化/反序列化的中间类型：

[[nodiscard]] MutableColumnPtr create_serialize_column() const override { return get_return_type()->create_column();}[[nodiscard]] DataTypePtr get_serialized_type() const override { return get_return_type(); }

Step 2：聚合操作

代码中须要将每行的数据取出来进行对应的聚合打算，详细是通重写 add 函数来实现的：

这里表示将第 row_num 行的数据丢给 AggregateFunctionMapAggData 来实行，这里可以看出来须要对 nullable 和非 nullable 的分开处理。
在 AggregateFunctionMapAggData 中，将 key 以及 value 分别存储在 _key_column 和 _value_column。
由于 key 不为 NULL，以是实行了 remove_nullable；由于 value 许可为 NULL，这里实行了 make_nullable，并通过 _map 来过滤了重复的 key。

详细的代码实现如下：

void AggregateFunctionMapAgg::add(AggregateDataPtr __restrict place, const IColumn columns, size_t row_num, Arena arena) const override { if (columns[0]->is_nullable()) { auto& nullable_col = assert_cast<const ColumnNullable&>(columns[0]); auto& nullable_map = nullable_col.get_null_map_data(); if (nullable_map[row_num]) { return; } Field value; columns[1]->get(row_num, value); this->data(place).add( assert_cast<const KeyColumnType&>(nullable_col.get_nested_column()) .get_data_at(row_num), value); } else { Field value; columns[1]->get(row_num, value); this->data(place).add( assert_cast<const KeyColumnType&>(columns[0]).get_data_at(row_num), value); }}AggregateFunctionMapAggData::add(const StringRef& key, const Field& value){ DCHECK(key.data != nullptr); if (UNLIKELY(_map.find(key) != _map.end())) { return; } ArenaKeyHolder key_holder {key, _arena}; if (key.size > 0) { key_holder_persist_key(key_holder); } _map.emplace(key_holder.key, _key_column->size()); _key_column->insert_data(key_holder.key.data, key_holder.key.size); _value_column->insert(value);}

Step 3：序列化/反序列化

由于中间传输的是 ColumnMap 类型，以是只需进行数据拷贝即可

void deserialize_from_column(AggregateDataPtr places, const IColumn& column, Arena arena, size_t num_rows) const override { auto& col = assert_cast<const ColumnMap&>(column); auto data = &(this->data(places)); for (size_t i = 0; i != num_rows; ++i) { auto map = doris::vectorized::get<Map>(col[i]); data->add(map[0], map[1]); }}void serialize_to_column(const std::vector<AggregateDataPtr>& places, size_t offset, MutableColumnPtr& dst, const size_t num_rows) const override { for (size_t i = 0; i != num_rows; ++i) { Data& data_ = this->data(places[i] + offset); data_.insert_result_into(dst); }}

Step 4：输出结果

insert_result_into 表示终极的返回，以是里面转换的类型要跟 return_type 里面的同等，以是可以看到我们将类型转换为 ColumnMap 进行处理。

void AggregateFunctionMapAgg::insert_result_into(ConstAggregateDataPtr __restrict place, IColumn& to) const override { this->data(place).insert_result_into(to);}void AggregateFunctionMapAggData::insert_result_into(IColumn& to) const { auto& dst = assert_cast<ColumnMap&>(to); size_t num_rows = _key_column->size(); auto& offsets = dst.get_offsets(); auto& dst_key_column = assert_cast<ColumnNullable&>(dst.get_keys()); dst_key_column.get_null_map_data().resize_fill(dst_key_column.get_null_map_data().size() + num_rows); dst_key_column.get_nested_column().insert_range_from(_key_column, 0, num_rows); dst.get_values().insert_range_from(_value_column, 0, num_rows); if (offsets.size() == 0) { offsets.push_back(num_rows); } else { offsets.push_back(offsets.back() + num_rows); }}

Step 5：掩护测试用例及文档

这块比较大略，可以参考官方文档 https://doris.apache.org/zh-CN/community/developer-guide/regression-testing/

array_agg 源码解析

笔者通过阅读 mag_agg 源码以及社区大佬 @mrhhsg 的答疑解惑，为 Apache Doris 增加了 array_agg 函数支持。
下文笔者将从 SQL 实行的角度阐述上文提到的函数实行流程及调用栈，详细代码可以阅读 https://github.com/apache/doris/pull/23474/files。

array_agg 利用先容

语法：ARRAY_AGG(col)

功能：将一列中的值（包括空值 null）串联成一个数组，可以用于多行转一行（行转列）。

须要把稳点：

数组中元素不担保顺序；返回转换天生的数组，数组中的元素类型与 col类型同等；须要显示NULL

实验 SQL 如下：

CREATE TABLE `test_array_agg` ( `id` int(11) NOT NULL, `label_name` varchar(32) default null, `value_field` string default null, ) ENGINE=OLAP DUPLICATE KEY(`id`) COMMENT 'OLAP' DISTRIBUTED BY HASH(`id`) BUCKETS 1 PROPERTIES ( "replication_allocation" = "tag.location.default: 1", "storage_format" = "V2", "light_schema_change" = "true", "disable_auto_compaction" = "false", "enable_single_replica_compaction" = "false" ); insert into `test_array_agg` values (1, "alex",NULL), (1, "LB", "V1_2"), (1, "LC", "V1_3"), (2, "LA", "V2_1"), (2, "LB", "V2_2"), (2, "LC", "V2_3"), (3, "LA", "V3_1"), (3, NULL, NULL), (3, "LC", "V3_3"), (4, "LA", "V4_1"), (4, "LB", "V4_2"), (4, "LC", "V4_3"), (5, "LA", "V5_1"), (5, "LB", "V5_2"), (5, "LC", "V5_3");实行流程

group by + 多阶段聚合

mysql> SELECT label_name, array_agg(label_name) FROM test_array_agg GROUP BY label_name;+------------+--------------------------------+| label_name | array_agg(`label_name`) |+------------+--------------------------------+| LC | ["LC", "LC", "LC", "LC", "LC"] || NULL | [NULL] || alex | ["alex"] || LB | ["LB", "LB", "LB", "LB"] || LA | ["LA", "LA", "LA", "LA"] |+------------+--------------------------------+5 rows in set (11.55 sec)#实行AggregationNode::_pre_agg_with_serialized_key-->add(实行15次，每次处理一行) +AggregationNode::_merge_with_serialized_key->deserialize_and_merge_vec(实行5次，每次merge一个分组)#取结果_serialize_with_serialized_key_result-->serialize_to_column 实行一次，处理5个分组_get_with_serialized_key_result-->insert_result_info 5次，每次处理一个分组

group by + 一阶段聚合

mysql> SELECT id, array_agg(label_name) FROM test_array_agg GROUP BY id;+------+-------------------------+| id | array_agg(`label_name`) |+------+-------------------------+| 1 | ["LC", "LB", "alex"] || 2 | ["LC", "LB", "LA"] || 3 | ["LC", NULL, "LA"] || 4 | ["LC", "LB", "LA"] || 5 | ["LC", "LB", "LA"] |+------+-------------------------+5 rows in set (20.12 sec)#实行AggregationNode::_execute_with_serialized_key-->add(实行15次，每次处理一行)#取结果 _get_with_serialized_key_result-->insert_result_info 一次循环，遍历处理5个分组

无group by + 多阶段聚合

mysql> SELECT array_agg(label_name) FROM test_array_agg;+----------------------------------------------------------------------------------------------+| array_agg(`label_name`) |+----------------------------------------------------------------------------------------------+| ["LC", "LB", "alex", "LC", "LB", "LA", "LC", NULL, "LA", "LC", "LB", "LA", "LC", "LB", "LA"] |+----------------------------------------------------------------------------------------------+1 row in set (1 min 21.01 sec) #实行 AggregationNode::_execute_without_key-->add(实行15次，每次处理一行) AggregationNode::_merge_without_key-->deserialize_and_merge_from_column(实行一次，只有一个分组，这个分组有15个元素) #取结果 AggregationNode::_serialize_without_key-->serialize_without_key_to_column AggregationNode::_get_without_key_result-->AggregateFunctionCollect::insert_result_into (实行一次，只有一个分组，这个分组有15个元素)函数调用栈

AggregationNode::init|--> //初始化_aggregate_evaluators|--> _aggregate_evaluators.reserve(tnode.agg_node.aggregate_functions.size());|--> //begin loop|--> for (int i = 0; i < tnode.agg_node.aggregate_functions.size(); ++i)|--> //为每个聚合函数天生一个evaluator|--> AggFnEvaluator::create(&evaluator)| |--> agg_fn_evaluator->_input_exprs_ctxs.push_back(ctx);|--> //将每个聚合函数的evaluator加到vector|--> _aggregate_evaluators.push_back(evaluator);|--> //end loopAggregationNode::prepare|--> ExecNode::prepare|--> AggregationNode::prepare_profile| |--> //begin loop| |--> for (int i = 0; i < _aggregate_evaluators.size(); ++i, ++j) | |--> //详细到某一个聚合函数| |--> _aggregate_evaluators[i]->prepare()| | |--> //初始化group by信息| | |--> VExpr::prepare()| | |--> //初始化| | |--> AggFnEvaluator::prepare| | | |--> //经由一些工厂函数的处理，终极调用到详细的聚合函数的创建| | | |--> create_aggregate_function_collect| | | | |--> create_agg_function_map_agg<String>(argument_types, result_is_nullable)| | | | | |--> //布局函数| | | | | |--> AggregateFunctionCollect(const DataTypes& argument_types_)| |--> //end loop| |--> //bind各种函数//调用AggregationNode::sink函数对输入block进行聚合，该步骤会利用前面分配的execute函数进行处理。 AggregationNode::sink|--> //in_block->rows()=15，便是数据的行数|--> _executor.execute(in_block)| |--> //group by+分桶id(一阶段聚合))即可触发| |--> //SELECT id, array_agg(label_name) FROM test_array_agg GROUP BY id;| |--> AggregationNode::_execute_with_serialized_key| | |--> AggregationNode::_execute_with_serialized_key_helper| | | |--> //这个时候num_rows便是所有记录的行数| | | |--> //但是这里循环了5次，由于有5个分组须要创建5个AggregateFunctionArrayAggData工具| | | |--> AggregationNode::_emplace_into_hash_table| | | | |--> PHHashMap<unsigned int, char, HashCRC32<unsigned int>, false>::lazy_emplace_keys| | | | | |--> //开始遍历所有的数据(keys.size()=15行)| | | | | |--> for (size_t i = 0; i < keys.size(); i++)| | | | | |--> _hash_map.lazy_emplace_with_hash(keys[i], hash_values[i]..)| | | | | | |--> //key不重复才会往下走，以是一共实行了5次AggregateFunctionMapAggData| | | | | | |--> creator-->AggregationNode::_create_agg_status| | | | | | | |--> AggregationNode::_create_agg_status| | | | | | | | |--> AggFnEvaluator::create| | | | | | | | | |--> AggregateFunctionCollect::create| | | | | | | | | | |--> //调用布局函数| | | | | | | | | | |--> AggregateFunctionArrayAggData()| | | | | |--> //结束遍历| | | |--> //begin loop| | | |--> for (int i = 0; i < _aggregate_evaluators.size(); ++i)| | | |--> //传入block，此时block有15行| | | |--> _aggregate_evaluators[i]->execute_batch_add | AggFnEvaluator::execute_batch_add| | | | |--> //block->rows()=17 offset=0 _agg_columns.data()有两列| | | | |--> IAggregateFunctionHelper::add_batch| | | | |--> //begin loop| | | | |--> //batch_size=15，实行15次add| | | | |--> for (size_t i = 0; i < batch_size; ++i)| | | | |--> AggregateFunctionCollect::add()| | | | |--> //end loop| | | |--> //end loop| || |--> //不带group by+一阶段聚合| |--> AggregationNode::_execute_without_key| | |--> AggFnEvaluator::execute_single_add| | | |--> IAggregateFunctionHelper::add_batch_single_place| | | |--> /begin loop/| | | |--> //实行15次| | | |--> for (size_t i = 0; i < batch_size; ++i)| | | |--> AggregateFunctionCollect::add()| | | |--> //end loop| |--> //group by+多阶段聚合| |--> AggregationNode::_merge_with_serialized_key| | |--> AggregateFunctionCollect::deserialize_and_merge_vec| |-->//无group by + 多阶段聚合| |-->//SELECT array_agg(label_name) FROM test_array_agg ;| |--> AggregationNode::_merge_without_key| | |--> AggregateFunctionCollect::deserialize_and_merge_from_column| |--> AggregationNode::_pre_agg_with_serialized_key| | |--> //如果聚合效果不佳，hash扩容达到阈值，则跳过聚合，直接把每一行输入当作一个分组| | |--> AggregateFunctionCollect::streaming_agg_serialize_to_column| | |--> //如果hash扩容没到阈值，还是采取朴素的方法| | |--> AggFnEvaluator::execute_batch_add| | | |--> //实行15次| | | |--> for (size_t i = 0; i < batch_size; ++i)| | | |--> AggregateFunctionCollect::add()| | | |--> //end loopAggregationNode::pull|--> //group by + 且须要finalize|--> //SELECT id, array_agg(label_name) FROM test_array_agg GROUP BY id;|--> AggregationNode::_get_with_serialized_key_result| |--> AggregationNode::_get_result_with_serialized_key_non_spill| | |--> //从block里面拿key的列，也便是group by的列| | |--> key_columns.emplace_back| | |--> //从block里面拿value的列| | |--> value_columns.emplace_back| | |--> //如果是一阶段聚合：这个时候num_rows=5，代表有5个分组| | |--> //SELECT id, array_agg(label_name) FROM test_array_agg GROUP BY id;| | |--> //如果是多阶段聚合：这个时候num_rows=1，须要在上层调用5次| | |--> //SELECT label_name, array_agg(label_name) FROM test_array_agg GROUP BY label_name;| | |--> AggFnEvaluator::insert_result_info(num_rows)| | | |--> for (size_t i = 0; i != num_rows; ++i)| | | |--> IAggregateFunctionHelper::insert_result_into| | | | |--> AggregateFunctionCollect::insert_result_into| | | | | |--> AggregateFunctionArrayAggData::insert_result_into| | | |--> //循环结束|--> //没有group by 且不须要finalize|--> AggregationNode::_serialize_without_key| |--> AggregateFunctionCollect::create_serialize_column| |--> AggregateFunctionCollect::serialize_without_key_to_column| | |--> AggregateFunctionArrayAggData::insert_result_into|--> //没有group by 且须要finalize|--> AggregationNode::_get_without_key_result| |--> AggregateFunctionCollect::insert_result_into| | |--> AggregateFunctionArrayAggData::insert_result_into|--> //group by + 且不须要finalize|--> AggregationNode::_serialize_with_serialized_key_result| |--> AggregationNode::_serialize_with_serialized_key_result_non_spill| | |--> //num_rows=5, 处理5个分组| | |--> AggregateFunctionCollect::serialize_to_column(num_rows)| | | |--> AggregateFunctionArrayAggData::insert_result_into

把稳点：

如果是两阶段聚合，在 execute 阶段一定会实行 execute+merge，即在会分别绑定 _merge_with 和 _execute_with，但是一阶段聚合只会绑定 _execute_with；如果是两阶段聚合，在 get_result 阶段会有多个 AggregationNode，会根据详细的情形判断是否 _needs_finalize；一阶段聚合只有一个 AggregationNode，会绑定 _needs_finalize。
总结

最近由于事情须要笔者开始调研和利用 Apache Doris，通过阅读聚合函数代码切入 Apache Doris 内核。
秉承着开源的精神，开拓了 array_agg 函数并贡献给社区。
希望通过这篇文章记录下对源码的一些理解，同时也方便后面的新人更快速地上手源码开拓。

在学习和节制 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