ClickhouseClickHouse 内部架构介绍

Posted 2022-03-18 九师兄

1.概述

转载：ClickHouse 内部架构介绍

官方原文链接：https://clickhouse.yandex/docs/en/development/architecture/

ClickHouse是一个完全面向列式的分布式数据库。数据通过列存储，在查询过程中，数据通过数组来处理(向量或者列Chunk)。当进行查询时，操作被转发到数组上，而不是在特定的值上。因此被称为”向量化查询执行”，相对于实际的数据处理成本，向量化处理具有更低的转发成本。

这个设计思路并不是新的思路理念。历史可以追溯到APL编程语言时代：A+, J, K, and Q。数组编程广泛用于科学数据处理领域。而在关系型数据库中：也应用了向量化系统。

在加速查询处理上，有两种的方法：向量化查询执行和运行时代码生成。为每种查询类型都进行代码生成，去除所有的间接和动态转发处理。这些方法并不比其他方法好，当多个操作一起执行时，运行时代码生成会更好，可以充分累用CPU执行单元和Pipeline管道。

向量化查询执行实用性并不那么高，因为它涉及到临时向量，必须写到缓存中，并读取回来。如果临时数据并不适合L2缓存，它可能是一个问题。但是向量化查询执行更容易利用CPU的SIMD能力。一个研究论文显示将两个方法结合到一起效果会更好。ClickHouse主要使用向量化查询执行和有限的运行时代码生成支持(仅GROUP BY内部循环第一阶段被编译)。

列

为了表示内存中的列(列的 chunks)，IColumn将被使用。这个接口提供了一些辅助方法来实现不同的关系操作符。几乎所有的操作符都是非更改的：他们不能更改原有的列，但是创建一个新的更新的列。例如，IColumn::filter方法接受一个过滤器字节掩码，同时创建一个新的过滤列。它被用在WHERE和HAVING的关系操作符上。额外的示例：IColumn::permute方法支持ORDER BY，IColumn::cut方法支持LIMIT等。

不同的IColumn实现(ColumnUInt8,ColumnString等)负责列的内存布局。内存布局通常是一个连续的数组。对于列的整型来说，它是一个连续的数组，如std::vector。对于String和Array列,这个是2个vectors:一个是所有的数组元素,连续放置,另一个是偏移量(offsets),位于每个数组的起始端。也有ColumnConst用于在内存中存储一个值,但是它看起来像一个列。

数据域

然而, 它也可能工作在单独的值上面。为了表示一个单独的值。数据域使用.Fieldis 这是一个UInt64,Int64,Float64,StringandArray可区分的集合。IColumn 有operator[]方法来获得n-th值作为一个数据域，insert[] 方法追加一个数据域到一个列的末尾。这些方法不是特别高效，因为他们需要处理临时的数据域对象，它代表一个单独的值。这是一个最高效的方法,例如insertFrom, insertRangeFrom等。

对于一个表，一个特定的数据类型，数据域没有足够的信息。例如 ,UInt8,UInt16,UInt32, 和 UInt64都用 UInt64表示。

抽象渗漏法则

IColumn有方法用于通用的关系型数据转换，但是它并不能满足所有需求。例如，ColumnUInt64没有方法来计算2个列的加和，ColumnString没有方法用于运行子字符串的搜索。一些进程是在IColumn之外实现的。

列中的不同函数能够以一个通用的方式来实现，使用IColumn方法来抽取数据域值，或者在特定的方法下使用数据的内部内存布局在特定的IColumn上实现。为了完成这个，函数将被转换成一个特定的IColumn类型，直接在内部进行处理。例如，ColumnUInt64有一个getData方法，将返回一个内存数组的引用，然后一个单独的进程读取或者直接填充这个数组。事实上，我们有一个抽象渗漏法则来允许不同进程的专用化。

数据类型

IDataType 负责序列化和反序列化: 读写这个列的值或者以二进制或文本的方式的值.IDataType 直接与表中的数据类型一致。例如，有DataTypeUInt32,DataTypeDateTime,DataTypeString等。

IDataType和IColumnare 互相是松耦合的。不同的数据类型能够在内存中表示，通过相同的IColumn 实现.。例如,DataTypeUInt32和DataTypeDateTime都是通过ColumnUInt32或者ColumnConstUInt32来表示。另外，相同的数据类型通过不同的IColumn实现来表示. 例如,DataTypeUInt8 能够通过ColumnUInt8或者ColumnConstUInt8.来表示。

IDataType 仅存储元数据。例如，DataTypeUInt8 根本不保存任何数据 (除了 vptr) ，同时DataTypeFixedString 保存justN(确定的字符串大小)。

IDataType 对于不同的数据格式都有协助方法。示例是有些方法可以序列化一个值, 序列化一个值到 JSON，序列化一个值到 XML 格式。没有直接的数据格式一一对应。例如，不同的数据格式Pretty和TabSeparated 能够使用相同的serializeTextEscaped协助方法，在IDataType接口中。

数据块

一个数据块是一个容器，代表了内存中一个表的子集。它也是三元组的集合:(IColumn,IDataType,columnname). 在查询执行过程中, 数据通过数据块来处理. 如果你有一个数据块， 我们有数据(在IColumn对象中), 我们有这个数据的类型(在IDataType中) 告诉我们怎样处理此列，同时我们有此列名称 (或者是原有列名, 或者是人工命名，得到计算的临时结果)。

在一个数据块中，当我们计算跨列某个函数时, 我们添加另外的带有结果的列到数据块中, 我们并不修改这个列，因为这些操作都是非变更的。然后，不需要的列将从数据块中删除，但不是修改。这个对于消除子表达式是便捷的。

数据块为了每个处理的数据 Chunk 创建的。 对于相同的计算类型，列名称和类型对于不同的数据块将保持一致, 只有列数据保持变化。这样有利于更好地从数据块头拆分数据，因为小的数据块大小将有高的临时字符串开销，当拷贝 shared_ptrs 和 column names时。

数据块流

数据块流用于处理数据。我们使用数据块的数据流从某处读取数据，执行数据转换或者写入数据到某处。IBlockInputStream 有一个read方法获取下一个数据块。IBlockOutputStream 有一个write方法发送数据块到某处。

数据流负责：

读写一个表。当读写数据块时，此表将返回一个数据流。

实现数据格式。例如，如果你想要输出数据以Pretty的格式到一个终端时。你将创建一个数据块输出流，然后格式化这个数据块。

执行数据转换。你有BlockInputStream 同时 想要创建一个过滤数据流。你创建FilterBlockInputStream，初始化它。然后当你从 FilterBlockInputStream拉取一个数据块时，它将从数据流中获得到一个数据块，，过滤它，然后返回已经过滤的数据块给你。查询执行的 Pipeline 将展示这个方式。

有一些更加综合的转换。例如，当你从AggregatingBlockInputStream拉取数据时，它将从数据源上读取所有的数据，聚合它，然后为你返回一个汇总数据流。另一个示例：UnionBlockInputStream接收很多输入数据源和一些线程。它启动了多个线程，从多个数据源中并行读取数据。

数据块流使用“pull” 的方式来控制数据流：当你从第一个数据流中拉取一个数据块时，它从嵌套的数据流中拉取所需要的数据块，整个执行 pipeline 将正常工作。其实“pull” 和 “push”都不是最佳方案，因为流控是隐式的，限制了不同特性的实现，如多个查询的并行执行(一起合并多个 pipeline)。此限制是协程或者运行互相等待的外部线程。我们也能够更多的可能性，如果我们进行显式的流控：如果我们定位这个逻辑，从一个计算单元传递数据到外部的一个计算单元。更多的想法，参考此文章。

查询执行流水线将在每个步骤创建临时数据。我们将保持数据块大小要足够小，因此临时数据要适合CPU缓存。假设，读写临时数据几乎是自由的，相对于其他计算来说。我们可以考虑一个替代方案，融合多个操作在 pipeline 中，让 pipeline 尽可能小，删除尽可能多的临时数据。这个可以是一个优势，也可能是个劣势。例如，一个拆分 pipeline 将容易实现缓存中间数据，从类似的查询中偷取中间数据，然后对于类似查询，合并 pipeline。

格式

数据格式用数据块流来实现。有种“显示性”格式仅适用于数据输出到客户端，例如 Pretty 格式, 它仅提供 IBlockOutputStream。有输入输出格式，例如 TabSeparated 或JSONEachRow 。

也有行数据流: IRowInputStream和 IRowOutputStream. 他们允许你按照行来推/拉数据, 而不是通过数据块. 他们仅被用于简化面向行格式的实现。封装器BlockInputStreamFromRowInputStream 和BlockOutputStreamFromRowOutputStream 允许你转换面向行的数据流到面向数据块的数据流。

I/O

对于面向字节的输入/输出。有 ReadBuffer 和 WriteBuffer 抽象类. 他们被用于替代C++ iostream。 不用担心：每个成熟的 C++ 工程都用更优的类库.

ReadBuffer 和 WriteBuffer 是一个连续的Buffer，游标指向Buffer的位置. 具体实现可能有或没有内存。有个虚方法来用如下数据填充Buffer填充。 (对于ReadBuffer) 或者刷新Buffer到某处 (对于 WriteBuffer). 虚方法很少被调用。

Implementations of ReadBuffer/WriteBuffer 的实现被用于文件，文件描述和网络套接字的处理，如实现压缩 (CompressedWriteBuffer 用另外的 WriteBuffer 来初始化，在写数据之前执行压缩），或者用于其他目的 – 名称ConcatReadBuffer, LimitReadBuffer, 和HashingWriteBuffer 等。

Read/WriteBuffers 仅用于处理字节，带有格式化的输入/输出 (例如, 以decimal的方式写入一个数字), 有一些函数是来自ReadHelpers 和WriteHelpers 头文件的。

让我们看一下当你想以Json的格式写入结果集到标准输出时发生了什么。你有一个结果集准备从IBlockInputStream获取。你创建了 WriteBufferFromFileDescriptor(STDOUT_FILENO) 写入字节到标准输出. 你创建JSONRowOutputStream, 用 WriteBuffer来初始化, 写入行到标准输出。你在行输出流之上创建 数据块输出流BlockOutputStreamFromRowOutputStream, 用IBlockOutputStream显示它. 然后调用 copyData从IBlockInputStream 到 IBlockOutputStream来传输数据. 从内部来看, JSONRowOutputStream 将写入不同的 JSON分隔符，调用 IDataType::serializeTextJSON 方法 引用到IColumn ，同时行数作为参数。然后，IDataType::serializeTextJSON将从 WriteHelpers.h调用一个方法：例如, 对于数字类型用writeText， 对于字符串类型用writeJSONString。

表

表通过IStorage接口来表示.对此接口不同的实现成为不同的表引擎. 例如 StorageMergeTree, StorageMemory, 等，这些类的实例是表。

最重要的IStorage 方法是读和写操作. 也有alter, rename, drop, 等操作. 读方法接受如下的参数：从表中读取的列集合, AST 查询, 返回需要的数据流的数量. 它返回一个或多个 IBlockInputStream 对象和有关数据处理阶段的信息，在查询的过程中在表引擎中完成。

在大多数情况下，read方法负责从表中读取特定的列，不进行后续的7数据处理。所有的进一步数据处理通过查询中断器来完成，这个在IStorage处理范围之外。

但是也有一些例外： - AST 查询被传递到read方法，表引擎使用它来衍生对索引的使用， 同时从一个表中读取少量数据. - 有时表引擎能够处理数据到一个特定的阶段。例如, StorageDistributed 能够发送一个查询到远程服务器，让他们处理数据到一个阶段，即来自不同远程服务器的数据能够被合并，同时返回预处理后数据 查询中断器随即结束对数据的处理。

表的read方法能够返回多个IBlockInputStream 对象允许并行处理数据. 这些多个数据块输入流能够从一个表中并行读取数据. 然后你能够用不同的转换来封装这些数据流(例如表达式评估，数据过滤) 能够被单独计算，同时在它们之上创建一个UnionBlockInputStream, 从多个数据流中并行读取。

也有一些TableFunction. 有一些函数返回临时的``IStorage 对象，用在查询的 FROM 语句中.

为了快速建立一个印象，怎样实现你自己的表引擎，如StorageMemory或StorageTinyLog。

作为read方法的结果, IStorage 返回 QueryProcessingStage – 此信息将返回哪个查询部分已经在Storage中被计算. 当前，我们仅有非常粗粒度的信息。对于存储来说，没有方法说“我已经处理了Where条件中的表达式部分，对于此数据范围” 。我们需要工作在其上。

解析器

一个查询通过手写的递归解析器被解析。例如， ParserSelectQuery递归调用如下的解析，对于不同的查询部分。解析器创建了一个AST. 这个AST通过节点来表示，它是一个IAST实例。

由于历史原因，解析器生成并没有被使用。

中断器

中断器负责从一个AST上创建查询执行Pipeline。有一些简单的中断器，例如 InterpreterExistsQuery和InterpreterDropQuery, 或者更复杂一些的 InterpreterSelectQuery. 此查询执行pipeline是数据块输入个输出流的结合体。例如，中断SELECT 查询的结果是IBlockInputStream 读取结果集； INSERT 查询的结果是 IBlockOutputStream 为了插入而写入数据；and the result of interpreting the中断 INSERT SELECT 查询的结果是在第一次读取时，返回一个空结果集, 但是同时从SELECT到INSERT拷贝数据。

InterpreterSelectQuery 使用了ExpressionAnalyzer和ExpressionActions 机制来查询分析和转换。 这是一个基于规则的查询优。ExpressionAnalyzer 是有点乱的，应该被重写: 不同的查询转换和优化应该被提取到不同的类，来允许模块化的转化和查询。

函数

有一些普通函数和聚合函数。 对于聚合函数，请查看下一个章节。

普通函数并不能改变行的数量 – 他们单独处理每个行。事实上，对于每个行，函数不能被调用，但是对于数据块的数据可实现向量化查询执行。

有一些 混合函数, 例如blockSize, rowNumberInBlock, 和runningAccumulate, 拓展了数据块处理，违反了行的独立性。

ClickHouse 有强类型，因此隐式类型转换不能执行。如果函数不支持一个特定的类型绑定，异常将会抛出。但是函数能够工作在很多不同的类型关联。例如， plus 函数 (实现了 + 操作符) 能够工作在任意的数字类型关联：UInt8 + Float32, UInt16 + Int8, 等。一些变种函数能够接收任意数量的参数，如concat 函数。

聚合函数

聚合函数是状态函数。 他们积累传递的值到某个状态, 允许你从这个状态获得结果。他们用IAggregateFunction来管理。状态可以很简单 (对于 AggregateFunctionCount 的状态是一个单UInt64值) 或者相当复杂 (AggregateFunctionUniqCombined 的状态是与线性数组相关, 一个哈希表， 一个 HyperLogLog 概率性数据结构)。

为了处理多个状态，当执行一个高基数 GROUP BY 查询, 状态被分配在Arena中(一个内存池), 或者他们能够以任意合适的内存分片被分配. 状态可以有一个非细碎的构造器和析构器：例如， 复杂的聚合状态能够自己分配额外的内存，这块需要注意，对于创建和销毁状态，同时传递他们的所属关系，追踪是谁和什么时候将销毁这个状态。

聚合状态能够序列化和反序列化来跨网络传递，在执行分布式查询期间，或者如果没有足够的内存情况下，将他们写入到磁盘. 他们甚至能够存储到表内，DataTypeAggregateFunction 允许增量聚合数据。

对于聚合函数状态，序列化的数据格式目前不是版本化的。如果聚合状态仅是临时存储，那是没问题的。但是对于增量聚合，我们有AggregatingMergeTreetable 引擎，同时很多用户已经在生产环境中使用他们了。这就是为什么我们应该增加向后兼容的支持，未来当为任意的聚合函数更改序列化格式时。

服务器

服务器实现了不同的接口：

1. 对于任意的外部客户端暴露一个HTTP接口

2. 对于本地客户端暴露一个TCP 接口，在分布式查询执行时，用于跨服务器通信

3. 一个接口用于传输同步数据

从内部来讲，这是一个基本的多线程服务器，没有携程, fibers, 等。服务器并没有为高频率短查询来设计，而是为了处理低频率的复杂查询， 这两种方式处理的数据量是不同的。

对于查询执行，服务器初始化上下文类，包括数据库列表，用户，访问权限，设置，集群，处理列表，查询日志，等。这个上下文环境被中断器使用。

对于服务器的TCP协议，我们维护了向前兼容和向后兼容：老客户端能访问新服务器，新客户端能访问老服务器。但是我们不想一直维护它们, 未来一年我们将停止对老版本的支持。

对于外部应用，我们推荐使用 HTTP 接口，因为它比较简单易用。TCP 协议与内部数据结构有很多关联耦合：它使用一个内部结构来传递数据块，使用自定义的帧来用于压缩。 对于此协议我们没有发布一个 C 的库，因为它需要连接大部分 ClickHouse 的代码库, 这么做不实际。

分布式查询执行

在一个集群设置中的服务器大部分是独立的。你能够在一个或所有的服务器上创建一个分布式表。此分布式表本身不存储数据—在集群的多个节点上，仅提供一个"视图"到所有的本地表。当你从分布式表进行查询时，它重写这个查询，根据负载均衡的设置，选择远程节点，发送查询给他们。

分布式表请求远程服务器来处理一个查询到一个阶段，此阶段从不同的服务器中继结果后进行合并。然后接收结果后合并这些结果。分布式表尝试分布尽可能多的工作到远程服务器，不能跨网络发送太多的中继数据。

当你进行 IN 或 JOIN 子查询时，情况变得更加复杂一些，每个子查询都使用一个分布式表。我们有不同的策略来执行这些查询。

对于分布式查询执行，没有一个全局的查询规划。每个节点有自己的本地查询规划作为任务的一部分。我们仅有一个简化的一步分布式查询执行：我们为远程节点发送查询，然后合并结果集。但是对于高基数的GROUP BY高难度查询是并不可行的，或者大量临时数据的 JOIN 查询。ClickHouse并不支持这种查询方式，我们需要进一步开发它。
合并树

合并树(MergeTree)是存储引擎的族，通过主键来支持索引. 主键可以是列或表达式的任意 tuple。在MergeTree表中的数据被存储在 “parts” 中. 每一部分按照主键顺序存储数据 (数据通过主键 tuple 来排序). 所有的表的列都在各自的column.bin文件中保存。 此文件由压缩的数据块组成。每个数据块大小从64 KB 到 1 MB，依赖于平均值的大小。数据块由列值组成，按顺序连续放置。对于每一列，列值在同一个顺序上 (顺序通过主键来定义), 因此，对于对应的列，当你通过多列迭代以后来获得值。

主键自身是"稀疏的"。它不定位到每个行 ，但是仅是一些数据范围。 对于每个N-th行， 一个单独的primary.idx 文件有主键的值, N 被称为 index_granularity(通常情况下, N = 8192). 对于每个列, 我们有column.mrk 文件 ，带有 “marks”标签，对于数据文件中的每个N-th行，它是一个偏移量 。每个标签都成成对儿出现的：文件中的偏移量到压缩数据块的起始端，解压缩数据块的偏移量到数据的起始端。 通常情况下，压缩的数据块通过"marks"标签来对齐，解压缩的数据块的偏移量是0。对于primary.idx的数据通常主流在存储中，对于column.mrk文件的数据放在缓存中。

当我们从MergeTree引擎中读取数据时，我们看到了 primary.idx 数据和定位了可能包含请求数据的范围， 然后进一步看column.mrk 数据，和计算偏移量从哪开始读取这些范围。因为稀疏性, 超额的数据可能被读取。 ClickHouse 并不适合高负载的点状查询，因为带有索引粒度行的整个范围必须被读取, 整个压缩数据块必须被解压缩。我们构建的结构是索引稀疏的，因为我们必须在单台服务器上维护数万亿条数据， 对于索引来说没有显著的内存消耗。因为主键是稀疏的，它并不是唯一的：在 INSERT时，它不能够检查键的存在。在一个表内，相同的键你可以有多个行。

当你插入大量数据进入MergeTree时，数据通过主键顺序来筛选，形成一个新的部分。为了保持数据块数是低位的，有一些背景线程周期性地查询这些数据块，将他们合并到一个排序好的数据块。

这就是为什么称为MergeTree。当然，合并意味着"写入净化"。所有的部分都是非修改的：他们仅创建和删除，但是不会更新。当SELECT运行时，它将获得一个表的快照。在合并之后，我们也保持旧的部分用于故障数据恢复，所以如果我们某些合并部分的文件损坏了，我们能够根据原来的部分进行替换。

MergeTree 不是一个LSM 树，因为它不包含 “memtable” 和 “log”: 插入的数据直接写入到文件系统。这个仅适合于批量的INSERT操作，并不是每行写入，同时不能过于频繁 – 每秒一次写入是 OK 的，每秒几千次写入是不可以的。 我们使用这种方式是为了简化，因为在生产环境中，我们主要以批量插入数据为主。

MergeTree表只有一个(主)索引：没有二级索引。它允许在一个逻辑表下的多个物理表示，例如，在多个物理表中存储数据，甚至允许沿着原有的数据带有预计算的表示。

有MergeTree引擎作为背景线程来做额外的合并。示例是CollapsingMergeTree和AggregatingMergeTree。他们作为对更新的特定支持来看待。这些并不是真的更新，在背景合并运行时，因为用户没法控制时间，在MergeTreetable中的数据经常被存储到多个部分，以非完全的合并形式。