nodedb-query 0.1.1

// SPDX-License-Identifier: Apache-2.0

//! Peer-aware running-aggregate for RANGE BETWEEN UNBOUNDED PRECEDING AND
//! CURRENT ROW.
//!
//! This is the specialised fast path for the most common ordered-window
//! pattern (the PostgreSQL default frame for windows with ORDER BY).
//!
//! Peer semantics: two rows with equal ORDER BY values are in the same peer
//! group. All peers in a group see the same aggregate result: the value
//! computed at the *last* peer in the group (i.e., they all include each
//! other). This matches PostgreSQL behaviour for RANGE CURRENT ROW.

use super::helpers::{as_f64, get_field, order_keys_equal, set_window_col};
use super::spec::WindowFuncSpec;

/// Apply a peer-aware running aggregate over a sorted partition.
///
/// `indices` is the sorted slice of row indices within the partition.
/// `order_by` is the `(column, ascending)` list from the window spec — used
/// to detect peer groups.
pub(super) fn running_aggregate(
    rows: &mut [(String, serde_json::Value)],
    indices: &[usize],
    spec: &WindowFuncSpec,
    field: &str,
) {
    let len = indices.len();
    if len == 0 {
        return;
    }

    // Accumulate state incrementally row-by-row, but defer writing results
    // until the end of each peer group (so all peers see the group's final
    // value). We track where the current peer group started.
    let mut running_sum = 0.0f64;
    let mut running_count = 0u64;
    let mut running_min: Option<f64> = None;
    let mut running_max: Option<f64> = None;

    // Indices of rows belonging to the *current* peer group (deferred write).
    let mut peer_start = 0usize;

    for pos in 0..len {
        let i = indices[pos];
        let val = get_field(&rows[i].1, field);
        if let Some(n) = as_f64(&val) {
            running_sum += n;
            running_count += 1;
            running_min = Some(running_min.map_or(n, |m: f64| m.min(n)));
            running_max = Some(running_max.map_or(n, |m: f64| m.max(n)));
        } else if spec.func_name == "count" {
            running_count += 1;
        }

        // Check if the *next* row starts a new peer group (or we're at the end).
        let is_last_in_group =
            pos + 1 == len || !order_keys_equal(rows, i, indices[pos + 1], &spec.order_by);

        if is_last_in_group {
            // Compute the result at the end of this peer group.
            let result = match spec.func_name.as_str() {
                "sum" => serde_json::json!(running_sum),
                "count" => serde_json::json!(running_count),
                "avg" => {
                    if running_count > 0 {
                        serde_json::json!(running_sum / running_count as f64)
                    } else {
                        serde_json::Value::Null
                    }
                }
                "min" => running_min
                    .map(|m| serde_json::json!(m))
                    .unwrap_or(serde_json::Value::Null),
                "max" => running_max
                    .map(|m| serde_json::json!(m))
                    .unwrap_or(serde_json::Value::Null),
                "first_value" => get_field(&rows[indices[0]].1, field),
                "last_value" => get_field(&rows[indices[pos]].1, field),
                _ => serde_json::Value::Null,
            };

            // Write the *same* result to every row in the peer group.
            for &peer_idx in &indices[peer_start..=pos] {
                set_window_col(&mut rows[peer_idx].1, &spec.alias, result.clone());
            }

            peer_start = pos + 1;
        }
    }
}