preserve order of tasks..

Author: arthurpassos

src/Storages/ExportReplicatedMergeTreePartitionTaskEntry.h

@@ -2,6 +2,11 @@
 
 #include <Storages/ExportReplicatedMergeTreePartitionManifest.h>
 #include <Storages/MergeTree/IMergeTreeDataPart.h>
+#include "Core/QualifiedTableName.h"
+#include <boost/multi_index_container.hpp>
+#include <boost/multi_index/hashed_index.hpp>
+#include <boost/multi_index/ordered_index.hpp>
+#include <boost/multi_index/mem_fun.hpp>
 
 namespace DB
 {
@@ -15,6 +20,38 @@ struct ExportReplicatedMergeTreePartitionTaskEntry
     /// It does not mean this replica will export all the parts
     /// There is also a chance this replica does not contain a given part and it is totally ok.
     std::vector<DataPartPtr> part_references;
+
+    std::string getCompositeKey() const
+    {
+        const auto qualified_table_name = QualifiedTableName {manifest.destination_database, manifest.destination_table};
+        return manifest.partition_id + "_" + qualified_table_name.getFullName();
+    }
+
+    /// Get create_time for sorted iteration
+    time_t getCreateTime() const
+    {
+        return manifest.create_time;
+    }
 };
 
+struct ExportPartitionTaskEntryTagByCompositeKey {};
+struct ExportPartitionTaskEntryTagByCreateTime {};
+
+// Multi-index container for export partition task entries
+// - Index 0 (TagByCompositeKey): hashed_unique on composite key for O(1) lookup
+// - Index 1 (TagByCreateTime): ordered_non_unique on create_time for sorted iteration
+using ExportPartitionTaskEntriesContainer = boost::multi_index_container<
+    ExportReplicatedMergeTreePartitionTaskEntry,
+    boost::multi_index::indexed_by<
+        boost::multi_index::hashed_unique<
+            boost::multi_index::tag<ExportPartitionTaskEntryTagByCompositeKey>,
+            boost::multi_index::const_mem_fun<ExportReplicatedMergeTreePartitionTaskEntry, std::string, &ExportReplicatedMergeTreePartitionTaskEntry::getCompositeKey>
+        >,
+        boost::multi_index::ordered_non_unique<
+            boost::multi_index::tag<ExportPartitionTaskEntryTagByCreateTime>,
+            boost::multi_index::const_mem_fun<ExportReplicatedMergeTreePartitionTaskEntry, time_t, &ExportReplicatedMergeTreePartitionTaskEntry::getCreateTime>
+        >
+    >
+>;
+
 }

src/Storages/MergeTree/ExportPartitionManifestUpdatingTask.cpp

@@ -98,12 +98,13 @@ void ExportPartitionManifestUpdatingTask::run()
 
         const auto metadata = ExportReplicatedMergeTreePartitionManifest::fromJsonString(metadata_json);
 
-        const auto local_entry = storage.export_merge_tree_partition_task_entries.find(key);
+        auto & entries_by_key = storage.export_merge_tree_partition_task_entries_by_key;
+        const auto local_entry = entries_by_key.find(key);
 
         /// If the zk entry has been replaced with export_merge_tree_partition_force_export, checking only for the export key is not enough
         /// we need to make sure it is the same transaction id. If it is not, it needs to be replaced.
-        bool has_local_entry_and_is_up_to_date = local_entry != storage.export_merge_tree_partition_task_entries.end()
-            && local_entry->second.manifest.transaction_id == metadata.transaction_id;
+        bool has_local_entry_and_is_up_to_date = local_entry != entries_by_key.end()
+            && local_entry->manifest.transaction_id == metadata.transaction_id;
 
         /// If the entry is up to date and we don't have the cleanup lock, early exit, nothing to be done.
         if (!cleanup_lock_acquired && has_local_entry_and_is_up_to_date)
@@ -125,7 +126,9 @@ void ExportPartitionManifestUpdatingTask::run()
             if (has_expired && is_not_pending)
             {
                 zk->tryRemoveRecursive(fs::path(entry_path));
-                storage.export_merge_tree_partition_task_entries.erase(key);
+                auto it = entries_by_key.find(key);
+                if (it != entries_by_key.end())
+                    entries_by_key.erase(it);
                 LOG_INFO(storage.log, "ExportPartition: Removed {}: expired", key);
                 continue;
             }
@@ -180,46 +183,46 @@ void ExportPartitionManifestUpdatingTask::run()
             }
         }
 
-        /// It is important to use the operator[] because it updates the existing entry if it already exists.
-        storage.export_merge_tree_partition_task_entries[key] = ExportReplicatedMergeTreePartitionTaskEntry {metadata, std::move(part_references)};
+        /// Insert or update entry. The multi_index container automatically maintains both indexes.
+        auto entry = ExportReplicatedMergeTreePartitionTaskEntry {metadata, std::move(part_references)};
+        auto it = entries_by_key.find(key);
+        if (it != entries_by_key.end())
+            entries_by_key.replace(it, entry);
+        else
+            entries_by_key.insert(entry);
     }
 
     /// Remove entries that were deleted by someone else
-    std::erase_if(storage.export_merge_tree_partition_task_entries,
-        [&](auto const & kv)
+    auto & entries_by_key = storage.export_merge_tree_partition_task_entries_by_key;
+    for (auto it = entries_by_key.begin(); it != entries_by_key.end();)
+    {
+        const auto & key = it->getCompositeKey();
+        if (zk_children.contains(key))
         {
-            if (zk_children.contains(kv.first))
-            {
-                return false;
-            }
+            ++it;
+            continue;
+        }
 
-            const auto & transaction_id = kv.second.manifest.transaction_id;
-            LOG_INFO(storage.log, "ExportPartition: Export task {} was deleted, calling killExportPartition for transaction {}", kv.first, transaction_id);
-            
-            try
-            {
-                storage.killExportPart(transaction_id);
-            }
-            catch (...)
-            {
-                tryLogCurrentException(storage.log, __PRETTY_FUNCTION__);
-            }
+        const auto & transaction_id = it->manifest.transaction_id;
+        LOG_INFO(storage.log, "ExportPartition: Export task {} was deleted, calling killExportPartition for transaction {}", key, transaction_id);
+        
+        try
+        {
+            storage.killExportPart(transaction_id);
+        }
+        catch (...)
+        {
+            tryLogCurrentException(storage.log, __PRETTY_FUNCTION__);
+        }
 
-            return true;
-        });
+        it = entries_by_key.erase(it);
+    }
 
     if (cleanup_lock_acquired)
     {
         zk->tryRemove(cleanup_lock_path);
     }
 
-    /// todo arthur remember to sort the entries by create_time
-    // std::sort(storage.export_merge_tree_partition_task_entries.begin(), storage.export_merge_tree_partition_task_entries.end(),
-    //     [](const auto & a, const auto & b)
-    //     {
-    //         return a.second.manifest.create_time < b.second.manifest.create_time;
-    //     });
-
     storage.export_merge_tree_partition_select_task->schedule();
 }
 

src/Storages/MergeTree/ExportPartitionTaskScheduler.cpp

@@ -20,9 +20,11 @@ void ExportPartitionTaskScheduler::run()
 
     auto zk = storage.getZooKeeper();
 
-    for (const auto & [key, entry] : storage.export_merge_tree_partition_task_entries)
+    // Iterate sorted by create_time
+    for (const auto & entry : storage.export_merge_tree_partition_task_entries_by_create_time)
     {
         const auto & manifest = entry.manifest;
+        const auto key = entry.getCompositeKey();
         const auto & database = storage.getContext()->resolveDatabase(manifest.destination_database);
         const auto & table = manifest.destination_table;
 

src/Storages/MergeTree/tests/gtest_export_partition_ordering.cpp

@@ -0,0 +1,74 @@
+#include <gtest/gtest.h>
+#include <Storages/ExportReplicatedMergeTreePartitionTaskEntry.h>
+
+namespace DB
+{
+
+// Test fixture that uses the shared container definition
+class ExportPartitionOrderingTest : public ::testing::Test
+{
+protected:
+    ExportPartitionTaskEntriesContainer container;
+    ExportPartitionTaskEntriesContainer::index<ExportPartitionTaskEntryTagByCompositeKey>::type & by_key;
+    ExportPartitionTaskEntriesContainer::index<ExportPartitionTaskEntryTagByCreateTime>::type & by_create_time;
+
+    ExportPartitionOrderingTest()
+        : by_key(container.get<ExportPartitionTaskEntryTagByCompositeKey>())
+        , by_create_time(container.get<ExportPartitionTaskEntryTagByCreateTime>())
+    {
+    }
+};
+
+TEST_F(ExportPartitionOrderingTest, IterationOrderMatchesCreateTime)
+{
+    // Create entries with different create_times (in reverse order)
+    time_t base_time = 1000;
+    
+    ExportReplicatedMergeTreePartitionManifest manifest1;
+    manifest1.partition_id = "2020";
+    manifest1.destination_database = "db1";
+    manifest1.destination_table = "table1";
+    manifest1.create_time = base_time + 300; // Latest
+    
+    ExportReplicatedMergeTreePartitionManifest manifest2;
+    manifest2.partition_id = "2021";
+    manifest2.destination_database = "db1";
+    manifest2.destination_table = "table1";
+    manifest2.create_time = base_time + 100; // Middle
+    
+    ExportReplicatedMergeTreePartitionManifest manifest3;
+    manifest3.partition_id = "2022";
+    manifest3.destination_database = "db1";
+    manifest3.destination_table = "table1";
+    manifest3.create_time = base_time; // Oldest
+
+    ExportReplicatedMergeTreePartitionTaskEntry entry1{manifest1, {}};
+    ExportReplicatedMergeTreePartitionTaskEntry entry2{manifest2, {}};
+    ExportReplicatedMergeTreePartitionTaskEntry entry3{manifest3, {}};
+
+    // Insert in reverse order
+    by_key.insert(entry1);
+    by_key.insert(entry2);
+    by_key.insert(entry3);
+
+    // Verify iteration order matches create_time (ascending)
+    auto it = by_create_time.begin();
+    ASSERT_NE(it, by_create_time.end());
+    EXPECT_EQ(it->manifest.partition_id, "2022"); // Oldest first
+    EXPECT_EQ(it->manifest.create_time, base_time);
+    
+    ++it;
+    ASSERT_NE(it, by_create_time.end());
+    EXPECT_EQ(it->manifest.partition_id, "2021");
+    EXPECT_EQ(it->manifest.create_time, base_time + 100);
+    
+    ++it;
+    ASSERT_NE(it, by_create_time.end());
+    EXPECT_EQ(it->manifest.partition_id, "2020");
+    EXPECT_EQ(it->manifest.create_time, base_time + 300);
+    
+    ++it;
+    EXPECT_EQ(it, by_create_time.end());
+}
+
+}

src/Storages/StorageReplicatedMergeTree.cpp

@@ -425,6 +425,8 @@ StorageReplicatedMergeTree::StorageReplicatedMergeTree(
     , merge_strategy_picker(*this)
     , queue(*this, merge_strategy_picker)
     , fetcher(*this)
+    , export_merge_tree_partition_task_entries_by_key(export_merge_tree_partition_task_entries.get<ExportPartitionTaskEntryTagByCompositeKey>())
+    , export_merge_tree_partition_task_entries_by_create_time(export_merge_tree_partition_task_entries.get<ExportPartitionTaskEntryTagByCreateTime>())
     , cleanup_thread(*this)
     , async_block_ids_cache(*this)
     , part_check_thread(*this)
@@ -8080,7 +8082,7 @@ void StorageReplicatedMergeTree::exportPartitionToTable(const PartitionCommand &
     const auto exports_path = fs::path(zookeeper_path) / "exports";
     Coordination::Requests ops;
 
-    const auto export_key = partition_id + "_" + dest_storage_id.getNameForLogs();
+    const auto export_key = partition_id + "_" + dest_storage_id.getQualifiedName().getFullName();
 
     const auto partition_exports_path = fs::path(exports_path) / export_key;
 

src/Storages/StorageReplicatedMergeTree.h

@@ -531,8 +531,11 @@ class StorageReplicatedMergeTree final : public MergeTreeData
 
     BackgroundSchedulePoolTaskHolder export_merge_tree_partition_select_task;
 
+    ExportPartitionTaskEntriesContainer export_merge_tree_partition_task_entries;
 
-    std::unordered_map<std::string, ExportReplicatedMergeTreePartitionTaskEntry> export_merge_tree_partition_task_entries;
+    // Convenience references to indexes
+    ExportPartitionTaskEntriesContainer::index<ExportPartitionTaskEntryTagByCompositeKey>::type & export_merge_tree_partition_task_entries_by_key;
+    ExportPartitionTaskEntriesContainer::index<ExportPartitionTaskEntryTagByCreateTime>::type & export_merge_tree_partition_task_entries_by_create_time;
     /// A thread that removes old parts, log entries, and blocks.
     ReplicatedMergeTreeCleanupThread cleanup_thread;