diff --git a/.claude/board/STATUS_BOARD.md b/.claude/board/STATUS_BOARD.md index a33af65e5..d64c7d12f 100644 --- a/.claude/board/STATUS_BOARD.md +++ b/.claude/board/STATUS_BOARD.md @@ -6,7 +6,7 @@ Plan: `.claude/plans/d-relalg-unified-v1.md`. Decisions recorded there: ADMISSIO |---|---|---|---| | **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 | 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-2** | U2: `let`-bound planes on the contract IR, immutable `Plane(i)`, versioned coordinates, no DAG inside a fold | MEASURED (`contract::set_program`, `relalg::program`; `entries/2026-10-21-u2-let-planes.md`): 16 queries sharing one conjunction at 1M rows: 3.1× faster, 14 vs 59 lane reads, 2 planes (250 kB) materialized, ≤ 2 scratch slots; answers equal recompute and a row oracle; other versions refused | 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 | | **D-RU-5** | Quack physical pins (dense vs sorted, materialize vs recompute) | Queued (after D-RU-3/4) | can-fire and can-stay-silent per pin | diff --git a/.claude/board/entries/2026-10-21-u2-let-planes.md b/.claude/board/entries/2026-10-21-u2-let-planes.md new file mode 100644 index 000000000..176e6f745 --- /dev/null +++ b/.claude/board/entries/2026-10-21-u2-let-planes.md @@ -0,0 +1,66 @@ +# 2026-10-21 — D-RU-2 (U2): let-bound planes over the contract set algebra + +**Status:** MEASURED (lab; `crates/lance-graph-contract/src/set_program.rs`, `crates/lance-graph-relalg/src/program.rs`). Plan: `.claude/plans/d-relalg-unified-v1.md`, stage table (U2). Builds on D-CMR-1 (`lance_graph_contract::set_algebra`, #1644). + +## What landed + +- `set_program::SetProgram { version, resident, lets, queries }`. + - `Sym(i)` with `i < resident` is a resident plane. + - `Sym(resident + k)` is `lets[k]`. + - A let may read only earlier lets; queries may read any let. + - `validate()` refuses forward and unknown references. + - A program describes one table version. +- `set_program::share(version, resident, queries)` lifts a junction subset into a let when two or more junctions of the same kind hold it. Rules: + - largest subset first; + - judged after `normalize`, because `normalize` flattens nested junctions, so identical subtrees rarely survive whole; + - lets are ordered by dependency and renumbered densely. +- `relalg::run_program`: + - evaluates each let once into a scratch plane of `n_rows / 8` bytes; + - runs every query over the resident planes plus those planes; + - a let's plane is one more resident plane, so every fold stays a tree. +- `relalg::run_recompute` is the baseline: each query with its lets inlined. +- Both refuse: + - a program for another version (`VersionMismatch`); + - a malformed program (`Shape`); + - the wrong number of resident planes; + - a measure count that does not match the queries. +- `ProgramCost` counts, from the folds that ran: programs, ops, lane reads, plane reads, planes materialized and their bytes, and the most scratch slots one fold used. + +## Measured + +`examples/program_bench`, release build: 16 queries that share one three-comparison conjunction; one of every three also reads a resident plane. Medians of 9 runs on 4 cores, load below 1. + +| rows | program ms | recompute ms | time | lane reads | ops | materialized | +|---|---|---|---|---|---|---| +| 100,000 | 0.35 | 1.15 | 3.3× | 14 vs 59 | 15 vs 59 | 2 planes, 25 kB | +| 1,000,000 | 4.77 | 14.87 | 3.1× | 14 vs 59 | 15 vs 59 | 2 planes, 250 kB | +| 4,000,000 | 20.98 | 69.35 | 3.3× | 14 vs 59 | 15 vs 59 | 2 planes, 1 MB | + +- The lane-read count is the row-traversal count: each comparison reads one whole lane of `n_rows` values. +- The program runs 18 mask-risc programs against 16 for recompute: one per let more, each a whole pass. +- With 4 queries, the program reads 6 lanes against 15 and is 2.1× faster at 1M rows. +- Both paths use at most 2 scratch slots per fold, inside the ≤ 3 target. + +**Correctness.** +- Three paths agree on every answer (count, sum, min, max) and on every kept mask: the program, recompute, and a row-at-a-time oracle over the inlined descriptions. +- Fixtures: hand-built sets with one shared conjunction; with nested and distinct lets; with nothing shared; and 60 generated sets. +- With nothing shared, the program's cost equals recompute's exactly. +- Versions: a program built for version 7 is refused against version 8. + +**Disables.** Eight, all red-then-green: +- wrong let evaluated; +- lets cannot read earlier lets; +- version check removed; +- `validate` removed; +- resident-count check removed; +- lane reads not counted; +- bytes not counted; +- `share` lifts nothing. + +The first and second were green at first: no fixture held a program with more than one let. `nested_and_distinct_lets_agree_with_recompute_and_the_oracle` closes that. + +## Open + +- `share` is greedy (largest subset first). It does not weigh a let's materialization cost against the reads it saves. A subset held by two junctions is lifted even when its comparisons are cheaper than the extra pass. Measured here only where sharing wins. +- The ops count includes the let passes. Total cost is bounded by time, which is measured; there is no byte-level traffic counter. +- U3 (cross-rule CSE in crepe-plan-probe) and U4 (join over `JoinDomain`) are unblocked. diff --git a/.claude/board/entries/README.md b/.claude/board/entries/README.md index 2a43cc4a9..b9943fbfa 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. -401 entries, 2026-08-06 .. 2026-10-21. +402 entries, 2026-08-06 .. 2026-10-21. | date | entry id | finding | file | |---|---|---|---| +| 2026-10-21 | `D-RU-2` | | [2026-10-21-u2-let-planes.md](2026-10-21-u2-let-planes.md) | | 2026-10-21 | `D-RU-6-LAB-1` | | [2026-10-21-d-ru-6-tile-walk-probe.md](2026-10-21-d-ru-6-tile-walk-probe.md) | | 2026-10-21 | `D-RU-6-LAB-2` | | [2026-10-21-d-ru-6-sparse-compute.md](2026-10-21-d-ru-6-sparse-compute.md) | | 2026-10-21 | `D-CMR-5` | | [2026-10-21-d-cmr-5-rescope.md](2026-10-21-d-cmr-5-rescope.md) | diff --git a/.claude/plans/d-relalg-unified-v1.md b/.claude/plans/d-relalg-unified-v1.md index cbad31659..3b5d03da8 100644 --- a/.claude/plans/d-relalg-unified-v1.md +++ b/.claude/plans/d-relalg-unified-v1.md @@ -298,4 +298,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. **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. +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 MEASURED (2026-10-21, `contract::set_program` + `relalg::program`, `entries/2026-10-21-u2-let-planes.md`):** each shared junction subset is evaluated once into a scratch plane; for 16 queries sharing one conjunction at 1M rows the program is 3.1× faster than recompute and reads 14 lanes instead of 59, at the cost of 2 materialized planes (250 kB). Answers equal recompute and a row oracle, and a program for another version is refused. Both are accepted only on measured row traversals, materializations and total cost (§ Decisions), not on operator counts. diff --git a/.github/workflows/crepe-plan-probe.yml b/.github/workflows/crepe-plan-probe.yml index 2e27a3f84..172f8d65e 100644 --- a/.github/workflows/crepe-plan-probe.yml +++ b/.github/workflows/crepe-plan-probe.yml @@ -71,3 +71,6 @@ jobs: cargo fmt --check cargo test cargo clippy --all-targets -- -D warnings + # D-RU-2 benchmark smoke run; an n that is not a multiple of 64 + # exercises the plane tail. + BENCH_N=100003 BENCH_REPS=1 cargo run --example program_bench diff --git a/crates/lance-graph-contract/src/lib.rs b/crates/lance-graph-contract/src/lib.rs index 2ec79c93a..6a18970df 100644 --- a/crates/lance-graph-contract/src/lib.rs +++ b/crates/lance-graph-contract/src/lib.rs @@ -191,6 +191,9 @@ pub mod scheduler; pub mod selection; pub mod sensorium; pub mod set_algebra; +/// D-RU-2 / U2: a batch of selections whose shared parts are named once +/// (`lets` over `set_algebra::SetExpr::Sym`). See module docs. +pub mod set_program; pub mod settlement; /// D-BLW-5 payload-law DTO (`ShapeRankPayload`, `RemeasureKey`, `RemeasureLedger`) — /// distribution shape × Prozentrang, never the raw statistic. See module docs. diff --git a/crates/lance-graph-contract/src/set_program.rs b/crates/lance-graph-contract/src/set_program.rs new file mode 100644 index 000000000..6cba4bfb7 --- /dev/null +++ b/crates/lance-graph-contract/src/set_program.rs @@ -0,0 +1,554 @@ +//! A batch of selections over one population with its shared parts named +//! once (D-RU-2 / U2, `.claude/plans/d-relalg-unified-v1.md` §5.1). +//! +//! A [`SetProgram`] holds `lets` and `queries`, each a [`SetExpr`]. A let is +//! the definition of one [`SetExpr::Sym`] id: `Sym(resident + k)` means +//! `lets[k]`. Ids below `resident` name sets the caller already holds (an +//! executor's resident planes). The meaning of every query is the query with +//! its let references replaced by their definitions ([`SetProgram::inline`]). +//! +//! # What this buys an executor +//! +//! An executor may compute each let once, keep the result, and let every +//! query read it. The alternative, sharing a node inside one expression, +//! would turn a fold's tree into a DAG; a fold that releases consumed +//! operands (relalg's `call_owned`) needs every operand used once, so the +//! shared value is kept outside the fold instead. Each fold stays a tree. +//! +//! # Rules +//! +//! - **Order.** A let may refer only to resident ids and to earlier lets; +//! a query may refer to any let. So no let depends on itself and an +//! executor can evaluate the lets in order ([`SetProgram::validate`]). +//! - **Immutability.** A let's value is fixed for the program. Nothing +//! assigns to an id twice. +//! - **Version.** A program describes one table at one version +//! ([`SetProgram::version`]). An executor must refuse to evaluate it +//! against another version; reusing a let's value across versions would +//! answer from stale rows. +//! +//! # Finding the shared parts +//! +//! [`share`] normalizes every query ([`normalize`]) and lifts into a let any +//! set of at least two members that two or more junctions of one kind hold +//! (`a ∧ b` in `a ∧ b ∧ c` and `a ∧ b ∧ d`; an identical junction is the +//! case where the set is all its members). Normalization flattens nested +//! conjunctions, so a shared `x ∧ y` written inside two rules is found only +//! as such a common subset, never as a node. The part with the most leaves +//! goes first, so a shared expression is not split into smaller lets that +//! only it uses. +//! +//! Zero dependencies. + +use std::collections::{BTreeMap, BTreeSet}; + +use crate::set_algebra::{normalize, SetExpr}; + +/// Selections over one table at one version, with shared parts as lets. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SetProgram { + /// The table version every let and query is described against. + pub version: u64, + /// Ids below this are resident sets the caller supplies. + pub resident: u32, + /// `lets[k]` defines `Sym(resident + k)`. + pub lets: Vec, + /// The selections the program answers. + pub queries: Vec, +} + +/// Why a program is malformed. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ProgramError { + /// `lets[at]` names `sym`, which is itself or a later let. + ForwardReference { at: usize, sym: u32 }, + /// A let or query names an id at or past `resident + lets.len()`. + UnknownSym { sym: u32 }, + /// `resident + lets.len()` does not fit a `u32` id. + TooManyIds, +} + +impl SetProgram { + /// The id `lets[k]` defines; `None` when it does not fit a `u32`. + pub fn let_sym(&self, k: usize) -> Option { + u32::try_from(k).ok()?.checked_add(self.resident) + } + + /// Checks that every reference is defined and that lets refer only + /// backwards. + pub fn validate(&self) -> Result<(), ProgramError> { + let bound = self + .let_sym(self.lets.len()) + .ok_or(ProgramError::TooManyIds)?; + for (k, l) in self.lets.iter().enumerate() { + // `k < lets.len()`, and `resident + lets.len()` fits. + let own = self.resident + k as u32; + for sym in syms(l) { + if sym >= bound { + return Err(ProgramError::UnknownSym { sym }); + } + if sym >= own { + return Err(ProgramError::ForwardReference { at: k, sym }); + } + } + } + for q in &self.queries { + if let Some(sym) = syms(q).into_iter().find(|&s| s >= bound) { + return Err(ProgramError::UnknownSym { sym }); + } + } + Ok(()) + } + + /// `e` with every let reference replaced by its definition, recursively: + /// what `e` means without the lets. Resident ids stay. + pub fn inline(&self, e: &SetExpr) -> SetExpr { + match e { + SetExpr::Sym(s) if *s >= self.resident => { + let k = (*s - self.resident) as usize; + match self.lets.get(k) { + Some(def) => self.inline(def), + None => e.clone(), + } + } + SetExpr::All | SetExpr::Empty | SetExpr::Sym(_) | SetExpr::Res(_) => e.clone(), + SetExpr::And(cs) => SetExpr::And(cs.iter().map(|c| self.inline(c)).collect()), + SetExpr::Or(cs) => SetExpr::Or(cs.iter().map(|c| self.inline(c)).collect()), + SetExpr::Not(c) => SetExpr::negated(self.inline(c)), + } + } +} + +/// Every `Sym` id `e` names, in visiting order. +fn syms(e: &SetExpr) -> Vec { + let mut out = Vec::new(); + fn walk(e: &SetExpr, out: &mut Vec) { + match e { + SetExpr::Sym(s) => out.push(*s), + SetExpr::And(cs) | SetExpr::Or(cs) => cs.iter().for_each(|c| walk(c, out)), + SetExpr::Not(c) => walk(c, out), + _ => {} + } + } + walk(e, &mut out); + out +} + +/// Which junction a node is: `true` for `And`, `false` for `Or`. +fn kind(e: &SetExpr) -> Option<(bool, &[SetExpr])> { + match e { + SetExpr::And(cs) => Some((true, cs)), + SetExpr::Or(cs) => Some((false, cs)), + _ => None, + } +} + +fn junction(and: bool, members: Vec) -> SetExpr { + if and { + SetExpr::And(members) + } else { + SetExpr::Or(members) + } +} + +/// Every junction node in `e`, nested ones included, as its kind and member +/// set. +fn junctions(e: &SetExpr, out: &mut Vec<(bool, BTreeSet)>) { + if let Some((and, cs)) = kind(e) { + out.push((and, cs.iter().cloned().collect())); + } + match e { + SetExpr::And(cs) | SetExpr::Or(cs) => cs.iter().for_each(|c| junctions(c, out)), + SetExpr::Not(c) => junctions(c, out), + _ => {} + } +} + +/// `e` with the members `part` taken out of every junction of the same kind +/// that holds all of them, and `sym` put in their place. A junction whose +/// members are exactly `part` becomes `sym`. +fn replace(e: &SetExpr, and: bool, part: &BTreeSet, sym: &SetExpr) -> SetExpr { + let rec = |cs: &[SetExpr]| -> Vec { + cs.iter().map(|c| replace(c, and, part, sym)).collect() + }; + match e { + SetExpr::And(cs) | SetExpr::Or(cs) => { + let this_and = matches!(e, SetExpr::And(_)); + if this_and == and && part.iter().all(|m| cs.contains(m)) { + let rest: Vec = cs.iter().filter(|c| !part.contains(c)).cloned().collect(); + if rest.is_empty() { + return sym.clone(); + } + let mut members = rec(&rest); + members.push(sym.clone()); + return junction(and, members); + } + junction(this_and, rec(cs)) + } + SetExpr::Not(c) => SetExpr::negated(replace(c, and, part, sym)), + _ => e.clone(), + } +} + +/// `e` with each id renamed through `rename` (ids not in it stay). +fn rename_syms(e: &SetExpr, rename: &BTreeMap) -> SetExpr { + match e { + SetExpr::Sym(s) => SetExpr::Sym(*rename.get(s).unwrap_or(s)), + SetExpr::And(cs) => SetExpr::And(cs.iter().map(|c| rename_syms(c, rename)).collect()), + SetExpr::Or(cs) => SetExpr::Or(cs.iter().map(|c| rename_syms(c, rename)).collect()), + SetExpr::Not(c) => SetExpr::negated(rename_syms(c, rename)), + _ => e.clone(), + } +} + +/// The next part to lift: among the member sets two junctions of one kind +/// have in common (at least two members), the one with the most leaves that +/// at least two junctions hold; ties go to more holders, then to the +/// expression order. +fn next_part(nodes: &[(bool, BTreeSet)]) -> Option<(bool, BTreeSet)> { + let mut parts: BTreeSet<(bool, BTreeSet)> = BTreeSet::new(); + for (i, (ka, ma)) in nodes.iter().enumerate() { + for (kb, mb) in &nodes[i + 1..] { + if ka != kb { + continue; + } + let common: BTreeSet = ma.intersection(mb).cloned().collect(); + if common.len() >= 2 { + parts.insert((*ka, common)); + } + } + } + parts + .into_iter() + .map(|(and, part)| { + let holders = nodes + .iter() + .filter(|(k, m)| *k == and && part.is_subset(m)) + .count(); + let leaves: usize = part.iter().map(SetExpr::leaves).sum(); + (leaves, holders, and, part) + }) + .max_by(|a, b| { + (a.0, a.1) + .cmp(&(b.0, b.1)) + .then_with(|| (&b.2, &b.3).cmp(&(&a.2, &a.3))) + }) + .map(|(_, _, and, part)| (and, part)) +} + +/// A program for `queries` (over `resident` caller-held sets, at `version`) +/// in which every part two junctions share after normalization is a let. +/// +/// The returned queries mean what the given ones mean +/// ([`SetProgram::inline`] of each, normalized, equals the given query +/// normalized), and the lets satisfy [`SetProgram::validate`]. +pub fn share(version: u64, resident: u32, queries: &[SetExpr]) -> SetProgram { + let mut qs: Vec = queries.iter().map(normalize).collect(); + // Lets in extraction order, each with the id it was given then. + let mut lets: Vec<(u32, SetExpr)> = Vec::new(); + // A cap, not a count: a round's part is held by two or more junctions + // and afterwards co-occurs only inside its let. If the cap stops the + // loop early the program is still correct; it just shares less. + let budget: usize = qs.iter().map(SetExpr::leaves).sum(); + for _ in 0..budget { + let mut nodes = Vec::new(); + for e in qs.iter().chain(lets.iter().map(|(_, l)| l)) { + junctions(e, &mut nodes); + } + let Some((and, part)) = next_part(&nodes) else { + break; + }; + // `validate` needs `resident + lets.len()` to fit a `u32`. If the + // next let would break that, stop: the program is still correct, it + // just shares less. + let Some(id) = u32::try_from(lets.len() + 1) + .ok() + .and_then(|k| k.checked_add(resident)) + .map(|end| end - 1) + else { + break; + }; + let sym = SetExpr::Sym(id); + for q in &mut qs { + *q = replace(q, and, &part, &sym); + } + for (_, l) in &mut lets { + *l = replace(l, and, &part, &sym); + } + lets.push((id, junction(and, part.into_iter().collect()))); + } + // A let extracted later can occur inside one extracted earlier, so order + // the lets by dependency and renumber them densely. + let order = dependency_order(&lets); + let rename: BTreeMap = order + .iter() + .enumerate() + // Every new id is at most an id already given out, so it fits. + .map(|(new, &old)| (lets[old].0, resident + new as u32)) + .collect(); + SetProgram { + version, + resident, + lets: order + .iter() + .map(|&i| normalize(&rename_syms(&lets[i].1, &rename))) + .collect(), + queries: qs + .iter() + .map(|q| normalize(&rename_syms(q, &rename))) + .collect(), + } +} + +/// Indices into `lets` such that every let comes after the lets it names. +/// The references form a DAG: a let names only lets replaced inside it, and +/// replacement never puts an id inside its own definition. +fn dependency_order(lets: &[(u32, SetExpr)]) -> Vec { + let index: BTreeMap = lets + .iter() + .enumerate() + .map(|(i, (id, _))| (*id, i)) + .collect(); + let mut placed = vec![false; lets.len()]; + let mut order = Vec::with_capacity(lets.len()); + fn visit( + i: usize, + lets: &[(u32, SetExpr)], + index: &BTreeMap, + placed: &mut [bool], + order: &mut Vec, + ) { + if placed[i] { + return; + } + placed[i] = true; + for s in syms(&lets[i].1) { + if let Some(&j) = index.get(&s) { + visit(j, lets, index, placed, order); + } + } + order.push(i); + } + for i in 0..lets.len() { + visit(i, lets, &index, &mut placed, &mut order); + } + order +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::set_algebra::{Cmp, Res}; + + fn a(t: i32) -> SetExpr { + SetExpr::i32(0, Cmp::Lt, t) + } + fn b(v: u32) -> SetExpr { + SetExpr::eq_u32(1, v) + } + fn c(t: i32) -> SetExpr { + SetExpr::i32(2, Cmp::Gt, t) + } + + /// The shared junction of the tests: `(a<10 ∨ b=3) ∧ c>5`. + fn shared() -> SetExpr { + SetExpr::And(vec![SetExpr::Or(vec![a(10), b(3)]), c(5)]) + } + + fn meaning(p: &SetProgram, q: &SetExpr) -> SetExpr { + normalize(&p.inline(q)) + } + + /// FAILS IF: a repeated junction is not lifted, or a lifted one changes + /// what a query means. + #[test] + fn a_repeated_junction_becomes_one_let_and_meanings_are_kept() { + let qs = vec![ + SetExpr::And(vec![shared(), b(7)]), + SetExpr::Or(vec![shared(), a(-4)]), + SetExpr::And(vec![a(0), c(1)]), + ]; + let p = share(9, 2, &qs); + assert_eq!(p.validate(), Ok(())); + assert_eq!(p.lets, vec![normalize(&shared())]); + for (q, orig) in p.queries.iter().zip(&qs) { + assert_eq!(meaning(&p, q), normalize(orig)); + } + // The two queries that share it name the let instead of repeating it. + assert!(syms(&p.queries[0]).contains(&2)); + assert!(syms(&p.queries[1]).contains(&2)); + assert!(syms(&p.queries[2]).is_empty()); + } + + /// FAILS IF: the share pass lifts a junction that occurs once. + /// Anti-vacuity: the queries have junctions; none repeats. + #[test] + fn nothing_is_lifted_when_nothing_repeats() { + let qs = vec![ + SetExpr::And(vec![a(1), b(2)]), + SetExpr::Or(vec![a(3), c(4)]), + ]; + let p = share(1, 0, &qs); + assert!(p.lets.is_empty()); + assert_eq!(p.queries, qs.iter().map(normalize).collect::>()); + } + + /// FAILS IF: the largest repeated junction is split into smaller lets. + /// Its inner `Or` repeats too, but only inside the shared junction, so + /// one let suffices. + #[test] + fn the_largest_repeated_junction_is_lifted_whole() { + let qs = vec![ + SetExpr::And(vec![shared(), b(7)]), + SetExpr::And(vec![shared(), b(8)]), + ]; + let p = share(1, 0, &qs); + assert_eq!(p.lets.len(), 1, "{:?}", p.lets); + assert_eq!(p.lets[0], normalize(&shared())); + } + + /// FAILS IF: a let that a larger let uses is placed after it, or ids are + /// not renumbered. The inner `Or` also occurs alone in a third query, so + /// both are lets and the larger one names the smaller. + #[test] + fn nested_lets_are_ordered_by_dependency() { + let inner = SetExpr::Or(vec![a(10), b(3)]); + let qs = vec![ + SetExpr::And(vec![shared(), b(7)]), + SetExpr::And(vec![shared(), b(8)]), + SetExpr::And(vec![inner.clone(), c(99)]), + SetExpr::And(vec![inner.clone(), b(5)]), + ]; + let p = share(1, 0, &qs); + assert_eq!(p.validate(), Ok(())); + assert_eq!(p.lets.len(), 2, "{:?}", p.lets); + assert_eq!(p.lets[0], normalize(&inner)); + assert!(syms(&p.lets[1]).contains(&0), "{:?}", p.lets[1]); + for (q, orig) in p.queries.iter().zip(&qs) { + assert_eq!(meaning(&p, q), normalize(orig)); + } + } + + /// FAILS IF: two queries that differ only in how they were written are + /// not recognised as one shared junction. + #[test] + fn sharing_is_judged_after_normalization() { + let x = SetExpr::And(vec![SetExpr::Or(vec![a(10), b(3)]), c(5)]); + let y = SetExpr::And(vec![c(5), SetExpr::Or(vec![b(3), a(10)])]); + let p = share(1, 0, &[x, y]); + assert_eq!(p.lets.len(), 1); + assert_eq!(p.queries, vec![SetExpr::Sym(0), SetExpr::Sym(0)]); + } + + /// FAILS IF: a forward or unknown reference passes validation. + #[test] + fn validation_refuses_forward_and_unknown_references() { + let fwd = SetProgram { + version: 1, + resident: 1, + lets: vec![SetExpr::And(vec![SetExpr::Sym(2), a(1)]), b(2)], + queries: vec![SetExpr::Sym(1)], + }; + assert_eq!( + fwd.validate(), + Err(ProgramError::ForwardReference { at: 0, sym: 2 }) + ); + let own = SetProgram { + lets: vec![SetExpr::Or(vec![SetExpr::Sym(1), a(1)])], + queries: vec![], + ..fwd.clone() + }; + assert_eq!( + own.validate(), + Err(ProgramError::ForwardReference { at: 0, sym: 1 }) + ); + let unknown = SetProgram { + lets: vec![], + queries: vec![SetExpr::Sym(1)], + ..fwd + }; + assert_eq!(unknown.validate(), Err(ProgramError::UnknownSym { sym: 1 })); + } + + /// FAILS IF: a set of conditions two conjunctions have in common is not + /// shared because neither conjunction holds it as a node. + #[test] + fn a_common_subset_of_two_conjunctions_is_shared() { + let qs = vec![ + SetExpr::And(vec![a(1), b(2), c(3)]), + SetExpr::And(vec![a(1), b(2), c(4)]), + ]; + let p = share(1, 0, &qs); + assert_eq!(p.lets, vec![normalize(&SetExpr::And(vec![a(1), b(2)]))]); + for (q, orig) in p.queries.iter().zip(&qs) { + assert_eq!(meaning(&p, q), normalize(orig)); + assert_eq!(q.leaves(), 2, "{q:?}"); + } + } + + /// FAILS IF: an `And` part is taken out of an `Or` that holds the same + /// members. `a ∨ b` and `a ∧ b` share members but not meaning. + #[test] + fn parts_are_shared_only_between_junctions_of_one_kind() { + let qs = vec![ + SetExpr::And(vec![a(1), b(2), c(3)]), + SetExpr::Or(vec![a(1), b(2), c(4)]), + ]; + let p = share(1, 0, &qs); + assert!(p.lets.is_empty(), "{:?}", p.lets); + } + + /// FAILS IF: resident ids are treated as lets. `Sym(0)` is resident here. + /// FAILS IF: `resident + lets.len()` overflows instead of being + /// refused, or `share` overflows instead of sharing less. + #[test] + fn ids_past_u32_are_refused_not_wrapped() { + let a = SetExpr::Res(Res::U32 { + lane: 0, + eq: true, + v: 1, + }); + let p = SetProgram { + version: 0, + resident: u32::MAX, + lets: vec![a.clone()], + queries: vec![SetExpr::Sym(0)], + }; + assert_eq!(p.validate(), Err(ProgramError::TooManyIds)); + assert_eq!(p.let_sym(1), None); + let fits = SetProgram { + resident: u32::MAX - 1, + ..p + }; + assert_eq!(fits.validate(), Ok(())); + assert_eq!(fits.let_sym(0), Some(u32::MAX - 1)); + + // `share` stops before an id would not fit. + let b = SetExpr::Res(Res::U32 { + lane: 1, + eq: true, + v: 2, + }); + let q = |c: u32| SetExpr::And(vec![a.clone(), b.clone(), SetExpr::Sym(c)]); + let full = share(0, u32::MAX, &[q(0), q(1)]); + assert!(full.lets.is_empty(), "{full:?}"); + assert_eq!(full.validate(), Ok(())); + let room = share(0, u32::MAX - 1, &[q(0), q(1)]); + assert_eq!(room.lets.len(), 1, "{room:?}"); + assert_eq!(room.validate(), Ok(())); + } + + #[test] + fn resident_ids_are_not_lets() { + let r = SetExpr::Sym(0); + let qs = vec![ + SetExpr::And(vec![r.clone(), a(1), b(2)]), + SetExpr::And(vec![r.clone(), a(1), b(2)]), + ]; + let p = share(1, 1, &qs); + assert_eq!(p.lets.len(), 1); + assert_eq!(p.queries, vec![SetExpr::Sym(1), SetExpr::Sym(1)]); + assert_eq!(p.inline(&SetExpr::Sym(0)), SetExpr::Sym(0)); + assert_eq!(p.validate(), Ok(())); + } +} diff --git a/crates/lance-graph-relalg/examples/program_bench.rs b/crates/lance-graph-relalg/examples/program_bench.rs new file mode 100644 index 000000000..b89511069 --- /dev/null +++ b/crates/lance-graph-relalg/examples/program_bench.rs @@ -0,0 +1,134 @@ +//! D-RU-2 benchmark: a batch of queries sharing one conjunction, run as a +//! [`SetProgram`] (each let once into a scratch plane) against recompute +//! (every query inlined). Reports time, lane reads, ops and materialized +//! bytes. `BENCH_N` rows (default 1M), `BENCH_REPS` repetitions (default 9), +//! `BENCH_Q` queries (default 16). + +use std::time::Instant; + +use lance_graph_contract::set_algebra::{Cmp, Res, SetExpr}; +use lance_graph_contract::set_program::share; +use lance_graph_mask_risc::{words_for, LaneRef, Planes}; +use lance_graph_relalg::{run_program, run_recompute, Measure}; + +fn env(name: &str, default: usize) -> usize { + std::env::var(name) + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(default) +} + +fn median(mut v: Vec) -> f64 { + v.sort_by(f64::total_cmp); + v[v.len() / 2] +} + +fn main() { + let n = env("BENCH_N", 1_000_000); + let reps = env("BENCH_REPS", 9); + let nq = env("BENCH_Q", 16); + + let x: Vec = (0..n).map(|i| (i % 50) as i32 - 20).collect(); + let y: Vec = (0..n).map(|i| ((i * 7) % 31) as i32).collect(); + let z: Vec = (0..n).map(|i| ((i * 13) % 97) as i32).collect(); + let v: Vec = (0..n).map(|i| (i % 13) as i32 - 6).collect(); + let k: Vec = (0..n).map(|i| (i % 5) as u32).collect(); + let mut p0: Vec = (0..words_for(n)) + .map(|w| 0x9249_2492_4924_9249u64.rotate_left(w as u32)) + .collect(); + // mask-risc refuses set bits past the last row. + if !n.is_multiple_of(64) { + *p0.last_mut().expect("n > 0") &= (1u64 << (n % 64)) - 1; + } + let lanes = [ + LaneRef::I32(&x), + LaneRef::I32(&y), + LaneRef::I32(&z), + LaneRef::I32(&v), + LaneRef::U32(&k), + ]; + let masks: [&[u64]; 1] = [&p0]; + let planes = Planes { + n_rows: n, + masks: &masks, + lanes: &lanes, + }; + + let core = vec![ + SetExpr::i32(0, Cmp::Ge, -5), + SetExpr::i32(1, Cmp::Lt, 20), + SetExpr::Res(Res::U32 { + lane: 4, + eq: true, + v: 2, + }), + ]; + let queries: Vec = (0..nq) + .map(|i| { + let mut cs = core.clone(); + cs.push(match i % 3 { + 0 => SetExpr::i32(2, Cmp::Lt, i as i32 * 5), + 1 => SetExpr::Sym(0), + _ => SetExpr::i32(2, Cmp::Gt, i as i32 * 3), + }); + SetExpr::And(cs) + }) + .collect(); + let p = share(1, 1, &queries); + let ms: Vec = (0..nq) + .map(|i| { + if i % 2 == 0 { + Measure::Count + } else { + Measure::Sum(3) + } + }) + .collect(); + + let mut none: Vec<&mut [u64]> = Vec::new(); + let mut t_prog = Vec::new(); + let mut t_rec = Vec::new(); + let mut last = None; + for _ in 0..reps { + let t = Instant::now(); + let a = run_program(&p, 1, &planes, &ms, &mut none).expect("program"); + t_prog.push(t.elapsed().as_secs_f64() * 1e3); + let t = Instant::now(); + let b = run_recompute(&p, 1, &planes, &ms, &mut none).expect("recompute"); + t_rec.push(t.elapsed().as_secs_f64() * 1e3); + assert_eq!(a.answers, b.answers, "answers diverged"); + last = Some((a.cost, b.cost)); + } + let (pc, rc) = last.expect("reps >= 1"); + println!( + "n_rows={n} queries={nq} lets={} reps={reps} avx512f={} avx2={}", + p.lets.len(), + cfg!(target_feature = "avx512f"), + cfg!(target_feature = "avx2") + ); + println!( + "program : {:>8.2} ms lane_reads={} passes={} programs={} planes={} bytes={} slots={}", + median(t_prog.clone()), + pc.lane_reads, + pc.passes, + pc.programs, + pc.planes_materialized, + pc.plane_bytes, + pc.max_scratch_slots + ); + println!( + "recompute : {:>8.2} ms lane_reads={} passes={} programs={} planes={} bytes={} slots={}", + median(t_rec.clone()), + rc.lane_reads, + rc.passes, + rc.programs, + rc.planes_materialized, + rc.plane_bytes, + rc.max_scratch_slots + ); + println!( + "ratio : time {:.2}x lane_reads {:.2}x", + median(t_rec) / median(t_prog), + rc.lane_reads as f64 / pc.lane_reads as f64 + ); +} diff --git a/crates/lance-graph-relalg/src/lib.rs b/crates/lance-graph-relalg/src/lib.rs index 06917a5eb..760708f94 100644 --- a/crates/lance-graph-relalg/src/lib.rs +++ b/crates/lance-graph-relalg/src/lib.rs @@ -32,7 +32,9 @@ pub mod ir; pub mod join_body; pub mod lower; pub mod normalize; +pub mod program; pub use ir::{Atom, Cmp, Measure, Query, Rel}; pub use lower::{run, run_as_written, run_with_program, Answer, RelError, Run}; pub use normalize::{normalize, symbolic}; +pub use program::{run_program, run_recompute, ProgramCost, ProgramError, ProgramRun}; diff --git a/crates/lance-graph-relalg/src/normalize.rs b/crates/lance-graph-relalg/src/normalize.rs index f2a19c0c2..cde5000f3 100644 --- a/crates/lance-graph-relalg/src/normalize.rs +++ b/crates/lance-graph-relalg/src/normalize.rs @@ -54,7 +54,7 @@ fn cmp_from(c: sa::Cmp) -> Cmp { } } -fn to_set(rel: &Rel) -> SetExpr { +pub(crate) fn to_set(rel: &Rel) -> SetExpr { match rel { Rel::All => SetExpr::All, Rel::Empty => SetExpr::Empty, @@ -77,7 +77,7 @@ fn to_set(rel: &Rel) -> SetExpr { /// Inverse of [`to_set`]. Every `Sym` the contract returns came from a /// `Plane`, so the narrowing never fails on a converted input. -fn from_set(e: &SetExpr) -> Rel { +pub(crate) fn from_set(e: &SetExpr) -> Rel { match e { SetExpr::All => Rel::All, SetExpr::Empty => Rel::Empty, diff --git a/crates/lance-graph-relalg/src/program.rs b/crates/lance-graph-relalg/src/program.rs new file mode 100644 index 000000000..799f6aaa4 --- /dev/null +++ b/crates/lance-graph-relalg/src/program.rs @@ -0,0 +1,224 @@ +//! Running a [`SetProgram`]: each let once into a scratch plane, then every +//! query over the resident planes and those scratch planes (D-RU-2 / U2). +//! +//! A let's plane is one more resident plane to the queries that read it, so +//! every fold stays a tree and keeps [`run`](crate::run)'s width-free scratch +//! bound. Nothing here loops over rows: each let and each query is one +//! [`run_with_program`] call. +//! +//! [`run_recompute`] is the reference: every query with its lets inlined, +//! run on its own. Both report the same [`ProgramCost`], so a caller can +//! compare what sharing saves (lane reads, ops) with what it costs (one +//! `n_rows / 8`-byte plane per let). + +use lance_graph_contract::set_program::{ProgramError as ShapeError, SetProgram}; +use lance_graph_mask_risc::{words_for, Planes}; + +use crate::ir::{Measure, Query, Rel}; +use crate::lower::{run_with_program, Answer, RelError}; +use crate::normalize::{from_set, normalize}; + +/// What running a program cost, counted from the folds that ran. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct ProgramCost { + /// mask-risc programs executed. + pub programs: usize, + /// mask-risc ops executed. + pub passes: usize, + /// Comparisons lowered: each reads one whole value lane. + pub lane_reads: usize, + /// Plane loads lowered: each reads one whole bit plane. + pub plane_reads: usize, + /// Lets computed into scratch planes. + pub planes_materialized: usize, + /// Bytes of those planes. + pub plane_bytes: usize, + /// The most scratch slots any one fold used. + pub max_scratch_slots: u32, +} + +/// The answers, in query order, and their cost. +#[derive(Debug, Clone, PartialEq)] +pub struct ProgramRun { + pub answers: Vec, + pub cost: ProgramCost, +} + +/// Why a program was refused. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ProgramError { + /// The program describes another version of the table. + VersionMismatch { program: u64, table: u64 }, + /// The program's references are malformed. + Shape(ShapeError), + /// The program expects `expected` resident planes; the caller gave `got`. + ResidentPlanes { expected: u32, got: usize }, + /// One measure per query is needed. + Measures { queries: usize, measures: usize }, + /// Resident planes plus lets exceed the `u16` plane index. + TooManyPlanes, + /// A fold was refused. + Rel(RelError), +} + +impl From for ProgramError { + fn from(e: RelError) -> Self { + ProgramError::Rel(e) + } +} + +fn check( + p: &SetProgram, + table_version: u64, + planes: &Planes<'_>, + measures: &[Measure], +) -> Result<(), ProgramError> { + if p.version != table_version { + return Err(ProgramError::VersionMismatch { + program: p.version, + table: table_version, + }); + } + // Before `validate`, so an oversized program is refused by the plane + // limit it cannot meet rather than by its id arithmetic. + if p.resident as usize + p.lets.len() > usize::from(u16::MAX) + 1 { + return Err(ProgramError::TooManyPlanes); + } + p.validate().map_err(ProgramError::Shape)?; + if planes.masks.len() != p.resident as usize { + return Err(ProgramError::ResidentPlanes { + expected: p.resident, + got: planes.masks.len(), + }); + } + if measures.len() != p.queries.len() { + return Err(ProgramError::Measures { + queries: p.queries.len(), + measures: measures.len(), + }); + } + Ok(()) +} + +fn plane_count(rel: &Rel) -> usize { + match rel { + Rel::Plane(_) => 1, + Rel::And(cs) | Rel::Or(cs) => cs.iter().map(plane_count).sum(), + Rel::Not(c) => plane_count(c), + _ => 0, + } +} + +/// Runs one fold and adds what it cost. +fn fold( + q: &Query, + planes: &Planes<'_>, + keep: Option<&mut [u64]>, + cost: &mut ProgramCost, +) -> Result { + let rel = normalize(&q.rel); + let (run, program) = run_with_program(q, planes, keep)?; + cost.programs += run.programs; + cost.passes += run.passes; + if run.programs > 0 { + cost.lane_reads += rel.atoms(); + cost.plane_reads += plane_count(&rel); + } + if let Some(p) = program { + cost.max_scratch_slots = cost.max_scratch_slots.max(p.scratch_slots); + } + Ok(run.answer) +} + +/// Runs the queries with `measures[i]` for `queries[i]`. A `Measure::Keep` +/// writes into the next sink of `keeps`, in query order. +/// +/// Refused when the program was built for another version, its references +/// are malformed, `planes` does not hold exactly its resident planes, or the +/// measures do not match the queries. +pub fn run_program( + p: &SetProgram, + table_version: u64, + planes: &Planes<'_>, + measures: &[Measure], + keeps: &mut [&mut [u64]], +) -> Result { + check(p, table_version, planes, measures)?; + let words = words_for(planes.n_rows); + let mut cost = ProgramCost::default(); + let mut scratch: Vec> = vec![vec![0; words]; p.lets.len()]; + for k in 0..p.lets.len() { + let (done, rest) = scratch.split_at_mut(k); + let masks: Vec<&[u64]> = planes + .masks + .iter() + .copied() + .chain(done.iter().map(Vec::as_slice)) + .collect(); + let view = Planes { + n_rows: planes.n_rows, + masks: &masks, + lanes: planes.lanes, + }; + let q = Query { + rel: from_set(&p.lets[k]), + measure: Measure::Keep, + }; + fold(&q, &view, Some(&mut rest[0]), &mut cost)?; + cost.planes_materialized += 1; + cost.plane_bytes += words * 8; + } + let masks: Vec<&[u64]> = planes + .masks + .iter() + .copied() + .chain(scratch.iter().map(Vec::as_slice)) + .collect(); + let view = Planes { + n_rows: planes.n_rows, + masks: &masks, + lanes: planes.lanes, + }; + let rels: Vec = p.queries.iter().map(from_set).collect(); + answer_all(&rels, measures, &view, keeps, &mut cost).map(|answers| ProgramRun { answers, cost }) +} + +/// The reference for [`run_program`]: each query with its lets inlined, run +/// on its own over the resident planes. Same checks, same answers. +pub fn run_recompute( + p: &SetProgram, + table_version: u64, + planes: &Planes<'_>, + measures: &[Measure], + keeps: &mut [&mut [u64]], +) -> Result { + check(p, table_version, planes, measures)?; + let mut cost = ProgramCost::default(); + let rels: Vec = p.queries.iter().map(|q| from_set(&p.inline(q))).collect(); + answer_all(&rels, measures, planes, keeps, &mut cost) + .map(|answers| ProgramRun { answers, cost }) +} + +fn answer_all( + rels: &[Rel], + measures: &[Measure], + planes: &Planes<'_>, + keeps: &mut [&mut [u64]], + cost: &mut ProgramCost, +) -> Result, ProgramError> { + let mut sinks = keeps.iter_mut(); + rels.iter() + .zip(measures) + .map(|(rel, &measure)| { + let sink = match measure { + Measure::Keep => Some(&mut **sinks.next().ok_or(RelError::NoKeepSink)?), + _ => None, + }; + let q = Query { + rel: rel.clone(), + measure, + }; + fold(&q, planes, sink, cost) + }) + .collect() +} diff --git a/crates/lance-graph-relalg/tests/program.rs b/crates/lance-graph-relalg/tests/program.rs new file mode 100644 index 000000000..ae6bcabab --- /dev/null +++ b/crates/lance-graph-relalg/tests/program.rs @@ -0,0 +1,424 @@ +//! D-RU-2 (U2): a [`SetProgram`] run with shared lets must give the same +//! answers as running every query with its lets inlined, and as a +//! row-at-a-time oracle — and must read fewer lanes when something is +//! shared. + +use lance_graph_contract::set_algebra::{Cmp, SetExpr}; +use lance_graph_contract::set_program::{share, ProgramError as ShapeError, SetProgram}; +use lance_graph_mask_risc::{words_for, LaneRef, Planes}; +use lance_graph_relalg::{run_program, run_recompute, Answer, Measure, ProgramError}; + +const N: usize = 3_000; +const VERSION: u64 = 7; +const RESIDENT: u32 = 2; + +// Lane indexes. +const X: u16 = 0; // i32, (i % 50) - 20 +const Y: u16 = 1; // i32, (i * 7) % 31 +const V: u16 = 2; // i32, the summed value +const K: u16 = 3; // u32, i % 5 + +struct Fixture { + x: Vec, + y: Vec, + v: Vec, + k: Vec, + p: [Vec; 2], +} + +fn plane(f: impl Fn(usize) -> bool) -> Vec { + let mut w = vec![0u64; words_for(N)]; + for r in (0..N).filter(|&r| f(r)) { + w[r / 64] |= 1 << (r % 64); + } + w +} + +fn fixture() -> Fixture { + Fixture { + x: (0..N).map(|i| (i % 50) as i32 - 20).collect(), + y: (0..N).map(|i| ((i * 7) % 31) as i32).collect(), + v: (0..N).map(|i| (i % 13) as i32 - 6).collect(), + k: (0..N).map(|i| (i % 5) as u32).collect(), + p: [plane(|r| r % 3 == 0), plane(|r| r % 7 < 3)], + } +} + +fn with_planes(fx: &Fixture, f: impl FnOnce(&Planes<'_>) -> T) -> T { + let lanes = [ + LaneRef::I32(&fx.x), + LaneRef::I32(&fx.y), + LaneRef::I32(&fx.v), + LaneRef::U32(&fx.k), + ]; + let masks: [&[u64]; 2] = [&fx.p[0], &fx.p[1]]; + let planes = Planes { + n_rows: N, + masks: &masks, + lanes: &lanes, + }; + f(&planes) +} + +// ── the row oracle, over resident-only descriptions ───────────────────────── + +fn holds(fx: &Fixture, e: &SetExpr, r: usize) -> bool { + use lance_graph_contract::set_algebra::Res; + match e { + SetExpr::All => true, + SetExpr::Empty => false, + SetExpr::Sym(i) => fx.p[*i as usize][r / 64] >> (r % 64) & 1 == 1, + SetExpr::Res(Res::I32 { lane, op, t }) => { + let x = [&fx.x, &fx.y, &fx.v][*lane as usize][r]; + match op { + Cmp::Lt => x < *t, + Cmp::Le => x <= *t, + Cmp::Gt => x > *t, + Cmp::Ge => x >= *t, + Cmp::Eq => x == *t, + Cmp::Ne => x != *t, + } + } + SetExpr::Res(Res::U32 { lane, eq, v }) => { + assert_eq!(*lane, K); + (fx.k[r] == *v) == *eq + } + SetExpr::And(cs) => cs.iter().all(|c| holds(fx, c, r)), + SetExpr::Or(cs) => cs.iter().any(|c| holds(fx, c, r)), + SetExpr::Not(c) => !holds(fx, c, r), + } +} + +fn oracle(fx: &Fixture, e: &SetExpr, m: Measure) -> (Answer, Vec) { + let rows: Vec = (0..N).filter(|&r| holds(fx, e, r)).collect(); + let mask = plane(|r| holds(fx, e, r)); + let vals = |l: u16| -> Vec { + let lane = [&fx.x, &fx.y, &fx.v][l as usize]; + rows.iter().map(|&r| lane[r]).collect() + }; + let a = match m { + Measure::Count => Answer::Count(rows.len() as u64), + Measure::Sum(l) => Answer::Sum(vals(l).iter().map(|&v| i64::from(v)).sum()), + Measure::Min(l) => Answer::Min(vals(l).into_iter().min()), + Measure::Max(l) => Answer::Max(vals(l).into_iter().max()), + Measure::Keep => Answer::Kept, + }; + (a, mask) +} + +// ── building blocks ───────────────────────────────────────────────────────── + +fn x(op: Cmp, t: i32) -> SetExpr { + SetExpr::i32(X, op, t) +} +fn y(op: Cmp, t: i32) -> SetExpr { + SetExpr::i32(Y, op, t) +} +fn k(v: u32) -> SetExpr { + SetExpr::Res(lance_graph_contract::set_algebra::Res::U32 { + lane: K, + eq: true, + v, + }) +} +fn and(cs: Vec) -> SetExpr { + SetExpr::And(cs) +} +fn or(cs: Vec) -> SetExpr { + SetExpr::Or(cs) +} + +/// Four queries that all carry the conjunction `x >= -5 ∧ y < 20 ∧ k == 2`. +fn shared_queries() -> Vec { + let core = vec![x(Cmp::Ge, -5), y(Cmp::Lt, 20), k(2)]; + let with = |extra: SetExpr| { + let mut cs = core.clone(); + cs.push(extra); + and(cs) + }; + vec![ + with(SetExpr::Sym(0)), + with(SetExpr::Sym(1)), + with(x(Cmp::Lt, 25)), + or(vec![and(core.clone()), y(Cmp::Eq, 3)]), + ] +} + +fn measures(n: usize) -> Vec { + let cycle = [ + Measure::Count, + Measure::Sum(V), + Measure::Min(X), + Measure::Max(Y), + Measure::Keep, + ]; + (0..n).map(|i| cycle[i % cycle.len()]).collect() +} + +/// Program, recompute and oracle must agree on every answer and kept mask. +fn assert_agree( + fx: &Fixture, + p: &SetProgram, + ms: &[Measure], +) -> ( + lance_graph_relalg::ProgramCost, + lance_graph_relalg::ProgramCost, +) { + let keeps_n = ms.iter().filter(|m| matches!(m, Measure::Keep)).count(); + let mut k1 = vec![vec![0u64; words_for(N)]; keeps_n]; + let mut k2 = vec![vec![0u64; words_for(N)]; keeps_n]; + let (a, b) = with_planes(fx, |pl| { + let mut s1: Vec<&mut [u64]> = k1.iter_mut().map(Vec::as_mut_slice).collect(); + let mut s2: Vec<&mut [u64]> = k2.iter_mut().map(Vec::as_mut_slice).collect(); + ( + run_program(p, VERSION, pl, ms, &mut s1).expect("program runs"), + run_recompute(p, VERSION, pl, ms, &mut s2).expect("recompute runs"), + ) + }); + assert_eq!(a.answers, b.answers, "program vs recompute"); + let mut keep_i = 0; + for (i, (q, &m)) in p.queries.iter().zip(ms).enumerate() { + let (want, mask) = oracle(fx, &p.inline(q), m); + assert_eq!(a.answers[i], want, "query {i} vs oracle"); + if m == Measure::Keep { + assert_eq!(k1[keep_i], mask, "query {i} program kept mask"); + assert_eq!(k2[keep_i], mask, "query {i} recompute kept mask"); + keep_i += 1; + } + } + (a.cost, b.cost) +} + +// ── tests ─────────────────────────────────────────────────────────────────── + +/// FAILS IF: a let is evaluated against the wrong planes, a query reads a +/// let's plane at the wrong index, or the sinks are paired with the wrong +/// queries. +#[test] +fn shared_lets_give_the_same_answers_as_recompute_and_the_oracle() { + let fx = fixture(); + let qs = shared_queries(); + let p = share(VERSION, RESIDENT, &qs); + assert!(!p.lets.is_empty(), "the shared conjunction must be lifted"); + for rot in 0..5 { + let mut ms = measures(qs.len() + rot); + ms.drain(..rot); + assert_agree(&fx, &p, &ms); + } +} + +/// FAILS IF: sharing does not cut lane reads, or the program does not +/// account for the planes it materializes. +#[test] +fn sharing_reads_fewer_lanes_and_pays_one_plane_per_let() { + let fx = fixture(); + let qs = shared_queries(); + let p = share(VERSION, RESIDENT, &qs); + let (prog, rec) = assert_agree(&fx, &p, &measures(qs.len())); + // The core's three comparisons are read once instead of four times. + assert_eq!(rec.lane_reads, 3 * 4 + 1 + 1, "recompute: {rec:?}"); + assert_eq!(prog.lane_reads, 3 + 1 + 1, "program: {prog:?}"); + assert!(prog.passes < rec.passes, "{prog:?} vs {rec:?}"); + assert_eq!(prog.planes_materialized, p.lets.len()); + assert_eq!(prog.plane_bytes, p.lets.len() * words_for(N) * 8); + assert_eq!(rec.planes_materialized, 0); + assert_eq!(rec.plane_bytes, 0); + assert_eq!(prog.programs, rec.programs + p.lets.len()); + // Trees, not DAGs: scratch stays small. + assert!(prog.max_scratch_slots <= 3, "{prog:?}"); + assert!(rec.max_scratch_slots <= 3, "{rec:?}"); +} + +/// A let that reads another let, plus a second, unrelated let. +fn nested_queries() -> Vec { + let ab = vec![x(Cmp::Ge, -5), y(Cmp::Lt, 20)]; + let cat = |extra: Vec| { + let mut cs = ab.clone(); + cs.extend(extra); + and(cs) + }; + let alt = or(vec![k(4), y(Cmp::Gt, 25)]); + vec![ + cat(vec![k(2), SetExpr::Sym(0)]), + cat(vec![k(2), SetExpr::Sym(1)]), + cat(vec![x(Cmp::Lt, 10)]), + cat(vec![y(Cmp::Gt, 4)]), + and(vec![alt.clone(), SetExpr::Sym(0)]), + and(vec![alt, x(Cmp::Lt, 0)]), + ] +} + +/// Whether some let reads an earlier let. +fn has_nested_let(p: &SetProgram) -> bool { + fn reads_let(e: &SetExpr, resident: u32) -> bool { + match e { + SetExpr::Sym(i) => *i >= resident, + SetExpr::And(cs) | SetExpr::Or(cs) => cs.iter().any(|c| reads_let(c, resident)), + SetExpr::Not(c) => reads_let(c, resident), + _ => false, + } + } + p.lets.iter().any(|l| reads_let(l, p.resident)) +} + +/// FAILS IF: a let is evaluated with another let's description, or cannot +/// read the plane of a let computed before it. +#[test] +fn nested_and_distinct_lets_agree_with_recompute_and_the_oracle() { + let fx = fixture(); + let qs = nested_queries(); + let p = share(VERSION, RESIDENT, &qs); + // Anti-vacuity: the fixture must yield several lets, one reading another. + assert!(p.lets.len() >= 3, "lets: {:?}", p.lets); + assert!(has_nested_let(&p), "no nested let: {:?}", p.lets); + let (prog, rec) = assert_agree(&fx, &p, &measures(qs.len())); + assert!(prog.lane_reads < rec.lane_reads, "{prog:?} vs {rec:?}"); + assert_eq!(prog.planes_materialized, p.lets.len()); +} + +/// FAILS IF: an unshared program changes cost, or materializes anything. +#[test] +fn nothing_shared_costs_exactly_what_recompute_costs() { + let fx = fixture(); + let qs = vec![ + and(vec![x(Cmp::Ge, 0), SetExpr::Sym(0)]), + or(vec![y(Cmp::Lt, 5), k(1)]), + SetExpr::Not(Box::new(SetExpr::Sym(1))), + ]; + let p = share(VERSION, RESIDENT, &qs); + assert!(p.lets.is_empty()); + let (prog, rec) = assert_agree(&fx, &p, &measures(qs.len())); + assert_eq!(prog, rec); +} + +/// FAILS IF: an answer computed on one version is served for another. +#[test] +fn a_program_for_another_version_is_refused() { + let fx = fixture(); + let p = share(VERSION, RESIDENT, &shared_queries()); + let ms = measures(p.queries.len()); + with_planes(&fx, |pl| { + let mut sink = vec![0u64; words_for(N)]; + let mut sinks: Vec<&mut [u64]> = vec![&mut sink]; + let want = ProgramError::VersionMismatch { + program: VERSION, + table: VERSION + 1, + }; + assert_eq!( + run_program(&p, VERSION + 1, pl, &ms, &mut sinks).unwrap_err(), + want + ); + assert_eq!( + run_recompute(&p, VERSION + 1, pl, &ms, &mut sinks).unwrap_err(), + want + ); + }); +} + +/// FAILS IF: a malformed program, the wrong resident planes, or too few +/// measures run anyway. +#[test] +fn malformed_programs_and_mismatched_inputs_are_refused() { + let fx = fixture(); + with_planes(&fx, |pl| { + let mut none: Vec<&mut [u64]> = Vec::new(); + let forward = SetProgram { + version: VERSION, + resident: RESIDENT, + lets: vec![SetExpr::Sym(RESIDENT + 1), x(Cmp::Lt, 0)], + queries: vec![SetExpr::Sym(RESIDENT)], + }; + assert_eq!( + run_program(&forward, VERSION, pl, &[Measure::Count], &mut none).unwrap_err(), + ProgramError::Shape(ShapeError::ForwardReference { + at: 0, + sym: RESIDENT + 1 + }) + ); + + let three = SetProgram { + version: VERSION, + resident: 3, + lets: vec![], + queries: vec![SetExpr::Sym(2)], + }; + assert_eq!( + run_program(&three, VERSION, pl, &[Measure::Count], &mut none).unwrap_err(), + ProgramError::ResidentPlanes { + expected: 3, + got: 2 + } + ); + + let huge = SetProgram { + version: VERSION, + resident: u32::MAX, + lets: vec![x(Cmp::Lt, 0)], + queries: vec![SetExpr::Sym(0)], + }; + assert_eq!( + run_program(&huge, VERSION, pl, &[Measure::Count], &mut none).unwrap_err(), + ProgramError::TooManyPlanes + ); + + let p = share(VERSION, RESIDENT, &shared_queries()); + assert_eq!( + run_program(&p, VERSION, pl, &[Measure::Count], &mut none).unwrap_err(), + ProgramError::Measures { + queries: p.queries.len(), + measures: 1 + } + ); + }); +} + +/// FAILS IF: any of many generated query sets disagrees between the three. +/// The pool is small so repeated junctions are common. +#[test] +fn generated_query_sets_agree() { + let fx = fixture(); + let pool = [ + x(Cmp::Ge, -5), + x(Cmp::Lt, 10), + y(Cmp::Lt, 20), + y(Cmp::Gt, 4), + k(2), + k(4), + SetExpr::Sym(0), + SetExpr::Sym(1), + ]; + let mut seed: u64 = 0x9E37_79B9_7F4A_7C15; + let mut next = |m: usize| { + seed ^= seed << 13; + seed ^= seed >> 7; + seed ^= seed << 17; + (seed % m as u64) as usize + }; + let mut shared_any = 0; + for _ in 0..60 { + let mut qs = Vec::new(); + for _ in 0..(2 + next(5)) { + let pick = |next: &mut dyn FnMut(usize) -> usize| -> Vec { + (0..(2 + next(3))) + .map(|_| pool[next(pool.len())].clone()) + .collect() + }; + let inner = pick(&mut next); + let q = match next(4) { + 0 => and(inner), + 1 => or(inner), + 2 => and(vec![or(inner), pool[next(pool.len())].clone()]), + _ => SetExpr::Not(Box::new(and(inner))), + }; + qs.push(q); + } + let p = share(VERSION, RESIDENT, &qs); + if !p.lets.is_empty() { + shared_any += 1; + } + let (prog, rec) = assert_agree(&fx, &p, &measures(qs.len())); + assert!(prog.lane_reads <= rec.lane_reads, "{prog:?} vs {rec:?}"); + } + // Anti-vacuity: the generator must actually exercise sharing. + assert!(shared_any > 10, "only {shared_any}/60 sets shared anything"); +}