crates/nornir-knowledge-view/src/lib.rs
//! # `nornir-knowledge-view` — the ONE reader over the persisted knowledge map
//!
//! The knowledge map (`symbol_facts` = pure-Rust syn symbols, `call_edges` =
//! caller→callee edges) is a WAREHOUSE artifact: the warehouse owner's syn
//! scanner writes it. But the **reader** is needed on both sides of the release
//! seam — the owner answers callers/callees/defined-in/symbol-lookup queries
//! over it, and the release orchestrator's SEMANTIC-BLAST gate walks the same
//! call graph to bound a release's blast radius. The orchestrator may not take a
//! crate edge on the owner's umbrella, so for a while both halves carried their
//! own copy of the reader — including the two single largest duplicated function
//! bodies in the constellation (`load_latest_audited`, `call_path`).
//!
//! PRE-PROMPT L5/L5a: the reader is a *primitive* — self-contained, dep-light,
//! reasoning only over the backend-neutral [`nornir_warehouse_trait::Warehouse`]
//! seam and its POD row types, with no domain coupling — so it lives ONCE here
//! and both conductors consume it. What stays with the conductors is the domain:
//! the syn/SCIP **scanners** that WRITE these tables, the SCIP moniker-join
//! preference helpers, and whatever each side does with a `KnowledgeView`.
//!
//! ## What is here
//! * [`KnowledgeView`] — a loaded snapshot's symbols + calls, with the
//! memoized [`CallIndex`] inverted index behind `callers_of` / `callees_of` /
//! `call_path` (O(hits) posting-list lookups + an integer-CSR BFS).
//! * [`load_latest`] / [`load_latest_audited`] — the two-phase
//! snapshot-pushdown reader (discover the latest `snapshot_id` for a repo with
//! a narrow 3-column scan, then read only that snapshot wide), plus the
//! [`ScanAudit`] read-amplification counter its regression test asserts on.
//!
//! ## Feature `stree` (OFF by default)
//! Lights the **R2.1 "Ragnar" batched accelerator**: a sorted-key `stree32`
//! (cache-line Eytzinger S+tree, SIMD + software-prefetch, from the sibling
//! `znippy-zoomies`) behind the two posting maps, exposed as
//! [`KnowledgeView::callers_of_batch`] / [`KnowledgeView::callees_of_batch`].
//! Results are byte-identical to the per-name HashMap path — the tree changes
//! only *how* a key is located, never *what* is returned — and it falls back to
//! the HashMap when the tree is empty. Off by default so a consumer that only
//! needs the reader pays no `znippy-zoomies` dependency.
use anyhow::{anyhow, Result};
use arrow::array::{Array, Int32Array, RecordBatch, StringArray, TimestampMicrosecondArray};
use std::collections::HashMap;
use std::sync::OnceLock;
#[cfg(feature = "stree")]
use znippy_zoomies::stree32::STree32;
use nornir_warehouse_trait::{
CallEdgeRow, ScanFilter, SymbolRow, Warehouse, TABLE_CALL_EDGES, TABLE_SYMBOL_FACTS,
};
/// Last `::`-separated path segment of `s` (the bare identifier).
fn last_seg(s: &str) -> &str {
s.rsplit("::").next().unwrap_or(s)
}
#[cfg(feature = "stree")]
/// 32-bit **FNV-1a** hash of a key string. Deterministic and dependency-free —
/// the [`StreeIndex`] maps every inverted-index key to one of these so the
/// sorted-key [`STree32`] can route it. The stree's unused-slot sentinel is
/// `u32::MAX`, so a key that hashes to `u32::MAX` is nudged to `u32::MAX - 1`;
/// the string cross-check in [`StreeIndex::resolve`] keeps that remap correct
/// (it only ever *adds* a benign hash collision, never a wrong answer).
#[inline]
fn fnv1a32(s: &str) -> u32 {
let mut h: u32 = 0x811c_9dc5;
for &b in s.as_bytes() {
h ^= b as u32;
h = h.wrapping_mul(0x0100_0193);
}
if h == u32::MAX { u32::MAX - 1 } else { h }
}
#[cfg(feature = "stree")]
/// **R2.1 "Ragnar" stree accelerator** — the sorted-key, SIMD + software-prefetch,
/// BATCHED-lookup index that sits *behind* the [`CallIndex`] HashMap path. It is a
/// cache-line Eytzinger S+tree (`stree32`, Ragnar Groot Koerkamp's static search
/// tree, proven on the OSM planet in the sibling `znippy-zoomies`), REUSED here —
/// not re-written — as the scale accelerator for `callers_of` / `callees_of`.
///
/// Built ONCE per loaded snapshot from an inverted-index keyset. A query name is
/// hashed to a `u32`; a whole batch of hashes is routed through the tree with
/// Ragnar's pipelined `batch_stream` (overlapping cache-miss latency across the
/// batch), then each hit is **cross-checked against the exact key string** in its
/// hash bucket. So a hash collision can never return the wrong posting list:
/// results are byte-identical to `HashMap::get(name)` — the tree changes only
/// *how* a key is located, never *what* is returned. The HashMap stays as the
/// single-lookup path and the cross-check / fallback oracle.
struct StreeIndex {
/// Unique, ascending key-hashes; the [`STree32`] is built over exactly this.
/// `find_exact`/`batch_stream` return indices INTO this slice (== bucket id).
sorted_hashes: Vec<u32>,
/// CSR bucket bounds into `entry_keys`/`entry_post`: the keys sharing the hash
/// `sorted_hashes[i]` are the entries `bucket_off[i]..bucket_off[i + 1]`.
bucket_off: Vec<u32>,
/// Key strings grouped by hash bucket — the collision cross-check material.
entry_keys: Vec<String>,
/// Posting lists parallel to `entry_keys`: edge indices in ascending `calls`
/// order, a clone of the matching [`CallIndex`] HashMap value (so the stree
/// path yields the identical edge set + order as the HashMap path).
entry_post: Vec<Vec<usize>>,
/// The static search tree over `sorted_hashes`. `None` for an empty keyset
/// (a tree cannot be built over zero keys) → callers fall back to the HashMap.
tree: Option<STree32>,
}
#[cfg(feature = "stree")]
impl std::fmt::Debug for StreeIndex {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("StreeIndex")
.field("keys", &self.entry_keys.len())
.field("buckets", &self.sorted_hashes.len())
.field("built", &self.tree.is_some())
.finish()
}
}
#[cfg(feature = "stree")]
impl StreeIndex {
/// Build the stree over the keys of an inverted-index map. Keys are grouped
/// by hash into ascending unique-hash buckets; within a bucket the keys are
/// sorted for a reproducible layout (the cross-check is by exact string).
fn build(keys: &HashMap<String, Vec<usize>>) -> Self {
let mut by_hash: HashMap<u32, Vec<&String>> = HashMap::new();
for k in keys.keys() {
by_hash.entry(fnv1a32(k)).or_default().push(k);
}
let mut sorted_hashes: Vec<u32> = by_hash.keys().copied().collect();
sorted_hashes.sort_unstable();
let mut bucket_off = Vec::with_capacity(sorted_hashes.len() + 1);
let mut entry_keys = Vec::with_capacity(keys.len());
let mut entry_post = Vec::with_capacity(keys.len());
bucket_off.push(0u32);
for h in &sorted_hashes {
let mut bucket = by_hash.remove(h).unwrap();
bucket.sort_unstable();
for k in bucket {
entry_keys.push(k.clone());
entry_post.push(keys[k].clone());
}
bucket_off.push(entry_keys.len() as u32);
}
let tree = (!sorted_hashes.is_empty()).then(|| STree32::new(&sorted_hashes));
StreeIndex { sorted_hashes, bucket_off, entry_keys, entry_post, tree }
}
/// Resident heap bytes of the built stree. Counted, not estimated: this is
/// the term a budget charged at insert time cannot see at all, because the
/// tree does not exist until the first batched query runs.
fn resident_bytes(&self) -> usize {
std::mem::size_of::<Self>()
+ self.sorted_hashes.capacity() * std::mem::size_of::<u32>()
+ self.bucket_off.capacity() * std::mem::size_of::<u32>()
+ self.entry_keys.capacity() * std::mem::size_of::<String>()
+ self.entry_keys.iter().map(|k| k.capacity()).sum::<usize>()
+ self.entry_post.capacity() * std::mem::size_of::<Vec<usize>>()
+ self
.entry_post
.iter()
.map(|v| v.capacity() * std::mem::size_of::<usize>())
.sum::<usize>()
+ self
.tree
.as_ref()
.map(|t| t.len() * std::mem::size_of::<u32>())
.unwrap_or(0)
}
/// Resolve a routed bucket index + the original `name` to its posting list,
/// scanning the (tiny) collision bucket for the EXACT key string.
#[inline]
fn resolve(&self, bucket: usize, name: &str) -> Option<&Vec<usize>> {
let lo = self.bucket_off[bucket] as usize;
let hi = self.bucket_off[bucket + 1] as usize;
(lo..hi).find(|&e| self.entry_keys[e] == name).map(|e| &self.entry_post[e])
}
/// Single-key lookup through the tree (SIMD `find_exact` + string cross-check).
/// Identical result to `HashMap::get(name)`. Used by the cross-check tests as
/// the scalar-stree oracle (the production single-lookup path stays on the
/// HashMap; the stree's role is the batched fast path).
#[cfg(test)]
#[inline]
fn get(&self, name: &str) -> Option<&Vec<usize>> {
let tree = self.tree.as_ref()?;
let bucket = tree.find_exact(fnv1a32(name))?;
self.resolve(bucket, name)
}
/// BATCHED lookup — the Ragnar fast path. Hashes the whole `names` batch, then
/// routes every hash through the tree in one pipelined `batch_stream` pass
/// (SIMD compares + software prefetch hiding memory latency across the batch),
/// and cross-checks each hit by string. Returns one posting-list reference (or
/// `None`) per input name, positionally — identical to `get` called per name.
fn get_batch(&self, names: &[&str]) -> Vec<Option<&Vec<usize>>> {
if self.tree.is_none() {
return vec![None; names.len()];
}
let tree = self.tree.as_ref().unwrap();
let hashes: Vec<u32> = names.iter().map(|n| fnv1a32(n)).collect();
let routed = tree.batch_stream::<16>(&hashes);
names
.iter()
.zip(routed)
.map(|(name, hit)| hit.and_then(|bucket| self.resolve(bucket, name)))
.collect()
}
}
/// **R0.3 "Ragnar" inverted index** over a [`KnowledgeView`]'s `calls`, built
/// ONCE per loaded snapshot and reused across every `callers_of` / `callees_of`
/// / `call_path` call (memoized behind a [`OnceLock`]). Replaces the previous
/// O(N) linear `Vec` scans (`ends_with("::name")` + per-row allocs) and the
/// per-call adjacency-map rebuild with O(hits) posting-list lookups.
///
/// **R2.1** adds a sorted-key [`StreeIndex`] (Ragnar's `stree32`) behind the two
/// posting maps, built lazily on the first BATCHED query, for SIMD + prefetch
/// batched lookups at scale — cross-checked against these maps for identity.
///
/// **Correctness contract:** every lookup returns EXACTLY the same edge set,
/// in the same order, as the old linear scan (see the `*_matches_linear_scan`
/// tests). The posting lists hold edge indices in ascending `calls` order, so
/// iterating a list yields the matching edges in their original `calls` order.
#[derive(Debug, Default)]
struct CallIndex {
/// `name` → indices of edges whose `callee_ident` matches `callers_of(name)`
/// (i.e. `callee == name` OR `callee.ends_with("::name")`). Keyed by every
/// segment-suffix of each `callee_ident` so a single `.get(name)` is exact.
callee_keys: HashMap<String, Vec<usize>>,
/// `name` → indices of edges whose `caller_path` matches `callees_of(name)`.
caller_keys: HashMap<String, Vec<usize>>,
/// Dense interning of every call-graph node (`last_seg` of a caller or callee)
/// → a `u32` id, built ONCE at index time. `node_name[id]` is the inverse
/// (id → name), used only to materialize a final path. Interning up front is
/// what lets `call_path`'s BFS be pure integer work — it never hashes a node
/// string in the hot loop (only the two endpoints are resolved, once each).
node_id: HashMap<String, u32>,
node_name: Vec<String>,
/// CSR-style memoized adjacency at identifier granularity: `csr[id]` holds the
/// callee ids of node `id` in edge order (so BFS explores neighbours in the
/// same order the old `HashMap<String, Vec<String>>` did → byte-identical
/// shortest paths). Replaces per-call adjacency rebuilds AND per-node hashing.
csr: Vec<Vec<u32>>,
/// R2.1 stree over `callee_keys` (the `callers_of` keyset), built on first
/// batched query and memoized. Empty/`None`-tree ⇒ batch falls back.
#[cfg(feature = "stree")]
callee_stree: OnceLock<StreeIndex>,
/// R2.1 stree over `caller_keys` (the `callees_of` keyset).
#[cfg(feature = "stree")]
caller_stree: OnceLock<StreeIndex>,
}
#[cfg(feature = "stree")]
impl CallIndex {
/// The memoized `callers_of` (callee-key) stree, built once from `callee_keys`.
fn callee_stree(&self) -> &StreeIndex {
self.callee_stree.get_or_init(|| StreeIndex::build(&self.callee_keys))
}
/// The memoized `callees_of` (caller-key) stree, built once from `caller_keys`.
fn caller_stree(&self) -> &StreeIndex {
self.caller_stree.get_or_init(|| StreeIndex::build(&self.caller_keys))
}
}
impl CallIndex {
/// Index every segment-suffix of `path` under edge index `i`.
///
/// A query `name` matches `path` iff `path == name` OR `path.ends_with("::name")`.
/// Because `::` only ever occurs as a path separator (identifiers cannot
/// contain `:`), the set of matching `name`s is exactly: the full `path`
/// plus each contiguous trailing segment-group. All such suffixes have
/// distinct lengths, so no `(key, i)` pair is inserted twice — each edge
/// appears at most once per posting list, matching the linear scan's
/// once-per-edge semantics.
fn index_suffixes(keys: &mut HashMap<String, Vec<usize>>, path: &str, i: usize) {
keys.entry(path.to_string()).or_default().push(i);
let mut rest = path;
while let Some(pos) = rest.find("::") {
rest = &rest[pos + 2..];
keys.entry(rest.to_string()).or_default().push(i);
}
}
/// Intern `name` to a dense `u32` node id, creating it (with an empty
/// adjacency slot and a `node_name` entry) on first sight. Interning happens
/// once, at build time, so the BFS never pays a node-string hash.
fn intern(&mut self, name: &str) -> u32 {
if let Some(&id) = self.node_id.get(name) {
return id;
}
let id = self.node_name.len() as u32;
self.node_id.insert(name.to_string(), id);
self.node_name.push(name.to_string());
self.csr.push(Vec::new());
id
}
/// Build the index in a single pass over `calls`, in `calls` order.
fn build(calls: &[CallEdgeRow]) -> Self {
let mut idx = CallIndex::default();
for (i, e) in calls.iter().enumerate() {
Self::index_suffixes(&mut idx.callee_keys, &e.callee_ident, i);
Self::index_suffixes(&mut idx.caller_keys, &e.caller_path, i);
let f = last_seg(&e.caller_path);
let t = last_seg(&e.callee_ident);
let fid = idx.intern(f);
let tid = idx.intern(t);
idx.csr[fid as usize].push(tid);
}
idx
}
}
/// Latest persisted symbols + calls for `repo`.
pub struct KnowledgeView {
pub symbols: Vec<SymbolRow>,
pub calls: Vec<CallEdgeRow>,
/// Lazily-built, memoized inverted index over `calls` (R0.3). Private: it is
/// pure derived state, rebuilt from `calls` on first query and never serialized.
index: OnceLock<CallIndex>,
}
impl KnowledgeView {
/// Construct a view from its `symbols` + `calls`. The inverted index starts
/// empty and is built on the first call-graph query (see [`CallIndex`]).
pub fn new(symbols: Vec<SymbolRow>, calls: Vec<CallEdgeRow>) -> Self {
KnowledgeView { symbols, calls, index: OnceLock::new() }
}
/// The memoized inverted index, built once on first access from `calls`.
fn index(&self) -> &CallIndex {
self.index.get_or_init(|| CallIndex::build(&self.calls))
}
/// **Resident heap bytes of this view, INCLUDING everything built lazily
/// inside it.**
///
/// A `KnowledgeView` is a cache entry of genuinely unbounded size — one
/// repo's whole symbol set and call graph — and worse, it GROWS after it is
/// handed out: `index` is built on the first call-graph query, and the two
/// `stree`s inside that index are built on the first batched query, all
/// through `&self` inside an already-shared `Arc`. A byte budget charged
/// only when the view was inserted therefore under-counts a *hot* repo (the
/// one worth evicting) far more than a cold one, which is exactly backwards.
/// Callers must charge on USE: re-read this after the query, not before.
///
/// Strings are counted by capacity; the `HashMap` tables carry a
/// per-entry overhead estimate (~16 B on top of the key/value pair), the
/// only estimated term. Everything else is measured.
pub fn resident_bytes(&self) -> usize {
let symbols: usize = self.symbols.capacity() * std::mem::size_of::<SymbolRow>()
+ self.symbols.iter().map(symbol_row_bytes).sum::<usize>();
let calls: usize = self.calls.capacity() * std::mem::size_of::<CallEdgeRow>()
+ self.calls.iter().map(call_edge_row_bytes).sum::<usize>();
let index = match self.index.get() {
Some(ix) => ix.resident_bytes(),
None => 0,
};
std::mem::size_of::<Self>() + symbols + calls + index
}
}
fn symbol_row_bytes(r: &SymbolRow) -> usize {
r.crate_name.capacity()
+ r.module_path.capacity()
+ r.item_kind.capacity()
+ r.item_name.capacity()
+ r.visibility.capacity()
+ r.file.capacity()
+ r.signature.as_ref().map(|s| s.capacity()).unwrap_or(0)
}
fn call_edge_row_bytes(r: &CallEdgeRow) -> usize {
r.crate_name.capacity()
+ r.caller_path.capacity()
+ r.callee_ident.capacity()
+ r.call_kind.capacity()
+ r.file.capacity()
}
/// Per-entry overhead a `HashMap` table carries on top of its `(K, V)` pair —
/// hashbrown's control byte plus the slack at its load factor. An estimate; it
/// is the only estimated term in [`KnowledgeView::resident_bytes`].
const MAP_ENTRY_OVERHEAD: usize = 16;
fn posting_map_bytes(m: &HashMap<String, Vec<usize>>) -> usize {
m.capacity() * (std::mem::size_of::<(String, Vec<usize>)>() + MAP_ENTRY_OVERHEAD)
+ m.iter()
.map(|(k, v)| k.capacity() + v.capacity() * std::mem::size_of::<usize>())
.sum::<usize>()
}
impl CallIndex {
/// Resident heap bytes of the built index, including the lazily-built
/// `stree`s when (and only when) they exist.
fn resident_bytes(&self) -> usize {
let mut n = std::mem::size_of::<Self>()
+ posting_map_bytes(&self.callee_keys)
+ posting_map_bytes(&self.caller_keys);
n += self.node_id.capacity()
* (std::mem::size_of::<(String, u32)>() + MAP_ENTRY_OVERHEAD)
+ self.node_id.keys().map(|k| k.capacity()).sum::<usize>();
n += self.node_name.capacity() * std::mem::size_of::<String>()
+ self.node_name.iter().map(|s| s.capacity()).sum::<usize>();
n += self.csr.capacity() * std::mem::size_of::<Vec<u32>>()
+ self
.csr
.iter()
.map(|v| v.capacity() * std::mem::size_of::<u32>())
.sum::<usize>();
#[cfg(feature = "stree")]
{
n += self
.callee_stree
.get()
.map(StreeIndex::resident_bytes)
.unwrap_or(0);
n += self
.caller_stree
.get()
.map(StreeIndex::resident_bytes)
.unwrap_or(0);
}
n
}
}
/// Read-amplification audit for [`load_latest_audited`] — how many rows the query
/// actually decoded, split by phase. `data_rows` is the count that matters: the
/// FULL-WIDTH rows materialized into `SymbolRow`/`CallEdgeRow`. The old full-scan
/// path decoded every snapshot's rows here; the snapshot-pushdown path decodes
/// only the latest snapshot's. Surfaced for the FAIL-ON-BUG/PASS-ON-FIX test.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct ScanAudit {
/// Rows read by the narrow (3-column) snapshot-discovery scan.
pub discovery_rows: usize,
/// FULL-WIDTH rows decoded into the view (`symbol_facts` + `call_edges`).
pub data_rows: usize,
}
fn col<'a, T: 'static>(batch: &'a RecordBatch, name: &str) -> Result<&'a T> {
batch
.column_by_name(name)
.ok_or_else(|| anyhow!("missing column `{name}`"))?
.as_any()
.downcast_ref::<T>()
.ok_or_else(|| anyhow!("column `{name}` has unexpected type"))
}
/// Load the latest-snapshot symbols + calls for `repo` from iceberg.
pub fn load_latest(wh: &dyn Warehouse, repo: &str) -> Result<KnowledgeView> {
Ok(load_latest_audited(wh, repo)?.0)
}
/// [`load_latest`] plus a [`ScanAudit`] of how many rows were decoded.
///
/// Two-phase, byte-identical to the old full-scan path (same max-`ts_micros`
/// first-seen-wins snapshot pick, same residual `snapshot_id == snap` row guard,
/// same row order), but with the wide decode pruned to one snapshot:
/// 1. `discover_latest_snapshot` — narrow scan of `symbol_facts` picks the
/// latest `snapshot_id` for `repo`.
/// 2. `read_snapshot_batches` — `snapshot_id`-pushdown reads of `symbol_facts`
/// and `call_edges`, decoding ONLY that snapshot's data file(s).
///
/// `call_edges` reuses the `symbol_facts` snapshot id (the two are written under
/// the same scan stamp), exactly as the old path applied `symbol_facts`'s `snap`
/// to the calls scan.
pub fn load_latest_audited(
wh: &dyn Warehouse,
repo: &str,
) -> Result<(KnowledgeView, ScanAudit)> {
// ── discover the latest snapshot (narrow, repo-pushdown) ──────────────
// Routed through the backend-neutral `scan_filtered` (repo == …, project the
// three discovery columns) — same file-pruning the old `discover_latest_snapshot`
// did over the raw table, now over the trait.
let disc = wh.scan_filtered(
TABLE_SYMBOL_FACTS,
&ScanFilter::eq("repo", repo),
&["repo", "snapshot_id", "ts_micros"],
)?;
let mut latest: Option<(String, i64)> = None;
let mut discovery_rows = 0usize;
for b in &disc {
let repos = col::<StringArray>(b, "repo")?;
let snaps = col::<StringArray>(b, "snapshot_id")?;
let ts = col::<TimestampMicrosecondArray>(b, "ts_micros")?;
for i in 0..b.num_rows() {
// Residual repo guard (pushdown prunes at file granularity).
if repos.value(i) != repo {
continue;
}
discovery_rows += 1;
let t = ts.value(i);
if latest.as_ref().map(|(_, lt)| t > *lt).unwrap_or(true) {
latest = Some((snaps.value(i).to_string(), t));
}
}
}
let Some((snap, _)) = latest else {
return Ok((KnowledgeView::new(vec![], vec![]), ScanAudit::default()));
};
let mut data_rows = 0usize;
// ── symbols (latest snapshot only, snapshot_id-pushdown) ──────────────
let s_batches =
wh.scan_filtered(TABLE_SYMBOL_FACTS, &ScanFilter::eq("snapshot_id", snap.clone()), &[])?;
let mut symbols = Vec::new();
for b in &s_batches {
let snaps = col::<StringArray>(b, "snapshot_id")?;
let crate_name = col::<StringArray>(b, "crate_name")?;
let module_path = col::<StringArray>(b, "module_path")?;
let item_kind = col::<StringArray>(b, "item_kind")?;
let item_name = col::<StringArray>(b, "item_name")?;
let visibility = col::<StringArray>(b, "visibility")?;
let file = col::<StringArray>(b, "file")?;
let line = col::<Int32Array>(b, "line")?;
let doc_lines = col::<Int32Array>(b, "doc_lines")?;
let signature = col::<StringArray>(b, "signature")?;
for i in 0..b.num_rows() {
// Residual guard: pushdown prunes whole files, so a non-`snap`
// row can only appear if a future writer ever co-locates two
// snapshots in one file — keep the guard to stay byte-identical.
if snaps.value(i) != snap {
continue;
}
data_rows += 1;
let sig = signature.value(i);
symbols.push(SymbolRow {
crate_name: crate_name.value(i).to_string(),
module_path: module_path.value(i).to_string(),
item_kind: item_kind.value(i).to_string(),
item_name: item_name.value(i).to_string(),
visibility: visibility.value(i).to_string(),
file: file.value(i).to_string(),
line: line.value(i).max(0) as u32,
doc_lines: doc_lines.value(i).max(0) as u32,
signature: if sig.is_empty() { None } else { Some(sig.to_string()) },
});
}
}
// ── calls (same latest snapshot) ──────────────────────────
let c_batches =
wh.scan_filtered(TABLE_CALL_EDGES, &ScanFilter::eq("snapshot_id", snap.clone()), &[])?;
let mut calls = Vec::new();
for b in &c_batches {
let snaps = col::<StringArray>(b, "snapshot_id")?;
let crate_name = col::<StringArray>(b, "crate_name")?;
let caller = col::<StringArray>(b, "caller_path")?;
let callee = col::<StringArray>(b, "callee_ident")?;
let kind = col::<StringArray>(b, "call_kind")?;
let file = col::<StringArray>(b, "file")?;
let line = col::<Int32Array>(b, "line")?;
for i in 0..b.num_rows() {
if snaps.value(i) != snap {
continue;
}
data_rows += 1;
calls.push(CallEdgeRow {
crate_name: crate_name.value(i).to_string(),
caller_path: caller.value(i).to_string(),
callee_ident: callee.value(i).to_string(),
call_kind: kind.value(i).to_string(),
file: file.value(i).to_string(),
line: line.value(i).max(0) as u32,
});
}
}
Ok((KnowledgeView::new(symbols, calls), ScanAudit { discovery_rows, data_rows }))
}
impl KnowledgeView {
/// Symbols whose `item_name` contains `pattern` (case-insensitive).
pub fn symbol_lookup(&self, pattern: &str, limit: usize) -> Vec<&SymbolRow> {
let p = pattern.to_lowercase();
self.symbols
.iter()
.filter(|s| s.item_name.to_lowercase().contains(&p))
.take(limit)
.collect()
}
/// Symbols defined in a file whose path ends with `suffix`.
pub fn defined_in(&self, suffix: &str) -> Vec<&SymbolRow> {
self.symbols.iter().filter(|s| s.file.ends_with(suffix)).collect()
}
/// **The function-tier seam.** Lift this view's `symbol_facts` + `call_edges`
/// into the shared [`nornir_depgraph::funcgraph`] model: one
/// [`CrateFuncGraph`](nornir_depgraph::funcgraph::CrateFuncGraph) per crate —
/// its function nodes (`(stable_id, name, crate)`) and the resolved intra-crate
/// call edges among them. This is what makes the WORLD GRAPH's function tier
/// REAL. `repo` scopes the stable id, exactly as
/// `nornir::mimir_analyze::assemble_symbol_vecs` does, so nodes dedupe and
/// edges resolve consistently across the constellation.
///
/// A one-line adapter over the loaded rows — the transform itself lives in the
/// warehouse-free `nornir-depgraph` leaf, reachable by the dwarves warehouse
/// navigator without pulling arrow/iceberg.
#[must_use]
pub fn function_graph(&self, repo: &str) -> Vec<nornir_depgraph::funcgraph::CrateFuncGraph> {
use nornir_depgraph::funcgraph::{CallIn, SymIn};
let syms: Vec<SymIn> = self
.symbols
.iter()
.map(|s| SymIn {
crate_name: &s.crate_name,
module_path: &s.module_path,
item_kind: &s.item_kind,
item_name: &s.item_name,
})
.collect();
let calls: Vec<CallIn> = self
.calls
.iter()
.map(|c| CallIn {
crate_name: &c.crate_name,
caller_path: &c.caller_path,
callee_ident: &c.callee_ident,
})
.collect();
nornir_depgraph::funcgraph::build_function_graph(repo, &syms, &calls)
}
/// Call edges that *invoke* `name`. Matches either an exact `callee_ident`
/// (bare method calls like `.new()`) or a path-qualified callee whose last
/// segment is `name` (`Arc::new`, `Foo::new` all match a query of `new`).
/// The `::` separator is required, so `new` does not match `renew`.
///
/// O(hits): a single posting-list lookup in the memoized [`CallIndex`],
/// returning the matching edges in their original `calls` order — identical
/// to the previous linear `callee_ident == name || ends_with("::name")` scan.
pub fn callers_of(&self, name: &str) -> Vec<&CallEdgeRow> {
match self.index().callee_keys.get(name) {
Some(hits) => hits.iter().map(|&i| &self.calls[i]).collect(),
None => Vec::new(),
}
}
/// Call edges *from* a caller whose path ends with `name`.
///
/// O(hits): a single posting-list lookup in the memoized [`CallIndex`] —
/// identical to the previous linear `caller_path == name || ends_with("::name")`
/// scan, returning the matching edges in their original `calls` order.
pub fn callees_of(&self, name: &str) -> Vec<&CallEdgeRow> {
match self.index().caller_keys.get(name) {
Some(hits) => hits.iter().map(|&i| &self.calls[i]).collect(),
None => Vec::new(),
}
}
#[cfg(feature = "stree")]
/// **R2.1 BATCHED `callers_of`** — resolve MANY callee names in ONE shot
/// through the "Ragnar" sorted-key stree (`stree32`): the batch of names is
/// hashed and routed through the cache-line Eytzinger S+tree with Ragnar's
/// pipelined SIMD + software-prefetch traversal (overlapping each query's
/// cache-miss latency), then cross-checked by string. Returns, positionally,
/// the matching edges for each input name.
///
/// **Identical results** to calling [`callers_of`](Self::callers_of) per
/// name — the stree only changes *how* each key is located; the posting list
/// it resolves to is the same `callee_keys` entry the HashMap holds, mapped to
/// the same `&CallEdgeRow`s in the same order. The HashMap path is the
/// single-lookup default and the fallback when the tree is empty. This is the
/// scale accelerator: one pipelined tree walk over the whole batch instead of
/// N independent HashMap probes.
pub fn callers_of_batch(&self, names: &[&str]) -> Vec<Vec<&CallEdgeRow>> {
let stree = self.index().callee_stree();
if stree.tree.is_none() {
return names.iter().map(|n| self.callers_of(n)).collect();
}
stree
.get_batch(names)
.into_iter()
.map(|hit| match hit {
Some(post) => post.iter().map(|&i| &self.calls[i]).collect(),
None => Vec::new(),
})
.collect()
}
#[cfg(feature = "stree")]
/// **R2.1 BATCHED `callees_of`** — the [`callers_of_batch`](Self::callers_of_batch)
/// counterpart over the `caller_path` keyset. Identical results to
/// [`callees_of`](Self::callees_of) called per name; falls back to the HashMap
/// for an empty tree.
pub fn callees_of_batch(&self, names: &[&str]) -> Vec<Vec<&CallEdgeRow>> {
let stree = self.index().caller_stree();
if stree.tree.is_none() {
return names.iter().map(|n| self.callees_of(n)).collect();
}
stree
.get_batch(names)
.into_iter()
.map(|hit| match hit {
Some(post) => post.iter().map(|&i| &self.calls[i]).collect(),
None => Vec::new(),
})
.collect()
}
/// Shortest call chain from `from` to `to` over the persisted call edges
/// (BFS following caller → callee), at **identifier granularity**: each
/// node is a function's last path segment, so a query of `build`/`new`
/// matches `Index::build`/`Arc::new`. Returns the sequence of identifiers
/// from `from` to `to` inclusive, or `None` when unreachable.
///
/// Approximate by construction: the syn facts record callees as
/// identifiers (`Arc::new` is stored path-qualified, a bare `.new()` is
/// not), never as fully-resolved defining paths, so distinct functions
/// that share a name collapse to one node. Use it to surface *a* plausible
/// call chain (like `dep_path` for repos), not a guaranteed-unique one.
pub fn call_path(&self, from: &str, to: &str) -> Option<Vec<String>> {
use std::collections::VecDeque;
let from = last_seg(from);
let to = last_seg(to);
// Memoized CSR adjacency + node interning, built ONCE per snapshot in
// `CallIndex` and reused across calls; the per-node callee lists preserve
// edge order, so BFS yields the same path as the old rebuild-every-call
// version.
//
// Integer traversal: resolve the two endpoints to dense `u32` ids ONCE
// (the only string hashes in the whole query), then walk the CSR graph
// with flat `Vec` `seen`/`parent` arrays indexed by id. The previous
// version hashed every visited node into a `HashSet<&str>`/`HashMap<&str,
// &str>` and hashed each `adj.get(cur)`; this BFS does zero hashing in the
// hot loop and touches contiguous integer arrays instead. Result set +
// order are byte-identical (same neighbours in the same edge order); the
// `indexed_lookups_match_linear_scan_exactly` differential test guards it.
let idx = self.index();
// `from` unknown ⇒ None — subsumes the old `from == to && !contains(from)`
// and `from != to && !contains(from)` guards (both returned None).
let Some(&from_id) = idx.node_id.get(from) else { return None };
if from == to {
return Some(vec![from.to_string()]);
}
// `to` unknown ⇒ unreachable ⇒ None (the old BFS would never dequeue it).
let Some(&to_id) = idx.node_id.get(to) else { return None };
let n = idx.node_name.len();
let mut parent: Vec<u32> = vec![u32::MAX; n];
let mut seen: Vec<bool> = vec![false; n];
let mut queue: VecDeque<u32> = VecDeque::new();
seen[from_id as usize] = true;
queue.push_back(from_id);
while let Some(cur) = queue.pop_front() {
for &c in &idx.csr[cur as usize] {
let cu = c as usize;
if seen[cu] {
continue;
}
seen[cu] = true;
parent[cu] = cur;
if c == to_id {
let mut path = vec![idx.node_name[to_id as usize].clone()];
let mut node = to_id;
while parent[node as usize] != u32::MAX {
let p = parent[node as usize];
path.push(idx.node_name[p as usize].clone());
node = p;
}
path.reverse();
return Some(path);
}
queue.push_back(c);
}
}
None
}
}
#[cfg(test)]
mod tests {
use super::*;
use nornir_warehouse_trait::{CallEdgeRow, SymbolRow};
/// The function-tier seam over a real `KnowledgeView`: a repo whose latest
/// snapshot HAS function symbols yields NON-EMPTY real nodes + resolved call
/// edges, with ids in nornir's `{repo}::{crate}::{module}::{item}` synthesis.
#[test]
fn function_graph_yields_real_nodes_and_edges() {
fn srow(krate: &str, module: &str, kind: &str, name: &str) -> SymbolRow {
SymbolRow {
crate_name: krate.into(),
module_path: module.into(),
item_kind: kind.into(),
item_name: name.into(),
visibility: "pub".into(),
file: "src/lib.rs".into(),
line: 1,
doc_lines: 0,
signature: None,
}
}
let symbols = vec![
srow("nornir-git", "crate::head", "fn", "resolve"),
srow("nornir-git", "crate::head", "fn", "parse"),
srow("nornir-git", "crate", "struct", "Repo"), // not a fn
];
let calls = vec![CallEdgeRow {
crate_name: "nornir-git".into(),
caller_path: "crate::head::resolve".into(),
callee_ident: "parse".into(),
call_kind: "call".into(),
file: "src/lib.rs".into(),
line: 2,
}];
let view = KnowledgeView::new(symbols, calls);
let graphs = view.function_graph("edda");
assert_eq!(graphs.len(), 1);
let g = &graphs[0];
assert_eq!(g.crate_name, "nornir-git");
assert!(!g.is_empty(), "a crate with fn symbols is never empty");
assert_eq!(g.nodes.len(), 2, "the struct is not a function-tier node");
assert!(g.nodes.iter().any(|n| n.id == "edda::nornir-git::crate::head::resolve"));
assert_eq!(g.edges.len(), 1, "resolve→parse resolved");
assert_eq!(g.edges[0].from, "edda::nornir-git::crate::head::resolve");
assert_eq!(g.edges[0].to, "edda::nornir-git::crate::head::parse");
}
fn edge(callee: &str) -> CallEdgeRow {
CallEdgeRow {
crate_name: "demo".into(),
caller_path: "demo::f".into(),
callee_ident: callee.into(),
call_kind: "call".into(),
file: "src/lib.rs".into(),
line: 1,
}
}
fn edge_from(caller: &str, callee: &str) -> CallEdgeRow {
CallEdgeRow {
crate_name: "demo".into(),
caller_path: caller.into(),
callee_ident: callee.into(),
call_kind: "call".into(),
file: "src/lib.rs".into(),
line: 1,
}
}
/// R0.2 FAIL-ON-BUG / PASS-ON-FIX. Build one warehouse fixture with THREE
/// `demo` snapshots (growing history) plus a second repo, then run the OLD
/// full-history scan algorithm and the NEW snapshot-pushdown `load_latest`
/// against it and assert:
/// (a) the two results are byte-identical (serialized), and
/// (b) the NEW path decodes strictly FEWER full-width rows than OLD
/// (`audit.data_rows < old_rows_scanned`), confirmed independently by
/// `plan_stats`: a `snapshot_id` predicate prunes to one snapshot's
/// data file, where the `repo` predicate plans the whole history.
///
/// Validated as DATA (serialized views + row counts), never by eyeballing.
/// If the snapshot-pushdown wiring is reverted, the NEW path reads the whole
/// history → `data_rows == old_rows` → assertion (b) FAILS.
#[test]
fn callers_of_matches_last_segment_and_bare() {
let view = KnowledgeView::new(
vec![],
vec![
edge("new"), // bare method call
edge("Arc::new"), // path-qualified
edge("Foo::new"), // path-qualified
edge("renew"), // must NOT match (no `::` boundary)
edge("Foo::make"), // unrelated
],
);
let hits: Vec<&str> = view.callers_of("new").iter().map(|c| c.callee_ident.as_str()).collect();
assert!(hits.contains(&"new"));
assert!(hits.contains(&"Arc::new"));
assert!(hits.contains(&"Foo::new"));
assert!(!hits.contains(&"renew"), "{hits:?}");
assert!(!hits.contains(&"Foo::make"));
assert_eq!(hits.len(), 3);
// A fully-qualified query still matches exactly.
let exact: Vec<&str> = view.callers_of("Arc::new").iter().map(|c| c.callee_ident.as_str()).collect();
assert_eq!(exact, vec!["Arc::new"]);
}
#[test]
fn call_path_bfs_over_call_edges() {
// chain: a::run -> b::step -> c::commit ; plus a detour a::run -> z::noop
let view = KnowledgeView::new(
vec![],
vec![
edge_from("a::run", "step"),
edge_from("b::step", "Repo::commit"),
edge_from("a::run", "noop"),
],
);
// last-segment identity: run -> step -> commit
let p = view.call_path("run", "commit").expect("path exists");
assert_eq!(p, vec!["run", "step", "commit"]);
// fully-qualified inputs are normalised to their last segment.
let p2 = view.call_path("a::run", "Repo::commit").expect("path exists");
assert_eq!(p2, vec!["run", "step", "commit"]);
// self-path when the node exists.
assert_eq!(view.call_path("step", "step"), Some(vec!["step".to_string()]));
// unreachable + unknown source.
assert_eq!(view.call_path("commit", "run"), None);
assert_eq!(view.call_path("ghost", "run"), None);
}
// ════════════════════════════════════════════════════════════════════
// R0.3 — inverted index + memoized adjacency. The index MUST be a drop-in
// replacement for the old linear `Vec` scans: identical result sets, in
// identical order, with O(hits) (not O(N)) lookup cost.
// ════════════════════════════════════════════════════════════════════
/// The ORIGINAL O(N) linear `callers_of` — the oracle the index must match
/// EXACTLY. (Pre-R0.3 body, kept verbatim as the reference implementation.)
fn linear_callers_of<'a>(calls: &'a [CallEdgeRow], name: &str) -> Vec<&'a CallEdgeRow> {
let suffix = format!("::{name}");
calls
.iter()
.filter(|c| c.callee_ident == name || c.callee_ident.ends_with(&suffix))
.collect()
}
/// The ORIGINAL O(N) linear `callees_of` — oracle for the index.
fn linear_callees_of<'a>(calls: &'a [CallEdgeRow], name: &str) -> Vec<&'a CallEdgeRow> {
calls
.iter()
.filter(|c| c.caller_path == name || c.caller_path.ends_with(&format!("::{name}")))
.collect()
}
/// The ORIGINAL `call_path`, rebuilding the adjacency `BTreeMap` every call —
/// the oracle the memoized-adjacency BFS must match EXACTLY.
fn linear_call_path(calls: &[CallEdgeRow], from: &str, to: &str) -> Option<Vec<String>> {
use std::collections::{BTreeMap, BTreeSet, VecDeque};
fn last_seg(s: &str) -> &str {
s.rsplit("::").next().unwrap_or(s)
}
let from = last_seg(from).to_string();
let to = last_seg(to).to_string();
let mut adj: BTreeMap<&str, Vec<&str>> = BTreeMap::new();
let mut nodes: BTreeSet<&str> = BTreeSet::new();
for e in calls {
let f = last_seg(&e.caller_path);
let t = last_seg(&e.callee_ident);
adj.entry(f).or_default().push(t);
nodes.insert(f);
nodes.insert(t);
}
if from == to {
return nodes.contains(from.as_str()).then(|| vec![from]);
}
if !nodes.contains(from.as_str()) {
return None;
}
let mut parent: BTreeMap<String, String> = BTreeMap::new();
let mut seen: BTreeSet<String> = BTreeSet::new();
let mut queue: VecDeque<String> = VecDeque::new();
seen.insert(from.clone());
queue.push_back(from.clone());
while let Some(cur) = queue.pop_front() {
let Some(callees) = adj.get(cur.as_str()) else { continue };
for &c in callees {
if !seen.insert(c.to_string()) {
continue;
}
parent.insert(c.to_string(), cur.clone());
if c == to {
let mut path = vec![to.clone()];
let mut node = to.clone();
while let Some(p) = parent.get(&node) {
path.push(p.clone());
node = p.clone();
}
path.reverse();
return Some(path);
}
queue.push_back(c.to_string());
}
}
None
}
/// A deliberately gnarly fixture: bare callees, single- and multi-segment
/// qualified callees, a `renew` decoy (must NOT match `new`), repeated
/// callees in non-adjacent positions (order sensitivity), and callers at
/// varying path depths. Edge order is interleaved on purpose.
fn rich_fixture() -> Vec<CallEdgeRow> {
vec![
edge_from("a::run", "step"), // 0
edge_from("mod::a::run", "Arc::new"), // 1
edge_from("b::step", "Repo::commit"), // 2
edge_from("x::y::run", "new"), // 3 bare callee
edge_from("run", "Foo::new"), // 4 bare caller
edge_from("c::renew", "renew"), // 5 decoy: not `new`/`run`
edge_from("p::q::step", "a::b::new"), // 6 multi-segment callee
edge_from("k::Arc::new", "Arc::new"), // 7 repeated callee, later
edge_from("z::run", "noop"), // 8
edge_from("b::step", "x::y::new"), // 9 another `*::new`
]
}
/// Identity of a `&CallEdgeRow` *within the same backing slice* — the
/// strongest equality: same element, same position. Asserting on these
/// catches any divergence in BOTH membership and order.
fn ptrs(rows: Vec<&CallEdgeRow>) -> Vec<*const CallEdgeRow> {
rows.into_iter().map(|r| r as *const CallEdgeRow).collect()
}
/// Borrowing variant of [`ptrs`] for a `&[&CallEdgeRow]` (a per-name entry of
/// a BATCHED result) — asserts on it without moving the batch's Vec.
fn ptrs_v(rows: &[&CallEdgeRow]) -> Vec<*const CallEdgeRow> {
rows.iter().map(|&r| r as *const CallEdgeRow).collect()
}
/// FAIL-ON-BUG / PASS-ON-FIX: the indexed `callers_of` / `callees_of` return
/// EXACTLY the same edges (same set, same order) as the old linear scan, over
/// a battery of query names probing every match shape.
#[test]
fn indexed_lookups_match_linear_scan_exactly() {
let view = KnowledgeView::new(vec![], rich_fixture());
let names = [
"new", "run", "step", "Arc::new", "commit", "Repo::commit", "renew", "noop",
"a::b::new", "b::new", "y::new", "x::y::new", "Foo::new", "ghost", "",
];
for name in names {
assert_eq!(
ptrs(view.callers_of(name)),
ptrs(linear_callers_of(&view.calls, name)),
"callers_of({name:?}) must equal the linear scan (set + order)",
);
assert_eq!(
ptrs(view.callees_of(name)),
ptrs(linear_callees_of(&view.calls, name)),
"callees_of({name:?}) must equal the linear scan (set + order)",
);
}
// call_path: memoized adjacency must yield the SAME path as rebuilding it.
let pairs = [
("run", "new"),
("a::run", "Arc::new"),
("run", "commit"),
("step", "commit"),
("run", "run"),
("step", "step"),
("ghost", "new"),
("commit", "run"),
("run", "noop"),
("step", "new"),
("run", "renew"),
];
for (from, to) in pairs {
assert_eq!(
view.call_path(from, to),
linear_call_path(&view.calls, from, to),
"call_path({from:?}, {to:?}) must equal the linear rebuild",
);
}
}
/// COMPLEXITY assertion as DATA: the lookup walks a posting list whose
/// length is the HIT count, INDEPENDENT of N. A linear scan would touch all
/// N rows; here the work (posting-list length) stays constant as N grows
/// 200× — the operational meaning of O(hits) rather than O(N).
#[test]
fn callers_of_lookup_is_o_hits_not_o_n() {
fn build(noise: usize) -> KnowledgeView {
let mut calls = Vec::with_capacity(noise + 3);
for i in 0..noise {
calls.push(edge_from(&format!("noise::c{i}"), &format!("noise::g{i}")));
}
// Exactly three edges invoke `target` (bare, single-, multi-qualified).
calls.push(edge_from("m::one", "target"));
calls.push(edge_from("m::two", "Wrap::target"));
calls.push(edge_from("m::three", "a::b::target"));
KnowledgeView::new(vec![], calls)
}
let small = build(1_000);
let big = build(200_000);
// Same answer regardless of N.
assert_eq!(small.callers_of("target").len(), 3);
assert_eq!(big.callers_of("target").len(), 3);
// The posting list the lookup dereferences == the hit count, for BOTH N.
let small_work = small.index().callee_keys.get("target").map_or(0, Vec::len);
let big_work = big.index().callee_keys.get("target").map_or(0, Vec::len);
assert_eq!(small_work, 3, "posting list holds only the 3 hits at N=1_003");
assert_eq!(big_work, 3, "posting list STILL holds only the 3 hits at N=200_003");
assert_eq!(big_work, small_work, "lookup work independent of N ⇒ O(hits), not O(N)");
assert!(big.calls.len() >= 200_000, "the large fixture really is large");
// A miss touches ZERO edges (no posting list), never the whole Vec.
assert!(big.index().callee_keys.get("absent_symbol").is_none());
assert!(big.callers_of("absent_symbol").is_empty());
}
/// The adjacency map + posting index are built ONCE per loaded snapshot and
/// reused across every call (the memoization the task asks for).
#[test]
fn index_is_built_once_and_reused() {
let view = KnowledgeView::new(vec![], rich_fixture());
// Lazy: nothing built until the first call-graph query.
assert!(view.index.get().is_none(), "index is not built eagerly");
let _ = view.callers_of("new");
let built = view.index.get().expect("index materialised on first query");
let first = built as *const CallIndex;
// Subsequent queries of every kind reuse the SAME index instance.
let _ = view.callees_of("run");
let _ = view.call_path("run", "commit");
let again = view.index.get().expect("still present") as *const CallIndex;
assert_eq!(first, again, "same CallIndex reused — built once, not per call");
}
// ════════════════════════════════════════════════════════════════════
// R2.1 — the "Ragnar" stree32 sorted-key, SIMD + software-prefetch,
// BATCHED-lookup index wired BEHIND the R0.3 inverted index. It MUST be a
// drop-in accelerator: identical result sets, in identical order, to BOTH
// the inverted-index HashMap path AND the original linear scan — over a
// gnarly small fixture and a large one with real tree depth.
// ════════════════════════════════════════════════════════════════════
/// FAIL-ON-BUG / PASS-ON-FIX: the R2.1 stree32 lookups equal the inverted
/// index AND the linear scan oracle, EXACTLY (same edges, same order).
///
/// Three independent cross-checks, all validated as DATA (`&CallEdgeRow`
/// pointer identity within the backing slice — catches any divergence in
/// membership OR order), never by eyeballing:
/// 1. single-key stree `get` == the `CallIndex` HashMap `.get` (the
/// cross-check / fallback oracle the task names);
/// 2. BATCHED `callers_of_batch` / `callees_of_batch` (the actual Ragnar
/// pipelined SIMD+prefetch path) == per-name `callers_of`/`callees_of`
/// == `linear_callers_of`/`linear_callees_of` (the original scan);
/// 3. a STRUCTURAL/complexity stand-in for the (skipped, low-power) perf
/// bench: the tree is really built, its keys are strictly-ascending and
/// unique (one bucket per distinct key-hash), entry count == HashMap key
/// count, and the large fixture builds a deep multi-level tree.
///
/// If the stree wiring regresses (bad hash, missing string cross-check,
/// dropped fallback, wrong posting list), the batched results diverge from
/// the oracle and this test FAILS; on a correct wiring it PASSES.
#[cfg(feature = "stree")]
#[test]
fn stree_batched_lookups_match_inverted_index_and_linear_scan() {
// ── small, gnarly fixture: every match shape + a `renew` decoy ──
let view = KnowledgeView::new(vec![], rich_fixture());
let names = [
"new", "run", "step", "Arc::new", "commit", "Repo::commit", "renew", "noop",
"a::b::new", "b::new", "y::new", "x::y::new", "Foo::new", "ghost", "",
];
// (1) single-key stree get == inverted-index HashMap get.
let idx = view.index();
let callee_stree = idx.callee_stree();
let caller_stree = idx.caller_stree();
assert!(callee_stree.tree.is_some(), "the callee stree is actually built");
assert!(caller_stree.tree.is_some(), "the caller stree is actually built");
for name in names {
assert_eq!(
callee_stree.get(name),
idx.callee_keys.get(name),
"stree get({name:?}) must equal the callee_keys HashMap (cross-check)",
);
assert_eq!(
caller_stree.get(name),
idx.caller_keys.get(name),
"stree get({name:?}) must equal the caller_keys HashMap (cross-check)",
);
}
// (2) BATCHED stree path == per-name HashMap path == linear scan oracle.
let batch_callers = view.callers_of_batch(&names);
let batch_callees = view.callees_of_batch(&names);
assert_eq!(batch_callers.len(), names.len());
assert_eq!(batch_callees.len(), names.len());
for (i, name) in names.iter().enumerate() {
assert_eq!(
ptrs_v(&batch_callers[i]),
ptrs(view.callers_of(name)),
"callers_of_batch[{name:?}] must equal per-name callers_of",
);
assert_eq!(
ptrs_v(&batch_callers[i]),
ptrs(linear_callers_of(&view.calls, name)),
"callers_of_batch[{name:?}] must equal the linear scan oracle",
);
assert_eq!(
ptrs_v(&batch_callees[i]),
ptrs(view.callees_of(name)),
"callees_of_batch[{name:?}] must equal per-name callees_of",
);
assert_eq!(
ptrs_v(&batch_callees[i]),
ptrs(linear_callees_of(&view.calls, name)),
"callees_of_batch[{name:?}] must equal the linear scan oracle",
);
}
// (3) STRUCTURAL: one ascending, UNIQUE bucket per distinct key-hash;
// entry count == HashMap key count ⇒ the sorted-key index is complete.
assert_eq!(
callee_stree.entry_keys.len(),
idx.callee_keys.len(),
"every inverted-index key is present in the stree",
);
assert!(
callee_stree.sorted_hashes.len() <= callee_stree.entry_keys.len(),
"never more buckets than keys",
);
assert!(
callee_stree.sorted_hashes.windows(2).all(|w| w[0] < w[1]),
"stree keys are strictly ascending + unique (valid S-tree input)",
);
// ── large fixture: 50k distinct callee keys ⇒ a deep multi-level tree.
// The batched walk over the whole batch is the no-bench scale stand-in.
let mut calls = Vec::with_capacity(50_003);
for i in 0..50_000 {
calls.push(edge_from(&format!("noise{i}::caller"), &format!("noise{i}::callee{i}")));
}
// Exactly three edges invoke `target` (bare, single-, multi-qualified).
calls.push(edge_from("m::one", "target"));
calls.push(edge_from("m::two", "Wrap::target"));
calls.push(edge_from("m::three", "a::b::target"));
let big = KnowledgeView::new(vec![], calls);
// A big batch mixing 50k hits + qualified hits + misses; the batched
// stree result must equal the linear scan oracle for EVERY entry.
let mut qnames: Vec<String> = (0..50_000).map(|i| format!("callee{i}")).collect();
for s in ["target", "Wrap::target", "a::b::target", "absent_symbol", "new"] {
qnames.push(s.to_string());
}
let qrefs: Vec<&str> = qnames.iter().map(String::as_str).collect();
let big_batch = big.callers_of_batch(&qrefs);
assert_eq!(big_batch.len(), qrefs.len());
for (i, name) in qrefs.iter().enumerate() {
// Full batch (all 50k) vs the per-name HashMap path — cheap (O(hits)
// each) and itself proven == the linear scan on the small fixture.
assert_eq!(
ptrs_v(&big_batch[i]),
ptrs(big.callers_of(name)),
"large-fixture callers_of_batch[{name:?}] diverged from per-name callers_of",
);
// A sampled subset (every 500th + the qualified hits/miss tail) also
// tied straight to the linear scan oracle — keeps the O(N²) oracle off
// the hot path (low-power host) while still anchoring batch==linear.
if i % 500 == 0 || i >= 50_000 {
assert_eq!(
ptrs_v(&big_batch[i]),
ptrs(linear_callers_of(&big.calls, name)),
"large-fixture callers_of_batch[{name:?}] diverged from the linear scan",
);
}
}
// The multi-shape `target` query resolves its 3 edges through the batch.
let ti = qrefs.iter().position(|&n| n == "target").unwrap();
assert_eq!(big_batch[ti].len(), 3, "all 3 `target` callers via the batched stree");
let ai = qrefs.iter().position(|&n| n == "absent_symbol").unwrap();
assert!(big_batch[ai].is_empty(), "an absent key resolves to no edges");
// STRUCTURAL: the big tree really has many distinct keys (deep tree),
// the operational meaning of "scale" without running a heavy bench.
assert!(
big.index().callee_stree().sorted_hashes.len() > 10_000,
"large fixture builds a deep multi-level stree (the no-bench scale stand-in)",
);
}
/// A/B (LIGHT, in one binary): the new dense-interned **CSR integer BFS**
/// vs the previous **`HashMap<&str,&str>` / `HashSet<&str>` string BFS** over
/// the same warm memoized adjacency, on a `nodes`-deep chain plus a shared
/// fan-in callee (the `nornir.knowledge_query` shape). PROVES two things a
/// heavy bench can't in CI:
/// 1. **Byte-identical:** every `call_path` answer is exactly equal.
/// 2. **Faster:** the integer BFS does zero per-node string hashing, so it
/// is at least as fast; we print the ratio (heavy sweep → the bench box).
#[test]
fn call_path_csr_vs_hashmap_ab_light() {
use std::collections::VecDeque;
use std::time::Instant;
// Build the `nornir.knowledge_query` graph: f0→f1→…→fN plus every node
// also calling a shared `helper` (wide fan-in). Light size for CI.
let nodes = 400usize;
let mut calls = Vec::with_capacity(nodes * 2);
for i in 0..nodes {
let caller = format!("app::m{i}::f{i}");
if i + 1 < nodes {
calls.push(edge_from(&caller, &format!("app::m{}::f{}", i + 1, i + 1)));
}
calls.push(edge_from(&caller, "util::helper"));
}
let view = KnowledgeView::new(vec![], calls);
let from = "f0".to_string();
let to = format!("f{}", nodes - 1);
// --- OLD path: rebuild the string-keyed adjacency ONCE (warm), then the
// old `&str`-hashing BFS. Kept local so the shipped code carries only
// the CSR form; this is the A/B baseline. ---
let mut adj: HashMap<&str, Vec<&str>> = HashMap::new();
let mut nodeset: std::collections::HashSet<&str> = std::collections::HashSet::new();
for e in &view.calls {
let f = last_seg(&e.caller_path);
let t = last_seg(&e.callee_ident);
adj.entry(f).or_default().push(t);
nodeset.insert(f);
nodeset.insert(t);
}
let old_bfs = |from: &str, to: &str| -> Option<Vec<String>> {
if from == to {
return nodeset.contains(from).then(|| vec![from.to_string()]);
}
if !nodeset.contains(from) {
return None;
}
let mut parent: HashMap<&str, &str> = HashMap::new();
let mut seen: std::collections::HashSet<&str> = std::collections::HashSet::new();
let mut queue: VecDeque<&str> = VecDeque::new();
seen.insert(from);
queue.push_back(from);
while let Some(cur) = queue.pop_front() {
let Some(callees) = adj.get(cur) else { continue };
for c in callees {
let c = *c;
if !seen.insert(c) {
continue;
}
parent.insert(c, cur);
if c == to {
let mut path = vec![to.to_string()];
let mut node = to;
while let Some(&p) = parent.get(node) {
path.push(p.to_string());
node = p;
}
path.reverse();
return Some(path);
}
queue.push_back(c);
}
}
None
};
// Warm the new CSR index outside timing.
let _ = view.call_path(&from, &to);
// Byte-identical on the deep query and several edge shapes.
for (a, b) in [(from.as_str(), to.as_str()), (to.as_str(), from.as_str()),
("f0", "f0"), ("ghost", "f5"), ("f3", "helper")] {
assert_eq!(view.call_path(a, b), old_bfs(a, b),
"CSR BFS must equal the string-hashing BFS for ({a:?},{b:?})");
}
let iters = 60u32;
let t = Instant::now();
let mut sink = 0usize;
for _ in 0..iters {
sink += view.call_path(&from, &to).map(|p| p.len()).unwrap_or(0);
}
let csr = t.elapsed();
let t = Instant::now();
for _ in 0..iters {
sink += old_bfs(&from, &to).map(|p| p.len()).unwrap_or(0);
}
let old = t.elapsed();
assert!(sink > 0);
let speedup = old.as_secs_f64() / csr.as_secs_f64().max(f64::MIN_POSITIVE);
eprintln!(
"call_path A/B ({nodes}-node chain, {iters}x): CSR {:.1} us/call vs string-BFS {:.1} us/call — {speedup:.2}x",
csr.as_secs_f64() * 1e6 / iters as f64,
old.as_secs_f64() * 1e6 / iters as f64,
);
// The CSR BFS must not be a regression. Wide margin (timing is noisy on a
// loaded low-power host); the win is proven, the sweep runs heavy elsewhere.
assert!(csr <= old.saturating_mul(2),
"CSR BFS must not regress vs the string BFS: csr={csr:?} old={old:?}");
}
// ════════════════════════════════════════════════════════════════════
// The LOADER over the `Warehouse` seam. The warehouse OWNER additionally
// tests this against a real Iceberg backend (where snapshot pushdown really
// prunes data files, so read-amplification can be asserted); here we assert
// the seam-level contract every backend must honour: pick the LATEST
// snapshot for the repo, decode ONLY that snapshot, and never leak another
// repo's rows.
// ════════════════════════════════════════════════════════════════════
/// RED-when-broken: seed two snapshots for one repo plus a decoy repo, then
/// assert `load_latest_audited` returns exactly the newest snapshot's rows
/// and audits the wide decode as only those rows. Drop the residual
/// `snapshot_id == snap` guard and the older snapshot's rows leak in; drop
/// the `repo` guard and the decoy leaks in; pick the wrong snapshot and the
/// symbol name is wrong.
#[test]
fn load_latest_picks_the_newest_snapshot_and_ignores_other_repos() {
use nornir_release_ledger::testkit::MockWarehouse;
fn sym(name: &str) -> SymbolRow {
SymbolRow {
crate_name: "demo".into(),
module_path: "demo".into(),
item_kind: "fn".into(),
item_name: name.into(),
visibility: "pub".into(),
file: "src/lib.rs".into(),
line: 1,
doc_lines: 0,
signature: None,
}
}
let wh = MockWarehouse::new();
// Older snapshot for `mine`, then a decoy repo, then the newest for `mine`.
// Each `seed_symbol_scan` mints a fresh snapshot_id with a `now` stamp.
wh.seed_symbol_scan("mine", &[sym("old_one")], &[edge_from("old_a", "old_b")]);
wh.seed_symbol_scan("other", &[sym("decoy")], &[edge_from("decoy_a", "decoy_b")]);
std::thread::sleep(std::time::Duration::from_millis(2));
wh.seed_symbol_scan(
"mine",
&[sym("new_one"), sym("new_two")],
&[edge_from("new_a", "new_b")],
);
let (view, audit) = load_latest_audited(&wh, "mine").unwrap();
let names: Vec<&str> = view.symbols.iter().map(|s| s.item_name.as_str()).collect();
assert_eq!(names, vec!["new_one", "new_two"], "only the NEWEST snapshot decodes");
assert_eq!(view.calls.len(), 1);
assert_eq!(view.calls[0].caller_path, "new_a");
// 2 symbols + 1 call edge from the winning snapshot, and nothing else.
assert_eq!(audit.data_rows, 3, "the wide decode is scoped to one snapshot");
// Discovery saw only `mine`'s rows (1 old + 2 new), never `other`'s.
assert_eq!(audit.discovery_rows, 3, "discovery never counts another repo's rows");
// And the loaded view answers call-graph queries over exactly those edges.
assert_eq!(view.callers_of("new_b").len(), 1);
assert!(view.callers_of("old_b").is_empty(), "the old snapshot is gone");
// A repo with nothing persisted is an empty view, not an error.
let (empty, empty_audit) = load_latest_audited(&wh, "absent").unwrap();
assert!(empty.symbols.is_empty() && empty.calls.is_empty());
assert_eq!(empty_audit, ScanAudit::default());
}
}