PolarDB並行查詢的前世今生

2025-04-22 07:54 阿里技術

本文會深入介紹PolarDB MySQL在並行查詢這一企業級查詢加速特性上做的技術探索、形態演進和相關元件的實現原理，所涉及功能隨PolarDB MySQL 8.0.2版本上線。

一  背景

1  PolarDB

雲的興起為古老而頑固的資料庫市場帶來了新的發展機遇，據Gartner預測，到 2022 年，所有資料庫中將有 75% 部署或遷移到雲平臺，雲原生資料庫的誕生為各個資料庫廠商、雲服務供應商提供了彎道超車的絕好機遇，看一下AWS在Re:invent 2020釋出的Babelfish，就能嗅到它在資料庫市場的野心有多大。
AWS在2017年發表的關於Aurora的這篇paper[1]，引領了雲原生關係型資料庫的發展趨勢，而作為國內最早佈局雲計算的廠商，阿里雲也在2018年推出了自己的雲原生關係資料庫PolarDB，和Aurora的理念一致，PolarDB深度融合了雲上基礎設施，目標是在為客戶提供雲上特有的擴充套件性、彈性、高可用性的同時，能夠具備更低的響應延遲和更高的併發吞吐，其基本架構如下：
底層的分散式共享儲存突破了單機儲存容量的限制，而且可以隨使用者的資料量增長自動彈性擴容，計算層則是一寫多讀的典型拓撲，利用RDMA提供的高速遠端訪問能力來抵消計算儲存分離帶來的額外網路開銷。

2  挑戰

從上圖可以看到，儲存層將允許遠大於單機的資料容量（目前是128T），甚至線上會出現一些使用者，單表容量達到xx T的級別，這在基於MySQL主從複製的傳統部署中是難以想象的。同時大量使用者會有對業務資料的即時分析訴求，如統計、報表等，但大家對MySQL的直觀印象就是：小事務處理快，併發能力強，分析能力弱，對於這些即時性分析查詢，該如何應對呢？

3  方案

首先要說的是，隨著網際網路的發展，資料量的爆炸，一定的資料分析能力、異構資料的處理能力開始成為事務型資料庫的標配，MySQL社群在8.0版本中也對自身的查詢處理能力做了補強，包括對子查詢的transformation、hash join、window function支援等，同時PolarDB MySQL最佳化器團隊也做了大量工作來提升對複雜查詢的處理能力，如統計資訊增強、子查詢更多樣的transformation、query cache等。
並行查詢(Parallel Query)是PolarDB MySQL在推出伊始就配備的查詢加速功能，本質上它解決的就是一個最核心的問題：MySQL的查詢執行是單執行緒的，無法充分利用現代多核大記憶體的硬體資源。透過多執行緒並行執行來降低包括IO以及CPU計算在內的處理時間，來實現響應時間的大幅下降。畢竟，對於使用者來說，一條查詢如果可以1分鐘用10個核完成，總比10分鐘用1個核完成更有意義。此外所有成熟的商業型資料庫也都具備並行查詢的能力。

二  並行查詢介紹

1  特性

並行查詢可以說是PolarDB MySQL在計算層最為重要複雜度也最高的功能元件，隨著PolarDB的推出已經線上穩定執行多年，而且一直在持續演進，它具備如下幾個特性：
  • 完全基於MySQL codebase，原生的MySQL 100%相容，這裡包括
    • 語法相容
    • 型別相容
    • 行為相容
  • 0 附加成本，隨產品釋出就攜帶的功能
    • 無需額外儲存資源
    • 無需額外計算節點
  • 0 維護成本，使用和普通查詢沒有任何差別，只是響應變快了
    • 隨叢集部署，開箱即用
    • 對業務無侵入
    • 單一配置引數（並行度）
  • 即時性分析，PolarDB原生的一部分，受惠於REDO物理複製的低延遲
    • 統一底層事務型資料
    • 提交即可見
  • 極致效能，隨著PQ的不斷完善，對於分析型運算元、複雜查詢結構的支援能力不斷提升
    • 全運算元並行
    • 高效流水線
    • 複雜SQL結構支援
  • 穩定可靠，作為企業級特性，這個毋庸置疑
    • 擴充套件MySQL測試體系
    • 線上多年積累
    • 完備診斷體系
上面這些聽起來像是廣告宣傳詞，但也確實是並行查詢的核心競爭力。

2  演進

並行查詢的功能是持續積累起來的，從最初的PQ1.0到PQ2.0，目前進入了跨節點並行的研發階段並且很快會上線釋出，這裡我們先不介紹跨節點並行能力，只關注已上線的情況。

PQ1.0

最早釋出的並行查詢能力，其基本的思路是計算的下推，將盡可能多的計算分發到多個worker上並行完成，這樣像IO這樣的重操作就可以同時進行，但和一般的share-nothing分散式資料庫不同，由於底層共享儲存，PolarDB並行中對於資料的分片是邏輯而非物理的，每個worker都可以看到全量的表資料，關於邏輯分片後面執行器部分會介紹。
並行拆分的計劃形態典型如下：
可以看到有這麼幾個特點：
  • 執行模式是簡單的scatter-gather，也就是隻有一個plan slice，多個worker完成相同的功能，彙總到leader
  • 儘可能的下推運算元到worker上
  • leader負責完成無法下推的計算
這個方案能夠解決很多線上的慢查詢問題，得到很好的加速效果，不過也存在著一定的侷限性
  • 計劃形態是單一的，導致運算元的並行方式單一，比如group by + aggregation，只能透過二階段的聚集來完成：worker先做partial aggregation，leader上做final aggregation
  • 一旦leader上完成聚集操作，後續如果有distinct / window function / order by等，都只能在leader上完成，形成單點瓶頸
  • 如果存在資料傾斜，會使部分worker沒有工作可做，導致並行擴充套件性差
  • 此外實現上還有一些待完善的地方，例如少量運算元不支援並行、一些複雜的查詢巢狀結構不支援並行
總得來說，PQ1.0的並行形態和PostgreSQL社群的方案比較像，還有改進空間，畢竟所有商業資料庫的並行形態都要更靈活複雜。

PQ2.0

PQ2.0彌補了上面說到的那些侷限性，從執行模式上對齊了Oracle/SQL Server，實現了更加強大的多階段並行。
計劃形態典型如下：
第一眼看到的變化是這裡存在多個worker group，PQ2.0的執行計劃是多階段的，計劃會被拆分為若干片段(plan slice)，每個slice由一組worker並行完成，在slice之間透過exchange資料通道傳遞中間結果，並觸發後續slice的流水線執行。其中一些補強的點包括：
  • 全新的Cost-based並行最佳化器，基於統計資訊和代價決定最優計劃形態
  • 全運算元的並行支援，包括上面提到的複雜的多層巢狀結構，也可以做到完全的並行
  • 引入exchange運算元，也就是支援shuffle/broadcast這樣的資料分發操作
  • 引入一定自適應能力，即使並行最佳化完成了，也可以根據資源負載情況做動態調整，如回退序列或降低並行度
這些改變意味著什麼呢？我們來看一個簡單且實際的例子：
SELECT t1.a, sum(t2.b)FROMt1 JOIN t2 ON t1.a = t2.aJOIN t3 ON t2.c = t3.cGROUPBY t1.aORDERBY t1.aLIMIT10;
對上面的簡單查詢，在經過最佳化後，PQ1.0會生成圖中的執行計劃。
  • 在join的表集合中，尋找一個可以做邏輯分片的表做拆分，如果3個表都不足以拆分足夠多的分片，那就選最多的那個，比如這裡選擇了t2，它可能拆出12個分片，但仍然無法滿足並行度16的要求，導致有4個worker讀不到資料而idle。
  • 聚集操作先在worker上做區域性聚集，leader上做彙總聚集，如果各個worker上分組的聚攏不佳，導致leader仍然會收到來自下面的大量分組，leader上就會仍然有很重的聚集計算，leader算的慢了，會來不及收worker資料，從而反壓worker的執行速度，導致查詢整體變慢。
而PQ2.0的執行計劃如下
  • 雖然仍然只能在t2上做資料分片，但12個worker只需要完成t1 join t2這個操作，在join完成後一般資料量會膨脹，透過Shuffle(Repartition)將更多的中間結果分發到後續的slice中，從而以更高的並行度完成與t3的join
  • 各worker完成區域性聚集後，如果分組仍很多，可以基於group by key做一次Shuffle來將資料打散到下一層slice，下一組worker會並行完成較重的聚集操作，以及隨後的order by區域性排序，最終leader只需要做一次merge sort的彙總
這樣就解決了單點瓶頸和資料量不足導致的擴充套件性問題，實現線性加速。
為什麼線性擴充套件如此重要？
從上圖可以看到，隨著並行度的增長，E2E的響應時間是線性下降的，這對於客戶有兩個重要作用：
  • 隨著業務增長資料不斷膨脹，透過相應提高並行度來使用匹配的計算資源，來持續得到穩定可預期的查詢效能
  • 始終快速的分析時間可以驅動快速的業務決策，使企業在快速變化的市場環境中保持競爭力
完美的線性加速就是，Parallel RT = Serial RT / CPU cores，當然這並不現實

3  架構

並行查詢元件的整體架構如下
核心部分包括在3層中，從上到下依次是：
  • Cost-based Parallel Optimizer，嵌入在MySQL的最佳化器框架中，完成並行最佳化部分
  • Parallel Plan Generator，根據抽象的並行計劃描述，生成可供worker執行的物理執行計劃
  • Parallel Executor，並行執行器元件，包括一些運算元內並行功能和資料分發功能等
具體每個元件的實現會在後面詳細介紹

4  效能

由於是個人文章這裡隱去了具體執行時間（可以網上搜索下），主要看下PQ2.0的查詢加速能力，這裡並行度是32（可能有同學會奇怪為什麼Q6/Q12的加速比超過了32，後面會具體講到）
總的數字是：100%的SQL可以被加速，總和加速比是18.8倍。

5  使用方式

從易用性的角度，使用者開啟並行查詢只需要設定一個引數：
set max_parallel_degree = xxx;
如果想檢視並行執行計劃，只需要和普通查詢一樣，執行EXPLAIN / EXPLAIN FORMAT=TREE 即可。
Explain 做了必要的增強來顯示並行相關的information，包括代價、並行模式、分發方式等。

三  並行查詢實現

上面是一些總體性的內容，沒有什麼技術細節，後面的章節會依次dive到每個模組中介紹下。

1  並行最佳化器

在PQ2.0中，由於計劃形態會變得更加多樣，如果拆分計劃只是依靠簡單規則和簡單統計是很難得到最優解的，因此我們重新實現了一套完全基於cost的並行最佳化器。
基本的流程是在MySQL序列最佳化後，進一步做並行拆分，這裡可能有同學疑惑為什麼不像Oracle或Greenplum那樣搞成一體化的，也就是在最佳化流程中統一考慮串/並行的執行策略。原因在於，MySQL的最佳化流程中，各個子步驟之間沒有清晰的邊界，而且深度遞迴的join ordering演算法以及嵌入其中的semi-join最佳化策略選擇等，都使得程式碼邏輯與結構更加複雜，很難在不大量侵入原生程式碼的前提下實現一體化最佳化，而一旦對社群程式碼破壞嚴重，就沒法follow社群後續的版本迭代，享受社群紅利。
因此採用了兩步走的最佳化流程，這也是業界常用的手法，如Spark、CockroachDB、SQL Server PDW、Oceanbase等都採用了類似的方案。

代價模型的增強

既然是基於cost的最佳化，在過程中就必然要能夠得到各個運算元並行執行的代價資訊。為此PolarDB也做了大量統計資訊增強的工作：
  1. 統計資訊自動更新
  2. 序列最佳化流程中做針對並行執行的補強，例如修正table掃描方式等，這也是上面效能資料中Q6/Q12會有超線性加速比的原因
  3. 全運算元統計資訊推導+代價計算，補充了一系列的cost formula和cardinality estimation推導機制
這裡只能展示下統計資訊增強帶來的效果，收益的不止是並行查詢，序列執行也會提升。

自適應執行策略

在早期版本中，序列最佳化和並行最佳化，並行最佳化和並行計劃生成之間存在一定的耦合性，導致的問題就是在開始並行最佳化後會無法退化回序列，如果系統中這樣的查詢併發較多，會同時佔用很多worker執行緒導致CPU打爆。新的並行最佳化器解決了這個問題。
  1. 序列最佳化與並行最佳化解耦，並行最佳化會重新構建抽象運算元樹，並以此為輸入開始enumeration
  2. 並行最佳化與並行計劃生成解耦，最佳化的結果是計劃子片段的抽象描述，作為輸出進行plan generation
這樣就使執行策略的靈活性成為可能，允許在資源不足情況下，要麼退回序列，要麼降低並行度，或者進入排程佇列排隊等資源。

基於代價的窮盡式列舉

這是一個比較大的話題，概略來說，並行最佳化是一個自底向上，基於動態規劃的窮盡式列舉過程，實現思路參考了SQL Server PDW paper[2]，在過程中會針對每個運算元，列舉可能的並行執行方式和資料分發方式，並基於輸出資料的phsical property(distribution + order)構建物理等價類，從而做區域性剪枝，獲取區域性子問題的最優解並向上層傳遞，最終到root operator獲取全域性最優解。
下圖是針對t1 NLJ t2這個運算元，做列舉過程的一個簡要示例：
在整體列舉完成後，計劃空間中會產生一系列帶有資料分發Exchange Enforcer的物理運算元樹，基於代價選擇最優樹即可，然後以Enforcer作為子計劃的切分點，可以構建出一系列的執行計劃抽象描述，輸出到plan generator中。

2  並行計劃生成

從工程實現角度，並行計劃生成可以說是整個元件中複雜度最高，坑最多的部分。這裡採用了physical plan clone的機制來實現，也就是說，根據最佳化器生成的並行計劃描述，從原始序列計劃clone出各個計劃片段的物理執行計劃。
為什麼要用這種方式呢？還是和MySQL本身機制相關，MySQL的最佳化和執行是耦合在一起的，並沒有一個清晰的邊界，也就是在最佳化過程中構建了相關的執行結構。所以沒有辦法根據一個獨立的計劃描述，直接構建出各個物理執行結構，只能從序列計劃中“clone”出來，這可以說是一切複雜度的根源。
MySQL的執行結構非常複雜，expression(Item)和query block(SELECT_LEX)的交叉引用，內外層查詢的關聯(Item_ref)等等，都使得這項任務難度大增，但在這個不斷填坑不斷完善的過程中，團隊也對MySQL的最佳化執行結構有了很深入的理解，還發現了社群不少bug…
以上圖中簡單的查詢為例
SELECT t1.a, sum(t2.b) sumbFROM t1 join t2ON t1.c = t2.cGROUPBY t1.aORDERBY sumb;
雖然社群對執行器做了基於Iterator model的重構，但本質上，物理執行計劃仍然是由QEP_TAB組成的序列，其中group by+aggr由一個tmp table1完成，order by由tmp table2完成。
在做plan generation時，有兩個核心的操作：
  • clone
根據序列physical plan和子slice的描述，將相對應的結構clone到各個worker執行緒中，如上圖右下部分，將在worker上執行的t1 join t2和下推的聚集操作clone了下來。
  • refix
原始的序列計劃需要轉換為leader計劃，因此要替掉不必要的執行結構並調整一些引用關係，如上圖右上部分，由於t1 join t2和部分聚集操作已經下推，leader上需要去掉不必要的結構，並替換為從一個collector table中讀取worker傳遞上來的資料，同時需要將後續步驟中引用的t1/t2表的結構轉為引用collector表的對應結構。
這裡只是舉了最為簡單的例子，還沒有涉及子查詢和多階段plan，實際的工程實現成本要高很多。

3  並行執行器

PQ實現了一系列運算元內並行的機制，如對錶的邏輯分割槽和並行掃描，parallel hash join等，來使並行執行成為可能或進一步提升效能，還有多樣化的子查詢處理機制等，這裡選一些具有代表性的來介紹。

parallel scan

PolarDB是共享儲存的，所有資料對所有節點均可見，這和sharding的分散式系統有所不同，不同worker處理哪一部分資料無法預先確定，因此採用了邏輯分割槽的方案：
在btree這個level，會將資料切分成很多小分片，不同worker負責不同分片來觸發並行執行，這裡有一些最佳化點：
  1. 儘量做細粒度的切分，使分片數 >> worker數，然後worker之間透過round robin的方式去“搶”分片來執行，這樣自然做到了能者多勞，避免由於資料分佈skew導致的負載不均衡問題，這是shared storage系統的一個天然優勢。
  2. 切分時可以不用dive到葉子節點，也就是以page作為最小分割槽單位，來加速初始分割槽速度。

parallel hash join

hash join是社群8.0為加速分析型查詢所引入的功能，並隨著版本演進對semi hash/anti hash/left hash join均做了支援，PolarDB也引入了這些patch來實現完整的hash join功能，並實現了多種並行執行策略。
parallel hash join在build/probe兩個階段均做了並行支援
  1. build階段，多個worker向同一個共享的lock-free hash table中插入資料。
  2. probe階段，多個worker並行到hash table做搜尋。
兩個階段沒有重疊，這樣就實現了全階段的並行，但parallel hash join也有自身的問題，例如共享hash table過大導致spill to disk問題，並行插入雖然無鎖，但仍有“同步”原語帶來的cache invalidation。

partition hash join

partition hash join則可以避免以上問題，但代價則是引入資料shuffle的開銷：
如圖所示，查詢的執行過程分為了3個階段
  1. build/probe兩側都根據join key做shuffle，將資料分發到目標partition；
  2. 在每個partition內，build側各自構建小hash table；
  3. 在每個partition內，probe側各自查詢對應的hash table；
這樣就在各個partition內，完成了co-located join，每個hash table都更小來避免落盤，此外也沒有了build中的併發問題。
以上兩個方案哪個更優？由並行最佳化器基於Cost決定。

子查詢並行 – pushdown exec

這裡子查詢是表示式中的一部分，可以存在於select list / where / having等子句中。
對於相關子查詢，唯一的並行方式是隨外層依賴的資料（表）下推到worker中，在每個worker內完整執行，但由於外層並行了，每個worker中子查詢執行次數還是可以等比例減少。
例如如下查詢：
SELECT c1 FROM t1WHEREEXISTS (SELECT c1 FROM t2 WHERE t2.a = t1.a <= EXISTS subquery )ORDERBY c1LIMIT10;
EXISTS子查詢完整的clone到各個worker中，隨著WHERE條件的evaluation反覆觸發執行。

子查詢並行 – pushdown shared

這種並行方式的子查詢可以是表示式的一部分，也可以是派生表(derived table)。
概略的來說，這種並行方式適用於非相關子查詢，因此可以提前並行物化掉，形成一個臨時結果表，後續外層在並行中，各worker引用該子查詢時可以直接從表中並行讀取結果資料。
例如如下查詢
SELECT c1 FROM t1WHERE t1.c2 IN (SELECT c2 FROM t2 WHERE t2.c1 < 15 <= IN subquery )ORDERBY c1LIMIT10;
另外線上使用者的報表類查詢中，一種非常常見的Query模式就是derived table的多層巢狀，對於這類SQL，pushdown shared策略可以很好的提升並行執行的效能，例如如下示例：
上圖中每個顏色的方塊，代表了一層query block，這裡就構成了多層derived table的巢狀邏輯，有些層中透過UNION ALL做了彙總，有些層則是多個表（包括derived table）的join，對於這樣的查詢，MySQL會對每個derived table做必要的物化，在外層形成一個臨時結果表參與後續計算，而PQ2.0對這種常見的查詢模式做了更普遍的支援，現在每一層查詢的執行都是並行完成的，力爭達到線性的加速效果。

Exchanges

要生成高效靈活的執行計劃，資料分發元件是必不可少的，目前PolarDB支援了Shuffle/Broadcast/Gather三種分發方式，實現上利用lock-free shared ring buffer，做到流水線模式的高效資料傳輸。
下圖展示了Shuffle(Repartition)的基本形態
到這裡，並行查詢的線上版本功能及實現已經大體介紹完了。
作為一款成熟的企業級功能特性，團隊還實現了一整套完善的輔助工具集，來配合提升產品的易用性，實現功能的可監控、可干預、可反饋，但這裡篇幅已經很大了，就先不介紹了。

四  未來規劃

這裡說是未來規劃並不確切因為團隊已經在跨節點並行上做了大量的工作並進入了開發週期的尾端，跨節點的並行會把針對海量資料的複雜查詢能力提升到另一個水平：
  • 打通節點間計算資源，實現更高的計算並行度
  • 突破單節點在IO / CPU上的瓶頸，充分利用分散式儲存的高吞吐能力
  • 結合全域性節點管理與資源檢視，平衡排程全域性計算資源，實現負載均衡的同時保證查詢效能
  • 結合全域性一致性檢視，保證對事務性資料的正確讀取
[1]https://www.allthingsdistributed.com/files/p1041-verbitski.pdf
[2]https://www.scinapse.io/papers/2160963784#fullText

網際網路技術實戰營·資料智慧專題

網際網路企業資料運營正在經歷從工具化到資料化，從資料化向智慧化演進的轉變。以資料智慧驅動科學運營，洞察使用者行為，最佳化運營策略，進而實現使用者的精細化和智慧化運營。
本次課程面向關注資料治理與使用者增長的互娛、電商、遊戲等網際網路企業，特邀阿里雲的技術專家分享最佳實踐和解決方案，您可按需選擇想要學習的課程。點選閱讀原文檢視詳情！
關鍵詞
問題
元件
節點
結構
資料庫


相關文章