private static StringData getCanalData(SourceRecord record, TableChange tableSchema, GenericRowData rowData) {
        // construct canal json
        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);
        // schemaName
        String schemaName = getMetaData(record, AbstractSourceInfo.SCHEMA_NAME_KEY);
        // opTs
        long opTs = (Long) sourceStruct.get(AbstractSourceInfo.TIMESTAMP_KEY);
        // ts
        long ts = (Long) messageStruct.get(FieldName.TIMESTAMP);
        // actual data
        GenericRowData data = rowData;
        Map<String, Object> field = (Map<String, Object>) data.getField(0);
        List<Map<String, Object>> dataList = new ArrayList<>();
        dataList.add(field);

        CanalJson canalJson = CanalJson.builder()
                .data(dataList)
                .database(databaseName)
                .schema(schemaName)
                .sql("")
                .es(opTs).isDdl(false)
                .pkNames(getPkNames(tableSchema))
                .table(tableName)
                .ts(ts)
                .postGreType(getType(tableSchema))
                .type(getCanalOpType(rowData))
                .incremental(isIncrementalRecord(sourceStruct))
                .sqlType(getSqlType(tableSchema))
                .build();
        try {
            ObjectMapper objectMapper = new ObjectMapper();
            return StringData.fromString(objectMapper.writeValueAsString(canalJson));
        } catch (Exception e) {
            throw new IllegalStateException("exception occurs when get meta data", e);
        }
    }
Generate a canal json message

@param record source record
@param tableSchema table schema
@param rowData row data with canal json or debezium json
@return
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/table/PostgreSQLReadableMetaData.java