Merge pull request ClickHouse#90059 from ClickHouse/unblock_ttl_drop_merges_in_cold_volumes

Michicosun · zvonand · commit 016b43a6d503 · 2026-02-03T18:05:27.000+01:00
Unblock ttl part drops for cold volumes
diff --git a/src/Storages/MergeTree/Compaction/MergePredicates/DistributedMergePredicate.h b/src/Storages/MergeTree/Compaction/MergePredicates/DistributedMergePredicate.h
@@ -82,6 +82,9 @@ class DistributedMergePredicate : public IMergePredicate
         if (left.info.isPatch() != right.info.isPatch())
             return std::unexpected(PreformattedMessage::create("One of parts ({}, {}) is patch part and another is regular part", left.name, right.name));
 
+        if (left.is_in_volume_where_merges_avoid || right.is_in_volume_where_merges_avoid)
+            return std::unexpected(PreformattedMessage::create("One of parts ({}, {}) lies on volume where merges should be avoided", left.name, right.name));
+
         int64_t left_max_block = left.info.max_block;
         int64_t right_min_block = right.info.min_block;
         chassert(left_max_block < right_min_block);
diff --git a/src/Storages/MergeTree/Compaction/MergePredicates/MergeTreeMergePredicate.cpp b/src/Storages/MergeTree/Compaction/MergePredicates/MergeTreeMergePredicate.cpp
@@ -37,6 +37,9 @@ std::expected<void, PreformattedMessage> MergeTreeMergePredicate::canMergeParts(
     if (left.info.isPatch() != right.info.isPatch())
         return std::unexpected(PreformattedMessage::create("One of parts ({}, {}) is patch part and another is regular part", left.name, right.name));
 
+    if (left.is_in_volume_where_merges_avoid || right.is_in_volume_where_merges_avoid)
+        return std::unexpected(PreformattedMessage::create("One of parts ({}, {}) lies on volume where merges should be avoided", left.name, right.name));
+
     if (left.projection_names != right.projection_names)
     {
         return std::unexpected(PreformattedMessage::create(
diff --git a/src/Storages/MergeTree/Compaction/MergeSelectors/TTLMergeSelector.cpp b/src/Storages/MergeTree/Compaction/MergeSelectors/TTLMergeSelector.cpp
@@ -213,6 +213,9 @@ time_t TTLRowDeleteMergeSelector::getTTLForPart(const PartProperties & part) con
 
 bool TTLRowDeleteMergeSelector::canConsiderPart(const PartProperties & part) const
 {
+    if (part.is_in_volume_where_merges_avoid)
+        return false;
+
     if (!part.general_ttl_info.has_value())
         return false;
 
@@ -231,6 +234,9 @@ time_t TTLRecompressMergeSelector::getTTLForPart(const PartProperties & part) co
 
 bool TTLRecompressMergeSelector::canConsiderPart(const PartProperties & part) const
 {
+    if (part.is_in_volume_where_merges_avoid)
+        return false;
+
     if (!part.recompression_ttl_info.has_value())
         return false;
 
diff --git a/src/Storages/MergeTree/Compaction/PartProperties.cpp b/src/Storages/MergeTree/Compaction/PartProperties.cpp
@@ -67,13 +67,15 @@ std::set<std::string> getCalculatedProjectionNames(const MergeTreeDataPartPtr &
 PartProperties buildPartProperties(
     const MergeTreeDataPartPtr & part,
     const StorageMetadataPtr & metadata_snapshot,
+    const StoragePolicyPtr & storage_policy,
     time_t current_time)
 {
     return PartProperties{
         .name = part->name,
         .info = part->info,
         .projection_names = getCalculatedProjectionNames(part),
         .all_ttl_calculated_if_any = part->checkAllTTLCalculated(metadata_snapshot),
+        .is_in_volume_where_merges_avoid = !part->shallParticipateInMerges(storage_policy),
         .size = part->getExistingBytesOnDisk(),
         .age = current_time - part->modification_time,
         .general_ttl_info = buildGeneralTTLInfo(metadata_snapshot, part),
diff --git a/src/Storages/MergeTree/Compaction/PartProperties.h b/src/Storages/MergeTree/Compaction/PartProperties.h
@@ -1,6 +1,9 @@
 #pragma once
 
 #include <Storages/MergeTree/MergeTreePartInfo.h>
+
+#include <Disks/IStoragePolicy.h>
+
 #include <Core/UUID.h>
 
 #include <optional>
@@ -24,6 +27,7 @@ struct PartProperties
     const std::set<std::string> projection_names = {};
 
     const bool all_ttl_calculated_if_any = false;
+    const bool is_in_volume_where_merges_avoid = false;
 
     /// Size of data part in bytes.
     const size_t size = 0;
@@ -56,6 +60,7 @@ using PartsRangeView = std::span<const PartProperties>;
 PartProperties buildPartProperties(
     const MergeTreeDataPartPtr & part,
     const StorageMetadataPtr & metadata_snapshot,
+    const StoragePolicyPtr & storage_policy,
     time_t current_time);
 
 }
diff --git a/src/Storages/MergeTree/Compaction/PartsCollectors/Common.cpp b/src/Storages/MergeTree/Compaction/PartsCollectors/Common.cpp
@@ -6,7 +6,10 @@ namespace DB
 {
 
 PartsRanges constructPartsRanges(
-    std::vector<MergeTreeDataPartsVector> && ranges, const StorageMetadataPtr & metadata_snapshot, const time_t & current_time)
+    std::vector<MergeTreeDataPartsVector> && ranges,
+    const StorageMetadataPtr & metadata_snapshot,
+    const StoragePolicyPtr & storage_policy,
+    const time_t & current_time)
 {
     PartsRanges properties_ranges;
     properties_ranges.reserve(ranges.size());
@@ -17,7 +20,7 @@ PartsRanges constructPartsRanges(
         properties_ranges.reserve(range.size());
 
         for (const auto & part : range)
-            properties_range.push_back(buildPartProperties(part, metadata_snapshot, current_time));
+            properties_range.push_back(buildPartProperties(part, metadata_snapshot, storage_policy, current_time));
 
         properties_ranges.push_back(std::move(properties_range));
     }
diff --git a/src/Storages/MergeTree/Compaction/PartsCollectors/Common.h b/src/Storages/MergeTree/Compaction/PartsCollectors/Common.h
@@ -55,7 +55,10 @@ std::expected<void, PreformattedMessage> checkAllPartsSatisfyPredicate(const std
 }
 
 PartsRanges constructPartsRanges(
-    std::vector<MergeTreeDataPartsVector> && ranges, const StorageMetadataPtr & metadata_snapshot, const time_t & current_time);
+    std::vector<MergeTreeDataPartsVector> && ranges,
+    const StorageMetadataPtr & metadata_snapshot,
+    const StoragePolicyPtr & storage_policy,
+    const time_t & current_time);
 
 MergeTreeDataPartsVector filterByPartitions(
     MergeTreeDataPartsVector && parts, const std::optional<PartitionIdsHint> & partitions_to_keep);
diff --git a/src/Storages/MergeTree/Compaction/PartsCollectors/MergeTreePartsCollector.cpp b/src/Storages/MergeTree/Compaction/PartsCollectors/MergeTreePartsCollector.cpp
@@ -77,9 +77,7 @@ MergeTreeDataPartsVector collectInitial(const MergeTreeData & data, const MergeT
 
 auto constructPreconditionsPredicate(const StoragePolicyPtr & storage_policy, const MergeTreeTransactionPtr & tx, const MergeTreeMergePredicatePtr & merge_pred)
 {
-    bool has_volumes_with_disabled_merges = storage_policy->hasAnyVolumeWithDisabledMerges();
-
-    auto predicate = [storage_policy, tx, merge_pred, has_volumes_with_disabled_merges](const MergeTreeDataPartPtr & part) -> std::expected<void, PreformattedMessage>
+    auto predicate = [storage_policy, tx, merge_pred](const MergeTreeDataPartPtr & part) -> std::expected<void, PreformattedMessage>
     {
         if (tx)
         {
@@ -93,9 +91,6 @@ auto constructPreconditionsPredicate(const StoragePolicyPtr & storage_policy, co
                 return std::unexpected(PreformattedMessage::create("Part {} is locked for removal", part->name));
         }
 
-        if (has_volumes_with_disabled_merges && !part->shallParticipateInMerges(storage_policy))
-            return std::unexpected(PreformattedMessage::create("Merges for part's {} volume are disabled", part->name));
-
         chassert(merge_pred);
         return merge_pred->canUsePartInMerges(part);
     };
@@ -135,7 +130,7 @@ PartsRanges MergeTreePartsCollector::grabAllPossibleRanges(
 {
     auto parts = filterByPartitions(collectInitial(storage, tx), partitions_hint);
     auto ranges = splitPartsByPreconditions(std::move(parts), storage_policy, tx, merge_pred, series_log);
-    return constructPartsRanges(std::move(ranges), metadata_snapshot, current_time);
+    return constructPartsRanges(std::move(ranges), metadata_snapshot, storage_policy, current_time);
 }
 
 std::expected<PartsRange, PreformattedMessage> MergeTreePartsCollector::grabAllPartsInsidePartition(
@@ -148,7 +143,7 @@ std::expected<PartsRange, PreformattedMessage> MergeTreePartsCollector::grabAllP
     if (auto result = checkAllParts(parts, storage_policy, tx, merge_pred); !result)
         return std::unexpected(std::move(result.error()));
 
-    auto ranges = constructPartsRanges({std::move(parts)}, metadata_snapshot, current_time);
+    auto ranges = constructPartsRanges({std::move(parts)}, metadata_snapshot, storage_policy, current_time);
     chassert(ranges.size() == 1);
 
     return std::move(ranges.front());
diff --git a/src/Storages/MergeTree/Compaction/PartsCollectors/ReplicatedMergeTreePartsCollector.cpp b/src/Storages/MergeTree/Compaction/PartsCollectors/ReplicatedMergeTreePartsCollector.cpp
@@ -17,14 +17,8 @@ MergeTreeDataPartsVector collectInitial(const MergeTreeData & data)
 
 auto constructPreconditionsPredicate(const StoragePolicyPtr & storage_policy, const ReplicatedMergeTreeMergePredicatePtr & merge_pred)
 {
-    bool has_volumes_with_disabled_merges = storage_policy->hasAnyVolumeWithDisabledMerges();
-
-    auto predicate = [storage_policy, merge_pred, has_volumes_with_disabled_merges](const MergeTreeDataPartPtr & part) -> std::expected<void, PreformattedMessage>
+    auto predicate = [storage_policy, merge_pred](const MergeTreeDataPartPtr & part) -> std::expected<void, PreformattedMessage>
     {
-        if (has_volumes_with_disabled_merges && !part->shallParticipateInMerges(storage_policy))
-            return std::unexpected(PreformattedMessage::create("Merges for part's {} volume are disabled", part->name));
-
-        chassert(merge_pred);
         return merge_pred->canUsePartInMerges(part);
     };
 
@@ -62,7 +56,7 @@ PartsRanges ReplicatedMergeTreePartsCollector::grabAllPossibleRanges(
 {
     auto parts = filterByPartitions(collectInitial(storage), partitions_hint);
     auto ranges = splitPartsByPreconditions(std::move(parts), storage_policy, merge_pred, series_log);
-    return constructPartsRanges(std::move(ranges), metadata_snapshot, current_time);
+    return constructPartsRanges(std::move(ranges), metadata_snapshot, storage_policy, current_time);
 }
 
 std::expected<PartsRange, PreformattedMessage> ReplicatedMergeTreePartsCollector::grabAllPartsInsidePartition(
@@ -75,7 +69,7 @@ std::expected<PartsRange, PreformattedMessage> ReplicatedMergeTreePartsCollector
     if (auto result = checkAllParts(parts, storage_policy, merge_pred); !result)
         return std::unexpected(std::move(result.error()));
 
-    auto ranges = constructPartsRanges({std::move(parts)}, metadata_snapshot, current_time);
+    auto ranges = constructPartsRanges({std::move(parts)}, metadata_snapshot, storage_policy, current_time);
     chassert(ranges.size() == 1);
 
     return std::move(ranges.front());
diff --git a/src/Storages/MergeTree/MergeTreeDataMergerMutator.cpp b/src/Storages/MergeTree/MergeTreeDataMergerMutator.cpp
@@ -300,10 +300,11 @@ MergeSelectorChoices chooseMergesFrom(
 
         for (size_t i = 0; i < choices.size(); ++i)
         {
+            const auto & merge_type = choices[i].merge_type;
             const auto & range = choices[i].range;
             const auto & range_patches = choices[i].range_patches;
             ProfileEvents::increment(ProfileEvents::MergerMutatorSelectRangePartsCount, range.size());
-            LOG_TRACE(log, "Merge #{} with {} parts from {} to {} with {} patches", i, range.size(), range.front().name, range.back().name, range_patches.size());
+            LOG_TRACE(log, "Merge #{} type {} with {} parts from {} to {} with {} patches", i, merge_type, range.size(), range.front().name, range.back().name, range_patches.size());
         }
     }
 
diff --git a/tests/integration/test_no_merges_volume_ttl/__init__.py b/tests/integration/test_no_merges_volume_ttl/__init__.py
diff --git a/tests/integration/test_no_merges_volume_ttl/configs/storage_config.xml b/tests/integration/test_no_merges_volume_ttl/configs/storage_config.xml
@@ -0,0 +1,25 @@
+<clickhouse>
+    <storage_configuration>
+        <disks>
+            <with_merges>
+                <path>/with_merges/</path>
+            </with_merges>
+            <no_merges>
+                <path>/no_merges/</path>
+            </no_merges>
+        </disks>
+        <policies>
+            <hot_cold_separation_policy>
+                <volumes>
+                    <with_merges>
+                        <disk>with_merges</disk>
+                    </with_merges>
+                    <no_merges>
+                        <disk>no_merges</disk>
+                        <prefer_not_to_merge>true</prefer_not_to_merge>
+                    </no_merges>
+                </volumes>
+            </hot_cold_separation_policy>
+        </policies>
+    </storage_configuration>
+</clickhouse>
diff --git a/tests/integration/test_no_merges_volume_ttl/test.py b/tests/integration/test_no_merges_volume_ttl/test.py
@@ -0,0 +1,80 @@
+import pytest
+import time
+
+from helpers.cluster import ClickHouseCluster
+
+cluster = ClickHouseCluster(__file__)
+node = cluster.add_instance(
+    "node",
+    main_configs=["configs/storage_config.xml"],
+    tmpfs=["/with_merges:size=200M", "/no_merges:size=200M"],
+)
+
+@pytest.fixture(scope="module")
+def start_cluster():
+    try:
+        cluster.start()
+        yield cluster
+
+    finally:
+        cluster.shutdown()
+
+
+def wait_parts_count(table, expected_number_of_parts):
+    for i in range(100):
+        print(f"Waiting {expected_number_of_parts} for table {table}. Iteration: {i}")
+        parts_count = int(node.query(f"select count() from system.parts where table = '{table}' and active"))
+
+        if parts_count == expected_number_of_parts:
+            break
+
+        time.sleep(1)
+
+    assert int(node.query(f"select count() from system.parts where table = '{table}' and active")) == expected_number_of_parts
+
+
+def test_no_merges_volume_ttl_merge(start_cluster):
+    node.query("create table t (time DateTime) engine = MergeTree order by tuple() ttl time settings storage_policy='hot_cold_separation_policy', merge_with_ttl_timeout=0")
+    table_uuid = node.query("select uuid from system.tables where table = 't'").strip()
+
+    node.query("system stop merges t")
+    node.query("insert into t values (now() - interval 1 day)")
+    assert node.query("select path from system.parts where table = 't' and active").strip() == f"/with_merges/store/{table_uuid[:3]}/{table_uuid}/all_1_1_0/"
+    assert int(node.query("select count() from t").strip()) == 1
+
+    node.query("alter table t move partition () to volume 'no_merges'")
+    assert node.query("select path from system.parts where table = 't' and active").strip() == f"/no_merges/store/{table_uuid[:3]}/{table_uuid}/all_1_1_0/"
+    assert int(node.query("select count() from t").strip()) == 1
+
+    node.query("system start merges t")
+    wait_parts_count("t", 0)
+
+    assert int(node.query("select count() from t").strip()) == 0
+    node.query("drop table t sync")
+
+
+def test_no_merges_volume_no_regular_merges(start_cluster):
+    node.query("create table t (a UInt64) engine = MergeTree order by tuple() settings storage_policy='hot_cold_separation_policy'")
+
+    node.query("system stop merges t")
+    node.query("insert into t select number from numbers(50) settings max_block_size=1, min_insert_block_size_bytes=1")
+    assert int(node.query("select count() from t").strip()) == 50
+    assert int(node.query("select count() from system.parts where table = 't' and active").strip()) == 50
+
+    node.query("alter table t move partition () to volume 'no_merges'")
+    assert int(node.query("select count() from t").strip()) == 50
+    assert int(node.query("select count() from system.parts where table = 't' and active").strip()) == 50
+
+    node.query("system start merges t")
+    node.query("optimize table t")
+    node.query("optimize table t")
+    node.query("optimize table t")
+    node.query("optimize table t")
+    node.query("optimize table t final")
+    node.query("optimize table t final")
+    node.query("optimize table t final")
+    node.query("optimize table t final")
+
+    assert int(node.query("select count() from t").strip()) == 50
+    assert int(node.query("select count() from system.parts where table = 't' and active").strip()) == 50
+    node.query("drop table t sync")

Original file line number	Diff line number	Diff line change
`@@ -37,6 +37,9 @@ std::expected<void, PreformattedMessage> MergeTreeMergePredicate::canMergeParts(`
`37`	`37`	`if (left.info.isPatch() != right.info.isPatch())`
`38`	`38`	`return std::unexpected(PreformattedMessage::create("One of parts ({}, {}) is patch part and another is regular part", left.name, right.name));`
`39`	`39`
	`40`	`+ if (left.is_in_volume_where_merges_avoid \|\| right.is_in_volume_where_merges_avoid)`
	`41`	`+ return std::unexpected(PreformattedMessage::create("One of parts ({}, {}) lies on volume where merges should be avoided", left.name, right.name));`
	`42`	`+`
`40`	`43`	`if (left.projection_names != right.projection_names)`
`41`	`44`	`{`
`42`	`45`	`return std::unexpected(PreformattedMessage::create(`
Original file line number	Diff line number	Diff line change
`@@ -6,7 +6,10 @@ namespace DB`
`6`	`6`	`{`
`7`	`7`
`8`	`8`	`PartsRanges constructPartsRanges(`
`9`		`- std::vector<MergeTreeDataPartsVector> && ranges, const StorageMetadataPtr & metadata_snapshot, const time_t & current_time)`
	`9`	`+ std::vector<MergeTreeDataPartsVector> && ranges,`
	`10`	`+ const StorageMetadataPtr & metadata_snapshot,`
	`11`	`+ const StoragePolicyPtr & storage_policy,`
	`12`	`+ const time_t & current_time)`
`10`	`13`	`{`
`11`	`14`	`PartsRanges properties_ranges;`
`12`	`15`	`properties_ranges.reserve(ranges.size());`
`@@ -17,7 +20,7 @@ PartsRanges constructPartsRanges(`
`17`	`20`	`properties_ranges.reserve(range.size());`
`18`	`21`
`19`	`22`	`for (const auto & part : range)`
`20`		`- properties_range.push_back(buildPartProperties(part, metadata_snapshot, current_time));`
	`23`	`+ properties_range.push_back(buildPartProperties(part, metadata_snapshot, storage_policy, current_time));`
`21`	`24`
`22`	`25`	`properties_ranges.push_back(std::move(properties_range));`
`23`	`26`	`}`
Original file line number	Diff line number	Diff line change
`@@ -77,9 +77,7 @@ MergeTreeDataPartsVector collectInitial(const MergeTreeData & data, const MergeT`
`77`	`77`
`78`	`78`	`auto constructPreconditionsPredicate(const StoragePolicyPtr & storage_policy, const MergeTreeTransactionPtr & tx, const MergeTreeMergePredicatePtr & merge_pred)`
`79`	`79`	`{`
`80`		`- bool has_volumes_with_disabled_merges = storage_policy->hasAnyVolumeWithDisabledMerges();`
`81`		`-`
`82`		`- auto predicate = [storage_policy, tx, merge_pred, has_volumes_with_disabled_merges](const MergeTreeDataPartPtr & part) -> std::expected<void, PreformattedMessage>`
	`80`	`+ auto predicate = [storage_policy, tx, merge_pred](const MergeTreeDataPartPtr & part) -> std::expected<void, PreformattedMessage>`
`83`	`81`	`{`
`84`	`82`	`if (tx)`
`85`	`83`	`{`
`@@ -93,9 +91,6 @@ auto constructPreconditionsPredicate(const StoragePolicyPtr & storage_policy, co`
`93`	`91`	`return std::unexpected(PreformattedMessage::create("Part {} is locked for removal", part->name));`
`94`	`92`	`}`
`95`	`93`
`96`		`- if (has_volumes_with_disabled_merges && !part->shallParticipateInMerges(storage_policy))`
`97`		`- return std::unexpected(PreformattedMessage::create("Merges for part's {} volume are disabled", part->name));`
`98`		`-`
`99`	`94`	`chassert(merge_pred);`
`100`	`95`	`return merge_pred->canUsePartInMerges(part);`
`101`	`96`	`};`
`@@ -135,7 +130,7 @@ PartsRanges MergeTreePartsCollector::grabAllPossibleRanges(`
`135`	`130`	`{`
`136`	`131`	`auto parts = filterByPartitions(collectInitial(storage, tx), partitions_hint);`
`137`	`132`	`auto ranges = splitPartsByPreconditions(std::move(parts), storage_policy, tx, merge_pred, series_log);`
`138`		`- return constructPartsRanges(std::move(ranges), metadata_snapshot, current_time);`
	`133`	`+ return constructPartsRanges(std::move(ranges), metadata_snapshot, storage_policy, current_time);`
`139`	`134`	`}`
`140`	`135`
`141`	`136`	`std::expected<PartsRange, PreformattedMessage> MergeTreePartsCollector::grabAllPartsInsidePartition(`
`@@ -148,7 +143,7 @@ std::expected<PartsRange, PreformattedMessage> MergeTreePartsCollector::grabAllP`
`148`	`143`	`if (auto result = checkAllParts(parts, storage_policy, tx, merge_pred); !result)`
`149`	`144`	`return std::unexpected(std::move(result.error()));`
`150`	`145`
`151`		`- auto ranges = constructPartsRanges({std::move(parts)}, metadata_snapshot, current_time);`
	`146`	`+ auto ranges = constructPartsRanges({std::move(parts)}, metadata_snapshot, storage_policy, current_time);`
`152`	`147`	`chassert(ranges.size() == 1);`
`153`	`148`
`154`	`149`	`return std::move(ranges.front());`
Original file line number	Diff line number	Diff line change
`@@ -300,10 +300,11 @@ MergeSelectorChoices chooseMergesFrom(`
`300`	`300`
`301`	`301`	`for (size_t i = 0; i < choices.size(); ++i)`
`302`	`302`	`{`
	`303`	`+ const auto & merge_type = choices[i].merge_type;`
`303`	`304`	`const auto & range = choices[i].range;`
`304`	`305`	`const auto & range_patches = choices[i].range_patches;`
`305`	`306`	`ProfileEvents::increment(ProfileEvents::MergerMutatorSelectRangePartsCount, range.size());`
`306`		`- LOG_TRACE(log, "Merge #{} with {} parts from {} to {} with {} patches", i, range.size(), range.front().name, range.back().name, range_patches.size());`
	`307`	`+ LOG_TRACE(log, "Merge #{} type {} with {} parts from {} to {} with {} patches", i, merge_type, range.size(), range.front().name, range.back().name, range_patches.size());`
`307`	`308`	`}`
`308`	`309`	`}`
`309`	`310`