ClickHouse源码笔记1:聚合函数的实现

由于工作的需求，后续笔者工作需要和开源的OLAP数据库ClickHouse打交道。ClickHouse是Yandex在2016年6月15日开源了一个分析型数据库，以强悍的单机处理能力被称道。
笔者在实际测试ClickHouse和阅读ClickHouse的源码过程之中，对"战斗民族"开发的数据库十分欣赏。ClickHouse不仅是一个很好的数据库学习材料，而且同时应用了大量的CPP17的新特性进行开发，也是一个大型的Modern CPP的教导资料。
笔者接下来会陆续将阅读ClickHouse的部分心得体会与通过源码阅读笔记的方式和大家分享，坦白说，这种源码阅读笔记很难写啊。(多一分繁琐，少一分就模糊了~~)
第一篇文章，我们就从聚合函数的实现开始聊起~~ 上车！

1.基础知识的梳理

什么是聚合函数？

聚合函数: 顾名思义就是对一组数据执行聚合计算并返回结果的函数。
这类函数在数据库之中很常见，如：count, max, min, sum等等。

ClickHouse的实现接口
  • IAggregateFunction接口
    在ClickHouse之中，定义了一个统一的聚合函数接口：IAggregateFunction.（在ClickHouse之中，所有的接口类都是以大写的I开头的。) 上文笔者提到的聚合函数，则都是作为抽象类IAggregateFunction的子类实现的。其中该接口最为核心的方法是下面这5个方法：
    • add函数：最为核心的调用接口，将对应AggregateDataPtr指针之中数据取出，与列columns中的第row_num的数据进行对应的聚合计算。(这里可以看到ClickHouse是一个纯粹的列式存储数据库，所有的操作都是基于列的数据结构。）
    • merge函数：将两个聚合结果进行合并的函数，通常用在并发执行聚合函数的过程之中，需要将对应的聚合结果进行合并。
    • serialize函数与deserialize函数：序列化与反序列化的函数，通常用于spill to disk或分布式场景需要保存或传输中间结果的。
    • addBatch函数：这是函数也是非常重要的，虽然它仅仅实现了一个for循环调用add函数。它通过这样的方式来减少虚函数的调用次数，并且增加了编译器内联的概率。（虚函数的调用需要一次访存指令，一次查表，最终才能定位到需要调用的函数上，这在传统的火山模型的实现上会带来极大的CPU开销。）
  /** Adds a value into aggregation data on which place points to.
     *  columns points to columns containing arguments of aggregation function.
     *  row_num is number of row which should be added.
     *  Additional parameter arena should be used instead of standard memory allocator if the addition requires memory allocation.
     */
    virtual void add(AggregateDataPtr place, const IColumn ** columns, size_t row_num, Arena * arena) const = 0;

    /// Merges state (on which place points to) with other state of current aggregation function.
    virtual void merge(AggregateDataPtr place, ConstAggregateDataPtr rhs, Arena * arena) const = 0;

    /// Serializes state (to transmit it over the network, for example).
    virtual void serialize(ConstAggregateDataPtr place, WriteBuffer & buf) const = 0;

    /// Deserializes state. This function is called only for empty (just created) states.
    virtual void deserialize(AggregateDataPtr place, ReadBuffer & buf, Arena * arena) const = 0;
    // /** Contains a loop with calls to "add" function. You can collect arguments into array "places"
      *  and do a single call to "addBatch" for devirtualization and inlining.
      */
    virtual void addBatch(size_t batch_size, AggregateDataPtr * places, size_t place_offset, const IColumn ** columns, Arena * arena) const = 0;
  • 抽象类IColumn
    上面的接口IAggregateFunction的函数使用到了ClickHouse的核心接口IColumn类，这里也进行简要的介绍。 IColumn 接口表达了所有数据在ClickHouse之中的用内存表达的数据结构，其他带有具体数据类型的如ColumnUInt8、ColumnArray 等, 都实现了对应的列接口，并且在子类之中具象实现了不同的内存布局。
    IColumn的子类实现细节很琐碎，笔者这里就暂时不展开讲了，笔者这里就简单讲讲涉及到聚合函数调用部分的IColumn接口的对应方法：
    这里columns是一个二维数组，通过columns[0]可以取到第一列。(这里只有涉及到一列，为什么columns是二维数组呢？因为处理array等列的时候，也是通过对应的接口，而array就需要应用二维数组了. )
    注意这里有一个强制的类型转换，column已经转换为ColVecType类型了，这是模板派生出IColumn的子类。
    然后通过IColumn子类实现的getData方法获取对应row_num行的数据进行add函数调用就完成了一次聚合函数的计算了。
    void add(AggregateDataPtr place, const IColumn ** columns, size_t row_num, Arena *) const override
    {
        const auto & column = static_cast<const ColVecType &>(*columns[0]);
        this->data(place).add(column.getData()[row_num]);
    }
  • IAggregateFunctionHelper接口
    这个接口是上面提到 IAggregateFunction的辅助子类接口，它很巧妙的通过模板的类型派生，将虚函数的调用转换为函数指针的调用，这个在实际聚合函数的实现过程之中能够大大提高计算的效率。
    函数addFree就实现了我上述所说的过程，但是它是一个private的函数，所以通常我们都是通过getAddressOfAddFunction获取对应的函数地址。这在聚合查询的过程之中能够提高20%左右的执行效率。
template <typename Derived>
class IAggregateFunctionHelper : public IAggregateFunction
{
private:
    static void addFree(const IAggregateFunction * that, AggregateDataPtr place, const IColumn ** columns, size_t row_num, Arena * arena)
    {
        static_cast<const Derived &>(*that).add(place, columns, row_num, arena);
    }

public:
    IAggregateFunctionHelper(const DataTypes & argument_types_, const Array & parameters_)
        : IAggregateFunction(argument_types_, parameters_) {}

    AddFunc getAddressOfAddFunction() const override { return &addFree; }
  • AggregateFunctionFactory类
    顾名思义，这个是一个生成聚合函数的工厂类。它的逻辑很简单，所有ClickHouse之中所相关的聚合函数都是通过这个工厂类注册并且获取，然后进行调用的。
class AggregateFunctionFactory final : private boost::noncopyable, public IFactoryWithAliases<AggregateFunctionCreator>
{
public:

    static AggregateFunctionFactory & instance();

    /// Register a function by its name.
    /// No locking, you must register all functions before usage of get.
    void registerFunction(
        const String & name,
        Creator creator,
        CaseSensitiveness case_sensitiveness = CaseSensitive);

    /// Throws an exception if not found.
    AggregateFunctionPtr get(
        const String & name,
        const DataTypes & argument_types,
        const Array & parameters = {},
        int recursion_level = 0) const;

2.聚合函数的注册流程

有了上述的背景知识，我们接下来举个栗子。来看看一个聚合函数的实现细节，以及它是如何被使用的。

AggregateFunctionSum

笔者这里选取了一个很简单的聚合算子Sum，我们来看看它实现的代码细节。
这里我们可以看到AggregateFunctionSum是个final类，无法被继承了。而它继承了上面提到的IAggregateFunctionHelp类的子类IAggregateFunctionDataHelper类。

这里我们就重点看，这个类override了getName方法，返回了对应的名字sum。并且实现了我们上文提到的四个核心的方法。

  • add
  • merge
  • seriable
  • deserialize
template <typename T, typename TResult, typename Data>
class AggregateFunctionSum final : public IAggregateFunctionDataHelper<Data, AggregateFunctionSum<T, TResult, Data>>
{
public:
    using ResultDataType = std::conditional_t<IsDecimalNumber<T>, DataTypeDecimal<TResult>, DataTypeNumber<TResult>>;
    using ColVecType = std::conditional_t<IsDecimalNumber<T>, ColumnDecimal<T>, ColumnVector<T>>;
    using ColVecResult = std::conditional_t<IsDecimalNumber<T>, ColumnDecimal<TResult>, ColumnVector<TResult>>;

    String getName() const override { return "sum"; }

    AggregateFunctionSum(const DataTypes & argument_types_)
        : IAggregateFunctionDataHelper<Data, AggregateFunctionSum<T, TResult, Data>>(argument_types_, {})
        , scale(0)
    {}

    AggregateFunctionSum(const IDataType & data_type, const DataTypes & argument_types_)
        : IAggregateFunctionDataHelper<Data, AggregateFunctionSum<T, TResult, Data>>(argument_types_, {})
        , scale(getDecimalScale(data_type))
    {}

    DataTypePtr getReturnType() const override
    {
        if constexpr (IsDecimalNumber<T>)
            return std::make_shared<ResultDataType>(ResultDataType::maxPrecision(), scale);
        else
            return std::make_shared<ResultDataType>();
    }

    void add(AggregateDataPtr place, const IColumn ** columns, size_t row_num, Arena *) const override
    {
        const auto & column = static_cast<const ColVecType &>(*columns[0]);
        this->data(place).add(column.getData()[row_num]);
    }

    void merge(AggregateDataPtr place, ConstAggregateDataPtr rhs, Arena *) const override
    {
        this->data(place).merge(this->data(rhs));
    }

    void serialize(ConstAggregateDataPtr place, WriteBuffer & buf) const override
    {
        this->data(place).write(buf);
    }

    void deserialize(AggregateDataPtr place, ReadBuffer & buf, Arena *) const override
    {
        this->data(place).read(buf);
    }

    void insertResultInto(ConstAggregateDataPtr place, IColumn & to) const override
    {
        auto & column = static_cast<ColVecResult &>(to);
        column.getData().push_back(this->data(place).get());
    }

private:
    UInt32 scale;
};

接下来，ClickHouse实现了两种聚合计算：AggregateFunctionSumData和AggregateFunctionSumKahanData。后者是用Kahan算法避免float类型精度损失的，我们可以暂时不细看。直接看SumData的实现。这是个模板类，之前我们讲到AggregateFunction的函数就是通过AggregateDataPtr指针来获取AggregateFunctionSumData的地址，来调用add实现聚合算子的。我们可以看到AggregateFunctionSumData实现了前文提到的add, merge, write,read四大方法，正好和接口一一对应上了。

template <typename T>
struct AggregateFunctionSumData
{
    T sum{};

    void add(T value)
    {
        sum += value;
    }

    void merge(const AggregateFunctionSumData & rhs)
    {
        sum += rhs.sum;
    }

    void write(WriteBuffer & buf) const
    {
        writeBinary(sum, buf);
    }

    void read(ReadBuffer & buf)
    {
        readBinary(sum, buf);
    }

    T get() const
    {
        return sum;
    }
};

ClickHouse在Server启动时。main函数之中会调用registerAggregateFunction的初始化函数注册所有的聚合函数。
然后调用到下面的函数：

void registerAggregateFunctionSum(AggregateFunctionFactory & factory)
{
    factory.registerFunction("sum", createAggregateFunctionSum<AggregateFunctionSumSimple>, AggregateFunctionFactory::CaseInsensitive);
    factory.registerFunction("sumWithOverflow", createAggregateFunctionSum<AggregateFunctionSumWithOverflow>);
    factory.registerFunction("sumKahan", createAggregateFunctionSum<AggregateFunctionSumKahan>);
}

这里又调用了factory.registerFunction("sum", createAggregateFunctionSum<AggregateFunctionSumSimple>, AggregateFunctionFactory::CaseInsensitive);来进行上述我们看到的聚合函数的注册。这里有一点很恶心的模板代码，笔者这里简化了一下，把注册的部分函数拉出来：

createAggregateFunctionSum(const std::string & name, const DataTypes & argument_types, const Array & parameters)
{
    AggregateFunctionPtr res;
    DataTypePtr data_type = argument_types[0];
    if (isDecimal(data_type))
        res.reset(createWithDecimalType<Function>(*data_type, *data_type, argument_types));
    else
        res.reset(createWithNumericType<Function>(*data_type, argument_types));
    return res;

这里的Function模板就是上面的AggregateFunctionSumSimple, 而它又是下面的模板类型：

template <typename T> using AggregateFunctionSumSimple = typename SumSimple<T>::Function;

template <typename T>
struct SumSimple
{
    /// @note It uses slow Decimal128 (cause we need such a variant). sumWithOverflow is faster for Decimal32/64
    using ResultType = std::conditional_t<IsDecimalNumber<T>, Decimal128, NearestFieldType<T>>;
    using AggregateDataType = AggregateFunctionSumData<ResultType>;
    using Function = AggregateFunctionSum<T, ResultType, AggregateDataType>;
};

不知道读者被绕晕了没，最终绕回来还是new出来这个AggregateFunctionSum<T, ResultType, AggregateDataType>
也就是完成了这个求和算子的注册，后续我们get出来就可以愉快的调用啦。（这里这部分的模板变化比较复杂，如果看不明白可以回到源码梳理一下~~~）

3. 小结

好了，关于聚合函数的基础信息，和它是如何实现并且通过工厂方法注册获取的流程算是搞明白了。
关于其他的聚合算子，也是大同小异的方式。笔者就不再赘述了，感兴趣的可以回到源码之中继续一探究竟。讲完了聚合函数的实现，下一篇笔者就要继续给探究聚合函数究竟在ClickHouse之中是如何和列存结合使用，并实现向量化的~~。
笔者是一个ClickHouse的初学者，对ClickHouse有兴趣的同学，也欢迎和笔者多多指教，交流。

4. 参考资料

官方文档
ClickHouse源代码

©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 204,053评论 6赞 478
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 85,527评论 2赞 381
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 150,779评论 0赞 337
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 54,685评论 1赞 276
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 63,699评论 5赞 366
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,609评论 1赞 281
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 37,989评论 3赞 396
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,654评论 0赞 258
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 40,890评论 1赞 298
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,634评论 2赞 321
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,716评论 1赞 330
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,394评论 4赞 319
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 38,976评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,950评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,191评论 1赞 260
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 44,849评论 2赞 349
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,458评论 2赞 342