API: Add snapshot reference create, replace, rename, and rename operations to ManageSnapshots API

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

@@ -80,4 +80,121 @@ public interface ManageSnapshots extends PendingUpdate<Snapshot> {
    * wapId
    */
   ManageSnapshots cherrypick(long snapshotId);
+
+  /**
+   * Create a new branch pointing to the given snapshot id.
+   *
+   * @param name branch name
+   * @param snapshotId id of the snapshot which will be the head of the branch
+   * @return this for method chaining
+   * @throws IllegalArgumentException if a branch with the given name already exists
+   */
+  ManageSnapshots createBranch(String name, long snapshotId);
+
+  /**
+   * Create a new tag pointing to the given snapshot id
+   *
+   * @param name tag name
+   * @param snapshotId snapshotId for the head of the new branch.
+   * @return this for method chaining
+   * @throws IllegalArgumentException if a tag with the given name already exists
+   */
+  ManageSnapshots createTag(String name, long snapshotId);
+
+  /**
+   * Remove a branch by name
+   *
+   * @param name branch name
+   * @return this for method chaining
+   * @throws IllegalArgumentException if the branch does not exist
+   */
+  ManageSnapshots removeBranch(String name);
+
+  /**
+   * Rename a branch
+   *
+   * @param name name of branch to rename
+   * @param newName the desired new name of the branch
+   * @throws IllegalArgumentException if the branch to rename does not exist or if there is already a branch
+   * with the same name as the desired new name.
+   */
+  ManageSnapshots renameBranch(String name, String newName);
+
+  /**
+   * Remove the tag with the given name.
+   *
+   * @param name tag name
+   * @return this for method chaining
+   * @throws IllegalArgumentException if the branch does not exist
+   */
+  ManageSnapshots removeTag(String name);
+
+  /**
+   * Replaces the tag with the given name to point to the specified snapshot.
+   *
+   * @param name Tag to replace
+   * @param snapshotId new snapshot id for the given tag
+   * @return this for method chaining
+   */
+  ManageSnapshots replaceTag(String name, long snapshotId);
+
+  /**
+   * Replaces the branch with the given name to point to the specified snapshot
+   *
+   * @param name Branch to replace
+   * @param snapshotId new snapshot id for the given branch
+   * @return this for method chaining
+   */
+  ManageSnapshots replaceBranch(String name, long snapshotId);
+
+  /**
+   * Replaces the branch with the given name to point to the source snapshot.
+   * The source branch will remain unchanged, the target branch will retain its retention properties.
+   *
+   * @param name Branch to replace
+   * @param source Source reference for the target to be replaced with
+   * @return this for method chaining
+   */
+  ManageSnapshots replaceBranch(String name, String source);
+
+  /**
+   * Performs a fast-forward of the given target branch up to the source snapshot if target is an ancestor of source.
+   * The source branch will remain unchanged, the target branch will retain its retention properties.
+   *
+   * @param name Branch to fast-forward
+   * @param source Source reference for the target to be fast forwarded to
+   * @return this for method chaining
+   * @throws IllegalArgumentException if the target branch is not an ancestor of source
+   */
+  ManageSnapshots fastForwardBranch(String name, String source);
+
+  /**
+   * Updates the minimum number of snapshots to keep for a branch.
+   *
+   * @param branchName branch name
+   * @param minSnapshotsToKeep minimum number of snapshots to retain on the branch
+   * @return this for method chaining
+   * @throws IllegalArgumentException if the branch does not exist
+   */
+  ManageSnapshots setMinSnapshotsToKeep(String branchName, int minSnapshotsToKeep);
+
+  /**
+   * Updates the max snapshot age for a branch.
+   *
+   * @param branchName branch name
+   * @param maxSnapshotAgeMs maximum snapshot age in milliseconds to retain on branch
+   * @return this for method chaining
+   * @throws IllegalArgumentException if the branch does not exist
+   */
+  ManageSnapshots setMaxSnapshotAgeMs(String branchName, long maxSnapshotAgeMs);
+
+  /**
+   * Updates the retention policy for a reference.
+   *
+   * @param name branch name
+   * @param maxRefAgeMs retention age in milliseconds of the tag reference itself
+   * @return this for method chaining
+   * @throws IllegalArgumentException if the reference does not exist
+   */
+  ManageSnapshots setMaxRefAgeMs(String name, long maxRefAgeMs);
 }

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

@@ -238,6 +238,14 @@ SetSnapshotOperation setBranchSnapshot() {
     return set;
   }
 
+  UpdateSnapshotReferencesOperation updateSnapshotReferencesOperation() {
+    checkLastOperationCommitted("UpdateSnapshotReferencesOperation");
+    UpdateSnapshotReferencesOperation manageSnapshotRefOperation =
+        new UpdateSnapshotReferencesOperation(transactionOps);
+    updates.add(manageSnapshotRefOperation);
+    return manageSnapshotRefOperation;
+  }
+
   @Override
   public void commitTransaction() {
     Preconditions.checkState(hasLastOpCommitted,

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

@@ -20,6 +20,7 @@
 package org.apache.iceberg;
 
 import java.io.Serializable;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
@@ -227,31 +228,60 @@ public String name() {
 
     @Override
     public void applyTo(TableMetadata.Builder metadataBuilder) {
-      // TODO: this should be generalized when tagging is supported
-      metadataBuilder.removeBranch(name);
+      metadataBuilder.removeRef(name);
     }
   }
 
   class SetSnapshotRef implements MetadataUpdate {
     private final String name;
-    private final long snapshotId;
-
-    public SetSnapshotRef(String name, long snapshotId) {
+    private final Long snapshotId;
+    private final SnapshotRefType type;
+    private Integer minSnapshotsToKeep;
+    private Long maxSnapshotAgeMs;
+    private Long maxRefAgeMs;
+
+    public SetSnapshotRef(String name, Long snapshotId, SnapshotRefType type, Integer minSnapshotsToKeep,
+                          Long maxSnapshotAgeMs, Long maxRefAgeMs) {
       this.name = name;
       this.snapshotId = snapshotId;
+      this.type = type;
+      this.minSnapshotsToKeep = minSnapshotsToKeep;
+      this.maxSnapshotAgeMs = maxSnapshotAgeMs;
+      this.maxRefAgeMs = maxRefAgeMs;
     }
 
     public String name() {
       return name;
     }
 
+    public String type() {
+      return type.name().toLowerCase(Locale.ROOT);
+    }
+
     public long snapshotId() {
       return snapshotId;
     }
 
+    public Integer minSnapshotsToKeep() {
+      return minSnapshotsToKeep;
+    }
+
+    public Long maxSnapshotAgeMs() {
+      return maxSnapshotAgeMs;
+    }
+
+    public Long maxRefAgeMs() {
+      return maxRefAgeMs;
+    }
+
     @Override
     public void applyTo(TableMetadata.Builder metadataBuilder) {
-      metadataBuilder.setBranchSnapshot(snapshotId, name);
+      SnapshotRef ref = SnapshotRef.builderFor(snapshotId, type)
+          .minSnapshotsToKeep(minSnapshotsToKeep)
+          .maxSnapshotAgeMs(maxSnapshotAgeMs)
+          .maxRefAgeMs(maxRefAgeMs)
+          .build();
+      metadataBuilder.setRef(name, ref);
     }
   }
 

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

@@ -24,6 +24,7 @@
 public class SnapshotManager implements ManageSnapshots {
 
   private final BaseTransaction transaction;
+  private UpdateSnapshotReferencesOperation updateSnapshotReferencesOperation;
 
   SnapshotManager(String tableName, TableOperations ops) {
     Preconditions.checkState(ops.current() != null, "Cannot manage snapshots: table %s does not exist", tableName);
@@ -32,35 +33,127 @@ public class SnapshotManager implements ManageSnapshots {
 
   @Override
   public ManageSnapshots cherrypick(long snapshotId) {
+    commitIfRefUpdatesExist();
     transaction.cherryPick().cherrypick(snapshotId).commit();
     return this;
   }
 
   @Override
   public ManageSnapshots setCurrentSnapshot(long snapshotId) {
+    commitIfRefUpdatesExist();
     transaction.setBranchSnapshot().setCurrentSnapshot(snapshotId).commit();
     return this;
   }
 
   @Override
   public ManageSnapshots rollbackToTime(long timestampMillis) {
+    commitIfRefUpdatesExist();
     transaction.setBranchSnapshot().rollbackToTime(timestampMillis).commit();
     return this;
   }
 
   @Override
   public ManageSnapshots rollbackTo(long snapshotId) {
+    commitIfRefUpdatesExist();
     transaction.setBranchSnapshot().rollbackTo(snapshotId).commit();
     return this;
   }
 
+  @Override
+  public ManageSnapshots createBranch(String name, long snapshotId) {
+    updateSnapshotReferencesOperation().createBranch(name, snapshotId);
+    return this;
+  }
+
+  @Override
+  public ManageSnapshots createTag(String name, long snapshotId) {
+    updateSnapshotReferencesOperation().createTag(name, snapshotId);
+    return this;
+  }
+
+  @Override
+  public ManageSnapshots removeBranch(String name) {
+    updateSnapshotReferencesOperation().removeBranch(name);
+    return this;
+  }
+
+  @Override
+  public ManageSnapshots removeTag(String name) {
+    updateSnapshotReferencesOperation().removeTag(name);
+    return this;
+  }
+
+  @Override
+  public ManageSnapshots setMinSnapshotsToKeep(String name, int minSnapshotsToKeep) {
+    updateSnapshotReferencesOperation().setMinSnapshotsToKeep(name, minSnapshotsToKeep);
+    return this;
+  }
+
+  @Override
+  public ManageSnapshots setMaxSnapshotAgeMs(String name, long maxSnapshotAgeMs) {
+    updateSnapshotReferencesOperation().setMaxSnapshotAgeMs(name, maxSnapshotAgeMs);
+    return this;
+  }
+
+  @Override
+  public ManageSnapshots setMaxRefAgeMs(String name, long maxRefAgeMs) {
+    updateSnapshotReferencesOperation().setMaxRefAgeMs(name, maxRefAgeMs);
+    return this;
+  }
+
+  @Override
+  public ManageSnapshots replaceTag(String name, long snapshotId)  {
+    updateSnapshotReferencesOperation().replaceTag(name, snapshotId);
+    return this;
+  }
+
+  @Override
+  public ManageSnapshots replaceBranch(String name, long snapshotId)  {
+    updateSnapshotReferencesOperation().replaceBranch(name, snapshotId);
+    return this;
+  }
+
+  @Override
+  public ManageSnapshots replaceBranch(String name, String source) {
+    updateSnapshotReferencesOperation().replaceBranch(name, source);
+    return this;
+  }
+
+  @Override
+  public ManageSnapshots fastForwardBranch(String name, String source) {
+    updateSnapshotReferencesOperation().fastForward(name, source);
+    return this;
+  }
+
+  @Override
+  public ManageSnapshots renameBranch(String name, String newName) {
+    updateSnapshotReferencesOperation().renameBranch(name, newName);
+    return this;
+  }
+
+  private UpdateSnapshotReferencesOperation updateSnapshotReferencesOperation() {
+    if (updateSnapshotReferencesOperation == null) {
+      this.updateSnapshotReferencesOperation = transaction.updateSnapshotReferencesOperation();
+    }
+
+    return updateSnapshotReferencesOperation;
+  }
+
+  private void commitIfRefUpdatesExist() {
+    if (updateSnapshotReferencesOperation != null) {
+      updateSnapshotReferencesOperation.commit();
+      updateSnapshotReferencesOperation = null;
+    }
+  }
+
   @Override
   public Snapshot apply() {
     return transaction.table().currentSnapshot();
   }
 
   @Override
   public void commit() {
+    commitIfRefUpdatesExist();
     transaction.commitTransaction();
   }
 }