概述

Facebook Velox 是一個(gè)針對(duì) SQL 運(yùn)行時(shí)的 C++ 庫(kù)，旨在統(tǒng)一 Facebook 各種計(jì)算流，包括 Spark 和 Presto，使用推的模式、支持向量計(jì)算。

Velox 接受一棵優(yōu)化過(guò)的 PlanNode Tree，然后將其切成一個(gè)個(gè)的線(xiàn)性的 Pipeline，Task 負(fù)責(zé)這個(gè)轉(zhuǎn)變過(guò)程，每個(gè) Task 針對(duì)一個(gè) PlanTree Segment。大多數(shù)算子是一對(duì)一翻譯的，但是有一些特殊的算子，通常出現(xiàn)在多個(gè) Pipeline 的切口處，通常來(lái)說(shuō)，這些切口對(duì)應(yīng)計(jì)劃樹(shù)的分叉處，如 HashJoinNode，CrossJoinNode， MergeJoinNode ，通常會(huì)翻譯成 XXProbe 和 XXBuild。但也有一些例外，比如 LocalPartitionNode 和 LocalMergeNode 。

邏輯計(jì)劃翻譯成物理計(jì)劃，可調(diào)整 Pipeline 并發(fā)度

為了提高執(zhí)行的并行度，Velox 引入了 LocalPartitionNode 節(jié)點(diǎn)，可以將一個(gè) Pipeline 進(jìn)行多線(xiàn)程（每個(gè)線(xiàn)程一個(gè)實(shí)例）并行運(yùn)行，并且互斥的消費(fèi)數(shù)據(jù)。其中每個(gè)實(shí)例稱(chēng)為 Driver。該算子在輸入計(jì)劃樹(shù)里并沒(méi)有分叉（即沒(méi)有多個(gè) source），但在翻譯成物理算子時(shí)，會(huì)在此節(jié)點(diǎn)處進(jìn)行切開(kāi)，并在切口前后改變執(zhí)行的并行度，對(duì)應(yīng)的物理算子是LocalPartition  和  LocalExchange。

 調(diào)整并發(fā)度算子，一個(gè)邏輯算子翻譯成兩個(gè)物理算子

還有一個(gè)特殊節(jié)點(diǎn)，稱(chēng)為 LocalMergeNode，該對(duì)輸入有要求：必須有序，然后會(huì)進(jìn)行單線(xiàn)程的歸并排序，從而使輸出全局有序。也因此，由其而切開(kāi)的消費(fèi) Pipeline 一定是單 Driver 的。翻譯成算子，對(duì)應(yīng)兩個(gè) CallbackSink 和 LocalMerge。

Merge 算子，也是一種邏輯翻譯成兩種物理算子

總結(jié)一下，上述五個(gè) PlanNode，HashJoinNode，CrossJoinNode， MergeJoinNode ，LocalPartitionNode ，LocalMergeNode 在翻譯時(shí)會(huì)造成切口，即將邏輯 PlanTree 切成多個(gè)物理 Pipeline，因此在切口處會(huì)將一個(gè)邏輯算子翻譯成多個(gè)物理算子，分到不同 Pipeline 上。每個(gè) Pipeline 會(huì)有一個(gè)從 0 開(kāi)始的編號(hào)：Pipeline ID，是全局粒度的。

并且，可以由 LocalPartitionNode 來(lái)按需改變每個(gè) Pipeline 并行度，其中 Pipeline 的每個(gè)線(xiàn)程由一個(gè) Driver 來(lái)執(zhí)行。每個(gè) Driver 也有一個(gè)從 0 開(kāi)始的編號(hào)：Driver ID，是 Pipeline 粒度的。

其他 PlanNode 到算子的翻譯基本都是一對(duì)一的，感興趣的可以看官方文檔的這個(gè)頁(yè)面：Plan Nodes and Operators。

下面展開(kāi)一些細(xì)節(jié)。

Splits

Velox 允許應(yīng)用層（即 Velox 的使用方）以 Splits （每個(gè)算子的輸入片段稱(chēng)為 Split）的方式給 Pipeline 喂數(shù)據(jù)，可以流式的喂，因此有兩個(gè) API：

1. Task::addSplit(planNodeId, split) ：喂一份數(shù)據(jù)給 Velox
2. Task::noMoreSplits() ：通知 Velox 我喂完了。

Velox 會(huì)使用一個(gè)隊(duì)列在緩存這些 Splits 數(shù)據(jù)。在數(shù)據(jù)喂完之前的任意一個(gè)時(shí)刻，Pipeline 的葉子算子（對(duì)的，外部喂數(shù)據(jù)只能發(fā)生在葉子節(jié)點(diǎn)，如 TableScan，Exchange 和 MergeExchange）都可以從隊(duì)列中取數(shù)據(jù)，對(duì)應(yīng) API 是 Task::getSplitOrFuture(planNodeId) ，返回值有兩種：

1. 如果隊(duì)列中有數(shù)據(jù)，則返回一個(gè) Split
2. 如果隊(duì)列中無(wú)數(shù)據(jù)，但還沒(méi)有收到喂完的信號(hào)，則返回一個(gè) Future （類(lèi)似于一個(gè)欠條，之后有數(shù)據(jù)之后，會(huì)憑該欠條兌付）。

Task 是 PlanTree Segment 執(zhí)行單位，可以通過(guò) Splits 方式流式喂數(shù)據(jù)

Join Bridges and Barriers

Join （HashJoinNode 和 CrossJoinNode）會(huì)翻譯成 XXProbe 和 XXBuild 兩個(gè)算子，并且通過(guò)一個(gè)共享的 Bridge 來(lái)溝通數(shù)據(jù)，兩側(cè) Pipeline 都可以通過(guò) Task::getHashJoinBridge() 函數(shù)來(lái)根據(jù) PlanNodeId 獲取該共享的 Bridge。

為了提高 build 速度，build 側(cè) Pipeline 通常使用多個(gè) Driver 并發(fā)執(zhí)行。但由于只有一個(gè) Bridge，每個(gè) Driver 在結(jié)束時(shí)可以調(diào)用 Task::allPeersFinished() （內(nèi)部是使用一個(gè) BarrierState 的結(jié)構(gòu)來(lái)實(shí)現(xiàn)的）來(lái)判斷自己是否為最后一個(gè) Driver，如果是，則將所有 Driver 的輸出進(jìn)行合并后送到 Bridge。

當(dāng)然，在 RIGHT and FULL OUTER join 情況下，Probe 側(cè)也需要將沒(méi)有 match 上的數(shù)據(jù)喂給 Bridge，此時(shí)也需要由最后一個(gè) Driver 來(lái)負(fù)責(zé)這件事，于是同樣需要調(diào)用  Task::allPeersFinished() 函數(shù)。

使用 Bridge 對(duì) Join 兩側(cè) Pipeline 進(jìn)行數(shù)據(jù)橋接（Build->Probe）

下面來(lái)詳細(xì)看下 Join 類(lèi)算子的切分細(xì)節(jié)。以 HashJoin 為例，Task 在切分 PlanTree 時(shí)，會(huì)將邏輯上的一個(gè) HashJoin 算子，轉(zhuǎn)化成物理上的一對(duì)算子：HashProbe 和 HashJoin，并且使用異步機(jī)制進(jìn)行通知：在 HashJoin 完成后，通知 HashProbe 所在 Pipeline 繼續(xù)執(zhí)行，在此之前，后者是阻塞等待的。

 Join 兩側(cè) Pipeline 是可以調(diào)整并發(fā)度的

如上圖，每個(gè) Pipeline 在實(shí)例化（邏輯 PlanNode 轉(zhuǎn)物理 Operator）的時(shí)候，可以生成多份，進(jìn)行并發(fā)執(zhí)行，互斥的消費(fèi)數(shù)據(jù)。并且，每個(gè) Pipeline 的并行粒度可以不一樣，如上圖 Probe Pipeline 實(shí)例化了兩份，而 Build Pipeline 實(shí)例化了三份。并且，Build Pipeline 組中最后一個(gè)運(yùn)行完的 Pipeline 負(fù)責(zé)將數(shù)據(jù)通過(guò) Bridge 發(fā)送給 Probe Pipeline。

Exchange Clients

Velox 使用 Exchange Clients 來(lái)獲取遠(yuǎn)程 worker 的數(shù)據(jù)。分兩個(gè)步驟：

第一步，Pipeline 中第一個(gè) Driver （driverId == 0） 的 Exchange 算子從 Task 中獲取一個(gè) Split，并且初始化一個(gè)共享 Exchange Client。

第二步，Exchange Client 會(huì)為上游每個(gè) Task 構(gòu)造一個(gè) Exchange Source，并行的拉取每個(gè)上游 Task 同一個(gè) Partition （圖中是 Partition-15）數(shù)據(jù)，然后將其放在 Client 的隊(duì)列 Queue 中。Exchange 的每個(gè) Driver 都會(huì)去隊(duì)列中拉取這些數(shù)據(jù)。

如何從上游 Task 拉取數(shù)據(jù)的邏輯，需要由用戶(hù)自定義實(shí)現(xiàn)  ExchangeSource 和 ExchangeSource::Factory  。每個(gè) ExchangeSource 接受一個(gè)上游 Task 的字符串 ID、Partition 編號(hào)和一個(gè)隊(duì)列作為參數(shù)。然后會(huì)從上游 Task 中拉取該 Partition 的數(shù)據(jù)，并且放到隊(duì)列中。

 向上游 Task 遠(yuǎn)程（跨進(jìn)程）拉取數(shù)據(jù)，也叫 MaterializePage

Local Exchange Queues

Local exchange 用于在一個(gè) Task 內(nèi)部調(diào)整數(shù)據(jù)并發(fā)度，會(huì)被翻譯成兩個(gè)物理算子：LocalPartition 和 LocalExchange。其中，LocalPartition 在生產(chǎn)側(cè) Pipeline，LocalExchange 在消費(fèi)側(cè) Pipeline。

中間通過(guò) LocalExchangeQueues 來(lái)溝通生產(chǎn)者和消費(fèi)者，這些隊(duì)列在 Task 類(lèi)中。對(duì)于每個(gè)消費(fèi)者（也即 LocalExchange 側(cè) Driver）Task 都會(huì)構(gòu)建一個(gè) LocalExchangeQueue 隊(duì)列；每個(gè)生產(chǎn)者 （LocalPartition）可以訪(fǎng)問(wèn)所有隊(duì)列。在產(chǎn)生一條數(shù)據(jù)是，會(huì)對(duì)其按照某種方式進(jìn)行 Partition，然后寫(xiě)到對(duì)應(yīng)隊(duì)列中。這個(gè)過(guò)程類(lèi)似于 MapReduce 中的 Shuffle 階段。

 本地改變并發(fā)度時(shí)，使用一個(gè)隊(duì)列進(jìn)行數(shù)據(jù)溝通

具體來(lái)說(shuō)，Local Exchange 可以有幾種方式改變并行度。如一改多、多改一。多改一，典型的例子如，并行 sort：先切成多個(gè)分片每個(gè)分片分別 sort，后通過(guò) Local Exchange 進(jìn)行 merge sort。不僅單個(gè) Pipeline 的多個(gè) Driver 在進(jìn)行數(shù)據(jù)合并時(shí)可以用 Local Exchange，多個(gè) Pipeline 的合并也可以用 Local Exchange，不妨稱(chēng)之為多并一。典型例子有，Union All，將多個(gè)數(shù)據(jù)集合并起來(lái)。

 多改一 

 多并一

一改多通常用在，在經(jīng)歷了某些必須使用單線(xiàn)程的算子后（比如一些 Shuffle 算子），重新對(duì)數(shù)據(jù)分片提高并發(fā)度，使用多線(xiàn)程運(yùn)行。

 一改多

Local Merge Sources

LocalMerge 算子和 LocalExchange 算子類(lèi)似，但對(duì)并發(fā)數(shù)和輸入都有限定。其所在 Pipeline 只會(huì)單線(xiàn)程運(yùn)行，但會(huì)接受多線(xiàn)程運(yùn)行的 Pipeline 的輸入。并且要求所有輸入有序，然后將輸入進(jìn)行歸并，保證輸出是有序的。

LocalMerge 算子通過(guò) Task::getLocalMergeSources() 來(lái)獲取所有待 Merge 的 sources。因此，每個(gè) LocalMergeNode 會(huì)初始化給定并發(fā)數(shù)個(gè) LocalMergeSource。

Merge Join Sources

MergeJoin 算子提供了某種接受右側(cè)輸入的方法。Task 會(huì)在右側(cè) Pipeline 增加一個(gè) CallbackSink 算子，來(lái)匯集數(shù)據(jù)。左側(cè)算子可以通過(guò) Task::getMergeJoinSource() 接口來(lái)獲取該 CallbackSink 的輸出。

擴(kuò)展性

Velox 允許用戶(hù)自定義 PlanNode 和 Operator，以及 Join 相關(guān)的 Operator 和 Bridge。自定義 Operator 可以訪(fǎng)問(wèn) task 中的 splits 并使用 barriers。

但 Exchange clients, local exchange queues 和 local merge sources、 merge join sources 等狀態(tài)由于不是通用的，因此訪(fǎng)問(wèn)不了。

總結(jié)

小節(jié)一下，Task 負(fù)責(zé)將由 PlanNode 組成的 PlanTree 翻譯成由 Operator 組成的 Pipeline，并且對(duì) Pipeline 進(jìn)行并發(fā)運(yùn)行。在此期間，Task 會(huì)維護(hù) Operator 間的共享狀態(tài)、協(xié)調(diào) Operator 間的運(yùn)行依賴(lài)。這些共享狀態(tài)包括：

• Splits
• Join bridges and barriers
• Exchange clients
• Local exchange queues
• Local merge sources
• Merge join sources

上述的每個(gè)狀態(tài)都是和特定 PlanNode 關(guān)聯(lián)的（即不是全局范圍的，而是和 PlanNode 綁定的），因此 Opeator 需要使用 PlanNodeID 來(lái)訪(fǎng)問(wèn)相關(guān)狀態(tài)。前兩個(gè)狀態(tài)是所有算子都有的，因此自定義算子可以訪(fǎng)問(wèn)到，后幾個(gè)狀態(tài)是某些算子特有的，因此自定義算子訪(fǎng)問(wèn)不到。