Clickhouse系列二：Join调优策略

背景

Clickhouse中对于Join的处理相对薄弱一些，特别是对于分布式表的Join操作。但是对于OLAP类型的数据库，Join的操作还是不可避免的。并且在业务场景下也存在多表Join等需求。所以对于Join的性能还是需要进行调优的。
Clickhouse中没有CBO(基于代价的优化器)引擎。所以对于使用者来说SQL调优要求的能力比较高。本文主要是对于Clickhouse的使用者对于Join方面的一些调优手段进行梳理总结。也便于后续Clickhouse使用者能够编写更加适用于Clickhouse的SQL语句。

Join场景一 

Join的连接条件是基于分布式表Sharding Key的情况下。 通常用户会使用主键作为Sharding Key，例如(cityHash64(主键))或者直接使用列名称，这样既可以保证数据在各个节点上面的负载均衡，也能保证相同的主键数据，一定落到同一个节点上面。 针对于这个特性。我们可以使用Colocated join(local join)的能力。

测试场景SQL描述

表的Schema信息如下所示

CREATE TABLE local.t1_local on cluster default_cluster
(
    id int,
    val int
)
ENGINE = ReplicatedMergeTree('/clickhouse/local/tables/t1_local/2/{shard}', '{replica}')
ORDER BY id
 
CREATE TABLE local.t1 on cluster default_cluster
(
    id int,
    val int
)
ENGINE =  Distributed('default_cluster', 'local', 't1_local', cityHash64(id))
 
 
CREATE TABLE local.t2_local on cluster default_cluster
(
    id int,
    val int
)
ENGINE = ReplicatedMergeTree('/clickhouse/local/tables/t2_local/2/{shard}', '{replica}')
ORDER BY id
 
CREATE TABLE local.t2 on cluster default_cluster
(
    id int,
    val int
)
ENGINE =  Distributed('default_cluster', 'local', 't2_local', cityHash64(id))

测试SQL如下所示

SELECT count(*)
FROM local.t1 AS t1
INNER JOIN local.t2 AS t2 ON t1.id = t2.id
 
Query id: e5485fde-d624-46fa-965c-2042c7078c0a
┌──count()─┐
│ 10000000 │
└──────────┘
1 rows in set. Elapsed: 1.589 sec. Processed 50.00 million rows, 200.00 MB (31.48 million rows/s., 125.90 MB/s.)

测试场景SQL调优描述

我们对于上述的SQL进行调优处理。因为Join的关联条件是基于分布式表的Sharding Key。所以使用Colocated join(local join)的能力。

Colocated join(local join)

SELECT count(*)
FROM t1
INNER JOIN
(
    SELECT id
    FROM t2 AS t
) AS t2 ON t1.id = t2.id
SETTINGS distributed_product_mode = 'local'
 
Query id: b87a924c-8475-43ae-afb2-edffc827cdc2
┌──count()─┐
│ 10000000 │
└──────────┘
1 rows in set. Elapsed: 0.508 sec. Processed 20.00 million rows, 80.00 MB (39.33 million rows/s., 157.34 MB/s.)

这里有一个点需要注意，如果发现使用Colocated join性能没有提升，一定要使用explain syntax select xxxx 看一下计划，是不是把右表变成了本地表了。如果没有变成本地表。则右表通常需要写成子查询才可以。

Join场景二

Join的连接条件不是基于分布式表Sharding Key的情况下。 这种场景通常没有办法使用Colocated join(local join)的能力。所以通常会从下面的几个方面进行优化处理。

测试场景SQL描述

我们使用SSB(Star Schema Benchmark)中的测试个案进行验证。表的Schema信息如下所示.