浅析OceanBase数据库的向量化执行引擎

本篇博客是偏数据库系统概念性的内容，不会深入到 OceanBase 中各个算子和表达式的在向量化中的详细设计和实现。

背景

为了提升OceanBase社区版用户解决问题的效率，OceanBase官方不久前推出了《OceanBase 从入门到实践》系列课程。在第七期直播课程后，我们收到了不少用户的提问，表示在查阅学习资料或笔记时，对其中提到的诸如“rowset=16”或“rowset=256”这样的信息存在理解上的困惑，不清楚这其具体含义。如下代码中的示例。这里的 rowset 信息和 OceanBase 执行引擎的向量化执行技术相关，这篇博客就正好作为 AP 性能专题的内容之一，在解答这个用户问题的同时，也为大家介绍一下 OceanBase 执行引擎的向量化执行技术。

obclient [test]> create table t1(c1 int, c2 int);
Query OK, 0 rows affected (0.203 sec)obclient [test]> explain select count(*) from t1 where c1 = 1000;
+------------------------------------------------------------------------------------+
| Query Plan                                                                         |
+------------------------------------------------------------------------------------+
| =================================================                                  |
| |ID|OPERATOR         |NAME|EST.ROWS|EST.TIME(us)|                                  |
| -------------------------------------------------                                  |
| |0 |SCALAR GROUP BY  |    |1       |4           |                                  |
| |1 |└─TABLE FULL SCAN|t1  |1       |4           |                                  |
| =================================================                                  |
| Outputs & filters:                                                                 |
| -------------------------------------                                              |
|   0 - output([T_FUN_COUNT_SUM(T_FUN_COUNT(*))]), filter(nil), rowset=16            |
|       group(nil), agg_func([T_FUN_COUNT_SUM(T_FUN_COUNT(*))])                      |
|   1 - output([T_FUN_COUNT(*)]), filter([t1.c1 = 1000]), rowset=16                  |
|       access([t1.c1]), partitions(p0)                                              |
|       is_index_back=false, is_global_index=false, filter_before_indexback[false],  |
|       range_key([t1.__pk_increment]), range(MIN ; MAX)always true                  |
+------------------------------------------------------------------------------------+
14 rows in set (0.033 sec)

火山模型执行引擎

向量化执行引擎是提升 AP 能力的利器之一，也为 2021 年 OceanBase TPCH 打榜夺冠做出了重要贡献。不过如果要学习向量化引擎，首先需要了解一下传统的数据库执行引擎模型 —— 火山模型。

火山模型（Volcano Model）也被称为迭代模型（Iterator Model），是最著名的查询执行模型，早在 1990 年就在论文《Volcano, an Extensible and Parallel Query Evaluation System》中被提出。大多数关系型数据库都采用了这种模型，例如 Oracle、MySQL、DB2、SQLServer 等等。

在火山模型中，一个查询计划会被分解为多个算子（Operator）。每个算子就是一个迭代器，都要实现一个 next() 接口，通常包括三个步骤：

- 调用儿子算子的 next() 方法，获取儿子算子返回的计算结果；
- 对儿子算子返回的计算结果，执行当前算子的计算动作，得到计算结果；
- 返回一行计算结果给父亲算子。

说明：
论文里算子的的 next() 接口，在 OceanBase 的代码里叫 ObOperator::get_next_row()。

通过火山模型，查询执行引擎可以优雅地将任意算子组装在一起，而不需要考虑每个算子的具体处理逻辑。查询执行时会由查询树自顶向下嵌套调用 get_next_row() ，数据则自底向上地被拉取处理。所以，这种处理方式也称为拉取执行模型（Pull Based）。为了更好地理解火山模型的拉取执行过程，这里继续看上面这个聚合计算的例子：

select count(*) from t1 where c1 = 1000;

说明：

图中的 Tuple 是一个元组，也就是下层算子向上层算子返回的一个计算结果行。

如上图所示：

- 步骤 1 - 步骤 3，聚合算子 Aggregate 先通过调用 get_next_row() 方法触发下面这些算子逐层调用孩子算子的 get_next_row() 方法。
- 步骤 4 - 步骤 6，TableScan 算子获取到存储层的数据，吐行给 Filter 算子，Filter 算子完成过滤条件 c1 = 1000 的计算，吐结果行给负责聚合计算的 Aggregate 算子。
- 步骤 7，Aggregate 算子通过反复调用 next 方法获取到需要的一批数据，最终完成聚合计算，返回计算结果。

把 OceanBase 的向量化开关关掉，就可以看到类似于上图的执行计划树。

-- 关掉向量化开关，让后面的 SQL 默认走类似于火山模型的单行计算模式
alter system set _rowsets_enabled = false;-- 大家可以发现，在下面这个计划里，看不到类似于上面那个计划里的 rowset 值了
explain select count(*) from t1 where c1 = 1000;
+------------------------------------------------------------------------------------+
| Query Plan                                                                         |
+------------------------------------------------------------------------------------+
| =================================================                                  |
| |ID|OPERATOR         |NAME|EST.ROWS|EST.TIME(us)|                                  |
| -------------------------------------------------                                  |
| |0 |SCALAR GROUP BY  |    |1       |6           |                                  |
| |1 |└─TABLE FULL SCAN|t1  |1       |6           |                                  |
| =================================================                                  |
| Outputs & filters:                                                                 |
| -------------------------------------                                              |
|   0 - output([T_FUN_COUNT_SUM(T_FUN_COUNT(*))]), filter(nil)                       |
|       group(nil), agg_func([T_FUN_COUNT_SUM(T_FUN_COUNT(*))])                      |
|   1 - output([T_FUN_COUNT(*)]), filter([t1.c1 = 1000])                             |
|       access([t1.c1]), partitions(p0)                                              |
|       is_index_back=false, is_global_index=false, filter_before_indexback[false],  |
|       range_key([t1.__pk_increment]), range(MIN ; MAX)always true                  |
+------------------------------------------------------------------------------------+
14 rows in set (0.010 sec)

OceanBase 的计划只有两个算子，比上面这张图上的看上去更简单些。因为 OceanBase 中的每一个算子都包含了 Filter 的功能，所以不需要独立的 Filter 算子。可以在上面的计划里看到，TABLE SCAN 算子里也有 filter([t1.c1 = 1000]) 的信息。计划里的 SCALAR GROUP BY 算子就是负责不需要进行 group by 分组的聚合计算，对应图中的 Aggregate 算子。

火山模型的优点是处理逻辑清晰，每个算子只要关心自己的处理逻辑即可，耦合性低。但是它的缺点也非常明显，主要是两点：

- get_next_row() 虚函数在每个算子处理每行数据时都要调用一次，调用次数过多，造成 CPU 资源的浪费。在 OLAP 查询数据量大的场景中，问题尤为明显。
- 数据以行为单位进行处理，不利于充分发挥现代 CPU 的特性。

向量化执行引擎及其优势

向量化模型最早是在 MonetDB-X100（Vectorwise）系统的论文《MonetDB/X100: Hyper-Pipelining Query Execution》中提出的，不同于传统的火山模型按行迭代的方式，向量化引擎采用按批迭代方式，可以在算子间一次传递一批数据。向量化引擎被提出后，因为可以有效提高 CPU 利用率并充分发挥现代 CPU 的特性，很快就被广泛应用到了现代数据库引擎设计中。

如上图所示，与数据库传统的火山模型迭代类似，向量化模型也是通过 PULL 模式从算子树的根节点层层拉取数据。区别于 get_next_row() 调用一次传递一行数据，向量化引擎通过 get_next_batch() 一次传递一批数据，并尽量保证该批数据在内存上紧凑排列。

减少虚函数调用的开销

向量化引擎第一个显而易见的优势，就是大幅减少了框架函数的调用次数。假设一张表有 1 亿行数据，按火山模型的处理方式，每个算子均需要执行 1 亿次 get_next_row() 的迭代才能完成查询。如果使用向量化引擎，假设一个向量的大小为 1024 行数据，那么在同样的查询中，get_next_batch() 的调用次数会小于 10 万次（ 1 亿 / 1024 = 97657 ），大大降低了虚函数调用次数，节省了 CPU 的开销。

在本文开头的用户问题里，计划中的 rowset 就是表示一个 batch（或者叫一个向量）中有多少行数据。

-- 打开向量化开关
alter system set _rowsets_enabled = true;-- 设置每个向量的大小是 16 行
alter system set _rowsets_max_rows = 16;-- 计划中可以看到执行时的向量值大小（rowset=16）
explain select count(*) from t1 where c1 = 1000;
+------------------------------------------------------------------------------------+
| Query Plan                                                                         |
+------------------------------------------------------------------------------------+
| =================================================                                  |
| |ID|OPERATOR         |NAME|EST.ROWS|EST.TIME(us)|                                  |
| -------------------------------------------------                                  |
| |0 |SCALAR GROUP BY  |    |1       |4           |                                  |
| |1 |└─TABLE FULL SCAN|t1  |1       |4           |                                  |
| =================================================                                  |
| Outputs & filters:                                                                 |
| -------------------------------------                                              |
|   0 - output([T_FUN_COUNT_SUM(T_FUN_COUNT(*))]), filter(nil), rowset=16            |
|       group(nil), agg_func([T_FUN_COUNT_SUM(T_FUN_COUNT(*))])                      |
|   1 - output([T_FUN_COUNT(*)]), filter([t1.c1 = 1000]), rowset=16                  |
|       access([t1.c1]), partitions(p0)                                              |
|       is_index_back=false, is_global_index=false, filter_before_indexback[false],  |
|       range_key([t1.__pk_increment]), range(MIN ; MAX)always true                  |
+------------------------------------------------------------------------------------+
14 rows in set (0.021 sec)

充分发挥现代 CPU 的特性

数据紧密排列，提升 Cache 友好性

OceanBase 在向量化执行过程中，会尽量保证该批数据在内存上紧凑排列，产生的中间数据会以列的形式进行组织。比如我们一个 batch 是 256 行数据, 那我们内存排布中 c1 的 256 行数据就会都连续存放, 然后 c2 的 256 行数据也连续存放, 计算 concat(c1, c2) 表达式还是 256 行一起计算，最后将结果放入到该表达式预分配的结果值的内存空间中。

由于中间数据连续，CPU 就可以通过预取指令快速把即将进行计算的数据提前加载到 L2 cache 中，以此减少 memory stall 的现象，提升 CPU 的利用率。在算子函数内部，函数也不再是一次处理一行数据，而通过批量处理连续数据的方式进行计算，可以提升 CPU DCache 和 ICache 的友好性，减少 Cache Miss。

减少分支预测失败对 CPU 流水线的破坏

论文《DBMSs On A Modern Processor: Where Does Time Go?》中介绍了分支预测失败对数据库性能的影响。由于 CPU 中断了流水执行，重新刷新流水线，因此分支预测失败对数据库处理性能的影响很大。SIGMOD13 的论文《Micro Adaptivity in Vectorwise》也对分支在不同选择率下的执行效率有详细论述，见下图。

由于数据库 SQL 引擎逻辑比较复杂，在火山模型下，条件判断逻辑出现频率普遍较高。

// 单行计算流程的伪代码，处理 256 行数据，要进行 256 次 if 的判断：
for (auto row_no : 256) {get_next_row() {if (A) {eval_func_A();} else if (B) {eval_func_B();}}
}

但在向量化执行时，可以在算子和表达式内部，最大限度地避免条件判断。例如避免在 for 循环内部出现 if 判断，从而避免分支预测失败对 CPU 流水线的破坏，大幅提升 CPU 的处理能力。

// 向量化计算流程伪代码，处理 256 行数据，只需要进行 1 次 if 的判断：
get_next_batch() {if (A) {for (auto row_no : 256) {eval_func_A();}} else if (B) {for (auto row_no : 256) {eval_func_B();}}
}

利用 SIMD 指令加速计算

由于向量化引擎处理内存连续数据，因此向量引擎可以很方便的把一批数据装载到向量寄存器中。然后通过 SIMD 指令，替换传统的标量（Scalar）算法，进行向量（Vector）计算。SIMD 指令可以使 CPU 同时对一批数据并行执行相同的计算动作，从而减少这批数据计算所需要的 CPU 的周期数。

上图右侧展示了典型的 SIMD 计算，两组四个连续存储的数据元素被并行计算，CPU 会根据 SIMD 指令，对每对相应的数据元素（A1 和 B1，A2 和 B2，A3 和 B3，A4 和 B4）同时执行相同的运算。四个并行计算的结果也会被连续存储。

假设我们的处理器支持 4 个元素的 SIMD 乘法操作，那就意味着它有特殊的寄存器（通常称为向量寄存器），可以同时存储四个整型数。因为 OceanBase 的向量化执行过程中数据会紧密存储，所以就可以按照如下流程编写 SIMD 代码：

数据加载（_mm_loadu_si128）：首先，我们将向量 A1 A2 A3 A4 和 B1 B2 B3 B4 的各个元素加载到两个 SIMD 寄存器中。
单一指令乘法（_mm_mullo_epi32）：接着，我们使用单一的 SIMD 乘法指令，同时对两个寄存器中的所有元素执行乘法操作。
数据存储（_mm_storeu_si128）：最后，我们将结果从 SIMD 寄存器存储回为结果分配的内存中，得到结果向量。

// SSE（针对x86架构）的 C++ 伪代码示例，使用 SIMD 技术进行整型向量的逐元素相乘
#include <immintrin.h> // 包含 SSE 指令集头文件// 函数用于计算两个整型向量的逐元素相乘
void simdIntVectorMultiply(const int* vec1, const int* vec2, int* result, size_t length) {// 确保向量长度是 4 的倍数，因为 SSE 寄存器一次处理 4 个 32 位整数assert(length % 4 == 0);// 使用 SSE 指令进行优化的循环for (size_t i = 0; i < length; i += 4) {// 加载四个整数到 xmm 寄存器（128位）__m128i vec1_simd = _mm_loadu_si128(reinterpret_cast<const __m128i*>(vec1 + i));__m128i vec2_simd = _mm_loadu_si128(reinterpret_cast<const __m128i*>(vec2 + i));// 执行向量乘法__m128i product_simd = _mm_mullo_epi32(vec1_simd, vec2_simd);// 存储结果回内存_mm_storeu_si128(reinterpret_cast<__m128i*>(result + i), product_simd);}
}