feat: add support to write metadata for avro and parquet

Author: wgtmac

src/iceberg/avro/avro_writer.cc

@@ -75,7 +75,7 @@ class AvroWriter::Impl {
                             CreateOutputStream(options, kDefaultBufferSize));
     arrow_output_stream_ = output_stream->arrow_output_stream();
     std::map<std::string, std::vector<uint8_t>> metadata;
-    for (const auto& [key, value] : options.properties) {
+    for (const auto& [key, value] : options.metadata) {
       std::vector<uint8_t> vec;
       vec.reserve(value.size());
       vec.assign(value.begin(), value.end());

src/iceberg/file_reader.h

@@ -42,7 +42,7 @@ class ICEBERG_EXPORT Reader {
   Reader& operator=(const Reader&) = delete;
 
   /// \brief Open the reader.
-  virtual Status Open(const struct ReaderOptions& options) = 0;
+  virtual Status Open(const ReaderOptions& options) = 0;
 
   /// \brief Close the reader.
   virtual Status Close() = 0;
@@ -93,6 +93,7 @@ struct ICEBERG_EXPORT ReaderOptions {
   /// that may have different field names than the current schema.
   std::shared_ptr<class NameMapping> name_mapping;
   /// \brief Format-specific or implementation-specific properties.
+  /// TODO(gangwu): replace it with a new class inherited from `ConfigBase`.
   std::unordered_map<std::string, std::string> properties;
 };
 

src/iceberg/file_writer.h

@@ -44,7 +44,10 @@ struct ICEBERG_EXPORT WriterOptions {
   /// to the specific FileIO implementation. By default, the `iceberg-bundle` library uses
   /// `ArrowFileSystemFileIO` as the default implementation.
   std::shared_ptr<class FileIO> io;
+  /// \brief Metadata to write to the file.
+  std::unordered_map<std::string, std::string> metadata;
   /// \brief Format-specific or implementation-specific properties.
+  /// TODO(gangwu): replace it with a new class inherited from `ConfigBase`.
   std::unordered_map<std::string, std::string> properties;
 };
 
@@ -57,7 +60,7 @@ class ICEBERG_EXPORT Writer {
   Writer& operator=(const Writer&) = delete;
 
   /// \brief Open the writer.
-  virtual Status Open(const struct WriterOptions& options) = 0;
+  virtual Status Open(const WriterOptions& options) = 0;
 
   /// \brief Close the writer.
   virtual Status Close() = 0;

src/iceberg/parquet/parquet_writer.cc

@@ -68,7 +68,8 @@ class ParquetWriter::Impl {
 
     ICEBERG_ASSIGN_OR_RAISE(output_stream_, OpenOutputStream(options));
     auto file_writer = ::parquet::ParquetFileWriter::Open(
-        output_stream_, std::move(schema_node), std::move(writer_properties));
+        output_stream_, std::move(schema_node), std::move(writer_properties),
+        std::make_shared<::arrow::KeyValueMetadata>(options.metadata));
     ICEBERG_ARROW_RETURN_NOT_OK(
         ::parquet::arrow::FileWriter::Make(pool_, std::move(file_writer), arrow_schema_,
                                            std::move(arrow_writer_properties), &writer_));

src/iceberg/test/avro_test.cc

@@ -123,9 +123,13 @@ class AvroReaderTest : public TempFileTestBase {
     auto export_result = ::arrow::ExportArray(*array, &arrow_array);
     ASSERT_TRUE(export_result.ok());
 
-    auto writer_result = WriterFactoryRegistry::Open(
-        FileFormatType::kAvro,
-        {.path = temp_avro_file_, .schema = schema, .io = file_io_});
+    std::unordered_map<std::string, std::string> metadata = {{"k1", "v1"}, {"k2", "v2"}};
+
+    auto writer_result =
+        WriterFactoryRegistry::Open(FileFormatType::kAvro, {.path = temp_avro_file_,
+                                                            .schema = schema,
+                                                            .io = file_io_,
+                                                            .metadata = metadata});
     ASSERT_TRUE(writer_result.has_value());
     auto writer = std::move(writer_result.value());
     ASSERT_THAT(writer->Write(&arrow_array), IsOk());
@@ -144,6 +148,15 @@ class AvroReaderTest : public TempFileTestBase {
     auto reader = std::move(reader_result.value());
     ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*reader, expected_string));
     ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader));
+
+    auto metadata_result = reader->Metadata();
+    ASSERT_THAT(metadata_result, IsOk());
+    auto read_metadata = std::move(metadata_result.value());
+    for (const auto& [key, value] : metadata) {
+      auto it = read_metadata.find(key);
+      ASSERT_NE(it, read_metadata.end());
+      ASSERT_EQ(it->second, value);
+    }
   }
 
   std::shared_ptr<::arrow::fs::LocalFileSystem> local_fs_;

src/iceberg/test/parquet_test.cc

@@ -63,7 +63,8 @@ Status WriteArray(std::shared_ptr<::arrow::Array> data,
 }
 
 Status ReadArray(std::shared_ptr<::arrow::Array>& out,
-                 const ReaderOptions& reader_options) {
+                 const ReaderOptions& reader_options,
+                 std::unordered_map<std::string, std::string>* metadata) {
   ICEBERG_ASSIGN_OR_RAISE(
       auto reader, ReaderFactoryRegistry::Open(FileFormatType::kParquet, reader_options));
   ICEBERG_ASSIGN_OR_RAISE(auto read_data, reader->Next());
@@ -77,6 +78,11 @@ Status ReadArray(std::shared_ptr<::arrow::Array>& out,
   ICEBERG_ASSIGN_OR_RAISE(ArrowSchema arrow_schema, reader->Schema());
   ICEBERG_ARROW_ASSIGN_OR_RETURN(out,
                                  ::arrow::ImportArray(&arrow_c_array, &arrow_schema));
+
+  if (metadata) {
+    ICEBERG_ASSIGN_OR_RAISE(*metadata, reader->Metadata());
+  }
+
   return {};
 }
 
@@ -85,18 +91,25 @@ void DoRoundtrip(std::shared_ptr<::arrow::Array> data, std::shared_ptr<Schema> s
   std::shared_ptr<FileIO> file_io = arrow::ArrowFileSystemFileIO::MakeMockFileIO();
   const std::string basePath = "base.parquet";
 
+  std::unordered_map<std::string, std::string> metadata = {{"k1", "v1"}, {"k2", "v2"}};
+
   auto writer_data = WriterFactoryRegistry::Open(
-      FileFormatType::kParquet, {.path = basePath, .schema = schema, .io = file_io});
+      FileFormatType::kParquet,
+      {.path = basePath, .schema = schema, .io = file_io, .metadata = metadata});
   ASSERT_THAT(writer_data, IsOk())
       << "Failed to create writer: " << writer_data.error().message;
   auto writer = std::move(writer_data.value());
   ASSERT_THAT(WriteArray(data, *writer), IsOk());
 
-  ASSERT_THAT(ReadArray(out, {.path = basePath,
-                              .length = writer->length(),
-                              .io = file_io,
-                              .projection = schema}),
+  std::unordered_map<std::string, std::string> read_metadata;
+  ASSERT_THAT(ReadArray(out,
+                        {.path = basePath,
+                         .length = writer->length(),
+                         .io = file_io,
+                         .projection = schema},
+                        &read_metadata),
               IsOk());
+  ASSERT_EQ(read_metadata, metadata);
 
   ASSERT_TRUE(out != nullptr) << "Reader.Next() returned no data";
 }

src/iceberg/type_fwd.h

@@ -139,6 +139,8 @@ class ManifestListWriter;
 class ManifestReader;
 class ManifestWriter;
 
+struct ReaderOptions;
+struct WriterOptions;
 class Reader;
 class Writer;