From 761590cf6f7bd6aeec8d2330a2064dcd097b384b Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 02:03:31 +0000 Subject: [PATCH 01/10] U3 (D-RU-3): cross-rule CSE in crepe-plan-probe cse.rs lays out the distinct sides of all rules' stars once, over one shared domain, writes each rule as And(Sym(side)), lifts parts two or more rules share into lets (contract::set_program::share) and runs them through relalg::run_program. The tuples are read per rule with native::tuples_over_mask, the per-rule pipeline's own read, moved into a helper (behaviour unchanged: the existing 41 tests pass). examples/rule_cse.rs: 256 rules Out(k,x,z) <- R_g(x,c), S_g(c,z), T(c,k) at a shared-core rate of 0/50/90 %. Arms A (per-rule pipeline), D0 (one table, no lets), D (one table, lets). Tests: every arm equals the reference evaluator (bags), crepe! and ascent! (sets), at 0, 50, 90 and 100 % shared; the shared core is one let and planes read are pinned; nothing is lifted at 0 %; a stale side is refused. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f --- crates/crepe-plan-probe/examples/rule_cse.rs | 496 +++++++++++++++++++ crates/crepe-plan-probe/src/cse.rs | 270 ++++++++++ crates/crepe-plan-probe/src/lib.rs | 3 + crates/crepe-plan-probe/src/native.rs | 28 +- 4 files changed, 792 insertions(+), 5 deletions(-) create mode 100644 crates/crepe-plan-probe/examples/rule_cse.rs create mode 100644 crates/crepe-plan-probe/src/cse.rs 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..30f81b85c --- /dev/null +++ b/crates/crepe-plan-probe/examples/rule_cse.rs @@ -0,0 +1,496 @@ +//! 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. +//! * **D0** — [`cse::evaluate_shared`] without lets: one table for all rules, +//! 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::{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; + +/// 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<(Vec<[u32; 2]>, Vec<[u32; 2]>)>, + /// `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)) % 10 == 0 { + 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::default(); + prog.g = w + .group + .iter() + .enumerate() + .map(|(k, &g)| (k as u32, g as u32)) + .collect(); + 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.t = w.t.iter().map(|&[c, k]| (c, k)).collect(); + 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, sharing: bool) -> (Outcome, SharedReport) { + let d = demand(w.out_rel(), vec![None; 3]); + evaluate_shared(&w.rules, &d, kind, OPTS, &w.store(), 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, false); + let (d, _) = arm_d(&w, bag, 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, sharing) in [("D0", false), ("D", true)] { + let mut t = (Vec::new(), Vec::new()); + let mut last = None; + for _ in 0..reps { + let (_, r) = arm_d(&w, kind, sharing); + t.0.push(r.prepare); + t.1.push(r.execute); + 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 + ); + } + } + } +} + +#[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, false); + let (d, _) = arm_d(&w, bag, 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); + assert_eq!(arm_d(&w, kind, true).0, want, "{kind:?}"); + assert_eq!(arm_d(&w, kind, false).0, want, "{kind:?}"); + } + } + } + + /// 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, false); + let (_, d) = arm_d(&w, ResultKind::Exists, 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); + // 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, false); + let (_, d) = arm_d(&w, ResultKind::Exists, true); + assert_eq!(d.lets, 0); + assert_eq!(d.cost, d0.cost); + } + + /// 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); + let mut generations: Vec = shared.prepared.keys.iter().map(|_| 1).collect(); + assert!(execute_shared(&planned, &shared, kind, &generations, true).is_ok()); + generations[0] = 2; + assert!(matches!( + execute_shared(&planned, &shared, kind, &generations, true), + Err(CseRefusal::Table(JoinRefusal::Stale { .. })) + )); + generations.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..049966150 --- /dev/null +++ b/crates/crepe-plan-probe/src/cse.rs @@ -0,0 +1,270 @@ +//! Cross-rule CSE (D-RU-3 / U3): every rule's star over ONE shared table, +//! and the parts several rules share 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 distinct sides ([`SideKey`]) of all stars 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; a center outside some side's domain is +//! absent from that side's plane, so the union domain changes no rule's +//! mask, only its width; +//! 2. each rule becomes `And(Sym(side), …)` over those planes, and +//! [`share`] lifts every part two or more rules hold into a let +//! (`lance_graph_contract::set_program`); +//! 3. [`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. +//! +//! With `sharing` off the same table is used and no let is formed, so the +//! two effects (one table, 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}; + +/// Every distinct side of a set of stars, laid out once over one domain. +pub struct SharedStars { + /// The table; its side `k` holds `keys[k]`. + pub prepared: PreparedStar, + /// For star `j`, the table side of each of its plan sides. + pub rule_sides: Vec>, +} + +/// Lays out the distinct sides of `stars` over the union of their centers. +pub fn prepare_shared(stars: &[StarPlan], store: &Store<'_>) -> SharedStars { + let mut keys: Vec = Vec::new(); + let rule_sides = stars + .iter() + .map(|s| { + s.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(), + }; + SharedStars { + prepared: prepare(&all, store), + rule_sides, + } +} + +/// Each star as the conjunction of its table sides, as set expressions over +/// `Sym(side)`. +pub fn rule_sets(shared: &SharedStars) -> Vec { + shared + .rule_sides + .iter() + .map(|sides| SetExpr::And(sides.iter().map(|&k| SetExpr::Sym(k as u32)).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 generations than the table has + /// sides. + Generations { + sides: usize, + given: usize, + }, + Program(ProgramError), +} + +/// What one shared evaluation cost, each part on its own. +#[derive(Debug, Clone, Copy, Default)] +pub struct SharedReport { + /// Sides laid out (distinct across all rules). + pub sides: usize, + /// Slots of the shared domain. + 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, + pub execute: Duration, +} + +/// The program the shared path runs: the rules over the table's sides, with +/// the shared parts as lets when `sharing` is on. +pub fn program(shared: &SharedStars, version: u64, sharing: bool) -> SetProgram { + let resident = shared.prepared.table.sides.len() as u32; + let queries = rule_sets(shared); + if sharing { + share(version, resident, &queries) + } else { + SetProgram { + version, + resident, + lets: Vec::new(), + queries, + } + } +} + +/// [`crate::pipeline::evaluate`] over one shared table. Same rewrites, same +/// answer; `sharing` decides whether shared parts are evaluated once. +pub fn evaluate_shared( + rules: &[Rule], + demand: &Demand, + kind: ResultKind, + opts: Options, + store: &Store<'_>, + 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); + let prepare_cost = t.elapsed(); + let generations: Vec = shared + .prepared + .keys + .iter() + .map(|k| store.view(k.rel).1) + .collect(); + let (outcome, mut report) = execute_shared(&planned, &shared, kind, &generations, sharing)?; + report.plan = plan_cost; + report.prepare = prepare_cost; + Ok((outcome, report)) +} + +/// The run half over a prepared shared table. `generations[k]` is the +/// current generation of table side `k`. +pub fn execute_shared( + planned: &Planned, + shared: &SharedStars, + kind: ResultKind, + generations: &[u64], + sharing: bool, +) -> Result<(Outcome, SharedReport), CseRefusal> { + let t = Instant::now(); + let table = &shared.prepared.table; + if generations.len() != table.sides.len() { + return Err(CseRefusal::Generations { + sides: table.sides.len(), + given: generations.len(), + }); + } + table.check(generations).map_err(CseRefusal::Table)?; + let version = table_version(generations); + let prog = program(shared, version, sharing); + + 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 filtered = planned.optimized.post_filter.iter().any(Option::is_some); + let mask_only = kind == ResultKind::Exists && !filtered; + let rules = prog.queries.len(); + let measure = if mask_only { + Measure::Count + } else { + Measure::Keep + }; + 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 run = run_program(&prog, version, &planes, &vec![measure; rules], &mut sinks) + .map_err(CseRefusal::Program)?; + + let outcome = if mask_only { + Outcome::Exists(run.answers.iter().any(|a| *a != Answer::Count(0))) + } else { + let mut bag: BTreeMap, u64> = BTreeMap::new(); + for (j, star) in planned.stars.iter().enumerate() { + let sides: Vec<&Side> = shared.rule_sides[j] + .iter() + .map(|&k| &table.sides[k]) + .collect(); + for (row, m) in tuples_over_mask(&table.domain, &sides, star, &kept[j], true) { + *bag.entry(row).or_insert(0) += m; + } + } + 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()), + } + }; + let report = SharedReport { + sides: table.sides.len(), + domain: n, + rows_prepared: shared.prepared.rows_prepared, + lets: prog.lets.len(), + cost: run.cost, + execute: t.elapsed(), + ..SharedReport::default() + }; + 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 From 54e402cf8ee97e4641b95ecf2297edef86abf83b Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 02:03:59 +0000 Subject: [PATCH 02/10] rule_cse: clippy (type alias, is_multiple_of, struct init) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f --- crates/crepe-plan-probe/examples/rule_cse.rs | 24 ++++++++++++-------- 1 file changed, 14 insertions(+), 10 deletions(-) diff --git a/crates/crepe-plan-probe/examples/rule_cse.rs b/crates/crepe-plan-probe/examples/rule_cse.rs index 30f81b85c..6fee7a488 100644 --- a/crates/crepe-plan-probe/examples/rule_cse.rs +++ b/crates/crepe-plan-probe/examples/rule_cse.rs @@ -59,10 +59,13 @@ const OPTS: Options = Options { }; 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<(Vec<[u32; 2]>, Vec<[u32; 2]>)>, + pub cores: Vec, /// `T(c, k)`. pub t: Vec<[u32; 2]>, /// `group[k]`: the core rule `k` uses. @@ -132,7 +135,7 @@ pub(crate) fn workload(rules: usize, shared: usize, centers: u32, rows: usize) - 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)) % 10 == 0 { + if mix((u64::from(c) << 32) | u64::from(k)).is_multiple_of(10) { t.push([c, k]); } } @@ -218,18 +221,19 @@ pub(crate) fn crepe_answer(w: &Workload) -> Tuples { } pub(crate) fn ascent_answer(w: &Workload) -> Tuples { - let mut prog = CseAscent::default(); - prog.g = w - .group - .iter() - .enumerate() - .map(|(k, &g)| (k as u32, g as u32)) - .collect(); + 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.t = w.t.iter().map(|&[c, k]| (c, k)).collect(); prog.run(); prog.out.into_iter().collect() } From 46cbb1b65c1ac817e066b7b18249f030c100bd0e Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 02:04:53 +0000 Subject: [PATCH 03/10] rule_cse: pin the post filter with bound demands Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f --- crates/crepe-plan-probe/examples/rule_cse.rs | 41 ++++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/crates/crepe-plan-probe/examples/rule_cse.rs b/crates/crepe-plan-probe/examples/rule_cse.rs index 6fee7a488..41f3cec73 100644 --- a/crates/crepe-plan-probe/examples/rule_cse.rs +++ b/crates/crepe-plan-probe/examples/rule_cse.rs @@ -432,6 +432,47 @@ mod tests { } } + /// 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(), 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] From dd06359adcade68bcef18beaf442a654cb919e69 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 02:05:13 +0000 Subject: [PATCH 04/10] CI: run the rule_cse example's tests Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f --- .github/workflows/crepe-plan-probe.yml | 4 ++++ 1 file changed, 4 insertions(+) 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 From 6459ab1609ee5c2717fc7bddce1f31f979e26ddd Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 02:07:38 +0000 Subject: [PATCH 05/10] cse: report share, fold and tuple time separately Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f --- crates/crepe-plan-probe/examples/rule_cse.rs | 10 ++++++++++ crates/crepe-plan-probe/src/cse.rs | 15 +++++++++++++++ 2 files changed, 25 insertions(+) diff --git a/crates/crepe-plan-probe/examples/rule_cse.rs b/crates/crepe-plan-probe/examples/rule_cse.rs index 41f3cec73..836945c0f 100644 --- a/crates/crepe-plan-probe/examples/rule_cse.rs +++ b/crates/crepe-plan-probe/examples/rule_cse.rs @@ -342,11 +342,15 @@ fn main() { ); for (arm, sharing) in [("D0", false), ("D", 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, 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"); @@ -363,6 +367,12 @@ fn main() { 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)) + ); } } } diff --git a/crates/crepe-plan-probe/src/cse.rs b/crates/crepe-plan-probe/src/cse.rs index 049966150..c8be1eae6 100644 --- a/crates/crepe-plan-probe/src/cse.rs +++ b/crates/crepe-plan-probe/src/cse.rs @@ -136,6 +136,13 @@ pub struct SharedReport { pub cost: ProgramCost, pub plan: Duration, pub prepare: Duration, + /// Building the program: 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, } @@ -203,7 +210,9 @@ pub fn execute_shared( } table.check(generations).map_err(CseRefusal::Table)?; let version = table_version(generations); + let ts = Instant::now(); let prog = program(shared, version, sharing); + let share_cost = ts.elapsed(); let n = table.domain.len(); let masks: Vec<&[u64]> = table.sides.iter().map(Side::present).collect(); @@ -226,9 +235,12 @@ pub fn execute_shared( 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)?; + let fold_cost = tf.elapsed(); + let tt = Instant::now(); let outcome = if mask_only { Outcome::Exists(run.answers.iter().any(|a| *a != Answer::Count(0))) } else { @@ -263,6 +275,9 @@ pub fn execute_shared( rows_prepared: shared.prepared.rows_prepared, lets: prog.lets.len(), cost: run.cost, + share: share_cost, + fold: fold_cost, + tuples: tt.elapsed(), execute: t.elapsed(), ..SharedReport::default() }; From 1fee3de09ce8b8b8754b26b05d2e98d62d58b931 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 02:13:50 +0000 Subject: [PATCH 06/10] cse: group rules by shared sides (Layout::Components) The first layout put every side of every rule over the union of all centers. Measured at 200k centers, that made a workload with no shared core slower than the per-rule path (prepare 2.84 s vs 2.09 s, tuples 293 ms vs 145 ms): each side is laid out densely over its table's domain. Layout::Components gives one table per group of rules that share a side, directly or through other rules; a rule that shares nothing keeps a table of its own. Layout::OneTable stays as the measured counter-example (arm U). Tests: every arm under both layouts equals the reference, crepe! and ascent!; at 0 % shared, Components lays out exactly the per-rule rows, one table per rule; at 90 %, one group for core 0 plus one per other rule. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f --- crates/crepe-plan-probe/examples/rule_cse.rs | 83 +++-- crates/crepe-plan-probe/src/cse.rs | 358 ++++++++++++------- 2 files changed, 297 insertions(+), 144 deletions(-) diff --git a/crates/crepe-plan-probe/examples/rule_cse.rs b/crates/crepe-plan-probe/examples/rule_cse.rs index 836945c0f..2c1517407 100644 --- a/crates/crepe-plan-probe/examples/rule_cse.rs +++ b/crates/crepe-plan-probe/examples/rule_cse.rs @@ -28,7 +28,11 @@ //! # Arms (plan §12) //! //! * **A** — [`pipeline::evaluate`]: one table and one fold per rule. -//! * **D0** — [`cse::evaluate_shared`] without lets: one table for all rules, +//! * **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. @@ -44,7 +48,7 @@ use std::collections::BTreeSet; use std::time::Duration; use crepe::crepe; -use crepe_plan_probe::cse::{SharedReport, evaluate_shared}; +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}; @@ -253,9 +257,9 @@ fn arm_a(w: &Workload, kind: ResultKind) -> (Outcome, Report) { evaluate(&w.rules, &d, kind, OPTS, &w.store()).expect("arm A") } -fn arm_d(w: &Workload, kind: ResultKind, sharing: bool) -> (Outcome, SharedReport) { +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(), sharing).expect("arm D") + evaluate_shared(&w.rules, &d, kind, OPTS, &w.store(), layout, sharing).expect("arm D") } fn median(mut v: Vec) -> Duration { @@ -291,8 +295,8 @@ fn main() { // Correctness, outside every timed region. let (a, _) = arm_a(&w, bag); - let (d0, _) = arm_d(&w, bag, false); - let (d, _) = arm_d(&w, bag, true); + 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 { @@ -340,12 +344,16 @@ fn main() { "-", 0 ); - for (arm, sharing) in [("D0", false), ("D", true)] { + 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, sharing); + let (_, r) = arm_d(&w, kind, layout, sharing); t.0.push(r.prepare); t.1.push(r.execute); split.0.push(r.share); @@ -414,8 +422,8 @@ mod tests { 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, false); - let (d, _) = arm_d(&w, bag, true); + 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") }; @@ -436,8 +444,10 @@ mod tests { ResultKind::Count(Semantics::Bag), ] { let (want, _) = arm_a(&w, kind); - assert_eq!(arm_d(&w, kind, true).0, want, "{kind:?}"); - assert_eq!(arm_d(&w, kind, false).0, want, "{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:?}"); + } } } } @@ -476,8 +486,16 @@ mod tests { .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(), sharing).expect("D"); + 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}"); } } @@ -490,8 +508,8 @@ mod tests { 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, false); - let (_, d) = arm_d(&w, ResultKind::Exists, true); + 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); @@ -505,6 +523,8 @@ mod tests { // 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!( @@ -520,10 +540,24 @@ mod tests { #[test] fn nothing_is_lifted_when_no_core_is_shared() { let w = workload(RULES, 0, CENTERS, ROWS); - let (_, d0) = arm_d(&w, ResultKind::Exists, false); - let (_, d) = arm_d(&w, ResultKind::Exists, true); + 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: the shared path runs against a side whose generation moved. @@ -534,15 +568,20 @@ mod tests { 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); - let mut generations: Vec = shared.prepared.keys.iter().map(|_| 1).collect(); + 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] = 2; + generations[0][0] = 2; assert!(matches!( execute_shared(&planned, &shared, kind, &generations, true), Err(CseRefusal::Table(JoinRefusal::Stale { .. })) )); - generations.pop(); + 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 index c8be1eae6..289c94d1c 100644 --- a/crates/crepe-plan-probe/src/cse.rs +++ b/crates/crepe-plan-probe/src/cse.rs @@ -1,5 +1,5 @@ -//! Cross-rule CSE (D-RU-3 / U3): every rule's star over ONE shared table, -//! and the parts several rules share evaluated once. +//! 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. //! @@ -7,20 +7,29 @@ //! fold per rule, so a core that 256 rules share is laid out and ANDed 256 //! times. Here: //! -//! 1. the distinct sides ([`SideKey`]) of all stars 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; a center outside some side's domain is -//! absent from that side's plane, so the union domain changes no rule's -//! mask, only its width; -//! 2. each rule becomes `And(Sym(side), …)` over those planes, and -//! [`share`] lifts every part two or more rules hold into a let -//! (`lance_graph_contract::set_program`); -//! 3. [`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 +//! 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. //! -//! With `sharing` off the same table is used and no let is formed, so the -//! two effects (one table, shared folds) are measured apart. +//! 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 @@ -41,58 +50,140 @@ use crate::native::{PreparedStar, SideKey, SidePlan, StarPlan, Store, prepare, t use crate::optimize::{Demand, Options, ResultKind}; use crate::pipeline::{Outcome, PipelineRefusal, Planned, plan}; -/// Every distinct side of a set of stars, laid out once over one domain. +/// 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 { - /// The table; its side `k` holds `keys[k]`. + 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, - /// For star `j`, the table side of each of its plan sides. + /// 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>, } -/// Lays out the distinct sides of `stars` over the union of their centers. -pub fn prepare_shared(stars: &[StarPlan], store: &Store<'_>) -> SharedStars { - let mut keys: Vec = Vec::new(); - let rule_sides = stars - .iter() - .map(|s| { - s.sides +/// 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(|side| match keys.iter().position(|k| *k == side.key) { - Some(i) => i, - None => { - keys.push(side.key); - keys.len() - 1 - } + .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() + .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(); - // 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(), - }; - SharedStars { - prepared: prepare(&all, store), - rule_sides, - } + SharedStars { groups } } -/// Each star as the conjunction of its table sides, as set expressions over -/// `Sym(side)`. -pub fn rule_sets(shared: &SharedStars) -> Vec { - shared +/// 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`]. @@ -112,8 +203,8 @@ pub enum CseRefusal { Pipeline(PipelineRefusal), /// A side is stale against the generations the caller passed. Table(JoinRefusal), - /// The caller passed another number of generations than the table has - /// sides. + /// 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, @@ -121,12 +212,16 @@ pub enum CseRefusal { Program(ProgramError), } -/// What one shared evaluation cost, each part on its own. +/// What one shared evaluation cost, each part on its own, summed over the +/// groups. #[derive(Debug, Clone, Copy, Default)] pub struct SharedReport { - /// Sides laid out (distinct across all rules). + /// Tables laid out. + pub groups: usize, + /// Sides laid out (distinct within each group). pub sides: usize, - /// Slots of the shared domain. + /// 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, @@ -136,7 +231,7 @@ pub struct SharedReport { pub cost: ProgramCost, pub plan: Duration, pub prepare: Duration, - /// Building the program: finding the shared parts (compile time). + /// Building the programs: finding the shared parts (compile time). pub share: Duration, /// The folds: lets and rules. pub fold: Duration, @@ -146,11 +241,11 @@ pub struct SharedReport { pub execute: Duration, } -/// The program the shared path runs: the rules over the table's sides, with -/// the shared parts as lets when `sharing` is on. -pub fn program(shared: &SharedStars, version: u64, sharing: bool) -> SetProgram { - let resident = shared.prepared.table.sides.len() as u32; - let queries = rule_sets(shared); +/// 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 { @@ -163,97 +258,126 @@ pub fn program(shared: &SharedStars, version: u64, sharing: bool) -> SetProgram } } -/// [`crate::pipeline::evaluate`] over one shared table. Same rewrites, same -/// answer; `sharing` decides whether shared parts are evaluated once. +/// [`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); + let shared = prepare_shared(&planned.stars, store, layout); let prepare_cost = t.elapsed(); - let generations: Vec = shared - .prepared - .keys - .iter() - .map(|k| store.view(k.rel).1) - .collect(); - let (outcome, mut report) = execute_shared(&planned, &shared, kind, &generations, sharing)?; + 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 run half over a prepared shared table. `generations[k]` is the -/// current generation of table side `k`. +/// 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: &[u64], + generations: &[Vec], sharing: bool, ) -> Result<(Outcome, SharedReport), CseRefusal> { let t = Instant::now(); - let table = &shared.prepared.table; - if generations.len() != table.sides.len() { + if generations.len() != shared.groups.len() { return Err(CseRefusal::Generations { - sides: table.sides.len(), + sides: shared.groups.len(), given: generations.len(), }); } - table.check(generations).map_err(CseRefusal::Table)?; - let version = table_version(generations); - let ts = Instant::now(); - let prog = program(shared, version, sharing); - let share_cost = 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 filtered = planned.optimized.post_filter.iter().any(Option::is_some); let mask_only = kind == ResultKind::Exists && !filtered; - let rules = prog.queries.len(); let measure = if mask_only { Measure::Count } else { Measure::Keep }; - let mut kept: Vec> = if mask_only { - Vec::new() - } else { - vec![vec![0u64; words_for(n)]; rules] + let mut report = SharedReport { + groups: shared.groups.len(), + ..SharedReport::default() }; - 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)?; + 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 fold_cost = 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(run.answers.iter().any(|a| *a != Answer::Count(0))) + Outcome::Exists(any) } else { - let mut bag: BTreeMap, u64> = BTreeMap::new(); - for (j, star) in planned.stars.iter().enumerate() { - let sides: Vec<&Side> = shared.rule_sides[j] - .iter() - .map(|&k| &table.sides[k]) - .collect(); - for (row, m) in tuples_over_mask(&table.domain, &sides, star, &kept[j], true) { - *bag.entry(row).or_insert(0) += m; - } - } bag.retain(|row, _| { row.iter() .zip(&planned.optimized.post_filter) @@ -269,17 +393,7 @@ pub fn execute_shared( ResultKind::Bindings(Semantics::Bag) => Outcome::Bindings(bag.into_iter().collect()), } }; - let report = SharedReport { - sides: table.sides.len(), - domain: n, - rows_prepared: shared.prepared.rows_prepared, - lets: prog.lets.len(), - cost: run.cost, - share: share_cost, - fold: fold_cost, - tuples: tt.elapsed(), - execute: t.elapsed(), - ..SharedReport::default() - }; + report.tuples += tt.elapsed(); + report.execute = t.elapsed(); Ok((outcome, report)) } From a337e21aa9d3f03c6c7f486effffad8e55766008 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 02:39:46 +0000 Subject: [PATCH 07/10] U3: board (STATUS_BOARD D-RU-3 MEASURED, entry, plan note, index) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f --- .claude/board/STATUS_BOARD.md | 2 +- .../entries/2026-10-21-u3-cross-rule-cse.md | 65 +++++++++++++++++++ .claude/board/entries/README.md | 1 + .claude/plans/d-relalg-unified-v1.md | 2 +- 4 files changed, 68 insertions(+), 2 deletions(-) create mode 100644 .claude/board/entries/2026-10-21-u3-cross-rule-cse.md 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..f1dde15c8 100644 --- a/.claude/board/entries/README.md +++ b/.claude/board/entries/README.md @@ -29,6 +29,7 @@ exactly one of them, never both. | 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. From b29150d43ad6155f53e1b38edcce54c4dc73c19b Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 07:48:23 +0000 Subject: [PATCH 08/10] cse: refuse tables prepared for another plan (Codex P2) execute_shared now checks a coordinate witness before reading any plane: every planned star is in exactly one group, and each of its sides maps to a table side holding that side's key. This is the shared path's counterpart of PreparedStar::witness. Before, tables prepared from a different plan of the same shape passed the generation checks and silently read unrelated relations. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f --- crates/crepe-plan-probe/examples/rule_cse.rs | 33 ++++++++++++++++++ crates/crepe-plan-probe/src/cse.rs | 35 ++++++++++++++++++++ 2 files changed, 68 insertions(+) diff --git a/crates/crepe-plan-probe/examples/rule_cse.rs b/crates/crepe-plan-probe/examples/rule_cse.rs index 2c1517407..a453dfec8 100644 --- a/crates/crepe-plan-probe/examples/rule_cse.rs +++ b/crates/crepe-plan-probe/examples/rule_cse.rs @@ -560,6 +560,39 @@ mod tests { ); } + /// 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 + /// missing a star. + #[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) + ); + 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() { diff --git a/crates/crepe-plan-probe/src/cse.rs b/crates/crepe-plan-probe/src/cse.rs index 289c94d1c..70e535758 100644 --- a/crates/crepe-plan-probe/src/cse.rs +++ b/crates/crepe-plan-probe/src/cse.rs @@ -210,6 +210,9 @@ pub enum CseRefusal { 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 @@ -283,6 +286,37 @@ pub fn evaluate_shared( 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( @@ -293,6 +327,7 @@ pub fn execute_shared( 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(), From ac583359b28f51be1858281e75a40c3195a6d896 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 07:49:04 +0000 Subject: [PATCH 09/10] rule_cse: pin the witness's coverage check (a star no group covers) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f --- crates/crepe-plan-probe/examples/rule_cse.rs | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/crates/crepe-plan-probe/examples/rule_cse.rs b/crates/crepe-plan-probe/examples/rule_cse.rs index a453dfec8..38758a2d2 100644 --- a/crates/crepe-plan-probe/examples/rule_cse.rs +++ b/crates/crepe-plan-probe/examples/rule_cse.rs @@ -562,7 +562,7 @@ mod tests { /// 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 - /// missing a star. + /// holding one star more or one fewer. #[test] fn tables_prepared_for_another_plan_are_refused() { let kind = ResultKind::Exists; @@ -584,6 +584,14 @@ mod tests { 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(); From 2bfacf94d04dab62fa983e8609658c0e20c80f55 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 07:50:27 +0000 Subject: [PATCH 10/10] Regenerate entries index after rebasing onto main Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f --- .claude/board/entries/README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.claude/board/entries/README.md b/.claude/board/entries/README.md index f1dde15c8..5dc69dd98 100644 --- a/.claude/board/entries/README.md +++ b/.claude/board/entries/README.md @@ -25,7 +25,7 @@ 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 | |---|---|---|---|