Merge pull request #23459 from CurtizJ/fix-replacing

Fix corner cases in vertical merges with ReplacingMergeTree

src/DataStreams/ColumnGathererStream.cpp

@@ -53,7 +53,7 @@ ColumnGathererStream::ColumnGathererStream(
 Block ColumnGathererStream::readImpl()
 {
     /// Special case: single source and there are no skipped rows
-    if (children.size() == 1 && row_sources_buf.eof())
+    if (children.size() == 1 && row_sources_buf.eof() && !source_to_fully_copy)
         return children[0]->read();
 
     if (!source_to_fully_copy && row_sources_buf.eof())

src/Processors/Merges/Algorithms/ReplacingSortedAlgorithm.cpp

@@ -93,6 +93,10 @@ IMergingAlgorithm::Status ReplacingSortedAlgorithm::merge()
         }
     }
 
+    /// If have enough rows, return block, because it prohibited to overflow requested number of rows.
+    if (merged_data.hasEnoughRows())
+        return Status(merged_data.pull());
+
     /// We will write the data for the last primary key.
     if (!selected_row.empty())
         insertRow();

tests/queries/0_stateless/00926_adaptive_index_granularity_replacing_merge_tree.reference

@@ -1,7 +1,7 @@
 4
 4
 8
-7
+8
 -----
 4
 2

tests/queries/0_stateless/01825_replacing_vertical_merge.reference

@@ -0,0 +1,4 @@
+1720	32
+220	17
+33558527	8193
+33550336	8192

tests/queries/0_stateless/01825_replacing_vertical_merge.sql

@@ -0,0 +1,48 @@
+SET optimize_on_insert = 0;
+
+DROP TABLE IF EXISTS replacing_table;
+
+CREATE TABLE replacing_table (a UInt32, b UInt32, c UInt32)
+ENGINE = ReplacingMergeTree ORDER BY a
+SETTINGS vertical_merge_algorithm_min_rows_to_activate = 1,
+    vertical_merge_algorithm_min_columns_to_activate = 1,
+    index_granularity = 16,
+    min_bytes_for_wide_part = 0,
+    merge_max_block_size = 16;
+
+SYSTEM STOP MERGES replacing_table;
+
+INSERT INTO replacing_table SELECT number, number, number from numbers(16);
+INSERT INTO replacing_table SELECT 100, number, number from numbers(16);
+
+SELECT sum(a), count() FROM replacing_table;
+
+SYSTEM START MERGES replacing_table;
+
+OPTIMIZE TABLE replacing_table FINAL;
+
+SELECT sum(a), count() FROM replacing_table;
+
+DROP TABLE IF EXISTS replacing_table;
+
+CREATE TABLE replacing_table
+(
+    key UInt64,
+    value UInt64
+)
+ENGINE = ReplacingMergeTree
+ORDER BY key
+SETTINGS
+    vertical_merge_algorithm_min_rows_to_activate=0,
+    vertical_merge_algorithm_min_columns_to_activate=0,
+    min_bytes_for_wide_part = 0;
+
+INSERT INTO replacing_table SELECT if(number == 8192, 8191, number), 1 FROM numbers(8193);
+
+SELECT sum(key), count() from replacing_table;
+
+OPTIMIZE TABLE replacing_table FINAL;
+
+SELECT sum(key), count() from replacing_table;
+
+DROP TABLE IF EXISTS replacing_table;