diff --git a/.claude/board/entries/2026-10-20-crepe-rule-optimizer-spike.md b/.claude/board/entries/2026-10-20-crepe-rule-optimizer-spike.md index 6ec648b7e..f73f2ba59 100644 --- a/.claude/board/entries/2026-10-20-crepe-rule-optimizer-spike.md +++ b/.claude/board/entries/2026-10-20-crepe-rule-optimizer-spike.md @@ -10,11 +10,12 @@ ## Measured -- 16 differential tests (A–M) against `crepe!`, datafrog, the MORK shape and a nested-loop bag reference; 9 disables each red at their named tests. +- 20 differential tests (A–N) against `crepe!`, datafrog, the MORK shape and a nested-loop bag reference; 9 disables each red at their named tests. - `Answer(17, z)`, 50,000 Parent rows, release: terminal 3.6 µs with rewrites vs 11.8 ms without; prepared rows 56,286 vs 118,802; summed cold cost lower with the rewrites at every size (break-even reuse count 0). Parity ok at 100/1k/10k/50k. - H14d absorption miss holds only when `x` is itself an `And`/`Or` (pinned, test M). ## OPEN - Dynamic MeTTa rules need crepe's analysis as a library: crepe is a proc-macro crate (follow-up A, `crepe-core` split). -- Preparation still reads the unrestricted side in full; semijoin reduction before preparation is the next step. +- Semijoin reduction built (`Options::semijoin`): laid-out rows 36 at every size instead of growing with n; preparation 3.47 ms → 0.28 ms at 50,000. The scan stays O(n). +- ndarray for the row filter, measured (`examples/row_filter.rs`): over the row-major store the strided compare is slower than the scalar test at every needle count, so it is not wired; over a contiguous column it is 5.8 µs per needle at 50,000 rows and beats the scalar test up to 64 needles. Next step: a column-major relation store. diff --git a/.claude/plans/crepe-rule-optimizer-spike-v1.md b/.claude/plans/crepe-rule-optimizer-spike-v1.md index b59ae4dd5..531874eff 100644 --- a/.claude/plans/crepe-rule-optimizer-spike-v1.md +++ b/.claude/plans/crepe-rule-optimizer-spike-v1.md @@ -13,7 +13,7 @@ Pipeline, one stage per module of the probe: |---|---|---| | 1 | crepe fork, `crepe_plan!` | `Context::new` (names, arity, `@input` goals, negation cycles, strata), then crepe's own left-to-right bound/free analysis as data. No `HashSet` store, `HashMap` index or fixpoint loop is emitted. | | 2 | `logic.rs` | labels to ids through the label table; flat, positive, non-recursive rules over `@input` relations, or a named refusal | -| 3 | `optimize.rs` | binding propagation; duplicate-atom and duplicate-rule elimination gated on duplicate-insensitive answers | +| 3 | `optimize.rs` | binding propagation; duplicate-atom and duplicate-rule elimination gated on duplicate-insensitive answers; the `semijoin` switch read at preparation | | 4 | `mork_api.rs` | each rule as a typed `query_multi (pattern, template)` (D-MPS-9 shape); nested, repeated-variable and cyclic shapes refused | | 5 | `native.rs` | a star on one shared variable laid out by the D-MPS-8 table; joint mask = relalg `And` of presence planes, one mask-risc program; terminals read the mask | | 6 | `reference.rs` | nested-loop evaluator over the unoptimized rules (bag) | @@ -27,7 +27,7 @@ Pipeline, one stage per module of the probe: - [x] Phase 4: relalg + mask-risc over the D-MPS-8 table; `relalg::run_with_program` (additive) to inspect the program - [x] Phase 5: differential tests A–M against `crepe!`, datafrog, MORK, reference; 9 disable runs - [x] Phase 6 (limited): Soufflé harvest table (§3) -- [ ] Follow-up (not built): semijoin reduction of the unrestricted sides before preparation (§6) +- [x] Semijoin reduction before preparation (`Options::semijoin`, `native::prepare_semijoin`; §2b, tests N) ## 1. Integration point @@ -65,7 +65,46 @@ Plan construction (`crepe_plan()`) 164–261 ns, resolution 273–458 ns, indepe Reading: - **Break-even reuse count is 0.** The rewrite itself costs ~210–470 ns more, but it removes a rule, so mapping and normalization get cheaper by more than that: summed cold cost (optimize + map + normalize) is lower with the rewrites at every size (2,347 vs 2,603 ns at n = 100; 1,439 vs 1,856 ns at 50,000). - **The win is in preparation and in the terminal, not the mask.** With the constant pushed into the body, the `Parent(17, y)` side is prepared from 8 rows instead of n, and the sponsor branch is never prepared: rows halve. The terminal reads one restricted side instead of enumerating every `x` and filtering: 3.6 µs vs 11.8 ms at 50,000 (≈3,300×). The joint mask itself is ~1 µs either way. -- **Preparation still dominates** (3.5 ms at 50,000) because `Parent(y, z)` is prepared in full. That is the next rewrite (§6), not a physical gap. +- **Preparation still dominates** (3.5 ms at 50,000) because `Parent(y, z)` is prepared in full. §2b removes that. + +## 2b. Semijoin reduction (2026-10-20, same harness) + +When a star side is restricted by a constant, every side keeps only the rows whose center all restricted sides have (sorted `Vec`, binary search). A center outside that set cannot reach the joint mask, so answers and bag multiplicities are unchanged. `no-semijoin` is `Options::ALL` with only the reduction off. + +| n | opts | prepare µs | mask ns | terminal µs | rows prepared | parity | +|---|---|---|---|---|---|---| +| 100 | all | 1.9 | 791 | 2.3 | 36 | ok | +| 100 | no-semijoin | 3.9 | 786 | 2.2 | 152 | ok | +| 1,000 | all | 6.8 | 821 | 2.3 | 36 | ok | +| 1,000 | no-semijoin | 33.6 | 868 | 2.3 | 1,161 | ok | +| 10,000 | all | 76.1 | 1,280 | 3.7 | 36 | ok | +| 10,000 | no-semijoin | 614.0 | 902 | 2.5 | 11,286 | ok | +| 50,000 | all | 277.6 | 819 | 2.4 | 36 | ok | +| 50,000 | no-semijoin | 3,473.2 | 1,070 | 3.3 | 56,286 | ok | + +- The laid-out rows no longer depend on `n` (36 at every size); preparation at 50,000 falls 12.5×. +- **Preparation is still O(n):** every row is still scanned once to test its center (1.9 → 6.8 → 76 → 278 µs). What became O(k) is the layout (domain, presence planes, payload), not the scan. Removing the scan needs an index by key column on the stored relation, which the in-memory `Store` does not have. +- Mask and terminal are unchanged (~1 µs, ~2–3 µs): the reduction does not change the program, only the domain it runs over. +- Interaction: a reduced table's domain holds only the reduced centers, so re-sealing a side with rows for other centers is refused as `OutsideDomain` (safe; prepare again). + +Tests N (4): same answers for set, bag, count and exists with exact prepared-row counts (11 vs 417); no change without a restricted side; an empty restricted side gives 0 rows and no answer; two restricted sides intersect (5 vs 7 rows, bag multiplicity 2 kept). Disables: no reduction → the three reduction tests red; first restricted side only → the intersection test red; filtering by the wrong column → A, B ×2, C, F and both N tests red. + +## 2c. The row filter through ndarray (measured, not wired) + +The semijoin filter (and the constant restriction) is a scalar test per row: binary search of the row's key in the sorted center set. `ndarray::simd` has the compare (`eq_u32_strided_to_mask` over the row bytes, `eq_u32_to_mask` over a contiguous column, OR-ed with `mask_or_assign`), one pass per needle. `cargo run --release --example row_filter`, 50,000 rows, target-cpu=native (avx512f/bw), medians of 31; all three agree on the kept rows: + +| needles | scalar µs | ndarray over the rows µs | ndarray over a contiguous column µs | +|---|---|---|---| +| 1 | 17.4 | 82.4 | 5.8 | +| 8 | 144.4 | 654.8 | 29.6 | +| 32 | 284.9 | 2,869.7 | 120.7 | +| 64 | 329.0 | 5,434.8 | 242.8 | +| 128 | 375.4 | 10,603.4 | 485.2 | + +- Over the store as it is (`[u32; 2]` rows, the D-MPS-8 `RelationView` shape) the strided kernel is slower than the scalar test at every needle count: its field loads are scalar by design, and it pays one pass per needle where the binary search pays one pass in total. It is therefore **not** used in `prepare`. +- Over a contiguous column the same compares are 14–20× faster than the strided ones and beat the scalar test up to 64 needles; at 128 the scalar test wins again (the per-needle passes add up). +- The one-pass alternative, a needle bitmap read through `mask_gather_u32`, is a gather and stays out (2026-10-16 ruling, test H). +- Consequence for the next step: a column-major relation store (one contiguous `u32` lane per column) is what makes ndarray applicable here, and it is the same change the key-column index needs. ## 3. Soufflé harvest (souffle-lang/souffle `f53dab8`, UPL-1.0; algorithms only, no source copied) @@ -86,7 +125,7 @@ Reading: ## 5. Tests and disables -16 tests (`tests/pipeline.rs`, A–M of the brief), all green. Every guard turned its tests red under a disable run (made after commit, restored after): +20 tests (`tests/pipeline.rs`: A–M of the brief plus N for the semijoin, §2b), all green. Every guard turned its tests red under a disable run (made after commit, restored after): | disable | red | |---|---| @@ -108,7 +147,8 @@ Known limits: rule dedup does not search body permutations; the star must have o ## 6. Follow-ups (not built) -- **Next (recommended):** semijoin reduction before preparation — restrict the unrestricted sides of a star to the center values the restricted side has (here: prepare `Parent(y, z)` only for `y ∈ {children of 17}`). Same answers, preparation from O(n) to O(k). Needs a demand-aware layout step, not a new mask op. +- ~~Semijoin reduction before preparation~~ — built, §2b. Layout O(k); the scan stays O(n). +- **Next (recommended):** a column-major relation store (contiguous `u32` lanes). It lets the restriction and semijoin filters run through `ndarray::simd::eq_u32_to_mask` (§2c: 4–6 µs per needle at 50,000 rows instead of a scalar or strided scan), and it is the layout a key-column index would sit on, which then removes the O(n) scan. - A: `crepe-core` library split in the fork for dynamic rules. - B: full magic sets for derived body relations and recursion (waits on D-MTP-3 / D-SRAM-4). - C: cost-informed ordering once a non-star shape is admitted. diff --git a/crates/crepe-plan-probe/Cargo.toml b/crates/crepe-plan-probe/Cargo.toml index b3a0fba2c..eef3e5c9e 100644 --- a/crates/crepe-plan-probe/Cargo.toml +++ b/crates/crepe-plan-probe/Cargo.toml @@ -23,6 +23,10 @@ lance-graph-mask-risc = { path = "../lance-graph-mask-risc" } mork-pathmap-probe = { path = "../mork-pathmap-probe" } [dev-dependencies] +# The SIMD facade, for the row-filter comparison in examples/row_filter.rs. +# Same coordinate as mask-risc and mork-pathmap-probe, so the graph keeps one +# ndarray package. +ndarray = { git = "https://github.com/AdaWorldAPI/ndarray", branch = "master", default-features = false, features = ["std"] } datafrog = { git = "https://github.com/AdaWorldAPI/datafrog", branch = "master" } [patch."https://github.com/AdaWorldAPI/lance-graph"] diff --git a/crates/crepe-plan-probe/examples/measure.rs b/crates/crepe-plan-probe/examples/measure.rs index 4d7b7a0a8..88d5eb691 100644 --- a/crates/crepe-plan-probe/examples/measure.rs +++ b/crates/crepe-plan-probe/examples/measure.rs @@ -17,7 +17,7 @@ use std::time::{Duration, Instant}; use crepe_plan_probe::logic::{Labels, Rule, resolve}; use crepe_plan_probe::mork_api::{admit, query_multi}; -use crepe_plan_probe::native::{StarPlan, Store, bindings, joint, prepare, star}; +use crepe_plan_probe::native::{StarPlan, Store, bindings, joint, prepare, prepare_semijoin, star}; use crepe_plan_probe::optimize::{Demand, Options, ResultKind, optimize}; use crepe_plan_probe::programs::family; use crepe_plan_probe::reference; @@ -128,7 +128,15 @@ fn main() { .collect::>() }); - for (name, opts) in [("all", Options::ALL), ("none", Options::NONE)] { + let no_semijoin = Options { + semijoin: false, + ..Options::ALL + }; + for (name, opts) in [ + ("all", Options::ALL), + ("no-semijoin", no_semijoin), + ("none", Options::NONE), + ] { let optimize_t = time(|| optimize(&rules, &demand, kind, opts)); let opt = optimize(&rules, &demand, kind, opts); let map_t = time(|| { @@ -151,8 +159,15 @@ fn main() { .map(|s| Rel::And((0..s.sides.len() as u16).map(Rel::Plane).collect())) .collect(); let normalize_t = time(|| rels.iter().map(normalize).collect::>()); - let prepare_t = time(|| stars.iter().map(|s| prepare(s, &st)).collect::>()); - let preps: Vec<_> = stars.iter().map(|s| prepare(s, &st)).collect(); + let prep = |s: &StarPlan| { + if opts.semijoin { + prepare_semijoin(s, &st) + } else { + prepare(s, &st) + } + }; + let prepare_t = time(|| stars.iter().map(prep).collect::>()); + let preps: Vec<_> = stars.iter().map(prep).collect(); let gens = |s: &StarPlan| -> Vec { vec![1; s.sides.len()] }; let mask_t = time(|| { stars diff --git a/crates/crepe-plan-probe/examples/row_filter.rs b/crates/crepe-plan-probe/examples/row_filter.rs new file mode 100644 index 000000000..a7f724d46 --- /dev/null +++ b/crates/crepe-plan-probe/examples/row_filter.rs @@ -0,0 +1,151 @@ +//! The row filter at preparation, two ways, over the same rows: the scalar +//! test (`binary_search` in a sorted needle set) and `ndarray::simd`'s strided +//! compare, one pass per needle OR-ed into a mask, over the row bytes and +//! over a contiguous column (what a column-major store would hand it). +//! Release, medians of 31. Not wired into `prepare`: over the row-major store +//! the strided pass is slower than the scalar test at every needle count +//! (plan §2c). +//! +//! ```text +//! cargo run --release --example row_filter +//! ``` + +use std::hint::black_box; +use std::time::{Duration, Instant}; + +use mork_pathmap_probe::join::for_each_set; +use ndarray::simd::{eq_u32_strided_to_mask, eq_u32_to_mask, mask_or_assign}; + +/// The rows' bytes, borrowed: `[u32; 2]` has no padding and `u8` has no +/// alignment requirement, so the view covers exactly the rows' memory. +fn row_bytes(rows: &[[u32; 2]]) -> &[u8] { + // SAFETY: `rows` is a live, initialized slice of `[u32; 2]`; its + // `size_of_val` bytes are readable for the borrow's lifetime, and every + // bit pattern is a valid `u8`. + unsafe { std::slice::from_raw_parts(rows.as_ptr().cast::(), std::mem::size_of_val(rows)) } +} + +// `eq_u32_strided_to_mask` reads each field as little-endian. +const _: () = assert!( + cfg!(target_endian = "little"), + "row_bytes hands native-endian u32 bytes to a little-endian kernel" +); + +/// Mask over `rows` of the rows whose column `col` is one of `needles`: one +/// `ndarray::simd` strided compare per needle (16 lanes per step), OR-ed +/// into the mask. `needles` empty gives the empty mask. +fn column_in_mask(rows: &[[u32; 2]], col: usize, needles: &[u32]) -> Vec { + let n = rows.len(); + let words = n.div_ceil(64); + let mut mask = vec![0u64; words]; + let mut hit = vec![0u64; words]; + let bytes = row_bytes(rows); + for &v in needles { + eq_u32_strided_to_mask(bytes, col * 4, 8, n, v, &mut hit); + mask_or_assign(&mut mask, &hit); + } + mask +} + +/// The rows a mask selects, in row order. +fn rows_under(rows: &[[u32; 2]], mask: &[u64]) -> Vec<[u32; 2]> { + let mut out = Vec::new(); + for_each_set(mask, rows.len(), |i| out.push(rows[i])); + out +} + +fn median(mut v: Vec) -> Duration { + v.sort(); + v[v.len() / 2] +} + +fn time(mut f: impl FnMut() -> T) -> Duration { + let mut v = Vec::with_capacity(31); + for _ in 0..31 { + let t = Instant::now(); + black_box(f()); + v.push(t.elapsed()); + } + median(v) +} + +/// The strided filter equals the scalar one on row counts around word +/// boundaries, with repeated and absent needles. +fn check_boundaries() { + let mut s = 0x2545_f491_4f6c_dd1d_u64; + let mut next = move || { + s ^= s << 13; + s ^= s >> 7; + s ^= s << 17; + (s % 40) as u32 + }; + for n in [0usize, 1, 15, 16, 17, 63, 64, 65, 200] { + let rows: Vec<[u32; 2]> = (0..n).map(|_| [next(), next()]).collect(); + for needles in [ + vec![], + vec![3], + vec![3, 3, 7], + vec![1000], + (0..40).collect(), + ] { + for col in [0, 1] { + let got = rows_under(&rows, &column_in_mask(&rows, col, &needles)); + let want: Vec<[u32; 2]> = rows + .iter() + .copied() + .filter(|r| needles.contains(&r[col])) + .collect(); + assert_eq!(got, want, "n={n} col={col} needles={needles:?}"); + } + } + } +} + +fn main() { + check_boundaries(); + println!("n,needles,kept,scalar_us,ndarray_us,compare_us,collect_us,contiguous_us"); + let mut s = 0x9E37_79B9_7F4A_7C15_u64; + let mut next = move || { + s = s.wrapping_mul(6_364_136_223_846_793_005).wrapping_add(1); + (s >> 33) as u32 + }; + for n in [1_000usize, 50_000] { + let rows: Vec<[u32; 2]> = (0..n).map(|_| [next() % 20_000, next()]).collect(); + for k in [1usize, 2, 4, 8, 16, 32, 64, 128] { + let mut needles: Vec = (0..k as u32).map(|i| i * 97 % 20_000).collect(); + needles.sort_unstable(); + needles.dedup(); + let scalar = || -> Vec<[u32; 2]> { + rows.iter() + .copied() + .filter(|r| needles.binary_search(&r[0]).is_ok()) + .collect() + }; + let simd = || rows_under(&rows, &column_in_mask(&rows, 0, &needles)); + assert_eq!(scalar(), simd()); + let mask = column_in_mask(&rows, 0, &needles); + // What a column-major store would give: the same compares over a + // contiguous u32 column (built once here, outside the timing). + let column: Vec = rows.iter().map(|r| r[0]).collect(); + let contiguous = || { + let mut m = vec![0u64; n.div_ceil(64)]; + let mut hit = vec![0u64; n.div_ceil(64)]; + for &v in &needles { + eq_u32_to_mask(&column, v, &mut hit); + mask_or_assign(&mut m, &hit); + } + m + }; + assert_eq!(contiguous(), mask); + println!( + "{n},{k},{},{:.1},{:.1},{:.1},{:.1},{:.1}", + scalar().len(), + time(scalar).as_secs_f64() * 1e6, + time(simd).as_secs_f64() * 1e6, + time(|| column_in_mask(&rows, 0, &needles)).as_secs_f64() * 1e6, + time(|| rows_under(&rows, &mask)).as_secs_f64() * 1e6, + time(contiguous).as_secs_f64() * 1e6, + ); + } + } +} diff --git a/crates/crepe-plan-probe/src/native.rs b/crates/crepe-plan-probe/src/native.rs index 005dd4972..a911d1d75 100644 --- a/crates/crepe-plan-probe/src/native.rs +++ b/crates/crepe-plan-probe/src/native.rs @@ -213,7 +213,20 @@ pub struct PreparedStar { /// Lays the plan's sides out over one shared domain. A side's restriction is /// applied to the rows it is prepared from. pub fn prepare(plan: &StarPlan, store: &Store<'_>) -> PreparedStar { - let owned: Vec<(Vec<[u32; 2]>, u64)> = plan + prepare_with(plan, store, false) +} + +/// [`prepare`] with semijoin reduction: when at least one side is restricted +/// by a constant, every side keeps only the rows whose center the restricted +/// sides all have. A center outside that set cannot reach the joint mask, so +/// the answer and its multiplicities are unchanged; the domain shrinks to the +/// reduced centers. Without a restricted side nothing changes. +pub fn prepare_semijoin(plan: &StarPlan, store: &Store<'_>) -> PreparedStar { + prepare_with(plan, store, true) +} + +fn prepare_with(plan: &StarPlan, store: &Store<'_>, semijoin: bool) -> PreparedStar { + let mut owned: Vec<(Vec<[u32; 2]>, u64)> = plan .sides .iter() .map(|s| { @@ -227,6 +240,31 @@ pub fn prepare(plan: &StarPlan, store: &Store<'_>) -> PreparedStar { (kept, g) }) .collect(); + if semijoin { + // Centers every restricted side has: sorted, deduplicated, intersected. + let mut centers: Option> = None; + for (s, (rows, _)) in plan.sides.iter().zip(&owned) { + if s.key.unary || s.key.restrict.is_none() { + continue; + } + let mut c: Vec = rows.iter().map(|r| r[s.key.key_col as usize]).collect(); + c.sort_unstable(); + c.dedup(); + centers = Some(match centers { + None => c, + Some(prev) => prev + .into_iter() + .filter(|v| c.binary_search(v).is_ok()) + .collect(), + }); + } + if let Some(centers) = centers { + for (s, (rows, _)) in plan.sides.iter().zip(owned.iter_mut()) { + let col = s.key.key_col as usize; + rows.retain(|r| centers.binary_search(&r[col]).is_ok()); + } + } + } let inputs: Vec<(RelationView<'_>, usize)> = owned .iter() .zip(&plan.sides) diff --git a/crates/crepe-plan-probe/src/optimize.rs b/crates/crepe-plan-probe/src/optimize.rs index 4bb1962bd..e93e4bbf5 100644 --- a/crates/crepe-plan-probe/src/optimize.rs +++ b/crates/crepe-plan-probe/src/optimize.rs @@ -59,16 +59,22 @@ pub struct Demand { pub struct Options { pub propagate: bool, pub dedup: bool, + /// Semijoin reduction at preparation: when a side of a star is restricted + /// by a constant, the other sides are laid out only for the centers that + /// side has. + pub semijoin: bool, } impl Options { pub const ALL: Options = Options { propagate: true, dedup: true, + semijoin: true, }; pub const NONE: Options = Options { propagate: false, dedup: false, + semijoin: false, }; } diff --git a/crates/crepe-plan-probe/src/pipeline.rs b/crates/crepe-plan-probe/src/pipeline.rs index ecbdfbe27..a7817add7 100644 --- a/crates/crepe-plan-probe/src/pipeline.rs +++ b/crates/crepe-plan-probe/src/pipeline.rs @@ -10,7 +10,8 @@ use mork_pathmap_probe::Semantics; use crate::logic::{RelId, Rule}; use crate::mork_api::{MorkRefusal, QueryMulti, admit, query_multi}; use crate::native::{ - NativeRefusal, PreparedStar, StarPlan, Store, bindings_run, joint, prepare, star, + NativeRefusal, PreparedStar, StarPlan, Store, bindings_run, joint, prepare, prepare_semijoin, + star, }; use crate::optimize::{Demand, Optimized, Options, ResultKind, optimize}; @@ -75,7 +76,16 @@ pub fn evaluate( let map_cost = t.elapsed(); let t = Instant::now(); - let prepared: Vec = stars.iter().map(|s| prepare(s, store)).collect(); + let prepared: Vec = stars + .iter() + .map(|s| { + if opts.semijoin { + prepare_semijoin(s, store) + } else { + prepare(s, store) + } + }) + .collect(); let prepare_cost = t.elapsed(); let t = Instant::now(); diff --git a/crates/crepe-plan-probe/tests/pipeline.rs b/crates/crepe-plan-probe/tests/pipeline.rs index 127092980..cd69b5a28 100644 --- a/crates/crepe-plan-probe/tests/pipeline.rs +++ b/crates/crepe-plan-probe/tests/pipeline.rs @@ -719,3 +719,113 @@ fn keys_name_relation_and_column() { let b = SideKey { key_col: 0, ..a }; assert_ne!(a, b); } + +// --------------------------------------------------------------------------- +// N. Semijoin reduction before preparation +// --------------------------------------------------------------------------- + +const NO_SEMIJOIN: Options = Options { + semijoin: false, + ..Options::ALL +}; + +/// A restricted side names the only centers that can answer; the other sides +/// are laid out for those centers only. Same answers, set and bag. +#[test] +fn n_semijoin_prepares_only_the_restricted_centers() { + let f = family_data(); + let (labels, rules) = resolved(&family::crepe_plan()); + let st = store(&labels, &f, [1, 1, 1]); + let d = answer_demand(&labels, 17); + for kind in [ + ResultKind::Bindings(SET), + ResultKind::Bindings(BAG), + ResultKind::Count(BAG), + ResultKind::Exists, + ] { + let (on, rep_on) = evaluate(&rules, &d, kind, Options::ALL, &st).unwrap(); + let (off, rep_off) = evaluate(&rules, &d, kind, NO_SEMIJOIN, &st).unwrap(); + assert_eq!(on, off, "{kind:?}"); + // Parent(17, y) has y in {1, 2, 3}: 3 rows; Parent(y, z) for those y: + // 5 rows; Allowed(y): 3 rows. Without the reduction every Parent and + // Allowed row is laid out. + assert_eq!(rep_on.rows_prepared, 3 + 5 + 3, "{kind:?}"); + assert_eq!( + rep_off.rows_prepared, + 3 + f.parent.len() + f.allowed.len(), + "{kind:?}" + ); + } + let want = reference_bag(&labels, &rules, &st, &d); + let (bag, _) = evaluate(&rules, &d, ResultKind::Bindings(BAG), Options::ALL, &st).unwrap(); + assert_eq!(bindings(bag), want.into_iter().collect::>()); +} + +/// No restricted side: nothing to reduce by, so the layout is unchanged. +#[test] +fn n_semijoin_without_a_restricted_side_changes_nothing() { + let f = family_data(); + let (labels, rules) = resolved(&family::crepe_plan()); + let st = store(&labels, &f, [1, 1, 1]); + let d = answer_demand(&labels, 17); + let only_semijoin = Options { + semijoin: true, + ..Options::NONE + }; + let kind = ResultKind::Bindings(SET); + let (a, ra) = evaluate(&rules, &d, kind, only_semijoin, &st).unwrap(); + let (b, rb) = evaluate(&rules, &d, kind, Options::NONE, &st).unwrap(); + assert_eq!(a, b); + assert_eq!(ra.rows_prepared, rb.rows_prepared); +} + +/// A restricted side with no rows empties the star; the reference agrees. +#[test] +fn n_semijoin_with_an_empty_restricted_side_answers_nothing() { + let f = family_data(); + let (labels, rules) = resolved(&family::crepe_plan()); + let st = store(&labels, &f, [1, 1, 1]); + let d = answer_demand(&labels, 999); + let (on, rep) = evaluate(&rules, &d, ResultKind::Bindings(BAG), Options::ALL, &st).unwrap(); + assert_eq!(bindings(on), Vec::<(Vec, u64)>::new()); + assert!(reference_bag(&labels, &rules, &st, &d).is_empty()); + assert_eq!(rep.rows_prepared, 0); +} + +/// Two restricted sides: only the centers both have survive. +#[test] +fn n_semijoin_intersects_two_restricted_sides() { + let plan = two_way::crepe_plan(); + let (labels, rules) = resolved(&plan); + // R(1, y): y in {10, 10, 11, 12}; S(y, 9): y in {10, 11, 13}. + let mut r: Vec<[u32; 2]> = vec![[1, 10], [1, 10], [1, 11], [1, 12], [2, 10]]; + let mut s: Vec<[u32; 2]> = vec![[10, 9], [11, 9], [11, 8], [12, 7], [13, 9]]; + for i in 0..50u32 { + r.push([100 + i, 200 + i]); + s.push([200 + i, 300 + i]); + } + let st = Store { + relations: vec![ + (labels.id("R").unwrap(), &r, 1), + (labels.id("S").unwrap(), &s, 1), + ], + }; + let d = Demand { + goal: labels.id("T").unwrap(), + bound: vec![Some(1), None, Some(9)], + }; + let want = reference::eval(&rules, d.goal, &d.bound, &raw(&labels, &st)); + assert_eq!( + want.clone().into_iter().collect::>(), + vec![(vec![1, 10, 9], 2), (vec![1, 11, 9], 1)], + "fixture answer" + ); + let kind = ResultKind::Bindings(BAG); + let (on, rep_on) = evaluate(&rules, &d, kind, Options::ALL, &st).unwrap(); + let (off, rep_off) = evaluate(&rules, &d, kind, NO_SEMIJOIN, &st).unwrap(); + assert_eq!(bindings(on.clone()), want.into_iter().collect::>()); + assert_eq!(on, off); + // y = 12 and y = 13 are each on one side only: dropped before layout. + assert_eq!(rep_on.rows_prepared, 3 + 2); + assert_eq!(rep_off.rows_prepared, 4 + 3); +}