Method mutateSomePartColumns

src/WorkerTasks/MergeTreeDataMutator.cpp:1145–1203

}

void MergeTreeDataMutator::mutateSomePartColumns(
    const MergeTreeDataPartPtr & source_part,
    const StorageMetadataPtr & metadata_snapshot,
    const std::set<MergeTreeIndexPtr> & indices_to_recalc,
    const std::set<MergeTreeProjectionPtr> & projections_to_recalc,
    const Block & mutation_header,
    MergeTreeData::MutableDataPartPtr new_data_part,
    BlockInputStreamPtr mutating_stream,
    time_t time_of_mutation,
    const CompressionCodecPtr & compression_codec,
    ManipulationListEntry & manipulation_entry,
    bool need_sync,
    const ReservationPtr & space_reservation,
    TableLockHolder & holder,
    ContextPtr context)
{
    if (mutating_stream == nullptr)
        throw Exception("Cannot mutate part columns with uninitialized mutations stream. It's a bug", ErrorCodes::LOGICAL_ERROR);

    if (new_data_part->versions->enable_compact_map_data)
        throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot mutate part {} to compact part. It's a bug.", source_part->name);

    IMergedBlockOutputStream::WrittenOffsetColumns unused_written_offsets;
    MergedColumnOnlyOutputStream out(
        new_data_part,
        metadata_snapshot,
        mutation_header,
        compression_codec,
        std::vector<MergeTreeIndexPtr>(indices_to_recalc.begin(), indices_to_recalc.end()),
        nullptr,
        source_part->index_granularity,
        &source_part->index_granularity_info
    );

    mutating_stream->readPrefix();
    out.writePrefix();

    std::vector<MergeTreeProjectionPtr> projections_to_build(projections_to_recalc.begin(), projections_to_recalc.end());
    auto row_exists_count = writeWithProjections(
        new_data_part,
        metadata_snapshot,
        projections_to_build,
        mutating_stream,
        out,
        time_of_mutation,
        manipulation_entry,
        space_reservation,
        holder,
        context);

    mutating_stream->readSuffix();

    auto changed_checksums = out.writeSuffixAndGetChecksums(new_data_part, *(new_data_part->getChecksums()), need_sync);
    new_data_part->checksums_ptr->add(std::move(changed_checksums));

    /// New part's row_exists_count should be updated if it's DELETE mutation, otherwise it use source part's value.
    if (row_exists_count.has_value())
        new_data_part->row_exists_count = row_exists_count;