Flink: Backport add StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE to Flink 1.19 (#12899)

backports #12839

Author: morhidi

flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/source/StreamingStartingStrategy.java

@@ -30,14 +30,21 @@ public enum StreamingStartingStrategy {
   /**
    * Start incremental mode from the latest snapshot inclusive.
    *
-   * <p>If it is an empty map, all future append snapshots should be discovered.
+   * <p>If it is an empty table, all future append snapshots should be discovered.
    */
   INCREMENTAL_FROM_LATEST_SNAPSHOT,
 
+  /**
+   * Start incremental mode from the latest snapshot exclusive.
+   *
+   * <p>If it is an empty table, all future append snapshots should be discovered.
+   */
+  INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE,
+
   /**
    * Start incremental mode from the earliest snapshot inclusive.
    *
-   * <p>If it is an empty map, all future append snapshots should be discovered.
+   * <p>If it is an empty table, all future append snapshots should be discovered.
    */
   INCREMENTAL_FROM_EARLIEST_SNAPSHOT,
 

flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/source/enumerator/ContinuousSplitPlannerImpl.java

@@ -165,7 +165,7 @@ private ContinuousEnumerationResult discoverInitialSplits() {
         "Get starting snapshot id {} based on strategy {}",
         startSnapshot.snapshotId(),
         scanContext.streamingStartingStrategy());
-    List<IcebergSourceSplit> splits;
+    List<IcebergSourceSplit> splits = Collections.emptyList();
     IcebergEnumeratorPosition toPosition;
     if (scanContext.streamingStartingStrategy()
         == StreamingStartingStrategy.TABLE_SCAN_THEN_INCREMENTAL) {
@@ -180,10 +180,17 @@ private ContinuousEnumerationResult discoverInitialSplits() {
       // For TABLE_SCAN_THEN_INCREMENTAL, incremental mode starts exclusive from the startSnapshot
       toPosition =
           IcebergEnumeratorPosition.of(startSnapshot.snapshotId(), startSnapshot.timestampMillis());
+    } else if (scanContext.streamingStartingStrategy()
+        == StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE) {
+      toPosition =
+          IcebergEnumeratorPosition.of(startSnapshot.snapshotId(), startSnapshot.timestampMillis());
+      LOG.info(
+          "Start incremental scan with start snapshot (exclusive): id = {}, timestamp = {}",
+          startSnapshot.snapshotId(),
+          startSnapshot.timestampMillis());
     } else {
       // For all other modes, starting snapshot should be consumed inclusively.
       // Use parentId to achieve the inclusive behavior. It is fine if parentId is null.
-      splits = Collections.emptyList();
       Long parentSnapshotId = startSnapshot.parentId();
       if (parentSnapshotId != null) {
         Snapshot parentSnapshot = table.snapshot(parentSnapshotId);
@@ -216,6 +223,7 @@ static Optional<Snapshot> startSnapshot(Table table, ScanContext scanContext) {
     switch (scanContext.streamingStartingStrategy()) {
       case TABLE_SCAN_THEN_INCREMENTAL:
       case INCREMENTAL_FROM_LATEST_SNAPSHOT:
+      case INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE:
         return Optional.ofNullable(table.currentSnapshot());
       case INCREMENTAL_FROM_EARLIEST_SNAPSHOT:
         return Optional.ofNullable(SnapshotUtil.oldestAncestor(table));

flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/source/enumerator/TestContinuousSplitPlannerImpl.java

@@ -44,6 +44,8 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
 
 public class TestContinuousSplitPlannerImpl {
   @TempDir protected Path temporaryFolder;
@@ -173,13 +175,14 @@ public void testTableScanThenIncrementalWithNonEmptyTable() throws Exception {
     }
   }
 
-  @Test
-  public void testIncrementalFromLatestSnapshotWithEmptyTable() throws Exception {
+  @ParameterizedTest
+  @EnumSource(
+      value = StreamingStartingStrategy.class,
+      names = {"INCREMENTAL_FROM_LATEST_SNAPSHOT", "INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE"})
+  public void testIncrementalFromLatestSnapshotWithEmptyTable(
+      StreamingStartingStrategy startingStrategy) throws Exception {
     ScanContext scanContext =
-        ScanContext.builder()
-            .startingStrategy(StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT)
-            .splitSize(1L)
-            .build();
+        ScanContext.builder().startingStrategy(startingStrategy).splitSize(1L).build();
     ContinuousSplitPlannerImpl splitPlanner =
         new ContinuousSplitPlannerImpl(TABLE_RESOURCE.tableLoader().clone(), scanContext, null);
 
@@ -256,6 +259,44 @@ public void testIncrementalFromLatestSnapshotWithNonEmptyTable() throws Exceptio
     }
   }
 
+  @Test
+  public void testIncrementalFromLatestSnapshotExclusiveWithNonEmptyTable() throws Exception {
+    appendTwoSnapshots();
+
+    ScanContext scanContext =
+        ScanContext.builder()
+            .startingStrategy(StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE)
+            .build();
+    ContinuousSplitPlannerImpl splitPlanner =
+        new ContinuousSplitPlannerImpl(TABLE_RESOURCE.tableLoader().clone(), scanContext, null);
+
+    ContinuousEnumerationResult initialResult = splitPlanner.planSplits(null);
+    assertThat(initialResult.splits()).isEmpty();
+    assertThat(initialResult.fromPosition()).isNull();
+    // For exclusive behavior, the initial result should point to snapshot2
+    assertThat(initialResult.toPosition().snapshotId().longValue())
+        .isEqualTo(snapshot2.snapshotId());
+    assertThat(initialResult.toPosition().snapshotTimestampMs().longValue())
+        .isEqualTo(snapshot2.timestampMillis());
+
+    // Then the next incremental scan shall discover no files
+    ContinuousEnumerationResult secondResult = splitPlanner.planSplits(initialResult.toPosition());
+    assertThat(initialResult.splits()).isEmpty();
+    assertThat(secondResult.fromPosition().snapshotId().longValue())
+        .isEqualTo(snapshot2.snapshotId());
+    assertThat(secondResult.fromPosition().snapshotTimestampMs().longValue())
+        .isEqualTo(snapshot2.timestampMillis());
+    assertThat(secondResult.toPosition().snapshotId().longValue())
+        .isEqualTo(snapshot2.snapshotId());
+    assertThat(secondResult.toPosition().snapshotTimestampMs().longValue())
+        .isEqualTo(snapshot2.timestampMillis());
+
+    IcebergEnumeratorPosition lastPosition = secondResult.toPosition();
+    for (int i = 0; i < 3; ++i) {
+      lastPosition = verifyOneCycle(splitPlanner, lastPosition).lastPosition;
+    }
+  }
+
   @Test
   public void testIncrementalFromEarliestSnapshotWithEmptyTable() throws Exception {
     ScanContext scanContext =

flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/source/enumerator/TestContinuousSplitPlannerImplStartStrategy.java

@@ -37,6 +37,8 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
 
 public class TestContinuousSplitPlannerImplStartStrategy {
   private static final FileFormat FILE_FORMAT = FileFormat.PARQUET;
@@ -88,13 +90,14 @@ public void testTableScanThenIncrementalStrategy() throws IOException {
     assertThat(startSnapshot.snapshotId()).isEqualTo(snapshot3.snapshotId());
   }
 
-  @Test
-  public void testForLatestSnapshotStrategy() throws IOException {
+  @ParameterizedTest
+  @EnumSource(
+      value = StreamingStartingStrategy.class,
+      names = {"INCREMENTAL_FROM_LATEST_SNAPSHOT", "INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE"})
+  public void testForLatestSnapshotStrategyWithEmptyTable(
+      StreamingStartingStrategy startingStrategy) throws IOException {
     ScanContext scanContext =
-        ScanContext.builder()
-            .streaming(true)
-            .startingStrategy(StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT)
-            .build();
+        ScanContext.builder().streaming(true).startingStrategy(startingStrategy).build();
 
     assertThat(ContinuousSplitPlannerImpl.startSnapshot(TABLE_RESOURCE.table(), scanContext))
         .isNotPresent();
@@ -105,6 +108,22 @@ public void testForLatestSnapshotStrategy() throws IOException {
     assertThat(startSnapshot.snapshotId()).isEqualTo(snapshot3.snapshotId());
   }
 
+  @ParameterizedTest
+  @EnumSource(
+      value = StreamingStartingStrategy.class,
+      names = {"INCREMENTAL_FROM_LATEST_SNAPSHOT", "INCREMENTAL_FROM_LATEST_SNAPSHOT_EXCLUSIVE"})
+  public void testForLatestSnapshotStrategyWithNonEmptyTable(
+      StreamingStartingStrategy startingStrategy) throws IOException {
+    appendThreeSnapshots();
+
+    ScanContext scanContext =
+        ScanContext.builder().streaming(true).startingStrategy(startingStrategy).build();
+
+    Snapshot startSnapshot =
+        ContinuousSplitPlannerImpl.startSnapshot(TABLE_RESOURCE.table(), scanContext).get();
+    assertThat(startSnapshot.snapshotId()).isEqualTo(snapshot3.snapshotId());
+  }
+
   @Test
   public void testForEarliestSnapshotStrategy() throws IOException {
     ScanContext scanContext =