并行执行简介

并行执行是一项优化 SQL 查询任务的策略，通过将一个查询任务分解成多个子任务，允许它们在多个处理器上同时运行，从而提升整个查询任务的执行效率。

在当今的计算机系统中，广泛采用多核处理器、多线程以及高速网络连接，使得并行执行成为一种高效的查询技术。这项技术极大地降低了计算密集型大查询的响应时间，并在离线数据仓库、实时报表、在线大数据分析等业务场景得到广泛应用，同时也在批量导数、快速构建索引表等领域展现了其强大的应用能力。

并行执行显著提升了以下 SQL 查询场景的性能：

• 处理大表扫描、大表连接、大数据量排序和聚合。
• 执行大表的 DDL 操作，例如修改主键、改变列类型和创建索引等。
• 从已有的大数据中建表（使用 Create Table As Select）。
• 执行批量插入、删除和更新数据操作。

适用场景

并行执行不仅适用于离线数据仓库、实时报表、在线大数据分析等分析型系统，而且在 OLTP 领域也具有一定作用，可用于加速 DDL 操作以及数据跑批工作等。

并行执行通过充分利用多个 CPU 和 IO 资源，以实现降低 SQL 执行时间的目标。在以下条件得到满足时，使用并行执行将优于串行执行：

• 数据访问量大
• SQL 查询并发度低
• 对延迟要求较低
• 具备充足的硬件资源

并行执行协同利用多个处理器并发处理同一任务，在具备以下系统环境时可提高性能：

• 多处理器系统（SMP）和集群
• IO 带宽充足
• 具备富余内存（可用于处理内存密集型操作，例如排序、建 Hash 表等）
• 系统负载适中，或者具有峰谷特征（例如系统负载通常保持在 30% 以下）

如果系统不符合上述特征，那么并行执行可能无法显著改善性能。在一些高负载、内存较小或者 IO 能力较弱的系统中，甚至可能导致并行执行效果不佳。

并行执行对硬件没有特殊要求，但需要注意的是，CPU 核数、内存大小、存储 I/O 性能、网络带宽都会对并行执行性能产生影响，其中任何一项成为瓶颈都可能影响整体性能。

工作原理

并行执行通过将一个 SQL 查询任务分解成多个子任务，并将这些子任务调度到多个处理器上运行，实现高效的并发执行。

一旦一个 SQL 查询被解析为并行执行计划，其执行过程按照以下步骤进行：

1. SQL 主线程（负责接收和解析 SQL 的线程）根据计划的形态，提前分配并行执行所需的线程资源。这些线程资源可能涉及多台机器上的集群。
2. SQL 主线程启用 并行调度算子（PX Coordinator）。
3. 并行调度算子解析计划，将其拆分为多个操作步骤，并按照自底向上的顺序调度这些操作。每个操作都被设计为能够最大程度地并行执行。
4. 所有操作完成并行执行后，并行调度算子接收计算结果，并传递给上层算子（例如 Aggregate 算子），以串行完成剩余不可并行的计算（例如最终的 SUM 计算）。

并行的粒度

并行数据扫描的基本工作单元被称为 Granule。

OceanBase 数据库将表扫描任务细分为多个称为 Granule，每个 Granule 定义了扫描任务的范围。由于每个 Granule 仅涵盖单个分区的数据，因此每个分区的数据对应一个独立的扫描任务。简而言之，每个 Granule 对应一个分区内的扫描任务。

基于 Granule 的粒度特征，可以分为以下两类：

• Partition Granule

  Partition Granule 描述的范围是一个完整的分区，扫描任务会涉及到多少个分区，就会划分出多少个 Partition Granule。这里的分区既可以是主表分区，也可以是索引分区。Partition Granule 最常用于 Partition Wise Join，确保两个表的对应分区由 Partition Granule 处理。

• Block Granule

  Block Granule 描述的范围是一个分区中的一段连续数据。在数据扫描场景中，通常使用 Block Granule 划分数据。每个分区被划分为若干个 Block，这些 Block 以一定的规则连接形成一个任务队列，供并行工作线程消耗。

  下图展示了 Block Granule 的工作原理。

  

在给定并行度的情况下，为了保持扫描任务的均衡，优化器会自动选择将数据划分为 Partition Granule 或 Block Granule。如果选择了 Block Granule，那么并行执行框架会在运行时动态决定 Block 的划分，总体原则是确保一个 Block 既不过大，也不过小。Block 过大可能导致数据倾斜，使一些线程工作不充分；Block 过小则会导致频繁的扫描切换开销。

一旦分区粒度（即，Block Granule）划分完毕，每个粒度对应一个扫描任务。Table Scan 扫描算子会依次处理这些扫描任务，逐个完成，直至所有任务执行完毕。

并行执行模型

生产者-消费者流水线模型

并行执行采用生产者-消费者模型进行流水线执行。

在并行调度算子解析计划后，计划会被切分成多个操作步骤，每个操作步骤被称为一个 DFO（Data Flow Operation）。

通常情况下，并行调度算子会同时启动两个 DFO，这两个 DFO 之间以生产者-消费者的方式连接，形成 DFO 间的并行执行。每个 DFO 使用一组线程来执行，这被称为 DFO 内的并行执行，而 DFO 使用的线程数被称为 DOP（Degree Of Parallisim）。

上一阶段的消费者 DFO 在下一阶段会变成生产者 DFO。在并行调度算子的协调下，消费者 DFO 和生产者 DFO 会同时启动。下图展示了生产者-消费者模型中 DFO 的流程。

• DFO A 生成的数据会即时传输给 DFO B 进行计算。
• DFO B 计算完成后，将数据存储在当前线程中，等待上层的 DFO C 启动。
• 当 DFO B 收到 DFO C 启动完成通知后，它会将自身的角色转变为生产者，开始向 DFO C 传输数据，DFO C 收到数据后开始计算。

在下面的查询示例中，SELECT 语句的执行计划首先对 game 表进行全表扫描，然后按照 team 表进行分组求和，最终计算出每个 team 表的总分数。

CREATE TABLE game (round INT PRIMARY KEY, team VARCHAR(10), score INT)
     PARTITION BY HASH(round) PARTITIONS 3;
INSERT INTO game VALUES (1, "CN", 4), (2, "CN", 5), (3, "JP", 3);
INSERT INTO game VALUES (4, "CN", 4), (5, "US", 4), (6, "JP", 4);
SELECT /*+ PARALLEL(3) */ team, SUM(score) TOTAL FROM game GROUP BY team;

obclient> EXPLAIN SELECT /*+ PARALLEL(3) */ team, SUM(score) TOTAL FROM game GROUP BY team;
obclient> EXPLAIN SELECT /*+ PARALLEL(3) */ team, SUM(score) TOTAL FROM game GROUP BY team;
+---------------------------------------------------------------------------------------------------------+
| Query Plan                                                                                              |
+---------------------------------------------------------------------------------------------------------+
| =================================================================                                       |
| |ID|OPERATOR                     |NAME    |EST.ROWS|EST.TIME(us)|                                       |
| -----------------------------------------------------------------                                       |
| |0 |PX COORDINATOR               |        |1       |4           |                                       |
| |1 | EXCHANGE OUT DISTR          |:EX10001|1       |4           |                                       |
| |2 |  HASH GROUP BY              |        |1       |4           |                                       |
| |3 |   EXCHANGE IN DISTR         |        |3       |3           |                                       |
| |4 |    EXCHANGE OUT DISTR (HASH)|:EX10000|3       |3           |                                       |
| |5 |     HASH GROUP BY           |        |3       |2           |                                       |
| |6 |      PX BLOCK ITERATOR      |        |1       |2           |                                       |
| |7 |       TABLE SCAN            |game    |1       |2           |                                       |
| =================================================================                                       |
| Outputs & filters:                                                                                      |
| -------------------------------------                                                                   |
|   0 - output([INTERNAL_FUNCTION(game.team, T_FUN_SUM(T_FUN_SUM(game.score)))]), filter(nil), rowset=256 |
|   1 - output([INTERNAL_FUNCTION(game.team, T_FUN_SUM(T_FUN_SUM(game.score)))]), filter(nil), rowset=256 |
|       dop=3                                                                                             |
|   2 - output([game.team], [T_FUN_SUM(T_FUN_SUM(game.score))]), filter(nil), rowset=256                  |
|       group([game.team]), agg_func([T_FUN_SUM(T_FUN_SUM(game.score))])                                  |
|   3 - output([game.team], [T_FUN_SUM(game.score)]), filter(nil), rowset=256                             |
|   4 - output([game.team], [T_FUN_SUM(game.score)]), filter(nil), rowset=256                             |
|       (#keys=1, [game.team]), dop=3                                                                     |
|   5 - output([game.team], [T_FUN_SUM(game.score)]), filter(nil), rowset=256                             |
|       group([game.team]), agg_func([T_FUN_SUM(game.score)])                                             |
|   6 - output([game.team], [game.score]), filter(nil), rowset=256                                        |
|   7 - output([game.team], [game.score]), filter(nil), rowset=256                                        |
|       access([game.team], [game.score]), partitions(p[0-2])                                             |
|       is_index_back=false, is_global_index=false,                                                       |
|       range_key([game.round]), range(MIN ; MAX)always true                                              |
+---------------------------------------------------------------------------------------------------------+
29 rows in set

上述查询示例的执行示意图如下：

从图中可以得知，该查询实际使用了六个线程，执行流程和任务分配如下：

• 第一步：前三个线程负责 game 表扫描，并在每个线程内对 game.team 数据进行预聚合。
• 第二步：后三个线程负责对预聚合的数据进行最终聚合。
• 第三步：最终聚合结果交由并行调度器，由它返回给客户端。

第一步的数据发给第二步时，需要用 game.team 字段做 Hash，决定将预聚合数据发给哪个线程。

生产者和消费者之间的数据分发方式

数据分发方式指的是，在一组并行执行工作线程（生产者）将数据发送给另一组工作线程（消费者）时使用的方法。优化器使用一系列优化策略来选择最优的数据重分布方式以达到最佳性能。

常见的并行执行中的数据分发方式包括：

• Hash Distribution

  使用 Hash Distribution 时，生产者根据 Distribution Key 对数据行进行哈希计算并取模，以确定将数据发送给哪个消费者工作线程。通常情况下，Hash Distribution 能够均匀分发数据给多个消费者线程。

• Pkey Distribution

  使用 Pkey Distribution 时，生产者计算数据行对应的目标表所在分区，然后将行数据发送给处理该分区的消费者线程。Pkey Distribution 常用于 Partitial Partitions Wise Join 场景，其中消费端的数据不需要进行重分布，可以直接与生产端的数据进行 Partition Wise Join，从而减少网络通信量，提高性能。

• Pkey Hash Distribution

  使用 Pkey Hash Distribution 时，生产者首先计算数据行对应的目标表所在的分区，然后根据 Distribution Key 对数据行进行哈希计算，以确定将其发送给哪个消费者线程来处理。

  Pkey Hash Distribution 常用于 Parallel DML 场景，其中一个分区可以同时被多个线程并发更新，因此需要使用 Pkey Hash Distribution 来确保相同值的数据行由同一个线程处理，不同值的数据行尽可能均匀分布到多个线程中。

• Broadcast Distribution

  使用 Broadcast Distribution 时，生产者将每个数据行发送给消费者端的每个线程，使得每个消费者线程都拥有生产者端的全量数据。Broadcast Distribution 常用于将小表数据复制到执行 Join 的所有节点上，然后在本地执行 Join 操作，以减少网络通信量。

• Broadcast to Host Distribution（BC2HOST）

  使用 Broadcast to Host Distribution 时，生产者将每个数据行发送给消费者端的每个节点，从而使每个消费者节点都拥有生产者端的全量数据。然后，节点内的消费者线程协同处理这份数据。

  Broadcast to Host Distribution 常用于 NESTED LOOP JOIN、SHARED HASH JOIN 场景。在 NESTED LOOP JOIN 场景中，每个消费者线程从共享数据中获取一部分数据作为驱动数据，用于在目标表上执行 Join 操作；在 SHARED HASH JOIN 场景中，每个消费者线程会基于共享数据共同构建 Hash 表，以避免每个线程独立构建相同的 Hash 表所带来的不必要开销。

• Range Distribution

  使用 Range Distribution 时，生产者根据 Range 范围对数据进行划分，使不同的消费者线程处理不同范围的数据。Range Distribution 常用于排序场景，每个消费者线程只需排序其分配的数据，以便在全局范围内保持有序。

• Random Distribution

  使用 Random Distribution 时，生产者会随机打散数据，然后将其发送给消费者线程，以确保每个消费者线程处理的数据量几乎相等，从而实现负载均衡。Random Distribution 常用于多线程并行 UNION ALL 场景，该场景只需数据被打散和负载均衡，而数据之间没有其他关联约束。

• Hybrid Hash Distribution

  Hybrid Hash Distribution 用于自适应的 Join 算法。结合收集的统计信息，OceanBase 数据库提供了一组配置项来定义常规值和高频值。Hybrid Hash Distribution 方法将 Join 两侧的常规值使用 Hash 分布，左侧的高频值使用 Broadcast 分布，右侧的高频值使用 Random 分布。

  

生产者和消费者之间的数据传输机制

在同一时刻，由并行调度算子启动的两个 DFO 之间以生产者-消费者的方式连接并行执行。为了在生产者和消费者之间传输数据，需要创建一个传输网络。

例如，如果生产者 DFO 使用 DOP = 2 进行数据扫描，而消费者 DFO 使用 DOP = 3 进行数据聚合计算，那么每个生产者线程都会创建 3 个虚拟链接连接到消费者线程，总共创建 6 个虚拟链接。如下图所示。

创建的虚拟传输网络称为数据传输层（Data Transfer Layer，简称 DTL）。在 OceanBase 数据库的并行执行框架中，所有的控制消息和行数据都通过 DTL 进行传输。每个工作线程可以建立数千个虚拟链接，具有高度的可扩展性。此外，DTL 还具有数据缓冲、批量数据发送和自动流量控制等功能。

当 DTL 的两端位于同一节点时，DTL 会通过内存拷贝的方式传递消息；而当 DTL 的两端位于不同节点时，DTL 将通过网络通信的方式传递消息。

工作线程

在并行查询中，有两种主要的线程：一个主线程和多个并行工作线程。主线程与普通的 TP 查询使用相同的线程池，而并行工作线程则来自专用线程池。

OceanBase 数据库采用专用线程池模型来分配并行工作线程。每个租户在其所属的各个节点上都有一个专属的并行执行线程池，所有并行查询工作线程都通过该线程池分配。

在并行调度算子在调度每个 DFO 之前，会在线程池中请求线程资源。当 DFO 执行完成后，线程资源会立即被释放。

线程池的初始大小为 0，根据需求动态增长，没有上限。为了避免过多的空闲线程，线程池引入了自动回收机制。对于任何线程：

• 如果空闲时间超过 10 分钟，并且线程池中的剩余线程数大于 8 个，则会被回收销毁。
• 如果空闲时间超过 60 分钟，则会无条件销毁。

尽管线程池的大小没有上限，通过以下两个机制，在大多数情况下形成了实际上的上限：

• 在并行执行开始之前，必须通过 Admission 模块预约线程资源，只有预约成功后才能投入执行。这个机制可以限制并发查询的数量。Admission 模块的详细内容请参考 并发控制与排队。
• 每次从线程池申请线程时，单次申请的线程数量不会超过 N。其中，N 等于租户 Unit 的 MIN_CPU 乘以 px_workers_per_cpu_quota，如果超过，则最多只会分配 N 个线程。集群配置项 px_workers_per_cpu_quota 默认值为 10。例如，一个 DFO 的 DOP = 100，它需要从 A 节点申请 30 个线程，从 B 节点申请 70 个线程。如果 Unit 的 MIN_CPU = 4，px_workers_per_cpu_quota = 10，那么 N = 4 * 10 = 40。最终，这个 DFO 在 A 节点上实际申请到 30 个线程，在 B 节点上实际申请到 40 个线程，其实际 DOP 为 70。

通过均衡负载优化性能

为了实现最佳性能，所有工作线程分配的任务应该尽量相等。

当 SQL 使用 Block Granule 划分任务时，工作任务会被动态地分配到工作线程之间。这可以最小化工作负载不均衡问题，即某些工作线程的工作量不会明显超过其他工作线程。

当 SQL 使用 Partition Granule 划分任务时，通过使任务数等于工作线程数的整数倍，可以优化性能。这对于 Partition Wise Join 和并行 DML 场景很有用。

假设一个表有 16 个分区，每个分区的数据量差不多。此时可以使用 16 个工作线程（DOP 等于 16），以大约十六分之一的时间完成工作；也可以使用五个工作线程，以五分之一的时间完成工作；或者使用两个线程，以一半的时间完成工作。但是，如果使用 15 个线程来处理 16 个分区，那么第一个线程在完成一个分区的工作后，就会开始处理第 16 个分区。而其他线程在完成工作后，它们就会变为空闲状态。当每个分区的数据量相近时，这种配置会导致性能不佳。而当每个分区的数据量有所差异时，实际性能会因情况而异。

类似地，假设使用 6 个线程来处理 16 个分区，每个分区的数据量相近。

在这种情况下，每个线程在完成第一个分区后，会处理第二个分区，但只有四个线程会处理第三个分区，而其他两个线程则保持空闲状态。

一般来说，不能假设在给定数量的分区（N）和给定数量的工作线程（P）上执行并行操作所需的时间等于 N 除以 P。这个公式没有考虑到一些线程可能需要等待其他线程完成最后的分区。然而，通过选择适当的 DOP，可以最小化工作负载不均衡问题并优化性能。

不适用场景

并行执行通常不建议用于以下场景：

• 系统中的典型 SQL 执行时间都在毫秒级别。

  由于并行查询本身有毫秒级的调度开销，对于短查询来说，由并行执行带来的收益可能会被调度开销所抵消。

• 系统负载很高。

  并行执行的设计目标是充分利用系统的空余资源，如果系统本身已经没有空余资源，那么并行执行可能无法带来额外的收益，反而可能影响系统整体性能。

串行执行使用单个线程执行数据库操作，在以下情况下，使用串行执行优于并行执行：

• Query 访问的数据量很小。
• 高并发。
• Query 执行时间小于 100 毫秒。

以下情况是局部无法并行执行的：

• 最顶层的 DFO 无需并行，它负责与客户端交互，以及执行最顶层无需并行的部分操作，例如 LIMIT、PX COORDINATOR 等。
• 包含 TABLE UDF 时，含该 UDF 的 DFO 只能串行执行，其余部分依然可以并行。
• 对于 OLTP 系统中的普通 SELECT 和 DML 语句，并行执行可能不适用。