crates/mimir-driver/tests/fixture-tables/wh/src/lib.rs
//! The warehouse API of the fixture — the shapes nornir's own has: a
//! warehouse struct that turns a NAME into the engine's ident, read / write
//! helpers that take the name or a loaded table, a backend trait, and a
//! typed handle that wraps the engine's table. NOTHING here is listed in the
//! driver: every fn's part in a table access is derived from what it passes
//! on to the engine.
use iceberg::{Catalog, MemoryCatalog, NamespaceIdent, Table, TableCreation, TableIdent, Transaction};
pub mod tables {
pub const TABLE_RUNS: &str = "runs";
pub const TABLE_EVENTS: &str = "events";
pub const TABLE_COVERAGE: &str = "coverage";
pub static TABLE_AUDIT: &str = "audit";
/// A const LIST of names: a loop over it touches every one.
pub const SWEPT: [&str; 2] = ["sweep_a", "sweep_b"];
/// A name BUILT at compile time is still a constant.
pub const TABLE_BUILT: &str = concat!("built", "_", "name");
}
pub type Result<T> = std::result::Result<T, iceberg::Error>;
pub struct Warehouse {
catalog: MemoryCatalog,
namespace: NamespaceIdent,
}
impl Warehouse {
pub fn open() -> Self {
Self { catalog: MemoryCatalog, namespace: NamespaceIdent(vec!["nornir".into()]) }
}
pub fn catalog(&self) -> &MemoryCatalog {
&self.catalog
}
/// name → ident: a conduit (no access of its own).
pub fn table_ident(&self, name: &str) -> TableIdent {
TableIdent::new(self.namespace.clone(), name.to_string())
}
/// The generic browser: reads whatever table it is given.
pub fn scan_preview(&self, table: &str, limit: usize) -> Result<Vec<String>> {
let t = self.catalog.load_table(&self.table_ident(table))?;
Ok(t.scan().rows().into_iter().take(limit).collect())
}
/// Runtime names: every table the catalog lists.
pub fn count_all(&self) -> Result<usize> {
let mut n = 0;
for ident in self.catalog.list_tables(&self.namespace)? {
n += self.catalog.load_table(&ident)?.scan().rows().len();
}
Ok(n)
}
/// An ACCESSOR with the table baked in: the edge is here, on the fn that
/// spells the name — its callers reach the table through CALLS. One arm
/// of the `match` leaves the fn: it stands for no table.
pub fn append_events(&self, rows: &[String]) -> Result<()> {
let t = match self.catalog.load_table(&self.table_ident(tables::TABLE_EVENTS)) {
Ok(t) => t,
Err(_) => return Ok(()),
};
append_batch(&self.catalog, t, rows)
}
/// A loop over a list of `(name, something)` pairs: only the names count.
pub fn ensure_layout(&self) -> Result<()> {
for (name, _format) in [(tables::TABLE_RUNS, 1u32), (tables::TABLE_EVENTS, 2)] {
create_table_if_missing(self, name)?;
}
Ok(())
}
/// The table travels inside `Some((table, rows))`: the rows beside it say
/// nothing about which table.
pub fn append_staged(&self, rows: &[String]) -> Result<()> {
let staged = {
let table = self.catalog.load_table(&self.table_ident(tables::TABLE_COVERAGE))?;
if rows.is_empty() { None } else { Some((table, rows.to_vec())) }
};
if let Some((table, rows)) = staged {
append_batch(&self.catalog, table, &rows)?;
}
Ok(())
}
pub fn drop_tables(&self, names: &[&str]) -> Result<()> {
for n in names {
self.catalog.drop_table(&self.table_ident(n))?;
}
Ok(())
}
}
/// READ helper taking the NAME (nornir's `load_and_read_all`).
pub fn load_and_read_all(wh: &Warehouse, table: &str) -> Result<Vec<String>> {
let t = wh.catalog().load_table(&wh.table_ident(table))?;
Ok(read_all(&t))
}
/// READ helper taking the loaded table (skade's `read_all`).
pub fn read_all(table: &Table) -> Vec<String> {
table.scan().rows()
}
/// WRITE helper taking the loaded table (nornir's `append_batch`).
pub fn append_batch(catalog: &dyn Catalog, table: Table, rows: &[String]) -> Result<()> {
Transaction::new(&table).fast_append(rows).commit(catalog)?;
Ok(())
}
/// Create-if-missing: a WRITE (DDL) on the table the name stands for.
pub fn create_table_if_missing(wh: &Warehouse, name: &str) -> Result<Table> {
let creation = TableCreation::builder().name(name.to_string()).build();
wh.catalog().create_table(&NamespaceIdent(vec!["nornir".into()]), creation)
}
/// The async forms (`.await`, `?`, an `async` block).
pub async fn load_async(wh: &Warehouse, table: &str) -> Result<Table> {
let ident = wh.table_ident(table);
let t = async { wh.catalog().load_table(&ident) }.await?;
Ok(t)
}
pub async fn read_async(wh: &Warehouse, table: &str) -> Result<Vec<String>> {
let t = load_async(wh, table).await?;
Ok(read_all(&t))
}
/// The backend seam (nornir's `Warehouse` trait): callers hold `&dyn Store`.
pub trait Store {
fn scan(&self, table: &str) -> Result<Vec<String>>;
fn append(&self, table: &str, rows: &[String]) -> Result<()>;
/// No impl touches a table through this one.
fn label(&self, text: &str) -> String;
}
impl Store for Warehouse {
fn scan(&self, table: &str) -> Result<Vec<String>> {
load_and_read_all(self, table)
}
fn append(&self, table: &str, rows: &[String]) -> Result<()> {
let ident = self.table_ident(table);
let t = self.catalog.load_table(&ident)?;
append_batch(&self.catalog, t, rows)
}
fn label(&self, text: &str) -> String {
format!("wh:{text}")
}
}
/// A TYPED handle: a wrapper struct around the engine's table.
pub struct Handle {
inner: Table,
pub rows_written: usize,
}
impl Handle {
pub fn open(wh: &Warehouse, name: &str) -> Result<Handle> {
let inner = wh.catalog().load_table(&wh.table_ident(name))?;
Ok(Handle { inner, rows_written: 0 })
}
pub fn append(&mut self, catalog: &dyn Catalog, rows: &[String]) -> Result<()> {
self.rows_written += rows.len();
append_batch(catalog, self.inner.clone(), rows)
}
pub fn read(&self) -> Vec<String> {
read_all(&self.inner)
}
/// Metadata only.
pub fn location(&self) -> String {
self.inner.location()
}
}
/// A fn that takes a string that is NOT a table.
pub fn log_line(text: &str) -> usize {
text.len()
}
pub type Ident = TableIdent;
pub type EngineTable = Table;
/// An extension trait over the engine's catalog (a sibling's own wrapper).
pub trait CatalogExt {
fn load(&self, ident: &TableIdent) -> Result<Table>;
}
impl CatalogExt for MemoryCatalog {
fn load(&self, ident: &TableIdent) -> Result<Table> {
self.load_table(ident)
}
}