tmp

Author: arthurpassos

src/Storages/MergeTree/BackgroundJobsAssignee.cpp

@@ -95,6 +95,10 @@ bool BackgroundJobsAssignee::scheduleCommonTask(ExecutableTaskPtr common_task, b
     return schedule_res;
 }
 
+std::size_t BackgroundJobsAssignee::getAvailableMoveExecutors() const
+{
+    return getContext()->getMovesExecutor()->getAvailableSlots();
+}
 
 String BackgroundJobsAssignee::toString(Type type)
 {

src/Storages/MergeTree/BackgroundJobsAssignee.h

@@ -55,6 +55,8 @@ class BackgroundJobsAssignee : public WithContext
     bool scheduleMoveTask(ExecutableTaskPtr move_task);
     bool scheduleCommonTask(ExecutableTaskPtr common_task, bool need_trigger);
 
+    std::size_t getAvailableMoveExecutors() const;
+
     /// Just call finish
     ~BackgroundJobsAssignee();
 

src/Storages/MergeTree/ExportPartFromPartitionExportTask.cpp

@@ -27,11 +27,14 @@ bool ExportPartFromPartitionExportTask::executeStep()
 {
     const auto zk = storage.getZooKeeper();
     const auto part_name = manifest.data_part->name;
-    
+
+    LOG_INFO(storage.log, "ExportPartition scheduler task: Attempting to lock part: {}", part_name);
+
     ProfileEvents::increment(ProfileEvents::ExportPartitionZooKeeperRequests);
     ProfileEvents::increment(ProfileEvents::ExportPartitionZooKeeperCreate);
     if (Coordination::Error::ZOK == zk->tryCreate(fs::path(storage.zookeeper_path) / "exports" / key / "locks" / part_name, storage.replica_name, zkutil::CreateMode::Ephemeral))
     {
+        LOG_INFO(storage.log, "ExportPartition scheduler task: Locked part: {}", part_name);
         export_part_task->executeStep();
         return false;
     }

src/Storages/MergeTree/ExportPartitionTaskScheduler.cpp

@@ -60,13 +60,30 @@ ExportPartitionTaskScheduler::ExportPartitionTaskScheduler(StorageReplicatedMerg
 
 void ExportPartitionTaskScheduler::run()
 {
+    const auto available_move_executors = storage.background_moves_assignee.getAvailableMoveExecutors();
+    if (available_move_executors == 0)
+    {
+        LOG_INFO(storage.log, "ExportPartition scheduler task: No available move executors, skipping");
+        return;
+    }
+
+    LOG_INFO(storage.log, "ExportPartition scheduler task: Available move executors: {}", available_move_executors);
+
+    std::size_t scheduled_exports_count = 0;
+
     std::lock_guard lock(storage.export_merge_tree_partition_mutex);
 
     auto zk = storage.getZooKeeper();
 
     // Iterate sorted by create_time
     for (auto & entry : storage.export_merge_tree_partition_task_entries_by_create_time)
     {
+        if (scheduled_exports_count >= available_move_executors)
+        {
+            LOG_INFO(storage.log, "ExportPartition scheduler task: Scheduled exports count is greater than available move executors, skipping");
+            break;
+        }
+
         const auto & manifest = entry.manifest;
         const auto key = entry.getCompositeKey();
         const auto database = storage.getContext()->resolveDatabase(manifest.destination_database);
@@ -144,6 +161,12 @@ void ExportPartitionTaskScheduler::run()
 
         for (const auto & zk_part_name : parts_in_processing_or_pending)
         {
+            if (scheduled_exports_count >= available_move_executors)
+            {
+                LOG_INFO(storage.log, "ExportPartition scheduler task: Scheduled exports count is greater than available move executors, skipping");
+                break;
+            }
+
             if (locked_parts_set.contains(zk_part_name))
             {
                 LOG_INFO(storage.log, "ExportPartition scheduler task: Part {} is locked, skipping", zk_part_name);
@@ -157,6 +180,8 @@ void ExportPartitionTaskScheduler::run()
                 continue;
             }
 
+            LOG_INFO(storage.log, "ExportPartition scheduler task: Scheduling part export: {}", zk_part_name);
+
             std::lock_guard part_export_lock(storage.export_manifests_mutex);
 
             auto context = getContextCopyWithTaskSettings(storage.getContext(), manifest);

src/Storages/MergeTree/MergeTreeBackgroundExecutor.cpp

@@ -136,6 +136,12 @@ size_t MergeTreeBackgroundExecutor<Queue>::getMaxTasksCount() const
     return max_tasks_count.load(std::memory_order_relaxed);
 }
 
+template <class Queue>
+size_t MergeTreeBackgroundExecutor<Queue>::getAvailableSlots() const
+{
+    return getMaxTasksCount() - CurrentMetrics::values[metric].load(std::memory_order_relaxed);
+}
+
 template <class Queue>
 bool MergeTreeBackgroundExecutor<Queue>::trySchedule(ExecutableTaskPtr task)
 {

src/Storages/MergeTree/MergeTreeBackgroundExecutor.h

@@ -279,6 +279,8 @@ class MergeTreeBackgroundExecutor final : boost::noncopyable
     /// can lead only to some postponing, not logical error.
     size_t getMaxTasksCount() const;
 
+    size_t getAvailableSlots() const;
+
     bool trySchedule(ExecutableTaskPtr task);
     void removeTasksCorrespondingToStorage(StorageID id);
     void wait();