@Override
    protected byte[] serializeInternal(Row row) {
        if (row == null) {
            return null;
        }

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

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

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

            // The extra fields will be dropped.
            for (int i = 0; i < fieldNames.length; ++i) {
                if (i >= row.getArity()) {
                    // The absent fields will be filled with null literal
                    fieldTexts[i] = nullLiteral == null ? "" : nullLiteral;
                } else {
                    String fieldText =
                            serializeBasicField(
                                    fieldNames[i],
                                    fieldFormatInfos[i],
                                    row.getField(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].", row), t);
        }
    }
The literal represented null values, default "".
inlong-sort/sort-formats/format-row/format-csv/src/main/java/org/apache/inlong/sort/formats/csv/CsvSerializationSchema.java