do not lock parts, only hold references so they are not deleted from disk. first attempt

Author: arthurpassos

src/Storages/MergeTree/MergeMutateSelectedEntry.h

@@ -51,14 +51,4 @@ struct MergeMutateSelectedEntry
 
 using MergeMutateSelectedEntryPtr = std::shared_ptr<MergeMutateSelectedEntry>;
 
-struct CurrentlyExportingPartsTagger
-{
-    DataPartPtr part_to_export;
-    StorageMergeTree & storage;
-    ~CurrentlyExportingPartsTagger();
-    CurrentlyExportingPartsTagger(DataPartPtr part_to_export_, StorageMergeTree & storage_);
-};
-
-using CurrentlyExportingPartsTaggerPtr = std::shared_ptr<CurrentlyExportingPartsTagger>;
-
 }

src/Storages/MergeTree/MergeTreeExportManifest.h

@@ -52,6 +52,7 @@ struct MergeTreeExportManifest
         String part_name;
         String remote_path; // empty until uploaded
         bool in_progress = false; /// this is just a hackish workaround for now
+        DataPartPtr part; // hold reference to part so it does not get deleted from disk even if it is outdated. Should be null once we are done with it
     };
 
 
@@ -82,12 +83,15 @@ struct MergeTreeExportManifest
         manifest->file_path = std::filesystem::path(path_prefix) / ("export_partition_" + partition_id_ + "_transaction_" + transaction_id_ + ".json");
         manifest->items.reserve(data_parts.size());
         for (const auto & data_part : data_parts)
-            manifest->items.push_back({data_part->name, ""});
+            manifest->items.emplace_back(data_part->name, "", false, data_part);
         manifest->write();
         return manifest;
     }
 
-    static std::shared_ptr<MergeTreeExportManifest> read(const DiskPtr & disk_, const String & file_path_)
+    /// will not fill parts ref, maybe I should.
+    static std::shared_ptr<MergeTreeExportManifest> read(
+        const DiskPtr & disk_,
+        const String & file_path_)
     {
         auto manifest = std::make_shared<MergeTreeExportManifest>();
         manifest->disk = disk_;

src/Storages/ObjectStorage/MergeTree/ExportPartPlainMergeTreeTask.cpp

@@ -101,23 +101,18 @@ bool ExportPartPlainMergeTreeTask::executeStep()
             manifest->status = MergeTreeExportManifest::Status::failed;
             manifest->write();
 
-            /// this is a mess, what if several fail? I need to re-think the architecture
-            /// I'll leave this commented out for now
-            // storage.already_exported_partition_ids.erase(manifest->partition_id);
-
-            storage.currently_merging_mutating_parts.erase(part_to_export);
+            /// doesn't sound ideal, but it is actually ok to allow this partition to be re-exported as soon as a single part fails
+            /// this is because the ongoing export will never commit, so it won't cause duplicates
+            storage.already_exported_partition_ids.erase(manifest->partition_id);
 
             return false;
         }
         case State::SUCCESS:
         {
-            storage.currently_merging_mutating_parts.erase(part_to_export);
-
             return false;
         }
     }
 
-    storage.currently_merging_mutating_parts.erase(part_to_export);
     return false;
 }
 

src/Storages/StorageMergeTree.cpp

@@ -47,6 +47,7 @@
 #include <Common/ProfileEventsScope.h>
 #include <Common/escapeForFileName.h>
 #include "Core/BackgroundSchedulePool.h"
+#include "Storages/MergeTree/MergeTreeDataPartState.h"
 #include "Storages/ObjectStorage/MergeTree/ExportPartPlainMergeTreeTask.h"
 #include <Core/Names.h>
 #include <Storages/ObjectStorage/StorageObjectStorage.h>
@@ -549,20 +550,12 @@ void StorageMergeTree::exportPartitionToTable(const PartitionCommand & command,
     }
 
     {
-        /// Do not put this in a scope because `CurrentlyExportingPartsTagger` instantiated above relies on this already being locked
-        /// shitty design I came up with huh
         std::lock_guard lock_background_mutex(currently_processing_in_background_mutex);
 
         if (!already_exported_partition_ids.emplace(partition_id).second)
         {
             throw Exception(ErrorCodes::PART_IS_LOCKED, "Partition {} has already been exported", partition_id);
         }
-
-        for (const auto & part : all_parts)
-        {
-            if (!currently_merging_mutating_parts.emplace(part).second)
-                throw Exception(ErrorCodes::LOGICAL_ERROR, "Tagging already tagged part {}. This is a bug.", part->name);
-        }
     }
 
     const auto transaction_id = std::to_string(generateSnowflakeID());
@@ -658,26 +651,6 @@ CurrentlyMergingPartsTagger::~CurrentlyMergingPartsTagger()
     storage.currently_processing_in_background_condition.notify_all();
 }
 
-CurrentlyExportingPartsTagger::CurrentlyExportingPartsTagger(DataPartPtr part_to_export_, StorageMergeTree & storage_)
-    : part_to_export(std::move(part_to_export_)), storage(storage_)
-{
-    /// assume it is already locked
-    if (!storage.currently_merging_mutating_parts.emplace(part_to_export).second)
-        throw Exception(ErrorCodes::LOGICAL_ERROR, "Tagging already tagged part {}. This is a bug.", part_to_export->name);
-}
-
-CurrentlyExportingPartsTagger::~CurrentlyExportingPartsTagger()
-{
-    std::lock_guard lock(storage.currently_processing_in_background_mutex);
-
-    if (!storage.currently_merging_mutating_parts.contains(part_to_export))
-        std::terminate();
-    storage.currently_merging_mutating_parts.erase(part_to_export);
-
-    storage.currently_processing_in_background_condition.notify_all();
-}
-
-
 Int64 StorageMergeTree::startMutation(const MutationCommands & commands, ContextPtr query_context)
 {
     /// Choose any disk, because when we load mutations we search them at each disk
@@ -1150,6 +1123,7 @@ void StorageMergeTree::loadMutations()
 
 void StorageMergeTree::readExportPartitionManifests()
 {
+    static const auto states = {MergeTreeDataPartState::Active, MergeTreeDataPartState::Deleting, MergeTreeDataPartState::Outdated, MergeTreeDataPartState::DeleteOnDestroy};
     for (const auto & disk : getDisks())
     {
         for (auto it = disk->iterateDirectory(relative_data_path); it->isValid(); it->next())
@@ -1177,6 +1151,28 @@ void StorageMergeTree::readExportPartitionManifests()
                         }
                     }
 
+                    for (auto & item : manifest->items)
+                    {
+                        /// if this part has not been pushed yet
+                        if (item.remote_path.empty())
+                        {
+                            item.part = getPartIfExists(item.part_name, states);
+
+                            if (!item.part)
+                            {
+                                LOG_ERROR(log, "Part {} is present in the manifest file {}, but not found in the storage {}",
+                                    item.part_name,
+                                    manifest->transaction_id,
+                                    getStorageID().getNameForLogs());
+
+                                manifest->status = MergeTreeExportManifest::Status::failed;
+                                manifest->write();
+                                already_exported_partition_ids.erase(manifest->partition_id);
+                                continue;
+                            }
+                        }
+                    }
+
                     export_partition_transaction_id_to_manifest.emplace(manifest->transaction_id, manifest);
 
                     LOG_DEBUG(log, "Loaded export transaction manifest: {} (transaction_id: {})", name, manifest->transaction_id);
@@ -1670,12 +1666,13 @@ bool StorageMergeTree::scheduleDataMovingJob(BackgroundJobsAssignee & assignee)
                 continue;
             }
 
+            static const auto states = {MergeTreeDataPartState::Active, MergeTreeDataPartState::Deleting, MergeTreeDataPartState::Outdated, MergeTreeDataPartState::DeleteOnDestroy};
             for (auto & item : manifest->items)
             {
                 if (item.in_progress)
                     continue;
 
-                auto part = getPartIfExists(item.part_name, {MergeTreeDataPartState::Active});
+                auto part = getPartIfExists(item.part_name, states);
                 if (!part)
                 {
                     LOG_ERROR(log, "Part {} is present in the manifest file {}, but not found in the storage {}",