ClickHouse 内幕（3）基于索引的查询优化

ClickHouse 索引采用唯一聚簇索引的方式，即 Part 内数据按照 order by keys 有序，在整个查询计划中，如果算子能够有效利用输入数据的有序性，对算子的执行性能将有巨大的提升。本文讨论 ClickHouse 基于索引的查询算子优化方式。

在整个查询计划中 Sort、Distinct、聚合这 3 个算子相比其他算子比如：过滤、projection 等有如下几个特点：1.算子需要再内存中保存状态，内存代价高；2.算子计算代价高；3.算子会阻断执行 pipeline，待所有数据计算完整后才会向下游输出数据。所以上算子往往是整个查询的瓶颈算子。

本文详细讨论，3 个算子基于索引的查询优化前后，在计算、内存和 pipeline 阻断上的影响。

实验前准备：

后续的讨论主要基于实验进行。

CREATE TABLE test_in_order(    `a` UInt64,    `b` UInt64,    `c` UInt64,    `d` UInt64)ENGINE = MergeTreeORDER BY (a, b);

表中总共有 3 个 part，每个 part 数据量 4 条。

PS： 用户可以在插入数据前提前关闭后台 merge，以避免 part 合并成一个，如果 part 合并成一个将影响查询并行度，可能对实验有影响，以下查询可以关闭后台 merge：system stop merges test_in_order

一、Sort 算子

如果 order by 查询的 order by 字段与表的 order by keys 的前缀列匹配，那么可以根据数据的有序特性对 Sort 算子进行优化。

1.Sort 算子实现方式

首先看下不能利用主键有序性的场景，即对于 order by 查询的 order by 字段与表的 order by keys 的前缀列不匹配。比如下面的查询：

query_1: EXPLAIN PIPELINE SELECT b FROM read_in_order ORDER BY b ASC

它的执行计划如下：

┌─explain───────────────────────────────┐│ (Expression)                          ││ ExpressionTransform                   ││   (Sorting)                           ││   MergingSortedTransform 3 → 1        ││     MergeSortingTransform × 3         ││       LimitsCheckingTransform × 3     ││         PartialSortingTransform × 3   ││           (Expression)                ││           ExpressionTransform × 3     ││             (ReadFromMergeTree)       ││             MergeTreeThread × 3 0 → 1 │└───────────────────────────────────────┘

排序算法由 3 个 Transform 组成，其中

1）PartialSortingTransform 对单个 Chunk 进行排序；

2）MergeSortingTransform 对单个 stream 进行排序；

3）MergingSortedTransform 合并多个有序的 stream 进行全局 sort-merge 排序

如果查询的 order by 字段与表的 order by keys 的前缀列匹配，那么可以根据数据的有序特性对查询进行优化，优化开关：optimize_read_in_order。

2.匹配索引列的查询

以下查询的 order by 字段与表的 order by keys 的前缀列匹配

query_3: EXPLAIN PIPELINE SELECT b FROM test_in_order ORDER BY a ASC, b ASCSETTINGS optimize_read_in_order = 0 -- 关闭read_in_order优化

查看 order by 语句的 pipeline 执行计划

┌─explain───────────────────────────┐│ (Expression)                      ││ ExpressionTransform               ││   (Sorting)                       ││   MergingSortedTransform 3 → 1    ││     MergeSortingTransform × 3     ││       (Expression)                ││       ExpressionTransform × 3     ││         (ReadFromMergeTree)       ││         MergeTreeThread × 3 0 → 1 │└───────────────────────────────────┘

此时 order by 算子的算法

1）首先 MergeSortingTransform 对输入的 stream 进行排序

2）然后 MergingSortedTransform 将多个排好序的 stream 进行合并，并输出一个整体有序的 stream，也是最终的排序结果。

这里有个疑问在关闭 read_in_order 优化的查询计划中，系统直接默认了 MergeSortingTransform 的输入在 Chunk 内是有序的，这里其实是一个默认优化，因为 order by 查询的 order by 字段与表的 order by keys 的前缀列匹配，所以数据在 Chunk 内部一定是有序的。

3. 开启优化 optimize_read_in_order

┌─explain──────────────────────────┐│ (Expression)                     ││ ExpressionTransform              ││   (Sorting)                      ││   MergingSortedTransform 3 → 1   ││     (Expression)                 ││     ExpressionTransform × 3      ││       (ReadFromMergeTree)        ││       MergeTreeInOrder × 3 0 → 1 │└──────────────────────────────────┘

4. 优化分析

打开 optimize_read_in_order 后：

1.对于计算方面：算法中只有一个 MergingSortedTransform，省略了单个 stream 内排序的步骤

2.由于内存方面：由于 MergeSortingTransform 是消耗内存最大的步骤，所以优化后可以节约大量的内存

3.对于 poipeline 阻塞：MergeSortingTransform 会阻塞整个 pipeline，所以优化后也消除了对 pipeline 的阻塞

二、Distinct 算子

如果 distinct 查询的 distinct 字段与表的 order by keys 的前缀列匹配，那么可以根据数据的有序特性对 Distinct 算子进行优化，优化开关：optimize_distinct_in_order。通过以下实验进行说明：

1. Distinct 算子实现方式

查看 distinct 语句的 pipeline 执行计划

query_2: EXPLAIN PIPELINE SELECT DISTINCT * FROM woo.test_in_order SETTINGS optimize_distinct_in_order = 0 -- 关闭distinct in order优化

┌─explain─────────────────────────────┐│ (Expression)                        ││ ExpressionTransform                 ││   (Distinct)                        ││   DistinctTransform                 ││     Resize 3 → 1                    ││       (Distinct)                    ││       DistinctTransform × 3         ││         (Expression)                ││         ExpressionTransform × 3     ││           (ReadFromMergeTree)       ││           MergeTreeThread × 3 0 → 1 │└─────────────────────────────────────┘

Distinct 算子采用两阶段的方式，首先第一个 DistinctTransform 在内部进行初步 distinct，其并行度为 3，可以简单的认为有 3 个线程在同时执行。然后第二个 DistinctTransform 进行 final distinct。

每个 DistinctTransform 的计算方式为：首先构建一个 HashSet 数据结构，然后根据 HashSet，构建一个 Filter Mask（如果当前 key 存在于 HashSet 中，则过滤掉），最后过滤掉不需要的数据。

2.开启优化 optimize_distinct_in_order

┌─explain────────────────────────────────┐│ (Expression)                           ││ ExpressionTransform                    ││   (Distinct)                           ││   DistinctTransform                    ││     Resize 3 → 1                       ││       (Distinct)                       ││       DistinctSortedChunkTransform × 3 ││         (Expression)                   ││         ExpressionTransform × 3        ││           (ReadFromMergeTree)          ││           MergeTreeThread × 3 0 → 1    │└────────────────────────────────────────┘

可以看到初步 distinct 和 final distinct 采用了不同的 transform，DistinctSortedChunkTransform 和 DistinctTransform。

DistinctSortedChunkTransform：对单个 stream 内的数据进行 distinct 操作，因为 distinct 列跟表的 order by keys 的前缀列匹配，scan 算子读取数据的时候一个 stream 只从一个 part 内读取数据，那么每个 distinct transform 输入的数据就是有序的。所以 distinct 算法有：

DistinctSortedChunkTransform 算法一：

Transform 中保留最后一个输入的数据作为状态，对于每个输入的新数据如果跟保留的状态相同，那么忽略，如果不同则将上一个状态输出给上一个算子，然后保留当前的数据最为状态。这种算法对于在整个 stream 内部全局去重时间和空间复杂度都有极大的降低。

DistinctSortedStreamTransform 算法二：(ClickHouse 采用的)

Transform 对与每个 Chunk（ClickHouse 中 Transform 数据处理的基本单位，默认大约 6.5w 行），首先将相同的数据划分成多个 Range，并设置一个 mask 数组，然后将相同的数据删除掉，最后返回删除重复数据的 Chunk。

3. 优化分析

打开 optimize_distinct_in_order 后：主要对于第一阶段的 distinct 步骤进行了优化，从基于 HashSet 过滤的算法到基于连续相同值的算法。

1.对于计算方面：优化后的算法，省去了 Hash 计算，但多了判断相等的步骤，在不同数据基数集大小下，各有优劣。

2.由于内存方面：优化后的算法，不需要存储 HashSet

3.对于 poipeline 阻塞：优化前后都不会阻塞 pipeline

三、聚合算子

如果 group by 查询的 order by 字段与表的 order by keys 的前缀列匹配，那么可以根据数据的有序特性对聚合算子进行优化，优化开关：optimize_aggregation_in_order。

1.聚合算子实现方式

查看 group by 语句的 pipeline 执行计划：

query_4: EXPLAIN PIPELINE SELECT a FROM test_in_order GROUP BY a SETTINGS optimize_aggregation_in_order = 0 -- 关闭read_in_order优化

┌─explain─────────────────────────────┐│ (Expression)                        ││ ExpressionTransform × 8             ││   (Aggregating)                     ││   Resize 3 → 8                      ││     AggregatingTransform × 3        ││       StrictResize 3 → 3            ││         (Expression)                ││         ExpressionTransform × 3     ││           (ReadFromMergeTree)       ││           MergeTreeThread × 3 0 → 1 │└─────────────────────────────────────┘

对于聚合算子的整体算法没有在执行计划中完整显示出来，其宏观上采用两阶段的聚合算法，其完整算法如下：1.AggregatingTransform 进行初步聚合，这一步可以并行计算；2.ConvertingAggregatedToChunksTransform 进行第二阶段聚合。（PS：为简化起见，忽略 two level HashMap，和 spill to disk 的介绍）。

2.开启优化 optimize_aggregation_in_order

执行计划如下：

┌─explain───────────────────────────────────────┐│ (Expression)                                  ││ ExpressionTransform × 8                       ││   (Aggregating)                               ││   MergingAggregatedBucketTransform × 8        ││     Resize 1 → 8                              ││       FinishAggregatingInOrderTransform 3 → 1 ││         AggregatingInOrderTransform × 3       ││           (Expression)                        ││           ExpressionTransform × 3             ││             (ReadFromMergeTree)               ││             MergeTreeInOrder × 3 0 → 1        │└───────────────────────────────────────────────┘

可以看到打开 optimize_aggregation_in_order 后 aggregating 算法由三个步骤组成：

1）首先 AggregatingInOrderTransform 会将 stream 内连续的相同的 key 进行预聚合，预聚合后在当前 stream 内相同 keys 的数据只会有一条；

2）FinishAggregatingInOrderTransform 将接收到的多个 stream 内的数据进行重新分组使得输出的 chunk 间数据是有序的，假设前一个 chunk 中 group by keys 最大的一条数据是 5，当前即将输出的 chunk 中没有大于 5 的数据；

3）MergingAggregatedBucketTransform 的作用是进行最终的 merge aggregating。

FinishAggregatingInOrderTransform 的分组算法如下：

假设有 3 个 stream 当前算子会维护 3 个 Chunk，每一次选取在当前的 3 个 Chunk 内找到最后一条数据的最小值，比如初始状态最小值是 5，然后将 3 个 Chunk 内所有小于 5 的数据一次性取走，如此反复如果一个 Chunk 被取光，需要从改 stream 内拉取新的 Chunk。

这种算法保证了每次 FinishAggregatingInOrderTransform 向下游输出的 Chunk 的最大值小于下一次 Chunk 的最小值，便于后续步骤的优化。

3.优化分析

打开 optimize_aggregation_in_order 后：主要对于第一阶段的聚合步骤进行了优化，从基于 HashMap 的算法到基于连续相同值的算法。

1.对于计算方面：优化后的算法，减少了 Hash 计算，但多了判断相等的步骤，在不同数据基数集大小下，各有优劣。

2.由于内存方面：优化前后无差别

3.对于 poipeline 阻塞：优化前后无差别

四、优化小结

在整个查询计划中 Sort、Distinct、聚合这 3 个算子算子往往是整个查询的瓶颈算子，所以值得对其进行深度优化。ClickHouse 通过利用算子输入数据的有序性，优化算子的算法或者选择不同的算法，在计算、内存和 pipeline 阻塞三个方面均有不同程度的优化。