serve/examples/loadtable_bench.rs
//! `loadTable` A/B: **what routing the metadata read through the catalog's own
//! `FileIO` costs the local-fs path that everyone already runs.**
//!
//! The object-store change moved `rest`/`grpc` off a direct
//! `tokio::fs::read(strip_prefix("file://"))` and onto
//! `table.file_io().new_input(loc).read()`. That is what lets an `s3://…`
//! metadata-location be fetched at all — but `loadTable` is on the hot path of
//! every Doris/Trino/Spark query PLAN, so a regression there is paid by every
//! query, not just the S3 ones. This bench measures the two reads against the
//! SAME file so the difference is the indirection and nothing else.
//!
//! Arms (each one `BenchResult` JSON line on stdout — the shared nordisk
//! envelope, not a bespoke format):
//!
//! * `serve.loadtable.metadata_read.tokio_fs` — the pre-change read.
//! * `serve.loadtable.metadata_read.file_io_localfs` — the post-change read,
//! same `file://` table. The regression-risk arm; `overhead_x` against the
//! `tokio_fs` arm is the number that matters.
//! * `serve.loadtable.metadata_read.file_io_objectstore` (feature
//! `object-store`) — the newly reachable read: an `s3://…` location served
//! through a skade object store. Backed by the embedded local-fs store so
//! the arm needs no MinIO and measures the code path, not a network.
//! * `serve.loadtable.rest.{local_fs,object_store}` — the same load
//! end-to-end over the REST front, where the HTTP hop is included and the
//! read is a small fraction. This is the arm an operator actually feels.
//!
//! Run: cargo run --release --example loadtable_bench
//! cargo run --release --features object-store --example loadtable_bench
//! Knobs: LTB_ITERS (reads per arm, default 500), LTB_WARMUP (default 50).
use std::time::Instant;
use anyhow::Result;
use iceberg::{Catalog, NamespaceIdent, TableIdent};
use nornir_bench::BenchResult;
use serde_json::{Map, json};
use skade_katalog_serve::ServeConfig;
fn env_usize(key: &str, default: usize) -> usize {
std::env::var(key)
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(default)
}
/// One arm's timings, reduced to the shared envelope.
struct Arm {
name: &'static str,
iters: usize,
total_s: f64,
bytes: usize,
}
impl Arm {
fn per_op_us(&self) -> f64 {
self.total_s * 1e6 / self.iters as f64
}
fn to_result(&self) -> BenchResult {
let mut m = Map::new();
m.insert("iters".into(), json!(self.iters));
m.insert("per_op_us".into(), json!(self.per_op_us()));
m.insert(
"ops_per_sec".into(),
json!(self.iters as f64 / self.total_s.max(1e-9)),
);
m.insert("metadata_bytes".into(), json!(self.bytes));
BenchResult::measured(format!("serve.loadtable.{}", self.name), m)
}
}
/// Write a warehouse with the data plane, then hand back the location of the
/// metadata document `loadTable` would serve, plus a live catalog to read it
/// through. Dropping the `Warehouse` first releases the redb write lock.
async fn seeded(cfg: &ServeConfig) -> Result<(iceberg::table::Table, String)> {
let cat = skade_katalog_serve::open_catalog(cfg).await?;
let table = cat
.load_table(&TableIdent::new(
NamespaceIdent::new("main".into()),
"personer".into(),
))
.await?;
let loc = table.metadata_location().unwrap_or_default().to_string();
Ok((table, loc))
}
/// The pre-change read: strip `file://`, hand the rest to the OS.
async fn read_tokio_fs(loc: &str) -> Result<usize> {
let path = loc.strip_prefix("file://").unwrap_or(loc);
Ok(tokio::fs::read(path).await?.len())
}
/// The post-change read: through the storage the catalog is configured with.
async fn read_file_io(table: &iceberg::table::Table, loc: &str) -> Result<usize> {
Ok(table.file_io().new_input(loc)?.read().await?.len())
}
async fn time<F, Fut>(name: &'static str, iters: usize, warmup: usize, mut op: F) -> Result<Arm>
where
F: FnMut() -> Fut,
Fut: std::future::Future<Output = Result<usize>>,
{
let mut bytes = 0;
for _ in 0..warmup {
bytes = op().await?;
}
let t0 = Instant::now();
for _ in 0..iters {
// The result is consumed so the read cannot be optimised away, and a
// short read would show up as a changed byte count in the row.
bytes = op().await?;
}
Ok(Arm {
name,
iters,
total_s: t0.elapsed().as_secs_f64(),
bytes,
})
}
/// One `loadTable` over the REST front, socket to parsed body. No HTTP client
/// dep in this crate — a raw `Connection: close` GET is the whole protocol here.
async fn rest_load(addr: &std::net::SocketAddr) -> Result<usize> {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let mut sock = tokio::net::TcpStream::connect(addr).await?;
sock.write_all(
format!(
"GET /v1/namespaces/main/tables/personer HTTP/1.1\r\nHost: {addr}\r\n\
Connection: close\r\n\r\n"
)
.as_bytes(),
)
.await?;
let mut raw = Vec::new();
sock.read_to_end(&mut raw).await?;
Ok(raw.len())
}
#[tokio::main(flavor = "multi_thread")]
async fn main() -> Result<()> {
let iters = env_usize("LTB_ITERS", 500);
let warmup = env_usize("LTB_WARMUP", 50);
let tmp = tempfile::TempDir::new()?;
let mut arms: Vec<Arm> = Vec::new();
// ── local-fs warehouse: the shape every existing deployment runs ──
let root = tmp.path().join("localfs");
{
let wh = skade::Warehouse::open(&root).await?;
let mut t = wh.table_or_create("personer", &sample_schema()).await?;
t.append(&[sample_batch(256)]).await?;
}
let mut local_cfg = ServeConfig::new(&root);
local_cfg.rest = Some("127.0.0.1:0".parse()?);
let (local_table, local_loc) = seeded(&local_cfg).await?;
arms.push(
time("metadata_read.tokio_fs", iters, warmup, || {
read_tokio_fs(&local_loc)
})
.await?,
);
arms.push(
time("metadata_read.file_io_localfs", iters, warmup, || {
read_file_io(&local_table, &local_loc)
})
.await?,
);
drop(local_table);
// ── object-store warehouse: the newly reachable shape ──
#[cfg(feature = "object-store")]
let _os_keepalive = {
use skade::object_store::ObjectStoreConfig;
use skade_katalog_serve::WarehouseStorage;
let db_path = tmp.path().join("os-catalog.redb");
let store_cfg = ObjectStoreConfig::LocalFs {
root: tmp.path().join("objectstore"),
};
let warehouse_uri = "s3://benchbucket/skade-warehouse";
{
let wh = skade::Warehouse::open_with_object_store_config(
&db_path,
warehouse_uri,
store_cfg.clone(),
)
.await?;
let mut t = wh.table_or_create("personer", &sample_schema()).await?;
t.append(&[sample_batch(256)]).await?;
}
let mut os_cfg = ServeConfig::new(tmp.path().join("os-root"));
os_cfg.db_path = db_path;
os_cfg.warehouse_uri = Some(warehouse_uri.to_string());
os_cfg.storage = WarehouseStorage::ObjectStore(store_cfg);
os_cfg.rest = Some("127.0.0.1:0".parse()?);
let (os_table, os_loc) = seeded(&os_cfg).await?;
arms.push(
time("metadata_read.file_io_objectstore", iters, warmup, || {
read_file_io(&os_table, &os_loc)
})
.await?,
);
let (rest, _) = skade_katalog_serve::spawn(&os_cfg).await?;
let rest = rest.expect("rest bound");
arms.push(
time("rest.object_store", iters / 5, warmup / 5, || {
rest_load(&rest)
})
.await?,
);
// The catalog holds the redb lock; keep it alive until the run is done.
os_table
};
let (rest, _) = skade_katalog_serve::spawn(&local_cfg).await?;
let rest = rest.expect("rest bound");
arms.push(time("rest.local_fs", iters / 5, warmup / 5, || rest_load(&rest)).await?);
// ── report ──
for arm in &arms {
println!("{}", serde_json::to_string(&arm.to_result())?);
}
let find = |n: &str| arms.iter().find(|a| a.name == n).map(Arm::per_op_us);
eprintln!("# arm per_op_us ops/s");
for arm in &arms {
eprintln!(
"# {:<38} {:>9.3} {:>8.0}",
arm.name,
arm.per_op_us(),
arm.iters as f64 / arm.total_s.max(1e-9)
);
}
if let (Some(old), Some(new)) = (
find("metadata_read.tokio_fs"),
find("metadata_read.file_io_localfs"),
) {
// The verdict this bench exists for: FileIO must not make the read that
// every existing deployment already does measurably slower.
eprintln!(
"# FileIO vs tokio::fs on the SAME file: {:.2}x ({})",
new / old.max(1e-9),
if new / old.max(1e-9) < 1.25 {
"no regression"
} else {
"REGRESSION"
}
);
}
Ok(())
}
fn sample_schema() -> std::sync::Arc<skade::arrow_schema::Schema> {
use skade::arrow_schema::{DataType, Field, Schema};
std::sync::Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("namn", DataType::Utf8, false),
]))
}
fn sample_batch(n: i64) -> skade::arrow_array::RecordBatch {
use skade::arrow_array::{Int64Array, RecordBatch, StringArray};
RecordBatch::try_new(
sample_schema(),
vec![
std::sync::Arc::new(Int64Array::from((0..n).collect::<Vec<_>>())),
std::sync::Arc::new(StringArray::from(
(0..n).map(|i| format!("rad-{i}")).collect::<Vec<_>>(),
)),
],
)
.expect("batch")
}