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實現了兩種聚合計算:AggregateFunctionSumDataAggregateFunctionSumKahanData。后者是用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源代碼


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM