Function execute_aggregate_scan

nodedb/src/event/alert/executor.rs:149–191

Dispatch an aggregate scan to the Data Plane via the SPSC bridge. Builds a `TimeseriesOp::Scan` with the alert's aggregate function and group-by columns, dispatches via `dispatch_async`, and decodes the MessagePack response into `(group_key, agg_value)` pairs.

(
    state: &Arc<SharedState>,
    tenant_id: TenantId,
    alert: &AlertDef,
    window_start: i64,
    window_end: i64,
)

/// group-by columns, dispatches via `dispatch_async`, and decodes the
/// MessagePack response into `(group_key, agg_value)` pairs.
async fn execute_aggregate_scan(
    state: &Arc<SharedState>,
    tenant_id: TenantId,
    alert: &AlertDef,
    window_start: i64,
    window_end: i64,
) -> crate::Result<Vec<(String, f64)>> {
    // Encode WHERE filter as MessagePack for the Data Plane filter evaluator.
    let filters = match &alert.where_filter {
        Some(f) => zerompk::to_msgpack_vec(f).unwrap_or_default(),
        None => Vec::new(),
    };

    let plan = PhysicalPlan::Timeseries(TimeseriesOp::Scan {
        collection: alert.collection.clone(),
        time_range: (window_start, window_end),
        projection: Vec::new(),
        limit: 10_000, // Safety cap on group cardinality.
        filters,
        bucket_interval_ms: 0, // No time bucketing — aggregate over entire window.
        group_by: alert.group_by.clone(),
        aggregates: vec![(
            alert.condition.agg_func.clone(),
            alert.condition.column.clone(),
        )],
        gap_fill: String::new(),
        computed_columns: Vec::new(),
        rls_filters: Vec::new(),
        system_as_of_ms: None,
        valid_at_ms: None,
    });

    let payload = sync_dispatch::dispatch_async(
        state,
        tenant_id,
        &alert.collection,
        plan,
        Duration::from_secs(30),
    )
    .await?;

    decode_aggregate_response(&payload, &alert.group_by, &alert.condition.agg_func)
}

/// Decode the Data Plane's MessagePack response into (group_key, agg_value) pairs.
///