private void doDeserialize(SourceRecord record, Collector<RowData> out, long deserializeStartTime)
            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);
            validator.validate(insert, RowKind.INSERT);
            insert.setRowKind(RowKind.INSERT);
            if (sourceExactlyMetric != null) {
                out = new MetricsCollector<>(out, sourceExactlyMetric);
            }
            emit(record, insert, out);
        } else if (op == Envelope.Operation.DELETE) {
            GenericRowData delete = extractBeforeRow(value, valueSchema);
            validator.validate(delete, RowKind.DELETE);
            delete.setRowKind(RowKind.DELETE);
            if (sourceExactlyMetric != null) {
                out = new MetricsCollector<>(out, sourceExactlyMetric);
            }
            emit(record, delete, out);
        } else {
            if (changelogMode == DebeziumChangelogMode.ALL) {
                GenericRowData before = extractBeforeRow(value, valueSchema);
                validator.validate(before, RowKind.UPDATE_BEFORE);
                before.setRowKind(RowKind.UPDATE_BEFORE);
                emit(record, before, out);
            }

            GenericRowData after = extractAfterRow(value, valueSchema);
            validator.validate(after, RowKind.UPDATE_AFTER);
            after.setRowKind(RowKind.UPDATE_AFTER);
            if (sourceExactlyMetric != null) {
                out = new MetricsCollector<>(out, sourceExactlyMetric);
            }
            emit(record, after, out);
        }
        if (sourceExactlyMetric != null) {
            sourceExactlyMetric.incNumDeserializeSuccess();
            sourceExactlyMetric.recordDeserializeDelay(System.currentTimeMillis() - deserializeStartTime);
        }
    }
Returns a builder to build {@link RowDataDebeziumDeserializeSchema}.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/postgre/RowDataDebeziumDeserializeSchema.java