本篇博客是偏数据库系统概念性的内容,不会深入到 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);}
}
TPCH 性能测试
利用 TPCH 30T 数据集对 OceanBase 数据库进行 TPCH 测试,向量化执行模式相比单行执行模式,执行性能整体可以提升 2.48 倍。对于计算密集的 Q1 查询, 性能提升可以达到 10 倍以上。
在最新的 4.3 版本中,OceanBase 还对从 3.x 版本就已经支持的向量化执行引擎,进行了一次新的优化和重构,详细介绍和性能测试可以参考:这篇博客。
结语
这篇博客是 OceanBase 向量化执行引擎入门级的介绍,希望无论是 DBA 同学,还是内核研发同学,在阅读之后都能够有所收获。如果大家希望继续和作者讨论有关向量化执行的任何问题,都欢迎在这篇博客的评论区进行留言,我会第一时间对大家的问题进行回复。