Improvement locking Iceberg iterator

src/Common/ProfileEvents.cpp

@@ -307,6 +307,7 @@
     M(IcebergAvroFileParsingMicroseconds, "Time spent for parsing avro metadata files for Iceberg tables.", ValueType::Microseconds) \
     M(IcebergJsonFileParsing, "Number of times json metadata files have been parsed.", ValueType::Number) \
     M(IcebergJsonFileParsingMicroseconds, "Time spent for parsing json metadata files for Iceberg tables.", ValueType::Microseconds) \
+    M(IcebergIteratorNextMicroseconds, "Time spent for getting next objects in Iceberg iterator.", ValueType::Microseconds) \
     \
     M(JoinBuildTableRowCount, "Total number of rows in the build table for a JOIN operation.", ValueType::Number) \
     M(JoinProbeTableRowCount, "Total number of rows in the probe table for a JOIN operation.", ValueType::Number) \

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

@@ -61,6 +61,7 @@ extern const Event IcebergPartitionPrunedFiles;
 extern const Event IcebergMinMaxIndexPrunedFiles;
 extern const Event IcebergMetadataReadWaitTimeMicroseconds;
 extern const Event IcebergMetadataReturnedObjectInfos;
+extern const Event IcebergIteratorNextMicroseconds;
 };
 
 
@@ -119,6 +120,8 @@ defineDeletesSpan(ManifestFileEntry data_object_, const std::vector<ManifestFile
 
 std::optional<ManifestFileEntry> SingleThreadIcebergKeysIterator::next()
 {
+    ProfileEventTimeIncrement<Microseconds> watch(ProfileEvents::IcebergIteratorNextMicroseconds);
+
     if (!data_snapshot)
     {
         return std::nullopt;

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

@@ -100,6 +100,8 @@ class IcebergIterator : public IObjectIterator
     size_t estimatedKeysCount() override;
     ~IcebergIterator() override;
 
+    bool has_concurrent_next() const override { return true; }
+
 private:
     std::unique_ptr<ActionsDAG> filter_dag;
     ObjectStoragePtr object_storage;

src/Storages/ObjectStorage/IObjectIterator.h

@@ -21,6 +21,7 @@ struct IObjectIterator
     virtual ObjectInfoPtr next(size_t) = 0;
     virtual size_t estimatedKeysCount() = 0;
     virtual std::optional<UInt64> getSnapshotVersion() const { return std::nullopt; }
+    virtual bool has_concurrent_next() const { return false; }
 };
 
 using ObjectIterator = std::shared_ptr<IObjectIterator>;

src/Storages/ObjectStorage/StorageObjectStorageStableTaskDistributor.cpp

@@ -154,6 +154,19 @@ ObjectInfoPtr StorageObjectStorageStableTaskDistributor::getMatchingFileFromIter
     {
         ObjectInfoPtr object_info;
 
+        if (iterator->has_concurrent_next())
+        {
+            object_info = iterator->next(0);
+
+            if (!object_info)
+            {
+                LOG_TEST(log, "Iterator is exhausted");
+                std::lock_guard lock(mutex);
+                iterator_exhausted = true;
+                break;
+            }
+        }
+        else
         {
             std::lock_guard lock(mutex);
             object_info = iterator->next(0);