diff --git a/.claude/board/STATUS_BOARD.md b/.claude/board/STATUS_BOARD.md index d0afa30fb..c5dad10ff 100644 --- a/.claude/board/STATUS_BOARD.md +++ b/.claude/board/STATUS_BOARD.md @@ -7,7 +7,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 | 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-3** | cross-rule CSE in crepe-plan-probe (256-rule benchmark, separate from the hop benchmark) | MEASURED (`crepe-plan-probe/src/cse.rs`, `examples/rule_cse.rs`; `entries/2026-10-21-u3-cross-rule-cse.md`): 256 rules, 90 % shared core, 200k centers: rows laid out 15.3 M → 6.2 M, prepare 1.78 s → 1.22 s; lets cut ops 512 → 283 but folds are ≤ 2.4 ms and `share` costs 7 ms; one table for all rules is slower than per-rule when nothing is shared, so tables are per group of rules sharing a side; answers equal crepe, ascent and the reference | 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 | | **D-RU-6** | recursion driver over BIND tiles (256-hop benchmark) | Queued/Lab (slot assignment must be argued as an attested frozen-codebook lookup under ADMISSION = B). Lab probe MEASURED (`crepe-plan-probe/examples/walk_tile.rs`, `entries/2026-10-21-d-ru-6-tile-walk-probe.md`): tuple sets equal ascent, crepe and walk_mask; execution gather-free; binding witness probe-local, so ADMISSION B is NOT demonstrated; count-only 1.0–9.3× slower than walk_mask, end to end 1.1–2.0× slower. Block-sparse arm C MEASURED (`examples/walk_sparse.rs`, `entries/2026-10-21-d-ru-6-sparse-compute.md`): faster than the dense tile path on every fixture (0.11–0.96× count-only); against walk_mask end to end it wins only on deep low-fan-out recursion (chain(230) 0.82×, chain(700) 0.59× UNCHECKED), is near parity on sparse graphs (1.03–1.07×) and loses on dense, cyclic, clustered and mixed-length graphs (1.20–2.43×) | least fixpoint == datafrog; ≥ 200 iterations asserted; cycles terminate. Benchmark requirement (operator, 2026-10-21; refs D-RMAP-2 §5b, #1642): tuple materialization, count-only evaluation, memory growth, address binding and value-derived reads by origin reported separately; tuple-set equality with crepe and ascent wherever they run | diff --git a/.claude/board/entries/2026-10-21-u3-cross-rule-cse.md b/.claude/board/entries/2026-10-21-u3-cross-rule-cse.md new file mode 100644 index 000000000..a2184bb8d --- /dev/null +++ b/.claude/board/entries/2026-10-21-u3-cross-rule-cse.md @@ -0,0 +1,65 @@ +# 2026-10-21 — D-RU-3 (U3): cross-rule CSE in crepe-plan-probe + +**Status:** MEASURED (lab; `crates/crepe-plan-probe/src/cse.rs`, `examples/rule_cse.rs`). Plan: `.claude/plans/d-relalg-unified-v1.md` §7, §12, stage table (U3). Builds on U2 (`contract::set_program`, `relalg::run_program`, #1649). + +## What landed + +- `cse::evaluate_shared(rules, demand, kind, opts, store, layout, sharing)`. It makes the same rewrites as `pipeline::evaluate` and gives the same answer. Per group of rules it does three things: + - lays out each distinct `SideKey` once, over one domain; + - writes every rule as `And(Sym(side))`; + - with `sharing` on, lifts shared parts into lets (`set_program::share`) and runs the group through `relalg::run_program`. +- `Layout::Components`: rules that share a side, directly or through other rules, share a table. A rule that shares nothing keeps a table of its own. `Layout::OneTable` puts every rule into one table. +- Tuples are read per rule by `native::tuples_over_mask`. This is the per-rule path's own read, moved into a helper. The 41 existing tests are unchanged. +- Refusals: a stale side (`JoinRefusal::Stale`), a generation vector of the wrong shape, and a program for another version. +- The shared path ignores `Options::semijoin`. A semijoin's reduced domain is per rule. + +## Measured + +Workload: 256 rules `Out(k,x,z) <- R_g(x,c), S_g(c,z), T(c,k)`. The first `p · 256` rules use core 0; every other rule has a core of its own. Release build, 4 cores, medians. +- Arms: A is the per-rule pipeline; U is one table with lets; D0 is components without lets; D is components with lets. +- `exists` runs only the folds. `bindings` reads the full bag. + +200,000 centers, 20,000 rows per core relation (3 reps): + +| core shared | arm | prepare | rows prepared | folds / ops | fold time | share (compile) | tuples | +|---|---|---|---|---|---|---|---| +| 0 % | A | 1851 ms | 15.3 M | 256 / 512 | — | — | 132 ms | +| 0 % | U | 2721 ms | 15.3 M | 256 / 512 | 2.38 ms | 1.1 ms | 259 ms | +| 0 % | D0 | 1794 ms | 15.3 M | 256 / 512 | 0.86 ms | — | 130 ms | +| 90 % | A | 1777 ms | 15.3 M | 256 / 512 | — | — | 142 ms | +| 90 % | D0 | 1284 ms | 6.2 M | 256 / 512 | 1.39 ms | — | 175 ms | +| 90 % | D | 1218 ms | 6.2 M | 257 / 283 | 1.18 ms | 7.1 ms | 169 ms | + +At 4,000 centers (5 reps) the pattern is the same: +- 90 %: prepare A 197 ms, D0 35 ms, D 36 ms; +- `exists` folds: D0 0.44 ms, D 0.41 ms; +- `share` 7.7 ms. + +**Findings.** +- **The win is the layout, not the fold.** At 90 % the shared core is laid out once instead of 230 times: 15.3 M rows become 6.2 M, and prepare falls 1.78 s → 1.22 s. That is −31 % end to end for `exists` and −27 % for `bindings`. +- **Lets cut ops 512 → 283, but fold time is negligible here** (≤ 2.4 ms against ≥ 1.2 s per run). `share` costs 7–8 ms of compile time at 90 %, more than the folds it saves. The pairwise subset search is quadratic in the number of rules. +- **One table for everything is wrong.** Every side is laid out densely over its table's domain. With nothing shared, U is slower than A: prepare 2.72 s vs 1.85 s, tuples 259 ms vs 132 ms. That is why `Layout::Components` exists. With it, D0 at 0 % lays out exactly A's rows, one table per rule. +- **A shared group pays width.** At 90 % the core-0 group spans almost all centers, so tuple reads are 175 ms vs 142 ms for A. The layout saving outweighs that. + +**Correctness.** +- Every arm, under both layouts, equals: + - the reference evaluator (bags); + - `crepe!` and `ascent!` (sets; the 256 rules written as one parametrized rule); + - arm A. +- Checked at 0/50/90/100 % shared in the tests, and at 0/50/90 % at 4,000 centers in the benchmark. +- Bound demands go through the post filter, with and without propagation. +- **Disables:** seven, all red: + - sharing ignored; + - wrong table side read; + - generation check removed; + - keys not deduplicated; + - post filter dropped; + - components never joined; + - components always one table. +- The post-filter disable first did not compile. The suite had no bound demand, so a test was added first. + +## Open + +- `share`'s subset search costs more than it saves when planes are this small. Two possible fixes, neither measured: a cheaper share for star rules (group by identical core keys), or skipping lets when one plane is cheaper than the search. +- Factorization (plan arm E) and a shared BIND (arm F) are not built. +- The JC batch through the same lets (arm H) is not built. diff --git a/.claude/board/entries/README.md b/.claude/board/entries/README.md index 3a29ff226..5dc69dd98 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. -403 entries, 2026-08-06 .. 2026-10-21. +404 entries, 2026-08-06 .. 2026-10-21. | date | entry id | finding | file | |---|---|---|---| +| 2026-10-21 | `D-RU-3` | | [2026-10-21-u3-cross-rule-cse.md](2026-10-21-u3-cross-rule-cse.md) | | 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) | diff --git a/.claude/plans/d-relalg-unified-v1.md b/.claude/plans/d-relalg-unified-v1.md index 3b5d03da8..ce31e40c2 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 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. +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. **U3 MEASURED (2026-10-21, `crepe-plan-probe/src/cse.rs` + `examples/rule_cse.rs`, `entries/2026-10-21-u3-cross-rule-cse.md`):** the §7 claim "one core join plus 256 ANDs" holds for the layout, not for the folds. At 90 % shared core the core is laid out once (15.3 M → 6.2 M rows, prepare 1.78 s → 1.22 s); the lets cut ops 512 → 283, but the folds cost ≤ 2.4 ms and the subset search 7 ms. One table for all rules is slower than per-rule tables when nothing is shared, so tables are formed per group of rules that share a side. 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 172f8d65e..030c45173 100644 --- a/.github/workflows/crepe-plan-probe.yml +++ b/.github/workflows/crepe-plan-probe.yml @@ -59,6 +59,10 @@ jobs: # D-RU-6 lab probe: the same program over RMAP tiles, its oracle # comparisons and refusals. Its benchmark is not run in CI either. - run: cargo test --example walk_tile + # D-RU-3 (U3): cross-rule CSE; its oracle comparisons and refusals. + # The benchmark (`cargo run --release --example rule_cse`) is not run + # in CI. + - run: cargo test --example rule_cse # D-RU-6 arm C: the block-sparse product, against arms A and B and the # oracles, its refusals and the semi-naive bound. - run: cargo test --example walk_sparse diff --git a/crates/crepe-plan-probe/examples/rule_cse.rs b/crates/crepe-plan-probe/examples/rule_cse.rs new file mode 100644 index 000000000..38758a2d2 --- /dev/null +++ b/crates/crepe-plan-probe/examples/rule_cse.rs @@ -0,0 +1,631 @@ +//! D-RU-3 (U3): cross-rule CSE over 256 rules, against today's per-rule +//! pipeline. Run from this crate's directory: +//! +//! ```text +//! cargo run --release --example rule_cse # the benchmark +//! cargo test --example rule_cse # the correctness fixtures +//! ``` +//! +//! # The rules +//! +//! Rule `k` (one per `k` in `0..RULES`): +//! +//! ```text +//! Out(k, x, z) <- R_g(x, c), S_g(c, z), T(c, k) with g = group(k) +//! ``` +//! +//! A *shared-core rate* `p` decides `group`: the first `p · RULES` rules use +//! group 0, every other rule has a core of its own. The rules are a star on +//! `c`; `T(c, k)` is a side restricted to `k`, so no two rules have the same +//! third side. Written as one parametrized rule, the same program is +//! +//! ```text +//! Out(k, x, z) <- G(k, g), RG(g, x, c), SG(g, c, z), T(c, k) +//! ``` +//! +//! which is what `crepe!` and `ascent!` evaluate as oracles. +//! +//! # Arms (plan §12) +//! +//! * **A** — [`pipeline::evaluate`]: one table and one fold per rule. +//! * **U** — [`cse::evaluate_shared`] with [`Layout::OneTable`] and lets: one +//! table for all rules, over the union of all centers. Kept as the measured +//! counter-example: when rules share nothing it is slower than A. +//! * **D0** — [`Layout::Components`] without lets: one table per group of +//! rules that share a side, +//! still one fold per rule. +//! * **D** — the same with lets: each part two or more rules share is ANDed +//! once into a plane. +//! +//! Every arm runs with binding propagation and dedup on and semijoin off, +//! because the shared path cannot apply a per-rule semijoin. +//! +//! Two answers are timed. `Exists` is mask-only: it isolates the folds. +//! `Bindings(Bag)` is the full answer with multiplicities; it adds reading +//! the tuples, which is the same work in every arm. + +use std::collections::BTreeSet; +use std::time::Duration; + +use crepe::crepe; +use crepe_plan_probe::cse::{Layout, SharedReport, evaluate_shared}; +use crepe_plan_probe::logic::{Atom, RelId, Rule, Term, VarId}; +use crepe_plan_probe::native::Store; +use crepe_plan_probe::optimize::{Options, ResultKind}; +use crepe_plan_probe::pipeline::{Outcome, Report, demand, evaluate}; +use mork_pathmap_probe::Semantics; + +/// Rewrites on, semijoin off: the arms differ only in sharing. +const OPTS: Options = Options { + propagate: true, + dedup: true, + semijoin: false, +}; +const REPS: usize = 5; + +/// One core: `R_g` (x, c) rows and `S_g` (c, z) rows. +type Core = (Vec<[u32; 2]>, Vec<[u32; 2]>); + +/// A generated workload: relations and rules. +pub(crate) struct Workload { + /// `rel[2g]` is `R_g` (x, c); `rel[2g + 1]` is `S_g` (c, z). + pub cores: Vec, + /// `T(c, k)`. + pub t: Vec<[u32; 2]>, + /// `group[k]`: the core rule `k` uses. + pub group: Vec, + pub rules: Vec, +} + +impl Workload { + fn t_rel(&self) -> RelId { + RelId(2 * self.cores.len() as u16) + } + fn out_rel(&self) -> RelId { + RelId(2 * self.cores.len() as u16 + 1) + } + + pub fn store(&self) -> Store<'_> { + let mut relations = Vec::new(); + for (g, (r, s)) in self.cores.iter().enumerate() { + relations.push((RelId(2 * g as u16), r.as_slice(), 1)); + relations.push((RelId(2 * g as u16 + 1), s.as_slice(), 1)); + } + relations.push((self.t_rel(), self.t.as_slice(), 1)); + Store { relations } + } +} + +/// SplitMix64, so fixtures are deterministic. +fn mix(mut z: u64) -> u64 { + z = z.wrapping_add(0x9E37_79B9_7F4A_7C15); + z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9); + z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); + z ^ (z >> 31) +} + +/// `rules` rules over `centers` centers, `rows` rows per core relation; +/// `shared` of the rules use core 0. +pub(crate) fn workload(rules: usize, shared: usize, centers: u32, rows: usize) -> Workload { + let mut group = Vec::with_capacity(rules); + let mut next = 1; + for k in 0..rules { + if k < shared { + group.push(0); + } else { + group.push(next); + next += 1; + } + } + let ngroups = next.max(1); + let rel = |g: usize, s: u64| -> Vec<[u32; 2]> { + let mut v: Vec<[u32; 2]> = (0..rows as u64) + .map(|i| { + let h = mix(s ^ ((g as u64) << 32) ^ i); + [(h % 50) as u32, ((h >> 20) % u64::from(centers)) as u32] + }) + .collect(); + v.sort_unstable(); + v.dedup(); + v + }; + let cores = (0..ngroups) + .map(|g| { + let r = rel(g, 1); + let s = rel(g, 2).into_iter().map(|[z, c]| [c, z]).collect(); + (r, s) + }) + .collect(); + let mut t = Vec::new(); + for c in 0..centers { + for k in 0..rules as u32 { + if mix((u64::from(c) << 32) | u64::from(k)).is_multiple_of(10) { + t.push([c, k]); + } + } + } + let mut w = Workload { + cores, + t, + group, + rules: Vec::new(), + }; + let (x, c, z) = (VarId(0), VarId(1), VarId(2)); + let t_rel = w.t_rel(); + let out = w.out_rel(); + w.rules = (0..rules) + .map(|k| { + let g = w.group[k] as u16; + Rule { + head: Atom { + rel: out, + terms: vec![Term::Const(k as u64), Term::Var(x), Term::Var(z)], + }, + body: vec![ + Atom { + rel: RelId(2 * g), + terms: vec![Term::Var(x), Term::Var(c)], + }, + Atom { + rel: RelId(2 * g + 1), + terms: vec![Term::Var(c), Term::Var(z)], + }, + Atom { + rel: t_rel, + terms: vec![Term::Var(c), Term::Const(k as u64)], + }, + ], + } + }) + .collect(); + w +} + +crepe! { + @input + struct G(u32, u32); + @input + struct RG(u32, u32, u32); + @input + struct SG(u32, u32, u32); + @input + struct T(u32, u32); + @output + struct Out(u32, u32, u32); + Out(k, x, z) <- G(k, g), RG(g, x, c), SG(g, c, z), T(c, k); +} + +ascent::ascent! { + struct CseAscent; + relation g(u32, u32); + relation rg(u32, u32, u32); + relation sg(u32, u32, u32); + relation t(u32, u32); + relation out(u32, u32, u32); + out(k, x, z) <-- g(k, gg), rg(gg, x, c), sg(gg, c, z), t(c, k); +} + +type Tuples = BTreeSet<(u32, u32, u32)>; + +pub(crate) fn crepe_answer(w: &Workload) -> Tuples { + let mut rt = Crepe::new(); + rt.extend( + w.group + .iter() + .enumerate() + .map(|(k, &g)| G(k as u32, g as u32)), + ); + for (g, (r, s)) in w.cores.iter().enumerate() { + rt.extend(r.iter().map(|&[x, c]| RG(g as u32, x, c))); + rt.extend(s.iter().map(|&[c, z]| SG(g as u32, c, z))); + } + rt.extend(w.t.iter().map(|&[c, k]| T(c, k))); + let (out,) = rt.run(); + out.into_iter().map(|Out(k, x, z)| (k, x, z)).collect() +} + +pub(crate) fn ascent_answer(w: &Workload) -> Tuples { + let mut prog = CseAscent { + g: w.group + .iter() + .enumerate() + .map(|(k, &g)| (k as u32, g as u32)) + .collect(), + t: w.t.iter().map(|&[c, k]| (c, k)).collect(), + ..CseAscent::default() + }; + for (g, (r, s)) in w.cores.iter().enumerate() { + prog.rg.extend(r.iter().map(|&[x, c]| (g as u32, x, c))); + prog.sg.extend(s.iter().map(|&[c, z]| (g as u32, c, z))); + } + prog.run(); + prog.out.into_iter().collect() +} + +pub(crate) fn as_set(o: &Outcome) -> Tuples { + match o { + Outcome::Bindings(b) => b + .iter() + .map(|(r, _)| (r[0] as u32, r[1] as u32, r[2] as u32)) + .collect(), + other => panic!("expected bindings, got {other:?}"), + } +} + +fn arm_a(w: &Workload, kind: ResultKind) -> (Outcome, Report) { + let d = demand(w.out_rel(), vec![None; 3]); + evaluate(&w.rules, &d, kind, OPTS, &w.store()).expect("arm A") +} + +fn arm_d(w: &Workload, kind: ResultKind, layout: Layout, sharing: bool) -> (Outcome, SharedReport) { + let d = demand(w.out_rel(), vec![None; 3]); + evaluate_shared(&w.rules, &d, kind, OPTS, &w.store(), layout, sharing).expect("arm D") +} + +fn median(mut v: Vec) -> Duration { + v.sort(); + v[v.len() / 2] +} + +fn ms(d: Duration) -> f64 { + d.as_secs_f64() * 1e3 +} + +fn env(name: &str, default: usize) -> usize { + std::env::var(name) + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(default) +} + +fn main() { + let rules = env("BENCH_RULES", 256); + let centers = env("BENCH_CENTERS", 4_000) as u32; + let rows = env("BENCH_ROWS", 8_000); + let reps = env("BENCH_REPS", REPS); + let oracle = env("BENCH_ORACLE", 1) == 1; + println!( + "rules={rules} centers={centers} rows/core relation={rows} reps={reps} avx512f={}", + cfg!(target_feature = "avx512f") + ); + for pct in [0usize, 50, 90] { + let shared = rules * pct / 100; + let w = workload(rules, shared, centers, rows); + let bag = ResultKind::Bindings(Semantics::Bag); + + // Correctness, outside every timed region. + let (a, _) = arm_a(&w, bag); + let (d0, _) = arm_d(&w, bag, Layout::Components, false); + let (d, _) = arm_d(&w, bag, Layout::Components, true); + assert_eq!(a, d0, "D0 vs A"); + assert_eq!(a, d, "D vs A"); + let checked = if oracle { + let set = as_set(&a); + assert_eq!(set, crepe_answer(&w), "A vs crepe"); + assert_eq!(set, ascent_answer(&w), "A vs ascent"); + format!("crepe = ascent = A = D0 = D ({} tuples)", set.len()) + } else { + "A = D0 = D (oracles skipped)".to_string() + }; + println!("\nshared core {pct}% ({shared} of {rules} rules): {checked}"); + println!( + "{:<4} {:>9} {:>10} {:>9} {:>11} {:>8} {:>7} {:>6} {:>10} {:>11}", + "arm", + "kind", + "prepare", + "execute", + "rows prep.", + "folds", + "ops", + "lets", + "planes rd", + "plane bytes" + ); + for (kname, kind) in [("exists", ResultKind::Exists), ("bindings", bag)] { + let mut ta = (Vec::new(), Vec::new()); + let mut last_a = None; + for _ in 0..reps { + let (_, r) = arm_a(&w, kind); + ta.0.push(r.costs.prepare); + ta.1.push(r.costs.execute); + last_a = Some(r); + } + let ra = last_a.expect("reps >= 1"); + println!( + "{:<4} {:>9} {:>8.2}ms {:>7.2}ms {:>11} {:>8} {:>7} {:>6} {:>10} {:>11}", + "A", + kname, + ms(median(ta.0)), + ms(median(ta.1)), + ra.rows_prepared, + ra.programs_run, + ra.programs.iter().map(|p| p.ops.len()).sum::(), + "-", + "-", + 0 + ); + for (arm, layout, sharing) in [ + ("U", Layout::OneTable, true), + ("D0", Layout::Components, false), + ("D", Layout::Components, true), + ] { + let mut t = (Vec::new(), Vec::new()); + let mut split = (Vec::new(), Vec::new(), Vec::new()); + let mut last = None; + for _ in 0..reps { + let (_, r) = arm_d(&w, kind, layout, sharing); + t.0.push(r.prepare); + t.1.push(r.execute); + split.0.push(r.share); + split.1.push(r.fold); + split.2.push(r.tuples); + last = Some(r); + } + let r = last.expect("reps >= 1"); + println!( + "{:<4} {:>9} {:>8.2}ms {:>7.2}ms {:>11} {:>8} {:>7} {:>6} {:>10} {:>11}", + arm, + kname, + ms(median(t.0)), + ms(median(t.1)), + r.rows_prepared, + r.cost.programs, + r.cost.passes, + r.lets, + r.cost.plane_reads, + r.cost.plane_bytes + ); + println!( + " execute = share {:.2}ms + folds {:.2}ms + tuples {:.2}ms", + ms(median(split.0)), + ms(median(split.1)), + ms(median(split.2)) + ); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crepe_plan_probe::cse::{CseRefusal, execute_shared, prepare_shared}; + use crepe_plan_probe::logic::RelId as Rid; + use crepe_plan_probe::pipeline::plan; + use crepe_plan_probe::reference::eval; + use mork_pathmap_probe::join::JoinRefusal; + + const RULES: usize = 24; + const CENTERS: u32 = 40; + const ROWS: usize = 80; + + fn reference_bag(w: &Workload) -> Vec<(Vec, u64)> { + let store = w.store(); + let rows = |r: Rid| -> Vec> { + store + .view(r) + .0 + .iter() + .map(|p| p.iter().map(|&v| u64::from(v)).collect()) + .collect() + }; + eval(&w.rules, w.out_rel(), &[None, None, None], &rows) + .into_iter() + .collect() + } + + /// FAILS IF: the shared table, the lets or the per-rule tuple read change + /// any answer or multiplicity, at any shared-core rate. + #[test] + fn every_arm_matches_the_reference_crepe_and_ascent() { + for shared in [0, RULES / 2, RULES * 9 / 10, RULES] { + let w = workload(RULES, shared, CENTERS, ROWS); + let bag = ResultKind::Bindings(Semantics::Bag); + let (a, _) = arm_a(&w, bag); + let (d0, _) = arm_d(&w, bag, Layout::Components, false); + let (d, _) = arm_d(&w, bag, Layout::Components, true); + let Outcome::Bindings(ab) = &a else { + panic!("bindings") + }; + assert!(ab.len() > 50, "fixture too thin: {} tuples", ab.len()); + assert!( + ab.iter().any(|(_, m)| *m > 1), + "fixture has no multiplicity above 1" + ); + assert_eq!(*ab, reference_bag(&w), "A vs reference, shared={shared}"); + assert_eq!(d0, a, "D0 vs A, shared={shared}"); + assert_eq!(d, a, "D vs A, shared={shared}"); + let set = as_set(&a); + assert_eq!(set, crepe_answer(&w), "crepe, shared={shared}"); + assert_eq!(set, ascent_answer(&w), "ascent, shared={shared}"); + for kind in [ + ResultKind::Exists, + ResultKind::Count(Semantics::Set), + ResultKind::Count(Semantics::Bag), + ] { + let (want, _) = arm_a(&w, kind); + for layout in [Layout::OneTable, Layout::Components] { + assert_eq!(arm_d(&w, kind, layout, true).0, want, "{kind:?}"); + assert_eq!(arm_d(&w, kind, layout, false).0, want, "{kind:?}"); + } + } + } + } + + /// FAILS IF: a demand that propagation does not push into the rules is + /// not applied to the shared path's answers (the post filter), or one + /// it does push changes them. + #[test] + fn a_bound_demand_filters_the_shared_answers() { + let w = workload(RULES, RULES / 2, CENTERS, ROWS); + let bag = ResultKind::Bindings(Semantics::Bag); + let x = w.cores[0].0[0][0]; + for (bound, propagate) in [ + (vec![None, Some(u64::from(x)), None], false), + (vec![None, Some(u64::from(x)), None], true), + (vec![Some(3), None, None], false), + ] { + let opts = Options { propagate, ..OPTS }; + let d = demand(w.out_rel(), bound.clone()); + let (a, _) = evaluate(&w.rules, &d, bag, opts, &w.store()).expect("A"); + let Outcome::Bindings(ab) = &a else { + panic!("bindings") + }; + assert!(!ab.is_empty(), "demand {bound:?} selects nothing"); + let store = w.store(); + let rows = |r: Rid| -> Vec> { + store + .view(r) + .0 + .iter() + .map(|p| p.iter().map(|&v| u64::from(v)).collect()) + .collect() + }; + let want: Vec<(Vec, u64)> = eval(&w.rules, w.out_rel(), &bound, &rows) + .into_iter() + .collect(); + assert_eq!(*ab, want, "A vs reference, {bound:?}"); + for sharing in [false, true] { + let (dd, _) = evaluate_shared( + &w.rules, + &d, + bag, + opts, + &w.store(), + Layout::Components, + sharing, + ) + .expect("D"); + assert_eq!(dd, a, "sharing={sharing}, {bound:?}, propagate={propagate}"); + } + } + } + + /// FAILS IF: a shared core is not lifted, is read once per rule anyway, + /// or the shared table lays the core out once per rule. + #[test] + fn a_shared_core_is_laid_out_once_and_anded_once() { + let shared = RULES * 9 / 10; + let w = workload(RULES, shared, CENTERS, ROWS); + let (_, a) = arm_a(&w, ResultKind::Exists); + let (_, d0) = arm_d(&w, ResultKind::Exists, Layout::Components, false); + let (_, d) = arm_d(&w, ResultKind::Exists, Layout::Components, true); + // One fold per rule in A and D0; D adds one per let. + assert_eq!(a.programs_run, RULES); + assert_eq!(d0.cost.programs, RULES); + assert_eq!(d.lets, 1, "the shared core R_0 ∧ S_0 is one let"); + assert_eq!(d.cost.programs, RULES + 1); + // D0 reads three planes per rule; D reads the core's two once, then + // two per shared rule (let, T) and three per unshared rule. + assert_eq!(d0.cost.plane_reads, 3 * RULES); + assert_eq!(d.cost.plane_reads, 2 + 2 * shared + 3 * (RULES - shared)); + assert_eq!(d.cost.planes_materialized, 1); + // Distinct sides: one core pair per group, one T side per rule. + let groups = 1 + (RULES - shared); + assert_eq!(d0.sides, 2 * groups + RULES); + // One table for the rules on core 0, one per other rule. + assert_eq!(d0.groups, groups); + // A lays the shared core out once per rule that uses it. + let core_rows = w.cores[0].0.len() + w.cores[0].1.len(); + assert!( + a.rows_prepared >= d0.rows_prepared + (shared - 1) * core_rows, + "A {} vs shared {}", + a.rows_prepared, + d0.rows_prepared + ); + } + + /// FAILS IF: lets are formed, or cost changes, when no two rules share + /// a part. + #[test] + fn nothing_is_lifted_when_no_core_is_shared() { + let w = workload(RULES, 0, CENTERS, ROWS); + let (_, d0) = arm_d(&w, ResultKind::Exists, Layout::Components, false); + let (_, d) = arm_d(&w, ResultKind::Exists, Layout::Components, true); + assert_eq!(d.lets, 0); + assert_eq!(d.cost, d0.cost); + // Every rule keeps a table of its own, laid out from exactly the rows + // the per-rule path lays out; one table would widen all of them. + let (_, a) = arm_a(&w, ResultKind::Exists); + assert_eq!(d0.groups, RULES); + assert_eq!(d0.rows_prepared, a.rows_prepared); + let (_, one) = arm_d(&w, ResultKind::Exists, Layout::OneTable, false); + assert_eq!(one.groups, 1); + // In one table every rule's fold runs at the union width. + assert!( + one.domain * RULES > d0.domain, + "union {} x {RULES} rules vs per-rule widths {}", + one.domain, + d0.domain + ); + } + + /// FAILS IF: tables prepared for one plan are run with another plan of + /// the same shape (same rule count, same generations), or with a plan + /// holding one star more or one fewer. + #[test] + fn tables_prepared_for_another_plan_are_refused() { + let kind = ResultKind::Exists; + let w1 = workload(RULES, RULES / 2, CENTERS, ROWS); + let w2 = workload(RULES, RULES / 2 - 1, CENTERS, ROWS); + let d1 = demand(w1.out_rel(), vec![None; 3]); + let d2 = demand(w2.out_rel(), vec![None; 3]); + let p1 = plan(&w1.rules, &d1, kind, OPTS).expect("plan 1"); + let p2 = plan(&w2.rules, &d2, kind, OPTS).expect("plan 2"); + assert_eq!(p1.stars.len(), p2.stars.len(), "same shape"); + let shared = prepare_shared(&p1.stars, &w1.store(), Layout::OneTable); + let gens: Vec> = shared + .groups + .iter() + .map(|g| g.prepared.keys.iter().map(|_| 1).collect()) + .collect(); + assert!(execute_shared(&p1, &shared, kind, &gens, true).is_ok()); + assert_eq!( + execute_shared(&p2, &shared, kind, &gens, true).err(), + Some(CseRefusal::NoCoordinateWitness) + ); + // A star no group covers. + let mut long = p1.clone(); + long.stars.push(long.stars[0].clone()); + long.optimized.rules.push(long.optimized.rules[0].clone()); + assert_eq!( + execute_shared(&long, &shared, kind, &gens, true).err(), + Some(CseRefusal::NoCoordinateWitness) + ); + let mut short = p1.clone(); + short.stars.pop(); + short.optimized.rules.pop(); + assert_eq!( + execute_shared(&short, &shared, kind, &gens, true).err(), + Some(CseRefusal::NoCoordinateWitness) + ); + } + + /// FAILS IF: the shared path runs against a side whose generation moved. + #[test] + fn a_stale_side_is_refused() { + let w = workload(RULES, RULES / 2, CENTERS, ROWS); + let store = w.store(); + let d = demand(w.out_rel(), vec![None; 3]); + let kind = ResultKind::Exists; + let planned = plan(&w.rules, &d, kind, OPTS).expect("plan"); + let shared = prepare_shared(&planned.stars, &store, Layout::Components); + let mut generations: Vec> = shared + .groups + .iter() + .map(|g| g.prepared.keys.iter().map(|_| 1).collect()) + .collect(); + assert!(execute_shared(&planned, &shared, kind, &generations, true).is_ok()); + generations[0][0] = 2; + assert!(matches!( + execute_shared(&planned, &shared, kind, &generations, true), + Err(CseRefusal::Table(JoinRefusal::Stale { .. })) + )); + generations[0][0] = 1; + generations[0].pop(); + assert!(matches!( + execute_shared(&planned, &shared, kind, &generations, true), + Err(CseRefusal::Generations { .. }) + )); + } +} diff --git a/crates/crepe-plan-probe/src/cse.rs b/crates/crepe-plan-probe/src/cse.rs new file mode 100644 index 000000000..70e535758 --- /dev/null +++ b/crates/crepe-plan-probe/src/cse.rs @@ -0,0 +1,434 @@ +//! Cross-rule CSE (D-RU-3 / U3): rules that share a side read one table, and +//! the parts several rules share are evaluated once. +//! +//! STATUS: PROBE (results only), like the rest of this crate. +//! +//! [`crate::pipeline::evaluate`] prepares one table per rule and runs one +//! fold per rule, so a core that 256 rules share is laid out and ANDed 256 +//! times. Here: +//! +//! 1. the stars are grouped ([`Layout`]). With [`Layout::Components`], stars +//! that share a side ([`SideKey`]), directly or through other stars, form +//! one group; a star that shares nothing is a group of its own; +//! 2. each group's distinct sides are laid out once, over one domain: the +//! union of their centers. A rule's joint mask is the AND of its sides' +//! presence planes, and a center a side lacks is absent from that side's +//! plane, so the wider domain changes no rule's mask, only its width; +//! 3. each rule becomes `And(Sym(side), …)` over its group's planes, and +//! [`share`] lifts every part two or more of the group's rules hold into +//! a let (`lance_graph_contract::set_program`); +//! 4. [`run_program`] evaluates each let once into a plane and every rule +//! over the sides plus those planes. The tuples are read per rule with +//! [`tuples_over_mask`], exactly as the per-rule path reads them. +//! +//! Width is the price. A side is laid out densely over its table's domain, +//! so a table pays per slot for every side it holds. [`Layout::OneTable`] +//! puts every side over the union of all centers. Measured in +//! `examples/rule_cse.rs`, that is slower than the per-rule path when the +//! rules share nothing. [`Layout::Components`] pays it only within a group +//! that shares something. +//! +//! With `sharing` off the same tables are used and no let is formed, so the +//! two effects (fewer tables, shared folds) are measured apart. +//! +//! The shared path ignores [`Options::semijoin`]: a semijoin reduces one +//! star's sides to the centers its restricted sides have, which differs per +//! rule and would give each rule its own domain again. + +use std::collections::BTreeMap; +use std::time::{Duration, Instant}; + +use lance_graph_contract::set_algebra::SetExpr; +use lance_graph_contract::set_program::{SetProgram, share}; +use lance_graph_mask_risc::{Planes, words_for}; +use lance_graph_relalg::{Answer, Measure, ProgramCost, ProgramError, run_program}; +use mork_pathmap_probe::Semantics; +use mork_pathmap_probe::join::{JoinRefusal, Side}; + +use crate::logic::{Rule, VarId}; +use crate::native::{PreparedStar, SideKey, SidePlan, StarPlan, Store, prepare, tuples_over_mask}; +use crate::optimize::{Demand, Options, ResultKind}; +use crate::pipeline::{Outcome, PipelineRefusal, Planned, plan}; + +/// How the stars are laid out. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Layout { + /// One table for all stars, over the union of all their centers. + OneTable, + /// One table per group of stars that share a side, directly or through + /// other stars; a star that shares nothing keeps a table of its own, as + /// in the per-rule path. + Components, +} + +/// Stars laid out as tables, each holding the distinct sides of its stars +/// over one domain. +pub struct SharedStars { + pub groups: Vec, +} + +/// One table and the stars that read it. +pub struct Group { + /// The table; its side `k` holds `prepared.keys[k]`. + pub prepared: PreparedStar, + /// The stars of this group, as indices into the planned stars. + pub stars: Vec, + /// For `stars[i]`, the table side of each of its plan sides. + pub rule_sides: Vec>, +} + +/// Star indices grouped by `layout`, each group in star order. +fn partition(stars: &[StarPlan], layout: Layout) -> Vec> { + if layout == Layout::OneTable || stars.is_empty() { + return vec![(0..stars.len()).collect()]; + } + // Union-find over stars, joined through each side key's first holder. + let mut parent: Vec = (0..stars.len()).collect(); + fn root(p: &mut [usize], mut i: usize) -> usize { + while p[i] != i { + p[i] = p[p[i]]; + i = p[i]; + } + i + } + let mut holder: BTreeMap<(u16, u8, Option, bool), usize> = BTreeMap::new(); + for (j, s) in stars.iter().enumerate() { + for side in &s.sides { + let k = side.key; + let key = (k.rel.0, k.key_col, k.restrict, k.unary); + match holder.get(&key) { + Some(&h) => { + let (a, b) = (root(&mut parent, h), root(&mut parent, j)); + parent[a.max(b)] = a.min(b); + } + None => { + holder.insert(key, j); + } + } + } + } + let mut groups: BTreeMap> = BTreeMap::new(); + for j in 0..stars.len() { + let r = root(&mut parent, j); + groups.entry(r).or_default().push(j); + } + groups.into_values().collect() +} + +/// Lays out the distinct sides of each group's stars over the union of that +/// group's centers. +pub fn prepare_shared(stars: &[StarPlan], store: &Store<'_>, layout: Layout) -> SharedStars { + let groups = partition(stars, layout) + .into_iter() + .map(|members| { + let mut keys: Vec = Vec::new(); + let rule_sides = members + .iter() + .map(|&j| { + stars[j] + .sides + .iter() + .map(|side| match keys.iter().position(|k| *k == side.key) { + Some(i) => i, + None => { + keys.push(side.key); + keys.len() - 1 + } + }) + .collect() + }) + .collect(); + // A star whose sides are the distinct keys: `prepare` lays them + // out over one domain with the same per-side restriction as the + // per-rule path. + let all = StarPlan { + center: VarId(0), + sides: keys + .iter() + .map(|&key| SidePlan { key, emit: None }) + .collect(), + head: Vec::new(), + }; + Group { + prepared: prepare(&all, store), + stars: members, + rule_sides, + } + }) + .collect(); + SharedStars { groups } +} + +/// Each star of the group as the conjunction of its table sides, as set +/// expressions over `Sym(side)`. +pub fn rule_sets(group: &Group) -> Vec { + group + .rule_sides + .iter() + .map(|sides| SetExpr::And(sides.iter().map(|&k| SetExpr::Sym(k as u32)).collect())) + .collect() +} + +/// The current generation of every side of every group, in table order. +pub fn generations(shared: &SharedStars, store: &Store<'_>) -> Vec> { + shared + .groups + .iter() + .map(|g| { + g.prepared + .keys + .iter() + .map(|k| store.view(k.rel).1) + .collect() + }) + .collect() +} + +/// The version a program over this table is built for: the generations of +/// its sides, in order. A program for other generations is refused by +/// [`run_program`]. +pub fn table_version(generations: &[u64]) -> u64 { + let mut h = 0xcbf2_9ce4_8422_2325_u64; + for g in generations { + for b in g.to_le_bytes() { + h = (h ^ u64::from(b)).wrapping_mul(0x100_0000_01b3); + } + } + h ^ generations.len() as u64 +} + +/// Why the shared path refused. +#[derive(Debug, Clone, PartialEq)] +pub enum CseRefusal { + Pipeline(PipelineRefusal), + /// A side is stale against the generations the caller passed. + Table(JoinRefusal), + /// The caller passed another number of generation vectors than there + /// are groups, or of generations than a group's table has sides. + Generations { + sides: usize, + given: usize, + }, + Program(ProgramError), + /// The tables were prepared for other stars than the plan holds: a star + /// in no group or in two, or a side the table holds under another key. + NoCoordinateWitness, +} + +/// What one shared evaluation cost, each part on its own, summed over the +/// groups. +#[derive(Debug, Clone, Copy, Default)] +pub struct SharedReport { + /// Tables laid out. + pub groups: usize, + /// Sides laid out (distinct within each group). + pub sides: usize, + /// Slots over all tables' domains: the width every fold of a group runs + /// at, summed. + pub domain: usize, + /// Rows laid out over all sides. + pub rows_prepared: usize, + /// Lets formed. + pub lets: usize, + /// Folds, ops, planes read and materialized. + pub cost: ProgramCost, + pub plan: Duration, + pub prepare: Duration, + /// Building the programs: finding the shared parts (compile time). + pub share: Duration, + /// The folds: lets and rules. + pub fold: Duration, + /// Reading the tuples over the kept masks. + pub tuples: Duration, + /// All of the run half: checks, share, folds, tuples. + pub execute: Duration, +} + +/// The program a group runs: its rules over the table's sides, with the +/// shared parts as lets when `sharing` is on. +pub fn program(group: &Group, version: u64, sharing: bool) -> SetProgram { + let resident = group.prepared.table.sides.len() as u32; + let queries = rule_sets(group); + if sharing { + share(version, resident, &queries) + } else { + SetProgram { + version, + resident, + lets: Vec::new(), + queries, + } + } +} + +/// [`crate::pipeline::evaluate`] over shared tables laid out by `layout`. +/// Same rewrites, same answer; `sharing` decides whether the parts a +/// group's rules share are evaluated once. +pub fn evaluate_shared( + rules: &[Rule], + demand: &Demand, + kind: ResultKind, + opts: Options, + store: &Store<'_>, + layout: Layout, + sharing: bool, +) -> Result<(Outcome, SharedReport), CseRefusal> { + let t = Instant::now(); + let planned = plan(rules, demand, kind, opts).map_err(CseRefusal::Pipeline)?; + let plan_cost = t.elapsed(); + let t = Instant::now(); + let shared = prepare_shared(&planned.stars, store, layout); + let prepare_cost = t.elapsed(); + let gens = generations(&shared, store); + let (outcome, mut report) = execute_shared(&planned, &shared, kind, &gens, sharing)?; + report.plan = plan_cost; + report.prepare = prepare_cost; + Ok((outcome, report)) +} + +/// The coordinate witness: every planned star is in exactly one group, and +/// each of its sides maps to a table side holding exactly that side's key. +/// Tables prepared for another plan are refused before any plane is read. +pub fn witness(planned: &Planned, shared: &SharedStars) -> Result<(), CseRefusal> { + let mut seen = vec![false; planned.stars.len()]; + for g in &shared.groups { + if g.stars.len() != g.rule_sides.len() { + return Err(CseRefusal::NoCoordinateWitness); + } + for (&j, sides) in g.stars.iter().zip(&g.rule_sides) { + let Some(star) = planned.stars.get(j) else { + return Err(CseRefusal::NoCoordinateWitness); + }; + if seen[j] || star.sides.len() != sides.len() { + return Err(CseRefusal::NoCoordinateWitness); + } + seen[j] = true; + for (side, &k) in star.sides.iter().zip(sides) { + if g.prepared.keys.get(k) != Some(&side.key) { + return Err(CseRefusal::NoCoordinateWitness); + } + } + } + } + if seen.iter().all(|&b| b) { + Ok(()) + } else { + Err(CseRefusal::NoCoordinateWitness) + } +} + +/// The run half over prepared shared tables. `generations[g][k]` is the +/// current generation of side `k` of group `g`. +pub fn execute_shared( + planned: &Planned, + shared: &SharedStars, + kind: ResultKind, + generations: &[Vec], + sharing: bool, +) -> Result<(Outcome, SharedReport), CseRefusal> { + let t = Instant::now(); + witness(planned, shared)?; + if generations.len() != shared.groups.len() { + return Err(CseRefusal::Generations { + sides: shared.groups.len(), + given: generations.len(), + }); + } + let filtered = planned.optimized.post_filter.iter().any(Option::is_some); + let mask_only = kind == ResultKind::Exists && !filtered; + let measure = if mask_only { + Measure::Count + } else { + Measure::Keep + }; + let mut report = SharedReport { + groups: shared.groups.len(), + ..SharedReport::default() + }; + let mut any = false; + let mut bag: BTreeMap, u64> = BTreeMap::new(); + for (group, gens) in shared.groups.iter().zip(generations) { + let table = &group.prepared.table; + if gens.len() != table.sides.len() { + return Err(CseRefusal::Generations { + sides: table.sides.len(), + given: gens.len(), + }); + } + table.check(gens).map_err(CseRefusal::Table)?; + let version = table_version(gens); + let ts = Instant::now(); + let prog = program(group, version, sharing); + report.share += ts.elapsed(); + + let n = table.domain.len(); + let masks: Vec<&[u64]> = table.sides.iter().map(Side::present).collect(); + let planes = Planes { + n_rows: n, + masks: &masks, + lanes: &[], + }; + let rules = prog.queries.len(); + let mut kept: Vec> = if mask_only { + Vec::new() + } else { + vec![vec![0u64; words_for(n)]; rules] + }; + let mut sinks: Vec<&mut [u64]> = kept.iter_mut().map(Vec::as_mut_slice).collect(); + let tf = Instant::now(); + let run = run_program(&prog, version, &planes, &vec![measure; rules], &mut sinks) + .map_err(CseRefusal::Program)?; + report.fold += tf.elapsed(); + + let tt = Instant::now(); + if mask_only { + any |= run.answers.iter().any(|a| *a != Answer::Count(0)); + } else { + for (i, &j) in group.stars.iter().enumerate() { + let sides: Vec<&Side> = group.rule_sides[i] + .iter() + .map(|&k| &table.sides[k]) + .collect(); + let star = &planned.stars[j]; + for (row, m) in tuples_over_mask(&table.domain, &sides, star, &kept[i], true) { + *bag.entry(row).or_insert(0) += m; + } + } + } + report.tuples += tt.elapsed(); + report.sides += table.sides.len(); + report.domain += n; + report.rows_prepared += group.prepared.rows_prepared; + report.lets += prog.lets.len(); + let c = run.cost; + report.cost.programs += c.programs; + report.cost.passes += c.passes; + report.cost.lane_reads += c.lane_reads; + report.cost.plane_reads += c.plane_reads; + report.cost.planes_materialized += c.planes_materialized; + report.cost.plane_bytes += c.plane_bytes; + report.cost.max_scratch_slots = report.cost.max_scratch_slots.max(c.max_scratch_slots); + } + let tt = Instant::now(); + let outcome = if mask_only { + Outcome::Exists(any) + } else { + bag.retain(|row, _| { + row.iter() + .zip(&planned.optimized.post_filter) + .all(|(v, f)| f.is_none_or(|f| f == *v)) + }); + match kind { + ResultKind::Exists => Outcome::Exists(!bag.is_empty()), + ResultKind::Count(Semantics::Set) => Outcome::Count(bag.len() as u64), + ResultKind::Count(Semantics::Bag) => Outcome::Count(bag.values().sum()), + ResultKind::Bindings(Semantics::Set) => { + Outcome::Bindings(bag.into_keys().map(|r| (r, 1)).collect()) + } + ResultKind::Bindings(Semantics::Bag) => Outcome::Bindings(bag.into_iter().collect()), + } + }; + report.tuples += tt.elapsed(); + report.execute = t.elapsed(); + Ok((outcome, report)) +} diff --git a/crates/crepe-plan-probe/src/lib.rs b/crates/crepe-plan-probe/src/lib.rs index 29a35828e..cb75078a4 100644 --- a/crates/crepe-plan-probe/src/lib.rs +++ b/crates/crepe-plan-probe/src/lib.rs @@ -24,11 +24,14 @@ //! - [`addressed`]: arm B, the addressed semijoin. Each relation is a lane //! in key order, attested with `SealedFacetLane`; a side reads only the //! ranges of its demanded keys. +//! - [`cse`]: every rule's star over one shared table, the parts several +//! rules share evaluated once (D-RU-3 / U3). //! 6. [`reference`]: an independent nested-loop evaluator over the resolved, //! unoptimized rules. The unmodified `crepe!` evaluator, datafrog and MORK //! are the other oracles (tests only). pub mod addressed; +pub mod cse; pub mod lance_store; pub mod logic; pub mod mork_api; diff --git a/crates/crepe-plan-probe/src/native.rs b/crates/crepe-plan-probe/src/native.rs index 37695486a..004a6d09a 100644 --- a/crates/crepe-plan-probe/src/native.rs +++ b/crates/crepe-plan-probe/src/native.rs @@ -28,7 +28,7 @@ use lance_graph_mask_risc::{MaskOp, Planes, Pred, Program, Terminal, words_for}; use lance_graph_relalg::{Answer, Measure, Query, Rel, RelError, Run, run_with_program}; use mork_pathmap_probe::RelationView; -use mork_pathmap_probe::join::{JoinRefusal, JoinTable, Side, for_each_set}; +use mork_pathmap_probe::join::{JoinDomain, JoinRefusal, JoinTable, Side, for_each_set}; use crate::logic::{RelId, VarId}; use crate::mork_api::{Expr, QueryMulti}; @@ -407,12 +407,30 @@ pub fn bindings_run( let n = prep.table.domain.len(); let mut mask = vec![0u64; words_for(n)]; let (run, program) = joint(prep, plan, current, Measure::Keep, Some(&mut mask))?; + let sides: Vec<&Side> = prep.table.sides.iter().collect(); + let rows = tuples_over_mask(&prep.table.domain, &sides, plan, &mask, bag); + Ok((rows, run, program)) +} + +/// The output tuples of `plan` at the slots `mask` names, with +/// multiplicities. `sides[i]` is the laid-out side for `plan.sides[i]`, over +/// `domain`. Each side is read at the slot the mask names, never at a +/// position taken from another side's values. Without `bag`, every +/// multiplicity is 1. +pub fn tuples_over_mask( + domain: &JoinDomain, + sides: &[&Side], + plan: &StarPlan, + mask: &[u64], + bag: bool, +) -> Vec<(Vec, u64)> { + let n = domain.len(); let mut out: Vec<(Vec, u64)> = Vec::new(); - for_each_set(&mask, n, |slot| { - let center = u64::from(prep.table.domain.id(slot)); + for_each_set(mask, n, |slot| { + let center = u64::from(domain.id(slot)); let mut tuples: Vec<(Vec>, u64)> = vec![(vec![None; plan.sides.len()], 1)]; for (i, s) in plan.sides.iter().enumerate() { - let vals = prep.table.sides[i].at(slot); + let vals = sides[i].at(slot); if s.emit.is_some() { tuples = tuples .into_iter() @@ -457,7 +475,7 @@ pub fn bindings_run( e.1 = 1; } } - Ok((merged, run, program)) + merged } /// True when the program contains an op that reads at a position taken from