Skip to main content

mira_core/
series.rs

1//! The metrics read path: match metrics, gather points, group into series.
2//!
3//! A different shape from [`crate::query::search`], and deliberately a separate
4//! function rather than a mode of it. A log search filters one root table and
5//! returns rows newest-first; a metrics query filters a *descriptor* table,
6//! gathers points from four point tables that share one id space, and then
7//! regroups them by a key that has to be stable across blocks. Folding both into
8//! one code path would mean inventing the abstraction that both are special
9//! cases of, which is a planner.
10//!
11//! The grouping key is the hard part and the reason this is not trivial. Ids are
12//! rebased per block — that is what makes the attribute joins array stores — so
13//! nothing block-local can identify a series across two blocks. The key is built
14//! from values instead: the metric's name, unit and kind, plus the merged
15//! attribute map from all four levels. That is one string built per point, which
16//! is the honest cost of a layout optimized for writing and pruning rather than
17//! for grouping.
18//!
19//! ## What V0 returns
20//!
21//! Gauges and sums come back as their own value. Histograms, exponential
22//! histograms and summaries come back as two derived series each, `<name>.count`
23//! and `<name>.sum` — the same convention Prometheus uses, and the same two
24//! numbers that answer "how often" and "how much". Bucket and quantile maths is
25//! not here: it is a heatmap feature, it needs the UI to exist first, and
26//! shipping it wrong would be worse than shipping it later. Everything needed
27//! for it is on disk already — `bucket_counts`, `bounds_id`, `scale`, `quantile`
28//! — so this is a read-path gap, not a storage one.
29//!
30//! Blocking: mmaps and page-faults, same as `search`. Callers on an async
31//! runtime must use `spawn_blocking`.
32
33use std::collections::{BTreeMap, BTreeSet, HashMap};
34use std::path::Path;
35use std::sync::Arc;
36
37use arrow_array::cast::AsArray;
38use arrow_array::types::{
39    Float64Type, Int64Type, TimestampNanosecondType, UInt8Type, UInt16Type, UInt32Type, UInt64Type,
40};
41use arrow_array::{Array, RecordBatch};
42
43use crate::block::{self, Src};
44use crate::error::Result;
45use crate::json::Json;
46use crate::query::{Attrs, Results, Stats, Term, attr_key, attr_parents, dict_index, emit_attr};
47use crate::schema::MetricKind;
48use crate::signal::Open;
49
50/// A metrics query.
51///
52/// Note what is missing: no `step`, no aggregation, no rate. Downsampling in the
53/// engine would need a fill policy, an alignment rule and a choice of aggregator
54/// per metric kind, all of which are presentation decisions the caller is better
55/// placed to make and all of which are wrong for somebody. Raw points, bounded
56/// by `max_points`, and the chart decides.
57#[derive(Debug, Clone)]
58pub struct SeriesQuery {
59    /// Exact metric name. `None` matches every metric, which is what the name
60    /// listing uses and what an exploratory query does before it knows better.
61    pub name: Option<String>,
62    /// Inclusive nanosecond bounds.
63    pub from: i64,
64    pub to: i64,
65    /// Attribute filters. A term is satisfied if it matches at *any* level —
66    /// resource, scope, metric or data point — because a caller filtering on
67    /// `service.name` should not have to know which of the four carries it.
68    pub terms: Vec<Term>,
69    pub max_series: usize,
70    /// Per series, not in total: truncating one busy series must not silently
71    /// empty the others charted beside it.
72    pub max_points: usize,
73}
74
75/// One point. Integers stay integers: an OTLP `as_int` is `sfixed64`, and a
76/// counter past 2^53 pushed through an f64 loses its low bits — which is exactly
77/// the moment a counter is interesting.
78#[derive(Debug, Clone, Copy)]
79enum Pt {
80    Int(i64),
81    Double(f64),
82}
83
84struct Series {
85    /// Rendered `"name":...,"unit":...,` etc. Built once when the series is
86    /// first seen; identical for every point in it by construction.
87    desc: String,
88    /// Rendered `{...}` attribute object.
89    attrs: String,
90    points: Vec<(i64, Pt)>,
91    /// Points dropped by `max_points`, reported rather than hidden. A chart
92    /// missing its spike because the engine quietly truncated is the failure
93    /// mode this exists to prevent.
94    dropped: usize,
95    /// The trace ids behind the points: `(time, rendered object)`, rendered
96    /// here because the block they were read from is unmapped before the
97    /// response is written.
98    exemplars: Vec<(i64, String)>,
99}
100
101impl Series {
102    /// Discard all but the newest `max` points.
103    ///
104    /// Which points a cap keeps is a correctness question, not a tuning one.
105    /// Points arrive in block-scan order, so refusing them once the buffer is
106    /// full kept whichever ones the directory listing happened to reach first
107    /// — an arbitrary subset that the sort at render time then dressed up as a
108    /// contiguous series. Newest-wins is the same rule `limit` already uses on
109    /// the record side, and it is a window a reader can reason about.
110    ///
111    /// Called at twice `max`, so every call throws away half of what it sorts
112    /// and the amortised cost is constant per point. A bounded heap would hold
113    /// exactly `max` at all times and pay `log max` on every point instead;
114    /// at these sizes the occasional sort is cheaper and much less code.
115    ///
116    /// ponytail: truncation, not downsampling. A chart that needs the whole
117    /// range at a lower resolution needs an aggregation window — until that
118    /// exists the response says `dropped_points` rather than pretending.
119    fn compact(&mut self, max: usize) {
120        // Descending, so `truncate` keeps the newest.
121        self.points.sort_unstable_by_key(|p| std::cmp::Reverse(p.0));
122        self.dropped += self.points.len() - max;
123        self.points.truncate(max);
124    }
125}
126
127/// Exemplars kept per series.
128///
129/// An exemplar is one sampled measurement per collection interval per bucket,
130/// so a well-behaved exporter sends few — but nothing in OTLP bounds it, and an
131/// unbounded array here would make one misconfigured SDK able to inflate every
132/// chart response. ponytail: a flat cap, not a reservoir sample; the first ones
133/// in a window are as good a sample as any until someone proves otherwise.
134const MAX_EXEMPLARS: usize = 64;
135
136/// One attribute as `(key, rendered JSON value)`. Rendered once and carried as
137/// a string because the same value is written into both the grouping key and
138/// the response.
139type Attr = (String, String);
140
141/// A descriptor row's cached contribution: its rendered JSON members, and the
142/// attributes every point beneath it inherits.
143type Prefix = (String, Vec<Attr>);
144
145/// The four point tables, and how many name suffixes each contributes.
146///
147/// One table per OTLP point type rather than one wide table: measured, the split
148/// layout is 2.34x smaller (73.0 vs 170.6 bytes/point), because a wide table
149/// pays for every column of every type on every row.
150const DP_TABLES: [&str; 4] = ["number_dp", "hist_dp", "exp_hist_dp", "summary_dp"];
151
152/// The suffixes a histogram's or summary's name is derived with. Named rather
153/// than written twice, because the name filter has to accept back exactly the
154/// names the renderer hands out and two literal lists would drift apart.
155const COUNT: &str = ".count";
156const SUM: &str = ".sum";
157const DERIVED: [&str; 2] = [COUNT, SUM];
158
159pub fn series(root: &Path, q: &SeriesQuery) -> Result<Results> {
160    series_open(root, q, &[])
161}
162
163/// As [`series`], but also reads the open block's snapshot — see
164/// [`crate::query::search_open`], which this mirrors exactly.
165pub fn series_open(root: &Path, q: &SeriesQuery, open: &[Arc<Open>]) -> Result<Results> {
166    let disk = block::scan(root, "metrics")?;
167    let refs = block::sources(&disk, open);
168    let mut stats = Stats {
169        blocks_total: refs.len(),
170        ..Default::default()
171    };
172    let mut out: BTreeMap<String, Series> = BTreeMap::new();
173    let mut dropped: BTreeSet<String> = BTreeSet::new();
174
175    for bref in &refs {
176        // The directory name is the whole index; a block outside the window is
177        // never opened.
178        if bref.max_ts < q.from || bref.min_ts > q.to {
179            continue;
180        }
181        stats.blocks_scanned += 1;
182        collect_block(bref, q, &mut out, &mut dropped, &mut stats)?;
183    }
184    stats.dropped_series = dropped.len();
185
186    // The last compaction of each series, and the only one for a series that
187    // never reached the trigger. Sorting ascending here as well means the
188    // render below reads `points` directly instead of cloning it.
189    for s in out.values_mut() {
190        if s.points.len() > q.max_points {
191            s.compact(q.max_points);
192        }
193        s.points.sort_unstable_by_key(|p| p.0);
194    }
195
196    // A `BTreeMap`, so this is already in key order: two identical queries
197    // produce byte-identical responses, which an ETag, a diff and a cache all
198    // depend on and which a `HashMap` iteration order does not provide. It is
199    // also already at most `max_series` long — see [`bound`].
200    let mut j = Json::new();
201    j.arr(|j| {
202        for s in out.values() {
203            let pts = &s.points;
204            j.obj(|j| {
205                j.raw(&s.desc);
206                j.key("attributes");
207                j.raw(&s.attrs);
208                if s.dropped > 0 {
209                    j.key("dropped_points");
210                    j.u64(s.dropped as u64);
211                }
212                j.key("points");
213                j.arr(|j| {
214                    for &(ts, v) in pts {
215                        j.arr(|j| {
216                            // Both 64-bit halves go out as strings, for the
217                            // reason [`Json::i64_str`] gives: a nanosecond
218                            // timestamp is ~1.7e18 and an OTLP `as_int` is an
219                            // `sfixed64`, and a counter past 2^53 read back
220                            // through a double loses exactly the low bits that
221                            // made it worth charting. A double stays a number
222                            // — it was never anything else.
223                            j.i64_str(ts);
224                            match v {
225                                Pt::Int(i) => j.i64_str(i),
226                                Pt::Double(d) => j.f64(d),
227                            }
228                        });
229                    }
230                });
231                // The answer to "which trace made this spike". Omitted rather
232                // than emitted empty, because most series have none and this is
233                // the response a chart polls.
234                if !s.exemplars.is_empty() {
235                    let mut ex = s.exemplars.clone();
236                    ex.sort_unstable_by_key(|e| e.0);
237                    j.key("exemplars");
238                    j.arr(|j| {
239                        for (_, rendered) in &ex {
240                            j.raw(rendered);
241                        }
242                    });
243                }
244            });
245        }
246    });
247    Ok(Results {
248        json: j.into_string(),
249        stats,
250        // Metrics are bounded by `max_series` and `max_points`, which cap what
251        // a chart can render rather than cut a list short. Nothing to page.
252        next: None,
253    })
254}
255
256/// Every metric name present in the window, with its unit and kind.
257///
258/// This is what a UI puts in a dropdown and what an agent reads before writing
259/// its first query, so it is a first-class endpoint rather than something to
260/// derive from an unfiltered `series` call — the descriptor table is tens of
261/// rows per block, while the points it describes are hundreds of thousands.
262pub fn names(root: &Path, from: i64, to: i64) -> Result<Results> {
263    names_open(root, from, to, &[])
264}
265
266/// As [`names`], but also reads the open block's snapshot. A metric name that
267/// has only ever been written to the open block is exactly the one a dropdown
268/// must not omit — it is the new one.
269pub fn names_open(root: &Path, from: i64, to: i64, open: &[Arc<Open>]) -> Result<Results> {
270    let disk = block::scan(root, "metrics")?;
271    let refs = block::sources(&disk, open);
272    let mut stats = Stats {
273        blocks_total: refs.len(),
274        ..Default::default()
275    };
276    let mut seen: HashMap<String, (String, u8)> = HashMap::new();
277
278    for bref in &refs {
279        if bref.max_ts < from || bref.min_ts > to {
280            continue;
281        }
282        stats.blocks_scanned += 1;
283        let Some(m) = load(bref, "metrics")? else {
284            continue;
285        };
286        stats.rows_scanned += m.num_rows();
287        for r in 0..m.num_rows() {
288            let name = dict_str(&m, "name", r).unwrap_or("").to_owned();
289            let unit = dict_str(&m, "unit", r).unwrap_or("").to_owned();
290            let kind = u8_col(&m, "kind", r);
291            seen.entry(name).or_insert((unit, kind));
292        }
293    }
294    stats.rows_matched = seen.len();
295
296    let mut names: Vec<&String> = seen.keys().collect();
297    names.sort_unstable();
298    let mut j = Json::new();
299    j.arr(|j| {
300        for n in names {
301            let (unit, kind) = &seen[n];
302            j.obj(|j| {
303                j.key("name");
304                j.str(n);
305                j.key("unit");
306                j.str(unit);
307                j.key("kind");
308                j.str(kind_name(*kind));
309            });
310        }
311    });
312    Ok(Results {
313        json: j.into_string(),
314        stats,
315        // Metrics are bounded by `max_series` and `max_points`, which cap what
316        // a chart can render rather than cut a list short. Nothing to page.
317        next: None,
318    })
319}
320
321fn kind_name(k: u8) -> &'static str {
322    match k {
323        x if x == MetricKind::Gauge as u8 => "gauge",
324        x if x == MetricKind::Sum as u8 => "sum",
325        x if x == MetricKind::Histogram as u8 => "histogram",
326        x if x == MetricKind::ExponentialHistogram as u8 => "exponential_histogram",
327        x if x == MetricKind::Summary as u8 => "summary",
328        _ => "unset",
329    }
330}
331
332fn load(bref: &Src, name: &str) -> Result<Option<RecordBatch>> {
333    bref.load(name)
334}
335
336/// Child rows grouped by `parent_id`, indexed by it.
337///
338/// Ids are rebased dense from zero per block, so the parent id *is* the slot —
339/// no hash map, and the whole thing is one pass over a `u32` buffer.
340pub(crate) fn index_by_parent(b: &RecordBatch) -> Vec<Vec<u32>> {
341    let Some(col) = b.column_by_name("parent_id") else {
342        return Vec::new();
343    };
344    let parents = col.as_primitive::<UInt32Type>().values();
345    let mut out: Vec<Vec<u32>> =
346        vec![Vec::new(); parents.iter().copied().max().unwrap_or(0) as usize + 1];
347    for (r, &p) in parents.iter().enumerate() {
348        out[p as usize].push(r as u32);
349    }
350    out
351}
352
353fn exemplar_time(b: &RecordBatch, row: u32) -> i64 {
354    b.column_by_name("time_unix_nano").map_or(0, |c| {
355        c.as_primitive::<TimestampNanosecondType>()
356            .value(row as usize)
357    })
358}
359
360fn dict_str<'a>(b: &'a RecordBatch, col: &str, row: usize) -> Option<&'a str> {
361    let c = b.column_by_name(col)?;
362    let d = c.as_dictionary::<UInt16Type>();
363    if d.is_null(row) {
364        return None;
365    }
366    Some(
367        d.values()
368            .as_string::<i32>()
369            .value(d.keys().value(row) as usize),
370    )
371}
372
373fn u8_col(b: &RecordBatch, col: &str, row: usize) -> u8 {
374    b.column_by_name(col)
375        .map_or(0, |c| c.as_primitive::<UInt8Type>().value(row))
376}
377
378/// Hold `out` to `max_series` entries by evicting its largest key.
379///
380/// The result is the same set `sort(keys).truncate(max_series)` produced, and
381/// the argument is short enough to check: the answer is the `max_series`
382/// smallest keys, and a key that belongs in it can never be the largest key of
383/// a map that is already one over — everything else in that map is smaller, so
384/// there are already `max_series` keys ahead of it. Evicting the maximum
385/// therefore only ever discards a key the sort would have truncated. It also
386/// cannot come back: the evicted key is greater than every key in the map, and
387/// the map's maximum only falls, so the same key arriving from a later block is
388/// evicted again on sight.
389///
390/// The difference is *when*, and that is the whole point. The old map grew one
391/// entry per distinct (name, unit, kind, temporality, monotonic, attributes)
392/// tuple until the query finished, each carrying a rendered descriptor, a
393/// rendered attribute object and a key concatenating both — about a kilobyte.
394/// `max_points` bounded the points inside a series and nothing bounded the
395/// series, so a `query_metric` with no `name` over a store with a request id in
396/// a data-point attribute allocated until the process died.
397///
398/// The evicted key is kept, without its payload, so the response can say how
399/// many series it is not showing. ponytail: that record is itself capped at
400/// `max_series` keys, so `dropped_series` saturates rather than counting an
401/// unbounded number of them exactly — remembering every key to count it is the
402/// allocation this function exists to prevent. `dropped_series == max_series`
403/// reads as "at least". Counting evictions instead of distinct keys was the
404/// other option and it is worse: a series above the cap is re-inserted and
405/// re-evicted once per point, so a store with 100 series and a cap of 64 would
406/// report tens of thousands dropped.
407fn bound(out: &mut BTreeMap<String, Series>, dropped: &mut BTreeSet<String>, max: usize) {
408    while out.len() > max {
409        let Some((k, _)) = out.pop_last() else { break };
410        if dropped.len() < max {
411            dropped.insert(k);
412        }
413    }
414}
415
416fn collect_block(
417    bref: &Src,
418    q: &SeriesQuery,
419    out: &mut BTreeMap<String, Series>,
420    dropped: &mut BTreeSet<String>,
421    stats: &mut Stats,
422) -> Result<()> {
423    // A missing descriptor table means retention is unlinking this block under
424    // us. Normal, not an error.
425    let Some(metrics) = load(bref, "metrics")? else {
426        return Ok(());
427    };
428    let n_metrics = metrics.num_rows();
429
430    // Name filter: resolve the string against the dictionary once, then compare
431    // u16 codes. A block whose dictionary lacks the name has no rows to check.
432    //
433    // Above the six loads below, not below them: each is an `mmap` and a CRC of
434    // the whole table body, so a block that does not hold the requested metric
435    // should cost one of them and not six.
436    let mut wanted = vec![true; n_metrics];
437    // Set when the requested name was a derived one, so only that half of the
438    // histogram is emitted rather than both.
439    let mut only_suffix = "";
440    if let Some(want) = &q.name {
441        let d = metrics
442            .column_by_name("name")
443            .map(|c| c.as_dictionary::<UInt16Type>());
444        let Some(d) = d else { return Ok(()) };
445        let names = d.values().as_string::<i32>();
446        // The descriptor dictionary holds `http.server.duration`; this function
447        // hands back series called `http.server.duration.count`. A caller
448        // pasting a name off its own previous answer — a chart legend, an agent
449        // reading the result it just got — must not get a silent empty series
450        // list, so a miss retries against the base name. A metric genuinely
451        // called `foo.count` matches on the first try and keeps winning.
452        let mut code = dict_index(names, want);
453        if code.is_none() {
454            for s in DERIVED {
455                if let Some(base) = want.strip_suffix(s) {
456                    code = dict_index(names, base);
457                    only_suffix = s;
458                    break;
459                }
460            }
461        }
462        let Some(code) = code else { return Ok(()) };
463        let codes = d.keys().values();
464        for (r, w) in wanted.iter_mut().enumerate() {
465            *w = codes[r] == code;
466        }
467    }
468
469    // Wrapped rather than raw, because `own_attrs` below runs once per matched
470    // point and `Attrs` turns each of those from a scan of the whole table into
471    // two binary searches.
472    let metric_attrs = load(bref, "metric_attrs")?.map(Attrs::new);
473    let dp_attrs = load(bref, "dp_attrs")?.map(Attrs::new);
474    // Data point ids are one dense space across all four point tables, so one
475    // list indexed by id serves every table below — the same property that lets
476    // `dp_attrs` carry no discriminant.
477    let exemplars = load(bref, "exemplars")?;
478    let by_point = exemplars.as_ref().map(index_by_parent).unwrap_or_default();
479    let resource_attrs = load(bref, "resource_attrs")?.map(Attrs::new);
480    let scope_attrs = load(bref, "scope_attrs")?.map(Attrs::new);
481
482    // For each term, which metric rows satisfy it *above* the point level. A
483    // point still qualifies if its own attributes satisfy the term, so this is
484    // one half of an OR evaluated per point below.
485    let above: Vec<Vec<bool>> = q
486        .terms
487        .iter()
488        .map(|t| {
489            let key = match &t.target {
490                crate::query::Target::Attr(k) | crate::query::Target::Field(k) => k,
491            };
492            let mut hit = vec![false; n_metrics];
493            // Metric level: parent_id is the descriptor row number.
494            if let Some(a) = &metric_attrs {
495                for pid in attr_parents(&a.rows, key, t.op, &t.value) {
496                    if let Some(s) = hit.get_mut(pid as usize) {
497                        *s = true;
498                    }
499                }
500            }
501            // Resource and scope level: parent_id is an entity id, and the
502            // descriptor row carries the foreign key.
503            for (table, fk) in [(&resource_attrs, "resource_id"), (&scope_attrs, "scope_id")] {
504                let (Some(a), Some(col)) = (table, metrics.column_by_name(fk)) else {
505                    continue;
506                };
507                let ids = attr_parents(&a.rows, key, t.op, &t.value);
508                let Some(&top) = ids.iter().max() else {
509                    continue;
510                };
511                let mut want = vec![false; top as usize + 1];
512                for id in ids {
513                    want[id as usize] = true;
514                }
515                for (r, &id) in col.as_primitive::<UInt16Type>().values().iter().enumerate() {
516                    if want.get(id as usize).copied().unwrap_or(false) {
517                        hit[r] = true;
518                    }
519                }
520            }
521            hit
522        })
523        .collect();
524
525    // Same, at the point level. Data point ids are one shared space across all
526    // four point tables — that is why `dp_attrs` needs no table discriminant —
527    // so one bitmap per term serves every table below.
528    let dp_hit: Vec<Vec<bool>> = q
529        .terms
530        .iter()
531        .map(|t| {
532            let key = match &t.target {
533                crate::query::Target::Attr(k) | crate::query::Target::Field(k) => k,
534            };
535            let mut hit = Vec::new();
536            if let Some(a) = &dp_attrs {
537                for pid in attr_parents(&a.rows, key, t.op, &t.value) {
538                    if hit.len() <= pid as usize {
539                        hit.resize(pid as usize + 1, false);
540                    }
541                    hit[pid as usize] = true;
542                }
543            }
544            hit
545        })
546        .collect();
547
548    // Attributes above the point level, rendered once per descriptor row rather
549    // than once per point. A metric with 10k points would otherwise re-render
550    // its resource attributes 10k times.
551    let mut prefix: Vec<Option<Prefix>> = vec![None; n_metrics];
552
553    for table in DP_TABLES {
554        let Some(dp) = load(bref, table)? else {
555            continue;
556        };
557        let n = dp.num_rows();
558        stats.rows_scanned += n;
559        let (Some(time), Some(mid), Some(did)) = (
560            dp.column_by_name("time_unix_nano")
561                .map(|c| &**c.as_primitive::<TimestampNanosecondType>().values()),
562            dp.column_by_name("metric_id")
563                .map(|c| &**c.as_primitive::<UInt32Type>().values()),
564            dp.column_by_name("id")
565                .map(|c| &**c.as_primitive::<UInt32Type>().values()),
566        ) else {
567            continue;
568        };
569        let vals = Values::for_table(table);
570
571        for r in 0..n {
572            let m = mid[r] as usize;
573            if !(q.from..=q.to).contains(&time[r]) || !wanted.get(m).copied().unwrap_or(false) {
574                continue;
575            }
576            let d = did[r] as usize;
577            if !(0..q.terms.len()).all(|t| {
578                above[t].get(m).copied().unwrap_or(false)
579                    || dp_hit[t].get(d).copied().unwrap_or(false)
580            }) {
581                continue;
582            }
583            stats.rows_matched += 1;
584
585            let (desc_prefix, upper) = prefix[m].get_or_insert_with(|| {
586                (
587                    describe(&metrics, m),
588                    upper_attrs(&metrics, m, &metric_attrs, &resource_attrs, &scope_attrs),
589                )
590            });
591            // The point's own attributes, merged over the inherited ones.
592            let own = own_attrs(&dp_attrs, d as u32);
593            let attrs = merge(upper, &own);
594
595            for (suffix, v) in vals.at(&dp, r) {
596                if !only_suffix.is_empty() && suffix != only_suffix {
597                    continue;
598                }
599                let key = format!("{desc_prefix}\u{1}{suffix}\u{1}{attrs}");
600                let s = out.entry(key).or_insert_with(|| Series {
601                    desc: with_suffix(desc_prefix, suffix),
602                    attrs: attrs.clone(),
603                    points: Vec::new(),
604                    dropped: 0,
605                    exemplars: Vec::new(),
606                });
607                s.points.push((time[r], v));
608                if s.points.len() >= 2 * q.max_points {
609                    s.compact(q.max_points);
610                }
611                // A histogram yields two derived series from one point, and its
612                // exemplars belong to both: whichever of `.count` and `.sum` is
613                // charted, the spike in it points at the same traces.
614                if let (Some(ex), Some(rows)) = (&exemplars, by_point.get(d)) {
615                    for &er in rows {
616                        if s.exemplars.len() >= MAX_EXEMPLARS {
617                            break;
618                        }
619                        let mut j = Json::new();
620                        j.obj(|j| crate::query::emit_fields(j, ex, er));
621                        s.exemplars.push((exemplar_time(ex, er), j.into_string()));
622                    }
623                }
624                // Applied per point, not once at the end: the whole reason
625                // `max_series` exists is that the map between here and the end
626                // is what runs the process out of memory.
627                bound(out, dropped, q.max_series);
628            }
629        }
630    }
631    Ok(())
632}
633
634/// The descriptor's identity as rendered JSON members, without the trailing
635/// comma. Doubles as the stable half of the series key.
636fn describe(metrics: &RecordBatch, row: usize) -> String {
637    let mut j = Json::new();
638    j.key("name");
639    j.str(dict_str(metrics, "name", row).unwrap_or(""));
640    j.key("unit");
641    j.str(dict_str(metrics, "unit", row).unwrap_or(""));
642    j.key("kind");
643    j.str(kind_name(u8_col(metrics, "kind", row)));
644    j.key("temporality");
645    j.u64(u8_col(metrics, "temporality", row) as u64);
646    j.key("monotonic");
647    j.bool(
648        metrics
649            .column_by_name("is_monotonic")
650            .is_some_and(|c| c.as_boolean().value(row)),
651    );
652    j.into_string()
653}
654
655/// `describe` with the derived-series suffix folded into the name, so a
656/// `.count` series reports the name a caller can query it back by.
657///
658/// That is only true because `collect_block` strips a [`DERIVED`] suffix when
659/// the descriptor dictionary does not hold the requested name. Emitting a name
660/// here that the filter there does not accept is the same bug as returning a
661/// cursor nobody can page with.
662fn with_suffix(desc: &str, suffix: &str) -> String {
663    if suffix.is_empty() {
664        return desc.to_owned();
665    }
666    // `describe` always emits `"name":"..."` first, so the closing quote of the
667    // name is the second unescaped quote after the colon. Splitting on the known
668    // prefix is cheaper and less fragile than re-rendering.
669    match desc.find("\",\"unit\"") {
670        Some(i) => format!("{}{suffix}{}", &desc[..i], &desc[i..]),
671        None => desc.to_owned(),
672    }
673}
674
675/// Attributes from the metric, resource and scope levels, sorted and deduped
676/// with the most specific level winning.
677fn upper_attrs(
678    metrics: &RecordBatch,
679    row: usize,
680    metric_attrs: &Option<Attrs>,
681    resource_attrs: &Option<Attrs>,
682    scope_attrs: &Option<Attrs>,
683) -> Vec<Attr> {
684    let fk = |name: &str| {
685        metrics
686            .column_by_name(name)
687            .map(|c| c.as_primitive::<UInt16Type>().value(row) as u32)
688    };
689    let mut v = Vec::new();
690    // Least specific first: `merge_into` keeps the last write per key.
691    for (table, parent) in [
692        (resource_attrs, fk("resource_id")),
693        (scope_attrs, fk("scope_id")),
694        (metric_attrs, Some(row as u32)),
695    ] {
696        let (Some(a), Some(p)) = (table, parent) else {
697            continue;
698        };
699        collect_attrs(a, p, &mut v);
700    }
701    dedup_last(&mut v);
702    v
703}
704
705fn own_attrs(dp_attrs: &Option<Attrs>, dp_id: u32) -> Vec<Attr> {
706    let mut v = Vec::new();
707    if let Some(a) = dp_attrs {
708        collect_attrs(a, dp_id, &mut v);
709    }
710    dedup_last(&mut v);
711    v
712}
713
714/// The rows one parent owns, rendered.
715///
716/// This runs once per matched point, so the search matters: `dp_attrs` on a
717/// block holding 7,752 points is thousands of rows, and scanning all of them
718/// per point is the quadratic shape [`Attrs`] exists to remove. Empty for a
719/// table whose parent column is not a `u32`, matching the read path — no schema
720/// in this tree writes one, and a block someone else wrote should not panic.
721fn collect_attrs(a: &Attrs, parent: u32, out: &mut Vec<Attr>) {
722    let parents = Attrs::parents(&a.rows).unwrap_or_default();
723    for r in a.run(parents, parent).filter(|&r| parents[r] == parent) {
724        let mut j = Json::new();
725        emit_attr(&mut j, &a.rows, r);
726        out.push((attr_key(&a.rows, r).to_owned(), j.into_string()));
727    }
728}
729
730fn dedup_last(v: &mut Vec<Attr>) {
731    // Stable, so entries pushed later — from the more specific level — sort
732    // after their earlier namesakes and win the dedup.
733    v.sort_by(|a, b| a.0.cmp(&b.0));
734    v.dedup_by(|a, b| {
735        if a.0 == b.0 {
736            std::mem::swap(a, b);
737            true
738        } else {
739            false
740        }
741    });
742}
743
744/// Render the union of two sorted attribute lists as a JSON object, `own`
745/// winning on collision. A merge, not a concatenation and a re-sort: `upper` is
746/// already sorted and shared by every point of the metric.
747fn merge(upper: &[Attr], own: &[Attr]) -> String {
748    let mut j = Json::new();
749    j.obj(|j| {
750        let (mut i, mut k) = (0, 0);
751        while i < upper.len() || k < own.len() {
752            let take_own = match (upper.get(i), own.get(k)) {
753                (Some(u), Some(o)) => {
754                    if u.0 == o.0 {
755                        i += 1;
756                    }
757                    o.0 <= u.0
758                }
759                (None, Some(_)) => true,
760                _ => false,
761            };
762            let (key, val) = if take_own {
763                k += 1;
764                &own[k - 1]
765            } else {
766                i += 1;
767                &upper[i - 1]
768            };
769            j.key(key);
770            j.raw(val);
771        }
772    });
773    j.into_string()
774}
775
776/// Which columns of a point table carry chartable values, resolved once per
777/// table instead of per row.
778enum Values {
779    /// `int` and `double`, exactly one set per row.
780    Number,
781    /// `count` and `sum`, emitted as `<name>.count` and `<name>.sum`.
782    CountSum,
783    None,
784}
785
786impl Values {
787    fn for_table(table: &str) -> Values {
788        match table {
789            "number_dp" => Values::Number,
790            "hist_dp" | "exp_hist_dp" | "summary_dp" => Values::CountSum,
791            _ => Values::None,
792        }
793    }
794
795    fn at(&self, dp: &RecordBatch, r: usize) -> Vec<(&'static str, Pt)> {
796        match self {
797            Values::Number => {
798                if let Some(c) = dp.column_by_name("int") {
799                    let a = c.as_primitive::<Int64Type>();
800                    if !a.is_null(r) {
801                        return vec![("", Pt::Int(a.value(r)))];
802                    }
803                }
804                if let Some(c) = dp.column_by_name("double") {
805                    let a = c.as_primitive::<Float64Type>();
806                    if !a.is_null(r) {
807                        return vec![("", Pt::Double(a.value(r)))];
808                    }
809                }
810                Vec::new()
811            }
812            Values::CountSum => {
813                let mut v = Vec::with_capacity(2);
814                if let Some(c) = dp.column_by_name("count") {
815                    let a = c.as_primitive::<UInt64Type>();
816                    if !a.is_null(r) {
817                        v.push((COUNT, Pt::Int(a.value(r) as i64)));
818                    }
819                }
820                if let Some(c) = dp.column_by_name("sum") {
821                    let a = c.as_primitive::<Float64Type>();
822                    if !a.is_null(r) {
823                        v.push((SUM, Pt::Double(a.value(r))));
824                    }
825                }
826                v
827            }
828            Values::None => Vec::new(),
829        }
830    }
831}
832
833#[cfg(test)]
834mod tests {
835    use super::*;
836    use arrow_array::{ArrayRef, Float64Array, Int64Array, UInt32Array, UInt64Array};
837    use mira_proto::collector::metrics::v1::ExportMetricsServiceRequest;
838    use mira_proto::common::v1::any_value::Value as AnyVal;
839    use mira_proto::common::v1::{AnyValue, InstrumentationScope, KeyValue};
840    use mira_proto::metrics::v1::metric::Data;
841    use mira_proto::metrics::v1::number_data_point::Value as NumValue;
842    use mira_proto::metrics::v1::{
843        Exemplar, ExponentialHistogram, ExponentialHistogramDataPoint, Gauge, Histogram,
844        HistogramDataPoint, Metric, NumberDataPoint, ResourceMetrics, ScopeMetrics, Sum, Summary,
845        SummaryDataPoint, exemplar,
846    };
847    use mira_proto::resource::v1::Resource;
848
849    /// A named temporary directory, emptied first so a crashed run does not
850    /// hand the next one a half-written store.
851    fn tmp(name: &str) -> std::path::PathBuf {
852        let d = std::env::temp_dir().join(format!("mira-{name}-{}", std::process::id()));
853        let _ = std::fs::remove_dir_all(&d);
854        d
855    }
856
857    fn kv(k: &str, v: &str) -> KeyValue {
858        KeyValue {
859            key: k.into(),
860            value: Some(AnyValue {
861                value: Some(AnyVal::StringValue(v.into())),
862            }),
863        }
864    }
865
866    /// Publish `metrics` as one block, and hand back its directory so a test
867    /// can damage it the way retention or an older writer would.
868    fn publish(
869        dir: &std::path::Path,
870        seq: u64,
871        resource: Option<Resource>,
872        scope: Option<InstrumentationScope>,
873        metrics: Vec<Metric>,
874    ) -> std::path::PathBuf {
875        let req = ExportMetricsServiceRequest {
876            resource_metrics: vec![ResourceMetrics {
877                resource,
878                scope_metrics: vec![ScopeMetrics {
879                    scope,
880                    metrics,
881                    ..Default::default()
882                }],
883                ..Default::default()
884            }],
885        };
886        let mut b = crate::metrics::MetricsBuilder::new();
887        b.append_request(&req).unwrap();
888        let sealed = b.finish().unwrap();
889        crate::block::publish(dir, "metrics", crate::block::node_id("a"), seq, 0, &sealed)
890            .unwrap()
891            .dir
892    }
893
894    /// A gauge of one integer point per timestamp, the simplest thing that
895    /// produces a series.
896    fn gauge(name: &str, points: &[(u64, i64)]) -> Metric {
897        Metric {
898            name: name.into(),
899            data: Some(Data::Gauge(Gauge {
900                data_points: points
901                    .iter()
902                    .map(|&(t, v)| NumberDataPoint {
903                        time_unix_nano: t,
904                        value: Some(NumValue::AsInt(v)),
905                        ..Default::default()
906                    })
907                    .collect(),
908            })),
909            ..Default::default()
910        }
911    }
912
913    fn query(dir: &std::path::Path, q: &SeriesQuery) -> Results {
914        series(dir, q).unwrap()
915    }
916
917    /// The query every test below varies one field of.
918    fn wide() -> SeriesQuery {
919        SeriesQuery {
920            name: None,
921            from: 0,
922            to: i64::MAX,
923            terms: Vec::new(),
924            max_series: 100,
925            max_points: 100,
926        }
927    }
928
929    /// A histogram comes back as two series called `<name>.count` and
930    /// `<name>.sum`, and those are the names a chart legend shows and an agent
931    /// reads off its own previous answer. Matching only the base name against
932    /// the descriptor dictionary made that round trip return an empty list with
933    /// a 200, which reads as "the metric stopped reporting".
934    #[test]
935    fn a_derived_series_name_queries_back_to_the_series_it_names() {
936        let dir = std::env::temp_dir().join(format!("mira-derived-{}", std::process::id()));
937        let _ = std::fs::remove_dir_all(&dir);
938
939        let req = ExportMetricsServiceRequest {
940            resource_metrics: vec![ResourceMetrics {
941                scope_metrics: vec![ScopeMetrics {
942                    metrics: vec![
943                        Metric {
944                            name: "http.server.duration".into(),
945                            data: Some(Data::Histogram(Histogram {
946                                data_points: vec![HistogramDataPoint {
947                                    time_unix_nano: 1_000,
948                                    count: 3,
949                                    sum: Some(1.5),
950                                    ..Default::default()
951                                }],
952                                ..Default::default()
953                            })),
954                            ..Default::default()
955                        },
956                        // A gauge whose name already ends in `.count`. The base
957                        // name has to win the lookup, or this metric becomes
958                        // unreachable the moment the stripping is added.
959                        Metric {
960                            name: "queue.depth.count".into(),
961                            data: Some(Data::Gauge(Gauge {
962                                data_points: vec![NumberDataPoint {
963                                    time_unix_nano: 1_000,
964                                    value: Some(NumValue::AsInt(42)),
965                                    ..Default::default()
966                                }],
967                            })),
968                            ..Default::default()
969                        },
970                    ],
971                    ..Default::default()
972                }],
973                ..Default::default()
974            }],
975        };
976
977        let mut b = crate::metrics::MetricsBuilder::new();
978        b.append_request(&req).unwrap();
979        let sealed = b.finish().unwrap();
980        crate::block::publish(&dir, "metrics", crate::block::node_id("a"), 0, 0, &sealed).unwrap();
981
982        let ask = |name: &str| {
983            series(
984                &dir,
985                &SeriesQuery {
986                    name: Some(name.into()),
987                    from: 0,
988                    to: 10_000,
989                    terms: Vec::new(),
990                    max_series: 100,
991                    max_points: 100,
992                },
993            )
994            .unwrap()
995            .json
996        };
997
998        // The base name still returns both halves.
999        let both = ask("http.server.duration");
1000        for half in [COUNT, SUM] {
1001            assert!(
1002                both.contains(&format!(r#""name":"http.server.duration{half}""#)),
1003                "{both}"
1004            );
1005        }
1006
1007        // Each derived name returns exactly the series it names, and only it.
1008        let c = ask("http.server.duration.count");
1009        assert!(c.contains(r#""name":"http.server.duration.count""#), "{c}");
1010        assert!(!c.contains(SUM), "{c}");
1011        // Both halves of a point are strings when the value is an integer: the
1012        // timestamp is nanoseconds and a `.count` is a `uint64`, so both are
1013        // 64-bit and both go out the OTLP/JSON way. `.sum` below is a double
1014        // and stays a bare number, which is what makes this pair worth having.
1015        assert!(c.contains(r#"["1000","3"]"#), "{c}");
1016        let s = ask("http.server.duration.sum");
1017        assert!(s.contains(r#""name":"http.server.duration.sum""#), "{s}");
1018        assert!(!s.contains(COUNT), "{s}");
1019        assert!(s.contains(r#"["1000",1.5]"#), "{s}");
1020
1021        // A metric that really is called `x.count` matches before the suffix is
1022        // stripped.
1023        let q = ask("queue.depth.count");
1024        assert!(q.contains(r#""name":"queue.depth.count""#), "{q}");
1025        assert!(q.contains(r#"["1000","42"]"#), "{q}");
1026
1027        // The retry widens the lookup; it does not make it match anything. A
1028        // derived name on a metric that has no derived series is still empty,
1029        // and so is a base name nothing carries.
1030        assert_eq!(ask("queue.depth.count.sum"), "[]");
1031        assert_eq!(ask("nope.count"), "[]");
1032
1033        let _ = std::fs::remove_dir_all(&dir);
1034    }
1035
1036    /// `max_series` bounds the map, not just the render.
1037    ///
1038    /// A high-cardinality attribute — a request id, a customer id, a pod name in
1039    /// a cluster that recycles them — makes one metric name expand into a series
1040    /// per distinct value. Accumulating them all and truncating at the end meant
1041    /// the cap the caller set had no effect on the memory the query took, and
1042    /// the store is the one thing here with no bound on how many values it
1043    /// holds.
1044    #[test]
1045    fn the_series_cap_bounds_the_map_and_says_what_it_refused() {
1046        let dir = std::env::temp_dir().join(format!("mira-cap-{}", std::process::id()));
1047        let _ = std::fs::remove_dir_all(&dir);
1048
1049        // Ten series of one metric, distinguished only by an attribute, and
1050        // named so that key order is the order a reader would guess.
1051        let req = ExportMetricsServiceRequest {
1052            resource_metrics: vec![ResourceMetrics {
1053                scope_metrics: vec![ScopeMetrics {
1054                    metrics: vec![Metric {
1055                        name: "rpc.duration".into(),
1056                        data: Some(Data::Gauge(Gauge {
1057                            data_points: (0..10)
1058                                .map(|i| NumberDataPoint {
1059                                    time_unix_nano: 1_000,
1060                                    value: Some(NumValue::AsInt(i)),
1061                                    attributes: vec![mira_proto::common::v1::KeyValue {
1062                                        key: "peer".into(),
1063                                        value: Some(mira_proto::common::v1::AnyValue {
1064                                            value: Some(
1065                                                mira_proto::common::v1::any_value::Value::StringValue(
1066                                                    format!("p{i}"),
1067                                                ),
1068                                            ),
1069                                        }),
1070                                    }],
1071                                    ..Default::default()
1072                                })
1073                                .collect(),
1074                        })),
1075                        ..Default::default()
1076                    }],
1077                    ..Default::default()
1078                }],
1079                ..Default::default()
1080            }],
1081        };
1082
1083        let mut b = crate::metrics::MetricsBuilder::new();
1084        b.append_request(&req).unwrap();
1085        let sealed = b.finish().unwrap();
1086        crate::block::publish(&dir, "metrics", crate::block::node_id("a"), 0, 0, &sealed).unwrap();
1087
1088        let ask = |max_series: usize| {
1089            series(
1090                &dir,
1091                &SeriesQuery {
1092                    name: None,
1093                    from: 0,
1094                    to: 10_000,
1095                    terms: Vec::new(),
1096                    max_series,
1097                    max_points: 100,
1098                },
1099            )
1100            .unwrap()
1101        };
1102
1103        // Under the cap: everything, and nothing refused.
1104        let r = ask(100);
1105        assert_eq!(r.json.matches(r#""peer""#).count(), 10, "{}", r.json);
1106        assert_eq!(r.stats.dropped_series, 0);
1107
1108        // Over it: the smallest keys survive, which is the same set the old
1109        // sort-then-truncate produced — that equivalence is the whole safety
1110        // argument for evicting the maximum as we go, so it is what this checks
1111        // rather than the count alone.
1112        let r = ask(8);
1113        for i in 0..10 {
1114            assert_eq!(
1115                r.json.contains(&format!(r#""peer":"p{i}""#)),
1116                i < 8,
1117                "p{i}: {}",
1118                r.json
1119            );
1120        }
1121        assert_eq!(r.stats.dropped_series, 2);
1122
1123        // The ponytail ceiling, asserted rather than left to be discovered: the
1124        // record of what was refused is itself capped at `max_series`, so with
1125        // six dropped and a cap of four the answer is four and reads as "at
1126        // least". Remembering every key exactly is the allocation the cap
1127        // exists to prevent.
1128        let r = ask(4);
1129        assert_eq!(r.json.matches(r#""peer""#).count(), 4, "{}", r.json);
1130        assert_eq!(r.stats.dropped_series, 4);
1131
1132        let _ = std::fs::remove_dir_all(&dir);
1133    }
1134
1135    /// A point's value lives in whichever column its table set, and a row that
1136    /// set none must yield no point rather than a zero. A gauge with no `value`
1137    /// on the wire means "not reported"; charting it as 0 invents a reading
1138    /// nobody took, and on a `.sum` it would drag an average down.
1139    ///
1140    /// The null cases are built by hand because today's writer never emits
1141    /// them — `count` is non-nullable and every gauge point has one of `int`
1142    /// and `double`. The reader is the half that outlives the writer that wrote
1143    /// the block, so it is the half that has to survive them.
1144    #[test]
1145    fn a_point_yields_the_value_of_whichever_column_its_table_actually_set() {
1146        for (table, want) in [
1147            ("number_dp", "number"),
1148            ("hist_dp", "countsum"),
1149            ("exp_hist_dp", "countsum"),
1150            ("summary_dp", "countsum"),
1151            // Not a point table. `collect_block` only ever asks about the four,
1152            // so this arm is what stops a fifth table added later from being
1153            // charted as whatever its first two columns happen to be called.
1154            ("logs", "none"),
1155        ] {
1156            let got = match Values::for_table(table) {
1157                Values::Number => "number",
1158                Values::CountSum => "countsum",
1159                Values::None => "none",
1160            };
1161            assert_eq!(got, want, "{table}");
1162        }
1163
1164        /// `(suffix, value)` as text, because `Pt` is deliberately not `Eq` —
1165        /// comparing two charted doubles for equality is a bug everywhere else.
1166        fn shape(v: Vec<(&'static str, Pt)>) -> Vec<String> {
1167            v.into_iter()
1168                .map(|(s, p)| match p {
1169                    Pt::Int(i) => format!("{s}=i{i}"),
1170                    Pt::Double(d) => format!("{s}=d{d}"),
1171                })
1172                .collect()
1173        }
1174
1175        let num = RecordBatch::try_from_iter(vec![
1176            (
1177                "int",
1178                Arc::new(Int64Array::from(vec![Some(7), None, None])) as ArrayRef,
1179            ),
1180            (
1181                "double",
1182                Arc::new(Float64Array::from(vec![None, Some(0.5), None])) as ArrayRef,
1183            ),
1184        ])
1185        .unwrap();
1186        assert_eq!(shape(Values::Number.at(&num, 0)), ["=i7"]);
1187        assert_eq!(shape(Values::Number.at(&num, 1)), ["=d0.5"]);
1188        assert!(shape(Values::Number.at(&num, 2)).is_empty());
1189
1190        // A writer that never emitted the column at all, rather than emitting
1191        // it null: the reader must fall through to the other one, not index a
1192        // column that is not there.
1193        let only_double = RecordBatch::try_from_iter(vec![(
1194            "double",
1195            Arc::new(Float64Array::from(vec![1.25])) as ArrayRef,
1196        )])
1197        .unwrap();
1198        assert_eq!(shape(Values::Number.at(&only_double, 0)), ["=d1.25"]);
1199
1200        let cs = RecordBatch::try_from_iter(vec![
1201            (
1202                "count",
1203                Arc::new(UInt64Array::from(vec![None, Some(4), Some(4), None])) as ArrayRef,
1204            ),
1205            (
1206                "sum",
1207                Arc::new(Float64Array::from(vec![Some(1.5), None, Some(2.5), None])) as ArrayRef,
1208            ),
1209        ])
1210        .unwrap();
1211        assert_eq!(shape(Values::CountSum.at(&cs, 0)), [".sum=d1.5"]);
1212        assert_eq!(shape(Values::CountSum.at(&cs, 1)), [".count=i4"]);
1213        assert_eq!(
1214            shape(Values::CountSum.at(&cs, 2)),
1215            [".count=i4", ".sum=d2.5"]
1216        );
1217        assert!(shape(Values::CountSum.at(&cs, 3)).is_empty());
1218
1219        // The half of a histogram that is missing is missing, not zero: a
1220        // `sum`-less point charts a `.count` series and no `.sum` series at
1221        // all. `starts_with`, not `==`: `shape` renders the value into the
1222        // string, so an equality against the bare suffix could never match and
1223        // would hold however many `.sum` points came back.
1224        assert!(
1225            !shape(Values::CountSum.at(&cs, 1))
1226                .iter()
1227                .any(|s| s.starts_with(SUM))
1228        );
1229        assert!(shape(Values::None.at(&cs, 2)).is_empty());
1230    }
1231
1232    /// Every kind OTLP can declare gets a name a caller can read, including the
1233    /// two that were added last and the descriptor with no `data` at all.
1234    ///
1235    /// This is the dropdown an agent reads before writing its first query. A
1236    /// kind rendered as the wrong word — or an exponential histogram rendered
1237    /// as "unset" because the match arm was never added — is a metric the
1238    /// caller cannot tell apart from one that is not reporting.
1239    #[test]
1240    fn every_metric_kind_reports_the_name_a_dropdown_shows() {
1241        let dir = tmp("series-kinds");
1242        let point = NumberDataPoint {
1243            time_unix_nano: 1_000,
1244            value: Some(NumValue::AsInt(1)),
1245            ..Default::default()
1246        };
1247        let named = |name: &str, unit: &str, data: Option<Data>| Metric {
1248            name: name.into(),
1249            unit: unit.into(),
1250            data,
1251            ..Default::default()
1252        };
1253        publish(
1254            &dir,
1255            0,
1256            None,
1257            None,
1258            vec![
1259                // No `data` oneof: a descriptor somebody's exporter believes in
1260                // that owns no points. It still belongs in the listing.
1261                named("a.declared", "1", None),
1262                named(
1263                    "b.gauge",
1264                    "By",
1265                    Some(Data::Gauge(Gauge {
1266                        data_points: vec![point.clone()],
1267                    })),
1268                ),
1269                named(
1270                    "c.sum",
1271                    "1",
1272                    Some(Data::Sum(Sum {
1273                        data_points: vec![point],
1274                        ..Default::default()
1275                    })),
1276                ),
1277                named(
1278                    "d.hist",
1279                    "s",
1280                    Some(Data::Histogram(Histogram {
1281                        data_points: vec![HistogramDataPoint {
1282                            time_unix_nano: 1_000,
1283                            count: 2,
1284                            sum: Some(3.0),
1285                            ..Default::default()
1286                        }],
1287                        ..Default::default()
1288                    })),
1289                ),
1290                named(
1291                    "e.exp",
1292                    "s",
1293                    Some(Data::ExponentialHistogram(ExponentialHistogram {
1294                        data_points: vec![ExponentialHistogramDataPoint {
1295                            time_unix_nano: 1_000,
1296                            count: 2,
1297                            sum: Some(3.0),
1298                            ..Default::default()
1299                        }],
1300                        ..Default::default()
1301                    })),
1302                ),
1303                named(
1304                    "f.summary",
1305                    "s",
1306                    Some(Data::Summary(Summary {
1307                        data_points: vec![SummaryDataPoint {
1308                            time_unix_nano: 1_000,
1309                            count: 2,
1310                            sum: 3.0,
1311                            ..Default::default()
1312                        }],
1313                    })),
1314                ),
1315            ],
1316        );
1317
1318        let r = names(&dir, 0, 10_000).unwrap();
1319        assert_eq!(
1320            r.json,
1321            concat!(
1322                r#"[{"name":"a.declared","unit":"1","kind":"unset"},"#,
1323                r#"{"name":"b.gauge","unit":"By","kind":"gauge"},"#,
1324                r#"{"name":"c.sum","unit":"1","kind":"sum"},"#,
1325                r#"{"name":"d.hist","unit":"s","kind":"histogram"},"#,
1326                r#"{"name":"e.exp","unit":"s","kind":"exponential_histogram"},"#,
1327                r#"{"name":"f.summary","unit":"s","kind":"summary"}]"#,
1328            )
1329        );
1330
1331        // The same words come back on the series themselves, and the three
1332        // aggregate kinds each split into the two derived halves.
1333        let json = query(&dir, &wide()).json;
1334        for (name, kind) in [
1335            ("b.gauge", "gauge"),
1336            ("c.sum", "sum"),
1337            ("d.hist.count", "histogram"),
1338            ("d.hist.sum", "histogram"),
1339            ("e.exp.count", "exponential_histogram"),
1340            ("e.exp.sum", "exponential_histogram"),
1341            ("f.summary.count", "summary"),
1342            ("f.summary.sum", "summary"),
1343        ] {
1344            assert!(
1345                json.contains(&format!(r#""name":"{name}","unit":"#))
1346                    && json.contains(&format!(r#""kind":"{kind}""#)),
1347                "{name}/{kind}: {json}"
1348            );
1349        }
1350        // A descriptor with no points charts nothing, however loudly it is
1351        // declared.
1352        assert!(!json.contains("a.declared"), "{json}");
1353
1354        let _ = std::fs::remove_dir_all(&dir);
1355    }
1356
1357    /// A block outside the window is never opened, and a series over the point
1358    /// cap keeps its newest points and says how many it dropped.
1359    ///
1360    /// Both are the difference between a chart and a lie. Opening every block
1361    /// makes the directory name — the whole index — worthless; keeping an
1362    /// arbitrary subset of points and sorting it at render time draws a
1363    /// contiguous line through whichever rows the directory listing reached
1364    /// first, with no sign that anything is missing.
1365    #[test]
1366    fn a_series_is_bounded_by_the_window_asked_for_and_the_points_it_can_chart() {
1367        let dir = tmp("series-bounds");
1368        let pts: Vec<(u64, i64)> = (1..=6).map(|i| (i as u64 * 1_000, i)).collect();
1369        publish(&dir, 0, None, None, vec![gauge("m", &pts)]);
1370        // Hours later, so it lands in its own partition and its own directory
1371        // name. Nothing in the window below can reach it.
1372        publish(
1373            &dir,
1374            1,
1375            None,
1376            None,
1377            vec![gauge("m", &[(9_000_000_000_000, 99)])],
1378        );
1379
1380        let mut q = wide();
1381        q.to = 10_000;
1382        q.max_points = 4;
1383        let r = query(&dir, &q);
1384
1385        assert_eq!(r.stats.blocks_total, 2);
1386        assert_eq!(r.stats.blocks_scanned, 1, "the far block was opened");
1387        // Six points offered, four charted, and the two dropped are the two
1388        // oldest — not the two the scan happened to see last.
1389        assert!(r.json.contains(r#""dropped_points":2"#), "{}", r.json);
1390        assert!(
1391            r.json
1392                .contains(r#""points":[["3000","3"],["4000","4"],["5000","5"],["6000","6"]]"#),
1393            "{}",
1394            r.json
1395        );
1396
1397        // Under the cap there is no `dropped_points` member at all: a chart
1398        // must be able to treat its presence as "something is missing".
1399        let mut q = wide();
1400        q.to = 10_000;
1401        let r = query(&dir, &q);
1402        assert!(!r.json.contains("dropped_points"), "{}", r.json);
1403        assert!(r.json.contains(r#"["1000","1"]"#), "{}", r.json);
1404
1405        let _ = std::fs::remove_dir_all(&dir);
1406    }
1407
1408    /// A term is satisfied at whichever of the four levels carries the
1409    /// attribute, and a block missing a level is filtered by the levels it has.
1410    ///
1411    /// Whether `service.name` is a resource attribute or a data point one is a
1412    /// detail of whoever configured the SDK. A filter that only looked at the
1413    /// point level would return nothing for the single most common metrics
1414    /// query there is, with a 200 and an empty list.
1415    #[test]
1416    fn a_term_is_satisfied_at_whichever_level_carries_the_attribute() {
1417        let dir = tmp("series-levels");
1418        publish(
1419            &dir,
1420            0,
1421            Some(Resource {
1422                attributes: vec![kv("service.name", "api")],
1423                ..Default::default()
1424            }),
1425            Some(InstrumentationScope {
1426                attributes: vec![kv("otel.lib", "sdk")],
1427                ..Default::default()
1428            }),
1429            vec![
1430                Metric {
1431                    metadata: vec![kv("tier", "gold")],
1432                    ..gauge("m1", &[(1_000, 1)])
1433                },
1434                Metric {
1435                    ..gauge("m2", &[(1_000, 2)])
1436                },
1437            ],
1438        );
1439        // No resource and no scope: both attribute tables are empty, so
1440        // `publish` never writes them. Filtering must fall back to the levels
1441        // this block does have rather than treating the absence as a match.
1442        publish(
1443            &dir,
1444            1,
1445            None,
1446            None,
1447            vec![Metric {
1448                metadata: vec![kv("tier", "gold")],
1449                ..gauge("m3", &[(1_000, 3)])
1450            }],
1451        );
1452
1453        // Point-level attributes, added after the fact so each metric above
1454        // stays readable.
1455        let with_pod = |name: &str, v: i64, pod: &str| Metric {
1456            data: Some(Data::Gauge(Gauge {
1457                data_points: vec![NumberDataPoint {
1458                    time_unix_nano: 1_000,
1459                    value: Some(NumValue::AsInt(v)),
1460                    attributes: vec![kv("pod", pod)],
1461                    ..Default::default()
1462                }],
1463            })),
1464            ..Metric {
1465                name: name.into(),
1466                ..Default::default()
1467            }
1468        };
1469        publish(
1470            &dir,
1471            2,
1472            Some(Resource {
1473                attributes: vec![kv("service.name", "db")],
1474                ..Default::default()
1475            }),
1476            None,
1477            vec![with_pod("m4", 4, "p4")],
1478        );
1479
1480        let ask = |key: &str, want: &str| {
1481            let mut q = wide();
1482            q.to = 10_000;
1483            q.terms = vec![Term {
1484                target: crate::query::Target::Attr(key.into()),
1485                op: crate::query::Op::Eq,
1486                value: crate::query::Value::Str(want.into()),
1487            }];
1488            let r = query(&dir, &q);
1489            let mut names: Vec<String> = Vec::new();
1490            for m in ["m1", "m2", "m3", "m4"] {
1491                if r.json.contains(&format!(r#""name":"{m}""#)) {
1492                    names.push(m.to_owned());
1493                }
1494            }
1495            names
1496        };
1497
1498        // Resource level. The second block has no resource attributes at all
1499        // and must not match on the strength of not having them.
1500        assert_eq!(ask("service.name", "api"), ["m1", "m2"]);
1501        assert_eq!(ask("service.name", "db"), ["m4"]);
1502        // Scope level.
1503        assert_eq!(ask("otel.lib", "sdk"), ["m1", "m2"]);
1504        // Metric level — `Metric.metadata`, which is per descriptor row and so
1505        // reaches both blocks that declare it.
1506        assert_eq!(ask("tier", "gold"), ["m1", "m3"]);
1507        // Point level.
1508        assert_eq!(ask("pod", "p4"), ["m4"]);
1509        // A term nothing carries at any level matches nothing, rather than
1510        // falling back to "no opinion" and matching everything.
1511        assert!(ask("tier", "bronze").is_empty());
1512        assert!(ask("nope", "x").is_empty());
1513
1514        let _ = std::fs::remove_dir_all(&dir);
1515    }
1516
1517    /// The four attribute levels merge with the most specific winning, and the
1518    /// merged map is what identifies the series.
1519    ///
1520    /// Attributes are half the grouping key, so getting the precedence wrong
1521    /// does not just mislabel a line — it splits one series into two, or fuses
1522    /// two into one, and neither is visible in the response.
1523    #[test]
1524    fn attributes_from_every_level_merge_with_the_most_specific_winning() {
1525        let dir = tmp("series-merge");
1526        publish(
1527            &dir,
1528            0,
1529            Some(Resource {
1530                attributes: vec![kv("env", "prod"), kv("region", "eu")],
1531                ..Default::default()
1532            }),
1533            None,
1534            vec![Metric {
1535                // Same key as the resource carries, one level down.
1536                metadata: vec![kv("env", "staging")],
1537                data: Some(Data::Gauge(Gauge {
1538                    data_points: vec![NumberDataPoint {
1539                        time_unix_nano: 1_000,
1540                        value: Some(NumValue::AsInt(1)),
1541                        // And again, one level further down.
1542                        attributes: vec![kv("env", "canary"), kv("pod", "x")],
1543                        ..Default::default()
1544                    }],
1545                })),
1546                ..Metric {
1547                    name: "m".into(),
1548                    ..Default::default()
1549                }
1550            }],
1551        );
1552
1553        let json = query(&dir, &wide()).json;
1554        // `env` resolves to the point's own value; the keys only an upper level
1555        // carries survive; and the losing values appear nowhere.
1556        assert!(
1557            json.contains(r#""attributes":{"env":"canary","pod":"x","region":"eu"}"#),
1558            "{json}"
1559        );
1560        assert!(
1561            !json.contains("prod") && !json.contains("staging"),
1562            "{json}"
1563        );
1564
1565        let _ = std::fs::remove_dir_all(&dir);
1566    }
1567
1568    /// Exemplars are capped per series, because nothing in OTLP bounds how many
1569    /// an exporter attaches to a point.
1570    ///
1571    /// Without the cap one misconfigured SDK inflates every chart response that
1572    /// touches its metric — and this is the response a chart polls on a timer.
1573    #[test]
1574    fn exemplars_are_capped_so_one_exporter_cannot_inflate_every_chart_response() {
1575        let dir = tmp("series-exemplars");
1576        publish(
1577            &dir,
1578            0,
1579            None,
1580            None,
1581            vec![Metric {
1582                data: Some(Data::Gauge(Gauge {
1583                    data_points: vec![NumberDataPoint {
1584                        time_unix_nano: 1_000,
1585                        value: Some(NumValue::AsInt(1)),
1586                        exemplars: (0..MAX_EXEMPLARS as i64 + 6)
1587                            .map(|i| Exemplar {
1588                                time_unix_nano: 1_000 + i as u64,
1589                                value: Some(exemplar::Value::AsInt(i)),
1590                                trace_id: vec![7u8; 16].into(),
1591                                ..Default::default()
1592                            })
1593                            .collect(),
1594                        ..Default::default()
1595                    }],
1596                })),
1597                ..Metric {
1598                    name: "m".into(),
1599                    ..Default::default()
1600                }
1601            }],
1602        );
1603
1604        let json = query(&dir, &wide()).json;
1605        // Points carry their timestamp as a bare array element, so every
1606        // occurrence of the key is one exemplar object.
1607        assert_eq!(
1608            json.matches(r#""time_unix_nano""#).count(),
1609            MAX_EXEMPLARS,
1610            "{json}"
1611        );
1612        // Capped, not dropped: the ones that are there carry the trace id that
1613        // is the whole point of an exemplar.
1614        assert!(json.contains(r#""trace_id":"07070707"#), "{json}");
1615
1616        let _ = std::fs::remove_dir_all(&dir);
1617    }
1618
1619    /// A table shaped differently from the one this reader writes contributes
1620    /// nothing, rather than panicking or charting garbage.
1621    ///
1622    /// Two things produce that shape and neither is an error: retention
1623    /// unlinking a block while a query walks it, and a block written by another
1624    /// version of the binary — the block directory is the manifest, so there is
1625    /// no schema version to check against and no coordinator to ask.
1626    #[test]
1627    fn a_table_the_reader_did_not_write_contributes_nothing_instead_of_panicking() {
1628        let dir = tmp("series-ragged");
1629        publish(&dir, 0, None, None, vec![gauge("ok", &[(1_000, 1)])]);
1630        let unlinked = publish(&dir, 1, None, None, vec![gauge("gone", &[(1_000, 2)])]);
1631        let ragged = publish(&dir, 2, None, None, vec![gauge("ragged", &[(1_000, 3)])]);
1632
1633        // Retention got here first: the descriptor table is gone, but the
1634        // directory is still listed.
1635        std::fs::remove_file(unlinked.join("metrics.arrow")).unwrap();
1636
1637        // An older writer's point table, with no `id` column to join
1638        // attributes and exemplars on. Staged and renamed, because a mapping
1639        // over a file being truncated is a SIGBUS, not an error.
1640        let table = ragged.join("number_dp.arrow");
1641        let mut b = crate::block::open_table_opt(&table)
1642            .unwrap()
1643            .unwrap()
1644            .batches[0]
1645            .clone();
1646        b.remove_column(b.schema().index_of("id").unwrap());
1647        let staged = ragged.join("number_dp.arrow.staged");
1648        crate::block::write_table(&staged, &b).unwrap();
1649        std::fs::rename(&staged, &table).unwrap();
1650
1651        let r = query(&dir, &wide());
1652        assert_eq!(r.stats.blocks_scanned, 3, "all three were in the window");
1653        assert!(r.json.contains(r#""name":"ok""#), "{}", r.json);
1654        assert!(!r.json.contains("gone"), "{}", r.json);
1655        assert!(!r.json.contains("ragged"), "{}", r.json);
1656
1657        // The name listing reads only the descriptor table, so the ragged block
1658        // still names its metric and the unlinked one names nothing.
1659        let n = names(&dir, 0, 10_000).unwrap();
1660        assert_eq!(
1661            n.json,
1662            concat!(
1663                r#"[{"name":"ok","unit":"","kind":"gauge"},"#,
1664                r#"{"name":"ragged","unit":"","kind":"gauge"}]"#,
1665            )
1666        );
1667
1668        // The same tolerance one layer down, where the missing column is the
1669        // one the whole index is built on.
1670        let no_parent = RecordBatch::try_from_iter(vec![(
1671            "id",
1672            Arc::new(UInt32Array::from(vec![0u32, 1])) as ArrayRef,
1673        )])
1674        .unwrap();
1675        assert!(index_by_parent(&no_parent).is_empty());
1676
1677        // And on a descriptor this renderer did not render: the suffix is
1678        // dropped rather than spliced into the middle of some other member.
1679        // Returning the input unchanged keeps the string valid JSON, which a
1680        // `find`-and-insert on a miss would not.
1681        assert_eq!(
1682            with_suffix(r#""kind":"histogram""#, COUNT),
1683            r#""kind":"histogram""#
1684        );
1685        assert_eq!(
1686            with_suffix(r#""name":"x","unit":"s""#, ""),
1687            r#""name":"x","unit":"s""#
1688        );
1689        assert_eq!(
1690            with_suffix(r#""name":"x","unit":"s""#, COUNT),
1691            r#""name":"x.count","unit":"s""#
1692        );
1693
1694        let _ = std::fs::remove_dir_all(&dir);
1695    }
1696
1697    /// What one data point costs to render, at the point count that makes the
1698    /// attribute join visible.
1699    ///
1700    /// Scaled rather than `#[ignore]`d, exactly like
1701    /// [`crate::query::scan_cost_per_row`]: the default 2,000 points run in the
1702    /// normal suite as a correctness check on the merge, and the same code is
1703    /// the measurement at a real point count.
1704    ///
1705    /// ```sh
1706    /// MIRA_BENCH_POINTS=50000 cargo test --release -p miradb-core \
1707    ///     --lib series_cost_per_point -- --nocapture
1708    /// ```
1709    ///
1710    /// The shape is the one a real exporter produces and the one the old
1711    /// [`collect_attrs`] was quadratic in: a few distinct attribute sets over
1712    /// many timestamps, so the series map stays small while `dp_attrs` grows
1713    /// with the point count. Scanning the whole table per point is `2n²`
1714    /// comparisons; the run search is two per point.
1715    #[test]
1716    fn series_cost_per_point() {
1717        let n: usize = std::env::var("MIRA_BENCH_POINTS")
1718            .ok()
1719            .and_then(|s| s.parse().ok())
1720            .unwrap_or(2_000);
1721        let dir = tmp("series-bench");
1722        const PODS: usize = 20;
1723        let points: Vec<NumberDataPoint> = (0..n)
1724            .map(|i| NumberDataPoint {
1725                time_unix_nano: 1_000 + i as u64,
1726                value: Some(NumValue::AsInt(i as i64)),
1727                attributes: vec![
1728                    kv("k8s.pod.name", &format!("checkout-{}", i % PODS)),
1729                    kv("cloud.availability_zone", &format!("z{}", i % 3)),
1730                ],
1731                ..Default::default()
1732            })
1733            .collect();
1734        publish(
1735            &dir,
1736            0,
1737            Some(Resource {
1738                attributes: vec![kv("service.name", "checkout")],
1739                ..Default::default()
1740            }),
1741            None,
1742            vec![Metric {
1743                name: "http.server.requests".into(),
1744                data: Some(Data::Gauge(Gauge {
1745                    data_points: points,
1746                })),
1747                ..Default::default()
1748            }],
1749        );
1750        let q = SeriesQuery {
1751            name: Some("http.server.requests".into()),
1752            max_points: n,
1753            ..wide()
1754        };
1755
1756        // Once to fault the block in, so the number below is the join and not
1757        // the first touch of every page.
1758        let warm = query(&dir, &q);
1759        assert_eq!(warm.stats.rows_matched, n);
1760
1761        let t = std::time::Instant::now();
1762        let r = query(&dir, &q);
1763        let el = t.elapsed();
1764
1765        // The correctness half, which is why this is not `#[ignore]`d: every
1766        // point kept its own pod and inherited the resource's service, so the
1767        // binary search found the right run and not its neighbour.
1768        let series = r.json.matches(r#""name":"#).count();
1769        assert_eq!(series, PODS * 3, "one series per distinct attribute set");
1770        for pod in 0..PODS {
1771            assert!(
1772                r.json
1773                    .contains(&format!(r#""k8s.pod.name":"checkout-{pod}""#)),
1774                "pod {pod} lost its own attributes"
1775            );
1776        }
1777        assert!(r.json.contains(r#""service.name":"checkout""#));
1778
1779        println!(
1780            "series: {n} points in {el:?}  {:.3} us/point  {series} series",
1781            el.as_secs_f64() * 1e6 / n as f64,
1782        );
1783        let _ = std::fs::remove_dir_all(&dir);
1784    }
1785}