src/warehouse/build_info.rs
//! Iceberg writer + reader for **build-info** (`build_info`) — the exploded
//! columnar record ABOUT a build (build-info flight-stream design §f3/§f6).
//!
//! Where [`super::build_provenance`] holds one flat row per recorded build, this
//! holds the per-fact rows: the Artifactory-parity **manifest** (`provenance` /
//! `dep` / `artifact`, derived from a [`BuildProvenance`] via
//! [`nornir_build_thing::build_info::manifest_rows`]) plus the live **feed**
//! (`log` / `metric` / `step`). The row type is reused verbatim from
//! `nornir-build-thing` (`BuildInfoRow`) — the same struct the dep-light emitter
//! and the at-rest znippy seal use, so the live wire, the sealed blob, and this
//! warehouse table are one shape (design invariant: "same schema at rest as
//! live — no second format").
//!
//! This is the NATIVE landing for a dwarves build's build-info: previously it
//! rode `test_results` as an interim; now it lands in its own table + surfaces in
//! the viz warehouse deck like any other table.
use std::sync::Arc;
use anyhow::{Result, anyhow};
use arrow::array::{Array, Int64Array, LargeBinaryArray, RecordBatch, StringArray};
use iceberg::Catalog;
use iceberg::arrow::schema_to_arrow_schema;
use nornir_build_thing::build_info::BuildInfoRow;
use super::iceberg::{IcebergWarehouse, TABLE_BUILD_INFO, append_batch};
// Column indices match `iceberg_schema::build_info` field order (§f3).
const COL_BUILD_ID: usize = 0;
const COL_STEP_ID: usize = 1;
const COL_TS_MICROS: usize = 2;
const COL_KIND: usize = 3;
const COL_LEVEL: usize = 4;
const COL_NAME: usize = 5;
const COL_VALUE: usize = 6;
const COL_CONTENT_ID: usize = 7;
const COL_BLOB: usize = 8;
/// The §f3 Arrow schema (columns in declared order, `blob` the only nullable
/// column) — the hand-built twin of `iceberg_schema::build_info()` for callers
/// that need an `arrow` schema WITHOUT an open warehouse (the at-rest seal). The
/// two MUST stay column-identical; the warehouse round-trip test proves the
/// iceberg-derived batch matches, and this shares the same column order.
pub fn build_info_arrow_schema() -> Arc<arrow::datatypes::Schema> {
use arrow::datatypes::{DataType, Field, Schema};
Arc::new(Schema::new(vec![
Field::new("build_id", DataType::Utf8, false),
Field::new("step_id", DataType::Utf8, false),
Field::new("ts_micros", DataType::Int64, false),
Field::new("kind", DataType::Utf8, false),
Field::new("level", DataType::Utf8, false),
Field::new("name", DataType::Utf8, false),
Field::new("value", DataType::Utf8, false),
Field::new("content_id", DataType::Utf8, false),
// iceberg's `Binary` maps to Arrow `LargeBinary` (via `schema_to_arrow_schema`),
// so the seal batch MUST match the warehouse batch — use LargeBinary here too.
Field::new("blob", DataType::LargeBinary, true),
]))
}
/// Build one Arrow [`RecordBatch`] over `rows`, matching the `build_info` table
/// schema (§f3 columns, `blob` the only nullable column). Reused by both the
/// warehouse append and the at-rest znippy seal so there is ONE columnizer.
pub fn build_info_record_batch(
arrow_schema: Arc<arrow::datatypes::Schema>,
rows: &[BuildInfoRow],
) -> Result<RecordBatch> {
let cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from_iter_values(
rows.iter().map(|r| r.build_id.as_str()),
)),
Arc::new(StringArray::from_iter_values(
rows.iter().map(|r| r.step_id.as_str()),
)),
Arc::new(Int64Array::from_iter_values(
rows.iter().map(|r| r.ts_micros),
)),
Arc::new(StringArray::from_iter_values(
rows.iter().map(|r| r.kind.as_str()),
)),
Arc::new(StringArray::from_iter_values(
rows.iter().map(|r| r.level.as_str()),
)),
Arc::new(StringArray::from_iter_values(
rows.iter().map(|r| r.name.as_str()),
)),
Arc::new(StringArray::from_iter_values(
rows.iter().map(|r| r.value.as_str()),
)),
Arc::new(StringArray::from_iter_values(
rows.iter().map(|r| r.content_id.as_str()),
)),
// The nullable column: None ⇒ SQL NULL, distinct from an empty blob.
// `LargeBinary` (i64 offsets) matches iceberg's `Binary` arrow mapping.
Arc::new(LargeBinaryArray::from_iter(
rows.iter().map(|r| r.blob.as_deref()),
)),
];
Ok(RecordBatch::try_new(arrow_schema, cols)?)
}
/// Append build-info rows (one Iceberg snapshot). No-op on an empty slice.
pub async fn append_build_info(wh: &IcebergWarehouse, rows: &[BuildInfoRow]) -> Result<()> {
if rows.is_empty() {
return Ok(());
}
let table = wh
.catalog()
.load_table(&wh.table_ident(TABLE_BUILD_INFO))
.await?;
let arrow_schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
let batch = build_info_record_batch(arrow_schema, rows)?;
append_batch(wh.catalog(), table, batch).await?;
Ok(())
}
/// Read build-info rows, optionally scoped to one `build_id`, sorted by
/// `(build_id, ts_micros, kind)` so a build's manifest reads back in a stable
/// order (Iceberg gives no scan order).
pub async fn query_build_info(
wh: &IcebergWarehouse,
build_id: Option<&str>,
) -> Result<Vec<BuildInfoRow>> {
let batches: Vec<RecordBatch> = super::iceberg::load_and_read_all(wh, TABLE_BUILD_INFO).await?;
let mut out: Vec<BuildInfoRow> = Vec::new();
for b in &batches {
let build_id_c = col_str(b, COL_BUILD_ID)?;
let step_id = col_str(b, COL_STEP_ID)?;
let ts = col_i64(b, COL_TS_MICROS)?;
let kind = col_str(b, COL_KIND)?;
let level = col_str(b, COL_LEVEL)?;
let name = col_str(b, COL_NAME)?;
let value = col_str(b, COL_VALUE)?;
let content_id = col_str(b, COL_CONTENT_ID)?;
let blob = col_bin(b, COL_BLOB)?;
for i in 0..b.num_rows() {
let row = BuildInfoRow {
build_id: build_id_c.value(i).to_string(),
step_id: step_id.value(i).to_string(),
ts_micros: ts.value(i),
kind: kind.value(i).to_string(),
level: level.value(i).to_string(),
name: name.value(i).to_string(),
value: value.value(i).to_string(),
content_id: content_id.value(i).to_string(),
blob: (!blob.is_null(i)).then(|| blob.value(i).to_vec()),
};
if build_id.map(|w| row.build_id == w).unwrap_or(true) {
out.push(row);
}
}
}
out.sort_by(|a, b| {
a.build_id
.cmp(&b.build_id)
.then_with(|| a.ts_micros.cmp(&b.ts_micros))
.then_with(|| a.kind.cmp(&b.kind))
.then_with(|| a.name.cmp(&b.name))
});
Ok(out)
}
// ─── column helpers ─────────────────────────────────────────────────────
fn col_str<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a StringArray> {
b.column(idx)
.as_any()
.downcast_ref::<StringArray>()
.ok_or_else(|| anyhow!("build_info col {idx} is not StringArray"))
}
fn col_i64<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a Int64Array> {
b.column(idx)
.as_any()
.downcast_ref::<Int64Array>()
.ok_or_else(|| anyhow!("build_info col {idx} is not Int64Array"))
}
fn col_bin<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a LargeBinaryArray> {
b.column(idx)
.as_any()
.downcast_ref::<LargeBinaryArray>()
.ok_or_else(|| anyhow!("build_info col {idx} is not LargeBinaryArray"))
}
#[cfg(test)]
mod tests {
use super::*;
use nornir_build_thing::build_info::{BuildInfoKind, manifest_rows};
use nornir_build_thing::provenance::{BuildProvenance, HostInfo, Outcome, Toolchain};
use nornir_build_thing::variant::{BuildVariant, Profile};
fn sample_prov() -> (tempfile::TempDir, BuildProvenance) {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let target = root.join("target/release");
std::fs::create_dir_all(&target).unwrap();
let art = target.join("dwarves-full");
std::fs::write(&art, b"ELF dwarves-full bytes").unwrap();
let v = BuildVariant {
name: "full".into(),
bin: Some("dwarves-full".into()),
features: vec!["full".into(), "cli".into()],
profile: Profile::Release,
..Default::default()
};
let rec = BuildProvenance::capture(
&v,
"dwarves",
root,
Toolchain {
rustc: "rustc 1.85.0".into(),
host: "x86_64".into(),
cargo: "cargo 1.85.0".into(),
// The constellation's pinned build image, as a real record carries it.
image: Some("nordisk-cargo-toolchain".into()),
image_sha256: Some(
"3d5115baa07d03ca51eb3e44ebb86b4e64e679bd88f38e156ef9da12c35d2143".into(),
),
},
HostInfo {
cores: 12,
memory_bytes: 64 * 1024 * 1024 * 1024,
},
Outcome::Success,
&[art],
)
.unwrap();
(dir, rec)
}
#[test]
fn append_and_query_round_trips_manifest_rows() {
let dir = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(dir.path()).unwrap();
let (_d, rec) = sample_prov();
// Manifest rows (provenance/dep/artifact) + a live-feed log row with a blob.
let mut rows = manifest_rows(&rec, 100);
rows.push(
BuildInfoRow::new(&rec.id, 101, BuildInfoKind::Log)
.step("cargo-build")
.level("info")
.blob(b"compiling dwarves".to_vec()),
);
wh.block_on(append_build_info(&wh, &rows)).unwrap();
let got = wh.block_on(query_build_info(&wh, Some(&rec.id))).unwrap();
assert_eq!(got.len(), rows.len(), "every build-info row landed");
// RED-when-broken: the artifact CAS id survives.
let art = got
.iter()
.find(|r| r.kind == "artifact")
.expect("artifact row");
assert_eq!(art.content_id, rec.artifacts[0].sha256);
assert_eq!(art.blob, None, "manifest rows carry no blob");
// dep rows = enabled features; provenance repo fact landed.
let deps: std::collections::BTreeSet<&str> = got
.iter()
.filter(|r| r.kind == "dep")
.map(|r| r.value.as_str())
.collect();
assert_eq!(deps, ["cli", "full"].into_iter().collect());
assert!(
got.iter()
.any(|r| r.kind == "provenance" && r.name == "repo" && r.value == "dwarves")
);
// The nullable blob round-trips distinct from None.
let log = got.iter().find(|r| r.kind == "log").expect("log row");
assert_eq!(log.blob.as_deref(), Some(b"compiling dwarves".as_slice()));
}
#[test]
fn query_scopes_to_one_build_id() {
let dir = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(dir.path()).unwrap();
wh.block_on(append_build_info(
&wh,
&[
BuildInfoRow::new("build-a", 1, BuildInfoKind::Provenance).kv("repo", "a"),
BuildInfoRow::new("build-b", 2, BuildInfoKind::Provenance).kv("repo", "b"),
],
))
.unwrap();
let a = wh.block_on(query_build_info(&wh, Some("build-a"))).unwrap();
assert_eq!(a.len(), 1);
assert_eq!(a[0].value, "a");
let all = wh.block_on(query_build_info(&wh, None)).unwrap();
assert_eq!(all.len(), 2, "None reads all builds");
}
}