crates/mimir-ingest/src/schema.rs
//! The log's own schemas. Generic over the payload: the node / edge payload
//! schemas come from `mimir-schema` (lane 3); this module only adds the event
//! prefix around them and defines the three bookkeeping tables.
use std::sync::Arc;
use anyhow::{Result, bail};
use skade::arrow_array::RecordBatch;
use skade::arrow_cast::can_cast_types;
use skade::arrow_schema::{DataType, Field, Schema, SchemaRef, TimeUnit};
/// The skade namespace every Mímir v2 table lives in. Old data is not migrated.
pub const NAMESPACE: &str = "mimir_v2";
// The SOURCE stream (mimir-driver dumps): written only by the source plan.
pub const CODE_NODE_EVENTS: &str = "code_node_events";
pub const CODE_EDGE_EVENTS: &str = "code_edge_events";
pub const CODE_SPAN_EVENTS: &str = "code_span_events";
/// The dependency ring's own tables (the `@universe` rows: dependency / std
/// crates, dumped once per crate@version + toolchain, never closed). Same
/// event schema as the source tables, unpartitioned. A lake born before the
/// split keeps its ring rows in the source tables until `EventLog::split_ring`
/// copies them here and the compaction drops them there.
pub const UNIVERSE_NODE_EVENTS: &str = "universe_node_events";
pub const UNIVERSE_EDGE_EVENTS: &str = "universe_edge_events";
pub const UNIVERSE_SPAN_EVENTS: &str = "universe_span_events";
/// Table property on `code_node_events`: `split` once the ring lives in its
/// own tables and the source tables are written per repo (partition `repo`).
/// Absent = the lake's legacy shape (ring rows in the source tables).
pub const PROP_RING: &str = "mimir.ring";
pub const RING_SPLIT: &str = "split";
// The BINARY stream (mimir-binary over linked artifacts): its own tables, never
// shared with the source stream (LAW 5, one writer per table). Provenance is
// derived at read time by joining the streams on the moniker.
pub const CODE_SYMBOL_EVENTS: &str = "code_symbol_events";
pub const BIN_NODE_EVENTS: &str = "bin_node_events";
pub const BIN_EDGE_EVENTS: &str = "bin_edge_events";
/// The binary stream's ledger: one row per (artifact, binary release).
pub const BIN_INGEST_RUNS: &str = "bin_ingest_runs";
/// The binary stream's offsets: per artifact, the release (`rev`) and commit
/// it was last written at — the next `rev` is read from here, not the ledger.
pub const BIN_OFFSETS: &str = "bin_offsets";
pub const CONSUMER_OFFSETS: &str = "consumer_offsets";
pub const INGEST_RUNS: &str = "ingest_runs";
pub const UNIVERSE_MANIFEST: &str = "universe_manifest";
/// The ledger of the non-graph feeds that ride the consumer's offsets (git
/// heat, sbom, code index, embeddings …): one row per (repo, sha, feed) run.
pub const FEED_RUNS: &str = "consumer_feed_runs";
/// The backfills asked of this log: one row when a member's history is asked
/// for from a commit (`requested`) and one when the consumer has walked from
/// there to the member's tip (`done`). What is still to do is DERIVED from it
/// and the member's offset — a restart resumes from the lake, not from a file
/// beside it.
pub const BACKFILL_RUNS: &str = "backfill_runs";
/// `repo` value of universe-ring events (dependency / std crates). They are
/// not a repo's state and never enter a repo's open set.
pub const UNIVERSE_REPO: &str = "@universe";
/// Event prefix columns, in order, ahead of the payload columns.
pub const COL_REPO: &str = "repo";
pub const COL_SEQ: &str = "seq";
pub const COL_SHA: &str = "sha";
pub const COL_OP: &str = "op";
pub const COL_KRATE: &str = "krate";
pub const COL_PAYLOAD_HASH: &str = "payload_hash";
/// Non-null only for a key the dump repeated with different payloads: the
/// `#<payload hash>` that tells those identities apart.
pub const COL_KEY_DUP: &str = "key_dup";
/// Binary-stream tables only: the binary release ordinal per repo. A binary
/// re-ingest of the SAME sha (artifact rebuilt) is a new release with the same
/// `seq`; the fold orders binary events by (repo, seq, rev).
pub const COL_REV: &str = "rev";
const PREFIX: [&str; 8] = [COL_REPO, COL_SEQ, COL_SHA, COL_OP, COL_KRATE, COL_PAYLOAD_HASH, COL_KEY_DUP, COL_REV];
/// Table property of an event table: the `mimir-schema` version its columns
/// are at (written when the table is created, moved by an evolution).
pub const PROP_SCHEMA: &str = mimir_schema::META_SCHEMA;
/// Table property: the columns added to the table in place, comma-joined
/// (see [`PayloadSpec::added`]). Absent = the table is as it was created.
pub const PROP_ADDED: &str = "mimir.added";
/// `op` of an event that opens (or replaces) a key's payload.
pub const OP_PLUS: &str = "+";
/// `op` of an event that closes a key.
pub const OP_MINUS: &str = "-";
/// Ledger statuses.
pub const STATUS_OK: &str = "ok";
pub const STATUS_SKIPPED_DOCS_ONLY: &str = "skipped_docs_only";
pub const STATUS_ERROR: &str = "error";
/// Some local crates did not compile: the ones that did were dumped, the
/// failed ones (named in `error`) keep their previous state; offset advances.
pub const STATUS_PARTIAL: &str = "partial";
/// `bin_ingest_runs`: the artifact is no longer linked (its bin left the
/// workspace, or it is linked under another artifact id): its keys closed.
pub const STATUS_CLOSED: &str = "closed";
/// `consumer_feed_runs.status`: the feed could not run here and says why (no embedder
/// compiled, no manifest) — not an error, and not `ok`: it stays behind.
pub const STATUS_SKIPPED: &str = "skipped";
/// `consumer_feed_runs.status` of the binary hook: no linked artifact of the
/// commit was where the hook looks (`detail` names the path). Not an error and
/// not `ok`: the hook looks again when an artifact appears there.
pub const STATUS_NO_ARTIFACT: &str = "no_artifact";
/// `backfill_runs.status`: asked for, not yet walked to the member's tip.
pub const STATUS_REQUESTED: &str = "requested";
/// `backfill_runs.status`: the consumer walked from the first commit to the tip.
pub const STATUS_DONE: &str = "done";
/// Separator joining a multi-column key into one map key. Refused inside key
/// values so the join is unambiguous.
pub(crate) const KEY_SEP: char = '\u{1f}';
/// Stands for a null key part inside a joined key.
pub(crate) const KEY_NULL: char = '\u{0}';
/// Joins the pieces of ONE composite key-column value (a span's edge key:
/// `from U+001E to U+001E kind`).
pub(crate) const PART_SEP: char = '\u{1e}';
/// Leads the disambiguating key part of a repeated key (see `key_dup`).
pub(crate) const DUP_MARK: char = '#';
/// A payload schema plus the columns that identify a row (the key the delta
/// and the fold group by): e.g. the moniker for nodes, `(from, to, kind)` for
/// edges.
#[derive(Debug, Clone)]
pub struct PayloadSpec {
pub schema: SchemaRef,
pub key_columns: Vec<String>,
/// Columns of the dumped batches the LOG owns and therefore does not
/// store in the payload (e.g. `repo`, which is the event prefix, or
/// `valid_from_sha` / `valid_to_sha`, which the log derives). Present or
/// not in a dumped batch, they are ignored and not hashed.
pub log_owned: Vec<String>,
/// Payload columns THIS lake's table did not have when it was created:
/// they were added in place (schema evolution, [`PROP_ADDED`]), so the
/// rows written before have no value for them. The payload hash covers
/// such a column only where it HAS a value: a row that is null there
/// hashes exactly as it did before the column existed, so an evolved lake
/// re-emits a row only when the producer now says something about it.
/// Empty for a spec that is not bound to a lake, and for a fresh lake.
pub added: Vec<String>,
/// Key columns are made STABLE before keying: every `rust-analyzer cargo`
/// moniker in a key part (or in a U+001E-joined piece of one) has its
/// version replaced by [`mimir_schema::moniker::STABLE_VERSION`], so the
/// same item at two dependency (or own) versions is ONE identity and a
/// version bump is a payload change, never a close + open. The stored
/// columns keep the real version. Set on the source-stream specs.
pub stable_keys: bool,
}
impl PayloadSpec {
/// Key columns must exist and be castable to text and back (strings,
/// integers, booleans); no payload column may reuse an event-prefix name.
pub fn new(schema: SchemaRef, key_columns: &[&str]) -> Result<Self> {
Self::with_log_owned(schema, key_columns, &[])
}
/// [`Self::new`], with `log_owned` columns dropped from the payload first.
pub fn with_log_owned(schema: SchemaRef, key_columns: &[&str], log_owned: &[&str]) -> Result<Self> {
if key_columns.is_empty() {
bail!("a payload needs at least one key column");
}
for o in log_owned {
if schema.field_with_name(o).is_err() {
bail!("log-owned column `{o}` is not in the payload schema");
}
if key_columns.contains(o) {
bail!("log-owned column `{o}` cannot also be a key column");
}
}
let kept: Vec<Field> = schema
.fields()
.iter()
.filter(|f| !log_owned.contains(&f.name().as_str()))
.map(|f| f.as_ref().clone())
.collect();
let schema = Arc::new(Schema::new(kept).with_metadata(schema.metadata().clone()));
for f in schema.fields() {
if PREFIX.contains(&f.name().as_str()) {
bail!(
"payload column `{}` collides with the event prefix {PREFIX:?}; name it log-owned",
f.name()
);
}
}
for k in key_columns {
match schema.field_with_name(k) {
Ok(f) => {
let t = f.data_type();
if !(can_cast_types(t, &DataType::Utf8) && can_cast_types(&DataType::Utf8, t)) || t.is_nested() {
bail!("key column `{k}` is {t:?}; key columns must round-trip through text");
}
}
Err(_) => bail!("key column `{k}` is not in the payload schema"),
}
}
Ok(Self {
schema,
key_columns: key_columns.iter().map(|k| k.to_string()).collect(),
log_owned: log_owned.iter().map(|k| k.to_string()).collect(),
added: Vec::new(),
stable_keys: false,
})
}
/// This payload keys by STABLE ids (see [`PayloadSpec::stable_keys`]).
pub fn stable(mut self) -> Self {
self.stable_keys = true;
self
}
/// The `mimir-schema` version of the payload (its schema metadata), if it
/// carries one.
pub fn version(&self) -> Option<&str> {
self.schema.metadata().get(mimir_schema::META_SCHEMA).map(String::as_str)
}
/// The event table's schema: prefix + every payload column, nullable (a
/// `-` event carries only its key columns).
pub fn event_schema(&self) -> Schema {
self.event_schema_with(false)
}
/// [`Self::event_schema`]; `rev` = the binary stream's tables, which carry
/// [`COL_REV`] after the prefix.
pub fn event_schema_with(&self, rev: bool) -> Schema {
let mut fields = vec![
Field::new(COL_REPO, DataType::Utf8, false),
Field::new(COL_SEQ, DataType::Int64, false),
Field::new(COL_SHA, DataType::Utf8, false),
Field::new(COL_OP, DataType::Utf8, false),
Field::new(COL_KRATE, DataType::Utf8, false),
Field::new(COL_PAYLOAD_HASH, DataType::Utf8, true),
Field::new(COL_KEY_DUP, DataType::Utf8, true),
];
if rev {
fields.push(Field::new(COL_REV, DataType::Int64, false));
}
fields.extend(
self.schema
.fields()
.iter()
.map(|f| Field::new(f.name(), storage_type(f.data_type()), true)),
);
Schema::new(fields)
}
/// `batch` re-ordered into this payload's column order, refusing a missing
/// column, a type that differs from the spec, or a column the spec does
/// not know (silently dropping data is not an option).
pub(crate) fn conform(&self, batch: &RecordBatch) -> Result<RecordBatch> {
for f in batch.schema().fields() {
if self.schema.field_with_name(f.name()).is_err() && !self.log_owned.iter().any(|o| o == f.name()) {
bail!("dumped column `{}` is not in the payload schema {:?}", f.name(), field_names(&self.schema));
}
}
let mut cols = Vec::with_capacity(self.schema.fields().len());
for f in self.schema.fields() {
let Some(col) = batch.column_by_name(f.name()) else {
bail!("dumped batch has no column `{}` (payload schema: {:?})", f.name(), field_names(&self.schema));
};
if col.data_type() != f.data_type() {
bail!(
"dumped column `{}` is {:?}, payload schema says {:?}",
f.name(),
col.data_type(),
f.data_type()
);
}
cols.push(col.clone());
}
let relaxed = Schema::new(
self.schema
.fields()
.iter()
.map(|f| Field::new(f.name(), f.data_type().clone(), true))
.collect::<Vec<_>>(),
);
Ok(RecordBatch::try_new(Arc::new(relaxed), cols)?)
}
}
/// The type a payload column is STORED as in an event table: Iceberg has no
/// unsigned ints, so they widen by VALUE (lossless, no bit reinterpret):
/// u8/u16 → i32, u32 → i64, u64 → decimal(20, 0). Everything else as is.
/// Hashes and keys are computed on the dumped (unwidened) values.
pub fn storage_type(t: &DataType) -> DataType {
match t {
DataType::UInt8 | DataType::UInt16 => DataType::Int32,
DataType::UInt32 => DataType::Int64,
DataType::UInt64 => DataType::Decimal128(20, 0),
other => other.clone(),
}
}
fn field_names(s: &Schema) -> Vec<&str> {
s.fields().iter().map(|f| f.name().as_str()).collect()
}
fn ts() -> DataType {
DataType::Timestamp(TimeUnit::Microsecond, Some("+00:00".into()))
}
/// `consumer_offsets`: append-only; a repo's offset is its row with the
/// highest `seq`.
pub fn consumer_offsets_schema() -> Schema {
Schema::new(vec![
Field::new("repo", DataType::Utf8, false),
Field::new("branch", DataType::Utf8, false),
Field::new("sha", DataType::Utf8, false),
Field::new("seq", DataType::Int64, false),
Field::new("ts", ts(), false),
])
}
/// `ingest_runs`: the ledger, one row per consumed commit (and one per
/// failed attempt).
pub fn ingest_runs_schema() -> Schema {
Schema::new(vec![
Field::new("repo", DataType::Utf8, false),
Field::new("seq", DataType::Int64, false),
Field::new("sha", DataType::Utf8, false),
Field::new("status", DataType::Utf8, false),
Field::new("crates_dumped", DataType::Int64, false),
Field::new("rows_plus", DataType::Int64, false),
Field::new("rows_minus", DataType::Int64, false),
Field::new("key_collisions", DataType::Int64, false),
Field::new("span_plus", DataType::Int64, false),
Field::new("span_minus", DataType::Int64, false),
Field::new("owner_disambiguations", DataType::Int64, false),
Field::new("ms", DataType::Int64, false),
Field::new("error", DataType::Utf8, true),
Field::new("ts", ts(), false),
])
}
/// `consumer_feed_runs`: the ledger of the post-commit feeds — what ran for which
/// (repo, sha), with which outcome. A feed's own offset is DERIVED: its last
/// `ok` row of the repo.
pub fn feed_runs_schema() -> Schema {
Schema::new(vec![
Field::new("repo", DataType::Utf8, false),
Field::new("seq", DataType::Int64, false),
Field::new("sha", DataType::Utf8, false),
Field::new("feed", DataType::Utf8, false),
Field::new("status", DataType::Utf8, false),
Field::new("rows", DataType::Int64, false),
Field::new("ms", DataType::Int64, false),
Field::new("detail", DataType::Utf8, true),
Field::new("ts", ts(), false),
])
}
/// `backfill_runs`: append-only. `from_sha` = the first commit of the history
/// asked for; `base_seq` = the member's consumer ordinal when it was asked (0:
/// never consumed) — the backfill has started once the offset's ordinal is
/// past it; `target_sha` / `total` = the member's tip and the first-parent
/// commits from `from_sha` to it, as they were then (the tip may move on).
pub fn backfill_runs_schema() -> Schema {
Schema::new(vec![
Field::new("repo", DataType::Utf8, false),
Field::new("from_sha", DataType::Utf8, false),
Field::new("base_seq", DataType::Int64, false),
Field::new("target_sha", DataType::Utf8, false),
Field::new("total", DataType::Int64, false),
Field::new("status", DataType::Utf8, false),
Field::new("detail", DataType::Utf8, true),
Field::new("ts", ts(), false),
])
}
/// `universe_manifest`: which dependency crate was dumped under which
/// toolchain, so it is dumped once. `rustc_release` = the toolchain's
/// `rustc -V` release (lang crates carry rust-analyzer's URL as their version,
/// so the release lives here); null for a crates.io crate dumped by a repo.
pub fn universe_manifest_schema() -> Schema {
Schema::new(vec![
Field::new("crate", DataType::Utf8, false),
Field::new("toolchain", DataType::Utf8, false),
Field::new("dumped_at", ts(), false),
Field::new("rows", DataType::Int64, false),
Field::new("rustc_release", DataType::Utf8, true),
])
}
/// `bin_offsets`: append-only; an artifact's offset is its row with the
/// highest `rev`, the repo's last binary release the highest `rev` of all.
pub fn bin_offsets_schema() -> Schema {
Schema::new(vec![
Field::new("repo", DataType::Utf8, false),
Field::new("artifact", DataType::Utf8, false),
Field::new("rev", DataType::Int64, false),
Field::new("seq", DataType::Int64, false),
Field::new("sha", DataType::Utf8, false),
Field::new("ts", ts(), false),
])
}
/// `bin_ingest_runs`: the binary stream's ledger — one row per artifact per
/// binary release (`rev`), with mimir-binary's join ledger as JSON.
pub fn bin_ingest_runs_schema() -> Schema {
Schema::new(vec![
Field::new("repo", DataType::Utf8, false),
Field::new("seq", DataType::Int64, false),
Field::new("rev", DataType::Int64, false),
Field::new("sha", DataType::Utf8, false),
Field::new("artifact", DataType::Utf8, false),
Field::new("status", DataType::Utf8, false),
Field::new("rows_plus", DataType::Int64, false),
Field::new("rows_minus", DataType::Int64, false),
Field::new("key_collisions", DataType::Int64, false),
Field::new("ms", DataType::Int64, false),
Field::new("ledger", DataType::Utf8, true),
Field::new("ts", ts(), false),
])
}
/// `ns.table` as skade parses it.
pub fn qualified(table: &str) -> String {
format!("{NAMESPACE}.{table}")
}
/// Columns of `mimir-schema` rows the LOG owns: `repo` is the event prefix,
/// `valid_from_sha` / `valid_to_sha` are derived by the fold.
pub const MIMIR_LOG_OWNED: [&str; 2] = ["valid_from_sha", "valid_to_sha"];
/// Node position columns: they are `code_spans` rows (the driver writes them
/// there too), never node identity or node payload — moving an item must not
/// produce a node event.
pub const NODE_POSITION_COLUMNS: [&str; 3] = ["file", "start_line", "end_line"];
/// `code_nodes` payload (mimir-schema) keyed by the node id (SCIP moniker);
/// positions excluded (stored nowhere in this table, hashed nowhere).
pub fn code_nodes_spec() -> Result<PayloadSpec> {
let mut owned = vec!["repo"];
owned.extend(MIMIR_LOG_OWNED);
owned.extend(NODE_POSITION_COLUMNS);
PayloadSpec::with_log_owned(mimir_schema::nodes_schema(), &["id"], &owned)
}
/// `code_edges` payload (mimir-schema): one SEMANTIC edge per
/// (from, to, kind); `count` (sites) is payload, sites are spans.
pub fn code_edges_spec() -> Result<PayloadSpec> {
PayloadSpec::with_log_owned(mimir_schema::edges_schema(), &["from", "to", "kind"], &MIMIR_LOG_OWNED)
}
/// `code_spans` payload (mimir-schema): a position is its own identity
/// (target, key, file, start, end) — moved code is a `-` and a `+` here.
pub fn code_spans_spec() -> Result<PayloadSpec> {
PayloadSpec::new(mimir_schema::spans_schema(), &["target", "key", "file", "start_line", "end_line"])
}
/// `provenance` is never stored by the binary stream: it is DERIVED at read
/// time (`EventLog::provenance`).
pub const DERIVED_PROVENANCE: &str = "provenance";
/// Binary-stream node payload (`bin_node_events`): mimir-schema `code_nodes`
/// rows of the nodes only the binary knows (mono_instance, ffi_import,
/// unmatched symbols), keyed by id; positions and provenance not stored.
pub fn code_bin_nodes_spec() -> Result<PayloadSpec> {
let mut owned = vec!["repo", DERIVED_PROVENANCE];
owned.extend(MIMIR_LOG_OWNED);
owned.extend(NODE_POSITION_COLUMNS);
PayloadSpec::with_log_owned(mimir_schema::nodes_schema(), &["id"], &owned)
}
/// Binary-stream edge payload (`bin_edge_events`): INSTANCE_OF edges.
pub fn code_bin_edges_spec() -> Result<PayloadSpec> {
let mut owned = vec![DERIVED_PROVENANCE];
owned.extend(MIMIR_LOG_OWNED);
PayloadSpec::with_log_owned(mimir_schema::edges_schema(), &["from", "to", "kind"], &owned)
}
/// `code_symbol_events` payload: mimir-schema `code_symbols` rows of one
/// artifact (the artifact is the event's owner), keyed by (symbol, id) — an
/// import with several source declarations is one row per declaration.
pub fn code_symbols_spec() -> Result<PayloadSpec> {
let mut owned = vec![DERIVED_PROVENANCE];
owned.extend(MIMIR_LOG_OWNED);
PayloadSpec::with_log_owned(mimir_schema::symbols_schema(), &["symbol", "id"], &owned)
}
/// The six payloads of the log: three per producer stream.
#[derive(Debug, Clone)]
pub struct Specs {
// source stream
pub nodes: PayloadSpec,
pub edges: PayloadSpec,
pub spans: PayloadSpec,
// binary stream
pub bin_nodes: PayloadSpec,
pub bin_edges: PayloadSpec,
pub symbols: PayloadSpec,
}
impl Specs {
pub(crate) fn get_mut(&mut self, kind: crate::delta::Kind) -> &mut PayloadSpec {
use crate::delta::Kind;
match kind {
Kind::Nodes => &mut self.nodes,
Kind::Edges => &mut self.edges,
Kind::Spans => &mut self.spans,
Kind::BinNodes => &mut self.bin_nodes,
Kind::BinEdges => &mut self.bin_edges,
Kind::Symbols => &mut self.symbols,
}
}
/// The mimir-schema payloads.
pub fn mimir() -> Result<Self> {
Ok(Self {
// The source stream keys by STABLE ids: a version bump is a
// payload change, never a close + open. The binary stream keeps
// raw keys (its owners are artifacts; the stable join is the
// binary lane's own step).
nodes: code_nodes_spec()?.stable(),
edges: code_edges_spec()?.stable(),
spans: code_spans_spec()?.stable(),
bin_nodes: code_bin_nodes_spec()?,
bin_edges: code_bin_edges_spec()?,
symbols: code_symbols_spec()?,
})
}
}