#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
use arrow::array::Array;
use chronix::chronix_core::histogram::{Bucket, CUSTOM_BUCKETS_SCHEMA, Histogram, ResetHint};
use chronix::prelude::*;
use std::collections::BTreeMap;
fn latency_histogram(scale: f64) -> Histogram {
let mut h = Histogram::empty(3);
h.zero_threshold = 1e-6;
h.reset_hint = ResetHint::NoReset;
for i in 1..=200 {
h.observe(f64::from(i) * scale / 200.0);
}
h
}
fn point_at(ts: i64, h: Histogram) -> Point {
let key = SeriesKey::new("http", tags! { "route" => "/api" }).unwrap();
let mut fields: BTreeMap<String, FieldValue> = BTreeMap::new();
fields.insert("latency".into(), FieldValue::Histogram(Box::new(h)));
Point::new(key, fields, ts).unwrap()
}
fn read_back(db: &Chronix) -> Vec<Histogram> {
let plan = db
.query()
.measurement("http")
.range(0, i64::MAX)
.build()
.unwrap();
let batches = db.execute_stream(&plan).unwrap();
let mut out = Vec::new();
for b in &batches {
let idx = b
.schema()
.index_of("latency")
.expect("the column is present");
let col = b.column(idx);
let bin = col
.as_any()
.downcast_ref::<arrow::array::BinaryArray>()
.expect("a histogram column reads back as Binary");
for i in 0..bin.len() {
if bin.is_null(i) {
continue;
}
out.push(postcard::from_bytes::<Histogram>(bin.value(i)).expect("decode"));
}
}
out
}
#[test]
fn a_histogram_survives_the_memtable_a_flush_and_a_restart() {
let dir = tempfile::tempdir().unwrap();
let original = latency_histogram(2.0);
{
let db = Chronix::open_small(dir.path()).unwrap();
db.insert(&point_at(1_700_000_000_000_000_000, original.clone()))
.unwrap();
let from_memtable = read_back(&db);
assert_eq!(from_memtable.len(), 1, "the unflushed point is queryable");
assert_eq!(from_memtable[0], original, "unchanged in the memtable");
db.flush().unwrap();
let from_segment = read_back(&db);
assert_eq!(from_segment.len(), 1, "the flushed point is queryable");
assert_eq!(from_segment[0], original, "unchanged through a segment");
db.close().unwrap();
}
let db = Chronix::open_small(dir.path()).unwrap();
let after_restart = read_back(&db);
assert_eq!(after_restart.len(), 1, "the point survived a reopen");
assert_eq!(after_restart[0], original, "unchanged across a restart");
assert!(
after_restart[0].validate().is_ok(),
"and it is still a valid histogram"
);
db.close().unwrap();
}
#[test]
fn the_schema_reports_a_histogram_column_as_one() {
let dir = tempfile::tempdir().unwrap();
let db = Chronix::open_small(dir.path()).unwrap();
db.insert(&point_at(1_700_000_000_000_000_000, latency_histogram(1.0)))
.unwrap();
let schema = db.schema("http").expect("the measurement exists");
let col = schema
.columns()
.iter()
.find(|c| c.name == "latency")
.expect("the field is in the schema");
assert_eq!(
col.column_type,
chronix::chronix_core::ColumnType::Histogram,
"a histogram field is typed as one, not inferred as a blob"
);
db.close().unwrap();
}
#[test]
fn a_custom_bucket_histogram_survives_a_flush() {
let dir = tempfile::tempdir().unwrap();
let db = Chronix::open_small(dir.path()).unwrap();
let mut h = Histogram::empty(CUSTOM_BUCKETS_SCHEMA);
h.custom_values = vec![
0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0,
];
for v in [0.003, 0.02, 0.4, 3.0, 20.0] {
h.observe(v);
}
db.insert(&point_at(1_700_000_000_000_000_000, h.clone()))
.unwrap();
db.flush().unwrap();
let back = read_back(&db);
assert_eq!(back.len(), 1);
assert_eq!(
back[0].custom_values, h.custom_values,
"boundaries survived"
);
assert_eq!(back[0], h);
db.close().unwrap();
}
#[test]
fn two_histograms_in_one_measurement_keep_their_own_buckets() {
let dir = tempfile::tempdir().unwrap();
let db = Chronix::open_small(dir.path()).unwrap();
let a = latency_histogram(1.0);
let b = latency_histogram(1000.0);
db.insert(&point_at(1_700_000_000_000_000_000, a.clone()))
.unwrap();
db.insert(&point_at(1_700_000_001_000_000_000, b.clone()))
.unwrap();
db.flush().unwrap();
let back = read_back(&db);
assert_eq!(back.len(), 2);
assert!(
back.contains(&a) && back.contains(&b),
"both survived intact"
);
assert_ne!(
back[0].positive, back[1].positive,
"two different distributions keep two different bucket sets"
);
db.close().unwrap();
}
#[test]
fn a_histogram_and_a_float_can_share_a_measurement() {
let dir = tempfile::tempdir().unwrap();
let db = Chronix::open_small(dir.path()).unwrap();
let key = SeriesKey::new("http", tags! { "route" => "/api" }).unwrap();
let mut fields: BTreeMap<String, FieldValue> = BTreeMap::new();
fields.insert(
"latency".into(),
FieldValue::Histogram(Box::new(latency_histogram(1.0))),
);
fields.insert("inflight".into(), FieldValue::F64(7.0));
db.insert(&Point::new(key, fields, 1_700_000_000_000_000_000).unwrap())
.unwrap();
db.flush().unwrap();
let plan = db
.query()
.measurement("http")
.range(0, i64::MAX)
.build()
.unwrap();
let batches = db.execute_stream(&plan).unwrap();
let b = &batches[0];
assert!(b.schema().index_of("latency").is_ok());
assert!(b.schema().index_of("inflight").is_ok());
assert_eq!(read_back(&db).len(), 1);
db.close().unwrap();
}
#[test]
fn an_invalid_histogram_is_refused_at_the_write_call() {
let dir = tempfile::tempdir().unwrap();
let db = Chronix::open_small(dir.path()).unwrap();
let mut bad = Histogram::empty(0);
bad.count = 1.0;
bad.positive = vec![
Bucket {
index: 5,
count: 1.0,
},
Bucket {
index: 2,
count: 1.0,
}, ];
let key = SeriesKey::new("http", tags! { "route" => "/api" }).unwrap();
let mut fields: BTreeMap<String, FieldValue> = BTreeMap::new();
fields.insert("latency".into(), FieldValue::Histogram(Box::new(bad)));
let err = Point::new(key, fields, 1_700_000_000_000_000_000)
.expect_err("an unsorted bucket list is not a histogram");
let msg = err.to_string();
assert!(
msg.contains("ascending") || msg.contains("histogram"),
"the refusal should name what is wrong: {msg}"
);
db.close().unwrap();
}
#[test]
fn a_rollup_over_a_histogram_column_merges_rather_than_dropping_it() {
let dir = tempfile::tempdir().unwrap();
let db = Chronix::open_small(dir.path()).unwrap();
db.create_rollup(
chronix::RollupBuilder::new()
.name("http_1m")
.source("http")
.target("http_1m")
.every("1m")
.aggregation(chronix::RollupAggFn::Sum)
.build()
.unwrap(),
)
.unwrap();
let base = 1_700_000_000_000_000_000;
let mut expected = Histogram::empty(3);
for k in 0..3i64 {
let mut h = Histogram::empty(3);
h.zero_threshold = 1e-6;
for i in 1..=50 {
h.observe(f64::from(i) / 25.0 + k as f64);
}
expected.merge(&h).unwrap();
db.insert(&point_at(base + k * 1_000_000_000, h)).unwrap();
}
db.insert(&point_at(
base + 4 * 3_600_000_000_000,
latency_histogram(1.0),
))
.unwrap();
db.flush().unwrap();
let n = db.materialise_rollups().unwrap();
assert!(n > 0, "the rollup pass materialised nothing at all");
let plan = db
.query()
.measurement("http_1m")
.range(0, i64::MAX)
.build()
.unwrap();
let batches = db.execute_stream(&plan).unwrap();
let mut rolled: Vec<Histogram> = Vec::new();
for b in &batches {
let Ok(idx) = b.schema().index_of("latency_sum") else {
continue;
};
let bin = b
.column(idx)
.as_any()
.downcast_ref::<arrow::array::BinaryArray>()
.expect("a rolled-up histogram is still a histogram column");
for i in 0..bin.len() {
if !bin.is_null(i) {
rolled.push(postcard::from_bytes(bin.value(i)).expect("decode"));
}
}
}
assert!(
!rolled.is_empty(),
"the rollup produced no histogram column — a tier that silently drops \
the column it exists to summarise is the failure this test is for"
);
let first = &rolled[0];
assert_eq!(
first.count, expected.count,
"the merged histogram must hold every observation of the bucket"
);
assert!(
(first.sum - expected.sum).abs() < 1e-6,
"and their sum: got {}, expected {}",
first.sum,
expected.sum
);
assert_eq!(
first.positive, expected.positive,
"bucket for bucket, the rollup equals merging the samples by hand"
);
assert!(first.validate().is_ok(), "and it is a valid histogram");
db.close().unwrap();
}
#[test]
fn a_rolled_up_histogram_still_answers_quantiles() {
let dir = tempfile::tempdir().unwrap();
let db = Chronix::open_small(dir.path()).unwrap();
let mut a = Histogram::empty(4);
let mut b = Histogram::empty(4);
for i in 1..=100 {
a.observe(f64::from(i) / 100.0);
b.observe(f64::from(i) / 100.0);
}
let mut merged = a.clone();
merged.merge(&b).unwrap();
assert!(
(merged.quantile(0.5) - a.quantile(0.5)).abs() < 1e-9,
"merging two identical distributions must not move the median: \
{} vs {}",
merged.quantile(0.5),
a.quantile(0.5)
);
assert_eq!(merged.count, a.count + b.count);
db.close().unwrap();
}
fn one_value(db: &Chronix, q: &str, at: i64) -> f64 {
match db.promql(q, at).unwrap_or_else(|e| panic!("{q}: {e}")) {
chronix::promql::ast::PromQLValue::Vector(v) => {
assert_eq!(v.len(), 1, "{q} returned {} series", v.len());
v[0].samples[0].value
}
other => panic!("{q} returned {other:?}"),
}
}
#[test]
fn promql_reads_a_stored_native_histogram() {
let dir = tempfile::tempdir().unwrap();
let db = Chronix::open_small(dir.path()).unwrap();
let at = 1_700_000_000_000_000_000;
let mut h = Histogram::empty(5);
h.zero_threshold = 1e-9;
for i in 1..=200 {
h.observe(f64::from(i) / 200.0);
}
let (count, sum, avg) = (h.count, h.sum, h.avg());
let (q50, q99) = (h.quantile(0.5), h.quantile(0.99));
let stddev = h.stddev();
db.insert(&point_at(at, h)).unwrap();
db.flush().unwrap();
assert!(
(one_value(&db, "histogram_count(http_latency)", at) - count).abs() < 1e-9,
"histogram_count must be the observation count"
);
assert!(
(one_value(&db, "histogram_sum(http_latency)", at) - sum).abs() < 1e-9,
"histogram_sum must be the observation sum"
);
assert!(
(one_value(&db, "histogram_avg(http_latency)", at) - avg).abs() < 1e-9,
"histogram_avg must be sum/count"
);
assert!(
(one_value(&db, "histogram_stddev(http_latency)", at) - stddev).abs() < 1e-9,
"histogram_stddev must match the type's own answer"
);
assert!(
(one_value(&db, "histogram_quantile(0.5, http_latency)", at) - q50).abs() < 1e-9,
"histogram_quantile must read the native buckets, not look for `le` labels"
);
assert!((one_value(&db, "histogram_quantile(0.99, http_latency)", at) - q99).abs() < 1e-9);
let p50 = one_value(&db, "histogram_quantile(0.5, http_latency)", at);
assert!(
(p50 - 0.5).abs() < 0.02,
"the median of a uniform [0.005, 1.0] distribution should be ~0.5, got {p50}"
);
let all = one_value(&db, "histogram_fraction(-Inf, +Inf, http_latency)", at);
assert!(
(all - 1.0).abs() < 1e-6,
"fraction over everything is 1, got {all}"
);
let half = one_value(&db, "histogram_fraction(-Inf, 0.5, http_latency)", at);
assert!(
(half - 0.5).abs() < 0.02,
"half the uniform distribution is below 0.5, got {half}"
);
db.close().unwrap();
}
#[test]
fn the_histogram_family_refuses_a_float_series_by_name() {
let dir = tempfile::tempdir().unwrap();
let db = Chronix::open_small(dir.path()).unwrap();
let at = 1_700_000_000_000_000_000;
let key = SeriesKey::new("cpu", tags! { "host" => "a" }).unwrap();
let mut fields: BTreeMap<String, FieldValue> = BTreeMap::new();
fields.insert("value".into(), FieldValue::F64(42.0));
db.insert(&Point::new(key, fields, at).unwrap()).unwrap();
db.flush().unwrap();
for q in [
"histogram_count(cpu)",
"histogram_sum(cpu)",
"histogram_avg(cpu)",
"histogram_stddev(cpu)",
"histogram_stdvar(cpu)",
"histogram_fraction(0, 1, cpu)",
] {
let err = db
.promql(q, at)
.expect_err(&format!("{q} must refuse a float series"));
let msg = err.to_string();
assert!(
msg.contains("histogram"),
"{q} should say what it wanted: {msg}"
);
}
db.close().unwrap();
}
#[test]
fn a_classic_histogram_still_answers_histogram_quantile() {
let dir = tempfile::tempdir().unwrap();
let db = Chronix::open_small(dir.path()).unwrap();
let at = 1_700_000_000_000_000_000;
for (le, count) in [("0.1", 1.0), ("0.5", 3.0), ("1", 5.0), ("+Inf", 5.0)] {
let key = SeriesKey::new("lat", tags! { "le" => le }).unwrap();
let mut fields: BTreeMap<String, FieldValue> = BTreeMap::new();
fields.insert("value".into(), FieldValue::F64(count));
db.insert(&Point::new(key, fields, at).unwrap()).unwrap();
}
db.flush().unwrap();
let v = one_value(&db, "histogram_quantile(0.5, lat)", at);
assert!(
v > 0.1 && v < 1.0,
"the classic path must still work: got {v}"
);
db.close().unwrap();
}
fn vector(db: &Chronix, q: &str, at: i64) -> Vec<(BTreeMap<String, String>, f64)> {
match db.promql(q, at).unwrap_or_else(|e| panic!("{q}: {e}")) {
chronix::promql::ast::PromQLValue::Vector(v) => v
.into_iter()
.map(|s| {
let labels: BTreeMap<String, String> = s.labels.into_iter().collect();
(labels, s.samples[0].value)
})
.collect(),
other => panic!("{q} returned {other:?}"),
}
}
fn classic_fixture(dir: &tempfile::TempDir) -> (Chronix, i64) {
let db = Chronix::open_small(dir.path()).unwrap();
let at = 1_700_000_000_000_000_000;
let mut h = Histogram::empty(CUSTOM_BUCKETS_SCHEMA);
h.custom_values = vec![0.1, 0.5, 1.0];
h.positive = vec![
Bucket {
index: 0,
count: 1.0,
},
Bucket {
index: 1,
count: 2.0,
},
Bucket {
index: 2,
count: 3.0,
},
Bucket {
index: 3,
count: 4.0,
},
];
h.count = 10.0;
h.sum = 6.5;
h.validate().unwrap();
let key = SeriesKey::new("lat", tags! { "job" => "api" }).unwrap();
let mut fields: BTreeMap<String, FieldValue> = BTreeMap::new();
fields.insert("value".into(), FieldValue::Histogram(Box::new(h)));
db.insert(&Point::new(key, fields, at).unwrap()).unwrap();
db.flush().unwrap();
(db, at)
}
#[test]
fn a_native_histogram_answers_the_classic_bucket_name() {
let dir = tempfile::tempdir().unwrap();
let (db, at) = classic_fixture(&dir);
let series = vector(&db, "lat_bucket", at);
let mut got: Vec<(String, f64)> = series.iter().map(|(l, v)| (l["le"].clone(), *v)).collect();
got.sort_by(|a, b| a.0.cmp(&b.0));
let mut expected = vec![
("0.1".to_string(), 1.0),
("0.5".to_string(), 3.0), ("1".to_string(), 6.0), ("+Inf".to_string(), 10.0),
];
expected.sort_by(|a, b| a.0.cmp(&b.0));
assert_eq!(
got, expected,
"counts are cumulative and the last boundary is `+Inf`, spelled the \
way a dashboard written years ago spells it"
);
for (labels, _) in &series {
assert_eq!(labels["__name__"], "lat_bucket");
assert_eq!(labels["job"], "api", "the original tags are kept");
}
db.close().unwrap();
}
#[test]
fn a_native_histogram_answers_the_classic_count_and_sum_names() {
let dir = tempfile::tempdir().unwrap();
let (db, at) = classic_fixture(&dir);
assert_eq!(one_value(&db, "lat_count", at), 10.0);
assert_eq!(one_value(&db, "lat_sum", at), 6.5);
assert_eq!(
one_value(&db, "lat_count", at),
one_value(&db, "histogram_count(lat)", at)
);
assert_eq!(
one_value(&db, "lat_sum", at),
one_value(&db, "histogram_sum(lat)", at)
);
db.close().unwrap();
}
#[test]
fn an_le_matcher_selects_one_boundary() {
let dir = tempfile::tempdir().unwrap();
let (db, at) = classic_fixture(&dir);
assert_eq!(one_value(&db, r#"lat_bucket{le="0.5"}"#, at), 3.0);
assert_eq!(one_value(&db, r#"lat_bucket{le="+Inf"}"#, at), 10.0);
assert!(
vector(&db, r#"lat_bucket{le="0.25"}"#, at).is_empty(),
"a boundary this histogram does not have selects nothing"
);
db.close().unwrap();
}
#[test]
fn the_classic_view_and_the_native_one_agree_on_the_bucket() {
let dir = tempfile::tempdir().unwrap();
let (db, at) = classic_fixture(&dir);
let native = one_value(&db, "histogram_quantile(0.5, lat)", at);
let classic = one_value(&db, "histogram_quantile(0.5, lat_bucket)", at);
assert!(
(native - classic).abs() < 1e-9,
"the same bytes read two ways: native {native}, classic {classic}"
);
db.close().unwrap();
}
#[test]
fn the_classic_names_are_not_offered_by_discovery() {
let dir = tempfile::tempdir().unwrap();
let (db, _at) = classic_fixture(&dir);
let names: Vec<String> = chronix::promql::all_metrics(db.schema_registry())
.into_iter()
.map(|m| m.name)
.collect();
assert!(names.contains(&"lat".to_string()));
for suffix in ["_bucket", "_count", "_sum"] {
let name = format!("lat{suffix}");
assert!(
!names.contains(&name),
"{name} is a view, not a stored metric, and listing it would make \
`{{__name__=~\".+\"}}` count the same observations twice: {names:?}"
);
}
db.close().unwrap();
}
#[test]
fn a_genuine_count_field_still_resolves_to_itself() {
let dir = tempfile::tempdir().unwrap();
let db = Chronix::open_small(dir.path()).unwrap();
let at = 1_700_000_000_000_000_000;
let key = SeriesKey::new("jobs", tags! { "queue" => "a" }).unwrap();
let mut fields: BTreeMap<String, FieldValue> = BTreeMap::new();
fields.insert("count".into(), FieldValue::F64(7.0));
db.insert(&Point::new(key, fields, at).unwrap()).unwrap();
db.flush().unwrap();
assert_eq!(one_value(&db, "jobs_count", at), 7.0);
db.close().unwrap();
}
#[test]
fn a_float_metric_has_no_classic_view() {
let dir = tempfile::tempdir().unwrap();
let db = Chronix::open_small(dir.path()).unwrap();
let at = 1_700_000_000_000_000_000;
let key = SeriesKey::new("cpu", tags! { "host" => "a" }).unwrap();
let mut fields: BTreeMap<String, FieldValue> = BTreeMap::new();
fields.insert("value".into(), FieldValue::F64(42.0));
db.insert(&Point::new(key, fields, at).unwrap()).unwrap();
db.flush().unwrap();
assert!(
vector(&db, "cpu_bucket", at).is_empty(),
"a gauge has no buckets, and inventing one is worse than an empty result"
);
db.close().unwrap();
}
#[test]
fn the_le_values_are_the_schemas_boundaries_not_the_ones_asked_for() {
let dir = tempfile::tempdir().unwrap();
let db = Chronix::open_small(dir.path()).unwrap();
let at = 1_700_000_000_000_000_000;
let mut h = Histogram::empty(0);
h.positive = vec![
Bucket {
index: 1,
count: 2.0,
},
Bucket {
index: 2,
count: 3.0,
},
];
h.count = 5.0;
h.sum = 12.0;
h.validate().unwrap();
let key = SeriesKey::new("exp", tags! {}).unwrap();
let mut fields: BTreeMap<String, FieldValue> = BTreeMap::new();
fields.insert("value".into(), FieldValue::Histogram(Box::new(h)));
db.insert(&Point::new(key, fields, at).unwrap()).unwrap();
db.flush().unwrap();
let mut les: Vec<String> = vector(&db, "exp_bucket", at)
.into_iter()
.map(|(l, _)| l["le"].clone())
.collect();
les.sort();
let mut expected = vec![
"2".to_string(), "4".to_string(), "+Inf".to_string(),
];
expected.sort();
assert_eq!(
les, expected,
"the boundaries are the schema's, computed from `(2^(2^-0))^i`"
);
assert!(
vector(&db, r#"exp_bucket{le="3"}"#, at).is_empty(),
"an exponential histogram has no bucket ending at 3, and answering \
with the nearest one would be a quantile nobody asked for"
);
let q = one_value(&db, "histogram_quantile(0.5, exp_bucket)", at);
assert!(
(2.0..=4.0).contains(&q),
"the median of counts 2 then 3 over (…,2] (2,4] is in the second \
bucket, got {q}"
);
db.close().unwrap();
}