[[399309]]

一、背景

隨著數據規模的不斷擴大，用戶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也是如此，在設計和開發并行查詢引擎時，我們充分利用現有優化器的技術積累和實現基礎，不斷改進，不斷打磨，最終形成了一個持續迭代的技術方案，以保證新的優化器的穩定運行和技術革新。