update partition export

arthurpassos · arthurpassos · commit c6eee2cd952e · 2025-12-15T19:50:27.000-03:00
diff --git a/src/Storages/ExportReplicatedMergeTreePartitionManifest.h b/src/Storages/ExportReplicatedMergeTreePartitionManifest.h
@@ -62,14 +62,14 @@ struct ExportReplicatedMergeTreePartitionProcessingPartEntry
 struct ExportReplicatedMergeTreePartitionProcessedPartEntry
 {
     String part_name;
-    String path_in_destination;
+    std::vector<String> paths_in_destination;
     String finished_by;
 
     std::string toJsonString() const
     {
         Poco::JSON::Object json;
         json.set("part_name", part_name);
-        json.set("path_in_destination", path_in_destination);
+        json.set("paths_in_destination", paths_in_destination);
         json.set("finished_by", finished_by);
         std::ostringstream oss;     // STYLE_CHECK_ALLOW_STD_STRING_STREAM
         oss.exceptions(std::ios::failbit);
@@ -86,7 +86,11 @@ struct ExportReplicatedMergeTreePartitionProcessedPartEntry
         ExportReplicatedMergeTreePartitionProcessedPartEntry entry;
 
         entry.part_name = json->getValue<String>("part_name");
-        entry.path_in_destination = json->getValue<String>("path_in_destination");
+
+        const auto paths_in_destination_array = json->getArray("paths_in_destination");
+        for (size_t i = 0; i < paths_in_destination_array->size(); ++i)
+            entry.paths_in_destination.emplace_back(paths_in_destination_array->getElement<String>(static_cast<unsigned int>(i)));
+
         entry.finished_by = json->getValue<String>("finished_by");
 
         return entry;
diff --git a/src/Storages/MergeTree/ExportPartTask.cpp b/src/Storages/MergeTree/ExportPartTask.cpp
@@ -196,7 +196,7 @@ bool ExportPartTask::executeStep()
         ProfileEvents::increment(ProfileEvents::PartsExportTotalMilliseconds, (*exports_list_entry)->watch.elapsedMilliseconds());
 
         if (manifest.completion_callback)
-            manifest.completion_callback(MergeTreePartExportManifest::CompletionCallbackResult::createSuccess((*exports_list_entry)->destination_file_paths.front()));
+            manifest.completion_callback(MergeTreePartExportManifest::CompletionCallbackResult::createSuccess((*exports_list_entry)->destination_file_paths));
     }
     catch (const Exception & e)
     {
@@ -226,12 +226,7 @@ bool ExportPartTask::executeStep()
 
                 if (manifest.completion_callback)
                 {
-                    if ((*exports_list_entry)->destination_file_paths.empty())
-                    {
-                        throw Exception(ErrorCodes::LOGICAL_ERROR, "No destination file paths found for part {}", manifest.data_part->name);
-                    }
-
-                    manifest.completion_callback(MergeTreePartExportManifest::CompletionCallbackResult::createSuccess((*exports_list_entry)->destination_file_paths.front()));
+                    manifest.completion_callback(MergeTreePartExportManifest::CompletionCallbackResult::createSuccess((*exports_list_entry)->destination_file_paths));
                 }
                     
                 return false;
diff --git a/src/Storages/MergeTree/ExportPartitionTaskScheduler.cpp b/src/Storages/MergeTree/ExportPartitionTaskScheduler.cpp
@@ -175,7 +175,7 @@ void ExportPartitionTaskScheduler::handlePartExportCompletion(
 
     if (result.success)
     {
-        handlePartExportSuccess(manifest, destination_storage, processing_parts_path, processed_part_path, part_name, export_path, zk, result.relative_path_in_destination_storage);
+        handlePartExportSuccess(manifest, destination_storage, processing_parts_path, processed_part_path, part_name, export_path, zk, result.relative_paths_in_destination_storage);
     }
     else
     {
@@ -191,12 +191,17 @@ void ExportPartitionTaskScheduler::handlePartExportSuccess(
     const std::string & part_name,
     const std::filesystem::path & export_path,
     const zkutil::ZooKeeperPtr & zk,
-    const String & relative_path_in_destination_storage
+    const std::vector<String> & relative_paths_in_destination_storage
 )
 {
-    LOG_INFO(storage.log, "ExportPartition scheduler task: Part {} exported successfully", relative_path_in_destination_storage);
+    LOG_INFO(storage.log, "ExportPartition scheduler task: Part {} exported successfully, paths size: {}", part_name, relative_paths_in_destination_storage.size());
 
-    if (!tryToMovePartToProcessed(export_path, processing_parts_path, processed_part_path, part_name, relative_path_in_destination_storage, zk))
+    for (const auto & relative_path_in_destination_storage : relative_paths_in_destination_storage)
+    {
+        LOG_INFO(storage.log, "ExportPartition scheduler task: {}", relative_path_in_destination_storage);
+    }
+
+    if (!tryToMovePartToProcessed(export_path, processing_parts_path, processed_part_path, part_name, relative_paths_in_destination_storage, zk))
     {
         LOG_INFO(storage.log, "ExportPartition scheduler task: Failed to move part to processed, will not commit export partition");
         return;
@@ -323,7 +328,7 @@ bool ExportPartitionTaskScheduler::tryToMovePartToProcessed(
     const std::filesystem::path & processing_parts_path,
     const std::filesystem::path & processed_part_path,
     const std::string & part_name,
-    const String & relative_path_in_destination_storage,
+    const std::vector<String> & relative_paths_in_destination_storage,
     const zkutil::ZooKeeperPtr & zk
 )
 {
@@ -348,7 +353,7 @@ bool ExportPartitionTaskScheduler::tryToMovePartToProcessed(
 
     ExportReplicatedMergeTreePartitionProcessedPartEntry processed_part_entry;
     processed_part_entry.part_name = part_name;
-    processed_part_entry.path_in_destination = relative_path_in_destination_storage;
+    processed_part_entry.paths_in_destination = relative_paths_in_destination_storage;
     processed_part_entry.finished_by = storage.replica_name;
 
     requests.emplace_back(zkutil::makeRemoveRequest(processing_parts_path / part_name, -1));
diff --git a/src/Storages/MergeTree/ExportPartitionTaskScheduler.h b/src/Storages/MergeTree/ExportPartitionTaskScheduler.h
@@ -37,7 +37,7 @@ class ExportPartitionTaskScheduler
         const std::string & part_name,
         const std::filesystem::path & export_path,
         const zkutil::ZooKeeperPtr & zk,
-        const String & relative_path_in_destination_storage
+        const std::vector<String> & relative_paths_in_destination_storage
     );
 
     void handlePartExportFailure(
@@ -53,7 +53,7 @@ class ExportPartitionTaskScheduler
         const std::filesystem::path & processing_parts_path,
         const std::filesystem::path & processed_part_path,
         const std::string & part_name,
-        const String & relative_path_in_destination_storage,
+        const std::vector<String> & relative_paths_in_destination_storage,
         const zkutil::ZooKeeperPtr & zk
     );
 
diff --git a/src/Storages/MergeTree/ExportPartitionUtils.cpp b/src/Storages/MergeTree/ExportPartitionUtils.cpp
@@ -55,7 +55,10 @@ namespace ExportPartitionUtils
 
             const auto processed_part_entry = ExportReplicatedMergeTreePartitionProcessedPartEntry::fromJsonString(responses[i].data);
 
-            exported_paths.emplace_back(processed_part_entry.path_in_destination);
+            for (const auto & path_in_destination : processed_part_entry.paths_in_destination)
+            {
+                exported_paths.emplace_back(path_in_destination);
+            }
         }
 
         return exported_paths;
diff --git a/src/Storages/MergeTree/MergeTreePartExportManifest.h b/src/Storages/MergeTree/MergeTreePartExportManifest.h
@@ -23,22 +23,22 @@ struct MergeTreePartExportManifest
     struct CompletionCallbackResult
     {
     private:
-        CompletionCallbackResult(bool success_, const String & relative_path_in_destination_storage_, std::optional<Exception> exception_)
-            : success(success_), relative_path_in_destination_storage(relative_path_in_destination_storage_), exception(std::move(exception_)) {}
+        CompletionCallbackResult(bool success_, const std::vector<String> & relative_paths_in_destination_storage_, std::optional<Exception> exception_)
+            : success(success_), relative_paths_in_destination_storage(relative_paths_in_destination_storage_), exception(std::move(exception_)) {}
     public:
 
-        static CompletionCallbackResult createSuccess(const String & relative_path_in_destination_storage_)
+        static CompletionCallbackResult createSuccess(const std::vector<String> & relative_paths_in_destination_storage_)
         {
-            return CompletionCallbackResult(true, relative_path_in_destination_storage_, std::nullopt);
+            return CompletionCallbackResult(true, relative_paths_in_destination_storage_, std::nullopt);
         }
 
         static CompletionCallbackResult createFailure(Exception exception_)
         {
-            return CompletionCallbackResult(false, "", std::move(exception_));
+            return CompletionCallbackResult(false, {}, std::move(exception_));
         }
 
         bool success = false;
-        String relative_path_in_destination_storage;
+        std::vector<String> relative_paths_in_destination_storage;
         std::optional<Exception> exception;
     };
 

Original file line number	Diff line number	Diff line change
`@@ -196,7 +196,7 @@ bool ExportPartTask::executeStep()`
`196`	`196`	`ProfileEvents::increment(ProfileEvents::PartsExportTotalMilliseconds, (*exports_list_entry)->watch.elapsedMilliseconds());`
`197`	`197`
`198`	`198`	`if (manifest.completion_callback)`
`199`		`- manifest.completion_callback(MergeTreePartExportManifest::CompletionCallbackResult::createSuccess((*exports_list_entry)->destination_file_paths.front()));`
	`199`	`+ manifest.completion_callback(MergeTreePartExportManifest::CompletionCallbackResult::createSuccess((*exports_list_entry)->destination_file_paths));`
`200`	`200`	`}`
`201`	`201`	`catch (const Exception & e)`
`202`	`202`	`{`
`@@ -226,12 +226,7 @@ bool ExportPartTask::executeStep()`
`226`	`226`
`227`	`227`	`if (manifest.completion_callback)`
`228`	`228`	`{`
`229`		`- if ((*exports_list_entry)->destination_file_paths.empty())`
`230`		`- {`
`231`		`- throw Exception(ErrorCodes::LOGICAL_ERROR, "No destination file paths found for part {}", manifest.data_part->name);`
`232`		`- }`
`233`		`-`
`234`		`- manifest.completion_callback(MergeTreePartExportManifest::CompletionCallbackResult::createSuccess((*exports_list_entry)->destination_file_paths.front()));`
	`229`	`+ manifest.completion_callback(MergeTreePartExportManifest::CompletionCallbackResult::createSuccess((*exports_list_entry)->destination_file_paths));`
`235`	`230`	`}`
`236`	`231`
`237`	`232`	`return false;`
Original file line number	Diff line number	Diff line change
`@@ -175,7 +175,7 @@ void ExportPartitionTaskScheduler::handlePartExportCompletion(`
`175`	`175`
`176`	`176`	`if (result.success)`
`177`	`177`	`{`
`178`		`- handlePartExportSuccess(manifest, destination_storage, processing_parts_path, processed_part_path, part_name, export_path, zk, result.relative_path_in_destination_storage);`
	`178`	`+ handlePartExportSuccess(manifest, destination_storage, processing_parts_path, processed_part_path, part_name, export_path, zk, result.relative_paths_in_destination_storage);`
`179`	`179`	`}`
`180`	`180`	`else`
`181`	`181`	`{`
`@@ -191,12 +191,17 @@ void ExportPartitionTaskScheduler::handlePartExportSuccess(`
`191`	`191`	`const std::string & part_name,`
`192`	`192`	`const std::filesystem::path & export_path,`
`193`	`193`	`const zkutil::ZooKeeperPtr & zk,`
`194`		`- const String & relative_path_in_destination_storage`
	`194`	`+ const std::vector<String> & relative_paths_in_destination_storage`
`195`	`195`	`)`
`196`	`196`	`{`
`197`		`- LOG_INFO(storage.log, "ExportPartition scheduler task: Part {} exported successfully", relative_path_in_destination_storage);`
	`197`	`+ LOG_INFO(storage.log, "ExportPartition scheduler task: Part {} exported successfully, paths size: {}", part_name, relative_paths_in_destination_storage.size());`
`198`	`198`
`199`		`- if (!tryToMovePartToProcessed(export_path, processing_parts_path, processed_part_path, part_name, relative_path_in_destination_storage, zk))`
	`199`	`+ for (const auto & relative_path_in_destination_storage : relative_paths_in_destination_storage)`
	`200`	`+ {`
	`201`	`+ LOG_INFO(storage.log, "ExportPartition scheduler task: {}", relative_path_in_destination_storage);`
	`202`	`+ }`
	`203`	`+`
	`204`	`+ if (!tryToMovePartToProcessed(export_path, processing_parts_path, processed_part_path, part_name, relative_paths_in_destination_storage, zk))`
`200`	`205`	`{`
`201`	`206`	`LOG_INFO(storage.log, "ExportPartition scheduler task: Failed to move part to processed, will not commit export partition");`
`202`	`207`	`return;`
`@@ -323,7 +328,7 @@ bool ExportPartitionTaskScheduler::tryToMovePartToProcessed(`
`323`	`328`	`const std::filesystem::path & processing_parts_path,`
`324`	`329`	`const std::filesystem::path & processed_part_path,`
`325`	`330`	`const std::string & part_name,`
`326`		`- const String & relative_path_in_destination_storage,`
	`331`	`+ const std::vector<String> & relative_paths_in_destination_storage,`
`327`	`332`	`const zkutil::ZooKeeperPtr & zk`
`328`	`333`	`)`
`329`	`334`	`{`
`@@ -348,7 +353,7 @@ bool ExportPartitionTaskScheduler::tryToMovePartToProcessed(`
`348`	`353`
`349`	`354`	`ExportReplicatedMergeTreePartitionProcessedPartEntry processed_part_entry;`
`350`	`355`	`processed_part_entry.part_name = part_name;`
`351`		`- processed_part_entry.path_in_destination = relative_path_in_destination_storage;`
	`356`	`+ processed_part_entry.paths_in_destination = relative_paths_in_destination_storage;`
`352`	`357`	`processed_part_entry.finished_by = storage.replica_name;`
`353`	`358`
`354`	`359`	`requests.emplace_back(zkutil::makeRemoveRequest(processing_parts_path / part_name, -1));`
Original file line number	Diff line number	Diff line change
`@@ -55,7 +55,10 @@ namespace ExportPartitionUtils`
`55`	`55`
`56`	`56`	`const auto processed_part_entry = ExportReplicatedMergeTreePartitionProcessedPartEntry::fromJsonString(responses[i].data);`
`57`	`57`
`58`		`- exported_paths.emplace_back(processed_part_entry.path_in_destination);`
	`58`	`+ for (const auto & path_in_destination : processed_part_entry.paths_in_destination)`
	`59`	`+ {`
	`60`	`+ exported_paths.emplace_back(path_in_destination);`
	`61`	`+ }`
`59`	`62`	`}`
`60`	`63`
`61`	`64`	`return exported_paths;`
Original file line number	Diff line number	Diff line change
`@@ -23,22 +23,22 @@ struct MergeTreePartExportManifest`
`23`	`23`	`struct CompletionCallbackResult`
`24`	`24`	`{`
`25`	`25`	`private:`
`26`		`- CompletionCallbackResult(bool success_, const String & relative_path_in_destination_storage_, std::optional<Exception> exception_)`
`27`		`- : success(success_), relative_path_in_destination_storage(relative_path_in_destination_storage_), exception(std::move(exception_)) {}`
	`26`	`+ CompletionCallbackResult(bool success_, const std::vector<String> & relative_paths_in_destination_storage_, std::optional<Exception> exception_)`
	`27`	`+ : success(success_), relative_paths_in_destination_storage(relative_paths_in_destination_storage_), exception(std::move(exception_)) {}`
`28`	`28`	`public:`
`29`	`29`
`30`		`- static CompletionCallbackResult createSuccess(const String & relative_path_in_destination_storage_)`
	`30`	`+ static CompletionCallbackResult createSuccess(const std::vector<String> & relative_paths_in_destination_storage_)`
`31`	`31`	`{`
`32`		`- return CompletionCallbackResult(true, relative_path_in_destination_storage_, std::nullopt);`
	`32`	`+ return CompletionCallbackResult(true, relative_paths_in_destination_storage_, std::nullopt);`
`33`	`33`	`}`
`34`	`34`
`35`	`35`	`static CompletionCallbackResult createFailure(Exception exception_)`
`36`	`36`	`{`
`37`		`- return CompletionCallbackResult(false, "", std::move(exception_));`
	`37`	`+ return CompletionCallbackResult(false, {}, std::move(exception_));`
`38`	`38`	`}`
`39`	`39`
`40`	`40`	`bool success = false;`
`41`		`- String relative_path_in_destination_storage;`
	`41`	`+ std::vector<String> relative_paths_in_destination_storage;`
`42`	`42`	`std::optional<Exception> exception;`
`43`	`43`	`};`
`44`	`44`