|
| 1 | +//! Post-join aggregation in the Control Plane. |
| 2 | +//! |
| 3 | +//! When a query has `GROUP BY` over a `JOIN` result, the Data Plane cores |
| 4 | +//! return raw join rows. This module aggregates them in the Control Plane. |
| 5 | +
|
| 6 | +use std::collections::HashMap; |
| 7 | + |
| 8 | +use crate::bridge::envelope::{Payload, Response}; |
| 9 | + |
| 10 | +/// Apply GROUP BY + aggregate functions on a join response payload. |
| 11 | +/// |
| 12 | +/// The input response payload is a JSON array of objects (merged from all cores). |
| 13 | +/// Returns a new response with aggregated results. |
| 14 | +pub fn apply_post_aggregation( |
| 15 | + resp: Response, |
| 16 | + group_by: &[String], |
| 17 | + aggregates: &[(String, String)], |
| 18 | +) -> crate::Result<Response> { |
| 19 | + let payload_bytes = resp.payload.as_bytes(); |
| 20 | + let json_text = std::str::from_utf8(payload_bytes).map_err(|e| crate::Error::PlanError { |
| 21 | + detail: format!("post-aggregation: invalid UTF-8: {e}"), |
| 22 | + })?; |
| 23 | + |
| 24 | + let rows: Vec<serde_json::Value> = |
| 25 | + serde_json::from_str(json_text).map_err(|e| crate::Error::PlanError { |
| 26 | + detail: format!("post-aggregation: JSON parse error: {e}"), |
| 27 | + })?; |
| 28 | + |
| 29 | + // Group rows by the GROUP BY columns. |
| 30 | + // Join results use "collection.field" keys, but GROUP BY may use bare field names. |
| 31 | + let mut groups: HashMap<Vec<String>, Vec<&serde_json::Value>> = HashMap::new(); |
| 32 | + for row in &rows { |
| 33 | + let key: Vec<String> = group_by |
| 34 | + .iter() |
| 35 | + .map(|col| { |
| 36 | + resolve_field(row, col) |
| 37 | + .map(|v| match v { |
| 38 | + serde_json::Value::String(s) => s.clone(), |
| 39 | + other => other.to_string(), |
| 40 | + }) |
| 41 | + .unwrap_or_default() |
| 42 | + }) |
| 43 | + .collect(); |
| 44 | + groups.entry(key).or_default().push(row); |
| 45 | + } |
| 46 | + |
| 47 | + // Compute aggregates per group. |
| 48 | + let mut result = Vec::with_capacity(groups.len()); |
| 49 | + for (key, group_rows) in &groups { |
| 50 | + let mut obj = serde_json::Map::new(); |
| 51 | + |
| 52 | + // Add GROUP BY columns. |
| 53 | + for (i, col) in group_by.iter().enumerate() { |
| 54 | + obj.insert(col.clone(), serde_json::Value::String(key[i].clone())); |
| 55 | + } |
| 56 | + |
| 57 | + // Compute each aggregate. |
| 58 | + for (op, field) in aggregates { |
| 59 | + let agg_key = format!("{op}({field})"); |
| 60 | + let value = compute_aggregate(op, field, group_rows); |
| 61 | + obj.insert(agg_key, value); |
| 62 | + } |
| 63 | + |
| 64 | + result.push(serde_json::Value::Object(obj)); |
| 65 | + } |
| 66 | + |
| 67 | + let output = serde_json::to_vec(&result).map_err(|e| crate::Error::PlanError { |
| 68 | + detail: format!("post-aggregation: serialize error: {e}"), |
| 69 | + })?; |
| 70 | + |
| 71 | + Ok(Response { |
| 72 | + payload: Payload::from_vec(output), |
| 73 | + ..resp |
| 74 | + }) |
| 75 | +} |
| 76 | + |
| 77 | +/// Compute a single aggregate over a group of rows. |
| 78 | +fn compute_aggregate(op: &str, field: &str, rows: &[&serde_json::Value]) -> serde_json::Value { |
| 79 | + match op { |
| 80 | + "count" => serde_json::Value::Number(serde_json::Number::from(rows.len() as u64)), |
| 81 | + "sum" => { |
| 82 | + let sum: f64 = rows.iter().filter_map(|r| extract_number(r, field)).sum(); |
| 83 | + serde_json::json!(sum) |
| 84 | + } |
| 85 | + "avg" => { |
| 86 | + let values: Vec<f64> = rows |
| 87 | + .iter() |
| 88 | + .filter_map(|r| extract_number(r, field)) |
| 89 | + .collect(); |
| 90 | + if values.is_empty() { |
| 91 | + serde_json::Value::Null |
| 92 | + } else { |
| 93 | + serde_json::json!(values.iter().sum::<f64>() / values.len() as f64) |
| 94 | + } |
| 95 | + } |
| 96 | + "min" => rows |
| 97 | + .iter() |
| 98 | + .filter_map(|r| extract_number(r, field)) |
| 99 | + .min_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal)) |
| 100 | + .map(|v| serde_json::json!(v)) |
| 101 | + .unwrap_or(serde_json::Value::Null), |
| 102 | + "max" => rows |
| 103 | + .iter() |
| 104 | + .filter_map(|r| extract_number(r, field)) |
| 105 | + .max_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal)) |
| 106 | + .map(|v| serde_json::json!(v)) |
| 107 | + .unwrap_or(serde_json::Value::Null), |
| 108 | + _ => serde_json::Value::Null, |
| 109 | + } |
| 110 | +} |
| 111 | + |
| 112 | +/// Extract a numeric value from a JSON object field. |
| 113 | +fn extract_number(row: &serde_json::Value, field: &str) -> Option<f64> { |
| 114 | + if field == "*" { |
| 115 | + return Some(1.0); |
| 116 | + } |
| 117 | + resolve_field(row, field).and_then(|v| match v { |
| 118 | + serde_json::Value::Number(n) => n.as_f64(), |
| 119 | + serde_json::Value::String(s) => s.parse().ok(), |
| 120 | + _ => None, |
| 121 | + }) |
| 122 | +} |
| 123 | + |
| 124 | +/// Resolve a field name in a JSON row, handling "collection.field" prefixed keys. |
| 125 | +/// |
| 126 | +/// Join results use keys like "users.name" but SQL refers to fields as just "name" |
| 127 | +/// or with alias "u.name". This tries exact match first, then suffix match. |
| 128 | +fn resolve_field<'a>(row: &'a serde_json::Value, field: &str) -> Option<&'a serde_json::Value> { |
| 129 | + // Exact match first. |
| 130 | + if let Some(v) = row.get(field) { |
| 131 | + return Some(v); |
| 132 | + } |
| 133 | + // Suffix match: look for any key ending with ".{field}". |
| 134 | + let suffix = format!(".{field}"); |
| 135 | + if let serde_json::Value::Object(map) = row { |
| 136 | + for (k, v) in map { |
| 137 | + if k.ends_with(&suffix) { |
| 138 | + return Some(v); |
| 139 | + } |
| 140 | + } |
| 141 | + } |
| 142 | + None |
| 143 | +} |
0 commit comments