## 課程目標(biāo)

這節(jié)課主要講的內(nèi)容，是如何并行執(zhí)行這些查詢

## 為什么要關(guān)心并行執(zhí)行?

因?yàn)楝F(xiàn)代 CPU 和 GPU 都有很多可用的 Core，利用多核可以有效提升性能(雖然并不絕對(duì))

同時(shí)也有更高的吞吐量，而這意味著，每秒可以執(zhí)行更多的查詢，能處理的數(shù)據(jù)更多，同時(shí)延遲更小

最終，整個(gè)數(shù)據(jù)庫(kù)系統(tǒng)能對(duì)請(qǐng)求進(jìn)行更快的響應(yīng)

TCO(Total Cost of Ownership)是用來(lái)考慮一個(gè)數(shù)據(jù)庫(kù)系統(tǒng)成本方面的術(shù)語(yǔ)，它指的不僅僅是購(gòu)買機(jī)器、購(gòu)買軟件許可證之類的成本，它指的是某一段時(shí)間內(nèi)，運(yùn)行這個(gè)數(shù)據(jù)庫(kù)所要花的成本。如果能用更少的硬件來(lái)做更多的工作，那么 TCO 這個(gè)成本將會(huì)降一大截

為什么需要多核，就是能省機(jī)器，也就是省錢!

## 并行和分布式的區(qū)別

相同點(diǎn): 都是將一個(gè)數(shù)據(jù)庫(kù)分散到多個(gè) **資源**(多臺(tái)機(jī)器、CPU、磁盤等) 上，這能夠改善數(shù)據(jù)庫(kù)系統(tǒng)的各個(gè)方面，比如性能、成本、延遲等

不同點(diǎn):?

- 并行

? ? 1. 可用資源在物理意義上彼此相鄰(比如多個(gè) CPU 之間通過(guò)高速總線進(jìn)行通信，這個(gè)距離很短)

? ? 2. 資源通信速度快、成本低，可靠性高(數(shù)據(jù)通信時(shí)不會(huì)丟失)

- 分布式

? ? 1. 資源間的距離可以很遠(yuǎn)

? ? 2. 資源間的通信，因?yàn)榫嚯x的原因，就得使用一個(gè)速度比較慢的通信信道(比如公共廣域網(wǎng)絡(luò))

? ? 3. 又因?yàn)槭褂昧司W(wǎng)絡(luò)協(xié)議做數(shù)據(jù)通信，所以可靠性不能保證，其他還有一些不能忽視的問(wèn)題比如成本之類

## Process Model

Process Model 就是一種通過(guò)組織系統(tǒng)，調(diào)度多個(gè) worker 來(lái)處理并發(fā)的請(qǐng)求方式

為什么需要調(diào)度多個(gè) worker ?一個(gè)應(yīng)用程序可能會(huì)發(fā)送一個(gè)很大的請(qǐng)求，或者同一時(shí)間發(fā)送了多個(gè)請(qǐng)求，給數(shù)據(jù)庫(kù)系統(tǒng)，這個(gè)時(shí)候就可以把這個(gè)請(qǐng)求打散掉，然后分配給多個(gè) worker 去處理

> 這里的 worker 就是一個(gè)任務(wù)的抽象，worker 可以是一個(gè)進(jìn)程也可以是一個(gè)線程，也可以是一個(gè)查詢優(yōu)化器的執(zhí)行任務(wù)

有 3 種 Process Model

### Process Per Worker

使用單個(gè) OS 進(jìn)程作為一個(gè) worker 使用

上圖就是這個(gè) Model 的模型，最左邊是數(shù)據(jù)庫(kù)，最右邊是應(yīng)用發(fā)出的查詢，OS 通過(guò)調(diào)度器，使用了一個(gè)進(jìn)程作為 worker 連接數(shù)據(jù)庫(kù)，然后用這個(gè) worker 處理查詢

但是這里有一個(gè)問(wèn)題，如果是多個(gè) worker 的情況，那么有可能，同一個(gè) page 會(huì)有多個(gè)副本保存在多個(gè) worker 代表的進(jìn)程中，這樣就會(huì)造成內(nèi)存的浪費(fèi)

那么如何解決?

可以使用共享內(nèi)存，這種方式可以允許這些不同進(jìn)程在內(nèi)存中，擁有各自獨(dú)立的地址空間，那么這些進(jìn)程，就可以共享這些全局?jǐn)?shù)據(jù)了

多進(jìn)程好處是，一個(gè)進(jìn)程掛了不影響其他進(jìn)程工作

為什么上述老牌數(shù)據(jù)庫(kù)系統(tǒng)都采用這種方式? 因?yàn)楫?dāng)時(shí)線程的 API 標(biāo)準(zhǔn)還沒(méi)出來(lái)(盡管各個(gè)操作系統(tǒng)里面都有線程)，為了最大的兼容性以及最小的代碼修改，就采用了這種方式，因?yàn)?fork 和 join 是當(dāng)時(shí)各個(gè) OS 的基本原語(yǔ)，也就是它們都支持進(jìn)程并且 API 相對(duì)一致，也就是同一個(gè)數(shù)據(jù)庫(kù)能跑在不同的 OS 中

### Process Pool

Process Pool 是 Process Per Worker 的改進(jìn)版本，關(guān)鍵的改變?cè)谟?，這里不會(huì)為進(jìn)來(lái)的每個(gè)連接都去創(chuàng)建一個(gè)進(jìn)程，而是里面會(huì)有一個(gè)進(jìn)程池

進(jìn)程池會(huì)分配進(jìn)程去處理查詢(同時(shí)進(jìn)程可以復(fù)用)，而不是每一個(gè)查詢進(jìn)來(lái)就 fork 一個(gè)去處理，因?yàn)槟菢哟鷥r(jià)很高

好處就是能做并行查詢，部分工作量可以分給其他進(jìn)程

> work-stealing，是指某些 worker 在處理數(shù)據(jù)時(shí)花費(fèi)了太多時(shí)間，還有一堆它需要處理的工作堆積在它要執(zhí)行的隊(duì)列中，那么調(diào)度器會(huì)從這個(gè) worker 手中拿走一些工作，交給其他 worker 處理

### Thread Per Worker

這是大多數(shù)現(xiàn)代操作系統(tǒng)的選擇，即一個(gè)線程就是一個(gè) worker

用一個(gè)進(jìn)程來(lái)運(yùn)行數(shù)據(jù)庫(kù)系統(tǒng)，在進(jìn)程里面的線程，根據(jù)需要去對(duì)任務(wù)進(jìn)行調(diào)度

上圖是這個(gè) Model 的模型

使用這種多線程模型才是正確的做法，這種方式非常省事，因?yàn)椴挥萌ス?OS 里面的共享內(nèi)存或者進(jìn)程管理之類的東西，并且上下文切換的開銷更低，因?yàn)檫M(jìn)程使用的是 OS 提供的共享內(nèi)存空間，很難做到保護(hù)。然后就是執(zhí)行任務(wù)會(huì)變得很快

## Scheduling

調(diào)度是決定將一個(gè)查詢拆分成多少任務(wù)，決定使用哪個(gè) CPU Core 來(lái)執(zhí)行這些任務(wù)，哪條線程要等其他相關(guān)線程執(zhí)行完之后再繼續(xù)往前執(zhí)行，的這么一個(gè)東西

并且，一旦某個(gè)線程執(zhí)行的任務(wù)生成來(lái)結(jié)果，這個(gè)結(jié)果應(yīng)該往哪里放，也是調(diào)度的內(nèi)容

## 并行查詢

分為 2 個(gè)

1. Inter-Query

? ? 可以在同一時(shí)間執(zhí)行多個(gè)做不同事情的查詢，能改善系統(tǒng)的吞吐量以及延遲

2. Intra-Query

? ? 將一個(gè)查詢拆分成多個(gè)子任務(wù)或者片段，然后在不同的資源上同時(shí)并行執(zhí)行這些任務(wù)

Inter-Query 的思想是，用多個(gè) worker(線程) 同時(shí)處理這些請(qǐng)求

但是也會(huì)有一個(gè)問(wèn)題，如果有多個(gè)線程同時(shí)對(duì)數(shù)據(jù)庫(kù)進(jìn)行更新操作，那么就要處理數(shù)據(jù)的沖突問(wèn)題，要加鎖

Intra-Query 適用于分析型查詢，利于并行計(jì)算。因?yàn)橛卸鄠€(gè) worker，它里面有多個(gè)線程或者 CPU Core，就可以使用它們來(lái)執(zhí)行同一個(gè)查詢

可以用 生產(chǎn)者-消費(fèi)者 模式來(lái)重新思考 Query Plan Tree 中的 operator 的并行算法

每個(gè) operator 不僅僅是數(shù)據(jù)的生產(chǎn)者(調(diào)用 next 生產(chǎn)一些數(shù)據(jù))，也是數(shù)據(jù)的消費(fèi)者(消費(fèi)在它下面的 operator 傳上來(lái)的數(shù)據(jù))

用多線程更新和構(gòu)建 hash table，用多線程來(lái)對(duì) hash table 進(jìn)行檢測(cè)

可以從當(dāng)前 operator 下面的 operator 傳上來(lái)的數(shù)據(jù)進(jìn)行分區(qū)，用多線程來(lái)處理這些數(shù)據(jù)分區(qū)

好處就是不需要對(duì)這些同時(shí)工作中的不同的 worker 進(jìn)行協(xié)調(diào)了

Intra-Query 也有 3 個(gè)方法

這 3 個(gè)方法并不是說(shuō)獨(dú)立使用，而是可以結(jié)合起來(lái)用

### 并行 Grace Hash Join

上圖是之前提到過(guò)的 Grace Hash Join 算法，那么怎么做并行呢

讓一個(gè) worker 對(duì)其對(duì)應(yīng)的一層 bucket 做處理，然后進(jìn)行 Join 操作，并生成輸出結(jié)果

麻煩的在于如何將這些輸出結(jié)果組合在一起

### Intra-Operator(Horizontal)

如果要對(duì)一張表進(jìn)行掃描，可以對(duì)其掃描生成結(jié)果后的多個(gè)實(shí)例，用不同的線程對(duì)不同的 fragment 進(jìn)行相同的處理，每個(gè)線程都會(huì)去掃描表的不同部分，然后將數(shù)據(jù)收集到一起

這里數(shù)據(jù)庫(kù)系統(tǒng)會(huì)使用一個(gè)叫做 exchange 的 operator 的玩意兒

這個(gè) exchange operator 會(huì)放在查詢計(jì)劃中的某個(gè)位置，提示數(shù)據(jù)庫(kù)系統(tǒng)，在這個(gè)位置，可以對(duì)這些 fragment 進(jìn)行并行處理，然后將處理完后的 fragment 結(jié)合到一起

例子

如圖所示，對(duì) A 進(jìn)行掃描，然后將結(jié)果傳入 filter operator

為了做到并行執(zhí)行查詢，可以將這個(gè)查詢計(jì)劃拆分為 3 個(gè)不同的 fragment，每個(gè) fragment 中都有一個(gè) scan operator 以及 filter operator

然后對(duì)數(shù)據(jù)庫(kù)進(jìn)行拆分，將它們拆分成多個(gè) page

然后在一個(gè)指定的 fragment 中，對(duì)不同的 page 進(jìn)行操作

對(duì)于 exchange operator，它會(huì)調(diào)用 Next，而后在 scan operator，它也會(huì)調(diào)用 Next，然后從一個(gè)特定的 page 上拿 data

對(duì)于其他的 fragment 也做同樣處理

exchange operator 的特征如下圖

1. Gather(收集)

? ? 將不同 worker 線程執(zhí)行任務(wù)所得到的結(jié)果，或者是這些 operator 生成的不同輸出進(jìn)行合并，合并成一個(gè)單一的輸出流，再向上傳給下一個(gè) operator

2. Repartition

? ? 有些時(shí)候需要將數(shù)據(jù)按區(qū)間進(jìn)行分塊，這樣就方便做并行掃描，然后將它們放進(jìn) repartition exchange operator 進(jìn)行處理，根據(jù)看到的值進(jìn)行拆分

3. Distribute

? ? 拿到 1 個(gè)輸入流，將它們拆分變成 2 個(gè)不同的輸出流

一個(gè)更復(fù)雜的例子

上圖中的查詢，經(jīng)過(guò)了

1. 對(duì) A 進(jìn)行掃描

2. 分配 3 個(gè)不同的 worker 來(lái)執(zhí)行工作

3. 在查詢計(jì)劃的 fragment 中，先對(duì) fragment 進(jìn)行 scan 操作，然后是 filter 操作，接著構(gòu)建 hash table

上面的 hash table 是一個(gè)由這 3 個(gè) fragment 共同維護(hù)的 hash table

然后這里會(huì)有一個(gè) exchange operator，只有等到 hash table 全部都得到更新之后，這個(gè) exchange operator 才會(huì)開始工作

然后對(duì) B 做類似的操作，這里通過(guò) partition 來(lái)拆分 B 的數(shù)據(jù)

然后再進(jìn)行 Join，可以用單線程也可以用多線程(這里使用多線程)

此時(shí)可以在 Join 里面將任務(wù)進(jìn)行拆分。同時(shí)也可以通過(guò)不同的線程來(lái)對(duì) B 那邊的 partition 進(jìn)行檢測(cè)

綜合來(lái)講，就是將數(shù)據(jù)進(jìn)行拆分，將拆分后的數(shù)據(jù)交給多個(gè)線程，由它們來(lái)執(zhí)行相同的邏輯，然后將這些線程執(zhí)行所得到的結(jié)果進(jìn)行合并。

最后，把這些過(guò)程整合為一個(gè)很大的樹形結(jié)構(gòu)，這樣就可以做并行執(zhí)行了

### Intra-Operator(Vertical)

對(duì)于每個(gè) operator tree 來(lái)說(shuō)，可以將每個(gè) operator 作為一個(gè)單獨(dú)的 worker 來(lái)執(zhí)行

例子

對(duì)于此處的 worker 來(lái)說(shuō)，可以使用一個(gè) CPU Core 或者一個(gè) worker 來(lái)執(zhí)行這個(gè) Join 操作，它只需要從它的子 operator 處獲取數(shù)據(jù)即可

Join 操作完成之后，結(jié)果會(huì)發(fā)送給另一個(gè) worker，上面這個(gè) operator 會(huì)拿到 Join operator 發(fā)送給它的數(shù)據(jù)，然后執(zhí)行條件判斷或者 projection 操作，最后將生成的結(jié)果進(jìn)一步發(fā)給查詢計(jì)劃樹

所以執(zhí)行這些 operator 的線程，都在自旋等待數(shù)據(jù)的傳入，也就是形成了 生產(chǎn)者-消費(fèi)者 模型

但是這里就會(huì)存在一個(gè)問(wèn)題，假設(shè) 2 個(gè)線程在訪問(wèn)同一個(gè)數(shù)據(jù)，那么它就會(huì)用到鎖，這樣就會(huì)導(dǎo)致，一個(gè)線程在讀時(shí)，另一個(gè)沒(méi)法寫，或者一個(gè)在寫時(shí)另一個(gè)沒(méi)法讀。

所以如果生成的 tuple 越多，這里就會(huì)越卡(但是可以用環(huán)形隊(duì)列 + 鎖 方式改善)

當(dāng)然還有一種方式是，在這個(gè) Join Operator 上使用水平并行，然后通過(guò)垂直并行將它和這個(gè) projection 操作整合在一起，即對(duì)于單個(gè)線程，高內(nèi)聚它的任務(wù)；對(duì)于多個(gè)線程，低耦合它們的任務(wù)

### Bushy Parallelism

一個(gè) inter-operator parallelism 的拓展版本

基本思路是，讓不同的 worker 在同一時(shí)刻對(duì)一個(gè)查詢計(jì)劃的不同部分進(jìn)行操作

這里依然使用 exchange operator 來(lái)在這些 operator 之間移動(dòng)數(shù)據(jù)

比如上圖這樣的 4-way Join 查詢

交給 worker 處理就是這樣，它們會(huì)并行處理這些 fragment，將處理完的結(jié)果向上傳遞給 exchange operator，然后再通過(guò)其他 worker 對(duì) exchange operator 后的結(jié)果進(jìn)行處理

## I/O Parallelism

如果處理請(qǐng)求時(shí)，硬盤 I/O 瓶頸始終存在，那么就只能使用 I/O parallelism 了

基本思想是，對(duì)數(shù)據(jù)庫(kù)系統(tǒng)的文件和數(shù)據(jù)進(jìn)行拆分，將它們分散到儲(chǔ)存設(shè)備上的不同位置處

有很多方法可以做到這一點(diǎn)，比如上圖中列出來(lái)的

### Multi-Disk Parallelism

這有點(diǎn)像是 RAID

> RAID: Redundant Arrays of Independent Disks，即?

獨(dú)立磁盤構(gòu)成的具有冗余能力的陣列

對(duì)系統(tǒng)進(jìn)行配置，讓多個(gè)存儲(chǔ)設(shè)備以單個(gè)邏輯設(shè)備的形式來(lái)供數(shù)據(jù)庫(kù)使用，可以用硬件也可以用軟件來(lái)做這個(gè)事情

一個(gè)關(guān)于 RAID 的例子，也被叫做 Stripping(數(shù)據(jù)分條技術(shù))

RAID 控制器會(huì)根據(jù) Round-Robin 策略來(lái)決定該往哪個(gè)存儲(chǔ)設(shè)備上寫數(shù)據(jù)，在它內(nèi)部會(huì)有一些元數(shù)據(jù)，這些數(shù)據(jù)保存了 page 在磁盤上的位置

另一種常見的做法就是做磁盤鏡像(Mirroring)，就是每個(gè)存儲(chǔ)設(shè)備上面都會(huì)保存一份該 page 的副本，可以通過(guò)某種糾錯(cuò)碼或者其他方式，來(lái)確保在一個(gè)磁盤掛掉的情況下，還可以通過(guò)其他沒(méi)掛掉的磁盤上保存的 page，進(jìn)行重新創(chuàng)建該 page

### Database Partitioning

將數(shù)據(jù)庫(kù)中的數(shù)據(jù)拆分為不相交的子集，然后將它們分配給離散的磁盤

同時(shí)可以設(shè)置軟鏈接/硬鏈接，來(lái)讓不同的目錄(一個(gè)目錄/文件代表一個(gè)數(shù)據(jù)庫(kù))指向不同的磁盤

log 文件保存的是對(duì)記錄所做的所有的修改，如果有多個(gè)不同的存儲(chǔ)設(shè)備，就需要對(duì) log 文件進(jìn)行分片

partition 的基本思想是，對(duì)于單個(gè)邏輯表，將該表的數(shù)據(jù)拆分成不相交的子集，然后將這些子集分別存放到不同的存儲(chǔ)設(shè)備上，對(duì)它們進(jìn)行管理

### Vertical Partitioning

這個(gè)垂直分區(qū)方式就是之前講的列式存儲(chǔ)方面的東西

假設(shè)有一個(gè)表，里面有 4 個(gè)屬性

那么可以拿走其中該表中的一個(gè)屬性，并將它保存在一個(gè)單獨(dú)的分區(qū)中，然后這個(gè)分區(qū)，會(huì)以一個(gè)單獨(dú)文件的形式，放在一個(gè)單獨(dú)的磁盤或者存儲(chǔ)設(shè)備上。雖然這種方式跟列存儲(chǔ)很相似，但是，如果這些列中的值都是相同的，Vertical Partitioning 并不會(huì)對(duì)這列的數(shù)據(jù)進(jìn)行壓縮處理

### Horizontal Partitioning

假設(shè)有一個(gè)表，里面有 4 個(gè)屬性

進(jìn)行水平分區(qū)后，一個(gè) tuple 中所有數(shù)據(jù)就會(huì)被放在一個(gè)分區(qū)中

## 結(jié)論