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

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

            @Override
            public DeserializationSchema<RowData> createRuntimeDecoder(
                    DynamicTableSource.Context context,
                    DataType physicalDataType) {
                final RowType rowType = (RowType) physicalDataType.getLogicalType();
                final TypeInformation<RowData> rowDataTypeInfo = context.createTypeInformation(physicalDataType);
                JsonRowDataDeserializationSchema.Builder builder =
                        JsonRowDataDeserializationSchema.builder(rowType, rowDataTypeInfo);
                return builder
                        .setFailOnMissingField(formatOptions.get(TextFormatOptions.FAIL_ON_MISSING_FIELD))
                        .setIgnoreParseErrors(formatOptions.get(TableFormatOptions.IGNORE_ERRORS))
                        .setTimestampFormat(formatOptions.get(TextFormatOptions.TIMESTAMP_FORMAT))
                        .setCharset(formatOptions.get(TextFormatOptions.CHARSET))
                        .setObjectMapperConfig(formatOptions)
                        .build();
            }

            @Override
            public ChangelogMode getChangelogMode() {
                return ChangelogMode.insertOnly();
            }
        };
    }
Table format factory for providing configured instances of JSON to RowData {@link
SerializationSchema} and {@link DeserializationSchema}.
inlong-sort/sort-formats/format-rowdata/format-rowdata-json/src/main/java/org/apache/inlong/sort/formats/json/JsonFormatFactory.java