nornir 0.5.4

Companion to cargo: dependency tracking, release gating, deploy, benchmarks, and documentation assembly. Project-agnostic.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252

//! 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::{anyhow, Result};
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::{append_batch, IcebergWarehouse, TABLE_BUILD_INFO};

// 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::{manifest_rows, BuildInfoKind};
    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() },
            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");
    }
}