Core, API: Performing operations on a snapshot branch ref

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

@@ -60,4 +60,10 @@ public interface SnapshotUpdate<ThisT> extends PendingUpdate<Snapshot> {
    * @return this for method chaining
    */
   ThisT scanManifestsWith(ExecutorService executorService);
+
+  /**
+   * Perform operations on a particular branch
+   * @param branch which is name of SnapshotRef of type branch.
+   */
+  ThisT toBranch(String branch);
 }

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

@@ -88,6 +88,7 @@ public void accept(String file) {
   private Consumer<String> deleteFunc = defaultDelete;
 
   private ExecutorService workerPool = ThreadPools.getWorkerPool();
+  private String targetBranch = SnapshotRef.MAIN_BRANCH;
 
   protected SnapshotProducer(TableOperations ops) {
     this.ops = ops;
@@ -116,6 +117,38 @@ public ThisT scanManifestsWith(ExecutorService executorService) {
     return self();
   }
 
+  @Override
+  public ThisT toBranch(String branch) {
+    throw new UnsupportedOperationException("Performing operations on a branch is currently not supported");
+  }
+
+  /***
+   * Will be used by snapshot producer operations to create a new ref if an invalid branch is passed
+   * @param branch ref name on which operation is to performed
+   */
+  protected void createNewRef(String branch) {
+    SnapshotRef branchRef = SnapshotRef.branchBuilder(this.current().currentSnapshot().snapshotId()).build();
+    TableMetadata.Builder updatedBuilder = TableMetadata.buildFrom(this.current());
+    updatedBuilder.setRef(branch, branchRef);
+    ops.commit(ops.current(), updatedBuilder.build());
+  }
+
+  /***
+   * A setter for the target branch on which snapshot producer operation should be performed
+   * @param branch to set as target branch
+   */
+  protected void setTargetBranch(String branch) {
+    this.targetBranch = branch;
+  }
+
+  /***
+   * A getter for the target branch on which snapshot producer operation should be performed
+   * @return target branch
+   */
+  protected String getTargetBranch() {
+    return targetBranch;
+  }
+
   protected ExecutorService workerPool() {
     return this.workerPool;
   }
@@ -167,8 +200,7 @@ protected void validate(TableMetadata currentMetadata) {
   @Override
   public Snapshot apply() {
     refresh();
-    Long parentSnapshotId = base.currentSnapshot() != null ?
-        base.currentSnapshot().snapshotId() : null;
+    Long parentSnapshotId = base.ref(targetBranch) != null ? base.ref(targetBranch).snapshotId() : null;
     long sequenceNumber = base.nextSequenceNumber();
 
     // run validations from the child operation
@@ -298,11 +330,11 @@ public void commit() {
             TableMetadata.Builder update = TableMetadata.buildFrom(base);
             if (base.snapshot(newSnapshot.snapshotId()) != null) {
               // this is a rollback operation
-              update.setBranchSnapshot(newSnapshot.snapshotId(), SnapshotRef.MAIN_BRANCH);
+              update.setBranchSnapshot(newSnapshot.snapshotId(), targetBranch);
             } else if (stageOnly) {
               update.addSnapshot(newSnapshot);
             } else {
-              update.setBranchSnapshot(newSnapshot, SnapshotRef.MAIN_BRANCH);
+              update.setBranchSnapshot(newSnapshot, targetBranch);
             }
 
             TableMetadata updated = update.build();

core/src/test/java/org/apache/iceberg/TestFastAppend.java

@@ -478,4 +478,13 @@ public void testIncludedPartitionSummaryLimit() {
     String changedPartitions = table.currentSnapshot().summary().get(SnapshotSummary.CHANGED_PARTITION_COUNT_PROP);
     Assert.assertEquals("Should set changed partition count", "2", changedPartitions);
   }
+
+  @Test(expected = UnsupportedOperationException.class)
+  public void testAppendToBranch() throws UnsupportedOperationException {
+    table.newFastAppend()
+            .appendFile(FILE_A)
+            .commit();
+    table.manageSnapshots().createBranch("ref", table.currentSnapshot().snapshotId()).commit();
+    table.newFastAppend().appendFile(FILE_B).toBranch("ref").commit();
+  }
 }