http://www.clickhouse.com.cn/topic/5a3df7b02141c2917483557c

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。有一些简单的中断器,例如 InterpreterExistsQueryInterpreterDropQuery, 或者更复杂一些的 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 引擎,同时很多用户已经在生产环境中使用他们了。这就是为什么我们应该增加向后兼容的支持,未来当为任意的聚合函数更改序列化格式时。

服务器


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

  • 对于任意的外部客户端暴露一个HTTP接口
  • 对于本地客户端暴露一个TCP 接口,在分布式查询执行时,用于跨服务器通信
  • 一个接口用于传输同步数据

从内部来讲,这是一个基本的多线程服务器,没有携程, 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中的数据经常被存储到多个部分,以非完全的合并形式。