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

    ScanContext scanContext =
        ScanContext.builder()
            .includeColumnStats(true)
            .startingStrategy(StreamingStartingStrategy.TABLE_SCAN_THEN_INCREMENTAL)
            .build();
    ContinuousSplitPlannerImpl splitPlanner =
        new ContinuousSplitPlannerImpl(TABLE_RESOURCE.tableLoader().clone(), scanContext, null);

    ContinuousEnumerationResult initialResult = splitPlanner.planSplits(null);
    assertThat(initialResult.splits()).hasSize(1);
    IcebergSourceSplit split = Iterables.getOnlyElement(initialResult.splits());
    assertThat(split.task().files()).hasSize(2);
    verifyStatCount(split, 3);

    IcebergEnumeratorPosition lastPosition = initialResult.toPosition();
    for (int i = 0; i < 3; ++i) {
      CycleResult result = verifyOneCycle(splitPlanner, lastPosition);
      verifyStatCount(result.split, 3);
      lastPosition = result.lastPosition;
    }
  }