Repository: risingwavelabs/risingwave
File: src/connector/src/parser/mod.rs (Rust)

Code before:
    mut parser: P,
    msg_stream: BoxSourceMessageStream,
    source_ctrl_opts: SourceCtrlOpts,
) {
    let mut chunk_builder =
        SourceStreamChunkBuilder::new(parser.columns().to_vec(), source_ctrl_opts);

    let mut direct_cdc_event_lag_latency_metrics = HashMap::new();

    #[for_await]
    for batch in msg_stream {
        // It's possible that the split is not active, which means the next batch may arrive
        // very lately, so we should prefer emitting all records in current batch before the end
        // of each iteration, instead of merging them with the next batch. An exception is when
        // a transaction is not committed yet, in which yield when the transaction is committed.

        let batch = batch?;
        let batch_len = batch.len();

        if batch_len == 0 {
            continue;
        }

        if batch.iter().all(|msg| msg.is_cdc_heartbeat()) {
            // This `.iter().all(...)` will short-circuit after seeing the first `false`, so in
            // normal cases, this should only involve a constant time cost.

            // Now we know that there is no data message in the batch, let's just emit the latest
            // heartbeat message. Note that all messages in `batch` should belong to the same
            // split, so we don't have to do a split to heartbeats mapping here.

            let heartbeat_msg = batch.last().unwrap();
            tracing::debug!(
                offset = heartbeat_msg.offset,
                "handling a heartbeat message"
            );
            chunk_builder.heartbeat(MessageMeta {
                source_meta: &heartbeat_msg.meta,
                split_id: &heartbeat_msg.split_id,
                offset: &heartbeat_msg.offset,
            });

            for chunk in chunk_builder.consume_ready_chunks() {
                yield chunk;
            }
            continue; // continue to next batch
        }

        // When we reach here, there is at least one data message in the batch. We should ignore all
        // heartbeat messages.


Review comment (performance):
Is the short circuit of `batch.iter().all(|msg| msg.is_cdc_heartbeat())` still necessary now? 

> let's just emit the latest heartbeat message

How much will it cost if we don't have such optimization? Will we frequently encounter a large number of heartbeats (only) in a chunk?

Code after:
async fn parse_message_stream<P: ByteStreamSourceParser>(
    mut parser: P,
    msg_stream: BoxSourceMessageStream,
    source_ctrl_opts: SourceCtrlOpts,
) {
    let mut chunk_builder =
        SourceStreamChunkBuilder::new(parser.columns().to_vec(), source_ctrl_opts);

    let mut direct_cdc_event_lag_latency_metrics = HashMap::new();

    #[for_await]
    for batch in msg_stream {
        // It's possible that the split is not active, which means the next batch may arrive
        // very lately, so we should prefer emitting all records in current batch before the end
        // of each iteration, instead of merging them with the next batch. An exception is when
        // a transaction is not committed yet, in which yield when the transaction is committed.

        let batch = batch?;
        let batch_len = batch.len();
        if batch_len == 0 {
            continue;
        }

        let mut txn_started_in_last_batch = chunk_builder.is_in_transaction();
        let process_time_ms = chrono::Utc::now().timestamp_millis();
        let mut is_heartbeat_emitted = false;
        for msg in batch {
            if msg.is_cdc_heartbeat() {
                if !is_heartbeat_emitted {
                    tracing::debug!(offset = msg.offset, "handling a heartbeat message");
                    chunk_builder.heartbeat(MessageMeta {
                        source_meta: &msg.meta,
                        split_id: &msg.split_id,
                        offset: &msg.offset,
                    });
                    for chunk in chunk_builder.consume_ready_chunks() {
                        yield chunk;
                    }
                    is_heartbeat_emitted = true;
                }
                continue;
            }

            // calculate process_time - event_time lag
            if let SourceMeta::DebeziumCdc(msg_meta) = &msg.meta {
                let lag_ms = process_time_ms - msg_meta.source_ts_ms;
                // report to promethus
                let full_table_name = msg_meta.full_table_name.clone();
                let direct_cdc_event_lag_latency = direct_cdc_event_lag_latency_metrics
                    .entry(full_table_name)
                    .or_insert_with(|| {