Function extract_group_key

nodedb/src/event/streaming_mv/processor.rs:128–153

Extract the GROUP BY key by concatenating column values from the event. Looks for column values in: 1. CdcEvent's own fields (event_type, collection, partition, etc.) 2. new_value JSON object fields

(event: &CdcEvent, group_by_columns: &[String])

/// 1. CdcEvent's own fields (event_type, collection, partition, etc.)
/// 2. new_value JSON object fields
fn extract_group_key(event: &CdcEvent, group_by_columns: &[String]) -> String {
    let mut parts = Vec::with_capacity(group_by_columns.len());

    for col in group_by_columns {
        let val = match col.as_str() {
            "event_type" | "op" => event.op.clone(),
            "collection" => event.collection.clone(),
            "partition" => event.partition.to_string(),
            "row_id" => event.row_id.clone(),
            "tenant_id" => event.tenant_id.to_string(),
            // Look in new_value JSON.
            field => event
                .new_value
                .as_ref()
                .and_then(|v| v.get(field))
                .map(|v| match v {
                    serde_json::Value::String(s) => s.clone(),
                    other => other.to_string(),
                })
                .unwrap_or_else(|| "NULL".to_string()),
        };
        parts.push(val);
    }

    parts.join(":")
}

/// Extract a numeric value for an aggregate function from the event.
///