导读

BaikalDB 作为服务百度商业产品的分布式存储系统，支撑了整个广告库海量物料的存储和 OLTP 事务处理。随着数据不断增长，离线计算时效性和资源需求压力突显，基于同一份数据进行 OLAP 处理也更为经济便捷，BaikalDB 如何在 OLTP 系统内实现适合大数据分析场景的查询引擎以应对挑战？

01 BaikalDB 应对 OLAP 场景的挑战

BaikalDB 是面向百度商业产品系统的需求而设计的分布式存储系统，过去多年把商业内部几十套存储系统全部统一到 BaikalDB，解决了异构存储带来的各种问题，支撑了整个广告库的海量物料存储和复杂的业务查询。BaikalDB 核心特点包括：

• 兼容 mysql 协议，支持分布式事务：基于 Raft 协议实现三副本强一致性，通过两阶段提交协议保障跨节点事务的原子性与持久性‌。

• 丰富检索能力：不仅支持传统的结构化索引、全文索引等，为解决 LLM 应用的向量需求，BaikalDB 通过内置向量索引方式实现向量数据的存储和检索，一套系统支持结构化检索、全文检索、向量检索等丰富的检索能力，综合满足 LLM 应用的各种记忆存储和检索需求。

• 高可用，弹性扩展：支持自动扩缩容和数据均衡，支持自动故障恢复和迁移，无单点。当前管理数千业务表与数十万亿行数据，日均处理百亿级请求‌。

△BaikalDB 架构

随着业务发展，离线分析难以满足诉求，实时多维分析需求对 BaikalDB 大数据处理能力的要求显著提高。BaikalDB 的查询引擎主要面向 OLTP（联机事务处理）场景设计的，以下双重关键瓶颈使其应对 OLAP （联机分析处理）有很大的挑战：

1. 计算性能瓶颈：传统火山模型使用行存结构破坏缓存局部性、逐行虚函数调用风暴频繁中断指令流水线、单调用链阻塞多核并行扩展等等弊端，导致大数据分析性能呈超线性劣化。

2. 计算资源瓶颈：Baikaldb 单节点计算资源有限，面对大规模数据计算，单节点 CPU、内存使用容易超限。

BaikalDB 从 OLTP 向 HTAP（混合事务/分析处理）架构演进亟需解决当前 OLTP 查询架构在面向大规模数据的计算性能瓶颈、计算资源瓶颈，并通过如向量化查询引擎、MPP 多机并行查询、列式存储等技术手段优化 OLAP 场景查询性能。

02 BaikalDB OLAP 查询引擎的目标

2.1 向量化查询引擎：解决 OLTP 查询引擎性能瓶颈

2.1.1 火山模型性能瓶颈

设计之初，由于 BaikalDB 主要面向 OLTP 场景，故而 BaikalDB 查询引擎是基于传统的火山模型而实现。

如下图所示，在火山模型里，SQL 的每个算子会抽象为一个 Operator Node，整个 SQL 执行计划构建一个 Operator Node 树，执行方式就是从根节点到叶子节点自上而下不断递归调用 next()函数获取一批数据进行计算处理。 由于火山模型简单易用，每个算子独立实现，不用关心其他算子逻辑等优点，使得火山模型非常适合 OLTP 场景。

△select id, name, age, (age - 30) * 50 as bonus from peope where age > 30 火山模型执行计划

但当火山模型处理大量数据时有着以下三大弊端，这些弊端是导致火山模型面对大数据量分析场景查询性能差的元凶。

1. 行式存储引发的缓存失效问题‌

2. 数据局部性缺失‌：行式存储（Row-based Storage）将整行数据连续存放，当查询仅需部分列时，系统被迫加载整行冗余数据，容易造成 Cache Miss。

3. 硬件资源浪费‌：现代 CPU 三级缓存容量有限，行存结构导致有效数据密度降低。

4. 逐行处理机制的性能衰减‌

5. ‌函数调用过载‌：火山模型要求每个算子逐行调用 next() 接口，处理百万级数据时产生百万次函数调用。

6. ‌CPU 流水线中断‌：频繁的上下文切换导致 CPU 分支预测失败率升高。

7. 执行模型的多核适配缺陷

8. 流水线阻塞‌：Pull-based 模型依赖自顶向下的单调用链，无法并行执行相邻算子。

9. ‌资源闲置浪费‌：现代服务器普遍具备 64 核以上计算能力，单调用链无法充分利用多核能力。

当查询模式从 OLTP 轻量操作转向 OLAP 海量扫描复杂计算时，上述三个弊端导致的查询性能衰减呈现级联放大效应。虽然能做一些如 batch 计算，算子内多线程计算等优化，但并不能解决根本问题，获得的收益也有限。BaikalDB 寻求从根本上解决瓶颈的方法，探寻新架构方案以突破性能瓶颈！

2.1.2 向量化查询引擎

多核时代下的现代数据库执行引擎，发展出了向量化查询引擎，解决上述火山模型的种种弊端，以支持 OLAP 大数据量查询性能。与火山模型弊端一一对应，向量化执行引擎特点是如下：

1. 列式存储与硬件加速协同优化‌

2. 列存数据紧凑布局‌：采用列式存储结构（Colum-based Storage），将同类型数据连续存储于内存，有效提升 CPU 缓存行利用率，减少 Memory Stall。

3. ‌SIMD 指令集加速‌：通过向量寄存器批量处理 128/256/512 位宽度的连续数据单元，允许 CPU 在单个指令周期内对多组数据进行相同的操作。

4. 批量处理提升缓存亲和性

5. ‌数据访问模式优化‌：算子/计算函数内部采用批量处理机制，每次处理一批连续数据块。这种连续内存访问模式提升了 CPU DCache 和 ICache 的友好性，减少 Cache Miss。

6. 流水线气泡消除‌：批处理大量减少上下文切换，降低 CPU 资源空闲周期占比，显著提升流水线吞吐量‌。

7. 多核并行计算架构创新

8. Morsel-Driven Parallelism 范式‌：将 Scan 输入的数据划分为多个数据块（称为 morsel），morsel 作为计算调度的最小单位，能均匀分发给多个 CPU core 并行处理。

9. Push-based 流水线重构‌：颠覆传统 Pull 模型，通过工作线程主动推送数据块至下游算子（算子树中的父节点），消除线程间等待开销。数据驱动执行，将算子树切分成多个线性的 Pipeline，多 Pipeline 之间能并行计算。

△向量化执行引擎 pipeline 并行计算

2.2 MPP 多机并行计算的启发：进一步提升 OLAP 查询性能

向量化执行引擎在 OLAP 场景有大幅度的性能提升，却仍然无法解决单台计算节点的 CPU 和内存受限的问题，同时随着处理数据量增大，单节点执行时间呈指数级增长。

为了打破这一单点限制，进一步提升 OLAP 查询性能，MPP（Massively Parallel Processing）大规模并行计算应运而生，其核心特点有：

1. ****分布式计算架构：****基于哈希、范围等策略将数据分布到不同计算节点，实现并行计算加速，能显著缩短 SQL 响应时间。

2. ****线性扩展：****各计算节点采用无共享架构，通过高速网络实现跨节点数据交换，天然具备横向扩展能力，可通过增加节点线性提升整体吞吐量‌。

△MPP: 3 台计算节点并行计算

03 BaikalDB HTAP 查询架构落地

3.1 巧妙结合 Arrow Acero 实现向量化查询引擎

BaikalDB 团队在向量化执行引擎设计过程中，秉持"避免重复造轮子"的技术理念，优先探索开源生态的优秀解决方案‌。基于 BaikalDB 已采用 Apache Arrow 列式存储格式实现全文索引的技术积淀‌，团队发现 Arrow 项目最新推出的 Acero 流式执行引擎子项目展现出三大核心功能：①支持 Arrow 列式存储格式向量化计算，支持 SIMD 加速；②Push-Based 流式执行框架支持 Pipeline 并行计算，能充分利用多核能力；③执行框架可扩展——这些特性与 BaikalDB 对向量化执行引擎的需求高度契合‌。最终团队选择基于 Arrow 列存格式和 Acero 流式计算引擎实现 BaikalDB 向量化执行引擎，不仅大幅度缩短了研发周期，更充分发挥了开源技术生态的协同优势‌。

BaikalDB 巧妙结合开源 Apache Arrow 列存格式和 Arrow Acero 向量化执行引擎，通过将火山模型计划翻译为 Acero 执行计划，全程使用 Arrow 列存格式，计算最终列存结果再列转行返回给用户，在 BaikalDB 内部实现了适合大量数据计算的向量化查询架构。

核心实现包括以下五点：

1. 数据格式：将源头 RocksDB 扫描出来的行存数据直接转为 Arrow 列存格式，SQL 计算全程使用 Arrow 列存，最终将列存格式的计算结果再转换成行存输出给用户。

2. 计算函数：将所有 BaikalDB 计算函数翻译成 Arrow Compute Expression，每个 Arrow 计算函数运行一次处理万行数据，同时能支持部分如 AVX2、SSE42 等 SIMD 指令集，支持 SIMD 加速。

3. 执行计划：将每个 SQL 生成的 BaikalDB 原生执行计划，转换成一个 Acero 执行计划 Exec Plan（包含所有 BaikalDB 算子翻译成的 Acero ExecNode Declaration 树），使用 Acero 向量化执行引擎替换火山模型执行。

4. 调度器：将进行数据扫描 SourceNode 的 IO Executor 和计算的 CPU Executor 独立开。

5. 每个 Baikaldb 和 BaikalStore 实例内置一个全局的 Pthread CPU IO Executor 池，支持计算阶段的 Push-based Pipeline 并行计算，同时支持算子内并行计算，如 Agg 并发聚合、Join 并发 Build/Probe 等。

6. 每个 SourceNode 都独自一个 Bthread IO Executor，支持 Hash Join 或 Union 的各个子节点能并发查表。

7. RPC 接口：Baikaldb 和 BaikalStore 之间 RPC 使用 Arrow 列存格式替换行存 pb 格式进行数据传输，大幅度降低传输数据大小，同时去掉了 pb 序列化和反序列化的开销，加速了数据传输效率。

△BaikalDB 火山模型（左） BaikalDB 向量化引擎（中：Index Lookup HashJoin，右：HashJoin）

3.2 拆解执行计划实现 MPP 多机并行查询引擎

BaikalDB 实现了向量化执行引擎后，大部分 OLAP 请求性能得到大幅度提升，但很快就遇到了受限于单计算节点资源和性能的 case。团队首先对业务场景进行了分析，发现资源/性能卡点主要在集中在最后单点 Agg/Join 的计算上。

为了破除这一单点限制，因 Agg/Join 都依赖内部 HashTable 进行计算，很容易想到按照 HashKey 将源数据 hash shuffle 到不同的计算节点进行并行计算来加速计算性能，平摊计算需要的内存和 CPU 资源，每个计算节点依然可以使用向量化执行引擎加速单机计算性能。如下图所示，这一方案需要解决的核心问题有以下两点：

1. 实现 Exchange 算子对进行跨节点数据 Shuffle。

2. 在执行计划合适的地方的插入 Exchange 算子对拆分为多个并行子计划 Fragment。

△BaikalDB Agg Mpp 执行示例

△BaikalDB Hash Join Mpp 执行示例

3.2.1  Exchange 算子跨节点数据 shuffle

Exchange 算子包括 ExchangeSenderNode/ExchangeReceiverNode 算子对，必须成对出现，进行跨节点数据分发和接收。ExchangeReceiverNode 核心功能主要是接收对应的下游 Fragment 所有 ExchangeSenderNode 发来的数据，需保证不重不丢，进行超时处理等。ExchangeSenderNode 核心功能是进行数据分发，其分发模式主要有以下三种，以支持不同的场景需求。

1. SinglePartition：一个或者多个 ExchangeSenderNode 将所有数据直接完整发送到单个上游 Fragment ExchangeReciverNode 实例，使用场景如 Fragment 0 单个计算节点收集最终结果输出给用户。

2. HashPartition：ExchangeSenderNode 将所有数据都先根据指定 HashKey 计算每一行 hash 值，根据 hash 值将每行数据 re-partition 到不同的发送队列里，最终发送到上游 Fragment 多个 ExchangeReceiverNode，使用场景如 Store Fragment 将 rocksdb 扫描出来的数据 shuffle 到上游多个实例进行 Agg/HashJoin 计算。

3. BroadcastPartition：ExchangeSenderNode 将所有数据都直接完整发送到多个上游 ExchangeReciverNode，使用场景如 Join 小表数据直接完整发送到上游多个实例进行 BroadcastJoin。

△BaikalDB Exchange 算子实现（HashPartition）

3.2.2  执行计划拆解多个并行子计划 Fragment

在插入 Exchange 算子对之前，需要给物理执行计划里的每个算子明确其分区属性 PartitionProperty，标志着该算子是否可以分发到多个计算节点进行并行加速。PartitionProperty 主要有三种：

1. AnyParitition：该节点没有任何要求，如 FilterNode，可以根据相邻节点分区属性情况单节点或多节点执行。

2. SinglePartition：该节点必须要在单个计算节点上执行，如用于将最终数据打包发送给用户的 PacketNode。

3. HashPartition：该节点可以按照指定 key 将数据分发到多个计算节点上并行执行，当前只有 AggNode 和 JoinNode 会产生 HashPartition。

如下图所示，明确了每个算子的分区属性后，需要加入 Exchange 算子对的位置就很清晰明了了：分区属性有冲突的相邻算子之间。

在物理计划中插入了 Exchange 算子对后，在一对 ExchangeSenderNode/ExchangeReceiverNode 之间进行拆解，即可以将单个物理执行计划拆分为多个子执行计划 Fragment。单机执行的 Fragment 在本机执行，多机执行的 Fragment 发送给多个同集群其他计算节点进行异步执行，store fragment 根据表数据分布情况分发给 BaikalStore 存储集群并行执行。

在 MPP 物理计划制定过程中，为了减少数据 shuffle 的开销，尽可能减少 SQL shuffle 数据量，BaikalDB 也实现了多种查询计划优化手段，如 limit 下推、hash partition fragment 合并等等。

△MPP 物理计划制定过程

3.3  自适应策略支持一套系统应对 TP/AP 请求

BaikalDB 设计 HTAP 架构的核心目标是一套系统能同时兼容 OLTP 和 OLAP 请求，无需业务进行任何改造，这需要系统拥有极强的兼容性。

目前 BaikalDB 内部有三种执行方式：火山模型、单机向量化执行引擎、MPP 多机执行引擎，不同的执行引擎适用不同数据量级的 SQL。随着数据量大小从小到大，适合的计算引擎是火山模型执行 → 单机向量化执行 → MPP 多机并行执行，原因有以下两点：

1. 火山模型执行对比向量化执行更为轻量，导致在 OLTP 小请求（如 SQL 涉及数据行数<1024）方面，传统火山模型性能比单机向量化执行性能更好。

2. MPP 执行会带来额外的网络开销（数据 shuffle）、CPU 开销（hash 计算和 re-partition），导致 Baikaldb 处理数据量需要超过一定阈值时，Baikaldb 多机并行执行才能 cover 额外的网络/CPU 开销，SQL 性能才能提升，否则单机向量化执行是性能最优。

BaikalDB 实现了智能执行引擎决策系统，支持 SQL 自适应选择合适的执行引擎，支持一套集群能同时满足业务 OLTP 和 OLAP 场景需求。技术实现包含两大核心机制：

1. 向量化引擎动态热切换：BaikalDB 支持 SQL 在执行过程中，随着数据量增大，从火山模型动态切换到向量化执行引擎执行。

2. 统计信息驱动 MPP 加速：BaikalDB 支持根据过去执行统计信息决策是否走 MPP 加速执行，当 SQL 统计的 99 分位处理数据行数/大小超过指定的阈值，则 SQL 直接通过 MPP 多机并行执行。

04  总结

4.1 项目收益

BaikalDB 通过架构创新打造了 HTAP 架构，期望一套系统支持线上 OLTP/OLAP 请求，其技术演进路径呈现『混合执行引擎架构自适应选择』的特征：SQL 自适应选择火山模型、向量化执行引擎、MPP 多机并行执行引擎。

目前 BaikalDB HTAP 架构已经应用到线上多个业务，大数据量查询取得大幅度的性能提升，同时 Baikaldb 单点内存使用峰值也得以大幅度下降：

• 单机向量化执行相对于火山模型执行，大数据量查询请求耗时平均下降 62%，最大能下降 97%（如 11s 优化到 300ms），内存使用峰值最大能降低 56%（57G->25G）。

• MPP 在单机向量化执行基础上，大数据量查询耗时还能进一步优化，查询耗时平均下降 54%，最大能下降 71%（如 42s 优化到 12s），内存使用峰值最大能降低 80%（5 hash partition）。

4.2 未来展望

BaikalDB HTAP 架构目前还在不断发展，包括性能优化、丰富支持算子和计算函数等等，未来还预期结合列式存储、CBO 等技术进一步提升 OLAP 场景性能：

• 结合列式存储提升数据扫描性能：目前 BaikalDB 向量化查询引擎和 MPP 查询引擎全程基于 Arrow 列存格式，但是底层数据存储仍然是行存，存在一次行转列的开销；并且 OLAP 场景，更适合使用列存格式作为底层数据存储格式。BaikalDB 当前在发展列存引擎，未来单机向量化和 MPP 能直接基于底层列式存储，进一步提升 OLAP 场景查询性能。

• 结合 CBO（Cost-Based Optimization）优化自适应 MPP 选择策略：当前 BaikalDB 是基于过去的执行统计信息判断 SQL 是否适合走 MPP，当 SQL 过去执行统计信息波动巨大时，自适应判断方法可能会失效，未来可能结合代价模型来进一步优化 MPP 选择策略。