Function per_row_aggregate

nodedb-query/src/window/aggregate.rs:56–107

Per-row frame evaluator. For each row position `pos` in the partition: 1. Resolve the concrete `[start_idx, end_idx]` frame slice via `evaluate_frame_bounds`. 2. Aggregate `field` over `indices[start_idx..=end_idx]`. 3. Write the result back under `spec.alias`.

(
    rows: &mut [(String, serde_json::Value)],
    indices: &[usize],
    spec: &WindowFuncSpec,
    field: &str,
)

/// 2. Aggregate `field` over `indices[start_idx..=end_idx]`.
/// 3. Write the result back under `spec.alias`.
fn per_row_aggregate(
    rows: &mut [(String, serde_json::Value)],
    indices: &[usize],
    spec: &WindowFuncSpec,
    field: &str,
) {
    let len = indices.len();
    if len == 0 {
        return;
    }

    // Extract order-by values for RANGE numeric offsets.
    let order_expr = spec.order_by.first().map(|(expr, _)| expr);
    let order_values: Vec<serde_json::Value> = indices
        .iter()
        .map(|&i| {
            order_expr
                .map(|expr| super::helpers::eval_expr_on_json(expr, &rows[i].1))
                .unwrap_or(serde_json::Value::Null)
        })
        .collect();

    // Peer groups needed for GROUPS mode (and for RANGE CurrentRow peer
    // awareness — reused from the frame module which handles both).
    let peer_groups: Vec<usize> = if spec.frame.mode == "groups" {
        build_peer_groups(&order_values)
    } else {
        Vec::new()
    };

    // Pre-collect all numeric values to avoid repeated borrow issues.
    let all_vals: Vec<Option<f64>> = indices
        .iter()
        .map(|&i| as_f64(&get_field(&rows[i].1, field)))
        .collect();

    // We need to write into `rows` after computing each result; collect
    // results first so we only borrow `rows` immutably during computation.
    let results: Vec<serde_json::Value> = (0..len)
        .map(|pos| {
            let (start_idx, end_idx) =
                evaluate_frame_bounds(&spec.frame, pos, len, &order_values, &peer_groups);

            aggregate_slice(&all_vals, indices, rows, field, spec, start_idx, end_idx)
        })
        .collect();

    for (pos, result) in results.into_iter().enumerate() {
        let row_idx = indices[pos];
        set_window_col(&mut rows[row_idx].1, &spec.alias, result);
    }
}

/// Aggregate `field` over the slice `indices[start_idx..=end_idx]`.
fn aggregate_slice(