feat: add Metadata method to Reader and its implementations

- Introduced a new virtual method `Metadata()` in the `Reader` class to retrieve file metadata.
- Implemented `Metadata()` in `AvroReader` to return key-value pairs from the Avro file's metadata.
- Implemented `Metadata()` in `ParquetReader` to extract and return key-value pairs from the Parquet file's metadata.

Author: nullccxsy

src/iceberg/avro/avro_reader.cc

@@ -173,6 +173,24 @@ class AvroReader::Impl {
     return arrow_schema;
   }
 
+  Result<std::unordered_map<std::string, std::string>> Metadata() {
+    if (reader_ == nullptr) {
+      return InvalidArgument("Reader is not opened");
+    }
+
+    auto metadata = reader_->metadata();
+
+    std::unordered_map<std::string, std::string> metadata_map;
+    metadata_map.reserve(metadata.size());
+
+    for (const auto& pair : metadata) {
+      metadata_map.try_emplace(pair.first,
+                               std::string(pair.second.begin(), pair.second.end()));
+    }
+
+    return metadata_map;
+  }
+
  private:
   Status InitReadContext() {
     context_ = std::make_unique<ReadContext>();
@@ -241,6 +259,10 @@ Result<std::optional<ArrowArray>> AvroReader::Next() { return impl_->Next(); }
 
 Result<ArrowSchema> AvroReader::Schema() { return impl_->Schema(); }
 
+Result<std::unordered_map<std::string, std::string>> AvroReader::Metadata() {
+  return impl_->Metadata();
+}
+
 Status AvroReader::Open(const ReaderOptions& options) {
   impl_ = std::make_unique<Impl>();
   return impl_->Open(options);

src/iceberg/avro/avro_reader.h

@@ -39,6 +39,8 @@ class ICEBERG_BUNDLE_EXPORT AvroReader : public Reader {
 
   Result<ArrowSchema> Schema() final;
 
+  Result<std::unordered_map<std::string, std::string>> Metadata() final;
+
  private:
   class Impl;
   std::unique_ptr<Impl> impl_;

src/iceberg/file_reader.h

@@ -54,6 +54,9 @@ class ICEBERG_EXPORT Reader {
 
   /// \brief Get the schema of the data.
   virtual Result<ArrowSchema> Schema() = 0;
+
+  /// \brief Get the metadata of the file
+  virtual Result<std::unordered_map<std::string, std::string>> Metadata() = 0;
 };
 
 /// \brief A split of the file to read.

src/iceberg/parquet/parquet_reader.cc

@@ -26,6 +26,7 @@
 #include <arrow/record_batch.h>
 #include <arrow/result.h>
 #include <arrow/type.h>
+#include <arrow/util/key_value_metadata.h>
 #include <parquet/arrow/reader.h>
 #include <parquet/arrow/schema.h>
 #include <parquet/file_reader.h>
@@ -185,6 +186,31 @@ class ParquetReader::Impl {
     return arrow_schema;
   }
 
+  Result<std::unordered_map<std::string, std::string>> Metadata() {
+    if (reader_ == nullptr) {
+      return InvalidArgument("Reader is not opened");
+    }
+
+    auto metadata = reader_->parquet_reader()->metadata();
+    if (!metadata) {
+      return InvalidArgument("Failed to get Parquet file metadata");
+    }
+
+    auto kv_metadata = metadata->key_value_metadata();
+    if (!kv_metadata) {
+      return std::unordered_map<std::string, std::string>{};
+    }
+
+    std::unordered_map<std::string, std::string> metadata_map;
+    metadata_map.reserve(kv_metadata->size());
+
+    for (int i = 0; i < kv_metadata->size(); ++i) {
+      metadata_map.try_emplace(kv_metadata->key(i), kv_metadata->value(i));
+    }
+
+    return metadata_map;
+  }
+
  private:
   Status InitReadContext() {
     context_ = std::make_unique<ReadContext>();
@@ -251,6 +277,10 @@ Result<std::optional<ArrowArray>> ParquetReader::Next() { return impl_->Next();
 
 Result<ArrowSchema> ParquetReader::Schema() { return impl_->Schema(); }
 
+Result<std::unordered_map<std::string, std::string>> ParquetReader::Metadata() {
+  return impl_->Metadata();
+}
+
 Status ParquetReader::Open(const ReaderOptions& options) {
   impl_ = std::make_unique<Impl>();
   return impl_->Open(options);

src/iceberg/parquet/parquet_reader.h

@@ -39,6 +39,8 @@ class ICEBERG_BUNDLE_EXPORT ParquetReader : public Reader {
 
   Result<ArrowSchema> Schema() final;
 
+  Result<std::unordered_map<std::string, std::string>> Metadata() final;
+
  private:
   class Impl;
   std::unique_ptr<Impl> impl_;