improvements

Author: arthurpassos

src/Storages/ExportReplicatedMergeTreePartitionManifest.h

@@ -162,22 +162,14 @@ struct ExportReplicatedMergeTreePartitionManifest
         manifest.max_threads = json->getValue<size_t>("max_threads");
         manifest.parallel_formatting = json->getValue<bool>("parallel_formatting");
         manifest.parquet_parallel_encoding = json->getValue<bool>("parquet_parallel_encoding");
-
-        if (json->has("file_already_exists_policy"))
+        const auto file_already_exists_policy = magic_enum::enum_cast<MergeTreePartExportManifest::FileAlreadyExistsPolicy>(json->getValue<String>("file_already_exists_policy"));
+        /// todo what to do if it's not a valid value?
+        if (file_already_exists_policy)
         {
-            const auto file_already_exists_policy = magic_enum::enum_cast<MergeTreePartExportManifest::FileAlreadyExistsPolicy>(json->getValue<String>("file_already_exists_policy"));
-            if (file_already_exists_policy)
-            {
-                manifest.file_already_exists_policy = file_already_exists_policy.value();
-            }
-
-            /// what to do if it's not a valid value?
+            manifest.file_already_exists_policy = file_already_exists_policy.value();
         }
 
-        if (json->has("lock_inside_the_task"))
-        {
-            manifest.lock_inside_the_task = json->getValue<bool>("lock_inside_the_task");
-        }
+        manifest.lock_inside_the_task = json->getValue<bool>("lock_inside_the_task");
 
         return manifest;
     }

src/Storages/MergeTree/ExportPartFromPartitionExportTask.h

@@ -8,6 +8,9 @@
 namespace DB
 {
 
+/*
+    Decorator around the ExportPartTask to lock the part inside the task
+*/
 class ExportPartFromPartitionExportTask : public IExecutableTask
 {
 public:

src/Storages/MergeTree/ExportPartitionTaskScheduler.cpp

@@ -131,7 +131,7 @@ void ExportPartitionTaskScheduler::run()
         if (status_in_zk.value() != ExportReplicatedMergeTreePartitionTaskEntry::Status::PENDING)
         {
             entry.status = status_in_zk.value();
-            LOG_INFO(storage.log, "ExportPartition scheduler task: Skipping... Status from zk is {}", entry.status);
+            LOG_INFO(storage.log, "ExportPartition scheduler task: Skipping {}... Status from zk is {}", key, magic_enum::enum_name(entry.status).data());
             continue;
         }
 

src/Storages/MergeTree/MergeTreePartExportManifest.h

@@ -73,6 +73,7 @@ struct MergeTreePartExportManifest
     std::function<void(CompletionCallbackResult)> completion_callback;
 
     time_t create_time;
+    /// Required to cancel export tasks
     mutable std::shared_ptr<IExecutableTask> task = nullptr;
 
     bool operator<(const MergeTreePartExportManifest & rhs) const 

tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py

@@ -245,6 +245,9 @@ def test_kill_export(cluster):
         # ZooKeeper operations (KILL) proceed quickly since only S3 is blocked
         node.query(f"KILL EXPORT PARTITION WHERE partition_id = '2020' and source_table = '{mt_table}' and destination_table = '{s3_table}'")
 
+        # sleep for a while to let the kill to be processed
+        time.sleep(2)
+
     # wait for 2021 to finish
     wait_for_export_status(node, mt_table, s3_table, "2021", "COMPLETED")