Method write

src/Storages/MergeTree/MergeTreeDataPartWriterWide.cpp:310–406

}

void MergeTreeDataPartWriterWide::write(const Block & block, const IColumnPermutation * permutation, Block * permuted_columns_cache)
{
    Block block_to_write = block;

    /// For some columns the set of streams may depend on the actual column data.
    /// For example: dynamic structure and statistics for JSON, Dynamic and Map (with adaptive number of buckets).
    /// We must ensure that all blocks will be written in the same set of streams, so we have to make some
    /// preparations to achieve it.
    prepareBlockForWriting(block_to_write);

    initStreamsIfNeeded();
    initColumnsSubstreamsIfNeeded();

    /// 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)
    if (compute_granularity)
    {
        size_t index_granularity_for_block = 0;
        if (auto constant_granularity = index_granularity->getConstantGranularity())
            index_granularity_for_block = *constant_granularity;
        else
            index_granularity_for_block = computeIndexGranularity(block_to_write);

        if (rows_written_in_last_mark > 0)
        {
            size_t rows_left_in_last_mark = index_granularity->getMarkRows(getCurrentMark()) - rows_written_in_last_mark;
            /// Previous granularity was much bigger than our new block's
            /// granularity let's adjust it, because we want add new
            /// heavy-weight blocks into small old granule.
            if (rows_left_in_last_mark > index_granularity_for_block)
            {
                /// We have already written more rows than granularity of our block.
                /// adjust last mark rows and flush to disk.
                if (rows_written_in_last_mark >= index_granularity_for_block)
                    adjustLastMarkIfNeedAndFlushToDisk(rows_written_in_last_mark);
                else /// We still can write some rows from new block into previous granule. So the granule size will be block granularity size.
                    adjustLastMarkIfNeedAndFlushToDisk(index_granularity_for_block);
            }
        }

        fillIndexGranularity(index_granularity_for_block, block_to_write.rows());
    }

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

    WrittenOffsetSubstreams offset_substreams = written_offset_substreams ? *written_offset_substreams : WrittenOffsetSubstreams{};

    Block primary_key_block;
    if (settings.rewrite_primary_key)
        primary_key_block = getIndexBlockAndPermute(block, metadata_snapshot->getPrimaryKeyColumns(), permutation, permuted_columns_cache);

    Block skip_indexes_block = getIndexBlockAndPermute(block, getSkipIndicesColumns(), permutation, permuted_columns_cache);

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