crates/mimir-room/tests/room/tables.rs
//! TOUCHES_TABLE in the room: a table is ONE node for the whole log, two
//! repos meet at it, the edge carries `access` (the union over the units that
//! describe it), and a table whose name nobody knows at compile time is a
//! node of its own — counted, never dropped.
//!
//! ```text
//! repo `store`, crate st@0.1.0 — units `lib` and `lib+test`
//! main → api → save CALLS
//! save TOUCHES bench_runs write src/lib.rs:12
//! load TOUCHES bench_runs read src/lib.rs:20
//! mixed TOUCHES bench_runs lib says read, lib+test says write → both
//! browse TOUCHES <unresolved> read ×2 src/lib.rs:40, 41
//! idle touches nothing
//! repo `web`, crate web@0.2.0
//! page → report CALLS
//! report TOUCHES bench_runs read src/lib.rs:7
//! ```
use std::collections::BTreeSet;
use std::path::Path;
use mimir_ingest::{CommitDump, Consumer, CrateDump, Dumper, EventLog};
use mimir_room::{AnswerState, EdgeSel, MaskSpec, Room, Scope};
use mimir_schema::{Access, EdgeKind, EdgeRow, Level, NodeRow, Origin, SPAN_EDGE, SpanRow, TABLE_UNRESOLVED, edge_key, kind, moniker};
use super::common::{open_log, repo_spec, scratch, serial};
const STORE: &str = "store";
const WEB: &str = "web";
fn st(d: &str) -> String {
moniker::moniker("st", "0.1.0", d)
}
fn web(d: &str) -> String {
moniker::moniker("web", "0.2.0", d)
}
fn runs() -> String {
moniker::table("bench_runs")
}
fn unresolved() -> String {
moniker::table(TABLE_UNRESOLVED)
}
fn func(id: &str, krate: &str, version: &str) -> NodeRow {
let name = id.rsplit(' ').next().unwrap().trim_end_matches("().").to_string();
NodeRow::new(id, kind::FN, Level::Item, name, krate, version, Origin::Local)
}
/// The row every dump writes for a table it touches.
fn table(name: &str) -> NodeRow {
NodeRow::new(moniker::table(name), kind::TABLE, Level::Item, name, moniker::TABLE_PACKAGE, moniker::TABLE_VERSION, Origin::Local)
}
fn touch(from: &str, to: &str, access: Access, count: u32) -> EdgeRow {
EdgeRow { access: Some(access), count, ..EdgeRow::new(from, to, EdgeKind::TouchesTable) }
}
fn site(from: &str, to: &str, line: u32) -> SpanRow {
SpanRow { target: SPAN_EDGE, key: edge_key(from, to, EdgeKind::TouchesTable), file: "src/lib.rs".into(), start_line: line, end_line: line }
}
struct Units(Vec<(&'static str, Vec<NodeRow>, Vec<EdgeRow>, Vec<SpanRow>)>);
impl Dumper for Units {
fn dump(&mut self, _: &Path, _: &str) -> anyhow::Result<CommitDump> {
let units = std::mem::take(&mut self.0);
let pkg = units[0].0.split('@').next().unwrap().to_string();
Ok(CommitDump {
toolchain: "nightly-fixture".into(),
workspace_crates: units.iter().map(|u| u.0.to_string()).collect(),
workspace_targets: [format!("{pkg}:lib")].into(),
failed_units: BTreeSet::new(),
owner_disambiguations: 0,
crates: units
.into_iter()
.map(|(owner, n, e, s)| CrateDump {
krate: owner.into(),
universe: false,
nodes: vec![mimir_schema::nodes_batch(&n).unwrap()],
edges: vec![mimir_schema::edges_batch(&e).unwrap()],
spans: vec![mimir_schema::spans_batch(&s).unwrap()],
})
.collect(),
binary: false,
binaries: vec![],
})
}
}
async fn consume(log: EventLog, tmp: &Path, name: &str, units: Units) -> EventLog {
scratch(&tmp.join(format!("src-{name}")), 1);
let mut c = Consumer::boot(log, units).await.unwrap();
let rep = c.run(&repo_spec(tmp, name)).await.unwrap();
assert!(rep.iter().all(|r| r.status == "ok"), "{rep:?}");
c.into_parts().0
}
async fn fixture(tmp: &Path) -> EventLog {
use EdgeKind::Calls;
let fns = || -> Vec<NodeRow> { ["main().", "api().", "save().", "load().", "mixed().", "browse().", "idle()."].iter().map(|d| func(&st(d), "st", "0.1.0")).chain([table("bench_runs"), table(TABLE_UNRESOLVED)]).collect() };
let edges = |mixed: Access| {
vec![
EdgeRow::new(st("main()."), st("api()."), Calls),
EdgeRow::new(st("api()."), st("save()."), Calls),
touch(&st("save()."), &runs(), Access::Write, 1),
touch(&st("load()."), &runs(), Access::Read, 1),
touch(&st("mixed()."), &runs(), mixed, 1),
touch(&st("browse()."), &unresolved(), Access::Read, 2),
]
};
let spans = || vec![site(&st("save()."), &runs(), 12), site(&st("load()."), &runs(), 20), site(&st("browse()."), &unresolved(), 40), site(&st("browse()."), &unresolved(), 41)];
let log = open_log(tmp).await;
let log = consume(log, tmp, STORE, Units(vec![("st@0.1.0:lib", fns(), edges(Access::Read), spans()), ("st@0.1.0:lib+test", fns(), edges(Access::Write), spans())])).await;
let web_nodes = vec![func(&web("page()."), "web", "0.2.0"), func(&web("report()."), "web", "0.2.0"), table("bench_runs")];
let web_edges = vec![EdgeRow::new(web("page()."), web("report()."), Calls), touch(&web("report()."), &runs(), Access::Read, 1)];
consume(log, tmp, WEB, Units(vec![("web@0.2.0:lib", web_nodes, web_edges, vec![site(&web("report()."), &runs(), 7)])])).await
}
fn names(r: &Room, ords: &[u64]) -> Vec<String> {
let mut v: Vec<String> = ords.iter().map(|o| r.node(*o).name.unwrap_or("?").to_string()).collect();
v.sort();
v
}
#[tokio::test(flavor = "current_thread")]
async fn two_repos_meet_at_one_table_node_and_the_edge_says_how() {
let _g = serial().await;
let tmp = tempfile::tempdir().unwrap();
let log = fixture(tmp.path()).await;
let r = Room::at(&log, &Scope::current()).await.unwrap();
// this lake was created with the `access` column: every edge row has it
assert!(!r.edge_column_added_in_place("access"));
// ---- ONE node per table name, whoever wrote it
let tables = r.mask(&MaskSpec::kinds(&[kind::TABLE])).unwrap();
assert_eq!(names(&r, tables.value.ordinals()), ["<unresolved>", "bench_runs"]);
let t = r.ordinal(&runs()).expect("the table node");
let n = r.node(t);
assert_eq!((n.kind, n.name, n.krate, n.defined), (Some("table"), Some("bench_runs"), Some("@warehouse"), true));
// three units of two repos wrote the same row: not a repeated key
let dups = r.duplicate_keys(None).unwrap();
assert!(dups.value.is_empty(), "{:?}", dups.value);
// ---- the edges, with access and site count
let e = r.edges(&EdgeSel::kinds(&[EdgeKind::TouchesTable])).unwrap();
assert_eq!(e.evidence.state, AnswerState::Complete, "{}", e.evidence);
let mut got: Vec<(String, String, Option<Access>, u32)> = e.value.edges.iter().map(|x| (r.node(x.from).name.unwrap().to_string(), r.node(x.to).name.unwrap().to_string(), x.access, x.count)).collect();
got.sort_by(|a, b| (&a.0, &a.1).cmp(&(&b.0, &b.1)));
assert_eq!(
got,
vec![
("browse".into(), "<unresolved>".into(), Some(Access::Read), 2),
("load".into(), "bench_runs".into(), Some(Access::Read), 1),
// lib says read, lib+test says write: the union
("mixed".into(), "bench_runs".into(), Some(Access::Both), 1),
("report".into(), "bench_runs".into(), Some(Access::Read), 1),
("save".into(), "bench_runs".into(), Some(Access::Write), 1),
]
);
assert_eq!(e.value.sites, 6);
let save = r.ordinal(&st("save().")).unwrap();
assert_eq!(r.access_of(EdgeKind::TouchesTable, save, t), Some(Access::Write));
// an edge of another kind has no access
let api = r.ordinal(&st("api().")).unwrap();
assert_eq!(r.access_of(EdgeKind::Calls, api, save), None);
let calls = r.edges(&EdgeSel::kinds(&[EdgeKind::Calls])).unwrap();
assert!(calls.value.edges.iter().all(|x| x.access.is_none()));
// ---- where: the sites of an edge
let sites: Vec<(String, u32)> = r.sites(save, t, EdgeKind::TouchesTable).iter().map(|s| (format!("{}:{}", s.repo, s.file), s.start_line)).collect();
assert_eq!(sites, vec![("store:src/lib.rs".to_string(), 12)]);
let browse = r.ordinal(&st("browse().")).unwrap();
let u = r.ordinal(&unresolved()).unwrap();
assert_eq!(r.sites(browse, u, EdgeKind::TouchesTable).iter().map(|s| s.start_line).collect::<Vec<_>>(), [40, 41]);
// ---- who reaches the table: the touchers, and their callers, in BOTH repos
let reach = r.dependents_by(&[runs().as_str()], &[EdgeKind::TouchesTable, EdgeKind::Calls], None, None).unwrap();
assert_eq!(names(&r, &reach.value.reached), ["api", "load", "main", "mixed", "page", "report", "save"]);
// one hop: the direct touchers only
let direct = r.dependents_by(&[runs().as_str()], &[EdgeKind::TouchesTable], Some(1), None).unwrap();
assert_eq!(names(&r, &direct.value.reached), ["load", "mixed", "report", "save"]);
// from the code side: what does `main` reach
let from_main = r.reach(&[st("main().").as_str()], &[EdgeKind::Calls, EdgeKind::TouchesTable], None, None).unwrap();
assert_eq!(names(&r, &from_main.value.reached), ["api", "bench_runs", "save"]);
let its_tables: Vec<u64> = from_main.value.reached.iter().copied().filter(|o| tables.value.contains(*o)).collect();
assert_eq!(names(&r, &its_tables), ["bench_runs"]);
// how far: main → api → save → the table
let near = r.min_hops_to(&MaskSpec::kinds(&[kind::TABLE]), &[EdgeKind::Calls, EdgeKind::TouchesTable], None).unwrap();
let hops = |d: &str| near.value.iter().find(|(o, _)| Some(*o) == r.ordinal(&st(d))).map(|x| x.1);
assert_eq!((hops("main()."), hops("api()."), hops("save()."), hops("idle().")), (Some(3), Some(2), Some(1), None));
// a fn that touches nothing reaches no table — a fact, with every input there
let idle = r.reach(&[st("idle().").as_str()], &[EdgeKind::Calls, EdgeKind::TouchesTable], None, None).unwrap();
assert!(idle.value.reached.is_empty());
assert_eq!(idle.evidence.state, AnswerState::EmptyFact, "{}", idle.evidence);
// ---- one repo's scope: the other repo's code is not in it, the table is
let only = Room::at(&log, &Scope::repo(STORE, None)).await.unwrap();
assert!(only.ordinal(&web("report().")).is_none());
let reach = only.dependents_by(&[runs().as_str()], &[EdgeKind::TouchesTable, EdgeKind::Calls], None, None).unwrap();
assert_eq!(names(&only, &reach.value.reached), ["api", "load", "main", "mixed", "save"]);
let web_only = Room::at(&log, &Scope::repo(WEB, None)).await.unwrap();
let reach = web_only.dependents_by(&[runs().as_str()], &[EdgeKind::TouchesTable, EdgeKind::Calls], None, None).unwrap();
assert_eq!(names(&web_only, &reach.value.reached), ["page", "report"]);
// … and the unresolved node is not in a scope whose code has no such site
assert!(web_only.ordinal(&unresolved()).is_none());
}
/// A lake whose edge table lacked `access` gains it in place: the room says
/// so, because there "no TOUCHES_TABLE edge" is not yet a fact about a repo
/// that was not consumed since.
#[tokio::test(flavor = "current_thread")]
async fn a_lake_that_gained_the_access_column_in_place_says_so() {
use mimir_ingest::{PayloadSpec, Specs};
use skade::arrow_schema::{Field, Schema};
use std::sync::Arc;
let _g = serial().await;
let tmp = tempfile::tempdir().unwrap();
// the edge payload without `access`, stamped at the identity epoch (a
// lake stamped before 5 is wiped at open, not evolved)
let full = mimir_schema::edges_schema();
let kept: Vec<Field> = full.fields().iter().filter(|f| f.name() != "access").map(|f| f.as_ref().clone()).collect();
let mut meta = full.metadata().clone();
meta.insert(mimir_schema::META_SCHEMA.to_string(), "5".to_string());
let old_edges = Arc::new(Schema::new(kept).with_metadata(meta));
let mut specs = Specs::mimir().unwrap();
specs.edges = PayloadSpec::with_log_owned(old_edges.clone(), &["from", "to", "kind"], &["valid_from_sha", "valid_to_sha"]).unwrap();
specs.bin_edges = PayloadSpec::with_log_owned(old_edges, &["from", "to", "kind"], &["provenance", "valid_from_sha", "valid_to_sha"]).unwrap();
let lake = tmp.path().join("lake");
let log = EventLog::open(skade::open(&lake).await.unwrap(), specs).await.unwrap();
// one commit of a producer that knows no table: two fns and a call
struct Old;
impl Dumper for Old {
fn dump(&mut self, _: &Path, _: &str) -> anyhow::Result<CommitDump> {
let edges = mimir_schema::edges_batch(&[EdgeRow::new(st("main()."), st("api()."), EdgeKind::Calls)]).unwrap();
let keep: Vec<usize> = (0..edges.num_columns()).filter(|i| edges.schema().field(*i).name() != "access").collect();
Ok(CommitDump {
toolchain: "nightly-fixture".into(),
workspace_crates: ["st@0.1.0:lib".to_string()].into(),
workspace_targets: ["st:lib".to_string()].into(),
failed_units: BTreeSet::new(),
owner_disambiguations: 0,
crates: vec![CrateDump {
krate: "st@0.1.0:lib".into(),
universe: false,
nodes: vec![mimir_schema::nodes_batch(&[func(&st("main()."), "st", "0.1.0"), func(&st("api()."), "st", "0.1.0")]).unwrap()],
edges: vec![edges.project(&keep).unwrap()],
spans: vec![],
}],
binary: false,
binaries: vec![],
})
}
}
scratch(&tmp.path().join(format!("src-{STORE}")), 1);
let mut c = Consumer::boot(log, Old).await.unwrap();
assert!(c.run(&repo_spec(tmp.path(), STORE)).await.unwrap().iter().all(|r| r.status == "ok"));
drop(c);
// opened by this build: evolved in place, and the room knows
let log = EventLog::open_lake(&lake).await.unwrap();
let r = Room::at(&log, &Scope::current()).await.unwrap();
assert!(r.edge_column_added_in_place("access"));
assert!(!r.edge_column_added_in_place("count"));
let e = r.edges(&EdgeSel::kinds(&[EdgeKind::TouchesTable])).unwrap();
assert_eq!(e.value.total, 0);
assert_eq!(e.evidence.state, AnswerState::EmptyGap, "{}", e.evidence);
assert!(e.evidence.missing_input("TOUCHES_TABLE edges")[0].why.contains("mimir-schema 4"), "{}", e.evidence);
// the old call is there, with no access
let calls = r.edges(&EdgeSel::kinds(&[EdgeKind::Calls])).unwrap();
assert_eq!((calls.value.total, calls.value.edges[0].access), (1, None));
}