EXCHANGE operators exchange data between threads.
EXCHANGE operators are useful in distributed scenarios and usually appear in pairs, with an EXCHANGE OUT operator at the source side and an EXCHANGE IN operator at the destination side.
EXCH-IN/OUT
Short for EXCHANGE IN/EXCHANGE OUT, the EXCH-IN/OUT operators aggregate data from multiple partitions and send them to the leader node involved in the query.
In the following example, the query accesses five partitions: p0, p1, p2, p3, and p4. Operator 1 receives the output of Operator 2 and sends the data out. Operator 0 receives the output of Operator 1 from multiple partitions, aggregates them, and generates the result.
obclient> CREATE TABLE t15 (c1 INT, c2 INT) PARTITION BY HASH(c1) PARTITIONS 5;
Query OK, 0 rows affected
obclient> EXPLAIN SELECT * FROM t15;
+---------------------------------------------------------------------------+
| Query Plan |
+---------------------------------------------------------------------------+
| ============================================================= |
| |ID|OPERATOR |NAME |EST.ROWS|EST.TIME(us)| |
| ------------------------------------------------------------- |
| |0 |PX COORDINATOR | |1 |20 | |
| |1 |└─EXCHANGE OUT DISTR |:EX10000|1 |20 | |
| |2 | └─PX PARTITION ITERATOR| |1 |19 | |
| |3 | └─TABLE FULL SCAN |t15 |1 |19 | |
| ============================================================= |
| Outputs & filters: |
| ------------------------------------- |
| 0 - output([INTERNAL_FUNCTION(t15.c1, t15.c2)]), filter(nil), rowset=16 |
| 1 - output([INTERNAL_FUNCTION(t15.c1, t15.c2)]), filter(nil), rowset=16 |
| dop=1 |
| 2 - output([t15.c1], [t15.c2]), filter(nil), rowset=16 |
| force partition granule |
| 3 - output([t15.c1], [t15.c2]), filter(nil), rowset=16 |
| access([t15.c1], [t15.c2]), partitions(p[0-4]) |
| is_index_back=false, is_global_index=false, |
| range_key([t15.__pk_increment]), range(MIN ; MAX)always true |
+---------------------------------------------------------------------------+
In the preceding example, the Outputs & filters section shows in detail the output information of the EXCH-IN/OUT operators.
| Field | Description |
|---|---|
| PX COORDINATOR | A special type of EXCHANGE IN operator, responsible for not only pulling back remote data, but also scheduling the execution of sub-plans. |
| Output | The output expressions of the operator. |
| filter | The filter conditions of the operator. In this example, filter is set to nil because no filter condition is configured for the EXCH-IN/OUT operators. |
EXCH-IN/OUT (REMOTE)
The EXCH-IN/OUT (REMOTE) operators pull remote data in a single partition back to the local server.
As shown in the following example, a non-partitioned table is created on Server A, and a query is executed on Server B to read data from that table. In this case, the operators 0 and 1 are assigned to the execution plan to fetch remote data. Operator 1 is executed on Server A to read the data from table t14 and send the data out. Operator 0 is executed on Server B to receive the output of Operator 1.
obclient> CREATE TABLE t14 (c1 INT, c2 INT);
Query OK, 0 rows affected
obclient> EXPLAIN SELECT * FROM t14;
+--------------------------------------------------------------------+
| Query Plan |
+--------------------------------------------------------------------+
| ===================================================== |
| |ID|OPERATOR |NAME|EST.ROWS|EST.TIME(us)| |
| ----------------------------------------------------- |
| |0 |EXCHANGE IN REMOTE | |1 |5 | |
| |1 |└─EXCHANGE OUT REMOTE| |1 |4 | |
| |2 | └─TABLE FULL SCAN |t14 |1 |3 | |
| ===================================================== |
| Outputs & filters: |
| ------------------------------------- |
| 0 - output([t14.C1], [t14.C2]), filter(nil) |
| 1 - output([t14.C1], [t14.C2]), filter(nil) |
| 2 - output([t14.C1], [t14.C2]), filter(nil), rowset=16 |
| access([t14.C1], [t14.C2]), partitions(p0) |
| is_index_back=false, is_global_index=false, |
| range_key([t14.__pk_increment]), range(MIN ; MAX)always true |
+--------------------------------------------------------------------+
In the preceding example, the Outputs & filters section shows in detail the output information of the EXCH-IN/OUT (REMOTE) operators. The fields of the operator have the same meaning as those of the EXCH-IN/OUT operators.
EXCH-IN/OUT (PKEY)
The EXCH-IN/OUT (PKEY) operators repartition data. They are generally used with a binary operator pair to repartition the data of one subnode by following the partitioning method of the other subnode.
In the following example, the query is to join two partitioned tables. The execution plan repartitions the data of table s15 by following the partitioning method of table t13. The input of Operator 4 is the scanning result of table s15. For every row of table s15, the operator determines its target node based on the partitioning of table t13 and the join condition of the query.
In addition, Operator 3 is EXCHANGE IN MERGE SORT DISTR, a special EXCHANGE IN operator that performs merge and sort operations when aggregating data from multiple partitions. In this execution plan, data received by Operator 3 from all partitions is sorted by c1, so the operator merges and sorts the received data to ensure that the output is also sorted by c1.
obclient> CREATE TABLE t13 (c1 INT, c2 INT) PARTITION BY HASH(c1) PARTITIONS 5;
Query OK, 0 rows affected
obclient> CREATE TABLE s15 (c1 INT PRIMARY KEY, c2 INT) PARTITION BY HASH(c1) PARTITIONS 4;
Query OK, 0 rows affected
obclient> EXPLAIN SELECT * FROM s15, t13 WHERE s15.c1 = t13.c1;
+-------------------------------------------------------------------------------------------+
| Query Plan |
+-------------------------------------------------------------------------------------------+
| ===================================================================== |
| |ID|OPERATOR |NAME |EST.ROWS|EST.TIME(us)| |
| --------------------------------------------------------------------- |
| |0 |PX COORDINATOR | |1 |38 | |
| |1 |└─EXCHANGE OUT DISTR |:EX10001|1 |37 | |
| |2 | └─HASH JOIN | |1 |36 | |
| |3 | ├─EXCHANGE IN DISTR | |1 |17 | |
| |4 | │ └─EXCHANGE OUT DISTR (PKEY)|:EX10000|1 |16 | |
| |5 | │ └─PX PARTITION ITERATOR | |1 |16 | |
| |6 | │ └─TABLE FULL SCAN |s15 |1 |16 | |
| |7 | └─PX PARTITION ITERATOR | |1 |19 | |
| |8 | └─TABLE FULL SCAN |t13 |1 |19 | |
| ===================================================================== |
| Outputs & filters: |
| ------------------------------------- |
| 0 - output([INTERNAL_FUNCTION(s15.c1, s15.c2, t13.c1, t13.c2)]), filter(nil), rowset=16 |
| 1 - output([INTERNAL_FUNCTION(s15.c1, s15.c2, t13.c1, t13.c2)]), filter(nil), rowset=16 |
| dop=1 |
| 2 - output([s15.c1], [t13.c1], [s15.c2], [t13.c2]), filter(nil), rowset=16 |
| equal_conds([s15.c1 = t13.c1]), other_conds(nil) |
| 3 - output([s15.c1], [s15.c2]), filter(nil), rowset=16 |
| 4 - output([s15.c1], [s15.c2]), filter(nil), rowset=16 |
| (#keys=1, [s15.c1]), dop=1 |
| 5 - output([s15.c1], [s15.c2]), filter(nil), rowset=16 |
| force partition granule |
| 6 - output([s15.c1], [s15.c2]), filter(nil), rowset=16 |
| access([s15.c1], [s15.c2]), partitions(p[0-3]) |
| is_index_back=false, is_global_index=false, |
| range_key([s15.c1]), range(MIN ; MAX)always true |
| 7 - output([t13.c1], [t13.c2]), filter(nil), rowset=16 |
| affinitize, force partition granule |
| 8 - output([t13.c1], [t13.c2]), filter(nil), rowset=16 |
| access([t13.c1], [t13.c2]), partitions(p[0-4]) |
| is_index_back=false, is_global_index=false, |
| range_key([t13.__pk_increment]), range(MIN ; MAX)always true |
+-------------------------------------------------------------------------------------------+
In the preceding example, the Outputs & filters section shows in detail the output information of the EXCH-IN/OUT (PKEY) operators.
| Field | Description |
|---|---|
| PX COORDINATOR | A special type of EXCHANGE IN operator, responsible for not only pulling back remote data, but also scheduling the execution of sub-plans. |
| Output | The output expressions of the operator. |
| filter | The filter conditions of the operator. In this example, filter is set to nil because no filter condition is configured for the EXCH-IN/OUT (PKEY) operators. |
| pkey | The column based on which the repartitioning is performed. For example, #keys=1, [s15.c1] indicates that the repartitioning is performed based on column c1. |
EXCH-IN/OUT (HASH)
The EXCH-IN/OUT (HASH) operators repartition data by using a set of hash functions.
In the execution plan shown in the following example, operators 3-5 and 7-8 are two sets of EXCHANGE operators that use hash functions for repartitioning. These two sets of operators partition the data of tables t12 and s14 into multiple parts based on a new set of hash functions. In the example, the columns t12.c2 and s14.c2 are taken for the hash operation to ensure that rows with the same value in column c2 are grouped in the same part. Based on the repartitioned data, Operator 2 HASH JOIN merges all parts based on the condition t12.c2= s14.c2.
Additionally, the plan shows dop = 2 because the query is executed with a degree of parallelism (DOP) of 2.
obclient> CREATE TABLE t12 (c1 INT, c2 INT) PARTITION BY HASH(c1) PARTITIONS 4;
Query OK, 0 rows affected
obclient> CREATE TABLE s14 (c1 INT, c2 INT) PARTITION BY HASH(c1) PARTITIONS 4;
Query OK, 0 rows affected
obclient> EXPLAIN SELECT /*+PARALLEL(2) LEADING(t12 s14) USE_HASH(s14) PQ_DISTRIBUTE(s14 HASH HASH)*/ * FROM t12, s14 WHERE t12.c2 = s14.c2;
+-------------------------------------------------------------------------------------------+
| Query Plan |
+-------------------------------------------------------------------------------------------+
| ===================================================================== |
| |ID|OPERATOR |NAME |EST.ROWS|EST.TIME(us)| |
| --------------------------------------------------------------------- |
| |0 |PX COORDINATOR | |1 |18 | |
| |1 |└─EXCHANGE OUT DISTR |:EX10002|1 |17 | |
| |2 | └─HASH JOIN | |1 |17 | |
| |3 | ├─EXCHANGE IN DISTR | |1 |9 | |
| |4 | │ └─EXCHANGE OUT DISTR (HASH)|:EX10000|1 |8 | |
| |5 | │ └─PX BLOCK ITERATOR | |1 |8 | |
| |6 | │ └─TABLE FULL SCAN |t12 |1 |8 | |
| |7 | └─EXCHANGE IN DISTR | |1 |9 | |
| |8 | └─EXCHANGE OUT DISTR (HASH)|:EX10001|1 |8 | |
| |9 | └─PX BLOCK ITERATOR | |1 |8 | |
| |10| └─TABLE FULL SCAN |s14 |1 |8 | |
| ===================================================================== |
| Outputs & filters: |
| ------------------------------------- |
| 0 - output([INTERNAL_FUNCTION(t12.c1, t12.c2, s14.c1, s14.c2)]), filter(nil), rowset=16 |
| 1 - output([INTERNAL_FUNCTION(t12.c1, t12.c2, s14.c1, s14.c2)]), filter(nil), rowset=16 |
| dop=2 |
| 2 - output([t12.c2], [s14.c2], [t12.c1], [s14.c1]), filter(nil), rowset=16 |
| equal_conds([t12.c2 = s14.c2]), other_conds(nil) |
| 3 - output([t12.c2], [t12.c1]), filter(nil), rowset=16 |
| 4 - output([t12.c2], [t12.c1]), filter(nil), rowset=16 |
| (#keys=1, [t12.c2]), dop=2 |
| 5 - output([t12.c1], [t12.c2]), filter(nil), rowset=16 |
| 6 - output([t12.c1], [t12.c2]), filter(nil), rowset=16 |
| access([t12.c1], [t12.c2]), partitions(p[0-3]) |
| is_index_back=false, is_global_index=false, |
| range_key([t12.__pk_increment]), range(MIN ; MAX)always true |
| 7 - output([s14.c2], [s14.c1]), filter(nil), rowset=16 |
| 8 - output([s14.c2], [s14.c1]), filter(nil), rowset=16 |
| (#keys=1, [s14.c2]), dop=2 |
| 9 - output([s14.c1], [s14.c2]), filter(nil), rowset=16 |
| 10 - output([s14.c1], [s14.c2]), filter(nil), rowset=16 |
| access([s14.c1], [s14.c2]), partitions(p[0-3]) |
| is_index_back=false, is_global_index=false, |
| range_key([s14.__pk_increment]), range(MIN ; MAX)always true |
+-------------------------------------------------------------------------------------------+
The PX PARTITION ITERATOR operator iterates data at the partition level. For more information, see GI.
In the preceding example, the Outputs & filters section shows in detail the output information of the EXCH-IN/OUT (HASH) operators.
| Field | Description |
|---|---|
| PX COORDINATOR | A special type of EXCHANGE IN operator, responsible for not only pulling back remote data, but also scheduling the execution of sub-plans. |
| Output | The output expressions of the operator. |
| filter | The filter conditions of the operator. In this example, filter is set to nil because no filter condition is configured for the EXCH-IN/OUT (HASH) operators. |
| pkey | The column based on which the hash repartitioning is performed. For example, #keys=1, [s14.c2] indicates that the hash repartitioning is performed based on column c2. |
EXCH-IN/OUT (BC2HOST)
The EXCH-IN/OUT (BC2HOST) operators repartition input data by using the BC2HOST method and broadcasts the data to other threads.
In the execution plan shown in the following example, the operators 3 and 4 are EXCHANGE operators that use BC2HOST repartitioning. They broadcast the data of table t11 to every thread and try to join each partition of table s13 with the broadcast data of table t11.
obclient> CREATE TABLE t11 (c1 INT, c2 INT) PARTITION BY HASH(c1) PARTITIONS 4;
Query OK, 0 rows affected
obclient> CREATE TABLE s13 (c1 INT, c2 INT) PARTITION BY HASH(c1) PARTITIONS 4;
Query OK, 0 rows affected
obclient> INSERT INTO s VALUES (1, 1), (2, 2), (3, 3), (4, 4);
Query OK, 1 rows affected
obclient> EXPALIN SELECT /*+PARALLEL(2) */ * FROM t11, s13 WHERE t11.c2 = s13.c2;
+-------------------------------------------------------------------------------------------+
| Query Plan |
+-------------------------------------------------------------------------------------------+
| ======================================================================== |
| |ID|OPERATOR |NAME |EST.ROWS|EST.TIME(us)| |
| ------------------------------------------------------------------------ |
| |0 |PX COORDINATOR | |1 |18 | |
| |1 |└─EXCHANGE OUT DISTR |:EX10001|1 |17 | |
| |2 | └─SHARED HASH JOIN | |1 |17 | |
| |3 | ├─EXCHANGE IN DISTR | |1 |9 | |
| |4 | │ └─EXCHANGE OUT DISTR (BC2HOST)|:EX10000|1 |8 | |
| |5 | │ └─PX BLOCK ITERATOR | |1 |8 | |
| |6 | │ └─TABLE FULL SCAN |t11 |1 |8 | |
| |7 | └─PX BLOCK ITERATOR | |4 |8 | |
| |8 | └─TABLE FULL SCAN |s13 |4 |8 | |
| ======================================================================== |
| Outputs & filters: |
| ------------------------------------- |
| 0 - output([INTERNAL_FUNCTION(t11.c1, t11.c2, s13.c1, s13.c2)]), filter(nil), rowset=16 |
| 1 - output([INTERNAL_FUNCTION(t11.c1, t11.c2, s13.c1, s13.c2)]), filter(nil), rowset=16 |
| dop=2 |
| 2 - output([t11.c2], [s13.c2], [t11.c1], [s13.c1]), filter(nil), rowset=16 |
| equal_conds([t11.c2 = s13.c2]), other_conds(nil) |
| 3 - output([t11.c2], [t11.c1]), filter(nil), rowset=16 |
| 4 - output([t11.c2], [t11.c1]), filter(nil), rowset=16 |
| dop=2 |
| 5 - output([t11.c1], [t11.c2]), filter(nil), rowset=16 |
| 6 - output([t11.c1], [t11.c2]), filter(nil), rowset=16 |
| access([t11.c1], [t11.c2]), partitions(p[0-3]) |
| is_index_back=false, is_global_index=false, |
| range_key([t11.__pk_increment]), range(MIN ; MAX)always true |
| 7 - output([s13.c1], [s13.c2]), filter(nil), rowset=16 |
| 8 - output([s13.c1], [s13.c2]), filter(nil), rowset=16 |
| access([s13.c1], [s13.c2]), partitions(p[0-3]) |
| is_index_back=false, is_global_index=false, |
| range_key([s13.__pk_increment]), range(MIN ; MAX)always true |
+-------------------------------------------------------------------------------------------+
In the preceding example, the Outputs & filters section shows in detail the output information of the EXCH-IN/OUT (BC2HOST) operators. The fields of the operators have the same meaning as those of the EXCH-IN/OUT operators.