ClickHouse源碼筆記5:聚合函數的源碼再梳理
- 2021 年 4 月 20 日
- 筆記
- ClickHouse
筆者在源碼筆記1之中分析過ClickHouse的聚合函數的實現，但是對於各個介面函數的實際如何共同工作的源碼，回頭看並沒有那麼明晰，主要原因是沒有結合Aggregator的類來一起分析聚合函數的是如果工作起來的。所以決定重新再完成一篇聚合函數的源碼梳理的文章，幫助大家進一步的理解ClickHouse之中聚合函數的工作原理。
本系列文章的源碼分析基於ClickHouse v19.16.2.2的版本。
1.IAggregateFunction介面梳理
話不多說，直接上程式碼，筆者這裡會將所有聚合函數的核心介面程式碼全部列出，一一梳理各個部分：
構造函數
IAggregateFunction(const DataTypes & argument_types_, const Array & parameters_)
: argument_types(argument_types_), parameters(parameters_) {}
上面的程式碼實現了IAggregateFunction介面的構造函數，初始化了該介面的兩個成員變數：
argument_type：函數的參數類型，比如函數select sum(a), sum(b), c from test group by c, 這裡a,b分別是UInt16類型與Double類型，那麼這個sum(a)與sum(b)的參數就不同。parameters: 參數，實際類型為std::vector<Field>。它代表著函數的除了數據的輸入參數之外的其他參數。比如聚合函數topk，其中需要傳入的k的值就在parameters之中。
記憶體分配介面
在Clickhouse的聚合執行過程之中，所有的聚合函數都是通過列來進行的。而這裡有兩個重要的問題：
- 列記憶體從哪裡分配
- 分配的記憶體結構，長度是如何的
筆者在梳理下面程式碼的過程之中給出解答,
/** Create empty data for aggregation with `placement new` at the specified location.
* You will have to destroy them using the `destroy` method.
*/
virtual void create(AggregateDataPtr place) const = 0;
/// Delete data for aggregation.
virtual void destroy(AggregateDataPtr place) const noexcept = 0;
IAggregateFunction定義的兩個介面create與destory介面完成了記憶體結構與長度的確定，這裡可能描述的不是很明白，這裡了解Doris聚合實現的同學可以這樣理解。create函數本身就是完成了Doris聚合函數之中init函數所完成的工作。這裡通過子類IAggregateFunctionDataHelper的實現程式碼來進一步理解它做了什麼事情：
void create(AggregateDataPtr place) const override
{
new (place) Data;
}
void destroy(AggregateDataPtr place) const noexcept override
{
data(place).~Data();
}
這部分程式碼很簡單，Data就是模板派生的類型，然後通過placement new與placement delete的方式完成了Data類型的構造與析構。而這個Data類型就是聚合函數存儲中間結果的類型，比如sum的聚合函數的派生類型是類AggregateFunctionSumData的記憶體結構，它不僅包含了聚合結果的數據sum同時也包含了一組進行聚合計算的函數介面add,merge等:
template <typename T>
struct AggregateFunctionSumData
{
T sum{};
void add(T value)
{
sum += value;
}
void merge(const AggregateFunctionSumData & rhs)
{
sum += rhs.sum;
}
T get() const
{
return sum;
}
};
這裡就是通過create與destory函數調用AggregateFunctionSumData的構造函數與析構函數。而問題又繞回第一個問題了，這部分記憶體是在哪裡分配的呢？
aggregate_data = aggregates_pool->alignedAlloc(total_size_of_aggregate_states, align_aggregate_states);
createAggregateStates(aggregate_data);
在進行聚合運算時，通過Aggregator之中的記憶體池進行單行所有的聚合函數的數據結果的記憶體分配。並且調用createAggregateStates依次調用各個聚合函數的create方法進行構造函數的調用。這部分可能有些難理解，我們接著看下面的流程圖，來更好的幫助理解：
通過上述流程圖可以看到，create這部分就是在構造聚合hash表時，進行記憶體初始化工作的，而這部分記憶體不僅僅包含了聚合函數的結果數據，還包含了對應聚合運算元的函數指針。後文我們分析計算介面的時候也會同樣看到。接下來，來看destory就很容易理解了，就是在聚合計算結束或取消時，遍歷hash表，並調用析構函數對hash表中存儲的Data類型調用析構函數，而最終的記憶體伴隨著aggregates_pool記憶體池的析構而同時釋放。
函數計算介面
接下來就是聚合函數最核心的部分，聚合函數的計算。
/** 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;
/** 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;
IAggregateFunction定義的3個介面:
add函數將對應AggregateDataPtr指針之中數據取出，與列columns中的第row_num的數據進行對應的聚合計算。addBatch函數：這是函數也是非常重要的，雖然它僅僅實現了一個for循環調用add函數。它通過這樣的方式來減少虛函數的調用次數，並且增加了編譯器內聯的概率，同樣，它實現了高效的向量化。merge函數：將兩個聚合結果進行合併的函數，通常用在並發執行聚合函數的過程之中，需要將對應的聚合結果進行合併。
這裡的兩個函數類似Doris之中聚合函數的update與merge。接下來我們看它是如何完成工作的。
首先看聚合節點Aggregetor是如何調用addBatch函數：
/// Add values to the aggregate functions.
for (AggregateFunctionInstruction * inst = aggregate_instructions; inst->that; ++inst)
inst->that->addBatch(rows, places.data(), inst->state_offset, inst->arguments, aggregates_pool);
這裡依次遍歷AggregateFunction，並調用addBatch介面。而addBatch介面就是一行行的遍歷列，將參數列inst->arguments與上文提到create函數構造的聚合數據結構的兩列列數據進行聚合計算:
void addBatch(size_t batch_size, AggregateDataPtr * places, size_t place_offset, const IColumn ** columns, Arena * arena) const override
{
for (size_t i = 0; i < batch_size; ++i)
static_cast<const Derived *>(this)->add(places[i] + place_offset, columns, i, arena);
}
這裡還是調用了add函數，我們通過AggregateFunctionSum作為子類來具體看一下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]);
}
這裡其實還是調用上文提到的AggregateFunctionSumData的記憶體結構的add函數完成聚合計算。而這個add函數就是一個簡單的相加邏輯，這樣就完成了簡單的一次聚合運算。
void add(T value)
{
sum += value;
}
而merge函數的實現邏輯類似於add函數，這裡就不展開再次分析了。
函數結果輸出介面
最後就是聚合函數結果輸出介面，將聚合計算的結果重新組織為列存。
/// Inserts results into a column.
virtual void insertResultInto(ConstAggregateDataPtr place, IColumn & to) const = 0;
首先看聚合節點Aggregator是如何調用insertResultInto函數的
data.forEachValue([&](const auto & key, auto & mapped)
{
method.insertKeyIntoColumns(key, key_columns, key_sizes);
for (size_t i = 0; i < params.aggregates_size; ++i)
aggregate_functions[i]->insertResultInto(
mapped + offsets_of_aggregate_states[i],
*final_aggregate_columns[i]);
});
Aggregetor同樣是遍歷hash表之中的結果，將key列先組織成列存，然後調用insertResultInto函數將聚合計算的結果也轉換為列存。
這裡我們找一個sum函數的實現，來看看insertResultInto函數介面是如何工作的：
void insertResultInto(ConstAggregateDataPtr place, IColumn & to) const override
{
auto & column = static_cast<ColVecResult &>(to);
column.getData().push_back(this->data(place).get());
}
其實很簡單，就是調用AggregateDataPtr,也就是AggregateFunctionSumData的get()函數獲取sum計算的結果，然後添加到列記憶體之中。
get函數介面的實現如下：
T get() const
{
return sum;
}
2.聚合函數的註冊流程
有了上述的背景知識，我們接下來舉個栗子。來看看一個聚合函數的實現細節，以及它是如何被使用的。
AggregateFunctionSum
這裡選取了一個很簡單的聚合運算元Sum，我們來看看它實現的程式碼細節。
這裡我們可以看到AggregateFunctionSum是個final類，無法被繼承了。而它繼承IAggregateFunctionHelp類與IAggregateFunctionDataHelper類。
IAggregateFunctionHelp類 通過CRTP讓父類可以直接調用子類的add函數指針而避免了虛函數調用的開銷。IAggregateFunctionHelper類則包含了Data的模板數據類型，也就是上文提及的AggregateFunctionSumData進行記憶體結構的create，destory等等。
這裡我們就重點看，這個類override了getName方法，返回了對應的名字時sum。並且實現了我們上文提到核心方法。
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;
};
之前我們講到AggregateFunction的函數就是通過AggregateDataPtr指針來獲取AggregateFunctionSumData的地址，來調用add實現聚合運算元的。我們可以看到AggregateFunctionSumData實現了前文提到的add, merge, write,read四大方法，正好與介面IAggregateFunction一一對應上了。
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的初始化函數註冊所有的聚合函數。
然後調用到下面的函數註冊sum的聚合函數：
void registerAggregateFunctionSum(AggregateFunctionFactory & factory)
{
factory.registerFunction("sum", createAggregateFunctionSum<AggregateFunctionSumSimple>, AggregateFunctionFactory::CaseInsensitive);
}
也就是完成了這個sum聚合函數的註冊，後續我們get出來就可以愉快的調用啦。（這部分有許多模板派生的複雜程式碼，建議與源碼結合梳理才能事半功倍~~）
3.要點梳理
第二小節解析了一個聚合函數與介面意義對應的流程，這裡重點梳理聚合函數實現的源碼要點：
- 各個聚合函數核心的實現
add,merge與序列化，記憶體結構初始化，記憶體結構釋放的介面。 - 各個函數的實現需要繼承IAggregateFunctionDataHelper的介面，而它的父類是
IAggregateFunctionHelper與IAggregateFunction介面。 - ClickHouse的聚合函數保證了每次循環遍歷一個Block只調用一個
IAggregateFunction的聚合函數，這樣最大程度上確保了向量化執行的可能性，減少了數據偏移與依賴。
4. 小結
好了，到這裡也就把ClickHouse聚合函數部分的程式碼梳理完了。
除了sum函數外，其他的函數的執行也是同樣通過類似的方式依次來實現和處理的，源碼閱讀的步驟也可以參照筆者的分析流程來參考。
筆者是一個ClickHouse的初學者，對ClickHouse有興趣的同學，歡迎多多指教，交流。
