作者 | 智鄰
來源 | 阿裡技術公衆号
一 背景
随着資料規模的不斷擴大，使用者SQL的執行時間越來越長，這不僅對資料庫的優化能力提出更高的要求，并且對資料庫的執行模式也提出了新的挑戰。随着資料庫在雲上的蓬勃發展，越來越多的傳統使用者遷移到雲上，享受雲上彈性擴充的紅利，但是随着業務的快速擴張，卻發現即使動态增加了很多資源，但SQL的執行時間還是越來越慢，并沒有随着資源的投入達到預期的效果。顯而易見，雖然新增了很多資源，但這些資源并沒用被充分利用，很多傳統的商業資料庫，如Oracle、SQL Server等都提供對并行查詢引擎的支援，以充分利用系統資源，達到加速SQL執行的效果。
本文主要介紹基于代價進行并行優化、并行執行的雲資料庫的并行查詢引擎的關鍵問題和核心技術。
二 如何将查詢并行起來
對于一個類OLAP的查詢，顯而易見的是它通常是對大批量資料的查詢，資料量大意味着資料遠大于資料庫的記憶體容量，大部分資料可能無法緩存到資料庫的緩沖區中，而必須在查詢執行時才動态加載到緩沖區中，這樣就會造成大量IO操作，而IO操作又是最耗時的，是以首先要考慮的就是如何能加速IO操作。
由于硬體的限制，每次IO的耗時基本是固定的，雖然還有順序IO和随機IO的差別，但在SSD已經盛行的今天，兩者的差異也在逐漸接近。那麼還有沒有其它方式可以加速IO呢? 顯然并行IO是一個簡單易行的方法，如果多個線程可以同時發起IO，每個線程隻讀取部分資料，這樣就可以快速的将資料讀到資料庫的緩沖區中。
并行讀取資料的示意如上圖所示，每個worker代表一個線程，如果資料已經有partition分區，可以每個線程讀取一個partition；也可以将全部資料按固定大小進行分片，比如按一個資料頁面大小，然後每個線程以Round-robin模式輪詢讀取一個分片。
這裡需要注意的是，按已有partition配置設定給不同worker可能會導緻每個worker處理的資料不均勻，而按Round-robin模式進行輪詢，如果分片設定的比較小，相對來說就比較容易做到每個worker處理的資料比較均勻。
如果隻是将資料讀取到緩沖區中，而不是立即進行後續處理，那麼這些資料就會因緩沖區爆滿導緻資料被換出，進而失去加速IO的意義。是以，在并行讀取資料的同時，必須同時并行的處理這些資料，這是并行查詢加速的基礎。
傳統的優化器隻能生成串行的執行計劃，為了實作并行讀取資料，同時并行處理資料，首先必須對現有的優化器進行改造，讓優化器可以生成我們需要的并行計劃。比如選擇哪個表或哪些表可以并行讀取，并且通過并行讀取會帶來足夠的收益；或者哪些操作可以并行執行，并且可以帶來足夠的收益。
并不是說并行化改造一定會有收益，比如對一個資料量很小的表，可能隻是幾行，如果也對它進行并行讀取的話，并行執行所需要的多線程建構再加上線程間的資料同步等所需要的代價可能遠大于所得到的收益，總體來說，并行執行會需要更多的資源和時間，這就得不償失了。是以查詢計劃的并行化必須是基于代價的，否則可能會導緻更嚴重的性能退化問題。
三 如何選擇并行掃描的表
選擇并行掃描的表是生成并行計劃的重要基礎，通過基于并行掃描代價的計算和比較，選擇可以并行掃描的表作為候選，是并行執行計劃疊代的第一步。基于新的并行代價，也許會有更優的JOIN順序選擇，尤其是當參與JOIN的表的數量比較多時，這需要更多額外的疊代空間，為防止優化過程消耗太多的時間，保持原有計劃的JOIN順序是一個不錯的選擇。另外，對于參與JOIN的每張表，因為表的通路方法不同，比如全表掃描、Ref索引掃描，Range索引掃描等，這些都會影響到最終并行掃描的代價。
通常我們選擇最大的那張表作為并行表，這樣并行掃描的收益最大，當然也可以選擇多個表同時做并行掃描，後面會繼續讨論更複雜的情況。
下面以查詢年度消費TOP 10的使用者為例：
SELECT c.c_name, sum(o.o_totalprice) as s
FROM customer c, orders o
WHERE c.c_custkey = o.o_custkey
AND o_orderdate >= '1996-01-01'
AND o_orderdate <= '1996-12-31'
GROUP BY c.c_name
ORDER BY s DESC
LIMIT 10; 其中orders表為訂單表，資料很多，這類表也被稱之為事實表，customer表為客戶表，資料相對較少，這類表也被稱之為次元表。那麼此SQL的并行執行計劃如下圖所示：
從計劃中可以看出orders表會做并行掃描，由32個workers線程來執行，每個worker隻掃描orders表的一部分資料分片，然後與customer表按o_custkey做index lookup進行JOIN，JOIN的結果發送到一個collector元件，然後由collector元件繼續做後續的GROUP BY、ORDER BY及LIMIT操作。
四 選擇多表并行的JOIN
将一張表做并行掃描之後，就會想為什麼隻能選擇一張表？如果SQL中有2張或更多的FACT表，能不能可以将FACT表都做并行掃描呢？答案是當然可以。以下面SQL為例：
SELECT o.o_custkey, sum(l.l_extendedprice) as s
FROM orders o, lineitem l
WHERE o.o_custkey = l.l_orderkey
GROUP BY o.o_custkey
ORDER BY s
LIMIT 10; 其中orders表和lineitem表都是資料量很大的事實表，此SQL的并行執行計劃如下圖所示：
從計劃中可以看到orders表和lineitem表都會做并行掃描，都由32個workers線程來執行。那麼多個表的并行是如何實作的呢？我們以2個表為例，當2個表執行JOIN時，通常的JOIN方式有Nested Loop JOIN、HASH JOIN等，對于不同的JOIN方式，為保證結果的正确性，必須選擇合理的表掃描方式。
以HASH JOIN為例，對于串行執行的HASH JOIN來說，首先選擇一個表建立HASH表稱之為Build表，然後讀取另一個Probe表，計算HASH，并在Build表中進行HASH比對，若比對成功，輸出結果，否則繼續讀取。如果改為并行HASH JOIN，并行優化器會對串行執行的HASH JOIN進行并行化改造，使之成為并行HASH JOIN，并行化改造的方案可以有以下兩種解決方案。
方案一是将2個表都按HASH key進行分區，相同HASH值的資料處于同一個分區内，由同一個線程執行HASH JOIN。方案二是建立一個共享的Build表，由所有執行HASH JOIN的線程共享，然後每個線程并行讀取屬于自己線程的另外一個表的分片，再執行HASH JOIN。最終選擇哪種方案，通過代價估算來決定。
圖2 并行HASH JOIN示意圖
- 對于方案一，需要讀取表中的所有資料，根據選中的HASH key，對資料進行分區，并将資料發送到不同的處理線程中，這需要額外增加一個Repartition算子，負責根據分區規則将資料發送到不同的處理線程。
- 對于方案二，需要并行建立共享的HASH build表，當build表建立成功後，每個線程讀取Probe表的一個分片，分别執行HASH JOIN，這裡的分片并不需要按照HASH key進行分片，每個線程分别讀取互不相交的分片即可。
五 分析統計的複雜算子的并行
對于一個分析統計的需求，GROUP BY操作是繞不開的操作，尤其對大量的JOIN結果再做GROUP BY操作，是整個SQL中最費時的一個過程，是以GROUP BY的并行也是并行查詢引擎必須優先解決的問題。
以年度消費TOP10客戶的SQL為例，對GROUP BY并行化後的并行執行計劃如下圖所示：
與之前的執行計劃相比，新的執行計劃中多了一個collector元件，總共有2個collector元件。首先我們看第二行的collector元件，它的extra資訊中有2條"Using temporary; Using filesort"，這表示它是對從workers接收到的資料執行GROUP BY，然後再按ORDER排序，因為隻有第一個collector元件在使用者的session中，是以這個collector也是在worker中并行執行，也就是說并行的做Group by和Order by以及Limit；然後看第一行的collector元件，它的extra資訊中隻有一條"Merge sort"，表示session線程對從workers接收到的資料執行一次merge sort，然後将結果傳回給使用者。這裡可能就有人會提出疑問，為什麼session線程隻做merge sort就可以完成GROUP BY操作呢？另外LIMIT在哪裡呢？
首先回答第2個問題，因為explain計劃顯示的問題，在正常模式下不顯示LIMIT操作，但在Tree模式下會顯示LIMIT操作。如下所示：
從Tree型計劃樹上可以清楚的看到LIMIT操作有2處，一處在計劃的頂端，也就是在session上，做完limit後将資料傳回給使用者；另外一處在計劃樹的中間位置，它其實是在worker線程的執行計劃上，在每個worker線程中在排序完成後也會做一次limit，這樣就可以極大減少worker傳回給session線程的資料量，進而提升整體性能。
下面來回答第一個問題，為什麼GROUP BY隻需要在worker線程上執行一次就可以保證結果的正确性。通常來說，每個worker隻有所有資料的一個分片，隻在一個資料分片上做GROUP BY是有極大的風險得到錯誤的GROUP BY結果的，因為同一GROUP分組的資料可能不隻是在本WORKER的資料分片上，也可能在其它WORKER的資料分片中，被其它WORKER所持有。但是如果我們可以保證同一GROUP分組的資料一定位于同一個資料分片，并且這個資料分片隻被一個WORKER線程所持有，那麼就可以保證GROUP BY結果的正确性。通過Tree型執行計劃可以看到，在并行JOIN之後，将JOIN的結果按GROUP分組的KEY值: c.c_name進行Repartition操作，将相同分組的資料分發到相同的WORKER，進而保證每個WORKER擁有的資料分片互不交叉，保證GROUP BY結果的正确性。
因為每個WORKER的GROUP BY操作已經是最終結果，是以還可以将ORDER BY和LIMIT也下推到WORKER來執行，進一步提升了并行執行的效率。
六 并行查詢引擎對TPCH的線性加速
附圖是一個并行查詢引擎對TPCH的加速效果，TPC-H中100%的SQL可以被加速，70%的SQL加速比超過8倍，總和加速近13倍，Q6和Q12加速甚至超過32倍。
七 總結
資料庫是應用系統的核心，而優化器是資料庫的核心，優化器的好壞幾乎可以決定一個資料庫産品的成敗。開發一個全新的優化器，對任何團隊都是一個巨大的挑戰，技術的複雜度暫且不提，就是想做到産品的足夠穩定就是一個非常難以克服的困難。是以即使傳統商業資料庫，也是在現有優化器的基礎上不斷改進，逐漸增加對并行的支援，最終成為一個成熟的并行優化器。對PolarDB也是如此，在設計和開發并行查詢引擎時，我們充分利用現有優化器的技術積累和實作基礎，不斷改進，不斷打磨，最終形成了一個持續疊代的技術方案，以保證新的優化器的穩定運作和技術革新。
關于我們
PolarDB 是阿裡巴巴自主研發的雲原生分布式關系型資料庫，于2020年進入Gartner全球資料庫Leader象限，并獲得了2020年中國電子學會頒發的科技進步一等獎。PolarDB 基于雲原生分布式資料庫架構，提供大規模線上事務處理能力，兼具對複雜查詢的并行處理能力，在雲原生分布式資料庫領域整體達到了國際領先水準，并且得到了廣泛的市場認可。在阿裡巴巴集團内部的最佳實踐中，PolarDB還全面支撐了2020年天貓雙十一，并重新整理了資料庫處理峰值記錄，高達1.4億TPS。歡迎有志之士加入我們，履歷請投遞到[email protected]，期待與您共同打造世界一流的下一代雲原生分布式關系型資料庫。
免費領取電子書
《阿裡巴巴 DevOps 實踐手冊》
火遍全球的 DevOps 到底是什麼？如何利用 DevOps 進行高效能研發？阿裡巴巴是怎樣快速落地 DevOps 的？如何享受 DevOps 紅利，打造自己的精英傳遞團隊？本書覆寫 DevOps 演進史、核心理念與阿裡巴巴最佳實踐的全方位解析，從 DevOps 到雲效架構師手把手教你搭建 DevOps 平台。
掃碼加阿裡妹好友，回複“do手冊”擷取吧~(英文字母小寫，若掃碼無效，可直接添加alimei4、alimei5、alimei6、alimei7)