@Override
  public ContinuousEnumerationResult planSplits(IcebergEnumeratorPosition lastPosition) {
    table.refresh();
    if (lastPosition != null) {
      return discoverIncrementalSplits(lastPosition);
    } else {
      return discoverInitialSplits();
    }
  }
@Test
  public void testMaxPlanningSnapshotCount() throws Exception {
    appendTwoSnapshots();
    // append 3 more snapshots
    for (int i = 2; i < 5; ++i) {
      appendSnapshot(i, 2);
    }

    ScanContext scanContext =
        ScanContext.builder()
            .startingStrategy(StreamingStartingStrategy.INCREMENTAL_FROM_EARLIEST_SNAPSHOT)
            // limit to 1 snapshot per discovery
            .maxPlanningSnapshotCount(1)
            .build();
    ContinuousSplitPlannerImpl splitPlanner =
        new ContinuousSplitPlannerImpl(TABLE_RESOURCE.tableLoader().clone(), scanContext, null);

    ContinuousEnumerationResult initialResult = splitPlanner.planSplits(null);
    assertThat(initialResult.fromPosition()).isNull();
    // For inclusive behavior, the initial result should point to snapshot1's parent,
    // which leads to null snapshotId and snapshotTimestampMs.
    assertThat(initialResult.toPosition().snapshotId()).isNull();
    assertThat(initialResult.toPosition().snapshotTimestampMs()).isNull();
    assertThat(initialResult.splits()).isEmpty();

    ContinuousEnumerationResult secondResult = splitPlanner.planSplits(initialResult.toPosition());
    // should discover dataFile1 appended in snapshot1
    verifyMaxPlanningSnapshotCountResult(
        secondResult, null, snapshot1, ImmutableSet.of(dataFile1.path().toString()));

    ContinuousEnumerationResult thirdResult = splitPlanner.planSplits(secondResult.toPosition());
    // should discover dataFile2 appended in snapshot2
    verifyMaxPlanningSnapshotCountResult(
        thirdResult, snapshot1, snapshot2, ImmutableSet.of(dataFile2.path().toString()));
  }