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.
(
rows: &mut [(String, serde_json::Value)],
indices: &[usize],
spec: &WindowFuncSpec,
field: &str,
)
| 20 | /// `order_by` is the `(column, ascending)` list from the window spec — used |
| 21 | /// to detect peer groups. |
| 22 | pub(super) fn running_aggregate( |
| 23 | rows: &mut [(String, serde_json::Value)], |
| 24 | indices: &[usize], |
| 25 | spec: &WindowFuncSpec, |
| 26 | field: &str, |
| 27 | ) { |
| 28 | let len = indices.len(); |
| 29 | if len == 0 { |
| 30 | return; |
| 31 | } |
| 32 | |
| 33 | // Accumulate state incrementally row-by-row, but defer writing results |
| 34 | // until the end of each peer group (so all peers see the group's final |
| 35 | // value). We track where the current peer group started. |
| 36 | let mut running_sum = 0.0f64; |
| 37 | let mut running_count = 0u64; |
| 38 | let mut running_min: Option<f64> = None; |
| 39 | let mut running_max: Option<f64> = None; |
| 40 | |
| 41 | // Indices of rows belonging to the *current* peer group (deferred write). |
| 42 | let mut peer_start = 0usize; |
| 43 | |
| 44 | for pos in 0..len { |
| 45 | let i = indices[pos]; |
| 46 | let val = get_field(&rows[i].1, field); |
| 47 | if let Some(n) = as_f64(&val) { |
| 48 | running_sum += n; |
| 49 | running_count += 1; |
| 50 | running_min = Some(running_min.map_or(n, |m: f64| m.min(n))); |
| 51 | running_max = Some(running_max.map_or(n, |m: f64| m.max(n))); |
| 52 | } else if spec.func_name == "count" { |
| 53 | running_count += 1; |
| 54 | } |
| 55 | |
| 56 | // Check if the *next* row starts a new peer group (or we're at the end). |
| 57 | let is_last_in_group = |
| 58 | pos + 1 == len || !order_keys_equal(rows, i, indices[pos + 1], &spec.order_by); |
| 59 | |
| 60 | if is_last_in_group { |
| 61 | // Compute the result at the end of this peer group. |
| 62 | let result = match spec.func_name.as_str() { |
| 63 | "sum" => serde_json::json!(running_sum), |
| 64 | "count" => serde_json::json!(running_count), |
| 65 | "avg" => { |
| 66 | if running_count > 0 { |
| 67 | serde_json::json!(running_sum / running_count as f64) |
| 68 | } else { |
| 69 | serde_json::Value::Null |
| 70 | } |
| 71 | } |
| 72 | "min" => running_min |
| 73 | .map(|m| serde_json::json!(m)) |
| 74 | .unwrap_or(serde_json::Value::Null), |
| 75 | "max" => running_max |
| 76 | .map(|m| serde_json::json!(m)) |
| 77 | .unwrap_or(serde_json::Value::Null), |
| 78 | "first_value" => get_field(&rows[indices[0]].1, field), |
| 79 | "last_value" => get_field(&rows[indices[pos]].1, field), |
no test coverage detected