fix: implement manifest entry metadata inheritance system

- Add InheritableMetadata interface and implementations for applying
  metadata inheritance rules to manifest entries
- Fix ManifestReader::Make methods to properly handle Result types
  from InheritableMetadataFactory methods
- Move constructor/destructor definitions from header to implementation
  to avoid redefinition errors
- Add missing ManifestListReaderImpl::Files() method implementation
- Update method names from MakeReader to Make for consistency

This ensures proper inheritance of snapshot_id, sequence_number, and
file_sequence_number from manifest metadata to individual entries
according to Iceberg specification.

Author: HeartLinked

src/iceberg/inheritable_metadata.cc

@@ -0,0 +1,102 @@
+/*
+ * 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/inheritable_metadata.h"
+
+#include <cassert>
+#include <utility>
+
+#include <iceberg/result.h>
+
+#include "iceberg/manifest_entry.h"
+
+namespace iceberg {
+
+BaseInheritableMetadata::BaseInheritableMetadata(int32_t spec_id, int64_t snapshot_id,
+                                                 int64_t sequence_number,
+                                                 std::string manifest_location)
+    : spec_id_(spec_id),
+      snapshot_id_(snapshot_id),
+      sequence_number_(sequence_number),
+      manifest_location_(std::move(manifest_location)) {}
+
+Result<ManifestEntry> BaseInheritableMetadata::Apply(ManifestEntry entry) {
+  if (!entry.snapshot_id.has_value()) {
+    entry.snapshot_id = snapshot_id_;
+  }
+
+  // In v1 tables, the data sequence number is not persisted and can be safely defaulted
+  // to 0 In v2 tables, the data sequence number should be inherited iff the entry status
+  // is ADDED
+  if (!entry.sequence_number.has_value() &&
+      (sequence_number_ == 0 || entry.status == ManifestStatus::kAdded)) {
+    entry.sequence_number = sequence_number_;
+  }
+
+  // In v1 tables, the file sequence number is not persisted and can be safely defaulted
+  // to 0 In v2 tables, the file sequence number should be inherited iff the entry status
+  // is ADDED
+  if (!entry.file_sequence_number.has_value() &&
+      (sequence_number_ == 0 || entry.status == ManifestStatus::kAdded)) {
+    entry.file_sequence_number = sequence_number_;
+  }
+
+  if (entry.data_file) {
+    entry.data_file->partition_spec_id = spec_id_;
+  }
+
+  return entry;
+}
+
+Result<ManifestEntry> EmptyInheritableMetadata::Apply(ManifestEntry entry) {
+  return entry;
+}
+
+CopyInheritableMetadata::CopyInheritableMetadata(int64_t snapshot_id)
+    : snapshot_id_(snapshot_id) {}
+
+Result<ManifestEntry> CopyInheritableMetadata::Apply(ManifestEntry entry) {
+  if (!entry.snapshot_id.has_value()) {
+    entry.snapshot_id = snapshot_id_;
+  }
+  return entry;
+}
+
+Result<std::unique_ptr<InheritableMetadata>> InheritableMetadataFactory::Empty() {
+  return std::make_unique<EmptyInheritableMetadata>();
+}
+
+Result<std::unique_ptr<InheritableMetadata>> InheritableMetadataFactory::FromManifest(
+    const ManifestFile& manifest) {
+  // Validate that the manifest has a snapshot ID assigned
+  if (manifest.added_snapshot_id == Snapshot::kInvalidSnapshotId) {
+    return InvalidArgument("Manifest file {} has no snapshot ID", manifest.manifest_path);
+  }
+
+  return std::make_unique<BaseInheritableMetadata>(
+      manifest.partition_spec_id, manifest.added_snapshot_id, manifest.sequence_number,
+      manifest.manifest_path);
+}
+
+Result<std::unique_ptr<InheritableMetadata>> InheritableMetadataFactory::ForCopy(
+    int64_t snapshot_id) {
+  return std::make_unique<CopyInheritableMetadata>(snapshot_id);
+}
+
+}  // namespace iceberg

src/iceberg/inheritable_metadata.h

@@ -0,0 +1,118 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+/// \file iceberg/inheritable_metadata.h
+/// Metadata inheritance system for manifest entries.
+
+#include <cstdint>
+#include <memory>
+#include <string>
+
+#include <iceberg/result.h>
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+
+namespace iceberg {
+
+/// \brief Interface for applying inheritable metadata to manifest entries.
+///
+/// When manifest entries have null values for certain fields (snapshot_id,
+/// data sequence number, file sequence number), these values should be inherited
+/// from the manifest file. This interface provides a way to apply such inheritance rules.
+class ICEBERG_EXPORT InheritableMetadata {
+ public:
+  virtual ~InheritableMetadata() = default;
+
+  /// \brief Apply inheritable metadata to a manifest entry.
+  /// \param entry The manifest entry to modify.
+  /// \return The modified manifest entry with inherited metadata applied.
+  virtual Result<ManifestEntry> Apply(ManifestEntry entry) = 0;
+};
+
+/// \brief Base implementation of InheritableMetadata that handles standard inheritance
+/// rules.
+class ICEBERG_EXPORT BaseInheritableMetadata : public InheritableMetadata {
+ public:
+  /// \brief Constructor for base inheritable metadata.
+  /// \param spec_id Partition spec ID from the manifest.
+  /// \param snapshot_id Snapshot ID from the manifest.
+  /// \param sequence_number Sequence number from the manifest.
+  /// \param manifest_location Path to the manifest file.
+  BaseInheritableMetadata(int32_t spec_id, int64_t snapshot_id, int64_t sequence_number,
+                          std::string manifest_location);
+
+  /// \brief Apply inheritance rules to a manifest entry.
+  /// \param entry The manifest entry to modify.
+  /// \return The modified manifest entry.
+  Result<ManifestEntry> Apply(ManifestEntry entry) override;
+
+ private:
+  int32_t spec_id_;
+  int64_t snapshot_id_;
+  int64_t sequence_number_;
+  std::string manifest_location_;
+};
+
+/// \brief Empty implementation that applies no inheritance.
+class ICEBERG_EXPORT EmptyInheritableMetadata : public InheritableMetadata {
+ public:
+  /// \brief Apply no inheritance - returns the entry unchanged.
+  Result<ManifestEntry> Apply(ManifestEntry entry) override;
+};
+
+/// \brief Metadata inheritance for copying manifests before commit.
+class ICEBERG_EXPORT CopyInheritableMetadata : public InheritableMetadata {
+ public:
+  /// \brief Constructor for copy metadata.
+  /// \param snapshot_id The snapshot ID to use for copying.
+  explicit CopyInheritableMetadata(int64_t snapshot_id);
+
+  /// \brief Apply copy inheritance rules.
+  Result<ManifestEntry> Apply(ManifestEntry entry) override;
+
+ private:
+  int64_t snapshot_id_;
+};
+
+/// \brief Factory for creating InheritableMetadata instances.
+class ICEBERG_EXPORT InheritableMetadataFactory {
+ public:
+  /// \brief Create an empty metadata instance that applies no inheritance.
+  static Result<std::unique_ptr<InheritableMetadata>> Empty();
+
+  /// \brief Create metadata instance from a manifest file.
+  /// \param manifest The manifest file to extract metadata from.
+  /// \return Inheritable metadata based on the manifest.
+  static Result<std::unique_ptr<InheritableMetadata>> FromManifest(
+      const ManifestFile& manifest);
+
+  /// \brief Create metadata instance for rewriting a manifest before commit.
+  /// \param snapshot_id The snapshot ID for the copy operation.
+  /// \return Inheritable metadata for copying.
+  static Result<std::unique_ptr<InheritableMetadata>> ForCopy(int64_t snapshot_id);
+
+ private:
+  InheritableMetadataFactory() = default;
+};
+
+}  // namespace iceberg

src/iceberg/manifest_reader.cc

@@ -19,6 +19,7 @@
 
 #include "iceberg/manifest_reader.h"
 
+#include "iceberg/file_reader.h"
 #include "iceberg/manifest_entry.h"
 #include "iceberg/manifest_list.h"
 #include "iceberg/manifest_reader_internal.h"
@@ -27,7 +28,33 @@
 
 namespace iceberg {
 
-Result<std::unique_ptr<ManifestReader>> ManifestReader::MakeReader(
+Result<std::unique_ptr<ManifestReader>> ManifestReader::Make(
+    const ManifestFile& manifest, std::shared_ptr<FileIO> file_io,
+    std::shared_ptr<Schema> partition_schema) {
+  // Validate manifest content type - only DATA manifests are supported by ManifestReader
+  if (manifest.content != ManifestFile::Content::kData) {
+    return InvalidArgument("Cannot read a delete manifest with a ManifestReader: {}",
+                           manifest.manifest_path);
+  }
+
+  auto manifest_entry_schema = ManifestEntry::TypeFromPartitionType(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 reader,
+      ReaderFactoryRegistry::Open(FileFormatType::kAvro, {.path = manifest.manifest_path,
+                                                          .io = std::move(file_io),
+                                                          .projection = schema}));
+  // Create inheritable metadata for this manifest
+  ICEBERG_ASSIGN_OR_RAISE(auto inheritable_metadata,
+                          InheritableMetadataFactory::FromManifest(manifest));
+
+  return std::make_unique<ManifestReaderImpl>(std::move(reader), std::move(schema),
+                                              std::move(inheritable_metadata));
+}
+
+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);
@@ -39,10 +66,12 @@ Result<std::unique_ptr<ManifestReader>> ManifestReader::MakeReader(
                                                {.path = std::string(manifest_location),
                                                 .io = std::move(file_io),
                                                 .projection = schema}));
-  return std::make_unique<ManifestReaderImpl>(std::move(reader), std::move(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));
 }
 
-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

@@ -27,6 +27,7 @@
 
 #include "iceberg/file_reader.h"
 #include "iceberg/iceberg_export.h"
+#include "iceberg/manifest_list.h"
 #include "iceberg/type_fwd.h"
 
 namespace iceberg {
@@ -37,11 +38,21 @@ class ICEBERG_EXPORT ManifestReader {
   virtual ~ManifestReader() = default;
   virtual Result<std::vector<ManifestEntry>> Entries() const = 0;
 
+  /// \brief Creates a reader for a manifest file.
+  /// \param manifest A ManifestFile object containing metadata about the manifest.
+  /// \param file_io File IO implementation to use.
+  /// \param partition_schema Schema for the partition.
+  /// \return A Result containing the reader or an error.
+  static Result<std::unique_ptr<ManifestReader>> Make(
+      const ManifestFile& manifest, std::shared_ptr<FileIO> file_io,
+      std::shared_ptr<Schema> partition_schema);
+
   /// \brief Creates a reader for a manifest file.
   /// \param manifest_location Path to the manifest file.
   /// \param file_io File IO implementation to use.
+  /// \param partition_schema Schema for the partition.
   /// \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 +67,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_reader_internal.cc

@@ -30,6 +30,7 @@
 #include "iceberg/schema.h"
 #include "iceberg/type.h"
 #include "iceberg/util/macros.h"
+#include "iceberg/util/visit_type.h"
 
 namespace iceberg {
 
@@ -541,6 +542,12 @@ Result<std::vector<ManifestEntry>> ManifestReaderImpl::Entries() const {
       break;
     }
   }
+
+  // Apply inheritance to all entries
+  for (auto& entry : manifest_entries) {
+    ICEBERG_ASSIGN_OR_RAISE(entry, inheritable_metadata_->Apply(std::move(entry)));
+  }
+
   return manifest_entries;
 }
 

src/iceberg/manifest_reader_internal.h

@@ -22,30 +22,37 @@
 /// \file iceberg/internal/manifest_reader_internal.h
 /// Reader implementation for manifest list files and manifest files.
 
+#include "iceberg/file_reader.h"
+#include "iceberg/inheritable_metadata.h"
 #include "iceberg/manifest_reader.h"
 
 namespace iceberg {
 
 /// \brief Read manifest entries from a manifest file.
 class ManifestReaderImpl : public ManifestReader {
  public:
-  explicit ManifestReaderImpl(std::unique_ptr<Reader> reader,
-                              std::shared_ptr<Schema> schema)
-      : schema_(std::move(schema)), reader_(std::move(reader)) {}
+  ManifestReaderImpl(std::unique_ptr<Reader> reader, std::shared_ptr<Schema> schema,
+                     std::unique_ptr<InheritableMetadata> inheritable_metadata)
+      : schema_(std::move(schema)),
+        reader_(std::move(reader)),
+        inheritable_metadata_(std::move(inheritable_metadata)) {}
+
+  ManifestReaderImpl() = default;
 
   Result<std::vector<ManifestEntry>> Entries() const override;
 
  private:
   std::shared_ptr<Schema> schema_;
   std::unique_ptr<Reader> reader_;
+  std::unique_ptr<InheritableMetadata> inheritable_metadata_;
 };
 
 /// \brief Read manifest files from a manifest list file.
 class ManifestListReaderImpl : public ManifestListReader {
  public:
-  explicit ManifestListReaderImpl(std::unique_ptr<Reader> reader,
-                                  std::shared_ptr<Schema> schema)
+  ManifestListReaderImpl(std::unique_ptr<Reader> reader, std::shared_ptr<Schema> schema)
       : schema_(std::move(schema)), reader_(std::move(reader)) {}
+  ManifestListReaderImpl() = default;
 
   Result<std::vector<ManifestFile>> Files() const override;