基于湖库一体架构，统一管理结构化、半结构化与非结构化等多模态数据，一个系统承载事务处理、实时分析与 AI 工作负载。
分布式计划的生成
更新时间：2026-08-26 14:06:34
编辑
分享
OceanBase 数据库的优化器会分为两大阶段来生成分布式的执行计划。
这两阶段生成执行计划的方式如下：
第一阶段：不考虑数据的物理分布，生成所有基于本地关系优化的最优执行计划。在本地计划生成后，优化器会检查数据是否访问了多个分区，或者是否访问的是本地单分区表同时用户使用 Hint 强制采用了并行查询执行。
对于分布式执行计划，分区可以提高查询性能。如果数据库关系表比较小，则不必要进行分区，如果关系表比较大，则需要根据上层业务需求谨慎选择分区键，以保证大多数查询能够使用分区键进行分区裁剪，从而减少数据访问量。
同时，对于有关联性的表，建议使用关联键作为分区键，并采用相同分区方式，使用 Table Group 将相同的分区配置在同样的节点上，以减少跨节点的数据交互。
第二阶段：生成分布式计划。根据执行计划树，在需要进行数据重分布的地方，插入
EXCHANGE节点，从而将原先的本地计划树变成分布式执行计划。
分布式执行计划的算子
生成分布式计划的过程就是在原始计划树上寻找恰当位置插入 EXCHANGE 算子的过程，在自顶向下遍历计划树的时候，需要根据相应算子的数据处理情况以及输入算子的数据分区情况，来决定是否需要插入 EXCHANGE 算子。
如下示例为最简单的单表扫描：
obclient> CREATE TABLE t1 (v1 INT, v2 INT) PARTITION BY HASH(v1) PARTITIONS 5;
Query OK, 0 rows affected
obclient> EXPLAIN SELECT * FROM t1;
Query Plan:
==============================================
|ID|OPERATOR |NAME |EST. ROWS|COST |
------------------------------------------------------
|0 |PX COORDINATOR | |500000 |545109|
|1 | EXCHANGE OUT DISTR |:EX10000|500000 |320292|
|2 | PX PARTITION ITERATOR| |500000 |320292|
|3 | TABLE SCAN |T1 |500000 |320292|
======================================================
Outputs & filters:
-------------------------------------
0 - output([T1.V1], [T1.V2]), filter(nil)
1 - output([T1.V1], [T1.V2]), filter(nil), dop=1
2 - output([T1.V1], [T1.V2]), filter(nil)
3 - output([T1.V1], [T1.V2]), filter(nil),
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。如果操作的键和输入数据的数据分布是不一致的，则需要做两阶段聚合操作，聚合算子需要做下压操作。
一阶段聚合操作如下例所示：
obclient> CREATE TABLE t2 (v1 INT, v2 INT) PARTITION BY HASH(v1) PARTITIONS 4;
Query OK, 0 rows affected
obclient> EXPLAIN SELECT SUM(v1) FROM t2 GROUP BY v1;
Query Plan:
| ======================================================
|ID|OPERATOR |NAME |EST. ROWS|COST |
------------------------------------------------------
|0 |PX COORDINATOR | |101 |357302|
|1 | EXCHANGE OUT DISTR |:EX10000|101 |357297|
|2 | PX PARTITION ITERATOR| |101 |357297|
|3 | MERGE GROUP BY | |101 |357297|
|4 | TABLE SCAN |t2 |400000 |247403|
======================================================
Outputs & filters:
-------------------------------------
0 - output([T_FUN_SUM(t2.v1)]), filter(nil)
1 - output([T_FUN_SUM(t2.v1)]), filter(nil), dop=1
2 - output([T_FUN_SUM(t2.v1)]), filter(nil)
3 - output([T_FUN_SUM(t2.v1)]), filter(nil),
group([t2.v1]), agg_func([T_FUN_SUM(t2.v1)])
4 - output([t2.v1]), filter(nil),
access([t2.v1]), partitions(p[0-3])
二阶段聚合操作如下例所示：
obclient> EXPLAIN SELECT SUM(v1) FROM t2 GROUP BY v2;
Query Plan:
| ============================================================
|ID|OPERATOR |NAME |EST. ROWS|COST |
------------------------------------------------------------
|0 |PX COORDINATOR | |101 |561383|
|1 | EXCHANGE OUT DISTR |:EX10001|101 |561374|
|2 | HASH GROUP BY | |101 |561374|
|3 | EXCHANGE IN DISTR | |101 |408805|
|4 | EXCHANGE OUT DISTR (HASH)|:EX10000|101 |408795|
|5 | HASH GROUP BY | |101 |408795|
|6 | PX PARTITION ITERATOR | |400000 |256226|
|7 | TABLE SCAN |t2 |400000 |256226|
============================================================
Outputs & filters:
-------------------------------------
0 - output([T_FUN_SUM(T_FUN_SUM(t2.v1))]), filter(nil)
1 - output([T_FUN_SUM(T_FUN_SUM(t2.v1))]), filter(nil), dop=1
2 - output([T_FUN_SUM(T_FUN_SUM(t2.v1))]), filter(nil),
group([t2.v2]), agg_func([T_FUN_SUM(T_FUN_SUM(t2.v1))])
3 - output([t2.v2], [T_FUN_SUM(t2.v1)]), filter(nil)
4 - (#keys=1, [t2.v2]), output([t2.v2], [T_FUN_SUM(t2.v1)]), filter(nil), dop=1
5 - output([t2.v2], [T_FUN_SUM(t2.v1)]), filter(nil),
group([t2.v2]), agg_func([T_FUN_SUM(t2.v1)])
6 - output([t2.v1], [t2.v2]), filter(nil)
7 - output([t2.v1], [t2.v2]), filter(nil),
access([t2.v1], [t2.v2]), partitions(p[0-3])
二元输入算子
二元输入算子主要考虑 JOIN 算子的情况。对于 JOIN 算子来说，主要基于规则来生成分布式执行计划和选择数据重分布方法。JOIN 算子主要有以下三种联接方式：
Partition-Wise Join
当左右表都是分区表且分区方式相同，物理分布一样，并且
JOIN的联接条件为分区键时，可以使用以分区为单位的联接方法。如下例所示：obclient> CREATE TABLE t3 (v1 INT, v2 INT) PARTITION BY HASH(v1) PARTITIONS 4; Query OK, 0 rows affected obclient> EXPLAIN SELECT * FROM t2, t3 WHERE t2.v1 = t3.v1; Query Plan: =========================================================== |ID|OPERATOR |NAME |EST. ROWS |COST | |0 |PX COORDINATOR | |1568160000|1227554264| |1 | EXCHANGE OUT DISTR |:EX10000|1568160000|930670004 | |2 | PX PARTITION ITERATOR| |1568160000|930670004 | |3 | MERGE JOIN | |1568160000|930670004 | |4 | TABLE SCAN |t2 |400000 |256226 | |5 | TABLE SCAN |t3 |400000 |256226 | =========================================================== Outputs & filters: ------------------------------------- 0 - output([t2.v1], [t2.v2], [t3.v1], [t3.v2]), filter(nil) 1 - output([t2.v1], [t2.v2], [t3.v1], [t3.v2]), filter(nil), dop=1 2 - output([t2.v1], [t2.v2], [t3.v1], [t3.v2]), filter(nil) 3 - output([t2.v1], [t2.v2], [t3.v1], [t3.v2]), filter(nil), equal_conds([t2.v1 = t3.v1]), other_conds(nil) 4 - output([t2.v1], [t2.v2]), filter(nil), access([t2.v1], [t2.v2]), partitions(p[0-3]) 5 - output([t3.v1], [t3.v2]), filter(nil), access([t3.v1], [t3.v2]), partitions(p[0-3])Partial Partition-Wise Join
当左右表中一个表为分区表，另一个表为非分区表，或者两者皆为分区表但是联接键仅和其中一个分区表的分区键相同的情况下，会以该分区表的分区分布为基准，重新分布另一个表的数据。如下例所示：
obclient> CREATE TABLE t4 (v1 INT, v2 INT) PARTITION BY HASH(v1) PARTITIONS 3; Query OK, 0 rows affected obclient> EXPLAIN SELECT * FROM t4, t2 WHERE t2.v1 = t4.v1; Query Plan: =========================================================== |ID|OPERATOR |NAME |EST. ROWS|COST | ----------------------------------------------------------- |0 |PX COORDINATOR | |11880 |17658| |1 | EXCHANGE OUT DISTR |:EX10001|11880 |15409| |2 | NESTED-LOOP JOIN | |11880 |15409| |3 | EXCHANGE IN DISTR | |3 |37 | |4 | EXCHANGE OUT DISTR (PKEY)|:EX10000|3 |37 | |5 | PX PARTITION ITERATOR | |3 |37 | |6 | TABLE SCAN |t4 |3 |37 | |7 | PX PARTITION ITERATOR | |3960 |2561 | |8 | TABLE SCAN |t2 |3960 |2561 | =========================================================== 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=1 2 - output([t4.v1], [t4.v2], [t2.v1], [t2.v2]), filter(nil), conds(nil), nl_params_([t4.v1]) 3 - output([t4.v1], [t4.v2]), filter(nil) 4 - (#keys=1, [t4.v1]), output([t4.v1], [t4.v2]), filter(nil), dop=1 5 - output([t4.v1], [t4.v2]), filter(nil) 6 - output([t4.v1], [t4.v2]), filter(nil), 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), 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; 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), 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), 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), access([t2.v1], [t2.v2]), partitions(p[0-3]) obclient> EXPLAIN SELECT /*+ PQ_DISTRIBUTE(t2 HASH HASH) PARALLEL(2)*/* FROM t4, t2 WHERE t2.v2 = t4.v2; Query Plan: ============================================================ |ID|OPERATOR |NAME |EST. ROWS|COST | ------------------------------------------------------------ |0 |PX COORDINATOR | |11880 |434727| |1 | EXCHANGE OUT DISTR |:EX10002|11880 |432478| |2 | HASH JOIN | |11880 |432478| |3 | EXCHANGE IN DISTR | |3 |37 | |4 | EXCHANGE OUT DISTR (HASH)|:EX10000|3 |37 | |5 | PX BLOCK ITERATOR | |3 |37 | |6 | TABLE SCAN |t4 |3 |37 | |7 | EXCHANGE IN DISTR | |400000 |294090| |8 | EXCHANGE OUT DISTR (HASH)|:EX10001|400000 |256226| |9 | PX PARTITION ITERATOR | |400000 |256226| |10| 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), equal_conds([t2.v2 = t4.v2]), other_conds(nil) 3 - output([t4.v1], [t4.v2]), filter(nil) 4 - (#keys=1, [t4.v2]), 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), access([t4.v1], [t4.v2]), partitions(p[0-2]) 7 - output([t2.v1], [t2.v2]), filter(nil) 8 - (#keys=1, [t2.v2]), output([t2.v1], [t2.v2]), filter(nil), dop=2 9 - output([t2.v1], [t2.v2]), filter(nil) 10 - output([t2.v1], [t2.v2]), filter(nil), access([t2.v1], [t2.v2]), partitions(p[0-3])
