Fix iterator locking

ianton-ru · ianton-ru · commit d01eb1575f46 · 2026-03-10T10:42:45.000+01:00
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergIterator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergIterator.cpp
@@ -319,8 +319,10 @@ IcebergIterator::IcebergIterator(
     std::sort(equality_deletes_files.begin(), equality_deletes_files.end());
     std::sort(position_deletes_files.begin(), position_deletes_files.end());
     producer_task.emplace(
-        [this]()
+        [this, thread_group = DB::CurrentThread::getGroup()]()
         {
+            ThreadGroupSwitcher switcher(thread_group, "IcebergKeysIterator");
+
             while (!blocking_queue.isFinished())
             {
                 std::optional<ManifestFileEntry> entry;
diff --git a/src/Storages/ObjectStorage/IObjectIterator.h b/src/Storages/ObjectStorage/IObjectIterator.h
@@ -41,6 +41,8 @@ class ObjectIteratorWithPathAndFileFilter : public IObjectIterator, private With
     size_t estimatedKeysCount() override { return iterator->estimatedKeysCount(); }
     std::optional<UInt64> getSnapshotVersion() const override { return iterator->getSnapshotVersion(); }
 
+    bool has_concurrent_next() const override { return iterator->has_concurrent_next(); }
+
 private:
     const ObjectIterator iterator;
     const std::string object_namespace;
@@ -62,6 +64,8 @@ class ObjectIteratorSplitByBuckets : public IObjectIterator, private WithContext
     size_t estimatedKeysCount() override { return iterator->estimatedKeysCount(); }
     std::optional<UInt64> getSnapshotVersion() const override { return iterator->getSnapshotVersion(); }
 
+    bool has_concurrent_next() const override { return iterator->has_concurrent_next(); }
+
 private:
     const ObjectIterator iterator;
     String format;

Original file line number	Diff line number	Diff line change
`@@ -319,8 +319,10 @@ IcebergIterator::IcebergIterator(`
`319`	`319`	`std::sort(equality_deletes_files.begin(), equality_deletes_files.end());`
`320`	`320`	`std::sort(position_deletes_files.begin(), position_deletes_files.end());`
`321`	`321`	`producer_task.emplace(`
`322`		`- [this]()`
	`322`	`+ [this, thread_group = DB::CurrentThread::getGroup()]()`
`323`	`323`	`{`
	`324`	`+ ThreadGroupSwitcher switcher(thread_group, "IcebergKeysIterator");`
	`325`	`+`
`324`	`326`	`while (!blocking_queue.isFinished())`
`325`	`327`	`{`
`326`	`328`	`std::optional<ManifestFileEntry> entry;`