TiFlash 原始碼閱讀（八）TiFlash 表示式的實現與設計

Volcano Model 源自 1994 年的論文 Volcano-An Extensible and Parallel Query Evaluation System 。Volcano Model 將 SQL 分解為若干個獨立的 Operator，Operator 串聯成一棵 Operator Tree。

如上圖所示，從最源頭的 Table Scan Operator 開始，一行一行地讀取資料，Operator 處理後，傳給上游 Operator。最終 Operator Tree 一行一行地輸出結果。

下面是對 Operator 介面的一個簡單的虛擬碼描述。

struct Operator
{
Row next()
{
    Row row = child.next();
    ....
    return row;
}
}

Volcano Model 提供了一個非常簡潔的執行模型，在工程上也非常容易實現。但是 Volcano Model 在現代編譯器和硬體下執行得慢。在後續幾年了誕生了對 Volcano Model 的兩個改進方案，Codegen 和 Vectorized。TiFlash 就是使用的 Vectorized 即向量化執行。

Block 是若干 Row 組合在一起的資料塊，Block 內部按 Column 儲存資料。

這樣設計的好處有幾個:

• 虛擬函式呼叫的開銷會被減小為 1 / (Block Size)。Volcano Model 中的 Operator 和表示式通常都是用多型來實現的，在其中就會引入虛擬函式呼叫。每次 Operator Tree 和內部的表示式被呼叫就是一系列的虛擬函式呼叫，在資料量大的情況下，虛擬函式開銷甚至會成為一個不可忽視的點。Block by Block 可以讓 Operator Tree 和內部的表示式的一次呼叫處理多行而不是一行資料，從而均攤了虛擬函式開銷。

• Cache Friendly。把會被連續處理的資料放在一個數據塊裡，提高在 Cache 上的空間區域性性。

左右滑動檢視

• execute

• 對 ExpressionAction::execute   的包裝。 用於執行表示式。

• add

• 用於外部組裝出一個 ExpressionActions 。

• ExpressionActions  會維護一個 Sample Block，在 Add Action 的過程中 Sample Block 會不斷更新，模擬實際 Block 的變化情況。

• 在 Add 的過程中，重複的 Action 會被跳過。重複 Action 的判斷條件是，該 Action 的執行結果是否已經出現在 Sample Block 裡了。

• finalize

• 分析 Block 內的 Column 引用情況，在合適的位置插入 Remove Action 來移除無用的 Column。

• 在 Column 引用數歸 0 的時候，就會插入對應 Column 的 Remove Action。

• 下圖是 (a + b) * (a + b) 的引用數分析和 Remove Action 插入情況。

ExpressionAction   是  ExpressionActions   內部的執行單元。

struct ExpressionAction
{
    Type type;
    …

    void execute(Block & block) const
    {
        switch (type)
        {
            case: APPLY_FUNCTION:
                …
            case: REMOVE_COLUMN:
               …
            case: ADD_COLUMN:
               …
        }
    }
}

ExpressionAction  有不同的 Type 用於對 Block 進行不同的處理

• REMOVE_COLUMN

• 即前文所講的 Remove Action

• ADD_COLUMN

• 用於執行 Literal  Expression，在 Block 插入一個 Const Column。

• ADD_COLUMN  會在 Block 中插入一個 Column。 對於 Literal，插入的會是  ColumnConst ，即常量 Column。

• block.insert({added_column->cloneResized(block.rows()), result_type, result_name});

• APPLY_FUNCTION

• 用於執行 Function Expression，由 ExpressionAction 持有的  IFunction   執行

• APPLY_FUNCTION  會讀取 Block 中的 Argument Columns 傳給  IFunction  做計算。 計算出結果 Column 後，插入到 Block 中。

String DAGExpressionAnalyzerHelper::buildFunction(
    DAGExpressionAnalyzer * analyzer,
    const tipb::Expr & expr,
    const ExpressionActionsPtr & actions)
{
    const String & func_name = getFunctionName(expr);
    if (function_builder_map.count(func_name) != 0)
    {
        return function_builder_map[func_name](analyzer, expr, actions);
    }
    else
    {
        return buildDefaultFunction(analyzer, expr, actions);
    }
}

Default Function Builder 的處理本身很簡單，就是先處理所有的函式入參，然後呼叫  applyFunction 生成對應的  IFunction 實現。

String DAGExpressionAnalyzerHelper::buildDefaultFunction(
    DAGExpressionAnalyzer * analyzer,
    const tipb::Expr & expr,
    const ExpressionActionsPtr & actions)
{
    const String & func_name = getFunctionName(expr);
    Names argument_names;
    for (const auto & child : expr.children())
    {
        String name = analyzer->getActions(child, actions);
        argument_names.push_back(name);
    }
    return analyzer->applyFunction(func_name, argument_names, actions, getCollatorFromExpr(expr));
}

對某些函式，有些特殊處理，這時就會使用到 Special Function Builder。

特殊處理的 Function Builder 會放在表裡，遇到對應的函式，會轉發過來。

下面是一些函式的特殊處理對映。

FunctionBuilderMap DAGExpressionAnalyzerHelper::function_builder_map(
    {...
     {"ifNull", DAGExpressionAnalyzerHelper::buildIfNullFunction},
     {"multiIf", DAGExpressionAnalyzerHelper::buildMultiIfFunction},
     ...
     {"bitAnd", DAGExpressionAnalyzerHelper::buildBitwiseFunction},
     {"bitOr", DAGExpressionAnalyzerHelper::buildBitwiseFunction},
     {"bitXor", DAGExpressionAnalyzerHelper::buildBitwiseFunction},
     {"bitNot", DAGExpressionAnalyzerHelper::buildBitwiseFunction},
     {"bitShiftLeft", DAGExpressionAnalyzerHelper::buildBitwiseFunction},
     {"bitShiftRight", DAGExpressionAnalyzerHelper::buildBitwiseFunction},
     ...
});

一個特殊的處理的情況是複用函式實現。某些函式可以由另一個函式來代理執行，比如

leftUTF8(str, len) = substrUTF8(str, 1, len) 。如果讓 substrUTF8 來代理執行 leftUTF8 ，那麼就可以省掉 leftUTF8 本身的開發實現工作。下面是 leftUTF8(str, len) = substrUTF8(str, 1, len) 的代理實現程式碼。為 substrUTF8 生成第二個引數 1 後，將 leftUTF8 的兩個引數傳入 substrUTF8 ， substrUTF8 就可以代理 leftUTF8 執行。

String DAGExpressionAnalyzerHelper::buildLeftUTF8Function(
    DAGExpressionAnalyzer * analyzer,
    const tipb::Expr & expr,
    const ExpressionActionsPtr & actions)
{
    const String & func_name = "substringUTF8";
    Names argument_names;

    // the first parameter: str
    String str = analyzer->getActions(expr.children()[0], actions, false);
    argument_names.push_back(str);

    // the second parameter: const(1)
    auto const_one = constructInt64LiteralTiExpr(1);
    auto col_const_one = analyzer->getActions(const_one, actions, false);
    argument_names.push_back(col_const_one);

    // the third parameter: len
    String name = analyzer->getActions(expr.children()[1], actions, false);
    argument_names.push_back(name);

    return analyzer->applyFunction(func_name, argument_names, actions, getCollatorFromExpr(expr));
}

接下來我們來看一下 IFunction   本身的介面。

• getName : 返回 Function 的 Name，Name 是作為 TiFlash 向量化函式的唯一標識來使用。

• getReturnTypeImpl : 負責做向量化函式的型別推導，因為輸入引數資料型別的變化可能會導致輸出資料型別變化。

• executeImpl : 負責向量化函式的執行邏輯，這也是一個向量化函式的主體部分。 一個 TiFlash 向量化函式夠不夠"向量化"，夠不夠快也就看這裡了。

接下來以 jsonLength(string) 為例子，講一下向量化函式的執行

1. 從 Block 中獲取 Json Column

2. 建立同等大小的 Json Length Column

3. Foreach Json Column，獲取每一個行的 Json

4. 呼叫 GetJsonLength(Json) 獲取 Json Length，將結果插入 Json Length Column 中的對應位置。

5. 將 Json Length Column 插入到 Block 中，完成單次計算

void executeImpl(Block & block, const ColumnNumbers & arguments, size_t result) const override
{
    // 1. 獲取 json column，json column 本身是 String 型別，所以 json column 用的是 ColumnString 這個 column 實現
    const ColumnPtr column = block.getByPosition(arguments[0]).column;
    if (const auto * col = checkAndGetColumn<ColumnString>(column.get()))
    {
        // 2.建立 json len column, json len 本身是 UInt64 型別，用的是 ColumnUInt64 這個 column 實現
        auto col_res = ColumnUInt64::create();
        typename ColumnUInt64::Container & vec_col_res = col_res->getData();
        {
            // 3. 遍歷 json column，ColumnString 提供了一些裸操作內部 string 的方法，可以提高效率
            const auto & data = col->getChars();
            const auto & offsets = col->getOffsets();
            const size_t size = offsets.size();
            vec_col_res.resize(size);

            ColumnString::Offset prev_offset = 0;
            for (size_t i = 0; i < size; ++i)
            {
                // 4. 呼叫 GetJsonLength，計算出 json 的 length，插入到 json len column 中。
                std::string_view sv(reinterpret_cast<const char *>(&data[prev_offset]), offsets[i] - prev_offset - 1);
                vec_col_res[i] = GetJsonLength(sv);
                prev_offset = offsets[i];
            }
        }
        // 5. 將 json len column 插入到 Block 中，完成單次計算。
        block.getByPosition(result).column = std::move(col_res);
    }
    else
        throw Exception(fmt::format("Illegal column {} of argument of function {}", column->getName(), getName()), ErrorCodes::ILLEGAL_COLUMN);
}

前段時間 TiFlash 有個社群活動，號召大家來參與 TiFlash 函式下推的工作。

以上關於標量函式的內容在社群活動的兩篇文章 TiFlash 函式下推必知必會 和  手把手教你實現 TiFlash 向量化函式 都有包含，大家可以通過這兩篇文章瞭解更多關於 TiFlash 標量函式的內容。也歡迎小夥伴們也參與到 TiFlash 函式下推的工作中來。

所以聚合函式不會使用 ExpressionActions 和  IFunction ，而是有獨立的執行體系。

聚合函式本身由 Aggregate 運算元來執行， Aggregate 運算元負責管理聚合函式的執行步驟，執行併發度等等。

Aggregate   運算元的執行有兩部分

• 前者 ExpressionActions 用於執行標量函式的部分。

• 後者 Aggregator 用於執行聚合函式的部分。

如下圖所示，對於 select max(a+b) ， ExpressionActions 執行 a+b ， Aggregator 執行 max 。

IAggregateFunction   是聚合函式的實現基類。 接下來我們來看一下  IAggregateFunction   本身的介面。

class IAggregateFunction
{
public:
    String getName() const = 0;

    DataTypePtr getReturnType() const = 0;

    void add(AggregateDataPtr __restrict place, const IColumn ** arg_columns, size_t row_num, Arena * arena) const = 0;

    void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs, Arena * arena) const = 0;

    void insertResultInto(ConstAggregateDataPtr __restrict place, IColumn & to, Arena * arena) const = 0;
};

• add 用於計算聚合結果並更新到 place 上

• merge 用於將 rhs 裡的聚合結果合併到 place

• insertResultInto 用於將 place 裡儲存的聚合結果輸出成 Block

接下來我們以 Sum 這個聚合函式為例子來看一下聚合函式的執行過程。

template <typename T>
using AggregateFunctionSumSimple = 
AggregateFunctionSum<T, typename NearestFieldType<T>::Type, AggregateFunctionSumData<typename NearestFieldType<T>::Type>>;

template <typename T, typename TResult, typename Data>
class AggregateFunctionSum final : public IAggregateFunctionDataHelper<Data, AggregateFunctionSum<T, TResult, Data>>
{
public:
    void add(AggregateDataPtr __restrict place, const IColumn ** columns, size_t row_num, Arena *) const override
    {
        this->data(place).add(column.getData()[row_num]);
    }

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

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

    template <typename U>
    void add(U value) { AggregateFunctionSumAddImpl<T>::add(sum, value); }

    void merge(const AggregateFunctionSumData & rhs) { AggregateFunctionSumAddImpl<T>::add(sum, rhs.sum); }
}

template <typename T>
struct AggregateFunctionSumAddImpl
{
    static void NO_SANITIZE_UNDEFINED ALWAYS_INLINE add(T & lhs, const T & rhs) { lhs += rhs; }
};

• 實際的 Sum 實現帶有很多優化，這裡選擇最簡化的實現

• 對於 Sum，AggregateData 的實現是 AggregateFunctionSumData ，內部維護一個 T Sum 儲存聚合結果

• 對於 Sum， add 和 merge 都是執行 += 操作。

下圖是對 Sum 執行過程的一個簡化描述圖。

add 階段， Aggregator 會從 Input 讀取資料，執行 sum += input ，更新 Aggregate Data 裡的聚合結果。在 add 完成後，會執行 sum0 += sum1 和 sum0 += sum2 ，將聚合結果合併成一個。最終輸出 sum0 作為最終結果。