@Override
    public byte[] serializeInternal(RowData rowData) {
        if (rowData == null) {
            return null;
        }

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

            if (rowData.getArity() != fieldFormatInfos.length) {
                LOG.warn("The number of fields mismatches: expected=[{}], actual=[{}]. Row=[{}].",
                        fieldNames.length, rowData.getArity(), rowData);
            }

            String[] fieldTexts = new String[fieldNames.length];

            // the extra fields will be dropped
            for (int i = 0; i < fieldNames.length; i++) {
                if (i >= rowData.getArity()) {
                    fieldTexts[i] = nullLiteral == null ? "" : nullLiteral;
                } else {
                    Object field = rowFieldConverters[i].convert(rowData, i);
                    String fieldText = serializeBasicField(
                            fieldNames[i],
                            fieldFormatInfos[i],
                            field,
                            nullLiteral);

                    fieldTexts[i] = fieldText;
                }
            }

            String text = concatKv(
                    fieldNames,
                    fieldTexts,
                    entryDelimiter,
                    kvDelimiter,
                    escapeChar,
                    quoteChar);

            return text.getBytes(Charset.forName(charset));
        } catch (Throwable t) {
            throw new RuntimeException(
                    String.format("Could not properly serialize kv. Row=[%s]. FormatInfo=[%s].",
                            rowData, rowFormatInfo),
                    t);
        }
    }
Serialization schema that serializes an object of Flink Table & SQL internal data structure into
a KV format bytes.

<p> Serialize the input row into a string in a form of "k1=v1&k2=v2" and converts it into
bytes.</p>

<p>Result <code>byte[]</code> messages can be deserialized using
{@link KvRowDataDeserializationSchema}.</p>
inlong-sort/sort-formats/format-rowdata/format-rowdata-kv/src/main/java/org/apache/inlong/sort/formats/kv/KvRowDataSerializationSchema.java