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

        Source source = Source.builder().db(getMetaData(record, AbstractSourceInfo.DATABASE_NAME_KEY))
                .table(getMetaData(record, AbstractSourceInfo.TABLE_NAME_KEY))
                .name(sourceStruct.getString(AbstractSourceInfo.SERVER_NAME_KEY))
                .sqlType(getSqlType(tableSchema))
                .pkNames(getPkNames(tableSchema))
                .mysqlType(getType(tableSchema))
                .build();
        DebeziumJson debeziumJson = DebeziumJson.builder().source(source)
                .tsMs(sourceStruct.getInt64(AbstractSourceInfo.TIMESTAMP_KEY)).op(getDebeziumOpType(data))
                .dataSourceName(sourceStruct.getString(AbstractSourceInfo.SERVER_NAME_KEY))
                .tableChange(tableSchema).incremental(!isSnapshotRecord(sourceStruct)).build();

        try {
            if (RecordUtils.isDdlRecord(messageStruct)) {
                String sql = (String) field.get(DDL_FIELD_NAME);
                debeziumJson.setDdl(sql);
                debeziumJson.setOperation(generateOperation(sql, getSqlType(tableSchema)));
                debeziumJson.setAfter(new HashMap<>());
            } else {
                debeziumJson.setAfter(field);
            }
            return StringData.fromString(OBJECT_MAPPER.writeValueAsString(debeziumJson));
        } catch (Exception e) {
            throw new IllegalStateException("exception occurs when get meta data {}", e);
        }

    }
get sql type from table schema, represents the jdbc data type

@param tableSchema table schema
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/utils/MetaDataUtils.java