crates/mimir-ingest/tests/ingest/readfast.rs
//! The READ-FAST zone: per repo, one interval table per source kind — one
//! row per node/edge/span VERSION with `[valid_from_seq, valid_to_seq)`,
//! rebuilt from the event log ([`EventLog::materialize`], the anchor) — and
//! the read ([`EventLog::readfast_rows`]): the interval rows live at `seq`
//! plus the event tail after the anchor. The fold over the full log is the
//! oracle: at every seq the two must hand out the same open rows.
use std::path::Path;
use mimir_ingest::{Consumer, EventLog, FoldScope, FoldedRows, Kind, RepoSpec};
use skade::arrow_array::{Array, StringArray};
use skade::arrow_schema::DataType;
use crate::common::{FakeDumper, Scratch, open_log};
const REPO: &str = "r";
/// The history, one step at a time: adds, a payload change + a new edge, a
/// removed crate and a changed signature — the compaction fixture's history.
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");
}
_ => 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,
}
}
/// The fixture specs' key columns ([`crate::common::specs`]).
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"),
}
}
/// The open rows of `folded` as sorted `krate|key…|hash` strings — the
/// canonical form both reads are compared in.
fn open_rows(folded: &FoldedRows, keys: &[&str]) -> Vec<String> {
let mut out = Vec::new();
for &(bi, ri) in &folded.open {
let b = &folded.batches[bi as usize];
let col = |name: &str| -> String {
let c = b
.column_by_name(name)
.unwrap_or_else(|| panic!("column `{name}` missing from a batch with {:?}", b.schema().fields().iter().map(|f| f.name().clone()).collect::<Vec<_>>()));
let c = skade::arrow_cast::cast(c, &DataType::Utf8).unwrap();
let c = c.as_any().downcast_ref::<StringArray>().unwrap();
if c.is_null(ri as usize) { "\u{2205}".to_string() } else { c.value(ri as usize).to_string() }
};
let mut row = col("krate");
for k in keys {
row.push('|');
row.push_str(&col(k));
}
row.push('|');
row.push_str(&col("payload_hash"));
out.push(row);
}
out.sort();
out
}
async fn fold_oracle(log: &EventLog, kind: Kind, upto: i64) -> Vec<String> {
let scope = FoldScope { repos: [(REPO.to_string(), upto)].into(), universe: false };
let folded = log.fold_rows(kind, &scope, &["payload_hash"]).await.unwrap();
open_rows(&folded, key_columns(kind))
}
/// Commit steps `1..=upto` and consume them.
async fn consumed(tmp: &Path, src: &Scratch, upto: usize) -> EventLog {
for step in 1..=upto {
commit_step(src, step);
}
let log = open_log(&tmp.join("lake")).await;
let mut c = Consumer::boot(log, FakeDumper::default()).await.unwrap();
c.run(&spec(tmp, src)).await.unwrap();
let (log, _) = c.into_parts();
log
}
/// Anchor at the tip: the interval rows alone are the fold, at the tip and
/// as-of every earlier seq (time travel without the event scan).
#[tokio::test(flavor = "current_thread")]
async fn the_interval_rows_are_the_fold_at_every_seq() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
let log = consumed(tmp.path(), &src, 3).await;
let report = log.materialize(REPO).await.unwrap();
assert_eq!(report.anchor_seq, 3, "{report:?}");
assert!(report.rows > 0 && report.closed_rows > 0, "{report:?}");
assert_eq!(log.readfast_anchor(REPO).await.unwrap(), Some(3));
for kind in Kind::SOURCE {
for seq in 1..=3i64 {
let fast = log.readfast_rows(kind, REPO, Some(seq), &["payload_hash"]).await.unwrap();
assert_eq!(
open_rows(&fast, key_columns(kind)),
fold_oracle(&log, kind, seq).await,
"{kind:?} as of seq {seq}"
);
}
}
}
/// Anchor BEHIND the tip: the read is interval rows + the event tail, and it
/// still equals the fold at every seq — including seqs before the anchor.
#[tokio::test(flavor = "current_thread")]
async fn the_tail_after_the_anchor_completes_the_read() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
// consume two commits, anchor there, then the third arrives
let log = consumed(tmp.path(), &src, 2).await;
let report = log.materialize(REPO).await.unwrap();
assert_eq!(report.anchor_seq, 2, "{report:?}");
commit_step(&src, 3);
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), "a consume moves no anchor");
for kind in Kind::SOURCE {
for seq in 1..=3i64 {
let fast = log.readfast_rows(kind, REPO, Some(seq), &["payload_hash"]).await.unwrap();
assert_eq!(
open_rows(&fast, key_columns(kind)),
fold_oracle(&log, kind, seq).await,
"{kind:?} as of seq {seq} (anchor 2)"
);
}
// no upto = the tip
let fast = log.readfast_rows(kind, REPO, None, &["payload_hash"]).await.unwrap();
assert_eq!(open_rows(&fast, key_columns(kind)), fold_oracle(&log, kind, 3).await, "{kind:?} at the tip");
}
}
/// A closed version keeps its interval: `h` (crate b, removed at seq 3) is a
/// row with `valid_to_seq = 3`; `f`'s changed payload is two versions,
/// `[1, 2)` and `[2, 3)` and `[3, —)` for the schema-3 signature change.
#[tokio::test(flavor = "current_thread")]
async fn a_closed_version_keeps_its_interval() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
let log = consumed(tmp.path(), &src, 3).await;
log.materialize(REPO).await.unwrap();
let rows = log.interval_rows(Kind::Nodes, REPO).await.unwrap();
let mut h = Vec::new();
let mut f = Vec::new();
for b in &rows {
let id = skade::arrow_cast::cast(b.column_by_name("moniker").unwrap(), &DataType::Utf8).unwrap();
let id = id.as_any().downcast_ref::<StringArray>().unwrap();
let from = b.column_by_name("valid_from_seq").unwrap();
let from = from.as_any().downcast_ref::<skade::arrow_array::Int64Array>().unwrap();
let to = b.column_by_name("valid_to_seq").unwrap();
let to = to.as_any().downcast_ref::<skade::arrow_array::Int64Array>().unwrap();
for i in 0..b.num_rows() {
let interval = (from.value(i), (!to.is_null(i)).then(|| to.value(i)));
match id.value(i) {
"b h" => h.push(interval),
"a f" => f.push(interval),
_ => {}
}
}
}
h.sort();
f.sort();
assert_eq!(h, vec![(1, Some(3))], "crate b's `h` lived [1, 3)");
assert_eq!(f, vec![(1, Some(2)), (2, Some(3)), (3, None)], "`f` changed at 2 (doc) and 3 (signature)");
}
/// A second materialize at the same tip changes nothing; after new events it
/// moves the anchor and the interval read follows.
#[tokio::test(flavor = "current_thread")]
async fn materialize_is_idempotent_and_moves_with_the_log() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
let log = consumed(tmp.path(), &src, 2).await;
let first = log.materialize(REPO).await.unwrap();
assert_eq!(first.anchor_seq, 2);
let again = log.materialize(REPO).await.unwrap();
assert!(again.skipped, "the anchor is at the tip already: {again:?}");
commit_step(&src, 3);
let mut c = Consumer::boot(log, FakeDumper::default()).await.unwrap();
c.run(&spec(tmp.path(), &src)).await.unwrap();
let (log, _) = c.into_parts();
let moved = log.materialize(REPO).await.unwrap();
assert_eq!((moved.anchor_seq, moved.skipped), (3, false), "{moved:?}");
for kind in Kind::SOURCE {
let fast = log.readfast_rows(kind, REPO, None, &["payload_hash"]).await.unwrap();
assert_eq!(open_rows(&fast, key_columns(kind)), fold_oracle(&log, kind, 3).await, "{kind:?} after the move");
}
}
/// Every version row of the repo's zone as sorted `krate|key…|from|to|hash`.
async fn zone_rows(log: &EventLog, kind: Kind) -> Vec<String> {
let mut out = Vec::new();
for b in log.interval_rows(kind, REPO).await.unwrap() {
let s = |name: &str, i: usize| -> String {
let c = skade::arrow_cast::cast(b.column_by_name(name).unwrap(), &DataType::Utf8).unwrap();
let c = c.as_any().downcast_ref::<StringArray>().unwrap();
if c.is_null(i) { "\u{2205}".to_string() } else { c.value(i).to_string() }
};
for i in 0..b.num_rows() {
let mut row = s("krate", i);
for k in key_columns(kind) {
row.push('|');
row.push_str(&s(k, i));
}
for c in ["valid_from_seq", "valid_to_seq", "payload_hash"] {
row.push('|');
row.push_str(&s(c, i));
}
out.push(row);
}
}
out.sort();
out
}
/// A consume's materialize ADVANCES the zone (its open rows + the tail)
/// instead of refolding the history, and the advanced zone is the one a
/// single fresh materialize at the tip writes, row for row; time travel over
/// it still equals the fold at every seq. RED before: every materialize
/// rescanned the repo's whole history (`incremental` false).
#[tokio::test(flavor = "current_thread")]
async fn materialize_advances_the_zone_and_equals_a_fresh_one() {
let tmp = tempfile::tempdir().unwrap();
let src = Scratch::new(tmp.path().join("src"));
let mut log = consumed(tmp.path(), &src, 1).await;
let first = log.materialize(REPO).await.unwrap();
assert!(!first.incremental, "the first build folds the history: {first:?}");
for step in 2..=3 {
commit_step(&src, step);
let mut c = Consumer::boot(log, FakeDumper::default()).await.unwrap();
c.run(&spec(tmp.path(), &src)).await.unwrap();
log = c.into_parts().0;
let r = log.materialize(REPO).await.unwrap();
assert!(r.incremental && r.anchor_seq == step as i64, "step {step}: {r:?}");
}
let tmp2 = tempfile::tempdir().unwrap();
let src2 = Scratch::new(tmp2.path().join("src"));
let fresh = consumed(tmp2.path(), &src2, 3).await;
assert!(!fresh.materialize(REPO).await.unwrap().incremental);
for kind in Kind::SOURCE {
assert_eq!(zone_rows(&log, kind).await, zone_rows(&fresh, kind).await, "{kind:?}: advanced zone = fresh zone");
for seq in 1..=3i64 {
let fast = log.readfast_rows(kind, REPO, Some(seq), &["payload_hash"]).await.unwrap();
assert_eq!(open_rows(&fast, key_columns(kind)), fold_oracle(&log, kind, seq).await, "{kind:?} as of seq {seq}");
}
}
}