various fixes:

- unify partition summary collection
- fix manifest reader to inherit metadata
- fix first row id round trip

Author: wgtmac

src/iceberg/manifest_adapter.cc

@@ -27,7 +27,6 @@
 #include "iceberg/arrow/nanoarrow_status_internal.h"
 #include "iceberg/manifest_entry.h"
 #include "iceberg/manifest_list.h"
-#include "iceberg/partition_summary_internal.h"
 #include "iceberg/result.h"
 #include "iceberg/schema.h"
 #include "iceberg/util/checked_cast.h"
@@ -321,16 +320,13 @@ Status ManifestEntryAdapter::AppendDataFile(
           ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArrayAppendNull(child_array, 1));
         }
         break;
-      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()));
+      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()));
         } 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));
@@ -396,8 +392,6 @@ Status ManifestEntryAdapter::AppendInternal(const ManifestEntry& entry) {
     return InvalidManifest("Missing required data_file field from manifest entry.");
   }
 
-  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

@@ -76,13 +76,7 @@ class ICEBERG_EXPORT ManifestEntryAdapter : public ManifestAdapter {
 
   const std::shared_ptr<PartitionSpec>& partition_spec() const { return partition_spec_; }
 
-  const std::unique_ptr<PartitionSummary>& partition_summary() const {
-    return partition_summary_;
-  }
-
-  /// \brief Create a ManifestFile object without setting file metadata, such as
-  /// location, file size, key metadata, etc.
-  Result<ManifestFile> ToManifestFile() const;
+  const std::shared_ptr<StructType>& partition_type() const { return partition_type_; }
 
  protected:
   Status AppendInternal(const ManifestEntry& entry);
@@ -106,9 +100,9 @@ class ICEBERG_EXPORT ManifestEntryAdapter : public ManifestAdapter {
   std::optional<int64_t> snapshot_id_;
   std::shared_ptr<PartitionSpec> partition_spec_;
   std::shared_ptr<Schema> current_schema_;
+  std::shared_ptr<StructType> partition_type_;
   std::shared_ptr<Schema> manifest_schema_;
   const ManifestContent content_;
-  std::unique_ptr<PartitionSummary> partition_summary_;
 };
 
 /// \brief Adapter for appending a list of `ManifestFile`s to an `ArrowArray`.

src/iceberg/manifest_entry.cc

@@ -22,37 +22,11 @@
 #include <memory>
 #include <vector>
 
-#include "iceberg/schema.h"
 #include "iceberg/schema_field.h"
 #include "iceberg/type.h"
 
 namespace iceberg {
 
-std::shared_ptr<DataFile> DataFile::Clone() const {
-  auto copy = std::make_shared<DataFile>();
-  copy->content = content;
-  copy->file_path = file_path;
-  copy->file_format = file_format;
-  copy->partition = partition;
-  copy->record_count = record_count;
-  copy->file_size_in_bytes = file_size_in_bytes;
-  copy->column_sizes = column_sizes;
-  copy->value_counts = value_counts;
-  copy->null_value_counts = null_value_counts;
-  copy->nan_value_counts = nan_value_counts;
-  copy->lower_bounds = lower_bounds;
-  copy->upper_bounds = upper_bounds;
-  copy->key_metadata = key_metadata;
-  copy->split_offsets = split_offsets;
-  copy->equality_ids = equality_ids;
-  copy->sort_order_id = sort_order_id;
-  copy->first_row_id = first_row_id;
-  copy->referenced_data_file = referenced_data_file;
-  copy->content_offset = content_offset;
-  copy->content_size_in_bytes = content_size_in_bytes;
-  return copy;
-}
-
 bool ManifestEntry::operator==(const ManifestEntry& other) const {
   return status == other.status && snapshot_id == other.snapshot_id &&
          sequence_number == other.sequence_number &&

src/iceberg/manifest_entry.h

@@ -268,8 +268,7 @@ struct ICEBERG_EXPORT DataFile {
 
   bool operator==(const DataFile& other) const = default;
 
-  std::shared_ptr<DataFile> Clone() const;
-
+  /// \brief Get the schema of the data file with the given partition type.
   static std::shared_ptr<StructType> Type(std::shared_ptr<StructType> partition_type);
 };
 
@@ -316,6 +315,10 @@ struct ICEBERG_EXPORT ManifestEntry {
   ManifestEntry AsAdded() const {
     ManifestEntry copy = *this;
     copy.status = ManifestStatus::kAdded;
+    if (copy.data_file->first_row_id.has_value()) {
+      copy.data_file = std::make_unique<DataFile>(*copy.data_file);
+      copy.data_file->first_row_id = std::nullopt;
+    }
     return copy;
   }
 

src/iceberg/manifest_reader.cc

@@ -48,7 +48,8 @@ Result<std::unique_ptr<ManifestReader>> ManifestReader::Make(
                           InheritableMetadataFactory::FromManifest(manifest));
 
   return std::make_unique<ManifestReaderImpl>(std::move(reader), std::move(schema),
-                                              std::move(inheritable_metadata));
+                                              std::move(inheritable_metadata),
+                                              manifest.first_row_id);
 }
 
 Result<std::unique_ptr<ManifestReader>> ManifestReader::Make(
@@ -66,7 +67,8 @@ Result<std::unique_ptr<ManifestReader>> ManifestReader::Make(
                                                 .projection = schema}));
   ICEBERG_ASSIGN_OR_RAISE(auto inheritable_metadata, InheritableMetadataFactory::Empty());
   return std::make_unique<ManifestReaderImpl>(std::move(reader), std::move(schema),
-                                              std::move(inheritable_metadata));
+                                              std::move(inheritable_metadata),
+                                              std::nullopt);
 }
 
 Result<std::unique_ptr<ManifestListReader>> ManifestListReader::Make(

src/iceberg/manifest_reader_internal.cc

@@ -328,7 +328,7 @@ Status ParseLiteral(ArrowArrayView* view_of_partition, int64_t row_idx,
 }
 
 Status ParseDataFile(const std::shared_ptr<StructType>& data_file_schema,
-                     ArrowArrayView* view_of_column,
+                     ArrowArrayView* view_of_column, std::optional<int64_t>& first_row_id,
                      std::vector<ManifestEntry>& manifest_entries) {
   if (view_of_column->storage_type != ArrowType::NANOARROW_TYPE_STRUCT) {
     return InvalidManifest("DataFile field should be a struct.");
@@ -432,10 +432,25 @@ Status ParseDataFile(const std::shared_ptr<StructType>& data_file_schema,
         PARSE_PRIMITIVE_FIELD(manifest_entries[row_idx].data_file->sort_order_id,
                               view_of_file_field, int32_t);
         break;
-      case 16:
+      case 16: {
         PARSE_PRIMITIVE_FIELD(manifest_entries[row_idx].data_file->first_row_id,
                               view_of_file_field, int64_t);
+        if (first_row_id.has_value()) {
+          std::ranges::for_each(manifest_entries, [&first_row_id](ManifestEntry& entry) {
+            if (entry.status != ManifestStatus::kDeleted &&
+                !entry.data_file->first_row_id.has_value()) {
+              entry.data_file->first_row_id = first_row_id.value();
+              first_row_id = first_row_id.value() + entry.data_file->record_count;
+            }
+          });
+        } else {
+          // data file's first_row_id is null when the manifest's first_row_id is null
+          std::ranges::for_each(manifest_entries, [](ManifestEntry& entry) {
+            entry.data_file->first_row_id = std::nullopt;
+          });
+        }
         break;
+      }
       case 17:
         PARSE_STRING_FIELD(manifest_entries[row_idx].data_file->referenced_data_file,
                            view_of_file_field);
@@ -455,9 +470,9 @@ Status ParseDataFile(const std::shared_ptr<StructType>& data_file_schema,
   return {};
 }
 
-Result<std::vector<ManifestEntry>> ParseManifestEntry(ArrowSchema* schema,
-                                                      ArrowArray* array_in,
-                                                      const Schema& iceberg_schema) {
+Result<std::vector<ManifestEntry>> ParseManifestEntry(
+    ArrowSchema* schema, ArrowArray* array_in, const Schema& iceberg_schema,
+    std::optional<int64_t>& first_row_id) {
   if (schema->n_children != array_in->n_children) {
     return InvalidManifest("Columns size not match between schema:{} and array:{}",
                            schema->n_children, array_in->n_children);
@@ -512,8 +527,8 @@ Result<std::vector<ManifestEntry>> ParseManifestEntry(ArrowSchema* schema,
       case 4: {
         auto data_file_schema =
             internal::checked_pointer_cast<StructType>(field.value()->get().type());
-        ICEBERG_RETURN_UNEXPECTED(
-            ParseDataFile(data_file_schema, view_of_column, manifest_entries));
+        ICEBERG_RETURN_UNEXPECTED(ParseDataFile(data_file_schema, view_of_column,
+                                                first_row_id, manifest_entries));
         break;
       }
       default:
@@ -533,7 +548,7 @@ Result<std::vector<ManifestEntry>> ManifestReaderImpl::Entries() const {
       internal::ArrowArrayGuard array_guard(&result.value());
       ICEBERG_ASSIGN_OR_RAISE(
           auto parse_result,
-          ParseManifestEntry(&arrow_schema, &result.value(), *schema_));
+          ParseManifestEntry(&arrow_schema, &result.value(), *schema_, first_row_id_));
       manifest_entries.insert(manifest_entries.end(),
                               std::make_move_iterator(parse_result.begin()),
                               std::make_move_iterator(parse_result.end()));

src/iceberg/manifest_reader_internal.h

@@ -33,10 +33,12 @@ class ManifestReaderImpl : public ManifestReader {
  public:
   explicit ManifestReaderImpl(std::unique_ptr<Reader> reader,
                               std::shared_ptr<Schema> schema,
-                              std::unique_ptr<InheritableMetadata> inheritable_metadata)
+                              std::unique_ptr<InheritableMetadata> inheritable_metadata,
+                              std::optional<int64_t> first_row_id)
       : schema_(std::move(schema)),
         reader_(std::move(reader)),
-        inheritable_metadata_(std::move(inheritable_metadata)) {}
+        inheritable_metadata_(std::move(inheritable_metadata)),
+        first_row_id_(first_row_id) {}
 
   Result<std::vector<ManifestEntry>> Entries() const override;
 
@@ -46,6 +48,7 @@ class ManifestReaderImpl : public ManifestReader {
   std::shared_ptr<Schema> schema_;
   std::unique_ptr<Reader> reader_;
   std::unique_ptr<InheritableMetadata> inheritable_metadata_;
+  mutable std::optional<int64_t> first_row_id_;
 };
 
 /// \brief Read manifest files from a manifest list file.