private Document toDocument(Table table) {
        Document document = Document.create();

        document.set("defaultCharsetName", table.defaultCharsetName());
        document.set("primaryKeyColumnNames", Array.create(table.primaryKeyColumnNames()));

        List<Document> columns =
                table.columns().stream().map(this::toDocument).collect(Collectors.toList());

        document.setArray("columns", Array.create(columns));

        return document;
    }
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/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/debezium/history/FlinkJsonTableChangeSerializer.java