PostgreSQL並行查詢是個什麼“鬼"?
【導語】2016年4月，PostgreSQL社群釋出了PostgreSQL 9.6 Beta 1，迎來了並行查詢（Parallel Query）這個新特性。在追求高效能運算和查詢的大資料時代，能提升效能的特性都會成為一個新的熱門話題。作為關注PostgreSQL發展的資料庫開發者，本文作者將分享對於一些PostgreSQL並行查詢特性相關話題的認識。
➤並行查詢的背景
隨著SSD等磁碟技術的平民化，以及動輒上百GB記憶體的普及，I/O層面的效能問題得到了有效緩解。提升資料庫的擴充套件效能，可以追求Scale Out的方式，增加機器，往分散式方向發展，也可以追求Scale Up，增加硬體元件，充分利用各個硬體的資源，把單機的效能發揮到最大效果。相較而言，Scale Up通過軟體加速效能，依賴軟體層面的優化，是低成本的擴充套件方案。
現代伺服器除了磁碟和記憶體資源的增強，多CPU的配置也足夠強大。資料庫的Join、聚合等操作記憶體耗費比較大，很多時間花在了資料的交換和快取上，CPU的利用率並不高，所以面向CPU的加速策略中，併發執行是一種常見的方法。
查詢的效能是評價OLAP型資料庫產品好壞的核心指標，而並行查詢可以聚焦在資料的讀和計算上，通過把Join、聚合、排序等操作分解成多個操作實現並行。
並行查詢的挑戰在於，為了要做並行而加入的資料分片過程、程序或執行緒間的通訊，以及併發控制方面帶來的系統開銷不但沒有增加效能，反而降低了原有效能。實現上，如何在優化器裡規劃好並行計劃也是很多資料庫做不到的。
PostgreSQL的並行查詢功能主要由PostgreSQL社群的核心開發者Robert Haas等人開發。從Robert Haas的個人部落格瞭解到，社群開發PostgreSQL的並行查詢特性時間表如下：
- 2013年10月，執行框架上做了Dynamic Background Workers和Dynamic Shared Memory兩個調整
- 2014年12月，Amit Kapila提交了一個簡單版的parallel sequential scan的patch；
- 2015年3月，正式版的parallel sequential scan的patch被提交；
- 2016年3月，支援parallel joins和parallel aggregation；
- 2016年4月，作為9.6的新特性發布。
PostgreSQL的並行查詢在大資料量（中間結果在GB以上）的Join、Merge場合，效果比較明顯。效果上，因為系統開銷，投入的資源跟效能提升並不是線性的，比如增加4個worker，效能則可能提升2倍左右，而不是4倍。通過TPCH的測試效果，表明在Ad-Hoc查詢場景，普遍都有加速效果。
➤並行查詢功能說明
現在支援的並行場景主要是以下3種：
- parallel sequential scan
- parallel join
- parallel aggregation
鑑於安全考慮，以下4種場景不支援並行：
- 公共表表達式（CTE）的掃描
- 臨時表的掃描
- 外部表的掃描（除非外部資料包裝器有一個IsForeignScanParallelSafeAPI）
- 對InitPlan或SubPlan的訪問
使用並行查詢，還有以下限制：
- 必須保證是嚴格的read only模式，不能改變database的狀態
- 查詢執行過程中，不能被掛起
- 隔離級別不能是SERIALIZABLE
- 不能呼叫PARALLEL UNSAFE函式
並行查詢有基於代價策略的判斷，譬如小資料量時預設還是普通執行。在PostgreSQL的配置引數中，提供了一些跟並行查詢相關的引數。我們想測試並行，一般設定下面兩個引數：
- force_parallel_mode：強制開啟並行模式的開關
- max_parallel_workers_per_gather：設定用於並行查詢的worker程序數
一個簡單的兩表Join查詢場景，使用並行查詢模式的查詢計劃如下：
test=# select count(*) from t1;
count
----------
10,000,000
(1 row)
test=# select count(*) from t2;
count
----------
10,000,000
(1 row)
test=# explain analyze select count(*) from t1,t2 where t1.id = t2.id ;
QUERY PLAN
------------------------------------------------------------------------------
Finalize Aggregate (cost=596009.38..596009.39 rows=1 width=8) (actual time=17129.158..17129.158 rows=1 loops=1)
-> Gather (cost=596009.17..596009.38 rows=2 width=8) (actual time=16907.462..17129.132 rows=3 loops=1)
Workers Planned: 2
Workers Launched: 2
-> Partial Aggregate (cost=595009.17..595009.18 rows=1 width=8) (actual time=17038.230..17038.231 rows=1 loops=3)
-> Hash Join (cost=308310.48..570009.22 rows=9999977 width=0) (actual time=8483.284..16703.813 rows=3333333 loops=3)
Hash Cond: (t1.id = t2.id)
-> Parallel Seq Scan on t1 (cost=0.00..85914.87 rows=4166687 width=4) (actual time=0.575..741.057 rows=3333333 loops=3)
-> Hash (cost=144247.77..144247.77 rows=9999977 width=4) (actual time=8449.743..8449.743 rows=10000000 loops=3)
Buckets: 131072 Batches: 256 Memory Usage: 2400kB
-> Seq Scan on t2 (cost=0.00..144247.77 rows=9999977 width=4) (actual time=0.294..2177.531 rows=10000000 loops=3)並行查詢開啟後，解析器會生成一份Gather…Partial風格的執行計劃，這意味著到Executor層，會將Partial部分的計劃並行執行。
執行計劃裡可以看到，在做並行查詢時，額外建立了2個worker程序，加上原來的master程序，總共3個程序。Join的驅動表資料被平均分配了3份，通過並行scan分散了I/O操作，之後跟大表資料分別做Join。
➤並行查詢的實現
PostgreSQL的並行由多個程序的機制完成。每個程序在內部稱之為1個worker，這些worker可以動態地建立、銷燬。PostgreSQL在SQL語句解析和生成查詢計劃階段並沒有並行。在執行器（Executor）模組，由多個worker併發執行被分片過的子任務。即使在查詢計劃被並行執行的環節，一直存在的程序也會充當一個worker來完成並行的子任務，我們可以稱之為主程序。同時，根據配置引數指定的worker數，再啟動n個worker程序來執行其他子計劃。
PostgreSQL內延續了共享記憶體的機制，在每個worker初始化時就為每個worker分配共享記憶體，用於worker各自獲取計劃資料和快取中間結果。這些worker間沒有複雜的通訊機制，而是都由主程序做簡單的通訊，來啟動和執行計劃。
PostgreSQL中並行的執行模型如圖1所示。
圖1 PostgreSQL並行查詢的框架
以上文的Hash Join的場景為例，在執行器層面，並行查詢的執行流程如圖2所示。
圖2 並行查詢的執行流程
各worker按照以下方式協同完成執行任務：
- 首先，每個worker節點做的任務相同。因為是Hash Join，worker節點使用一個數據量小的表作為驅動表，做Hash表。每個worker節點都會維護這樣一個Hash表，而大表被平均分之後跟Hash表做資料Join。
- 最底層的並行是磁碟的並行scan，worker程序可以從磁碟block裡獲取自己要scan的block。
- Hash Join後的資料是全部資料的子集。對於count()這種聚合函式，資料子集上可以分別做計算，最後再合併，結果上可以保證正確。
- 資料整合後，做一次總的聚合操作。
worker程序又是如何建立和執行的？首先來看worker的建立邏輯（參見圖3）。
圖3 PostgreSQL的worker建立
PostgreSQL的並行處理，以worker動態建立為前提。worker可以由主程序初始化出來，並且在上下文中，先指定好入口函式。
並行查詢中，入口函式被指定為ParallelWorkerMain。而ParallelWorkerMain函式裡，在完成一系列訊號代理設定後，會呼叫ParallelQueryMain來執行查詢。ParallelQueryMain建立了一個新的執行器上下文，遞迴執行並行子查詢計劃。
用來並行查詢的worker程序接收主程序的訊號，比如一旦傳送建立程序的訊號，worker程序就會啟動，緊接著執行ParallelWorkerMain函式。進而，ParallelQueryMain也會執行，各個worker程序獨立執行子計劃，執行結果會存在共享記憶體裡。所有程序執行結束後，master程序會去搜集共享記憶體裡的結果資料（tuple），做資料整合。
➤並行查詢的改進
並行查詢的特性公佈後，不乏對並行的評價和之後的改進計劃。社群並行查詢的開發者在部落格中提到準備做一個大的共享Hash Table，這樣Hash Join操作的並行度會進一步提升。
圖4 建立大的Hash表共享資料
另外，對PostgreSQL而言，反倒是基於其folk出來的一些資料庫產品先於它做了並行查詢的特性，可以學習參考：
- Postgres-XC的分散式框架
- GreenPlum的MPP架構
- CitusDB的分散式
- VitesseDB基於多執行緒的並行
- Fujitsu的Fujitsu Enterprise PostgreSQL的並行
其中開源資料庫GreenPlum並行架構很有借鑑意義。GreenPlum的並行查詢設計了一個專門的排程器來協調查詢任務的分配，而PostgreSQL沒有這樣的設計。關於GreenPlum的執行框架，簡單講是以下三層結構：
- 排程器（QD）：排程器傳送優化後的查詢計劃給所有資料節點（Segments）上的執行器（QE）。排程器負責任務的執行，包括執行器的建立、銷燬、錯誤處理、任務取消、狀態更新等。
- 執行器（QE）：執行器收到排程器傳送的查詢計劃後，開始執行自己負責的那部分計劃。典型的操作包括資料掃描、雜湊關聯、排序、聚集等。
- Interconnect：負責叢集中各個節點間的資料傳輸。
GreenPlum會根據資料分佈情況做資料的廣播和重分佈，這是PostgreSQL的並行模型可以借鑑的。
僅僅是一個大的Hash Table，在資料訪問上有序列的開銷，worker的並行仍然受限。如圖5所示，大表和小表Join的場景參考GreenPlum的資料廣播機制，驅動表的資料可以給每個worker程序準備一個拷貝，相當於廣播了一份資料。這樣資料被高度共享，並行的效果會更好。
除了PostgreSQL生態的資料庫，關係型資料庫老大哥Oracle在並行查詢上已經積累了30年的經驗，也需要借鑑。在Oracle的官方手冊中，有對其並行查詢機制做出的說明。
圖5 借鑑GreenPlum的廣播機制提升並行效果
Oracle在每個操作環節，都能把資料高度分片，可以參考圖6所示的Hash Join的並行。
圖6 Oracle的Hash Join操作的並行流程
而在內部並行控制上，資料被分組後，不管是scan還是排序，幾組worker對分組的資料都能分治。
也就是說Oracle做到了操作符（Operator）Level的並行。在每個操作中，把資料分片後動態的並行運算。可以看到Oracle的並行查詢在做Operator級別的並行，每個操作環節，都能把資料分片後分而治之，並行程度非常高。這對資料的流轉要求也很高，資料和操作既能水平分治也能垂直分治。
PostgreSQL目前是任務級別的並行，將原先的執行計劃垂直拆分成幾個可以分離的子任務，並行實現簡單，但在大資料量時並行度不夠，而且共享記憶體的訪問負荷加重，效能提升不明顯。
圖7 Oracle內部動態的並行操作
參考Oracle的方式，按上圖改進後，worker不再是單獨執行1個任務，而是隨時被呼叫執行操作。資料根據操作分層、分片、廣播，worker程序為資料操作服務，而不是資料為worker服務。這樣在超大規模資料的場景，驅動表作為producer做資料partition，外表作為consumer做operator運算。多組這樣的操作產生的平行計算更自由，效能也更有想象空間，也是我們團隊目前在嘗試的方向。
圖8 通過資料分組和worker分組提升PostgreSQL的並行
筆者對資料庫實現的理解深度有限，立足自己的經驗分享了關於並行查詢的以上認識。關注社群郵件，可以看到PostgreSQL社群非常積極地加入更多並行查詢的特性，比如parallel bitmap index等，相信並行查詢的特性會更豐富。期待後面越來越強大的平行計算，以及隨之而來效能加速的無限可能。