Add object_metadata

Author: kssenii

src/Storages/ObjectStorage/DataLakes/DeltaLakeMetadata.cpp

@@ -600,6 +600,7 @@ DeltaLakeMetadata::DeltaLakeMetadata(ObjectStoragePtr object_storage_, Configura
     data_files = result.data_files;
     schema = result.schema;
     partition_columns = result.partition_columns;
+    object_storage = object_storage_;
 
     LOG_TRACE(impl.log, "Found {} data files, {} partition files, schema: {}",
              data_files.size(), partition_columns.size(), schema.toString());
@@ -712,6 +713,14 @@ Field DeltaLakeMetadata::getFieldValue(const String & value, DataTypePtr data_ty
     throw Exception(ErrorCodes::BAD_ARGUMENTS, "Unsupported DeltaLake type for {}", check_type->getColumnType());
 }
 
+ObjectIterator DeltaLakeMetadata::iterate(
+    const ActionsDAG * filter_dag,
+    FileProgressCallback callback,
+    size_t /* list_batch_size */) const
+{
+    return createKeysIterator(getDataFiles(filter_dag), object_storage, callback);
+}
+
 }
 
 #endif

src/Storages/ObjectStorage/DataLakes/DeltaLakeMetadata.h

@@ -79,12 +79,18 @@ class DeltaLakeMetadata final : public IDataLakeMetadata
     static Field getFieldValue(const String & value, DataTypePtr data_type);
 
 protected:
-    Strings getDataFiles(const ActionsDAG *) const override { return data_files; }
+    ObjectIterator iterate(
+        const ActionsDAG * filter_dag,
+        FileProgressCallback callback,
+        size_t list_batch_size) const override;
 
 private:
     mutable Strings data_files;
     NamesAndTypesList schema;
     DeltaLakePartitionColumns partition_columns;
+    ObjectStoragePtr object_storage;
+
+    Strings getDataFiles(const ActionsDAG *) const { return data_files; }
 };
 
 }

src/Storages/ObjectStorage/DataLakes/HudiMetadata.cpp

@@ -98,4 +98,12 @@ Strings HudiMetadata::getDataFiles(const ActionsDAG *) const
     return data_files;
 }
 
+ObjectIterator HudiMetadata::iterate(
+    const ActionsDAG * filter_dag,
+    FileProgressCallback callback,
+    size_t /* list_batch_size */) const
+{
+    return createKeysIterator(getDataFiles(filter_dag), object_storage, callback);
+}
+
 }

src/Storages/ObjectStorage/DataLakes/HudiMetadata.h

@@ -38,14 +38,18 @@ class HudiMetadata final : public IDataLakeMetadata, private WithContext
     }
 
 protected:
-    Strings getDataFiles(const ActionsDAG * filter_dag) const override;
+    ObjectIterator iterate(
+        const ActionsDAG * filter_dag,
+        FileProgressCallback callback,
+        size_t list_batch_size) const override;
 
 private:
     const ObjectStoragePtr object_storage;
     const ConfigurationObserverPtr configuration;
     mutable Strings data_files;
 
     Strings getDataFilesImpl() const;
+    Strings getDataFiles(const ActionsDAG * filter_dag) const;
 };
 
 }

src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.cpp

@@ -12,8 +12,10 @@ class KeysIterator : public IObjectIterator
 public:
     KeysIterator(
         Strings && data_files_,
+        ObjectStoragePtr object_storage_,
         IDataLakeMetadata::FileProgressCallback callback_)
         : data_files(data_files_)
+        , object_storage(object_storage_)
         , callback(callback_)
     {
     }
@@ -32,35 +34,30 @@ class KeysIterator : public IObjectIterator
                 return nullptr;
 
             auto key = data_files[current_index];
-            ObjectMetadata object_metadata;
-            // auto object_metadata = object_storage->getObjectMetadata(key);
+            auto object_metadata = object_storage->getObjectMetadata(key);
 
-            // if (callback)
-            //     callback(FileProgress(0, object_metadata.size_bytes));
+            if (callback)
+                callback(FileProgress(0, object_metadata.size_bytes));
 
             return std::make_shared<ObjectInfo>(key, std::move(object_metadata));
         }
     }
 
 private:
     Strings data_files;
+    ObjectStoragePtr object_storage;
     std::atomic<size_t> index = 0;
     IDataLakeMetadata::FileProgressCallback callback;
 };
 
 }
 
-ObjectIterator IDataLakeMetadata::iterate(
-    const ActionsDAG * filter_dag,
-    FileProgressCallback callback,
-    size_t /* list_batch_size */) const
+ObjectIterator IDataLakeMetadata::createKeysIterator(
+    Strings && data_files_,
+    ObjectStoragePtr object_storage_,
+    IDataLakeMetadata::FileProgressCallback callback_) const
 {
-    return std::make_shared<KeysIterator>(getDataFiles(filter_dag), callback);
-}
-
-Strings IDataLakeMetadata::getDataFiles(const ActionsDAG * /* filter_dag */) const
-{
-    throwNotImplemented("getDataFiles");
+    return std::make_shared<KeysIterator>(std::move(data_files_), object_storage_, callback_);
 }
 
 }

src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.h

@@ -25,7 +25,7 @@ class IDataLakeMetadata : boost::noncopyable
     virtual ObjectIterator iterate(
         const ActionsDAG * /* filter_dag */,
         FileProgressCallback /* callback */,
-        size_t /* list_batch_size */) const;
+        size_t /* list_batch_size */) const = 0;
 
     /// Table schema from data lake metadata.
     virtual NamesAndTypesList getTableSchema() const = 0;
@@ -50,9 +50,10 @@ class IDataLakeMetadata : boost::noncopyable
     virtual std::optional<size_t> totalBytes() const { return {}; }
 
 protected:
-    /// List all data files.
-    /// For better parallelization, iterate() method should be used.
-    virtual Strings getDataFiles(const ActionsDAG * filter_dag) const;
+    ObjectIterator createKeysIterator(
+        Strings && data_files_,
+        ObjectStoragePtr object_storage_,
+        IDataLakeMetadata::FileProgressCallback callback_) const;
 
     [[noreturn]] void throwNotImplemented(std::string_view method) const
     {

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

@@ -660,6 +660,14 @@ std::optional<size_t> IcebergMetadata::totalBytes() const
     return result;
 }
 
+ObjectIterator IcebergMetadata::iterate(
+    const ActionsDAG * filter_dag,
+    FileProgressCallback callback,
+    size_t /* list_batch_size */) const
+{
+    return createKeysIterator(getDataFiles(filter_dag), object_storage, callback);
+}
+
 }
 
 #endif

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

@@ -84,7 +84,10 @@ class IcebergMetadata : public IDataLakeMetadata, private WithContext
     std::optional<size_t> totalBytes() const override;
 
 protected:
-    Strings getDataFiles(const ActionsDAG * filter_dag) const override;
+    ObjectIterator iterate(
+        const ActionsDAG * filter_dag,
+        FileProgressCallback callback,
+        size_t list_batch_size) const override;
 
 private:
     using ManifestEntryByDataFile = std::unordered_map<String, Iceberg::ManifestFilePtr>;
@@ -114,6 +117,8 @@ class IcebergMetadata : public IDataLakeMetadata, private WithContext
 
     mutable std::optional<Strings> cached_unprunned_files_for_last_processed_snapshot;
 
+    Strings getDataFiles(const ActionsDAG * filter_dag) const;
+
     void updateState(const ContextPtr & local_context);
 
     void updateSnapshot();