并行查詢的執行 · k8s原理與實踐

[TOC] # 并行查詢的執行并行執行（Parallel Execution）是將一個較大的任務切分為多個較小的任務，啟動多個線程或者進程來并行處理這些小任務，這樣可以利用更多的 CPU 與 IO 資源來縮短操作的響應時間。并行執行分為并行查詢（Parallel Query）、并行 DDL（Parallel DDL）和并行 DML（Parallel DML）。目前 OceanBase 數據庫僅支持并行查詢，并行 DDL 與并行 DML 還未支持。啟動并行查詢的方式有以下兩種： * 通過 PARALLEL HINT 指定并行度（dop）的方式啟動并行查詢。 * 針對查詢分區數大于 1 的分區表會自動啟動并行查詢。 ## 啟用分區表并行查詢針對分區表的查詢，如果查詢的目標分區數大于 1，系統會自動啟用并行查詢，dop 的值由系統默認指定為 1。如下例所示，創建一個分區表 ptable，對 ptable 進行全表數據的掃描操作，通過 EXPLAIN 命令查看生成的執行計劃。 ~~~ obclient>CREATE TABLE PTABLE(c1 INT , c2 INT) PARTITION BY HASH(c1) PARTITIONS 16; Query OK, 0 rows affected (0.12 sec) obclient>EXPLAIN SELECT * FROM ptable\G; *************************** 1. row *************************** Query Plan: ======================================================= |ID|OPERATOR |NAME |EST. ROWS|COST | ------------------------------------------------------- |0 |EXCHANGE IN DISTR | |1600000 |1246946| |1 | EXCHANGE OUT DISTR |:EX10000|1600000 |1095490| |2 | PX PARTITION ITERATOR| |1600000 |1095490| |3 | TABLE SCAN |ptable |1600000 |1095490| ======================================================= Outputs & filters: ------------------------------------- 0 - output([ptable.c1], [ptable.c2]), filter(nil) 1 - output([ptable.c1], [ptable.c2]), filter(nil), dop=1 2 - output([ptable.c1], [ptable.c2]), filter(nil) 3 - output([ptable.c1], [ptable.c2]), filter(nil), access([ptable.c1], [ptable.c2]), partitions(p[0-15]) ~~~ 通過執行計劃可以看出，分區表默認的并行查詢的 dop 為 1。如果 OceanBase 集群一共有 3 個 OBServer，表 ptable 的 16 個分區分散在 3 個 OBServer 中，那么每一個 OBServer 都會啟動一個工作線程（Worker Thread）來執行分區數據的掃描工作，一共需要啟動 3 個工作線程來執行表的掃描工作。針對分區表，添加 PARALLEL HINT 啟動并行查詢，并指定 dop，通過 EXPLAIN 命令查看生成的執行計劃。 ~~~ obclient>EXPLAIN SELECT /*+ PARALLEL(8) */ * FROM ptable\G; *************************** 1. row *************************** Query Plan: ======================================================= |ID|OPERATOR |NAME |EST. ROWS|COST | ------------------------------------------------------- |0 |EXCHANGE IN DISTR | |1600000 |1246946| |1 | EXCHANGE OUT DISTR |:EX10000|1600000 |1095490| |2 | PX PARTITION ITERATOR| |1600000 |1095490| |3 | TABLE SCAN |ptable |1600000 |1095490| ======================================================= Outputs & filters: ------------------------------------- 0 - output([ptable.c1], [ptable.c2]), filter(nil) 1 - output([ptable.c1], [ptable.c2]), filter(nil), dop=8 2 - output([ptable.c1], [ptable.c2]), filter(nil) 3 - output([ptable.c1], [ptable.c2]), filter(nil), access([ptable.c1], [ptable.c2]), partitions(p[0-15]) ~~~ 通過執行計劃可以看出，并行查詢的 dop 為 8。如果查詢分區所在的 OBServer 的個數小于等于 dop，那么工作線程（總個數等于 dop）會按照一定的策略分配到涉及的 OBServer 上；如果查詢分區所在的 OBServer 的個數大于 dop，那么每一個 OBServer 都會至少啟動一個工作線程，一共需要啟動的工作線程的數目會大于 dop。例如，當`dop=8`，如果 16 個分區均勻的分布在 4 臺 OBServer 節點上，那么每一個 OBServer 上都會啟動 2 個工作線程來掃描其對應的分區（一共啟動 8 個工作線程）；如果 16 個分區分布在 16 臺 OBServer 節點上（每一個節點一個分區），那么每一臺 OBServer 上都會啟動 1 個工作線程來掃描其對應的分區（一共啟動 16 個工作線程）。如果針對分區表的查詢，查詢分區數目小于等于 1，系統不會啟動并行查詢。如下例所示，對 ptable 的查詢添加一個過濾條件`c1=1`。 ~~~ obclient>EXPLAIN SELECT * FROM ptable WHERE c1 = 1\G; *************************** 1. row *************************** Query Plan: ====================================== |ID|OPERATOR |NAME |EST. ROWS|COST | -------------------------------------- |0 |TABLE SCAN|ptable|990 |85222| ====================================== Outputs & filters: ------------------------------------- 0 - output([ptable.c1], [ptable.c2]), filter([ptable.c1 = 1]), access([ptable.c1], [ptable.c2]), partitions(p1) ~~~ 通過計劃可以看出，查詢的目標分區個數為 1，系統沒有啟動并行查詢。如果希望針對一個分區的查詢也能夠進行并行執行，就只能通過添加 PARALLEL HINT 的方式進行分區內并行查詢，通過 EXPLAIN 命令查看生成的執行計劃。 ~~~ obclient>EXPLAIN SELECT /*+ PARALLEL(8) */ * FROM ptable WHERE c1 = 1\G; *************************** 1. row *************************** Query Plan: ================================================= |ID|OPERATOR |NAME |EST. ROWS|COST | ------------------------------------------------- |0 |EXCHANGE IN DISTR | |990 |85316| |1 | EXCHANGE OUT DISTR|:EX10000|990 |85222| |2 | PX BLOCK ITERATOR| |990 |85222| |3 | TABLE SCAN |ptable |990 |85222| ================================================= Outputs & filters: ------------------------------------- 0 - output([ptable.c1], [ptable.c2]), filter(nil) 1 - output([ptable.c1], [ptable.c2]), filter(nil), dop=8 2 - output([ptable.c1], [ptable.c2]), filter(nil) 3 - output([ptable.c1], [ptable.c2]), filter([ptable.c1 = 1]), access([ptable.c1], [ptable.c2]), partitions(p1) ~~~ **注意** * 如果希望在查詢分區數等于 1 的情況下，能夠采用 HINT 的方式進行分區內并行查詢，需要對應的 dop 的值大于等于 2。 * 如果 dop 的值為空或者小于 2 將不啟動并行查詢。 ## 啟用非分區表并行查詢非分區表本質上是只有 1 個分區的分區表，因此針對非分區表的查詢，只能通過添加 PARALLEL HINT 的方式啟動分區內并行查詢，否則不會啟動并行查詢。如下例所示，創建一個非分區表 stable，對 stable 進行全表數據的掃描操作，通過 EXPLAIN 命令查看生成的執行計劃。 ~~~ obclient>CREATE TABLE stable(c1 INT, c2 INT); Query OK, 0 rows affected (0.12 sec) obclient>EXPLAIN SELECT * FROM stable\G; *************************** 1. row *************************** Query Plan: ====================================== |ID|OPERATOR |NAME |EST. ROWS|COST | -------------------------------------- |0 |TABLE SCAN|stable|100000 |68478| ====================================== Outputs & filters: ------------------------------------- 0 - output([stable.c1], [stable.c2]), filter(nil), access([stable.c1], [stable.c2]), partitions(p0) ~~~ 通過執行計劃可以看出，非分區表不使用 HINT 的情況下，不會啟動并行查詢。針對非分區表，添加 PARALLEL HINT 啟動分區內并行查詢，并指定 dop（大于等于 2），通過 EXPLAIN 命令查看生成的執行計劃。 ~~~ obclient>EXPLAIN SELECT /*+ PARALLEL(4)*/ * FROM stable\G; *************************** 1. row *************************** Query Plan: ================================================= |ID|OPERATOR |NAME |EST. ROWS|COST | ------------------------------------------------- |0 |EXCHANGE IN DISTR | |100000 |77944| |1 | EXCHANGE OUT DISTR|:EX10000|100000 |68478| |2 | PX BLOCK ITERATOR| |100000 |68478| |3 | TABLE SCAN |stable |100000 |68478| ================================================= Outputs & filters: ------------------------------------- 0 - output([stable.c1], [stable.c2]), filter(nil) 1 - output([stable.c1], [stable.c2]), filter(nil), dop=4 2 - output([stable.c1], [stable.c2]), filter(nil) 3 - output([stable.c1], [stable.c2]), filter(nil), access([stable.c1], [stable.c2]), partitions(p0) ~~~ ## 啟用多表并行查詢在查詢中，多表 JOIN 查詢最為常見。如下例所示，首先創建兩張分區表 p1table 和 p2table： ~~~ obclient>CREATE TABLE p1table(c1 INT ,c2 INT) PARTITION BY HASH(c1) PARTITIONS 2; Query OK, 0 rows affected (0.02 sec) obclient>CREATE TABLE p2table(c1 INT ,c2 INT) PARTITION BY HASH(c1) PARTITIONS 4; Query OK, 0 rows affected (0.02 sec) ~~~ 查詢 p1table 與 p2table 的 JOIN 結果，JOIN 條件是`p1table.c1=p2table.c2`，執行計劃如下： ~~~ obclient>EXPLAIN SELECT * FROM p1table p1 JOIN p2table p2 ON p1.c1=p2.c2\G; *************************** 1. row *************************** Query Plan: ==================================================================== |ID|OPERATOR |NAME |EST. ROWS|COST | -------------------------------------------------------------------- |0 |EXCHANGE IN DISTR | |784080000|614282633| |1 | EXCHANGE OUT DISTR |:EX10001|784080000|465840503| |2 | HASH JOIN | |784080000|465840503| |3 | EXCHANGE IN DISTR | |200000 |155887 | |4 | EXCHANGE OUT DISTR (BROADCAST)|:EX10000|200000 |136955 | |5 | PX PARTITION ITERATOR | |200000 |136955 | |6 | TABLE SCAN |p1 |200000 |136955 | |7 | PX PARTITION ITERATOR | |400000 |273873 | |8 | TABLE SCAN |p2 |400000 |273873 | ==================================================================== Outputs & filters: ------------------------------------- 0 - output([p1.c1], [p1.c2], [p2.c1], [p2.c2]), filter(nil) 1 - output([p1.c1], [p1.c2], [p2.c1], [p2.c2]), filter(nil), dop=1 2 - output([p1.c1], [p1.c2], [p2.c1], [p2.c2]), filter(nil), equal_conds([p1.c1 = p2.c2]), other_conds(nil) 3 - output([p1.c1], [p1.c2]), filter(nil) 4 - output([p1.c1], [p1.c2]), filter(nil), dop=1 5 - output([p1.c1], [p1.c2]), filter(nil) 6 - output([p1.c1], [p1.c2]), filter(nil), access([p1.c1], [p1.c2]), partitions(p[0-1]) 7 - output([p2.c1], [p2.c2]), filter(nil) 8 - output([p2.c1], [p2.c2]), filter(nil), access([p2.c1], [p2.c2]), partitions(p[0-3]) ~~~ 默認情況下針對 p1table 與 p2table（兩張表需要查詢的分區數都大于 1）都會采用并行查詢，默認的 dop 為 1。同樣，也可以通過使用 PARALLEL HINT 的方式來改變并行度。如下例所示，改變 JOIN 的條件為`p1table.c1=p2table.c2`和`p2table.c1=1`，這樣針對 p2table 僅僅會選擇單個分區，執行計劃如下所示： ~~~ obclient>EXPLAIN SELECT * FROM p1table p1 JOIN p2table p2 ON p1.c1=p2.c2 AND p2.c1=1\G; *************************** 1. row *************************** Query Plan: ============================================================= |ID|OPERATOR |NAME |EST. ROWS|COST | ------------------------------------------------------------- |0 |EXCHANGE IN DISTR | |1940598 |1807515| |1 | EXCHANGE OUT DISTR |:EX10001|1940598 |1440121| |2 | HASH JOIN | |1940598 |1440121| |3 | EXCHANGE IN DISTR | |990 |85316 | |4 | EXCHANGE OUT DISTR (PKEY)|:EX10000|990 |85222 | |5 | TABLE SCAN |p2 |990 |85222 | |6 | PX PARTITION ITERATOR | |200000 |136955 | |7 | TABLE SCAN |p1 |200000 |136955 | ============================================================= Outputs & filters: ------------------------------------- 0 - output([p1.c1], [p1.c2], [p2.c1], [p2.c2]), filter(nil) 1 - output([p1.c1], [p1.c2], [p2.c1], [p2.c2]), filter(nil), dop=1 2 - output([p1.c1], [p1.c2], [p2.c1], [p2.c2]), filter(nil), equal_conds([p1.c1 = p2.c2]), other_conds(nil) 3 - output([p2.c1], [p2.c2]), filter(nil) 4 - (#keys=1, [p2.c2]), output([p2.c1], [p2.c2]), filter(nil), dop=1 5 - output([p2.c1], [p2.c2]), filter([p2.c1 = 1]), access([p2.c1], [p2.c2]), partitions(p1) 6 - output([p1.c1], [p1.c2]), filter(nil) 7 - output([p1.c1], [p1.c2]), filter(nil), access([p1.c1], [p1.c2]), partitions(p[0-1]) ~~~ 通過計劃可以看出，p2table 僅需要掃描一個分區，在默認情況下不進行并行查詢；p1table 需要掃描兩個分區，默認情況下進行并行查詢。同樣，也可以通過添加 PARALLEL HINT 的方式改變并行度，使 p2table 針對一個分區的查詢變為分區內并行查詢。 ## 關閉并行查詢分區表在查詢的時候會自動啟動并行查詢（查詢分區個數大于 1），如果不想啟動并行查詢，可以使用添加 HINT`/*+ NO_USE_PX */`來關閉并行查詢。例如，針對分區表 ptable，添加 HINT`/*+ NO_USE_PX */`來關閉并行查詢，通過生成的執行計劃可以看出對 ptable 表的掃描沒有進行并行查詢。 ~~~ obclient>EXPLAIN SELECT /*+ NO_USE_PX */ * FROM ptable\G; *************************** 1. row *************************** Query Plan: ================================================= |ID|OPERATOR |NAME |EST. ROWS|COST | ------------------------------------------------- |0 |EXCHANGE IN DISTR | |1600000 |1246946| |1 | EXCHANGE OUT DISTR| |1600000 |1095490| |2 | TABLE SCAN |ptable|1600000 |1095490| ================================================= Outputs & filters: ------------------------------------- 0 - output([ptable.c1], [ptable.c2]), filter(nil) 1 - output([ptable.c1], [ptable.c2]), filter(nil) 2 - output([ptable.c1], [ptable.c2]), filter(nil), access([ptable.c1], [ptable.c2]), partitions(p[0-15]) ~~~ ## 并行執行相關的系統視圖 OceanBase 數據庫提供了系統視圖`gv$sql_audit/v$sql_audit`來查看并行執行的運行狀態以及一些統計信息。 `gv$sql_audit/v$sql_audit`包含字段較多，其中與并行執行相關的字段為：qc\_id、dfo\_id、sqc\_id 和 worker\_id。詳細信息請參考[(g)v$sql\_audit 介紹](https://open.oceanbase.com/docs/community/oceanbase-database/V3.1.0/g-v-sql_audit-introduction-1)。