private ContinuousEnumerationResult planSplits() {
    ContinuousEnumerationResult result = doPlanSplits();
    if (dimTable && temporalJoinSplits == null) {
      temporalJoinSplits = new TemporalJoinSplits(result.splits(), context.metricGroup());
      // the first SourceEvent may be faster than plan splits
      if (result.isEmpty() && sourceEventBeforeFirstPlan) {
        notifyReaders();
      }
    }
    return result;
  }
snapshotId for the last enumerated snapshot. next incremental enumeration should be based off
this as the starting position.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/hybrid/enumerator/MixedFormatSourceEnumerator.java