原文地址：https://ikenchina.github.io/

论文

论文地址
https://research.google/pubs/pub36726/

概述

设计percolator的目的
为大数据集群进行增量处理更新的系统，主要用于google网页搜索索引服务。使用基于Percolator的增量处理系统代替原有的批处理索引系统后，Google在处理同样数据量的文档时，将文档的平均搜索延时降低了50%。

基于bigtable单行事务实现跨行事务
Percolator 在 Bigtable 之上实现的，以client library 的方式实现。
Percolator 利用 Bigtable 的单行事务能力，依靠client的协议和一个全局的授时服务器 TSO 以及两阶段提交协议来实现了跨机器的多行事务。

MVCC与snapshot isolation
Percolator依据全局时间戳和MVCC实现 Snapshot Isolation 隔离级别的并发控制协议。

percolator特点

事务: 跨行、跨表的、基于快照隔离的ACID事务
观察者(observers)：一种类似触发器的通知机制

设计

隔离等级

通过MVCC来实现SI(Snapshop isolation)隔离等级。

Percolator 使用Bigtable的时间戳记维度实现了数据的多版本化。优点如下：

读操作：可以读取任何指定时间戳版本的记录
写操作，能很好的应对写写冲突：若两个事务操作同一记录，只有一个会提交成功

I存在write skew(写偏斜)

锁机制

Because it is built as a client library accessing Bigtable, rather than controlling access to storage itself, Percolator faces a different set of challenges implementing distributed transactions than traditional PDBMSs. Other parallel databases integrate locking into the system component that manages access to the disk: since each node already mediates access to data on the disk it can grant locks on requests and deny accesses that violate locking requirements.

Percolator锁的管理必须满足以下条件：

能应对机器故障：若一个锁在两阶段提交时消失，系统可能将两个有冲突的事务都提交
高吞吐量：上千台机器会同时请求获取锁
低延时

锁服务要实现：

多副本： survive failure
distributed and balanced ： handle load
写入持久化存储系统

时间戳

Timestamp Oracle(不是Oracle数据库)： TSO通过统一中心授权可以保证按照递增的方式分配逻辑时钟,任何事件申请的时钟都不会重复，能够保证事务版本号的单调递增，确保分布式事务的时序。

所以TSO是一个分配严格的单调递增时间戳的服务器。

优化
因为每个事务都需要调用oracle两次，所以这个服务必须有很好的可伸缩性。
Oracle会定期分配一个范围的时间戳，然后将范围中的最大值写入持久化，Oracle在内存中原子递增来快速分配时间戳，查询时不涉及磁盘I/O。如果oracle重启，将以存储中的最大值作为开始值。
worker会维持一个长连接RPC到oracle，低频率的、批量的获取时间戳。

性能
Oracle中单台机器每秒向外分配接近两百万的时间戳。

关于批量获取时间戳

批量获取时间戳并不会造成乱序问题，因为就算事务A先获取时间戳T1，事务B后获取时间戳T2，T1<T2，那么分布式系统中，也无法保证事务A先执行，事务B后执行。
如果事务B先执行，那么事务A势必能发现写冲突从而rollback。

单点

为了保证单调递增的特性，所以很多TSO的开源实现都存在单点问题。如tidb的TSO。
而且，一般TSO也存在跨数据中心高延迟的问题。

其他时序方案

Logic Clock: dynamoDB
True Time : spanner
Hybrid Logic Clock : cockroachDB，没有单点问题，但是为了解决时钟误差而无法避免的时延问题。

数据存储

percolator定义了5个列

Column	Use
c:lock	An uncommitted transaction is writing this cell; contains the location of primary lock
c:write	Committed data present; stores the Bigtable timestamp of the data
c:data	Stores the data itself
c:notify Hint:	observers may need to run
c:ack	O Observer “O” has run ; stores start timestamp of successful last run

Lock

事务的锁，key value映射

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

key：数据的key
start_ts：事务开始时间
primary：该锁的primary的引用。事务从待修改的keys中选择一个作为primary,其余的则作为secondary，secondary的primary_key指向primary的key，事务的上锁和解锁都由primary key决定。

Write
已提交的数据对应的时间戳。key value映射

(key,commit_ts) ==> (start_ts)

key：数据的key
commit_ts：事务的提交时间
start_ts：事务的开始时间（此数据在data中的时间戳版本）

Data

存储数据的列，key value映射

(key,start_ts) ==> (value)

key：对应的主键
start_ts：事务的开始时间
value：除主键外的数据列

Notify
notify列仅仅是一个hint值（可能是个bool值），表示是否需要触发通知。

Ack
ack列是一个简单的时间戳值，表示最近执行通知的观察者的开始时间。

案例

以银行转账为案例
Bob 向 Joe 转账7元。
事务开始时间：start timestamp =7 ，提交时间：commit timestamp=8。

percolator-demo1.png

Bob有10元：查询column write获取最新时间戳版本的数据(data@5),然后从column data里面获取时间戳为5的数据( $10），Joe($ 2)

percolator-demo2.png

stat timestamp=7 作为当前事务的开始时间戳，将Bob选为此事务的primary key，再写入column:lock对Bob上锁，同时将column:data列更新为7:$3。

percolator-demo3.png

start timestamp=7作为锁定Joe账户的时间戳，更新其column:data为$9，其锁是secondary指向primary

percolator-demo4.png

当前时间戳commit timestamp=8作为事务提交时间戳：删除primary所在的lock，在write列中写入commit_ts：data@7

percolator-demo5.png

在所有secondary中写入column:write且清理column:lock，事务完成。

流程

伪代码

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;
    }
};

事务

第一阶段
- 获取时间戳T1,
- 写入column:data和锁：时间戳都为T1
第二阶段
- 获取时间戳T2
- 写入column:write：key:commit_ts:start_ts (key:T2:T1)
- 删除锁

读取

获取时间戳Tx
从column:write 读取key[0, Tx]的最大时间戳数据(获取到事务写入的commit_ts=T2)
从T2中提取出start_ts为T1
从column:data中读取 key:T1的数据

所以读取到的数据是commit时间戳的数据。

清理锁

若客户端在Commit一个事务时，出现了异常，Prepare时产生的锁会被留下。为避免将新事务挂住，Percolator必须清理这些锁。

Percolator用lazy方式来处理未处理的锁：当事务在执行时，发现其他事务造成的锁未处理掉，事务将决定其他事务是否失败，以及清理其他事务的那些锁。

当客户端在执行两阶段提交的commit阶段crash时，事务会留下一个提交点commit point(至少已经写入一条write记录)，但可能会留下一些lock未被处理掉

如果priarmy lock 已被write所替代：意味着该事务已被提交，事务需要roll forword，也就是对所有涉及到的、未完成提交的数据，用write记录替代标准的锁standed lock。
如果primary lock存在：事务将roll back(因为总是最先提交primary，所以primary未被提交时，可以安全地执行回滚)

这些都是基于bigtable的事务中的。

清理操作在primary锁上是同步的，所以清理alive客户端持有的锁是安全的；然而回滚会强迫事务取消，这会严重影响性能。所以，一个事务将不会清理一个锁除非它猜测这个锁属于一个僵死的worker。
Percolator使用简单的机制来确定另一个事务的活跃度。运行中的worker会写一个token到Chubby锁服务来指示他们属于本系统，token会被其他worker视为一个代表活跃度的信号（退出时token会被自动删除）。有些worker是活跃的，但不在运行中，为了处理这种情况，我们附加的写入一个wall time到锁中；一个锁的wall time如果太老，即使token有效也会被清理。有些操作运行很长时间才会提交，针对这种情况，在整个提交过程中worker会周期的更新wall time。

通知

用户对感兴趣的列编写观察者function注册到percolator，当列发生改变时，percolator通知percolator的worker运行用户function。

通知与写操作不是原子的
通知类似于数据库中的触发器，然而不同的是，通知在其他事务(worker)中执行，所以写操作与观察者执行不是原子的，且观察者的执行会有时效性问题。

这和传统关系型数据库的ACID的C有一定的差别，我猜: 所以这里不叫trigger而是通知的原因

通知与观察者无限循环
编写观察者时，用户要自己考虑通知与观察者进入无限循环的情况(通知->观察者->通知->观察者.....)。

通知的"丢失"
一个列的多次更改只会触发一次通知，所以通知和操作系统的中断一样会存在“丢失”的问题。

实现通知机制

为了实现通知机制，Percolator需要高效找到被观察的脏cell。
Percolator在Bigtable维护一个“notify”列(notify列为一个独立的Bigtable locality group)，表示此cell是否为脏。当事务修改被观察的cell时，则设置cell的notify。worker对notify列执行一个分布式扫描来找到脏cell。找到notify则触发观察者并且等到观察者事务提交成功后，会删除对应的notify cell。

tidb的实现

percolator定义了5个列：data, write, lock, ack, notify。
tidb定义了其中3个：data, write, lock。所以tidb没有实现notify功能(tidb不需要增量处理能力)。

tidb定义了3个rocksdb的column family：

CF_DEFAULT：对应percolator的data列
CF_LOCK：对应percolator的lock列
CF_WRITE：对应percolator的write列

CF_DEFAULT

(key, start_ts) ==> value

CF_LOCK

key ==> lock_info

同一时刻一个key最多只有一个锁，所以，tidb的锁没有start_ts。

CF_WRITE

(key, commit_ts) ==> write_info

tidb

1 client 向 tidb 发起开启事务 begin
2 tidb 向 pd 获取 tso 作为当前事务的 start_ts
3 client 向 tidb 执行以下请求：
- 读操作，从 tikv 读取版本 start_ts 对应具体数据.
- 写操作，写入 memory 中。
4 client 向 tidb 发起 commit 提交事务请求
5 tidb 开始两阶段提交。
6 tidb 按照 region 对需要写的数据进行分组。
7 tidb 开始 prewrite 操作：向所有涉及改动的 region 并发执行 prewrite 请求。若其中某个prewrite 失败，根据错误类型决定处理方式：
- KeyIsLock：尝试 Resolve Lock 后，若成功，则重试当前 region 的 prewrite[步骤7]。否则，重新获取 tso 作为 start_ts 启动 2pc 提交（步骤5）。
- WriteConfict 有其它事务在写当前 key, 重新获取 tso 作为 start_ts 启动 2pc 提交（步骤5）。
- 其它错误，向 client 返回失败。
8 commit : tidb 向 pd 获取 tso 作为当前事务的 commit_ts。
9 tidb 开始 commit:tidb 向 primary 所在 region 发起 commit。若 commit primary 失败，则先执行 rollback keys,然后根据错误判断是否重试:
- LockNotExist 重新获取 tso 作为 start_ts 启动 2pc 提交（步骤5）。
- 其它错误，向 client 返回失败。
10 tidb 向 tikv 异步并发向剩余 region 发起 commit。
11 tidb 向 client 返回事务提交成功信息。

所有涉及重新获取 tso 重启事务的两阶段提交的地方，会先检查当前事务是否可以满足重试条件：只有单条语句组成的事务才可以重新获取tso作为start_ts。

tikv

Prewrite

伪代码

-> 代表rpc调用, 例如tidb->tikv.Prewrite tidb调用tikv的Prewrite接口
. 代表进程内调用, 例如memory.Put往内存模型写数据

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
}

Commit

伪代码

// 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
}

Rollback

当事务在两阶段提交过程中失败时， tidb 会向当前事务涉及到的所有 tikv 发起回滚操作。

伪代码

// 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
}

Resolve Lock

若客户端在Commit一个事务时，出现了异常，Prepare 时产生的锁会被留下。为避免将新事务hang住，Percolator必须清理这些锁。

Percolator用lazy方式处理这些锁：当事务A在执行时，发现事务B造成的锁冲突，事务A将决定事务B是否失败，以及清理事务B的那些锁。

tidb 在执行 prewrite, get 过程中，若遇到锁，在锁超时的情况下，会向 tikv 发起清锁操作。

// 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
}

Get

// 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)
    }
}

GC

TiDB 的事务的实现采用了MVCC机制，当新写入的数据覆盖旧的数据时，旧的数据不会被替换掉，而是与新写入的数据同时保留，并以时间戳来区分版本。
GC 的任务便是清理不再需要的旧数据。

一个 TiDB 集群中会有一个 TiDB 实例被选举为 GC leader，GC 的运行由 GC leader 来控制。

GC 会被定期触发。每次 GC 时，首先，TiDB 会计算一个称为 safe point 的时间戳，接下来 TiDB 会在保证 safe point 之后的快照全部拥有正确数据的前提下，删除更早的过期数据。

每一轮 GC 分为以下三个步骤：

Resolve Locks：该阶段会对所有 Region 扫描 safe point 之前的锁，并清理这些锁
Delete Ranges：该阶段快速地删除由于 DROP TABLE/DROP INDEX 等操作产生的整区间的废弃数据
Do GC：该阶段每个 TiKV 节点将会各自扫描该节点上的数据，并对每一个 key 删除其不再需要的旧版本

// 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];
    }
}

优化

Parallel Prewrite

tikv分批的并发进行prewrite，不会像percolator要先prewrite primary，再去prewrite secondary。
如果事务冲突，导致rollback，在tikv的rollback实现中，其会留下rollback记录，这样就会导致事务的prewrite失败，而不会产生副作用。

Short Value

对于percolator，先读取column:write 列，提取到key的start_ts，再去column:data列读取key数据本身。
这样会造成两次读取，tidb的优化是，如果数据本身很小，那么就直接存储在colulmn:write中，只需读取一次即可。

Point Read Without Timestamp

为了减少一次RPC调用和减轻TSO压力，对于单点读，并不需要获取timestamp。

因为单点读不存在跨行一致性问题(读取多行数据时，必须是同一个版本的数据)，所以直接可以读取最新的数据即可。

Calculated Commit Timestamp

如果不通过TSO获取commit_ts，则会减少一次RPC交互从而降低事务的时延。

然而，为了实现SI的RR特性(repeatable read)，所以commit_ts需要确保其他事务多次读取的值是一样的。那么commit_ts就和其他事务的读取有相关性。

下面公式可以计算出一个commit_ts

max{start_ts, max_read_ts_of_written_keys} < commit_ts <= now

由于不可能记录每个key的最大的读取时间，但是可以记录每个region的最大读取时间，所以公式转换为：

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

region_x_max_read_ts : 事务涉及到的key的region。

Single Region 1PC

对于事务只涉及到一个Region，那么其实是没有必要走2PC流程的。直接提交事务即可。

percolator的理解与开源实现分析