@Override
    public RowData deserializeInternal(@Nullable byte[] message) throws Exception {
        if (message == null) {
            return null;
        }
        String text = new String(message, Charset.forName(charset));

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

            String[] fieldTexts = splitCsv(text, delimiter, escapeChar, quoteChar);
            if (fieldTexts.length != fieldNames.length) {
                LOG.warn("The number of fields mismatches: expected=[{}], actual=[{}]. Text=[{}].",
                        fieldNames.length, fieldTexts.length, text);
            }
            GenericRowData rowData = new GenericRowData(fieldNames.length);

            for (int i = 0; i < fieldNames.length; ++i) {
                if (i >= fieldTexts.length) {
                    rowData.setField(i, null);
                } else {
                    Object field =
                            TableFormatUtils.deserializeBasicField(
                                    fieldNames[i],
                                    fieldFormatInfos[i],
                                    fieldTexts[i],
                                    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 csv. Text=[{}].", text), t));
        }
        return null;
    }
Creates a CSV deserialization schema for the given {@link TypeInformation} with optional
parameters.
inlong-sort/sort-formats/format-rowdata/format-rowdata-csv/src/main/java/org/apache/inlong/sort/formats/csv/CsvRowDataDeserializationSchema.java