Update benchmark results and improve code clarity in various modules

Author: argahsuknesib

README.md

@@ -128,23 +128,26 @@ current performance guarantee.
 ## Representative benchmark results
 
 The versioned [result-analysis package](./data/result-analysis/) contains the
-CSV summaries and figures behind the following comparison with Oxigraph. The
-historical-access values are mean ± standard deviation; the storage values are
-medians from 35 iterations. They describe the included workloads and should
-not be interpreted as a general-purpose database benchmark.
-
-| Workload | Janus | Oxigraph | Relative result |
-| --- | ---: | ---: | --- |
-| Point lookup, 1M quads | 0.068 ± 0.004 ms | 818.002 ± 8.683 ms | 12,029× lower mean latency |
-| Fixed 60-second range, 1M quads | 1.247 ± 0.060 ms | 845.388 ± 6.130 ms | 678× lower mean latency |
-| 50% historical range, 1M quads | 541.342 ± 6.438 ms | 1,125.296 ± 5.510 ms | 2.08× lower mean latency |
-| Full historical range, 1M quads | 1,075.730 ± 2.685 ms | 1,454.357 ± 8.035 ms | 1.35× lower mean latency |
-| Persistent footprint, 1M events | 23.14 MB | 302.83 MB | 13.1× smaller median footprint |
-| Storage ingestion, 1M events | 1.06M events/s | 0.121M events/s | 8.8× higher median throughput |
-
-See the [historical-access CSV](./data/result-analysis/historical_access_latency_parsed.csv),
-[storage-footprint CSV](./data/result-analysis/storage_footprint_summary.csv),
-and the accompanying [historical-access figure](./data/result-analysis/historical_access_latency_5panel_shared_yaxis.png).
+figures for the included Janus/Oxigraph workloads. They are workload- and
+machine-dependent, not general-purpose database guarantees.
+
+### Historical access latency
+
+![Historical access latency comparison](./data/result-analysis/historical_access_latency_5panel_shared_yaxis.png)
+
+### Storage footprint
+
+![Storage footprint comparison](./data/result-analysis/resource_storage_footprint_combined.png)
+
+### Memory and CPU
+
+![Median memory and CPU comparison](./data/result-analysis/memory_cpu_1m_median_side_by_side.png)
+
+![Peak memory comparison](./data/result-analysis/peak_memory_1m_lineplot.png)
+
+![Median peak memory comparison](./data/result-analysis/peak_memory_1m_median_lineplot.png)
+
+![Compact median peak memory comparison](./data/result-analysis/peak_memory_1m_median_lineplot_compact.png)
 
 ## Development
 

benches/historical_fixed.rs

@@ -17,8 +17,8 @@ fn setup(n: usize) -> (Arc<StreamingSegmentedStorage>, WindowDefinition) {
     populate_storage(&storage, n, 1_000, 1, GRAPH_URI);
     let window = WindowDefinition {
         window_name: "w".to_string(),
-        source_kind: SourceKind::Stream,
-        stream_name: GRAPH_URI.to_string(),
+        source_kind: SourceKind::Log,
+        source_name: GRAPH_URI.to_string(),
         width: n as u64,
         slide: n as u64,
         offset: None,

benches/historical_sliding.rs

@@ -26,8 +26,8 @@ fn setup(n: usize) -> (Arc<StreamingSegmentedStorage>, WindowDefinition) {
     populate_storage(&storage, n, start_ts, step_ms, GRAPH_URI);
     let window = WindowDefinition {
         window_name: "w".to_string(),
-        source_kind: SourceKind::Stream,
-        stream_name: GRAPH_URI.to_string(),
+        source_kind: SourceKind::Log,
+        source_name: GRAPH_URI.to_string(),
         width: RANGE_MS,
         slide: SLIDE_MS,
         offset: Some(OFFSET_MS),

src/bin/hybrid_scaling_combined.rs

@@ -1,3 +1,5 @@
+#![allow(clippy::too_many_arguments)]
+
 use clap::Parser;
 use janus::core::RDFEvent;
 use janus::execution::result_converter::parse_rsprs_binding_string;
@@ -393,9 +395,10 @@ fn normalize_binding_term(raw: &str) -> String {
             return without_prefix[..end].to_string();
         }
     }
-    if trimmed.starts_with('<') && trimmed.ends_with('>') && trimmed.len() > 2 {
-        trimmed[1..trimmed.len() - 1].to_string()
-    } else if trimmed.starts_with('"') && trimmed.ends_with('"') && trimmed.len() > 2 {
+    if ((trimmed.starts_with('<') && trimmed.ends_with('>'))
+        || (trimmed.starts_with('"') && trimmed.ends_with('"')))
+        && trimmed.len() > 2
+    {
         trimmed[1..trimmed.len() - 1].to_string()
     } else {
         trimmed.to_string()
@@ -750,7 +753,11 @@ fn congestion_value_for_historical(index: usize) -> f64 {
 fn congestion_value_for_live(index: usize) -> f64 {
     let sensor_idx = index % SENSOR_COUNT;
     let sample_idx = index / SENSOR_COUNT;
-    let bias = if sensor_idx % 2 == 0 { 8.0 } else { -8.0 };
+    let bias = if sensor_idx.is_multiple_of(2) {
+        8.0
+    } else {
+        -8.0
+    };
     let oscillation = ((sample_idx * 5 + sensor_idx) % 5) as f64 - 2.0;
     historical_sensor_base(sensor_idx) + bias + oscillation
 }
@@ -824,14 +831,14 @@ fn safe_takeaway_label(system: &str, query_type: &str) -> &'static str {
 fn format_size_label(size: usize) -> String {
     match size {
         1_000_000 => "1M quads".to_string(),
-        1_000..=999_999 if size % 1_000 == 0 => format!("{}k quads", size / 1_000),
+        1_000..=999_999 if size.is_multiple_of(1_000) => format!("{}k quads", size / 1_000),
         _ => format!("{size} quads"),
     }
 }
 
-fn find_first_baseline_definition<'a>(
-    parsed: &'a ParsedJanusQuery,
-) -> Result<(&'a BaselineDefinition, &'a BaselineGraphTemplate), Box<dyn std::error::Error>> {
+fn find_first_baseline_definition(
+    parsed: &ParsedJanusQuery,
+) -> Result<(&BaselineDefinition, &BaselineGraphTemplate), Box<dyn std::error::Error>> {
     let definition = parsed
         .ast
         .baseline_definitions
@@ -1040,10 +1047,11 @@ fn run_janus_unified(
                 first_hybrid_result_ms = Some(*received_at);
                 window_processing_overhead_ms = Some(*overhead);
             }
-        } else if slides.len() >= 2 && res.timestamp_to == slides[1] {
-            if main_window_result_ms.is_none() {
-                main_window_result_ms = Some(*received_at);
-            }
+        } else if slides.len() >= 2
+            && res.timestamp_to == slides[1]
+            && main_window_result_ms.is_none()
+        {
+            main_window_result_ms = Some(*received_at);
         }
     }
 
@@ -1172,12 +1180,12 @@ fn run_decomposed_oxigraph(
         {
             Term::NamedNode(NamedNode::new(&event.object)?)
         } else {
-            let literal = if let Ok(_) = event.object.parse::<f64>() {
+            let literal = if event.object.parse::<f64>().is_ok() {
                 oxigraph::model::Literal::new_typed_literal(
                     &event.object,
                     NamedNode::new("http://www.w3.org/2001/XMLSchema#decimal").unwrap(),
                 )
-            } else if let Ok(_) = event.object.parse::<i64>() {
+            } else if event.object.parse::<i64>().is_ok() {
                 oxigraph::model::Literal::new_typed_literal(
                     &event.object,
                     NamedNode::new("http://www.w3.org/2001/XMLSchema#integer").unwrap(),
@@ -1192,7 +1200,7 @@ fn run_decomposed_oxigraph(
         // Quad 2: Event timestamp in Default Graph
         let event_node = NamedNode::new(&event_graph_uri)?;
         let ts_literal = Term::Literal(oxigraph::model::Literal::new_typed_literal(
-            &event.timestamp.to_string(),
+            event.timestamp.to_string(),
             NamedNode::new("http://www.w3.org/2001/XMLSchema#integer").unwrap(),
         ));
         store.insert(&Quad::new(
@@ -1221,7 +1229,7 @@ fn run_decomposed_oxigraph(
     let evaluator = build_evaluator();
     let parsed_query = evaluator
         .parse_query(&sparql_query)
-        .map_err(|e| oxigraph::sparql::QueryEvaluationError::from(e))?;
+        .map_err(oxigraph::sparql::QueryEvaluationError::from)?;
     let results = parsed_query.on_store(&store).execute()?;
 
     let mut external_bindings = Vec::new();
@@ -1341,10 +1349,11 @@ fn run_decomposed_oxigraph(
                 window_processing_overhead_ms = Some(*overhead);
                 external_merge_ms_total = Some(*merge_ms);
             }
-        } else if slides.len() >= 2 && res.timestamp_to == slides[1] {
-            if main_window_result_ms.is_none() {
-                main_window_result_ms = Some(*received_at + *merge_ms);
-            }
+        } else if slides.len() >= 2
+            && res.timestamp_to == slides[1]
+            && main_window_result_ms.is_none()
+        {
+            main_window_result_ms = Some(*received_at + *merge_ms);
         }
     }
 
@@ -1429,7 +1438,7 @@ fn write_reports(
     let mut query_types = rows.iter().map(|r| r.historical_query_type.clone()).collect::<Vec<_>>();
     query_types.sort_unstable();
     query_types.dedup();
-    let preferred_order = vec![
+    let preferred_order = [
         "point_lookup".to_string(),
         "fixed_60s".to_string(),
         "range_10_percent".to_string(),

src/execution/historical_executor.rs

@@ -435,10 +435,7 @@ impl<'a> Iterator for SlidingWindowIterator<'a> {
 
     fn next(&mut self) -> Option<Self::Item> {
         let window_start = self.current_start;
-        let window_end = match window_start.checked_add(self.width) {
-            Some(window_end) => window_end,
-            None => return None,
-        };
+        let window_end = window_start.checked_add(self.width)?;
 
         if window_end > self.evaluation_time {
             return None;

src/paper_bench/harness/data_gen.rs

@@ -209,5 +209,5 @@ fn congestion_value_for_live(index: usize) -> f64 {
 }
 
 fn historical_end_timestamp(start_ts: u64, events: usize) -> u64 {
-    start_ts + events.saturating_sub(1) as u64 * HISTORICAL_INTERVAL_MS
+    start_ts + events.max(1) as u64 * HISTORICAL_INTERVAL_MS
 }

src/paper_bench/harness/helpers.rs

@@ -1,3 +1,5 @@
+#![allow(clippy::implicit_hasher)]
+
 use super::types::{
     SustainedRunConfig, TimeMode, BASELINE_NS, CONGESTION_PREDICATE, GRAPH_URI, LIVE_STREAM_URI,
 };
@@ -213,20 +215,19 @@ pub fn hybrid_query(start_ts: u64, end_ts: u64) -> String {
 }
 
 pub fn historical_baseline_sparql_query() -> Result<String, Box<dyn std::error::Error>> {
-    Ok(format!(
-        r#"
+    Ok(r#"
         PREFIX ex: <http://example.org/>
 
         SELECT ?sensor
                (AVG(?historicalCongestion) AS ?historicalAvgCongestion)
-        WHERE {{
-            GRAPH ex:citybench {{
+        WHERE {
+            GRAPH ex:citybench {
                 ?sensor ex:congestionLevel ?historicalCongestion .
-            }}
-        }}
+            }
+        }
         GROUP BY ?sensor
         "#
-    ))
+    .to_string())
 }
 
 pub fn live_only_rspql() -> String {

src/paper_bench/query_defined_baseline/rdf.rs

@@ -1,3 +1,5 @@
+#![allow(clippy::implicit_hasher)]
+
 use oxigraph::model::{BlankNode, GraphName, NamedNode, NamedOrBlankNode, Quad, Term};
 use std::collections::{HashMap, HashSet};
 

src/paper_bench/query_defined_baseline/validation.rs

@@ -1,3 +1,5 @@
+#![allow(clippy::implicit_hasher)]
+
 use std::collections::{BTreeMap, HashMap};
 
 use super::rdf::{normalize_binding_term, parse_numeric};

src/paper_bench/storage_footprint.rs

@@ -690,7 +690,7 @@ fn median(values: &[f64]) -> f64 {
     sorted.sort_by(f64::total_cmp);
     let mid = sorted.len() / 2;
     if sorted.len() % 2 == 0 {
-        (sorted[mid - 1] + sorted[mid]) / 2.0
+        f64::midpoint(sorted[mid - 1], sorted[mid])
     } else {
         sorted[mid]
     }