@Override
    public DeserializationSchema<RowData> createRuntimeDecoder(
            DynamicTableSource.Context context, DataType physicalDataType) {
        final List<ReadableMetadata> readableMetadata = metadataKeys.stream()
                .map(k -> Stream.of(ReadableMetadata.values())
                        .filter(rm -> rm.key.equals(k))
                        .findFirst()
                        .orElseThrow(IllegalStateException::new))
                .collect(Collectors.toList());
        final List<DataTypes.Field> metadataFields = readableMetadata.stream()
                .map(m -> DataTypes.FIELD(m.key, m.dataType))
                .collect(Collectors.toList());
        final DataType producedDataType = DataTypeUtils.appendRowFields(physicalDataType, metadataFields);
        final TypeInformation<RowData> producedTypeInfo = context.createTypeInformation(producedDataType);
        return CanalJsonEnhancedDeserializationSchema.builder(physicalDataType, readableMetadata, producedTypeInfo)
                .setDatabase(database)
                .setTable(table)
                .setIgnoreParseErrors(ignoreParseErrors)
                .setTimestampFormat(timestampFormat)
                .build();
    }
{@link DecodingFormat} for Canal using JSON encoding.
different from flink:1.13.5. This support more metadata.
inlong-sort/sort-formats/format-row/format-json-v1.15/src/main/java/org/apache/inlong/sort/formats/json/canal/CanalJsonEnhancedDecodingFormat.java