From 2ef558f55fd5305ec922315e71ab95b2da6d7ed8 Mon Sep 17 00:00:00 2001 From: Xinli shang Date: Mon, 7 Sep 2026 16:55:16 -0700 Subject: [PATCH] feat: add CherryPickOperation and wire SnapshotManager::Cherrypick Implements cherry-picking an append or dynamic partition overwrite onto the current state, with source-snapshot-id and WAP publish tracking, non-ancestor and replaced-partition validation. SnapshotManager routes fast-forwardable picks to SetSnapshot, since those produce no snapshot. Closes the CherryPickOperation item in #637. --- src/iceberg/CMakeLists.txt | 1 + src/iceberg/meson.build | 1 + src/iceberg/test/CMakeLists.txt | 1 + .../test/cherry_pick_operation_test.cc | 476 ++++++++++++++++++ src/iceberg/transaction.cc | 8 + src/iceberg/transaction.h | 7 + src/iceberg/type_fwd.h | 1 + src/iceberg/update/cherry_pick_operation.cc | 333 ++++++++++++ src/iceberg/update/cherry_pick_operation.h | 113 +++++ src/iceberg/update/meson.build | 1 + src/iceberg/update/set_snapshot.cc | 11 + src/iceberg/update/set_snapshot.h | 9 + src/iceberg/update/snapshot_manager.cc | 28 +- src/iceberg/update/snapshot_manager.h | 7 +- src/iceberg/util/snapshot_util.cc | 9 + src/iceberg/util/snapshot_util_internal.h | 11 + 16 files changed, 1014 insertions(+), 3 deletions(-) create mode 100644 src/iceberg/test/cherry_pick_operation_test.cc create mode 100644 src/iceberg/update/cherry_pick_operation.cc create mode 100644 src/iceberg/update/cherry_pick_operation.h diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index 8a98274ff..122877cec 100644 --- a/src/iceberg/CMakeLists.txt +++ b/src/iceberg/CMakeLists.txt @@ -119,6 +119,7 @@ set(ICEBERG_SOURCES transform.cc transform_function.cc type.cc + update/cherry_pick_operation.cc update/delete_files.cc update/expire_snapshots.cc update/fast_append.cc diff --git a/src/iceberg/meson.build b/src/iceberg/meson.build index 989f4ae03..9c80201eb 100644 --- a/src/iceberg/meson.build +++ b/src/iceberg/meson.build @@ -172,6 +172,7 @@ iceberg_sources = files( 'transform.cc', 'transform_function.cc', 'type.cc', + 'update/cherry_pick_operation.cc', 'update/delete_files.cc', 'update/expire_snapshots.cc', 'update/fast_append.cc', diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index 5ca9fd915..1af30fb26 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -227,6 +227,7 @@ if(ICEBERG_BUILD_BUNDLE) add_iceberg_test(table_update_test USE_BUNDLE SOURCES + cherry_pick_operation_test.cc delete_files_test.cc expire_snapshots_test.cc fast_append_test.cc diff --git a/src/iceberg/test/cherry_pick_operation_test.cc b/src/iceberg/test/cherry_pick_operation_test.cc new file mode 100644 index 000000000..f8d22f01f --- /dev/null +++ b/src/iceberg/test/cherry_pick_operation_test.cc @@ -0,0 +1,476 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iceberg/update/cherry_pick_operation.h" + +#include +#include +#include +#include + +#include +#include + +#include "iceberg/avro/avro_register.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/partition_spec.h" +#include "iceberg/row/partition_values.h" +#include "iceberg/schema.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/test/matchers.h" +#include "iceberg/test/update_test_base.h" +#include "iceberg/transaction.h" +#include "iceberg/update/delete_files.h" +#include "iceberg/update/fast_append.h" +#include "iceberg/update/overwrite_files.h" +#include "iceberg/update/replace_partitions.h" +#include "iceberg/update/snapshot_manager.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +// The base table (TableMetadataV2ValidMinimal.json) has schema {x: long (id 1), +// y: long (id 2), z: long (id 3)} and partitions by identity(x) as spec 0. +class CherryPickOperationTest : public UpdateTestBase { + protected: + static void SetUpTestSuite() { avro::RegisterAll(); } + + std::string MetadataResource() const override { + return "TableMetadataV2ValidMinimal.json"; + } + + void SetUp() override { + UpdateTestBase::SetUp(); + + ICEBERG_UNWRAP_OR_FAIL(spec_, table_->spec()); + ICEBERG_UNWRAP_OR_FAIL(schema_, table_->schema()); + + file_a_ = MakeDataFile("/data/file_a.parquet", /*partition_x=*/1L); + file_b_ = MakeDataFile("/data/file_b.parquet", /*partition_x=*/2L); + replacement_a_ = MakeDataFile("/data/file_a_replacement.parquet", /*partition_x=*/1L); + conflict_a_ = MakeDataFile("/data/file_a_conflict.parquet", /*partition_x=*/1L); + } + + std::shared_ptr MakeDataFile(const std::string& path, int64_t partition_x) { + auto f = std::make_shared(); + f->content = DataFile::Content::kData; + f->file_path = table_location_ + path; + f->file_format = FileFormatType::kParquet; + f->partition = PartitionValues(std::vector{Literal::Long(partition_x)}); + f->file_size_in_bytes = 1024; + f->record_count = 100; + f->partition_spec_id = spec_->spec_id(); + return f; + } + + // Live (non-deleted) data file paths across the current snapshot's manifests. + Result> LiveDataFilePaths() { + std::vector paths; + ICEBERG_ASSIGN_OR_RAISE(auto snapshot, table_->current_snapshot()); + SnapshotCache cache(snapshot.get()); + ICEBERG_ASSIGN_OR_RAISE(auto manifests, cache.DataManifests(file_io_)); + for (const auto& manifest : manifests) { + ICEBERG_ASSIGN_OR_RAISE( + auto spec, table_->metadata()->PartitionSpecById(manifest.partition_spec_id)); + ICEBERG_ASSIGN_OR_RAISE(auto reader, + ManifestReader::Make(manifest, file_io_, schema_, spec)); + ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->LiveEntries()); + for (const auto& entry : entries) { + if (entry.data_file) { + paths.push_back(entry.data_file->file_path); + } + } + } + return paths; + } + + int64_t CommitAppend(const std::shared_ptr& file) { + auto fa = table_->NewFastAppend(); + EXPECT_TRUE(fa.has_value()); + fa.value()->AppendFile(file); + EXPECT_THAT(fa.value()->Commit(), IsOk()); + EXPECT_THAT(table_->Refresh(), IsOk()); + auto snap = table_->current_snapshot(); + EXPECT_TRUE(snap.has_value()); + return snap.value()->snapshot_id; + } + + // Commit a staged append and return the staged snapshot's ID. The table's + // current snapshot is unchanged. + int64_t StageAppend(const std::shared_ptr& file, + const std::string& wap_id = "") { + auto fa = table_->NewFastAppend(); + EXPECT_TRUE(fa.has_value()); + fa.value()->StageOnly(); + if (!wap_id.empty()) { + fa.value()->Set(SnapshotSummaryFields::kWAPId, wap_id); + } + fa.value()->AppendFile(file); + EXPECT_THAT(fa.value()->Commit(), IsOk()); + EXPECT_THAT(table_->Refresh(), IsOk()); + return table_->metadata()->snapshots.back()->snapshot_id; + } + + // Commit a staged dynamic partition overwrite and return its snapshot ID. + int64_t StageReplacePartitions(const std::shared_ptr& file) { + auto ctx = TransactionContext::Make(table_, TransactionKind::kUpdate); + EXPECT_TRUE(ctx.has_value()); + auto op = ReplacePartitions::Make(TableName(), std::move(ctx.value())); + EXPECT_TRUE(op.has_value()); + op.value()->StageOnly(); + op.value()->AddFile(file); + EXPECT_THAT(op.value()->Commit(), IsOk()); + EXPECT_THAT(table_->Refresh(), IsOk()); + return table_->metadata()->snapshots.back()->snapshot_id; + } + + // Commit a staged overwrite that is not a dynamic partition overwrite, and + // return its snapshot ID. + int64_t StageOverwrite(const std::shared_ptr& added, + const std::shared_ptr& removed) { + auto ctx = TransactionContext::Make(table_, TransactionKind::kUpdate); + EXPECT_TRUE(ctx.has_value()); + auto op = OverwriteFiles::Make(TableName(), std::move(ctx.value())); + EXPECT_TRUE(op.has_value()); + op.value()->StageOnly(); + op.value()->AddFile(added); + op.value()->DeleteFile(removed); + EXPECT_THAT(op.value()->Commit(), IsOk()); + EXPECT_THAT(table_->Refresh(), IsOk()); + return table_->metadata()->snapshots.back()->snapshot_id; + } + + void RollbackTo(int64_t snapshot_id) { + ICEBERG_UNWRAP_OR_FAIL(auto manager, table_->NewSnapshotManager()); + manager->RollbackTo(snapshot_id); + EXPECT_THAT(manager->Commit(), IsOk()); + EXPECT_THAT(table_->Refresh(), IsOk()); + } + + void CommitDelete(const std::string& path) { + ICEBERG_UNWRAP_OR_FAIL(auto delete_files, table_->NewDeleteFiles()); + delete_files->DeleteFile(path); + EXPECT_THAT(delete_files->Commit(), IsOk()); + EXPECT_THAT(table_->Refresh(), IsOk()); + } + + Status Cherrypick(int64_t snapshot_id) { + ICEBERG_ASSIGN_OR_RAISE(auto manager, table_->NewSnapshotManager()); + manager->Cherrypick(snapshot_id); + ICEBERG_RETURN_UNEXPECTED(manager->Commit()); + return table_->Refresh(); + } + + std::shared_ptr spec_; + std::shared_ptr schema_; + std::shared_ptr file_a_; + std::shared_ptr file_b_; + std::shared_ptr replacement_a_; + std::shared_ptr conflict_a_; +}; + +// A staged dynamic overwrite is re-applied onto a state that moved on, so a new +// snapshot is produced rather than a fast-forward. +TEST_F(CherryPickOperationTest, CherryPickDynamicOverwrite) { + CommitAppend(file_a_); + int64_t staged_id = StageReplacePartitions(replacement_a_); + CommitAppend(file_b_); + + EXPECT_THAT(Cherrypick(staged_id), IsOk()); + + ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot()); + EXPECT_NE(snapshot->snapshot_id, staged_id); + EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kSourceSnapshotId), + std::to_string(staged_id)); + + ICEBERG_UNWRAP_OR_FAIL(auto live, LiveDataFilePaths()); + EXPECT_THAT(live, ::testing::UnorderedElementsAre(file_b_->file_path, + replacement_a_->file_path)); +} + +// The same pick works when the staged overwrite has no parent snapshot. +TEST_F(CherryPickOperationTest, CherryPickDynamicOverwriteWithoutParent) { + int64_t staged_id = StageReplacePartitions(replacement_a_); + CommitAppend(file_b_); + + EXPECT_THAT(Cherrypick(staged_id), IsOk()); + + ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot()); + EXPECT_NE(snapshot->snapshot_id, staged_id); + + ICEBERG_UNWRAP_OR_FAIL(auto live, LiveDataFilePaths()); + EXPECT_THAT(live, ::testing::UnorderedElementsAre(file_b_->file_path, + replacement_a_->file_path)); +} + +// A file added concurrently into a replaced partition blocks the pick. +TEST_F(CherryPickOperationTest, CherryPickDynamicOverwriteConflict) { + CommitAppend(file_a_); + int64_t staged_id = StageReplacePartitions(replacement_a_); + int64_t last_snapshot_id = CommitAppend(conflict_a_); + + EXPECT_THAT( + Cherrypick(staged_id), + ::testing::AllOf( + IsError(ErrorKind::kValidationFailed), + HasErrorMessage("Cannot cherry-pick replace partitions with changed partition: " + "x=1"))); + + ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot()); + EXPECT_EQ(snapshot->snapshot_id, last_snapshot_id); + ICEBERG_UNWRAP_OR_FAIL(auto live, LiveDataFilePaths()); + EXPECT_THAT( + live, ::testing::UnorderedElementsAre(file_a_->file_path, conflict_a_->file_path)); +} + +// A file the staged overwrite removed must still be present at pick time. +TEST_F(CherryPickOperationTest, CherryPickDynamicOverwriteDeleteConflict) { + CommitAppend(file_a_); + int64_t staged_id = StageReplacePartitions(replacement_a_); + CommitAppend(file_b_); + CommitDelete(file_a_->file_path); + ICEBERG_UNWRAP_OR_FAIL(auto before, table_->current_snapshot()); + int64_t last_snapshot_id = before->snapshot_id; + + EXPECT_THAT(Cherrypick(staged_id), IsError(ErrorKind::kValidationFailed)); + + EXPECT_THAT(table_->Refresh(), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot()); + EXPECT_EQ(snapshot->snapshot_id, last_snapshot_id); + ICEBERG_UNWRAP_OR_FAIL(auto live, LiveDataFilePaths()); + EXPECT_THAT(live, ::testing::UnorderedElementsAre(file_b_->file_path)); +} + +// A staged append is re-applied on top of the current state. +TEST_F(CherryPickOperationTest, CherryPickAppend) { + CommitAppend(file_a_); + int64_t staged_id = StageAppend(replacement_a_); + CommitAppend(file_b_); + + EXPECT_THAT(Cherrypick(staged_id), IsOk()); + + ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot()); + EXPECT_NE(snapshot->snapshot_id, staged_id); + EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kSourceSnapshotId), + std::to_string(staged_id)); + EXPECT_FALSE(snapshot->summary.contains(SnapshotSummaryFields::kPublishedWAPId)); + + ICEBERG_UNWRAP_OR_FAIL(auto live, LiveDataFilePaths()); + EXPECT_THAT(live, + ::testing::UnorderedElementsAre(file_a_->file_path, file_b_->file_path, + replacement_a_->file_path)); +} + +// When the picked snapshot's parent is the current snapshot, the pick moves the +// current snapshot instead of creating one. +TEST_F(CherryPickOperationTest, FastForwardSetsCurrentSnapshot) { + CommitAppend(file_a_); + int64_t staged_id = StageAppend(file_b_); + + EXPECT_THAT(Cherrypick(staged_id), IsOk()); + + ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot()); + EXPECT_EQ(snapshot->snapshot_id, staged_id); + ICEBERG_UNWRAP_OR_FAIL(auto live, LiveDataFilePaths()); + EXPECT_THAT(live, + ::testing::UnorderedElementsAre(file_a_->file_path, file_b_->file_path)); +} + +// An overwrite without "replace-partitions" is not pickable, but it is still +// fast-forwarded to when it is a child of the current snapshot. +TEST_F(CherryPickOperationTest, FastForwardOverwriteSetsCurrentSnapshot) { + CommitAppend(file_a_); + int64_t staged_id = StageOverwrite(/*added=*/replacement_a_, /*removed=*/file_a_); + + EXPECT_THAT(Cherrypick(staged_id), IsOk()); + + ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot()); + EXPECT_EQ(snapshot->snapshot_id, staged_id); + EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kOperation), + DataOperation::kOverwrite); + // A fast-forward publishes the staged snapshot itself, so it carries no + // source-snapshot-id. + EXPECT_FALSE(snapshot->summary.contains(SnapshotSummaryFields::kSourceSnapshotId)); + + ICEBERG_UNWRAP_OR_FAIL(auto live, LiveDataFilePaths()); + EXPECT_THAT(live, ::testing::UnorderedElementsAre(replacement_a_->file_path)); +} + +// A dynamic overwrite whose parent has been rolled off the current history +// cannot be picked, because the partitions it replaced cannot be checked. +TEST_F(CherryPickOperationTest, CherryPickDynamicOverwriteParentNotAncestor) { + int64_t first_id = CommitAppend(file_a_); + CommitAppend(file_b_); + int64_t staged_id = StageReplacePartitions(replacement_a_); + RollbackTo(first_id); + + EXPECT_THAT(Cherrypick(staged_id), + ::testing::AllOf( + IsError(ErrorKind::kValidationFailed), + HasErrorMessage(std::format("Cannot cherry-pick overwrite not based on " + "an ancestor of the current state: {}", + staged_id)))); + + ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot()); + EXPECT_EQ(snapshot->snapshot_id, first_id); +} + +// A WAP id already published by an ancestor is rejected even when the staged +// snapshot is a child of the current one and would otherwise fast-forward. +TEST_F(CherryPickOperationTest, DuplicateWapPublishOnFastForwardRejected) { + CommitAppend(file_a_); + int64_t first_staged_id = StageAppend(file_b_, /*wap_id=*/"wap-123"); + + EXPECT_THAT(Cherrypick(first_staged_id), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto picked, table_->current_snapshot()); + EXPECT_EQ(picked->snapshot_id, first_staged_id); + + // Staged on top of the snapshot just published, so this is a fast-forward. + int64_t second_staged_id = StageAppend(conflict_a_, /*wap_id=*/"wap-123"); + + EXPECT_THAT( + Cherrypick(second_staged_id), + ::testing::AllOf(IsError(ErrorKind::kValidationFailed), + HasErrorMessage("Duplicate request to cherry pick wap id that " + "was published already: wap-123"))); + + EXPECT_THAT(table_->Refresh(), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot()); + EXPECT_EQ(snapshot->snapshot_id, first_staged_id); +} + +// A fast-forward staged against one state must not be applied to another. The +// commit below is retried after a concurrent append, and re-applying the +// fast-forward there would discard that append. +TEST_F(CherryPickOperationTest, FastForwardInvalidatedByConcurrentCommitRejected) { + CommitAppend(file_a_); + int64_t staged_id = StageAppend(file_b_); + + // Stage the fast-forward, but hold the transaction open. + ICEBERG_UNWRAP_OR_FAIL(auto txn, Transaction::Make(table_, TransactionKind::kUpdate)); + ICEBERG_UNWRAP_OR_FAIL(auto manager, SnapshotManager::Make(txn)); + manager->Cherrypick(staged_id); + EXPECT_THAT(manager->Commit(), IsOk()); + + // A separate table handle advances the current snapshot in the catalog, so + // the staged snapshot's parent is no longer current. + ICEBERG_UNWRAP_OR_FAIL(auto other_table, catalog_->LoadTable(table_ident_)); + ICEBERG_UNWRAP_OR_FAIL(auto append, other_table->NewFastAppend()); + append->AppendFile(conflict_a_); + ASSERT_THAT(append->Commit(), IsOk()); + + EXPECT_THAT(txn->Commit(), + HasErrorMessage(std::format( + "Cannot fast-forward to {}: not a child of the current table state", + staged_id))); + + // The concurrent append survives; the staged snapshot was not published. + EXPECT_THAT(table_->Refresh(), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot()); + EXPECT_NE(snapshot->snapshot_id, staged_id); + ICEBERG_UNWRAP_OR_FAIL(auto live, LiveDataFilePaths()); + EXPECT_THAT( + live, ::testing::UnorderedElementsAre(file_a_->file_path, conflict_a_->file_path)); +} + +// A snapshot already in the current history cannot be picked again. +TEST_F(CherryPickOperationTest, CherryPickAncestorRejected) { + int64_t first_id = CommitAppend(file_a_); + CommitAppend(file_b_); + + EXPECT_THAT(Cherrypick(first_id), + ::testing::AllOf( + IsError(ErrorKind::kValidationFailed), + HasErrorMessage(std::format( + "Cannot cherrypick snapshot {}: already an ancestor", first_id)))); + + ICEBERG_UNWRAP_OR_FAIL(auto live, LiveDataFilePaths()); + EXPECT_THAT(live, + ::testing::UnorderedElementsAre(file_a_->file_path, file_b_->file_path)); +} + +// The same WAP id cannot be picked twice. +TEST_F(CherryPickOperationTest, DuplicateWapPublishRejected) { + CommitAppend(file_a_); + int64_t first_staged_id = StageAppend(file_b_, /*wap_id=*/"wap-123"); + int64_t second_staged_id = StageAppend(conflict_a_, /*wap_id=*/"wap-123"); + + EXPECT_THAT(Cherrypick(first_staged_id), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto picked, table_->current_snapshot()); + EXPECT_EQ(picked->snapshot_id, first_staged_id); + + EXPECT_THAT( + Cherrypick(second_staged_id), + ::testing::AllOf(IsError(ErrorKind::kValidationFailed), + HasErrorMessage("Duplicate request to cherry pick wap id that " + "was published already: wap-123"))); +} + +// A picked snapshot with no WAP id records only the source snapshot. +TEST_F(CherryPickOperationTest, NonWapCherrypick) { + CommitAppend(file_a_); + int64_t staged_id = StageAppend(replacement_a_); + CommitAppend(file_b_); + + EXPECT_THAT(Cherrypick(staged_id), IsOk()); + + ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot()); + EXPECT_FALSE(snapshot->summary.contains(SnapshotSummaryFields::kPublishedWAPId)); + EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kSourceSnapshotId), + std::to_string(staged_id)); +} + +// A snapshot that is neither an append nor a dynamic overwrite can only be +// fast-forwarded. +TEST_F(CherryPickOperationTest, NonPickableOperationRejected) { + CommitAppend(file_a_); + CommitAppend(file_b_); + + auto delete_files = table_->NewDeleteFiles(); + ASSERT_TRUE(delete_files.has_value()); + delete_files.value()->StageOnly(); + delete_files.value()->DeleteFile(file_a_->file_path); + ASSERT_THAT(delete_files.value()->Commit(), IsOk()); + ASSERT_THAT(table_->Refresh(), IsOk()); + int64_t staged_id = table_->metadata()->snapshots.back()->snapshot_id; + + CommitAppend(conflict_a_); + + EXPECT_THAT(Cherrypick(staged_id), + ::testing::AllOf( + IsError(ErrorKind::kValidationFailed), + HasErrorMessage(std::format( + "Cannot cherry-pick snapshot {}: not append, dynamic overwrite, " + "or fast-forward", + staged_id)))); +} + +TEST_F(CherryPickOperationTest, UnknownSnapshotRejected) { + CommitAppend(file_a_); + + EXPECT_THAT( + Cherrypick(/*snapshot_id=*/-99), + ::testing::AllOf(IsError(ErrorKind::kValidationFailed), + HasErrorMessage("Cannot cherry-pick unknown snapshot ID: -99"))); +} + +} // namespace iceberg diff --git a/src/iceberg/transaction.cc b/src/iceberg/transaction.cc index 80d39c8a8..2eb937f53 100644 --- a/src/iceberg/transaction.cc +++ b/src/iceberg/transaction.cc @@ -32,6 +32,7 @@ #include "iceberg/table_requirement.h" #include "iceberg/table_requirements.h" #include "iceberg/table_update.h" +#include "iceberg/update/cherry_pick_operation.h" #include "iceberg/update/delete_files.h" #include "iceberg/update/expire_snapshots.h" #include "iceberg/update/fast_append.h" @@ -485,6 +486,13 @@ Result> Transaction::NewUpdateLocation() { return update_location; } +Result> Transaction::NewCherryPickOperation() { + ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr cherry_pick, + CherryPickOperation::Make(ctx_->table->name().name, ctx_)); + ICEBERG_RETURN_UNEXPECTED(AddUpdate(cherry_pick)); + return cherry_pick; +} + Result> Transaction::NewSetSnapshot() { ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr set_snapshot, SetSnapshot::Make(ctx_)); diff --git a/src/iceberg/transaction.h b/src/iceberg/transaction.h index 007b1057e..1ea3d1b03 100644 --- a/src/iceberg/transaction.h +++ b/src/iceberg/transaction.h @@ -131,6 +131,13 @@ class ICEBERG_EXPORT Transaction : public std::enable_shared_from_this> NewSnapshotManager(); + /// \brief Create a new CherryPickOperation to apply the changes of an existing + /// snapshot onto the current state. + /// + /// \note Intended for use by SnapshotManager. Prefer + /// SnapshotManager::Cherrypick(), which also handles the fast-forward case. + Result> NewCherryPickOperation(); + /// \brief Create a new SetSnapshot to set the current snapshot or rollback to a /// previous snapshot and commit the changes. Result> NewSetSnapshot(); diff --git a/src/iceberg/type_fwd.h b/src/iceberg/type_fwd.h index 0b19adaf5..453d0512c 100644 --- a/src/iceberg/type_fwd.h +++ b/src/iceberg/type_fwd.h @@ -250,6 +250,7 @@ class Transaction; class TransactionContext; /// \brief Update family. +class CherryPickOperation; class DeleteFiles; class ExpireSnapshots; class FastAppend; diff --git a/src/iceberg/update/cherry_pick_operation.cc b/src/iceberg/update/cherry_pick_operation.cc new file mode 100644 index 000000000..1042e0226 --- /dev/null +++ b/src/iceberg/update/cherry_pick_operation.cc @@ -0,0 +1,333 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iceberg/update/cherry_pick_operation.h" + +#include +#include + +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_list.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/schema.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/transaction.h" +#include "iceberg/util/error_collector.h" +#include "iceberg/util/macros.h" +#include "iceberg/util/snapshot_util_internal.h" + +namespace iceberg { + +namespace { + +Result> MakeManifestReader( + const ManifestFile& manifest, const std::shared_ptr& file_io, + const TableMetadata& metadata) { + ICEBERG_ASSIGN_OR_RAISE(auto schema, metadata.Schema()); + TableMetadataCache metadata_cache(&metadata); + ICEBERG_ASSIGN_OR_RAISE(auto specs_by_id, metadata_cache.GetPartitionSpecsById()); + return ManifestReader::Make(manifest, file_io, std::move(schema), specs_by_id.get()); +} + +/// \brief Data files added and removed by a snapshot. +struct SnapshotChanges { + std::vector> added; + std::vector> removed; +}; + +/// \brief Read the data files that the given snapshot added and removed. +/// +/// Only manifests written by the snapshot carry its own entries, so manifests +/// inherited from earlier snapshots are skipped. +Result ReadSnapshotChanges(const Snapshot& snapshot, + const std::shared_ptr& file_io, + const TableMetadata& metadata) { + SnapshotChanges changes; + SnapshotCache cache(&snapshot); + ICEBERG_ASSIGN_OR_RAISE(auto manifests, cache.DataManifests(file_io)); + for (const auto& manifest : manifests) { + if (manifest.added_snapshot_id != snapshot.snapshot_id) { + continue; + } + ICEBERG_ASSIGN_OR_RAISE(auto reader, MakeManifestReader(manifest, file_io, metadata)); + ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->Entries()); + for (const auto& entry : entries) { + if (!entry.data_file) { + continue; + } + if (entry.status == ManifestStatus::kAdded) { + changes.added.push_back(entry.data_file); + } else if (entry.status == ManifestStatus::kDeleted) { + changes.removed.push_back(entry.data_file); + } + } + } + return changes; +} + +std::string StagedWapId(const Snapshot& snapshot) { + auto it = snapshot.summary.find(SnapshotSummaryFields::kWAPId); + return it == snapshot.summary.end() ? std::string() : it->second; +} + +std::string PublishedWapId(const Snapshot& snapshot) { + auto it = snapshot.summary.find(SnapshotSummaryFields::kPublishedWAPId); + return it == snapshot.summary.end() ? std::string() : it->second; +} + +Result> CurrentAncestorIds(const TableMetadata& metadata) { + std::vector ids; + if (metadata.current_snapshot_id == kInvalidSnapshotId) { + return ids; + } + ICEBERG_ASSIGN_OR_RAISE( + auto ancestors, SnapshotUtil::AncestorsOf(metadata, metadata.current_snapshot_id)); + ids.reserve(ancestors.size()); + for (const auto& ancestor : ancestors) { + ids.push_back(ancestor->snapshot_id); + } + return ids; +} + +/// \brief Fail if the WAP id staged on the picked snapshot was already +/// published, and return that id when the snapshot has one. +Result ValidateWapPublish(const TableMetadata& metadata, + int64_t wap_snapshot_id) { + ICEBERG_ASSIGN_OR_RAISE(auto snapshot, metadata.SnapshotById(wap_snapshot_id)); + std::string wap_id = StagedWapId(*snapshot); + if (wap_id.empty()) { + return wap_id; + } + + ICEBERG_ASSIGN_OR_RAISE(auto ancestor_ids, CurrentAncestorIds(metadata)); + for (int64_t ancestor_id : ancestor_ids) { + ICEBERG_ASSIGN_OR_RAISE(auto ancestor, metadata.SnapshotById(ancestor_id)); + if (wap_id == StagedWapId(*ancestor) || wap_id == PublishedWapId(*ancestor)) { + return CommitFailed( + "Duplicate request to cherry pick wap id that was published already: {}", + wap_id); + } + } + return wap_id; +} + +bool IsReplacePartitions(const Snapshot& snapshot) { + auto it = snapshot.summary.find(SnapshotSummaryFields::kReplacePartitions); + return it != snapshot.summary.end() && it->second == "true"; +} + +} // namespace + +Result> CherryPickOperation::Make( + std::string table_name, std::shared_ptr ctx) { + ICEBERG_PRECHECK(!table_name.empty(), "Table name cannot be empty"); + ICEBERG_PRECHECK(ctx != nullptr, "Cannot create CherryPickOperation without a context"); + return std::unique_ptr( + new CherryPickOperation(std::move(table_name), std::move(ctx))); +} + +CherryPickOperation::CherryPickOperation(std::string table_name, + std::shared_ptr ctx) + : MergingSnapshotUpdate(std::move(table_name), std::move(ctx)) {} + +std::string CherryPickOperation::operation() { + if (cherrypick_snapshot_ == nullptr) { + return DataOperation::kAppend; + } + auto op = cherrypick_snapshot_->Operation(); + return op.has_value() ? std::string(*op) : DataOperation::kAppend; +} + +Status CherryPickOperation::ValidateFastForward(const TableMetadata& metadata, + const Snapshot& snapshot) { + // Java runs the WAP check only for the two pickable operations; any other + // snapshot reaches a fast-forward without one. + const auto operation = snapshot.Operation(); + const bool is_pickable = + operation == DataOperation::kAppend || + (operation == DataOperation::kOverwrite && IsReplacePartitions(snapshot)); + if (!is_pickable) { + return {}; + } + + ICEBERG_ASSIGN_OR_RAISE(std::ignore, + ValidateWapPublish(metadata, snapshot.snapshot_id)); + return {}; +} + +bool CherryPickOperation::IsFastForward() const { + return cherrypick_snapshot_ != nullptr && + SnapshotUtil::CanFastForward(ctx_->current(), *cherrypick_snapshot_); +} + +CherryPickOperation& CherryPickOperation::Cherrypick(int64_t snapshot_id) { + const TableMetadata& metadata = ctx_->current(); + ICEBERG_BUILDER_ASSIGN_OR_RETURN_WITH_ERROR( + cherrypick_snapshot_, metadata.SnapshotById(snapshot_id), + "Cannot cherry-pick unknown snapshot ID: {}", snapshot_id); + + const auto picked_operation = cherrypick_snapshot_->Operation(); + const bool is_append = picked_operation == DataOperation::kAppend; + const bool is_dynamic_overwrite = picked_operation == DataOperation::kOverwrite && + IsReplacePartitions(*cherrypick_snapshot_); + + if (!is_append && !is_dynamic_overwrite) { + ICEBERG_BUILDER_CHECK( + IsFastForward(), + "Cannot cherry-pick snapshot {}: not append, dynamic overwrite, or fast-forward", + snapshot_id); + return *this; + } + + if (is_dynamic_overwrite) { + // The replaced partitions can only be checked against files added since the + // picked snapshot's parent, so that parent must still be in the history. + if (cherrypick_snapshot_->parent_snapshot_id.has_value()) { + ICEBERG_BUILDER_ASSIGN_OR_RETURN( + bool is_ancestor, + SnapshotUtil::IsAncestorOf(metadata, + cherrypick_snapshot_->parent_snapshot_id.value())); + ICEBERG_BUILDER_CHECK(is_ancestor, + "Cannot cherry-pick overwrite not based on an ancestor of " + "the current state: {}", + snapshot_id); + } + } + + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto wap_id, + ValidateWapPublish(metadata, snapshot_id)); + if (!wap_id.empty()) { + Set(SnapshotSummaryFields::kPublishedWAPId, wap_id); + } + Set(SnapshotSummaryFields::kSourceSnapshotId, std::to_string(snapshot_id)); + + auto io = ctx_->table->io(); + ICEBERG_BUILDER_ASSIGN_OR_RETURN( + auto changes, ReadSnapshotChanges(*cherrypick_snapshot_, io, metadata)); + + if (is_dynamic_overwrite) { + // A replace can only be re-applied if the files it removed are all present. + FailMissingDeletePaths(); + replaced_partitions_.emplace(); + } + + for (const auto& added : changes.added) { + ICEBERG_BUILDER_RETURN_IF_ERROR(AddDataFile(added)); + if (replaced_partitions_.has_value()) { + ICEBERG_BUILDER_CHECK(added->partition_spec_id.has_value(), + "Data file must have partition spec ID"); + replaced_partitions_->add(added->partition_spec_id.value(), added->partition); + } + } + + if (is_dynamic_overwrite) { + for (const auto& removed : changes.removed) { + ICEBERG_BUILDER_RETURN_IF_ERROR(DeleteDataFile(removed)); + } + } + + return *this; +} + +Status CherryPickOperation::ValidateNonAncestor(const TableMetadata& metadata, + int64_t snapshot_id) const { + ICEBERG_ASSIGN_OR_RAISE(bool is_ancestor, + SnapshotUtil::IsAncestorOf(metadata, snapshot_id)); + if (is_ancestor) { + return CommitFailed("Cannot cherrypick snapshot {}: already an ancestor", + snapshot_id); + } + + const std::string snapshot_id_str = std::to_string(snapshot_id); + ICEBERG_ASSIGN_OR_RAISE(auto ancestor_ids, CurrentAncestorIds(metadata)); + for (int64_t ancestor_id : ancestor_ids) { + ICEBERG_ASSIGN_OR_RAISE(auto ancestor, metadata.SnapshotById(ancestor_id)); + auto it = ancestor->summary.find(SnapshotSummaryFields::kSourceSnapshotId); + if (it != ancestor->summary.end() && it->second == snapshot_id_str) { + return CommitFailed( + "Cannot cherrypick snapshot {}: already picked to create ancestor {}", + snapshot_id, ancestor_id); + } + } + return {}; +} + +Status CherryPickOperation::ValidateReplacedPartitions( + const TableMetadata& metadata) const { + if (!replaced_partitions_.has_value() || + metadata.current_snapshot_id == kInvalidSnapshotId) { + return {}; + } + + const auto parent_id = cherrypick_snapshot_->parent_snapshot_id; + if (parent_id.has_value()) { + ICEBERG_ASSIGN_OR_RAISE(bool is_ancestor, + SnapshotUtil::IsAncestorOf(metadata, parent_id.value())); + if (!is_ancestor) { + return ValidationFailed( + "Cannot cherry-pick overwrite, based on non-ancestor of the current state: {}", + parent_id.value()); + } + } + + // Walk back from the current snapshot to the picked snapshot's parent and + // reject any file added into a partition this pick replaces. + auto io = ctx_->table->io(); + ICEBERG_ASSIGN_OR_RAISE( + auto ancestors, SnapshotUtil::AncestorsOf(metadata, metadata.current_snapshot_id)); + for (const auto& ancestor : ancestors) { + if (parent_id.has_value() && ancestor->snapshot_id == parent_id.value()) { + break; + } + ICEBERG_ASSIGN_OR_RAISE(auto changes, ReadSnapshotChanges(*ancestor, io, metadata)); + for (const auto& added : changes.added) { + if (!added->partition_spec_id.has_value() || + !replaced_partitions_->contains(added->partition_spec_id.value(), + added->partition)) { + continue; + } + ICEBERG_ASSIGN_OR_RAISE(auto spec, + metadata.PartitionSpecById(*added->partition_spec_id)); + ICEBERG_ASSIGN_OR_RAISE(auto partition_path, spec->PartitionPath(added->partition)); + return ValidationFailed( + "Cannot cherry-pick replace partitions with changed partition: {}", + partition_path); + } + } + return {}; +} + +Status CherryPickOperation::Validate(const TableMetadata& current_metadata, + const std::shared_ptr& snapshot) { + if (cherrypick_snapshot_ == nullptr || IsFastForward()) { + return {}; + } + + ICEBERG_RETURN_UNEXPECTED( + ValidateNonAncestor(current_metadata, cherrypick_snapshot_->snapshot_id)); + ICEBERG_RETURN_UNEXPECTED(ValidateReplacedPartitions(current_metadata)); + ICEBERG_ASSIGN_OR_RAISE( + std::ignore, + ValidateWapPublish(current_metadata, cherrypick_snapshot_->snapshot_id)); + return {}; +} + +} // namespace iceberg diff --git a/src/iceberg/update/cherry_pick_operation.h b/src/iceberg/update/cherry_pick_operation.h new file mode 100644 index 000000000..9ddaabff8 --- /dev/null +++ b/src/iceberg/update/cherry_pick_operation.h @@ -0,0 +1,113 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#pragma once + +/// \file iceberg/update/cherry_pick_operation.h + +#include +#include +#include +#include + +#include "iceberg/iceberg_export.h" +#include "iceberg/result.h" +#include "iceberg/type_fwd.h" +#include "iceberg/update/merging_snapshot_update.h" +#include "iceberg/util/partition_value_util.h" + +namespace iceberg { + +/// \brief Cherry-picks the changes of a snapshot onto the current state. +/// +/// This update is not exposed through the Table API. It is part of the +/// Transaction API intended for use in SnapshotManager. +/// +/// Three kinds of snapshot can be picked. An append snapshot has its added +/// data files re-applied on top of the current state. An overwrite snapshot +/// carrying "replace-partitions"="true" has both its added and its removed +/// data files re-applied, and can only be picked while the partitions it +/// replaced are unchanged. Any other snapshot can only be fast-forwarded. +/// +/// A fast-forward moves the current state to the picked snapshot without +/// producing a new one, which this operation cannot express because Apply() +/// always builds a new snapshot. Callers detect that case with +/// SnapshotUtil::CanFastForward() and set the current snapshot instead of +/// committing this operation; ValidateFastForward() applies the checks this +/// operation would otherwise have run. +/// +/// The new snapshot records the picked snapshot in "source-snapshot-id". When +/// the picked snapshot carries a "wap.id", that id is recorded in +/// "published-wap-id" and the pick fails if the id was already published. +class ICEBERG_EXPORT CherryPickOperation : public MergingSnapshotUpdate { + public: + /// \brief Create a new CherryPickOperation instance. + /// + /// \param table_name The name of the table + /// \param ctx The transaction context + /// \return A new CherryPickOperation instance + static Result> Make( + std::string table_name, std::shared_ptr ctx); + + /// \brief Apply the changes of the given snapshot to the current state. + /// + /// \param snapshot_id The ID of the snapshot whose changes to apply + /// \return Reference to this for method chaining + CherryPickOperation& Cherrypick(int64_t snapshot_id); + + /// \brief Run the checks that apply when the given snapshot is fast-forwarded + /// to rather than picked. + /// + /// A fast-forward publishes the picked snapshot itself, so the WAP id staged + /// on it must not already have been published. Mirrors Java, where this check + /// runs in cherrypick() for append and dynamic overwrite snapshots, before + /// apply() elects to fast-forward. + /// + /// \param metadata The table metadata to fast-forward + /// \param snapshot The snapshot to fast-forward to + static Status ValidateFastForward(const TableMetadata& metadata, + const Snapshot& snapshot); + + std::string operation() override; + + protected: + Status Validate(const TableMetadata& current_metadata, + const std::shared_ptr& snapshot) override; + + private: + explicit CherryPickOperation(std::string table_name, + std::shared_ptr ctx); + + /// \brief Whether the snapshot passed to Cherrypick() can be fast-forwarded. + bool IsFastForward() const; + + /// \brief Fail if the picked snapshot is already part of the current history, + /// either directly or as the source of an earlier pick. + Status ValidateNonAncestor(const TableMetadata& metadata, int64_t snapshot_id) const; + + /// \brief Fail if any partition replaced by the picked snapshot received new + /// files after the picked snapshot's parent. + Status ValidateReplacedPartitions(const TableMetadata& metadata) const; + + std::shared_ptr cherrypick_snapshot_; + // Set only when the picked snapshot is a dynamic partition overwrite. + std::optional replaced_partitions_; +}; + +} // namespace iceberg diff --git a/src/iceberg/update/meson.build b/src/iceberg/update/meson.build index 6b2cf6f33..05b03d869 100644 --- a/src/iceberg/update/meson.build +++ b/src/iceberg/update/meson.build @@ -17,6 +17,7 @@ install_headers( [ + 'cherry_pick_operation.h', 'delete_files.h', 'expire_snapshots.h', 'fast_append.h', diff --git a/src/iceberg/update/set_snapshot.cc b/src/iceberg/update/set_snapshot.cc index 79662890b..e2c18d3a0 100644 --- a/src/iceberg/update/set_snapshot.cc +++ b/src/iceberg/update/set_snapshot.cc @@ -85,6 +85,11 @@ SetSnapshot& SetSnapshot::RollbackTo(int64_t snapshot_id) { return SetCurrentSnapshot(snapshot_id); } +SetSnapshot& SetSnapshot::RequireFastForward() { + require_fast_forward_ = true; + return *this; +} + Result SetSnapshot::Apply() { ICEBERG_RETURN_UNEXPECTED(CheckErrors()); @@ -102,6 +107,12 @@ Result SetSnapshot::Apply() { "Cannot roll back to unknown snapshot id: {}", target_snapshot_id_.value()); + if (require_fast_forward_) { + ICEBERG_CHECK(SnapshotUtil::CanFastForward(base_metadata, *snapshot_result.value()), + "Cannot fast-forward to {}: not a child of the current table state", + target_snapshot_id_.value()); + } + // If this is a rollback, validate that the target is still an ancestor if (is_rollback_) { ICEBERG_ASSIGN_OR_RAISE( diff --git a/src/iceberg/update/set_snapshot.h b/src/iceberg/update/set_snapshot.h index 431e636b2..b48abcf4c 100644 --- a/src/iceberg/update/set_snapshot.h +++ b/src/iceberg/update/set_snapshot.h @@ -50,6 +50,14 @@ class ICEBERG_EXPORT SetSnapshot : public PendingUpdate { /// \brief Rollback table's state to a specific Snapshot identified by id. SetSnapshot& RollbackTo(int64_t snapshot_id); + /// \brief Require the target snapshot to still be a fast-forward when applied. + /// + /// Without this, a commit retried against refreshed metadata would move the + /// current snapshot even if a concurrent commit has since advanced it, + /// discarding that commit. Used by SnapshotManager::Cherrypick(), which + /// expresses a cherry-pick fast-forward as a set of the current snapshot. + SetSnapshot& RequireFastForward(); + Kind kind() const final { return Kind::kSetSnapshot; } bool IsRetryable() const override { return true; } @@ -67,6 +75,7 @@ class ICEBERG_EXPORT SetSnapshot : public PendingUpdate { std::optional target_snapshot_id_; bool is_rollback_{false}; + bool require_fast_forward_{false}; }; } // namespace iceberg diff --git a/src/iceberg/update/snapshot_manager.cc b/src/iceberg/update/snapshot_manager.cc index 5473f3033..a62067526 100644 --- a/src/iceberg/update/snapshot_manager.cc +++ b/src/iceberg/update/snapshot_manager.cc @@ -23,10 +23,12 @@ #include "iceberg/table.h" #include "iceberg/table_metadata.h" #include "iceberg/transaction.h" +#include "iceberg/update/cherry_pick_operation.h" #include "iceberg/update/fast_append.h" #include "iceberg/update/set_snapshot.h" #include "iceberg/update/update_snapshot_reference.h" #include "iceberg/util/macros.h" +#include "iceberg/util/snapshot_util_internal.h" namespace iceberg { @@ -55,8 +57,30 @@ SnapshotManager::~SnapshotManager() = default; SnapshotManager& SnapshotManager::Cherrypick(int64_t snapshot_id) { ICEBERG_BUILDER_RETURN_IF_ERROR(CommitIfRefUpdatesExist()); - // TODO(anyone): Implement cherrypick operation - ICEBERG_BUILDER_CHECK(false, "Cherrypick operation not yet implemented"); + const TableMetadata& metadata = transaction_->current(); + ICEBERG_BUILDER_ASSIGN_OR_RETURN_WITH_ERROR( + auto snapshot, metadata.SnapshotById(snapshot_id), + "Cannot cherry-pick unknown snapshot ID: {}", snapshot_id); + + // A fast-forward produces no new snapshot, so move the current snapshot + // instead of creating a cherry-pick operation. + if (SnapshotUtil::CanFastForward(metadata, *snapshot)) { + ICEBERG_BUILDER_RETURN_IF_ERROR( + CherryPickOperation::ValidateFastForward(metadata, *snapshot)); + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto set_snapshot, transaction_->NewSetSnapshot()); + // The commit may be retried against refreshed metadata, where a concurrent + // commit can have moved the current snapshot. Without this the retry would + // apply as a plain branch move and discard that commit. + set_snapshot->RequireFastForward(); + set_snapshot->SetCurrentSnapshot(snapshot_id); + ICEBERG_BUILDER_RETURN_IF_ERROR(set_snapshot->Commit()); + return *this; + } + + ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto cherry_pick, + transaction_->NewCherryPickOperation()); + cherry_pick->Cherrypick(snapshot_id); + ICEBERG_BUILDER_RETURN_IF_ERROR(cherry_pick->Commit()); return *this; } diff --git a/src/iceberg/update/snapshot_manager.h b/src/iceberg/update/snapshot_manager.h index fd81f8339..3dfdb25f3 100644 --- a/src/iceberg/update/snapshot_manager.h +++ b/src/iceberg/update/snapshot_manager.h @@ -46,7 +46,9 @@ namespace iceberg { /// instead of the one that was current when the audited changes were created. This class /// adds support for cherry-picking the changes from an orphan snapshot by applying them /// to the current snapshot. The output of the operation is a new snapshot with the -/// changes from cherry-picked snapshot. +/// changes from cherry-picked snapshot, except when the picked snapshot is a child of +/// the current one, in which case the current state fast-forwards to it and no snapshot +/// is created. class ICEBERG_EXPORT SnapshotManager : public ErrorCollector { public: /// \brief Create a SnapshotManager that owns its own transaction. @@ -63,6 +65,9 @@ class ICEBERG_EXPORT SnapshotManager : public ErrorCollector { /// \brief Apply supported changes in given snapshot and create a new snapshot which /// will be set as the current snapshot on commit. /// + /// If the given snapshot is a child of the current snapshot, the current state + /// fast-forwards to it on commit and no new snapshot is created. + /// /// \param snapshot_id a Snapshot ID whose changes to apply /// \return Reference to this for method chaining SnapshotManager& Cherrypick(int64_t snapshot_id); diff --git a/src/iceberg/util/snapshot_util.cc b/src/iceberg/util/snapshot_util.cc index 49019408b..642513279 100644 --- a/src/iceberg/util/snapshot_util.cc +++ b/src/iceberg/util/snapshot_util.cc @@ -70,6 +70,15 @@ Result SnapshotUtil::IsAncestorOf(const TableMetadata& metadata, return IsAncestorOf(metadata, current->snapshot_id, ancestor_snapshot_id); } +bool SnapshotUtil::CanFastForward(const TableMetadata& metadata, + const Snapshot& snapshot) { + if (metadata.current_snapshot_id == kInvalidSnapshotId) { + return !snapshot.parent_snapshot_id.has_value(); + } + return snapshot.parent_snapshot_id.has_value() && + snapshot.parent_snapshot_id.value() == metadata.current_snapshot_id; +} + Result SnapshotUtil::IsAncestorOf(const Table& table, int64_t snapshot_id, int64_t ancestor_snapshot_id) { return IsAncestorOf(*table.metadata(), snapshot_id, ancestor_snapshot_id); diff --git a/src/iceberg/util/snapshot_util_internal.h b/src/iceberg/util/snapshot_util_internal.h index 8a3158185..f45e2b68c 100644 --- a/src/iceberg/util/snapshot_util_internal.h +++ b/src/iceberg/util/snapshot_util_internal.h @@ -78,6 +78,17 @@ class ICEBERG_EXPORT SnapshotUtil { static Result IsAncestorOf(const TableMetadata& metadata, int64_t ancestor_snapshot_id); + /// \brief Returns whether the given snapshot can be fast-forwarded onto the metadata's + /// current state, that is, its parent is the current snapshot or both are absent. + /// + /// A fast-forward moves the current state to an existing snapshot rather than creating + /// a new one. + /// + /// \param metadata The table metadata to fast-forward + /// \param snapshot The snapshot to fast-forward to + /// \return true if snapshot can be fast-forwarded onto the current state + static bool CanFastForward(const TableMetadata& metadata, const Snapshot& snapshot); + /// \brief Returns whether ancestor_snapshot_id is an ancestor of snapshot_id. /// /// \param table The table to check