并行查询的执行 - OceanBase 数据库 V5.0.1

并行查询（Parallel Query）是并行执行在查询场景下的具体应用。通过并行查询，可以将 SELECT 查询拆分为多个子任务，启动多个 Worker 线程并行处理，从而缩短单个查询的响应时间。关于并行执行与分布式执行、并行查询的概念关系，请参见分布式执行与并行执行概述。

启动并行查询的方式有以下两种：

通过 PARALLEL Hint 指定并行度（DOP）的方式启动并行查询。
针对查询分区数大于 1 的分区表会自动启动并行查询。

并行查询技术介绍

并行查询是指通过对查询计划的改造，提升对每一个查询计划的 CPU 和 IO 处理能力，从而缩短单个查询的响应时间。并行查询技术可以用于分布式执行计划，也可以用于本地查询计划。

当单个查询的访问数据不在同一个节点上时，需要通过数据重分布的方式，将相关的数据分布到相同的节点进行计算。以每一次的数据重分布节点为上下界，OceanBase 数据库的执行计划在垂直方向上被划分为多个 DFO（Data Flow Operation），而每一个 DFO 可以被切分为指定并行度的任务，通过并发执行以提高执行效率。

一般来说，当并行度提高时，查询的响应时间会缩短，更多的 CPU、IO 和内存资源会被用于执行查询命令。对于支持大数据量查询处理的 DSS（Decision Support Systems）系统或者数据仓库型应用来说，查询时间的提升尤为明显。

整体来说，并行查询的总体思路和分布式执行计划有相似之处，即将执行计划分解之后，将执行计划的每个部分由多个执行线程执行，通过一定的调度的方式，实现执行计划的 DFO 之间的并发执行和 DFO 内部的并发执行。

当系统满足以下条件时，并行查询可以有效提升系统处理性能：

充足的 IO 带宽
系统 CPU 负载较低
充足的内存资源

如果系统没有充足的资源进行额外的并行处理，使用并行查询或者提高并行度并不能提高执行性能。相反，在系统过载的情况下，操作系统会被迫进行更多的调度，例如，执行上下文切换或者页面交换，可能会导致性能的下降。

通常在 DSS 系统中，需要访问大量分区和数据仓库环境，这时并行执行能够提升执行响应时间。对于简单的 DML 操作或者分区内查询以及涉及分区数比较小的查询来说，使用并行查询并不能很明显的降低查询响应时间。

启用分区表并行查询

针对分区表的查询，如果查询的目标分区数大于 1，系统会自动启用并行查询，dop 值由系统默认指定为 1。

如下例所示，创建一个分区表 ptable，对 ptable 进行全表数据的扫描操作，通过 EXPLAIN 命令查看生成的执行计划。

obclient> CREATE TABLE PTABLE(c1 INT , c2 INT) PARTITION BY HASH(c1) PARTITIONS 16;
Query OK, 0 rows affected

obclient> EXPLAIN SELECT * FROM ptable;

Query Plan:
=======================================================
|ID|OPERATOR               |NAME    |EST. ROWS|COST   |
-------------------------------------------------------
|0 |EXCHANGE IN DISTR      |        |1600000  |1246946|
|1 | EXCHANGE OUT DISTR    |:EX10000|1600000  |1095490|
|2 |  PX PARTITION ITERATOR|        |1600000  |1095490|
|3 |   TABLE SCAN          |ptable  |1600000  |1095490|
=======================================================

Outputs & filters:
-------------------------------------
  0 - output([ptable.c1], [ptable.c2]), filter(nil)
  1 - output([ptable.c1], [ptable.c2]), filter(nil), dop=1
  2 - output([ptable.c1], [ptable.c2]), filter(nil)
  3 - output([ptable.c1], [ptable.c2]), filter(nil),
      access([ptable.c1], [ptable.c2]), partitions(p[0-15])

通过执行计划可以看出，分区表默认的并行查询的 dop 为 1。如果 OceanBase 集群一共有 3 个 OBServer 节点，表 ptable 的 16 个分区分散在 3 个 OBServer 节点中，那么每一个 OBServer 节点都会启动一个工作线程（Worker Thread）来执行分区数据的扫描工作，一共需要启动 3 个工作线程来执行表的扫描工作。

针对分区表，添加 PARALLEL Hint 启动并行查询，并指定 dop 值，通过 EXPLAIN 命令查看生成的执行计划。

obclient> EXPLAIN SELECT /*+ PARALLEL(8) */ * FROM ptable;

Query Plan:
=======================================================
|ID|OPERATOR               |NAME    |EST. ROWS|COST   |
-------------------------------------------------------
|0 |EXCHANGE IN DISTR      |        |1600000  |1246946|
|1 | EXCHANGE OUT DISTR    |:EX10000|1600000  |1095490|
|2 |  PX PARTITION ITERATOR|        |1600000  |1095490|
|3 |   TABLE SCAN          |ptable  |1600000  |1095490|
=======================================================

Outputs & filters:
-------------------------------------
  0 - output([ptable.c1], [ptable.c2]), filter(nil)
  1 - output([ptable.c1], [ptable.c2]), filter(nil), dop=8
  2 - output([ptable.c1], [ptable.c2]), filter(nil)
  3 - output([ptable.c1], [ptable.c2]), filter(nil),
      access([ptable.c1], [ptable.c2]), partitions(p[0-15])

通过执行计划可以看出，并行查询的 dop 值为 8。如果查询分区所在的 OBServer 节点的个数小于等于 dop 值，那么工作线程（总个数等于 dop 值）会按照一定的策略分配到涉及的 OBServer 节点上；如果查询分区所在的 OBServer 节点的个数大于 dop 值，那么每一个 OBServer 都会至少启动一个工作线程，一共需要启动的工作线程的数目会大于 dop 值。

例如，当 dop=8，如果 16 个分区均匀的分布在 4 台 OBServer 节点上，那么每一个 OBServer 节点上都会启动 2 个工作线程来扫描其对应的分区（一共启动 8 个工作线程）；如果 16 个分区分布在 16 台 OBServer 节点上（每一个节点一个分区），那么每一台 OBServer 节点上都会启动 1 个工作线程来扫描其对应的分区（一共启动 16 个工作线程）。

如果针对分区表的查询，查询分区数目小于等于 1，系统不会启动并行查询。如下例所示，对 ptable 的查询添加一个过滤条件 c1=1。

obclient> EXPLAIN SELECT * FROM ptable WHERE c1 = 1;

Query Plan:
======================================
|ID|OPERATOR  |NAME  |EST. ROWS|COST |
--------------------------------------
|0 |TABLE SCAN|ptable|990      |85222|
======================================

Outputs & filters:
-------------------------------------
  0 - output([ptable.c1], [ptable.c2]), filter([ptable.c1 = 1]),
      access([ptable.c1], [ptable.c2]), partitions(p1)

通过计划可以看出，查询的目标分区个数为 1，系统没有启动并行查询。如果希望针对一个分区的查询也能够进行并行执行，就只能通过添加 PARALLEL Hint 的方式进行分区内并行查询，通过 EXPLAIN 命令查看生成的执行计划。

obclient> EXPLAIN SELECT /*+ PARALLEL(8) */ * FROM ptable WHERE c1 = 1;

Query Plan:
=================================================
|ID|OPERATOR           |NAME    |EST. ROWS|COST |
-------------------------------------------------
|0 |EXCHANGE IN DISTR  |        |990      |85316|
|1 | EXCHANGE OUT DISTR|:EX10000|990      |85222|
|2 |  PX BLOCK ITERATOR|        |990      |85222|
|3 |   TABLE SCAN      |ptable  |990      |85222|
=================================================

Outputs & filters:
-------------------------------------
  0 - output([ptable.c1], [ptable.c2]), filter(nil)
  1 - output([ptable.c1], [ptable.c2]), filter(nil), dop=8
  2 - output([ptable.c1], [ptable.c2]), filter(nil)
  3 - output([ptable.c1], [ptable.c2]), filter([ptable.c1 = 1]),
      access([ptable.c1], [ptable.c2]), partitions(p1)

注意

如果希望在查询分区数等于 1 的情况下，能够采用 Hint 的方式进行分区内并行查询，需要对应的 dop 值大于等于 2。
如果 dop 值为空或者小于 2 将不启动并行查询。

启用非分区表并行查询

非分区表本质上是只有 1 个分区的分区表，因此针对非分区表的查询，只能通过添加 PARALLEL Hint 的方式启动分区内并行查询，否则不会启动并行查询。

如下例所示，创建一个非分区表 stable，对 stable 进行全表数据的扫描操作，通过 EXPLAIN 命令查看生成的执行计划。

obclient> CREATE TABLE stable(c1 INT, c2 INT);
Query OK, 0 rows affected

obclient> EXPLAIN SELECT * FROM stable;

Query Plan:
======================================
|ID|OPERATOR  |NAME  |EST. ROWS|COST |
--------------------------------------
|0 |TABLE SCAN|stable|100000   |68478|
======================================

Outputs & filters:
-------------------------------------
  0 - output([stable.c1], [stable.c2]), filter(nil),
      access([stable.c1], [stable.c2]), partitions(p0)

通过执行计划可以看出，非分区表不使用 Hint 的情况下，不会启动并行查询。

针对非分区表，添加 PARALLEL Hint 启动分区内并行查询，并指定 dop 值（大于等于 2），通过 EXPLAIN 命令查看生成的执行计划。

obclient> EXPLAIN SELECT /*+ PARALLEL(4)*/  * FROM stable;

Query Plan:
=================================================
|ID|OPERATOR           |NAME    |EST. ROWS|COST |
-------------------------------------------------
|0 |EXCHANGE IN DISTR  |        |100000   |77944|
|1 | EXCHANGE OUT DISTR|:EX10000|100000   |68478|
|2 |  PX BLOCK ITERATOR|        |100000   |68478|
|3 |   TABLE SCAN      |stable  |100000   |68478|
=================================================

Outputs & filters:
-------------------------------------
  0 - output([stable.c1], [stable.c2]), filter(nil)
  1 - output([stable.c1], [stable.c2]), filter(nil), dop=4
  2 - output([stable.c1], [stable.c2]), filter(nil)
  3 - output([stable.c1], [stable.c2]), filter(nil),
      access([stable.c1], [stable.c2]), partitions(p0)

启用多表并行查询

在查询中，多表 JOIN 查询最为常见。

如下例所示，首先创建两张分区表 p1table 和 p2table：

obclient> CREATE TABLE p1table(c1 INT ,c2 INT) PARTITION BY HASH(c1) PARTITIONS 2;
Query OK, 0 rows affected

obclient> CREATE TABLE p2table(c1 INT ,c2 INT) PARTITION BY HASH(c1) PARTITIONS 4;
Query OK, 0 rows affected

查询 p1table 与 p2table 的 JOIN 结果，JOIN 条件是 p1table.c1=p2table.c2，执行计划如下：

obclient> EXPLAIN SELECT  * FROM p1table p1 JOIN p2table p2 ON p1.c1=p2.c2;

Query Plan:
====================================================================
|ID|OPERATOR                          |NAME    |EST. ROWS|COST     |
--------------------------------------------------------------------
|0 |EXCHANGE IN DISTR                 |        |784080000|614282633|
|1 | EXCHANGE OUT DISTR               |:EX10001|784080000|465840503|
|2 |  HASH JOIN                       |        |784080000|465840503|
|3 |   EXCHANGE IN DISTR              |        |200000   |155887   |
|4 |    EXCHANGE OUT DISTR (BROADCAST)|:EX10000|200000   |136955   |
|5 |     PX PARTITION ITERATOR        |        |200000   |136955   |
|6 |      TABLE SCAN                  |p1      |200000   |136955   |
|7 |   PX PARTITION ITERATOR          |        |400000   |273873   |
|8 |    TABLE SCAN                    |p2      |400000   |273873   |
====================================================================

Outputs & filters:
-------------------------------------
  0 - output([p1.c1], [p1.c2], [p2.c1], [p2.c2]), filter(nil)
  1 - output([p1.c1], [p1.c2], [p2.c1], [p2.c2]), filter(nil), dop=1
  2 - output([p1.c1], [p1.c2], [p2.c1], [p2.c2]), filter(nil),
      equal_conds([p1.c1 = p2.c2]), other_conds(nil)
  3 - output([p1.c1], [p1.c2]), filter(nil)
  4 - output([p1.c1], [p1.c2]), filter(nil), dop=1
  5 - output([p1.c1], [p1.c2]), filter(nil)
  6 - output([p1.c1], [p1.c2]), filter(nil),
      access([p1.c1], [p1.c2]), partitions(p[0-1])
  7 - output([p2.c1], [p2.c2]), filter(nil)
  8 - output([p2.c1], [p2.c2]), filter(nil),
      access([p2.c1], [p2.c2]), partitions(p[0-3])

默认情况下针对 p1table 与 p2table（两张表需要查询的分区数都大于 1）都会采用并行查询，默认的 dop 值为 1。同样，也可以通过使用 PARALLEL Hint 的方式来改变并行度。

如下例所示，改变 JOIN 的条件为 p1table.c1=p2table.c2 和 p2table.c1=1，这样针对 p2table 仅仅会选择单个分区，执行计划如下所示：

obclient> EXPLAIN SELECT  * FROM p1table p1 JOIN p2table p2 ON p1.c1=p2.c2 AND p2.c1=1;

Query Plan:
=============================================================
|ID|OPERATOR                     |NAME    |EST. ROWS|COST   |
-------------------------------------------------------------
|0 |EXCHANGE IN DISTR            |        |1940598  |1807515|
|1 | EXCHANGE OUT DISTR          |:EX10001|1940598  |1440121|
|2 |  HASH JOIN                  |        |1940598  |1440121|
|3 |   EXCHANGE IN DISTR         |        |990      |85316  |
|4 |    EXCHANGE OUT DISTR (PKEY)|:EX10000|990      |85222  |
|5 |     TABLE SCAN              |p2      |990      |85222  |
|6 |   PX PARTITION ITERATOR     |        |200000   |136955 |
|7 |    TABLE SCAN               |p1      |200000   |136955 |
=============================================================

Outputs & filters:
-------------------------------------
  0 - output([p1.c1], [p1.c2], [p2.c1], [p2.c2]), filter(nil)
  1 - output([p1.c1], [p1.c2], [p2.c1], [p2.c2]), filter(nil), dop=1
  2 - output([p1.c1], [p1.c2], [p2.c1], [p2.c2]), filter(nil),
      equal_conds([p1.c1 = p2.c2]), other_conds(nil)
  3 - output([p2.c1], [p2.c2]), filter(nil)
  4 - (#keys=1, [p2.c2]), output([p2.c1], [p2.c2]), filter(nil), dop=1
  5 - output([p2.c1], [p2.c2]), filter([p2.c1 = 1]),
      access([p2.c1], [p2.c2]), partitions(p1)
  6 - output([p1.c1], [p1.c2]), filter(nil)
  7 - output([p1.c1], [p1.c2]), filter(nil),
      access([p1.c1], [p1.c2]), partitions(p[0-1])

通过计划可以看出，p2table 仅需要扫描一个分区，在默认情况下不进行并行查询；p1table 需要扫描两个分区，默认情况下进行并行查询。同样，也可以通过添加 PARALLEL Hint 的方式改变并行度，使 p2table 针对一个分区的查询变为分区内并行查询。

并行执行相关的系统视图

OceanBase 数据库提供了系统视图 (G)V$OB_SQL_AUDIT 来查看并行执行的运行状态以及一些统计信息。

(G)V$OB_SQL_AUDIT 包含字段较多，其中与并行执行相关的字段为：qc_id、dfo_id、sqc_id 和 worker_id。

详细信息请参考 (G)V$OB_SQL_AUDIT。