Fix projection merges and mutations.

[amosbird]

I hereby agree to the terms of the CLA available at: https://yandex.ru/legal/cla/?lang=en

Changelog category (leave one):

• Bug Fix (user-visible misbehaviour in official stable or prestable release)

Changelog entry (a user-readable short description of the changes that goes to CHANGELOG.md):
Fix vertical merges of projection parts. This fixes #29253 . This PR also fixes several projection merge/mutation issues introduced in #25165

Detailed description / Documentation draft:
.

[amosbird]

btw, this PR should be backport to earlier versions without MergeTask and MutateTask. I don't know what's the proper way to achieve this.

[desertfury]

Hi!
We are at 22.8.5.29 but still can reproduce this bug

2023.07.17 14:10:36.230305 [ 3947378 ] {} void DB::MergeTreeBackgroundExecutorDB::MergeMutateRuntimeQueue::routine(DB::TaskRuntimeDataPtr) [Queue = DB::MergeMutateRuntimeQueue]: Code: 49. DB::Exception: Number of rows in source parts (256704) differs from number of bytes written to rows_sources file (214282). It is a bug. (LOGICAL_ERROR), Stack trace (when copying this message, always include the lines below):

DB::Exception::Exception(std::__1::basic_string<char, std::__1::char_traits, std::__1::allocator > const&, int, bool) @ 0xa3c597a in /usr/bin/clickhouse
DB::MergeTask::VerticalMergeStage::prepareVerticalMergeForAllColumns() const @ 0x1575b648 in /usr/bin/clickhouse
bool std::__1::__function::__policy_invoker<bool ()>::__call_impl<std::__1::__function::__default_alloc_func<DB::MergeTask::VerticalMergeStage::subtasks::'lambda'(), bool ()> >(std::__1::__function::__policy_storage const*) @ 0x15763b69 in /usr/bin/clickhouse
DB::MergeTask::VerticalMergeStage::execute() @ 0x157601ab in /usr/bin/clickhouse
DB::MergeTask::execute() @ 0x1575fd7a in /usr/bin/clickhouse
DB::MergePlainMergeTreeTask::executeStep() @ 0x1574f592 in /usr/bin/clickhouse
DB::MergeTreeBackgroundExecutorDB::MergeMutateRuntimeQueue::routine(std::__1::shared_ptrDB::TaskRuntimeData) @ 0xa39013b in /usr/bin/clickhouse
DB::MergeTreeBackgroundExecutorDB::MergeMutateRuntimeQueue::threadFunction() @ 0xa38fb70 in /usr/bin/clickhouse
ThreadPoolImpl::worker(std::__1::__list_iterator<ThreadFromGlobalPool, void*>) @ 0xa489a26 in /usr/bin/clickhouse
void std::__1::__function::__policy_invoker<void ()>::__call_impl<std::__1::__function::__default_alloc_func<ThreadFromGlobalPool::ThreadFromGlobalPool<void ThreadPoolImpl::scheduleImpl(std::__1::function<void ()>, int, std::__1::optional)::'lambda0'()>(void&&)::'lambda'(), void ()> >(std::__1::__function::__policy_storage const*) @ 0xa48b377 in /usr/bin/clickhouse
ThreadPoolImplstd::__1::thread::worker(std::__1::__list_iterator<std::__1::thread, void*>) @ 0xa487348 in /usr/bin/clickhouse
? @ 0xa48a55d in /usr/bin/clickhouse
? @ 0x7f0fcb97c609 in ?
clone @ 0x7f0fcb8a3103 in ?
(version 22.8.5.29 (official build))

This is the table schema

│ CREATE TABLE charts.points_v1
(
    `task_id` String CODEC(ZSTD(1)),
    `line` String CODEC(ZSTD(1)),
    `file_hash` UInt32 CODEC(DoubleDelta),
    `created_at` DateTime CODEC(Delta(4), ZSTD(1)),
    `seq_num` UInt64 CODEC(DoubleDelta),
    `timestamp` DateTime64(6) CODEC(Delta(4), ZSTD(1)),
    `receive_time` DateTime64(6) CODEC(Delta(4), ZSTD(1)),
    `value` Float64 CODEC(DoubleDelta),
    `meta` String CODEC(ZSTD(1)),
    `seq_num1` UInt64 CODEC(DoubleDelta),
    `timestamp1` DateTime64(6) CODEC(Delta(4), ZSTD(1)),
    `receive_time1` DateTime64(6) CODEC(Delta(4), ZSTD(1)),
    `value1` Float64 CODEC(DoubleDelta),
    `meta1` String CODEC(ZSTD(1)),
    `seq_num2` UInt64 CODEC(DoubleDelta),
    `timestamp2` DateTime64(6) CODEC(Delta(4), ZSTD(1)),
    `receive_time2` DateTime64(6) CODEC(Delta(4), ZSTD(1)),
    `value2` Float64 CODEC(DoubleDelta),
    `meta2` String CODEC(ZSTD(1)),
    `seq_num3` UInt64 CODEC(DoubleDelta),
    `timestamp3` DateTime64(6) CODEC(Delta(4), ZSTD(1)),
    `receive_time3` DateTime64(6) CODEC(Delta(4), ZSTD(1)),
    `value3` Float64 CODEC(DoubleDelta),
    `meta3` String CODEC(ZSTD(1)),
    `seq_num4` UInt64 CODEC(DoubleDelta),
    `timestamp4` DateTime64(6) CODEC(Delta(4), ZSTD(1)),
    `receive_time4` DateTime64(6) CODEC(Delta(4), ZSTD(1)),
    `value4` Float64 CODEC(DoubleDelta),
    `meta4` String CODEC(ZSTD(1)),
    `seq_num5` UInt64 CODEC(DoubleDelta),
    `timestamp5` DateTime64(6) CODEC(Delta(4), ZSTD(1)),
    `receive_time5` DateTime64(6) CODEC(Delta(4), ZSTD(1)),
    `value5` Float64 CODEC(DoubleDelta),
    `meta5` String CODEC(ZSTD(1)),
    `seq_num6` UInt64 CODEC(DoubleDelta),
    `timestamp6` DateTime64(6) CODEC(Delta(4), ZSTD(1)),
    `receive_time6` DateTime64(6) CODEC(Delta(4), ZSTD(1)),
    `value6` Float64 CODEC(DoubleDelta),
    `meta6` String CODEC(ZSTD(1)),
    `seq_num7` UInt64 CODEC(DoubleDelta),
    `timestamp7` DateTime64(6) CODEC(Delta(4), ZSTD(1)),
    `receive_time7` DateTime64(6) CODEC(Delta(4), ZSTD(1)),
    `value7` Float64 CODEC(DoubleDelta),
    `meta7` String CODEC(ZSTD(1)),
    `seq_num8` UInt64 CODEC(DoubleDelta),
    `timestamp8` DateTime64(6) CODEC(Delta(4), ZSTD(1)),
    `receive_time8` DateTime64(6) CODEC(Delta(4), ZSTD(1)),
    `value8` Float64 CODEC(DoubleDelta),
    `meta8` String CODEC(ZSTD(1)),
    `seq_num9` UInt64 CODEC(DoubleDelta),
    `timestamp9` DateTime64(6) CODEC(Delta(4), ZSTD(1)),
    `receive_time9` DateTime64(6) CODEC(Delta(4), ZSTD(1)),
    `value9` Float64 CODEC(DoubleDelta),
    `meta9` String CODEC(ZSTD(1))
)
ENGINE = ReplacingMergeTree(created_at)
PRIMARY KEY (task_id, file_hash, line, seq_num)
ORDER BY (task_id, file_hash, line, seq_num)
TTL created_at + toIntervalDay(10)
SETTINGS index_granularity = 8192