Merge pull request #1042 from Altinity/bugfix/antalya-25.6.5/lock_object_storage_task_distribution_ms_lost_host

Antalya 25.6: Fix lock_object_storage_task_distribution_ms, Changed lock_object_storage_task_distribution_ms value to 500

Author: Enmk

src/Core/Settings.cpp

@@ -7001,8 +7001,18 @@ Default number of tasks for parallel reading in distributed query. Tasks are spr
     DECLARE(Bool, distributed_plan_optimize_exchanges, true, R"(
 Removes unnecessary exchanges in distributed query plan. Disable it for debugging.
 )", 0) \
-    DECLARE(UInt64, lock_object_storage_task_distribution_ms, 0, R"(
-In object storage distribution queries do not distibute tasks on non-prefetched nodes until prefetched node is active.
+    DECLARE(UInt64, lock_object_storage_task_distribution_ms, 500, R"(
+In object storage distribution queries do not distribute tasks on non-prefetched nodes until prefetched node is active.
+Determines how long the free executor node (one that finished processing all of it assigned tasks) should wait before "stealing" tasks from queue of currently busy executor nodes.
+
+Possible values:
+
+- 0  - steal tasks immediately after freeing up.
+- >0 - wait for specified period of time before stealing tasks.
+
+Having this `>0` helps with cache reuse and might improve overall query time.
+Because busy node might have warmed-up caches for this specific task, while free node needs to fetch lots of data from S3.
+Which might take longer than just waiting for the busy node and generate extra traffic.
 )", EXPERIMENTAL) \
     DECLARE(String, distributed_plan_force_exchange_kind, "", R"(
 Force specified kind of Exchange operators between distributed query stages.

src/Core/SettingsChangesHistory.cpp

@@ -39,6 +39,10 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
         /// controls new feature and it's 'true' by default, use 'false' as previous_value).
         /// It's used to implement `compatibility` setting (see https://github.com/ClickHouse/ClickHouse/issues/35972)
         /// Note: please check if the key already exists to prevent duplicate entries.
+        addSettingsChanges(settings_changes_history, "25.8.1.20364",
+        {
+            {"lock_object_storage_task_distribution_ms", 500, 500, "Raised the value to 500 to avoid hoping tasks between executors."},
+        });
         addSettingsChanges(settings_changes_history, "25.8",
         {
             {"output_format_json_quote_64bit_integers", true, false, "Disable quoting of the 64 bit integers in JSON by default"},

src/Storages/ObjectStorage/StorageObjectStorageStableTaskDistributor.cpp

@@ -189,7 +189,7 @@ ObjectInfoPtr StorageObjectStorageStableTaskDistributor::getMatchingFileFromIter
         // Queue file for its assigned replica
         {
             std::lock_guard lock(mutex);
-            unprocessed_files.emplace(file_path, std::make_pair(object_info, number_of_current_replica));
+            unprocessed_files.emplace(file_path, std::make_pair(object_info, file_replica_idx));
             connection_to_files[file_replica_idx].push_back(object_info);
         }
     }
@@ -281,7 +281,11 @@ void StorageObjectStorageStableTaskDistributor::rescheduleTasksFromReplica(size_
 
     replica_to_files_to_be_processed.erase(number_of_current_replica);
     for (const auto & file : processed_file_list_ptr->second)
-        unprocessed_files.emplace(file->getPath(), std::make_pair(file, getReplicaForFile(file->getPath())));
+    {
+        auto file_replica_idx = getReplicaForFile(file->getPath());
+        unprocessed_files.emplace(file->getPath(), std::make_pair(file, file_replica_idx));
+        connection_to_files[file_replica_idx].push_back(file);
+    }
 }
 
 }

tests/integration/test_s3_cache_locality/test.py

@@ -86,8 +86,7 @@ def check_s3_gets(cluster, node, expected_result, cluster_first, cluster_second,
         "filesystem_cache_name": "'raw_s3_cache'",
         }
 
-    if lock_object_storage_task_distribution_ms > 0:
-        settings["lock_object_storage_task_distribution_ms"] = lock_object_storage_task_distribution_ms
+    settings["lock_object_storage_task_distribution_ms"] = lock_object_storage_task_distribution_ms
 
     query_id_first = str(uuid.uuid4())
     result_first = node.query(