This is an automated email from the ASF dual-hosted git repository.
wgtmac pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 44e29e10 feat: add CherryPickOperation and wire
SnapshotManager::Cherrypick (#926)
44e29e10 is described below
commit 44e29e109e461e7c9069855c9a90442002580d23
Author: Xinli Shang <[email protected]>
AuthorDate: Sun Oct 11 09:30:30 2026 +0900
feat: add CherryPickOperation and wire SnapshotManager::Cherrypick (#926)
Adds `CherryPickOperation` and implements `SnapshotManager::Cherrypick()`,
which was a TODO stub.
Follows Java's `CherryPickOperation`: appends re-apply their added files,
dynamic overwrites (`replace-partitions`) re-apply adds and deletes and are
rejected if a replaced partition changed, and anything else can only be
fast-forward. Records `source-snapshot-id` and `published-wap-id`, and rejects
a WAP id that was already published.
Like Java, the operation is reachable only through `Transaction`, not
`Table`.
Fast-forwards also follow Java: `Apply()` returns the picked snapshot
itself when it is a child of the current state, so no new snapshot is created.
Because that decision is re-evaluated on every apply, a commit retry against
moved metadata replays the pick on top of the new state instead of being
rejected.
Tests ported from Java `TestSnapshotManager` and `TestWapWorkflow`, plus
regression coverage for commit retries and the fast-forward/replay transitions.
Part of #637.
---
src/iceberg/CMakeLists.txt | 1 +
src/iceberg/cpp20_compatibility_internal.h | 1 +
src/iceberg/test/CMakeLists.txt | 1 +
src/iceberg/test/cherry_pick_operation_test.cc | 805 +++++++++++++++++++++++++
src/iceberg/test/snapshot_util_test.cc | 11 +
src/iceberg/transaction.cc | 13 +-
src/iceberg/transaction.h | 4 +
src/iceberg/type_fwd.h | 1 +
src/iceberg/update/cherry_pick_operation.cc | 363 +++++++++++
src/iceberg/update/cherry_pick_operation.h | 85 +++
src/iceberg/update/snapshot_manager.cc | 7 +-
src/iceberg/update/snapshot_manager.h | 7 +-
src/iceberg/update/snapshot_update.h | 3 +-
src/iceberg/util/snapshot_util.cc | 8 +-
14 files changed, 1302 insertions(+), 8 deletions(-)
diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt
index b5fb4874..7394ce82 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/cpp20_compatibility_internal.h
b/src/iceberg/cpp20_compatibility_internal.h
index 5c54b544..8c4a465f 100644
--- a/src/iceberg/cpp20_compatibility_internal.h
+++ b/src/iceberg/cpp20_compatibility_internal.h
@@ -139,6 +139,7 @@
#include "iceberg/transform_function.h"
#include "iceberg/type.h"
#include "iceberg/type_fwd.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"
diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt
index 956a0cbd..058a9dd3 100644
--- a/src/iceberg/test/CMakeLists.txt
+++ b/src/iceberg/test/CMakeLists.txt
@@ -234,6 +234,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 00000000..cb55de9d
--- /dev/null
+++ b/src/iceberg/test/cherry_pick_operation_test.cc
@@ -0,0 +1,805 @@
+/*
+ * 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 <format>
+#include <memory>
+#include <optional>
+#include <string>
+#include <vector>
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+#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/table_properties.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/update/update_partition_spec.h"
+#include "iceberg/update/update_properties.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+class CherryPickOperationTest : public UpdateTestBase {
+ public:
+ static void SetUpTestSuite() { avro::RegisterAll(); }
+
+ protected:
+ /// Most cherry-pick behavior is version independent; only tests that assert
+ /// version-dependent metadata derive from CherryPickFormatVersionTest.
+ virtual int8_t format_version() const { return 2; }
+
+ std::string MetadataResource() const override {
+ return format_version() == 3 ? "TableMetadataV3ValidMinimal.json"
+ : "TableMetadataV2ValidMinimal.json";
+ }
+
+ void SetUp() override {
+ UpdateTestBase::SetUp();
+ if (format_version() == 1) {
+ auto metadata = std::make_shared<TableMetadata>(*table_->metadata());
+ metadata->format_version = 1;
+ metadata->last_sequence_number = 0;
+ const auto metadata_location =
+ std::format("{}/metadata/v1.metadata.json", table_location_);
+ ASSERT_THAT(TableMetadataUtil::Write(*file_io_, metadata_location,
*metadata),
+ IsOk());
+ ASSERT_THAT(catalog_->DropTable(table_ident_, /*purge=*/false), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(table_,
+ catalog_->RegisterTable(table_ident_,
metadata_location));
+ }
+
+ 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<DataFile> MakeDataFile(const std::string& path, int64_t
partition_x) {
+ auto f = std::make_shared<DataFile>();
+ f->content = DataFile::Content::kData;
+ f->file_path = table_location_ + path;
+ f->file_format = FileFormatType::kParquet;
+ f->partition =
PartitionValues(std::vector<Literal>{Literal::Long(partition_x)});
+ f->file_size_in_bytes = 1024;
+ f->record_count = 100;
+ f->partition_spec_id = spec_->spec_id();
+ return f;
+ }
+
+ Result<std::vector<ManifestEntry>> LiveDataEntries(const Snapshot& snapshot)
{
+ std::vector<ManifestEntry> entries;
+ SnapshotReader snapshot_reader(&snapshot);
+ ICEBERG_ASSIGN_OR_RAISE(auto manifests,
snapshot_reader.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 live, reader->LiveEntries());
+ entries.insert(entries.end(), live.begin(), live.end());
+ }
+ return entries;
+ }
+
+ Result<std::vector<std::string>> LiveDataFilePaths() {
+ ICEBERG_ASSIGN_OR_RAISE(auto snapshot, table_->current_snapshot());
+ ICEBERG_ASSIGN_OR_RAISE(auto entries, LiveDataEntries(*snapshot));
+ std::vector<std::string> paths;
+ for (const auto& entry : entries) {
+ paths.push_back(entry.data_file->file_path);
+ }
+ return paths;
+ }
+
+ Result<int64_t> CommitAppend(const std::shared_ptr<DataFile>& file) {
+ ICEBERG_ASSIGN_OR_RAISE(auto append, table_->NewFastAppend());
+ append->AppendFile(file);
+ ICEBERG_RETURN_UNEXPECTED(append->Commit());
+ ICEBERG_RETURN_UNEXPECTED(table_->Refresh());
+ ICEBERG_ASSIGN_OR_RAISE(auto snapshot, table_->current_snapshot());
+ return snapshot->snapshot_id;
+ }
+
+ Result<int64_t> CommitAppendToBranch(const std::string& branch,
+ const std::shared_ptr<DataFile>& file) {
+ ICEBERG_ASSIGN_OR_RAISE(auto append, table_->NewFastAppend());
+ // ToBranch() returns SnapshotUpdate& in C++20 compat builds, so keep the
+ // FastAppend-only call separate.
+ append->ToBranch(branch);
+ append->AppendFile(file);
+ ICEBERG_RETURN_UNEXPECTED(append->Commit());
+ ICEBERG_RETURN_UNEXPECTED(table_->Refresh());
+ return table_->metadata()->refs.at(branch)->snapshot_id;
+ }
+
+ Result<int64_t> StageAppend(const std::shared_ptr<DataFile>& file,
+ const std::optional<std::string>& wap_id =
std::nullopt) {
+ ICEBERG_ASSIGN_OR_RAISE(auto append, table_->NewFastAppend());
+ append->StageOnly();
+ if (wap_id.has_value()) {
+ append->Set(SnapshotSummaryFields::kWAPId, *wap_id);
+ }
+ append->AppendFile(file);
+ ICEBERG_RETURN_UNEXPECTED(append->Commit());
+ ICEBERG_RETURN_UNEXPECTED(table_->Refresh());
+ return table_->metadata()->snapshots.back()->snapshot_id;
+ }
+
+ Result<int64_t> StageReplacePartitions(const std::shared_ptr<DataFile>&
file) {
+ ICEBERG_ASSIGN_OR_RAISE(auto transaction, table_->NewTransaction());
+ ICEBERG_ASSIGN_OR_RAISE(auto replace, transaction->NewReplacePartitions());
+ replace->StageOnly();
+ replace->Set(SnapshotSummaryFields::kReplacePartitions, "true");
+ replace->AddFile(file);
+ ICEBERG_RETURN_UNEXPECTED(replace->Commit());
+ ICEBERG_RETURN_UNEXPECTED(transaction->Commit());
+ ICEBERG_RETURN_UNEXPECTED(table_->Refresh());
+ return table_->metadata()->snapshots.back()->snapshot_id;
+ }
+
+ Result<int64_t> StageOverwrite(const std::shared_ptr<DataFile>& added,
+ const std::shared_ptr<DataFile>& removed) {
+ ICEBERG_ASSIGN_OR_RAISE(auto transaction, table_->NewTransaction());
+ ICEBERG_ASSIGN_OR_RAISE(auto overwrite, transaction->NewOverwrite());
+ overwrite->StageOnly();
+ overwrite->AddFile(added).DeleteFile(removed);
+ ICEBERG_RETURN_UNEXPECTED(overwrite->Commit());
+ ICEBERG_RETURN_UNEXPECTED(transaction->Commit());
+ ICEBERG_RETURN_UNEXPECTED(table_->Refresh());
+ return table_->metadata()->snapshots.back()->snapshot_id;
+ }
+
+ Status RollbackTo(int64_t snapshot_id) {
+ ICEBERG_ASSIGN_OR_RAISE(auto manager, table_->NewSnapshotManager());
+ manager->RollbackTo(snapshot_id);
+ ICEBERG_RETURN_UNEXPECTED(manager->Commit());
+ return table_->Refresh();
+ }
+
+ Status CommitDelete(const std::string& path) {
+ ICEBERG_ASSIGN_OR_RAISE(auto delete_files, table_->NewDeleteFiles());
+ delete_files->DeleteFile(path);
+ ICEBERG_RETURN_UNEXPECTED(delete_files->Commit());
+ return table_->Refresh();
+ }
+
+ Result<std::shared_ptr<Transaction>> StageCherrypick(int64_t snapshot_id) {
+ ICEBERG_ASSIGN_OR_RAISE(auto transaction, table_->NewTransaction());
+ ICEBERG_ASSIGN_OR_RAISE(auto manager, transaction->NewSnapshotManager());
+ manager->Cherrypick(snapshot_id);
+ ICEBERG_RETURN_UNEXPECTED(manager->Commit());
+ return transaction;
+ }
+
+ 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();
+ }
+
+ /// Commit an append through a second table handle while another transaction
is still
+ /// open, so that transaction must rebase and replay at commit time.
+ Result<int64_t> CommitFromOtherHandle(const std::shared_ptr<DataFile>& file)
{
+ ICEBERG_ASSIGN_OR_RAISE(auto other_table,
catalog_->LoadTable(table_ident_));
+ ICEBERG_ASSIGN_OR_RAISE(auto append, other_table->NewFastAppend());
+ append->AppendFile(file);
+ ICEBERG_RETURN_UNEXPECTED(append->Commit());
+ ICEBERG_RETURN_UNEXPECTED(other_table->Refresh());
+ ICEBERG_ASSIGN_OR_RAISE(auto snapshot, other_table->current_snapshot());
+ return snapshot->snapshot_id;
+ }
+
+ /// A fast-forward publishes the staged snapshot itself: no new snapshot, no
new
+ /// sequence number or row ids, and one new snapshot-log entry for it.
+ void ExpectFastForward(const std::shared_ptr<TableMetadata>& before,
+ int64_t staged_id) {
+ ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
+ EXPECT_EQ(snapshot->snapshot_id, staged_id);
+ EXPECT_EQ(table_->metadata()->snapshots.size(), before->snapshots.size());
+ EXPECT_EQ(table_->metadata()->last_sequence_number,
before->last_sequence_number);
+ EXPECT_EQ(table_->metadata()->next_row_id, before->next_row_id);
+ ASSERT_EQ(table_->metadata()->snapshot_log.size(),
before->snapshot_log.size() + 1);
+ EXPECT_EQ(table_->metadata()->snapshot_log.back().snapshot_id, staged_id);
+
EXPECT_FALSE(snapshot->summary.contains(SnapshotSummaryFields::kSourceSnapshotId));
+
EXPECT_FALSE(snapshot->summary.contains(SnapshotSummaryFields::kPublishedWAPId));
+ }
+
+ /// A pick replayed on top of a moved state writes a new snapshot that is
linked to the
+ /// picked one and parented on the state it was applied to.
+ void ExpectReplayedPick(int64_t staged_id, int64_t parent_snapshot_id) {
+ ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
+ EXPECT_NE(snapshot->snapshot_id, staged_id);
+ EXPECT_EQ(snapshot->parent_snapshot_id, parent_snapshot_id);
+ EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kSourceSnapshotId),
+ std::to_string(staged_id));
+ }
+
+ std::shared_ptr<PartitionSpec> spec_;
+ std::shared_ptr<Schema> schema_;
+ std::shared_ptr<DataFile> file_a_;
+ std::shared_ptr<DataFile> file_b_;
+ std::shared_ptr<DataFile> replacement_a_;
+ std::shared_ptr<DataFile> conflict_a_;
+};
+
+class CherryPickFormatVersionTest : public CherryPickOperationTest,
+ public
::testing::WithParamInterface<int8_t> {
+ protected:
+ int8_t format_version() const override { return GetParam(); }
+};
+
+// 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) {
+ ASSERT_THAT(CommitAppend(file_a_), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto staged_id,
StageReplacePartitions(replacement_a_));
+ ASSERT_THAT(CommitAppend(file_b_), IsOk());
+
+ ASSERT_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) {
+ ICEBERG_UNWRAP_OR_FAIL(auto staged_id,
StageReplacePartitions(replacement_a_));
+ ASSERT_THAT(CommitAppend(file_b_), IsOk());
+
+ ASSERT_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) {
+ ASSERT_THAT(CommitAppend(file_a_), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto staged_id,
StageReplacePartitions(replacement_a_));
+ ICEBERG_UNWRAP_OR_FAIL(auto 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) {
+ ASSERT_THAT(CommitAppend(file_a_), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto staged_id,
StageReplacePartitions(replacement_a_));
+ ASSERT_THAT(CommitAppend(file_b_), IsOk());
+ ASSERT_THAT(CommitDelete(file_a_->file_path), IsOk());
+ 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));
+
+ ASSERT_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));
+}
+
+TEST_P(CherryPickFormatVersionTest, CherryPickAppend) {
+ ASSERT_THAT(CommitAppend(file_a_), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto staged_id, StageAppend(file_b_));
+ ICEBERG_UNWRAP_OR_FAIL(auto staged, table_->SnapshotById(staged_id));
+ ASSERT_THAT(CommitAppend(conflict_a_), IsOk());
+ const auto before = table_->metadata();
+ ICEBERG_UNWRAP_OR_FAIL(auto parent, table_->current_snapshot());
+
+ ASSERT_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));
+ EXPECT_EQ(snapshot->parent_snapshot_id, parent->snapshot_id);
+ const int64_t expected_sequence =
+ GetParam() == 1 ? 0 : before->last_sequence_number + 1;
+ EXPECT_EQ(snapshot->sequence_number, expected_sequence);
+ EXPECT_EQ(table_->metadata()->last_sequence_number, expected_sequence);
+ EXPECT_EQ(table_->metadata()->snapshots.size(), before->snapshots.size() +
1);
+ ASSERT_EQ(table_->metadata()->snapshot_log.size(),
before->snapshot_log.size() + 1);
+ EXPECT_EQ(table_->metadata()->snapshot_log.back().snapshot_id,
snapshot->snapshot_id);
+ ExpectBranch(std::string(SnapshotRef::kMainBranch), snapshot->snapshot_id);
+ EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kAddedDataFiles), "1");
+ EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kTotalDataFiles), "3");
+ EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kTotalRecords), "300");
+
+ ICEBERG_UNWRAP_OR_FAIL(auto entries, LiveDataEntries(*snapshot));
+ ASSERT_EQ(entries.size(), 3);
+ int added_files = 0;
+ for (const auto& entry : entries) {
+ if (entry.data_file->file_path == file_b_->file_path) {
+ ++added_files;
+ EXPECT_EQ(entry.status, ManifestStatus::kAdded);
+ EXPECT_EQ(entry.snapshot_id, snapshot->snapshot_id);
+ EXPECT_EQ(entry.sequence_number, expected_sequence);
+ EXPECT_EQ(entry.file_sequence_number, expected_sequence);
+ if (GetParam() == 3) {
+ EXPECT_EQ(entry.data_file->first_row_id, before->next_row_id);
+ EXPECT_NE(entry.data_file->first_row_id, staged->first_row_id);
+ } else {
+ EXPECT_FALSE(entry.data_file->first_row_id.has_value());
+ }
+ }
+ }
+ EXPECT_EQ(added_files, 1);
+ ICEBERG_UNWRAP_OR_FAIL(auto live, LiveDataFilePaths());
+ EXPECT_THAT(live, ::testing::UnorderedElementsAre(
+ file_a_->file_path, file_b_->file_path,
conflict_a_->file_path));
+ if (GetParam() == 3) {
+ EXPECT_EQ(snapshot->first_row_id, before->next_row_id);
+ EXPECT_EQ(snapshot->added_rows, file_b_->record_count);
+ EXPECT_EQ(table_->metadata()->next_row_id,
+ before->next_row_id + file_b_->record_count);
+ }
+}
+
+// Main is empty, but the source snapshot has a parent on another branch.
+TEST_F(CherryPickOperationTest, CherryPickAppendOntoEmptyMain) {
+ ASSERT_THAT(CommitAppendToBranch("b1", file_a_), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto staged_id, CommitAppendToBranch("b1", file_b_));
+
+ ASSERT_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));
+}
+
+// When the picked snapshot's parent is the current snapshot, the pick moves
the
+// current snapshot instead of creating one.
+// When the picked snapshot's parent is the current snapshot, the pick moves
the
+// current snapshot instead of creating one.
+TEST_F(CherryPickOperationTest, FastForwardSetsCurrentSnapshot) {
+ ASSERT_THAT(CommitAppend(file_a_), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto staged_id, StageAppend(file_b_));
+ const auto before = table_->metadata();
+
+ ASSERT_THAT(Cherrypick(staged_id), IsOk());
+
+ ExpectFastForward(before, 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.
+// 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) {
+ ASSERT_THAT(CommitAppend(file_a_), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto staged_id,
+ StageOverwrite(/*added=*/replacement_a_,
/*removed=*/file_a_));
+ const auto before = table_->metadata();
+
+ ASSERT_THAT(Cherrypick(staged_id), IsOk());
+
+ ExpectFastForward(before, staged_id);
+ ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
+ EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kOperation),
+ DataOperation::kOverwrite);
+ 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) {
+ ICEBERG_UNWRAP_OR_FAIL(auto first_id, CommitAppend(file_a_));
+ ASSERT_THAT(CommitAppend(file_b_), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto staged_id,
StageReplacePartitions(replacement_a_));
+ ASSERT_THAT(RollbackTo(first_id), IsOk());
+
+ 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) {
+ ASSERT_THAT(CommitAppend(file_a_), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto first_staged_id,
+ StageAppend(file_b_, /*wap_id=*/"wap-123"));
+
+ ASSERT_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.
+ ICEBERG_UNWRAP_OR_FAIL(auto 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")));
+
+ ASSERT_THAT(table_->Refresh(), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
+ EXPECT_EQ(snapshot->snapshot_id, first_staged_id);
+}
+
+TEST_F(CherryPickOperationTest,
NonPickableFastForwardInvalidatedByConcurrentCommit) {
+ ASSERT_THAT(CommitAppend(file_a_), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto staged_id, StageOverwrite(replacement_a_,
file_a_));
+ ICEBERG_UNWRAP_OR_FAIL(auto transaction, StageCherrypick(staged_id));
+ ICEBERG_UNWRAP_OR_FAIL(auto concurrent_id, CommitFromOtherHandle(file_b_));
+
+ EXPECT_THAT(transaction->Commit(),
+ ::testing::AllOf(
+ IsError(ErrorKind::kValidationFailed),
+ HasErrorMessage(std::format(
+ "Cannot cherry-pick snapshot {}: not append, dynamic
overwrite, "
+ "or fast-forward",
+ staged_id))));
+ ASSERT_THAT(table_->Refresh(), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
+ EXPECT_EQ(snapshot->snapshot_id, concurrent_id);
+ ICEBERG_UNWRAP_OR_FAIL(auto live, LiveDataFilePaths());
+ EXPECT_THAT(live,
+ ::testing::UnorderedElementsAre(file_a_->file_path,
file_b_->file_path));
+}
+
+// A snapshot already in the current history cannot be picked again.
+TEST_F(CherryPickOperationTest, CherryPickAncestorRejected) {
+ ICEBERG_UNWRAP_OR_FAIL(auto first_id, CommitAppend(file_a_));
+ ASSERT_THAT(CommitAppend(file_b_), IsOk());
+
+ 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) {
+ ASSERT_THAT(CommitAppend(file_a_), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto first_staged_id,
+ StageAppend(file_b_, /*wap_id=*/"wap-123"));
+ ICEBERG_UNWRAP_OR_FAIL(auto second_staged_id,
+ StageAppend(conflict_a_, /*wap_id=*/"wap-123"));
+
+ ASSERT_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 snapshot that is neither an append nor a dynamic overwrite can only be
+// fast-forwarded.
+TEST_F(CherryPickOperationTest, NonPickableOperationRejected) {
+ ASSERT_THAT(CommitAppend(file_a_), IsOk());
+ ASSERT_THAT(CommitAppend(file_b_), IsOk());
+
+ 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;
+
+ ASSERT_THAT(CommitAppend(conflict_a_), IsOk());
+
+ 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) {
+ ASSERT_THAT(CommitAppend(file_a_), IsOk());
+
+ EXPECT_THAT(
+ Cherrypick(/*snapshot_id=*/-99),
+ ::testing::AllOf(IsError(ErrorKind::kValidationFailed),
+ HasErrorMessage("Cannot cherry-pick unknown snapshot
ID: -99")));
+}
+
+TEST_F(CherryPickOperationTest, FastForwardOntoEmptyMain) {
+ ICEBERG_UNWRAP_OR_FAIL(auto staged_id, StageAppend(file_a_, "wap-123"));
+ const auto before = table_->metadata();
+
+ ASSERT_THAT(Cherrypick(staged_id), IsOk());
+
+ ExpectFastForward(before, staged_id);
+ ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
+ EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kWAPId), "wap-123");
+ ICEBERG_UNWRAP_OR_FAIL(auto live, LiveDataFilePaths());
+ EXPECT_THAT(live, ::testing::ElementsAre(file_a_->file_path));
+}
+
+TEST_F(CherryPickOperationTest,
FastForwardAppendReplayedAfterConcurrentAppend) {
+ ASSERT_THAT(CommitAppend(file_a_), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto staged_id, StageAppend(file_b_, "wap-123"));
+ ICEBERG_UNWRAP_OR_FAIL(auto transaction, StageCherrypick(staged_id));
+ ICEBERG_UNWRAP_OR_FAIL(auto concurrent_id,
CommitFromOtherHandle(conflict_a_));
+
+ ASSERT_THAT(transaction->Commit(), IsOk());
+ ASSERT_THAT(table_->Refresh(), IsOk());
+
+ ExpectReplayedPick(staged_id, concurrent_id);
+ ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
+ EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kPublishedWAPId),
"wap-123");
+ ICEBERG_UNWRAP_OR_FAIL(auto live, LiveDataFilePaths());
+ EXPECT_THAT(live, ::testing::UnorderedElementsAre(
+ file_a_->file_path, file_b_->file_path,
conflict_a_->file_path));
+}
+
+TEST_F(CherryPickOperationTest,
FastForwardDynamicOverwriteReplayedWithoutConflict) {
+ ASSERT_THAT(CommitAppend(file_a_), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto staged_id,
StageReplacePartitions(replacement_a_));
+ ICEBERG_UNWRAP_OR_FAIL(auto transaction, StageCherrypick(staged_id));
+ ICEBERG_UNWRAP_OR_FAIL(auto concurrent_id, CommitFromOtherHandle(file_b_));
+
+ ASSERT_THAT(transaction->Commit(), IsOk());
+ ASSERT_THAT(table_->Refresh(), IsOk());
+
+ ExpectReplayedPick(staged_id, concurrent_id);
+ ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
+ EXPECT_EQ(snapshot->Operation(), DataOperation::kOverwrite);
+ ICEBERG_UNWRAP_OR_FAIL(auto live, LiveDataFilePaths());
+ EXPECT_THAT(live, ::testing::UnorderedElementsAre(file_b_->file_path,
+
replacement_a_->file_path));
+}
+
+TEST_F(CherryPickOperationTest,
FastForwardDynamicOverwriteReplayedWithConflict) {
+ ASSERT_THAT(CommitAppend(file_a_), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto staged_id,
StageReplacePartitions(replacement_a_));
+ ICEBERG_UNWRAP_OR_FAIL(auto transaction, StageCherrypick(staged_id));
+ ICEBERG_UNWRAP_OR_FAIL(auto concurrent_id,
CommitFromOtherHandle(conflict_a_));
+
+ EXPECT_THAT(transaction->Commit(),
+ ::testing::AllOf(
+ IsError(ErrorKind::kValidationFailed),
+ HasErrorMessage("Cannot cherry-pick replace partitions with
changed "
+ "partition")));
+ ASSERT_THAT(table_->Refresh(), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
+ EXPECT_EQ(snapshot->snapshot_id, concurrent_id);
+ ICEBERG_UNWRAP_OR_FAIL(auto live, LiveDataFilePaths());
+ EXPECT_THAT(
+ live, ::testing::UnorderedElementsAre(file_a_->file_path,
conflict_a_->file_path));
+}
+
+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());
+ 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);
+ ASSERT_THAT(file_io_->ReadFile(draft->manifest_list, std::nullopt), IsOk());
+
+ ICEBERG_UNWRAP_OR_FAIL(auto other_table, catalog_->LoadTable(table_ident_));
+ ICEBERG_UNWRAP_OR_FAIL(auto rollback, other_table->NewSnapshotManager());
+ rollback->RollbackTo(parent_id);
+ ASSERT_THAT(rollback->Commit(), IsOk());
+ ASSERT_THAT(other_table->Refresh(), IsOk());
+ const auto before = other_table->metadata();
+
+ ASSERT_THAT(transaction->Commit(), IsOk());
+ ASSERT_THAT(table_->Refresh(), IsOk());
+
+ ExpectFastForward(before, staged_id);
+ ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
+ EXPECT_THAT(file_io_->ReadFile(draft->manifest_list, std::nullopt),
+ IsError(ErrorKind::kIOError));
+ EXPECT_THAT(file_io_->ReadFile(snapshot->manifest_list, std::nullopt),
IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto live, LiveDataFilePaths());
+ EXPECT_THAT(live,
+ ::testing::UnorderedElementsAre(file_a_->file_path,
file_b_->file_path));
+}
+
+TEST_F(CherryPickOperationTest, FastForwardSucceedsAfterCommitRetries) {
+ ASSERT_THAT(CommitAppend(file_a_), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto staged_id, StageAppend(file_b_, "wap-123"));
+ const auto before = table_->metadata();
+ FailCommits(2);
+
+ ASSERT_THAT(Cherrypick(staged_id), IsOk());
+
+ ExpectFastForward(before, staged_id);
+ ICEBERG_UNWRAP_OR_FAIL(auto live, LiveDataFilePaths());
+ EXPECT_THAT(live,
+ ::testing::UnorderedElementsAre(file_a_->file_path,
file_b_->file_path));
+}
+
+TEST_F(CherryPickOperationTest, CherryPickRecordsPublishedWapId) {
+ ASSERT_THAT(CommitAppend(file_a_), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto staged_id, StageAppend(file_b_, "wap-123"));
+ ICEBERG_UNWRAP_OR_FAIL(auto second_staged_id, StageAppend(replacement_a_,
"wap-123"));
+ ASSERT_THAT(CommitAppend(conflict_a_), IsOk());
+
+ ASSERT_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_EQ(snapshot->summary.at(SnapshotSummaryFields::kPublishedWAPId),
"wap-123");
+ EXPECT_FALSE(snapshot->summary.contains(SnapshotSummaryFields::kWAPId));
+ EXPECT_THAT(Cherrypick(second_staged_id),
+ HasErrorMessage("Duplicate request to cherry pick wap id"));
+ ASSERT_THAT(table_->Refresh(), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto after, table_->current_snapshot());
+ EXPECT_EQ(after->snapshot_id, snapshot->snapshot_id);
+ ICEBERG_UNWRAP_OR_FAIL(auto live, LiveDataFilePaths());
+ EXPECT_THAT(live, ::testing::UnorderedElementsAre(
+ file_a_->file_path, file_b_->file_path,
conflict_a_->file_path));
+
+ // An explicitly empty wap id is still recorded as published-wap-id.
+ auto extra = MakeDataFile("/data/file_extra.parquet", /*partition_x=*/4L);
+ auto bridge = MakeDataFile("/data/file_bridge.parquet", /*partition_x=*/5L);
+ ICEBERG_UNWRAP_OR_FAIL(auto empty_wap_id, StageAppend(extra, /*wap_id=*/""));
+ ASSERT_THAT(CommitAppend(bridge), IsOk());
+
+ ASSERT_THAT(Cherrypick(empty_wap_id), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto empty_wap_snapshot, table_->current_snapshot());
+
EXPECT_EQ(empty_wap_snapshot->summary.at(SnapshotSummaryFields::kPublishedWAPId),
"");
+
EXPECT_EQ(empty_wap_snapshot->summary.at(SnapshotSummaryFields::kSourceSnapshotId),
+ std::to_string(empty_wap_id));
+}
+
+TEST_F(CherryPickOperationTest, DuplicateSourceSnapshotRejected) {
+ ASSERT_THAT(CommitAppend(file_a_), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto staged_id, StageAppend(file_b_));
+ ASSERT_THAT(CommitAppend(conflict_a_), IsOk());
+ ASSERT_THAT(Cherrypick(staged_id), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto picked, table_->current_snapshot());
+
+ EXPECT_THAT(Cherrypick(staged_id),
+ ::testing::AllOf(
+ IsError(ErrorKind::kValidationFailed),
+ HasErrorMessage(std::format(
+ "Cannot cherrypick snapshot {}: already picked to create
ancestor "
+ "{}",
+ staged_id, picked->snapshot_id))));
+
+ ASSERT_THAT(table_->Refresh(), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
+ EXPECT_EQ(snapshot->snapshot_id, picked->snapshot_id);
+ ICEBERG_UNWRAP_OR_FAIL(auto live, LiveDataFilePaths());
+ EXPECT_THAT(live, ::testing::UnorderedElementsAre(
+ file_a_->file_path, file_b_->file_path,
conflict_a_->file_path));
+}
+
+TEST_F(CherryPickOperationTest, DuplicateWapPublishedDuringReplayRejected) {
+ ASSERT_THAT(CommitAppend(file_a_), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto first_id, StageAppend(file_b_, "wap-123"));
+ ICEBERG_UNWRAP_OR_FAIL(auto second_id, StageAppend(replacement_a_,
"wap-123"));
+ ICEBERG_UNWRAP_OR_FAIL(auto transaction, StageCherrypick(second_id));
+
+ ICEBERG_UNWRAP_OR_FAIL(auto other_table, catalog_->LoadTable(table_ident_));
+ ICEBERG_UNWRAP_OR_FAIL(auto publish, other_table->NewSnapshotManager());
+ publish->Cherrypick(first_id);
+ ASSERT_THAT(publish->Commit(), IsOk());
+
+ EXPECT_THAT(
+ transaction->Commit(),
+ ::testing::AllOf(IsError(ErrorKind::kValidationFailed),
+ HasErrorMessage("Duplicate request to cherry pick wap
id")));
+ ASSERT_THAT(table_->Refresh(), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
+ EXPECT_EQ(snapshot->snapshot_id, first_id);
+ ICEBERG_UNWRAP_OR_FAIL(auto live, LiveDataFilePaths());
+ EXPECT_THAT(live,
+ ::testing::UnorderedElementsAre(file_a_->file_path,
file_b_->file_path));
+}
+
+TEST_F(CherryPickOperationTest, CherryPickDynamicOverwriteFromMergedManifest) {
+ ICEBERG_UNWRAP_OR_FAIL(auto properties, table_->NewUpdateProperties());
+ properties->Set(TableProperties::kManifestMinMergeCount.key(), "1");
+ ASSERT_THAT(properties->Commit(), IsOk());
+ ASSERT_THAT(CommitAppend(file_a_), IsOk());
+ ASSERT_THAT(CommitAppend(file_b_), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto staged_id,
StageReplacePartitions(replacement_a_));
+ ICEBERG_UNWRAP_OR_FAIL(auto staged, table_->SnapshotById(staged_id));
+ ICEBERG_UNWRAP_OR_FAIL(auto staged_entries, LiveDataEntries(*staged));
+ ASSERT_THAT(staged_entries, ::testing::Contains(::testing::Field(
+ &ManifestEntry::status,
ManifestStatus::kExisting)));
+ auto new_file = MakeDataFile("/data/another_partition.parquet", 3L);
+ ASSERT_THAT(CommitAppend(new_file), IsOk());
+
+ ASSERT_THAT(Cherrypick(staged_id), IsOk());
+
+ ICEBERG_UNWRAP_OR_FAIL(auto live, LiveDataFilePaths());
+ EXPECT_THAT(
+ live, ::testing::UnorderedElementsAre(file_b_->file_path,
replacement_a_->file_path,
+ new_file->file_path));
+}
+
+INSTANTIATE_TEST_SUITE_P(FormatVersions, CherryPickFormatVersionTest,
+ ::testing::Values(1, 2, 3),
+ [](const ::testing::TestParamInfo<int8_t>& info) {
+ return std::format("V{}", info.param);
+ });
+
+} // namespace iceberg
diff --git a/src/iceberg/test/snapshot_util_test.cc
b/src/iceberg/test/snapshot_util_test.cc
index 25aa59ab..3b6204ff 100644
--- a/src/iceberg/test/snapshot_util_test.cc
+++ b/src/iceberg/test/snapshot_util_test.cc
@@ -222,6 +222,17 @@ TEST_F(SnapshotUtilTest, IsAncestorOf) {
EXPECT_FALSE(result5);
}
+TEST_F(SnapshotUtilTest, IsAncestorOfNoCurrentSnapshot) {
+ TableMetadata metadata;
+ metadata.current_snapshot_id = kInvalidSnapshotId;
+ metadata.snapshots = {
+ CreateSnapshot(base_snapshot_id_, std::nullopt, 1, base_timestamp_)};
+
+ ICEBERG_UNWRAP_OR_FAIL(auto result,
+ SnapshotUtil::IsAncestorOf(metadata,
base_snapshot_id_));
+ EXPECT_FALSE(result);
+}
+
TEST_F(SnapshotUtilTest, CurrentAncestors) {
ICEBERG_UNWRAP_OR_FAIL(auto ancestors,
SnapshotUtil::CurrentAncestors(*table_));
auto ids = ExtractSnapshotIds(ancestors);
diff --git a/src/iceberg/transaction.cc b/src/iceberg/transaction.cc
index 391d6f6c..0eaa3c97 100644
--- a/src/iceberg/transaction.cc
+++ b/src/iceberg/transaction.cc
@@ -35,6 +35,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"
@@ -356,11 +357,14 @@ Status Transaction::ApplyUpdateSnapshot(SnapshotUpdate&
update) {
const auto& base = ctx_->metadata_builder->current();
ICEBERG_ASSIGN_OR_RAISE(auto result, update.Apply());
+ if (result.snapshot == nullptr) {
+ return {};
+ }
// Create a temp builder to check if this is an empty update
auto temp_update = TableMetadataBuilder::BuildFrom(&base);
if (base.SnapshotById(result.snapshot->snapshot_id).has_value()) {
- // This is a rollback operation
+ // Move the branch to an existing snapshot.
temp_update->SetBranchSnapshot(result.snapshot->snapshot_id,
result.target_branch);
} else if (result.stage_only) {
temp_update->AddSnapshot(result.snapshot);
@@ -635,6 +639,13 @@ Result<std::shared_ptr<UpdateLocation>>
Transaction::NewUpdateLocation() {
return update_location;
}
+Result<std::shared_ptr<CherryPickOperation>>
Transaction::NewCherryPickOperation() {
+ ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<CherryPickOperation> cherry_pick,
+ CherryPickOperation::Make(ctx_->table->name().name,
ctx_));
+ ICEBERG_RETURN_UNEXPECTED(AddUpdate(cherry_pick));
+ return cherry_pick;
+}
+
Result<std::shared_ptr<SetSnapshot>> Transaction::NewSetSnapshot() {
ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<SetSnapshot> set_snapshot,
SetSnapshot::Make(ctx_));
diff --git a/src/iceberg/transaction.h b/src/iceberg/transaction.h
index 75e3c831..47678b15 100644
--- a/src/iceberg/transaction.h
+++ b/src/iceberg/transaction.h
@@ -153,6 +153,10 @@ class ICEBERG_EXPORT Transaction : public
std::enable_shared_from_this<Transacti
/// \brief Create a new SnapshotManager to manage snapshots.
Result<std::shared_ptr<SnapshotManager>> NewSnapshotManager();
+ /// \brief Create a new CherryPickOperation to apply the changes of an
existing
+ /// snapshot onto the current state.
+ Result<std::shared_ptr<CherryPickOperation>> NewCherryPickOperation();
+
/// \brief Create a new SetSnapshot to set the current snapshot or rollback
to a
/// previous snapshot and commit the changes.
Result<std::shared_ptr<SetSnapshot>> NewSetSnapshot();
diff --git a/src/iceberg/type_fwd.h b/src/iceberg/type_fwd.h
index f45dd011..84a99190 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 00000000..d0a33076
--- /dev/null
+++ b/src/iceberg/update/cherry_pick_operation.cc
@@ -0,0 +1,363 @@
+/*
+ * 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 <string>
+#include <vector>
+
+#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/content_file_util.h"
+#include "iceberg/util/error_collector.h"
+#include "iceberg/util/macros.h"
+#include "iceberg/util/snapshot_util.h"
+#include "iceberg/util/string_util.h"
+
+namespace iceberg {
+
+namespace {
+
+Result<std::unique_ptr<ManifestReader>> MakeManifestReader(
+ const ManifestFile& manifest, const std::shared_ptr<FileIO>& 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<std::shared_ptr<DataFile>> added;
+ std::vector<std::shared_ptr<DataFile>> 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<SnapshotChanges> ReadSnapshotChanges(const Snapshot& snapshot,
+ const std::shared_ptr<FileIO>&
file_io,
+ const TableMetadata& metadata) {
+ SnapshotChanges changes;
+ SnapshotReader snapshot_reader(&snapshot);
+ ICEBERG_ASSIGN_OR_RAISE(auto manifests,
snapshot_reader.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->EntriesStream());
+ while (true) {
+ ICEBERG_ASSIGN_OR_RAISE(auto entry, entries->Next());
+ if (!entry.has_value()) {
+ break;
+ }
+ if (!entry->data_file) {
+ continue;
+ }
+ if (entry->status == ManifestStatus::kAdded) {
+ changes.added.push_back(std::move(entry->data_file));
+ } else if (entry->status == ManifestStatus::kDeleted) {
+ ContentFileUtil::DropAllStats(*entry->data_file);
+ changes.removed.push_back(std::move(entry->data_file));
+ }
+ }
+ }
+ return changes;
+}
+
+std::optional<std::string> StagedWapId(const Snapshot& snapshot) {
+ auto it = snapshot.summary.find(SnapshotSummaryFields::kWAPId);
+ return it == snapshot.summary.end() ? std::nullopt :
std::make_optional(it->second);
+}
+
+std::optional<std::string> PublishedWapId(const Snapshot& snapshot) {
+ auto it = snapshot.summary.find(SnapshotSummaryFields::kPublishedWAPId);
+ return it == snapshot.summary.end() ? std::nullopt :
std::make_optional(it->second);
+}
+
+Result<std::vector<int64_t>> CurrentAncestorIds(const TableMetadata& metadata)
{
+ std::vector<int64_t> 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<std::optional<std::string>> ValidateWapPublish(const TableMetadata&
metadata,
+ int64_t wap_snapshot_id)
{
+ ICEBERG_ASSIGN_OR_RAISE(auto snapshot,
metadata.SnapshotById(wap_snapshot_id));
+ auto wap_id = StagedWapId(*snapshot);
+ if (!wap_id.has_value() || 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 ValidationFailed(
+ "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() &&
+ StringUtils::EqualsIgnoreCase(it->second, "true");
+}
+
+} // namespace
+
+Result<std::unique_ptr<CherryPickOperation>> CherryPickOperation::Make(
+ std::string table_name, std::shared_ptr<TransactionContext> 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<CherryPickOperation>(
+ new CherryPickOperation(std::move(table_name), std::move(ctx)));
+}
+
+CherryPickOperation::CherryPickOperation(std::string table_name,
+ std::shared_ptr<TransactionContext>
ctx)
+ : MergingSnapshotUpdate(std::move(table_name), std::move(ctx)) {}
+
+std::string CherryPickOperation::operation() {
+ ICEBERG_DCHECK(cherrypick_snapshot_ != nullptr, "Uninitialized cherry-pick
operation");
+ return std::string(cherrypick_snapshot_->Operation().value());
+}
+
+bool CherryPickOperation::IsFastForward(const TableMetadata& base) const {
+ if (base.current_snapshot_id == kInvalidSnapshotId) {
+ return !cherrypick_snapshot_->parent_snapshot_id.has_value();
+ }
+ return cherrypick_snapshot_->parent_snapshot_id == base.current_snapshot_id;
+}
+
+Result<SnapshotUpdate::ApplyResult> CherryPickOperation::Apply() {
+ ICEBERG_RETURN_UNEXPECTED(CheckErrors());
+ const auto& base = this->base();
+ if (cherrypick_snapshot_ == nullptr) {
+ ICEBERG_ASSIGN_OR_RAISE(auto current,
+ SnapshotUtil::OptionalLatestSnapshot(
+ base, std::string(SnapshotRef::kMainBranch)));
+ return ApplyResult{.snapshot = std::move(current),
+ .target_branch = std::string(SnapshotRef::kMainBranch)};
+ }
+
+ const bool is_fast_forward = IsFastForward(base);
+ if (require_fast_forward_ || is_fast_forward) {
+ ICEBERG_CHECK(
+ is_fast_forward,
+ "Cannot cherry-pick snapshot {}: not append, dynamic overwrite, or
fast-forward",
+ cherrypick_snapshot_->snapshot_id);
+ ICEBERG_ASSIGN_OR_RAISE(auto snapshot,
+
base.SnapshotById(cherrypick_snapshot_->snapshot_id));
+ return ApplyResult{.snapshot = std::move(snapshot),
+ .target_branch = std::string(SnapshotRef::kMainBranch)};
+ }
+ return SnapshotUpdate::Apply();
+}
+
+CherryPickOperation& CherryPickOperation::Cherrypick(int64_t snapshot_id) {
+ const TableMetadata& current = ctx_->current();
+ ICEBERG_BUILDER_ASSIGN_OR_RETURN_WITH_ERROR(
+ cherrypick_snapshot_, current.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(current),
+ "Cannot cherry-pick snapshot {}: not append, dynamic overwrite, or
fast-forward",
+ snapshot_id);
+ require_fast_forward_ = true;
+ return *this;
+ }
+
+ if (is_dynamic_overwrite) {
+ // Partition conflicts are checked against files added since the parent
snapshot.
+ // The parent must be a current ancestor, or absent if the overwrite was
based on
+ // an empty table.
+ if (cherrypick_snapshot_->parent_snapshot_id.has_value()) {
+ ICEBERG_BUILDER_ASSIGN_OR_RETURN(
+ bool is_ancestor,
+ SnapshotUtil::IsAncestorOf(current,
+
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(current,
snapshot_id));
+ if (wap_id.has_value()) {
+ Set(SnapshotSummaryFields::kPublishedWAPId, *wap_id);
+ }
+ Set(SnapshotSummaryFields::kSourceSnapshotId, std::to_string(snapshot_id));
+
+ if (is_dynamic_overwrite) {
+ // A replace can only be re-applied if the files it removed are all
present.
+ FailMissingDeletePaths();
+ replaced_partitions_.emplace();
+ }
+
+ auto io = ctx_->table->io();
+ ICEBERG_BUILDER_ASSIGN_OR_RETURN(
+ auto changes, ReadSnapshotChanges(*cherrypick_snapshot_, io, current));
+
+ for (const auto& added_file : changes.added) {
+ ICEBERG_BUILDER_RETURN_IF_ERROR(AddDataFile(added_file));
+ if (replaced_partitions_.has_value()) {
+ ICEBERG_BUILDER_CHECK(added_file->partition_spec_id.has_value(),
+ "Data file must have partition spec ID");
+ replaced_partitions_->add(added_file->partition_spec_id.value(),
+ added_file->partition);
+ }
+ }
+
+ if (is_dynamic_overwrite) {
+ for (const auto& deleted_file : changes.removed) {
+ ICEBERG_BUILDER_RETURN_IF_ERROR(DeleteDataFile(deleted_file));
+ }
+ }
+
+ return *this;
+}
+
+Status CherryPickOperation::ValidateNonAncestor(const TableMetadata& meta,
+ int64_t snapshot_id) const {
+ ICEBERG_ASSIGN_OR_RAISE(bool is_ancestor,
+ SnapshotUtil::IsAncestorOf(meta, snapshot_id));
+ if (is_ancestor) {
+ return ValidationFailed("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(meta));
+ for (int64_t ancestor_id : ancestor_ids) {
+ ICEBERG_ASSIGN_OR_RAISE(auto ancestor, meta.SnapshotById(ancestor_id));
+ auto it = ancestor->summary.find(SnapshotSummaryFields::kSourceSnapshotId);
+ if (it != ancestor->summary.end() && it->second == snapshot_id_str) {
+ return ValidationFailed(
+ "Cannot cherrypick snapshot {}: already picked to create ancestor
{}",
+ snapshot_id, ancestor_id);
+ }
+ }
+ return {};
+}
+
+Status CherryPickOperation::ValidateReplacedPartitions(const TableMetadata&
meta) const {
+ if (!replaced_partitions_.has_value() ||
+ meta.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(meta,
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 snapshots, SnapshotUtil::AncestorsBetween(
+ meta, meta.current_snapshot_id,
parent_id));
+ // TODO: Parallelize added-file validation using the planning executor.
+ // Bound active readers and wait for in-flight tasks before returning.
+ for (const auto& snap : snapshots) {
+ SnapshotReader snapshot_reader(snap.get());
+ ICEBERG_ASSIGN_OR_RAISE(auto manifests, snapshot_reader.DataManifests(io));
+ for (const auto& manifest : manifests) {
+ if (manifest.added_snapshot_id != snap->snapshot_id) {
+ continue;
+ }
+ ICEBERG_ASSIGN_OR_RAISE(auto reader, MakeManifestReader(manifest, io,
meta));
+ ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->EntriesStream());
+ while (true) {
+ ICEBERG_ASSIGN_OR_RAISE(auto entry, entries->Next());
+ if (!entry.has_value()) {
+ break;
+ }
+ if (entry->status != ManifestStatus::kAdded || !entry->data_file) {
+ continue;
+ }
+ const auto& new_file = entry->data_file;
+ if (!new_file->partition_spec_id.has_value() ||
+
!replaced_partitions_->contains(new_file->partition_spec_id.value(),
+ new_file->partition)) {
+ continue;
+ }
+ ICEBERG_ASSIGN_OR_RAISE(auto spec,
+
meta.PartitionSpecById(*new_file->partition_spec_id));
+ ICEBERG_ASSIGN_OR_RAISE(auto partition_path,
+ spec->PartitionPath(new_file->partition));
+ return ValidationFailed(
+ "Cannot cherry-pick replace partitions with changed partition: {}",
+ partition_path);
+ }
+ }
+ }
+ return {};
+}
+
+Status CherryPickOperation::Validate(
+ const TableMetadata& base,
+ [[maybe_unused]] const std::shared_ptr<Snapshot>& snapshot) {
+ if (IsFastForward(base)) {
+ return {};
+ }
+
+ ICEBERG_RETURN_UNEXPECTED(ValidateNonAncestor(base,
cherrypick_snapshot_->snapshot_id));
+ ICEBERG_RETURN_UNEXPECTED(ValidateReplacedPartitions(base));
+ ICEBERG_RETURN_UNEXPECTED(ValidateWapPublish(base,
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 00000000..5c3211c4
--- /dev/null
+++ b/src/iceberg/update/cherry_pick_operation.h
@@ -0,0 +1,85 @@
+/*
+ * 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 <cstdint>
+#include <memory>
+#include <optional>
+#include <string>
+
+#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 or fast-forwards the current state to a snapshot.
+///
+/// This update is not exposed through the Table API. It is part of the
+/// Transaction API intended for use in SnapshotManager.
+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<std::unique_ptr<CherryPickOperation>> Make(
+ std::string table_name, std::shared_ptr<TransactionContext> ctx);
+
+ /// \brief Cherry-pick the changes of the given snapshot, or fast-forward to
it.
+ ///
+ /// \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);
+
+ protected:
+ Result<ApplyResult> Apply() override;
+
+ std::string operation() override;
+
+ Status Validate(const TableMetadata& base,
+ const std::shared_ptr<Snapshot>& snapshot) override;
+
+ private:
+ explicit CherryPickOperation(std::string table_name,
+ std::shared_ptr<TransactionContext> ctx);
+
+ /// \brief Whether the snapshot passed to Cherrypick() can be fast-forwarded.
+ bool IsFastForward(const TableMetadata& base) 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& meta, int64_t snapshot_id)
const;
+
+ /// \brief If there is a current snapshot, require an ancestral parent (if
any)
+ /// and reject files added to replaced partitions since that parent.
+ Status ValidateReplacedPartitions(const TableMetadata& meta) const;
+
+ std::shared_ptr<Snapshot> cherrypick_snapshot_;
+ bool require_fast_forward_ = false;
+ std::optional<PartitionSet> replaced_partitions_;
+};
+
+} // namespace iceberg
diff --git a/src/iceberg/update/snapshot_manager.cc
b/src/iceberg/update/snapshot_manager.cc
index 405b029f..d825403b 100644
--- a/src/iceberg/update/snapshot_manager.cc
+++ b/src/iceberg/update/snapshot_manager.cc
@@ -23,6 +23,7 @@
#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"
@@ -55,8 +56,10 @@ 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");
+ 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 fd81f833..3dfdb25f 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/update/snapshot_update.h
b/src/iceberg/update/snapshot_update.h
index cfb05188..3f06c8b3 100644
--- a/src/iceberg/update/snapshot_update.h
+++ b/src/iceberg/update/snapshot_update.h
@@ -62,6 +62,7 @@ class ICEBERG_EXPORT SnapshotUpdate : public PendingUpdate {
public:
/// \brief Result of applying a snapshot update
struct ApplyResult {
+ // Null when applying an unconfigured update to an empty table.
std::shared_ptr<Snapshot> snapshot;
std::string target_branch;
bool stage_only = false;
@@ -200,7 +201,7 @@ class ICEBERG_EXPORT SnapshotUpdate : public PendingUpdate {
friend class Transaction;
/// Build snapshot state for the transaction's current metadata.
- Result<ApplyResult> Apply();
+ virtual Result<ApplyResult> Apply();
Status Finalize(const TableMetadata& committed) override;
Status CleanStaged() override;
diff --git a/src/iceberg/util/snapshot_util.cc
b/src/iceberg/util/snapshot_util.cc
index 6aed2826..885189c2 100644
--- a/src/iceberg/util/snapshot_util.cc
+++ b/src/iceberg/util/snapshot_util.cc
@@ -66,9 +66,11 @@ Result<bool> SnapshotUtil::IsAncestorOf(const Table& table,
Result<bool> SnapshotUtil::IsAncestorOf(const TableMetadata& metadata,
int64_t ancestor_snapshot_id) {
- ICEBERG_ASSIGN_OR_RAISE(auto current, metadata.Snapshot());
- ICEBERG_CHECK(current != nullptr, "Current snapshot is null");
- return IsAncestorOf(metadata, current->snapshot_id, ancestor_snapshot_id);
+ // An empty current state has no ancestors.
+ if (metadata.current_snapshot_id == kInvalidSnapshotId) {
+ return false;
+ }
+ return IsAncestorOf(metadata, metadata.current_snapshot_id,
ancestor_snapshot_id);
}
Result<bool> SnapshotUtil::IsAncestorOf(const Table& table, int64_t
snapshot_id,
