@Override
    public TableChanges deserialize(Array array, boolean useCatalogBeforeSchema) {
        TableChanges tableChanges = new TableChanges();

        for (Entry entry : array) {
            TableChange change =
                    fromDocument(entry.getValue().asDocument(), useCatalogBeforeSchema);

            if (change.getType() == TableChangeType.CREATE) {
                tableChanges.create(change.getTable());
            } else if (change.getType() == TableChangeType.ALTER) {
                tableChanges.alter(change.getTable());
            } else if (change.getType() == TableChangeType.DROP) {
                tableChanges.drop(change.getTable());
            }
        }

        return tableChanges;
    }
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