fix(manifest): preserve snapshot ID inheritance state

Represent ManifestFile::added_snapshot_id as optional to distinguish an
inheritance-ready null ID from a literal -1.

Reuse manifests only when the ID is null, rewrite explicit -1 manifests
with the commit snapshot ID, and continue requiring the field when
writing manifest lists.

Author: wgtmac

src/iceberg/inheritable_metadata.cc

@@ -97,13 +97,13 @@ Result<std::unique_ptr<InheritableMetadata>> InheritableMetadataFactory::Empty()
 Result<std::unique_ptr<InheritableMetadata>> InheritableMetadataFactory::FromManifest(
     const ManifestFile& manifest) {
   // Validate that the manifest has a snapshot ID assigned
-  if (manifest.added_snapshot_id == kInvalidSnapshotId) {
+  if (!manifest.added_snapshot_id.has_value()) {
     return InvalidManifest("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);
+      manifest.partition_spec_id, manifest.added_snapshot_id.value(),
+      manifest.sequence_number, manifest.manifest_path);
 }
 
 Result<std::unique_ptr<InheritableMetadata>> InheritableMetadataFactory::ForCopy(

src/iceberg/manifest/manifest_adapter.cc

@@ -448,6 +448,10 @@ Result<std::optional<int64_t>> ManifestFileAdapter::GetFirstRowId(
 }
 
 Status ManifestFileAdapter::AppendInternal(const ManifestFile& file) {
+  if (!file.added_snapshot_id.has_value()) {
+    return InvalidManifestList("Cannot write manifest with unassigned snapshot ID: {}",
+                               file.manifest_path);
+  }
   const auto& fields = manifest_list_schema_->fields();
   for (size_t i = 0; i < fields.size(); i++) {
     const auto& field = fields[i];
@@ -478,7 +482,7 @@ Status ManifestFileAdapter::AppendInternal(const ManifestFile& file) {
         break;
       }
       case ManifestFile::kAddedSnapshotIdFieldId:
-        ICEBERG_RETURN_UNEXPECTED(AppendInt(array, file.added_snapshot_id));
+        ICEBERG_RETURN_UNEXPECTED(AppendInt(array, file.added_snapshot_id.value()));
         break;
       case ManifestFile::kAddedFilesCountFieldId:
         if (file.added_files_count.has_value()) {

src/iceberg/manifest/manifest_list.h

@@ -107,7 +107,9 @@ struct ICEBERG_EXPORT ManifestFile {
   int64_t min_sequence_number = TableMetadata::kInitialSequenceNumber;
   /// Field id: 503
   /// ID of the snapshot where the manifest file was added
-  int64_t added_snapshot_id = kInvalidSnapshotId;
+  /// May be null before commit so callers can distinguish inheritance from
+  /// kInvalidSnapshotId.
+  std::optional<int64_t> added_snapshot_id;
   /// Field id: 504
   /// Number of entries in the manifest that have status ADDED (1), when null this is
   /// assumed to be non-zero

src/iceberg/manifest/manifest_merge_manager.cc

@@ -177,7 +177,8 @@ Result<std::vector<ManifestFile>> ManifestMergeManager::MergeGroup(
         ICEBERG_ASSIGN_OR_RAISE(auto merged, FlushBin(bin, metadata, writer_factory));
         merged_manifests_.Add(bin, merged);
         for (const auto* manifest : bin) {
-          if (manifest->added_snapshot_id != snapshot_id) {
+          if (!manifest->added_snapshot_id.has_value() ||
+              manifest->added_snapshot_id.value() != snapshot_id) {
             ++replaced_manifests_count_;
           }
         }
@@ -202,8 +203,8 @@ Result<ManifestFile> ManifestMergeManager::FlushBin(
 
   const int64_t snapshot_id = snapshot_id_supplier_();
   for (const auto* manifest : bin) {
-    bool is_committed = manifest->added_snapshot_id != kInvalidSnapshotId &&
-                        manifest->added_snapshot_id != snapshot_id;
+    bool is_committed = manifest->added_snapshot_id.has_value() &&
+                        manifest->added_snapshot_id.value() != snapshot_id;
     ICEBERG_ASSIGN_OR_RAISE(auto reader, ManifestReader::Make(*manifest, file_io_, schema,
                                                               spec, is_committed));
     ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->Entries());
@@ -276,7 +277,8 @@ Result<int32_t> ManifestMergeManager::MergedManifestCache::CleanUncommitted(
 
     std::ignore = delete_file(merged.manifest_path);
     for (const auto& manifest : bin.bin) {
-      if (manifest.added_snapshot_id != snapshot_id) {
+      if (!manifest.added_snapshot_id.has_value() ||
+          manifest.added_snapshot_id.value() != snapshot_id) {
         ++removed_replaced_manifests_count;
       }
     }

src/iceberg/manifest/manifest_writer.cc

@@ -239,7 +239,7 @@ Result<ManifestFile> ManifestWriter::ToManifestFile() const {
       // have been written, so the min data sequence number is the one that will be
       // assigned when this is committed. pass kUnassignedSequenceNumber to inherit it.
       .min_sequence_number = min_sequence_number_.value_or(kUnassignedSequenceNumber),
-      .added_snapshot_id = adapter_->snapshot_id().value_or(kInvalidSnapshotId),
+      .added_snapshot_id = adapter_->snapshot_id(),
       .added_files_count = add_files_count_,
       .existing_files_count = existing_files_count_,
       .deleted_files_count = delete_files_count_,

src/iceberg/manifest/manifest_writer.h

@@ -121,7 +121,7 @@ class ICEBERG_EXPORT ManifestWriter {
   /// \brief Factory function to create a writer for a manifest file based on format
   /// version.
   /// \param format_version The format version (1, 2, 3, etc.).
-  /// \param snapshot_id ID of the snapshot.
+  /// \param snapshot_id ID written to new entries; null writes inheritable snapshot IDs.
   /// \param manifest_location Path to the manifest file.
   /// \param file_io File IO implementation to use.
   /// \param partition_spec Partition spec for the manifest.

src/iceberg/manifest/v2_metadata.cc

@@ -158,14 +158,16 @@ Status ManifestFileAdapterV2::Append(const ManifestFile& file) {
 }
 
 Result<int64_t> ManifestFileAdapterV2::GetSequenceNumber(const ManifestFile& file) const {
+  ICEBERG_DCHECK(file.added_snapshot_id.has_value(),
+                 "added_snapshot_id must be assigned before writing a manifest list");
   if (file.sequence_number == kUnassignedSequenceNumber) {
     // if the sequence number is being assigned here, then the manifest must be created by
     // the current operation. to validate this, check that the snapshot id matches the
     // current commit
-    if (snapshot_id_ != file.added_snapshot_id) {
+    if (snapshot_id_ != file.added_snapshot_id.value()) {
       return InvalidManifestList(
           "Found unassigned sequence number for a manifest from snapshot: {}",
-          file.added_snapshot_id);
+          file.added_snapshot_id.value());
     }
     return sequence_number_;
   }
@@ -174,12 +176,14 @@ Result<int64_t> ManifestFileAdapterV2::GetSequenceNumber(const ManifestFile& fil
 
 Result<int64_t> ManifestFileAdapterV2::GetMinSequenceNumber(
     const ManifestFile& file) const {
+  ICEBERG_DCHECK(file.added_snapshot_id.has_value(),
+                 "added_snapshot_id must be assigned before writing a manifest list");
   if (file.min_sequence_number == kUnassignedSequenceNumber) {
     // same sanity check as above
-    if (snapshot_id_ != file.added_snapshot_id) {
+    if (snapshot_id_ != file.added_snapshot_id.value()) {
       return InvalidManifestList(
           "Found unassigned sequence number for a manifest from snapshot: {}",
-          file.added_snapshot_id);
+          file.added_snapshot_id.value());
     }
     // if the min sequence number is not determined, then there was no assigned sequence
     // number for any file written to the wrapped manifest. replace the unassigned

src/iceberg/manifest/v3_metadata.cc

@@ -206,14 +206,16 @@ Status ManifestFileAdapterV3::Append(const ManifestFile& file) {
 }
 
 Result<int64_t> ManifestFileAdapterV3::GetSequenceNumber(const ManifestFile& file) const {
+  ICEBERG_DCHECK(file.added_snapshot_id.has_value(),
+                 "added_snapshot_id must be assigned before writing a manifest list");
   if (file.sequence_number == kUnassignedSequenceNumber) {
     // if the sequence number is being assigned here, then the manifest must be created by
     // the current operation. to validate this, check that the snapshot id matches the
     // current commit
-    if (snapshot_id_ != file.added_snapshot_id) {
+    if (snapshot_id_ != file.added_snapshot_id.value()) {
       return InvalidManifestList(
           "Found unassigned sequence number for a manifest from snapshot: {}",
-          file.added_snapshot_id);
+          file.added_snapshot_id.value());
     }
     return sequence_number_;
   }
@@ -222,12 +224,14 @@ Result<int64_t> ManifestFileAdapterV3::GetSequenceNumber(const ManifestFile& fil
 
 Result<int64_t> ManifestFileAdapterV3::GetMinSequenceNumber(
     const ManifestFile& file) const {
+  ICEBERG_DCHECK(file.added_snapshot_id.has_value(),
+                 "added_snapshot_id must be assigned before writing a manifest list");
   if (file.min_sequence_number == kUnassignedSequenceNumber) {
     // same sanity check as above
-    if (snapshot_id_ != file.added_snapshot_id) {
+    if (snapshot_id_ != file.added_snapshot_id.value()) {
       return InvalidManifestList(
           "Found unassigned sequence number for a manifest from snapshot: {}",
-          file.added_snapshot_id);
+          file.added_snapshot_id.value());
     }
     // if the min sequence number is not determined, then there was no assigned sequence
     // number for any file written to the wrapped manifest. replace the unassigned

src/iceberg/row/manifest_wrapper.cc

@@ -206,7 +206,7 @@ Result<Scalar> ManifestFileStructLike::GetField(size_t pos) const {
     case ManifestFileField::kMinSequenceNumber:
       return manifest_file.min_sequence_number;
     case ManifestFileField::kAddedSnapshotId:
-      return manifest_file.added_snapshot_id;
+      return FromOptional(manifest_file.added_snapshot_id);
     case ManifestFileField::kAddedFilesCount:
       return FromOptional(manifest_file.added_files_count);
     case ManifestFileField::kExistingFilesCount:

src/iceberg/table_scan.cc

@@ -804,10 +804,12 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> IncrementalAppendScan::PlanFi
   for (const auto& snapshot : append_snapshots) {
     SnapshotCache snapshot_cache(snapshot.get());
     ICEBERG_ASSIGN_OR_RAISE(auto manifests, snapshot_cache.DataManifests(io_));
-    std::ranges::copy_if(manifests, std::inserter(data_manifests, data_manifests.end()),
-                         [&snapshot_ids](const ManifestFile& manifest) {
-                           return snapshot_ids.contains(manifest.added_snapshot_id);
-                         });
+    std::ranges::copy_if(
+        manifests, std::inserter(data_manifests, data_manifests.end()),
+        [&snapshot_ids](const ManifestFile& manifest) {
+          return manifest.added_snapshot_id.has_value() &&
+                 snapshot_ids.contains(manifest.added_snapshot_id.value());
+        });
   }
   if (data_manifests.empty()) {
     return std::vector<std::shared_ptr<FileScanTask>>{};
@@ -902,7 +904,8 @@ IncrementalChangelogScan::PlanFiles(std::optional<int64_t> from_snapshot_id_excl
   for (const auto& snapshot : changelog_snapshots) {
     ICEBERG_ASSIGN_OR_RAISE(auto manifests, snapshot.second->DataManifests(io_));
     for (auto& manifest : manifests) {
-      if (snapshot_ids.contains(manifest.added_snapshot_id) &&
+      if (manifest.added_snapshot_id.has_value() &&
+          snapshot_ids.contains(manifest.added_snapshot_id.value()) &&
           seen_manifest_paths.insert(manifest.manifest_path).second) {
         data_manifests.push_back(manifest);
       }