TiDB: A Raft-based HTAP Database

发表于

TIDB

ABSTRACT

Hybrid Transactional and Analytical Processing (HTAP，混合事务和分析处理)数据库要求独立地处理事务和分析查询，避免相互干扰。为了实现这一点，需要为两类查询维护不同的数据副本。然而，为存储系统中的分布式副本提供一致性视图并不容易。在该系统中，分析请求可以大规模地、高可靠性地从事务工作负载中读取一致的实时数据。

INTRODUCTION

关系型数据库（RDBMS）一直因其关系模型、强力的事务保证和SQL接口而广受好评，但它不具备高可扩展性和高可用性。因此NoSQL就应运而生，像Google bigtable和DynamoDB之类的放宽了一致性要求，提供了更高的可扩展性。但由于业务又需要事务处理能力、数据一致性和SQL接口等，就慢慢出现了像CockroachDB和Spanner之类的NewSQL。此外，像许多架构在Hadoop之上的SQL系统一样，在线分析处理系统（OLAP）也在迅速发展。

以前一般认为针对OLAP和OLTP应该采用的不同的数据模型和技术方案，但维护多个系统的成本很高。业界开始探索OLTP和OLAP的混合系统HTAP。HTAP系统需要满足几个关键点：一是数据新鲜度，即OLAP需要拿到最新的数据；二是隔离，即为单独的OLTP或者OLAP查询提供不同的硬件资源处理，避免性能相互影响。

本文就是基于上面的考虑，提出了一个以Raft为共识算法的HTAP数据库——TiDB，在Raft中引入了一个专门的节点Learner，Learner会异步地复制Leader节点的事务日志，并为OLAP查询构造新副本，并将日志中的行格式元组转换为列格式，便于查询。

RAFT-BASED HTAP

使用共识算法如Raft、Paxos等可以基于复制状态机在服务器之间实时可靠地复制数据，通过调整，该论文的做法可以针对不同的 HTAP 工作负载将数据复制到不同的服务器，并且保持资源隔离和数据新鲜度。

如上图所示，TiDB将数据按行格式存储在多个Raft Group里，每个Raft Group由一个Leader和多个Follower组成，另外为每个组增加一个Learner角色，可以异步复制Leader的数据，并将行格式转换为列格式。另外，扩展的查询优化器用来构建访问列存和行存的物理计划。

在该实现中，Leader不参与Leader选举，也不计入日志复制的个数，不参与Quorum。读数据时，需要保证Leader和Learner之间的强一致性和日志复制的低延迟。行列格式的转换也有优点，行格式可以利用索引来高效进行事务查询，列格式可以有效利用数据压缩和矢量化处理。由于Learner部署在单独的物理资源中，OLAP和OLTP可以在独立的资源中得到处理。

总的来说，这个设计克服了几个困难：一是基于Raft的系统如何应对高并发读写；二是资源独立如何保证数据新鲜度；另外就是如何构建查询计划在行式存储和列式存储中选择最优的。

ARCHITECTURE

如下图所示，这是TIDB的架构，兼容MySQL协议，由三个核心组件组成：分布式存储层、Placement Driver布局驱动器和计算引擎层。

  • 分布式存储层

由行式存储（TiKV）和列式存储（TiFlash）组成，存储在TiKV的数据是有序的key-value对，key由两个整数table id和row id组成，其中row id就是主键列，value就是真实的行数据。

1
2
Key: {table{tableID} record{rowID}}
Value: {col0, col1, col2, col3}

为了保证可扩展性，TiDB采用了range partition的策略，将kv对映射分割成若干个连续的范围，每个范围称为一个region，每个region都有多个副本来保证可用性，Raft就是用于维护每个region中若干个副本的一致性的。

PD负责管理region，包括提供每个key对应的region和其物理位置，以及自动转移region以平衡负载。同时PD也是时间戳分配器，提供了严格递增全剧唯一的时间戳。PD不具备持久状态。

计算引擎层是无状态、可扩展的，其SQL引擎由cost-based query optimizer和distributed query executor组成。另外TiDB基于Percolator实现了2PC协议。

除此之外，TiDB还与Spark集成，方便集成TiDB和HDFS中存储的数据。

MULTI-RAFT STORAGE

下图展示了分布式存储层的架构，其由TiKV和TiFlash组成。每个Region的副本之间都使用Raft来维护数据一致性。当数据复制到TiFlash的时候，为了方便扫描，多个Regions会被合并成一个Partition。

Row-based Storage

TiKV是由多个TiKV服务器组成的，每个TiKV服务器都可以是不同Region的Raft Leader或者Follower，另外在TiKV服务器上，数据和原数据都会保存在RocksDB上。

基于Raft算法响应读写请求的过程如下：

  1. Region的Leader从SQL引擎层接受请求；
  2. Leader将请求Append到日志中；
  3. Leader向Follower发送新的日志条目，Follower会将接收到的日志Append到自己的日志中；
  4. Leader等待Follower回应，满足指定数量的节点响应成功后，Leader会在本地commit并Apply；
  5. Leader将结果发送给客户端；

Optimization between Leaders and Followers

为了提高吞吐，可以在Leaders和Followers之间的操作做一些优化。首先是，上述过程的第二步和第三步可以并行进行，即便第二步失败了，只要第三步成功了仍然可以提交日志。另外就是，第三步中Leader可以缓冲这些日志条目，并批量发送。并且发送日志后也不需要等待Follower响应，可以假设发送成功，并利用预测的日志索引发送后来的日志。即便出现错误，Leader可以调整日志索引进行重发。还有就是，第四步中，Leader应用日志可以交给其他线程去做。整体流程就变成：

  1. Region的Leader从SQL引擎层接受请求；
  2. Leader并行地向Follower发送日志，并同时Append本地日志；
  3. Leader继续接收请求，重复2；
  4. Leader commit日志，并将应用逻辑交给另外的线程去做；
  5. 应用日志后，Leader返回结果；

Accelerating Read Requests from Clients

从Leader读取数据具有线性化的语义，但通过常规的Raft流程来保证会导致很高的网络IO开销。为了提高性能，可以避免日志同步阶段。

Raft保证：一旦Leader成功写入数据，就可以响应任何读取请求，而无需同步日志。但Leader是可能改变的。为了实现从Leader读取，TiKV做了以下优化：

  • read index：Leader响应请求时，会将当前的提交索引记录为本地read index，然后向Follower发送heartbeat确认Leader地位。如果确实是Leader，并且应用的索引大于或等于read index，就可以返回值。
  • lease read：为了减少heartbeat开销，Leader和Follower之间维护一个Lease期限，Follower在这期间不发出选举请求，因此Leader在此期间也无需与Follower进行heartbeat交流。

Follower也可以响应client的读请求，但需要向Leader询问read index，如果本地应用索引等于或大于read index，则Follower可以将值返回给客户端。

Managing Massive Regions

为了实现跨机器平衡Region，Plancement Driver会对Region副本数量和位置施予限制。一个就是必须要在不同的TiKV实例上放置至少三个Region副本。PD通过Heartbeat从服务器收集信息、监控服务器负载，并将热Region进行转移。

另一方面，维护大量Region涉及Heartbeat信息和元数据管理导致的大量的网络和存储开销，会被PD根据负载情况调整心跳频率。

Dynamic Region Split and Merge

这里主要设计Region的拆分和合并。热点Region或者大型Region会被拆成小Region，小或者访问少的Region，会被合并成大Region，以减少由于维护小Region心跳和元数据带来的网络和CPU开销。

PD操作Region，是通过向TiKV发送拆分和合并指令，然后以Raft流程来完成更新请求。Region拆分比较简单，只需要更改元数据。合并的话，PD会移动两个Region的副本，放到单独的服务器上。然后通过两阶段操作，在每台服务器上本地合并两个Region的并置副本：即停止其一Region的服务并将其与另一Region合并。

Column-based Storage (TiFlash)

考虑到TiKV中的行存数据并不适合OLAP，因此将列存储TiFlash合并到TiDB中。TiFlash由Learner节点组成，仅从Raft组接收Raft日志，并将行格式的元祖转换为列存数据。

用户可以通过SQL语句为表设置列格式副本，ALTER TABLE x SET TiFLASH REPLICA n;其中x是表名，n是副本数量。在TiFlash中，每个表会按partitions进行划分，每个partitions包括几个连续Regions，更大的Regions便于范围扫描。

当初始化一个TiFlash实例时，相关Region的Leader开始讲数据复制到新的Learner。一旦初始化完成后，TiFlash开始监听Raft组的更新。Learner收到日志后，会将日志应用到本子状态机，包括日志重放、转换数据格式和更新本地存储中的引用值。

Log Replayer

由于在Raft中，Learner接收到的日志时可线性化的，因此重放日志也会按照FIFO的策略重放日志，具体步骤包括：

  • 压缩日志：事务日志分为三种状态：预写、提交或回滚。回滚的日志不需要写盘；
  • 解码元组：缓冲区中的日志被解码为行格式的元组，去除关于事务的冗余信息；
  • 转换数据格式：行元组会被转换为列数据；

更具体的步骤可以参考上图。

Schema Synchronization

为了实时将元组转换为列格式，Learner需要知道最新的schema，因此TiFlash会通过Schema syncer与TiKV中最新的Schema同步。同时为了减少TiFlash向TiKV的请求次数，每个Learner都会维护一个schema cache。这里采取两阶段策略：

  • Regular synchronization：定期同步；
  • Compulsive synchronization：synced检测到不匹配的schema，就会主动从TiKV拉去最新的Schema；

Columnar Delta Tree

TiFlash设计了一个新的列存储引擎——Delta Tree，该引擎支持快速追加增量更新，然后将其与每个Partitions的稳定版本合并。如下图所示，在Stable space中，Partitions以chunks的形式存储，每个chunk都覆盖了一个较小的元组范围。TiFlash将元组的列数据及其元数据存储到不同的文件中，以同时更新文件。

新进来的增量更新时插入或者删除数据的原子批处理，这些增量会缓存到内存中，并持久化道磁盘。另外TiFlash会定期将小增量压缩成大增量，并持久化。传入增量的内存副本有助于读取最新数据。

另外，读取的时候由于需要合并增量文件和稳定空间中的数据，而且增量文件本身也可能存在空间放大的问题。TiFlash需要定期将增量合并到稳定空间中，每个增量文件及其相关chunks被读入内存进行合并，合并后的chunks自动替换磁盘中的原始chunks。

由于相关的键在增量空间中是无序的，合并成本较高，并且也会影响合并读的速度，因此会在增量空间构建B+ Tree索引，每个增量更新项都被插入到 B+ Tree 中，并按其关键字和时间戳进行排序。便于快速响应读请求，和有序数据也更易与稳定块合并。

Read Process

与Follower read类似，Learner提供snapshot isolation，在接收到读取请求后，Learner向其Leader发送read index请求，以获取涵盖所请求时间戳的最新数据。Learner拿到日志后，回放日志，写入Delta Tree，就可以读到特定数据响应请求了。

HTAP ENGINES

为了提供OLAP和OLTP查询，TiDB引入了SQL引擎来为了查询计划做决策。SQL引擎使用Percolator模型来实现分布式集群的乐观和悲观锁，并基于优化器、索引和下推算子来支持OLAP查询。

Transactional Processing

TiDB为ACID事务提供了snapshot isolation语义或者repeatable read语义，前者允许每个请求读取版本一致的数据，后者则是事务中的不同语句可能为同一个key读取不同的值，但重复读取将会读取到相同的值。这是基于MVCC实现。

如下图，TiDB中的事务由SQL引擎、TiKV和PD三个组件共同完成：

TiDB对于悲观锁和乐观锁的实现启发自Percolator模型。

  1. client接收到Begin命令后，SQL引擎向PD请求一个start_ts时间戳；
  2. SQL引擎从TiKV读取数据并在本地内存中执行SQL DMLs。TiKV提供在start_ts之前最近的commit_ts数据；
  3. SQL引擎收到client的commit命令后，启动2PC协议。随机选择一个主键，并行锁定所有的key，并将prewrite发送TiKV节点；
  4. 如果所有的prewrite都成功了，SQL引擎会向PD请求一个commit_ts，并向TiKV发送commit命令。TiKV commit主键，并响应成功回到SQL引擎；
  5. SQL引擎向Client响应成功；
  6. SQL引擎向TiKV发送commit命令，异步并行地提交从键和清除锁；

悲观事务和乐观事务的区别在于获取锁的实际，前者是在第二步执行DMLs的时候获取，后者则是第三步prewrite的时候。

Analytical Processing

TiDB通过两阶段查询优化来实现优化器：rule-based optimization生成逻辑计划， cost-based optimization将逻辑计划转换为物理计划。由于TiDB有两类存储、因此扫描表往往有三种选项：扫描TiKV的行格式表、扫描TiKV中有索引的表和骚婊TiFlash的列。

索引可以提高点查询和范围查询的性能，TiDB实现了可扩展性的索引，由于维护索引会消耗大量资源，因此会在后台以非同步的形式构建或删除索引。索引是以与数据相同的方式按Regions分割，并作为键值存储在TiKV 中。唯一键索引上的索引项编码为：

1
2
Key: {table{tableID} index{indexID} indexedColValue}
Value: {rowID}

非唯一索引上的索引项被解码为

1
2
Key: {table{tableID} index{indexID} indexedColValue rowID}
Value: {null}

物理计划的执行是由SQL引擎层使用pulling iterator model进行的，通过将部分计算下推到存储层，可以进一步优化执行。在存储层，执行计算的组件称为coprocessor，coprocessor在不同的服务器上并行执行substrees of an execution 破烂，这减少了必须从存储层发送到引擎层的元组数量。

总结

本文主要是介绍了一个投入生产环境的HTAP数据库——TiDB。TiDB建立在TiKV上，这是一个基于Raft的分布式行式存储，并引入一个TiFlash组件用于实时分析，作为Raft的learner从TiKV异步复制日志，并将行格式的数据转换为列格式。