improve system.exports, show failed exports

arthurpassos · arthurpassos · commit 502b5018d5b3 · 2025-09-03T13:29:20.000-03:00
diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp
@@ -6877,9 +6877,9 @@ Experimental timeSeries* aggregate functions for Prometheus-like timeseries resa
     DECLARE_WITH_ALIAS(Bool, allow_experimental_export_merge_tree_partition, false, R"(
 Experimental export merge tree partition.
 )", EXPERIMENTAL, allow_experimental_export_merge_tree_partition) \
-    DECLARE_WITH_ALIAS(Bool, export_merge_tree_partition_individual_part_executor, true, R"(
+    DECLARE_WITH_ALIAS(Bool, export_merge_tree_partition_executor, false, R"(
     Use the part task instead of the partition task
-)", EXPERIMENTAL, export_merge_tree_partition_background_execution) \
+)", EXPERIMENTAL, export_merge_tree_partition_executor) \
     \
     /* ####################################################### */ \
     /* ############ END OF EXPERIMENTAL FEATURES ############# */ \
diff --git a/src/Storages/MergeTree/MergeTreeData.h b/src/Storages/MergeTree/MergeTreeData.h
@@ -18,6 +18,7 @@
 #include <Storages/MergeTree/MergeTreeIndices.h>
 #include <Storages/MergeTree/MergeTreePartInfo.h>
 #include <Storages/MergeTree/MergeTreeMutationStatus.h>
+#include <Storages/MergeTree/MergeTreeExportStatus.h>
 #include <Storages/MergeTree/MergeList.h>
 #include <Storages/MergeTree/IMergeTreeDataPart.h>
 #include <Storages/MergeTree/MergeTreeDataPartBuilder.h>
@@ -966,6 +967,7 @@ class MergeTreeData : public IStorage, public WithMutableContext
         bool must_on_same_disk);
 
     virtual std::vector<MergeTreeMutationStatus> getMutationsStatus() const = 0;
+    virtual std::vector<MergeTreeExportStatus> getExportsStatus() const { return {}; }
 
     /// Returns true if table can create new parts with adaptive granularity
     /// Has additional constraint in replicated version
diff --git a/src/Storages/MergeTree/MergeTreeExportManifest.h b/src/Storages/MergeTree/MergeTreeExportManifest.h
@@ -14,6 +14,8 @@
 #include <Poco/JSON/Stringifier.h>
 #include <Poco/Dynamic/Var.h>
 #include <Storages/MergeTree/IMergeTreeDataPart.h>
+#include <ctime>
+#include <magic_enum.hpp>
 
 namespace DB
 {
@@ -25,29 +27,39 @@ namespace DB
  *   "transaction_id": "<id>",
  *   "partition_id": "<partition_id>",
  *   "destination": "<database>.<table>",
- *   "completed": <true/false>,
+ *   "create_time": <unix_timestamp>,
+ *   "status": "<pending|completed|failed>",
  *   "parts": [ {"part_name": "name", "remote_path": "path-or-empty"}, ... ]
  * }
  */
 struct MergeTreeExportManifest
 {
     using DataPartPtr = std::shared_ptr<const IMergeTreeDataPart>;
+    
+    enum class Status {
+        pending,
+        completed, 
+        failed
+    };
 
     MergeTreeExportManifest()
         : destination_storage_id(StorageID::createEmpty())
+        , status(Status::pending)
     {}
 
     struct Item
     {
         String part_name;
         String remote_path; // empty until uploaded
     };
+    
 
     String transaction_id;
     String partition_id;
     StorageID destination_storage_id;
+    time_t create_time = 0;
     std::vector<Item> items;
-    bool completed = false;
+    Status status = Status::pending;
 
     std::filesystem::path file_path;
     DiskPtr disk;
@@ -65,6 +77,7 @@ struct MergeTreeExportManifest
         manifest->transaction_id = transaction_id_;
         manifest->partition_id = partition_id_;
         manifest->destination_storage_id = destination_storage_id_;
+        manifest->create_time = std::time(nullptr);
         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)
@@ -94,7 +107,11 @@ struct MergeTreeExportManifest
         manifest->partition_id = root->getValue<String>("partition_id");
         const auto destination = root->getValue<String>("destination");
         manifest->destination_storage_id = StorageID(QualifiedTableName::parseFromString(destination));
-        manifest->completed = root->getValue<bool>("completed");
+        
+        manifest->create_time = root->getValue<UInt64>("create_time");
+        
+        String status_str = root->getValue<String>("status");
+        manifest->status = magic_enum::enum_cast<Status>(status_str).value();
 
         manifest->items.clear();
         auto parts = root->get("parts").extract<Poco::JSON::Array::Ptr>();
@@ -118,7 +135,8 @@ struct MergeTreeExportManifest
         root->set("transaction_id", transaction_id);
         root->set("partition_id", partition_id);
         root->set("destination", destination_storage_id.getQualifiedName().getFullName());
-        root->set("completed", completed);
+        root->set("create_time", static_cast<UInt64>(create_time));
+        root->set("status", String(magic_enum::enum_name(status)));
 
         Poco::JSON::Array::Ptr parts(new Poco::JSON::Array());
         for (const auto & i : items)
@@ -138,7 +156,7 @@ struct MergeTreeExportManifest
         out->sync();
     }
 
-    void deleteFile()
+    void deleteFile() const
     {
         disk->removeFile(file_path);
     }
diff --git a/src/Storages/MergeTree/MergeTreeExportStatus.cpp b/src/Storages/MergeTree/MergeTreeExportStatus.cpp
diff --git a/src/Storages/MergeTree/MergeTreeExportStatus.h b/src/Storages/MergeTree/MergeTreeExportStatus.h
@@ -0,0 +1,25 @@
+#pragma once
+
+#include <base/types.h>
+#include <vector>
+#include <ctime>
+#include <Storages/MergeTree/MergeTreeExportManifest.h>
+
+
+namespace DB
+{
+
+struct MergeTreeExportStatus
+{
+    String source_database;
+    String source_table;
+    String destination_database;
+    String destination_table;
+    String transaction_id;
+    time_t create_time = 0;
+    std::vector<String> parts_to_do_names;
+    MergeTreeExportManifest::Status status;
+};
+
+}
+
diff --git a/src/Storages/ObjectStorage/MergeTree/ExportPartPlainMergeTreeTask.cpp b/src/Storages/ObjectStorage/MergeTree/ExportPartPlainMergeTreeTask.cpp
@@ -57,40 +57,51 @@ bool ExportPartPlainMergeTreeTask::executeStep()
             if (executeExport())
             {
                 state = State::NEED_COMMIT;
-                return true;
             }
-
-            if (retry_count < max_retries)
+            else if (retry_count < max_retries)
             {
                 retry_count++;
                 LOG_INFO(getLogger("ExportPartPlainMergeTreeTask"),
                 "Retrying export attempt {} for part {}",
                 retry_count, exports_tagger->parts_to_export[0]->name);
                 state = State::NEED_EXECUTE;
-
-                return true;
+            }
+            else
+            {
+                state = State::FAILED;
             }
 
-            return false;
+            return true;
         }
         case State::NEED_COMMIT:
         {
             if (commitExport())
             {
                 state = State::SUCCESS;
-                return true;
             }
-
-            if (retry_count < max_retries)
+            else if (retry_count < max_retries)
             {
                 retry_count++;
                 LOG_INFO(getLogger("ExportPartPlainMergeTreeTask"),
                 "Retrying export attempt {} for part {}",
                 retry_count, exports_tagger->parts_to_export[0]->name);
                 state = State::NEED_COMMIT;
-
-                return true;
             }
+            else
+            {
+                state = State::FAILED;
+            }
+
+            return true;
+        }
+        case State::FAILED:
+        {
+            std::lock_guard lock(storage.export_partition_transaction_id_to_manifest_mutex);
+
+            manifest->status = MergeTreeExportManifest::Status::failed;
+            manifest->write();
+
+            storage.already_exported_partition_ids.erase(manifest->partition_id);
 
             return false;
         }
@@ -171,7 +182,7 @@ bool ExportPartPlainMergeTreeTask::commitExport()
             manifest->partition_id,
             manifest->exportedPaths(),
             context);
-        manifest->completed = true;
+        manifest->status = MergeTreeExportManifest::Status::completed;
         manifest->write();
         storage.export_partition_transaction_id_to_manifest.erase(manifest->transaction_id);
         LOG_INFO(getLogger("ExportMergeTreePartitionToObjectStorageTask"),
diff --git a/src/Storages/ObjectStorage/MergeTree/ExportPartPlainMergeTreeTask.h b/src/Storages/ObjectStorage/MergeTree/ExportPartPlainMergeTreeTask.h
@@ -43,6 +43,7 @@ class ExportPartPlainMergeTreeTask : public IExecutableTask
         NEED_PREPARE,
         NEED_EXECUTE,
         NEED_COMMIT,
+        FAILED,
         SUCCESS
     };
 
diff --git a/src/Storages/ObjectStorage/MergeTree/ExportPartitionPlainMergeTreeTask.cpp b/src/Storages/ObjectStorage/MergeTree/ExportPartitionPlainMergeTreeTask.cpp
@@ -189,7 +189,7 @@ bool ExportPartitionPlainMergeTreeTask::commitExport()
         manifest->exportedPaths(),
         context);
     
-    manifest->completed = true;
+    manifest->status = MergeTreeExportManifest::Status::completed;
     manifest->write();
 
     storage.export_partition_transaction_id_to_manifest.erase(manifest->transaction_id);
@@ -204,7 +204,7 @@ bool ExportPartitionPlainMergeTreeTask::commitExport()
 void ExportPartitionPlainMergeTreeTask::onCompleted()
 {
     std::lock_guard lock(storage.export_partition_transaction_id_to_manifest_mutex);
-    task_result_callback(manifest->completed);
+    task_result_callback(manifest->status == MergeTreeExportManifest::Status::completed);
 }
 
 void ExportPartitionPlainMergeTreeTask::cancel() noexcept
diff --git a/src/Storages/StorageMergeTree.cpp b/src/Storages/StorageMergeTree.cpp
@@ -82,7 +82,7 @@ namespace Setting
     extern const SettingsBool throw_on_unsupported_query_inside_transaction;
     extern const SettingsUInt64 max_parts_to_move;
     extern const SettingsBool allow_experimental_export_merge_tree_partition;
-    extern const SettingsBool export_merge_tree_partition_individual_part_executor;
+    extern const SettingsBool export_merge_tree_partition_executor;
 }
 
 namespace MergeTreeSetting
@@ -559,8 +559,6 @@ void StorageMergeTree::exportPartitionToTable(const PartitionCommand & command,
         throw Exception(ErrorCodes::PART_IS_LOCKED, "Partition {} has already been exported", partition_id);
     }
 
-    
-
     const auto transaction_id = std::to_string(generateSnowflakeID());
 
     /// TODO missing parts lock here with tagger
@@ -578,7 +576,7 @@ void StorageMergeTree::exportPartitionToTable(const PartitionCommand & command,
         export_partition_transaction_id_to_manifest[transaction_id] = manifest;
     }
 
-    if (getContext()->getSettingsRef()[Setting::export_merge_tree_partition_individual_part_executor])
+    if (!getContext()->getSettingsRef()[Setting::export_merge_tree_partition_executor])
     {
         for (const auto & part : all_parts)
         {
@@ -981,6 +979,33 @@ std::map<std::string, MutationCommands> StorageMergeTree::getUnfinishedMutationC
     return result;
 }
 
+
+std::vector<MergeTreeExportStatus> StorageMergeTree::getExportsStatus() const
+{
+    std::lock_guard lock(export_partition_transaction_id_to_manifest_mutex);
+    std::vector<MergeTreeExportStatus> result;
+
+    auto source_database = getStorageID().database_name;
+    auto source_table = getStorageID().table_name;
+
+    for (const auto & [transaction_id, manifest] : export_partition_transaction_id_to_manifest)
+    {
+        MergeTreeExportStatus status;
+
+        status.transaction_id = transaction_id;
+        status.source_database = source_database;
+        status.source_table = source_table;
+        status.destination_database = manifest->destination_storage_id.database_name;
+        status.destination_table = manifest->destination_storage_id.table_name;
+        status.create_time = 0;
+        status.parts_to_do_names = manifest->pendingParts();
+        status.status = manifest->status;
+
+        result.emplace_back(std::move(status));
+    }
+    return result;
+}
+
 std::vector<MergeTreeMutationStatus> StorageMergeTree::getMutationsStatus() const
 {
     std::lock_guard lock(currently_processing_in_background_mutex);
@@ -1157,17 +1182,20 @@ void StorageMergeTree::readExportPartitionManifests()
                 {
                     auto manifest = MergeTreeExportManifest::read(disk, fs::path(relative_data_path) / name);
 
-                    already_exported_partition_ids.insert(manifest->partition_id);
-
-                    if (manifest->completed)
+                    if (manifest->status != MergeTreeExportManifest::Status::failed)
                     {
-                        LOG_INFO(
-                            log,
-                            "Export transaction {} of partition {} to destination storage {} already completed, skipping",
-                            manifest->transaction_id,
-                            manifest->partition_id,
-                            manifest->destination_storage_id.getNameForLogs());
-                        continue;
+                        already_exported_partition_ids.insert(manifest->partition_id);
+
+                        if (manifest->status == MergeTreeExportManifest::Status::completed)
+                        {
+                            LOG_INFO(
+                                log,
+                                "Export transaction {} of partition {} to destination storage {} already completed, skipping",
+                                manifest->transaction_id,
+                                manifest->partition_id,
+                                manifest->destination_storage_id.getNameForLogs());
+                            continue;
+                        }
                     }
 
                     export_partition_transaction_id_to_manifest.emplace(manifest->transaction_id, manifest);
@@ -1189,6 +1217,9 @@ void StorageMergeTree::resumeExportPartitionTasks()
     /// but it turns out the background executor schedules tasks based on their priority, so it is likely this is not needed anymore.
     for (const auto & [transaction_id, manifest] : export_partition_transaction_id_to_manifest)
     {
+        if (manifest->status != MergeTreeExportManifest::Status::pending)
+            continue;
+
         auto destination_storage = DatabaseCatalog::instance().tryGetTable(manifest->destination_storage_id, getContext());
         if (!destination_storage)
         {
@@ -1216,7 +1247,7 @@ void StorageMergeTree::resumeExportPartitionTasks()
             parts_to_export.emplace_back(part);
         }
 
-        if (getContext()->getSettingsRef()[Setting::export_merge_tree_partition_individual_part_executor])
+        if (!getContext()->getSettingsRef()[Setting::export_merge_tree_partition_executor])
         {
             for (const auto & part : parts_to_export)
             {
diff --git a/src/Storages/StorageMergeTree.h b/src/Storages/StorageMergeTree.h
@@ -96,6 +96,8 @@ class StorageMergeTree final : public MergeTreeData
     /// Return introspection information about currently processing or recently processed mutations.
     std::vector<MergeTreeMutationStatus> getMutationsStatus() const override;
 
+    std::vector<MergeTreeExportStatus> getExportsStatus() const override;
+
     CancellationCode killMutation(const String & mutation_id) override;
 
     /// Makes backup entries to backup the data of the storage.
@@ -160,7 +162,7 @@ class StorageMergeTree final : public MergeTreeData
     std::map<UInt64, MergeTreeMutationEntry> current_mutations_by_version;
     
     std::map<String, std::shared_ptr<MergeTreeExportManifest>> export_partition_transaction_id_to_manifest;
-    std::mutex export_partition_transaction_id_to_manifest_mutex;
+    mutable std::mutex export_partition_transaction_id_to_manifest_mutex; /// mutable because of getExportsStatus
 
     /// Unfinished mutations that are required for AlterConversions.
     MutationCounters mutation_counters;
diff --git a/src/Storages/System/StorageSystemExports.cpp b/src/Storages/System/StorageSystemExports.cpp

Original file line number	Diff line number	Diff line change
`@@ -189,7 +189,7 @@ bool ExportPartitionPlainMergeTreeTask::commitExport()`
`189`	`189`	`manifest->exportedPaths(),`
`190`	`190`	`context);`
`191`	`191`
`192`		`- manifest->completed = true;`
	`192`	`+ manifest->status = MergeTreeExportManifest::Status::completed;`
`193`	`193`	`manifest->write();`
`194`	`194`
`195`	`195`	`storage.export_partition_transaction_id_to_manifest.erase(manifest->transaction_id);`
`@@ -204,7 +204,7 @@ bool ExportPartitionPlainMergeTreeTask::commitExport()`
`204`	`204`	`void ExportPartitionPlainMergeTreeTask::onCompleted()`
`205`	`205`	`{`
`206`	`206`	`std::lock_guard lock(storage.export_partition_transaction_id_to_manifest_mutex);`
`207`		`- task_result_callback(manifest->completed);`
	`207`	`+ task_result_callback(manifest->status == MergeTreeExportManifest::Status::completed);`
`208`	`208`	`}`
`209`	`209`
`210`	`210`	`void ExportPartitionPlainMergeTreeTask::cancel() noexcept`