java基础知识367



1.基础知识的梳理

ClickHouse的实现接口

• Block类 前文我们聊到ClickHouse是一个列式存储数据库，在内存之中用IColumn接口来作为数据结构表示数据。 而Block则是这些列的集合，也就是说Block包含了一组列，而无数个Block就构成了我们通常理解的表了。 在ClickHouse进行查询之中，数据的最小处理单位是 Block 。由下面代码可以看到，Block就是由一组列以及列名对应列的偏移map组成的。

复制

这是一个很重要的类，实现的也并不复杂。Block类作为ClickHouse的核心，后续的工作都是基于Block类展开的。

• 抽象类IBlockInputStream 由名字可以看出，IBlockInputStream是一个实现接口。 这也同样是一个十分重要的接口，ClickHouse的调用模型就建立在IBlockInputStream接口之上。该接口最为核心的就是方法便是，它返回一个被对应Stream处理过的Block。 想必看到这里应该明白了，ClickHouse就是通过IBlockInputStream实现的火山模型，每一个不同的Stream处理不同的查询逻辑，最后层层迭代，完成最终输出流就是用户需要的结果了。 IBlockInputStream类还有一个孪生兄弟IBlockoutputStream，顾名思义，需要进行写操作的时候就要用到它了。

复制

• AggregatingBlockInputStream类 终于引出我们的主角了，AggregatingBlockInputStream类，作为上面IBlockInputStream的子类，也就是我们今天要重点分析的类。

复制

首先看它的构造方法，参数有：

• BlockInputStreamPtr: 这个很好理解，就是它的子流，也就是实际产生数据的流，后续的聚合计算将会在子流返回的结果上展开。
• params: 聚合参数，这个参数十分重要。它记录了那些key属于聚合，调用那些聚合参数等核心信息。并且aggregator也就是执行聚合的类，也是通过该参数构造的，它是的内部类。
• final: 指明该Stream是否是最终结果，java基础知识367还是要继续进行计算。

这里最为核心的就是AggregatingBlockInputStream类通过继承override对应的readImpl()的接口来实现对应的具体逻辑。AggregatingBlockInputStream类还有一个孪生兄弟：ParallelAggregatingBlockInputStream类，通过并行化来进一步加快聚合流程的执行效率。(通过笔者进行的测试，在简单查询聚合查询下，并行化能够提高近一倍的效率～～)

• Aggregator::Params类 Aggregator::Params类是Aggregator的内部类。这个类是整个聚合过程之中最重要的类，查询解析优化后生成聚合查询的执行计划。 而对应的执行计划的参数都通过Aggregator::Params类来初始化，比如那些列要进行聚合，选取的聚合算子等等，并传递给对应的Aggregator来实现对应的聚合逻辑。

复制

• Aggregator类 顾名思义，这个是一个实际进行聚合工作展开的类。它最为核心的方法是下面两个函数：

• execute函数：将输入流的stream依照次序进行blcok迭代处理，将聚合的结果写入result之中。
• mergeAndConvertToBlocks函数：将聚合的结果转换为输入流，并通过输入流的read函数将结果继续返回给上一层。 通过上面两个函数的调用，我们就可以完成被聚合的数据输入-》 数据聚合 -》 数据输出的流程。具体的细节笔者会在下一章详细的进行剖析。

复制

2.聚合流程的实现

这里我们就从上文提到的Aggregator::execute(const BlockInputStreamPtr & stream, AggregatedDataVariants & result)函数作为起点来梳理一下ClickHouse的聚合实现：

复制

由上述代码可以看出，这里就是依次读取子节点流生成的Block，然后继续调用executeOnBlock方法来执行聚合流程处理每一个Block的聚合。接着我们按图索骥，继续看下去，这个函数比较长，我们拆分成几个部分，并且把无关紧要的代码先去掉：这部分主要完成的工作就是将param之中指定的key列与聚合列的指针作为参数提取出来，并且和聚合函数一起封装到AggregateFunctionInstructions的结构之中。

复制

将需要准备的参数准备好了之后，后续就通过按部就班的调用executeImpl(*result.NAME, result.aggregates_pool, num_rows, key_columns, aggregate_functions_instructions.data(), no_more_keys, overflow_row_ptr)聚合运算了。我们来看看它的实现，它是一个模板函数，内部通过调用了 executeImplBatch(method, state, aggregates_pool, rows, aggregate_instructions)来实现的，数据库都会通过Batch的形式，一次性提交一组需要操作的数据来减少虚函数调用的开销。

复制

那我们就继续看下去，executeImplBatch同样也是一个模板函数。

• 首先，它构造了一个AggregateDataPtr的数组places，这里是这就是后续我们实际聚合结果存放的地方。这个数据的长度也就是这个Batch的长度，也就是说，聚合结果的指针也作为一组列式的数据，参与到后续的聚合运算之中。
• 接下来，通过一个for循环，依次调用state.emplaceKey，计算每列聚合key的hash值，进行分类，并且将对应结果依次和places对应。
• 最后，通过一个for循环，调用聚合函数的addBatch方法，（这个函数我们在上一篇之中介绍过）。每个AggregateFunctionInstruction都有一个制定的places_offset和对应属于进行聚合计算的value列，这里通过一个for循环调用AddBatch，将places之中对应的数据指针和聚合value列进行聚合，最终形成所有的聚合计算的结果。

到这里，整个聚合计算的核心流程算是完成了，后续就是将result的结果通过上面的convertToBlock的方式转换为BlockStream流，继续返回给上层的调用方。

复制

3. 小结

好了，到这里也就把ClickHouse聚合流程的代码梳理完了。 除了聚合计算外，其他的物理执行操作符也是同样通过流的方式依次对接处理的，源码阅读的步骤也可以参照笔者的分析流程来参考。

4. 参考资料

版权声明：

本文来源网络，所有图片文章版权属于原作者，如有侵权，联系删除。

本文网址：https://www.bianchenghao6.com/h6javajc/20578.html