@Override
    public EncodingFormat<SerializationSchema<RowData>> createEncodingFormat(
            DynamicTableFactory.Context context,
            ReadableConfig formatOptions) {

        FactoryUtil.validateFactoryOptions(this, formatOptions);
        KvCommons.validateFormatOptions(formatOptions);
        return new EncodingFormat<SerializationSchema<RowData>>() {

            @Override
            public SerializationSchema<RowData> createRuntimeEncoder(
                    DynamicTableSink.Context context,
                    DataType dataType) {

                final RowFormatInfo projectedRowFormatInfo = TableFormatForRowDataUtils.projectRowFormatInfo(
                        deserializeRowFormatInfo(formatOptions.get(ROW_FORMAT_INFO)),
                        dataType);
                KvRowDataSerializationSchema.Builder schemaBuilder =
                        new KvRowDataSerializationSchema.Builder(projectedRowFormatInfo);
                configureSerializationSchema(formatOptions, schemaBuilder);
                return schemaBuilder.build();
            }

            @Override
            public ChangelogMode getChangelogMode() {
                return ChangelogMode.insertOnly();
            }
        };
    }
Table format factory for providing configured instances of KV-to-RowData serializer and
deserializer.
inlong-sort/sort-formats/format-rowdata/format-rowdata-kv/src/main/java/org/apache/inlong/sort/formats/kv/KvFormatFactory.java