crates/mimir-room/src/snapshot.rs
//! Event log → snapshot: fold both producer streams under a scope, give
//! every moniker a dense ordinal (lexicographic, so two snapshots map into
//! their union monotonically), build one boolean CSR matrix per edge kind
//! through the sorted fast path, derive provenance per node.
use std::collections::{BTreeMap, HashMap};
/// The fold's string maps (monikers, files, interned values): foldhash, not
/// SipHash — millions of lookups per build, and no attacker picks the keys.
pub(crate) type FastMap<K, V> = HashMap<K, V, foldhash::fast::RandomState>;
use std::sync::{Mutex, OnceLock};
use std::hash::BuildHasher;
use std::time::Instant;
use anyhow::{Context, Result, bail};
use mimir_graph::Matrix;
use mimir_graph::arrow::{ImportPath, from_edge_chunks};
use mimir_ingest::schema::UNIVERSE_REPO;
use mimir_ingest::{EventLog, FoldDelta, FoldScope, FoldedRows, Kind, PayloadSpec};
use skade::ScanFilter;
use mimir_schema::{BinaryFacts, EDGE_KEY_SEP, EdgeKind, Level, Origin, Provenance, SPAN_EDGE, SPAN_NODE, derive_provenance};
use serde::Serialize;
use skade::arrow_array::{Array, BooleanArray, Int64Array, RecordBatch, StringArray, UInt64Array};
use skade::arrow_schema::DataType;
use crate::evidence::Input;
use crate::ring;
use crate::scope::ResolvedScope;
// ---- node flags (what the streams say about a moniker)
/// An open source node row (definition or reference stub).
pub(crate) const F_SRC: u8 = 1;
/// A source row with `defined_here` (a definition, not a stub).
pub(crate) const F_DEFINED: u8 = 2;
/// The binary stream owns a node with this id.
pub(crate) const F_BIN_NODE: u8 = 4;
/// A linked symbol carries this id.
pub(crate) const F_SYMBOL: u8 = 8;
/// ≥1 INSTANCE_OF edge points at this id.
pub(crate) const F_INSTANCES: u8 = 16;
/// Defined in the universe ring, found by the provenance lookup while the
/// ring itself is outside the scope.
pub(crate) const F_UNIVERSE_DEF: u8 = 32;
/// The describing row says `is_test` (a `#[test]` / bench fn of a harness unit).
pub(crate) const F_TEST: u8 = 64;
/// An edge row of a LOCAL repo (not the universe ring) leaves this node:
/// the node set whose out-edges a lake delta of the consumed repos can move.
pub(crate) const F_LOCAL_OUT: u8 = 128;
/// No value (unknown kind / crate / origin).
pub(crate) const NONE: u32 = u32::MAX;
pub(crate) const NONE8: u8 = u8::MAX;
/// A record the fold keeps a SLOT for (the track's slots stay aligned) but
/// the room never sees: a `mir_block` row (its payload is read per repo
/// from the lake, [`crate::MirSide`]) — no moniker, no string interned.
const DROPPED: u32 = u32::MAX - 1;
/// Does the room keep the `mir_block` rows and their CFG_SUCC edges? Off
/// (the default): they were most of the room's bytes for one lens.
/// `MIMIR_ROOM_MIR_BLOCKS=in` keeps the old room (a measurement switch).
fn keeps_mir_blocks() -> bool {
static KEEP: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
*KEEP.get_or_init(|| std::env::var("MIMIR_ROOM_MIR_BLOCKS").is_ok_and(|v| v == "in"))
}
/// Every moniker of the room, sorted; ordinal = position.
#[derive(Debug, Default)]
pub struct Monikers {
blob: String,
offs: Vec<u64>,
}
impl Monikers {
pub(crate) fn from_sorted<'a>(names: impl Iterator<Item = &'a str>, n: usize) -> Self {
let mut m = Monikers { blob: String::new(), offs: Vec::with_capacity(n + 1) };
m.offs.push(0);
for s in names {
m.blob.push_str(s);
m.offs.push(m.blob.len() as u64);
}
m
}
pub fn len(&self) -> usize {
self.offs.len().saturating_sub(1)
}
pub fn is_empty(&self) -> bool {
self.len() == 0
}
pub fn get(&self, i: u64) -> &str {
let i = i as usize;
&self.blob[self.offs[i] as usize..self.offs[i + 1] as usize]
}
/// Ordinal of `moniker` (binary search).
pub fn find(&self, moniker: &str) -> Option<u64> {
let (mut lo, mut hi) = (0usize, self.len());
while lo < hi {
let mid = (lo + hi) / 2;
match self.get(mid as u64).cmp(moniker) {
std::cmp::Ordering::Less => lo = mid + 1,
std::cmp::Ordering::Greater => hi = mid,
std::cmp::Ordering::Equal => return Some(mid as u64),
}
}
None
}
pub(crate) fn bytes(&self) -> usize {
self.blob.capacity() + self.offs.capacity() * 8
}
}
#[derive(Debug, Default)]
struct Interner {
map: FastMap<String, u32>,
list: Vec<String>,
}
impl Interner {
fn id(&mut self, s: &str) -> u32 {
if let Some(i) = self.map.get(s) {
return *i;
}
let i = self.list.len() as u32;
self.list.push(s.to_string());
self.map.insert(s.to_string(), i);
i
}
}
/// A string column: values concatenated, addressed by row.
#[derive(Debug)]
pub(crate) struct Blob {
bytes: String,
offs: Vec<u64>,
}
impl Default for Blob {
fn default() -> Self {
Self { bytes: String::new(), offs: vec![0] }
}
}
impl Blob {
pub(crate) fn push(&mut self, s: &str) {
self.bytes.push_str(s);
self.offs.push(self.bytes.len() as u64);
}
pub(crate) fn get(&self, i: usize) -> &str {
&self.bytes[self.offs[i] as usize..self.offs[i + 1] as usize]
}
fn heap(&self) -> usize {
self.bytes.capacity() + self.offs.capacity() * 8
}
}
/// A payload column of `code_node_events` this build has no typed field for
/// (a producer added it later — visibility, doc lines …): folded as it is
/// and read by name ([`crate::Room::attr`]). No API change when one appears.
#[derive(Debug)]
pub(crate) struct ExtraCol {
pub(crate) name: String,
pub(crate) data: ExtraData,
}
#[derive(Debug)]
pub(crate) enum ExtraData {
/// Index into [`NodeTable::strs`], `NONE` = null.
Str(Vec<u32>),
/// `i64::MIN` = null.
Int(Vec<i64>),
/// 0 / 1, 2 = null.
Bool(Vec<u8>),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ExtraType {
Str,
Int,
Bool,
}
/// One source file some span of the room lies in.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub(crate) struct FileEntry {
/// The repo whose dump wrote the span.
pub(crate) repo: String,
/// Path as the driver wrote it (workspace-root relative for local code).
pub(crate) path: String,
}
/// One site of an edge: sorted by (kind, from, to, file, line).
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub(crate) struct Site {
pub(crate) kind: u8,
pub(crate) from: u32,
pub(crate) to: u32,
pub(crate) file: u32,
pub(crate) line: u32,
}
/// The binary stream's linked symbols, sorted by (node, symbol).
#[derive(Debug, Default)]
pub(crate) struct SymbolTable {
pub(crate) node: Vec<u32>,
pub(crate) symbol: Blob,
/// Empty = the row carries none.
pub(crate) demangled: Blob,
/// Indexes into `strs`; `NONE` = null.
pub(crate) role: Vec<u32>,
pub(crate) artifact: Vec<u32>,
pub(crate) binary: Vec<u32>,
pub(crate) section: Vec<u32>,
/// `u64::MAX` = null (an import has no size).
pub(crate) size: Vec<u64>,
pub(crate) strs: Vec<String>,
}
impl SymbolTable {
pub(crate) fn len(&self) -> usize {
self.node.len()
}
fn heap(&self) -> usize {
self.symbol.heap() + self.demangled.heap() + self.len() * (5 * 4 + 8)
}
}
/// A node id that more than one open row describes DIFFERENTLY, or that one
/// dump repeated (`key_dup`): the fold merges them into one node; this is
/// the record of what it merged.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct DupKey {
pub node: u64,
/// Which class of rows describes the node (the highest present):
/// `definition`, `binary`, `stub` or `symbol`.
pub rows_of: &'static str,
/// Open rows of that class (one per owner describing the id, plus one
/// per repeat inside a unit).
pub rows: u32,
/// Distinct payload hashes among them: > 1 = the rows DISAGREE about
/// what the item is.
pub distinct_payloads: u32,
/// Rows the dump itself marked as a repeated key (`key_dup`): one id,
/// several items inside ONE compiled unit.
pub key_dup_rows: u32,
/// Owners of the rows (`crate@version:kind`, or the artifact).
pub owners: Vec<String>,
}
/// The node table: one row per ordinal, columns as arrays, strings interned.
#[derive(Debug, Default)]
pub struct NodeTable {
pub(crate) names: Monikers,
pub(crate) kind: Vec<u32>,
pub(crate) krate: Vec<u32>,
pub(crate) version: Vec<u32>,
pub(crate) repo: Vec<u32>,
pub(crate) origin: Vec<u8>,
pub(crate) level: Vec<u8>,
pub(crate) flags: Vec<u8>,
/// 0 = none, else 1 + index into [`Provenance::ALL`].
pub(crate) prov: Vec<u8>,
/// The semantic payload hash of the node's describing row(s) (the
/// highest-priority rows: definition > binary node > stub; distinct
/// hashes combined). A change means the item itself changed.
pub(crate) phash: Vec<u64>,
pub(crate) kinds: Vec<String>,
pub(crate) crates: Vec<String>,
pub(crate) versions: Vec<String>,
pub(crate) repos: Vec<String>,
/// The item's own name (index into `strs`).
pub(crate) name: Vec<u32>,
/// Ordinal of the node the `parent` column names; `NONE` when the row
/// names none or the room does not hold it.
pub(crate) parent: Vec<u32>,
/// The `cfg` gate of the item as written (index into `strs`).
pub(crate) cfg: Vec<u32>,
/// Printed type (params, fields, consts, statics, aliases).
pub(crate) ty: Vec<u32>,
/// Interned strings of name / cfg / ty / string extras.
pub(crate) strs: Vec<String>,
pub(crate) extras: Vec<ExtraCol>,
/// The node's span: index into `files` (`NONE` = no span row), lines.
pub(crate) span_file: Vec<u32>,
pub(crate) span_start: Vec<u32>,
pub(crate) span_end: Vec<u32>,
/// Sorted by (repo, path).
pub(crate) files: Vec<FileEntry>,
/// Further spans of nodes with more than one (node, file, start, end), sorted.
pub(crate) more_spans: Vec<(u32, u32, u32, u32)>,
}
impl NodeTable {
pub fn len(&self) -> usize {
self.names.len()
}
pub fn is_empty(&self) -> bool {
self.names.is_empty()
}
/// A string payload column read by name (the columns the room has no
/// array of its own for).
pub(crate) fn extra_str(&self, i: usize, name: &str) -> Option<&str> {
match &self.extras.iter().find(|e| e.name == name)?.data {
ExtraData::Str(d) => (d[i] != NONE).then(|| self.strs[d[i] as usize].as_str()),
_ => None,
}
}
pub(crate) fn extra_bool(&self, i: usize, name: &str) -> Option<bool> {
match &self.extras.iter().find(|e| e.name == name)?.data {
ExtraData::Bool(d) => (d[i] < 2).then_some(d[i] == 1),
_ => None,
}
}
pub(crate) fn provenance(&self, i: usize) -> Option<Provenance> {
match self.prov[i] {
0 => None,
p => Some(Provenance::ALL[p as usize - 1]),
}
}
fn bytes(&self) -> usize {
let extras = self.extras.iter().map(|e| match &e.data { ExtraData::Str(_) => 4, ExtraData::Int(_) => 8, ExtraData::Bool(_) => 1 }).sum::<usize>();
self.names.bytes()
+ self.len() * (4 * 4 + 4 + 8 + 4 * 4 + 3 * 4 + extras)
+ self.strs.iter().map(|s| s.capacity() + 24).sum::<usize>()
+ self.files.iter().map(|f| f.path.capacity() + f.repo.capacity() + 48).sum::<usize>()
+ self.more_spans.capacity() * 16
}
}
/// Per repo: how many open rows each stream holds for it in this room.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)]
pub struct RepoRows {
pub source: u64,
pub binary: u64,
}
/// Wall clock of one build.
#[derive(Debug, Clone, Default, Serialize)]
pub struct BuildStats {
pub fold_ms: u64,
pub index_ms: u64,
/// Inside `index_ms`: the names sort + ordinal ranking, the node-table
/// fill (set_rec over every open record), the symbol table.
#[serde(default)]
pub sort_ms: u64,
#[serde(default)]
pub fill_ms: u64,
#[serde(default)]
pub symbols_ms: u64,
pub provenance_ms: u64,
pub matrix_ms: u64,
/// Positions: `code_span_events` folded into node spans and edge sites.
pub span_ms: u64,
pub total_ms: u64,
#[serde(serialize_with = "ser_import_paths")]
pub import_paths: BTreeMap<EdgeKind, ImportPath>,
/// Node table + monikers + spans, sites and symbols, bytes (matrices
/// live in GraphBLAS' memory).
pub node_bytes: usize,
}
fn ser_import_paths<S: serde::Serializer>(m: &BTreeMap<EdgeKind, ImportPath>, s: S) -> std::result::Result<S::Ok, S::Error> {
s.collect_map(m.iter().map(|(k, v)| (k.as_str(), format!("{v:?}"))))
}
impl BuildStats {
/// The phase split on one line, for the consumer's "room built" log —
/// which of the phases carries a slow build (the fold scales with the
/// log's whole history, the rest with the room's size).
pub fn phase_note(&self) -> String {
format!(
"fold {} ms, index {} ms (sort {} + fill {} + symbols {}), provenance {} ms, matrices {} ms, spans {} ms",
self.fold_ms, self.index_ms, self.sort_ms, self.fill_ms, self.symbols_ms, self.provenance_ms, self.matrix_ms, self.span_ms
)
}
}
/// The adjacency per edge kind, plus the derived matrices queries asked for
/// (kind unions, transposes, the Jikes walk matrix), built once and kept.
///
/// Why a mutex and not shared read access: SuiteSparse:GraphBLAS 7.4 may
/// MUTATE a matrix that is only an INPUT. Since 7.3.0 a hypersparse matrix
/// carries a lazily built hyper_hash `A->Y` (Doc/ChangeLog v7.3.0: "added
/// A->Y hyper_hash for hypersparse case, speeds up GrB_mxm …"), constructed
/// on first use by `GB_hyper_hash_build` ("construct A->Y if not already
/// constructed", it assigns `A->Y`), and `GrB_wait` does NOT build it — 7.4's
/// Source/GB_wait.c: "Instead, it is not computed and left pending (as
/// NULL)". `GrB_wait(A, GrB_MATERIALIZE)` therefore does not make an input
/// read-only in 7.4, and two threads using the same hypersparse matrix as an
/// input could race on `A->Y`. (The C API spec's GrB_COMPLETE only
/// "establishes a happens-before relation" — GraphBLAS.h 7.4 — for handing
/// an object to another thread, not for concurrent use.) Our adjacencies
/// are n×n with most rows empty, which GraphBLAS stores hypersparse. Parallelism lives inside
/// each call (OpenMP).
pub(crate) struct Graphs {
pub(crate) n: u64,
pub(crate) a: BTreeMap<EdgeKind, Matrix<bool>>,
unions: HashMap<KindMask, Matrix<bool>>,
transposes: HashMap<KindMask, Matrix<bool>>,
/// The Jikes walk matrix (built on first use; inner None = no edge).
jikes: Option<Option<Matrix<bool>>>,
}
/// A set of edge kinds: bit `i` = `EdgeKind::ALL[i]`. 32 bits since the
/// 17th kind (SYSCALLS) — 16 were full with BUILT_FROM.
pub(crate) type KindMask = u32;
// the kind sets of a walk are KindMask bit sets
const _: () = assert!(EdgeKind::ALL.len() <= KindMask::BITS as usize, "more edge kinds than KindMask bits: widen the kind masks");
/// Edge flag: every mention of the gate sits under an odd number of `not(…)`.
pub(crate) const E_NEGATED: u8 = 1;
/// Edge flag: the gate covers something INSIDE the item, not the item.
pub(crate) const E_PARTIAL: u8 = 2;
pub(crate) fn kind_index(k: EdgeKind) -> u8 {
EdgeKind::ALL.iter().position(|x| *x == k).expect("kind in ALL") as u8
}
pub(crate) fn kind_bit(k: EdgeKind) -> KindMask {
1 << kind_index(k)
}
impl Graphs {
/// The adjacency of the kinds in `mask` that the room holds (union when
/// several), `Aᵀ` when `transposed`; `None` if none of them has an edge.
pub(crate) fn get(&mut self, mask: KindMask, transposed: bool) -> Result<Option<&Matrix<bool>>> {
let present: Vec<EdgeKind> = self.a.keys().copied().filter(|k| mask & kind_bit(*k) != 0).collect();
if present.is_empty() {
return Ok(None);
}
let key: KindMask = present.iter().map(|k| kind_bit(*k)).fold(0, |a, b| a | b);
if !transposed {
if present.len() == 1 {
return Ok(self.a.get(&present[0]));
}
if !self.unions.contains_key(&key) {
let mut acc = self.a[&present[0]].dup()?;
for k in &present[1..] {
acc = mimir_graph::ops::union(&acc, &self.a[k])?;
}
acc.wait()?;
self.unions.insert(key, acc);
}
return Ok(self.unions.get(&key));
}
if !self.transposes.contains_key(&key) {
let t = self.get(key, false)?.context("kinds present")?.transpose()?;
t.wait()?;
self.transposes.insert(key, t);
}
Ok(self.transposes.get(&key))
}
/// The adjacency of `mask` AND its cached transpose (built on first use):
/// a walk that steps forward and looks predecessors up by row.
pub(crate) fn get_pair(&mut self, mask: KindMask) -> Result<Option<(&Matrix<bool>, &Matrix<bool>)>> {
if self.get(mask, true)?.is_none() {
return Ok(None);
}
let present: Vec<EdgeKind> = self.a.keys().copied().filter(|k| mask & kind_bit(*k) != 0).collect();
let key: KindMask = present.iter().map(|k| kind_bit(*k)).fold(0, |a, b| a | b);
let fwd = if present.len() == 1 { self.a.get(&present[0]) } else { self.unions.get(&key) };
Ok(fwd.zip(self.transposes.get(&key)))
}
/// The Jikes walk as ONE adjacency: `Jᵀ ∨ diag(impls)·CONTAINS` — walk
/// back along the `mask` kinds (who calls / uses / implements / is impl
/// for), and from every impl block reached step DOWN to the items it
/// contains (a changed type invalidates the methods of its impls, and
/// from there their callers).
pub(crate) fn jikes(&mut self, mask: KindMask, impls: &[u64]) -> Result<Option<&Matrix<bool>>> {
if self.jikes.is_none() {
let back = self.get(mask, true)?.map(|m| m.dup()).transpose()?;
let down = match self.a.get(&EdgeKind::Contains) {
Some(c) if !impls.is_empty() => Some(mimir_graph::ops::mask_rows(c, &mimir_graph::Vector::<bool>::from_indices(self.n, impls)?)?),
_ => None,
};
let w = match (back, down) {
(Some(b), Some(d)) => Some(mimir_graph::ops::union(&b, &d)?),
(b, d) => b.or(d),
};
if let Some(w) = &w {
w.wait()?;
}
self.jikes = Some(w);
}
Ok(self.jikes.as_ref().and_then(Option::as_ref))
}
/// How many derived matrices are cached (kind unions + the Jikes
/// matrix, transposes).
pub(crate) fn cached(&self) -> (usize, usize) {
(self.unions.len() + usize::from(matches!(self.jikes, Some(Some(_)))), self.transposes.len())
}
}
/// One built room state.
pub struct Snapshot {
/// Unique per build in this process ([`crate::Room::version`]).
pub(crate) version: u64,
/// The lake state this room is 1:1 with — STABLE across processes (the
/// log identity, the scope and the event tables' snapshot ids), unlike
/// `version`. What a lake-kept answer is keyed by.
pub(crate) state_key: String,
/// The log's identity (which lake this room is of).
pub(crate) identity: String,
pub(crate) scope: ResolvedScope,
pub(crate) nodes: NodeTable,
pub(crate) graphs: Mutex<Graphs>,
pub(crate) edge_counts: BTreeMap<EdgeKind, u64>,
pub(crate) inputs: Vec<Input>,
pub(crate) repo_rows: BTreeMap<String, RepoRows>,
pub(crate) universe_rows: u64,
/// Edge kinds the log holds that this build does not know (skipped).
pub(crate) unknown_kinds: BTreeMap<String, u64>,
/// Monikers known only as an edge endpoint (no node row in any stream).
pub(crate) endpoint_only: u64,
/// Ring on: (ring owners in the scope's DEPENDS_ON closure, ring owners in the log).
pub(crate) ring_owners: Option<(usize, usize)>,
/// Ring off but linked ring items: the ring definitions were looked up
/// (`Some(true)` = from the ring cache).
pub(crate) ring_lookup: Option<bool>,
/// `universe_manifest` snapshot the build saw.
pub(crate) manifest: Option<i64>,
/// Ring on: FNV-1a over the sorted ring owners the scan read (the
/// DEPENDS_ON closure) — with `manifest`, the identity of the ring rows.
pub(crate) ring_closure: Option<u64>,
pub(crate) stats: BuildStats,
/// Site counts of the edges that have more or fewer than ONE site, per
/// kind, sorted by `(from << 32 | to)`; every other edge has exactly one.
pub(crate) counts: BTreeMap<EdgeKind, Vec<(u64, u32)>>,
/// `E_NEGATED` / `E_PARTIAL` of the edges that carry one (GATED_BY), per
/// kind, sorted by `(from << 32 | to)`.
pub(crate) edge_flags: BTreeMap<EdgeKind, Vec<(u64, u8)>>,
/// `mimir_schema::Access` bits of the edges that carry an `access`
/// (TOUCHES_TABLE), per kind, sorted by `(from << 32 | to)`.
pub(crate) edge_access: BTreeMap<EdgeKind, Vec<(u64, u8)>>,
/// Edge payload columns the lake's edge table gained IN PLACE: it was
/// created by an older `mimir-schema`, so rows written before the
/// evolution say nothing for them.
pub(crate) edge_columns_added: Vec<String>,
/// Every positioned edge site.
pub(crate) sites: Vec<Site>,
/// (impl block, its self type) from FOR_TYPE, sorted by impl.
pub(crate) impl_self: Vec<(u32, u32)>,
pub(crate) symbols: SymbolTable,
pub(crate) dups: Vec<DupKey>,
/// Span rows naming a node / edge the room does not hold.
pub(crate) orphan_spans: u64,
/// Node payload columns this build could not read (type it has no reader for).
pub(crate) unread_columns: Vec<String>,
/// Node ordinals with a `name`, sorted by (name, ordinal) — built on first lookup.
pub(crate) name_idx: OnceLock<Vec<u32>>,
/// Node ordinals with a span, sorted by (file, start, end, ordinal).
pub(crate) file_idx: OnceLock<Vec<u32>>,
/// Symbol rows sorted by symbol name.
pub(crate) symbol_idx: OnceLock<Vec<u32>>,
/// (stable form, ordinal) of every node whose moniker is NOT its own
/// stable form, sorted — how a STABLE id (the ingest keys by them since
/// mimir-schema 5) finds the node(s) whose stored moniker keeps its real
/// version. Built on first lookup.
pub(crate) stable_idx: OnceLock<Vec<(String, u32)>>,
}
/// Utf8 view of a column (cast when the reader returned another string type).
fn utf8(b: &RecordBatch, name: &str) -> Result<StringArray> {
let c = b.column_by_name(name).with_context(|| format!("column `{name}` missing"))?;
let c = if c.data_type() == &DataType::Utf8 { c.clone() } else { skade::arrow_cast::cast(c, &DataType::Utf8)? };
Ok(c.as_any().downcast_ref::<StringArray>().context("utf8 downcast")?.clone())
}
fn boolean(b: &RecordBatch, name: &str) -> Result<BooleanArray> {
let c = b.column_by_name(name).with_context(|| format!("column `{name}` missing"))?;
Ok(c.as_any().downcast_ref::<BooleanArray>().with_context(|| format!("`{name}` is not boolean"))?.clone())
}
/// Any integer column as Int64 (unsigned payload ints are stored widened).
fn int64(b: &RecordBatch, name: &str) -> Result<Int64Array> {
let c = b.column_by_name(name).with_context(|| format!("column `{name}` missing"))?;
let c = if c.data_type() == &DataType::Int64 { c.clone() } else { skade::arrow_cast::cast(c, &DataType::Int64)? };
Ok(c.as_any().downcast_ref::<Int64Array>().with_context(|| format!("`{name}` is not an integer"))?.clone())
}
fn opt_utf8(b: &RecordBatch, name: &str) -> Result<Option<StringArray>> {
if b.column_by_name(name).is_none() { Ok(None) } else { utf8(b, name).map(Some) }
}
/// One extra column of a batch, typed as [`extra_columns`] classified it.
pub(crate) enum ExtraArr {
Str(StringArray),
Int(Int64Array),
Bool(BooleanArray),
Absent,
}
/// The payload columns of the node tables the room has a typed field for.
const KNOWN_NODE_COLUMNS: [&str; 13] =
["id", "kind", "level", "name", "parent", "crate", "version", "cfg", "origin", "provenance", "ty", "is_test", "defined_here"];
/// Typed, but read only when the spec has them (a test schema may not).
const OPTIONAL_NODE_COLUMNS: [&str; 5] = ["name", "parent", "cfg", "ty", "is_test"];
/// The node payload columns beyond [`KNOWN_NODE_COLUMNS`], with the reader
/// each gets; a type without a reader is returned in the second list.
pub(crate) fn extra_columns(spec: &PayloadSpec) -> (Vec<(String, ExtraType)>, Vec<String>) {
let (mut extras, mut unread) = (Vec::new(), Vec::new());
for f in spec.schema.fields() {
if KNOWN_NODE_COLUMNS.contains(&f.name().as_str()) {
continue;
}
match f.data_type() {
DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View => extras.push((f.name().clone(), ExtraType::Str)),
DataType::Boolean => extras.push((f.name().clone(), ExtraType::Bool)),
t if t.is_integer() => extras.push((f.name().clone(), ExtraType::Int)),
_ => unread.push(f.name().clone()),
}
}
(extras, unread)
}
/// What the room projects out of a node event table: [`NODE_READ`], the
/// optional typed columns and the extras — each only if `spec` has it.
pub(crate) fn node_projection<'a>(spec: &PayloadSpec, extras: &'a [(String, ExtraType)]) -> Vec<&'a str> {
let has = |c: &str| spec.schema.field_with_name(c).is_ok();
let mut out: Vec<&str> = NODE_READ.to_vec();
out.extend(OPTIONAL_NODE_COLUMNS.iter().copied().filter(|c| has(c)));
out.extend(extras.iter().map(|(n, _)| n.as_str()).filter(|c| has(c)));
out
}
/// Columns of one node-table batch the room reads.
pub(crate) struct NodeCols {
pub(crate) repo: StringArray,
/// The event's owner (`crate@version:kind`).
pub(crate) owner: StringArray,
pub(crate) id: StringArray,
pub(crate) kind: StringArray,
pub(crate) level: StringArray,
pub(crate) krate: StringArray,
pub(crate) version: StringArray,
pub(crate) origin: StringArray,
pub(crate) defined: BooleanArray,
/// The payload hash (hex): identity of the node's semantic payload,
/// positions excluded.
pub(crate) hash: StringArray,
/// Non-null: the dump repeated this key with another payload.
pub(crate) dup: StringArray,
pub(crate) name: Option<StringArray>,
pub(crate) parent: Option<StringArray>,
pub(crate) cfg: Option<StringArray>,
pub(crate) ty: Option<StringArray>,
pub(crate) is_test: Option<BooleanArray>,
/// Aligned with the extras list the batch was read for.
pub(crate) extras: Vec<ExtraArr>,
}
const NODE_COLS: [&str; 6] = ["kind", "level", "crate", "version", "origin", "defined_here"];
/// What the room reads of a node event: payload columns + its payload hash.
pub(crate) const NODE_READ: [&str; 7] = ["kind", "level", "crate", "version", "origin", "defined_here", "payload_hash"];
/// The first 64 bits of a row's payload hash (0 when absent).
pub(crate) fn hash64(h: &StringArray, r: usize) -> u64 {
if h.is_null(r) {
return 0;
}
let v = h.value(r);
u64::from_str_radix(&v[..v.len().min(16)], 16).unwrap_or(0)
}
pub(crate) fn node_cols(rows: &FoldedRows, extras: &[(String, ExtraType)]) -> Result<Vec<NodeCols>> {
rows.batches
.iter()
.map(|b| {
let ex = extras
.iter()
.map(|(n, t)| {
Ok(match (b.column_by_name(n).is_some(), t) {
(false, _) => ExtraArr::Absent,
(true, ExtraType::Str) => ExtraArr::Str(utf8(b, n)?),
(true, ExtraType::Int) => ExtraArr::Int(int64(b, n)?),
(true, ExtraType::Bool) => ExtraArr::Bool(boolean(b, n)?),
})
})
.collect::<Result<Vec<_>>>()?;
Ok(NodeCols {
dup: utf8(b, "key_dup")?,
name: opt_utf8(b, "name")?,
parent: opt_utf8(b, "parent")?,
cfg: opt_utf8(b, "cfg")?,
ty: opt_utf8(b, "ty")?,
is_test: if b.column_by_name("is_test").is_some() { Some(boolean(b, "is_test")?) } else { None },
extras: ex,
repo: utf8(b, "repo")?,
owner: utf8(b, "krate")?,
id: utf8(b, "id")?,
kind: utf8(b, "kind")?,
level: utf8(b, "level")?,
krate: utf8(b, "crate")?,
version: utf8(b, "version")?,
origin: utf8(b, "origin")?,
defined: boolean(b, "defined_here")?,
hash: utf8(b, "payload_hash")?,
})
})
.collect()
}
pub(crate) struct EdgeCols {
pub(crate) repo: StringArray,
pub(crate) owner: StringArray,
pub(crate) from: StringArray,
pub(crate) to: StringArray,
pub(crate) kind: StringArray,
/// Distinct sites carrying the edge (absent in a schema without it).
pub(crate) count: Option<Int64Array>,
/// GATED_BY only (null elsewhere).
pub(crate) negated: Option<BooleanArray>,
pub(crate) partial: Option<BooleanArray>,
/// TOUCHES_TABLE only (null elsewhere; absent before mimir-schema 4).
pub(crate) access: Option<StringArray>,
}
fn edge_cols(rows: &FoldedRows) -> Result<Vec<EdgeCols>> {
rows.batches
.iter()
.map(|b| {
Ok(EdgeCols {
repo: utf8(b, "repo")?,
owner: utf8(b, "krate")?,
from: utf8(b, "from")?,
to: utf8(b, "to")?,
kind: utf8(b, "kind")?,
count: if b.column_by_name("count").is_some() { Some(int64(b, "count")?) } else { None },
negated: if b.column_by_name("negated").is_some() { Some(boolean(b, "negated")?) } else { None },
partial: if b.column_by_name("partial").is_some() { Some(boolean(b, "partial")?) } else { None },
access: if b.column_by_name("access").is_some() { Some(utf8(b, "access")?) } else { None },
})
})
.collect()
}
struct SymCols {
repo: StringArray,
/// The artifact (`name@triple:profile`) that links the symbol.
owner: StringArray,
id: StringArray,
krate: StringArray,
version: StringArray,
symbol: StringArray,
demangled: Option<StringArray>,
role: Option<StringArray>,
binary: Option<StringArray>,
section: Option<StringArray>,
size: Option<UInt64Array>,
}
/// What the room reads of a symbol event beyond its key (`symbol`, `id`).
const SYMBOL_READ: [&str; 6] = ["crate", "version", "demangled", "role", "binary", "section"];
fn sym_cols(rows: &FoldedRows) -> Result<Vec<SymCols>> {
rows.batches
.iter()
.map(|b| {
// u64 payload columns are stored as decimal(20, 0)
let size = match b.column_by_name("size") {
Some(c) => Some(skade::arrow_cast::cast(c, &DataType::UInt64)?.as_any().downcast_ref::<UInt64Array>().context("size as u64")?.clone()),
None => None,
};
Ok(SymCols {
repo: utf8(b, "repo")?,
owner: utf8(b, "krate")?,
id: utf8(b, "id")?,
krate: utf8(b, "crate")?,
version: utf8(b, "version")?,
symbol: utf8(b, "symbol")?,
demangled: opt_utf8(b, "demangled")?,
role: opt_utf8(b, "role")?,
binary: opt_utf8(b, "binary")?,
section: opt_utf8(b, "section")?,
size,
})
})
.collect()
}
struct SpanCols {
repo: StringArray,
target: StringArray,
key: StringArray,
file: StringArray,
start: Int64Array,
end: Int64Array,
}
const SPAN_COLS: [&str; 5] = ["target", "key", "file", "start_line", "end_line"];
fn span_cols(rows: &FoldedRows) -> Result<Vec<SpanCols>> {
rows.batches
.iter()
.map(|b| {
Ok(SpanCols {
repo: utf8(b, "repo")?,
target: utf8(b, "target")?,
key: utf8(b, "key")?,
file: utf8(b, "file")?,
start: int64(b, "start_line")?,
end: int64(b, "end_line")?,
})
})
.collect()
}
/// The room needs the mimir-schema payloads (not a test-only schema).
fn check_specs(log: &EventLog) -> Result<()> {
for kind in [Kind::Nodes, Kind::BinNodes] {
let s = &log.spec(kind).schema;
for c in NODE_COLS {
if s.field_with_name(c).is_err() {
bail!("{} has no `{c}` column: the room reads the mimir-schema payloads (Specs::mimir())", EventLog::table_name(kind));
}
}
}
for c in ["crate", "version"] {
if log.spec(Kind::Symbols).schema.field_with_name(c).is_err() {
bail!("code_symbol_events has no `{c}` column");
}
}
for c in SPAN_COLS {
if log.spec(Kind::Spans).schema.field_with_name(c).is_err() {
bail!("code_span_events has no `{c}` column: the room reads the mimir-schema span payload");
}
}
Ok(())
}
/// Owners a ring-on scan names at a time (`krate IN (…)`, `id IN (…)`).
const RING_IN_CHUNK: usize = 1 << 16;
/// The ring owners in `fs`'s scope ([`ring::allowed_owners`]) and how many
/// ring owners the log has, worked out from what the closure needs only: the
/// crate nodes, the nodes the binaries link, the DEPENDS_ON edges.
async fn ring_closure(
log: &EventLog,
fs: &FoldScope,
node_proj: &[&str],
edge_read: &[&str],
extras: &[(String, ExtraType)],
linked: &std::collections::HashSet<&str>,
) -> Result<(Vec<String>, usize)> {
let mut nodes = log.scan_rows_where(Kind::Nodes, &ScanFilter::is_in("kind", [mimir_schema::kind::CRATE]), node_proj).await?;
let linked_ids: Vec<&str> = linked.iter().copied().collect();
for ids in linked_ids.chunks(RING_IN_CHUNK) {
let more = log.scan_rows_where(Kind::Nodes, &ScanFilter::is_in("id", ids.iter().copied()), node_proj).await?;
nodes.batches.extend(more.batches);
}
log.fold_scanned(Kind::Nodes, fs, &mut nodes)?;
let mut edges = log.scan_rows_where(Kind::Edges, &ScanFilter::is_in("kind", [EdgeKind::DependsOn.as_str()]), edge_read).await?;
log.fold_scanned(Kind::Edges, fs, &mut edges)?;
let (nc, ec) = (node_cols(&nodes, extras)?, edge_cols(&edges)?);
let (allowed, all) = ring::allowed_owners(&nodes.open, &nc, &edges.open, &ec, linked);
let mut allowed: Vec<String> = allowed.into_iter().map(str::to_string).collect();
allowed.sort();
Ok((allowed, all.len()))
}
// ---- the streamed build's tables and records
/// Strings by id, each kept once: one arena, its ends, a table of ids
/// hashed by the string. What a streamed row leaves of its strings.
#[derive(Default)]
struct StrTable {
bytes: String,
ends: Vec<u64>,
index: hashbrown::HashTable<u32>,
hasher: foldhash::fast::FixedState,
}
impl StrTable {
fn len(&self) -> usize {
self.ends.len()
}
fn get(&self, i: u32) -> &str {
let i = i as usize;
let from = if i == 0 { 0 } else { self.ends[i - 1] as usize };
&self.bytes[from..self.ends[i] as usize]
}
fn id(&mut self, s: &str) -> u32 {
let h = self.hasher.hash_one(s);
let StrTable { bytes, ends, index, hasher } = self;
let at = |i: u32| {
let i = i as usize;
let from = if i == 0 { 0 } else { ends[i - 1] as usize };
&bytes[from..ends[i] as usize]
};
if let Some(&i) = index.find(h, |&i| at(i) == s) {
return i;
}
let id = u32::try_from(ends.len()).expect("a room's strings past 2^32");
bytes.push_str(s);
ends.push(bytes.len() as u64);
let (bytes, ends) = (&*bytes, &*ends);
index.insert_unique(h, id, |&i| {
let i = i as usize;
let from = if i == 0 { 0 } else { ends[i - 1] as usize };
hasher.hash_one(&bytes[from..ends[i] as usize])
});
id
}
fn opt(&mut self, s: Option<&str>) -> u32 {
s.map_or(NONE, |s| self.id(s))
}
}
/// One node-describing row, as `Builder::set` reads it — every string by
/// its id in the build's tables (`raw`; the moniker in `mon`).
struct NodeRec {
hash: u64,
node: u32,
kind: u32,
krate: u32,
version: u32,
repo: u32,
name: u32,
parent: u32,
cfg: u32,
ty: u32,
owner: u32,
flags: u8,
prio: u8,
level: u8,
origin: u8,
test: bool,
key_dup: bool,
}
/// The node records, and their extra payload columns by column.
struct NodeRecs {
recs: Vec<NodeRec>,
ext: Vec<ExtVals>,
}
enum ExtVals {
/// `raw` ids, `NONE` = null.
Str(Vec<u32>),
/// `i64::MIN` = null.
Int(Vec<i64>),
/// 0 / 1, 2 = null.
Bool(Vec<u8>),
}
impl NodeRecs {
fn new(extras: &[(String, ExtraType)]) -> Self {
let ext = extras
.iter()
.map(|(_, t)| match t {
ExtraType::Str => ExtVals::Str(Vec::new()),
ExtraType::Int => ExtVals::Int(Vec::new()),
ExtraType::Bool => ExtVals::Bool(Vec::new()),
})
.collect();
Self { recs: Vec::new(), ext }
}
fn len(&self) -> usize {
self.recs.len()
}
/// A row with the node table's columns: a source node (`binary` false:
/// defined or a stub) or a binary-stream node.
fn push_row(&mut self, c: &NodeCols, r: usize, binary: bool, mon: &mut StrTable, raw: &mut StrTable) -> u32 {
fn opt(a: &StringArray, r: usize) -> Option<&str> {
(!a.is_null(r)).then(|| a.value(r))
}
fn col(x: &Option<StringArray>, r: usize) -> Option<&str> {
x.as_ref().and_then(|a| opt(a, r))
}
if !binary && !keeps_mir_blocks() && opt(&c.kind, r) == Some(mimir_schema::kind::MIR_BLOCK) {
return self.push_dropped();
}
let (flags, prio) = if binary {
(F_BIN_NODE, P_BIN)
} else {
let defined = !c.defined.is_null(r) && c.defined.value(r);
(F_SRC | if defined { F_DEFINED } else { 0 }, if defined { P_DEFINED } else { P_STUB })
};
let node = mon.id(c.id.value(r));
self.recs.push(NodeRec {
hash: hash64(&c.hash, r),
node,
kind: raw.opt(opt(&c.kind, r)),
krate: raw.id(opt(&c.krate, r).unwrap_or("")),
version: raw.id(opt(&c.version, r).unwrap_or("")),
repo: raw.id(c.repo.value(r)),
name: raw.opt(col(&c.name, r)),
parent: raw.opt(col(&c.parent, r)),
cfg: raw.opt(col(&c.cfg, r)),
ty: raw.opt(col(&c.ty, r)),
owner: raw.id(c.owner.value(r)),
flags,
prio,
level: opt(&c.level, r).and_then(Level::parse).map_or(NONE8, |l| Level::ALL.iter().position(|x| *x == l).unwrap_or(0) as u8),
origin: opt(&c.origin, r).and_then(Origin::parse).map_or(NONE8, |o| Origin::ALL.iter().position(|x| *x == o).unwrap_or(0) as u8),
test: c.is_test.as_ref().is_some_and(|t| !t.is_null(r) && t.value(r)),
key_dup: !c.dup.is_null(r),
});
for (x, vals) in self.ext.iter_mut().enumerate() {
let arr = c.extras.get(x);
match vals {
ExtVals::Str(v) => v.push(match arr {
Some(ExtraArr::Str(s)) if !s.is_null(r) => raw.id(s.value(r)),
_ => NONE,
}),
ExtVals::Int(v) => v.push(match arr {
Some(ExtraArr::Int(s)) if !s.is_null(r) => s.value(r),
_ => i64::MIN,
}),
ExtVals::Bool(v) => v.push(match arr {
Some(ExtraArr::Bool(s)) if !s.is_null(r) => s.value(r) as u8,
_ => 2,
}),
}
}
node
}
/// A dropped row's slot ([`DROPPED`]): nothing interned.
fn push_dropped(&mut self) -> u32 {
self.recs.push(NodeRec {
hash: 0,
node: DROPPED,
kind: NONE,
krate: NONE,
version: NONE,
repo: NONE,
name: NONE,
parent: NONE,
cfg: NONE,
ty: NONE,
owner: NONE,
flags: 0,
prio: 0,
level: NONE8,
origin: NONE8,
test: false,
key_dup: false,
});
for vals in &mut self.ext {
match vals {
ExtVals::Str(v) => v.push(NONE),
ExtVals::Int(v) => v.push(i64::MIN),
ExtVals::Bool(v) => v.push(2),
}
}
DROPPED
}
/// A linked symbol: crate, version and repo of the row; a symbol's
/// address and size move with every link, they are not node payload.
fn push_symbol(&mut self, c: &SymCols, r: usize, mon: &mut StrTable, raw: &mut StrTable) -> u32 {
let node = mon.id(c.id.value(r));
self.recs.push(NodeRec {
hash: 0,
node,
kind: NONE,
krate: raw.id(c.krate.value(r)),
version: raw.id(c.version.value(r)),
repo: raw.id(c.repo.value(r)),
name: NONE,
parent: NONE,
cfg: NONE,
ty: NONE,
owner: raw.id(c.owner.value(r)),
flags: F_SYMBOL,
prio: P_SYMBOL,
level: NONE8,
origin: NONE8,
test: false,
key_dup: false,
});
for vals in &mut self.ext {
match vals {
ExtVals::Str(v) => v.push(NONE),
ExtVals::Int(v) => v.push(i64::MIN),
ExtVals::Bool(v) => v.push(2),
}
}
node
}
}
const E_REC_ODD: u8 = 1;
const E_REC_HAS_NEG: u8 = 2;
const E_REC_NEG: u8 = 4;
const E_REC_HAS_PART: u8 = 8;
const E_REC_PART: u8 = 16;
const E_REC_BINARY: u8 = 32;
/// A CFG_SUCC row the room drops (see [`DROPPED`]).
const E_REC_DROPPED: u8 = 64;
const NO_ACCESS: u8 = u8::MAX;
/// One edge row, as the index reads it.
struct EdgeRec {
from: u32,
to: u32,
/// The kind as written (`raw`): parsed once per distinct string.
kind: u32,
repo: u32,
/// The site count, when the row states one other than 1 (`E_REC_ODD`).
count: u32,
bits: u8,
access: u8,
}
impl EdgeRec {
fn of(c: &EdgeCols, r: usize, binary: bool, mon: &mut StrTable, raw: &mut StrTable) -> Self {
if !binary && !keeps_mir_blocks() && c.kind.value(r) == EdgeKind::CfgSucc.as_str() {
// between two dropped block rows: a slot, nothing interned
return EdgeRec { from: DROPPED, to: DROPPED, kind: NONE, repo: NONE, count: 0, bits: E_REC_DROPPED, access: NO_ACCESS };
}
let mut bits = if binary { E_REC_BINARY } else { 0 };
let mut count = 0;
if let Some(n) = c.count.as_ref().filter(|n| !n.is_null(r)).map(|n| n.value(r))
&& n != 1
{
bits |= E_REC_ODD;
count = n.clamp(0, u32::MAX as i64) as u32;
}
let flag = |col: &Option<BooleanArray>| col.as_ref().filter(|a| !a.is_null(r)).map(|a| a.value(r));
match flag(&c.negated) {
Some(true) => bits |= E_REC_HAS_NEG | E_REC_NEG,
Some(false) => bits |= E_REC_HAS_NEG,
None => {}
}
match flag(&c.partial) {
Some(true) => bits |= E_REC_HAS_PART | E_REC_PART,
Some(false) => bits |= E_REC_HAS_PART,
None => {}
}
let access = c.access.as_ref().filter(|a| !a.is_null(r)).and_then(|a| mimir_schema::Access::parse(a.value(r))).map_or(NO_ACCESS, |a| a.bits());
EdgeRec { from: mon.id(c.from.value(r)), to: mon.id(c.to.value(r)), kind: raw.id(c.kind.value(r)), repo: raw.id(c.repo.value(r)), count, bits, access }
}
}
const SPAN_REC_NODE: u8 = 0;
const SPAN_REC_EDGE: u8 = 1;
const SPAN_REC_OTHER: u8 = 2;
/// One span row: the monikers it names (`mon` ids — ordinals do not exist
/// at fold time, [`assemble`] resolves them; `a` = `NONE`: an orphan), its
/// file and lines.
struct SpanRec {
a: u32,
b: u32,
repo: u32,
path: u32,
start: u32,
end: u32,
target: u8,
kind: u8,
}
impl SpanRec {
fn of(c: &SpanCols, r: usize, mon: &mut StrTable, raw: &mut StrTable) -> Self {
let line = |a: &Int64Array| a.value(r).clamp(0, u32::MAX as i64) as u32;
let key = c.key.value(r);
let (target, a, b, kind) = match c.target.value(r) {
SPAN_NODE => (SPAN_REC_NODE, mon.id(key), NONE, 0),
SPAN_EDGE => {
let mut parts = key.split(EDGE_KEY_SEP);
match (parts.next(), parts.next(), parts.next().and_then(EdgeKind::parse)) {
(Some(from), Some(to), Some(k)) => (SPAN_REC_EDGE, mon.id(from), mon.id(to), kind_index(k)),
_ => (SPAN_REC_EDGE, NONE, NONE, 0),
}
}
_ => (SPAN_REC_OTHER, NONE, NONE, 0),
};
SpanRec { a, b, repo: raw.id(c.repo.value(r)), path: raw.id(c.file.value(r)), start: line(&c.start), end: line(&c.end), target, kind }
}
}
/// The scans of one streamed table: the scope's repos, and — ring on — the
/// ring rows of the owners `ring` names (`krate IN (…)`, trimmed at read
/// time to the ring's rows).
fn scope_scans(local: &FoldScope, ring: Option<&[String]>) -> Vec<mimir_ingest::Scan> {
use mimir_ingest::Scan;
use mimir_ingest::schema::{COL_KRATE, COL_REPO};
let mut scans = Vec::new();
if !local.repos.is_empty() {
scans.push(Scan { filter: ScanFilter::is_in(COL_REPO, local.repos.keys().map(String::as_str)), repo: None });
}
if let Some(owners) = ring.filter(|o| !o.is_empty()) {
scans.push(Scan { filter: ScanFilter::is_in(COL_KRATE, owners.iter().map(String::as_str)), repo: Some(UNIVERSE_REPO.to_string()) });
}
scans
}
impl Builder {
/// [`Builder::set`] of a record: the same rule (the highest priority row
/// wins, the first among equals; every row's payload hash counts), its
/// strings interned as they are set, from the build's table `raw`.
fn set_rec(&mut self, i: usize, rec: &NodeRec, recs: &NodeRecs, at: usize, raw: &StrTable, ordinal_of: &dyn Fn(&str) -> Option<u32>) {
// every id below goes through a raw-id → interned-id cache: a distinct
// string is hashed ONCE per interner, not once per record (the fill
// was 5.7 s of repeated interner lookups on dwarf, 2026-10-19)
let owner = self.cache.owners.via(rec.owner, &mut self.owners, raw);
self.phash_rows.push(HashRow { node: i as u32, prio: rec.prio, hash: rec.hash, owner, key_dup: rec.key_dup });
if self.prio[i] >= rec.prio {
return;
}
self.prio[i] = rec.prio;
self.t.kind[i] = self.cache.kinds.via_opt(rec.kind, &mut self.kinds, raw);
self.t.level[i] = rec.level;
self.t.krate[i] = self.cache.crates.via(rec.krate, &mut self.crates, raw);
self.t.version[i] = self.cache.versions.via(rec.version, &mut self.versions, raw);
self.t.origin[i] = rec.origin;
self.t.repo[i] = self.cache.repos.via(rec.repo, &mut self.repos, raw);
self.t.name[i] = self.cache.strs.via_opt(rec.name, &mut self.strs, raw);
self.t.parent[i] = self.cache.parents.ordinal_via(rec.parent, raw, ordinal_of);
self.t.cfg[i] = self.cache.strs.via_opt(rec.cfg, &mut self.strs, raw);
self.t.ty[i] = self.cache.strs.via_opt(rec.ty, &mut self.strs, raw);
self.t.flags[i] = (self.t.flags[i] & !F_TEST) | if rec.test { F_TEST } else { 0 };
for (x, col) in self.t.extras.iter_mut().enumerate() {
match (&mut col.data, recs.ext.get(x)) {
(ExtraData::Str(v), Some(ExtVals::Str(e))) => v[i] = self.cache.strs.via_opt(e[at], &mut self.strs, raw),
(ExtraData::Int(v), Some(ExtVals::Int(e))) => v[i] = e[at],
(ExtraData::Bool(v), Some(ExtVals::Bool(e))) => v[i] = e[at],
(ExtraData::Str(v), _) => v[i] = NONE,
(ExtraData::Int(v), _) => v[i] = i64::MIN,
(ExtraData::Bool(v), _) => v[i] = 2,
}
}
}
}
/// A raw-id → interned-id translation table (lazily filled; `UNFILLED`
/// marks a slot not asked yet). Grown to the raw table's size on first use.
#[derive(Default)]
struct Trans {
map: Vec<u32>,
}
const UNFILLED: u32 = u32::MAX;
/// `ordinal_via`'s "asked, resolves to none".
const RESOLVED_NONE: u32 = u32::MAX - 1;
impl Trans {
fn slot(&mut self, id: u32, raw: &StrTable) -> &mut u32 {
if self.map.len() < raw.len() {
self.map.resize(raw.len(), UNFILLED);
}
&mut self.map[id as usize]
}
/// The interned id of raw `id` (never `NONE` in).
fn via(&mut self, id: u32, interner: &mut Interner, raw: &StrTable) -> u32 {
let slot = self.slot(id, raw);
if *slot == UNFILLED {
*slot = interner.id(raw.get(id));
}
*slot
}
/// Like [`Trans::via`], `NONE` passing through.
fn via_opt(&mut self, id: u32, interner: &mut Interner, raw: &StrTable) -> u32 {
if id == NONE {
return NONE;
}
self.via(id, interner, raw)
}
/// The ordinal the raw moniker `id` resolves to (`NONE` when it is `NONE`
/// or resolves to no room node) — resolved once per distinct moniker.
fn ordinal_via(&mut self, id: u32, raw: &StrTable, ordinal_of: &dyn Fn(&str) -> Option<u32>) -> u32 {
if id == NONE {
return NONE;
}
let slot = self.slot(id, raw);
if *slot == UNFILLED {
*slot = ordinal_of(raw.get(id)).unwrap_or(RESOLVED_NONE);
}
if *slot == RESOLVED_NONE { NONE } else { *slot }
}
}
/// [`Builder::set_rec`]'s translation caches, one per target interner.
#[derive(Default)]
struct TransCache {
kinds: Trans,
crates: Trans,
versions: Trans,
repos: Trans,
strs: Trans,
owners: Trans,
parents: Trans,
}
/// The node columns of one batch (as [`node_cols`] reads a fold's).
fn batch_node_cols(b: &RecordBatch, extras: &[(String, ExtraType)]) -> Result<NodeCols> {
Ok(node_cols(&FoldedRows { batches: vec![b.clone()], ..Default::default() }, extras)?.pop().expect("one batch"))
}
fn batch_edge_cols(b: &RecordBatch) -> Result<EdgeCols> {
Ok(edge_cols(&FoldedRows { batches: vec![b.clone()], ..Default::default() })?.pop().expect("one batch"))
}
fn batch_span_cols(b: &RecordBatch) -> Result<SpanCols> {
Ok(span_cols(&FoldedRows { batches: vec![b.clone()], ..Default::default() })?.pop().expect("one batch"))
}
fn input(kind: Kind, stream: &'static str, rows: &FoldedRows) -> Input {
Input { table: EventLog::table_name(kind).into(), stream: stream.into(), events_read: rows.events_read, open_rows: rows.open.len() as u64 }
}
/// Attribute priority of a row kind (higher wins).
const P_DEFINED: u8 = 4;
const P_BIN: u8 = 3;
const P_STUB: u8 = 2;
const P_SYMBOL: u8 = 1;
struct Attrs<'a> {
kind: Option<&'a str>,
level: Option<&'a str>,
krate: &'a str,
version: &'a str,
origin: Option<&'a str>,
repo: &'a str,
name: Option<&'a str>,
/// The row the rest of the payload is read from (parent, cfg, ty,
/// is_test, extras); `None` for a row kind that has none (a symbol, a
/// ring definition looked up from the ring cache).
row: Option<(&'a NodeCols, usize)>,
}
/// One describing row of a node, for the payload hash and the duplicate report.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
struct HashRow {
node: u32,
prio: u8,
hash: u64,
owner: u32,
key_dup: bool,
}
struct Builder {
/// raw-id → interned-id tables ([`Trans`]): a string hashes once, not once per record.
cache: TransCache,
t: NodeTable,
prio: Vec<u8>,
/// Every describing row.
phash_rows: Vec<HashRow>,
kinds: Interner,
crates: Interner,
versions: Interner,
repos: Interner,
strs: Interner,
owners: Interner,
}
impl Builder {
/// `ordinal_of`: the room ordinal of a moniker (for `parent`).
fn set(&mut self, i: usize, p: u8, a: Attrs<'_>, h: u64, owner: &str, key_dup: bool, ordinal_of: &dyn Fn(&str) -> Option<u32>) {
let owner = self.owners.id(owner);
self.phash_rows.push(HashRow { node: i as u32, prio: p, hash: h, owner, key_dup });
if self.prio[i] >= p {
return;
}
self.prio[i] = p;
self.t.kind[i] = a.kind.map_or(NONE, |k| self.kinds.id(k));
self.t.level[i] = a.level.and_then(Level::parse).map_or(NONE8, |l| Level::ALL.iter().position(|x| *x == l).unwrap_or(0) as u8);
self.t.krate[i] = self.crates.id(a.krate);
self.t.version[i] = self.versions.id(a.version);
self.t.origin[i] = a.origin.and_then(Origin::parse).map_or(NONE8, |o| Origin::ALL.iter().position(|x| *x == o).unwrap_or(0) as u8);
self.t.repo[i] = self.repos.id(a.repo);
self.t.name[i] = a.name.map_or(NONE, |n| self.strs.id(n));
// the rest of the payload: always the winning row's, never a mix
fn opt(col: &Option<StringArray>, r: usize) -> Option<&str> {
col.as_ref().filter(|c| !c.is_null(r)).map(|c| c.value(r))
}
let (mut parent, mut cfg, mut ty, mut test) = (NONE, NONE, NONE, false);
if let Some((c, r)) = a.row {
parent = opt(&c.parent, r).and_then(ordinal_of).unwrap_or(NONE);
cfg = opt(&c.cfg, r).map_or(NONE, |s| self.strs.id(s));
ty = opt(&c.ty, r).map_or(NONE, |s| self.strs.id(s));
test = c.is_test.as_ref().is_some_and(|t| !t.is_null(r) && t.value(r));
}
self.t.parent[i] = parent;
self.t.cfg[i] = cfg;
self.t.ty[i] = ty;
self.t.flags[i] = (self.t.flags[i] & !F_TEST) | if test { F_TEST } else { 0 };
for (x, col) in self.t.extras.iter_mut().enumerate() {
let arr = a.row.and_then(|(c, r)| c.extras.get(x).map(|e| (e, r)));
match &mut col.data {
ExtraData::Str(v) => {
v[i] = match arr {
Some((ExtraArr::Str(s), r)) if !s.is_null(r) => self.strs.id(s.value(r)),
_ => NONE,
}
}
ExtraData::Int(v) => {
v[i] = match arr {
Some((ExtraArr::Int(s), r)) if !s.is_null(r) => s.value(r),
_ => i64::MIN,
}
}
ExtraData::Bool(v) => {
v[i] = match arr {
Some((ExtraArr::Bool(s), r)) if !s.is_null(r) => s.value(r) as u8,
_ => 2,
}
}
}
}
}
/// Per node: the distinct hashes of its top-priority rows, combined —
/// and the ids whose rows disagree or were repeated by a dump.
fn finish_hashes(&mut self) -> Vec<DupKey> {
let mut rows = std::mem::take(&mut self.phash_rows);
rows.sort_unstable();
let mut ph = vec![0u64; self.prio.len()];
let mut dups = Vec::new();
let mut at = 0;
while at < rows.len() {
let node = rows[at].node;
let end = at + rows[at..].iter().take_while(|r| r.node == node).count();
let top = self.prio[node as usize];
let mine: Vec<&HashRow> = rows[at..end].iter().filter(|r| r.prio == top).collect();
let mut hashes: Vec<u64> = mine.iter().map(|r| r.hash).collect();
hashes.dedup();
let acc = &mut ph[node as usize];
for h in &hashes {
*acc = acc.rotate_left(17) ^ h;
}
let key_dup_rows = mine.iter().filter(|r| r.key_dup).count() as u32;
if hashes.len() > 1 || key_dup_rows > 0 {
let mut owners: Vec<String> = mine.iter().map(|r| self.owners.list[r.owner as usize].clone()).collect();
owners.sort();
owners.dedup();
dups.push(DupKey {
node: node as u64,
rows_of: match top {
P_DEFINED => "definition",
P_BIN => "binary",
P_STUB => "stub",
_ => "symbol",
},
rows: mine.len() as u32,
distinct_payloads: hashes.len() as u32,
key_dup_rows,
owners,
});
}
at = end;
}
self.t.phash = ph;
dups
}
}
/// What the lake fold of one scope produced — every open record with its
/// strings interned, no ordinal assigned yet. [`assemble`] packs a
/// [`Snapshot`] from this alone (plus the per-process ring caches): the fold
/// is the one phase that scales with the log's whole history, so an advance
/// will re-apply a delta onto a kept `FoldState` instead of re-reading it.
pub(crate) struct FoldState {
identity: String,
manifest: Option<i64>,
extras: Vec<(String, ExtraType)>,
unread_columns: Vec<String>,
bnodes: FoldedRows,
bedges: FoldedRows,
syms: FoldedRows,
mon: StrTable,
raw: StrTable,
nrecs: NodeRecs,
erecs: Vec<EdgeRec>,
srecs: Vec<SpanRec>,
node_open: Vec<u32>,
edge_open: Vec<u32>,
span_open: Vec<u32>,
node_events: u64,
edge_events: u64,
span_events: u64,
src_recs: u32,
src_edges: u32,
/// Node records past the source block that are the binary stream's
/// (bin nodes + linked symbols): `src_recs..bin_recs_until`. A delta's
/// records land after.
bin_recs_until: u32,
/// Edge records past the source block that are the binary stream's.
bin_edges_until: u32,
sym_mon: Vec<u32>,
ring_owners: Option<(usize, usize)>,
/// Ring on: the sorted owners the ring scan read (`krate IN (…)`). An
/// advance stands only while the closure recomputed at the new head seqs
/// equals this list — a grown closure means rows the kept scan never
/// read belong now, a shrunk one that kept rows no longer do.
ring_allowed: Option<Vec<String>>,
/// The log's stamp at fold (or last advance) time: with the identity and
/// the scope it keys the lake state this fold is 1:1 with.
stamp: mimir_ingest::Stamp,
/// `Some` = the fold was TRACKED (event tables scanned whole, no zone):
/// [`advance`] can re-apply the lake's delta onto it.
tracks: Option<FoldTracks>,
started: Instant,
fold_ms: u64,
}
/// The identity maps of a tracked fold, per stream, and the log's stamp at
/// fold time (the binary components gate the advance: an unchanged binary
/// stream reuses its kept rows wholesale).
pub(crate) struct FoldTracks {
nodes: mimir_ingest::FoldTrack,
edges: mimir_ingest::FoldTrack,
spans: mimir_ingest::FoldTrack,
stamp: mimir_ingest::Stamp,
}
/// The key of ONE lake state under ONE scope: FNV-1a over the log identity,
/// the scope's repos+seqs+universe and the event tables' snapshot ids —
/// deterministic across processes and runs, which the build counter is not
/// (a fresh process restarts it at 1 and would read an older state's rows).
fn state_key_of(identity: &str, scope: &ResolvedScope, stamp: &mimir_ingest::Stamp) -> String {
fn eat(h: &mut u64, bytes: &[u8]) {
for &b in bytes {
*h = (*h ^ b as u64).wrapping_mul(0x100_0000_01b3);
}
}
let mut h: u64 = 0xcbf2_9ce4_8422_2325;
eat(&mut h, identity.as_bytes());
eat(&mut h, &[0x1f]);
for (r, n) in &scope.repos {
eat(&mut h, r.as_bytes());
eat(&mut h, &n.to_le_bytes());
eat(&mut h, &[0x1e]);
}
eat(&mut h, &[scope.universe as u8, 0x1f]);
for slot in stamp.0 {
match slot {
Some(id) => eat(&mut h, &id.to_le_bytes()),
None => eat(&mut h, &[0xff]),
}
eat(&mut h, &[0x1e]);
}
format!("{h:016x}")
}
/// The binary components of a stamp ([`Kind::ALL`] order: BinNodes,
/// BinEdges, Symbols sit at 3..6).
fn bin_stamp(s: &mimir_ingest::Stamp) -> [Option<i64>; 3] {
[s.0[3], s.0[4], s.0[5]]
}
/// `NORNIR_LENS_PROFILE=1`: the process's resident memory now and its PEAK
/// since the last note (VmHWM, reset after each read via clear_refs), per
/// phase of a fold — where a build's memory goes. Diagnosis only.
pub(crate) fn rss_phase(universe: bool, phase: &str) {
static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
if !*ON.get_or_init(|| std::env::var("NORNIR_LENS_PROFILE").is_ok_and(|v| v == "1")) {
return;
}
let s = std::fs::read_to_string("/proc/self/status").unwrap_or_default();
let kb = |k: &str| s.lines().find_map(|l| l.strip_prefix(k)).and_then(|v| v.split_whitespace().next()).and_then(|v| v.parse::<f64>().ok()).unwrap_or(0.0) / 1048576.0;
eprintln!("mimir-room: {} fold {phase}: rss {:.2} GB, peak since last note {:.2} GB", if universe { "ring" } else { "room" }, kb("VmRSS:"), kb("VmHWM:"));
let _ = std::fs::write("/proc/self/clear_refs", "5");
}
/// Fold `scope` out of `log` and build the snapshot.
pub(crate) async fn build(log: &EventLog, scope: ResolvedScope) -> Result<Snapshot> {
let st = fold(log, &scope, false).await?;
Ok(assemble(log, scope, st, false).await?.0)
}
/// [`build`], KEEPING the fold when the scope can advance (AT THE LOG'S
/// HEAD — only then does an unchanged binary stamp guarantee the kept
/// binary rows still hold: a fold behind the head may have seq-bounded away
/// rows that already existed). Ring on keeps the fold too: [`advance`]
/// gates its carry on an unchanged `universe_manifest` snapshot and an
/// unchanged dependency closure. The caller holds the [`FoldState`] beside
/// the snapshot and hands it to [`advance`] when the log moves, instead of
/// refolding the whole history.
pub(crate) async fn build_kept(log: &EventLog, scope: ResolvedScope, at_head: bool) -> Result<(Snapshot, Option<FoldState>)> {
let tracked = at_head;
let st = fold(log, &scope, tracked).await?;
assemble(log, scope, st, tracked).await
}
/// **Advance a kept fold by the lake's delta.** `true` = `st` is now the
/// log's current state for `scope` and [`assemble`] packs its snapshot;
/// `false` = a delta cannot stand in for a fresh fold (an untracked fold, a
/// moved binary stream, a payload-spec change, a lake window holding a
/// delete or a rewrite the delta cannot skip, or — ring on — a moved
/// `universe_manifest` snapshot or a changed dependency closure) — the
/// caller folds afresh and DROPS `st`, which a refused advance may have
/// half-applied. A lake COMPACTION between advances does NOT refuse: its
/// stamped row-preserving replace changes no row the track holds, so the
/// delta skips it ([`skade::COMPACTION_PROP`]) and the fold survives.
pub(crate) async fn advance(log: &EventLog, scope: &ResolvedScope, st: &mut FoldState) -> Result<bool> {
check_specs(log)?;
// a ring refusal is NAMED (live diagnosis: a silent refusal and a lost
// fold both end in the same minutes-long refold)
let refuse = |why: &str| {
if scope.universe {
eprintln!("mimir-room: ring advance refused: {why} — the full refold follows");
}
};
// the payload spec the kept records were read under must still stand
let (extras_now, mut unread_now) = extra_columns(log.spec(Kind::Nodes));
for c in extra_columns(log.spec(Kind::BinNodes)).1 {
if !unread_now.contains(&c) {
unread_now.push(c);
}
}
if extras_now != st.extras || unread_now != st.unread_columns {
refuse("the node payload spec changed");
return Ok(false);
}
let stamp = log.stamp().await?;
let manifest = ring::manifest_snapshot(log).await?;
// the snapshot assembled from this advance times ITS build, not the
// original fold's: the delta read is its fold phase
let t_delta = Instant::now();
st.started = t_delta;
let FoldState {
extras,
bnodes,
bedges,
syms,
mon,
raw,
nrecs,
erecs,
srecs,
node_open,
edge_open,
span_open,
node_events,
edge_events,
span_events,
ring_allowed,
manifest: kept_manifest,
stamp: kept_stamp,
tracks,
fold_ms,
..
} = st;
let Some(tracks) = tracks.as_mut() else {
refuse("the kept fold is untracked");
return Ok(false);
};
// the binary stream: only an unchanged one reuses its kept rows
if bin_stamp(&stamp) != bin_stamp(&tracks.stamp) {
refuse("the binary stream moved");
return Ok(false);
}
let bin_fs = FoldScope { repos: scope.repos.clone(), universe: false };
let node_proj = node_projection(log.spec(Kind::Nodes), extras);
let edge_read: Vec<&str> = ["count", "negated", "partial", "access"].into_iter().filter(|c| log.spec(Kind::Edges).schema.field_with_name(c).is_ok()).collect();
let scans = scope_scans(&bin_fs, None);
// the dependency universe half (ring on): the kept ring rows and their
// lake-side `krate IN owners` trim stand only while (a) the ring tables
// did not move — every ring write moves the `universe_manifest` snapshot
// in the same atomic release — and (b) the closure recomputed at the NEW
// head seqs names exactly the owners the kept scan read. A moved
// manifest (a new toolchain / dependency snapshot) or a changed closure
// (a DEPENDS_ON edge pulled a dumped crate in, or dropped one out)
// refuses the carry: the event-level delta cannot re-apply the trim.
// With both gates green the ring contributes NOTHING to the delta, so
// the streams below read the local repos only (`bin_fs`, no ring scan).
if scope.universe {
if manifest != *kept_manifest {
refuse(&format!("the universe manifest moved ({kept_manifest:?} -> {manifest:?}): the dependency universe itself changed"));
return Ok(false);
}
let Some(kept_allowed) = ring_allowed.as_ref() else {
refuse("the kept fold holds no ring owner list");
return Ok(false);
};
let (bnc, bec, sc) = (node_cols(bnodes, extras)?, edge_cols(bedges)?, sym_cols(syms)?);
let mut linked: std::collections::HashSet<&str> = std::collections::HashSet::new();
linked.extend(syms.open.iter().map(|&(b, r)| sc[b as usize].id.value(r as usize)));
linked.extend(bnodes.open.iter().map(|&(b, r)| bnc[b as usize].id.value(r as usize)));
linked.extend(bedges.open.iter().map(|&(b, r)| bec[b as usize].to.value(r as usize)));
let (allowed, _) = ring_closure(log, &scope.fold(), &node_proj, &edge_read, extras, &linked).await?;
if &allowed != kept_allowed {
refuse(&format!("the dependency closure changed ({} -> {} owners)", kept_allowed.len(), allowed.len()));
return Ok(false);
}
}
let d = log
.fold_stream_delta(Kind::Nodes, &bin_fs, &scans, &node_proj, &mut tracks.nodes, |b, rows| {
let cols = batch_node_cols(b, extras)?;
for &r in rows {
nrecs.push_row(&cols, r as usize, false, mon, raw);
}
Ok(())
})
.await?;
let FoldDelta::Advanced { open, events } = d else {
refuse("the node stream's window holds a commit a delta cannot express");
return Ok(false);
};
*node_open = open;
*node_events += events;
let d = log
.fold_stream_delta(Kind::Edges, &bin_fs, &scans, &edge_read, &mut tracks.edges, |b, rows| {
let c = batch_edge_cols(b)?;
for &r in rows {
erecs.push(EdgeRec::of(&c, r as usize, false, mon, raw));
}
Ok(())
})
.await?;
let FoldDelta::Advanced { open, events } = d else {
refuse("the edge stream's window holds a commit a delta cannot express");
return Ok(false);
};
*edge_open = open;
*edge_events += events;
let d = log
.fold_stream_delta(Kind::Spans, &bin_fs, &scans, &[], &mut tracks.spans, |b, rows| {
let c = batch_span_cols(b)?;
for &r in rows {
srecs.push(SpanRec::of(&c, r as usize, mon, raw));
}
Ok(())
})
.await?;
let FoldDelta::Advanced { open, events } = d else {
refuse("the span stream's window holds a commit a delta cannot express");
return Ok(false);
};
*span_open = open;
*span_events += events;
// replayed windows regrow the records (a stamped replace's merged files
// mix in rows the fold already holds; each re-seen row re-takes its slot
// and leaves its old record as garbage): past twice the open set the
// kept fold refreshes itself — the refusal here is the periodic fresh
// fold that reclaims it
let bloated = nrecs.len() > 2 * node_open.len() + (1 << 16)
|| erecs.len() > 2 * edge_open.len() + (1 << 16)
|| srecs.len() > 2 * span_open.len() + (1 << 16);
if bloated {
refuse("the kept fold's records grew past twice its open rows (replayed windows): refreshing");
return Ok(false);
}
*kept_manifest = manifest;
*kept_stamp = stamp;
tracks.stamp = stamp;
*fold_ms = t_delta.elapsed().as_millis() as u64;
Ok(true)
}
/// The lake half: every open row of the scope folded to records
/// ([`FoldState`]). `tracked` keeps the identity maps so [`advance`] can
/// re-apply a lake delta later (event tables scanned whole — no zone).
async fn fold(log: &EventLog, scope: &ResolvedScope, tracked: bool) -> Result<FoldState> {
check_specs(log)?;
let t0 = Instant::now();
rss_phase(scope.universe, "start");
let stamp = log.stamp().await?;
let identity = log.identity().await?;
let manifest = ring::manifest_snapshot(log).await?;
let fs = scope.fold();
// the binary stream has no universe ring
let bin_fs = FoldScope { repos: scope.repos.clone(), universe: false };
let (extras, mut unread_columns) = extra_columns(log.spec(Kind::Nodes));
for c in extra_columns(log.spec(Kind::BinNodes)).1 {
if !unread_columns.contains(&c) {
unread_columns.push(c);
}
}
// the binary stream first: what it links is part of the ring's closure
let bnodes = log.fold_rows(Kind::BinNodes, &bin_fs, &node_projection(log.spec(Kind::BinNodes), &extras)).await?;
let bedges = log.fold_rows(Kind::BinEdges, &bin_fs, &[]).await?;
let sym_read: Vec<&str> = SYMBOL_READ.iter().copied().chain(["size"]).filter(|c| log.spec(Kind::Symbols).schema.field_with_name(c).is_ok()).collect();
let syms = log.fold_rows(Kind::Symbols, &bin_fs, &sym_read).await?;
rss_phase(scope.universe, "binary stream");
let (bnc, bec, sc) = (node_cols(&bnodes, &extras)?, edge_cols(&bedges)?, sym_cols(&syms)?);
let node_proj = node_projection(log.spec(Kind::Nodes), &extras);
let edge_read: Vec<&str> = ["count", "negated", "partial", "access"].into_iter().filter(|c| log.spec(Kind::Edges).schema.field_with_name(c).is_ok()).collect();
// ---- the ring belongs to the scope only as far as its local crates
// depend on it (DEPENDS_ON closure at their seqs) or its binaries link
// it: time travel with the ring on never shows a crate first needed later.
// The closure is worked out from the crate nodes, the DEPENDS_ON edges and
// the linked items alone, and only the ring rows of those owners are read:
// the whole ring read and trimmed after was 15.7 M nodes / 37.7 M edges
// for dwarf's 21 repos (a 22 GB peak) where the room keeps 6.3 M nodes.
let mut ring_owners = None;
let ring_scan = if scope.universe {
// what the binary stream links (symbols, its own nodes, INSTANCE_OF targets)
let mut linked: std::collections::HashSet<&str> = std::collections::HashSet::new();
linked.extend(syms.open.iter().map(|&(b, r)| sc[b as usize].id.value(r as usize)));
linked.extend(bnodes.open.iter().map(|&(b, r)| bnc[b as usize].id.value(r as usize)));
linked.extend(bedges.open.iter().map(|&(b, r)| bec[b as usize].to.value(r as usize)));
let (allowed, all) = ring_closure(log, &fs, &node_proj, &edge_read, &extras, &linked).await?;
rss_phase(scope.universe, "ring closure");
ring_owners = Some((allowed.len(), all));
Some(allowed)
} else {
None
};
let local_fs = FoldScope { repos: scope.repos.clone(), universe: false };
// ---- the index, built as the rows stream by. Nothing of a scan is kept:
// a row leaves its strings in two tables (`mon`: the monikers, `raw`:
// every other string) and a fixed-size record; the batch goes. The
// monikers become the room's names once every row is in; the records
// are applied to the node table then, in the order the rows came — what
// `Builder::set` did over the kept scans.
let mut mon = StrTable::default();
let mut raw = StrTable::default();
let mut nrecs = NodeRecs::new(&extras);
let node_scans = scope_scans(&local_fs, ring_scan.as_deref());
let mut emit_nodes = |b: &RecordBatch, rows: &[u32]| {
let cols = batch_node_cols(b, &extras)?;
for &r in rows {
nrecs.push_row(&cols, r as usize, false, &mut mon, &mut raw);
}
Ok(())
};
let (node_open, node_events, mut node_track) = if tracked {
let (o, e, t) = log.fold_stream_tracked(Kind::Nodes, &fs, &node_scans, &node_proj, &mut emit_nodes).await?;
(o, e, Some(t))
} else {
let (o, e) = log.fold_stream(Kind::Nodes, &fs, &node_scans, &node_proj, &mut emit_nodes).await?;
(o, e, None)
};
drop(emit_nodes);
rss_phase(scope.universe, "nodes");
// the binary stream's nodes and the linked symbols: every open row
let src_recs = nrecs.len() as u32;
for &(b, r) in &bnodes.open {
nrecs.push_row(&bnc[b as usize], r as usize, true, &mut mon, &mut raw);
}
let sym_mon: Vec<u32> = syms.open.iter().map(|&(b, r)| nrecs.push_symbol(&sc[b as usize], r as usize, &mut mon, &mut raw)).collect();
let bin_recs_until = nrecs.len() as u32;
if let Some(t) = node_track.as_mut() {
t.skip_to(bin_recs_until);
}
let mut erecs: Vec<EdgeRec> = Vec::new();
let mut emit_edges = |b: &RecordBatch, rows: &[u32]| {
let c = batch_edge_cols(b)?;
for &r in rows {
erecs.push(EdgeRec::of(&c, r as usize, false, &mut mon, &mut raw));
}
Ok(())
};
let (edge_open, edge_events, mut edge_track) = if tracked {
let (o, e, t) = log.fold_stream_tracked(Kind::Edges, &fs, &node_scans, &edge_read, &mut emit_edges).await?;
(o, e, Some(t))
} else {
let (o, e) = log.fold_stream(Kind::Edges, &fs, &node_scans, &edge_read, &mut emit_edges).await?;
(o, e, None)
};
drop(emit_edges);
rss_phase(scope.universe, "edges");
let src_edges = erecs.len() as u32;
for &(b, r) in &bedges.open {
erecs.push(EdgeRec::of(&bec[b as usize], r as usize, true, &mut mon, &mut raw));
}
let bin_edges_until = erecs.len() as u32;
if let Some(t) = edge_track.as_mut() {
t.skip_to(bin_edges_until);
}
// the span rows too (the scope's repos only — the ring's spans are not
// folded: dependency sources are not this room's files, and they are
// most of the log); a row leaves its monikers, its file and its lines
let mut srecs: Vec<SpanRec> = Vec::new();
let span_scans = scope_scans(&bin_fs, None);
let mut emit_spans = |b: &RecordBatch, rows: &[u32]| {
let c = batch_span_cols(b)?;
for &r in rows {
srecs.push(SpanRec::of(&c, r as usize, &mut mon, &mut raw));
}
Ok(())
};
let (span_open, span_events, span_track) = if tracked {
let (o, e, t) = log.fold_stream_tracked(Kind::Spans, &bin_fs, &span_scans, &[], &mut emit_spans).await?;
(o, e, Some(t))
} else {
let (o, e) = log.fold_stream(Kind::Spans, &bin_fs, &span_scans, &[], &mut emit_spans).await?;
(o, e, None)
};
drop(emit_spans);
rss_phase(scope.universe, "spans");
let tracks = match (node_track, edge_track, span_track) {
(Some(nodes), Some(edges), Some(spans)) => Some(FoldTracks { nodes, edges, spans, stamp }),
_ => None,
};
let fold_ms = t0.elapsed().as_millis() as u64;
Ok(FoldState {
identity,
manifest,
extras,
unread_columns,
bnodes,
bedges,
syms,
mon,
raw,
nrecs,
erecs,
srecs,
node_open,
edge_open,
span_open,
node_events,
edge_events,
span_events,
src_recs,
src_edges,
bin_recs_until,
bin_edges_until,
sym_mon,
ring_owners,
ring_allowed: ring_scan,
stamp,
tracks,
started: t0,
fold_ms,
})
}
/// The CPU half: ordinals, the node table, matrices, spans, provenance —
/// a [`Snapshot`] packed from a [`FoldState`] without touching the lake
/// (the ring definitions come from the per-process ring cache). `keep`
/// hands the fold back for a later [`advance`]; without it each piece is
/// dropped as soon as the pack is done with it (the build-peak law: the
/// index keeps no scanned row).
pub(crate) async fn assemble(log: &EventLog, scope: ResolvedScope, st: FoldState, keep: bool) -> Result<(Snapshot, Option<FoldState>)> {
let FoldState {
identity,
manifest,
extras,
unread_columns,
bnodes,
bedges,
syms,
mon,
raw,
nrecs,
erecs,
srecs,
node_open,
edge_open,
span_open,
node_events,
edge_events,
span_events,
src_recs,
src_edges,
bin_recs_until,
bin_edges_until,
sym_mon,
ring_owners,
ring_allowed,
stamp,
tracks,
started,
fold_ms,
} = st;
let ring_closure = ring_allowed.as_ref().map(|owners| {
let mut h: u64 = 0xcbf2_9ce4_8422_2325;
for o in owners.iter() {
for &b in o.as_bytes().iter().chain([0x1e].iter()) {
h = (h ^ b as u64).wrapping_mul(0x100_0000_01b3);
}
}
h
});
let (mut mon, mut raw) = (Some(mon), Some(raw));
let (mut nrecs, mut erecs, mut srecs) = (Some(nrecs), Some(erecs), Some(srecs));
let (mut bnodes, mut bedges, mut syms) = (Some(bnodes), Some(bedges), Some(syms));
let (mut node_open, mut edge_open, mut span_open, mut sym_mon) = (Some(node_open), Some(edge_open), Some(span_open), Some(sym_mon));
let sc = sym_cols(syms.as_ref().expect("kept"))?;
let t1 = Instant::now();
let mut inputs = vec![
Input { table: EventLog::table_name(Kind::Nodes).into(), stream: "source".into(), events_read: node_events, open_rows: node_open.as_ref().expect("kept").len() as u64 },
Input { table: EventLog::table_name(Kind::Edges).into(), stream: "source".into(), events_read: edge_events, open_rows: edge_open.as_ref().expect("kept").len() as u64 },
input(Kind::BinNodes, "binary", bnodes.as_ref().expect("kept")),
input(Kind::BinEdges, "binary", bedges.as_ref().expect("kept")),
input(Kind::Symbols, "binary", syms.as_ref().expect("kept")),
];
// the records that count, in the order their rows came
let node_order: Vec<u32> = {
let recs = &nrecs.as_ref().expect("kept").recs;
node_open.as_ref().expect("kept").iter().copied().chain(src_recs..bin_recs_until).filter(|&k| recs[k as usize].node != DROPPED).collect()
};
let edge_order: Vec<u32> = {
let recs = erecs.as_ref().expect("kept");
edge_open.as_ref().expect("kept").iter().copied().chain(src_edges..bin_edges_until).filter(|&k| recs[k as usize].bits & E_REC_DROPPED == 0).collect()
};
if !keep {
node_open = None;
edge_open = None;
}
// ---- the room's names: the monikers an open row names, sorted
const UNUSED: u32 = u32::MAX;
let mon_r = mon.as_ref().expect("kept");
let raw_r = raw.as_ref().expect("kept");
let nrecs_r = nrecs.as_ref().expect("kept");
let erecs_r = erecs.as_ref().expect("kept");
let mut rank = vec![UNUSED; mon_r.len()];
for &k in &node_order {
rank[nrecs_r.recs[k as usize].node as usize] = 0;
}
let n_with_rows = rank.iter().filter(|r| **r == 0).count();
let mut unknown_kinds: BTreeMap<String, u64> = BTreeMap::new();
// an edge's kind, parsed once per distinct string
let mut edge_kind: FastMap<u32, Option<EdgeKind>> = FastMap::default();
let kind_of = |raw: &StrTable, cache: &mut FastMap<u32, Option<EdgeKind>>, id: u32| -> Option<EdgeKind> { *cache.entry(id).or_insert_with(|| EdgeKind::parse(raw.get(id))) };
for &k in &edge_order {
let e = &erecs_r[k as usize];
if kind_of(raw_r, &mut edge_kind, e.kind).is_some() {
rank[e.from as usize] = 0;
rank[e.to as usize] = 0;
}
}
let mut order: Vec<u32> = (0..mon_r.len() as u32).filter(|&i| rank[i as usize] == 0).collect();
// sorted by an 8-byte big-endian prefix first (one u64 compare settles
// most pairs), the full string only on a prefix tie — EXACTLY the
// lexicographic order the plain compare gives (byte order == BE u64
// order on the prefix; equal prefixes fall through to the full bytes)
let prefix = |m: u32| -> u64 {
let b = mon_r.get(m).as_bytes();
let mut p = [0u8; 8];
let n = b.len().min(8);
p[..n].copy_from_slice(&b[..n]);
u64::from_be_bytes(p)
};
let mut keyed: Vec<(u64, u32)> = order.iter().map(|&m| (prefix(m), m)).collect();
keyed.sort_unstable_by(|a, b| a.0.cmp(&b.0).then_with(|| mon_r.get(a.1).cmp(mon_r.get(b.1))));
for (i, (_, m)) in keyed.iter().enumerate() {
order[i] = *m;
}
drop(keyed);
let n = order.len();
if n as u64 >= u32::MAX as u64 {
bail!("{n} monikers: the room packs ordinals in 32 bits");
}
for (pos, &m) in order.iter().enumerate() {
rank[m as usize] = pos as u32;
}
let kept_names = Monikers::from_sorted(order.iter().map(|&m| mon_r.get(m)), n);
drop(order);
let sort_ms = t1.elapsed().as_millis() as u64;
rss_phase(scope.universe, "index sort");
if !keep {
mon = None;
}
// moniker → ordinal by the moniker's 64-bit hash (12 B a name, not a
// copy of the names); a hash two names share falls back to the sorted
// names' binary search
let hasher = foldhash::fast::FixedState::with_seed(0x6d69_6d69);
let mut by_hash: FastMap<u64, u32> = FastMap::with_capacity_and_hasher(n, Default::default());
for i in 0..n as u32 {
by_hash.entry(hasher.hash_one(kept_names.get(i as u64))).and_modify(|o| *o = u32::MAX).or_insert(i);
}
let ordinal_of = |m: &str| -> Option<u32> {
match by_hash.get(&hasher.hash_one(m)) {
Some(&o) if o != u32::MAX => (kept_names.get(o as u64) == m).then_some(o),
Some(_) => kept_names.find(m).map(|o| o as u32),
None => None,
}
};
let mut bld = Builder {
t: NodeTable {
names: Monikers::default(),
kind: vec![NONE; n],
krate: vec![NONE; n],
version: vec![NONE; n],
repo: vec![NONE; n],
origin: vec![NONE8; n],
level: vec![NONE8; n],
flags: vec![0; n],
prov: vec![0; n],
name: vec![NONE; n],
parent: vec![NONE; n],
cfg: vec![NONE; n],
ty: vec![NONE; n],
span_file: vec![NONE; n],
span_start: vec![0; n],
span_end: vec![0; n],
extras: extras
.iter()
.map(|(name, t)| ExtraCol {
name: name.clone(),
data: match t {
ExtraType::Str => ExtraData::Str(vec![NONE; n]),
ExtraType::Int => ExtraData::Int(vec![i64::MIN; n]),
ExtraType::Bool => ExtraData::Bool(vec![2; n]),
},
})
.collect(),
..Default::default()
},
prio: vec![0; n],
phash_rows: Vec::with_capacity(node_order.len()),
cache: TransCache::default(),
kinds: Interner::default(),
crates: Interner::default(),
versions: Interner::default(),
repos: Interner::default(),
strs: Interner::default(),
owners: Interner::default(),
};
let mut repo_rows: BTreeMap<String, RepoRows> = scope.repos.keys().map(|r| (r.clone(), RepoRows::default())).collect();
let mut universe_rows = 0u64;
let mut count = |repo: &str, binary: bool| {
if repo == UNIVERSE_REPO {
universe_rows += 1;
} else if let Some(rr) = repo_rows.get_mut(repo) {
if binary { rr.binary += 1 } else { rr.source += 1 }
}
};
for &k in &node_order {
let rec = &nrecs_r.recs[k as usize];
let i = rank[rec.node as usize] as usize;
bld.t.flags[i] |= rec.flags;
bld.set_rec(i, rec, nrecs_r, k as usize, raw_r, &ordinal_of);
count(raw_r.get(rec.repo), rec.flags & (F_BIN_NODE | F_SYMBOL) != 0);
}
drop(node_order);
let mut by_kind: BTreeMap<EdgeKind, Vec<(u32, u32)>> = BTreeMap::new();
// (kind, from, to, sites) of the edges whose site count is not 1
let mut odd_counts: Vec<(EdgeKind, u32, u32, u32)> = Vec::new();
// (kind, from, to, flags) of every row that states a flag (true or false)
let mut flag_rows: Vec<(EdgeKind, u32, u32, u8)> = Vec::new();
// (kind, from, to, access bits) of every row that states an access
let mut access_rows: Vec<(EdgeKind, u32, u32, u8)> = Vec::new();
for &k in &edge_order {
let e = &erecs_r[k as usize];
count(raw_r.get(e.repo), e.bits & E_REC_BINARY != 0);
let Some(kind) = kind_of(raw_r, &mut edge_kind, e.kind) else {
*unknown_kinds.entry(raw_r.get(e.kind).to_string()).or_default() += 1;
continue;
};
let (fi, ti) = (rank[e.from as usize], rank[e.to as usize]);
by_kind.entry(kind).or_default().push((fi, ti));
if e.bits & E_REC_ODD != 0 {
odd_counts.push((kind, fi, ti, e.count));
}
if e.bits & (E_REC_HAS_NEG | E_REC_HAS_PART) != 0 {
flag_rows.push((kind, fi, ti, if e.bits & E_REC_NEG != 0 { E_NEGATED } else { 0 } | if e.bits & E_REC_PART != 0 { E_PARTIAL } else { 0 }));
}
if e.access != NO_ACCESS {
access_rows.push((kind, fi, ti, e.access));
}
if e.bits & E_REC_BINARY != 0 && kind == EdgeKind::InstanceOf {
bld.t.flags[ti as usize] |= F_INSTANCES;
}
if raw_r.get(e.repo) != UNIVERSE_REPO {
bld.t.flags[fi as usize] |= F_LOCAL_OUT;
}
}
drop(edge_order);
if !keep {
nrecs = None;
erecs = None;
}
let endpoint_only = (n - n_with_rows) as u64;
let fill_ms = t1.elapsed().as_millis() as u64 - sort_ms;
rss_phase(scope.universe, "index fill");
// ---- the linked symbols, by node
let mut symbols = SymbolTable::default();
{
let syms_r = syms.as_ref().expect("kept");
let sym_mon_r = sym_mon.as_ref().expect("kept");
let mut st = Interner::default();
let mut order: Vec<(u32, &str, usize)> =
syms_r.open.iter().enumerate().map(|(k, &(b, r))| (rank[sym_mon_r[k] as usize], sc[b as usize].symbol.value(r as usize), k)).collect();
order.sort_unstable();
for (node, name, k) in order {
let (b, r) = syms_r.open[k];
let (c, r) = (&sc[b as usize], r as usize);
let mut s = |col: &Option<StringArray>| col.as_ref().filter(|a| !a.is_null(r)).map_or(NONE, |a| st.id(a.value(r)));
symbols.node.push(node);
symbols.symbol.push(name);
symbols.role.push(s(&c.role));
symbols.binary.push(s(&c.binary));
symbols.section.push(s(&c.section));
symbols.artifact.push(st.id(c.owner.value(r)));
symbols.demangled.push(c.demangled.as_ref().filter(|a| !a.is_null(r)).map_or("", |a| a.value(r)));
symbols.size.push(c.size.as_ref().filter(|a| !a.is_null(r)).map_or(u64::MAX, |a| a.value(r)));
}
symbols.strs = st.list;
}
let index_ms = t1.elapsed().as_millis() as u64;
let symbols_ms = index_ms - fill_ms - sort_ms;
drop(sc);
if !keep {
sym_mon = None;
bnodes = None;
bedges = None;
syms = None;
}
// ---- positions: node spans and edge sites, from the fold's span
// records — their monikers resolve to ordinals only now (`rank`; an
// unresolved one is an orphan).
let t_span = Instant::now();
inputs.push(Input { table: EventLog::table_name(Kind::Spans).into(), stream: "source".into(), events_read: span_events, open_rows: span_open.as_ref().expect("kept").len() as u64 });
let ord = |m: u32| if m == NONE { NONE } else { rank[m as usize] };
let mut file_ids: FastMap<(u32, u32), u32> = FastMap::default();
let mut file_list: Vec<(u32, u32)> = Vec::new();
let mut node_spans: Vec<(u32, u32, u32, u32)> = Vec::new();
let mut sites: Vec<Site> = Vec::new();
let mut orphan_spans = 0u64;
{
let srecs_r = srecs.as_ref().expect("kept");
for &k in span_open.as_ref().expect("kept") {
let s = &srecs_r[k as usize];
let repo = bld.repos.id(raw_r.get(s.repo));
let file = *file_ids.entry((repo, s.path)).or_insert_with(|| {
file_list.push((repo, s.path));
file_list.len() as u32 - 1
});
let (a, b) = (ord(s.a), ord(s.b));
match s.target {
SPAN_REC_NODE if a != NONE => node_spans.push((a, file, s.start, s.end)),
SPAN_REC_EDGE if a != NONE && b != NONE => sites.push(Site { kind: s.kind, from: a, to: b, file, line: s.start }),
_ => orphan_spans += 1,
}
}
}
drop(rank);
// canonical file ids: sorted by (repo, path)
let mut files: Vec<(FileEntry, u32)> =
file_list.iter().enumerate().map(|(i, (repo, path))| (FileEntry { repo: bld.repos.list[*repo as usize].clone(), path: raw_r.get(*path).to_string() }, i as u32)).collect();
files.sort();
if !keep {
srecs = None;
span_open = None;
raw = None;
}
let mut file_rank = vec![0u32; files.len()];
for (pos, (_, old)) in files.iter().enumerate() {
file_rank[*old as usize] = pos as u32;
}
bld.t.files = files.into_iter().map(|(f, _)| f).collect();
for s in &mut node_spans {
s.1 = file_rank[s.1 as usize];
}
for s in &mut sites {
s.file = file_rank[s.file as usize];
}
node_spans.sort_unstable();
node_spans.dedup();
let mut prev = NONE;
for &(o, f, s, e) in &node_spans {
if o == prev {
bld.t.more_spans.push((o, f, s, e));
} else {
bld.t.span_file[o as usize] = f;
bld.t.span_start[o as usize] = s;
bld.t.span_end[o as usize] = e;
}
prev = o;
}
sites.sort_unstable();
sites.dedup();
let span_ms = t_span.elapsed().as_millis() as u64;
rss_phase(scope.universe, "spans placed");
// ---- provenance (derived: L9-dec). With the universe ring outside the
// scope, a linked dependency / std item is still `both` when the ring
// defines it — same rule as EventLog::provenance. The ring's definitions
// come from the per-process ring cache (one fold per manifest snapshot).
let t2 = Instant::now();
let bin_fact = F_BIN_NODE | F_SYMBOL | F_INSTANCES;
let needs_lookup = !scope.universe && bld.t.flags.iter().any(|f| f & bin_fact != 0 && f & F_DEFINED == 0);
let mut ring_lookup = None;
if needs_lookup {
let (defs, cached) = ring::defs(log, &identity, manifest).await?;
let mut hits = 0u64;
for i in 0..n {
let f = bld.t.flags[i];
if f & bin_fact == 0 || f & F_DEFINED != 0 {
continue;
}
let Some(d) = defs.ids.find(kept_names.get(i as u64)) else { continue };
let (d, at) = (d as usize, defs.attrs[d as usize]);
bld.t.flags[i] |= F_UNIVERSE_DEF;
let a = Attrs {
kind: defs.s(at[0]),
level: defs.s(at[1]),
krate: defs.s(at[2]).unwrap_or(""),
version: defs.s(at[3]).unwrap_or(""),
origin: defs.s(at[4]),
repo: UNIVERSE_REPO,
name: defs.s(at[5]),
row: None,
};
bld.set(i, P_DEFINED, a, defs.hash[d], UNIVERSE_REPO, false, &ordinal_of);
hits += 1;
}
inputs.push(Input {
table: "code_node_events (ring definitions)".into(),
stream: "source".into(),
events_read: if cached { 0 } else { defs.events_read },
open_rows: hits,
});
ring_lookup = Some(cached);
}
let dups = bld.finish_hashes();
for i in 0..n {
let f = bld.t.flags[i];
let facts = BinaryFacts { has_symbol: f & F_SYMBOL != 0, instances_linked: f & F_INSTANCES != 0, binary_node: f & F_BIN_NODE != 0 };
let defined = f & (F_DEFINED | F_UNIVERSE_DEF) != 0;
bld.t.prov[i] = derive_provenance(defined, facts).map_or(0, |p| 1 + Provenance::ALL.iter().position(|x| *x == p).unwrap_or(0) as u8);
}
let provenance_ms = t2.elapsed().as_millis() as u64;
rss_phase(scope.universe, "provenance");
// ---- matrices: per kind, (from, to) packed into one u64, sorted, deduplicated
let t3 = Instant::now();
// the edges carry room ordinals already
let pack = |f: u32, t: u32| ((f as u64) << 32) | t as u64;
let mut counts: BTreeMap<EdgeKind, Vec<(u64, u32)>> = BTreeMap::new();
for (k, f, t, c) in odd_counts {
counts.entry(k).or_default().push((pack(f, t), c));
}
for v in counts.values_mut() {
// one edge described by several owners (a crate's lib and its test
// harness unit): the larger site count, never the sum
v.sort_unstable_by(|a, b| a.0.cmp(&b.0).then(b.1.cmp(&a.1)));
v.dedup_by_key(|e| e.0);
}
// an edge several owners carry is negated / partial only if EVERY row says
// so (one unit gating the item itself makes the gate a full one)
let mut edge_flags: BTreeMap<EdgeKind, Vec<(u64, u8)>> = BTreeMap::new();
for (k, f, t, fl) in flag_rows {
edge_flags.entry(k).or_default().push((pack(f, t), fl));
}
for v in edge_flags.values_mut() {
v.sort_unstable();
let mut folded: Vec<(u64, u8)> = Vec::with_capacity(v.len());
for (key, fl) in v.drain(..) {
match folded.last_mut() {
Some(last) if last.0 == key => last.1 &= fl,
_ => folded.push((key, fl)),
}
}
folded.retain(|e| e.1 != 0);
*v = folded;
}
// an edge several owners carry (a lib and its test harness unit) is read
// if ANY row reads and written if any row writes: the union
let mut edge_access: BTreeMap<EdgeKind, Vec<(u64, u8)>> = BTreeMap::new();
for (k, f, t, bits) in access_rows {
edge_access.entry(k).or_default().push((pack(f, t), bits));
}
for v in edge_access.values_mut() {
v.sort_unstable();
let mut folded: Vec<(u64, u8)> = Vec::with_capacity(v.len());
for (key, bits) in v.drain(..) {
match folded.last_mut() {
Some(last) if last.0 == key => last.1 |= bits,
_ => folded.push((key, bits)),
}
}
*v = folded;
}
let mut a = BTreeMap::new();
let mut edge_counts = BTreeMap::new();
let mut import_paths = BTreeMap::new();
let mut impl_self: Vec<(u32, u32)> = Vec::new();
for (k, pairs) in by_kind {
let mut packed: Vec<u64> = pairs.iter().map(|(f, t)| pack(*f, *t)).collect();
drop(pairs);
packed.sort_unstable();
packed.dedup();
let from: Vec<u64> = packed.iter().map(|p| p >> 32).collect();
let to: Vec<u64> = packed.iter().map(|p| p & 0xffff_ffff).collect();
drop(packed);
if k == EdgeKind::ForType {
impl_self = from.iter().zip(&to).map(|(f, t)| (*f as u32, *t as u32)).collect();
impl_self.dedup_by_key(|e| e.0);
}
let (m, path) = from_edge_chunks(&[(&from[..], &to[..])], n as u64)?;
m.wait()?;
edge_counts.insert(k, from.len() as u64);
import_paths.insert(k, path);
a.insert(k, m);
}
let matrix_ms = t3.elapsed().as_millis() as u64;
rss_phase(scope.universe, "matrices");
bld.t.names = kept_names;
let mut t = bld.t;
t.kinds = bld.kinds.list;
t.crates = bld.crates.list;
t.versions = bld.versions.list;
t.repos = bld.repos.list;
t.strs = bld.strs.list;
let node_bytes = t.bytes() + sites.capacity() * std::mem::size_of::<Site>() + symbols.heap();
// the fold handed back for a later advance — before the snapshot owns
// what both need
let fold = keep.then(|| FoldState {
identity: identity.clone(),
manifest,
extras,
unread_columns: unread_columns.clone(),
bnodes: bnodes.take().expect("kept"),
bedges: bedges.take().expect("kept"),
syms: syms.take().expect("kept"),
mon: mon.take().expect("kept"),
raw: raw.take().expect("kept"),
nrecs: nrecs.take().expect("kept"),
erecs: erecs.take().expect("kept"),
srecs: srecs.take().expect("kept"),
node_open: node_open.take().expect("kept"),
edge_open: edge_open.take().expect("kept"),
span_open: span_open.take().expect("kept"),
node_events,
edge_events,
span_events,
src_recs,
src_edges,
bin_recs_until,
bin_edges_until,
sym_mon: sym_mon.take().expect("kept"),
ring_owners,
ring_allowed,
stamp,
tracks,
started,
fold_ms,
});
static BUILDS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
let state_key = state_key_of(&identity, &scope, &stamp);
let snap = Snapshot {
version: BUILDS.fetch_add(1, std::sync::atomic::Ordering::Relaxed),
state_key,
identity: identity.clone(),
scope,
nodes: t,
graphs: Mutex::new(Graphs { n: n as u64, a, unions: HashMap::new(), transposes: HashMap::new(), jikes: None }),
edge_counts,
inputs,
repo_rows,
universe_rows,
unknown_kinds,
endpoint_only,
ring_owners,
ring_lookup,
manifest,
ring_closure,
stats: BuildStats { fold_ms, index_ms, sort_ms, fill_ms, symbols_ms, provenance_ms, matrix_ms, span_ms, total_ms: started.elapsed().as_millis() as u64, import_paths, node_bytes },
counts,
edge_flags,
edge_access,
edge_columns_added: log.spec(Kind::Edges).added.clone(),
sites,
impl_self,
symbols,
dups,
orphan_spans,
unread_columns,
name_idx: OnceLock::new(),
file_idx: OnceLock::new(),
symbol_idx: OnceLock::new(),
stable_idx: OnceLock::new(),
};
Ok((snap, fold))
}