Don't create empty parts on INSERT

src/Storages/MergeTree/MergeTreeBlockOutputStream.cpp

@@ -29,6 +29,12 @@ void MergeTreeBlockOutputStream::write(const Block & block)
         Stopwatch watch;
 
         MergeTreeData::MutableDataPartPtr part = storage.writer.writeTempPart(current_block, metadata_snapshot, optimize_on_insert);
+
+        /// If optimize_on_insert setting is true, current_block could become empty after merge
+        /// and we didn't create part.
+        if (!part)
+            continue;
+
         storage.renameTempPartAndAdd(part, &storage.increment);
 
         PartLog::addNewPart(storage.global_context, part, watch.elapsed());

src/Storages/MergeTree/MergeTreeDataWriter.cpp

@@ -327,6 +327,11 @@ MergeTreeData::MutableDataPartPtr MergeTreeDataWriter::writeTempPart(BlockWithPa
     /// Size of part would not be greater than block.bytes() + epsilon
     size_t expected_size = block.bytes();
 
+    /// If optimize_on_insert is true, block may become empty after merge.
+    /// There is no need to create empty part.
+    if (expected_size == 0)
+        return nullptr;
+
     DB::IMergeTreeDataPart::TTLInfos move_ttl_infos;
     const auto & move_ttl_entries = metadata_snapshot->getMoveTTLs();
     for (const auto & ttl_entry : move_ttl_entries)

src/Storages/MergeTree/ReplicatedMergeTreeBlockOutputStream.cpp

@@ -144,6 +144,11 @@ void ReplicatedMergeTreeBlockOutputStream::write(const Block & block)
 
         MergeTreeData::MutableDataPartPtr part = storage.writer.writeTempPart(current_block, metadata_snapshot, optimize_on_insert);
 
+        /// If optimize_on_insert setting is true, current_block could become empty after merge
+        /// and we didn't create part.
+        if (!part)
+            continue;
+
         String block_id;
 
         if (deduplicate)

tests/queries/0_stateless/01560_optimize_on_insert.reference

@@ -11,3 +11,4 @@ Summing Merge Tree
 Aggregating Merge Tree
 1	5	2020-01-01 00:00:00
 2	5	2020-01-02 00:00:00
+Check creating empty parts

tests/queries/0_stateless/01560_optimize_on_insert.sql

@@ -33,3 +33,10 @@ INSERT INTO aggregating_merge_tree VALUES (1, 1, '2020-01-01'), (2, 1, '2020-01-
 SELECT * FROM aggregating_merge_tree ORDER BY key;
 DROP TABLE aggregating_merge_tree;
 
+SELECT 'Check creating empty parts';
+DROP TABLE IF EXISTS empty;
+CREATE TABLE empty (key UInt32, val UInt32, date Datetime) ENGINE=SummingMergeTree(val) PARTITION BY date ORDER BY key;
+INSERT INTO empty VALUES (1, 1, '2020-01-01'), (1, 1, '2020-01-01'), (1, -2, '2020-01-01');
+SELECT * FROM empty ORDER BY key;
+SELECT table, partition, active FROM system.parts where table = 'empty' and active = 1;
+DROP TABLE empty;