use nodedb_types::{NdbDateTime, read_instant};
use crate::expr::{EvalError, GroupKeySpec, SqlExpr};
use crate::msgpack_scan::field::extract_field;
use crate::msgpack_scan::index::FieldIndex;
use crate::msgpack_scan::reader::{read_f64, read_integer, read_null, read_str};
pub fn build_group_key(doc: &[u8], group_keys: &[GroupKeySpec]) -> Result<String, EvalError> {
if group_keys.is_empty() {
return Ok("__all__".to_string());
}
let mut key_buf = String::new();
key_buf.push('[');
let mut written = 0usize;
for spec in group_keys {
if let Some(field) = spec.field.as_deref() {
if written > 0 {
key_buf.push(',');
}
append_field_value(&mut key_buf, doc, field);
written += 1;
} else if let Some(expr) = spec.expr.as_ref() {
if written > 0 {
key_buf.push(',');
}
append_computed_value(&mut key_buf, doc, expr)?;
written += 1;
}
}
key_buf.push(']');
Ok(key_buf)
}
pub fn build_group_key_indexed(
doc: &[u8],
group_keys: &[GroupKeySpec],
idx: &FieldIndex,
) -> Result<String, EvalError> {
if group_keys.is_empty() {
return Ok("__all__".to_string());
}
let mut key_buf = String::new();
key_buf.push('[');
let mut written = 0usize;
for spec in group_keys {
if let Some(field) = spec.field.as_deref() {
if written > 0 {
key_buf.push(',');
}
let range = idx.get(field);
append_field_value_range(&mut key_buf, doc, range);
written += 1;
} else if let Some(expr) = spec.expr.as_ref() {
if written > 0 {
key_buf.push(',');
}
append_computed_value(&mut key_buf, doc, expr)?;
written += 1;
}
}
key_buf.push(']');
Ok(key_buf)
}
fn append_value_at(buf: &mut String, doc: &[u8], start: usize, end: usize) {
if read_null(doc, start) {
buf.push_str("null");
} else if let Some(s) = read_str(doc, start) {
buf.push('"');
buf.push_str(s);
buf.push('"');
} else if let Some((_, micros)) = read_instant(doc, start) {
buf.push('"');
buf.push_str(&NdbDateTime::from_micros(micros).to_iso8601());
buf.push('"');
} else if let Some(n) = read_integer(doc, start) {
use std::fmt::Write;
let _ = write!(buf, "{n}");
} else if let Some(n) = read_f64(doc, start) {
use std::fmt::Write;
let _ = write!(buf, "{n}");
} else {
buf.push_str(&hex::encode(&doc[start..end]));
}
}
fn append_field_value(buf: &mut String, doc: &[u8], field: &str) {
match extract_field(doc, 0, field) {
Some((start, end)) => append_value_at(buf, doc, start, end),
None => buf.push_str("null"),
}
}
fn append_field_value_range(buf: &mut String, doc: &[u8], range: Option<(usize, usize)>) {
match range {
Some((start, end)) => append_value_at(buf, doc, start, end),
None => buf.push_str("null"),
}
}
fn append_computed_value(buf: &mut String, doc: &[u8], expr: &SqlExpr) -> Result<(), EvalError> {
let Ok(doc_val) = nodedb_types::json_msgpack::value_from_msgpack(doc) else {
buf.push_str("null");
return Ok(());
};
let val = expr.eval(&doc_val)?;
match nodedb_types::json_msgpack::value_to_msgpack(&val) {
Ok(vb) => append_value_at(buf, &vb, 0, vb.len()),
Err(_) => buf.push_str("null"),
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
fn encode(v: &serde_json::Value) -> Vec<u8> {
nodedb_types::json_msgpack::json_to_msgpack(v).expect("encode")
}
fn keys(fields: &[&str]) -> Vec<GroupKeySpec> {
fields.iter().map(|f| GroupKeySpec::column(*f)).collect()
}
#[test]
fn single_string_field() {
let doc = encode(&json!({"name": "alice", "age": 30}));
let key = build_group_key(&doc, &keys(&["name"])).unwrap();
assert_eq!(key, r#"["alice"]"#);
}
#[test]
fn single_int_field() {
let doc = encode(&json!({"status": 200}));
let key = build_group_key(&doc, &keys(&["status"])).unwrap();
assert_eq!(key, "[200]");
}
#[test]
fn large_integers_key_exactly() {
let doc = encode(&json!({"v": u64::MAX}));
let key = build_group_key(&doc, &keys(&["v"])).unwrap();
assert_eq!(key, "[18446744073709551615]");
let a = encode(&json!({"v": 9_007_199_254_740_993_i64}));
let b = encode(&json!({"v": 9_007_199_254_740_992_i64}));
assert_ne!(
build_group_key(&a, &keys(&["v"])).unwrap(),
build_group_key(&b, &keys(&["v"])).unwrap()
);
}
#[test]
fn multiple_fields() {
let doc = encode(&json!({"city": "ny", "year": 2024}));
let key = build_group_key(&doc, &keys(&["city", "year"])).unwrap();
assert_eq!(key, r#"["ny",2024]"#);
}
#[test]
fn missing_field_is_null() {
let doc = encode(&json!({"x": 1}));
let key = build_group_key(&doc, &keys(&["missing"])).unwrap();
assert_eq!(key, "[null]");
}
#[test]
fn empty_group_fields() {
let doc = encode(&json!({"x": 1}));
let key = build_group_key(&doc, &[]).unwrap();
assert_eq!(key, "__all__");
}
#[test]
fn null_field_value() {
let doc = encode(&json!({"v": null}));
let key = build_group_key(&doc, &keys(&["v"])).unwrap();
assert_eq!(key, "[null]");
}
#[test]
fn float_field() {
let doc = encode(&json!({"temp": 36.6}));
let key = build_group_key(&doc, &keys(&["temp"])).unwrap();
assert_eq!(key, "[36.6]");
}
#[test]
fn instant_field_keys_as_iso8601() {
use nodedb_types::InstantKind;
let mut doc = Vec::new();
crate::msgpack_scan::write_map_header(&mut doc, 2);
crate::msgpack_scan::write_kv_instant(
&mut doc,
"ts",
InstantKind::Naive,
1_583_402_400_000_000,
);
crate::msgpack_scan::write_kv_i64(&mut doc, "n", 1);
let key = build_group_key(&doc, &keys(&["ts", "n"])).unwrap();
assert_eq!(key, r#"["2020-03-05T10:00:00.000000Z",1]"#);
let idx = FieldIndex::build(&doc, 0).unwrap_or_else(FieldIndex::empty);
assert_eq!(
build_group_key_indexed(&doc, &keys(&["ts", "n"]), &idx).unwrap(),
key
);
}
fn upper_label_spec() -> GroupKeySpec {
GroupKeySpec {
output_name: "u".to_string(),
field: None,
expr: Some(SqlExpr::Function {
name: "upper".to_string(),
args: vec![SqlExpr::Column("label".to_string())],
}),
}
}
#[test]
fn computed_key_folds_expression_value() {
let lower = encode(&json!({"label": "alpha", "score": 7}));
let upper = encode(&json!({"label": "ALPHA", "score": 3}));
assert_eq!(
build_group_key(&lower, &[upper_label_spec()]).unwrap(),
r#"["ALPHA"]"#
);
assert_eq!(
build_group_key(&upper, &[upper_label_spec()]).unwrap(),
r#"["ALPHA"]"#
);
}
#[test]
fn computed_key_division_by_zero_errors() {
let doc = encode(&json!({"n": 10, "d": 0}));
let spec = GroupKeySpec {
output_name: "q".to_string(),
field: None,
expr: Some(SqlExpr::BinaryOp {
left: Box::new(SqlExpr::Column("n".to_string())),
op: crate::expr::BinaryOp::Div,
right: Box::new(SqlExpr::Column("d".to_string())),
}),
};
assert_eq!(
build_group_key(&doc, std::slice::from_ref(&spec)),
Err(EvalError::DivisionByZero)
);
let idx = FieldIndex::build(&doc, 0).unwrap_or_else(FieldIndex::empty);
assert_eq!(
build_group_key_indexed(&doc, std::slice::from_ref(&spec), &idx),
Err(EvalError::DivisionByZero)
);
}
#[test]
fn computed_key_indexed_matches_unindexed() {
let doc = encode(&json!({"label": "beta", "score": 5}));
let specs = [upper_label_spec()];
let idx = FieldIndex::build(&doc, 0).unwrap_or_else(FieldIndex::empty);
assert_eq!(
build_group_key(&doc, &specs).unwrap(),
build_group_key_indexed(&doc, &specs, &idx).unwrap(),
);
}
#[test]
fn mixed_column_and_computed_keys() {
let doc = encode(&json!({"region": "us", "label": "west"}));
let specs = [GroupKeySpec::column("region"), upper_label_spec()];
assert_eq!(build_group_key(&doc, &specs).unwrap(), r#"["us","WEST"]"#);
}
}