@Override
    public Array serialize(TableChanges tableChanges) {
        List<Value> values =
                StreamSupport.stream(tableChanges.spliterator(), false)
                        .map(this::toDocument)
                        .map(Value::create)
                        .collect(Collectors.toList());

        return Array.create(values);
    }
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