@Override
  public MixedFormatSplit deserialize(int version, byte[] serialized) throws IOException {
    if (serialized.length == 0) {
      return null;
    }
    try {
      byte flag = serialized[0];
      if (version == VERSION) {
        byte[] content = Bytes.subByte(serialized, 1, serialized.length - 1);
        if (flag == MOR_SPLIT_FLAG) {
          return InstantiationUtil.<MergeOnReadSplit>deserializeObject(
              content, MergeOnReadSplit.class.getClassLoader());
        } else if (flag == SNAPSHOT_SPLIT_FLAG) {
          return InstantiationUtil.<SnapshotSplit>deserializeObject(
              content, SnapshotSplit.class.getClassLoader());
        } else if (flag == CHANGELOG_SPLIT_FLAG) {
          return InstantiationUtil.<ChangelogSplit>deserializeObject(
              content, ChangelogSplit.class.getClassLoader());
        } else {
          throw new IllegalArgumentException(
              String.format(
                  "this flag split %s is unsupported. available: %s, %s, and %s.",
                  flag, SNAPSHOT_SPLIT_FLAG, CHANGELOG_SPLIT_FLAG, MOR_SPLIT_FLAG));
        }
      }
    } catch (ClassNotFoundException e) {
      throw new FlinkRuntimeException("deserialize split failed", e);
    }
    throw new FlinkRuntimeException(
        String.format("this version %s is not supported during deserialize split.", version));
  }
Serializer that serializes and deserializes {@link MixedFormatSplit}.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/hybrid/split/MixedFormatSplitSerializer.java