public CloseableIterator<T> next() {
    MixedFormatSplit split = splitQueue.poll();
    if (split == null) {
      throw new IllegalStateException("next() called, but no more valid splits");
    }

    LOG.info("Fetching data by this split:{}.", split);
    if (split.isMergeOnReadSplit()) {
      return flinkMORDataReader.readData(split.asMergeOnReadSplit().keyedTableScanTask());
    }
    return readerFunction.createDataIterator(split);
  }
This is a mixed-format table(mixed iceberg, mixed-hive) incremental loader.

<p>This loader is used to load data by the merge on read approach first, then by the incremental
pull approach.

<p>Merge on read approach only contain INSERT rows.

<p>Incremental pull approach contains INSERT, DELETE, UPDATE_BEFORE, and UPDATE_AFTER.

<p>Support projection and filter push-down to speed up the loading process.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/MixedIncrementalLoader.java