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 the t14 table 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 subnode on the other side.
In the following example, the query joins two partitioned tables. The execution plan repartitions the data of the s15 table by using the partitioning method of the t13 table. The input of Operator 4 is the scanning result of the s15 table. For every row of the s15 table, the operator determines the destination node based on the partitioning of the t13 table 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 repartitioning is performed. For example, #keys=1, [s15.c1] indicates that repartitioning is performed based on the c1 column. |
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, 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 joins 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 hash repartitioning is performed. For example, #keys=1, [s14.c2] indicates that hash repartitioning is performed based on the c2 column. |
EXCH-IN/OUT (BC2HOST)
The EXCH-IN/OUT (BC2HOST) operators repartition input data by using the BC2HOST method and broadcast the data to other threads.
In the execution plan shown in the following example, operators 3 and 4 are EXCHANGE operators that use BC2HOST repartitioning. The operators broadcast the data of the t11 table to every thread and try to join each partition of the s13 table with the data broadcast from the t11 table.
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 operator have the same meaning as those of the EXCH-IN/OUT operators.