【ClickHouse源码】各类存储引擎的实现原理

落日映苍穹つ 2022-12-22 09:29 237阅读 0赞

## 注册各类存储引擎 ##

在`ClickHouse`的`Server`端启动时会注册很多内容，包括存储引擎、函数、表函数等等，源码[Server.cpp][]，如下：

int Server::main(const std::vector<std::string> & /*args*/)
    {
        ......
        registerFunctions();
        registerAggregateFunctions();
        registerTableFunctions();
        registerStorages();
        registerDictionaries();
        ......
    }

其中，`registerStorages()`就是进行存储引擎的注册，源码[registerStorages.cpp][]，如下：

void registerStorages()
    {
        auto & factory = StorageFactory::instance();
    
        registerStorageLog(factory);
        registerStorageTinyLog(factory);
        registerStorageStripeLog(factory);
        registerStorageMergeTree(factory);
        registerStorageNull(factory);
        registerStorageMerge(factory);
        registerStorageBuffer(factory);
        registerStorageDistributed(factory);
        registerStorageMemory(factory);
        registerStorageFile(factory);
        registerStorageURL(factory);
        registerStorageS3(factory);
        registerStorageDictionary(factory);
        registerStorageSet(factory);
        registerStorageJoin(factory);
        registerStorageView(factory);
        registerStorageMaterializedView(factory);
        registerStorageLiveView(factory);
    
        #if USE_HDFS
        registerStorageHDFS(factory);
        #endif
    
        #if USE_POCO_SQLODBC || USE_POCO_DATAODBC
        registerStorageODBC(factory);
        #endif
        registerStorageJDBC(factory);
    
    
        #if USE_MYSQL
        registerStorageMySQL(factory);
        #endif
    
        #if USE_RDKAFKA
        registerStorageKafka(factory);
        #endif
    }

可以看出方法上半部分会注册`Log`、`MergeTree`、`S3`等存储引擎，下半部分会根据宏定义决定是否会注册，各个宏值为：

#define USE_ICU 0
    #define USE_MYSQL 1
    #define USE_RDKAFKA 1
    #define USE_EMBEDDED_COMPILER 0
    #define USE_POCO_SQLODBC 0
    #define USE_POCO_DATAODBC 1
    #define USE_POCO_MONGODB 1
    #define USE_POCO_REDIS 1
    #define USE_INTERNAL_LLVM_LIBRARY 0
    #define USE_SSL 1

所以MySQL和Kafka存储引擎默认是会注册的。

取其中一个`registerStorageMergeTree(factory)`方法细看一下，源码[registerStorageMergeTree.cpp][]，如下：

void registerStorageMergeTree(StorageFactory & factory)
    {
        factory.registerStorage("MergeTree", create);
        factory.registerStorage("CollapsingMergeTree", create);
        factory.registerStorage("ReplacingMergeTree", create);
        factory.registerStorage("AggregatingMergeTree", create);
        factory.registerStorage("SummingMergeTree", create);
        factory.registerStorage("GraphiteMergeTree", create);
        factory.registerStorage("VersionedCollapsingMergeTree", create);
    
        factory.registerStorage("ReplicatedMergeTree", create);
        factory.registerStorage("ReplicatedCollapsingMergeTree", create);
        factory.registerStorage("ReplicatedReplacingMergeTree", create);
        factory.registerStorage("ReplicatedAggregatingMergeTree", create);
        factory.registerStorage("ReplicatedSummingMergeTree", create);
        factory.registerStorage("ReplicatedGraphiteMergeTree", create);
        factory.registerStorage("ReplicatedVersionedCollapsingMergeTree", create);
    }

该方法负责注册各种`MergeTree`和`ReplicatedMergeTree`。既然所有引擎都注册到了工厂方法中，那在创建存储引擎时就会通过该`factory`来创建，实现了标准化封装。

## 创建各类存储引擎 ##

创建存储引擎是在`StorageFactory`类中统一实现的，`get`方法在`StorageFactory.h`，如下：

StoragePtr get(
            ASTCreateQuery & query,
            const String & data_path,
            const String & table_name,
            const String & database_name,
            Context & local_context,
            Context & context,
            const ColumnsDescription & columns,
            const ConstraintsDescription & constraints,
            bool attach,
            bool has_force_restore_data_flag) const;

`get`方法通过`ASTCreateQuery`来决定创建什么存储引擎，并返回`Storage`指针后续使用。

这里的`StoragePtr`就是`IStorage`的指针，所以所有的存储引擎都是继承`IStorage`来实现不同的功能

## 各类存储引擎的实现 ##

首先看一下实现了该基类的各个存储引擎类：

> `StorageBuffer`、`StorageDictionary`、**`StorageDistributed`**、`StorageDistributedFake`、`StorageFile`、`StorageFromMergeTreeDataPart`、`StorageHDFS`、`StorageInput`、`StorageJoin`、`StorageKafka`、`StorageLiveView`、`StorageLog`、`StorageMaterializedView`、`StorageMemory`、`StorageMerge`、`StorageMergeTree`、`StorageMySQL`、`StorageNull`、**`StorageReplicatedMergeTree`**、**`StorageS3`**、`StorageSet`、`StorageSetOrJoinBase`、`StorageStripeLog`、`StorageSystemColumns`、`StorageSystemDetachedParts`、`StorageSystemDisks`、`StorageSystemNumbers`、`StorageSystemOne`、`StorageSystemParts`、`StorageSystemPartsBase`、`StorageSystemPartsColumns`、`StorageSystemReplicas`、`StorageSystemStoragePolicies`、`StorageSystemTables`、`StorageTinyLog`、`StorageURL`、`StorageValues`、`StorageView`、`StorageXDBC`

可以看到标粗的（`StorageReplicatedMergeTree`、`StorageDistributed`、`StorageS3`）就是经常会涉及到的存储引擎，这些引擎都是实现`IStorage`中的虚函数来实现不同引擎的存取的（即多态）。

### 基类`IStorage`的实现 ###

先看下`IStorage`中到底有都有什么方法，源码[IStorage.h][]，如下：

class IStorage : public std::enable_shared_from_this<IStorage>, public TypePromotion<IStorage>
    {
    public:
        ......
        virtual std::string getName() const = 0;
        virtual std::string getTableName() const = 0;
        virtual std::string getDatabaseName() const { return {}; }
        ......
    
    public: 
        ......
        Block getSampleBlock() const; 
        Block getSampleBlockWithVirtuals() const; 
        Block getSampleBlockNonMaterialized() const; 
       
        void check(const Names & column_names, bool include_virtuals = false) const;
        void check(const NamesAndTypesList & columns) const;
        void check(const NamesAndTypesList & columns, const Names & column_names) const;
        void check(const Block & block, bool need_all = false) const;
        ......
    
    public:
        virtual BlockInputStreams watch(
            const Names & /*column_names*/,
            const SelectQueryInfo & /*query_info*/,
            const Context & /*context*/,
            QueryProcessingStage::Enum & /*processed_stage*/,
            size_t /*max_block_size*/,
            unsigned /*num_streams*/)
        {
            throw Exception("Method watch is not supported by storage " + getName(), ErrorCodes::NOT_IMPLEMENTED);
        }
    
        virtual BlockInputStreams read(
            const Names & /*column_names*/,
            const SelectQueryInfo & /*query_info*/,
            const Context & /*context*/,
            QueryProcessingStage::Enum /*processed_stage*/,
            size_t /*max_block_size*/,
            unsigned /*num_streams*/);
    
        virtual BlockOutputStreamPtr write(
            const ASTPtr & /*query*/,
            const Context & /*context*/)
        {
            throw Exception("Method write is not supported by storage " + getName(), ErrorCodes::NOT_IMPLEMENTED);
        }
    
        virtual void startup() {}
        virtual void shutdown() {}
    
    private:
        mutable RWLock alter_intention_lock = RWLockImpl::create();
        mutable RWLock new_data_structure_lock = RWLockImpl::create();
        mutable RWLock structure_lock = RWLockImpl::create();
    }

这个类的虚函数比较多，这里保留了写比较核心的方法，下面做下简要说明：

*  第一个`public`中主要是包含回去该存储引擎本身信息的一些方法；
 *  第二个`public`中主要是获取`Block`和验证`Block`的方法；
 *  第三个`public`中主要是`read`、`write`、`watch`方法及`startup`和`shutdown`方法（用于在引擎创建启动和删除后的一些额外操作，比如启动/删除`ReplicatedMergeTree`的监听`zk`线程、自动`merge`线程等，类似统一hook）；
 *  最后一个private中是流程中涉及到的锁；

### `StorageReplicatedMergeTree`的实现 ###

该类就是继承了`IStorage`的部分方法实现的，类在[StorageReplicatedMergeTree.h][]，这里不再详细列举，主要是实现一些核心方法及`StorageReplicatedMergeTree`相关`task`（`queue_updating_task`、`mutations_updating_task`、`queue_task_handle`、`move_parts_task_handle`、`merge_selecting_task`、`mutations_finalizing_task`）的初始化。

下面看下`ReplicatedMergeTree`如何写入到磁盘:

BlockOutputStreamPtr StorageReplicatedMergeTree::write()
                                     ↓ 
               void ReplicatedMergeTreeBlockOutputStream::write()
                                     ↓
                        storage.writer.writeTempPart()
                                     ↓
                MergedBlockOutputStrea out.writeWithPermutation()
                                     ↓
         std::pair<size_t, size_t> IMergedBlockOutputStream::writeColumn()
                                     ↓  
             size_t IMergedBlockOutputStream::writeSingleGranule()
                                     ↓
       void DataTypeLowCardinality::serializeBinaryBulkWithMultipleStreams()
                                     ↓
                     DataTypeString::serializeBinaryBulk()
                                     ↓
                           WriteBuffer ostr.write()

主要流程：

1.  调用`StorageReplicatedMergeTree::write()`方法时会创建`OutputStream`，最终由`OutputStream`的`write`方法实现写逻辑；
2.  写数据时先写临时`part`文件，其中的`writeWithPermutation()`方法是为了节省`RAM`；
3.  ck每个数据文件都是一个列，`writeColumn()`来实现接下来的数据落盘，并且根据`Granule`间隔（默认8192）一块一块操作；
4.  接下来就是序列化过程，对于`String`、`Array`、`Tuple`等都是有不同的实现方式；
5.  最后写入`WriteBuffer`，并按照数据量阈值或时间阈值实现真正的数据落盘；

### `StorageS3`的实现 ###

`S3`的实现就比较简单，只是实现几个核心方法，源码[StorageS3.h][]

写盘流程如下：

BlockOutputStreamPtr StorageS3::write()
                                     ↓ 
                  void StorageS3BlockOutputStream::write()
                                     ↓
                      void WriteBufferFromS3::writePart()

### `StorageHDFS`的实现 ###

与S3的实现类似，也是实现几个核心方法，源码[StorageHDFS.h][]

写盘流程如下：

BlockOutputStreamPtr StorageHDFS::write()
                                 ↓ 
                 void HDFSBlockOutputStream::write()
                                 ↓
                         tSize hdfsWrite()

操作`hdfs`时有专门的实现类库[libhdfs3][]，源码[hdfs.h][]，包含了`hdfsWrite`、`hdfsRead`、`hdfsOpenFile`等。

### 总结 ###

通过分析ck中的核心的`ReplicatedMergeTree`以及扩展的`S3`和`HDFS`的`Storage`实现，可以发现所有的实现都是继承`IStorage`基类的；

`ReplicatedMergeTree`由于是直接与标准文件系统接口交互，会有许多复杂的逻辑实现，如序列化/反序列化、`Buffer`刷盘等，还有`*MergeTree`和`Replicated*`的一些特性，所以实现会相对复杂，而`S3`和`HDFS`有各自独特的一套接口协议，使用了不同的接口实现了`IStorage`中的必要方法；

如果要实现一种表引擎，ck的模式是通过实现`IStorage`基类的方法作为标准化规范，再实现一套对应的数据流读写方法，这一层也同样要通过实现`IBlockInputStream`/`IBlockOutputStream`中的一些列标准方法，最后是有一套新存储系统的`client`标准类库作为支撑；

## `WriteBuffer`的实现 ##

如果想要知道`clickhouse`如何和Linux文件系统交互的，可以继续从`StorageReplicatedMergeTree` 中的`WriteBuffer ostr.write()`继续深入分析。在`clickhouse`中的`buffer`有很多实现，但都是基于`WriteBuffer`实现的。源码[WriteBuffer.h][]

class WriteBuffer : public BufferBase
    {
    public:
        WriteBuffer(Position ptr, size_t size) : BufferBase(ptr, size, 0) {}
        void set(Position ptr, size_t size) { BufferBase::set(ptr, size, 0); }
    
        inline void next()
        {
            if (!offset() && available())
                return;
            bytes += offset();
    
            try
            {
                nextImpl();
            }
            catch (...)
            {
                pos = working_buffer.begin();
                throw;
            }
    
            pos = working_buffer.begin();
        }
    
        virtual ~WriteBuffer() {}
    
        inline void nextIfAtEnd()
        {
            if (!hasPendingData())
                next();
        }
    
    
        void write(const char * from, size_t n)
        {
            size_t bytes_copied = 0;
    
            while (bytes_copied < n)
            {
                nextIfAtEnd();
                size_t bytes_to_copy = std::min(static_cast<size_t>(working_buffer.end() - pos), n - bytes_copied);
                memcpy(pos, from + bytes_copied, bytes_to_copy);
                pos += bytes_to_copy;
                bytes_copied += bytes_to_copy;
            }
        }
    
    
        inline void write(char x)
        {
            nextIfAtEnd();
            *pos = x;
            ++pos;
        }
    
    private:
        virtual void nextImpl() { throw Exception("Cannot write after end of buffer.", ErrorCodes::CANNOT_WRITE_AFTER_END_OF_BUFFER); }
    }

主要的核心方法有`next()`、`write()`、`set()`等。

> 这里先看一下`set()`，是由`BufferBase`类的`set`实现的，那`BufferBase`是做什么用的呢？看一下描述，源码[BufferBase.h][]
> 
>  *  `ReadBuffer`和`WriteBuffer`分别类似于`istream`和`ostream`；
>  *  使用`iostream`不可能有效地实现某些操作，例如使用`istream`，无法从制表符分隔的文件中快速读取字符串值；
>  *  `ReadBuffer`和`WriteBuffer`提供了对内部缓冲区的直接访问；
>  *  `ReadBuffer`和`WriteBuffer`可以拥有或不拥有自己的内存；
> 
> 可知，`BufferBase`是为了实现一种高效的buffer，提供给`clickhouse`更快的读写效率。

着重看一下`next()`，其中主要就是调用了`nextImpl()`，`nextImpl()`由不同的buffer做了不同的实现，以`WriteBufferFromOStream`和`WriteBufferFromFileDescriptor`为例看一下具体实现。源码[WriteBufferFromOStream.cpp][]、[WriteBufferFromFileDescriptor.cpp][]

### `WriteBufferFromOStream::nextImpl()` ###

void WriteBufferFromOStream::nextImpl()
    {
        if (!offset())
            return;
    
        ostr->write(working_buffer.begin(), offset());
        ostr->flush();
    
        if (!ostr->good())
            throw Exception("Cannot write to ostream at offset " + std::to_string(count()),
                ErrorCodes::CANNOT_WRITE_TO_OSTREAM);
    }

`OStream`中的`write()`就类似`std::ostream`的`write()`，会将数据以流的形式写入文件，并执行`flush()`.

### `WriteBufferFromFileDescriptor::nextImpl()` ###

void WriteBufferFromFileDescriptor::nextImpl()
    {
        if (!offset())
            return;
    
        Stopwatch watch;
    
        size_t bytes_written = 0;
        while (bytes_written != offset())
        {
            ProfileEvents::increment(ProfileEvents::WriteBufferFromFileDescriptorWrite);
    
            ssize_t res = 0;
            {
                CurrentMetrics::Increment metric_increment{CurrentMetrics::Write};
                res = ::write(fd, working_buffer.begin() + bytes_written, offset() - bytes_written);
            }
    
            ......
    
            if (res > 0)
                bytes_written += res;
        }
    
        ......
    }

`FileDescriptor`中的`::write()`就是`unistd.h`的`write()`方法，通过`fd`操作数据写入文件。

### 总结 ###

`clickhouse`根据自身需要，通过`BufferBase`构建了特殊的`buffer`，最终通过各个实现类不同的`nextImpl()`方法实现不同的功能的差异化；

`Stream`中的`write()`就类似`std::ostream`的`write()`，这里是隐式调用了`open()`等方法，直接的文件操作就是是直接的带`fd`参数的`write()`

[Server.cpp]: https://github.com/ClickHouse/ClickHouse/blob/master/programs/server/Server.cpp
[registerStorages.cpp]: https://github.com/ClickHouse/ClickHouse/blob/master/src/Storages/registerStorages.cpp
[registerStorageMergeTree.cpp]: https://github.com/ClickHouse/ClickHouse/blob/master/src/Storages/MergeTree/registerStorageMergeTree.cpp
[IStorage.h]: https://github.com/ClickHouse/ClickHouse/blob/master/src/Storages/IStorage.h
[StorageReplicatedMergeTree.h]: https://github.com/ClickHouse/ClickHouse/blob/master/src/Storages/StorageReplicatedMergeTree.h
[StorageS3.h]: https://github.com/ClickHouse/ClickHouse/blob/master/src/Storages/StorageS3.h
[StorageHDFS.h]: https://github.com/ClickHouse/ClickHouse/blob/master/src/Storages/StorageHDFS.h
[libhdfs3]: https://github.com/ClickHouse-Extras/libhdfs3
[hdfs.h]: https://github.com/ClickHouse-Extras/libhdfs3/blob/master/src/client/hdfs.h
[WriteBuffer.h]: https://github.com/ClickHouse/ClickHouse/blob/master/src/IO/WriteBuffer.h
[BufferBase.h]: https://github.com/ClickHouse/ClickHouse/blob/master/src/IO/BufferBase.h
[WriteBufferFromOStream.cpp]: https://github.com/ClickHouse/ClickHouse/blob/master/src/IO/WriteBufferFromOStream.cpp
[WriteBufferFromFileDescriptor.cpp]: https://github.com/ClickHouse/ClickHouse/blob/master/src/IO/WriteBufferFromFileDescriptor.cpp