关于数据库:OceanBase-40-解读分布式查询性能提升我们是如何思考的

对于作者

王国平

OceanBase 高级技术专家

目前 OceanBase SQL 引擎的技术负责人。2016 年退出 OceanBase，负责 SQL 引擎的研发工作。2008 年毕业于哈尔滨工业大学，2014 年在新加坡国立大学取得博士学位，博士期间次要钻研方向是数据库畛域的 (多) 查问优化和解决。在退出 OceanBase 之前，已经在华为从事数据库的研发工作。

性能是掂量数据库系统的重要指标之一，也是数据库系统畛域始终备受关注的话题。在 OceanBase 3.x 版本中，OceanBase 曾经实现了绝对欠缺的优化器引擎、单机执行引擎、并行执行引擎和向量化执行引擎。在 2021 年 5 月份，OceanBase 用这个版本打榜了 TPC-H，在数据分析型基准测试榜单的 30000GB 后果一栏，OceanBase 占据性能排行首位，其中代表着数据库外围性能的每小时执行申请数综合指标达到了 1526 万 QphH@30,000GB。这次打榜充分证明了 OceanBase 的分布式查问能力性能，而且具备线性可扩大。

尽管如此，在整个 3.x 版本的大规模利用中，咱们在局部业务场景中还是遭逢到了一些性能问题，比方在特定的分布式场景中生成了不优的执行打算、执行引擎对于不优的执行打算的容错能力、特定场景下没法充分利用所有的并行度来放慢查问的执行等。为了解决这些问题，在 OceanBase 4.0 设计之初，咱们就始终在思考，OceanBase 应该如何改良 SQL 引擎来晋升分布式查问性能。分布式查问优化和分布式执行引擎从根本上决定了 SQL 引擎的分布式查问性能，上面咱们从这两个方面来聊一聊咱们的思考。

OceanBase 4.0 如何做分布式查问优化？

家喻户晓，查问优化是数据库内核开发的重点和难点，也是数据库查问性能的关键点。查问优化的作用是给帮忙用户写的每一条 SQL，抉择一个最优的执行打算。通常来说，一条 SQL 会有很多等价的执行打算，不同执行打算的性能可能会有数量级别的差别，所以查问优化很多时候从根本上就决定了查问的性能。OceanBase 是一个分布式关系数据库系统，这就意味着 OceanBase 天生就须要解决分布式查问优化的问题。在整个关系数据库系统中，查问优化始终是开发的难点，而分布式的查问优化就更加加剧了优化的难度。接下来咱们来聊聊相比于单机查问优化，分布式查问优化的挑战在哪里。

▋ 分布式查问优化的挑战

分布式查问优化大大晋升了打算枚举空间

在查问优化中，优化器的其中一个指标是须要给执行打算中的每个算子抉择一种具体的实现办法。在单机的场景下，算子的实现办法只须要思考单机的实现，然而在分布式的场景中，算子的实现办法除了要思考单机实现之外，还须要思考其分布式的实现。就拿数据库中的连贯算子而言，在单机的场景中，通常的实现办法有 hash join、merge join 和 nested loop join。在分布式的场景中，通常的实现办法有 partition wise join、partitial partition wise join、hash-hash distribution join 和 broadcast distribution join。这些分布式的实现办法正交上单机的实现办法就会大大增加分布式查问优化的打算枚举空间，会让整个分布式查问优化变得更加有挑战。

分布式查问优化须要保护更多的物理属性

在单机的查问优化中，算子序是一个十分重要的物理属性，因为算子的序可能会用来减速后续的一些算子的执行。算子序实质上就是运行完这个算子之后，数据库中的元组是不是依照特定的序输入的。举个简略的例子，对于索引 (a,b,c) 的扫描，因为在 OceanBase 中索引扫描是保序扫描，所以这个索引扫描之后的序就是(a,b,c)。算子序跟特定的算子实现有关系，而且它可能会影响后续算子的代价，所以在每个算子执行之后，查问优化都会保护序这个物理属性，并且在做打算裁剪的时候会保留有用序的执行打算。

在分布式查问优化中，除了序这个物理属性之外，另外一个物理属性就是分区信息。分区信息次要包含数据的分区形式以及每个分区的物理地位信息。分区信息从根本上决定了一个算子的分布式算法的抉择，比方一个连贯能不能做 partition wise join 是取决于连贯键和表的分区信息的，所以分区信息同样可能也会影响后续算子的代价，所以在分布式查问优化中，除了保护序这个物理属性之外，咱们还须要保护分区信息这个物理属性。分区信息的保护最终会影响打算裁剪和打算抉择，同时也减少了整个分布式查问优化的复杂性。

分布式查问优化须要更加精准的分布式代价模型

在查问优化中，代价是掂量一个执行打算好坏的规范，通常代价代表了一个执行打算的执行工夫或者对数据库系统资源的占用量，包含 CPU 资源、IO 资源、网络资源等。在单机执行中，代价模型通常只须要思考 CPU 和 IO 就能够。然而在分布式的场景中，除了思考 CPU 和 IO 的代价之外，还须要思考网络传输代价、查问的并行度以及一些分布式特定优化场景的代价，比方 bloom filter 的代价计算等。这些因素从根本上晋升了分布式代价模型设计和拟合的复杂性，也从肯定水平上减少了整个分布式查问优化的复杂性。

▋ OceanBase 3.x 的二阶段分布式查问优化办法

为了解决分布式查问优化带来的复杂性，跟业界的大部分解决方案相似，OceanBase 3.x 的版本采纳二阶段的分布式查问优化办法。

第一阶段: 假如所有的表都是本地的，依赖已有的单机查问优化能力抉择一个本地最优的执行打算。

第二阶段: 在固定连贯程序和本地算法的根底上，基于简略的分布式代价模型为每一个算子抉择一个分布式算法。

下图展现了一个二阶段的分布式查问优化办法的例子，其中右边表代表的是第一阶段生成的本地最优的执行打算，左边代表的是第二阶段生成的分布式打算。对于 Q1，在第一阶段，单机优化器抉择了一个如右边所示的本地最优的执行打算，其中 MJ、HJ 和 HGBY 别离代表了 merge join、hash join 和 hash group by 的本地算法。在第二阶段，在固定间断程序和本地算法的根底上，基于简略的分布式代价模型，为每一个算子抉择了一个分布式算法。在这个例子中，为 MJ 节点抉择了一个 partition wise join 的分布式连贯算法，为 HJ 节点抉择了一个 hash-hash 重分区的分布式连贯算法。

create table R1(a int primary key, b int, c int, d int) partition by hash(a) partitions 4;
create table R2(a int primary key, b int, c int, d int) partition by hash(a) partitions 4;
create table R3(a int primary key, b int, c int, d int) partition by hash(b) partitions 5;
select R2.c, sum(R3.d) from R1, R2, R3 where R1.a = R2.a and R2.C = R3.C group by R2.C;

二阶段分布式查问优化放大大简化了整个分布式查问优化的复杂度，然而 OceanBase 3.x 在大规模商用的过程中也遇到了很多因为二阶段导致的分布式查问优化不优的状况，上面咱们总结了比较突出的两大类问题。

没有思考分区信息导致抉择了不优的本地算法

二阶段的分布式查问优化通常会因为第一阶段优化时没有思考分区信息而抉择了不优的本地算法。思考如下图所示的一个查问 Q2 和它的第一阶段的打算，在第一阶段本地优化的时候，如果谓词 R1.c = 100 的选择率比拟低，那么满足这个条件的 R1 的行数会比拟少，这个时候优化器会抉择 nested loop join 来执行这个查问，即对于满足条件的 R1 中的每一行，通过 R2 上的索引 idx 疾速的获取满足条件的 R2 数据。然而在实在的执行过程中，咱们发现 nested loop join 的执行工夫远远比优化器预计的要大很多，起因是因为 R2 是一个蕴含 100 个分区的分区表，在执行 nested loop join 的过程中，对于 R1 中的每一行，都须要在 R2 的每个分区都执行一遍，那么这个执行工夫其实会扩充 100 倍。如果咱们把这个扩充 100 的执行工夫思考进去，那么最优的打算可能就是 hash join 而不是 nested loop join 了。在这个场景中，因为第一阶段的优化没有思考分区信息，所以在第一阶段会谬误的预计单机算子的代价，从而导致抉择了不优的本地算法。

create table R1(a int primary key, b int, c int);
create table R2(a int primary key, b int, c int, index idx(b)) partition by hash(a) partitions 100;
Q2: select * from R1, R2 where R2.b = R1.b and R1.c = 100;
/* 一阶段打算 */
| =============================================

|ID|OPERATOR        |NAME   |EST. ROWS|COST |
---------------------------------------------

|0 |NESTED-LOOP JOIN|       |970299   |85622|
|1 | TABLE SCAN     |r1     |990      |40790|

|2 | TABLE SCAN     |r2(idx)|1        |44   |
=============================================

Outputs & filters:
-------------------------------------

  0 - output([r1.a], [r1.b], [r1.c], [r2.a], [r2.b], [r2.c]), filter(nil),
      conds(nil), nl_params_([r1.b])
  1 - output([r1.b], [r1.c], [r1.a]), filter([r1.c = 100]),
      access([r1.b], [r1.c], [r1.a]), partitions(p0)
  2 - output([r2.b], [r2.a], [r2.c]), filter(nil),
      access([r2.b], [r2.a], [r2.c]), partitions(p0)

没有思考分区信息导致抉择了不优的连贯程序

二阶段的分布式查问优化通常因为在第一阶段没有思考分区信息而抉择了不优的连贯程序。思考如下的一个查问 Q3 和它所对应的两个本地打算和分布式打算，其中第一个打算抉择了 ((R2, R3), R1) 的连贯程序，第二个打算抉择了 ((R1, R2), R3) 的连贯程序。如果不思考分区信息，在第一阶段优化器可能会抉择 ((R2, R3), R1) 这样的连贯程序，然而这个连贯程序通过第二阶段之后可能会产生更多的网络传输代价，如下图所示，表 R1、R2、R3 以及 R2 和 R3 的连贯后果都须要通过网络传输。一个更好的间断程序可能是 ((R1,R2), R3)，因为这个连贯程序通过第二阶段之后只须要传输 R3 以及 R1 和 R2 的连贯后果 (R1 和 R2 因为能够做 partition wise join，所以是不须要做网络传输的)。这种因为没有思考分区信息而导致选错了谬误的连贯程序的场景在咱们的业务场景中也大量存在。

create table R1(a int primary key, b int, c int, d int) partition by hash(a) partitions 4;create table R2(a int primary key, b int, c int, d int) partition by hash(a) partitions 4;create table R3(a int primary key, b int, c int, d int) partition by hash(b) partitions 5;Q3: select R2.c, sum(R3.d) from R1, R2, R3 where R1.a = R2.a and R2.b = R3.b;

在如上的两个场景中，究其实质就是因为在第一阶段做优化的时候没有思考分区信息而抉择了不优的连贯程序和本地算法。通过这两个场景咱们也理解到了二阶段的分布式查问优化办法的毛病是不言而喻的，接下来咱们来聊一聊 OceanBase 4.0 是如何做分布式查问优化来解决这个问题的。

▋ OceanBase 4.0 的分布式查问优化

咱们认为分布式查问优化肯定要应用一阶段的办法，即要同时枚举本地算法和分布式算法并且应用分布式代价模型来计算代价，而不是通过分阶段的形式来枚举本地算法和分布式算法。OceanBase 4.0 重构了整个分布式查问优化办法，从原先的二阶段变成了一阶段的分布式查问优化办法。

为了不便咱们形容一阶段的分布式查问优化办法，这里咱们简略介绍一下 System-R 的 Bottom-up 的动静布局办法。给定一个 SQL 语句，System-R 用 bottom-up 的动静布局的办法来进行连贯枚举和连贯算法的抉择。给定一个 N 张表的连贯，该办法以 size 为驱动枚举每一个子集的执行打算。对于每一个枚举的子集，该办法通过如下的形式来获取最优的打算：

• 枚举所有单机的连贯算法，保护序这个物理属性，应用单机代价模型来计算代价。
• 保留代价最小的打算和存在有用序的打算，一个打算的序是有用的当且仅当该序对后续算子的调配有用。

下图展现了一个 4 张表的连贯枚举例子。该算法首先会枚举大小为 1 的基表的打算，对于每一张基表，该办法会枚举所有的索引并且保留代价最小和存在有用序的打算。而后该算法为枚举每个大小为 2 的子集的打算，比方在枚举 {R1,R2} 这两张表的连贯的时候，该办法会思考所有的单机的连贯算法，而后再正交上所有 R1 和 R2 保留的打算，最终达到枚举所有执行打算的目标。以此类推，该算法会持续枚举直至大小为 4 的子集的打算都曾经枚举实现。

基于已有的单机的 System-R 的查问优化办法，OceanBase 4.0 的分布式查问优化依照如下的形式工作:

1. 对于每一个枚举的子集，枚举所有算子的分布式算法，对于每一个分布式算法，OceanBase 应用分布式代价模型来计算代价，同时 OceanBase 会同时保护序和分区信息这两个物理属性。
2. 对于每一个枚举子集，除了保留代价最小的打算，保留存在有用序的打算，同时还须要保留有存在有用分区信息的打算。一个分区信息是有用的当且仅当它对后续的算子有用。思考下图所示的场景, 在该场景中，P1 采纳了 HASH-HASH 重分区的 HASH JOIN 办法, P2 采纳了对 R2 做 BROADCAST 的 HASH JOIN 办法，尽管 P2 的代价比 P1 的代价高，然而 P2 继承了 R1 的分区信息，对后续的 group by 算子是有用的，因而 P2 这个打算也会被保留。

create table R1(a int primary key, b int, c int, d int) partition by hash(a) partitions 4;
create table R2(a int primary key, b int, c int, d int) partition by hash(a) partitions 4;
select R1.a, SUM(R2.c) from R1, R2 where R1.b = R2.b group by R1.a;

OceanBase 4.0 应用了一阶段的分布式查问优化办法，相比于单机的查问优化，分布式查问优化的打算空间是十分大的。为了解决打算空间大的问题，OceanBase 4.0 创造了很多疾速裁剪打算的办法以及新增了新的连贯枚举算法来反对超大规模表的分布式打算枚举。 通过这些技术，OceanBase 4.0 大大减少了分布式打算空间，晋升了分布式查问优化的性能。同时咱们的试验后果也表明 OceanBase 4.0 能够在秒级内实现 50 张表的分布式打算的枚举。

OceanBase 4.0 如何晋升分布式执行引擎性能？

相比于 OceanBase 3.x 版本，OceanBase 4.0 在执行引擎方面做了很多方面的工作，其中包含实现了新的分布式和单机算法(比方 null-aware hash anti-join、shared broadcast hash join、hash-based window function、partition bloom filter 等)，欠缺了整个向量化引擎的实现，开发了极致的并行下压技术，开启了自适应技术的开发。这些引擎方面的工作都大大晋升了分布式查问和单机查问的性能。在这里咱们次要介绍一下 OceanBase 4.0 的自适应技术和并行下压技术。

▋ OceanBase 4.0 执行引擎开始朝着自适应的方向倒退

在 OceanBase 的业务场景中，咱们发现 OceanBase 执行引擎对优化器产生的不优的执行打算没有任何的容错能力，即一旦优化器产生了不优的执行打算，那么执行引擎在执行的时候是没方法做一些打算上的调整从而达到晋升性能的目标。尽管咱们通常说优化器的目标是给数据库的查问抉择一个最优的执行打算，然而从数据库倒退的历程来看，优化器本身存在很多解决不了的难题，比方优化器始终解决不了估行不精确的问题，所以优化器有可能会选到一个不优的执行打算甚至是一个十分差的执行打算。

为了解决这个问题，OceanBase 4.0 执行引擎开始朝着自适应的方向倒退。自适应技术是指执行引擎依据以后的执行状态来辨认进去一部分打算不优的场景，通过动静调整执行打算从而达到晋升执行性能的目标。咱们认为一个执行引擎倒退到肯定阶段肯定要通过自适应技术来尽量解决优化器产生的不优的执行打算的问题，当然咱们也不认为自适应技术可能解决掉所有的打算不优的场景。

OceanBase 4.0 实现了自适应的 Group by/Distinct 并行下压技术，它能够解决 Group by/Distinct 并行下压场景中因为打算不优而导致的性能回退问题。在正式介绍该自适应技术之前，咱们首先简略介绍一下 Group by/Distinct 并行下压技术。Group by/Distinct 并行下压技术是分布式执行中一种常见的并行下压技术，它的核心思想是提前把 Group by 算子下压上来做局部的数据预聚合，通过预聚合的形式能够缩小网络传输从而达到晋升性能的目标。 下图展现了一个 Group by 并行下压的执行打算的例子，其中 5 号算子就是下压的 Group by 算子，通过 5 号算子的预聚合能够缩小 4 号算子网络传输从而达到性能晋升的目标。然而这里须要留神的是 Group by 并行下压不肯定会带来性能上的晋升，有时候也会导致性能回退，次要起因是因为下压的 Group By 算子会引来额定的计算代价，所以只有当网络传输带来的性能晋升超过下压的 Group By 带来的计算开销，Group by 的并行下压才会带来收益。

create table R1(a int primary key, b int, c int) partition by hash(a) partitions 4;
explain select b, sum(c) from R1 group by b;
| ==========================================================

|ID|OPERATOR                     |NAME    |EST. ROWS|COST|
----------------------------------------------------------

|0 |PX COORDINATOR               |        |1        |10  |
|1 | EXCHANGE OUT DISTR          |:EX10001|1        |10  |
|2 |  HASH GROUP BY              |        |1        |9   |
|3 |   EXCHANGE IN DISTR         |        |1        |9   |
|4 |    EXCHANGE OUT DISTR (HASH)|:EX10000|1        |8   |
|5 |     HASH GROUP BY           |        |1        |8   |
|6 |      PX PARTITION ITERATOR  |        |1        |7   |

|7 |       TABLE SCAN            |r1      |1        |7   |
==========================================================

Outputs & filters:
-------------------------------------

  0 - output([INTERNAL_FUNCTION(r1.b, T_FUN_SUM(T_FUN_SUM(r1.c)))]), filter(nil), rowset=256
  1 - output([INTERNAL_FUNCTION(r1.b, T_FUN_SUM(T_FUN_SUM(r1.c)))]), filter(nil), rowset=256, dop=1
  2 - output([r1.b], [T_FUN_SUM(T_FUN_SUM(r1.c))]), filter(nil), rowset=256,
      group([r1.b]), agg_func([T_FUN_SUM(T_FUN_SUM(r1.c))])
  3 - output([r1.b], [T_FUN_SUM(r1.c)]), filter(nil), rowset=256
  4 - (#keys=1, [r1.b]), output([r1.b], [T_FUN_SUM(r1.c)]), filter(nil), rowset=256, dop=1
  5 - output([r1.b], [T_FUN_SUM(r1.c)]), filter(nil), rowset=256,
      group([r1.b]), agg_func([T_FUN_SUM(r1.c)])
  6 - output([r1.b], [r1.c]), filter(nil), rowset=256
  7 - output([r1.b], [r1.c]), filter(nil), rowset=256,
      access([r1.b], [r1.c]), partitions(p[0-3])

OceanBase 在之前的版本中都是优化器通过计算代价来决定是否要下压 Group by 算子，然而因为优化器有时会谬误的预计行数，会导致呈现没有正确的下压 Group by 算子或者谬误的下压了 Group by 算子的场景，最终导致执行性能次优。为了解决这个问题，OceanBase 4.0 引入了自适应的 Group by/Distinct 并行下压技术，其核心思想是让优化器总是下压 Group by/Distinct 算子，而后在执行的时候通过采样下压算子的一部分数据来决定是否跳过下压的 Group by/Distinct 算子。该技术的难点在于如何判断下压的算子是否具备足够好的预聚合能力。OceanBase 采纳了管制下压算子的 HASH 表在 L3 cache 之内 (管制 Hash 表的性能) 以及多轮采样的策略 (确保数据间断非聚合性带来的误判) 来判断下压算子是否具备足够好的预聚合能力。其核心思想如下:

• 下压算子 hash 表尽量维持在 L2 cache (1M) 内, 如果预聚合成果不好，标记该 hash 表状态为舍弃。如果预聚合成果很好, 能够将 hash 表扩张到 L3 cache(10 M)，如果执行过程中发现须要更大的内存，标记该 hash 表为舍弃状态。
• 如果以后 hash 表的状态是舍弃状态，返回 hash 表内所有行并开释，从新建 hash 表，开启下一轮的采样查看。
• 如果间断 5 次采样查看预聚合成果都不好，就跳过以后下压的 Group by 算子。

这里须要留神的是，相比于齐全不下压的场景，自适应的 Group by/Distinct 并行下压会引入一些额定的 overhead，次要是在执行时须要对下压的 Group By/Distinct 算子做一些采样和计算来判断是否须要跳过该算子，然而通过咱们对各种数据分布的测试，这个额定的 overhead 基本上能够管制在 10% 之内，然而获取的性能晋升是十分大的。

除了自适应的 Group by/Distinct 下压技术之外，以后 OceanBase 4.0 也在摸索和实现更多新的自适应技术，包含自适应的创立和探测 bloom filter、自适应地调整 nested loop join 和 hash join，自适应地调整分布式的 broadcast 连贯和分布式的 hash-hash 重分区连贯等技术。咱们置信这些自适应的技术会把 OceanBase 的执行引擎能力晋升到一个新的级别，可能使整个执行引擎更加强壮，可能在优化器生成不优执行打算或者十分差的执行打算的时候晋升整个查问的性能。

▋ OceanBase 4.0 朝着极致的并行下压技术的方向倒退

分布式场景中的并行下压技术是指通过下压算子的计算从而达到晋升性能的目标。并行下压技术通常通过最大限度地利用并行度或者缩小数据网络传输来晋升分布式查问的性能。并行下压技术对分布式的查问性能晋升是非常明显的，在很多场景中都有数量级别的性能晋升。 前一个章节中介绍的 Group By/Distinct 并行下压技术就是一个比拟典型的并行下压的场景。相比于 OceanBase 3.x 的版本，OceanBase 4.0 实现了一套十分欠缺的并行下压技术，基本上笼罩了剖析类场景中的所有算子，包含 Group/Rollup/Window Function/Distinct 等。

上面这个表格比拟了 OceanBase 在 3.x 版本和 4.0 版本的并行下压技术上的区别。

OceanBase 4.0 中每个算子的并行下压技术的实现都是不一样的，思考到并行执行的复杂性，每种实现都面临不一样的挑战。因为文章篇幅的起因，这里咱们不一一介绍每一种并行下压技术，咱们通过 OceanBase 对于解决蕴含 distinct 去重的聚合函数的三阶段并行下压技术来介绍一下并行下压技术的劣势。思考下图的例子，其中 Q1 蕴含了两个 distinct 去重的汇合函数，在 OceanBase 3.x 的版本中，Q1 是没方法做任何的并行下压的，从 Q1 的执行打算中也能够看进去，所有的去重逻辑和聚合逻辑都是在 0 号算子中计算，而且 0 号算子是不具备任何并行的能力的，这会导致整体的执行性能很差。

create table R1(a int, b int, c int, d int, primary key(a,b)) partition by hash(b) partitions 4;
Q1: select sum(distinct c), sum(distinct d) from R1 where a = 5;
| =====================================================

|ID|OPERATOR                |NAME    |EST. ROWS|COST|
-----------------------------------------------------

|0 |SCALAR GROUP BY         |        |1        |2365|
|1 | PX COORDINATOR         |        |3960     |2122|
|2 |  EXCHANGE OUT DISTR    |:EX10000|3960     |1532|
|3 |   PX PARTITION ITERATOR|        |3960     |1532|

|4 |    TABLE SCAN          |r1      |3960     |1532|
=====================================================

Outputs & filters:
-------------------------------------

  0 - output([T_FUN_SUM(distinct r1.c)], [T_FUN_SUM(distinct r1.d)]), filter(nil),
      group(nil), agg_func([T_FUN_SUM(distinct r1.c)], [T_FUN_SUM(distinct r1.d)])
  1 - output([r1.c], [r1.d]), filter(nil)
  2 - output([r1.c], [r1.d]), filter(nil), dop=1
  3 - output([r1.c], [r1.d]), filter(nil)
  4 - output([r1.c], [r1.d]), filter(nil),
      access([r1.c], [r1.d]), partitions(p[0-3])

为了解决这种蕴含 distinct 的聚合函数的分布式执行性能，OceanBase 在 4.0 引入了三阶段并行下压的逻辑。咱们用下图中蕴含一个 distinct 去重的聚合函数的场景来简略介绍一下三阶段并行下压的大体逻辑。三阶段并行下压逻辑次要包含三个阶段：

第一阶段: 下压 distinct 逻辑去做数据局部去重，这里对应了下图中的 6 号算子。

第二阶段： 依照去重列做一次数据重分区，而后做齐全去重和局部预聚合计算，这里对应了下图中的 3~5 号算子。

第三阶段： 把第二阶段的后果做最终的聚合，这里对应了下图中的 0-2 号算子。
相比于不做任何的下压，这里三阶段并行下压有两个性能上的益处。首先三阶段并行下压能够最大限度地利用并行度去做数据去重和数据预聚合。其次通过下压 distinct 做数据局部去重能够缩小网络传输。

create table R1(a int, b int, c int, d int, primary key(a,b)) partition by hash(b) partitions 4;
select sum(distinct c) from R1 where a = 5;
| ===========================================================

|ID|OPERATOR                      |NAME    |EST. ROWS|COST|
-----------------------------------------------------------

|0 |SCALAR GROUP BY               |        |1        |1986|
|1 | PX COORDINATOR               |        |1        |1835|
|2 |  EXCHANGE OUT DISTR          |:EX10001|1        |1835|
|3 |   MERGE GROUP BY             |        |1        |1835|
|4 |    EXCHANGE IN DISTR         |        |1        |1683|
|5 |     EXCHANGE OUT DISTR (HASH)|:EX10000|1        |1683|
|6 |      HASH GROUP BY           |        |1        |1683|
|7 |       PX PARTITION ITERATOR  |        |3960     |1532|

|8 |        TABLE SCAN            |r1      |3960     |1532|
===========================================================

Outputs & filters:
-------------------------------------

  0 - output([T_FUN_SUM(T_FUN_SUM(distinct r1.c))]), filter(nil),
      group(nil), agg_func([T_FUN_SUM(T_FUN_SUM(distinct r1.c))])
  1 - output([T_FUN_SUM(distinct r1.c)]), filter(nil)
  2 - output([T_FUN_SUM(distinct r1.c)]), filter(nil), dop=1
  3 - output([T_FUN_SUM(distinct r1.c)]), filter(nil),
      group(nil), agg_func([T_FUN_SUM(distinct r1.c)])
  4 - output([r1.c]), filter(nil)
  5 - (#keys=1, [r1.c]), output([r1.c]), filter(nil), dop=1
  6 - output([r1.c]), filter(nil),
      group([r1.c]), agg_func(nil)
  7 - output([r1.c]), filter(nil)
  8 - output([r1.c]), filter(nil),
      access([r1.c]), partitions(p[0-3]

下面咱们介绍了只包含一个 distinct 去重的聚合函数的三阶段并行下压解决，这里有一个问题是如果蕴含多个 distinct 的聚合函数，三阶段下压技术是否还能够工作？答案是必定的，这里的解决技巧在于对于蕴含 N 个 distinct 去重的聚合函数的场景，在第一阶段的时候，为每一个蕴含 distinct 的聚合函数，咱们会冗余一份数据并且标记这一份数据属于这个聚合函数的，剩下的第二阶段和第三阶段的解决基本上都是相似的，会有一些实现上的小差异。下图展现了 OceanBase 中蕴含 2 个 distinct 的聚合函数的三阶段下压例子，其中 aggr_code 就是用来标记不同的 distinct 所冗余的数据。

create table R1(a int, b int, c int, d int, primary key(a,b)) partition by hash(b) partitions 4;select sum(distinct c), sum(distinct d) from R1 where a = 5;| ===========================================================|ID|OPERATOR                      |NAME    |EST. ROWS|COST|-----------------------------------------------------------|0 |SCALAR GROUP BY               |        |1        |13  ||1 | PX COORDINATOR               |        |2        |13  ||2 |  EXCHANGE OUT DISTR          |:EX10001|2        |12  ||3 |   HASH GROUP BY              |        |2        |11  ||4 |    EXCHANGE IN DISTR         |        |2        |10  ||5 |     EXCHANGE OUT DISTR (HASH)|:EX10000|2        |9   ||6 |      HASH GROUP BY           |        |2        |8   ||7 |       PX PARTITION ITERATOR  |        |1        |7   ||8 |        TABLE SCAN            |r1      |1        |7   |===========================================================Outputs & filters:-------------------------------------  0 - output([T_FUN_SUM(T_FUN_SUM(dup(r1.c)))], [T_FUN_SUM(T_FUN_SUM(dup(r1.d)))]), filter(nil), rowset=256,      group(nil), agg_func([T_FUN_SUM(T_FUN_SUM(dup(r1.c)))], [T_FUN_SUM(T_FUN_SUM(dup(r1.d)))])  1 - output([AGGR_CODE], [T_FUN_SUM(dup(r1.c))], [T_FUN_SUM(dup(r1.d))]), filter(nil), rowset=256  2 - output([AGGR_CODE], [T_FUN_SUM(dup(r1.c))], [T_FUN_SUM(dup(r1.d))]), filter(nil), rowset=256, dop=1  3 - output([AGGR_CODE], [T_FUN_SUM(dup(r1.c))], [T_FUN_SUM(dup(r1.d))]), filter(nil), rowset=256,      group([AGGR_CODE]), agg_func([T_FUN_SUM(dup(r1.c))], [T_FUN_SUM(dup(r1.d))])  4 - output([AGGR_CODE], [dup(r1.c)], [dup(r1.d)]), filter(nil), rowset=256  5 - (#keys=3, [AGGR_CODE], [dup(r1.c)], [dup(r1.d)]), output([AGGR_CODE], [dup(r1.c)], [dup(r1.d)]), filter(nil), rowset=256, dop=1  6 - output([AGGR_CODE], [dup(r1.c)], [dup(r1.d)]), filter(nil), rowset=256,      group([AGGR_CODE], [dup(r1.c)], [dup(r1.d)]), agg_func(nil)  7 - output([r1.c], [r1.d]), filter(nil), rowset=256  8 - output([r1.c], [r1.d]), filter(nil), rowset=256,      access([r1.c], [r1.d]), partitions(p[0-3])

分布式并行下压的场景是一个比拟常见的客户场景，在 OceanBase 3.x 的版本中，咱们也遇到了不少因为并行下压性能的不欠缺导致的分布式查问性能问题。咱们置信在 OceanBase 4.0 能够很好地解决这类问题，晋升分布式查问的性能。

写在最初

文章的最初，咱们心愿和大家分享，OceanBase 4.0 的分布式性能晋升实际效果。相比于 OceanBase 3.x 版本，OceanBase 4.0 实现了全新的分布式代价模型和分布式查问优化框架、开发了一套十分欠缺的并行下压技术，开启了自适应技术的开发。这些技术的开发驱动一方面来自于咱们对客户需要的了解，另一方面也来自于咱们本人对分布式系统的了解。

为测试 Oceanbase 4.0 版本这些技术的工作成果，咱们在 TPC-DS 100GB 上进行了测试，试验结果表明 OceanBase 4.0 的分布式性能晋升效果显著，TPC-DS 100GB 的 99 个查问的执行工夫总和从 918s 降落到了 270s，在本文的最初，大家也能够看到 TPC-DS 100GB 上其中一部分查问在 OceanBase 3.x 版本和 4.0 版本的理论性能比照。

TPC-DS 100GB 性能测试比照（OceanBase 3.x vs. 4.0)

以上是咱们对 OceanBase 4.0 分布式性能查问价值及技术演进的思考。数据库的实质是根底软件，站在软件「使用者」的角度来看，咱们心愿在将来的 4.x 版本中，通过分布式查问优化和执行引擎技术的创新能力，帮忙用户带来更易用的应用体验和更疾速的查问性能。