crates/mimir-ingest/src/provenance.rs
//! Provenance, DERIVED at read time (plan L9-dec): the source stream
//! (`code_node_events`) and the binary stream (`bin_node_events`,
//! `bin_edge_events`, `code_symbol_events`) never share a table and never
//! write a provenance; this query folds both and joins them on the moniker
//! ([`mimir_schema::derive_provenance`]):
//!
//! * a source item DEFINED in the repo (or in the universe ring: dependency /
//! toolchain crates) that a linked artifact carries — its own symbol, or for
//! a generic item ≥1 INSTANCE_OF — is `both`;
//! * a source item no artifact carries is `source_only` (inlined, dead, cfg'd
//! out, never instantiated);
//! * a binary node no source defines (mono instance, ffi import, unmatched
//! symbol) is `binary_only`.
//!
//! Reference stubs (`defined_here = false`) are not source definitions.
use std::collections::{BTreeMap, HashMap, HashSet};
use anyhow::{Context, Result, bail};
use mimir_schema::{BinaryFacts, EdgeKind, Provenance, derive_provenance};
use skade::ScanFilter;
use skade::arrow_array::{Array, BooleanArray, StringArray};
use crate::delta::{Kind, RepoOpen, unhex};
use crate::log::{EventLog, i64_col, identity, str_col};
use crate::schema::{COL_KEY_DUP, COL_KRATE, COL_OP, COL_PAYLOAD_HASH, COL_REPO, COL_SEQ, KEY_SEP, OP_PLUS, UNIVERSE_REPO, qualified};
/// The node payload column that tells a definition from a reference stub.
pub const DEFINED_HERE: &str = "defined_here";
/// Derived provenance per moniker of one repo (as of a consumer ordinal).
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ProvenanceView {
/// Every id the repo's source defines, plus every id the repo's binary
/// stream knows (universe definitions included when the binary links them).
pub by_id: BTreeMap<String, Provenance>,
}
impl ProvenanceView {
pub fn get(&self, id: &str) -> Option<Provenance> {
self.by_id.get(id).copied()
}
/// How many ids have each provenance.
pub fn counts(&self) -> BTreeMap<Provenance, usize> {
let mut out = BTreeMap::new();
for p in self.by_id.values() {
*out.entry(*p).or_default() += 1;
}
out
}
}
/// The `idx`-th key part of an identity (U+001F-joined).
fn part(identity: &str, idx: usize) -> &str {
identity.split(KEY_SEP).nth(idx).unwrap_or_default()
}
fn key_pos(log: &EventLog, kind: Kind, col: &str) -> Result<usize> {
log.spec(kind)
.key_columns
.iter()
.position(|k| k == col)
.with_context(|| format!("{} has no key column `{col}`", EventLog::table_name(kind)))
}
impl EventLog {
/// The derived provenance of `repo` as of consumer ordinal `upto_seq`
/// (`None` = now). See the module docs.
pub async fn provenance(&self, repo: &str, upto_seq: Option<i64>) -> Result<ProvenanceView> {
if repo == UNIVERSE_REPO {
bail!("provenance is per repo; the universe ring has no binary stream");
}
let open = self.fold(Some(repo), upto_seq).await?;
// ---- the binary stream's facts
let mut facts: HashMap<String, BinaryFacts> = HashMap::new();
if let Some(o) = open.repo(Kind::BinNodes, repo) {
// the node key's first column is the id
for (_, ident, _) in o.iter() {
facts.entry(part(&ident, 0).to_string()).or_default().binary_node = true;
}
}
if let Some(o) = open.repo(Kind::Symbols, repo) {
let id = key_pos(self, Kind::Symbols, "id")?;
for (_, ident, _) in o.iter() {
facts.entry(part(&ident, id).to_string()).or_default().has_symbol = true;
}
}
if let Some(o) = open.repo(Kind::BinEdges, repo) {
let (to, kind) = (key_pos(self, Kind::BinEdges, "to")?, key_pos(self, Kind::BinEdges, "kind")?);
for (_, ident, _) in o.iter() {
if part(&ident, kind) == EdgeKind::InstanceOf.as_str() {
facts.entry(part(&ident, to).to_string()).or_default().instances_linked = true;
}
}
}
// ---- the source stream's definitions: the repo's open ones, and the
// universe ring's for the ids the binary links
let mut defined = self.defined_ids(repo, upto_seq, open.repo(Kind::Nodes, repo), None).await?;
let wanted: HashSet<&str> = facts.keys().map(String::as_str).filter(|id| !defined.contains(*id)).collect();
if !wanted.is_empty() {
defined.extend(self.defined_ids(UNIVERSE_REPO, None, None, Some(&wanted)).await?);
}
let mut view = ProvenanceView::default();
for id in defined.iter().chain(facts.keys()) {
if view.by_id.contains_key(id) {
continue;
}
let f = facts.get(id).copied().unwrap_or_default();
if let Some(p) = derive_provenance(defined.contains(id), f) {
view.by_id.insert(id.clone(), p);
}
}
Ok(view)
}
/// Ids DEFINED (not reference stubs) by the source node events of `repo`:
/// with `open`, only identities still open with that payload; without
/// (the universe ring: append-only), every `+`. `only` restricts the ids.
async fn defined_ids(
&self,
repo: &str,
upto_seq: Option<i64>,
open: Option<&RepoOpen>,
only: Option<&HashSet<&str>>,
) -> Result<HashSet<String>> {
let spec = self.spec(Kind::Nodes);
let has_defined = spec.schema.field_with_name(DEFINED_HERE).is_ok();
let mut cols: Vec<&str> = vec![COL_REPO, COL_SEQ, COL_OP, COL_KRATE, COL_PAYLOAD_HASH, COL_KEY_DUP];
cols.extend(spec.key_columns.iter().map(String::as_str));
if has_defined {
cols.push(DEFINED_HERE);
}
let t = self.warehouse().table(&qualified(self.table_for(Kind::Nodes, repo))).await?;
let batches = skade::read_filtered(t.inner(), &ScanFilter::eq(COL_REPO, repo), &cols).await?;
let mut out = HashSet::new();
for b in &batches {
let (r, seq, op, krate, hash, dup) =
(str_col(b, COL_REPO)?, i64_col(b, COL_SEQ)?, str_col(b, COL_OP)?, str_col(b, COL_KRATE)?, str_col(b, COL_PAYLOAD_HASH)?, str_col(b, COL_KEY_DUP)?);
let keys: Vec<StringArray> = spec.key_columns.iter().map(|k| str_col(b, k)).collect::<Result<_>>()?;
let here: Option<BooleanArray> = if has_defined {
let c = b.column_by_name(DEFINED_HERE).context("defined_here")?;
Some(c.as_any().downcast_ref::<BooleanArray>().context("defined_here is not boolean")?.clone())
} else {
None
};
for i in 0..b.num_rows() {
if r.value(i) != repo || op.value(i) != OP_PLUS || upto_seq.is_some_and(|n| seq.value(i) > n) {
continue;
}
if here.as_ref().is_some_and(|h| h.is_null(i) || !h.value(i)) {
continue;
}
// the RAW identity carries the id the binary facts name;
// the open set keys by the STABLE one
let ident = identity(false, &keys, &dup, i);
let id = part(&ident, 0);
if only.is_some_and(|w| !w.contains(id)) {
continue;
}
if let Some(o) = open {
let h = unhex(hash.value(i))?;
let stable = identity(spec.stable_keys, &keys, &dup, i);
if o.get(krate.value(i), &stable) != Some(&h) {
continue;
}
}
out.insert(id.to_string());
}
}
Ok(out)
}
}