Function update_mv

nodedb/src/event/streaming_mv/processor.rs:103–121

Update a single MV's state from a CdcEvent.

(event: &CdcEvent, mv_state: &MvState)

/// Update a single MV's state from a CdcEvent.
fn update_mv(event: &CdcEvent, mv_state: &MvState) {
    // Extract GROUP BY key from the event.
    let group_key = extract_group_key(event, &mv_state.group_by_columns);

    // Extract aggregate values from the event.
    let agg_values: Vec<f64> = mv_state
        .aggregates
        .iter()
        .map(|agg| extract_agg_value(event, agg.function, &agg.input_expr))
        .collect();

    mv_state.update_with_time(&group_key, &agg_values, event.event_time);

    trace!(
        mv = %mv_state.name,
        group_key = %group_key,
        "streaming MV updated"
    );
}

/// Extract the GROUP BY key by concatenating column values from the event.
///