MatrixOne：HTAP數據庫中的OLAP設計
一、MatrixOne整體架構
MatrixOne早期的架構是一個典型的share nothing架構，數據存放在一個Multi Raft集群上面，數據的每一個切片存在一個Raft上面，不同的Raft Group之間的數據是完全沒有重疊的。
早期架構存在著一些無法解決的問題，比如在擴展性上，每擴展一個節點，就需要同時擴展存算的資源，因為計算和存儲沒有完全分開。而且每擴展一個節點，需要大量的數據遷移工作。另外因為每一份數據都要保存至少 3個副本，從擴展節點到完成的時間會非常久。在性能方面，Raft協議所包含的leader角色，容易造成熱點；在性能較差的存儲下，數據庫整體性能下降會超過預期；多種引擎各自用途不同，性能各異，無法有效應對HTAP場景。成本方面，數據保存3副本，隨節點規模增長，成本不斷攀升，云上版本更甚；只有高配存儲才能發揮數據庫的預期性能。
為了更好地滿足客戶的需求，我們升級了新的架構。
升級后的架構從整體上看，分為三個部分：
- 最上面是計算層，計算層里面的每一個單位是一個Compute Node (CN) 計算節點。
- 在計算層下面是數據層，由Data Node (DN) 組成。
- 再下面是File service，支持各種不同的文件系統。
下面分別詳細介紹一下每一部分。
- 先從最底下的File service說起。File service支持各種不同的文件系統，比如用戶自己的本地的磁盤，NFS、HDFS、或者對象存儲。File service對上層提供一個統一的接口，對用戶來說，他并不需要關心最底下的數據本身是存在什么樣的介質上面。
- 這邊還會有個log service，因為我們知道在file service插入數據的時候，只能一整塊一整塊地寫，尤其像S3，一個object非常大，不可能一行一行地寫，一般是積累到一定的量再往里面寫一個整塊，那么這些還沒形成整塊的部分，會放在一個單獨的log service里面。log service是一個Multi Raft的集群。已經形成整塊的部分的數據完整性和一致性是用S3或者HDFS的功能來保證。還沒有形成單個block的數據的一致性完整性則是通過Multi Raft group來保證。
- 再上面是存儲節點DN，存儲節點里只存放的是元數據信息，比如每個表會分成大的segment，每個segment又會分成很多小的block。比如用S3來存放數據，元數據里會存放每個block，存放對應的具體S3 object。還有其他元數據信息比如row map，或者是次級索引bloom filter這些信息也會存在DN上。這里可以看到有多個DN，它們是怎么分布的呢？比如一張表有一個主鍵，那么我們可以根據主鍵來做一個分區，目前是按hash來做分區，也可以按range來做都是沒有問題的。對于數據規模較小的客戶來說，存放元數據其實只需要一個DN就足夠了。
- 再上面就是計算節點CN，具體執行計算任務的節點。CN可以分成各種不同類型，比如專門做TP查詢用的，專門做AP查詢用的，專門做streaming用的，還有一些對用戶不可見的數據庫自己后臺的一些任務。
- 還有一個組件叫HA Keeper，跟ZooKeeper功能類似，在節點之間互相通知上線下線這些信息，維護整個集群的可用信息。
新的share storage的架構優勢在于：
- DN節點不保存任何數據，只保存元數據信息。這樣就不會讓單個DN成為瓶頸，如果需要做彈性擴縮容，比如DN要新增或者刪除一個節點。做數據遷移只需要交換一些元數據的信息，不需要把所有數據都做搬遷的操作，這大大簡化了數據遷移量，并提高了擴縮容的效率。
- 完整性和一致性主要是通過S3來保證，S3本身就具備這一功能，并且從成本上來看也是非常低的，比用戶自己去搭建Multi Raft集群的成本會低很多。
- 計算任務也可以很好地分離開，比如TP和AP的計算，可以放在不同的CN節點上去做。因為現在數據已經不跟DN綁定在一起，所以計算的節點也可以完全解耦。
我們來看一個典型的查詢里面讀數據的操作是怎樣的。數據查詢從一個CN節點開始，首先去訪問DN的信息，讀取元信息，判斷某張表在哪些block，甚至會先通過過濾條件用row map來做過濾，剩下的是實際上需要去讀的那些block信息，拿到之后直接去那個file service下面去讀取數據。
在這里DN節點只提供了元數據信息，基本上不會成為性能瓶頸。因為更多的數據是CN節點去直接向File service讀取的，CN節點上面還會維護一個metadata cache。假設這個數據新鮮度還沒有過期，甚至可以不需要訪問DN，直接去File service下面拉取數據。
最后是存儲引擎，TAE，T和A分別是指TP和AP，T和A表示它既有事務的能力也可以很好的處理分析性查詢，用同一套引擎同時支持AP和TP。
在結構上，我們實際上還是可以把它看成是基于列存，不同的列之間，數據仍然是單獨分開存放的，每一列會按照8192行分成一個小的block。對于大多數的數字類型的block，8192行數據可以在L1 cache里面直接裝下，在后面的批量計算的時候會對計算引擎比較友好。多個block會組成一個segment，segment的作用是假設這個表它有主鍵或者排序鍵的情況下，一個segment內部會通過排序鍵和主鍵去做排序，這樣數據存儲在每一個segment內部是保持有序的，但segment之間可能會有重疊。這跟LSM的存儲有些相似的地方。但如果我們之后做了partition功能，可能會把一個partition內的所有segment也去做一次compaction操作。即把它們重新拿出來，做一個歸變排序再放進去。
以上就是MatrixOne最新的架構。
二、MatrixOne OLAP引擎
MatrixOne的計算引擎分為四個部分：
- 第一個是parser：把一個SQL語句解析成一個AST樹。
- 第二個是planner：把AST樹轉化成一個邏輯計劃。
- 第三個是optimizer：把邏輯計劃通過各種優化器規則或者是通過一些基于代價估算的方式轉化成更好的邏輯計劃。
- 最后是execution：把具體的邏輯計劃轉化成可執行的pipeline，然后去具體的CPU上面執行。
語法解析器（Parser）
各大開源數據庫大多都不會去手寫一個parser,至少是用MySQL或者PG的parser。比如DuckDB就是直接使用Postgres的parser代碼。即使我們不直照搬，也可以用一些YACC的工具去生成一個parser。測試之后發現用YACC生成的parser，并不會成為性能瓶頸，它的耗時非常少，所以我們沒有必要去手寫一個parser。（除了ClickHouse的parser是手寫的）。parser生成AST樹之后，會通過邏輯計劃器，把AST樹轉換成一個可以執行的邏輯計劃。
邏輯計劃器（Planner）
邏輯計劃器主要包含兩個部分：Bind(Algebrizer) 和子查詢消除。因為我們并不支持像SQL Server一樣將子查詢轉換成apply join，或者像MySQL一樣完全從父查詢里面拿出一行，再帶入子查詢里面，把子查詢完整地執行一次?？紤]到在AP查詢的場景，這樣的一個執行計劃是不可接受的，就干脆完全不支持這種apply join的方式，所以我們在planner這一步，把子查詢的消除給做掉。
優化器（Optimizer）
優化器部分，通常會有一個RBO基于規則的優化，基于規則對大部分查詢已經是夠用的。因為優化器通常分為兩種，一種是減少數據IO，它會減少實際從磁盤文件系統讀取的數據量；還有一種是對于CPU，在計算的過程中減少計算的代價。下面會具體舉一些例子來說明MatrixOne是如何設計的。
（1）第一部分是減少IO
包括以下部分：
- 列裁剪，讀一張表，有很多列，但我實際上只用到其中一列，那么其它列是不用讀取的。
- 謂詞下推，就是把一些過濾條件直接推到讀取數據這一部分，這樣可以盡量少的讀取數據。
- 謂詞推斷，主要會影響TPC-H里面的Q7和Q19，在后面會再舉例詳細介紹。
- Runtime filter ，比如大表跟小表做join操作，在小表構建完hash表之后，可能hash表的計數非常小，這樣我們可以直接通過hash表里面不同的詞去大表里面通過runtime或者元數據信息進行過濾，這樣在運行時就大大減少了需要讀取的大表的block數量。
（2）第二部分是減少計算
它并不能減少實際從磁盤里面讀取的數據，但是會在計算過程中減少計算量，和減少中間結果的數據量：
join order的join定序。通常使用TPC-H做OLAP 的benchmark，join order會影響很多不同的查詢，如果join order做的不好，這些查詢都可能會以數量級的變慢。
聚合函數下推和上拉的操作。假設聚合函數是在一個join上面，如果是先做join，之后再去做聚合，那么在join這里，數據可能會膨脹的非常多。但是如果可以把聚合函數推到join下面去做，即在join之前先聚合，數據已經減少很多，這也可以大大的減少計算。
簡單介紹謂詞推斷：
謂詞下推是已經確定顯式的可以下推的一個位置。但謂詞推斷可能是需要做一些邏輯上的變化之后，才能得到一些新的謂詞，這些新的謂詞才可以下推下去。比如TPC-H Q19的過濾條件是三個很長的謂詞用or連接起來，通過觀察實際上這三個or里面有很多共同的部分，可以把共同的部分提取出來，變成右圖的樣子?？梢杂^察到，首先part這張表上面有一個可以下推的謂詞，lineitem這張表上面也有兩個可以下推的謂詞。這樣可以先把這兩個謂詞下推到每個表的table scan上面去。然后還多出來一個在part和lineitem上面用主鍵做連接的謂詞。如果原來不對這個執行計劃做優化直接去執行，可能會先做一個笛卡爾積，再去做過濾操作，這樣的效率會非常低?，F在我們可以把它變成一個join操作。
簡單介紹一下MatrixOne使用的join order算法：
在各大開源數據庫上，join order的算法實現主要包括貪心法和動態規劃。其中動態規劃有很多不同方法，也有很多論文可以參考。但是動態規劃存在一個問題，當表的數量稍微多一些，狀態搜索空間就會以指數級膨脹。比如StarRocks的文檔里面提到過，10個表以上的計算就沒辦法使用動態規劃來計算，只能使用貪心法來計算。
我們在MatrixOne里對這個問題做了思考，在大于10張表時，可以先用貪心的方法來做一個剪枝操作，讓搜索空間大大減小，在貪心法之后再做動態規劃。
貪心法分三步：
- 第一步是確定事實表和維度表，因為一般的OLAP查詢的數據通常會把表分成事實表和維度表，之間用維度表的主鍵做join。因此拿到一個查詢之后，我們可以把事實表和維度表給找出來。
- 下一步事實表先與其維度表join成子樹，因為事實表維度表之間始終是通過維度表的主鍵去做join。做這樣的join的結果的函數始終是不大于他本來輸入的函數，所以做這么一個join并不會造成很多OLAP開銷，不會造成數據膨脹。在做事實表維度表join的過程中，我們會考慮事實表先與過濾性好的維度表做join。就優化器而言，越早減少數據量是越好的。
- 最后一步是在子樹之間，像TPC DS的表，會有多個維度表，維度表互相之間都不是以主鍵來做join。那么在子樹之間，我們再使用經典的join order的算法，比如動態規劃等等。
這樣我們把Join order的算法從之前只能10表以下做動態規劃，擴展到10個事實表之下都可以使用動態規劃來做。
舉個例子，TPC-H Q5中有customer、 orders、lineitem、supplier、nation、region這么六張表，它是一個比較典型的星型結構。
我們將orders和region這兩個表標紅，因為它們上面有過濾條件，需要單獨考慮。然后對于本身的join條件，標上一些帶箭頭的線。這里從事實表到維度表，用維度表的主鍵做join的條件?？梢钥闯觯还灿?個條件都是用主鍵來做join，其中還有比較特殊的一個條件，用畫的虛線標記，它的兩邊互相都不是用主鍵來做的join。
這個join算法下一步還會有一個聯合優化，最后可以跟謂詞推斷聯合使用，形成一個新的優化。因為可以看到supplier和customer有一個join條件，supplier跟nation也有一個join條件，用的都是同一個類型T來做決定。我們可以推導出來nation和customer表之間是以類型key做為join的條件，因此我們用黃色的選項表示。
最后實際生成的最優的join順序是從region開始，先跟nation表join，再跟customer表join，再跟orders 表join，再和lineitem表 join，最后跟supplier表 join。這是我們新推斷出來的條件。那么為什么是走這么一個路線，而不是先把上面這一條路join完成后再join下面這邊的表，是因為我們考慮到orders，region這兩張表都有過濾條件，放在一起過濾效果會更好。這樣會讓lineitem這張表的行數減少的速率更快。
執行器（Execution）
最后是執行器部分，從邏輯計劃轉化到實際可執行的pipeline，執行器的好壞對OLAP系統的執行效率影響是非常大的，下面會做詳細介紹。
眾所周知，執行器有一個經典的火山模型。對于每一行它是一個典型的pull模型，從最上層的計算的operator開始，每次調用一個NEXT函數從下面的節點去拿一行新的數據出來，做完計算之后，再等待更上層的那個計算節點去調用next從它這里取走。
火山模型存在一些問題，首先它并沒有做并行化，而是一行一行地處理；而且每一行在不同層級之間做一次調用，實際上會產生虛函數的開銷。因為next在不同編程語言肯定是要做函數重載，就算是虛函數開銷也很有可能是比實際計算的開銷還大，在整體開銷上會占相當的比例；同時對緩存也是非常不友好的，因為一行數據會跑多個不同的operation，可能在取下一行的時候，原來的緩存已經被清洗掉了。
MatrixOne的執行器是基于push模型，可以把幾個連續的operator組成一個流水線，而且流水線里面流動的數據，并不是一行一行的數據，而是前面提到的TAE存儲引擎里面的一個block，包含8192行數據，對于一般的數字類型是可以直接放進L1 Cache里面的，對緩存非常友好。每一個operator每一次要處理完這8192行才會喂給下一個operator，再加上調度是從最下面的，實際讀取每一張表的table scan 那個節點開始，往上面push。
對于push模型，是以數據為中心而不是以operator為中心，它的生成過程是對上一步的planner和optimizer生成的邏輯計劃，作一個后續遍歷，后續遍歷之后就可以得到一個基礎的pipeline結構，這個基礎的pipeline結構還沒有帶上比如每臺機器有多少個CPU或者需要在多少個CN節點上去執行這樣的信息。在后面實際執行的時候，再動態地根據這些基礎信息去做擴展。
舉個簡單的例子，假設一個簡單的查詢，有R、S、T三張表做join，其中R表是最大的一個表，S表和T表相對比較小，并且每個表都有過濾條件。對于一個典型的hash join，我們會把S和T這兩張表去構建hash表，然后R表在這兩個hash表上面依次去做探測（Probe）操作，得到join之后的數據。這么一個邏輯計劃至少需要插上三個pipeline，S表的讀取數據，做完過濾之后再建完hash表就在這里終止，T表也是在建完hash表之后就在join算子上面終結掉。但是最大的R表它始終是要做probe的，這張表的pipeline就可以往上走很多步。比如先做完過濾，再跟S表join之后，仍然以批量數量。然后這個批會繼續往下走，在下一個join中，仍然在同一個pipeline的下一個operator里面，再跟T表做一個join。所以一個3表join通常會拆成3個pipeline出來。
右圖還包括了數據并行的信息，比如S表可能會使用go語言里的三個協程并行的去讀數據，再做合并操作，合并完之后構建hash表，T表也是用三個協程去并行的讀，讀完之后，然后送到這里的構建hash表。R表因為比較大，pipeline會展開出更多的實際pipeline出來。我們可以看到就是R表這個pipeline是不會被阻斷的，通過hash operator之后，會繼續進到下一個join節點。
如上圖，我們的pipeline提供了這些算子，比較典型的有聚合、分組和各種join操作。這里把merge和connector、dispatch的顏色標識成不一樣，因為它們和其它operator的區別是其它算子都是只能在一個pipeline的中間，接受的數據是從上一個算子傳過來，發送的數據就直接發送給下一個算子去做后續的計算。而標成灰色的這一部分是在一個pipeline的數據的source或者sink，即入口或者出口的地方，比如merge它會去其他的pipeline去接收數據，把所有pipeline的數據合并成一個，返回給用戶。同理group by或者order by算子，也會執行merge操作。
發送也分為兩類：
- connector算子是一對一的發送。
- dispatch算子是一對多的發送。
dispatch會有很多不同的模式：
- 一種是廣播的模式，假設S表是一個很小的表，構建完hash表之后，會把hash表廣播到不同的pipeline出來去做計算；
- 一種是做shuffle，假如S表和R表都比較大，因此要做shuffle join，那么直接會通過一個shuffle dispatch算子把數據發送到不同的對應的一個pipeline上面去。
對于OLAP系統，從語義上來說通常跟SQL本身沒什么關系。但是OLAP的分析性查詢會是比較復雜的計算任務，有一些SQL能力是必須具備的，比如多表join、子查詢、窗口函數，還有CTE 和Recursive CTE，以及用戶自定義函數等。MatrixOne目前已經具備這些能力。
