private SnapshotPendingSplitsState deserializeSnapshotPendingSplitsState(int version,
            int splitVersion, DataInputDeserializer in) throws IOException {
        List<TableId> alreadyProcessedTables = readTableIds(in);
        List<MySqlSnapshotSplit> remainingSplits = readMySqlSnapshotSplits(splitVersion, in);
        Map<String, MySqlSnapshotSplit> assignedSnapshotSplits =
                readAssignedSnapshotSplits(splitVersion, in);
        Map<String, BinlogOffset> finishedOffsets = readFinishedOffsets(splitVersion, in);
        AssignerStatus assignerStatus;
        if (version >= 3) {
            assignerStatus = fromStatusCode(in.readInt());
        } else {
            boolean isAssignerFinished = in.readBoolean();
            if (isAssignerFinished) {
                assignerStatus = AssignerStatus.INITIAL_ASSIGNING_FINISHED;
            } else {
                assignerStatus = AssignerStatus.INITIAL_ASSIGNING;
            }
        }
        List<TableId> remainingTableIds = readTableIds(in);
        boolean isTableIdCaseSensitive = in.readBoolean();
        final List<MySqlSchemalessSnapshotSplit> remainingSchemaLessSplits = new ArrayList<>();
        final Map<String, MySqlSchemalessSnapshotSplit> assignedSchemaLessSnapshotSplits =
                new HashMap<>();
        final Map<TableId, TableChange> tableSchemas = new HashMap<>();
        remainingSplits.forEach(
                split -> {
                    tableSchemas.putAll(split.getTableSchemas());
                    remainingSchemaLessSplits.add(split.toSchemaLessSnapshotSplit());
                });
        assignedSnapshotSplits
                .forEach((key, value) -> {
                    tableSchemas.putAll(value.getTableSchemas());
                    assignedSchemaLessSnapshotSplits.put(
                            key, value.toSchemaLessSnapshotSplit());
                });
        if (version >= 4) {
            tableSchemas.putAll(readTableSchemas(splitVersion, in));
        }
        return new SnapshotPendingSplitsState(
                alreadyProcessedTables,
                remainingSchemaLessSplits,
                assignedSchemaLessSnapshotSplits,
                finishedOffsets,
                assignerStatus,
                remainingTableIds,
                isTableIdCaseSensitive,
                true, tableSchemas);
    }
The {@link SimpleVersionedSerializer Serializer} for the {@link PendingSplitsState} of MySQL CDC
source.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/state/PendingSplitsStateSerializer.java