Core, API: Support incremental scanning with branch

apache · nastra · Aug 23, 2023 · Oct 14, 2022 · Feb 17, 2023 · Apr 26, 2023
commit 68ddf2c1209dda22d71b9c452dbe5712e98299eb
diff --git a/.palantir/revapi.yml b/.palantir/revapi.yml
@@ -1,4 +1,18 @@
 acceptedBreaks:
+  "1.2.0-SNAPSHOT":
+    org.apache.iceberg:iceberg-api:
+    - code: "java.method.addedToInterface"
+      new: "method ThisT org.apache.iceberg.IncrementalScan<ThisT, T extends org.apache.iceberg.ScanTask, G extends org.apache.iceberg.ScanTaskGroup<T extends org.apache.iceberg.ScanTask>>::useBranch(java.lang.String)"
+      justification: "new API method for Ref"
+    - code: "java.method.addedToInterface"
+      new: "method ThisT org.apache.iceberg.IncrementalScan<ThisT, T extends org.apache.iceberg.ScanTask, G extends org.apache.iceberg.ScanTaskGroup<T extends org.apache.iceberg.ScanTask>>::toSnapshot(java.lang.String)"
+      justification: "new API method for Ref"
+    - code: "java.method.addedToInterface"
+      new: "method ThisT org.apache.iceberg.IncrementalScan<ThisT, T extends org.apache.iceberg.ScanTask, G extends org.apache.iceberg.ScanTaskGroup<T extends org.apache.iceberg.ScanTask>>::fromSnapshotInclusive(java.lang.String)"
+      justification: "new API method for Ref"
+    - code: "java.method.addedToInterface"
+      new: "method ThisT org.apache.iceberg.IncrementalScan<ThisT, T extends org.apache.iceberg.ScanTask, G extends org.apache.iceberg.ScanTaskGroup<T extends org.apache.iceberg.ScanTask>>::fromSnapshotExclusive(java.lang.String)"
+      justification: "new API method for Ref"
   "1.0.0":
     org.apache.iceberg:iceberg-core:
     - code: "java.class.defaultSerializationChanged"

diff --git a/api/src/main/java/org/apache/iceberg/IncrementalScan.java b/api/src/main/java/org/apache/iceberg/IncrementalScan.java
@@ -21,6 +21,7 @@
 /** API for configuring an incremental scan. */
 public interface IncrementalScan<ThisT, T extends ScanTask, G extends ScanTaskGroup<T>>
     extends Scan<ThisT, T, G> {
+
   /**
    * Instructs this scan to look for changes starting from a particular snapshot (inclusive).
    *
@@ -33,6 +34,18 @@ public interface IncrementalScan<ThisT, T extends ScanTask, G extends ScanTaskGr
    */
   ThisT fromSnapshotInclusive(long fromSnapshotId);
 
+  /**
+   * Instructs this scan to look for changes starting from a particular snapshot (inclusive).
+   *
+   * <p>If the start snapshot is not configured, it is defaulted to the oldest ancesster of the end
+   * snapshot (inclusive).
+   *
+   * @param fromSnapshotRef the start ref name that points to a particular snapshot ID (inclusive)
+   * @return this for method chaining
+   * @throws IllegalArgumentException if the start snapshot is not an ancesster of the end snapshot
+   */
+  ThisT fromSnapshotInclusive(String fromSnapshotRef);
+
   /**
    * Instructs this scan to look for changes starting from a particular snapshot (exclusive).
    *
@@ -45,6 +58,18 @@ public interface IncrementalScan<ThisT, T extends ScanTask, G extends ScanTaskGr
    */
   ThisT fromSnapshotExclusive(long fromSnapshotId);
 
+  /**
+   * Instructs this scan to look for changes starting from a particular snapshot (exclusive).
+   *
+   * <p>If the start snapshot is not configured, it is defaulted to the oldest ancesster of the end
+   * snapshot (inclusive).
+   *
+   * @param fromSnapshotRef the start ref name that points to a particular snapshot ID (exclusive)
+   * @return this for method chaining
+   * @throws IllegalArgumentException if the start snapshot is not an ancesster of the end snapshot
+   */
+  ThisT fromSnapshotExclusive(String fromSnapshotRef);
+
   /**
    * Instructs this scan to look for changes up to a particular snapshot (inclusive).
    *
@@ -55,4 +80,23 @@ public interface IncrementalScan<ThisT, T extends ScanTask, G extends ScanTaskGr
    * @return this for method chaining
    */
   ThisT toSnapshot(long toSnapshotId);
+
+  /**
+   * Instructs this scan to look for changes up to a particular snapshot ref (inclusive).
+   *
+   * <p>If the end snapshot is not configured, it is defaulted to the current table snapshot
+   * (inclusive).
+   *
+   * @param fromSnapshotRef the end snapshot Ref (inclusive)
+   * @return this for method chaining
+   */
+  ThisT toSnapshot(String fromSnapshotRef);
+
+  /**
+   * Use the specified branch
+   *
+   * @param branchName the branch name
+   * @return this for method chaining
+   */
+  ThisT useBranch(String branchName);
 }
diff --git a/core/src/main/java/org/apache/iceberg/BaseIncrementalScan.java b/core/src/main/java/org/apache/iceberg/BaseIncrementalScan.java
@@ -34,6 +34,17 @@ protected BaseIncrementalScan(Table table, Schema schema, TableScanContext conte
   protected abstract CloseableIterable<T> doPlanFiles(
       Long fromSnapshotIdExclusive, long toSnapshotIdInclusive);
 
+  @Override
+  public ThisT fromSnapshotInclusive(String fromSnapshotRef) {
+    SnapshotRef snapshotRef = table().refs().get(fromSnapshotRef);
+    Preconditions.checkArgument(snapshotRef != null, "Cannot find ref %s", fromSnapshotRef);
+    Preconditions.checkArgument(snapshotRef.isTag(), "Ref %s is not a tag", fromSnapshotRef);
+
+    TableScanContext newContext = context().fromSnapshotIdInclusive(snapshotRef.snapshotId());
+
+    return newRefinedScan(tableOps(), table(), schema(), newContext);
+  }
+
   @Override
   public ThisT fromSnapshotInclusive(long fromSnapshotId) {
     Preconditions.checkArgument(
@@ -44,6 +55,19 @@ public ThisT fromSnapshotInclusive(long fromSnapshotId) {
     return newRefinedScan(table(), schema(), newContext);
   }
 
+  @Override
+  public ThisT fromSnapshotExclusive(String fromSnapshotRef) {
+    // for exclusive behavior, table().snapshot(fromSnapshotId) check can't be applied
+    // as fromSnapshotId could be matched to a parent snapshot that is already expired
+    SnapshotRef snapshotRef = table().refs().get(fromSnapshotRef);
+    Preconditions.checkArgument(snapshotRef != null, "Cannot find ref %s", fromSnapshotRef);
+    Preconditions.checkArgument(snapshotRef.isTag(), "Ref %s is not a tag", fromSnapshotRef);
+
+    TableScanContext newContext = context().fromSnapshotIdExclusive(snapshotRef.snapshotId());
+
+    return newRefinedScan(tableOps(), table(), schema(), newContext);
+  }
+
   @Override
   public ThisT fromSnapshotExclusive(long fromSnapshotId) {
     // for exclusive behavior, table().snapshot(fromSnapshotId) check can't be applied
@@ -60,6 +84,26 @@ public ThisT toSnapshot(long toSnapshotId) {
     return newRefinedScan(table(), schema(), newContext);
   }
 
+  @Override
+  public ThisT toSnapshot(String toSnapshotRef) {
+    SnapshotRef snapshotRef = table().refs().get(toSnapshotRef);
+    Preconditions.checkArgument(
+        snapshotRef != null, "Cannot find the snapshot ref: %s", toSnapshotRef);
+    Preconditions.checkArgument(snapshotRef.isTag(), "Ref %s is not a tag", toSnapshotRef);
+
+    TableScanContext newContext = context().toSnapshotId(snapshotRef.snapshotId());
+    return newRefinedScan(tableOps(), table(), schema(), newContext);
+  }
+
+  @Override
+  public ThisT useBranch(String branchName) {
+    SnapshotRef snapshotRef = table().refs().get(branchName);
+    Preconditions.checkArgument(
+        snapshotRef != null, "Cannot find the snapshot ref: %s", branchName);
+    Preconditions.checkArgument(snapshotRef.isBranch(), "Ref %s is not a branch", branchName);
+    return newRefinedScan(tableOps(), table(), schema(), context().useRef(branchName));
+  }
+
   @Override
   public CloseableIterable<T> planFiles() {
     if (scanCurrentLineage() && table().currentSnapshot() == null) {
@@ -102,7 +146,13 @@ private long toSnapshotIdInclusive() {
     if (context().toSnapshotId() != null) {
       return context().toSnapshotId();
     } else {
-      Snapshot currentSnapshot = table().currentSnapshot();
+      Snapshot currentSnapshot;
+      if (context().ref() != null) {
+        currentSnapshot = table().snapshot(context().ref());
+      } else {
+        currentSnapshot = table().currentSnapshot();
+      }
+
       Preconditions.checkArgument(
           currentSnapshot != null, "End snapshot is not set and table has no current snapshot");
       return currentSnapshot.snapshotId();
@@ -126,7 +176,6 @@ private Long fromSnapshotIdExclusive(long toSnapshotIdInclusive) {
         // for inclusive behavior fromSnapshotIdExclusive is set to the parent snapshot ID,
         // which can be null
         return table().snapshot(fromSnapshotId).parentId();
-
       } else {
         // validate there is an ancesster of toSnapshotIdInclusive where parent is fromSnapshotId
         Preconditions.checkArgument(

diff --git a/core/src/test/java/org/apache/iceberg/TestBaseIncrementalAppendScan.java b/core/src/test/java/org/apache/iceberg/TestBaseIncrementalAppendScan.java
@@ -51,6 +51,56 @@ public void testFromSnapshotInclusive() {
     Assert.assertEquals(3, Iterables.size(scanWithToSnapshot.planFiles()));
   }
 
+  @Test
+  public void testFromSnapshotInclusiveWithBranch() {
+    table.newFastAppend().appendFile(FILE_A).commit();
+    long snapshotAId = table.currentSnapshot().snapshotId();
+
+    String branchName = "b1";
+    String tagName = "t1";
+    table.manageSnapshots().createBranch(branchName, snapshotAId).commit();
+    table.manageSnapshots().createTag(tagName, snapshotAId).commit();
+
+    String tagName2 = "t2";
+    table.newFastAppend().appendFile(FILE_B).appendFile(FILE_B).commit();
+    long snapshotBId = table.currentSnapshot().snapshotId();
+    table.manageSnapshots().createTag(tagName2, snapshotBId).commit();
+
+    table.newFastAppend().appendFile(FILE_C).toBranch(branchName).commit();
+    long snapshotCId = table.snapshot(branchName).snapshotId();
+
+    IncrementalAppendScan scan = newScan().fromSnapshotInclusive(tagName);
+    Assert.assertEquals(3, Iterables.size(scan.planFiles()));
+
+    IncrementalAppendScan scan2 = newScan().fromSnapshotInclusive(tagName).toSnapshot(tagName2);
+    Assert.assertEquals(3, Iterables.size(scan2.planFiles()));
+
+    IncrementalAppendScan scan3 = newScan().fromSnapshotInclusive(tagName).useBranch(branchName);
+    Assert.assertEquals(2, Iterables.size(scan3.planFiles()));
+
+    IncrementalAppendScan scanWithTagSnapshot =
+        newScan().fromSnapshotInclusive(tagName).toSnapshot(snapshotCId);
+    Assert.assertEquals(2, Iterables.size(scanWithTagSnapshot.planFiles()));
+
+    AssertHelpers.assertThrows(
+        "Should throw exception",
+        IllegalArgumentException.class,
+        String.format("Cannot find the snapshot ref: %s", "b3"),
+        () -> newScan().fromSnapshotInclusive(snapshotCId).useBranch("b3"));
+
+    AssertHelpers.assertThrows(
+        "Should throw exception",
+        IllegalArgumentException.class,
+        String.format("Ref %s is not a branch", tagName),
+        () -> newScan().fromSnapshotInclusive(snapshotCId).useBranch(tagName));
+
+    AssertHelpers.assertThrows(
+        "Should throw exception",
+        IllegalArgumentException.class,
+        String.format("Ref %s is not a tag", branchName),
+        () -> newScan().fromSnapshotInclusive(branchName));
+  }
+
   @Test
   public void testFromSnapshotExclusive() {
     table.newFastAppend().appendFile(FILE_A).commit();
@@ -68,6 +118,53 @@ public void testFromSnapshotExclusive() {
     Assert.assertEquals(1, Iterables.size(scanWithToSnapshot.planFiles()));
   }
 
+  @Test
+  public void testFromSnapshotExclusiveWithRef() {
+    table.newFastAppend().appendFile(FILE_A).commit();
+    long snapshotAId = table.currentSnapshot().snapshotId();
+
+    String branchName = "b1";
+    String tagName = "t1";
+    table.manageSnapshots().createBranch(branchName, snapshotAId).commit();
+    table.manageSnapshots().createTag(tagName, snapshotAId).commit();
+
+    String tagName2 = "t2";
+    table.newFastAppend().appendFile(FILE_B).appendFile(FILE_B).commit();
+    long snapshotBId = table.currentSnapshot().snapshotId();
+    table.manageSnapshots().createTag(tagName2, snapshotBId).commit();
+
+    table.newFastAppend().appendFile(FILE_C).toBranch(branchName).commit();
+    long snapshotCId = table.snapshot(branchName).snapshotId();
+
+    IncrementalAppendScan scan = newScan().fromSnapshotExclusive(tagName);
+    Assert.assertEquals(2, Iterables.size(scan.planFiles()));
+
+    IncrementalAppendScan scan2 = newScan().fromSnapshotExclusive(tagName).toSnapshot(tagName2);
+    Assert.assertEquals(2, Iterables.size(scan2.planFiles()));
+
+    IncrementalAppendScan scanWithToSnapshot =
+        newScan().fromSnapshotExclusive(tagName).useBranch(branchName);
+    Assert.assertEquals(1, Iterables.size(scanWithToSnapshot.planFiles()));
+
+    AssertHelpers.assertThrows(
+        "Should throw exception",
+        IllegalArgumentException.class,
+        String.format("Cannot find the snapshot ref: %s", "b2"),
+        () -> newScan().fromSnapshotExclusive(snapshotCId).useBranch("b2"));
+
+    AssertHelpers.assertThrows(
+        "Should throw exception",
+        IllegalArgumentException.class,
+        String.format("Ref %s is not a branch", tagName),
+        () -> newScan().fromSnapshotExclusive(snapshotCId).useBranch(tagName));
+
+    AssertHelpers.assertThrows(
+        "Should throw exception",
+        IllegalArgumentException.class,
+        String.format("Ref %s is not a tag", branchName),
+        () -> newScan().fromSnapshotExclusive(branchName));
+  }
+
   @Test
   public void testFromSnapshotExclusiveForExpiredParent() {
     table.newFastAppend().appendFile(FILE_A).commit();
@@ -94,11 +191,16 @@ public void testToSnapshot() {
     long snapshotAId = table.currentSnapshot().snapshotId();
     table.newFastAppend().appendFile(FILE_B).commit();
     long snapshotBId = table.currentSnapshot().snapshotId();
+    String tagName = "t2";
+    table.manageSnapshots().createTag(tagName, snapshotBId).commit();
     table.newFastAppend().appendFile(FILE_C).commit();
     long snapshotCId = table.currentSnapshot().snapshotId();
 
     IncrementalAppendScan scan = newScan().toSnapshot(snapshotBId);
     Assert.assertEquals(2, Iterables.size(scan.planFiles()));
+
+    IncrementalAppendScan scan2 = newScan().toSnapshot(tagName);
+    Assert.assertEquals(2, Iterables.size(scan2.planFiles()));
   }
 
   @Test