摘要：在事務提交結束之后，事務可能提交成功，也可能提交失敗。需要把這個狀態(tài)告知如果發(fā)生了，那么輸出的類型就為，如果成功提交，那么輸出的類型就為。，當完成自己所有的狀態(tài)變更之后，會把的狀態(tài)改為。

作者：姚維

這篇文章不是講 TiDB Binlog 組件的源碼，而是講 TiDB 在執(zhí)行 DML/DDL 語句過程中，如何將 Binlog 數據 發(fā)送給 TiDB Binlog 集群的 Pump 組件。目前 TiDB 在 DML 上的 Binlog 用的類似 Row-based 的格式。具體 Binlog 具體的架構細節(jié)可以參考這篇 文章。

這里只描述 TiDB 中的代碼實現(xiàn)。

TiDB 采用 protobuf 來編碼 binlog，具體的格式可以見 binlog.proto。這里討論 TiDB 寫 Binlog 的機制，以及 Binlog 對 TiDB 寫入的影響。

TiDB 會在 DML 語句提交，以及 DDL 語句完成的時候，向 pump 輸出 Binlog。

DML 語句包括 Insert/Replace、Update、Delete，這里挑 Insert 語句來闡述，其他的語句行為都類似。首先在 Insert 語句執(zhí)行完插入（未提交）之前，會把自己新增的數據記錄在 binlog.TableMutation 結構體中。

// TableMutation 存儲表中數據的變化
message TableMutation {
        // 表的 id，唯一標識一個表
        optional int64 table_id      = 1 [(gogoproto.nullable) = false]; 
        
        // 保存插入的每行數據
        repeated bytes inserted_rows = 2;
        
        // 保存修改前和修改后的每行的數據
        repeated bytes updated_rows  = 3;
        
        // 已廢棄
        repeated int64 deleted_ids   = 4;
        
        // 已廢棄
        repeated bytes deleted_pks   = 5;
         
        // 刪除行的數據
        repeated bytes deleted_rows  = 6;
        
        // 記錄數據變更的順序
        repeated MutationType sequence = 7;
}

這個結構體保存于跟每個 Session 鏈接相關的事務上下文結構體中 TxnState.mutations。 一張表對應一個 TableMutation 對象，TableMutation 里面保存了這個事務對這張表的所有變更數據。Insert 會把當前語句插入的行，根據 RowID + Row-value 的格式編碼之后，追加到 TableMutation.InsertedRows 中：

func (t *Table) addInsertBinlog(ctx context.Context, h int64, row []types.Datum, colIDs []int64) error {
    mutation := t.getMutation(ctx)
    pk, err := codec.EncodeValue(ctx.GetSessionVars().StmtCtx, nil, types.NewIntDatum(h))
    if err != nil {
        return errors.Trace(err)
    }
    value, err := tablecodec.EncodeRow(ctx.GetSessionVars().StmtCtx, row, colIDs, nil, nil)
    if err != nil {
        return errors.Trace(err)
    }
    bin := append(pk, value...)
    mutation.InsertedRows = append(mutation.InsertedRows, bin)
    mutation.Sequence = append(mutation.Sequence, binlog.MutationType_Insert)
    return nil
}

等到所有的語句都執(zhí)行完之后，在 TxnState.mutations 中就保存了當前事務對所有表的變更數據。

對于 DML 而言，TiDB 的事務采用 2-phase-commit 算法，一次事務提交會分為 Prewrite 階段，以及 Commit 階段。這里分兩個階段來看看 TiDB 具體的行為。

在 session.doCommit 函數中，TiDB 會構造 binlog.PrewriteValue：

message PrewriteValue {
    optional int64         schema_version = 1 [(gogoproto.nullable) = false];
    repeated TableMutation mutations      = 2 [(gogoproto.nullable) = false];
}

這個 PrewriteValue 中包含了跟這次變動相關的所有行數據，TiDB 會填充一個類型為 binlog.BinlogType_Prewrite 的 Binlog：

info := &binloginfo.BinlogInfo{
    Data: &binlog.Binlog{
        Tp:            binlog.BinlogType_Prewrite,
        PrewriteValue: prewriteData,
    },
    Client: s.sessionVars.BinlogClient.(binlog.PumpClient),
}

TiDB 這里用一個事務的 Option kv.BinlogInfo 來把 BinlogInfo 綁定到當前要提交的 transaction 對象中：

s.txn.SetOption(kv.BinlogInfo, info)

在 twoPhaseCommitter.execute 中，在把數據 prewrite 到 TiKV 的同時，會調用 twoPhaseCommitter.prewriteBinlog，這里會把關聯(lián)的 binloginfo.BinlogInfo 取出來，把 Binlog 的 binlog.PrewriteValue 輸出到 Pump。

binlogChan := c.prewriteBinlog()
err := c.prewriteKeys(NewBackoffer(prewriteMaxBackoff, ctx), c.keys)
if binlogChan != nil {
    binlogErr := <-binlogChan // 等待 write prewrite binlog 完成
    if binlogErr != nil {
        return errors.Trace(binlogErr)
    }
}

這里值得注意的是，在 prewrite 階段，是需要等待 write prewrite binlog 完成之后，才能繼續(xù)做接下去的提交的，這里是為了保證 TiDB 成功提交的事務，Pump 至少一定能收到 Prewrite Binlog。

在 twoPhaseCommitter.execute 事務提交結束之后，事務可能提交成功，也可能提交失敗。TiDB 需要把這個狀態(tài)告知 Pump：

err = committer.execute(ctx)
if err != nil {
    committer.writeFinishBinlog(binlog.BinlogType_Rollback, 0)
    return errors.Trace(err)
}
committer.writeFinishBinlog(binlog.BinlogType_Commit, int64(committer.commitTS))

如果發(fā)生了 error，那么輸出的 Binlog 類型就為 binlog.BinlogType_Rollback，如果成功提交，那么輸出的 Binlog 類型就為 binlog.BinlogType_Commit。

func (c *twoPhaseCommitter) writeFinishBinlog(tp binlog.BinlogType, commitTS int64) {
    if !c.shouldWriteBinlog() {
        return
    }
    binInfo := c.txn.us.GetOption(kv.BinlogInfo).(*binloginfo.BinlogInfo)
    binInfo.Data.Tp = tp
    binInfo.Data.CommitTs = commitTS
    go func() {
        err := binInfo.WriteBinlog(c.store.clusterID)
        if err != nil {
            log.Errorf("failed to write binlog: %v", err)
        }
    }()
}

值得注意的是，這里 WriteBinlog 是多帶帶啟動 goroutine 異步完成的，也就是 Commit 階段，是不再需要等待寫 binlog 完成的。這里可以節(jié)省一點 commit 的等待時間，這里不需要等待是因為 Pump 即使接收不到這個 Commit Binlog，在超過 timeout 時間后，Pump 會自行根據 Prewrite Binlog 到 TiKV 中確認當條事務的提交狀態(tài)。

一個 DDL 有如下幾個狀態(tài)：

const (
    JobStateNone            JobState = 0
    JobStateRunning         JobState = 1
    JobStateRollingback      JobState = 2
    JobStateRollbackDone     JobState = 3
    JobStateDone             JobState = 4
    JobStateSynced             JobState = 6
    JobStateCancelling         JobState = 7
)

這些狀態(tài)代表了一個 DDL 任務所處的狀態(tài)：

JobStateNone，代表 DDL 任務還在處理隊列，TiDB 還沒有開始做這個 DDL。

JobStateRunning，當 DDL Owner 開始處理這個任務的時候，會把狀態(tài)設置為 JobStateRunning，之后 DDL 會開始變更，TiDB 的 Schema 可能會涉及多個狀態(tài)的變更，這中間不會改變 DDL job 的狀態(tài)，只會變更 Schema 的狀態(tài)。

JobStateDone， 當 TiDB 完成自己所有的 Schema 狀態(tài)變更之后，會把 Job 的狀態(tài)改為 Done。

JobStateSynced，當 TiDB 每做一次 schema 狀態(tài)變更，就會需要跟集群中的其他 TiDB 做一次同步，但是當 Job 狀態(tài)為 JobStateDone 之后，在 TiDB 等到所有的 TiDB 節(jié)點同步之后，會將狀態(tài)修改為 JobStateSynced。

JobStateCancelling，TiDB 提供語法 ADMIN CANCEL DDL JOBS job_ids 用于取消某個正在執(zhí)行或者還未執(zhí)行的 DDL 任務，當成功執(zhí)行這個命令之后，DDL 任務的狀態(tài)會變?yōu)?JobStateCancelling。

JobStateRollingback，當 DDL Owner 發(fā)現(xiàn) Job 的狀態(tài)變?yōu)?JobStateCancelling 之后，它會將 job 的狀態(tài)改變?yōu)?JobStateRollingback，以示已經開始處理 cancel 請求。

JobStateRollbackDone，在做 cancel 的過程，也會涉及 Schema 狀態(tài)的變更，也需要經歷 Schema 的同步，等到狀態(tài)回滾已經做完了，TiDB 會將 Job 的狀態(tài)設置為 JobStateRollbackDone。

對于 Binlog 而言，DDL 的 Binlog 輸出機制，跟 DML 語句也是類似的，只有開始處理事務提交階段，才會開始寫 Binlog 出去。那么對于 DDL 來說，跟 DML 不一樣，DML 有事務的概念，對于 DDL 來說，SQL 的事務是不影響 DDL 語句的。但是 DDL 里面，上面提到的 Job 的狀態(tài)變更，是作為一個事務來提交的（保證狀態(tài)一致性）。所以在每個狀態(tài)變更，都會有一個事務與之對應，但是上面提到的中間狀態(tài)，DDL 并不會往外寫 Binlog，只有 JobStateRollbackDone 以及 JobStateDone 這兩種狀態(tài)，TiDB 會認為 DDL 語句已經完成，會對外發(fā)送 Binlog，發(fā)送之前，會把 Job 的狀態(tài)從 JobStateDone 修改為 JobStateSynced，這次修改，也涉及一次事務提交。這塊邏輯的代碼如下：

worker.handleDDLJobQueue():

if job.IsDone() || job.IsRollbackDone() {
        binloginfo.SetDDLBinlog(d.binlogCli, txn, job.ID, job.Query)
        if !job.IsRollbackDone() {
            job.State = model.JobStateSynced
        }
        err = w.finishDDLJob(t, job)
        return errors.Trace(err)
}

type Binlog struct {
    DdlQuery []byte
    DdlJobId         int64
}

DdlQuery 會設置為原始的 DDL 語句，DdlJobId 會設置為 DDL 的任務 ID。

對于最后一次 Job 狀態(tài)的提交，會有兩條 Binlog 與之對應，這里有幾種情況：

如果事務提交成功，類型分別為 binlog.BinlogType_Prewrite 和 binlog.BinlogType_Commit。

如果事務提交失敗，類型分別為 binlog.BinlogType_Prewrite 和 binlog.BinlogType_Rollback。

所以，Pumps 收到的 DDL Binlog，如果類型為 binlog.BinlogType_Rollback 應該只認為如下狀態(tài)是合法的：

JobStateDone （因為修改為 JobStateSynced 還未成功）

JobStateRollbackDone

如果類型為 binlog.BinlogType_Commit，應該只認為如下狀態(tài)是合法的：

JobStateSynced

JobStateRollbackDone

當 TiDB 在提交最后一個 Job 狀態(tài)的時候，如果事務提交失敗了，那么 TiDB Owner 會嘗試繼續(xù)修改這個 Job，直到成功。也就是對于同一個 DdlJobId，后續(xù)還可能會有多次 Binlog，直到出現(xiàn) binlog.BinlogType_Commit。