@Override
  public DataIterator<RowData> createDataIterator(MixedFormatSplit split) {
    if (split.isMergeOnReadSplit()) {
      FlinkKeyedMORDataReader morDataReader =
          new FlinkKeyedMORDataReader(
              io,
              tableSchema,
              readSchema,
              primaryKeySpec,
              nameMapping,
              caseSensitive,
              RowDataUtil::convertConstant,
              reuse);
      return new MergeOnReadDataIterator(
          morDataReader, split.asMergeOnReadSplit().keyedTableScanTask(), io);
    } else if (split.isSnapshotSplit()) {
      FileScanTaskReader<RowData> rowDataReader =
          new FlinkUnkyedDataReader(
              io,
              tableSchema,
              readSchema,
              primaryKeySpec,
              nameMapping,
              caseSensitive,
              RowDataUtil::convertConstant,
              Collections.singleton(split.dataTreeNode()),
              reuse);
      return new DataIterator<>(
          rowDataReader,
          split.asSnapshotSplit().insertTasks(),
          rowData -> Long.MIN_VALUE,
          this::removeMixedFormatMetaColumn);
    } else if (split.isChangelogSplit()) {
      FileScanTaskReader<RowData> rowDataReader =
          new FlinkUnkyedDataReader(
              io,
              wrapFileOffsetColumnMeta(tableSchema),
              wrapFileOffsetColumnMeta(readSchema),
              primaryKeySpec,
              nameMapping,
              caseSensitive,
              RowDataUtil::convertConstant,
              Collections.singleton(split.dataTreeNode()),
              reuse);
      return new ChangeLogDataIterator<>(
          rowDataReader,
          split.asChangelogSplit().insertTasks(),
          split.asChangelogSplit().deleteTasks(),
          this::mixedFormatFileOffset,
          this::removeMixedFormatMetaColumn,
          this::transformRowKind);
    } else {
      throw new IllegalArgumentException(
          String.format(
              "As of now this split %s is not supported.", split.getClass().getSimpleName()));
    }
  }
The index of the mixed-format file offset field in the read schema Refer to {@link
this#wrapFileOffsetColumnMeta}
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/hybrid/reader/RowDataReaderFunction.java