ClickHouse架构概述

ClickHouse架构概述

ClickHouse是一个真正面向列的DBMS。数据按列存储，并在执行数组（向量或列块）期间存储。只要有可能，就会在数组上调度操作，而不是在单个值上调度操作。这称为“矢量化查询执行”，它有助于降低实际数据处理的成本。

这个想法并不新鲜。它可以追溯到APL编程语言及其后代：A +，J，K和Q.数组编程用于科学数据处理。这个想法在关系数据库中也不是新的东西：例如，它在Vectorwise系统中使用。

加速查询处理有两种不同的方法：矢量化查询执行和运行时代码生成。在后者中，代码是为动态的每种查询生成的，删除所有间接和动态分派。这两种方法旗鼓相当。运行时代码生成可以更好地将多个操作融合在一起，从而充分利用CPU执行单元和管道。矢量化查询执行可能不太实际，因为它涉及必须写入缓存并读回的临时向量。如果临时数据不适合L2缓存，则会出现问题。但矢量化查询执行更容易利用CPU的SIMD功能。我们的朋友写的一篇研究报告表明，将这两种方法结合起来会更好。 ClickHouse使用矢量化查询执行，并且对运行时代码生成的初始支持有限。
Columns (列)
要表示内存中的列（实际上是列的块），请使用IColumn接口。该接口提供了用于实现各种关系运算符的辅助方法。几乎所有操作都是不可变的：它们不会修改原始列，而是创建一个新的修改后的列。例如，IColumn :: filter方法接受过滤字节掩码。它用于WHERE和HAVING关系运算符。其他示例：支持ORDER BY的IColumn :: permute方法，支持LIMIT的IColumn :: cut方法，等等。 各种IColumn实现（ColumnUInt8，ColumnString等）负责列的内存布局。内存布局通常是一个连续的数组。对于整数类型的列，它只是一个连续的数组，如std :: vector。对于String和Array列，它是两个向量：一个用于所有数组元素，连续放置，另一个用于偏移到每个数组的开头。还有ColumnConst只在内存中存储一个值，但看起来像一个列。

Field (字段)
尽管如此，也可以使用单个值。为了表示单个值，使用Field。 Field只是UInt64，Int64，Float64，String和Array的可辨别联合。 IColumn使用operator []方法将第n个值作为Field获取，使用insert方法将Field附加到列的末尾。这些方法效率不高，因为它们需要处理表示单个值的临时Field对象。有更有效的方法，例如insertFrom，insertRangeFrom等。

对于表的特定数据类型，字段没有足够的相关信息。例如，UInt8，UInt16，UInt32和UInt64在Field中都表示为UInt64。

Leaky Abstractions (抽象漏洞)
IColumn具有用于数据的常见关系转换的方法，但它们不满足所有需求。例如，ColumnUInt64没有计算两列总和的方法，而ColumnString没有运行子字符串搜索的方法。这些无数的常规方法在IColumn之外实现。 列上的各种函数可以使用IColumn方法以通用的，非有效的方式实现，以提取字段值。或者利用了解内部存储器布局的知识以特定的方式通过特殊的IColumn实现。为此，函数将转换为特定的IColumn类型，并直接在内部处理。例如，ColumnUInt64具有getData方法，该方法返回对内部数组的引用，然后单独的程序直接读取或填充该数组。事实上，我们有“抽象漏洞”，以允许各种程序的有效的扩展。

Data Types(数据类型)
IDataType负责序列化和反序列化：用于读取和写入二进制或文本形式的列或单个值的块。 IDataType直接对应于表中的数据类型。例如，有DataTypeUInt32，DataTypeDateTime，DataTypeString等。 IDataType和IColumn只是松散地相互关联。不同的数据类型可以通过相同的IColumn实现在内存中表示。例如，DataTypeUInt32和DataTypeDateTime都由ColumnUInt32或ColumnConstUInt32表示。此外，相同的数据类型可以由不同的IColumn实现表示。例如，DataTypeUInt8可以由ColumnUInt8或ColumnConstUInt8表示。 IDataType仅存储元数据。例如，DataTypeUInt8根本不存储任何内容（vptr除外），DataTypeFixedString只存储N（固定大小字符串的大小）。 IDataType具有各种数据格式的辅助方法。示例使用的是可能引用的序列化值，为JSON序列化值以及将值序列化为XML格式的一部分的方法。没有与数据格式的直接对应关系。例如，Pretty和TabSeparated的不同数据格式可以使用IDataType接口中的相同serializeTextEscaped辅助方法。

Block (块)
块是一个容器，表示内存中表的子集（块）。它只是一组三元组：（IColumn，IDataType，列名）。在查询执行期间，数据由Blocks处理。如果我们有一个Block，我们有数据（在IColumn对象中），我们有关于它的类型（在IDataType中）的信息告诉我们如何处理该列，并且我们有列名（要么是表中的原始列名，或者是用于获取临时计算结果的一些人工指定的临时名称）。 当我们在块中的列上计算某些函数时，我们将在块中新增一列用于存放其结果，并且我们不会触及函数参数的列，因为操作是不可变的。之后，可以从块中删除不需要的列，但不会修改列。这对于消除常见的子表达式很方便。 每个处理过的数据块都会被创建Blocks。
请注意，对于相同类型的计算，对于不同的数据块列名称和数据类型仍然保持相同，只有列的数据被更改。最好从块的头部分割块数据，因为小的块大小(small block sizes)会占高开销的临时字符串空间，用于复制shared_ptrs和列名..

Block Streams(流式数据块)
流式数据块用于处理数据。我们使用块流来从某个地方读取数据，执行数据转换或将数据写入某个地方。 IBlockInputStream具有read方法以在可用时获取下一个块。 IBlockOutputStream具有将块推送到某处的write方法。

Streams负责：
1. 读或写表。该表只返回一个用于读取或写入块的流。
2. 实现数据格式化。例如，如果要以漂亮格式将数据输出到终端，则可以创建块输出流，然后按块进行格式化。
3. 执行数据转换。假设你有IBlockInputStream并想要创建一个过滤流。您创建FilterBlockInputStream并使用您的流初始化它。然后，当您从FilterBlockInputStream中拉出一个块时，它会从您的流中提取一个块，对其进行过滤，然后将过滤后的块返回给您。查询执行管道以这种方式表示。

还有更复杂的转换。例如，当您从AggregatingBlockInputStream中提取时，它会从其源读取所有数据，聚合它，然后为您返回聚合数据流。另一个例子：UnionBlockInputStream接受构造函数中的许多输入源以及许多线程。它启动多个线程并从多个源并行读取。
块流使用“拉”方法来控制流：当您从第一个流中提取块时，它会从嵌套流中提取所需的块，整个执行管道将起作用。 “pull”和“push”都不是最佳解决方案，因为控制流是隐含的，并且限制了各种功能的实现，例如同时执行多个查询（将许多管道合并在一起）。可以通过协同程序或仅运行等待彼此的额外线程来克服此限制。如果我们明确控制流，我们可能有更多的可能性：如果我们找到将数据从一个计算单元传递到那些计算单元之外的另一个计算单元的逻辑。阅读本文以获取更多想法。

我们应该注意，查询执行管道在每一步创建临时数据。我们尝试将块大小保持足够小，以便临时数据适合CPU缓存。有了这个假设，与其他计算相比，写入和读取临时数据几乎是免费的。我们可以考虑一种替代方案，即将流水线中的许多操作融合在一起，使流水线尽可能短，并删除大部分临时数据。这可能是一个优点，但它也有缺点。例如，拆分管道可以轻松实现缓存中间数据，窃取同时运行的类似查询的中间数据，以及合并类似查询的管道。

Formats (格式化)
数据格式是通过流式数据块实现的。这些“表象”的格式仅适用于向客户端输出数据，例如Pretty格式，其仅提供IBlockOutputStream。还有输入/输出格式，例如TabSeparated或JSONEachRow。 还有基于行的流：IRowInputStream和IRowOutputStream。它们允许您按单个行而不是块来提取/推送数据。它们只需要简化面向行的格式的实现。包装器BlockInputStreamFromRowInputStream和BlockOutputStreamFromRowOutputStream允许您将面向行的流转换为常规的面向块的流。


I /O (输入/输出)
对于面向字节的输入/输出，有ReadBuffer和WriteBuffer抽象类。它们被用来代替C ++ iostream。不要担心：每个成熟的C ++项目都使用除了iostream之外的其他东西。 ReadBuffer和WriteBuffer只是一个连续的缓冲区和一个指向该缓冲区中位置的游标。实现可能拥有或不拥有缓冲区的内存。有一个虚拟方法用以下数据填充缓冲区（对于ReadBuffer）或在某处刷新缓冲区（对于WriteBuffer）。很少调用虚方法。 ReadBuffer / WriteBuffer的实现用于处理文件和文件描述符和网络套接字，用于实现压缩（CompressedWriteBuffer用另一个WriteBuffer初始化并在向其写入数据之前执行压缩），以及用于其他目的 , ConcatReadBuffer，LimitReadBuffer和HashingWriteBuffer这些方法名称已说明一切。 Read / WriteBuffers只处理字节。为了帮助格式化输入/输出（例如，以十进制格式写入数字），ReadHelpers和WriteHelpers头文件中都有函数。 让我们看一下当你想把JSON格式的结果集写入stdout时会发生什么。您已准备好从IBlockInputStream获取结果集。您创建WriteBufferFromFileDescriptor（STDOUT_FILENO）以将字节写入stdout。您创建使用该WriteBuffer初始化的JSONRowOutputStream，以将JSON中的行写入stdout。您可以在其上创建BlockOutputStreamFromRowOutputStream，以将其表示为IBlockOutputStream。然后调用copyData将数据从IBlockInputStream传输到IBlockOutputStream，一切正常。在内部，JSONRowOutputStream将编写各种JSON分隔符并调用IDataType :: serializeTextJSON方法，并引用IColumn和行号作为参数。因此，IDataType :: serializeTextJSON将从WriteHelpers.h调用一个方法：例如，writeText用于数字类型，writeJSONString用于DataTypeString。

Tables (表)
表由IStorage接口表示。不同的表引擎由不同的表接口实现。例如StorageMergeTree，StorageMemory等。这些类的实例只是表格。
最重要的IStorage方法是read和write。同样重要的还有alter，rename，drop等等。 read方法接受以下参数：要从表中读取的列的集合，要考虑的AST查询以及要返回的所需数量的流。它返回一个或多个IBlockInputStream对象以及有关在查询执行期间在表引擎内完成的数据处理阶段的信息。 在大多数情况下，read方法仅负责从表中读取指定的列，不会用于任何进一步的数据处理。所有进一步的数据处理都不在IStorage的职责范围内，转而由查询解释器完成。
但是也有一些值得注意的例外情况：
 AST查询被传递给read方法，表引擎可以使用它来获取索引使用情况从而能从表中读取更少的数据。
 有时，表引擎自身可以将数据本身处理到特定阶段。例如，StorageDistributed可以向远程服务器发送查询，要求它们将数据处理到可以合并来自不同远程服务器的数据的阶段，并返回该预处理数据。然后，查询解释器完成数据处理。

表的read方法可以返回多个IBlockInputStream对象以允许并行数据处理。这些多个块输入流可以并行读取表。然后，您可以独立计算的各种转换（例如表达式求值或过滤）来包装这些流，并在它们之上创建UnionBlockInputStream，以并行读取多个流。 还有TableFunctions。这些函数返回临时IStorage对象以在查询的FROM子句中使用。 要快速了解如何实现自己的表引擎，请查看简单的内容，如StorageMemory或StorageTinyLog。

作为read方法的结果，IStorage返回QueryProcessingStage - 有关查询的哪些部分已在存储内计算的信息。目前，我们对该信息只有非常粗略的粒度。对于这个数据范围，存储无法说“我已经在WHERE中处理了表达式的这一部分”。我们需要努力。

Parsers (解析器)
一条查询是由一个手写的递归下降解析器解析的。例如，ParserSelectQuery只是递归地调用查询的各个部分的底层解析器。AST由解析器创建。 AST由节点表示，节点是IAST的实例。

由于历史原因，不使用分析器生成器。


Interpreters (解释器)
解释器负责从AST创建查询执行管道。有简单的解释器，如InterpreterExistsQuery和InterpreterDropQuery，也有更复杂的InterpreterSelectQuery。查询执行管道是块输入或输出流的组合。例如，解释SELECT查询的结果是IBlockInputStream从中读取结果集; INSERT查询的结果是IBlockOutputStream来写入要插入的数据;解释INSERT SELECT查询的结果是IBlockInputStream，它在第一次读取时返回一个空结果集，但同时将数据从SELECT复制到INSERT。 InterpreterSelectQuery使用ExpressionAnalyzer和ExpressionActions机制进行查询分析和转换。这是大多数基于规则的查询优化的地方。 ExpressionAnalyzer非常混乱，应该重写：应该提取各种查询转换和优化来分离类，以允许模块化转换或查询。

Functions (函数)
Clickhouse提供普通的函数和聚合函数。有关聚合函数，请参阅下一节。 普通函数不会改变行数 - 它们的工作方式就像它们独立处理每一行一样。实际上，不会为单个行调用函数，而是以向量化查询执行的方式使用Block数据。 有一些杂项函数，如blockSize，rowNumberInBlock和runningAccumulate，它们利用块处理并违反行的独立性。 ClickHouse具有强类型，因此不会发生隐式类型转换。如果函数不支持特定的类型组合，则会抛出异常。但是对于许多不同类型的组合，函数可以工作（被重载）。例如，plus函数（用于实现+运算符）适用于任何数字类型组合：UInt8 + Float32，UInt16 + Int8等。此外，一些可变函数可以接受任意数量的参数，例如concat函数。
实现函数可能稍微不方便，因为函数显式调度支持的数据类型和支持的IColumns。例如，plus函数拥有实例化C ++模板提供的代码生成特性,以组合各种数值类型。以及用于常量或非常量补齐参数的代码。

这是实现运行时代码生成以避免模板代码膨胀的好地方。此外，它可以添加融合函数，如融合乘法 - 加法，或在一个循环迭代中进行多重比较。

由于矢量化查询执行，函数不是短路的。例如，如果写入WHERE f（x）AND g（y），则当f（x）为零时也会计算两边（除非f（x）是零常量表达式）, 甚至整行。但是如果f（x）条件的选择性很高，并且f（x）的计算比g（y）便宜得多，那么最好的实现方法是使用多层过滤计算(multi-pass calculation)：首先计算f（x），接着过滤结果，最后计算g（y）的时候只仅针对较小的，经过过滤后的数据块。

Aggregate Functions (聚合函数)
聚合函数是有状态函数。它们将传递的值累积到某个状态，并允许您从该状态获得结果。它们使用IAggregateFunction接口进行管理。状态可以相当简单（AggregateFunctionCount的状态只是单个UInt64值）或相当复杂（AggregateFunctionUniqCombined的状态是线性数组，散列表和HyperLogLog概率数据结构的组合）。

为了在执行高基数GROUP BY查询时处理多个状态，在Arena（内存池）中分配状态，或者可以在任何合适的内存中分配状态。状态可以有一个非平凡(non-trivial)的构造函数和析构函数：例如，复杂的聚合状态可以自己分配额外的内存。这需要注意创造和销毁状态并正确地传递其所有权，以便跟踪谁以及何时将销毁状态。

聚合状态可以被序列化和反序列化，以在分布式查询执行期间通过网络传递，或者将它们写入没有足够RAM的磁盘上。它们甚至可以存储在具有DataTypeAggregateFunction的表中，以允许数据的增量聚合。

聚合函数状态的序列化数据格式目前尚未版本化。如果仅是暂时存储聚合状态，尚可接受。但我们提供了AggregatingMergeTree表引擎用于增量聚合，人们已经在生产中使用它。这就是为什么我们应该在将来更改任何聚合函数的序列化格式时添加对向后兼容性的支持。

Server (服务)
服务实现了几个不同的接口：
 用于任外部客户端的HTTP接口。
 用于原生ClickHouse客户端，以及分布式查询执行期间的跨服务通信的TCP接口，。
 用于传输数据以进行复制的接口。

内在的讲，它只是一个没有协程，光纤等硬性要求的基本的多线程服务。由于服务不是为处理高速率的简单查询设计的，而是为了处理相对较低速率的复杂查询而设计的，因此每个服务都可以处理大量的数据用于分析。

服务使用必要的查询执行环境初始化Context类：可用数据库列表，用户和访问权限，设置，集群，进程列表，查询日志等。此执行环节会被解释器使用。

我们维护服务TCP协议的完全向后和向前兼容性：旧客户端可以与新服务通信，新客户端可以与旧服务通信。但是我们不想永久维护它，并且我们在大约一年后就删除了对旧版本的支持。

对于所有外部应用程序，我们建议使用HTTP接口，因为它简单易用。 TCP协议与内部数据结构的联系更紧密：它使用内部格式传递数据块，并使用自定义帧来压缩数据。我们还没有发布该协议的C库，因为它需要链接大部分ClickHouse代码库，这是不切实际的。

Distributed Query Execution (分布式查询执行)
群集设置中的服务器大多是独立的。您可以在群集中的一个或所有服务器上创建分布式表。 Distributed表本身不存储数据 - 它只为集群的多个节点上的所有本地表提供“视图”。从分布式表中进行SELECT时，它会重写该查询，根据负载平衡设置选择远程节点，并将查询发送给它们。 Distributed表请求远程服务器处理查询，直到可以合并来自不同服务器的中间结果的阶段。然后它接收中间结果并合并它们。分布式表尝试将尽可能多的工作分配给远程服务器，并且不会通过网络发送太多中间数据。

当您在IN或JOIN子句中有子查询并且每个子查询都使用分布式表时，事情会变得更加复杂。我们有不同的策略来执行这些查询。

分布式查询执行没有全局查询计划。每个节点都有自己的部分作业本地查询计划。我们只有简单的一次性分布式查询执行：我们发送远程节点的查询，然后合并结果。但对于具有高基数GROUP BY或具有大量JOIN临时数据的困难查询，这是不可行的：在这种情况下，我们需要在服务器之间“重新洗牌”(reshuffle)数据，这需要额外的协调。 ClickHouse不支持这种查询执行，我们需要在将来支持处理它。

Merge Tree (合并树)
MergeTree是一系列存储引擎，支持按主键索引。主键可以是列或表达式的任意元组。 MergeTree表中的数据存储在“分片”(parts)中。每个部分以主键顺序存储数据（数据按主键元组按字典顺序排序）。所有表的列信息都存储在这些部分中的单独column.bin文件中。文件由压缩块组成。每个块通常是64 KB到1 MB的未压缩数据，具体取决于平均值大小。这些块由一个接一个地连续放置的列值组成。每列的列值的顺序相同（顺序由主键定义），因此当您按多列进行迭代时，您将获得相应行的值。

主键本身是“稀疏的”。它不是针对每一行，而是针对某些范围的数据。单独的primary.idx文件具有每个第N行的主键值，其中N称为index_granularity（通常，N = 8192）。此外，对于每一列，我们都有带有“标记”的column.mrk文件，这些文件是数据文件中每个第N行的偏移量。每个标记都是一对：文件中的偏移量到压缩块的开头，以及解压缩块中的偏移量到数据的开头。通常，压缩块通过标记对齐，并且解压缩块中的偏移量为零。 primary.idx的数据始终驻留在内存中，并缓存column.mrk文件的数据。

当我们要从MergeTree中的某个部分读取内容时，我们会查看primary.idx数据并查找可能包含请求数据的范围，然后查看column.mrk数据并计算从哪里开始读取这些范围的偏移量。由于稀疏性，可能会读取过多的数据。 ClickHouse不适用于高负载的简单点查询，因为必须为每个键读取具有index_granularity行的整个范围，并且必须为每列解压缩整个压缩块。我们使索引稀疏，因为我们必须能够为每个服务器维护数万亿行，而索引没有明显的内存消耗。此外，由于主键是稀疏的，因此它不是唯一的：它无法在INSERT时检查表中是否存在键。您可以在表中使用相同的键创建许多行。
当您将一堆数据插入MergeTree时，该组按主键顺序排序并形成一个新part。为了保持part数量相对较低，有background线程定期选择一些分片(parts)并将它们合并到一个排序分片。这就是它被称为MergeTree的原因。当然，合并会导致“写入放大”。所有部分都是不可变的：它们仅被创建和删除，但不会被修改。运行SELECT时，它保存表的快照（一组分片）。在合并之后，我们还会将旧分片保留一段时间以便在故障后更容易恢复，因此如果我们看到某些合并分片可能已损坏，我们可以将其替换为其源分片。

MergeTree不是LSM树，因为它不包含“memtable”和“log”：插入的数据直接写入文件系统。这使得它仅适用于批量插入数据，而不是频繁的通过单独的行- 大约每秒一次是可行的，但是每秒一千次是不可取的。我们这样做是为了简单，也因为我们已经在我们的应用程序中应用了这种批量插入模式。

MergeTree表只能有一个（主）索引：没有任何辅助索引。这将适用于在一个逻辑表下允许多个物理表示的场景，例如，以多个物理顺序存储数据，甚至允许具有从原始数据中预聚合数据。

有些MergeTree引擎在后台合并期间会进行额外的工作。示例是CollapsingMergeTree和AggregatingMergeTree。这可以被视为对更新的特殊支持。请记住，这些不是真正的更新，因为用户通常无法控制执行后台合并的时间，并且MergeTree表中的数据几乎总是存储在多个部分中，而不是完全合并的形式。

Replication (副本/复制)
ClickHouse中的复制是基于每个表实现的。您可以在同一台服务器上拥有一些复制的表和一些非复制的表。您还可以以不同方式复制表，例如一个表具有双因子复制，另一个表具有三因子复制。

复制在ReplicatedMergeTree存储引擎中实现。 ZooKeeper中的路径被指定为存储引擎的参数。 ZooKeeper中具有相同路径的所有表都成为彼此的副本：它们同步数据并保持一致性。只需创建或删除表，即可动态添加和删除副本。

复制使用异步多主机方案。您可以将数据插入到与ZooKeeper进行会话的任何副本中，并将数据异步复制到所有其他副本。由于ClickHouse不支持UPDATE，因此复制是无冲突的。由于没有对插入的仲裁确认，如果一个节点发生故障，刚刚插入的数据可能会丢失。

用于复制的元数据存储在ZooKeeper中。有一个复制日志列出了要执行的操作。步骤是：get part; merge parts; drop partition等等。每个副本将复制日志复制到其队列，然后执行队列中的操作。例如，在插入时，会在日志中创建“get part”操作，并且每个副本都会下载该part。在副本之间协调合并以获得字节相同的结果。所有parts在所有副本上以相同的方式合并。为实现此目的，一个副本被选为领导者，该副本启动merges，并将“merge parts”操作写入日志。

复制是物理的：只有压缩的部分在节点之间传输，而不是查询。为了降低网络成本（以避免网络放大），在大多数情况下，在每个副本上独立地处理合并。只有在存在显着的复制延迟的情况下，才会通过网络发送Large merged parts。

此外，每个副本在ZooKeeper中将其状态存储为一组parts及其checksums。当本地文件系统上的状态与ZooKeeper中的引用状态不同时，副本通过从其他副本下载缺失和损坏的parts来恢复其一致性。当本地文件系统中存在一些意外或损坏的数据时，ClickHouse不会将其删除，而是将其移动到单独的目录并忘记它。

ClickHouse集群由独立的分片(shards)组成，每个分片由副本组成。群集不是弹性的，因此在添加新分片后，数据不会自动在分片之间重新平衡。相反，群集负载将不均匀。此实现为您提供了更多控制，对于相对较小的集群（例如数十个节点）来说它很好。但对于我们在生产中使用的具有数百个节点的集群，这种方法成为一个重大缺陷。我们应该实现一个表引擎，该表引擎将跨集群跨越其数据，并具有动态复制的区域，这些区域可以在集群之间自动拆分和平衡。