nordisk/nornir
src/warehouse/iceberg.rs
//! Iceberg-backed warehouse implementation.
//!
//! Files live under `<root>/warehouse/<namespace>/<table>/data/...`,
//! catalog metadata in `<root>/catalog.redb` (single ACID file via
//! [`skade_katalog`]). The sync [`Warehouse`] trait is bridged
//! to the async iceberg API by a private tokio runtime owned by this
//! warehouse instance.
//!
//! No services, no SQL server, no JVM. Layout is straight Iceberg —
//! a pyiceberg / Spark / DuckDB reader pointed at the same warehouse
//! root + catalog can read every row.
use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;
use anyhow::{Context, Result, anyhow};
use arrow::array::{
Array, BooleanArray, Float64Array, Int32Array, Int64Array, RecordBatch, StringArray,
TimestampMicrosecondArray,
};
use chrono::{DateTime, NaiveDate, NaiveDateTime, TimeZone, Utc};
use iceberg::arrow::schema_to_arrow_schema;
use iceberg::io::LocalFsStorageFactory;
use iceberg::spec::{PartitionSpec, Transform, UnboundPartitionSpec};
use iceberg::table::Table;
use iceberg::{Catalog, CatalogBuilder, NamespaceIdent, TableCreation, TableIdent};
use skade_katalog::{HealOutcome, RedbCatalog, RedbCatalogBuilder};
use tokio::runtime::Runtime;
use uuid::Uuid;
use super::iceberg_schema;
use super::{BenchFilter, ScanFilter, Warehouse};
use crate::bench::{BenchResult, BenchRun, TestOutcome};
const NAMESPACE: &str = "nornir";
const TABLE_BENCH_RUNS: &str = "bench_runs";
const TABLE_BENCH_RESULTS: &str = "bench_results";
// Per-bench CPU/mem telemetry (one row per (run_id, bench)). Partition by `repo`
// like bench_runs so per-repo reads prune.
const TABLE_BENCH_TELEMETRY: &str = "bench_telemetry";
const TABLE_TEST_OUTCOMES: &str = "test_outcomes";
// First-class test-matrix results (one row per (run_id, suite, test_name)).
// Written by `nornir test`; read by the viz Test pane + `nornir test history`.
// Partitioned by `repo` like bench_runs so per-repo reads prune.
pub(crate) const TABLE_TEST_RESULTS: &str = "test_results";
// build-provenance ledger (task #36): one row per recorded `nornir build
// --attest`. Partitioned by `repo` so per-repo history reads prune.
pub(crate) const TABLE_BUILD_PROVENANCE: &str = "build_provenance";
// build-info manifest + live-feed rows (§f3 of the build-info flight-stream
// design): one row per (build_id, kind, name) — `provenance` / `dep` / `artifact`
// (the Artifactory-parity manifest) plus `log` / `metric` / `step` (the feed).
// The sibling of `build_provenance`: that ledger has one flat row PER BUILD, this
// has the exploded columnar record ABOUT the build. Unpartitioned — reads scope
// by `build_id` (one build is a handful of rows).
pub(crate) const TABLE_BUILD_INFO: &str = "build_info";
pub(crate) const TABLE_DEP_GRAPH_EDGES: &str = "dep_graph_edges";
// The three RELEASE table NAMES come from the shared `nornir-release-ledger`
// leaf, which also owns their schemas and Arrow codecs — the release
// orchestrator addresses the same three tables in this same warehouse, so the
// name and the shape must come from one place or they can drift apart (L5).
// TABLE_RELEASE_LINEAGE — per-(release, repo) outcome ledger.
// TABLE_RELEASE_EVENTS — row-level DAG of a release run (one row per
// component×op×phase boundary); read by `nornir
// release events` and the Release Ops viz tab.
// TABLE_RELEASE_CHANGES — reversible release-change ledger (task #32): one row
// per mutating release step with enough before-state to
// undo it, SHA-keyed, grouped by `release_id`. Written
// by the bump/patch-strip/branch/publish hooks; read by
// `release undo` (newest-first reverse) + the viz.
pub(crate) use nornir_release_ledger::schema::{
TABLE_RELEASE_CHANGES, TABLE_RELEASE_EVENTS, TABLE_RELEASE_LINEAGE,
};
// Row-level DAG of an airgap pack/move/unfold/install/start run (one row per
// product×op×phase boundary). The airgap twin of release_events. Written by
// `nornir airgap start`; read by `nornir airgap events` / a future viz tab.
pub(crate) const TABLE_AIRGAP_EVENTS: &str = "airgap_events";
// Per-member clone/fetch/republish populate outcomes (PLAN #6): one row per
// monitored-member attempt so a thin viz/CLI client can read *why* a member
// didn't populate (auth, vanished repo, network) instead of only a server log.
// Written by the monitor's fetch/republish path; read by `nornir workspace
// events` + the 🧬 nornir root pane via `Viz.CloneEvents`.
pub(crate) const TABLE_CLONE_EVENTS: &str = "clone_events";
pub(crate) const TABLE_WORKSPACE_EVENTS: &str = "workspace_events";
pub(crate) const TABLE_FUNNEL_EVENTS: &str = "funnel_events";
pub(crate) const TABLE_TANTIVY_INDEX_SNAPSHOTS: &str = "tantivy_index_snapshots";
pub(crate) const TABLE_TANTIVY_INDEX_BLOBS: &str = "tantivy_index_blobs";
// Urðr time-machine: future-artifact stubs. Tables are registered now
// so the warehouse schema is forward-compatible; no capture logic ships
// yet. Same shape as tantivy_index_* — populated by the upcoming
// `nornir debuginfo snapshot` workflow.
pub(crate) const TABLE_DWARF_SNAPSHOTS: &str = "dwarf_snapshots";
pub(crate) const TABLE_DWARF_BLOBS: &str = "dwarf_blobs";
pub(crate) const TABLE_GIMLI_SNAPSHOTS: &str = "gimli_snapshots";
pub(crate) const TABLE_GIMLI_BLOBS: &str = "gimli_blobs";
// Rustdoc-JSON snapshots: same artifact-snapshot/blob shape used by
// tantivy/dwarf/gimli; populated by feature 7 (rustdoc-json semver diff).
pub(crate) const TABLE_RUSTDOC_JSON_SNAPSHOTS: &str = "rustdoc_json_snapshots";
pub(crate) const TABLE_RUSTDOC_JSON_BLOBS: &str = "rustdoc_json_blobs";
// Vector / semantic-search: materialized embeddings + index identity +
// per-occurrence manifest. Lets semantic search at any historical git_sha be
// a warehouse read (no re-embed, no git walk).
pub(crate) const TABLE_EMBEDDINGS: &str = "embeddings";
pub(crate) const TABLE_EMBEDDING_SNAPSHOTS: &str = "embedding_snapshots";
pub(crate) const TABLE_EMBEDDING_MANIFEST: &str = "embedding_manifest";
// One vector per function node: node id → content_hash (events) + the run ledger.
pub(crate) const TABLE_EMBEDDING_NODE_EVENTS: &str = "embedding_node_events";
pub(crate) const TABLE_EMBEDDING_NODE_RUNS: &str = "embedding_node_runs";
// The dependency ring's function vectors: the schema of `embeddings`, a table
// of its own. A ring is 10^5–10^6 bodies; every reader of `embeddings` scans
// the model's rows, so the ring must not sit among a workspace's own vectors.
pub(crate) const TABLE_EMBEDDINGS_RING: &str = "embeddings_ring";
// The clone signals: the normalised body per function node (exact key, MinHash),
// and the vectors of the normalised bodies (the `embeddings` schema, keyed by
// the exact key) — a model space without the names.
pub(crate) const TABLE_FN_SHAPE_EVENTS: &str = "fn_shape_events";
pub(crate) const TABLE_EMBEDDINGS_NORM: &str = "embeddings_norm";
pub(crate) const TABLE_FN_ROLE_VECTORS: &str = "fn_role_vectors";
// The group twins (`mimir_groupclone`): vectors of the normalised semantic
// GROUPS inside function bodies — the `embeddings` schema, keyed by the
// group's exact key, the same convention as `embeddings_norm` one level down.
pub(crate) const TABLE_EMBEDDINGS_GROUPS: &str = "embeddings_groups";
// Cargo-pipeline fact tables (release-scoped, dep-graph-scoped, or events).
pub(crate) const TABLE_PATH_DEP_AUDITS: &str = "path_dep_audits";
pub(crate) const TABLE_PATCH_STRIP_EVENTS: &str = "patch_strip_events";
pub(crate) const TABLE_PUBLISH_ATTEMPTS: &str = "publish_attempts";
pub(crate) const TABLE_CRATE_METADATA_CHECKS: &str = "crate_metadata_checks";
pub(crate) const TABLE_CRATE_TARBALL_STATS: &str = "crate_tarball_stats";
pub(crate) const TABLE_YANK_EVENTS: &str = "yank_events";
pub(crate) const TABLE_SEMVER_DIFFS: &str = "semver_diffs";
/// The per-SYMBOL grain of a `compatible?` edge — which public item was
/// added/removed/changed, with both join keys (plain name → `call_edges`,
/// SCIP moniker → `scip_occurrences`). See
/// [`iceberg_schema::semver_symbol_diffs`] for the snapshot-keying argument.
pub(crate) const TABLE_SEMVER_SYMBOL_DIFFS: &str = "semver_symbol_diffs";
pub(crate) const TABLE_LINKS_DECLARATIONS: &str = "links_declarations";
pub(crate) const TABLE_RESOLVED_FEATURES: &str = "resolved_features";
pub(crate) const TABLE_MSRV_PROBES: &str = "msrv_probes";
pub(crate) const TABLE_TEST_IMPACTED_SELECTIONS: &str = "test_impacted_selections";
pub(crate) const TABLE_TEST_QUARANTINES: &str = "test_quarantines";
pub(crate) const TABLE_RELEASE_COMMITS: &str = "release_commits";
pub(crate) const TABLE_REGISTRY_MIRRORS: &str = "registry_mirrors";
pub(crate) const TABLE_VERSION_BUMP_PLANS: &str = "version_bump_plans";
pub(crate) const TABLE_VERSION_BUMP_TARGETS: &str = "version_bump_targets";
// Knowledge map (gix producer). The syn-era tables (`symbol_facts`,
// `call_edges`, `feature_gate_facts`, `test_inventory`, `scip_occurrences`,
// `warehouse_access_edges`, `architecture_wiring`, `knowledge_scans`,
// `semantic_clusters`) lost their producers with Mímir v2 and are no longer
// registered; an existing warehouse keeps them on disk, never read again.
pub(crate) const TABLE_GIT_HEAT_FACTS: &str = "git_heat_facts";
// Rendered documentation: export history (inline bytes) + a dedicated
// Tantivy index (snapshot/blob pattern), kept separate from the code index.
pub(crate) const TABLE_DOC_EXPORTS: &str = "doc_exports";
// MCP server call telemetry (metrics only — tool, status, latency, ts).
pub(crate) const TABLE_MCP_REQUESTS: &str = "mcp_requests";
pub(crate) const TABLE_VULN_FINDINGS: &str = "vuln_findings";
// Lens answers kept per room version (1:1 with the consumed iceberg transaction
// of the event log): the lake IS the lens cache — a restarted server reads a
// kept answer back instead of folding the lens again. Partitioned by
// `room_version` so a read prunes to the one version it asks for. NOT `lens_*`:
// that prefix is v1's per-lens result tables, which Repopulate purges
// (`code_room::status::is_v1_table`) — this v2 table must never match it.
pub(crate) const TABLE_LENS_ANSWERS: &str = "mimir_lens_answers";
pub(crate) const TABLE_DOCS_INDEX_SNAPSHOTS: &str = "docs_index_snapshots";
pub(crate) const TABLE_DOCS_INDEX_BLOBS: &str = "docs_index_blobs";
// SBOM components captured from a deep-scan / cargo-metadata sweep (one row per
// resolved dep), so the CycloneDX SBOM is a warehouse read.
pub(crate) const TABLE_SBOM_COMPONENTS: &str = "sbom_components";
// Binary-analysis facts (task #65): one header row per analysed artifact
// (`binary_facts`) + the per-section (`section_sizes`) and per-symbol
// (`symbol_sizes`) size breakdowns, all keyed by `content_sha256` (the
// build_info.content_id join key). See `.nornir/binary-analysis-design.md` §5.
pub(crate) const TABLE_BINARY_FACTS: &str = "binary_facts";
pub(crate) const TABLE_SECTION_SIZES: &str = "section_sizes";
pub(crate) const TABLE_SYMBOL_SIZES: &str = "symbol_sizes";
// Artifact-to-artifact diff (task #65 slice 2): one narrow row per change
// between two analysed artifacts, keyed on the pair
// (from_content_sha256, to_content_sha256). See §5.4.
pub(crate) const TABLE_ARTIFACT_DIFF: &str = "artifact_diff";
// H5 local-LLM bake-off: one row per (run_id, model, prompt_id) answer.
pub(crate) const TABLE_AGENT_MODEL_RUNS: &str = "agent_model_runs";
// N5 viz action trail — durable sink for the viz user-action log (the twin of
// the `$NORNIR_VIZ_ACTIONLOG` /tmp file). One row per pushed action.
pub(crate) const TABLE_VIZ_ACTIONS: &str = "viz_actions";
// autonom completeness gate (AUT2 / n-005): one row per discovered surface node
// × mode × workspace × verdict — the persisted gate verdict, mirroring
// `tests/mcp_tool_coverage.json` generalized to the whole surface.
pub(crate) const TABLE_SURFACE_COVERAGE: &str = "surface_coverage";
/// Precomputed kNN neighbour lists — the GPU scan's output, landed as a table so
/// the pure lens folds can read neighbours from `Sources` instead of calling a
/// device. See [`iceberg_schema::knn_neighbors`].
pub(crate) const TABLE_KNN_NEIGHBORS: &str = "knn_neighbors";
// Source-based line/region coverage (feature N — maven-style). `coverage` holds
// the overall + per-crate + per-file rollup rows; `coverage_fn` the per-function
// rows tagged with their ui/grpc/emitter boundary. Partitioned by `repo`.
pub(crate) const TABLE_COVERAGE: &str = "coverage";
pub(crate) const TABLE_COVERAGE_FN: &str = "coverage_fn";
// EPIC SAGA (task #25) — record/replay: one row per captured user-journey event
// (`nornir_saga::SagaEvent`). The matrix reads a session's rows back and replays
// the nav/click path as a robot test. Partitioned by `session_id` so a
// per-session read prunes to that journey's files.
pub(crate) const TABLE_SAGA_EVENTS: &str = "saga_events";
// UI snapshot reference store (render-oracle reference images, WebP, moved out
// of git). Read scoped to one `repo`'s references, so partitioned by `repo`.
pub(crate) const TABLE_UI_SNAPSHOTS: &str = "ui_snapshots";
// Yggdrasil INFRA CROWN + RUNTIME band (previously "MISSING ⛔" in
// `yggdrasil-supergraph.md`): the machines/services/VMs/containers where the
// constellation actually runs (`infra_nodes`/`infra_edges`) plus the general
// runtime-trace band (`runtime_spans`). Snapshot-enveloped like
// `dep_graph_edges`; reads scope by `workspace_name`. Seeded from the jera
// JobRecord ledger (see `warehouse::yggdrasil`).
pub(crate) const TABLE_INFRA_NODES: &str = "infra_nodes";
pub(crate) const TABLE_INFRA_EDGES: &str = "infra_edges";
pub(crate) const TABLE_RUNTIME_SPANS: &str = "runtime_spans";
// FEATURE ACTIVATION graph parsed out of `Cargo.toml` (one row per
// feature → dep/feature edge). WHY a table and not a `cargo metadata` read:
// cargo's `features` map is already FLATTENED and RESOLVED, so the conditional
// structure — which feature would still pull a dep if another were switched off,
// and above all the `pkg?/feat` WEAK form that only activates when `pkg` is
// enabled by some other route — is destroyed before cargo hands the data over.
// Only the manifest has it, so it is parsed and persisted here. Partitioned by
// `repo` like its `#[cfg]`-side sibling `feature_gate_facts`.
pub(crate) const TABLE_FEATURE_DEP_EDGES: &str = "feature_dep_edges";
pub struct IcebergWarehouse {
catalog: Arc<RedbCatalog>,
rt: Option<Runtime>,
namespace: NamespaceIdent,
root: std::path::PathBuf,
/// When this warehouse was opened from a copied-aside snapshot of
/// `catalog.redb` (because a live `nornir-server` holds the exclusive
/// redb lock on the real file), the temp dir that backs the copy is
/// owned here so it lives exactly as long as the catalog and is cleaned
/// up on drop. `None` for a normal exclusive open of the live catalog.
_snapshot: Option<tempfile::TempDir>,
}
/// True when `err` (anywhere in its source chain) is redb's exclusive-lock
/// rejection — i.e. another process (typically the live `nornir-server`)
/// already holds the single-writer lock on `catalog.redb`. skade-katalog
/// surfaces this as an `anyhow`/`iceberg` error whose `Display` carries
/// redb's `DatabaseAlreadyOpen` wording, so we match on the message across
/// the whole chain rather than a concrete error type we don't re-export.
pub fn is_catalog_lock_error(err: &anyhow::Error) -> bool {
err.chain().any(|e| {
let m = e.to_string();
m.contains("Database already open") || m.contains("Cannot acquire lock")
})
}
/// True when `err` is an Iceberg "blank/empty JSON" parse failure — a table
/// whose on-disk metadata (or a manifest/manifest-list) JSON was left **empty**
/// (a torn write, an interrupted commit). iceberg-rust surfaces this as a
/// `DataInvalid` wrapping serde's `EOF while parsing a value at line 1 column 0`.
///
/// A blank metadata file is not a real "this table is broken" condition the viz
/// should abort on — it means "no committed snapshot the reader can see", which
/// for a preview/scan is just **zero rows**. The 🕸 Call Graph lens scans
/// `symbol_facts`; one blank metadata file used to take the whole pane down with
/// an Internal error instead of degrading to an empty graph.
pub(crate) fn is_blank_metadata_json_error(err: &anyhow::Error) -> bool {
err.chain().any(|e| {
let m = e.to_string();
// serde_json's EOF-on-empty-input message, however iceberg-rust wraps it.
m.contains("EOF while parsing a value at line 1 column 0")
|| (m.contains("Failed to parse json string") && m.contains("EOF"))
})
}
/// The number of records a table's OWN current Iceberg snapshot advertises
/// (`total-records` in the snapshot summary) — metadata only, no data read.
/// `None` when the table has never been written (no snapshot) or its summary
/// omits the key, i.e. when the metadata has no opinion to contradict.
pub(crate) fn declared_records(t: &Table) -> Option<i64> {
t.metadata()
.current_snapshot()?
.summary()
.additional_properties
.get("total-records")?
.parse::<i64>()
.ok()
}
/// **Refuse a data-bearing read that came back empty over a table whose own
/// metadata says it holds rows.** ONE predicate, called by both
/// [`IcebergWarehouse::scan_preview_bounded`] and
/// [`IcebergWarehouse::scan_arrow_bounded`] (LAW 5: one gate, two callers).
///
/// `Ok(vec![])` and "this table is empty" are the same value, and that is the
/// warehouse's signature failure: on 2026-08-22 a read of the live `symbol_facts`
/// answered `Ok`, 0 rows, no error, no warning, while the table's metadata
/// recorded 3 660 247 records over 32 data files — and the `mimir analyze`
/// lenses downstream reported "no duplicates" and exited 0. A reader that finds
/// nothing where the writer recorded millions has NOT found an empty table; it
/// has failed, and it must say which table and by how much.
///
/// Deliberately narrow, so it can only ever fire on a genuine contradiction:
/// * silent when any row came back (a partial read is a different problem —
/// `cut` already reports it),
/// * silent when the scan was `truncated` by its own byte/row ceiling before
/// the first row landed,
/// * silent when the table advertises 0 records or has no snapshot at all — an
/// honestly empty table stays honestly empty.
pub(crate) fn refuse_silent_empty(
table: &str,
rows_read: usize,
truncated: bool,
declared: Option<i64>,
) -> Result<()> {
if rows_read > 0 || truncated {
return Ok(());
}
match declared {
Some(n) if n > 0 => anyhow::bail!(
"nornir: read of table `{table}` returned 0 rows, but that table's own current \
Iceberg snapshot advertises {n} record(s). That is a BROKEN READ, not an empty \
table, and it is refused instead of being handed on as success — a lens fed this \
answer reports \"no findings\" and exits 0. Check, in this order: (1) the \
warehouse root — `<root>/warehouse/<namespace>` is the DATA dir, the ROOT is the \
directory holding `catalog.redb`; (2) that the parquet under `<root>/warehouse/` \
is READABLE by this user (the server may run as a different one); (3) that the \
data files the manifest names still exist on disk."
),
_ => Ok(()),
}
}
/// Copy the locked `live` `catalog.redb` → `dst` so the copy is a **complete,
/// openable** image reflecting the latest committed catalog state, even while
/// the `nornir-server` is committing to `live`.
///
/// Why a plain `fs::copy` is not safe here: `live` is held open and mutated in
/// place by the server. redb grows the file in **regions** — to add a region it
/// updates the on-disk header to advertise the larger layout *and* extends the
/// file. Those two steps are not atomic w.r.t. an outside `fs::copy`: if the
/// copy snapshots the file after the header was written but before (or partway
/// through) the file was extended, the copy's byte length is **shorter than the
/// layout its own header claims**. Opening that copy trips redb's
/// `raw_file_len() >= header.layout().len()` invariant — in redb 2.6 an
/// **assertion that panics** (`page_manager.rs`), crashing the reader, or, in
/// milder mismatches, forces a recovery to an older/short state. The reader the
/// caller believes is "latest" then crashes or reads empty (the fat-viz
/// empty-Timeline bug).
///
/// The fix is to take the copy only when the source is **size-stable** — copy,
/// then re-stat the source and re-copy until its length does not change across
/// the copy window. A window with no region growth cannot leave the header
/// advertising bytes the copy is missing, so the resulting copy always satisfies
/// redb's length invariant and opens to the latest committed transaction. (redb
/// is robust to a *longer* file via recovery, and to a torn-but-complete copy
/// via its two-phase god-slot — both safe; only the short-file race is the
/// hazard, and size-stability closes exactly that.) Growth is rare and the
/// server fsyncs per commit, so this converges in one pass in the steady state
/// and a few retries while a region is being added; bounded by `MAX_ATTEMPTS`.
fn copy_catalog_consistent(live: &Path, dst: &Path) -> Result<()> {
copy_catalog_consistent_with(live, dst, |s, d| {
std::fs::copy(s, d)
.map(|_| ())
.with_context(|| format!("copy {} -> {}", s.display(), d.display()))
})
}
/// Core of [`copy_catalog_consistent`] with the copy step injected, so tests can
/// deterministically drive the size-stability retry (force a short/growing copy
/// on early attempts, a stable one later) instead of racing a real writer.
fn copy_catalog_consistent_with(
live: &Path,
dst: &Path,
mut copy: impl FnMut(&Path, &Path) -> Result<()>,
) -> Result<()> {
const MAX_ATTEMPTS: usize = 64;
let len_of = |p: &Path| -> Result<u64> {
Ok(std::fs::metadata(p)
.with_context(|| format!("stat {}", p.display()))?
.len())
};
for attempt in 0..MAX_ATTEMPTS {
let before = len_of(live)?;
copy(live, dst)?;
let after = len_of(live)?;
// Accept only a copy taken across a window with no region growth AND
// whose own length is not short relative to the source: then it cannot
// be shorter than the header's advertised layout, so redb opens it to
// the latest committed transaction. (`len_of(dst) >= after` rejects a
// torn/short copy even if the source size happened to look unchanged.)
if before == after && len_of(dst)? >= after {
return Ok(());
}
// A region grew (or the copy came up short) across the window; let the
// extension land, then retry.
std::thread::sleep(std::time::Duration::from_millis(2 + attempt as u64 / 4));
}
anyhow::bail!(
"snapshot of locked catalog {} never reached a size-stable window after \
{MAX_ATTEMPTS} attempts (catalog growing under a very hot writer)",
live.display()
)
}
impl Drop for IcebergWarehouse {
fn drop(&mut self) {
// Dropping a tokio Runtime synchronously inside another tokio
// runtime (e.g. when this warehouse is owned by an async MCP
// server) panics. Hand the inner runtime off to be shut down
// in the background so the outer runtime's drop is safe.
if let Some(rt) = self.rt.take() {
rt.shutdown_background();
}
}
}
impl IcebergWarehouse {
pub fn open(root: &Path) -> Result<Self> {
std::fs::create_dir_all(root)
.with_context(|| format!("create warehouse root {}", root.display()))?;
let warehouse_dir = root.join("warehouse");
std::fs::create_dir_all(&warehouse_dir)
.with_context(|| format!("create warehouse data dir {}", warehouse_dir.display()))?;
let db_path = root.join("catalog.redb");
let warehouse_uri = format!("file://{}", warehouse_dir.canonicalize()?.display());
Self::open_with_catalog(root, &db_path, &warehouse_uri, None)
}
/// **Read-only** open of an EXISTING warehouse. It copies `catalog.redb`
/// aside into a temp dir and opens *that* copy, yielding a point-in-time
/// snapshot of the catalog. Table data files are referenced by absolute
/// `file://` paths in the catalog metadata, so they still resolve to the
/// live warehouse data — only the catalog index is copied (typically a few
/// MB). It therefore never takes the exclusive redb write lock, never heals,
/// never creates, and cannot deadlock against — or disturb — a live
/// `nornir-server` holding the same warehouse.
///
/// This lets dev-side `nornir` CLI reads (the `docs_fresh` gate, docs
/// render, `release trace`, the `mimir analyze` lenses, the viz panes)
/// coexist with a running server.
///
/// Callers that mutate the warehouse must use [`open`](Self::open) — a
/// snapshot copy is read-only by construction (writes would land in the
/// throwaway temp dir, not the live catalog), and [`ensure_writable`] makes
/// that refusal explicit.
///
/// # Why it refuses a root with no `catalog.redb`, and why it stopped
/// # calling [`open`](Self::open) first
///
/// It used to be "try [`open`](Self::open), fall back to a snapshot only on
/// a lock error", which is two defects wearing one name:
///
/// 1. **It was not read-only.** On an uncontended warehouse it took the
/// redb *write* lock and ran `heal_all_tables`, which rewrites committed
/// catalog pointers. A function called `open_read_only` mutated the
/// thing it was asked only to read.
/// 2. **It MANUFACTURED an empty warehouse at any path handed to it.**
/// [`open`](Self::open) does `create_dir_all(root)` +
/// `create_dir_all(root/warehouse)` + a fresh `catalog.redb`, and then
/// `ensure_layout` registers every known table — all EMPTY. Every later
/// scan then honestly returned `Ok(0 rows)`: of a warehouse that had been
/// created seconds earlier, not of the one the caller meant.
///
/// MEASURED on oden 2026-08-22: pointing a read at
/// `<root>/warehouse/nornir` — the DATA dir, one level too deep — instead of
/// erroring, created `catalog.redb` (3.6 MB) plus a whole
/// `<root>/warehouse/nornir/warehouse/nornir/<table>/metadata/00000-*.json`
/// tree inside the LIVE server's warehouse directory, listed 85 tables, and
/// answered `Ok`, 0 rows, no error for `symbol_facts` — whose real metadata
/// records 3 660 247 rows over 32 data files. That read was diagnosed for
/// hours as catalog corruption. It was this function creating the emptiness
/// it then reported. A read that cannot find a warehouse must say so.
pub fn open_read_only(root: &Path) -> Result<Self> {
let db = root.join("catalog.redb");
// Distinguish "not there" from "cannot see it": `is_file()` is false for
// BOTH, and reporting a permission failure as a missing warehouse sends
// the reader hunting for the wrong thing. MEASURED 2026-08-22: the live
// warehouse sits under `/home/dwarves/.nornir`, mode 0700 — every read of
// it from another account stops at the directory, not at the catalog.
if let Err(e) = std::fs::metadata(&db) {
if e.kind() != std::io::ErrorKind::NotFound {
return Err(anyhow::Error::new(e)).with_context(|| {
format!(
"cannot read `catalog.redb` at {} — this is an ACCESS failure, not an \
empty warehouse. The server may run as a different user (the nornir \
data root is mode 0700), so check the uid this process runs as before \
concluding anything about the data.",
db.display()
)
});
}
anyhow::bail!(
"no `catalog.redb` at {} — refusing to read it as a warehouse. \
A nornir warehouse ROOT is the directory that HOLDS `catalog.redb` \
next to a `warehouse/` data dir; `<root>/warehouse/<namespace>` is \
the DATA dir and is one level too deep. (For a workspace the root is \
`<registry>/<workspace>/builds` — see `nornir_config::warehouse_root`.) \
Nothing was created: a read-only open never brings a warehouse into \
existence, because an empty warehouse it made itself is indistinguishable \
from an empty answer.",
root.display()
);
}
Self::open_snapshot(root).with_context(|| {
format!("open read-only catalog snapshot of warehouse at {}", root.display())
})
}
/// Copy `catalog.redb` aside and open the copy. Used by
/// [`open_read_only`](Self::open_read_only) when the live catalog is locked.
fn open_snapshot(root: &Path) -> Result<Self> {
let live_db = root.join("catalog.redb");
if !live_db.exists() {
// Nothing to snapshot — the lock error wasn't about our catalog.
anyhow::bail!("no catalog.redb at {} to snapshot", live_db.display());
}
// Data files are referenced by absolute paths in the catalog, so the
// data dir must stay pointed at the *real* warehouse. Only the catalog
// index file is copied into the temp dir.
let tmp = tempfile::Builder::new()
.prefix("nornir-catalog-snapshot-")
.tempdir()
.context("create temp dir for catalog snapshot")?;
let snap_db = tmp.path().join("catalog.redb");
// The live `catalog.redb` is being mutated *under us* by the server that
// holds the lock. A single blind `fs::copy` can race redb's region growth
// and capture a file shorter than its own header's advertised layout —
// which redb then either panics on (its length assertion) or recovers to
// a stale/short state, so the reader crashes or sees an empty Timeline.
// `copy_catalog_consistent` instead snapshots only across a size-stable
// window, yielding a complete image that opens to the latest committed
// state (see its doc comment for the full rationale).
copy_catalog_consistent(&live_db, &snap_db).with_context(|| {
format!(
"copy locked catalog {} -> snapshot {}",
live_db.display(),
snap_db.display()
)
})?;
let warehouse_dir = root.join("warehouse");
let warehouse_uri = format!("file://{}", warehouse_dir.canonicalize()?.display());
Self::open_with_catalog(root, &snap_db, &warehouse_uri, Some(tmp))
}
/// Shared constructor: open `db_path` as the redb catalog while pointing
/// table data at `warehouse_uri`. `snapshot` ties an owning temp dir to the
/// warehouse lifetime when `db_path` is a copied-aside snapshot.
fn open_with_catalog(
root: &Path,
db_path: &Path,
warehouse_uri: &str,
snapshot: Option<tempfile::TempDir>,
) -> Result<Self> {
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.context("build tokio runtime")?;
let catalog = rt.block_on(async {
RedbCatalogBuilder::default()
.db_path(db_path)
.warehouse_location(warehouse_uri)
.with_storage_factory(Arc::new(LocalFsStorageFactory))
.load("nornir", HashMap::new())
.await
})?;
let catalog = Arc::new(catalog);
let namespace = NamespaceIdent::new(NAMESPACE.to_string());
// Crash recovery, BEFORE anything reads or appends. An unclean shutdown
// can leave the durable catalog pointer aimed at a metadata JSON that
// parses but names a **0-byte manifest** — every later scan of that table
// then dies on `Failed to load manifest … DataInvalid`, forever, because
// nothing re-writes a committed pointer. skade's `heal_all_tables` rolls
// each such pointer back to the newest snapshot whose manifests are all
// whole; the next append re-adds what the torn commit was carrying.
//
// This is wired here because it was wired NOWHERE: `heal_table` shipped
// with zero callers, and oden's live warehouse consequently stayed torn
// from the 2026-08-16 20:11 crash to 2026-08-22 — six days of
// `nornir-embed … embed FAILED`, 19 repos every cycle, past a repair that
// already existed. MEASURED on that warehouse: 0.46 s for all 73 tables
// in a debug build, so it is unconditional on the write path.
//
// NOT on a read-only snapshot open: healing a copied-aside catalog would
// rewind the throwaway copy and hide the torn state from the live one.
if snapshot.is_none() {
for (ident, outcome) in rt.block_on(catalog.heal_all_tables(&namespace))? {
match outcome {
HealOutcome::Healed { from, to } => eprintln!(
"WARNING: nornir: table `{}` pointed at torn metadata after an unclean \
shutdown; rolled back {} -> {}. Rows committed after that snapshot are \
orphaned and will be re-appended by the next populate.",
ident.name(),
from.rsplit('/').next().unwrap_or(&from),
to.rsplit('/').next().unwrap_or(&to),
),
HealOutcome::Unrecoverable { location } => eprintln!(
"ERROR: nornir: table `{}` points at torn metadata ({location}) and NO \
intact snapshot was found in its metadata dir — every read of it will \
fail until it is dropped and repopulated.",
ident.name(),
),
HealOutcome::Healthy | HealOutcome::Unknown | HealOutcome::Skipped => {}
}
}
}
// NOT on a read-only snapshot open, for the same reason `heal_all_tables`
// is not: only `catalog.redb` is copied aside — every table's
// `warehouse_uri` still points at the LIVE data dir. `ensure_layout`
// CREATES each missing table, so running it here would write fresh
// `00000-*.metadata.json` files into the live warehouse from a path that
// is supposed to be pure read, and would register those empty tables in
// the throwaway catalog so the caller reads 0 rows and is told nothing.
// A snapshot reader takes the layout exactly as the writer left it; a
// table genuinely missing from the catalog surfaces as a named
// `load_table` error, which is the honest answer.
if snapshot.is_none() {
rt.block_on(ensure_layout(&catalog, &namespace))?;
}
Ok(Self {
catalog,
rt: Some(rt),
namespace,
root: root.to_path_buf(),
_snapshot: snapshot,
})
}
/// The warehouse root this was opened from (parent of `warehouse/` +
/// `catalog.redb`). Lets the timeline locate the sibling `../git/` clones
/// for the git-history fallback when a workspace has no release backlog.
pub fn root(&self) -> &Path {
&self.root
}
/// True when this warehouse was opened from a copied-aside read-only
/// snapshot of a locked `catalog.redb` (via [`open_read_only`](Self::open_read_only)).
///
/// Such a warehouse is **read-only**, and dangerously so if unguarded: only
/// `catalog.redb` is copied aside — the table `warehouse_uri` still points at
/// the **live** data dir. A write would therefore drop new parquet into the
/// live warehouse while committing the snapshot pointer into the throwaway
/// temp catalog → silent data loss + orphan files. Every mutating entry point
/// guards on this via [`ensure_writable`](Self::ensure_writable).
pub fn is_snapshot(&self) -> bool {
self._snapshot.is_some()
}
/// Refuse to mutate a read-only snapshot warehouse (see [`is_snapshot`](Self::is_snapshot)).
/// Called at the top of every write/schema-mutation entry point so a copied-aside
/// catalog can never silently lose writes into the live data dir.
fn ensure_writable(&self) -> Result<()> {
if self.is_snapshot() {
anyhow::bail!(
"read-only snapshot warehouse: refusing to mutate a copied-aside catalog \
(the live catalog is locked by another process, e.g. nornir-server)"
);
}
Ok(())
}
pub(crate) fn table_ident(&self, name: &str) -> TableIdent {
TableIdent::new(self.namespace.clone(), name.to_string())
}
/// Load `name` for an APPEND, self-healing a blank/torn metadata JSON by
/// dropping + recreating the (0-readable-row) table. The method form for
/// sibling modules (e.g. `index::snapshot`) that write extra tables without
/// the catalog/namespace in hand. See the free `load_table_for_append`.
pub(crate) async fn load_table_for_append(
&self,
name: &str,
schema: iceberg::spec::Schema,
partition_cols: &[&str],
) -> Result<Table> {
load_table_for_append(&self.catalog, &self.namespace, name, schema, partition_cols).await
}
}
impl IcebergWarehouse {
/// Access the underlying catalog. Used by sibling modules
/// (e.g. `warehouse::dep_graph`) that read/write additional tables
/// without going through the `Warehouse` trait.
pub fn catalog(&self) -> &Arc<RedbCatalog> {
&self.catalog
}
/// The Iceberg namespace every warehouse table lives under. Used by
/// [`crate::warehouse::sql`] to build a DataFusion schema over the tables.
pub fn namespace(&self) -> &NamespaceIdent {
&self.namespace
}
/// Block on `fut` using this warehouse's runtime. Sibling modules
/// share the runtime so the sync `Warehouse` trait stays sync.
pub fn block_on<F: std::future::Future>(&self, fut: F) -> F::Output {
self.rt.as_ref().expect("rt present").block_on(fut)
}
}
async fn ensure_layout(catalog: &RedbCatalog, ns: &NamespaceIdent) -> Result<()> {
if !catalog.namespace_exists(ns).await? {
catalog.create_namespace(ns, HashMap::new()).await?;
}
// bench_runs is the highest-traffic scoped-read table — partition by `repo`
// so `query_bench_runs(for_repo(..))` prunes to one repo's files.
create_partitioned_table_if_missing(
catalog,
ns,
TABLE_BENCH_RUNS,
iceberg_schema::bench_runs()?,
&["repo"],
)
.await?;
create_table_if_missing(
catalog,
ns,
TABLE_BENCH_RESULTS,
iceberg_schema::bench_results()?,
)
.await?;
// Per-bench telemetry: partitioned by `repo` like bench_runs so per-repo
// reads prune to one repo's files.
create_partitioned_table_if_missing(
catalog,
ns,
TABLE_BENCH_TELEMETRY,
iceberg_schema::bench_telemetry()?,
&["repo"],
)
.await?;
create_table_if_missing(
catalog,
ns,
TABLE_TEST_OUTCOMES,
iceberg_schema::test_outcomes()?,
)
.await?;
// C6 test-matrix: standalone green/red board (not bench-coupled). Partition
// by `repo` so `nornir test history <repo>` + the viz Test pane prune to one
// repo's files, exactly like bench_runs.
create_partitioned_table_if_missing(
catalog,
ns,
TABLE_TEST_RESULTS,
iceberg_schema::test_results()?,
&["repo"],
)
.await?;
// build-provenance ledger — partition by `repo` like test_results.
create_partitioned_table_if_missing(
catalog,
ns,
TABLE_BUILD_PROVENANCE,
iceberg_schema::build_provenance()?,
&["repo"],
)
.await?;
// build-info record (§f3) — the exploded columnar record ABOUT each build,
// the sibling of build_provenance. Unpartitioned; reads scope by `build_id`.
create_table_if_missing(catalog, ns, TABLE_BUILD_INFO, iceberg_schema::build_info()?).await?;
create_table_if_missing(
catalog,
ns,
TABLE_DEP_GRAPH_EDGES,
iceberg_schema::dep_graph_edges()?,
)
.await?;
create_table_if_missing(
catalog,
ns,
TABLE_RELEASE_LINEAGE,
iceberg_schema::release_lineage()?,
)
.await?;
// Release-op DAG. Reads scope by `run_id` (one release run) and roll the
// long rows into a topo view; unpartitioned (a run is a handful of rows, the
// residual `run_id` filter in the reader is enough).
create_table_if_missing(
catalog,
ns,
TABLE_RELEASE_EVENTS,
iceberg_schema::release_events()?,
)
.await?;
// Reversible release-change ledger (task #32) — unpartitioned, reads scope by
// `release_id` (one run's records), reversed newest-first by `undo`.
create_table_if_missing(
catalog,
ns,
TABLE_RELEASE_CHANGES,
iceberg_schema::release_changes()?,
)
.await?;
create_table_if_missing(
catalog,
ns,
TABLE_FUNNEL_EVENTS,
iceberg_schema::funnel_events()?,
)
.await?;
// Airgap-op DAG (EPIC AIRGAP / AIRGAP8). Same shape/reasoning as
// release_events — unpartitioned, reads scope by `run_id`.
create_table_if_missing(
catalog,
ns,
TABLE_AIRGAP_EVENTS,
iceberg_schema::airgap_events()?,
)
.await?;
// Clone/fetch/republish populate outcomes (PLAN #6) — unpartitioned,
// append-only; reads scope by workspace + recency.
create_table_if_missing(
catalog,
ns,
TABLE_CLONE_EVENTS,
iceberg_schema::clone_events()?,
)
.await?;
// Per-STEP populate outcome stream (EPIC #21): snapshot / deep-scan /
// test-surface / arch-generate / populate. Sibling of clone_events.
create_table_if_missing(
catalog,
ns,
TABLE_WORKSPACE_EVENTS,
iceberg_schema::workspace_events()?,
)
.await?;
// Artifact snapshot/blob pairs. Snapshots are looked up by `repo` (find the
// latest for a repo); blobs are fetched by `snapshot_id` (one snapshot's
// bytes) — partition each by its read key so restore prunes to one snapshot
// instead of O(all bytes ever). (`*_blobs` carries no `repo` column.)
for (snap_name, blob_name) in [
(TABLE_TANTIVY_INDEX_SNAPSHOTS, TABLE_TANTIVY_INDEX_BLOBS),
(TABLE_DWARF_SNAPSHOTS, TABLE_DWARF_BLOBS),
(TABLE_GIMLI_SNAPSHOTS, TABLE_GIMLI_BLOBS),
(TABLE_RUSTDOC_JSON_SNAPSHOTS, TABLE_RUSTDOC_JSON_BLOBS),
(TABLE_DOCS_INDEX_SNAPSHOTS, TABLE_DOCS_INDEX_BLOBS),
] {
create_partitioned_table_if_missing(
catalog,
ns,
snap_name,
iceberg_schema::artifact_snapshots()?,
&["repo"],
)
.await?;
create_partitioned_table_if_missing(
catalog,
ns,
blob_name,
iceberg_schema::artifact_blobs()?,
&["snapshot_id"],
)
.await?;
}
// Rendered-document export history (inline bytes).
create_table_if_missing(
catalog,
ns,
TABLE_DOC_EXPORTS,
iceberg_schema::doc_exports()?,
)
.await?;
// Vector / semantic-search tables.
create_table_if_missing(catalog, ns, TABLE_EMBEDDINGS, iceberg_schema::embeddings()?).await?;
create_table_if_missing(
catalog,
ns,
TABLE_EMBEDDING_SNAPSHOTS,
iceberg_schema::embedding_snapshots()?,
)
.await?;
create_table_if_missing(
catalog,
ns,
TABLE_EMBEDDING_MANIFEST,
iceberg_schema::embedding_manifest()?,
)
.await?;
create_table_if_missing(catalog, ns, TABLE_EMBEDDING_NODE_EVENTS, iceberg_schema::embedding_node_events()?).await?;
create_table_if_missing(catalog, ns, TABLE_EMBEDDING_NODE_RUNS, iceberg_schema::embedding_node_runs()?).await?;
create_table_if_missing(catalog, ns, TABLE_EMBEDDINGS_RING, iceberg_schema::embeddings()?).await?;
create_table_if_missing(catalog, ns, TABLE_FN_SHAPE_EVENTS, iceberg_schema::fn_shape_events()?).await?;
create_table_if_missing(catalog, ns, TABLE_EMBEDDINGS_NORM, iceberg_schema::embeddings()?).await?;
create_table_if_missing(catalog, ns, TABLE_FN_ROLE_VECTORS, iceberg_schema::fn_role_vectors()?).await?;
create_table_if_missing(catalog, ns, TABLE_EMBEDDINGS_GROUPS, iceberg_schema::embeddings()?).await?;
// Cargo-pipeline tables (feature 1-15).
for (name, schema) in [
(TABLE_PATH_DEP_AUDITS, iceberg_schema::path_dep_audits()?),
(
TABLE_PATCH_STRIP_EVENTS,
iceberg_schema::patch_strip_events()?,
),
(TABLE_PUBLISH_ATTEMPTS, iceberg_schema::publish_attempts()?),
(
TABLE_CRATE_METADATA_CHECKS,
iceberg_schema::crate_metadata_checks()?,
),
(
TABLE_CRATE_TARBALL_STATS,
iceberg_schema::crate_tarball_stats()?,
),
(TABLE_YANK_EVENTS, iceberg_schema::yank_events()?),
(TABLE_SEMVER_DIFFS, iceberg_schema::semver_diffs()?),
(
TABLE_LINKS_DECLARATIONS,
iceberg_schema::links_declarations()?,
),
(
TABLE_RESOLVED_FEATURES,
iceberg_schema::resolved_features()?,
),
(TABLE_MSRV_PROBES, iceberg_schema::msrv_probes()?),
(
TABLE_TEST_IMPACTED_SELECTIONS,
iceberg_schema::test_impacted_selections()?,
),
(TABLE_TEST_QUARANTINES, iceberg_schema::test_quarantines()?),
(TABLE_RELEASE_COMMITS, iceberg_schema::release_commits()?),
(TABLE_REGISTRY_MIRRORS, iceberg_schema::registry_mirrors()?),
(
TABLE_VERSION_BUMP_PLANS,
iceberg_schema::version_bump_plans()?,
),
(
TABLE_VERSION_BUMP_TARGETS,
iceberg_schema::version_bump_targets()?,
),
(TABLE_MCP_REQUESTS, iceberg_schema::mcp_requests()?),
(TABLE_VULN_FINDINGS, iceberg_schema::vuln_findings()?),
// autonom completeness gate (n-005): per surface-node verdict rows.
(TABLE_SURFACE_COVERAGE, iceberg_schema::surface_coverage()?),
] {
create_table_if_missing(catalog, ns, name, schema).await?;
}
// Per-repo fact tables are written by single-repo producers and read scoped
// to a repo, so partition them by `repo` to prune.
for (name, schema) in [
// Neighbour lists are produced per repo and read per repo, so they
// partition by `repo`.
(TABLE_KNN_NEIGHBORS, iceberg_schema::knn_neighbors()?),
// Per-symbol semver diffs are read scoped to one world repo
// (`<crate>@<to_version>`) exactly like the knowledge tables they join
// against, so partition by `repo` to prune the same way.
(
TABLE_SEMVER_SYMBOL_DIFFS,
iceberg_schema::semver_symbol_diffs()?,
),
(TABLE_GIT_HEAT_FACTS, iceberg_schema::git_heat_facts()?),
// SBOM components are read scoped to a single `repo`, so partition them
// by `repo` to prune scans.
(TABLE_SBOM_COMPONENTS, iceberg_schema::sbom_components()?),
// Source-based coverage (feature N) — read scoped to a single `repo`
// (the viz Test pane + docs render one repo's latest run), so partition
// by `repo` like the other per-repo fact tables.
(TABLE_COVERAGE, iceberg_schema::coverage()?),
(TABLE_COVERAGE_FN, iceberg_schema::coverage_fn()?),
// UI snapshot reference store (render oracle) — read scoped to one repo.
(TABLE_UI_SNAPSHOTS, iceberg_schema::ui_snapshots()?),
// Traced runs and profiles (Runa `Trace` / `Profile`): listed per repo.
(super::fn_trace::TABLE_FN_TRACE_RUNS, super::fn_trace::runs_schema()?),
(super::fn_trace::TABLE_FN_PROFILE, super::fn_trace::profile_schema()?),
] {
create_partitioned_table_if_missing(catalog, ns, name, schema, &["repo"]).await?;
}
// A traced run's events are read one run at a time (the 3D map's replay
// window), so partition by `run_id`: a read prunes straight to that run.
create_partitioned_table_if_missing(
catalog,
ns,
super::fn_trace::TABLE_FN_TRACE_EVENTS,
super::fn_trace::events_schema()?,
&["run_id"],
)
.await?;
// Lens answers are read for ONE room version (the current one), so partition
// by `room_version`: a restarted server's read prunes past every old
// version's bodies instead of scanning them.
create_partitioned_table_if_missing(
catalog,
ns,
TABLE_LENS_ANSWERS,
iceberg_schema::lens_answers()?,
&["room_version"],
)
.await?;
// Binary-analysis facts (task #65): the header + section + symbol size rows
// are all read scoped to ONE artifact by `content_sha256` (the content_id
// join key), and one `analyze` writes exactly one artifact's rows in a single
// batch, so partition by `content_sha256` — one append = one partition, a
// read-back prunes straight to that artifact's files.
for (name, schema) in [
(TABLE_BINARY_FACTS, iceberg_schema::binary_facts()?),
(TABLE_SECTION_SIZES, iceberg_schema::section_sizes()?),
(TABLE_SYMBOL_SIZES, iceberg_schema::symbol_sizes()?),
] {
create_partitioned_table_if_missing(catalog, ns, name, schema, &["content_sha256"]).await?;
}
// Artifact-to-artifact diff (task #65 slice 2): one diff writes every change
// row for a (base, new) pair in one batch, and a diff read-back scopes to the
// NEW artifact then filters the base — so partition by `to_content_sha256`:
// one append = one partition, a read prunes straight to that pair's files.
create_partitioned_table_if_missing(
catalog,
ns,
TABLE_ARTIFACT_DIFF,
iceberg_schema::artifact_diff()?,
&["to_content_sha256"],
)
.await?;
// H5 bake-off: one row per (run_id, model, prompt) answer. A bake-off writes
// every model's row in one batch, and the leaderboard reads a whole run back,
// so partition by `run_id` — one append = one partition, and a run read prunes
// to that run's files (a per-model read does a residual `model` filter).
create_partitioned_table_if_missing(
catalog,
ns,
TABLE_AGENT_MODEL_RUNS,
iceberg_schema::agent_model_runs()?,
&["run_id"],
)
.await?;
// N5 viz action trail. A viz session appends its actions as they happen and
// reads its own recent history back scoped to one `session_id`, so partition
// by `session_id` — a per-session read prunes to that launch's files (a
// cross-session read does a residual filter).
create_partitioned_table_if_missing(
catalog,
ns,
TABLE_VIZ_ACTIONS,
iceberg_schema::viz_actions()?,
&["session_id"],
)
.await?;
// EPIC SAGA (task #25) record/replay. A capture drain appends one session's
// events and the matrix reads one session's rows back, so partition by
// `session_id` — a per-session read prunes to that journey's files.
create_partitioned_table_if_missing(
catalog,
ns,
TABLE_SAGA_EVENTS,
iceberg_schema::saga_events()?,
&["session_id"],
)
.await?;
// Yggdrasil infra crown + runtime band. Snapshot-enveloped and read scoped
// by `workspace_name` (a constellation is a handful of machines/services), so
// unpartitioned like `dep_graph_edges` — the residual `workspace_name` filter
// in the reader is enough.
create_table_if_missing(
catalog,
ns,
TABLE_INFRA_NODES,
iceberg_schema::infra_nodes()?,
)
.await?;
create_table_if_missing(
catalog,
ns,
TABLE_INFRA_EDGES,
iceberg_schema::infra_edges()?,
)
.await?;
create_table_if_missing(
catalog,
ns,
TABLE_RUNTIME_SPANS,
iceberg_schema::runtime_spans()?,
)
.await?;
// Feature ACTIVATION graph from `Cargo.toml` (table 72, appended at the END of
// the registration order so every existing ordinal keeps its number). One
// manifest parse writes one repo's edges and every read is scoped to one repo
// (the UI asks "for THIS repo, what pulls dep X"), so partition by `repo` —
// the same treatment as `feature_gate_facts`.
create_partitioned_table_if_missing(
catalog,
ns,
TABLE_FEATURE_DEP_EDGES,
iceberg_schema::feature_dep_edges()?,
&["repo"],
)
.await?;
Ok(())
}
async fn create_table_if_missing(
catalog: &RedbCatalog,
ns: &NamespaceIdent,
name: &str,
schema: iceberg::spec::Schema,
) -> Result<()> {
create_partitioned_table_if_missing(catalog, ns, name, schema, &[]).await
}
/// Like [`create_table_if_missing`] but lays the table out partitioned by the
/// identity of `partition_cols` (each must be a top-level column in `schema`).
/// Partitioning makes scoped reads prune whole data files at plan time instead
/// of scanning all history (see plan.md §"Iceberg warehouse efficiency").
async fn create_partitioned_table_if_missing(
catalog: &RedbCatalog,
ns: &NamespaceIdent,
name: &str,
schema: iceberg::spec::Schema,
partition_cols: &[&str],
) -> Result<()> {
let ident = TableIdent::new(ns.clone(), name.to_string());
if catalog.table_exists(&ident).await? {
// Existing table: do NOT touch it here. Additive column evolution happens
// safely at APPEND time (each append path calls `ensure_table_schema`, which
// only ADDS missing columns and preserves rows — that's the S6b scip fix).
// An earlier version reconciled "incompatible" drift by dropping+recreating,
// but the type-equality check FALSE-POSITIVED on identical round-tripped
// schemas (nested/list element ids differ across persist) and silently WIPED
// real-data tables (funnel_events / release_events / embeddings). Never drop
// a table out from under its data: an incompatible change must be an explicit,
// per-table decision, not a blanket auto-drop.
return Ok(());
}
// Declare the write-time sort key (if any) as the table's Iceberg SortOrder,
// so the scan planner + future compaction know the physical row order the
// writer maintains (the data-skipping trick — see `warehouse_sort_cols`).
let sort_cols = warehouse_sort_cols(name);
let sort_order = sort_order_for(&schema, sort_cols)?;
// `TableCreation::builder()` is a typed-builder (each setter changes the
// type), so branch rather than mutate a single binding.
let creation = if partition_cols.is_empty() {
TableCreation::builder()
.name(name.to_string())
.schema(schema)
.sort_order(sort_order)
.build()
} else {
// Build the (unbound) spec from a borrow before `schema` is moved in.
let spec = identity_partition_spec(&schema, partition_cols)?;
TableCreation::builder()
.name(name.to_string())
.schema(schema)
.partition_spec(spec)
.sort_order(sort_order)
.build()
};
catalog.create_table(ns, creation).await?;
Ok(())
}
/// Load `ident` for an APPEND, **self-healing a blank/torn metadata JSON**.
///
/// A torn-commit blank `metadata.json` makes iceberg-rust's `load_table` fail
/// with `DataInvalid: EOF while parsing a value at line 1 column 0` — and unlike
/// a *read* (which degrades to 0 rows), a *write* can't proceed against an
/// unloadable table, so the `?` used to bubble and kill `Workspaces.Fetch`'s
/// republish. But a blank-metadata table has **no committed snapshot a reader can
/// see** — i.e. 0 readable rows — so dropping and recreating it (with the same
/// schema + partitioning) loses nothing readable and gets the writer unstuck.
/// This is the warehouse-native write seam: "torn metadata ⇒ fresh empty table",
/// never a fatal append. `partition_cols` must match how the table was created.
pub(crate) async fn load_table_for_append(
catalog: &RedbCatalog,
ns: &NamespaceIdent,
name: &str,
schema: iceberg::spec::Schema,
partition_cols: &[&str],
) -> Result<Table> {
let ident = TableIdent::new(ns.clone(), name.to_string());
match catalog.load_table(&ident).await {
Ok(t) => Ok(t),
Err(e) => {
let e: anyhow::Error = e.into();
if is_blank_metadata_json_error(&e) {
eprintln!(
"WARNING: nornir: table `{name}` has blank/torn Iceberg metadata JSON \
({e:#}); it has 0 readable rows, so dropping + recreating it empty so \
the append (republish) can proceed instead of failing the fetch."
);
// Best-effort drop (the table entry may already be half-gone), then
// recreate from the canonical schema + partitioning.
let _ = catalog.drop_table(&ident).await;
create_partitioned_table_if_missing(catalog, ns, name, schema, partition_cols)
.await?;
Ok(catalog.load_table(&ident).await?)
} else {
Err(e)
}
}
}
}
/// An identity [`UnboundPartitionSpec`] over `columns`, bound against `schema`
/// (to resolve names → field ids) then returned unbound for `TableCreation`.
fn identity_partition_spec(
schema: &iceberg::spec::Schema,
columns: &[&str],
) -> Result<UnboundPartitionSpec> {
let mut b = PartitionSpec::builder(Arc::new(schema.clone()));
for c in columns {
b = b.add_partition_field(c, (*c).to_string(), Transform::Identity)?;
}
Ok(b.build()?.into_unbound())
}
/// The **write-time sort key** for a warehouse table, by table name — the
/// predicate columns a point-lookup / scan-a-repo query filters on.
///
/// THE ICEBERG DATA-SKIPPING TRICK: rows are lexicographically sorted by these
/// columns *before* each Parquet file is written (see [`sort_batch_by`]), so
/// every data file's column min/max bounds end up TIGHT and largely
/// NON-OVERLAPPING across files. Parquet writes per-column statistics by default
/// (`EnabledStatistics::Page`), iceberg-rust folds those into the manifest's
/// per-file `lower_bounds`/`upper_bounds`, and the scan planner then prunes any
/// file whose `[lower,upper]` can't contain the queried value — a file-level
/// skip with no read. Unsorted, the bounds span the whole repo and nothing is
/// skippable; sorted, they partition the value space.
///
/// The same key is declared to the table as an Iceberg [`SortOrder`] at creation
/// (see [`sort_order_for`]) so the planner and any future compaction know the
/// physical order. Returns `&[]` for tables with no useful point-lookup key
/// (event logs, telemetry), which then write unsorted as before.
pub(crate) fn warehouse_sort_cols(table: &str) -> &'static [&'static str] {
match table {
// A neighbour read is "give me everything for THIS symbol, in rank
// order" — a point lookup on `src_symbol` followed by a short ordered
// run, so sorting by the pair keeps each symbol's list in one place.
TABLE_KNN_NEIGHBORS => &["repo", "src_symbol", "rank"],
// "what broke in this bump" is a point lookup on (repo, edge, symbol);
// `to_snapshot_id` leads inside the repo because a repo can carry several
// incoming edges and the query always names one.
TABLE_SEMVER_SYMBOL_DIFFS => &["repo", "to_snapshot_id", "symbol"],
// Binary-analysis reads scope to one artifact by content_sha256; the child
// size tables further sort by their natural roll-up key so a per-crate /
// per-section scan skips cleanly.
TABLE_BINARY_FACTS => &["content_sha256"],
// A replay window reads one run in sequence order.
super::fn_trace::TABLE_FN_TRACE_EVENTS => &["run_id", "seq_no"],
TABLE_SECTION_SIZES => &["content_sha256", "section_name"],
TABLE_SYMBOL_SIZES => &["content_sha256", "crate_name", "symbol_mangled"],
// Diff reads scope to the new artifact then the base; sort by the pair
// then `kind` so a per-kind scan (e.g. only abi_removed) skips cleanly.
TABLE_ARTIFACT_DIFF => &["to_content_sha256", "from_content_sha256", "kind"],
// Feature-activation edges have TWO point-lookup patterns and a single
// lexicographic key can only make ONE of them file-skippable:
// forward (repo, crate_name, feature) — "what does feature F of crate C turn on"
// reverse (repo, enables) — "which features pull dep X"
// `enables` LEADS (after the partition col) because the reverse lookup is
// the one that actually needs the skip. Reverse is a needle-in-haystack
// point probe over every edge in the repo, and `enables` is the
// highest-cardinality column here (every dep name + every feature name of
// every member — hundreds of distinct values vs a few dozen crate names),
// so it partitions the value space into many tight, non-overlapping
// per-file ranges. The forward lookup is NOT a needle: a consumer opening
// one crate's feature panel wants that crate's whole subgraph, which is a
// bounded read once `repo` has already pruned, and `crate_name`/`feature`
// still tie-break WITHIN each `enables` run so the residual filter stays
// cheap. Reversing this (crate_name first) would leave `enables` bounds
// spanning the whole repo in every file — the reverse question, the one
// this table exists to answer, would prune nothing.
TABLE_FEATURE_DEP_EDGES => &["repo", "enables", "crate_name", "feature"],
_ => &[],
}
}
/// Build a bound Iceberg [`SortOrder`] over `cols` (all ascending,
/// nulls-last, identity transform) against `schema`. Empty `cols` ⇒
/// [`SortOrder::unsorted_order`]. Declaring this on the table tells the scan
/// planner + future compaction the physical row order the writer maintains.
fn sort_order_for(
schema: &iceberg::spec::Schema,
cols: &[&str],
) -> Result<iceberg::spec::SortOrder> {
use iceberg::spec::{NullOrder, SortDirection, SortField, SortOrder, Transform};
if cols.is_empty() {
return Ok(SortOrder::unsorted_order());
}
let mut b = SortOrder::builder();
for c in cols {
let field = schema
.field_by_name(c)
.ok_or_else(|| anyhow!("sort column `{c}` is not in the schema"))?;
b.with_sort_field(
SortField::builder()
.source_id(field.id)
.transform(Transform::Identity)
.direction(SortDirection::Ascending)
.null_order(NullOrder::Last)
.build(),
);
}
Ok(b.build(schema)?)
}
/// Lexicographically sort a [`RecordBatch`]'s rows by the named string columns
/// (ascending). A no-op when `cols` is empty or any column is absent. This is
/// the write-time half of the data-skipping trick: tight per-file bounds.
pub(crate) fn sort_batch_by_pub(batch: &RecordBatch, cols: &[&str]) -> Result<RecordBatch> {
sort_batch_by(batch, cols)
}
fn sort_batch_by(batch: &RecordBatch, cols: &[&str]) -> Result<RecordBatch> {
use arrow::compute::{SortColumn, SortOptions, lexsort_to_indices, take_record_batch};
if cols.is_empty() || batch.num_rows() < 2 {
return Ok(batch.clone());
}
let mut sort_cols = Vec::with_capacity(cols.len());
for c in cols {
let Some(values) = batch.column_by_name(c) else {
// Schema drift / older table missing the column: skip sorting rather
// than fail the write.
return Ok(batch.clone());
};
sort_cols.push(SortColumn {
values: values.clone(),
options: Some(SortOptions {
descending: false,
nulls_first: false,
}),
});
}
let indices = lexsort_to_indices(&sort_cols, None)?;
Ok(take_record_batch(batch, &indices)?)
}
impl IcebergWarehouse {
/// Async variant of [`Warehouse::append_bench_run`] — call this
/// directly from async contexts (e.g. `release::pipeline`) to
/// avoid the nested `block_on` that would otherwise panic when
/// the sync trait method is invoked from inside a running tokio
/// runtime.
pub async fn append_bench_run_async(&self, repo: &str, run: &BenchRun) -> Result<Uuid> {
self.append_bench_run_keyed_async(repo, run, Uuid::new_v4(), None).await
}
/// [`Self::append_bench_run_async`] under a run id the PRODUCER minted and
/// the commit it measured — what a forge's Runa `Bench` step files
/// (`Telemetry.SubmitStepFeed`, keyed by (repo, commit)). Idempotent: a run
/// id `bench_runs` already holds is not written again (`Ok(false)`).
pub async fn append_bench_run_filed_async(
&self,
repo: &str,
run: &BenchRun,
run_id: Uuid,
commit: &str,
) -> Result<bool> {
self.ensure_writable()?;
if self.run_filed_async(TABLE_BENCH_RUNS, &run_id.to_string()).await? {
return Ok(false);
}
let commit = Some(commit).filter(|c| !c.trim().is_empty());
self.append_bench_run_keyed_async(repo, run, run_id, commit).await?;
Ok(true)
}
/// Does `table` already hold a row of `run_id`? (The idempotency check of a
/// producer-keyed filing: a retried job must not write its run twice.)
pub(crate) async fn run_filed_async(&self, table: &str, run_id: &str) -> Result<bool> {
let t = self.catalog.load_table(&self.table_ident(table)).await?;
let batches = read_runs_filtered(&t, &[run_id.to_string()]).await?;
Ok(batches.iter().any(|b| b.num_rows() > 0))
}
async fn append_bench_run_keyed_async(
&self,
repo: &str,
run: &BenchRun,
run_id: Uuid,
commit: Option<&str>,
) -> Result<Uuid> {
self.ensure_writable()?;
if run.machine.trim().is_empty() {
anyhow::bail!("BenchRun.machine is required");
}
let ts = resolve_timestamp(run)?;
let ident = self.table_ident(TABLE_BENCH_RUNS);
let table = self.catalog.load_table(&ident).await?;
// `commit` (field 8) joined the table 2026-10-17: a table created before
// it gains the column on this write; older rows read NULL.
let table = ensure_table_schema(&self.catalog, &ident, table, &iceberg_schema::bench_runs()?).await?;
let batch = build_bench_runs_batch(&table, run_id, repo, ts, run, commit)?;
append_batch(&self.catalog, table, batch).await?;
let results_batch_opt = {
let table = self
.catalog
.load_table(&self.table_ident(TABLE_BENCH_RESULTS))
.await?;
let batch = build_bench_results_batch(&table, run_id, run)?;
if batch.num_rows() > 0 {
Some((table, batch))
} else {
None
}
};
if let Some((table, batch)) = results_batch_opt {
append_batch(&self.catalog, table, batch).await?;
}
// Per-bench telemetry (CPU/mem footprint) lifted out of each result's
// `telem_*` metrics into its own partitioned table, keyed by run_id+bench.
let telem_batch_opt = {
let ident = self.table_ident(TABLE_BENCH_TELEMETRY);
let table = self.catalog.load_table(&ident).await?;
// Additive forward-evolution: a table created before the
// process-scoped columns (12-15) existed gains them here, on the
// first write that has values for them. Old rows keep NULL.
let table = ensure_table_schema(
&self.catalog,
&ident,
table,
&iceberg_schema::bench_telemetry()?,
)
.await?;
let batch = build_bench_telemetry_batch(&table, run_id, repo, run)?;
if batch.num_rows() > 0 {
Some((table, batch))
} else {
None
}
};
if let Some((table, batch)) = telem_batch_opt {
append_batch(&self.catalog, table, batch).await?;
}
let tests_batch_opt = {
let ident = self.table_ident(TABLE_TEST_OUTCOMES);
let table = self.catalog.load_table(&ident).await?;
let table = ensure_table_schema(&self.catalog, &ident, table, &iceberg_schema::test_outcomes()?).await?;
let batch = build_test_outcomes_batch(&table, run_id, run)?;
if batch.num_rows() > 0 {
Some((table, batch))
} else {
None
}
};
if let Some((table, batch)) = tests_batch_opt {
append_batch(&self.catalog, table, batch).await?;
}
Ok(run_id)
}
/// **File a Runa test step's outcomes** into `test_outcomes` — one row per
/// case under the producer's `run_id`, with the repo, crate, commit and full
/// status on the row itself (a test step has no `bench_runs` row). Idempotent
/// on `run_id`: `Ok(0)` when the run is already filed. Returns the rows written.
pub async fn append_step_test_outcomes_async(
&self,
repo: &str,
commit: &str,
run_id: &str,
cases: &[StepTestOutcome],
) -> Result<usize> {
self.ensure_writable()?;
if cases.is_empty() || self.run_filed_async(TABLE_TEST_OUTCOMES, run_id).await? {
return Ok(0);
}
let ident = self.table_ident(TABLE_TEST_OUTCOMES);
let table = self.catalog.load_table(&ident).await?;
let table = ensure_table_schema(&self.catalog, &ident, table, &iceberg_schema::test_outcomes()?).await?;
let batch = build_step_test_outcomes_batch(&table, repo, commit, run_id, cases)?;
append_batch(&self.catalog, table, batch).await?;
Ok(cases.len())
}
/// Idempotency lookup: the `snapshot_id` already recorded for `(repo,
/// git_sha)` in an artifact `*_snapshots` table, or `None`. Lets a capture
/// phase skip re-writing a projection that's already in the warehouse at this
/// SHA (iceberg-rust 0.9 ships no `expire_snapshots`/`rewrite_data_files`, so
/// not writing redundant blobs is one growth lever; the metadata-layer
/// [`super::maintenance`] `compact`/`expire` ops — `nornir warehouse compact`
/// / `expire` — are the other, reclaiming small files + aged snapshots after
/// the fact). Partition-pruned by `repo`, so it reads one repo's
/// small snapshot registry, never the blob bytes. A missing table (first
/// capture ever) reads as "not present".
async fn artifact_snapshot_id_for(
&self,
table_name: &str,
repo: &str,
git_sha: &str,
) -> Result<Option<String>> {
let table = match self.catalog.load_table(&self.table_ident(table_name)).await {
Ok(t) => t,
Err(_) => return Ok(None),
};
let batches = scan_repo_filtered(&table, Some(repo)).await?;
// artifact_snapshots layout: col 0 = snapshot_id, col 3 = git_sha.
let mut found = None;
for b in &batches {
let sid = downcast::<StringArray>(b, 0)?;
let sha = downcast::<StringArray>(b, 3)?;
for i in 0..b.num_rows() {
if sha.value(i) == git_sha {
found = Some(sid.value(i).to_string());
}
}
}
Ok(found)
}
/// `snapshot_id` of an existing Tantivy index capture at `(repo, git_sha)`.
pub async fn index_snapshot_id_for(&self, repo: &str, git_sha: &str) -> Result<Option<String>> {
self.artifact_snapshot_id_for(TABLE_TANTIVY_INDEX_SNAPSHOTS, repo, git_sha)
.await
}
/// `snapshot_id` of an existing DWARF capture at `(repo, git_sha)`.
pub async fn dwarf_snapshot_id_for(&self, repo: &str, git_sha: &str) -> Result<Option<String>> {
self.artifact_snapshot_id_for(TABLE_DWARF_SNAPSHOTS, repo, git_sha)
.await
}
/// Every table name in the warehouse namespace, sorted — the inventory the
/// generic viz browser lists. (Catalog metadata only, no data read.)
pub fn table_names(&self) -> Result<Vec<String>> {
self.rt.as_ref().expect("rt present").block_on(async {
let mut names: Vec<String> = self
.catalog
.list_tables(&self.namespace)
.await?
.into_iter()
.map(|t| t.name().to_string())
.collect();
names.sort();
Ok(names)
})
}
/// **Drop tables through the SERVED handle.** Returns the names that were
/// actually there and are now gone, sorted — a name that did not exist is
/// silently absent from the result rather than an error, because "already
/// not there" is the state the caller asked for.
///
/// This is the write half `Workspaces.Repopulate { purge }` needs and the one
/// `nornir warehouse reset` cannot give it: reset takes the warehouse ROOT and
/// opens `catalog.redb` itself, which a live server already holds exclusively
/// (redb is single-writer) — so reset refuses exactly when the server is up.
/// Going through the open handle is the same discipline every other server-side
/// write follows.
///
/// It drops what it is TOLD to drop and decides nothing: the derived-vs-raw
/// judgement lives in [`crate::lens_chain::derived_tables`], which reads the
/// lens registry, so a fourteenth lens needs no edit here.
pub fn drop_tables(&self, names: &[String]) -> Result<Vec<String>> {
self.rt.as_ref().expect("rt present").block_on(async {
let present: std::collections::HashSet<String> = self
.catalog
.list_tables(&self.namespace)
.await?
.into_iter()
.map(|t| t.name().to_string())
.collect();
let mut dropped = Vec::new();
for name in names {
if !present.contains(name) {
continue;
}
let ident = TableIdent::new(self.namespace.clone(), name.clone());
self.catalog.drop_table(&ident).await?;
dropped.push(name.clone());
}
dropped.sort();
Ok(dropped)
})
}
/// Scan `table` and return up to `limit` rows as display strings (column
/// names + stringified cells) — the generic "show any warehouse table in
/// egui" path. Blob/timestamp columns render as a `<type>` placeholder
/// rather than dumping bytes.
pub fn scan_preview(&self, table: &str, limit: usize) -> Result<TablePreview> {
self.scan_preview_bounded(table, limit, usize::MAX)
.map(|(preview, _)| preview)
}
/// [`Self::scan_preview`] with a **hard byte ceiling** on the decoded preview,
/// and an honest account of whether the table was cut.
///
/// The stringified `Vec<Vec<String>>` shape costs ~50–80 bytes per CELL once
/// `String` headers and allocator rounding are counted, so a "generous" row cap
/// alone is no bound at all: `5_000_000` rows of a 13-column table is ~4 GiB
/// RESIDENT, and thirteen such tables warmed back-to-back at populate end were
/// the measured 44 GB RSS plateau of the fat `dwarves-server` on oden
/// (2026-08-21, 1 s RSS sampler vs the monitor journal). The decode therefore
/// stops at whichever comes first — `limit` rows or `max_bytes` of estimated
/// resident size — and REPORTS the cut so the caller can say so out loud
/// instead of serving a silently partial table. `max_bytes == usize::MAX`
/// disables the byte ceiling (the plain [`Self::scan_preview`] behaviour).
///
/// The estimate is [`preview_resident_bytes`]-exact for what is pushed (String
/// header + heap capacity per cell, `Vec` header per row); it does not model
/// allocator slack, so the real RSS is somewhat ABOVE the figure — a ceiling
/// that under-counts is still a ceiling, but pick the budget with that in mind.
///
/// ## The ceiling bounds the ARROW half too, and until 2026-09-06 it did not
///
/// This decode used to open with `let batches = scan_limited(&t, limit).await?`
/// — `skade::read_limited`, which returns a fully materialised `Vec<RecordBatch>`
/// of up to `limit` rows. `limit` on the served path is `HOT_PREVIEW_LIMIT` /
/// `SCAN_CAP` = **5,000,000**, so the ceiling was applied to the smaller of two
/// simultaneously-live copies while the bigger one — full-width Arrow for five
/// million rows — was ALREADY resident and charged to nothing. The module note
/// on the (since removed) working-set cache claimed the opposite in as many words
/// ("This bounds the TRANSIENT as well as the resident size"); a row cap is no
/// byte bound, which is the very lesson the rest of that file is about.
/// MEASURED by the guard `the_preview_ceiling_bounds_the_arrow_transient_not_only_the_strings`:
/// **400 Arrow rows pulled to keep 39.**
///
/// So the stream is now driven here, exactly as the sibling
/// [`Self::scan_arrow_bounded`] drives it: each batch is stringified as it
/// arrives and dropped, and the first batch whose rows cross `max_bytes` (or
/// `limit`) ends the loop — dropping the stream, which cancels the scan so no
/// further parquet file is opened. The Arrow transient is therefore ONE batch,
/// not the whole cap. `skade::read_limited` cannot carry the ceiling because it
/// hands back the finished `Vec`; the bound has to live where the rows are
/// consumed. [`PREVIEW_ARROW_ROWS_DECODED`] makes the transient observable.
pub fn scan_preview_bounded(
&self,
table: &str,
limit: usize,
max_bytes: usize,
) -> Result<(TablePreview, PreviewCut)> {
use futures::StreamExt;
let result: Result<(TablePreview, PreviewCut)> =
self.rt.as_ref().expect("rt present").block_on(async {
let t = self.catalog.load_table(&self.table_ident(table)).await?;
// `select_all` preserves column order (positional `downcast` stays
// valid); the batch size is the sibling's, so "one batch beyond the
// ceiling" means the same thing on both paths.
let mut stream = t
.scan()
.select_all()
.with_batch_size(Some(PREVIEW_BATCH_ROWS))
.build()?
.to_arrow()
.await?;
let mut columns: Vec<String> = Vec::new();
let mut rows: Vec<Vec<String>> = Vec::new();
let mut bytes: usize = 0;
let mut cut = PreviewCut::None;
'outer: while let Some(b) = stream.next().await {
let b = b?;
PREVIEW_ARROW_ROWS_DECODED
.fetch_add(b.num_rows(), std::sync::atomic::Ordering::Relaxed);
if columns.is_empty() {
columns = b
.schema()
.fields()
.iter()
.map(|f| f.name().to_string())
.collect();
bytes += columns.iter().map(string_resident_bytes).sum::<usize>();
}
for i in 0..b.num_rows() {
if rows.len() >= limit {
cut = PreviewCut::Rows { limit };
break 'outer;
}
let row: Vec<String> = (0..b.num_columns())
.map(|c| cell_to_string(b.column(c).as_ref(), i))
.collect();
let row_bytes = std::mem::size_of::<Vec<String>>()
+ row.iter().map(string_resident_bytes).sum::<usize>();
if bytes + row_bytes > max_bytes {
cut = PreviewCut::Bytes { max_bytes };
break 'outer;
}
bytes += row_bytes;
rows.push(row);
}
// The batch just consumed is dropped HERE, before the next one
// is pulled — that is the whole point of the loop.
}
// Dropping the stream cancels the rest of the scan (no further
// parquet file is opened) — the same cancellation `scan_arrow_bounded`
// relies on.
drop(stream);
// A zero out of the READER must never be handed on as a zero out
// of the WAREHOUSE. See `refuse_silent_empty`.
refuse_silent_empty(
table,
rows.len(),
!matches!(cut, PreviewCut::None),
declared_records(&t),
)?;
Ok((TablePreview { columns, rows }, cut))
});
match result {
Ok(p) => Ok(p),
// A blank/torn metadata JSON (`EOF at line 1 col 0`) is "no readable
// snapshot" — degrade to an EMPTY preview instead of aborting the whole
// scan with an Internal error. (See `is_blank_metadata_json_error`.)
Err(e) if is_blank_metadata_json_error(&e) => {
eprintln!(
"WARNING: nornir: table `{table}` has blank/torn Iceberg metadata JSON \
({e:#}); reading it as an empty preview (0 rows)."
);
Ok((TablePreview::default(), PreviewCut::None))
}
Err(e) => Err(e),
}
}
/// Whole-table Arrow read with a **hard byte ceiling**: batches are pulled
/// off the scan stream and kept while their summed `get_array_memory_size`
/// stays within `max_bytes`; the batch that would cross it — and everything
/// behind it — is dropped (dropping the stream cancels the rest of the scan:
/// no further parquet files are opened). Returns the batches and whether the
/// table was cut. `max_bytes == usize::MAX` is a plain full read.
///
/// Why: a full Arrow decode of this workspace's `symbol_facts` (98 M rows of
/// append-only history) measured **29.11 GiB** on oden 2026-08-21 — pulled in
/// by the populate-end warm for a reader that did not exist. A decode with no
/// ceiling is the OOM-sized transient, whatever the cache later retains.
pub fn scan_arrow_bounded(
&self,
table: &str,
max_bytes: usize,
) -> Result<(Vec<RecordBatch>, bool)> {
use futures::StreamExt;
let result: Result<(Vec<RecordBatch>, bool)> =
self.rt.as_ref().expect("rt present").block_on(async {
let t = self.catalog.load_table(&self.table_ident(table)).await?;
let mut stream = t
.scan()
.select_all()
.with_batch_size(Some(8192))
.build()?
.to_arrow()
.await?;
let mut batches: Vec<RecordBatch> = Vec::new();
let mut bytes = 0usize;
let mut cut = false;
while let Some(b) = stream.next().await {
let b = b?;
let nb = b.get_array_memory_size();
if bytes + nb > max_bytes {
cut = true;
break;
}
bytes += nb;
batches.push(b);
}
// Same gate as the preview path — one predicate, two callers.
refuse_silent_empty(
table,
batches.iter().map(|b| b.num_rows()).sum(),
cut,
declared_records(&t),
)?;
Ok((batches, cut))
});
match result {
Ok(r) => Ok(r),
Err(e) if is_blank_metadata_json_error(&e) => {
eprintln!(
"WARNING: nornir: table `{table}` has blank/torn Iceberg metadata JSON \
({e:#}); reading it as empty (0 rows)."
);
Ok((Vec::new(), false))
}
Err(e) => Err(e),
}
}
/// Bytes of data files behind `table`'s CURRENT snapshot, straight off the
/// Iceberg snapshot summary (`total-files-size`) — metadata only, no data
/// scan. `None` when the table has no snapshot, the summary lacks the key, or
/// the table is unreadable. Parquet is compressed, so this is a LOWER bound on
/// what a full Arrow decode of the table will occupy in RAM — which makes it a
/// cheap, honest pre-check for "would decoding this table ever fit the cache".
pub fn table_disk_bytes(&self, table: &str) -> Option<u64> {
self.rt.as_ref().expect("rt present").block_on(async {
let t = self.catalog.load_table(&self.table_ident(table)).await.ok()?;
let md = t.metadata();
let snap = md.current_snapshot()?;
snap.summary()
.additional_properties
.get("total-files-size")?
.parse::<u64>()
.ok()
})
}
/// Per-table populate stats for EVERY table in the namespace — the numbers the
/// Mímir "Warehouse Health" backoffice page shows (row count · byte size ·
/// last-populated snapshot) and the populate-completeness oracle reads. CHEAP:
/// read straight off each table's current Iceberg **snapshot summary**
/// (`total-records` / `total-files-size` / `total-data-files`) — NO data scan.
/// A table with no snapshot yet reports `row_count = -1` (unknown/empty) and a
/// zero `last_snapshot_ms`. A blank/torn metadata JSON degrades to an
/// unknown-stat row rather than aborting the whole sweep.
pub fn table_stats(&self) -> Result<Vec<TableStat>> {
let names = self.table_names()?;
self.rt.as_ref().expect("rt present").block_on(async {
let mut out = Vec::with_capacity(names.len());
for name in names {
let stat = match self.catalog.load_table(&self.table_ident(&name)).await {
Ok(t) => {
let md = t.metadata();
match md.current_snapshot() {
Some(snap) => {
let props = &snap.summary().additional_properties;
let num = |k: &str| -> i64 {
props
.get(k)
.and_then(|v| v.parse::<i64>().ok())
.unwrap_or(-1)
};
TableStat {
table: name.clone(),
row_count: num("total-records"),
byte_size: num("total-files-size"),
data_files: num("total-data-files").max(0),
last_snapshot_ms: snap.timestamp_ms(),
}
}
// Table exists but has never been written to.
None => TableStat {
table: name.clone(),
row_count: 0,
byte_size: 0,
data_files: 0,
last_snapshot_ms: 0,
},
}
}
Err(_) => TableStat {
table: name.clone(),
row_count: -1,
byte_size: -1,
data_files: 0,
last_snapshot_ms: 0,
},
};
out.push(stat);
}
Ok(out)
})
}
}
/// One warehouse table's populate stats — row count, on-disk byte size, and the
/// timestamp of its last-populated Iceberg snapshot. Read cheaply from the
/// snapshot summary by [`IcebergWarehouse::table_stats`]. `row_count`/`byte_size`
/// are `-1` when unknown (no snapshot / unreadable metadata); a table that exists
/// but was never written reports `0`. `is_empty()` treats both as empty.
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct TableStat {
pub table: String,
pub row_count: i64,
pub byte_size: i64,
pub data_files: i64,
pub last_snapshot_ms: i64,
}
impl TableStat {
/// A table is empty when it has no rows AND no data files. `row_count == -1`
/// (unknown) with zero data files also reads empty — a never-populated table.
pub fn is_empty(&self) -> bool {
self.row_count <= 0 && self.data_files <= 0
}
}
/// **Does `table` hold any Parquet data file under `root`? — WITHOUT the catalog.**
///
/// Answers the one question a caller may need to ask about a warehouse it does
/// NOT hold open: "was this wiped?". Every catalog-based answer to that question
/// is unavailable exactly when it matters:
///
/// * [`IcebergWarehouse::open`] takes redb's single-writer lock, which the live
/// server already holds — and it CREATES the warehouse it was asked to inspect.
/// * [`IcebergWarehouse::open_read_only`] copies `catalog.redb` aside through
/// [`copy_catalog_consistent`], which needs a size-stable window. MEASURED on
/// oden 2026-09-10: on a warehouse being written by the boot feed + lens chains
/// it did not get one, so three consecutive monitor ticks over a provably empty
/// warehouse got an `Err` instead of an answer — and a probe that cannot answer
/// under load is a probe that never fires.
///
/// So this reads the DIRECTORY. Iceberg's own layout is
/// `<root>/warehouse/<namespace>/<table>/{metadata,data}/…`; a table that has
/// never been written has a `metadata/` and no Parquet anywhere beneath it, which
/// is precisely the post-wipe state (the server re-creates every table as an
/// empty schema the moment it boots). No lock, no copy, no runtime — one walk
/// that stops at the first `.parquet`.
///
/// The namespace is not named by the caller: every namespace directly under
/// `warehouse/` is searched, so this cannot go quietly false by disagreeing with
/// [`IcebergWarehouse::namespace`] about which one is in use.
///
/// It answers the presence of DATA, never its correctness — an interrupted
/// populate can leave a table with fewer files than a finished one, and this says
/// "has data" for both.
pub fn table_has_data_files(root: &Path, table: &str) -> bool {
let Ok(namespaces) = std::fs::read_dir(root.join("warehouse")) else {
return false;
};
for ns in namespaces.flatten() {
let dir = ns.path().join(table);
if dir.is_dir() && dir_holds_a_parquet(&dir) {
return true;
}
}
false
}
/// Depth-first walk under one table directory, stopping at the first `.parquet`.
/// Iterative (an explicit stack) so a deeply partitioned table cannot recurse the
/// monitor thread's stack away, and silent about unreadable entries: this is a
/// "did I see one" question, and an unreadable subtree is simply not a sighting.
fn dir_holds_a_parquet(dir: &Path) -> bool {
let mut stack = vec![dir.to_path_buf()];
while let Some(d) = stack.pop() {
let Ok(entries) = std::fs::read_dir(&d) else {
continue;
};
for e in entries.flatten() {
let p = e.path();
match e.file_type() {
Ok(t) if t.is_dir() => stack.push(p),
Ok(_) if p.extension().is_some_and(|x| x == "parquet") => return true,
_ => {}
}
}
}
false
}
/// A stringified preview of one warehouse table for the egui browser. The POD
/// now lives in the `nornir-warehouse-trait` edda leaf (so a consumer can use it
/// without depending on nornir); re-exported here at its old path
/// (`crate::warehouse::iceberg::TablePreview`) so call sites are untouched.
pub use nornir_warehouse_trait::TablePreview;
/// Stringify one arrow cell for display. Covers the scalar types the warehouse
/// schemas use; anything else (timestamps, blobs, lists) renders as a
/// `<DataType>` placeholder so the browser never dumps raw bytes.
/// How (or whether) a bounded preview decode was cut short — see
/// [`IcebergWarehouse::scan_preview_bounded`]. Carried out to the caller so the
/// cut is NEVER silent: the working-set cache logs it with the table name and the
/// knob that raises the ceiling.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PreviewCut {
/// Every row the scan produced was kept.
None,
/// The row cap was reached with more rows behind it.
Rows { limit: usize },
/// The byte ceiling was reached; the row that would have crossed it (and
/// everything after) was dropped.
Bytes { max_bytes: usize },
}
impl PreviewCut {
pub fn is_cut(self) -> bool {
!matches!(self, PreviewCut::None)
}
}
/// Rows per Arrow batch on the bounded preview path — the same figure
/// [`IcebergWarehouse::scan_arrow_bounded`] streams at, so "at most one batch
/// beyond the ceiling" is the same overshoot on both readers.
pub(crate) const PREVIEW_BATCH_ROWS: usize = 8192;
/// **Arrow rows pulled off the scan stream by [`IcebergWarehouse::scan_preview_bounded`]
/// since process start.** The counter that makes the preview ceiling's TRANSIENT
/// half observable.
///
/// The byte ceiling can only be believed if the Arrow decode that feeds it is
/// bounded too: the stringified rows the cache retains are the SMALLER of the two
/// live copies during a decode. This counter is what a guard asserts on — rows
/// KEPT is already visible in the `TablePreview`, rows DECODED was not, and the
/// gap between them was the whole defect (see the method's doc).
pub static PREVIEW_ARROW_ROWS_DECODED: std::sync::atomic::AtomicUsize =
std::sync::atomic::AtomicUsize::new(0);
/// Read [`PREVIEW_ARROW_ROWS_DECODED`] — take a delta across a call to know what
/// ONE decode pulled.
pub fn preview_arrow_rows_decoded() -> usize {
PREVIEW_ARROW_ROWS_DECODED.load(std::sync::atomic::Ordering::Relaxed)
}
/// Resident bytes of one `String`: the 24-byte header plus its heap capacity.
/// (Allocator rounding is NOT modelled — see [`IcebergWarehouse::scan_preview_bounded`].)
pub fn string_resident_bytes(s: &String) -> usize {
std::mem::size_of::<String>() + s.capacity()
}
/// Resident bytes of a whole decoded [`TablePreview`] — header + heap for every
/// column name and cell, plus the `Vec` header of every row. The figure the
/// working-set byte budget is charged with.
pub fn preview_resident_bytes(p: &TablePreview) -> usize {
std::mem::size_of::<TablePreview>()
+ p.columns.iter().map(string_resident_bytes).sum::<usize>()
+ p.rows
.iter()
.map(|r| {
std::mem::size_of::<Vec<String>>() + r.iter().map(string_resident_bytes).sum::<usize>()
})
.sum::<usize>()
}
fn cell_to_string(arr: &dyn Array, row: usize) -> String {
if arr.is_null(row) {
return String::new();
}
if let Some(a) = arr.as_any().downcast_ref::<StringArray>() {
return a.value(row).to_string();
}
if let Some(a) = arr.as_any().downcast_ref::<Int64Array>() {
return a.value(row).to_string();
}
if let Some(a) = arr.as_any().downcast_ref::<Int32Array>() {
return a.value(row).to_string();
}
if let Some(a) = arr.as_any().downcast_ref::<Float64Array>() {
return format!("{:.4}", a.value(row));
}
if let Some(a) = arr.as_any().downcast_ref::<BooleanArray>() {
return a.value(row).to_string();
}
// TIMESTAMPS. Without this arm every `ts_micros` cell rendered as the literal
// type name `<Timestamp(Microsecond, Some("+00:00"))>`, and that is not a
// cosmetic defect: `mimir_lens::rows::ts_micros` parses this string, got 0 for
// EVERY row, and so `latest_snapshot_rows_of`'s `(ts, snapshot_id)` ordering
// collapsed onto the tie-break — the snapshot UUID. "Latest snapshot per repo"
// silently meant "lexicographically greatest random UUID per repo", for every
// table in `SNAPSHOT_TABLES` (`symbol_facts`, `call_edges`, `feature_gate_facts`,
// `git_heat_facts`, `test_inventory`, `scip_occurrences`).
//
// Measured on the live oden warehouse 2026-07-31: after re-ingesting all 14
// repos with the document-scoping fix, `Mimir.SymbolIdentity` still reported 4
// of them (funnel-core, ordning, skade, znippy-zoomies) as carrying pre-fix
// rows. Their newest snapshot was clean; they were exactly the 4 whose OLDEST
// snapshot happened to have the largest UUID. A stale-but-plausible answer, with
// no error anywhere — the FALSE-GREEN class inverted into a false RED, inside
// the reader the anti-false-green gate itself is built on.
//
// RFC 3339 (not epoch micros) so the string is both machine-ordered and the same
// shape `warehouse query` prints, and so it stays sortable as text.
if let Some(a) = arr.as_any().downcast_ref::<TimestampMicrosecondArray>() {
if let Some(dt) = a.value_as_datetime(row) {
return DateTime::<Utc>::from_naive_utc_and_offset(dt, Utc)
.to_rfc3339_opts(chrono::SecondsFormat::Micros, true);
}
}
// NON-SCALAR COLUMNS. Anything with no arm above cannot be a preview cell, and
// this catch-all is the mechanism by which `embeddings.vector` — a 768-float
// `list<float32>` — never reached a thin client: every row of that column rendered
// as the SAME constant Arrow type name, so the dupes SEMANTIC tier and leafwatch's
// semantic leg were permanently unavailable over gRPC (hash tiers only, blind to
// "the same thing, written differently").
//
// The fix is NOT to widen this string — a vector is bulk data and belongs on the
// typed `Warehouse::scan_vectors` / `Warehouse.ScanVectors` route. What IS fixed
// here is the placeholder: it now names the type in Arrow's own compact syntax, its
// element count, and where the real values live — so a reader (human or agent) is
// told the column has a route instead of being handed `<List(Field { name: "item",
// data_type: Float32, nullable: true, dict_id: 0, … })>` and left to guess.
//
// It stays deliberately UNPARSEABLE as a number run, so no consumer can ever
// mistake a placeholder for data.
if let Some(a) = arr.as_any().downcast_ref::<arrow::array::ListArray>() {
let elem = a.value(row);
return format!(
"<list<{}>[{}] — typed values via Warehouse.ScanVectors>",
elem.data_type(),
elem.len()
);
}
if let Some(a) = arr
.as_any()
.downcast_ref::<arrow::array::FixedSizeListArray>()
{
let elem = a.value(row);
return format!(
"<fixed_size_list<{}>[{}] — typed values via Warehouse.ScanVectors>",
elem.data_type(),
elem.len()
);
}
format!("<{}>", arr.data_type())
}
#[async_trait::async_trait]
impl Warehouse for IcebergWarehouse {
// ── generic Arrow core ──────────────────────────────────────────────────
// Each delegates to the existing engine bodies (which already wrap skade);
// ZERO behavior change — this is the P1 trait seam, not a flip.
fn append_arrow(&self, table: &str, batch: RecordBatch) -> Result<()> {
self.ensure_writable()?;
self.block_on(async {
let ident = self.table_ident(table);
let t = self.catalog.load_table(&ident).await?;
append_batch(&self.catalog, t, batch).await
})
}
fn scan_arrow(&self, table: &str) -> Result<Vec<RecordBatch>> {
// Guard at load AND read: a torn metadata.json fails inside `load_table`.
self.block_on(load_and_read_all(self, table))
}
fn scan_filtered(
&self,
table: &str,
filter: &ScanFilter,
columns: &[&str],
) -> Result<Vec<RecordBatch>> {
self.block_on(async {
let t = match self.catalog.load_table(&self.table_ident(table)).await {
Ok(t) => t,
// A not-yet-created table reads as empty (matches the inherent
// readers that treat a missing table as "no rows").
Err(_) => return Ok(Vec::new()),
};
// A blank/torn metadata JSON degrades to 0 rows (see `empty_on_blank_metadata`).
empty_on_blank_metadata(
table,
skade::read_filtered(&t, filter, columns)
.await
.map_err(Into::into),
)
})
}
fn scan_limited(&self, table: &str, max_rows: usize) -> Result<Vec<RecordBatch>> {
self.block_on(async {
// Guard at load AND read: a torn metadata.json fails inside `load_table`.
empty_on_blank_metadata(
table,
async {
let t = self.catalog.load_table(&self.table_ident(table)).await?;
scan_limited(&t, max_rows).await
}
.await,
)
})
}
// ── table / catalog lifecycle ───────────────────────────────────────────
fn ensure_table(
&self,
table: &str,
schema: iceberg::spec::Schema,
partition_cols: &[&str],
) -> Result<()> {
self.ensure_writable()?;
self.block_on(create_partitioned_table_if_missing(
&self.catalog,
&self.namespace,
table,
schema,
partition_cols,
))
}
fn ensure_columns(&self, table: &str, canonical: &iceberg::spec::Schema) -> Result<()> {
self.ensure_writable()?;
self.block_on(async {
let ident = self.table_ident(table);
let t = self.catalog.load_table(&ident).await?;
ensure_table_schema(&self.catalog, &ident, t, canonical).await?;
Ok(())
})
}
fn table_names(&self) -> Result<Vec<String>> {
// The inherent method already does the sorted catalog listing under
// `block_on`; the trait method mirrors it exactly.
IcebergWarehouse::table_names(self)
}
fn scan_preview(&self, table: &str, limit: usize) -> Result<TablePreview> {
// Delegates to the inherent `scan_preview` (inherent methods win method
// resolution over trait methods, so this is NOT a recursion) — ZERO
// behavior change; the browser reads exactly as before.
IcebergWarehouse::scan_preview(self, table, limit)
}
/// TYPED vector read — the route `scan_preview` cannot be (see
/// [`cell_to_string`]'s non-scalar catch-all and the trait doc).
///
/// A table that does not exist yet is `Ok` with zero vectors: that is the honest
/// "the warehouse holds no embeddings" state (the embedder never ran), and it is
/// deliberately DIFFERENT from the error a transport that cannot carry vectors
/// returns. A missing *column* on an existing table is still an error, because that
/// is a schema mismatch and not an empty warehouse.
fn scan_vectors(
&self,
table: &str,
key_col: &str,
vec_col: &str,
limit: usize,
) -> Result<nornir_warehouse_trait::VectorScan> {
let table = if table.is_empty() {
TABLE_EMBEDDINGS
} else {
table
};
let key_col = if key_col.is_empty() {
"content_hash"
} else {
key_col
};
let vec_col = if vec_col.is_empty() {
"vector"
} else {
vec_col
};
if !self.table_names()?.iter().any(|t| t == table) {
return Ok(nornir_warehouse_trait::VectorScan::default());
}
// A biting limit uses the early-break scan so a constellation-scale table is
// not materialized to hand back 1000 rows; limit 0 means the WHOLE table,
// because the semantic tier is all-pairs and a silently short corpus would give
// a thin client a different verdict from a fat one.
let batches = if limit > 0 {
Warehouse::scan_limited(self, table, limit)?
} else {
Warehouse::scan_arrow(self, table)?
};
nornir_warehouse_trait::vectors_from_batches(
&batches,
key_col,
vec_col,
"model_profile",
limit,
)
.with_context(|| format!("scan_vectors({table}.{vec_col})"))
}
fn describe_columns(&self, table: &str) -> Result<Vec<crate::warehouse::sql::ColumnInfo>> {
use crate::warehouse::sql::ColumnInfo;
let known = self.table_names()?;
if !known.iter().any(|t| t == table) {
anyhow::bail!("no warehouse table named {table:?} — known tables: {}", {
let mut k = known;
k.sort();
k.join(", ")
});
}
let catalog = self.catalog().clone();
let ident = self.table_ident(table);
let name = table.to_string();
self.block_on(async move {
let t = catalog
.load_table(&ident)
.await
.with_context(|| format!("load table {name}"))?;
let schema = schema_to_arrow_schema(t.metadata().current_schema())
.with_context(|| format!("arrow schema for {name}"))?;
Ok(schema
.fields()
.iter()
.map(|f| ColumnInfo {
name: f.name().clone(),
data_type: f.data_type().to_string(),
nullable: f.is_nullable(),
})
.collect())
})
}
// ── named per-table convenience wrappers (thin) ─────────────────────────
fn append_bench_run(&self, repo: &str, run: &BenchRun) -> Result<Uuid> {
self.rt
.as_ref()
.expect("rt present")
.block_on(self.append_bench_run_async(repo, run))
}
fn query_bench_runs(&self, filter: &BenchFilter) -> Result<Vec<BenchRun>> {
self.rt
.as_ref()
.expect("rt present")
.block_on(self.query_bench_runs_async(filter))
}
// ── async surface ────────────────────────────────────────────────────────
// Each delegates to the existing inherent async body (fully-qualified so the
// INHERENT method is selected, never this same-named trait method → no
// recursion). These do real async work — no `block_on` — so they compose from
// inside an existing runtime.
async fn append_arrow_async(&self, table: &str, batch: RecordBatch) -> Result<()> {
self.ensure_writable()?;
let ident = self.table_ident(table);
let t = self.catalog.load_table(&ident).await?;
append_batch(&self.catalog, t, batch).await
}
async fn scan_arrow_async(&self, table: &str) -> Result<Vec<RecordBatch>> {
load_and_read_all(self, table).await
}
async fn append_bench_run_async(&self, repo: &str, run: &BenchRun) -> Result<Uuid> {
IcebergWarehouse::append_bench_run_async(self, repo, run).await
}
async fn query_bench_runs_async(&self, filter: &BenchFilter) -> Result<Vec<BenchRun>> {
IcebergWarehouse::query_bench_runs_async(self, filter).await
}
async fn append_git_heat_scan_async(
&self,
scan: &crate::knowledge::git_heat::GitHeatScan,
) -> Result<()> {
IcebergWarehouse::append_git_heat_scan_async(self, scan).await
}
fn as_any(&self) -> &dyn std::any::Any {
self
}
}
impl IcebergWarehouse {
/// Async form of [`Warehouse::query_bench_runs`]. Call this from code
/// already on the warehouse runtime (e.g. inside `run_pipeline`, itself
/// driven by `wh.block_on(...)`): the sync trait method does its own
/// `block_on` and would panic if nested.
pub async fn query_bench_runs_async(&self, filter: &BenchFilter) -> Result<Vec<BenchRun>> {
Ok(self.query_bench_runs_audited_async(filter).await?.0)
}
/// [`query_bench_runs_async`] plus the count of FULL `bench_results` +
/// `test_outcomes` rows decoded — the read-amplification the join pays.
///
/// The old path `scan_all`'d both child tables (every repo's runs) and
/// hash-joined in RAM, discarding rows whose `run_id` was not among the
/// `repo`/`machine`-matched runs. This path instead pushes
/// `run_id IN (matched run ids)` into the child scans, so the reader's
/// `RowFilter` skips other runs' rows at decode time. Byte-identical output
/// (the discarded rows never contributed), fewer rows materialized. The
/// child tables carry no `repo` column and are unsorted on `run_id`, so this
/// is a row-group/`RowFilter` win, not a partition prune (`run_id`-min/max
/// bounds span everything) — measured via the returned row count, not
/// `plan_stats`.
pub(crate) async fn query_bench_runs_audited_async(
&self,
filter: &BenchFilter,
) -> Result<(Vec<BenchRun>, usize)> {
async {
let mut by_id: std::collections::HashMap<Uuid, BenchRun> =
std::collections::HashMap::new();
// The raw stored `run_id` strings of the runs that survive the
// repo/machine filter — pushed verbatim into the child scans so the
// pushdown literals byte-match the persisted values (no UUID
// re-format round-trip).
let mut wanted_run_ids: Vec<String> = Vec::new();
// bench_runs
let table = self
.catalog
.load_table(&self.table_ident(TABLE_BENCH_RUNS))
.await?;
for batch in scan_repo_filtered(&table, filter.repo.as_deref()).await? {
let ids = downcast::<StringArray>(&batch, 0)?;
let repos = downcast::<StringArray>(&batch, 1)?;
let ts = downcast::<TimestampMicrosecondArray>(&batch, 2)?;
let dates = downcast::<StringArray>(&batch, 3)?;
let versions = downcast::<StringArray>(&batch, 4)?;
let machines = downcast::<StringArray>(&batch, 5)?;
let cores = downcast::<Int32Array>(&batch, 6)?;
for i in 0..batch.num_rows() {
if let Some(want) = &filter.repo {
if repos.value(i) != want {
continue;
}
}
if let Some(want) = &filter.machine {
if machines.value(i) != want {
continue;
}
}
let uid = Uuid::parse_str(ids.value(i))?;
let ts_dt = Utc
.timestamp_micros(ts.value(i))
.single()
.context("invalid micro timestamp")?;
wanted_run_ids.push(ids.value(i).to_string());
by_id.insert(
uid,
BenchRun {
// The row's OWN identity, read back from the table's `id`
// column (`uid`), not minted here. Added 2026-09-20 with
// `BenchRun::id`; this reader was the one initializer the
// field's introduction missed, so the lib did not compile
// at all with `--no-default-features --features
// vector,embed-tract,embed-ort`.
id: uid,
// `bench_runs` carries no machine-facts column; a row read
// back from Iceberg genuinely does not know them, and
// inventing an empty `Machine` would claim it did.
machine_facts: None,
date: dates.value(i).to_string(),
timestamp: Some(ts_dt.to_rfc3339()),
version: versions.value(i).to_string(),
machine: machines.value(i).to_string(),
cores: cores.value(i).max(0) as u32,
results: Vec::new(),
tests: Vec::new(),
},
);
}
}
let mut data_rows = 0usize;
// bench_results — `run_id IN (matched)` pushdown (RowFilter skip).
let table = self
.catalog
.load_table(&self.table_ident(TABLE_BENCH_RESULTS))
.await?;
for batch in read_runs_filtered(&table, &wanted_run_ids).await? {
data_rows += batch.num_rows();
let ids = downcast::<StringArray>(&batch, 0)?;
let names = downcast::<StringArray>(&batch, 1)?;
let metrics = downcast::<StringArray>(&batch, 2)?;
let values = downcast::<Float64Array>(&batch, 3)?;
for i in 0..batch.num_rows() {
let uid = Uuid::parse_str(ids.value(i))?;
if let Some(run) = by_id.get_mut(&uid) {
let name = names.value(i).to_string();
let entry = run.results.iter_mut().find(|r| r.name == name);
let target = if let Some(e) = entry {
e
} else {
run.results.push(BenchResult {
name,
metrics: Default::default(),
..Default::default()
});
run.results.last_mut().unwrap()
};
target.metrics.insert(
metrics.value(i).to_string(),
serde_json::Value::from(values.value(i)),
);
}
}
}
// test_outcomes — same `run_id IN (matched)` pushdown.
let table = self
.catalog
.load_table(&self.table_ident(TABLE_TEST_OUTCOMES))
.await?;
for batch in read_runs_filtered(&table, &wanted_run_ids).await? {
data_rows += batch.num_rows();
let ids = downcast::<StringArray>(&batch, 0)?;
let names = downcast::<StringArray>(&batch, 1)?;
let passed = downcast::<BooleanArray>(&batch, 2)?;
let durations = downcast::<Float64Array>(&batch, 3)?;
let messages = downcast::<StringArray>(&batch, 4)?;
for i in 0..batch.num_rows() {
let uid = Uuid::parse_str(ids.value(i))?;
if let Some(run) = by_id.get_mut(&uid) {
run.tests.push(TestOutcome {
name: names.value(i).to_string(),
passed: passed.value(i),
duration_ms: if durations.is_null(i) {
None
} else {
Some(durations.value(i))
},
message: if messages.is_null(i) {
None
} else {
Some(messages.value(i).to_string())
},
});
}
}
}
let mut out: Vec<BenchRun> = by_id.into_values().collect();
out.sort_by(|a, b| a.timestamp.cmp(&b.timestamp));
if let Some(n) = filter.limit {
let drop_n = out.len().saturating_sub(n);
out.drain(..drop_n);
}
anyhow::Ok((out, data_rows))
}
.await
}
}
impl IcebergWarehouse {
/// Read `bench_telemetry` rows, optionally pruned to one `repo`. One row per
/// `(run_id, bench)` — the CPU/mem footprint the in-bench sampler captured.
/// Sync wrapper; see [`IcebergWarehouse::query_bench_telemetry_async`].
pub fn query_bench_telemetry(&self, repo: Option<&str>) -> Result<Vec<BenchTelemetryRow>> {
self.rt
.as_ref()
.expect("rt present")
.block_on(self.query_bench_telemetry_async(repo))
}
/// Async form of [`IcebergWarehouse::query_bench_telemetry`].
pub async fn query_bench_telemetry_async(
&self,
repo: Option<&str>,
) -> Result<Vec<BenchTelemetryRow>> {
let table = self
.catalog
.load_table(&self.table_ident(TABLE_BENCH_TELEMETRY))
.await?;
let mut out = Vec::new();
for batch in scan_repo_filtered(&table, repo).await? {
let run_ids = downcast::<StringArray>(&batch, 0)?;
let repos = downcast::<StringArray>(&batch, 1)?;
let benches = downcast::<StringArray>(&batch, 2)?;
let n_cores = downcast::<Int32Array>(&batch, 3)?;
let cpu_avg = downcast::<Float64Array>(&batch, 4)?;
let cpu_max = downcast::<Float64Array>(&batch, 5)?;
let busy_avg = downcast::<Float64Array>(&batch, 6)?;
let busy_max = downcast::<Int32Array>(&batch, 7)?;
let mem_peak = downcast::<Float64Array>(&batch, 8)?;
let mem_pct = downcast::<Float64Array>(&batch, 9)?;
let elapsed = downcast::<Float64Array>(&batch, 10)?;
// Fields 12-15 (process-scoped) are read BY NAME and tolerate
// absence: a table that predates them has 11 columns, and a data
// file written before the evolution has NULLs in them. Both must
// read back as `None` — "not measured" — never as `0.0`.
let opt_f64 = |name: &str| -> Option<&Float64Array> {
batch
.column_by_name(name)
.and_then(|c| c.as_any().downcast_ref::<Float64Array>())
};
let proc_cpu = opt_f64("proc_cpu_ms");
let child_cpu = opt_f64("child_cpu_ms");
let proc_avg = opt_f64("proc_cores_avg");
let proc_max = opt_f64("proc_cores_max");
let at = |a: Option<&Float64Array>, i: usize| -> Option<f64> {
a.filter(|c| !c.is_null(i)).map(|c| c.value(i))
};
for i in 0..batch.num_rows() {
if let Some(want) = repo {
if repos.value(i) != want {
continue;
}
}
out.push(BenchTelemetryRow {
run_id: run_ids.value(i).to_string(),
repo: repos.value(i).to_string(),
bench: benches.value(i).to_string(),
n_cores: n_cores.value(i).max(0) as u32,
cpu_pct_avg: cpu_avg.value(i),
cpu_pct_max: cpu_max.value(i),
cores_busy_avg: busy_avg.value(i),
cores_busy_max: busy_max.value(i).max(0) as u32,
mem_peak_mb: mem_peak.value(i),
mem_pct_max: mem_pct.value(i),
elapsed_ms: elapsed.value(i),
proc_cpu_ms: at(proc_cpu, i),
child_cpu_ms: at(child_cpu, i),
proc_cores_avg: at(proc_avg, i),
proc_cores_max: at(proc_max, i),
});
}
}
Ok(out)
}
}
/// One persisted `bench_telemetry` row — the CPU/mem footprint of a single
/// bench unit. Mirrors [`crate::bench::telemetry::Telemetry`] plus the keys
/// (`run_id`, `repo`, `bench`).
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct BenchTelemetryRow {
pub run_id: String,
pub repo: String,
pub bench: String,
pub n_cores: u32,
/// **HOST-WIDE** (`/proc/stat`) — the whole box, every process on it.
/// Ambient context; never a statement about this bench. See
/// [`crate::warehouse::iceberg_schema::bench_telemetry`].
pub cpu_pct_avg: f64,
pub cpu_pct_max: f64,
/// **HOST-WIDE.** Cores >50% busy *for any reason*. A strictly serial arm
/// on a loaded 32-core box reads 32 here. Judge saturation on
/// [`Self::proc_cores_max`] instead.
pub cores_busy_avg: f64,
pub cores_busy_max: u32,
pub mem_peak_mb: f64,
pub mem_pct_max: f64,
pub elapsed_ms: f64,
/// **PROCESS-SCOPED** (`/proc/self/stat`): CPU ms of this process's own
/// threads. `None` on a row written before 2026-08-03 — not measured.
#[serde(default)]
pub proc_cpu_ms: Option<f64>,
/// CPU ms of child processes **reaped** during the unit. An arm whose work
/// sits in a still-running child contributes nothing here, so a large
/// share of the total living in this column (or a suspiciously small
/// `proc_cores_*` for an arm known to shell out) is the tell.
#[serde(default)]
pub child_cpu_ms: Option<f64>,
/// **The saturation number** — task-clock ÷ wall, mean cores this process
/// (+ reaped children) occupied. Host-independent. `None` = not measured.
#[serde(default)]
pub proc_cores_avg: Option<f64>,
/// Peak of the same ratio over the sampler's intervals. `None` = not
/// measured, which a gate must NOT read as `0.0`.
#[serde(default)]
pub proc_cores_max: Option<f64>,
}
/// The Parquet codec for **all** warehouse data-file writes — both the
/// single-batch [`append_batch`] path and the all-core
/// [`IcebergWarehouse::ingest_symbol_scans_async`] bulk path.
///
/// skade's plain `append`/`ingest_parallel` default to `UNCOMPRESSED`, which is
/// wasteful for the warehouse: `symbol_facts.signature`/`doc`, `scip`
/// occurrences and every other text-heavy column store raw. zstd at level 3 is
/// the high-ratio/low-CPU sweet spot, and in `ingest_parallel` the encode (and
/// thus the compression) runs across all cores, so the codec is parallelised
/// for free (the GATLING-law saver path). One knob, applied everywhere.
pub(crate) fn warehouse_compression() -> skade::parquet::basic::Compression {
use skade::parquet::basic::{Compression, ZstdLevel};
// `try_new(3)` only fails on an out-of-range level; 3 is always valid, so the
// fallback to the default level is purely defensive (never taken).
Compression::ZSTD(ZstdLevel::try_new(3).unwrap_or_default())
}
/// Bench/test-only override for [`warehouse_write_props`]'s duplicate-file
/// precheck. `None` (the default) means "use the `WriteProps` default", which is
/// **off** — see [`skade::WriteProps::check_duplicate_files`] for why that is
/// safe on skade's write paths (every data-file name comes from a fresh
/// per-call `unique_prefix()`, so a collision with an already-referenced path is
/// unreachable).
///
/// This exists so the `iceberg_commit_deep_*` bench arms can measure the SAME
/// append path with the check on and off in one process, and so
/// `duplicate_check_law` can assert the check still fires when asked for. It is
/// never set on any production path.
#[doc(hidden)]
pub static FORCE_CHECK_DUPLICATE: std::sync::atomic::AtomicU8 = std::sync::atomic::AtomicU8::new(0); // 0 = default, 1 = force off, 2 = force on
pub(crate) async fn append_batch(
catalog: &RedbCatalog,
table: Table,
batch: RecordBatch,
) -> Result<()> {
// Sort rows by the table's predicate columns FIRST (the data-skipping trick):
// tight, non-overlapping per-file min/max bounds in the Iceberg manifest. The
// sort key is keyed off the table name; `&[]` for tables without one (no-op).
let sort_cols = warehouse_sort_cols(table.identifier().name());
let batch = sort_batch_by(&batch, sort_cols)?;
// The whole ParquetWriter → RollingFileWriter → DataFileWriter → fast_append
// dance (and identity-partition-key tagging) lives in `skade::append_parallel_with`;
// the batch is recast to the table's field-id schema there. One call, with
// the warehouse zstd codec applied to the Parquet data file.
//
// GATLING-law write path (core-saturation): a serial `skade::append_with`
// encodes the whole batch into ONE Parquet file on ONE core — the zstd(3)
// encode of a text-heavy fact table (`symbol_facts.signature`/`doc`, scip
// occurrences) then dominates Mimir populate. `append_parallel_with` splits
// the (already predicate-sorted) rows into ~one-file-per-core ROW-ALIGNED
// zero-copy slices, fans the zstd encode across all cores on the ONE gatling
// fork-join engine, and commits every file in a SINGLE `fast_append` snapshot.
// Because the split preserves row order, each file inherits a contiguous
// sorted range — the tight non-overlapping per-file min/max bounds the sort
// was for are PRESERVED (even sharpened). skade falls back to the
// byte-identical single-file `append_props` below its ~8k-row threshold
// (`APPEND_PARALLEL_MIN_ROWS`), so small appends stay exactly as before; only
// large batches split. LAYOUT change (N files vs 1), NOT byte-identical, but
// FUNCTIONALLY equivalent: same rows, same per-column values, same snapshot —
// every order-independent read (scans, filtered/point lookups, HNSW/RRF
// search, aggregates) returns an identical result.
skade::append_parallel_with(
catalog,
&table,
std::slice::from_ref(&batch),
warehouse_compression(),
)
.await?;
Ok(())
}
/// Append many `batches` to `table` via skade's all-core
/// [`skade::ingest_parallel_with`] — the gatling topology: **every core encodes**
/// (one in-memory Parquet file per batch, warehouse zstd applied inside the
/// parallel encode), and only the `fast_append` **COMMIT is serial** (one commit
/// for the whole call). Use for a heavy multi-row write whose single-file
/// [`append_batch`] encode would pin one core — e.g. snapshot CONTENT blobs (the
/// tantivy-segment bytes), so the parquet encode of the payload column fans
/// across all cores while the writer does commit-only work.
///
/// Each batch is sorted by the table's predicate columns first (data-skipping —
/// a no-op for tables without a sort key, like the blob tables). Same
/// single-partition rule as [`append_batch`]: every batch in one call must share
/// the table's partition key (snapshot blobs all carry the same `snapshot_id`).
pub(crate) async fn ingest_batches_parallel(
catalog: &RedbCatalog,
table: Table,
batches: Vec<RecordBatch>,
) -> Result<()> {
if batches.is_empty() {
return Ok(());
}
let sort_cols = warehouse_sort_cols(table.identifier().name());
let groups: Vec<Vec<RecordBatch>> = batches
.into_iter()
.map(|b| sort_batch_by(&b, sort_cols).map(|s| vec![s]))
.collect::<Result<_>>()?;
let n_groups = groups.len().max(1);
// `files_per_commit = n_groups` ⇒ exactly ONE commit for the whole call.
skade::ingest_parallel_with(catalog, table, groups, n_groups, warehouse_compression()).await?;
Ok(())
}
pub(crate) async fn ensure_table_schema(
catalog: &RedbCatalog,
ident: &TableIdent,
table: Table,
canonical: &iceberg::spec::Schema,
) -> Result<Table> {
use std::collections::HashSet;
use iceberg::TableUpdate;
let current = table.metadata().current_schema();
let have: HashSet<&str> = current
.as_struct()
.fields()
.iter()
.map(|f| f.name.as_str())
.collect();
let missing: Vec<&str> = canonical
.as_struct()
.fields()
.iter()
.map(|f| f.name.as_str())
.filter(|n| !have.contains(n))
.collect();
if missing.is_empty() {
// Steady state: the table already carries every canonical column.
return Ok(table);
}
// Evolve straight to the **canonical** schema. Our `iceberg_schema::*` fns
// are strictly append-only with stable field ids (the file-header rule), so
// every column the stale table already has keeps the exact id it was created
// with, and the only difference is the added (trailing) columns — a pure
// Iceberg add-column. The canonical schema is internally consistent (its
// nested list/struct element ids are unique by construction), so handing it
// to `AddSchema` whole avoids any field-id arithmetic of our own. We only
// re-stamp the schema_id (current + 1) so `AddSchema` registers a new
// schema rather than colliding with the existing one.
use iceberg::spec::Schema;
let new_schema_id = current.schema_id() + 1;
let evolved = Schema::builder()
.with_schema_id(new_schema_id)
.with_fields(canonical.as_struct().fields().to_vec())
.build()?;
let updates = vec![
TableUpdate::AddSchema { schema: evolved },
// `-1` = "the schema just added" (iceberg metadata-builder convention).
TableUpdate::SetCurrentSchema { schema_id: -1 },
];
eprintln!(
"nornir: evolving warehouse table `{}` schema — adding column(s) [{}] \
(Iceberg add-column migration)",
ident.name(),
missing.join(", "),
);
catalog
.commit_table(ident.clone(), Vec::new(), updates)
.await?;
// Re-load so the caller appends against the new current schema.
Ok(catalog.load_table(ident).await?)
}
/// Convert a blank/torn-metadata-JSON scan error (`EOF at line 1 col 0`,
/// however iceberg-rust wraps it — see [`is_blank_metadata_json_error`]) into
/// an **empty** result rather than propagating it.
///
/// Every data read in the warehouse funnels through the three scan helpers
/// below. A table whose on-disk metadata/manifest/manifest-list JSON was left
/// empty (a torn write, an interrupted commit, or an empty table left behind by
/// the arrow-58 migration) means "no committed snapshot the reader can see" —
/// which for any scan is just **zero rows**, not a fatal `DataInvalid` that
/// takes down `Workspaces.Fetch` / `Warehouse.Scan` with a JSON-EOF Internal
/// error. Centralizing the guard here makes EVERY read path
/// (`scan_arrow`/`scan_limited`/`scan_filtered`/`scan_repo_filtered` and the
/// per-table query helpers that call them — `knowledge_scan_exists`,
/// `query_test_inventory`, the bench/test queries, …) degrade to empty.
pub(crate) fn empty_on_blank_metadata(
table: &str,
r: Result<Vec<RecordBatch>>,
) -> Result<Vec<RecordBatch>> {
match r {
Ok(v) => Ok(v),
Err(e) if is_blank_metadata_json_error(&e) => {
eprintln!(
"WARNING: nornir: table `{table}` has blank/torn Iceberg metadata JSON \
({e:#}); reading it as 0 rows instead of failing the scan."
);
Ok(Vec::new())
}
Err(e) => Err(e),
}
}
/// Load `table` from the catalog and `read_all` it, degrading a blank/torn
/// metadata JSON to **0 rows** at BOTH the load AND the read step.
///
/// The sibling per-table query helpers (`query_test_results`,
/// `query_test_inventory`, `release_events`, `clone_events`, …) used to do
/// `catalog.load_table(..).await?` then guard only the read — but a torn-commit
/// blank metadata JSON fails at *load* time (iceberg-rust parses the metadata
/// JSON inside `load_table`), so the `?` bubbled the `DataInvalid: EOF` before
/// the read guard ever ran and took down `Workspaces.Fetch` / `test history`.
/// Routing the WHOLE load+read through `empty_on_blank_metadata` (the table name
/// comes from the caller's const since there may be no table handle) makes every
/// such reader degrade to empty. This is the warehouse-native read seam:
/// "no readable snapshot ⇒ 0 rows", never a fatal read.
pub(crate) async fn load_and_read_all(
wh: &IcebergWarehouse,
table: &str,
) -> Result<Vec<RecordBatch>> {
empty_on_blank_metadata(
table,
async {
let t = wh.catalog().load_table(&wh.table_ident(table)).await?;
skade::read_all(&t).await.map_err(Into::into)
}
.await,
)
}
/// Like [`load_and_read_all`] but pushes `filter`/`columns` into the scan. Same
/// blank-metadata-at-load-or-read → 0 rows degradation.
pub(crate) async fn load_and_read_filtered(
wh: &IcebergWarehouse,
table: &str,
filter: &ScanFilter,
columns: &[&str],
) -> Result<Vec<RecordBatch>> {
empty_on_blank_metadata(
table,
async {
let t = wh.catalog().load_table(&wh.table_ident(table)).await?;
skade::read_filtered(&t, filter, columns)
.await
.map_err(Into::into)
}
.await,
)
}
async fn scan_all(table: &Table) -> Result<Vec<RecordBatch>> {
// Full-snapshot scan → Arrow, via skade's reader. A blank/torn metadata
// JSON degrades to 0 rows rather than EOF-erroring the whole read path.
empty_on_blank_metadata(
table.identifier().name(),
skade::read_all(table).await.map_err(Into::into),
)
}
/// Scan `table` but **stop streaming once `limit` rows are in hand** instead of
/// materializing the whole table (what `skade::read_all` does — it `try_collect`s
/// every data file before the caller can truncate).
///
/// This is the bottleneck the viz callgraph/dep-graph pays on big workspaces:
/// `njord`'s `symbol_facts`/`call_edges` are ~15 GB across ~6 k Iceberg data
/// files, so `read_all` reads all of it (~5 s/table) before the preview keeps
/// the first 8 000 rows. By driving the Iceberg scan's Arrow **stream** and
/// breaking as soon as we've accumulated `limit` rows, we read only the first
/// few data files — the remaining batches (and their parquet reads) are never
/// pulled. A small `with_batch_size` keeps the first batch from overshooting the
/// limit by much. `limit == 0` falls back to a full scan.
///
/// Caveat: this returns *some* `limit` rows, not a deterministic top-N — Iceberg
/// gives no ordering guarantee. The preview never promised one (it already
/// truncated arbitrarily), and the callgraph re-filters/-caps the rows itself,
/// so an arbitrary prefix is fine.
async fn scan_limited(table: &Table, limit: usize) -> Result<Vec<RecordBatch>> {
// The limit / early-break streaming read now lives in skade as the public,
// backend-neutral `read_limited` (it drives the Arrow stream and breaks once
// `max_rows` rows are in hand, then drops the stream to cancel the rest of
// the scan). `max_rows == 0` falls back to a full read there too, so the
// `limit == 0 -> full scan` behavior is preserved without a guard here.
// A blank/torn metadata JSON degrades to 0 rows (see `empty_on_blank_metadata`).
empty_on_blank_metadata(
table.identifier().name(),
skade::read_limited(table, limit).await.map_err(Into::into),
)
}
/// Like [`scan_all`] but pushes a `repo == <repo>` predicate into the scan
/// planner when `repo` is `Some`, so the engine prunes other repos' data
/// files / row-groups instead of reading them. No projection is applied, so
/// column order is preserved and positional [`downcast`] stays valid. The
/// caller must keep its residual per-row guard: pushdown prunes at file
/// granularity, not per row.
async fn scan_repo_filtered(table: &Table, repo: Option<&str>) -> Result<Vec<RecordBatch>> {
// The pushdown read now lives in skade as the public, backend-neutral
// `read_filtered`: it lowers the `ScanFilter` to the engine predicate,
// prunes data files / row-groups, and (empty `columns`) `select_all`s so
// column order is preserved and positional `downcast` stays valid. When
// `repo` is `None` we pass an unconstrained scan (no filter) — a full read,
// matching the previous behavior.
let columns: &[&str] = &[];
match repo {
// A blank/torn metadata JSON degrades to 0 rows (see `empty_on_blank_metadata`).
Some(r) => empty_on_blank_metadata(
table.identifier().name(),
skade::read_filtered(table, &skade::ScanFilter::eq("repo", r), columns)
.await
.map_err(Into::into),
),
// No predicate: `read_all` is the unfiltered, column-order-preserving read
// (already blank-metadata tolerant).
None => scan_all(table).await,
}
}
// ── latest-snapshot read (the `load_latest*` speed lever) ───────────────────
//
// The knowledge tables (`symbol_facts`, `call_edges`, `scip_occurrences`) are
// append-only: every `scan_repo`/`append_scip_scan` writes ONE Iceberg append
// whose rows all carry the SAME data-column `snapshot_id` + `ts_micros` (the
// scan stamp). Old appends are never deleted, so a repo's partition accrues the
// WHOLE scan history. `load_latest*` only wants the *newest* snapshot's rows.
//
// The old path `try_collect`ed the entire repo partition (all snapshots, all 12
// wide text columns) then RAM-filtered to the max-`ts_micros` snapshot — i.e. it
// decoded the full history to keep one slice of it. The two helpers below split
// that into (1) a NARROW discovery scan that reads only the 3 key columns to pick
// the latest `snapshot_id`, and (2) a `snapshot_id`-pushdown read that decodes
// ONLY the latest snapshot's data file(s) at full width. Because each append is
// one snapshot with a uniform `snapshot_id` per data file, the manifest min/max
// bounds prune every other snapshot's files at plan time — so the wide decode
// drops from "all history" to "one snapshot". See R0.2 in
// `warehouse-ragnar-fast.md`.
/// Read a child bench table (`bench_results` / `test_outcomes`) restricted to the
/// given `run_ids` by pushing `run_id IN (run_ids)` into the scan — the engine's
/// `RowFilter` skips other runs' rows at decode time. Column order is preserved
/// (`select_all`) so the bench join's positional [`downcast`]s stay valid. An
/// empty `run_ids` lowers to an always-false predicate (`IN ()` matches nothing),
/// so no rows are read — matching the old `scan_all` + `by_id` join, which would
/// have discarded every row anyway.
async fn read_runs_filtered(table: &Table, run_ids: &[String]) -> Result<Vec<RecordBatch>> {
empty_on_blank_metadata(
table.identifier().name(),
skade::read_filtered(
table,
&skade::ScanFilter::is_in("run_id", run_ids.iter().cloned()),
&[],
)
.await
.map_err(Into::into),
)
}
fn downcast<T: 'static>(batch: &RecordBatch, idx: usize) -> Result<&T> {
batch
.column(idx)
.as_any()
.downcast_ref::<T>()
.ok_or_else(|| {
anyhow!(
"column {idx} has unexpected type {:?}",
batch.column(idx).data_type()
)
})
}
/// Like [`downcast`] but by column name — used by the SCIP read-back and the
/// latest-snapshot discovery/read helpers where projection order is not assumed.
fn downcast_named<'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"))
}
fn resolve_timestamp(run: &BenchRun) -> Result<DateTime<Utc>> {
if let Some(s) = &run.timestamp {
return DateTime::parse_from_rfc3339(s)
.map(|dt| dt.with_timezone(&Utc))
.with_context(|| format!("parse timestamp {s}"));
}
let nd = NaiveDate::parse_from_str(&run.date, "%Y-%m-%d")
.with_context(|| format!("parse date {}", run.date))?;
let ndt: NaiveDateTime = nd.and_hms_opt(0, 0, 0).unwrap();
Ok(Utc.from_utc_datetime(&ndt))
}
/// Build the bench_runs RecordBatch using the table's iceberg-derived
/// Arrow schema (which carries `PARQUET:field_id` metadata).
fn build_bench_runs_batch(
table: &Table,
run_id: Uuid,
repo: &str,
ts: DateTime<Utc>,
run: &BenchRun,
commit: Option<&str>,
) -> Result<RecordBatch> {
let s = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
let mut cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(vec![run_id.to_string()])),
Arc::new(StringArray::from(vec![repo.to_string()])),
Arc::new(
TimestampMicrosecondArray::from(vec![ts.timestamp_micros()]).with_timezone("+00:00"),
),
Arc::new(StringArray::from(vec![run.date.clone()])),
Arc::new(StringArray::from(vec![run.version.clone()])),
Arc::new(StringArray::from(vec![run.machine.clone()])),
Arc::new(Int32Array::from(vec![run.cores as i32])),
];
// field 8 `commit`, when the table carries it (the writer evolves it first)
if s.fields().len() > cols.len() {
cols.push(Arc::new(StringArray::from(vec![commit.map(str::to_string)])));
}
Ok(RecordBatch::try_new(s, cols)?)
}
fn build_bench_results_batch(table: &Table, run_id: Uuid, run: &BenchRun) -> Result<RecordBatch> {
let s = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
let id_str = run_id.to_string();
let mut ids = Vec::new();
let mut names = Vec::new();
let mut metrics = Vec::new();
let mut values = Vec::new();
for r in &run.results {
for (k, v) in &r.metrics {
// `telem_*` keys are the in-bench sampler's CPU/mem footprint; they
// ride the metrics map only to survive the stdout-JSON hop and are
// persisted into the dedicated `bench_telemetry` table instead, so
// they never pollute the scalar metric set / docs comparison.
if k.starts_with(crate::bench::telemetry::TELEM_PREFIX) {
continue;
}
let f = match v {
serde_json::Value::Number(n) => n.as_f64(),
_ => continue,
};
if let Some(f) = f {
ids.push(id_str.clone());
names.push(r.name.clone());
metrics.push(k.clone());
values.push(f);
}
}
}
let cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(ids)),
Arc::new(StringArray::from(names)),
Arc::new(StringArray::from(metrics)),
Arc::new(Float64Array::from(values)),
];
Ok(RecordBatch::try_new(s, cols)?)
}
/// Build the `bench_telemetry` batch: one row per result that carries `telem_*`
/// metrics (the in-bench sampler's footprint). Results without telemetry (legacy
/// runs, or benches that produced no telemetry) contribute no row, so a missing
/// sampler degrades to an empty table rather than zero-filled noise.
///
/// The process-scoped columns (12-15) are only appended when the loaded table
/// actually carries them — the caller runs `ensure_table_schema` first, so in
/// practice it always does, but a table that somehow did not evolve keeps
/// writing its 11 columns instead of failing the whole append.
fn build_bench_telemetry_batch(
table: &Table,
run_id: Uuid,
repo: &str,
run: &BenchRun,
) -> Result<RecordBatch> {
let s = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
let has_proc = s.column_with_name("proc_cores_max").is_some();
let id_str = run_id.to_string();
let mut ids = Vec::new();
let mut repos = Vec::new();
let mut benches = Vec::new();
let mut n_cores = Vec::new();
let mut cpu_avg = Vec::new();
let mut cpu_max = Vec::new();
let mut busy_avg = Vec::new();
let mut busy_max = Vec::new();
let mut mem_peak = Vec::new();
let mut mem_pct = Vec::new();
let mut elapsed = Vec::new();
let mut proc_cpu: Vec<Option<f64>> = Vec::new();
let mut child_cpu: Vec<Option<f64>> = Vec::new();
let mut proc_avg: Vec<Option<f64>> = Vec::new();
let mut proc_max: Vec<Option<f64>> = Vec::new();
for r in &run.results {
let Some(t) = crate::bench::telemetry::from_metrics(&r.metrics) else {
continue;
};
ids.push(id_str.clone());
repos.push(repo.to_string());
benches.push(r.name.clone());
n_cores.push(t.n_cores as i32);
cpu_avg.push(t.cpu_pct_avg);
cpu_max.push(t.cpu_pct_max);
busy_avg.push(t.cores_busy_avg);
busy_max.push(t.cores_busy_max as i32);
mem_peak.push(t.mem_peak_mb);
mem_pct.push(t.mem_pct_max);
elapsed.push(t.elapsed_ms);
// `None` stays NULL — "not measured" — never a fabricated 0.0.
proc_cpu.push(t.proc_cpu_ms);
child_cpu.push(t.child_cpu_ms);
proc_avg.push(t.proc_cores_avg);
proc_max.push(t.proc_cores_max);
}
let mut cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(ids)),
Arc::new(StringArray::from(repos)),
Arc::new(StringArray::from(benches)),
Arc::new(Int32Array::from(n_cores)),
Arc::new(Float64Array::from(cpu_avg)),
Arc::new(Float64Array::from(cpu_max)),
Arc::new(Float64Array::from(busy_avg)),
Arc::new(Int32Array::from(busy_max)),
Arc::new(Float64Array::from(mem_peak)),
Arc::new(Float64Array::from(mem_pct)),
Arc::new(Float64Array::from(elapsed)),
];
if has_proc {
cols.push(Arc::new(Float64Array::from(proc_cpu)));
cols.push(Arc::new(Float64Array::from(child_cpu)));
cols.push(Arc::new(Float64Array::from(proc_avg)));
cols.push(Arc::new(Float64Array::from(proc_max)));
}
Ok(RecordBatch::try_new(s, cols)?)
}
fn build_test_outcomes_batch(table: &Table, run_id: Uuid, run: &BenchRun) -> Result<RecordBatch> {
let s = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
let id_str = run_id.to_string();
let n = run.tests.len();
let ids: Vec<String> = run.tests.iter().map(|_| id_str.clone()).collect();
let names: Vec<String> = run.tests.iter().map(|t| t.name.clone()).collect();
let passed: Vec<bool> = run.tests.iter().map(|t| t.passed).collect();
let durations: Vec<Option<f64>> = run.tests.iter().map(|t| t.duration_ms).collect();
let messages: Vec<Option<String>> = run.tests.iter().map(|t| t.message.clone()).collect();
// Fields 6-10 are optional metadata for impact selection + flake
// quarantine — the bench harness doesn't populate them yet, but the
// schema requires the columns to be present (null-filled).
let nulls_string: Vec<Option<String>> = vec![None; n];
let nulls_int: Vec<Option<i32>> = vec![None; n];
let nulls_bool: Vec<Option<bool>> = vec![None; n];
let mut cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(ids)),
Arc::new(StringArray::from(names)),
Arc::new(BooleanArray::from(passed)),
Arc::new(Float64Array::from(durations)),
Arc::new(StringArray::from(messages)),
Arc::new(StringArray::from(nulls_string.clone())), // source_repo
Arc::new(StringArray::from(nulls_string.clone())), // source_crate
Arc::new(Int32Array::from(nulls_int.clone())), // attempt_idx
Arc::new(StringArray::from(nulls_string.clone())), // retry_of_run_id
Arc::new(BooleanArray::from(nulls_bool)), // quarantined
];
// fields 11-12 (`status`, `commit`): a bench-coupled row's `bench_runs` row
// says both, so they stay NULL here
while s.fields().len() > cols.len() {
cols.push(Arc::new(StringArray::from(nulls_string.clone())));
}
Ok(RecordBatch::try_new(s, cols)?)
}
/// One case a Runa test step ran, as [`IcebergWarehouse::append_step_test_outcomes_async`] files it.
#[derive(Debug, Clone, PartialEq)]
pub struct StepTestOutcome {
/// Its path inside its crate (`cart::tests::adds_up`).
pub name: String,
/// The crate it lives in.
pub krate: String,
/// `pass` | `fail` | `ignored`.
pub status: String,
pub duration_ms: Option<f64>,
pub message: Option<String>,
}
fn build_step_test_outcomes_batch(
table: &Table,
repo: &str,
commit: &str,
run_id: &str,
cases: &[StepTestOutcome],
) -> Result<RecordBatch> {
let s = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
let n = cases.len();
let text = |f: &dyn Fn(&StepTestOutcome) -> Option<String>| -> Arc<dyn Array> {
Arc::new(StringArray::from(cases.iter().map(f).collect::<Vec<_>>()))
};
let commit = Some(commit.to_string()).filter(|c| !c.trim().is_empty());
let cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(vec![run_id.to_string(); n])),
Arc::new(StringArray::from(cases.iter().map(|c| c.name.clone()).collect::<Vec<_>>())),
Arc::new(BooleanArray::from(cases.iter().map(|c| c.status == "pass").collect::<Vec<_>>())),
Arc::new(Float64Array::from(cases.iter().map(|c| c.duration_ms).collect::<Vec<_>>())),
text(&|c| c.message.clone().filter(|m| !m.is_empty())),
text(&|_| Some(repo.to_string())), // source_repo
text(&|c| Some(c.krate.clone()).filter(|k| !k.is_empty())), // source_crate
Arc::new(Int32Array::from(vec![None::<i32>; n])), // attempt_idx
text(&|_| None), // retry_of_run_id
Arc::new(BooleanArray::from(vec![None::<bool>; n])), // quarantined
text(&|c| Some(c.status.clone())), // status
text(&|_| commit.clone()), // commit
];
Ok(RecordBatch::try_new(s, cols)?)
}
// Silence dead-code warnings for items used only via the public re-export.
#[allow(dead_code)]
fn _unused_imports_anchor() {}
// ─── knowledge-map writers ──────────────────────────────────────────────
impl IcebergWarehouse {
/// Persist a cross-repo dependency-graph snapshot (sync wrapper over the
/// async [`crate::warehouse::dep_graph::record_dep_graph`]) on this
/// warehouse's private runtime. Returns the snapshot id.
pub fn record_dep_graph(
&self,
workspace_name: &str,
graph: &crate::warehouse::dep_graph::WorkspaceGraph,
) -> Result<uuid::Uuid> {
self.rt.as_ref().expect("rt present").block_on(
crate::warehouse::dep_graph::record_dep_graph(self, workspace_name, graph),
)
}
/// Persist a [`crate::knowledge::git_heat::GitHeatScan`] into `git_heat_facts`.
pub fn append_git_heat_scan(
&self,
scan: &crate::knowledge::git_heat::GitHeatScan,
) -> Result<()> {
self.rt
.as_ref()
.expect("rt present")
.block_on(self.append_git_heat_scan_async(scan))
}
pub async fn append_git_heat_scan_async(
&self,
scan: &crate::knowledge::git_heat::GitHeatScan,
) -> Result<()> {
if scan.files.is_empty() {
return Ok(());
}
let ts = scan.ts.timestamp_micros();
let snap = scan.snapshot_id.to_string();
let table = load_table_for_append(
&self.catalog,
&self.namespace,
TABLE_GIT_HEAT_FACTS,
iceberg_schema::git_heat_facts()?,
&["repo"],
)
.await?;
let s = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
let n = scan.files.len();
let cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(vec![snap; n])),
Arc::new(TimestampMicrosecondArray::from(vec![ts; n]).with_timezone("+00:00")),
Arc::new(StringArray::from(vec![scan.repo.clone(); n])),
Arc::new(StringArray::from(
scan.files
.iter()
.map(|r| r.file.clone())
.collect::<Vec<_>>(),
)),
Arc::new(Int64Array::from(
scan.files
.iter()
.map(|r| r.commits_total)
.collect::<Vec<_>>(),
)),
Arc::new(Int64Array::from(
scan.files.iter().map(|r| r.commits_30d).collect::<Vec<_>>(),
)),
Arc::new(Int64Array::from(
scan.files.iter().map(|r| r.commits_90d).collect::<Vec<_>>(),
)),
Arc::new(Int64Array::from(
scan.files
.iter()
.map(|r| r.authors_total)
.collect::<Vec<_>>(),
)),
Arc::new(
TimestampMicrosecondArray::from(
scan.files
.iter()
.map(|r| r.last_commit_ts.timestamp_micros())
.collect::<Vec<_>>(),
)
.with_timezone("+00:00"),
),
];
let batch = RecordBatch::try_new(s, cols)?;
append_batch(&self.catalog, table, batch).await?;
Ok(())
}
/// Append a batch of MCP tool-call telemetry rows (one snapshot). Metrics
/// only — no payloads. Called by the nornir-mcp server's logging thread.
pub async fn append_mcp_calls_async(&self, calls: &[McpCall]) -> Result<()> {
if calls.is_empty() {
return Ok(());
}
let table = self
.catalog
.load_table(&self.table_ident(TABLE_MCP_REQUESTS))
.await?;
let s = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
let cols: Vec<Arc<dyn Array>> = vec![
Arc::new(
TimestampMicrosecondArray::from(
calls.iter().map(|c| c.ts_micros).collect::<Vec<_>>(),
)
.with_timezone("+00:00"),
),
Arc::new(StringArray::from(
calls.iter().map(|c| c.tool.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
calls.iter().map(|c| c.status.clone()).collect::<Vec<_>>(),
)),
Arc::new(Int64Array::from(
calls.iter().map(|c| c.latency_ms).collect::<Vec<_>>(),
)),
];
let batch = RecordBatch::try_new(s, cols)?;
append_batch(&self.catalog, table, batch).await?;
Ok(())
}
/// Sync wrapper over [`append_mcp_calls_async`](Self::append_mcp_calls_async).
pub fn append_mcp_calls(&self, calls: &[McpCall]) -> Result<()> {
self.rt
.as_ref()
.expect("rt present")
.block_on(self.append_mcp_calls_async(calls))
}
/// **Keep a lens answer in the lake** under the STATE KEY of the room it
/// was folded from (`Room::state_key`: identity + scope + lake stamp) —
/// the lake IS the lens cache (the owner, 2026-10-18): a restarted server
/// reads this row back instead of folding the lens again. The partition
/// column keeps its `room_version` name; its values are state keys (the
/// old per-process counter collided across restarts and could serve an
/// older state's answer).
pub async fn append_lens_answer_async(
&self,
state_key: &str,
key: &str,
state: &str,
body: &str,
) -> Result<()> {
let table = self
.catalog
.load_table(&self.table_ident(TABLE_LENS_ANSWERS))
.await?;
let s = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
let ts = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_micros() as i64)
.unwrap_or(0);
let cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(vec![state_key.to_string()])),
Arc::new(StringArray::from(vec![key.to_string()])),
Arc::new(StringArray::from(vec![state.to_string()])),
Arc::new(StringArray::from(vec![body.to_string()])),
Arc::new(TimestampMicrosecondArray::from(vec![ts]).with_timezone("+00:00")),
];
let batch = RecordBatch::try_new(s, cols)?;
append_batch(&self.catalog, table, batch).await?;
Ok(())
}
/// Sync wrapper over [`append_lens_answer_async`](Self::append_lens_answer_async).
pub fn append_lens_answer(
&self,
state_key: &str,
key: &str,
state: &str,
body: &str,
) -> Result<()> {
self.rt
.as_ref()
.expect("rt present")
.block_on(self.append_lens_answer_async(state_key, key, state, body))
}
/// **Read a kept lens answer back** for exactly this (state key, key) —
/// the latest row wins. `None` means the lake holds no answer for this
/// lake state: the caller folds and keeps one. The scan prunes to the
/// state's partition, so other states' bodies are never read.
pub fn read_lens_answer(&self, state_key: &str, key: &str) -> Result<Option<String>> {
let batches = self.scan_filtered(
TABLE_LENS_ANSWERS,
&ScanFilter::eq("room_version", state_key.to_string()),
&[],
)?;
let mut best: Option<(i64, String)> = None;
for b in &batches {
let k = downcast::<StringArray>(b, 1)?;
let body = downcast::<StringArray>(b, 3)?;
let ts = downcast::<TimestampMicrosecondArray>(b, 4)?;
for i in 0..b.num_rows() {
if k.value(i) == key && best.as_ref().is_none_or(|(t, _)| ts.value(i) >= *t) {
best = Some((ts.value(i), body.value(i).to_string()));
}
}
}
Ok(best.map(|(_, body)| body))
}
/// Aggregate the `mcp_requests` table into per-tool usage stats (call count,
/// error count, mean latency, last-seen), sorted by call count descending.
pub async fn query_mcp_stats_async(&self) -> Result<Vec<McpToolStat>> {
let table = self
.catalog
.load_table(&self.table_ident(TABLE_MCP_REQUESTS))
.await?;
let batches = scan_all(&table).await?;
struct Acc {
calls: u64,
errors: u64,
lat_sum: i128,
last_ts: i64,
}
let mut map: std::collections::HashMap<String, Acc> = std::collections::HashMap::new();
for b in &batches {
let ts = downcast::<TimestampMicrosecondArray>(b, 0)?;
let tool = downcast::<StringArray>(b, 1)?;
let status = downcast::<StringArray>(b, 2)?;
let lat = downcast::<Int64Array>(b, 3)?;
for i in 0..b.num_rows() {
let e = map.entry(tool.value(i).to_string()).or_insert(Acc {
calls: 0,
errors: 0,
lat_sum: 0,
last_ts: i64::MIN,
});
e.calls += 1;
if status.value(i) != "ok" {
e.errors += 1;
}
e.lat_sum += lat.value(i) as i128;
e.last_ts = e.last_ts.max(ts.value(i));
}
}
let mut out: Vec<McpToolStat> = map
.into_iter()
.map(|(tool, a)| McpToolStat {
tool,
calls: a.calls,
errors: a.errors,
avg_latency_ms: if a.calls > 0 {
a.lat_sum as f64 / a.calls as f64
} else {
0.0
},
last_ts_micros: a.last_ts,
})
.collect();
out.sort_by(|a, b| b.calls.cmp(&a.calls).then_with(|| a.tool.cmp(&b.tool)));
Ok(out)
}
/// Sync wrapper over [`query_mcp_stats_async`](Self::query_mcp_stats_async).
pub fn query_mcp_stats(&self) -> Result<Vec<McpToolStat>> {
self.rt
.as_ref()
.expect("rt present")
.block_on(self.query_mcp_stats_async())
}
/// The `vuln_findings` cache: `crate@version → (advisory ids, summary)`. The
/// security scan reads this warehouse-first; only un-cached crate@versions
/// hit OSV. Empty `ids` = checked and clean (still a cache hit).
pub async fn query_vuln_findings_async(
&self,
) -> Result<std::collections::HashMap<(String, String), (Vec<String>, String, i64)>> {
let table = self
.catalog
.load_table(&self.table_ident(TABLE_VULN_FINDINGS))
.await?;
let batches = scan_all(&table).await?;
// Value = (advisory ids, summary, checked_at_micros) — keep the LATEST
// check per crate@version (the table is append-only, so re-checks add rows).
let mut map: std::collections::HashMap<(String, String), (Vec<String>, String, i64)> =
std::collections::HashMap::new();
for b in &batches {
let kr = downcast::<StringArray>(b, 0)?;
let ver = downcast::<StringArray>(b, 1)?;
let ids = downcast::<StringArray>(b, 2)?;
let sum = downcast::<StringArray>(b, 3)?;
let ts = downcast::<TimestampMicrosecondArray>(b, 4)?;
for i in 0..b.num_rows() {
let key = (kr.value(i).to_string(), ver.value(i).to_string());
let checked = ts.value(i);
let entry = map
.entry(key)
.or_insert((Vec::new(), String::new(), i64::MIN));
if checked >= entry.2 {
let id_vec: Vec<String> = ids
.value(i)
.split(',')
.filter(|s| !s.is_empty())
.map(String::from)
.collect();
*entry = (id_vec, sum.value(i).to_string(), checked);
}
}
}
Ok(map)
}
/// Sync wrapper over [`query_vuln_findings_async`](Self::query_vuln_findings_async).
pub fn query_vuln_findings(
&self,
) -> Result<std::collections::HashMap<(String, String), (Vec<String>, String, i64)>> {
self.rt
.as_ref()
.expect("rt present")
.block_on(self.query_vuln_findings_async())
}
/// Persist freshly-checked `crate@version` vuln rows (fills the cache).
pub async fn append_vuln_findings_async(&self, rows: &[VulnFinding]) -> Result<()> {
if rows.is_empty() {
return Ok(());
}
let table = self
.catalog
.load_table(&self.table_ident(TABLE_VULN_FINDINGS))
.await?;
let s = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
let cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(
rows.iter()
.map(|r| r.crate_name.clone())
.collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.version.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.ids.join(",")).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.summary.clone()).collect::<Vec<_>>(),
)),
Arc::new(
TimestampMicrosecondArray::from(
rows.iter().map(|r| r.checked_at_micros).collect::<Vec<_>>(),
)
.with_timezone("+00:00"),
),
];
let batch = RecordBatch::try_new(s, cols)?;
append_batch(&self.catalog, table, batch).await?;
Ok(())
}
/// Sync wrapper over [`append_vuln_findings_async`](Self::append_vuln_findings_async).
pub fn append_vuln_findings(&self, rows: &[VulnFinding]) -> Result<()> {
self.rt
.as_ref()
.expect("rt present")
.block_on(self.append_vuln_findings_async(rows))
}
/// Persist a freshly-resolved SBOM component set for `repo` under a single
/// capture `snapshot_id` (one row per component). Append-only: the latest
/// snapshot per repo is the authoritative one (see
/// [`query_sbom_components`](Self::query_sbom_components)).
pub async fn append_sbom_components_async(
&self,
repo: &str,
snapshot_id: Uuid,
components: &[SbomComponentRow],
) -> Result<()> {
if components.is_empty() {
return Ok(());
}
let table = load_table_for_append(
&self.catalog,
&self.namespace,
TABLE_SBOM_COMPONENTS,
iceberg_schema::sbom_components()?,
&["repo"],
)
.await?;
let s = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
let n = components.len();
let snap = snapshot_id.to_string();
let ts = Utc::now().timestamp_micros();
let cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(vec![snap; n])),
Arc::new(TimestampMicrosecondArray::from(vec![ts; n]).with_timezone("+00:00")),
Arc::new(StringArray::from(vec![repo.to_string(); n])),
Arc::new(StringArray::from(
components
.iter()
.map(|c| c.name.clone())
.collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
components
.iter()
.map(|c| c.version.clone())
.collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
components
.iter()
.map(|c| c.license.clone())
.collect::<Vec<_>>(),
)),
// derived_from — `source` for a blank value (backward-compatible).
Arc::new(StringArray::from(
components
.iter()
.map(|c| {
if c.derived_from.is_empty() {
"source".to_string()
} else {
c.derived_from.clone()
}
})
.collect::<Vec<_>>(),
)),
];
let batch = RecordBatch::try_new(s, cols)?;
append_batch(&self.catalog, table, batch).await?;
Ok(())
}
/// Sync wrapper over [`append_sbom_components_async`](Self::append_sbom_components_async).
pub fn append_sbom_components(
&self,
repo: &str,
snapshot_id: Uuid,
components: &[SbomComponentRow],
) -> Result<()> {
self.rt
.as_ref()
.expect("rt present")
.block_on(self.append_sbom_components_async(repo, snapshot_id, components))
}
/// The latest captured SBOM component set for `repo` (the rows of the
/// most-recent `snapshot_id`, by capture ts). `None` when the repo has never
/// been captured — the caller should then fall back to live `cargo metadata`.
pub async fn query_sbom_components_async(
&self,
repo: &str,
) -> Result<Option<Vec<SbomComponentRow>>> {
// Blank/torn metadata JSON → 0 rows, guarded at load AND read.
let batches: Vec<RecordBatch> = load_and_read_filtered(
self,
TABLE_SBOM_COMPONENTS,
&skade::ScanFilter::eq("repo", repo),
&[],
)
.await?;
// Find the latest snapshot_id for this repo (max ts_micros).
let mut latest: Option<(String, i64)> = None;
for b in &batches {
let snaps = downcast::<StringArray>(b, 0)?;
let ts = downcast::<TimestampMicrosecondArray>(b, 1)?;
let repos = downcast::<StringArray>(b, 2)?;
for i in 0..b.num_rows() {
if repos.value(i) != repo {
continue;
}
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(None);
};
let mut out = Vec::new();
for b in &batches {
let snaps = downcast::<StringArray>(b, 0)?;
let name = downcast::<StringArray>(b, 3)?;
let version = downcast::<StringArray>(b, 4)?;
let license = downcast::<StringArray>(b, 5)?;
// Read `derived_from` by name so a table written before the column
// (task #65 slice 4a) reads back as `source` rather than erroring.
let derived = downcast_named::<StringArray>(b, "derived_from").ok();
for i in 0..b.num_rows() {
if snaps.value(i) != snap {
continue;
}
let derived_from = match derived {
Some(d) if !d.is_null(i) => d.value(i).to_string(),
_ => "source".to_string(),
};
out.push(SbomComponentRow {
name: name.value(i).to_string(),
version: version.value(i).to_string(),
license: license.value(i).to_string(),
derived_from,
});
}
}
Ok(Some(out))
}
/// Sync wrapper over [`query_sbom_components_async`](Self::query_sbom_components_async).
pub fn query_sbom_components(&self, repo: &str) -> Result<Option<Vec<SbomComponentRow>>> {
self.rt
.as_ref()
.expect("rt present")
.block_on(self.query_sbom_components_async(repo))
}
// ── Binary-analysis facts (task #65) — header + section + symbol sizes,
// all keyed by `content_sha256` (the build_info.content_id join key). Writes
// mirror `append_sbom_components`; reads return the latest capture snapshot
// for a `content_sha256`. See `.nornir/binary-analysis-design.md` §5. ──
/// Persist the `binary_facts` header row for one analysed artifact under a
/// single capture `snapshot_id`.
pub async fn append_binary_facts_async(
&self,
snapshot_id: Uuid,
row: &BinaryFactRow,
) -> Result<()> {
let table = load_table_for_append(
&self.catalog,
&self.namespace,
TABLE_BINARY_FACTS,
iceberg_schema::binary_facts()?,
&["content_sha256"],
)
.await?;
let s = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
let ts = Utc::now().timestamp_micros();
let cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(vec![snapshot_id.to_string()])),
Arc::new(TimestampMicrosecondArray::from(vec![ts]).with_timezone("+00:00")),
Arc::new(StringArray::from(vec![row.repo.clone()])),
Arc::new(StringArray::from(vec![row.git_sha.clone()])),
Arc::new(StringArray::from(vec![row.build_id.clone()])),
Arc::new(StringArray::from(vec![row.content_sha256.clone()])),
Arc::new(StringArray::from(vec![row.artifact_path.clone()])),
Arc::new(StringArray::from(vec![row.format.clone()])),
Arc::new(StringArray::from(vec![row.arch.clone()])),
Arc::new(StringArray::from(vec![row.class.clone()])),
Arc::new(Int64Array::from(vec![row.total_bytes])),
Arc::new(BooleanArray::from(vec![row.stripped])),
Arc::new(BooleanArray::from(vec![row.has_debug_info])),
Arc::new(BooleanArray::from(vec![row.pie])),
Arc::new(StringArray::from(vec![row.relro.clone()])),
Arc::new(BooleanArray::from(vec![row.nx])),
Arc::new(BooleanArray::from(vec![row.stack_canary])),
Arc::new(BooleanArray::from(vec![row.fortify])),
Arc::new(BooleanArray::from(vec![row.rpath])),
Arc::new(BooleanArray::from(vec![row.runpath])),
Arc::new(StringArray::from(vec![row.build_id_note.clone()])),
Arc::new(Int64Array::from(vec![row.n_sections])),
Arc::new(Int64Array::from(vec![row.n_sym_defined])),
Arc::new(Int64Array::from(vec![row.n_sym_undefined])),
Arc::new(Int64Array::from(vec![row.n_dyn_deps])),
];
let batch = RecordBatch::try_new(s, cols)?;
append_batch(&self.catalog, table, batch).await?;
Ok(())
}
/// Persist the `section_sizes` rows for one artifact under `snapshot_id`.
pub async fn append_section_sizes_async(
&self,
snapshot_id: Uuid,
content_sha256: &str,
rows: &[SectionSizeRow],
) -> Result<()> {
if rows.is_empty() {
return Ok(());
}
let table = load_table_for_append(
&self.catalog,
&self.namespace,
TABLE_SECTION_SIZES,
iceberg_schema::section_sizes()?,
&["content_sha256"],
)
.await?;
let s = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
let n = rows.len();
let ts = Utc::now().timestamp_micros();
let cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(vec![snapshot_id.to_string(); n])),
Arc::new(TimestampMicrosecondArray::from(vec![ts; n]).with_timezone("+00:00")),
Arc::new(StringArray::from(vec![content_sha256.to_string(); n])),
Arc::new(StringArray::from(
rows.iter()
.map(|r| r.section_name.clone())
.collect::<Vec<_>>(),
)),
Arc::new(Int64Array::from(
rows.iter().map(|r| r.size_bytes).collect::<Vec<_>>(),
)),
Arc::new(Int64Array::from(
rows.iter().map(|r| r.vaddr).collect::<Vec<_>>(),
)),
Arc::new(BooleanArray::from(
rows.iter().map(|r| r.alloc).collect::<Vec<_>>(),
)),
Arc::new(BooleanArray::from(
rows.iter().map(|r| r.exec).collect::<Vec<_>>(),
)),
Arc::new(BooleanArray::from(
rows.iter().map(|r| r.write).collect::<Vec<_>>(),
)),
Arc::new(Int64Array::from(
rows.iter().map(|r| r.file_offset).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.kind.clone()).collect::<Vec<_>>(),
)),
];
let batch = RecordBatch::try_new(s, cols)?;
append_batch(&self.catalog, table, batch).await?;
Ok(())
}
/// Persist the `symbol_sizes` rows for one artifact under `snapshot_id`.
pub async fn append_symbol_sizes_async(
&self,
snapshot_id: Uuid,
content_sha256: &str,
rows: &[SymbolSizeRow],
) -> Result<()> {
if rows.is_empty() {
return Ok(());
}
let table = load_table_for_append(
&self.catalog,
&self.namespace,
TABLE_SYMBOL_SIZES,
iceberg_schema::symbol_sizes()?,
&["content_sha256"],
)
.await?;
let s = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
let n = rows.len();
let ts = Utc::now().timestamp_micros();
let cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(vec![snapshot_id.to_string(); n])),
Arc::new(TimestampMicrosecondArray::from(vec![ts; n]).with_timezone("+00:00")),
Arc::new(StringArray::from(vec![content_sha256.to_string(); n])),
Arc::new(StringArray::from(
rows.iter()
.map(|r| r.symbol_mangled.clone())
.collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter()
.map(|r| r.symbol_demangled.clone())
.collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter()
.map(|r| r.symbol_demangled_no_generics.clone())
.collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter()
.map(|r| r.crate_name.clone())
.collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.section.clone()).collect::<Vec<_>>(),
)),
Arc::new(Int64Array::from(
rows.iter().map(|r| r.size_bytes).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.kind.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.binding.clone()).collect::<Vec<_>>(),
)),
Arc::new(BooleanArray::from(
rows.iter()
.map(|r| r.is_monomorphization)
.collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter()
.map(|r| r.source_file.clone())
.collect::<Vec<_>>(),
)),
Arc::new(Int32Array::from(
rows.iter()
.map(|r| r.source_line.map(|v| v as i32))
.collect::<Vec<_>>(),
)),
];
let batch = RecordBatch::try_new(s, cols)?;
append_batch(&self.catalog, table, batch).await?;
Ok(())
}
/// Sync wrapper: persist a full binary analysis (header + sections + symbols)
/// for one artifact under a single capture `snapshot_id`.
pub fn append_binary_analysis(
&self,
snapshot_id: Uuid,
header: &BinaryFactRow,
sections: &[SectionSizeRow],
symbols: &[SymbolSizeRow],
) -> Result<()> {
let rt = self.rt.as_ref().expect("rt present");
rt.block_on(self.append_binary_facts_async(snapshot_id, header))?;
rt.block_on(self.append_section_sizes_async(
snapshot_id,
&header.content_sha256,
sections,
))?;
rt.block_on(self.append_symbol_sizes_async(snapshot_id, &header.content_sha256, symbols))?;
Ok(())
}
/// The latest-captured `binary_facts` header for `content_sha256`, or `None`
/// if that artifact was never analysed.
pub async fn query_binary_facts_async(
&self,
content_sha256: &str,
) -> Result<Option<BinaryFactRow>> {
let batches: Vec<RecordBatch> = load_and_read_filtered(
self,
TABLE_BINARY_FACTS,
&skade::ScanFilter::eq("content_sha256", content_sha256),
&[],
)
.await?;
// Latest snapshot for this artifact (max ts_micros).
let mut latest: Option<(usize, usize, i64)> = None; // (batch, row, ts)
for (bi, b) in batches.iter().enumerate() {
let content = downcast::<StringArray>(b, 5)?;
let ts = downcast::<TimestampMicrosecondArray>(b, 1)?;
for i in 0..b.num_rows() {
if content.value(i) != content_sha256 {
continue;
}
let t = ts.value(i);
if latest.as_ref().map(|(_, _, lt)| t > *lt).unwrap_or(true) {
latest = Some((bi, i, t));
}
}
}
let Some((bi, i, _)) = latest else {
return Ok(None);
};
let b = &batches[bi];
Ok(Some(BinaryFactRow {
repo: downcast::<StringArray>(b, 2)?.value(i).to_string(),
git_sha: downcast::<StringArray>(b, 3)?.value(i).to_string(),
build_id: downcast::<StringArray>(b, 4)?.value(i).to_string(),
content_sha256: downcast::<StringArray>(b, 5)?.value(i).to_string(),
artifact_path: downcast::<StringArray>(b, 6)?.value(i).to_string(),
format: downcast::<StringArray>(b, 7)?.value(i).to_string(),
arch: downcast::<StringArray>(b, 8)?.value(i).to_string(),
class: downcast::<StringArray>(b, 9)?.value(i).to_string(),
total_bytes: downcast::<Int64Array>(b, 10)?.value(i),
stripped: downcast::<BooleanArray>(b, 11)?.value(i),
has_debug_info: downcast::<BooleanArray>(b, 12)?.value(i),
pie: downcast::<BooleanArray>(b, 13)?.value(i),
relro: downcast::<StringArray>(b, 14)?.value(i).to_string(),
nx: downcast::<BooleanArray>(b, 15)?.value(i),
stack_canary: downcast::<BooleanArray>(b, 16)?.value(i),
fortify: downcast::<BooleanArray>(b, 17)?.value(i),
rpath: downcast::<BooleanArray>(b, 18)?.value(i),
runpath: downcast::<BooleanArray>(b, 19)?.value(i),
build_id_note: downcast::<StringArray>(b, 20)?.value(i).to_string(),
n_sections: downcast::<Int64Array>(b, 21)?.value(i),
n_sym_defined: downcast::<Int64Array>(b, 22)?.value(i),
n_sym_undefined: downcast::<Int64Array>(b, 23)?.value(i),
n_dyn_deps: downcast::<Int64Array>(b, 24)?.value(i),
}))
}
/// Sync wrapper over [`query_binary_facts_async`](Self::query_binary_facts_async).
pub fn query_binary_facts(&self, content_sha256: &str) -> Result<Option<BinaryFactRow>> {
self.rt
The file goes on; the console shows all of it.