feat(bench): add the B1 predicate-pushdown bench + zstd|grep reference

crates/ourios-bench/Cargo.toml

@@ -68,6 +68,10 @@ criterion = "0.5"
 ourios-querier = { path = "../ourios-querier" }
 tokio = { version = "1", default-features = false, features = ["rt"] }
 
+[[bench]]
+name = "b1"
+harness = false
+
 [[bench]]
 name = "b2"
 harness = false

crates/ourios-bench/benches/b1.rs

@@ -0,0 +1,195 @@
+//! B1 — predicate-pushdown query latency vs. the `zstdcat | grep`
+//! reference (`docs/benchmarks.md` §3 B1; RFC0007.1).
+//!
+//! Supportive **wall-clock** evidence for the B1 thesis gate
+//! ("Ourios ≥ 10× faster than `zstdcat files_in_range.zst | grep ERROR
+//! | wc -l`"). The structural side — that the selective query prunes
+//! row groups rather than scanning them — is proven deterministically
+//! in `ourios-querier`'s `rfc0007_1_*` tests; here we measure the thing
+//! the structural test can't: the latency ratio.
+//!
+//! `b1/synthetic` (always runs) — the controlled instrument. A query
+//! window holds ERROR rows (one hour/file) + INFO rows (another
+//! hour/file); out-of-window ERROR filler sits in a later hour. The
+//! Ourios query (`tenant` + time window + `severity_text='ERROR'`)
+//! prunes the INFO row group (severity statistics) **and** the
+//! out-of-window file (time statistics), so it reads only the in-window
+//! ERROR row group. The reference is given the **same in-window file
+//! set** (fairness — see [`ourios_bench::ReferenceCorpus`]) and greps
+//! every line. Two timings land in the group, `ourios` and
+//! `zstd-grep-reference`; the B1 ratio is `reference / ourios`.
+//!
+//! `b1/otel-demo` is deferred to the corpus run: it needs the OTLP/JSON
+//! corpus loader's per-record severity (the plain-text loader forces
+//! INFO) plus in-window raw-line extraction with a real time window —
+//! wired when the staged corpus lands, not here.
+
+use std::hint::black_box;
+use std::path::Path;
+
+use criterion::{Criterion, criterion_group, criterion_main};
+
+use ourios_bench::ReferenceCorpus;
+use ourios_core::audit::ParamType;
+use ourios_core::record::{BodyKind, MinedRecord, Param};
+use ourios_core::tenant::TenantId;
+use ourios_parquet::{PartitionKey, Writer};
+use ourios_querier::{Querier, QueryRequest};
+
+/// 2026-04-02T10:58:00 UTC (hour 10) — stable partition anchor.
+const TS0: u64 = 1_775_127_480_000_000_000;
+const HOUR_NS: u64 = 3_600_000_000_000;
+
+/// In-window ERROR rows (the B1 result) and INFO rows (pruned by
+/// severity), plus out-of-window ERROR filler (pruned by time).
+const ERROR_ROWS: u64 = 2_000;
+const INFO_ROWS: u64 = 2_000;
+const OUT_OF_WINDOW_ROWS: u64 = 4_000;
+/// zstd level for the reference `*.zst` (matches A1's reference codec).
+const ZSTD_LEVEL: i32 = 19;
+
+fn rec(template_id: u64, ts_ns: u64, severity: &str) -> MinedRecord {
+    let severity_number: u8 = match severity {
+        "INFO" => 9,
+        "ERROR" => 17,
+        _ => 0,
+    };
+    MinedRecord {
+        tenant_id: TenantId::new("a"),
+        template_id,
+        template_version: 1,
+        severity_number,
+        severity_text: Some(severity.to_string()),
+        scope_name: Some("lib.cart".to_string()),
+        scope_version: Some("1.0.0".to_string()),
+        time_unix_nano: ts_ns,
+        observed_time_unix_nano: Some(ts_ns + 1_000),
+        attributes: Vec::new(),
+        dropped_attributes_count: 0,
+        resource_attributes: Vec::new(),
+        trace_id: None,
+        span_id: None,
+        flags: 0x01,
+        event_name: None,
+        body_kind: BodyKind::String,
+        params: vec![Param {
+            type_tag: ParamType::Num,
+            value: "42".to_string(),
+        }],
+        separators: vec![String::new(), " ".to_string()],
+        body: None,
+        confidence: 1.0,
+        lossy_flag: false,
+    }
+}
+
+/// Write `records` (sharing one partition) as a single file.
+fn write_one_file(bucket: &Path, records: &[MinedRecord]) {
+    let part = PartitionKey::derive(&records[0]).expect("derive partition");
+    let mut w = Writer::open(bucket, part).expect("open writer");
+    w.append_records(records).expect("append");
+    w.close().expect("close");
+}
+
+/// Build the synthetic store and the matching reference corpus. The
+/// store gets structured records (severity set); the reference gets the
+/// raw lines of the **same in-window files** (hour 10 ERROR + hour 11
+/// INFO) — the out-of-window hour-20 file is not part of the reference,
+/// just as `files_in_range.zst` wouldn't name it.
+fn build(bucket: &Path) -> ReferenceCorpus {
+    // hour 10 — the in-window ERROR file (the B1 result).
+    let errors: Vec<MinedRecord> = (0..ERROR_ROWS)
+        .map(|i| rec(1, TS0 + i * 1_000, "ERROR"))
+        .collect();
+    write_one_file(bucket, &errors);
+
+    // hour 11 — in-window INFO file (pruned by the severity predicate).
+    let infos: Vec<MinedRecord> = (0..INFO_ROWS)
+        .map(|i| rec(2, TS0 + HOUR_NS + i * 1_000, "INFO"))
+        .collect();
+    write_one_file(bucket, &infos);
+
+    // hour 20 — out-of-window ERROR filler (pruned by the time bound).
+    let filler: Vec<MinedRecord> = (0..OUT_OF_WINDOW_ROWS)
+        .map(|i| rec(3, TS0 + 10 * HOUR_NS + i * 1_000, "ERROR"))
+        .collect();
+    write_one_file(bucket, &filler);
+
+    // Reference: the raw lines of the two IN-WINDOW files only. The grep
+    // token "ERROR" matches the error file's lines, not the info file's.
+    let error_lines: Vec<String> = (0..ERROR_ROWS)
+        .map(|i| format!("ERROR request failed id={i}"))
+        .collect();
+    let info_lines: Vec<String> = (0..INFO_ROWS)
+        .map(|i| format!("INFO request ok id={i}"))
+        .collect();
+    ReferenceCorpus::compress(&[error_lines, info_lines], ZSTD_LEVEL).expect("compress reference")
+}
+
+/// The B1 query: tenant "a", the half-open window covering hours 10–11,
+/// `level='ERROR'`.
+fn b1_query() -> QueryRequest {
+    QueryRequest {
+        tenant: TenantId::new("a"),
+        time_range: Some((TS0, TS0 + 2 * HOUR_NS)),
+        template_id: None,
+        severity_text: Some("ERROR".to_string()),
+    }
+}
+
+fn synthetic(c: &mut Criterion) {
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .build()
+        .expect("tokio runtime");
+
+    let bucket = tempfile::TempDir::new().expect("temp bucket");
+    let reference = build(bucket.path());
+    let querier = Querier::new(bucket.path());
+
+    // Sanity + visibility: both sides return the same B1 result, and the
+    // Ourios scan prunes (the comparison is meaningless otherwise).
+    let probe = rt.block_on(querier.run(b1_query())).expect("probe query");
+    let ref_count = reference
+        .count_lines_containing("ERROR")
+        .expect("reference");
+    assert_eq!(probe.rows, ERROR_ROWS, "Ourios B1 result");
+    assert_eq!(ref_count, ERROR_ROWS, "reference B1 result matches Ourios");
+    // The bench premise: the Ourios query *prunes* (the INFO + the
+    // out-of-window files), so it isn't silently doing a full scan that
+    // would make the latency comparison meaningless.
+    assert!(
+        probe.stats.row_groups_pruned >= 1,
+        "Ourios B1 query must prune row groups; stats={:?}",
+        probe.stats,
+    );
+    let total_rg = probe.stats.row_groups_scanned + probe.stats.row_groups_pruned;
+    eprintln!(
+        "b1/synthetic: result={} rows; ourios pruned {}/{} row groups, read {} B; \
+         reference scans {} in-window compressed B",
+        probe.rows,
+        probe.stats.row_groups_pruned,
+        total_rg,
+        probe.stats.bytes_read,
+        reference.compressed_bytes(),
+    );
+
+    let mut group = c.benchmark_group("b1/synthetic");
+    group.bench_function("ourios", |b| {
+        b.iter(|| {
+            let r = rt.block_on(querier.run(b1_query())).expect("query");
+            black_box(r.rows);
+        });
+    });
+    group.bench_function("zstd-grep-reference", |b| {
+        b.iter(|| {
+            let n = reference
+                .count_lines_containing("ERROR")
+                .expect("reference");
+            black_box(n);
+        });
+    });
+    group.finish();
+}
+
+criterion_group!(benches, synthetic);
+criterion_main!(benches);

crates/ourios-bench/src/lib.rs

@@ -35,9 +35,11 @@ mod c1;
 mod c2;
 mod corpus;
 mod harness;
+mod reference;
 mod report;
 mod store;
 
+pub use reference::ReferenceCorpus;
 pub use report::{update_status_section, write_results_json};
 pub use store::{BuiltStore, build_query_store};
 

crates/ourios-bench/src/reference.rs

@@ -0,0 +1,170 @@
+//! In-process B1 baseline — the `zstdcat files_in_range.zst | grep TOKEN
+//! | wc -l` reference (`docs/benchmarks.md` §3 B1), without shelling out
+//! to a system `zstd`/`grep` (so the bench is reproducible on any host;
+//! same bundled `zstd` the A1 reference codec uses).
+//!
+//! **Fairness.** The caller passes only the raw lines from the files that
+//! fall in the query's time window — the file-level pruning the baseline
+//! pipeline gets "for free" by naming `files_in_range.zst`. Within those
+//! files the reference scans **every** line (decode + substring match),
+//! whereas the Ourios query additionally skips row groups via column
+//! statistics. So the measured ratio reflects *within-window* pruning,
+//! not a strawman full-corpus scan against a no-pruning baseline.
+
+/// A B1 reference corpus: the in-window raw log lines, `zstd`-compressed
+/// one block per file (mirroring stored `*.zst` segments the baseline
+/// `zstdcat`s).
+pub struct ReferenceCorpus {
+    blocks: Vec<Vec<u8>>,
+}
+
+impl ReferenceCorpus {
+    /// Compress each in-window file's raw lines (`files[i]` is one file's
+    /// lines) at `level`. Only in-window files should be passed (see the
+    /// module's fairness note).
+    ///
+    /// # Errors
+    ///
+    /// Propagates a `zstd` compression I/O error (not expected for an
+    /// in-memory buffer, but the encoder's signature is fallible).
+    pub fn compress(files: &[Vec<String>], level: i32) -> std::io::Result<Self> {
+        use std::io::Write;
+
+        let mut blocks = Vec::with_capacity(files.len());
+        for lines in files {
+            // Stream each line + `\n` straight into the encoder rather
+            // than materialising the whole file as one `String` first:
+            // setup stays memory-bounded at GiB-window scale and mirrors
+            // how a real `*.zst` segment is produced. Newline-*terminate*
+            // every line (as a real log file is), so the decode below and
+            // a `zstdcat | wc -l` pipeline agree with no off-by-one.
+            let mut encoder = zstd::stream::write::Encoder::new(Vec::new(), level)?;
+            for line in lines {
+                encoder.write_all(line.as_bytes())?;
+                encoder.write_all(b"\n")?;
+            }
+            blocks.push(encoder.finish()?);
+        }
+        Ok(Self { blocks })
+    }
+
+    /// `zstdcat <blocks> | grep -F token | wc -l`: stream-decode every
+    /// block and count the lines containing `token`. Matches at the
+    /// **byte** level (like `grep -F`) over a reused line buffer — no
+    /// UTF-8 decode and no per-line allocation — so the baseline isn't
+    /// artificially slowed (which would bias the B1 ratio in Ourios's
+    /// favour). Streaming keeps memory bounded at GiB corpus scale. This
+    /// is the timed reference work the B1 bench compares Ourios against.
+    ///
+    /// # Errors
+    ///
+    /// Propagates a `zstd` decompression / read error if a block is
+    /// corrupt or truncated.
+    pub fn count_lines_containing(&self, token: &str) -> std::io::Result<u64> {
+        use std::io::BufRead;
+
+        let needle = token.as_bytes();
+        let mut matches = 0u64;
+        // One line buffer reused across every block (and every line).
+        let mut line = Vec::new();
+        for block in &self.blocks {
+            let decoder = zstd::stream::read::Decoder::new(block.as_slice())?;
+            let mut reader = std::io::BufReader::new(decoder);
+            loop {
+                line.clear();
+                if reader.read_until(b'\n', &mut line)? == 0 {
+                    break;
+                }
+                // Match the line *content*, like `grep` — without the
+                // trailing `\n` (and a `\r` for CRLF input) the delimiter
+                // would otherwise leave in the buffer.
+                let mut content = line.as_slice();
+                if let [rest @ .., b'\n'] = content {
+                    content = rest;
+                }
+                if let [rest @ .., b'\r'] = content {
+                    content = rest;
+                }
+                if contains_subslice(content, needle) {
+                    matches = matches.saturating_add(1);
+                }
+            }
+        }
+        Ok(matches)
+    }
+
+    /// Total compressed size across the in-window blocks — the `*.zst`
+    /// size a `zstdcat` baseline would read. (Held in memory here, so
+    /// this measures size, not disk I/O.)
+    ///
+    /// # Panics
+    ///
+    /// Panics only if a single block's length exceeds `u64` (`usize >
+    /// u64`), which is unreachable on any supported target.
+    #[must_use]
+    pub fn compressed_bytes(&self) -> u64 {
+        self.blocks.iter().fold(0_u64, |acc, b| {
+            acc.saturating_add(
+                u64::try_from(b.len()).expect("usize fits in u64 on every supported Rust target"),
+            )
+        })
+    }
+}
+
+/// Whether `haystack` contains `needle` as a byte substring — the
+/// `grep -F` fixed-string match, without UTF-8 decoding. Anchors on the
+/// needle's first byte and only compares the remainder on a hit, so it
+/// avoids the redundant full-window comparison `windows().any()` would
+/// do at every position (keeps the baseline close to `grep -F`'s
+/// optimized search rather than artificially slow).
+fn contains_subslice(haystack: &[u8], needle: &[u8]) -> bool {
+    let Some((&first, rest)) = needle.split_first() else {
+        return true; // an empty needle matches anywhere
+    };
+    if needle.len() > haystack.len() {
+        return false;
+    }
+    let last_start = haystack.len() - needle.len();
+    // `.get()` (not indexing) keeps this panic-free by construction.
+    (0..=last_start).any(|i| {
+        haystack.get(i) == Some(&first) && haystack.get(i + 1..i + needle.len()) == Some(rest)
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn counts_only_in_window_lines_containing_the_token() {
+        // Arrange — two in-window "files": one with 3 ERROR lines + 1
+        // INFO, one all INFO. (Out-of-window files are simply not passed.)
+        let files = vec![
+            vec![
+                "ERROR boom a".to_string(),
+                "INFO ok".to_string(),
+                "ERROR boom b".to_string(),
+                "ERROR boom c".to_string(),
+            ],
+            vec!["INFO ok".to_string(), "INFO also ok".to_string()],
+        ];
+
+        // Act
+        let reference = ReferenceCorpus::compress(&files, 19).expect("compress");
+        let n = reference.count_lines_containing("ERROR").expect("count");
+
+        // Assert — exactly the three ERROR lines, across both files.
+        assert_eq!(n, 3);
+        assert!(reference.compressed_bytes() > 0, "blocks hold bytes");
+    }
+
+    #[test]
+    fn empty_corpus_counts_zero() {
+        // Arrange / Act
+        let reference = ReferenceCorpus::compress(&[], 19).expect("compress");
+
+        // Assert
+        assert_eq!(reference.count_lines_containing("ERROR").expect("count"), 0);
+        assert_eq!(reference.compressed_bytes(), 0);
+    }
+}