private MixedFormatSourceEnumState deserializeV1(byte[] serialized) throws IOException {
    DataInputDeserializer in = new DataInputDeserializer(serialized);

    MixedFormatEnumeratorOffset enumeratorOffset = null;
    if (in.readBoolean()) {
      int version = in.readInt();
      byte[] positionBytes = new byte[in.readInt()];
      in.read(positionBytes);
      enumeratorOffset = offsetSerializer.deserialize(version, positionBytes);
    }

    int splitSerializerVersion = in.readInt();
    int splitCount = in.readInt();
    Collection<MixedFormatSplitState> pendingSplits = Lists.newArrayListWithCapacity(splitCount);
    for (int i = 0; i < splitCount; ++i) {
      byte[] splitBytes = new byte[in.readInt()];
      in.read(splitBytes);
      MixedFormatSplit split = splitSerializer.deserialize(splitSerializerVersion, splitBytes);
      pendingSplits.add(new MixedFormatSplitState(split));
    }

    long[] shuffleSplitRelation = null;
    if (in.readBoolean()) {
      int length = in.readInt();
      shuffleSplitRelation = new long[length];
      for (int i = 0; i < length; i++) {
        shuffleSplitRelation[i] = in.readLong();
      }
    }

    TemporalJoinSplits temporalJoinSplits = null;
    if (in.readBoolean()) {
      byte[] bytes = new byte[in.readInt()];
      in.read(bytes);
      try {
        temporalJoinSplits =
            InstantiationUtil.deserializeObject(bytes, TemporalJoinSplits.class.getClassLoader());
      } catch (ClassNotFoundException e) {
        throw new RuntimeException("deserialize FirstSplit error", e);
      }
    }

    return new MixedFormatSourceEnumState(
        pendingSplits, enumeratorOffset, shuffleSplitRelation, temporalJoinSplits);
  }
Serializer that serializes and deserializes mixed-format enumerator {@link
MixedFormatSourceEnumState}.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/hybrid/enumerator/MixedFormatSourceEnumStateSerializer.java