Add benchmarks for historical fixed, historical sliding, live injection, and storage write throughput

Author: argahsuknesib

.zed/settings.json

@@ -42,6 +42,25 @@ path = "src/main.rs"
 name = "http_server"
 path = "src/bin/http_server.rs"
 
+[dev-dependencies]
+criterion = { version = "0.5", features = ["html_reports"] }
+
+[[bench]]
+name = "storage_write"
+harness = false
+
+[[bench]]
+name = "historical_fixed"
+harness = false
+
+[[bench]]
+name = "historical_sliding"
+harness = false
+
+[[bench]]
+name = "live_injection"
+harness = false
+
 [profile.release]
 opt-level = 3
 lto = true

Cargo.toml

@@ -0,0 +1,79 @@
+use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion};
+use janus::{
+    execution::historical_executor::HistoricalExecutor,
+    parsing::janusql_parser::{WindowDefinition, WindowType},
+    querying::oxigraph_adapter::OxigraphAdapter,
+    storage::{segmented_storage::StreamingSegmentedStorage, util::StreamingConfig},
+};
+use std::sync::{
+    atomic::{AtomicU64, Ordering},
+    Arc,
+};
+use std::time::{SystemTime, UNIX_EPOCH};
+
+static COUNTER: AtomicU64 = AtomicU64::new(0);
+
+fn unique_config() -> StreamingConfig {
+    let id = COUNTER.fetch_add(1, Ordering::Relaxed);
+    let ts = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos();
+    StreamingConfig {
+        segment_base_path: format!("/tmp/janus_bench_fixed_{}_{}", ts, id),
+        max_batch_events: 1_000_000,
+        max_batch_age_seconds: 3600,
+        max_batch_bytes: 1_000_000_000,
+        sparse_interval: 64,
+        entries_per_index_block: 256,
+    }
+}
+
+/// Write N events at timestamps [1000, 1000+N) into a fresh storage.
+/// These land in the in-memory batch buffer — no flush needed before querying.
+fn setup(n: usize) -> (Arc<StreamingSegmentedStorage>, WindowDefinition) {
+    let storage = StreamingSegmentedStorage::new(unique_config()).unwrap();
+    for i in 0..n as u64 {
+        storage
+            .write_rdf(
+                1_000 + i,
+                &format!("http://example.org/sensor{}", i % 5),
+                "http://saref.etsi.org/core/hasValue",
+                &format!("{}", 20 + (i % 10)),
+                "http://example.org/graph",
+            )
+            .unwrap();
+    }
+    let window = WindowDefinition {
+        window_name: "w".to_string(),
+        stream_name: "http://example.org/stream".to_string(),
+        width: n as u64,
+        slide: n as u64,
+        offset: None,
+        start: Some(1_000),
+        end: Some(1_000 + n as u64 - 1),
+        window_type: WindowType::HistoricalFixed,
+    };
+    (Arc::new(storage), window)
+}
+
+const SPARQL: &str = "SELECT ?s ?p ?o WHERE { ?s ?p ?o }";
+
+fn historical_fixed(c: &mut Criterion) {
+    let mut group = c.benchmark_group("historical/fixed_window");
+
+    for &n in &[100usize, 1_000, 10_000] {
+        group.bench_with_input(BenchmarkId::new("events", n), &n, |b, &n| {
+            b.iter_batched(
+                || setup(n),
+                |(storage, window)| {
+                    let executor = HistoricalExecutor::new(storage, OxigraphAdapter::new());
+                    black_box(executor.execute_fixed_window(&window, SPARQL).unwrap())
+                },
+                criterion::BatchSize::SmallInput,
+            );
+        });
+    }
+
+    group.finish();
+}
+
+criterion_group!(benches, historical_fixed);
+criterion_main!(benches);

benches/historical_fixed.rs

@@ -0,0 +1,92 @@
+use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion};
+use janus::{
+    execution::historical_executor::HistoricalExecutor,
+    parsing::janusql_parser::{WindowDefinition, WindowType},
+    querying::oxigraph_adapter::OxigraphAdapter,
+    storage::{segmented_storage::StreamingSegmentedStorage, util::StreamingConfig},
+};
+use std::sync::{
+    atomic::{AtomicU64, Ordering},
+    Arc,
+};
+use std::time::{SystemTime, UNIX_EPOCH};
+
+static COUNTER: AtomicU64 = AtomicU64::new(0);
+
+fn unique_config() -> StreamingConfig {
+    let id = COUNTER.fetch_add(1, Ordering::Relaxed);
+    let ts = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos();
+    StreamingConfig {
+        segment_base_path: format!("/tmp/janus_bench_sliding_{}_{}", ts, id),
+        max_batch_events: 1_000_000,
+        max_batch_age_seconds: 3600,
+        max_batch_bytes: 1_000_000_000,
+        sparse_interval: 64,
+        entries_per_index_block: 256,
+    }
+}
+
+// Window config: OFFSET=10_000ms, RANGE=2_000ms, SLIDE=1_000ms
+// SlidingWindowIterator scans [now-10000, now] with 8 overlapping windows.
+// Data is written at [now-8000, now-2000] — solidly within the scan range.
+const OFFSET_MS: u64 = 10_000;
+const RANGE_MS: u64 = 2_000;
+const SLIDE_MS: u64 = 1_000;
+const DATA_START_BEFORE_NOW_MS: u64 = 8_000;
+const DATA_SPAN_MS: u64 = 6_000;
+
+fn setup(n: usize) -> (Arc<StreamingSegmentedStorage>, WindowDefinition) {
+    let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis() as u64;
+    let storage = StreamingSegmentedStorage::new(unique_config()).unwrap();
+    let n64 = n as u64;
+    for i in 0..n64 {
+        let ts = now - DATA_START_BEFORE_NOW_MS + i * DATA_SPAN_MS / n64.max(1);
+        storage
+            .write_rdf(
+                ts,
+                &format!("http://example.org/sensor{}", i % 5),
+                "http://saref.etsi.org/core/hasValue",
+                &format!("{}", 20 + (i % 10)),
+                "http://example.org/graph",
+            )
+            .unwrap();
+    }
+    let window = WindowDefinition {
+        window_name: "w".to_string(),
+        stream_name: "http://example.org/stream".to_string(),
+        width: RANGE_MS,
+        slide: SLIDE_MS,
+        offset: Some(OFFSET_MS),
+        start: None,
+        end: None,
+        window_type: WindowType::HistoricalSliding,
+    };
+    (Arc::new(storage), window)
+}
+
+const SPARQL: &str = "SELECT ?s ?p ?o WHERE { ?s ?p ?o }";
+
+fn historical_sliding(c: &mut Criterion) {
+    let mut group = c.benchmark_group("historical/sliding_window");
+
+    for &n in &[100usize, 1_000, 10_000] {
+        group.bench_with_input(BenchmarkId::new("events", n), &n, |b, &n| {
+            b.iter_batched(
+                || setup(n),
+                |(storage, window)| {
+                    let executor = HistoricalExecutor::new(storage, OxigraphAdapter::new());
+                    // Collect all window results — the iterator is finite and exits naturally
+                    let results: Vec<_> =
+                        executor.execute_sliding_windows(&window, SPARQL).collect();
+                    black_box(results)
+                },
+                criterion::BatchSize::SmallInput,
+            );
+        });
+    }
+
+    group.finish();
+}
+
+criterion_group!(benches, historical_sliding);
+criterion_main!(benches);

benches/historical_sliding.rs

@@ -0,0 +1,80 @@
+use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion};
+use janus::{core::RDFEvent, stream::live_stream_processing::LiveStreamProcessing};
+use std::time::Instant;
+
+const STREAM_URI: &str = "http://example.org/stream1";
+
+// RSP-QL query: 10s range, 1s step window over stream1
+const RSPQL: &str = r#"
+    PREFIX ex: <http://example.org/>
+    REGISTER RStream <output> AS
+    SELECT ?s ?p ?o
+    FROM NAMED WINDOW ex:w ON STREAM ex:stream1 [RANGE 10000 STEP 1000]
+    WHERE {
+        WINDOW ex:w { ?s ?p ?o }
+    }
+"#;
+
+fn make_event(timestamp_ms: u64, i: u64) -> RDFEvent {
+    RDFEvent::new(
+        timestamp_ms,
+        &format!("http://example.org/sensor{}", i % 5),
+        "http://saref.etsi.org/core/hasValue",
+        &format!("{}", 20 + (i % 10)),
+        "",
+    )
+}
+
+/// Wait for the first live result with a 10-second hard deadline.
+/// Panics with a clear message if nothing arrives — indicates the RSP engine
+/// is not emitting results for the injected events.
+fn wait_for_result(proc: &LiveStreamProcessing) -> rsp_rs::BindingWithTimestamp {
+    let deadline = Instant::now() + std::time::Duration::from_secs(10);
+    loop {
+        if let Some(result) = proc.try_receive_result().unwrap() {
+            return result;
+        }
+        assert!(
+            Instant::now() < deadline,
+            "live_injection: no result within 10s — RSP engine did not emit for injected events"
+        );
+        std::thread::yield_now();
+    }
+}
+
+fn live_injection(c: &mut Criterion) {
+    let mut group = c.benchmark_group("live/event_injection");
+    // Lower sample size: each iteration spawns an RSP engine thread
+    group.sample_size(20);
+
+    for &n in &[1usize, 10, 100] {
+        group.bench_with_input(BenchmarkId::new("events_per_window", n), &n, |b, &n| {
+            b.iter_batched(
+                || {
+                    let mut proc = LiveStreamProcessing::new(RSPQL.to_string()).unwrap();
+                    proc.register_stream(STREAM_URI).unwrap();
+                    proc.start_processing().unwrap();
+                    proc
+                },
+                |proc| {
+                    // Spread N events evenly across [0, 9000] ms (inside the RANGE 10000 window)
+                    let n64 = n as u64;
+                    for i in 0..n64 {
+                        let ts = if n64 > 1 { i * 9_000 / (n64 - 1) } else { 0 };
+                        proc.add_event(STREAM_URI, make_event(ts, i)).unwrap();
+                    }
+                    // Sentinel at 20_000 ms closes all open windows
+                    proc.add_event(STREAM_URI, make_event(20_000, 999)).unwrap();
+                    // Block until first result arrives
+                    black_box(wait_for_result(&proc))
+                },
+                criterion::BatchSize::SmallInput,
+            );
+        });
+    }
+
+    group.finish();
+}
+
+criterion_group!(benches, live_injection);
+criterion_main!(benches);

benches/live_injection.rs

@@ -0,0 +1,52 @@
+use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion, Throughput};
+use janus::storage::{segmented_storage::StreamingSegmentedStorage, util::StreamingConfig};
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::{SystemTime, UNIX_EPOCH};
+
+static COUNTER: AtomicU64 = AtomicU64::new(0);
+
+fn unique_config() -> StreamingConfig {
+    let id = COUNTER.fetch_add(1, Ordering::Relaxed);
+    let ts = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos();
+    StreamingConfig {
+        segment_base_path: format!("/tmp/janus_bench_write_{}_{}", ts, id),
+        // Large thresholds so no flush happens during measurement
+        max_batch_events: 1_000_000,
+        max_batch_age_seconds: 3600,
+        max_batch_bytes: 1_000_000_000,
+        sparse_interval: 64,
+        entries_per_index_block: 256,
+    }
+}
+
+fn storage_write(c: &mut Criterion) {
+    let mut group = c.benchmark_group("storage/write_throughput");
+
+    for &n in &[100usize, 1_000, 10_000, 100_000] {
+        group.throughput(Throughput::Elements(n as u64));
+        group.bench_with_input(BenchmarkId::from_parameter(n), &n, |b, &n| {
+            b.iter_batched(
+                || StreamingSegmentedStorage::new(unique_config()).unwrap(),
+                |storage| {
+                    for i in 0..n as u64 {
+                        storage
+                            .write_rdf(
+                                black_box(1_000 + i),
+                                &format!("http://example.org/sensor{}", i % 5),
+                                "http://saref.etsi.org/core/hasValue",
+                                &format!("{}", 20 + (i % 10)),
+                                "http://example.org/graph",
+                            )
+                            .unwrap();
+                    }
+                },
+                criterion::BatchSize::SmallInput,
+            );
+        });
+    }
+
+    group.finish();
+}
+
+criterion_group!(benches, storage_write);
+criterion_main!(benches);