摘要：在先前的源碼閱讀系列文章四中，我們介紹了語句，想必大家已經了解了是如何寫入數據，本篇文章介紹一下語句是如何執行。最終語句被解析成結構對于本文所提到的語句會被解析為，被解析為字段，被解析為字段。的實現會不斷獲取每個返回的，把結果寫入。

在先前的 TiDB 源碼閱讀系列文章（四） 中，我們介紹了 Insert 語句，想必大家已經了解了 TiDB 是如何寫入數據，本篇文章介紹一下 Select 語句是如何執行。相比 Insert，Select 語句的執行流程會更復雜，本篇文章會第一次進入優化器、Coprocessor 模塊進行介紹。

表結構沿用上篇文章的：

CREATE TABLE t {
  id   VARCHAR(31),
  name VARCHAR(50),
  age  int,
  key id_idx (id)
};

Select 語句只會講解最簡單的情況：全表掃描+過濾，暫時不考慮索引等復雜情況，更復雜的情況會在后續章節中介紹。語句為：

SELECT name FROM t WHERE age > 10;

相比 Insert 的處理流程，Select 的處理流程中有 3 個明顯的不同：

需要經過 Optimize

Insert 是比較簡單語句，在查詢計劃這塊并不能做什么事情（對于 Insert into Select 語句這種，實際上只對 Select 進行優化），而 Select 語句可能會無比復雜，不同的查詢計劃之間性能天差地別，需要非常仔細的進行優化。

需要和存儲引擎中的計算模塊交互

Insert 語句只涉及對 Key-Value 的 Set 操作，Select 語句可能要查詢大量的數據，如果通過 KV 接口操作存儲引擎，會過于低效，必須要通過計算下推的方式，將計算邏輯發送到存儲節點，就近進行處理。

需要對客戶端返回結果集數據

Insert 語句只需要返回是否成功以及插入了多少行即可，而 Select 語句需要返回結果集。

本篇文章會重點說明這些不同的地方，而相同的步驟會盡量化簡。

Select 語句的語法解析規則在 這里。相比 Insert 語句，要復雜很多，大家可以對著 MySQL 文檔 看一下具體的解析實現。需要特別注意的是 From 字段，這里可能會非常復雜，其語法定義是遞歸的。

最終語句被解析成 ast.SelectStmt 結構：

type SelectStmt struct {
        dmlNode
        resultSetNode
        // SelectStmtOpts wraps around select hints and switches.
        *SelectStmtOpts
        // Distinct represents whether the select has distinct option.
        Distinct bool
        // From is the from clause of the query.
        From *TableRefsClause
        // Where is the where clause in select statement.
        Where ExprNode
        // Fields is the select expression list.
        Fields *FieldList
        // GroupBy is the group by expression list.
        GroupBy *GroupByClause
        // Having is the having condition.
        Having *HavingClause
        // OrderBy is the ordering expression list.
        OrderBy *OrderByClause
        // Limit is the limit clause.
        Limit *Limit
        // LockTp is the lock type
        LockTp SelectLockType
        // TableHints represents the level Optimizer Hint
        TableHints [](#)*TableOptimizerHint
}

對于本文所提到的語句 SELECT name FROM t WHERE age > 10;? name 會被解析為 Fields，WHERE age > 10 被解析為 Where 字段，FROM t 被解析為 From 字段。

在 planBuilder.buildSelect() 方法中，我們可以看到 ast.SelectStmt 是如何轉換成一個 plan 樹，最終的結果是一個 LogicalPlan，每一個語法元素都被轉換成一個邏輯查詢計劃單元，例如 WHERE c > 10 會被處理為一個 plan.LogicalSelection 的結構：

????if sel.Where != nil {
????????p = b.buildSelection(p, sel.Where, nil)
????????if b.err != nil 
????????????return nil
????????}
????}??

具體的結構如下：

// LogicalSelection represents a where or having predicate.
type LogicalSelection struct {
    baseLogicalPlan

    // Originally the WHERE or ON condition is parsed into a single expression,
    // but after we converted to CNF(Conjunctive normal form), it can be
    // split into a list of AND conditions.
    Conditions []expression.Expression
}

其中最重要的就是這個 Conditions 字段，代表了 Where 語句需要計算的表達式，這個表達式求值結果為 True 的時候，表明這一行符合條件。

其他字段的 AST 轉 LogicalPlan 讀者可以執行研究一下，經過個這個 buildSelect() 函數后，AST 變成一個 Plan 的樹狀結構樹，下一步會在這個結構上進行優化。

讓我們回到 plan.Optimize() 函數，Select 語句得到的 Plan 是一個 LogicalPlan，所以 這里 可以進入 doOptimize 這個函數，這個函數比較短，其內容如下：

func doOptimize(flag uint64, logic LogicalPlan) (PhysicalPlan, error) {
    logic, err := logicalOptimize(flag, logic)
    if err != nil {
        return nil, errors.Trace(err)
    }
    if !AllowCartesianProduct && existsCartesianProduct(logic) {
        return nil, errors.Trace(ErrCartesianProductUnsupported)
    }
    physical, err := dagPhysicalOptimize(logic)
    if err != nil {
        return nil, errors.Trace(err)
    }
    finalPlan := eliminatePhysicalProjection(physical)
    return finalPlan, nil
}

大家可以關注來兩個步驟：logicalOptimize 和 dagPhysicalOptimize，分別代表邏輯優化和物理優化，這兩種優化的基本概念和區別本文不會描述，請大家自行研究（這個是數據庫的基礎知識）。下面分別介紹一下這兩個函數做了什么事情。

邏輯優化由一系列優化規則組成，對于這些規則會按順序不斷應用到傳入的 LogicalPlan Tree 中，見 logicalOptimize() 函數：

func logicalOptimize(flag uint64, logic LogicalPlan) (LogicalPlan, error) {
    var err error
    for i, rule := range optRuleList {
        // The order of flags is same as the order of optRule in the list.
        // We use a bitmask to record which opt rules should be used. If the i-th bit is 1, it means we should
        // apply i-th optimizing rule.
        if flag&(1<
目前 TiDB 已經支持下列優化規則：
var optRuleList = []logicalOptRule{
    &columnPruner{}, 
    &maxMinEliminator{},
    &projectionEliminater{},
    &buildKeySolver{},
    &decorrelateSolver{},
    &ppdSolver{},
    &aggregationOptimizer{},
    &pushDownTopNOptimizer{},
}
這些規則并不會考慮數據的分布，直接無腦的操作 Plan 樹，因為大多數規則應用之后，一定會得到更好的 Plan（不過上面有一個規則并不一定會更好，讀者可以想一下是哪個）。
這里選一個規則介紹一下，其他優化規則請讀者自行研究或者是等到后續文章。
columnPruner（列裁剪） 規則，會將不需要的列裁剪掉，考慮這個 SQL: select c from t; 對于 from t 這個全表掃描算子（也可能是索引掃描）來說，只需要對外返回 c 這一列的數據即可，這里就是通過列裁剪這個規則實現，整個 Plan 樹從樹根到葉子節點遞歸調用這個規則，每層節點只保留上面節點所需要的列即可。
經過邏輯優化，我們可以得到這樣一個查詢計劃：

其中 FROM t 變成了 DataSource 算子，WHERE age > 10 變成了 Selection 算子，這里留一個思考題，SELECT name 中的列選擇去哪里了？
物理優化
在物理優化階段，會考慮數據的分布，決定如何選擇物理算子，比如對于 FROM t WHERE age > 10 這個語句，假設在 age 字段上有索引，需要考慮是通過 TableScan + Filter 的方式快還是通過 IndexScan 的方式比較快，這個選擇取決于統計信息，也就是 age > 10 這個條件究竟能過濾掉多少數據。
我們看一下 dagPhysicalOptimize 這個函數：
func dagPhysicalOptimize(logic LogicalPlan) (PhysicalPlan, error) {
    logic.preparePossibleProperties()
    logic.deriveStats()
    t, err := logic.convert2PhysicalPlan(&requiredProp{taskTp: rootTaskType, expectedCnt: math.MaxFloat64})
    if err != nil {
        return nil, errors.Trace(err)
    }
    p := t.plan()
    p.ResolveIndices()
    return p, nil
}
這里的 convert2PhysicalPlan 會遞歸調用下層節點的 convert2PhysicalPlan 方法，生成物理算子并且估算其代價，然后從中選擇代價最小的方案，這兩個函數比較重要：
// convert2PhysicalPlan implements LogicalPlan interface.
func (p *baseLogicalPlan) convert2PhysicalPlan(prop *requiredProp) (t task, err error) {
    // Look up the task with this prop in the task map.
    // It"s used to reduce double counting.
    t = p.getTask(prop)
    if t != nil {
        return t, nil
    }
    t = invalidTask
    if prop.taskTp != rootTaskType {
        // Currently all plan cannot totally push down.
        p.storeTask(prop, t)
        return t, nil
    }
    for _, pp := range p.self.genPhysPlansByReqProp(prop) {
        t, err = p.getBestTask(t, pp)
        if err != nil {
            return nil, errors.Trace(err)
        }
    }
    p.storeTask(prop, t)
    return t, nil
}

func (p *baseLogicalPlan) getBestTask(bestTask task, pp PhysicalPlan) (task, error) {
    tasks := make([]task, 0, len(p.children))
    for i, child := range p.children {
        childTask, err := child.convert2PhysicalPlan(pp.getChildReqProps(i))
        if err != nil {
            return nil, errors.Trace(err)
        }
        tasks = append(tasks, childTask)
    }
    resultTask := pp.attach2Task(tasks...)
    if resultTask.cost() < bestTask.cost() {
        bestTask = resultTask
    }
    return bestTask, nil
}
上面兩個方法的返回值都是一個叫 task 的結構，而不是物理計劃，這里引入一個概念，叫 Task，TiDB 的優化器會將 PhysicalPlan 打包成為 Task。Task 的定義在 task.go 中，我們看一下注釋：
// task is a new version of `PhysicalPlanInfo`. It stores cost information for a task.
// A task may be CopTask, RootTask, MPPTask or a ParallelTask.
type task interface {
    count() float64
    addCost(cost float64)
    cost() float64
    copy() task
    plan() PhysicalPlan
    invalid() bool
}
在 TiDB 中，Task 的定義是能在單個節點上不依賴于和其他節點進行數據交換即可進行的一些列操作，目前只實現了兩種 Task：

CopTask 是需要下推到存儲引擎（TiKV）上進行計算的物理計劃，每個收到請求的 TiKV 節點都會做相同的操作
RootTask 是保留在 TiDB 中進行計算的那部分物理計劃

如果了解過 TiDB 的 Explain 結果，那么可以看到每個 Operator 都會標明屬于哪種 Task，比如下面這個例子：

整個流程是一個樹形動態規劃的算法，大家有興趣可以跟一下相關的代碼自行研究或者等待后續的文章。
經過整個優化過程，我們已經得到一個物理查詢計劃，這個 SELECT name FROM t WHERE age > 10; 語句能夠指定出來的查詢計劃大概是這樣子的：

讀者可能會比較奇怪，為什么只剩下這樣一個這一個物理算子？WHERR age > 10 哪里去了？實際上 age > 10 這個過濾條件被合并進了 PhysicalTableScan，因為 age > 10 這個表達式可以下推到 TiKV 上進行計算，所以會把 TableScan 和 Filter 這樣兩個操作合在一起。哪些表達式會被下推到 TiKV 上的 Coprocessor 模塊進行計算呢？對于這個 Query 是在下面 這個地方 進行識別：
// PredicatePushDown implements LogicalPlan PredicatePushDown interface.
func (ds *DataSource) PredicatePushDown(predicates []expression.Expression) ([]expression.Expression, LogicalPlan) {
    _, ds.pushedDownConds, predicates = expression.ExpressionsToPB(ds.ctx.GetSessionVars().StmtCtx, predicates, ds.ctx.GetClient())
    return predicates, ds
}
在 expression.ExpressionsToPB 這個方法中，會把能下推 TiKV 上的表達式識別出來（TiKV 還沒有實現所有的表達式，特別是內建函數只實現了一部分），放到 DataSource.pushedDownConds 字段中。接下來我們看一下 DataSource 是如何轉成 PhysicalTableScan，見 DataSource.convertToTableScan() 方法。這個方法會構建出 PhysicalTableScan，并且調用 addPushDownSelection() 方法，將一個 PhysicalSelection 加到 PhysicalTableScan 之上，一起放進 copTask 中。
這個查詢計劃是一個非常簡單計劃，不過我們可以用這個計劃來說明 TiDB 是如何執行查詢操作。
Executing
一個查詢計劃如何變成一個可執行的結構以及如何驅動這個結構執行查詢已經在前面的兩篇文章中做了描述，這里不再敷述，這一節我會重點介紹如何具體的執行過程以及 TiDB 的分布式執行框架。
Coprocessor 框架
Coprocessor 這個概念是從 HBase 中借鑒而來，簡單來說是一段注入在存儲引擎中的計算邏輯，等待 SQL 層發來的計算請求（序列化后的物理執行計劃），處理本地數據并返回計算結果。在 TiDB 中，計算是以 Region 為單位進行，SQ
L 層會分析出要處理的數據的 Key Range，再將這些 Key Range 根據 PD 中拿到的 Region 信息劃分成若干個 Key Range，最后將這些請求發往對應的 Region。
SQL 層會將多個 Region 返回的結果進行匯總，在經過所需的 Operator 處理，生成最終的結果集。
DistSQL
請求的分發與匯總會有很多復雜的處理邏輯，比如出錯重試、獲取路由信息、控制并發度以及結果返回順序，為了避免這些復雜的邏輯與 SQL 層耦合在一起，TiDB 抽象了一個統一的分布式查詢接口，稱為 DistSQL API，位于 distsql 這個包中。
其中最重要的方法是 SelectDAG 這個函數：
// SelectDAG sends a DAG request, returns SelectResult.
// In kvReq, KeyRanges is required, Concurrency/KeepOrder/Desc/IsolationLevel/Priority are optional.
func SelectDAG(goCtx goctx.Context, ctx context.Context, kvReq *kv.Request, fieldTypes []*types.FieldType) (SelectResult, error) {
    // kvReq 中包含了計算所涉及的數據的 KeyRanges
    // 這里通過 TiKV Client 向 TiKV 集群發送計算請求
    resp := ctx.GetClient().Send(goCtx, kvReq)
    if resp == nil {
        err := errors.New("client returns nil response")
        return nil, errors.Trace(err)
    }

    if kvReq.Streaming {
        return &streamResult{
            resp:       resp,
            rowLen:     len(fieldTypes),
            fieldTypes: fieldTypes,
            ctx:        ctx,
        }, nil
    }
    // 這里將結果進行了封裝
    return &selectResult{
        label:      "dag",
        resp:       resp,
        results:    make(chan newResultWithErr, kvReq.Concurrency),
        closed:     make(chan struct{}),
        rowLen:     len(fieldTypes),
        fieldTypes: fieldTypes,
        ctx:        ctx,
    }, nil
}
TiKV Client 中的具體邏輯我們暫時跳過，這里只關注 SQL 層拿到了這個 selectResult 后如何讀取數據，下面這個接口是關鍵。
// SelectResult is an iterator of coprocessor partial results.
type SelectResult interface {
    // NextRaw gets the next raw result.
    NextRaw(goctx.Context) ([]byte, error)
    // NextChunk reads the data into chunk.
    NextChunk(goctx.Context, *chunk.Chunk) error
    // Close closes the iterator.
    Close() error
    // Fetch fetches partial results from client.
    // The caller should call SetFields() before call Fetch().
    Fetch(goctx.Context)
    // ScanKeys gets the total scan row count.
    ScanKeys() int64
selectResult 實現了 SelectResult 這個接口，代表了一次查詢的所有結果的抽象，計算是以 Region 為單位進行，所以這里全部結果會包含所有涉及到的 Region 的結果。調用 Chunk 方法可以讀到一個 Chunk 的數據，通過不斷調用 NextChunk 方法，直到 Chunk 的 NumRows 返回 0 就能拿到所有結果。NextChunk 的實現會不斷獲取每個 Region 返回的 SelectResponse，把結果寫入 Chunk。
Root Executor
能推送到 TiKV 上的計算請求目前有 TableScan、IndexScan、Selection、TopN、Limit、PartialAggregation 這樣幾個，其他更復雜的算子，還是需要在單個 tidb-server 上進行處理。所以整個計算是一個多 tikv-server 并行處理 + 單個 tidb-server 進行匯總的模式。
總結
Select 語句的處理過程中最復雜的地方有兩點，一個是查詢優化，一個是如何分布式地執行，這兩部分后續都會有文章來更進一步介紹。下一篇文章會脫離具體的 SQL 邏輯，介紹一下如何看懂某一個特定的模塊。
作者：申礫