[WIP] Add stable file to node mapping in StorageObjectStorageCluster - rename node -> replica

Author: adikus

src/Storages/ObjectStorage/StorageObjectStorageStableTaskDistributor.cpp

@@ -15,15 +15,15 @@ StorageObjectStorageStableTaskDistributor::StorageObjectStorageStableTaskDistrib
 
     LOG_INFO(
         log,
-        "Initialized StorageObjectStorageStableTaskDistributor to distribute files across {} unique nodes",
-        total_nodes
+        "Initialized StorageObjectStorageStableTaskDistributor to distribute files across {} unique replicas",
+        total_replicas
     );
 }
 
 void StorageObjectStorageStableTaskDistributor::initializeConnectionMapping(const ClusterPtr & cluster)
 {
-    connection_key_to_node.clear();
-    node_to_connection_key.clear();
+    connection_key_to_replica.clear();
+    replica_to_connection_key.clear();
 
     const auto & addresses_with_failover = cluster->getShardsAddresses();
 
@@ -35,52 +35,52 @@ void StorageObjectStorageStableTaskDistributor::initializeConnectionMapping(cons
         {
             String connection_key = address.host_name + ":" + std::to_string(address.port);
 
-            if (connection_key_to_node.contains(connection_key))
+            if (connection_key_to_replica.contains(connection_key))
                 continue;
 
-            Int32 node_idx = static_cast<Int32>(node_to_connection_key.size());
-            connection_key_to_node[connection_key] = node_idx;
-            node_to_connection_key.push_back(connection_key);
+            Int32 replica_idx = static_cast<Int32>(replica_to_connection_key.size());
+            connection_key_to_replica[connection_key] = replica_idx;
+            replica_to_connection_key.push_back(connection_key);
 
             LOG_TRACE(
                 log,
-                "Discovered shard {} replica with connection key {} (node_idx: {})",
+                "Discovered shard {} replica with connection key {} (replica_idx: {})",
                 shard_idx,
                 connection_key,
-                node_idx
+                replica_idx
             );
         }
     }
 
-    total_nodes = static_cast<Int32>(node_to_connection_key.size());
+    total_replicas = static_cast<Int32>(replica_to_connection_key.size());
 
     LOG_INFO(
         log,
-        "Mapping connections to {} unique nodes",
-        total_nodes
+        "Mapping connections to {} unique replicas",
+        total_replicas
     );
 }
 
 String StorageObjectStorageStableTaskDistributor::getNextTask(Connection * connection)
 {
     String connection_key = "default";
-    Int32 node_idx = -1;
+    Int32 replica_idx = -1;
 
     if (connection)
     {
         connection_key = connection->getHost() + ":" + std::to_string(connection->getPort());
-        auto it = connection_key_to_node.find(connection_key);
-        if (it != connection_key_to_node.end())
+        auto it = connection_key_to_replica.find(connection_key);
+        if (it != connection_key_to_replica.end())
         {
-            node_idx = it->second;
+            replica_idx = it->second;
         }
     }
 
     LOG_TRACE(
         log,
-        "Received a new connection ({}, node_idx: {}) looking for a file",
+        "Received a new connection ({}, replica_idx: {}) looking for a file",
         connection_key,
-        node_idx
+        replica_idx
     );
 
     // 1. Check pre-queued files first
@@ -89,26 +89,26 @@ String StorageObjectStorageStableTaskDistributor::getNextTask(Connection * conne
         return file;
 
     // 2. Try to find a matching file from the iterator
-    file = getMatchingFileFromIterator(connection_key, node_idx);
+    file = getMatchingFileFromIterator(connection_key, replica_idx);
     if (!file.empty())
         return file;
 
     if (!unprocessed_files.empty()) {
-        // Prevent initiator from stealing jobs from other nodes 
+        // Prevent initiator from stealing jobs from other replicas 
         sleepForMilliseconds(50);
     }
 
     // 3. Process unprocessed files if iterator is exhausted
     return getAnyUnprocessedFile(connection_key);
 }
 
-Int32 StorageObjectStorageStableTaskDistributor::getNodeForFile(const String & file_path, Int32 total_nodes)
+Int32 StorageObjectStorageStableTaskDistributor::getReplicaForFile(const String & file_path, Int32 total_replicas)
 {
-    if (total_nodes <= 0)
+    if (total_replicas <= 0)
         return 0;
 
     UInt64 hash_value = sipHash64(file_path);
-    return static_cast<Int32>(hash_value % total_nodes);
+    return static_cast<Int32>(hash_value % total_replicas);
 }
 
 String StorageObjectStorageStableTaskDistributor::getPreQueuedFile(const String & connection_key)
@@ -143,7 +143,7 @@ String StorageObjectStorageStableTaskDistributor::getPreQueuedFile(const String
     return "";
 }
 
-String StorageObjectStorageStableTaskDistributor::getMatchingFileFromIterator(const String & connection_key, Int32 node_idx)
+String StorageObjectStorageStableTaskDistributor::getMatchingFileFromIterator(const String & connection_key, Int32 replica_idx)
 {
     while (!iterator_exhausted)
     {
@@ -161,9 +161,9 @@ String StorageObjectStorageStableTaskDistributor::getMatchingFileFromIterator(co
         }
 
         String file_path = object_info->getPath();
-        Int32 file_node_idx = getNodeForFile(file_path, total_nodes);
+        Int32 file_replica_idx = getReplicaForFile(file_path, total_replicas);
 
-        if (file_node_idx == node_idx)
+        if (file_replica_idx == replica_idx)
         {
             std::lock_guard lock(mutex);
             total_files_processed++;
@@ -179,14 +179,14 @@ String StorageObjectStorageStableTaskDistributor::getMatchingFileFromIterator(co
             return file_path;
         }
 
-        // Queue file for its assigned node
+        // Queue file for its assigned replica
         {
             std::lock_guard lock(mutex);
 
             unprocessed_files.insert(file_path);
-            if (file_node_idx < total_nodes)
+            if (file_replica_idx < total_replicas)
             {
-                String target_connection_key = node_to_connection_key[file_node_idx];
+                String target_connection_key = replica_to_connection_key[file_replica_idx];
                 connection_to_files[target_connection_key].push_back(file_path);
             }
         }

src/Storages/ObjectStorage/StorageObjectStorageStableTaskDistributor.h

@@ -16,8 +16,8 @@ namespace DB
 class StorageObjectStorageStableTaskDistributor
 {
 public:
-    using ConnectionKeyToNodeMap = std::unordered_map<String, Int32>;
-    using NodeToConnectionKeyMap = std::vector<String>;
+    using ConnectionKeyToReplicaMap = std::unordered_map<String, Int32>;
+    using ReplicaToConnectionKeyMap = std::vector<String>;
 
     StorageObjectStorageStableTaskDistributor(
         std::shared_ptr<IObjectIterator> iterator_,
@@ -28,15 +28,15 @@ class StorageObjectStorageStableTaskDistributor
 private:
     void initializeConnectionMapping(const ClusterPtr & cluster);
 
-    static Int32 getNodeForFile(const String & file_path, Int32 total_nodes);
+    static Int32 getReplicaForFile(const String & file_path, Int32 total_replicas);
     String getPreQueuedFile(const String & connection_key);
-    String getMatchingFileFromIterator(const String & connection_key, Int32 node_idx);
+    String getMatchingFileFromIterator(const String & connection_key, Int32 replica_idx);
     String getAnyUnprocessedFile(const String & connection_key);
 
     std::shared_ptr<IObjectIterator> iterator;
-    ConnectionKeyToNodeMap connection_key_to_node;
-    NodeToConnectionKeyMap node_to_connection_key;
-    Int32 total_nodes;
+    ConnectionKeyToReplicaMap connection_key_to_replica;
+    ReplicaToConnectionKeyMap replica_to_connection_key;
+    Int32 total_replicas;
 
     std::mutex mutex;
     std::unordered_map<String, std::vector<String>> connection_to_files;