Merge pull request #18483 from ClickHouse/backport/20.11/18381

Backport #18381 to 20.11: Restrict merges from wide to compact parts

src/Storages/MergeTree/MergeTreeDataMergerMutator.cpp

@@ -79,13 +79,16 @@ void FutureMergedMutatedPart::assign(MergeTreeData::DataPartsVector parts_)
 
     size_t sum_rows = 0;
     size_t sum_bytes_uncompressed = 0;
+    MergeTreeDataPartType future_part_type = MergeTreeDataPartType::UNKNOWN;
     for (const auto & part : parts_)
     {
         sum_rows += part->rows_count;
         sum_bytes_uncompressed += part->getTotalColumnsSize().data_uncompressed;
+        future_part_type = std::min(future_part_type, part->getType());
     }
 
-    auto future_part_type = parts_.front()->storage.choosePartTypeOnDisk(sum_bytes_uncompressed, sum_rows);
+    auto chosen_type = parts_.front()->storage.choosePartTypeOnDisk(sum_bytes_uncompressed, sum_rows);
+    future_part_type = std::min(future_part_type, chosen_type);
     assign(std::move(parts_), future_part_type);
 }
 

src/Storages/MergeTree/MergeTreeDataPartType.h

@@ -18,7 +18,7 @@ class MergeTreeDataPartType
         /// Data of all columns is stored in one file. Marks are also stored in single file.
         COMPACT,
 
-        /// Format with buffering data in RAM. Not implemented yet.
+        /// Format with buffering data in RAM.
         IN_MEMORY,
 
         UNKNOWN,
@@ -37,6 +37,16 @@ class MergeTreeDataPartType
         return !(*this == other);
     }
 
+    bool operator<(const MergeTreeDataPartType & other) const
+    {
+        return value < other.value;
+    }
+
+    bool operator>(const MergeTreeDataPartType & other) const
+    {
+        return value > other.value;
+    }
+
     void fromString(const String & str);
     String toString() const;
 

tests/queries/0_stateless/01606_merge_from_wide_to_compact.reference

@@ -0,0 +1,9 @@
+all_1_1_0	Wide
+all_2_2_0	Wide
+all_3_3_0	Wide
+all_1_3_1	Wide
+300000
+all_1_3_1	Wide
+all_4_4_0	Compact
+all_1_4_2	Wide
+400000

tests/queries/0_stateless/01606_merge_from_wide_to_compact.sql

@@ -0,0 +1,36 @@
+DROP TABLE IF EXISTS wide_to_comp;
+
+CREATE TABLE wide_to_comp (a Int, b Int, c Int)
+    ENGINE = MergeTree ORDER BY a
+    settings vertical_merge_algorithm_min_rows_to_activate = 1,
+    vertical_merge_algorithm_min_columns_to_activate = 1,
+    min_bytes_for_wide_part = 0;
+
+SYSTEM STOP merges wide_to_comp;
+
+INSERT INTO wide_to_comp SELECT number, number, number FROM numbers(100000);
+INSERT INTO wide_to_comp SELECT number, number, number FROM numbers(100000);
+INSERT INTO wide_to_comp SELECT number, number, number FROM numbers(100000);
+
+SELECT name, part_type FROM system.parts WHERE table = 'wide_to_comp' AND database = currentDatabase() AND active ORDER BY name;
+
+ALTER TABLE wide_to_comp MODIFY setting min_rows_for_wide_part = 10000000;
+SYSTEM START merges wide_to_comp;
+OPTIMIZE TABLE wide_to_comp FINAL;
+
+SELECT name, part_type FROM system.parts WHERE table = 'wide_to_comp' AND database = currentDatabase() AND active ORDER BY name;
+SELECT count() FROM wide_to_comp WHERE not ignore(*);
+
+SYSTEM STOP merges wide_to_comp;
+INSERT INTO wide_to_comp SELECT number, number, number FROM numbers(100000);
+
+SELECT name, part_type FROM system.parts WHERE table = 'wide_to_comp' AND database = currentDatabase() AND active ORDER BY name;
+
+ALTER TABLE wide_to_comp MODIFY setting min_rows_for_wide_part = 10000000;
+SYSTEM START merges wide_to_comp;
+OPTIMIZE TABLE wide_to_comp FINAL;
+
+SELECT name, part_type FROM system.parts WHERE table = 'wide_to_comp' AND database = currentDatabase() AND active ORDER BY name;
+SELECT count() FROM wide_to_comp WHERE not ignore(*);
+
+DROP TABLE wide_to_comp;