src/mimir_lens/rows.rs
//! **The warehouse rows the Mimir lenses consume** — the stringified read shape
//! plus the typed POD rows parsed out of it.
//!
//! LIFTED VERBATIM from `dwarves/crates/dwarves/src/coverage/rows.rs` (the QUERY
//! half of the lenses belongs in nornir — `dwarves::mimir` ARCHITECTURE L5a). The
//! `Table` shape is byte-identical to [`crate::warehouse::iceberg::TablePreview`]
//! (columns + rows, all cells rendered to display strings), so a warehouse scan
//! feeds this parser with no field mapping. Kept a faithful copy rather than a
//! rewrite so the two front doors (dwarves' GATE, nornir's RPC) cannot drift.
use std::collections::BTreeMap;
/// One table scan: stringified `columns` + `rows` (one cell per column).
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Table {
pub columns: Vec<String>,
pub rows: Vec<Vec<String>>,
}
impl Table {
pub fn new(columns: Vec<String>, rows: Vec<Vec<String>>) -> Self {
Self { columns, rows }
}
/// Is this table empty — i.e. did the warehouse carry NO rows for it?
pub fn is_empty(&self) -> bool {
self.rows.is_empty()
}
/// Column index by name.
pub fn col(&self, name: &str) -> Option<usize> {
self.columns.iter().position(|c| c == name)
}
/// A cursor that resolves the named columns ONCE and then reads rows by
/// position — so a 500k-row `call_edges` scan does not do a string compare per
/// cell per row.
pub fn cursor(&self, want: &[&str]) -> Cursor<'_> {
Cursor {
table: self,
idx: want.iter().map(|w| self.col(w)).collect(),
}
}
}
/// Resolved column positions over a [`Table`] (see [`Table::cursor`]).
pub struct Cursor<'a> {
table: &'a Table,
idx: Vec<Option<usize>>,
}
impl<'a> Cursor<'a> {
/// Read the `i`-th requested column of `row`; `""` when the column is absent
/// or the cell is null.
pub fn get(&self, row: &'a [String], i: usize) -> &'a str {
match self.idx.get(i).copied().flatten() {
Some(c) => row.get(c).map(String::as_str).unwrap_or(""),
None => "",
}
}
pub fn rows(&self) -> &'a [Vec<String>] {
&self.table.rows
}
/// Did EVERY requested column resolve?
pub fn complete(&self) -> bool {
self.idx.iter().all(Option::is_some)
}
}
// ─── typed rows ─────────────────────────────────────────────────────────────
/// One `symbol_facts` row, trimmed to the columns the lenses need.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SymbolFact {
pub snapshot_id: String,
pub ts_micros: i64,
pub repo: String,
pub crate_name: String,
pub module_path: String,
pub item_kind: String,
pub item_name: String,
pub visibility: String,
pub file: String,
pub line: u32,
}
impl SymbolFact {
/// The reachability identity: `<module_path>::<item_name>`.
pub fn fqn(&self) -> String {
format!("{}::{}", self.module_path, self.item_name)
}
}
/// One `bench_runs` row — one bench invocation, host-stamped.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct BenchRunFact {
pub run_id: String,
pub repo: String,
pub ts_micros: i64,
pub date: String,
pub version: String,
pub machine: String,
pub cores: u32,
/// The checkout's HEAD the run measured (`""` = a door that did not say).
pub commit: String,
}
impl BenchRunFact {
/// The code version the run measured: its commit when known, else its stamp.
pub fn code_version(&self) -> &str {
if self.commit.is_empty() { &self.version } else { &self.commit }
}
}
/// One `bench_results` row — a metric of one arm of one run.
#[derive(Debug, Clone, PartialEq)]
pub struct BenchResultFact {
pub run_id: String,
pub result_name: String,
pub metric_name: String,
pub metric_value: f64,
}
/// One `test_outcomes` row — one case in one run.
///
/// Two producers write the table. A bench run's own tests carry only
/// `run_id` / `test_name` / `passed` and are placed by their `bench_runs` row. A
/// Runa test step's cases (`Telemetry.SubmitStepFeed`) have no `bench_runs` row
/// and say it all themselves: `repo` (`source_repo`), `krate`
/// (`source_crate`), `status` and `commit`.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct TestOutcomeFact {
pub run_id: String,
pub test_name: String,
pub passed: bool,
/// `source_repo` — `""` on a bench-coupled row.
pub repo: String,
/// `source_crate`.
pub krate: String,
/// `pass` | `fail` | `ignored`; `""` on a bench-coupled row (`passed` says it).
pub status: String,
/// The checkout's HEAD the case ran against; `""` on a bench-coupled row.
pub commit: String,
}
impl TestOutcomeFact {
/// Did the case RUN? An ignored case says nothing about the code, and
/// neither does one an `only: Affected` step skipped (nothing it reaches
/// changed: it was not run, it is not missing).
pub fn ran(&self) -> bool {
!matches!(self.status.as_str(), "ignored" | "skipped")
}
}
/// One `git_heat_facts` row — per-file churn at scan time.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct GitHeatFact {
pub repo: String,
pub file: String,
pub commits_90d: u64,
}
// ─── parsing ────────────────────────────────────────────────────────────────
fn num<T: std::str::FromStr + Default>(s: &str) -> T {
s.trim().parse::<T>().unwrap_or_default()
}
fn boolean(s: &str) -> bool {
matches!(s.trim(), "true" | "True" | "1" | "TRUE")
}
/// Timestamps arrive either as µs since epoch or as an RFC3339 string. Both sort
/// correctly once folded to µs; an unparseable stamp yields `0` (sorts OLDEST).
///
/// RFC3339 is parsed EXACTLY (`chrono`), not by the digit heuristic below. The
/// heuristic keeps the first 14 digits — `2026-07-31T14:16:27.711509Z` folds to
/// `20260731141627`, i.e. SECOND resolution — so two snapshots of the same repo
/// written in the same second compared EQUAL and the caller's tie-break decided
/// which one was "latest". [`latest_snapshot_rows_of`]'s tie-break is the snapshot
/// UUID, which is random, so the fold silently chose a stale generation. The
/// heuristic remains as a last resort for shapes chrono cannot read.
pub fn ts_micros(s: &str) -> i64 {
let s = s.trim();
if let Ok(v) = s.parse::<i64>() {
return v;
}
if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(s) {
return dt.timestamp_micros();
}
let digits: String = s.chars().filter(char::is_ascii_digit).take(14).collect();
digits.parse::<i64>().unwrap_or(0)
}
pub fn parse_bench_runs(t: &Table) -> Vec<BenchRunFact> {
let c = t.cursor(&[
"run_id",
"repo",
"ts_micros",
"date",
"version",
"machine",
"cores",
"commit",
]);
c.rows()
.iter()
.map(|r| BenchRunFact {
run_id: c.get(r, 0).to_string(),
repo: c.get(r, 1).to_string(),
ts_micros: ts_micros(c.get(r, 2)),
date: c.get(r, 3).to_string(),
version: c.get(r, 4).to_string(),
machine: c.get(r, 5).to_string(),
cores: num(c.get(r, 6)),
commit: cell(c.get(r, 7)),
})
.collect()
}
pub fn parse_bench_results(t: &Table) -> Vec<BenchResultFact> {
let c = t.cursor(&["run_id", "result_name", "metric_name", "metric_value"]);
c.rows()
.iter()
.map(|r| BenchResultFact {
run_id: c.get(r, 0).to_string(),
result_name: c.get(r, 1).to_string(),
metric_name: c.get(r, 2).to_string(),
metric_value: c.get(r, 3).trim().parse().unwrap_or(0.0),
})
.collect()
}
pub fn parse_test_outcomes(t: &Table) -> Vec<TestOutcomeFact> {
let c = t.cursor(&["run_id", "test_name", "passed", "source_repo", "source_crate", "status", "commit"]);
c.rows()
.iter()
.map(|r| TestOutcomeFact {
run_id: c.get(r, 0).to_string(),
test_name: c.get(r, 1).to_string(),
passed: boolean(c.get(r, 2)),
repo: cell(c.get(r, 3)),
krate: cell(c.get(r, 4)),
status: cell(c.get(r, 5)),
commit: cell(c.get(r, 6)),
})
.collect()
}
/// One `coverage_fn` row, trimmed to what a lens joins: where the function is
/// and whether a test executed it.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct CoverageFnFact {
pub run_id: String,
pub ts_micros: i64,
pub repo: String,
pub file: String,
pub name: String,
/// First / last source line (`0` = the row says none).
pub line: u32,
pub end_line: u32,
pub regions: i64,
pub regions_covered: i64,
pub lines_covered: i64,
pub commit: String,
/// An `only: Affected` run, merged: the commit whose run measured these
/// counts (`""` = this run did).
pub carried_from: String,
/// An `only: Affected` run: the function changed since the baseline
/// (`None` = a run of every test, which does not say).
pub changed: Option<bool>,
}
impl CoverageFnFact {
/// Did any test execute the function at all?
pub fn exercised(&self) -> bool {
self.regions_covered > 0 || self.lines_covered > 0
}
/// It changed since the baseline and no selected test executed it.
pub fn changed_uncovered(&self) -> bool {
self.changed == Some(true) && !self.exercised()
}
}
pub fn parse_coverage_fn(t: &Table) -> Vec<CoverageFnFact> {
let c = t.cursor(&["run_id", "ts_micros", "repo", "file", "name", "line", "end_line", "regions", "regions_covered", "lines_covered", "commit", "carried_from", "changed"]);
c.rows()
.iter()
.map(|r| CoverageFnFact {
run_id: c.get(r, 0).to_string(),
ts_micros: ts_micros(c.get(r, 1)),
repo: c.get(r, 2).to_string(),
file: c.get(r, 3).to_string(),
name: c.get(r, 4).to_string(),
line: num(c.get(r, 5)),
end_line: num(c.get(r, 6)),
regions: num(c.get(r, 7)),
regions_covered: num(c.get(r, 8)),
lines_covered: num(c.get(r, 9)),
commit: cell(c.get(r, 10)),
carried_from: cell(c.get(r, 11)),
changed: match cell(c.get(r, 12)).as_str() {
"true" => Some(true),
"false" => Some(false),
_ => None,
},
})
.collect()
}
/// Each repo's NEWEST coverage run (by `ts_micros`, then `run_id`): a function
/// measured in an older run says what the code WAS.
pub fn latest_coverage_runs(rows: Vec<CoverageFnFact>) -> Vec<CoverageFnFact> {
let mut newest: BTreeMap<String, (i64, String)> = BTreeMap::new();
for r in &rows {
let e = newest.entry(r.repo.clone()).or_insert((i64::MIN, String::new()));
if (r.ts_micros, &r.run_id) > (e.0, &e.1) {
*e = (r.ts_micros, r.run_id.clone());
}
}
rows.into_iter().filter(|r| newest.get(&r.repo).is_some_and(|(_, id)| id == &r.run_id)).collect()
}
/// A text cell: the warehouse renders a NULL as `""` or as `null`/`NULL` — all
/// three mean "not said".
fn cell(s: &str) -> String {
match s.trim() {
"null" | "NULL" | "None" => String::new(),
v => v.to_string(),
}
}
pub fn parse_git_heat(t: &Table) -> Vec<GitHeatFact> {
let c = t.cursor(&["repo", "file", "commits_90d"]);
c.rows()
.iter()
.map(|r| GitHeatFact {
repo: c.get(r, 0).to_string(),
file: c.get(r, 1).to_string(),
commits_90d: num(c.get(r, 2)),
})
.collect()
}
// ─── latest-snapshot discipline ─────────────────────────────────────────────
/// **Retain only the latest-snapshot rows of a knowledge-scan table**, keyed by
/// its `repo` + `snapshot_id` columns (with `ts_micros` deciding "latest"). A
/// table without those columns (e.g. `symbol_sizes`, keyed by `content_sha256`)
/// is returned unchanged — the append-only double-count only afflicts the
/// `nornir knowledge scan` tables.
pub fn latest_snapshot_rows(t: &Table) -> Table {
latest_snapshot_rows_of(&t.columns, &t.rows)
}
/// The core of [`latest_snapshot_rows`] over **borrowed** `(columns, rows)`, so a
/// caller holding an `Arc<TablePreview>` (the Tier-0 working-set cache) can filter
/// to the latest snapshot per repo while cloning ONLY the surviving rows — instead
/// of first deep-cloning the whole preview into a `Table` and then cloning the
/// survivors again. That removes the read path's second full copy of the biggest
/// tables (`symbol_facts`, `call_edges`) on every lens/dup RPC (finding #2). The
/// preview shape is byte-identical to [`Table`], so this is a drop-in.
pub fn latest_snapshot_rows_of(columns: &[String], rows: &[Vec<String>]) -> Table {
let col = |name: &str| columns.iter().position(|c| c == name);
let (Some(ri), Some(si)) = (col("repo"), col("snapshot_id")) else {
// Not a snapshot-keyed table — hand back an owned copy unchanged.
return Table::new(columns.to_vec(), rows.to_vec());
};
let ti = col("ts_micros");
// latest snapshot_id per repo
let mut best: BTreeMap<&str, (i64, &str)> = BTreeMap::new();
for r in rows {
let repo = r.get(ri).map(String::as_str).unwrap_or("");
let snap = r.get(si).map(String::as_str).unwrap_or("");
let ts = ti.and_then(|i| r.get(i)).map(|s| ts_micros(s)).unwrap_or(0);
let e = best.entry(repo).or_insert((i64::MIN, ""));
if (ts, snap) > (e.0, e.1) {
*e = (ts, snap);
}
}
let kept = rows
.iter()
.filter(|r| {
let repo = r.get(ri).map(String::as_str).unwrap_or("");
let snap = r.get(si).map(String::as_str).unwrap_or("");
best.get(repo).map(|(_, s)| *s == snap).unwrap_or(false)
})
.cloned()
.collect();
Table::new(columns.to_vec(), kept)
}
#[cfg(test)]
mod latest_snapshot_ordering_tests {
use super::*;
/// The REAL defect, reproduced from the live oden warehouse (2026-07-31).
///
/// `skade` had three `scip_occurrences` snapshots: an OLD 12:54 one written
/// before the document-scoping fix, and two clean ones at 13:39 and 14:16. The
/// fold is supposed to keep the 14:16 rows. It kept the 12:54 rows — because
/// the preview rendered every `ts_micros` cell as the literal type name
/// `<Timestamp(...)>`, [`ts_micros`] folded all three to `0`, and the
/// `(ts, snapshot_id)` ordering fell through to the UUID tie-break, where
/// `c60fdc74…` (the OLDEST) sorts above `b732c62e…` (the NEWEST).
///
/// So `Mimir.SymbolIdentity` reported 4 repos as still carrying pre-fix rows
/// AFTER they had all been re-ingested clean: a stale answer that looked
/// exactly like a real finding, with no error raised anywhere.
#[test]
fn the_newest_snapshot_wins_even_when_the_oldest_has_the_larger_uuid() {
let cols: Vec<String> = ["repo", "snapshot_id", "ts_micros", "symbol"]
.iter()
.map(|s| s.to_string())
.collect();
let row = |snap: &str, ts: &str, sym: &str| {
vec![
"skade".to_string(),
snap.to_string(),
ts.to_string(),
sym.to_string(),
]
};
let rows = vec![
// OLDEST, but the lexicographically LARGEST uuid — the trap.
row(
"c60fdc74-f5fa-4555-9567-09e1fc191426",
"2026-07-31T12:54:39.610047Z",
"local 0",
),
row(
"b237882a-f96d-4f14-a95c-6251cc214396",
"2026-07-31T13:39:12.348006Z",
"local a.rs#0",
),
// NEWEST, smaller uuid.
row(
"b732c62e-a235-47ab-9b39-ec7b22c83154",
"2026-07-31T14:16:27.711509Z",
"local b.rs#0",
),
];
let out = latest_snapshot_rows_of(&cols, &rows);
assert_eq!(
out.rows.len(),
1,
"exactly one snapshot survives: {:?}",
out.rows
);
assert_eq!(
out.rows[0][1], "b732c62e-a235-47ab-9b39-ec7b22c83154",
"the NEWEST snapshot must win, not the largest uuid"
);
}
/// Sub-second resolution is load-bearing: the resolved-ingest fan-out writes
/// several members' snapshots inside one second, and a repo re-resolved twice
/// in a cycle would otherwise tie and fall back to the random uuid again.
#[test]
fn two_snapshots_in_the_same_second_still_order_by_their_micros() {
assert!(
ts_micros("2026-07-31T14:16:27.711509Z") > ts_micros("2026-07-31T14:16:27.260818Z"),
"sub-second digits must not be truncated away"
);
let cols: Vec<String> = ["repo", "snapshot_id", "ts_micros"]
.iter()
.map(|s| s.to_string())
.collect();
let rows = vec![
vec![
"r".to_string(),
"zzz".to_string(),
"2026-07-31T14:16:27.260818Z".to_string(),
],
vec![
"r".to_string(),
"aaa".to_string(),
"2026-07-31T14:16:27.711509Z".to_string(),
],
];
let out = latest_snapshot_rows_of(&cols, &rows);
assert_eq!(out.rows.len(), 1);
assert_eq!(
out.rows[0][1], "aaa",
"later micros wins over the larger uuid"
);
}
/// The two wire shapes both fold to real epoch µs, so they compare correctly,
/// and an unreadable stamp still sorts OLDEST rather than newest — a stamp we
/// cannot read must never be allowed to claim it is the freshest.
#[test]
fn ts_micros_reads_rfc3339_exactly_and_epoch_micros_unchanged() {
assert_eq!(ts_micros("1785507387711509"), 1785507387711509);
assert_eq!(
ts_micros("2026-07-31T14:16:27.711509Z"),
chrono::DateTime::parse_from_rfc3339("2026-07-31T14:16:27.711509Z")
.unwrap()
.timestamp_micros()
);
// The exact string the un-fixed `cell_to_string` produced.
assert_eq!(ts_micros("<Timestamp(Microsecond, Some(\"+00:00\"))>"), 0);
assert_eq!(ts_micros(""), 0);
}
}