fix: review comments

Author: zhjwpku

src/iceberg/avro/avro_writer.cc

@@ -154,19 +154,19 @@ Status AvroWriter::Close() {
   return {};
 }
 
-std::optional<Metrics> AvroWriter::metrics() {
+Result<Metrics> AvroWriter::metrics() {
   if (impl_->Closed()) {
     // TODO(xiao.dong) implement metrics
     return {};
   }
-  return std::nullopt;
+  return Invalid("AvroWriter is not closed");
 }
 
-std::optional<int64_t> AvroWriter::length() {
+Result<int64_t> AvroWriter::length() {
   if (impl_->Closed()) {
     return impl_->length();
   }
-  return std::nullopt;
+  return Invalid("AvroWriter is not closed");
 }
 
 std::vector<int64_t> AvroWriter::split_offsets() { return {}; }

src/iceberg/avro/avro_writer.h

@@ -37,9 +37,9 @@ class ICEBERG_BUNDLE_EXPORT AvroWriter : public Writer {
 
   Status Write(ArrowArray* data) final;
 
-  std::optional<Metrics> metrics() final;
+  Result<Metrics> metrics() final;
 
-  std::optional<int64_t> length() final;
+  Result<int64_t> length() final;
 
   std::vector<int64_t> split_offsets() final;
 

src/iceberg/file_writer.h

@@ -97,11 +97,11 @@ class ICEBERG_EXPORT Writer {
 
   /// \brief Get the file statistics.
   /// Only valid after the file is closed.
-  virtual std::optional<Metrics> metrics() = 0;
+  virtual Result<Metrics> metrics() = 0;
 
   /// \brief Get the file length.
   /// Only valid after the file is closed.
-  virtual std::optional<int64_t> length() = 0;
+  virtual Result<int64_t> length() = 0;
 
   /// \brief Returns a list of recommended split locations, if applicable, empty
   /// otherwise. When available, this information is used for planning scan tasks whose

src/iceberg/manifest_adapter.cc

@@ -164,32 +164,7 @@ ManifestEntryAdapter::~ManifestEntryAdapter() {
   }
 }
 
-Status ManifestEntryAdapter::AddEntry(ManifestEntry& entry) {
-  ICEBERG_RETURN_UNEXPECTED(CheckDataFile(*entry.data_file));
-  entry.status = ManifestStatus::kAdded;
-  entry.snapshot_id = snapshot_id_;
-  if (entry.sequence_number.has_value() &&
-      entry.sequence_number.value() < TableMetadata::kInitialSequenceNumber) {
-    entry.sequence_number = std::nullopt;
-  }
-  entry.file_sequence_number = std::nullopt;
-  return AddEntryInternal(entry);
-}
-
-Status ManifestEntryAdapter::AddDeleteEntry(ManifestEntry& entry) {
-  ICEBERG_RETURN_UNEXPECTED(CheckDataFile(*entry.data_file));
-  entry.status = ManifestStatus::kDeleted;
-  entry.snapshot_id = snapshot_id_;
-  return AddEntryInternal(entry);
-}
-
-Status ManifestEntryAdapter::AddExistingEntry(ManifestEntry& entry) {
-  ICEBERG_RETURN_UNEXPECTED(CheckDataFile(*entry.data_file));
-  entry.status = ManifestStatus::kExisting;
-  return AddEntryInternal(entry);
-}
-
-ManifestFile ManifestEntryAdapter::ToManifestFile() const {
+Result<ManifestFile> ManifestEntryAdapter::ToManifestFile() const {
   ManifestFile manifest_file;
   manifest_file.partition_spec_id = partition_spec_->spec_id();
   manifest_file.content = content_;
@@ -199,75 +174,17 @@ ManifestFile ManifestEntryAdapter::ToManifestFile() const {
   manifest_file.min_sequence_number =
       min_sequence_number_.value_or(TableMetadata::kInvalidSequenceNumber);
   manifest_file.existing_files_count = existing_files_count_;
-  manifest_file.added_snapshot_id = snapshot_id_.value_or(Snapshot::kInvalidSnapshotId);
   manifest_file.added_files_count = add_files_count_;
   manifest_file.existing_files_count = existing_files_count_;
   manifest_file.deleted_files_count = delete_files_count_;
   manifest_file.added_rows_count = add_rows_count_;
   manifest_file.existing_rows_count = existing_rows_count_;
   manifest_file.deleted_rows_count = delete_rows_count_;
-  manifest_file.partitions = std::move(partition_summary_->Summaries());
+  ICEBERG_ASSIGN_OR_RAISE(auto partition_summary, partition_summary_->Summaries());
+  manifest_file.partitions = std::move(partition_summary);
   return manifest_file;
 }
 
-Status ManifestEntryAdapter::CheckDataFile(const DataFile& file) const {
-  switch (content_) {
-    case ManifestContent::kData:
-      if (file.content != DataFile::Content::kData) {
-        return InvalidArgument(
-            "Manifest content type: data, data file content should be: data, but got: {}",
-            ToString(file.content));
-      }
-      break;
-    case ManifestContent::kDeletes:
-      if (file.content != DataFile::Content::kPositionDeletes &&
-          file.content != DataFile::Content::kEqualityDeletes) {
-        return InvalidArgument(
-            "Manifest content type: deletes, data file content should be: "
-            "position_deletes or equality_deletes, but got: {}",
-            ToString(file.content));
-      }
-      break;
-    default:
-      std::unreachable();
-  }
-  return {};
-}
-
-Status ManifestEntryAdapter::AddEntryInternal(const ManifestEntry& entry) {
-  if (entry.data_file == nullptr) [[unlikely]] {
-    return InvalidManifest("Missing required data_file field from manifest entry.");
-  }
-
-  switch (entry.status) {
-    case ManifestStatus::kAdded:
-      add_files_count_++;
-      add_rows_count_ += entry.data_file->record_count;
-      break;
-    case ManifestStatus::kExisting:
-      existing_files_count_++;
-      existing_rows_count_ += entry.data_file->record_count;
-      break;
-    case ManifestStatus::kDeleted:
-      delete_files_count_++;
-      delete_rows_count_ += entry.data_file->record_count;
-      break;
-    default:
-      std::unreachable();
-  }
-
-  ICEBERG_RETURN_UNEXPECTED(partition_summary_->Update(entry.data_file->partition));
-
-  if (entry.IsAlive() && entry.sequence_number.has_value()) {
-    if (!min_sequence_number_.has_value() ||
-        entry.sequence_number.value() < min_sequence_number_.value()) {
-      min_sequence_number_ = entry.sequence_number.value();
-    }
-  }
-
-  return AppendInternal(entry);
-}
-
 Status ManifestEntryAdapter::AppendPartitionValues(
     ArrowArray* array, const std::shared_ptr<StructType>& partition_type,
     const std::vector<Literal>& partition_values) {
@@ -425,13 +342,16 @@ Status ManifestEntryAdapter::AppendDataFile(
           ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArrayAppendNull(child_array, 1));
         }
         break;
-      case 142:  // first_row_id (optional int64)
-        if (file.first_row_id.has_value()) {
-          ICEBERG_RETURN_UNEXPECTED(AppendField(child_array, file.first_row_id.value()));
+      case 142: {
+        // first_row_id (optional int64)
+        ICEBERG_ASSIGN_OR_RAISE(auto first_row_id, GetFirstRowId(file));
+        if (first_row_id.has_value()) {
+          ICEBERG_RETURN_UNEXPECTED(AppendField(child_array, first_row_id.value()));
         } else {
           ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArrayAppendNull(child_array, 1));
         }
         break;
+      }
       case 143: {
         // referenced_data_file (optional string)
         ICEBERG_ASSIGN_OR_RAISE(auto referenced_data_file, GetReferenceDataFile(file));
@@ -493,6 +413,29 @@ Result<std::optional<int64_t>> ManifestEntryAdapter::GetContentSizeInBytes(
 }
 
 Status ManifestEntryAdapter::AppendInternal(const ManifestEntry& entry) {
+  if (entry.data_file == nullptr) [[unlikely]] {
+    return InvalidManifest("Missing required data_file field from manifest entry.");
+  }
+
+  switch (entry.status) {
+    case ManifestStatus::kAdded:
+      add_files_count_++;
+      add_rows_count_ += entry.data_file->record_count;
+      break;
+    case ManifestStatus::kExisting:
+      existing_files_count_++;
+      existing_rows_count_ += entry.data_file->record_count;
+      break;
+    case ManifestStatus::kDeleted:
+      delete_files_count_++;
+      delete_rows_count_ += entry.data_file->record_count;
+      break;
+    default:
+      std::unreachable();
+  }
+
+  ICEBERG_RETURN_UNEXPECTED(partition_summary_->Update(entry.data_file->partition));
+
   const auto& fields = manifest_schema_->fields();
   for (size_t i = 0; i < fields.size(); i++) {
     const auto& field = fields[i];

src/iceberg/manifest_adapter.h

@@ -66,39 +66,19 @@ class ICEBERG_EXPORT ManifestEntryAdapter : public ManifestAdapter {
 
   ~ManifestEntryAdapter() override;
 
-  /// \brief Add a new entry to the manifest.
-  ///
-  /// This method will update following status of the entry:
-  /// - Update the entry status to `Added`
-  /// - Set the snapshot id to the current snapshot id
-  /// - Set the sequence number to nullopt if it is invalid(smaller than 0)
-  /// - Set the file sequence number to nullopt
-  virtual Status AddEntry(ManifestEntry& entry);
-
-  /// \brief Add a delete entry to the manifest.
-  ///
-  /// This method will update following status of the entry:
-  /// - Update the entry status to `Deleted`
-  /// - Set the snapshot id to the current snapshot id
-  virtual Status AddDeleteEntry(ManifestEntry& entry);
-
-  /// \brief Add an existing entry to the manifest.
-  ///
-  /// This method will update following status of the entry:
-  /// - Update the entry status to `Existing`
-  virtual Status AddExistingEntry(ManifestEntry& entry);
+  virtual Status Append(const ManifestEntry& entry) = 0;
 
   const std::shared_ptr<Schema>& schema() const { return manifest_schema_; }
 
   ManifestContent content() const { return content_; }
 
+  std::optional<int64_t> snapshot_id() const { return snapshot_id_; }
+
   /// \brief Create a ManifestFile object without setting file metadata, such as
   /// location, file size, key metadata, etc.
-  ManifestFile ToManifestFile() const;
+  Result<ManifestFile> ToManifestFile() const;
 
  protected:
-  Status CheckDataFile(const DataFile& file) const;
-  Status AddEntryInternal(const ManifestEntry& entry);
   Status AppendInternal(const ManifestEntry& entry);
   Status AppendDataFile(ArrowArray* array,
                         const std::shared_ptr<StructType>& data_file_type,
@@ -139,7 +119,7 @@ class ICEBERG_EXPORT ManifestFileAdapter : public ManifestAdapter {
   ManifestFileAdapter() = default;
   ~ManifestFileAdapter() override;
 
-  virtual Status Append(ManifestFile& file) = 0;
+  virtual Status Append(const ManifestFile& file) = 0;
 
   const std::shared_ptr<Schema>& schema() const { return manifest_list_schema_; }
 

src/iceberg/manifest_entry.h

@@ -311,9 +311,21 @@ struct ICEBERG_EXPORT ManifestEntry {
     return status == ManifestStatus::kAdded || status == ManifestStatus::kExisting;
   }
 
-  /// \brief Create a copy of this manifest entry.
-  ManifestEntry Copy() const {
+  ManifestEntry AsAdded() const {
     ManifestEntry copy = *this;
+    copy.status = ManifestStatus::kAdded;
+    return copy;
+  }
+
+  ManifestEntry AsExisting() const {
+    ManifestEntry copy = *this;
+    copy.status = ManifestStatus::kExisting;
+    return copy;
+  }
+
+  ManifestEntry AsDeleted() const {
+    ManifestEntry copy = *this;
+    copy.status = ManifestStatus::kDeleted;
     return copy;
   }
 

src/iceberg/manifest_list.h

@@ -198,12 +198,6 @@ struct ICEBERG_EXPORT ManifestFile {
   bool operator==(const ManifestFile& other) const = default;
 
   static const std::shared_ptr<Schema>& Type();
-
-  /// \brief Create a copy of this manifest file.
-  ManifestFile Copy() const {
-    ManifestFile copy = *this;
-    return copy;
-  }
 };
 
 /// Snapshots are embedded in table metadata, but the list of manifests for a snapshot are