fix comment

Author: xiao.dong

src/iceberg/manifest_writer.cc

@@ -27,11 +27,12 @@
 
 namespace iceberg {
 
-Result<std::unique_ptr<ManifestWriter>> ManifestWriter::MakeWriter(
-    int32_t format_version, int64_t snapshot_id, int64_t first_row_id,
+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(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);
@@ -48,17 +49,21 @@ Result<std::unique_ptr<ManifestWriter>> ManifestWriter::MakeWriter(
       return std::make_unique<ManifestWriterV2>(snapshot_id, std::move(writer),
                                                 std::move(schema));
     case 3:
-      return std::make_unique<ManifestWriterV3>(snapshot_id, first_row_id,
+      // 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 InvalidArgument("Unsupported manifest format version: {}", format_version);
+      return NotSupported("Unsupported manifest format version: {}", format_version);
   }
 }
 
-Result<std::unique_ptr<ManifestListWriter>> ManifestListWriter::MakeWriter(
+Result<std::unique_ptr<ManifestListWriter>> ManifestListWriter::Make(
     int32_t format_version, int64_t snapshot_id, int64_t parent_snapshot_id,
-    int64_t sequence_number, int64_t first_row_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());
@@ -71,20 +76,25 @@ Result<std::unique_ptr<ManifestListWriter>> ManifestListWriter::MakeWriter(
   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, std::move(writer),
-                                                    std::move(schema));
-    case 3:
-      return std::make_unique<ManifestListWriterV3>(snapshot_id, parent_snapshot_id,
-                                                    sequence_number, first_row_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 InvalidArgument("Unsupported manifest list format version: {}",
-                             format_version);
+      return NotSupported("Unsupported manifest list format version: {}", format_version);
   }
 }
 

src/iceberg/manifest_writer.h

@@ -36,10 +36,15 @@ class ICEBERG_EXPORT ManifestWriter {
  public:
   virtual ~ManifestWriter() = default;
 
-  /// \brief Write manifest entry to file
+  /// \brief Write manifest entry to file.
   /// \param entry Manifest entry to write.
-  /// \return Status::OK() if all entry was written successfully
-  virtual Status WriteManifestEntry(const ManifestEntry& entry) const = 0;
+  /// \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;
@@ -51,8 +56,8 @@ class ICEBERG_EXPORT ManifestWriter {
   /// \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(
-      int32_t format_version, int64_t snapshot_id, int64_t first_row_id,
+  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);
 };
@@ -62,10 +67,15 @@ class ICEBERG_EXPORT ManifestListWriter {
  public:
   virtual ~ManifestListWriter() = default;
 
-  /// \brief Write manifest file list to manifest list file.
+  /// \brief Write manifest file to manifest list file.
   /// \param file Manifest file to write.
-  /// \return Status::OK() if all file was written successfully
-  virtual Status WriteManifestFile(const ManifestFile& file) const = 0;
+  /// \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;
@@ -79,9 +89,9 @@ class ICEBERG_EXPORT ManifestListWriter {
   /// \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,
-      int64_t sequence_number, int64_t first_row_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

@@ -25,45 +25,105 @@
 
 namespace iceberg {
 
-Status ManifestWriterV1::WriteManifestEntry(const ManifestEntry& entry) const {
+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 {}; }
 
-Status ManifestWriterV2::WriteManifestEntry(const ManifestEntry& entry) const {
+ManifestEntry ManifestWriterV1::prepare(const ManifestEntry& entry) {
+  return 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 {}; }
 
-Status ManifestWriterV3::WriteManifestEntry(const ManifestEntry& entry) const {
+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 {}; }
 
-Status ManifestListWriterV1::WriteManifestFile(const ManifestFile& file) const {
+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 {}; }
 
-Status ManifestListWriterV2::WriteManifestFile(const ManifestFile& file) const {
+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 {}; }
 
-Status ManifestListWriterV3::WriteManifestFile(const ManifestFile& file) const {
+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

src/iceberg/manifest_writer_internal.h

@@ -36,6 +36,9 @@ class ManifestWriterImpl : public ManifestWriter {
                               std::shared_ptr<Schema> schema)
       : schema_(std::move(schema)), writer_(std::move(writer)) {}
 
+ protected:
+  virtual ManifestEntry prepare(const ManifestEntry& entry) = 0;
+
  private:
   std::shared_ptr<Schema> schema_;
   std::unique_ptr<Writer> writer_;
@@ -48,10 +51,13 @@ class ManifestWriterV1 : public ManifestWriterImpl {
                             std::shared_ptr<Schema> schema)
       : ManifestWriterImpl(snapshot_id, std::move(writer), std::move(schema)) {}
 
-  Status WriteManifestEntry(const ManifestEntry& entry) const override;
-
+  Status Add(const ManifestEntry& entry) override;
+  Status AddAll(const std::vector<ManifestEntry>& entries) override;
   Status Close() override;
 
+ protected:
+  ManifestEntry prepare(const ManifestEntry& entry) override;
+
  private:
   V1MetaData::ManifestEntryWrapper wrapper_;
 };
@@ -64,10 +70,14 @@ class ManifestWriterV2 : public ManifestWriterImpl {
       : ManifestWriterImpl(snapshot_id, std::move(writer), std::move(schema)),
         wrapper_(snapshot_id) {}
 
-  Status WriteManifestEntry(const ManifestEntry& entry) const override;
+  Status Add(const ManifestEntry& entry) override;
+  Status AddAll(const std::vector<ManifestEntry>& entries) override;
 
   Status Close() override;
 
+ protected:
+  ManifestEntry prepare(const ManifestEntry& entry) override;
+
  private:
   V2MetaData::ManifestEntryWrapper wrapper_;
 };
@@ -81,10 +91,14 @@ class ManifestWriterV3 : public ManifestWriterImpl {
       : ManifestWriterImpl(snapshot_id, std::move(writer), std::move(schema)),
         wrapper_(snapshot_id) {}
 
-  Status WriteManifestEntry(const ManifestEntry& entry) const override;
+  Status Add(const ManifestEntry& entry) override;
+  Status AddAll(const std::vector<ManifestEntry>& entries) override;
 
   Status Close() override;
 
+ protected:
+  ManifestEntry prepare(const ManifestEntry& entry) override;
+
  private:
   V3MetaData::ManifestEntryWrapper wrapper_;
 };
@@ -97,6 +111,9 @@ class ManifestListWriterImpl : public ManifestListWriter {
                                   std::shared_ptr<Schema> schema)
       : schema_(std::move(schema)), writer_(std::move(writer)) {}
 
+ protected:
+  virtual ManifestFile prepare(const ManifestFile& file) = 0;
+
  private:
   std::shared_ptr<Schema> schema_;
   std::unique_ptr<Writer> writer_;
@@ -112,10 +129,13 @@ class ManifestListWriterV1 : public ManifestListWriterImpl {
       : ManifestListWriterImpl(snapshot_id, parent_snapshot_id, std::move(writer),
                                std::move(schema)) {}
 
-  Status WriteManifestFile(const ManifestFile& file) const override;
-
+  Status Add(const ManifestFile& file) override;
+  Status AddAll(const std::vector<ManifestFile>& files) override;
   Status Close() override;
 
+ protected:
+  ManifestFile prepare(const ManifestFile& file) override;
+
  private:
   V1MetaData::ManifestFileWrapper wrapper_;
 };
@@ -130,10 +150,14 @@ class ManifestListWriterV2 : public ManifestListWriterImpl {
                                std::move(schema)),
         wrapper_(snapshot_id, sequence_number) {}
 
-  Status WriteManifestFile(const ManifestFile& file) const override;
+  Status Add(const ManifestFile& file) override;
+  Status AddAll(const std::vector<ManifestFile>& files) override;
 
   Status Close() override;
 
+ protected:
+  ManifestFile prepare(const ManifestFile& file) override;
+
  private:
   V2MetaData::ManifestFileWrapper wrapper_;
 };
@@ -149,11 +173,16 @@ class ManifestListWriterV3 : public ManifestListWriterImpl {
                                std::move(schema)),
         wrapper_(snapshot_id, sequence_number) {}
 
-  Status WriteManifestFile(const ManifestFile& file) const override;
+  Status Add(const ManifestFile& file) override;
+  Status AddAll(const std::vector<ManifestFile>& files) override;
 
   Status Close() override;
 
+ protected:
+  ManifestFile prepare(const ManifestFile& file) override;
+
  private:
+  int64_t next_row_id_ = 0;
   V3MetaData::ManifestFileWrapper wrapper_;
 };
 

src/iceberg/v1_metadata.h

@@ -35,14 +35,14 @@ class V1MetaData {
   struct ManifestFileWrapper : public ManifestFile {
     ManifestFileWrapper() = default;
 
-    ManifestFile wrap(ManifestFile file, int64_t first_row_id) { return *this; }
+    ManifestFile Wrap(ManifestFile file) { return *this; }
   };
 
   /// \brief v1 manifest entry wrapper.
   struct ManifestEntryWrapper : public ManifestEntry {
     ManifestEntryWrapper() = default;
 
-    ManifestEntry wrap(ManifestEntry entry) { return *this; }
+    ManifestEntry Wrap(ManifestEntry entry) { return *this; }
   };
 
   static ManifestFileWrapper manifestFileWrapper() { return {}; }