设置并行执行并行度 - OceanBase Database AI V4.6.2

并行度（Degree Of Parallelism，简称 DOP）是指单个 DFO（Data Flow Operation）在执行时所使用的工作线程数。

并行执行的设计目的是为了高效利用多核资源。OceanBase 并行执行框架提供了多种指定并行度的方式，可以手动指定也可以让数据库自动选择。

并行度取值建议

一个简单的原则是：PARALLEL 的设定目的是为了充分利用 CPU 资源，而不是简单追求最大值。

示例

假设租户的 CPU 数为 20：

对于单表的简单操作（如单表扫描过滤、单表增删改），PARALLEL 的理论值可设置为 20。
对于多表 JOIN 的查询、带有全局索引的 PDML，PARALLEL 的理论值可设置为 10。
对于一些更加复杂的计划，形态为右深树（即，执行计划中的操作树右侧更深层次的部分）时，PARALLEL 的理论值可设置为 7 左右。

示例说明

假如当前租户的 CPU 数为 20 的情况下：

对于单表操作，只有一个 DFO，可以将全部 20 个 CPU 分配给该 DFO 使用。

例如，如果该 DFO 大部分时间都在进行 I/O 操作，可以将并行度设置略高一些，如 25，以充分利用 CPU。

微调后，对于单表简单操作（如单表扫描过滤、单表增删改），PARALLEL 实际值为 30。
对于多表 JOIN，同一时间会启动 2 个 DFO，形成数据流水线（一个 DFO 是 producer，另一个是 consumer），因此每个 DFO 可分配 10 个 CPU。

实际上，并不能保证每个 DFO 都能将分配给它的 CPU 使用到极限。因此，可以适当调整并行度，比如设为 15，以更充分地利用 CPU。然而，不应无限制地增加并行度，设置为 50 根本不会带来实质性的收益，反而会增加线程调度开销和框架调度开销。

微调后，对于多表 JOIN 的查询、带全局索引的 PDML，PARALLEL 实际值为 15。
对于右深树计划，让我们以一个简单的 SQL 查询为例，以说明右深树计划的情况：
```
SELECT /*+ parallel(10) */ column1, column2
FROM table1
WHERE condition1;
```
在这个查询中，我们使用了 parallel(10) 提示，希望系统使用并行度为 10 的计划执行。假设执行计划如下：
```
======================================================================
| ID  | OPERATOR                 | NAME    | EST.ROWS  | EST.TIME(us)|
----------------------------------------------------------------------
| 0   | SELECT STATEMENT          |         |         |              |
| 1   |  PX COORDINATOR           |         |         |              |
| 2   |   PX SEND QC (RANDOM)     | :EX10000|         |              |
| 3   |    PX BLOCK ITERATOR      |         | N rows  |              |
| 4   |     TABLE ACCESS (FULL)   | table1  | N rows  |              |
======================================================================
```
在这个执行计划中，我们可以看到右深树计划的部分是在操作树的右侧，例如操作符号为 PX SEND QC 和 PX BLOCK ITERATOR 的部分。在这个阶段，可能会启动多个 DFO，每个 DFO 负责执行计划的一部分。

假设系统决定在这个右深树计划的某个局部同时启动 3 个 DFO，而每个 DFO 分配到 7 个 CPU。这样做的目的是确保每个 DFO 能够充分利用计算资源，提高整体查询性能。

微调后，通过设置并行度为 10 ~ 15，我们为系统提供了更大的并行执行空间，以适应不同部分的计划可能需要的并行度。这样的调整可以在处理大型数据集或者需要高并行度执行的查询时提高整体性能。

手动指定并行度

您可以手动指定表的并行度，以确保扫描该表时始终使用并行执行。语法如下。

手动指定并行度

通过表属性指定并行度

以下语句分别指定了主表和一个索引的扫描并行度。

ALTER TABLE table_name PARALLEL 4;
ALTER TABLE table_name ALTER INDEX index_name PARALLEL 2;

如果一个 SQL 中只涉及一张表，当 SQL 中查询了主表时，除了主表所在的 DFO，其余 DFO 也会使用并行度 4 来执行；当 SQL 查询了索引表时，除了索引表所在的 DFO，其余 DFO 也会使用并行度 2 来执行。

如果一个 SQL 中涉及多张表，则会使用 PARALLEL 最大值作为整个计划的 DOP。

通过 PARALLEL HINT 指定并行度

可以通过全局 PARALLEL HINT 来指定整个 SQL 的并行度，也可以通过表级的 PARALLEL HINT 来指定特定表的并行度。当一个 SQL 里给定了多个表的 PARALLEL HINT，各个表所在 DFO 的 DOP 由各个表的 Parallel 值决定。如果一个 DFO 里包含多张表，则会取它们的 PARALLEL 最大值作为 DFO 的 DOP。

全局 HINT：

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

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

表级 HINT：

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

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

obclient> EXPLAIN SELECT /*+ parallel(t1 3)+ parallel(t2 2)*/ * FROM t1, t2 WHERE t1.c1 = t2.c1;
obclient [t]> EXPLAIN SELECT /*+ parallel(t1 3)+ parallel(t2 2)*/ * FROM t1, t2 WHERE t1.c1 = t2.c1;
+---------------------------------------------------------------------------------------+
| Query Plan                                                                            |
+---------------------------------------------------------------------------------------+
| ========================================================================              |
| |ID|OPERATOR                            |NAME    |EST.ROWS|EST.TIME(us)|              |
| ------------------------------------------------------------------------              |
| |0 |PX COORDINATOR                      |        |1       |6           |              |
| |1 |└─EXCHANGE OUT DISTR                |:EX10001|1       |5           |              |
| |2 |  └─SHARED HASH JOIN                |        |1       |4           |              |
| |3 |    ├─EXCHANGE IN DISTR             |        |1       |2           |              |
| |4 |    │ └─EXCHANGE OUT DISTR (BC2HOST)|:EX10000|1       |2           |              |
| |5 |    │   └─PX BLOCK ITERATOR         |        |1       |2           |              |
| |6 |    │     └─TABLE FULL SCAN         |t1      |1       |2           |              |
| |7 |    └─PX BLOCK ITERATOR             |        |1       |2           |              |
| |8 |      └─TABLE FULL SCAN             |t2      |1       |2           |              |
| ========================================================================              |
| Outputs & filters:                                                                    |
| -------------------------------------                                                 |
|   0 - output([INTERNAL_FUNCTION(t1.c1, t1.c2, t2.c1, t2.c2)]), filter(nil), rowset=16 |
|   1 - output([INTERNAL_FUNCTION(t1.c1, t1.c2, t2.c1, t2.c2)]), filter(nil), rowset=16 |
|       dop=2                                                                           |
|   2 - output([t1.c1], [t2.c1], [t1.c2], [t2.c2]), filter(nil), rowset=16              |
|       equal_conds([t1.c1 = t2.c1]), other_conds(nil)                                  |
|   3 - output([t1.c1], [t1.c2]), filter(nil), rowset=16                                |
|   4 - output([t1.c1], [t1.c2]), filter(nil), rowset=16                                |
|       dop=3                                                                           |
|   5 - output([t1.c1], [t1.c2]), filter(nil), rowset=16                                |
|   6 - output([t1.c1], [t1.c2]), filter(nil), rowset=16                                |
|       access([t1.c1], [t1.c2]), partitions(p0)                                        |
|       is_index_back=false, is_global_index=false,                                     |
|       range_key([t1.__pk_increment]), range(MIN ; MAX)always true                     |
|   7 - output([t2.c1], [t2.c2]), filter(nil), rowset=16                                |
|   8 - output([t2.c1], [t2.c2]), filter(nil), rowset=16                                |
|       access([t2.c1], [t2.c2]), partitions(p0)                                        |
|       is_index_back=false, is_global_index=false,                                     |
|       range_key([t2.__pk_increment]), range(MIN ; MAX)always true                     |
+---------------------------------------------------------------------------------------+
33 rows in set

在 SESSION 上指定并行度

一旦在当前 session 上指定了并行度，该 session 中的所有查询语句都将以指定的并行度执行。请注意，即使是单行查询的 SQL，也会使用该并行度，可能导致性能下降。

set _force_parallel_query_dop = 3;

对于 DML 语句，使用上述命令只会使 DML 语句的查询部分并行执行，写入部分仍为串行。若希望写入部分也并行，需使用以下命令：

set _force_parallel_dml_dop = 3;

当 _force_parallel_query_dop 被修改为大于 1 时，当前 session 上的查询都会以大于 1 的并行度执行。当 _force_parallel_query_dop 被修改为 1 时，当前 session 上的 SQL 则不开启并行执行。

需要注意的是，这里的设置对于整个 session 中的查询都是全局性的，包括 DML 语句。因此，在使用这些设置时，务必谨慎，确保了解其对整个 session 执行的影响。

自动并行度

有关自动并行度的详细介绍，请参阅 Auto DOP。

并行度优先级

TABLE HINT > 全局 HINT > SESSION 并行度 > TABLE DOP

比较全局 HINT 和 TABLE HINT 的优先级。可以看到，有 TABLE HINT 时，全局 HINT 不生效。

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

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

对比 SESSION 和 TABLE HINT 的优先级。可以看到，使用 TABLE 级的 HINT 时，SESSION 设定的 DOP 不生效。

obclient> ALTER SESSION FORCE PARALLEL QUERY PARALLEL 4;
Query OK, 0 rows affected (0.001 sec)

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

obclient> EXPLAIN SELECT * FROM t1;
+-------------------------------------------------------------------------+
| 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|T1      |1       |1           |               |
| =========================================================               |
| Outputs & filters:                                                      |
| -------------------------------------                                   |
|   0 - output([INTERNAL_FUNCTION(T1.C1, T1.C2)]), filter(nil), rowset=16 |
|   1 - output([INTERNAL_FUNCTION(T1.C1, T1.C2)]), filter(nil), rowset=16 |
|       dop=4                                                             |
|   2 - output([T1.C1], [T1.C2]), filter(nil), rowset=16                  |
|   3 - output([T1.C1], [T1.C2]), filter(nil), rowset=16                  |
|       access([T1.C1], [T1.C2]), partitions(p0)                          |
|       is_index_back=false, is_global_index=false,                       |
|       range_key([T1.C1]), range(MIN ; MAX)always true                   |
+-------------------------------------------------------------------------+
18 rows in set

对比 SESSION 和 TABLE PARALLEL 的优先级。可以看到 TABLE PARALLEL 属性的优先级低于 SESSION 指定 DOP 的优先级。

obclient> ALTER TABLE t1 PARALLEL 5;
Query OK, 0 rows affected

obclient> EXPLAIN SELECT * FROM t1;
+-------------------------------------------------------------------------+
| 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|T1      |1       |1           |               |
| =========================================================               |
| Outputs & filters:                                                      |
| -------------------------------------                                   |
|   0 - output([INTERNAL_FUNCTION(T1.C1, T1.C2)]), filter(nil), rowset=16 |
|   1 - output([INTERNAL_FUNCTION(T1.C1, T1.C2)]), filter(nil), rowset=16 |
|       dop=4                                                             |
|   2 - output([T1.C1], [T1.C2]), filter(nil), rowset=16                  |
|   3 - output([T1.C1], [T1.C2]), filter(nil), rowset=16                  |
|       access([T1.C1], [T1.C2]), partitions(p0)                          |
|       is_index_back=false, is_global_index=false,                       |
|       range_key([T1.C1]), range(MIN ; MAX)always true                   |
+-------------------------------------------------------------------------+
18 rows in set

obclient> ALTER SESSION FORCE PARALLEL QUERY PARALLEL 4;
Query OK, 0 rows affected

obclient> EXPLAIN SELECT * FROM t1;
+-------------------------------------------------------------------------+
| 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|T1      |1       |1           |               |
| =========================================================               |
| Outputs & filters:                                                      |
| -------------------------------------                                   |
|   0 - output([INTERNAL_FUNCTION(T1.C1, T1.C2)]), filter(nil), rowset=16 |
|   1 - output([INTERNAL_FUNCTION(T1.C1, T1.C2)]), filter(nil), rowset=16 |
|       dop=4                                                             |
|   2 - output([T1.C1], [T1.C2]), filter(nil), rowset=16                  |
|   3 - output([T1.C1], [T1.C2]), filter(nil), rowset=16                  |
|       access([T1.C1], [T1.C2]), partitions(p0)                          |
|       is_index_back=false, is_global_index=false,                       |
|       range_key([T1.C1]), range(MIN ; MAX)always true                   |
+-------------------------------------------------------------------------+
18 rows in set