@Override
  protected ContinuousEnumerationResult discoverInitialSplits(List<Expression> filters) {
    Snapshot changeSnapshot = table.changeTable().currentSnapshot();

    List<MixedFormatSplit> mixedFormatSplits =
        FlinkSplitPlanner.mergeOnReadPlan(table, filters, SPLIT_COUNT);

    long changeStartSnapshotId =
        changeSnapshot != null ? changeSnapshot.snapshotId() : EARLIEST_SNAPSHOT_ID;
    if (changeSnapshot == null && CollectionUtils.isEmpty(mixedFormatSplits)) {
      LOG.info("There have no change snapshot, and no base splits in table: {}.", table);
      return ContinuousEnumerationResult.EMPTY;
    }

    return new ContinuousEnumerationResult(
        mixedFormatSplits, null, MixedFormatEnumeratorOffset.of(changeStartSnapshotId, null));
  }
A planner for merge-on-read scanning by {@link this#discoverInitialSplits} and incremental
scanning by {@link this#discoverIncrementalSplits(MixedFormatEnumeratorOffset, List)}.

<p>{@link ContinuousEnumerationResult#splits()} includes the {@link MergeOnReadSplit}s and {@link
ChangelogSplit}s.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/hybrid/enumerator/MergeOnReadIncrementalPlanner.java