Cherrypick snapshot feature

api/src/main/java/org/apache/iceberg/ManageSnapshots.java

@@ -0,0 +1,83 @@
+/*
+ * 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.
+ */
+
+package org.apache.iceberg;
+
+import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.exceptions.DuplicateWAPCommitException;
+import org.apache.iceberg.exceptions.ValidationException;
+
+/**
+ * API for managing snapshots. Allows rolling table data back to a stated at an older table {@link Snapshot snapshot}.
+ * Rollback:
+ *  <p>
+ *  This API does not allow conflicting calls to {@link #setCurrentSnapshot(long)} and
+ *  {@link #rollbackToTime(long)}.
+ *  <p>
+ *  When committing, these changes will be applied to the current table metadata. Commit conflicts
+ *  will not be resolved and will result in a {@link CommitFailedException}.
+ * Cherrypick:
+ *  <p>
+ *  In an audit workflow, new data is written to an orphan {@link Snapshot snapshot} that is not committed as
+ *  the table's current state until it is audited. After auditing a change, it may need to be applied or cherry-picked
+ *  on top of the latest snapshot 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.
+ *  <p>
+ */
+public interface ManageSnapshots extends PendingUpdate<Snapshot> {
+
+  /**
+   * Roll this table's data back to a specific {@link Snapshot} identified by id.
+   *
+   * @param snapshotId long id of the snapshot to roll back table data to
+   * @return this for method chaining
+   * @throws IllegalArgumentException If the table has no snapshot with the given id
+   */
+  ManageSnapshots setCurrentSnapshot(long snapshotId);
+
+  /**
+   * Roll this table's data back to the last {@link Snapshot} before the given timestamp.
+   *
+   * @param timestampMillis a long timestamp, as returned by {@link System#currentTimeMillis()}
+   * @return this for method chaining
+   * @throws IllegalArgumentException If the table has no old snapshot before the given timestamp
+   */
+  ManageSnapshots rollbackToTime(long timestampMillis);
+
+  /**
+   * Rollback table's state to a specific {@link Snapshot} identified by id.
+   * @param snapshotId long id of snapshot id to roll back table to. Must be an ancestor of the current snapshot
+   * @throws IllegalArgumentException If the table has no snapshot with the given id
+   * @throws ValidationException If given snapshot id is not an ancestor of the current state
+   */
+  ManageSnapshots rollbackTo(long snapshotId);
+
+  /**
+   * Apply supported changes in given snapshot and create a new snapshot which will be set as the
+   * current snapshot on commit.
+   * @param snapshotId a snapshotId whose changes to apply
+   * @return this for method chaining
+   * @throws IllegalArgumentException If the table has no snapshot with the given id
+   * @throws DuplicateWAPCommitException In case of a WAP workflow and if the table has has a duplicate commit with same
+   * wapId
+   */
+  ManageSnapshots cherrypick(long snapshotId);
+}

api/src/main/java/org/apache/iceberg/Rollback.java

@@ -38,7 +38,9 @@ public interface Rollback extends PendingUpdate<Snapshot> {
    * @param snapshotId long id of the snapshot to roll back table data to
    * @return this for method chaining
    * @throws IllegalArgumentException If the table has no snapshot with the given id
+   * @deprecated Replaced by {@link ManageSnapshots#setCurrentSnapshot(long)}
    */
+  @Deprecated
   Rollback toSnapshotId(long snapshotId);
 
   /**
@@ -47,7 +49,8 @@ public interface Rollback extends PendingUpdate<Snapshot> {
    * @param timestampMillis a long timestamp, as returned by {@link System#currentTimeMillis()}
    * @return this for method chaining
    * @throws IllegalArgumentException If the table has no old snapshot before the given timestamp
+   * @deprecated Replaced by {@link ManageSnapshots#rollbackToTime(long)}
    */
+  @Deprecated
   Rollback toSnapshotAtTime(long timestampMillis);
-
 }

api/src/main/java/org/apache/iceberg/Table.java

@@ -203,9 +203,17 @@ default AppendFiles newFastAppend() {
    * Create a new {@link Rollback rollback API} to roll back to a previous snapshot and commit.
    *
    * @return a new {@link Rollback}
+   * @deprecated Replaced by {@link #manageSnapshots()}
    */
+  @Deprecated
   Rollback rollback();
 
+  /**
+   * Create a new {@link ManageSnapshots manage snapshots API} to manage snapshots in this table and commit.
+   * @return a new {@link ManageSnapshots}
+   */
+  ManageSnapshots manageSnapshots();
+
   /**
    * Create a new {@link Transaction transaction API} to commit multiple table operations at once.
    *

api/src/main/java/org/apache/iceberg/exceptions/CherrypickAncestorCommitException.java

@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+
+package org.apache.iceberg.exceptions;
+
+/**
+ * This exception occurs when one cherrypicks an ancestor or when the picked snapshot is already linked to
+ * a published ancestor. This additionally helps avoid duplicate cherrypicks on non-WAP snapshots.
+ */
+public class CherrypickAncestorCommitException extends ValidationException {
+
+  public CherrypickAncestorCommitException(long snapshotId) {
+    super("Cannot cherrypick snapshot %s: already an ancestor", String.valueOf(snapshotId));
+  }
+
+  public CherrypickAncestorCommitException(long snapshotId, long publishedAncestorId) {
+    super("Cannot cherrypick snapshot %s: already picked to create ancestor %s",
+        String.valueOf(snapshotId), String.valueOf(publishedAncestorId));
+  }
+}

api/src/main/java/org/apache/iceberg/exceptions/DuplicateWAPCommitException.java

@@ -0,0 +1,31 @@
+/*
+ * 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.
+ */
+
+package org.apache.iceberg.exceptions;
+
+/**
+ * This exception occurs when the WAP workflow detects a duplicate wap commit. This helps clients
+ * to detect duplicate snapshots that are connected by the same wap id.
+ */
+public class DuplicateWAPCommitException extends ValidationException {
+
+  public DuplicateWAPCommitException(String wapId) {
+    super("Duplicate request to cherry pick wap id that was published already: %s", wapId);
+  }
+}

core/src/main/java/org/apache/iceberg/BaseMetadataTable.java

@@ -142,6 +142,11 @@ public Rollback rollback() {
     throw new UnsupportedOperationException("Cannot roll back a metadata table");
   }
 
+  @Override
+  public ManageSnapshots manageSnapshots() {
+    throw new UnsupportedOperationException("Cannot manage snapshots in a metadata table");
+  }
+
   @Override
   public Transaction newTransaction() {
     throw new UnsupportedOperationException("Cannot create transactions for a metadata table");

core/src/main/java/org/apache/iceberg/BaseSnapshot.java

@@ -163,7 +163,7 @@ private void cacheChanges() {
         manifest -> Objects.equal(manifest.snapshotId(), snapshotId));
     try (CloseableIterable<ManifestEntry> entries = new ManifestGroup(io, changedManifests)
         .ignoreExisting()
-        .select(ManifestReader.CHANGE_WITH_STATS_COLUMNS)
+        .select(ManifestReader.ALL_COLUMNS)
         .entries()) {
       for (ManifestEntry entry : entries) {
         switch (entry.status()) {

core/src/main/java/org/apache/iceberg/BaseTable.java

@@ -159,6 +159,11 @@ public Rollback rollback() {
     return new RollbackToSnapshot(ops);
   }
 
+  @Override
+  public ManageSnapshots manageSnapshots() {
+    return new SnapshotManager(ops);
+  }
+
   @Override
   public Transaction newTransaction() {
     return Transactions.newTransaction(ops);

core/src/main/java/org/apache/iceberg/BaseTransaction.java

@@ -585,6 +585,11 @@ public Rollback rollback() {
       throw new UnsupportedOperationException("Transaction tables do not support rollback");
     }
 
+    @Override
+    public ManageSnapshots manageSnapshots() {
+      throw new UnsupportedOperationException("Transaction tables do not support managing snapshots");
+    }
+
     @Override
     public Transaction newTransaction() {
       throw new UnsupportedOperationException("Cannot create a transaction within a transaction");

core/src/main/java/org/apache/iceberg/ManifestReader.java

@@ -49,7 +49,7 @@
 public class ManifestReader extends CloseableGroup implements Filterable<FilteredManifest> {
   private static final Logger LOG = LoggerFactory.getLogger(ManifestReader.class);
 
-  private static final ImmutableList<String> ALL_COLUMNS = ImmutableList.of("*");
+  static final ImmutableList<String> ALL_COLUMNS = ImmutableList.of("*");
   static final ImmutableList<String> CHANGE_COLUMNS = ImmutableList.of(
       "file_path", "file_format", "partition", "record_count", "file_size_in_bytes");
   static final ImmutableList<String> CHANGE_WITH_STATS_COLUMNS = ImmutableList.<String>builder()

core/src/main/java/org/apache/iceberg/RollbackToSnapshot.java

@@ -19,61 +19,21 @@
 
 package org.apache.iceberg;
 
-import com.google.common.base.Preconditions;
-import org.apache.iceberg.exceptions.ValidationException;
-
-class RollbackToSnapshot implements Rollback {
-  private final TableOperations ops;
-  private TableMetadata base = null;
-  private Long targetSnapshotId = null;
+class RollbackToSnapshot extends SnapshotManager implements Rollback {
 
   RollbackToSnapshot(TableOperations ops) {
-    this.ops = ops;
-    this.base = ops.current(); // do not retry
+    super(ops);
   }
 
   @Override
   public Rollback toSnapshotId(long snapshotId) {
-    Preconditions.checkArgument(base.snapshot(snapshotId) != null,
-        "Cannot roll back to unknown snapshot id: %s", snapshotId);
-
-    this.targetSnapshotId = snapshotId;
-
+    super.setCurrentSnapshot(snapshotId);
     return this;
   }
 
   @Override
   public Rollback toSnapshotAtTime(long timestampMillis) {
-    long snapshotId = 0;
-    long snapshotTimestamp = 0;
-    // find the latest snapshot by timestamp older than timestampMillis
-    for (Snapshot snapshot : base.snapshots()) {
-      if (snapshot.timestampMillis() < timestampMillis &&
-          snapshot.timestampMillis() > snapshotTimestamp) {
-        snapshotId = snapshot.snapshotId();
-        snapshotTimestamp = snapshot.timestampMillis();
-      }
-    }
-
-    Preconditions.checkArgument(base.snapshot(snapshotId) != null,
-        "Cannot roll back, no valid snapshot older than: %s", timestampMillis);
-
-    this.targetSnapshotId = snapshotId;
-
+    super.rollbackToTime(timestampMillis);
     return this;
   }
-
-  @Override
-  public Snapshot apply() {
-    ValidationException.check(targetSnapshotId != null,
-        "Cannot roll back to unknown version: call toSnapshotId or toSnapshotAtTime");
-    return base.snapshot(targetSnapshotId);
-  }
-
-  @Override
-  public void commit() {
-    // rollback does not refresh or retry. it only operates on the state of the table when rollback
-    // was called to create the transaction.
-    ops.commit(base, base.rollbackTo(apply()));
-  }
 }