public static StringData getCanalData(SourceRecord record, GenericRowData rowData,
            TableChange tableSchema) {
        Struct messageStruct = (Struct) record.value();
        Struct sourceStruct = messageStruct.getStruct(FieldName.SOURCE);
        // tableName
        String tableName = getMetaData(record, AbstractSourceInfo.TABLE_NAME_KEY);
        // databaseName
        String databaseName = getMetaData(record, AbstractSourceInfo.DATABASE_NAME_KEY);
        // opTs
        long opTs = (Long) sourceStruct.get(AbstractSourceInfo.TIMESTAMP_KEY);
        // actual data
        Map<String, Object> field = (Map<String, Object>) rowData.getField(0);
        List<Map<String, Object>> dataList = new ArrayList<>();

        CanalJson canalJson = CanalJson.builder()
                .database(databaseName)
                .es(opTs).pkNames(getPkNames(tableSchema))
                .mysqlType(getType(tableSchema)).table(tableName)
                .incremental(!isSnapshotRecord(sourceStruct))
                .dataSourceName(sourceStruct.getString(AbstractSourceInfo.SERVER_NAME_KEY))
                .type(getCanalOpType(rowData))
                .sqlType(getSqlType(tableSchema))
                .build();

        try {
            if (RecordUtils.isDdlRecord(messageStruct)) {
                String sql = (String) field.get(DDL_FIELD_NAME);
                canalJson.setSql(sql);
                canalJson.setOperation(generateOperation(sql, getSqlType(tableSchema)));
                canalJson.setDdl(true);
                canalJson.setData(dataList);
            } else {
                canalJson.setDdl(false);
                canalJson.setTs((Long) messageStruct.get(FieldName.TIMESTAMP));
                dataList.add(field);
                canalJson.setData(dataList);
            }
            LOG.debug("canal json: {}", canalJson);
            return StringData.fromString(OBJECT_MAPPER.writeValueAsString(canalJson));
        } 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