Bucket Shuffle Join
Bucket Shuffle Join 是在 Doris 0.14 版本中正式加入的新功能。旨在為某些 Join 查詢提供本地性優(yōu)化，來減少數(shù)據(jù)在節(jié)點間的傳輸耗時，來加速查詢。
它的設計、實現(xiàn)和效果可以參閱?
上面的圖片展示了Bucket Shuffle Join的工作原理。SQL語句為 A表 join B表，并且join的等值表達式命中了A的數(shù)據(jù)分布列。而Bucket Shuffle Join會根據(jù)A表的數(shù)據(jù)分布信息，將B表的數(shù)據(jù)發(fā)送到對應的A表的數(shù)據(jù)存儲計算節(jié)點。Bucket Shuffle Join開銷如下：
- 網(wǎng)絡開銷：?
B < min(3B, A + B) - 內(nèi)存開銷：?
B <= min(3B, B)
可見，相比于Broadcast Join與Shuffle Join， Bucket Shuffle Join有著較為明顯的性能優(yōu)勢。減少數(shù)據(jù)在節(jié)點間的傳輸耗時和Join時的內(nèi)存開銷。相對于Doris原有的Join方式，它有著下面的優(yōu)點
- 首先，Bucket-Shuffle-Join降低了網(wǎng)絡與內(nèi)存開銷，使一些Join查詢具有了更好的性能。尤其是當FE能夠執(zhí)行左表的分區(qū)裁剪與桶裁剪時。
- 其次，同時與Colocate Join不同，它對于表的數(shù)據(jù)分布方式并沒有侵入性，這對于用戶來說是透明的。對于表的數(shù)據(jù)分布沒有強制性的要求，不容易導致數(shù)據(jù)傾斜的問題。
- 最后，它可以為Join Reorder提供更多可能的優(yōu)化空間。
使用方式?
設置Session變量?
將session變量enable_bucket_shuffle_join設置為true，則FE在進行查詢規(guī)劃時就會默認將能夠轉換為Bucket Shuffle Join的查詢自動規(guī)劃為Bucket Shuffle Join。
set enable_bucket_shuffle_join = true;
在FE進行分布式查詢規(guī)劃時，優(yōu)先選擇的順序為 Colocate Join -> Bucket Shuffle Join -> Broadcast Join -> Shuffle Join。但是如果用戶顯式hint了Join的類型，如：
select * from test join [shuffle] baseall on test.k1 = baseall.k1;
則上述的選擇優(yōu)先順序則不生效。
該session變量在0.14版本默認為true, 而0.13版本需要手動設置為true。
查看Join的類型?
可以通過explain命令來查看Join是否為Bucket Shuffle Join：
| 2:HASH JOIN |
| | join op: INNER JOIN (BUCKET_SHUFFLE) |
| | hash predicates: |
| | colocate: false, reason: table not in the same group |
| | equal join conjunct: `test`.`k1` = `baseall`.`k1`
在Join類型之中會指明使用的Join方式為：BUCKET_SHUFFLE。
Bucket Shuffle Join的規(guī)劃規(guī)則?
在絕大多數(shù)場景之中，用戶只需要默認打開session變量的開關就可以透明的使用這種Join方式帶來的性能提升，但是如果了解Bucket Shuffle Join的規(guī)劃規(guī)則，可以幫助我們利用它寫出更加高效的SQL。
- Bucket Shuffle Join只生效于Join條件為等值的場景，原因與Colocate Join類似，它們都依賴hash來計算確定的數(shù)據(jù)分布。
- 在等值Join條件之中包含兩張表的分桶列，當左表的分桶列為等值的Join條件時，它有很大概率會被規(guī)劃為Bucket Shuffle Join。
- 由于不同的數(shù)據(jù)類型的hash值計算結果不同，所以Bucket Shuffle Join要求左表的分桶列的類型與右表等值join列的類型需要保持一致，否則無法進行對應的規(guī)劃。
- Bucket Shuffle Join只作用于Doris原生的OLAP表，對于ODBC，MySQL，ES等外表，當其作為左表時是無法規(guī)劃生效的。
- 對于分區(qū)表，由于每一個分區(qū)的數(shù)據(jù)分布規(guī)則可能不同，所以Bucket Shuffle Join只能保證左表為單分區(qū)時生效。所以在SQL執(zhí)行之中，需要盡量使用
where條件使分區(qū)裁剪的策略能夠生效。 - 假如左表為Colocate的表，那么它每個分區(qū)的數(shù)據(jù)分布規(guī)則是確定的，Bucket Shuffle Join能在Colocate表上表現(xiàn)更好。
Colocation Join
Colocation Join 是在 Doris 0.9 版本中引入的新功能。旨在為某些 Join 查詢提供本地性優(yōu)化，來減少數(shù)據(jù)在節(jié)點間的傳輸耗時，加速查詢。
最初的設計、實現(xiàn)和效果可以參閱?ISSUE 245。
Colocation Join 功能經(jīng)過一次改版，設計和使用方式和最初設計稍有不同。本文檔主要介紹 Colocation Join 的原理、實現(xiàn)、使用方式和注意事項。
注意：這個屬性不會被CCR同步，如果這個表是被CCR復制而來的，即PROPERTIES中包含is_being_synced = true時，這個屬性將會在這個表中被擦除。
名詞解釋?
- Colocation Group（CG）：一個 CG 中會包含一張及以上的 Table。在同一個 Group 內(nèi)的 Table 有著相同的 Colocation Group Schema，并且有著相同的數(shù)據(jù)分片分布。
- Colocation Group Schema（CGS）：用于描述一個 CG 中的 Table，和 Colocation 相關的通用 Schema 信息。包括分桶列類型，分桶數(shù)以及副本數(shù)等。
原理?
Colocation Join 功能，是將一組擁有相同 CGS 的 Table 組成一個 CG。并保證這些 Table 對應的數(shù)據(jù)分片會落在同一個 BE 節(jié)點上。使得當 CG 內(nèi)的表進行分桶列上的 Join 操作時，可以通過直接進行本地數(shù)據(jù) Join，減少數(shù)據(jù)在節(jié)點間的傳輸耗時。
一個表的數(shù)據(jù)，最終會根據(jù)分桶列值 Hash、對桶數(shù)取模的后落在某一個分桶內(nèi)。假設一個 Table 的分桶數(shù)為 8，則共有?[0, 1, 2, 3, 4, 5, 6, 7]?8 個分桶（Bucket），我們稱這樣一個序列為一個?BucketsSequence。每個 Bucket 內(nèi)會有一個或多個數(shù)據(jù)分片（Tablet）。當表為單分區(qū)表時，一個 Bucket 內(nèi)僅有一個 Tablet。如果是多分區(qū)表，則會有多個。
為了使得 Table 能夠有相同的數(shù)據(jù)分布，同一 CG 內(nèi)的 Table 必須保證以下屬性相同：
-
分桶列和分桶數(shù)
分桶列，即在建表語句中?
DISTRIBUTED BY HASH(col1, col2, ...)?中指定的列。分桶列決定了一張表的數(shù)據(jù)通過哪些列的值進行 Hash 劃分到不同的 Tablet 中。同一 CG 內(nèi)的 Table 必須保證分桶列的類型和數(shù)量完全一致，并且桶數(shù)一致，才能保證多張表的數(shù)據(jù)分片能夠一一對應的進行分布控制。 -
副本數(shù)
同一個 CG 內(nèi)所有表的所有分區(qū)（Partition）的副本數(shù)必須一致。如果不一致，可能出現(xiàn)某一個 Tablet 的某一個副本，在同一個 BE 上沒有其他的表分片的副本對應。
同一個 CG 內(nèi)的表，分區(qū)的個數(shù)、范圍以及分區(qū)列的類型不要求一致。
在固定了分桶列和分桶數(shù)后，同一個 CG 內(nèi)的表會擁有相同的 BucketsSequence。而副本數(shù)決定了每個分桶內(nèi)的 Tablet 的多個副本，存放在哪些 BE 上。假設 BucketsSequence 為?[0, 1, 2, 3, 4, 5, 6, 7]，BE 節(jié)點有?[A, B, C, D]?4個。則一個可能的數(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ù)都會按照上面的規(guī)則進行統(tǒng)一分布，這樣就保證了，分桶列值相同的數(shù)據(jù)都在同一個 BE 節(jié)點上，可以進行本地數(shù)據(jù) Join。
使用方式?
建表?
建表時，可以在?PROPERTIES?中指定屬性?"colocate_with" = "group_name"，表示這個表是一個 Colocation Join 表，并且歸屬于一個指定的 Colocation Group。
示例：
CREATE TABLE tbl (k1 int, v1 int sum)
DISTRIBUTED BY HASH(k1)
BUCKETS 8
PROPERTIES(
"colocate_with" = "group1"
);
如果指定的 Group 不存在，則 Doris 會自動創(chuàng)建一個只包含當前這張表的 Group。如果 Group 已存在，則 Doris 會檢查當前表是否滿足 Colocation Group Schema。如果滿足，則會創(chuàng)建該表，并將該表加入 Group。同時，表會根據(jù)已存在的 Group 中的數(shù)據(jù)分布規(guī)則創(chuàng)建分片和副本。 Group 歸屬于一個 Database，Group 的名字在一個 Database 內(nèi)唯一。在內(nèi)部存儲是 Group 的全名為?dbId_groupName，但用戶只感知 groupName。
SinceVersion?dev
2.0 版本中，Doris 支持了跨Database的 Group。在建表時，需使用關鍵詞?__global__?作為 Group 名稱的前綴。如：
CREATE TABLE tbl (k1 int, v1 int sum)
DISTRIBUTED BY HASH(k1)
BUCKETS 8
PROPERTIES(
"colocate_with" = "__global__group1"
);
__global__?前綴的 Group 不再歸屬于一個 Database，其名稱也是全局唯一的。
通過創(chuàng)建 Global Group，可以實現(xiàn)跨 Database 的 Colocate Join。
刪表?
當 Group 中最后一張表徹底刪除后（徹底刪除是指從回收站中刪除。通常，一張表通過?DROP TABLE?命令刪除后，會在回收站默認停留一天的時間后，再刪除），該 Group 也會被自動刪除。
查看 Group?
以下命令可以查看集群內(nèi)已存在的 Group 信息。
SHOW PROC '/colocation_group';
+-------------+--------------+--------------+------------+----------------+----------+----------+
| GroupId | GroupName | TableIds | BucketsNum | ReplicationNum | DistCols | IsStable |
+-------------+--------------+--------------+------------+----------------+----------+----------+
| 10005.10008 | 10005_group1 | 10007, 10040 | 10 | 3 | int(11) | true |
+-------------+--------------+--------------+------------+----------------+----------+----------+
- GroupId： 一個 Group 的全集群唯一標識，前半部分為 db id，后半部分為 group id。
- GroupName： Group 的全名。
- TabletIds： 該 Group 包含的 Table 的 id 列表。
- BucketsNum： 分桶數(shù)。
- ReplicationNum： 副本數(shù)。
- DistCols： Distribution columns，即分桶列類型。
- IsStable： 該 Group 是否穩(wěn)定（穩(wěn)定的定義，見?
Colocation 副本均衡和修復?一節(jié)）。
通過以下命令可以進一步查看一個 Group 的數(shù)據(jù)分布情況：
SHOW PROC '/colocation_group/10005.10008';
+-------------+---------------------+
| BucketIndex | BackendIds |
+-------------+---------------------+
| 0 | 10004, 10002, 10001 |
| 1 | 10003, 10002, 10004 |
| 2 | 10002, 10004, 10001 |
| 3 | 10003, 10002, 10004 |
| 4 | 10002, 10004, 10003 |
| 5 | 10003, 10002, 10001 |
| 6 | 10003, 10004, 10001 |
| 7 | 10003, 10004, 10002 |
+-------------+---------------------+
- BucketIndex： 分桶序列的下標。
- BackendIds： 分桶中數(shù)據(jù)分片所在的 BE 節(jié)點 id 列表。
以上命令需要 ADMIN 權限。暫不支持普通用戶查看。
修改表 Colocate Group 屬性?
可以對一個已經(jīng)創(chuàng)建的表，修改其 Colocation Group 屬性。示例：
ALTER TABLE tbl SET ("colocate_with" = "group2");
- 如果該表之前沒有指定過 Group，則該命令檢查 Schema，并將該表加入到該 Group（Group 不存在則會創(chuàng)建）。
- 如果該表之前有指定其他 Group，則該命令會先將該表從原有 Group 中移除，并加入新 Group（Group 不存在則會創(chuàng)建）。
也可以通過以下命令，刪除一個表的 Colocation 屬性：
ALTER TABLE tbl SET ("colocate_with" = "");
其他相關操作?
當對一個具有 Colocation 屬性的表進行增加分區(qū)（ADD PARTITION）、修改副本數(shù)時，Doris 會檢查修改是否會違反 Colocation Group Schema，如果違反則會拒絕。
Colocation 副本均衡和修復?
Colocation 表的副本分布需要遵循 Group 中指定的分布，所以在副本修復和均衡方面和普通分片有所區(qū)別。
Group 自身有一個 Stable 屬性，當 Stable 為 true 時，表示當前 Group 內(nèi)的表的所有分片沒有正在進行變動，Colocation 特性可以正常使用。當 Stable 為 false 時（Unstable），表示當前 Group 內(nèi)有部分表的分片正在做修復或遷移，此時，相關表的 Colocation Join 將退化為普通 Join。
副本修復?
副本只能存儲在指定的 BE 節(jié)點上。所以當某個 BE 不可用時（宕機、Decommission 等），需要尋找一個新的 BE 進行替換。Doris 會優(yōu)先尋找負載最低的 BE 進行替換。替換后，該 Bucket 內(nèi)的所有在舊 BE 上的數(shù)據(jù)分片都要做修復。遷移過程中，Group 被標記為 Unstable。
副本均衡?
Doris 會盡力將 Colocation 表的分片均勻分布在所有 BE 節(jié)點上。對于普通表的副本均衡，是以單副本為粒度的，即單獨為每一個副本尋找負載較低的 BE 節(jié)點即可。而 Colocation 表的均衡是 Bucket 級別的，即一個 Bucket 內(nèi)的所有副本都會一起遷移。我們采用一個簡單的均衡算法，即在不考慮副本實際大小，而只根據(jù)副本數(shù)量，將 BucketsSequence 均勻的分布在所有 BE 上。具體算法可以參閱?ColocateTableBalancer.java?中的代碼注釋。
注1：當前的 Colocation 副本均衡和修復算法，對于異構部署的 Doris 集群效果可能不佳。所謂異構部署，即 BE 節(jié)點的磁盤容量、數(shù)量、磁盤類型（SSD 和 HDD）不一致。在異構部署情況下，可能出現(xiàn)小容量的 BE 節(jié)點和大容量的 BE 節(jié)點存儲了相同的副本數(shù)量。
注2：當一個 Group 處于 Unstable 狀態(tài)時，其中的表的 Join 將退化為普通 Join。此時可能會極大降低集群的查詢性能。如果不希望系統(tǒng)自動均衡，可以設置 FE 的配置項?
disable_colocate_balance?來禁止自動均衡。然后在合適的時間打開即可。（具體參閱?高級操作?一節(jié)）
查詢?
對 Colocation 表的查詢方式和普通表一樣，用戶無需感知 Colocation 屬性。如果 Colocation 表所在的 Group 處于 Unstable 狀態(tài)，將自動退化為普通 Join。
舉例說明：
表1：
CREATE TABLE `tbl1` (
`k1` date NOT NULL COMMENT "",
`k2` int(11) NOT NULL COMMENT "",
`v1` int(11) SUM NOT NULL COMMENT ""
) ENGINE=OLAP
AGGREGATE KEY(`k1`, `k2`)
PARTITION BY RANGE(`k1`)
(
PARTITION p1 VALUES LESS THAN ('2019-05-31'),
PARTITION p2 VALUES LESS THAN ('2019-06-30')
)
DISTRIBUTED BY HASH(`k2`) BUCKETS 8
PROPERTIES (
"colocate_with" = "group1"
);
表2：
CREATE TABLE `tbl2` (
`k1` datetime NOT NULL COMMENT "",
`k2` int(11) NOT NULL COMMENT "",
`v1` double SUM NOT NULL COMMENT ""
) ENGINE=OLAP
AGGREGATE KEY(`k1`, `k2`)
DISTRIBUTED BY HASH(`k2`) BUCKETS 8
PROPERTIES (
"colocate_with" = "group1"
);
查看查詢計劃：
DESC SELECT * FROM tbl1 INNER JOIN tbl2 ON (tbl1.k2 = tbl2.k2);
+----------------------------------------------------+
| Explain String |
+----------------------------------------------------+
| PLAN FRAGMENT 0 |
| OUTPUT EXPRS:`tbl1`.`k1` | |
| PARTITION: RANDOM |
| |
| RESULT SINK |
| |
| 2:HASH JOIN |
| | join op: INNER JOIN |
| | hash predicates: |
| | colocate: true |
| | `tbl1`.`k2` = `tbl2`.`k2` |
| | tuple ids: 0 1 |
| | |
| |----1:OlapScanNode |
| | TABLE: tbl2 |
| | PREAGGREGATION: OFF. Reason: null |
| | partitions=0/1 |
| | rollup: null |
| | buckets=0/0 |
| | cardinality=-1 |
| | avgRowSize=0.0 |
| | numNodes=0 |
| | tuple ids: 1 |
| | |
| 0:OlapScanNode |
| TABLE: tbl1 |
| PREAGGREGATION: OFF. Reason: No AggregateInfo |
| partitions=0/2 |
| rollup: null |
| buckets=0/0 |
| cardinality=-1 |
| avgRowSize=0.0 |
| numNodes=0 |
| tuple ids: 0 |
+----------------------------------------------------+
如果 Colocation Join 生效，則 Hash Join 節(jié)點會顯示?colocate: true。
如果沒有生效，則查詢計劃如下：
+----------------------------------------------------+
| Explain String |
+----------------------------------------------------+
| PLAN FRAGMENT 0 |
| OUTPUT EXPRS:`tbl1`.`k1` | |
| PARTITION: RANDOM |
| |
| RESULT SINK |
| |
| 2:HASH JOIN |
| | join op: INNER JOIN (BROADCAST) |
| | hash predicates: |
| | colocate: false, reason: group is not stable |
| | `tbl1`.`k2` = `tbl2`.`k2` |
| | tuple ids: 0 1 |
| | |
| |----3:EXCHANGE |
| | tuple ids: 1 |
| | |
| 0:OlapScanNode |
| TABLE: tbl1 |
| PREAGGREGATION: OFF. Reason: No AggregateInfo |
| partitions=0/2 |
| rollup: null |
| buckets=0/0 |
| cardinality=-1 |
| avgRowSize=0.0 |
| numNodes=0 |
| tuple ids: 0 |
| |
| PLAN FRAGMENT 1 |
| OUTPUT EXPRS: |
| PARTITION: RANDOM |
| |
| STREAM DATA SINK |
| EXCHANGE ID: 03 |
| UNPARTITIONED |
| |
| 1:OlapScanNode |
| TABLE: tbl2 |
| PREAGGREGATION: OFF. Reason: null |
| partitions=0/1 |
| rollup: null |
| buckets=0/0 |
| cardinality=-1 |
| avgRowSize=0.0 |
| numNodes=0 |
| tuple ids: 1 |
+----------------------------------------------------+
HASH JOIN 節(jié)點會顯示對應原因：colocate: false, reason: group is not stable。同時會有一個 EXCHANGE 節(jié)點生成。
高級操作?
FE 配置項?
-
disable_colocate_relocate
是否關閉 Doris 的自動 Colocation 副本修復。默認為 false，即不關閉。該參數(shù)只影響 Colocation 表的副本修復，不影響普通表。
-
disable_colocate_balance
是否關閉 Doris 的自動 Colocation 副本均衡。默認為 false，即不關閉。該參數(shù)只影響 Colocation 表的副本均衡，不影響普通表。
以上參數(shù)可以動態(tài)修改，設置方式請參閱?HELP ADMIN SHOW CONFIG;?和?HELP ADMIN SET CONFIG;。
-
disable_colocate_join
是否關閉 Colocation Join 功能。在 0.10 及之前的版本，默認為 true，即關閉。在之后的某個版本中將默認為 false，即開啟。
-
use_new_tablet_scheduler
在 0.10 及之前的版本中，新的副本調(diào)度邏輯與 Colocation Join 功能不兼容，所以在 0.10 及之前版本，如果?
disable_colocate_join = false，則需設置?use_new_tablet_scheduler = false，即關閉新的副本調(diào)度器。之后的版本中，use_new_tablet_scheduler?將衡為 true。
HTTP Restful API?
Doris 提供了幾個和 Colocation Join 有關的 HTTP Restful API，用于查看和修改 Colocation Group。
該 API 實現(xiàn)在 FE 端，使用?fe_host:fe_http_port?進行訪問。需要 ADMIN 權限。
-
查看集群的全部 Colocation 信息
GET /api/colocate 返回以 Json 格式表示內(nèi)部 Colocation 信息。 { "msg": "success", "code": 0, "data": { "infos": [ ["10003.12002", "10003_group1", "10037, 10043", "1", "1", "int(11)", "true"] ], "unstableGroupIds": [], "allGroupIds": [{ "dbId": 10003, "grpId": 12002 }] }, "count": 0 } -
將 Group 標記為 Stable 或 Unstable
-
標記為 Stable
POST /api/colocate/group_stable?db_id=10005&group_id=10008 返回：200 -
標記為 Unstable
DELETE /api/colocate/group_stable?db_id=10005&group_id=10008 返回：200
-
-
設置 Group 的數(shù)據(jù)分布
該接口可以強制設置某一 Group 的數(shù)分布。
POST /api/colocate/bucketseq?db_id=10005&group_id=10008 Body: [[10004,10002],[10003,10002],[10002,10004],[10003,10002],[10002,10004],[10003,10002],[10003,10004],[10003,10004],[10003,10004],[10002,10004]] 返回 200其中 Body 是以嵌套數(shù)組表示的 BucketsSequence 以及每個 Bucket 中分片分布所在 BE 的 id。
注意，使用該命令，可能需要將 FE 的配置?
disable_colocate_relocate?和?disable_colocate_balance?設為 true。即關閉系統(tǒng)自動的 Colocation 副本修復和均衡。否則可能在修改后，會被系統(tǒng)自動重置。
?文章來源地址http://www.zghlxwxcb.cn/news/detail-666124.html
Runtime Filter
Runtime Filter 是在 Doris 0.15 版本中正式加入的新功能。旨在為某些 Join 查詢在運行時動態(tài)生成過濾條件，來減少掃描的數(shù)據(jù)量，避免不必要的I/O和網(wǎng)絡傳輸，從而加速查詢。
它的設計、實現(xiàn)和效果可以參閱?ISSUE 6116。
名詞解釋?
- 左表：Join查詢時，左邊的表。進行Probe操作?？杀籎oin Reorder調(diào)整順序。
- 右表：Join查詢時，右邊的表。進行Build操作?？杀籎oin Reorder調(diào)整順序。
- Fragment：FE會將具體的SQL語句的執(zhí)行轉化為對應的Fragment并下發(fā)到BE進行執(zhí)行。BE上執(zhí)行對應Fragment，并將結果匯聚返回給FE。
- Join on clause:?
A join B on A.a=B.b中的A.a=B.b，在查詢規(guī)劃時基于此生成join conjuncts，包含join Build和Probe使用的expr，其中Build expr在Runtime Filter中稱為src expr，Probe expr在Runtime Filter中稱為target expr。
原理?
Runtime Filter在查詢規(guī)劃時生成，在HashJoinNode中構建，在ScanNode中應用。
舉個例子，當前存在T1表與T2表的Join查詢，它的Join方式為HashJoin，T1是一張事實表，數(shù)據(jù)行數(shù)為100000，T2是一張維度表，數(shù)據(jù)行數(shù)為2000，Doris join的實際情況是:
| > HashJoinNode <
| | |
| | 100000 | 2000
| | |
| OlapScanNode OlapScanNode
| ^ ^
| | 100000 | 2000
| T1 T2
|
顯而易見對T2掃描數(shù)據(jù)要遠遠快于T1，如果我們主動等待一段時間再掃描T1，等T2將掃描的數(shù)據(jù)記錄交給HashJoinNode后，HashJoinNode根據(jù)T2的數(shù)據(jù)計算出一個過濾條件，比如T2數(shù)據(jù)的最大和最小值，或者構建一個Bloom Filter，接著將這個過濾條件發(fā)給等待掃描T1的ScanNode，后者應用這個過濾條件，將過濾后的數(shù)據(jù)交給HashJoinNode，從而減少probe hash table的次數(shù)和網(wǎng)絡開銷，這個過濾條件就是Runtime Filter，效果如下:
| > HashJoinNode <
| | |
| | 6000 | 2000
| | |
| OlapScanNode OlapScanNode
| ^ ^
| | 100000 | 2000
| T1 T2
|
如果能將過濾條件（Runtime Filter）下推到存儲引擎，則某些情況下可以利用索引來直接減少掃描的數(shù)據(jù)量，從而大大減少掃描耗時，效果如下:
| > HashJoinNode <
| | |
| | 6000 | 2000
| | |
| OlapScanNode OlapScanNode
| ^ ^
| | 6000 | 2000
| T1 T2
|
可見，和謂詞下推、分區(qū)裁剪不同，Runtime Filter是在運行時動態(tài)生成的過濾條件，即在查詢運行時解析join on clause確定過濾表達式，并將表達式廣播給正在讀取左表的ScanNode，從而減少掃描的數(shù)據(jù)量，進而減少probe hash table的次數(shù)，避免不必要的I/O和網(wǎng)絡傳輸。
Runtime Filter主要用于大表join小表的優(yōu)化，如果左表的數(shù)據(jù)量太小，或者右表的數(shù)據(jù)量太大，則Runtime Filter可能不會取得預期效果。
使用方式?
Runtime Filter查詢選項?
與Runtime Filter相關的查詢選項信息，請參閱以下部分:
- 第一個查詢選項是調(diào)整使用的Runtime Filter類型，大多數(shù)情況下，您只需要調(diào)整這一個選項，其他選項保持默認即可。
-
runtime_filter_type: 包括Bloom Filter、MinMax Filter、IN predicate、IN Or Bloom Filter、Bitmap Filter，默認會使用IN Or Bloom Filter，部分情況下同時使用Bloom Filter、MinMax Filter、IN predicate時性能更高。
-
- 其他查詢選項通常僅在某些特定場景下，才需進一步調(diào)整以達到最優(yōu)效果。通常只在性能測試后，針對資源密集型、運行耗時足夠長且頻率足夠高的查詢進行優(yōu)化。
-
runtime_filter_mode: 用于調(diào)整Runtime Filter的下推策略，包括OFF、LOCAL、GLOBAL三種策略，默認設置為GLOBAL策略 -
runtime_filter_wait_time_ms: 左表的ScanNode等待每個Runtime Filter的時間，默認1000ms -
runtime_filters_max_num: 每個查詢可應用的Runtime Filter中Bloom Filter的最大數(shù)量，默認10 -
runtime_bloom_filter_min_size: Runtime Filter中Bloom Filter的最小長度，默認1048576（1M） -
runtime_bloom_filter_max_size: Runtime Filter中Bloom Filter的最大長度，默認16777216（16M） -
runtime_bloom_filter_size: Runtime Filter中Bloom Filter的默認長度，默認2097152（2M） -
runtime_filter_max_in_num: 如果join右表數(shù)據(jù)行數(shù)大于這個值，我們將不生成IN predicate，默認1024
-
下面對查詢選項做進一步說明。
1.runtime_filter_type?
使用的Runtime Filter類型。
類型: 數(shù)字(1, 2, 4, 8, 16)或者相對應的助記符字符串(IN, BLOOM_FILTER, MIN_MAX,?IN_OR_BLOOM_FILTER, BITMAP_FILTER)，默認8(IN_OR_BLOOM_FILTER)，使用多個時用逗號分隔，注意需要加引號，或者將任意多個類型的數(shù)字相加，例如:
set runtime_filter_type="BLOOM_FILTER,IN,MIN_MAX";
等價于:
set runtime_filter_type=7;
使用注意事項
-
IN or Bloom Filter: 根據(jù)右表在執(zhí)行過程中的真實行數(shù)，由系統(tǒng)自動判斷使用 IN predicate 還是 Bloom Filter
- 默認在右表數(shù)據(jù)行數(shù)少于102400時會使用IN predicate（可通過session變量中的
runtime_filter_max_in_num調(diào)整），否則使用Bloom filter。
- 默認在右表數(shù)據(jù)行數(shù)少于102400時會使用IN predicate（可通過session變量中的
-
Bloom Filter: 有一定的誤判率，導致過濾的數(shù)據(jù)比預期少一點，但不會導致最終結果不準確，在大部分情況下Bloom Filter都可以提升性能或?qū)π阅軟]有顯著影響，但在部分情況下會導致性能降低。
- Bloom Filter構建和應用的開銷較高，所以當過濾率較低時，或者左表數(shù)據(jù)量較少時，Bloom Filter可能會導致性能降低。
- 目前只有左表的Key列應用Bloom Filter才能下推到存儲引擎，而測試結果顯示Bloom Filter不下推到存儲引擎時往往會導致性能降低。
- 目前Bloom Filter僅在ScanNode上使用表達式過濾時有短路(short-circuit)邏輯，即當假陽性率過高時，不繼續(xù)使用Bloom Filter，但當Bloom Filter下推到存儲引擎后沒有短路邏輯，所以當過濾率較低時可能導致性能降低。
-
MinMax Filter: 包含最大值和最小值，從而過濾小于最小值和大于最大值的數(shù)據(jù)，MinMax Filter的過濾效果與join on clause中Key列的類型和左右表數(shù)據(jù)分布有關。
- 當join on clause中Key列的類型為int/bigint/double等時，極端情況下，如果左右表的最大最小值相同則沒有效果，反之右表最大值小于左表最小值，或右表最小值大于左表最大值，則效果最好。
- 當join on clause中Key列的類型為varchar等時，應用MinMax Filter往往會導致性能降低。
-
IN predicate: 根據(jù)join on clause中Key列在右表上的所有值構建IN predicate，使用構建的IN predicate在左表上過濾，相比Bloom Filter構建和應用的開銷更低，在右表數(shù)據(jù)量較少時往往性能更高。
- 目前IN predicate已實現(xiàn)合并方法。
- 當同時指定In predicate和其他filter，并且in的過濾數(shù)值沒達到runtime_filter_max_in_num時，會嘗試把其他filter去除掉。原因是In predicate是精確的過濾條件，即使沒有其他filter也可以高效過濾，如果同時使用則其他filter會做無用功。目前僅在Runtime filter的生產(chǎn)者和消費者處于同一個fragment時才會有去除非in filter的邏輯。
-
Bitmap Filter:
- 當前僅當in subquery操作中的子查詢返回bitmap列時會使用bitmap filter.
- 當前僅在向量化引擎中支持bitmap filter.
2.runtime_filter_mode?
用于控制Runtime Filter在instance之間傳輸?shù)姆秶?/p>
類型: 數(shù)字(0, 1, 2)或者相對應的助記符字符串(OFF, LOCAL, GLOBAL)，默認2(GLOBAL)。
使用注意事項
LOCAL：相對保守，構建的Runtime Filter只能在同一個instance（查詢執(zhí)行的最小單元）上同一個Fragment中使用，即Runtime Filter生產(chǎn)者（構建Filter的HashJoinNode）和消費者（使用RuntimeFilter的ScanNode）在同一個Fragment，比如broadcast join的一般場景；
GLOBAL：相對激進，除滿足LOCAL策略的場景外，還可以將Runtime Filter合并后通過網(wǎng)絡傳輸?shù)讲煌琲nstance上的不同F(xiàn)ragment中使用，比如Runtime Filter生產(chǎn)者和消費者在不同F(xiàn)ragment，比如shuffle join。
大多數(shù)情況下GLOBAL策略可以在更廣泛的場景對查詢進行優(yōu)化，但在有些shuffle join中生成和合并Runtime Filter的開銷超過給查詢帶來的性能優(yōu)勢，可以考慮更改為LOCAL策略。
如果集群中涉及的join查詢不會因為Runtime Filter而提高性能，您可以將設置更改為OFF，從而完全關閉該功能。
在不同F(xiàn)ragment上構建和應用Runtime Filter時，需要合并Runtime Filter的原因和策略可參閱?ISSUE 6116(opens new window)
3.runtime_filter_wait_time_ms?
Runtime Filter的等待耗時。
類型: 整數(shù)，默認1000，單位ms
使用注意事項
在開啟Runtime Filter后，左表的ScanNode會為每一個分配給自己的Runtime Filter等待一段時間再掃描數(shù)據(jù)，即如果ScanNode被分配了3個Runtime Filter，那么它最多會等待3000ms。
因為Runtime Filter的構建和合并均需要時間，ScanNode會嘗試將等待時間內(nèi)到達的Runtime Filter下推到存儲引擎，如果超過等待時間后，ScanNode會使用已經(jīng)到達的Runtime Filter直接開始掃描數(shù)據(jù)。
如果Runtime Filter在ScanNode開始掃描之后到達，則ScanNode不會將該Runtime Filter下推到存儲引擎，而是對已經(jīng)從存儲引擎掃描上來的數(shù)據(jù)，在ScanNode上基于該Runtime Filter使用表達式過濾，之前已經(jīng)掃描的數(shù)據(jù)則不會應用該Runtime Filter，這樣得到的中間數(shù)據(jù)規(guī)模會大于最優(yōu)解，但可以避免嚴重的裂化。
如果集群比較繁忙，并且集群上有許多資源密集型或長耗時的查詢，可以考慮增加等待時間，以避免復雜查詢錯過優(yōu)化機會。如果集群負載較輕，并且集群上有許多只需要幾秒的小查詢，可以考慮減少等待時間，以避免每個查詢增加1s的延遲。
4.runtime_filters_max_num?
每個查詢生成的Runtime Filter中Bloom Filter數(shù)量的上限。
類型: 整數(shù)，默認10
使用注意事項?目前僅對Bloom Filter的數(shù)量進行限制，因為相比MinMax Filter和IN predicate，Bloom Filter構建和應用的代價更高。
如果生成的Bloom Filter超過允許的最大數(shù)量，則保留選擇性大的Bloom Filter，選擇性大意味著預期可以過濾更多的行。這個設置可以防止Bloom Filter耗費過多的內(nèi)存開銷而導致潛在的問題。
選擇性=(HashJoinNode Cardinality / HashJoinNode left child Cardinality)
-- 因為目前FE拿到Cardinality不準，所以這里Bloom Filter計算的選擇性與實際不準，因此最終可能只是隨機保留了部分Bloom Filter。
僅在對涉及大表間join的某些長耗時查詢進行調(diào)優(yōu)時，才需要調(diào)整此查詢選項。
5.Bloom Filter長度相關參數(shù)?
包括runtime_bloom_filter_min_size、runtime_bloom_filter_max_size、runtime_bloom_filter_size，用于確定Runtime Filter使用的Bloom Filter數(shù)據(jù)結構的大?。ㄒ宰止?jié)為單位）。
類型: 整數(shù)
使用注意事項?因為需要保證每個HashJoinNode構建的Bloom Filter長度相同才能合并，所以目前在FE查詢規(guī)劃時計算Bloom Filter的長度。
如果能拿到join右表統(tǒng)計信息中的數(shù)據(jù)行數(shù)(Cardinality)，會嘗試根據(jù)Cardinality估計Bloom Filter的最佳大小，并四舍五入到最接近的2的冪(以2為底的log值)。如果無法拿到右表的Cardinality，則會使用默認的Bloom Filter長度runtime_bloom_filter_size。runtime_bloom_filter_min_size和runtime_bloom_filter_max_size用于限制最終使用的Bloom Filter長度最小和最大值。
更大的Bloom Filter在處理高基數(shù)的輸入集時更有效，但需要消耗更多的內(nèi)存。假如查詢中需要過濾高基數(shù)列（比如含有數(shù)百萬個不同的取值），可以考慮增加runtime_bloom_filter_size的值進行一些基準測試，這有助于使Bloom Filter過濾的更加精準，從而獲得預期的性能提升。
Bloom Filter的有效性取決于查詢的數(shù)據(jù)分布，因此通常僅對一些特定查詢額外調(diào)整其Bloom Filter長度，而不是全局修改，一般僅在對涉及大表間join的某些長耗時查詢進行調(diào)優(yōu)時，才需要調(diào)整此查詢選項。
查看query生成的Runtime Filter?
explain命令可以顯示的查詢計劃中包括每個Fragment使用的join on clause信息，以及Fragment生成和使用Runtime Filter的注釋，從而確認是否將Runtime Filter應用到了期望的join on clause上。
- 生成Runtime Filter的Fragment包含的注釋例如
runtime filters: filter_id[type] <- table.column。 - 使用Runtime Filter的Fragment包含的注釋例如
runtime filters: filter_id[type] -> table.column。
下面例子中的查詢使用了一個ID為RF000的Runtime Filter。
CREATE TABLE test (t1 INT) DISTRIBUTED BY HASH (t1) BUCKETS 2 PROPERTIES("replication_num" = "1");
INSERT INTO test VALUES (1), (2), (3), (4);
CREATE TABLE test2 (t2 INT) DISTRIBUTED BY HASH (t2) BUCKETS 2 PROPERTIES("replication_num" = "1");
INSERT INTO test2 VALUES (3), (4), (5);
EXPLAIN SELECT t1 FROM test JOIN test2 where test.t1 = test2.t2;
+-------------------------------------------------------------------+
| Explain String |
+-------------------------------------------------------------------+
| PLAN FRAGMENT 0 |
| OUTPUT EXPRS:`t1` |
| |
| 4:EXCHANGE |
| |
| PLAN FRAGMENT 1 |
| OUTPUT EXPRS: |
| PARTITION: HASH_PARTITIONED: `default_cluster:ssb`.`test`.`t1` |
| |
| 2:HASH JOIN |
| | join op: INNER JOIN (BUCKET_SHUFFLE) |
| | equal join conjunct: `test`.`t1` = `test2`.`t2` |
| | runtime filters: RF000[in] <- `test2`.`t2` |
| | |
| |----3:EXCHANGE |
| | |
| 0:OlapScanNode |
| TABLE: test |
| runtime filters: RF000[in] -> `test`.`t1` |
| |
| PLAN FRAGMENT 2 |
| OUTPUT EXPRS: |
| PARTITION: HASH_PARTITIONED: `default_cluster:ssb`.`test2`.`t2` |
| |
| 1:OlapScanNode |
| TABLE: test2 |
+-------------------------------------------------------------------+
-- 上面`runtime filters`的行顯示了`PLAN FRAGMENT 1`的`2:HASH JOIN`生成了ID為RF000的IN predicate，
-- 其中`test2`.`t2`的key values僅在運行時可知，
-- 在`0:OlapScanNode`使用了該IN predicate用于在讀取`test`.`t1`時過濾不必要的數(shù)據(jù)。
SELECT t1 FROM test JOIN test2 where test.t1 = test2.t2;
-- 返回2行結果[3, 4];
-- 通過query的profile（set enable_profile=true;）可以查看查詢內(nèi)部工作的詳細信息，
-- 包括每個Runtime Filter是否下推、等待耗時、以及OLAP_SCAN_NODE從prepare到接收到Runtime Filter的總時長。
RuntimeFilter:in:
- HasPushDownToEngine: true
- AWaitTimeCost: 0ns
- EffectTimeCost: 2.76ms
-- 此外，在profile的OLAP_SCAN_NODE中還可以查看Runtime Filter下推后的過濾效果和耗時。
- RowsVectorPredFiltered: 9.320008M (9320008)
- VectorPredEvalTime: 364.39ms
Runtime Filter的規(guī)劃規(guī)則?
- 只支持對join on clause中的等值條件生成Runtime Filter，不包括Null-safe條件，因為其可能會過濾掉join左表的null值。
- 不支持將Runtime Filter下推到left outer、full outer、anti join的左表；
- 不支持src expr或target expr是常量；
- 不支持src expr和target expr相等；
- 不支持src expr的類型等于
HLL或者BITMAP； - 目前僅支持將Runtime Filter下推給OlapScanNode；
- 不支持target expr包含NULL-checking表達式，比如
COALESCE/IFNULL/CASE，因為當outer join上層其他join的join on clause包含NULL-checking表達式并生成Runtime Filter時，將這個Runtime Filter下推到outer join的左表時可能導致結果不正確； - 不支持target expr中的列（slot）無法在原始表中找到某個等價列；
- 不支持列傳導，這包含兩種情況：
- 一是例如join on clause包含A.k = B.k and B.k = C.k時，目前C.k只可以下推給B.k，而不可以下推給A.k；
- 二是例如join on clause包含A.a + B.b = C.c，如果A.a可以列傳導到B.a，即A.a和B.a是等價的列，那么可以用B.a替換A.a，然后可以嘗試將Runtime Filter下推給B（如果A.a和B.a不是等價列，則不能下推給B，因為target expr必須與唯一一個join左表綁定）；
- Target expr和src expr的類型必須相等，因為Bloom Filter基于hash，若類型不等則會嘗試將target expr的類型轉換為src expr的類型；
- 不支持
PlanNode.Conjuncts生成的Runtime Filter下推，與HashJoinNode的eqJoinConjuncts和otherJoinConjuncts不同，PlanNode.Conjuncts生成的Runtime Filter在測試中發(fā)現(xiàn)可能會導致錯誤的結果，例如IN子查詢轉換為join時，自動生成的join on clause將保存在PlanNode.Conjuncts中，此時應用Runtime Filter可能會導致結果缺少一些行。
Doris Join 優(yōu)化原理
Doris 支持兩種物理算子，一類是?Hash Join，另一類是?Nest Loop Join。
- Hash Join：在右表上根據(jù)等值 Join 列建立哈希表，左表流式的利用哈希表進行 Join 計算，它的限制是只能適用于等值 Join。
- Nest Loop Join：通過兩個 for 循環(huán)，很直觀。然后它適用的場景就是不等值的 Join，例如：大于小于或者是需要求笛卡爾積的場景。它是一個通用的 Join 算子，但是性能表現(xiàn)差。
作為分布式的 MPP 數(shù)據(jù)庫， 在 Join 的過程中是需要進行數(shù)據(jù)的 Shuffle。數(shù)據(jù)需要進行拆分調(diào)度，才能保證最終的 Join 結果是正確的。舉個簡單的例子，假設關系S 和 R 進行Join，N 表示參與 Join 計算的節(jié)點的數(shù)量；T 則表示關系的 Tuple 數(shù)目。
Doris Shuffle 方式?
Doris 支持 4 種 Shuffle 方式
-
Broadcast Join
它要求把右表全量的數(shù)據(jù)都發(fā)送到左表上，即每一個參與 Join 的節(jié)點，它都擁有右表全量的數(shù)據(jù)，也就是 T(R)。
它適用的場景是比較通用的，同時能夠支持 Hash Join 和 Nest loop Join，它的網(wǎng)絡開銷 N * T(R)。
左表數(shù)據(jù)不移動，右表數(shù)據(jù)發(fā)送到左表數(shù)據(jù)的掃描節(jié)點。
-
Shuffle Join
當進行 Hash Join 時候，可以通過 Join 列計算對應的 Hash 值，并進行 Hash 分桶。
它的網(wǎng)絡開銷則是：T（S） + T（R），但它只能支持 Hash Join，因為它是根據(jù) Join 的條件也去做計算分桶的。
左右表數(shù)據(jù)根據(jù)分區(qū)，計算的結果發(fā)送到不同的分區(qū)節(jié)點上。
-
Bucket Shuffle Join
Doris 的表數(shù)據(jù)本身是通過 Hash 計算分桶的，所以就可以利用表本身的分桶列的性質(zhì)來進行 Join 數(shù)據(jù)的 Shuffle。假如兩張表需要做 Join，并且 Join 列是左表的分桶列，那么左表的數(shù)據(jù)其實可以不用去移動右表通過左表的數(shù)據(jù)分桶發(fā)送數(shù)據(jù)就可以完成 Join 的計算。
它的網(wǎng)絡開銷則是：T（R）相當于只 Shuffle 右表的數(shù)據(jù)就可以了。
左表數(shù)據(jù)不移動，右表數(shù)據(jù)根據(jù)分區(qū)計算的結果發(fā)送到左表掃表的節(jié)點
-
Colocate
它與 Bucket Shuffle Join 相似，相當于在數(shù)據(jù)導入的時候，根據(jù)預設的 Join 列的場景已經(jīng)做好了數(shù)據(jù)的 Shuffle。那么實際查詢的時候就可以直接進行 Join 計算而不需要考慮數(shù)據(jù)的 Shuffle 問題了。
數(shù)據(jù)已經(jīng)預先分區(qū)，直接在本地進行 Join 計算
四種 Shuffle 方式對比?
| Shuffle方式 | 網(wǎng)絡開銷 | 物理算子 | 適用場景 |
|---|---|---|---|
| BroadCast | N * T(R) | Hash Join / Nest Loop Join | 通用 |
| Shuffle | T(S) + T(R) | Hash Join | 通用 |
| Bucket Shuffle | T(R) | Hash Join | Join條件中存在左表的分布式列，且左表執(zhí)行時為單分區(qū) |
| Colocate | 0 | Hash Join | Join條件中存在左表的分布式列，且左右表同屬于一個Colocate Group |
N ： 參與 Join 計算的 Instance 個數(shù)
T(關系) : 關系的 Tuple 數(shù)目
上面這 4 種方式靈活度是從高到低的，它對這個數(shù)據(jù)分布的要求是越來越嚴格，但 Join 計算的性能也是越來越好的。
Runtime Filter Join 優(yōu)化?
Doris 在進行 Hash Join 計算時會在右表構建一個哈希表，左表流式的通過右表的哈希表從而得出 Join 結果。而 RuntimeFilter 就是充分利用了右表的 Hash 表，在右表生成哈希表的時候，同時生成一個基于哈希表數(shù)據(jù)的一個過濾條件，然后下推到左表的數(shù)據(jù)掃描節(jié)點。通過這樣的方式，Doris 可以在運行時進行數(shù)據(jù)過濾。
假如左表是一張大表，右表是一張小表，那么利用右表生成的過濾條件就可以把絕大多數(shù)在 Join 層要過濾的數(shù)據(jù)在數(shù)據(jù)讀取時就提前過濾，這樣就能大幅度的提升 Join 查詢的性能。
當前 Doris 支持三種類型 RuntimeFilter
- 一種是 IN，很好理解，將一個 hashset 下推到數(shù)據(jù)掃描節(jié)點。
- 第二種就是 BloomFilter，就是利用哈希表的數(shù)據(jù)構造一個 BloomFilter，然后把這個 BloomFilter 下推到查詢數(shù)據(jù)的掃描節(jié)點。
- 最后一種就是 MinMax，就是個 Range 范圍，通過右表數(shù)據(jù)確定 Range 范圍之后，下推給數(shù)據(jù)掃描節(jié)點。
Runtime Filter 適用的場景有兩個要求：
- 第一個要求就是左表大右表小，因為構建 Runtime Filter是需要承擔計算成本的，包括一些內(nèi)存的開銷。
- 第二個要求就是左右表 Join 出來的結果很少，說明這個 Join 可以過濾掉左表的絕大部分數(shù)據(jù)。
當符合上面兩個條件的情況下，開啟 Runtime Filter 就能收獲比較好的效果
當 Join 列為左表的 Key 列時，RuntimeFilter 會下推到存儲引擎。Doris 本身支持延遲物化，
延遲物化簡單來說是這樣的：假如需要掃描 A、B、C 三列，在 A 列上有一個過濾條件: A 等于 2，要掃描 100 行的話，可以先把 A 列的 100 行掃描出來，再通過 A = 2 這個過濾條件過濾。之后通過過濾完成后的結果，再去讀取 B、C 列，這樣就能極大的降低數(shù)據(jù)的讀取 IO。所以說 Runtime Filter 如果在 Key 列上生成，同時利用 Doris 本身的延遲物化來進一步提升查詢的性能。
Runtime Filter 類型?
Doris 提供了三種不同的 Runtime Filter 類型：
- IN?的優(yōu)點是過濾效果明顯，且快速。它的缺點首先第一個它只適用于 BroadCast，第二，它右表超過一定數(shù)據(jù)量的時候就失效了，當前 Doris 目前配置的是1024，即右表如果大于 1024，IN 的 Runtime Filter 就直接失效了。
- MinMax?的優(yōu)點是開銷比較小。它的缺點就是對數(shù)值列還有比較好的效果，但對于非數(shù)值列，基本上就沒什么效果。
- Bloom Filter?的特點就是通用，適用于各種類型、效果也比較好。缺點就是它的配置比較復雜并且計算較高。
Join Reorder?
數(shù)據(jù)庫一旦涉及到多表 Join，Join 的順序?qū)φ麄€ Join 查詢的性能是影響很大的。假設有三張表 Join，參考下面這張圖，左邊是 a 表跟 b 張表先做 Join，中間結果的有 2000 行，然后與 c 表再進行 Join 計算。
接下來看右圖，把 Join 的順序調(diào)整了一下。把 a 表先與 c 表 Join，生成的中間結果只有 100，然后最終再與 b 表 Join 計算。最終的 Join 結果是一樣的，但是它生成的中間結果有 20 倍的差距，這就會產(chǎn)生一個很大的性能 Diff 了。
Doris 目前支持基于規(guī)則的 Join Reorder 算法。它的邏輯是：
- 讓大表、跟小表盡量做 Join，它生成的中間結果是盡可能小的。
- 把有條件的 Join 表往前放，也就是說盡量讓有條件的 Join 表進行過濾
- Hash Join 的優(yōu)先級高于 Nest Loop Join，因為 Hash join 本身是比 Nest Loop Join 快很多的。
Doris Join 調(diào)優(yōu)方法?
Doris Join 調(diào)優(yōu)的方法：
- 利用 Doris 本身提供的 Profile，去定位查詢的瓶頸。Profile 會記錄 Doris 整個查詢當中各種信息，這是進行性能調(diào)優(yōu)的一手資料。
- 了解 Doris 的 Join 機制，這也是第二部分跟大家分享的內(nèi)容。知其然知其所以然、了解它的機制，才能分析它為什么比較慢。
- 利用 Session 變量去改變 Join 的一些行為，從而實現(xiàn) Join 的調(diào)優(yōu)。
- 查看 Query Plan 去分析這個調(diào)優(yōu)是否生效。
上面的 4 步基本上完成了一個標準的 Join 調(diào)優(yōu)流程，接著就是實際去查詢驗證它，看看效果到底怎么樣。
如果前面 4 種方式串聯(lián)起來之后，還是不奏效。這時候可能就需要去做 Join 語句的改寫，或者是數(shù)據(jù)分布的調(diào)整、需要重新去 Recheck 整個數(shù)據(jù)分布是否合理，包括查詢 Join 語句，可能需要做一些手動的調(diào)整。當然這種方式是心智成本是比較高的，也就是說要在嘗試前面方式不奏效的情況下，才需要去做進一步的分析。
調(diào)優(yōu)案例實戰(zhàn)?
案例一?
一個四張表 Join 的查詢，通過 Profile 的時候發(fā)現(xiàn)第二個 Join 耗時很高，耗時 14 秒。
進一步分析 Profile 之后，發(fā)現(xiàn) BuildRows，就是右表的數(shù)據(jù)量是大概 2500 萬。而 ProbeRows （ ProbeRows 是左表的數(shù)據(jù)量）只有 1 萬多。這種場景下右表是遠遠大于左表，這顯然是個不合理的情況。這顯然說明 Join 的順序出現(xiàn)了一些問題。這時候嘗試改變 Session 變量，開啟 Join Reorder。
set enable_cost_based_join_reorder = true
這次耗時從 14 秒降到了 4 秒，性能提升了 3 倍多。
此時再 Check Profile 的時候，左右表的順序已經(jīng)調(diào)整正確，即右表是小表，左表是大表。基于小表去構建哈希表，開銷是很小的，這就是典型的一個利用 Join Reorder 去提升 Join 性能的一個場景
案例二?
存在一個慢查詢，查看 Profile 之后，整個 Join 節(jié)點耗時大概44秒。它的右表有 1000 萬，左表有 6000 萬，最終返回的結果也只有 6000 萬。
這里可以大致的估算出過濾率是很高的，那為什么 Runtime Filter 沒有生效呢？通過 Query Plan 去查看它，發(fā)現(xiàn)它只開啟了 IN 的 Runtime Filter。
當右表超過1024行的話， IN 是不生效的，所以根本起不到什么過濾的效果，所以嘗試調(diào)整 RuntimeFilter 的類型。
這里改為了 BloomFilter，左表的 6000 萬條數(shù)據(jù)過濾了 5900 萬條?；旧?99% 的數(shù)據(jù)都被過濾掉了，這個效果是很顯著的。查詢也從原來的 44 秒降到了 13 秒，性能提升了大概也是三倍多。
案例三?
下面是一個比較極端的 Case，通過一些環(huán)境變量調(diào)優(yōu)也沒有辦法解決，因為它涉及到 SQL Rewrite，所以這里列出來了原始的 SQL 。
select 100.00 * sum (case
when P_type like 'PROMOS'
then 1 extendedprice * (1 - 1 discount)
else 0
end ) / sum(1 extendedprice * (1 - 1 discount)) as promo revenue
from lineitem, part
where
1_partkey = p_partkey
and 1_shipdate >= date '1997-06-01'
and 1 shipdate < date '1997-06-01' + interval '1' month
這個 Join 查詢是很簡單的，單純的一個左右表的 Join 。當然它上面有一些過濾條件，打開 Profile 的時候，發(fā)現(xiàn)整個查詢 Hash Join 執(zhí)行了三分多鐘，它是一個 BroadCast 的 Join，它的右表有 2 億條，左表只有 70 萬。在這種情況下選擇了 Broadcast Join 是不合理的，這相當于要把 2 億條做一個 Hash Table，然后用 70 萬條遍歷兩億條的 Hash Table ，這顯然是不合理的。
為什么會產(chǎn)生不合理的 Join 順序呢？其實這個左表是一個 10 億條級別的大表，它上面加了兩個過濾條件，加完這兩個過濾條件之后， 10 億條的數(shù)據(jù)就剩 70 萬條了。但 Doris 目前沒有一個好的統(tǒng)計信息收集的框架，所以它不知道這個過濾條件的過濾率到底怎么樣。所以這個 Join 順序安排的時候，就選擇了錯誤的 Join 的左右表順序，導致它的性能是極其低下的。
下圖是改寫完成之后的一個 SQL 語句，在 Join 后面添加了一個Join Hint，在Join 后面加一個方括號，然后把需要的 Join 方式寫入。這里選擇了 Shuffle Join，可以看到右邊它實際查詢計劃里面看到這個數(shù)據(jù)確實是做了 Partition ，原先 3 分鐘的耗時通過這樣的改寫完之后只剩下 7 秒，性能提升明顯
Doris Join 調(diào)優(yōu)建議?
最后我們總結 Doris Join 優(yōu)化調(diào)優(yōu)的四點建議：文章來源：http://www.zghlxwxcb.cn/news/detail-666124.html
- 第一點：在做 Join 的時候，要盡量選擇同類型或者簡單類型的列，同類型的話就減少它的數(shù)據(jù) Cast，簡單類型本身 Join 計算就很快。
- 第二點：盡量選擇 Key 列進行 Join， 原因前面在 Runtime Filter 的時候也介紹了，Key 列在延遲物化上能起到一個比較好的效果。
- 第三點：大表之間的 Join ，盡量讓它 Colocation ，因為大表之間的網(wǎng)絡開銷是很大的，如果需要去做 Shuffle 的話，代價是很高的。
- 第四點：合理的使用 Runtime Filter，它在 Join 過濾率高的場景下效果是非常顯著的。但是它并不是萬靈藥，而是有一定副作用的，所以需要根據(jù)具體的 SQL 的粒度做開關。
- 最后：要涉及到多表 Join 的時候，需要去判斷 Join 的合理性。盡量保證左表為大表，右表為小表，然后 Hash Join 會優(yōu)于 Nest Loop Join。必要的時可以通過 SQL Rewrite，利用 Hint 去調(diào)整 Join 的順序。
?
到了這里，關于Apache Doris 入門教程34：Join 優(yōu)化的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關文章，希望大家以后多多支持TOY模板網(wǎng)！
