public static StringData getDebeziumData(SourceRecord record, TableChange tableSchema,
            RowData rowData) {
        // construct debezium json
        Struct messageStruct = (Struct) record.value();
        GenericRowData data = (GenericRowData) rowData;
        Map<String, Object> field = (Map<String, Object>) data.getField(0);
        Map<String, String> mongoDbType = (Map<String, String>) data.getField(1);

        String database = getMetaData(record, MongoDBEnvelope.NAMESPACE_DATABASE_FIELD);
        String table = getMetaData(record, MongoDBEnvelope.NAMESPACE_COLLECTION_FIELD);
        long ts = (Long) messageStruct.get(FieldName.TIMESTAMP);
        String debeziumOp = getDebeziumOpType(rowData);

        Source source = Source.builder()
                .db(database)
                .table(table)
                .name("mongodb_cdc_source")
                .mysqlType(mongoDbType)
                .sqlType(getSqlType(rowData))
                .pkNames(Collections.singletonList(MONGODB_DEFAULT_PRIMARY_KEY))
                .build();
        DebeziumJson debeziumJson = DebeziumJson.builder()
                .source(source)
                .after(field)
                .tsMs(ts)
                .op(debeziumOp)
                .tableChange(tableSchema)
                .build();
        try {
            return StringData.fromString(OBJECT_MAPPER.writeValueAsString(debeziumJson));
        } catch (Exception e) {
            throw new IllegalStateException("exception occurs when get meta data", e);
        }
    }
get sql type from row data, represents the jdbc data type
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/source/utils/MetaDataUtils.java