本篇文章选自数据库内核杂谈系列文章。

欢迎阅读新一期的数据库内核杂谈。首先和大家道个歉，拖更太久了。下半年工作太忙了，一直没能空出时间来更新。终于到了感恩节假期，赶紧来填一下上一期留下的坑：假设给一个单机的数据库系统实现，在这个基础上如何把它扩建成分布式数据库系统。

先来说说这个坑的趣事，当时在原公司参与系统设计面试，但不幸和同事撞题了。。。这个题是我在推开面试会议室门的时候临时想的(被自己机智到了！)记得，当时是这么对面试者说的，“你好，今天我们要做的设计题是，假定现在给你一个单体的数据库系统实现，比如开源的 Postgres 或者 MySql 的实现，把它拓展成一个分布式数据库。要求是，一，分布式设置对客户端屏蔽，即，对客户端来说，还是连接到了一个逻辑的数据库实体；二，实现基本的 SQL 实现。”。后来我仔细想想，觉得这个设计题还是蛮不错的。足够开放，可以横向聊全局，也可以纵向聊细节。看到这里，大家可以暂停一下，思考一下，你会怎么做？

全局分配

题目的第一个要求是，分布式设置对客户端屏蔽。也就是说，客户端在连接到数据库之后，并不知道这是一个分布式数据库。不难想到，无论这个分布式数据库设置多少台机器，应该只有一台机器负责和客户端沟通，来屏蔽分布式的实现。 沿着这个思路想下去，这台机器接受客户端的指令，然后协调统筹其他机器完成操作。自然而然地就引出了一个 leader 节点带领着一群 worker 节点协同工作的全局架构(布局如下图所示)。这个设置中，由 leader 节点全权负责和客户端沟通：建立连接，增删改查；只需要暴露 leader 节点的地址和端口，客户端完全不需要知道有 worker 节点存在。如此，满足了要求一。另一个好处就是方便协调管理。Leader 节点作为整个数据库的单一大脑，负责协调所有工作，worker 节点只要听指挥工作即可。缺点也明显，就是 leader 节点是数据库的单点故障瓶颈(这里就暂不扩展如何解决这个瓶颈了)。  具体 leader 这个大脑负责哪些工作，后面的章节会做讨论。

由于互联网行业的兴起，数据量成级数增长。现有的数据库需要不断升级来支持海量数据。但纵向扩展(vertical scaling)是有上限的，并且性价比越来越低。分布式数据库解决的一大痛点就是，如何可持续地支持数据增长。分布式数据库采用的是横向扩展(horizontal scaling)，通过增加新的 worker 节点，把工作分摊开来。从性价比和可持续发展的角度来看，都比纵向扩展更好一些。但是天下没有免费的午餐，分布式系统对于软件设计复杂度也大大提高，介绍分布式系统复杂性的文章太多了，这里就不扩展了。

存储

聊完了全局节点分配，来聊一下存储。分布式数据库解决的第一个问题就是，如何存储单机数据库无法存放下的海量数据。既然有了多个 worker 节点，自然想到，要把数据分发存储到各个 worker 节点上。这是一个典型的分布式存储问题，如何 shard/distribute(分发)数据。最容易想到的就是 random shard(随机分发)。对于每一行(row)数据，随机分配到一个 worker 节点上存储。随机分发的好处也显而易见。一，实现简单，每次分发相当于只需要"掷个骰子"。二，分布均匀，随机分发可以均匀地把数据分发到每个节点上，并且，保证不会出现热点数据。

提到了随机分发，大家肯定会想到 hash 分发:对于每行数据，计算一个 hash 值，然后根据 hash 值决定数据去到哪一个节点。可以选择 mod 哈希值这种简单操作，也可以选择 consistent hashing 这类方便未来做扩展的。具体如何计算哈希值，一是选择 hash 函数(哈希函数的选择就不在这延伸了，主要是要做到性能和均衡)。二就是，决定哈希函数的输入。常见的方法就是，选择一个或多个 column 作为输入，比如用主键。追求的依然是把数据均匀的分布到各个节点上。主键由于通常都是唯一的，更容易做到 hash 后均匀分布。因此，尽量不要用 cardinality 比较小的 column。除了追求数据均匀分布，还可以根据业务需求，选择其他的 column。为什么这么说呢，先在这卖个关子，等到讲处理数据的时候再说。

小结一下，分布式通过数据分发策略，将数据分摊到各个 worker 节点中存储，突破了单体数据库的存储瓶颈。常见的分发策略有随机分发和 hash 分发。

元数据

聊完了数据存储，简单聊一下元数据的存储。首先，分布式数据库的元数据，相对于单体数据库，多了刚才用到的分布式信息，这层信息必须存储在 leader 节点上。对于普通的元数据，比如表的信息，列的信息，索引等等，每个节点需要存储这些元数据，在执行 sql 语句的时候会用到。此外，通常 leader 节点是整个数据库的大脑，需要对 sql 语句进行编译，因此，这些元数据也应该保存一份在 leader 节点上。总结而言，leader 节点上会存储全部的元数据，分布式元数据和基础元数据，用于数据分发、sql 语句的编译及优化。每个 worker 节点需要保留本地数据的元数据，用于 sql 语句的执行。

数据处理：Sql 语句执行

介绍完了数据和元数据存储，可以来讲如何执行客户端发来的 SQL 语句啦。在第四章-执行模式，一起学习过执行 SQL 语句的四个步骤：编译(parsing)，绑定(binding)，优化(optimizing)，执行(executing)。在分布式环境下，依然是这些步骤。具体哪些节点应该执行哪些操作呢？因为每个 worker 节点都存储了数据，这些节点至少需要执行物理计划来获取数据。另外，在上文提到过，leader 节点存储了所有的元数据，通常作为数据库的大脑，可以对 sql 语句进行编译，因此，leader 节点负责对 SQL 语句的编译，绑定和优化。leader 节点需要做执行步骤吗？答案是需要的。因为最终结果集(resultset)是通过 leader 节点传输给客户端，因此 leader 节点至少需要负责收集(merge)所有 worker 节点的结果。当然，在优化生成物理执行计划的时候，应该尽量减少 leader 节点的执行算子复杂度，因为 leader 节点需要处理所有的客户端请求，并且对这些语句进行，编译，绑定和优化。

编译和绑定，相对于单体数据库而言，分布式数据库的处理没有任何区别。重点在于优化和执行。优化的步骤是优化器在众多候选计划中选出最优的物理算子执行计划，然后交给执行器执行。在分布式环境下，仅 leader 节点需要有优化器的实例，所有的节点都会有执行器的实例。

在第五章和第六章里，一期学习过物理算子的实现：一个算子，先从下层算子中读取数据，做相应处理，然后输出到上层算子。要让物理执行计划在分布式环境下执行，需要引入下面这一类分布式算子，这些算子实现了节点间的数据传输的不同需求。和普通算子相比，分布式算子是节点间相互通讯，需要 RPC 调用，调用成本是相对更高。

Merge 算子：worker 节点向 leader 节点输送数据。对于 worker 节点来说，merge 算子将下层算子中读取的数据不断发送到 leader 节点。对于 leader 节点来说，merge 算子替代了单体数据库中的 scan 算子用来从 worker 节点获取数据。

Redistribute 算子：worker 节点之间传输数据的算子之一，由 worker 节点将数据发送到另一个 worker 节点。特别强调“一个”，是要和下面介绍的 Broadcast 算子区分开。具体发送到哪个 worker 节点，和上文说过的存储分发是异曲同工的，可以选择随机分发，也可以选择 hash 分发(主要是 hash 分发)。

Broadcast 算子：worker 节点之间传输数据的算子之二，由 worker 节点将数据广播到所有 worker 节点(包含自己)。读者可能有疑问，有了 redistribute 算子，为什么还需要 Broadcast 算子来广播数据，而且从性能开销角度看，Broadcast 也比 redistribute 更高。在下面介绍常见 SQL 语句的执行计划时，会聊到 broadcast 算子的作用。

这三个分布式算子和普通算子共同构建了分布式物理执行计划。万事俱备，来看如何用这些算子实现基本的 SQL 语句。

Select (select * from table)

还是先从简单的 Select 全表开始，由于表数据分摊到了所有的 worker 节点上，因此需要每个 worker 节点在本地扫描全部数据，然后通过 merge 算子发送给 leader 节点，leader 节点在接受到数据后即可返回给客户端。下图给出了物理执行计划示例。

Filter(select * from table where CONDITION)

相比于全表扫描，加入了 where condition，因此需要引入 filter 算子。很容易想到把 filter 算子放在 leader 节点，用来过滤掉不满足 condition 的数据(执行计划如下图所示)。

但是有没有更优化的执行计划呢？是有的。在优化器那一章讲过算子下推(predicate push down)。显然，优化器可以选择把 filter 算子下推到 worker 节点(执行计划如下图所示)。Worker 节点在扫描数据的同时可以直接过滤不满足 condition 的数据。相比于上面的执行计划，新计划在 merge 过程中，传输更少的数据(只传输满足 condition 的数据)。举个极端的例子:select * from super_large_table where false，后者几乎不需要传输任何数据。具体选择哪个执行计划，就是优化器的工作啦。

Sort(select * from table order by COLUMN)

再来看一下如何支持排序操作。一个无脑的操作就是 worker 节点将所有的数据发送给 leader 节点，然后由 leader 节点进行排序再返回结果集。但这个执行计划显然是低效的，不仅需要将所有数据都传输给 leader 节点，排序工作也一并交给 leader，这样和单体数据库执行没有任何区别。如何利用 worker 节点来协助排序呢？这里可以借鉴 mergeKSortedList，分而治之。每个 worker 节点先对本地数据进行排序；排序完以后用 merge 算子发送给 leader 节点，leader 节点通过维护 一个长度等于 worker 节点数量的最小化堆对数据进行堆排序。这样不仅利用了所有 worker 节点来帮助排序，同时 leader 节点的堆排序完全可以放在内存中进行，避免了外排序带来的 IO 开销。执行计划见下图。

Group by(select GroupByCol1, count(1) from table group by 1)

第五章排序和聚合(link)提过排序和聚合有很多共通之处。对于分布式的实现来说，亦是如此。因此，决定把聚合的实现留作思考题给大家，我会后续在评论区贴出答案。

Join (select * from table1, table2 where table1.col1 = table2.col2)

终于到今天最后介绍，也是最复杂的 join 算子的实现了。为什么 join 最复杂，因为它和单体数据库的执行计划最不同。两个，或者多个表的 join 牵涉到所有数据的两两比较，但表的数据本身又分布在所有 worker 节点上。因此，每个 worker 节点单独做 join 从语义上讲是错误的，或者是不完整的。怎么才能解决这个问题呢？这就要用到上文提到的 Redistribute 和 Broadcast 算子了。用上述的 join condition 做示例，要找到 table1.col1 = table2.col2 的数据。可以对每个 worker 节点 table1 数据，根据 col1 的 hash 值做 redistribute，同时对 table2 数据 col2 的 hash 值做 redistribute。如此这般，两个表中相同 col1 及 col2 的数据都会汇聚到同一个 worker 节点上，这时，只需要 local join 操作就可以得到正确的结果。最后再把结果汇聚给 leader 节点即可。具体的执行计划示例如下。

有没有优化的余地呢？是有的。讲存储的时候，对于选择哪个 column 作为 hash 分发的 key 时，卖了个关子。现在迷底揭晓，假设，table1 的数据本身就以 col1 作为 hash 分发的 key，那就可以省略 redistribute table1 的数据了。如果 table2 是以 col2 为 key 的话，那两边的 redistribute 就都可以省去了。因此，在选择表的 hash 分发 key 时，应该根据业务需求来决定。比如，对于非常大的数据表(俗称 fact table)，可以根据经常需要被 join 的 column 来做 hash 分发的 key，这可以避免大表在执行 join 的过程中占用大量的数据传输(redistribute)。

那如果某些 join 涉及到大表，但是并没有相对应的 hash key 存储支持，还有其他办法可以优化吗？答案是有的。轮到 Broadcast 算子出场了。当 join 涉及一个大表(例如 billion 级别数据)和一个相对较小的表(例如 10K 级别)，优化器可以选择将小表的数据 broadcast 到每个 worker 节点(相当于每个 worker 节点保留了一份全部的小表数据)。这样，worker 节点就可以 local join 本节点的大表数据和所有小表数据得到 join 结果并返回给 leader 节点。并且，由于小表数据量小，可以存放在内存中做 hash join，相当于对本地的大表数据只需要一次 IO 读取即可。最重要的是，完全避免了重新 redistribute 大表的数据。执行计划如下图所示。

至此，一起学习了如何构造分布式执行计划来实现一些常见的 sql 语句。这些语句仅仅是最基本的 sql 语法，常见的业务语句会需要生成更加复杂的执行计划(但是，万变不离其宗，再复杂的语句也可以由这些基本算子构建而成)。分布式设置，无疑给优化器增加了一个量级的复杂度来计算最优计划。优化器不仅要确保语义的正确性，普通算子的搭配，还要考虑如何分配分布式算子来减少开销。

总结

总结一下，这一篇讨论了如何把一个单机数据库改造成分布式数据库。首先，介绍了 leader-worker 模式的全局分配。然后，讨论了数据和元数据的存储。最后，介绍了分布式算子以及它们如何和普通算子共同构建分布式执行计划来运行 SQL 语句。当然，分布式数据库存在多种架构，比如 PingCAP 的 TiDB 这类 NewSQL 系统，是基于底层分布式 key-value 存储的。以后有机会再给大家介绍。感谢阅读。