From e205346a32ad50b441a6923e60ec68533b72e165 Mon Sep 17 00:00:00 2001 From: Gang Wu Date: Mon, 9 Nov 2026 09:49:30 +0800 Subject: [PATCH 1/2] fix: preserve staged manifest dependencies during transaction replay Transaction replay currently calls `CleanupUpdates()` before reapplying pending updates. This deletes staged files that later updates in the same transaction may still reference. For example: 1. A staged append creates manifest M0. 2. A later `RewriteManifests` replaces M0 with a caller-provided manifest. 3. An unrelated property update forces the transaction to replay. 4. Cleanup deletes M0, and the append generates a different manifest path. 5. The rewrite fails because its declared source manifest no longer exists. Simply retaining all cached state is also unsafe. A manifest can keep the same path while inheriting different sequence numbers or row IDs on replay. Derived caches and explicit replacements must not silently carry values from an earlier attempt into the committed snapshot. Separate replay preparation from terminal file cleanup: - Add `PrepareForReplay()` to reset attempt-dependent state without deleting staged files. FastAppend and MergingSnapshotUpdate reuse their immutable append outputs. MergingSnapshotUpdate clears derived filter/merge caches, and RewriteManifests rebuilds its rewrite outputs. - Validate the sequence number, minimum sequence number, and first row ID of explicitly replaced source manifests. Changed inheritance metadata causes a validation failure. - Preserve copied-manifest summaries across attempts. Resume incomplete copy batches from the first uncopied input, and merge each copy's summary only after that copy succeeds. - Centralize generated-file cleanup in SnapshotUpdate's staging registry, including owned merged-DV files. After success, protect files referenced by every retained snapshot added by the transaction, including intermediate and staged-only snapshots. - Derive expected snapshot IDs from the final `AddSnapshot` changes. Skip snapshot-file cleanup if committed metadata omits any expected snapshot or file discovery fails, so a partial keep-set cannot delete live files. Known failures and aborts clean staged files; unknown outcomes preserve them. - Reset per-attempt reporting snapshots and expiration cleanup plans. A cherry-pick that becomes a fast-forward must not report an abandoned draft, and a no-op expiration must not execute an earlier cleanup plan. Redundant cleanup helpers and file tracking are removed, and the cleanup contract now explicitly leaves physical deletion to the staging registry. Regression coverage includes repeated append-to-rewrite replay, inherited metadata changes across branches, stale replacement rejection, copied manifest reuse and copy resumption, intermediate snapshot retention, and cherry-pick transitions to fast-forward. --- .../test/cherry_pick_operation_test.cc | 13 ++ src/iceberg/test/fast_append_test.cc | 21 ++- .../test/merging_snapshot_update_test.cc | 97 ++++++++++++- src/iceberg/test/rewrite_manifests_test.cc | 81 +++++++++++ src/iceberg/transaction.cc | 101 +++++++++++++- src/iceberg/update/expire_snapshots.cc | 2 + src/iceberg/update/fast_append.cc | 58 +++----- src/iceberg/update/fast_append.h | 4 +- src/iceberg/update/merging_snapshot_update.cc | 130 ++++++++---------- src/iceberg/update/merging_snapshot_update.h | 12 +- src/iceberg/update/pending_update.h | 4 + src/iceberg/update/rewrite_manifests.cc | 71 +++++----- src/iceberg/update/rewrite_manifests.h | 3 +- src/iceberg/update/snapshot_update.cc | 71 ++++------ src/iceberg/update/snapshot_update.h | 31 +++-- 15 files changed, 480 insertions(+), 219 deletions(-) diff --git a/src/iceberg/test/cherry_pick_operation_test.cc b/src/iceberg/test/cherry_pick_operation_test.cc index cb55de9d0..bb0069552 100644 --- a/src/iceberg/test/cherry_pick_operation_test.cc +++ b/src/iceberg/test/cherry_pick_operation_test.cc @@ -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" @@ -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(); + 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); diff --git a/src/iceberg/test/fast_append_test.cc b/src/iceberg/test/fast_append_test.cc index d75e1ef23..3efceaa78 100644 --- a/src/iceberg/test/fast_append_test.cc +++ b/src/iceberg/test/fast_append_test.cc @@ -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()), @@ -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])); @@ -487,12 +495,13 @@ TEST_F(FastAppendTest, RebaseCopiesAppendManifestAgain) { std::unordered_set 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()); } @@ -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); } } diff --git a/src/iceberg/test/merging_snapshot_update_test.cc b/src/iceberg/test/merging_snapshot_update_test.cc index d31a1eb29..0b587589d 100644 --- a/src/iceberg/test/merging_snapshot_update_test.cc +++ b/src/iceberg/test/merging_snapshot_update_test.cc @@ -1380,7 +1380,62 @@ TEST_F(MergingSnapshotUpdateTest, AddManifestCopiesManifestWithAssignedSnapshotI EXPECT_NE(data_manifests[0].manifest_path, path); } -TEST_F(MergingSnapshotUpdateTest, AddManifestRetryCopiesManifestAgain) { +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; @@ -1390,12 +1445,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); }); @@ -1412,8 +1475,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()); @@ -1424,6 +1486,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_})); diff --git a/src/iceberg/test/rewrite_manifests_test.cc b/src/iceberg/test/rewrite_manifests_test.cc index 1bf4dd0d3..9c746ba75 100644 --- a/src/iceberg/test/rewrite_manifests_test.cc +++ b/src/iceberg/test/rewrite_manifests_test.cc @@ -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"; diff --git a/src/iceberg/transaction.cc b/src/iceberg/transaction.cc index eada4eb90..88672af1d 100644 --- a/src/iceberg/transaction.cc +++ b/src/iceberg/transaction.cc @@ -21,11 +21,16 @@ #include #include #include +#include +#include #include +#include #include "iceberg/catalog.h" #include "iceberg/location_provider.h" #include "iceberg/logging/log_macros.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_reader.h" #include "iceberg/schema.h" #include "iceberg/snapshot.h" #include "iceberg/statistics_file.h" @@ -99,6 +104,60 @@ std::string FormatCommittedSnapshots( return detail; } +/// Collect files that must survive transaction finalization. +/// +/// The set includes manifests and manifest lists from all snapshots created by +/// this transaction. It also includes owned content files still referenced by +/// retained delete manifests. +/// +/// \param metadata The metadata after the transaction commit. +/// \param io FileIO used to read manifest lists and delete manifests. +/// \param snapshot_ids Snapshot IDs in the committed AddSnapshot changes. +/// \param content_paths Owned content files that may need live-entry checks. +Result> CommittedTransactionFiles( + const TableMetadata& metadata, const std::shared_ptr& io, + std::unordered_set snapshot_ids, + const std::unordered_set& content_paths) { + std::unordered_set committed; + for (const auto& snapshot : metadata.snapshots) { + if (!snapshot || snapshot_ids.erase(snapshot->snapshot_id) == 0) { + continue; + } + committed.insert(snapshot->manifest_list); + SnapshotReader snapshot_reader(snapshot.get()); + ICEBERG_ASSIGN_OR_RAISE(auto manifests, snapshot_reader.Manifests(io)); + for (const auto& manifest : manifests) { + if (!committed.insert(manifest.manifest_path).second || content_paths.empty() || + manifest.content != ManifestContent::kDeletes) { + continue; + } + // A merged DV can be shared by several retained snapshots. + 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 reader, + ManifestReader::Make(manifest, io, schema, spec)); + ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->LiveEntriesStream()); + while (true) { + ICEBERG_ASSIGN_OR_RAISE(auto entry, entries->Next()); + if (!entry) { + break; + } + ICEBERG_CHECK(entry->data_file != nullptr, + "Missing content file in committed manifest {}", + manifest.manifest_path); + if (content_paths.contains(entry->data_file->file_path)) { + committed.insert(entry->data_file->file_path); + } + } + } + } + // A catalog may omit snapshots. Never clean files with a partial keep-set. + ICEBERG_CHECK(snapshot_ids.empty(), "Committed metadata is missing {} added snapshots", + snapshot_ids.size()); + return committed; +} + } // namespace // --------------------------------------------------------------------------- @@ -222,6 +281,7 @@ Status Transaction::ReplayUpdates() { ICEBERG_CHECK(state_ == TransactionState::kReady, "Replay requires a ready transaction"); for (const auto& update : pending_updates_) { + ICEBERG_RETURN_UNEXPECTED(update->PrepareForReplay()); ICEBERG_RETURN_UNEXPECTED(ApplyUpdate(*update)); } return {}; @@ -357,6 +417,8 @@ Status Transaction::ApplyUpdateSchema(UpdateSchema& update) { Status Transaction::ApplyUpdateSnapshot(SnapshotUpdate& update) { const auto& base = ctx_->metadata_builder->current(); + // Report only snapshots generated by this attempt. + update.staged_snapshot_.reset(); ICEBERG_ASSIGN_OR_RAISE(auto result, update.Apply()); if (result.snapshot == nullptr) { return {}; @@ -375,10 +437,8 @@ Status Transaction::ApplyUpdateSnapshot(SnapshotUpdate& update) { ICEBERG_RETURN_UNEXPECTED(temp_update->CheckErrors()); if (temp_update->changes().empty()) { - // A no-op may still write temporary files. Clean them now; the update remains - // registered so it can be replayed after a refresh. - internal::LogAndIgnoreFailure("Update staging cleanup", - [&update] { return update.CleanStaged(); }); + // A later replay may reuse these files. Terminal cleanup will remove leftovers. + update.staged_snapshot_.reset(); return {}; } @@ -534,7 +594,39 @@ void Transaction::CleanupUpdates() noexcept { } void Transaction::FinalizeUpdates(const TableMetadata& committed) noexcept { + std::optional> committed_files; + internal::LogAndIgnoreFailure("Transaction file discovery", [&]() -> Status { + std::unordered_set snapshot_ids; + for (const auto& change : ctx_->metadata_builder->changes()) { + if (change->kind() == TableUpdate::Kind::kAddSnapshot) { + const auto& snapshot = + internal::checked_cast(*change).snapshot(); + snapshot_ids.insert(snapshot->snapshot_id); + } + } + std::unordered_set content_paths; + for (const auto& update : pending_updates_) { + if (update->kind() != PendingUpdate::Kind::kUpdateSnapshot) { + continue; + } + auto& snapshot_update = internal::checked_cast(*update); + std::lock_guard lock(snapshot_update.staging_mutex_); + content_paths.insert(snapshot_update.staged_content_files_.begin(), + snapshot_update.staged_content_files_.end()); + } + ICEBERG_ASSIGN_OR_RAISE(committed_files, CommittedTransactionFiles( + committed, ctx_->table->io(), + std::move(snapshot_ids), content_paths)); + return {}; + }); for (const auto& update : pending_updates_) { + if (update->kind() == PendingUpdate::Kind::kUpdateSnapshot && committed_files) { + // Failed discovery must never turn into an empty keep-set. + internal::LogAndIgnoreFailure("Snapshot cleanup", [&] { + return internal::checked_cast(*update).CleanUncommittedFiles( + *committed_files); + }); + } internal::LogAndIgnoreFailure("Update finalization", [&update, &committed] { return update->Finalize(committed); }); @@ -556,7 +648,6 @@ Result> Transaction::CommitOnce(bool is_first_attempt) { if (metadata_changed || standalone_retry) { ICEBERG_CHECK(CanRetry(), "Cannot rebase a transaction containing a non-retryable update"); - CleanupUpdates(); ctx_->metadata_builder = TableMetadataBuilder::BuildFrom(ctx_->table->metadata().get()); auto applied = ReplayUpdates(); diff --git a/src/iceberg/update/expire_snapshots.cc b/src/iceberg/update/expire_snapshots.cc index fd14308e8..036531dd9 100644 --- a/src/iceberg/update/expire_snapshots.cc +++ b/src/iceberg/update/expire_snapshots.cc @@ -918,6 +918,8 @@ Result ExpireSnapshots::ComputeRetainedRefs( } Result ExpireSnapshots::Apply() { + // A no-op apply must not leave an earlier cleanup plan active. + apply_result_.reset(); ICEBERG_RETURN_UNEXPECTED(CheckErrors()); const TableMetadata& base = this->base(); diff --git a/src/iceberg/update/fast_append.cc b/src/iceberg/update/fast_append.cc index 1cb1e41cf..60c63e57b 100644 --- a/src/iceberg/update/fast_append.cc +++ b/src/iceberg/update/fast_append.cc @@ -97,13 +97,12 @@ Result> FastAppend::Apply( } ICEBERG_ASSIGN_OR_RAISE(auto new_written_manifests, WriteNewManifests()); - // A retry cleanup deletes copied append manifests and clears the rewritten - // list; rebuild them from the original appended manifests before re-applying. - if (rewritten_append_manifests_.empty() && !append_manifests_to_copy_.empty()) { - for (const auto& manifest : append_manifests_to_copy_) { - ICEBERG_ASSIGN_OR_RAISE(auto copied_manifest, CopyManifest(manifest)); - rewritten_append_manifests_.push_back(std::move(copied_manifest)); - } + // Immutable copies and their summaries are reused across replay attempts. + for (size_t i = rewritten_append_manifests_.size(); + i < append_manifests_to_copy_.size(); ++i) { + ICEBERG_ASSIGN_OR_RAISE(auto copied_manifest, + CopyManifest(append_manifests_to_copy_[i])); + rewritten_append_manifests_.push_back(std::move(copied_manifest)); } manifests.reserve(new_written_manifests.size() + append_manifests_.size() + @@ -147,6 +146,7 @@ std::unordered_map FastAppend::Summary() { base().properties.Get(TableProperties::kWritePartitionSummaryLimit)); summary_.Merge(added_data_files_summary_); summary_.Merge(appended_manifests_summary_); + summary_.Merge(copied_manifests_summary_); for (const auto& [property, value] : custom_summary_properties_) { summary_.Set(property, value); } @@ -160,28 +160,11 @@ void FastAppend::SetSummaryProperty(const std::string& property, SnapshotUpdate::SetSummaryProperty(property, value); } -Status FastAppend::CleanUncommitted(const std::unordered_set& committed) { - // Clean up new manifests that were written but not committed - if (!new_manifests_.empty()) { - for (const auto& manifest : new_manifests_) { - if (!committed.contains(manifest.manifest_path)) { - std::ignore = DeleteFile(manifest.manifest_path); - } - } - new_manifests_.clear(); - } - - // Clean up only rewritten append manifests as they are always owned by the table - // Don't clean up append manifests as they are added to the manifest list and are - // not compacted - if (!rewritten_append_manifests_.empty()) { - for (const auto& manifest : rewritten_append_manifests_) { - if (!committed.contains(manifest.manifest_path)) { - std::ignore = DeleteFile(manifest.manifest_path); - } - } - rewritten_append_manifests_.clear(); - } +Status FastAppend::CleanUncommitted( + const std::unordered_set& /*committed*/) { + new_manifests_.clear(); + rewritten_append_manifests_.clear(); + copied_manifests_summary_.Clear(); return {}; } @@ -199,18 +182,19 @@ Result FastAppend::CopyManifest(const ManifestFile& manifest) { std::string new_manifest_path = ManifestPath(); int64_t snapshot_id = SnapshotId(); - // Copy the manifest with the new snapshot ID. - return CopyAppendManifest(manifest, ctx_->table->io(), schema, spec, snapshot_id, - new_manifest_path, current.format_version, - &appended_manifests_summary_); + // A failed copy must not leave partial counts in the cached summary. + SnapshotSummaryBuilder summary; + ICEBERG_ASSIGN_OR_RAISE( + auto copied, + CopyAppendManifest(manifest, ctx_->table->io(), schema, spec, snapshot_id, + new_manifest_path, current.format_version, &summary)); + copied_manifests_summary_.Merge(summary); + return copied; } Result> FastAppend::WriteNewManifests() { - // If there are new files and manifests were already written, clean them up + // New additions invalidate the cache. Old paths stay staged until cleanup. if (has_new_files_ && !new_manifests_.empty()) { - for (const auto& manifest : new_manifests_) { - std::ignore = DeleteFile(manifest.manifest_path); - } new_manifests_.clear(); } diff --git a/src/iceberg/update/fast_append.h b/src/iceberg/update/fast_append.h index e20576654..81682578a 100644 --- a/src/iceberg/update/fast_append.h +++ b/src/iceberg/update/fast_append.h @@ -106,10 +106,12 @@ class ICEBERG_EXPORT FastAppend : public SnapshotUpdate { // Stable input summaries for retry-safe summary_ rebuilds. SnapshotSummaryBuilder added_data_files_summary_; SnapshotSummaryBuilder appended_manifests_summary_; + // Copies may be reused across attempts, so their summary must survive too. + SnapshotSummaryBuilder copied_manifests_summary_; // User-provided summary properties restored after summary_ rebuilds. std::unordered_map custom_summary_properties_; std::vector append_manifests_; - // Original manifests kept to recreate copied manifests after retry cleanup. + // Original inputs for manifests that must be copied. std::vector append_manifests_to_copy_; std::vector rewritten_append_manifests_; std::vector new_manifests_; diff --git a/src/iceberg/update/merging_snapshot_update.cc b/src/iceberg/update/merging_snapshot_update.cc index 154a27b91..0033fd4fa 100644 --- a/src/iceberg/update/merging_snapshot_update.cc +++ b/src/iceberg/update/merging_snapshot_update.cc @@ -411,8 +411,10 @@ MergingSnapshotUpdate::MergingSnapshotUpdate(std::string table_name, delete_expression_(Expressions::AlwaysFalse()) { auto file_io = ctx_->table->io(); auto data_filter_manager = ManifestFilterManager::Make( - ManifestContent::kData, file_io, - [this](const std::string& location) { return DeleteFile(location); }); + ManifestContent::kData, file_io, [this](const std::string& location) { + RegisterStagedFile(location); + return Status{}; + }); if (!data_filter_manager.has_value()) { AddError(data_filter_manager.error()); } else { @@ -420,8 +422,10 @@ MergingSnapshotUpdate::MergingSnapshotUpdate(std::string table_name, } auto delete_filter_manager = ManifestFilterManager::Make( - ManifestContent::kDeletes, file_io, - [this](const std::string& location) { return DeleteFile(location); }); + ManifestContent::kDeletes, file_io, [this](const std::string& location) { + RegisterStagedFile(location); + return Status{}; + }); if (!delete_filter_manager.has_value()) { AddError(delete_filter_manager.error()); } else { @@ -437,7 +441,10 @@ MergingSnapshotUpdate::MergingSnapshotUpdate(std::string table_name, auto data_merge_manager = ManifestMergeManager::Make( ManifestContent::kData, target_size_bytes, min_count_to_merge, merge_enabled, file_io, [this] { return SnapshotId(); }, - [this](const std::string& location) { return DeleteFile(location); }); + [this](const std::string& location) { + RegisterStagedFile(location); + return Status{}; + }); if (!data_merge_manager.has_value()) { AddError(data_merge_manager.error()); } else { @@ -447,7 +454,10 @@ MergingSnapshotUpdate::MergingSnapshotUpdate(std::string table_name, auto delete_merge_manager = ManifestMergeManager::Make( ManifestContent::kDeletes, target_size_bytes, min_count_to_merge, merge_enabled, file_io, [this] { return SnapshotId(); }, - [this](const std::string& location) { return DeleteFile(location); }); + [this](const std::string& location) { + RegisterStagedFile(location); + return Status{}; + }); if (!delete_merge_manager.has_value()) { AddError(delete_merge_manager.error()); } else { @@ -675,8 +685,14 @@ Result MergingSnapshotUpdate::CopyManifest(const ManifestFile& man ICEBERG_ASSIGN_OR_RAISE(auto spec, current.PartitionSpecById(manifest.partition_spec_id)); std::string path = ManifestPath(); - return CopyAppendManifest(manifest, ctx_->table->io(), schema, spec, SnapshotId(), path, - current.format_version, &appended_manifests_summary_); + // A failed copy must not leave partial counts in the cached summary. + SnapshotSummaryBuilder summary; + ICEBERG_ASSIGN_OR_RAISE( + auto copied, + CopyAppendManifest(manifest, ctx_->table->io(), schema, spec, SnapshotId(), path, + current.format_version, &summary)); + copied_manifests_summary_.Merge(summary); + return copied; } // ------------------------------------------------------------------------- @@ -729,11 +745,8 @@ ManifestWriterFactory MergingSnapshotUpdate::MakeWriterFactory( } Result> MergingSnapshotUpdate::WriteNewDataManifests() { - // If new files were staged after the cache was populated (commit retry), invalidate. + // New additions invalidate the cache. Old paths stay staged until cleanup. if (has_new_data_files_ && !cached_new_data_manifests_.empty()) { - for (const auto& m : cached_new_data_manifests_) { - std::ignore = DeleteFile(m.manifest_path); - } cached_new_data_manifests_.clear(); } @@ -819,10 +832,9 @@ MergingSnapshotUpdate::MergeDVs() { auto output_path = location_provider->NewDataLocation( std::format("merged-dvs-{}-{}.puffin", SnapshotId(), ++dv_merge_attempt_)); - RegisterStagedFile(output_path); + RegisterStagedContentFile(output_path); auto merged_files = DVUtil::MergeAndWriteDVs(groups, output_path, ctx_->table->io()); if (!merged_files) { - std::ignore = DeleteFile(output_path); return ::iceberg::unexpected(std::move(merged_files.error())); } @@ -846,30 +858,16 @@ MergingSnapshotUpdate::MergeDVs() { groups[pos].referenced_data_file); PendingDeleteFile merged{.file = std::move(iter->second), .data_sequence_number = data_sequence_numbers[pos]}; - merged_dvs_.push_back(merged); result.push_back(std::move(merged)); } - // The operation-specific merged DV cache now owns cleanup for this file. - UnregisterStagedFile(output_path); - return result; } Result> MergingSnapshotUpdate::WriteNewDeleteManifests() { - // If new files were staged after the cache was populated (commit retry), invalidate. + // New additions invalidate the cache. Old paths stay staged until cleanup. if (has_new_delete_files_ && !cached_new_delete_manifests_.empty()) { - for (const auto& m : cached_new_delete_manifests_) { - std::ignore = DeleteFile(m.manifest_path); - } cached_new_delete_manifests_.clear(); - std::unordered_set deleted_merged_dv_paths; - for (const auto& dv : merged_dvs_) { - if (deleted_merged_dv_paths.insert(dv.file->file_path).second) { - std::ignore = DeleteFile(dv.file->file_path); - } - } - merged_dvs_.clear(); added_delete_files_summary_.Clear(); } @@ -921,6 +919,12 @@ Result> MergingSnapshotUpdate::WriteNewDeleteManifests return result; } +Status MergingSnapshotUpdate::PrepareForReplay() { + // The same path can inherit new sequence numbers and row IDs on replay. + // Drop derived caches, but keep their files staged for terminal cleanup. + return CleanUncommittedManagers({}); +} + Result> MergingSnapshotUpdate::Apply( const TableMetadata& metadata_to_update, const std::shared_ptr& snapshot) { appended_manifests_summary_.Clear(); @@ -990,11 +994,10 @@ Result> MergingSnapshotUpdate::Apply( // Step 4: Write (or retrieve cached) new data manifests. ICEBERG_ASSIGN_OR_RAISE(auto written_data_manifests, WriteNewDataManifests()); - if (rewritten_append_manifests_.empty() && !append_manifests_to_copy_.empty()) { - for (const auto& manifest : append_manifests_to_copy_) { - ICEBERG_ASSIGN_OR_RAISE(auto copied, CopyManifest(manifest)); - rewritten_append_manifests_.push_back(std::move(copied)); - } + for (size_t i = rewritten_append_manifests_.size(); + i < append_manifests_to_copy_.size(); ++i) { + ICEBERG_ASSIGN_OR_RAISE(auto copied, CopyManifest(append_manifests_to_copy_[i])); + rewritten_append_manifests_.push_back(std::move(copied)); } // Incorporate append manifests (from AddManifest), stamping each with the @@ -1027,6 +1030,7 @@ Result> MergingSnapshotUpdate::Apply( summary_builder().Merge(added_data_files_summary_); summary_builder().Merge(added_delete_files_summary_); summary_builder().Merge(appended_manifests_summary_); + summary_builder().Merge(copied_manifests_summary_); for (const auto& [property, value] : custom_summary_properties_) { summary_builder().Set(property, value); } @@ -1061,57 +1065,39 @@ Result> MergingSnapshotUpdate::Apply( Status MergingSnapshotUpdate::CleanUncommitted( const std::unordered_set& committed) { + ICEBERG_RETURN_UNEXPECTED(CleanUncommittedManagers(committed)); + CleanUncommittedAppends(committed); + return {}; +} + +Status MergingSnapshotUpdate::CleanUncommittedManagers( + const std::unordered_set& committed) { ICEBERG_RETURN_UNEXPECTED(ManagersReady()); ICEBERG_RETURN_UNEXPECTED(data_merge_manager_->CleanUncommitted(committed)); ICEBERG_RETURN_UNEXPECTED(data_filter_manager_->CleanUncommitted(committed)); ICEBERG_RETURN_UNEXPECTED(delete_merge_manager_->CleanUncommitted(committed)); - ICEBERG_RETURN_UNEXPECTED(delete_filter_manager_->CleanUncommitted(committed)); - ICEBERG_RETURN_UNEXPECTED(CleanUncommittedAppends(committed)); - return {}; + return delete_filter_manager_->CleanUncommitted(committed); } -Status MergingSnapshotUpdate::CleanUncommittedAppends( +void MergingSnapshotUpdate::CleanUncommittedAppends( const std::unordered_set& committed) { - ICEBERG_RETURN_UNEXPECTED( - DeleteUncommitted(cached_new_data_manifests_, committed, /*clear=*/true)); - ICEBERG_RETURN_UNEXPECTED( - DeleteUncommitted(cached_new_delete_manifests_, committed, /*clear=*/true)); - // rewritten_append_manifests_ are always owned by the table. - ICEBERG_RETURN_UNEXPECTED( - DeleteUncommitted(rewritten_append_manifests_, committed, /*clear=*/true)); - if (committed.empty()) { - std::unordered_set deleted_merged_dv_paths; - for (const auto& dv : merged_dvs_) { - if (deleted_merged_dv_paths.insert(dv.file->file_path).second) { - std::ignore = DeleteFile(dv.file->file_path); - } - } - } - merged_dvs_.clear(); + // Generated files are already tracked by SnapshotUpdate. + cached_new_data_manifests_.clear(); + cached_new_delete_manifests_.clear(); + rewritten_append_manifests_.clear(); + copied_manifests_summary_.Clear(); - // append_manifests_ are only owned by the table if the commit succeeded. + // External append manifests become table-owned only after a successful commit. if (!committed.empty()) { - ICEBERG_RETURN_UNEXPECTED( - DeleteUncommitted(append_manifests_, committed, /*clear=*/false)); + for (const auto& manifest : append_manifests_) { + if (!committed.contains(manifest.manifest_path)) { + RegisterStagedFile(manifest.manifest_path); + } + } } has_new_data_files_ = false; has_new_delete_files_ = false; - return {}; -} - -Status MergingSnapshotUpdate::DeleteUncommitted( - std::vector& manifests, - const std::unordered_set& committed, bool clear) { - for (const auto& manifest : manifests) { - if (!committed.contains(manifest.manifest_path)) { - std::ignore = DeleteFile(manifest.manifest_path); - } - } - if (clear) { - manifests.clear(); - } - return {}; } std::unordered_map MergingSnapshotUpdate::Summary() { diff --git a/src/iceberg/update/merging_snapshot_update.h b/src/iceberg/update/merging_snapshot_update.h index e1cd7303b..3938a9d1e 100644 --- a/src/iceberg/update/merging_snapshot_update.h +++ b/src/iceberg/update/merging_snapshot_update.h @@ -71,6 +71,7 @@ class ICEBERG_EXPORT MergingSnapshotUpdate : public SnapshotUpdate { const std::shared_ptr& snapshot) override; Status CleanUncommitted(const std::unordered_set& committed) override; + Status PrepareForReplay() override; std::unordered_map Summary() override; @@ -348,10 +349,8 @@ class ICEBERG_EXPORT MergingSnapshotUpdate : public SnapshotUpdate { /// \brief Write new delete manifests for staged delete files; caches the result. Result> WriteNewDeleteManifests(); - Status CleanUncommittedAppends(const std::unordered_set& committed); - - Status DeleteUncommitted(std::vector& manifests, - const std::unordered_set& committed, bool clear); + Status CleanUncommittedManagers(const std::unordered_set& committed); + void CleanUncommittedAppends(const std::unordered_set& committed); // Used for commit event notifications and diagnostic log messages. std::string table_name_; @@ -363,6 +362,8 @@ class ICEBERG_EXPORT MergingSnapshotUpdate : public SnapshotUpdate { SnapshotSummaryBuilder added_data_files_summary_; SnapshotSummaryBuilder added_delete_files_summary_; SnapshotSummaryBuilder appended_manifests_summary_; + // Copies may be reused across attempts, so their summary must survive too. + SnapshotSummaryBuilder copied_manifests_summary_; std::unordered_map custom_summary_properties_; std::unique_ptr data_filter_manager_; @@ -378,7 +379,7 @@ class ICEBERG_EXPORT MergingSnapshotUpdate : public SnapshotUpdate { // Manifests passed via AddManifest(): inherit path (no copy needed) and // rewrite path (must be copied with the current snapshot ID). std::vector append_manifests_; - // Original manifests kept to recreate copied manifests after retry cleanup. + // Original inputs for manifests that must be copied. std::vector append_manifests_to_copy_; std::vector rewritten_append_manifests_; @@ -389,7 +390,6 @@ class ICEBERG_EXPORT MergingSnapshotUpdate : public SnapshotUpdate { std::vector cached_new_data_manifests_; std::vector cached_new_delete_manifests_; - std::vector merged_dvs_; int32_t dv_merge_attempt_ = 0; }; diff --git a/src/iceberg/update/pending_update.h b/src/iceberg/update/pending_update.h index 81bf4563e..972f77c9e 100644 --- a/src/iceberg/update/pending_update.h +++ b/src/iceberg/update/pending_update.h @@ -93,6 +93,10 @@ class ICEBERG_EXPORT PendingUpdate : public ErrorCollector, /// \brief Discard this generation's staging, retaining applied operation intent. virtual Status CleanStaged() { return {}; } + + /// \brief Prepare retry-safe state after the transaction metadata is refreshed. + virtual Status PrepareForReplay() { return {}; } + /// \brief Complete a known successful commit. Called only by Transaction with the /// final committed table metadata. virtual Status Finalize(const TableMetadata& committed); diff --git a/src/iceberg/update/rewrite_manifests.cc b/src/iceberg/update/rewrite_manifests.cc index 653663b61..38abe8d93 100644 --- a/src/iceberg/update/rewrite_manifests.cc +++ b/src/iceberg/update/rewrite_manifests.cc @@ -84,6 +84,19 @@ Status CatchStatus(std::string_view operation, Fn&& fn) { } } +Status FinishWithCloseStatus(Status rewrite_status, Status close_status) { + // Keep the rewrite error when both operations fail. + if (!rewrite_status.has_value()) { + auto error = std::move(rewrite_status).error(); + if (!close_status.has_value()) { + error.message += "; additionally failed to close manifest writer: "; + error.message += close_status.error().message; + } + return ::iceberg::unexpected(std::move(error)); + } + return close_status; +} + } // namespace class RewriteManifests::WriterWrapper { @@ -177,7 +190,7 @@ RewriteManifests& RewriteManifests::AddManifest(const ManifestFile& manifest) { if (can_inherit_snapshot_id() && !manifest.added_snapshot_id.has_value()) { added_manifests_.push_back(manifest); } else { - // Retain the input so Apply can recreate the copy after retry cleanup. + // Retain the input so replay can rebuild the copy. added_manifests_to_copy_.push_back(manifest); } return *this; @@ -185,6 +198,11 @@ RewriteManifests& RewriteManifests::AddManifest(const ManifestFile& manifest) { std::string RewriteManifests::operation() { return DataOperation::kReplace; } +Status RewriteManifests::PrepareForReplay() { + // Rebuild against the latest manifest list on every replay. + return CleanUncommitted({}); +} + Result> RewriteManifests::Apply( const TableMetadata& /*metadata_to_update*/, const std::shared_ptr& snapshot) { @@ -262,12 +280,9 @@ std::unordered_map RewriteManifests::Summary() { } Status RewriteManifests::CleanUncommitted( - const std::unordered_set& committed) { - DeleteUncommitted(new_manifests_, committed); - DeleteUncommitted(rewritten_added_manifests_, committed); - entry_count_ = 0; - kept_manifests_.clear(); - rewritten_manifests_.clear(); + const std::unordered_set& /*committed*/) { + Reset(); + rewritten_added_manifests_.clear(); return {}; } @@ -311,11 +326,20 @@ Status RewriteManifests::ValidateDeletedManifests( int64_t current_snapshot_id) const { // Directly deleted manifests must still be present in the current snapshot. for (const auto& manifest : deleted_manifests_) { - if (!current_manifests.contains(manifest)) { + auto current = current_manifests.find(manifest); + if (current == current_manifests.end()) { return ValidationFailed( "Deleted manifest {} could not be found in the latest snapshot {}", manifest.manifest_path, current_snapshot_id); } + // A prior update may reuse this path with new inherited values on replay. + // The caller's replacement still contains the original values. + if (manifest.sequence_number != current->sequence_number || + manifest.min_sequence_number != current->min_sequence_number || + manifest.first_row_id != current->first_row_id) { + return ValidationFailed("Deleted manifest metadata changed for {} in snapshot {}", + manifest.manifest_path, current_snapshot_id); + } } return {}; } @@ -467,36 +491,19 @@ Status RewriteManifests::PerformRewrite(std::span current_ma return process_batch(); }); - auto close_status = CloseWriters(); - - if (!write_status.has_value()) { - auto error = std::move(write_status).error(); - if (!close_status.has_value()) { - error.message += "; additionally failed to close manifest writer: "; - error.message += close_status.error().message; - } - return ::iceberg::unexpected(std::move(error)); - } - if (!close_status.has_value()) { - return close_status; + // Close every writer before returning the first failure. + auto status = FinishWithCloseStatus(std::move(write_status), CloseWriters()); + if (!status.has_value()) { + return status; } rewrite_dirty_ = false; return {}; } -void RewriteManifests::DeleteUncommitted( - std::vector& manifests, - const std::unordered_set& committed) { - for (const auto& manifest : manifests) { - if (!committed.contains(manifest.manifest_path)) { - std::ignore = DeleteFile(manifest.manifest_path); - } - } - manifests.clear(); -} - void RewriteManifests::Reset() { - DeleteUncommitted(new_manifests_, {}); + // Replay keeps staged outputs alive until the transaction reaches a terminal + // state. Clear only the in-memory generation. + new_manifests_.clear(); entry_count_ = 0; kept_manifests_.clear(); rewritten_manifests_.clear(); diff --git a/src/iceberg/update/rewrite_manifests.h b/src/iceberg/update/rewrite_manifests.h index 669e0fb1f..3aa968cb9 100644 --- a/src/iceberg/update/rewrite_manifests.h +++ b/src/iceberg/update/rewrite_manifests.h @@ -102,6 +102,7 @@ class ICEBERG_EXPORT RewriteManifests : public SnapshotUpdate { const std::shared_ptr& snapshot) override; std::unordered_map Summary() override; Status CleanUncommitted(const std::unordered_set& committed) override; + Status PrepareForReplay() override; explicit RewriteManifests(std::string table_name, std::shared_ptr ctx); @@ -139,8 +140,6 @@ class ICEBERG_EXPORT RewriteManifests : public SnapshotUpdate { const std::shared_ptr& spec); Status CloseWriters(); - void DeleteUncommitted(std::vector& manifests, - const std::unordered_set& committed); void Reset(); private: diff --git a/src/iceberg/update/snapshot_update.cc b/src/iceberg/update/snapshot_update.cc index d6a1066ac..f493c85dc 100644 --- a/src/iceberg/update/snapshot_update.cc +++ b/src/iceberg/update/snapshot_update.cc @@ -384,43 +384,33 @@ Result SnapshotUpdate::Apply() { .stage_only = stage_only_}; } -Status SnapshotUpdate::Finalize([[maybe_unused]] const TableMetadata& metadata) { - if (staged_snapshot_ == nullptr) { - return {}; - } - - // Only files created by this update are tracked, and committed paths are kept below. - internal::LogAndIgnoreFailure("Snapshot cleanup", [this]() -> Status { - auto snapshot_reader = SnapshotReader(staged_snapshot_.get()); - ICEBERG_ASSIGN_OR_RAISE(auto manifests, snapshot_reader.Manifests(ctx_->table->io())); - auto committed = manifests | std::views::transform([](const auto& manifest) { - return manifest.manifest_path; - }) | - std::ranges::to>(); - internal::LogAndIgnoreFailure("Snapshot operation cleanup", [this, &committed] { - return CleanUncommitted(committed); - }); - committed.insert(staged_snapshot_->manifest_list); - std::vector unused; - { - std::lock_guard lock(staging_mutex_); - for (const auto& path : staged_files_) { - if (!committed.contains(path)) { - unused.push_back(path); - } - } - } - for (const auto& path : unused) { - std::ignore = DeleteFile(path); - } - return {}; +Status SnapshotUpdate::CleanUncommittedFiles( + const std::unordered_set& committed) { + // All retained snapshots share this set. A later update may reuse our output. + internal::LogAndIgnoreFailure("Snapshot operation cleanup", [this, &committed] { + return CleanUncommitted(committed); }); - + std::vector unused; { std::lock_guard lock(staging_mutex_); - staged_files_.clear(); + for (const auto& path : staged_files_) { + if (!committed.contains(path)) { + unused.push_back(path); + } + } + for (const auto& path : committed) { + staged_files_.erase(path); + staged_content_files_.erase(path); + } + } + for (const auto& path : unused) { + std::ignore = DeleteStagedFile(path); } - auto report_status = ReportCommit(); + return {}; +} + +Status SnapshotUpdate::Finalize([[maybe_unused]] const TableMetadata& committed) { + auto report_status = staged_snapshot_ ? ReportCommit() : Status{}; staged_snapshot_.reset(); summary_.Clear(); return report_status; @@ -479,14 +469,14 @@ Result> SnapshotUpdate::ComputeSumm Status SnapshotUpdate::CleanStaged() { internal::LogAndIgnoreFailure("Snapshot staging cleanup", [this] { return CleanUncommitted({}); }); - // Retry paths that operation-specific cleanup did not delete. + // Delete all remaining staged paths, including partial writes. std::vector paths; { std::lock_guard lock(staging_mutex_); paths.assign(staged_files_.begin(), staged_files_.end()); } for (const auto& path : paths) { - std::ignore = DeleteFile(path); + std::ignore = DeleteStagedFile(path); } staged_snapshot_.reset(); summary_.Clear(); @@ -498,26 +488,25 @@ void SnapshotUpdate::RegisterStagedFile(const std::string& path) { staged_files_.insert(path); } -void SnapshotUpdate::UnregisterStagedFile(const std::string& path) { +void SnapshotUpdate::RegisterStagedContentFile(const std::string& path) { std::lock_guard lock(staging_mutex_); - staged_files_.erase(path); + staged_files_.insert(path); + staged_content_files_.insert(path); } -Status SnapshotUpdate::DeleteFile(const std::string& path) noexcept { +Status SnapshotUpdate::DeleteStagedFile(const std::string& path) noexcept { try { auto result = delete_func_ ? delete_func_(path) : ctx_->table->io()->DeleteFile(path); if (!result && result.error().kind != ErrorKind::kNotFound) { - RegisterStagedFile(path); ICEBERG_LOG_WARN("Cannot clean staged file {}: {}", path, result.error().message); return {}; } std::lock_guard lock(staging_mutex_); staged_files_.erase(path); + staged_content_files_.erase(path); } catch (const std::exception& e) { - RegisterStagedFile(path); ICEBERG_LOG_WARN("Cannot clean staged file {}: {}", path, e.what()); } catch (...) { - RegisterStagedFile(path); ICEBERG_LOG_WARN("Cannot clean staged file {}: unknown exception", path); } return {}; diff --git a/src/iceberg/update/snapshot_update.h b/src/iceberg/update/snapshot_update.h index 18e81e4f9..42b1ae984 100644 --- a/src/iceberg/update/snapshot_update.h +++ b/src/iceberg/update/snapshot_update.h @@ -236,16 +236,14 @@ class ICEBERG_EXPORT SnapshotUpdate : public PendingUpdate { int32_t attempt() const { return attempt_.load(std::memory_order_relaxed); } int64_t target_manifest_size_bytes() const { return target_manifest_size_bytes_; } - /// \brief Clean up any uncommitted manifests that were created. + /// \brief Clear operation caches and register any remaining owned files. /// - /// Manifests may not be committed if Apply is called multiple times - /// because a commit conflict has occurred. Implementations may keep - /// around manifests because the same changes will be made by both - /// Apply calls. This method instructs the implementation to clean up - /// those manifests and passes the paths of the manifests that were - /// actually committed. + /// Do not delete files here. SnapshotUpdate deletes uncommitted paths through + /// its staging registry. Generated paths must be registered before writing. + /// External inputs become table-owned only after a successful commit. /// - /// \param committed A set of manifest paths that were actually committed + /// \param committed File paths protected by retained transaction snapshots. + /// Empty when discarding all staged files. virtual Status CleanUncommitted(const std::unordered_set& committed) = 0; /// \brief A string that describes the action that produced the new snapshot. @@ -287,11 +285,9 @@ class ICEBERG_EXPORT SnapshotUpdate : public PendingUpdate { /// \brief Get or generate the snapshot ID for the new snapshot. int64_t SnapshotId(); - /// Best-effort delete. Failed paths remain registered for later cleanup. - Status DeleteFile(const std::string& path) noexcept; void RegisterStagedFile(const std::string& path); - /// Transfer cleanup ownership to the derived update. - void UnregisterStagedFile(const std::string& path); + /// Track an owned content file as well as its cleanup path. + void RegisterStagedContentFile(const std::string& path); std::string ManifestPath(); std::string ManifestListPath(); @@ -306,6 +302,8 @@ class ICEBERG_EXPORT SnapshotUpdate : public PendingUpdate { const TableMetadata& previous); Status ReportCommit() const; + Status CleanUncommittedFiles(const std::unordered_set& committed); + Status DeleteStagedFile(const std::string& path) noexcept; protected: SnapshotSummaryBuilder summary_; @@ -364,9 +362,16 @@ class ICEBERG_EXPORT SnapshotUpdate : public PendingUpdate { int32_t write_manifest_parallelism_{1}; std::atomic manifest_count_{0}; std::atomic attempt_{0}; - // All paths are registered before writes, including partial/failed writes. + // Protect both staging sets. All paths are registered before writes, including + // partial and failed writes. std::mutex staging_mutex_; + // Every file path created or owned by this update: manifests, manifest lists, + // and content files. Terminal cleanup may delete paths from this set. std::unordered_set staged_files_; + // A subset of staged_files_ containing owned content files, such as merged + // deletion vectors. Successful cleanup keeps these only when a retained + // manifest still references them. + std::unordered_set staged_content_files_; const int64_t target_manifest_size_bytes_; std::optional snapshot_id_; OptionalExecutor plan_executor_; From 1d981106ce9306b140ee01438ca05aeafb3a0efe Mon Sep 17 00:00:00 2001 From: Gang Wu Date: Mon, 9 Nov 2026 11:20:04 +0800 Subject: [PATCH 2/2] address copilot comments --- .../test/merging_snapshot_update_test.cc | 77 +++++++++++++++++++ src/iceberg/transaction.cc | 15 ++-- src/iceberg/update/snapshot_update.cc | 14 ++++ 3 files changed, 99 insertions(+), 7 deletions(-) diff --git a/src/iceberg/test/merging_snapshot_update_test.cc b/src/iceberg/test/merging_snapshot_update_test.cc index 0b587589d..64b1b8866 100644 --- a/src/iceberg/test/merging_snapshot_update_test.cc +++ b/src/iceberg/test/merging_snapshot_update_test.cc @@ -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" @@ -1380,6 +1382,81 @@ TEST_F(MergingSnapshotUpdateTest, AddManifestCopiesManifestWithAssignedSnapshotI EXPECT_NE(data_manifests[0].manifest_path, path); } +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()); diff --git a/src/iceberg/transaction.cc b/src/iceberg/transaction.cc index 88672af1d..53b5f1bd3 100644 --- a/src/iceberg/transaction.cc +++ b/src/iceberg/transaction.cc @@ -113,11 +113,11 @@ std::string FormatCommittedSnapshots( /// \param metadata The metadata after the transaction commit. /// \param io FileIO used to read manifest lists and delete manifests. /// \param snapshot_ids Snapshot IDs in the committed AddSnapshot changes. -/// \param content_paths Owned content files that may need live-entry checks. +/// \param content_paths Owned files whose live references have not been found. Result> CommittedTransactionFiles( const TableMetadata& metadata, const std::shared_ptr& io, std::unordered_set snapshot_ids, - const std::unordered_set& content_paths) { + std::unordered_set content_paths) { std::unordered_set committed; for (const auto& snapshot : metadata.snapshots) { if (!snapshot || snapshot_ids.erase(snapshot->snapshot_id) == 0) { @@ -138,7 +138,7 @@ Result> CommittedTransactionFiles( ICEBERG_ASSIGN_OR_RAISE(auto reader, ManifestReader::Make(manifest, io, schema, spec)); ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->LiveEntriesStream()); - while (true) { + while (!content_paths.empty()) { ICEBERG_ASSIGN_OR_RAISE(auto entry, entries->Next()); if (!entry) { break; @@ -146,7 +146,7 @@ Result> CommittedTransactionFiles( ICEBERG_CHECK(entry->data_file != nullptr, "Missing content file in committed manifest {}", manifest.manifest_path); - if (content_paths.contains(entry->data_file->file_path)) { + if (content_paths.erase(entry->data_file->file_path) != 0) { committed.insert(entry->data_file->file_path); } } @@ -614,9 +614,10 @@ void Transaction::FinalizeUpdates(const TableMetadata& committed) noexcept { content_paths.insert(snapshot_update.staged_content_files_.begin(), snapshot_update.staged_content_files_.end()); } - ICEBERG_ASSIGN_OR_RAISE(committed_files, CommittedTransactionFiles( - committed, ctx_->table->io(), - std::move(snapshot_ids), content_paths)); + ICEBERG_ASSIGN_OR_RAISE( + committed_files, + CommittedTransactionFiles(committed, ctx_->table->io(), std::move(snapshot_ids), + std::move(content_paths))); return {}; }); for (const auto& update : pending_updates_) { diff --git a/src/iceberg/update/snapshot_update.cc b/src/iceberg/update/snapshot_update.cc index f493c85dc..c2604a9c0 100644 --- a/src/iceberg/update/snapshot_update.cc +++ b/src/iceberg/update/snapshot_update.cc @@ -23,6 +23,7 @@ #include #include #include +#include #include "iceberg/constants.h" #include "iceberg/file_io.h" @@ -193,6 +194,18 @@ Result AddMetadata(const ManifestFile& manifest, std::shared_ptr manifests, + const TableMetadata& metadata) { + std::unordered_set checked_spec_ids; + for (const auto& manifest : manifests) { + if (checked_spec_ids.insert(manifest.partition_spec_id).second) { + ICEBERG_RETURN_UNEXPECTED(metadata.PartitionSpecById(manifest.partition_spec_id)); + } + } + return {}; +} + } // anonymous namespace SnapshotUpdate::~SnapshotUpdate() = default; @@ -322,6 +335,7 @@ Result SnapshotUpdate::Apply() { ICEBERG_RETURN_UNEXPECTED(Validate(base(), parent_snapshot)); ICEBERG_ASSIGN_OR_RAISE(auto manifests, Apply(base(), parent_snapshot)); + ICEBERG_RETURN_UNEXPECTED(ValidateManifestPartitionSpecs(manifests, base())); auto metadata_tasks = TaskGroup().SetExecutor(plan_executor_); for (auto& manifest : manifests) { if (manifest.added_snapshot_id.has_value()) {