diff --git a/Cargo.lock b/Cargo.lock index 6ca3a2c1..6ede6bc0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -61,6 +61,12 @@ dependencies = [ "libc", ] +[[package]] +name = "anes" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299" + [[package]] name = "anstream" version = "1.0.0" @@ -544,6 +550,12 @@ version = "1.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" +[[package]] +name = "cast" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5" + [[package]] name = "cc" version = "1.2.62" @@ -583,6 +595,33 @@ dependencies = [ "phf", ] +[[package]] +name = "ciborium" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42e69ffd6f0917f5c029256a24d0161db17cea3997d185db0d35926308770f0e" +dependencies = [ + "ciborium-io", + "ciborium-ll", + "serde", +] + +[[package]] +name = "ciborium-io" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05afea1e0a06c9be33d539b876f1ce3692f4afea2cb41f740e7743225ed1c757" + +[[package]] +name = "ciborium-ll" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57663b653d948a338bfb3eeba9bb2fd5fcfaecb9e199e87e1eda4d9e8b240fd9" +dependencies = [ + "ciborium-io", + "half", +] + [[package]] name = "clap" version = "4.6.1" @@ -710,6 +749,61 @@ dependencies = [ "rustc_version", ] +[[package]] +name = "criterion" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2b12d017a929603d80db1831cd3a24082f8137ce19c69e6447f54f5fc8d692f" +dependencies = [ + "anes", + "cast", + "ciborium", + "clap", + "criterion-plot", + "is-terminal", + "itertools 0.10.5", + "num-traits", + "once_cell", + "oorandom", + "plotters", + "rayon", + "regex", + "serde", + "serde_derive", + "serde_json", + "tinytemplate", + "walkdir", +] + +[[package]] +name = "criterion-plot" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b50826342786a51a89e2da3a28f1c32b06e387201bc2d19791f622c673706b1" +dependencies = [ + "cast", + "itertools 0.10.5", +] + +[[package]] +name = "crossbeam-deque" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9dd111b7b7f7d55b72c0a6ae361660ee5853c9af73f70c3c2ef6858b950e2e51" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-epoch" +version = "0.9.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-utils" version = "0.8.21" @@ -802,7 +896,7 @@ dependencies = [ "datafusion-physical-plan", "datafusion-session", "futures", - "itertools", + "itertools 0.14.0", "log", "object_store", "parking_lot", @@ -833,7 +927,7 @@ dependencies = [ "datafusion-physical-plan", "datafusion-session", "futures", - "itertools", + "itertools 0.14.0", "log", "object_store", "parking_lot", @@ -858,7 +952,7 @@ dependencies = [ "datafusion-physical-expr-common", "datafusion-physical-plan", "futures", - "itertools", + "itertools 0.14.0", "log", "object_store", ] @@ -876,7 +970,7 @@ dependencies = [ "half", "hashbrown 0.16.1", "indexmap", - "itertools", + "itertools 0.14.0", "libc", "log", "object_store", @@ -918,7 +1012,7 @@ dependencies = [ "datafusion-session", "futures", "glob", - "itertools", + "itertools 0.14.0", "log", "object_store", "rand", @@ -945,7 +1039,7 @@ dependencies = [ "datafusion-physical-plan", "datafusion-session", "futures", - "itertools", + "itertools 0.14.0", "object_store", "tokio", ] @@ -1019,7 +1113,7 @@ dependencies = [ "datafusion-pruning", "datafusion-session", "futures", - "itertools", + "itertools 0.14.0", "log", "object_store", "parking_lot", @@ -1072,7 +1166,7 @@ dependencies = [ "datafusion-functions-window-common", "datafusion-physical-expr-common", "indexmap", - "itertools", + "itertools 0.14.0", "paste", "serde_json", ] @@ -1086,7 +1180,7 @@ dependencies = [ "arrow", "datafusion-common", "indexmap", - "itertools", + "itertools 0.14.0", "paste", ] @@ -1108,7 +1202,7 @@ dependencies = [ "datafusion-expr-common", "datafusion-macros", "hex", - "itertools", + "itertools 0.14.0", "log", "memchr", "num-traits", @@ -1221,7 +1315,7 @@ dependencies = [ "datafusion-expr-common", "datafusion-physical-expr", "indexmap", - "itertools", + "itertools 0.14.0", "log", "regex", "regex-syntax", @@ -1243,7 +1337,7 @@ dependencies = [ "half", "hashbrown 0.16.1", "indexmap", - "itertools", + "itertools 0.14.0", "parking_lot", "paste", "petgraph", @@ -1262,7 +1356,7 @@ dependencies = [ "datafusion-functions", "datafusion-physical-expr", "datafusion-physical-expr-common", - "itertools", + "itertools 0.14.0", ] [[package]] @@ -1278,7 +1372,7 @@ dependencies = [ "datafusion-expr-common", "hashbrown 0.16.1", "indexmap", - "itertools", + "itertools 0.14.0", "parking_lot", ] @@ -1297,7 +1391,7 @@ dependencies = [ "datafusion-physical-expr-common", "datafusion-physical-plan", "datafusion-pruning", - "itertools", + "itertools 0.14.0", ] [[package]] @@ -1324,7 +1418,7 @@ dependencies = [ "half", "hashbrown 0.16.1", "indexmap", - "itertools", + "itertools 0.14.0", "log", "num-traits", "parking_lot", @@ -1345,7 +1439,7 @@ dependencies = [ "datafusion-physical-expr", "datafusion-physical-expr-common", "datafusion-physical-plan", - "itertools", + "itertools 0.14.0", "log", ] @@ -1655,6 +1749,12 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" +[[package]] +name = "hermit-abi" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" + [[package]] name = "hex" version = "0.4.3" @@ -1827,12 +1927,32 @@ version = "3.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8bb03732005da905c88227371639bf1ad885cc712789c011c31c5fb3ab3ccf02" +[[package]] +name = "is-terminal" +version = "0.4.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" +dependencies = [ + "hermit-abi", + "libc", + "windows-sys", +] + [[package]] name = "is_terminal_polyfill" version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" +[[package]] +name = "itertools" +version = "0.10.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0fd2260e829bddf4cb6ea802289de2f86d6a7a690192fbe91b3f46e0f2c8473" +dependencies = [ + "either", +] + [[package]] name = "itertools" version = "0.14.0" @@ -2085,7 +2205,7 @@ dependencies = [ "futures-util", "http", "humantime", - "itertools", + "itertools 0.14.0", "parking_lot", "percent-encoding", "thiserror", @@ -2109,6 +2229,12 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" +[[package]] +name = "oorandom" +version = "11.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e" + [[package]] name = "opentelemetry" version = "0.32.0" @@ -2160,13 +2286,16 @@ name = "ourios-bench" version = "0.0.0" dependencies = [ "clap", + "criterion", "opentelemetry-proto", "ourios-core", "ourios-miner", "ourios-parquet", + "ourios-querier", "serde", "serde_json", "tempfile", + "tokio", "zstd", ] @@ -2367,6 +2496,34 @@ version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e" +[[package]] +name = "plotters" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5aeb6f403d7a4911efb1e33402027fc44f29b5bf6def3effcc22d7bb75f2b747" +dependencies = [ + "num-traits", + "plotters-backend", + "plotters-svg", + "wasm-bindgen", + "web-sys", +] + +[[package]] +name = "plotters-backend" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df42e13c12958a16b3f7f4386b9ab1f3e7933914ecea48da7139435263a4172a" + +[[package]] +name = "plotters-svg" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "51bae2ac328883f7acdfea3d66a7c35751187f870bc81f94563733a154d7a670" +dependencies = [ + "plotters-backend", +] + [[package]] name = "portable-atomic" version = "1.13.1" @@ -2442,7 +2599,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "27c6023962132f4b30eb4c172c91ce92d933da334c59c23cddee82358ddafb0b" dependencies = [ "anyhow", - "itertools", + "itertools 0.14.0", "proc-macro2", "quote", "syn", @@ -2507,6 +2664,26 @@ dependencies = [ "rand_core", ] +[[package]] +name = "rayon" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb39b166781f92d482534ef4b4b1b2568f42613b53e5b6c160e24cfbfa30926d" +dependencies = [ + "either", + "rayon-core", +] + +[[package]] +name = "rayon-core" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91" +dependencies = [ + "crossbeam-deque", + "crossbeam-utils", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -2799,6 +2976,16 @@ dependencies = [ "zerovec", ] +[[package]] +name = "tinytemplate" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "be4d6b5f19ff7664e8c98d03e2139cb510db9b0a60b55f8e8709b689d939b6bc" +dependencies = [ + "serde", + "serde_json", +] + [[package]] name = "tokio" version = "1.52.3" @@ -3083,6 +3270,16 @@ dependencies = [ "semver", ] +[[package]] +name = "web-sys" +version = "0.3.98" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b572dff8bcf38bad0fa19729c89bb5748b2b9b1d8be70cf90df697e3a8f32aa" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "web-time" version = "1.1.0" diff --git a/crates/ourios-bench/Cargo.toml b/crates/ourios-bench/Cargo.toml index f9bd2fc2..2bc007a2 100644 --- a/crates/ourios-bench/Cargo.toml +++ b/crates/ourios-bench/Cargo.toml @@ -57,5 +57,20 @@ zstd = "0.13" # `tests/`, so no separate `[dev-dependencies]` mirror is needed. tempfile = "3" +[dev-dependencies] +# B2 latency bench (RFC 0007 §6 — supportive, non-gating evidence +# for the RFC0007.2 thesis gate that ourios-querier proves +# structurally). criterion gives warmup + a statistical median, +# which is what B2's "median latency" target wants; ourios-querier +# is the unit under test (and pulls DataFusion); a current-thread +# tokio runtime drives its async `Querier::run`. +criterion = "0.5" +ourios-querier = { path = "../ourios-querier" } +tokio = { version = "1", default-features = false, features = ["rt"] } + +[[bench]] +name = "b2" +harness = false + [lints] workspace = true diff --git a/crates/ourios-bench/benches/b2.rs b/crates/ourios-bench/benches/b2.rs new file mode 100644 index 00000000..b61947d6 --- /dev/null +++ b/crates/ourios-bench/benches/b2.rs @@ -0,0 +1,208 @@ +//! B2 — template-exact query latency vs corpus size (RFC 0007 §6). +//! +//! Supportive, **non-gating** wall-clock evidence for the RFC0007.2 +//! thesis gate that `ourios-querier` already proves *structurally* +//! (deterministically, via `QueryStats`) in its test suite. Here we +//! measure the thing the structural test can't: actual latency. +//! +//! Two groups: +//! +//! - `b2/synthetic` (always runs) — the controlled instrument. The +//! *result* size is held constant (`TARGET_ROWS` of one template) +//! while the *corpus* grows 1×/10×/50× with filler under other +//! templates in other hours (own files ⇒ own row groups, pruned +//! by `template_id` statistics). If the inverted-index-collapse +//! thesis holds, latency stays ~flat as the corpus grows. +//! - `b2/otel-demo` (only when `OURIOS_B2_CORPUS_DIRS` is set) — the +//! real corpora (`corpus/otel-demo-v*`), one bench id per dir. See +//! the env var's use below. + +use std::hint::black_box; +use std::path::Path; + +use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main}; + +use ourios_bench::build_query_store; +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}; + +/// Anchored in 2026 so the derived partition path is stable. +const TS0: u64 = 1_775_127_480_000_000_000; +const HOUR_NS: u64 = 3_600_000_000_000; + +/// Rows under the queried template — the result size, held constant +/// across corpus sizes so latency reflects result, not corpus. +const TARGET_ROWS: u64 = 2_000; +/// Filler rows per hour-file (each a distinct non-target template). +const FILLER_PER_FILE: u64 = 4_000; + +fn rec(template_id: u64, ts_ns: u64) -> MinedRecord { + MinedRecord { + tenant_id: TenantId::new("a"), + template_id, + template_version: 1, + severity_number: 9, + severity_text: Some("INFO".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` that all share one partition into 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 a store with `TARGET_ROWS` rows of template 1 (hour 0) plus +/// `filler_rows` rows of distinct non-target templates spread across +/// later hours, so a `template_id = 1` query returns a fixed result +/// while total corpus size is `TARGET_ROWS + filler_rows`. +fn build_synthetic(bucket: &Path, filler_rows: u64) { + let target: Vec = (0..TARGET_ROWS).map(|i| rec(1, TS0 + i * 1_000)).collect(); + write_one_file(bucket, &target); + + let mut written = 0u64; + let mut hour = 1u64; + while written < filler_rows { + let n = (filler_rows - written).min(FILLER_PER_FILE); + let base = TS0 + hour * HOUR_NS; + // A distinct template per file ⇒ its row group is pruned by + // a `template_id = 1` query. + let template = 1_000 + hour; + let file: Vec = (0..n).map(|i| rec(template, base + i * 1_000)).collect(); + write_one_file(bucket, &file); + written += n; + hour += 1; + } +} + +fn template_exact(template_id: u64) -> QueryRequest { + QueryRequest { + tenant: TenantId::new("a"), + time_range: None, + template_id: Some(template_id), + } +} + +fn synthetic(c: &mut Criterion) { + let rt = tokio::runtime::Builder::new_current_thread() + .build() + .expect("tokio runtime"); + + let mut group = c.benchmark_group("b2/synthetic"); + for mult in [1u64, 10, 50] { + let filler = TARGET_ROWS * (mult - 1); + let total = TARGET_ROWS + filler; + + let bucket = tempfile::TempDir::new().expect("temp bucket"); + build_synthetic(bucket.path(), filler); + let querier = Querier::new(bucket.path()); + + // Sanity: the result is the fixed TARGET_ROWS regardless of + // corpus — otherwise the latency comparison is meaningless. + let probe = rt + .block_on(querier.run(template_exact(1))) + .expect("probe query"); + assert_eq!( + probe.rows, TARGET_ROWS, + "result size must be constant across corpus sizes", + ); + + group.bench_with_input(BenchmarkId::from_parameter(total), &total, |b, _| { + b.iter(|| { + let r = rt.block_on(querier.run(template_exact(1))).expect("query"); + black_box(r.rows); + }); + }); + } + group.finish(); +} + +/// Real corpora, one bench id per directory in `OURIOS_B2_CORPUS_DIRS` +/// (comma-separated). Each dir is loaded → mined → written to a temp +/// Parquet store, then the busiest template is queried. Skipped +/// entirely when the env var is unset (the corpora aren't committed; +/// CI / a local operator stages them and points this at them, e.g. +/// `OURIOS_B2_CORPUS_DIRS=corpus/otel-demo-v1,corpus/otel-demo-v4`). +fn otel_demo(c: &mut Criterion) { + let Ok(raw) = std::env::var("OURIOS_B2_CORPUS_DIRS") else { + eprintln!( + "b2/otel-demo: OURIOS_B2_CORPUS_DIRS unset — skipping (synthetic group still runs)" + ); + return; + }; + let dirs: Vec<&str> = raw + .split(',') + .map(str::trim) + .filter(|s| !s.is_empty()) + .collect(); + if dirs.is_empty() { + return; + } + + let rt = tokio::runtime::Builder::new_current_thread() + .build() + .expect("tokio runtime"); + + let mut group = c.benchmark_group("b2/otel-demo"); + for dir in dirs { + let path = Path::new(dir); + if !path.is_dir() { + eprintln!("b2/otel-demo: {dir} is not a directory — skipping"); + continue; + } + let bucket = tempfile::TempDir::new().expect("temp bucket"); + let built = build_query_store(path, bucket.path()).expect("build query store"); + if built.busiest_template_rows == 0 { + eprintln!("b2/otel-demo: {dir} produced no rows — skipping"); + continue; + } + let querier = Querier::new(bucket.path()); + eprintln!( + "b2/otel-demo: {dir} — {} rows, {} files; querying template {} ({} rows)", + built.rows, built.files, built.busiest_template_id, built.busiest_template_rows, + ); + + let id = path + .file_name() + .and_then(|s| s.to_str()) + .unwrap_or(dir) + .to_string(); + group.bench_function(BenchmarkId::new("corpus", id), |b| { + b.iter(|| { + let r = rt + .block_on(querier.run(template_exact(built.busiest_template_id))) + .expect("query"); + black_box(r.rows); + }); + }); + } + group.finish(); +} + +criterion_group!(benches, synthetic, otel_demo); +criterion_main!(benches); diff --git a/crates/ourios-bench/src/lib.rs b/crates/ourios-bench/src/lib.rs index 304e23fe..7f5a0687 100644 --- a/crates/ourios-bench/src/lib.rs +++ b/crates/ourios-bench/src/lib.rs @@ -36,8 +36,10 @@ mod c2; mod corpus; mod harness; mod report; +mod store; pub use report::{update_status_section, write_results_json}; +pub use store::{BuiltStore, build_query_store}; /// Configuration for one bench invocation. /// @@ -297,6 +299,29 @@ fn resolve_bucket( /// `bytes(ourios_output)`. Missing subtrees are fine (nothing /// to collide with). fn ensure_bucket_has_no_parquet(bucket_root: &std::path::Path) -> Result<(), BenchError> { + if let Some(path) = find_published_parquet(bucket_root)? { + return Err(BenchError::Cli { + detail: format!( + "bucket {} already contains a Parquet file ({}); A1 would \ + count it in bytes(ourios_output). Point --bucket-dir at an \ + empty directory or omit it to use a fresh scratch dir.", + bucket_root.display(), + path.display(), + ), + }); + } + Ok(()) +} + +/// First committed `*.parquet` under the bucket's `data/` or +/// `audit/` subtree, or `None` if the bucket holds no Parquet. +/// Shared by [`ensure_bucket_has_no_parquet`] (the A1 path) and +/// [`store::build_query_store`] (the B2 path), which each wrap it +/// in a context-appropriate error — a reused bucket skews A1's +/// byte count and mixes corpora into the B2 query alike. +pub(crate) fn find_published_parquet( + bucket_root: &std::path::Path, +) -> Result, BenchError> { for sub in ["data", "audit"] { let dir = bucket_root.join(sub); let mut stack = vec![dir]; @@ -338,20 +363,12 @@ fn ensure_bucket_has_no_parquet(bucket_root: &std::path::Path) -> Result<(), Ben .extension() .is_some_and(|e| e.eq_ignore_ascii_case("parquet")) { - return Err(BenchError::Cli { - detail: format!( - "bucket {} already contains a Parquet file ({}); A1 would \ - count it in bytes(ourios_output). Point --bucket-dir at an \ - empty directory or omit it to use a fresh scratch dir.", - bucket_root.display(), - path.display(), - ), - }); + return Ok(Some(path)); } } } } - Ok(()) + Ok(None) } /// Format `SystemTime::now()` as a §3.6 millisecond-precision diff --git a/crates/ourios-bench/src/store.rs b/crates/ourios-bench/src/store.rs new file mode 100644 index 00000000..30044dec --- /dev/null +++ b/crates/ourios-bench/src/store.rs @@ -0,0 +1,165 @@ +//! Build a queryable RFC 0005 Parquet store from a corpus. +//! +//! The A1 path measures the bytes a corpus compresses to; the B2 +//! latency bench needs to *query* the same corpus, so it needs the +//! mined records laid down as a real partitioned Parquet store it +//! can point a [`ourios_querier::Querier`] at. This is the only +//! public store-builder the bench exposes — it reuses the same +//! corpus loader and miner harness the gates run on (so the store +//! matches what A1 measured), then writes every emitted record via +//! per-partition [`Writer`]s (the same streaming write A1 uses). + +use std::collections::HashMap; +use std::path::Path; + +use ourios_core::record::MinedRecord; +use ourios_parquet::{PartitionKey, Writer}; + +use crate::{BenchError, corpus, harness}; + +/// What [`build_query_store`] wrote, enough for the B2 bench to pick +/// a populated query and report the result-vs-corpus relationship. +#[derive(Debug, Clone, Copy)] +pub struct BuiltStore { + /// Total rows written across all partitions. + pub rows: u64, + /// Number of partition files written (one per `*.parquet`). + pub files: u64, + /// The `template_id` with the most rows — a query for it is + /// guaranteed to return a non-empty, representative result. + pub busiest_template_id: u64, + /// How many rows that busiest template has (the result size a + /// `template_id = busiest_template_id` query returns). + pub busiest_template_rows: u64, +} + +/// Load the corpus at `corpus_dir`, mine it, and write the emitted +/// records as a partitioned RFC 0005 Parquet store under +/// `bucket_root` (which a [`ourios_querier::Querier`] can then be +/// rooted at). Returns a [`BuiltStore`] summary. +/// +/// # Errors +/// +/// - [`BenchError::Corpus`] if the corpus can't be loaded. +/// - [`BenchError::Pipeline`] if partition derivation or a Parquet +/// write/close fails. +/// +/// # Panics +/// +/// Panics if the partition count exceeds `u64` (`usize > u64`), +/// which can't happen on any supported target — same documented +/// assumption as [`crate::run`]. +pub fn build_query_store(corpus_dir: &Path, bucket_root: &Path) -> Result { + // A reused bucket would let the querier enumerate a prior run's + // Parquet too, mixing corpora and skewing both the row counts + // and the latency measurement. Reject up front (the A1 path + // guards the same way via `ensure_bucket_has_no_parquet`). + if let Some(existing) = crate::find_published_parquet(bucket_root)? { + return Err(BenchError::Pipeline { + detail: format!( + "bucket {} already contains a Parquet file ({}); build_query_store \ + needs an empty bucket so the querier doesn't mix corpora", + bucket_root.display(), + existing.display(), + ), + }); + } + + let load = corpus::load(corpus_dir)?; + + let mut writers: HashMap = HashMap::new(); + let mut counts: HashMap = HashMap::new(); + let mut rows: u64 = 0; + // The harness callback returns `()`, so a write error is stashed + // (first wins) and surfaced after the run — the same pattern + // `a1::A1Accumulator` uses. + let mut first_err: Option = None; + + harness::run(&load, false, |_input, emitted, _snap| { + if first_err.is_some() { + return; + } + match append_record(&mut writers, bucket_root, emitted) { + Ok(()) => { + rows += 1; + *counts.entry(emitted.template_id).or_insert(0) += 1; + } + Err(e) => first_err = Some(e), + } + })?; + + if let Some(e) = first_err { + return Err(e); + } + + let files = u64::try_from(writers.len()).expect("usize fits in u64 on every supported target"); + for (_partition, writer) in writers { + writer.close().map_err(|e| BenchError::Pipeline { + detail: format!("parquet close: {e}"), + })?; + } + + let (busiest_template_id, busiest_template_rows) = + counts.into_iter().max_by_key(|&(_, n)| n).unwrap_or((0, 0)); + + Ok(BuiltStore { + rows, + files, + busiest_template_id, + busiest_template_rows, + }) +} + +/// Append one record into its partition's writer, opening one on the +/// first record for a partition (mirrors `a1::A1Accumulator`). +fn append_record( + writers: &mut HashMap, + bucket_root: &Path, + emitted: &MinedRecord, +) -> Result<(), BenchError> { + let partition = PartitionKey::derive(emitted).map_err(|e| BenchError::Pipeline { + detail: format!("partition derive failed: {e}"), + })?; + if let Some(writer) = writers.get_mut(&partition) { + return writer + .append_records(std::slice::from_ref(emitted)) + .map_err(|e| BenchError::Pipeline { + detail: format!("parquet append_records: {e}"), + }); + } + let mut writer = + Writer::open(bucket_root, partition.clone()).map_err(|e| BenchError::Pipeline { + detail: format!("parquet open: {e}"), + })?; + writer + .append_records(std::slice::from_ref(emitted)) + .map_err(|e| BenchError::Pipeline { + detail: format!("parquet append_records: {e}"), + })?; + writers.insert(partition, writer); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A bucket that already holds a published `*.parquet` is + /// rejected, so a reused dir can't silently mix corpora into the + /// B2 query (regression guard for the second-build case). + #[test] + fn rejects_a_non_empty_bucket() { + let corpus = tempfile::TempDir::new().expect("corpus dir"); + std::fs::write(corpus.path().join("c.txt"), b"user 42 logged in\n").expect("write corpus"); + let bucket = tempfile::TempDir::new().expect("bucket dir"); + + let first = build_query_store(corpus.path(), bucket.path()).expect("first build"); + assert!(first.rows >= 1, "the one corpus line is written"); + + let second = build_query_store(corpus.path(), bucket.path()); + assert!( + matches!(second, Err(BenchError::Pipeline { .. })), + "a reused, non-empty bucket must be rejected, got {second:?}", + ); + } +} diff --git a/docs/benchmarks.md b/docs/benchmarks.md index 43c79d64..86dfc5c3 100644 --- a/docs/benchmarks.md +++ b/docs/benchmarks.md @@ -124,6 +124,17 @@ exploit structure the tree extracted. corpus size, template mining is buying compression but not query locality — the inverted-index collapse thesis (`CLAUDE.md` §2) is wrong in practice. Open an RFC. +- **Instruments**: B2 is proven *structurally* (deterministically) + by `ourios-querier`'s `rfc0007_2_*` test — for a fixed result the + scanned row groups + bytes stay flat as the corpus grows. The + `criterion` bench `crates/ourios-bench/benches/b2.rs` adds the + *wall-clock* view: a `b2/synthetic` group (result held constant, + corpus scaled 1×/10×/50×) and a `b2/otel-demo` group over real + corpora (set `OURIOS_B2_CORPUS_DIRS` to a comma-separated list of + corpus dirs; skipped when unset, since the corpora aren't + committed). Run with `cargo bench -p ourios-bench --bench b2`. + These laptop/CI latencies are **indicative**, not §1-baseline + canonical, so they are not recorded in §9. ### B3 — Substring queries (the hard case) @@ -307,8 +318,11 @@ hardware baseline (`baseline-8vcpu-32gib`), against an OTel-Demo corpus that is **shape-representative** (real multi-service template + envelope diversity) but **not size-representative** — every corpus is well below §8's ≥ 1 GiB canonical minimum, so -this run is intentionally diagnostic, not a thesis verdict. B1 / -B2 (query-side) remain unmeasured pending the querier (RFC 0007). +this run is intentionally diagnostic, not a thesis verdict. The +query-side gates now have instruments — B1 and B2 are proven +structurally in `ourios-querier`, and B2 has a `criterion` latency +bench (§B2 "Instruments") — but canonical query numbers on the §1 +hardware against a ≥ 1 GiB corpus are still pending. Reviewers: a PR that materially affects the hot path must either (a) cite the benchmark result and its delta against the relevant