Function apply_v_per_row_aggregate

nodedb-query/src/window/value_agg.rs:111–170

(
    rows: &mut [Vec<Value>],
    indices: &[usize],
    column_index: &HashMap<String, usize>,
    spec: &WindowFuncSpec,
    write_col: usize,
)

}

fn apply_v_per_row_aggregate(
    rows: &mut [Vec<Value>],
    indices: &[usize],
    column_index: &HashMap<String, usize>,
    spec: &WindowFuncSpec,
    write_col: usize,
) {
    let len = indices.len();
    if len == 0 {
        return;
    }

    let order_expr = spec.order_by.first().map(|(expr, _)| expr);
    let order_values: Vec<Value> = indices
        .iter()
        .map(|&i| {
            order_expr
                .and_then(|expr| {
                    rows.get(i)
                        .map(|row| eval_arg_for_row(expr, row, column_index))
                })
                .unwrap_or(Value::Null)
        })
        .collect();

    let peer_groups: Vec<usize> = if spec.frame.mode == "groups" {
        build_v_peer_groups(&order_values)
    } else {
        Vec::new()
    };

    let all_vals: Vec<Option<f64>> = indices
        .iter()
        .map(|&i| {
            rows.get(i)
                .map(|row| eval_arg(spec, row, column_index).as_f64())
                .unwrap_or(None)
        })
        .collect();

    let results: Vec<Value> = (0..len)
        .map(|pos| {
            let (start_idx, end_idx) =
                evaluate_v_frame_bounds(&spec.frame, pos, len, &order_values, &peer_groups);
            aggregate_v_slice(
                &all_vals,
                indices,
                rows,
                column_index,
                spec,
                start_idx,
                end_idx,
            )
        })
        .collect();

    for (pos, result) in results.into_iter().enumerate() {
        set_cell(rows, indices[pos], write_col, result);