bench/examples/nornir-bench.rs
//! nornir bencher entry point (the contract in `nornir::bench::api`).
//!
//! `cargo run --release --example nornir-bench` — or, end-to-end,
//! `nornir bench run nornir-catalog` from `workspace_nornir-catalog/` — prints
//! one `BenchRun` JSON line on stdout. nornir parses it, persists the run into
//! the `bench_runs` Iceberg table, and `nornir docs render nornir-catalog` folds
//! the results into `.nornir/README.md`'s `benches` regions → root `README.md`.
//!
//! Each `Bencher` produces one named `BenchResult`; the README sections filter by
//! result-name substring (`table_exists`; `skade_katalog_embedded` minus `table_exists`;
//! `data_pipe`; `commit_burst`). The result ids match the `required_results` in
//! `workspace_nornir-catalog/release/nornir.toml`. S3/Nessie/Polaris benchers need
//! their containers (`cargo run --bin bench-containers up all`); if a container is
//! down that bench degrades to a recorded failure, not a run abort. The three
//! skade S3 data-plane arms (`data_pipe_skade_s3` / `commit_burst_skade_s3` /
//! `tpch_cmp_skade_s3`) go one better: built WITHOUT `--features s3` they emit an
//! honest, `BenchSource::Static` **skip** row (`skipped=1` + a `--features s3`
//! reason, no throughput number) instead of a red failure — a clean self-skip
//! that can never read as a false-green measurement. Build `--features s3` (with
//! RustFS up) to actually measure them.
//!
//! Env knobs: `NORNIR_BENCH_FULL=1` (10k-table read scale), `NORNIR_BENCH_TABLES`
//! / `NORNIR_BENCH_ITERS` (read scale), `BENCH_DATA_ROWS` / `BENCH_DATA_BATCH`
//! (data-pipe), `NORNIR_BURST_TABLES` / `NORNIR_BURST_COMMITS` (commit-burst),
//! `BENCH_REST_URI` (Nessie endpoint), `OSM_GEOPARQUET` / `OSM_MAX_ROWS`
//! (OSM GeoParquet ingest; the `osm_ingest_*` benchers skip-as-failure if unset).
use std::sync::Arc;
use std::time::Instant;
use anyhow::Result;
use iceberg::{Catalog, NamespaceIdent, TableIdent};
use serde_json::{json, Map, Value};
use nornir::bench::api::{run_main_json, Bencher};
use nornir::bench::BenchResult;
use nornir::register_bench;
use skade_katalog_bench::scenarios::{CatalogResult, Scale};
use skade_katalog_bench::{data, factory, pipeline, scenarios};
#[cfg(feature = "rest")]
use skade_katalog_bench::rest_shim;
fn main() -> Result<()> {
run_main_json()
}
// ---- shared helpers --------------------------------------------------------
fn rt() -> tokio::runtime::Runtime {
tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.expect("tokio runtime")
}
fn env_usize(k: &str, d: usize) -> usize {
std::env::var(k).ok().and_then(|v| v.parse().ok()).unwrap_or(d)
}
/// PREVIEW mode (`NORNIR_BENCH_PREVIEW=1`) — the fast verification loop.
///
/// Every arm still runs, through the SAME code paths, emitting the SAME metric keys;
/// only the corpus scale shrinks (fewer OSM rows, fewer decode iterations), so a
/// broken arm — or a broken persistence chain — surfaces in seconds instead of the
/// tens of minutes a real OSM ingest takes. The harness stamps the emitted run onto
/// `<host>-preview`, so a preview number can never land in — or overwrite — the real
/// host's README row. Every explicit `OSM_*` / `ZSTD_*` env knob still WINS over the
/// preview default, so you can preview at any scale.
fn preview() -> bool {
nornir_bench_api_preview()
}
/// The harness's preview flag, reached through nornir's re-export.
fn nornir_bench_api_preview() -> bool {
nornir::bench::api::preview_run()
}
/// A size default: the heavy one, or the small one in preview mode.
fn scaled(heavy: usize, light: usize) -> usize {
if preview() { light } else { heavy }
}
/// Read-latency scale: small + fast by default (latency is scale-insensitive);
/// `NORNIR_BENCH_FULL=1` uses the 10k-table headline scale.
fn read_scale() -> Scale {
if std::env::var("NORNIR_BENCH_FULL").is_ok() {
Scale::full()
} else {
Scale {
tables: env_usize("NORNIR_BENCH_TABLES", 100),
iters: env_usize("NORNIR_BENCH_ITERS", 2_000),
threads: vec![],
}
}
}
fn bench(name: &str, metrics: Map<String, Value>) -> BenchResult {
BenchResult::measured(name, metrics)
}
fn latency_metrics(r: &CatalogResult) -> Map<String, Value> {
let mut m = Map::new();
m.insert("ops_sec".into(), json!(r.ops_sec));
m.insert("min_us".into(), json!(r.min_us));
m.insert("mean_us".into(), json!(r.mean_us));
m.insert("p50_us".into(), json!(r.p50_us));
m.insert("p90_us".into(), json!(r.p90_us));
m.insert("p99_us".into(), json!(r.p99_us));
m.insert("p999_us".into(), json!(r.p999_us));
m.insert("max_us".into(), json!(r.max_us));
m
}
fn pctl_metrics(extra: &[(&str, f64)], mut samples_ns: Vec<u128>, wall_ns: u128) -> Map<String, Value> {
samples_ns.sort_unstable();
let ops = samples_ns.len() as u64;
let pct = |q: f64| -> f64 {
let idx = (((samples_ns.len().max(1) - 1) as f64) * q).round() as usize;
samples_ns.get(idx).copied().unwrap_or(0) as f64 / 1000.0
};
let mean_us = if ops == 0 {
0.0
} else {
(samples_ns.iter().sum::<u128>() as f64 / ops as f64) / 1000.0
};
let mut m = Map::new();
for (k, v) in extra {
m.insert((*k).to_string(), json!(v));
}
m.insert("ops_sec".into(), json!(if wall_ns > 0 { ops as f64 * 1e9 / wall_ns as f64 } else { 0.0 }));
m.insert("min_us".into(), json!(samples_ns.first().copied().unwrap_or(0) as f64 / 1000.0));
m.insert("mean_us".into(), json!(mean_us));
m.insert("p50_us".into(), json!(pct(0.50)));
m.insert("p90_us".into(), json!(pct(0.90)));
m.insert("p99_us".into(), json!(pct(0.99)));
m.insert("p999_us".into(), json!(pct(0.999)));
m.insert("max_us".into(), json!(samples_ns.last().copied().unwrap_or(0) as f64 / 1000.0));
m
}
// ---- data-pipe / commit-burst cores (reused across targets) ----------------
/// `write_concurrency`: 1 → the single-writer-sequential path (optimal for local
/// NVMe/RAM where a write is a memcpy-class syscall). >1 → the GATLING-style
/// concurrent multi-PUT path (`run_concurrent_writer_ingest`, N PUTs in flight)
/// to hide S3/network per-request latency. The S3/Nessie benchers pass
/// `BENCH_PIPE_WRITE_CONCURRENCY` (default 16); local passes 1.
async fn data_pipe_metrics(cat: Arc<dyn Catalog>, write_concurrency: usize) -> Result<Map<String, Value>> {
let rows = env_usize("BENCH_DATA_ROWS", 500_000);
let per_batch = env_usize("BENCH_DATA_BATCH", 50_000);
let ns = NamespaceIdent::new("bench".to_string());
if !cat.namespace_exists(&ns).await.unwrap_or(false) {
cat.create_namespace(&ns, Default::default()).await?;
}
let ident = TableIdent::new(ns.clone(), "osm_nodes_pipe".to_string());
if cat.table_exists(&ident).await.unwrap_or(false) {
let _ = cat.drop_table(&ident).await;
}
let table = data::create_node_table(cat.as_ref(), &ident).await?;
let schema = data::node_arrow_schema_standalone()?;
let groups: Vec<Vec<_>> = data::synthetic_batches(schema.clone(), rows, per_batch)
.into_iter()
.map(|b| vec![b])
.collect();
let (table, stats) = if write_concurrency > 1 {
pipeline::run_concurrent_writer_ingest(Arc::clone(&cat), table, groups, schema, 8, 8, "pipe", write_concurrency).await?
} else {
pipeline::run_single_writer_ingest(Arc::clone(&cat), table, groups, schema, 8, 8, "pipe").await?
};
let (scanned, _) = data::scan_count(&table).await?;
anyhow::ensure!(scanned == stats.rows, "scan mismatch: {} != {}", scanned, stats.rows);
let mut m = Map::new();
m.insert("rows_per_sec".into(), json!(stats.rows_per_sec()));
m.insert("files".into(), json!(stats.files as f64));
m.insert("commits".into(), json!(stats.commits as f64));
m.insert("encode_workers".into(), json!(stats.encode_workers as f64));
m.insert("write_concurrency".into(), json!(stats.write_concurrency as f64));
m.insert("peak_inflight".into(), json!(stats.peak_inflight as f64));
Ok(m)
}
/// Concurrent-write level for S3/network targets (GATLING multi-PUT). Local
/// targets pass 1 (sequential is optimal there). Only referenced by the S3 /
/// REST data-plane arms, so it is gated to their feature set (else the default
/// build flags it dead).
#[cfg(any(feature = "s3", feature = "rest"))]
fn s3_write_concurrency() -> usize {
env_usize("BENCH_PIPE_WRITE_CONCURRENCY", 16).max(1)
}
/// The honest **skip** row an `_s3` bench arm emits when the binary was built
/// without `--features s3`. Wraps [`factory::s3_not_compiled_skip_metrics`] in a
/// `BenchSource::Static` result: the no-regression gate ignores Static rows and
/// no consumer reads one as a timed measurement, and the payload carries only a
/// `skipped` flag + a `--features s3` reason (never a throughput number), so the
/// skip can never masquerade as a false-green result. The `bench_arms` test
/// asserts that honesty invariant.
#[cfg(not(feature = "s3"))]
fn s3_not_compiled_skip(id: &str) -> BenchResult {
BenchResult::static_capabilities(id, factory::s3_not_compiled_skip_metrics())
}
async fn commit_burst_metrics(cat: Arc<dyn Catalog>) -> Result<Map<String, Value>> {
use iceberg::transaction::{ApplyTransactionAction, Transaction};
let tables = env_usize("NORNIR_BURST_TABLES", 32);
let per_table = env_usize("NORNIR_BURST_COMMITS", 100);
let ns = NamespaceIdent::new("burst".to_string());
if !cat.namespace_exists(&ns).await.unwrap_or(false) {
cat.create_namespace(&ns, Default::default()).await?;
}
let mut tabs = Vec::with_capacity(tables);
for t in 0..tables {
let ident = TableIdent::new(ns.clone(), format!("burst_{t:03}"));
if cat.table_exists(&ident).await.unwrap_or(false) {
let _ = cat.drop_table(&ident).await;
}
tabs.push(data::create_node_table(cat.as_ref(), &ident).await?);
}
let wall = Instant::now();
let mut handles = Vec::with_capacity(tables);
for table in tabs {
let cat = Arc::clone(&cat);
handles.push(tokio::spawn(async move {
let mut table = table;
let mut lat = Vec::with_capacity(per_table);
for i in 0..per_table {
let tx = Transaction::new(&table);
let action = tx.update_table_properties().set("bench.seq".to_string(), i.to_string());
let tx = action.apply(tx)?;
let t0 = Instant::now();
table = tx.commit(cat.as_ref()).await?;
lat.push(t0.elapsed().as_nanos());
}
Ok::<Vec<u128>, anyhow::Error>(lat)
}));
}
let mut all: Vec<u128> = Vec::new();
for h in handles {
all.extend(h.await??);
}
let secs = wall.elapsed().as_secs_f64().max(1e-9);
let commits = all.len() as f64;
Ok(pctl_metrics(&[("commits_per_sec", commits / secs)], all, wall.elapsed().as_nanos()))
}
// ---- embedded read scenarios ----------------------------------------------
struct EmbeddedTableExists;
impl Bencher for EmbeddedTableExists {
fn id(&self) -> &'static str { "skade_katalog_embedded_table_exists" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let sc = read_scale();
let (cat, _tmp) = factory::embedded().await?;
let ns = scenarios::seed(&cat, &sc).await?;
let r = scenarios::table_exists_latency(&cat, &ns, &sc).await;
Ok(bench(self.id(), latency_metrics(&r)))
})
}
}
register_bench!(EmbeddedTableExists);
struct EmbeddedResolveMetadata;
impl Bencher for EmbeddedResolveMetadata {
fn id(&self) -> &'static str { "skade_katalog_embedded_resolve_metadata" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let sc = read_scale();
let (cat, _tmp) = factory::embedded().await?;
let ns = scenarios::seed(&cat, &sc).await?;
let n = sc.tables.max(1);
for i in 0..100 {
let _ = cat
.resolve_metadata(&TableIdent::new(ns.clone(), format!("bench_t_{:06}", i % n)))
.await;
}
let mut samples = Vec::with_capacity(sc.iters);
let wall = Instant::now();
for i in 0..sc.iters {
let ident = TableIdent::new(ns.clone(), format!("bench_t_{:06}", i % n));
let t = Instant::now();
let _ = cat.resolve_metadata(&ident).await;
samples.push(t.elapsed().as_nanos());
}
Ok(bench(self.id(), pctl_metrics(&[], samples, wall.elapsed().as_nanos())))
})
}
}
register_bench!(EmbeddedResolveMetadata);
struct EmbeddedLoadTable;
impl Bencher for EmbeddedLoadTable {
fn id(&self) -> &'static str { "skade_katalog_embedded_load_table" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let sc = read_scale();
let (cat, _tmp) = factory::embedded().await?;
let ns = scenarios::seed(&cat, &sc).await?;
let r = scenarios::load_table_latency(&cat, &ns, &sc).await;
Ok(bench(self.id(), latency_metrics(&r)))
})
}
}
register_bench!(EmbeddedLoadTable);
// ---- static capabilities (issue #13 seam) ----------------------------------
// The front-page competitive matrix (skade vs Iceberg-Java / PyIceberg / Delta)
// emitted as a `BenchSource::Static` row — NOT a timed measurement. It carries
// the ✓/✗/◐/NA matrix as numeric coverage codes (see
// `skade_katalog_bench::capabilities`), tagged Static so the warehouse persists
// it and the no-regression gate skips it. The same matrix definition renders the
// README-full capability table, so the two can't drift.
struct SkadeCapabilities;
impl Bencher for SkadeCapabilities {
fn id(&self) -> &'static str { skade_katalog_bench::capabilities::RESULT_NAME }
fn run(&self) -> Result<BenchResult> {
Ok(BenchResult::static_capabilities(
self.id(),
skade_katalog_bench::capabilities::numeric_metrics(),
))
}
}
register_bench!(SkadeCapabilities);
// ---- catalog-key micro-bench (in-proc, no containers/S3) -------------------
// Times the redb key builders (`keys::{table_key,namespace_key,table_prefix}`) —
// the `pub(crate)` hot-path fns reached out-of-crate via skade-katalog's `bench`
// feature hook (`skade_katalog::bench_keys`). Every commit + every catalog op
// builds one of these keys; the zero-copy rewrite (one pre-sized alloc vs
// clone+join+`format!`) is byte-identical and ~8-10× on this box. A LIGHT,
// pure-in-proc run; the heavy N sweep goes to the quiet bench box via
// `BENCH_CATALOG_KEY_ITERS` (Loki/Odin). See `.nornir/catalog-key-zerocopy.md`.
struct CatalogKeyBuild;
impl Bencher for CatalogKeyBuild {
fn id(&self) -> &'static str { "skade_catalog_key_build" }
fn run(&self) -> Result<BenchResult> {
use std::hint::black_box;
use iceberg::{NamespaceIdent, TableIdent};
use skade_katalog::bench_keys;
let n = env_usize("BENCH_CATALOG_KEY_ITERS", 500_000);
let ns = NamespaceIdent::from_strs(["analytics", "warehouse"])
.expect("namespace ident");
let tbl = TableIdent::new(ns.clone(), "orders".to_string());
// Batched timing: a key build is a ~30ns op, so per-iteration
// `Instant::now()` overhead would swamp it — time the whole loop and
// divide. `black_box` keeps the builder (and its alloc) from being
// elided; the running `sink` pins the result live.
let mut sink = 0usize;
let warm = (n / 10).max(1);
let mut time_ns = |f: &dyn Fn() -> String| -> f64 {
for _ in 0..warm {
sink = sink.wrapping_add(black_box(f()).len());
}
let t = Instant::now();
for _ in 0..n {
sink = sink.wrapping_add(black_box(f()).len());
}
t.elapsed().as_nanos() as f64 / n as f64
};
let table_key_ns = time_ns(&|| bench_keys::table_key("my_catalog", &tbl));
let namespace_key_ns = time_ns(&|| bench_keys::namespace_key("my_catalog", &ns));
let table_prefix_ns = time_ns(&|| bench_keys::table_prefix("my_catalog", &ns));
let mut m = Map::new();
// `ops_sec` keyed off the headline builder (`table_key`, one per commit)
// so the warehouse no-regression gate has a single throughput number.
m.insert("ops_sec".into(), json!(if table_key_ns > 0.0 { 1e9 / table_key_ns } else { 0.0 }));
m.insert("table_key_ns_per_op".into(), json!(table_key_ns));
m.insert("namespace_key_ns_per_op".into(), json!(namespace_key_ns));
m.insert("table_prefix_ns_per_op".into(), json!(table_prefix_ns));
m.insert("iters".into(), json!(n as f64));
// Keep the accumulator observable so the optimizer can't drop the loops.
m.insert("_sink".into(), json!(sink as f64));
Ok(bench(self.id(), m))
}
}
register_bench!(CatalogKeyBuild);
// ---- spatial radius micro-bench (in-proc, no containers/S3) ----------------
// Times `GeoIndex::query_radius` over a synthetic point cloud — the OSM/GeoParquet
// nearest-features hot path. The candidate walk now carries each point's coords
// straight from the covering cells, so a k-hit radius is O(k) instead of the old
// O(k·n) `entries.iter().find(id)` re-scan (byte-identical result set — guarded
// by `skade/tests/spatial.rs::radius_matches_brute_force_after_zero_copy_rewrite`).
// LIGHT by default; the heavy N sweep goes to the quiet bench box via
// `BENCH_SPATIAL_POINTS` / `BENCH_SPATIAL_QUERIES` (Loki/Odin).
struct SpatialRadiusQuery;
impl Bencher for SpatialRadiusQuery {
fn id(&self) -> &'static str { "skade_spatial_radius" }
fn run(&self) -> Result<BenchResult> {
use std::hint::black_box;
use skade::spatial::GeoIndex;
let points = env_usize("BENCH_SPATIAL_POINTS", 50_000);
let queries = env_usize("BENCH_SPATIAL_QUERIES", 2_000);
// Deterministic pseudo-random cloud (same integer recurrence as the
// spatial tests) so the bench is byte-stable run to run.
let mut idx = GeoIndex::new(9);
for i in 0..points as u64 {
let lat = -80.0 + (i as f64 * 0.017) % 160.0;
let lon = -175.0 + (i as f64 * 0.031) % 350.0;
idx.insert(lat, lon, i);
}
idx.build();
// A tight radius: touches a small candidate set, so a per-candidate O(n)
// re-find (the old code) would dominate — this bench pins that cost.
let radius_m = 25_000.0;
let query_at = |q: usize| -> (f64, f64) {
let lat = -70.0 + (q as f64 * 3.7).rem_euclid(140.0);
let lon = -170.0 + (q as f64 * 5.9).rem_euclid(340.0);
(lat, lon)
};
let warm = (queries / 10).max(1);
let mut sink = 0usize;
for q in 0..warm {
let (la, lo) = query_at(q);
sink = sink.wrapping_add(black_box(idx.query_radius(la, lo, radius_m)).len());
}
let t = Instant::now();
for q in 0..queries {
let (la, lo) = query_at(q);
sink = sink.wrapping_add(black_box(idx.query_radius(la, lo, radius_m)).len());
}
let ns_per_query = t.elapsed().as_nanos() as f64 / queries.max(1) as f64;
let mut m = Map::new();
m.insert("ops_sec".into(), json!(if ns_per_query > 0.0 { 1e9 / ns_per_query } else { 0.0 }));
m.insert("ns_per_query".into(), json!(ns_per_query));
m.insert("points".into(), json!(points as f64));
m.insert("queries".into(), json!(queries as f64));
m.insert("radius_m".into(), json!(radius_m));
m.insert("_sink".into(), json!(sink as f64));
Ok(bench(self.id(), m))
}
}
register_bench!(SpatialRadiusQuery);
// ---- git-like refs micro-bench (in-proc embedded, no containers/S3) --------
// Times the git-like catalog surface (`src/git.rs`): the lock-free ref-read hot
// path (`ref_snapshot_id`, `list_refs`) over a table carrying a linear snapshot
// history and many branches/tags, plus the ref-move write path (`fast_forward`
// / `rollback_to` alternation — one redb commit + fsync each). Reads route
// through the same L1 pointer mirror + L0 metadata cache as `resolve_metadata`,
// so a warm ref lookup is a metadata-map read, not a redb hit. LIGHT by default;
// the heavy sweep goes to the quiet bench box via `BENCH_GIT_REFS` (refs seeded)
// / `BENCH_GIT_READ_ITERS` / `BENCH_GIT_WRITE_ITERS` (Loki/Odin).
// See `.nornir/git-refs-design.md`.
struct GitRefOps;
impl GitRefOps {
// A synthetic append-snapshot on a ref, through the normal commit_table path.
async fn append(
cat: &skade_katalog::RedbCatalog,
ident: &TableIdent,
ref_name: &str,
snapshot_id: i64,
parent: Option<i64>,
seq: i64,
) -> Result<()> {
use iceberg::spec::{Operation, Snapshot, SnapshotReference, SnapshotRetention, Summary};
use iceberg::TableUpdate;
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0);
let snap = Snapshot::builder()
.with_snapshot_id(snapshot_id)
.with_parent_snapshot_id(parent)
.with_sequence_number(seq)
.with_timestamp_ms(now_ms + seq)
.with_manifest_list(format!("file:///dev/null/m-{snapshot_id}.avro"))
.with_schema_id(0)
.with_summary(Summary {
operation: Operation::Append,
additional_properties: std::collections::HashMap::new(),
})
.build();
let updates = vec![
TableUpdate::AddSnapshot { snapshot: snap },
TableUpdate::SetSnapshotRef {
ref_name: ref_name.to_string(),
reference: SnapshotReference::new(
snapshot_id,
SnapshotRetention::Branch {
min_snapshots_to_keep: None,
max_snapshot_age_ms: None,
max_ref_age_ms: None,
},
),
},
];
cat.commit_table(ident.clone(), vec![], updates).await?;
Ok(())
}
}
impl Bencher for GitRefOps {
fn id(&self) -> &'static str { "skade_git_refs" }
fn run(&self) -> Result<BenchResult> {
use std::hint::black_box;
use iceberg::spec::{NestedField, PrimitiveType, Schema, Type};
let n_refs = env_usize("BENCH_GIT_REFS", 256);
let read_iters = env_usize("BENCH_GIT_READ_ITERS", 500_000);
let write_iters = env_usize("BENCH_GIT_WRITE_ITERS", 2_000);
rt().block_on(async move {
let (cat, _tmp) = factory::embedded().await?;
let ns = NamespaceIdent::new("git".to_string());
cat.create_namespace(&ns, Default::default()).await?;
let schema = Schema::builder()
.with_schema_id(0)
.with_fields(vec![
NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(),
])
.build()?;
let creation = iceberg::TableCreation::builder()
.name("t".to_string())
.schema(schema)
.build();
cat.create_table(&ns, creation).await?;
let ident = TableIdent::new(ns, "t".to_string());
// Linear snapshot history 1000..1000+depth on main.
let depth: i64 = 64;
let mut parent = None;
for i in 0..depth {
let sid = 1000 + i;
Self::append(&cat, &ident, "main", sid, parent, i + 1).await?;
parent = Some(sid);
}
let head = 1000 + depth - 1;
// Seed n_refs branches + tags pinned across the history.
let ref_names: Vec<String> = (0..n_refs)
.map(|i| if i % 2 == 0 { format!("branch_{i:05}") } else { format!("tag_{i:05}") })
.collect();
for (i, name) in ref_names.iter().enumerate() {
let sid = 1000 + (i as i64 % depth);
if i % 2 == 0 {
cat.create_branch(&ident, name, Some(sid), skade_katalog::BranchRetention::default()).await?;
} else {
cat.create_tag(&ident, name, Some(sid), None).await?;
}
}
// ---- READ hot path: ref_snapshot_id resolution (lock-free) --------
let warm = (read_iters / 10).max(1);
let mut sink: i64 = 0;
for i in 0..warm {
let name = &ref_names[i % ref_names.len()];
sink = sink.wrapping_add(black_box(cat.ref_snapshot_id(&ident, name).await?).unwrap_or(0));
}
let t = Instant::now();
for i in 0..read_iters {
let name = &ref_names[i % ref_names.len()];
sink = sink.wrapping_add(black_box(cat.ref_snapshot_id(&ident, name).await?).unwrap_or(0));
}
let ref_read_ns = t.elapsed().as_nanos() as f64 / read_iters.max(1) as f64;
// ---- READ: list_refs (build + sort the whole refs map) ------------
let list_warm = (read_iters / 100).max(1);
for _ in 0..list_warm {
sink = sink.wrapping_add(black_box(cat.list_refs(&ident).await?).len() as i64);
}
let list_n = (read_iters / 50).max(1);
let t2 = Instant::now();
for _ in 0..list_n {
sink = sink.wrapping_add(black_box(cat.list_refs(&ident).await?).len() as i64);
}
let list_refs_ns = t2.elapsed().as_nanos() as f64 / list_n as f64;
// ---- WRITE path: ref-move commits ---------------------------------
// Each iteration is two redb commits (fsync each) through
// commit_table: create a tag then drop it. Measures the git ref
// mutation throughput (the WAP/branch-lifecycle write cost).
let _ = head;
let w = Instant::now();
for i in 0..write_iters {
let sid = 1000 + (i as i64 % depth);
cat.create_tag(&ident, "churn", Some(sid), None).await?;
cat.drop_ref(&ident, "churn").await?;
}
let write_wall = w.elapsed().as_secs_f64().max(1e-9);
let ref_writes_per_sec = (write_iters as f64 * 2.0) / write_wall;
let mut m = Map::new();
// Headline throughput: the lock-free ref read (the git read surface).
m.insert("ops_sec".into(), json!(if ref_read_ns > 0.0 { 1e9 / ref_read_ns } else { 0.0 }));
m.insert("ref_read_ns_per_op".into(), json!(ref_read_ns));
m.insert("list_refs_ns_per_op".into(), json!(list_refs_ns));
m.insert("ref_writes_per_sec".into(), json!(ref_writes_per_sec));
m.insert("refs_seeded".into(), json!(n_refs as f64));
m.insert("history_depth".into(), json!(depth as f64));
m.insert("read_iters".into(), json!(read_iters as f64));
m.insert("write_iters".into(), json!(write_iters as f64));
m.insert("_sink".into(), json!(sink as f64));
Ok(bench(self.id(), m))
})
}
}
register_bench!(GitRefOps);
// ---- nornir-rest (in-proc axum shim, no container) -------------------------
#[cfg(feature = "rest")]
struct RestTableExists;
#[cfg(feature = "rest")]
impl Bencher for RestTableExists {
fn id(&self) -> &'static str { "skade_katalog_rest_table_exists" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let sc = read_scale();
let (cat, _tmp) = factory::embedded().await?;
let bound = rest_shim::spawn(cat, "127.0.0.1:0".parse().unwrap()).await?;
let rest = factory::rest(&format!("http://{bound}"), "warehouse").await?;
let ns = scenarios::seed(&rest, &sc).await?;
let r = scenarios::table_exists_latency(&rest, &ns, &sc).await;
Ok(bench(self.id(), latency_metrics(&r)))
})
}
}
#[cfg(feature = "rest")]
register_bench!(RestTableExists);
// ---- skade gRPC (in-proc tonic front, no container) ------------------------
// The third skade front: the SAME embedded catalog behind the native gRPC
// server, driven through the `GrpcCatalog` adapter by the identical scenarios.
// No feature gate — the gRPC crate links the same arrow-58 iceberg fork.
struct GrpcTableExists;
impl Bencher for GrpcTableExists {
fn id(&self) -> &'static str { "skade_katalog_grpc_table_exists" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let sc = read_scale();
let (cat, _tmp) = factory::embedded().await?;
let bound = skade_katalog_grpc::spawn(cat, "127.0.0.1:0".parse().unwrap()).await?;
let grpc = skade_katalog_bench::grpc_catalog::GrpcCatalog::connect(bound).await?;
let ns = scenarios::seed(&grpc, &sc).await?;
let r = scenarios::table_exists_latency(&grpc, &ns, &sc).await;
Ok(bench(self.id(), latency_metrics(&r)))
})
}
}
register_bench!(GrpcTableExists);
struct GrpcLoadTable;
impl Bencher for GrpcLoadTable {
fn id(&self) -> &'static str { "skade_katalog_grpc_load_table" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let sc = read_scale();
let (cat, _tmp) = factory::embedded().await?;
let bound = skade_katalog_grpc::spawn(cat, "127.0.0.1:0".parse().unwrap()).await?;
let grpc = skade_katalog_bench::grpc_catalog::GrpcCatalog::connect(bound).await?;
let ns = scenarios::seed(&grpc, &sc).await?;
let r = scenarios::load_table_latency(&grpc, &ns, &sc).await;
Ok(bench(self.id(), latency_metrics(&r)))
})
}
}
register_bench!(GrpcLoadTable);
// ---- iceberg-rust MemoryCatalog (serverless opponent, no container) --------
// The fair no-service rival for the embedded column: iceberg-rust's only
// serverless catalog (0.9.1 ships no file-based/Hadoop catalog), run through the
// identical scenarios.
struct MemoryTableExists;
impl Bencher for MemoryTableExists {
fn id(&self) -> &'static str { "iceberg_memory_table_exists" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let sc = read_scale();
let (cat, _tmp) = factory::memory().await?;
let ns = scenarios::seed(&cat, &sc).await?;
let r = scenarios::table_exists_latency(&cat, &ns, &sc).await;
Ok(bench(self.id(), latency_metrics(&r)))
})
}
}
register_bench!(MemoryTableExists);
// ---- Nessie (needs RustFS + Nessie containers) -----------------------------
#[cfg(feature = "rest")]
fn nessie_uri() -> String {
std::env::var("BENCH_REST_URI").unwrap_or_else(|_| "http://localhost:19120/iceberg".to_string())
}
#[cfg(feature = "rest")]
struct NessieTableExists;
#[cfg(feature = "rest")]
impl Bencher for NessieTableExists {
fn id(&self) -> &'static str { "nessie_table_exists" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let sc = read_scale();
let rest = factory::rest_s3(&nessie_uri(), "warehouse").await?;
let ns = scenarios::seed(&rest, &sc).await?;
let r = scenarios::table_exists_latency(&rest, &ns, &sc).await;
Ok(bench(self.id(), latency_metrics(&r)))
})
}
}
#[cfg(feature = "rest")]
register_bench!(NessieTableExists);
// ---- Polaris (control-plane only: table_exists; needs the Polaris container) -
#[cfg(feature = "rest")]
struct PolarisTableExists;
#[cfg(feature = "rest")]
impl Bencher for PolarisTableExists {
fn id(&self) -> &'static str { "polaris_table_exists" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let sc = read_scale();
let uri = std::env::var("BENCH_POLARIS_URI")
.unwrap_or_else(|_| "http://localhost:8181/api/catalog".to_string());
let rest = factory::rest_polaris(&uri).await?;
// Storage-free RPC: just the namespace; tables aren't created (the
// Polaris FILE warehouse 503s under a bind mount), so this times the
// exists-RPC round-trip, the apples-to-apples comparable.
let ns = scenarios::seed_namespace(&rest).await?;
let r = scenarios::table_exists_latency(&rest, &ns, &sc).await;
Ok(bench(self.id(), latency_metrics(&r)))
})
}
}
#[cfg(feature = "rest")]
register_bench!(PolarisTableExists);
// ---- data-pipe across targets ----------------------------------------------
// Two file destinations for the local-FS data plane: NVMe (/home/rickard/work)
// and RAM (/dev/shm). Both run the identical pipeline; the gap is pure storage.
struct DataPipeNornirNvme;
impl Bencher for DataPipeNornirNvme {
fn id(&self) -> &'static str { "data_pipe_skade_nvme" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let (cat, _tmp) = factory::embedded().await?;
Ok(bench(self.id(), data_pipe_metrics(Arc::new(cat), 1).await?))
})
}
}
register_bench!(DataPipeNornirNvme);
struct DataPipeNornirRam;
impl Bencher for DataPipeNornirRam {
fn id(&self) -> &'static str { "data_pipe_skade_ram" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let tmp = factory::tempdir_in(&factory::ram_dir())?;
let cat = factory::embedded_in(&tmp).await?;
Ok(bench(self.id(), data_pipe_metrics(Arc::new(cat), 1).await?))
})
}
}
register_bench!(DataPipeNornirRam);
struct DataPipeNornirS3;
impl Bencher for DataPipeNornirS3 {
fn id(&self) -> &'static str { "data_pipe_skade_s3" }
#[cfg(feature = "s3")]
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let (cat, _tmp) = factory::embedded_s3().await?;
Ok(bench(self.id(), data_pipe_metrics(Arc::new(cat), s3_write_concurrency()).await?))
})
}
// Built without `--features s3`: self-skip cleanly (honest Static row), not a
// recorded failure. See `s3_not_compiled_skip`.
#[cfg(not(feature = "s3"))]
fn run(&self) -> Result<BenchResult> {
Ok(s3_not_compiled_skip(self.id()))
}
}
register_bench!(DataPipeNornirS3);
#[cfg(feature = "rest")]
struct DataPipeNessie;
#[cfg(feature = "rest")]
impl Bencher for DataPipeNessie {
fn id(&self) -> &'static str { "data_pipe_nessie" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let rest = factory::rest_s3(&nessie_uri(), "warehouse").await?;
Ok(bench(self.id(), data_pipe_metrics(Arc::new(rest), s3_write_concurrency()).await?))
})
}
}
#[cfg(feature = "rest")]
register_bench!(DataPipeNessie);
#[cfg(feature = "rest")]
struct DataPipePolaris;
#[cfg(feature = "rest")]
impl Bencher for DataPipePolaris {
fn id(&self) -> &'static str { "data_pipe_polaris" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
// Polaris serves a FILE warehouse end-to-end (its S3 vending 301s on
// RustFS — see TpchPolarisFile), so the full create→ingest→scan pipe
// runs through the REST control plane just like Nessie does.
let uri = std::env::var("BENCH_POLARIS_URI")
.unwrap_or_else(|_| "http://localhost:8181/api/catalog".to_string());
let rest = factory::rest_polaris(&uri).await?;
Ok(bench(self.id(), data_pipe_metrics(Arc::new(rest), s3_write_concurrency()).await?))
})
}
}
#[cfg(feature = "rest")]
register_bench!(DataPipePolaris);
// ---- OSM GeoParquet ingest --------------------------------------------------
// Ingest a real `nodes.parquet` (OSM nodes converted to GeoParquet by the
// `katana-osm` / `osm2geoparquet` project) into a fresh RedbCatalog Iceberg
// table via the single-writer pipeline, then scan it back to verify. Unlike the
// synthetic `data_pipe_*` benchers this measures end-to-end ingest of authentic
// OSM data (id/geometry-WKB/tags-JSON/version/changeset/timestamp). Point
// `OSM_GEOPARQUET` at the file; `OSM_MAX_ROWS` caps rows (default: all).
fn osm_geoparquet_path() -> Option<String> {
std::env::var("OSM_GEOPARQUET").ok().filter(|s| !s.is_empty())
}
async fn geoparquet_ingest_metrics(cat: Arc<dyn Catalog>, path: &str) -> Result<Map<String, Value>> {
let max_rows = env_usize("OSM_MAX_ROWS", scaled(usize::MAX, 200_000));
let per_batch = env_usize("BENCH_DATA_BATCH", 1_000_000);
// Stream the file in row-group windows of ~this many rows so peak memory is
// one window, not the whole file — lets us ingest all 3.75B Europe nodes
// (~290GB) without OOM. Default 50M ≈ a few GB resident.
let window_rows = env_usize("OSM_WINDOW_ROWS", 50_000_000);
let threads = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(8);
// Layout only (no data read yet): derive the Iceberg schema + per-row-group
// sizes so we can window the file.
let (ice, _file_schema, rg_rows) = data::parquet_layout(path)?;
let windows = data::rowgroup_windows(&rg_rows, window_rows, max_rows);
anyhow::ensure!(!windows.is_empty(), "no row groups in {path}");
let ns = NamespaceIdent::new("bench".to_string());
if !cat.namespace_exists(&ns).await.unwrap_or(false) {
cat.create_namespace(&ns, Default::default()).await?;
}
let ident = TableIdent::new(ns.clone(), "osm_geoparquet".to_string());
if cat.table_exists(&ident).await.unwrap_or(false) {
let _ = cat.drop_table(&ident).await;
}
let creation = iceberg::TableCreation::builder()
.name("osm_geoparquet".to_string())
.schema(ice)
.build();
let mut table = cat.create_table(&ns, creation).await?;
let target = data::table_arrow_schema(&table)?;
// Stream: read one window of row groups (parallel) → recast → ingest+commit →
// free → next. Time the whole read+write end-to-end (decode is real work).
let t0 = Instant::now();
let mut total_rows = 0u64;
let mut total_files = 0u64;
let mut total_commits = 0u64;
let mut total_bytes = 0u64;
let mut read_ns: u128 = 0;
let mut encode_workers = 0u64;
let nwin = windows.len();
for (wi, rgs) in windows.into_iter().enumerate() {
let tr = Instant::now();
let src = data::read_row_groups(path, &rgs, per_batch, threads)?;
read_ns += tr.elapsed().as_nanos();
if src.is_empty() {
continue;
}
let groups: Vec<Vec<_>> = data::recast(&src, target.clone())?.into_iter().map(|b| vec![b]).collect();
drop(src);
// Per-window prefix so file names are unique across windows (each call
// restarts the pipeline's seq at 0; without this they'd collide).
let prefix = format!("w{wi:04}");
let (t2, stats) =
pipeline::run_single_writer_ingest(Arc::clone(&cat), table, groups, target.clone(), 8, 8, &prefix).await?;
table = t2;
total_rows += stats.rows;
total_files += stats.files;
total_commits += stats.commits;
total_bytes += stats.bytes_written;
encode_workers = stats.encode_workers as u64;
eprintln!(" osm window {}/{}: +{} rows ({} total)", wi + 1, nwin, stats.rows, total_rows);
}
anyhow::ensure!(total_rows > 0, "no rows ingested from {path}");
// scan_count streams (folds row counts over the arrow stream), so verifying
// the whole table back stays memory-bounded too.
let (scanned, _) = data::scan_count(&table).await?;
anyhow::ensure!(scanned == total_rows, "scan mismatch: {} != {}", scanned, total_rows);
let e2e_s = t0.elapsed().as_secs_f64().max(1e-9);
let read_s = (read_ns as f64 / 1e9).max(1e-9);
let write_s = (e2e_s - read_s).max(1e-9);
let rows = total_rows as f64;
let mut m = Map::new();
// Headline = end-to-end (read source → committed in the lake). Read vs write
// split out so the catalog speed isn't masked by decode.
m.insert("rows_per_sec".into(), json!(rows / e2e_s));
m.insert("write_rows_per_sec".into(), json!(rows / write_s));
m.insert("read_rows_per_sec".into(), json!(rows / read_s));
m.insert("rows".into(), json!(rows));
m.insert("windows".into(), json!(nwin as f64));
m.insert("mb_per_sec".into(), json!(total_bytes as f64 / 1e6 / write_s));
m.insert("read_threads".into(), json!(threads as f64));
m.insert("files".into(), json!(total_files as f64));
m.insert("commits".into(), json!(total_commits as f64));
m.insert("encode_workers".into(), json!(encode_workers as f64));
Ok(m)
}
struct OsmIngestNornirNvme;
impl Bencher for OsmIngestNornirNvme {
fn id(&self) -> &'static str { "osm_ingest_skade_nvme" }
fn run(&self) -> Result<BenchResult> {
let path = osm_geoparquet_path().ok_or_else(|| {
anyhow::anyhow!("set OSM_GEOPARQUET=/path/to/nodes.parquet (katana-osm osm2geoparquet output)")
})?;
rt().block_on(async {
let (cat, _tmp) = factory::embedded().await?;
Ok(bench(self.id(), geoparquet_ingest_metrics(Arc::new(cat), &path).await?))
})
}
}
register_bench!(OsmIngestNornirNvme);
struct OsmIngestNornirRam;
impl Bencher for OsmIngestNornirRam {
fn id(&self) -> &'static str { "osm_ingest_skade_ram" }
fn run(&self) -> Result<BenchResult> {
let path = osm_geoparquet_path().ok_or_else(|| {
anyhow::anyhow!("set OSM_GEOPARQUET=/path/to/nodes.parquet (katana-osm osm2geoparquet output)")
})?;
rt().block_on(async {
let tmp = factory::tempdir_in(&factory::ram_dir())?;
let cat = factory::embedded_in(&tmp).await?;
Ok(bench(self.id(), geoparquet_ingest_metrics(Arc::new(cat), &path).await?))
})
}
}
register_bench!(OsmIngestNornirRam);
// ---- zstd decode: zstd-sys-rs (zero-copy) vs the `zstd` crate ----------------
// Both call the SAME statically-linked libzstd 1.5.7, so this validates the
// user's bindings on real OSM data (WKB geometry bytes from europe nodes.parquet)
// and isolates the zero-copy path — a reused ZSTD_DCtx decoding into one reused
// output buffer (no per-frame Vec) — from the stock one-shot decode.
struct ZstdCorpus {
frames: Vec<Vec<u8>>, // zstd-compressed chunks of real OSM geometry bytes
max_raw: usize, // largest decompressed chunk (= reused output buffer size)
raw_total: u64, // decompressed bytes in one pass over all frames
}
fn zstd_corpus() -> Result<&'static ZstdCorpus> {
use std::sync::OnceLock;
static C: OnceLock<Option<ZstdCorpus>> = OnceLock::new();
C.get_or_init(|| {
let path = osm_geoparquet_path()?;
let threads = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(8);
let chunk = env_usize("ZSTD_CHUNK_KB", 64) * 1024; // ~parquet-page sized frames
let raw = data::osm_value_chunks(&path, env_usize("OSM_MAX_ROWS", scaled(20_000_000, 200_000)), threads, chunk).ok()?;
let mut frames = Vec::with_capacity(raw.len());
let (mut max_raw, mut raw_total) = (0usize, 0u64);
for chunk in &raw {
max_raw = max_raw.max(chunk.len());
raw_total += chunk.len() as u64;
frames.push(zstd::bulk::compress(chunk, 3).ok()?);
}
Some(ZstdCorpus { frames, max_raw, raw_total })
})
.as_ref()
.ok_or_else(|| anyhow::anyhow!("set OSM_GEOPARQUET=/path/to/nodes.parquet (katana-osm) for the zstd corpus"))
}
fn run_zstd_decode(name: &str, zero_copy_sys: bool) -> Result<BenchResult> {
let c = zstd_corpus()?;
let iters = env_usize("ZSTD_BENCH_ITERS", scaled(10, 2));
let threads = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(8);
let n = c.frames.len();
let per = n.div_ceil(threads);
// One pass = decode every frame, fanned out across all cores (an ingest
// engine decompresses many pages at once, so aggregate bandwidth is what
// matters). Each thread owns its context + output buffer.
let decode_pass = || {
std::thread::scope(|s| {
for t in 0..threads {
let (lo, hi) = (t * per, ((t + 1) * per).min(n));
if lo >= hi {
continue;
}
let frames = &c.frames[lo..hi];
let max_raw = c.max_raw.max(1);
s.spawn(move || {
let mut buf = vec![0u8; max_raw];
if zero_copy_sys {
let mut d = zstd_sys_rs::Decompressor::new(); // reused ctx, no per-frame alloc
for f in frames {
let _ = d.decompress_into(f, &mut buf).expect("zstd-sys-rs decode");
}
} else {
for f in frames {
let _ = zstd::bulk::decompress(f, max_raw).expect("zstd crate decode"); // allocs/frame
}
}
});
}
});
};
decode_pass(); // warmup
let mut samples = Vec::with_capacity(iters);
for _ in 0..iters {
let t = Instant::now();
decode_pass();
samples.push(t.elapsed().as_nanos());
}
let wall: u128 = samples.iter().sum();
let mb_per_sec = c.raw_total as f64 / 1e6 * iters as f64 / (wall as f64 / 1e9).max(1e-9);
let extra = [
("mb_per_sec", mb_per_sec),
("decompressed_mb", c.raw_total as f64 / 1e6),
("frames", c.frames.len() as f64),
("threads", threads as f64),
];
Ok(bench(name, pctl_metrics(&extra, samples, wall)))
}
struct ZstdDecodeCrate;
impl Bencher for ZstdDecodeCrate {
fn id(&self) -> &'static str { "zstd_decode_crate" }
fn run(&self) -> Result<BenchResult> { run_zstd_decode(self.id(), false) }
}
register_bench!(ZstdDecodeCrate);
struct ZstdDecodeSysZeroCopy;
impl Bencher for ZstdDecodeSysZeroCopy {
fn id(&self) -> &'static str { "zstd_decode_sys_zerocopy" }
fn run(&self) -> Result<BenchResult> { run_zstd_decode(self.id(), true) }
}
register_bench!(ZstdDecodeSysZeroCopy);
// ---- commit-burst across targets -------------------------------------------
struct CommitBurstNornirNvme;
impl Bencher for CommitBurstNornirNvme {
fn id(&self) -> &'static str { "commit_burst_skade_nvme" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let (cat, _tmp) = factory::embedded().await?;
Ok(bench(self.id(), commit_burst_metrics(Arc::new(cat)).await?))
})
}
}
register_bench!(CommitBurstNornirNvme);
struct CommitBurstNornirRam;
impl Bencher for CommitBurstNornirRam {
fn id(&self) -> &'static str { "commit_burst_skade_ram" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let tmp = factory::tempdir_in(&factory::ram_dir())?;
let cat = factory::embedded_in(&tmp).await?;
Ok(bench(self.id(), commit_burst_metrics(Arc::new(cat)).await?))
})
}
}
register_bench!(CommitBurstNornirRam);
struct CommitBurstNornirS3;
impl Bencher for CommitBurstNornirS3 {
fn id(&self) -> &'static str { "commit_burst_skade_s3" }
#[cfg(feature = "s3")]
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let (cat, _tmp) = factory::embedded_s3().await?;
Ok(bench(self.id(), commit_burst_metrics(Arc::new(cat)).await?))
})
}
// Built without `--features s3`: self-skip cleanly (honest Static row).
#[cfg(not(feature = "s3"))]
fn run(&self) -> Result<BenchResult> {
Ok(s3_not_compiled_skip(self.id()))
}
}
register_bench!(CommitBurstNornirS3);
#[cfg(feature = "rest")]
struct CommitBurstNessie;
#[cfg(feature = "rest")]
impl Bencher for CommitBurstNessie {
fn id(&self) -> &'static str { "commit_burst_nessie" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let rest = factory::rest_s3(&nessie_uri(), "warehouse").await?;
Ok(bench(self.id(), commit_burst_metrics(Arc::new(rest)).await?))
})
}
}
#[cfg(feature = "rest")]
register_bench!(CommitBurstNessie);
// ---- multi-table atomic commit (skade's flagship `atomic_release`) ----------
// `commit_burst_*` above measures per-table `iceberg::Transaction::commit`
// (single-table atomicity). This bencher measures skade's differentiator: the
// all-or-nothing `RedbCatalog::atomic_release_raw` that flips N table pointers
// in ONE redb write transaction. Each batch stages N property updates, writes
// N metadata blobs, then swaps all N pointers atomically — the exact hot path
// the `.nornir` docs call "publish bench_runs + dep_graph + components
// together". Reports batches/sec, table-flips/sec, and per-batch latency
// percentiles over a fresh embedded warehouse (tempdir on the NVMe work dir).
// Env: NORNIR_ATOMIC_TABLES (default 8), NORNIR_ATOMIC_BATCHES (default 50).
async fn atomic_release_metrics(cat: &skade_katalog::RedbCatalog) -> Result<Map<String, Value>> {
use std::collections::HashMap;
let tables = env_usize("NORNIR_ATOMIC_TABLES", 8).max(1);
let batches = env_usize("NORNIR_ATOMIC_BATCHES", 50).max(1);
let ns = NamespaceIdent::new("atomic".to_string());
if !cat.namespace_exists(&ns).await.unwrap_or(false) {
cat.create_namespace(&ns, Default::default()).await?;
}
let mut idents = Vec::with_capacity(tables);
for t in 0..tables {
let ident = TableIdent::new(ns.clone(), format!("rel_{t:03}"));
if cat.table_exists(&ident).await.unwrap_or(false) {
let _ = cat.drop_table(&ident).await;
}
data::create_node_table(cat, &ident).await?;
idents.push(ident);
}
let wall = Instant::now();
let mut lat = Vec::with_capacity(batches);
for b in 0..batches {
let commits: Vec<_> = idents
.iter()
.map(|ident| {
let mut updates = HashMap::new();
updates.insert("skade.release_seq".to_string(), b.to_string());
(
ident.clone(),
Vec::<iceberg::TableRequirement>::new(),
vec![iceberg::TableUpdate::SetProperties { updates }],
)
})
.collect();
let t0 = Instant::now();
cat.atomic_release_raw(commits).await?;
lat.push(t0.elapsed().as_nanos());
}
let secs = wall.elapsed().as_secs_f64().max(1e-9);
let batches_f = lat.len() as f64;
let flips = batches_f * tables as f64;
Ok(pctl_metrics(
&[
("batches_per_sec", batches_f / secs),
("table_flips_per_sec", flips / secs),
("tables_per_batch", tables as f64),
],
lat,
wall.elapsed().as_nanos(),
))
}
struct SkadeAtomicRelease;
impl Bencher for SkadeAtomicRelease {
fn id(&self) -> &'static str { "skade.atomic_release" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let tmp = factory::tempdir_in(&factory::nvme_dir())?;
let cat = factory::embedded_in(&tmp).await?;
Ok(bench(self.id(), atomic_release_metrics(&cat).await?))
})
}
}
register_bench!(SkadeAtomicRelease);
// ---- skade.* throughput (migrated from the old criterion suite) -------------
// These drive the high-level `skade::Table` API (append / ingest /
// ingest_parallel / ingest_pipelined / read / recast / compression codec) over
// a fresh embedded warehouse on a temp dir — a different code path from the
// catalog-level `data_pipe_*` / `commit_burst_*` benchers above (those go
// through the raw `pipeline` module over an `iceberg::Catalog`). Each emits
// throughput keys (`*_per_sec` / `*_mbs`), `elapsed_ms`, and raw counts.
// No external infra needed: a tempdir on the NVMe work dir (factory::nvme_dir).
use skade_katalog_bench::skade_throughput::{batch as sk_batch, schema as sk_schema};
/// Median wall-clock over `iters` timed runs of `f` (warm up once first).
fn timed_median(iters: usize, mut f: impl FnMut() -> std::time::Duration) -> std::time::Duration {
f(); // warmup
let mut samples: Vec<std::time::Duration> = (0..iters.max(1)).map(|_| f()).collect();
samples.sort_unstable();
samples[samples.len() / 2]
}
/// Fresh embedded skade warehouse + an (optionally partitioned) table on a temp
/// dir under the NVMe work dir. Returns the warehouse handle (kept alive via the
/// returned TempDir) and the table.
async fn sk_open_table(
tmp: &tempfile::TempDir,
partition: Option<&[&str]>,
) -> Result<skade::Table> {
let wh = skade::open(tmp.path()).await?;
let t = match partition {
Some(cols) => {
wh.create_partitioned_table("t", sk_schema().as_ref(), cols)
.await?
}
None => wh.create_table("t", sk_schema().as_ref()).await?,
};
Ok(t)
}
fn sk_tmp() -> Result<tempfile::TempDir> {
factory::tempdir_in(&factory::nvme_dir())
}
/// Single-batch `Table::append` (one snapshot/commit per call) — commit-bound.
/// Sweeps 1k + 10k rows; reports rows/sec at each.
struct SkadeAppend;
impl Bencher for SkadeAppend {
fn id(&self) -> &'static str { "skade.append" }
fn run(&self) -> Result<BenchResult> {
let rt = rt();
let iters = env_usize("SKADE_BENCH_ITERS", 20);
let mut m = Map::new();
let mut headline = 0.0f64;
for &rows in &[1_000usize, 10_000] {
let b = sk_batch(rows);
let tmp = sk_tmp()?;
let mut table = rt.block_on(sk_open_table(&tmp, None))?;
let dur = timed_median(iters, || {
let t = Instant::now();
rt.block_on(async { table.append(std::slice::from_ref(&b)).await.unwrap() });
t.elapsed()
});
let rps = rows as f64 / dur.as_secs_f64().max(1e-9);
m.insert(format!("rows_{rows}_per_sec"), json!(rps));
m.insert(format!("rows_{rows}_elapsed_ms"), json!(dur.as_secs_f64() * 1e3));
headline = headline.max(rps);
}
m.insert("rows_per_sec".into(), json!(headline));
Ok(bench(self.id(), m))
}
}
register_bench!(SkadeAppend);
/// Partitioned (`repo`) single-batch append — adds skade's per-commit identity
/// `partition_key_for` over the unpartitioned `skade.append` path.
struct SkadeAppendPartitioned;
impl Bencher for SkadeAppendPartitioned {
fn id(&self) -> &'static str { "skade.append_partitioned" }
fn run(&self) -> Result<BenchResult> {
let rt = rt();
let iters = env_usize("SKADE_BENCH_ITERS", 20);
let rows = 10_000usize;
let b = sk_batch(rows);
let tmp = sk_tmp()?;
let mut table = rt.block_on(sk_open_table(&tmp, Some(&["repo"])))?;
let dur = timed_median(iters, || {
let t = Instant::now();
rt.block_on(async { table.append(std::slice::from_ref(&b)).await.unwrap() });
t.elapsed()
});
let mut m = Map::new();
m.insert("rows_per_sec".into(), json!(rows as f64 / dur.as_secs_f64().max(1e-9)));
m.insert("rows".into(), json!(rows as f64));
m.insert("elapsed_ms".into(), json!(dur.as_secs_f64() * 1e3));
Ok(bench(self.id(), m))
}
}
register_bench!(SkadeAppendPartitioned);
/// Bulk `Table::ingest` — many batches, commit every 8. Amortises the fsync, so
/// this is the real bulk-load throughput of the high-level ingest API.
struct SkadeIngest;
impl Bencher for SkadeIngest {
fn id(&self) -> &'static str { "skade.ingest" }
fn run(&self) -> Result<BenchResult> {
let rt = rt();
let iters = env_usize("SKADE_BENCH_ITERS", 10);
let n_batches = env_usize("SKADE_INGEST_BATCHES", 64);
let rows = env_usize("SKADE_INGEST_ROWS", 1_000);
let total = (n_batches * rows) as f64;
let dur = timed_median(iters, || {
// Fresh warehouse + table per iteration (ingest mutates state).
let tmp = sk_tmp().unwrap();
let mut table = rt.block_on(sk_open_table(&tmp, None)).unwrap();
let batches: Vec<_> = (0..n_batches).map(|_| sk_batch(rows)).collect();
let t = Instant::now();
rt.block_on(async { table.ingest(batches, 8).await.unwrap() });
let e = t.elapsed();
drop(tmp);
e
});
let mut m = Map::new();
m.insert("rows_per_sec".into(), json!(total / dur.as_secs_f64().max(1e-9)));
m.insert("rows".into(), json!(total));
m.insert("batches".into(), json!(n_batches as f64));
m.insert("commit_every".into(), json!(8.0));
m.insert("elapsed_ms".into(), json!(dur.as_secs_f64() * 1e3));
Ok(bench(self.id(), m))
}
}
register_bench!(SkadeIngest);
/// Full-snapshot `Table::read` (scan → Arrow) across table sizes — 10k / 100k
/// rows by default (`SKADE_READ_FULL=1` adds the 1M-row scale). Absorbs the old
/// criterion `read` group (100k scan) into a size sweep.
struct SkadeReadScan;
impl Bencher for SkadeReadScan {
fn id(&self) -> &'static str { "skade.read_scan" }
fn run(&self) -> Result<BenchResult> {
let rt = rt();
let iters = env_usize("SKADE_BENCH_ITERS", 10);
let mut sizes = vec![10_000usize, 100_000];
if std::env::var("SKADE_READ_FULL").is_ok() {
sizes.push(1_000_000);
}
let mut m = Map::new();
let mut headline = 0.0f64;
for rows in sizes {
let tmp = sk_tmp()?;
let mut table = rt.block_on(sk_open_table(&tmp, None))?;
let per = 10_000usize;
let groups: Vec<Vec<_>> = (0..rows / per).map(|_| vec![sk_batch(per)]).collect();
rt.block_on(async { table.ingest_parallel(groups, 16).await })?;
let dur = timed_median(iters, || {
let t = Instant::now();
let got = rt.block_on(async { table.read().await.unwrap() });
assert_eq!(got.iter().map(|b| b.num_rows()).sum::<usize>(), rows);
t.elapsed()
});
let rps = rows as f64 / dur.as_secs_f64().max(1e-9);
m.insert(format!("rows_{rows}_per_sec"), json!(rps));
m.insert(format!("rows_{rows}_elapsed_ms"), json!(dur.as_secs_f64() * 1e3));
headline = headline.max(rps);
drop(tmp);
}
m.insert("rows_per_sec".into(), json!(headline));
Ok(bench(self.id(), m))
}
}
register_bench!(SkadeReadScan);
/// Parallel-encode saturation — `ingest_parallel` fans the Parquet encode across
/// all cores via the ONE fork-join engine (znippy-zoomies `gatling_forkjoin`),
/// with one sequential writer committing. Encode-heavy on purpose (32 files ×
/// 20k rows). Reports throughput plus the **core-saturation** number: the avg
/// cores the encode kept busy (measured only around the `ingest_parallel` call,
/// so table-open/batch-gen don't dilute it). The gatling payoff is a `cores_busy`
/// near `cores_total`. (Was a rayon-pool 1-core-vs-all comparison; rayon is gone
/// — the engine self-dispatches its own scoped threads, so we measure saturation
/// directly instead of throttling a global pool.)
struct SkadeIngestParallelScaling;
impl Bencher for SkadeIngestParallelScaling {
fn id(&self) -> &'static str { "skade.ingest_parallel_scaling" }
fn run(&self) -> Result<BenchResult> {
let rt = rt();
let iters = env_usize("SKADE_BENCH_ITERS", 10);
let n_files = env_usize("SKADE_SCALE_FILES", 32);
let rows = env_usize("SKADE_SCALE_ROWS", 20_000);
let total = (n_files * rows) as f64;
// Accumulate CPU + wall ONLY around the encode+commit (not setup) so
// cores_busy reflects the gatling fan-out, not single-threaded batch gen.
let mut cpu_acc = 0.0f64;
let mut wall_acc = 0.0f64;
let dur = timed_median(iters, || {
let tmp = sk_tmp().unwrap();
let mut table = rt.block_on(sk_open_table(&tmp, None)).unwrap();
let groups: Vec<Vec<_>> = (0..n_files).map(|_| vec![sk_batch(rows)]).collect();
let cpu0 = proc_cpu_secs();
let t = Instant::now();
rt.block_on(async { table.ingest_parallel(groups, 8).await.unwrap() });
let e = t.elapsed();
cpu_acc += proc_cpu_secs() - cpu0;
wall_acc += e.as_secs_f64();
drop(tmp);
e
});
let cores_busy = if wall_acc > 0.0 { cpu_acc / wall_acc } else { 0.0 };
let ct = cores_total();
let rps = total / dur.as_secs_f64().max(1e-9);
let mut m = Map::new();
m.insert("rows_per_sec".into(), json!(rps));
m.insert("elapsed_ms".into(), json!(dur.as_secs_f64() * 1e3));
m.insert("files".into(), json!(n_files as f64));
m.insert("cores_busy".into(), json!(cores_busy));
m.insert("cores_used".into(), json!(cores_busy));
m.insert("cores_total".into(), json!(ct as f64));
m.insert("core_saturation".into(), json!(cores_busy / ct.max(1) as f64));
Ok(bench(self.id(), m))
}
}
register_bench!(SkadeIngestParallelScaling);
/// Cross-table single-fsync populate — `skade::ingest_multi` (all N tables
/// flipped in ONE redb txn = ONE fsync) vs the per-table baseline (N separate
/// `ingest_parallel` calls = N fsyncs), on a symbol-scan-shaped flush (default
/// 4 tables × 8 files × 20k rows, env-overridable). Both encode the identical
/// data (byte-identical files); only the commit transactions collapse from N to
/// 1. Reports rows/sec for each lane, the measured **fsync count** of each
/// (`skade_katalog::redb_commit_count`, so `multi_fsyncs` == 1 and `sep_fsyncs`
/// == N are visible + regression-catchable), the fsync reduction, and the
/// speedup. LIGHT here; the authoritative numbers come from the Loki serial
/// HEAVY pass (`SKADE_MULTI_TABLES` / `SKADE_MULTI_FILES` / `SKADE_MULTI_ROWS`).
struct SkadeIngestMulti;
impl Bencher for SkadeIngestMulti {
fn id(&self) -> &'static str { "skade.ingest_multi" }
fn run(&self) -> Result<BenchResult> {
use skade::skade_katalog::redb_commit_count;
let rt = rt();
let iters = env_usize("SKADE_BENCH_ITERS", 8);
let n_tables = env_usize("SKADE_MULTI_TABLES", 4);
let n_files = env_usize("SKADE_MULTI_FILES", 8);
let rows = env_usize("SKADE_MULTI_ROWS", 20_000);
let total = (n_tables * n_files * rows) as f64;
// Build one warehouse with n_tables, populate every table via ONE
// ingest_multi (one fsync for the whole flush). Fresh warehouse/iteration.
let mut multi_fsyncs = 0u64;
let multi = timed_median(iters, || {
let tmp = sk_tmp().unwrap();
let mut tables = Vec::with_capacity(n_tables);
rt.block_on(async {
let wh = skade::open(tmp.path()).await.unwrap();
for t in 0..n_tables {
let tab = wh
.create_table(&format!("t{t}"), sk_schema().as_ref())
.await
.unwrap();
let groups: Vec<Vec<_>> =
(0..n_files).map(|_| vec![sk_batch(rows)]).collect();
tables.push((tab.inner().clone(), groups));
}
let cat = wh.catalog();
let before = redb_commit_count();
let t0 = Instant::now();
skade::ingest_multi(cat.as_ref(), tables, &skade::WriteProps::default())
.await
.unwrap();
let e = t0.elapsed();
multi_fsyncs = redb_commit_count() - before;
e
})
});
// Baseline: same shape, but N separate per-table ingest_parallel calls.
let mut sep_fsyncs = 0u64;
let sep = timed_median(iters, || {
let tmp = sk_tmp().unwrap();
rt.block_on(async {
let wh = skade::open(tmp.path()).await.unwrap();
let cat = wh.catalog();
// Pre-generate tables + groups OUTSIDE the timer (symmetric with
// the multi lane, which stages its input before `t0`).
let mut work = Vec::with_capacity(n_tables);
for t in 0..n_tables {
let tab = wh
.create_table(&format!("t{t}"), sk_schema().as_ref())
.await
.unwrap()
.inner()
.clone();
let groups: Vec<Vec<_>> =
(0..n_files).map(|_| vec![sk_batch(rows)]).collect();
work.push((tab, groups));
}
let before = redb_commit_count();
let t0 = Instant::now();
for (tab, groups) in work {
skade::ingest_parallel(cat.as_ref() as &dyn Catalog, tab, groups, 16)
.await
.unwrap();
}
let e = t0.elapsed();
sep_fsyncs = redb_commit_count() - before;
e
})
});
let multi_rps = total / multi.as_secs_f64().max(1e-9);
let sep_rps = total / sep.as_secs_f64().max(1e-9);
let mut m = Map::new();
m.insert("rows_per_sec".into(), json!(multi_rps));
m.insert("multi_rows_per_sec".into(), json!(multi_rps));
m.insert("separate_rows_per_sec".into(), json!(sep_rps));
m.insert("multi_fsyncs".into(), json!(multi_fsyncs as f64));
m.insert("separate_fsyncs".into(), json!(sep_fsyncs as f64));
m.insert(
"fsync_reduction".into(),
json!(if multi_fsyncs > 0 { sep_fsyncs as f64 / multi_fsyncs as f64 } else { 0.0 }),
);
m.insert("speedup".into(), json!(if sep_rps > 0.0 { multi_rps / sep_rps } else { 0.0 }));
m.insert("tables".into(), json!(n_tables as f64));
m.insert("files_per_table".into(), json!(n_files as f64));
m.insert("rows".into(), json!(total));
m.insert("multi_elapsed_ms".into(), json!(multi.as_secs_f64() * 1e3));
m.insert("separate_elapsed_ms".into(), json!(sep.as_secs_f64() * 1e3));
Ok(bench(self.id(), m))
}
}
register_bench!(SkadeIngestMulti);
/// **Bulk-import durability** — the fsync-side populate lever. A many-commit
/// populate (many small files, small `files_per_commit` → many `fast_append`
/// snapshots → many redb commits) run through a warehouse opened at
/// `WriteDurability::Immediate` (fsync per commit — the serving default) vs
/// `WriteDurability::None` (no per-commit fsync — the bulk-import setting). Same
/// data, same snapshots; only the per-commit fsync tax differs. Reports rows/sec
/// for each lane, the commit count (equal), and the speedup. LIGHT here — the
/// authoritative numbers come from the Loki serial HEAVY pass
/// (`SKADE_DUR_FILES` / `SKADE_DUR_ROWS` / `SKADE_DUR_FPC`). On local NVMe the
/// win is modest (fsync is cheap); it grows where fsync latency dominates.
struct SkadeBulkImportDurability;
impl Bencher for SkadeBulkImportDurability {
fn id(&self) -> &'static str { "skade.bulk_import_durability" }
fn run(&self) -> Result<BenchResult> {
use skade::WriteDurability;
let rt = rt();
let iters = env_usize("SKADE_BENCH_ITERS", 8);
let n_files = env_usize("SKADE_DUR_FILES", 64);
let rows = env_usize("SKADE_DUR_ROWS", 8_000);
let fpc = env_usize("SKADE_DUR_FPC", 2); // many commits → fsync-bound
let total = (n_files * rows) as f64;
let mut m = Map::new();
let run_at = |durability: WriteDurability| -> (std::time::Duration, u64) {
let mut commits = 0u64;
let d = timed_median(iters, || {
let tmp = sk_tmp().unwrap();
let e = rt.block_on(async {
let wh = skade::Warehouse::open_with_durability(tmp.path(), durability)
.await
.unwrap();
let mut t = wh.create_table("t", sk_schema().as_ref()).await.unwrap();
let groups: Vec<Vec<_>> =
(0..n_files).map(|_| vec![sk_batch(rows)]).collect();
let t0 = Instant::now();
let s = t.ingest_pipelined(groups, fpc, 8).await.unwrap();
let e = t0.elapsed();
commits = s.commits;
e
});
drop(tmp);
e
});
(d, commits)
};
let (imm, imm_commits) = run_at(WriteDurability::Immediate);
let (none, none_commits) = run_at(WriteDurability::None);
let imm_rps = total / imm.as_secs_f64().max(1e-9);
let none_rps = total / none.as_secs_f64().max(1e-9);
m.insert("immediate_rows_per_sec".into(), json!(imm_rps));
m.insert("immediate_elapsed_ms".into(), json!(imm.as_secs_f64() * 1e3));
m.insert("none_rows_per_sec".into(), json!(none_rps));
m.insert("none_elapsed_ms".into(), json!(none.as_secs_f64() * 1e3));
m.insert("immediate_commits".into(), json!(imm_commits as f64));
m.insert("none_commits".into(), json!(none_commits as f64));
m.insert("speedup".into(), json!(if imm_rps > 0.0 { none_rps / imm_rps } else { 0.0 }));
m.insert("rows_per_sec".into(), json!(none_rps.max(imm_rps)));
m.insert("files".into(), json!(n_files as f64));
Ok(bench(self.id(), m))
}
}
register_bench!(SkadeBulkImportDurability);
/// `skade::recast` — the zero-copy fast path. When a batch already matches the
/// table's field-id schema, recast re-stamps the schema over the same Arc'd
/// buffers (zero copy); only type-mismatched columns are cast+allocated. This
/// contrasts the two over 100k rows: `match_zerocopy` should be orders of
/// magnitude faster than `needs_cast` (which re-materialises one column).
struct SkadeRecast;
impl Bencher for SkadeRecast {
fn id(&self) -> &'static str { "skade.recast" }
fn run(&self) -> Result<BenchResult> {
use skade::arrow_array::{Float64Array, Int32Array, Int64Array, RecordBatch, StringArray};
use skade::arrow_schema::{DataType, Field, Schema};
let rows = env_usize("SKADE_RECAST_ROWS", 100_000);
let iters = env_usize("SKADE_BENCH_ITERS", 50);
let target = Arc::new(Schema::new(vec![
Field::new("a", DataType::Int64, false),
Field::new("b", DataType::Float64, false),
Field::new("c", DataType::Utf8, false),
]));
let b = Arc::new(Float64Array::from((0..rows).map(|i| i as f64).collect::<Vec<_>>()));
let s = Arc::new(StringArray::from((0..rows).map(|i| format!("r{i}")).collect::<Vec<_>>()));
let match_batch = RecordBatch::try_new(
target.clone(),
vec![Arc::new(Int64Array::from((0..rows as i64).collect::<Vec<_>>())), b.clone(), s.clone()],
)?;
let cast_schema = Arc::new(Schema::new(vec![
Field::new("a", DataType::Int32, false), // differs → forces a cast
Field::new("b", DataType::Float64, false),
Field::new("c", DataType::Utf8, false),
]));
let cast_batch = RecordBatch::try_new(
cast_schema,
vec![Arc::new(Int32Array::from((0..rows as i32).collect::<Vec<_>>())), b, s],
)?;
let zc = timed_median(iters, || {
let t = Instant::now();
skade::recast(std::slice::from_ref(&match_batch), target.clone()).unwrap();
t.elapsed()
});
let cast = timed_median(iters, || {
let t = Instant::now();
skade::recast(std::slice::from_ref(&cast_batch), target.clone()).unwrap();
t.elapsed()
});
let mut m = Map::new();
let zc_rps = rows as f64 / zc.as_secs_f64().max(1e-9);
let cast_rps = rows as f64 / cast.as_secs_f64().max(1e-9);
m.insert("match_zerocopy_rows_per_sec".into(), json!(zc_rps));
m.insert("match_zerocopy_elapsed_ms".into(), json!(zc.as_secs_f64() * 1e3));
m.insert("needs_cast_rows_per_sec".into(), json!(cast_rps));
m.insert("needs_cast_elapsed_ms".into(), json!(cast.as_secs_f64() * 1e3));
m.insert("zerocopy_speedup".into(), json!(if cast_rps > 0.0 { zc_rps / cast_rps } else { 0.0 }));
m.insert("rows_per_sec".into(), json!(zc_rps));
m.insert("rows".into(), json!(rows as f64));
Ok(bench(self.id(), m))
}
}
register_bench!(SkadeRecast);
/// Append throughput per Parquet codec — the CPU cost of compression on write
/// (none vs snappy vs zstd). Each is a single-batch append on a fresh table
/// configured with that codec.
struct SkadeCompressionAppend;
impl Bencher for SkadeCompressionAppend {
fn id(&self) -> &'static str { "skade.compression_append" }
fn run(&self) -> Result<BenchResult> {
let rt = rt();
let iters = env_usize("SKADE_BENCH_ITERS", 20);
let rows = 10_000usize;
let b = sk_batch(rows);
let mut m = Map::new();
let mut headline = 0.0f64;
for (name, codec) in [
("none", skade::Compression::UNCOMPRESSED),
("snappy", skade::Compression::SNAPPY),
("zstd", skade::Compression::ZSTD(Default::default())),
] {
let tmp = sk_tmp()?;
let mut table = rt.block_on(async {
let wh = skade::open(tmp.path()).await?;
Ok::<_, anyhow::Error>(
wh.create_table("t", sk_schema().as_ref()).await?.compression(codec),
)
})?;
let dur = timed_median(iters, || {
let t = Instant::now();
rt.block_on(async { table.append(std::slice::from_ref(&b)).await.unwrap() });
t.elapsed()
});
let rps = rows as f64 / dur.as_secs_f64().max(1e-9);
m.insert(format!("{name}_rows_per_sec"), json!(rps));
m.insert(format!("{name}_elapsed_ms"), json!(dur.as_secs_f64() * 1e3));
if name == "zstd" {
headline = rps;
}
drop(tmp);
}
m.insert("rows_per_sec".into(), json!(headline));
m.insert("rows".into(), json!(rows as f64));
Ok(bench(self.id(), m))
}
}
register_bench!(SkadeCompressionAppend);
/// **Single-snapshot append: serial vs all-core encode A/B** — the write-path
/// lever for the Mimir-populate hot shape. A lone large append (`Table::append` /
/// `skade::append_with`) encodes the whole write into ONE Parquet file on ONE
/// core — a serial zstd encode that strands the other ~11 cores (the
/// core-saturation law). `Table::append_parallel` splits the rows into ~one file
/// per core (zero-copy row-aligned slices), fans the encode across all cores on
/// the ONE gatling fork-join engine, and commits every file in a **single**
/// `fast_append` snapshot. Both write the same rows into one snapshot (functionally
/// equivalent — same multiset, same aggregates; NOT byte-identical: the parallel
/// snapshot holds several files). zstd on 200k rows so the encode is CPU-heavy
/// enough for the fan-out to bite. Reports rows/sec + the **core-saturation**
/// (`*_cores_busy`) of each lane: `serial_cores_busy` should sit near 1 (one core
/// encoding), `parallel_cores_busy` well above it (toward `cores_total`), and the
/// parallel lane should be the faster. LIGHT here — the authoritative numbers come
/// from the Loki serial HEAVY pass (`SKADE_APE_ROWS`).
struct SkadeAppendParallelEncode;
impl Bencher for SkadeAppendParallelEncode {
fn id(&self) -> &'static str { "skade.append_parallel_encode" }
fn run(&self) -> Result<BenchResult> {
let rt = rt();
let iters = env_usize("SKADE_BENCH_ITERS", 10);
let rows = env_usize("SKADE_APE_ROWS", 200_000);
let codec = skade::Compression::ZSTD(Default::default());
let b = sk_batch(rows);
// Accumulate CPU + wall ONLY around the append call (not table-open) so
// cores_busy reflects the encode, not single-threaded setup.
// Lane A: serial single-file append (encode pins ONE core).
let (mut a_cpu, mut a_wall) = (0.0f64, 0.0f64);
let serial = timed_median(iters, || {
let tmp = sk_tmp().unwrap();
let mut t = rt.block_on(sk_open_table(&tmp, None)).unwrap();
t.set_compression(codec);
let cpu0 = proc_cpu_secs();
let i = Instant::now();
rt.block_on(async { t.append(std::slice::from_ref(&b)).await.unwrap() });
let e = i.elapsed();
a_cpu += proc_cpu_secs() - cpu0;
a_wall += e.as_secs_f64();
drop(tmp);
e
});
// Lane B: parallel multi-file append (encode fans across all cores).
let (mut bb_cpu, mut bb_wall) = (0.0f64, 0.0f64);
let parallel = timed_median(iters, || {
let tmp = sk_tmp().unwrap();
let mut t = rt.block_on(sk_open_table(&tmp, None)).unwrap();
t.set_compression(codec);
let cpu0 = proc_cpu_secs();
let i = Instant::now();
rt.block_on(async { t.append_parallel(std::slice::from_ref(&b)).await.unwrap() });
let e = i.elapsed();
bb_cpu += proc_cpu_secs() - cpu0;
bb_wall += e.as_secs_f64();
drop(tmp);
e
});
let ct = cores_total();
let ser_rps = rows as f64 / serial.as_secs_f64().max(1e-9);
let par_rps = rows as f64 / parallel.as_secs_f64().max(1e-9);
let ser_cores = if a_wall > 0.0 { a_cpu / a_wall } else { 0.0 };
let par_cores = if bb_wall > 0.0 { bb_cpu / bb_wall } else { 0.0 };
let mut m = Map::new();
m.insert("serial_rows_per_sec".into(), json!(ser_rps));
m.insert("parallel_rows_per_sec".into(), json!(par_rps));
m.insert("serial_elapsed_ms".into(), json!(serial.as_secs_f64() * 1e3));
m.insert("parallel_elapsed_ms".into(), json!(parallel.as_secs_f64() * 1e3));
m.insert("serial_cores_busy".into(), json!(ser_cores));
m.insert("parallel_cores_busy".into(), json!(par_cores));
m.insert("cores_busy".into(), json!(par_cores));
m.insert("cores_total".into(), json!(ct as f64));
m.insert("core_saturation".into(), json!(par_cores / ct.max(1) as f64));
m.insert("speedup".into(), json!(if ser_rps > 0.0 { par_rps / ser_rps } else { 0.0 }));
m.insert("rows_per_sec".into(), json!(par_rps.max(ser_rps)));
m.insert("rows".into(), json!(rows as f64));
Ok(bench(self.id(), m))
}
}
register_bench!(SkadeAppendParallelEncode);
/// `ingest_parallel` (encode-all-then-write) vs `ingest_pipelined` (no-barrier:
/// encoders run ahead of one writer through a bounded channel — overlaps encode
/// with the writer's commit + bounds memory). Same encode-heavy workload.
struct SkadeIngestStrategies;
impl Bencher for SkadeIngestStrategies {
fn id(&self) -> &'static str { "skade.ingest_strategies" }
fn run(&self) -> Result<BenchResult> {
let rt = rt();
let iters = env_usize("SKADE_BENCH_ITERS", 10);
let n_files = env_usize("SKADE_SCALE_FILES", 32);
let rows = env_usize("SKADE_SCALE_ROWS", 20_000);
let total = (n_files * rows) as f64;
let mut m = Map::new();
// parallel
let par = timed_median(iters, || {
let tmp = sk_tmp().unwrap();
let mut t = rt.block_on(sk_open_table(&tmp, None)).unwrap();
let groups: Vec<Vec<_>> = (0..n_files).map(|_| vec![sk_batch(rows)]).collect();
let i = Instant::now();
rt.block_on(async { t.ingest_parallel(groups, 8).await.unwrap() });
let e = i.elapsed();
drop(tmp);
e
});
// pipelined
let pipe = timed_median(iters, || {
let tmp = sk_tmp().unwrap();
let mut t = rt.block_on(sk_open_table(&tmp, None)).unwrap();
let groups: Vec<Vec<_>> = (0..n_files).map(|_| vec![sk_batch(rows)]).collect();
let i = Instant::now();
rt.block_on(async { t.ingest_pipelined(groups, 8, 8).await.unwrap() });
let e = i.elapsed();
drop(tmp);
e
});
let par_rps = total / par.as_secs_f64().max(1e-9);
let pipe_rps = total / pipe.as_secs_f64().max(1e-9);
m.insert("parallel_rows_per_sec".into(), json!(par_rps));
m.insert("parallel_elapsed_ms".into(), json!(par.as_secs_f64() * 1e3));
m.insert("pipelined_rows_per_sec".into(), json!(pipe_rps));
m.insert("pipelined_elapsed_ms".into(), json!(pipe.as_secs_f64() * 1e3));
m.insert("rows_per_sec".into(), json!(par_rps.max(pipe_rps)));
m.insert("files".into(), json!(n_files as f64));
Ok(bench(self.id(), m))
}
}
register_bench!(SkadeIngestStrategies);
/// Un-serialized committer — `ingest_pipelined` (serial commit: write-ALL then
/// commit-ALL) vs `ingest_pipelined_overlap` (the commit fsyncs of already
/// written file-groups overlap the FileIO writes of the later groups). Both
/// produce a byte/row-identical table with the SAME snapshot grouping; only the
/// commit *timing* differs. A **commit-bound shape** (many small files, small
/// `files_per_commit` → many fsyncs) so the hidden fsync tail is observable.
/// Reports rows/sec for each lane, the commit count of each (should be equal),
/// and the speedup. LIGHT here — the authoritative numbers come from the Loki
/// serial HEAVY pass (`SKADE_OVL_FILES` / `SKADE_OVL_ROWS` / `SKADE_OVL_FPC` /
/// `SKADE_OVL_DEPTH`). On local NVMe the win is small (writes are fast relative
/// to fsync); it grows as commit/fsync latency dominates (slow/remote FileIO).
struct SkadeCommitterOverlap;
impl Bencher for SkadeCommitterOverlap {
fn id(&self) -> &'static str { "skade.committer_overlap" }
fn run(&self) -> Result<BenchResult> {
let rt = rt();
let iters = env_usize("SKADE_BENCH_ITERS", 10);
let n_files = env_usize("SKADE_OVL_FILES", 64);
let rows = env_usize("SKADE_OVL_ROWS", 8_000);
let fpc = env_usize("SKADE_OVL_FPC", 2); // many commits → commit-bound
let depth = env_usize("SKADE_OVL_DEPTH", 4);
let total = (n_files * rows) as f64;
let mut m = Map::new();
// Serial-commit pipeline (write-all → commit-all).
let mut ser_commits = 0u64;
let ser = timed_median(iters, || {
let tmp = sk_tmp().unwrap();
let mut t = rt.block_on(sk_open_table(&tmp, None)).unwrap();
let groups: Vec<Vec<_>> = (0..n_files).map(|_| vec![sk_batch(rows)]).collect();
let i = Instant::now();
let s = rt.block_on(async { t.ingest_pipelined(groups, fpc, depth).await.unwrap() });
let e = i.elapsed();
ser_commits = s.commits;
drop(tmp);
e
});
// Overlapped committer (commit fsync ‖ write I/O).
let mut ovl_commits = 0u64;
let ovl = timed_median(iters, || {
let tmp = sk_tmp().unwrap();
let mut t = rt.block_on(sk_open_table(&tmp, None)).unwrap();
let groups: Vec<Vec<_>> = (0..n_files).map(|_| vec![sk_batch(rows)]).collect();
let i = Instant::now();
let s = rt.block_on(async {
t.ingest_pipelined_overlap(groups, fpc, depth).await.unwrap()
});
let e = i.elapsed();
ovl_commits = s.commits;
drop(tmp);
e
});
let ser_rps = total / ser.as_secs_f64().max(1e-9);
let ovl_rps = total / ovl.as_secs_f64().max(1e-9);
m.insert("serial_rows_per_sec".into(), json!(ser_rps));
m.insert("serial_elapsed_ms".into(), json!(ser.as_secs_f64() * 1e3));
m.insert("overlap_rows_per_sec".into(), json!(ovl_rps));
m.insert("overlap_elapsed_ms".into(), json!(ovl.as_secs_f64() * 1e3));
m.insert("serial_commits".into(), json!(ser_commits as f64));
m.insert("overlap_commits".into(), json!(ovl_commits as f64));
m.insert("speedup".into(), json!(if ser_rps > 0.0 { ovl_rps / ser_rps } else { 0.0 }));
m.insert("rows_per_sec".into(), json!(ovl_rps.max(ser_rps)));
m.insert("files".into(), json!(n_files as f64));
Ok(bench(self.id(), m))
}
}
register_bench!(SkadeCommitterOverlap);
/// **Write stat-build placement A/B** — the `ingest_parallel` all-core footer-stat
/// win. skade's bulk write encodes every file across all cores (gatling), then
/// builds each file's prunable `DataFile` by parsing its Parquet footer. That
/// footer parse used to run **serially in the writer tail** (all other cores idle
/// — the core-saturation law); it now rides the **all-core encode burn**. Both
/// lanes here encode identically across cores (so bytes are byte-identical); they
/// differ ONLY in where the per-file footer parse runs:
/// * `serial_stat` lane — parallel encode, then a serial loop parsing each
/// file's footer (the OLD placement);
/// * `parallel_stat` lane — encode + footer parse together inside each gatling
/// worker (the NEW placement, [`skade::write::encode_and_stat_parallel`]).
/// Reports rows/sec + MB/s + the **core-saturation** (`*_cores_busy`) of each
/// lane: the parallel-stat lane should keep more cores busy (the parse no longer
/// idles them) and be ≥ as fast. Encode-heavy shape on purpose (many files). LIGHT
/// here — the authoritative numbers come from the Loki serial HEAVY pass
/// (`SKADE_WSB_FILES` / `SKADE_WSB_ROWS` / `SKADE_WSB_RGSIZE`). The real function
/// (`ingest_parallel`) is also covered by `skade.ingest_parallel_scaling`, whose
/// `cores_busy` should rise with this change.
struct SkadeWriteStatBuild;
impl Bencher for SkadeWriteStatBuild {
fn id(&self) -> &'static str { "skade.write_stat_build" }
fn run(&self) -> Result<BenchResult> {
use bytes::Bytes;
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
use parquet::arrow::ArrowWriter;
use parquet::file::properties::WriterProperties;
use znippy_zoomies::gatling_forkjoin::gatling_map_owned;
let iters = env_usize("SKADE_BENCH_ITERS", 10);
let n_files = env_usize("SKADE_WSB_FILES", 32);
let rows = env_usize("SKADE_WSB_ROWS", 20_000);
let rg_size = env_usize("SKADE_WSB_RGSIZE", 5_000);
let total = (n_files * rows) as f64;
let schema = sk_schema();
// Encode one group to an in-memory Parquet file (CPU only) — the shared
// work both lanes fan across cores. Small row groups → several row-group
// stats per file, so the footer parse has real per-rg work (like skade's).
let encode = {
let schema = schema.clone();
move |b: &skade::arrow_array::RecordBatch| -> Vec<u8> {
let props = WriterProperties::builder()
.set_max_row_group_row_count(Some(rg_size))
.build();
let mut buf: Vec<u8> = Vec::new();
let mut w = ArrowWriter::try_new(&mut buf, schema.clone(), Some(props)).unwrap();
w.write(b).unwrap();
w.close().unwrap();
buf
}
};
// Parse a file's footer + touch its per-column stats — the "build the
// prunable DataFile" work whose *placement* this arm isolates.
let stat = |bytes: &Bytes| -> u64 {
let rdr = ParquetRecordBatchReaderBuilder::try_new(bytes.clone()).unwrap();
let md = rdr.metadata();
let mut acc = 0u64;
for rg in md.row_groups() {
acc += rg.num_rows() as u64;
for c in rg.columns() {
acc = acc.wrapping_add(c.compressed_size() as u64);
}
}
acc
};
let mut m = Map::new();
// Lane A: parallel encode, then SERIAL footer parse (old placement).
let mut a_cpu = 0.0f64;
let mut a_wall = 0.0f64;
let mut bytes_total = 0u64;
let serial = timed_median(iters, || {
let batches: Vec<_> = (0..n_files).map(|_| sk_batch(rows)).collect();
let cpu0 = proc_cpu_secs();
let t = Instant::now();
let encoded: Vec<Bytes> =
gatling_map_owned(batches, |b| Bytes::from(encode(&b)));
let mut acc = 0u64;
for b in &encoded {
acc = acc.wrapping_add(stat(b));
}
std::hint::black_box(acc);
let e = t.elapsed();
a_cpu += proc_cpu_secs() - cpu0;
a_wall += e.as_secs_f64();
bytes_total = encoded.iter().map(|b| b.len() as u64).sum();
e
});
// Lane B: encode + footer parse together in each gatling worker (new).
let mut b_cpu = 0.0f64;
let mut b_wall = 0.0f64;
let parallel = timed_median(iters, || {
let batches: Vec<_> = (0..n_files).map(|_| sk_batch(rows)).collect();
let cpu0 = proc_cpu_secs();
let t = Instant::now();
let done: Vec<(Bytes, u64)> = gatling_map_owned(batches, |b| {
let bytes = Bytes::from(encode(&b));
let s = stat(&bytes);
(bytes, s)
});
let acc: u64 = done.iter().fold(0u64, |a, (_, s)| a.wrapping_add(*s));
std::hint::black_box(acc);
let e = t.elapsed();
b_cpu += proc_cpu_secs() - cpu0;
b_wall += e.as_secs_f64();
e
});
let ct = cores_total();
let ser_rps = total / serial.as_secs_f64().max(1e-9);
let par_rps = total / parallel.as_secs_f64().max(1e-9);
let ser_cores = if a_wall > 0.0 { a_cpu / a_wall } else { 0.0 };
let par_cores = if b_wall > 0.0 { b_cpu / b_wall } else { 0.0 };
let mbs = |rps: f64| (bytes_total as f64) * rps / total / (1024.0 * 1024.0);
m.insert("serial_stat_rows_per_sec".into(), json!(ser_rps));
m.insert("parallel_stat_rows_per_sec".into(), json!(par_rps));
m.insert("serial_stat_elapsed_ms".into(), json!(serial.as_secs_f64() * 1e3));
m.insert("parallel_stat_elapsed_ms".into(), json!(parallel.as_secs_f64() * 1e3));
m.insert("serial_stat_cores_busy".into(), json!(ser_cores));
m.insert("parallel_stat_cores_busy".into(), json!(par_cores));
m.insert("cores_total".into(), json!(ct as f64));
m.insert("parallel_stat_mbs".into(), json!(mbs(par_rps)));
m.insert("serial_stat_mbs".into(), json!(mbs(ser_rps)));
m.insert("speedup".into(), json!(if ser_rps > 0.0 { par_rps / ser_rps } else { 0.0 }));
m.insert("file_bytes".into(), json!(bytes_total as f64));
m.insert("files".into(), json!(n_files as f64));
m.insert("rows_per_sec".into(), json!(par_rps.max(ser_rps)));
Ok(bench(self.id(), m))
}
}
register_bench!(SkadeWriteStatBuild);
/// **Pipelined stat-build placement A/B** — the `ingest_pipelined[_overlap]`
/// all-core footer-stat win (the write-path complement to `skade.write_stat_build`,
/// which covered the non-pipelined `ingest_parallel`). The pipelined paths encode
/// every file across all cores (gatling), then — in the OLD placement — parsed each
/// file's Parquet footer to build its prunable `DataFile` **inside the async write
/// jobs on the tokio runtime** (off the strictly-serial block, but a CPU parse on
/// the runtime, driven by `run_ordered`'s single reorder task → the parses run one
/// at a time, ~1 core busy). It now rides the **all-core encode burn** inside
/// [`skade::write::encode_pipelined`] (via the shared `encode_stat_one` core), so
/// the footer parse fans across cores like the encode. Both lanes encode identically
/// across cores (byte-identical bytes); they differ ONLY in where the footer parse
/// runs:
/// * `runtime_stat` lane — parallel encode, then parse each footer inside async
/// jobs driven on the tokio runtime by `run_ordered` (the OLD placement);
/// * `gatling_stat` lane — encode + footer parse together inside each gatling
/// worker (the NEW placement).
/// Reports rows/sec + MB/s + the **core-saturation** (`*_stat_cores_busy`) of each
/// lane: `gatling_stat_cores_busy` should be well above `runtime_stat_cores_busy`
/// (which stays ~1, the parse serialised on the runtime task) and be ≥ as fast.
/// LIGHT here — authoritative numbers come from the Loki serial HEAVY pass
/// (`SKADE_PSB_FILES` / `SKADE_PSB_ROWS` / `SKADE_PSB_RGSIZE` / `SKADE_PSB_DEPTH`).
/// The real functions (`ingest_pipelined`/`_overlap`) also flow through
/// `skade.committer_overlap`, whose write phase now parses on-core too.
struct SkadePipelinedStatBuild;
impl Bencher for SkadePipelinedStatBuild {
fn id(&self) -> &'static str { "skade.pipelined_stat_build" }
fn run(&self) -> Result<BenchResult> {
use bytes::Bytes;
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
use parquet::arrow::ArrowWriter;
use parquet::file::properties::WriterProperties;
use znippy_zoomies::gatling_forkjoin::gatling_map_owned;
let rt = rt();
let iters = env_usize("SKADE_BENCH_ITERS", 10);
let n_files = env_usize("SKADE_PSB_FILES", 32);
let rows = env_usize("SKADE_PSB_ROWS", 20_000);
let rg_size = env_usize("SKADE_PSB_RGSIZE", 5_000);
let depth = env_usize("SKADE_PSB_DEPTH", 4);
let total = (n_files * rows) as f64;
let schema = sk_schema();
// Encode one group to an in-memory Parquet file (CPU only) — the shared
// work both lanes fan across cores (byte-identical output).
let encode = {
let schema = schema.clone();
move |b: &skade::arrow_array::RecordBatch| -> Vec<u8> {
let props = WriterProperties::builder()
.set_max_row_group_row_count(Some(rg_size))
.build();
let mut buf: Vec<u8> = Vec::new();
let mut w = ArrowWriter::try_new(&mut buf, schema.clone(), Some(props)).unwrap();
w.write(b).unwrap();
w.close().unwrap();
buf
}
};
// Parse a file's footer + touch its per-column stats — the "build the
// prunable DataFile" work whose *placement* this arm isolates.
let stat = |bytes: &Bytes| -> u64 {
let rdr = ParquetRecordBatchReaderBuilder::try_new(bytes.clone()).unwrap();
let md = rdr.metadata();
let mut acc = 0u64;
for rg in md.row_groups() {
acc += rg.num_rows() as u64;
for c in rg.columns() {
acc = acc.wrapping_add(c.compressed_size() as u64);
}
}
acc
};
let mut m = Map::new();
// Lane A: parallel encode, then footer parse inside async jobs on the tokio
// runtime (the OLD pipelined placement). `run_ordered` polls the parse
// futures on its single reorder task → the CPU parses run one at a time.
let mut a_cpu = 0.0f64;
let mut a_wall = 0.0f64;
let mut bytes_total = 0u64;
let runtime_stat = timed_median(iters, || {
let batches: Vec<_> = (0..n_files).map(|_| sk_batch(rows)).collect();
let cpu0 = proc_cpu_secs();
let t = Instant::now();
let encoded: Vec<Bytes> =
gatling_map_owned(batches, |b| Bytes::from(encode(&b)));
let acc = rt.block_on(async {
let jobs = encoded.iter().cloned().map(|bytes| {
let stat = &stat;
async move { Ok::<u64, anyhow::Error>(stat(&bytes)) }
});
let done: Vec<u64> =
znippy_zoomies::gatling::io::run_ordered(jobs, depth.max(1), |x| x)
.await
.unwrap();
done.iter().fold(0u64, |a, s| a.wrapping_add(*s))
});
std::hint::black_box(acc);
let e = t.elapsed();
a_cpu += proc_cpu_secs() - cpu0;
a_wall += e.as_secs_f64();
bytes_total = encoded.iter().map(|b| b.len() as u64).sum();
e
});
// Lane B: encode + footer parse together in each gatling worker (the NEW
// placement — the parse rides the all-core burn).
let mut b_cpu = 0.0f64;
let mut b_wall = 0.0f64;
let gatling_stat = timed_median(iters, || {
let batches: Vec<_> = (0..n_files).map(|_| sk_batch(rows)).collect();
let cpu0 = proc_cpu_secs();
let t = Instant::now();
let done: Vec<(Bytes, u64)> = gatling_map_owned(batches, |b| {
let bytes = Bytes::from(encode(&b));
let s = stat(&bytes);
(bytes, s)
});
let acc: u64 = done.iter().fold(0u64, |a, (_, s)| a.wrapping_add(*s));
std::hint::black_box(acc);
let e = t.elapsed();
b_cpu += proc_cpu_secs() - cpu0;
b_wall += e.as_secs_f64();
e
});
let ct = cores_total();
let a_rps = total / runtime_stat.as_secs_f64().max(1e-9);
let b_rps = total / gatling_stat.as_secs_f64().max(1e-9);
let a_cores = if a_wall > 0.0 { a_cpu / a_wall } else { 0.0 };
let b_cores = if b_wall > 0.0 { b_cpu / b_wall } else { 0.0 };
let mbs = |rps: f64| (bytes_total as f64) * rps / total / (1024.0 * 1024.0);
m.insert("runtime_stat_rows_per_sec".into(), json!(a_rps));
m.insert("gatling_stat_rows_per_sec".into(), json!(b_rps));
m.insert("runtime_stat_elapsed_ms".into(), json!(runtime_stat.as_secs_f64() * 1e3));
m.insert("gatling_stat_elapsed_ms".into(), json!(gatling_stat.as_secs_f64() * 1e3));
m.insert("runtime_stat_cores_busy".into(), json!(a_cores));
m.insert("gatling_stat_cores_busy".into(), json!(b_cores));
m.insert("cores_total".into(), json!(ct as f64));
m.insert("runtime_stat_mbs".into(), json!(mbs(a_rps)));
m.insert("gatling_stat_mbs".into(), json!(mbs(b_rps)));
m.insert("speedup".into(), json!(if a_rps > 0.0 { b_rps / a_rps } else { 0.0 }));
m.insert("file_bytes".into(), json!(bytes_total as f64));
m.insert("files".into(), json!(n_files as f64));
m.insert("rows_per_sec".into(), json!(b_rps.max(a_rps)));
Ok(bench(self.id(), m))
}
}
register_bench!(SkadePipelinedStatBuild);
/// **Read decode strategy A/B** — the `read_all_gatling` local-fd streaming win.
/// `read_all_gatling` on a local `file://` warehouse now decodes each data file
/// straight from its fd (pages read lazily) instead of first slurping every
/// file's whole bytes into memory before decoding a row. Both strategies are
/// byte/row-identical (same reader, same order, same per-file fan-out) — this
/// arm isolates the decode-strategy difference over the SAME files:
/// * `bytes_slurp` lane — read each file whole into a heap `Bytes`, then decode
/// (`try_new(Bytes)`), the old path / the S3-FileIO fallback still uses;
/// * `local_stream` lane — `File::open` + decode straight from the fd
/// (`try_new(File)`), the new local fast path.
/// Both lanes decode sequentially per file (the gatling fan-out is identical
/// across lanes, so this ratio isolates slurp-vs-stream). `slurp_peak_bytes` is
/// the total file bytes the slurp lane holds in memory at once — exactly the
/// peak-RSS the stream lane avoids. LIGHT here; the authoritative numbers +
/// process peak-RSS come from the Loki serial HEAVY pass (`SKADE_READLS_FILES` /
/// `SKADE_READLS_ROWS` / `SKADE_READLS_RGSIZE`).
struct SkadeReadLocalStream;
impl Bencher for SkadeReadLocalStream {
fn id(&self) -> &'static str { "skade.read_local_stream" }
fn run(&self) -> Result<BenchResult> {
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
use parquet::arrow::ArrowWriter;
use parquet::file::properties::WriterProperties;
let iters = env_usize("SKADE_BENCH_ITERS", 10);
let n_files = env_usize("SKADE_READLS_FILES", 16);
let rows = env_usize("SKADE_READLS_ROWS", 100_000);
let rg_size = env_usize("SKADE_READLS_RGSIZE", 20_000);
let total = (n_files * rows) as f64;
let mut m = Map::new();
// Stage `n_files` real Parquet files once (outside the timer). Each holds
// one `sk_batch(rows)`, split into `rg_size`-row row groups.
let tmp = sk_tmp()?;
let schema = sk_schema();
let batch = sk_batch(rows);
let mut paths: Vec<std::path::PathBuf> = Vec::with_capacity(n_files);
let mut slurp_peak_bytes = 0u64;
for k in 0..n_files {
let p = tmp.path().join(format!("data_{k}.parquet"));
let file = std::fs::File::create(&p)?;
let props = WriterProperties::builder()
.set_max_row_group_row_count(Some(rg_size))
.build();
let mut wr = ArrowWriter::try_new(file, schema.clone(), Some(props))?;
wr.write(&batch)?;
wr.close()?;
slurp_peak_bytes = slurp_peak_bytes.max(std::fs::metadata(&p)?.len());
paths.push(p);
}
// The slurp lane holds ALL files' bytes at once (old read_all_gatling
// collected the whole dataset into `raw` before decoding); the stream
// lane holds none of it. Report the whole-dataset byte footprint.
let dataset_bytes: u64 = paths.iter().filter_map(|p| std::fs::metadata(p).ok()).map(|md| md.len()).sum();
// Lane A: whole-file Bytes slurp, then decode.
let slurp = timed_median(iters, || {
let t = Instant::now();
let mut got = 0usize;
for p in &paths {
let raw = bytes::Bytes::from(std::fs::read(p).unwrap());
let rdr = ParquetRecordBatchReaderBuilder::try_new(raw).unwrap().build().unwrap();
for b in rdr { got += b.unwrap().num_rows(); }
}
assert_eq!(got as f64, total);
t.elapsed()
});
// Lane B: stream straight from the fd (no whole-file slurp).
let stream = timed_median(iters, || {
let t = Instant::now();
let mut got = 0usize;
for p in &paths {
let f = std::fs::File::open(p).unwrap();
let rdr = ParquetRecordBatchReaderBuilder::try_new(f).unwrap().build().unwrap();
for b in rdr { got += b.unwrap().num_rows(); }
}
assert_eq!(got as f64, total);
t.elapsed()
});
let slurp_rps = total / slurp.as_secs_f64().max(1e-9);
let stream_rps = total / stream.as_secs_f64().max(1e-9);
m.insert("bytes_slurp_rows_per_sec".into(), json!(slurp_rps));
m.insert("bytes_slurp_elapsed_ms".into(), json!(slurp.as_secs_f64() * 1e3));
m.insert("local_stream_rows_per_sec".into(), json!(stream_rps));
m.insert("local_stream_elapsed_ms".into(), json!(stream.as_secs_f64() * 1e3));
m.insert("speedup".into(), json!(if slurp_rps > 0.0 { stream_rps / slurp_rps } else { 0.0 }));
m.insert("slurp_peak_file_bytes".into(), json!(slurp_peak_bytes as f64));
m.insert("slurp_dataset_bytes".into(), json!(dataset_bytes as f64));
m.insert("files".into(), json!(n_files as f64));
m.insert("rows_total".into(), json!(total));
m.insert("rows_per_sec".into(), json!(stream_rps.max(slurp_rps)));
drop(tmp);
Ok(bench(self.id(), m))
}
}
register_bench!(SkadeReadLocalStream);
// ---- TPC-H: all 22 queries, per-query latency over RedbCatalog -------------
// Shared 8-table fixture (generate + ingest + DataFusion wiring) is built once
// and reused by all 22 query benchers, so setup isn't paid 22×.
#[path = "../tpch_shared.rs"]
mod tpch;
struct TpchFixture {
rt: tokio::runtime::Runtime,
ctx: datafusion::prelude::SessionContext,
_tmp: tempfile::TempDir,
}
fn tpch_fixture() -> &'static TpchFixture {
use std::sync::OnceLock;
static F: OnceLock<TpchFixture> = OnceLock::new();
F.get_or_init(|| {
let rt = tokio::runtime::Builder::new_multi_thread().enable_all().build().expect("tpch rt");
let sf = std::env::var("TPCH_SF").ok().and_then(|v| v.parse().ok()).unwrap_or(0.01);
let (ctx, tmp, _rows) = rt.block_on(tpch::build_ctx(sf)).expect("tpch fixture setup");
TpchFixture { rt, ctx, _tmp: tmp }
})
}
fn tpch_query_result(name: &str, n: i32) -> Result<BenchResult> {
let fx = tpch_fixture();
let sql = tpch::query_sql(n);
let iters = env_usize("TPCH_BENCH_ITERS", 5);
let exec = |q: &str| -> Result<usize> {
let b = fx
.rt
.block_on(async { fx.ctx.sql(q).await?.collect().await })
.map_err(anyhow::Error::from)?;
Ok(b.iter().map(|x| x.num_rows()).sum())
};
let _ = exec(&sql)?; // warmup
let mut samples = Vec::with_capacity(iters);
let mut rows = 0usize;
for _ in 0..iters {
let t = Instant::now();
rows = exec(&sql)?;
samples.push(t.elapsed().as_nanos());
}
let wall: u128 = samples.iter().sum();
Ok(bench(name, pctl_metrics(&[("rows", rows as f64)], samples, wall)))
}
macro_rules! tpch_q {
($s:ident, $id:literal, $n:expr) => {
struct $s;
impl Bencher for $s {
fn id(&self) -> &'static str { $id }
fn run(&self) -> Result<BenchResult> { tpch_query_result($id, $n) }
}
register_bench!($s);
};
}
tpch_q!(TpchQ01, "tpch_q01", 1);
tpch_q!(TpchQ02, "tpch_q02", 2);
tpch_q!(TpchQ03, "tpch_q03", 3);
tpch_q!(TpchQ04, "tpch_q04", 4);
tpch_q!(TpchQ05, "tpch_q05", 5);
tpch_q!(TpchQ06, "tpch_q06", 6);
tpch_q!(TpchQ07, "tpch_q07", 7);
tpch_q!(TpchQ08, "tpch_q08", 8);
tpch_q!(TpchQ09, "tpch_q09", 9);
tpch_q!(TpchQ10, "tpch_q10", 10);
tpch_q!(TpchQ11, "tpch_q11", 11);
tpch_q!(TpchQ12, "tpch_q12", 12);
tpch_q!(TpchQ13, "tpch_q13", 13);
tpch_q!(TpchQ14, "tpch_q14", 14);
tpch_q!(TpchQ15, "tpch_q15", 15);
tpch_q!(TpchQ16, "tpch_q16", 16);
tpch_q!(TpchQ17, "tpch_q17", 17);
tpch_q!(TpchQ18, "tpch_q18", 18);
tpch_q!(TpchQ19, "tpch_q19", 19);
tpch_q!(TpchQ20, "tpch_q20", 20);
tpch_q!(TpchQ21, "tpch_q21", 21);
tpch_q!(TpchQ22, "tpch_q22", 22);
// ---- TPC-H large warehouse (runs by default; scalable) ----------------------
// A full warehouse run: ingest the 8-table warehouse at a scale factor across all
// cores (partitioned parallel ingest), then run all 22 queries over it, reporting
// build vs query time. Runs by default at a moderate scale; tune / disable with:
// TPCH_WAREHOUSE_SF scale factor (default 10 ≈ 60M lineitem rows, a few min;
// 100-200 for the 10-60 min big-iron workout)
// TPCH_WAREHOUSE set to 0 to disable this bench entirely
// TPCH_WAREHOUSE_PARTS ingest partitions (default = all cores)
struct TpchWarehouse;
impl Bencher for TpchWarehouse {
fn id(&self) -> &'static str { "tpch_warehouse" }
fn run(&self) -> Result<BenchResult> {
let sf: f64 = std::env::var("TPCH_WAREHOUSE_SF").ok().and_then(|v| v.parse().ok()).unwrap_or(10.0);
let disabled = std::env::var("TPCH_WAREHOUSE").map(|v| v == "0").unwrap_or(false) || sf <= 0.0;
if disabled {
anyhow::bail!("tpch_warehouse disabled (TPCH_WAREHOUSE=0 or TPCH_WAREHOUSE_SF=0)");
}
let cores = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(8);
let parts = env_usize("TPCH_WAREHOUSE_PARTS", cores).max(1) as i32;
rt().block_on(async move {
let t0 = Instant::now();
let (ctx, _tmp, rows) = tpch::build_ctx_scaled(sf, parts).await?;
let build_s = t0.elapsed().as_secs_f64();
let tq = Instant::now();
let mut result_rows = 0u64;
let (mut slowest_q, mut slowest_s) = (0i32, 0.0f64);
for n in 1..=22 {
let sql = tpch::query_sql(n);
let qt = Instant::now();
let b = ctx.sql(&sql).await?.collect().await?;
let s = qt.elapsed().as_secs_f64();
result_rows += b.iter().map(|x| x.num_rows()).sum::<usize>() as u64;
if s > slowest_s { slowest_s = s; slowest_q = n; }
eprintln!(" warehouse Q{n:<2} {s:8.2}s");
}
let query_s = tq.elapsed().as_secs_f64();
let mut m = Map::new();
m.insert("scale_factor".into(), json!(sf));
m.insert("parts".into(), json!(parts as f64));
m.insert("ingest_rows".into(), json!(rows as f64));
m.insert("build_s".into(), json!(build_s));
m.insert("ingest_rows_per_sec".into(), json!(rows as f64 / build_s.max(1e-9)));
m.insert("query_s".into(), json!(query_s));
m.insert("total_s".into(), json!(build_s + query_s));
m.insert("slowest_query".into(), json!(slowest_q as f64));
m.insert("slowest_query_s".into(), json!(slowest_s));
m.insert("result_rows".into(), json!(result_rows as f64));
Ok(bench("tpch_warehouse", m))
})
}
}
register_bench!(TpchWarehouse);
// ---- TPC-H across catalogs (nornir / Nessie / Polaris) ----------------------
// The same SF≥0.5 warehouse built + all-22 queried over each catalog, so the
// analytical SQL path is comparable apples-to-apples (not just the table_exists
// RPC). nornir embedded runs anywhere; the REST targets need their container +
// the shared RustFS S3 warehouse (BENCH_S3_*), else they skip-as-failure.
// TPCH_COMPARE_SF sets the scale (default 0.5).
fn compare_sf() -> f64 {
// Default to the canonical TPC-H SF1 (~1GB / 8.66M rows); bump for a bigger run.
std::env::var("TPCH_COMPARE_SF").ok().and_then(|v| v.parse().ok()).unwrap_or(1.0)
}
fn compare_parts() -> i32 {
std::thread::available_parallelism().map(|n| n.get()).unwrap_or(8) as i32
}
async fn tpch_suite_metrics(cat: Arc<dyn Catalog>, sf: f64, parts: i32) -> Result<Map<String, Value>> {
let t0 = Instant::now();
let (ctx, rows) = tpch::build_ctx_in(cat, sf, parts).await?;
let build_s = t0.elapsed().as_secs_f64();
let tq = Instant::now();
let (mut slowest_q, mut slowest_s) = (0i32, 0.0f64);
let mut per_query: Vec<(String, f64)> = Vec::with_capacity(22);
for n in 1..=22 {
let sql = tpch::query_sql(n);
let qt = Instant::now();
let _ = ctx.sql(&sql).await?.collect().await?;
let s = qt.elapsed().as_secs_f64();
// per-query time on THIS backend, so the docs renderer can compute
// nornir-vs-competitor speedup per query and surface the most dramatic.
per_query.push((format!("q{n:02}_s"), s));
if s > slowest_s { slowest_s = s; slowest_q = n; }
}
let query_s = tq.elapsed().as_secs_f64();
let mut m = Map::new();
for (k, s) in per_query {
m.insert(k, json!(s));
}
m.insert("scale_factor".into(), json!(sf));
m.insert("rows".into(), json!(rows as f64));
m.insert("build_s".into(), json!(build_s));
m.insert("query_s".into(), json!(query_s));
m.insert("total_s".into(), json!(build_s + query_s));
m.insert("slowest_query".into(), json!(slowest_q as f64));
m.insert("slowest_query_s".into(), json!(slowest_s));
Ok(m)
}
// Storage matrix — each (catalog × storage) combination that's actually
// possible. nornir runs on all three (file-NVMe, file-RAM, S3); Nessie rejects a
// local file warehouse so it's S3-only; Polaris serves FILE end-to-end (its S3
// vending 301s on RustFS — see containers.rs). The REST targets skip-as-failure
// without their container + RustFS S3.
struct TpchNornirFileNvme;
impl Bencher for TpchNornirFileNvme {
fn id(&self) -> &'static str { "tpch_cmp_skade_file_nvme" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let (cat, _tmp) = factory::embedded().await?;
Ok(bench(self.id(), tpch_suite_metrics(Arc::new(cat), compare_sf(), compare_parts()).await?))
})
}
}
register_bench!(TpchNornirFileNvme);
struct TpchNornirFileRam;
impl Bencher for TpchNornirFileRam {
fn id(&self) -> &'static str { "tpch_cmp_skade_file_ram" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let tmp = factory::tempdir_in(&factory::ram_dir())?;
let cat = factory::embedded_in(&tmp).await?;
Ok(bench(self.id(), tpch_suite_metrics(Arc::new(cat), compare_sf(), compare_parts()).await?))
})
}
}
register_bench!(TpchNornirFileRam);
struct TpchNornirS3;
impl Bencher for TpchNornirS3 {
fn id(&self) -> &'static str { "tpch_cmp_skade_s3" }
#[cfg(feature = "s3")]
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let (cat, _tmp) = factory::embedded_s3().await?;
Ok(bench(self.id(), tpch_suite_metrics(Arc::new(cat), compare_sf(), compare_parts()).await?))
})
}
// Built without `--features s3`: self-skip cleanly (honest Static row).
#[cfg(not(feature = "s3"))]
fn run(&self) -> Result<BenchResult> {
Ok(s3_not_compiled_skip(self.id()))
}
}
register_bench!(TpchNornirS3);
#[cfg(feature = "rest")]
struct TpchNessieS3;
#[cfg(feature = "rest")]
impl Bencher for TpchNessieS3 {
fn id(&self) -> &'static str { "tpch_cmp_nessie_s3" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let cat = factory::rest_s3(&nessie_uri(), "warehouse").await?;
Ok(bench(self.id(), tpch_suite_metrics(Arc::new(cat), compare_sf(), compare_parts()).await?))
})
}
}
#[cfg(feature = "rest")]
register_bench!(TpchNessieS3);
#[cfg(feature = "rest")]
struct TpchPolarisFile;
#[cfg(feature = "rest")]
impl Bencher for TpchPolarisFile {
fn id(&self) -> &'static str { "tpch_cmp_polaris_file" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let uri = std::env::var("BENCH_POLARIS_URI")
.unwrap_or_else(|_| "http://localhost:8181/api/catalog".to_string());
let cat = factory::rest_polaris(&uri).await?;
Ok(bench(self.id(), tpch_suite_metrics(Arc::new(cat), compare_sf(), compare_parts()).await?))
})
}
}
#[cfg(feature = "rest")]
register_bench!(TpchPolarisFile);
// ==== SEARCH / DATA-SKIPPING — the warehouse-perf trade ======================
// Proves the warehouse-perf knobs (zstd + per-column bloom + small row-groups +
// per-file min/max bounds): writes are MODESTLY slower + files SMALLER, and
// point/predicate READS skip data files via the Iceberg manifest's min/max
// bounds — collapsing a full scan to a few-file read.
//
// Tables are built with `append` (iceberg's DataFileWriter, which records the
// per-column min/max bounds the scan planner prunes on). The fast
// `ingest_parallel`/`ingest_pipelined` paths build the DataFile by hand and DO
// NOT emit those bounds, so a table written that way CANNOT be file-pruned — the
// knobs only pay off through a bounds-emitting writer (flagged in skade/.nornir).
/// (repos, rows_per_repo) — small + fast by default; `NORNIR_BENCH_FULL=1` scales up.
fn search_params() -> (usize, usize) {
if std::env::var("NORNIR_BENCH_FULL").is_ok() {
(env_usize("SEARCH_REPOS", 32), env_usize("SEARCH_ROWS_PER_REPO", 250_000))
} else {
(env_usize("SEARCH_REPOS", 16), env_usize("SEARCH_ROWS_PER_REPO", 50_000))
}
}
fn search_schema() -> skade::arrow_schema::Schema {
use skade::arrow_schema::{DataType, Field, Schema};
Schema::new(vec![
Field::new("repo", DataType::Utf8, false),
Field::new("symbol", DataType::Utf8, false),
Field::new("value", DataType::Int64, false),
])
}
/// One file's rows, all for a single `repo` (file `repo` min==max) with a
/// disjoint, contiguous `symbol`/`value` range per repo (so min/max bounds are
/// tight and a point lookup prunes to one file).
fn search_repo_batch(repo_idx: usize, rows: usize) -> Result<skade::arrow_array::RecordBatch> {
use skade::arrow_array::{Int64Array, RecordBatch, StringArray};
let base = (repo_idx * rows) as i64;
let repo = format!("repo{repo_idx:02}");
let repos: Vec<&str> = std::iter::repeat(repo.as_str()).take(rows).collect();
let syms: Vec<String> = (0..rows).map(|i| format!("SYM{:08}", base + i as i64)).collect();
let vals: Vec<i64> = (0..rows).map(|i| base + i as i64).collect();
Ok(RecordBatch::try_new(
std::sync::Arc::new(search_schema()),
vec![
std::sync::Arc::new(StringArray::from(repos)),
std::sync::Arc::new(StringArray::from(syms)),
std::sync::Arc::new(Int64Array::from(vals)),
],
)?)
}
/// Sum the bytes of every `.parquet` data file under `dir` (on-disk table size).
fn search_lake_bytes(dir: &std::path::Path) -> u64 {
let mut total = 0u64;
let mut stack = vec![dir.to_path_buf()];
while let Some(d) = stack.pop() {
let Ok(rd) = std::fs::read_dir(&d) else { continue };
for entry in rd.flatten() {
let p = entry.path();
if p.is_dir() {
stack.push(p);
} else if p.extension().and_then(|e| e.to_str()) == Some("parquet") {
total += std::fs::metadata(&p).map(|m| m.len()).unwrap_or(0);
}
}
}
total
}
/// Build a table with `props`, one `append` per repo (one bounds-carrying data
/// file per repo). Returns the handle + the wall time to write all rows.
async fn search_build(
wh: &skade::Warehouse,
name: &str,
props: skade::WriteProps,
repos: usize,
rows: usize,
) -> Result<(skade::Table, std::time::Duration)> {
let mut t = wh.create_table(name, &search_schema()).await?.write_props(props);
let t0 = Instant::now();
for r in 0..repos {
t.append(&[search_repo_batch(r, rows)?]).await?;
}
let elapsed = t0.elapsed();
anyhow::ensure!(t.count().await? == (repos * rows) as u64, "all rows persisted");
Ok((t, elapsed))
}
// ---- WRITE side: tuned (zstd+bloom+small-rg) vs baseline (uncompressed) ------
struct SearchWriteCost;
impl Bencher for SearchWriteCost {
fn id(&self) -> &'static str { "search_write_cost" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
let (repos, rows) = search_params();
let total = (repos * rows) as f64;
let base_tmp = factory::tempdir_in(&factory::nvme_dir())?;
let base_wh = skade::open(base_tmp.path().join("lake")).await?;
let (_bt, base_t) = search_build(
&base_wh, "base",
skade::WriteProps::new(skade::Compression::UNCOMPRESSED),
repos, rows,
).await?;
let base_bytes = search_lake_bytes(base_tmp.path());
let tuned_tmp = factory::tempdir_in(&factory::nvme_dir())?;
let tuned_wh = skade::open(tuned_tmp.path().join("lake")).await?;
let (_tt, tuned_t) = search_build(
&tuned_wh, "tuned",
skade::WriteProps::new(skade::Compression::ZSTD(Default::default()))
.bloom_columns(["symbol"])
.row_group_size(8192),
repos, rows,
).await?;
let tuned_bytes = search_lake_bytes(tuned_tmp.path());
anyhow::ensure!(base_bytes > 0 && tuned_bytes > 0, "both tables wrote parquet bytes");
let mut m = Map::new();
m.insert("baseline_rows_per_sec".into(), json!(total / base_t.as_secs_f64().max(1e-9)));
m.insert("tuned_rows_per_sec".into(), json!(total / tuned_t.as_secs_f64().max(1e-9)));
m.insert("write_slowdown_pct".into(),
json!((tuned_t.as_secs_f64() / base_t.as_secs_f64().max(1e-9) - 1.0) * 100.0));
m.insert("baseline_mb".into(), json!(base_bytes as f64 / 1e6));
m.insert("tuned_mb".into(), json!(tuned_bytes as f64 / 1e6));
m.insert("size_reduction_pct".into(),
json!((1.0 - tuned_bytes as f64 / base_bytes as f64) * 100.0));
m.insert("rows".into(), json!(total));
Ok(bench(self.id(), m))
})
}
}
register_bench!(SearchWriteCost);
// ---- READ side: file-skipping via manifest min/max bounds (the win) ---------
struct SearchDataSkip;
impl Bencher for SearchDataSkip {
fn id(&self) -> &'static str { "search_data_skip" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
use skade::ScanFilter;
let (repos, rows) = search_params();
let tmp = factory::tempdir_in(&factory::nvme_dir())?;
let wh = skade::open(tmp.path().join("lake")).await?;
let (t, _) = search_build(
&wh, "syms",
skade::WriteProps::new(skade::Compression::ZSTD(Default::default()))
.bloom_columns(["symbol"])
.row_group_size(8192),
repos, rows,
).await?;
let target = format!("repo{:02}", repos / 2);
let full = t.plan_stats(None).await?;
let pruned = t.plan_stats(Some(&ScanFilter::eq("repo", target.as_str()))).await?;
anyhow::ensure!(full.data_files == repos as u64, "one file per repo: {}", full.data_files);
anyhow::ensure!(
pruned.data_files < full.data_files,
"data-skipping must prune files (planned {}/{}); bounds missing?",
pruned.data_files, full.data_files
);
// Symbol point-lookup: a single value in the target repo's range.
let sym = format!("SYM{:08}", ((repos / 2) * rows) as i64 + (rows as i64 / 2));
let sym_plan = t.plan_stats(Some(&ScanFilter::eq("symbol", sym.as_str()))).await?;
// Wall: full scan vs pruned read (mean of K reads), and verify the
// filtered read returns EXACTLY the target repo's rows (correctness).
let iters = env_usize("SEARCH_READ_ITERS", 5).max(1);
let mut full_ns = 0u128;
let mut pruned_ns = 0u128;
let mut got = 0usize;
for _ in 0..iters {
let a = Instant::now();
let _ = t.read().await?;
full_ns += a.elapsed().as_nanos();
let b = Instant::now();
let rb = t.read_filtered(&ScanFilter::eq("repo", target.as_str()), &[]).await?;
pruned_ns += b.elapsed().as_nanos();
got = rb.iter().map(|x| x.num_rows()).sum();
}
anyhow::ensure!(got == rows, "filtered read returns exactly {} rows, got {}", rows, got);
let full_ms = (full_ns as f64 / iters as f64) / 1e6;
let pruned_ms = (pruned_ns as f64 / iters as f64) / 1e6;
let mut m = Map::new();
m.insert("total_data_files".into(), json!(full.data_files as f64));
m.insert("repo_pruned_files".into(), json!(pruned.data_files as f64));
m.insert("repo_files_skipped".into(), json!((full.data_files - pruned.data_files) as f64));
m.insert("repo_files_skip_pct".into(),
json!((full.data_files - pruned.data_files) as f64 / full.data_files as f64 * 100.0));
m.insert("symbol_pruned_files".into(), json!(sym_plan.data_files as f64));
m.insert("symbol_files_skipped".into(), json!((full.data_files - sym_plan.data_files) as f64));
m.insert("full_rows_planned".into(), json!(full.rows_planned as f64));
m.insert("pruned_rows_planned".into(), json!(pruned.rows_planned as f64));
m.insert("full_scan_ms".into(), json!(full_ms));
m.insert("pruned_scan_ms".into(), json!(pruned_ms));
m.insert("scan_speedup_x".into(), json!(full_ms / pruned_ms.max(1e-9)));
Ok(bench(self.id(), m))
})
}
}
register_bench!(SearchDataSkip);
// ---- READ side: bloom-pruned POINT LOOKUP (the 5–50× point-lookup win) -------
// Isolates the written-bloom read path (`Table::lookup` → `read::lookup_gatling`,
// which probes each key column's Parquet bloom filter and decodes only the
// surviving row groups). Two tables with IDENTICAL compression + row-group
// layout are built; the ONLY difference is a per-column bloom on `symbol`. A
// point lookup on the bloom table skips every row group the key can't be in and
// decodes one; the no-bloom table must decode every row group of every file. The
// gap IS the bloom win. Both must resolve the SAME row (bloom never drops a
// present key) and both must return `None` for a definitely-absent key.
fn row_i64(b: &skade::arrow_array::RecordBatch, col: &str) -> i64 {
use skade::arrow_array::{Array, Int64Array};
b.column_by_name(col)
.unwrap()
.as_any()
.downcast_ref::<Int64Array>()
.unwrap()
.value(0)
}
struct SearchPointLookup;
impl Bencher for SearchPointLookup {
fn id(&self) -> &'static str { "search_point_lookup" }
fn run(&self) -> Result<BenchResult> {
rt().block_on(async {
use skade::Scalar;
let (repos, rows) = search_params();
let rg = 8192usize;
let bloom_tmp = factory::tempdir_in(&factory::nvme_dir())?;
let bloom_wh = skade::open(bloom_tmp.path().join("lake")).await?;
let (bloom_t, _) = search_build(
&bloom_wh, "bloom",
skade::WriteProps::new(skade::Compression::ZSTD(Default::default()))
.bloom_columns(["symbol"])
.row_group_size(rg),
repos, rows,
).await?;
// Baseline: same layout + codec, NO bloom — so the only variable is
// whether the point lookup can prune row groups.
let plain_tmp = factory::tempdir_in(&factory::nvme_dir())?;
let plain_wh = skade::open(plain_tmp.path().join("lake")).await?;
let (plain_t, _) = search_build(
&plain_wh, "plain",
skade::WriteProps::new(skade::Compression::ZSTD(Default::default()))
.row_group_size(rg),
repos, rows,
).await?;
// A symbol living in exactly one repo's file (one row group).
let sym = format!("SYM{:08}", ((repos / 2) * rows) as i64 + (rows as i64 / 2));
let key = [("symbol", Scalar::Str(sym.clone()))];
// Correctness: identical row from both paths; absent key ⇒ None on both.
let bloom_row = bloom_t.lookup(&key, None).await?.expect("symbol present (bloom)");
let plain_row = plain_t.lookup(&key, None).await?.expect("symbol present (plain)");
anyhow::ensure!(
row_i64(&bloom_row, "value") == row_i64(&plain_row, "value"),
"bloom and plain lookups must resolve the same row"
);
let absent = [("symbol", Scalar::Str("SYM-DEFINITELY-ABSENT".to_string()))];
anyhow::ensure!(
bloom_t.lookup(&absent, None).await?.is_none(),
"a definitely-absent key must resolve to None"
);
let iters = env_usize("SEARCH_LOOKUP_ITERS", 20).max(1);
let (mut bloom_ns, mut plain_ns) = (0u128, 0u128);
for _ in 0..iters {
let a = Instant::now();
let _ = bloom_t.lookup(&key, None).await?;
bloom_ns += a.elapsed().as_nanos();
let b = Instant::now();
let _ = plain_t.lookup(&key, None).await?;
plain_ns += b.elapsed().as_nanos();
}
let bloom_us = (bloom_ns as f64 / iters as f64) / 1e3;
let plain_us = (plain_ns as f64 / iters as f64) / 1e3;
let mut m = Map::new();
m.insert("ops_sec".into(), json!(if bloom_us > 0.0 { 1e6 / bloom_us } else { 0.0 }));
m.insert("bloom_lookup_us".into(), json!(bloom_us));
m.insert("plain_lookup_us".into(), json!(plain_us));
m.insert("lookup_speedup_x".into(), json!(plain_us / bloom_us.max(1e-9)));
m.insert("repos".into(), json!(repos as f64));
m.insert("rows_per_repo".into(), json!(rows as f64));
m.insert("row_group_size".into(), json!(rg as f64));
Ok(bench(self.id(), m))
})
}
}
register_bench!(SearchPointLookup);
// ---- WRITE side: partition-scan accelerator (uniform-collapse) --------------
// Times the `partition_key_for` single-partition proof on the identity-partition
// write hot path: the offsets+`memcmp` uniform-collapse (`uniform_str_value`) vs
// the naive per-row `value(i)` scan it replaces, over one uniform string column.
// A/B in-proc via the `skade::bench_partition` bench hook (feature `bench`), the
// same pattern as `skade_catalog_key_build`. LIGHT by default; the heavy N sweep
// goes to the quiet bench box via `BENCH_PARTITION_ROWS` (Loki/Odin).
struct PartitionKeyScan;
impl Bencher for PartitionKeyScan {
fn id(&self) -> &'static str { "skade_partition_key_scan" }
fn run(&self) -> Result<BenchResult> {
use std::hint::black_box;
use skade::arrow_array::StringArray;
use skade::bench_partition::{naive_str_value, uniform_str_value};
let rows = env_usize("BENCH_PARTITION_ROWS", 250_000);
let iters = env_usize("BENCH_PARTITION_ITERS", 200);
// The identity-partition write shape: one repo value across every row.
let col = StringArray::from(vec!["repo42"; rows]);
// Both must agree on the value for the uniform column (fast == naive).
anyhow::ensure!(
uniform_str_value(&col) == naive_str_value(&col) && uniform_str_value(&col) == Some("repo42"),
"fast and naive partition scans must agree on the uniform value"
);
let time = |f: &dyn Fn() -> usize| -> f64 {
let warm = (iters / 10).max(1);
for _ in 0..warm { black_box(f()); }
let t = Instant::now();
let mut s = 0usize;
for _ in 0..iters { s = s.wrapping_add(black_box(f())); }
black_box(s);
t.elapsed().as_nanos() as f64 / iters as f64
};
let fast_ns = time(&|| uniform_str_value(&col).map(|s| s.len()).unwrap_or(0));
let naive_ns = time(&|| naive_str_value(&col).map(|s| s.len()).unwrap_or(0));
let mut m = Map::new();
m.insert("ops_sec".into(), json!(if fast_ns > 0.0 { 1e9 / fast_ns } else { 0.0 }));
m.insert("fast_ns_per_scan".into(), json!(fast_ns));
m.insert("naive_ns_per_scan".into(), json!(naive_ns));
m.insert("speedup_x".into(), json!(naive_ns / fast_ns.max(1e-9)));
m.insert("rows".into(), json!(rows as f64));
Ok(bench(self.id(), m))
}
}
register_bench!(PartitionKeyScan);
// ==== OURS vs RIVALS — skade vs reference Iceberg vs raw Parquet =============
// A head-to-head of four write+read data planes over ONE identical dataset and
// ONE storage root, reported so that a reader can separate the *decode* cost
// from the *fan-out* contribution:
//
// • skade — ours: `skade::Table::{ingest_pipelined,read}` — all-core
// gatling encode + snapshot commit, all-core gatling scan.
// • iceberg — rival: upstream iceberg-rust's stock writer stack
// (`DataFileWriterBuilder` + `ParquetWriterBuilder` +
// `fast_append`) and stock scan (`table.scan()…to_arrow()`),
// driven over the *same* RedbCatalog + LocalFsStorageFactory
// (`data::ingest` / `data::scan_count`).
// • parquet — rival (the floor): raw `parquet::arrow::ArrowWriter` files,
// NO catalog / NO snapshots / NO ACID, read back
// SINGLE-THREADED with `ParquetRecordBatchReaderBuilder`.
// • parquet_par — the SAME raw files, read back with the SAME gatling
// row-group fan-out skade's scan uses. `parquet_par` minus
// `parquet` is the fan-out contribution; `skade` minus
// `parquet_par` is what the catalog + snapshot layer costs.
//
// ---- WHAT IS HELD EQUAL, AND WHAT IS NOT ------------------------------------
// Equal, by construction:
// · the dataset — `data::synthetic_batches`, byte-for-byte identical rows;
// · the storage root, the tempdir-per-pass discipline, the warmup+median
// protocol, and the correctness gate (every pass must scan back exactly the
// rows it wrote);
// · the Parquet codec — skade's `WriteProps::default()` is UNCOMPRESSED +
// dictionary-on, which is exactly `WriterProperties::builder().build()`;
// · the READ OUTPUT CONTRACT — every arm materialises a `Vec<RecordBatch>`
// holding all rows, because that is what `skade::Table::read` returns. (The
// parquet arm used to decode-and-drop each batch, which never holds the
// result and so recycles one hot allocation — a silent advantage.)
// · the DECODE BATCH SIZE — `OXR_SCAN_BATCH_ROWS` (8192) everywhere, matching
// skade's `read::SCAN_BATCH_ROWS`. (The parquet arm used to take the reader
// default of 1024 rows — a silent disadvantage.)
// · the ROW-GROUP GRANULARITY — `OXR_BATCH` rows per row group, which is the
// unit of read parallelism for both skade's scan and `parquet_par`. skade's
// writer emits one row group per input batch; the parquet arm now sets
// `set_max_row_group_row_count(OXR_BATCH)` so its files carve the same way.
//
// NOT equal, and INTRINSIC to what each writer is — do not "fix" these, read the
// `files` / `row_groups` metrics beside each number:
// · FILE CADENCE. skade's `ingest_pipelined` writes ONE FILE PER INPUT BATCH
// (20 files at the defaults): that per-file split IS its encode fan-out.
// iceberg-rust's `DataFileWriter` writes one data file per commit
// (`commit_every` batches → 3 files), and the raw-Parquet arms follow the
// same commit cadence (3 files). The row-group count is equal (20) even
// though the file count is not, so the read fan-out has the same number of
// work units to hand out in every arm.
// · WRITE PARALLELISM. skade's write is deliberately all-core (that is the
// ingest story); iceberg's and raw Parquet's are single-writer. This is why
// `write_cores_busy` and `read_cores_busy` are reported SEPARATELY — see
// below.
// · the iceberg arm's decode batch size is whatever its scan chooses; it is
// not a knob the caller reaches.
//
// ---- WHY THE CORE COUNT IS PER PHASE ----------------------------------------
// The single `cores_busy` this bench used to publish was Δcpu/Δwall over the
// WHOLE pass — write AND read. Dividing the *read* throughput by it therefore
// charged skade's all-core *encode* against its scan, and produced a per-core
// figure that meant nothing. Each phase now samples `CLOCK_PROCESS_CPUTIME_ID`
// (nanosecond process-wide CPU, all threads) around itself, so
// `read_cores_busy` is the scan's own core count and
// `read_rows_per_sec_per_core` is a number you can actually compare across rows
// of the table. `cores_busy` is retained (whole-pass) for warehouse continuity.
//
// Env: OXR_ROWS (default 1_000_000), OXR_BATCH (50_000), OXR_COMMIT_EVERY (8),
// OXR_ITERS (5), OXR_RAM=1 (tmpfs root).
use std::path::Path;
use std::path::PathBuf as OxrPathBuf;
use std::time::Duration as OxrDuration;
/// Decode batch size for every arm we control — skade's `read::SCAN_BATCH_ROWS`.
/// The Parquet reader's own default is 1024; leaving the control arms there was
/// an unequal knob, not a property of the format.
const OXR_SCAN_BATCH_ROWS: usize = 8192;
struct OxrParams {
rows: usize,
batch: usize,
commit_every: usize,
iters: usize,
ram: bool,
}
fn oxr_params() -> OxrParams {
OxrParams {
rows: env_usize("OXR_ROWS", 1_000_000),
batch: env_usize("OXR_BATCH", 50_000),
commit_every: env_usize("OXR_COMMIT_EVERY", 8).max(1),
iters: env_usize("OXR_ITERS", 5).max(1),
ram: std::env::var("OXR_RAM").is_ok(),
}
}
fn oxr_root(ram: bool) -> std::path::PathBuf {
if ram { factory::ram_dir() } else { factory::nvme_dir() }
}
/// The one dataset every engine writes — deterministic, so all three see
/// byte-identical rows.
fn oxr_dataset(p: &OxrParams) -> Result<Vec<arrow_array::RecordBatch>> {
let schema = data::node_arrow_schema_standalone()?;
Ok(data::synthetic_batches(schema, p.rows, p.batch))
}
fn oxr_median(mut v: Vec<OxrDuration>) -> f64 {
v.sort_unstable();
v[v.len() / 2].as_secs_f64()
}
/// Process CPU-seconds consumed so far (Linux `/proc/self/stat` `utime`+`stime`,
/// clock ticks ÷ 100). Whole-pass granularity only — the tick is 10 ms, which is
/// twice a whole read phase at these sizes, so the per-phase numbers use
/// [`proc_cpu_secs_fine`] instead. Kept for the retained whole-pass `cores_busy`.
fn proc_cpu_secs() -> f64 {
// Field 2 (comm) can contain spaces/parens, so split *after* the trailing
// ')': field 3 (state) is then index 0, utime (14) → 11, stime (15) → 12.
let s = std::fs::read_to_string("/proc/self/stat").unwrap_or_default();
let Some(i) = s.rfind(')') else { return 0.0 };
let f: Vec<&str> = s[i + 1..].split_whitespace().collect();
let utime: f64 = f.get(11).and_then(|x| x.parse().ok()).unwrap_or(0.0);
let stime: f64 = f.get(12).and_then(|x| x.parse().ok()).unwrap_or(0.0);
(utime + stime) / 100.0
}
/// Process-wide CPU-seconds at **nanosecond** resolution
/// (`CLOCK_PROCESS_CPUTIME_ID` sums every thread's on-CPU time). `/proc/self/stat`
/// quantises to a 10 ms tick, and a read phase here is ~5 ms — measuring a
/// per-phase core count off the tick counter would be pure noise.
fn proc_cpu_secs_fine() -> f64 {
let mut ts = libc::timespec { tv_sec: 0, tv_nsec: 0 };
// SAFETY: `clock_gettime` writes only the `timespec` we hand it.
if unsafe { libc::clock_gettime(libc::CLOCK_PROCESS_CPUTIME_ID, &mut ts) } != 0 {
return 0.0;
}
ts.tv_sec as f64 + ts.tv_nsec as f64 * 1e-9
}
/// Total logical cores on the box (denominator for `core_saturation`).
fn cores_total() -> usize {
std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1)
}
/// Time one phase and the CPU-seconds it burned across ALL threads.
/// Returns `(wall, cpu, value)`.
fn oxr_phase<T>(f: impl FnOnce() -> Result<T>) -> Result<(OxrDuration, f64, T)> {
let cpu0 = proc_cpu_secs_fine();
let t0 = Instant::now();
let v = f()?;
let wall = t0.elapsed();
let cpu = proc_cpu_secs_fine() - cpu0;
Ok((wall, cpu, v))
}
/// The physical layout an arm actually produced — reported beside its numbers so
/// "identical cadence" is a measured claim, not a comment. `row_groups` is the
/// number of work units a row-group-parallel reader can hand out.
#[derive(Clone, Copy, Default)]
struct OxrLayout {
files: u64,
row_groups: u64,
/// Total on-disk Parquet bytes — the scan's INPUT volume. With
/// `out_bytes` (the Arrow the scan materialises) this bounds the memory
/// traffic a scan can possibly generate, which is what settles the
/// "is the scan bandwidth-bound?" question against the box's STREAM ceiling.
bytes: u64,
}
/// Walk `dir` for `*.parquet` and total the files + their row groups (footer
/// reads only, no data). Called ONLY on the warmup pass, outside every timer.
fn oxr_layout_of(dir: &Path) -> OxrLayout {
fn walk(dir: &Path, out: &mut Vec<OxrPathBuf>) {
let Ok(rd) = std::fs::read_dir(dir) else { return };
for e in rd.flatten() {
let p = e.path();
if p.is_dir() {
walk(&p, out);
} else if p.extension().is_some_and(|x| x == "parquet") {
out.push(p);
}
}
}
let mut paths = Vec::new();
walk(dir, &mut paths);
let mut l = OxrLayout { files: paths.len() as u64, row_groups: 0, bytes: 0 };
for p in &paths {
if let Ok((_, _, rg_rows)) = skade::parquet_layout(p) {
l.row_groups += rg_rows.len() as u64;
}
l.bytes += std::fs::metadata(p).map(|m| m.len()).unwrap_or(0);
}
l
}
/// Bytes of Arrow the scan materialised (`get_array_memory_size` over every
/// column of every batch) — the scan's OUTPUT volume. Measured on the warmup
/// pass only, outside every timer.
fn oxr_out_bytes(batches: &[arrow_array::RecordBatch]) -> u64 {
batches
.iter()
.map(|b| b.columns().iter().map(|c| c.get_array_memory_size() as u64).sum::<u64>())
.sum()
}
/// One timed pass: both phases' wall AND CPU, the rows scanned back, and (warmup
/// pass only) the layout that was written.
struct OxrPass {
write: OxrDuration,
write_cpu: f64,
read: OxrDuration,
read_cpu: f64,
rows: u64,
layout: OxrLayout,
/// Arrow bytes the scan materialised (warmup pass only; 0 otherwise).
out_bytes: u64,
}
/// What [`oxr_measure`] hands to [`oxr_metrics`].
struct OxrStats {
write_s: f64,
read_s: f64,
/// Avg cores the WRITE phase kept busy (Σ write cpu / Σ write wall).
write_cores: f64,
/// Avg cores the READ phase kept busy (Σ read cpu / Σ read wall) — the
/// denominator of `read_rows_per_sec_per_core`.
read_cores: f64,
/// Whole-pass Δcpu/Δwall, the historical `cores_busy`. Retained so the
/// warehouse series does not break; NOT the per-core denominator.
cores_busy: f64,
layout: OxrLayout,
out_bytes: u64,
}
/// Run `pass` once as warmup (row-count checked, layout captured), then `iters`
/// timed passes. Each pass MUST scan back exactly `expect` rows.
fn oxr_measure(
iters: usize,
expect: u64,
mut pass: impl FnMut(bool) -> Result<OxrPass>,
) -> Result<OxrStats> {
let warm = pass(true)?;
anyhow::ensure!(warm.rows == expect, "warmup scanned {} rows, expected {expect}", warm.rows);
let mut ws = Vec::with_capacity(iters);
let mut rs = Vec::with_capacity(iters);
let (mut w_wall, mut w_cpu, mut r_wall, mut r_cpu) = (0.0f64, 0.0f64, 0.0f64, 0.0f64);
// Whole-pass sample (historical `cores_busy`).
let cpu0 = proc_cpu_secs();
let wall0 = Instant::now();
for _ in 0..iters {
let p = pass(false)?;
anyhow::ensure!(p.rows == expect, "pass scanned {} rows, expected {expect}", p.rows);
w_wall += p.write.as_secs_f64();
w_cpu += p.write_cpu;
r_wall += p.read.as_secs_f64();
r_cpu += p.read_cpu;
ws.push(p.write);
rs.push(p.read);
}
let wall = wall0.elapsed().as_secs_f64();
let cores_busy = if wall > 0.0 { (proc_cpu_secs() - cpu0) / wall } else { 0.0 };
Ok(OxrStats {
write_s: oxr_median(ws),
read_s: oxr_median(rs),
write_cores: if w_wall > 0.0 { w_cpu / w_wall } else { 0.0 },
read_cores: if r_wall > 0.0 { r_cpu / r_wall } else { 0.0 },
cores_busy,
layout: warm.layout,
out_bytes: warm.out_bytes,
})
}
fn oxr_metrics(p: &OxrParams, s: &OxrStats) -> Map<String, Value> {
let rows = p.rows as f64;
let total = cores_total();
let read_rps = rows / s.read_s.max(1e-9);
let write_rps = rows / s.write_s.max(1e-9);
let mut m = Map::new();
m.insert("write_rows_per_sec".into(), json!(write_rps));
m.insert("read_rows_per_sec".into(), json!(read_rps));
m.insert("write_ms".into(), json!(s.write_s * 1e3));
m.insert("read_ms".into(), json!(s.read_s * 1e3));
m.insert("rows".into(), json!(rows));
m.insert("batches".into(), json!(p.rows.div_ceil(p.batch.max(1)) as f64));
m.insert("commit_every".into(), json!(p.commit_every as f64));
m.insert("iters".into(), json!(p.iters as f64));
m.insert("storage".into(), json!(if p.ram { "ram" } else { "nvme" }));
// --- PER-PHASE core saturation. The read number is the one that belongs in
// the head-to-head table; the whole-pass `cores_busy` mixes the all-core
// encode into the scan and is kept only for series continuity.
m.insert("write_cores_busy".into(), json!(s.write_cores));
m.insert("read_cores_busy".into(), json!(s.read_cores));
m.insert("cores_busy".into(), json!(s.cores_busy));
m.insert("cores_used".into(), json!(s.cores_busy));
m.insert("cores_total".into(), json!(total as f64));
m.insert("core_saturation".into(), json!(s.cores_busy / total.max(1) as f64));
m.insert("read_core_saturation".into(), json!(s.read_cores / total.max(1) as f64));
// --- PER-CORE throughput: first-class, because an absolute rows/s beside a
// core count that differs 6× between rows invites exactly one misreading.
m.insert("read_rows_per_sec_per_core".into(), json!(read_rps / s.read_cores.max(1e-9)));
m.insert("write_rows_per_sec_per_core".into(), json!(write_rps / s.write_cores.max(1e-9)));
// --- LAYOUT actually written (files ≠ equal across arms by design;
// row_groups — the read fan-out's work units — are equal).
m.insert("files".into(), json!(s.layout.files as f64));
m.insert("row_groups".into(), json!(s.layout.row_groups as f64));
// --- MEMORY TRAFFIC the scan can possibly generate: Parquet bytes read in
// plus Arrow bytes materialised out, over the read wall. An UPPER bound on
// useful traffic (real DRAM traffic adds decode intermediates but subtracts
// whatever stays in the 128 MiB of L3) — compare it against the box's
// measured STREAM ceiling to decide whether "bandwidth-bound" is the story.
m.insert("file_bytes".into(), json!(s.layout.bytes as f64));
if s.out_bytes > 0 {
// Only for arms whose scan hands the batches back to us (the iceberg
// arm's `scan_count` consumes its own stream, so it reports no output
// volume and no bandwidth — an absent number, not a guessed one).
let traffic = (s.layout.bytes + s.out_bytes) as f64;
m.insert("scan_out_bytes".into(), json!(s.out_bytes as f64));
m.insert("read_gb_per_sec".into(), json!(traffic / s.read_s.max(1e-9) / 1e9));
}
// Headline = write throughput (the primary ingest story); read is the second column.
m.insert("rows_per_sec".into(), json!(write_rps));
m
}
/// skade (ours): the flagship **no-barrier gatling pipeline**
/// (`skade::Table::ingest_pipelined`) — the ONE fork-join engine (znippy-zoomies
/// `gatling_forkjoin`) fans the Parquet encode across all cores while a single
/// sequential writer commits, overlapped through a bounded channel. Each input
/// batch becomes its own file so the encode has per-core fan-out (that is what
/// drives `write_cores_busy` toward `cores_total`, and it is also why this arm
/// has 20 files where the rivals have 3 — see the layout note in the section
/// header). Read-back is the all-core `skade::Table::read` scan.
struct OxrSkade;
impl Bencher for OxrSkade {
fn id(&self) -> &'static str { "oxr_skade" }
fn run(&self) -> Result<BenchResult> {
let p = oxr_params();
let root = oxr_root(p.ram);
let batches = oxr_dataset(&p)?;
let rt = rt();
let expect = p.rows as u64;
let schema = data::node_arrow_schema_standalone()?;
// Bounded-channel depth = cores, so up to `cores_total` files are encoded
// in flight ahead of the lone writer (the full channel is the backpressure).
let depth = cores_total();
let stats = oxr_measure(p.iters, expect, |want_layout| {
let tmp = factory::tempdir_in(&root)?;
let lake = tmp.path().join("lake");
let out: Result<OxrPass> = rt.block_on(async {
let wh = skade::open(lake.clone()).await?;
let mut table = wh.create_table("t", schema.as_ref()).await?;
// One file per batch → per-core encode fan-out across the gatling pool.
let groups: Vec<Vec<_>> = batches.iter().map(|b| vec![b.clone()]).collect();
// `oxr_phase` is sync; the two async phases sample the same
// nanosecond process CPU clock inline.
let cpu0 = proc_cpu_secs_fine();
let t0 = Instant::now();
table.ingest_pipelined(groups, p.commit_every, depth).await?;
let write = t0.elapsed();
let write_cpu = proc_cpu_secs_fine() - cpu0;
let cpu1 = proc_cpu_secs_fine();
let t1 = Instant::now();
let got = table.read().await?;
let read = t1.elapsed();
let read_cpu = proc_cpu_secs_fine() - cpu1;
let rows: u64 = got.iter().map(|b| b.num_rows() as u64).sum();
let out_bytes = if want_layout { oxr_out_bytes(&got) } else { 0 };
drop(got);
Ok(OxrPass {
write, write_cpu, read, read_cpu, rows,
layout: OxrLayout::default(), out_bytes,
})
});
let mut out = out?;
if want_layout {
out.layout = oxr_layout_of(&lake);
}
Ok(out)
})?;
Ok(bench(self.id(), oxr_metrics(&p, &stats)))
}
}
register_bench!(OxrSkade);
/// skade (ours), **single-core read**: the SAME `skade::Table::read` scan as
/// `oxr_skade`, over the SAME one-file-per-batch layout, but run through
/// `skade::with_scan_serial` — its manifest parse, footer load, and Parquet
/// decode dispatch on ONE core instead of fanning across the gatling pool. The
/// write phase is unchanged (all-core encode — `with_scan_serial` gates only the
/// read fast path), so the ONLY difference from `oxr_skade` is the scan's fan
/// width. This is skade's own single↔multi read scaling made visible, the exact
/// analogue of `oxr_parquet` (serial) beside `oxr_parquet_par` (row-group
/// parallel): `oxr_skade`'s `read_rows_per_sec` ÷ this arm's is skade's measured
/// scan scaling on this box, and `read_cores_busy` should read ~1 here and
/// well above it in `oxr_skade`. Row order is byte-identical to the all-core
/// scan (gatling preserves input order; the serial map IS that order), so the
/// correctness gate passes both.
struct OxrSkadeSerial;
impl Bencher for OxrSkadeSerial {
fn id(&self) -> &'static str { "oxr_skade_serial" }
fn run(&self) -> Result<BenchResult> {
let p = oxr_params();
let root = oxr_root(p.ram);
let batches = oxr_dataset(&p)?;
let rt = rt();
let expect = p.rows as u64;
let schema = data::node_arrow_schema_standalone()?;
let depth = cores_total();
let stats = oxr_measure(p.iters, expect, |want_layout| {
let tmp = factory::tempdir_in(&root)?;
let lake = tmp.path().join("lake");
// `with_scan_serial` sets a thread-local read on the block_on thread —
// the very thread that polls the read future's synchronous decode — so
// the scan runs single-core. It does NOT touch the write path (that
// gatling engine never consults the flag), so the encode stays
// all-core and only the READ differs from `oxr_skade`.
let out: Result<OxrPass> = skade::with_scan_serial(true, || {
rt.block_on(async {
let wh = skade::open(lake.clone()).await?;
let mut table = wh.create_table("t", schema.as_ref()).await?;
let groups: Vec<Vec<_>> = batches.iter().map(|b| vec![b.clone()]).collect();
let cpu0 = proc_cpu_secs_fine();
let t0 = Instant::now();
table.ingest_pipelined(groups, p.commit_every, depth).await?;
let write = t0.elapsed();
let write_cpu = proc_cpu_secs_fine() - cpu0;
let cpu1 = proc_cpu_secs_fine();
let t1 = Instant::now();
let got = table.read().await?;
let read = t1.elapsed();
let read_cpu = proc_cpu_secs_fine() - cpu1;
let rows: u64 = got.iter().map(|b| b.num_rows() as u64).sum();
let out_bytes = if want_layout { oxr_out_bytes(&got) } else { 0 };
drop(got);
Ok(OxrPass {
write, write_cpu, read, read_cpu, rows,
layout: OxrLayout::default(), out_bytes,
})
})
});
let mut out = out?;
if want_layout {
out.layout = oxr_layout_of(&lake);
}
Ok(out)
})?;
Ok(bench(self.id(), oxr_metrics(&p, &stats)))
}
}
register_bench!(OxrSkadeSerial);
/// Reference Iceberg (rival): upstream iceberg-rust stock writer + scan over the
/// SAME RedbCatalog + LocalFsStorageFactory (`data::ingest` / `data::scan_count`).
/// One data file per commit (3 at the defaults) — the writer's own cadence.
struct OxrIceberg;
impl Bencher for OxrIceberg {
fn id(&self) -> &'static str { "oxr_iceberg" }
fn run(&self) -> Result<BenchResult> {
let p = oxr_params();
let root = oxr_root(p.ram);
let batches = oxr_dataset(&p)?;
let rt = rt();
let expect = p.rows as u64;
let stats = oxr_measure(p.iters, expect, |want_layout| {
let tmp = factory::tempdir_in(&root)?;
let dir = tmp.path().to_path_buf();
let out: Result<OxrPass> = rt.block_on(async {
let cat = factory::embedded_in(&tmp).await?;
let ns = NamespaceIdent::new("bench".to_string());
if !cat.namespace_exists(&ns).await.unwrap_or(false) {
cat.create_namespace(&ns, Default::default()).await?;
}
let ident = TableIdent::new(ns, "t".to_string());
let table = data::create_node_table(&cat, &ident).await?;
let cpu0 = proc_cpu_secs_fine();
let t0 = Instant::now();
let (table, _stats) =
data::ingest(&cat, table, batches.iter().cloned(), p.commit_every).await?;
let write = t0.elapsed();
let write_cpu = proc_cpu_secs_fine() - cpu0;
let cpu1 = proc_cpu_secs_fine();
let (rows, read) = data::scan_count(&table).await?;
let read_cpu = proc_cpu_secs_fine() - cpu1;
// `scan_count` consumes its own stream — no batches come back
// here, so this arm reports no Arrow output volume.
Ok(OxrPass {
write, write_cpu, read, read_cpu, rows,
layout: OxrLayout::default(), out_bytes: 0,
})
});
let mut out = out?;
if want_layout {
out.layout = oxr_layout_of(&dir);
}
Ok(out)
})?;
Ok(bench(self.id(), oxr_metrics(&p, &stats)))
}
}
register_bench!(OxrIceberg);
/// Write the raw-Parquet layout the two floor arms share: one file per
/// `commit_every` batches (the same commit cadence the catalog arms use), with
/// the row group capped at `OXR_BATCH` rows so the files carve into the SAME
/// number of row groups — the same read-parallelism work units — skade's
/// one-file-per-batch layout produces. Returns the files in write order.
fn oxr_parquet_write(
dir: &Path,
batches: &[arrow_array::RecordBatch],
schema: &arrow_schema::SchemaRef,
p: &OxrParams,
) -> Result<Vec<OxrPathBuf>> {
use parquet::arrow::ArrowWriter;
use parquet::file::properties::WriterProperties;
// UNCOMPRESSED + dictionary-on is `WriterProperties` default AND
// skade's `WriteProps::default()` — the codec is genuinely equal.
let props = WriterProperties::builder()
.set_max_row_group_row_count(Some(p.batch.max(1)))
.build();
let mut files = Vec::new();
for (gi, chunk) in batches.chunks(p.commit_every).enumerate() {
let path = dir.join(format!("part-{gi:05}.parquet"));
let file = std::fs::File::create(&path)?;
let mut wr = ArrowWriter::try_new(file, schema.clone(), Some(props.clone()))?;
for b in chunk {
wr.write(b)?;
}
wr.close()?;
files.push(path);
}
Ok(files)
}
/// Raw Parquet (rival / the floor): `ArrowWriter` files, no catalog / no
/// snapshots, read back **single-threaded** with `ParquetRecordBatchReaderBuilder`
/// at skade's own batch size, materialising the full `Vec<RecordBatch>` skade's
/// `read()` returns. This is the honest per-core decode cost of the format.
struct OxrParquet;
impl Bencher for OxrParquet {
fn id(&self) -> &'static str { "oxr_parquet" }
fn run(&self) -> Result<BenchResult> {
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
let p = oxr_params();
let root = oxr_root(p.ram);
let batches = oxr_dataset(&p)?;
let expect = p.rows as u64;
let schema = batches.first().map(|b| b.schema()).ok_or_else(|| anyhow::anyhow!("empty dataset"))?;
let stats = oxr_measure(p.iters, expect, |want_layout| {
let tmp = factory::tempdir_in(&root)?;
let dir: &Path = tmp.path();
let (write, write_cpu, files) = oxr_phase(|| oxr_parquet_write(dir, &batches, &schema, &p))?;
// READ — one stream, in file order, materialised like `Table::read`.
let (read, read_cpu, out) = oxr_phase(|| {
let mut out: Vec<arrow_array::RecordBatch> = Vec::new();
for path in &files {
let file = std::fs::File::open(path)?;
let rdr = ParquetRecordBatchReaderBuilder::try_new(file)?
.with_batch_size(OXR_SCAN_BATCH_ROWS)
.build()?;
for b in rdr {
out.push(b?);
}
}
Ok(out)
})?;
let rows: u64 = out.iter().map(|b| b.num_rows() as u64).sum();
let out_bytes = if want_layout { oxr_out_bytes(&out) } else { 0 };
drop(out);
let layout = if want_layout { oxr_layout_of(dir) } else { OxrLayout::default() };
Ok(OxrPass { write, write_cpu, read, read_cpu, rows, layout, out_bytes })
})?;
Ok(bench(self.id(), oxr_metrics(&p, &stats)))
}
}
register_bench!(OxrParquet);
/// Raw Parquet, **row-group-parallel** read (the fan-out control): the exact same
/// files `oxr_parquet` writes, decoded through the ONE sanctioned fan-out
/// primitive (znippy-zoomies `gatling_forkjoin::gatling_map_owned`) over the same
/// unit shape skade's scan uses — a contiguous run of row groups per worker,
/// footer parsed once per file, each worker on its own fd, results kept in input
/// order so the row order matches the serial read.
///
/// This arm exists because "skade 3× raw Parquet" is unreadable without it: the
/// raw arm reads single-threaded, so that gap conflates skade's decode with a
/// fan-out the control simply declined to do. `parquet_par − parquet` is the
/// fan-out; `skade − parquet_par` is what the catalog + snapshot layer adds.
///
/// `OXR_PAR_UNITS` (default = logical cores) caps the number of decode units, so
/// sweeping it 1,2,4,…,32 traces the scan's SCALING CURVE on this exact data. A
/// curve that keeps climbing to the unit ceiling says the scan is NOT
/// bandwidth-bound; one that flattens early says it is. That sweep is the
/// evidence behind the bandwidth verdict, and it needs no PMU.
struct OxrParquetPar;
impl Bencher for OxrParquetPar {
fn id(&self) -> &'static str { "oxr_parquet_par" }
fn run(&self) -> Result<BenchResult> {
use parquet::arrow::arrow_reader::{
ArrowReaderMetadata, ArrowReaderOptions, ParquetRecordBatchReaderBuilder,
};
use znippy_zoomies::gatling_forkjoin::gatling_map_owned;
let p = oxr_params();
let root = oxr_root(p.ram);
let batches = oxr_dataset(&p)?;
let expect = p.rows as u64;
let schema = batches.first().map(|b| b.schema()).ok_or_else(|| anyhow::anyhow!("empty dataset"))?;
let target = env_usize("OXR_PAR_UNITS", cores_total()).max(1);
let stats = oxr_measure(p.iters, expect, |want_layout| {
let tmp = factory::tempdir_in(&root)?;
let dir: &Path = tmp.path();
let (write, write_cpu, files) = oxr_phase(|| oxr_parquet_write(dir, &batches, &schema, &p))?;
let (read, read_cpu, out) = oxr_phase(|| {
// Footers ONCE per file (Arc-backed, cheap clone per unit).
let mut metas: Vec<(OxrPathBuf, ArrowReaderMetadata, usize)> =
Vec::with_capacity(files.len());
let mut total_rg = 0usize;
for path in &files {
let f = std::fs::File::open(path)?;
let meta = ArrowReaderMetadata::load(&f, ArrowReaderOptions::default())?;
let n_rg = meta.metadata().num_row_groups();
total_rg += n_rg;
metas.push((path.clone(), meta, n_rg));
}
if total_rg == 0 {
return Ok(Vec::new());
}
// Units = contiguous row-group runs, each file split in
// proportion to its share of the total — skade's own topology.
let mut units: Vec<(OxrPathBuf, ArrowReaderMetadata, Vec<usize>)> = Vec::new();
for (path, meta, n_rg) in &metas {
let want = (target * *n_rg).div_ceil(total_rg).clamp(1, *n_rg);
let (base, rem) = (*n_rg / want, *n_rg % want);
let mut start = 0usize;
for b in 0..want {
let len = base + usize::from(b < rem);
if len == 0 {
continue;
}
units.push((path.clone(), meta.clone(), (start..start + len).collect()));
start += len;
}
}
let decoded = gatling_map_owned(units, |(path, meta, rgs)| -> Result<Vec<arrow_array::RecordBatch>> {
let f = std::fs::File::open(&path)?;
let rdr = ParquetRecordBatchReaderBuilder::new_with_metadata(f, meta)
.with_row_groups(rgs)
.with_batch_size(OXR_SCAN_BATCH_ROWS)
.build()?;
Ok(rdr.collect::<std::result::Result<Vec<_>, _>>()?)
});
let mut out: Vec<arrow_array::RecordBatch> = Vec::new();
for d in decoded {
out.extend(d?);
}
Ok(out)
})?;
let rows: u64 = out.iter().map(|b| b.num_rows() as u64).sum();
let out_bytes = if want_layout { oxr_out_bytes(&out) } else { 0 };
drop(out);
let layout = if want_layout { oxr_layout_of(dir) } else { OxrLayout::default() };
Ok(OxrPass { write, write_cpu, read, read_cpu, rows, layout, out_bytes })
})?;
Ok(bench(self.id(), oxr_metrics(&p, &stats)))
}
}
register_bench!(OxrParquetPar);