這篇簡短的paper從非常high level的角度描述了下Oracle 10g對于parallel query所做的重新設計和其中的一些優化，由于Oracle RAC特殊的share-disk架構，使其在并行計算上與普通的MPP資料庫有一些不同，例如對于worker的排程和配置設定方式以及對于資源/資料的動态調整。
PolarDB和Oracle一樣都基于底層的共享存儲，這讓它們共享了一些基本的設計原則和架構優勢，但在查詢的優化和執行層面的細節上則存在不小的差别，後續我會寫一篇文章專門介紹PolarDB的并行查詢技術，現在先讓我們看下Oracle的做法。（文中有一些我沒有搞懂的點，請大神們批評指正）
基本介紹
新的parallel execution(PX) engine希望具有更好的靈活性，擴充性，對資源的高效利用和易管理性。
- 靈活性
由于底層share-disk的架構，使得使用者不需要去關心資料的分區與節點間的映射關系，是以簡化了應用的開發，此外由于在任意節點都可以通路所有資料，使得Oracle可以更加靈活和動态的方式去将資料片段配置設定到不同的worker上運作。PX engine其實是和底層的硬體特性以及資料的實體分區解耦的，因為每個節點都可以看到整體的資料視圖。
- 擴充性
通過生成高效的parallel plan，并且在多個worker之間實作負載的均衡分布來達到好的擴充性，前者需要好的優化能力而後者則通過執行中的動态調整來實作。
- 資源高效利用
通過一些execution constructs來限制worker的配置設定，以及對每個worker的工作負載進行限制，進而對一些關鍵的執行資源（memory/lock/cpu...)實作更合理的利用。
- 易管理性
為了提升易管理性，引入了Parallel Single Cusor(PSC)模型，也就是對于一個SQL statement來說，所有的執行節點都共享一個相同的全局parallel plan，這樣友善收集統計資訊，監控執行狀态。（這個點沒有get到，為什麼必須全局計劃一樣才好管理？？感覺和Oracle的曆史實作方案相關。。。）
基本概念
PX engine支援節點内/節點間并行，節點内并行則通過share memory進行互動，節點間要通過網絡互動。每個query的執行需要一個Parallel Execution Coordinator(PEC)和一組Parallel Execution Servers(PES)組成，也就是一個leader和一組workers。
PX可以支援各種類型的操作，包括：
- 各類關系運算(scan/join/aggregation...)
- DML (insert/update/delete...)
- DDL (create table/index/materialized view...)
- data reorganization
- data load/unload
一個Parallel Execution Plan(PEP)包含4個要素
- Dataflow operators(DFO)，分布式計劃的一個片段，是一組算子的集合，等同于PolarDB的slice
- Table Queues(TQs)，執行資料分發操作，等同于PolarDB的exchange
- Granule Iterators (GRA)，控制對object(table/index)的動态分區，等同于PolarDB中的parallel scan
- Parallelizer row source，Oracle并行計劃的形态就是一棵由DFO組成的樹。Parallelizer腐惡排程DFO子樹如何執行，每個DFO都由一組PES來并行執行。注意和普通的MPP并行資料庫不同，Oracle的并行執行排程嚴格遵循2-DFOs的模式，也就是任意時間隻有兩個DFO在運作，一個producer一個consumer，當producer執行完成時，其PES空閑出來去執行下一個DFO，原來的consumer開始變為producer，而新的DFO成為了consumer。這種方式意味着2件事：
- PX的CPU占用最大為2 * DOP，每個DFO占用DOP個CPU
- 在producer和consumer之間存在temp table做中間結果的緩存，這樣才能盡快切走producer把執行資源空出來，也就是DFO之間無法做到pipelining
Parallel Single Cursor模型
由PEC完成并行plan的優化，PEC/PES間共享相同的plan。
Single PEP的生成
這裡對于優化流程講的非常模糊讓人十分郁悶，總的來說優化流程分為邏輯 + 實體兩個步驟，
邏輯優化包含2個pass：
- 在給定的DOP下，确定join order和table的access method來使代價最低，這裡計算cost時會考慮叢集中node的個數，各個table/index的partition個數，以及預設的distribution方法（這個指什麼？個人了解時每個算子有個預設的并行執行方式，例如group by按照group列做分布？hash join按照join key做分布？也就是按照這種分布來計算代價，然後決定access method和join order ? ）
- 真正為每個算子決定最優的distribution方法（這時才會考慮資料的statistics嗎，例如資料量，可能的NDV？）
實體優化則将邏輯計劃轉換為實際的實體算子并建構為DFO，同時考慮一些基于資料實體屬性的優化（clustering/ordering/data-fragmentation)。
吐槽下paper中的講解。。。如果真的是個人了解的那樣，為啥要搞成這種方案？不再考慮join order的同時考慮資料的各種分布方式？為啥實體屬性的影響要到計劃形态已确定時才考慮？
例如下圖展示了一個簡單的2表parallel hash join的計劃
Parallelizer在PEC上負責整個計劃的排程執行，Serial的部分負責對小表的串行scan通過hash redistribution分發到其他PES上（也在PEC上完成）。DFO 1負責大表的并行scan（由GRA負責多個PES掃表時對資料邏輯分片的配置設定，保證各個PES之間負載均衡)，然後将資料分發到DFO 2的各個worker中進行并行的hash join。
Single Global PEP的意義
一般的并行計算架構，在跨機傳遞執行計劃時，會采用2種方案：
- 傳遞plan segment的某種編碼形式（序列化...），例如Greenplum，PolarDB
- 傳遞描述plan segment的SQL語句，例如SQL Server PDW, MemSQL
Oracle認為兩者都存在一定的缺陷
- 随着plan複雜度的上升，以及PX能夠執行的并行功能越來與廣泛，維護編碼的方案将會越來越複雜且容易出錯，這确實是一個問題，隻要增加新的并行算子，就需要實作對應的編碼/解碼。
- 為每個DFO生成SQL可能無法準确描述DFO實際的實體執行計劃，可能需要非常詳盡的hint機制，而這也具有很高的開發+維護成本。
是以它采用了PSC的方案，即PEC在本地完成全局并行計劃的優化，在相同node内的PES，共享同一個physical plan，而其他node則通過把原始SQL傳遞過去，完成相同的優化得到相同的parallel plan，并在那個node内與其上的PES共享。。。這樣所有PES + PEC看到的都是全局一緻的plan。這樣有3個好處
- 易于監控各個DFO PES的執行，因為都面對相同的plan，各個PES在執行中的統計資訊可以先記錄在本地plan的row source當中，最後彙總到PEC上。
- 加入新的并行算子支援不需要考慮編/解碼，隻需要定義算子的預設分布方式（優化時用？）
- 使用SQL傳遞節省了記憶體，去掉中間表示。
但這裡應該是有兩個前提因素的：
- 實體執行計劃和執行結構的解耦，使得多個PES能共享一個plan描述，這點MySQL就做不到
- 在各個節點上優化，能夠生成完全相同的parallel plan，這點沒想好是怎麼做的，如何保證統計資訊完全相同呢？還是優化中不考慮統計資訊？
PSC的執行
通過GRA可以實作對data object(table/index)的并行讀取比較均衡，具體方式是每個PES向PEC上的GRA請求一塊object fragment information進行讀取（多個page)，當其讀取完成後會向PEC請求後續的granule，這樣就可以實作根據各個PES消耗granule的速度，在多個PES間做到負載均衡。
最前面已經提到由于共享存儲的特點，PX engine并不受底層data的分布限制，這樣可以對資料形成某種virtual partitioning，而不受限與資料實際的user partition情況，但同時也可以利用實際的user partition做一些優化，例如paper中Full Partition-wise Join：每個PES隻負載scan + join 一對partition的資料，完全避免了在PES之間的資料互動。和share-nothing架構不同，每個PES處理的partition和PES所在的計算node無關，不需要綁定在data“所在”的節點。
Cluster-aware PX
這裡是指在優化中會考慮cluster的拓撲資訊，例如上節提到的full partition-wise join，其形态如下：
這樣确實沒有任何data redistribution，但很明顯并行PES的個數将受到partition個數的限制。
如上圖所示，如果引入了hash redistribution，而且資料分發的成本，可以用更高的PES并行度來彌補，那麼就可以提升查詢的響應速度。但這裡資料分發的量明顯大了很多，還有更好的方法：
上圖是前面2種方法的混合，利用user partition的schema，來避免跨節點的data redistribution，可以看到通過限制資料分發的目标PES，來實作隻在部分partition之間做重分布，相當于把partition按照Node做了分組，每個node負責一部分partitions。這樣看就變成了一個share nothing的結構，每個node綁定一些partition，在node内部完成并行join。
這種優化可以在存在partition-wise join且partition數量不足的情況下使用。當然具體采用哪種方案需要通過cost決定。
上面說的到對PES進行分組的限制是通過一個執行器的construct : PES mapper來實作了，它會對每條tuple進行檢查，并限制其可能分發的PES分組。
Resource-aware PX
Memory
這裡舉了parallel insert partition table的例子，在并行insert時每個PES會為每個partition預留一個buffer來實作異步批量寫入，那麼為了避免占用記憶體過多，通過PES mapper限制每個PES所寫入的partition數量（預設是每個PES都向所有partition寫），來實作減少記憶體占用。
Locking
類似Memory，每個data block能夠并發修改的process數量是有上限的，通過PES mapper來把PES分組，每個data block隻能被某個分組的PES修改，進而限制了PES的數量。
Adaptive DOP and PES allocation
在每個query開始時，利用叢集中的負載資訊 + cpu數量來限制DOP，在配置設定PES到各個node時，要考慮node之間的負載均衡情況。
Granule allocation
在用GRA配置設定granule給各個PES時，應該考慮data和node的affinity或者data locality(這是指data是否在buffer pool中？還要到GCS中看block的owner嗎？）
其他功能
Flexible data distribution and pipelining
對SQL做并行的一個核心要素就是對data object(靜态table/index)或者data flow(動态的中間結果）實作分片。PX優化中會使用實體算子所具有的distribution requirement來完成這種分片。這裡舉例了window function的partition by，group by應該也是同樣道理。同時優化中會考慮不同算子在實體屬性上的相容性，例如如果group by key和order by key是相容的（group by a, b order by a)，則将執行group by的hash distribution變為range distribution，來避免再單獨一個DFO做并行order by。
Cost-based Parallelization of Subqueries
Oracle對于大多數的相關子查詢都可以實作去相關化，而對于無法unnesting的，則基于cost決定是否執行并行：
- subquery已經在PES中執行（在子計劃片段中），自身就不能再并行了
- subquery在PEC上，但如果對每行輸入，并行執行的cost太高，則也不并行
Recursive and Iterative Computation
由于producer-consumer之間存在temp table緩存中間結果，可以利用這個特點實作一些高效計算，例如在producer 生成資料時，通過redistribution把資料分布出去，而consumer在消費這些資料時，也可以不考慮temp table的分布特性（share-disk）,重新做動态的virtual partitioning，實作更優的并行方案。
Load-Balanced Table Queues
在資料分發過程中，可以通過一些動态政策來防止data skew，使資料均勻分發，例如對于range distribution，可以通過預采樣的方式發現分布不均勻，進而動态調整range邊界。
總結
這是一篇很老的paper了，相信其中的很多技術和現在Oracle的政策有所不同，但其充分share-disk架構的靈活性，實作靈活的執行排程 + 負載均衡政策，還是有一些借鑒意義的。
還是要吐槽下Oracle的論文風格，一如既往的粗糙。。。