Skip to main content

mira_core/
query.rs

1//! The read path: prune blocks, scan columns, materialize JSON.
2//!
3//! There is no query planner and no expression tree here, and that is a
4//! decision rather than an omission. Mira answers a small, known set of
5//! questions — "the last N records matching these filters", "every span of this
6//! trace", "this metric bucketed over time" — and each is a hand-written scan
7//! over a layout designed for it. A general planner spends its budget
8//! rediscovering at runtime what this module knows at compile time: which
9//! column holds the timestamp, that `parent_id` is a row index, that block
10//! directory names already carry the pruning key. That is most of what the 151
11//! transitive crates of a general engine buy.
12//!
13//! Three properties the layout hands the scan, in the order they matter:
14//!
15//! * **Blocks prune by name.** `<min_ts>-<max_ts>-<node>-<seq>-<wal_hi>` is the
16//!   whole index. A time-bounded query opens no file it will not read, and with
17//!   blocks visited newest-first a `limit` stops the scan early.
18//! * **`parent_id` is a row index.** Ids are rebased dense per block at ingest,
19//!   so attaching an attribute to its record is an array store, not a hash
20//!   join.
21//! * **Dictionary columns compare as `u16`.** `severity_text = "ERROR"` resolves
22//!   the string once against the dictionary, then scans a buffer of 16-bit
23//!   codes and never touches string data again.
24//!
25//! Everything runs against mmap'd buffers, so a cold block costs demand paging
26//! and a warm one costs memory bandwidth. Nothing allocates per row until
27//! materialization, which happens after `limit` has cut the result to size.
28//!
29//! With no write-ahead log, nothing here needs to read the open, unsealed
30//! block: ingest acknowledges an export only after the block containing it has
31//! been fsynced and renamed into place, so read-your-writes falls out of the
32//! durability rule. Turn the log on and the acknowledgement moves ahead of the
33//! seal, which breaks that — so [`search_open`] takes the open block's snapshot
34//! alongside the directory scan and merges the two into one ordered page. See
35//! [`crate::signal::Open`]; the argument is ARCHITECTURE section 4.
36
37use std::path::Path;
38use std::sync::Arc;
39// `Relaxed` alone rather than `Ordering`, which is `std::cmp::Ordering` in this
40// file.
41use std::sync::atomic::{AtomicUsize, Ordering::Relaxed};
42
43use arrow_array::cast::AsArray;
44use arrow_array::types::{
45    Float64Type, Int32Type, Int64Type, TimestampNanosecondType, UInt8Type, UInt16Type, UInt32Type,
46    UInt64Type,
47};
48use arrow_array::{Array, BooleanArray, RecordBatch, StringArray};
49use arrow_schema::DataType;
50use mira_proto::common::v1::AnyValue;
51use prost::Message;
52
53use crate::block::{self, Src};
54use crate::error::Result;
55use crate::json::Json;
56use crate::schema::AttrType;
57use crate::signal::Open;
58
59/// Comparison operator. `Contains` is substring matching on strings and matches
60/// nothing on any other type: a filter that cannot apply returns no rows rather
61/// than an error, because a query spanning signals with slightly different
62/// columns is a normal thing for an agent to try.
63#[derive(Debug, Clone, Copy, PartialEq, Eq)]
64pub enum Op {
65    Eq,
66    Ne,
67    Lt,
68    Lte,
69    Gt,
70    Gte,
71    Contains,
72}
73
74impl Op {
75    pub fn parse(s: &str) -> Option<Op> {
76        Some(match s {
77            "eq" | "=" | "==" => Op::Eq,
78            "ne" | "!=" => Op::Ne,
79            "lt" | "<" => Op::Lt,
80            "lte" | "<=" => Op::Lte,
81            "gt" | ">" => Op::Gt,
82            "gte" | ">=" => Op::Gte,
83            "contains" | "~" => Op::Contains,
84            _ => return None,
85        })
86    }
87
88    fn test_ord(self, ord: std::cmp::Ordering) -> bool {
89        use std::cmp::Ordering::*;
90        match self {
91            Op::Eq => ord == Equal,
92            Op::Ne => ord != Equal,
93            Op::Lt => ord == Less,
94            Op::Lte => ord != Greater,
95            Op::Gt => ord == Greater,
96            Op::Gte => ord != Less,
97            // Never reached: every caller handles Contains before comparing.
98            Op::Contains => false,
99        }
100    }
101}
102
103/// A scalar from a query document.
104///
105/// Deliberately loose about numbers. Queries arrive as JSON from a browser or
106/// an LLM, and both write `"200"` for `http.response.status_code` about as often
107/// as they write `200`. Rejecting the quoted form would be defensible and
108/// useless, so coercion happens at comparison time, once the column's real type
109/// is known.
110#[derive(Debug, Clone, PartialEq)]
111pub enum Value {
112    Str(String),
113    Int(i64),
114    Double(f64),
115    Bool(bool),
116}
117
118impl Value {
119    fn as_i64(&self) -> Option<i64> {
120        match self {
121            Value::Int(i) => Some(*i),
122            // A whole-valued float is the same number; a fractional one is not,
123            // and truncating it would make `duration > 0.5` mean `> 0`.
124            Value::Double(d) if d.fract() == 0.0 => Some(*d as i64),
125            Value::Str(s) => s.parse().ok(),
126            Value::Double(_) | Value::Bool(_) => None,
127        }
128    }
129
130    fn as_f64(&self) -> Option<f64> {
131        match self {
132            Value::Int(i) => Some(*i as f64),
133            Value::Double(d) => Some(*d),
134            Value::Str(s) => s.parse().ok(),
135            Value::Bool(_) => None,
136        }
137    }
138
139    fn as_bool(&self) -> Option<bool> {
140        match self {
141            Value::Bool(b) => Some(*b),
142            Value::Str(s) if s == "true" => Some(true),
143            Value::Str(s) if s == "false" => Some(false),
144            Value::Str(_) | Value::Int(_) | Value::Double(_) => None,
145        }
146    }
147
148    fn as_str(&self) -> Option<&str> {
149        match self {
150            Value::Str(s) => Some(s),
151            _ => None,
152        }
153    }
154}
155
156/// What a term filters on.
157#[derive(Debug, Clone)]
158pub enum Target {
159    /// A column of the signal's root table, by name.
160    Field(String),
161    /// An attribute key, looked up at every level the signal has — record,
162    /// resource, scope, and on traces the span's own events and links — and
163    /// unioned.
164    ///
165    /// Whether `service.name` is a resource attribute is a detail of whoever
166    /// configured the SDK, and users do not know it. Searching all of them is
167    /// what every usable tracing UI does; the star schema makes it a handful of
168    /// scans of tables that are tiny next to the root.
169    ///
170    /// The child levels are not a nicety. `Span.recordException` — the one API
171    /// call behind most of the spans anyone goes looking for — writes
172    /// `exception.type` to a span *event*, so leaving events out made the
173    /// commonest question in a tracing UI return an empty list while the value
174    /// was visible in `events[].attributes` of the very same response. A match
175    /// on a child selects the span it hangs off, which is the row the caller
176    /// asked for.
177    Attr(String),
178}
179
180/// One conjunct. Terms are AND-ed. There is no OR in V0: a disjunction over
181/// attributes is rare enough that supporting it means building a planner for a
182/// query nobody has typed yet.
183#[derive(Debug, Clone)]
184pub struct Term {
185    pub target: Target,
186    pub op: Op,
187    pub value: Value,
188}
189
190/// Which signal, and therefore which tables and which time column.
191#[derive(Debug, Clone, Copy, PartialEq, Eq)]
192pub enum Signal {
193    Logs,
194    Traces,
195}
196
197impl Signal {
198    pub fn parse(s: &str) -> Option<Signal> {
199        match s {
200            "logs" => Some(Signal::Logs),
201            "traces" | "spans" => Some(Signal::Traces),
202            _ => None,
203        }
204    }
205
206    pub fn dir(self) -> &'static str {
207        match self {
208            Signal::Logs => "logs",
209            Signal::Traces => "traces",
210        }
211    }
212
213    /// The root table's file stem.
214    fn root(self) -> &'static str {
215        match self {
216            Signal::Logs => "logs",
217            Signal::Traces => "spans",
218        }
219    }
220
221    /// The record-level attribute table.
222    fn attrs(self) -> &'static str {
223        match self {
224            Signal::Logs => "log_attrs",
225            Signal::Traces => "span_attrs",
226        }
227    }
228
229    /// The column that orders results and bounds the query.
230    fn time_col(self) -> &'static str {
231        match self {
232            Signal::Logs => "time_unix_nano",
233            Signal::Traces => "start_time_unix_nano",
234        }
235    }
236}
237
238/// A record search: filter, order by time descending, take `limit`.
239#[derive(Debug, Clone)]
240pub struct Search {
241    pub signal: Signal,
242    /// Inclusive nanosecond bounds.
243    pub from: i64,
244    pub to: i64,
245    pub terms: Vec<Term>,
246    pub limit: usize,
247    /// Start after this row. See [`Cursor`].
248    pub after: Option<Cursor>,
249}
250
251/// Where the previous page stopped.
252///
253/// Keyset, not offset, and not for tidiness: `offset: 20000` forces the engine
254/// to find and discard twenty thousand rows on every page, which turns the
255/// early exit below into a full scan and makes the last page the most expensive
256/// one. It is also *wrong* on a store that is still being written to — a batch
257/// arriving between two pages shifts every row down and the reader sees a row
258/// twice or never.
259///
260/// This is the sort key of the last row returned, so "the next page" is
261/// "everything that sorts after this", which is exact whatever else has landed
262/// meanwhile. `(node, seq)` identifies the block globally with no coordination
263/// (see [`block::node_id`]) and `row` is its offset inside it, so the key is
264/// intrinsic to the record rather than to the query that found it.
265///
266/// Rendered as `ts.node.seq.row`, in decimal, on purpose: an agent reading a
267/// response can tell what it is holding, and a human debugging a stuck reader
268/// can tell where it stopped.
269#[derive(Debug, Clone, Copy, PartialEq, Eq)]
270pub struct Cursor {
271    pub ts: i64,
272    pub node: u32,
273    pub seq: u64,
274    pub row: u32,
275}
276
277impl Cursor {
278    /// Descending: newest first, and for rows sharing a nanosecond, the
279    /// higher-numbered block and row first. Any total order would do; what
280    /// matters is that it is total, so no row can hide in a tie.
281    fn key(&self) -> std::cmp::Reverse<(i64, u32, u64, u32)> {
282        std::cmp::Reverse((self.ts, self.node, self.seq, self.row))
283    }
284}
285
286impl std::fmt::Display for Cursor {
287    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
288        write!(f, "{}.{}.{}.{}", self.ts, self.node, self.seq, self.row)
289    }
290}
291
292impl std::str::FromStr for Cursor {
293    type Err = String;
294
295    fn from_str(s: &str) -> std::result::Result<Cursor, String> {
296        let bad = || format!("{s:?} is not a cursor; pass back the `next` field verbatim");
297        let mut p = s.split('.');
298        let mut next = |f: &dyn Fn(&str) -> bool| p.next().filter(|v| f(v)).ok_or_else(bad);
299        // `ts` may be negative; nothing else may. Parsing the pieces by hand
300        // rather than trusting `parse` to reject `+1` or `1_000`.
301        let ts = next(&|v: &str| !v.is_empty())?.parse().map_err(|_| bad())?;
302        let digits = |v: &str| !v.is_empty() && v.bytes().all(|b| b.is_ascii_digit());
303        let c = Cursor {
304            ts,
305            node: next(&digits)?.parse().map_err(|_| bad())?,
306            seq: next(&digits)?.parse().map_err(|_| bad())?,
307            row: next(&digits)?.parse().map_err(|_| bad())?,
308        };
309        match p.next() {
310            Some(_) => Err(bad()),
311            None => Ok(c),
312        }
313    }
314}
315
316/// One matching record, kept only long enough to sort and materialize.
317///
318/// Holds a block index and a row number, not the row's data: a query over a
319/// wide window can match millions of rows and keep a hundred, and copying the
320/// other 999,900 out of the mapping in order to discard them is the easiest way
321/// to make a columnar engine slow.
322pub(crate) struct Hit {
323    ts: i64,
324    pub(crate) block: usize,
325    pub(crate) row: u32,
326}
327
328/// What a search cost, reported alongside the rows.
329///
330/// Not decoration. "How much did that cost" is the first question when a query
331/// is slow, and it is also what tells an agent its filter was too broad.
332#[derive(Debug, Default, Clone)]
333pub struct Stats {
334    pub blocks_total: usize,
335    pub blocks_scanned: usize,
336    pub rows_scanned: usize,
337    /// Rows satisfying the whole query — window, terms and `after` — not rows
338    /// on this page. `limit` cuts the page; this says how much there was to
339    /// cut, so a caller can tell "my filter is right and I am reading page one
340    /// of forty" from "my filter is wrong".
341    pub rows_matched: usize,
342    /// Series the metrics cap refused (`series::bound`); always zero on a
343    /// record search, which has no such cap.
344    ///
345    /// The counterpart to the per-series `dropped_points`, and here rather than
346    /// in the response body because the response body is a JSON array with
347    /// nowhere to hang a number that belongs to all of it. A chart with a
348    /// series missing and no way to know it is the failure mode both of these
349    /// exist to prevent.
350    pub dropped_series: usize,
351}
352
353pub struct Results {
354    /// A JSON array of row objects.
355    pub json: String,
356    pub stats: Stats,
357    /// Pass back as `after` for the next page. `None` means this was the last
358    /// one — not "ask again and see", which is the ambiguity that makes readers
359    /// poll forever.
360    pub next: Option<Cursor>,
361}
362
363/// The trace id this search pins down exactly, if it pins one down.
364///
365/// Only `trace_id = <32 hex chars>` qualifies, on either signal — logs blocks
366/// carry the same filter as span blocks, so "the logs for this trace" prunes
367/// exactly as hard as "the spans for it". Terms are AND-ed, so one such term is
368/// enough no matter what else is in the list: a block that cannot hold the id
369/// cannot hold a row satisfying the conjunction.
370fn trace_needle(q: &Search) -> Option<[u8; 16]> {
371    q.terms.iter().find_map(|t| match (&t.target, t.op) {
372        (Target::Field(f), Op::Eq) if f == "trace_id" => unhex(t.value.as_str()?)?.try_into().ok(),
373        _ => None,
374    })
375}
376
377/// The attribute equalities in a search, in the form a block filter answers.
378///
379/// Terms are AND-ed, so any single one the block cannot satisfy rules the whole
380/// block out — which is why a list of independent probes is enough and no
381/// expression tree is needed.
382///
383/// `numeric` records that the query scalar could be read as a number. Doubles
384/// are not in the index (see [`crate::bloom::HAS_DOUBLE`]), so a block that
385/// holds any must be scanned for a numeric query even when the text probe misses.
386struct AttrProbe {
387    hash: (u64, u64),
388    numeric: bool,
389}
390
391impl AttrProbe {
392    /// The block's own answer to "could a row here satisfy this term?".
393    fn maybe(&self, f: &crate::bloom::Filter) -> bool {
394        f.may_contain(self.hash) || (self.numeric && f.flags & crate::bloom::HAS_DOUBLE != 0)
395    }
396}
397
398fn attr_probes(q: &Search) -> Vec<AttrProbe> {
399    q.terms
400        .iter()
401        .filter(|t| t.op == Op::Eq)
402        .filter_map(|t| match &t.target {
403            Target::Attr(key) => Some(AttrProbe {
404                hash: crate::bloom::attr_hash(key, canon(&t.value).as_bytes()),
405                numeric: t.value.as_f64().is_some(),
406            }),
407            Target::Field(_) => None,
408        })
409        .collect()
410}
411
412/// The comparisons in a search that a zone map can answer.
413///
414/// Only an *unquoted* number qualifies, and that is a rule about the scan
415/// rather than about the index. A quoted scalar means a text comparison on a
416/// `str`-typed attribute (see [`AttrPred`]) and a numeric one on an `int`
417/// column, so the same term is lexicographic against one row and arithmetic
418/// against the next — two orderings, and [`crate::zone`] describes one. Since
419/// an absent key is read as "prune", a probe that describes the wrong ordering
420/// would not merely mislead, it would drop rows. `Ne` and `Contains` are out
421/// for the reason in [`crate::zone`]: they prune nothing worth the branch.
422fn range_probes(q: &Search) -> Vec<crate::zone::Probe> {
423    q.terms
424        .iter()
425        .filter(|t| !matches!(t.op, Op::Ne | Op::Contains))
426        .filter(|t| matches!(t.value, Value::Int(_) | Value::Double(_)))
427        .map(|t| crate::zone::Probe {
428            key: match &t.target {
429                Target::Attr(k) => crate::zone::attr_key(k),
430                Target::Field(f) => crate::zone::field_key(f),
431            },
432            op: t.op,
433            // The same two readings `attr_matches` and `field_pred` take of the
434            // scalar, so the interval consulted is the one the row would be
435            // compared against.
436            int: t.value.as_i64(),
437            float: t.value.as_f64(),
438        })
439        .collect()
440}
441
442/// The text an attribute of this value would have been indexed under.
443///
444/// This is the contract between [`AttrPred`] and the block index, and both
445/// sides are written against it: `Op::Eq` on a stored string is *defined* as
446/// equality with this text, and [`crate::attrs::index`] writes exactly these
447/// bytes for the str, int and bool types. Doubles are not in the index at all —
448/// the `numeric` flag covers them — so a stored double is reached through
449/// `HAS_DOUBLE` rather than through this string.
450///
451/// Every value has a text form, including a fractional double: nothing forces a
452/// producer to send `0.5` as a double rather than as `"0.5"`, and returning
453/// `None` here to mean "unindexable" made every fractional-double equality scan
454/// every block.
455fn canon(v: &Value) -> String {
456    match v {
457        Value::Str(s) => s.clone(),
458        Value::Int(i) => i.to_string(),
459        Value::Bool(b) => b.to_string(),
460        // `1.0` is written `1` by the int arm of the indexer and would be
461        // written `1` by anything rendering the number for a human, so a whole
462        // float has to canonicalise the same way or `eq: 1.0` misses `1`.
463        Value::Double(d) if d.fract() == 0.0 => (*d as i64).to_string(),
464        Value::Double(d) => d.to_string(),
465    }
466}
467
468/// Run a search against the blocks under `root`.
469///
470/// Blocking: this mmaps and page-faults. Callers on an async runtime must go
471/// through `spawn_blocking` — a hard fault stalls the whole OS thread with no
472/// yield point and no signal to the scheduler.
473pub fn search(root: &Path, q: &Search) -> Result<Results> {
474    search_open(root, q, &[])
475}
476
477/// As [`search`], but also reads `open` — snapshots of blocks that have been
478/// acknowledged and not yet published.
479///
480/// This is what keeps read-your-writes true once the write-ahead log moves the
481/// acknowledgement ahead of the seal (ARCHITECTURE section 4): without it a client that
482/// got a `200` and queried immediately would get nothing back for up to
483/// `max_block_age`. See [`crate::signal::Open`] for why a cursor stays valid
484/// across the seal.
485pub fn search_open(root: &Path, q: &Search, open_blocks: &[Arc<Open>]) -> Result<Results> {
486    let disk = block::scan(root, q.signal.dir())?;
487    let mut refs = block::sources(&disk, open_blocks);
488    let mut stats = Stats {
489        blocks_total: refs.len(),
490        ..Default::default()
491    };
492    refs.retain(|b| b.overlaps(q.from, q.to));
493    // A block whose oldest row is newer than the cursor is entirely on a page
494    // already delivered. Pruning here rather than per row is what keeps deep
495    // paging as cheap as the first page.
496    if let Some(c) = &q.after {
497        refs.retain(|b| b.min_ts <= c.ts);
498    }
499    // Newest first, so `limit` can cut the scan short.
500    refs.sort_by_key(|b| std::cmp::Reverse((b.max_ts, b.seq)));
501    let refs = &refs[..];
502
503    let scan = Scan::new(q);
504    let mut hits: Vec<Hit> = Vec::new();
505    let mut open: Vec<Option<Block>> = Vec::with_capacity(refs.len());
506
507    // Waves, widening. The first is one block wide because the commonest query
508    // there is — "the last `limit` records" — is answered by the first block and
509    // exits, and reading eleven more in parallel to throw them away would make
510    // the cheap case eleven times dearer to make the dear case faster. Doubling
511    // reaches full width after four waves and fifteen blocks, which is noise
512    // against the scan this is for.
513    let mut width = 1;
514    let mut i = 0;
515    while i < refs.len() {
516        // The early exit. With `limit` hits held, a block whose newest row
517        // predates the oldest hit cannot contribute — and since blocks are
518        // ordered by max_ts descending, neither can any block after it.
519        //
520        // Checked at the head of the wave rather than per block, which is what
521        // parallelism costs: a wave that begins before the limit is reached runs
522        // to the end even if its first block would have satisfied it. So
523        // `blocks_scanned` is now "what the scan read", not "the fewest blocks
524        // that could have answered" — the two were the same number when the loop
525        // was serial and are within one wave of each other now.
526        if hits.len() >= q.limit && hits.last().is_some_and(|w| refs[i].max_ts < w.ts) {
527            break;
528        }
529        let asked = (i + width).min(refs.len());
530        let answers = scan.wave(refs, i, asked);
531        let end = i + answers.len();
532        for done in answers {
533            let (block, found) = done?;
534            if let Some(b) = &block {
535                stats.blocks_scanned += 1;
536                stats.rows_scanned += b.root.num_rows();
537            }
538            // Counted *after* the cursor, which is what makes it a property of
539            // the query rather than of the scan. The block retain above already
540            // drops every block that lies wholly ahead of the cursor, so
541            // counting before the cursor made `rows_matched` shrink from page to
542            // page by whatever that pruning happened to remove — a number that
543            // moves when an optimisation fires is reporting the optimisation,
544            // not the query. Counting behind the cursor instead gives one
545            // meaning that holds on every page: how many matching rows are still
546            // to be read from here.
547            //
548            // It stays a lower bound in exactly one case — the early exit above,
549            // which cannot fire until `limit` hits are held and therefore cannot
550            // fire on a response whose `next` is `None`. So the count is exact
551            // whenever the answer is complete, and short only when the reader has
552            // already been told to page.
553            stats.rows_matched += found.len();
554            hits.extend(found);
555            open.push(block);
556        }
557
558        // Trim between waves, so memory is bounded by `limit` times the wave
559        // width rather than by the match count, which is not bounded by
560        // anything.
561        //
562        // Partition before sorting. `limit` is a hundred and a broad filter over
563        // one block is hundreds of thousands, so sorting the match set to throw
564        // away all but its head is the dominant cost of the commonest query
565        // there is — "the last 100 records", which matches every row it reads.
566        // `select_nth_unstable` is linear and leaves the head in the first
567        // `limit` slots; only those get ordered.
568        if hits.len() > q.limit {
569            hits.select_nth_unstable_by_key(q.limit, |h| cursor(&refs[h.block], h).key());
570            hits.truncate(q.limit);
571        }
572        hits.sort_unstable_by_key(|h| cursor(&refs[h.block], h).key());
573        i = end;
574        width = (width * 2).min(MAX_FANOUT);
575    }
576
577    let mut j = Json::new();
578    j.arr(|j| {
579        for h in &hits {
580            let b = open[h.block].as_ref().expect("a hit implies an open block");
581            b.emit_row(j, h.row);
582        }
583    });
584    Ok(Results {
585        json: j.into_string(),
586        stats,
587        // A short page is the last page. Saying so costs nothing here and saves
588        // every reader one round trip that returns nothing.
589        next: (hits.len() == q.limit)
590            .then(|| hits.last().map(|h| cursor(&refs[h.block], h)))
591            .flatten(),
592    })
593}
594
595/// The widest a single search will fan out across blocks.
596///
597/// Past this the win is not CPU. A block scan is mostly page faults, the device
598/// answers them at its own rate, and every extra thread is one more spawn and
599/// one more shootdown for the same IO.
600const MAX_FANOUT: usize = 16;
601
602/// Threads a search may borrow beyond its own, for the whole process.
603///
604/// Fan-out is only worth anything when there is a core going spare, and this is
605/// the measurement that says so. One search over 37 blocks and 19M rows: 1.81 s
606/// serial, 0.20 s fanned out. Eight concurrent searches on the same twelve
607/// cores: throughput unchanged to within noise, and the short classes' p99 an
608/// order of magnitude worse — the cores were already busy, so every extra
609/// thread was scheduler work and nothing else.
610///
611/// So the budget is shared rather than per-search. A search takes what is idle
612/// and runs serially when nothing is, which makes the two cases above the same
613/// code with no mode to pick and nothing to configure. Bounded by the core
614/// count less the caller's own thread, because that is what "idle" means here.
615///
616/// Deliberately *not* a fair queue: a search never waits for the budget, it
617/// only asks. Waiting would trade the thing being optimised — latency — for a
618/// share of a resource the caller is about to finish with anyway.
619static SPARE: std::sync::LazyLock<AtomicUsize> = std::sync::LazyLock::new(|| {
620    AtomicUsize::new(
621        std::thread::available_parallelism()
622            .map_or(1, |n| n.get())
623            .saturating_sub(1)
624            .min(MAX_FANOUT),
625    )
626});
627
628/// Threads borrowed from [`SPARE`], returned on drop — including the drop that
629/// unwinds a panicking scan. Leaking one would degrade the process to serial
630/// scans permanently, with nothing to point at.
631pub(crate) struct Helpers(usize);
632
633impl Helpers {
634    /// `pub(crate)` for one caller that is not a scan: a test claiming the
635    /// whole budget, so it can run the same search with the fan-out off and
636    /// with it on and demand the same bytes back.
637    pub(crate) fn claim(want: usize) -> Helpers {
638        let mut got = 0;
639        let _ = SPARE.fetch_update(Relaxed, Relaxed, |n| {
640            got = n.min(want);
641            (got > 0).then(|| n - got)
642        });
643        Helpers(got)
644    }
645}
646
647impl Drop for Helpers {
648    fn drop(&mut self) {
649        SPARE.fetch_add(self.0, Relaxed);
650    }
651}
652
653/// Everything the per-block half of a search reads, and nothing it writes.
654///
655/// One struct rather than five arguments because it crosses a thread boundary:
656/// shared by reference, immutable for the whole scan, and therefore `Sync`
657/// without a lock. What is left out is as deliberate — `hits` and `stats` are
658/// the only state a block scan produces, and both are returned rather than
659/// accumulated, which is what makes the blocks independent.
660pub(crate) struct Scan<'a> {
661    q: &'a Search,
662    needle: Option<[u8; 16]>,
663    probes: Vec<AttrProbe>,
664    ranges: Vec<crate::zone::Probe>,
665    after: std::cmp::Reverse<(i64, u32, u64, u32)>,
666}
667
668impl<'a> Scan<'a> {
669    /// The per-block half of a query, prepared once.
670    ///
671    /// Everything here is derived from `q` alone, which is what lets one `Scan`
672    /// serve every block of a search and every thread scanning them.
673    pub(crate) fn new(q: &'a Search) -> Scan<'a> {
674        Scan {
675            q,
676            needle: trace_needle(q),
677            probes: attr_probes(q),
678            ranges: range_probes(q),
679            // With no cursor, every row is after the start — which is the key
680            // that sorts ahead of all of them.
681            after: q.after.map_or(
682                std::cmp::Reverse((i64::MAX, u32::MAX, u64::MAX, u32::MAX)),
683                |c| c.key(),
684            ),
685        }
686    }
687
688    /// Blocks `[from, to)` scanned at once, answered in block order.
689    ///
690    /// `std::thread::scope` and not a pool: the threads borrow the mapping, the
691    /// query and each other's nothing, and a scope is the one construct that
692    /// lets them do that without an `Arc` around every field. They are also the
693    /// wrong threads to pool — a pool exists to amortise spawns against short
694    /// tasks, and a block scan is milliseconds against a spawn's microseconds.
695    ///
696    /// The caller's own thread takes the first block, so a wave with no helpers
697    /// — the first wave of every search, and every wave on a busy node — spawns
698    /// nothing at all.
699    ///
700    /// `to` is the wave the ramp asked for and the answer may be shorter: what
701    /// [`SPARE`] hands out decides how many blocks this wave actually covers,
702    /// and the caller advances by the number of answers rather than by the
703    /// width it asked for.
704    pub(crate) fn wave(&self, refs: &[Src<'_>], from: usize, to: usize) -> Vec<Result<Scanned>> {
705        let helpers = Helpers::claim(to - from - 1);
706        if helpers.0 == 0 {
707            return vec![self.block(from, &refs[from])];
708        }
709        let to = from + 1 + helpers.0;
710        std::thread::scope(|s| {
711            let rest: Vec<_> = (from + 1..to)
712                .map(|k| s.spawn(move || self.block(k, &refs[k])))
713                .collect();
714            let mut out = Vec::with_capacity(to - from);
715            out.push(self.block(from, &refs[from]));
716            // A panic in a block scan is a bug in this file, not a bad query.
717            // Re-raising it on the caller's thread puts it where the runtime
718            // will turn it into a 500 with the original message attached;
719            // swallowing it would return a silently short answer instead.
720            out.extend(
721                rest.into_iter()
722                    .map(|h| h.join().unwrap_or_else(|e| std::panic::resume_unwind(e))),
723            );
724            out
725        })
726    }
727
728    /// One block: prune it, or open it and find the rows that match.
729    pub(crate) fn block(&self, i: usize, bref: &Src<'_>) -> Result<Scanned> {
730        // The sidecar filters (section 7.4). Both cover the case the block name cannot:
731        // a query with no useful time bound, either because it names a trace or
732        // because it names an attribute value that is rare or absent. Without
733        // them the scan runs to the end of retention to prove a negative. A
734        // 20 KB read is two orders of magnitude cheaper than the block it skips,
735        // and every damaged or missing filter reads as "scan me".
736        //
737        // `is_some_and`/`is_ok_and` rather than a `let` chain: those are stable
738        // since 1.70 and the chains are stable since 1.88, which is three years
739        // past the MSRV this workspace declares.
740        let no_trace = bref.dir.zip(self.needle.as_ref()).is_some_and(|(dir, id)| {
741            std::fs::read(dir.join(crate::bloom::TRACE_IDX))
742                .is_ok_and(|f| !crate::bloom::may_contain(&f, id))
743        });
744        if no_trace {
745            return Ok((None, Vec::new()));
746        }
747        let no_attr = !self.probes.is_empty()
748            && bref.dir.is_some_and(|dir| {
749                std::fs::read(dir.join(crate::bloom::ATTR_IDX)).is_ok_and(|bytes| {
750                    crate::bloom::Filter::open(&bytes)
751                        .is_some_and(|f| !self.probes.iter().all(|p| p.maybe(&f)))
752                })
753            });
754        if no_attr {
755            return Ok((None, Vec::new()));
756        }
757        // And the orderings, which the Bloom filter cannot help with at all:
758        // "slower than a second", "returned 5xx". Same shape, same fail-open —
759        // a block published before the zone map existed has no file here and is
760        // scanned, exactly as it was before.
761        let out_of_range = !self.ranges.is_empty()
762            && bref.dir.is_some_and(|dir| {
763                std::fs::read(dir.join(crate::zone::ZONE_IDX)).is_ok_and(|bytes| {
764                    crate::zone::Map::open(&bytes)
765                        .is_some_and(|m| !self.ranges.iter().all(|p| p.maybe(&m)))
766                })
767            });
768        if out_of_range {
769            return Ok((None, Vec::new()));
770        }
771
772        let Some(b) = Block::open(bref, self.q.signal)? else {
773            return Ok((None, Vec::new()));
774        };
775        let sel = b.select(self.q, bref);
776        let hits = b.time().map_or_else(Vec::new, |time| {
777            sel.iter()
778                .filter_map(|&row| {
779                    let h = Hit {
780                        ts: time[row as usize],
781                        block: i,
782                        row,
783                    };
784                    (cursor(bref, &h).key() > self.after).then_some(h)
785                })
786                .collect()
787        });
788        Ok((Some(b), hits))
789    }
790}
791
792/// What one block contributed: the mapping, if it was opened, and its matches.
793///
794/// The mapping comes back because the hits are row numbers into it and the
795/// rendering pass at the end of the search needs it still open. Dropping it and
796/// re-opening later would page the block in twice.
797pub(crate) type Scanned = (Option<Block>, Vec<Hit>);
798
799/// The sort key of one hit, which is also the cursor a caller pages on.
800fn cursor(bref: &Src, h: &Hit) -> Cursor {
801    Cursor {
802        ts: h.ts,
803        node: bref.node,
804        seq: bref.seq,
805        row: h.row,
806    }
807}
808
809/// The tables of one block, opened and ready to scan.
810///
811/// No `Mmap` handle is kept: every Arrow buffer read out of a block owns an
812/// `Arc<Mmap>` of its own (see `block::open_table`), so holding the
813/// `RecordBatch` is what holds the mapping. That is also why row data can be
814/// borrowed straight out of these batches with ordinary lifetimes.
815pub(crate) struct Block {
816    signal: Signal,
817    pub(crate) root: RecordBatch,
818    /// Record-level attributes; `parent_id` indexes `root` directly.
819    attrs: Option<Attrs>,
820    resource_attrs: Option<Attrs>,
821    scope_attrs: Option<Attrs>,
822    /// Rows that hang off a root row rather than being one: a span's events and
823    /// its links. Empty for logs.
824    children: Vec<Child>,
825}
826
827/// A child table and the attribute table keyed by its `id`.
828///
829/// Both indexes are built once when the block is opened, because both are read
830/// once per emitted span and once per attribute term: rebuilding either per
831/// span made emission quadratic in the number of spans in the block, which is
832/// the one place a `limit` does not bound the work.
833struct Child {
834    /// What the array is called in the emitted row.
835    label: &'static str,
836    rows: RecordBatch,
837    attrs: Option<Attrs>,
838    /// Child rows grouped by the root row they hang off, indexed by it.
839    by_parent: Vec<Vec<u32>>,
840    /// The root row each child `id` belongs to, indexed by the id. Child ids
841    /// are dense from zero per block, so this is an array store like every
842    /// other join here — but it is not the identity, because `id` numbers a
843    /// child and the row number is where it happens to sit.
844    parent_of_id: Vec<u32>,
845}
846
847/// An attribute table, plus the one fact about it that makes rendering a row
848/// cheap: whether `parent_id` ascends.
849///
850/// Every builder with the [`crate::schema::ATTRS`] shape appends one parent's
851/// attributes in one go, parents in ascending order, so a parent's rows are a
852/// contiguous run and two binary searches find it. [`emit_attrs`] used to scan
853/// the whole table per emitted row per level — on a 330 K-row logs block that
854/// is 66 million comparisons to render a hundred records, and it measured as
855/// *most* of an unfiltered `limit 100`, more than the scan and more than the
856/// paging docs/architecture.md section 11 attributes it to. Same shape
857/// [`Block::emit_children`] already fixed for the child tables; this is the
858/// other half of it.
859///
860/// Checked at open rather than assumed, because a binary search over unsorted
861/// parents does not fail — it silently drops attributes, which is the one
862/// outcome nobody would notice.
863///
864/// [`crate::series`] reads the same shape off its own tables and so uses this
865/// rather than a second copy of the search.
866pub(crate) struct Attrs {
867    pub(crate) rows: RecordBatch,
868    ordered: bool,
869}
870
871impl Attrs {
872    pub(crate) fn new(rows: RecordBatch) -> Attrs {
873        let ordered = Attrs::parents(&rows).is_some_and(|p| p.windows(2).all(|w| w[0] <= w[1]));
874        Attrs { rows, ordered }
875    }
876
877    pub(crate) fn parents(rows: &RecordBatch) -> Option<&[u32]> {
878        rows.column(0)
879            .as_primitive_opt::<UInt32Type>()
880            .map(|c| &**c.values())
881    }
882
883    /// The rows belonging to `parent`, as a range the caller still filters —
884    /// exact when the table is ordered, the whole table when it is not.
885    ///
886    /// Takes the column the caller already downcast rather than repeating it,
887    /// since this runs once per emitted row per level.
888    ///
889    /// ponytail: that fallback is the linear scan this replaced, kept for a
890    /// table no builder in this tree produces. It is O(rows) per emitted row;
891    /// if one ever turns up, the fix is to sort it once at open rather than to
892    /// make this cleverer.
893    pub(crate) fn run(&self, parents: &[u32], parent: u32) -> std::ops::Range<usize> {
894        if !self.ordered {
895            return 0..parents.len();
896        }
897        let lo = parents.partition_point(|&p| p < parent);
898        lo..lo + parents[lo..].partition_point(|&p| p == parent)
899    }
900}
901
902/// The child tables of each signal: emitted name, row table, attribute table.
903///
904/// Links are the *out*-edge of the correlation graph — a link's `trace_id`
905/// points at another trace, usually one this node never saw — so returning them
906/// is the whole point of storing them. Events are a span's log lines and are
907/// what a waterfall shows when a row is expanded.
908fn child_tables(signal: Signal) -> &'static [(&'static str, &'static str, &'static str)] {
909    match signal {
910        Signal::Logs => &[],
911        Signal::Traces => &[
912            ("events", "span_events", "span_event_attrs"),
913            ("links", "span_links", "span_link_attrs"),
914        ],
915    }
916}
917
918impl Block {
919    fn open(bref: &Src, signal: Signal) -> Result<Option<Block>> {
920        let load = |name: &str| bref.load(name);
921
922        // No root table means the directory is being torn down by retention
923        // underneath us. Treat it as absent rather than as an error: racing a
924        // deletion of an expired block is normal, not a query failure.
925        let Some(root) = load(signal.root())? else {
926            return Ok(None);
927        };
928        let mut children = Vec::new();
929        for &(label, table, attrs) in child_tables(signal) {
930            // A block whose spans carried no events writes no `span_events`
931            // file at all (`publish` skips empty tables), which is absence, not
932            // damage.
933            if let Some(rows) = load(table)? {
934                children.push(Child {
935                    label,
936                    by_parent: crate::series::index_by_parent(&rows),
937                    parent_of_id: index_parent_of_id(&rows),
938                    rows,
939                    attrs: load(attrs)?.map(Attrs::new),
940                });
941            }
942        }
943        Ok(Some(Block {
944            signal,
945            attrs: load(signal.attrs())?.map(Attrs::new),
946            resource_attrs: load("resource_attrs")?.map(Attrs::new),
947            scope_attrs: load("scope_attrs")?.map(Attrs::new),
948            children,
949            root,
950        }))
951    }
952
953    /// The root's time column as a raw slice, straight out of the mapping.
954    fn time(&self) -> Option<&[i64]> {
955        self.root
956            .column_by_name(self.signal.time_col())
957            .map(|c| &**c.as_primitive::<TimestampNanosecondType>().values())
958    }
959
960    /// Row numbers of `root` matching the query, ascending.
961    fn select(&self, q: &Search, bref: &Src) -> Vec<u32> {
962        let n = self.root.num_rows();
963        let Some(time) = self.time() else {
964            return Vec::new();
965        };
966
967        // Time first: cheapest filter, and on a block the query only partly
968        // covers, usually the most selective. When the block sits wholly inside
969        // the window the comparison is skipped entirely — the directory name
970        // already proved it, which is the point of putting the range there.
971        // ponytail: the selection is now sized for the whole block even when
972        // the time filter throws most of it away, where the filtered collect
973        // this replaced grew to the match count — 4 bytes per row of one block,
974        // for as long as the block's hits take to build, against a fan-out of
975        // sixteen. Worth it for a 2.6x on the one predicate every query has; if
976        // the transient ever matters, `shrink_to_fit` after the time filter
977        // buys it back for one realloc.
978        let mut sel: Vec<u32> = (0..n as u32).collect();
979        if !(q.from <= bref.min_ts && q.to >= bref.max_ts) {
980            // The `&[i64]` loop docs/architecture.md section 10 names as what
981            // replaces intrinsics on the scan, and the one column every query
982            // has a predicate on. `None` for the validity because the time
983            // column is non-nullable in every signal's schema — and the scan it
984            // replaced read the raw slice too, so a nullable one would compare
985            // the same bytes it always did.
986            keep(&mut sel, None, time, |t| (q.from..=q.to).contains(&t));
987        }
988
989        for term in &q.terms {
990            if sel.is_empty() {
991                break;
992            }
993            match &term.target {
994                Target::Field(name) => {
995                    let ok = self
996                        .root
997                        .column_by_name(name)
998                        .is_some_and(|c| field_filter(&mut sel, c.as_ref(), term.op, &term.value));
999                    // Unknown column, or a value that cannot be compared
1000                    // against this column's type at all.
1001                    if !ok {
1002                        sel.clear();
1003                    }
1004                }
1005                Target::Attr(key) => {
1006                    let matched = self.attr_rows(key, term.op, &term.value, n);
1007                    sel.retain(|&i| matched[i as usize]);
1008                }
1009            }
1010        }
1011        sel
1012    }
1013
1014    /// A bitmap over root rows: does this record carry `key op value` at any
1015    /// attribute level it has — record, resource, scope, or one of its own
1016    /// child rows?
1017    fn attr_rows(&self, key: &str, op: Op, value: &Value, n: usize) -> Vec<bool> {
1018        let mut out = vec![false; n];
1019
1020        // Record level: parent_id *is* the root row number, so this is a store,
1021        // not a join. That is what rebasing ids at ingest bought.
1022        if let Some(a) = &self.attrs {
1023            for pid in attr_parents(&a.rows, key, op, value) {
1024                if let Some(slot) = out.get_mut(pid as usize) {
1025                    *slot = true;
1026                }
1027            }
1028        }
1029
1030        // Resource and scope level: parent_id is an entity id, and a foreign key
1031        // column on the root says which rows point at it. Entity ids are dense
1032        // from zero and number in the tens, so the reverse lookup is a small
1033        // boolean array rather than a hash set.
1034        for (table, fk) in [
1035            (&self.resource_attrs, "resource_id"),
1036            (&self.scope_attrs, "scope_id"),
1037        ] {
1038            let (Some(a), Some(col)) = (table, self.root.column_by_name(fk)) else {
1039                continue;
1040            };
1041            let ids = attr_parents(&a.rows, key, op, value);
1042            let Some(&top) = ids.iter().max() else {
1043                continue;
1044            };
1045            let mut wanted = vec![false; top as usize + 1];
1046            for id in ids {
1047                wanted[id as usize] = true;
1048            }
1049            for (i, &id) in col.as_primitive::<UInt16Type>().values().iter().enumerate() {
1050                if wanted.get(id as usize).copied().unwrap_or(false) {
1051                    out[i] = true;
1052                }
1053            }
1054        }
1055
1056        // Child level: a span's events and links carry attributes of their own,
1057        // and `recordException` is the reason this matters — the OTel API puts
1058        // `exception.type`, `exception.message` and `exception.stacktrace` on an
1059        // *event*, so "which spans threw NullPointerException" is a child-level
1060        // filter and nothing else. The row that matches is the span the event
1061        // hangs off: `attr_parents` gives the event's own id, and
1062        // `parent_of_id` turns that into the root row in one indexed load.
1063        //
1064        // The attribute Bloom sidecar has always covered these tables
1065        // (`attrs::index` walks every table with the ATTRS schema), so a block
1066        // holding the value was already being opened and scanned for it. What
1067        // was missing was the last hop.
1068        for c in &self.children {
1069            let Some(a) = &c.attrs else { continue };
1070            for id in attr_parents(&a.rows, key, op, value) {
1071                // Two `get`s and no `let` chain, for the MSRV reason given in
1072                // `search` above.
1073                if let Some(slot) = c
1074                    .parent_of_id
1075                    .get(id as usize)
1076                    .and_then(|&root| out.get_mut(root as usize))
1077                {
1078                    *slot = true;
1079                }
1080            }
1081        }
1082        out
1083    }
1084
1085    /// Write one root row as a JSON object, attributes merged in and child rows
1086    /// nested under it.
1087    fn emit_row(&self, j: &mut Json, row: u32) {
1088        j.obj(|j| {
1089            emit_fields(j, &self.root, row);
1090            j.key("attributes");
1091            emit_attrs(
1092                j,
1093                &[
1094                    (&self.resource_attrs, self.fk(row, "resource_id")),
1095                    (&self.scope_attrs, self.fk(row, "scope_id")),
1096                    (&self.attrs, Some(row)),
1097                ],
1098            );
1099            for c in &self.children {
1100                self.emit_children(j, c, row);
1101            }
1102        });
1103    }
1104
1105    /// The `events` / `links` array of one span.
1106    ///
1107    /// Reads the index built when the block was opened. It used to be a linear
1108    /// pass over the child table per emitted row, on the argument that `limit`
1109    /// bounds the row count — but the other factor is the block's event count,
1110    /// so the product is `limit` × events-in-block, and a block holds hundreds
1111    /// of thousands of events. One pass at open time replaces all of them.
1112    fn emit_children(&self, j: &mut Json, c: &Child, row: u32) {
1113        // No key at all rather than an empty array: most spans have neither
1114        // events nor links, and two empty arrays per span is most of the
1115        // response.
1116        let hits = match c.by_parent.get(row as usize) {
1117            Some(h) if !h.is_empty() => h,
1118            _ => return,
1119        };
1120        j.key(c.label);
1121        j.arr(|j| {
1122            for &r in hits {
1123                let r = r as usize;
1124                j.obj(|j| {
1125                    emit_fields(j, &c.rows, r as u32);
1126                    // The child's own `id`, which its attribute table keys on —
1127                    // not the span's row number.
1128                    let id = c
1129                        .rows
1130                        .column_by_name("id")
1131                        .map(|col| col.as_primitive::<UInt32Type>().value(r));
1132                    j.key("attributes");
1133                    emit_attrs(j, &[(&c.attrs, id)]);
1134                });
1135            }
1136        });
1137    }
1138
1139    fn fk(&self, row: u32, name: &str) -> Option<u32> {
1140        self.root
1141            .column_by_name(name)
1142            .map(|c| c.as_primitive::<UInt16Type>().value(row as usize) as u32)
1143    }
1144}
1145
1146/// The `parent_id` of each child row, indexed by that row's own `id`.
1147///
1148/// The inverse of [`crate::series::index_by_parent`] and the same trick: ids
1149/// are dense from zero per block, so the id is the slot. A table missing either
1150/// column indexes as empty, and every lookup then misses — which is the same
1151/// answer a scan of it would give.
1152fn index_parent_of_id(b: &RecordBatch) -> Vec<u32> {
1153    let (Some(ids), Some(parents)) = (b.column_by_name("id"), b.column_by_name("parent_id")) else {
1154        return Vec::new();
1155    };
1156    let ids = ids.as_primitive::<UInt32Type>().values();
1157    let parents = parents.as_primitive::<UInt32Type>().values();
1158    let mut out = vec![u32::MAX; ids.iter().copied().max().unwrap_or(0) as usize + 1];
1159    for (&id, &parent) in ids.iter().zip(parents) {
1160        out[id as usize] = parent;
1161    }
1162    out
1163}
1164
1165/// Every non-null column of one row, by name.
1166///
1167/// The block-local ids are skipped: the caller asked for a log line or a span
1168/// event, not for the row numbers that found it.
1169pub(crate) fn emit_fields(j: &mut Json, b: &RecordBatch, row: u32) {
1170    for (i, f) in b.schema().fields().iter().enumerate() {
1171        if matches!(
1172            f.name().as_str(),
1173            "id" | "parent_id" | "resource_id" | "scope_id"
1174        ) {
1175            continue;
1176        }
1177        let col = b.column(i);
1178        if col.is_null(row as usize) {
1179            continue;
1180        }
1181        j.key(f.name());
1182        // `body_ser` is the one Binary column that is not opaque bytes: it holds
1183        // the protobuf encoding of a non-string log body, written precisely so
1184        // that nothing is lost. Sending it through the generic Binary arm below
1185        // renders a map-valued body as a hex dump, which loses it on the way out
1186        // instead of on the way in.
1187        if f.name() == "body_ser" {
1188            emit_any(j, col.as_binary::<i32>().value(row as usize));
1189        } else {
1190            emit_value(j, col.as_ref(), row as usize);
1191        }
1192    }
1193}
1194
1195/// One `{...}` merging the attributes of several levels, most specific last.
1196fn emit_attrs(j: &mut Json, levels: &[(&Option<Attrs>, Option<u32>)]) {
1197    j.obj(|j| {
1198        let mut merged: Vec<(&str, &RecordBatch, usize)> = Vec::new();
1199        for &(table, parent) in levels {
1200            let (Some(a), Some(parent)) = (table, parent) else {
1201                continue;
1202            };
1203            // Empty for a table whose parent column is not a `u32` — which no
1204            // schema in this tree produces, and which then emits nothing rather
1205            // than panicking on a block someone else wrote.
1206            let parents = Attrs::parents(&a.rows).unwrap_or_default();
1207            for r in a.run(parents, parent).filter(|&r| parents[r] == parent) {
1208                merged.push((attr_key(&a.rows, r), &a.rows, r));
1209            }
1210        }
1211        // Sorted so output is deterministic, and stably so that within one key
1212        // the last level pushed — the most specific one — is the entry that
1213        // survives the dedup below.
1214        merged.sort_by_key(|(k, _, _)| *k);
1215        for (i, &(k, a, r)) in merged.iter().enumerate() {
1216            if merged.get(i + 1).is_some_and(|nxt| nxt.0 == k) {
1217                continue;
1218            }
1219            j.key(k);
1220            emit_attr(j, a, r);
1221        }
1222    });
1223}
1224
1225/// `parent_id`s of the attribute rows whose key matches and whose value
1226/// satisfies `op value`.
1227pub(crate) fn attr_parents(a: &RecordBatch, key: &str, op: Op, value: &Value) -> Vec<u32> {
1228    let keys = a.column(1).as_dictionary::<UInt16Type>();
1229    // Resolve the key string once. Everything after this compares u16 codes.
1230    let Some(code) = dict_index(keys.values().as_string::<i32>(), key) else {
1231        return Vec::new();
1232    };
1233    let codes = keys.keys().values();
1234    let parents = &**a.column(0).as_primitive::<UInt32Type>().values();
1235    let types = &**a.column(2).as_primitive::<UInt8Type>().values();
1236    let pred = AttrPred::new(a, op, value);
1237
1238    // Three raw buffers zipped, so the key test — which rejects most rows in a
1239    // table holding every key of every record — is a `u16` compare against a
1240    // slice the bounds check is already gone from.
1241    codes
1242        .iter()
1243        .zip(types)
1244        .zip(parents)
1245        .enumerate()
1246        .filter_map(|(i, ((&c, &ty), &p))| (c == code && pred.test(ty, i)).then_some(p))
1247        .collect()
1248}
1249
1250/// One attribute term with everything that does not depend on the row resolved
1251/// once: the four value columns downcast, the query scalar canonicalized, and —
1252/// for the string column — the predicate already evaluated against the
1253/// dictionary.
1254///
1255/// This used to be `attr_matches`, which did all of it *per row*: two
1256/// `downcast_ref`s, a `String` allocation for [`canon`], and a `parse::<f64>()`
1257/// on every ordered comparison. That measured 30.9 ns per root row against 6.1
1258/// for the allocation-free integer arm, which made an attribute filter fifteen
1259/// times dearer than a column one on the same block.
1260struct AttrPred<'a> {
1261    op: Op,
1262    /// The `str` column's dictionary codes, and which dictionary entries
1263    /// satisfy the predicate.
1264    ///
1265    /// The same trick as [`field_filter`]'s `Dictionary` arm: attribute values
1266    /// are where telemetry repeats (see [`crate::schema::ATTRS`]), so a few
1267    /// hundred string comparisons replace one per row and the row loop never
1268    /// touches string data.
1269    ///
1270    /// ponytail: evaluated over the whole dictionary, which every key in the
1271    /// table shares — so filtering on a rare key in a block whose values are
1272    /// nearly all distinct pays one comparison per distinct value to reject a
1273    /// handful of rows. The ceiling is a value column with no repetition in it,
1274    /// which is the case its dictionary encoding is already the wrong layout
1275    /// for; the upgrade path is a lazily filled memo over the same array.
1276    strs: Option<(&'a [u32], Vec<bool>)>,
1277    ints: Option<(&'a [i64], i64)>,
1278    doubles: Option<(&'a [f64], f64)>,
1279    bools: Option<(&'a BooleanArray, bool)>,
1280}
1281
1282impl AttrPred<'_> {
1283    fn new<'a>(a: &'a RecordBatch, op: Op, v: &Value) -> AttrPred<'a> {
1284        let text = canon(v);
1285        // A quoted query scalar asked for a text comparison and gets one; so
1286        // does anything that is not a number. Decided once rather than per row.
1287        let numeric = !matches!(v, Value::Str(_));
1288        // `_opt` on all four: a table whose columns are not the ATTRS shape
1289        // matches nothing here instead of panicking inside a scan thread.
1290        let strs = a
1291            .column(3)
1292            .as_dictionary_opt::<UInt32Type>()
1293            .and_then(|d| Some((d, d.values().as_string_opt::<i32>()?)))
1294            .map(|(d, values)| {
1295                let ok = (0..values.len())
1296                    .map(|i| {
1297                        let s = values.value(i);
1298                        match op {
1299                            Op::Contains => s.contains(text.as_str()),
1300                            // Equality against a string column is *defined* as
1301                            // equality with `canon`, because that is the text
1302                            // the block index holds (see [`canon`]). Widening
1303                            // it any further — say, matching the stored string
1304                            // "200.0" against `eq: 200` because both parse to
1305                            // the same number — would make the filter prune
1306                            // away blocks that do contain a match, which is the
1307                            // one failure mode a sidecar is not allowed to
1308                            // have.
1309                            Op::Eq | Op::Ne => op.test_ord(s.cmp(text.as_str())),
1310                            // Ordering is not in the index — `attr_probes` only
1311                            // takes `Op::Eq` — so there is nothing here to
1312                            // disagree with, and a number written as a string
1313                            // can be ordered as the number it is. It has to be:
1314                            // half the SDKs that emit
1315                            // `http.response.status_code` emit it as text, and
1316                            // lexicographically "1000" sorts below "400", so
1317                            // `gte: 400` would otherwise mean something
1318                            // different on each of them.
1319                            _ => match (s.parse::<f64>(), v.as_f64()) {
1320                                (Ok(x), Some(y)) if numeric => {
1321                                    x.partial_cmp(&y).is_some_and(|o| op.test_ord(o))
1322                                }
1323                                _ => op.test_ord(s.cmp(text.as_str())),
1324                            },
1325                        }
1326                    })
1327                    .collect();
1328                (&**d.keys().values(), ok)
1329            });
1330        AttrPred {
1331            op,
1332            strs,
1333            ints: a
1334                .column(4)
1335                .as_primitive_opt::<Int64Type>()
1336                .zip(v.as_i64())
1337                .map(|(c, y)| (&**c.values(), y)),
1338            doubles: a
1339                .column(5)
1340                .as_primitive_opt::<Float64Type>()
1341                .zip(v.as_f64())
1342                .map(|(c, y)| (&**c.values(), y)),
1343            bools: a.column(6).as_boolean_opt().zip(v.as_bool()),
1344        }
1345    }
1346
1347    /// Compare one attribute row against the query scalar, dispatching on the
1348    /// stored `type` rather than on the query's — the column decides what it
1349    /// is.
1350    ///
1351    /// Empty, Bytes, Slice and Map are returned in results but not filterable
1352    /// in V0, and a column whose type does not match the schema reads the same
1353    /// way: no match.
1354    fn test(&self, ty: u8, row: usize) -> bool {
1355        const STR: u8 = AttrType::Str as u8;
1356        const INT: u8 = AttrType::Int as u8;
1357        const DOUBLE: u8 = AttrType::Double as u8;
1358        const BOOL: u8 = AttrType::Bool as u8;
1359        match ty {
1360            STR => self.strs.as_ref().is_some_and(|(codes, ok)| {
1361                codes
1362                    .get(row)
1363                    .and_then(|&c| ok.get(c as usize))
1364                    .copied()
1365                    .unwrap_or(false)
1366            }),
1367            INT => self
1368                .ints
1369                .is_some_and(|(xs, y)| xs.get(row).is_some_and(|&x| self.op.test_ord(x.cmp(&y)))),
1370            DOUBLE => self.doubles.is_some_and(|(xs, y)| {
1371                xs.get(row)
1372                    .and_then(|x| x.partial_cmp(&y))
1373                    .is_some_and(|o| self.op.test_ord(o))
1374            }),
1375            BOOL => self
1376                .bools
1377                .is_some_and(|(xs, y)| row < xs.len() && self.op.test_ord(xs.value(row).cmp(&y))),
1378            _ => false,
1379        }
1380    }
1381}
1382
1383pub(crate) fn attr_key(a: &RecordBatch, row: usize) -> &str {
1384    let d = a.column(1).as_dictionary::<UInt16Type>();
1385    d.values()
1386        .as_string::<i32>()
1387        .value(d.keys().value(row) as usize)
1388}
1389
1390/// Position of `needle` in a dictionary's value array.
1391///
1392/// Linear over the dictionary, which is at most 65536 entries and in practice a
1393/// few dozen. Doing it once here is what keeps the row scan off the string data
1394/// entirely.
1395pub(crate) fn dict_index(values: &StringArray, needle: &str) -> Option<u16> {
1396    (0..values.len())
1397        .find(|&i| values.value(i) == needle)
1398        .map(|i| i as u16)
1399}
1400
1401/// Narrow `sel` to the rows of `col` satisfying `op value`.
1402///
1403/// `false` means the query value cannot be compared against this column at all,
1404/// which the caller turns into an empty result. Nothing is written to `sel`
1405/// before that decision, so a refusal leaves it untouched.
1406///
1407/// One monomorphic loop per column type, where this used to build a
1408/// `Box<dyn Fn(u32) -> bool>` and pay an indirect call per row.
1409/// docs/architecture.md section 10 rejects hand-written intrinsics on the scan
1410/// and names what replaces them: "a tight loop over `&[i64]` with no bounds
1411/// checks and no branches, which LLVM turns into NEON unasked". That is what
1412/// [`keep`] is; this function's only job is to hand it a values slice and a
1413/// comparison that inlines into it.
1414fn field_filter(sel: &mut Vec<u32>, col: &dyn Array, op: Op, v: &Value) -> bool {
1415    macro_rules! ints {
1416        ($t:ty) => {{
1417            let Some(target) = v.as_i64() else {
1418                return false;
1419            };
1420            let vals = col.as_primitive::<$t>().values();
1421            keep(sel, col.nulls(), vals, |x| {
1422                op.test_ord((x as i64).cmp(&target))
1423            });
1424        }};
1425    }
1426
1427    match col.data_type() {
1428        DataType::Timestamp(_, _) => ints!(TimestampNanosecondType),
1429        DataType::Int64 => ints!(Int64Type),
1430        DataType::Int32 => ints!(Int32Type),
1431        DataType::UInt64 => ints!(UInt64Type),
1432        DataType::UInt32 => ints!(UInt32Type),
1433        DataType::UInt16 => ints!(UInt16Type),
1434        DataType::UInt8 => ints!(UInt8Type),
1435        DataType::Float64 => {
1436            let Some(target) = v.as_f64() else {
1437                return false;
1438            };
1439            let vals = col.as_primitive::<Float64Type>().values();
1440            keep(sel, col.nulls(), vals, |x: f64| {
1441                x.partial_cmp(&target).is_some_and(|o| op.test_ord(o))
1442            });
1443        }
1444        DataType::Dictionary(_, _) => {
1445            let d = col.as_dictionary::<UInt16Type>();
1446            let values = d.values().as_string::<i32>();
1447            let Some(needle) = v.as_str() else {
1448                return false;
1449            };
1450            // Evaluate the predicate against the dictionary, not the rows: a
1451            // few dozen string comparisons replace one per row, and the scan
1452            // below is a lookup table indexed by a u16.
1453            let ok: Vec<bool> = (0..values.len())
1454                .map(|i| {
1455                    let s = values.value(i);
1456                    match op {
1457                        Op::Contains => s.contains(needle),
1458                        _ => op.test_ord(s.cmp(needle)),
1459                    }
1460                })
1461                .collect();
1462            keep(sel, col.nulls(), d.keys().values(), |c| {
1463                ok.get(c as usize).copied().unwrap_or(false)
1464            });
1465        }
1466        DataType::Boolean => {
1467            let a = col.as_boolean();
1468            let Some(target) = v.as_bool() else {
1469                return false;
1470            };
1471            keep_by(sel, a.len(), col.nulls(), |i| {
1472                op.test_ord(a.value(i).cmp(&target))
1473            });
1474        }
1475        DataType::Utf8 => {
1476            let a = col.as_string::<i32>();
1477            let Some(target) = v.as_str() else {
1478                return false;
1479            };
1480            keep_by(sel, a.len(), col.nulls(), |i| {
1481                let s = a.value(i);
1482                match op {
1483                    Op::Contains => s.contains(target),
1484                    _ => op.test_ord(s.cmp(target)),
1485                }
1486            });
1487        }
1488        DataType::FixedSizeBinary(_) => {
1489            // Trace and span ids: hex in the query, bytes on disk. Decoding the
1490            // needle once beats hex-encoding every row.
1491            let a = col.as_fixed_size_binary();
1492            let Some(target) = v.as_str().and_then(unhex) else {
1493                return false;
1494            };
1495            keep_by(sel, a.len(), col.nulls(), |i| {
1496                op.test_ord(a.value(i).cmp(target.as_slice()))
1497            });
1498        }
1499        _ => return false,
1500    }
1501    true
1502}
1503
1504/// Narrow `sel` to the rows where `p(vals[row])` holds.
1505///
1506/// Three shapes, because the first one is the whole point. A column with no
1507/// nulls, not yet narrowed by an earlier term — which is every first term on a
1508/// block the time range covers whole — walks `vals` contiguously and writes the
1509/// surviving row number unconditionally, advancing the cursor by the boolean.
1510/// No indirect call, no bounds check on the load, and no branch in the body, so
1511/// the comparison itself stays in vector registers.
1512///
1513/// `sel.len() == vals.len()` is what proves the selection is still the identity
1514/// `0..n`: [`Block::select`] only ever removes from it, and it is built
1515/// ascending, so a full-length selection has nothing missing from it.
1516///
1517/// ponytail: the two narrowed shapes stay a gather under `retain` and do not
1518/// vectorise. Making them would mean the selection becoming a bitmap so every
1519/// term reads and writes contiguously — a different engine, and worth it only
1520/// once a query with several selective terms shows up in a profile. Terms are
1521/// applied cheapest-first-by-accident today, and most queries carry one.
1522fn keep<T: Copy>(
1523    sel: &mut Vec<u32>,
1524    nulls: Option<&arrow_buffer::NullBuffer>,
1525    vals: &[T],
1526    p: impl Fn(T) -> bool,
1527) {
1528    match nulls {
1529        None if sel.len() == vals.len() => {
1530            let mut k = 0;
1531            for (i, &x) in vals.iter().enumerate() {
1532                sel[k] = i as u32;
1533                k += p(x) as usize;
1534            }
1535            sel.truncate(k);
1536        }
1537        None => sel.retain(|&i| vals.get(i as usize).is_some_and(|&x| p(x))),
1538        // An `is_null` per row is what kept this loop scalar, which is why the
1539        // case above exists at all. Every column a sealed block filters on is
1540        // non-null in practice; this is the path that stays correct when one is
1541        // not.
1542        Some(n) => {
1543            sel.retain(|&i| n.is_valid(i as usize) && vals.get(i as usize).is_some_and(|&x| p(x)));
1544        }
1545    }
1546}
1547
1548/// [`keep`] for a column with no values slice to walk: a bitmap, a variable
1549/// offset array or a fixed stride, none of which a vector register helps with.
1550/// The win here is only the closure being monomorphic rather than boxed.
1551fn keep_by(
1552    sel: &mut Vec<u32>,
1553    len: usize,
1554    nulls: Option<&arrow_buffer::NullBuffer>,
1555    p: impl Fn(usize) -> bool,
1556) {
1557    match nulls {
1558        None => sel.retain(|&i| (i as usize) < len && p(i as usize)),
1559        Some(n) => sel.retain(|&i| n.is_valid(i as usize) && p(i as usize)),
1560    }
1561}
1562
1563/// Decode hex, either case. `None` on an odd length or any non-hex byte, so a
1564/// malformed trace id matches nothing rather than matching a truncated prefix.
1565pub fn unhex(s: &str) -> Option<Vec<u8>> {
1566    if s.len() % 2 != 0 {
1567        return None;
1568    }
1569    let b = s.as_bytes();
1570    (0..b.len() / 2)
1571        .map(|i| {
1572            let hi = (b[i * 2] as char).to_digit(16)?;
1573            let lo = (b[i * 2 + 1] as char).to_digit(16)?;
1574            Some((hi * 16 + lo) as u8)
1575        })
1576        .collect()
1577}
1578
1579/// Emit one cell of a root table.
1580fn emit_value(j: &mut Json, col: &dyn Array, row: usize) {
1581    match col.data_type() {
1582        // Nanoseconds, never a formatted date: every other representation
1583        // either loses precision or picks a timezone on the user's behalf, and
1584        // the UI formats them, which is where that belongs. As a *string*,
1585        // because that is what OTLP/JSON says a 64-bit integer is and because
1586        // 1.7e18 is twenty times past what a JSON number survives — see
1587        // [`Json::i64_str`].
1588        DataType::Timestamp(_, _) => {
1589            j.i64_str(col.as_primitive::<TimestampNanosecondType>().value(row));
1590        }
1591        DataType::Int64 => j.i64_str(col.as_primitive::<Int64Type>().value(row)),
1592        // 32 bits and narrower stay bare: they fit a double exactly, and a
1593        // reader doing arithmetic on `severity_number` or `status_code` should
1594        // not have to parse it first.
1595        DataType::Int32 => j.i64(col.as_primitive::<Int32Type>().value(row) as i64),
1596        DataType::UInt64 => j.u64_str(col.as_primitive::<UInt64Type>().value(row)),
1597        DataType::UInt32 => j.u64(col.as_primitive::<UInt32Type>().value(row) as u64),
1598        DataType::UInt16 => j.u64(col.as_primitive::<UInt16Type>().value(row) as u64),
1599        DataType::UInt8 => j.u64(col.as_primitive::<UInt8Type>().value(row) as u64),
1600        DataType::Float64 => j.f64(col.as_primitive::<Float64Type>().value(row)),
1601        DataType::Boolean => j.bool(col.as_boolean().value(row)),
1602        DataType::Utf8 => j.str(col.as_string::<i32>().value(row)),
1603        DataType::Binary => j.hex(col.as_binary::<i32>().value(row)),
1604        // `as_fixed_size_binary` panics on a type mismatch, exactly like the
1605        // `as_primitive`, `as_string` and `as_binary` arms above it. The match
1606        // is on `data_type()`, so a mismatch would mean the array disagreeing
1607        // with its own type — a broken Arrow build, not a broken block.
1608        DataType::FixedSizeBinary(_) => j.hex(col.as_fixed_size_binary().value(row)),
1609        DataType::Dictionary(_, _) => {
1610            let d = col.as_dictionary::<UInt16Type>();
1611            j.str(
1612                d.values()
1613                    .as_string::<i32>()
1614                    .value(d.keys().value(row) as usize),
1615            );
1616        }
1617        DataType::List(_) => {
1618            let inner = col.as_list::<i32>().value(row);
1619            j.arr(|j| {
1620                for i in 0..inner.len() {
1621                    if inner.is_null(i) {
1622                        j.null();
1623                    } else {
1624                        emit_value(j, inner.as_ref(), i);
1625                    }
1626                }
1627            });
1628        }
1629        _ => j.null(),
1630    }
1631}
1632
1633/// Emit one attribute's value from whichever column its `type` names.
1634pub(crate) fn emit_attr(j: &mut Json, a: &RecordBatch, row: usize) {
1635    const STR: u8 = AttrType::Str as u8;
1636    const INT: u8 = AttrType::Int as u8;
1637    const DOUBLE: u8 = AttrType::Double as u8;
1638    const BOOL: u8 = AttrType::Bool as u8;
1639    const BYTES: u8 = AttrType::Bytes as u8;
1640    const SLICE: u8 = AttrType::Slice as u8;
1641    const MAP: u8 = AttrType::Map as u8;
1642    match a.column(2).as_primitive::<UInt8Type>().value(row) {
1643        STR => j.str(crate::attrs::str_column(a).value(row)),
1644        // Int attributes go out as JSON strings, because that is the OTLP/JSON
1645        // encoding of an `int64` and principle 3 says OTLP decides the wire
1646        // form. The property is round-trip symmetry, not browser etiquette:
1647        // ingest reads these as strings, so what Mira emits Mira must accept —
1648        // and it is the same property in reverse that makes it safe, since a
1649        // reader that wants the number back has the exact digits rather than
1650        // whatever a double rounded them to.
1651        INT => j.i64_str(a.column(4).as_primitive::<Int64Type>().value(row)),
1652        DOUBLE => j.f64(a.column(5).as_primitive::<Float64Type>().value(row)),
1653        BOOL => j.bool(a.column(6).as_boolean().value(row)),
1654        BYTES => j.hex(a.column(7).as_binary::<i32>().value(row)),
1655        // `ser` holds the protobuf encoding of the whole `AnyValue`, so an array
1656        // or a map is a decode away rather than a reconstruction. They are not
1657        // exotic: `process.command_args` and `http.request.header.*` are arrays
1658        // by semantic convention, and `gen_ai.input.messages` — the first-class
1659        // case in ARCHITECTURE section 1 — is a kvlist.
1660        SLICE | MAP => emit_any(j, a.column(8).as_binary::<i32>().value(row)),
1661        // AttrType::Empty: the key arrived with no value at all, which is what
1662        // `null` means.
1663        _ => j.null(),
1664    }
1665}
1666
1667/// Render a protobuf-encoded `AnyValue` — the attribute `ser` column, and a log
1668/// body that was not a string — as the JSON it describes.
1669///
1670/// A decode failure becomes `null`. These bytes were written by this process
1671/// from an already-decoded message, so a failure here means the block is
1672/// damaged, and one damaged value must not take the whole response with it.
1673pub(crate) fn emit_any(j: &mut Json, bytes: &[u8]) {
1674    match AnyValue::decode(bytes) {
1675        Ok(v) => emit_any_value(j, v.value.as_ref()),
1676        Err(_) => j.null(),
1677    }
1678}
1679
1680/// Recursion is bounded by prost's own decode recursion limit (100), which
1681/// `AnyValue::decode` above has already enforced on these bytes — a hostile
1682/// client cannot nest deeply enough here to reach the stack.
1683fn emit_any_value(j: &mut Json, v: Option<&mira_proto::common::v1::any_value::Value>) {
1684    use mira_proto::common::v1::any_value::Value as Av;
1685    match v {
1686        None => j.null(),
1687        Some(Av::StringValue(s)) => j.str(s),
1688        // A string for the same reason [`emit_attr`]'s `INT` arm is one: these
1689        // bytes are an OTLP `AnyValue`, and OTLP/JSON writes its `int_value` as
1690        // a string.
1691        Some(Av::IntValue(i)) => j.i64_str(*i),
1692        Some(Av::DoubleValue(d)) => j.f64(*d),
1693        Some(Av::BoolValue(b)) => j.bool(*b),
1694        Some(Av::BytesValue(b)) => j.hex(b),
1695        Some(Av::ArrayValue(a)) => j.arr(|j| {
1696            for e in &a.values {
1697                emit_any_value(j, e.value.as_ref());
1698            }
1699        }),
1700        Some(Av::KvlistValue(m)) => j.obj(|j| {
1701            for e in &m.values {
1702                j.key(&e.key);
1703                emit_any_value(j, e.value.as_ref().and_then(|v| v.value.as_ref()));
1704            }
1705        }),
1706    }
1707}
1708
1709#[cfg(test)]
1710mod tests {
1711    use super::*;
1712    use arrow_array::builder::StringDictionaryBuilder;
1713    use arrow_array::{
1714        BinaryArray, BooleanArray, FixedSizeBinaryArray, Float64Array, Int32Array, Int64Array,
1715        ListArray, TimestampNanosecondArray, UInt8Array, UInt16Array, UInt32Array, UInt64Array,
1716    };
1717    use std::sync::Arc;
1718
1719    /// `u32::MAX` for "this column cannot be compared against this value",
1720    /// which the scan turns into an empty result rather than into every row.
1721    fn hits(col: &dyn Array, op: Op, v: &Value) -> Vec<u32> {
1722        let mut sel: Vec<u32> = (0..col.len() as u32).collect();
1723        if field_filter(&mut sel, col, op, v) {
1724            sel
1725        } else {
1726            vec![u32::MAX]
1727        }
1728    }
1729
1730    /// The predicate builder is a type-dispatch table, and a column type missing
1731    /// from it does not error — it returns no rows. So the only way an arm can be
1732    /// wrong and stay quiet is if nothing exercises it.
1733    ///
1734    /// Every arm gets the same three rows — below, equal, above — so one
1735    /// expectation checks the arm, the ordering and the null handling at once.
1736    #[test]
1737    fn every_column_type_compares_the_same_way() {
1738        let cols: Vec<(&str, Arc<dyn Array>)> = vec![
1739            (
1740                "timestamp",
1741                Arc::new(TimestampNanosecondArray::from(vec![
1742                    Some(1),
1743                    Some(2),
1744                    Some(3),
1745                    None,
1746                ])),
1747            ),
1748            (
1749                "i64",
1750                Arc::new(Int64Array::from(vec![Some(1), Some(2), Some(3), None])),
1751            ),
1752            (
1753                "i32",
1754                Arc::new(Int32Array::from(vec![Some(1), Some(2), Some(3), None])),
1755            ),
1756            (
1757                "u64",
1758                Arc::new(UInt64Array::from(vec![Some(1), Some(2), Some(3), None])),
1759            ),
1760            (
1761                "u32",
1762                Arc::new(UInt32Array::from(vec![Some(1), Some(2), Some(3), None])),
1763            ),
1764            (
1765                "u16",
1766                Arc::new(UInt16Array::from(vec![Some(1), Some(2), Some(3), None])),
1767            ),
1768            (
1769                "u8",
1770                Arc::new(UInt8Array::from(vec![Some(1), Some(2), Some(3), None])),
1771            ),
1772            (
1773                "f64",
1774                Arc::new(Float64Array::from(vec![
1775                    Some(1.0),
1776                    Some(2.0),
1777                    Some(3.0),
1778                    None,
1779                ])),
1780            ),
1781        ];
1782        for (name, col) in &cols {
1783            let c = col.as_ref();
1784            assert_eq!(hits(c, Op::Eq, &Value::Int(2)), [1], "{name} eq");
1785            assert_eq!(hits(c, Op::Ne, &Value::Int(2)), [0, 2], "{name} ne");
1786            assert_eq!(hits(c, Op::Lt, &Value::Int(2)), [0], "{name} lt");
1787            assert_eq!(hits(c, Op::Lte, &Value::Int(2)), [0, 1], "{name} lte");
1788            assert_eq!(hits(c, Op::Gt, &Value::Int(2)), [2], "{name} gt");
1789            assert_eq!(hits(c, Op::Gte, &Value::Int(2)), [1, 2], "{name} gte");
1790            // A null is not less than anything, and `ne` is where that bites:
1791            // the naive reading would return it.
1792            assert!(!hits(c, Op::Ne, &Value::Int(9)).contains(&3), "{name} null");
1793            // Quoted, because a browser and an LLM both write "2" as often as 2.
1794            assert_eq!(hits(c, Op::Eq, &Value::Str("2".into())), [1], "{name} str");
1795            // Nothing to compare against: no rows, not an error.
1796            assert_eq!(
1797                hits(c, Op::Eq, &Value::Bool(true)),
1798                [u32::MAX],
1799                "{name} bool"
1800            );
1801        }
1802
1803        // A fractional target against an integer column has no integer to be
1804        // equal to. Truncating would make `duration > 0.5` mean `duration > 0`.
1805        assert_eq!(
1806            hits(cols[1].1.as_ref(), Op::Gt, &Value::Double(1.5)),
1807            [u32::MAX]
1808        );
1809        assert_eq!(hits(cols[1].1.as_ref(), Op::Gt, &Value::Double(2.0)), [2]);
1810        // Floats compare as floats, and NaN is unordered rather than equal.
1811        let f = Float64Array::from(vec![Some(1.5), Some(f64::NAN)]);
1812        assert_eq!(hits(&f, Op::Gt, &Value::Double(1.0)), [0]);
1813        assert_eq!(
1814            hits(&f, Op::Eq, &Value::Double(f64::NAN)),
1815            Vec::<u32>::new()
1816        );
1817
1818        let b = BooleanArray::from(vec![Some(true), Some(false), None]);
1819        assert_eq!(hits(&b, Op::Eq, &Value::Bool(true)), [0]);
1820        assert_eq!(hits(&b, Op::Eq, &Value::Str("false".into())), [1]);
1821        assert_eq!(hits(&b, Op::Eq, &Value::Int(1)), [u32::MAX]);
1822
1823        let s = StringArray::from(vec![Some("alpha"), Some("beta"), None]);
1824        assert_eq!(hits(&s, Op::Eq, &Value::Str("beta".into())), [1]);
1825        assert_eq!(hits(&s, Op::Contains, &Value::Str("et".into())), [1]);
1826        assert_eq!(hits(&s, Op::Lt, &Value::Str("b".into())), [0]);
1827        assert_eq!(hits(&s, Op::Eq, &Value::Int(1)), [u32::MAX]);
1828
1829        let mut d = StringDictionaryBuilder::<UInt16Type>::new();
1830        for v in ["ERROR", "INFO", "ERROR"] {
1831            d.append_value(v);
1832        }
1833        let d = d.finish();
1834        assert_eq!(hits(&d, Op::Eq, &Value::Str("ERROR".into())), [0, 2]);
1835        // Resolved against the dictionary once. A value that is not in it cannot
1836        // match any row, and saying so costs no row scan at all.
1837        assert_eq!(
1838            hits(&d, Op::Eq, &Value::Str("TRACE".into())),
1839            Vec::<u32>::new()
1840        );
1841
1842        // Ids arrive as hex and live as bytes. The needle is decoded once, so a
1843        // needle that is not hex at all is no rows rather than every row.
1844        let ids = FixedSizeBinaryArray::try_from_iter([[1u8, 2], [3, 4]].into_iter()).unwrap();
1845        assert_eq!(hits(&ids, Op::Eq, &Value::Str("0102".into())), [0]);
1846        assert_eq!(hits(&ids, Op::Gt, &Value::Str("0102".into())), [1]);
1847        assert_eq!(hits(&ids, Op::Eq, &Value::Str("zz".into())), [u32::MAX]);
1848        assert_eq!(hits(&ids, Op::Eq, &Value::Str("010".into())), [u32::MAX]);
1849
1850        // A type the table does not know is not a panic and not an error.
1851        let l = ListArray::from_iter_primitive::<Int64Type, _, _>(vec![Some(vec![Some(1)])]);
1852        assert_eq!(hits(&l, Op::Eq, &Value::Int(1)), [u32::MAX]);
1853    }
1854
1855    /// Materialization is the same dispatch table read the other way, and its
1856    /// failure mode is worse: a column emitted under the wrong JSON type is a
1857    /// reader's bug, not ours.
1858    #[test]
1859    fn every_column_type_materializes_as_the_json_type_it_is() {
1860        let cell = |col: &dyn Array| {
1861            let mut j = Json::new();
1862            j.arr(|j| emit_value(j, col, 0));
1863            let s = j.into_string();
1864            s[1..s.len() - 1].to_owned()
1865        };
1866        let l = ListArray::from_iter_primitive::<Int64Type, _, _>(vec![Some(vec![
1867            Some(1),
1868            None,
1869            Some(3),
1870        ])]);
1871        let mut d = StringDictionaryBuilder::<UInt16Type>::new();
1872        d.append_value("ERROR");
1873        let cases: Vec<(Arc<dyn Array>, &str)> = vec![
1874            // The 64-bit trio is quoted and the narrower integers are not. The
1875            // timestamp is the case that would be silently wrong as a number:
1876            // as a double it reads back 1700000000000000000, one nanosecond off
1877            // and no parser anywhere complains.
1878            (
1879                Arc::new(TimestampNanosecondArray::from(vec![
1880                    1_700_000_000_000_000_001i64,
1881                ])),
1882                r#""1700000000000000001""#,
1883            ),
1884            (Arc::new(Int64Array::from(vec![-7i64])), r#""-7""#),
1885            (Arc::new(Int32Array::from(vec![-7i32])), "-7"),
1886            (
1887                Arc::new(UInt64Array::from(vec![u64::MAX])),
1888                r#""18446744073709551615""#,
1889            ),
1890            (Arc::new(UInt32Array::from(vec![7u32])), "7"),
1891            (Arc::new(UInt16Array::from(vec![7u16])), "7"),
1892            (Arc::new(UInt8Array::from(vec![7u8])), "7"),
1893            (Arc::new(Float64Array::from(vec![0.5f64])), "0.5"),
1894            (Arc::new(BooleanArray::from(vec![true])), "true"),
1895            (Arc::new(StringArray::from(vec!["a\"b"])), r#""a\"b""#),
1896            (
1897                Arc::new(BinaryArray::from(vec![&b"\xab\xcd"[..]])),
1898                r#""abcd""#,
1899            ),
1900            // A trace id: hex, and the same hex the query takes as a needle.
1901            (
1902                Arc::new(
1903                    FixedSizeBinaryArray::try_from_iter([[0xabu8, 0xcd]].into_iter()).unwrap(),
1904                ),
1905                r#""abcd""#,
1906            ),
1907            (Arc::new(d.finish()), r#""ERROR""#),
1908            // A null inside a list stays a null; the surrounding array does not
1909            // collapse to one. The elements are `Int64`, so they are quoted for
1910            // the same reason a scalar `Int64` is — the list arm recurses into
1911            // this same table rather than having a second encoding.
1912            (Arc::new(l), r#"["1",null,"3"]"#),
1913        ];
1914        for (col, want) in cases {
1915            assert_eq!(cell(col.as_ref()), want, "{:?}", col.data_type());
1916        }
1917
1918        // A type with no representation is null rather than a guess.
1919        let m = arrow_array::Int8Array::from(vec![1i8]);
1920        assert_eq!(cell(&m), "null");
1921    }
1922
1923    /// Everything the scalar helpers promise in their own doc comments, in one
1924    /// place, because each of them is a coercion rule a query author will hit
1925    /// and none of them is guessable from the type.
1926    #[test]
1927    fn a_scalar_coerces_to_what_the_column_needs_or_to_nothing() {
1928        for (s, want) in [
1929            ("eq", Op::Eq),
1930            ("=", Op::Eq),
1931            ("==", Op::Eq),
1932            ("ne", Op::Ne),
1933            ("!=", Op::Ne),
1934            ("lt", Op::Lt),
1935            ("<", Op::Lt),
1936            ("lte", Op::Lte),
1937            ("<=", Op::Lte),
1938            ("gt", Op::Gt),
1939            (">", Op::Gt),
1940            ("gte", Op::Gte),
1941            (">=", Op::Gte),
1942            ("contains", Op::Contains),
1943            ("~", Op::Contains),
1944        ] {
1945            assert_eq!(Op::parse(s), Some(want), "{s}");
1946        }
1947        assert_eq!(Op::parse("=~"), None);
1948        // Documented as unreachable — every caller handles Contains first — so
1949        // the guarantee is that it stays inert if one day a caller does not.
1950        use std::cmp::Ordering::*;
1951        for ord in [Less, Equal, Greater] {
1952            assert!(!Op::Contains.test_ord(ord));
1953        }
1954
1955        assert_eq!(Value::Int(3).as_i64(), Some(3));
1956        assert_eq!(Value::Double(3.0).as_i64(), Some(3));
1957        assert_eq!(Value::Double(3.5).as_i64(), None);
1958        assert_eq!(Value::Str("3".into()).as_i64(), Some(3));
1959        assert_eq!(Value::Str("3.5".into()).as_i64(), None);
1960        assert_eq!(Value::Bool(true).as_i64(), None);
1961
1962        assert_eq!(Value::Int(3).as_f64(), Some(3.0));
1963        assert_eq!(Value::Double(3.5).as_f64(), Some(3.5));
1964        assert_eq!(Value::Str("3.5".into()).as_f64(), Some(3.5));
1965        assert_eq!(Value::Str("x".into()).as_f64(), None);
1966        assert_eq!(Value::Bool(true).as_f64(), None);
1967
1968        assert_eq!(Value::Bool(false).as_bool(), Some(false));
1969        assert_eq!(Value::Str("true".into()).as_bool(), Some(true));
1970        assert_eq!(Value::Str("false".into()).as_bool(), Some(false));
1971        assert_eq!(Value::Str("TRUE".into()).as_bool(), None);
1972        assert_eq!(Value::Int(1).as_bool(), None);
1973        assert_eq!(Value::Double(1.0).as_bool(), None);
1974
1975        assert_eq!(Value::Str("x".into()).as_str(), Some("x"));
1976        assert_eq!(Value::Int(1).as_str(), None);
1977
1978        // Ids arrive from a URL, a log line or a model, so both cases and
1979        // neither-of-them all have to land somewhere predictable.
1980        assert_eq!(unhex("0aFf"), Some(vec![0x0a, 0xff]));
1981        assert_eq!(unhex(""), Some(vec![]));
1982        assert_eq!(unhex("abc"), None);
1983        assert_eq!(unhex("0g"), None);
1984        assert_eq!(unhex("0 1"), None);
1985
1986        let vals = StringArray::from(vec!["a", "b"]);
1987        assert_eq!(dict_index(&vals, "b"), Some(1));
1988        assert_eq!(dict_index(&vals, "c"), None);
1989    }
1990
1991    /// Every `AnyValue` arm, including the ones no attribute in the scan below
1992    /// carries, and bytes that are not an `AnyValue` at all. The renderer is a
1993    /// recursive decoder over attacker-supplied structure, which is the one
1994    /// shape in this file where a missing arm is a silently wrong answer.
1995    #[test]
1996    fn every_any_value_arm_renders_and_damage_renders_as_null() {
1997        use mira_proto::common::v1::any_value::Value as Av;
1998        use mira_proto::common::v1::{ArrayValue, KeyValue, KeyValueList};
1999
2000        let render = |v: Option<Av>| {
2001            let mut j = Json::new();
2002            emit_any(&mut j, &AnyValue { value: v }.encode_to_vec());
2003            j.into_string()
2004        };
2005        assert_eq!(render(None), "null");
2006        assert_eq!(render(Some(Av::StringValue("s".into()))), r#""s""#);
2007        // Quoted: an `AnyValue`'s `int_value` is an `int64`, and OTLP/JSON
2008        // writes those as strings in both directions.
2009        assert_eq!(render(Some(Av::IntValue(-1))), r#""-1""#);
2010        assert_eq!(render(Some(Av::DoubleValue(0.5))), "0.5");
2011        assert_eq!(render(Some(Av::BoolValue(true))), "true");
2012        assert_eq!(
2013            render(Some(Av::BytesValue(vec![0xbe, 0xef].into()))),
2014            r#""beef""#
2015        );
2016        // Nested both ways round, because the recursion is the only part of
2017        // this that can be wrong in a way a flat value would not show.
2018        assert_eq!(
2019            render(Some(Av::ArrayValue(ArrayValue {
2020                values: vec![
2021                    AnyValue { value: None },
2022                    AnyValue {
2023                        value: Some(Av::KvlistValue(KeyValueList {
2024                            values: vec![KeyValue {
2025                                key: "k".into(),
2026                                value: Some(AnyValue {
2027                                    value: Some(Av::IntValue(2)),
2028                                }),
2029                            }],
2030                        })),
2031                    },
2032                ],
2033            }))),
2034            r#"[null,{"k":"2"}]"#
2035        );
2036
2037        // A field number and wire type no protobuf carries. The bytes only get
2038        // here by being on disk, so this is a damaged block, not a bad request.
2039        let mut j = Json::new();
2040        emit_any(&mut j, &[0xff, 0xff, 0xff]);
2041        assert_eq!(j.into_string(), "null");
2042    }
2043
2044    /// The inverse index a child's attribute join reads: the child's own `id`
2045    /// is the slot and the value is the row it hangs off. A table missing
2046    /// either column indexes as empty and every lookup then misses, which is
2047    /// the same answer scanning it would give and not a panic.
2048    #[test]
2049    fn a_child_row_finds_its_parent_by_its_own_id() {
2050        let ids: Arc<dyn Array> = Arc::new(UInt32Array::from(vec![2u32, 0]));
2051        let parents: Arc<dyn Array> = Arc::new(UInt32Array::from(vec![7u32, 5]));
2052        let b = RecordBatch::try_from_iter([("id", ids.clone()), ("parent_id", parents)]).unwrap();
2053        let idx = index_parent_of_id(&b);
2054        assert_eq!(idx.len(), 3, "dense from zero to the largest id");
2055        assert_eq!(idx[0], 5);
2056        assert_eq!(idx[1], u32::MAX, "an id no row claims has no parent");
2057        assert_eq!(idx[2], 7);
2058
2059        let orphan = RecordBatch::try_from_iter([("id", ids)]).unwrap();
2060        assert!(index_parent_of_id(&orphan).is_empty());
2061    }
2062
2063    /// A block on disk, scanned. The predicate tables above are exercised in
2064    /// isolation; this is the path that reaches them — the time prefilter, the
2065    /// three attribute levels, and the merge that decides which of two levels
2066    /// setting the same key the caller actually sees.
2067    #[test]
2068    fn a_scan_narrows_by_time_then_by_terms_and_the_most_specific_level_wins() {
2069        use mira_proto::collector::logs::v1::ExportLogsServiceRequest;
2070        use mira_proto::common::v1::any_value::Value as Av;
2071        use mira_proto::common::v1::{
2072            AnyValue, ArrayValue, InstrumentationScope, KeyValue, KeyValueList,
2073        };
2074        use mira_proto::logs::v1::{LogRecord, ResourceLogs, ScopeLogs};
2075        use mira_proto::resource::v1::Resource;
2076
2077        let kv = |k: &str, v: Av| KeyValue {
2078            key: k.into(),
2079            value: Some(AnyValue { value: Some(v) }),
2080        };
2081        let base = 1_700_000_000_000_000_000u64;
2082        let req = ExportLogsServiceRequest {
2083            resource_logs: vec![ResourceLogs {
2084                resource: Some(Resource {
2085                    attributes: vec![
2086                        kv("service.name", Av::StringValue("checkout".into())),
2087                        // Also set per-record below, which is the whole point:
2088                        // an SDK default that one record overrides.
2089                        kv("deploy.env", Av::StringValue("prod".into())),
2090                    ],
2091                    ..Default::default()
2092                }),
2093                scope_logs: vec![ScopeLogs {
2094                    scope: Some(InstrumentationScope {
2095                        name: "mira.test".into(),
2096                        attributes: vec![kv("scope.kind", Av::StringValue("lib".into()))],
2097                        ..Default::default()
2098                    }),
2099                    log_records: (0..3)
2100                        .map(|i| LogRecord {
2101                            time_unix_nano: base + i * 1_000_000_000,
2102                            severity_number: 9 + i as i32,
2103                            severity_text: "INFO".into(),
2104                            // OTel Events carry the identity of the event here
2105                            // and not in an attribute; empty on the others,
2106                            // which is how OTLP says "this is a plain log".
2107                            event_name: if i == 0 {
2108                                "user.login".into()
2109                            } else {
2110                                String::new()
2111                            },
2112                            // The last record's body is a map rather than a
2113                            // string, which is the case `body_ser` exists for.
2114                            body: Some(AnyValue {
2115                                value: Some(if i == 2 {
2116                                    Av::KvlistValue(KeyValueList {
2117                                        values: vec![kv(
2118                                            "msg",
2119                                            Av::StringValue("structured body".into()),
2120                                        )],
2121                                    })
2122                                } else {
2123                                    Av::StringValue(format!("line {i}"))
2124                                }),
2125                            }),
2126                            attributes: vec![
2127                                kv("deploy.env", Av::StringValue("canary".into())),
2128                                kv("attempt", Av::IntValue(i as i64)),
2129                                // None of these three is filterable in V0; all
2130                                // three still have to come back in the row.
2131                                kv("payload", Av::BytesValue(vec![0xde, 0xad].into())),
2132                                kv(
2133                                    "tags",
2134                                    Av::ArrayValue(ArrayValue {
2135                                        values: vec![
2136                                            AnyValue {
2137                                                value: Some(Av::StringValue("a".into())),
2138                                            },
2139                                            AnyValue {
2140                                                value: Some(Av::IntValue(7)),
2141                                            },
2142                                        ],
2143                                    }),
2144                                ),
2145                                kv(
2146                                    "gen_ai.input.messages",
2147                                    Av::KvlistValue(KeyValueList {
2148                                        values: vec![kv("role", Av::StringValue("user".into()))],
2149                                    }),
2150                                ),
2151                                // A key that arrived with no value at all.
2152                                // Legal OTLP, and `null` is what it means.
2153                                KeyValue {
2154                                    key: "trace.hint".into(),
2155                                    value: None,
2156                                },
2157                            ],
2158                            ..Default::default()
2159                        })
2160                        .collect(),
2161                    ..Default::default()
2162                }],
2163                ..Default::default()
2164            }],
2165        };
2166
2167        let dir = std::env::temp_dir().join(format!("mira-scan-{}", std::process::id()));
2168        let _ = std::fs::remove_dir_all(&dir);
2169        let mut b = crate::logs::LogsBuilder::new();
2170        b.append_request(&req).unwrap();
2171        let sealed = b.finish().unwrap();
2172        let bref =
2173            crate::block::publish(&dir, "logs", crate::block::node_id("a"), 0, 0, &sealed).unwrap();
2174
2175        let base = base as i64;
2176        let scan = |from: i64, to: i64, terms: Vec<Term>| {
2177            search(
2178                &dir,
2179                &Search {
2180                    signal: Signal::Logs,
2181                    from,
2182                    to,
2183                    terms,
2184                    limit: 100,
2185                    after: None,
2186                },
2187            )
2188            .unwrap()
2189        };
2190        let field = |n: &str, op: Op, v: Value| Term {
2191            target: Target::Field(n.into()),
2192            op,
2193            value: v,
2194        };
2195        let attr = |n: &str, op: Op, v: Value| Term {
2196            target: Target::Attr(n.into()),
2197            op,
2198            value: v,
2199        };
2200
2201        // A window the block only partly covers: the directory name proved the
2202        // block is worth opening, and only then does the timestamp get read.
2203        let r = scan(base + 500_000_000, base + 1_500_000_000, vec![]);
2204        assert_eq!(r.stats.rows_matched, 1, "{}", r.json);
2205        assert!(r.json.contains("line 1"), "{}", r.json);
2206
2207        let all = base + 10_000_000_000;
2208        assert_eq!(scan(base, all, vec![]).stats.rows_matched, 3);
2209
2210        // A column this signal does not have is no rows, not an error and not
2211        // every row — a query spanning signals is a normal thing to try.
2212        assert_eq!(
2213            scan(
2214                base,
2215                all,
2216                vec![field("duration_nano", Op::Gt, Value::Int(0))]
2217            )
2218            .stats
2219            .rows_matched,
2220            0
2221        );
2222        // Same for a value that cannot be compared against the column's type.
2223        assert_eq!(
2224            scan(
2225                base,
2226                all,
2227                vec![field("severity_number", Op::Eq, Value::Bool(true))]
2228            )
2229            .stats
2230            .rows_matched,
2231            0
2232        );
2233        // Once nothing is selected the remaining terms are skipped, so a term
2234        // that would have been expensive costs nothing.
2235        assert_eq!(
2236            scan(
2237                base,
2238                all,
2239                vec![
2240                    field("severity_number", Op::Gt, Value::Int(99)),
2241                    attr("service.name", Op::Eq, Value::Str("checkout".into())),
2242                ]
2243            )
2244            .stats
2245            .rows_matched,
2246            0
2247        );
2248        // Attributes are found without being told which level they live at.
2249        for (k, v) in [("service.name", "checkout"), ("scope.kind", "lib")] {
2250            assert_eq!(
2251                scan(base, all, vec![attr(k, Op::Eq, Value::Str(v.into()))])
2252                    .stats
2253                    .rows_matched,
2254                3,
2255                "{k}"
2256            );
2257        }
2258        assert_eq!(
2259            scan(base, all, vec![attr("attempt", Op::Gte, Value::Int(1))])
2260                .stats
2261                .rows_matched,
2262            2
2263        );
2264        // A bytes attribute is not filterable in V0. No rows, and no panic on
2265        // the way to deciding that.
2266        assert_eq!(
2267            scan(
2268                base,
2269                all,
2270                vec![attr("payload", Op::Eq, Value::Str("dead".into()))]
2271            )
2272            .stats
2273            .rows_matched,
2274            0
2275        );
2276        // That one never opened the block — `attr_probes` answered it from the
2277        // filter. `contains` is not probed, so the same term on the same
2278        // attribute reaches the row scan, which is where the decision that a
2279        // bytes value is not comparable actually has to be made.
2280        let r = scan(
2281            base,
2282            all,
2283            vec![attr("payload", Op::Contains, Value::Str("dead".into()))],
2284        );
2285        assert_eq!(r.stats.blocks_scanned, 1, "no probe, so the block is read");
2286        assert_eq!(r.stats.rows_matched, 0);
2287
2288        let row = scan(base, all, vec![]).json;
2289        // Record level beats resource level for the same key.
2290        assert!(row.contains(r#""deploy.env":"canary""#), "{row}");
2291        assert!(!row.contains("prod"), "{row}");
2292        assert!(row.contains(r#""service.name":"checkout""#), "{row}");
2293        assert!(row.contains(r#""payload":"dead""#), "{row}");
2294        // Slice and Map round-trip through `ser` as the JSON they were, because
2295        // a value nothing can read back is a value that was not stored.
2296        assert!(row.contains(r#""tags":["a","7"]"#), "{row}");
2297        assert!(
2298            row.contains(r#""gen_ai.input.messages":{"role":"user"}"#),
2299            "{row}"
2300        );
2301        // Same for a non-string body, and for the field that says the record is
2302        // an OTel Event rather than a log line.
2303        assert!(
2304            row.contains(r#""body_ser":{"msg":"structured body"}"#),
2305            "{row}"
2306        );
2307        assert!(row.contains(r#""event_name":"user.login""#), "{row}");
2308        // Empty on the wire is absent in the row, not an empty string.
2309        assert_eq!(row.matches("event_name").count(), 1, "{row}");
2310        // An attribute that arrived with no value is a key with a `null`, not a
2311        // key that vanished: the exporter sent it, so the record has it.
2312        assert!(row.contains(r#""trace.hint":null"#), "{row}");
2313
2314        // A block written before `event_name` was added to the schema. Once a
2315        // block is on disk a schema change is not revertible, so the claim that
2316        // the reader is `column_by_name` all the way down has to be checked
2317        // rather than asserted: dropping the column reproduces the old writer
2318        // exactly, and the only difference in the answer must be the field.
2319        let table = bref.dir.join("logs.arrow");
2320        let mut old = block::open_table_opt(&table).unwrap().unwrap().batches[0].clone();
2321        old.remove_column(old.schema().index_of("event_name").unwrap());
2322        // Staged and renamed rather than truncated in place: a mapping over a
2323        // truncated file is a SIGBUS, which is why `write_table` says so.
2324        let staged = bref.dir.join("logs.arrow.new");
2325        crate::block::write_table(&staged, &old).unwrap();
2326        std::fs::rename(&staged, &table).unwrap();
2327        let r = scan(base, all, vec![]);
2328        assert_eq!(r.stats.rows_matched, 3, "{}", r.json);
2329        assert!(!r.json.contains("event_name"), "{}", r.json);
2330        assert!(
2331            r.json.contains(r#""body_ser":{"msg":"structured body"}"#),
2332            "{}",
2333            r.json
2334        );
2335
2336        // One column deeper, and the one with no fallback: the time column is
2337        // what `select` filters on before it looks at anything else. A root
2338        // table without it selects nothing, so a table this build cannot make
2339        // sense of costs its own rows and no more.
2340        let mut old = block::open_table_opt(&table).unwrap().unwrap().batches[0].clone();
2341        old.remove_column(old.schema().index_of("time_unix_nano").unwrap());
2342        crate::block::write_table(&staged, &old).unwrap();
2343        std::fs::rename(&staged, &table).unwrap();
2344        let r = scan(base, all, vec![]);
2345        assert_eq!(r.json, "[]");
2346        assert_eq!(r.stats.rows_matched, 0);
2347
2348        // Retention can delete a block between the directory listing and the
2349        // read. The block still counts as present and simply contributes
2350        // nothing, rather than failing the query.
2351        std::fs::remove_file(bref.dir.join("logs.arrow")).unwrap();
2352        let r = scan(base, all, vec![]);
2353        assert_eq!((r.stats.blocks_total, r.stats.blocks_scanned), (1, 0));
2354        assert_eq!(r.json, "[]");
2355    }
2356
2357    /// The fourth attribute level: a span's own events and links.
2358    ///
2359    /// This is not a completeness exercise. The OTel API puts the fields anyone
2360    /// actually searches for on an *event* — `recordException` writes
2361    /// `exception.type`, `exception.message` and `exception.stacktrace` onto one
2362    /// — so "which spans threw a NullPointerException" is a child-level filter
2363    /// and there is no span-level equivalent to fall back on. The row came back
2364    /// with the value visible under `events[].attributes` while a filter for the
2365    /// same key returned nothing, which is the worst shape a search can have.
2366    #[test]
2367    fn an_attribute_on_a_span_event_or_link_selects_the_span_it_hangs_off() {
2368        use mira_proto::collector::trace::v1::ExportTraceServiceRequest;
2369        use mira_proto::common::v1::any_value::Value as Av;
2370        use mira_proto::common::v1::{AnyValue, InstrumentationScope, KeyValue};
2371        use mira_proto::resource::v1::Resource;
2372        use mira_proto::trace::v1::span::{Event, Link};
2373        use mira_proto::trace::v1::{ResourceSpans, ScopeSpans, Span};
2374
2375        let kv = |k: &str, v: &str| KeyValue {
2376            key: k.into(),
2377            value: Some(AnyValue {
2378                value: Some(Av::StringValue(v.into())),
2379            }),
2380        };
2381        let event = |name: &str, attrs: Vec<KeyValue>| Event {
2382            time_unix_nano: 10_050,
2383            name: name.into(),
2384            attributes: attrs,
2385            ..Default::default()
2386        };
2387        let span = |i: u64, events: Vec<Event>, links: Vec<Link>| Span {
2388            trace_id: vec![1u8; 16].into(),
2389            span_id: vec![i as u8 + 1; 8].into(),
2390            name: format!("span {i}"),
2391            start_time_unix_nano: 10_000 + i,
2392            end_time_unix_nano: 10_100 + i,
2393            attributes: vec![kv("http.method", "GET")],
2394            events,
2395            links,
2396            ..Default::default()
2397        };
2398
2399        // Span 0 carries two events and span 1 carries the throw, so the
2400        // throwing event's own id is 2 while the span it belongs to is row 1.
2401        // An implementation that treats a child id as a root row number — the
2402        // obvious mistake, since every other join in this file is exactly that —
2403        // answers "span 2" here and passes any test where each span has one
2404        // event.
2405        let req = ExportTraceServiceRequest {
2406            resource_spans: vec![ResourceSpans {
2407                resource: Some(Resource {
2408                    attributes: vec![kv("service.name", "checkout")],
2409                    ..Default::default()
2410                }),
2411                scope_spans: vec![ScopeSpans {
2412                    scope: Some(InstrumentationScope {
2413                        name: "mira.test".into(),
2414                        ..Default::default()
2415                    }),
2416                    spans: vec![
2417                        span(
2418                            0,
2419                            vec![
2420                                event("cache.miss", vec![kv("cache.key", "cart:7")]),
2421                                event("retrying", vec![]),
2422                            ],
2423                            vec![],
2424                        ),
2425                        span(
2426                            1,
2427                            vec![event(
2428                                "exception",
2429                                vec![kv("exception.type", "NullPointerException")],
2430                            )],
2431                            vec![],
2432                        ),
2433                        span(
2434                            2,
2435                            vec![],
2436                            vec![Link {
2437                                trace_id: vec![9u8; 16].into(),
2438                                span_id: vec![8u8; 8].into(),
2439                                attributes: vec![kv("link.kind", "follows_from")],
2440                                ..Default::default()
2441                            }],
2442                        ),
2443                    ],
2444                    ..Default::default()
2445                }],
2446                ..Default::default()
2447            }],
2448        };
2449
2450        let dir = std::env::temp_dir().join(format!("mira-child-attr-{}", std::process::id()));
2451        let _ = std::fs::remove_dir_all(&dir);
2452        let mut b = crate::traces::TracesBuilder::new();
2453        b.append_request(&req).unwrap();
2454        let sealed = b.finish().unwrap();
2455        let bref = crate::block::publish(&dir, "traces", crate::block::node_id("a"), 0, 0, &sealed)
2456            .unwrap();
2457
2458        let find = |key: &str, value: &str| {
2459            search(
2460                &dir,
2461                &Search {
2462                    signal: Signal::Traces,
2463                    from: 0,
2464                    to: i64::MAX,
2465                    terms: vec![Term {
2466                        target: Target::Attr(key.into()),
2467                        op: Op::Eq,
2468                        value: Value::Str(value.into()),
2469                    }],
2470                    limit: 10,
2471                    after: None,
2472                },
2473            )
2474            .unwrap()
2475        };
2476
2477        // The case the bug report named, and the two either side of it: an
2478        // event attribute on a span that has siblings, an event attribute on a
2479        // span whose events are not the first in the block, and a *link*
2480        // attribute, which travels the same path through a different table.
2481        for (key, value, want) in [
2482            ("exception.type", "NullPointerException", "span 1"),
2483            ("cache.key", "cart:7", "span 0"),
2484            ("link.kind", "follows_from", "span 2"),
2485        ] {
2486            let r = find(key, value);
2487            assert_eq!(r.stats.rows_matched, 1, "{key}: {}", r.json);
2488            assert!(
2489                r.json.contains(&format!(r#""name":"{want}""#)),
2490                "{}",
2491                r.json
2492            );
2493        }
2494
2495        // The other three levels still answer, and a value nothing carries at
2496        // any of the four still matches nothing — the union must widen the
2497        // search, not defeat it.
2498        assert_eq!(find("http.method", "GET").stats.rows_matched, 3);
2499        assert_eq!(find("service.name", "checkout").stats.rows_matched, 3);
2500        assert_eq!(find("exception.type", "IOError").stats.rows_matched, 0);
2501
2502        // And the value is still rendered where it was found, so the filter and
2503        // the row now agree about where an event attribute lives.
2504        let r = find("exception.type", "NullPointerException");
2505        assert!(
2506            r.json
2507                .contains(r#""attributes":{"exception.type":"NullPointerException"}"#),
2508            "{}",
2509            r.json
2510        );
2511
2512        // Rewrite one table without one column, the way a writer that predates
2513        // it left the block. Staged and renamed rather than edited in place: a
2514        // mapping over a file being shortened is a SIGBUS, not an error.
2515        let strip = |table: &str, col: &str| {
2516            let path = bref.dir.join(format!("{table}.arrow"));
2517            let mut b = block::open_table_opt(&path).unwrap().unwrap().batches[0].clone();
2518            b.remove_column(b.schema().index_of(col).unwrap());
2519            let staged = bref.dir.join(format!("{table}.staged"));
2520            crate::block::write_table(&staged, &b).unwrap();
2521            std::fs::rename(&staged, &path).unwrap();
2522        };
2523
2524        // The last hop above is an index built from `span_events.parent_id`. Without
2525        // that column every lookup has to miss, which is the answer a scan of
2526        // the table would give — and emphatically not row zero, which is what
2527        // an index defaulting to the dense-id trick would hand back.
2528        strip("span_events", "parent_id");
2529        let r = find("exception.type", "NullPointerException");
2530        assert_eq!(r.stats.rows_matched, 0, "{}", r.json);
2531        assert!(
2532            !r.json.contains("span"),
2533            "an unjoinable event picked a span"
2534        );
2535        // The levels that do not go through that index are untouched, so this
2536        // is one broken join and not a broken block.
2537        assert_eq!(find("link.kind", "follows_from").stats.rows_matched, 1);
2538        assert_eq!(find("http.method", "GET").stats.rows_matched, 3);
2539
2540        // The root table without the column every scan starts from. No rows —
2541        // not the whole block unfiltered, which is what a scan that treated a
2542        // missing time column as "no time filter" would return.
2543        strip("spans", "start_time_unix_nano");
2544        let r = find("http.method", "GET");
2545        assert_eq!(r.stats.blocks_scanned, 1, "the block was still opened");
2546        assert_eq!(r.json, "[]");
2547        assert_eq!(r.stats.rows_matched, 0);
2548
2549        let _ = std::fs::remove_dir_all(&dir);
2550    }
2551
2552    /// Empty, bytes, array and map attributes come back in the row and match no
2553    /// filter, and those two halves have to stay in step.
2554    ///
2555    /// `gen_ai.input.messages` is a kvlist and is the first-class case in
2556    /// ARCHITECTURE section 1, so these are not exotic types nobody stores. V0 cannot
2557    /// order or substring-match them, and the only safe answer to a filter on
2558    /// one is *no rows*: matching everything would silently widen a query, and
2559    /// the row still carries the value so a caller can see what is there.
2560    ///
2561    /// `Op::Contains` rather than `Op::Eq` on purpose — an `Eq` term builds an
2562    /// attribute Bloom probe that prunes the block before the comparison is
2563    /// reached, so it never proves what the comparison does.
2564    #[test]
2565    fn an_attribute_v0_cannot_filter_still_renders_and_still_matches_nothing() {
2566        use mira_proto::collector::logs::v1::ExportLogsServiceRequest;
2567        use mira_proto::common::v1::any_value::Value as Av;
2568        use mira_proto::common::v1::{AnyValue, ArrayValue, KeyValue, KeyValueList};
2569        use mira_proto::logs::v1::{LogRecord, ResourceLogs, ScopeLogs};
2570
2571        let kv = |k: &str, v: Option<Av>| KeyValue {
2572            key: k.into(),
2573            value: v.map(|v| AnyValue { value: Some(v) }),
2574        };
2575        let req = ExportLogsServiceRequest {
2576            resource_logs: vec![ResourceLogs {
2577                scope_logs: vec![ScopeLogs {
2578                    log_records: vec![LogRecord {
2579                        time_unix_nano: 1_000,
2580                        attributes: vec![
2581                            // No value at all on the wire. Not an empty string,
2582                            // and not the key being absent.
2583                            kv("note", None),
2584                            kv("payload", Some(Av::BytesValue(vec![0xde, 0xad].into()))),
2585                            kv(
2586                                "tags",
2587                                Some(Av::ArrayValue(ArrayValue {
2588                                    values: vec![AnyValue {
2589                                        value: Some(Av::StringValue("a".into())),
2590                                    }],
2591                                })),
2592                            ),
2593                            kv(
2594                                "gen_ai.input.messages",
2595                                Some(Av::KvlistValue(KeyValueList {
2596                                    values: vec![kv("role", Some(Av::StringValue("user".into())))],
2597                                })),
2598                            ),
2599                            // The control: a type V0 *can* filter, carrying a
2600                            // value every probe below is a substring of.
2601                            kv("level", Some(Av::StringValue("a dead role".into()))),
2602                        ],
2603                        ..Default::default()
2604                    }],
2605                    ..Default::default()
2606                }],
2607                ..Default::default()
2608            }],
2609        };
2610
2611        let dir = std::env::temp_dir().join(format!("mira-unfilterable-{}", std::process::id()));
2612        let _ = std::fs::remove_dir_all(&dir);
2613        let mut b = crate::logs::LogsBuilder::new();
2614        b.append_request(&req).unwrap();
2615        let sealed = b.finish().unwrap();
2616        crate::block::publish(&dir, "logs", crate::block::node_id("a"), 0, 0, &sealed).unwrap();
2617
2618        let scan = |terms: Vec<Term>| {
2619            search(
2620                &dir,
2621                &Search {
2622                    signal: Signal::Logs,
2623                    from: 0,
2624                    to: i64::MAX,
2625                    terms,
2626                    limit: 10,
2627                    after: None,
2628                },
2629            )
2630            .unwrap()
2631        };
2632        let contains = |k: &str, v: &str| {
2633            vec![Term {
2634                target: Target::Attr(k.into()),
2635                op: Op::Contains,
2636                value: Value::Str(v.into()),
2637            }]
2638        };
2639
2640        // The same operator and a probe that really is inside the stored value,
2641        // so a zero here is the attribute's *type* and not the operator or the
2642        // needle.
2643        assert_eq!(scan(contains("level", "dead")).stats.rows_matched, 1);
2644        for (key, needle) in [
2645            ("note", ""),
2646            ("payload", "dead"),
2647            ("tags", "a"),
2648            ("gen_ai.input.messages", "role"),
2649        ] {
2650            let r = scan(contains(key, needle));
2651            assert_eq!(r.stats.rows_matched, 0, "{key}: {}", r.json);
2652            // The block was opened and the comparison really ran: a zero that
2653            // came from the sidecar pruning the block would prove nothing about
2654            // what the comparison decides.
2655            assert_eq!(r.stats.blocks_scanned, 1, "{key}");
2656        }
2657
2658        // And every one of them is still in the row. An empty attribute renders
2659        // as `null`, which is what "the key arrived with no value" means — the
2660        // key disappearing would be a different statement.
2661        let row = scan(Vec::new()).json;
2662        assert!(row.contains(r#""note":null"#), "{row}");
2663        assert!(row.contains(r#""payload":"dead""#), "{row}");
2664        assert!(row.contains(r#""tags":["a"]"#), "{row}");
2665        assert!(
2666            row.contains(r#""gen_ai.input.messages":{"role":"user"}"#),
2667            "{row}"
2668        );
2669
2670        // The index that turns a child id into a root row, on a table that has
2671        // neither column: empty, so every lookup misses. The alternative — a
2672        // vector sized off whichever column *is* present — would join rows to
2673        // whatever happened to sit at that slot.
2674        let ids = Arc::new(UInt32Array::from(vec![0u32, 1])) as Arc<dyn Array>;
2675        let only_ids = RecordBatch::try_from_iter(vec![("id", ids.clone())]).unwrap();
2676        let only_parents = RecordBatch::try_from_iter(vec![("parent_id", ids)]).unwrap();
2677        assert!(index_parent_of_id(&only_ids).is_empty());
2678        assert!(index_parent_of_id(&only_parents).is_empty());
2679
2680        let _ = std::fs::remove_dir_all(&dir);
2681    }
2682
2683    /// `rows_matched` is a property of the query, not of the page.
2684    ///
2685    /// The UI prints it next to the page and the MCP tool hands it to a model as
2686    /// "how much there was", so a number that counts rows an earlier page
2687    /// already delivered reads as a result set that grows as you page through
2688    /// it.
2689    #[test]
2690    fn rows_matched_counts_what_is_left_behind_the_cursor() {
2691        use mira_proto::collector::logs::v1::ExportLogsServiceRequest;
2692        use mira_proto::common::v1::any_value::Value as Av;
2693        use mira_proto::common::v1::{AnyValue, InstrumentationScope};
2694        use mira_proto::logs::v1::{LogRecord, ResourceLogs, ScopeLogs};
2695
2696        let base = 1_700_000_000_000_000_000u64;
2697        let req = ExportLogsServiceRequest {
2698            resource_logs: vec![ResourceLogs {
2699                scope_logs: vec![ScopeLogs {
2700                    scope: Some(InstrumentationScope {
2701                        name: "mira.test".into(),
2702                        ..Default::default()
2703                    }),
2704                    log_records: (0..5)
2705                        .map(|i| LogRecord {
2706                            time_unix_nano: base + i * 1_000_000_000,
2707                            body: Some(AnyValue {
2708                                value: Some(Av::StringValue(format!("line {i}"))),
2709                            }),
2710                            ..Default::default()
2711                        })
2712                        .collect(),
2713                    ..Default::default()
2714                }],
2715                ..Default::default()
2716            }],
2717        };
2718
2719        let dir = std::env::temp_dir().join(format!("mira-paged-count-{}", std::process::id()));
2720        let _ = std::fs::remove_dir_all(&dir);
2721        let mut b = crate::logs::LogsBuilder::new();
2722        b.append_request(&req).unwrap();
2723        let sealed = b.finish().unwrap();
2724        crate::block::publish(&dir, "logs", crate::block::node_id("a"), 0, 0, &sealed).unwrap();
2725
2726        let page = |after: Option<Cursor>| {
2727            search(
2728                &dir,
2729                &Search {
2730                    signal: Signal::Logs,
2731                    from: 0,
2732                    to: i64::MAX,
2733                    terms: Vec::new(),
2734                    limit: 2,
2735                    after,
2736                },
2737            )
2738            .unwrap()
2739        };
2740
2741        // Five rows, two per page. Each page reports what is still ahead of the
2742        // reader including the rows it is holding, so the sequence falls by the
2743        // page size and the last page is exact rather than being the whole
2744        // block over again.
2745        let mut cursor = None;
2746        for want in [5, 3, 1] {
2747            let r = page(cursor);
2748            assert_eq!(r.stats.rows_matched, want, "{}", r.json);
2749            cursor = r.next;
2750        }
2751        // A short page is the last page, so there is nothing to ask for again.
2752        assert_eq!(cursor, None);
2753
2754        let _ = std::fs::remove_dir_all(&dir);
2755    }
2756
2757    /// [`keep`] is three loops where there used to be one boxed closure, and
2758    /// only one of them is reachable from a full selection — so the thing to
2759    /// pin is that all three answer the same question.
2760    ///
2761    /// The contiguous branch-free loop is taken when the column has no nulls
2762    /// *and* nothing has narrowed the selection yet; a second term on the same
2763    /// block, or any null at all, gathers through `sel` instead. Every
2764    /// column-type case above enters through the first of those, which is
2765    /// exactly why the other two need their own test.
2766    #[test]
2767    fn narrowing_a_selection_agrees_with_scanning_one_whole() {
2768        let plain = Int64Array::from(vec![1, 2, 3, 4, 5, 6]);
2769        let holed = Int64Array::from(vec![Some(1), None, Some(3), Some(4), None, Some(6)]);
2770        let gt2 = |sel: &mut Vec<u32>, col: &dyn Array| {
2771            assert!(field_filter(sel, col, Op::Gt, &Value::Int(2)));
2772        };
2773
2774        let mut sel: Vec<u32> = (0..6).collect();
2775        gt2(&mut sel, &plain);
2776        assert_eq!(sel, [2, 3, 4, 5], "whole block, no nulls");
2777
2778        let mut sel = vec![1, 3, 5];
2779        gt2(&mut sel, &plain);
2780        assert_eq!(sel, [3, 5], "already narrowed by an earlier term");
2781
2782        let mut sel: Vec<u32> = (0..6).collect();
2783        gt2(&mut sel, &holed);
2784        assert_eq!(sel, [2, 3, 5], "a null is not a match, whole block");
2785
2786        let mut sel = vec![1, 3, 4];
2787        gt2(&mut sel, &holed);
2788        assert_eq!(sel, [3], "a null is not a match, narrowed");
2789
2790        // Nothing selected stays nothing selected — the length test that picks
2791        // the fast path must not read an empty selection as a full one.
2792        let mut sel: Vec<u32> = Vec::new();
2793        gt2(&mut sel, &plain);
2794        assert!(sel.is_empty());
2795
2796        // A refusal writes nothing: `select` clears the selection itself, and
2797        // a half-filtered one left behind would be an answer, not an empty
2798        // result.
2799        let mut sel: Vec<u32> = (0..6).collect();
2800        assert!(!field_filter(
2801            &mut sel,
2802            &plain,
2803            Op::Eq,
2804            &Value::Str("beta".into())
2805        ));
2806        assert_eq!(sel.len(), 6);
2807
2808        // And the same both ways round: a number against a dictionary column
2809        // has no encoding to look up, so it is a refusal rather than a miss.
2810        // `severity_text: 2` must not quietly mean "no logs", which is what a
2811        // plain `false` per row would have made it.
2812        let mut d = StringDictionaryBuilder::<UInt16Type>::new();
2813        d.append_value("warn");
2814        d.append_value("info");
2815        let dict = d.finish();
2816        let mut sel: Vec<u32> = (0..2).collect();
2817        assert!(!field_filter(&mut sel, &dict, Op::Eq, &Value::Int(2)));
2818        assert_eq!(sel.len(), 2);
2819    }
2820
2821    /// Rendering a row binary-searches its attributes, which is only right
2822    /// because every builder appends parents in ascending order.
2823    ///
2824    /// A table that is not in that order has to keep working, and the reason is
2825    /// the failure mode rather than the likelihood: a binary search over
2826    /// unsorted parents does not fail, it returns some of the rows, and an
2827    /// attribute quietly missing from a response is what nobody would notice.
2828    #[test]
2829    fn an_attribute_table_out_of_parent_order_falls_back_to_the_scan() {
2830        let table = |parents: &'static [u32]| {
2831            let mut b = crate::attrs::AttrsBuilder::new("t.key");
2832            for &p in parents {
2833                b.append(p, "k", None).unwrap();
2834            }
2835            let a = Attrs::new(b.finish().unwrap());
2836            // The slice `run` is handed below is the caller's own, so pin it to
2837            // the column the builder actually wrote before trusting it.
2838            assert_eq!(Attrs::parents(&a.rows), Some(parents));
2839            (a, parents)
2840        };
2841
2842        let (ordered, p) = table(&[0, 0, 1, 3, 3]);
2843        assert!(ordered.ordered);
2844        assert_eq!(ordered.run(p, 0), 0..2);
2845        assert_eq!(ordered.run(p, 1), 2..3);
2846        // A parent with no attributes of its own, and one past the end: both
2847        // are empty runs rather than a panic or somebody else's rows.
2848        assert_eq!(ordered.run(p, 2), 3..3);
2849        assert_eq!(ordered.run(p, 9), 5..5);
2850
2851        let (jumbled, p) = table(&[3, 0, 1, 0, 3]);
2852        assert!(!jumbled.ordered);
2853        // The whole table, which the caller still filters row by row — the
2854        // scan this replaced, reached only by a block nothing here writes.
2855        assert_eq!(jumbled.run(p, 0), 0..5);
2856
2857        // An empty table is ordered by vacuous truth and has no rows for
2858        // anyone, which is the same answer either way.
2859        let (empty, p) = table(&[]);
2860        assert_eq!(empty.run(p, 0), 0..0);
2861    }
2862
2863    /// One log block of `rows` records, through the real encoder, held in
2864    /// memory as an [`Open`] snapshot so a scan of it touches no file.
2865    fn bench_block(rows: usize) -> Open {
2866        use mira_proto::collector::logs::v1::ExportLogsServiceRequest;
2867        use mira_proto::common::v1::{InstrumentationScope, KeyValue, any_value};
2868        use mira_proto::logs::v1::{LogRecord, ResourceLogs, ScopeLogs};
2869        use mira_proto::resource::v1::Resource;
2870
2871        let sv = |k: &str, v: &str| KeyValue {
2872            key: k.into(),
2873            value: Some(AnyValue {
2874                value: Some(any_value::Value::StringValue(v.into())),
2875            }),
2876        };
2877        const SEV: [(i32, &str); 4] = [(5, "DEBUG"), (9, "INFO"), (13, "WARN"), (17, "ERROR")];
2878        const ROUTES: [&str; 6] = [
2879            "/checkout",
2880            "/cart",
2881            "/search",
2882            "/api/v1/orders",
2883            "/healthz",
2884            "/metrics",
2885        ];
2886
2887        let mut b = crate::logs::LogsBuilder::new();
2888        let mut done = 0usize;
2889        while done < rows {
2890            let take = (rows - done).min(8192);
2891            let log_records = (0..take)
2892                .map(|k| {
2893                    let i = done + k;
2894                    LogRecord {
2895                        time_unix_nano: 1_700_000_000_000_000_000 + i as u64 * 1_000,
2896                        severity_number: SEV[i % 4].0,
2897                        severity_text: SEV[i % 4].1.into(),
2898                        body: Some(AnyValue {
2899                            value: Some(any_value::Value::StringValue(format!(
2900                                "GET /api/v1/orders/{i} -> 200 in {}ms for tenant t{}",
2901                                i % 997,
2902                                i % 64
2903                            ))),
2904                        }),
2905                        attributes: vec![
2906                            sv("http.route", ROUTES[i % ROUTES.len()]),
2907                            KeyValue {
2908                                key: "http.status_code".into(),
2909                                value: Some(AnyValue {
2910                                    value: Some(any_value::Value::IntValue(
2911                                        200 + (i % 5) as i64 * 100,
2912                                    )),
2913                                }),
2914                            },
2915                        ],
2916                        ..Default::default()
2917                    }
2918                })
2919                .collect();
2920            b.append_request(&ExportLogsServiceRequest {
2921                resource_logs: vec![ResourceLogs {
2922                    resource: Some(Resource {
2923                        attributes: vec![sv("service.name", "checkout"), sv("env", "prod")],
2924                        ..Default::default()
2925                    }),
2926                    scope_logs: vec![ScopeLogs {
2927                        scope: Some(InstrumentationScope {
2928                            name: "http".into(),
2929                            attributes: vec![sv("tier", "gold")],
2930                            ..Default::default()
2931                        }),
2932                        log_records,
2933                        ..Default::default()
2934                    }],
2935                    ..Default::default()
2936                }],
2937            })
2938            .unwrap();
2939            done += take;
2940        }
2941        Open {
2942            node: 1,
2943            seq: 1,
2944            sealed: b.finish().unwrap(),
2945        }
2946    }
2947
2948    /// What one row of one block costs to filter, per predicate kind, with no
2949    /// paging and no CRC in the number.
2950    ///
2951    /// Scaled rather than `#[ignore]`d: the default 4096 rows run in the normal
2952    /// suite, so every line here is a covered correctness check on the six
2953    /// predicate shapes, and the same code is the measurement at a real row
2954    /// count.
2955    ///
2956    /// ```sh
2957    /// MIRA_BENCH_ROWS=2000000 cargo test --release -p miradb-core \
2958    ///     --lib scan_cost_per_row -- --nocapture
2959    /// ```
2960    #[test]
2961    fn scan_cost_per_row() {
2962        let rows: usize = std::env::var("MIRA_BENCH_ROWS")
2963            .ok()
2964            .and_then(|s| s.parse().ok())
2965            .unwrap_or(4_096);
2966        let open = bench_block(rows);
2967        let (lo, hi) = (open.sealed.min_ts, open.sealed.max_ts);
2968        let src = Src::open(&open);
2969        let b = Block::open(&src, Signal::Logs).unwrap().unwrap();
2970        let n = b.root.num_rows();
2971        assert_eq!(n, rows);
2972
2973        let all = |terms: Vec<Term>| Search {
2974            signal: Signal::Logs,
2975            from: 0,
2976            to: i64::MAX,
2977            terms,
2978            limit: 100,
2979            after: None,
2980        };
2981        let field = |name: &str, op, value| {
2982            all(vec![Term {
2983                target: Target::Field(name.into()),
2984                op,
2985                value,
2986            }])
2987        };
2988        let attr = |key: &str, op, value| {
2989            all(vec![Term {
2990                target: Target::Attr(key.into()),
2991                op,
2992                value,
2993            }])
2994        };
2995        let cases: Vec<(&str, Search)> = vec![
2996            ("no term (whole block)", all(Vec::new())),
2997            (
2998                "time range (per-row)",
2999                Search {
3000                    from: lo + (hi - lo) / 4,
3001                    to: hi,
3002                    ..all(Vec::new())
3003                },
3004            ),
3005            (
3006                "int field  severity_number gte",
3007                field("severity_number", Op::Gte, Value::Int(9)),
3008            ),
3009            (
3010                "utf8 field body contains",
3011                field("body", Op::Contains, Value::Str("tenant t7".into())),
3012            ),
3013            (
3014                "dict field severity_text eq",
3015                field("severity_text", Op::Eq, Value::Str("ERROR".into())),
3016            ),
3017            (
3018                "attr record http.route eq",
3019                attr("http.route", Op::Eq, Value::Str("/checkout".into())),
3020            ),
3021            (
3022                "attr record http.status_code gte",
3023                attr("http.status_code", Op::Gte, Value::Int(500)),
3024            ),
3025            (
3026                "attr resource service.name eq",
3027                attr("service.name", Op::Eq, Value::Str("checkout".into())),
3028            ),
3029        ];
3030
3031        let reps = if rows > 100_000 { 5 } else { 1 };
3032        for (name, s) in &cases {
3033            let hit = b.select(s, &src).len();
3034            let t = std::time::Instant::now();
3035            for _ in 0..reps {
3036                std::hint::black_box(b.select(s, &src));
3037            }
3038            let ns = t.elapsed().as_nanos() as f64 / (reps * n) as f64;
3039            println!("{name:34} {ns:8.3} ns/row  {hit} of {n}");
3040            assert!(hit > 0, "{name} matched nothing");
3041        }
3042
3043        // And the same block through the whole read path, published and warm,
3044        // so `select` can be read as a fraction of what a query actually costs.
3045        // Everything outside it — the mmap's minor faults, the CRC32 of every
3046        // table body, the dictionary scan, `Block::open`'s two child indexes,
3047        // the cursor filter and the JSON of `limit` rows — is the "no term"
3048        // line, and that is the claim in docs/architecture.md section 11 that
3049        // the per-block cost is paging rather than scanning.
3050        let dir = std::env::temp_dir().join(format!("mira-scan-cost-{}", std::process::id()));
3051        let _ = std::fs::remove_dir_all(&dir);
3052        crate::block::publish(&dir, "logs", 1, 1, 0, &open.sealed).unwrap();
3053        let one = (
3054            "no term, limit 1",
3055            Search {
3056                limit: 1,
3057                ..all(Vec::new())
3058            },
3059        );
3060        for (name, s) in [&cases[0], &one, &cases[3], &cases[5]] {
3061            let _ = search(&dir, s).unwrap();
3062            let t = std::time::Instant::now();
3063            for _ in 0..reps {
3064                std::hint::black_box(search(&dir, s).unwrap());
3065            }
3066            let ns = t.elapsed().as_nanos() as f64 / (reps * n) as f64;
3067            println!("  full search: {name:21} {ns:8.3} ns/row");
3068        }
3069        let _ = std::fs::remove_dir_all(&dir);
3070    }
3071}