feat: add CherryPickOperation and wire SnapshotManager::Cherrypick

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

src/iceberg/test/CMakeLists.txt

@@ -232,6 +232,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

src/iceberg/test/snapshot_util_test.cc

@@ -222,6 +222,19 @@ TEST_F(SnapshotUtilTest, IsAncestorOf) {
   EXPECT_FALSE(result5);
 }
 
+// No current snapshot means an empty ancestor history, so nothing can be an
+// ancestor of it; this must return false rather than raising "No current
+// snapshot", matching Java's SnapshotUtil.ancestorsOf(null, ...).
+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);

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"
@@ -570,6 +571,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_));

src/iceberg/transaction.h

@@ -153,6 +153,13 @@ 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.
+  ///
+  /// \note Intended for use by SnapshotManager. Prefer
+  /// SnapshotManager::Cherrypick(), which also handles the fast-forward case.
+  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();

src/iceberg/type_fwd.h

@@ -250,6 +250,7 @@ class Transaction;
 class TransactionContext;
 
 /// \brief Update family.
+class CherryPickOperation;
 class DeleteFiles;
 class ExpireSnapshots;
 class FastAppend;