基本概念
SQL執行器是PolarDB-X中執行邏輯層算子的元件。對于簡單的點查SQL,往往可以整體下推存儲層MySQL執行,因而感覺不到執行器的存在,MySQL的結果經過簡單的解包封包又被回傳給使用者。但是對于較複雜的SQL,往往無法将SQL中的算子全部下推,這時候就需要PolarDB-X執行器執行無法下推的計算。
SELECT l_orderkey, sum(l_extendedprice *(1 - l_discount)) AS revenue
FROM CUSTOMER, ORDERS, LINEITEM
WHERE c_mktsegment = 'AUTOMOBILE'
and c_custkey = o_custkey
and l_orderkey = o_orderkey
and o_orderdate < '1995-03-13'
and l_shipdate > '1995-03-13'
GROUP BY l_orderkey;
通過EXPLAIN指令看到PolarDB-X的執行計劃如下:
HashAgg(group="l_orderkey", revenue="SUM(*)")
HashJoin(condition="o_custkey = c_custkey", type="inner")
Gather(concurrent=true)
LogicalView(tables="ORDERS_[0-7],LINEITEM_[0-7]", shardCount=8, sql="SELECT `ORDERS`.`o_custkey`, `LINEITEM`.`l_orderkey`, (`LINEITEM`.`l_extendedprice` * (? - `LINEITEM`.`l_discount`)) AS `x` FROM `ORDERS` AS `ORDERS` INNER JOIN `LINEITEM` AS `LINEITEM` ON (((`ORDERS`.`o_orderkey` = `LINEITEM`.`l_orderkey`) AND (`ORDERS`.`o_orderdate` < ?)) AND (`LINEITEM`.`l_shipdate` > ?))")
Gather(concurrent=true)
LogicalView(tables="CUSTOMER_[0-7]", shardCount=8, sql="SELECT `c_custkey` FROM `CUSTOMER` AS `CUSTOMER` WHERE (`c_mktsegment` = ?)")
如下圖所示,LogicalView的SQL在執行時被下發給MySQL,而不能下推的部分(除LogicalView以外的算子)由PolarDB-X執行器進行計算,得到最終使用者SQL需要的結果。
![](https://img.laitimes.com/img/__Qf2AjLwojIjJCLyojI0JCLicmbw5iY2U2N2AjNzEGO5ATMjRDO1MmM0E2NlNGOmV2M0IWNy8CX5d2bs92Yl1iclB3bsVmdlR2LcNWaw9CXt92Yu4GZjlGbh5yYjV3Lc9CX6MHc0RHaiojIsJye.png)
執行模型
與傳統資料庫采用Volcano執行模型不一樣,PolarDB-X采樣的是Pull~Push混合執行模型。所有算子按照計算過程中是否需要緩存臨時表,将執行過程切分成多個pipeline,pipeline内部采樣next()接口,按批擷取資料,完成在pipeline内部的計算,pipeline間采用push接口,上遊pipeline在計算完成後,會将資料源源不斷推送給下遊pipeline做計算。下面的例子中,被切分成兩個pipeline,在pipeline-A中掃描Table-A資料,完成建構哈希表。Pipeline-B掃描Table-B的資料,然後在HashJoin算子内部做關聯得到JOIN結果,再傳回用戶端。
執行模式
目前 PolarDB-X 支援了三種執行模式:
- 單機單線程(TP_LOCAL):查詢過程中,是單線程計算,TP負載的查詢涉及到的掃描行數比較少,往往會采用這種執行模式,比如基于主鍵的點查。
- 單機并行(AP_LOCAL):查詢過程中,會利用節點的多核資源做并行計算,如果您沒有配置隻讀執行個體,針對AP負載的查詢,往往會采樣這種執行模式,一般也稱之為Parallel Query模式。
- 多機并行(MPP):您如果配置了隻讀執行個體,針對AP負載的查詢,可以協調隻讀執行個體上多個節點的多核做分布式多機并行加速。
為了準确知道執行模式,在原有EXPLAIN和執行計劃的基礎上,擴充了
EXPLAIN PHYSICAL例如以下查詢,通過指令可以檢視目前查詢采樣的是MPP模式,此外還可以擷取到每個執行片段的并發數。
mysql> explain physical select a.k, count(*) cnt from sbtest1 a, sbtest1 b where a.id = b.k and a.id > 1000 group by k having cnt > 1300 or
der by cnt limit 5, 10;
+-------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| PLAN |
+-------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| ExecutorType: MPP |
| The Query's MaxConcurrentParallelism: 2 |
| Fragment 1 |
| Shuffle Output layout: [BIGINT, BIGINT] Output layout: [BIGINT, BIGINT] |
| Output partitioning: SINGLE [] Parallelism: 1 |
| TopN(sort="cnt ASC", offset=?2, fetch=?3) |
| Filter(condition="cnt > ?1") |
| HashAgg(group="k", cnt="COUNT()") |
| BKAJoin(condition="k = id", type="inner") |
| RemoteSource(sourceFragmentIds=[0], type=RecordType(INTEGER_UNSIGNED id, INTEGER_UNSIGNED k)) |
| Gather(concurrent=true) |
| LogicalView(tables="[000000-000003].sbtest1_[00-15]", shardCount=16, sql="SELECT `k` FROM `sbtest1` AS `sbtest1` WHERE ((`k` > ?) AND (`k` IN (...)))") |
| Fragment 0 |
| Shuffle Output layout: [BIGINT, BIGINT] Output layout: [BIGINT, BIGINT] |
| Output partitioning: SINGLE [] Parallelism: 1 Splits: 16 |
| LogicalView(tables="[000000-000003].sbtest1_[00-15]", shardCount=16, sql="SELECT `id`, `k` FROM `sbtest1` AS `sbtest1` WHERE (`id` > ?)") |
+-------------------------------------------------------------------------------------------------------------------------------------------------------------------+
同樣的也允許您通過
HINT EXECUTOR_MODE
指定執行模式。比如主執行個體空閑資源很多,可以考慮強制設定為單機或者多機并行模式來加速。
mysql> explain physical /*+TDDL:EXECUTOR_MODE=AP_LOCAL*/select a.k, count(*) cnt from sbtest1 a, sbtest1 b where a.id = b.k and a.id > 1000 group by k having cnt > 1300 order by cnt limit 5, 10; |
+-------------------------------------------------------------------------------------------------------------------------------------------------------------+
| ExecutorMode: AP_LOCAL |
| Fragment 0 dependency: [] parallelism: 4 |
| BKAJoin(condition="k = id", type="inner") |
| Gather(concurrent=true) |
| LogicalView(tables="[000000-000003].sbtest1_[00-15]", shardCount=16, sql="SELECT `id`, `k` FROM `sbtest1` AS `sbtest1` WHERE (`id` > ?)") |
| Gather(concurrent=true) |
| LogicalView(tables="[000000-000003].sbtest1_[00-15]", shardCount=16, sql="SELECT `k` FROM `sbtest1` AS `sbtest1` WHERE ((`k` > ?) AND (`k` IN (...)))") |
| Fragment 1 dependency: [] parallelism: 8 |
| LocalBuffer |
| RemoteSource(sourceFragmentIds=[0], type=RecordType(INTEGER_UNSIGNED id, INTEGER_UNSIGNED k, INTEGER_UNSIGNED k0)) |
| Fragment 2 dependency: [0, 1] parallelism: 8 |
| Filter(condition="cnt > ?1") |
| HashAgg(group="k", cnt="COUNT()") |
| RemoteSource(sourceFragmentIds=[1], type=RecordType(INTEGER_UNSIGNED id, INTEGER_UNSIGNED k, INTEGER_UNSIGNED k0)) |
| Fragment 3 dependency: [0, 1] parallelism: 1 |
| LocalBuffer |
| RemoteSource(sourceFragmentIds=[2], type=RecordType(INTEGER_UNSIGNED k, BIGINT cnt)) |
| Fragment 4 dependency: [2, 3] parallelism: 1 |
| TopN(sort="cnt ASC", offset=?2, fetch=?3) |
| RemoteSource(sourceFragmentIds=[3], type=RecordType(INTEGER_UNSIGNED k, BIGINT cnt)) |
+-------------------------------------------------------------------------------------------------------------------------------------------------------------+
在多機并行MPP執行模式的并發度是根據實體掃描行數、執行個體規格和計算所涉及到表的分表數來計算出來的,整體的并行度要考慮高并發場景,是以并行度的計算會偏保守,您可以通過上述
EXPLAIN PHYSICAL
指令檢視并行度。也同樣支援
HINT MPP_PARALLELISM
強制指定并行度。
/*+TDDL:EXECUTOR_MODE=MPP MPP_PARALLELISM=8*/select a.k, count(*) cnt from sbtest1 a, sbtest1 b where a.id = b.k and a