向量化执行引擎的 SIMD 优化：从标量到向量的查询加速

国医中兴

155人浏览 · 2026-06-10 23:47:22

国医中兴 · 2026-06-10 23:47:22 发布

向量化执行引擎的 SIMD 优化：从标量到向量的查询加速

cover

一、标量执行的 CPU 瓶颈：逐行处理的性能天花板

传统数据库执行引擎采用 Volcano 模型——每个算子逐行调用 next() 获取数据，逐行处理。这种模型虽然实现简单，但 CPU 利用率极低——每次处理一行都需要函数调用开销、分支预测失败和缓存未命中。现代 CPU 的 SIMD 寄存器（AVX-512 可同时处理 16 个 32 位整数）在逐行处理模式下完全闲置。

生产环境中，向量化执行面临三个核心痛点：第一，内存布局不适配——列式存储是向量化的前提，但许多系统仍使用行式存储；第二，分支预测惩罚——条件过滤（WHERE 子句）导致 SIMD 通道内的分支发散，性能退化；第三，SIMD 编程复杂——手写 SIMD intrinsics 可读性差、可维护性低，且不同 CPU 架构需要不同实现。

这个问题的本质是：向量化执行需要从"逐行处理"转变为"批量处理"——一次处理一批数据（通常 1024 行），利用 SIMD 指令并行计算，将 CPU 吞吐量提升 4-16 倍。

二、向量化执行的底层机制

flowchart TB
    subgraph Volcano模型["Volcano 逐行模型"]
        V1[next() → 处理1行] --> V2[next() → 处理1行]
        V2 --> V3[next() → 处理1行]
        V3 --> V4[...N次函数调用]
    end

    subgraph 向量化模型["向量化批量模型"]
        VEC1[next_batch() → 处理1024行] --> VEC2[next_batch() → 处理1024行]
    end

    subgraph SIMD并行["SIMD 并行计算"]
        SCALAR[标量: 1次加法/周期]
        SSE[SSE: 4次加法/周期<br/>128位寄存器]
        AVX2[AVX2: 8次加法/周期<br/>256位寄存器]
        AVX512[AVX-512: 16次加法/周期<br/>512位寄存器]
    end

关键机制解析：

批量处理：向量化引擎每次处理一批数据（batch），而非逐行。批大小通常为 1024 或 4096 行，与 CPU L1 缓存大小匹配。
SIMD 指令：单指令多数据——一条指令同时对多个数据执行相同操作。例如 _mm256_add_ps 一条指令同时完成 8 个单精度浮点加法。
分支消除：条件过滤使用掩码（mask）替代分支——SIMD 通道内所有元素都执行计算，通过掩码选择有效结果，避免分支预测惩罚。

三、向量化执行的实现

3.1 向量化过滤算子

import numpy as np

def vectorized_filter(column: np.ndarray, predicate) -> tuple:
    """
    向量化过滤算子
    使用NumPy的SIMD加速实现
    """
    # 批量评估谓词，生成布尔掩码
    mask = predicate(column)

    # 使用掩码提取满足条件的行
    filtered = column[mask]

    return filtered, mask


# 示例：WHERE amount > 100 AND status = 'PAID'
data = np.array([50, 150, 200, 80, 300, 120, 60, 250])
status = np.array(['PENDING', 'PAID', 'PAID', 'PAID',
                    'PAID', 'PENDING', 'PAID', 'PAID'])

# 向量化条件过滤
mask_amount = data > 100
mask_status = status == 'PAID'
combined_mask = mask_amount & mask_status

result = data[combined_mask]
# result: [200, 300, 250]

3.2 向量化聚合算子

def vectorized_aggregate(
    group_keys: np.ndarray,
    values: np.ndarray,
    agg_func: str = "sum",
) -> dict:
    """
    向量化聚合算子
    使用NumPy的分组操作实现
    """
    unique_keys = np.unique(group_keys)
    result = {}

    for key in unique_keys:
        mask = group_keys == key
        group_values = values[mask]

        if agg_func == "sum":
            result[key] = np.sum(group_values)
        elif agg_func == "avg":
            result[key] = np.mean(group_values)
        elif agg_func == "count":
            result[key] = len(group_values)
        elif agg_func == "min":
            result[key] = np.min(group_values)
        elif agg_func == "max":
            result[key] = np.max(group_values)

    return result