@Override
  public CloseableIterator<RecordsWithSplitIds<MixedFormatRecordWithOffset<T>>> apply(
      MixedFormatSplit split) {
    DataIterator<T> inputIterator = createDataIterator(split);
    if (inputIterator instanceof MergeOnReadDataIterator) {
      inputIterator.seek(0, split.asMergeOnReadSplit().recordOffset());
    } else if (inputIterator instanceof ChangeLogDataIterator) {
      ChangeLogDataIterator<T> changelogInputIterator = (ChangeLogDataIterator<T>) inputIterator;
      ChangelogSplit changelogSplit = split.asChangelogSplit();
      changelogInputIterator.seek(
          changelogSplit.insertFileOffset(),
          changelogSplit.deleteFileOffset(),
          changelogSplit.insertRecordOffset(),
          changelogSplit.deleteRecordOffset());
    } else {
      inputIterator.seek(
          split.asSnapshotSplit().insertFileOffset(), split.asSnapshotSplit().insertRecordOffset());
    }
    return batcher.batch(split.splitId(), inputIterator);
  }
A {@link ReaderFunction} implementation that uses {@link DataIterator}.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/hybrid/reader/DataIteratorReaderFunction.java