此博客文章是系列文章的一部分
- ClickHouse 支持的 JOIN 类型
- ClickHouse Join 内幕 - 哈希 Join、并行哈希 Join、Grace 哈希 Join
- ClickHouse Join 内幕 - 完全排序合并 Join、部分合并 Join
- 选择正确的 Join 算法
这篇文章完成了我们对 ClickHouse 开发的 6 种不同 Join 算法的探索。作为提醒：这些算法决定了 Join 查询的计划和执行方式。ClickHouse 可以配置为自适应地选择并在运行时动态更改要使用的 Join 算法，具体取决于资源可用性和使用情况。但是，ClickHouse 也允许用户指定所需的 Join 算法。此图表概述了这些算法，基于它们的相对内存消耗和执行时间： 
在我们的第二篇文章中，我们详细描述并比较了上图中基于内存哈希表的三个 ClickHouse Join 算法
提醒一下：哈希 Join 和 并行哈希 Join 速度很快，但受内存限制。来自右侧表的连接数据需要装入内存。Grace 哈希 Join 是一个不受内存限制的版本，它将数据临时溢出到磁盘，而无需对数据进行任何排序。这克服了其他 Join 算法的一些性能挑战，这些算法将数据溢出到磁盘并需要预先对数据进行排序。
在第三篇文章中，我们探索并比较了上图中基于外部排序的两个算法
提醒一下：完全排序合并 Join 是不受内存限制的，并且基于内存或外部排序，并且可以利用连接表的物理行顺序，并跳过排序阶段。在这种情况下，Join 性能可以与上图中的一些哈希 Join 算法相媲美，同时通常需要少得多的主内存。部分合并 Join 针对最小化连接大表时的内存使用量进行了优化，并且始终首先通过外部排序完全排序右表。左表也始终按块在内存中排序。如果左表的物理行顺序与 Join 键排序顺序匹配，则 Join 匹配过程将更有效率。
我们将最好的留到了最后，并在本文中完成对 ClickHouse Join 算法的探索，描述上图中 ClickHouse 最快的 Join 算法
- 直接 Join
当右侧表的底层存储支持低延迟键值请求时，可以使用直接 Join 算法。特别是对于大型右表，直接 Join 在执行时间方面具有显着改进，击败了所有其他 ClickHouse Join 算法。
测试设置
我们使用与第二篇文章中介绍的相同的两个表和相同的 ClickHouse Cloud 服务实例。
对于所有示例查询运行，我们使用 max_threads 的默认设置。执行查询的节点有 30 个 CPU 核心，因此默认的 max_threads 设置为 30。对于所有 查询管道 可视化，为了保持简洁和可读性，我们使用设置 max_threads = 2 人为地限制 ClickHouse 查询管道中使用的并行级别。
直接 Join
描述
当右侧表的底层存储支持低延迟键值请求时，可以使用直接 Join 算法。ClickHouse 有三种 表引擎 提供此功能：Join（这基本上是一个预先计算的 哈希表）、EmbeddedRocksDB 和 Dictionary。我们将在此处基于字典描述直接 Join 算法，但所有三个引擎的机制都相同。
字典是 ClickHouse 的一个 关键功能，它提供了来自各种内部和外部 源的数据的内存键值表示，针对超低延迟查找查询进行了优化。
这在各种场景中都很方便，例如，用于动态丰富摄取的数据，而不会减慢摄取过程，以及用于提高一般查询的性能，尤其是 JOIN 特别受益。
我们在下面草绘了直接 Join 算法的 查询管道： 
直接 Join 算法要求右表由字典支持，这样来自该表的要连接的数据已经以低延迟键值数据结构的形式存在于内存中。然后 ① 来自左表的所有数据都由 2 个流（因为 max_threads = 2）并行流式传输到查询引擎中，并且行通过对右表的基础字典进行查找，由两个 Join 阶段 ② 并行连接。
支持的 Join 类型
仅支持 LEFT ANY Join 类型。请注意，Join 键需要与底层键值存储的键属性匹配。
示例
为了演示直接 Join，我们需要首先创建一个字典。为此，我们需要选择一个布局，该布局确定字典内容在内存中的存储方式。我们将使用 flat 选项，并为了比较，也使用 hashed 布局。两种布局都要求键属性的数据类型与 UInt64 类型兼容。flat 布局在所有布局选项中提供最佳性能，并分配一个内存数组，该数组具有与键属性的最大值一样多的条目空间。例如，如果最大值为 10 万，则数组将有 10 万个条目的空间。这种数据布局允许极快的键值查找，O(1) 时间复杂度，因为只需要一个简单的数组偏移量查找。偏移量只是提供的键的值，数组中该偏移量位置的条目包含相应的值。这非常适合我们的演员和角色数据，其中我们在键列（id 和 actor_id， 分别）在我们的源表中。因此，每个分配的数组条目都将被使用。使用 hashed 布局，字典内容存储在哈希表中。hashed 布局更通用。例如，对于非密集键属性值（不从 0 开始），不会在内存中分配不必要的空间。但是，正如我们稍后将看到的，访问速度会慢 2-5 倍。
我们创建一个具有 flat 布局的字典，该字典将来自 roles 表的内容完全加载到内存中，以进行低延迟键值查找。我们使用 actor_id 作为键属性。请注意，我们使用 max_array_size 设置来指定初始和最大数组大小（默认值 500,000 太小）。我们还通过将 LIFETIME 设置为 0 来禁用字典的 内容更新
CREATE DICTIONARY imdb_large.roles_dict_flat
(
created_at DateTime,
actor_id UInt32,
movie_id UInt32,
role String
)
PRIMARY KEY actor_id
SOURCE(CLICKHOUSE(db 'imdb_large' table 'roles'))
LIFETIME(0)
LAYOUT(FLAT(INITIAL_ARRAY_SIZE 1_000_000 MAX_ARRAY_SIZE 1_000_000));
接下来，我们创建一个类似的字典，但使用 hashed 布局
CREATE DICTIONARY imdb_large.roles_dict_hashed
(
created_at DateTime,
actor_id UInt32,
movie_id UInt32,
role String
)
PRIMARY KEY actor_id
SOURCE(CLICKHOUSE(db 'imdb_large' table 'roles'))
LIFETIME(0)
LAYOUT(hashed());
请注意，在 ClickHouse Cloud 中，字典将自动在所有节点上创建。对于 OSS，如果使用 Replicated 数据库，则此行为是可能的。其他配置将需要在所有节点上手动或通过使用 ON CLUSTER 子句创建字典。
我们查询 dictionaries 系统表 以检查一些指标
SELECT
name,
status,
formatReadableSize(bytes_allocated) AS memory_allocated,
formatReadableTimeDelta(loading_duration) AS loading_duration
FROM system.dictionaries
WHERE startsWith(name, 'roles_dict_')
ORDER BY name;
┌─name──────────────┬─status─┬─memory_allocated─┬─loading_duration─┐
│ roles_dict_flat │ LOADED │ 1.52 GiB │ 12 seconds │
│ roles_dict_hashed │ LOADED │ 128.00 MiB │ 6 seconds │
└───────────────────┴────────┴──────────────────┴──────────────────┘
loading_duration 列显示将源表内容加载到字典的内存布局中花费了多长时间。status 指示加载已完成。我们可以看到为字典分配了多少主内存空间。
使用上述字典 DDL 创建字典会自动创建一个表，该表具有由字典支持的 dictionary 表引擎。我们通过查询 tables 系统表来验证这一点
SELECT
name,
engine
FROM system.tables
WHERE startsWith(name, 'roles_dict_')
ORDER BY name;
┌─name──────────────┬─engine─────┐
│ roles_dict_flat │ Dictionary │
│ roles_dict_hashed │ Dictionary │
└───────────────────┴────────────┘
有了这样的表，字典可以作为一流的表实体进行处理，并且可以使用熟悉的 SELECT 子句直接读取数据。
请注意，与普通（MergeTree 引擎系列）ClickHouse 表相比，键属性在字典中是（自动）唯一的。例如，roles 表包含许多具有相同 actor_id 值的行，因为通常演员/女演员有多个角色。当这些行加载到以 actor_id 作为键属性的字典中时，具有相同键值的行会相互覆盖。实际上，字典中仅包含为特定 actor_id 插入的最后一行的数据。
我们可以通过从字典的 roles 源表和自动创建的字典表中选择计数来验证这一点
SELECT formatReadableQuantity(count()) as count FROM roles;
┌─count──────────┐
│ 100.00 million │
└────────────────┘
SELECT formatReadableQuantity(count()) as count FROM roles_dict_flat;
┌─count────────┐
│ 1.00 million │
└──────────────┘
100 万正是 actors 表中唯一演员的数量。这意味着 roles 字典包含每个演员/女演员一个角色的数据
SELECT formatReadableQuantity(count()) as count FROM actors;
┌─count────────┐
│ 1.00 million │
└──────────────┘
现在我们使用字典来丰富来自 actors 表的行，其中包含来自 roles 表的信息。请注意，我们使用 dictGet 函数执行低延迟键值查找。对于来自 actors 表的每一行，我们在字典中使用来自 id 列的值执行查找，并请求 created_at、movie_id 和 role 值，以 元组 的形式返回
WITH T1 AS (
SELECT
id,
first_name,
last_name,
gender,
dictGet('roles_dict_flat', ('created_at', 'movie_id', 'role'), id) as t
FROM actors)
SELECT
id,
first_name,
last_name,
gender,
id AS actor_id,
t.1 AS created_at,
t.2 AS movie_id,
t.3 AS role
FROM T1
LIMIT 1
FORMAT Vertical;
Row 1:
──────
id: 393216
first_name: Wissia
last_name: Breitenreiter
gender: F
actor_id: 393216
created_at: 2023-05-12 13:03:09
movie_id: 373614
role: Gaston Binet
1 row in set. Elapsed: 0.019 sec. Processed 327.68 thousand rows, 12.74 MB (17.63 million rows/s., 685.25 MB/s.)
请注意，如果字典不包含特定演员 id 值的键条目，则为请求的属性返回配置的 默认值。此外，由于如上所述，字典根据 actor_id 列对加载的数据进行重复数据删除，因此实际上只返回第一个找到的匹配项。因此，上述查询的行为等同于 LEFT ANY JOIN。
在 ClickHouse 中，有一种更简单、更紧凑的方式来制定上述查询。我们之前展示过，当创建一个具有特定名称的字典时，ClickHouse 会自动创建一个同名的表，该表通过字典表引擎由字典支持。此表允许我们通过将 Join 查询与 direct Join 算法结合使用来表达与上述查询相同的逻辑
SELECT *
FROM actors AS a
JOIN roles_dict_flat AS r ON a.id = r.actor_id
LIMIT 1
SETTINGS join_algorithm='direct'
FORMAT Vertical;
Row 1:
──────
id: 393216
first_name: Wissia
last_name: Breitenreiter
gender: F
actor_id: 393216
created_at: 2023-05-12 13:03:09
movie_id: 373614
role: Gaston Binet
1 row in set. Elapsed: 0.023 sec. Processed 327.68 thousand rows, 12.74 MB (14.28 million rows/s., 555.30 MB/s.)
在内部，ClickHouse 正在使用高效的键值查找来实现 Join，查找进入支持右侧表的字典。这类似于上面使用 dictGet 函数进行查找的查询。我们可以通过使用 EXPLAIN PLAN 子句内省 Join 查询的 查询计划 来验证这一点
EXPLAIN PLAN
SELECT *
FROM actors AS a
JOIN roles_dict_flat AS r ON a.id = r.actor_id
SETTINGS join_algorithm='direct';
┌─explain───────────────────────────────────────────────┐
│ Expression ((Projection + Before ORDER BY)) │
│ FilledJoin (JOIN) │
│ Expression ((Convert JOIN columns + Before JOIN)) │
│ ReadFromMergeTree (imdb_large.actors) │
└───────────────────────────────────────────────────────┘
我们可以看到 ClickHouse 正在使用一个特殊的 FilledJoin 步骤，表明无需执行任何操作来准备或加载右侧表，因为它的内容已经以非常快速的键值查找数据结构的形式存在于内存中。准备就绪，非常适合执行 Join。
为了进行比较，我们可以内省使用哈希算法的相同 Join 查询的查询计划
EXPLAIN PLAN
SELECT *
FROM actors AS a
JOIN roles AS r ON a.id = r.actor_id
SETTINGS join_algorithm='hash';
┌─explain──────────────────────────────────────────────────────────────────────────────────────┐
│ Expression ((Projection + Before ORDER BY)) │
│ Join (JOIN FillRightFirst) │
│ Expression (Before JOIN) │
│ ReadFromMergeTree (imdb_large.actors) │
│ Expression ((Joined actions + (Rename joined columns + (Projection + Before ORDER BY)))) │
│ ReadFromMergeTree (imdb_large.roles) │
└──────────────────────────────────────────────────────────────────────────────────────────────
现在我们看到一个 JOIN FillRightFirst 步骤，表明来自右侧表的数据将首先加载到内存中（到哈希表中），然后才能执行哈希 Join。
我们现在将比较使用以下算法的相同 Join 查询的执行时间
请注意，如上所述，与字典支持的右侧表进行的直接 Join 实际上是 LEFT ANY JOIN。为了公平比较，因此，我们在使用哈希算法的查询运行时使用此 Join 类型。
我们运行哈希 Join
SELECT *
FROM actors AS a
LEFT ANY JOIN roles AS r ON a.id = r.actor_id
SETTINGS join_algorithm='hash'
FORMAT Null;
0 rows in set. Elapsed: 1.133 sec. Processed 101.00 million rows, 3.67 GB (89.13 million rows/s., 3.24 GB/s.)
我们运行并行哈希 Join
SELECT *
FROM actors AS a
LEFT ANY JOIN roles AS r ON a.id = r.actor_id
SETTINGS join_algorithm='parallel_hash'
FORMAT Null;
0 rows in set. Elapsed: 0.690 sec. Processed 101.00 million rows, 3.67 GB (146.38 million rows/s., 5.31 GB/s.)
我们运行直接 Join，右侧表具有带 hashed 内存布局的底层字典
SELECT *
FROM actors AS a
JOIN roles_dict_hashed AS r ON a.id = r.actor_id
SETTINGS join_algorithm='direct'
FORMAT Null;
0 rows in set. Elapsed: 0.113 sec. Processed 1.00 million rows, 38.87 MB (8.87 million rows/s., 344.76 MB/s.)
最后，我们运行直接 Join，右侧表具有带 flat 内存布局的底层字典
SELECT *
FROM actors AS a
JOIN roles_dict_flat AS r ON a.id = r.actor_id
SETTINGS join_algorithm='direct'
FORMAT Null;
0 rows in set. Elapsed: 0.044 sec. Processed 1.00 million rows, 38.87 MB (22.97 million rows/s., 892.85 MB/s.)
现在让我们检查最后四个查询运行的运行时统计信息
SELECT
query,
query_duration_ms,
(query_duration_ms / 1000)::String || ' s' AS query_duration_s,
formatReadableSize(memory_usage) AS memory_usage,
formatReadableQuantity(read_rows) AS read_rows,
formatReadableSize(read_bytes) AS read_data
FROM clusterAllReplicas(default, system.query_log)
WHERE (type = 'QueryFinish') AND (hasAll(tables, ['imdb_large.actors', 'imdb_large.roles']) OR arrayExists(t -> startsWith(t, 'imdb_large.roles_dict_'), tables))
ORDER BY initial_query_start_time DESC
LIMIT 4
FORMAT Vertical;
Row 1:
──────
query: SELECT *
FROM actors AS a
JOIN roles_dict_flat AS r ON a.id = r.actor_id
SETTINGS join_algorithm='direct'
FORMAT Null;
query_duration_ms: 44
query_duration_s: 0.044 s
memory_usage: 83.66 MiB
read_rows: 1.00 million
read_data: 37.07 MiB
Row 2:
──────
query: SELECT *
FROM actors AS a
JOIN roles_dict_hashed AS r ON a.id = r.actor_id
SETTINGS join_algorithm='direct'
FORMAT Null;
query_duration_ms: 113
query_duration_s: 0.113 s
memory_usage: 102.90 MiB
read_rows: 1.00 million
read_data: 37.07 MiB
Row 3:
──────
query: SELECT *
FROM actors AS a
LEFT ANY JOIN roles AS r ON a.id = r.actor_id
SETTINGS join_algorithm='parallel_hash'
FORMAT Null;
query_duration_ms: 689
query_duration_s: 0.689 s
memory_usage: 4.78 GiB
read_rows: 101.00 million
read_data: 3.41 GiB
Row 4:
──────
query: SELECT *
FROM actors AS a
LEFT ANY JOIN roles AS r ON a.id = r.actor_id
SETTINGS join_algorithm='hash'
FORMAT Null;
query_duration_ms: 1084
query_duration_s: 1.084 s
memory_usage: 4.44 GiB
read_rows: 101.00 million
read_data: 3.41 GiB
来自第 1 行的直接 Join 运行，右侧表由具有 flat 内存布局的字典支持，比来自第 3 行的并行哈希 Join 运行快约 15 倍，比来自第 4 行的哈希 Join 运行快约 25 倍，并且比来自第 2 行的直接 Join 运行快约 2.5 倍，其中右侧表由具有 hashed 内存布局的字典支持。真快！
主要原因是右侧表的数据已在内存中。相反，哈希和并行哈希算法需要首先将数据加载到内存中。此外，如前所述，具有 flat 布局的字典的内存数组允许极快的键值查找，O(1) 时间复杂度，因为只需要一个简单的数组偏移量查找。
请注意，来自 query_log 系统表的 memory_usage 列不考虑字典本身分配的内存。因此，为了进行公平的峰值内存消耗比较，我们需要添加 dictionaries 系统表的 bytes_allocated 列中的相应值 - 请参阅我们上面对该系统表的查询。我们在本文的摘要部分中进一步执行此操作。正如您将看到的，即使将字典的 bytes_allocated 添加到直接 Join 运行的 memory_usage 中，峰值内存消耗也明显低于哈希和并行哈希 Join 运行。
查询管道
让我们内省 max_threads 设置为 2 的直接 Join 查询的实际查询管道
clickhouse client --host ekyyw56ard.us-west-2.aws.clickhouse.cloud --secure --port 9440 --password <PASSWORD> --database=imdb_large --query "
EXPLAIN pipeline graph=1, compact=0
SELECT *
FROM actors AS a
JOIN roles_dict_flat AS r ON a.id = r.actor_id
SETTINGS max_threads = 2, join_algorithm = 'direct';" | dot -Tpdf > pipeline.pdf
我们使用上面抽象图中使用的相同带圈数字注释了管道，稍微简化了主要阶段的名称，并添加了字典和左表，以便对齐两个图表： 
我们看到真实的查询管道与我们上面的抽象版本匹配。
摘要
这篇博客文章描述了 ClickHouse 最快的 Join 算法：直接 Join。当右侧表的底层存储支持低延迟键值请求时，此算法适用。特别是对于大型右侧表，直接 Join 在执行时间方面具有显着改进，击败了所有其他 ClickHouse Join 算法。
下图总结并比较了本文 Join 查询运行的内存使用量和执行时间。为此，我们始终运行相同的查询，连接相同的数据，右侧的表较大，在具有 30 个 CPU 核心的节点上（因此 max_threads 设置为 30）： 
上图非常清楚。direct Join 速度最快。① 对于由具有 flat 内存布局的字典支持的右侧表，该算法比 hash Join 快约 25 倍，比 parallel hash 快约 15 倍，并且比 ② 直接 Join 快约 2.5 倍，其中右侧表由具有 hashed 内存布局的字典支持。无论字典布局类型如何，总体峰值内存消耗（包括添加到直接 Join 运行的 memory_usage 中的字典的 bytes_allocated）都低于哈希算法运行。
这结束了我们对 6 个 ClickHouse Join 算法的三部分深入探讨。
在本系列的下一篇文章中，我们将总结并直接比较所有 6 个 ClickHouse Join 算法。我们还将提供一个方便的决策树 + Join 类型支持概述，您可以将其用于决定哪种 Join 算法最适合您的特定场景。
敬请关注！
