Skip to content

Commit 6452715

Browse files
authored
fix: MergingSnapshotUpdate should not contain deleted file during retry (#707)
1 parent 6f50a39 commit 6452715

3 files changed

Lines changed: 50 additions & 5 deletions

File tree

src/iceberg/test/merging_snapshot_update_test.cc

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -882,6 +882,38 @@ TEST_F(MergingSnapshotUpdateTest, AddManifestCopiesManifestWithAssignedSnapshotI
882882
EXPECT_NE(data_manifests[0].manifest_path, path);
883883
}
884884

885+
TEST_F(MergingSnapshotUpdateTest, AddManifestRetryCopiesManifestAgain) {
886+
auto path = table_location_ + "/metadata/retry-input.avro";
887+
ICEBERG_UNWRAP_OR_FAIL(auto manifest, WriteManifest(path, {file_a_}));
888+
manifest.added_snapshot_id = 12345;
889+
890+
ICEBERG_UNWRAP_OR_FAIL(auto op, NewMergeAppend());
891+
EXPECT_THAT(op->AppendManifest(manifest), IsOk());
892+
893+
ICEBERG_UNWRAP_OR_FAIL(auto first_apply, static_cast<SnapshotUpdate&>(*op).Apply());
894+
SnapshotCache first_snapshot_cache(first_apply.snapshot.get());
895+
ICEBERG_UNWRAP_OR_FAIL(auto first_manifests,
896+
first_snapshot_cache.DataManifests(file_io_));
897+
ASSERT_EQ(first_manifests.size(), 1U);
898+
EXPECT_NE(first_manifests[0].manifest_path, path);
899+
900+
ICEBERG_UNWRAP_OR_FAIL(auto second_apply, static_cast<SnapshotUpdate&>(*op).Apply());
901+
SnapshotCache second_snapshot_cache(second_apply.snapshot.get());
902+
ICEBERG_UNWRAP_OR_FAIL(auto second_manifests,
903+
second_snapshot_cache.DataManifests(file_io_));
904+
ASSERT_EQ(second_manifests.size(), 1U);
905+
EXPECT_NE(second_manifests[0].manifest_path, path);
906+
EXPECT_NE(second_manifests[0].manifest_path, first_manifests[0].manifest_path);
907+
908+
std::vector<ManifestFile> second_manifest_vector(second_manifests.begin(),
909+
second_manifests.end());
910+
ICEBERG_UNWRAP_OR_FAIL(auto entries,
911+
ReadAllEntries(second_manifest_vector, *table_->metadata()));
912+
ASSERT_EQ(entries.size(), 1U);
913+
ASSERT_NE(entries[0].data_file, nullptr);
914+
EXPECT_EQ(entries[0].data_file->file_path, file_a_->file_path);
915+
}
916+
885917
TEST_F(MergingSnapshotUpdateTest, AddManifestRejectsManifestWithFirstRowId) {
886918
auto path = table_location_ + "/metadata/rowid.avro";
887919
ICEBERG_UNWRAP_OR_FAIL(auto manifest, WriteManifest(path, {file_a_}));

src/iceberg/update/merging_snapshot_update.cc

Lines changed: 14 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -649,20 +649,23 @@ Status MergingSnapshotUpdate::AddManifest(ManifestFile manifest) {
649649
appended_manifests_summary_.AddedManifest(manifest);
650650
append_manifests_.push_back(std::move(manifest));
651651
} else {
652-
ICEBERG_ASSIGN_OR_RAISE(auto copied, CopyManifest(manifest));
652+
ICEBERG_ASSIGN_OR_RAISE(auto copied, CopyManifest(manifest, /*update_summary=*/true));
653+
append_manifests_to_copy_.push_back(std::move(manifest));
653654
rewritten_append_manifests_.push_back(std::move(copied));
654655
}
655656
return {};
656657
}
657658

658-
Result<ManifestFile> MergingSnapshotUpdate::CopyManifest(const ManifestFile& manifest) {
659+
Result<ManifestFile> MergingSnapshotUpdate::CopyManifest(const ManifestFile& manifest,
660+
bool update_summary) {
659661
const TableMetadata& current = base();
660662
ICEBERG_ASSIGN_OR_RAISE(auto schema, SnapshotUtil::SchemaFor(current, target_branch()));
661663
ICEBERG_ASSIGN_OR_RAISE(auto spec,
662664
current.PartitionSpecById(manifest.partition_spec_id));
663665
std::string path = ManifestPath();
664666
return CopyAppendManifest(manifest, ctx_->table->io(), schema, spec, SnapshotId(), path,
665-
current.format_version, &appended_manifests_summary_);
667+
current.format_version,
668+
update_summary ? &appended_manifests_summary_ : nullptr);
666669
}
667670

668671
// -------------------------------------------------------------------------
@@ -890,6 +893,13 @@ Result<std::vector<ManifestFile>> MergingSnapshotUpdate::Apply(
890893

891894
// Step 4: Write (or retrieve cached) new data manifests.
892895
ICEBERG_ASSIGN_OR_RAISE(auto written_data_manifests, WriteNewDataManifests());
896+
if (rewritten_append_manifests_.empty() && !append_manifests_to_copy_.empty()) {
897+
for (const auto& manifest : append_manifests_to_copy_) {
898+
ICEBERG_ASSIGN_OR_RAISE(auto copied, CopyManifest(manifest,
899+
/*update_summary=*/false));
900+
rewritten_append_manifests_.push_back(std::move(copied));
901+
}
902+
}
893903

894904
// Incorporate append manifests (from AddManifest), stamping each with the
895905
// current snapshot ID. append_manifests_ are used directly (inherit path);
@@ -972,7 +982,7 @@ Status MergingSnapshotUpdate::CleanUncommittedAppends(
972982
DeleteUncommitted(cached_new_delete_manifests_, committed, /*clear=*/true));
973983
// rewritten_append_manifests_ are always owned by the table.
974984
ICEBERG_RETURN_UNEXPECTED(
975-
DeleteUncommitted(rewritten_append_manifests_, committed, /*clear=*/false));
985+
DeleteUncommitted(rewritten_append_manifests_, committed, /*clear=*/true));
976986

977987
// append_manifests_ are only owned by the table if the commit succeeded.
978988
if (!committed.empty()) {

src/iceberg/update/merging_snapshot_update.h

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -318,7 +318,8 @@ class ICEBERG_EXPORT MergingSnapshotUpdate : public SnapshotUpdate {
318318

319319
/// \brief Copy a manifest with the current snapshot ID, for use when snapshot
320320
/// ID inheritance is not possible.
321-
Result<ManifestFile> CopyManifest(const ManifestFile& manifest);
321+
/// \param update_summary Whether to add copied entries to the append summary
322+
Result<ManifestFile> CopyManifest(const ManifestFile& manifest, bool update_summary);
322323

323324
Status AddDeleteFile(std::shared_ptr<DataFile> file,
324325
std::optional<int64_t> data_sequence_number);
@@ -371,6 +372,8 @@ class ICEBERG_EXPORT MergingSnapshotUpdate : public SnapshotUpdate {
371372
// Manifests passed via AddManifest(): inherit path (no copy needed) and
372373
// rewrite path (must be copied with the current snapshot ID).
373374
std::vector<ManifestFile> append_manifests_;
375+
// Original manifests kept to recreate copied manifests after retry cleanup.
376+
std::vector<ManifestFile> append_manifests_to_copy_;
374377
std::vector<ManifestFile> rewritten_append_manifests_;
375378

376379
// Set to true when new files are staged after the cache was populated, so the

0 commit comments

Comments
 (0)