CREATE TABLE mt (
    EventDate Date,
    OrderID Int32,
    BannerID UInt64,
    UserId String,
    GoalNum Int8,

    INDEX UserIdIndex (UserId) TYPE bloom_filter GRANULARITY 4
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(EventDate) 
ORDER BY (OrderID, BannerID)

$ ls /var/lib/clickhouse/data/default/mt
201809_2_2_0  201809_3_3_0  201810_1_1_0  201810_4_4_0 201810_1_4_1
detached  format_version.txt

$ ls /var/lib/clickhouse/data/default/mt/201810_1_4_1
GoalNum.bin GoalNum.mrk BannerID.bin ...
primary.idx checksums.txt count.txt columns.txt
partition.dat minmax_EventDate.idx

~/Documents/master/ClickHouse/src master !17 ?17 ❯ cloc  .                                                                                                                                         00:25:25
    4271 text files.
    4269 unique files.                                          
      25 files ignored.

github.com/AlDanial/cloc v 1.90  T=2.11 s (2019.9 files/s, 368444.2 lines/s)
-------------------------------------------------------------------------------
Language                     files          blank        comment           code
-------------------------------------------------------------------------------
C++                           2250          79043          19631         458480
C/C++ Header                  1946          44246          22895         150382
CMake                           51            304             52           1177
Python                           2             66              3            218
SQL                              4              0              0            159
Protocol Buffers                 1             35             55            107
ANTLR Grammar                    1             15              1            105
Bourne Shell                     4             20             11             63
Markdown                         1              1              0              2
-------------------------------------------------------------------------------
SUM:                          4260         123730          42648         610693
-------------------------------------------------------------------------------

~/Documents/master/ClickHouse/src/Storages/MergeTree master !17 ?17 ❯ cloc  .                                                                                                                      00:28:08
     225 text files.
     225 unique files.                                          
       0 files ignored.

github.com/AlDanial/cloc v 1.90  T=0.14 s (1662.5 files/s, 395988.9 lines/s)
-------------------------------------------------------------------------------
Language                     files          blank        comment           code
-------------------------------------------------------------------------------
C++                            106           6805           2306          31543
C/C++ Header                   117           2789           2005           8139
CMake                            2              0              0              5
-------------------------------------------------------------------------------
SUM:                           225           9594           4311          39687
-------------------------------------------------------------------------------

void MergeTreeSink::consume(Chunk chunk)
{
    auto block = getHeader().cloneWithColumns(chunk.detachColumns());

    // 按 Partition By 子句指定的分区，将 block 拆成每个 Partition 一个 Block, block -> []block

    auto part_blocks = storage.writer.splitBlockIntoParts(block, max_parts_per_block, metadata_snapshot, context);
    for (auto & current_block : part_blocks)
    {
        Stopwatch watch;

        // 每个 Block 创建新的 Part
        // 一个 MergeTree 表包含多个 Part, Part 会在后台定期合并 (Merge)
        // 新建 Part 时会写到 Temp 目录，以防止查询时读到不完整的写入

        MergeTreeData::MutableDataPartPtr part = storage.writer.writeTempPart(current_block, metadata_snapshot, context);

        /// If optimize_on_insert setting is true, current_block could become empty after merge
        /// and we didn't create part.
        if (!part)
            continue;

        // 挂载 Temp Part 到表空间下

        /// Part can be deduplicated, so increment counters and add to part log only if it's really added
        if (storage.renameTempPartAndAdd(part, &storage.increment, nullptr, storage.getDeduplicationLog()))
        {
            // 记录 part metrics 到 system.part_log 表
            PartLog::addNewPart(storage.getContext(), part, watch.elapsed());

            // 触发后台 Merge 任务

            /// Initiate async merge - it will be done if it's good time for merge and if there are space in 'background_pool'.
            storage.background_operations_assignee.trigger();
        }
    }
}

BlocksWithPartition MergeTreeDataWriter::splitBlockIntoParts(
        const Block & block, size_t max_parts, const StorageMetadataPtr & metadata_snapshot, ContextPtr context)
{
    // 省略 ...

    // 未分区时, 快速跳过返回
    if (!metadata_snapshot->hasPartitionKey()) /// Table is not partitioned.
    {
        result.emplace_back(Block(block), Row{});
        return result;
    }

    Block block_copy = block;

    // executePartitionByExpression 对 Block 执行 MergeTree Partition By 子句, 并且将新增的 key 字段添加到 block_copy 中
    // partition_key_names_and_types 仅包含分区相关字段

    /// After expression execution partition key columns will be added to block_copy with names regarding partition function.
    auto partition_key_names_and_types = MergeTreePartition::executePartitionByExpression(metadata_snapshot, block_copy, context);

    ColumnRawPtrs partition_columns;
    partition_columns.reserve(partition_key_names_and_types.size());
    for (const auto & element : partition_key_names_and_types)
        partition_columns.emplace_back(block_copy.getByName(element.name).column.get());

    // buildScatterSelector 建立了 row_id -> partition_id 之间的映射
    // selector 等价于 []int 类型, selector[row_id] = partition_id
    PODArray<size_t> partition_num_to_first_row;
    IColumn::Selector selector;
    buildScatterSelector(partition_columns, partition_num_to_first_row, selector, max_parts);

    // 省略 ...

    // IColumn::scatter 算子可以将一个 Column 一拆多
    // 对每个 Column 一拆多后，组合为多个 Block
    for (size_t col = 0; col < block.columns(); ++col)
    {
        MutableColumns scattered = block.getByPosition(col).column->scatter(partitions_count, selector);
        for (size_t i = 0; i < partitions_count; ++i)
            result[i].block.getByPosition(col).column = std::move(scattered[i]);
    }

    return result;
}

MergeTreeData::MutableDataPartPtr MergeTreeDataWriter::writeTempPart(
    BlockWithPartition & block_with_partition, const StorageMetadataPtr & metadata_snapshot, ContextPtr context)
{
    Block & block = block_with_partition.block;
    static const String TMP_PREFIX = "tmp_insert_";

    // 生成一个 Part ID, 每一个新的 Part 插入都会生成一个自增的 Part ID
    /// This will generate unique name in scope of current server process.
    Int64 temp_index = data.insert_increment.get();

    // 计算 partition 相关字段的 min-max 统计
    auto minmax_idx = std::make_shared<IMergeTreeDataPart::MinMaxIndex>();
    minmax_idx->update(block, data.getMinMaxColumnsNames(metadata_snapshot->getPartitionKey()));

    MergeTreePartition partition(std::move(block_with_partition.partition));

    MergeTreePartInfo new_part_info(partition.getID(metadata_snapshot->getPartitionKey().sample_block), temp_index, temp_index, 0);
    String part_name;
    if (data.format_version < MERGE_TREE_DATA_MIN_FORMAT_VERSION_WITH_CUSTOM_PARTITIONING)
    {
        // 省略, 这里为旧版 MergeTree 格式 ... 
    }
    else
        part_name = new_part_info.getPartName();

    /// If we need to calculate some columns to sort.
    if (metadata_snapshot->hasSortingKey() || metadata_snapshot->hasSecondaryIndices())
        // SORT BY toYYYYMM(time) 这种情况需要计算 Expression 的值
        data.getSortingKeyAndSkipIndicesExpression(metadata_snapshot)->execute(block);

    // 省略 ... 

    ProfileEvents::increment(ProfileEvents::MergeTreeDataWriterBlocks);

    /// Sort
    IColumn::Permutation * perm_ptr = nullptr;
    IColumn::Permutation perm;
    if (!sort_description.empty())
    {
        // isAlreadySorted 仅对 Block 未排序情况下进行排序
        if (!isAlreadySorted(block, sort_description))
        {
            // stableGetPermutation 为稳定排序(尽可能保持插入顺序), 实现为 std::stable_sort
            // 注意这里并未对所有列都进行了排序, 而是建立了原始数据与排序数据 row_id 的映射
            stableGetPermutation(block, sort_description, perm);
            perm_ptr = &perm;
        }
        else
            ProfileEvents::increment(ProfileEvents::MergeTreeDataWriterBlocksAlreadySorted);
    }

    // 省略 ... 

    // updateTTL 更新 TTL 信息, TTL 为 MergeTree 提供的数据生命周期管理功能
    // 记录当前 Part 与 TTL 相关字段的 min,max 统计，用于后续方便快速判断一个 Part 是否满足 TTL 条件
    for (const auto & ttl_entry : move_ttl_entries)
        updateTTL(ttl_entry, move_ttl_infos, move_ttl_infos.moves_ttl[ttl_entry.result_column], block, false);

    // 省略 ... 

    // createPart
    auto new_data_part = data.createPart(
        part_name,
        // 注意 choosePartType 会按照 Block 大小和行数决定是否使用 Compact/Memory 存储方式
        data.choosePartType(expected_size, block.rows()),
        new_part_info,
        createVolumeFromReservation(reservation, volume),
        TMP_PREFIX + part_name);

    // 省略 ... 

    SyncGuardPtr sync_guard;
    if (new_data_part->isStoredOnDisk())
    {
        // 省略 ... 

        // sync_guard 会在析构函数执行时完成目录 Sync 到磁盘
        if (data.getSettings()->fsync_part_directory)
            sync_guard = disk->getDirectorySyncGuard(full_path);
    }

    // 省略, 记录额外的 TTL 信息 ...
    // [TTL expr
    //   [DELETE|TO DISK 'xxx'|TO VOLUME 'xxx' [, ...] ]
    //   [WHERE conditions]
    //   [GROUP BY key_expr [SET v1 = aggr_func(v1) [, v2 = aggr_func(v2) ...]] ] ] 
    new_data_part->ttl_infos.update(move_ttl_infos);

    /// This effectively chooses minimal compression method:
    ///  either default lz4 or compression method with zero thresholds on absolute and relative part size.
    auto compression_codec = data.getContext()->chooseCompressionCodec(0, 0);

    // MergedBlockOutputStream 将 Block 序列化到磁盘
    const auto & index_factory = MergeTreeIndexFactory::instance();
    MergedBlockOutputStream out(new_data_part, metadata_snapshot,columns,
        index_factory.getMany(metadata_snapshot->getSecondaryIndices()), compression_codec);

    bool sync_on_insert = data_settings->fsync_after_insert;

    // 写入数据主体, Columns / Indexs
    out.writeWithPermutation(block, perm_ptr);

    // 处理 PROJECTION 字段, PROJECTION projection_name_1 (SELECT <COLUMN LIST EXPR> [GROUP BY] [ORDER BY])
    for (const auto & projection : metadata_snapshot->getProjections())
    {
        auto projection_block = projection.calculate(block, context);
        if (projection_block.rows())
            new_data_part->addProjectionPart(
                projection.name, writeProjectionPart(data, log, projection_block, projection, new_data_part.get()));
    }

    // 写入收尾, Checksums、ttl info 等信息
    out.writeSuffixAndFinalizePart(new_data_part, sync_on_insert);

    // 省略, 添加写入指标 ...

    return new_data_part;
}

void MergedBlockOutputStream::writeImpl(const Block & block, const IColumn::Permutation * permutation)
{
    // Check 各个列行数相同
    block.checkNumberOfRows();
    size_t rows = block.rows();
    if (!rows)
        return;

    // 调用具体的 MergeTreeDataPartWriter 实现
    // 后文以 MergeTreeDataPartWriterWide.cpp 为例
    writer->write(block, permutation);
    if (reset_columns)
        new_serialization_infos.add(block);

    rows_count += rows;
}

void MergeTreeDataPartWriterWide::write(const Block & block, const IColumn::Permutation * permutation)
{
    /// Fill index granularity for this block
    /// if it's unknown (in case of insert data or horizontal merge,
    /// but not in case of vertical part of vertical merge)
    // 在插入数据是，compute_granularity 默认为 true
    // index_granularity 含义见第二节的介绍
    if (compute_granularity)
    {
        // 计算出此 Part 对应的 index_granularity
        size_t index_granularity_for_block = computeIndexGranularity(block);

        // 省略 ...

        // 根据 index_granularity_for_block，计算每个 granular 对应的 row 范围
        fillIndexGranularity(index_granularity_for_block, block.rows());
    }

    Block block_to_write = block;

    auto granules_to_write = getGranulesToWrite(index_granularity, block_to_write.rows(), getCurrentMark(), rows_written_in_last_mark);

    auto offset_columns = written_offset_columns ? *written_offset_columns : WrittenOffsetColumns{};
    Block primary_key_block;
    if (settings.rewrite_primary_key)
        primary_key_block = getBlockAndPermute(block, metadata_snapshot->getPrimaryKeyColumns(), permutation);

    // 将所有 Column 排序
    Block skip_indexes_block = getBlockAndPermute(block, getSkipIndicesColumns(), permutation);

    auto it = columns_list.begin();
    for (size_t i = 0; i < columns_list.size(); ++i, ++it)
    {
        auto & column = block_to_write.getByName(it->name);

        if (data_part->getSerialization(*it)->getKind() != ISerialization::Kind::SPARSE)
            column.column = recursiveRemoveSparse(column.column);

        if (permutation)
        {
            if (primary_key_block.has(it->name))
            {
                const auto & primary_column = *primary_key_block.getByName(it->name).column;
                writeColumn(*it, primary_column, offset_columns, granules_to_write);
            }
            else if (skip_indexes_block.has(it->name))
            {
                const auto & index_column = *skip_indexes_block.getByName(it->name).column;
                writeColumn(*it, index_column, offset_columns, granules_to_write);
            }
            else
            {
                /// We rearrange the columns that are not included in the primary key here; Then the result is released - to save RAM.
                ColumnPtr permuted_column = column.column->permute(*permutation, 0);
                writeColumn(*it, *permuted_column, offset_columns, granules_to_write);
            }
        }
        else
        {
            // writeColumn 执行最终的写入
            // 后文详细说明写入过程
            writeColumn(*it, *column.column, offset_columns, granules_to_write);
        }
    }

    if (settings.rewrite_primary_key)
        // calculateAndSerializePrimaryIndex 写入主键索引 primary.idx
        calculateAndSerializePrimaryIndex(primary_key_block, granules_to_write);

    // calculateAndSerializeSkipIndices 写入 Column 对应的二级索引
    calculateAndSerializeSkipIndices(skip_indexes_block, granules_to_write);

    shiftCurrentMark(granules_to_write);
}

static size_t computeIndexGranularityImpl(
    const Block & block,
    size_t index_granularity_bytes,
    size_t fixed_index_granularity_rows,
    bool blocks_are_granules,
    bool can_use_adaptive_index_granularity)
{
    size_t rows_in_block = block.rows();
    size_t index_granularity_for_block;
    if (!can_use_adaptive_index_granularity)
        // fixed_index_granularity_rows 来自于 Setting 配置 index_granularity
        index_granularity_for_block = fixed_index_granularity_rows;
    else
    {
        size_t block_size_in_memory = block.bytes();
        if (blocks_are_granules)
            index_granularity_for_block = rows_in_block;
        else if (block_size_in_memory >= index_granularity_bytes)
        {
            // 按照 index_granularity_bytes 计算每个 Granular 需要多少行
            size_t granules_in_block = block_size_in_memory / index_granularity_bytes;
            index_granularity_for_block = rows_in_block / granules_in_block;
        }
        else
        {
            size_t size_of_row_in_bytes = std::max(block_size_in_memory / rows_in_block, 1UL);
            index_granularity_for_block = index_granularity_bytes / size_of_row_in_bytes;
        }
    }
    if (index_granularity_for_block == 0) /// very rare case when index granularity bytes less then single row
        index_granularity_for_block = 1;

    /// We should be less or equal than fixed index granularity
    index_granularity_for_block = std::min(fixed_index_granularity_rows, index_granularity_for_block);
    return index_granularity_for_block;
}

Block getBlockAndPermute(const Block & block, const Names & names, const IColumn::Permutation * permutation)
{
    Block result;
    for (size_t i = 0, size = names.size(); i < size; ++i)
    {
        const auto & name = names[i];
        result.insert(i, block.getByName(name));

        /// Reorder primary key columns in advance and add them to `primary_key_columns`.
        if (permutation)
        {
            auto & column = result.getByPosition(i);
            // 执行对 Column 的排序
            column.column = column.column->permute(*permutation, 0);
        }
    }

    return result;
}

void MergeTreeDataPartWriterWide::writeColumn(
    const NameAndTypePair & name_and_type,
    const IColumn & column,
    WrittenOffsetColumns & offset_columns,
    const Granules & granules)
{
    // 省略 ...

    if (inserted)
    {
        ISerialization::SerializeBinaryBulkSettings serialize_settings;
        serialize_settings.getter = createStreamGetter(name_and_type, offset_columns);
        serialization->serializeBinaryBulkStatePrefix(serialize_settings, it->second);
    }

    // 省略 ...

    for (const auto & granule : granules)
    {
        // 省略 ...

        // 写入单个 Granule 到 xxx.bin
        writeSingleGranule(
           name_and_type,
           column,
           offset_columns,
           it->second,
           serialize_settings,
           granule
        );

        if (granule.is_complete)
        {
            // 省略 ...

            for (const auto & mark : marks_it->second)
                // 写入 Column 对应的 .mrk 索引文件
                flushMarkToFile(mark, index_granularity.getMarkRows(granule.mark_number));
            // 省略 ...
        }

        // 省略 ...
    }

    // 省略 ...
}

void MergeTreeDataPartWriterWide::writeSingleGranule(
    const NameAndTypePair & name_and_type,
    const IColumn & column,
    WrittenOffsetColumns & offset_columns,
    ISerialization::SerializeBinaryBulkStatePtr & serialization_state,
    ISerialization::SerializeBinaryBulkSettings & serialize_settings,
    const Granule & granule)
{
    // 获得字段对应的序列化实例
    const auto & serialization = data_part->getSerialization(name_and_type);
    // 调用 ISerialization::serializeBinaryBulk() 完成字段序列化
    serialization->serializeBinaryBulkWithMultipleStreams(column, granule.start_row, granule.rows_to_write, serialize_settings, serialization_state);

    // 省略 ...
}

void SerializationString::serializeBinaryBulk(const IColumn & column, WriteBuffer & ostr, size_t offset, size_t limit) const
{
    const ColumnString & column_string = typeid_cast<const ColumnString &>(column);
    const ColumnString::Chars & data = column_string.getChars();            // 将所有字符串拼接起来
    const ColumnString::Offsets & offsets = column_string.getOffsets();     // 可用来得到每个 String 在 data 中的 Offset
    // 在实际存储中，offsets 为 Maps i'th position to offset to i+1'th element. Last offset maps to the end of all chars (is the size of all chars).

    // 省略 ...

    if (offset == 0)
    {
        UInt64 str_size = offsets[0] - 1;
        writeVarUInt(str_size, ostr);   // 写入第一个 String 的长度
        ostr.write(reinterpret_cast<const char *>(data.data()), str_size);    // 写入第一个 String 内容

        ++offset;
    }

    for (size_t i = offset; i < end; ++i)
    {
        UInt64 str_size = offsets[i] - offsets[i - 1] - 1; 
        writeVarUInt(str_size, ostr);    // 继续写入后续 String 的长度
        ostr.write(reinterpret_cast<const char *>(&data[offsets[i - 1]]), str_size); // 继续写入后续 String 的内容
    }
}

bool isAlreadySorted(const Block & block, const SortDescription & description)
{
    // 省略 ...

    /** If the rows are not too few, then let's make a quick attempt to verify that the block is not sorted.
     * Constants - at random.
     */

    // 抽样 10 个数(等距离分散采样)，快速检查是否排好序
    // 如果已经排好序，只需要 O(N) 的代价即可完成排序
    // 如果尚未排好序，这十个数的检查代价也可以接受
    // 这是一个非常好的面向常见特殊场景优化的例子
    static constexpr size_t num_rows_to_try = 10;
    if (rows > num_rows_to_try * 5)
    {
        for (size_t i = 1; i < num_rows_to_try; ++i)
        {
            size_t prev_position = rows * (i - 1) / num_rows_to_try;
            size_t curr_position = rows * i / num_rows_to_try;

            if (less(curr_position, prev_position))
                return false;
        }
    }

    for (size_t i = 1; i < rows; ++i)
        if (less(i, i - 1))
            return false;

    return true;
}

LocalDirectorySyncGuard::~LocalDirectorySyncGuard()
{
    try
    {
#if defined(OS_DARWIN)
        if (fcntl(fd, F_FULLFSYNC, 0))
            throwFromErrno("Cannot fcntl(F_FULLFSYNC)", ErrorCodes::CANNOT_FSYNC);
#else
        if (-1 == ::fdatasync(fd))
            throw Exception("Cannot fdatasync", ErrorCodes::CANNOT_FSYNC);
#endif
        if (-1 == ::close(fd))
            throw Exception("Cannot close file", ErrorCodes::CANNOT_CLOSE_FILE);
    }
    catch (...)
    {
        tryLogCurrentException(__PRETTY_FUNCTION__);
    }
}