public MixedFormatSplit toSourceSplit() {
    if (mixedFormatSplit.isMergeOnReadSplit()) {
      MergeOnReadSplit mergeOnReadSplit = (MergeOnReadSplit) mixedFormatSplit;
      mergeOnReadSplit.updateOffset(
          new Object[] {currentInsertFileOffset, currentInsertRecordOffset});
      return mergeOnReadSplit;
    } else if (mixedFormatSplit.isSnapshotSplit()) {
      SnapshotSplit snapshotSplit = (SnapshotSplit) mixedFormatSplit;
      snapshotSplit.updateOffset(new Object[] {currentInsertFileOffset, currentInsertRecordOffset});
      return snapshotSplit;
    } else if (mixedFormatSplit.isChangelogSplit()) {
      ChangelogSplit changelogSplit = (ChangelogSplit) mixedFormatSplit;
      changelogSplit.updateOffset(
          new Object[] {
            currentInsertFileOffset,
            currentInsertRecordOffset,
            currentDeleteFileOffset,
            currentDeleteRecordOffset
          });
      return changelogSplit;
    }

    throw new FlinkRuntimeException(
        String.format(
            "As of now this source split is unsupported %s, available split are %s, %s, and %s",
            mixedFormatSplit.getClass().getSimpleName(),
            SnapshotSplit.class.getSimpleName(),
            ChangelogSplit.class.getSimpleName(),
            MergeOnReadSplit.class.getSimpleName()));
  }
This is the mutable state for per mixed-format source split.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/hybrid/split/MixedFormatSplitState.java