crates/mimir-ingest/src/delta.rs
//! The delta computer and the open set.
//!
//! A row's identity in the log is (owner crate, key): the crate whose dump
//! emitted it, and the payload's key columns. Owner is part of the identity
//! because one crate's dump legitimately carries rows keyed like another
//! crate's (reference stubs of items it points at), and "crate no longer
//! present → close all its keys" needs the owner anyway.
//!
//! The open set is the fold of the event log: per repo, per (owner, key), the
//! payload hash of its last `+` event (an identity whose last event is `-`
//! is not in it). It is a cache DERIVED from the log — rebuilt by
//! [`crate::EventLog::fold`] — never a second truth.
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::sync::Arc;
use std::hash::BuildHasher;
use hashbrown::{DefaultHashBuilder, HashTable};
use anyhow::{Context, Result, bail};
use arrow_row::{RowConverter, SortField};
use skade::arrow_array::{Array, ArrayRef, Int64Array, RecordBatch, StringArray, new_null_array};
use skade::arrow_schema::DataType;
use crate::atoms::Atoms;
use crate::schema::{DUP_MARK, KEY_NULL, KEY_SEP, OP_MINUS, OP_PLUS, PART_SEP, PayloadSpec, storage_type};
/// 128 bits of blake3 over the arrow-row encoding of a payload row.
pub type PayloadHash = [u8; 16];
/// The open identities of one repo for one table (nodes or edges):
/// owner crate → key → payload hash.
///
/// A key is not kept as its text but as the ids of its interned parts
/// ([`crate::atoms`]) — 2.6 GB of key text on dwarf was most of the server's
/// resident set. A crate's identities are shared with every clone of the set
/// until one side changes them (copy on write): the base rewrite writes a
/// clone while the consumer goes on applying.
#[derive(Clone)]
pub struct RepoOpen {
atoms: Arc<Atoms>,
by_krate: HashMap<String, Arc<KrateOpen>>,
}
/// One owner crate's open identities: the part ids of every key back to
/// back in `ids`, a table of (where its ids are, payload hash).
#[derive(Clone, Default)]
struct KrateOpen {
hasher: DefaultHashBuilder,
ids: Vec<u32>,
table: HashTable<Slot>,
/// Words of `ids` no slot points at any more (closed keys).
dead: usize,
}
#[derive(Clone, Copy)]
struct Slot {
at: u32,
len: u8,
hash: PayloadHash,
}
impl KrateOpen {
fn key(&self, s: &Slot) -> &[u32] {
&self.ids[s.at as usize..s.at as usize + s.len as usize]
}
fn get(&self, key: &[u32]) -> Option<&PayloadHash> {
let h = self.hasher.hash_one(key);
self.table.find(h, |s| self.key(s) == key).map(|s| &s.hash)
}
fn open(&mut self, key: &[u32], hash: PayloadHash) {
let h = self.hasher.hash_one(key);
let KrateOpen { hasher, ids, table, .. } = self;
if let Some(s) = table.find_mut(h, |s| &ids[s.at as usize..s.at as usize + s.len as usize] == key) {
s.hash = hash;
return;
}
let at = u32::try_from(ids.len()).expect("open set: one crate's keys past 2^32 words");
ids.extend_from_slice(key);
let ids = &*ids;
table.insert_unique(h, Slot { at, len: key.len() as u8, hash }, |s| hasher.hash_one(&ids[s.at as usize..s.at as usize + s.len as usize]));
}
fn close(&mut self, key: &[u32]) {
let h = self.hasher.hash_one(key);
let KrateOpen { ids, table, dead, .. } = self;
if let Ok(e) = table.find_entry(h, |s| &ids[s.at as usize..s.at as usize + s.len as usize] == key) {
let (s, _) = e.remove();
*dead += s.len as usize;
}
// the words of closed keys: given back once they are most of them
if self.dead > 1024 && self.dead * 2 > self.ids.len() {
self.compact();
}
}
/// `ids` again with only the keys still open.
fn compact(&mut self) {
let mut ids = Vec::with_capacity(self.ids.len() - self.dead);
for s in self.table.iter_mut() {
let at = ids.len() as u32;
ids.extend_from_slice(&self.ids[s.at as usize..s.at as usize + s.len as usize]);
s.at = at;
}
self.ids = ids;
self.dead = 0;
}
fn shrink(&mut self) {
if self.dead > 0 {
self.compact();
}
let KrateOpen { hasher, ids, table, .. } = self;
ids.shrink_to_fit();
let ids = &*ids;
table.shrink_to_fit(|s| hasher.hash_one(&ids[s.at as usize..s.at as usize + s.len as usize]));
}
fn heap_bytes(&self) -> usize {
self.ids.capacity() * 4 + self.table.capacity() * (std::mem::size_of::<Slot>() + 1)
}
}
impl RepoOpen {
fn new(atoms: Arc<Atoms>) -> Self {
Self { atoms, by_krate: HashMap::new() }
}
pub fn len(&self) -> usize {
self.by_krate.values().map(|k| k.table.len()).sum()
}
pub fn is_empty(&self) -> bool {
self.by_krate.is_empty()
}
/// The payload hash of `key` owned by `krate`, if open. Multi-column keys
/// are joined with U+001F.
pub fn get(&self, krate: &str, key: &str) -> Option<&PayloadHash> {
let k = self.by_krate.get(krate)?;
k.get(self.atoms.find(key)?.as_slice())
}
/// Every open (owner, key, hash). The key is spelled out for each row.
pub fn iter(&self) -> impl Iterator<Item = (&String, String, &PayloadHash)> {
self.by_krate.iter().flat_map(move |(krate, k)| k.table.iter().map(move |s| (krate, self.atoms.resolve(k.key(s)), &s.hash)))
}
pub fn krates(&self) -> impl Iterator<Item = &String> {
self.by_krate.keys()
}
/// The open (key, hash) of `krate`, the key spelled out.
pub fn rows_of(&self, krate: &str) -> impl Iterator<Item = (String, &PayloadHash)> {
self.by_krate.get(krate).into_iter().flat_map(move |k| k.table.iter().map(move |s| (self.atoms.resolve(k.key(s)), &s.hash)))
}
/// The open keys of `krate`, spelled out.
pub fn keys_of(&self, krate: &str) -> impl Iterator<Item = String> {
self.by_krate.get(krate).into_iter().flat_map(move |k| k.table.iter().map(move |s| self.atoms.resolve(k.key(s))))
}
pub(crate) fn open(&mut self, krate: &str, key: impl AsRef<str>, hash: PayloadHash) {
let ids = self.atoms.intern(key.as_ref());
if !self.by_krate.contains_key(krate) {
self.by_krate.insert(krate.to_string(), Arc::default());
}
let k = self.by_krate.get_mut(krate).expect("inserted above");
Arc::make_mut(k).open(ids.as_slice(), hash);
}
pub(crate) fn close(&mut self, krate: &str, key: &str) {
let Some(ids) = self.atoms.find(key) else { return };
if let Some(k) = self.by_krate.get_mut(krate) {
if k.get(ids.as_slice()).is_none() {
return;
}
let k = Arc::make_mut(k);
k.close(ids.as_slice());
if k.table.is_empty() {
self.by_krate.remove(krate);
}
}
}
fn shrink(&mut self) {
for k in self.by_krate.values_mut() {
Arc::make_mut(k).shrink();
}
self.by_krate.shrink_to_fit();
}
fn heap_bytes(&self) -> usize {
self.by_krate.iter().map(|(n, k)| n.capacity() + std::mem::size_of::<KrateOpen>() + k.heap_bytes()).sum()
}
}
/// The same identities with the same payloads (whatever the interning).
impl PartialEq for RepoOpen {
fn eq(&self, other: &Self) -> bool {
if self.by_krate.len() != other.by_krate.len() {
return false;
}
self.by_krate.iter().all(|(krate, a)| {
let Some(b) = other.by_krate.get(krate) else { return false };
if Arc::ptr_eq(a, b) && Arc::ptr_eq(&self.atoms, &other.atoms) {
return true;
}
a.table.len() == b.table.len()
&& a.table.iter().all(|s| {
let key = self.atoms.resolve(a.key(s));
other.atoms.find(&key).and_then(|ids| b.get(ids.as_slice())) == Some(&s.hash)
})
})
}
}
impl Eq for RepoOpen {}
impl std::fmt::Debug for RepoOpen {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let mut rows: Vec<(&String, String, &PayloadHash)> = self.iter().collect();
rows.sort();
f.debug_list().entries(rows).finish()
}
}
/// Which event table a payload belongs to.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub enum Kind {
// source stream
Nodes,
Edges,
Spans,
// binary stream
BinNodes,
BinEdges,
Symbols,
}
/// The producer stream that owns an event table (LAW 5: one writer per table).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum Stream {
/// mimir-driver's source dumps.
Source,
/// mimir-binary over linked artifacts.
Binary,
}
impl Kind {
/// Every event table, in commit order.
pub const ALL: [Kind; 6] = [Kind::Nodes, Kind::Edges, Kind::Spans, Kind::BinNodes, Kind::BinEdges, Kind::Symbols];
/// The source stream's tables.
pub const SOURCE: [Kind; 3] = [Kind::Nodes, Kind::Edges, Kind::Spans];
/// The binary stream's tables.
pub const BINARY: [Kind; 3] = [Kind::BinNodes, Kind::BinEdges, Kind::Symbols];
/// The ONE stream that writes this table.
pub fn stream(self) -> Stream {
match self {
Kind::Nodes | Kind::Edges | Kind::Spans => Stream::Source,
Kind::BinNodes | Kind::BinEdges | Kind::Symbols => Stream::Binary,
}
}
pub(crate) fn idx(self) -> usize {
self as usize
}
}
/// The whole open set: per table kind, per repo. Its keys are interned in
/// ONE table of atoms shared by its repos (and its clones).
#[derive(Clone, Default)]
pub struct OpenSet {
atoms: Arc<Atoms>,
maps: [BTreeMap<String, RepoOpen>; 6],
}
impl PartialEq for OpenSet {
fn eq(&self, other: &Self) -> bool {
self.maps == other.maps
}
}
impl Eq for OpenSet {}
impl std::fmt::Debug for OpenSet {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("OpenSet").field("maps", &self.maps).finish()
}
}
impl OpenSet {
pub fn repo(&self, kind: Kind, repo: &str) -> Option<&RepoOpen> {
self.map(kind).get(repo)
}
/// Every open identity, all tables and repos.
pub fn rows(&self) -> usize {
self.maps.iter().flat_map(BTreeMap::values).map(RepoOpen::len).sum()
}
/// Bytes the set holds on the heap (capacities: what it costs, not what
/// it uses), the interned atoms included.
pub fn heap_bytes(&self) -> usize {
self.atoms.heap_bytes() + self.maps.iter().flat_map(BTreeMap::iter).map(|(r, o)| r.capacity() + o.heap_bytes()).sum::<usize>()
}
/// Give back what the tables grew past their use — after a load or a
/// fold, which grow them a row at a time.
pub fn shrink(&mut self) {
self.atoms.shrink();
for o in self.maps.iter_mut().flat_map(BTreeMap::values_mut) {
o.shrink();
}
}
pub(crate) fn map(&self, kind: Kind) -> &BTreeMap<String, RepoOpen> {
&self.maps[kind.idx()]
}
pub(crate) fn repo_mut(&mut self, kind: Kind, repo: &str) -> &mut RepoOpen {
let atoms = &self.atoms;
self.maps[kind.idx()].entry(repo.to_string()).or_insert_with(|| RepoOpen::new(atoms.clone()))
}
/// Apply a computed delta (after its commit landed).
pub(crate) fn apply(&mut self, kind: Kind, repo: &str, delta: &KindDelta) {
let open = self.repo_mut(kind, repo);
for p in &delta.plus {
open.open(&p.krate, &p.key, p.hash);
}
for m in &delta.minus {
open.close(&m.krate, &m.key);
}
}
/// Apply one folded event.
pub(crate) fn apply_event(&mut self, kind: Kind, repo: &str, op: &str, krate: &str, key: String, hash: Option<PayloadHash>) -> Result<()> {
let open = self.repo_mut(kind, repo);
match (op, hash) {
(OP_PLUS, Some(hash)) => open.open(krate, key, hash),
(OP_PLUS, None) => bail!("`+` event for {key:?} in {repo} has no payload_hash"),
(OP_MINUS, _) => open.close(krate, &key),
(other, _) => bail!("unknown op {other:?} in the event log"),
}
Ok(())
}
}
/// A `+` event to write: payload row `row` of conformed batch `batch`.
#[derive(Debug, Clone)]
pub(crate) struct Plus {
pub krate: String,
pub key: String,
pub hash: PayloadHash,
pub batch: usize,
pub row: usize,
}
/// A `-` event to write.
#[derive(Debug, Clone)]
pub(crate) struct Minus {
pub krate: String,
pub key: String,
}
/// The delta of one table kind for one repo at one commit.
#[derive(Debug, Default)]
pub(crate) struct KindDelta {
/// The conformed payload batches `plus` rows point into.
pub batches: Vec<RecordBatch>,
pub plus: Vec<Plus>,
pub minus: Vec<Minus>,
/// Keys disambiguated by payload hash (see [`Keyed`]).
pub collisions: usize,
}
/// The rows of one dump: identity (owner, key) → (hash, batch, row).
///
/// A key repeated with the SAME payload collapses to one row. A key repeated
/// with DIFFERENT payloads (a producer collision, e.g. rust-analyzer-shaped
/// monikers that are not unique) is neither dropped nor fatal: each distinct
/// payload becomes its own identity, `key ⟨U+001F⟩ #<payload hash>` — the
/// suffix is stored in the event's `key_dup` column so the fold rebuilds the
/// same identity — and the collision is counted (`collisions`, ledger
/// `key_collisions`).
pub(crate) struct Keyed {
pub batches: Vec<RecordBatch>,
pub rows: BTreeMap<(String, String), (PayloadHash, usize, usize)>,
/// Keys that carried more than one distinct payload in this dump.
pub collisions: usize,
}
pub(crate) fn key_rows<'a>(
spec: &PayloadSpec,
per_crate: impl IntoIterator<Item = (&'a str, &'a [RecordBatch])>,
) -> Result<Keyed> {
let mut batches = Vec::new();
// (owner, key) → distinct payloads (hash → first row carrying it).
let mut seen: BTreeMap<(String, String), BTreeMap<PayloadHash, (usize, usize)>> = BTreeMap::new();
for (krate, crate_batches) in per_crate {
for b in crate_batches {
if b.num_rows() == 0 {
continue;
}
let b = spec.conform(b).with_context(|| format!("crate {krate}"))?;
let hashes = row_hashes(spec, &b)?;
let keys = row_keys(spec, &b)?;
let bi = batches.len();
for (ri, (key, hash)) in keys.into_iter().zip(hashes).enumerate() {
seen.entry((krate.to_string(), key)).or_default().entry(hash).or_insert((bi, ri));
}
batches.push(b);
}
}
let mut out = Keyed { batches, rows: BTreeMap::new(), collisions: 0 };
for ((krate, key), payloads) in seen {
if payloads.len() == 1 {
let (hash, (bi, ri)) = payloads.into_iter().next().expect("one payload");
out.rows.insert((krate, key), (hash, bi, ri));
continue;
}
out.collisions += 1;
for (hash, (bi, ri)) in payloads {
let id = format!("{key}{KEY_SEP}{DUP_MARK}{}", hex(&hash));
out.rows.insert((krate.clone(), id), (hash, bi, ri));
}
}
Ok(out)
}
/// The `key_dup` part of an identity (`#<hash>`), if it has one.
pub(crate) fn dup_part(spec: &PayloadSpec, identity: &str) -> Option<String> {
identity.split(KEY_SEP).nth(spec.key_columns.len()).map(str::to_string)
}
/// The delta of `keyed` (the rows of every LOCAL crate dumped this commit)
/// against `open`, the repo's open identities:
/// * an identity that is new, or whose payload hash changed → `+`;
/// * an open identity of a re-dumped crate the dump no longer has → `-`;
/// * every open identity of a crate in `gone` (left the workspace) → `-`.
///
/// Crates neither dumped nor gone keep their identities untouched.
pub(crate) fn delta(keyed: Keyed, dumped: &BTreeSet<String>, gone: &BTreeSet<String>, open: Option<&RepoOpen>) -> KindDelta {
let mut d = KindDelta { batches: keyed.batches, collisions: keyed.collisions, ..Default::default() };
for ((krate, key), (hash, batch, row)) in &keyed.rows {
if open.and_then(|o| o.get(krate, key)) != Some(hash) {
d.plus.push(Plus { krate: krate.clone(), key: key.clone(), hash: *hash, batch: *batch, row: *row });
}
}
if let Some(open) = open {
for krate in dumped.iter().chain(gone.iter()) {
let mut keys: Vec<String> = open
.keys_of(krate)
.map(|k| (krate.clone(), k))
.filter(|id| !keyed.rows.contains_key(id))
.map(|(_, k)| k)
.collect();
keys.sort();
d.minus.extend(keys.into_iter().map(|key| Minus { krate: krate.clone(), key }));
}
}
d
}
/// Every row of a universe crate as `+` (nothing is open for the universe).
pub(crate) fn universe_delta(keyed: Keyed) -> KindDelta {
let mut d = KindDelta { batches: keyed.batches, collisions: keyed.collisions, ..Default::default() };
for ((krate, key), (hash, batch, row)) in keyed.rows {
d.plus.push(Plus { krate, key, hash, batch, row });
}
d
}
/// Blake3-128 of each row's canonical arrow-row encoding — over the columns
/// the lake's table was created with. A column added in place later
/// ([`PayloadSpec::added`]) joins a row's hash only where the row HAS a value
/// for it (name + value, in name order): null there is the same as the column
/// not existing, so the rows written before the evolution keep their hashes.
pub(crate) fn row_hashes(spec: &PayloadSpec, batch: &RecordBatch) -> Result<Vec<PayloadHash>> {
let schema = batch.schema();
let (mut base, mut added) = (Vec::new(), Vec::new());
for (i, f) in schema.fields().iter().enumerate() {
if spec.added.iter().any(|a| a == f.name()) {
added.push(i);
} else {
base.push(i);
}
}
added.sort_by(|a, b| schema.field(*a).name().cmp(schema.field(*b).name()));
let encode = |cols: &[usize]| -> Result<arrow_row::Rows> {
let fields: Vec<SortField> = cols.iter().map(|i| SortField::new(schema.field(*i).data_type().clone())).collect();
let conv = RowConverter::new(fields).context("payload row encoding")?;
let cols: Vec<ArrayRef> = cols.iter().map(|i| batch.column(*i).clone()).collect();
conv.convert_columns(&cols).context("encode payload rows")
};
let rows = encode(&base)?;
let late: Vec<(usize, arrow_row::Rows)> = added.iter().map(|i| Ok((*i, encode(&[*i])?))).collect::<Result<_>>()?;
Ok((0..batch.num_rows())
.map(|r| {
let mut h = blake3::Hasher::new();
h.update(rows.row(r).as_ref());
for (i, values) in &late {
if batch.column(*i).is_null(r) {
continue;
}
let (name, value) = (schema.field(*i).name().as_bytes(), values.row(r));
h.update(&[KEY_SEP as u8]);
h.update(&(name.len() as u32).to_le_bytes());
h.update(name);
h.update(&(value.as_ref().len() as u32).to_le_bytes());
h.update(value.as_ref());
}
let mut out = [0u8; 16];
out.copy_from_slice(&h.finalize().as_bytes()[..16]);
out
})
.collect())
}
/// Key of each row: the key columns rendered as text (null = U+0000),
/// joined with U+001F.
fn row_keys(spec: &PayloadSpec, batch: &RecordBatch) -> Result<Vec<String>> {
let cols: Vec<StringArray> = spec
.key_columns
.iter()
.map(|k| {
let c = batch.column_by_name(k).with_context(|| format!("key column `{k}` missing"))?;
let c = skade::arrow_cast::cast(c, &DataType::Utf8).with_context(|| format!("key column `{k}` as text"))?;
Ok(c.as_any().downcast_ref::<StringArray>().context("utf8 downcast")?.clone())
})
.collect::<Result<_>>()?;
let mut out = Vec::with_capacity(batch.num_rows());
for row in 0..batch.num_rows() {
let mut key = String::new();
for (i, c) in cols.iter().enumerate() {
if i > 0 {
key.push(KEY_SEP);
}
if c.is_null(row) {
key.push(KEY_NULL);
continue;
}
let v = c.value(row);
if v.contains(KEY_SEP) || v.contains(KEY_NULL) {
bail!("key column `{}` value {v:?} contains U+001F or U+0000", spec.key_columns[i]);
}
if spec.stable_keys {
push_stable(&mut key, v);
} else {
key.push_str(v);
}
}
out.push(key);
}
Ok(out)
}
/// A key part made STABLE ([`PayloadSpec::stable_keys`]): every moniker in it
/// — the value itself, or each U+001E-joined piece of a composite value (a
/// span's edge key) — keyed with its version as `*`. Anything that is not a
/// `rust-analyzer cargo` moniker is appended byte for byte.
pub(crate) fn push_stable(key: &mut String, value: &str) {
for (i, piece) in value.split(PART_SEP).enumerate() {
if i > 0 {
key.push(PART_SEP);
}
key.push_str(&mimir_schema::moniker::stable(piece));
}
}
/// [`push_stable`] appending bytes — the fold's side of the same rule, so an
/// identity hashed from stored columns (which keep their real versions)
/// matches the key the delta wrote.
pub(crate) fn stable_into(value: &str, buf: &mut Vec<u8>) {
let mut ch = [0u8; 4];
for (i, piece) in value.split(PART_SEP).enumerate() {
if i > 0 {
buf.extend_from_slice(PART_SEP.encode_utf8(&mut ch).as_bytes());
}
buf.extend_from_slice(mimir_schema::moniker::stable(piece).as_bytes());
}
}
pub(crate) fn hex(h: &PayloadHash) -> String {
const D: &[u8; 16] = b"0123456789abcdef";
let mut s = String::with_capacity(32);
for b in h {
s.push(D[(b >> 4) as usize] as char);
s.push(D[(b & 0xf) as usize] as char);
}
s
}
pub(crate) fn unhex(s: &str) -> Result<PayloadHash> {
let b = s.as_bytes();
if b.len() != 32 {
bail!("payload_hash {s:?} is not 32 hex digits");
}
let nib = |c: u8| -> Result<u8> {
Ok(match c {
b'0'..=b'9' => c - b'0',
b'a'..=b'f' => c - b'a' + 10,
_ => bail!("payload_hash {s:?} is not lowercase hex"),
})
};
let mut out = [0u8; 16];
for (i, o) in out.iter_mut().enumerate() {
*o = (nib(b[2 * i])? << 4) | nib(b[2 * i + 1])?;
}
Ok(out)
}
/// The event rows of one delta as ONE batch in `spec.event_schema()`: `+`
/// rows (identity order) then `-` rows (identity order). A `-` row carries its
/// key columns (cast back from their text form) and nulls elsewhere. `rev` =
/// a binary-stream table (carries the binary release ordinal).
pub(crate) fn event_batch(spec: &PayloadSpec, repo: &str, seq: i64, rev: Option<i64>, sha: &str, d: &KindDelta) -> Result<Option<RecordBatch>> {
let n = d.plus.len() + d.minus.len();
if n == 0 {
return Ok(None);
}
let schema = Arc::new(spec.event_schema_with(rev.is_some()));
let payload_fields = spec.schema.fields();
let plus_payload: Option<RecordBatch> = if d.plus.is_empty() {
None
} else {
let refs: Vec<&RecordBatch> = d.batches.iter().collect();
let idx: Vec<(usize, usize)> = d.plus.iter().map(|p| (p.batch, p.row)).collect();
Some(arrow_select::interleave::interleave_record_batch(&refs, &idx).context("gather + rows")?)
};
let mut payload_cols: Vec<ArrayRef> = Vec::with_capacity(payload_fields.len());
for (ci, f) in payload_fields.iter().enumerate() {
let minus_col: ArrayRef = match spec.key_columns.iter().position(|k| k == f.name()) {
Some(part) => {
let text: StringArray = d
.minus
.iter()
.map(|m| m.key.split(KEY_SEP).nth(part).filter(|p| *p != "\u{0}"))
.collect();
skade::arrow_cast::cast(&text, f.data_type()).with_context(|| format!("key column `{}` back from text", f.name()))?
}
None => new_null_array(f.data_type(), d.minus.len()),
};
let col = match &plus_payload {
Some(p) => arrow_select::concat::concat(&[p.column(ci).as_ref(), minus_col.as_ref()])?,
None => minus_col,
};
let stored = storage_type(f.data_type());
let col = if &stored == f.data_type() { col } else { skade::arrow_cast::cast(&col, &stored)? };
payload_cols.push(col);
}
let ops = d.plus.iter().map(|_| OP_PLUS).chain(d.minus.iter().map(|_| OP_MINUS));
let krates = d.plus.iter().map(|p| p.krate.as_str()).chain(d.minus.iter().map(|m| m.krate.as_str()));
let hashes = d.plus.iter().map(|p| Some(hex(&p.hash))).chain(d.minus.iter().map(|_| None));
let mut cols: Vec<ArrayRef> = vec![
Arc::new(StringArray::from_iter_values(std::iter::repeat_n(repo, n))),
Arc::new(Int64Array::from(vec![seq; n])),
Arc::new(StringArray::from_iter_values(std::iter::repeat_n(sha, n))),
Arc::new(StringArray::from_iter_values(ops)),
Arc::new(StringArray::from_iter_values(krates)),
Arc::new(hashes.collect::<StringArray>()),
Arc::new(
d.plus
.iter()
.map(|p| dup_part(spec, &p.key))
.chain(d.minus.iter().map(|m| dup_part(spec, &m.key)))
.collect::<StringArray>(),
),
];
if let Some(rev) = rev {
cols.push(Arc::new(Int64Array::from(vec![rev; n])));
}
cols.extend(payload_cols);
Ok(Some(RecordBatch::try_new(schema, cols).context("assemble event batch")?))
}
#[cfg(test)]
mod tests {
use super::*;
/// **What the open set costs a row.** dwarf 2026-10-16: 14 M rows held
/// 3.6 GB — the keys as text, 2.6 GB of it the same monikers over and
/// over. A set of moniker-shaped rows like the live one (nodes, edges
/// `from␟to␟kind`, spans `edge␟from␞to␞kind␟file␟l␟l`; the same monikers
/// in many keys) must hold under 100 bytes a row, counted at the
/// allocator on this thread. Keys as `String`s held 304.
#[test]
fn the_open_set_holds_a_row_in_under_100_bytes() {
let mut seed = 0x9e37_79b9_7f4a_7c15u64;
let mut rnd = move |n: usize| {
seed = seed.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407);
(seed >> 33) as usize % n
};
let crates = ["nordisk-pairing", "nornir-ingest", "skade-katalog", "facett-graphview", "dwarves-embed", "knut-rows", "edda-jera", "korp-collect"];
let modules = ["bootlog", "installer/wire", "receiver", "cache/journal", "room/snapshot", "delta", "consumer/loop", "schema"];
let kinds = ["CALLS", "USES_TYPE", "READS_FIELD", "IMPLEMENTS"];
let mut monikers: Vec<(usize, String)> = Vec::new();
for (c, krate) in crates.iter().enumerate() {
for m in 0..2500 {
let module = modules[rnd(modules.len())];
let item = match rnd(3) {
0 => format!("{module}/impl#[Type{}][Trait{}]method_{m}().", rnd(200), rnd(40)),
1 => format!("{module}/Type{}#field_{m}.", rnd(300)),
_ => format!("{module}/function_{m}()."),
};
monikers.push((c, format!("rust-analyzer cargo {krate} 0.{c}.{} {item}", rnd(20))));
}
}
for i in 0..200 {
monikers.push((usize::MAX, format!("rust-analyzer cargo core https://github.com/rust-lang/rust/library/core module_{}/Item{i}#", i % 17)));
}
let owner = |c: usize| format!("{}@0.1.0:lib", crates[c]);
let before = crate::alloc_count::held();
let mut set = OpenSet::default();
let mut hash = [0u8; 16];
for (i, (c, m)) in monikers.iter().enumerate() {
if *c == usize::MAX {
continue;
}
hash[..8].copy_from_slice(&(i as u64).to_le_bytes());
set.apply_event(Kind::Nodes, "nordisk", OP_PLUS, &owner(*c), m.clone(), Some(hash)).unwrap();
for e in 0..3 {
let to = &monikers[rnd(monikers.len())].1;
let kind = kinds[rnd(kinds.len())];
hash[8] = e;
set.apply_event(Kind::Edges, "nordisk", OP_PLUS, &owner(*c), format!("{m}\u{1f}{to}\u{1f}{kind}"), Some(hash)).unwrap();
let line = rnd(900) + 1;
let span = format!("edge\u{1f}{m}\u{1e}{to}\u{1e}{kind}\u{1f}crates/{}/src/m{}.rs\u{1f}{line}\u{1f}{line}", crates[*c], rnd(8));
set.apply_event(Kind::Spans, "nordisk", OP_PLUS, &owner(*c), span, Some(hash)).unwrap();
}
}
drop(monikers);
let held = crate::alloc_count::held() - before;
// (an edge drawn twice is one row)
let rows = set.rows();
assert!(rows > 130_000, "{rows} rows");
let per_row = held as f64 / rows as f64;
eprintln!("open set: {rows} rows hold {held} B: {per_row:.1} B a row");
assert!(per_row < 100.0, "the open set holds {per_row:.1} bytes a row ({rows} rows, {held} B) — the keys as text again?");
}
use skade::arrow_schema::{Field, Schema};
fn spec() -> PayloadSpec {
let s = Schema::new(vec![
Field::new("from", DataType::Utf8, false),
Field::new("line", DataType::UInt32, true),
Field::new("note", DataType::Utf8, true),
]);
PayloadSpec::new(Arc::new(s), &["from", "line"]).unwrap()
}
fn batch(from: &[&str], line: &[Option<u32>], note: &[&str]) -> RecordBatch {
RecordBatch::try_new(
Arc::new(spec().schema.as_ref().clone()),
vec![
Arc::new(StringArray::from(from.to_vec())),
Arc::new(skade::arrow_array::UInt32Array::from(line.to_vec())),
Arc::new(StringArray::from(note.to_vec())),
],
)
.unwrap()
}
#[test]
fn same_dump_against_its_own_fold_is_an_empty_delta() {
let b = vec![batch(&["x", "y"], &[Some(1), None], &["a", "b"])];
let dumped: BTreeSet<String> = ["k".to_string()].into();
let first = delta(key_rows(&spec(), [("k", b.as_slice())]).unwrap(), &dumped, &BTreeSet::new(), None);
assert_eq!((first.plus.len(), first.minus.len()), (2, 0));
let mut open = OpenSet::default();
open.apply(Kind::Edges, "r", &first);
let again = delta(key_rows(&spec(), [("k", b.as_slice())]).unwrap(), &dumped, &BTreeSet::new(), open.repo(Kind::Edges, "r"));
assert_eq!((again.plus.len(), again.minus.len()), (0, 0));
}
#[test]
fn minus_rows_carry_typed_keys_including_null() {
let b = vec![batch(&["x", "y"], &[Some(7), None], &["a", "b"])];
let dumped: BTreeSet<String> = ["k".to_string()].into();
let first = delta(key_rows(&spec(), [("k", b.as_slice())]).unwrap(), &dumped, &BTreeSet::new(), None);
let mut open = OpenSet::default();
open.apply(Kind::Edges, "r", &first);
// The crate left the workspace: both identities close.
let gone: BTreeSet<String> = ["k".to_string()].into();
let d = delta(key_rows(&spec(), std::iter::empty()).unwrap(), &BTreeSet::new(), &gone, open.repo(Kind::Edges, "r"));
let ev = event_batch(&spec(), "r", 2, None, "s", &d).unwrap().unwrap();
let line = ev.column_by_name("line").unwrap().as_any().downcast_ref::<skade::arrow_array::Int64Array>().unwrap();
let from = ev.column_by_name("from").unwrap().as_any().downcast_ref::<StringArray>().unwrap();
let note = ev.column_by_name("note").unwrap();
assert_eq!((from.value(0), line.value(0)), ("x", 7));
assert_eq!(from.value(1), "y");
assert!(line.is_null(1), "a null key part must come back null, not 0");
assert_eq!(note.null_count(), 2);
}
#[test]
fn a_key_twice_with_different_payloads_becomes_two_counted_identities() {
let b = vec![batch(&["x", "x", "x"], &[Some(1), Some(1), Some(1)], &["a", "b", "a"])];
let keyed = key_rows(&spec(), [("k", b.as_slice())]).unwrap();
assert_eq!(keyed.collisions, 1);
let ids: Vec<&String> = keyed.rows.keys().map(|(_, k)| k).collect();
assert_eq!(ids.len(), 2, "two distinct payloads, the repeated one collapsed");
assert!(ids.iter().all(|k| dup_part(&spec(), k).is_some_and(|d| d.starts_with('#'))));
let d = delta(keyed, &["k".to_string()].into(), &BTreeSet::new(), None);
let ev = event_batch(&spec(), "r", 1, None, "s", &d).unwrap().unwrap();
let dup = ev.column_by_name("key_dup").unwrap();
assert_eq!(dup.null_count(), 0);
let same = vec![batch(&["x", "x"], &[Some(1), Some(1)], &["a", "a"])];
let one = key_rows(&spec(), [("k", same.as_slice())]).unwrap();
assert_eq!((one.rows.len(), one.collisions), (1, 0));
assert!(dup_part(&spec(), &one.rows.keys().next().unwrap().1).is_none());
}
#[test]
fn a_version_bump_keeps_the_identity() {
let m1 = "rust-analyzer cargo arrow 59.2.0 array/RecordBatch#";
let m2 = "rust-analyzer cargo arrow 59.3.1 array/RecordBatch#";
let sp = spec().stable();
let dumped: BTreeSet<String> = ["k".to_string()].into();
let b1 = vec![batch(&[m1], &[Some(1)], &["a"])];
let first = delta(key_rows(&sp, [("k", b1.as_slice())]).unwrap(), &dumped, &BTreeSet::new(), None);
assert_eq!((first.plus.len(), first.minus.len()), (1, 0));
assert!(first.plus[0].key.starts_with("rust-analyzer cargo arrow * "), "identity is version-free: {}", first.plus[0].key);
let mut open = OpenSet::default();
open.apply(Kind::Edges, "r", &first);
// The dependency bumped: same item, new version in the stored column.
let b2 = vec![batch(&[m2], &[Some(1)], &["a"])];
let again = delta(key_rows(&sp, [("k", b2.as_slice())]).unwrap(), &dumped, &BTreeSet::new(), open.repo(Kind::Edges, "r"));
assert_eq!(again.minus.len(), 0, "a version bump must not close the identity");
assert_eq!(again.plus.len(), 1, "the bump is one payload re-emit on the same identity");
assert_eq!(again.plus[0].key, first.plus[0].key);
// Re-dumped unchanged: an empty delta.
open.apply(Kind::Edges, "r", &again);
let third = delta(key_rows(&sp, [("k", b2.as_slice())]).unwrap(), &dumped, &BTreeSet::new(), open.repo(Kind::Edges, "r"));
assert_eq!((third.plus.len(), third.minus.len()), (0, 0));
}
#[test]
fn stable_keys_reach_each_part_of_a_composite_value() {
// A span-like key column holds several monikers joined by U+001E.
let v1 = "rust-analyzer cargo a 1.0.0 m/f().\u{1e}rust-analyzer cargo b 2.0.0 g/\u{1e}CALLS";
let v2 = "rust-analyzer cargo a 1.0.1 m/f().\u{1e}rust-analyzer cargo b 2.0.0 g/\u{1e}CALLS";
let sp = spec().stable();
let k1 = key_rows(&sp, [("k", vec![batch(&[v1], &[Some(1)], &["a"])].as_slice())]).unwrap();
let k2 = key_rows(&sp, [("k", vec![batch(&[v2], &[Some(1)], &["a"])].as_slice())]).unwrap();
let (id1, id2) = (k1.rows.keys().next().unwrap(), k2.rows.keys().next().unwrap());
assert_eq!(id1, id2, "every moniker inside a composite key part is made stable");
assert!(id1.1.contains(" a * "), "{}", id1.1);
assert!(id1.1.ends_with("CALLS\u{1f}1"), "a non-moniker part is untouched: {}", id1.1);
assert!(id1.1.contains('\u{1e}'), "the composite joiners survive: {}", id1.1);
}
#[test]
fn an_unstable_spec_keys_as_before() {
let m1 = "rust-analyzer cargo arrow 59.2.0 array/RecordBatch#";
let keyed = key_rows(&spec(), [("k", vec![batch(&[m1], &[Some(1)], &["a"])].as_slice())]).unwrap();
assert!(keyed.rows.keys().next().unwrap().1.contains("59.2.0"));
}
#[test]
fn hex_round_trips() {
let h: PayloadHash = core::array::from_fn(|i| (i * 17) as u8);
assert_eq!(unhex(&hex(&h)).unwrap(), h);
assert!(unhex("zz").is_err());
}
}