Incorporating Partitioning and Parallel Plans into the SCOPE optimizer

這篇paper中讨論是的Microsoft的cosmos DB，其本身是一個海量資料的大規模計算平台，有些類似hadoop，使用的是一種類SQL的腳本，叫做SCOPE，針對SCOPE的優化器負責生成最優的執行計劃。在1998年前後Microsoft基本丢棄了Sybase原有的優化器實作，并由Graefe主導重寫了基于cascades的優化器。是以和Microsoft所有其他的資料庫産品一樣，SCOPE optimizer也是基于Cascades的transformation-based的優化器。

本paper介紹的是如何在SCOPE的優化過程中，無縫接入對于并行計劃的考慮，同時利用functional dependency +等價列等概念，利用partitioning/grouping/ordering等資訊盡量減少/避免分區，排序，分組等操作，提高執行效率。PolarDB的并行優化方案中，在屬性統一描述 + 相容性判斷 + 屬性推導等方面，參考了這篇paper的思路。

綜述

概略來說，它擴充了cascades中property的概念，把partitioning/grouping/sorting，用一個統一的structural property來進行描述，并利用屬性的相容性來生成更高效的并行plan。例如下面這個簡單的例子：

對于上圖中的SQL query，最基礎直覺的并行執行方式如(a)所示，兩個分區表先通過在join key上的repartition操作，形成co-locate join，并行join後根據後續group by key做二次repartition，完成aggregation的并行計算。這是一個合理的plan，但由于有多次的repartition而不那麼高效。但如果基于R.a = S.a這個join條件，隻在R.a / S.a上做第一輪repartition，且存在這樣的functional dependency: R.c -> R.a(R.c是R的主鍵)，則并行join後的結果也在R.c上分布，可以省去第二次repartition直接完成join。

可以看到通過利用FD等來推導分布的相容性，就可以生成更簡化的plan。

這篇paper就是以一種系統化的方式，統一且規範的描述了partitioning/grouping/sorting三方面的屬性極其相容性推導，然後融合到cascades的optimization rules中。此外，還引入了一系列利用FD / 資料限制的inference rules。

并行plan + exchange算子

data exchange operator是一個描述資料重新分布的算子，在并行/MPP環境下，exchange是保證資料并行處理的基本邏輯操作，具體實作中，exchange包含發(partition) + 收(merge)兩個操作，在多台機器上同時執行。

分發拓撲

這裡主要考慮Initial Partitioning(1->n) , Full Repartitioning(m -> n)， Full Merge (n -> 1) 這三種

Partitioning 方案

一個公共前提是，Paritition本身是FIFO的方式，是以如果原來前後的兩個tuple r1, r2，在分區之後如果還在一個分區内，則仍然保持這種前後關系。

1. Hash Partitioning：基于partition key的hash value進行分發，partition之間無序，且不保證資料的均勻。
2. Range Partitioning： 将partition key的domain分成若幹不相交的range，partition之間整體有序，也不保證資料均勻。
3. Non-deterministic Partitioning： 例如Round-robin/random分發，可以較好應對data skew。
4. Broadcast： 将全量資料分發到所有目标節點，适合資料量較小的場景。

Merging方案

将來自多個資料源的分發資料，彙總到單一節點。

1. Random Merge： 從任一Input上擷取資料，各input内部的順序可以保留，各input之間的順序無法保留。
2. Sort Merge： 隻有當各輸入，在sort列上各自有序時，輸出可以保證全體有序，使用例如多路歸并的排序算法。
3. Concat Merge： 一個input一個input的處理，各個input内部順序可以保留，各input之間無法保證。
4. Sort Concat Merge： 先确定各個input之間的順序，然後按序對各input做處理，各個input内部順序可以保留，input之間也可以全局有序。

optimizer中利用property

在架構中，query expression用來描述某個特定的算子子樹，其中的算子包括physical / logical operator，logical operator描述算子的操作類型，而physical operator則确定算子使用的實體算法。優化過程分為2種特定操作，Logical exploration和physical optimization。logical exploration應用transformation rules生成新的logical expressions，而physical optimization應用implementation rules将logical operator轉換為physical operator。

如下算法描述了給定一個初始query expression以及對最終輸出的property requirements後，超級簡化的遞歸優化過程：

上圖中有下劃線的幾個函數是優化中和physical property最為相關的幾個操作：

Determining child required properties

parent實體算子會對目前算子的輸出施加某種property requirement，這個req必須被滿足。例如如果parent是使用ordered group by的計算方式，會要求目前算子在group by key上具有有序性。同時目前算子的實體實作也會對其input children算子的輸出提出某種property requirement，也同樣需要被滿足。

這個函數就是用來決定children的property requirement，它由父算子對目前算子的requirment和目前算子的實體實作決定。

Deriving delivered properties

這個函數根據輸入資料的實體屬性和目前算子的實體實作算法，推導出其輸出資料的實體屬性。

Property matching

一旦輸出實體屬性被推導出來，就需要判斷其與目前算子的property requirement是否match，如果無法match則目前plan就是無效的，需要被丢棄。

注意所謂match并不要求完全一緻，這裡有一定的相容性規則，後面會具體說明。

規範化描述

Functional dependency / 限制 / 等價性

FD的含義是： 一組column set R與一組column set S，如果對其中任意兩個tuple，其在R上的值相同，則在S上的值一定相同，則R -> S。

FD可能來源于幾個方面：

1. R -> R’ ，隻要R’是R的子集
2. key限制，一個relation的主鍵可以FD決定relation的所有其他列
3. 等值謂詞 col1 = col2 ，意味着 col1 -> col2 并且 col2 -> col1
4. 等值常量 col1 = const，意味着 

   

5. grouping column，在做完group by後，grouping column成為結果的key

column等價類的含義在之前的文章 已經提及了，如果對于一個relation中的所有tuple，在某些column set上都具有相同的值，則這組column set構成column等價類，等價類中也可以包含常量，這和MySQL中的MEP概念一緻。

structural property

用一個統一結構來描述partitioning ,grouping, sorting這三方面的實體屬性，屬性根據其作用域分為了2個類别：parititioning是全局屬性，描述全局的資料如何分布；grouping/sorting則是局部屬性，描述了每個partition内部，資料的實體特性。(paper中使用了很多複雜的數學符号，其實概念是很簡單的，為了便于說明這裡就不一一列舉了，隻是口述下概念)

是以是property是從global/local兩個次元，綜合起來考慮。其中grouping是一個列的集合，分組列之間沒有順序要求，sorting則是一個列的清單，列之間的前後順序不可變。

1. 局部的property使用一組固定順序的Action序列 {A1, A2 ... Am} 來描述，表示一個Action一個Action的進行操作，每個action都是基于前面action序列的結果進行操作，不破壞前面序列的屬性：比如A1是分組{C1, C2}，表示在C1,C2上分組，A2是排序C3，表示在每個C1,C2的分組内，進行排序。其中每個action要麼是針對一組column做分組操作，要麼針對單個column做排序操作。
2. 全局的property描述partitioning方式，主要包含2種：ordered / non-ordered，non-ordered partition隻能保證，在partition column上具有相同value的tuple會在同一個partition中；而ordered partition還可以額外保證，不同partition會覆寫disjoint的key range且partition間有序，也就是說，一個partition内tuple全部都小于另一個partition中的tuple。

把以上2個方面結合起來就構成了對structural property的規範描述：

前面是分區操作，其中 

表示有序/無序的全局分區，後面則是一系列操作序列，用來建構分區内局部屬性。而相關處理就在這2個正交的次元上，各自獨立進行。

例如一個relation有C1, C2, C3列以及結構化屬性{ 

 }，也就是說在C1上分區，而在每個分區内，資料首先在{C1, C2} 列上分組，而在每個分組内，按C3列有序。如下資料

Inference Rules

Rule 1. 局部屬性可以從尾部truncate掉，這個很容易了解，因為每個action總是基于前序的action來操作，是以前序部分總可以被保留。

Rule 2. 全局屬性可以expand，如果資料在C1列上做partition，它同樣在{C1, C2}列上做partition，因為具有相同(c1,c2)組合值的元素具有相同c1值，是以必然在同一分區。

更廣泛來說，在C1…Cm上分布，可以推導出在C1…Cm,Cm+1分布。

Rule 3. 在C列上有序，可以推導出在C列上分組，反過來不成立

Rule 4. 利用FD，可以盡量化簡grouping列或者order列

對于group列集合，如果其中一些列可以functional的決定其他一些列，被決定的列可以從group列集合中去除。這裡沒有對列的順序要求。

對于order列的集合，如果Cj的字首部分可以functional的決定Cj，則Cj可以從order列集合中去除。注意這裡是有列的順序要求的，必須是字首。

Rule 5. 對于local屬性，還有一種化簡方式

如果A1…Ai-1的操作序列所影響的列，可以完全決定Ai操作所影響的列，則Ai操作是無用的，可以去掉。

生成structural property

Partition operator産生的property

前面已經提到，Partition本身是FIFO的方式，是以任一個分區内，tuple的順序在partition前後是不會改變的，也就是partition隻會影響全局屬性，不會影響局部屬性。exchange的output property會內建input property的局部屬性，而改變其全局屬性。

1. 如果做hash distribution，則輸出在partition列上呈現分組特定（相同資料位于同一分區）。
2. 如果做range partition，則partition間整體有序，即ordered partition。
3. 如果是随機分發，則 

   

   表示沒有特定屬性
4. Broadcast時， 

   

    表示所有資料重複

Merge operator産生的property

Merge操作會産生針對多個輸入産生單一輸出，其局部屬性取決于input的局部屬性和merge操作的類型。

1. 對于non-ordered partitioning輸入

random merge無法保留任何局部屬性

Sort merge當輸入的局部屬性的有序列和全局的排序列一緻，可以保證全局有序，否則無序

Concat merge當輸入的局部屬性已經在分區列上進行了分組，concat之後這個分組可以保留

Sort Concat merge同上

2. 對于ordered partitioning (range)輸入：

Sort Concat merge，如果局部屬性的排序列就是全局的分區列，則可以保證全局有序

Repartition operator産生的property

repartition就是partition + merge的組合操作，是一個完整的exchange算子實作，是以其産生的屬性也是前面2項的各種組合，如下圖所示

生成property requirement

不同的算子的不同實體實作算法，會根據算子本身是串行或并行執行，對其輸入的資料流的 property有不同的要求，列舉在下圖中

從上圖可以看出，table scan/select/project對于輸入無要求，主要是order by/group by/join。

串行輸入

1. hash group by對輸入無要求
2. stream group by要求輸入在分組列上已完成具有分組屬性
3. NL/HS join對輸入無要求
4. MergeJoin 要求輸入在join列上有序

并行輸入

1. hash group by要求輸入的分區列是分組列的子集或相同。
2. streaming group by除了這個要求外，還要求各個局部輸入在分組列上已具有分組屬性。
3. NL/HS join要求輸入的分區列，是join列的子集，且兩側的輸入列要對應。
4. MergeJoin 除了以上要求，還要求兩側的輸入，在各個分區内按照join列有序

有了output property和property requirement，下面就是兩者之間如何比對了。在評估output和requirement時，可以在global/local兩個正交次元分别比較即可，方法類似于DB2中的order optimization的思路：

1. 利用FD + 等價列，轉換為最簡且統一的normalized形式，具體就是首先用等價列中的head來統一替換其他列。
2. 利用前面提到的各種推斷規則，進行轉換(truncate/expand/Co->Cg）和推導，判斷是否可以從輸出property => 要求property，如果可以，則說明兩者match。

舉個例子更容易說明白：

輸出屬性是

屬性要求是

給定的FD是 { C6, C2 } -> { C3 }，此外有兩個等價列 { "C1", C6 }, { "C2", C7 }，引号内的是head列。

先用等價列替換，變為

可以看到，這裡不僅替換了屬性描述本身，也替換了FD中的列資訊！

然後應用{ C1, C2 } -> { C3 }這個FD，輸出屬性變為

先看全局分區屬性，由于其可以expand，是以{C2,C1} => {C1,C2,C4}。

再看局部屬性，由于其可以truncate，可以将尾部的C5排序去除掉，是以得到 

，同時由于inference rule 3，有序可以得到分組，是以得到 

。

全局和局部屬性都可以match，是以requirement可以被output滿足。

Enforcer rules

在有了這些Property/Rules/Matching等之後，在Cascades的優化過程中，去內建parallel/serial的plan選項，就比較直接了：

• 在枚舉每個實體算子實作時，既要枚舉串行的實作，也要枚舉不同的并行實作，不同的并行方式，可以産生不同的輸出property，也會對輸入有不同的property要求
• 利用enforcer rules，引入不同的enforcer (sort/exchange)，并進行requirement的傳導

enforcer rules包括對sorting/grouping/partition不同屬性的處理，但思路都是一樣的：

1. 如果目前operator的實體實作，可以産生要求的property，則直接枚舉該operator，然後遞歸到其child上，對child的要求就是該實體實作對于input的要求
2. 如果目前operator的實體實作，隻能維持輸入的property，則枚舉該operator，并遞歸到其child上，對child的要求，除了實體實作本身對于input的要求，還有透傳下來的上層要求
3. 如果目前operator的實體實作，無法産生要求的property，則枚舉該operator，并在其上加入一個enforcer！這樣改變了對該operator的輸出要求，重新對該operator進行判斷

以上過程其實就是cascades framework的處理流程的一部分，所有這些選擇都要枚舉到，然後基于cost選擇最優。不過由于選項太多，會有一些heuristic，比如不同rules之間的優先級，進而做一些pruning。

總結

這篇paper提出了這種統一的property處理架構，并內建到了cascades的優化流程中。

核心要素就是實體屬性的推斷規則 + 算子輸出屬性的生成 + 對輸入屬性的要求 + match推導。PolarDB的并行優化器參考了cascades的設計思路，是以也具備多屬性的統一描述結構，其處理方式很多都參考了這篇paper，不過目前還沒有這麼完善的推斷規則，後續有持續改進的空間。