(key,start_ts) ==> (primary_key,lock_type)

(key,commit_ts) ==> (start_ts)

(key,start_ts) ==> (value)

class Transaction {    struct Write{ Row row; Column: col; string value;};    vector<Write> writes_;    int start_ts_;
    Transaction():start_ts_(oracle.GetTimestamp()) {}        // 往事务中添加一行数据    void Set(Write w) {writes_.push_back(w);}        bool Get(Row row, Column c, string* value) {        while(true) {                    // 开启bigtable事务            bigtable::Txn = bigtable::StartRowTransaction(row);                        // Check for locks that signal concurrent writes.            // 检查[0, start_ts]内是否有锁，有则等待再重试            if (T.Read(row, c+"locks", [0, start_ts_])) {                // There is a pending lock; try to clean it and wait                BackoffAndMaybeCleanupLock(row, c);                continue;            }        }
        // Find the latest write below our start_timestamp.        // 读取column:wriet的[0, start_ts]的最新提交记录版本        latest_write = T.Read(row, c+"write", [0, start_ts_]);        if(!latest_write.found()) return false; // no data                // 根据此时间戳版本去column:data读取数据        int data_ts = latest_write.start_timestamp();        *value = T.Read(row, c+"data", [data_ts, data_ts]);        return true;    }        // prewrite : 尝试上锁+写数据，如果锁冲突则返回失败    bool Prewrite(Write w, Write primary) {        Column c = w.col;                // bigtable 事务        bigtable::Txn T = bigtable::StartRowTransaction(w.row);
        // 检查写写冲突：从column:write获取最新数据，如果commit_ts>=事务开始时间，则冲突        if (T.Read(w.row, c+"write", [start_ts_, max])) return false;                // 检查column:lock，如果存在已存在，则锁冲突        if (T.Read(w.row, c+"lock", [0, max])) return false;                // 往column:lock写入锁，若为secondary，则指向primary        T.Write(w.row, c+"lock", start_ts_, {primary.row, primary.col})                // 往column:data写入(key, start_ts = value)        T.Write(w.row, c+"data", start_ts_, w.value);                // 提交bigtable事务        return T.Commit();    }        // 此commit整合 2PC的prewrite和commit，    // 对外caller而言不需要关注是2PC提交，直接commit就好    bool Commit() {        // ***** Prewrite *****//        // 2PC 第一阶段：prewrite        // 第一行为primary        Write primary = write_[0];        // 其余为secondary        vector<Write> secondaries(write_.begin() + 1, write_.end());                if (!Prewrite(primary, primary)) return false;        for (Write w : secondaries)            if (!Prewrite(w, primary)) return false;                // ***** Commit *****//        // 2PC第二阶段：commit                // 从oracle获取commit timestamp        int commit_ts = oracle.GetTimestamp();
        // Commit primary first.         Write p = primary;                // primary : 先开始bigtable的事务        bigtable::Txn T = bigtable::StartRowTransaction(p.row);                // primary : 如果primiary的锁不存在，则abort        if (!T.Read(p.row, p.col+"lock", [start_ts_, start_ts_]))            return false; // aborted while working                // primary : 在column:write 列写入开始时间（Pointer to data written at start_ts_）        T.Write(p.row, p.col+"write", commit_ts, start_ts_);         // primary : 移除column:lock 列        T.Erase(p.row, p.col+"lock", commit_ts);                // primary : 提交bigtable 事务        if(!T.Commit()) return false;
        // Second phase: write our write records for secondary cells.        // 遍历所有secondary，写入column:write数据同时删除column:lock        for (Write w:secondaries) {            bigtable::write(w.row, w.col+"write", commit_ts, start_ts_);            bigtable::Erase(w.row, w.col+"lock", commit_ts);        }        return true;    }};

(key, start_ts) ==> value

key ==> lock_info

(key, commit_ts) ==> write_info

start_ts = tidb->pd.GetTso() // get start_ts
// tidb调用tikv prewrite接口 tidb->tikv.Prewrite(start_ts, data_list){ // tikv prewrite实现    keyIsLockedArray = []        // prewrite each key with start_ts in memory，中间出现失败，则整个prewrite失败    for key in data_list     {        // check write conflict: 通过raft获取key的数据       record = raft.Get(WriteColumn, key, start_ts)       if record.commit_ts >= start_ts       {           return error(write conflict, END)       }              // check lock       lock = raft.Get(LockColumn, key)       if lock != null && lock.ts != start_ts // lock已存在且为其他tx的锁       {           keyIsLockedArray.append(key)           continue       }       // 往内存中的 lock 列写入 lock(start_ts,key) 为当前key加锁,       //   若当前key被选为 primary, 则标记为 primary,       //   若为secondary,则标明指向primary的信息。       memory.Put(LockColumn, key, start_ts, (primary|secondary), ttl, short_value)       memory.Put(DataColumn, key, start_ts, long_value)    }    if len(keyIsLockedArray) > 0    {        return error(keyIsLockedArray)    }    // 将此事务在内存模型中写入的数据 持久化到raft中    raft.Commit(memory_data)    return ok}

// tidb调用rikv的Commit接口，进行2PC的Commit阶段tidb->tikv.Commit(keys, start_ts, commit_ts){    for key in keys // do commit    {        lock = raft.Get(LockColumn, key)        // lock存在且匹配，则提交        if lock != null && lock.ts == start_ts         {            memory.Put(WriteColumn, key, commit_ts, start_ts)            memory.Del(LockColumn, key, start_ts)           }       // lock does not exist or tx dismatch       else if lock == null || lock.ts != start_ts        {           record = raft.Get(WriteColumn, key, start_ts, commit_ts)           if record != null && record.write_type == (PUT|DELETE|Lock)           {               continue; // already commited           } else if record == null || record.write_type == RollBack           {               return error(tx conflict, lock not exist)           }       }           }    // commit to raft    raft.Save(memory.data)    return ok}

// tidb调用tikv的Rollback接口tidb->tikv.Rollback(keys){   // Rollback接口实现
    // 检查合法性    for key in keys    {        // 检查当前key的锁        lock=memory.GetLockColumn(start_ts, key)        if lock != null and lock.ts = start_ts        {   // 如果锁还存在且是之前的锁，则删除锁，写入的数据            // 且在WriteColumn写入rollback记录防止后面commit请求的到来            memory.Del(DataColumn, key, start_ts)            memory.Put(WriteColumn, key, start_ts, rollback)            meomry.Del(LockColumn, key, start_ts)            continue        }               // 检查提交情况        record = raft.Get(WriteColumn, key, start_ts)        if record != null          {           if record.status == (PUT|DELETE)           {               return error(transaction is already commited.)           } else if record.status == RollBack           {               continue;  // already rollbacked           }        } else { // record is null           // 提交纪录不存在，说明当前 key 尚未被 prewrite 过，           // 为预防 prewrite 在rollback之后过来(可能网络原因)，           // 在这里留下 (key,start_ts,rollback)记录           memory.Put(WriteColumn, key, start_ts, rollback)           continue        }                // persist        raft->Save(memory.data)    }    return ok}

// tidb调用tikv的ResolveLock接口tidb->tikv.ResolveLock(start_ts, commit_ts){    // 找出所有 lock.ts==start_ts 的锁并执行清锁操作    locks = raft.Scan(LockColumn, lock.ts = start_ts)    for (lock in locks)    {        // commit_ts存在，则说明已提交        if (commit_ts != null)        {            // 对已上锁的key进行提交            memory.Commit(lock.key, commit_ts)        } else {            memory.Rollback(lock.key)        }           }    raft->Save(memory.data)        return ok}

// tidb调用tikv的Get接口tidb->tikv.Get(key, start_ts){    // check lock: 如果锁存在且在此之前加的锁，则返回锁冲突    lock = raft->Get(LockColumn, key)    if (lock != null && lock.ts <= start_ts)        return error(isLocked);    version = start_ts - 1;    RAFTGET:    // get writeColumn： 获取小于start_ts的最新提交记录    data = raft->Get(WriteColumn, key, version)    if (data != null)    {        if (data.writeType == PUT)        {            if (data.isShortValue)                return data.shortValue;            // long value            return raft->Get(DataColumn, key, start_ts)        } else if (data.writeType == DELETE)        {            // 没有出现过该值，或该值最近已被删除，返回tidb空            return ok(None)；        } else if (data.writeType == "LOCK | ROLLBACK")        {            // version=commit_ts-1, 继续查找下一个最近版本            version=commit_ts-1            goto RAFTGET        }    } else {        return ok(None)    }}

// tidb 向 tikv 发起 GC操作，要求清理 safe-point 版本之前的所有无意义版本
// tidb调用tikv的GC接口tidb->tikv.Gc(savePoint){    startKey=null;    for {        // scan (startKey, maxKey]        keys = raft.Scan(WriteColumn, startKey, batchSize)        if (batch == null)            return ok;                // 对每个key进行gc操作        for (key in keys)        {            bool remove_older = false;                        // 清理write中commit_ts <= startPoint的每个item            // item: 一个key的所有write记录            for (item in key)            {                if (remove_older is true)                {                    memory.Del(WriteColumn, item);                    memory.Del(DataColumn, item);                }                                    // check if is ordest version to save                if (writeType == PUT)                {   // 保留该提交，清理所有该提交之前的数据                    memory.Set(memoryOrder, true);                } else if (writeType == DELETE)                {   //  清理所有safe-point 之前的数据                    remove_older = true;                    memory.Del(WriteColumn, item);                } else if (writeType == ROLLBACK | LOCK)                {   // 清理所有小于 safe-point 的 Rollback 和 Lock                    memory.Del(WriteColumn, item);                }            }        }        startKey = keys[keys.length() - 1];    }}

max{start_ts, max_read_ts_of_written_keys} < commit_ts <= now

commit_ts = max{start_ts, region_1_max_read_ts, region_2_max_read_ts, ...} + 1