crates/mimir-room/src/window.rs
//! One repo's commit window (seq N−1 → N) straight from the READ-FAST
//! zone: the node versions whose validity starts or ends at N are the
//! candidates; each candidate's node is then rebuilt on both sides from
//! its interval rows by the rules the room's build applies (the highest
//! priority row wins, the first among equals; the payload hash folds the
//! distinct hashes of the top-priority rows), and a candidate without a
//! node row on a side is still THERE when an edge names it (an endpoint).
//! What [`Room::diff`] of the two as-of rooms says of every item — without
//! folding either room.
//!
//! It answers only where the two rooms would be built from the zone and the
//! source stream alone: both seqs inside the anchor (no event tail) and no
//! binary stream for the repo at either seq (binary rows and linked symbols
//! join a node's rows, and the ring lookup they trigger). Elsewhere it
//! returns `None` and the caller folds the rooms.
use std::collections::{BTreeMap, HashMap, HashSet};
use anyhow::Result;
use mimir_ingest::{EventLog, FoldScope, Kind};
use mimir_schema::{EdgeKind, Level, Origin};
use skade::arrow_array::{Array, BooleanArray, Int64Array, RecordBatch, StringArray};
use crate::snapshot::hash64;
use crate::Room;
/// A node of one side of the window, as the room of that side holds it —
/// what a change ranking reads of it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WindowNode {
/// `None` = only an edge endpoint on this side (no node row).
pub krate: Option<String>,
pub level: Option<Level>,
pub origin: Option<Origin>,
/// A source definition describes it.
pub defined: bool,
pub effective_visibility: Option<String>,
}
/// `seq − 1 → seq` of one repo, by moniker, sorted — what `Room::diff`
/// lists, each with its node on the side the diff reads it from.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct CommitWindow {
/// Present at `seq`, not at `seq − 1` (the node at `seq`).
pub added: Vec<(String, WindowNode)>,
/// Present at `seq − 1`, not at `seq` (the node at `seq − 1`).
pub removed: Vec<(String, WindowNode)>,
/// Present at both under another payload hash (the node at `seq`).
pub changed: Vec<(String, WindowNode)>,
/// Interval rows read (nodes, plus edges when an endpoint was asked).
pub rows_read: u64,
}
const P_DEFINED: u8 = 4;
const P_STUB: u8 = 2;
/// One side's rows of one candidate, in read order.
#[derive(Default)]
struct Side {
/// (prio, hash64, batch, row)
rows: Vec<(u8, u64, u32, u32)>,
}
fn utf8<'a>(b: &'a RecordBatch, name: &str) -> Option<&'a StringArray> {
b.column_by_name(name).and_then(|c| c.as_any().downcast_ref::<StringArray>())
}
fn col_str<'a>(a: Option<&'a StringArray>, r: usize) -> Option<&'a str> {
a.filter(|a| !a.is_null(r)).map(|a| a.value(r))
}
impl Room {
/// The commit window `seq − 1 → seq` of `repo` from the read-fast zone
/// (see the module docs); `None` = not answerable there, fold the rooms.
/// One committed state of the lake, like a room's open.
pub async fn commit_window(log: &EventLog, repo: &str, seq: i64) -> Result<Option<CommitWindow>> {
let log = &log.pin().await?;
commit_window(log, repo, seq).await
}
/// [`Room::commit_window`] for a caller with no async context (see [`Room::at_blocking`]).
pub fn commit_window_blocking(log: &EventLog, repo: &str, seq: i64) -> Result<Option<CommitWindow>> {
std::thread::scope(|s| s.spawn(|| log.block_on(Self::commit_window(log, repo, seq))).join().unwrap_or_else(|p| std::panic::resume_unwind(p)))
}
}
async fn commit_window(log: &EventLog, repo: &str, seq: i64) -> Result<Option<CommitWindow>> {
if seq < 1 {
return Ok(None);
}
// the binary stream joins a node's rows: a repo with any at either side is folded
for s in [seq - 1, seq] {
let scope = FoldScope { repos: [(repo.to_string(), s)].into(), universe: false };
for kind in [Kind::BinNodes, Kind::BinEdges, Kind::Symbols] {
if !log.fold_rows(kind, &scope, &[]).await?.open.is_empty() {
return Ok(None);
}
}
}
let ev = "effective_visibility";
let has_ev = log.spec(Kind::Nodes).schema.field_with_name(ev).is_ok();
let mut cols = vec!["id", "kind", "level", "crate", "origin", "defined_here"];
if has_ev {
cols.push(ev);
}
let Some((anchor, nodes)) = log.interval_rows_at(Kind::Nodes, repo, &cols).await? else { return Ok(None) };
if anchor < seq {
return Ok(None);
}
let (lo, hi) = (seq - 1, seq);
let live = |from: &Int64Array, to: &Int64Array, i: usize, s: i64| from.value(i) <= s && (to.is_null(i) || to.value(i) > s);
let mut out = CommitWindow::default();
// the candidates: monikers with a version that starts or ends at `seq`
let mut cands: HashSet<&str> = HashSet::new();
for b in &nodes {
let (id, from, to) = (utf8(b, "id").expect("id"), int64(b, "valid_from_seq"), int64(b, "valid_to_seq"));
out.rows_read += b.num_rows() as u64;
for i in 0..b.num_rows() {
if from.value(i) == seq || (!to.is_null(i) && to.value(i) == seq) {
cands.insert(id.value(i));
}
}
}
// every row of a candidate live on a side, in read order
let mut sides: HashMap<&str, [Side; 2]> = cands.iter().map(|m| (*m, Default::default())).collect();
for (bi, b) in nodes.iter().enumerate() {
let (id, from, to) = (utf8(b, "id").expect("id"), int64(b, "valid_from_seq"), int64(b, "valid_to_seq"));
let (defined, hash) = (boolean(b, "defined_here"), utf8(b, "payload_hash").expect("payload_hash"));
for i in 0..b.num_rows() {
let Some(s) = sides.get_mut(id.value(i)) else { continue };
let prio = if !defined.is_null(i) && defined.value(i) { P_DEFINED } else { P_STUB };
let row = (prio, hash64(hash, i), bi as u32, i as u32);
if live(from, to, i, lo) {
s[0].rows.push(row);
}
if live(from, to, i, hi) {
s[1].rows.push(row);
}
}
}
// a side without a node row: the moniker is still there when a live edge names it
let bare: HashSet<&str> = sides.iter().filter(|(_, s)| s[0].rows.is_empty() || s[1].rows.is_empty()).map(|(m, _)| *m).collect();
let mut endpoint: HashMap<&str, [bool; 2]> = HashMap::new();
if !bare.is_empty() {
let Some((ea, edges)) = log.interval_rows_at(Kind::Edges, repo, &["from", "to", "kind"]).await? else { return Ok(None) };
if ea < seq {
return Ok(None);
}
for b in &edges {
let (f, t, k, from, to) = (utf8(b, "from").expect("from"), utf8(b, "to").expect("to"), utf8(b, "kind").expect("kind"), int64(b, "valid_from_seq"), int64(b, "valid_to_seq"));
out.rows_read += b.num_rows() as u64;
for i in 0..b.num_rows() {
let (fm, tm) = (f.value(i), t.value(i));
if !bare.contains(fm) && !bare.contains(tm) {
continue;
}
let (l, h) = (live(from, to, i, lo), live(from, to, i, hi));
if !(l || h) || EdgeKind::parse(k.value(i)).is_none() {
continue;
}
for m in [fm, tm] {
if let Some(&m) = bare.get(m) {
let e = endpoint.entry(m).or_default();
e[0] |= l;
e[1] |= h;
}
}
}
}
}
// each candidate's node on each side, by the room build's rules
let view = |s: &Side| -> (u64, WindowNode) {
let top = s.rows.iter().map(|r| r.0).max().unwrap_or(0);
let mut hashes: Vec<u64> = s.rows.iter().filter(|r| r.0 == top).map(|r| r.1).collect();
hashes.sort_unstable();
hashes.dedup();
let phash = hashes.iter().fold(0u64, |acc, h| acc.rotate_left(17) ^ h);
let &(_, _, bi, ri) = s.rows.iter().find(|r| r.0 == top).expect("a row");
let (b, r) = (&nodes[bi as usize], ri as usize);
let node = WindowNode {
krate: Some(col_str(utf8(b, "crate"), r).unwrap_or("").to_string()),
level: col_str(utf8(b, "level"), r).and_then(Level::parse),
origin: col_str(utf8(b, "origin"), r).and_then(Origin::parse),
defined: top == P_DEFINED,
effective_visibility: if has_ev { col_str(utf8(b, ev), r).map(str::to_string) } else { None },
};
(phash, node)
};
let bare_node = WindowNode { krate: None, level: None, origin: None, defined: false, effective_visibility: None };
let mut sorted: BTreeMap<&str, &[Side; 2]> = BTreeMap::new();
for (m, s) in &sides {
sorted.insert(m, s);
}
for (m, s) in sorted {
let ends = endpoint.get(m).copied().unwrap_or_default();
let side = |k: usize| -> Option<(u64, WindowNode)> {
if !s[k].rows.is_empty() {
Some(view(&s[k]))
} else if ends[k] {
Some((0, bare_node.clone()))
} else {
None
}
};
match (side(0), side(1)) {
(Some((_, n)), None) => out.removed.push((m.to_string(), n)),
(None, Some((_, n))) => out.added.push((m.to_string(), n)),
(Some((ha, _)), Some((hb, n))) if ha != hb => out.changed.push((m.to_string(), n)),
_ => {}
}
}
Ok(Some(out))
}
fn int64<'a>(b: &'a RecordBatch, name: &str) -> &'a Int64Array {
b.column_by_name(name).and_then(|c| c.as_any().downcast_ref::<Int64Array>()).expect("an int64 interval column")
}
fn boolean<'a>(b: &'a RecordBatch, name: &str) -> &'a BooleanArray {
b.column_by_name(name).and_then(|c| c.as_any().downcast_ref::<BooleanArray>()).expect("a boolean column")
}