TiDB Flashback 之 MVCC Query 的实现思路

// planner/core/logical_plan_builder.go

func (b *PlanBuilder) buildDataSource(ctx context.Context, tn *ast.TableName, asName *model.CIStr) (LogicalPlan, error) {
    // ...
    // add _tidb_mvcc_ts to columns.
    mvccTsCol := ds.newExtraMVCCTsSchemaCol()
    ds.Columns = append(ds.Columns, model.NewExtraMVCCTsColInfo())
    schema.Append(mvccTsCol)
    names = append(names, &types.FieldName{
        DBName:      dbName,
        TblName:     tableInfo.Name,
        ColName:     model.ExtraMVCCTsName,
        OrigColName: model.ExtraMVCCTsName,
    })
    ds.TblCols = append(ds.TblCols, mvccTsCol)
    // add _tidb_mvcc_op to columns.
    mvccOpCol := ds.newExtraMVCCOpSchemaCol()
    ds.Columns = append(ds.Columns, model.NewExtraMVCCOpColInfo())
    schema.Append(mvccOpCol)
    names = append(names, &types.FieldName{
        DBName:      dbName,
        TblName:     tableInfo.Name,
        ColName:     model.ExtraMVCCOpName,
        OrigColName: model.ExtraMVCCOpName,
    })
    ds.TblCols = append(ds.TblCols, mvccOpCol)
    // ...
}

// planner/core/expression_rewriter.go

// Leave implements Visitor interface.
func (er *expressionRewriter) Leave(originInNode ast.Node) (retNode ast.Node, ok bool) {
    // ...
    case v := inNode.(type) {
        // ...
        case *ast.ColumnNameExpr:
            if v.Name.Name.L == model.ExtraMVCCOpName.L || v.Name.Name.L == model.ExtraMVCCTsName.L {
                er.sctx.GetSessionVars().NeedMvcc = true
            }
        // ...
    }
    // ...
}

// planner/core/plan_to_pb.go

// ToPB implements PhysicalPlan ToPB interface.
func (p *PhysicalTableScan) ToPB(ctx sessionctx.Context, storeType kv.StoreType) (*tipb.Executor, error) {
    tsExec := tables.BuildTableScanFromInfos(p.Table, p.Columns, ctx.GetSessionVars().NeedMvcc)
    // clear the need mvcc flag
    ctx.GetSessionVars().NeedMvcc = false
    // ...
}

// table/tables/tables.go

// BuildTableScanFromInfos build tipb.TableScan with *model.TableInfo and *model.ColumnInfo.
func BuildTableScanFromInfos(tableInfo *model.TableInfo, columnInfos []*model.ColumnInfo, needMvcc bool) *tipb.TableScan {
    pkColIds := TryGetCommonPkColumnIds(tableInfo)
    tsExec := &tipb.TableScan{
        TableId:          tableInfo.ID,
        Columns:          util.ColumnsToProto(columnInfos, tableInfo.PKIsHandle),
        PrimaryColumnIds: pkColIds,
        NeedMvcc:         needMvcc,
    }
    // ...
}

// tablecodec/tablecodec.go

// EncodeRow encode row data and column ids into a slice of byte.
// valBuf and values pass by caller, for reducing EncodeRow allocates temporary bufs. If you pass valBuf and values as nil,
// EncodeRow will allocate it.
func EncodeRow(sc *stmtctx.StatementContext, row []types.Datum, colIDs []int64, valBuf []byte, values []types.Datum, e *rowcodec.Encoder) ([]byte, error) {
    if len(row) != len(colIDs) {
        return nil, errors.Errorf("EncodeRow error: data and columnID count not match %d vs %d", len(row), len(colIDs))
    }
    // For hackthon: disable the new encoding mode.
    // if e.Enable {
    // 	return e.Encode(sc, colIDs, row, valBuf)
    // }
    return EncodeOldRow(sc, row, colIDs, valBuf, values)
}

//  planner/core/logical_plan_builder.go

func (b *PlanBuilder) buildUpdate(ctx context.Context, update *ast.UpdateStmt) (Plan, error) {
    // ...
    updt := Update{
        OrderedList:               orderedList,
        AllAssignmentsAreConstant: allAssignmentsAreConstant,
        VirtualAssignmentsOffset:  len(update.List),
        RawUpdate:                 b.ctx.GetSessionVars().NeedMvcc,
    }.Init(b.ctx)
    // ...
    return updt, err
}

// executor/update.go 

func (e *UpdateExec) exec(ctx context.Context, schema *expression.Schema, row, newData []types.Datum) error {
    // ...
    commitTs := uint64(0)
    op := byte('P')
    if e.rawUpdate {
        for j, col := range schema.Columns {
            if col.ID == model.ExtraMVCCTsID {
                commitTs = row[j].GetUint64()
            } else if col.ID == model.ExtraMVCCOpID {
                op = row[j].GetString()[0]
            }
        }
    }
    // Update row
    changed, err1 := updateRecord(
        ctx, 
        e.ctx, 
        handle, 
        oldData, 
        newTableData, 
        flags, 
        tbl, 
        false, 
        e.memTracker, 
        e.rawUpdate, 
        commitTs, 
        op)
    // ...
}

// table/tables/tables.go

// RawUpdateRecord is similar to UpdateRecord, but use RawPut.
func (t *TableCommon) RawUpdateRecord(ctx context.Context, sctx sessionctx.Context, h kv.Handle, oldData, newData []types.Datum, touched []bool, commitTs uint64, op byte) error {
    // ...
    key := t.RecordKey(h)
    encodedKey := codec.EncodeBytes([]byte{}, key)
    // compose the writeCF key
    commitTsBytes := make([]byte, 8)
    if commitTs == 0 {
        return nil
    }
    binary.BigEndian.PutUint64(commitTsBytes, ^commitTs)
    encodedKey = append(encodedKey, commitTsBytes...)

    sc, rd := sessVars.StmtCtx, &sessVars.RowEncoder
    value, err := tablecodec.EncodeRow(sc, row, colIDs, nil, nil, rd)
    if err != nil {
        return err
    }
    // construct the writeCF value
    // encoded value: OP _FLAG + START_TS + LONG_VALUE_SUFFIX + LONG_VALUE_LEN + VALUE
    // encoded value: []byte{'P'(or 'D'), VarU64(0)..., 'V', U32(len(value)))..., value...}
    encodedValue := make([]byte, 7)
    encodedValue[0] = op
    encodedValue[1] = 0
    encodedValue[2] = 'V'
    binary.BigEndian.PutUint32(encodedValue[3:7], uint32(len(value)))
    encodedValue = append(encodedValue, value...)

    // Raw input
    addrs := []string{"127.0.0.1:2379"}
    // get pd addr
    if store, ok := sctx.GetStore().(interface{ EtcdAddrs() ([]string, error) }); ok {
        if addrs, err = store.EtcdAddrs(); err != nil {
            return err
        }
    }
    cli, err := rawkv.NewClient(ctx, addrs, config.DefaultConfig().Security)
    if err != nil {
        return err
    }
    err = cli.PutWithCF(ctx, encodedKey, encodedValue, "write")
    if err != nil {
        return err
    }
    // ...
}