并行 DML

并行 DML 通过使用并行执行机制来提高对大型数据库的表和索引执行插入、更新、删除等操作，以提高执行效率。对于决策支持系统 (DSS)的数据库，并行 DML 提供了查询和更新功能，是对并行查询功能的补充。对于 OLTP 数据库，并行 DML 操作可以加速批处理作业的运行。

配置并行 DML

OceanBase 数据库支持在 SQL 语句或会话中显式启用和禁用并行 DML。

在 SQL 语句中启用并行 DML

在 SQL 语句中启用并行 DML，请在语句中插入如下 Hint：

/*+ ENABLE_PARALLEL_DML PARALLEL(3) */

一般情况下，ENABLE_PARALLEL_DML Hint 和 PARALLEL Hint 必须配合使用才能开启并行 DML。不过，当目标表的 Schema 上指定了表级别的并行度时，仅需指定 ENABLE_PARALLEL_DML Hint。

示例如下：

同时使用 ENABLE_PARALLEL_DML Hint 和 PARALLEL(n) 参数指定并行度 n，并且 n > 1，并行度 dop=2。

1. 创建表 t1。

   CREATE TABLE t1 (c1 INT PRIMARY KEY, c2 INT) NOPARALLEL;
   

2. 创建表 t2。

   CREATE TABLE t2 (c1 INT PRIMARY KEY, c2 INT) PARALLEL 11 PARTITION BY HASH(c1) PARTITIONS 3;
   

3. 创建表 t3。

   CREATE TABLE t3 (c1 INT PRIMARY KEY, c2 INT) PARALLEL 10 PARTITION BY HASH(c1) PARTITIONS 4;
   

4. 查看执行计划。

   EXPLAIN INSERT /*+ ENABLE_PARALLEL_DML PARALLEL(2) */ INTO t1 SELECT * FROM t3;
   

   返回结果如下：

   +-------------------------------------------------------------------------------------------------------------------------------------------------------+
   | Query Plan                                                                                                                                            |
   +-------------------------------------------------------------------------------------------------------------------------------------------------------+
   | =========================================================================                                                                             |
   | |ID|OPERATOR                      |NAME           |EST.ROWS|EST.TIME(us)|                                                                             |
   | -------------------------------------------------------------------------                                                                             |
   | |0 |OPTIMIZER STATS MERGE         |               |1       |18          |                                                                             |
   | |1 | PX COORDINATOR               |               |1       |18          |                                                                             |
   | |2 |  EXCHANGE OUT DISTR          |:EX10001       |1       |18          |                                                                             |
   | |3 |   INSERT                     |               |1       |17          |                                                                             |
   | |4 |    EXCHANGE IN DISTR         |               |1       |4           |                                                                             |
   | |5 |     EXCHANGE OUT DISTR (HASH)|:EX10000       |1       |4           |                                                                             |
   | |6 |      OPTIMIZER STATS GATHER  |               |1       |4           |                                                                             |
   | |7 |       SUBPLAN SCAN           |ANONYMOUS_VIEW1|1       |4           |                                                                             |
   | |8 |        PX BLOCK ITERATOR     |               |1       |4           |                                                                             |
   | |9 |         TABLE SCAN           |t3             |1       |4           |                                                                             |
   | =========================================================================                                                                             |
   | Outputs & filters:                                                                                                                                    |
   | -------------------------------------                                                                                                                 |
   |   0 - output(nil), filter(nil), rowset=256                                                                                                            |
   |   1 - output([column_conv(INT,PS:(11,0),NOT NULL,ANONYMOUS_VIEW1.c1)], [column_conv(INT,PS:(11,0),NULL,ANONYMOUS_VIEW1.c2)]), filter(nil), rowset=256 |
   |   2 - output([column_conv(INT,PS:(11,0),NOT NULL,ANONYMOUS_VIEW1.c1)], [column_conv(INT,PS:(11,0),NULL,ANONYMOUS_VIEW1.c2)]), filter(nil), rowset=256 |
   |       dop=2                                                                                                                                           |
   |   3 - output([column_conv(INT,PS:(11,0),NOT NULL,ANONYMOUS_VIEW1.c1)], [column_conv(INT,PS:(11,0),NULL,ANONYMOUS_VIEW1.c2)]), filter(nil)             |
   |       columns([{t1: ({t1: (t1.c1, t1.c2)})}]), partitions(p0),                                                                                        |
   |       column_values([column_conv(INT,PS:(11,0),NOT NULL,ANONYMOUS_VIEW1.c1)], [column_conv(INT,PS:(11,0),NULL,ANONYMOUS_VIEW1.c2)])                   |
   |   4 - output([column_conv(INT,PS:(11,0),NOT NULL,ANONYMOUS_VIEW1.c1)], [column_conv(INT,PS:(11,0),NULL,ANONYMOUS_VIEW1.c2)]), filter(nil), rowset=256 |
   |   5 - output([column_conv(INT,PS:(11,0),NOT NULL,ANONYMOUS_VIEW1.c1)], [column_conv(INT,PS:(11,0),NULL,ANONYMOUS_VIEW1.c2)]), filter(nil), rowset=256 |
   |       (#keys=1, [column_conv(INT,PS:(11,0),NOT NULL,ANONYMOUS_VIEW1.c1)]), dop=2                                                                      |
   |   6 - output([column_conv(INT,PS:(11,0),NOT NULL,ANONYMOUS_VIEW1.c1)], [column_conv(INT,PS:(11,0),NULL,ANONYMOUS_VIEW1.c2)]), filter(nil), rowset=256 |
   |   7 - output([ANONYMOUS_VIEW1.c1], [ANONYMOUS_VIEW1.c2]), filter(nil), rowset=256                                                                     |
   |       access([ANONYMOUS_VIEW1.c1], [ANONYMOUS_VIEW1.c2])                                                                                              |
   |   8 - output([t3.c1], [t3.c2]), filter(nil), rowset=256                                                                                               |
   |   9 - output([t3.c1], [t3.c2]), filter(nil), rowset=256                                                                                               |
   |       access([t3.c1], [t3.c2]), partitions(p[0-3])                                                                                                    |
   |       is_index_back=false, is_global_index=false,                                                                                                     |
   |       range_key([t3.c1]), range(MIN ; MAX)always true                                                                                                 |
   +-------------------------------------------------------------------------------------------------------------------------------------------------------+
   34 rows in set
   

在 SQL 语句中禁用并行 DML

如果要禁用并行 DML，请在语句中插入如下 Hint：

/*+ DISABLE_PARALLEL_DML */

即使会话中启用了并行 DML，您也可以在指定 SQL 语句中使用 DISABLE_PARALLEL_DML Hint 禁用并行 DML。

在会话中启用并行 DML

默认情况下，即使 SQL 语句中使用了 PARALLEL Hint，并行 DML 也是不开启的，所以还需要通过在会话上开启并行 DML。

执行以下 SQL 语句在会话中启用并行 DML：

SET _FORCE_PARALLEL_DML_DOP = n;

ALTER SESSION ENABLE PARALLEL DML;

ALTER SESSION FORCE PARALLEL DML PARALLEL n;

在会话中禁用并行 DML

执行以下 SQL 语句在会话中禁用并行 DML：

SET _FORCE_PARALLEL_DML_DOP = 1;

ALTER SESSION DISABLE PARALLEL DML;

示例与执行计划解析

分区级并行更新

将根据以下 SQL 语句进行展示说明表分区并行处理功能。

1. 创建测试表 branch_sp_tbl_src。

   CREATE TABLE branch_sp_tbl_src(id INT PRIMARY KEY, v INT) PARTITION BY KEY(id) PARTITIONS 4;
   

2. 创建测试表 branch_sp_tbl_dest。

   CREATE TABLE branch_sp_tbl_dest LIKE branch_sp_tbl_src;
   

3. 查看执行计划。

   执行下面 SQL 语句，将展示如何执行该插入操作。

   obclient [test]> EXPLAIN BASIC INSERT /*+enable_parallel_dml parallel(100) query_timeout(1000000000)*/ INTO branch_sp_tbl_dest SELECT id, v FROM branch_sp_tbl_src ON DUPLICATE KEY UPDATE v = v + 1;
   

   返回结果如下：

   +-------------------------------------------------------------------------------------------------------------------------------------+
   | Query Plan                                                                                                                          |
   +-------------------------------------------------------------------------------------------------------------------------------------+
   | ================================================                                                                                    |
   | |ID|OPERATOR                 |NAME             |                                                                                    |
   | ------------------------------------------------                                                                                    |
   | |0 |PX COORDINATOR           |                 |                                                                                    |
   | |1 |└─EXCHANGE OUT DISTR     |:EX10000         |                                                                                    |
   | |2 |  └─PX PARTITION ITERATOR|                 |                                                                                    |
   | |3 |    └─INSERT_UP          |                 |                                                                                    |
   | |4 |      └─SUBPLAN SCAN     |ANONYMOUS_VIEW1  |                                                                                    |
   | |5 |        └─TABLE FULL SCAN|branch_sp_tbl_src|                                                                                    |
   | ================================================                                                                                    |
   | Outputs & filters:                                                                                                                  |
   | -------------------------------------                                                                                               |
   |   0 - output(nil), filter(nil), rowset=16                                                                                           |
   |   1 - output(nil), filter(nil), rowset=16                                                                                           |
   |       dop=100                                                                                                                       |
   |   2 - output(nil), filter(nil), rowset=16                                                                                           |
   |       partition wise, force partition granule                                                                                       |
   |   3 - output(nil), filter(nil)                                                                                                      |
   |       columns([{branch_sp_tbl_dest: ({branch_sp_tbl_dest: (branch_sp_tbl_dest.id, branch_sp_tbl_dest.v)})}]), partitions(p[0-3]),   |
   |       column_values([column_conv(INT,PS:(11,0),NOT NULL,ANONYMOUS_VIEW1.id)], [column_conv(INT,PS:(11,0),NULL,ANONYMOUS_VIEW1.v)]), |
   |       update([branch_sp_tbl_dest.v=column_conv(INT,PS:(11,0),NULL,cast(branch_sp_tbl_dest.v + 1, INT(-1, 0)))])                     |
   |   4 - output([ANONYMOUS_VIEW1.id], [ANONYMOUS_VIEW1.v]), filter(nil), rowset=16                                                     |
   |       access([ANONYMOUS_VIEW1.id], [ANONYMOUS_VIEW1.v])                                                                             |
   |   5 - output([branch_sp_tbl_src.id], [branch_sp_tbl_src.v]), filter(nil), rowset=16                                                 |
   |       access([branch_sp_tbl_src.id], [branch_sp_tbl_src.v]), partitions(p[0-3])                                                     |
   |       is_index_back=false, is_global_index=false,                                                                                   |
   |       range_key([branch_sp_tbl_src.id]), range(MIN ; MAX)always true                                                                |
   +-------------------------------------------------------------------------------------------------------------------------------------+
   27 rows in set
   

   查询计划中算子分析如下：

   • 0 号算子：表示这是一个并行执行协调器，它负责管理并行执行的进程。
   • 1 号算子：表示数据将在不同的执行节点之间进行分发。
   • 2 号算子：表示查询会遍历分区，这里 “partition wise” 意味着查询会智能地处理各个分区之间的数据。
   • 3 号算子：表示插入或更新操作。如果插入的键在表中不存在，它将执行插入操作；如果存在，则执行更新操作。
   • 4 和 5 号算子：表示对名为 branch_sp_tbl_src 的表进行全表扫描。这个表是数据源，从中选择数据来进行插入操作。

并行 DML 关联更新

关联更新指在 UPDATE 语句中使用多表连接操作，基于关联表的数据更新目标表。通过并行 DML 功能，可以显著提升大规模数据更新操作的性能。

示例如下：

以下步骤展示了如何使用并行 DML 实现关联更新功能。

1. 创建两个测试表 customers 和 orders。

   CREATE TABLE customers (
       id INT PRIMARY KEY,
       name VARCHAR(50),
       customer_level VARCHAR(10)
   );
   

   CREATE TABLE orders (
       order_id INT PRIMARY KEY,
       customer_id INT,
       discount DECIMAL(3, 2),
       amount DECIMAL(10, 2)
   );
   

2. 插入测试数据。

   • 向 customers 表插入客户数据：

     INSERT INTO customers (id, name, customer_level) VALUES
     (1, 'Alice', 'VIP'),
     (2, 'Bob', 'NORMAL'),
     (3, 'Charlie', 'OTHER');
     

   • 向 orders 表插入订单数据：

     INSERT INTO orders (order_id, customer_id, discount, amount) VALUES
     (101, 1, NULL, 1000.00),
     (102, 2, NULL, 2000.00),
     (103, 3, NULL, 1500.00),
     (104, 1, NULL, 3000.00),
     (105, 2, NULL, 2500.00);
     

3. 并行 DML 关联更新。

   通过 EXPLAIN 查看并行 DML 关联更新的执行计划：通过 PARALLEL Hint 指定并行度为 4，根据 customers 表中的客户等级（customer_level），更新 orders 表中的折扣字段（discount）。

   EXPLAIN UPDATE /*+ PARALLEL(4) */
       orders o JOIN customers c ON o.customer_id = c.id
   SET o.discount = CASE
       WHEN c.customer_level = 'VIP' THEN 0.9
       WHEN c.customer_level = 'NORMAL' THEN 0.95
       ELSE 1.0
       END;
   

   返回结果如下：

   +------------------------------------------------------------------------------------------------------------------------------------------------------------+
   | Query Plan                                                                                                                                                 |
   +------------------------------------------------------------------------------------------------------------------------------------------------------------+
   | ==========================================================================                                                                                 |
   | |ID|OPERATOR                              |NAME    |EST.ROWS|EST.TIME(us)|                                                                                 |
   | --------------------------------------------------------------------------                                                                                 |
   | |0 |DISTRIBUTED UPDATE                    |        |3       |50          |                                                                                 |
   | |1 |└─PX COORDINATOR                      |        |3       |3           |                                                                                 |
   | |2 |  └─EXCHANGE OUT DISTR                |:EX10001|3       |3           |                                                                                 |
   | |3 |    └─SHARED HASH JOIN                |        |3       |3           |                                                                                 |
   | |4 |      ├─EXCHANGE IN DISTR             |        |3       |2           |                                                                                 |
   | |5 |      │ └─EXCHANGE OUT DISTR (BC2HOST)|:EX10000|3       |2           |                                                                                 |
   | |6 |      │   └─PX BLOCK ITERATOR         |        |3       |1           |                                                                                 |
   | |7 |      │     └─TABLE FULL SCAN         |c       |3       |1           |                                                                                 |
   | |8 |      └─PX BLOCK ITERATOR             |        |5       |1           |                                                                                 |
   | |9 |        └─TABLE FULL SCAN             |o       |5       |1           |                                                                                 |
   | ==========================================================================                                                                                 |
   | Outputs & filters:                                                                                                                                         |
   | -------------------------------------                                                                                                                      |
   |   0 - output(nil), filter(nil)                                                                                                                             |
   |       table_columns([{o: ({orders: (o.order_id, o.customer_id, o.discount, o.amount)})}]),                                                                 |
   |       update([o.discount=column_conv(DECIMAL_INT,PS:(3,2),NULL,CASE WHEN c.customer_level = 'VIP' THEN cast(0.9, DECIMAL_INT(3, 2)) WHEN c.customer_level  |
   |       = 'NORMAL' THEN 0.95 ELSE cast(1.0, DECIMAL_INT(3, 2)) END)])                                                                                        |
   |   1 - output([o.order_id], [o.customer_id], [o.discount], [o.amount], [c.customer_level]), filter(nil), rowset=16                                          |
   |   2 - output([o.order_id], [o.customer_id], [o.discount], [o.amount], [c.customer_level]), filter(nil), rowset=16                                          |
   |       dop=4                                                                                                                                                |
   |   3 - output([o.order_id], [o.customer_id], [o.discount], [o.amount], [c.customer_level]), filter(nil), rowset=16                                          |
   |       equal_conds([o.customer_id = c.id]), other_conds(nil)                                                                                                |
   |   4 - output([c.id], [c.customer_level]), filter(nil), rowset=16                                                                                           |
   |   5 - output([c.id], [c.customer_level]), filter(nil), rowset=16                                                                                           |
   |       dop=4                                                                                                                                                |
   |   6 - output([c.id], [c.customer_level]), filter(nil), rowset=16                                                                                           |
   |   7 - output([c.id], [c.customer_level]), filter(nil), rowset=16                                                                                           |
   |       access([c.id], [c.customer_level]), partitions(p0)                                                                                                   |
   |       is_index_back=false, is_global_index=false,                                                                                                          |
   |       range_key([c.id]), range(MIN ; MAX)always true                                                                                                       |
   |   8 - output([o.order_id], [o.customer_id], [o.discount], [o.amount]), filter(nil), rowset=16                                                              |
   |   9 - output([o.order_id], [o.customer_id], [o.discount], [o.amount]), filter(nil), rowset=16                                                              |
   |       access([o.order_id], [o.customer_id], [o.discount], [o.amount]), partitions(p0)                                                                      |
   |       is_index_back=false, is_global_index=false,                                                                                                          |
   |       range_key([o.order_id]), range(MIN ; MAX)always true                                                                                                 |
   +------------------------------------------------------------------------------------------------------------------------------------------------------------+
   38 rows in set
   

使用限制与注意事项

• 禁用并行 DML 时，即使在 SQL 语句中使用 PARALLEL Hint，也不会执行并行 DML。
• 在会话中启用并行 DML 时，则并行执行对此会话中的所有 DML 语句都会生效。如果 SQL 语句通过 ENABLE_PARALLEL_DML Hint 启用并行 DML 时，则并行执行仅对指定语句生效。但是，如果没有具有并行属性的表，或者违反了并行操作的限制，即使启用了并行 DML，DML 操作仍然会串行执行。

并行 DML 支持的场景

INSERT 的写入表有自增列时，假如自增列是主键或者分区键时，并且该列被指定时不支持并行 DML。

示例如下：

1. 创建表 tbl1，其中列 col1 为自增列并指定为主键。

   CREATE TABLE tbl1(col1 INT AUTO_INCREMENT PRIMARY KEY, col2 INT, col3 INT);
   

2. 创建表 tbl2，其中列 col1 为自增列。

   CREATE TABLE tbl1(col1 INT AUTO_INCREMENT, col2 INT, col3 INT);
   

3. 创建表 tbl3。

   CREATE TABLE tbl3(col1 INT, col2 INT, col3 INT);
   

4. INSERT 的写入数据时，因为自增列 col1 是主键，所以不支持并行 DML。

   INSERT /*+enable_parallel_dml parallel(3)*/ INTO tbl1 SELECT * FROM tbl3;
   

5. INSERT 的写入数据时，因为未指定列 col1，所以支持并行 DML。

   INSERT /*+enable_parallel_dml parallel(3)*/ INTO tbl1(col2, col3) SELECT col2, col3 FROM tbl3;
   

6. INSERT 的写入数据时，因为列 col1 只是自增列，所以支持并行 DML。

   INSERT /*+enable_parallel_dml parallel(3)*/ INTO tbl2 SELECT * FROM tbl3;