feat: complete v1_metadata

src/iceberg/CMakeLists.txt

@@ -47,6 +47,8 @@ set(ICEBERG_SOURCES
     type.cc
     manifest_reader.cc
     manifest_reader_internal.cc
+    manifest_writer.cc
+    manifest_writer_internal.cc
     arrow_c_data_guard_internal.cc
     util/murmurhash3_internal.cc
     util/timepoint.cc

src/iceberg/manifest_reader.cc

@@ -27,7 +27,7 @@
 
 namespace iceberg {
 
-Result<std::unique_ptr<ManifestReader>> ManifestReader::MakeReader(
+Result<std::unique_ptr<ManifestReader>> ManifestReader::Make(
     std::string_view manifest_location, std::shared_ptr<FileIO> file_io,
     std::shared_ptr<Schema> partition_schema) {
   auto manifest_entry_schema = ManifestEntry::TypeFromPartitionType(partition_schema);
@@ -42,7 +42,7 @@ Result<std::unique_ptr<ManifestReader>> ManifestReader::MakeReader(
   return std::make_unique<ManifestReaderImpl>(std::move(reader), std::move(schema));
 }
 
-Result<std::unique_ptr<ManifestListReader>> ManifestListReader::MakeReader(
+Result<std::unique_ptr<ManifestListReader>> ManifestListReader::Make(
     std::string_view manifest_list_location, std::shared_ptr<FileIO> file_io) {
   std::vector<SchemaField> fields(ManifestFile::Type().fields().begin(),
                                   ManifestFile::Type().fields().end());

src/iceberg/manifest_reader.h

@@ -41,7 +41,7 @@ class ICEBERG_EXPORT ManifestReader {
   /// \param manifest_location Path to the manifest file.
   /// \param file_io File IO implementation to use.
   /// \return A Result containing the reader or an error.
-  static Result<std::unique_ptr<ManifestReader>> MakeReader(
+  static Result<std::unique_ptr<ManifestReader>> Make(
       std::string_view manifest_location, std::shared_ptr<FileIO> file_io,
       std::shared_ptr<Schema> partition_schema);
 };
@@ -56,7 +56,7 @@ class ICEBERG_EXPORT ManifestListReader {
   /// \param manifest_list_location Path to the manifest list file.
   /// \param file_io File IO implementation to use.
   /// \return A Result containing the reader or an error.
-  static Result<std::unique_ptr<ManifestListReader>> MakeReader(
+  static Result<std::unique_ptr<ManifestListReader>> Make(
       std::string_view manifest_list_location, std::shared_ptr<FileIO> file_io);
 };
 

src/iceberg/manifest_writer.cc

@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/manifest_writer.h"
+
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/manifest_writer_internal.h"
+#include "iceberg/schema.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+Result<std::unique_ptr<ManifestWriter>> ManifestWriter::Make(
+    int32_t format_version, int64_t snapshot_id, std::optional<int64_t> first_row_id,
+    std::string_view manifest_location, std::shared_ptr<FileIO> file_io,
+    std::shared_ptr<Schema> partition_schema) {
+  auto manifest_entry_schema =
+      ManifestEntry::TypeFromPartitionType(std::move(partition_schema));
+  auto fields_span = manifest_entry_schema->fields();
+  std::vector<SchemaField> fields(fields_span.begin(), fields_span.end());
+  auto schema = std::make_shared<Schema>(fields);
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto writer, WriterFactoryRegistry::Open(FileFormatType::kAvro,
+                                               {.path = std::string(manifest_location),
+                                                .schema = schema,
+                                                .io = std::move(file_io)}));
+  switch (format_version) {
+    case 1:
+      return std::make_unique<ManifestWriterV1>(snapshot_id, std::move(writer),
+                                                std::move(schema));
+    case 2:
+      return std::make_unique<ManifestWriterV2>(snapshot_id, std::move(writer),
+                                                std::move(schema));
+    case 3:
+      // first_row_id is required for V3 manifest entry
+      if (!first_row_id.has_value()) {
+        return InvalidManifest("first_row_id is required for V3 manifest entry");
+      }
+      return std::make_unique<ManifestWriterV3>(snapshot_id, first_row_id.value(),
+                                                std::move(writer), std::move(schema));
+
+    default:
+      return NotSupported("Unsupported manifest format version: {}", format_version);
+  }
+}
+
+Result<std::unique_ptr<ManifestListWriter>> ManifestListWriter::Make(
+    int32_t format_version, int64_t snapshot_id, int64_t parent_snapshot_id,
+    std::optional<int64_t> sequence_number, std::optional<int64_t> first_row_id,
+    std::string_view manifest_list_location, std::shared_ptr<FileIO> file_io) {
+  std::vector<SchemaField> fields(ManifestFile::Type().fields().begin(),
+                                  ManifestFile::Type().fields().end());
+  auto schema = std::make_shared<Schema>(fields);
+  ICEBERG_ASSIGN_OR_RAISE(auto writer, WriterFactoryRegistry::Open(
+                                           FileFormatType::kAvro,
+                                           {.path = std::string(manifest_list_location),
+                                            .schema = schema,
+                                            .io = std::move(file_io)}));
+  switch (format_version) {
+    case 1:
+      return std::make_unique<ManifestListWriterV1>(snapshot_id, parent_snapshot_id,
+                                                    std::move(writer), std::move(schema));
+    case 2:
+      return std::make_unique<ManifestListWriterV2>(snapshot_id, parent_snapshot_id,
+                                                    sequence_number.value(),
+                                                    std::move(writer), std::move(schema));
+    case 3:
+      // sequence_number&first_row_id is required for V3 manifest list
+      if (!sequence_number.has_value()) {
+        return InvalidManifestList("sequence_number is required for V3 manifest list");
+      }
+      if (!first_row_id.has_value()) {
+        return InvalidManifestList("first_row_id is required for V3 manifest list");
+      }
+      return std::make_unique<ManifestListWriterV3>(
+          snapshot_id, parent_snapshot_id, sequence_number.value(), first_row_id.value(),
+          std::move(writer), std::move(schema));
+
+    default:
+      return NotSupported("Unsupported manifest list format version: {}", format_version);
+  }
+}
+
+}  // namespace iceberg

src/iceberg/manifest_writer.h

@@ -35,14 +35,29 @@ namespace iceberg {
 class ICEBERG_EXPORT ManifestWriter {
  public:
   virtual ~ManifestWriter() = default;
-  virtual Status WriteManifestEntries(
-      const std::vector<ManifestEntry>& entries) const = 0;
+
+  /// \brief Write manifest entry to file.
+  /// \param entry Manifest entry to write.
+  /// \return Status::OK() if entry was written successfully
+  virtual Status Add(const ManifestEntry& entry) = 0;
+
+  /// \brief Write manifest entries to file.
+  /// \param entries Manifest entries to write.
+  /// \return Status::OK() if all entries were written successfully
+  virtual Status AddAll(const std::vector<ManifestEntry>& entries) = 0;
+
+  /// \brief Close writer and flush to storage.
+  virtual Status Close() = 0;
 
   /// \brief Creates a writer for a manifest file.
+  /// \param format_version Format version of the manifest.
+  /// \param snapshot_id ID of the snapshot.
+  /// \param first_row_id First row ID of the snapshot.
   /// \param manifest_location Path to the manifest file.
   /// \param file_io File IO implementation to use.
   /// \return A Result containing the writer or an error.
-  static Result<std::unique_ptr<ManifestWriter>> MakeWriter(
+  static Result<std::unique_ptr<ManifestWriter>> Make(
+      int32_t format_version, int64_t snapshot_id, std::optional<int64_t> first_row_id,
       std::string_view manifest_location, std::shared_ptr<FileIO> file_io,
       std::shared_ptr<Schema> partition_schema);
 };
@@ -51,13 +66,32 @@ class ICEBERG_EXPORT ManifestWriter {
 class ICEBERG_EXPORT ManifestListWriter {
  public:
   virtual ~ManifestListWriter() = default;
-  virtual Status WriteManifestFiles(const std::vector<ManifestFile>& files) const = 0;
+
+  /// \brief Write manifest file to manifest list file.
+  /// \param file Manifest file to write.
+  /// \return Status::OK() if file was written successfully
+  virtual Status Add(const ManifestFile& file) = 0;
+
+  /// \brief Write manifest file list to manifest list file.
+  /// \param files Manifest file list to write.
+  /// \return Status::OK() if all files were written successfully
+  virtual Status AddAll(const std::vector<ManifestFile>& files) = 0;
+
+  /// \brief Close writer and flush to storage.
+  virtual Status Close() = 0;
 
   /// \brief Creates a writer for the manifest list.
+  /// \param format_version Format version of the manifest list.
+  /// \param snapshot_id ID of the snapshot.
+  /// \param parent_snapshot_id ID of the parent snapshot.
+  /// \param sequence_number Sequence number of the snapshot.
+  /// \param first_row_id First row ID of the snapshot.
   /// \param manifest_list_location Path to the manifest list file.
   /// \param file_io File IO implementation to use.
   /// \return A Result containing the writer or an error.
-  static Result<std::unique_ptr<ManifestListWriter>> MakeWriter(
+  static Result<std::unique_ptr<ManifestListWriter>> Make(
+      int32_t format_version, int64_t snapshot_id, int64_t parent_snapshot_id,
+      std::optional<int64_t> sequence_number, std::optional<int64_t> first_row_id,
       std::string_view manifest_list_location, std::shared_ptr<FileIO> file_io);
 };
 

src/iceberg/manifest_writer_internal.cc

@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "manifest_writer_internal.h"
+
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/v1_metadata.h"
+
+namespace iceberg {
+
+Status ManifestWriterV1::Add(const ManifestEntry& entry) {
+  // TODO(xiao.dong) convert entries to arrow data
+  return {};
+}
+
+Status ManifestWriterV1::AddAll(const std::vector<ManifestEntry>& files) {
+  // TODO(xiao.dong) convert entries to arrow data
+  return {};
+}
+
+Status ManifestWriterV1::Close() { return {}; }
+
+ManifestEntry ManifestWriterV1::prepare(const ManifestEntry& entry) {
+  return static_cast<ManifestEntry>(wrapper_.Wrap(entry));
+}
+
+Status ManifestWriterV2::Add(const ManifestEntry& entry) {
+  // TODO(xiao.dong) convert entries to arrow data
+  return {};
+}
+
+Status ManifestWriterV2::AddAll(const std::vector<ManifestEntry>& files) {
+  // TODO(xiao.dong) convert entries to arrow data
+  return {};
+}
+
+Status ManifestWriterV2::Close() { return {}; }
+
+ManifestEntry ManifestWriterV2::prepare(const ManifestEntry& entry) {
+  return wrapper_.Wrap(entry);
+}
+
+Status ManifestWriterV3::Add(const ManifestEntry& entry) {
+  // TODO(xiao.dong) convert entries to arrow data
+  return {};
+}
+
+Status ManifestWriterV3::AddAll(const std::vector<ManifestEntry>& files) {
+  // TODO(xiao.dong) convert entries to arrow data
+  return {};
+}
+
+Status ManifestWriterV3::Close() { return {}; }
+
+ManifestEntry ManifestWriterV3::prepare(const ManifestEntry& entry) {
+  return wrapper_.Wrap(entry);
+}
+
+Status ManifestListWriterV1::Add(const ManifestFile& file) {
+  // TODO(xiao.dong) convert manifest files to arrow data
+  return {};
+}
+
+Status ManifestListWriterV1::AddAll(const std::vector<ManifestFile>& files) {
+  // TODO(xiao.dong) convert manifest files to arrow data
+  return {};
+}
+
+Status ManifestListWriterV1::Close() { return {}; }
+
+ManifestFile ManifestListWriterV1::prepare(const ManifestFile& file) {
+  return wrapper_.Wrap(file);
+}
+
+Status ManifestListWriterV2::Add(const ManifestFile& file) {
+  // TODO(xiao.dong) convert manifest files to arrow data
+  return {};
+}
+
+Status ManifestListWriterV2::AddAll(const std::vector<ManifestFile>& files) {
+  // TODO(xiao.dong) convert manifest files to arrow data
+  return {};
+}
+
+Status ManifestListWriterV2::Close() { return {}; }
+
+ManifestFile ManifestListWriterV2::prepare(const ManifestFile& file) {
+  return wrapper_.Wrap(file);
+}
+
+Status ManifestListWriterV3::Add(const ManifestFile& file) {
+  // TODO(xiao.dong) convert manifest files to arrow data
+  return {};
+}
+
+Status ManifestListWriterV3::AddAll(const std::vector<ManifestFile>& files) {
+  // TODO(xiao.dong) convert manifest files to arrow data
+  return {};
+}
+
+Status ManifestListWriterV3::Close() { return {}; }
+
+ManifestFile ManifestListWriterV3::prepare(const ManifestFile& file) {
+  if (file.content != ManifestFile::Content::kData || file.first_row_id.has_value()) {
+    return wrapper_.Wrap(file, std::nullopt);
+  }
+  auto result = wrapper_.Wrap(file, next_row_id_);
+  next_row_id_ +=
+      file.existing_rows_count.value_or(0) + file.added_rows_count.value_or(0);
+  return result;
+}
+}  // namespace iceberg