crates/mimir-ingest/tests/ingest/diff.rs
//! The commit diff from the READ-FAST zone ([`EventLog::interval_diff`]):
//! what changed in a repo between two seqs, answered by range predicates on
//! the interval tables (+ the event tail past the anchor) — never two full
//! folds. THE fold at each end is the oracle: added / removed / changed must
//! be exactly the identity-set difference of the two folds, with `changed`
//! the identities open at both ends under another payload hash.
use std::collections::BTreeMap;
use std::path::Path;
use mimir_ingest::{Consumer, EventLog, IntervalDiff, Kind, RepoSpec};
use skade::arrow_array::StringArray;
use skade::arrow_schema::DataType;
use crate::common::{FakeDumper, Scratch, open_log};
const REPO: &str = "r";
/// Commits 1..=4: adds; a payload change + a new edge; a removed crate + a
/// signature change; a REVERT of the signature (a changed-then-restored
/// payload is NOT a change between 2 and 4).
fn commit_step(src: &Scratch, step: usize) {
match step {
1 => {
src.write("crates/a/src/lib.rs", "fn f (x: i32) | one\nfn g () | two\ncall g -> f\n")
.write("crates/b/src/lib.rs", "fn h () | three\ncall h -> h\n");
src.commit("one");
}
2 => {
src.write("crates/a/src/lib.rs", "fn f (x: i32) | uno\nfn g () | two\ncall g -> f\ncall f -> g\n");
src.commit("two");
}
3 => {
src.write("crates/a/src/lib.rs", "fn f (x: i64) | uno\nfn g () | two\ncall f -> g\n");
src.remove("crates/b");
src.commit("three");
}
4 => {
src.write("crates/a/src/lib.rs", "fn f (x: i32) | uno\nfn g () | two\ncall f -> g\n");
src.commit("four");
}
_ => unreachable!(),
}
}
fn spec(tmp: &Path, src: &Scratch) -> RepoSpec {
RepoSpec {
name: REPO.into(),
source: src.root.to_string_lossy().into_owned(),
branch: "main".into(),
worktree: tmp.join("clone"),
initial_offset: None,
}
}
fn key_columns(kind: Kind) -> &'static [&'static str] {
match kind {
Kind::Nodes => &["moniker"],
Kind::Edges => &["from", "to", "kind"],
Kind::Spans => &["target", "key", "file", "start_line", "end_line"],
_ => unreachable!("a source kind"),
}
}
/// `(krate, key…)` of one referenced row, for comparing.
fn row_key(d: &IntervalDiff, r: (u32, u32), keys: &[&str]) -> String {
let b = &d.batches[r.0 as usize];
let col = |name: &str| -> String {
let c = b.column_by_name(name).unwrap_or_else(|| panic!("column `{name}` missing"));
let c = skade::arrow_cast::cast(c, &DataType::Utf8).unwrap();
c.as_any().downcast_ref::<StringArray>().unwrap().value(r.1 as usize).to_string()
};
let mut out = col("krate");
for k in keys {
out.push('|');
out.push_str(&col(k));
}
out
}
fn rendered(d: &IntervalDiff, refs: &[(u32, u32)], keys: &[&str]) -> Vec<String> {
let mut v: Vec<String> = refs.iter().map(|&r| row_key(d, r, keys)).collect();
v.sort();
v
}
/// The oracle: the identity sets of THE fold at `from` and at `to`.
async fn fold_diff(log: &EventLog, kind: Kind, from: i64, to: i64) -> (Vec<String>, Vec<String>, Vec<String>) {
let mut sides = Vec::new();
for seq in [from, to] {
let set = log.fold(Some(REPO), Some(seq)).await.unwrap();
let mut rows: BTreeMap<String, mimir_ingest::PayloadHash> = BTreeMap::new();
if let Some(open) = set.repo(kind, REPO) {
for (krate, key, hash) in open.iter() {
rows.insert(format!("{krate}|{}", key.replace('\u{1f}', "|")), *hash);
}
}
sides.push(rows);
}
let (a, b) = (&sides[0], &sides[1]);
let added = b.keys().filter(|k| !a.contains_key(*k)).cloned().collect();
let removed = a.keys().filter(|k| !b.contains_key(*k)).cloned().collect();
let changed = b.iter().filter(|(k, h)| a.get(*k).is_some_and(|ha| &ha != h)).map(|(k, _)| k.clone()).collect();
(added, removed, changed)
}
async fn check_every_window(log: &EventLog, upto: i64) {
for kind in Kind::SOURCE {
let keys = key_columns(kind);
for from in 0..upto {
for to in (from + 1)..=upto {
let d = log.interval_diff(kind, REPO, from, to).await.unwrap();
let (added, removed, changed) = fold_diff(log, kind, from, to).await;
assert_eq!(rendered(&d, &d.added, keys), added, "{kind:?} added, {from} -> {to}");
assert_eq!(rendered(&d, &d.removed, keys), removed, "{kind:?} removed, {from} -> {to}");
assert_eq!(rendered(&d, &d.changed, keys), changed, "{kind:?} changed, {from} -> {to}");
}
}
}
}
/// Anchor at the tip: every (from, to) window of every source kind equals
/// the two-fold diff — the keys the fold renders are what the rows carry.
#[tokio::test(flavor = "current_thread")]
async fn the_interval_diff_is_the_fold_diff_at_every_window() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
for step in 1..=4 {
commit_step(&src, step);
}
let log = open_log(&tmp.path().join("lake")).await;
let mut c = Consumer::boot(log, FakeDumper::default()).await.unwrap();
c.run(&spec(tmp.path(), &src)).await.unwrap();
let (log, _) = c.into_parts();
log.materialize(REPO).await.unwrap();
check_every_window(&log, 4).await;
}
/// Anchor BEHIND the tip (and windows wholly past it): the tail completes
/// the diff, and the same equality holds for every window.
#[tokio::test(flavor = "current_thread")]
async fn the_tail_completes_the_diff_past_the_anchor() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
for step in 1..=2 {
commit_step(&src, step);
}
let log = open_log(&tmp.path().join("lake")).await;
let mut c = Consumer::boot(log, FakeDumper::default()).await.unwrap();
c.run(&spec(tmp.path(), &src)).await.unwrap();
let (log, _) = c.into_parts();
log.materialize(REPO).await.unwrap();
for step in 3..=4 {
commit_step(&src, step);
}
let mut c = Consumer::boot(log, FakeDumper::default()).await.unwrap();
c.run(&spec(tmp.path(), &src)).await.unwrap();
let (log, _) = c.into_parts();
assert_eq!(log.readfast_anchor(REPO).await.unwrap(), Some(2));
check_every_window(&log, 4).await;
}
/// No zone at all: the diff still answers (the fold is its own floor).
#[tokio::test(flavor = "current_thread")]
async fn a_repo_without_a_zone_still_diffs() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
for step in 1..=3 {
commit_step(&src, step);
}
let log = open_log(&tmp.path().join("lake")).await;
let mut c = Consumer::boot(log, FakeDumper::default()).await.unwrap();
c.run(&spec(tmp.path(), &src)).await.unwrap();
let (log, _) = c.into_parts();
check_every_window(&log, 3).await;
}
/// The window's own semantics: a payload changed and restored inside the
/// window is NOT a change (2 -> 4 restored `f`'s signature to its seq-2
/// shape, but its body stayed `uno`-changed from seq 2 — so f IS unchanged
/// between 2 and 4 only if every payload column matches; here it does).
#[tokio::test(flavor = "current_thread")]
async fn a_change_reverted_inside_the_window_is_no_change() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
for step in 1..=4 {
commit_step(&src, step);
}
let log = open_log(&tmp.path().join("lake")).await;
let mut c = Consumer::boot(log, FakeDumper::default()).await.unwrap();
c.run(&spec(tmp.path(), &src)).await.unwrap();
let (log, _) = c.into_parts();
log.materialize(REPO).await.unwrap();
let d = log.interval_diff(Kind::Nodes, REPO, 2, 4).await.unwrap();
let keys = key_columns(Kind::Nodes);
assert!(!rendered(&d, &d.changed, keys).iter().any(|k| k.ends_with("|a f")), "seq 4 restored f to its seq-2 payload: {:?}", rendered(&d, &d.changed, keys));
}