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}