1. 简介
ClickHouse MergeTree表引擎主要用于海量数据分析，支持数据分区、存储有序、主键索引、稀疏索引、数据TTL等。
1.1. 关于分层合并
MergeTree存储的数据按照增序排序，由多个part组成。part通过启发式算法在后端做merge，不同分区的part不会合并。合并示意图如下：
merge
由此可以发现，MergeTree和LSM结构类似，只不过没有memtable和log，写操作直接操作硬盘，所以适合大批量插入数据，而不是单行的频繁插入。
ClickHouse提供了不同的合并方式，分别对应不同类型的MergeTree：
- MergeTree
- CollapsingMergeTree
- ReplacingMergeTree
- AggregatingMergeTree
- SummingMergeTree
- GraphiteMergeTree
- VersionedCollapsingMergeTree
1.2. 关于数据更改
另外，MergeTree对UPDATE和DELETE命令没有直接支持，是通过ALTER的变种——突变（mutation）——来间接支持：
ALTER TABLE [db.]table DELETE WHERE filter_expr;
ALTER TABLE [db.]table UPDATE column1 = expr1 [, …] WHERE filter_expr;
更新功能不支持更新有关主键或分区键的列
MergeTree通过重写整个数据部分来执行mutation，该mutation没有原子性，mutation执行开始后，select的执行可以看到已经修改的数据和没有修改的数据。
1.3. 关于主键
MergeTree的主键与通常理解的主键有差异，这里的主键并不提供唯一性保证，主要用于加速查询，并且主键是稀疏索引（默认8192行数据生成一个索引，类似于RocksDB SST 文件的block索引方式）
1.4. 关于part
一个part对应一个目录，是组成MergeTree的基本单位，每一个insert操作就会生成一个part，后台再不断的对part进行merge，每个part的数据按照主键排序。
1.5. 关于skip index
MergeTree目前推出的一个实验特性，支持对指定列建立skip index。有skip index的列会进行稀疏的预计算，当查询条件中有使用该列进行过滤的时候，就可以跳过不相关的列，大幅提升查询效率。
2. 实现分析
2.1. Part数据存储目录
目录名格式如下：
partiiton-id _ min-id _ max-id _ level
举例说明：
CREATE TABLE visits (`VisitDate` Date, `Hour` UInt8, `ClientID` UInt64, INDEX index_id ClientID TYPE minmax GRANULARITY 3) ENGINE = MergeTree() PARTITION BY toYYYYMM(VisitDate) ORDER BY Hour;
建表如上述语句，然后插入数据若干，如下查看分区情况：
insert into visits values ('2019-01-01',2,22);
insert into visits values ('2019-01-01',10,22);
insert into visits values ('2019-01-01',30,22);
select partition, name, active from system.parts where table = 'visits';
┌─partition─┬─name─────────┬─active─┐
│ 201901 │ 201901_1_1_0 │ 1 │
│ 201901 │ 201901_2_2_0 │ 1 │
│ 201901 │ 201901_3_3_0 │ 1 │
└───────────┴──────────────┴────────┘
partition列可见有1个分区201901，下面有三个可用part，分别对应三行数据。
一次写入会生成一个或多个（和分区数对应）part目录，一个part只能属于一个分区，且只有一个分区的part才可以进行合并。
假如merge操作发起后，分区会变成如下：
┌─partition─┬─name─────────┬─active─┐
│ 201901 │ 201901_1_1_0 │ 0 │
│ 201901 │ 201901_1_3_1 │ 1 │
│ 201901 │ 201901_2_2_0 │ 0 │
│ 201901 │ 201901_3_3_0 │ 0 │
└───────────┴──────────────┴────────┘
201901分区老的三个part都成为无效状态，新的有效part包含了失效part的block_id范围。
2.2. part存储数据文件
在上述目录下，文件如下：
- checksums.txt：所有文件的列表，包括文件的大小和checksum
- columns.txt：包含所有列及其类型的列表
- primary.idx：包含主键
- [Column].bin：包含压缩后的列数据
- [Column].mrk：标记，允许跳过n*k行进行位置定位
- count.txt：记录本part的行数
- partition.txt：记录分区表达式的值
- minmax_[Column].idx：根据记录分区表达式生成的MinMax 索引（见MergeTreeDataPart::MinMaxIndex class）
- skip_idx_[index name].idx：skip index的值
- skip_idx_[index name].mrk2：skip index的索引标记
还是以上述例子为例，其中一个part目录的文件如下：
-rw-r----- 1 clickhouse clickhouse 479 2月 16 14:08 checksums.txt
-rw-r----- 1 clickhouse clickhouse 39 2月 16 14:08 ClientID.bin
-rw-r----- 1 clickhouse clickhouse 48 2月 16 14:08 ClientID.mrk2
-rw-r----- 1 clickhouse clickhouse 85 2月 16 14:08 columns.txt
-rw-r----- 1 clickhouse clickhouse 1 2月 16 14:08 count.txt
-rw-r----- 1 clickhouse clickhouse 29 2月 16 14:08 Hour.bin
-rw-r----- 1 clickhouse clickhouse 48 2月 16 14:08 Hour.mrk2
-rw-r----- 1 clickhouse clickhouse 4 2月 16 14:08 minmax_VisitDate.idx
-rw-r----- 1 clickhouse clickhouse 4 2月 16 14:08 partition.dat
-rw-r----- 1 clickhouse clickhouse 2 2月 16 14:08 primary.idx
-rw-r----- 1 clickhouse clickhouse 39 2月 16 14:08 skp_idx_index_id.idx
-rw-r----- 1 clickhouse clickhouse 24 2月 16 14:08 skp_idx_index_id.mrk2
-rw-r----- 1 clickhouse clickhouse 32 2月 16 14:08 VisitDate.bin
-rw-r----- 1 clickhouse clickhouse 48 2月 16 14:08 VisitDate.mrk2
2.3. 代码走读
类图如下：
MergeTree类图
在单个节点上，一个table对应一个StorageMergeTree实例，MergeTree的类型通过StorageMergeTree.merging_params确定。
每一个MergeTreeDataPart类，内部有一个MergeTreePartition对象用于存储分区信息，不同MergeTreeDataPart实例上的MergeTreePartition实例可能是一样的，也就是说，逻辑上的partition包含多个part，一个part只能属于一个partition。
2.3.1. insert操作
代码执行路径如下：
- excuteQuery() // called in TCPHandler.cpp
- executeQueryImpl()
- parseQuery()
- interpreter->execute() // InterpreterInsertQuery
- StorageMergeTree.write // 构造一个MergeTreeBlockOutputStream
- copyData(in, out) // out就是是MergeTreeBlockOutputStream
- MergeTreeBlockOutputStream.write()
- MergeTreeDataWriter.splitBlockIntoParts
- MergeTreeDataWriter.writeTempPart
- MergedBlockOutputStream.writePrefix // MergedBlockOutputStream用于写数据到磁盘
- MergedBlockOutputStream.writeWithPermutation
- MergedBlockOutputStream.calculateAndSerializeSkipIndices // 写skip indies
- MergedBlockOutputStream.writeSuffixAndFinalizePart // 落盘
- StorageMergeTree. renameTempPartAndAdd // min_id 和 max_id一样
- MergeTreeBlockOutputStream.write()
- executeQueryImpl()
- processInsertQuery()
- NativeBlockOutputStream.write() // 发送结果到client
2.3.2. select
代码执行路径如下：
- excuteQuery() // called in TCPHandler.cpp
- executeQueryImpl()
- parseQuery()
- interpreter->execute() // InterpreterSelectQuery
- InterpreterSelectQuery.executeImpl()
- executeFetchColumns()
- StorageMergeTree.readWithProcessors
- MergeTreeDataSelectExecutor.read
- MergeTreeDataSelectExecutor.readFromParts // 通过skip索引过滤构建输入流
- StorageMergeTree.readWithProcessors
- executeFetchColumns()
- InterpreterSelectQuery.executeImpl()
- executeQueryImpl()
- processOrdinaryQuery() // Pull query execution result, if exists, and send it to network.
2.3.3 mutation
代码执行路径如下：
mutate()
- 实例化MergeTreeMutationEntry对象，创建tmp_mutation_{xxx}.txt，写入command、format version、create time到文件
- tmp_mutation_{xxx}.txt重命名为mutation_{version}.txt
- MergeTreeMutationEntry对象放入current_mutations_by_id和current_mutations_by_version
- 唤醒后台线程 ，执行mutation函数 mergeMutateTask
- clearOldPartsFromFilesystem
- 找出outdated状态的part
- 从文件系统删除
- 从内存中删除（ data_parts_indexes）
- 写part log
- clearOldTemporaryDirectories
- clearOldMutations
- merge()
- merge失败则执行tryMutatePart()，否则直接返回
- 找出outdated状态的part
- clearOldPartsFromFilesystem
- 等待线程执行结束
2.3.4 merge
merge()
- MergeTreeDataMergeMutator.selectPartsToMerge // 选出最适合merge的分区的所有parts
- FutureMergedMutatedPart.assign // 将所有parts的信息进行合并，初始化将要生成的新part对象
- block_id合并
- level加1
- MergeTreeDataMergeMutator.mergePartsToTemparyPart // 落盘到临时目录
- MergeTreeDataMergeMutator.renameMergedTemporaryPart // mv到正式目录
附件
@startuml
scale 1.75
together {
class MergingParams {
// For Collapsing and VersionedCollapsing mode.
String sign_column;
// For Summing mode.
Names columns_to_sum;
// For Replacing and VersionedCollapsing mode.
String version_column;
// For Graphite mode.
Graphite::Params graphite_params;
}
class MergeTreePartition {
Row value;
void load(/*storage*/,
/*part_path*/);
void store();
}
class MergeTreeDataPart {
MinMaxIndex minmax_idx;
MergeTreeIndexGranularity
index_granularity;
Columns index;
MergeTreePartition partition;
}
class MergeTreeData {
Names minmax_idx_columns;
Names primary_key_columns;
Names sorting_key_columns;
ExpressionActionsPtr partition_key_expr;
ExpressionActionsPtr minmax_idx_expr;
// Secondary indices for MergeTree skipping
MergeTreeIndices skip_indices;
// multi-index container of parts
DataPartsIndexes data_parts_indexes;
MergingParams merging_params;
void alterDataParts();
void loadDataParts();
}
}
interface IStorage {
ColumnsDescription columns;
IndicesDescription indices;
ConstraintsDescription constraints;
BlockOutputStreamPtr write(/*querey*/, /*context*/);
void mutate(const MutationCommands &, const Context &);
void alter(const AlterCommands &, const Context &, /*lock holder*/);
BlockInputStreams read(const Names &, const SelectQueryInfo &, ...);
Pipes readWithProcessors (const Names &, const SelectQueryInfo &, ...);
}
note left: table is a IStorage
class StorageMergeTree {
MergeTreeDataSelectExecutor reader;
MergeTreeDataWriter writer;
MergeTreeDataMergerMutator merger_mutator;
BackgroundProcessingPool::TaskHandle
merging_mutating_task_handle;
void mutate(const MutationCommands &, const Context &);
void alter(const AlterCommands &, const Context &, /*lock holder*/);
BlockOutputStreamPtr write(/*querey*/, /*context*/);
Pipes readWithProcessors (const Names &, const SelectQueryInfo &, ...);
}
IStorage <|.. MergeTreeData
MergeTreeData <|-- StorageMergeTree
MergeTreePartition o- MergeTreeDataPart : > aggregated by some
MergeTreeDataPart -* MergeTreeData : < have some
MergeTreeData *- MergingParams : have 1 >
@enduml
