private void extractRowAndEmitRecord(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 (op == Envelope.Operation.CREATE || op == Envelope.Operation.READ) {
            GenericRowData insert = extractAfterRow(value, valueSchema, tableSchema);
            validator.validate(insert, RowKind.INSERT);
            insert.setRowKind(RowKind.INSERT);
            emit(record, insert, tableSchema, out);
        } else if (op == Envelope.Operation.DELETE) {
            GenericRowData delete = extractBeforeRow(value, valueSchema, tableSchema);
            validator.validate(delete, RowKind.DELETE);
            delete.setRowKind(RowKind.DELETE);
            emit(record, delete, tableSchema, out);
        } else {
            if (!appendSource) {
                GenericRowData before = extractBeforeRow(value, valueSchema, tableSchema);
                validator.validate(before, RowKind.UPDATE_BEFORE);
                before.setRowKind(RowKind.UPDATE_BEFORE);
                emit(record, before, tableSchema, out);
            }

            GenericRowData after = extractAfterRow(value, valueSchema, tableSchema);
            validator.validate(after, RowKind.UPDATE_AFTER);
            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/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/debezium/table/RowDataDebeziumDeserializeSchema.java