From f7216919b4ce456155bcd7feda35ae2e4cd12deb Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 20 Oct 2026 23:43:19 +0000 Subject: [PATCH 1/3] jc-batch-probe: U1 statistical batch over shared power-sum passes (WIP, pre-disable) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f --- crates/jc-batch-probe/Cargo.toml | 27 ++ crates/jc-batch-probe/src/lib.rs | 560 +++++++++++++++++++++++++++ crates/jc-batch-probe/tests/batch.rs | 432 +++++++++++++++++++++ 3 files changed, 1019 insertions(+) create mode 100644 crates/jc-batch-probe/Cargo.toml create mode 100644 crates/jc-batch-probe/src/lib.rs create mode 100644 crates/jc-batch-probe/tests/batch.rs 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/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); +} From 02886314ed08acb2bf4cf4856318df04d17f8015 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 20 Oct 2026 23:47:58 +0000 Subject: [PATCH 2/3] jc-batch-probe: U1 benchmark, workspace exclude and board (D-RU-1 measured) 252 requests over 2 selections reduce to 2 selection and 42 cross passes. At 1M rows the batch is 6.4x faster than answering each request alone, with 11.5x fewer rows traversed, 8.5x fewer values read and 126x fewer mask bytes; answers are bit-identical. Six disable runs turn the tests red. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f --- .claude/board/STATUS_BOARD.md | 2 +- .../board/entries/2026-10-20-u1-jc-batch.md | 54 ++++++ .claude/board/entries/README.md | 3 +- .claude/plans/d-relalg-unified-v1.md | 2 +- Cargo.toml | 4 + crates/jc-batch-probe/examples/bench.rs | 161 ++++++++++++++++++ 6 files changed, 223 insertions(+), 3 deletions(-) create mode 100644 .claude/board/entries/2026-10-20-u1-jc-batch.md create mode 100644 crates/jc-batch-probe/examples/bench.rs 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/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/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 + ); +} From 7c9ead98c8ff61b4c0965b35ad935a35045d5f03 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 20 Oct 2026 23:53:15 +0000 Subject: [PATCH 3/3] jc-batch-probe: CI workflow (fmt, clippy, tests, benchmark smoke run) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f --- .github/workflows/jc-batch-probe.yml | 54 ++++++++++++++++++++++++++++ 1 file changed, 54 insertions(+) create mode 100644 .github/workflows/jc-batch-probe.yml 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