启用和关闭并行查询 - OceanBase 数据库 V4.3.5

本文主要通过具体示例介绍 OceanBase 数据库如何启用分区表并行查询、非分区表并行查询、多表并行查询和关闭并行查询，以及并行执行相关的系统视图。

启用分区表并行查询

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

如下例所示，创建一个分区表 ptable，对 ptable 进行全表数据的扫描操作，通过 EXPLAIN 命令查看生成的执行计划。通过执行计划可以看出，分区表默认的并行查询的 dop 值为 1。如果 OceanBase 集群一共有 3 个 OBServer，表 ptable 的 16 个分区分散在 3 个 OBServer 中，那么每一个 OBServer 都会启动一个工作线程（Worker Thread）来执行分区数据的扫描工作，一共需要启动 3 个工作线程来执行表的扫描工作。

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|EST.TIME(us)|                   |
| -------------------------------------------------------------                   |
| |0 |PX COORDINATOR           |        |1       |62          |                   |
| |1 |└─EXCHANGE OUT DISTR     |:EX10000|1       |62          |                   |
| |2 |  └─PX PARTITION ITERATOR|        |1       |61          |                   |
| |3 |    └─TABLE FULL SCAN    |ptable  |1       |61          |                   |
| =============================================================                   |
| Outputs & filters:                                                              |
| -------------------------------------                                           |
|   0 - output([INTERNAL_FUNCTION(ptable.c1, ptable.c2)]), filter(nil), rowset=16 |
|   1 - output([INTERNAL_FUNCTION(ptable.c1, ptable.c2)]), filter(nil), rowset=16 |
|       dop=1                                                                     |
|   2 - output([ptable.c1], [ptable.c2]), filter(nil), rowset=16                  |
|       force partition granule                                                   |
|   3 - output([ptable.c1], [ptable.c2]), filter(nil), rowset=16                  |
|       access([ptable.c1], [ptable.c2]), partitions(p[0-15])                     |
|       is_index_back=false, is_global_index=false,                               |
|       range_key([ptable.__pk_increment]), range(MIN ; MAX)always true           |
+---------------------------------------------------------------------------------+
19 rows in set

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

obclient> EXPLAIN SELECT /*+ PARALLEL(8) */ * FROM ptable;
Query Plan: ==================================================
|ID|OPERATOR           |NAME    |EST. ROWS|COST  |
--------------------------------------------------
|0 |PX COORDINATOR     |        |1600000  |737992|
|1 | EXCHANGE OUT DISTR|:EX10000|1600000  |618888|
|2 |  PX BLOCK ITERATOR|        |1600000  |618888|
|3 |   TABLE SCAN      |ptable  |1600000  |618888|
==================================================

Outputs & filters:
-------------------------------------
  0 - output([INTERNAL_FUNCTION(ptable.c1, ptable.c2)]), filter(nil)
  1 - output([INTERNAL_FUNCTION(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|EST.TIME(us)|                            |
| -------------------------------------------------------                            |
| |0 |EXCHANGE IN REMOTE   |      |1       |5           |                            |
| |1 |└─EXCHANGE OUT REMOTE|      |1       |5           |                            |
| |2 |  └─TABLE FULL SCAN  |ptable|1       |4           |                            |
| =======================================================                            |
| Outputs & filters:                                                                 |
| -------------------------------------                                              |
|   0 - output([ptable.c1], [ptable.c2]), filter(nil)                                |
|   1 - output([ptable.c1], [ptable.c2]), filter(nil)                                |
|   2 - output([ptable.c1], [ptable.c2]), filter([ptable.c1 = 1]), rowset=16         |
|       access([ptable.c1], [ptable.c2]), partitions(p1)                             |
|       is_index_back=false, is_global_index=false, filter_before_indexback[false],  |
|       range_key([ptable.__pk_increment]), range(MIN ; MAX)always true              |
+------------------------------------------------------------------------------------+
15 rows in set

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

obclient> EXPLAIN SELECT /*+ PARALLEL(8) */ * FROM ptable WHERE c1 = 1;
+------------------------------------------------------------------------------------+
| Query Plan                                                                         |
+------------------------------------------------------------------------------------+
| =========================================================                          |
| |ID|OPERATOR             |NAME    |EST.ROWS|EST.TIME(us)|                          |
| ---------------------------------------------------------                          |
| |0 |PX COORDINATOR       |        |1       |1           |                          |
| |1 |└─EXCHANGE OUT DISTR |:EX10000|1       |1           |                          |
| |2 |  └─PX BLOCK ITERATOR|        |1       |1           |                          |
| |3 |    └─TABLE FULL SCAN|ptable  |1       |1           |                          |
| =========================================================                          |
| Outputs & filters:                                                                 |
| -------------------------------------                                              |
|   0 - output([INTERNAL_FUNCTION(ptable.c1, ptable.c2)]), filter(nil), rowset=16    |
|   1 - output([INTERNAL_FUNCTION(ptable.c1, ptable.c2)]), filter(nil), rowset=16    |
|       dop=8                                                                        |
|   2 - output([ptable.c1], [ptable.c2]), filter(nil), rowset=16                     |
|   3 - output([ptable.c1], [ptable.c2]), filter([ptable.c1 = 1]), rowset=16         |
|       access([ptable.c1], [ptable.c2]), partitions(p1)                             |
|       is_index_back=false, is_global_index=false, filter_before_indexback[false],  |
|       range_key([ptable.__pk_increment]), range(MIN ; MAX)always true              |
+------------------------------------------------------------------------------------+
18 rows in set

说明

如果希望在查询分区数等于 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|EST.TIME(us)|               |
| -------------------------------------------------------               |
| |0 |EXCHANGE IN REMOTE   |      |1       |5           |               |
| |1 |└─EXCHANGE OUT REMOTE|      |1       |5           |               |
| |2 |  └─TABLE FULL SCAN  |stable|1       |4           |               |
| =======================================================               |
| Outputs & filters:                                                    |
| -------------------------------------                                 |
|   0 - output([stable.c1], [stable.c2]), filter(nil)                   |
|   1 - output([stable.c1], [stable.c2]), filter(nil)                   |
|   2 - output([stable.c1], [stable.c2]), filter(nil), rowset=16        |
|       access([stable.c1], [stable.c2]), partitions(p0)                |
|       is_index_back=false, is_global_index=false,                     |
|       range_key([stable.__pk_increment]), range(MIN ; MAX)always true |
+-----------------------------------------------------------------------+
15 rows in set

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

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

obclient> EXPLAIN SELECT /*+ PARALLEL(4)*/  * FROM stable;
+---------------------------------------------------------------------------------+
| Query Plan                                                                      |
+---------------------------------------------------------------------------------+
| =========================================================                       |
| |ID|OPERATOR             |NAME    |EST.ROWS|EST.TIME(us)|                       |
| ---------------------------------------------------------                       |
| |0 |PX COORDINATOR       |        |1       |2           |                       |
| |1 |└─EXCHANGE OUT DISTR |:EX10000|1       |2           |                       |
| |2 |  └─PX BLOCK ITERATOR|        |1       |1           |                       |
| |3 |    └─TABLE FULL SCAN|stable  |1       |1           |                       |
| =========================================================                       |
| Outputs & filters:                                                              |
| -------------------------------------                                           |
|   0 - output([INTERNAL_FUNCTION(stable.c1, stable.c2)]), filter(nil), rowset=16 |
|   1 - output([INTERNAL_FUNCTION(stable.c1, stable.c2)]), filter(nil), rowset=16 |
|       dop=4                                                                     |
|   2 - output([stable.c1], [stable.c2]), filter(nil), rowset=16                  |
|   3 - output([stable.c1], [stable.c2]), filter(nil), rowset=16                  |
|       access([stable.c1], [stable.c2]), partitions(p0)                          |
|       is_index_back=false, is_global_index=false,                               |
|       range_key([stable.__pk_increment]), range(MIN ; MAX)always true           |
+---------------------------------------------------------------------------------+
18 rows in set

启用多表并行查询

在查询中，多表 JOIN 查询最为常见。对于多表的场景，如果查询的分区数都大于 1，每张表都会采用并行查询。

如下例所示，首先创建两张分区表 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 p1table.c1, p2table.c1 AS p2_c1 FROM p1table JOIN p2table ON p1table.c1 = p2table.c2;
+-----------------------------------------------------------------------------------+
| Query Plan                                                                        |
+-----------------------------------------------------------------------------------+
| =====================================================================             |
| |ID|OPERATOR                         |NAME    |EST.ROWS|EST.TIME(us)|             |
| ---------------------------------------------------------------------             |
| |0 |PX COORDINATOR                   |        |1       |25          |             |
| |1 |└─EXCHANGE OUT DISTR             |:EX10001|1       |25          |             |
| |2 |  └─HASH JOIN                    |        |1       |25          |             |
| |3 |    ├─PX PARTITION ITERATOR      |        |1       |8           |             |
| |4 |    │ └─TABLE FULL SCAN          |p1table |1       |8           |             |
| |5 |    └─EXCHANGE IN DISTR          |        |1       |17          |             |
| |6 |      └─EXCHANGE OUT DISTR (PKEY)|:EX10000|1       |16          |             |
| |7 |        └─PX PARTITION ITERATOR  |        |1       |16          |             |
| |8 |          └─TABLE FULL SCAN      |p2table |1       |16          |             |
| =====================================================================             |
| Outputs & filters:                                                                |
| -------------------------------------                                             |
|   0 - output([INTERNAL_FUNCTION(p1table.c1, p2table.c1)]), filter(nil), rowset=16 |
|   1 - output([INTERNAL_FUNCTION(p1table.c1, p2table.c1)]), filter(nil), rowset=16 |
|       dop=1                                                                       |
|   2 - output([p1table.c1], [p2table.c1]), filter(nil), rowset=16                  |
|       equal_conds([p1table.c1 = p2table.c2]), other_conds(nil)                    |
|   3 - output([p1table.c1]), filter(nil), rowset=16                                |
|       affinitize, force partition granule                                         |
|   4 - output([p1table.c1]), filter(nil), rowset=16                                |
|       access([p1table.c1]), partitions(p[0-1])                                    |
|       is_index_back=false, is_global_index=false,                                 |
|       range_key([p1table.__pk_increment]), range(MIN ; MAX)always true            |
|   5 - output([p2table.c2], [p2table.c1]), filter(nil), rowset=16                  |
|   6 - output([p2table.c2], [p2table.c1]), filter(nil), rowset=16                  |
|       (#keys=1, [p2table.c2]), dop=1                                              |
|   7 - output([p2table.c1], [p2table.c2]), filter(nil), rowset=16                  |
|       force partition granule                                                     |
|   8 - output([p2table.c1], [p2table.c2]), filter(nil), rowset=16                  |
|       access([p2table.c1], [p2table.c2]), partitions(p[0-3])                      |
|       is_index_back=false, is_global_index=false,                                 |
|       range_key([p2table.__pk_increment]), range(MIN ; MAX)always true            |
+-----------------------------------------------------------------------------------+
35 rows in set

默认情况下针对 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|EST.TIME(us)|                 |
| ---------------------------------------------------------------------                 |
| |0 |PX COORDINATOR                   |        |1       |15          |                 |
| |1 |└─EXCHANGE OUT DISTR             |:EX10001|1       |15          |                 |
| |2 |  └─HASH JOIN                    |        |1       |13          |                 |
| |3 |    ├─PX PARTITION ITERATOR      |        |1       |8           |                 |
| |4 |    │ └─TABLE FULL SCAN          |p1      |1       |8           |                 |
| |5 |    └─EXCHANGE IN DISTR          |        |1       |5           |                 |
| |6 |      └─EXCHANGE OUT DISTR (PKEY)|:EX10000|1       |5           |                 |
| |7 |        └─TABLE FULL SCAN        |p2      |1       |4           |                 |
| =====================================================================                 |
| Outputs & filters:                                                                    |
| -------------------------------------                                                 |
|   0 - output([INTERNAL_FUNCTION(p1.c1, p1.c2, p2.c1, p2.c2)]), filter(nil), rowset=16 |
|   1 - output([INTERNAL_FUNCTION(p1.c1, p1.c2, p2.c1, p2.c2)]), filter(nil), rowset=16 |
|       dop=1                                                                           |
|   2 - output([p1.c1], [p2.c2], [p1.c2], [p2.c1]), filter(nil), rowset=16              |
|       equal_conds([p1.c1 = p2.c2]), other_conds(nil)                                  |
|   3 - output([p1.c1], [p1.c2]), filter(nil), rowset=16                                |
|       affinitize, force partition granule                                             |
|   4 - output([p1.c1], [p1.c2]), filter(nil), rowset=16                                |
|       access([p1.c1], [p1.c2]), partitions(p[0-1])                                    |
|       is_index_back=false, is_global_index=false,                                     |
|       range_key([p1.__pk_increment]), range(MIN ; MAX)always true                     |
|   5 - output([p2.c2], [p2.c1]), filter(nil), rowset=16                                |
|   6 - output([p2.c2], [p2.c1]), filter(nil), rowset=16                                |
|       (#keys=1, [p2.c2]), is_single, dop=1                                            |
|   7 - output([p2.c1], [p2.c2]), filter([p2.c1 = 1]), rowset=16                        |
|       access([p2.c1], [p2.c2]), partitions(p1)                                        |
|       is_index_back=false, is_global_index=false, filter_before_indexback[false],     |
|       range_key([p2.__pk_increment]), range(MIN ; MAX)always true                     |
+---------------------------------------------------------------------------------------+
32 rows in set

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

并行执行相关的系统视图

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

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

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