private static Table fromDocument(TableId id, Document document) {
        TableEditor editor =
                Table.editor()
                        .tableId(id)
                        .setDefaultCharsetName(document.getString("defaultCharsetName"));

        document.getArray("columns")
                .streamValues()
                .map(Value::asDocument)
                .map(
                        v -> {
                            ColumnEditor columnEditor =
                                    Column.editor()
                                            .name(v.getString("name"))
                                            .jdbcType(v.getInteger("jdbcType"));

                            Integer nativeType = v.getInteger("nativeType");
                            if (nativeType != null) {
                                columnEditor.nativeType(nativeType);
                            }

                            columnEditor
                                    .type(v.getString("typeName"), v.getString("typeExpression"))
                                    .charsetName(v.getString("charsetName"));

                            Integer length = v.getInteger("length");
                            if (length != null) {
                                columnEditor.length(length);
                            }

                            Integer scale = v.getInteger("scale");
                            if (scale != null) {
                                columnEditor.scale(scale);
                            }

                            columnEditor
                                    .position(v.getInteger("position"))
                                    .optional(v.getBoolean("optional"))
                                    .autoIncremented(v.getBoolean("autoIncremented"))
                                    .generated(v.getBoolean("generated"));

                            // BEGIN FLINK MODIFICATION
                            Array enumValues = v.getArray("enumValues");
                            if (enumValues != null && !enumValues.isEmpty()) {
                                columnEditor.enumValues(
                                        enumValues
                                                .streamValues()
                                                .map(Value::asString)
                                                .collect(Collectors.toList()));
                            }
                            // END FLINK MODIFICATION

                            return columnEditor.create();
                        })
                .forEach(editor::addColumn);

        editor.setPrimaryKeyNames(
                document.getArray("primaryKeyColumnNames")
                        .streamValues()
                        .map(Value::asString)
                        .collect(Collectors.toList()));

        return editor.create();
    }
The serializer responsible for converting of {@link TableChanges} into a JSON format. Copied from
io.debezium.relational.history.JsonTableChangeSerializer, but add serialization/deserialization
for column's enumValues
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/debezium/history/FlinkJsonTableChangeSerializer.java