Method write

src/DataStreams/PartitionedBlockOutputStream.cpp:50–113

}

void PartitionedBlockOutputStream::write(const Block & block)
{
    const auto & columns_with_type_and_name = block.getColumnsWithTypeAndName();
    const auto & columns = block.getColumns();

    Block block_with_partition_by_expr = sample_block.cloneWithoutColumns();
    block_with_partition_by_expr.setColumns(columns);
    partition_by_expr->execute(block_with_partition_by_expr);

    const auto * partition_by_result_column = block_with_partition_by_expr.getByName(partition_by_column_name).column.get();

    size_t rows_size = block.rows();
    block_row_index_to_partition_index.resize(rows_size);

    partition_id_to_block_index.clear();

    for (size_t row = 0; row < rows_size; ++row)
    {
        auto partition_key = partition_by_result_column->getDataAt(row);
        auto [it, inserted] = partition_id_to_block_index.insert(makePairNoInit(partition_key, partition_id_to_block_index.size()));
        if (inserted)
            it->value.first = copyStringInArena(partition_keys_arena, partition_key);

        block_row_index_to_partition_index[row] = it->getMapped();
    }

    size_t columns_size = columns.size();
    size_t partitions_size = partition_id_to_block_index.size();

    Blocks partition_index_to_block;
    partition_index_to_block.reserve(partitions_size);

    for (size_t column_index = 0; column_index < columns_size; ++column_index)
    {
        MutableColumns partition_index_to_column_split
            = columns[column_index]->scatter(partitions_size, block_row_index_to_partition_index);

        /// add empty block into partition_index_to_block in first loop
        if (column_index == 0)
        {
            size_t size = 0;
            while (size++ < partitions_size)
            {
                partition_index_to_block.emplace_back(Block());
            }
        }

        auto type = columns_with_type_and_name[column_index].type;
        auto name = columns_with_type_and_name[column_index].name;
        for (size_t partition_index = 0; partition_index < partitions_size; ++partition_index)
        {
            ColumnWithTypeAndName column_with_type_and_name(std::move(partition_index_to_column_split[partition_index]), type, name);
            partition_index_to_block[partition_index].insert(std::move(column_with_type_and_name));
        }
    }

    for (const auto & partition : partition_id_to_block_index)
    {