Written by Anju Garg

When a SQL statement is executed in parallel, certain operations, such as sorts, aggregations, and joins require data to be redistributed among the parallel server processes executing the statement. The distribution method chosen by the optimizer depends on various factors like the operation to be carried out, the number of parallel server processes involved, and the expected number of rows to be distributed. Until Oracle Database 12c, the optimizer decided the distribution method based on its cardinality estimate which relies on optimizer statistics.

Incorrect estimate by the optimizer can lead to suboptimal distribution method getting chosen which can affect the performance of the parallel query adversely. For example, if optimizer incorrectly estimates few rows to be returned and chooses broadcast method, a small number of parallel processes would have to distribute a large no. of rows to each of the parallel server processes. Similarly if hash distribution is incorrectly chosen, a large no. of parallel processes will be underutilized as they will distribute a small no. of rows to ensure that each parallel server process receives an equal number of rows.

Oracle Database 12c introduces a new adaptive parallel distribution method called Hybrid Hash distribution. With this new method, the optimizer does not decide the final data distribution method until execution time when it will have more information on the actual number of rows involved. The optimizer inserts Statistic Collectors in front of the parallel server processes on the producer side of the operation. If no. of rows buffered is less than a threshold, defined as twice the degree of parallelism (DOP) chosen for the operation, the distribution method chosen is Broadcast for left input and Round-Robin for right input . On the other hand, If no. of rows buffered reach the threshold (2*DOP), Hash Distribution method is chosen for both left and right input. It's important to emphasize that the decision to choose between Hash / Broadcast distribution is made for each execution of the statement so the same statement can employ Broadcast/Hash distribution during various executions depending on the number of rows detected by the Statistics Collector operator. 

Demonstration

To illustrate Hybrid Hash distribution technique, I have created two tables HR.TAB1 and HR.TAB2 having identical structure. Table HR.TAB1 has 2424 rows with values in ID column ranging from 1 to 2424. Table HR.TAB2 has 999 rows with values in ID column ranging from 1 to 999.

I will join the two tables on column ID while employing a DOP of 6. We will observe that the optimizer chooses Hybrid Hash distribution technique to distribute the data among six parallel server processes. When buffered rows are less than the threshold of 12 rows (2*DOP), Broadcast distribution is used, else Hash distribution is employed.

Case-I : Returned Rows = 2 (< 2*DOP)

In this case I have included the condition t1.id in (1,2) in the where clause so that only two rows will be buffered which are much less than the threshold of 12 (= 2*DOP) rows. Table 1 shows the execution plan (edited to remove some columns) for the join between TAB1 and TAB2 that is executed in parallel. One set of parallel server processes (producers) scan the two tables and send the rows to another set of parallel server processes (consumers) that actually perform the join.

The first table accessed in this join is the TAB1 table. A Statistics Collector has been inserted before the parallel server processes scanning the producer TAB1 which buffers the rows coming out of the TAB1 table until the threshold (12 rows) is exceeded or the final row is fetched. The distribution method Hybrid Hash is chosen for both TAB1 and TAB2.

SQL>select /*+ parallel(6) gather_plan_statistics monitor */
    t1.id, t2.txt from hr.tab1 t1, hr.tab2 t2
    where t1.id = t2.id
    and t1.id  in (1,2);

    select * from table(dbms_xplan.display_cursor(format=> 'PARALLEL'));

PLAN_TABLE_OUTPUT
-----------------------------------------------------------------------------
SQL_ID  65m7asn4szcwy, child number 0
-------------------------------------
select /*+ parallel(6) gather_plan_statistics monitor */      t1.id,
t2.txt from hr.tab1 t1, hr.tab2 t2 where t1.id = t2.id  and t1.id  in
(1,2)

Plan hash value: 1086986504
-----------------------------------------------------------------------------
|Id |Operation                 |Name    |Rows|Bytes| TQ  |IN-OUT|PQ Distrib |
-----------------------------------------------------------------------------
| 0
|SELECT STATEMENT              |        |    |     |     |      |           |
| 1 | PX COORDINATOR           |        |    |     |     |      |           |
| 2 |  PX SEND QC (RANDOM)     |:TQ10002|  2 |2018 |Q1,02| P->S |QC (RAND)  |
|*3 |   HASH JOIN BUFFERED     |        |  2 |2018 |Q1,02| PCWP |           |
| 4 |    PX RECEIVE            |        |  2 |   8 |Q1,02| PCWP |           |
| 5 |     PX SEND HYBRID HASH  |:TQ10000|  2 |   8 |Q1,00| P->P |HYBRID HASH|
| 6 |      STATISTICS COLLECTOR|        |    |     |Q1,00| PCWC |           |
| 7 |       PX BLOCK ITERATOR  |        |  2 |   8 |Q1,00| PCWC |           |
|*8 |        TABLE ACCESS FULL |TAB1    |  2 |   8 |Q1,00| PCWP |           |
| 9 |    PX RECEIVE            |        |  2 |2010 |Q1,02| PCWP |           |
|10 |     PX SEND HYBRID HASH  |:TQ10001|  2 |2010 |Q1,01| P->P |HYBRID HASH|
|11 |      PX BLOCK ITERATOR   |        |  2 |2010 |Q1,01| PCWC |           |
|*12|       TABLE ACCESS FULL  |TAB2    |  2 |2010 |Q1,01| PCWP |           |
-----------------------------------------------------------------------------

Predicate Information (identified by operation id):
---------------------------------------------------

   3 - access("T1"."ID"="T2"."ID")
   8 - access(:Z>=:Z AND :Z<=:Z)
       filter(("T1"."ID"=1 OR "T1"."ID"=2))
  12 - access(:Z>=:Z AND :Z<=:Z)
       filter(("T2"."ID"=1 OR "T2"."ID"=2))

Note
-----
   - Degree of Parallelism is 6 because of hint

Table 1

 

To find out which distribution method was finally chosen at runtime, I will query V$PQ_TQSTAT.
 

In this example, the number of rows returned from the scan of the TAB1 table is 2. Since the threshold of 12 (2*DOP = 2*6 = 12) has not been reached, the 2 rows returned from the TAB1 table are BROADCAST to each of the 6 parallel server processes responsible for completing the join, resulting in 12 rows (2 *6) being processed by the distribution step in the plan (Table 2). Since a Broadcast distribution was used for rows coming from the TAB1 table, the rows coming from the TAB2 table are distributed via Round-Robin method. This means, one row from the TAB2 table will be sent to each of the 6 parallel server processes in turn, until all of the rows have been distributed.

SQL>select tq_id, server_type, process, num_rows
         from v$pq_tqstat
         order by tq_id, server_type, process;

     TQ_ID SERVER_TYP PROCES   NUM_ROWS
---------- ---------- ------ ----------
         0 Consumer   P000            2 |
         0 Consumer   P001            2 |
         0 Consumer   P002            2 |--> 2 rows of TAB1 received by each
         0 Consumer   P003            2 |    consumer(broadcast)
         0 Consumer   P004            2 |
         0 Consumer   P005            2 |

         0 Producer   P006              |
         0 Producer   P007           12 |    
         0 Producer   P008            0 |--> 12 rows of TAB1
         0 Producer   P009            0 |    (2 rows*6 consumer processes)   
         0 Producer   P010            0 |    sent to consumers by one
         0 Producer   P011            0 |    producer process

         1 Consumer   P000            1 |
         1 Consumer   P001            1 |
         1 Consumer   P002            0 |--> 2 rows of TAB2 distributed to
         1 Consumer   P003            0 |    consumers in round robin method
         1 Consumer   P004            0 |
         1 Consumer   P005            0 |

         1 Producer   P006            0 |
         1 Producer   P007            0 |
         1 Producer   P008            0 |--> 2 rows of TAB2 read by one
         1 Producer   P009            2 |    producer
         1 Producer   P010            0 |
         1 Producer   P011            0

         2 Consumer   QC              2

         2 Producer   P000            1 |
         2 Producer   P001            1 |
         2 Producer   P002            0 |--> 2 rows of TAB1 and TAB2 joined
         2 Producer   P003            0 |
         2 Producer   P004            0 |
         2 Producer   P005            0 |

Table 2

 

Case-II : Returned rows > Threshold (2*DOP)

To demonstrate Hash distribution,   I have joined tables TAB1 and TAB2 on column ID using a DOP of 6, so that 2424 rows from TAB1 and 999 rows from TAB2 will be returned, which are much higher than the threshold of 12 (2*DOP = 2*6) rows.

It can be seen in Table 3   that Statistics Collector has been inserted in front of the parallel server processes scanning TAB1 and distribution method chosen is Hybrid Hash for both TAB1 and TAB2.

select /*+ parallel(6) gather_plan_statistics monitor */
     t1.id, t2.txt from hr.tab1 t1, hr.tab2 t2
where t1.id = t2.id;

select * from table(dbms_xplan.display_cursor(format=> 'PARALLEL'));

PLAN_TABLE_OUTPUT
-----------------------------------------------------------------------------
SQL_ID  4zzum7cyxt19k, child number 0
-------------------------------------
select /*+ parallel(6) gather_plan_statistics monitor */      t1.id,
t2.txt from hr.tab1 t1, hr.tab2 t2 where t1.id = t2.id

Plan hash value: 1086986504

-----------------------------------------------------------------------------
| Id|Operation                  | Name  |Rows|Bytes| TQ  |IN-OUT| PQ Distrib|
-----------------------------------------------------------------------------
|  0|SELECT STATEMENT           |        |    |    |     |      |           |           
|  1|  PX COORDINATOR           |        |    |    |     |      |           |         
|  2|   PX SEND QC (RANDOM)     |:TQ10002| 999|984K|Q1,02| P->S | QC (RAND) |
| *3|    HASH JOIN BUFFERED     |        | 999|984K|Q1,02| PCWP |           |
|  4|     PX RECEIVE            |        |2424|9696|Q1,02| PCWP |           |
|  5|      PX SEND HYBRID HASH  |:TQ10000|2424|9696|Q1,00| P->P |HYBRID HASH|
|  6|       STATISTICS COLLECTOR|        |    |    |Q1,00| PCWC |           |
|  7|        PX BLOCK ITERATOR  |        |2424|9696|Q1,00| PCWC |           |
| *8|         TABLE ACCESS FULL |TAB1    |2424|9696|Q1,00| PCWP |           |
|  9|     PX RECEIVE            |        | 999|980K|Q1,02| PCWP |           |
| 10|      PX SEND HYBRID HASH  |:TQ10001| 999|980K|Q1,01| P->P |HYBRID HASH|
| 11|       PX BLOCK ITERATOR   |        | 999|980K|Q1,01| PCWC |           |
|*12|        TABLE ACCESS FULL  |TAB2    | 999|980K|Q1,01| PCWP |           |
-----------------------------------------------------------------------------
Predicate Information (identified by operation id):
---------------------------------------------------

   3 - access("T1"."ID"="T2"."ID")
   8 - access(:Z>=:Z AND :Z<=:Z)
  12 - access(:Z>=:Z AND :Z<=:Z)

Note
-----
   - Degree of Parallelism is 6 because of hint

Table 3

 

Again to find out the distribution method finally chosen at runtime, I will query V$PQ_TQSTAT. While scanning TAB1 which returned 2424 rows, since the threshold of 12 was reached, the rows returned from TAB1 are distributed by HASH among the 6 parallel server processes responsible for completing the join, resulting in 2424 rows being processed by the distribution step in the plan (Table 4). Since a HASH distribution has been used for rows coming from the TAB1 table, the rows coming from the TAB2 table are also distributed via HASH.

select tq_id, server_type, process, num_rows
from v$pq_tqstat
order by tq_id, server_type, process;

     TQ_ID SERVER_TYP PROCES   NUM_ROWS
---------- ---------- ------ ----------
         0 Consumer   P000          398 |
         0 Consumer   P001          416 |
         0 Consumer   P002          411 |-->(2424 rows of TAB1 distributed
         0 Consumer   P003          423 |    using hash algorithm across 6
         0 Consumer   P004          371 |    consumers )
         0 Consumer   P005          405 |

         0 Producer   P006          413 |
         0 Producer   P007          401 |
         0 Producer   P008          392 |--> (2424 rows read from TAB1 by 6  
         0 Producer   P009          385 |     producers )
         0 Producer   P010          399 |
         0 Producer   P011          434 |

         1 Consumer   P000          165 |
         1 Consumer   P001          166 |
         1 Consumer   P002          174 |--> (999 rows of TAB2 distributed
         1 Consumer   P003          178 |     using hash algorithm across 6
         1 Consumer   P004          146 |     consumers )
         1 Consumer   P005          170 |

         1 Producer   P006          161 |
         1 Producer   P007          182 |
         1 Producer   P008          147 |--> (999 rows read from TAB2 by 6  
         1 Producer   P009          175 |     producers )
         1 Producer   P010          161 |
         1 Producer   P011          173 |

         2 Consumer   QC            999

         2 Producer   P000          165 |
         2 Producer   P001          166 |
         2 Producer   P002          174 |--> (999 rows of TAB1 and TAB2
         2 Producer   P003          178 |     joined)
         2 Producer   P004          146 |
         2 Producer   P005          170 |

Table 4 

 

It is worth mentioning here that currently Hybrid Hash distribution method is applicable only for those parallel queries for which the optimizer decides to employ hash distribution at parse time i.e. it is designed for cases where the left side cardinality is overestimated. In such cases, it replaces the earlier Hash-Hash distribution method. In case of queries where the optimizer underestimates cardinality on the left side and decides to employ Broadcast distribution, still continue the same way and do not get Hybrid Hash.

 

References:

 

https://blogs.oracle.com/datawarehousing/entry/finding_the_distribution_method_in

http://www.oracle.com/technetwork/database/bi-datawarehousing/twp-optimizer-with-oracledb-12c-1963236.pdf

http://kerryosborne.oracle-guy.com/papers/12c_Adaptive_Optimization.pdf