@Override
    public DecodingFormat<DeserializationSchema<RowData>> createDecodingFormat(
            DynamicTableFactory.Context context, ReadableConfig formatOptions) {
        FactoryUtil.validateFactoryOptions(this, formatOptions);
        validateFormatOptions(formatOptions);

        return new DecodingFormat<DeserializationSchema<RowData>>() {

            @Override
            public DeserializationSchema<RowData> createRuntimeDecoder(
                    DynamicTableSource.Context context, DataType producedDataType) {

                InLongMsgTlogCsvDeserializationSchema.Builder builder =
                        new InLongMsgTlogCsvDeserializationSchema.Builder(
                                deserializeRowFormatInfo(formatOptions.get(ROW_FORMAT_INFO)));
                configureDeserializationSchema(formatOptions, builder);
                return builder.build();
            }

            @Override
            public ChangelogMode getChangelogMode() {
                return ChangelogMode.insertOnly();
            }
        };
    }
Table format factory for providing configured instances of InLongMsgTlogCsv-to-row
serializer and deserializer.
inlong-sort/sort-formats/format-rowdata/format-inlongmsg-rowdata-tlogcsv/src/main/java/org/apache/inlong/sort/formats/inlongmsgtlogcsv/InLongMsgTlogCsvFormatFactory.java