Core, API: Append to branch Impl

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 SanshotRef of type branch.
+   */
+  ThisT toBranch(String branch);
 }

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

@@ -683,7 +683,7 @@ protected Map<String, String> summary() {
 
   @Override
   public List<ManifestFile> apply(TableMetadata base) {
-    Snapshot current = base.currentSnapshot();
+    Snapshot current = toBranch != null ? base.snapshot(base.ref(toBranch).snapshotId()) : base.currentSnapshot();
 
     // filter any existing manifests
     List<ManifestFile> filtered = filterManager.filterManifests(

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

@@ -89,6 +89,8 @@ public void accept(String file) {
 
   private ExecutorService workerPool = ThreadPools.getWorkerPool();
 
+  protected String toBranch = null;
+
   protected SnapshotProducer(TableOperations ops) {
     this.ops = ops;
     this.base = ops.current();
@@ -116,6 +118,19 @@ public ThisT scanManifestsWith(ExecutorService executorService) {
     return self();
   }
 
+  @Override
+  public ThisT toBranch(String branch){
+      Preconditions.checkArgument(branch != null,"branch cannot be null");
+      if (ops.current().ref(branch) == null) {
+        SnapshotRef branchRef = SnapshotRef.branchBuilder(ops.current().currentSnapshot().snapshotId()).build();
+        TableMetadata.Builder updatedBuilder = TableMetadata.buildFrom(base);
+        updatedBuilder.setRef(branch, branchRef).build();
+      }
+      Preconditions.checkArgument(ops.current().ref(branch).type() != SnapshotRefType.BRANCH, "%s is not a ref to type branch", branch);
+      this.toBranch = branch;
+      return self();
+  }
+
   protected ExecutorService workerPool() {
     return this.workerPool;
   }
@@ -167,8 +182,10 @@ protected void validate(TableMetadata currentMetadata) {
   @Override
   public Snapshot apply() {
     refresh();
-    Long parentSnapshotId = base.currentSnapshot() != null ?
-        base.currentSnapshot().snapshotId() : null;
+    Long parentSnapshotId = base.currentSnapshot() != null ? base.currentSnapshot().snapshotId() : null;
+    if(toBranch != null){
+      parentSnapshotId = base.ref(toBranch).snapshotId();
+    }
     long sequenceNumber = base.nextSequenceNumber();
 
     // run validations from the child operation
@@ -282,6 +299,7 @@ protected TableMetadata refresh() {
   @Override
   public void commit() {
     // this is always set to the latest commit attempt's snapshot id.
+    String branch = toBranch == null ? SnapshotRef.MAIN_BRANCH : toBranch;
     AtomicLong newSnapshotId = new AtomicLong(-1L);
     try {
       Tasks.foreach(ops)
@@ -298,11 +316,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(), branch);
             } else if (stageOnly) {
               update.addSnapshot(newSnapshot);
             } else {
-              update.setBranchSnapshot(newSnapshot, SnapshotRef.MAIN_BRANCH);
+              update.setBranchSnapshot(newSnapshot, branch);
             }
 
             TableMetadata updated = update.build();

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

@@ -478,4 +478,37 @@ public void testIncludedPartitionSummaryLimit() {
     String changedPartitions = table.currentSnapshot().summary().get(SnapshotSummary.CHANGED_PARTITION_COUNT_PROP);
     Assert.assertEquals("Should set changed partition count", "2", changedPartitions);
   }
+
+  @Test
+  public void testAppendToBranch() throws UnsupportedOperationException {
+    table.newFastAppend()
+            .appendFile(FILE_A)
+            .commit();
+
+    Long currSnapshot = table.currentSnapshot().snapshotId();
+    table.manageSnapshots().createBranch("ref", table.currentSnapshot().snapshotId()).commit();
+    table.newFastAppend().toBranch("ref").appendFile(FILE_B).commit();
+    Snapshot branch = table.snapshot(table.ops().current().ref("ref").snapshotId());
+    Assert.assertEquals(currSnapshot, branch.parentId());
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testAppendToNullBranch() {
+    table.newFastAppend()
+            .appendFile(FILE_A)
+            .commit();
+
+    table.manageSnapshots().createBranch("ref", table.currentSnapshot().snapshotId()).commit();
+    table.newDelete().toBranch(null).deleteFile(FILE_A);
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testAppendToInValidBranch() {
+    table.newFastAppend()
+            .appendFile(FILE_A)
+            .commit();
+
+    table.manageSnapshots().createBranch("ref", table.currentSnapshot().snapshotId()).commit();
+    table.newDelete().toBranch("newBranch").deleteFile(FILE_A).commit();
+  }
 }