public static StringData getCanalData(SourceRecord record, RowData rowData,
            TableChange tableSchema) {
        // construct canal json
        Struct messageStruct = (Struct) record.value();
        Struct sourceStruct = messageStruct.getStruct(Envelope.FieldName.SOURCE);
        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 opTs = (Long) sourceStruct.get(AbstractSourceInfo.TIMESTAMP_KEY);
        long ts = (Long) messageStruct.get(FieldName.TIMESTAMP);

        List<Map<String, Object>> dataList = new ArrayList<>();
        dataList.add(field);
        CanalJson canalJson = CanalJson.builder()
                .data(dataList)
                .database(database)
                .sql("")
                .es(opTs)
                .isDdl(false)
                .pkNames(Collections.singletonList(MONGODB_DEFAULT_PRIMARY_KEY))
                .mysqlType(mongoDbType)
                .table(table)
                .ts(ts)
                .type(getCanalOpType(rowData))
                .sqlType(getSqlType(data))
                .build();
        try {
            return StringData.fromString(OBJECT_MAPPER.writeValueAsString(canalJson));
        } 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