@Override
  public RecordsWithSplitIds<MixedFormatRecordWithOffset<T>> fetch() throws IOException {
    if (currentReader == null) {
      if (splits.isEmpty()) {
        return new RecordsBySplits<>(Collections.emptyMap(), Collections.emptySet());
      }
      MixedFormatSplit mixedFormatSplit = splits.poll();
      currentReader = openSplitFunction.apply(mixedFormatSplit);
      currentSplitId = mixedFormatSplit.splitId();
    }
    if (currentReader.hasNext()) {
      // Because Iterator#next() doesn't support checked exception,
      // we need to wrap and unwrap the checked IOException with UncheckedIOException
      try {
        return currentReader.next();
      } catch (UncheckedIOException e) {
        throw e.getCause();
      }
    } else {
      return finishSplit();
    }
  }
A hybrid source split reader that could read {@link SnapshotSplit} and {@link ChangelogSplit}.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/hybrid/reader/HybridSplitReader.java