private GenericRowData extractMongoDBDdlData(Struct value, String keyFiled, String ddlType) {
        Struct documentTo = (Struct) value.get(keyFiled);
        String newDb = documentTo.getString(MongoDBEnvelope.NAMESPACE_DATABASE_FIELD);
        String newColl = documentTo.getString(MongoDBEnvelope.NAMESPACE_COLLECTION_FIELD);
        String renameFormatStr = newDb + "." + newColl;
        Map<String, Object> data = new HashMap<>();
        Map<String, String> dataType = new HashMap<>();
        data.put(ddlType, renameFormatStr);
        dataType.put(ddlType, "VARCHAR");
        GenericRowData ddlRowData = new GenericRowData(2);
        ddlRowData.setField(0, data);
        ddlRowData.setField(1, dataType);
        return ddlRowData;

    }
Validator to validate the row value.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/debezium/table/MongoDBConnectorDeserializationSchema.java