private byte[] serializeV1(MixedFormatSourceEnumState enumState) throws IOException {
    DataOutputSerializer out = SERIALIZER_CACHE.get();

    out.writeBoolean(enumState.lastEnumeratedOffset() != null);
    if (enumState.lastEnumeratedOffset() != null) {
      out.writeInt(offsetSerializer.getVersion());
      byte[] positionBytes = offsetSerializer.serialize(enumState.lastEnumeratedOffset());
      out.writeInt(positionBytes.length);
      out.write(positionBytes);
    }

    out.writeInt(splitSerializer.getVersion());
    out.writeInt(enumState.pendingSplits().size());
    for (MixedFormatSplitState splitState : enumState.pendingSplits()) {
      byte[] splitBytes = splitSerializer.serialize(splitState.toSourceSplit());
      out.writeInt(splitBytes.length);
      out.write(splitBytes);
    }

    out.writeBoolean(enumState.shuffleSplitRelation() != null);
    if (enumState.shuffleSplitRelation() != null) {
      long[] shuffleSplitRelation = enumState.shuffleSplitRelation();
      out.writeInt(Objects.requireNonNull(shuffleSplitRelation).length);
      for (long l : shuffleSplitRelation) {
        out.writeLong(l);
      }
    }

    out.writeBoolean(enumState.temporalJoinSplits() != null);
    if (enumState.temporalJoinSplits() != null) {
      byte[] temporalJoinSplits = InstantiationUtil.serializeObject(enumState.temporalJoinSplits());
      out.writeInt(temporalJoinSplits.length);
      out.write(temporalJoinSplits);
    }

    byte[] result = out.getCopyOfBuffer();
    out.clear();
    return result;
  }
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