@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()) {
                    // The absent fields will be filled with null literal
                    fieldTexts[i] = nullLiteral == null ? "" : nullLiteral;
                } else {
                    String fieldText =
                            TableFormatForRowDataUtils.serializeBasicField(
                                    fieldNames[i],
                                    fieldFormatInfos[i],
                                    rowFieldConverters[i].convert(rowData, i),
                                    nullLiteral);
                    fieldTexts[i] = fieldText;
                }
            }

            String result =
                    concatCsv(fieldTexts, delimiter, escapeChar, quoteChar);

            return result.getBytes(Charset.forName(charset));
        } catch (Throwable t) {
            throw new RuntimeException(
                    String.format("Could not properly serialize csv. Row=[%s].", rowData), t);
        }
    }
Creates a {@link CsvRowDataSerializationSchema} expecting the given {@link RowType}.

@param rowFormatInfo logical row format info used to create schema.
inlong-sort/sort-formats/format-rowdata/format-rowdata-csv/src/main/java/org/apache/inlong/sort/formats/csv/CsvRowDataSerializationSchema.java