diff --git a/.claude/board/STATUS_BOARD.md b/.claude/board/STATUS_BOARD.md index b7371f1df..4a12986ba 100644 --- a/.claude/board/STATUS_BOARD.md +++ b/.claude/board/STATUS_BOARD.md @@ -5,7 +5,7 @@ Plan: `.claude/plans/d-relalg-unified-v1.md`. Decisions recorded there: ADMISSIO | D-id | scope | status | gate / falsifier | |---|---|---|---| | **D-RU-0** | decisions: ADMISSION = B, D-RA-2 = A | DECIDED (operator, 2026-10-20) | — | -| **D-RU-1** | U1: JC statistical batch, distinct (selection, key, lane pair) triples over shared power-sum terminals | Queued (authorized; first) | rows traversed and terminal passes below the per-request baseline; results bit-exact on integer lanes; overflow refused, not wrapped; R references | +| **D-RU-1** | U1: JC statistical batch, distinct (selection, key, lane pair) triples over shared power-sum terminals | MEASURED (`crates/jc-batch-probe`; `entries/2026-10-20-u1-jc-batch.md`): 252 requests → 2 selection + 42 cross passes; at 1M rows 6.4× faster, 11.5× fewer rows traversed, 8.5× fewer values read, 126× fewer mask bytes; bit-identical to per-request; no R reference (R absent) | rows traversed and terminal passes below the per-request baseline; results bit-exact on integer lanes; overflow refused, not wrapped; R references | | **D-RU-2** | U2: `let`-bound planes on the contract IR, immutable `Plane(i)`, versioned coordinates, no DAG inside a fold | Queued (after D-CMR-1) | rows traversed and planes materialized against recompute; width test ≤ 3 slots; answers and versions unchanged | | **D-RU-3** | cross-rule CSE in crepe-plan-probe (256-rule benchmark, separate from the hop benchmark) | Queued (after D-RU-2) | answers == crepe and ascent; traversals and materializations measured | | **D-RU-4** | co-addressed join operator over `JoinDomain` | Queued (after D-RU-2) | parity with the star path; no position taken from another table's values | diff --git a/.claude/board/entries/2026-10-20-u1-jc-batch.md b/.claude/board/entries/2026-10-20-u1-jc-batch.md new file mode 100644 index 000000000..bc4770185 --- /dev/null +++ b/.claude/board/entries/2026-10-20-u1-jc-batch.md @@ -0,0 +1,54 @@ +# 2026-10-20 — D-RU-1 (U1): a JC statistics batch over shared power-sum passes + +**Status:** MEASURED (lab probe `crates/jc-batch-probe`). Plan: `.claude/plans/d-relalg-unified-v1.md` §3 and the stage table (U1). + +## What landed + +- `answer_separately`: each request evaluates its relalg selection into a keep mask, then runs one power or cross pass over it. This is the baseline. +- `answer_batched`: requests are reduced to the distinct selections after `relalg::normalize`. For each selection, each distinct unordered lane pair gets one `GroupCrossPowerSumsI32` pass keyed by the group lane. A lane that no pair covers gets one `GroupPowerSumsI32` pass. + - An ungrouped answer merges the groups with `checked_merge`. + - A univariate answer reads the `x` or `y` side of the cross sums. + - A reversed pair swaps the sides. +- Statistics: mean, variance, one-way ANOVA, covariance, Pearson, R², and the simple regression line in both directions. +- Both paths refuse a request built for another version (`BatchError::VersionMismatch`). +- `Cost` is counted from the programs that actually run: programs, selection passes, sum passes, rows traversed, values read, mask bytes. +- `examples/bench.rs` (`BENCH_N`, `BENCH_REPS`). + +## Measured + +Battery: 252 requests over 2 selections and 7 `I32` lanes. It reduces to 2 selection passes and 42 cross passes; no univariate pass is needed. The table gives medians of 5 warm runs on 4 cores, load below 1. + +| rows | separate ms | batched ms | time | rows traversed | values read | mask bytes | +|---|---|---|---|---|---|---| +| 100,000 | 40.2 | 5.8 | 6.9× | 11.5× | 8.5× | 126× | +| 1,000,000 | 424.5 | 66.2 | 6.4× | 11.5× | 8.5× | 126× | +| 4,000,000 | 1,818.9 | 279.1 | 6.5× | 11.5× | 8.5× | 126× | + +At 1M rows the counts are: +- programs: 504 → 44; +- rows traversed: 504M → 44M; +- values read: 1,092M → 129M; +- mask bytes: 31.5 MB → 250 kB. + +The time ratio is below the values-read ratio because a cross pass folds five sums per row. + +**Correctness.** The batched answers equal the separate answers bit for bit. They also agree with an independent f64 oracle over directly selected rows: JC's `pearson` and `anova_one_way`, and two-pass mean and variance. + +**Integer range.** +- At `i32::MIN`/`i32::MAX` values, `Σx²` passes `u64` and `Σxy` passes `i64`; the test asserts both. Mean and variance still equal the exact `i128` values. +- At an offset of 10⁹, the naive one-pass f64 Pearson is off by more than 10⁻³. The exact path is within 10⁻¹² of the two-pass value. + +**Tests and disables.** 8 tests pass. Six disable runs each turned red, with every disable patch's anchor asserted: +- swap the reversed pair; +- read the `y` side as `x`; +- drop the pair dedup; +- skip `normalize`; +- merge only the first group; +- drop the version check. + +## OPEN + +- No R reference: R is not installed in this container. The reference is JC's slice functions plus exact integer sums. +- A merge overflow cannot happen below the 2³²-row cap the terminals enforce, so `MergeOverflow` is untested; the cap itself is mask-risc's test. +- The timing of one battery shape. Batches with few requests per pair, so less sharing, are not measured, and neither is the break-even request count. +- Selections are shared only when they are equal after normalization; overlapping selections (U2's let-planes) are not shared yet. diff --git a/.claude/board/entries/README.md b/.claude/board/entries/README.md index f1648aaf3..6aa11f140 100644 --- a/.claude/board/entries/README.md +++ b/.claude/board/entries/README.md @@ -25,10 +25,11 @@ index row, (3) no duplicate entry id. Checks 1 and 2 are deliberately opposite directions; the stranding this convention prevents shows up in exactly one of them, never both. -395 entries, 2026-08-06 .. 2026-10-20. +396 entries, 2026-08-06 .. 2026-10-20. | date | entry id | finding | file | |---|---|---|---| +| 2026-10-20 | `D-RU-1` | | [2026-10-20-u1-jc-batch.md](2026-10-20-u1-jc-batch.md) | | 2026-10-20 | `thought-git-replay-api-write-rubicon` | | [2026-10-20-thought-git-replay-api-write-rubicon.md](2026-10-20-thought-git-replay-api-write-rubicon.md) | | 2026-10-20 | `ADR-2026-10-20-SIX` | | [2026-10-20-six-decisions.md](2026-10-20-six-decisions.md) | | 2026-10-20 | `relstore-quack-petgraph-plan` | | [2026-10-20-relstore-quack-petgraph-plan.md](2026-10-20-relstore-quack-petgraph-plan.md) | diff --git a/.claude/plans/d-relalg-unified-v1.md b/.claude/plans/d-relalg-unified-v1.md index 655cef9ec..f4490fe6a 100644 --- a/.claude/plans/d-relalg-unified-v1.md +++ b/.claude/plans/d-relalg-unified-v1.md @@ -296,4 +296,4 @@ One workload, run under three toggles. Each toggle changes one mechanism only. | U7 | prefetch and staging of Arm B ranges | U6 or a non-recursive multi-stage plan | critical-path overlap measured; bytes read unchanged | | U8 | Wikidata and SAP-shaped fixtures; completeness refusals | U6, U7 | refusal tests; witness carried end to end | -U1 needs nothing new and comes first. U2 builds on D-CMR-1. Both are accepted only on measured row traversals, materializations and total cost (§ Decisions), not on operator counts. +U1 needs nothing new and comes first. **U1 MEASURED (2026-10-20, `crates/jc-batch-probe`, `entries/2026-10-20-u1-jc-batch.md`):** 252 requests reduce to 2 selection and 42 cross passes; at 1M rows the batch is 6.4× faster with 11.5× fewer rows traversed, 8.5× fewer values read and 126× fewer mask bytes, and its answers equal the per-request answers bit for bit. The R reference is replaced by JC's slice functions because R is not installed. U2 builds on D-CMR-1. Both are accepted only on measured row traversals, materializations and total cost (§ Decisions), not on operator counts. diff --git a/.github/workflows/jc-batch-probe.yml b/.github/workflows/jc-batch-probe.yml new file mode 100644 index 000000000..331aba999 --- /dev/null +++ b/.github/workflows/jc-batch-probe.yml @@ -0,0 +1,54 @@ +name: JC batch probe +# `crates/jc-batch-probe` (D-RU-1 / U1) is excluded from the workspace and has +# its own [workspace], so no workspace job builds it. It path-deps jc, relalg +# and mask-risc. jc and relalg name ndarray as a sibling checkout +# (`../../../ndarray` from their directory), so the job clones it first. +on: + pull_request: + paths: + - 'crates/jc-batch-probe/**' + - 'crates/jc/**' + - 'crates/lance-graph-relalg/**' + - 'crates/lance-graph-r2il/**' + - 'crates/lance-graph-mask-risc/**' + - 'crates/lance-graph-quack/**' + - 'crates/lance-graph-contract/**' + - '.github/workflows/jc-batch-probe.yml' + - 'rust-toolchain.toml' + push: + branches: [main] + paths: + - 'crates/jc-batch-probe/**' + - 'crates/jc/**' + - 'crates/lance-graph-relalg/**' + - 'crates/lance-graph-r2il/**' + - 'crates/lance-graph-mask-risc/**' + - 'crates/lance-graph-quack/**' + - 'crates/lance-graph-contract/**' + - '.github/workflows/jc-batch-probe.yml' + - 'rust-toolchain.toml' + workflow_dispatch: +permissions: + contents: read +jobs: + probe: + runs-on: ubuntu-latest + timeout-minutes: 45 + defaults: + run: + working-directory: crates/jc-batch-probe + env: + CARGO_PROFILE_DEV_DEBUG: "0" + CARGO_INCREMENTAL: "0" + steps: + - uses: actions/checkout@v4 + - name: Setup rust toolchain (pinned by rust-toolchain.toml) + run: rustup show + - name: Sibling ndarray checkout + run: git clone --depth 1 https://github.com/AdaWorldAPI/ndarray ../../../ndarray + - run: cargo fmt --check + - run: cargo clippy --release --all-targets -- -D warnings + - run: cargo test --release + # The benchmark asserts that both paths agree before it times them. + - name: Benchmark smoke run + run: BENCH_N=100000 BENCH_REPS=1 cargo run --release --example bench diff --git a/Cargo.toml b/Cargo.toml index 6f5de67e4..bbe702b9e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -174,6 +174,10 @@ exclude = [ # -> MORK query_multi -> relalg over the D-MPS-8 table. Own workspace; run # from its directory so `.cargo/config.toml` applies. "crates/crepe-plan-probe", + # D-RU-1 (U1) JC statistical batch: one battery of means/variances/ANOVA/ + # covariance/Pearson/regression answered from shared power-sum passes vs + # request by request. Lab probe, own workspace: `cd crates/jc-batch-probe && cargo test --release`. + "crates/jc-batch-probe", # D-TGB-0 thought-git pre-registration: fixtures, the from-scratch oracle, # and signal checks over lance-graph-contract only. Lab probe; run from # its directory: `cd crates/thought-git-probe && cargo test`. diff --git a/crates/jc-batch-probe/Cargo.toml b/crates/jc-batch-probe/Cargo.toml new file mode 100644 index 000000000..bc5a90f3f --- /dev/null +++ b/crates/jc-batch-probe/Cargo.toml @@ -0,0 +1,27 @@ +[package] +name = "jc-batch-probe" +version = "0.1.0" +edition = "2021" +publish = false +description = "Lab probe (D-RU-1 / U1): a batch of JC statistical requests answered from shared power-sum passes, against one pass per request. Results only; no production path." + +# Its own workspace root (it path-deps relalg, which git-deps OGAR). Run with +# `cargo test --manifest-path crates/jc-batch-probe/Cargo.toml`. +[workspace] + +[dependencies] +# The selection algebra and its lowering. +lance-graph-relalg = { path = "../lance-graph-relalg" } +# The only bulk engine: the power-sum terminals. +lance-graph-mask-risc = { path = "../lance-graph-mask-risc" } +# Statistical semantics: projections from the sums. +jc = { path = "../jc" } +# The coordinate mask-risc uses; patched below to the sibling jc builds on, +# so `PowerSums` is one type across all three. +ndarray = { git = "https://github.com/AdaWorldAPI/ndarray", branch = "master", default-features = false, features = ["std"] } + +[patch."https://github.com/AdaWorldAPI/lance-graph"] +lance-graph-contract = { path = "../lance-graph-contract" } + +[patch."https://github.com/AdaWorldAPI/ndarray"] +ndarray = { path = "../../../ndarray" } diff --git a/crates/jc-batch-probe/examples/bench.rs b/crates/jc-batch-probe/examples/bench.rs new file mode 100644 index 000000000..c8e84195b --- /dev/null +++ b/crates/jc-batch-probe/examples/bench.rs @@ -0,0 +1,161 @@ +//! U1 benchmark: one statistical battery answered request by request versus +//! from shared power-sum passes. Run in release from this crate's directory: +//! +//! ```text +//! cargo run --release --example bench +//! ``` +//! +//! `BENCH_N` rows (default 1,000,000), 7 `I32` statistic lanes, a selection +//! lane, a category lane, an 8-value group lane and a zero lane. The battery +//! is the one of `tests/batch.rs` over two selections: every mean, variance +//! and ANOVA, and every covariance, Pearson, R² and regression line (both +//! directions) of every pair: 252 requests. Both paths must return equal +//! answers before either is timed. Times are medians of `BENCH_REPS` +//! (default 5) warm runs. + +use std::hint::black_box; +use std::time::{Duration, Instant}; + +use jc_batch_probe::{answer_batched, answer_separately, Request, Stat, Table}; +use lance_graph_mask_risc::LaneRef; +use lance_graph_relalg::{Cmp, Rel}; + +const STAT_LANES: u16 = 7; +const SEL: u16 = 7; +const CAT: u16 = 8; +const GROUP: u16 = 9; +const ZERO: u16 = 10; +const GROUPS: usize = 8; +const VERSION: u64 = 1; + +fn env(k: &str, d: usize) -> usize { + std::env::var(k) + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(d) +} + +fn lcg(s: &mut u64) -> u64 { + *s = s + .wrapping_mul(6364136223846793005) + .wrapping_add(1442695040888963407); + *s >> 33 +} + +fn battery(mut select: impl FnMut() -> Rel) -> Vec { + let mut r = Vec::new(); + let mut push = |stat| { + r.push(Request { + version: VERSION, + select: select(), + stat, + }) + }; + for x in 0..STAT_LANES { + push(Stat::Mean(x)); + push(Stat::Variance(x)); + push(Stat::Anova(x)); + } + for x in 0..STAT_LANES { + for y in (x + 1)..STAT_LANES { + push(Stat::Covariance(x, y)); + push(Stat::Pearson(x, y)); + push(Stat::RSquared(x, y)); + push(Stat::Regression(x, y)); + push(Stat::Regression(y, x)); + } + } + r +} + +fn median(mut v: Vec) -> Duration { + v.sort(); + v[v.len() / 2] +} + +fn main() { + let n = env("BENCH_N", 1_000_000); + let reps = env("BENCH_REPS", 5); + let mut s = 0xBE7C_u64; + let base: Vec = (0..n) + .map(|_| (lcg(&mut s) % 2_001) as i32 - 1_000) + .collect(); + let stats: Vec> = (0..STAT_LANES as i32) + .map(|j| { + base.iter() + .map(|&b| b * j + (lcg(&mut s) % 401) as i32 - 200) + .collect() + }) + .collect(); + let sel: Vec = (0..n).map(|_| (lcg(&mut s) % 100) as i32).collect(); + let cat: Vec = (0..n).map(|_| (lcg(&mut s) % 4) as u32).collect(); + let group: Vec = (0..n) + .map(|_| (lcg(&mut s) % GROUPS as u64) as u32) + .collect(); + let zero = vec![0u32; n]; + let mut lanes: Vec> = stats.iter().map(|v| LaneRef::I32(v)).collect(); + lanes.extend([ + LaneRef::I32(&sel), + LaneRef::U32(&cat), + LaneRef::U32(&group), + LaneRef::U32(&zero), + ]); + let t = Table { + version: VERSION, + n_rows: n, + lanes: &lanes, + group: GROUP, + groups: GROUPS, + zero: ZERO, + }; + + let low = || Rel::i32(SEL, Cmp::Lt, 60); + let mut reqs = battery(low); + reqs.extend(battery(|| Rel::And(vec![low(), Rel::eq_u32(CAT, 2)]))); + + let (a_sep, c_sep) = answer_separately(&t, &reqs).unwrap(); + let (a_bat, c_bat, red) = answer_batched(&t, &reqs).unwrap(); + assert_eq!(a_sep, a_bat, "the two paths must agree before timing"); + + let time = |f: &dyn Fn()| { + let v: Vec = (0..reps) + .map(|_| { + let st = Instant::now(); + f(); + st.elapsed() + }) + .collect(); + median(v) + }; + let t_sep = time(&|| { + black_box(answer_separately(&t, &reqs).unwrap()); + }); + let t_bat = time(&|| { + black_box(answer_batched(&t, &reqs).unwrap()); + }); + + println!("rows {n}, requests {}, {red:?}", reqs.len()); + println!( + "{:>10} {:>9} {:>9} {:>9} {:>14} {:>14} {:>12}", + "path", "ms", "programs", "sum pass", "rows trav.", "values read", "mask bytes" + ); + for (name, ms, c) in [("separate", t_sep, c_sep), ("batched", t_bat, c_bat)] { + println!( + "{:>10} {:>9.1} {:>9} {:>9} {:>14} {:>14} {:>12}", + name, + ms.as_secs_f64() * 1e3, + c.programs, + c.sum_passes, + c.rows_traversed, + c.values_read, + c.mask_bytes + ); + } + println!( + "ratio separate/batched: time {:.1}x, rows {:.1}x, values {:.1}x, masks {:.1}x", + t_sep.as_secs_f64() / t_bat.as_secs_f64(), + c_sep.rows_traversed as f64 / c_bat.rows_traversed as f64, + c_sep.values_read as f64 / c_bat.values_read as f64, + c_sep.mask_bytes as f64 / c_bat.mask_bytes as f64 + ); +} diff --git a/crates/jc-batch-probe/src/lib.rs b/crates/jc-batch-probe/src/lib.rs new file mode 100644 index 000000000..e08f5541d --- /dev/null +++ b/crates/jc-batch-probe/src/lib.rs @@ -0,0 +1,560 @@ +//! D-RU-1 (U1) lab probe: a batch of JC statistical requests answered from +//! shared power-sum passes. Plan: `.claude/plans/d-relalg-unified-v1.md` §3. +//! +//! STATUS: PROBE (results only). Nothing here is wired into JC, Quack or a +//! production path. +//! +//! A request names a row selection (a [`relalg::Rel`]) and a statistic over +//! `I32` lanes. Every statistic here is a projection of grouped sufficient +//! statistics, the sums mask-risc folds exactly: +//! +//! - mean, variance and one-way ANOVA read `(n, Σx, Σx²)`; +//! - covariance, Pearson, the simple regression line and its R² read +//! `(n, Σx, Σy, Σx², Σy², Σxy)`. +//! +//! Two ways to answer a batch: +//! +//! - [`answer_separately`]: every request on its own. It evaluates its +//! selection into a mask, then runs one power-sum pass over it. +//! - [`answer_batched`]: requests are reduced to the distinct selections and, +//! per selection, to the lane pairs they need. Each selection is evaluated +//! once, and each needed pair is folded once, keyed by the batch's group +//! lane. One cross pass over `(x, y)` keyed by group answers every request +//! that needs `x`, `y` or the pair, grouped or not: an ungrouped statistic +//! merges the groups exactly ([`CrossPowerSums::checked_merge`]), and a +//! univariate one reads the `x` side of the cross sums. Lanes no pair +//! covers get one univariate pass. +//! +//! **Exactness.** The sums are integers (`u64`, `i64`, `u128`, `i128`) and +//! the executor refuses a plane past `2^32` rows instead of wrapping. Merging +//! groups is checked addition. Both arms therefore hand JC bit-identical +//! inputs, and every projection is the same `f64` arithmetic on the same +//! integers: the batched answers equal the separate ones exactly, not within +//! a tolerance. +//! +//! **Versions.** A batch reads one snapshot. Every request carries the version +//! it was built for, and a batch over a table at another version, or mixing +//! versions, is refused ([`BatchError::VersionMismatch`]). +//! +//! **Cost accounting.** Every mask-risc program the probe executes spans all +//! `n_rows` rows, so [`Cost`] counts executed programs, the values each one +//! reads (from the lanes the program names), and the mask bytes written. The +//! counts come from the code that runs the programs, not from a model. + +use jc::stats::{ + anova_from_power_sums, pearson_from_cross_power_sums, r_squared_from_cross_power_sums, + sample_covariance_from_cross_power_sums, simple_regression_from_cross_power_sums, Anova, + SimpleRegression, +}; +use lance_graph_mask_risc::{ + execute_into, scratch_words_for, words_for, ExecError, Foreign, GroupKey, LaneRef, Operand, + Out, Planes, Program, Scratch, Terminal, Value, +}; +use lance_graph_relalg::{normalize, run, Atom, Measure, Query, Rel, RelError}; +use ndarray::simd::{CrossPowerSums, PowerSums}; + +/// One statistic over `I32` lanes of the table. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Stat { + /// `Σx / n`. + Mean(u16), + /// The unbiased sample variance. + Variance(u16), + /// One-way ANOVA of the lane across the table's groups. + Anova(u16), + /// The unbiased sample covariance. + Covariance(u16, u16), + /// Pearson's r. + Pearson(u16, u16), + /// The simple least-squares line `y = α + β·x` (first lane is `x`). + Regression(u16, u16), + /// `R²` of that line. + RSquared(u16, u16), +} + +impl Stat { + /// The lanes the statistic reads, `(x, Some(y))` for a pair. + fn lanes(self) -> (u16, Option) { + match self { + Stat::Mean(x) | Stat::Variance(x) | Stat::Anova(x) => (x, None), + Stat::Covariance(x, y) + | Stat::Pearson(x, y) + | Stat::Regression(x, y) + | Stat::RSquared(x, y) => (x, Some(y)), + } + } + + /// Whether the statistic is per group (needs the group lane). + fn grouped(self) -> bool { + matches!(self, Stat::Anova(_)) + } +} + +/// One request: a statistic over a row selection, built for one version. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Request { + /// The table version the request was built against. + pub version: u64, + /// The rows the statistic is taken over. + pub select: Rel, + /// The statistic. + pub stat: Stat, +} + +/// An answer. `None` inside means the statistic is undefined on the +/// selection (JC's degeneracy rules: too few rows, a constant lane, …). +#[derive(Debug, Clone, Copy, PartialEq)] +pub enum Answer { + /// A scalar statistic. + Scalar(Option), + /// A regression line. + Line(Option), + /// An ANOVA result. + Anova(Option), +} + +/// The table a batch runs over: resident lanes owned by the caller. +#[derive(Debug, Clone, Copy)] +pub struct Table<'a> { + /// The version the lanes are at. + pub version: u64, + /// Row count every lane spans. + pub n_rows: usize, + /// The lanes. Statistic lanes are `I32`; `group` and `zero` are `U32`. + pub lanes: &'a [LaneRef<'a>], + /// The `U32` group lane grouped statistics are taken over. + pub group: u16, + /// Group universe: every value of `group` is below it. + pub groups: usize, + /// A `U32` lane of zeros, the key of an ungrouped pass when a request is + /// answered on its own. Caller-owned, as the other lanes are. + pub zero: u16, +} + +/// Why a batch was refused. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum BatchError { + /// A request was built for another version than the table holds. + VersionMismatch { + request: usize, + request_version: u64, + table_version: u64, + }, + /// The selection was decided from its description alone (all rows or + /// none). The probe runs only selections that need a pass. + SymbolicSelection { request: usize }, + /// relalg refused the selection. + Relalg(RelError), + /// mask-risc refused a pass. + Exec(ExecError), + /// Merging group sums overflowed (cannot happen below `2^32` rows). + MergeOverflow, +} + +impl From for BatchError { + fn from(e: RelError) -> Self { + BatchError::Relalg(e) + } +} + +impl From for BatchError { + fn from(e: ExecError) -> Self { + BatchError::Exec(e) + } +} + +/// What answering a batch cost, counted where the programs run. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct Cost { + /// mask-risc programs executed (selection passes plus power-sum passes). + pub programs: usize, + /// Of those, selection passes (relalg evaluating a selection to a mask). + pub selection_passes: usize, + /// Of those, power-sum passes. + pub sum_passes: usize, + /// Rows traversed: every program spans the whole table. + pub rows_traversed: u64, + /// Lane values read: per program, `n_rows` per lane it names. + pub values_read: u64, + /// Mask bytes materialized (one selection mask per selection pass). + pub mask_bytes: u64, +} + +impl Cost { + fn selection(&mut self, n: usize, lanes: usize) { + self.programs += 1; + self.selection_passes += 1; + self.rows_traversed += n as u64; + self.values_read += (n * lanes) as u64; + self.mask_bytes += (words_for(n) * 8) as u64; + } + + fn sums(&mut self, n: usize, lanes: usize) { + self.programs += 1; + self.sum_passes += 1; + self.rows_traversed += n as u64; + self.values_read += (n * lanes) as u64; + } +} + +/// The distinct lanes a normalized selection reads. +fn selection_lanes(rel: &Rel, out: &mut Vec) { + match rel { + Rel::All | Rel::Empty | Rel::Plane(_) => {} + Rel::Atom(Atom::I32 { lane, .. } | Atom::U32 { lane, .. }) => { + if !out.contains(lane) { + out.push(*lane); + } + } + Rel::And(xs) | Rel::Or(xs) => xs.iter().for_each(|x| selection_lanes(x, out)), + Rel::Not(x) => selection_lanes(x, out), + } +} + +fn check_versions(t: &Table<'_>, reqs: &[Request]) -> Result<(), BatchError> { + for (i, r) in reqs.iter().enumerate() { + if r.version != t.version { + return Err(BatchError::VersionMismatch { + request: i, + request_version: r.version, + table_version: t.version, + }); + } + } + Ok(()) +} + +/// Evaluates `select` into a mask (one relalg program). +fn select_mask( + t: &Table<'_>, + select: &Rel, + request: usize, + cost: &mut Cost, +) -> Result, BatchError> { + let planes = Planes { + n_rows: t.n_rows, + masks: &[], + lanes: t.lanes, + }; + let mut keep = vec![0u64; words_for(t.n_rows)]; + let r = run( + &Query { + rel: select.clone(), + measure: Measure::Keep, + }, + &planes, + Some(&mut keep), + )?; + if r.programs == 0 { + return Err(BatchError::SymbolicSelection { request }); + } + let mut lanes = Vec::new(); + selection_lanes(&normalize(select), &mut lanes); + cost.selection(t.n_rows, lanes.len()); + Ok(keep) +} + +fn exec(t: &Table<'_>, mask: &[u64], terminal: Terminal, out: Out<'_>) -> Result<(), BatchError> { + let masks = [mask]; + let planes = Planes { + n_rows: t.n_rows, + masks: &masks, + lanes: t.lanes, + }; + let program = Program::new(Vec::new(), terminal); + let words = words_for(t.n_rows).max(1); + let slots = program.scratch_slots as usize; + let mut buf = vec![0u64; scratch_words_for(words, slots).expect("scratch size")]; + let mut scratch = Scratch::over(&mut buf, words, slots).expect("scratch carves"); + let v = execute_into(&program, &planes, &Foreign::NONE, &mut scratch, out)?; + debug_assert!(matches!( + v, + Value::GroupPowerSums | Value::GroupCrossPowerSums + )); + Ok(()) +} + +/// Grouped `(n, Σx, Σx²)` of lane `x` under `mask`, keyed by `key`. +fn power_pass( + t: &Table<'_>, + mask: &[u64], + key: u16, + groups: usize, + x: u16, + cost: &mut Cost, +) -> Result, BatchError> { + let mut out = vec![PowerSums::default(); groups]; + exec( + t, + mask, + Terminal::GroupPowerSumsI32 { + mask: Operand::Plane(0), + key: GroupKey::Lane(key), + val: x, + }, + Out::PowerSums(&mut out), + )?; + cost.sums(t.n_rows, 2); + Ok(out) +} + +/// Grouped cross sums of `(x, y)` under `mask`, keyed by `key`. +fn cross_pass( + t: &Table<'_>, + mask: &[u64], + key: u16, + groups: usize, + x: u16, + y: u16, + cost: &mut Cost, +) -> Result, BatchError> { + let mut out = vec![CrossPowerSums::default(); groups]; + exec( + t, + mask, + Terminal::GroupCrossPowerSumsI32 { + mask: Operand::Plane(0), + key: GroupKey::Lane(key), + x, + y, + }, + Out::CrossPowerSums(&mut out), + )?; + cost.sums(t.n_rows, 3); + Ok(out) +} + +/// The `x` side of cross sums, as univariate sums. +fn x_side(c: &CrossPowerSums) -> PowerSums { + PowerSums { + n: c.n, + sum: c.sum_x, + sum_sq: c.sum_x_sq, + } +} + +/// The `y` side of cross sums, as univariate sums. +fn y_side(c: &CrossPowerSums) -> PowerSums { + PowerSums { + n: c.n, + sum: c.sum_y, + sum_sq: c.sum_y_sq, + } +} + +/// The cross sums with `x` and `y` swapped. +fn swapped(c: &CrossPowerSums) -> CrossPowerSums { + CrossPowerSums { + n: c.n, + sum_x: c.sum_y, + sum_y: c.sum_x, + sum_x_sq: c.sum_y_sq, + sum_y_sq: c.sum_x_sq, + sum_xy: c.sum_xy, + } +} + +fn merge_power(g: &[PowerSums]) -> Result { + g.iter() + .try_fold(PowerSums::default(), |a, b| a.checked_merge(*b)) + .ok_or(BatchError::MergeOverflow) +} + +fn merge_cross(g: &[CrossPowerSums]) -> Result { + g.iter() + .try_fold(CrossPowerSums::default(), |a, b| a.checked_merge(*b)) + .ok_or(BatchError::MergeOverflow) +} + +/// `Σx / n`, `None` for an empty selection. +pub fn mean(m: &PowerSums) -> Option { + (m.n > 0).then(|| m.sum as f64 / m.n as f64) +} + +/// The unbiased sample variance, formed exactly: `(nΣx² − (Σx)²)` in `i128` +/// before the one division. `None` for `n < 2`. +/// +/// Candidate for `jc::stats` (it has no univariate projection yet); kept here +/// so the probe does not change JC. +pub fn variance(m: &PowerSums) -> Option { + if m.n < 2 { + return None; + } + let n = i128::from(m.n); + let sum = i128::from(m.sum); + let sum_sq = i128::try_from(m.sum_sq).ok()?; + let centered = n.checked_mul(sum_sq)?.checked_sub(sum.checked_mul(sum)?)?; + Some(centered as f64 / (n * (n - 1)) as f64) +} + +fn univariate_answer(stat: Stat, total: &PowerSums, groups: &[PowerSums]) -> Answer { + match stat { + Stat::Mean(_) => Answer::Scalar(mean(total)), + Stat::Variance(_) => Answer::Scalar(variance(total)), + Stat::Anova(_) => Answer::Anova(anova_from_power_sums(groups)), + _ => unreachable!("univariate"), + } +} + +fn pair_answer(stat: Stat, m: &CrossPowerSums) -> Answer { + match stat { + Stat::Covariance(..) => Answer::Scalar(sample_covariance_from_cross_power_sums(m)), + Stat::Pearson(..) => Answer::Scalar(pearson_from_cross_power_sums(m)), + Stat::Regression(..) => Answer::Line(simple_regression_from_cross_power_sums(m)), + Stat::RSquared(..) => Answer::Scalar(r_squared_from_cross_power_sums(m)), + _ => unreachable!("pair"), + } +} + +/// Every request on its own: its selection, then one pass for its +/// statistic (keyed by the group lane only when the statistic is grouped). +pub fn answer_separately( + t: &Table<'_>, + reqs: &[Request], +) -> Result<(Vec, Cost), BatchError> { + check_versions(t, reqs)?; + let mut cost = Cost::default(); + let mut out = Vec::with_capacity(reqs.len()); + for (i, r) in reqs.iter().enumerate() { + let mask = select_mask(t, &r.select, i, &mut cost)?; + let (key, groups) = if r.stat.grouped() { + (t.group, t.groups) + } else { + (t.zero, 1) + }; + out.push(match r.stat.lanes() { + (x, None) => { + let g = power_pass(t, &mask, key, groups, x, &mut cost)?; + univariate_answer(r.stat, &merge_power(&g)?, &g) + } + (x, Some(y)) => { + let g = cross_pass(t, &mask, key, groups, x, y, &mut cost)?; + pair_answer(r.stat, &merge_cross(&g)?) + } + }); + } + Ok((out, cost)) +} + +/// What a batch was reduced to. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct Reduction { + /// Distinct selections. + pub selections: usize, + /// Cross passes (one per distinct unordered pair per selection). + pub cross_passes: usize, + /// Univariate passes (lanes no pair covers). + pub power_passes: usize, +} + +/// The batch reduced to distinct selections and, per selection, to one pass +/// per distinct unordered lane pair plus one per lane no pair covers. Every +/// pass is keyed by the group lane, so grouped and ungrouped requests share +/// it. +pub fn answer_batched( + t: &Table<'_>, + reqs: &[Request], +) -> Result<(Vec, Cost, Reduction), BatchError> { + check_versions(t, reqs)?; + let mut cost = Cost::default(); + let mut red = Reduction::default(); + + // Distinct selections, compared after normalization. + let mut sels: Vec<(Rel, usize)> = Vec::new(); + let sel_of: Vec = reqs + .iter() + .enumerate() + .map(|(i, r)| { + let n = normalize(&r.select); + match sels.iter().position(|(s, _)| *s == n) { + Some(p) => p, + None => { + sels.push((n, i)); + sels.len() - 1 + } + } + }) + .collect(); + red.selections = sels.len(); + + let mut out: Vec> = vec![None; reqs.len()]; + for (s, (_, first)) in sels.iter().enumerate() { + let members: Vec = (0..reqs.len()).filter(|&i| sel_of[i] == s).collect(); + let mask = select_mask(t, &reqs[*first].select, *first, &mut cost)?; + + // Distinct unordered pairs, then the lanes no pair covers. + let mut pairs: Vec<(u16, u16)> = Vec::new(); + for &i in &members { + if let (x, Some(y)) = reqs[i].stat.lanes() { + let p = (x.min(y), x.max(y)); + if !pairs.contains(&p) { + pairs.push(p); + } + } + } + let mut singles: Vec = Vec::new(); + for &i in &members { + if let (x, None) = reqs[i].stat.lanes() { + if !pairs.iter().any(|&(a, b)| a == x || b == x) && !singles.contains(&x) { + singles.push(x); + } + } + } + + let mut cross: Vec<((u16, u16), Vec)> = Vec::with_capacity(pairs.len()); + for &(a, b) in &pairs { + cross.push(( + (a, b), + cross_pass(t, &mask, t.group, t.groups, a, b, &mut cost)?, + )); + } + red.cross_passes += pairs.len(); + let mut power: Vec<(u16, Vec)> = Vec::with_capacity(singles.len()); + for &x in &singles { + power.push((x, power_pass(t, &mask, t.group, t.groups, x, &mut cost)?)); + } + red.power_passes += singles.len(); + + // Per-group univariate sums of `x`, from a pair pass or its own. + let groups_of = |x: u16| -> Vec { + for ((a, b), g) in &cross { + if *a == x { + return g.iter().map(x_side).collect(); + } + if *b == x { + return g.iter().map(y_side).collect(); + } + } + power + .iter() + .find(|(l, _)| *l == x) + .map(|(_, g)| g.clone()) + .expect("every lane is covered") + }; + + for &i in &members { + let stat = reqs[i].stat; + out[i] = Some(match stat.lanes() { + (x, None) => { + let g = groups_of(x); + univariate_answer(stat, &merge_power(&g)?, &g) + } + (x, Some(y)) => { + let (_, g) = cross + .iter() + .find(|(p, _)| *p == (x.min(y), x.max(y))) + .expect("every pair is covered"); + let m = merge_cross(g)?; + pair_answer(stat, &if x <= y { m } else { swapped(&m) }) + } + }); + } + } + Ok(( + out.into_iter() + .map(|a| a.expect("every request answered")) + .collect(), + cost, + red, + )) +} diff --git a/crates/jc-batch-probe/tests/batch.rs b/crates/jc-batch-probe/tests/batch.rs new file mode 100644 index 000000000..fc4129a0e --- /dev/null +++ b/crates/jc-batch-probe/tests/batch.rs @@ -0,0 +1,432 @@ +//! D-RU-1 (U1): a batch answered from shared power-sum passes equals every +//! request answered on its own, costs fewer traversals, values read and mask +//! bytes, and keeps the integer sums exact. + +use jc::reliability::pearson; +use jc::stats::anova_one_way; +use jc_batch_probe::{answer_batched, answer_separately, Answer, BatchError, Request, Stat, Table}; +use lance_graph_mask_risc::LaneRef; +use lance_graph_relalg::{Cmp, Rel}; + +/// Statistic lanes 0..7 (`I32`), a selection lane 7 (`I32`), a category lane +/// 8 (`U32`), the group lane 9 (`U32`, `GROUPS` values) and a zero lane 10. +const STAT_LANES: u16 = 7; +const SEL: u16 = 7; +const CAT: u16 = 8; +const GROUP: u16 = 9; +const ZERO: u16 = 10; +const GROUPS: usize = 8; +const VERSION: u64 = 42; + +struct Fixture { + n: usize, + stats: Vec>, + sel: Vec, + cat: Vec, + group: Vec, + zero: Vec, +} + +fn lcg(s: &mut u64) -> u64 { + *s = s + .wrapping_mul(6364136223846793005) + .wrapping_add(1442695040888963407); + *s >> 33 +} + +impl Fixture { + /// `n` rows; lane `j` is correlated with lane 0 by `j`, plus noise, so + /// the correlations differ from lane to lane. + fn new(n: usize, seed: u64) -> Self { + let mut s = seed; + let base: Vec = (0..n) + .map(|_| (lcg(&mut s) % 2_001) as i32 - 1_000) + .collect(); + let stats = (0..STAT_LANES as usize) + .map(|j| { + base.iter() + .map(|&b| b * j as i32 + (lcg(&mut s) % 401) as i32 - 200) + .collect() + }) + .collect(); + Fixture { + n, + stats, + sel: (0..n).map(|_| (lcg(&mut s) % 100) as i32).collect(), + cat: (0..n).map(|_| (lcg(&mut s) % 4) as u32).collect(), + group: (0..n) + .map(|_| (lcg(&mut s) % GROUPS as u64) as u32) + .collect(), + zero: vec![0; n], + } + } + + fn lanes(&self) -> Vec> { + let mut l: Vec> = self.stats.iter().map(|v| LaneRef::I32(v)).collect(); + l.push(LaneRef::I32(&self.sel)); + l.push(LaneRef::U32(&self.cat)); + l.push(LaneRef::U32(&self.group)); + l.push(LaneRef::U32(&self.zero)); + l + } + + fn table<'a>(&self, lanes: &'a [LaneRef<'a>]) -> Table<'a> { + Table { + version: VERSION, + n_rows: self.n, + lanes, + group: GROUP, + groups: GROUPS, + zero: ZERO, + } + } + + /// The rows a selection keeps, computed directly (not through relalg). + fn rows(&self, which: Which) -> Vec { + (0..self.n) + .filter(|&i| match which { + Which::Low => self.sel[i] < 60, + Which::LowCat => self.sel[i] < 60 && self.cat[i] == 2, + }) + .collect() + } +} + +#[derive(Clone, Copy)] +enum Which { + Low, + LowCat, +} + +/// `sel < 60`. +fn low() -> Rel { + Rel::i32(SEL, Cmp::Lt, 60) +} + +/// `sel < 60 AND cat == 2`, written in two orders. +fn low_cat(flipped: bool) -> Rel { + if flipped { + Rel::And(vec![Rel::eq_u32(CAT, 2), low()]) + } else { + Rel::And(vec![low(), Rel::eq_u32(CAT, 2)]) + } +} + +/// A full battery over one selection: means, variances and ANOVAs of every +/// lane; covariance, Pearson and R² of every pair; and the regression line +/// of every pair in both directions. +fn battery(mut select: impl FnMut() -> Rel) -> Vec { + let mut r = Vec::new(); + let mut push = |stat| { + r.push(Request { + version: VERSION, + select: select(), + stat, + }) + }; + for x in 0..STAT_LANES { + push(Stat::Mean(x)); + push(Stat::Variance(x)); + push(Stat::Anova(x)); + } + for x in 0..STAT_LANES { + for y in (x + 1)..STAT_LANES { + push(Stat::Covariance(x, y)); + push(Stat::Pearson(x, y)); + push(Stat::RSquared(x, y)); + push(Stat::Regression(x, y)); + push(Stat::Regression(y, x)); + } + } + r +} + +/// Two selections, the second written in two orders across its requests. +fn requests() -> Vec { + let mut r = battery(low); + let mut flip = false; + r.extend(battery(move || { + flip = !flip; + low_cat(flip) + })); + r +} + +/// FAILS IF: any batched answer differs from the request answered on its own +/// (bit for bit; the integer sums are exact, so equality is the contract). +/// Anti-vacuity: the battery has defined answers of every kind. +#[test] +fn batched_answers_equal_separate_answers_exactly() { + let fx = Fixture::new(5_000, 0xBA7C); + let lanes = fx.lanes(); + let t = fx.table(&lanes); + let reqs = requests(); + let (sep, _) = answer_separately(&t, &reqs).unwrap(); + let (bat, _, _) = answer_batched(&t, &reqs).unwrap(); + assert_eq!(sep.len(), reqs.len()); + assert_eq!(bat, sep); + let defined = |f: fn(&Answer) -> bool| bat.iter().any(f); + assert!(defined(|a| matches!(a, Answer::Scalar(Some(_))))); + assert!(defined(|a| matches!(a, Answer::Line(Some(_))))); + assert!(defined(|a| matches!(a, Answer::Anova(Some(_))))); +} + +/// FAILS IF: the batch does not reduce to 2 selections and 21 pair passes +/// per selection, or does not cost less than the separate path in programs, +/// rows traversed, values read and mask bytes. +#[test] +fn batched_cost_is_lower_in_traversals_values_and_masks() { + let fx = Fixture::new(5_000, 0xC057); + let lanes = fx.lanes(); + let t = fx.table(&lanes); + let reqs = requests(); + let (_, sep) = answer_separately(&t, &reqs).unwrap(); + let (_, bat, red) = answer_batched(&t, &reqs).unwrap(); + let pairs = (STAT_LANES as usize) * (STAT_LANES as usize - 1) / 2; + assert_eq!(red.selections, 2); + assert_eq!(red.cross_passes, 2 * pairs); + assert_eq!(red.power_passes, 0, "every lane is covered by a pair"); + + let n = fx.n as u64; + assert_eq!(sep.programs, 2 * reqs.len()); + assert_eq!(sep.selection_passes, reqs.len()); + assert_eq!(bat.selection_passes, 2); + assert_eq!(bat.sum_passes, 2 * pairs); + assert_eq!(bat.rows_traversed, (2 + 2 * pairs) as u64 * n); + assert_eq!(bat.mask_bytes, 2 * (fx.n.div_ceil(64) * 8) as u64); + assert!( + bat.rows_traversed * 5 < sep.rows_traversed, + "{bat:?} vs {sep:?}" + ); + assert!(bat.values_read * 5 < sep.values_read, "{bat:?} vs {sep:?}"); + assert!(bat.mask_bytes * 100 < sep.mask_bytes, "{bat:?} vs {sep:?}"); +} + +fn rel_close(a: f64, b: f64) -> bool { + (a - b).abs() <= 1e-9 * (1.0 + a.abs().max(b.abs())) +} + +/// FAILS IF: a batched statistic disagrees with an independent computation +/// over the selected rows in `f64` (JC's slice functions and two-pass +/// formulas; rows selected directly, not through relalg). +#[test] +fn batched_answers_agree_with_an_independent_f64_oracle() { + let fx = Fixture::new(5_000, 0x0C1E); + let lanes = fx.lanes(); + let t = fx.table(&lanes); + let reqs = requests(); + let (bat, _, _) = answer_batched(&t, &reqs).unwrap(); + let half = reqs.len() / 2; + let mut checked = 0; + for (i, (r, a)) in reqs.iter().zip(&bat).enumerate() { + let rows = fx.rows(if i < half { Which::Low } else { Which::LowCat }); + let col = |l: u16| -> Vec { + rows.iter() + .map(|&k| f64::from(fx.stats[l as usize][k])) + .collect() + }; + match (r.stat, a) { + (Stat::Mean(x), Answer::Scalar(Some(m))) => { + let v = col(x); + assert!(rel_close(*m, v.iter().sum::() / v.len() as f64)); + checked += 1; + } + (Stat::Variance(x), Answer::Scalar(Some(s2))) => { + let v = col(x); + let mu = v.iter().sum::() / v.len() as f64; + let want = + v.iter().map(|z| (z - mu) * (z - mu)).sum::() / (v.len() - 1) as f64; + assert!(rel_close(*s2, want), "{s2} vs {want}"); + checked += 1; + } + (Stat::Pearson(x, y), Answer::Scalar(Some(rr))) => { + let want = pearson(&col(x), &col(y)).unwrap(); + assert!(rel_close(*rr, want), "{rr} vs {want}"); + checked += 1; + } + (Stat::Anova(x), Answer::Anova(Some(an))) => { + let mut g = vec![Vec::new(); GROUPS]; + for &k in &rows { + g[fx.group[k] as usize].push(f64::from(fx.stats[x as usize][k])); + } + let want = anova_one_way(&g).unwrap(); + assert!(rel_close(an.f, want.f), "{} vs {}", an.f, want.f); + checked += 1; + } + _ => {} + } + } + assert!(checked > 50, "checked {checked}"); +} + +/// FAILS IF: the sums are not exact at the `i32` extremes. Every value is +/// `i32::MIN` or `i32::MAX`, so `Σx²` passes `u64::MAX` and an `i64` +/// accumulator of `Σx·y` would overflow (checked below); the batched mean +/// and variance must equal the exact `i128` values. +#[test] +fn sums_stay_exact_at_the_i32_extremes() { + let n = 4_096; + let mut fx = Fixture::new(n, 0xE7E7); + for (j, lane) in fx.stats.iter_mut().enumerate() { + for (i, v) in lane.iter_mut().enumerate() { + *v = if (i + j) % 3 == 0 { i32::MIN } else { i32::MAX }; + } + } + fx.sel = vec![0; n]; // every row selected by `low()` + let lanes = fx.lanes(); + let t = fx.table(&lanes); + let reqs: Vec = (0..STAT_LANES) + .flat_map(|x| { + [ + Stat::Mean(x), + Stat::Variance(x), + Stat::Covariance(x, (x + 1) % STAT_LANES), + ] + }) + .map(|stat| Request { + version: VERSION, + select: low(), + stat, + }) + .collect(); + let (bat, _, _) = answer_batched(&t, &reqs).unwrap(); + let (sep, _) = answer_separately(&t, &reqs).unwrap(); + assert_eq!(bat, sep); + + // Anti-vacuity: the naive accumulators really overflow on this data. + let x0 = &fx.stats[0]; + let sq_u64 = x0.iter().try_fold(0u64, |a, &v| { + a.checked_add((i64::from(v) * i64::from(v)) as u64) + }); + assert!( + sq_u64.is_none(), + "Σx² must exceed u64 for the fixture to bite" + ); + let xy_i64 = x0.iter().zip(&fx.stats[1]).try_fold(0i64, |a, (&p, &q)| { + a.checked_add(i64::from(p) * i64::from(q)) + }); + assert!( + xy_i64.is_none(), + "Σxy must exceed i64 for the fixture to bite" + ); + + for (r, a) in reqs.iter().zip(&bat) { + let x = match r.stat { + Stat::Mean(x) | Stat::Variance(x) => x, + _ => continue, + }; + let v = &fx.stats[x as usize]; + let n128 = n as i128; + let s: i128 = v.iter().map(|&z| i128::from(z)).sum(); + let s2: i128 = v.iter().map(|&z| i128::from(z) * i128::from(z)).sum(); + let want = match r.stat { + Stat::Mean(_) => s as f64 / n as f64, + _ => (n128 * s2 - s * s) as f64 / (n128 * (n128 - 1)) as f64, + }; + assert_eq!(*a, Answer::Scalar(Some(want)), "{:?}", r.stat); + } +} + +/// FAILS IF: the exact path loses the accuracy a naive one-pass `f64` +/// formula loses. Lanes sit at an offset of 10⁹ with a small spread; the +/// naive `Σxy/n − x̄ȳ` correlation is visibly wrong there (anti-vacuity), and +/// the exact one matches the two-pass reference. +#[test] +fn exact_sums_keep_accuracy_where_a_naive_f64_pass_loses_it() { + let n = 20_000; + let mut fx = Fixture::new(n, 0xACC0); + let mut s = 7u64; + let off = 1_000_000_000; + fx.stats[0] = (0..n).map(|_| off + (lcg(&mut s) % 1_000) as i32).collect(); + fx.stats[1] = fx.stats[0] + .iter() + .map(|&x| x + (lcg(&mut s) % 300) as i32) + .collect(); + fx.sel = vec![0; n]; + let lanes = fx.lanes(); + let t = fx.table(&lanes); + let reqs = [Request { + version: VERSION, + select: low(), + stat: Stat::Pearson(0, 1), + }]; + let (bat, _, _) = answer_batched(&t, &reqs).unwrap(); + let Answer::Scalar(Some(exact)) = bat[0] else { + panic!("defined") + }; + let x: Vec = fx.stats[0].iter().map(|&v| f64::from(v)).collect(); + let y: Vec = fx.stats[1].iter().map(|&v| f64::from(v)).collect(); + let two_pass = pearson(&x, &y).unwrap(); + let nf = n as f64; + let (sx, sy) = (x.iter().sum::(), y.iter().sum::()); + let sxx: f64 = x.iter().map(|v| v * v).sum(); + let syy: f64 = y.iter().map(|v| v * v).sum(); + let sxy: f64 = x.iter().zip(&y).map(|(a, b)| a * b).sum(); + let naive = (nf * sxy - sx * sy) / ((nf * sxx - sx * sx).sqrt() * (nf * syy - sy * sy).sqrt()); + assert!( + (naive - two_pass).abs() > 1e-3, + "the fixture must defeat the naive formula: {naive} vs {two_pass}" + ); + assert!((exact - two_pass).abs() < 1e-12, "{exact} vs {two_pass}"); +} + +/// FAILS IF: a request built for another version is answered, by either path. +#[test] +fn a_request_for_another_version_is_refused() { + let fx = Fixture::new(500, 0x7E57); + let lanes = fx.lanes(); + let t = fx.table(&lanes); + let mut reqs = requests(); + reqs[3].version = VERSION + 1; + let want = Err(BatchError::VersionMismatch { + request: 3, + request_version: VERSION + 1, + table_version: VERSION, + }); + assert_eq!(answer_batched(&t, &reqs).map(|_| ()), want); + assert_eq!(answer_separately(&t, &reqs).map(|_| ()), want); +} + +/// FAILS IF: two selections equal after normalization are evaluated twice, +/// or two different selections are merged. The second selection is written +/// in two atom orders across its requests. +#[test] +fn selections_are_shared_after_normalization_and_kept_apart_otherwise() { + let fx = Fixture::new(1_000, 0x5E15); + let lanes = fx.lanes(); + let t = fx.table(&lanes); + let reqs = requests(); + assert!(reqs.iter().any(|r| r.select == low_cat(true))); + assert!(reqs.iter().any(|r| r.select == low_cat(false))); + let (_, cost, red) = answer_batched(&t, &reqs).unwrap(); + assert_eq!(red.selections, 2); + assert_eq!(cost.selection_passes, 2); +} + +/// FAILS IF: a lane no pair request covers is not given its own pass, or +/// that pass answers differently from the separate path. +#[test] +fn a_lane_only_univariate_requests_need_gets_one_pass() { + let fx = Fixture::new(2_000, 0x51A6); + let lanes = fx.lanes(); + let t = fx.table(&lanes); + let reqs: Vec = [ + Stat::Pearson(0, 1), + Stat::Mean(4), + Stat::Variance(4), + Stat::Anova(4), + Stat::Mean(0), + ] + .into_iter() + .map(|stat| Request { + version: VERSION, + select: low(), + stat, + }) + .collect(); + let (bat, cost, red) = answer_batched(&t, &reqs).unwrap(); + let (sep, _) = answer_separately(&t, &reqs).unwrap(); + assert_eq!(bat, sep); + assert_eq!((red.cross_passes, red.power_passes), (1, 1)); + assert_eq!(cost.sum_passes, 2); +}