Merge pull request #34316 from ClickHouse/probably-fix-data-race-in-WriteBufferFromS3

Probably fix data race in WriteBufferFromS3 destructor.

src/IO/WriteBufferFromS3.cpp

@@ -201,9 +201,12 @@ void WriteBufferFromS3::writePart()
                 std::lock_guard lock(bg_tasks_mutex);
                 task->is_finised = true;
                 ++num_finished_bg_tasks;
-            }
 
-            bg_tasks_condvar.notify_one();
+                /// Notification under mutex is important here.
+                /// Othervies, WriteBuffer could be destroyed in between
+                /// Releasing lock and condvar notification.
+                bg_tasks_condvar.notify_one();
+            }
         });
     }
     else
@@ -305,9 +308,13 @@ void WriteBufferFromS3::makeSinglepartUpload()
             {
                 std::lock_guard lock(bg_tasks_mutex);
                 put_object_task->is_finised = true;
+
+                /// Notification under mutex is important here.
+                /// Othervies, WriteBuffer could be destroyed in between
+                /// Releasing lock and condvar notification.
+                bg_tasks_condvar.notify_one();
             }
 
-            bg_tasks_condvar.notify_one();
         });
     }
     else

src/Storages/MergeTree/MergeTreeDataPartWriterWide.cpp

@@ -412,6 +412,11 @@ void MergeTreeDataPartWriterWide::validateColumnOfFixedSize(const NameAndTypePai
     String escaped_name = escapeForFileName(name);
     String mrk_path = part_path + escaped_name + marks_file_extension;
     String bin_path = part_path + escaped_name + DATA_FILE_EXTENSION;
+
+    /// Some columns may be removed because of ttl. Skip them.
+    if (!disk->exists(mrk_path))
+        return;
+
     auto mrk_in = disk->readFile(mrk_path);
     DB::CompressedReadBufferFromFile bin_in(disk->readFile(bin_path));
     bool must_be_last = false;

src/Storages/MergeTree/MergedBlockOutputStream.cpp

@@ -66,10 +66,9 @@ struct MergedBlockOutputStream::Finalizer::Impl
 
 void MergedBlockOutputStream::Finalizer::finish()
 {
-    if (impl)
-        impl->finish();
-
-    impl.reset();
+    std::unique_ptr<Impl> to_finish = std::move(impl);
+    if (to_finish)
+        to_finish->finish();
 }
 
 void MergedBlockOutputStream::Finalizer::Impl::finish()