最近一年一直在做PolarDB的並行最佳化器，過程中調研了各種分散式資料庫系統的最佳化和執行框架，後續幾篇文章將一一分享，首先介紹對PolarDB MySQL的並行最佳化框架影響最大的，也就是SQL Server PDW。

SQL Server PDW介紹

SQL Server PDW（Parallel DataWarehouse）是SQL Server的MPP版本，目前已經演進為Azure DataWarehouse部署在雲上，用來儲存大容量資料並處理分析型查詢。總體上是一個share nothing的經典MPP架構，類似於Greenplum，它也會利用單機SQL Server作為其sharding data和meta data的儲存+計算例項。

基本架構

叢集中每個節點都部署單個SQL Server instance + DMS服務，節點具備2種角色：

Control node 是叢集的入口點，前端應用於control node連線併發送請求，其上有一個PDW engine，做全域性性的管理控制：distributed query最佳化、執行排程管理，DMS管理，許可權檢查，對外介面。內部的SQL server上有一個shell database，儲存全域性資訊：global metadata/global statistics/資料分佈/許可權資訊，和GP一樣沒有user data。

compute node 儲存user data，並負責分散式子計劃的執行。

表資料的分片方式包括hash-partitioned和replicated（複製表）。

使用者query到達control node後，最佳化生成分散式執行計劃，稱為DSQL。和MemSQL不同，PDW的分散式plan可以用一系列的step來描述，每個step之間不構成pipelining，step直接順序依次執行，中間結果資料要物化到temp table後，才能開始下一step，step內是並行。

DSQL plan包含4種operation，每種operation都用SQL來描述（和MemSQL類似）

SQL operation，描述step的操作，直接在compute node的instance上執行SQL，返回資料

DMS operation，step執行請求傳送到DMS服務上，DMS對所屬instance執行SQL獲取源資料，根據分發方式傳送，接收方DMS把資料存入本地instance的temp table中

Temp table operation ： setup temp table，並用來接收遠端資料，供本地讀取

Return operation，結果返回client

PDW 並行最佳化

整體來說，PDW的查詢最佳化分為了2個步驟，分別在2個不同的元件中完成。

PDW Parser，獲取使用者query（T-SQL）做語法解析，生成AST，傳遞給本地SQL Server例項。

在Control node的SQL Server例項中，利用shell database中儲存的全域性metadata + 統計資訊等，完成單機執行計劃的最佳化，包括：

apply logical transformation

cardinality estimation，基於shell database statistics

apply physical implementation，為運算元列舉物理執行方式，計算相應代價並做一些基本的剪枝

由於SQL Server的單機最佳化器是基於Cascades的，最佳化的結果儲存在一個Memo的結構中，有關Cascades的原理，可參考之前的paper解讀以及Orca的實現。

但有所不同的是，這裡並不選出最優的序列執行計劃，而是將Memo中整個search space都保留下來

。

3。 將Memo傳遞給XML generator做序列化。

4。 傳遞Memo給PDW optimizer，它基於memo中各種alternative的plans，根據資料分佈情況，列舉可能的分散式執行方式+資料分發方式。相當於擴充套件了search space的新維度，加入了分佈相關的資訊，生成了新的候選plan集合，並基於cost，從中選出最優結果。

這種2-pass的方式有很多好處，首先，它可以完全複用SQL Server強大的單機最佳化器能力，眾所周知SQL Server的QO應該是眾多commercial database中最為優秀的，這樣實現完全解耦了並行最佳化和單機最佳化，工程難度也降低了1個數量級。而且由於儲存了整個Memo，不會導致串 -> 並帶來的計劃次優性問題。

在分散式元件PDW optimizer中，只需要去擴充套件DMS cost model，擴充套件physical property加入distribution即可。

PDW Query Optimizer實現

從上圖的流程可以看到，概念上這個過程並不複雜，但工程實現的挑戰還是很多的，下面一一介紹。

對現有Cascades最佳化器的增強

能夠將Memo輸出

擴充套件SQL語法，使其能夠接受一些PDW相關的hints

調整單機最佳化流程，在enumeration過程中，也會傾向對分佈執行更有利的一些最佳化方式（這裡沒有細講，個人猜想比如更傾向於subquery的展開？）

擴充套件更多的邏輯/物理運算元，比如partial aggregation/final aggregation，以及每種operator分散式實現，join可以是local/directed/broadcast/repartition， aggregation可以是repartition/partial+final的形式。

擴展出DMS enforcer，不同的分發方式形成不同物理operator

Plan Enumeration

這是本篇paper的核心，也就是如何列舉分散式運算元，基本原理如下

引入針對特定列（join列/group列）的distribution propery：（哪列，如何分佈）

自底向上

，對Memo中的group進行預處理：

修正一些分散式執行運算元的cardinality estimation，例如partial/final aggregation，在單機節點上雖然枚舉了出來但沒有分散式的statistics，這個card是不準確的，需要校正。

從分散式的角度，對每個group，合併等價的group expression。

自頂向下，生成每個group的interesting property（擴充套件了System-R的interesting order，添加了distribution屬性）。

自底向上

，對每個group列舉分佈計劃

針對input group中所有可能expression，列舉本group內所有可能的分散式執行方式，生成新的group expression。

基於cost做剪枝，對每個physical equivalent class（相同具有輸出property），保留一個lowest cost plan，此外保留一個總體的best子計劃。

基於之前生成的interesting property，列舉可能的move enforcer，然後再做和步驟2類似的cost-based pruning。

提取最優plan

從root group中獲取最優的plan tree

生成DSQL plan

Cost model

從paper中的描述來看，PDW的cost model只考慮資料傳輸+DMS做temp table物化的代價，而不考慮執行SQL操作的代價，這有幾個原因

這部分代價確實佔據了執行中的絕大多數時間

由於考慮的更少，使得cost model處理的範圍更小，易管理，開發、除錯、測試都會簡化

其基本假設是

DSQL step順序執行沒有流水線，這有cost的累計就對應了執行時間的累加

不考慮併發query的影響

認為各node在硬體上同質的

資料在節點間均勻分佈，也就是不考慮data skew

但很遺憾paper沒有給出設計細節。

DMS operation

DMS支援多種資料分發方式，這和PDW的並行執行能力相關，例如

Shuffle Move（N ： N） ，最經典的redistribution，基於shuffle key的hash value

Partition Move（N ： 1），相當於彙總

Broadcast Move （N ： ALL），資料從部分compute node分發到所有compute node

Replicated broadcast（1 ： ALL），從1個compute node分發到所有compute node

。。。

總的來說，DMS被分為source + target 兩個component，source傳送和targe接收是同時進行的：

source從本地instance獲取資料 + network（網路分發） ， 本地read和network分發是非同步的

target從Network接收資料放入本地buffer + bulkcpy（批次寫入本地instance temp table）， 本地收 和 bulkcpy也是非同步的，代價公式是

由於以上操作都是非同步的，而cost衡量的就是query執行時間，其cost formula是

Csource = max（Creader， Cnetwork）。

Ctarget = max（Cwriter， CSQLBulkCpy）。

Cdms = max（Csource， Ctarget）。

在這篇著名的paper

［1］

中，我們都看到這樣一個結論：由於Cardinality Estimation通常具有很大的誤差，導致cost model的精確性變得不那麼重要。

這裡PDW的設計者們也做出了類似的判斷，認為精細的cost model會使得plan對於資料分佈/統計資訊/執行環境等的微小變化都更為敏感，反而不夠robust，此外維護和除錯成本也更高，因此這裡每個Cxx的計算都非常簡單：Cxx = B * λ，B是操作處理的位元組數，λ是不同操作對應的常量代價因子，可以看到cost model保持了儘可能的簡單。

一個栗子

paper中給出了一個簡單的例子，SQL是

SELECT

*

FROM

CUSTOMER

C

，

ORDERS

O

WHERE

C

。

C_CUSTKEY

=

O

。

O_CUSTKEY

AND

O

。

O_TOTALPRICE

>

1000

；

上圖非常清晰的描述了4個主要步驟都發生了什麼：

（a）使用者輸入到PDW engine

（b）Parser生成AST

（c）在單機SQL Server中完成單機最佳化，形成Final （Serial） Memo，傳遞給PDW optimizer，列舉Shuffle/Replicate等運算元，生成Augmented （Parallel） Memo。

（d）從c的Parallel Memo中選出最優分散式計劃

（e）轉換為DSQL的plan描述，其中包括DMS operation + SQL operation。。。

總結

SQL Server PDW的並行最佳化方案還是非常優雅的，從本質上，它利用Cascades的top-down strategy完成單機序列的最佳化，再利用System-R的bottom-up完成並行的分散式最佳化，將兩者結合了起來，是非常聰明的解決方案。不需要太多調整原有的Cascades元件，又簡潔高效的生成了分散式計劃。

參考

^

2015， How Good Are Query Optimizers， Really？， VLDB