Method writeWithProjections

src/WorkerTasks/MergeTreeDataMutator.cpp:874–1054

1. get projection pipeline and a sink to write parts 2. build an executor that can write block to the input stream (actually we can write through it to generate as many parts as possible) 3. finalize the pipeline so that all parts are merged into one part

// 2. build an executor that can write block to the input stream (actually we can write through it to generate as many parts as possible)
// 3. finalize the pipeline so that all parts are merged into one part
std::optional<size_t> MergeTreeDataMutator::writeWithProjections(
    MergeTreeData::MutableDataPartPtr new_data_part,
    const StorageMetadataPtr & metadata_snapshot,
    const MergeTreeProjections & projections_to_build,
    BlockInputStreamPtr mutating_stream,
    IMergedBlockOutputStream & out,
    time_t time_of_mutation,
    ManipulationListEntry & manipulation_entry,
    const ReservationPtr & space_reservation,
    TableLockHolder & holder,
    ContextPtr context,
    IMergeTreeDataPart::MinMaxIndex * minmax_idx)
{
    std::optional<size_t> row_exists_count = is_delete_command ? std::make_optional(0ull) : std::nullopt;
    size_t block_num = 0;
    Block block;
    std::map<String, MergeTreeData::MutableDataPartsVector> projection_parts;
    std::vector<SquashingTransform> projection_squashes;
    for (size_t i = 0, size = projections_to_build.size(); i < size; ++i)
    {
        projection_squashes.emplace_back(65536, 65536 * 256);
    }

    LOG_DEBUG(log, "Begin to write with projections, part name is {}, need to rebuild {} projections", new_data_part->name, projections_to_build.size());

    while (checkOperationIsNotCanceled(manipulation_entry) && (block = mutating_stream->read()))
    {  
        if (minmax_idx)
            minmax_idx->update(block, data.getMinMaxColumnsNames(metadata_snapshot->getPartitionKey()));

        /// Count how many rows are existing when writing blocks. It will be stored in part metadata.
        if (is_delete_command)
        {
            auto row_exists_column = block.getByName(RowExistsColumn::ROW_EXISTS_COLUMN.name).column;
            for (size_t i = 0; i < block.rows(); ++i)
            {
                if (row_exists_column->getBool(i))
                    ++*row_exists_count;
            }
        }

        out.write(block);

        for (size_t i = 0, size = projections_to_build.size(); i < size; ++i)
        {
            const auto & projection = projections_to_build[i]->projection;
            auto in = InterpreterSelectQuery(
                          projection.query_ast,
                          context,
                          Pipe(std::make_shared<SourceFromSingleChunk>(block, Chunk(block.getColumns(), block.rows()))),
                          SelectQueryOptions{
                              projection.type == ProjectionDescription::Type::Normal ? QueryProcessingStage::FetchColumns : QueryProcessingStage::WithMergeableState})
                          .execute()
                          .getInputStream();
            in = std::make_shared<SquashingBlockInputStream>(in, block.rows(), std::numeric_limits<UInt64>::max());
            in->readPrefix();
            auto & projection_squash = projection_squashes[i];
            auto projection_block = projection_squash.add(in->read());