@Override
  public byte[] serialize(MixedFormatSplit split) throws IOException {
    if (split == null) {
      return new byte[0];
    }
    if (split.isMergeOnReadSplit()) {
      MergeOnReadSplit mergeOnReadSplit = (MergeOnReadSplit) split;
      byte[] content = InstantiationUtil.serializeObject(mergeOnReadSplit);
      return Bytes.mergeByte(new byte[] {MOR_SPLIT_FLAG}, content);
    } else if (split.isSnapshotSplit()) {
      SnapshotSplit snapshotSplit = (SnapshotSplit) split;
      byte[] content = InstantiationUtil.serializeObject(snapshotSplit);
      return Bytes.mergeByte(new byte[] {SNAPSHOT_SPLIT_FLAG}, content);
    } else if (split.isChangelogSplit()) {
      ChangelogSplit changelogSplit = (ChangelogSplit) split;
      byte[] content = InstantiationUtil.serializeObject(changelogSplit);
      return Bytes.mergeByte(new byte[] {CHANGELOG_SPLIT_FLAG}, content);
    } else {
      throw new IllegalArgumentException(
          String.format(
              "This mixed-format split is not supported, class %s.",
              split.getClass().getSimpleName()));
    }
  }
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