前言

在线处理分析场景对于 YMatrix 而言具有十分重要的意义，如何在该场景上使性能有数量级的提升呢？我们通过向量化执行引擎和列存系统，实现了性能在 TPC-H 基准测试上达到 GPDB 的十倍。本文将以 Hash Join 为切入点，介绍向量化执行引擎的基本原理，并深入分析、探讨 Hash Join 的算法和工程实践。

作者：数据库内核研发工程师 申磊

TPC-H 是一套业内常用的决策支持基准测试，由 TPC 委员会制定发布，用于评测数据库的分析型查询能力。TPC-H 查询包含 8 张数据表、22 条复杂的 SQL 查询，大多数查询包含若干表 Join、子查询和 Group-by 聚合等。所选查询和数据库中的数据具有广泛的行业适用性。这个基准测试展示了决策支持系统的能力，可以检查大量数据、执行高度复杂查询、回答关键业务问题；同时反映了数据库系统处理查询的多方面能力。正因如此，我们选择 TPC-H 作为我们优化目标。

Greenplum（以下简称 GPDB）是全球领先的开源、并行大数据平台，专为分析、机器学习和 AI 而打造。GPDB 大数据平台基于 MPP（大规模并行处理）架构，具有强大的内核技术，包括数据水平分布、并行查询执行、专业优化器、线性扩展能力、多态存储、资源管理、高可用、高速数据加载等。从商业智能（BI）、文本、GIS、图、图像、流式数据处理到各种机器学习算法都能支持。此外，GPDB 提供的查询优化器是业界第一个开源的基于代价的查询优化器，专为大数据负载而设计。它可以将交互式和批处理模式分析扩展到 PB 级的大型数据集，而不会降低查询性能和吞吐量。

YMatrix 是基于 PostgreSQL / Greenplum 经典系开源数据库开发的超融合数据库产品。当前主要针对时序场景，而大部分时序数据库的查询场景可以认为是在线分析处理（OLAP）场景，因此我们选择业界领先的分析数据平台 GPDB 作为比较对象。

在打造向量化执行引擎开始时，我们就定了第一个小目标：性能是 GPDB 十倍。原因很简单，数据库作为基础软件，替换成本比较高，包括前期评估、测试成本、数据层面迁移成本、业务层面迁移成本、对新数据库完全掌控的成本、对业务迭代影响的成本，以及相关配套工具改造成本等等。整个过程需要 DBA、基础架构团队和业务团队协作，如果只是为了得到 50% 或一倍的性能提升，并不值得大动干戈。

本文以 TPC-H 查询为主线，讨论分析 YMatrix 如何在这套基准测试上达到GPDB性能的 10 倍。（测试报告全文）

一、向量化执行引擎原理

首先，我们看一下 GPDB 的执行模型——火山模型——的优劣势。

1.1 火山模型

火山模型是数据库领域非常成熟的解释计算模型。该模型将关系代数中每一种操作抽象为一个 Operator，将整个查询构建成一个 Operator 树，从根节点到叶子节点自上而下地递归调用 next() 函数。

优势是实现简单，每个 Operator 单独实现即可。

缺点也很明显：

• 函数调用开销：每条 Tuple 在各个节点之间流动时会涉及大量函数调用，虽然对于单条 Tuple 而言开销并不大，但考虑到数据规模，火山模型整体函数调用开销是显著的。如果涉及虚函数、函数指针，开销会更大一些；

• 缓存不友好：火山模型每次处理一个 Tuple 的模式和过多的控制语句、函数调用等使得缓存失效的概率增加；

• 无法充分利用现代 CPU 超标量能力。

几十年前，这个模型与当时的硬件能力相适应，能够匹配当时 CPU 和内存之间的速度比例。经过几十年发展，硬件有了长足的进步，值得注意的有：

• 相比内存性能的提高，CPU 性能提高的更多；

• 磁盘容量增长比 I/O 带宽增长更快，带宽是更稀缺的资源。

为了弥补火山模型的不足，并充分发挥现代硬件的能力，列存系统和向量化执行引擎应运而生。它们通过改变架构解决了上述两点导致的性能瓶颈转移问题，从最早的研究原型，转变为如今的数据库标配。

所以，如果想要使 YMatrix 性能在 TPC-H 基准测试上达到 GPDB 的十倍，打造列存系统和向量化执行引擎是必然选择。本文关注点是向量化执行引擎。

1.2 基于火山模型的向量化执行引擎

和常见的向量化执行引擎一样，我们的做法是基于火山模型的微批处理模式。从逻辑控制角度看，仍旧是火山模型；从数据角度看，一次处理一批数据（N 个 Tuple）而不是一个 Tuple。

数据按列存放，带类型信息，紧密排布，一个简单的 for 循环就能高效地处理一批数据，如下图所示。

这就弥补了火山模型的劣势，也是向量化的优势：

减少函数调用开销：显而易见，函数调用开销被均摊了，只有原先的 1/N 。除了火山模型中最常见的 next() 之外，很多和业务相关的函数开销也被均摊了。比如写缓冲区，如果一次处理一个 Tuple，每次都必须检查缓冲区是否有足够的剩余空间；如果一次处理一批，成本就会被均摊，相当于 1/N 个 Tuple 检查一次缓冲区。 更好的缓存局部性：数据跳转和指令跳转是流水线的大敌，更好的局部性意味着更好的性能。比如数据都在 L1 中，直接读取比从 L2 读取数据快 14 倍（数据来源参考：https://gist.github.com/jboner/2841832）从数据缓存角度看，一个批次的大小设置成能放进缓存的大小是合适的。如果过大，超过 L1 缓存大小甚至超过全部共享缓存大小，数据需要从内存（或共享缓存）传输数据到 L1 缓存，使查询执行速度变慢。从指令缓存角度看，向量化函数的命令会迭代和批大小一样的次数，而不是像面向行的执行引擎那样执行 A 函数之后接着执行 B 函数等，指令局部性更好。

图以 select a + b from t where a % 2 = 0 为例演示了火山模型和向量化的执行流程。左上角方框表示数据在内存中的布局，一个是行存，一个是列存。算子左侧方框表示当前正在处理的数据。从图中可以看出，向量化数据和指令的局部性都更好，函数调用次数和 if-else 分支都更少。

此外，向量化的优势还有：

• 充分利用编译器能力：如上所述，通过循环遍历数据紧密排布的原始类型数组及一些编程技巧，我们可以让编译器将我们的 C/C++ 代码自动编译为 SIMD 指令等更高效的形式。后续小节会详细介绍 SIMD 指令；

• 自适应执行路径：我们根据运算的复杂程度选择不同的执行路径。对于像加法运算这样开销很低的算子，无论对应 Tuple 是否被过滤，向量化执行引擎都会对所有数据做处理。虽然会有些额外的运算，但是因为循环中没有 if-else 分支，所以能够生成SIMD 指令，执行速度比针对有效数据一个一个计算效率要高。下面讲解微批的优势，同时也是一个动态选择路径的例子；

• 性能分析：由于均摊，收集性能数据的成本也比一次处理一个 Tuple 要小很多，这使得向量化引擎能够提供更详细的关于 CPU 开销的性能指标分析。如果面向行的执行引擎做同样细致分析，分析本身相对于处理的占比会增高，影响性能。通过更详细的性能分析做数据支持，更有助于做出正确的优化决策，从而提升性能，形成正向反馈;

• 并行内存访问：在现代 CPU 上向量化循环执行内存访问的算法能够针对向量中不同值生成多个待处理的缓存未命中（ outstanding cache misses ）。这是因为当缓存未命中发生时，现代 CPU 能够提前推测，可以同时有多个待处理的缓存未命中。现代计算机对此支持的很好，基于此，内存带宽才能更好地被利用。工程实践的第一小节“搜索优化”是一个很好的例子。

一批处理多少数据呢？上限应该满足输入数据、输出数据外加辅助信息都能够放到 L1 缓存中这个条件，目的是避免频繁读写内存，充分利用缓存将这一批数据处理完；下限是至少能够利用 SIMD 指令做计算，同时能够达到有效均摊附加开销的目的。

不管为表示一批数据是否为 NULL，还是为表示这批数据是否被过滤条件选中，都使用一个 bitmap 字段表示，本质上是一个 uint64 数组，其中每个比特表示一个 Tuple 是否为 NULL 或者被过滤掉了。多个算子的结果基于 bitmap 位运算完成。当需要对这些数据做处理时，以很低代价判断出数据特征，选择不同方式执行计算，尽可能减少分支跳转，同时利用编译器优化能力达到更好的性能。

if (bitmap.Full() && bitmap.AllSet()) { // 满批，且全匹配
    // 编译器有机会进行循环展开、生成 SIMD 指令等优化处理
    // 同时，这里不需要再对 bitmap 进行检查
    for (auto i = 0; rowid  0)
    tochecktmp = tocheck
    tocheck.clear();
    foreach rowid in tochecktmp
        if(!CheckEqual())
            match[rowid] = next[match[rowid]];
            if (match[rowid] != 0)
                tocheck.add(rowid);

2.2.2 Join Key 类型

在第一版中，HashMap 中 Key 的类型只有两种，一种是 long，目的是处理 Join Key 数量唯一且其类型是整数的情况，另一种是 string，处理其他情况，比如两个 Join Key，先把这两列数据追加到 string 中，然后将其视作二进制来计算哈希值和判等。

仍旧以 Q9 为例，这里涉及五个 Join，四个是单列 Join，类型包含 int32 和 int16，一个是两列 Join，这两列的类型都是 int32。第一版设计的 Key 类型虽然能覆盖所有场景，但是和具体查询有一点差别，如果能消除这个差距的话性能肯定会有提升。

如果 Join Key 类型是 int，下层算子返回的是 BATCH_SIZE 个紧密排列的数 int，但是往 HashMap 插入时，参数是 long 数组，这里不得不拷贝一次，一行一行赋值。解法是拓展 Key 的类型，以避免拷贝，或者只需块级别拷贝一次。

如果 Join Key 有两个，都是整数，使用 string 作为 Key 类型，也需要拷贝，并且计算哈希值（详见下一小节）和判等都会比较慢。再次拓展 Key 类型，将两个整数 Key 放到一起，这时额外需要一个 bool，表示该给哪个 Key 进行赋值。这里还是有数据的拷贝，不过向 string 中拷贝是把整数当做二进制，使用 std::copy 处理，而新的 Key 类型是带类型的整数集合，直接用 = 赋值即可，编译结果是 mov 指令，比内存拷贝快很多。除了计算哈希值能快外，判等也是两个带类型的整数比较，比按二进制比较 string 快很多。

2.2.3 哈希值的计算

在整个 Hash Join 中，计算哈希值是很重要的，性能只是一个方面，哈希函数的质量（碰撞和随机性）也是重要指标。经过对性能和质量两个方面的实际测试，我们选择 xxHash 的 XXH3 生成 std::uint64_t 类型的哈希值。

通过分析 TPC-H 查询，绝大多数都只是单列等值 Join，且类型是整数。如果能够提升计算整数类型哈希值的性能，基本上所有查询都能得到性能提升。

在一开始的实现中，不管是 string 还是整数值类型，都交给 xxHash 计算哈希值。xxHash 性能很好，不过对于整数值，也是当做二进制计算，没有利用到类型信息。简单地乘以一个大的质数作为哈希值是否可行呢？是否能更快呢？

经过性能测试，乘以大质数的速度是 xxHash 的 7 倍，能够进一步提升性能。利用 xxHash 自带的质量检测工具，质量和 xxHash 性能一致，甚至在有些高比特位上表现更好。

对于 std::uint8_t 和 std::uint16_t 两个类型，直接返回数值本身作为哈希值，同时，桶数设置为 $$256=2^8$$,$$65536=2^{16}$$ 个。

读者可能会有疑问，如果 Join Key 是 std::uint32_t 或者 uint64_t 类型，为什么不能直接返回数值本身作为哈希值呢？

这两个类型的值域非常大，不可能像一字节和两字节整数一样，桶数等于值域集合大小，所以不得不从哈希值映射到对应的桶。我们选择 2 的幂次作为桶数，得到哈希值之后，通过与运算得到 bucketid，放到对应的桶里面。简单的与运算只用了低若干位的信息，而高位信息被丢弃了。

如果 Join Key 列的数值完全随机分散，直接使用整数值做为哈希值没有任何问题。但是实际生产场景分布往往与业务相关，在某些情况下，数据隐含某种规律，那么直接取低若干位可能会导致冲突率升高。如果计算一次哈希值，利用到所有比特的信息，会改变数据分布，更趋于完全随机分散，冲突率就会降低。对于 HashMap 而言，冲突率的高低是影响性能的一个重要因素。

对于上一个小节提到的将两个整数打包的 Hash Key 类型，计算哈希值的算法是

std::uint64_t hashcode() const {
    auto h1 = data1_.hashcode();
    auto h2 = data2_.hashcode();
    return ((h1  '2022-01-01' AND region = 2;

从存储引擎角度看，如果一个块的 ts 上最大值小于 2022-01-01，存储引擎不再读取这一块，节省 I/O 资源，同时，也减少了执行引擎的计算量。

向量化执行引擎会也能利用 minmax 信息加速计算。比如某块 region 的最大值和最小值都是 2，执行引擎只会在该块应用谓词 ts > '2022-01-01' 过滤出符合条件的数据，而不会进行任何和 region 有关的运算，提高性能。

这两个点也告诉我们，性能具备整体性。单打独斗，只实现一个极致的向量化执行引擎是远远不够的。

四、展望

对性能的追求没有尽头，我们的目标始终是更快更快还是更快！

随着 YMatrix 5.0 发布，向量化执行引擎面世，但这只是起点，未来还需进一步提升性能，超越过去的自己。

从算法层面到工程实践，都有很多需要做的事情。比如当前还没有很好地实现 Merge Join 算法，而当数据已经有序或者上层算子需要有序输出时， Merge Join 可能是更好的选择。再比如文中提到的微批化处理的优劣势，可以大块（1000 行以上为一个处理单元）与微批相结合，利用好各自的优势。

除了执行引擎本身，还可以考虑和存储引擎深度整合。比如利用好列存系统的统计信息，简化诸如聚集等计算；也可以给压缩数据增加一些元数据，利用这些实现不解压直接计算，提升性能。