crates/nornir-warehouse-trait/src/lib.rs
//! nornir-warehouse-trait — the durable, backend-neutral **Warehouse seam**.
//!
//! This crate holds ONLY the abstraction, extracted verbatim from nornir so a
//! consumer (dwarves, a future SkadeWarehouse, …) can depend on the trait and
//! its data types WITHOUT depending on nornir:
//!
//! * the [`Warehouse`] async trait (the storage seam),
//! * its POD arg/return types: [`BenchFilter`], [`TablePreview`], [`ColumnInfo`],
//! * the pure knowledge MODEL PODs — [`GitHeatScan`]/[`GitHeatRow`] (the trait's
//! async surface names them) and the symbol / call ROW model
//! [`SymbolScan`]/[`SymbolRow`]/[`CallEdgeRow`]/[`FeatureGateRow`]/[`TestDefRow`]
//! and [`ScipScan`]/[`ScipRow`].
//!
//! The gix **scanner** that BUILDS the git-heat PODs stays in nornir and imports
//! them from here; nornir re-exports every type at its old module path so
//! existing `crate::warehouse::…` / `crate::knowledge::…` call sites are
//! untouched.
//!
//! The symbol / call rows are a MODEL only since Mímir v2 (2026-10-13): nornir's
//! syn scan and its `rust-analyzer scip` ingest are deleted, so NOTHING writes
//! `symbol_facts` / `call_edges` / `scip_occurrences` any more and the trait no
//! longer carries a symbol-scan writer or a SCIP reader. The code graph is
//! nornir's `mimir_v2` lake, a catalog of its own that this seam does not read.
use anyhow::Result;
use arrow::array::RecordBatch;
use uuid::Uuid;
use chrono::{DateTime, Utc};
use nornir_bench::BenchRun;
// Re-export skade's backend-neutral scan predicate so trait callers can build
// pushdown filters (`ScanFilter::eq("repo", repo)`, `is_in`, `range`) without
// naming any engine type. This is the durable predicate surface a future
// `SkadeWarehouse` / DuckDB backend implements identically; the Iceberg backend
// lowers it to an `iceberg::expr::Predicate` inside skade.
pub use skade::{Scalar, ScanFilter};
// ─── knowledge-map table names ──────────────────────────────────────────
//
// A table name is part of the seam for the same reason the trait is: if two
// sides spell it in two places they can drift. Named once, here.
//
// NO WRITER since Mímir v2 (2026-10-13): all three were written by nornir's
// deleted syn scan / SCIP ingest. A warehouse made before that still holds
// them — stale — until its owner purges them; a new one never has them. A
// reader of one of these names must say so, never show the 0 rows (or the old
// rows) as an answer.
/// The `symbol_facts` table: one row per syn-discovered symbol. No writer.
pub const TABLE_SYMBOL_FACTS: &str = "symbol_facts";
/// The `call_edges` table: one row per syn-discovered caller→callee edge. No writer.
pub const TABLE_CALL_EDGES: &str = "call_edges";
/// The `scip_occurrences` table: resolved (SCIP) occurrences. No writer.
pub const TABLE_SCIP_OCCURRENCES: &str = "scip_occurrences";
/// The durable storage seam for nornir's warehouse (bench/test/release/symbol
/// history, etc.). **Sync** by design — every method blocks internally on the
/// backend's own runtime (`IcebergWarehouse` owns a tokio `Runtime` and calls
/// `block_on`), so the ~150 sync call sites never have to become async.
///
/// Two layers:
///
/// 1. A **generic Arrow core** ([`append_arrow`](Warehouse::append_arrow),
/// [`scan_arrow`](Warehouse::scan_arrow),
/// [`scan_filtered`](Warehouse::scan_filtered),
/// [`scan_limited`](Warehouse::scan_limited)) plus table/catalog lifecycle
/// ([`ensure_table`](Warehouse::ensure_table),
/// [`ensure_columns`](Warehouse::ensure_columns),
/// [`table_names`](Warehouse::table_names)). This is what the sibling
/// modules (`dep_graph`, `release_events`, `test_results`, …) need: build a
/// `RecordBatch` from `iceberg_schema::*` and append / scan it by table
/// name, without reaching for the raw catalog. Backends only have to make
/// these honest and every typed method composes over them.
///
/// 2. **Named per-table convenience methods** ([`append_bench_run`] /
/// [`query_bench_runs`]) kept as thin wrappers so existing callers are
/// untouched. (P1 keeps just the two that were already on the trait; the
/// rest stay as inherent `IcebergWarehouse` methods until the next phase
/// migrates consumers onto the trait.)
///
/// `IcebergWarehouse` is the **default** (and today the only) implementation: it
/// delegates each method to its existing inherent body, which already wraps
/// skade (`skade::append` / `read_*` / `ingest_parallel`). A future
/// `SkadeWarehouse` (reusing a `skade::Warehouse` + `Table` handle) or a DuckDB
/// backend plugs in behind this same trait.
///
/// `iceberg_schema.rs` remains the per-table shape source of truth; the trait
/// is table-name + `RecordBatch` based and stays schema-agnostic.
///
/// ## Async surface (release orchestrator)
///
/// Alongside the sync core, the trait carries a small **async** surface
/// (`#[async_trait]`): the async Arrow twins (`append_arrow_async` /
/// `scan_arrow_async`) plus the named async writers/readers the release
/// subsystem drives (`append_bench_run_async`, `query_bench_runs_async`,
/// `append_git_heat_scan_async`). These exist
/// because the release pipeline (and the server's release-event handlers) are
/// **already inside a runtime** when they touch the warehouse: calling a sync
/// method there would re-enter `block_on` and panic (nested runtime). The async
/// twins do the real async work directly, so they compose from any runtime.
#[async_trait::async_trait]
pub trait Warehouse: Send + Sync {
// ── generic Arrow core ──────────────────────────────────────────────────
/// Append one Arrow [`RecordBatch`] to `table`. The batch must already match
/// the table's current `iceberg_schema::*` shape — `append_arrow` only recasts
/// to the stored schema and does NOT evolve it, so a batch carrying columns
/// the table lacks is an ERROR. Callers that need add-column evolution must
/// run [`ensure_columns`](Self::ensure_columns) first to migrate the schema,
/// then append. (No silent forward schema evolution here.)
fn append_arrow(&self, table: &str, batch: RecordBatch) -> Result<()>;
/// Full-snapshot scan of `table` → Arrow, column order preserved (so
/// positional downcasts stay valid).
fn scan_arrow(&self, table: &str) -> Result<Vec<RecordBatch>>;
/// Pushdown scan: prune to rows matching `filter` (file / row-group
/// granularity — keep a residual per-row guard for exactness) and project
/// `columns` (empty = all columns, order preserved). Returns an empty result
/// when the table does not exist yet.
fn scan_filtered(
&self,
table: &str,
filter: &ScanFilter,
columns: &[&str],
) -> Result<Vec<RecordBatch>>;
/// Limit / early-break scan: stop once `max_rows` rows are in hand instead
/// of materializing the whole table (`max_rows == 0` → full scan). Returns
/// *some* `max_rows` rows, not a deterministic top-N (no ordering guarantee).
fn scan_limited(&self, table: &str, max_rows: usize) -> Result<Vec<RecordBatch>>;
// ── table / catalog lifecycle ───────────────────────────────────────────
/// Create `table` if missing, partitioned by `partition_cols` (identity
/// transform; empty = unpartitioned). No-op when the table already exists.
fn ensure_table(
&self,
table: &str,
schema: ::iceberg::spec::Schema,
partition_cols: &[&str],
) -> Result<()>;
/// Evolve `table`'s stored schema to `canonical` (Iceberg add-column
/// migration) if it is missing any of `canonical`'s top-level columns.
/// No-op when the table already carries every canonical column.
fn ensure_columns(&self, table: &str, canonical: &::iceberg::spec::Schema) -> Result<()>;
/// Every table name in the warehouse namespace, sorted (catalog metadata
/// only, no data read).
fn table_names(&self) -> Result<Vec<String>>;
/// Stringified preview of `table` (≤ `limit` rows, `0` = backend default) —
/// the generic warehouse *browser* surface behind the `Warehouse.Scan` gRPC.
/// Backend-agnostic: the Iceberg backend reads Parquet and stringifies cells;
/// a `RemoteWarehouse` maps the `Warehouse.Scan` response 1:1. This is the
/// read seam that makes the warehouse browser (and the server's own
/// `Warehouse.Scan` handler) source-agnostic — an Iceberg tree or a remote
/// server both answer it.
fn scan_preview(&self, table: &str, limit: usize) -> Result<TablePreview>;
/// **TYPED vector read** — the one column shape [`scan_preview`](Self::scan_preview)
/// structurally cannot carry.
///
/// `scan_preview` stringifies every cell through one renderer whose catch-all is
/// `format!("<{}>", arr.data_type())`. A `list<float32>` matches no scalar arm, so
/// `embeddings.vector` comes back as the literal Arrow TYPE NAME — the same
/// constant for every row, carrying zero floats. Not truncated, not dropped:
/// replaced by something shaped like a cell. Any consumer that reads vectors off
/// the preview is therefore blind by construction, and the dupes SEMANTIC tier was.
///
/// This method is the honest route: vectors stay `f32`, end to end. `limit == 0`
/// means EVERY row — the semantic tier is an all-pairs sweep and a silently
/// truncated corpus would give a thin client a different verdict from a fat one,
/// which is the whole thing the seam exists to prevent. A backend that cannot
/// answer must return an ERROR (`RemoteWarehouse` returns its typed
/// `Unsupported`), never `Ok(VectorScan::default())`: "this transport cannot carry
/// vectors" and "the warehouse holds 0 embeddings" are different states and a
/// caller has to be able to tell them apart.
fn scan_vectors(
&self,
table: &str,
key_col: &str,
vec_col: &str,
limit: usize,
) -> Result<VectorScan>;
/// Describe `table`'s columns (name · Arrow logical type · nullability) — the
/// backend-neutral schema-*discovery* surface behind `nornir warehouse tables
/// <table>`. The Iceberg backend maps its stored schema to Arrow
/// (catalog-metadata only, no Parquet read); a `RemoteWarehouse` returns a
/// typed `Unsupported` — column *types* are lost on the stringified
/// `Warehouse.Scan` wire, so a typed describe is the Arrow Flight follow-up
/// (the same boundary as [`scan_filtered`](Self::scan_filtered) /
/// [`query_bench_runs`](Self::query_bench_runs)).
fn describe_columns(&self, table: &str) -> Result<Vec<ColumnInfo>>;
// ── named per-table convenience wrappers (thin) ─────────────────────────
fn append_bench_run(&self, repo: &str, run: &BenchRun) -> Result<Uuid>;
fn query_bench_runs(&self, filter: &BenchFilter) -> Result<Vec<BenchRun>>;
// ── async surface (safe to await from inside an existing runtime) ────────
/// Async twin of [`append_arrow`](Self::append_arrow). Awaited by the async
/// release writers (`persist_lineage`, `append_release_events`) where the
/// sync method's internal `block_on` would panic (nested runtime).
async fn append_arrow_async(&self, table: &str, batch: RecordBatch) -> Result<()>;
/// Async twin of [`scan_arrow`](Self::scan_arrow). Awaited by the async
/// release readers (`query_release_history`, `query_release_events`).
async fn scan_arrow_async(&self, table: &str) -> Result<Vec<RecordBatch>>;
/// Async twin of [`append_bench_run`](Self::append_bench_run) — driven by
/// `release::pipeline::run_pipeline` (itself `async`).
async fn append_bench_run_async(&self, repo: &str, run: &BenchRun) -> Result<Uuid>;
/// Async twin of [`query_bench_runs`](Self::query_bench_runs) — the gate's
/// bench-history read from inside `run_pipeline`.
async fn query_bench_runs_async(&self, filter: &BenchFilter) -> Result<Vec<BenchRun>>;
/// Persist a git-heat scan (async — `run_pipeline`'s knowledge-capture step).
async fn append_git_heat_scan_async(&self, scan: &GitHeatScan) -> Result<()>;
/// Concrete-backend escape hatch (generic Any downcast). Retained as a general
/// seam.
fn as_any(&self) -> &dyn std::any::Any;
}
#[derive(Debug, Default, Clone)]
pub struct BenchFilter {
pub repo: Option<String>,
pub machine: Option<String>,
pub limit: Option<usize>,
}
impl BenchFilter {
pub fn for_repo(repo: impl Into<String>) -> Self {
Self { repo: Some(repo.into()), machine: None, limit: None }
}
}
/// A stringified preview of one warehouse table for the egui browser.
///
/// **Scalars only, by construction.** Cells are rendered by one `cell_to_string` whose
/// catch-all prints the Arrow TYPE NAME, so a non-scalar column (notably
/// `embeddings.vector`, a `list<float32>`) is not previewable — it comes back as a
/// constant placeholder. Vectors go through [`Warehouse::scan_vectors`], which keeps
/// them `f32`.
#[derive(Debug, Clone, Default)]
pub struct TablePreview {
pub columns: Vec<String>,
pub rows: Vec<Vec<String>>,
}
/// One embedding read back TYPED through the seam.
#[derive(Debug, Clone, PartialEq)]
pub struct VectorRow {
/// The row's identity column (`embeddings.content_hash`).
pub key: String,
/// The vector itself — `f32`, never a rendered string.
pub vector: Vec<f32>,
/// The weights-SHA-sensitive embedding-space id (`embeddings.model_profile`).
/// Carried because a cosine measured in one vector space says NOTHING about
/// another; mixing spaces produces confident nonsense. Reported, not enforced.
pub model_profile: String,
}
/// The result of a [`Warehouse::scan_vectors`] read.
///
/// An EMPTY `vectors` here means one specific thing: the warehouse answered and holds
/// no embedding rows. "The transport cannot carry vectors" is an `Err`, never this.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct VectorScan {
pub vectors: Vec<VectorRow>,
/// Source rows read (rows whose vector was null/unreadable are counted here but
/// absent from `vectors`, so a partial read is visible rather than silent).
pub rows_scanned: u64,
/// A non-zero `limit` cut the scan short — the corpus is INCOMPLETE and any
/// all-pairs verdict computed over it is not comparable with a full read.
pub clipped: bool,
}
/// One column of a warehouse table's schema — the discovery companion to the SQL
/// `query` surface. A consumer that knows the table + column names (from
/// `list_tables` / `describe_table`) can then write `SELECT`s against them.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
pub struct ColumnInfo {
/// Column name (as it appears in SQL).
pub name: String,
/// Arrow logical type (`Utf8`, `Int64`, `Timestamp(Microsecond, …)`, …).
pub data_type: String,
/// Whether the column admits NULLs.
pub nullable: bool,
}
// ── knowledge MODEL PODs (the gix SCANNER stays in nornir; the symbol / call
// rows have no scanner since Mímir v2 — a row model only) ─────────────────
#[derive(Debug, Default)]
pub struct SymbolScan {
pub snapshot_id: Uuid,
pub ts: DateTime<Utc>,
pub repo: String,
pub symbols: Vec<SymbolRow>,
pub calls: Vec<CallEdgeRow>,
pub features: Vec<FeatureGateRow>,
/// Every `#[test]`/`#[tokio::test]`/`#[test_case]`/`#[rstest]` fn discovered
/// during the (BUILD-FREE) syn parse — the test INVENTORY. This is what lets
/// the 🧪 matrix show every test as `X` (not-run) the instant a workspace is
/// added, with zero `cargo build` (discovery is free; only EXECUTION needs a
/// build, and that stays gated to `nornir release` / `nornir test run`).
pub tests: Vec<TestDefRow>,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct SymbolRow {
pub crate_name: String,
pub module_path: String,
pub item_kind: String,
pub item_name: String,
pub visibility: String,
pub file: String,
pub line: u32,
pub doc_lines: u32,
pub signature: Option<String>,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct CallEdgeRow {
pub crate_name: String,
pub caller_path: String,
pub callee_ident: String,
pub call_kind: String,
pub file: String,
pub line: u32,
}
/// One discovered test function — the build-free test INVENTORY row. Emitted by
/// the syn walker for any `fn` carrying a test attribute (`#[test]`,
/// `#[tokio::test]`, `#[test_case(...)]`, `#[rstest]`, …). Heavy tests (the
/// `#[ignore]`-marked container/corpus/matrix suites) are flagged `is_heavy` so
/// the viz can tag them and the gating audit can keep them out of the poll loop.
#[derive(Debug, Clone, serde::Serialize)]
pub struct TestDefRow {
pub crate_name: String,
pub module_path: String,
pub test_name: String,
pub file: String,
pub line: u32,
/// `#[ignore]`-marked → the heavy container/corpus/matrix tests. These run
/// ONLY in `nornir release` (default) or on explicit `nornir test run`, never
/// on initial load or in the poll loop.
pub is_heavy: bool,
/// `async fn` and/or `#[tokio::test]` — informational, surfaced in the viz.
pub is_async: bool,
}
#[derive(Debug, Clone)]
pub struct FeatureGateRow {
pub crate_name: String,
pub module_path: String,
pub item_name: String,
pub cfg_expr: String,
pub file: String,
pub line: u32,
}
#[derive(Debug, Default)]
pub struct GitHeatScan {
pub snapshot_id: Uuid,
pub ts: DateTime<Utc>,
pub repo: String,
pub files: Vec<GitHeatRow>,
}
#[derive(Debug, Clone)]
pub struct GitHeatRow {
pub file: String,
pub commits_total: i64,
pub commits_30d: i64,
pub commits_90d: i64,
pub authors_total: i64,
pub last_commit_ts: DateTime<Utc>,
}
// ── RESOLVED (SCIP) knowledge PODs — the SCIP row model ─────────────────────
//
// Plain data: the moniker id scheme nornir keeps. The `rust-analyzer scip`
// INGEST that produced these rows is deleted (Mímir v2), and with it the
// trait's reader of them. nornir re-exports both at
// `crate::knowledge::scip::{ScipScan, ScipRow}` so its call sites are untouched.
/// One ingested SCIP occurrence — a single resolved defs/ref token.
#[derive(Debug, Clone, serde::Serialize, PartialEq, Eq)]
pub struct ScipRow {
/// The RESOLVED SCIP symbol moniker (globally unique). For a local symbol
/// it is `local N`; for a global it encodes scheme + package + descriptors.
pub symbol: String,
/// Decoded role label(s) — `definition`, `reference`, `import`, `write`,
/// `read`, `generated`, `test` — joined by `+` when several bits are set.
pub role: String,
/// `true` when the `Definition` role bit is set (the defining occurrence).
pub is_definition: bool,
/// Short display name from the matching `SymbolInformation`, when present.
pub display_name: String,
/// SCIP `SymbolInformation.Kind` (`Function`, `Method`, `Struct`, `Trait`,
/// …) when the symbol is defined in this index; `""` for refs to externals.
pub kind: String,
/// Document-relative source path (forward-slashed).
pub file: String,
/// 1-based line of the occurrence start (SCIP ranges are 0-based).
pub start_line: u32,
/// 1-based column of the occurrence start.
pub start_col: u32,
/// 1-based START line of this occurrence's `enclosing_range` — the span of
/// the whole defining item (fn/method body). Only meaningful for DEFINITION
/// occurrences of a Function/Method (else `0`). This is what lets us decide
/// WHICH function CONTAINS a reference (i.e. who the caller is).
pub enc_start_line: u32,
/// 1-based END line of the `enclosing_range` (inclusive in the containment
/// test); `0` when absent.
pub enc_end_line: u32,
}
/// A whole ingested SCIP index, ready to persist / read back.
#[derive(Debug, Default)]
pub struct ScipScan {
pub snapshot_id: Uuid,
pub ts: DateTime<Utc>,
pub repo: String,
pub git_sha: String,
pub rows: Vec<ScipRow>,
}
/// In-memory queries over an ingested SCIP scan — the RESOLVED find-usages
/// surface. Because every reference carries the *same* resolved `symbol` moniker
/// as its definition, find-usages is an exact symbol match (no last-segment name
/// heuristic, no cross-name collisions). Pure (std-only), so they live with the
/// POD in the leaf; nornir's `crate::knowledge::scip` re-exports the type + these
/// methods verbatim.
impl ScipScan {
/// The defining occurrence(s) of every symbol whose display name or symbol
/// moniker contains `pattern` (case-insensitive).
pub fn definitions_matching<'a>(&'a self, pattern: &str) -> Vec<&'a ScipRow> {
let p = pattern.to_lowercase();
self.rows
.iter()
.filter(|r| r.is_definition)
.filter(|r| {
r.display_name.to_lowercase().contains(&p)
|| r.symbol.to_lowercase().contains(&p)
})
.collect()
}
/// Resolve `pattern` to the unique symbol moniker(s) it *defines*. Used to
/// turn a human name into the precise key for [`Self::usages_of`].
pub fn resolve_symbols(&self, pattern: &str) -> Vec<String> {
let mut syms: Vec<String> =
self.definitions_matching(pattern).iter().map(|r| r.symbol.clone()).collect();
syms.sort();
syms.dedup();
syms
}
/// Every *reference* occurrence of an exact resolved `symbol` — IDE-grade
/// find-usages. Excludes the definition itself.
pub fn usages_of<'a>(&'a self, symbol: &str) -> Vec<&'a ScipRow> {
self.rows
.iter()
.filter(|r| r.symbol == symbol && !r.is_definition)
.collect()
}
/// All occurrences (defs + refs) of an exact resolved `symbol`.
pub fn occurrences_of<'a>(&'a self, symbol: &str) -> Vec<&'a ScipRow> {
self.rows.iter().filter(|r| r.symbol == symbol).collect()
}
}
// ─── the ONE Arrow → f32 vector extractor ────────────────────────────────
//
// Every backend that stores vectors stores them the same way (`list<float32>`, or a
// `fixed_size_list<float32>` after a schema evolution), so the read is written ONCE
// here rather than per backend. It is also the only place that knows what a
// *missing* vector means: skipped, and visible in `rows_scanned` vs `vectors.len()`.
/// Pull `(key_col, vec_col)` out of Arrow batches as typed [`VectorRow`]s.
///
/// Accepts `list<float32>` and `fixed_size_list<float32>` (the two shapes an evolved
/// Iceberg vector column can take) and, so a caller is never silently handed half a
/// table, counts EVERY source row in [`VectorScan::rows_scanned`] while only rows with
/// a readable vector land in `vectors`.
///
/// `limit == 0` = every row. A non-zero limit that actually bites sets
/// [`VectorScan::clipped`]; an all-pairs consumer must treat that as INCOMPLETE rather
/// than as a smaller warehouse.
///
/// `model_profile_col` is optional (`""` = don't read it); a missing column yields an
/// empty profile string rather than an error, so an older table still reads.
pub fn vectors_from_batches(
batches: &[RecordBatch],
key_col: &str,
vec_col: &str,
model_profile_col: &str,
limit: usize,
) -> Result<VectorScan> {
use arrow::array::{Array, AsArray};
use arrow::datatypes::{DataType, Float32Type};
let mut out = VectorScan::default();
for b in batches {
let schema = b.schema();
let Ok(kf) = schema.index_of(key_col) else {
anyhow::bail!(
"vector scan: no `{key_col}` column (have: {})",
schema.fields().iter().map(|f| f.name().as_str()).collect::<Vec<_>>().join(", ")
);
};
let Ok(vf) = schema.index_of(vec_col) else {
anyhow::bail!(
"vector scan: no `{vec_col}` column (have: {})",
schema.fields().iter().map(|f| f.name().as_str()).collect::<Vec<_>>().join(", ")
);
};
let pf = (!model_profile_col.is_empty())
.then(|| schema.index_of(model_profile_col).ok())
.flatten();
let keys = b.column(kf);
let vecs = b.column(vf);
for i in 0..b.num_rows() {
if limit > 0 && out.vectors.len() >= limit {
out.clipped = true;
return Ok(out);
}
out.rows_scanned += 1;
let key = match keys.data_type() {
DataType::Utf8 => keys.as_string::<i32>().value(i).to_string(),
DataType::LargeUtf8 => keys.as_string::<i64>().value(i).to_string(),
_ => continue,
};
if key.is_empty() || vecs.is_null(i) {
continue;
}
let values = match vecs.data_type() {
DataType::List(_) => vecs.as_list::<i32>().value(i),
DataType::LargeList(_) => vecs.as_list::<i64>().value(i),
DataType::FixedSizeList(_, _) => vecs.as_fixed_size_list().value(i),
_ => continue,
};
if values.data_type() != &DataType::Float32 {
continue;
}
let f = values.as_primitive::<Float32Type>();
let vector: Vec<f32> = (0..f.len()).map(|j| f.value(j)).collect();
if vector.is_empty() {
continue;
}
let model_profile = pf
.filter(|_| matches!(b.column(pf.unwrap()).data_type(), DataType::Utf8))
.map(|p| b.column(p).as_string::<i32>().value(i).to_string())
.unwrap_or_default();
out.vectors.push(VectorRow { key, vector, model_profile });
}
}
Ok(out)
}
#[cfg(test)]
mod vector_extract_tests {
use super::*;
use arrow::array::{ArrayRef, Float32Builder, ListBuilder, StringArray};
use arrow::datatypes::{DataType, Field, Schema as ArrowSchema};
use std::sync::Arc;
fn batch(rows: &[(&str, Option<Vec<f32>>)]) -> RecordBatch {
let keys: ArrayRef =
Arc::new(StringArray::from(rows.iter().map(|(k, _)| *k).collect::<Vec<_>>()));
let mut lb = ListBuilder::new(Float32Builder::new());
for (_, v) in rows {
match v {
Some(v) => {
for f in v {
lb.values().append_value(*f);
}
lb.append(true);
}
None => lb.append(false),
}
}
let vecs: ArrayRef = Arc::new(lb.finish());
let schema = ArrowSchema::new(vec![
Field::new("content_hash", DataType::Utf8, false),
Field::new("vector", vecs.data_type().clone(), true),
]);
RecordBatch::try_new(Arc::new(schema), vec![keys, vecs]).unwrap()
}
/// The floats must arrive BIT-IDENTICAL. A `list<float32>` rendered through a
/// stringified preview loses them entirely (the preview's catch-all prints the
/// Arrow type name), which is the defect this path exists to close — so the
/// assertion is on the values, not on "we got a row".
#[test]
fn list_float32_arrives_as_exact_f32() {
let b = batch(&[("aaa", Some(vec![0.1, -0.25, 3.5])), ("bbb", Some(vec![1.0, 2.0, 4.0]))]);
let scan = vectors_from_batches(&[b], "content_hash", "vector", "", 0).unwrap();
assert_eq!(scan.rows_scanned, 2);
assert_eq!(scan.vectors.len(), 2);
assert_eq!(scan.vectors[0].key, "aaa");
assert_eq!(scan.vectors[0].vector, vec![0.1_f32, -0.25, 3.5]);
assert_eq!(scan.vectors[1].vector, vec![1.0_f32, 2.0, 4.0]);
assert!(!scan.clipped);
}
/// A null vector is SKIPPED but still counted, so `rows_scanned > vectors.len()`
/// makes a partially-embedded table visible instead of looking like a small one.
#[test]
fn a_null_vector_is_skipped_and_still_counted() {
let b = batch(&[("aaa", Some(vec![1.0, 2.0])), ("bbb", None)]);
let scan = vectors_from_batches(&[b], "content_hash", "vector", "", 0).unwrap();
assert_eq!(scan.rows_scanned, 2, "both rows were read");
assert_eq!(scan.vectors.len(), 1, "only one carried a vector");
}
/// A biting limit must SAY it bit. An all-pairs consumer given a silently clipped
/// corpus reports a different verdict from one given the whole table.
#[test]
fn a_biting_limit_reports_clipped() {
let b = batch(&[("a", Some(vec![1.0, 2.0])), ("b", Some(vec![3.0, 4.0]))]);
let scan = vectors_from_batches(&[b.clone()], "content_hash", "vector", "", 1).unwrap();
assert!(scan.clipped, "limit 1 over 2 rows is clipped");
assert_eq!(scan.vectors.len(), 1);
let full = vectors_from_batches(&[b], "content_hash", "vector", "", 0).unwrap();
assert!(!full.clipped, "limit 0 = every row, never clipped");
assert_eq!(full.vectors.len(), 2);
}
/// A wrong column name is an ERROR naming what IS there — never an empty `Ok`,
/// which would read exactly like an unembedded warehouse.
#[test]
fn a_missing_column_is_an_error_not_an_empty_ok() {
let b = batch(&[("a", Some(vec![1.0, 2.0]))]);
let err = vectors_from_batches(&[b], "content_hash", "embedding", "", 0).unwrap_err();
let msg = format!("{err}");
assert!(msg.contains("no `embedding` column"), "{msg}");
assert!(msg.contains("vector"), "the error names the columns that DO exist: {msg}");
}
}