Fix file identifier in rescheduleTasksFromReplica

Author: ianton-ru

src/Storages/ObjectStorage/StorageObjectStorageStableTaskDistributor.cpp

@@ -71,7 +71,7 @@ ObjectInfoPtr StorageObjectStorageStableTaskDistributor::getNextTask(size_t numb
         auto processed_file_list_ptr = replica_to_files_to_be_processed.find(number_of_current_replica);
         if (processed_file_list_ptr == replica_to_files_to_be_processed.end())
         { // It is possible that replica was lost after check in the begining of the method
-            auto file_identifier = file->getAbsolutePath().value_or(file->getPath());
+            auto file_identifier = getFileIdentifier(file);
             auto file_replica_idx = getReplicaForFile(file_identifier);
             unprocessed_files.emplace(file_identifier, std::make_pair(file, file_replica_idx));
             connection_to_files[file_replica_idx].push_back(file);
@@ -136,7 +136,7 @@ ObjectInfoPtr StorageObjectStorageStableTaskDistributor::getPreQueuedFile(size_t
         auto next_file = files.back();
         files.pop_back();
 
-        auto file_identifier = send_over_whole_archive ? next_file->getPathOrPathToArchiveIfArchive() : next_file->getIdentifier();
+        auto file_identifier = getFileIdentifier(next_file);
         auto it = unprocessed_files.find(file_identifier);
         if (it == unprocessed_files.end())
             continue;
@@ -194,18 +194,7 @@ ObjectInfoPtr StorageObjectStorageStableTaskDistributor::getMatchingFileFromIter
             }
         }
 
-        String file_identifier;
-        if (send_over_whole_archive && object_info->isArchive())
-        {
-            file_identifier = object_info->getPathOrPathToArchiveIfArchive();
-            LOG_TEST(log, "Will send over the whole archive {} to replicas. "
-                     "This will be suboptimal, consider turning on "
-                     "cluster_function_process_archive_on_multiple_nodes setting", file_identifier);
-        }
-        else
-        {
-            file_identifier = object_info->getIdentifier();
-        }
+        String file_identifier = getFileIdentifier(object_info, true);
 
         size_t file_replica_idx;
 
@@ -269,7 +258,7 @@ ObjectInfoPtr StorageObjectStorageStableTaskDistributor::getAnyUnprocessedFile(s
                 auto next_file = it->second.first;
                 unprocessed_files.erase(it);
 
-                auto file_path = send_over_whole_archive ? next_file->getPathOrPathToArchiveIfArchive() : next_file->getAbsolutePath().value_or(next_file->getPath());
+                auto file_path = getFileIdentifier(next_file);
                 LOG_TRACE(
                     log,
                     "Iterator exhausted. Assigning unprocessed file {} to replica {} from matched replica {}",
@@ -333,10 +322,27 @@ void StorageObjectStorageStableTaskDistributor::rescheduleTasksFromReplica(size_
     replica_to_files_to_be_processed.erase(number_of_current_replica);
     for (const auto & file : files)
     {
-        auto file_replica_idx = getReplicaForFile(file->getAbsolutePath().value_or(file->getPath()));
-        unprocessed_files.emplace(file->getAbsolutePath().value_or(file->getPath()), std::make_pair(file, file_replica_idx));
+        auto file_identifier = getFileIdentifier(file);
+        auto file_replica_idx = getReplicaForFile(file_identifier);
+        unprocessed_files.emplace(file_identifier, std::make_pair(file, file_replica_idx));
         connection_to_files[file_replica_idx].push_back(file);
     }
 }
 
+String StorageObjectStorageStableTaskDistributor::getFileIdentifier(ObjectInfoPtr file_object, bool write_to_log) const
+{
+    if (send_over_whole_archive && file_object->isArchive())
+    {
+        auto file_identifier = file_object->getPathOrPathToArchiveIfArchive();
+        if (write_to_log)
+        {
+            LOG_TEST(log, "Will send over the whole archive {} to replicas. "
+                        "This will be suboptimal, consider turning on "
+                        "cluster_function_process_archive_on_multiple_nodes setting", file_identifier);
+        }
+        return file_identifier;
+    }
+    return file_object->getIdentifier();
+}
+
 }

src/Storages/ObjectStorage/StorageObjectStorageStableTaskDistributor.h

@@ -40,6 +40,8 @@ class StorageObjectStorageStableTaskDistributor
 
     void saveLastNodeActivity(size_t number_of_current_replica);
 
+    String getFileIdentifier(ObjectInfoPtr file_object, bool write_to_log = false) const;
+
     const std::shared_ptr<IObjectIterator> iterator;
     const bool send_over_whole_archive;