@Override
  public void start() {
    if (keyedTable == null) {
      keyedTable = loadMixedTable(loader).asKeyedTable();
    }
    if (enumeratorPosition.get() == null
        && SCAN_STARTUP_MODE_LATEST.equalsIgnoreCase(scanContext.scanStartupMode())) {
      keyedTable.refresh();
      Snapshot snapshot = keyedTable.changeTable().currentSnapshot();
      long snapshotId = snapshot == null ? EARLIEST_SNAPSHOT_ID : snapshot.snapshotId();
      enumeratorPosition.set(MixedFormatEnumeratorOffset.of(snapshotId, null));
      LOG.info(
          "{} is {}, the current snapshot id of the change table {}  is {}.",
          SCAN_STARTUP_MODE.key(),
          SCAN_STARTUP_MODE_LATEST,
          keyedTable.id(),
          snapshotId);
    }
    if (snapshotDiscoveryIntervalMs > 0) {
      LOG.info(
          "Starting the MixedFormatSourceEnumerator with mixed-format table {} snapshot discovery interval of {} ms.",
          keyedTable,
          snapshotDiscoveryIntervalMs);
      context.callAsync(
          this::planSplits, this::handleResultOfSplits, 0, snapshotDiscoveryIntervalMs);
    }
  }
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