一文理解 Presto 兩種 JOIN 算法實現
我們在 《Presto 中支持的七種 Join 類型》 這篇文章中介紹了 Presto 可用的 JOIN 操作的基礎知識，以及如何在 SQL 查詢中使用它們。有了這些知識，我們現在可以瞭解 Presto 的內部結構以及它如何在內部執行 JOIN 操作。本文將介紹 Presto 如何執行 JOIN 操作以及用於 JOIN 的算法。
JOIN 的實現
幾乎所有的數據庫引擎一次只 JOIN 兩個表。即使在 SQL 查詢中有兩個以上的表要聯接，數據庫也會聯接前兩個表並將輸出與第三個表聯接起來，然後對其餘表繼續這樣做。數據庫工程師將連接操作中涉及的這兩個表稱爲構建表（Build Table）和探測表（Probe Table）。
Build Table
構建表是用於創建內存索引的表。通常，在讀取探測表之前必須完整讀取構建表。
Probe Table
一旦構建表被讀取並存儲在內存中，探測表就會被逐行讀取。從探測表讀取的每一行都將根據 join criteria 與構建表進行連接。
如果想及時瞭解 Spark、Hadoop 或者 HBase 相關的文章，歡迎關注微信公衆號：過往記憶大數據
Presto 使用優化後的邏輯計劃中的右表作爲構建表，將邏輯計劃中的左表作爲探測表。請注意，邏輯計劃中的表不必與它們在 SQL 查詢中的順序相同。Presto 有一些基於成本的優化器，它們可以重新排序連接以將最小的表（即構建表）保留在右側，以便它可以放入內存中。如果連接重新排序優化器被禁用或連接器特定的統計信息（例如 Hive 統計信息）被禁用，則 Presto 將不會對連接查詢重新排序。在這種情況下，建議將最小的表保留在連接的右側，以便 Presto 可以將其用作構建表。
JOIN 算法
數據庫根據數據類型和連接類型使用不同的算法來連接兩個表。例如，SQL Server 使用 Nested Loop 算法、Merge Join 算法、Hash Join 算法和 Adaptive Join 算法。在撰寫本文時，開源的 Presto SQL 引擎採用 Nested Loop 算法和 Hash Join 算法來支持 Presto 中所有不同聯接類型。本節簡要說明 Nested Loop 算法和 Hash Join 算法，並討論其他算法在 Presto 中的適用性以提高性能。
Nested Loop Algorithm
顧名思義，嵌套循環算法使用嵌套循環連接兩個表。下面使用一個數組連接示例來解釋嵌套循環連接算法。假設你有兩個整數數組，並要求你打印這些數組的笛卡爾積，你會如何解決這個問題？下面給出了一種簡單的方法來打印兩個數組的笛卡爾積。
public class IteblogNestedLoop {
public static void main(String[] args) {
// Construct two arrays
int[] tableA = {1, 2, 3, 4, 5, 6};
int[] tableB = {10, 20, 30, 40};
// Nested loop to print the Cartesian product of two arrays
for (int x : tableA) {
for (int y : tableB) {
System.out.println(x + ", " + y);
}
}
}
}上面的代碼使用兩個循環來打印兩個數組的笛卡爾積。嵌套循環算法的時間複雜度爲 O(n²)，因爲它必須將探測表中的每一行與構建表中的每一行連接起來。由於需要每個組合，交叉連接操作的執行時間複雜度不能超過 O(n²)。Presto 使用嵌套循環算法來執行 cross join 操作，這就是爲什麼如果連接表非常大，cross join 需要很長時間。由於 O(n²) 時間複雜度，不建議在沒有連接條件的情況下連接兩個大表。
Hash Join Algorithm
哈希連接算法爲構建表中的列生成哈希鍵，這些列是用於 JOIN 條件中的，比如 left.x = right.y AND left.z = right.w。每個這樣的相等條件稱爲連接相等條件（join equi criteria）。儘管 equi criteria 術語在數據庫領域被廣泛使用，但它們也被稱爲相等條件。爲了使用哈希算法，讓我們考慮一個打印所有客戶及其訂單信息的問題。這個問題中使用的 Customer 和 Order 類定義如下。請注意，這兩個類都有一個共同的屬性：custKey。
class Order {
String orderKey;
String custKey;
double totalPrice;
public Order(String orderKey, String custKey, double totalPrice) {
this.orderKey = orderKey;
this.custKey = custKey;
this.totalPrice = totalPrice;
}
@Override
public String toString() {
return "Order: " + orderKey + ", " + custKey + ", " + totalPrice;
}
}
class Customer {
String custKey;
String name;
public Customer(String custKey, String name) {
this.custKey = custKey;
this.name = name;
}
@Override
public String toString() {
return "Customer: " + name + ", " + custKey;
}
}回到問題：我們如何打印所有客戶及其訂單？瞭解嵌套循環算法後，可以簡單地在循環內應用帶有 if 條件的嵌套循環算法，如下所示：
import java.util.*;
public class IteblogHashJoin {
public static void main(String[] args) {
List<Customer> probe = List.of(new Customer("c_001", "Alice"),
new Customer("c_002", "Bob"),
new Customer("c_003", "David"));
List<Order> build = List.of(new Order("o_01", "c_001", 100.0),
new Order("o_01", "c_001", 100.0),
new Order("o_02", "c_001", 150.0),
new Order("o_03", "c_002", 90.0),
new Order("o_04", "c_003", 120.0));
// Nested loop join
for (Customer customer : probe) {
for (Order order : build) {
if (Objects.equals(customer.custKey, order.custKey)) {
System.out.println(customer + " -> " + order);
}
}
}
}
}儘管嵌套循環連接可以達到我們的要求，但它的效率很低，因爲它在給定 n 個客戶和 n 個訂單的情況下迭代 n² 次。一個有效的解決方案可以使用一個 Hashtable 來存儲所有訂單，使用相同的連接條件：custKey 作爲哈希鍵。然後在遍歷 Customer 列表時，可以生成 Customer 的散列值。獲取具有相同 custKey 的訂單列表，如下所示：
import java.util.*;
public class IteblogHashJoin {
public static void main(String[] args) {
List<Customer> probe = List.of(new Customer("c_001", "Alice"),
new Customer("c_002", "Bob"),
new Customer("c_003", "David"));
List<Order> build = List.of(new Order("o_01", "c_001", 100.0),
new Order("o_01", "c_001", 100.0),
new Order("o_02", "c_001", 150.0),
new Order("o_03", "c_002", 90.0),
new Order("o_04", "c_003", 120.0));
// Build the hash map index
Map<Integer, List<Order>> index = new Hashtable<>();
for (Order order : build) {
int hash = Objects.hash(order.custKey);
index.putIfAbsent(hash, new LinkedList<>());
index.get(hash).add(order);
}
// Hash Join algorithm
for (Customer customer : probe) {
int hash = Objects.hash(customer.custKey);
List<Order> orders = index.get(hash);
if (orders != null) {
for (Order order : orders) {
if (Objects.equals(customer.custKey, order.custKey)) {
System.out.println(customer + " -> " + order);
}
}
}
}
}
}在上述算法中，使用單獨的 LinkedList 來避免哈希衝突，因爲同一客戶下多個訂單的可能性很高。使用 equijoin criteria 裏面列的哈希值用於將構建表存儲在存儲桶中。然後將相同的散列算法應用於探測表的 equijoin criteria 列以查找包含匹配項的桶。儘管 Hash Join 算法的最壞情況時間複雜度是 O(n²)，但平均情況下預計爲 O(n)。
上述問題可以定義爲下面給出的 SQL 查詢，以將 Customer 表與 Orders 表連接起來。
SELECT *
FROM iteblog.customer c
LEFT JOIN iteblog.orders o
ON c.custkey=o.orderkey;具有等連接條件的所有連接操作都使用 Presto 中的哈希連接算法執行。然而，連接操作並不侷限於等效連接標準。例如，如果列值大於或小於另一列的值，則可以連接兩個表，如下查詢所示:
所有具有 equijoin criteria 的連接操作都使用 Presto 中的哈希連接算法執行。但是，連接操作不限於 equijoin criteria。例如，如果列值大於或小於另一列的值，則可以連接兩個表，如下面的查詢：
SELECT o.orderkey, l.linenumber
FROM iteblog.orderkey o
LEFT JOIN iteblog.lineitem l
ON o.orderdate < l.shipdate;Hash Join 算法不適用於具有不等式約束的 join 條件。首先，很難提出一個完美的散列算法來保持輸入的不等式屬性（即給定 x > b 並不能保證 hash(a) > hash(b)）。其次，即使我們提出了一個滿足不等式要求的散列函數，我們也不能簡單地連接一個桶中的所有值。要加入不相等的行，應該匹配大於 / 小於給定列的每一行。因此，Presto 使用帶 filter 的嵌套循環算法而不是散列連接算法來執行具有非等連接條件的連接。
儘管開源的 Presto SQL 僅使用 Nested Loop 算法和 Hash Join 算法進行連接操作，但 Merge Join 是關係數據庫中使用的另一種衆所周知的算法，有一些大數據計算引擎也支持 Merge Join ，比如 Spark。以下部分介紹了 Merge Join 算法，並解釋了 Presto 社區爲何不考慮添加對 Merge Join 算法的支持。
Merge Join
Merge Join 算法來自著名的 Merge-Sort 算法。歸併排序算法有兩個階段：排序和合並。假設兩個數組已經排序，它們可以以 O(n) 的時間複雜度合併。Presto 可以通過使用 equijoin criteria 中使用的列對構建表和探測表進行排序，然後通過執行合併操作來實現該算法。忽略排序部分，merge join 算法的性能有望優於上述算法，但 Presto 社區發現它需要在內存中對兩個表進行排序，這在大數據世界中很耗時，考慮到有限的內存，甚至可能是不可行的。但是，如果有機會在底層數據源中對數據進行排序，則合併連接算法可能是一個更好的候選算法。
在我看來，如果構建表足夠小可以容納在內存中，那麼對它進行排序並使用二分搜索算法將探測錶行與構建表進行比較不會是一個糟糕的選擇。它可以改進具有不等式條件（例如大於或小於）的連接操作。Presto 還支持關係數據庫，與大數據存儲相比，這些數據庫的數據量通常較少。如果連接來自關係數據庫的兩個表，或者來自關係數據庫的表與來自 Hadoop 文件存儲的表連接，則有機會要求底層關係數據庫返回排序結果。因此，我覺得即使在大數據領域，Merge Join 仍然是一個值得考慮的候選。
本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/1Yrmr8RzqVYKtAtqm8DVyA