@Override
    public void deserialize(SourceRecord record, Collector<RowData> out,
            TableChange tableSchema)
            throws Exception {
        Envelope.Operation op = Envelope.operationFor(record);
        Struct value = (Struct) record.value();
        Schema valueSchema = record.valueSchema();

        if (schemaChange && RecordUtils.isDdlRecord(value)) {
            extractDdlRecord(record, out, tableSchema, value);
            return;
        }

        if (op == Envelope.Operation.CREATE || op == Envelope.Operation.READ) {
            GenericRowData insert = extractAfterRow(value, valueSchema);
            insert.setRowKind(RowKind.INSERT);
            emit(record, insert, tableSchema, out);
        } else if (op == Envelope.Operation.DELETE) {
            GenericRowData delete = extractBeforeRow(value, valueSchema);
            delete.setRowKind(RowKind.DELETE);
            emit(record, delete, tableSchema, out);
        } else {
            if (!appendSource) {
                GenericRowData before = extractBeforeRow(value, valueSchema);
                if (before != null) {
                    before.setRowKind(RowKind.UPDATE_BEFORE);
                    emit(record, before, tableSchema, out);
                }
            }
            GenericRowData after = extractAfterRow(value, valueSchema);
            after.setRowKind(RowKind.UPDATE_AFTER);
            emit(record, after, tableSchema, out);
        }
    }
extract the data with the format provided by debezium

@param fieldValue
@param schemaName
@return the extracted data with schema
inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/debezium/table/RowDataDebeziumDeserializeSchema.java