@Override
  public ContinuousEnumerationResult planSplits(IcebergEnumeratorPosition lastPosition) {
    table.refresh();
    if (lastPosition != null) {
      return discoverIncrementalSplits(lastPosition);
    } else {
      return discoverInitialSplits();
    }
  }
@Test
  public void testIncrementalFromSnapshotId() throws Exception {
    appendTwoSnapshots();

    ScanContext scanContext =
        ScanContext.builder()
            .startingStrategy(StreamingStartingStrategy.INCREMENTAL_FROM_SNAPSHOT_ID)
            .startSnapshotId(snapshot2.snapshotId())
            .build();
    ContinuousSplitPlannerImpl splitPlanner =
        new ContinuousSplitPlannerImpl(TABLE_RESOURCE.tableLoader().clone(), scanContext, null);

    ContinuousEnumerationResult initialResult = splitPlanner.planSplits(null);
    assertThat(initialResult.fromPosition()).isNull();
    // For inclusive behavior of snapshot2, the initial result should point to snapshot1 (as
    // snapshot2's parent)
    assertThat(initialResult.toPosition().snapshotId().longValue())
        .isEqualTo(snapshot1.snapshotId());
    assertThat(initialResult.toPosition().snapshotTimestampMs().longValue())
        .isEqualTo(snapshot1.timestampMillis());
    assertThat(initialResult.splits()).isEmpty();

    ContinuousEnumerationResult secondResult = splitPlanner.planSplits(initialResult.toPosition());
    assertThat(secondResult.fromPosition().snapshotId().longValue())
        .isEqualTo(snapshot1.snapshotId());
    assertThat(secondResult.fromPosition().snapshotTimestampMs().longValue())
        .isEqualTo(snapshot1.timestampMillis());
    assertThat(secondResult.toPosition().snapshotId().longValue())
        .isEqualTo(snapshot2.snapshotId());
    assertThat(secondResult.toPosition().snapshotTimestampMs().longValue())
        .isEqualTo(snapshot2.timestampMillis());
    IcebergSourceSplit split = Iterables.getOnlyElement(secondResult.splits());
    assertThat(split.task().files()).hasSize(1);
    Set<String> discoveredFiles =
        split.task().files().stream()
            .map(fileScanTask -> fileScanTask.file().path().toString())
            .collect(Collectors.toSet());
    // should  discover dataFile2 appended in snapshot2
    Set<String> expectedFiles = ImmutableSet.of(dataFile2.path().toString());
    assertThat(discoveredFiles).containsExactlyElementsOf(expectedFiles);

    IcebergEnumeratorPosition lastPosition = secondResult.toPosition();
    for (int i = 0; i < 3; ++i) {
      lastPosition = verifyOneCycle(splitPlanner, lastPosition).lastPosition;
    }
  }