OceanBase 数据库的优化器会分为以下两大阶段来生成分布式的执行计划。

1. 第一阶段：不考虑数据的物理分布，生成所有基于本地关系优化的最优执行计划。在本地计划生成后，优化器会检查数据是否访问了多个分区，或者是否访问的是本地单分区表但是用户使用 HINT 强制采用了并行查询执行。

2. 第二阶段：生成分布式计划。根据执行计划树，在需要进行数据重分布的地方，插入 EXCHANGE 节点，从而将原先的本地计划树变成分布式执行计划。

分布式执行计划的算子

生成分布式计划的过程就是在原始计划树上寻找恰当位置插入 EXCHANGE 算子的过程，在自顶向下遍历计划树的时候，需要根据相应算子的数据处理情况以及输入算子的数据分区情况，来决定是否需要插入 EXCHANGE 算子。

如下示例为最简单的单表扫描：

  1. obclient>CREATE TABLE t1 (v1 INT, v2 INT) PARTITION BY HASH(v1) PARTITIONS 5;
  2. Query OK, 0 rows affected (0.12 sec)
  4. obclient>EXPLAIN SELECT * FROM t1\G;
  5. *************************** 1. row ***************************
  6. Query Plan:
  7. ==============================================
  8. |ID|OPERATOR               |NAME    |EST. ROWS|COST  |
  9. ------------------------------------------------------
  10. |0 |PX COORDINATOR         |        |500000   |545109|
  11. |1 | EXCHANGE OUT DISTR    |:EX10000|500000   |320292|
  12. |2 |  PX PARTITION ITERATOR|        |500000   |320292|
  13. |3 |   TABLE SCAN          |T1      |500000   |320292|
  14. ======================================================
  16. Outputs & filters:
  17. -------------------------------------
  18.   0 - output([T1.V1], [T1.V2]), filter(nil)
  19.   1 - output([T1.V1], [T1.V2]), filter(nil), dop=1
  20.   2 - output([T1.V1], [T1.V2]), filter(nil)
  21.   3 - output([T1.V1], [T1.V2]), filter(nil),
  22.       access([T1.V1], [T1.V2]), partitions(p[0-4])

当 t1 是一个分区表，可以在 TABLE SCAN 上插入配对的 EXCHANGE 算子，从而将 TABLE SCAN 和 EXCHANGE OUT 封装成一个 job，可以用于并行的执行。

单输入可下压算子

单输入可下压算子主要包括 AGGREGATION、SORT、GROUP BY 和 LIMIT 算子等，除了 LIMIT 算子以外，其余所列举的算子都会有一个操作的键，如果操作的键和输入数据的数据分布是一致的，则可以做一阶段聚合操作，也即 Partition Wise Aggregation。如果操作的键和输入数据的数据分布是不一致的，则需要做两阶段聚合操作，聚合算子需要做下压操作。

一阶段聚合操作如下例所示：

  1. obclient>CREATE TABLE t2 (v1 INT, v2 INT) PARTITION BY HASH(v1) PARTITIONS 4;
  2. Query OK, 0 rows affected (0.12 sec)
  4. obclient>EXPLAIN SELECT SUM(v1) FROM t2 GROUP BY v1\G;
  5. *************************** 1. row ***************************
  6. Query Plan:
  7. | ======================================================
  8. |ID|OPERATOR               |NAME    |EST. ROWS|COST  |
  9. ------------------------------------------------------
  10. |0 |PX COORDINATOR         |        |101      |357302|
  11. |1 | EXCHANGE OUT DISTR    |:EX10000|101      |357297|
  12. |2 |  PX PARTITION ITERATOR|        |101      |357297|
  13. |3 |   MERGE GROUP BY      |        |101      |357297|
  14. |4 |    TABLE SCAN         |t2      |400000   |247403|
  15. ======================================================
  17. Outputs & filters:
  18. -------------------------------------
  19.   0 - output([T_FUN_SUM(t2.v1)]), filter(nil)
  20.   1 - output([T_FUN_SUM(t2.v1)]), filter(nil), dop=1
  21.   2 - output([T_FUN_SUM(t2.v1)]), filter(nil)
  22.   3 - output([T_FUN_SUM(t2.v1)]), filter(nil),
  23.       group([t2.v1]), agg_func([T_FUN_SUM(t2.v1)])
  24.   4 - output([t2.v1]), filter(nil),
  25.       access([t2.v1]), partitions(p[0-3])

二阶段聚合操作如下例所示：

  1. | ============================================================
  2. |ID|OPERATOR                     |NAME    |EST. ROWS|COST  |
  3. ------------------------------------------------------------
  4. |0 |PX COORDINATOR               |        |101      |561383|
  5. |1 | EXCHANGE OUT DISTR          |:EX10001|101      |561374|
  6. |2 |  HASH GROUP BY              |        |101      |561374|
  7. |3 |   EXCHANGE IN DISTR         |        |101      |408805|
  8. |4 |    EXCHANGE OUT DISTR (HASH)|:EX10000|101      |408795|
  9. |5 |     HASH GROUP BY           |        |101      |408795|
  10. |6 |      PX PARTITION ITERATOR  |        |400000   |256226|
  11. |7 |       TABLE SCAN            |t2      |400000   |256226|
  12. ============================================================
  14. Outputs & filters:
  15. -------------------------------------
  16.   0 - output([T_FUN_SUM(T_FUN_SUM(t2.v1))]), filter(nil)
  17.   1 - output([T_FUN_SUM(T_FUN_SUM(t2.v1))]), filter(nil), dop=1
  18.   2 - output([T_FUN_SUM(T_FUN_SUM(t2.v1))]), filter(nil),
  19.       group([t2.v2]), agg_func([T_FUN_SUM(T_FUN_SUM(t2.v1))])
  20.   3 - output([t2.v2], [T_FUN_SUM(t2.v1)]), filter(nil)
  21.   4 - (#keys=1, [t2.v2]), output([t2.v2], [T_FUN_SUM(t2.v1)]), filter(nil), dop=1
  22.   5 - output([t2.v2], [T_FUN_SUM(t2.v1)]), filter(nil),
  23.       group([t2.v2]), agg_func([T_FUN_SUM(t2.v1)])
  24.   6 - output([t2.v1], [t2.v2]), filter(nil)
  25.   7 - output([t2.v1], [t2.v2]), filter(nil),
  26.       access([t2.v1], [t2.v2]), partitions(p[0-3])

二元输入算子

二元输入算子主要考虑 JOIN 算子的情况。对于 JOIN 算子来说，主要基于规则来生成分布式执行计划和选择数据重分布方法。JOIN 算子主要有以下三种联接方式：

  • Partition-Wise Join

    当左右表都是分区表且分区方式相同，物理分布一样，并且 JOIN 的联接条件为分区键时，可以使用以分区为单位的联接方法。如下例所示：

    1. obclient>CREATE TABLE t3 (v1 INT, v2 INT) PARTITION BY HASH(v1) PARTITIONS 4;
    2. Query OK, 0 rows affected (0.12 sec)
    4. obclient>EXPLAIN SELECT * FROM t2, t3 WHERE t2.v1 = t3.v1\G;
    5. *************************** 1. row ***************************
    6. Query Plan: 
    7. ===========================================================
    8. |ID|OPERATOR               |NAME    |EST. ROWS |COST      |
    9. |0 |PX COORDINATOR         |        |1568160000|1227554264|
    10. |1 | EXCHANGE OUT DISTR    |:EX10000|1568160000|930670004 |
    11. |2 |  PX PARTITION ITERATOR|        |1568160000|930670004 |
    12. |3 |   MERGE JOIN          |        |1568160000|930670004 |
    13. |4 |    TABLE SCAN         |t2      |400000    |256226    |
    14. |5 |    TABLE SCAN         |t3      |400000    |256226    |
    15. ===========================================================
    17. Outputs & filters:
    18. -------------------------------------
    19.   0 - output([t2.v1], [t2.v2], [t3.v1], [t3.v2]), filter(nil)
    20.   1 - output([t2.v1], [t2.v2], [t3.v1], [t3.v2]), filter(nil), dop=1
    21.   2 - output([t2.v1], [t2.v2], [t3.v1], [t3.v2]), filter(nil)
    22.   3 - output([t2.v1], [t2.v2], [t3.v1], [t3.v2]), filter(nil),
    23.       equal_conds([t2.v1 = t3.v1]), other_conds(nil)
    24.   4 - output([t2.v1], [t2.v2]), filter(nil),
    25.       access([t2.v1], [t2.v2]), partitions(p[0-3])
    26.   5 - output([t3.v1], [t3.v2]), filter(nil),
    27.       access([t3.v1], [t3.v2]), partitions(p[0-3])

  • Partial Partition-Wise Join

    当左右表中一个表为分区表，另一个表为非分区表，或者两者皆为分区表但是联接键仅和其中一个分区表的分区键相同的情况下，会以该分区表的分区分布为基准，重新分布另一个表的数据。如下例所示：

    1. obclient>CREATE TABLE t4 (v1 INT, v2 INT) PARTITION BY HASH(v1) PARTITIONS 3;
    2. Query OK, 0 rows affected (0.12 sec)
    4. obclient>EXPLAIN SELECT * FROM t4, t2 WHERE t2.v1 = t4.v1\G;
    5. *************************** 1. row ***************************
    6. Query Plan:
    7.  ===========================================================
    8. |ID|OPERATOR                     |NAME    |EST. ROWS|COST |
    9. -----------------------------------------------------------
    10. |0 |PX COORDINATOR               |        |11880    |17658|
    11. |1 | EXCHANGE OUT DISTR          |:EX10001|11880    |15409|
    12. |2 |  NESTED-LOOP JOIN           |        |11880    |15409|
    13. |3 |   EXCHANGE IN DISTR         |        |3        |37   |
    14. |4 |    EXCHANGE OUT DISTR (PKEY)|:EX10000|3        |37   |
    15. |5 |     PX PARTITION ITERATOR   |        |3        |37   |
    16. |6 |      TABLE SCAN             |t4      |3        |37   |
    17. |7 |   PX PARTITION ITERATOR     |        |3960     |2561 |
    18. |8 |    TABLE SCAN               |t2      |3960     |2561 |
    19. ===========================================================
    21. Outputs & filters:
    22. -------------------------------------
    23.   0 - output([t4.v1], [t4.v2], [t2.v1], [t2.v2]), filter(nil)
    24.   1 - output([t4.v1], [t4.v2], [t2.v1], [t2.v2]), filter(nil), dop=1
    25.   2 - output([t4.v1], [t4.v2], [t2.v1], [t2.v2]), filter(nil),
    26.       conds(nil), nl_params_([t4.v1])
    27.   3 - output([t4.v1], [t4.v2]), filter(nil)
    28.   4 - (#keys=1, [t4.v1]), output([t4.v1], [t4.v2]), filter(nil), dop=1
    29.   5 - output([t4.v1], [t4.v2]), filter(nil)
    30.   6 - output([t4.v1], [t4.v2]), filter(nil),
    31.       access([t4.v1], [t4.v2]), partitions(p[0-2])
    32.   7 - output([t2.v1], [t2.v2]), filter(nil)
    33.   8 - output([t2.v1], [t2.v2]), filter(nil),
    34.       access([t2.v1], [t2.v2]), partitions(p[0-3])

  • 数据重分布

    当联接键和左右表的分区键都没有关系的情况下，可以根据规则计算来选择使用 BROADCAST 还是 HASH HASH 的数据重分布方式，如下例所示：

    注意

    只有在并行度大于 1 时, 以下示例中两种数据重分发方式才有可能被选中。

    ``` obclient>EXPLAIN SELECT /+ PARALLEL(2)/ FROM t4, t2 WHERE t2.v2 = t4.v2\G; ** 1. row *

    Query Plan:

    |ID|OPERATOR |NAME |EST. ROWS|COST |

    |0 |PX COORDINATOR | |11880 |396863| |1 | EXCHANGE OUT DISTR |:EX10001|11880 |394614| |2 | HASH JOIN | |11880 |394614| |3 | EXCHANGE IN DISTR | |3 |37 | |4 | EXCHANGE OUT DISTR (BROADCAST)|:EX10000|3 |37 | |5 | PX BLOCK ITERATOR | |3 |37 | |6 | TABLE SCAN |t4 |3 |37 | |7 | PX PARTITION ITERATOR | |400000 |256226|

    |8 | TABLE SCAN |t2 |400000 |256226|

    Outputs & filters:

    0 - output([t4.v1], [t4.v2], [t2.v1], [t2.v2]), filter(nil) 1 - output([t4.v1], [t4.v2], [t2.v1], [t2.v2]), filter(nil), dop=2 2 - output([t4.v1], [t4.v2], [t2.v1], [t2.v2]), filter(nil),

    1.   equal_conds([t2.v2 = t4.v2]), other_conds(nil)

    3 - output([t4.v1], [t4.v2]), filter(nil) 4 - output([t4.v1], [t4.v2]), filter(nil), dop=2 5 - output([t4.v1], [t4.v2]), filter(nil) 6 - output([t4.v1], [t4.v2]), filter(nil),

    1.   access([t4.v1], [t4.v2]), partitions(p[0-2])

    7 - output([t2.v1], [t2.v2]), filter(nil) 8 - output([t2.v1], [t2.v2]), filter(nil),

    1.   access([t2.v1], [t2.v2]), partitions(p[0-3])
  1. obclient>EXPLAIN SELECT /*+ PQ_DISTRIBUTE(t2 HASH HASH) PARALLEL(2)*/* FROM t4, t2 
  2.               WHERE t2.v2 = t4.v2\G;
  3. *************************** 1. row ***************************
  4. Query Plan:
  5.  ============================================================
  6. |ID|OPERATOR                     |NAME    |EST. ROWS|COST  |
  7. ------------------------------------------------------------
  8. |0 |PX COORDINATOR               |        |11880    |434727|
  9. |1 | EXCHANGE OUT DISTR          |:EX10002|11880    |432478|
  10. |2 |  HASH JOIN                  |        |11880    |432478|
  11. |3 |   EXCHANGE IN DISTR         |        |3        |37    |
  12. |4 |    EXCHANGE OUT DISTR (HASH)|:EX10000|3        |37    |
  13. |5 |     PX BLOCK ITERATOR       |        |3        |37    |
  14. |6 |      TABLE SCAN             |t4      |3        |37    |
  15. |7 |   EXCHANGE IN DISTR         |        |400000   |294090|
  16. |8 |    EXCHANGE OUT DISTR (HASH)|:EX10001|400000   |256226|
  17. |9 |     PX PARTITION ITERATOR   |        |400000   |256226|
  18. |10|      TABLE SCAN             |t2      |400000   |256226|
  19. ============================================================
  21. Outputs & filters:
  22. -------------------------------------
  23.   0 - output([t4.v1], [t4.v2], [t2.v1], [t2.v2]), filter(nil)
  24.   1 - output([t4.v1], [t4.v2], [t2.v1], [t2.v2]), filter(nil), dop=2
  25.   2 - output([t4.v1], [t4.v2], [t2.v1], [t2.v2]), filter(nil),
  26.       equal_conds([t2.v2 = t4.v2]), other_conds(nil)
  27.   3 - output([t4.v1], [t4.v2]), filter(nil)
  28.   4 - (#keys=1, [t4.v2]), output([t4.v1], [t4.v2]), filter(nil), dop=2
  29.   5 - output([t4.v1], [t4.v2]), filter(nil)
  30.   6 - output([t4.v1], [t4.v2]), filter(nil),
  31.       access([t4.v1], [t4.v2]), partitions(p[0-2])
  32.   7 - output([t2.v1], [t2.v2]), filter(nil)
  33.   8 - (#keys=1, [t2.v2]), output([t2.v1], [t2.v2]), filter(nil), dop=2
  34.   9 - output([t2.v1], [t2.v2]), filter(nil)
  35.   10 - output([t2.v1], [t2.v2]), filter(nil),
  36.       access([t2.v1], [t2.v2]), partitions(p[0-3])
  37. ```