目錄

• 背景 & 目標(biāo)
• Doris 數(shù)據(jù)劃分
• Partition
• Bucket
• Join 方式
• 總覽
• Broadcast / Shuffle Join
• Bucket Shuffle Join
• Plan Rule
• Colocate Join
• Runtime Filter 優(yōu)化
• Join Reorder 優(yōu)化
• Join 調(diào)優(yōu)建議

背景 & 目標(biāo)

• 掌握 Apache Doris Join 優(yōu)化手段及其實(shí)現(xiàn)原理
• 為代碼閱讀提供理論基礎(chǔ)

Doris 數(shù)據(jù)劃分

不同的 Join 方式非常依賴(lài)于對(duì) Doris 中數(shù)據(jù)劃分方式的透徹理解。因此先在這里列舉出必要的基礎(chǔ)知識(shí)。

首先，在 Doris 中數(shù)據(jù)都以表（Table）的形式進(jìn)行邏輯上的描述。

在 Doris 的存儲(chǔ)引擎中，用戶(hù)數(shù)據(jù)被水平劃分為若干個(gè)數(shù)據(jù)分片（Tablet，也稱(chēng)作數(shù)據(jù)分桶 Bucket）。每個(gè) Tablet 包含若干數(shù)據(jù)行。各個(gè) Tablet 之間的數(shù)據(jù)沒(méi)有交集，并且在物理上是獨(dú)立存儲(chǔ)的。

一個(gè) Tablet 只屬于一個(gè)數(shù)據(jù)分區(qū)（Partition）。而一個(gè) Partition 包含若干個(gè) Tablet。因?yàn)?Tablet 在物理上是獨(dú)立存儲(chǔ)的，所以可以視為 Partition 在物理上也是獨(dú)立的。Tablet 是數(shù)據(jù)移動(dòng)、復(fù)制等操作的最小物理存儲(chǔ)單元。

若干個(gè) Partition 組成一個(gè) Table。Partition 可以視為是邏輯上最小的管理單元。數(shù)據(jù)的導(dǎo)入與刪除，僅能針對(duì)一個(gè) Partition 進(jìn)行。

Doris 支持兩層的數(shù)據(jù)劃分。第一層是 Partition，支持 Range 和 List 的劃分方式。第二層是 Bucket（Tablet），僅支持 Hash 的劃分方式。也可以?xún)H使用一層分區(qū)。使用一層分區(qū)時(shí)，只支持 Bucket 劃分。

下圖說(shuō)明 Table、Partition、Bucket（Tablet） 的關(guān)系：

• Table 按照 Range 的方式按照 date 字段進(jìn)行分區(qū)，得到了 N 個(gè) Partition
• 每個(gè) Partition 通過(guò)相同的 Hash 方式將其中的數(shù)據(jù)劃分為 M 個(gè) Bucket（Tablet）
• 從邏輯上來(lái)說(shuō)，Bucket 1 可以包含 N 個(gè) Partition 中劃分得到的數(shù)據(jù)，比如下圖中的 Tablet 11、Tablet 21、Tablet N1

特別注意：

Doris 中的 Partition 和 Bucket 定義可能和某些其它數(shù)據(jù)庫(kù)系統(tǒng)的定義有一些差異，下面配以一個(gè)具體的建表語(yǔ)句為例來(lái)說(shuō)明：

CREATE TABLE IF NOT EXISTS example_db.expamle_range_tbl(    `user_id` LARGEINT NOT NULL COMMENT "用戶(hù)id",    `date` DATE NOT NULL COMMENT "數(shù)據(jù)灌入日期時(shí)間",    `timestamp` DATETIME NOT NULL COMMENT "數(shù)據(jù)灌入的時(shí)間戳",    `city` VARCHAR(20) COMMENT "用戶(hù)所在城市",    `age` SMALLINT COMMENT "用戶(hù)年齡",    `sex` TINYINT COMMENT "用戶(hù)性別",    `last_visit_date` DATETIME REPLACE DEFAULT "1970-01-01 00:00:00" COMMENT "用戶(hù)最后一次訪(fǎng)問(wèn)時(shí)間",    `cost` BIGINT SUM DEFAULT "0" COMMENT "用戶(hù)總消費(fèi)",    `max_dwell_time` INT MAX DEFAULT "0" COMMENT "用戶(hù)最大停留時(shí)間",    `min_dwell_time` INT MIN DEFAULT "99999" COMMENT "用戶(hù)最小停留時(shí)間")ENGINE=OLAPAGGREGATE KEY(`user_id`, `date`, `timestamp`, `city`, `age`, `sex`)PARTITION BY RANGE(`date`)(    PARTITION `p201701` VALUES LESS THAN ("2017-02-01"),    PARTITION `p201702` VALUES LESS THAN ("2017-03-01"),    PARTITION `p201703` VALUES LESS THAN ("2017-04-01"))DISTRIBUTED BY HASH(`user_id`) BUCKETS 16PROPERTIES(    "replication_num" = "3");

綠色高亮：Partition，此例中使用一個(gè) date 字段進(jìn)行分區(qū)

藍(lán)色高亮：Bucket，此例中使用 user_id 字段為作為分布列

Partition

• Partition 列可以指定一列或多列，分區(qū)列必須為 KEY 列
• 分區(qū)數(shù)量理論上沒(méi)有上限
• 當(dāng)不使用 Partition 建表時(shí)，系統(tǒng)會(huì)自動(dòng)生成一個(gè)和表名同名的，全值范圍的 Partition。該 Partition 對(duì)用戶(hù)不可見(jiàn)，并且不可刪改

創(chuàng)建分區(qū)時(shí)不可添加范圍重疊的分區(qū)

有兩種分區(qū)方式：

分區(qū)方式一般用法Range通常按時(shí)間分區(qū)，以方便地管理新舊數(shù)據(jù)List支持的類(lèi)型更豐富，分區(qū)值為枚舉值。只有當(dāng)數(shù)據(jù)為目標(biāo)分區(qū)枚舉值其中之一時(shí)，才可以命中分區(qū)

Bucket

• 如果使用了 Partition，則 DISTRIBUTED 語(yǔ)句描述的是數(shù)據(jù)在各個(gè)分區(qū)內(nèi)的劃分規(guī)則。如果不使用 Partition，則描述的是對(duì)整個(gè)表的數(shù)據(jù)劃分規(guī)則
• 分桶列的選擇，是在 查詢(xún)吞吐 和 查詢(xún)并發(fā) 之間的一種權(quán)衡：
• 如果選擇多個(gè)分桶列，則數(shù)據(jù)分布更均勻。如果一個(gè)查詢(xún)條件不包含所有分桶列的等值條件（意味著無(wú)法做桶裁剪以減少數(shù)據(jù)查詢(xún)范圍），那么該查詢(xún)會(huì)觸發(fā)所有分桶同時(shí)掃描，這樣查詢(xún)的吞吐會(huì)增加，單個(gè)查詢(xún)的延遲隨之降低。這個(gè)方式適合大吞吐低并發(fā)的查詢(xún)場(chǎng)景
• 如果僅選擇一個(gè)或少數(shù)分桶列，則對(duì)應(yīng)的點(diǎn)查詢(xún)可以?xún)H觸發(fā)一個(gè)分桶掃描（意味著可以做桶裁剪以減少數(shù)據(jù)查詢(xún)范圍）。此時(shí)，當(dāng)多個(gè)點(diǎn)查詢(xún)并發(fā)時(shí)，這些查詢(xún)有較大的概率分別觸發(fā)不同的分桶掃描，各個(gè)查詢(xún)之間的 IO 影響較小，尤其當(dāng)不同桶分布在不同磁盤(pán)上時(shí)），所以這種方式適合高并發(fā)的點(diǎn)查詢(xún)場(chǎng)景
• 分桶的數(shù)量理論上沒(méi)有上限

Join 方式

總覽

作為分布式的 MPP 數(shù)據(jù)庫(kù)， 在 Join 的過(guò)程中是需要進(jìn)行數(shù)據(jù)的 Shuffle。數(shù)據(jù)需要進(jìn)行拆分調(diào)度，才能保證最終的 Join 結(jié)果是正確的。舉個(gè)簡(jiǎn)單的例子，假設(shè)關(guān)系 S 和 R 進(jìn)行Join，N 表示參與 Join 計(jì)算的節(jié)點(diǎn)的數(shù)量；T 則表示關(guān)系的 Tuple 數(shù)目。

目前 Doris 支持的 Join 方式有以上 4 種，這 4 種方式靈活度和適用性是從高到低的，對(duì)數(shù)據(jù)分布的要求越來(lái)越嚴(yán)，但 Join 計(jì)算的性能則通過(guò)降低網(wǎng)絡(luò)開(kāi)銷(xiāo)而越來(lái)越好。

Join 方式的選擇是 FE 生成分布式計(jì)劃階段會(huì)考慮的事項(xiàng)之一。在 FE 進(jìn)行分布式計(jì)劃時(shí)，優(yōu)先選擇的順序?yàn)椋偸菚?huì)優(yōu)先選擇預(yù)期性能最好的）：Colocate Join -> Bucket Shuffle Join -> Broadcast Join -> Shuffle Join。

Colocate 以及 Bucket Shuffle 是可遇不可求的。當(dāng)無(wú)法使用它們時(shí)，Doris會(huì)自動(dòng)嘗試進(jìn)行 Broadcast Join，如果預(yù)估小表過(guò)大則會(huì)自動(dòng)切換至 Shuffle Join。

但是用戶(hù)可以通過(guò)顯式 Hint 來(lái)強(qiáng)制使用期望的 Join 類(lèi)型，比如：

select * from test join [shuffle] baseall on test.k1 = baseall.k1;

Broadcast / Shuffle Join

原理比較簡(jiǎn)單，這里不展開(kāi)。

Bucket Shuffle Join

當(dāng) Join 條件命中了左表的數(shù)據(jù)分布列時(shí)，Broadcast 以及 Shuffle Join 會(huì)有非必要的網(wǎng)絡(luò)傳輸開(kāi)銷(xiāo)。而 Bucket Shuffle Join 旨在解決這類(lèi)問(wèn)題，通過(guò)對(duì)左表實(shí)現(xiàn)本地性計(jì)算優(yōu)化，來(lái)減少左表數(shù)據(jù)在節(jié)點(diǎn)間的傳輸耗時(shí)，從而加速查詢(xún)。

以上的例子中，Join 的等值表達(dá)式命中了表 A（左表）的數(shù)據(jù)分布列。Bucket Shuffle Join 會(huì)根據(jù)表 A 的數(shù)據(jù)分布信息，將表 B（右表）的數(shù)據(jù)發(fā)送到對(duì)應(yīng)表 A 的數(shù)據(jù)計(jì)算節(jié)點(diǎn)。

定性分析上：

• 降低了網(wǎng)絡(luò)與內(nèi)存開(kāi)銷(xiāo)（相比 Broadcast 以及 Shuffle Join 都不會(huì)更差），使一類(lèi) Join 查詢(xún)有更好的性能。尤其是當(dāng) FE 能夠執(zhí)行左表的分區(qū)裁剪與桶裁剪時(shí)
• 與 Colocate Join 不同，它對(duì)于表的數(shù)據(jù)分布方式?jīng)]有侵入性，對(duì)于用戶(hù)來(lái)說(shuō)是透明的。對(duì)于表的數(shù)據(jù)分布沒(méi)有強(qiáng)制性的要求（體現(xiàn)在建表語(yǔ)句中不需要顯式地設(shè)置 colocate_with 屬性），不容易導(dǎo)致數(shù)據(jù)傾斜的問(wèn)題
• 可以為 Join Reorder 提供更多可能的優(yōu)化空間

Plan Rule

• Bucket Shuffle Join 只生效于 Join 條件為等值的場(chǎng)景，原因與 Colocate Join 類(lèi)似，它們都依賴(lài) Hash 來(lái)計(jì)算確定的數(shù)據(jù)分布
• 在等值 Join 條件之中包含兩張表的分桶列，當(dāng)左表的分桶列為等值的 Join 條件時(shí)，它有很大概率會(huì)被規(guī)劃為 Bucket Shuffle Join
• 由于不同的數(shù)據(jù)類(lèi)型的 Hash 值計(jì)算結(jié)果不同，所以 Bucket Shuffle Join 要求左表的分桶列的類(lèi)型與右表等值 Join 列的類(lèi)型需要保持一致，否則無(wú)法進(jìn)行對(duì)應(yīng)的規(guī)劃
• Bucket Shuffle Join 只作用于 Doris 原生的 OLAP 表，對(duì)于 ODBC，MySQL，ES 等外表，當(dāng)其作為左表時(shí)是無(wú)法規(guī)劃生效的
• 對(duì)于分區(qū)表，由于每一個(gè)分區(qū)的數(shù)據(jù)分布規(guī)則可能不同，所以 Bucket Shuffle Join 只能保證左表為單分區(qū)時(shí)生效。所以在 SQL 執(zhí)行之中，需要盡量使用 where 條件使分區(qū)裁剪的策略能夠生效
• 假如左表為 Colocate 的表，那么它每個(gè)分區(qū)的數(shù)據(jù)分布規(guī)則是確定的，Bucket Shuffle Join 能在Colocate 表上表現(xiàn)更好

Colocate Join

可以理解為在數(shù)據(jù)分布滿(mǎn)足一定條件的前提下，減少一切不必要的網(wǎng)絡(luò)傳輸開(kāi)銷(xiāo)，實(shí)現(xiàn)完全的計(jì)算本地化來(lái)加速查詢(xún)。同時(shí)因?yàn)闆](méi)有網(wǎng)絡(luò)傳輸開(kāi)銷(xiāo)，BE 節(jié)點(diǎn)可以擁有更高的并發(fā)度，從而進(jìn)一步提升 Join 性能。

要理解這個(gè)算法，需要先了解兩個(gè)術(shù)語(yǔ)：

• Colocation Group（CG）：一個(gè) CG 中會(huì)包含一張及以上的 Table。在同一個(gè) Group 內(nèi)的 Table 有著相同的 Colocation Group Schema，并且有著相同的數(shù)據(jù)分片分布
• Colocation Group Schema（CGS）：用于描述一個(gè) CG 中的 Table，和 Colocation 相關(guān)的通用 Schema 信息。包括分桶列類(lèi)型，分桶數(shù)以及副本數(shù)等

和 Buckets Sequence 這一概念：

一個(gè)表的數(shù)據(jù)，最終會(huì)根據(jù)分桶列值 Hash、對(duì)桶數(shù)取模后落在某一個(gè)分桶內(nèi)。假設(shè)一個(gè) Table 的分桶數(shù)為 8，則共有 [0, 1, 2, 3, 4, 5, 6, 7] 8 個(gè)分桶（Bucket），我們稱(chēng)這樣一個(gè)序列為一個(gè) BucketsSequence。每個(gè) Bucket 內(nèi)會(huì)有一個(gè)或多個(gè)數(shù)據(jù)分片（Tablet）。當(dāng)表為單分區(qū)表時(shí)，一個(gè) Bucket 內(nèi)僅有一個(gè) Tablet。如果是多分區(qū)表，則會(huì)有多個(gè)（因?yàn)槎鄠€(gè) Partition 中的不同 Tablet 會(huì)被劃分到相同的 Bucket）。

Colocation Join 功能，是將一組擁有相同 CGS 的 Table 組成一個(gè) CG。并保證這些 Table 對(duì)應(yīng)的數(shù)據(jù)分片會(huì)落在同一個(gè) BE 節(jié)點(diǎn)上。使得當(dāng) CG 內(nèi)的表進(jìn)行分桶列上的 Join 操作時(shí)，可以通過(guò)直接進(jìn)行本地?cái)?shù)據(jù) Join，減少數(shù)據(jù)在節(jié)點(diǎn)間的傳輸耗時(shí)。

因此關(guān)鍵問(wèn)題就轉(zhuǎn)變?yōu)榱恕溉绾伪ＷC這些 Table 對(duì)應(yīng)的數(shù)據(jù)分片會(huì)落在同一個(gè) BE 節(jié)點(diǎn)上？」

通過(guò)同一 CG 內(nèi)的 Table 必須保證以下屬性相同實(shí)現(xiàn)：

• 分桶列和分桶數(shù)

分桶列，即在建表語(yǔ)句中 DISTRIBUTED BY HASH(col1, col2, ...) 中指定的列。分桶列決定了一張表的數(shù)據(jù)通過(guò)哪些列的值進(jìn)行 Hash 劃分到不同的 Tablet 中。同一 CG 內(nèi)的 Table 必須保證分桶列的類(lèi)型和數(shù)量完全一致，并且桶數(shù)一致，才能保證多張表的數(shù)據(jù)分片能夠一一對(duì)應(yīng)的進(jìn)行分布控制。

• 副本數(shù)

同一個(gè) CG 內(nèi)所有表的所有分區(qū)（Partition）的副本數(shù)必須一致。如果不一致，可能出現(xiàn)某一個(gè) Tablet 的某一個(gè)副本，在同一個(gè) BE 上沒(méi)有其他的表分片的副本對(duì)應(yīng)。不過(guò)，同一個(gè) CG 內(nèi)的表，分區(qū)的個(gè)數(shù)、范圍以及分區(qū)列的類(lèi)型不要求一致。

在固定了分桶列和分桶數(shù)后，同一個(gè) CG 內(nèi)的表會(huì)擁有相同的 BucketsSequence。而副本數(shù)決定了每個(gè)分桶內(nèi)的 Tablet 的多個(gè)副本，存放在哪些 BE 上。假設(shè) BucketsSequence 為 [0, 1, 2, 3, 4, 5, 6, 7]，BE 節(jié)點(diǎn)有 [A, B, C, D] 4個(gè)。則一個(gè)可能的數(shù)據(jù)分布如下：

+---+ +---+ +---+ +---+ +---+ +---+ +---+ +---+| 0 | | 1 | | 2 | | 3 | | 4 | | 5 | | 6 | | 7 |+---+ +---+ +---+ +---+ +---+ +---+ +---+ +---+| A | | B | | C | | D | | A | | B | | C | | D ||   | |   | |   | |   | |   | |   | |   | |   || B | | C | | D | | A | | B | | C | | D | | A ||   | |   | |   | |   | |   | |   | |   | |   || C | | D | | A | | B | | C | | D | | A | | B |+---+ +---+ +---+ +---+ +---+ +---+ +---+ +---+

CG 內(nèi)所有表的數(shù)據(jù)都會(huì)按照上面的規(guī)則進(jìn)行統(tǒng)一分布，這樣就保證了，分桶列值相同的數(shù)據(jù)都在同一個(gè) BE 節(jié)點(diǎn)上，可以進(jìn)行本地?cái)?shù)據(jù) Join。其核心思想是「兩次映射」，保證相同的 Distributed Key 的數(shù)據(jù)會(huì)被映射到相同的 Bucket Sequence，再保證 Bucket Sequence 對(duì)應(yīng)的 Bucket 映射到相同的 BE 節(jié)點(diǎn)：

通過(guò)查詢(xún)計(jì)劃可以檢查一個(gè)查詢(xún)是否使用了 Colocate Join，同時(shí)計(jì)劃中的 Exchange Node 也被去掉了，會(huì)將 ScanNode 直接設(shè)置為 Hash Join Node 的孩子節(jié)點(diǎn)。

DESC SELECT * FROM tbl1 INNER JOIN tbl2 ON (tbl1.k2 = tbl2.k2);-- 在 Hash Join 節(jié)點(diǎn)會(huì)顯示：-- colocate: true/false

Colocate Join 十分適合幾張表按照相同字段分桶，并高頻根據(jù)固定的字段 Join 的場(chǎng)景。這樣可以將數(shù)據(jù)預(yù)先存儲(chǔ)到相同的分桶中，實(shí)現(xiàn)本地計(jì)算。

Runtime Filter 優(yōu)化

Doris 在進(jìn)行 Hash Join 計(jì)算時(shí)會(huì)在右表構(gòu)建一個(gè) Hash Table，左表流式地通過(guò)右表的 Hash Table 從而得出 Join 結(jié)果。而 Runtime Filter 就是充分利用了右表的 Hash Table 構(gòu)建階段去做一些額外的事情。

在右表生成 Hash Table 的時(shí)，同時(shí)生成一個(gè)基于 Hash Table 數(shù)據(jù)的一個(gè)過(guò)濾條件，然后下推到左表的數(shù)據(jù)掃描節(jié)點(diǎn)。通過(guò)這樣的方式，Doris 可以在運(yùn)行時(shí)進(jìn)行數(shù)據(jù)過(guò)濾。

假如左表是一張大表，右表是一張小表，那么利用下推到左表的過(guò)濾條件就可以把絕大多數(shù) Join 層要過(guò)濾的數(shù)據(jù)在數(shù)據(jù)讀取時(shí)就提前過(guò)濾（如果能夠下推到引擎層，還能夠利用 Doris 針對(duì) Key 列過(guò)濾的延遲物化），從而大幅度地提升 Join 查詢(xún)的性能。

Runtime Filter 在查詢(xún)規(guī)劃時(shí)生成，在 HashJoinNode 中構(gòu)建，在 ScanNode 中應(yīng)用。比如 T1（行數(shù) 10w） 和 T2（行數(shù) 2k） 的 Join 操作：

|  >      HashJoinNode     <| | || | 100000  | 2000| | ||   OlapScanNode      OlapScanNode| ^ ^   | | 100000  | 2000|T1T2|

顯而易見(jiàn)對(duì) T2 掃描數(shù)據(jù)要遠(yuǎn)遠(yuǎn)快于 T1，如果我們主動(dòng)等待一段時(shí)間再掃描 T1，等 T2 將掃描的數(shù)據(jù)記錄交給 HashJoinNode 后，HashJoinNode 根據(jù) T2 的數(shù)據(jù)計(jì)算出一個(gè)過(guò)濾條件，比如 T2 數(shù)據(jù)的最大和最小值，或者構(gòu)建一個(gè) Bloom Filter，接著將這個(gè)過(guò)濾條件發(fā)給等待掃描 T1 的 ScanNode，后者應(yīng)用這個(gè)過(guò)濾條件，將過(guò)濾后的數(shù)據(jù)交給 HashJoinNode，從而減少 probe hash table 的次數(shù)和網(wǎng)絡(luò)開(kāi)銷(xiāo)，這個(gè)過(guò)濾條件就是 Runtime Filter，效果如下:

|  >      HashJoinNode     <| | || | 6000    | 2000| | ||   OlapScanNode      OlapScanNode| ^ ^   | | 100000  | 2000|T1T2|

如果能將過(guò)濾條件（Runtime Filter）下推到存儲(chǔ)引擎，則某些情況可以利用索引（比如 Join 列為 Key 列，可以利用延遲物化能力）來(lái)直接減少掃描的數(shù)據(jù)量，從而大大減少掃描耗時(shí)，效果如下：

|  >      HashJoinNode     <| | || | 6000    | 2000| | ||   OlapScanNode      OlapScanNode| ^ ^   | | 6000    | 2000|T1T2|

可見(jiàn)，和謂詞下推、分區(qū)裁剪不同，Runtime Filter 是在運(yùn)行時(shí)動(dòng)態(tài)生成的過(guò)濾條件，即在查詢(xún)運(yùn)行時(shí)解析 Join 條件確定過(guò)濾表達(dá)式，并將表達(dá)式下推給正在讀取左表的 ScanNode，從而減少掃描的數(shù)據(jù)量，進(jìn)而減少 probe hash table 的次數(shù)，避免不必要的 IO 和網(wǎng)絡(luò)傳輸。因?yàn)槠溥\(yùn)行時(shí)生效的特性，官方認(rèn)為它是 Adaptive Query Execution 的一種應(yīng)用。

根據(jù)上面的例子，可以推導(dǎo)出場(chǎng)景滿(mǎn)足以下的條件時(shí)，使用 Runtime Filter 的效果會(huì)比較好：

• 左表大右表?。ó?dāng)右表上還有額外的過(guò)濾條件會(huì)更理想），因?yàn)闃?gòu)建 Runtime Filter 是需要承擔(dān)計(jì)算成本的，包括一些內(nèi)存的開(kāi)銷(xiāo)，而開(kāi)銷(xiāo)直接取決于右表的實(shí)際數(shù)據(jù)量
• 左右表 Join 出來(lái)的結(jié)果很少，說(shuō)明通過(guò) Runtime Filter 可以過(guò)濾掉左表的絕大部分?jǐn)?shù)據(jù)

Doris 支持 3 種 Runtime Filter：

• 一種是 IN，很好理解，將一個(gè) hashset 下推到數(shù)據(jù)掃描節(jié)點(diǎn)。
• 第二種就是 BloomFilter，就是利用哈希表的數(shù)據(jù)構(gòu)造一個(gè) BloomFilter，然后把這個(gè) BloomFilter 下推到查詢(xún)數(shù)據(jù)的掃描節(jié)點(diǎn)。
• 最后一種就是 MinMax，就是個(gè) Range 范圍，通過(guò)右表數(shù)據(jù)確定 Range 范圍之后，下推給數(shù)據(jù)掃描節(jié)點(diǎn)。

工作原理和優(yōu)劣總結(jié)如下：

Runtime Filter 類(lèi)型工作原理適用場(chǎng)景優(yōu)點(diǎn)缺點(diǎn)IN子查詢(xún)的方式，實(shí)現(xiàn)上是將一個(gè) Hashset 下推到 Scan 節(jié)點(diǎn)Broadcast Join開(kāi)銷(xiāo)小，過(guò)濾效果明顯且快速右表超過(guò)一定數(shù)據(jù)量時(shí)會(huì)失效，目前 Doris 配置的閾值是 1024Min/Max通過(guò)右表構(gòu)建一個(gè) Range 范圍，然后將它下推到 Scan 節(jié)點(diǎn)通用開(kāi)銷(xiāo)小僅對(duì)數(shù)值類(lèi)型有效果；對(duì)數(shù)值以外類(lèi)型無(wú)法使用BloomFilter通過(guò)右表構(gòu)建一個(gè) BloomFilter，然后將它下推到 Scan 節(jié)點(diǎn)通用通用性較好，適用于各種類(lèi)型、效果也較好配置比較復(fù)雜且計(jì)算成本較高；當(dāng)過(guò)濾率較低或者左表數(shù)據(jù)較少時(shí)，可能導(dǎo)致性能降低

一些使用的注意事項(xiàng)（比較細(xì)節(jié)了，后面考慮結(jié)合代碼再深入理解）：

開(kāi)啟 Runtime Filter 后，左表的 ScanNode 會(huì)為每一個(gè)分配給自己的 Runtime Filter 等待一段時(shí)間再掃描數(shù)據(jù)，即如果 ScanNode 被分配了 3 個(gè) Runtime Filter，那么它最多會(huì)等待 3000ms。

因?yàn)?Runtime Filter 的構(gòu)建和合并均需要時(shí)間，ScanNode 會(huì)嘗試將等待時(shí)間內(nèi)到達(dá)的 Runtime Filter 下推到存儲(chǔ)引擎，如果超過(guò)等待時(shí)間后，ScanNode 會(huì)使用已經(jīng)到達(dá)的 Runtime Filter 直接開(kāi)始掃描數(shù)據(jù)。

如果 Runtime Filter 在 ScanNode 開(kāi)始掃描之后到達(dá)，則 ScanNode 不會(huì)將該 Runtime Filter 下推到存儲(chǔ)引擎，而是對(duì)已經(jīng)從存儲(chǔ)引擎掃描上來(lái)的數(shù)據(jù)，在 ScanNode 上基于該 Runtime Filter 使用表達(dá)式過(guò)濾，之前已經(jīng)掃描的數(shù)據(jù)則不會(huì)應(yīng)用該 Runtime Filter，這樣得到的中間數(shù)據(jù)規(guī)模會(huì)大于最優(yōu)解，但可以避免嚴(yán)重的劣化。

如果集群比較繁忙，并且集群上有許多資源密集型或長(zhǎng)耗時(shí)的查詢(xún)，可以考慮增加等待時(shí)間，以避免復(fù)雜查詢(xún)錯(cuò)過(guò)優(yōu)化機(jī)會(huì)。如果集群負(fù)載較輕，并且集群上有許多只需要幾秒的小查詢(xún)，可以考慮減少等待時(shí)間，以避免每個(gè)查詢(xún)?cè)黾?1s 的延遲。

Join Reorder 優(yōu)化

有了前面兩表 Join 的 Runtime Filter 鋪墊，再來(lái)看 Join Reorder 的優(yōu)化，邏輯關(guān)系上就能夠理順了。

Doris 目前的 Join Reorder 算法是基于 RBO 的，邏輯描述如下：

• 盡量讓大表跟小表做 Join，它生成的中間結(jié)果是盡可能小的
• 把有條件的 Join 表往前放，也就是說(shuō)盡量讓有條件的 Join 表進(jìn)行過(guò)濾
• Hash Join 的優(yōu)先級(jí)高于 Nest Loop Join，因?yàn)?Hash join 本身是比 Nest Loop Join 快很多的

可以發(fā)現(xiàn)前兩條，都是在朝著讓「右表」更小的方向去優(yōu)化，而最后一條則是從算法的性能上來(lái)考慮。

Join 調(diào)優(yōu)建議

• Join 列最好是相同的簡(jiǎn)單類(lèi)型；同類(lèi)型避免 Cast 操作，簡(jiǎn)單類(lèi)型則有不錯(cuò)的 Join 計(jì)算性能
• Join 列最好是 Key 列，原因是 Key 列能夠充分利用 Doris 延遲物化的特性，減少 IO 提升性能
• 大表之間的 Join 最好能夠利用上 Colocate，相當(dāng)于已經(jīng)做好了預(yù) Shuffle，實(shí)際查詢(xún)的時(shí)候可以直接 Join 計(jì)算不再有 Shuffle 操作，徹底避免了大表的 Shuffle 網(wǎng)絡(luò)開(kāi)銷(xiāo)
• 利用 Runtime Filter，Join 過(guò)濾性高時(shí)效果顯著。根據(jù) 3 種 Runtime Filter 特點(diǎn)選擇最適合的
• 涉及多表 Join，需要判斷 Join 的合理性。盡量保證“左大右小”的原則，HashJoin 優(yōu)于 NLJ。必要時(shí)可以通過(guò) SQL Rewrite，通過(guò) Hint 來(lái)調(diào)整 Join 順序

REF

以上就是Apache Doris Join 優(yōu)化原理詳解的詳細(xì)內(nèi)容，更多關(guān)于Apache Doris Join 優(yōu)化的資料請(qǐng)關(guān)注其它相關(guān)文章！