cleanup export manifests at shutdown

Author: arthurpassos

src/Storages/MergeTree/MergeTreeData.cpp

@@ -6283,14 +6283,7 @@ void MergeTreeData::killExportPart(const String & transaction_id)
 
     std::erase_if(export_manifests, [&](const auto & manifest)
     {
-        if (manifest.transaction_id == transaction_id)
-        {
-            if (manifest.task)
-                manifest.task->cancel();
-
-            return true;
-        }
-        return false;
+        return manifest.transaction_id == transaction_id;
     });
 }
 
@@ -9137,17 +9130,13 @@ bool MergeTreeData::scheduleDataMovingJob(BackgroundJobsAssignee & assignee)
         context_copy->setCurrentQueryId(manifest.transaction_id);
         context_copy->setBackgroundOperationTypeForContext(ClientInfo::BackgroundOperationType::EXPORT_PART);
 
-        auto task = std::make_shared<ExportPartTask>(*this, manifest, context_copy);
-
-        manifest.in_progress = assignee.scheduleMoveTask(task);
+        manifest.in_progress = assignee.scheduleMoveTask(std::make_shared<ExportPartTask>(*this, manifest, context_copy));
 
         if (!manifest.in_progress)
         {
             continue;
         }
 
-        manifest.task = task;
-
         return true;
     }
 

src/Storages/MergeTree/MergeTreePartExportManifest.h

@@ -73,7 +73,6 @@ struct MergeTreePartExportManifest
 
     time_t create_time;
     mutable bool in_progress = false;
-    mutable std::shared_ptr<ExportPartTask> task = nullptr;
 
     bool operator<(const MergeTreePartExportManifest & rhs) const 
     {

src/Storages/StorageMergeTree.cpp

@@ -256,6 +256,11 @@ void StorageMergeTree::shutdown(bool)
 
     if (deduplication_log)
         deduplication_log->shutdown();
+
+    {
+        std::lock_guard lock(export_manifests_mutex);
+        export_manifests.clear();
+    }
 }
 
 

src/Storages/StorageReplicatedMergeTree.cpp

@@ -6014,6 +6014,17 @@ void StorageReplicatedMergeTree::shutdown(bool)
         /// Wait for all of them
         std::lock_guard lock(data_parts_exchange_ptr->rwlock);
     }
+
+    {
+        std::lock_guard lock(export_manifests_mutex);
+        export_manifests.clear();
+    }
+
+    {
+        std::lock_guard lock(export_merge_tree_partition_mutex);
+        export_merge_tree_partition_task_entries.clear();
+    }
+
     LOG_TRACE(log, "Shutdown finished");
 }