【问底】陈焕生：深入理解Oracle 的并行执行

Oracle并行执行是一种分而治之的方法。执行一个sql 时，分配多个并行进程同时执行数据扫描，连接以及聚合等操作，使用更多的资源，得到更快的sql 响应时间。并行执行是充分利用硬件资源，处理大量数据时的核心技术。 在本文中，在一个简单的星型模型上，我会

Oracle并行执行是一种分而治之的方法。执行一个sql 时，分配多个并行进程同时执行数据扫描，连接以及聚合等操作，使用更多的资源，得到更快的sql 响应时间。并行执行是充分利用硬件资源，处理大量数据时的核心技术。

在本文中，在一个简单的星型模型上，我会使用大量例子和sql monitor 报告，力求以最直观简单的方式，向读者阐述并行执行的核心内容：

? Oracle 并行执行为什么使用生产者——消费者模型。
? 如何阅读并行执行计划。
? 不同的数据分发方式分别适合什么样的场景。
? 使用partition wise join 和并行执行的组合提高性能。
? 数据倾斜会对不同的分发方式带来什么影响。
? 由于生产者--‐消费者模型的限制，执行计划中可能出现阻塞点。
? 布隆过滤是如何提高并行执行性能的。
? 现实世界中，使用并行执行时最常见的问题。

术语说明：

1. S： 时间单位秒。
2. K： 数量单位一千。
3. M：  数量单位一百万，  或者时间单位分钟。
4. DoP： Degree of Parallelism， 并行执行的并行度。
5. QC： 并行查询的 Query Coordinator。
6. PX 进程： Parallel Execution Slaves。
7. AAS： Average active session， 并行执行时平均的活动会话数。
8. 分发： pq distribution method， 并行执行的分发方式， 包括 replicate， broadcast， hash 和 adaptive分发等 4 种方式， 其中 adaptive 分发是 12c 引入的的新特性， 我将在本篇文章中一一阐述。
9. Hash join 的左边： 驱动表， the build side of hash join， 一般为小表。
10. Hash join 的右边： 被驱动表， the probe side of hash join， 一般为大表。
11. 布隆过滤： bloom filter， 一种内存数据结构， 用于判断一个元素是否属于一个集合。

测试环境和数据

Oracle版本为12.1.0.2.2，两个节点的RAC，硬件为ExadataX3--‐8。

这是一个典型的星型模型，事实表lineorder有3亿行记录，维度表part/customer分别包含1.2M

和1.5M行记录，3个表都没有进行分区，lineorder大小接近30GB。

select
owner seg_owner，
segment_name seg_segment_name， round(bytes/1048576，2) SEG_MB
from
 
dba_segments where
owner = 'SID'
and segment_name in ('LINEORDER'，'PART'，'CUSTOMER')
/

OWNER	SEGMENT_NAME SEGMENT_TYPE	SEG_MB
------ ------------ ------------ -------- SID	LINEORDER	TABLE	30407.75
SID	CUSTOMER	TABLE	168
SID	PART	TABLE	120

本篇文章所有的测试，除非特别的说明，我关闭了12c的adaptive plan特性，参数optimizer_adaptive_features被默认设置为false。Adaptive相关的特性如cardinality feedback，adaptive distribution method，adaptive join都不会启用。如果检查执行计划的outline数据，你会发现7个优化器相关的隐含参数被设置为关闭状态。事实上，12c优化器因为引入adaptive plan特性，比以往版本复杂得多，剖析12c的优化器的各种新特性，我觉得非常具有挑战性，或许我会在另一篇文章里尝试一下。

select * from table(dbms_xplan.display_cursor('77457qc9a324k',0,’outline’));
...
Outline Data
-------------
/*+
BEGIN_OUTLINE_DATA
IGNORE_OPTIM_EMBEDDED_HINTS
OPTIMIZER_FEATURES_ENABLE('12.1.0.2')
DB_VERSION('12.1.0.2')
OPT_PARAM('_optimizer_use_feedback' 'false')
OPT_PARAM('_px_adaptive_dist_method' 'off')
OPT_PARAM('_optimizer_dsdir_usage_control' 0)
OPT_PARAM('_optimizer_adaptive_plans' 'false')
OPT_PARAM('_optimizer_strans_adaptive_pruning' 'false')
OPT_PARAM('_optimizer_gather_feedback' 'false')
OPT_PARAM('_optimizer_nlj_hj_adaptive_join' 'false')
OPT_PARAM('optimizer_dynamic_sampling' 11)
ALL_ROWS
……
END_OUTLINE_DATA
*/

---

并行初体验

串行执行

以下sql对customers和lineorder连接之后，计算所有订单的全部利润。 串行执行时不使用parallel hint：

select /*+ monitor */
sum(lo_revenue)
from
lineorder, customer
where
lo_custkey = c_custkey;

串行执行时，sql执行时间为1.5分钟，dbtime为1.5分钟。执行计划有5行，一个用户进程工作完成了对customer，lineorder两个表的扫描，hashjoin，聚合以及返回数据的所有操作。此时AAS(average active sessions)为1，sql执行时间等于db time。几乎所有的dbtime都为db cpu，72%的cpu花在了第二行的hash join操作。因为测试机器为一台Exadata X3——8，30GB的IO请求在一秒之内处理完成。Celloffload Efficiency等于87%意味着经过存储节点扫描，过滤不需要的列，最终返回计算节点的数据大小只有30GB的13%。

并行执行

使用hint parallel(4)，指定DoP=4并行执行同样的sql：

select /*+ monitor parallel(4)*/
sum(lo_revenue)
from
lineorder, customer
where
lo_custkey = c_custkey;

SQL执行时间为21s，db time为1.4分钟。DoP=4，在两个实例上执行。执行计划从5行增加为9行，从下往上分别多了’PXBLOCKITERATOR’， ‘SORTAGGREGATE’， ‘PXSENDQC(RANDOM)’ 和 ’PXCOORDINATOR’ 这四个操作。

其中3到8行的操作为并行处理，sql的执行顺序为：每个PX进程扫描维度表customer(第6行)，以数据块地址区间作为单位(第7行)扫描四分之一的事实表lineorder(第8行)，接着进行hash join(第5行)，然后对连接之后的数据做预先聚合(第4行)，最后把结果给QC(第三行)。QC接收数据(第2行)之后，做进一步的汇总(第1行)，最后返回数据(第0行)。

SQL执行时间比原来快了4倍，因为最消耗时间的操作，比如对lineorder的全表扫描，hashjoin和聚合，我们使用4个进程并行处理，因此最终sql执行时间为串行执行的1/4。另一方面，dbtime并没有明显下降，并行时1.4m，串行时为1.5m，从系统的角度看，两次执行消耗的系统资源是一样的。

              

DoP=4时，因为没有涉及数据的分发(distribution)，QC只需分配一组PX进程，四个PX进程分别为实例1和2的p000/p0001。我们可以从系统上查看这4个PX进程。每个PX进程消耗大致一样的db time，CPU和IO资源。AAS=4，这是最理想的情况，每个PX进程完成同样的工作量，一直保持活跃。没有串行点，没有并行执行倾斜。

AAS=4，查看活动信息时，为了更好的展示活动信息，注意点掉”CPU Cores”这个复选框。

在Linux系统上显示这四个PX进程。

[oracle@exa01db01 sidney]$ ps -ef | egrep "p00[01]_SSB"
oracle 20888 1 4 2014 ? 18:50:59 ora_p000_SSB1
oracle 20892 1 4 2014 ? 19:01:29 ora_p001_SSB1
[oracle@exa01db01 sidney]$ ssh exa01db02 'ps -ef | egrep "p00[01]_SSB"'
oracle 56910 1 4 2014 ? 19:01:03 ora_p000_SSB2
oracle 56912 1 4 2014 ? 18:53:30 ora_p001_SSB2

小结

本节的例子中，DoP=4，并行执行时分配了4个PX进程，带来4倍的性能提升。SQL monitor报告包含了并行执行的总体信息和各种细节，比如QC，DoP，并行执行所在的实例，每个PX进程消耗的资源，以及执行SQL时AAS。

---

生产者-消费者模型

在上面并行执行的例子中，每个px进程都会扫描一遍维度表customer，然后扫描事实表lineorder进行hash join。这时没有数据需要进行分发，只需要分配一组px进程。这种replicate维度表的行为，是12c的新特性，由参数_px_replication_enabled控制。

更常见情况是并行执行时，QC需要分配两组PX进程，互为生产者和消费者协同工作，完成并行执行计划。架构图1如下：

             

Broadcast分发，一次数据分发

为了举例说明两组px进程如何协作的，设置_px_replication_enabled为false。QC会分配两组PX进程，一组为生产者，一组为消费者。

见下图，此时sql执行时间为23s，执行时间变慢了2s，dbtime仍为1.5分钟。

            

最大的变化来自执行计划，现在执行计划有12行。增加了对customer的并行扫描 PXBLOCKITERATOR (第8行)，分发’PXSENDBROADCAST’和接收’PXRECEIVE’。执行计划中出现了两组PX进程，除了之前蓝色的多人标志，现在出现了红色的多人标志。此时，SQL的执行顺序为：

1. 4个红色的PX进程扮演生产者角色，扫描维度表customer，把数据通过broadcast的方式分发给每一个扮演消费者的蓝色PX进程。因为DoP=4，每一条被扫描出来的记录被复制了4份，从sqlmonitor的第9行，customer全表扫描返回1。5m行数据，第8行的分发和第7行的接受之时，变成了6m行记录，每个作为消费者的蓝色px进程都持有了一份完整包含所有custome记录的数据，并准备好第5行hashjoin的buildtable。
2. 4个作为消费者的蓝色PX进程，以数据块地址区间为单位扫描事实表lineorder(第10/11行);同时和已经持有的customer表的数据进hashjoin(第5行)，然后对满足join条件的数据做预聚合(第4行)，因为我们查询的目标是对所有lo_revenue求和，聚合之后每个PX进程只需输出一个总数。
3. 4个蓝色的PX进程反过来作为生产者，把聚合的数据发给消费者QC(第3行和第2行)。由QC对接收到4行记录做最后的聚合，然后返回给用户。
4. 使用broadcast的分发方式，只需要把customer的数据广播给每个消费者。Lineorder的数不需要重新分发。因为lineorder的数据量比customer大的多，应该避免对lineorder的数据进行分发，这种执行计划非常适合星型模型的数据。

               
               

生产者-消费者模型工作原理

并行查询之后，可以通过视图V$PQ_TQSTAT，验证以上描述的执行过程。

1. 实例1、2上的p002/p003进程作为生产者，几乎平均扫描customer的1/4记录，把每一条记录广播给4个消费者PX进程，发送的记录数之和为6m行。通过table queue0(TQ_ID=0)，每个作为消费者的p000/p001进程，接收了完整的1。5m行customer记录，接收的记录数之和为 6m行。
2. 实例1、2上的p000/p0001进程作为生产者，通过table queue1(TQ_ID=1)，把聚合的一条结果记录发给作为消费者的QC。QC作为消费者，接收了4行记录。

SELECT
dfo_number, tq_id, server_type, instance, process, num_rows
FROM
V$PQ_TQSTAT
ORDER BY
dfo_number DESC, tq_id, server_type desc, instance, process;
DFO_NUMBER TQ_ID SERVER_TYPE INSTANCE PROCESS NUM_ROWS
---------- ---------- -------------- ---------- --------- ----------
1 0 Producer 1 P002 1461932
1 0 Producer 1 P003 1501892
1 0 Producer 2 P002 1575712
1 0 Producer 2 P003 1460464
1 0 Consumer 1 P000 1500000
1 0 Consumer 1 P001 1500000
1 0 Consumer 2 P000 1500000
1 0 Consumer 2 P001 1500000
1 1 Producer 1 P000 1
1 1 Producer 1 P001 1
1 1 Producer 2 P000 1
1 1 Producer 2 P001 1
1 1 Consumer 1 QC 4
13 rows selected.

那么，以上的输出中，DFO_NUMBER和TQ_ID这两列表示什么意思呢?

1. DFO代表Data Flow Operator，是执行计划中可以并行执行的操作。一个QC代表一棵DFO树(tree)，包含多个DFO;同一个QC中所有并行操作的DFO_NUMBER是相同的，此例中，所有DFO_NUMBER为1。执行计划包含多个QC的例子也不少见，比如使用unionall的语句，unionall每个分支都是独立的DFO树，不同的DFO树之间可以并行执行。本篇文章仅讨论执行计划只有一个QC的情况。
2. TQ代表table queue，用以PX进程之间或者和QC通信连接。以上执行计划中，table queue0为PX进程之间的连接，table queue1为PX进程和QC之间的连接。生产者通过table queue分发数据，消费者从tablequeue接收数据。不同的table queue编号，代表了不同的数据分发。通过table queue，我们可以理解Oracle并行执行使用生产者--‐消费者模型的本质：

• 同一棵DFO树中，最多只有两组PX进程。每个生产者进程都存在一个和每个消费者进程的连接，每个PX进程和QC都存在一个连接。假设DoP=n，连接总数为(n*n+2*n)，随着n的增长，连接总数会爆炸型增长。Oracle并行执行设计时，采用生产者和消费者模型，考虑到连接数的复杂度，每个DFO最多只分配两组PX进程。假设DoP=100时，两组PX进程之间的连接总数为10000。假设可以分配三组PX进程一起完成并行执行计划，那么三组PX之间连接总数会等于1百万，维护这么多连接，是一个不可能的任务。
• 同一棵DFO树中，两组PX进程之间，同一时间只存在一个活跃的数据分发。如果执行路径很长，数据需要多次分发，两组PX进程会变换生产者消费者角色，相互协作，完成所有并行操作。每次数据分发，对应的tablequeue的编号不同。一个活跃的数据分发过程，需要两组PX进程都参与，一组为生产者发送数据，一组为消费者接收数据。因为一个DFO里最多只有两组PX进程，意味着，PX进程之间，同一时间只能有一个活跃的数据分发。如果PX进程在执行计划中需要多次分发数据，可能需要在执行计划插入一些阻塞点，比如BUFFERSORT和HASHJOINBUFFERED这两个操作，保证上一次的数据分发完成之后，才开始下一次分发。在后面的章节，我将会说明这些阻塞点带来什么影响。这个例子中，tablequeue0和1可以同时工作是因为：tablequeue0是两组PX进程之间的链接，tablequeue1为PX进程和QC之间的连接，tablequeue0与tablequeue1是相互独立的，因此 可以同时进行。
• PX进程之间或者与QC的连接至少存在一个(单节点下至多三个，RAC环境下至多四个)消息缓冲区用于进程间数据交互，该消息缓冲区默认在Largepool中分配(如果没有配置Largepool则在Sharedpool中分配)。多个缓冲区是为了实现异步通信，提高性能。
• 每个消息缓冲区的大小由参数parallel_execution_message_size控制，默认为16k。
• 当两个进程都在同一个节点的时候，通过在Largepool(如果没有配置Largepool则Sharedpool)中传递和接收消息缓冲进行数据交互。当两个进程位于不同节点时。通过RAC心跳网络进行数据交互，其中一方接收的数据需要缓存在本地Largepool(如果没有配置Largepool则Sharedpool)里面。

小结

为了说明并行执行的生产者--消费者模型是如何工作的，我使用了broad cast分发，QC分配两组PX进程，一组为生产者，一组为消费者。QC和PX进程之间，两组PX进程之间通过table queue进行数据分发，协同完成整个并行执行计划。视图V$PQ_TQSTAT记录了并行执行过程中，数据是如何分发的。通过对DFO，table queue的描述，我阐述生产者--‐消费者模型的工作原理和通信过程，或许有些描述对你来说过于突然，不用担心，后面的章节我会通过更多的例子来辅助理解。

---

如何阅读并行执行计划

Table queue 的编号代表了并行执行计划中，数据分发的顺序。理解执行计划中的并行操作是如何被执行的，原则很简单：跟随Tablequeue的顺序。

通过sqlmonitor报告判断sql的执行顺序，需要结合name列的tablequeue名字比如：TQ10000(代表DFO=1，tablequeue0)，：TQ10001(代表DFO=1，tablequeue1)，还有PX进程的颜色，进行确定。

下面的例子为dbms_xplan。display_cursor 的输出。对于并行执行计划，会多出来三列：

1. TQ列：为Q1：00或者Q1：01，其中Q1代表第一个DFO，00或者01代表tablequeue的编号。

a. ID7~9的操作的TQ列为Q1，00，该组PX进程，作为生产者首先执行，然后通过broadcast 的分发方式，把数据发给消费者。

b. ID10~11，3~6的操作的TQ列为Q1，01，该组PX进程作为消费者接受customer的数据之后，扫描lineorder，hashjoin，聚合之后，又作为生产者通过tablequeue2把数据 发给QC。

2. In--‐out 列：表明数据的流动和分发。

      ?  PCWC：parallelcombinewithchild。

      ?  PCWP：parallelcombinewithparent。

      ?  P--‐>P：  paralleltoparallel。

      ?  P--‐>S：  paralleltoSerial。

3. PQDistribute 列：数据的分发方式。此执行计划中，我们使用了broadcast 的方式，下面的章节

    我会讲述其他的分发方式。

HASH分发方式， 两次数据分发

除了broadcast分发方式，另一种常见的并行分发方式为hash。为了观察使用hash分发时sql的 执行情况，我对sql使用pq_distributehint 。

select /*+ monitor parallel(4)
         leading(customer lineorder)
         use_hash(lineorder)
         pq_distribute(lineorder hash hash) */
    sum(lo_revenue)
from
    lineorder, customer
where
    lo_custkey = c_custkey;

使用hash分发方式时，sql的执行时间为29s，dbtime为2.6m。相对于broadcast方式，sql的执行时间和dbtime都增加了大约40%。

执行计划如下，执行计划为14行，增加了对lineorder的hash分发，第11行的’PXSENDHASH’对3亿行数据通过hash函数分发，第10行的’PXRECEIVE’通过tablequeue1接收3亿行数据，这两个操作消耗了38%的dbcpu。这就是为什么SQL执行时间和dbtime变长的原因。此时，SQL的执行顺序为：

1. 红色的PX进程作为生产者，并行扫描customer(第8~9行)，对于连接键c_custkey运用函数，根据每行记录的hash值，通过table queue0，发给4个蓝色消费者的其中一个(第7行)。Hash分发方式并不会复制数据，sql monitor报告的第6~9行，actual rows列都为1.5m。
2. 红色的PX进程作为生产者，并行扫描li neorder(第12~13行)，对于连接键l o_custkey运用同样的 dhash函数，通过tablequeue1 ，发给4个蓝色消费者的其中一个(第11行)。同样的hash函数保证了customer和li neorder相同的连接键会发给同一个消费者，保证hashj oin结果的正确。因为3亿行数据都需要经过hash函数计算，然后分发(这是进程间的通信，或者需要通过RAC心跳网络通信)，这些巨大的额外开销，就是增加38% cpu的原因。

3. 4个蓝色的PX进程作为消费者接收了customer的1.5M行记录(第 6 行)，和lineorder的3亿行记录(第10行)，进行hash join(第5行)，预聚合(第4行)。

4. 4个蓝色的PX进程反过来作为生产者，通过table queue2，把聚合的数据发给消费者QC(第3 行和第2行)。由QC对接收到4行记录做最后的聚合， 然后返回给用户(第1和0行)。

因为涉及3亿行数据的分发和接收，作为生产者的红色PX进程和作为消费者的蓝色PX进程需要同时活跃，SQL monitor报告中的activity信息显示大部分时间，AAS超过并行度4，意味这两组PX进程同时工作。不像replicate或者broadcast分发时，AAS为4，只有一组PX进程保持活跃。

SELECT
dfo_number, tq_id, server_type, instance, process, num_rows
FROM
V$PQ_TQSTAT
ORDER BY
dfo_number DESC, tq_id, server_type desc, instance, process;
DFO_NUMBER TQ_ID SERVER_TYPE INSTANCE PROCESS NUM_ROWS
---------- ---------- -------------- ---------- --------- ----------
1 0 Producer 1 P002 299928364
1 0 Producer 1 P003 299954384
1 0 Producer 2 P002 300188788
1 0 Producer 2 P003 299951708
1 0 Consumer 1 P000 300005811
1 0 Consumer 1 P001 300005811
1 0 Consumer 2 P000 300005811
1 0 Consumer 2 P001 300005811
1 1 Producer 1 P000 1
1 1 Producer 1 P001 1
1 1 Producer 2 P000 1
1 1 Producer 2 P001 1
1 1 Consumer 1 QC 4
13 rows selected.

select /*+ monitor parallel(4)*/
sum(lo1.lo_revenue)
from
lineorder_hash32 lo1, lineorder_hash32 lo2
where
lo1.lo_orderkey = lo2.lo_orderkey;

并行查询之后，通过视图V$PQ_TQSTAT，进一步验证以上描述的执行过程。并行执行过程涉及3

个tablequeue0/1/2，V$PQ_TQSTAT包含21行记录。

1.　实例1、2上的p002/p003进程作为生产者，平均扫描customer的1/4记录，然后通过tablequeue0(TQ_ID=0)，发给作为消费者的p000/p001进程。发送和接收的customer记录之和都为 1.5m。

        ?  发送的记录数：1500000= 365658+364899+375679+393764

        ?  接收的记录数：1500000= 374690+374924+375709+374677

2. 　实例1、2上的p002/p0003进程作为生产者，平均扫描lineorder的1/4记录，通过table queue1(TQ_ID=1) ，发给作为消费者的p000/p001进程。发送和接收的lineorder 记录之和都为300005811。

        ?  发送的记录数：300005811= 74987629+75053393+74979748+74985041 

        ?  接收的记录数：300005811= 74873553+74968719+75102151+75061388

3.　实例1、2上的p000/p0001进程作为生产者，通过tablequeue2(TQ_ID=2)，把聚合的一条结果记 录发给作为消费者的QC。QC作为消费者，接收了4行记录。 

SELECT
dfo_number, tq_id, server_type, instance, process, num_rows
FROM
V$PQ_TQSTAT
ORDER BY
dfo_number DESC, tq_id, server_type desc, instance, process;
DFO_NUMBER TQ_ID SERVER_TYPE INSTANCE PROCESS NUM_ROWS
---------- ---------- ---------------- ---------- --------- ----------
1 0 Producer 1 P002 365658
1 0 Producer 1 P003 364899
1 0 Producer 2 P002 375679
1 0 Producer 2 P003 393764
1 0 Consumer 1 P000 374690
1 0 Consumer 1 P001 374924
1 0 Consumer 2 P000 375709
1 0 Consumer 2 P001 374677
1 1 Producer 1 P002 74987629
1 1 Producer 1 P003 75053393
1 1 Producer 2 P002 74979748
1 1 Producer 2 P003 74985041
1 1 Consumer 1 P000 74873553
1 1 Consumer 1 P001 74968719
1 1 Consumer 2 P000 75102151
1 1 Consumer 2 P001 75061388
1 2 Producer 1 P000 1
1 2 Producer 1 P001 1
1 2 Producer 2 P000 1
1 2 Producer 2 P001 1
1 2 Consumer 1 QC 4
21 rows selected.

小结

数组大小m，可以把错误判断的几率控制在很小的范围之内。

我们观察hash分发时sql的并行执行过程。Hash分发与broadcast最大的区分在于对hashjoin的 两边都进行分发。这个例子中，对lineorder 的hash分发会增加明显的dbcpu 。下一节，我将使用另一个例子，说明hash分发适用的场景。

---

Replicate，Broadcast和Hash的选择

我们已经测试过replicate，broadcast，和hash这三种分发方式。

1. Replicate ：每个PX进程重复扫描hashjoin 的左边，buffercache 被用来缓存hashjoin 左边的小表，减少重复扫描所需的物理读。相对于broadcast 分发，replicate 方式只需一组PX进程。但是repli cate不能替换br oadcast分发。因为repli cate仅限于hashj oin左边是表的情况，如果 hashjoin的左边的结果集来自其他操作，比如j oin或者视图，那么此时无法使用repli cate。

2. Broadcast分发：作为生产者的PX进程通过广播的方式，把hashjoin左边的结果集分发给每 个作为消费者的PX进程。一般适用于hashjoin 左边结果集比右边小得多的场景，比如星型模型。

3. Hash分发的本质：把hashjoin的左边和右边(两个数据源)，通过同样hash函数重新分发，切 分为N个工作单元(假设DoP=N)，再进行join ，目的是减少PX进程进行join 操作时，需要连接的数据量。Hash分发的代价需要对hashjoin 的两边都进行分发。对于customer连接li neorder的例子，因为维度表customer的数据量比事实表li neorder小得多，对customer进行repli cate或者broadcast 分发显然是更好的选择，因为这两种方式不用对lineorder 进行重新分发。如果是两个大表join 的话，join操作会是整个执行计划的瓶颈所在，hash分发是唯一合适的方式。为了减低j oin的代价，对hashj oin左边和右边都进行hash分发的代价是可以接 受的。

Hash分发，有时是唯一合理的选择

我们使用lineorder上的自连接来演示，为什么有时hash分发是唯一合理的选择。测试的SQL如 下：

select /*+ monitor parallel(4)*/
sum(lo1.lo_revenue)
from
lineorder lo1, lineorder lo2
where
lo1.lo_orderkey = lo2.lo_orderkey;

SQL执行时间为2.4分钟，dbtime为10.5分钟。

优化器默认选择hash分发方式，执行计划为14行，结构与之前的Hash分发的例子是一致的。不 同的是，第5行的hash join消耗了73%的db time，使用了9GB的临时表空间，表空间的IO占12%的db time。大约15%的db time用于Lineorder的两次hash分发和接收，相对上一个例子的占38%比例，这两次HASH分发的整体影响降低了一倍多。

红色的PX进程为实例1、2上的p002/p003进程，蓝色的PX进程为p000/p001进程。作为生产者的红色PX进程占总db time的15%左右。

SQL执行开始，对lineorder两次hash分发时，AAS大于4，分发完成之后，只有蓝色的PX进程进行 hash join操作，AAS=4。

从V$PQ_TQSTAT视图可以确认，对于lineorder的存在两次分发，通过table queue0和1，作为消费者的4个PX进程接收到的两次数据是一样的，保证重新分发不会影响join结果的正确性。每个蓝色PX 进程需要hash join的左边和右边均为3亿行数据的1/4，通过hash分发，3亿行记录连接3亿行记录的工作平均的分配四个独立PX进程各自处理，每个PX进程处理75M行记录连接75M行记录。

SELECT
dfo_number, tq_id, server_type, instance, process, num_rows
FROM
V$PQ_TQSTAT
ORDER BY
dfo_number DESC, tq_id, server_type desc, instance, process;
DFO_NUMBER TQ_ID SERVER_TYPE INSTANCE PROCESS NUM_ROWS
---------- ---------- -------------- ---------- --------- ----------
1 0 Producer 1 P002 75055725
1 0 Producer 1 P003 74977459
1 0 Producer 2 P002 74995276
1 0 Producer 2 P003 74977351
1 0 Consumer 1 P000 74998419
1 0 Consumer 1 P001 74995836
1 0 Consumer 2 P000 74976974
1 0 Consumer 2 P001 75034582
1 1 Producer 1 P002 74986798
1 1 Producer 1 P003 74985268
1 1 Producer 2 P002 74984883
1 1 Producer 2 P003 75048862
1 1 Consumer 1 P000 74998419
1 1 Consumer 1 P001 74995836
1 1 Consumer 2 P000 74976974
1 1 Consumer 2 P001 75034582
1 2 Producer 1 P000 1
1 2 Producer 1 P001 1
1 2 Producer 2 P000 1
1 2 Producer 2 P001 1
1 2 Consumer 1 QC 4
21 rows selected.

使用 broadcast 分发，糟糕的性能

对于lineorder,lineorder的自连接， 如果我们使用broadcast分发，会出现什么情况呢? 我们测试一下：

select /*+ monitor parallel(4)
leading(lo1 lo2)
use_hash(lo2)
pq_distribute(lo2 broadcast none) */
15
sum(lo1.lo_revenue)
from
lineorder lo1, lineorder lo2
where
lo1.lo_orderkey = lo2.lo_orderkey;

使用broadcase分发，SQL的执行时间为5.9分钟，db time为23.8分钟。相比hash分发，执行时间和 db time都增加了接近1.5倍。

红色的PX进程作为生产者，对lineorder进行并行扫描之后，3亿行记录通过tablequeue0广播给4个作为消费者的蓝色PX进程(第6~9行)，相当于复制了4份，每个蓝色的PX进程都接收了3亿行记录.这次broadcast分发消耗了11%的db time，因为需要每行记录传输给每个蓝色PX进程，消耗的db cpu比使用hash分发时两次hash分发所消耗的还多。

第5行的hash join的所消耗的临时表空间上升到27GB，临时表空间IO占的db time的38%。因为每个蓝色PX进程进行hash join的数据变大了，hash join的左边为3亿行数据，hash join的右边为3亿行记录的1/4.

蓝色PX进程为消费者负责hash join，所消耗的db time都大幅增加了。

hash join时，临时表空间读等待事件’direct path read temp’明显增加了。

V$PQ_TQSTAT的输出中，实例1、2上的p000/p001进程作为消费者，都接收了3亿行数据，造成后续hash join的急剧变慢。Broadcast分发对hash join左边进行广播的机制，决定了它不适合hash join两边都为大表的情况。

SELECT
dfo_number, tq_id, server_type, instance, process, num_rows
FROM
V$PQ_TQSTAT
ORDER BY
dfo_number DESC, tq_id, server_type desc, instance, process;
DFO_NUMBER TQ_ID SERVER_TYPE INSTANCE PROCESS NUM_ROWS
---------- ---------- -------------- ---------- --------- ----------
1 0 Producer 1 P002 299928364
1 0 Producer 1 P003 299954384
1 0 Producer 2 P002 300188788
1 0 Producer 2 P003 299951708
1 0 Consumer 1 P000 300005811
1 0 Consumer 1 P001 300005811
1 0 Consumer 2 P000 300005811
1 0 Consumer 2 P001 300005811
1 1 Producer 1 P000 1
1 1 Producer 1 P001 1
1 1 Producer 2 P000 1
1 1 Producer 2 P001 1
1 1 Consumer 1 QC 4
13 rows selected.

小结，Broadcast和Hash分发的陷阱

通过前一节和本节的例子，我们知道，如果选择了不合理的分发方式，SQL执行时性能会明显下降

1. 对于broadcast分发：只对hash join的左边进行分发，但是采用广播分发，hash join时左边的数据量并没有减少，如果hash join左边的包含大量数据，并行对hash join性能改善有限。对大量数据的broadcast分发也会消耗额外的db cpu，比如本节中lineorder自连接的例子。  Replicate 同理。

2. 对于hash分发：对hash join的两边都进行分发，使每个PX进程进行hash join时，左边和右边的数据量都为原始的1/N，N为并行度。Hash分发的潜在陷阱在于：

   ?两次分发，尤其对大表的分发，可能带来明显的额外开销，比如前一节customer连接lineorder 的例子。使用Partition wise join可以消除分发的需要，后面会举例说明。

   ?如果数据存在倾斜，连接键上的少数值占了大部分的数据，通过hash分发，同一个键值的记录会分发给同一个PX进程，某一个PX进程会处理大部分数据的hash join，引起并行执行倾斜。我会在后面的章节说明这种情况和解决方法。

SQL解析时，优化器会根据hash join左边和右边估算的cardinality，并行度等信息，选择具体何种分发方式。维护正确的统计信息，对于优化器产生合理的并行执行计划是至关重要的。

---

Partition Wise Join，消除分发的额外开销

无论对于broadcast或者hash分发，数据需要通过进程或者节点之间通信的完成传输，分发的数据越多，消耗的db cpu越多。并行执行时，数据需要分发，本质上是因为Oracle采用share---everything的集中存储架构，任何数据对每个实例的PX进程都是共享的。为了对hash join操作分而治之，切分为N个独立的工作单元(假设 DoP=N)，必须提前对数据重新分发，数据的分发操作就是并行带来的额外开销。

使用full或者partial partition wise join技术，可以完全消除分发的额外开销，或者把这种开销降到最低。如果hash join有一边在连接键上做hash分区，那么优化器可以选择对分区表不分发，因为hash分区已经对数据完成切分，这只需要hash分发hash join的其中一边，这是partial partition wise join。如果hash join的两边都在连接键上做了hash join分区，那么每个PX进程可以独立的处理对等的hash分区， 没有数据需要分发，这是full partition wise join。hash分区时，hash join的工作单元就是对等hash分区包含的数据量，应该控制每个分区的大小，hash join时就可能消除临时表空间的使用，大幅减少所需的PGA。

Partition Wise Join，不需要数据分发。

如果在lineorder的列lo_orderkey上做hash分区，分区数为32个。每个分区的大小接近1G。

SEGMENT_NAME SEG_PART_NAME SEGMENT_TYPE SEG_MB
------------------ --------------- -------------------- ----------
LINEORDER_HASH32 SYS_P3345 TABLE PARTITION 960
LINEORDER_HASH32 SYS_P3344 TABLE PARTITION 960
...
LINEORDER_HASH32 SYS_P3315 TABLE PARTITION 960
LINEORDER_HASH32 SYS_P3314 TABLE PARTITION 960
----------
30720
32 rows selected.

使用lo_orderkey 连接时，lineorder不需要再分发。我们继续使用自连接的sql，演示full partition wise join。

select /*+ monitor parallel(4)*/
sum(lo1.lo_revenue)
from
lineorder_hash32 lo1, lineorder_hash32 lo2
where
lo1.lo_orderkey = lo2.lo_orderkey;

此时sql执行时间为1.6分钟，dbtime 6分钟;不分区使用hash分发时，执行时间为2.4分钟，db time 10.5 分钟。使用Partition Wise join快了三分之一。执行计划中只有一组蓝色的PX进程，不需要对数据进行分发。因为lineorder_hash32的3亿行数据被切分为32个分区。虽然并行度为4，每个PX进程hash join时，工作单元为一对匹配的hash分区，两边的数据量都为3亿的1/32。更小的工作单元，使整个hash join消耗的临时表空间下降为 448MB。每个PX进程消耗8对hash分区，可以预见，当我们把并行度提高到8/16/32，每个PX进程处理的hash分区对数，应该分别为4/2/1，sql执行时间会线性的下降。

蓝色的PX进程为、的p000/p001进程。每个PX进程消耗的db time是平均的，每个PX进程均处理了8对分区的扫描和hash join。

AAS绝大部分时间都为4。

唯一的数据连接为tablequeue0，每个PX进程向QC发送一行记录。

SELECT
dfo_number, tq_id, server_type, instance, process, num_rows
FROM
V$PQ_TQSTAT
ORDER BY
dfo_number DESC, tq_id, server_type desc, instance, process;
DFO_NUMBER TQ_ID SERVER_TYPE INSTANCE PROCESS NUM_ROWS
---------- ---------- ------------- ---------- ---------- ----------
1 0 Producer 1 P000 1
1 0 Producer 1 P001 1
1 0 Producer 2 P000 1
1 0 Producer 2 P001 1
1 0 Consumer 1 QC 4
5 rows selected

当DoP大于分区数时，Partition Wise Join不会发生

当并行执行的DoP大于hash分区数时，partition wise join不会发生，这时优化器会使用 broadcast local的分发。使用DoP=64执行同样的sql：

select /*+ monitor parallel(64)*/ sum(lo1。lo_revenue)
from
lineorder_hash32 lo1， lineorder_hash32 lo2 where
lo1。lo_orderkey = lo2。lo_orderkey

DoP=64，查询执行时间为15秒，db time为11.3分钟。

执行计划中出现了两组PX进程。优化器选择对hash join的右边进行broadcast local分发。如果hash join的左边比较小的话，broadcast local会发生在hash join的左边。因为DoP是分区数的两倍，hash join两边的lineorder_hash64的每个分区，由2个PX进程共同处理。处理一对匹配分区的两个蓝色的PX进程和两个红色的PX进程，会处在同一个实例上。数据只会在同一个实例的PX进程之间，不会跨实例传输，降低数据分发成本，这是broadcast local的含义。SQL的执行顺序如下：

1. 以数据库地址区间为单位，蓝色的PX进程并行扫描hash join左边的lineorder_hash32(第7行)，因为DoP是分区数的两倍，每个分区由两个蓝色PX进程共同扫描，这两个PX进程在同一个实例上。每个蓝色的PX进程大约扫描每个分区一半的数据，大约4.7M行记录，并准备好第5行hash join的build table。
2. 红色的PX进程并行扫描hash join右边的lineorder_hash32，每个红色的PX进程大概扫描4.7M行记录，然后tablequeue0，以broadcast local的方式，分发给本实例两个红色的PX进程(数据分发时，映射到本实例某些PX进程，避免跨节点传输的特性，称为slaves mapping，除了broadcast local，还有hash local，random local等分发方式)。通过broadcast local分发，数据量从300M行变成600M行。
3. 每个蓝色的PX进程通过tablequeue0接收了大概9.4M行数据，这是整个匹配分区的数据量。然后进行hash join，以及之后的聚合操作。每个蓝色的PX进程hash join操作时，左边的数据量为lineorder_hash32的1/64(=1/DoP)，右边的数据为lineorder_hash32的1/32(=1/分区数)。如果继续提高DoP，只有hash join左边的数据量减少，右边的数据量并不会减少; 同时，更多的PX进程处理同一个分区，会提高broadcast分发成本。所以当DoP大于分区数时，并行执行的随着DoP的提高，扩展性并不好。

查看一个蓝色的PX进程，实例1p005进程的执行信息，可以确认hash join的左边为lineorder_hash32的1/64，hash join的右边为lineorder_hash32的1/32。

小结

数据仓库设计时，为了取得最佳的性能，应该使用partition wise join和并行执行的组合。在大表最常用的连接键上，进行hash分区，hash join时使优化器有机会选择partition wise join。Range-hash或者list-hash是常见的分区组合策略，一级分区根据业务特点，利用时间范围或者列表对数据做初步的切分，二级分区使用hash分区。查询时，对一级分区裁剪之后，优化器可以选择partition wise join。

设计partition wise join时，应该尽可能提高hash分区数，控制每个分区的大小。Partition wise join时，每对匹配的分区由一个PX进程处理，如果分区数据太多，可能导致join操作时使用临时空间，影响性能。另一方面，如果分区数太少，当DoP大于分区数时，partition wise join会失效，使用更大的DoP对性能改善非常有限。

---

数据倾斜对不同分发方式的影响

数据倾斜是指某一列上的大部分数据都是少数热门的值(Popular Value)。Hash join时，如果hash join的右边连接键上的数据是倾斜的，数据分发导致某个PX进程需要处理所有热门的数据，拖长sql执行时间，这种情况称为并行执行倾斜。如果优化器选择了hash分发，此时join两边的数据都进行hash分发，数据倾斜会导致执行倾斜。同值记录的hash值也是一样的，会被分发到同一PX进程进行hash join。工作分配不均匀，某个不幸的PX进程需要完成大部分的工作，消耗的db time会比其他PX进程多，SQL执行时间会因此被明显延长。对于replicate或者broadcast分发，则不存在这种执行倾斜的风险，因为hash join右边(一般为大表)的数据不用进行分发，PX进程使用基于数据块地址区间或者基于分区的granule，平均扫描hash join右边的数据，再进行join操作。

为了演示数据倾斜和不同分发的关系，新建两个表，customer_skew包含一条c_custkey=-1 的记录，lineorder_skew 90%的记录，两亿七千万行记录lo_custkey=-1。

sid@SSB> select count(*) from customer_skew where c_custkey = -1;
COUNT(*)
----------
1
sid@SSB> select count(*) from customer_skew;
COUNT(*)
----------
1500000
sid@SSB> select count(*) from lineorder_skew where lo_custkey = -1;
COUNT(*)
----------
270007612
sid@SSB> select count(*) from lineorder_skew;
COUNT(*)
----------
21
300005811

Replicate方式，不受数据倾斜的影响

测试sql如下：

select /*+ monitor parallel(4) */
sum(lo_revenue)
from
lineorder_skew, customer_skew
where
lo_custkey = c_custkey;

SQL执行时间为23秒，db time为1.5m。优化器默认的执行计划选择replicate的方式，只需分配一组PX进程，与broadcast分发的方式类似。每个蓝色的PX进程重复扫描customer，并行扫描lineorder_skew时，是采用基于地址区间的granule为扫描单位，见第7行的’PX BLOCK ITERATOR’。

4个蓝色的PX进程消耗的db time是平均的，对于replicate方式，lineorder_skew的数据倾斜并没有造成4个PX进程的执行倾斜。

当优化器使用replicate方式时，可以通过执行计划中outline中的hint PQ_REPLICATE确认。以下部分dbms_xplan。display_cursor输出没有显示，只显示outline数据。

select * from table(dbms_xplan.display_cursor('77457qc9a324k',0,’outline’));
Plan hash value: 4050055921
...
Outline Data
-------------
/*+
BEGIN_OUTLINE_DATA
22
IGNORE_OPTIM_EMBEDDED_HINTS
OPTIMIZER_FEATURES_ENABLE('12.1.0.2')
DB_VERSION('12.1.0.2')
……
ALL_ROWS
OUTLINE_LEAF(@"SEL$1")
FULL(@"SEL$1" "CUSTOMER_SKEW"@"SEL$1")
FULL(@"SEL$1" "LINEORDER_SKEW"@"SEL$1")
LEADING(@"SEL$1" "CUSTOMER_SKEW"@"SEL$1" "LINEORDER_SKEW"@"SEL$1")
USE_HASH(@"SEL$1" "LINEORDER_SKEW"@"SEL$1")
PQ_DISTRIBUTE(@"SEL$1" "LINEORDER_SKEW"@"SEL$1" BROADCAST NONE)
PQ_REPLICATE(@"SEL$1" "LINEORDER_SKEW"@"SEL$1")
END_OUTLINE_DATA
*/

Hash分发，数据倾斜造成执行倾斜

通过hint使用hash分发，测试sql如下：

select /*+ monitor parallel(4)
leading(customer_skew lineorder_skew)
use_hash(lineorder_skew)
pq_distribute(lineorder_skew hash hash) */
sum(lo_revenue)
from
lineorder_skew, customer_skew
where
lo_custkey = c_custkey;

使用hash分发，SQL执行时间为58秒，dbtime 2.1分钟。对于replicate时sql执行时间23秒，dbtime 1.5分钟。有趣的是，整个sql消耗的db time只增加了37秒，而执行时间确增加了35秒，意味着所增加的dbtime并不是平均到每个PX进程的。如果增加的dbtime平均到每个PX进程，而且并行执行没有倾斜的话，那么sql执行时间应该增加37/4，约9秒，而不是现在的35秒。红色的PX 进程作为生产者，分别对customer_skew和lineorder_skew 完成并行扫描并通过tablequeue0/1，hash分发给蓝色的PX进程。对lineorder_skew的分发，占了45%的db cpu。

实例2的蓝色PX进程p001消耗了57.1秒的dbtime，sql执行时间58秒，这个PX进程在sql执 行过程中一直是活跃状态。可以预见，lineorder_skew所有lo_custkey=-1的数据都分发到这个进程处理。而作为生产者的红色PX进程，负责扫描lineorder_skew并进行分发，它们的工作量是平均的。

大部分时候AAS=2，只有实例2的p001进程不断的从4个生产者接收数据并进行hash join。

从V$PQ_TQSTAT视图我们可以确认，对hash join右边分发时，通过tablequeue1，作为消费者的实例2的P001，接收了两亿七千多万的数据。这就是该PX进程在整个sql执行过程中一直保持活跃的原因。

SELECT
dfo_number, tq_id, server_type, instance, process, num_rows
FROM
V$PQ_TQSTAT
ORDER BY
dfo_number DESC, tq_id, server_type desc, instance, process;
DFO_NUMBER TQ_ID SERVER_TYPE INSTANCE PROCESS NUM_ROWS
---------- ---------- ------------- ---------- ---------- ----------
1 0 Producer 1 P004 375754
1 0 Producer 1 P005 365410
1 0 Producer 2 P003 393069
1 0 Producer 2 P004 365767
1 0 Consumer 1 P002 375709
1 0 Consumer 1 P003 374677
1 0 Consumer 2 P001 374690
1 0 Consumer 2 P002 374924
1 1 Producer 1 P004 75234478
1 1 Producer 1 P005 74926098
1 1 Producer 2 P003 74923913
1 1 Producer 2 P004 74921322
1 1 Consumer 1 P002 7497409
1 1 Consumer 1 P003 7467378
1 1 Consumer 2 P001 277538575
1 1 Consumer 2 P002 7502449
24
1 2 Producer 1 P002 1
1 2 Producer 1 P003 1
1 2 Producer 2 P001 1
1 2 Producer 2 P002 1
1 2 Consumer 1 QC 4
21 rows selected.

12c的sqlmonitor报告作了增强，并行执行倾斜时，包含了消耗最大的PX进程的采样信息。在plan statistics页面，下拉菜单选择’Parallel Server 3(instance 2，p001)’， 从执行计划的第10行，‘PX RECEIVE’，以及Actual Rows列的数据278M，也可以确认实例2的p001进程接收了两亿七千多万数据。

小节

对于实际的应用，处理数据倾斜是一个复杂的主题。比如在倾斜列上使用绑定变量进行过滤，绑定变量窥视(bind peeking)可能造成执行计划不稳定。本节讨论了数据倾斜对不同分发方式的带来影响：

1. 通常，replicate或者broadcast分发不受数据倾斜的影响。
2. 对于hash分发，hash join两边连接键的最热门数据，会被分发到同一PX进程进行join操作，容易造成明显的并行执行倾斜。

---

HASH JOIN BUFFERED，连续hash分发时执行计划中的阻塞点

到目前为止，所有的测试只涉及两个表的连接。如果多于两个表，就需要至少两次的hash join，数据分发次数变多，生产者消费者的角色可能互换，执行计划将不可避免变得复杂。执行路径变长，为了保证并行执行的正常进行，执行计划可能会插入相应的阻塞点，在hash join时，把符合join条件的数据缓存到临时表，暂停数据继续分发。本节我使用一个三表连接的sql来说明连续hash join时，不同分发方式的不同行为。

使用Broadcast分发，没有阻塞点。

测试三个表连接的sql如下，加入part表，使用hint让优化器两次hash join都使用broadcast分发。Replicate SQL查询性能类似。

select /*+ monitor parallel(4)
LEADING(CUSTOMER LINEORDER PART)
USE_HASH(LINEORDER)
USE_HASH(PART)
SWAP_JOIN_INPUTS(PART)
PQ_DISTRIBUTE(PART NONE BROADCAST)
NO_PQ_REPLICATE(PART)
PQ_DISTRIBUTE(LINEORDER BROADCAST NONE)
NO_PQ_REPLICATE(LINEORDER)
25
*/
sum(lo_revenue)
from
lineorder, customer, part
where
lo_custkey = c_custkey
and lo_partkey = p_partkey;

SQL执行时间为42秒，dbtime为2.6分钟。

AAS=(sql db time)/(sql 执行时间)=(2.6*60)/42=3.7，接近4，说明4个PX进程基本一直保持活跃。

执行计划是一颗完美的右深树，这是星型模型查询时执行计划的典型形式。生产者对两个维度进行broadcast分发，消费者接受数据之后准备好两次hash join的build table， 最后扫描事实表，并进行hash join。我们通过跟随table queue顺序的原则，阅读这个执行计划。

1. 红色PX进程作为生产者并行扫描part，通过tablequeue0广播给每个蓝色的消费者PX进程 (第7~9行)。每个蓝色的PX进程接收part的完整数据(第6行)，1.2M行记录，并准备好第5行hash join的build table。
2. 红色PX进程作为生产者并行扫描customer，通过tablequeue1广播broadcast给每个蓝色的 消费者PX进程(第12~14行)。每个蓝色的PX进程接收customer的完整数据(第11行)，1.5M行记录，并准备好第10行hash join的build table。
3. 蓝色的PX进程并行扫描事实表lineorder，对每条符合扫描条件(如果sql语句包含对lineorder的过滤条件)的3亿行记录，进行第10行的hash join，对于每一条通过第10行的 hash join的记录，马上进行第5行的hash join，接着再进行聚合。从sql monitor报告的 Timeline列信息，对lineorder的扫描和两个hash join操作是同时进行的。执行计划中没有阻塞点，数据在执行