Function extract_agg_value

nodedb/src/event/streaming_mv/processor.rs:160–189

Extract a numeric value for an aggregate function from the event. For COUNT, always returns 1.0 (each event counts as one). For SUM/MIN/MAX/AVG, parses the input expression as a field path and extracts the numeric value from new_value.

(event: &CdcEvent, func: AggFunction, input_expr: &str)

/// For SUM/MIN/MAX/AVG, parses the input expression as a field path
/// and extracts the numeric value from new_value.
fn extract_agg_value(event: &CdcEvent, func: AggFunction, input_expr: &str) -> f64 {
    if func == AggFunction::Count {
        return 1.0; // Each event counts as one.
    }

    // Parse input_expr as a field name (simple case).
    // Supports: "field_name" or "doc_get(new_value, '$.field')".
    let field_name = if input_expr.contains("doc_get") {
        // Extract field path from doc_get(new_value, '$.field').
        input_expr
            .split("'$.")
            .nth(1)
            .and_then(|s| s.split('\'').next())
            .unwrap_or(input_expr)
    } else {
        input_expr.trim()
    };

    // Look up the field in new_value.
    event
        .new_value
        .as_ref()
        .and_then(|v| v.get(field_name))
        .and_then(|v| match v {
            serde_json::Value::Number(n) => n.as_f64(),
            serde_json::Value::String(s) => s.parse::<f64>().ok(),
            _ => None,
        })
        .unwrap_or(f64::NAN) // NaN → skipped by GroupState::update.
}

#[cfg(test)]
mod tests {