@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();
            }
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