[WIP] Add stable file to node mapping in StorageObjectStorageCluster - add setting

Author: adikus

src/Core/Settings.cpp

@@ -495,6 +495,9 @@ Ignore absence of file if it does not exist when reading certain keys.
 Possible values:
 - 1 — `SELECT` returns empty result.
 - 0 — `SELECT` throws an exception.
+)", 0) \
+    DECLARE(Bool, object_storage_stable_cluster_task_distribution, false, R"(
+Use stable task distribution for object storage cluster table functions in order to better utilize filesystem cache.
 )", 0) \
     DECLARE(Bool, hdfs_ignore_file_doesnt_exist, false, R"(
 Ignore absence of file if it does not exist when reading certain keys.

src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp

@@ -19,6 +19,7 @@ namespace DB
 namespace Setting
 {
     extern const SettingsBool use_hive_partitioning;
+    extern const SettingsBool object_storage_stable_cluster_task_distribution;
 }
 
 namespace ErrorCodes
@@ -120,14 +121,29 @@ RemoteQueryExecutor::Extension StorageObjectStorageCluster::getTaskIteratorExten
         local_context, predicate, virtual_columns, nullptr, local_context->getFileProgressCallback());
 
     auto cluster = getCluster(local_context);
-    auto task_distributor = std::make_shared<StorageObjectStorageStableTaskDistributor>(iterator, cluster);
 
-    auto callback = std::make_shared<TaskIterator>(
-        [task_distributor](Connection * connection) mutable -> String {
-            return task_distributor->getNextTask(connection);
-        });
-    
-    return RemoteQueryExecutor::Extension{ .task_iterator = std::move(callback) };
+    if (local_context->getSettingsRef()[Setting::object_storage_stable_cluster_task_distribution])
+    {
+        auto task_distributor = std::make_shared<StorageObjectStorageStableTaskDistributor>(iterator, cluster);
+        
+        auto callback = std::make_shared<TaskIterator>(
+            [task_distributor](Connection * connection) mutable -> String {
+                return task_distributor->getNextTask(connection);
+            });
+        
+        return RemoteQueryExecutor::Extension{ .task_iterator = std::move(callback) };
+    }
+    else
+    {
+        auto callback = std::make_shared<TaskIterator>(
+            [iterator](Connection *) mutable -> String {
+                if (auto object_info = iterator->next(0))
+                    return object_info->getPath();
+                return "";
+            });
+        
+        return RemoteQueryExecutor::Extension{ .task_iterator = std::move(callback) };
+    }
 }
 
 }