@Override
    public RowData deserializeInternal(byte[] bytes) throws Exception {
        String text = new String(bytes, Charset.forName(charset));
        GenericRowData rowData = null;
        try {
            List<Map<String, String>> fieldTexts =
                    splitKv(text, entryDelimiter, kvDelimiter, escapeChar, quoteChar, null,
                            true);

            String[] fieldNames = rowFormatInfo.getFieldNames();
            FormatInfo[] fieldFormatInfos = rowFormatInfo.getFieldFormatInfos();

            rowData = new GenericRowData(fieldFormatInfos.length);
            for (int i = 0; i < fieldFormatInfos.length; i++) {
                String fieldName = fieldNames[i];
                FormatInfo fieldFormatInfo = fieldFormatInfos[i];

                String fieldText = fieldTexts.get(0).get(fieldName);

                Object field = deserializeBasicField(
                        fieldName,
                        fieldFormatInfo,
                        fieldText,
                        nullLiteral, failureHandler);
                rowData.setField(i, converters[i].convert(field));
            }
            return rowData;
        } catch (Throwable t) {
            failureHandler.onParsingMsgFailure(text, new RuntimeException(
                    String.format("Could not properly deserialize kv. Text=[{}].", text), t));
        }
        return null;
    }
Deserialization schema from KV string bytes to Flink Table & SQL internal data structures.

<p>Deserializes a <code>byte[]</code> messages as a Map and converts it to a {@link RowData}.</p>

<p>Failure during deserialization are forwarded as wrapped {@link IOException}.</p>
inlong-sort/sort-formats/format-rowdata/format-rowdata-kv/src/main/java/org/apache/inlong/sort/formats/kv/KvRowDataDeserializationSchema.java