xtask/examples/nornir-bench.rs
//! nornir bench suite for the knut workspace.
//!
//! Invoked by `nornir bench run knut` (or `cargo bench -p xtask`), which
//! parses stdout as one `nornir::bench::BenchRun` JSON line.
//!
//! Everything here is self-contained: synthetic batches, tempdirs on local
//! disk (`/tmp`), no containers. Container-bound performance (FalkorDB
//! ingest, Spark reads) lives in the `#[ignore]` integration tests, not in
//! the release gate.
use std::sync::Arc;
use std::time::Instant;
use anyhow::Result;
use arrow_array::{Float64Array, Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType as ArrowDataType, Field, Schema};
use nornir_bench::BenchResult;
use nornir_bench::api::Bencher;
use serde_json::json;
const ROWS: usize = 100_000;
fn sample_batch(rows: usize) -> RecordBatch {
let ids: Vec<i64> = (0..rows as i64).collect();
let names: Vec<String> = (0..rows).map(|i| format!("node_{i}")).collect();
let scores: Vec<f64> = (0..rows).map(|i| i as f64 * 0.5).collect();
RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("id", ArrowDataType::Int64, false),
Field::new("name", ArrowDataType::Utf8, false),
Field::new("score", ArrowDataType::Float64, false),
])),
vec![
Arc::new(Int64Array::from(ids)),
Arc::new(StringArray::from(names)),
Arc::new(Float64Array::from(scores)),
],
)
.unwrap()
}
fn metrics(
rows: usize,
bytes: u64,
write_s: f64,
read_s: f64,
) -> serde_json::Map<String, serde_json::Value> {
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(rows));
m.insert("file_bytes".into(), json!(bytes));
m.insert("write_s".into(), json!(write_s));
m.insert("read_s".into(), json!(read_s));
m.insert("write_rows_per_s".into(), json!(rows as f64 / write_s));
m.insert("read_rows_per_s".into(), json!(rows as f64 / read_s));
m
}
/// Round-trip one format through `graphar::io` on a 100k-row batch.
fn format_round_trip(file_type: graphar::FileType) -> Result<BenchResult> {
let dir = tempfile::tempdir()?; // /tmp — local disk, never T9
let path = dir.path().join(format!("chunk0.{}", file_type.extension()));
let batch = sample_batch(ROWS);
let t = Instant::now();
graphar::io::write_chunk(&path, std::slice::from_ref(&batch), &file_type)?;
let write_s = t.elapsed().as_secs_f64();
let bytes = std::fs::metadata(&path)?.len();
let t = Instant::now();
let read = graphar::io::read_chunk(&path, &file_type)?;
let read_s = t.elapsed().as_secs_f64();
let rows: usize = read.iter().map(|b| b.num_rows()).sum();
anyhow::ensure!(rows == ROWS, "row count mismatch: {rows}");
Ok(BenchResult::measured(
format!("format_{}", file_type.extension()),
metrics(ROWS, bytes, write_s, read_s),
))
}
macro_rules! format_bencher {
($name:ident, $id:literal, $ft:expr) => {
struct $name;
impl Bencher for $name {
fn id(&self) -> &'static str {
$id
}
fn run(&self) -> Result<BenchResult> {
format_round_trip($ft)
}
}
nornir_bench::register_bench!($name);
};
}
format_bencher!(
FormatParquet,
"knut.format_parquet",
graphar::FileType::Parquet
);
format_bencher!(FormatCsv, "knut.format_csv", graphar::FileType::Csv);
format_bencher!(FormatJson, "knut.format_json", graphar::FileType::Json);
format_bencher!(FormatOrc, "knut.format_orc", graphar::FileType::Orc);
format_bencher!(
FormatIpc,
"knut.format_arrow_ipc",
graphar::FileType::ArrowIpc
);
/// VerticesBuilder → chunked parquet on disk (the GraphAr write path).
struct BuilderDump;
impl Bencher for BuilderDump {
fn id(&self) -> &'static str {
"knut.builder_dump"
}
fn run(&self) -> Result<BenchResult> {
use graphar::{
DataType, FileType, Property, PropertyGroup, PropertyValue, Vertex, VertexInfo,
VerticesBuilder,
};
let dir = tempfile::tempdir()?;
let vi = VertexInfo::new("Person", 8192).with_property_group(PropertyGroup::new(
vec![
Property::new("id", DataType::Int64).primary(),
Property::new("name", DataType::String),
],
FileType::Parquet,
));
let t = Instant::now();
let mut builder = VerticesBuilder::new(vi, dir.path());
for i in 0..ROWS as i64 {
builder.add_vertex(
Vertex::new()
.add_property("id", PropertyValue::Int64(i))
.add_property("name", PropertyValue::String(format!("person_{i}"))),
);
}
let chunks = builder.dump()?;
let total_s = t.elapsed().as_secs_f64();
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(ROWS));
m.insert("chunks".into(), json!(chunks));
m.insert("total_s".into(), json!(total_s));
m.insert("rows_per_s".into(), json!(ROWS as f64 / total_s));
Ok(BenchResult::measured("builder_dump", m))
}
}
nornir_bench::register_bench!(BuilderDump);
/// VerticesBuilder → dump → **VertexReader::read_merged** — the GraphAr graph
/// LOAD path (read every vertex chunk back off disk and merge the property
/// groups into one row-aligned batch), the read-side counterpart of
/// `builder_dump`. Writes a real chunked GraphAr dataset to a tempdir, then
/// times reading the whole thing back. Reports read rows/sec and the on-disk
/// bytes/sec the reader sustains.
struct GraphArVertexLoad;
impl Bencher for GraphArVertexLoad {
fn id(&self) -> &'static str {
"knut.graphar_vertex_load"
}
fn run(&self) -> Result<BenchResult> {
use graphar::{
DataType, FileType, Property, PropertyGroup, PropertyValue, Vertex, VertexInfo,
VertexReader, VerticesBuilder,
};
let dir = tempfile::tempdir()?;
let make_vi = || {
VertexInfo::new("Person", 8192).with_property_group(PropertyGroup::new(
vec![
Property::new("id", DataType::Int64).primary(),
Property::new("name", DataType::String),
],
FileType::Parquet,
))
};
// Write phase: materialise a real chunked GraphAr vertex dataset.
let mut builder = VerticesBuilder::new(make_vi(), dir.path());
for i in 0..ROWS as i64 {
builder.add_vertex(
Vertex::new()
.add_property("id", PropertyValue::Int64(i))
.add_property("name", PropertyValue::String(format!("person_{i}"))),
);
}
let chunks = builder.dump()?;
let on_disk_bytes = dir_size_bytes(dir.path());
// Read phase (measured): load every chunk back and merge groups.
let reader = VertexReader::new(make_vi(), dir.path(), ROWS as u64);
let t = Instant::now();
let merged = reader.read_merged()?;
let read_s = t.elapsed().as_secs_f64();
anyhow::ensure!(
merged.num_rows() == ROWS,
"row count mismatch on load: {} != {ROWS}",
merged.num_rows()
);
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(ROWS));
m.insert("chunks".into(), json!(chunks));
m.insert("on_disk_bytes".into(), json!(on_disk_bytes));
m.insert("read_s".into(), json!(read_s));
m.insert("rows_per_s".into(), json!(ROWS as f64 / read_s));
m.insert(
"read_mb_s".into(),
json!((on_disk_bytes as f64 / (1024.0 * 1024.0)) / read_s),
);
Ok(BenchResult::measured("graphar_vertex_load", m))
}
}
nornir_bench::register_bench!(GraphArVertexLoad);
/// Total size in bytes of all regular files under `root` (a GraphAr dataset dir).
fn dir_size_bytes(root: &std::path::Path) -> u64 {
let mut acc = 0u64;
let mut stack = vec![root.to_path_buf()];
while let Some(p) = stack.pop() {
if let Ok(rd) = std::fs::read_dir(&p) {
for e in rd.flatten() {
let path = e.path();
if path.is_dir() {
stack.push(path);
} else if let Ok(meta) = e.metadata() {
acc += meta.len();
}
}
}
}
acc
}
/// merge_property_groups on two 100k-row groups.
struct MergeGroups;
impl Bencher for MergeGroups {
fn id(&self) -> &'static str {
"knut.merge_property_groups"
}
fn run(&self) -> Result<BenchResult> {
let a = sample_batch(ROWS);
let b = {
let cities: Vec<String> = (0..ROWS).map(|i| format!("city_{}", i % 100)).collect();
RecordBatch::try_new(
Arc::new(Schema::new(vec![Field::new(
"city",
ArrowDataType::Utf8,
false,
)])),
vec![Arc::new(StringArray::from(cities))],
)?
};
let t = Instant::now();
let merged = graphar::merge_property_groups(&[vec![a], vec![b]])?;
let merge_s = t.elapsed().as_secs_f64();
anyhow::ensure!(merged.num_columns() == 4);
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(ROWS));
m.insert("merge_s".into(), json!(merge_s));
m.insert("rows_per_s".into(), json!(ROWS as f64 / merge_s));
Ok(BenchResult::measured("merge_property_groups", m))
}
}
nornir_bench::register_bench!(MergeGroups);
/// Bifröst fabric: 100k table rows → vertex + edge mutation (no sink).
struct BifrostFabric;
impl Bencher for BifrostFabric {
fn id(&self) -> &'static str {
"knut.bifrost_fabric"
}
fn run(&self) -> Result<BenchResult> {
use knut_bifrost::{
TableContext, Transform, VertexTransform,
transform::{EdgeTransform, EndpointSpec},
};
let company_ids: Vec<i64> = (0..ROWS as i64).map(|i| i % 1000).collect();
let batch = {
let base = sample_batch(ROWS);
let mut cols = base.columns().to_vec();
let mut fields: Vec<Field> = base
.schema()
.fields()
.iter()
.map(|f| f.as_ref().clone())
.collect();
fields.push(Field::new("company_id", ArrowDataType::Int64, false));
cols.push(Arc::new(Int64Array::from(company_ids)));
RecordBatch::try_new(Arc::new(Schema::new(fields)), cols)?
};
let vertex = VertexTransform {
label: "Person".into(),
id_column: "id".into(),
properties: vec!["name".into(), "score".into()],
rename: Default::default(),
};
let edge = EdgeTransform {
edge_type: "WORKS_AT".into(),
src: EndpointSpec {
label: "Person".into(),
column: "id".into(),
id_column: None,
},
dst: EndpointSpec {
label: "Company".into(),
column: "company_id".into(),
id_column: None,
},
properties: vec![],
rename: Default::default(),
key: Vec::new(),
replace: None,
};
let ctx = TableContext {
table: "bench".into(),
};
let t = Instant::now();
let mut mutation = vertex.apply(&ctx, std::slice::from_ref(&batch))?;
mutation.merge(edge.apply(&ctx, std::slice::from_ref(&batch))?);
let fabric_s = t.elapsed().as_secs_f64();
anyhow::ensure!(mutation.vertices.len() == 1 && mutation.edges.len() == 1);
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(ROWS));
m.insert("fabric_s".into(), json!(fabric_s));
m.insert("rows_per_s".into(), json!(ROWS as f64 / fabric_s));
Ok(BenchResult::measured("bifrost_fabric", m))
}
}
nornir_bench::register_bench!(BifrostFabric);
/// FileSource scan: K parquet chunks in a watched dir → batches, the Airflow
/// AIP-82 file-listener read path. Synthetic chunks on /tmp, cold ledger.
struct FileSourceScan;
impl Bencher for FileSourceScan {
fn id(&self) -> &'static str {
"knut.file_source_scan"
}
fn run(&self) -> Result<BenchResult> {
use knut_bifrost::{
SyncMode, TableState,
source::{FileSource, TableSource, TableSpec},
};
const CHUNKS: usize = 4;
let per_chunk = ROWS / CHUNKS;
let dir = tempfile::tempdir()?; // /tmp — local disk, never T9
let table_dir = dir.path().join("lake.bench.nodes");
std::fs::create_dir_all(&table_dir)?;
let mut bytes = 0u64;
for c in 0..CHUNKS {
let path = table_dir.join(format!("chunk{c}.parquet"));
graphar::io::write_chunk(
&path,
std::slice::from_ref(&sample_batch(per_chunk)),
&graphar::FileType::Parquet,
)?;
bytes += std::fs::metadata(&path)?.len();
}
let source = FileSource::new(dir.path());
let spec = TableSpec::new("lake.bench.nodes", SyncMode::Incremental);
let rt = tokio::runtime::Builder::new_current_thread().build()?;
let t = Instant::now();
let got = rt
.block_on(source.fetch_new(&spec, &TableState::default()))?
.expect("chunks present → a round");
let scan_s = t.elapsed().as_secs_f64();
let rows: usize = got.0.iter().map(|b| b.num_rows()).sum();
anyhow::ensure!(rows == CHUNKS * per_chunk, "row count mismatch: {rows}");
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(rows));
m.insert("chunks".into(), json!(CHUNKS));
m.insert("file_bytes".into(), json!(bytes));
m.insert("scan_s".into(), json!(scan_s));
m.insert("rows_per_s".into(), json!(rows as f64 / scan_s));
Ok(BenchResult::measured("file_source_scan", m))
}
}
nornir_bench::register_bench!(FileSourceScan);
/// Full ingest lifecycle: K parquet drops → `Listener.sync_once` (FileSource →
/// transform fabric → counting null sink) → graph mutation, end to end. This
/// is the create→detect→ingest path the lifecycle tests drive, measured for
/// throughput (rows/s) so the sink/transform plumbing can be tuned. No DB:
/// the null sink isolates everything *up to* the FalkorDB round trip.
struct IngestLifecycle;
impl Bencher for IngestLifecycle {
fn id(&self) -> &'static str {
"knut.ingest_lifecycle"
}
fn run(&self) -> Result<BenchResult> {
use async_trait::async_trait;
use knut_bifrost::{
GraphMutation, Listener, MappingSpec, SinkReport, SyncState, TransformRegistry,
sink::GraphSink, source::FileSource,
};
#[derive(Default)]
struct NullSink;
#[async_trait]
impl GraphSink for NullSink {
async fn apply(&mut self, m: &GraphMutation) -> knut_bifrost::Result<SinkReport> {
Ok(SinkReport {
vertices_written: m.vertices.iter().map(|s| s.batch.num_rows()).sum(),
edges_written: m.edges.iter().map(|s| s.batch.num_rows()).sum(),
..Default::default()
})
}
}
const CHUNKS: usize = 4;
let per_chunk = ROWS / CHUNKS;
let dir = tempfile::tempdir()?; // /tmp — local disk, never T9
let table_dir = dir.path().join("nodes");
std::fs::create_dir_all(&table_dir)?;
for c in 0..CHUNKS {
graphar::io::write_chunk(
table_dir.join(format!("chunk{c}.parquet")),
std::slice::from_ref(&sample_batch(per_chunk)),
&graphar::FileType::Parquet,
)?;
}
let mapping = MappingSpec::from_yaml(
"version: 1\ntables:\n - table: nodes\n mode: incremental\n pipeline:\n - vertex: { label: Node, id_column: id, properties: [name, score] }\n",
)?;
let mut listener = Listener::new(
mapping,
TransformRegistry::new(),
FileSource::new(dir.path()),
NullSink,
SyncState::open(dir.path().join(".state.json"))?,
);
let rt = tokio::runtime::Builder::new_current_thread().build()?;
let t = Instant::now();
let round = rt.block_on(listener.sync_once())?;
let ingest_s = t.elapsed().as_secs_f64();
let rows = round.synced.first().map(|s| s.1).unwrap_or(0);
anyhow::ensure!(rows == ROWS, "row count mismatch: {rows}");
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(rows));
m.insert("chunks".into(), json!(CHUNKS));
m.insert("ingest_s".into(), json!(ingest_s));
m.insert("rows_per_s".into(), json!(rows as f64 / ingest_s));
Ok(BenchResult::measured("ingest_lifecycle", m))
}
}
nornir_bench::register_bench!(IngestLifecycle);
/// Arrow → Cypher rendering: the FalkorDB-ingest hot path. Renders 100k rows ×
/// 4 columns to Cypher map literals via the column-resolved writer (downcast
/// once per column, direct buffer writes — no per-cell String/Vec/join).
struct CypherRender;
impl Bencher for CypherRender {
fn id(&self) -> &'static str {
"knut.cypher_render"
}
fn run(&self) -> Result<BenchResult> {
use graphar_falkordb::cypher::{ColRenderer, write_row_map};
// id: Int64, name: Utf8, score: Float64, active: Boolean.
let ids: Vec<i64> = (0..ROWS as i64).collect();
let names: Vec<String> = (0..ROWS).map(|i| format!("node_{i}")).collect();
let scores: Vec<f64> = (0..ROWS).map(|i| i as f64 * 0.5).collect();
let active: Vec<bool> = (0..ROWS).map(|i| i % 2 == 0).collect();
let batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("id", ArrowDataType::Int64, false),
Field::new("name", ArrowDataType::Utf8, false),
Field::new("score", ArrowDataType::Float64, false),
Field::new("active", ArrowDataType::Boolean, false),
])),
vec![
Arc::new(Int64Array::from(ids)),
Arc::new(StringArray::from(names)),
Arc::new(Float64Array::from(scores)),
Arc::new(arrow_array::BooleanArray::from(active)),
],
)?;
let schema = batch.schema();
let cols: Vec<(&str, ColRenderer)> = schema
.fields()
.iter()
.zip(batch.columns())
.map(|(f, c)| (f.name().as_str(), ColRenderer::resolve(c.as_ref())))
.collect();
let mut buf = String::with_capacity(ROWS * 64);
let t = Instant::now();
for row in 0..ROWS {
let gar = row.to_string();
write_row_map(&mut buf, &cols, row, &[("_gar_id", &gar)])
.map_err(|e| anyhow::anyhow!("{e}"))?;
buf.push(',');
}
let render_s = t.elapsed().as_secs_f64();
anyhow::ensure!(buf.len() > ROWS, "rendered something");
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(ROWS));
m.insert("bytes".into(), json!(buf.len()));
m.insert("render_s".into(), json!(render_s));
m.insert("rows_per_s".into(), json!(ROWS as f64 / render_s));
Ok(BenchResult::measured("cypher_render", m))
}
}
nornir_bench::register_bench!(CypherRender);
/// `GRAPH.BULK` blob encode: Arrow batches → FalkorDB's binary bulk wire, the
/// other half of the FalkorDB-ingest boundary next to `cypher_render`. DB-free:
/// measures `build_bulk_payload` (header + typed cells, blobs presized from the
/// Arrow data, fanned over gatling) on 100k nodes + 100k edges.
struct FalkorBulkEncode;
impl Bencher for FalkorBulkEncode {
fn id(&self) -> &'static str {
"knut.falkor_bulk_encode"
}
fn run(&self) -> Result<BenchResult> {
use graphar_falkordb::{BulkEdgeGroup, BulkNodeGroup, build_bulk_payload};
let nodes = sample_batch(ROWS); // id Int64, name Utf8, score Float64
let src: Vec<i64> = (0..ROWS as i64).collect();
let dst: Vec<i64> = (0..ROWS as i64).map(|i| (i + 1) % ROWS as i64).collect();
let weight: Vec<f64> = (0..ROWS).map(|i| i as f64 * 0.25).collect();
let edges = RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("src", ArrowDataType::Int64, false),
Field::new("dst", ArrowDataType::Int64, false),
Field::new("weight", ArrowDataType::Float64, false),
])),
vec![
Arc::new(Int64Array::from(src)),
Arc::new(Int64Array::from(dst)),
Arc::new(Float64Array::from(weight)),
],
)?;
let ng = BulkNodeGroup {
label: "Person",
batches: std::slice::from_ref(&nodes),
};
let eg = BulkEdgeGroup {
src_label: "Person",
edge_label: "KNOWS",
dst_label: "Person",
batches: std::slice::from_ref(&edges),
};
let t = Instant::now();
let payload = build_bulk_payload(&[ng], &[eg]).map_err(|e| anyhow::anyhow!("{e}"))?;
let encode_s = t.elapsed().as_secs_f64();
let rows = (payload.node_count + payload.relation_count) as usize;
anyhow::ensure!(rows == 2 * ROWS, "entity count mismatch: {rows}");
let bytes: usize = payload.blobs.iter().map(Vec::len).sum();
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(rows));
m.insert("bytes".into(), json!(bytes));
m.insert("encode_s".into(), json!(encode_s));
m.insert("rows_per_s".into(), json!(rows as f64 / encode_s));
Ok(BenchResult::measured("falkor_bulk_encode", m))
}
}
nornir_bench::register_bench!(FalkorBulkEncode);
/// Results-back decode: a canned FalkorDB `GRAPH.QUERY` reply (`redis::Value`,
/// RESP tree) → Arrow `RecordBatch` via `response_to_batch_auto` — the read
/// half of the FalkorDB boundary. DB-free: the fixture is 100k rows × 4
/// columns (Int, BulkString, Double, Boolean) with a sprinkling of Nils,
/// exactly the shape the Flight SQL service decodes per query.
struct FalkorDecode;
impl Bencher for FalkorDecode {
fn id(&self) -> &'static str {
"knut.falkor_decode"
}
fn run(&self) -> Result<BenchResult> {
use redis::Value;
let header = Value::Array(
["id", "name", "score", "active"]
.iter()
.map(|n| Value::BulkString(n.as_bytes().to_vec()))
.collect(),
);
let rows: Vec<Value> = (0..ROWS)
.map(|i| {
Value::Array(vec![
Value::Int(i as i64),
if i % 97 == 0 {
Value::Nil
} else {
Value::BulkString(format!("node_{i}").into_bytes())
},
Value::Double(i as f64 * 0.5),
Value::Boolean(i % 2 == 0),
])
})
.collect();
let resp = Value::Array(vec![header, Value::Array(rows), Value::Array(vec![])]);
let t = Instant::now();
let batch = graphar_flight::convert::response_to_batch_auto(resp)
.map_err(|e| anyhow::anyhow!("{e}"))?;
let decode_s = t.elapsed().as_secs_f64();
anyhow::ensure!(batch.num_rows() == ROWS, "row count mismatch");
anyhow::ensure!(batch.num_columns() == 4, "column count mismatch");
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(ROWS));
m.insert("cols".into(), json!(4));
m.insert("decode_s".into(), json!(decode_s));
m.insert("rows_per_s".into(), json!(ROWS as f64 / decode_s));
Ok(BenchResult::measured("falkor_decode", m))
}
}
nornir_bench::register_bench!(FalkorDecode);
/// FalkorDB **parallel-dispatch pipelining** win — DB-free, the client-side half
/// of the ~1-core ingest fix (commit `de4e7fb`). The old serial loader awaited
/// one `UNWIND-CREATE` round-trip at a time, so between windows the client sat
/// idle for a full RTT — pinning ingest at ~1 core no matter the box. The
/// parallel path (`insert_*_par`) routes every window through
/// `graphar_falkordb::loader::buffered_dispatch`, which keeps up to
/// `available_parallelism()` windows in flight over the (`Clone`) multiplexed
/// connection. This arm models each window as a fixed simulated RTT and times
/// the SERIAL one-at-a-time loop against `buffered_dispatch` at full concurrency,
/// reporting `speedup_x` (→ ~concurrency). No FalkorDB container: it isolates the
/// scheduling win the fix delivers; the authoritative live nodes/s + `cores_busy`
/// on a real container is the `#[ignore]` integration path / Loki bencher's.
struct FalkorParallelDispatch {
windows: usize,
rtt_us: u64,
}
impl Bencher for FalkorParallelDispatch {
fn id(&self) -> &'static str {
"knut.falkor_parallel_dispatch"
}
fn run(&self) -> Result<BenchResult> {
use graphar_falkordb::loader::buffered_dispatch;
let concurrency = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(4);
let rtt = std::time::Duration::from_micros(self.rtt_us);
// A single-reactor runtime — exactly the MultiplexedConnection model
// (many in-flight commands, one link). enable_time for the RTT sleeps.
let rt = tokio::runtime::Builder::new_current_thread()
.enable_time()
.build()?;
rt.block_on(async {
// SERIAL baseline: await each simulated round-trip one at a time — the
// pre-fix loader's shape (the ~1-core stall).
let t = Instant::now();
let mut serial_rows = 0usize;
for w in 0..self.windows {
tokio::time::sleep(rtt).await;
serial_rows += w + 1;
}
let serial_s = t.elapsed().as_secs_f64();
// PARALLEL: the shipped path — up to `concurrency` windows in flight.
let items: Vec<usize> = (0..self.windows).map(|w| w + 1).collect();
let mut par_rows = 0usize;
let t = Instant::now();
buffered_dispatch(
items,
concurrency,
|rows| async move {
tokio::time::sleep(rtt).await;
Ok::<usize, graphar_falkordb::FalkorError>(rows)
},
|n| par_rows += n,
)
.await
.map_err(|e| anyhow::anyhow!("{e}"))?;
let parallel_s = t.elapsed().as_secs_f64();
anyhow::ensure!(
serial_rows == par_rows,
"serial and parallel must dispatch identical total rows ({serial_rows} != {par_rows})"
);
anyhow::ensure!(
parallel_s < serial_s,
"pipelining must beat the serial one-RTT-at-a-time loop ({parallel_s}s !< {serial_s}s)"
);
let host_cores = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1);
let mut m = serde_json::Map::new();
m.insert("windows".into(), json!(self.windows));
m.insert("rtt_us".into(), json!(self.rtt_us));
m.insert("concurrency".into(), json!(concurrency));
m.insert("serial_s".into(), json!(serial_s));
m.insert("parallel_s".into(), json!(parallel_s));
m.insert(
"serial_windows_per_s".into(),
json!(self.windows as f64 / serial_s.max(f64::MIN_POSITIVE)),
);
m.insert(
"parallel_windows_per_s".into(),
json!(self.windows as f64 / parallel_s.max(f64::MIN_POSITIVE)),
);
m.insert(
"speedup_x".into(),
json!(serial_s / parallel_s.max(f64::MIN_POSITIVE)),
);
m.insert("host_cores".into(), json!(host_cores));
m.insert("host_mem_bytes".into(), json!(host_mem_bytes()));
m.insert("elapsed_ms".into(), json!((serial_s + parallel_s) * 1000.0));
Ok(BenchResult::measured("falkor_parallel_dispatch", m))
})
}
}
nornir_bench::register_bench!(FalkorParallelDispatch {
windows: 512,
rtt_us: 500,
});
/// Graph → Iceberg export via **skade** (feature `skade`): the arrow bridge —
/// now an identity clone since knut and skade share arrow 58 (was a knut arrow
/// 56 → parquet → skade arrow 57 cross-version bridge) — plus a
/// `fast_append` Iceberg commit, for a 100k-row batch. Measures exactly the
/// write path the skade integration adds (`append_arrow_to_iceberg`), so a
/// future zero-copy bridge can be quantified against it. No container — skade's
/// catalog is an embedded redb warehouse in a tempdir.
#[cfg(feature = "skade")]
struct SkadeIcebergExport;
#[cfg(feature = "skade")]
impl Bencher for SkadeIcebergExport {
fn id(&self) -> &'static str {
"knut.skade_iceberg_export"
}
fn run(&self) -> Result<BenchResult> {
let batch = sample_batch(ROWS);
let dir = tempfile::tempdir()?; // /tmp — local disk, never T9
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?;
let t = Instant::now();
let rows = rt.block_on(graphar_flight::export::append_arrow_to_iceberg(
dir.path(),
"bench",
batch,
))?;
let export_s = t.elapsed().as_secs_f64();
anyhow::ensure!(rows == ROWS, "row count mismatch: {rows}");
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(rows));
m.insert("export_s".into(), json!(export_s));
m.insert("rows_per_s".into(), json!(rows as f64 / export_s));
m.insert("warehouse_bytes".into(), json!(dir_size(dir.path())));
Ok(BenchResult::measured("skade_iceberg_export", m))
}
}
#[cfg(feature = "skade")]
nornir_bench::register_bench!(SkadeIcebergExport);
/// The same export through the **parquet temp-file** bridge instead of the
/// zero-copy C Data Interface — the baseline `knut.skade_iceberg_export`
/// improves on. Running both shows the bridge delta.
#[cfg(feature = "skade")]
struct SkadeIcebergExportParquet;
#[cfg(feature = "skade")]
impl Bencher for SkadeIcebergExportParquet {
fn id(&self) -> &'static str {
"knut.skade_iceberg_export_parquet"
}
fn run(&self) -> Result<BenchResult> {
let batch = sample_batch(ROWS);
let dir = tempfile::tempdir()?;
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?;
let t = Instant::now();
let rows = rt.block_on(graphar_flight::export::append_arrow_to_iceberg_via_parquet(
dir.path(),
"bench",
batch,
))?;
let export_s = t.elapsed().as_secs_f64();
anyhow::ensure!(rows == ROWS, "row count mismatch: {rows}");
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(rows));
m.insert("export_s".into(), json!(export_s));
m.insert("rows_per_s".into(), json!(rows as f64 / export_s));
Ok(BenchResult::measured("skade_iceberg_export_parquet", m))
}
}
#[cfg(feature = "skade")]
nornir_bench::register_bench!(SkadeIcebergExportParquet);
/// **Iceberg read pushdown** (feature `skade`): the `SyncMode::Full` read that
/// used to be a full-width scan, measured three ways over the SAME warehouse.
///
/// knut's `IcebergSource` called `skade::Table::read()` for every Full round —
/// every column of every data file — while skade's `read_columns` /
/// `read_filtered` / `plan_stats` sat unused. This arm quantifies what that cost:
///
/// * **full** — the pre-fix baseline (`TableSpec` with no projection/filter);
/// * **projected** — the columns a `vertex: {id_column: id, properties: [name]}`
/// mapping actually reads, out of a 11-column table (2 of 11);
/// * **filtered** — the same projection plus `country == SE`, one data file of
/// four surviving the plan.
///
/// Reports `rows_per_s` per arm, `projection_speedup_x` / `filter_speedup_x`, and
/// the plan-time skipping (`files_baseline` vs `files_planned`) so the win is a
/// number rather than a claim. No container: an embedded skade warehouse in a
/// tempdir (/tmp — local disk, never T9).
#[cfg(feature = "skade")]
struct IcebergPushdownRead;
#[cfg(feature = "skade")]
impl Bencher for IcebergPushdownRead {
fn id(&self) -> &'static str {
"knut.iceberg_pushdown_read"
}
fn run(&self) -> Result<BenchResult> {
use knut_bifrost::mapping::{RowFilter, SyncMode};
use knut_bifrost::source::{IcebergSource, TableSource, TableSpec};
use knut_bifrost::state::TableState;
use skade::arrow_array::{ArrayRef, Int64Array, RecordBatch as B58, StringArray};
use skade::arrow_schema::{DataType, Field, Schema};
/// Rows per country file; four countries → 4 data files.
const PER: i64 = 250_000;
const COUNTRIES: [&str; 4] = ["SE", "NO", "DK", "FI"];
/// Payload columns the graph mapping never reads.
const PAYLOADS: usize = 8;
fn wide(country: &str, ids: std::ops::Range<i64>) -> B58 {
let ids: Vec<i64> = ids.collect();
let n = ids.len();
let names: Vec<String> = ids.iter().map(|i| format!("person_{i}")).collect();
let countries: Vec<String> = std::iter::repeat_n(country.to_string(), n).collect();
let mut fields = vec![
Field::new("id", DataType::Int64, false),
Field::new("name", DataType::Utf8, false),
Field::new("country", DataType::Utf8, false),
];
let mut cols: Vec<ArrayRef> = vec![
Arc::new(Int64Array::from(ids.clone())),
Arc::new(StringArray::from(names)),
Arc::new(StringArray::from(countries)),
];
for c in 0..PAYLOADS {
let p: Vec<String> = ids
.iter()
.map(|i| format!("payload_{c}_{i}_{}", "x".repeat(64)))
.collect();
fields.push(Field::new(format!("payload_{c}"), DataType::Utf8, false));
cols.push(Arc::new(StringArray::from(p)));
}
B58::try_new(Arc::new(Schema::new(fields)), cols).expect("wide batch")
}
let dir = tempfile::tempdir()?;
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()?;
rt.block_on(async {
// ── seed: one data file per country ──────────────────────────────
{
let wh = skade::Warehouse::open(dir.path()).await?;
let first = wide(COUNTRIES[0], 0..1);
let mut table = wh.table_or_create("people", &first.schema()).await?;
for (i, c) in COUNTRIES.iter().enumerate() {
let lo = i as i64 * PER;
table.append(&[wide(c, lo..lo + PER)]).await?;
}
} // drop → redb lock released for the source
let src = IcebergSource::new(dir.path());
let total_rows = PER * COUNTRIES.len() as i64;
// Warm the page cache identically for every arm: one untimed read.
let _ = src
.fetch_new(
&TableSpec::new("people", SyncMode::Full),
&TableState::default(),
)
.await
.map_err(|e| anyhow::anyhow!("{e}"))?;
// ── arm 1: FULL — the pre-fix baseline ───────────────────────────
let t = Instant::now();
let (full, _) = src
.fetch_new(
&TableSpec::new("people", SyncMode::Full),
&TableState::default(),
)
.await
.map_err(|e| anyhow::anyhow!("{e}"))?
.ok_or_else(|| anyhow::anyhow!("no news from the full read"))?;
let full_s = t.elapsed().as_secs_f64();
let full_rows: usize = full.iter().map(|b| b.num_rows()).sum();
let full_width = full.first().map(|b| b.num_columns()).unwrap_or(0);
anyhow::ensure!(
full_rows as i64 == total_rows,
"full read row count {full_rows}"
);
drop(full);
// ── arm 2: PROJECTED — the columns the mapping reads ─────────────
let proj_spec = TableSpec::new("people", SyncMode::Full)
.with_columns(vec!["id".into(), "name".into()]);
let t = Instant::now();
let (proj, _) = src
.fetch_new(&proj_spec, &TableState::default())
.await
.map_err(|e| anyhow::anyhow!("{e}"))?
.ok_or_else(|| anyhow::anyhow!("no news from the projected read"))?;
let proj_s = t.elapsed().as_secs_f64();
let proj_rows: usize = proj.iter().map(|b| b.num_rows()).sum();
let proj_width = proj.first().map(|b| b.num_columns()).unwrap_or(0);
anyhow::ensure!(
proj_rows == full_rows,
"the projection must drop columns, not rows: {proj_rows} vs {full_rows}"
);
anyhow::ensure!(proj_width == 2, "projected width {proj_width}");
drop(proj);
// ── arm 3: FILTERED — projection + `country == SE` ───────────────
let filt_spec = TableSpec::new("people", SyncMode::Full)
.with_columns(vec!["id".into(), "name".into()])
.with_filter(Some(RowFilter::eq("country", "SE")));
let t = Instant::now();
let (filt, _) = src
.fetch_new(&filt_spec, &TableState::default())
.await
.map_err(|e| anyhow::anyhow!("{e}"))?
.ok_or_else(|| anyhow::anyhow!("no news from the filtered read"))?;
let filt_s = t.elapsed().as_secs_f64();
let filt_rows: usize = filt.iter().map(|b| b.num_rows()).sum();
anyhow::ensure!(
filt_rows as i64 == PER,
"filtered read row count {filt_rows}"
);
drop(filt);
// ── plan-time skipping (metadata only, no data read) ─────────────
let baseline = src
.plan_stats("people", None)
.await
.map_err(|e| anyhow::anyhow!("{e}"))?;
let pruned = src
.plan_stats("people", Some(&RowFilter::eq("country", "SE")))
.await
.map_err(|e| anyhow::anyhow!("{e}"))?;
let host_cores = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1);
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(full_rows));
m.insert("columns_total".into(), json!(full_width));
m.insert("columns_projected".into(), json!(proj_width));
m.insert("full_s".into(), json!(full_s));
m.insert("projected_s".into(), json!(proj_s));
m.insert("filtered_s".into(), json!(filt_s));
m.insert(
"full_rows_per_s".into(),
json!(full_rows as f64 / full_s.max(f64::MIN_POSITIVE)),
);
m.insert(
"projected_rows_per_s".into(),
json!(proj_rows as f64 / proj_s.max(f64::MIN_POSITIVE)),
);
m.insert(
"filtered_rows_per_s".into(),
json!(filt_rows as f64 / filt_s.max(f64::MIN_POSITIVE)),
);
m.insert(
"projection_speedup_x".into(),
json!(full_s / proj_s.max(f64::MIN_POSITIVE)),
);
m.insert(
"filter_speedup_x".into(),
json!(full_s / filt_s.max(f64::MIN_POSITIVE)),
);
m.insert("files_baseline".into(), json!(baseline.data_files));
m.insert("files_planned".into(), json!(pruned.data_files));
m.insert(
"files_skipped".into(),
json!(baseline.data_files.saturating_sub(pruned.data_files)),
);
m.insert("rows_baseline_planned".into(), json!(baseline.rows_planned));
m.insert("rows_planned".into(), json!(pruned.rows_planned));
m.insert("warehouse_bytes".into(), json!(dir_size(dir.path())));
m.insert("peak_rss_bytes".into(), json!(peak_rss_bytes()));
m.insert("host_cores".into(), json!(host_cores));
m.insert("host_mem_bytes".into(), json!(host_mem_bytes()));
m.insert(
"elapsed_ms".into(),
json!((full_s + proj_s + filt_s) * 1000.0),
);
Ok(BenchResult::measured("iceberg_pushdown_read", m))
})
}
}
#[cfg(feature = "skade")]
nornir_bench::register_bench!(IcebergPushdownRead);
/// **CDC read pushdown** (feature `skade`): the change-feed read that used to be
/// full-width, measured four ways over the SAME warehouse and the SAME change
/// window.
///
/// `SyncMode::Full` gained projection/predicate pushdown in `00f6598`, but
/// `SyncMode::Incremental` / `SyncMode::Cdc` did **not** — they went through
/// `skade::Table::read_changelog`, which knut called unprojected, so every CDC
/// round paid for every column. This arm quantifies closing that:
///
/// * **full** — the pre-fix baseline: unprojected changelog read;
/// * **projected** — narrowed to the columns a `vertex: {id_column: id,
/// properties: [name]}` mapping actually reads (2 of 11);
/// * **projected + filtered** — the same projection plus `country == SE`;
/// * **renderable** — the projected read encoded all the way to the UI's
/// `(op, before, after, snapshot, ts)` Arrow rows, so the cost of the shape
/// a change pane actually binds to is a number and not an assumption.
///
/// HONEST ACCOUNTING, because "pushdown" means two different things: skade's
/// `ChangelogBatch::project_columns` is a **post-decode** slice (its
/// manifest-delta scan decodes whole parquet files), and the predicate is a
/// **post-scan** row prune. So this measures what knut can actually win from
/// here — encode, transform, sink and memory — NOT parquet I/O skipping. The
/// plan-time file skipping in `knut.iceberg_pushdown_read` is a different,
/// genuinely plan-level number and the two must not be conflated.
///
/// No container: an embedded skade warehouse in a tempdir.
#[cfg(feature = "skade")]
struct CdcPushdownRead;
#[cfg(feature = "skade")]
impl Bencher for CdcPushdownRead {
fn id(&self) -> &'static str {
"knut.cdc_pushdown_read"
}
fn run(&self) -> Result<BenchResult> {
use knut_bifrost::cdc::{CdcBackend, ChangeRequest, SkadeCdc};
use knut_bifrost::mapping::RowFilter;
use skade::arrow_array::{
ArrayRef, Int64Array as SkI64, RecordBatch as B58, StringArray as SkStr,
};
use skade::arrow_schema::{DataType, Field as SkField, Schema as SkSchema};
/// Rows per country file in the CHANGE WINDOW; four countries → 4 files.
const PER: i64 = 250_000;
const COUNTRIES: [&str; 4] = ["SE", "NO", "DK", "FI"];
/// Payload columns the graph mapping never reads.
const PAYLOADS: usize = 8;
fn wide(country: &str, ids: std::ops::Range<i64>) -> B58 {
let ids: Vec<i64> = ids.collect();
let n = ids.len();
let names: Vec<String> = ids.iter().map(|i| format!("person_{i}")).collect();
let countries: Vec<String> = std::iter::repeat_n(country.to_string(), n).collect();
let mut fields = vec![
SkField::new("id", DataType::Int64, false),
SkField::new("name", DataType::Utf8, false),
SkField::new("country", DataType::Utf8, false),
];
let mut cols: Vec<ArrayRef> = vec![
Arc::new(SkI64::from(ids.clone())),
Arc::new(SkStr::from(names)),
Arc::new(SkStr::from(countries)),
];
for c in 0..PAYLOADS {
let p: Vec<String> = ids
.iter()
.map(|i| format!("payload_{c}_{i}_{}", "x".repeat(64)))
.collect();
fields.push(SkField::new(format!("payload_{c}"), DataType::Utf8, false));
cols.push(Arc::new(SkStr::from(p)));
}
B58::try_new(Arc::new(SkSchema::new(fields)), cols).expect("wide batch")
}
let dir = tempfile::tempdir()?;
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()?;
rt.block_on(async {
// ── the cursor: a small base snapshot the window starts AFTER ────
let base = {
let wh = skade::Warehouse::open(dir.path()).await?;
let first = wide(COUNTRIES[0], -1..0);
let mut table = wh.table_or_create("people", &first.schema()).await?;
table.append(&[first]).await?;
table.current_snapshot_id().expect("base snapshot")
};
// ── the CHANGE WINDOW: 4 appends (1000000 rows) + a real
// equality-delete snapshot, so the feed carries deletes too ────
let latest = {
let wh = skade::Warehouse::open(dir.path()).await?;
let mut table = wh.table("people").await?;
for (i, c) in COUNTRIES.iter().enumerate() {
let lo = i as i64 * PER;
table.append(&[wide(c, lo..lo + PER)]).await?;
}
let key = B58::try_new(
Arc::new(SkSchema::new(vec![SkField::new(
"id",
DataType::Int64,
false,
)])),
vec![Arc::new(SkI64::from(vec![0i64, 1, 2]))],
)?;
table.delete_equality(&key, &["id"]).await?;
table.current_snapshot_id().expect("window head")
}; // drop → redb lock released for the reader
let cdc = SkadeCdc::new(dir.path());
let window = |cols: Vec<String>, filter: Option<RowFilter>| {
ChangeRequest::new("people", Some(base))
.to(Some(latest))
.with_columns(cols)
.with_filter(filter)
};
let count = |bs: &[arrow_array::RecordBatch]| -> usize {
bs.iter().map(|b| b.num_rows()).sum()
};
// Warm the page cache identically for every arm: one untimed read.
let _ = cdc
.changes(&window(vec![], None))
.await
.map_err(|e| anyhow::anyhow!("{e}"))?
.collect()
.await
.map_err(|e| anyhow::anyhow!("{e}"))?;
// ── arm 1: FULL-WIDTH — the pre-fix baseline ─────────────────────
let t = Instant::now();
let full = cdc
.changes(&window(vec![], None))
.await
.map_err(|e| anyhow::anyhow!("{e}"))?
.collect()
.await
.map_err(|e| anyhow::anyhow!("{e}"))?;
let full_s = t.elapsed().as_secs_f64();
let full_rows = count(&full);
let full_width = match full[0].schema().field(1).data_type() {
arrow_schema::DataType::Struct(f) => f.len(),
_ => 0,
};
anyhow::ensure!(
full_rows as i64 == PER * COUNTRIES.len() as i64 + 3,
"the window must be 1000000 inserts + 3 deletes, got {full_rows}",
);
drop(full);
// ── arm 2: PROJECTED — the columns the mapping reads ─────────────
let proj_req = window(vec!["id".into(), "name".into()], None);
let t = Instant::now();
let proj = cdc
.changes(&proj_req)
.await
.map_err(|e| anyhow::anyhow!("{e}"))?
.collect()
.await
.map_err(|e| anyhow::anyhow!("{e}"))?;
let proj_s = t.elapsed().as_secs_f64();
let proj_rows = count(&proj);
let proj_width = match proj[0].schema().field(1).data_type() {
arrow_schema::DataType::Struct(f) => f.len(),
_ => 0,
};
anyhow::ensure!(
proj_rows == full_rows,
"the projection must drop columns, not change rows: {proj_rows} vs {full_rows}",
);
drop(proj);
// ── arm 3: PROJECTED + FILTERED — `country == SE` ────────────────
let filt_req = window(
vec!["id".into(), "name".into()],
Some(RowFilter::eq("country", "SE")),
);
let t = Instant::now();
let filt = cdc
.changes(&filt_req)
.await
.map_err(|e| anyhow::anyhow!("{e}"))?
.collect()
.await
.map_err(|e| anyhow::anyhow!("{e}"))?;
let filt_s = t.elapsed().as_secs_f64();
let filt_rows = count(&filt);
// The 3 deletes have a null `country` and are KEPT — never lose a delete.
anyhow::ensure!(
filt_rows as i64 == PER + 3,
"filtered CDC read must be SE's 250000 + the 3 deletes, got {filt_rows}",
);
drop(filt);
// ── arm 4: RENDERABLE — projected, with update pairing on `id` ───
let render_req =
window(vec!["id".into(), "name".into()], None).with_identity(vec!["id".into()]);
let t = Instant::now();
let render = cdc
.changes(&render_req)
.await
.map_err(|e| anyhow::anyhow!("{e}"))?
.collect()
.await
.map_err(|e| anyhow::anyhow!("{e}"))?;
let render_s = t.elapsed().as_secs_f64();
let render_rows = count(&render);
drop(render);
let host_cores = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1);
let mut m = serde_json::Map::new();
m.insert("change_rows".into(), json!(full_rows));
m.insert("columns_total".into(), json!(full_width));
m.insert("columns_projected".into(), json!(proj_width));
m.insert("full_s".into(), json!(full_s));
m.insert("projected_s".into(), json!(proj_s));
m.insert("filtered_s".into(), json!(filt_s));
m.insert("renderable_s".into(), json!(render_s));
m.insert("filtered_change_rows".into(), json!(filt_rows));
m.insert("renderable_change_rows".into(), json!(render_rows));
m.insert(
"full_rows_per_s".into(),
json!(full_rows as f64 / full_s.max(f64::MIN_POSITIVE)),
);
m.insert(
"projected_rows_per_s".into(),
json!(proj_rows as f64 / proj_s.max(f64::MIN_POSITIVE)),
);
m.insert(
"filtered_rows_per_s".into(),
json!(filt_rows as f64 / filt_s.max(f64::MIN_POSITIVE)),
);
m.insert(
"renderable_rows_per_s".into(),
json!(render_rows as f64 / render_s.max(f64::MIN_POSITIVE)),
);
m.insert(
"projection_speedup_x".into(),
json!(full_s / proj_s.max(f64::MIN_POSITIVE)),
);
m.insert(
"filter_speedup_x".into(),
json!(full_s / filt_s.max(f64::MIN_POSITIVE)),
);
// Stated so nobody reads this as plan-time file skipping.
m.insert(
"pushdown_kind".into(),
json!("post_decode_projection_and_post_scan_predicate"),
);
m.insert("warehouse_bytes".into(), json!(dir_size(dir.path())));
m.insert("peak_rss_bytes".into(), json!(peak_rss_bytes()));
m.insert("host_cores".into(), json!(host_cores));
m.insert("host_mem_bytes".into(), json!(host_mem_bytes()));
m.insert(
"elapsed_ms".into(),
json!((full_s + proj_s + filt_s + render_s) * 1000.0),
);
Ok(BenchResult::measured("cdc_pushdown_read", m))
})
}
}
#[cfg(feature = "skade")]
nornir_bench::register_bench!(CdcPushdownRead);
/// Incremental delta ingest vs full re-ingest, Iceberg → FalkorDB, via skade.
///
/// This is the bench twin of the correctness test
/// `knut-bifrost/tests/delta_iceberg_falkordb.rs` — it writes a base Iceberg
/// snapshot then a second (delta) snapshot of new rows, and times two ingest
/// strategies into a live FalkorDB:
/// * INCREMENTAL: a `Listener` syncs the base, then `sync_once` again to ship
/// only the delta rows (cursor advances, no reload).
/// * FULL: a fresh graph + fresh cursor re-ingests *every* row (base + delta)
/// from scratch — the cost the delta path avoids.
/// Reports `incremental_rows_per_sec`, `full_rows_per_sec`, and `speedup_x`.
///
/// FIXTURE-GATE (not `--heavy`): FalkorDB is a container *fixture*, not scale.
/// The bench connects to `KNUT_FALKORDB_URL` (default `redis://127.0.0.1:6379`)
/// and SKIPS GRACEFULLY (`Err("skipped: no FalkorDB ...")`) if nothing answers —
/// exactly how the planet benches skip; the runner records the skip, not fatal.
#[cfg(feature = "skade")]
struct DeltaIngest;
#[cfg(feature = "skade")]
impl Bencher for DeltaIngest {
fn id(&self) -> &'static str {
"knut.delta_ingest"
}
fn run(&self) -> Result<BenchResult> {
use knut_bifrost::{
Listener, MappingSpec, SyncState, TransformRegistry, sink::FalkorDbSink,
source::IcebergSource,
};
use skade::arrow_array::{Int64Array, RecordBatch as B57, StringArray};
use skade::arrow_schema::{DataType, Field, Schema};
// Scale knobs: BASE rows ingested first, then DELTA new rows. Larger than
// the (downscaled) correctness test so rows/s is meaningful, but still a
// fixture bench — not `--heavy` scale.
const BASE: i64 = 20_000;
const DELTA: i64 = 5_000;
const MAPPING: &str = r#"
version: 1
tables:
- table: people
mode: incremental
pipeline:
- vertex:
label: Person
id_column: id
properties: [name]
"#;
fn people57(ids: std::ops::Range<i64>) -> B57 {
let ids: Vec<i64> = ids.collect();
let names: Vec<String> = ids.iter().map(|i| format!("person_{i}")).collect();
B57::try_new(
Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("name", DataType::Utf8, false),
])),
vec![
Arc::new(Int64Array::from(ids)),
Arc::new(StringArray::from(names)),
],
)
.unwrap()
}
async fn append(dir: &std::path::Path, ids: std::ops::Range<i64>) -> Result<()> {
let wh = skade::Warehouse::open(dir).await?;
let batch = people57(ids);
let mut table = wh.table_or_create("people", &batch.schema()).await?;
table.append(&[batch]).await?;
Ok(()) // `wh`/`table` drop here → redb lock released for IcebergSource.
}
let url =
std::env::var("KNUT_FALKORDB_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".into());
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?;
rt.block_on(async {
// ── FIXTURE-GATE: can we reach FalkorDB? If not, skip gracefully. ──
// Connecting opens a multiplexed redis connection; a dead/absent
// FalkorDB errors here and we turn it into a recorded skip, never a
// fatal bench failure.
FalkorDbSink::connect(&url, "delta_ingest_probe")
.await
.map_err(|e| anyhow::anyhow!("skipped: no FalkorDB at {url}: {e}"))?;
// ── shared Iceberg warehouse: base snapshot + delta snapshot ──────
let warehouse = tempfile::tempdir()?;
append(warehouse.path(), 0..BASE).await?;
// ── INCREMENTAL: ingest base, then sync_once ships only the delta ─
let inc_state = tempfile::tempdir()?;
let mut inc = Listener::new(
MappingSpec::from_yaml(MAPPING)?,
TransformRegistry::new(),
IcebergSource::new(warehouse.path()),
FalkorDbSink::connect(&url, "delta_ingest_inc")
.await
.map_err(|e| anyhow::anyhow!("{e}"))?,
SyncState::open(inc_state.path().join("state.json"))?,
);
// Prime the cursor with the base (not timed — the delta is the win).
let base_round = inc.sync_once().await.map_err(|e| anyhow::anyhow!("{e}"))?;
anyhow::ensure!(
base_round.synced.first().map(|s| s.1) == Some(BASE as usize),
"base ingest shipped {:?} rows, expected {BASE}",
base_round.synced.first().map(|s| s.1),
);
// Now append the delta snapshot and TIME the incremental sync.
append(warehouse.path(), BASE..BASE + DELTA).await?;
let t = Instant::now();
let delta_round = inc.sync_once().await.map_err(|e| anyhow::anyhow!("{e}"))?;
let inc_s = t.elapsed().as_secs_f64();
let inc_rows = delta_round.synced.first().map(|s| s.1).unwrap_or(0);
anyhow::ensure!(
inc_rows == DELTA as usize,
"incremental shipped {inc_rows} rows, expected only the delta {DELTA}",
);
// ── FULL: fresh graph + fresh cursor re-ingests base + delta ──────
let full_state = tempfile::tempdir()?;
let mut full = Listener::new(
MappingSpec::from_yaml(MAPPING)?,
TransformRegistry::new(),
IcebergSource::new(warehouse.path()),
FalkorDbSink::connect(&url, "delta_ingest_full")
.await
.map_err(|e| anyhow::anyhow!("{e}"))?,
SyncState::open(full_state.path().join("state.json"))?,
);
let t = Instant::now();
let full_round = full.sync_once().await.map_err(|e| anyhow::anyhow!("{e}"))?;
let full_s = t.elapsed().as_secs_f64();
let full_rows = full_round.synced.first().map(|s| s.1).unwrap_or(0);
anyhow::ensure!(
full_rows == (BASE + DELTA) as usize,
"full re-ingest shipped {full_rows} rows, expected base+delta {}",
BASE + DELTA,
);
let inc_rps = inc_rows as f64 / inc_s.max(f64::MIN_POSITIVE);
let full_rps = full_rows as f64 / full_s.max(f64::MIN_POSITIVE);
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(full_rows));
m.insert("incremental_rows".into(), json!(inc_rows));
m.insert("full_rows".into(), json!(full_rows));
m.insert("incremental_s".into(), json!(inc_s));
m.insert("full_s".into(), json!(full_s));
m.insert("elapsed_ms".into(), json!((inc_s + full_s) * 1000.0));
m.insert("incremental_rows_per_sec".into(), json!(inc_rps));
m.insert("full_rows_per_sec".into(), json!(full_rps));
m.insert(
"speedup_x".into(),
json!(if inc_s > 0.0 { full_s / inc_s } else { 0.0 }),
);
Ok(BenchResult::measured("delta_ingest", m))
})
}
}
#[cfg(feature = "skade")]
nornir_bench::register_bench!(DeltaIngest);
/// **Flight SQL DoPut dispatch**: the serial one-round-trip-at-a-time insert the
/// DoPut handler used to call, the parallel `insert_*_par` it now calls, and the
/// IN-PROCESS reference — all three against a live FalkorDB, same rows, same
/// renderer.
///
/// This arm exists to keep the claim honest. The DoPut handler really was on the
/// serial path while the in-process loader was on the concurrent one, and the fix
/// routes it across; what the routing BUYS depends entirely on where FalkorDB is:
///
/// * on **loopback** the round-trip the pipeline hides is tens of microseconds
/// and FalkorDB serialises writes per graph, so the measured win is small —
/// this arm reports it rather than a marketing number;
/// * over a **network** hop the hidden round-trip dominates, which is what the
/// DB-free `knut.falkor_parallel_dispatch` arm models (≈concurrency×).
///
/// `inprocess_rows_per_s` is the reference the DoPut path is compared against, so
/// `doput_vs_inprocess_x` says how much the Flight transport itself costs.
///
/// FIXTURE-GATE (not `--heavy`): connects to `KNUT_FALKORDB_URL` (default
/// `redis://127.0.0.1:6379`) and SKIPS GRACEFULLY if nothing answers.
struct DoPutDispatch {
rows: i64,
}
impl Bencher for DoPutDispatch {
fn id(&self) -> &'static str {
"knut.doput_dispatch"
}
fn run(&self) -> Result<BenchResult> {
use arrow_array::{Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema};
use graphar_falkordb::FalkorDbLoader;
let rows = self.rows;
let ids: Vec<i64> = (0..rows).collect();
let names: Vec<String> = ids.iter().map(|i| format!("person_{i}")).collect();
let cities: Vec<String> = ids
.iter()
.map(|i| ["Stockholm", "Göteborg", "Malmö", "Uppsala"][(i % 4) as usize].to_string())
.collect();
let batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("name", DataType::Utf8, false),
Field::new("city", DataType::Utf8, false),
])),
vec![
Arc::new(Int64Array::from(ids)),
Arc::new(StringArray::from(names)),
Arc::new(StringArray::from(cities)),
],
)?;
let url =
std::env::var("KNUT_FALKORDB_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".into());
let concurrency = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1);
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()?;
rt.block_on(async {
// FIXTURE-GATE.
FalkorDbLoader::connect(&url, "doput_probe")
.await
.map_err(|e| anyhow::anyhow!("skipped: no FalkorDB at {url}: {e}"))?;
async fn wipe(url: &str, graph: &str) {
if let Ok(c) = redis::Client::open(url)
&& let Ok(mut conn) = c.get_multiplexed_async_connection().await
{
let _: std::result::Result<redis::Value, _> = redis::cmd("GRAPH.DELETE")
.arg(graph)
.query_async(&mut conn)
.await;
}
}
// Each arm gets a fresh graph so none pays another's page-cache or
// index state.
let mut timings = Vec::new();
for (graph, conc) in [
("bench_doput_serial", 1usize),
("bench_doput_parallel", concurrency),
("bench_doput_inproc", concurrency),
] {
wipe(&url, graph).await;
let loader = FalkorDbLoader::connect(&url, graph)
.await
.map_err(|e| anyhow::anyhow!("{e}"))?;
let t = Instant::now();
let n = loader
.insert_vertices_par("Person", std::slice::from_ref(&batch), conc)
.await
.map_err(|e| anyhow::anyhow!("{e}"))?;
let s = t.elapsed().as_secs_f64();
anyhow::ensure!(n as i64 == rows, "{graph} inserted {n} of {rows}");
timings.push(s);
wipe(&url, graph).await;
}
let (serial_s, parallel_s, inproc_s) = (timings[0], timings[1], timings[2]);
let host_cores = concurrency;
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(rows));
m.insert("concurrency".into(), json!(concurrency));
m.insert("serial_s".into(), json!(serial_s));
m.insert("parallel_s".into(), json!(parallel_s));
m.insert("inprocess_s".into(), json!(inproc_s));
m.insert(
"serial_rows_per_s".into(),
json!(rows as f64 / serial_s.max(f64::MIN_POSITIVE)),
);
m.insert(
"parallel_rows_per_s".into(),
json!(rows as f64 / parallel_s.max(f64::MIN_POSITIVE)),
);
m.insert(
"inprocess_rows_per_s".into(),
json!(rows as f64 / inproc_s.max(f64::MIN_POSITIVE)),
);
m.insert(
"speedup_x".into(),
json!(serial_s / parallel_s.max(f64::MIN_POSITIVE)),
);
m.insert(
"doput_vs_inprocess_x".into(),
json!(parallel_s / inproc_s.max(f64::MIN_POSITIVE)),
);
m.insert("host_cores".into(), json!(host_cores));
m.insert("host_mem_bytes".into(), json!(host_mem_bytes()));
m.insert(
"elapsed_ms".into(),
json!((serial_s + parallel_s + inproc_s) * 1000.0),
);
Ok(BenchResult::measured("doput_dispatch", m))
})
}
}
nornir_bench::register_bench!(DoPutDispatch { rows: 40_000 });
/// Recursive on-disk size of a directory (the Iceberg warehouse), for the
/// skade bencher's `warehouse_bytes` metric.
#[cfg(feature = "skade")]
fn dir_size(path: &std::path::Path) -> u64 {
let mut total = 0;
if let Ok(rd) = std::fs::read_dir(path) {
for entry in rd.flatten() {
let p = entry.path();
if p.is_dir() {
total += dir_size(&p);
} else if let Ok(md) = entry.metadata() {
total += md.len();
}
}
}
total
}
/// Process peak resident set size (bytes) from `/proc/self/status` `VmHWM`.
/// `0` when unavailable (non-Linux / unreadable) — a bench never fails on it.
fn peak_rss_bytes() -> u64 {
std::fs::read_to_string("/proc/self/status")
.ok()
.and_then(|s| {
s.lines()
.find_map(|l| l.strip_prefix("VmHWM:"))
.and_then(|v| v.split_whitespace().next())
.and_then(|kb| kb.parse::<u64>().ok())
.map(|kb| kb * 1024)
})
.unwrap_or(0)
}
/// Current resident set size (bytes) from `/proc/self/status` `VmRSS`.
fn current_rss_bytes() -> u64 {
std::fs::read_to_string("/proc/self/status")
.ok()
.and_then(|s| {
s.lines()
.find_map(|l| l.strip_prefix("VmRSS:"))
.and_then(|v| v.split_whitespace().next())
.and_then(|kb| kb.parse::<u64>().ok())
.map(|kb| kb * 1024)
})
.unwrap_or(0)
}
/// Host physical memory (bytes) from `/proc/meminfo` `MemTotal`. `0` if unknown.
fn host_mem_bytes() -> u64 {
std::fs::read_to_string("/proc/meminfo")
.ok()
.and_then(|s| {
s.lines()
.find_map(|l| l.strip_prefix("MemTotal:"))
.and_then(|v| v.split_whitespace().next())
.and_then(|kb| kb.parse::<u64>().ok())
.map(|kb| kb * 1024)
})
.unwrap_or(0)
}
/// Þund incremental streaming: **recompute vs incremental** over a stream with
/// a fixed fraction of late rows (design `thund-streaming-state.md`, phase 1
/// bench arm `knut.thund_watermark_drop`). Both arms consume the SAME seeded
/// micro-batch stream through the real DataFusion native backend:
/// * RECOMPUTE (default): the watermark is advisory, every row is kept.
/// * INCREMENTAL (`state_root` + `StreamingMode::Incremental`): the event-time
/// watermark drops rows below `W` before the query — the late rows.
/// Reports throughput (rows/s) for each arm plus `peak_rss_bytes` and the host
/// facts (`host_cores`, `host_mem_bytes`), so the recompute-vs-incremental cost
/// is comparable across machines.
struct ThundWatermarkDrop {
rows: usize,
triggers: usize,
late_pct: usize,
}
impl Bencher for ThundWatermarkDrop {
fn id(&self) -> &'static str {
"knut.thund_watermark_drop"
}
fn run(&self) -> Result<BenchResult> {
use datafusion::arrow::array::{Int64Array, RecordBatch as DfBatch};
use datafusion::arrow::datatypes::{DataType, Field, Schema as DfSchema};
use knut_thund::backend::ExecBackend;
use knut_thund::backend::native::{NativeBackend, StreamingMode};
use knut_thund::ir::{
Dataset, Flow, FlowKind, OutputMode, OutputType, Pipeline, SourceSpec, Trigger,
Watermark,
};
let per_trigger = self.rows / self.triggers;
let late_per_trigger = per_trigger * self.late_pct / 100;
let schema = Arc::new(DfSchema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("et", DataType::Int64, false),
]));
// Build `triggers` micro-batches. On-time rows advance event time by a
// fixed step per trigger; the late fraction carries et=0 (far behind the
// running watermark), so the incremental arm drops exactly those.
let mut micro_batches: Vec<Vec<DfBatch>> = Vec::with_capacity(self.triggers);
let mut late_total = 0u64;
let mut next_id: i64 = 0;
for t in 0..self.triggers {
let base = (t as i64 + 1) * 1_000;
let mut ids = Vec::with_capacity(per_trigger);
let mut ets = Vec::with_capacity(per_trigger);
for r in 0..per_trigger {
ids.push(next_id);
next_id += 1;
if r < late_per_trigger {
ets.push(0); // late: below the watermark once it advances
late_total += 1;
} else {
// On-time rows cluster at the trigger's event time (jitter
// well within the 100ms allowed lateness), so they stay above
// the watermark; only the et=0 late rows fall below it.
ets.push(base + (r as i64 % 50));
}
}
let b = DfBatch::try_new(
schema.clone(),
vec![
Arc::new(Int64Array::from(ids)),
Arc::new(Int64Array::from(ets)),
],
)?;
micro_batches.push(vec![b]);
}
// One pipeline, watermark on `et` (allowed lateness 100ms), append mode.
let pipeline = || {
let mut flow = Flow::streaming(
"f_events",
"events_out",
SourceSpec::Kafka {
bootstrap: "localhost:9092".into(),
topic: "events".into(),
format: "json".into(),
},
)
.with_query("SELECT id, et FROM events")
.with_watermark(Watermark {
event_time_column: "et".into(),
allowed_lateness_ms: 100,
idle_timeout_ms: 0,
});
if let FlowKind::Streaming {
output_mode,
trigger,
..
} = &mut flow.kind
{
*output_mode = OutputMode::Append;
*trigger = Trigger::Continuous;
}
Pipeline::new("thund_wm_bench")
.with_dataset(Dataset::new("events_out", OutputType::Table).incremental())
.with_flow(flow)
};
// ── RECOMPUTE arm (default mode): keeps every row. ──
let t = Instant::now();
let recompute = NativeBackend::new()
.with_stream_input("events", micro_batches.clone())
.run(&pipeline())
.map_err(|e| anyhow::anyhow!("recompute run: {e}"))?;
let recompute_s = t.elapsed().as_secs_f64();
let recompute_rows = recompute.row_count("events_out");
let recompute_rss = current_rss_bytes();
// ── INCREMENTAL arm: watermark drops the late rows. ──
let state = tempfile::tempdir()?;
let state_uri = format!("file://{}", state.path().display());
let t = Instant::now();
let incremental = NativeBackend::new()
.with_streaming_mode(StreamingMode::Incremental)
.with_state_root(state_uri)
.with_stream_input("events", micro_batches)
.run(&pipeline())
.map_err(|e| anyhow::anyhow!("incremental run: {e}"))?;
let incremental_s = t.elapsed().as_secs_f64();
let incremental_rows = incremental.row_count("events_out");
let incremental_rss = current_rss_bytes();
anyhow::ensure!(
recompute_rows == self.rows,
"recompute should keep all {} rows, kept {recompute_rows}",
self.rows
);
anyhow::ensure!(
incremental_rows as u64 == self.rows as u64 - late_total,
"incremental should drop {late_total} late rows ({} -> {}), kept {incremental_rows}",
self.rows,
self.rows as u64 - late_total
);
let host_cores = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1);
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(self.rows));
m.insert("triggers".into(), json!(self.triggers));
m.insert("late_pct".into(), json!(self.late_pct));
m.insert("late_rows_dropped".into(), json!(late_total));
m.insert("recompute_s".into(), json!(recompute_s));
m.insert("incremental_s".into(), json!(incremental_s));
m.insert(
"recompute_rows_per_s".into(),
json!(self.rows as f64 / recompute_s.max(f64::MIN_POSITIVE)),
);
m.insert(
"incremental_rows_per_s".into(),
json!(self.rows as f64 / incremental_s.max(f64::MIN_POSITIVE)),
);
m.insert("recompute_rss_bytes".into(), json!(recompute_rss));
m.insert("incremental_rss_bytes".into(), json!(incremental_rss));
m.insert("peak_rss_bytes".into(), json!(peak_rss_bytes()));
m.insert("host_cores".into(), json!(host_cores));
m.insert("host_mem_bytes".into(), json!(host_mem_bytes()));
m.insert(
"elapsed_ms".into(),
json!((recompute_s + incremental_s) * 1000.0),
);
Ok(BenchResult::measured("thund_watermark_drop", m))
}
}
nornir_bench::register_bench!(ThundWatermarkDrop {
rows: 50_000,
triggers: 10,
late_pct: 20,
});
/// Þund windowed streaming: **recompute vs incremental** over a long windowed
/// stream (design `thund-streaming-state.md`, phase 2 bench arm
/// `knut.thund_window_close`). Both arms consume the SAME seeded micro-batch
/// stream — one tumbling window's worth of rows per trigger — through the real
/// DataFusion native backend, running a per-window `COUNT(*)`:
/// * RECOMPUTE (default): re-runs the query over ALL rows accumulated so far
/// every trigger; the working set (and its peak RSS) grows with the stream.
/// * INCREMENTAL (`state_root` + `StreamingMode::Incremental`): closes each
/// window on the event-time watermark, emits its aggregate once, and EVICTS
/// the closed window's rows — so the open buffer (and peak RSS) stays flat.
/// The headline is `peak_rss_grows_bounded`: the incremental arm's peak-RSS
/// delta stays far below recompute's. Reports throughput (rows/s) per arm plus
/// `peak_rss_bytes` + host facts (`host_cores`, `host_mem_bytes`).
struct ThundWindowClose {
rows: usize,
triggers: usize,
window_ms: u64,
}
impl Bencher for ThundWindowClose {
fn id(&self) -> &'static str {
"knut.thund_window_close"
}
fn run(&self) -> Result<BenchResult> {
use datafusion::arrow::array::{Int64Array, RecordBatch as DfBatch};
use datafusion::arrow::datatypes::{DataType, Field, Schema as DfSchema};
use knut_thund::backend::ExecBackend;
use knut_thund::backend::native::{NativeBackend, StreamingMode};
use knut_thund::ir::{
Dataset, Flow, FlowKind, OutputMode, OutputType, Pipeline, SourceSpec, Trigger,
Watermark, WindowSpec,
};
let per_trigger = self.rows / self.triggers;
let win = self.window_ms as i64;
// Allowed lateness half a window; row jitter kept under it so no on-time
// row is dropped, and a window lags ~2 triggers behind close.
let lateness = win / 2;
let jitter = lateness; // et jitter span within a trigger, < window
let schema = Arc::new(DfSchema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("et", DataType::Int64, false),
]));
// Trigger `t` fills window [t*win, t*win+win): all its rows carry event
// times in [t*win, t*win + jitter). As the stream advances, older windows
// pass the watermark, close, and evict.
let mut micro_batches: Vec<Vec<DfBatch>> = Vec::with_capacity(self.triggers);
let mut next_id: i64 = 0;
for t in 0..self.triggers {
let base = t as i64 * win;
let mut ids = Vec::with_capacity(per_trigger);
let mut ets = Vec::with_capacity(per_trigger);
for r in 0..per_trigger {
ids.push(next_id);
next_id += 1;
ets.push(base + (r as i64 % jitter));
}
let b = DfBatch::try_new(
schema.clone(),
vec![
Arc::new(Int64Array::from(ids)),
Arc::new(Int64Array::from(ets)),
],
)?;
micro_batches.push(vec![b]);
}
// One windowed pipeline: tumbling window on `et`, per-window COUNT(*).
let pipeline = || {
let mut flow = Flow::streaming(
"f_events",
"win_out",
SourceSpec::Kafka {
bootstrap: "localhost:9092".into(),
topic: "events".into(),
format: "json".into(),
},
)
.with_query("SELECT COUNT(*) AS n FROM events")
.with_watermark(Watermark {
event_time_column: "et".into(),
allowed_lateness_ms: lateness as u64,
idle_timeout_ms: 0,
})
.with_window(WindowSpec::Tumbling {
size_ms: self.window_ms,
});
if let FlowKind::Streaming {
output_mode,
trigger,
..
} = &mut flow.kind
{
*output_mode = OutputMode::Append;
*trigger = Trigger::Continuous;
}
Pipeline::new("thund_window_bench")
.with_dataset(Dataset::new("win_out", OutputType::Table).incremental())
.with_flow(flow)
};
let base_peak = peak_rss_bytes();
// ── INCREMENTAL arm first: window close + evict keeps the buffer flat. ──
let state = tempfile::tempdir()?;
let state_uri = format!("file://{}", state.path().display());
let t = Instant::now();
let incremental = NativeBackend::new()
.with_streaming_mode(StreamingMode::Incremental)
.with_state_root(state_uri)
.with_stream_input("events", micro_batches.clone())
.run(&pipeline())
.map_err(|e| anyhow::anyhow!("incremental run: {e}"))?;
let incremental_s = t.elapsed().as_secs_f64();
let inc_peak = peak_rss_bytes();
let windows_closed = incremental.windows_closed("win_out");
let peak_open_rows = incremental.peak_open_rows("win_out");
// ── RECOMPUTE arm (default): accumulates every row, re-runs each trigger. ──
let t = Instant::now();
let recompute = NativeBackend::new()
.with_stream_input("events", micro_batches)
.run(&pipeline())
.map_err(|e| anyhow::anyhow!("recompute run: {e}"))?;
let recompute_s = t.elapsed().as_secs_f64();
let rec_peak = peak_rss_bytes();
// Correctness: incremental closes one window per trigger (each once);
// recompute collapses to a single running COUNT row.
anyhow::ensure!(
windows_closed == self.triggers,
"incremental should close {} windows once each, closed {windows_closed}",
self.triggers
);
anyhow::ensure!(
recompute.row_count("win_out") == 1,
"recompute COUNT(*) collapses to one row, got {}",
recompute.row_count("win_out")
);
// Working-set size is the deterministic cause of the memory difference,
// and the honest one: process VmHWM is global and already high by the
// time this arm runs in the suite, so per-arm RSS deltas are noise. The
// incremental arm's buffered open-window state is bounded (it evicts
// closed windows); the recompute arm accumulates EVERY row seen and
// re-runs over it each trigger, so its working set is the whole stream.
let incremental_working_set_rows = peak_open_rows;
let recompute_working_set_rows = self.rows; // recompute accumulates all
let bounded = incremental_working_set_rows < recompute_working_set_rows;
anyhow::ensure!(
bounded,
"incremental open state ({incremental_working_set_rows}) must stay below recompute's accumulated working set ({recompute_working_set_rows})"
);
// Informational peak-RSS deltas (VmHWM is monotonic + process-global).
let incremental_peak_delta = inc_peak.saturating_sub(base_peak);
let recompute_peak_delta = rec_peak.saturating_sub(inc_peak);
let host_cores = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1);
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(self.rows));
m.insert("triggers".into(), json!(self.triggers));
m.insert("window_ms".into(), json!(self.window_ms));
m.insert("windows_closed".into(), json!(windows_closed));
m.insert("recompute_s".into(), json!(recompute_s));
m.insert("incremental_s".into(), json!(incremental_s));
m.insert(
"recompute_rows_per_s".into(),
json!(self.rows as f64 / recompute_s.max(f64::MIN_POSITIVE)),
);
m.insert(
"incremental_rows_per_s".into(),
json!(self.rows as f64 / incremental_s.max(f64::MIN_POSITIVE)),
);
// Headline: the evicting incremental arm's working set stays bounded far
// below recompute's, which grows with the whole accumulated stream.
m.insert(
"incremental_working_set_rows".into(),
json!(incremental_working_set_rows),
);
m.insert(
"recompute_working_set_rows".into(),
json!(recompute_working_set_rows),
);
m.insert("working_set_bounded".into(), json!(bounded));
m.insert(
"incremental_peak_rss_delta_bytes".into(),
json!(incremental_peak_delta),
);
m.insert(
"recompute_peak_rss_delta_bytes".into(),
json!(recompute_peak_delta),
);
m.insert("peak_rss_bytes".into(), json!(peak_rss_bytes()));
m.insert("host_cores".into(), json!(host_cores));
m.insert("host_mem_bytes".into(), json!(host_mem_bytes()));
m.insert(
"elapsed_ms".into(),
json!((recompute_s + incremental_s) * 1000.0),
);
Ok(BenchResult::measured("thund_window_close", m))
}
}
nornir_bench::register_bench!(ThundWindowClose {
rows: 2_000_000,
triggers: 100,
window_ms: 1_000,
});
/// Build one windowed micro-batch stream: trigger `t` fills tumbling window
/// `[t*win, t*win+win)`; event times cluster inside a sub-window jitter so no
/// on-time row is late. Shared by the checkpoint + recovery bench arms.
fn thund_windowed_micro_batches(
rows: usize,
triggers: usize,
win: i64,
) -> Vec<Vec<datafusion::arrow::array::RecordBatch>> {
use datafusion::arrow::array::{Int64Array, RecordBatch as DfBatch};
use datafusion::arrow::datatypes::{DataType, Field, Schema as DfSchema};
let per = rows / triggers;
let jitter = (win / 2).max(1);
let schema = Arc::new(DfSchema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("et", DataType::Int64, false),
]));
let mut mb = Vec::with_capacity(triggers);
let mut next: i64 = 0;
for t in 0..triggers {
let base = t as i64 * win;
let mut ids = Vec::with_capacity(per);
let mut ets = Vec::with_capacity(per);
for r in 0..per {
ids.push(next);
next += 1;
ets.push(base + (r as i64 % jitter));
}
mb.push(vec![
DfBatch::try_new(
schema.clone(),
vec![
Arc::new(Int64Array::from(ids)),
Arc::new(Int64Array::from(ets)),
],
)
.unwrap(),
]);
}
mb
}
/// A tumbling windowed `COUNT(*)` pipeline over `events`, watermark on `et`.
fn thund_windowed_pipeline(window_ms: u64) -> knut_thund::ir::Pipeline {
use knut_thund::ir::{
Dataset, Flow, FlowKind, OutputMode, OutputType, Pipeline, SourceSpec, Trigger, Watermark,
WindowSpec,
};
let mut flow = Flow::streaming(
"f_events",
"win_out",
SourceSpec::Kafka {
bootstrap: "localhost:9092".into(),
topic: "events".into(),
format: "json".into(),
},
)
.with_query("SELECT COUNT(*) AS n FROM events")
.with_watermark(Watermark {
event_time_column: "et".into(),
allowed_lateness_ms: window_ms / 2,
idle_timeout_ms: 0,
})
.with_window(WindowSpec::Tumbling { size_ms: window_ms });
if let FlowKind::Streaming {
output_mode,
trigger,
..
} = &mut flow.kind
{
*output_mode = OutputMode::Append;
*trigger = Trigger::Continuous;
}
Pipeline::new("thund_cp_bench")
.with_dataset(Dataset::new("win_out", OutputType::Table).incremental())
.with_flow(flow)
}
/// Þund barrier/epoch **checkpointing cost** (design phase 3 bench arm
/// `knut.thund_checkpoint`): the SAME windowed stream driven WITH durable
/// checkpoints (incremental + `state_root` — one epoch snapshotted to Parquet
/// per trigger) vs WITHOUT (recompute baseline, no state). Reports throughput
/// (rows/s) for each arm, the checkpoints committed, and the per-row checkpoint
/// overhead, plus `peak_rss_bytes` + host facts.
struct ThundCheckpointCommit {
rows: usize,
triggers: usize,
window_ms: u64,
}
impl Bencher for ThundCheckpointCommit {
fn id(&self) -> &'static str {
"knut.thund_checkpoint"
}
fn run(&self) -> Result<BenchResult> {
use knut_thund::backend::ExecBackend;
use knut_thund::backend::native::{NativeBackend, StreamingMode};
let win = self.window_ms as i64;
let mb = thund_windowed_micro_batches(self.rows, self.triggers, win);
// ── CHECKPOINTED arm: incremental, one committed epoch per trigger. ──
let state = tempfile::tempdir()?;
let state_uri = format!("file://{}", state.path().display());
let t = Instant::now();
let checkpointed = NativeBackend::new()
.with_streaming_mode(StreamingMode::Incremental)
.with_state_root(state_uri)
.with_stream_input("events", mb.clone())
.run(&thund_windowed_pipeline(self.window_ms))
.map_err(|e| anyhow::anyhow!("checkpointed run: {e}"))?;
let checkpointed_s = t.elapsed().as_secs_f64();
let committed = checkpointed.checkpoints_committed("win_out");
let cp_peak = peak_rss_bytes();
// ── BASELINE arm: recompute (default), no checkpoint store. ──
let t = Instant::now();
let baseline = NativeBackend::new()
.with_stream_input("events", mb)
.run(&thund_windowed_pipeline(self.window_ms))
.map_err(|e| anyhow::anyhow!("baseline run: {e}"))?;
let baseline_s = t.elapsed().as_secs_f64();
let _ = baseline.row_count("win_out");
anyhow::ensure!(
committed == self.triggers,
"expected {} committed epochs, got {committed}",
self.triggers
);
let host_cores = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1);
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(self.rows));
m.insert("triggers".into(), json!(self.triggers));
m.insert("window_ms".into(), json!(self.window_ms));
m.insert("checkpoints_committed".into(), json!(committed));
m.insert("checkpointed_s".into(), json!(checkpointed_s));
m.insert("baseline_recompute_s".into(), json!(baseline_s));
m.insert(
"checkpointed_rows_per_s".into(),
json!(self.rows as f64 / checkpointed_s.max(f64::MIN_POSITIVE)),
);
m.insert(
"baseline_rows_per_s".into(),
json!(self.rows as f64 / baseline_s.max(f64::MIN_POSITIVE)),
);
m.insert(
"checkpoint_overhead_us_per_epoch".into(),
json!((checkpointed_s - baseline_s).max(0.0) * 1e6 / committed.max(1) as f64),
);
m.insert(
"peak_rss_bytes".into(),
json!(cp_peak.max(peak_rss_bytes())),
);
m.insert("host_cores".into(), json!(host_cores));
m.insert("host_mem_bytes".into(), json!(host_mem_bytes()));
m.insert(
"elapsed_ms".into(),
json!((checkpointed_s + baseline_s) * 1000.0),
);
Ok(BenchResult::measured("thund_checkpoint", m))
}
}
nornir_bench::register_bench!(ThundCheckpointCommit {
rows: 1_000_000,
triggers: 100,
window_ms: 1_000,
});
/// Þund **crash recovery** (design phase 4 bench arm `knut.thund_recovery`):
/// process half the windowed stream then "crash" (`with_max_triggers`), and
/// RESUME from the latest committed checkpoint over the full stream — vs a cold
/// recompute-from-scratch over the whole stream. Reports the restore+resume
/// time, the epoch resumed from, the windows closed after recovery, and the
/// cold-recompute baseline, plus host facts. Recovery resumes from the committed
/// source offset, so no already-committed trigger is re-read (no double-work).
struct ThundRecovery {
rows: usize,
triggers: usize,
window_ms: u64,
}
impl Bencher for ThundRecovery {
fn id(&self) -> &'static str {
"knut.thund_recovery"
}
fn run(&self) -> Result<BenchResult> {
use knut_thund::backend::ExecBackend;
use knut_thund::backend::native::{NativeBackend, StreamingMode};
let win = self.window_ms as i64;
let mb = thund_windowed_micro_batches(self.rows, self.triggers, win);
let half = self.triggers / 2;
// ── COLD recompute-from-scratch baseline over the whole stream. ──
let t = Instant::now();
let cold = NativeBackend::new()
.with_stream_input("events", mb.clone())
.run(&thund_windowed_pipeline(self.window_ms))
.map_err(|e| anyhow::anyhow!("cold recompute: {e}"))?;
let cold_s = t.elapsed().as_secs_f64();
let _ = cold.row_count("win_out");
// ── CRASH: process the first half, checkpointing each trigger. ──
let state = tempfile::tempdir()?;
let state_uri = format!("file://{}", state.path().display());
let t = Instant::now();
let crashed = NativeBackend::new()
.with_streaming_mode(StreamingMode::Incremental)
.with_state_root(state_uri.clone())
.with_max_triggers(Some(half))
.with_stream_input("events", mb.clone())
.run(&thund_windowed_pipeline(self.window_ms))
.map_err(|e| anyhow::anyhow!("crashed run: {e}"))?;
let crash_s = t.elapsed().as_secs_f64();
let committed_before_crash = crashed.checkpoints_committed("win_out");
// ── RESUME from the checkpoint over the full stream. ──
let t = Instant::now();
let resumed = NativeBackend::new()
.with_streaming_mode(StreamingMode::Incremental)
.with_state_root(state_uri)
.with_stream_input("events", mb)
.run(&thund_windowed_pipeline(self.window_ms))
.map_err(|e| anyhow::anyhow!("resumed run: {e}"))?;
let resume_s = t.elapsed().as_secs_f64();
let resumed_epoch = resumed.resumed_from_epoch("win_out");
let closed_after_recovery = resumed.windows_closed("win_out");
anyhow::ensure!(
resumed_epoch.is_some(),
"resume must recover from a checkpoint"
);
let host_cores = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1);
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(self.rows));
m.insert("triggers".into(), json!(self.triggers));
m.insert("window_ms".into(), json!(self.window_ms));
m.insert(
"committed_before_crash".into(),
json!(committed_before_crash),
);
m.insert("resumed_from_epoch".into(), json!(resumed_epoch));
m.insert(
"windows_closed_after_recovery".into(),
json!(closed_after_recovery),
);
m.insert("cold_recompute_s".into(), json!(cold_s));
m.insert("crash_run_s".into(), json!(crash_s));
m.insert("restore_resume_s".into(), json!(resume_s));
m.insert(
"restore_resume_rows_per_s".into(),
json!(self.rows as f64 / resume_s.max(f64::MIN_POSITIVE)),
);
m.insert("peak_rss_bytes".into(), json!(peak_rss_bytes()));
m.insert("host_cores".into(), json!(host_cores));
m.insert("host_mem_bytes".into(), json!(host_mem_bytes()));
m.insert(
"elapsed_ms".into(),
json!((cold_s + crash_s + resume_s) * 1000.0),
);
Ok(BenchResult::measured("thund_recovery", m))
}
}
nornir_bench::register_bench!(ThundRecovery {
rows: 1_000_000,
triggers: 100,
window_ms: 1_000,
});
/// Þund **incremental aggregation** (design phase 5 bench arm
/// `knut.thund_incremental_agg`): a GROUPED windowed aggregate
/// (`SELECT g, COUNT(*), SUM(x) GROUP BY g`) driven by maintained accumulators
/// (no per-window SQL, raw rows never buffered — state bounded by the group
/// cardinality × concurrently-open windows) vs the recompute default (re-runs
/// the query over the whole growing accumulation each trigger). Reports
/// throughput (rows/s) per arm and the bounded incremental working set (open
/// accumulators) vs recompute's whole-stream working set, plus host facts.
struct ThundIncrementalAgg {
rows: usize,
triggers: usize,
window_ms: u64,
groups: i64,
}
impl Bencher for ThundIncrementalAgg {
fn id(&self) -> &'static str {
"knut.thund_incremental_agg"
}
fn run(&self) -> Result<BenchResult> {
use datafusion::arrow::array::{Int64Array, RecordBatch as DfBatch};
use datafusion::arrow::datatypes::{DataType, Field, Schema as DfSchema};
use knut_thund::backend::ExecBackend;
use knut_thund::backend::native::{NativeBackend, StreamingMode};
use knut_thund::ir::{
Dataset, Flow, FlowKind, OutputMode, OutputType, Pipeline, SourceSpec, Trigger,
Watermark, WindowSpec,
};
let per = self.rows / self.triggers;
let win = self.window_ms as i64;
let lateness = win; // >= a window: no on-time row dropped
let jitter = (win / 2).max(1);
let schema = Arc::new(DfSchema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("et", DataType::Int64, false),
Field::new("g", DataType::Int64, false),
Field::new("x", DataType::Int64, false),
]));
// Trigger t fills window [t*win, t*win+win); group = id % groups.
let mut mb: Vec<Vec<DfBatch>> = Vec::with_capacity(self.triggers);
let mut next: i64 = 0;
for t in 0..self.triggers {
let base = t as i64 * win;
let (mut ids, mut ets, mut gs, mut xs) = (
Vec::with_capacity(per),
Vec::with_capacity(per),
Vec::with_capacity(per),
Vec::with_capacity(per),
);
for r in 0..per {
ids.push(next);
ets.push(base + (r as i64 % jitter));
gs.push(next % self.groups);
xs.push((r as i64 % 1000) + 1);
next += 1;
}
mb.push(vec![DfBatch::try_new(
schema.clone(),
vec![
Arc::new(Int64Array::from(ids)),
Arc::new(Int64Array::from(ets)),
Arc::new(Int64Array::from(gs)),
Arc::new(Int64Array::from(xs)),
],
)?]);
}
let query = "SELECT g, COUNT(*) AS n, SUM(x) AS s, MIN(x) AS mn, MAX(x) AS mx FROM events GROUP BY g";
let pipeline = || {
let mut flow = Flow::streaming(
"f_events",
"agg_out",
SourceSpec::Kafka {
bootstrap: "localhost:9092".into(),
topic: "events".into(),
format: "json".into(),
},
)
.with_query(query)
.with_watermark(Watermark {
event_time_column: "et".into(),
allowed_lateness_ms: lateness as u64,
idle_timeout_ms: 0,
})
.with_window(WindowSpec::Tumbling {
size_ms: self.window_ms,
});
if let FlowKind::Streaming {
output_mode,
trigger,
..
} = &mut flow.kind
{
*output_mode = OutputMode::Append;
*trigger = Trigger::Continuous;
}
Pipeline::new("thund_agg_bench")
.with_dataset(Dataset::new("agg_out", OutputType::Table).incremental())
.with_flow(flow)
};
// ── INCREMENTAL-AGG arm: maintained accumulators, bounded state. ──
let state = tempfile::tempdir()?;
let state_uri = format!("file://{}", state.path().display());
let t = Instant::now();
let incremental = NativeBackend::new()
.with_streaming_mode(StreamingMode::Incremental)
.with_state_root(state_uri)
.with_stream_input("events", mb.clone())
.run(&pipeline())
.map_err(|e| anyhow::anyhow!("incremental-agg run: {e}"))?;
let incremental_s = t.elapsed().as_secs_f64();
let used_agg = incremental.used_incremental_agg("agg_out");
let windows_closed = incremental.windows_closed("agg_out");
let peak_open_acc = incremental.peak_open_rows("agg_out");
// ── RECOMPUTE arm (default): re-run over the whole growing accumulation. ──
let t = Instant::now();
let recompute = NativeBackend::new()
.with_stream_input("events", mb)
.run(&pipeline())
.map_err(|e| anyhow::anyhow!("recompute run: {e}"))?;
let recompute_s = t.elapsed().as_secs_f64();
let _ = recompute.row_count("agg_out");
anyhow::ensure!(
used_agg,
"the grouped aggregate must use the incremental-agg operator"
);
// The incremental working set is bounded by the concurrently-open
// accumulators (<= groups * open windows), never the whole stream.
let incremental_working_set = peak_open_acc;
let recompute_working_set = self.rows;
anyhow::ensure!(
incremental_working_set < recompute_working_set,
"incremental accumulators ({incremental_working_set}) must stay below recompute's accumulated rows ({recompute_working_set})"
);
let host_cores = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1);
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(self.rows));
m.insert("triggers".into(), json!(self.triggers));
m.insert("window_ms".into(), json!(self.window_ms));
m.insert("groups".into(), json!(self.groups));
m.insert("used_incremental_agg".into(), json!(used_agg));
m.insert("windows_closed".into(), json!(windows_closed));
m.insert("incremental_s".into(), json!(incremental_s));
m.insert("recompute_s".into(), json!(recompute_s));
m.insert(
"incremental_rows_per_s".into(),
json!(self.rows as f64 / incremental_s.max(f64::MIN_POSITIVE)),
);
m.insert(
"recompute_rows_per_s".into(),
json!(self.rows as f64 / recompute_s.max(f64::MIN_POSITIVE)),
);
m.insert(
"incremental_working_set_accumulators".into(),
json!(incremental_working_set),
);
m.insert(
"recompute_working_set_rows".into(),
json!(recompute_working_set),
);
m.insert("peak_rss_bytes".into(), json!(peak_rss_bytes()));
m.insert("host_cores".into(), json!(host_cores));
m.insert("host_mem_bytes".into(), json!(host_mem_bytes()));
m.insert(
"elapsed_ms".into(),
json!((incremental_s + recompute_s) * 1000.0),
);
Ok(BenchResult::measured("thund_incremental_agg", m))
}
}
nornir_bench::register_bench!(ThundIncrementalAgg {
rows: 2_000_000,
triggers: 100,
window_ms: 1_000,
groups: 50,
});
fn main() -> Result<()> {
nornir_bench::api::run_main_json()
}
// ── knut.doris_stream_load (feature `doris`) ────────────────────────────────
/// **Doris Stream Load delivery** (feature `doris`): the encode-and-dispatch
/// half of the CDC delivery path — Arrow change rows → chunked JSON bodies →
/// labelled Stream Loads — measured against an in-process fake transport.
///
/// This is deliberately the half that does **not** need a cluster. The network
/// half (the FE→BE 307 hop and the BE's own write) is integration work against
/// `red`; what this arm tracks continuously is whether the encode/chunk/label
/// path stays cheap enough that the delivery cost is dominated by the network
/// rather than by us. See `.nornir/doris-sink-design.md` f8: if
/// `read_changelog` + Stream Load is not an order of magnitude cheaper than a
/// `COMPLETE` MV refresh, the daemon should not run at all — and this arm is one
/// of the two numbers that decides it.
#[cfg(feature = "doris")]
struct DorisStreamLoad;
#[cfg(feature = "doris")]
impl nornir_bench::api::Bencher for DorisStreamLoad {
fn id(&self) -> &'static str {
"knut.doris_stream_load"
}
fn run(&self) -> Result<BenchResult> {
use knut_bifrost::sink::{
DorisConfig, DorisSink, SnapshotWindow, StreamLoadRequest, StreamLoadResponse,
StreamLoadTransport, TableWindow, deliver_window,
};
use std::sync::atomic::{AtomicUsize, Ordering};
/// Accepts everything, counts bytes. No network, no cluster.
struct NullTransport {
bytes: AtomicUsize,
loads: AtomicUsize,
}
#[async_trait::async_trait]
impl StreamLoadTransport for NullTransport {
async fn send(
&self,
req: &StreamLoadRequest,
) -> knut_bifrost::Result<StreamLoadResponse> {
if req.method == "GET" {
return Ok(StreamLoadResponse::ok(
r#"{"status":200,"data":{"properties":[]}}"#,
));
}
// One FE hop, then the BE. Mirror the documented 307 so the
// bench pays for the same number of round trips the real path
// does.
if req.url.starts_with("http://fe") {
return Ok(StreamLoadResponse::redirect(
"http://be:8040/api/b/t/_stream_load",
));
}
self.bytes.fetch_add(req.body.len(), Ordering::Relaxed);
self.loads.fetch_add(1, Ordering::Relaxed);
let n = req.body.iter().filter(|b| **b == b'}').count();
Ok(StreamLoadResponse::ok(format!(
r#"{{"Status":"Success","NumberLoadedRows":{n},"NumberFilteredRows":0}}"#
)))
}
}
let batch = sample_batch(ROWS);
let dir = tempfile::tempdir()?; // /tmp — local disk, never T9
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?;
let transport = NullTransport {
bytes: AtomicUsize::new(0),
loads: AtomicUsize::new(0),
};
let cfg = DorisConfig::new("http://fe:8030", "bench", "rows", "u", "p")
.with_max_chunk_bytes(8 * 1024 * 1024);
let mut sink = DorisSink::new(cfg, transport);
let mut state = knut_bifrost::SyncState::open(dir.path().join("state.json"))?;
let ups = [batch];
let window = TableWindow {
table: "bench.rows",
snapshots: SnapshotWindow::new(Some(1), 2),
upserts: &ups,
deletes: &[],
needs_full_reload: false,
};
let t = Instant::now();
let ack = rt
.block_on(deliver_window(&mut sink, &mut state, &window))
.map_err(|e| anyhow::anyhow!("{e}"))?;
let load_s = t.elapsed().as_secs_f64();
anyhow::ensure!(ack.rows_loaded == ROWS, "row count mismatch: {}", ack.rows_loaded);
// A replay must cost the same encode and write nothing — measured, so a
// regression that starts double-loading shows up as a row-count change
// rather than as silence.
let t2 = Instant::now();
let replay = rt
.block_on(deliver_window(&mut sink, &mut state, &window))
.map_err(|e| anyhow::anyhow!("{e}"))?;
let replay_s = t2.elapsed().as_secs_f64();
anyhow::ensure!(
replay.labels == ack.labels,
"replay derived different labels — replay safety is broken"
);
let mut m = serde_json::Map::new();
m.insert("rows".into(), json!(ROWS));
m.insert("chunks".into(), json!(ack.labels.len()));
m.insert("load_s".into(), json!(load_s));
m.insert("rows_per_s".into(), json!(ROWS as f64 / load_s));
m.insert("replay_s".into(), json!(replay_s));
m.insert("replay_rows_loaded".into(), json!(replay.rows_loaded));
Ok(BenchResult::measured("doris_stream_load", m))
}
}
#[cfg(feature = "doris")]
nornir_bench::register_bench!(DorisStreamLoad);