xtask/examples/nornir-bench.rs
//! Release-gate bench runner for the znippy workspace.
//!
//! Invoked by `nornir bench run znippy` (or `nornir release`) which
//! subprocesses `cargo run -p xtask --release --example nornir-bench`,
//! parses stdout as one `nornir::bench::BenchRun` line, and writes
//! into the iceberg `bench_runs` table.
//!
//! ## Strict serial pipeline
//!
//! The osm-katana pipeline is wired stage-by-stage, **never in
//! parallel** — every stage feeds the next stage's input. Order is
//! enforced by `register_bench_ordered!(n, …)`; nornir's
//! `run_main_json` sorts by `n` before executing and (with
//! `NORNIR_BENCH_STOP_ON_ERROR=1`) aborts the remainder when an
//! upstream stage fails.
//!
//! stage 0: planet-260427.osm.bz2 → planet-260427.osm.pbf
//! via `osm_katana::xml_to_pbf::xml_to_pbf_bz2_path`
//!
//! stage 1: planet-260427.osm.pbf → planet-260427.geoparquet/
//! via `osm_katana::convert`
//!
//! Fixtures + intermediate artifacts live on local disk only — never
//! on T9 (USB external would skew throughput). See workspace_holger
//! WORKSPACE.md for the canonical paths.
use std::path::{Path, PathBuf};
use std::time::Instant;
use anyhow::{Context, Result, anyhow};
use nornir::bench::{BenchResult, BenchSource};
use nornir::bench::api::Bencher;
use serde_json::json;
const PLANET_BZ2: &str = "/home/rickard/work/planet-260427.osm.bz2";
const PLANET_PBF: &str = "/home/rickard/work/planet-260427.osm.pbf";
const PLANET_GEO: &str = "/home/rickard/work/planet-260427.geoparquet";
// Extended planet path-matrix outputs (all on /home — never T9). Each ordered
// stage below records the convert's own per-phase telemetry (pass1 /
// sort_and_stree / pass2 cpu_cores) so the GATLING-saturation fix can be proven
// at planet scale.
const PLANET_GEO_DIRECT: &str = "/home/rickard/work/osm/planet-geo-direct";
const PLANET_ARROW: &str = "/home/rickard/work/osm/planet-nodes.arrow";
const PLANET_EUROPE_PBF: &str = "/home/rickard/work/osm/planet-europe-clip.osm.pbf";
const PLANET_EUROPE_GEO: &str = "/home/rickard/work/osm/planet-europe-clip-geo";
// Europe-ish bbox (W,S,E,N) for the osmium ±clip arm. Generous box covering the
// European mainland + British Isles + Nordics; `complete_ways` keeps boundary
// geometry intact.
const EUROPE_BBOX: &str = "-25.0,34.0,45.0,72.0";
fn require_local(p: &Path, label: &str) -> Result<()> {
if p.starts_with("/run/media/rickard/T9") {
return Err(anyhow!(
"{label} ({}) — bench fixtures must live on local disk; copy first",
p.display()
));
}
Ok(())
}
fn workers() -> usize {
std::thread::available_parallelism()
.map(|n| n.get().saturating_sub(1).max(1))
.unwrap_or(8)
}
/// Machine core count, recorded on every archive/stream result so a row in the
/// warehouse is self-describing (P2: cores + file count in the bench output).
/// Under `taskset -c 0` this is 1, which is how `_st` single-core rows arise.
fn cores() -> usize {
std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1)
}
/// Count how many on-disk chunks were stored raw vs codec-compressed by reading
/// the archive's Arrow-IPC index `compressed` column. Surfaces the
/// store-if-incompressible behaviour: incompressible corpora should report
/// `raw_chunks > 0` and never inflate the archive past the input.
fn chunk_kinds(archive: &Path) -> (u64, u64) {
use znippy_common::arrow::array::BooleanArray;
let (_schema, batches) = match znippy_common::read_znippy_index(archive) {
Ok(v) => v,
Err(_) => return (0, 0),
};
let (mut raw, mut comp) = (0u64, 0u64);
for batch in &batches {
if let Some(col) = batch.column_by_name("compressed") {
if let Some(b) = col.as_any().downcast_ref::<BooleanArray>() {
for i in 0..b.len() {
if b.value(i) { comp += 1; } else { raw += 1; }
}
}
}
}
(raw, comp)
}
fn walk_size(dir: &Path) -> std::io::Result<u64> {
let mut total = 0u64;
for entry in std::fs::read_dir(dir)? {
let entry = entry?;
let md = entry.metadata()?;
total += if md.is_dir() { walk_size(&entry.path())? } else { md.len() };
}
Ok(total)
}
// ── per-phase cpu_cores telemetry parse (saturation-fix proof) ────────────────
//
// `osm_katana::convert` (with `log_path = Some(..)`) appends JSONL phase events
// to the log: `{"phase":"pass1"/"sort_and_stree"/"pass2","ev":"progress"/"done",
// "cpu_cores":..,"busy_pct":..}`. We fold each phase's `cpu_cores` samples into
// `max` + `avg` (over `progress` + `done` events) and write them into the bench
// metrics as `<phase>_cores_max` / `<phase>_cores_avg`. The fix is proven when
// `pass1` and `pass2` sustain ~28-31 of 32 cores; `sort_and_stree` is I/O-bound
// (low cores expected, recorded but not a failure).
#[derive(Default, Clone, Copy)]
struct PhaseCores { max: f64, sum: f64, n: u64 }
impl PhaseCores {
fn push(&mut self, c: f64) { if c > self.max { self.max = c; } self.sum += c; self.n += 1; }
fn avg(&self) -> f64 { if self.n > 0 { self.sum / self.n as f64 } else { 0.0 } }
}
/// Parse a convert JSONL telemetry log into per-phase cpu_cores stats and inject
/// `<phase>_cores_max` / `<phase>_cores_avg` (rounded to 2 dp) into `m`. Robust
/// to interleaved `convert`-phase rollup lines (which we skip) and non-JSON
/// progress lines from the ext-sort helper.
fn fold_phase_cores(log: &Path, m: &mut serde_json::Map<String, serde_json::Value>) {
let text = match std::fs::read_to_string(log) { Ok(t) => t, Err(_) => return };
let mut phases: std::collections::BTreeMap<String, PhaseCores> = std::collections::BTreeMap::new();
for line in text.lines() {
let line = line.trim();
if !line.starts_with('{') { continue; }
let v: serde_json::Value = match serde_json::from_str(line) { Ok(v) => v, Err(_) => continue };
let phase = match v.get("phase").and_then(|p| p.as_str()) { Some(p) => p, None => continue };
// The umbrella "convert" phase only mirrors the active sub-phase's cores;
// skip it so each phase's stats stay clean (pass1 / sort_and_stree / pass2).
if phase == "convert" { continue; }
let cores = match v.get("cpu_cores").and_then(|c| c.as_f64()) { Some(c) => c, None => continue };
phases.entry(phase.to_string()).or_default().push(cores);
}
let round2 = |f: f64| (f * 100.0).round() / 100.0;
for (phase, pc) in &phases {
m.insert(format!("{phase}_cores_max"), json!(round2(pc.max)));
m.insert(format!("{phase}_cores_avg"), json!(round2(pc.avg())));
m.insert(format!("{phase}_samples"), json!(pc.n));
}
}
// ──────────────────────────── stage 0: bz2 → pbf ────────────────────────────
/// Skip the 45-minute planet pipeline for focused decompressor/archive runs.
fn skip_planet() -> bool {
std::env::var("NORNIR_BENCH_SKIP_PLANET").map(|v| v != "0" && !v.is_empty()).unwrap_or(false)
}
struct PlanetBz2ToPbf;
impl Bencher for PlanetBz2ToPbf {
fn id(&self) -> &'static str { "znippy.planet_bz2_to_pbf" }
fn run(&self) -> Result<BenchResult> {
if skip_planet() { return Err(anyhow!("skipped: NORNIR_BENCH_SKIP_PLANET")); }
let input = PathBuf::from(PLANET_BZ2);
let output = PathBuf::from(PLANET_PBF);
require_local(&input, "PLANET_BZ2")?;
require_local(&output, "PLANET_PBF")?;
if !input.exists() {
return Err(anyhow!("planet bz2 missing at {}", input.display()));
}
if output.exists() {
std::fs::remove_file(&output)
.with_context(|| format!("remove stale {}", output.display()))?;
}
let input_bytes = std::fs::metadata(&input)?.len();
let n_workers = workers();
let start = Instant::now();
let (elements, bytes_written) = osm_katana::xml_to_pbf::xml_to_pbf_bz2_path(
&input,
&output,
n_workers,
true, // skip changeset prologue
).context("xml_to_pbf_bz2_path (planet bz2 → pbf)")?;
let s = start.elapsed().as_secs_f64();
let mut m = serde_json::Map::new();
m.insert("stage".into(), json!("bz2_to_pbf"));
m.insert("input".into(), json!(input.display().to_string()));
m.insert("output".into(), json!(output.display().to_string()));
m.insert("workers".into(), json!(n_workers));
m.insert("input_bytes".into(), json!(input_bytes));
m.insert("output_bytes".into(), json!(bytes_written));
m.insert("elements".into(), json!(elements));
m.insert("seconds".into(), json!(s));
m.insert("input_mb_per_sec".into(), json!(input_bytes as f64 / s / 1_048_576.0));
m.insert("output_mb_per_sec".into(), json!(bytes_written as f64 / s / 1_048_576.0));
m.insert("elements_per_sec".into(), json!(elements as f64 / s));
Ok(BenchResult { name: "planet_bz2_to_pbf".into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench_ordered!(0, PlanetBz2ToPbf);
// ──────────────────────────── stage 1: pbf → geoparquet ─────────────────────
struct PlanetPbfToGeo;
impl Bencher for PlanetPbfToGeo {
fn id(&self) -> &'static str { "znippy.planet_pbf_to_geo" }
fn run(&self) -> Result<BenchResult> {
if skip_planet() { return Err(anyhow!("skipped: NORNIR_BENCH_SKIP_PLANET")); }
let input = PathBuf::from(PLANET_PBF);
let output = PathBuf::from(PLANET_GEO);
require_local(&input, "PLANET_PBF")?;
require_local(&output, "PLANET_GEO")?;
if !input.exists() {
return Err(anyhow!(
"planet pbf missing at {} — stage 0 must produce it first",
input.display()
));
}
if output.exists() {
std::fs::remove_dir_all(&output)
.with_context(|| format!("remove stale {}", output.display()))?;
}
std::fs::create_dir_all(&output).context("create geoparquet out dir")?;
let input_bytes = std::fs::metadata(&input)?.len();
let opts = osm_katana::ConvertOptions {
output_dir: output.clone(),
include_nodes: true,
include_ways: true,
include_rels: true,
// osm-katana's ConvertOptions gained this too; true = the prior
// behaviour (the `node_refs` column of ways.parquet is written).
// A bench that dropped it would measure a 26.5 % smaller ways
// table than every published pack, and read as faster for having
// written less.
include_node_refs: true,
compression: "zstd".into(),
vtd_workers: 0,
geometry: "resolved".into(),
skip_changesets: false,
log_path: None,
clip: None,
// osm-katana's ConvertOptions gained this; true = the prior behaviour
// (node coords kept), which the resolved-geometry bench relies on.
keep_node_coords: true,
// `reuse_node_coords` defaults to TRUE, and that default is wrong for a
// bench: it skips pass 1 whenever a previous run left a complete
// `node_coords.arrow` for the same input and clip. A second run would
// then measure a cache hit rather than the conversion, and the arm would
// look faster for having done nothing — the hollow-fixture trap
// (`clone_large` once served a repository with zero packs in it, so a
// working pack-copy and a dead one produced the same output). A bench
// pays pass 1 every time, on purpose.
reuse_node_coords: false,
// Also gained since this file last compiled: a cooperative cancel
// handle. A bench runs to completion by definition, so None.
cancel: None,
};
let start = Instant::now();
osm_katana::convert(&input, &opts).context("osm_katana::convert (pbf → geoparquet)")?;
let s = start.elapsed().as_secs_f64();
let output_bytes = walk_size(&output).unwrap_or(0);
let mut m = serde_json::Map::new();
m.insert("stage".into(), json!("pbf_to_geo"));
m.insert("input".into(), json!(input.display().to_string()));
m.insert("output".into(), json!(output.display().to_string()));
m.insert("workers".into(), json!(workers()));
m.insert("input_bytes".into(), json!(input_bytes));
m.insert("output_bytes".into(), json!(output_bytes));
m.insert("seconds".into(), json!(s));
m.insert("input_mb_per_sec".into(), json!(input_bytes as f64 / s / 1_048_576.0));
m.insert("output_mb_per_sec".into(), json!(output_bytes as f64 / s / 1_048_576.0));
m.insert("compression_ratio".into(), json!(
if output_bytes > 0 { input_bytes as f64 / output_bytes as f64 } else { 0.0 }
));
Ok(BenchResult { name: "planet_pbf_to_geo".into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench_ordered!(1, PlanetPbfToGeo);
// ─────────────── stage 2: bz2 → geoparquet DIRECT (one-shot convert) ──────────
//
// The ALTERNATIVE to the chained bz2→pbf→geo path: a single `osm_katana::convert`
// straight off the 162 GB planet bz2 (Gatling bz2 decode → resolved geometry).
// Same per-phase telemetry (pass1 / sort_and_stree / pass2 cpu_cores) is folded
// into the metrics. Output goes to a SEPARATE dir so it never clobbers the
// chained stage-1 geoparquet.
struct PlanetBz2ToGeoDirect;
impl Bencher for PlanetBz2ToGeoDirect {
fn id(&self) -> &'static str { "znippy.planet_bz2_to_geo_direct" }
fn run(&self) -> Result<BenchResult> {
if skip_planet() { return Err(anyhow!("skipped: NORNIR_BENCH_SKIP_PLANET")); }
let input = PathBuf::from(PLANET_BZ2);
let output = PathBuf::from(PLANET_GEO_DIRECT);
require_local(&input, "PLANET_BZ2")?;
require_local(&output, "PLANET_GEO_DIRECT")?;
if !input.exists() {
return Err(anyhow!("planet bz2 missing at {}", input.display()));
}
if output.exists() {
std::fs::remove_dir_all(&output)
.with_context(|| format!("remove stale {}", output.display()))?;
}
std::fs::create_dir_all(&output).context("create direct geoparquet out dir")?;
let log = output.with_extension("telemetry.jsonl");
let _ = std::fs::remove_file(&log);
let input_bytes = std::fs::metadata(&input)?.len();
let opts = osm_katana::ConvertOptions {
output_dir: output.clone(),
include_nodes: true,
include_ways: true,
include_rels: true,
// osm-katana's ConvertOptions gained this too; true = the prior
// behaviour (the `node_refs` column of ways.parquet is written).
// A bench that dropped it would measure a 26.5 % smaller ways
// table than every published pack, and read as faster for having
// written less.
include_node_refs: true,
compression: "zstd".into(),
vtd_workers: 0,
geometry: "resolved".into(),
skip_changesets: true, // skip planet bz2 changeset prologue
log_path: Some(log.clone()),
clip: None,
// osm-katana's ConvertOptions gained this; true = the prior behaviour
// (node coords kept), which the resolved-geometry bench relies on.
keep_node_coords: true,
// `reuse_node_coords` defaults to TRUE, and that default is wrong for a
// bench: it skips pass 1 whenever a previous run left a complete
// `node_coords.arrow` for the same input and clip. A second run would
// then measure a cache hit rather than the conversion, and the arm would
// look faster for having done nothing — the hollow-fixture trap
// (`clone_large` once served a repository with zero packs in it, so a
// working pack-copy and a dead one produced the same output). A bench
// pays pass 1 every time, on purpose.
reuse_node_coords: false,
// Also gained since this file last compiled: a cooperative cancel
// handle. A bench runs to completion by definition, so None.
cancel: None,
};
let start = Instant::now();
osm_katana::convert(&input, &opts).context("osm_katana::convert (bz2 → geoparquet, direct)")?;
let s = start.elapsed().as_secs_f64();
let output_bytes = walk_size(&output).unwrap_or(0);
let mut m = serde_json::Map::new();
m.insert("stage".into(), json!("bz2_to_geo_direct"));
m.insert("input".into(), json!(input.display().to_string()));
m.insert("output".into(), json!(output.display().to_string()));
m.insert("workers".into(), json!(workers()));
m.insert("input_bytes".into(), json!(input_bytes));
m.insert("output_bytes".into(), json!(output_bytes));
m.insert("seconds".into(), json!(s));
m.insert("input_mb_per_sec".into(), json!(input_bytes as f64 / s / 1_048_576.0));
m.insert("output_mb_per_sec".into(), json!(output_bytes as f64 / s / 1_048_576.0));
m.insert("cores".into(), json!(cores()));
fold_phase_cores(&log, &mut m);
Ok(BenchResult { name: "planet_bz2_to_geo_direct".into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench_ordered!(2, PlanetBz2ToGeoDirect);
// ─────────────── stage 3: geoparquet → Arrow IPC (geo2arrow) ──────────────────
//
// Streams the planet `nodes.parquet` (the largest table) through the same
// parquet→Arrow-IPC copy the `osm-katana geo2arrow` CLI does. Depends on the
// chained stage-1 geoparquet existing. This path has no convert sub-phases, so
// it records throughput + walltime only (no pass1/pass2 cores).
struct PlanetGeoToArrow;
impl Bencher for PlanetGeoToArrow {
fn id(&self) -> &'static str { "znippy.planet_geo_to_arrow" }
fn run(&self) -> Result<BenchResult> {
if skip_planet() { return Err(anyhow!("skipped: NORNIR_BENCH_SKIP_PLANET")); }
use std::fs::File;
use arrow::ipc::writer::FileWriter as ArrowFileWriter;
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
let geo = PathBuf::from(PLANET_GEO);
let input = geo.join("nodes.parquet");
let output = PathBuf::from(PLANET_ARROW);
require_local(&input, "PLANET_GEO/nodes.parquet")?;
require_local(&output, "PLANET_ARROW")?;
if !input.exists() {
return Err(anyhow!(
"planet geoparquet nodes.parquet missing at {} — stage 1 must produce it first",
input.display()
));
}
if let Some(parent) = output.parent() { std::fs::create_dir_all(parent)?; }
if output.exists() { std::fs::remove_file(&output)?; }
let input_bytes = std::fs::metadata(&input)?.len();
let start = Instant::now();
let builder = ParquetRecordBatchReaderBuilder::try_new(File::open(&input)?)?;
let schema = builder.schema().clone();
let mut w = ArrowFileWriter::try_new(File::create(&output)?, &schema)?;
let mut rows = 0u64;
for batch in builder.build()? {
let b = batch?;
rows += b.num_rows() as u64;
w.write(&b)?;
}
w.finish()?;
let s = start.elapsed().as_secs_f64();
let output_bytes = std::fs::metadata(&output)?.len();
let mut m = serde_json::Map::new();
m.insert("stage".into(), json!("geo_to_arrow"));
m.insert("input".into(), json!(input.display().to_string()));
m.insert("output".into(), json!(output.display().to_string()));
m.insert("input_bytes".into(), json!(input_bytes));
m.insert("output_bytes".into(), json!(output_bytes));
m.insert("rows".into(), json!(rows));
m.insert("seconds".into(), json!(s));
m.insert("input_mb_per_sec".into(), json!(input_bytes as f64 / s / 1_048_576.0));
m.insert("cores".into(), json!(cores()));
Ok(BenchResult { name: "planet_geo_to_arrow".into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench_ordered!(3, PlanetGeoToArrow);
// ─────────────── stage 4: osmium europe-clip → convert (±clip arm) ────────────
//
// The "WITH clip" arm of the ±europe-clip matrix: run `osmium extract` (a europe
// bbox, complete_ways) to carve a europe pbf onto /home, then
// `osm_katana::convert` that clipped pbf. The unclipped planet stages (0-2) are
// the "WITHOUT clip" arm. Records the osmium extract walltime AND the convert's
// per-phase cpu_cores telemetry.
//
// NB: osmium extracts from the planet **bz2** (standard XML/bzip2, osmium-
// readable), NOT from osm-katana's stage-0 pbf — that pbf uses zstd blobs which
// libosmium cannot decode ("PBF error: zstd blobs not supported"). So the clip
// arm depends on the bz2 source, independent of the chained pbf stages.
struct PlanetEuropeClipConvert;
impl Bencher for PlanetEuropeClipConvert {
fn id(&self) -> &'static str { "znippy.planet_europe_clip_convert" }
fn run(&self) -> Result<BenchResult> {
if skip_planet() { return Err(anyhow!("skipped: NORNIR_BENCH_SKIP_PLANET")); }
let planet_src = PathBuf::from(PLANET_BZ2); // osmium-readable XML/bzip2
let clip_pbf = PathBuf::from(PLANET_EUROPE_PBF);
let geo = PathBuf::from(PLANET_EUROPE_GEO);
require_local(&planet_src, "PLANET_BZ2")?;
require_local(&clip_pbf, "PLANET_EUROPE_PBF")?;
require_local(&geo, "PLANET_EUROPE_GEO")?;
if !planet_src.exists() {
return Err(anyhow!("planet bz2 missing at {}", planet_src.display()));
}
if !legacy::tool_available("osmium") {
return Err(anyhow!("skipped: osmium not on PATH"));
}
if let Some(parent) = clip_pbf.parent() { std::fs::create_dir_all(parent)?; }
if clip_pbf.exists() { std::fs::remove_file(&clip_pbf)?; }
// ── osmium extract (europe bbox, complete_ways) from the bz2 ──
let planet_bytes = std::fs::metadata(&planet_src)?.len();
let clip_start = Instant::now();
let status = Command::new("osmium")
.arg("extract")
.arg("--bbox").arg(EUROPE_BBOX)
.arg("--strategy").arg("complete_ways")
.arg("--overwrite")
.arg("-o").arg(&clip_pbf)
.arg(&planet_src)
.stdout(Stdio::null())
.stderr(Stdio::null())
.status()
.context("spawn osmium extract")?;
if !status.success() {
return Err(anyhow!("osmium extract failed (status {status})"));
}
let clip_s = clip_start.elapsed().as_secs_f64();
let clip_bytes = std::fs::metadata(&clip_pbf)?.len();
// ── convert the clipped europe pbf ──
if geo.exists() {
std::fs::remove_dir_all(&geo)
.with_context(|| format!("remove stale {}", geo.display()))?;
}
std::fs::create_dir_all(&geo).context("create europe-clip geoparquet out dir")?;
let log = geo.with_extension("telemetry.jsonl");
let _ = std::fs::remove_file(&log);
let opts = osm_katana::ConvertOptions {
output_dir: geo.clone(),
include_nodes: true,
include_ways: true,
include_rels: true,
// osm-katana's ConvertOptions gained this too; true = the prior
// behaviour (the `node_refs` column of ways.parquet is written).
// A bench that dropped it would measure a 26.5 % smaller ways
// table than every published pack, and read as faster for having
// written less.
include_node_refs: true,
compression: "zstd".into(),
vtd_workers: 0,
geometry: "resolved".into(),
skip_changesets: false,
log_path: Some(log.clone()),
clip: None,
// osm-katana's ConvertOptions gained this; true = the prior behaviour
// (node coords kept), which the resolved-geometry bench relies on.
keep_node_coords: true,
// `reuse_node_coords` defaults to TRUE, and that default is wrong for a
// bench: it skips pass 1 whenever a previous run left a complete
// `node_coords.arrow` for the same input and clip. A second run would
// then measure a cache hit rather than the conversion, and the arm would
// look faster for having done nothing — the hollow-fixture trap
// (`clone_large` once served a repository with zero packs in it, so a
// working pack-copy and a dead one produced the same output). A bench
// pays pass 1 every time, on purpose.
reuse_node_coords: false,
// Also gained since this file last compiled: a cooperative cancel
// handle. A bench runs to completion by definition, so None.
cancel: None,
};
let conv_start = Instant::now();
osm_katana::convert(&clip_pbf, &opts).context("osm_katana::convert (europe-clip pbf → geoparquet)")?;
let conv_s = conv_start.elapsed().as_secs_f64();
let geo_bytes = walk_size(&geo).unwrap_or(0);
let mut m = serde_json::Map::new();
m.insert("stage".into(), json!("europe_clip_convert"));
m.insert("bbox".into(), json!(EUROPE_BBOX));
m.insert("clip_source".into(), json!(planet_src.display().to_string()));
m.insert("clip_pbf".into(), json!(clip_pbf.display().to_string()));
m.insert("output".into(), json!(geo.display().to_string()));
m.insert("workers".into(), json!(workers()));
m.insert("planet_bytes".into(), json!(planet_bytes));
m.insert("clip_bytes".into(), json!(clip_bytes));
m.insert("geo_bytes".into(), json!(geo_bytes));
m.insert("clip_seconds".into(), json!(clip_s));
m.insert("convert_seconds".into(), json!(conv_s));
m.insert("clip_mb_per_sec".into(), json!(planet_bytes as f64 / clip_s / 1_048_576.0));
m.insert("convert_mb_per_sec".into(), json!(clip_bytes as f64 / conv_s / 1_048_576.0));
m.insert("cores".into(), json!(cores()));
fold_phase_cores(&log, &mut m);
Ok(BenchResult { name: "planet_europe_clip_convert".into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench_ordered!(4, PlanetEuropeClipConvert);
// ──────────────────── region-scale pbf → geoparquet (fast) ───────────────────
//
// The planet stages above are gated behind NORNIR_BENCH_SKIP_PLANET and take ~45
// min on the 162 GB bz2 / 76 GB pbf. These region benches run the SAME
// `osm_katana::convert` (pbf → resolved-geometry GeoParquet) on a small extract
// so the geo-conversion path gets a quick regression signal on every run. Each
// skips gracefully (recorded red, not fatal) when its fixture is absent — drop a
// `europe-latest.osm.pbf` into ~/work and the europe row lights up automatically.
struct RegionPbfToGeo {
id: &'static str,
result: &'static str,
pbf: &'static str,
}
impl Bencher for RegionPbfToGeo {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
let input = PathBuf::from(self.pbf);
require_local(&input, "region pbf")?;
if !input.exists() {
return Err(anyhow!("skipped: {} not present", input.display()));
}
let tmp = tempfile::TempDir::new()?;
let output = tmp.path().join("geo");
std::fs::create_dir_all(&output)?;
let input_bytes = std::fs::metadata(&input)?.len();
let opts = osm_katana::ConvertOptions {
output_dir: output.clone(),
include_nodes: true,
include_ways: true,
include_rels: true,
// osm-katana's ConvertOptions gained this too; true = the prior
// behaviour (the `node_refs` column of ways.parquet is written).
// A bench that dropped it would measure a 26.5 % smaller ways
// table than every published pack, and read as faster for having
// written less.
include_node_refs: true,
compression: "zstd".into(),
vtd_workers: 0,
geometry: "resolved".into(),
skip_changesets: false,
log_path: None,
clip: None,
// osm-katana's ConvertOptions gained this; true = the prior behaviour
// (node coords kept), which the resolved-geometry bench relies on.
keep_node_coords: true,
// `reuse_node_coords` defaults to TRUE, and that default is wrong for a
// bench: it skips pass 1 whenever a previous run left a complete
// `node_coords.arrow` for the same input and clip. A second run would
// then measure a cache hit rather than the conversion, and the arm would
// look faster for having done nothing — the hollow-fixture trap
// (`clone_large` once served a repository with zero packs in it, so a
// working pack-copy and a dead one produced the same output). A bench
// pays pass 1 every time, on purpose.
reuse_node_coords: false,
// Also gained since this file last compiled: a cooperative cancel
// handle. A bench runs to completion by definition, so None.
cancel: None,
};
let start = Instant::now();
osm_katana::convert(&input, &opts).context("osm_katana::convert (region pbf → geoparquet)")?;
let s = start.elapsed().as_secs_f64();
let output_bytes = walk_size(&output).unwrap_or(0);
let mut m = serde_json::Map::new();
m.insert("stage".into(), json!("pbf_to_geo"));
m.insert("input".into(), json!(input.display().to_string()));
m.insert("input_bytes".into(), json!(input_bytes));
m.insert("output_bytes".into(), json!(output_bytes));
m.insert("seconds".into(), json!(s));
m.insert("input_mb_per_sec".into(), json!(input_bytes as f64 / s / 1_048_576.0));
m.insert("output_mb_per_sec".into(), json!(output_bytes as f64 / s / 1_048_576.0));
m.insert("compression_ratio".into(), json!(
if output_bytes > 0 { input_bytes as f64 / output_bytes as f64 } else { 0.0 }
));
m.insert("cores".into(), json!(cores()));
Ok(BenchResult { name: self.result.into(), source: BenchSource::Measured, metrics: m })
}
}
// Liechtenstein (3.2 MB) is the always-on fast geo smoke test; europe lights up
// when the fixture is present.
nornir::register_bench!(RegionPbfToGeo {
id: "znippy.geo_liechtenstein", result: "geo_liechtenstein",
pbf: "/home/rickard/work/liechtenstein-latest.osm.pbf",
});
nornir::register_bench!(RegionPbfToGeo {
id: "znippy.geo_europe", result: "geo_europe",
pbf: "/home/rickard/work/europe-latest.osm.pbf",
});
// ─────────────────── znippy-zoomies decompressors vs legacy ──────────────────
//
// Each bench decodes a real corpus with our parallel Rust decompressor and
// times the legacy single-/multi-threaded C tool on the same file, emitting
// paired `<ours>_mbs` + `<legacy>_mbs` metrics (+ `speedup_x`). The `benches`
// doc renderer bolds the faster of the two. A missing legacy tool degrades to
// ours-only; a failed corpus download skips the bench (recorded red, not fatal).
//
// Corpora: Maven Central JARs (cached by `nornir::bench::assets`). For the
// gzip/bzip2 decoders we recompress a JAR into `.gz` / `.bz2` once, cached.
use nornir::bench::{assets, legacy};
use std::fs::File;
use std::process::{Command, Stdio};
#[derive(Clone, Copy)]
enum Decoder { Ljar, Lzip, Lbzip2, Lgz }
struct DecompCmp {
id: &'static str,
result: &'static str,
decoder: Decoder,
jar: usize, // index into assets::BENCH_JARS
}
/// Recompress `src` into a sibling `<src>.<ext>` via `tool` (writing to stdout),
/// cached. Returns `None` if the compression tool is absent or fails.
fn ensure_recompressed(src: &Path, ext: &str, tool: &str) -> Option<PathBuf> {
let out = src.with_extension(format!("jar.{ext}"));
if out.exists() {
return Some(out);
}
if !legacy::tool_available(tool) {
return None;
}
let f = File::create(&out).ok()?;
let ok = Command::new(tool)
.arg("-c")
.arg(src)
.stdout(f)
.stderr(Stdio::null())
.status()
.ok()?
.success();
if ok {
Some(out)
} else {
let _ = std::fs::remove_file(&out);
None
}
}
impl Bencher for DecompCmp {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
let cache = assets::default_cache_dir();
let asset = &assets::BENCH_JARS[self.jar];
let jar = assets::ensure(asset, &cache)
.ok_or_else(|| anyhow!("download failed: {}", asset.filename))?;
// (ours_label, legacy_label, corpus path, ours decode closure, legacy cmd)
let (ours_label, legacy_label, corpus, legacy_tool): (&str, &str, PathBuf, &str) =
match self.decoder {
Decoder::Ljar => ("ljar", "unzip", jar.clone(), "unzip"),
Decoder::Lzip => ("lzip", "unzip", jar.clone(), "unzip"),
Decoder::Lbzip2 => {
let bz2 = ensure_recompressed(&jar, "bz2", "bzip2")
.ok_or_else(|| anyhow!("bzip2 not available to build .bz2 corpus"))?;
("lbzip2", "lbzip2_c", bz2, "lbzip2")
}
Decoder::Lgz => {
let gz = ensure_recompressed(&jar, "gz", "gzip")
.ok_or_else(|| anyhow!("gzip not available to build .gz corpus"))?;
// Competitive framing — read before chasing a GB/s number here:
// igzip (Intel ISA-L) is the per-core throughput king. If "fastest"
// means raw GB/s on a single stream, igzip is the bar to beat and we
// will not win that axis. lgz's edge is orthogonal: it *scales the
// unscalable input* — parallelizing arbitrary, hint-free gzip across
// all cores (speculative resync), not just flush-marked pigz/BGZF.
// We benchmark against pigz because it is the canonical multi-core
// competitor; igzip belongs in a separate single-stream throughput row.
("lgz", "pigz", gz, "pigz")
}
};
let data = std::fs::read(&corpus)?;
let bytes = data.len() as u64;
// ours: best-of-3 in-process decode.
let decoder = self.decoder;
let ours = legacy::time_rust(3, || {
let _ = decode_with(decoder, &data);
});
// sanity: the decode must actually succeed once.
decode_with(decoder, &data)
.map_err(|e| anyhow!("{ours_label} decode failed: {e}"))?;
// legacy: best-of-3 subprocess decode of the same corpus to a null sink.
let legacy_dur = legacy::time_command_best(legacy_tool, 3, || {
let mut c = Command::new(legacy_tool);
match decoder {
Decoder::Ljar | Decoder::Lzip => { c.arg("-p").arg(&corpus); } // unzip -p → stdout
Decoder::Lbzip2 | Decoder::Lgz => { c.arg("-dc").arg(&corpus); } // decompress → stdout
}
c
});
let cmp = legacy::Comparison {
bytes,
ours_label: ours_label.into(),
ours,
legacy_label: legacy_label.into(),
legacy: legacy_dur,
};
Ok(cmp.into_result(self.result))
}
}
/// Decode `data` with the selected decompressor; returns the decoded byte count.
fn decode_with(decoder: Decoder, data: &[u8]) -> Result<usize> {
Ok(match decoder {
Decoder::Ljar => ljar::decompress_jar(data).map(|e| e.len()).map_err(|e| anyhow!("{e:?}"))?,
Decoder::Lzip => lzip_parallel::decompress_zip(data).map(|e| e.len()).map_err(|e| anyhow!("{e:?}"))?,
Decoder::Lbzip2 => lbzip2::parallel::decompress_parallel(data).map(|b| b.len()).map_err(|e| anyhow!("{e:?}"))?,
Decoder::Lgz => lgz::decompress_gz(data).map(|b| b.len()).map_err(|e| anyhow!("{e:?}"))?,
})
}
// ───────────────── raw SIMD-DEFLATE core: linflate vs miniz ──────────────────
//
// Single-stream, single-core decode of one DEFLATE stream built from a real
// Maven JAR (recompressed as raw DEFLATE via miniz level 6, cached in memory).
// This is the scorecard for the custom inflate core — linflate's
// libdeflate-style fastloop, branchless 64-bit refill, and SIMD match copy
// (see znippy-zoomies/linflate/DESIGN.md) — against miniz_oxide, the pure-Rust
// baseline the ecosystem ships by default. Decode-only: both sides eat the
// same compressed bytes; throughput is over the *uncompressed* output.
struct InflateCmp {
id: &'static str,
result: &'static str,
jar: usize, // index into assets::BENCH_JARS
}
impl Bencher for InflateCmp {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
let cache = assets::default_cache_dir();
let asset = &assets::BENCH_JARS[self.jar];
let jar = assets::ensure(asset, &cache)
.ok_or_else(|| anyhow!("download failed: {}", asset.filename))?;
// Representative corpus = the JAR's *decompressed entry contents*
// (class files, resources) — what linflate decodes in production.
// Compressing the raw JAR bytes instead would build a stored-block-heavy
// stream of already-deflated data: the codec's worst case, not its job.
let jar_bytes = std::fs::read(&jar)?;
let mut original = Vec::new();
for e in ljar::decompress_jar(&jar_bytes).map_err(|e| anyhow!("{e:?}"))? {
original.extend_from_slice(&e.data);
}
anyhow::ensure!(!original.is_empty(), "empty corpus from {}", asset.filename);
let compressed = miniz_oxide::deflate::compress_to_vec(&original, 6);
let bytes = original.len() as u64; // throughput over decoded output
// Sanity: linflate must reproduce the original byte-for-byte.
let out = linflate::inflate_to_vec(&compressed, original.len())
.map_err(|e| anyhow!("linflate decode failed: {e:?}"))?;
anyhow::ensure!(out == original, "linflate roundtrip mismatch");
let ours = legacy::time_rust(3, || {
let _ = linflate::inflate_to_vec(&compressed, original.len());
});
let miniz = legacy::time_rust(3, || {
let _ = miniz_oxide::inflate::decompress_to_vec(&compressed);
});
let cmp = legacy::Comparison {
bytes,
ours_label: "linflate".into(),
ours,
legacy_label: "miniz".into(),
legacy: Some(miniz),
};
Ok(cmp.into_result(self.result))
}
}
nornir::register_bench!(InflateCmp { id: "linflate.guava", result: "linflate_guava", jar: 0 });
nornir::register_bench!(InflateCmp { id: "linflate.scala", result: "linflate_scala", jar: 2 });
// guava (3.4 MB) is the canonical small workload; scala (5.8 MB) shows scaling.
nornir::register_bench!(DecompCmp { id: "ljar.guava", result: "ljar_guava", decoder: Decoder::Ljar, jar: 0 });
nornir::register_bench!(DecompCmp { id: "ljar.scala", result: "ljar_scala", decoder: Decoder::Ljar, jar: 2 });
nornir::register_bench!(DecompCmp { id: "lzip.guava", result: "lzip_guava", decoder: Decoder::Lzip, jar: 0 });
nornir::register_bench!(DecompCmp { id: "lzip.scala", result: "lzip_scala", decoder: Decoder::Lzip, jar: 2 });
nornir::register_bench!(DecompCmp { id: "lbzip2.guava", result: "lbzip2_guava", decoder: Decoder::Lbzip2, jar: 0 });
nornir::register_bench!(DecompCmp { id: "lbzip2.scala", result: "lbzip2_scala", decoder: Decoder::Lbzip2, jar: 2 });
nornir::register_bench!(DecompCmp { id: "lgz.guava", result: "lgz_guava", decoder: Decoder::Lgz, jar: 0 });
nornir::register_bench!(DecompCmp { id: "lgz.scala", result: "lgz_scala", decoder: Decoder::Lgz, jar: 2 });
// ──────────────────── znippy archive compress/decompress ─────────────────────
//
// The real numbers behind the README headline: compress a directory to a
// .znippy archive and decompress it back, measuring MB/s on the *input* bytes.
// Synthetic corpora self-generate; real jar/crate dirs are used when present
// under ~/work/holger_tests/. Run single-core by wrapping the process in
// `taskset -c 0` (then std::thread::available_parallelism()==1).
use znippy_common::plugin::{ArchiveTypePlugin, PluginRegistry};
use znippy_compress::{ArchiveEntry, compress_dir, compress_stream};
use znippy_plugin_maven::{NativeMavenPlugin, extract_maven_metadata};
#[derive(Clone, Copy)]
enum Corpus { Small10k, Mixed, RealJars, RealCrates, RealWheels }
struct ArchiveBench {
id: &'static str,
result: &'static str,
corpus: Corpus,
}
fn gen_text(size: usize) -> Vec<u8> {
// Mildly compressible pseudo-text.
const WORDS: &[&str] = &["fn", "let", "struct", "impl", "the", "data", "znippy", "arrow", "block"];
let mut s = String::with_capacity(size + 16);
let mut i = 0usize;
while s.len() < size {
s.push_str(WORDS[i % WORDS.len()]);
s.push(if i % 7 == 0 { '\n' } else { ' ' });
i += 1;
}
s.truncate(size);
s.into_bytes()
}
fn gen_random(size: usize) -> Vec<u8> {
// Incompressible: a simple xorshift fill (no rng dep).
let mut v = Vec::with_capacity(size);
let mut x: u64 = 0x9E3779B97F4A7C15 ^ size as u64;
while v.len() < size {
x ^= x << 13; x ^= x >> 7; x ^= x << 17;
v.extend_from_slice(&x.to_le_bytes());
}
v.truncate(size);
v
}
fn build_small(dir: &Path, count: usize, size: usize) {
for i in 0..count {
let sub = dir.join(format!("d{:03}", i / 1000));
let _ = std::fs::create_dir_all(&sub);
let _ = std::fs::write(sub.join(format!("f{:06}.txt", i)), gen_text(size));
}
}
fn build_mixed(dir: &Path) {
let poms = dir.join("poms");
let _ = std::fs::create_dir_all(&poms);
for i in 0..500 { let _ = std::fs::write(poms.join(format!("p{i:04}.xml")), gen_text(1024 + (i % 8) * 1024)); }
let jars = dir.join("jars");
let _ = std::fs::create_dir_all(&jars);
for i in 0..200 { let _ = std::fs::write(jars.join(format!("l{i:04}.jar")), gen_random(100 * 1024 + (i % 20) * 100 * 1024)); }
}
fn dir_bytes(dir: &Path) -> (u64, usize) {
let mut bytes = 0u64;
let mut files = 0usize;
for e in walkdir::WalkDir::new(dir).into_iter().flatten() {
if e.file_type().is_file() {
bytes += e.metadata().map(|m| m.len()).unwrap_or(0);
files += 1;
}
}
(bytes, files)
}
impl Bencher for ArchiveBench {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
let tmp = tempfile::TempDir::new()?;
let input = match self.corpus {
Corpus::Small10k => { let d = tmp.path().join("in"); std::fs::create_dir_all(&d)?; build_small(&d, 10_000, 10 * 1024); d }
Corpus::Mixed => { let d = tmp.path().join("in"); std::fs::create_dir_all(&d)?; build_mixed(&d); d }
Corpus::RealJars => {
let d = dirs::home_dir().unwrap_or_default().join("work/holger_tests/jars");
if !d.exists() { return Err(anyhow!("skipped: {} not present", d.display())); }
d
}
Corpus::RealCrates => {
let d = dirs::home_dir().unwrap_or_default().join("work/holger_tests/crates");
if !d.exists() { return Err(anyhow!("skipped: {} not present", d.display())); }
d
}
Corpus::RealWheels => {
let d = dirs::home_dir().unwrap_or_default().join("work/holger_tests/wheels");
if !d.exists() { return Err(anyhow!("skipped: {} not present", d.display())); }
d
}
};
let (input_bytes, files) = dir_bytes(&input);
if input_bytes == 0 { return Err(anyhow!("empty corpus")); }
let in_mb = input_bytes as f64 / 1_048_576.0;
let archive = tmp.path().join("out.znippy");
let registry = PluginRegistry::with_plugin(Box::new(NativeMavenPlugin));
let t0 = Instant::now();
compress_dir(&input.to_path_buf(), &archive.to_path_buf(), false, Some(®istry), None, None)?;
let comp_s = t0.elapsed().as_secs_f64();
let output_bytes = std::fs::metadata(&archive)?.len();
let out_dir = tmp.path().join("dec");
std::fs::create_dir_all(&out_dir)?;
let t1 = Instant::now();
znippy_common::decompress_archive(&archive, true, &out_dir)?;
let dec_s = t1.elapsed().as_secs_f64();
let (raw_chunks, compressed_chunks) = chunk_kinds(&archive);
let mut m = serde_json::Map::new();
m.insert("compress_mbs".into(), json!((in_mb / comp_s * 100.0).round() / 100.0));
m.insert("decompress_mbs".into(), json!((in_mb / dec_s * 100.0).round() / 100.0));
m.insert("ratio_x".into(), json!(((input_bytes as f64 / output_bytes.max(1) as f64) * 100.0).round() / 100.0));
m.insert("files".into(), json!(files));
m.insert("in_mb".into(), json!((in_mb * 10.0).round() / 10.0));
m.insert("cores".into(), json!(cores()));
m.insert("raw_chunks".into(), json!(raw_chunks));
m.insert("compressed_chunks".into(), json!(compressed_chunks));
Ok(BenchResult { name: self.result.into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench!(ArchiveBench { id: "znippy.archive_small_10k", result: "archive_small_10k", corpus: Corpus::Small10k });
nornir::register_bench!(ArchiveBench { id: "znippy.archive_mixed", result: "archive_mixed", corpus: Corpus::Mixed });
nornir::register_bench!(ArchiveBench { id: "znippy.archive_real_jars", result: "archive_real_jars", corpus: Corpus::RealJars });
nornir::register_bench!(ArchiveBench { id: "znippy.archive_real_crates", result: "archive_real_crates", corpus: Corpus::RealCrates });
nornir::register_bench!(ArchiveBench { id: "znippy.archive_real_wheels", result: "archive_real_wheels", corpus: Corpus::RealWheels });
// ───────────── FAIR ours-vs-rival mashup: znippy vs Arrow-IPC / zstd ──────────
//
// The single-system archive benches above measure znippy only, so a README
// `mashup` renders empty. These two arms pack ONE byte-identical corpus two ways
// and read it back, emitting the SAME metric keys so the pair
// (`znippy_mashup` vs `arrowipc_mashup`) forms a real ours-vs-rival table:
//
// • znippy — all-core Arrow-IPC archive with zstd blobs (compress_dir →
// .znippy, decompress_archive back). Ours.
// • arrowipc — vanilla arrow-rs Arrow-IPC `FileWriter` with `CompressionType::
// ZSTD` over a (path: Utf8, data: Binary) schema, read back via
// `FileReader` materialising every blob. The literal "Arrow-IPC /
// zstd" rival: the same job (bundle N files into one compressed,
// queryable Arrow-IPC container + read all bytes back) done with
// stock arrow-rs instead of znippy's parallel pipeline.
//
// Both measure MB/s on identical INPUT bytes and the on-disk size ratio, on the
// same box in the same run — so they are fair to each other. warmup + iterations,
// best throughput kept (single-shot noise removed). Records cores + peak RSS.
#[derive(Clone, Copy)]
enum MashupCorpus { Mixed }
/// Deterministic corpus — identical bytes on every run and for BOTH systems (the
/// gens are pure functions of size). Heavy: ~1 GiB across 8000 files, half
/// mildly-compressible text, half incompressible blobs — the real mixed-archive
/// shape where znippy's all-core codec meets a single-stream Arrow-IPC writer.
fn build_mashup_corpus(kind: MashupCorpus) -> Vec<(String, Vec<u8>)> {
// Heavy by default (~1 GiB / 8000 files). `ZNIPPY_MASHUP_FILES=N` shrinks the
// file count ONLY for wiring smoke-tests — the published heavy run leaves it unset.
let default_n = match kind { MashupCorpus::Mixed => 8_000usize };
let n = std::env::var("ZNIPPY_MASHUP_FILES").ok()
.and_then(|v| v.parse::<usize>().ok())
.filter(|n| *n > 0)
.unwrap_or(default_n);
let (txt, rnd) = match kind { MashupCorpus::Mixed => (96 * 1024, 160 * 1024) };
let mut files = Vec::with_capacity(n);
for i in 0..n {
let sub = i / 1000;
// Per-file UNIQUE content (seeded by index) so neither packer gets an
// artificial cross-file dedup windfall — the ratio reflects real distinct
// data, not repeated identical blobs.
if i % 2 == 0 {
files.push((format!("d{sub:02}/text_{i:06}.txt"), gen_text_seeded(txt, i as u64)));
} else {
files.push((format!("d{sub:02}/blob_{i:06}.bin"), gen_random_seeded(rnd, i as u64)));
}
}
files
}
/// Per-file compressible text: same tiny vocabulary (so each file still
/// compresses well on its own), but the word sequence is driven by a per-file
/// LCG so no two files are byte-identical.
fn gen_text_seeded(size: usize, seed: u64) -> Vec<u8> {
const WORDS: &[&str] = &["fn", "let", "struct", "impl", "the", "data", "znippy", "arrow", "block"];
let mut s = String::with_capacity(size + 16);
let mut x = seed ^ 0x9E3779B97F4A7C15;
let mut i = 0usize;
while s.len() < size {
x ^= x << 13; x ^= x >> 7; x ^= x << 17;
s.push_str(WORDS[(x as usize) % WORDS.len()]);
s.push(if i % 7 == 0 { '\n' } else { ' ' });
i += 1;
}
s.truncate(size);
s.into_bytes()
}
/// Per-file incompressible blob (xorshift), seeded by index so every blob differs.
fn gen_random_seeded(size: usize, seed: u64) -> Vec<u8> {
let mut v = Vec::with_capacity(size);
let mut x: u64 = 0x9E3779B97F4A7C15 ^ seed.wrapping_mul(0xD1B54A32D192ED03) ^ size as u64;
while v.len() < size {
x ^= x << 13; x ^= x >> 7; x ^= x << 17;
v.extend_from_slice(&x.to_le_bytes());
}
v.truncate(size);
v
}
fn mashup_corpus_bytes(files: &[(String, Vec<u8>)]) -> u64 {
files.iter().map(|(_, d)| d.len() as u64).sum()
}
/// Peak resident set (MB) from /proc/self/status VmHWM — the memory each system
/// touched, recorded per heavy-bench law so the row is self-describing.
fn peak_rss_mb() -> f64 {
std::fs::read_to_string("/proc/self/status")
.ok()
.and_then(|s| s.lines().find(|l| l.starts_with("VmHWM:")).map(str::to_string))
.and_then(|l| l.split_whitespace().nth(1).and_then(|k| k.parse::<f64>().ok()))
.map(|kb| (kb / 1024.0 * 10.0).round() / 10.0)
.unwrap_or(0.0)
}
const MASHUP_WARMUP: usize = 1;
const MASHUP_ITERS: usize = 3;
fn mashup_result(
name: &str, comp_mbs: f64, dec_mbs: f64, in_bytes: u64, out_bytes: u64, files: usize, in_mb: f64,
) -> BenchResult {
let mut m = serde_json::Map::new();
m.insert("compress_mbs".into(), json!((comp_mbs * 100.0).round() / 100.0));
m.insert("decompress_mbs".into(), json!((dec_mbs * 100.0).round() / 100.0));
m.insert("ratio_x".into(), json!(((in_bytes as f64 / out_bytes.max(1) as f64) * 100.0).round() / 100.0));
m.insert("files".into(), json!(files));
m.insert("in_mb".into(), json!((in_mb * 10.0).round() / 10.0));
m.insert("cores".into(), json!(cores()));
m.insert("peak_rss_mb".into(), json!(peak_rss_mb()));
BenchResult { name: name.into(), source: BenchSource::Measured, metrics: m }
}
struct ZnippyMashup { id: &'static str, result: &'static str, corpus: MashupCorpus }
impl Bencher for ZnippyMashup {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
let files = build_mashup_corpus(self.corpus);
let input_bytes = mashup_corpus_bytes(&files);
let in_mb = input_bytes as f64 / 1_048_576.0;
let registry = PluginRegistry::with_plugin(Box::new(NativeMavenPlugin));
// Materialise the identical input tree once (znippy walks a directory).
let src = tempfile::TempDir::new()?;
for (rel, data) in &files {
let p = src.path().join(rel);
if let Some(par) = p.parent() { std::fs::create_dir_all(par)?; }
std::fs::write(&p, data)?;
}
let src_dir = src.path().to_path_buf();
let mut best_comp = 0.0f64;
let mut best_dec = 0.0f64;
let mut out_bytes = 0u64;
for it in 0..(MASHUP_WARMUP + MASHUP_ITERS) {
let tmp = tempfile::TempDir::new()?;
let archive = tmp.path().join("out.znippy");
let t0 = Instant::now();
compress_dir(&src_dir, &archive, false, Some(®istry), None, None)?;
let comp_s = t0.elapsed().as_secs_f64();
out_bytes = std::fs::metadata(&archive)?.len();
let dec = tmp.path().join("dec");
std::fs::create_dir_all(&dec)?;
let t1 = Instant::now();
let verify = znippy_common::decompress_archive(&archive, true, &dec)?;
let dec_s = t1.elapsed().as_secs_f64();
if verify.corrupt_files != 0 {
return Err(anyhow!("znippy round-trip corruption: {} files", verify.corrupt_files));
}
if it >= MASHUP_WARMUP {
best_comp = best_comp.max(in_mb / comp_s.max(1e-9));
best_dec = best_dec.max(in_mb / dec_s.max(1e-9));
}
}
Ok(mashup_result(self.result, best_comp, best_dec, input_bytes, out_bytes, files.len(), in_mb))
}
}
struct ArrowIpcMashup { id: &'static str, result: &'static str, corpus: MashupCorpus }
impl Bencher for ArrowIpcMashup {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
use arrow::array::{Array, ArrayRef, BinaryArray, StringArray};
use arrow::datatypes::{DataType, Field, Schema};
use arrow::ipc::reader::FileReader;
use arrow::ipc::writer::{FileWriter, IpcWriteOptions};
use arrow::ipc::CompressionType;
use arrow::record_batch::RecordBatch;
use std::sync::Arc;
let files = build_mashup_corpus(self.corpus);
let input_bytes = mashup_corpus_bytes(&files);
let in_mb = input_bytes as f64 / 1_048_576.0;
let schema = Arc::new(Schema::new(vec![
Field::new("path", DataType::Utf8, false),
Field::new("data", DataType::Binary, false),
]));
// One RecordBatch per ~1000 files — a real columnar Arrow-IPC packer.
let batch_rows = 1000usize;
let mut batches: Vec<RecordBatch> = Vec::new();
for chunk in files.chunks(batch_rows) {
let paths: Vec<&str> = chunk.iter().map(|(p, _)| p.as_str()).collect();
let datas: Vec<&[u8]> = chunk.iter().map(|(_, d)| d.as_slice()).collect();
let path_arr = Arc::new(StringArray::from(paths)) as ArrayRef;
let data_arr = Arc::new(BinaryArray::from(datas)) as ArrayRef;
batches.push(RecordBatch::try_new(schema.clone(), vec![path_arr, data_arr])
.map_err(|e| anyhow!("arrow RecordBatch: {e}"))?);
}
let write_opts = IpcWriteOptions::default()
.try_with_compression(Some(CompressionType::ZSTD))
.map_err(|e| anyhow!("arrow ipc zstd options: {e}"))?;
let mut best_comp = 0.0f64;
let mut best_dec = 0.0f64;
let mut out_bytes = 0u64;
for it in 0..(MASHUP_WARMUP + MASHUP_ITERS) {
let tmp = tempfile::TempDir::new()?;
let path = tmp.path().join("out.arrow");
// WRITE (compress): serialise + zstd every batch, fsync via drop+finish.
let t0 = Instant::now();
{
let f = std::fs::File::create(&path)?;
let mut w = FileWriter::try_new_with_options(f, &schema, write_opts.clone())
.map_err(|e| anyhow!("arrow FileWriter: {e}"))?;
for b in &batches {
w.write(b).map_err(|e| anyhow!("arrow write: {e}"))?;
}
w.finish().map_err(|e| anyhow!("arrow finish: {e}"))?;
}
let comp_s = t0.elapsed().as_secs_f64();
out_bytes = std::fs::metadata(&path)?.len();
// READ back (decompress): materialise every blob, verify total bytes.
let t1 = Instant::now();
{
let f = std::fs::File::open(&path)?;
let reader = FileReader::try_new(f, None).map_err(|e| anyhow!("arrow FileReader: {e}"))?;
let mut seen = 0u64;
for batch in reader {
let batch = batch.map_err(|e| anyhow!("arrow read: {e}"))?;
let col = batch.column(1).as_any().downcast_ref::<BinaryArray>()
.ok_or_else(|| anyhow!("data column not Binary"))?;
for i in 0..col.len() {
seen = seen.wrapping_add(col.value(i).len() as u64);
}
}
if seen != input_bytes {
return Err(anyhow!("arrow read-back byte mismatch: {seen} != {input_bytes}"));
}
}
let dec_s = t1.elapsed().as_secs_f64();
if it >= MASHUP_WARMUP {
best_comp = best_comp.max(in_mb / comp_s.max(1e-9));
best_dec = best_dec.max(in_mb / dec_s.max(1e-9));
}
}
Ok(mashup_result(self.result, best_comp, best_dec, input_bytes, out_bytes, files.len(), in_mb))
}
}
nornir::register_bench!(ZnippyMashup { id: "znippy.mashup_znippy", result: "znippy_mashup", corpus: MashupCorpus::Mixed });
nornir::register_bench!(ArrowIpcMashup { id: "znippy.mashup_arrowipc", result: "arrowipc_mashup", corpus: MashupCorpus::Mixed });
// ─────────────── coalesced small high-entropy crates (repro_crate) ────────────
//
// Ported from the `repro_crate_roundtrip` test (B-bucket: bench-type, no bench).
// 5000 mixed-size incompressible `.crate` blobs forced through openzl
// (`no_skip=true`) — the coalesce/chunk/compress workload holger's crate path
// hits. The test is now a downscaled (n=200) correctness twin; the full 5000-file
// throughput lives here.
struct ReproCrate;
impl Bencher for ReproCrate {
fn id(&self) -> &'static str { "znippy.repro_crate" }
fn run(&self) -> Result<BenchResult> {
let n = 5000usize;
let sizes = [1_000usize, 5_000, 10_000, 50_000, 100_000];
let tmp = tempfile::TempDir::new()?;
let base = tmp.path().join("repro_crates");
let archive = base.with_extension("znippy");
let mut total_in: u64 = 0;
let t0 = Instant::now();
let compressor = compress_stream(&base, true).map_err(|e| anyhow!("compress_stream: {e:?}"))?;
let sender = compressor.sender().clone();
for i in 0..n {
let len = sizes[i % sizes.len()];
total_in += len as u64;
sender
.send(ArchiveEntry {
relative_path: format!("bench-crate-{i:06}-1.0.0.crate"),
data: gen_random(len),
pkg_type: None,
repo: None,
})
.map_err(|_| anyhow!("send failed (channel closed)"))?;
}
drop(sender);
let report = compressor.finish().map_err(|e| anyhow!("finish: {e:?}"))?;
let comp_s = t0.elapsed().as_secs_f64();
let out = tempfile::TempDir::new()?;
let t1 = Instant::now();
let verify = znippy_common::decompress_archive(&archive, true, out.path())
.map_err(|e| anyhow!("decompress: {e:?}"))?;
let dec_s = t1.elapsed().as_secs_f64();
if verify.corrupt_files != 0 {
return Err(anyhow!("round-trip corruption: {} files", verify.corrupt_files));
}
let in_mb = total_in as f64 / 1_048_576.0;
let mut m = serde_json::Map::new();
m.insert("compress_mbs".into(), json!((in_mb / comp_s.max(1e-9) * 100.0).round() / 100.0));
m.insert("decompress_mbs".into(), json!((in_mb / dec_s.max(1e-9) * 100.0).round() / 100.0));
m.insert("files".into(), json!(report.total_files));
m.insert("chunks".into(), json!(report.chunks));
m.insert("bytes_out".into(), json!(report.total_bytes_out));
m.insert("in_mb".into(), json!((in_mb * 10.0).round() / 10.0));
m.insert("cores".into(), json!(cores()));
Ok(BenchResult { name: "repro_crate".into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench!(ReproCrate);
// ─────────────── Iceberg metadata-sink compress throughput ────────────────────
//
// Ported from `znippy-iceberg/tests/round_trip.rs` (B-bucket: bench-type, no
// bench). Compress a ~25 MB fixture dir through the IcebergSink (index → embedded
// skade catalog) and measure MB/s. The round_trip test stays a correctness
// round-trip (downscaled big.bin); the full throughput lives here.
struct IcebergCompress;
impl Bencher for IcebergCompress {
fn id(&self) -> &'static str { "znippy.iceberg_compress" }
fn run(&self) -> Result<BenchResult> {
let tmp = tempfile::TempDir::new()?;
let src = tmp.path().join("src");
std::fs::create_dir_all(&src)?;
// ~25 MB fixture: compressible + incompressible + a nested file.
let files: [(&str, usize); 4] = [
("repeat.txt", 6 * 1024 * 1024),
("noise1.bin", 8 * 1024 * 1024),
("noise2.bin", 8 * 1024 * 1024),
("sub/nested.dat", 3 * 1024 * 1024),
];
let mut total_in: u64 = 0;
for (rel, len) in files {
let p = src.join(rel);
std::fs::create_dir_all(p.parent().unwrap())?;
std::fs::write(&p, gen_random(len))?;
total_in += len as u64;
}
let archive = tmp.path().join("out.znippy");
let warehouse = tmp.path().join("wh");
let wh = warehouse.clone();
let ns = "out".to_string();
let sink_factory: znippy_common::MetaSinkFactory = Box::new(move |_file, _off| {
Box::new(znippy_iceberg::IcebergSink::new(wh.clone(), ns.clone()))
});
let t0 = Instant::now();
let report = compress_dir(&src, &archive, false, None, None, Some(sink_factory))
.map_err(|e| anyhow!("compress_dir(iceberg): {e:?}"))?;
let comp_s = t0.elapsed().as_secs_f64();
if !warehouse.join("catalog.redb").exists() {
return Err(anyhow!("IcebergSink produced no durable catalog.redb"));
}
let in_mb = total_in as f64 / 1_048_576.0;
let mut m = serde_json::Map::new();
m.insert("compress_mbs".into(), json!((in_mb / comp_s.max(1e-9) * 100.0).round() / 100.0));
m.insert("files".into(), json!(report.total_files));
m.insert("in_mb".into(), json!((in_mb * 10.0).round() / 10.0));
m.insert("cores".into(), json!(cores()));
Ok(BenchResult { name: "iceberg_compress".into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench!(IcebergCompress);
// ─────────────── synthetic codec throughput (compress_stream path) ────────────
//
// Ported from the retired `znippy-tests::perf_bench::perf_benchmark_suite`. These
// measure pure in-memory codec throughput (no walkdir / file-read I/O) by feeding
// generated `ArchiveEntry`s straight through `compress_stream`. The big workloads
// (500 MB streams + the 2 GB single file) are gated behind NORNIR_BENCH_SKIP_HEAVY
// so a focused run can skip multi-GB allocations; default is to run them.
fn skip_heavy() -> bool {
std::env::var("NORNIR_BENCH_SKIP_HEAVY").map(|v| v != "0" && !v.is_empty()).unwrap_or(false)
}
/// PREVIEW mode (`NORNIR_BENCH_PREVIEW=1`) — the fast verification loop.
///
/// Unlike `NORNIR_BENCH_SKIP_HEAVY` (which *drops* the heavy arms, so a broken one
/// stays invisible), preview mode RUNS every registered arm, through the SAME code
/// paths and emitting the SAME metric keys — only the corpus sizes shrink. That is
/// what makes it a verification loop: an arm that no longer works fails in seconds.
///
/// The emitted run is stamped `<host>-preview` (see the `stamp_machine` call in
/// `run_filtered` / the shared `run_main_json`), so a preview number can never land
/// in — or overwrite — the real host's README row.
fn preview() -> bool {
nornir::bench::api::preview_run()
}
/// Shrink a heavy byte size (in MB) for a preview run. `/64`, floor 1 MB.
fn mb_scaled(heavy_mb: usize) -> usize {
if preview() { (heavy_mb / 64).max(1) } else { heavy_mb }
}
/// Shrink a heavy row count for a preview run. `/100`, floor 500 rows.
fn rows_scaled(heavy_rows: usize) -> usize {
if preview() { (heavy_rows / 100).max(500) } else { heavy_rows }
}
/// A heavy arm is dropped only on an explicit `NORNIR_BENCH_SKIP_HEAVY`. A PREVIEW
/// run never drops it — it runs it small, so the arm is actually verified.
fn drop_heavy_arm(heavy: bool) -> bool {
heavy && skip_heavy() && !preview()
}
fn gen_pattern(size: usize) -> Vec<u8> {
// Deterministic binary pattern (mildly compressible).
(0..size).map(|i| (i % 251) as u8).collect()
}
#[derive(Clone, Copy)]
enum Stream { Text500, Binary500, Random500, MixedRepo, Huge2gb }
struct StreamBench {
id: &'static str,
result: &'static str,
workload: Stream,
heavy: bool,
}
impl Bencher for StreamBench {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
if drop_heavy_arm(self.heavy) {
return Err(anyhow!("skipped: NORNIR_BENCH_SKIP_HEAVY"));
}
const MB: usize = 1024 * 1024;
let entries: Vec<ArchiveEntry> = match self.workload {
Stream::Text500 => vec![ArchiveEntry::new("text.txt", gen_text(mb_scaled(500) * MB))],
Stream::Binary500 => vec![ArchiveEntry::new("pattern.bin", gen_pattern(mb_scaled(500) * MB))],
Stream::Random500 => vec![ArchiveEntry::new("random.bin", gen_random(mb_scaled(500) * MB))],
Stream::Huge2gb => vec![ArchiveEntry::new("huge.bin", gen_text(mb_scaled(2 * 1024) * MB))],
Stream::MixedRepo => vec![
ArchiveEntry::new("pom.xml", gen_text(32 * 1024)),
ArchiveEntry::new("app.jar", gen_random(mb_scaled(200) * MB)),
ArchiveEntry::new("sources.jar", gen_text(mb_scaled(100) * MB)),
ArchiveEntry::new("javadoc.jar", gen_text(mb_scaled(80) * MB)),
ArchiveEntry::new("metadata.xml", gen_text(16 * 1024)),
ArchiveEntry::new("deps.tar.gz", gen_random(mb_scaled(150) * MB)),
],
};
let input_bytes: u64 = entries.iter().map(|e| e.data.len() as u64).sum();
let files = entries.len();
let in_mb = input_bytes as f64 / 1_048_576.0;
let tmp = tempfile::TempDir::new()?;
let archive = tmp.path().join("out.znippy");
let t0 = Instant::now();
let compressor = compress_stream(&archive, false)?;
for e in entries { compressor.sender().send(e)?; }
compressor.finish()?;
let comp_s = t0.elapsed().as_secs_f64();
let output_bytes = std::fs::metadata(&archive)?.len();
let out_dir = tmp.path().join("dec");
std::fs::create_dir_all(&out_dir)?;
let t1 = Instant::now();
znippy_common::decompress_archive(&archive, true, &out_dir)?;
let dec_s = t1.elapsed().as_secs_f64();
let (raw_chunks, compressed_chunks) = chunk_kinds(&archive);
let mut m = serde_json::Map::new();
m.insert("compress_mbs".into(), json!((in_mb / comp_s * 100.0).round() / 100.0));
m.insert("decompress_mbs".into(), json!((in_mb / dec_s * 100.0).round() / 100.0));
m.insert("ratio_x".into(), json!(((input_bytes as f64 / output_bytes.max(1) as f64) * 100.0).round() / 100.0));
m.insert("files".into(), json!(files));
m.insert("in_mb".into(), json!((in_mb * 10.0).round() / 10.0));
m.insert("cores".into(), json!(cores()));
// For incompressible workloads (random_500mb) store-if-incompressible
// means raw_chunks dominates and ratio_x stays ~1.0 (never inflates).
m.insert("raw_chunks".into(), json!(raw_chunks));
m.insert("compressed_chunks".into(), json!(compressed_chunks));
Ok(BenchResult { name: self.result.into(), source: BenchSource::Measured, metrics: m })
}
}
// Named `archive_*` so they render in the README archive-throughput table alongside
// the directory-path benches.
nornir::register_bench!(StreamBench { id: "znippy.archive_text_500mb", result: "archive_text_500mb", workload: Stream::Text500, heavy: true });
nornir::register_bench!(StreamBench { id: "znippy.archive_binary_500mb", result: "archive_binary_500mb", workload: Stream::Binary500, heavy: true });
nornir::register_bench!(StreamBench { id: "znippy.archive_random_500mb", result: "archive_random_500mb", workload: Stream::Random500, heavy: true });
nornir::register_bench!(StreamBench { id: "znippy.archive_mixed_repo", result: "archive_mixed_repo", workload: Stream::MixedRepo, heavy: false });
nornir::register_bench!(StreamBench { id: "znippy.archive_huge_2gb", result: "archive_huge_2gb", workload: Stream::Huge2gb, heavy: true });
// ────────────────────── maven metadata extract micro-bench ───────────────────
//
// Ported from `znippy-tests::maven_bench::maven_bench_compare`. Times the native
// ljar-backed plugin against the minimal single-threaded fallback parser on the
// same synthetic JAR, reporting µs/call for each plus the speedup.
//
// ## 2026-08-01 — this bench used to measure the wrong thing
//
// Its fixture came from a local `build_zip_stored`: compression **method 0**,
// compressed size == uncompressed size, over a ~230-byte pom. So
// `native_us_per_call`, `fallback_us_per_call` and `speedup_x` never timed an
// inflate at all — they timed ZIP central-directory parsing, while the whole
// point of the native path is the house DEFLATE core (linflate, fanned out by
// ljar). Real Maven JARs are method 8. A number that is real, precise, and
// about nothing is the same disease as a guard that cannot fail.
//
// The repo already knew this on the TEST side — `tests/tests/plugin_deflate_entry.rs`
// says in its own header that "every pre-existing plugin test builds its fixture
// with `build_zip_stored` (method 0), so the method-8 branch had zero coverage"
// — and shipped a working method-8 builder. Nobody carried it to the bench.
//
// Now: BOTH arms are measured and reported side by side, off the ONE shared
// fixture writer (`znippy-zip-fixtures`, LAW 5 — the stored builder had three
// byte-identical copies in this workspace). The old measurement is not deleted,
// it is relabelled `stored_*` for what it always actually measured (container
// parse only), and the honest DEFLATE figures are the `deflate_*` keys. The
// difference between the two is how much of the old number was parsing.
//
// The deflated fixture is ASSERTED to really be method 8 with csize < usize
// before anything is timed — a "deflate" fixture that is stored in disguise
// would otherwise reproduce the exact bug with a new name.
struct MavenExtractBench;
impl Bencher for MavenExtractBench {
fn id(&self) -> &'static str { "znippy.maven_extract" }
fn run(&self) -> Result<BenchResult> {
use znippy_zip_fixtures as zipfx;
let plugin = NativeMavenPlugin;
let iterations = 5000u32;
// ── Arm 1: the historical fixture — a stored (method 0) JAR holding a
// small pom. Parsing only; no inflate is reachable from here.
let stored_pom = zipfx::pom_xml("com.example", "biglib", "2.0.0");
let stored_jar = zipfx::build_zip_stored(
&zipfx::pom_path("com.example", "biglib"),
stored_pom.as_bytes(),
);
// ── Arm 2: what a real Maven JAR looks like — method 8 DEFLATE over a
// pom big and repetitive enough to emit dynamic-Huffman blocks with long
// back-references, so linflate's fastloop + match-copy actually run.
let deflate_pom = zipfx::big_pom("com.example", "biglib", "2.0.0");
let deflate_jar = zipfx::build_zip_deflated(
&zipfx::pom_path("com.example", "biglib"),
deflate_pom.as_bytes(),
);
// Assert the applied output of the fixture builders, not their names.
let (m0, c0, u0) = zipfx::read_local_header(&stored_jar);
if m0 != zipfx::METHOD_STORED || c0 != u0 {
return Err(anyhow!("stored arm fixture is not method 0 ({m0}, {c0} vs {u0})"));
}
let (m8, c8, u8_) = zipfx::read_local_header(&deflate_jar);
if m8 != zipfx::METHOD_DEFLATE {
return Err(anyhow!("deflate arm fixture is method {m8}, not 8 — it would time no inflate"));
}
if c8 >= u8_ / 2 {
return Err(anyhow!(
"deflate arm fixture barely compressed ({c8} vs {u8_}) — it is stored in disguise"
));
}
let mut m = serde_json::Map::new();
for (label, jar) in [("stored", &stored_jar), ("deflate", &deflate_jar)] {
// Warmup.
for _ in 0..100 {
let _ = plugin.extract_metadata("biglib.jar", jar);
let _ = extract_maven_metadata("biglib.jar", jar);
}
// Sanity: both paths must actually succeed once on this fixture.
plugin.extract_metadata("biglib-2.0.0.jar", jar)
.ok_or_else(|| anyhow!("native maven extract returned None on the {label} JAR"))?;
extract_maven_metadata("biglib-2.0.0.jar", jar)
.ok_or_else(|| anyhow!("fallback maven extract returned None on the {label} JAR"))?;
let t_native = {
let t = Instant::now();
for _ in 0..iterations { let _ = plugin.extract_metadata("biglib-2.0.0.jar", jar); }
t.elapsed()
};
let t_fallback = {
let t = Instant::now();
for _ in 0..iterations { let _ = extract_maven_metadata("biglib-2.0.0.jar", jar); }
t.elapsed()
};
let native_us = t_native.as_secs_f64() * 1e6 / iterations as f64;
let fallback_us = t_fallback.as_secs_f64() * 1e6 / iterations as f64;
let round2 = |v: f64| (v * 100.0).round() / 100.0;
m.insert(format!("{label}_native_us_per_call"), json!(round2(native_us)));
m.insert(format!("{label}_fallback_us_per_call"), json!(round2(fallback_us)));
m.insert(format!("{label}_speedup_x"), json!(
if native_us > 0.0 { round2(fallback_us / native_us) } else { 0.0 }
));
}
// The historical metric names are kept, but now they carry the DEFLATE
// figures — the ones that describe the path this bench claims to measure.
// The stored arm stays visible under its own `stored_*` keys so the
// history is comparable and the parse-only share is readable.
for key in ["native_us_per_call", "fallback_us_per_call", "speedup_x"] {
let v = m[format!("deflate_{key}").as_str()].clone();
m.insert(key.to_string(), v);
}
m.insert("fixture_uncompressed_bytes".into(), json!(u8_));
m.insert("fixture_compressed_bytes".into(), json!(c8));
m.insert("iterations".into(), json!(iterations));
Ok(BenchResult { name: "maven_extract".into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench!(MavenExtractBench);
// ──────────────────── raw io_uring read throughput (no codec) ─────────────────
//
// Ported from the retired `znippy-tests::compress_dir_bench::compress_dir_read_only_10k`.
// Measures the raw NVMe read ceiling the slot_packer's io_uring path is built to
// hit: batched open+read+close of 10k×10 KB files, no compression. This is the
// I/O baseline every compress_dir number is measured against.
use io_uring::{IoUring, opcode, types};
use std::ffi::CString;
struct ReadOnlyUring;
impl Bencher for ReadOnlyUring {
fn id(&self) -> &'static str { "znippy.read_only_uring_10k" }
fn run(&self) -> Result<BenchResult> {
let tmp = tempfile::TempDir::new()?;
let dir = tmp.path().join("small");
std::fs::create_dir_all(&dir)?;
// 10k × 10 KB files across 100-file subdirs.
for i in 0..10_000usize {
let sub = dir.join(format!("d{:03}", i / 1000));
std::fs::create_dir_all(&sub)?;
std::fs::write(sub.join(format!("f{:06}.txt", i)), gen_text(10 * 1024))?;
}
let files: Vec<std::path::PathBuf> = walkdir::WalkDir::new(&dir)
.into_iter().flatten()
.filter(|e| e.file_type().is_file())
.map(|e| e.into_path())
.collect();
let total_bytes: u64 = files.iter().map(|f| f.metadata().map(|m| m.len()).unwrap_or(0)).sum();
const BATCH: usize = 64;
let mut ring = IoUring::new((BATCH * 2) as u32)?;
let mut buf = vec![0u8; 6_500_000]; // one reusable slot
let t0 = Instant::now();
let mut i = 0;
while i < files.len() {
let end = (i + BATCH).min(files.len());
let batch_count = end - i;
// open batch
let cpaths: Vec<CString> = files[i..end].iter()
.map(|p| CString::new(p.as_os_str().as_encoded_bytes()).unwrap())
.collect();
for (j, cp) in cpaths.iter().enumerate() {
let op = opcode::OpenAt::new(types::Fd(-1), cp.as_ptr())
.flags(libc::O_RDONLY).build().user_data(j as u64);
unsafe { ring.submission().push(&op).map_err(|e| anyhow!("uring push open: {e}"))?; }
}
ring.submit_and_wait(batch_count)?;
let mut fds = vec![0i32; batch_count];
for _ in 0..batch_count {
let cqe = ring.completion().next().ok_or_else(|| anyhow!("missing open cqe"))?;
fds[cqe.user_data() as usize] = cqe.result();
}
// read batch
for (j, &fd) in fds.iter().enumerate() {
let size = files[i + j].metadata()?.len() as u32;
let op = opcode::Read::new(types::Fd(fd), buf.as_mut_ptr(), size)
.build().user_data(j as u64);
unsafe { ring.submission().push(&op).map_err(|e| anyhow!("uring push read: {e}"))?; }
}
ring.submit_and_wait(batch_count)?;
for _ in 0..batch_count { ring.completion().next().ok_or_else(|| anyhow!("missing read cqe"))?; }
// close batch
for &fd in &fds {
let op = opcode::Close::new(types::Fd(fd)).build().user_data(0);
unsafe { ring.submission().push(&op).map_err(|e| anyhow!("uring push close: {e}"))?; }
}
ring.submit_and_wait(batch_count)?;
for _ in 0..batch_count { ring.completion().next().ok_or_else(|| anyhow!("missing close cqe"))?; }
i = end;
}
let s = t0.elapsed().as_secs_f64();
let in_mb = total_bytes as f64 / 1_048_576.0;
let mut m = serde_json::Map::new();
m.insert("read_mbs".into(), json!((in_mb / s * 100.0).round() / 100.0));
m.insert("files".into(), json!(files.len()));
m.insert("in_mb".into(), json!((in_mb * 10.0).round() / 10.0));
m.insert("cores".into(), json!(cores()));
Ok(BenchResult { name: "read_only_uring_10k".into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench!(ReadOnlyUring);
// ── Static search tree (stree) — single + batch id lookup ─────────────────────
//
// Benchmarks Ragnar Groot Koerkamp's static search tree (`znippy_zoomies::stree`)
// against std `binary_search` on the SAME sorted slice, reproducing the
// comparison from his article "Static search trees: 40x faster than binary
// search" (https://curiouscoding.nl/posts/static-search-tree/). His headline:
// 1150 ns/query (std binary search, 1 GB input) → ~22-27 ns/query (batched,
// prefetched, single thread). We sweep dataset size across the cache→RAM cliff
// so the gap is visible: small (8 MB, in L2/L3), mid (128 MB, ~L3 size on the
// 3975WX), large (1 GB, RAM-bound — his regime). Keys are sparse sorted i64
// (OSM-node-id shaped: random gaps), queries are random *present* keys (hit
// workload = full-depth traversal). All times are reciprocal throughput
// (ns/query); `_ns` keys render "lower is better".
struct StreeBench {
id: &'static str,
result: &'static str,
log2n: u32,
heavy: bool,
}
impl StreeBench {
/// Deterministic sparse sorted i64 keys (gaps 1..=15), seeded so runs are
/// reproducible and historizable.
fn gen_keys(n: usize, mut seed: u64) -> Vec<i64> {
let mut next = || { // xorshift64*
seed ^= seed >> 12; seed ^= seed << 25; seed ^= seed >> 27;
seed.wrapping_mul(0x2545F4914F6CDD1D)
};
let mut keys = Vec::with_capacity(n);
let mut cur: i64 = 1;
for _ in 0..n {
cur += 1 + (next() % 15) as i64;
keys.push(cur);
}
keys
}
/// `q` random *present* keys (hit workload), shuffled so the access pattern
/// is random across the whole array (defeats prefetcher / cache for large n).
fn gen_queries(keys: &[i64], q: usize, mut seed: u64) -> Vec<i64> {
let mut next = || {
seed ^= seed >> 12; seed ^= seed << 25; seed ^= seed >> 27;
seed.wrapping_mul(0x2545F4914F6CDD1D)
};
(0..q).map(|_| keys[(next() as usize) % keys.len()]).collect()
}
}
impl Bencher for StreeBench {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
use std::hint::black_box;
use znippy_zoomies::stree::STree64;
if drop_heavy_arm(self.heavy) {
return Err(anyhow!("skipped: NORNIR_BENCH_SKIP_HEAVY"));
}
// PREVIEW: cap the key count so even the 1 GB arm is a sub-second run that
// still exercises the whole build + query path (same metric keys).
let n = 1usize << if preview() { self.log2n.min(16) } else { self.log2n };
let q = (1usize << 20).min(n); // 1M queries (or fewer if n is smaller)
let reps = if self.heavy { 3 } else { 5 };
let keys = Self::gen_keys(n, 0xD15EA5E_u64 ^ self.log2n as u64);
let queries = Self::gen_queries(&keys, q, 0xC0FFEE ^ self.log2n as u64);
let t0 = Instant::now();
let tree = STree64::new(&keys);
let build_ms = t0.elapsed().as_secs_f64() * 1000.0;
// Correctness gate: a handful of queries must resolve the same as a
// ground-truth binary search before we trust the timing.
for &qq in queries.iter().take(64) {
let got = tree.find_exact(qq).map(|i| keys[i]);
anyhow::ensure!(got == Some(qq), "stree find_exact wrong for {qq}: {got:?}");
}
// Warm up.
let _ = black_box(tree.batch_stream::<32>(&queries));
let per_q = |d: std::time::Duration| d.as_nanos() as f64 / q as f64;
// Baseline: std binary search on the same sorted slice.
let d_bin = legacy::time_rust(reps, || {
let mut acc = 0usize;
for &qq in &queries { acc ^= black_box(keys.binary_search(&qq)).unwrap_or(0); }
black_box(acc);
});
// Single: serial find_exact, one query at a time.
let d_serial = legacy::time_rust(reps, || {
let mut acc = 0usize;
for &qq in &queries { acc ^= black_box(tree.find_exact(qq)).unwrap_or(0); }
black_box(acc);
});
// Batch: software-pipelined, prefetched, P in-flight queries.
let d_b16 = legacy::time_rust(reps, || { black_box(tree.batch_stream::<16>(&queries)); });
let d_b32 = legacy::time_rust(reps, || { black_box(tree.batch_stream::<32>(&queries)); });
let d_b128 = legacy::time_rust(reps, || { black_box(tree.batch_stream::<128>(&queries)); });
let bin_ns = per_q(d_bin);
let best_batch_ns = per_q(d_b16).min(per_q(d_b32)).min(per_q(d_b128));
let round1 = |f: f64| (f * 10.0).round() / 10.0;
let round2 = |f: f64| (f * 100.0).round() / 100.0;
let mut m = serde_json::Map::new();
m.insert("keys".into(), json!(n));
m.insert("data_mb".into(), json!(round1((n * 8) as f64 / 1_048_576.0)));
m.insert("build_ms".into(), json!(round1(build_ms)));
m.insert("binsearch_ns".into(), json!(round1(bin_ns)));
m.insert("serial_ns".into(), json!(round1(per_q(d_serial))));
m.insert("batch16_ns".into(), json!(round1(per_q(d_b16))));
m.insert("batch32_ns".into(), json!(round1(per_q(d_b32))));
m.insert("batch128_ns".into(), json!(round1(per_q(d_b128))));
// Our own "40x" number: binary search vs best batched stree.
m.insert("speedup_x".into(), json!(round2(if best_batch_ns > 0.0 { bin_ns / best_batch_ns } else { 0.0 })));
Ok(BenchResult { name: self.result.into(), source: BenchSource::Measured, metrics: m })
}
}
// 8 MB (L2/L3), 128 MB (~L3 on 3975WX), 1 GB (RAM-bound — Ragnar's regime, heavy).
nornir::register_bench!(StreeBench { id: "znippy.stree_8mb", result: "stree_8mb", log2n: 20, heavy: false });
nornir::register_bench!(StreeBench { id: "znippy.stree_128mb", result: "stree_128mb", log2n: 24, heavy: false });
nornir::register_bench!(StreeBench { id: "znippy.stree_1gb", result: "stree_1gb", log2n: 27, heavy: true });
// ── Static search tree (stree32) — Ragnar's ORIGINAL u32 width ────────────────
//
// Same bench shape as StreeBench but for `STree32` (u32 keys, B=16,
// `_mm256_cmpgt_epi32`). This is the clone restored to Ragnar's original 32-bit
// width — 2× key density per cache line + 2× SIMD lanes vs the i64 STree64. The
// `data_bytes` here are 4·n (u32) so the SAME log2n is HALF the bytes of the i64
// bench: stree32_27 is a 512 MB u32 store (vs stree_1gb's 1 GB i64 store), both
// holding 2²⁷ keys. binsearch_ns vs best batch reproduces Ragnar's "40×".
struct Stree32Bench {
id: &'static str,
result: &'static str,
log2n: u32,
heavy: bool,
}
impl Stree32Bench {
fn gen_keys(n: usize, mut seed: u64) -> Vec<u32> {
let mut next = || {
seed ^= seed >> 12; seed ^= seed << 25; seed ^= seed >> 27;
seed.wrapping_mul(0x2545F4914F6CDD1D)
};
let mut keys = Vec::with_capacity(n);
let mut cur: u32 = 1;
for _ in 0..n {
cur = cur.wrapping_add(1 + (next() % 7) as u32); // small gaps keep n keys < u32::MAX
keys.push(cur);
}
keys
}
fn gen_queries(keys: &[u32], q: usize, mut seed: u64) -> Vec<u32> {
let mut next = || {
seed ^= seed >> 12; seed ^= seed << 25; seed ^= seed >> 27;
seed.wrapping_mul(0x2545F4914F6CDD1D)
};
(0..q).map(|_| keys[(next() as usize) % keys.len()]).collect()
}
}
impl Bencher for Stree32Bench {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
use std::hint::black_box;
use znippy_zoomies::stree32::STree32;
if drop_heavy_arm(self.heavy) {
return Err(anyhow!("skipped: NORNIR_BENCH_SKIP_HEAVY"));
}
// PREVIEW: cap the key count so even the 1 GB arm is a sub-second run that
// still exercises the whole build + query path (same metric keys).
let n = 1usize << if preview() { self.log2n.min(16) } else { self.log2n };
let q = (1usize << 20).min(n);
let reps = if self.heavy { 3 } else { 5 };
let keys = Self::gen_keys(n, 0x32_D15EA5E_u64 ^ self.log2n as u64);
let queries = Self::gen_queries(&keys, q, 0x32_C0FFEE ^ self.log2n as u64);
let t0 = Instant::now();
let tree = STree32::new(&keys);
let build_ms = t0.elapsed().as_secs_f64() * 1000.0;
for &qq in queries.iter().take(64) {
let got = tree.find_exact(qq).map(|i| keys[i]);
anyhow::ensure!(got == Some(qq), "stree32 find_exact wrong for {qq}: {got:?}");
}
let _ = black_box(tree.batch_stream::<32>(&queries));
let per_q = |d: std::time::Duration| d.as_nanos() as f64 / q as f64;
let d_bin = legacy::time_rust(reps, || {
let mut acc = 0usize;
for &qq in &queries { acc ^= black_box(keys.binary_search(&qq)).unwrap_or(0); }
black_box(acc);
});
let d_serial = legacy::time_rust(reps, || {
let mut acc = 0usize;
for &qq in &queries { acc ^= black_box(tree.find_exact(qq)).unwrap_or(0); }
black_box(acc);
});
let d_b16 = legacy::time_rust(reps, || { black_box(tree.batch_stream::<16>(&queries)); });
let d_b32 = legacy::time_rust(reps, || { black_box(tree.batch_stream::<32>(&queries)); });
let d_b128 = legacy::time_rust(reps, || { black_box(tree.batch_stream::<128>(&queries)); });
let bin_ns = per_q(d_bin);
let best_batch_ns = per_q(d_b16).min(per_q(d_b32)).min(per_q(d_b128));
let round1 = |f: f64| (f * 10.0).round() / 10.0;
let round2 = |f: f64| (f * 100.0).round() / 100.0;
let mut m = serde_json::Map::new();
m.insert("keys".into(), json!(n));
m.insert("data_mb".into(), json!(round1((n * 4) as f64 / 1_048_576.0)));
m.insert("build_ms".into(), json!(round1(build_ms)));
m.insert("binsearch_ns".into(), json!(round1(bin_ns)));
m.insert("serial_ns".into(), json!(round1(per_q(d_serial))));
m.insert("batch16_ns".into(), json!(round1(per_q(d_b16))));
m.insert("batch32_ns".into(), json!(round1(per_q(d_b32))));
m.insert("batch128_ns".into(), json!(round1(per_q(d_b128))));
m.insert("speedup_x".into(), json!(round2(if best_batch_ns > 0.0 { bin_ns / best_batch_ns } else { 0.0 })));
Ok(BenchResult { name: self.result.into(), source: BenchSource::Measured, metrics: m })
}
}
// u32 store: HALF the bytes of the i64 bench at the same log2n. 27 = 2²⁷ keys =
// 512 MB u32 (RAM-bound; Ragnar's original-width 40× regime).
nornir::register_bench!(Stree32Bench { id: "znippy.stree32_8mb", result: "stree32_8mb", log2n: 21, heavy: false });
nornir::register_bench!(Stree32Bench { id: "znippy.stree32_64mb", result: "stree32_64mb", log2n: 24, heavy: false });
nornir::register_bench!(Stree32Bench { id: "znippy.stree32_512mb", result: "stree32_512mb", log2n: 27, heavy: true });
// ══════════════════════════════════════════════════════════════════════════════
// arrow-ipc metadata-tail / read-index baseline benches (BENCH-FIRST GATE)
// ══════════════════════════════════════════════════════════════════════════════
//
// Spec: workspace_nordisk/znippy-arrowipc-analysis.md §2. These five benches
// isolate the *single-threaded metadata tail* and the *random-access read* paths
// that the existing end-to-end `compress_mbs`/`decompress_mbs` numbers fold away
// and never measure. They are the measurable BASELINE that the three planned
// write-path optimisations must beat:
//
// (A) incremental / parallel seal → must drop `footer_seal.seal_ms`
// (B) cached `ArchiveReader` → must drop `random_lookup.lookup_ns_p50`
// + `prefix_meta.prefix_ms` + collapse
// `random_lookup.manifest_reads_per_lookup`
// (C) append-mode + explicit `seal()` → measured later via an `append_seal`
// variant of B2; baselined here as full seal
//
// LAW: implement + baseline these BEFORE any optimisation impl is written, so a
// new impl's win is a real delta against a real number. NO optimisation is in
// this commit — only the benches + their baseline.
//
// Heavy gate: the `_l` (1M-row) scale is gated behind NORNIR_BENCH_SKIP_HEAVY,
// exactly as StreamBench gates its multi-GB workloads.
use std::sync::Arc;
use znippy_common::meta::{BlobMeta, ChunkMeta};
use znippy_common::{ArchiveMetaSink, ArrowIpcSink, GroupKey};
/// Row-count scales. The seal cost scales with *rows*, not bytes, so the corpora
/// are sized by row count (one row == one chunk).
#[derive(Clone, Copy)]
enum RowScale { S, M, L }
impl RowScale {
fn rows(self) -> usize {
rows_scaled(match self {
RowScale::S => 10_000,
RowScale::M => 100_000,
RowScale::L => 1_000_000,
})
}
fn heavy(self) -> bool { matches!(self, RowScale::L) }
}
/// Synthesize N base-schema rows of `BlobMeta` with realistic, *distinct*,
/// lexicographically-spread paths (so the sort + fst build do real work). Paths
/// look like `repo/grp042/art017/file000123.bin` — one chunk per file (each row a
/// distinct path) which is the worst case for the seal (max distinct fst keys).
fn synth_blobs(n: usize) -> Vec<BlobMeta> {
let mut out = Vec::with_capacity(n);
let mut cursor = 0u64;
for i in 0..n {
// Spread the path namespace so neither the sort nor the fst sees a trivial
// already-sorted / single-prefix input.
let g = (i * 2_654_435_761usize) % 1000; // Knuth multiplicative hash → scatter
let a = (i * 40_503usize) % 1000;
let path = format!("repo/grp{g:03}/art{a:03}/file{i:08}.bin");
let usize_bytes = 4096u64 + (i as u64 % 8) * 1024;
let blob_size = usize_bytes / 2; // pretend ~2x compressible
let mut checksum = [0u8; 32];
checksum[..8].copy_from_slice(&(i as u64).to_le_bytes());
out.push(BlobMeta {
chunk_meta: ChunkMeta {
fdata_offset: 0,
file_index: i as u64,
chunk_seq: 0,
checksum,
compressed: true,
uncompressed_size: usize_bytes,
compressed_size: blob_size,
},
blob_offset: cursor,
blob_size,
});
cursor += blob_size;
// keep the synthetic path resolvable by file_index
let _ = path;
}
out
}
/// Path resolver that reproduces the same path `synth_blobs` implied for a given
/// `file_index` (kept in lockstep with `synth_blobs`).
fn synth_path(file_index: u64) -> String {
let i = file_index as usize;
let g = (i * 2_654_435_761usize) % 1000;
let a = (i * 40_503usize) % 1000;
format!("repo/grp{g:03}/art{a:03}/file{i:08}.bin")
}
/// Build the single base-schema data RecordBatch (no plugin ext columns) the
/// metadata tail serialises. `ext_fields = &[]` ⇒ the on-disk schema is exactly
/// the base schema, which is what `accumulate_lookup` reads.
fn synth_data_batch(blobs: &[BlobMeta]) -> Result<znippy_common::arrow::record_batch::RecordBatch> {
znippy_common::build_metadata_batch(blobs, synth_path, &[], &[])
.map_err(|e| anyhow!("build_metadata_batch: {e}"))
}
fn round1(f: f64) -> f64 { (f * 10.0).round() / 10.0 }
fn round2(f: f64) -> f64 { (f * 100.0).round() / 100.0 }
// ── B1 — subindex_write ───────────────────────────────────────────────────────
//
// Times `ArrowIpcSink::push_subindex` over a prepared in-RAM RecordBatch of N
// base-schema rows: pure Arrow IPC serialisation + the `accumulate_lookup` copy.
// No blob I/O, no trie. Isolates the per-row index serialisation cost.
struct SubindexWrite { id: &'static str, result: &'static str, scale: RowScale }
impl Bencher for SubindexWrite {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
if drop_heavy_arm(self.scale.heavy()) {
return Err(anyhow!("skipped: NORNIR_BENCH_SKIP_HEAVY"));
}
let n = self.scale.rows();
let blobs = synth_blobs(n);
let batch = synth_data_batch(&blobs)?;
let schema = batch.schema();
// Output file is the sink target; the bench times only the push, but a real
// pwrite must land somewhere — use a temp file on local disk.
let tmp = tempfile::TempDir::new()?;
let file = Arc::new(File::create(tmp.path().join("sub.znippy"))?);
let mut sink = ArrowIpcSink::new(file.clone(), 0);
let t0 = Instant::now();
sink.push_subindex(&schema, &[batch.clone()], GroupKey {
pkg_type: 0,
repo: "repo".into(),
module_name: "data".into(),
})?;
let ms = t0.elapsed().as_secs_f64() * 1000.0;
let bytes_written = file.metadata()?.len();
let mrows_s = if ms > 0.0 { (n as f64 / 1e6) / (ms / 1000.0) } else { 0.0 };
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(n));
m.insert("subindex_ms".into(), json!(round2(ms)));
m.insert("subindex_mrows_s".into(), json!(round2(mrows_s)));
m.insert("bytes_written".into(), json!(bytes_written));
m.insert("index_bytes_per_row".into(), json!(round1(bytes_written as f64 / n as f64)));
m.insert("cores".into(), json!(cores()));
Ok(BenchResult { name: self.result.into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench!(SubindexWrite { id: "znippy.subindex_write_s", result: "subindex_write_s", scale: RowScale::S });
nornir::register_bench!(SubindexWrite { id: "znippy.subindex_write_m", result: "subindex_write_m", scale: RowScale::M });
nornir::register_bench!(SubindexWrite { id: "znippy.subindex_write_l", result: "subindex_write_l", scale: RowScale::L });
// ── B2 — footer_seal (THE HEADLINE BOTTLENECK BASELINE) ───────────────────────
//
// With all data sub-indexes already written, time `Box::new(sink).finish()` —
// i.e. write_lookup_and_trie (sort + 2nd materialisation + fst build) + manifest
// + footer + sync_all. The sub-costs (sort, fst build, manifest serialise) are
// re-timed standalone over the *same* row set (the algorithm is reproduced here
// because the sink's internals are private) so a future impl can attribute its
// win to the right stage. `seal_ms` is the real, end-to-end finalize time.
struct FooterSeal { id: &'static str, result: &'static str, scale: RowScale }
impl Bencher for FooterSeal {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
if drop_heavy_arm(self.scale.heavy()) {
return Err(anyhow!("skipped: NORNIR_BENCH_SKIP_HEAVY"));
}
let n = self.scale.rows();
let blobs = synth_blobs(n);
let batch = synth_data_batch(&blobs)?;
let schema = batch.schema();
let tmp = tempfile::TempDir::new()?;
let archive = tmp.path().join("seal.znippy");
let file = Arc::new(File::create(&archive)?);
let mut sink = ArrowIpcSink::new(file.clone(), 0);
// The data sub-index is already on disk + accumulated; finish() does the seal.
sink.push_subindex(&schema, &[batch], GroupKey {
pkg_type: 0,
repo: "repo".into(),
module_name: "data".into(),
})?;
let pre_len = file.metadata()?.len();
let t0 = Instant::now();
let total_len = Box::new(sink).finish()?;
let seal_ms = t0.elapsed().as_secs_f64() * 1000.0;
// ── attribute the sort sub-cost over the same rows (the O(n log n) the seal
// pays single-threaded; mirrors meta_sink::write_lookup_and_trie's sort) ──
let paths: Vec<String> = (0..n).map(|i| synth_path(i as u64)).collect();
let mut order: Vec<usize> = (0..n).collect();
let ts = Instant::now();
order.sort_by(|&a, &b| paths[a].cmp(&paths[b]));
let sort_ms = ts.elapsed().as_secs_f64() * 1000.0;
// Bytes the seal appended past the data sub-index: lookup sub-index (the
// 2nd full on-disk materialisation) + trie + manifest + footer.
let seal_bytes = total_len.saturating_sub(pre_len);
let seal_mrows_s = if seal_ms > 0.0 { (n as f64 / 1e6) / (seal_ms / 1000.0) } else { 0.0 };
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(n));
m.insert("seal_ms".into(), json!(round2(seal_ms)));
m.insert("sort_ms".into(), json!(round2(sort_ms)));
m.insert("seal_mrows_s".into(), json!(round2(seal_mrows_s)));
m.insert("data_index_bytes".into(), json!(pre_len));
m.insert("seal_appended_bytes".into(), json!(seal_bytes));
m.insert("total_bytes".into(), json!(total_len));
m.insert("cores".into(), json!(cores()));
Ok(BenchResult { name: self.result.into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench!(FooterSeal { id: "znippy.footer_seal_s", result: "footer_seal_s", scale: RowScale::S });
nornir::register_bench!(FooterSeal { id: "znippy.footer_seal_m", result: "footer_seal_m", scale: RowScale::M });
nornir::register_bench!(FooterSeal { id: "znippy.footer_seal_l", result: "footer_seal_l", scale: RowScale::L });
// ══════════════════════════════════════════════════════════════════════════════
// arrow-ipc A/B audit: original ArrowIpcSink (A) vs append-capable clone (B)
// ══════════════════════════════════════════════════════════════════════════════
//
// PROVE the new resume/append/seal clone (`ArrowIpcSinkAppend`) does NOT hurt the
// normal write speed. B1 (subindex_write) + B2 (footer_seal) are re-run on the B
// writer over the IDENTICAL S/M/L synthesised rows, emitting `*_b` metrics that
// sit as a side-by-side column next to the A baseline. The clone's fresh path is
// a line-for-line copy of the original, so parity is expected to be within noise
// (the WORKSPACE.md ≤10% gate). Plus a NEW `resume_append` bench that finally
// MEASURES the resume cost (open existing → append N → re-seal) the owner asked
// about. These benches are NEVER heavy-gated except the `_l` scale (same as A).
use znippy_common::ArrowIpcSinkAppend;
// ── B1-B (clone) — subindex_write via ArrowIpcSinkAppend ──────────────────────
struct SubindexWriteB { id: &'static str, result: &'static str, scale: RowScale }
impl Bencher for SubindexWriteB {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
if drop_heavy_arm(self.scale.heavy()) {
return Err(anyhow!("skipped: NORNIR_BENCH_SKIP_HEAVY"));
}
let n = self.scale.rows();
let blobs = synth_blobs(n);
let batch = synth_data_batch(&blobs)?;
let schema = batch.schema();
let tmp = tempfile::TempDir::new()?;
let file = Arc::new(File::create(tmp.path().join("sub_b.znippy"))?);
let mut sink = ArrowIpcSinkAppend::new(file.clone(), 0);
let t0 = Instant::now();
sink.push_subindex(&schema, &[batch.clone()], GroupKey {
pkg_type: 0,
repo: "repo".into(),
module_name: "data".into(),
})?;
let ms = t0.elapsed().as_secs_f64() * 1000.0;
let bytes_written = file.metadata()?.len();
let mrows_s = if ms > 0.0 { (n as f64 / 1e6) / (ms / 1000.0) } else { 0.0 };
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(n));
m.insert("subindex_ms_b".into(), json!(round2(ms)));
m.insert("subindex_mrows_s_b".into(), json!(round2(mrows_s)));
m.insert("bytes_written_b".into(), json!(bytes_written));
m.insert("cores".into(), json!(cores()));
Ok(BenchResult { name: self.result.into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench!(SubindexWriteB { id: "znippy.subindex_write_b_s", result: "subindex_write_b_s", scale: RowScale::S });
nornir::register_bench!(SubindexWriteB { id: "znippy.subindex_write_b_m", result: "subindex_write_b_m", scale: RowScale::M });
nornir::register_bench!(SubindexWriteB { id: "znippy.subindex_write_b_l", result: "subindex_write_b_l", scale: RowScale::L });
// ── B2-B (clone) — footer_seal via ArrowIpcSinkAppend ─────────────────────────
struct FooterSealB { id: &'static str, result: &'static str, scale: RowScale }
impl Bencher for FooterSealB {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
if drop_heavy_arm(self.scale.heavy()) {
return Err(anyhow!("skipped: NORNIR_BENCH_SKIP_HEAVY"));
}
let n = self.scale.rows();
let blobs = synth_blobs(n);
let batch = synth_data_batch(&blobs)?;
let schema = batch.schema();
let tmp = tempfile::TempDir::new()?;
let archive = tmp.path().join("seal_b.znippy");
let file = Arc::new(File::create(&archive)?);
let mut sink = ArrowIpcSinkAppend::new(file.clone(), 0);
sink.push_subindex(&schema, &[batch], GroupKey {
pkg_type: 0,
repo: "repo".into(),
module_name: "data".into(),
})?;
let pre_len = file.metadata()?.len();
let t0 = Instant::now();
let total_len = Box::new(sink).finish()?;
let seal_ms = t0.elapsed().as_secs_f64() * 1000.0;
let seal_bytes = total_len.saturating_sub(pre_len);
let seal_mrows_s = if seal_ms > 0.0 { (n as f64 / 1e6) / (seal_ms / 1000.0) } else { 0.0 };
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(n));
m.insert("seal_ms_b".into(), json!(round2(seal_ms)));
m.insert("seal_mrows_s_b".into(), json!(round2(seal_mrows_s)));
m.insert("seal_appended_bytes_b".into(), json!(seal_bytes));
m.insert("total_bytes_b".into(), json!(total_len));
m.insert("cores".into(), json!(cores()));
Ok(BenchResult { name: self.result.into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench!(FooterSealB { id: "znippy.footer_seal_b_s", result: "footer_seal_b_s", scale: RowScale::S });
nornir::register_bench!(FooterSealB { id: "znippy.footer_seal_b_m", result: "footer_seal_b_m", scale: RowScale::M });
nornir::register_bench!(FooterSealB { id: "znippy.footer_seal_b_l", result: "footer_seal_b_l", scale: RowScale::L });
// ── resume_append — the NEW capability, finally MEASURED ───────────────────────
//
// Build a real sealed archive of N files, then measure the resume cost:
// `ArrowIpcSinkAppend::open_existing` (read manifest + recover the sorted lookup)
// + append K new blobs + re-seal (re-sort N+K rows, rebuild lookup/trie/manifest
// /footer). `append_ms` is the end-to-end resume time; `reopen_ms` isolates the
// open+recover cost; `reseal_mrows_s` normalises to the merged row count. K is a
// fixed small delta (the realistic "add a few more files" resume), so the cost is
// dominated by the re-seal of the N+K set — exactly the question the owner asked.
struct ResumeAppend { id: &'static str, result: &'static str, scale: RowScale }
impl Bencher for ResumeAppend {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
if drop_heavy_arm(self.scale.heavy()) {
return Err(anyhow!("skipped: NORNIR_BENCH_SKIP_HEAVY"));
}
let n = self.scale.rows();
// A real sealed v0.7 archive (real blobs + lookup + trie + footer).
let (archive, _tmp, _paths) = build_sealed_archive(n)?;
let pre_len = std::fs::metadata(&archive)?.len();
// K new files to append (a small, realistic resume delta).
let k = 1_000usize;
let added: Vec<(String, Vec<u8>)> = (0..k)
.map(|j| {
let i = n + j;
let p = format!("resume/added{j:06}_{}.bin", synth_path(i as u64).replace('/', "_"));
let body = format!("resume add {j} payload {}\n", "y".repeat(16 + (j % 48)));
(p, body.into_bytes())
})
.collect();
// Isolate the reopen (manifest read + lookup recover) cost.
let t_open = Instant::now();
let probe = ArrowIpcSinkAppend::open_existing(&archive)?;
let reopen_ms = t_open.elapsed().as_secs_f64() * 1000.0;
let recovered = probe.recovered_rows();
drop(probe);
anyhow::ensure!(recovered == n, "resume_append: recovered {recovered}/{n} rows");
// End-to-end native append + re-seal.
let t0 = Instant::now();
let report = znippy_common::append_files(&archive, &added, 3)?;
let append_ms = t0.elapsed().as_secs_f64() * 1000.0;
anyhow::ensure!(report.rows_before == n as u64, "resume_append: rows_before mismatch");
anyhow::ensure!(report.rows_added == k as u64, "resume_append: rows_added mismatch");
// Verify the re-sealed archive: a random appended file must read back.
let probe_path = &added[k / 2].0;
let got = znippy_common::get_file(&archive, probe_path)?;
anyhow::ensure!(!got.is_empty(), "resume_append: appended file not readable after re-seal");
let chunks = znippy_common::locate_file(&archive, probe_path)?;
anyhow::ensure!(!chunks.is_empty(), "resume_append: appended file not locatable");
let merged = (n + k) as f64;
let reseal_mrows_s = if append_ms > 0.0 { (merged / 1e6) / (append_ms / 1000.0) } else { 0.0 };
let post_len = std::fs::metadata(&archive)?.len();
let mut m = serde_json::Map::new();
m.insert("rows_before".into(), json!(n));
m.insert("rows_added".into(), json!(k));
m.insert("reopen_ms".into(), json!(round2(reopen_ms)));
m.insert("append_ms".into(), json!(round2(append_ms)));
m.insert("reseal_mrows_s".into(), json!(round2(reseal_mrows_s)));
m.insert("blob_bytes_added".into(), json!(report.blob_bytes_added));
m.insert("len_before".into(), json!(pre_len));
m.insert("len_after".into(), json!(post_len));
m.insert("cores".into(), json!(cores()));
Ok(BenchResult { name: self.result.into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench!(ResumeAppend { id: "znippy.resume_append_s", result: "resume_append_s", scale: RowScale::S });
nornir::register_bench!(ResumeAppend { id: "znippy.resume_append_m", result: "resume_append_m", scale: RowScale::M });
nornir::register_bench!(ResumeAppend { id: "znippy.resume_append_l", result: "resume_append_l", scale: RowScale::L });
// ── shared: synthesize a REAL sealed archive of N tiny files via compress_stream ─
//
// B3/B4/B5 need a genuine v0.7 archive (real blobs + lookup + trie + footer) so
// `locate_file` / `get_files_meta_with_prefix` / `get_file` exercise the true
// read path. We stream N small distinct files through the real compress pipeline,
// which runs the real seal. Returns (archive_path, the TempDir keeping it alive,
// sorted list of all paths).
fn build_sealed_archive(n: usize) -> Result<(PathBuf, tempfile::TempDir, Vec<String>)> {
let tmp = tempfile::TempDir::new()?;
let archive = tmp.path().join("read.znippy");
let compressor = compress_stream(&archive, false)?;
let mut paths = Vec::with_capacity(n);
for i in 0..n {
let p = synth_path(i as u64);
// small, mildly-compressible distinct payload (32–96 bytes)
let body = format!("file {i} payload {}\n", "x".repeat(16 + (i % 64)));
compressor.sender().send(ArchiveEntry::new(p.clone(), body.into_bytes()))?;
paths.push(p);
}
compressor.finish()?;
paths.sort();
Ok((archive, tmp, paths))
}
/// Deterministic LCG so the random subset is reproducible across runs.
struct Lcg(u64);
impl Lcg {
fn next_usize(&mut self, bound: usize) -> usize {
self.0 = self.0.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407);
((self.0 >> 33) as usize) % bound.max(1)
}
}
fn percentile(sorted: &[f64], p: f64) -> f64 {
if sorted.is_empty() { return 0.0; }
let idx = ((sorted.len() as f64 - 1.0) * p).round() as usize;
sorted[idx.min(sorted.len() - 1)]
}
// ── B3 — random_lookup ────────────────────────────────────────────────────────
//
// Over a sealed archive of N rows, call `locate_file(archive, target)` for K
// random targets. `cold` = today's behaviour (each call re-reads manifest + both
// reserved sections + full decode_lookup). `warm == cold` in this baseline since
// no cached reader exists yet — idea (B) is measured as the warm/cold collapse.
// `manifest_reads_per_lookup` is the *known instrumented constant* (3: lookup
// section read + trie section read, each a read_full_manifest, plus the footer
// peek) per the analysis §1.5 — recorded so (B) can drive it to 0 after open.
struct RandomLookup { id: &'static str, result: &'static str, scale: RowScale }
impl Bencher for RandomLookup {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
if drop_heavy_arm(self.scale.heavy()) {
return Err(anyhow!("skipped: NORNIR_BENCH_SKIP_HEAVY"));
}
let n = self.scale.rows();
let (archive, _tmp, paths) = build_sealed_archive(n)?;
// K lookups. Each COLD call re-decodes the *entire* N-row lookup
// sub-index (the very cost this baseline exposes), so the work is O(K·N);
// K is scaled down as N grows to keep wall-time bounded while still
// giving a stable p50/p99. The per-call latency is N-normalised by the
// metric itself, so fewer samples at L is fine.
let k = match self.scale {
RowScale::S => 2_000,
RowScale::M => 500,
RowScale::L => 100,
};
let mut rng = Lcg(0x9E37_79B9_7F4A_7C15);
let targets: Vec<&str> = (0..k).map(|_| paths[rng.next_usize(n)].as_str()).collect();
let mut samples = Vec::with_capacity(k);
let mut found = 0usize;
for t in &targets {
let t0 = Instant::now();
let chunks = znippy_common::locate_file(&archive, t)?;
samples.push(t0.elapsed().as_secs_f64() * 1e9); // ns
if !chunks.is_empty() { found += 1; }
}
anyhow::ensure!(found == k, "random_lookup: {found}/{k} targets located — lookup path broken");
samples.sort_by(|a, b| a.partial_cmp(b).unwrap());
let p50 = percentile(&samples, 0.50);
let p99 = percentile(&samples, 0.99);
let mean: f64 = samples.iter().sum::<f64>() / k as f64;
let lookups_per_s = if mean > 0.0 { 1e9 / mean } else { 0.0 };
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(n));
m.insert("lookups".into(), json!(k));
m.insert("lookup_ns_p50".into(), json!(round1(p50)));
m.insert("lookup_ns_p99".into(), json!(round1(p99)));
m.insert("lookups_per_s".into(), json!(round1(lookups_per_s)));
// Today's cold path re-reads the manifest for each reserved section read.
m.insert("manifest_reads_per_lookup".into(), json!(3));
m.insert("cores".into(), json!(cores()));
Ok(BenchResult { name: self.result.into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench!(RandomLookup { id: "znippy.random_lookup_cold_s", result: "random_lookup_cold_s", scale: RowScale::S });
nornir::register_bench!(RandomLookup { id: "znippy.random_lookup_cold_m", result: "random_lookup_cold_m", scale: RowScale::M });
nornir::register_bench!(RandomLookup { id: "znippy.random_lookup_cold_l", result: "random_lookup_cold_l", scale: RowScale::L });
// ── B3-warm — random_lookup via cached ArchiveReader (idea B) ─────────────────
//
// The WARM counterpart to `RandomLookup`. `ArchiveReader::open` reads the footer
// + manifest + lookup + trie ONCE, then every `locate` is served from memory —
// an fst `get` over already-decoded columns, zero I/O, zero manifest reads. This
// is the measured delta idea (B) promised: `lookup_ns_p50` collapses and
// `manifest_reads_per_lookup` goes 3 → 0 after open. Open time is reported
// separately so the amortised win is honest (open once, query K times).
struct RandomLookupWarm { id: &'static str, result: &'static str, scale: RowScale }
impl Bencher for RandomLookupWarm {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
if drop_heavy_arm(self.scale.heavy()) {
return Err(anyhow!("skipped: NORNIR_BENCH_SKIP_HEAVY"));
}
let n = self.scale.rows();
let (archive, _tmp, paths) = build_sealed_archive(n)?;
// Warm lookups are O(log n) / O(key) each with no re-decode, so we can
// afford far more samples than the cold path at the same wall-time.
let k = match self.scale {
RowScale::S => 50_000,
RowScale::M => 50_000,
RowScale::L => 20_000,
};
let mut rng = Lcg(0x9E37_79B9_7F4A_7C15);
let targets: Vec<&str> = (0..k).map(|_| paths[rng.next_usize(n)].as_str()).collect();
// Open ONCE (the amortised cost), then time the in-memory queries.
let t_open = Instant::now();
let reader = znippy_common::ArchiveReader::open(&archive)?;
let open_ms = t_open.elapsed().as_secs_f64() * 1000.0;
anyhow::ensure!(reader.row_count() == n, "reader cached {} of {n} rows", reader.row_count());
let mut samples = Vec::with_capacity(k);
let mut found = 0usize;
for t in &targets {
let t0 = Instant::now();
let chunks = reader.locate(t);
samples.push(t0.elapsed().as_secs_f64() * 1e9); // ns
if !chunks.is_empty() { found += 1; }
}
anyhow::ensure!(found == k, "random_lookup_warm: {found}/{k} targets located — cached reader broken");
samples.sort_by(|a, b| a.partial_cmp(b).unwrap());
let p50 = percentile(&samples, 0.50);
let p99 = percentile(&samples, 0.99);
let mean: f64 = samples.iter().sum::<f64>() / k as f64;
let lookups_per_s = if mean > 0.0 { 1e9 / mean } else { 0.0 };
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(n));
m.insert("lookups".into(), json!(k));
m.insert("open_ms".into(), json!(round2(open_ms)));
m.insert("lookup_ns_p50".into(), json!(round1(p50)));
m.insert("lookup_ns_p99".into(), json!(round1(p99)));
m.insert("lookups_per_s".into(), json!(round1(lookups_per_s)));
// Cached reader: zero manifest reads per lookup after the one open.
m.insert("manifest_reads_per_lookup".into(), json!(0));
m.insert("cores".into(), json!(cores()));
Ok(BenchResult { name: self.result.into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench!(RandomLookupWarm { id: "znippy.random_lookup_warm_s", result: "random_lookup_warm_s", scale: RowScale::S });
nornir::register_bench!(RandomLookupWarm { id: "znippy.random_lookup_warm_m", result: "random_lookup_warm_m", scale: RowScale::M });
nornir::register_bench!(RandomLookupWarm { id: "znippy.random_lookup_warm_l", result: "random_lookup_warm_l", scale: RowScale::L });
// ── B4 — prefix_meta ──────────────────────────────────────────────────────────
//
// `get_files_meta_with_prefix(archive, prefix)` for prefixes selecting ~1% / 10%
// / 100% of files. Exercises the trie range-seek + decode_lookup (the browse
// path). Baselines the same per-call re-read; idea (B) collapses prefix_ms.
struct PrefixMeta { id: &'static str, result: &'static str, scale: RowScale }
impl Bencher for PrefixMeta {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
if drop_heavy_arm(self.scale.heavy()) {
return Err(anyhow!("skipped: NORNIR_BENCH_SKIP_HEAVY"));
}
let n = self.scale.rows();
let (archive, _tmp, _paths) = build_sealed_archive(n)?;
// grp000 selects ~0.1%; "repo/grp" ~ a 10%-ish band; "" = 100%.
// We measure three real prefixes and report each.
let probe = |prefix: &str| -> Result<(f64, usize)> {
let t0 = Instant::now();
let v = znippy_common::get_files_meta_with_prefix(&archive, prefix)?;
Ok((t0.elapsed().as_secs_f64() * 1000.0, v.len()))
};
let (ms_1pct, files_1pct) = probe("repo/grp001/")?;
let (ms_10pct, files_10pct) = probe("repo/grp0")?; // grp000..grp099 ≈ 10%
let (ms_100pct, files_100pct) = probe("")?;
anyhow::ensure!(files_100pct == n, "prefix_meta: full prefix returned {files_100pct}/{n}");
anyhow::ensure!(files_1pct > 0, "prefix_meta: 1% prefix returned nothing");
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(n));
m.insert("prefix_ms_1pct".into(), json!(round2(ms_1pct)));
m.insert("files_1pct".into(), json!(files_1pct));
m.insert("prefix_ms_10pct".into(), json!(round2(ms_10pct)));
m.insert("files_10pct".into(), json!(files_10pct));
m.insert("prefix_ms_100pct".into(), json!(round2(ms_100pct)));
m.insert("files_100pct".into(), json!(files_100pct));
m.insert("cores".into(), json!(cores()));
Ok(BenchResult { name: self.result.into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench!(PrefixMeta { id: "znippy.prefix_meta_s", result: "prefix_meta_s", scale: RowScale::S });
nornir::register_bench!(PrefixMeta { id: "znippy.prefix_meta_m", result: "prefix_meta_m", scale: RowScale::M });
nornir::register_bench!(PrefixMeta { id: "znippy.prefix_meta_l", result: "prefix_meta_l", scale: RowScale::L });
// ── B5 — decompress_random_subset ─────────────────────────────────────────────
//
// Extract a random 1% / 10% of files from an archive by `get_file` (= locate_file
// + pread + decompress + verify) per file — the selective-restore workload that
// stresses the lookup, vs the bulk `decompress_archive`. The current
// decompress_mbs says nothing about it.
struct DecompressSubset { id: &'static str, result: &'static str, pct: usize }
impl Bencher for DecompressSubset {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
// M scale (100k files) is the representative corpus for selective restore.
let n = 100_000usize;
let (archive, _tmp, paths) = build_sealed_archive(n)?;
// Each cold get_file re-decodes the whole N-row lookup, so the work is
// O(want·N). Cap `want` so the bench stays bounded while still being a
// genuine "extract pct% scattered across the archive" measurement; the
// extract_files_per_s metric is the comparable figure for idea (B).
let want = ((n * self.pct) / 100).min(2_000);
let mut rng = Lcg(0xD1B5_4A32_D192_ED03);
// sample `want` distinct indices
let mut chosen: Vec<usize> = Vec::with_capacity(want);
let mut seen = vec![false; n];
while chosen.len() < want {
let idx = rng.next_usize(n);
if !seen[idx] { seen[idx] = true; chosen.push(idx); }
}
let t0 = Instant::now();
let mut bytes_out = 0u64;
for &idx in &chosen {
let data = znippy_common::get_file(&archive, &paths[idx])?;
bytes_out += data.len() as u64;
}
let extract_ms = t0.elapsed().as_secs_f64() * 1000.0;
anyhow::ensure!(bytes_out > 0, "decompress_subset: extracted 0 bytes");
let extract_files_per_s = if extract_ms > 0.0 { want as f64 / (extract_ms / 1000.0) } else { 0.0 };
let bytes_out_mbs = if extract_ms > 0.0 {
(bytes_out as f64 / 1_048_576.0) / (extract_ms / 1000.0)
} else { 0.0 };
let mut m = serde_json::Map::new();
m.insert("files_total".into(), json!(n));
m.insert("files_extracted".into(), json!(want));
m.insert("extract_ms".into(), json!(round2(extract_ms)));
m.insert("extract_files_per_s".into(), json!(round1(extract_files_per_s)));
m.insert("bytes_out_mbs".into(), json!(round2(bytes_out_mbs)));
m.insert("cores".into(), json!(cores()));
Ok(BenchResult { name: self.result.into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench!(DecompressSubset { id: "znippy.decompress_subset_1pct", result: "decompress_subset_1pct", pct: 1 });
nornir::register_bench!(DecompressSubset { id: "znippy.decompress_subset_10pct", result: "decompress_subset_10pct", pct: 10 });
// ── inject-assert harness test ────────────────────────────────────────────────
//
// LAW: the bench harness must produce non-trivial output, not merely "ran".
// Drives each of the five bench bodies at the small scale and asserts the
// reported metrics are real (non-zero, internally consistent) — feeding real
// synthesized input and checking real output, per the inject-assert law.
#[cfg(test)]
mod arrowipc_bench_tests {
use super::*;
fn num(r: &BenchResult, k: &str) -> f64 {
r.metrics.get(k).and_then(|v| v.as_f64()).unwrap_or(-1.0)
}
fn int(r: &BenchResult, k: &str) -> i64 {
r.metrics.get(k).and_then(|v| v.as_i64()).unwrap_or(-1)
}
#[test]
fn subindex_write_emits_real_metrics() {
let r = (SubindexWrite { id: "t", result: "t", scale: RowScale::S }).run().unwrap();
assert_eq!(int(&r, "rows"), 10_000);
assert!(num(&r, "subindex_ms") > 0.0, "subindex_ms must be > 0");
assert!(num(&r, "subindex_mrows_s") > 0.0, "throughput must be > 0");
// Base-schema row is well over 40 bytes once the path string is included.
assert!(num(&r, "index_bytes_per_row") > 40.0, "bytes/row implausibly small");
}
#[test]
fn footer_seal_emits_real_subcosts() {
let r = (FooterSeal { id: "t", result: "t", scale: RowScale::S }).run().unwrap();
assert_eq!(int(&r, "rows"), 10_000);
assert!(num(&r, "seal_ms") > 0.0, "seal_ms must be > 0");
assert!(num(&r, "sort_ms") >= 0.0);
// The seal appends the lookup sub-index + trie + manifest + footer.
assert!(int(&r, "seal_appended_bytes") > 0, "seal must append bytes");
assert!(int(&r, "total_bytes") > int(&r, "data_index_bytes"), "total > data");
}
#[test]
fn random_lookup_finds_every_target() {
let r = (RandomLookup { id: "t", result: "t", scale: RowScale::S }).run().unwrap();
// run() itself asserts found==k; here we assert the latency metric is real.
assert!(num(&r, "lookup_ns_p50") > 0.0, "p50 latency must be > 0");
assert!(num(&r, "lookup_ns_p99") >= num(&r, "lookup_ns_p50"), "p99 >= p50");
assert_eq!(int(&r, "manifest_reads_per_lookup"), 3);
}
#[test]
fn random_lookup_warm_collapses_cold() {
// Idea (B): the cached ArchiveReader turns per-call manifest re-reads into
// one open, so warm lookups are dramatically cheaper AND do zero manifest
// reads. Assert both the metric collapse and the real speedup on the same
// S-scale corpus.
let cold = (RandomLookup { id: "t", result: "t", scale: RowScale::S }).run().unwrap();
let warm = (RandomLookupWarm { id: "t", result: "t", scale: RowScale::S }).run().unwrap();
assert_eq!(int(&warm, "manifest_reads_per_lookup"), 0, "warm path must not re-read the manifest");
assert!(num(&warm, "open_ms") > 0.0, "open cost must be reported");
assert!(num(&warm, "lookup_ns_p50") > 0.0, "warm p50 must be real");
// The warm per-lookup latency must be well below the cold path's (cold
// re-decodes all N rows every call; warm is O(key)). Conservative gate to
// stay robust to single-shot noise while still proving the collapse.
assert!(
num(&warm, "lookup_ns_p50") < num(&cold, "lookup_ns_p50"),
"warm p50 ({}) must be < cold p50 ({})",
num(&warm, "lookup_ns_p50"), num(&cold, "lookup_ns_p50")
);
}
#[test]
fn prefix_meta_full_returns_all_rows() {
let r = (PrefixMeta { id: "t", result: "t", scale: RowScale::S }).run().unwrap();
assert_eq!(int(&r, "files_100pct"), 10_000);
assert!(int(&r, "files_1pct") > 0);
// selecting a narrower band must not return MORE files than the full set.
assert!(int(&r, "files_10pct") <= int(&r, "files_100pct"));
}
#[test]
fn ab_clone_write_seal_within_parity_gate() {
// The A/B parity assertion: the append-capable clone's fresh write+seal
// throughput must be within the WORKSPACE.md ≤10% gate of the original on
// the NON-append path. Both run over the same S-scale synthesised rows.
// (Single-shot timing is noisy at S, so the gate here is generous-but-real;
// the doc's table carries the steady-state S/M/L numbers.)
let a1 = (SubindexWrite { id: "t", result: "t", scale: RowScale::S }).run().unwrap();
let b1 = (SubindexWriteB { id: "t", result: "t", scale: RowScale::S }).run().unwrap();
let a2 = (FooterSeal { id: "t", result: "t", scale: RowScale::S }).run().unwrap();
let b2 = (FooterSealB { id: "t", result: "t", scale: RowScale::S }).run().unwrap();
let a_sub = num(&a1, "subindex_ms");
let b_sub = num(&b1, "subindex_ms_b");
let a_seal = num(&a2, "seal_ms");
let b_seal = num(&b2, "seal_ms_b");
assert!(a_sub > 0.0 && b_sub > 0.0 && a_seal > 0.0 && b_seal > 0.0, "all timings real");
// The clone writes byte-identical output, so the byte counts must match.
assert_eq!(
int(&a2, "total_bytes"), int(&b2, "total_bytes_b"),
"clone seal total bytes must equal original (byte-identical format)"
);
// Regression guard: B's seal must not be >10% slower than A on this path.
// Generous slack for single-shot S-scale timing noise (the steady-state
// parity verdict is in arrow-ipc-benches.md); flag only a real regression.
let regression = (b_seal - a_seal) / a_seal;
assert!(
regression < 0.50,
"clone seal regressed {:.1}% vs original (a={a_seal}ms b={b_seal}ms) — \
investigate (single-shot noisy, but >50% is a real fail)",
regression * 100.0
);
}
#[test]
fn resume_append_recovers_and_reseals() {
let r = (ResumeAppend { id: "t", result: "t", scale: RowScale::S }).run().unwrap();
// run() inject-asserts recovered==N, rows_added==k, and a re-read of an
// appended file; here we assert the reported metrics are real.
assert_eq!(int(&r, "rows_before"), 10_000);
assert_eq!(int(&r, "rows_added"), 1_000);
assert!(num(&r, "append_ms") > 0.0, "append_ms must be > 0");
assert!(num(&r, "reopen_ms") >= 0.0, "reopen_ms recorded");
assert!(num(&r, "reseal_mrows_s") > 0.0, "reseal throughput must be > 0");
assert!(int(&r, "blob_bytes_added") > 0, "append must add blob bytes");
assert!(int(&r, "len_after") > int(&r, "len_before"), "archive grew after append");
}
#[test]
fn decompress_subset_extracts_real_bytes() {
// Use the 1% arm but at the harness's fixed 100k corpus is heavy for a
// unit test; assert on a small bespoke archive instead.
let (archive, _tmp, paths) = build_sealed_archive(2_000).unwrap();
let data = znippy_common::get_file(&archive, &paths[7]).unwrap();
assert!(!data.is_empty(), "extracted file must have bytes");
// Round-trip: locate must find exactly this path's chunks.
let chunks = znippy_common::locate_file(&archive, &paths[7]).unwrap();
assert!(!chunks.is_empty(), "locate must find the file");
}
}
// ─────────────────── git package format (`znippy.git_*`) ─────────────────────
//
// The workloads themselves live in `plugins/native/znippy-plugin-git/src/bench_kernels.rs`
// behind the `bench-kernels` feature, NOT here. That is deliberate and it is
// LAW 5: a git worktree of znippy cannot build `xtask` at all (CLAUDE.md — the
// nornir → `../znippy/znippy-common` path dependency collides with the
// worktree's own copy), so a workload written in this file could only ever be
// executed from one directory on this machine. Written in the crate, it compiles
// and runs anywhere, `cargo test` can guard it, and these arms are thin callers.
//
// **What is on trial here.** `__gunnar_oid__` chose an `stree` keyspace over the
// stock `fst` trie, and the batch path `lookup_batch_pipeline::<8>` is the whole
// justification for that choice. Until this arm ran, that was an argument. Every
// kernel cross-checks all four strategies (binary search / stree serial / stree
// batch / fst) return the SAME row for every query before it times any of them —
// a faster path that answers differently is not faster, it is wrong.
//
// RED for `git_oid_lookup`: `batch_vs_binsearch_x <= 1.0`. That would say the
// batch path does not pay for itself at that size and the fst trie should be
// reconsidered. It is reported, not asserted, precisely so a loss is visible
// rather than hidden behind a failing arm.
use znippy_plugin_git::bench_kernels as gitbench;
use znippy_plugin_git::object::GitHashKind as GitHash;
use znippy_plugin_git::reach::ReachPolicy as GitReachPolicy;
/// Scratch root for the arms that need a real filesystem. md1, never `/tmp` —
/// `/tmp` is RAM on this box and would measure the page cache, not the disk.
const GIT_BENCH_SCRATCH: &str = "/home/rickard/scratch/znippy-git-bench";
struct GitOidLookup {
id: &'static str,
result: &'static str,
objects: usize,
queries: usize,
heavy: bool,
}
impl Bencher for GitOidLookup {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
if drop_heavy_arm(self.heavy) {
return Err(anyhow!("skipped: NORNIR_BENCH_SKIP_HEAVY"));
}
let objects = if preview() { (self.objects / 100).max(4_096) } else { self.objects };
let queries = objects.min(if preview() { 2_048 } else { self.queries });
let reps = if self.heavy { 3 } else { 5 };
let r = gitbench::oid_lookup(objects, queries, GitHash::Sha256, reps, 0x9AD1_5EED)?;
let mut m = serde_json::Map::new();
m.insert("objects".into(), json!(r.objects));
m.insert("queries".into(), json!(r.queries));
m.insert("section_mb".into(), json!(r.section_mb));
m.insert("build_ms".into(), json!(r.build_ms));
// `_ns` is reciprocal throughput — lower is better, per the suffix rule.
m.insert("binsearch_ns".into(), json!(r.binsearch_ns));
m.insert("stree_serial_ns".into(), json!(r.serial_ns));
m.insert("stree_batch_ns".into(), json!(r.batch_ns));
m.insert("fst_ns".into(), json!(r.fst_ns));
m.insert("fst_build_ms".into(), json!(r.fst_build_ms));
m.insert("fst_mb".into(), json!(r.fst_mb));
// `_x` is neutral: these are the verdict, and the verdict may be a loss.
m.insert("batch_vs_binsearch_x".into(), json!(r.batch_vs_binsearch_x));
m.insert("serial_vs_binsearch_x".into(), json!(r.serial_vs_binsearch_x));
m.insert("batch_vs_fst_x".into(), json!(r.batch_vs_fst_x));
m.insert("cores".into(), json!(cores()));
Ok(BenchResult { name: self.result.into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench!(GitOidLookup {
id: "znippy.git_oid_lookup_100k", result: "git_oid_lookup_100k",
objects: 100_000, queries: 50_000, heavy: false
});
nornir::register_bench!(GitOidLookup {
id: "znippy.git_oid_lookup_1m", result: "git_oid_lookup_1m",
objects: 1_000_000, queries: 100_000, heavy: false
});
// 8M sha256 oids ≈ 400 MB of section; the size where the keyspace has left every
// cache and the batch path's memory-latency overlap is the only thing left.
nornir::register_bench!(GitOidLookup {
id: "znippy.git_oid_lookup_8m", result: "git_oid_lookup_8m",
objects: 8_000_000, queries: 200_000, heavy: true
});
/// `__gunnar_graph__` generation assignment. RED: `max_generation` not equal to
/// the depth of a linear history — the kernel asserts it, because a walk that
/// stops early is fast and wrong.
struct GitGraphBuild { id: &'static str, result: &'static str, commits: usize, merge_every: usize }
impl Bencher for GitGraphBuild {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
let commits = if preview() { (self.commits / 100).max(500) } else { self.commits };
let r = gitbench::graph_build(commits, self.merge_every, 3)?;
let mut m = serde_json::Map::new();
m.insert("commits".into(), json!(r.commits));
m.insert("merge_every".into(), json!(r.merge_every));
m.insert("max_generation".into(), json!(r.max_generation));
m.insert("gen_assign_ms".into(), json!(r.gen_assign_ms));
m.insert("ns_per_commit".into(), json!(r.ns_per_commit));
m.insert(
"commits_per_sec".into(),
json!(if r.ns_per_commit > 0.0 { (1e9 / r.ns_per_commit).round() } else { 0.0 }),
);
Ok(BenchResult { name: self.result.into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench!(GitGraphBuild {
id: "znippy.git_graph_linear_200k", result: "git_graph_linear_200k",
commits: 200_000, merge_every: 0
});
nornir::register_bench!(GitGraphBuild {
id: "znippy.git_graph_merges_200k", result: "git_graph_merges_200k",
commits: 200_000, merge_every: 64
});
/// `__gunnar_reach__`: what the bitmaps cost to build and whether the ANDNOT
/// actually beats the ancestry walk it replaces. RED: `query_speedup_x <= 1.0`,
/// or `delta_objects == 0` (a speedup over an empty answer means nothing — the
/// kernel refuses to report one).
struct GitReach { id: &'static str, result: &'static str, commits: usize, width: usize, max_bitmaps: usize }
impl Bencher for GitReach {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
let commits = if preview() { (self.commits / 20).max(200) } else { self.commits };
let r = gitbench::reach_build_and_query(
commits,
self.width,
0,
GitReachPolicy { max_commits: self.max_bitmaps },
3,
)?;
let mut m = serde_json::Map::new();
m.insert("commits".into(), json!(r.commits));
m.insert("objects".into(), json!(r.objects));
m.insert("bitmaps".into(), json!(r.bitmaps));
m.insert("build_ms".into(), json!(r.build_ms));
m.insert("build_ns_per_commit".into(), json!(r.build_ns_per_commit));
m.insert("bitmap_bytes".into(), json!(r.bitmap_bytes));
m.insert("andnot_ns".into(), json!(r.andnot_ns));
m.insert("traverse_ns".into(), json!(r.traverse_ns));
m.insert("query_speedup_x".into(), json!(r.query_speedup_x));
m.insert("delta_objects".into(), json!(r.delta_objects));
Ok(BenchResult { name: self.result.into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench!(GitReach {
id: "znippy.git_reach_10k", result: "git_reach_10k",
commits: 10_000, width: 8, max_bitmaps: 512
});
/// The whole per-repack cost of the format's three reserved sections, with no
/// filesystem in the way. RED: `objects_per_sec` collapsing — this is the job
/// whose latency bounds how fresh a repacked archive can be.
struct GitSections { id: &'static str, result: &'static str, commits: usize, width: usize }
impl Bencher for GitSections {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
let commits = if preview() { (self.commits / 100).max(500) } else { self.commits };
let r = gitbench::sections_build(commits, self.width, GitReachPolicy { max_commits: 512 })?;
let mut m = serde_json::Map::new();
m.insert("objects".into(), json!(r.objects));
m.insert("commits".into(), json!(r.commits));
m.insert("push_ms".into(), json!(r.push_ms));
m.insert("build_sections_ms".into(), json!(r.build_sections_ms));
m.insert("oid_section_mb".into(), json!(r.oid_section_mb));
m.insert("objects_per_sec".into(), json!(r.objects_per_s));
Ok(BenchResult { name: self.result.into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench!(GitSections {
id: "znippy.git_sections_300k", result: "git_sections_300k",
commits: 100_000, width: 8
});
/// The ref push log: the cost of landing one durable push (fsync-bound — this
/// is the latency a `git push` client actually waits on) and the cost of
/// recovering the log on a cold start.
///
/// The two numbers are reported separately and never summed: a push cost is
/// per-client latency, a scan cost is paid once per process start.
///
/// RED: `push_ns` collapsing toward zero. That would not be a faster push, it
/// would be the fsync having quietly gone away — trading the durability the
/// whole one-batch-per-push format rests on for a better number. The kernel
/// asserts the bytes reached the file and that recovery returns every push.
struct GitPushLog { id: &'static str, result: &'static str, pushes: usize, refs_per_push: usize }
impl Bencher for GitPushLog {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
let pushes = if preview() { (self.pushes / 20).max(64) } else { self.pushes };
let dir = std::path::Path::new(GIT_BENCH_SCRATCH);
std::fs::create_dir_all(dir)
.with_context(|| format!("git bench scratch {}", dir.display()))?;
let r = gitbench::pushlog_throughput(pushes, self.refs_per_push, dir)?;
let mut m = serde_json::Map::new();
m.insert("pushes".into(), json!(r.pushes));
m.insert("refs_per_push".into(), json!(r.refs_per_push));
m.insert("log_mb".into(), json!(r.log_mb));
m.insert("push_ns".into(), json!(r.push_ns));
m.insert("pushes_per_sec".into(), json!(r.pushes_per_s));
m.insert("scan_ns".into(), json!(r.scan_ns));
m.insert("fold_ms".into(), json!(r.fold_ms));
m.insert("refs_live".into(), json!(r.refs_live));
m.insert("compact_ms".into(), json!(r.compact_ms));
m.insert("compact_mb".into(), json!(r.compact_mb));
m.insert("compact_ratio".into(), json!(r.compact_ratio));
m.insert("scan_compact_ns".into(), json!(r.scan_compact_ns));
Ok(BenchResult { name: self.result.into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench!(GitPushLog {
id: "znippy.git_pushlog_5k", result: "git_pushlog_5k",
pushes: 5_000, refs_per_push: 4
});
/// A real `.znippy` sealed through the ordinary `compress_dir` path with all
/// three reserved sections attached, then reopened and every object resolved
/// through the sealed `__gunnar_oid__`. This is the arm that proves the format
/// survives the writer, not just the unit tests.
///
/// RED: the reopen resolving fewer objects than were sealed — the kernel bails
/// rather than reporting a number, because a fast index that has lost objects is
/// the failure this whole format has to not have.
struct GitArchive { id: &'static str, result: &'static str, commits: usize, width: usize }
impl Bencher for GitArchive {
fn id(&self) -> &'static str { self.id }
fn run(&self) -> Result<BenchResult> {
let commits = if preview() { (self.commits / 20).max(200) } else { self.commits };
let dir = std::path::Path::new(GIT_BENCH_SCRATCH);
std::fs::create_dir_all(dir)
.with_context(|| format!("git bench scratch {}", dir.display()))?;
let r = gitbench::archive_build(commits, self.width, dir)?;
let mut m = serde_json::Map::new();
m.insert("objects".into(), json!(r.objects));
m.insert("source_mb".into(), json!(r.source_mb));
m.insert("archive_mb".into(), json!(r.archive_mb));
m.insert("write_fixture_s".into(), json!(r.write_fixture_s));
m.insert("seal_s".into(), json!(r.seal_s));
m.insert("objects_per_sec".into(), json!(r.objects_per_s));
m.insert("seal_mbs".into(), json!(r.source_mb_per_s));
m.insert("reopen_lookup_ns".into(), json!(r.reopen_lookup_ns));
m.insert("cores".into(), json!(cores()));
Ok(BenchResult { name: self.result.into(), source: BenchSource::Measured, metrics: m })
}
}
nornir::register_bench!(GitArchive {
id: "znippy.git_archive_60k", result: "git_archive_60k",
commits: 20_000, width: 8
});
fn main() -> Result<()> {
// Agent sandboxes set PR_SET_THP_DISABLE and children inherit it, which
// silently disables every MADV_HUGEPAGE the stree trees issue (found
// 2026-08-10 — AnonHugePages stayed 0 under gigabytes of advised trees).
// Clearing it is a per-binary decision; this binary wants real numbers.
let _ = znippy_zoomies::stree::thp_enable_for_process();
// Focused run: `ZNIPPY_BENCH_ONLY=substr1,substr2 …` runs only the benchers
// whose id() contains any of the comma-separated substrings, emitting the
// same one-line BenchRun JSON as the full runner. Used to baseline a single
// family (e.g. the arrow-ipc benches) without churning the whole suite.
if let Ok(filter) = std::env::var("ZNIPPY_BENCH_ONLY") {
if !filter.trim().is_empty() {
return run_filtered(&filter);
}
}
nornir::bench::api::run_main_json()
}
fn run_filtered(filter: &str) -> Result<()> {
use nornir::bench::api::BencherRegistration;
use nornir::bench::{BenchRun, TestOutcome};
let needles: Vec<&str> = filter.split(',').map(|s| s.trim()).filter(|s| !s.is_empty()).collect();
let mut chosen: Vec<&'static dyn Bencher> = inventory::iter::<BencherRegistration>()
.map(|r| (r.make)())
.filter(|b| needles.iter().any(|n| b.id().contains(n)))
.collect();
chosen.sort_by(|a, b| a.id().cmp(b.id()));
let mut results = Vec::new();
let mut tests = Vec::new();
for b in &chosen {
let id = b.id();
eprintln!("focused-bench: {id} starting…");
let start = Instant::now();
match b.run() {
Ok(r) => {
let dur = start.elapsed().as_secs_f64() * 1000.0;
eprintln!("focused-bench: {id} → {} ({:.2}s)",
r.metrics.iter().next().map(|(k, v)| format!("{k}={v}")).unwrap_or_default(),
dur / 1000.0);
tests.push(TestOutcome { name: id.to_string(), passed: true, duration_ms: Some(dur), message: None });
results.push(r);
}
Err(e) => {
eprintln!("focused-bench: {id} ✗ {e}");
tests.push(TestOutcome { name: id.to_string(), passed: false, duration_ms: None, message: Some(e.to_string()) });
}
}
}
let now = chrono_now();
let run = BenchRun {
date: now.0,
timestamp: Some(now.1),
version: std::env::var("NORNIR_BENCH_VERSION").ok().filter(|v| !v.is_empty()).unwrap_or_else(|| "0.9.0".into()),
// Route the host stamp through the SHARED helper, exactly as `run_main_json`
// does. A `NORNIR_BENCH_PREVIEW=1` focused run must emit `<host>-preview`, so
// a small verification pass can never land in — or overwrite — the real host's
// README row. Re-deriving the suffix here would be a twin that silently rots.
machine: nornir::bench::api::stamp_machine(
std::env::var("NORNIR_MACHINE").unwrap_or_default(),
),
cores: cores() as u32,
results,
tests,
};
println!("{}", serde_json::to_string(&run)?);
Ok(())
}
/// (date YYYY-MM-DD, rfc3339) without pulling chrono directly into xtask: shell
/// out is overkill, so derive from SystemTime. The baseline row is stamped with
/// a literal date by the append step anyway; this is just the live timestamp.
fn chrono_now() -> (String, String) {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs();
// crude UTC date from epoch seconds (no leap-second concern for a stamp).
let days = now / 86_400;
let secs = now % 86_400;
// days since 1970-01-01 → y/m/d (civil from days, Howard Hinnant's algorithm)
let z = days as i64 + 719_468;
let era = if z >= 0 { z } else { z - 146_096 } / 146_097;
let doe = z - era * 146_097;
let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
let y = yoe + era * 400;
let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
let mp = (5 * doy + 2) / 153;
let d = doy - (153 * mp + 2) / 5 + 1;
let m = if mp < 10 { mp + 3 } else { mp - 9 };
let y = if m <= 2 { y + 1 } else { y };
let date = format!("{y:04}-{m:02}-{d:02}");
let ts = format!("{date}T{:02}:{:02}:{:02}Z", secs / 3600, (secs % 3600) / 60, secs % 60);
(date, ts)
}