nordisk/knut

crates/knut-rel2graph-bench/src/bin/timing.rs

//! **Timing bencher** — a deterministic, seeded timed loop (no external bench
//! crate, no wall-clock randomness: the DATA is seeded; only elapsed wall time
//! is measured, via the monotonic `Instant`).
//!
//! Measures, at scales 1x / 10x / 100x:
//!   * `GraphPlan::from_mapping`  — canonical plan construction,
//!   * `to_thund_pipeline`        — the thund projection,
//!   * `build_mutation`           — running the transform fabric over the data,
//! and reports graph-element throughput (nodes+edges per second) for the
//! mutation build. Run with `cargo run -p knut-rel2graph-bench --release --bin
//! rel2graph-timing`.

use std::time::{Duration, Instant};

use knut_bifrost::{GraphPlan, codegen::CodegenOptions};
use knut_rel2graph_bench::{
    DEFAULT_SEED, GeneratedDb, build_mutation, generate, mapping_spec, summarize,
};

/// Run `f` `iters` times (after `warmup` runs) and return the median duration.
fn median_time<T>(warmup: usize, iters: usize, mut f: impl FnMut() -> T) -> Duration {
    for _ in 0..warmup {
        std::hint::black_box(f());
    }
    let mut samples: Vec<Duration> = Vec::with_capacity(iters);
    for _ in 0..iters {
        let t = Instant::now();
        std::hint::black_box(f());
        samples.push(t.elapsed());
    }
    samples.sort();
    samples[samples.len() / 2]
}

fn per_sec(count: usize, d: Duration) -> f64 {
    if d.as_secs_f64() == 0.0 {
        f64::INFINITY
    } else {
        count as f64 / d.as_secs_f64()
    }
}

fn main() {
    let scales = [1usize, 10, 100];
    let spec = mapping_spec();
    let opts = CodegenOptions::default();

    println!("rel2graph timing harness — seed {DEFAULT_SEED:#x}, 12-table ERP schema\n");
    println!(
        "{:>6} | {:>10} | {:>10} | {:>10} | {:>12} | {:>10} | {:>10} | {:>10} | {:>14}",
        "scale",
        "rel.rows",
        "nodes",
        "edges",
        "from_mapping",
        "to_thund",
        "build_mut",
        "verify",
        "elems/sec*"
    );
    println!("{}", "-".repeat(120));

    for &scale in &scales {
        // Data is generated ONCE per scale (not part of any timed section).
        let db: GeneratedDb = generate(scale, DEFAULT_SEED);
        let rel_rows = db.total_rows();

        // A summary (untimed) for the element counts used in throughput.
        let summary = summarize(&build_mutation(&spec, &db));
        let elems = summary.total_nodes() + summary.total_edges();

        // Pick iteration counts so even the big scale finishes quickly.
        let (warmup, iters) = if scale >= 100 { (1, 5) } else { (3, 25) };

        let t_plan = median_time(warmup, iters, || {
            GraphPlan::from_mapping(&spec, &opts).expect("plan")
        });
        let plan = GraphPlan::from_mapping(&spec, &opts).expect("plan");
        let t_thund = median_time(warmup, iters, || {
            knut_thund::rel2graph::to_thund_pipeline(&plan).expect("thund")
        });
        let t_build = median_time(warmup, iters, || build_mutation(&spec, &db));
        // The row-touching pass: full endpoint resolution over every element —
        // the one stage whose cost genuinely scales with data volume.
        let mutation = build_mutation(&spec, &db);
        let t_verify = median_time(warmup, iters, || summarize(&mutation));

        println!(
            "{:>6} | {:>10} | {:>10} | {:>10} | {:>12} | {:>10} | {:>10} | {:>10} | {:>14.0}",
            format!("{scale}x"),
            rel_rows,
            summary.total_nodes(),
            summary.total_edges(),
            fmt(t_plan),
            fmt(t_thund),
            fmt(t_build),
            fmt(t_verify),
            per_sec(elems, t_verify),
        );
    }
    println!(
        "\nNotes:\n\
         * from_mapping / to_thund are shape-only — O(schema), scale-invariant.\n\
         * build_mut is ZERO-COPY (Arrow columns Arc-cloned into VertexSet/EdgeSet),\n\
         \x20 so it is O(#pipeline steps), NOT O(rows) — per-row work lives in the sink.\n\
         * verify is the correctness harness's full per-element endpoint resolution;\n\
         \x20 it touches every node+edge, so elems/sec* is measured against THAT pass."
    );
}

/// Human-friendly duration.
fn fmt(d: Duration) -> String {
    let us = d.as_secs_f64() * 1e6;
    if us < 1000.0 {
        format!("{us:.1}us")
    } else {
        format!("{:.2}ms", us / 1000.0)
    }
}