You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ga...@apache.org on 2026/09/20 09:43:17 UTC
(iceberg-cpp) branch main updated: fix(manifest): preserve snapshot ID inheritance semantics (#947)
This is an automated email from the ASF dual-hosted git repository.
wgtmac pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 168cb52b fix(manifest): preserve snapshot ID inheritance semantics (#947)
168cb52b is described below
commit 168cb52b90c279679c37f1c7cf481194a839148b
Author: Gang Wu <us...@gmail.com>
AuthorDate: Sun Sep 20 17:43:12 2026 +0800
fix(manifest): preserve snapshot ID inheritance semantics (#947)
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.
---
src/iceberg/inheritable_metadata.cc | 6 +--
src/iceberg/manifest/manifest_adapter.cc | 6 ++-
src/iceberg/manifest/manifest_list.h | 4 +-
src/iceberg/manifest/manifest_merge_manager.cc | 10 +++--
src/iceberg/manifest/manifest_writer.cc | 2 +-
src/iceberg/manifest/manifest_writer.h | 2 +-
src/iceberg/manifest/v2_metadata.cc | 12 ++++--
src/iceberg/manifest/v3_metadata.cc | 12 ++++--
src/iceberg/row/manifest_wrapper.cc | 2 +-
src/iceberg/table_scan.cc | 13 +++---
src/iceberg/test/fast_append_test.cc | 54 +++++++++++++++++++++---
src/iceberg/test/manifest_list_versions_test.cc | 12 ++++++
src/iceberg/test/merge_append_test.cc | 34 +++++++++++++--
src/iceberg/test/merging_snapshot_update_test.cc | 7 ++-
src/iceberg/update/expire_snapshots.cc | 8 +++-
src/iceberg/update/fast_append.cc | 6 ++-
src/iceberg/update/merge_append.cc | 3 +-
src/iceberg/update/merging_snapshot_update.cc | 2 +-
src/iceberg/update/snapshot_update.cc | 17 +++++---
19 files changed, 163 insertions(+), 49 deletions(-)
diff --git a/src/iceberg/inheritable_metadata.cc b/src/iceberg/inheritable_metadata.cc
index 8f6d53f5..e36b075e 100644
--- a/src/iceberg/inheritable_metadata.cc
+++ b/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(
diff --git a/src/iceberg/manifest/manifest_adapter.cc b/src/iceberg/manifest/manifest_adapter.cc
index 6b4b86d6..b26249a6 100644
--- a/src/iceberg/manifest/manifest_adapter.cc
+++ b/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()) {
diff --git a/src/iceberg/manifest/manifest_list.h b/src/iceberg/manifest/manifest_list.h
index 65271368..c6b0b7a4 100644
--- a/src/iceberg/manifest/manifest_list.h
+++ b/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
diff --git a/src/iceberg/manifest/manifest_merge_manager.cc b/src/iceberg/manifest/manifest_merge_manager.cc
index b2becf92..fc9183d1 100644
--- a/src/iceberg/manifest/manifest_merge_manager.cc
+++ b/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;
}
}
diff --git a/src/iceberg/manifest/manifest_writer.cc b/src/iceberg/manifest/manifest_writer.cc
index fbe3d79b..bdc6ba2a 100644
--- a/src/iceberg/manifest/manifest_writer.cc
+++ b/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_,
diff --git a/src/iceberg/manifest/manifest_writer.h b/src/iceberg/manifest/manifest_writer.h
index 0eaf478d..d8d9afdb 100644
--- a/src/iceberg/manifest/manifest_writer.h
+++ b/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.
diff --git a/src/iceberg/manifest/v2_metadata.cc b/src/iceberg/manifest/v2_metadata.cc
index d42e9f07..2e1e9e7c 100644
--- a/src/iceberg/manifest/v2_metadata.cc
+++ b/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
diff --git a/src/iceberg/manifest/v3_metadata.cc b/src/iceberg/manifest/v3_metadata.cc
index 9148e067..ad577b24 100644
--- a/src/iceberg/manifest/v3_metadata.cc
+++ b/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
diff --git a/src/iceberg/row/manifest_wrapper.cc b/src/iceberg/row/manifest_wrapper.cc
index 18be8223..2460c0a0 100644
--- a/src/iceberg/row/manifest_wrapper.cc
+++ b/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:
diff --git a/src/iceberg/table_scan.cc b/src/iceberg/table_scan.cc
index 37b52330..e4a85e9a 100644
--- a/src/iceberg/table_scan.cc
+++ b/src/iceberg/table_scan.cc
@@ -804,10 +804,12 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> IncrementalAppendScan::PlanFi
for (const auto& snapshot : append_snapshots) {
SnapshotReader snapshot_reader(snapshot.get());
ICEBERG_ASSIGN_OR_RAISE(auto manifests, snapshot_reader.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);
}
diff --git a/src/iceberg/test/fast_append_test.cc b/src/iceberg/test/fast_append_test.cc
index 54e40537..d75e1ef2 100644
--- a/src/iceberg/test/fast_append_test.cc
+++ b/src/iceberg/test/fast_append_test.cc
@@ -142,12 +142,13 @@ class FastAppendTest : public UpdateTestBase {
return data_file;
}
- Result<ManifestFile> WriteManifest(
- const std::string& path, const std::vector<std::shared_ptr<DataFile>>& files) {
+ Result<ManifestFile> WriteManifest(const std::string& path,
+ const std::vector<std::shared_ptr<DataFile>>& files,
+ std::optional<int64_t> snapshot_id = std::nullopt) {
ICEBERG_ASSIGN_OR_RAISE(
- auto writer, ManifestWriter::MakeWriter(table_->metadata()->format_version,
- kInvalidSnapshotId, path, file_io_, spec_,
- schema_, ManifestContent::kData));
+ auto writer,
+ ManifestWriter::MakeWriter(table_->metadata()->format_version, snapshot_id, path,
+ file_io_, spec_, schema_, ManifestContent::kData));
for (const auto& file : files) {
ManifestEntry entry;
entry.status = ManifestStatus::kAdded;
@@ -495,6 +496,49 @@ TEST_F(FastAppendTest, RebaseCopiesAppendManifestAgain) {
EXPECT_THAT(file_io_->ReadFile(attempt_lists[2], std::nullopt), IsOk());
}
+TEST_F(FastAppendTest, AppendManifestWithSnapshotIdInheritance) {
+ const auto path = table_location_ + "/metadata/inherited.avro";
+ ICEBERG_UNWRAP_OR_FAIL(auto manifest, WriteManifest(path, {file_a_, file_b_}));
+ ASSERT_FALSE(manifest.added_snapshot_id.has_value());
+
+ ICEBERG_UNWRAP_OR_FAIL(auto append, table_->NewFastAppend());
+ append->AppendManifest(manifest);
+ EXPECT_THAT(append->Commit(), IsOk());
+
+ EXPECT_THAT(table_->Refresh(), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
+ ICEBERG_UNWRAP_OR_FAIL(auto manifests, CurrentDataManifests());
+ ASSERT_THAT(manifests, ::testing::SizeIs(1));
+ EXPECT_EQ(manifests[0].manifest_path, path);
+ EXPECT_EQ(manifests[0].added_snapshot_id, snapshot->snapshot_id);
+ ICEBERG_UNWRAP_OR_FAIL(auto entries, ReadEntries(manifests[0]));
+ ASSERT_THAT(entries, ::testing::SizeIs(2));
+ EXPECT_EQ(entries[0].snapshot_id, snapshot->snapshot_id);
+ EXPECT_EQ(entries[1].snapshot_id, snapshot->snapshot_id);
+}
+
+TEST_F(FastAppendTest, AppendManifestWithExplicitInvalidSnapshotIdIsCopied) {
+ const auto path = table_location_ + "/metadata/invalid.avro";
+ ICEBERG_UNWRAP_OR_FAIL(auto manifest,
+ WriteManifest(path, {file_a_, file_b_}, kInvalidSnapshotId));
+ ASSERT_EQ(manifest.added_snapshot_id, kInvalidSnapshotId);
+
+ ICEBERG_UNWRAP_OR_FAIL(auto append, table_->NewFastAppend());
+ append->AppendManifest(manifest);
+ EXPECT_THAT(append->Commit(), IsOk());
+
+ EXPECT_THAT(table_->Refresh(), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
+ ICEBERG_UNWRAP_OR_FAIL(auto manifests, CurrentDataManifests());
+ ASSERT_THAT(manifests, ::testing::SizeIs(1));
+ EXPECT_NE(manifests[0].manifest_path, path);
+ EXPECT_EQ(manifests[0].added_snapshot_id, snapshot->snapshot_id);
+ ICEBERG_UNWRAP_OR_FAIL(auto entries, ReadEntries(manifests[0]));
+ ASSERT_THAT(entries, ::testing::SizeIs(2));
+ EXPECT_EQ(entries[0].snapshot_id, snapshot->snapshot_id);
+ EXPECT_EQ(entries[1].snapshot_id, snapshot->snapshot_id);
+}
+
TEST_F(FastAppendTest, AppendDuplicateFile) {
std::shared_ptr<FastAppend> fast_append;
ICEBERG_UNWRAP_OR_FAIL(fast_append, table_->NewFastAppend());
diff --git a/src/iceberg/test/manifest_list_versions_test.cc b/src/iceberg/test/manifest_list_versions_test.cc
index ce0eb9da..20ce4716 100644
--- a/src/iceberg/test/manifest_list_versions_test.cc
+++ b/src/iceberg/test/manifest_list_versions_test.cc
@@ -223,6 +223,18 @@ TEST_F(TestManifestListVersions, TestV1Write) {
EXPECT_EQ(manifest.deleted_rows_count, kDeletedRows);
}
+TEST_F(TestManifestListVersions, RejectsUnassignedSnapshotId) {
+ ICEBERG_UNWRAP_OR_FAIL(auto writer, ManifestListWriter::MakeWriter(
+ /*format_version=*/2, kSnapshotId,
+ /*parent_snapshot_id=*/std::nullopt,
+ CreateManifestListPath(), file_io_, kSeqNum));
+
+ ManifestFile manifest;
+ manifest.manifest_path = "unassigned.avro";
+ EXPECT_THAT(writer->Add(manifest),
+ HasErrorMessage("Cannot write manifest with unassigned snapshot ID"));
+}
+
TEST_F(TestManifestListVersions, TestV2Write) {
auto manifest = WriteAndReadManifestList(/*format_version=*/2);
diff --git a/src/iceberg/test/merge_append_test.cc b/src/iceberg/test/merge_append_test.cc
index c3112695..5e20fa3a 100644
--- a/src/iceberg/test/merge_append_test.cc
+++ b/src/iceberg/test/merge_append_test.cc
@@ -107,11 +107,12 @@ class MergeAppendTestBase : public MinimalUpdateTestBase {
return file;
}
- Result<ManifestFile> WriteManifest(
- const std::string& path, const std::vector<std::shared_ptr<DataFile>>& files) {
+ Result<ManifestFile> WriteManifest(const std::string& path,
+ const std::vector<std::shared_ptr<DataFile>>& files,
+ std::optional<int64_t> snapshot_id = std::nullopt) {
ICEBERG_ASSIGN_OR_RAISE(
auto writer, ManifestWriter::MakeWriter(
- format_version(), std::nullopt, path, file_io_, spec_, schema_,
+ format_version(), snapshot_id, path, file_io_, spec_, schema_,
ManifestContent::kData, /*first_row_id=*/std::nullopt));
for (const auto& file : files) {
ManifestEntry entry;
@@ -573,6 +574,7 @@ TEST_P(MergeAppendTest, EmptyTableAppendFilesWithDifferentSpecs) {
TEST_P(MergeAppendTest, EmptyTableAppendManifest) {
auto path = table_location_ + "/metadata/input.avro";
ICEBERG_UNWRAP_OR_FAIL(auto manifest, WriteManifest(path, {file_a_, file_b_}));
+ ASSERT_FALSE(manifest.added_snapshot_id.has_value());
ICEBERG_UNWRAP_OR_FAIL(auto append, NewBranchMergeAppend());
append->AppendManifest(manifest);
@@ -1238,6 +1240,7 @@ TEST_P(MergeAppendTest, AppendManifestWithSnapshotIdInheritance) {
ICEBERG_UNWRAP_OR_FAIL(auto data_manifests, CurrentDataManifests());
ASSERT_EQ(data_manifests.size(), 1U);
EXPECT_EQ(data_manifests[0].manifest_path, manifest.manifest_path);
+ EXPECT_EQ(data_manifests[0].added_snapshot_id, snapshot->snapshot_id);
ExpectManifestEntries(data_manifests[0], {file_a_, file_b_},
{ManifestStatus::kAdded, ManifestStatus::kAdded},
{snapshot->snapshot_id, snapshot->snapshot_id},
@@ -1249,6 +1252,31 @@ TEST_P(MergeAppendTest, AppendManifestWithSnapshotIdInheritance) {
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kTotalRecords), "200");
}
+TEST_P(MergeAppendTest, AppendManifestWithExplicitInvalidSnapshotIdIsCopied) {
+ SetSnapshotIdInheritanceEnabled();
+
+ auto path = table_location_ + "/metadata/input.avro";
+ ICEBERG_UNWRAP_OR_FAIL(auto manifest,
+ WriteManifest(path, {file_a_, file_b_}, kInvalidSnapshotId));
+ ASSERT_EQ(manifest.added_snapshot_id, kInvalidSnapshotId);
+
+ ICEBERG_UNWRAP_OR_FAIL(auto append, NewBranchMergeAppend());
+ append->AppendManifest(manifest);
+ EXPECT_THAT(append->Commit(), IsOk());
+
+ EXPECT_THAT(table_->Refresh(), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto snapshot, CurrentSnapshot());
+ ICEBERG_UNWRAP_OR_FAIL(auto data_manifests, CurrentDataManifests());
+ ASSERT_EQ(data_manifests.size(), 1U);
+ EXPECT_NE(data_manifests[0].manifest_path, manifest.manifest_path);
+ EXPECT_EQ(data_manifests[0].added_snapshot_id, snapshot->snapshot_id);
+ ExpectManifestEntries(data_manifests[0], {file_a_, file_b_},
+ {ManifestStatus::kAdded, ManifestStatus::kAdded},
+ {snapshot->snapshot_id, snapshot->snapshot_id},
+ {snapshot->sequence_number, snapshot->sequence_number},
+ {snapshot->sequence_number, snapshot->sequence_number});
+}
+
TEST_P(MergeAppendTest, MergedAppendManifestCleanupWithSnapshotIdInheritance) {
SetSnapshotIdInheritanceEnabled();
SetManifestMinMergeCount(1);
diff --git a/src/iceberg/test/merging_snapshot_update_test.cc b/src/iceberg/test/merging_snapshot_update_test.cc
index 2c7c31fa..d31a1eb2 100644
--- a/src/iceberg/test/merging_snapshot_update_test.cc
+++ b/src/iceberg/test/merging_snapshot_update_test.cc
@@ -441,14 +441,13 @@ class MergingSnapshotUpdateTest : public MinimalUpdateTestBase {
}
// Write a manifest file containing the given data files.
- // Returns a ManifestFile with added_snapshot_id = kInvalidSnapshotId so it
- // is eligible for snapshot ID inheritance.
+ // Returns a manifest whose entries can inherit the snapshot ID assigned at commit.
Result<ManifestFile> WriteManifest(
const std::string& path, const std::vector<std::shared_ptr<DataFile>>& files) {
ICEBERG_ASSIGN_OR_RAISE(
auto writer,
- ManifestWriter::MakeWriter(/*format_version=*/2, kInvalidSnapshotId, path,
- file_io_, spec_, schema_, ManifestContent::kData));
+ ManifestWriter::MakeWriter(/*format_version=*/2, std::nullopt, path, file_io_,
+ spec_, schema_, ManifestContent::kData));
for (const auto& f : files) {
ManifestEntry entry;
entry.status = ManifestStatus::kAdded;
diff --git a/src/iceberg/update/expire_snapshots.cc b/src/iceberg/update/expire_snapshots.cc
index d7e13766..fd14308e 100644
--- a/src/iceberg/update/expire_snapshots.cc
+++ b/src/iceberg/update/expire_snapshots.cc
@@ -478,7 +478,9 @@ class IncrementalFileCleanup : public FileCleanupStrategy {
for (auto& manifest : manifests) {
valid_manifest_result.insert(manifest.manifest_path);
- int64_t writer_id = manifest.added_snapshot_id;
+ ICEBERG_CHECK(manifest.added_snapshot_id.has_value(),
+ "Manifest {} has no snapshot ID", manifest.manifest_path);
+ int64_t writer_id = manifest.added_snapshot_id.value();
bool from_valid_snapshots = valid_ids.contains(writer_id);
bool is_from_ancestor = ancestor_ids.contains(writer_id);
bool is_picked = picked_ancestor_snapshot_ids.contains(writer_id);
@@ -554,7 +556,9 @@ class IncrementalFileCleanup : public FileCleanupStrategy {
}
manifests_to_delete.insert(manifest.manifest_path);
- int64_t writer_id = manifest.added_snapshot_id;
+ ICEBERG_CHECK(manifest.added_snapshot_id.has_value(),
+ "Manifest {} has no snapshot ID", manifest.manifest_path);
+ int64_t writer_id = manifest.added_snapshot_id.value();
bool is_from_ancestor = ancestor_ids.contains(writer_id);
bool is_from_expiring_snapshot = expired_snapshot_ids.contains(writer_id);
diff --git a/src/iceberg/update/fast_append.cc b/src/iceberg/update/fast_append.cc
index c5352e86..1cb1e41c 100644
--- a/src/iceberg/update/fast_append.cc
+++ b/src/iceberg/update/fast_append.cc
@@ -69,12 +69,14 @@ FastAppend& FastAppend::AppendManifest(const ManifestFile& manifest) {
"Cannot append manifest with existing files");
ICEBERG_BUILDER_CHECK(!manifest.has_deleted_files(),
"Cannot append manifest with deleted files");
- ICEBERG_BUILDER_CHECK(manifest.added_snapshot_id == kInvalidSnapshotId,
+ ICEBERG_BUILDER_CHECK(!manifest.added_snapshot_id.has_value() ||
+ manifest.added_snapshot_id == kInvalidSnapshotId,
"Snapshot id must be assigned during commit");
ICEBERG_BUILDER_CHECK(manifest.sequence_number == kInvalidSequenceNumber,
"Sequence number must be assigned during commit");
- if (can_inherit_snapshot_id() && manifest.added_snapshot_id == kInvalidSnapshotId) {
+ // Only a missing ID is inheritance-ready; an explicit -1 must be rewritten.
+ if (can_inherit_snapshot_id() && !manifest.added_snapshot_id.has_value()) {
append_manifests_.push_back(manifest);
} else {
// The manifest must be rewritten with this update's snapshot ID
diff --git a/src/iceberg/update/merge_append.cc b/src/iceberg/update/merge_append.cc
index fd6cc35a..c8b87349 100644
--- a/src/iceberg/update/merge_append.cc
+++ b/src/iceberg/update/merge_append.cc
@@ -50,7 +50,8 @@ MergeAppend& MergeAppend::AppendManifest(const ManifestFile& manifest) {
"Cannot append manifest with existing files");
ICEBERG_BUILDER_CHECK(!manifest.has_deleted_files(),
"Cannot append manifest with deleted files");
- ICEBERG_BUILDER_CHECK(manifest.added_snapshot_id == kInvalidSnapshotId,
+ ICEBERG_BUILDER_CHECK(!manifest.added_snapshot_id.has_value() ||
+ manifest.added_snapshot_id == kInvalidSnapshotId,
"Snapshot id must be assigned during commit");
ICEBERG_BUILDER_CHECK(manifest.sequence_number == kInvalidSequenceNumber,
"Sequence number must be assigned during commit");
diff --git a/src/iceberg/update/merging_snapshot_update.cc b/src/iceberg/update/merging_snapshot_update.cc
index d9757fec..ae1577a8 100644
--- a/src/iceberg/update/merging_snapshot_update.cc
+++ b/src/iceberg/update/merging_snapshot_update.cc
@@ -657,7 +657,7 @@ Status MergingSnapshotUpdate::AddManifest(ManifestFile manifest) {
if (manifest.content != ManifestContent::kData) {
return InvalidArgument("Cannot append delete manifest: {}", manifest.manifest_path);
}
- if (can_inherit_snapshot_id() && manifest.added_snapshot_id == kInvalidSnapshotId) {
+ if (can_inherit_snapshot_id() && !manifest.added_snapshot_id.has_value()) {
if (manifest.first_row_id.has_value()) {
return InvalidArgument("Cannot append manifest with assigned first row ID: {}",
manifest.manifest_path);
diff --git a/src/iceberg/update/snapshot_update.cc b/src/iceberg/update/snapshot_update.cc
index e5d7bf69..d6a1066a 100644
--- a/src/iceberg/update/snapshot_update.cc
+++ b/src/iceberg/update/snapshot_update.cc
@@ -26,6 +26,7 @@
#include "iceberg/constants.h"
#include "iceberg/file_io.h"
+#include "iceberg/inheritable_metadata.h"
#include "iceberg/logging/log_macros.h"
#include "iceberg/manifest/manifest_entry.h"
#include "iceberg/manifest/manifest_list.h"
@@ -113,17 +114,21 @@ Result<std::vector<ManifestFile>> WriteManifestGroups(OptionalExecutor executor,
// Add metadata to a manifest file by reading it and extracting statistics.
Result<ManifestFile> AddMetadata(const ManifestFile& manifest, std::shared_ptr<FileIO> io,
const TableMetadata& metadata) {
- ICEBERG_PRECHECK(manifest.added_snapshot_id != kInvalidSnapshotId,
+ ICEBERG_PRECHECK(!manifest.added_snapshot_id.has_value(),
"Manifest {} already has assigned a snapshot id: {}",
- manifest.manifest_path, manifest.added_snapshot_id);
+ manifest.manifest_path, manifest.added_snapshot_id.value());
ICEBERG_ASSIGN_OR_RAISE(auto schema, metadata.Schema());
ICEBERG_ASSIGN_OR_RAISE(auto spec,
metadata.PartitionSpecById(manifest.partition_spec_id));
ICEBERG_ASSIGN_OR_RAISE(auto partition_type, spec->PartitionType(*schema));
- ICEBERG_ASSIGN_OR_RAISE(auto reader,
- ManifestReader::Make(manifest, std::move(io), schema, spec));
+ ICEBERG_ASSIGN_OR_RAISE(auto inheritable_metadata, InheritableMetadataFactory::Empty());
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto reader,
+ ManifestReader::Make(manifest.manifest_path, manifest.manifest_length,
+ std::move(io), schema, spec, std::move(inheritable_metadata),
+ manifest.first_row_id));
ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->Entries());
PartitionSummary stats(*partition_type);
@@ -319,7 +324,7 @@ Result<SnapshotUpdate::ApplyResult> SnapshotUpdate::Apply() {
ICEBERG_ASSIGN_OR_RAISE(auto manifests, Apply(base(), parent_snapshot));
auto metadata_tasks = TaskGroup().SetExecutor(plan_executor_);
for (auto& manifest : manifests) {
- if (manifest.added_snapshot_id != kInvalidSnapshotId) {
+ if (manifest.added_snapshot_id.has_value()) {
continue;
}
metadata_tasks.Submit([&manifest, this]() -> Status {
@@ -539,7 +544,7 @@ SnapshotSummaryBuilder SnapshotUpdate::BuildManifestCountSummary(
for (const auto& manifest : manifests) {
if (manifest.added_snapshot_id == snapshot_id) {
++manifests_created;
- } else if (manifest.added_snapshot_id != kInvalidSnapshotId) {
+ } else if (manifest.added_snapshot_id.has_value()) {
++manifests_kept;
}
}