EXCHANGE · k8s原理與實踐

[TOC] # EXCHANGE EXCHANGE 算子用于線程間進行數據交互的算子。 EXCHANGE 算子適用于在分布式場景，一般都是成對出現的，數據源端有一個 OUT 算子，目的端會有一個 IN 算子。 ## EXCH-IN/OUT EXCH-IN/OUT 即 EXCHANGE IN/ EXCHANGE OUT 用于將多個分區上的數據匯聚到一起，發送到查詢所在的主節點上。如下例所示，下面的查詢中訪問了 5 個分區的數據（p0-p4），其中 1 號算子接受 2 號算子產生的輸出，并將數據傳出；0 號算子接收多個分區上 1 號算子產生的輸出，并將結果匯總輸出。 ~~~ obclient>CREATE TABLE t (c1 INT, c2 INT) PARTITION BY HASH(c1) PARTITIONS 5; Query OK, 0 rows affected (0.12 sec) obclient>EXPLAIN SELECT * FROM t\G; *************************** 1. row *************************** Query Plan: ============================================== |ID|OPERATOR |NAME|EST. ROWS|COST | ---------------------------------------------- |0 |EXCHANGE IN DISTR | |500000 |545109| |1 | EXCHANGE OUT DISTR| |500000 |320292| |2 | TABLE SCAN |T |500000 |320292| ============================================== Outputs & filters: ------------------------------------- 0 - output([T.C1], [T.C2]), filter(nil) 1 - output([T.C1], [T.C2]), filter(nil) 2 - output([T.C1], [T.C2]), filter(nil), access([T.C1], [T.C2]), partitions(p[0-4]) ~~~ 上述示例的執行計劃展示中的 outputs & filters 詳細列出了 EXCH-IN/OUT 算子的輸出信息如下： <table data-tag="table" id="table-r2s-ra3-31b" class="table"><colgroup span="1" width="200" data-tag="col" id="col-bc7-0ww-mlj" colwidth="1*" colnum="1" colname="col1" style="width:50%" class="col"></colgroup><colgroup span="1" width="540" data-tag="col" id="col-x8o-w8s-k5h" colwidth="1*" colnum="2" colname="col2" style="width:50%" class="col"></colgroup><thead id="thead-z6r-7qu-4hh" class="thead"><tr id="tr-8l9-pe4-z5n"><th id="td-cop-k71-dj4"><p id="p-xl0-7k4-559">信息名稱</p></th><th id="td-72i-wtg-goq"><p id="p-zgq-njb-lz6">含義</p></th></tr></thead><tbody data-tag="tbody" id="tbody-i9z-tvy-nx2" class="tbody"><tr data-tag="tr" id="tr-k0d-10d-ayp" class="tr"><td data-tag="td" id="td-7q9-w4g-ijl" class="td"><p id="p-yc9-poh-myx">output</p></td><td data-tag="td" id="td-0gg-lgw-mye" class="td"><p id="p-1o5-5n7-jrz">該算子輸出的表達式。</p></td></tr><tr data-tag="tr" id="tr-0ke-ug0-v24" class="tr"><td data-tag="td" id="td-n3c-1br-0j2" class="td"><p id="p-iar-ov7-8xx">filter</p></td><td data-tag="td" id="td-g4q-u97-dke" class="td"><p id="p-ee3-rbc-gex">該算子上的過濾條件。</p><p id="p-451-gvd-kef">由于示例中 EXCH-IN/OUT 算子沒有設置 filter，所以為 nil。</p></td></tr></tbody></table> ## EXCH-IN/OUT (REMOTE) EXCH-IN/OUT (REMOTE) 算子用于將遠程的數據（單個分區的數據）拉回本地。如下例所示，在 A 機器上創建了一張非分區表，在 B 機器上執行查詢，讀取該表的數據。此時，由于待讀取的數據在遠程，執行計劃中分配了 0 號算子和 1 號算子來拉取遠程的數據。其中，1 號算子在 A 機器上執行，讀取 t 表的數據，并將數據傳出；0 號算子在 B 機器上執行，接收 1 號算子產生的輸出。 ~~~ obclient>CREATE TABLE t (c1 INT, c2 INT); Query OK, 0 rows affected (0.12 sec) obclient>EXPLAIN SELECT * FROM t\G; *************************** 1. row *************************** Query Plan: =============================================== |ID|OPERATOR |NAME|EST. ROWS|COST | ----------------------------------------------- |0 |EXCHANGE IN REMOTE | |100000 |109029| |1 | EXCHANGE OUT REMOTE| |100000 |64066 | |2 | TABLE SCAN |T |100000 |64066 | =============================================== Outputs & filters: ------------------------------------- 0 - output([T.C1], [T.C2]), filter(nil) 1 - output([T.C1], [T.C2]), filter(nil) 2 - output([T.C1], [T.C2]), filter(nil), access([T.C1], [T.C2]), partitions(p0) ~~~ 上述示例的執行計劃展示中的 outputs & filters 詳細列出了 EXCH-IN/OUT (REMOTE) 算子的輸出信息，字段的含義與 EXCH-IN/OUT 算子相同。 ## EXCH-IN/OUT (PKEY) EXCH-IN/OUT (PKEY) 算子用于數據重分區。它通常用于二元算子中，將一側孩子節點的數據按照另外一些孩子節點的分區方式進行重分區。如下示例中，該查詢是對兩個分區表的數據進行聯接，執行計劃將 s 表的數據按照 t 的分區方式進行重分區，4 號算子的輸入是 s 表掃描的結果，對于 s 表的每一行，該算子會根據 t 表的數據分區，以及根據查詢的聯接條件，確定一行數據應該發送到哪個節點進行。此外，可以看到 3 號算子是一個 EXCHANGE IN MERGE SORT DISTR，它是一個特殊的 EXCHANGE IN 算子，它用于在匯總多個分區的數據時，會進行一定的歸并排序，在這個執行計劃中，3 號算子接收到的每個分區的數據都是按照 c1 有序排列的，它會對每個接收到的數據進行歸并排序，從而保證結果輸出結果也是按照 c1 有序排列的。 ~~~ obclient>CREATE TABLE t (c1 INT, c2 INT) PARTITION BY HASH(c1) PARTITIONS 5; Query OK, 0 rows affected (0.12 sec) obclient>CREATE TABLE s (c1 INT PRIMARY KEY, c2 INT) PARTITION BY HASH(c1) PARTITIONS 4; Query OK, 0 rows affected (0.12 sec) obclient>EXPLAIN SELECT * FROM s, t WHERE s.c1 = t.c1\G; *************************** 1. row *************************** Query Plan: =============================================================== |ID|OPERATOR |NAME|EST. ROWS |COST | --------------------------------------------------------------- |0 |EXCHANGE IN DISTR | |1960200000|3090308367| |1 | EXCHANGE OUT DISTR | |1960200000|1327558071| |2 | MERGE JOIN | |1960200000|1327558071| |3 | EXCHANGE IN MERGE SORT DISTR| |400000 |436080 | |4 | EXCHANGE OUT DISTR (PKEY) | |400000 |256226 | |5 | TABLE SCAN |S |400000 |256226 | |6 | TABLE SCAN |T |500000 |320292 | =============================================================== Outputs & filters: ------------------------------------- 0 - output([S.C1], [S.C2], [T.C1], [T.C2]), filter(nil) 1 - output([S.C1], [S.C2], [T.C1], [T.C2]), filter(nil) 2 - output([S.C1], [S.C2], [T.C1], [T.C2]), filter(nil), equal_conds([S.C1 = T.C1]), other_conds(nil) 3 - output([S.C1], [S.C2]), filter(nil), sort_keys([S.C1, ASC]) 4 - (#keys=1, [S.C1]), output([S.C1], [S.C2]), filter(nil) 5 - output([S.C1], [S.C2]), filter(nil), access([S.C1], [S.C2]), partitions(p[0-3]) 6 - output([T.C1], [T.C2]), filter(nil), access([T.C1], [T.C2]), partitions(p[0-4]) ~~~ 上述示例的執行計劃展示中的 outputs & filters 詳細列出了 EXCH-IN/OUT (PKEY) 算子的輸出信息如下： <table data-tag="table" id="table-y58-uwx-zz9" class="table"><colgroup span="1" width="200" data-tag="col" id="col-i0l-jbv-xo7" colwidth="1*" colnum="1" colname="col1" style="width:50%" class="col"></colgroup><colgroup span="1" width="540" data-tag="col" id="col-2kd-y1l-rqi" colwidth="1*" colnum="2" colname="col2" style="width:50%" class="col"></colgroup><thead id="thead-im2-pwz-alv" class="thead"><tr id="tr-fab-ydv-ny8"><th id="td-qhn-ula-51y"><p id="p-ko7-ult-hhs">信息名稱</p></th><th id="td-myw-gzs-dad"><p id="p-ajb-upm-pbj">含義</p></th></tr></thead><tbody data-tag="tbody" id="tbody-adb-yw3-qz1" class="tbody"><tr data-tag="tr" id="tr-jci-sxr-jlh" class="tr"><td data-tag="td" id="td-510-kj0-5rn" class="td"><p id="p-6tx-3tq-ehj">output</p></td><td data-tag="td" id="td-1c4-fi3-96y" class="td"><p id="p-iol-zui-4al">該算子輸出的表達式。</p></td></tr><tr data-tag="tr" id="tr-ljn-gdw-9xf" class="tr"><td data-tag="td" id="td-8zu-mj9-5h6" class="td"><p id="p-a20-z5k-8tq">filter</p></td><td data-tag="td" id="td-3xg-qe3-rc5" class="td"><p id="p-6fu-627-6gc">該算子上的過濾條件。</p><p id="p-fqr-ihc-61w">由于示例中 EXCH-IN/OUT（PKEY）算子沒有設置 filter，所以為 nil。</p></td></tr><tr data-tag="tr" id="tr-9wx-qu5-cyl" class="tr"><td id="td-bdg-t2z-loj"><p id="p-l42-77e-inz">pkey</p></td><td id="td-kr3-l9p-0lm"><p id="p-q17-irl-7xi">按照哪一列進行重分區。</p><p id="p-nof-78y-tlz">例如，<code data-tag="code" class="code">#keys=1, [s.c1]</code> 表示按照 c1 這一列重分區</p></td></tr></tbody></table> ## EXCH-IN/OUT (HASH) EXCH-IN/OUT (HASH) 算子用于對數據使用一組 HASH 函數進行重分區。如下例所示的執行計劃中，3-5 號以及 7-8 號是兩組使用 HASH 重分區的 EXCHANGE 算子。這兩組算子的作用是把 t 表和 s 表的數據按照一組新的 HASH 函數打散成多份，在這個例子中 HASH 的列為 t.c2 和 s.c2，這保證了 c2 取值相同的行會被分發到同一份中。基于重分區之后的數據，2 號算子 HASH JOIN 會對每一份數據按照`t.c2= s.c2`進行聯接。此外，由于查詢中執行了并行度為 2，計劃中展示了 dop = 2 （dop 是 Degree of Parallelism 的縮寫)。 ~~~ obclient>CREATE TABLE t (c1 INT, c2 INT) PARTITION BY HASH(c1) PARTITIONS 4; Query OK, 0 rows affected (0.12 sec) obclient>CREATE TABLE s (c1 INT, c2 INT) PARTITION BY HASH(c1) PARTITIONS 4; Query OK, 0 rows affected (0.12 sec) obclient>EXPLAIN SELECT /*+PQ_DISTRIBUTE(@"SEL$1" ("TEST.S"@"SEL$1" ) HASH HASH), PARALLEL(2)*/ * FROM t, s WHERE t.c2 = s.c2\G; *************************** 1. row *************************** Query Plan: ================================================================= |ID|OPERATOR |NAME |EST. ROWS |COST | ----------------------------------------------------------------- |0 |PX COORDINATOR | |1568160000|2473629500| |1 | EXCHANGE OUT DISTR |:EX10002|1568160000|1063429263| |2 | HASH JOIN | |1568160000|1063429263| |3 | EXCHANGE IN DISTR | |400000 |436080 | |4 | EXCHANGE OUT DISTR (HASH)|:EX10000|400000 |256226 | |5 | PX PARTITION ITERATOR | |400000 |256226 | |6 | TABLE SCAN |T |400000 |256226 | |7 | EXCHANGE IN DISTR | |400000 |436080 | |8 | EXCHANGE OUT DISTR (HASH)|:EX10001|400000 |256226 | |9 | PX PARTITION ITERATOR | |400000 |256226 | |10| TABLE SCAN |S |400000 |256226 | ================================================================= Outputs & filters: ------------------------------------- 0 - output([T.C1], [T.C2], [S.C1], [S.C2]), filter(nil) 1 - output([T.C1], [T.C2], [S.C1], [S.C2]), filter(nil), dop=2 2 - output([T.C1], [T.C2], [S.C1], [S.C2]), filter(nil), equal_conds([T.C2 = S.C2]), other_conds(nil) 3 - output([T.C1], [T.C2]), filter(nil) 4 - (#keys=1, [T.C2]), output([T.C1], [T.C2]), filter(nil), dop=2 5 - output([T.C1], [T.C2]), filter(nil) 6 - output([T.C1], [T.C2]), filter(nil), access([T.C1], [T.C2]), partitions(p[0-3]) 7 - output([S.C1], [S.C2]), filter(nil) 8 - (#keys=1, [S.C2]), output([S.C1], [S.C2]), filter(nil), dop=2 9 - output([S.C1], [S.C2]), filter(nil) 10 - output([S.C1], [S.C2]), filter(nil), access([S.C1], [S.C2]), partitions(p[0-3]) ~~~ 其中，PX PARTITION ITERATO 算子用于按照分區粒度迭代數據，詳細信息請參見[GI](https://open.oceanbase.com/docs/community/oceanbase-database/V3.1.0/GI-1-2)。上述示例的執行計劃展示中的 outputs & filters 詳細列出了 EXCH-IN/OUT (HASH) 算子的輸出信息如下： <table data-tag="table" id="table-aoh-27c-kxz" class="table"><colgroup span="1" width="200" data-tag="col" id="col-cmr-hqo-zvx" colwidth="1*" colnum="1" colname="col1" style="width:50%" class="col"></colgroup><colgroup span="1" width="540" data-tag="col" id="col-5k0-01h-m1j" colwidth="1*" colnum="2" colname="col2" style="width:50%" class="col"></colgroup><thead id="thead-ow6-5km-ka4" class="thead"><tr id="tr-u7s-4kf-8sr"><th id="td-atw-ra4-wn8"><p id="p-1ht-dl6-eev">信息名稱</p></th><th id="td-hjt-wg2-1bk"><p id="p-oau-6xw-u5d">含義</p></th></tr></thead><tbody data-tag="tbody" id="tbody-qfd-tdk-vcr" class="tbody"><tr data-tag="tr" id="tr-fj3-cwm-r3l" class="tr"><td data-tag="td" id="td-5se-18g-io5" class="td"><p id="p-s4w-9d7-5wb">output</p></td><td data-tag="td" id="td-qt3-gzg-w11" class="td"><p id="p-fp9-jjq-eqy">該算子輸出的表達式。</p></td></tr><tr data-tag="tr" id="tr-ws8-gjb-fop" class="tr"><td data-tag="td" id="td-xqs-gti-6q8" class="td"><p id="p-yoh-tsl-g8z">filter</p></td><td data-tag="td" id="td-zuq-46a-x5d" class="td"><p id="p-yb8-4vx-xj8">該算子上的過濾條件。</p><p id="p-3r0-zwt-lrb">由于示例中 EXCH-IN/OUT (HASH) 算子沒有設置 filter，所以為 nil。</p></td></tr><tr data-tag="tr" id="tr-pj8-0ak-kvi" class="tr"><td id="td-18c-f9y-po5"><p id="p-345-xem-2p5">pkey</p></td><td id="td-66t-otm-8zd"><p id="p-36o-2u3-r7u">按照哪一列進行 HASH 重分區。</p><p id="p-dkl-9t7-jul">例如，<code data-tag="code" class="code">#keys=1, [s.c2]</code> 表示按照 c2 這一列進行 HASH 重分區。</p></td></tr></tbody></table> ## EXCH-IN/OUT（BROADCAST） EXCH-IN/OUT(BROADCAST) 算子用于對輸入數據使用 BROADCAST 的方法進行重分區，它會將數據廣播到其他線程上。如下示例的執行計劃中，3-4 號是一組使用 BROADCAST 重分區方式的 EXCHANGE 算子。它會將 t 表的數據廣播到每個線程上，s 表每個分區的數據都會嘗試和被廣播的 t 表數據進行聯接。 ~~~ obclient>CREATE TABLE t (c1 INT, c2 INT) PARTITION BY HASH(c1) PARTITIONS 4; Query OK, 0 rows affected (0.12 sec) obclient>CREATE TABLE s (c1 INT, c2 INT) PARTITION BY HASH(c1) PARTITIONS 4; Query OK, 0 rows affected (0.12 sec) obclient>INSERT INTO s VALUES (1, 1), (2, 2), (3, 3), (4, 4); Query OK, 1 rows affected (0.12 sec) obclient>EXPALIN SELECT /*+PARALLEL(2) */ * FROM t, s WHERE t.c2 = s.c2\G; *************************** 1. row *************************** Query Plan: ====================================================================== |ID|OPERATOR |NAME |EST. ROWS |COST | ---------------------------------------------------------------------- |0 |PX COORDINATOR | |1568160000|2473449646| |1 | EXCHANGE OUT DISTR |:EX10001|1568160000|1063249409| |2 | HASH JOIN | |1568160000|1063249409| |3 | EXCHANGE IN DISTR | |400000 |436080 | |4 | EXCHANGE OUT DISTR (BROADCAST)|:EX10000|400000 |256226 | |5 | PX PARTITION ITERATOR | |400000 |256226 | |6 | TABLE SCAN |T |400000 |256226 | |7 | PX PARTITION ITERATOR | |400000 |256226 | |8 | TABLE SCAN |S |400000 |256226 | ====================================================================== Outputs & filters: ------------------------------------- 0 - output([T.C1], [T.C2], [S.C1], [S.C2]), filter(nil) 1 - output([T.C1], [T.C2], [S.C1], [S.C2]), filter(nil), dop=2 2 - output([T.C1], [T.C2], [S.C1], [S.C2]), filter(nil), equal_conds([T.C2 = S.C2]), other_conds(nil) 3 - output([T.C1], [T.C2]), filter(nil) 4 - output([T.C1], [T.C2]), filter(nil), dop=2 5 - output([T.C1], [T.C2]), filter(nil) 6 - output([T.C1], [T.C2]), filter(nil), access([T.C1], [T.C2]), partitions(p[0-3]) 7 - output([S.C1], [S.C2]), filter(nil) 8 - output([S.C1], [S.C2]), filter(nil), access([S.C1], [S.C2]), partitions(p[0-3]) ~~~ 上述示例的執行計劃展示中的 outputs & filters 詳細列出了 EXCH-IN/OUT (BROADCAST) 算子的信息，字段的含義與 EXCH-IN/OUT 算子相同。