Skip to content

fix: preserve staged manifest dependencies during transaction replay #1104

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wgtmac wants to merge 2 commits into
base: main
Choose a base branch
from
+572 −219
Open

fix: preserve staged manifest dependencies during transaction replay #1104

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e205346
fix: preserve staged manifest dependencies during transaction replay
wgtmac Nov 9, 2026
1d98110
address copilot comments
wgtmac Nov 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions src/iceberg/test/cherry_pick_operation_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
#include "iceberg/avro/avro_register.h"
#include "iceberg/manifest/manifest_entry.h"
#include "iceberg/manifest/manifest_reader.h"
#include "iceberg/metrics/metrics_reporter.h"
#include "iceberg/partition_spec.h"
#include "iceberg/row/partition_values.h"
#include "iceberg/schema.h"
Expand Down Expand Up @@ -653,6 +654,18 @@ TEST_F(CherryPickOperationTest, CherryPickReplayedAsFastForwardAfterRollback) {
ICEBERG_UNWRAP_OR_FAIL(auto parent_id, CommitAppend(file_a_));
ICEBERG_UNWRAP_OR_FAIL(auto staged_id, StageAppend(file_b_));
ASSERT_THAT(CommitAppend(conflict_a_), IsOk());

class MockReporter : public MetricsReporter {
public:
MOCK_METHOD(Status, Report, (const MetricsReport&), (override));
};
auto reporter = std::make_shared<MockReporter>();
EXPECT_CALL(*reporter, Report(::testing::_)).Times(0);
ICEBERG_UNWRAP_OR_FAIL(
table_, Table::Make(table_->name(), table_->metadata(),
std::string(table_->metadata_file_location()), file_io_,
catalog_, table_->full_name(), reporter));

ICEBERG_UNWRAP_OR_FAIL(auto transaction, StageCherrypick(staged_id));
ICEBERG_UNWRAP_OR_FAIL(auto draft, transaction->current().Snapshot());
ASSERT_NE(draft->snapshot_id, staged_id);
Expand Down
21 changes: 15 additions & 6 deletions src/iceberg/test/fast_append_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -429,7 +429,7 @@ TEST_F(FastAppendTest, TransactionApplyFailureCleansUpStagedFiles) {
EXPECT_THAT(deleted_paths, ::testing::SizeIs(2U));
}

TEST_F(FastAppendTest, RebaseCopiesAppendManifestAgain) {
TEST_F(FastAppendTest, RebaseReusesCopiedAppendManifest) {
table_->metadata()->format_version = 1;
ASSERT_THAT(
TableMetadataUtil::Write(*file_io_, std::string(table_->metadata_file_location()),
Expand All @@ -449,12 +449,20 @@ TEST_F(FastAppendTest, RebaseCopiesAppendManifestAgain) {
ASSERT_THAT(properties->Commit(), IsOk());
}
ICEBERG_UNWRAP_OR_FAIL(auto snapshot, txn->current().Snapshot());
EXPECT_THAT(snapshot->summary,
::testing::Contains(::testing::Pair(
std::string(SnapshotSummaryFields::kAddedRecords), "100")));
SnapshotReader snapshot_reader(snapshot.get());
ICEBERG_UNWRAP_OR_FAIL(auto manifests, snapshot_reader.DataManifests(file_io_));
ASSERT_THAT(manifests, ::testing::SizeIs(1));
const auto& manifest_path = manifests[0].manifest_path;
EXPECT_NE(manifest_path, path);
EXPECT_THAT(attempt_manifests, ::testing::Not(::testing::Contains(manifest_path)));
if (!attempt_manifests.empty()) {
EXPECT_EQ(manifest_path, attempt_manifests.front());
}
// The copy is immutable, but each attempt needs a new manifest list.
EXPECT_THAT(attempt_lists,
::testing::Not(::testing::Contains(snapshot->manifest_list)));
attempt_manifests.push_back(manifest_path);
attempt_lists.push_back(snapshot->manifest_list);
ICEBERG_UNWRAP_OR_FAIL(auto entries, ReadEntries(manifests[0]));
Expand Down Expand Up @@ -487,12 +495,13 @@ TEST_F(FastAppendTest, RebaseCopiesAppendManifestAgain) {
std::unordered_set<std::string> unique_deleted_paths(deleted_paths.begin(),
deleted_paths.end());
EXPECT_THAT(unique_deleted_paths,
::testing::UnorderedElementsAre(attempt_manifests[0], attempt_lists[0],
attempt_manifests[1], attempt_lists[1]));
::testing::UnorderedElementsAre(attempt_lists[0], attempt_lists[1]));
EXPECT_THAT(deleted_paths, ::testing::SizeIs(2));
EXPECT_THAT(deleted_paths, ::testing::Not(::testing::Contains(path)));
EXPECT_THAT(deleted_paths,
::testing::Not(::testing::Contains(manifests[0].manifest_path)));
EXPECT_THAT(file_io_->ReadFile(path, std::nullopt), IsOk());
EXPECT_THAT(file_io_->ReadFile(manifests[0].manifest_path, std::nullopt), IsOk());
EXPECT_THAT(file_io_->ReadFile(attempt_lists[2], std::nullopt), IsOk());
}

Expand Down Expand Up @@ -1241,10 +1250,10 @@ TEST_F(FastAppendTest, ReplayBecomingNoopCleansStagingWithoutReporting) {
EXPECT_EQ(committed->properties().configs().at("effective"), "value");
EXPECT_EQ(txn->state(), TransactionState::kCommitted);
}
EXPECT_EQ(deletes, 4);
EXPECT_EQ(deletes, 3);
EXPECT_EQ(reports, 0);
EXPECT_THAT(append->Commit(), IsError(ErrorKind::kValidationFailed));
EXPECT_EQ(deletes, 4);
EXPECT_EQ(deletes, 3);
}
}

Expand Down
174 changes: 170 additions & 4 deletions src/iceberg/test/merging_snapshot_update_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,10 +56,12 @@
#include "iceberg/test/update_test_base.h"
#include "iceberg/transaction.h"
#include "iceberg/update/delete_files.h"
#include "iceberg/update/expire_snapshots.h"
#include "iceberg/update/fast_append.h"
#include "iceberg/update/merge_append.h"
#include "iceberg/update/row_delta.h"
#include "iceberg/update/snapshot_manager.h"
#include "iceberg/update/update_partition_spec.h"
#include "iceberg/update/update_properties.h"
#include "iceberg/util/macros.h"

Expand Down Expand Up @@ -1380,7 +1382,137 @@ TEST_F(MergingSnapshotUpdateTest, AddManifestCopiesManifestWithAssignedSnapshotI
EXPECT_NE(data_manifests[0].manifest_path, path);
}

TEST_F(MergingSnapshotUpdateTest, AddManifestRetryCopiesManifestAgain) {
TEST_F(MergingSnapshotUpdateTest, ReplayRejectsRemovedPartitionSpec) {
for (bool fast_append : {true, false}) {
SCOPED_TRACE(fast_append);
RegisterTableFromResource(MetadataResource());
const auto original_metadata = table_->metadata();
ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
if (fast_append) {
ICEBERG_UNWRAP_OR_FAIL(auto append, txn->NewFastAppend());
append->AppendFile(file_a_);
ASSERT_THAT(append->Commit(), IsOk());
} else {
ICEBERG_UNWRAP_OR_FAIL(auto append, txn->NewMergeAppend());
append->AppendFile(file_a_);
ASSERT_THAT(append->Commit(), IsOk());
}

ICEBERG_UNWRAP_OR_FAIL(auto update_spec, table_->NewUpdatePartitionSpec());
update_spec->AddField("y");
ASSERT_THAT(update_spec->Commit(), IsOk());
ASSERT_THAT(table_->Refresh(), IsOk());
ICEBERG_UNWRAP_OR_FAIL(auto expire, table_->NewExpireSnapshots());
expire->CleanExpiredMetadata(true);
ASSERT_THAT(expire->Commit(), IsOk());
ASSERT_THAT(table_->Refresh(), IsOk());
ASSERT_THAT(table_->metadata()->PartitionSpecById(spec_->spec_id()),
IsError(ErrorKind::kNotFound));
ASSERT_NE(original_metadata, table_->metadata());

EXPECT_THAT(txn->Commit(), ::testing::AllOf(IsError(ErrorKind::kValidationFailed),
HasErrorMessage("Partition spec")));
EXPECT_EQ(txn->state(), TransactionState::kFailed);
EXPECT_TRUE(ReloadMetadata()->snapshots.empty());
}
}

TEST_F(MergingSnapshotUpdateTest, ReplayKeepsMergedDVReferencedOnlyByStagedSnapshot) {
UpgradeTableToV3();
CommitFileA();
const auto main_snapshot_id = table_->metadata()->current_snapshot_id;
ICEBERG_UNWRAP_OR_FAIL(
auto first,
WriteDeletionVector(table_location_ + "/data/first.puffin", file_a_, {1}));
ICEBERG_UNWRAP_OR_FAIL(
auto second,
WriteDeletionVector(table_location_ + "/data/second.puffin", file_a_, {2, 3}));
FailCommits(1, [&](int attempt) {
if (attempt != 0) return;
ICEBERG_UNWRAP_OR_FAIL(auto concurrent, catalog_->LoadTable(table_ident_));
ICEBERG_UNWRAP_OR_FAIL(auto properties, concurrent->NewUpdateProperties());
properties->Set("replay-conflict", "true");
ASSERT_THAT(properties->Commit(), IsOk());
});

ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
ICEBERG_UNWRAP_OR_FAIL(auto delta, txn->NewRowDelta());
delta->AddDeletes(first).AddDeletes(second).StageOnly();
ASSERT_THAT(delta->Commit(), IsOk());
auto staged = txn->current().snapshots.back();
ICEBERG_UNWRAP_OR_FAIL(auto committed, txn->Commit());
ICEBERG_UNWRAP_OR_FAIL(auto retained,
committed->metadata()->SnapshotById(staged->snapshot_id));
SnapshotReader reader(retained.get());
ICEBERG_UNWRAP_OR_FAIL(auto manifests, reader.DeleteManifests(file_io_));
ICEBERG_UNWRAP_OR_FAIL(
auto entries,
ReadAllEntries({manifests.begin(), manifests.end()}, *committed->metadata()));
ASSERT_THAT(entries, ::testing::SizeIs(1));
EXPECT_THAT(entries[0].data_file->file_path, ::testing::HasSubstr("merged-dvs-"));
ICEBERG_UNWRAP_OR_FAIL(auto positions, DVUtil::ReadDV(entries[0].data_file, file_io_));
EXPECT_EQ(positions.Cardinality(), 3);
EXPECT_EQ(committed->metadata()->current_snapshot_id, main_snapshot_id);
EXPECT_NE(retained->manifest_list, staged->manifest_list);
EXPECT_FALSE(file_io_->ReadFile(staged->manifest_list, std::nullopt).has_value());
}

TEST_F(MergingSnapshotUpdateTest, ReplayRefreshesInheritedManifestMetadata) {
UpgradeTableToV3();
ICEBERG_UNWRAP_OR_FAIL(auto properties, table_->NewUpdateProperties());
properties->Set(TableProperties::kManifestMinMergeCount.key(), "1");
ASSERT_THAT(properties->Commit(), IsOk());
ASSERT_THAT(table_->Refresh(), IsOk());
ICEBERG_UNWRAP_OR_FAIL(auto initial, table_->NewFastAppend());
initial->AppendFile(file_c_);
ASSERT_THAT(initial->Commit(), IsOk());
ASSERT_THAT(table_->Refresh(), IsOk());
ICEBERG_UNWRAP_OR_FAIL(auto base_snapshot, table_->current_snapshot());
ICEBERG_UNWRAP_OR_FAIL(auto manager, table_->NewSnapshotManager());
manager->CreateBranch("audit", base_snapshot->snapshot_id);
ASSERT_THAT(manager->Commit(), IsOk());
ASSERT_THAT(table_->Refresh(), IsOk());

FailCommits(2, [&](int attempt) {
if (attempt == 2) return;
ICEBERG_UNWRAP_OR_FAIL(auto concurrent, catalog_->LoadTable(table_ident_));
ICEBERG_UNWRAP_OR_FAIL(auto append, concurrent->NewFastAppend());
// First change the merge bin. Then change only its inherited metadata.
append->ToBranch(attempt == 0 ? "main" : "audit");
append->AppendFile(
MakeDataFile(std::format("/data/concurrent-{}.parquet", attempt), 4));
ASSERT_THAT(append->Commit(), IsOk());
});
ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
ICEBERG_UNWRAP_OR_FAIL(auto first, txn->NewFastAppend());
first->AppendFile(file_a_);
ASSERT_THAT(first->Commit(), IsOk());
ICEBERG_UNWRAP_OR_FAIL(auto staged, txn->current().Snapshot());
ICEBERG_UNWRAP_OR_FAIL(auto merge, txn->NewMergeAppend());
merge->AppendFile(file_b_);
ASSERT_THAT(merge->Commit(), IsOk());
ICEBERG_UNWRAP_OR_FAIL(auto committed, txn->Commit());

ICEBERG_UNWRAP_OR_FAIL(auto replayed_append,
committed->metadata()->SnapshotById(staged->snapshot_id));
ICEBERG_UNWRAP_OR_FAIL(auto snapshot, committed->current_snapshot());
SnapshotReader reader(snapshot.get());
ICEBERG_UNWRAP_OR_FAIL(auto manifests, reader.DataManifests(file_io_));
ICEBERG_UNWRAP_OR_FAIL(
auto entries,
ReadAllEntries({manifests.begin(), manifests.end()}, *committed->metadata()));
auto entry = std::ranges::find_if(entries, [&](const ManifestEntry& entry) {
return entry.data_file->file_path == file_a_->file_path;
});
ASSERT_NE(entry, entries.end());
EXPECT_EQ(entry->snapshot_id, replayed_append->snapshot_id);
EXPECT_EQ(entry->sequence_number, replayed_append->sequence_number);
EXPECT_EQ(entry->file_sequence_number, replayed_append->sequence_number);
EXPECT_EQ(entry->data_file->first_row_id, replayed_append->first_row_id);
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kManifestsReplaced), "3");
}

TEST_F(MergingSnapshotUpdateTest, AddManifestRetryReusesCopiedManifest) {
auto path = table_location_ + "/metadata/retry-input.avro";
ICEBERG_UNWRAP_OR_FAIL(auto manifest, WriteManifest(path, {file_a_}));
manifest.added_snapshot_id = 12345;
Expand All @@ -1390,12 +1522,20 @@ TEST_F(MergingSnapshotUpdateTest, AddManifestRetryCopiesManifestAgain) {
FailCommits(2, [&](int attempt) {
SCOPED_TRACE(attempt);
ICEBERG_UNWRAP_OR_FAIL(auto snapshot, op->StagedSnapshot());
EXPECT_THAT(snapshot->summary,
::testing::Contains(::testing::Pair(
std::string(SnapshotSummaryFields::kAddedRecords), "100")));
SnapshotReader snapshot_reader(snapshot.get());
ICEBERG_UNWRAP_OR_FAIL(auto manifests, snapshot_reader.DataManifests(file_io_));
ASSERT_THAT(manifests, ::testing::SizeIs(1));
const auto& manifest_path = manifests[0].manifest_path;
EXPECT_NE(manifest_path, path);
EXPECT_THAT(attempt_manifests, ::testing::Not(::testing::Contains(manifest_path)));
if (!attempt_manifests.empty()) {
EXPECT_EQ(manifest_path, attempt_manifests.front());
}
// The copy is immutable, but each attempt needs a new manifest list.
EXPECT_THAT(attempt_lists,
::testing::Not(::testing::Contains(snapshot->manifest_list)));
attempt_manifests.push_back(manifest_path);
attempt_lists.push_back(snapshot->manifest_list);
});
Expand All @@ -1412,8 +1552,7 @@ TEST_F(MergingSnapshotUpdateTest, AddManifestRetryCopiesManifestAgain) {
ASSERT_THAT(attempt_lists, ::testing::SizeIs(3));
EXPECT_EQ(manifests[0].manifest_path, attempt_manifests[2]);
EXPECT_THAT(deleted,
::testing::UnorderedElementsAre(attempt_manifests[0], attempt_lists[0],
attempt_manifests[1], attempt_lists[1]));
::testing::UnorderedElementsAre(attempt_lists[0], attempt_lists[1]));
EXPECT_THAT(deleted, ::testing::Not(::testing::Contains(path)));
EXPECT_THAT(deleted, ::testing::Not(::testing::Contains(manifests[0].manifest_path)));
EXPECT_THAT(file_io_->ReadFile(path, std::nullopt), IsOk());
Expand All @@ -1424,6 +1563,33 @@ TEST_F(MergingSnapshotUpdateTest, AddManifestRetryCopiesManifestAgain) {
EXPECT_EQ(entries[0].data_file->file_path, file_a_->file_path);
}

TEST_F(MergingSnapshotUpdateTest, ResumesManifestCopiesAfterApplyFailure) {
ICEBERG_UNWRAP_OR_FAIL(auto op, NewMergeAppend());
auto first_path = table_location_ + "/metadata/copy-first.avro";
auto second_path = table_location_ + "/metadata/copy-second.avro";
ICEBERG_UNWRAP_OR_FAIL(auto first, WriteManifest(first_path, {file_a_}));
ICEBERG_UNWRAP_OR_FAIL(auto second, WriteManifest(second_path, {file_b_}));
first.added_snapshot_id = second.added_snapshot_id = 12345;
ASSERT_THAT(op->AppendManifest(first), IsOk());
ASSERT_THAT(op->AppendManifest(second), IsOk());

ICEBERG_UNWRAP_OR_FAIL(auto contents, file_io_->ReadFile(second_path, std::nullopt));
ASSERT_THAT(file_io_->DeleteFile(second_path), IsOk());
EXPECT_FALSE(op->ApplyForTest(*table_->metadata(), nullptr).has_value());
ASSERT_THAT(file_io_->WriteFile(second_path, contents), IsOk());

ICEBERG_UNWRAP_OR_FAIL(auto manifests, op->CommitManifests());
ASSERT_THAT(manifests, ::testing::SizeIs(2));
ASSERT_THAT(table_->Refresh(), IsOk());
ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kAddedRecords), "200");
ICEBERG_UNWRAP_OR_FAIL(auto entries, ReadAllEntries(manifests, *table_->metadata()));
ASSERT_THAT(entries, ::testing::SizeIs(2));
EXPECT_THAT(
(std::vector{entries[0].data_file->file_path, entries[1].data_file->file_path}),
::testing::UnorderedElementsAre(file_a_->file_path, file_b_->file_path));
}

TEST_F(MergingSnapshotUpdateTest, AddManifestRejectsManifestWithFirstRowId) {
auto path = table_location_ + "/metadata/rowid.avro";
ICEBERG_UNWRAP_OR_FAIL(auto manifest, WriteManifest(path, {file_a_}));
Expand Down
81 changes: 81 additions & 0 deletions src/iceberg/test/rewrite_manifests_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2488,6 +2488,87 @@ TEST_P(RewriteManifestsTest, RewriteManifestsOnBranch) {
EXPECT_EQ(std::ranges::distance(audit_manifests), 1);
}

TEST_P(RewriteManifestsTest, ReplaysReplacementAfterEarlierStagedAppend) {
SetCommitRetryProperties(2);
FailCommits(2, [&](int attempt) {
if (attempt < 2) {
ICEBERG_UNWRAP_OR_FAIL(auto concurrent, catalog_->LoadTable(table_ident_));
ICEBERG_UNWRAP_OR_FAIL(auto properties, concurrent->NewUpdateProperties());
properties->Set(std::format("replay-conflict-{}", attempt), "true");
ASSERT_THAT(properties->Commit(), IsOk());
}
});

ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
ICEBERG_UNWRAP_OR_FAIL(auto append, txn->NewFastAppend());
append->AppendFile(file_a_);
ASSERT_THAT(append->Commit(), IsOk());

ICEBERG_UNWRAP_OR_FAIL(auto staged_snapshot, txn->current().Snapshot());
SnapshotReader staged_reader(staged_snapshot.get());
ICEBERG_UNWRAP_OR_FAIL(auto staged_manifests, staged_reader.DataManifests(file_io_));
ASSERT_THAT(staged_manifests, ::testing::SizeIs(1));
ICEBERG_UNWRAP_OR_FAIL(auto entries, ReadManifestEntries(staged_manifests[0]));
ASSERT_THAT(entries, ::testing::SizeIs(1));
ICEBERG_UNWRAP_OR_FAIL(auto replacement,
WriteExistingManifest("txn-replay-replacement", entries));

ICEBERG_UNWRAP_OR_FAIL(auto rewrite, txn->NewRewriteManifests());
rewrite->DeleteManifest(staged_manifests[0]).AddManifest(replacement);
ASSERT_THAT(rewrite->Commit(), IsOk());

ASSERT_THAT(txn->Commit(), IsOk());
ICEBERG_UNWRAP_OR_FAIL(auto manifests, CurrentManifests());
ASSERT_THAT(manifests, ::testing::SizeIs(1));
ICEBERG_UNWRAP_OR_FAIL(auto final_entries, ReadManifestEntries(manifests[0]));
ASSERT_THAT(final_entries, ::testing::SizeIs(1));
EXPECT_EQ(final_entries[0].data_file->file_path, file_a_->file_path);

// The append snapshot still needs the manifest replaced by the rewrite.
ICEBERG_UNWRAP_OR_FAIL(auto committed_append,
table_->metadata()->SnapshotById(staged_snapshot->snapshot_id));
SnapshotReader committed_reader(committed_append.get());
EXPECT_THAT(committed_reader.DataManifests(file_io_), IsOk());
EXPECT_TRUE(FileExists(staged_manifests[0].manifest_path));
}

TEST_P(RewriteManifestsTest, RejectsReplacementAfterStagedManifestMetadataChanges) {
if (GetParam() == 1) {
GTEST_SKIP() << "Sequence and row-ID inheritance require format v2 or higher";
}
ASSERT_THAT(AppendFiles({file_c_}), IsOk());
ICEBERG_UNWRAP_OR_FAIL(auto base_snapshot, table_->current_snapshot());
ICEBERG_UNWRAP_OR_FAIL(auto manager, table_->NewSnapshotManager());
manager->CreateBranch("audit", base_snapshot->snapshot_id);
ASSERT_THAT(manager->Commit(), IsOk());
ASSERT_THAT(table_->Refresh(), IsOk());

ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
ICEBERG_UNWRAP_OR_FAIL(auto append, txn->NewFastAppend());
append->AppendFile(file_a_);
ASSERT_THAT(append->Commit(), IsOk());
ICEBERG_UNWRAP_OR_FAIL(auto staged, txn->current().Snapshot());
SnapshotReader reader(staged.get());
ICEBERG_UNWRAP_OR_FAIL(auto manifests, reader.DataManifests(file_io_));
ICEBERG_UNWRAP_OR_FAIL(auto entries, ReadManifestEntries(manifests.front()));
ICEBERG_UNWRAP_OR_FAIL(auto replacement,
WriteExistingManifest("stale-inheritance", entries));
ICEBERG_UNWRAP_OR_FAIL(auto rewrite, txn->NewRewriteManifests());
rewrite->DeleteManifest(manifests.front()).AddManifest(replacement);
ASSERT_THAT(rewrite->Commit(), IsOk());

ICEBERG_UNWRAP_OR_FAIL(auto concurrent, table_->NewFastAppend());
concurrent->ToBranch("audit");
concurrent->AppendFile(file_d_);
ASSERT_THAT(concurrent->Commit(), IsOk());
ASSERT_THAT(table_->Refresh(), IsOk());
auto result = txn->Commit();
EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed));
EXPECT_THAT(result, HasErrorMessage("Deleted manifest metadata changed"));
EXPECT_EQ(txn->state(), TransactionState::kFailed);
EXPECT_TRUE(FileExists(replacement.manifest_path));
}

TEST_P(RewriteManifestsTest, RewriteDataManifestsPreservesDeletes) {
if (GetParam() == 1) {
GTEST_SKIP() << "Delete manifests require format version 2 or higher";
Expand Down
Loading
Loading