Merge pull request ClickHouse#89147 from ClickHouse/backport/25.8/89040

Backport ClickHouse#89040 to 25.8: Replace SchedulePool with a separate thread

Author: divanik

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergIterator.cpp

@@ -278,39 +278,7 @@ IcebergIterator::IcebergIterator(
           data_snapshot_,
           persistent_components_)
     , blocking_queue(100)
-    , producer_task(local_context_->getSchedulePool().createTask(
-          "IcebergMetaReaderThread",
-          [this]
-          {
-              while (!blocking_queue.isFinished())
-              {
-                  std::optional<ManifestFileEntry> entry;
-                  try
-                  {
-                      entry = data_files_iterator.next();
-                  }
-                  catch (...)
-                  {
-                      std::lock_guard lock(exception_mutex);
-                      if (!exception)
-                      {
-                          exception = std::current_exception();
-                      }
-                      blocking_queue.finish();
-                      break;
-                  }
-                  if (!entry.has_value())
-                      break;
-                  while (!blocking_queue.push(std::move(entry.value())))
-                  {
-                      if (blocking_queue.isFinished())
-                      {
-                          break;
-                      }
-                  }
-              }
-              blocking_queue.finish();
-          }))
+    , producer_task(std::nullopt)
     , callback(std::move(callback_))
     , format(configuration_.lock()->format)
     , compression_method(configuration_.lock()->compression_method)
@@ -330,8 +298,38 @@ IcebergIterator::IcebergIterator(
     }
     std::sort(equality_deletes_files.begin(), equality_deletes_files.end());
     std::sort(position_deletes_files.begin(), position_deletes_files.end());
-
-    producer_task->activateAndSchedule();
+    producer_task.emplace(
+        [this]()
+        {
+            while (!blocking_queue.isFinished())
+            {
+                std::optional<ManifestFileEntry> entry;
+                try
+                {
+                    entry = data_files_iterator.next();
+                }
+                catch (...)
+                {
+                    std::lock_guard lock(exception_mutex);
+                    if (!exception)
+                    {
+                        exception = std::current_exception();
+                    }
+                    blocking_queue.finish();
+                    break;
+                }
+                if (!entry.has_value())
+                    break;
+                while (!blocking_queue.push(std::move(entry.value())))
+                {
+                    if (blocking_queue.isFinished())
+                    {
+                        break;
+                    }
+                }
+            }
+            blocking_queue.finish();
+        });
 }
 
 ObjectInfoPtr IcebergIterator::next(size_t)
@@ -372,7 +370,10 @@ size_t IcebergIterator::estimatedKeysCount()
 IcebergIterator::~IcebergIterator()
 {
     blocking_queue.finish();
-    producer_task->deactivate();
+    if (producer_task)
+    {
+        producer_task->join();
+    }
 }
 }
 

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergIterator.h

@@ -103,7 +103,7 @@ class IcebergIterator : public IObjectIterator
     Iceberg::SingleThreadIcebergKeysIterator data_files_iterator;
     Iceberg::SingleThreadIcebergKeysIterator deletes_iterator;
     ConcurrentBoundedQueue<Iceberg::ManifestFileEntry> blocking_queue;
-    BackgroundSchedulePool::TaskHolder producer_task;
+    std::optional<ThreadFromGlobalPool> producer_task;
     IDataLakeMetadata::FileProgressCallback callback;
     const String format;
     const String compression_method;