crates/mimir-ingest/tests/ingest/evolve.rs
//! Schema evolution of a lake in place: a lake written with FEWER payload
//! columns opens under the current schema (the columns are added, the old
//! rows read null), the next consume writes an event only for a row the
//! producer now says something new about, and a difference that is not
//! "nullable columns added" is still the typed refusal.
use std::path::Path;
use std::sync::Arc;
use super::common::*;
use mimir_ingest::{Consumer, EventLog, Kind, LakeSchemaMismatch, PayloadSpec, RepoSpec, Specs, Unevolvable};
use mimir_schema::{EdgeKind, EdgeRow, Level, NodeRow, Origin};
use skade::arrow_array::{Array, RecordBatch};
use skade::arrow_schema::{DataType, Field, Schema};
const REPO: &str = "fixture";
const NODE_TABLE: &str = "mimir_v2.code_node_events";
/// The edge payload columns `mimir-schema` 1 did not have.
const SINCE_1_EDGES: [&str; 3] = ["negated", "partial", "access"];
/// The node payload columns `mimir-schema` 1 did not have: `visibility` and
/// everything after it (so the test follows the schema when it grows again).
fn node_columns_since_1() -> Vec<String> {
mimir_schema::nodes_schema().fields().iter().map(|f| f.name().clone()).skip_while(|n| n != "visibility").collect()
}
/// `schema` without `cols`, stamped at the identity epoch (`5`, the first
/// version whose lakes are evolved in place at all — one stamped earlier is
/// wiped on open, see `a_pre_stable_identity_lake_is_wiped_on_open`).
fn without(schema: &Schema, cols: &[&str]) -> Schema {
let kept: Vec<Field> = schema.fields().iter().filter(|f| !cols.contains(&f.name().as_str())).map(|f| f.as_ref().clone()).collect();
let mut meta = schema.metadata().clone();
meta.insert(mimir_schema::META_SCHEMA.to_string(), "5".to_string());
Schema::new(kept).with_metadata(meta)
}
fn strip(b: &RecordBatch, cols: &[&str]) -> RecordBatch {
let schema = b.schema();
let keep: Vec<usize> = (0..b.num_columns()).filter(|i| !cols.contains(&schema.field(*i).name().as_str())).collect();
b.project(&keep).unwrap()
}
/// The payloads of a lake that LACKS columns: today's, minus every column
/// `mimir-schema` 1 did not have, stamped at the identity epoch — the shape
/// a future version that only adds nullable columns will meet.
fn sparse_specs() -> Specs {
let since = node_columns_since_1();
let since: Vec<&str> = since.iter().map(String::as_str).collect();
let nodes = Arc::new(without(&mimir_schema::nodes_schema(), &since));
let edges = Arc::new(without(&mimir_schema::edges_schema(), &SINCE_1_EDGES));
let owned = ["repo", "valid_from_sha", "valid_to_sha", "file", "start_line", "end_line"];
let bin_owned = ["repo", "provenance", "valid_from_sha", "valid_to_sha", "file", "start_line", "end_line"];
Specs {
nodes: PayloadSpec::with_log_owned(nodes.clone(), &["id"], &owned).unwrap(),
edges: PayloadSpec::with_log_owned(edges.clone(), &["from", "to", "kind"], &["valid_from_sha", "valid_to_sha"]).unwrap(),
spans: mimir_ingest::code_spans_spec().unwrap(),
bin_nodes: PayloadSpec::with_log_owned(nodes, &["id"], &bin_owned).unwrap(),
bin_edges: PayloadSpec::with_log_owned(edges, &["from", "to", "kind"], &["provenance", "valid_from_sha", "valid_to_sha"]).unwrap(),
symbols: mimir_ingest::code_symbols_spec().unwrap(),
}
}
/// Dumps the given batches, one entry per commit.
struct Batches(Vec<(RecordBatch, RecordBatch)>);
impl mimir_ingest::Dumper for Batches {
fn dump(&mut self, _: &Path, _: &str) -> anyhow::Result<mimir_ingest::CommitDump> {
let (nodes, edges) = self.0.remove(0);
Ok(mimir_ingest::CommitDump {
toolchain: TOOLCHAIN.into(),
workspace_crates: ["k@0.1.0:lib".to_string()].into(),
workspace_targets: ["k:lib".to_string()].into(),
failed_units: Default::default(),
owner_disambiguations: 0,
crates: vec![mimir_ingest::CrateDump { krate: "k@0.1.0:lib".into(), universe: false, nodes: vec![nodes], edges: vec![edges], spans: vec![] }],
binary: false,
binaries: vec![],
})
}
}
fn node(id: &str) -> NodeRow {
NodeRow::new(id, "fn", Level::Item, id, "k", "0.1.0", Origin::Local)
}
fn repo(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 }
}
async fn node_table(log: &EventLog) -> skade::Table {
log.warehouse().table(NODE_TABLE).await.unwrap()
}
#[tokio::test(flavor = "current_thread")]
async fn a_lake_lacking_added_columns_is_evolved_in_place() {
let tmp = tempfile::tempdir().unwrap();
let lake = tmp.path().join("lake");
let src = Scratch::new(tmp.path().join("src"));
src.write("src/lib.rs", "1\n");
src.commit("one");
let rs = repo(tmp.path(), &src);
let since = node_columns_since_1();
let since_refs: Vec<&str> = since.iter().map(String::as_str).collect();
let call = vec![EdgeRow::new("f", "g", EdgeKind::Calls)];
// The lake as a sparser build wrote it: f, g and one call, no column
// added since.
let old = EventLog::open(skade::open(&lake).await.unwrap(), sparse_specs()).await.unwrap();
assert_eq!(node_table(&old).await.properties().get("mimir.schema").map(String::as_str), Some("5"));
let at_1 = Batches(vec![(
strip(&mimir_schema::nodes_batch(&[node("f"), node("g")]).unwrap(), &since_refs),
strip(&mimir_schema::edges_batch(&call).unwrap(), &SINCE_1_EDGES),
)]);
let mut c = Consumer::boot(old, at_1).await.unwrap();
let report = c.run(&rs).await.unwrap();
assert_eq!((report[0].rows_plus, report[0].rows_minus), (3, 0));
let live = c.open_set().clone();
drop(c);
// Opened by THIS build: no refusal, no wipe — the columns are added.
let log = EventLog::open_lake(&lake).await.unwrap_or_else(|e| panic!("an older lake must open by evolution: {e:#}"));
let t = node_table(&log).await;
let stored = t.arrow_schema().unwrap();
for c in &since {
assert!(stored.field_with_name(c).is_ok(), "`{c}` was not added");
}
assert_eq!(t.properties().get("mimir.schema").map(String::as_str), Some(mimir_schema::SCHEMA_VERSION), "the stamp follows the evolution");
assert_eq!(t.properties().get("mimir.added"), Some(&since.join(",")));
// (a table handle keeps the lake's catalog open)
drop(t);
assert_eq!(log.spec(Kind::Nodes).added, since);
assert_eq!(log.spec(Kind::Edges).added, SINCE_1_EDGES);
assert!(log.spec(Kind::Spans).added.is_empty(), "a table that lacked nothing is not touched");
// the old rows are all there and read the new columns as null
let folded = log.fold(None, None).await.unwrap();
for kind in [Kind::Nodes, Kind::Edges] {
assert_eq!(folded.repo(kind, REPO), live.repo(kind, REPO), "the old events fold to the same open set");
}
let rows = log.events(Kind::Nodes).await.unwrap();
assert_eq!(rows.iter().map(RecordBatch::num_rows).sum::<usize>(), 2);
for b in &rows {
for c in &since {
assert_eq!(b.column_by_name(c).unwrap().null_count(), b.num_rows(), "`{c}` of a row written before the evolution");
}
}
// a reader that asks for an added column by name (the room does) gets
// it, null, also before any release was written at the new schema
let scope = mimir_ingest::FoldScope { repos: [(REPO.to_string(), i64::MAX)].into(), universe: false };
let folded = log.fold_rows(Kind::Nodes, &scope, &["kind", "visibility"]).await.unwrap();
assert_eq!(folded.open.len(), 2);
for b in &folded.batches {
assert_eq!(b.column_by_name("visibility").unwrap().null_count(), b.num_rows());
assert_eq!(b.column_by_name("kind").unwrap().null_count(), 0);
}
// the streamed fold reads the evolved table as the fold does
super::common::fold_each_agrees_with_fold_rows(&log, REPO).await;
// The next consume, at this build's schema: `f` says nothing new (every
// added column null) → NO event; `g` is now known to be `pub` → one; `h`
// is new → one. The call (no flag set) → none.
src.write("src/lib.rs", "2\n");
src.commit("two");
let mut g = node("g");
g.visibility = Some("pub".into());
g.has_doc = Some(false);
let now = (mimir_schema::nodes_batch(&[node("f"), g, node("h")]).unwrap(), mimir_schema::edges_batch(&call).unwrap());
let mut c = Consumer::boot(log, Batches(vec![now.clone(), now.clone()])).await.unwrap();
let report = c.run(&rs).await.unwrap();
assert_eq!(report.len(), 1);
assert_eq!((report[0].rows_plus, report[0].rows_minus), (2, 0), "only what the producer now says is an event");
let (log, _) = c.into_parts();
let evs = shape(&events(&log, Kind::Nodes).await);
let key = |seq: i64, id: &str| (REPO.to_string(), seq, "+".to_string(), "k:lib".to_string(), id.to_string());
assert_eq!(evs, vec![key(1, "f"), key(1, "g"), key(2, "g"), key(2, "h")]);
assert_eq!(events(&log, Kind::Edges).await.len(), 1, "the edge was not written again");
// the new row carries the value, in whatever order the table has its columns
let mut seen = Vec::new();
for b in log.events(Kind::Nodes).await.unwrap() {
let (id, seq, vis) = (b.column_by_name("id").unwrap(), b.column_by_name("seq").unwrap(), b.column_by_name("visibility").unwrap());
let id = skade::arrow_cast::cast(id, &DataType::Utf8).unwrap();
let vis = skade::arrow_cast::cast(vis, &DataType::Utf8).unwrap();
let (id, vis) = (id.as_any().downcast_ref::<skade::arrow_array::StringArray>().unwrap(), vis.as_any().downcast_ref::<skade::arrow_array::StringArray>().unwrap());
let seq = seq.as_any().downcast_ref::<skade::arrow_array::Int64Array>().unwrap();
for i in 0..b.num_rows() {
seen.push((id.value(i).to_string(), seq.value(i), (!vis.is_null(i)).then(|| vis.value(i).to_string())));
}
}
seen.sort();
assert_eq!(seen[2], ("g".to_string(), 2, Some("pub".to_string())));
assert_eq!(seen[1], ("g".to_string(), 1, None));
let live = log.fold(None, None).await.unwrap();
let at = node_table(&log).await.inner().metadata_location().map(str::to_string);
drop(log);
// Opening it again changes nothing (no second evolution), and the same
// dump is an empty delta.
let log = EventLog::open_lake(&lake).await.unwrap();
assert_eq!(node_table(&log).await.inner().metadata_location().map(str::to_string), at, "a lake at this build's schema is not committed to on open");
assert_eq!(log.spec(Kind::Nodes).added, since);
assert_eq!(log.fold(None, None).await.unwrap(), live);
src.write("src/lib.rs", "3\n");
src.commit("three");
let mut c = Consumer::boot(log, Batches(vec![now])).await.unwrap();
let report = c.run(&rs).await.unwrap();
assert_eq!((report[0].rows_plus, report[0].rows_minus), (0, 0));
}
/// The payloads of a lake whose edge table lacks ONE column
/// (`code_edges.access`, the column schema 4 added), stamped at the identity
/// epoch.
fn specs_without_access() -> Specs {
let mut meta = mimir_schema::edges_schema().metadata().clone();
meta.insert(mimir_schema::META_SCHEMA.to_string(), "5".to_string());
let kept: Vec<Field> = mimir_schema::edges_schema().fields().iter().filter(|f| f.name() != "access").map(|f| f.as_ref().clone()).collect();
let edges = Arc::new(Schema::new(kept).with_metadata(meta));
let mut specs = Specs::mimir().unwrap();
specs.edges = PayloadSpec::with_log_owned(edges.clone(), &["from", "to", "kind"], &["valid_from_sha", "valid_to_sha"]).unwrap();
specs.bin_edges = PayloadSpec::with_log_owned(edges, &["from", "to", "kind"], &["provenance", "valid_from_sha", "valid_to_sha"]).unwrap();
specs
}
/// A lake whose edge table has no `access` gains it in place. The edges
/// written before read it as null and are NOT written again; the first
/// TOUCHES_TABLE edge lands with its value.
#[tokio::test(flavor = "current_thread")]
async fn a_lake_without_access_gains_it_and_the_first_table_touch_lands_with_it() {
let tmp = tempfile::tempdir().unwrap();
let lake = tmp.path().join("lake");
let src = Scratch::new(tmp.path().join("src"));
src.write("src/lib.rs", "1\n");
src.commit("one");
let rs = repo(tmp.path(), &src);
let call = EdgeRow::new("f", "g", EdgeKind::Calls);
let old = EventLog::open(skade::open(&lake).await.unwrap(), specs_without_access()).await.unwrap();
let at_3 = Batches(vec![(mimir_schema::nodes_batch(&[node("f"), node("g")]).unwrap(), strip(&mimir_schema::edges_batch(std::slice::from_ref(&call)).unwrap(), &["access"]))]);
let mut c = Consumer::boot(old, at_3).await.unwrap();
assert_eq!(c.run(&rs).await.unwrap()[0].rows_plus, 3);
drop(c);
// opened by this build: the one column is added, nothing else moves
let log = EventLog::open_lake(&lake).await.unwrap_or_else(|e| panic!("a lake lacking a nullable column must open by evolution: {e:#}"));
assert_eq!(log.spec(Kind::Edges).added, ["access"]);
assert!(log.spec(Kind::Nodes).added.is_empty(), "the node table lacked nothing");
for b in log.events(Kind::Edges).await.unwrap() {
assert_eq!(b.column_by_name("access").unwrap().null_count(), b.num_rows(), "an edge written before the evolution says nothing");
}
// the next commit: the same call (no event), and `f` now touches a table
src.write("src/lib.rs", "2\n");
src.commit("two");
let table = mimir_schema::moniker::table("bench_runs");
let touch = EdgeRow { access: Some(mimir_schema::Access::Write), ..EdgeRow::new("f", table.clone(), EdgeKind::TouchesTable) };
let table_node = NodeRow::new(table.clone(), mimir_schema::kind::TABLE, Level::Item, "bench_runs", mimir_schema::moniker::TABLE_PACKAGE, mimir_schema::moniker::TABLE_VERSION, Origin::Local);
let now = (mimir_schema::nodes_batch(&[node("f"), node("g"), table_node]).unwrap(), mimir_schema::edges_batch(&[call, touch]).unwrap());
let mut c = Consumer::boot(log, Batches(vec![now])).await.unwrap();
let report = c.run(&rs).await.unwrap();
assert_eq!((report[0].rows_plus, report[0].rows_minus), (2, 0), "the table node and the touch; the old call is not an event");
let (log, _) = c.into_parts();
// read back by name, as the room reads it
let scope = mimir_ingest::FoldScope { repos: [(REPO.to_string(), i64::MAX)].into(), universe: false };
let folded = log.fold_rows(Kind::Edges, &scope, &["access"]).await.unwrap();
let mut seen: Vec<(String, Option<String>)> = Vec::new();
for &(b, r) in &folded.open {
let batch = &folded.batches[b as usize];
let col = |name: &str| {
let c = skade::arrow_cast::cast(batch.column_by_name(name).unwrap(), &DataType::Utf8).unwrap();
let c = c.as_any().downcast_ref::<skade::arrow_array::StringArray>().unwrap().clone();
(!c.is_null(r as usize)).then(|| c.value(r as usize).to_string())
};
seen.push((col("kind").unwrap(), col("access")));
}
seen.sort();
assert_eq!(seen, vec![("CALLS".to_string(), None), ("TOUCHES_TABLE".to_string(), Some("write".to_string()))]);
}
fn mismatch(err: &anyhow::Error) -> &LakeSchemaMismatch {
err.downcast_ref::<LakeSchemaMismatch>().unwrap_or_else(|| panic!("not typed: {err:#}"))
}
async fn reopen(lake: &Path, specs: Specs) -> anyhow::Error {
EventLog::open(skade::open(lake).await.unwrap(), specs).await.err().expect("must refuse")
}
#[tokio::test(flavor = "current_thread")]
async fn a_lake_with_a_column_this_build_does_not_have_is_refused() {
let tmp = tempfile::tempdir().unwrap();
let lake = tmp.path().join("lake");
drop(open_log(&lake).await);
// this build's node payload dropped `kind`, `signature`, `body` (or the
// lake is from a build that has them)
let mut sp = specs();
sp.nodes = PayloadSpec::new(Arc::new(Schema::new(vec![Field::new("moniker", DataType::Utf8, false)])), &["moniker"]).unwrap();
let err = reopen(&lake, sp).await;
let m = mismatch(&err);
assert_eq!((m.table.as_str(), &m.why), (NODE_TABLE, &Unevolvable::Columns));
assert!(m.have.contains(&"signature".to_string()) && !m.want.contains(&"signature".to_string()));
let said = format!("{err:#}");
assert!(said.contains("not migrated") && said.contains("delete the lake directory") && said.contains("[kind, signature, body]"), "{said}");
}
#[tokio::test(flavor = "current_thread")]
async fn a_retyped_column_is_refused() {
let tmp = tempfile::tempdir().unwrap();
let lake = tmp.path().join("lake");
drop(open_log(&lake).await);
let fields = node_spec().schema.fields().iter().map(|f| if f.name() == "body" { Field::new("body", DataType::Int64, false) } else { f.as_ref().clone() }).collect::<Vec<_>>();
let mut sp = specs();
sp.nodes = PayloadSpec::new(Arc::new(Schema::new(fields)), &["moniker"]).unwrap();
let err = reopen(&lake, sp).await;
assert!(matches!(&mismatch(&err).why, Unevolvable::Retyped { column, .. } if column == "body"), "{err:#}");
assert!(format!("{err:#}").contains("delete the lake directory"));
}
#[tokio::test(flavor = "current_thread")]
async fn a_missing_key_column_is_refused_not_added() {
let tmp = tempfile::tempdir().unwrap();
let lake = tmp.path().join("lake");
// the lake's edges are keyed (from, to); this build's also by `kind`
let mut old = specs();
old.edges = PayloadSpec::new(Arc::new(Schema::new(vec![Field::new("from", DataType::Utf8, false), Field::new("to", DataType::Utf8, false)])), &["from", "to"]).unwrap();
drop(EventLog::open(skade::open(&lake).await.unwrap(), old).await.unwrap());
let err = reopen(&lake, specs()).await;
let m = mismatch(&err);
assert_eq!((m.table.as_str(), &m.why), ("mimir_v2.code_edge_events", &Unevolvable::Required { column: "kind".into() }));
}
#[tokio::test(flavor = "current_thread")]
async fn a_lake_newer_than_the_build_is_refused() {
let tmp = tempfile::tempdir().unwrap();
let lake = tmp.path().join("lake");
// same columns, written by a later mimir-schema (new enum kinds need no
// column): only the stamp can tell
let mut newer = Specs::mimir().unwrap();
let mut meta = newer.nodes.schema.metadata().clone();
meta.insert(mimir_schema::META_SCHEMA.to_string(), "999".to_string());
newer.nodes.schema = Arc::new(newer.nodes.schema.as_ref().clone().with_metadata(meta));
drop(EventLog::open(skade::open(&lake).await.unwrap(), newer).await.unwrap());
let err = EventLog::open_lake(&lake).await.err().expect("must refuse");
assert_eq!(mismatch(&err).why, Unevolvable::Newer { lake: "999".into(), build: mimir_schema::SCHEMA_VERSION.into() });
}
/// Version 5 changed what an identity IS (stable keys and owners), not a
/// column: a lake stamped before it cannot be evolved. `open` refuses it
/// typed ([`Unevolvable::Older`]); `open_lake` answers that refusal itself —
/// the lake is derived data — by wiping the directory and starting fresh.
#[tokio::test(flavor = "current_thread")]
async fn a_pre_stable_identity_lake_is_wiped_on_open() {
let tmp = tempfile::tempdir().unwrap();
let lake = tmp.path().join("lake");
// same columns, written at mimir-schema 4: its rows key and own by
// VERSIONED names — only the stamp can tell
let mut at_4 = Specs::mimir().unwrap();
let mut meta = at_4.nodes.schema.metadata().clone();
meta.insert(mimir_schema::META_SCHEMA.to_string(), "4".to_string());
at_4.nodes.schema = Arc::new(at_4.nodes.schema.as_ref().clone().with_metadata(meta));
let old = EventLog::open(skade::open(&lake).await.unwrap(), at_4).await.unwrap();
let old_identity = old.identity().await.unwrap();
drop(old);
// an in-memory open is told why, and told it typed
let err = reopen(&lake, Specs::mimir().unwrap()).await;
assert_eq!(mismatch(&err).why, Unevolvable::Older { lake: "4".into(), build: mimir_schema::SCHEMA_VERSION.into() });
assert!(format!("{err:#}").contains("stable identities"), "{err:#}");
// the on-disk open wipes and reopens: a NEW lake (new table UUIDs), empty
let log = EventLog::open_lake(&lake).await.unwrap_or_else(|e| panic!("a pre-stable lake must be wiped, not refused: {e:#}"));
assert_ne!(log.identity().await.unwrap(), old_identity, "the lake was not wiped");
assert_eq!(node_table(&log).await.properties().get("mimir.schema").map(String::as_str), Some(mimir_schema::SCHEMA_VERSION));
assert!(log.events(Kind::Nodes).await.unwrap().iter().all(|b| b.num_rows() == 0), "a wiped lake holds no events");
}