From a31c61bb84207591360802076f7c886b0b043544 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 20 Oct 2026 19:05:55 +0000 Subject: [PATCH 1/6] crepe-plan-probe: masked Lance 13 reads and the addressed semijoin Stage 3 (D-HOLO-H14d): each star side is read from a Lance 13 table (lancedb 0.40) under a relalg-built address mask with with_row_addr_prefilter. A side scans exactly the rows of its mask. Arm B: each relation is stored in key order and its keys are attested once with SealedFacetLane::attest_sorted. A side then reads only the lane ranges of its demanded keys, located with seek_from. It reads no mask column and adds no index, opcode or gather. It refuses a demand on the second column, a stale lane, an unsorted table and a table split over several fragments. Fixes found on the way: - The relalg Or holds one scratch slot per eq atom and a fold may use eight, so the relalg membership mask takes at most 4 centers (5 fail with ScratchBufferTooSmall). It was set to 64. - With two restricted sides the Lance path did not trim the restricted sides to the intersected centers, so it prepared 7 rows where the memory path prepared 5. - row_filter's contiguous arm compared a mask with rows (Codex P2). CI: a workflow builds, lints and tests the probe, which sits outside the workspace. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f --- .../plans/crepe-rule-optimizer-spike-v1.md | 25 + .github/workflows/crepe-plan-probe.yml | 48 ++ crates/crepe-plan-probe/Cargo.toml | 13 + .../crepe-plan-probe/examples/row_filter.rs | 8 +- crates/crepe-plan-probe/src/addressed.rs | 307 ++++++++++ crates/crepe-plan-probe/src/lance_store.rs | 390 +++++++++++++ crates/crepe-plan-probe/src/lib.rs | 8 + crates/crepe-plan-probe/src/native.rs | 6 + crates/crepe-plan-probe/src/pipeline.rs | 174 +++++- crates/crepe-plan-probe/tests/pipeline.rs | 534 ++++++++++++++++++ 10 files changed, 1493 insertions(+), 20 deletions(-) create mode 100644 .github/workflows/crepe-plan-probe.yml create mode 100644 crates/crepe-plan-probe/src/addressed.rs create mode 100644 crates/crepe-plan-probe/src/lance_store.rs diff --git a/.claude/plans/crepe-rule-optimizer-spike-v1.md b/.claude/plans/crepe-rule-optimizer-spike-v1.md index 531874eff..17a2b2e86 100644 --- a/.claude/plans/crepe-rule-optimizer-spike-v1.md +++ b/.claude/plans/crepe-rule-optimizer-spike-v1.md @@ -28,6 +28,8 @@ Pipeline, one stage per module of the probe: - [x] Phase 5: differential tests A–M against `crepe!`, datafrog, MORK, reference; 9 disable runs - [x] Phase 6 (limited): Soufflé harvest table (§3) - [x] Semijoin reduction before preparation (`Options::semijoin`, `native::prepare_semijoin`; §2b, tests N) +- [ ] Stage 3 over Lance 13 / lancedb 0.40: masked read per side, predictions P1–P5 (§2d) +- [ ] CI workflow for this probe ## 1. Integration point @@ -106,6 +108,29 @@ The semijoin filter (and the constant restriction) is a scalar test per row: bin - 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. +## 2d. Stage 3 over Lance 13 (pre-registered 2026-10-20, before any run) + +DECISION (operator, 2026-10-20): "ja, bau Stufe 3 mit Lance 13 und CI-Workflow" / "Lance 13 Lancedb 0.40 is mandatory". + +The relations are stored column-major in Lance: one table per relation, columns `c0`, `c1` (`UInt32`), written through lancedb 0.40 and read through lance 13. Preparation becomes H14d's stages 2 and 3 per side: + +1. Read the one column the address mask needs (the restriction column, or the key column for the semijoin), with `_rowaddr`. +2. Build the address mask `R` through relalg over that column as a `U32` lane: `eq_u32(col, k)` for a restriction, `Or` of `eq_u32(key, c)` for the semijoin centers (lowered by mask-risc onto `ndarray::simd::eq_u32_to_mask`). Above 4 centers the mask is built by the scalar membership test instead: each `eq` atom of the `Or` takes one scratch slot and a fold may use at most 8 (`lance_graph_r2il::MAX_SLOTS`), so 5 centers fail with `ScratchBufferTooSmall`. This is relalg's ceiling, not the §2c speed crossover, which is ndarray's row filter. +3. Read the side's columns with `with_row_addr_prefilter(R)`. +4. Hand those rows to `JoinTable::prepare_star`, as the in-memory path does. + +A side's generation is its Lance table version. + +Predictions, each with its falsifier: + +- P1. Answers equal the in-memory path and the nested-loop reference on every existing fixture (A, B, C, F, J, N) for set, bag, count and exists. Falsifier: any difference. +- P2. For every masked read, Lance's `rows_scanned` equals `|R|` exactly, and the rows returned equal `|R|`. Falsifier: any read that scans more rows than `R` admits. +- P3. Rows handed to `prepare_star` equal the in-memory semijoin path's `rows_prepared` exactly (11 and 5 on the N fixtures). Falsifier: any difference. +- P4. A write that adds a table version makes a table prepared under the old version stale (`JoinRefusal::Stale`). Falsifier: the old table is accepted. +- P5 (measured, not predicted as a number): bytes read and wall time per stage at 1,000 / 10,000 / 50,000 rows, against reading every row. No direction is predicted for bytes: H14d found bytes follow page locality, not `|R|`. + +Disables: build `R` from the wrong column (P1, P2 red); drop the prefilter (P2 red); skip the version check (P4 red). + ## 3. Soufflé harvest (souffle-lang/souffle `f53dab8`, UPL-1.0; algorithms only, no source copied) | Soufflé | transformation | precondition | in crepe | in lance-graph | taken | diff --git a/.github/workflows/crepe-plan-probe.yml b/.github/workflows/crepe-plan-probe.yml new file mode 100644 index 000000000..2811c8f17 --- /dev/null +++ b/.github/workflows/crepe-plan-probe.yml @@ -0,0 +1,48 @@ +name: Crepe plan probe +# `crates/crepe-plan-probe` is excluded from the workspace (own [workspace], +# git deps on the crepe and MORK forks, lance 13 / lancedb 0.40), so no +# workspace job builds it. It runs from its own directory so the cargo config +# above it and its own `.cargo/config.toml` apply. It path-deps relalg, +# mask-risc and mork-pathmap-probe, so changes there run it too. +on: + pull_request: + paths: + - 'crates/crepe-plan-probe/**' + - 'crates/lance-graph-relalg/**' + - 'crates/lance-graph-mask-risc/**' + - 'crates/mork-pathmap-probe/**' + - '.github/workflows/crepe-plan-probe.yml' + - 'rust-toolchain.toml' + push: + branches: [main] + paths: + - 'crates/crepe-plan-probe/**' + - 'crates/lance-graph-relalg/**' + - 'crates/lance-graph-mask-risc/**' + - 'crates/mork-pathmap-probe/**' + - '.github/workflows/crepe-plan-probe.yml' + - 'rust-toolchain.toml' + workflow_dispatch: +permissions: + contents: read +jobs: + probe: + runs-on: ubuntu-latest + timeout-minutes: 60 + defaults: + run: + working-directory: crates/crepe-plan-probe + env: + CARGO_PROFILE_DEV_DEBUG: "0" + CARGO_INCREMENTAL: "0" + steps: + - uses: actions/checkout@v4 + - name: Setup rust toolchain (pinned by rust-toolchain.toml) + run: rustup show + # lance's build compiles .proto files: `protoc` alone is not enough, the + # well-known includes (libprotobuf-dev) are needed too. + - name: Install protoc and the protobuf includes (for lance) + run: sudo apt-get update && sudo apt-get install -y protobuf-compiler libprotobuf-dev + - run: cargo fmt --check + - run: cargo clippy --all-targets -- -D warnings + - run: cargo test diff --git a/crates/crepe-plan-probe/Cargo.toml b/crates/crepe-plan-probe/Cargo.toml index eef3e5c9e..36de55f1a 100644 --- a/crates/crepe-plan-probe/Cargo.toml +++ b/crates/crepe-plan-probe/Cargo.toml @@ -21,12 +21,25 @@ lance-graph-mask-risc = { path = "../lance-graph-mask-risc" } # The D-MPS-8 versioned relationship table (shared y domain, presence planes, # generations) and the MORK/PathMap oracle behind boundary.rs. mork-pathmap-probe = { path = "../mork-pathmap-probe" } +# Arm B: the attested key-ordered lane (`SealedFacetLane`, `OrderedLaneWitness`) +# and the FacetCascade keys. +lance-graph-contract = { path = "../lance-graph-contract" } +# Stage 3 (D-HOLO-H14d): the relations live in a Lance table and are read +# under the address mask (`with_row_addr_prefilter`). lance 13 / lancedb 0.40 +# move in lockstep, patch floated, as in the workspace manifest. +lance = { version = "13.*", default-features = false } +lancedb = { version = "0.40.*", default-features = false, features = ["remote"] } +arrow-array = "58" +arrow-schema = "58" +futures = "0.3" +tokio = { version = "1.37", features = ["macros", "rt-multi-thread"] } [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"] } +tempfile = "3" datafrog = { git = "https://github.com/AdaWorldAPI/datafrog", branch = "master" } [patch."https://github.com/AdaWorldAPI/lance-graph"] diff --git a/crates/crepe-plan-probe/examples/row_filter.rs b/crates/crepe-plan-probe/examples/row_filter.rs index a7f724d46..11108866f 100644 --- a/crates/crepe-plan-probe/examples/row_filter.rs +++ b/crates/crepe-plan-probe/examples/row_filter.rs @@ -1,7 +1,8 @@ //! 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). +//! over a contiguous column (what a column-major store would hand it). Every +//! arm returns the selected rows, so the timings cover the same work. //! 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). @@ -134,9 +135,10 @@ fn main() { eq_u32_to_mask(&column, v, &mut hit); mask_or_assign(&mut m, &hit); } - m + // The same output as the scalar arm: the rows, not the mask. + rows_under(&rows, &m) }; - assert_eq!(contiguous(), mask); + assert_eq!(contiguous(), scalar()); println!( "{n},{k},{},{:.1},{:.1},{:.1},{:.1},{:.1}", scalar().len(), diff --git a/crates/crepe-plan-probe/src/addressed.rs b/crates/crepe-plan-probe/src/addressed.rs new file mode 100644 index 000000000..c5111c0d2 --- /dev/null +++ b/crates/crepe-plan-probe/src/addressed.rs @@ -0,0 +1,307 @@ +//! Arm B: the addressed semijoin (plan §2e). +//! +//! A relation's rows are stored in key order: row `(a, b)` carries the key +//! `FacetCascade::from_semantic_tiles([0, 0, a_hi, a_lo, b_hi, b_lo, 0, 0])`, +//! default class (no concept is minted for a probe relation), and the table is +//! written in numeric projection order. The order is attested, never produced +//! here: [`attest`] reads the keys once and hands them to +//! `SealedFacetLane::attest_sorted`, which refuses an unsorted lane. +//! +//! The rows of one leading value `a` are then a contiguous range, located by +//! `SealedFacetLane::seek_from` under the lane's `OrderedLaneWitness`. A side +//! is read with `with_row_addr_prefilter` over the union of its ranges, so the +//! rows Lance visits are the rows of the demanded keys and nothing else. No +//! value is translated to another coordinate space: the addresses are the +//! lane's own positions, in NodeGuid order. +//! +//! What this arm refuses, by name: +//! - a side whose restriction or demand is on the second column (the lane is +//! ordered by the first; the reverse order is a second lane, not built); +//! - a lane whose table moved on since it was attested (`Stale`); +//! - a table whose rows are not in key order (`NotSorted`, at attest); +//! - a table that is not one contiguous fragment (`NotContiguous`, at attest). + +use std::sync::Arc; + +use arrow_array::{Array, UInt32Array, UInt64Array}; +use lance::dataset::Dataset; +use lance::dataset::scanner::{RowAddrMask, RowAddrTreeMap}; +use lance_graph_contract::facet::{FacetCascade, SemanticPrefix}; +use lance_graph_contract::ordered_lane::{OrderedLaneWitness, SealedFacetLane, WitnessError}; + +use crate::lance_store::{LanceStore, LanceStoreError, SideIo, read_rows}; +use crate::logic::RelId; +use crate::native::{PreparedStar, StarPlan, prepare_from_rows}; + +/// The key of row `(a, b)`. +pub fn key(a: u32, b: u32) -> FacetCascade { + let hi = |v: u32| (v >> 16) as u16; + let lo = |v: u32| v as u16; + FacetCascade::from_semantic_tiles([0, 0, hi(a), lo(a), hi(b), lo(b), 0, 0]) +} + +/// The prefix that fixes the leading value `a` (class tiles and `a`). +pub fn leading(a: u32) -> SemanticPrefix { + SemanticPrefix::of(key(a, 0), 4) +} + +#[derive(Debug)] +pub enum AddressedRefusal { + Lance(LanceStoreError), + Witness(WitnessError), + /// The table is not one contiguous fragment, so lane position `i` is not + /// row address `i`. + NotContiguous { + rel: RelId, + }, + /// The table changed since its lane was attested. + Stale { + rel: RelId, + attested: u64, + current: u64, + }, + /// No lane was attested for this relation. + NoLane { + rel: RelId, + }, + /// The side would need the lane ordered by its second column. + NotLeadingColumn { + side: usize, + }, +} + +impl From for AddressedRefusal { + fn from(e: LanceStoreError) -> Self { + AddressedRefusal::Lance(e) + } +} + +/// One relation's attested lane. +pub struct AddressedRelation { + pub rel: RelId, + pub lane: SealedFacetLane, + pub witness: OrderedLaneWitness, +} + +/// Reads a relation's keys once and attests their order at the table's +/// current version. This is the cold cost `P` of the arm. +pub async fn attest(store: &LanceStore, rel: RelId) -> Result { + let ds = store.dataset(rel).await?; + let (keys, contiguous) = read_keys(&ds).await?; + if !contiguous { + return Err(AddressedRefusal::NotContiguous { rel }); + } + let lane = SealedFacetLane::attest_sorted(keys, ds.version().version) + .map_err(AddressedRefusal::Witness)?; + let witness = lane.witness(); + Ok(AddressedRelation { rel, lane, witness }) +} + +async fn read_keys(ds: &Dataset) -> Result<(Vec, bool), LanceStoreError> { + let mut s = ds.scan(); + s.project(&["c0", "c1"]) + .map_err(|e| LanceStoreError::Lance(e.to_string()))?; + s.with_row_address(); + let b = s + .try_into_batch() + .await + .map_err(|e| LanceStoreError::Lance(e.to_string()))?; + let col = |n: &str| { + b.column_by_name(n) + .and_then(|c| c.as_any().downcast_ref::()) + .expect("u32 column") + .clone() + }; + let (c0, c1) = (col("c0"), col("c1")); + let addr = b + .column_by_name("_rowaddr") + .and_then(|c| c.as_any().downcast_ref::()) + .expect("_rowaddr"); + let contiguous = (0..b.num_rows()).all(|i| addr.value(i) == i as u64); + let keys = (0..b.num_rows()) + .map(|i| key(c0.value(i), c1.value(i))) + .collect(); + Ok((keys, contiguous)) +} + +/// What the addressed arm did on one side, beyond Lance's counters. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct Lookup { + /// Distinct leading values sought. + pub keys: usize, + /// Rows inside the located ranges. + pub rows_in_ranges: usize, +} + +/// The rows of every leading value in `wanted` (sorted, deduplicated) as lane +/// ranges, found with galloping seeks from the previous range's end. +fn ranges( + a: &AddressedRelation, + wanted: &[u32], +) -> Result<(Vec<(u32, u32)>, usize), AddressedRefusal> { + let mut out = Vec::with_capacity(wanted.len()); + let mut from = 0u32; + let mut rows = 0usize; + for &v in wanted { + let lo = a + .lane + .seek_from(&a.witness, from, &leading(v).lo_key()) + .map_err(AddressedRefusal::Witness)?; + let hi = match v.checked_add(1) { + Some(next) => a + .lane + .seek_from(&a.witness, lo, &leading(next).lo_key()) + .map_err(AddressedRefusal::Witness)?, + None => a.lane.n_rows(), + }; + if hi > lo { + out.push((lo, hi)); + rows += (hi - lo) as usize; + } + from = hi; + } + Ok((out, rows)) +} + +fn range_mask(ranges: &[(u32, u32)]) -> RowAddrMask { + let mut m = RowAddrTreeMap::new(); + for &(lo, hi) in ranges { + m.insert_range(u64::from(lo)..u64::from(hi)); + } + RowAddrMask::from_allowed(m) +} + +fn lane_for( + lanes: &[AddressedRelation], + rel: RelId, +) -> Result<&AddressedRelation, AddressedRefusal> { + lanes + .iter() + .find(|l| l.rel == rel) + .ok_or(AddressedRefusal::NoLane { rel }) +} + +/// The side's attested lane and dataset, refused when the table moved on. +async fn current_lane<'a>( + plan: &StarPlan, + i: usize, + store: &LanceStore, + lanes: &'a [AddressedRelation], +) -> Result<(&'a AddressedRelation, Arc), AddressedRefusal> { + let rel = plan.sides[i].key.rel; + let a = lane_for(lanes, rel)?; + let ds = store.dataset(rel).await?; + let current = ds.version().version; + if current != a.lane.version() { + return Err(AddressedRefusal::Stale { + rel, + attested: a.lane.version(), + current, + }); + } + Ok((a, ds)) +} + +/// Prepares `plan` reading only the rows of the demanded keys: restricted +/// sides by their constant, the others by the centers every restricted side +/// has. Without a restricted side every lane is read in full (the arm has no +/// demand to address by). +pub async fn prepare_addressed( + plan: &StarPlan, + store: &LanceStore, + lanes: &[AddressedRelation], +) -> Result<(PreparedStar, Vec, Vec), AddressedRefusal> { + let n_sides = plan.sides.len(); + let mut rows: Vec, u64)>> = vec![None; n_sides]; + let mut io = vec![SideIo::default(); n_sides]; + let mut lookups = vec![Lookup::default(); n_sides]; + let restricted = |i: usize| { + let k = plan.sides[i].key; + (!k.unary).then_some(k.restrict).flatten() + }; + + for i in (0..n_sides).filter(|&i| restricted(i).is_some()) { + let key = plan.sides[i].key; + // The constant must be on the lane's leading column, so the center is + // the second column. + if key.key_col != 1 { + return Err(AddressedRefusal::NotLeadingColumn { side: i }); + } + let k = u32::try_from(restricted(i).expect("restricted")) + .map_err(|_| AddressedRefusal::NotLeadingColumn { side: i })?; + let (a, ds) = current_lane(plan, i, store, lanes).await?; + let (r, n) = ranges(a, &[k])?; + let (got, rows_io) = read_rows(&ds, Some(range_mask(&r))).await?; + lookups[i] = Lookup { + keys: 1, + rows_in_ranges: n, + }; + io[i] = SideIo { + mask_column: None, + mask_rows: Some(n), + rows: rows_io, + }; + rows[i] = Some((got, a.lane.version())); + } + + let mut centers: Option> = None; + for i in (0..n_sides).filter(|&i| restricted(i).is_some()) { + let mut c: Vec = rows[i] + .as_ref() + .expect("read") + .0 + .iter() + .map(|r| r[1]) + .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(), + }); + } + + // Restricted sides keep only the intersected centers (rows already read). + if let Some(c) = ¢ers { + for i in (0..n_sides).filter(|&i| restricted(i).is_some()) { + if let Some((r, _)) = rows[i].as_mut() { + r.retain(|row| c.binary_search(&row[1]).is_ok()); + } + } + } + + for i in (0..n_sides).filter(|&i| restricted(i).is_none()) { + let key = plan.sides[i].key; + if key.key_col != 0 { + return Err(AddressedRefusal::NotLeadingColumn { side: i }); + } + let (a, ds) = current_lane(plan, i, store, lanes).await?; + let (mask, n, sought) = match ¢ers { + Some(c) => { + let (r, n) = ranges(a, c)?; + (Some(range_mask(&r)), n, c.len()) + } + None => (None, a.lane.n_rows() as usize, 0), + }; + let (got, rows_io) = read_rows(&ds, mask).await?; + lookups[i] = Lookup { + keys: sought, + rows_in_ranges: n, + }; + io[i] = SideIo { + mask_column: None, + mask_rows: centers.as_ref().map(|_| n), + rows: rows_io, + }; + rows[i] = Some((got, a.lane.version())); + } + + let owned = rows + .into_iter() + .map(|r| r.expect("every side read")) + .collect(); + Ok((prepare_from_rows(plan, owned), io, lookups)) +} diff --git a/crates/crepe-plan-probe/src/lance_store.rs b/crates/crepe-plan-probe/src/lance_store.rs new file mode 100644 index 000000000..90b2ff778 --- /dev/null +++ b/crates/crepe-plan-probe/src/lance_store.rs @@ -0,0 +1,390 @@ +//! Stage 3 of D-HOLO-H14d for the star: the relations live column-major in +//! Lance (one table per relation, columns `c0` and `c1`, written through +//! lancedb 0.40, read through lance 13), and every side is read under an +//! address mask (plan §2d). +//! +//! Per side: +//! +//! 1. the one column the mask needs is read with `_rowaddr`; +//! 2. the address mask `R` is built by relalg over that column as a `U32` +//! lane (`eq_u32` for a restriction, `Or` of `eq_u32` for the semijoin +//! centers), lowered by mask-risc onto `ndarray::simd`; +//! 3. the side is read with `with_row_addr_prefilter(R)`; +//! 4. the rows go to `JoinTable::prepare_star`, as on the in-memory path. +//! +//! A side's generation is its table's Lance version. + +use std::sync::{Arc, Mutex}; + +use arrow_array::{Array, RecordBatch, UInt32Array, UInt64Array}; +use arrow_schema::{DataType, Field, Schema}; +use lance::dataset::Dataset; +use lance::dataset::scanner::{RowAddrMask, RowAddrTreeMap}; +use lance_graph_mask_risc::{LaneRef, Planes, words_for}; +use lance_graph_relalg::{Measure, Query, Rel, RelError, run_with_program}; +use mork_pathmap_probe::join::for_each_set; + +use crate::logic::RelId; +use crate::native::{PreparedStar, StarPlan, prepare_from_rows}; + +/// Up to this many semijoin centers, the address mask is an `Or` of `eq_u32` +/// atoms through relalg; above it, the scalar membership test builds it. +/// +/// The limit is relalg's, not a speed crossover: each `eq` atom of the `Or` +/// holds its own scratch slot, and one fold may use at most +/// `lance_graph_r2il::MAX_SLOTS` (8) of them. Measured: 4 centers fit, 5 +/// fail with `ScratchBufferTooSmall { need_words: 145, have_words: 129 }` at +/// 1,000 rows. (The 64-needle crossover in plan §2c is ndarray's row filter, +/// a different mechanism.) +pub const SEMIJOIN_ATOMS_MAX: usize = 4; + +#[derive(Debug)] +pub enum LanceStoreError { + Lance(String), + UnknownRelation(RelId), + Relalg(RelError), +} + +fn lance_err(e: impl std::fmt::Display) -> LanceStoreError { + LanceStoreError::Lance(e.to_string()) +} + +/// One relation per table. +pub struct LanceStore { + tables: Vec<(RelId, lancedb::Table)>, +} + +fn schema() -> Arc { + Arc::new(Schema::new(vec![ + Field::new("c0", DataType::UInt32, false), + Field::new("c1", DataType::UInt32, false), + ])) +} + +fn batch(rows: &[[u32; 2]]) -> RecordBatch { + let c0 = UInt32Array::from_iter_values(rows.iter().map(|r| r[0])); + let c1 = UInt32Array::from_iter_values(rows.iter().map(|r| r[1])); + RecordBatch::try_new(schema(), vec![Arc::new(c0), Arc::new(c1)]).expect("two u32 columns") +} + +impl LanceStore { + /// Creates one table per relation under `uri`, each written in key order + /// (`(c0, c1)` ascending, which is the numeric projection order of the + /// keys `addressed::key` gives the rows). The write path orders; readers + /// attest the order, they never sort. + pub async fn write( + uri: &str, + relations: &[(RelId, &str, &[[u32; 2]])], + ) -> Result { + let ordered: Vec<(RelId, &str, Vec<[u32; 2]>)> = relations + .iter() + .map(|&(rel, name, rows)| { + let mut r = rows.to_vec(); + r.sort_unstable(); + (rel, name, r) + }) + .collect(); + let views: Vec<(RelId, &str, &[[u32; 2]])> = ordered + .iter() + .map(|(rel, name, r)| (*rel, *name, r.as_slice())) + .collect(); + Self::write_as_given(uri, &views).await + } + + /// Creates the tables with the rows in the order given (for the test that + /// an unordered table is refused at attest). + pub async fn write_as_given( + uri: &str, + relations: &[(RelId, &str, &[[u32; 2]])], + ) -> Result { + let db = lancedb::connect(uri).execute().await.map_err(lance_err)?; + let mut tables = Vec::new(); + for &(rel, name, rows) in relations { + let t = db + .create_table(name, batch(rows)) + .execute() + .await + .map_err(lance_err)?; + tables.push((rel, t)); + } + Ok(Self { tables }) + } + + fn table(&self, rel: RelId) -> Result<&lancedb::Table, LanceStoreError> { + self.tables + .iter() + .find(|(r, _)| *r == rel) + .map(|(_, t)| t) + .ok_or(LanceStoreError::UnknownRelation(rel)) + } + + /// Appends rows to a relation; returns its new version. + pub async fn append(&self, rel: RelId, rows: &[[u32; 2]]) -> Result { + let t = self.table(rel)?; + t.add(batch(rows)).execute().await.map_err(lance_err)?; + t.version().await.map_err(lance_err) + } + + /// The relation's current Lance version. + pub async fn version(&self, rel: RelId) -> Result { + self.table(rel)?.version().await.map_err(lance_err) + } + + /// The current version of each side of `plan`. + pub async fn generations(&self, plan: &StarPlan) -> Result, LanceStoreError> { + let mut out = Vec::with_capacity(plan.sides.len()); + for s in &plan.sides { + out.push(self.version(s.key.rel).await?); + } + Ok(out) + } + + pub(crate) async fn dataset(&self, rel: RelId) -> Result, LanceStoreError> { + let t = self.table(rel)?; + let wrapper = t + .dataset() + .ok_or_else(|| LanceStoreError::Lance("not a native table".into()))?; + wrapper.get().await.map_err(lance_err) + } +} + +/// Lance's own statistics for one read. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct Io { + pub rows_scanned: usize, + pub rows_returned: usize, + pub bytes_read: usize, +} + +/// What preparing one side read. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct SideIo { + /// The read of the column the mask was built from; `None` when the side + /// was read without a mask. + pub mask_column: Option, + /// `|R|`, when a mask was built. + pub mask_rows: Option, + /// The read of the side itself (both columns). + pub rows: Io, +} + +fn stats_sink() -> ( + Arc>, + lance::dataset::scanner::ExecutionStatsCallback, +) { + let stats: Arc> = Arc::default(); + let sink = stats.clone(); + let cb: lance::dataset::scanner::ExecutionStatsCallback = Arc::new(move |c| { + let mut io = sink.lock().expect("stats lock"); + io.bytes_read = c.bytes_read; + io.rows_scanned = c.all_counts.get("rows_scanned").copied().unwrap_or(0); + }); + (stats, cb) +} + +fn u32_col<'b>(b: &'b RecordBatch, name: &str) -> &'b UInt32Array { + b.column_by_name(name) + .and_then(|c| c.as_any().downcast_ref::()) + .expect("u32 column") +} + +/// One column, every row, with each row's address. +async fn read_column( + ds: &Dataset, + col: usize, +) -> Result<(Vec, Vec, Io), LanceStoreError> { + let name = format!("c{col}"); + let (stats, cb) = stats_sink(); + let mut s = ds.scan(); + s.project(&[name.as_str()]).map_err(lance_err)?; + s.with_row_address(); + s.scan_stats_callback(cb); + let b = s.try_into_batch().await.map_err(lance_err)?; + let values = u32_col(&b, &name).values().to_vec(); + let addr = b + .column_by_name("_rowaddr") + .and_then(|c| c.as_any().downcast_ref::()) + .expect("_rowaddr") + .values() + .to_vec(); + let mut io = *stats.lock().expect("stats lock"); + io.rows_returned = b.num_rows(); + Ok((values, addr, io)) +} + +/// Both columns, under `mask` when given. +pub(crate) async fn read_rows( + ds: &Dataset, + mask: Option, +) -> Result<(Vec<[u32; 2]>, Io), LanceStoreError> { + let (stats, cb) = stats_sink(); + let mut s = ds.scan(); + s.project(&["c0", "c1"]).map_err(lance_err)?; + if let Some(m) = mask { + s.with_row_addr_prefilter(m); + } + s.scan_stats_callback(cb); + let b = s.try_into_batch().await.map_err(lance_err)?; + let c0 = u32_col(&b, "c0"); + let c1 = u32_col(&b, "c1"); + let rows = (0..b.num_rows()) + .map(|i| [c0.value(i), c1.value(i)]) + .collect(); + let mut io = *stats.lock().expect("stats lock"); + io.rows_returned = b.num_rows(); + Ok((rows, io)) +} + +/// `rel` over `values` as lane 0: the relalg description, lowered by +/// mask-risc, run over the column. +pub fn address_mask(values: &[u32], rel: &Rel) -> Result, RelError> { + let mut keep = vec![0u64; words_for(values.len())]; + let lanes = [LaneRef::U32(values)]; + let planes = Planes { + n_rows: values.len(), + masks: &[], + lanes: &lanes, + }; + run_with_program( + &Query { + rel: rel.clone(), + measure: Measure::Keep, + }, + &planes, + Some(&mut keep), + )?; + Ok(keep) +} + +/// Rows whose value is one of `centers` (sorted, deduplicated). +pub fn membership_mask(values: &[u32], centers: &[u32]) -> Result, RelError> { + if centers.is_empty() { + return Ok(vec![0; words_for(values.len())]); + } + if centers.len() <= SEMIJOIN_ATOMS_MAX { + let rel = Rel::Or(centers.iter().map(|&c| Rel::eq_u32(0, c)).collect()); + return address_mask(values, &rel); + } + let mut keep = vec![0u64; words_for(values.len())]; + for (i, v) in values.iter().enumerate() { + if centers.binary_search(v).is_ok() { + keep[i / 64] |= 1 << (i % 64); + } + } + Ok(keep) +} + +fn row_addr_mask(mask: &[u64], n: usize, addr: &[u64]) -> (RowAddrMask, usize) { + let mut set = Vec::new(); + for_each_set(mask, n, |i| set.push(addr[i])); + let k = set.len(); + (RowAddrMask::from_allowed(RowAddrTreeMap::from_iter(set)), k) +} + +/// Prepares `plan` from the Lance tables, reading each side under its +/// address mask. With `semijoin`, an unrestricted side is read only for the +/// centers every restricted side has (as `native::prepare_semijoin`). +pub async fn prepare_lance( + plan: &StarPlan, + store: &LanceStore, + semijoin: bool, +) -> Result<(PreparedStar, Vec), LanceStoreError> { + let n_sides = plan.sides.len(); + let mut rows: Vec, u64)>> = vec![None; n_sides]; + let mut io = vec![SideIo::default(); n_sides]; + let restricted = |i: usize| { + let k = plan.sides[i].key; + (!k.unary).then_some(k.restrict).flatten() + }; + + // Restricted sides first: their rows name the centers. + for i in (0..n_sides).filter(|&i| restricted(i).is_some()) { + let key = plan.sides[i].key; + let ds = store.dataset(key.rel).await?; + let other = 1 - key.key_col as usize; + let (values, addr, col_io) = read_column(&ds, other).await?; + let k = u32::try_from(restricted(i).expect("restricted")) + .map_err(|_| LanceStoreError::Lance(format!("restriction outside u32 on side {i}")))?; + let mask = address_mask(&values, &Rel::eq_u32(0, k)).map_err(LanceStoreError::Relalg)?; + let (m, r) = row_addr_mask(&mask, values.len(), &addr); + let (got, rows_io) = read_rows(&ds, Some(m)).await?; + io[i] = SideIo { + mask_column: Some(col_io), + mask_rows: Some(r), + rows: rows_io, + }; + rows[i] = Some((got, ds.version().version)); + } + + let centers: Option> = if semijoin { + let mut acc: Option> = None; + for i in (0..n_sides).filter(|&i| restricted(i).is_some()) { + let col = plan.sides[i].key.key_col as usize; + let mut c: Vec = rows[i] + .as_ref() + .expect("read") + .0 + .iter() + .map(|r| r[col]) + .collect(); + c.sort_unstable(); + c.dedup(); + acc = Some(match acc { + None => c, + Some(prev) => prev + .into_iter() + .filter(|v| c.binary_search(v).is_ok()) + .collect(), + }); + } + // A restricted side keeps only the centers every restricted side has, + // as in `native::prepare_semijoin`. Its rows were already read by its + // restriction; this trims them, it reads nothing. + if let Some(c) = &acc { + for i in (0..n_sides).filter(|&i| restricted(i).is_some()) { + let col = plan.sides[i].key.key_col as usize; + if let Some((r, _)) = rows[i].as_mut() { + r.retain(|row| c.binary_search(&row[col]).is_ok()); + } + } + } + acc + } else { + None + }; + + for i in (0..n_sides).filter(|&i| restricted(i).is_none()) { + let key = plan.sides[i].key; + let ds = store.dataset(key.rel).await?; + let version = ds.version().version; + match ¢ers { + Some(c) => { + let (values, addr, col_io) = read_column(&ds, key.key_col as usize).await?; + let mask = membership_mask(&values, c).map_err(LanceStoreError::Relalg)?; + let (m, r) = row_addr_mask(&mask, values.len(), &addr); + let (got, rows_io) = read_rows(&ds, Some(m)).await?; + io[i] = SideIo { + mask_column: Some(col_io), + mask_rows: Some(r), + rows: rows_io, + }; + rows[i] = Some((got, version)); + } + None => { + let (got, rows_io) = read_rows(&ds, None).await?; + io[i] = SideIo { + mask_column: None, + mask_rows: None, + rows: rows_io, + }; + rows[i] = Some((got, version)); + } + } + } + + let owned = rows + .into_iter() + .map(|r| r.expect("every side read")) + .collect(); + Ok((prepare_from_rows(plan, owned), io)) +} diff --git a/crates/crepe-plan-probe/src/lib.rs b/crates/crepe-plan-probe/src/lib.rs index 15bf6bebf..29a35828e 100644 --- a/crates/crepe-plan-probe/src/lib.rs +++ b/crates/crepe-plan-probe/src/lib.rs @@ -18,10 +18,18 @@ //! 5. [`native`]: the star read out of the pattern, laid out by the D-MPS-8 //! relationship table, the joint mask computed by relalg (`And` of presence //! planes) as one mask-risc program; terminals read the mask. +//! - [`lance_store`]: the same star over Lance 13 tables (written through +//! lancedb 0.40), each side read under a relalg-built address mask with +//! `with_row_addr_prefilter` (D-HOLO-H14d stage 3). +//! - [`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. //! 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 lance_store; pub mod logic; pub mod mork_api; pub mod native; diff --git a/crates/crepe-plan-probe/src/native.rs b/crates/crepe-plan-probe/src/native.rs index a911d1d75..37695486a 100644 --- a/crates/crepe-plan-probe/src/native.rs +++ b/crates/crepe-plan-probe/src/native.rs @@ -265,6 +265,12 @@ fn prepare_with(plan: &StarPlan, store: &Store<'_>, semijoin: bool) -> PreparedS } } } + prepare_from_rows(plan, owned) +} + +/// Lays out rows already selected per side (`(rows, generation)`, in plan +/// side order) over one shared domain. +pub fn prepare_from_rows(plan: &StarPlan, owned: Vec<(Vec<[u32; 2]>, u64)>) -> PreparedStar { let inputs: Vec<(RelationView<'_>, usize)> = owned .iter() .zip(&plan.sides) diff --git a/crates/crepe-plan-probe/src/pipeline.rs b/crates/crepe-plan-probe/src/pipeline.rs index a7817add7..85405bb6a 100644 --- a/crates/crepe-plan-probe/src/pipeline.rs +++ b/crates/crepe-plan-probe/src/pipeline.rs @@ -7,6 +7,8 @@ use std::time::{Duration, Instant}; use lance_graph_mask_risc::Program; use mork_pathmap_probe::Semantics; +use crate::addressed::{AddressedRelation, Lookup, prepare_addressed}; +use crate::lance_store::{LanceStore, LanceStoreError, SideIo, prepare_lance}; use crate::logic::{RelId, Rule}; use crate::mork_api::{MorkRefusal, QueryMulti, admit, query_multi}; use crate::native::{ @@ -17,8 +19,19 @@ use crate::optimize::{Demand, Optimized, Options, ResultKind, optimize}; #[derive(Debug, Clone, PartialEq, Eq)] pub enum PipelineRefusal { - Mork { rule: usize, why: MorkRefusal }, - Native { rule: usize, why: NativeRefusal }, + Mork { + rule: usize, + why: MorkRefusal, + }, + Native { + rule: usize, + why: NativeRefusal, + }, + /// A Lance read or write failed (stage 3 only). + Lance(String), + /// The addressed arm refused a side (Arm B only): stale lane, unsorted + /// table, a demand on the second column, or no attested lane. + Addressed(String), } #[derive(Debug, Clone, PartialEq, Eq)] @@ -38,6 +51,16 @@ pub struct Costs { pub execute: Duration, } +/// The output of [`plan`]: what is evaluated, before any row is read. +#[derive(Debug, Clone)] +pub struct Planned { + pub optimized: Optimized, + pub patterns: Vec, + pub stars: Vec, + pub optimize_cost: Duration, + pub map_cost: Duration, +} + /// Everything one evaluation produced. #[derive(Debug, Clone)] pub struct Report { @@ -52,14 +75,13 @@ pub struct Report { pub costs: Costs, } -/// Rewrites, maps, prepares and runs `rules` for `demand`. -pub fn evaluate( +/// The cold half: rewrites and maps `rules` for `demand`; reads no row. +pub fn plan( rules: &[Rule], demand: &Demand, kind: ResultKind, opts: Options, - store: &Store<'_>, -) -> Result<(Outcome, Report), PipelineRefusal> { +) -> Result { let t = Instant::now(); let optimized = optimize(rules, demand, kind, opts); let optimize_cost = t.elapsed(); @@ -75,8 +97,27 @@ pub fn evaluate( } let map_cost = t.elapsed(); + Ok(Planned { + optimized, + patterns, + stars, + optimize_cost, + map_cost, + }) +} + +/// Rewrites, maps, prepares and runs `rules` for `demand`. +pub fn evaluate( + rules: &[Rule], + demand: &Demand, + kind: ResultKind, + opts: Options, + store: &Store<'_>, +) -> Result<(Outcome, Report), PipelineRefusal> { + let planned = plan(rules, demand, kind, opts)?; let t = Instant::now(); - let prepared: Vec = stars + let prepared: Vec = planned + .stars .iter() .map(|s| { if opts.semijoin { @@ -88,13 +129,36 @@ pub fn evaluate( .collect(); let prepare_cost = t.elapsed(); + let generations: Vec> = planned + .stars + .iter() + .map(|s| { + s.sides + .iter() + .map(|side| store.view(side.key.rel).1) + .collect() + }) + .collect(); + execute(planned, prepared, prepare_cost, kind, &generations) +} + +/// The run half: the joint mask per prepared star, then the terminal. +/// `generations[j]` is the current generation of each side of star `j`. +pub fn execute( + planned: Planned, + prepared: Vec, + prepare_cost: Duration, + kind: ResultKind, + generations: &[Vec], +) -> Result<(Outcome, Report), PipelineRefusal> { + let Planned { + optimized, + patterns, + stars, + optimize_cost, + map_cost, + } = planned; let t = Instant::now(); - let generations = |s: &StarPlan| -> Vec { - s.sides - .iter() - .map(|side| store.view(side.key.rel).1) - .collect() - }; let native = |rule: usize, why| PipelineRefusal::Native { rule, why }; let mut programs = Vec::new(); let mut programs_run = 0; @@ -102,11 +166,16 @@ pub fn evaluate( let outcome = if kind == ResultKind::Exists && !filtered { // Mask-only terminal: no payload is read. let mut any = false; - for ((i, _), (s, p)) in optimized.rules.iter().zip(stars.iter().zip(&prepared)) { + for (j, ((i, _), (s, p))) in optimized + .rules + .iter() + .zip(stars.iter().zip(&prepared)) + .enumerate() + { let (run, prog) = joint( p, s, - &generations(s), + &generations[j], lance_graph_relalg::Measure::Count, None, ) @@ -118,11 +187,16 @@ pub fn evaluate( Outcome::Exists(any) } else { let mut bag: BTreeMap, u64> = BTreeMap::new(); - for ((i, _), (s, p)) in optimized.rules.iter().zip(stars.iter().zip(&prepared)) { + for (j, ((i, _), (s, p))) in optimized + .rules + .iter() + .zip(stars.iter().zip(&prepared)) + .enumerate() + { // One program per rule: the Keep mask that the bindings are read // from; no separate count pass. let (rows, run, prog) = - bindings_run(p, s, &generations(s), true).map_err(|w| native(*i, w))?; + bindings_run(p, s, &generations[j], true).map_err(|w| native(*i, w))?; programs_run += run.programs; programs.extend(prog); for (row, m) in rows { @@ -163,6 +237,72 @@ pub fn evaluate( Ok((outcome, report)) } +/// [`evaluate`] over Lance tables: each side is read under its address mask +/// (stage 3, `lance_store::prepare_lance`), and a side's generation is its +/// table's version at execution. Also returns what each side read. +pub async fn evaluate_lance( + rules: &[Rule], + demand: &Demand, + kind: ResultKind, + opts: Options, + store: &LanceStore, +) -> Result<(Outcome, Report, Vec>), PipelineRefusal> { + let planned = plan(rules, demand, kind, opts)?; + let lance = |e: LanceStoreError| PipelineRefusal::Lance(format!("{e:?}")); + let t = Instant::now(); + let mut prepared = Vec::with_capacity(planned.stars.len()); + let mut io = Vec::with_capacity(planned.stars.len()); + for s in &planned.stars { + let (p, side_io) = prepare_lance(s, store, opts.semijoin) + .await + .map_err(lance)?; + prepared.push(p); + io.push(side_io); + } + let prepare_cost = t.elapsed(); + let mut generations = Vec::with_capacity(planned.stars.len()); + for s in &planned.stars { + generations.push(store.generations(s).await.map_err(lance)?); + } + let (outcome, report) = execute(planned, prepared, prepare_cost, kind, &generations)?; + Ok((outcome, report, io)) +} + +/// [`evaluate`] through the addressed arm (Arm B): each side is read only at +/// the lane ranges of its demanded keys, located on the attested +/// `SealedFacetLane`s in `lanes`. Also returns what each side read and the +/// lookups each side made. +pub async fn evaluate_addressed( + rules: &[Rule], + demand: &Demand, + kind: ResultKind, + opts: Options, + store: &LanceStore, + lanes: &[AddressedRelation], +) -> Result<(Outcome, Report, Vec>, Vec>), PipelineRefusal> { + let planned = plan(rules, demand, kind, opts)?; + let lance = |e: LanceStoreError| PipelineRefusal::Lance(format!("{e:?}")); + let t = Instant::now(); + let mut prepared = Vec::with_capacity(planned.stars.len()); + let mut io = Vec::with_capacity(planned.stars.len()); + let mut lookups = Vec::with_capacity(planned.stars.len()); + for s in &planned.stars { + let (p, side_io, l) = prepare_addressed(s, store, lanes) + .await + .map_err(|e| PipelineRefusal::Addressed(format!("{e:?}")))?; + prepared.push(p); + io.push(side_io); + lookups.push(l); + } + let prepare_cost = t.elapsed(); + let mut generations = Vec::with_capacity(planned.stars.len()); + for s in &planned.stars { + generations.push(store.generations(s).await.map_err(lance)?); + } + let (outcome, report) = execute(planned, prepared, prepare_cost, kind, &generations)?; + Ok((outcome, report, io, lookups)) +} + /// The goal's relation id, for building a [`Demand`]. pub fn demand(goal: RelId, bound: Vec>) -> Demand { Demand { goal, bound } diff --git a/crates/crepe-plan-probe/tests/pipeline.rs b/crates/crepe-plan-probe/tests/pipeline.rs index cd69b5a28..4d971f727 100644 --- a/crates/crepe-plan-probe/tests/pipeline.rs +++ b/crates/crepe-plan-probe/tests/pipeline.rs @@ -829,3 +829,537 @@ fn n_semijoin_intersects_two_restricted_sides() { assert_eq!(rep_on.rows_prepared, 3 + 2); assert_eq!(rep_off.rows_prepared, 4 + 3); } + +// --------------------------------------------------------------------------- +// S. Stage 3 over Lance 13 (plan §2d, predictions P1–P4) +// --------------------------------------------------------------------------- + +mod stage3 { + use super::*; + use crepe_plan_probe::lance_store::{ + LanceStore, SEMIJOIN_ATOMS_MAX, SideIo, membership_mask, prepare_lance, + }; + use crepe_plan_probe::pipeline::evaluate_lance; + + async fn lance_family(labels: &Labels, f: &Family) -> (tempfile::TempDir, LanceStore) { + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap().to_string(); + let st = LanceStore::write( + &uri, + &[ + (labels.id("Parent").unwrap(), "parent", &f.parent), + (labels.id("Allowed").unwrap(), "allowed", &f.allowed), + (labels.id("Sponsor").unwrap(), "sponsor", &f.sponsor), + ], + ) + .await + .unwrap(); + (dir, st) + } + + fn kinds() -> [ResultKind; 5] { + [ + ResultKind::Bindings(SET), + ResultKind::Bindings(BAG), + ResultKind::Count(SET), + ResultKind::Count(BAG), + ResultKind::Exists, + ] + } + + /// P1: the Lance path answers what the in-memory path and the reference + /// answer, for every option set and result kind, on both demands. + #[tokio::test(flavor = "multi_thread")] + async fn p1_lance_answers_equal_the_memory_path_and_the_reference() { + let f = family_data(); + let (labels, rules) = resolved(&family::crepe_plan()); + let mem = store(&labels, &f, [1, 1, 1]); + let (_dir, lance) = lance_family(&labels, &f).await; + for x in [17, 5, 999] { + let d = answer_demand(&labels, x); + let want = reference_bag(&labels, &rules, &mem, &d); + for opts in [Options::ALL, NO_SEMIJOIN, Options::NONE] { + for kind in kinds() { + let (m, _) = evaluate(&rules, &d, kind, opts, &mem).unwrap(); + let (l, _, _) = evaluate_lance(&rules, &d, kind, opts, &lance) + .await + .unwrap(); + assert_eq!(l, m, "x={x} {opts:?} {kind:?}"); + } + let (bag, _, _) = + evaluate_lance(&rules, &d, ResultKind::Bindings(BAG), opts, &lance) + .await + .unwrap(); + assert_eq!(bindings(bag), want.clone().into_iter().collect::>()); + } + } + } + + /// P2: every masked read scans and returns exactly `|R|` rows. + #[tokio::test(flavor = "multi_thread")] + async fn p2_every_masked_read_scans_exactly_the_mask() { + let f = family_data(); + let (labels, rules) = resolved(&family::crepe_plan()); + let (_dir, lance) = lance_family(&labels, &f).await; + let d = answer_demand(&labels, 17); + let (_, _, io) = + evaluate_lance(&rules, &d, ResultKind::Bindings(SET), Options::ALL, &lance) + .await + .unwrap(); + let masked: Vec<(usize, &SideIo)> = io + .iter() + .flatten() + .filter_map(|s| s.mask_rows.map(|r| (r, s))) + .collect(); + // Parent(17, y), Parent(y, z) and Allowed(y) are all read under a mask. + assert_eq!(masked.len(), 3, "{io:?}"); + for (r, s) in &masked { + assert_eq!(s.rows.rows_scanned, *r, "{s:?}"); + assert_eq!(s.rows.rows_returned, *r, "{s:?}"); + } + // Not vacuous: the masks are much smaller than the tables. + let total: usize = masked.iter().map(|(r, _)| r).sum(); + assert_eq!(total, 3 + 5 + 3); + assert!( + masked + .iter() + .any(|(_, s)| s.mask_column.unwrap().rows_returned == f.parent.len()) + ); + } + + /// P3: the rows handed to `prepare_star` equal the in-memory semijoin's. + #[tokio::test(flavor = "multi_thread")] + async fn p3_prepared_rows_equal_the_memory_path() { + let f = family_data(); + let (labels, rules) = resolved(&family::crepe_plan()); + let mem = store(&labels, &f, [1, 1, 1]); + let (_dir, lance) = lance_family(&labels, &f).await; + let d = answer_demand(&labels, 17); + for opts in [Options::ALL, NO_SEMIJOIN] { + let kind = ResultKind::Bindings(BAG); + let (_, rm) = evaluate(&rules, &d, kind, opts, &mem).unwrap(); + let (_, rl, _) = evaluate_lance(&rules, &d, kind, opts, &lance) + .await + .unwrap(); + assert_eq!(rl.rows_prepared, rm.rows_prepared, "{opts:?}"); + } + // Two restricted sides, intersected. + let plan = two_way::crepe_plan(); + let (labels, rules) = resolved(&plan); + let r: Vec<[u32; 2]> = vec![[1, 10], [1, 10], [1, 11], [1, 12], [2, 10], [7, 70]]; + let s: Vec<[u32; 2]> = vec![[10, 9], [11, 9], [11, 8], [12, 7], [13, 9], [70, 1]]; + let dir = tempfile::tempdir().unwrap(); + let lance = LanceStore::write( + dir.path().to_str().unwrap(), + &[ + (labels.id("R").unwrap(), "r", &r), + (labels.id("S").unwrap(), "s", &s), + ], + ) + .await + .unwrap(); + let mem = 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 kind = ResultKind::Bindings(BAG); + let (om, rm) = evaluate(&rules, &d, kind, Options::ALL, &mem).unwrap(); + let (ol, rl, _) = evaluate_lance(&rules, &d, kind, Options::ALL, &lance) + .await + .unwrap(); + assert_eq!(ol, om); + assert_eq!((rl.rows_prepared, rm.rows_prepared), (5, 5)); + } + + /// P4: a write that adds a version makes a table prepared before it stale. + #[tokio::test(flavor = "multi_thread")] + async fn p4_an_append_makes_the_prepared_table_stale() { + let f = family_data(); + let (labels, rules) = resolved(&family::crepe_plan()); + let (_dir, lance) = lance_family(&labels, &f).await; + let d = answer_demand(&labels, 17); + let planned = + crepe_plan_probe::pipeline::plan(&rules, &d, ResultKind::Bindings(SET), Options::ALL) + .unwrap(); + let star = &planned.stars[0]; + let (prep, _) = prepare_lance(star, &lance, true).await.unwrap(); + let before = lance.generations(star).await.unwrap(); + assert!(joint(&prep, star, &before, Measure::Count, None).is_ok()); + let parent = labels.id("Parent").unwrap(); + let v = lance.append(parent, &[[17, 4]]).await.unwrap(); + let after = lance.generations(star).await.unwrap(); + assert_eq!(after[0], v); + assert!(matches!( + joint(&prep, star, &after, Measure::Count, None), + Err(NativeRefusal::Table(JoinRefusal::Stale { side: 0, .. })) + )); + } + + /// The semijoin mask is built through relalg up to `SEMIJOIN_ATOMS_MAX` + /// centers and by the scalar test above it; both give the same rows. + #[test] + fn the_membership_mask_agrees_on_both_sides_of_the_crossover() { + let values: Vec = (0..1_000u32).map(|i| (i * 37) % 500).collect(); + for k in [ + 0usize, + 1, + 7, + SEMIJOIN_ATOMS_MAX, + SEMIJOIN_ATOMS_MAX + 1, + 200, + ] { + let centers: Vec = (0..k as u32).map(|c| c * 3).collect(); + let got = membership_mask(&values, ¢ers).unwrap(); + for (i, v) in values.iter().enumerate() { + let bit = got[i / 64] >> (i % 64) & 1 == 1; + assert_eq!(bit, centers.contains(v), "k={k} i={i}"); + } + } + } +} + +// --------------------------------------------------------------------------- +// B. Arm B: the addressed semijoin over attested lanes (plan §2e) +// --------------------------------------------------------------------------- + +mod arm_b { + use super::*; + use crepe_plan_probe::addressed::{AddressedRefusal, AddressedRelation, attest, key}; + use crepe_plan_probe::lance_store::{LanceStore, SideIo}; + use crepe_plan_probe::pipeline::{PipelineRefusal, evaluate_addressed, evaluate_lance}; + use lance_graph_contract::ordered_lane::WitnessError; + + struct Fixture { + _dir: tempfile::TempDir, + lance: LanceStore, + lanes: Vec, + } + + async fn fixture(labels: &Labels, f: &Family) -> Fixture { + let dir = tempfile::tempdir().unwrap(); + let rels = [ + (labels.id("Parent").unwrap(), "parent", f.parent.as_slice()), + ( + labels.id("Allowed").unwrap(), + "allowed", + f.allowed.as_slice(), + ), + ( + labels.id("Sponsor").unwrap(), + "sponsor", + f.sponsor.as_slice(), + ), + ]; + let lance = LanceStore::write(dir.path().to_str().unwrap(), &rels) + .await + .unwrap(); + let mut lanes = Vec::new(); + for (rel, _, _) in rels { + lanes.push(attest(&lance, rel).await.unwrap()); + } + Fixture { + _dir: dir, + lance, + lanes, + } + } + + fn kinds() -> [ResultKind; 5] { + [ + ResultKind::Bindings(SET), + ResultKind::Bindings(BAG), + ResultKind::Count(SET), + ResultKind::Count(BAG), + ResultKind::Exists, + ] + } + + /// The written key order is the numeric order of the rows: `key(a, b)` + /// compares as `(a, b)`. + #[test] + fn the_key_orders_as_its_pair() { + let pairs = [ + [0u32, 0], + [0, 1], + [0, u32::MAX], + [1, 0], + [65_535, 7], + [65_536, 0], + [u32::MAX, 0], + ]; + for w in pairs.windows(2) { + assert!( + key(w[0][0], w[0][1]).cmp_numeric_projection(&key(w[1][0], w[1][1])) + == std::cmp::Ordering::Less, + "{w:?}" + ); + } + } + + /// B1: Arm B answers what the memory path, Arm A and the reference + /// answer, for every result kind, on a hit, a sponsor branch and a miss. + #[tokio::test(flavor = "multi_thread")] + async fn b1_addressed_answers_equal_memory_arm_a_and_the_reference() { + let f = family_data(); + let (labels, rules) = resolved(&family::crepe_plan()); + let mem = store(&labels, &f, [1, 1, 1]); + let fx = fixture(&labels, &f).await; + for x in [17, 5, 999] { + let d = answer_demand(&labels, x); + let want = reference_bag(&labels, &rules, &mem, &d); + for kind in kinds() { + let (m, _) = evaluate(&rules, &d, kind, Options::ALL, &mem).unwrap(); + let (a, _, _) = evaluate_lance(&rules, &d, kind, Options::ALL, &fx.lance) + .await + .unwrap(); + let (b, _, _, _) = + evaluate_addressed(&rules, &d, kind, Options::ALL, &fx.lance, &fx.lanes) + .await + .unwrap(); + assert_eq!(b, m, "x={x} {kind:?}"); + assert_eq!(b, a, "x={x} {kind:?}"); + } + let (bag, _, _, _) = evaluate_addressed( + &rules, + &d, + ResultKind::Bindings(BAG), + Options::ALL, + &fx.lance, + &fx.lanes, + ) + .await + .unwrap(); + assert_eq!(bindings(bag), want.into_iter().collect::>()); + } + } + + /// B2: no hidden scan. Every side Arm B reads scans exactly the rows of + /// its located ranges, and those are the rows Arm A keeps; Arm A also + /// reads the whole mask column, Arm B reads no column at all. + #[tokio::test(flavor = "multi_thread")] + async fn b2_lance_scans_exactly_the_located_ranges() { + let f = family_data(); + let (labels, rules) = resolved(&family::crepe_plan()); + let fx = fixture(&labels, &f).await; + let d = answer_demand(&labels, 17); + let kind = ResultKind::Bindings(SET); + let (_, _, io_a) = evaluate_lance(&rules, &d, kind, Options::ALL, &fx.lance) + .await + .unwrap(); + let (_, _, io_b, lookups) = + evaluate_addressed(&rules, &d, kind, Options::ALL, &fx.lance, &fx.lanes) + .await + .unwrap(); + let sides_b: Vec<&SideIo> = io_b.iter().flatten().collect(); + let looks: Vec<_> = lookups.iter().flatten().collect(); + assert_eq!(sides_b.len(), 3, "{io_b:?}"); + for (s, l) in sides_b.iter().zip(&looks) { + assert!(s.mask_column.is_none(), "{s:?}"); + assert_eq!(s.rows.rows_scanned, l.rows_in_ranges, "{s:?} {l:?}"); + assert_eq!(s.rows.rows_returned, l.rows_in_ranges, "{s:?} {l:?}"); + } + // Parent(17, y): 3 rows. Parent(y, z) for y in {1, 2, 3}: 5. Allowed(y): 3. + let scanned_b: usize = sides_b.iter().map(|s| s.rows.rows_scanned).sum(); + assert_eq!(scanned_b, 3 + 5 + 3); + let scanned_a: usize = io_a + .iter() + .flatten() + .map(|s| s.rows.rows_scanned + s.mask_column.map_or(0, |c| c.rows_scanned)) + .sum(); + // Not vacuous: Arm A reads every mask column in full. + assert_eq!(scanned_a, 3 + 5 + 3 + 2 * f.parent.len() + f.allowed.len()); + assert!(scanned_b * 10 < scanned_a, "{scanned_b} vs {scanned_a}"); + } + + /// B3: a demand that matches nothing reads nothing. + #[tokio::test(flavor = "multi_thread")] + async fn b3_a_miss_reads_no_rows() { + let f = family_data(); + let (labels, rules) = resolved(&family::crepe_plan()); + let fx = fixture(&labels, &f).await; + let d = answer_demand(&labels, 999); + let (out, _, io, _) = evaluate_addressed( + &rules, + &d, + ResultKind::Count(BAG), + Options::ALL, + &fx.lance, + &fx.lanes, + ) + .await + .unwrap(); + assert_eq!(out, Outcome::Count(0)); + let scanned: usize = io.iter().flatten().map(|s| s.rows.rows_scanned).sum(); + assert_eq!(scanned, 0, "{io:?}"); + } + + /// B4: a high-degree center. Every row of the hub is read once, and the + /// answer equals the memory path. + #[tokio::test(flavor = "multi_thread")] + async fn b4_a_hub_is_read_once_per_row() { + let mut f = family_data(); + for h in 0..500u32 { + f.parent.push([7, 10_000 + h]); + f.parent.push([10_000 + h, 9]); + f.allowed.push([10_000 + h, 10_000 + h]); + } + let (labels, rules) = resolved(&family::crepe_plan()); + let mem = store(&labels, &f, [1, 1, 1]); + let fx = fixture(&labels, &f).await; + let d = answer_demand(&labels, 7); + let kind = ResultKind::Count(BAG); + let (m, _) = evaluate(&rules, &d, kind, Options::ALL, &mem).unwrap(); + let (b, _, io, lookups) = + evaluate_addressed(&rules, &d, kind, Options::ALL, &fx.lance, &fx.lanes) + .await + .unwrap(); + assert_eq!(b, m); + assert_eq!(b, Outcome::Count(500)); + let scanned: usize = io.iter().flatten().map(|s| s.rows.rows_scanned).sum(); + assert_eq!(scanned, 3 * 500); + // The two unrestricted sides each sought the 500 distinct centers. + let sought: Vec = lookups.iter().flatten().map(|l| l.keys).collect(); + assert_eq!(sought, vec![1, 500, 500]); + } + + /// B5: a demanded key repeated by the restricted side is sought once, and + /// the bag multiplicity of the stored facts survives. + #[tokio::test(flavor = "multi_thread")] + async fn b5_a_repeated_center_is_sought_once_and_multiplicity_is_kept() { + let mut f = family_data(); + // Parent(17, 2) twice: center 2 appears twice on the restricted side. + f.parent.push([17, 2]); + let (labels, rules) = resolved(&family::crepe_plan()); + let mem = store(&labels, &f, [1, 1, 1]); + let fx = fixture(&labels, &f).await; + let d = answer_demand(&labels, 17); + let kind = ResultKind::Bindings(BAG); + let (m, _) = evaluate(&rules, &d, kind, Options::ALL, &mem).unwrap(); + let (b, _, _, lookups) = + evaluate_addressed(&rules, &d, kind, Options::ALL, &fx.lance, &fx.lanes) + .await + .unwrap(); + assert_eq!(b, m); + let sought: Vec = lookups.iter().flatten().map(|l| l.keys).collect(); + assert_eq!(sought, vec![1, 3, 3], "centers {{1, 2, 3}}, each once"); + // Not vacuous: z = 10 comes once through y = 1 and 2 x 2 times through + // y = 2 (Parent(17, 2) twice, Allowed(2) twice); z = 12 only through + // y = 2. Without the duplicate fact this is 1 + 2 + 2 = 5. + let Outcome::Bindings(rows) = b else { panic!() }; + let mult = |z: u64| rows.iter().find(|(t, _)| t[1] == z).map_or(0, |(_, n)| *n); + assert_eq!((mult(10), mult(12)), (5, 4), "{rows:?}"); + } + + /// B6: a table written out of key order gets no witness. + #[tokio::test(flavor = "multi_thread")] + async fn b6_an_unsorted_table_is_refused_at_attest() { + const R0: crepe_plan_probe::logic::RelId = crepe_plan_probe::logic::RelId(0); + let dir = tempfile::tempdir().unwrap(); + let rows = [[5u32, 1], [1, 2], [3, 3]]; + let lance = LanceStore::write_as_given(dir.path().to_str().unwrap(), &[(R0, "r", &rows)]) + .await + .unwrap(); + assert!(matches!( + attest(&lance, R0).await, + Err(AddressedRefusal::Witness(WitnessError::NotSorted { + first_inversion_at: 1 + })) + )); + // Silence twin: the same rows in key order are attested. + let dir2 = tempfile::tempdir().unwrap(); + let sorted = LanceStore::write(dir2.path().to_str().unwrap(), &[(R0, "r", &rows)]) + .await + .unwrap(); + assert!(attest(&sorted, R0).await.is_ok()); + } + + /// B7: an append after attest makes the lane stale; re-attesting serves + /// the new rows. + #[tokio::test(flavor = "multi_thread")] + async fn b7_an_append_makes_the_lane_stale() { + let f = family_data(); + let (labels, rules) = resolved(&family::crepe_plan()); + let fx = fixture(&labels, &f).await; + let d = answer_demand(&labels, 17); + let kind = ResultKind::Count(BAG); + let parent = labels.id("Parent").unwrap(); + fx.lance.append(parent, &[[999_999, 1]]).await.unwrap(); + let got = evaluate_addressed(&rules, &d, kind, Options::ALL, &fx.lance, &fx.lanes).await; + assert!( + matches!(&got, Err(PipelineRefusal::Addressed(s)) if s.starts_with("Stale")), + "{got:?}" + ); + // Re-attesting is refused too: the append landed in a second + // fragment, so lane position is no longer row address. The arm needs + // the table compacted (or rewritten) before it can address it again. + assert!(matches!( + attest(&fx.lance, parent).await, + Err(AddressedRefusal::NotContiguous { rel }) if rel == parent + )); + } + + /// B8: a side whose restriction or demand is on the second column is + /// refused by name; the lane is ordered by the first. + #[tokio::test(flavor = "multi_thread")] + async fn b8_a_demand_on_the_second_column_is_refused() { + let plan = two_way::crepe_plan(); + let (labels, rules) = resolved(&plan); + let r: Vec<[u32; 2]> = vec![[1, 10], [1, 11], [2, 10]]; + let s: Vec<[u32; 2]> = vec![[10, 9], [11, 8]]; + let dir = tempfile::tempdir().unwrap(); + let (rr, ss) = (labels.id("R").unwrap(), labels.id("S").unwrap()); + let lance = LanceStore::write( + dir.path().to_str().unwrap(), + &[(rr, "r", &r), (ss, "s", &s)], + ) + .await + .unwrap(); + let lanes = vec![ + attest(&lance, rr).await.unwrap(), + attest(&lance, ss).await.unwrap(), + ]; + let goal = labels.id("T").unwrap(); + let kind = ResultKind::Count(BAG); + // S(y, 9): the constant is on S's second column. + let d = Demand { + goal, + bound: vec![Some(1), None, Some(9)], + }; + let got = evaluate_addressed(&rules, &d, kind, Options::ALL, &lance, &lanes).await; + assert!( + matches!(&got, Err(PipelineRefusal::Addressed(e)) if e.starts_with("NotLeadingColumn")), + "{got:?}" + ); + // Silence twin: the constant on R's first column only is served. + let d = Demand { + goal, + bound: vec![Some(1), None, None], + }; + let got = evaluate_addressed(&rules, &d, kind, Options::ALL, &lance, &lanes).await; + assert_eq!(got.unwrap().0, Outcome::Count(2)); + } + + /// B9: the arm adds no program and no foreign hop. + #[tokio::test(flavor = "multi_thread")] + async fn b9_the_addressed_arm_adds_no_program() { + let f = family_data(); + let (labels, rules) = resolved(&family::crepe_plan()); + let mem = store(&labels, &f, [1, 1, 1]); + let fx = fixture(&labels, &f).await; + let d = answer_demand(&labels, 17); + let kind = ResultKind::Bindings(SET); + let (_, rm) = evaluate(&rules, &d, kind, Options::ALL, &mem).unwrap(); + let (_, rb, _, _) = + evaluate_addressed(&rules, &d, kind, Options::ALL, &fx.lance, &fx.lanes) + .await + .unwrap(); + assert_eq!(rb.programs.len(), rm.programs.len()); + assert_eq!(rb.rows_prepared, rm.rows_prepared); + } +} From cc578f269c2a16e39c0ade52a2cfd1d587ddd5dc Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 20 Oct 2026 19:29:14 +0000 Subject: [PATCH 2/6] crepe-plan-probe: Arm A vs Arm B measurement by selectivity Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f --- crates/crepe-plan-probe/examples/addressed.rs | 216 ++++++++++++++++++ 1 file changed, 216 insertions(+) create mode 100644 crates/crepe-plan-probe/examples/addressed.rs diff --git a/crates/crepe-plan-probe/examples/addressed.rs b/crates/crepe-plan-probe/examples/addressed.rs new file mode 100644 index 000000000..f494fb5ec --- /dev/null +++ b/crates/crepe-plan-probe/examples/addressed.rs @@ -0,0 +1,216 @@ +//! Arm A (scan semijoin) against Arm B (addressed semijoin) over Lance 13 +//! tables, by demand selectivity. Run in release from this crate's directory: +//! +//! ```text +//! cargo run --release --example addressed +//! ``` +//! +//! The query is `Answer(17, z)` over the `family` program. `Parent` holds +//! `N = 50,000` background edges `[y, z]` with distinct `y`, every `y` is +//! `Allowed`, and `x = 17` has `k` children chosen among those `y`. So the +//! semijoin centers are exactly `k` and the answer is `k` rows. +//! +//! Modes, each a median of `REPS` runs: +//! - `full`: semijoin off; every unrestricted side is read in full. +//! - `arm A`: the scan semijoin; each side reads its key column in full, +//! builds the address mask and reads the masked rows. +//! - `arm B warm`: the addressed semijoin with the lanes already attested. +//! - `attest`: reading the keys of the three relations and attesting them +//! (Arm B's one-time cost `P`; `arm B cold` = `attest` + `arm B warm`). +//! +//! Break-even: Arm B's attest pays for itself after `ceil(P / S)` queries, +//! where `S` is Arm A's per-query time minus Arm B warm's. + +use std::time::{Duration, Instant}; + +use crepe_plan_probe::addressed::{AddressedRelation, attest}; +use crepe_plan_probe::lance_store::{LanceStore, SideIo}; +use crepe_plan_probe::logic::{Labels, Rule, resolve}; +use crepe_plan_probe::optimize::{Demand, Options, ResultKind}; +use crepe_plan_probe::pipeline::{Outcome, evaluate_addressed, evaluate_lance}; +use crepe_plan_probe::programs::family; +use mork_pathmap_probe::Semantics; + +const N: u32 = 50_000; +const REPS: usize = 9; +const NO_SEMIJOIN: Options = Options { + propagate: true, + dedup: true, + semijoin: false, +}; +const KIND: ResultKind = ResultKind::Count(Semantics::Bag); + +fn median(mut v: Vec) -> Duration { + v.sort(); + v[v.len() / 2] +} + +#[derive(Default, Clone, Copy)] +struct Read { + rows: usize, + bytes: usize, +} + +fn read_of(io: &[Vec]) -> Read { + let mut r = Read::default(); + for s in io.iter().flatten() { + r.rows += s.rows.rows_scanned; + r.bytes += s.rows.bytes_read; + if let Some(c) = s.mask_column { + r.rows += c.rows_scanned; + r.bytes += c.bytes_read; + } + } + r +} + +fn ms(d: Duration) -> f64 { + d.as_secs_f64() * 1e3 +} + +struct Mode { + time: Duration, + prepare: Duration, + read: Read, +} + +async fn lance_mode(rules: &[Rule], d: &Demand, opts: Options, st: &LanceStore, k: u64) -> Mode { + let mut times = Vec::new(); + let mut preps = Vec::new(); + let mut read = Read::default(); + for _ in 0..REPS { + let t = Instant::now(); + let (out, rep, io) = evaluate_lance(rules, d, KIND, opts, st).await.unwrap(); + times.push(t.elapsed()); + preps.push(rep.costs.prepare); + assert_eq!(out, Outcome::Count(k)); + read = read_of(&io); + } + Mode { + time: median(times), + prepare: median(preps), + read, + } +} + +async fn arm_b( + rules: &[Rule], + d: &Demand, + st: &LanceStore, + lanes: &[AddressedRelation], + k: u64, +) -> Mode { + let mut times = Vec::new(); + let mut preps = Vec::new(); + let mut read = Read::default(); + for _ in 0..REPS { + let t = Instant::now(); + let (out, rep, io, _) = evaluate_addressed(rules, d, KIND, Options::ALL, st, lanes) + .await + .unwrap(); + times.push(t.elapsed()); + preps.push(rep.costs.prepare); + assert_eq!(out, Outcome::Count(k)); + read = read_of(&io); + } + Mode { + time: median(times), + prepare: median(preps), + read, + } +} + +#[tokio::main(flavor = "multi_thread")] +async fn main() { + let plan = family::crepe_plan(); + let labels = Labels::of(&plan); + let rules: Vec = resolve(&plan, &labels) + .into_iter() + .map(|r| r.expect("admitted rule")) + .collect(); + let rels = [ + labels.id("Parent").unwrap(), + labels.id("Allowed").unwrap(), + labels.id("Sponsor").unwrap(), + ]; + let d = Demand { + goal: labels.id("Answer").unwrap(), + bound: vec![Some(17), None], + }; + let ks: [u32; 7] = [0, 1, 8, N / 100, N / 10, N / 2, N]; + + println!( + "{:>7} | {:>9} {:>9} {:>9} {:>9} | {:>9} {:>9} {:>9} | {:>9} {:>9} {:>9} | {:>6}", + "k", + "full ms", + "A ms", + "B ms", + "attest", + "full rows", + "A rows", + "B rows", + "A prep", + "B prep", + "B bytes", + "N*" + ); + for k in ks { + let mut parent: Vec<[u32; 2]> = (0..N).map(|i| [100_000 + i, 300_000 + i]).collect(); + // x = 17 sits below every background y, so the restricted range is + // the table's first k rows; spread the chosen children over the + // background instead of taking a prefix. + let step = if k == 0 { 1 } else { N / k }; + parent.extend((0..k).map(|j| [17, 100_000 + j * step])); + let allowed: Vec<[u32; 2]> = (0..N).map(|i| [100_000 + i, 100_000 + i]).collect(); + let sponsor: Vec<[u32; 2]> = vec![[20, 20], [21, 21]]; + let dir = tempfile::tempdir().unwrap(); + let st = LanceStore::write( + dir.path().to_str().unwrap(), + &[ + (rels[0], "parent", &parent), + (rels[1], "allowed", &allowed), + (rels[2], "sponsor", &sponsor), + ], + ) + .await + .unwrap(); + + let mut attests = Vec::new(); + let mut lanes = Vec::new(); + for _ in 0..REPS { + let t = Instant::now(); + let mut l = Vec::new(); + for &r in &rels { + l.push(attest(&st, r).await.unwrap()); + } + attests.push(t.elapsed()); + lanes = l; + } + let p = median(attests); + + let full = lance_mode(&rules, &d, NO_SEMIJOIN, &st, u64::from(k)).await; + let a = lance_mode(&rules, &d, Options::ALL, &st, u64::from(k)).await; + let b = arm_b(&rules, &d, &st, &lanes, u64::from(k)).await; + let saving = a.time.saturating_sub(b.time); + let break_even = if saving.is_zero() { + "never".to_string() + } else { + (p.as_nanos().div_ceil(saving.as_nanos())).to_string() + }; + println!( + "{:>7} | {:>9.3} {:>9.3} {:>9.3} {:>9.3} | {:>9} {:>9} {:>9} | {:>9.3} {:>9.3} {:>9} | {:>6}", + k, + ms(full.time), + ms(a.time), + ms(b.time), + ms(p), + full.read.rows, + a.read.rows, + b.read.rows, + ms(a.prepare), + ms(b.prepare), + b.read.bytes, + break_even + ); + } +} From d0f308ae6bc9169930300584c471c6513aa89247 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 20 Oct 2026 19:46:00 +0000 Subject: [PATCH 3/6] crepe-plan-probe: Arm B measured by selectivity; row filter re-measured MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Plan §2e records the addressed semijoin at 50,000 rows: - Arm B scans 3k rows where Arm A scans at least 150,000. - Up to 1% of centers it is 2.5-12x faster than Arm A, and its 7 ms attest pays for itself after 2-3 queries. - From 10% of centers up a full scan wins; with all centers Arm B never breaks even. §2c is re-measured with the row_filter fix. The contiguous column beats the scalar test up to 64 needles in both runs. At 128 needles the order differs between runs, so that crossover is not pinned. The example also passes clippy (manual_checked_ops). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f --- .../2026-10-20-crepe-rule-optimizer-spike.md | 4 +- .../plans/crepe-rule-optimizer-spike-v1.md | 56 ++++++++++++++++--- crates/crepe-plan-probe/examples/addressed.rs | 9 ++- 3 files changed, 55 insertions(+), 14 deletions(-) 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 f73f2ba59..b93dca172 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 @@ -18,4 +18,6 @@ - Dynamic MeTTa rules need crepe's analysis as a library: crepe is a proc-macro crate (follow-up A, `crepe-core` split). - 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. +- 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–6 µs per needle at 50,000 rows and beats the scalar test up to 64 needles; at 128 needles the order differs between runs (re-measured after the #1632 Codex fix). Next step: a column-major relation store. +- Stage 3 over Lance 13 / lancedb 0.40 built (`lance_store`): each side read under a relalg address mask with `with_row_addr_prefilter`; every masked read scans exactly its mask (P1–P4 green). The relalg membership mask holds 4 centers at most (one scratch slot per `eq` atom, `MAX_SLOTS = 8`); above that a scalar test builds it. +- Arm B built (`addressed`): lanes in key order, attested with `SealedFacetLane::attest_sorted`, read only at the `seek_from` ranges of the demanded keys. At 50,000 rows it scans `3k` rows where Arm A scans at least 150,000. It is 2.5–12× faster than Arm A up to 1% of centers and pays back its 7 ms attest after 2–3 queries. From 10% of centers up a full scan wins, and with all centers Arm B never breaks even. Open: the selectivity crossover between 1% and 10%; multi-fragment lanes (an append currently forces a compaction); a second lane for demands on the second column. diff --git a/.claude/plans/crepe-rule-optimizer-spike-v1.md b/.claude/plans/crepe-rule-optimizer-spike-v1.md index 17a2b2e86..0c2835e0b 100644 --- a/.claude/plans/crepe-rule-optimizer-spike-v1.md +++ b/.claude/plans/crepe-rule-optimizer-spike-v1.md @@ -28,8 +28,9 @@ Pipeline, one stage per module of the probe: - [x] Phase 5: differential tests A–M against `crepe!`, datafrog, MORK, reference; 9 disable runs - [x] Phase 6 (limited): Soufflé harvest table (§3) - [x] Semijoin reduction before preparation (`Options::semijoin`, `native::prepare_semijoin`; §2b, tests N) -- [ ] Stage 3 over Lance 13 / lancedb 0.40: masked read per side, predictions P1–P5 (§2d) -- [ ] CI workflow for this probe +- [x] Stage 3 over Lance 13 / lancedb 0.40: masked read per side, P1–P4 green (§2d); P5 measured at 50,000 rows only (§2e) +- [x] CI workflow for this probe (`.github/workflows/crepe-plan-probe.yml`) +- [x] Arm B, the addressed semijoin over attested lanes (§2e, tests B1–B9) ## 1. Integration point @@ -97,14 +98,16 @@ The semijoin filter (and the constant restriction) is a scalar test per row: bin | 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 | +| 1 | 30.9 | 82.7 | 5.4 | +| 8 | 146.4 | 665.9 | 30.5 | +| 32 | 225.3 | 2,892.2 | 114.9 | +| 64 | 278.2 | 4,988.1 | 238.0 | +| 128 | 518.1 | 10,043.3 | 462.0 | + +Re-measured 2026-10-20 after the Codex P2 fix on #1632. The contiguous arm used to return the mask; it now collects the kept rows like the other two arms, and the example asserts that its rows equal the scalar test's. Collecting the rows costs at most 1.8 µs per run. - 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). +- Over a contiguous column the same compares are 14–22× faster than the strided ones, and they beat the scalar test up to 64 needles in both runs. At 128 needles the order differs between runs: this run 462 vs 518 µs, the first run 485 vs 375 µs. The scalar time moves by up to 1.8× between runs (17.4 vs 30.9 µs at one needle), so the crossover lies near 64–128 needles and is not stable enough to pin. At 1,000 rows the scalar test wins at 128 (10.0 vs 7.2 µs). - 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. @@ -131,6 +134,40 @@ Predictions, each with its falsifier: Disables: build `R` from the wrong column (P1, P2 red); drop the prefilter (P2 red); skip the version check (P4 red). +## 2e. Arm B: the addressed semijoin (measured 2026-10-20) + +DECISION (operator, 2026-10-20): "They are ordered by design. The only issue is the relationship maps proposed earlier pulling in global to local which would make the nodeguid order unusual. Build arm b". + +A relation is written in key order: row `(a, b)` carries `FacetCascade::from_semantic_tiles([0, 0, a_hi, a_lo, b_hi, b_lo, 0, 0])`, which compares as `(a, b)`. The probe never orders a lane itself. `addressed::attest` reads the keys once and calls `SealedFacetLane::attest_sorted`, which refuses an unsorted lane. A side then reads only the lane ranges of its demanded keys, located with `seek_from` under the lane's `OrderedLaneWitness`, through `with_row_addr_prefilter`. The arm adds no mask column, index, opcode, gather or global-to-local mapping. + +The arm refuses, by name: +- `NotLeadingColumn`: a restriction or demand on the second column (that would need a second lane, ordered by that column); +- `Stale`: a table that moved on since its lane was attested; +- `NotSorted`: an unsorted table, at attest; +- `NotContiguous`: a table split over more than one fragment, at attest. An append adds a fragment, so after one the table must be compacted before it can be addressed again. + +Measured with `cargo run --release --example addressed` (lance 13, lancedb 0.40, rustc 1.99.0). The query is `Answer(17, z)` over `family`. There are N = 50,000 background edges with distinct `y`, all `Allowed`, and `k` children of 17 spread over them, so there are exactly `k` centers and `k` answers. Times are medians of 9 runs. "Rows" counts rows Lance scanned, mask columns included. + +| k | full ms | Arm A ms | Arm B warm ms | attest ms | full rows | Arm A rows | Arm B rows | break-even | +|---|---|---|---|---|---|---|---|---| +| 0 | 9.81 | 4.46 | 0.37 | 7.28 | 150,000 | 150,000 | 0 | 2 | +| 1 | 9.95 | 5.02 | 2.19 | 6.83 | 150,003 | 150,005 | 3 | 3 | +| 8 | 12.30 | 6.60 | 2.24 | 6.67 | 150,024 | 150,040 | 24 | 2 | +| 500 (1%) | 10.42 | 8.86 | 3.47 | 6.71 | 151,500 | 152,500 | 1,500 | 2 | +| 5,000 (10%) | 11.70 | 15.02 | 12.11 | 7.81 | 165,000 | 175,000 | 15,000 | 3 | +| 25,000 (50%) | 18.55 | 41.39 | 34.54 | 7.74 | 225,000 | 275,000 | 75,000 | 2 | +| 50,000 (all) | 30.71 | 36.15 | 38.98 | 9.84 | 300,000 | 400,000 | 150,000 | never | + +Break-even is `ceil(P / S)`, where `P` is the attest time and `S` is Arm A's per-query time minus Arm B warm's. + +What this says: +- **Source-row reads do go down, by construction, and exactly.** Arm B scans `3k` rows: one row per demanded key on each of the three sides. Arm A scans at least `150,000`, because each side reads its key column in full to build the mask. The answer to "does demand-aware semijoin reduction decrease source-row reads" is yes for Arm B and no for Arm A. Arm A reduces the rows laid out (§2b), not the rows read. +- **Time follows the rows only at low selectivity.** Up to 1% of centers, Arm B is 2.5–12× faster than Arm A and the attest pays for itself after 2–3 queries. Its floor is about 2 ms for three masked reads, even for 3 rows; with no centers it reads nothing and takes 0.37 ms. +- **From 10% of centers up, every masked read loses to a full scan.** Arm A is already slower than the full read at 10%, and both arms are slower at 50%. With all centers, Arm B never breaks even. Many small ranges cost Lance more per row than one sequential scan. A planner must switch to the full read above a crossover between 1% and 10%; that crossover is not measured more finely here. +- `P` (reading and attesting 100,002 keys over three tables) is 7–10 ms: a cold cost the size of one full query. + +P5 is measured at 50,000 rows only; 1,000 and 10,000 were not run. + ## 3. Soufflé harvest (souffle-lang/souffle `f53dab8`, UPL-1.0; algorithms only, no source copied) | Soufflé | transformation | precondition | in crepe | in lance-graph | taken | @@ -173,6 +210,9 @@ Known limits: rule dedup does not search body permutations; the star must have o ## 6. Follow-ups (not built) - ~~Semijoin reduction before preparation~~ — built, §2b. Layout O(k); the scan stays O(n). +- ~~Addressed read~~ — built, §2e. Reads O(k) rows over attested lanes. +- Selectivity crossover: measure between 1% and 10% of centers and switch Arm B to the full read above it. +- Multi-fragment lanes: map lane positions to row addresses per fragment, so that an append does not force a compaction before Arm B can read the table again. - **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). diff --git a/crates/crepe-plan-probe/examples/addressed.rs b/crates/crepe-plan-probe/examples/addressed.rs index f494fb5ec..4904c2692 100644 --- a/crates/crepe-plan-probe/examples/addressed.rs +++ b/crates/crepe-plan-probe/examples/addressed.rs @@ -159,7 +159,7 @@ async fn main() { // x = 17 sits below every background y, so the restricted range is // the table's first k rows; spread the chosen children over the // background instead of taking a prefix. - let step = if k == 0 { 1 } else { N / k }; + let step = N.checked_div(k).unwrap_or(1); parent.extend((0..k).map(|j| [17, 100_000 + j * step])); let allowed: Vec<[u32; 2]> = (0..N).map(|i| [100_000 + i, 100_000 + i]).collect(); let sponsor: Vec<[u32; 2]> = vec![[20, 20], [21, 21]]; @@ -192,10 +192,9 @@ async fn main() { let a = lance_mode(&rules, &d, Options::ALL, &st, u64::from(k)).await; let b = arm_b(&rules, &d, &st, &lanes, u64::from(k)).await; let saving = a.time.saturating_sub(b.time); - let break_even = if saving.is_zero() { - "never".to_string() - } else { - (p.as_nanos().div_ceil(saving.as_nanos())).to_string() + let break_even = match saving.as_nanos() { + 0 => "never".to_string(), + s => p.as_nanos().div_ceil(s).to_string(), }; println!( "{:>7} | {:>9.3} {:>9.3} {:>9.3} {:>9.3} | {:>9} {:>9} {:>9} | {:>9.3} {:>9.3} {:>9} | {:>6}", From 068afb8a83610475eccc32d235b272802e5bb337 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 20 Oct 2026 19:46:00 +0000 Subject: [PATCH 4/6] thought-git: state what the probe implements The full thought git API (thought-git-api-v1.md) is still a proposal. crates/thought-git-probe implements the slot pool, the oracle and the tests. It does not implement the production land, rebase or persistence pipeline; D-TGB-2..5 stay Queued. D-TGB-0 and D-TGB-1 merged with #1630/#1631, so their board rows now read Shipped (lab) instead of In PR. The module doc no longer says that no thought-git code exists. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f --- .claude/board/STATUS_BOARD.md | 4 ++-- .../board/entries/2026-10-20-d-tgb-thought-git-basics.md | 4 ++++ crates/thought-git-probe/src/lib.rs | 6 +++++- 3 files changed, 11 insertions(+), 3 deletions(-) diff --git a/.claude/board/STATUS_BOARD.md b/.claude/board/STATUS_BOARD.md index 3042116ef..de03cf737 100644 --- a/.claude/board/STATUS_BOARD.md +++ b/.claude/board/STATUS_BOARD.md @@ -355,8 +355,8 @@ Plan: `.claude/plans/ce64-hypothesis-layers-v1.md`. Plateau: `.claude/knowledge/ | **D-TGA-4** | land through the revision gate into the next Lance version (D-HL-8) | Queued | landing a thought that is behind is refused | | **D-TGA-5** | squash_into (partial squash) and blame | Queued | blame returns original sources after slot reuse | | **D-TGA-6** | MergeQueue | Queued | independent thoughts == landed by hand in FIFO order; a thought reading another negated lands after it; a cycle through negation refused at enqueue | -| **D-TGB-0** | thought git basics: fixtures per write-intention outcome, squash-stack and anomaly fixtures, absent-not-zero fixture, from-scratch oracle (`thought-git-basics-v1.md`) | In PR (lab probe `crates/thought-git-probe`, 19 tests; `PREREGISTRATION.md`) | 5 oracle disables verified red-then-green; every signal has a can-fire and a can-stay-silent case on real contract code; the pearl half on edges moves to D-TGB-3 | -| **D-TGB-1** | slot pool, autonomous: taken by `branch`, freed by `squash`/`discard`/`land` (the write is the Rubicon decision, 2026-10-20) | In PR (lab probe `crates/thought-git-probe/src/slots.rs`) | at 63 taken `branch` picks a FIFO or least-used candidate, squashes it along its dependency chain if it has one, else evicts it (DECISIONS 2026-10-20; was: squash unremarkable first, else refuse); stale handles refused; 6 tests | +| **D-TGB-0** | thought git basics: fixtures per write-intention outcome, squash-stack and anomaly fixtures, absent-not-zero fixture, from-scratch oracle (`thought-git-basics-v1.md`) | Shipped (lab, #1630; probe `crates/thought-git-probe`, 19 tests; `PREREGISTRATION.md`) | 5 oracle disables verified red-then-green; every signal has a can-fire and a can-stay-silent case on real contract code; the pearl half on edges moves to D-TGB-3 | +| **D-TGB-1** | slot pool, autonomous: taken by `branch`, freed by `squash`/`discard`/`land` (the write is the Rubicon decision, 2026-10-20) | Shipped (lab, #1630/#1631; `crates/thought-git-probe/src/slots.rs`). Slot pool only: no production land, rebase or persistence (D-TGB-2..5 Queued) | at 63 taken `branch` picks a FIFO or least-used candidate, squashes it along its dependency chain if it has one, else evicts it (DECISIONS 2026-10-20; was: squash unremarkable first, else refuse); stale handles refused; 6 tests | | **D-TGB-2** | `w_slot` lane labels (W = thought layer, not a pointer) qualified by base version / slot generation | Queued (lab) | no live cell carries a landed slot; old-generation edge refused | | **D-TGB-3** | `Commit` lands one sparse batch through `LanceCycleWriter`; `DurableRef` = dataset + `Ref::Version(branch, version)` | Queued (lab) | empty cycle writes nothing; re-submit reconciles; two branches at version 7 told apart | | **D-TGB-4** | receipt (base ref, program id, seeds, slot generation) instead of history | Queued (lab) | replay from receipt reproduces writes cell for cell | diff --git a/.claude/board/entries/2026-10-20-d-tgb-thought-git-basics.md b/.claude/board/entries/2026-10-20-d-tgb-thought-git-basics.md index 5e09a66f3..66260a364 100644 --- a/.claude/board/entries/2026-10-20-d-tgb-thought-git-basics.md +++ b/.claude/board/entries/2026-10-20-d-tgb-thought-git-basics.md @@ -14,3 +14,7 @@ OPEN: card granularity (mailbox card or per-move card); the PR entry under MemWA Correction (2026-10-20, review #1628): `LanceCycleWriter` commits one batch per supervisor cycle, so every `Commit` in a cycle coalesces into one batch and one version; a landed thought is (version, thought id within the batch). D-TGB-2 is blocked on the `w_slot` reading decision. Correction (2026-10-20, operator): D-TGB-2 is not blocked. W = thought layer 0..63 is the 2026-10-17 decision (#1597); D-TGB-2 carries the migration of the competing readers (`MailboxSoA::apply_edges`, `deepnsm-v2`'s `WitnessTable` index). + +## Status note (2026-10-20, later) + +The full thought git API (`thought-git-api-v1.md`, D-TGA) is still a PROPOSAL. `crates/thought-git-probe` implements the slot pool (D-TGB-1), the from-scratch oracle and the tests (D-TGB-0). It does not implement the production land, rebase or persistence pipeline (D-TGB-2..5 are Queued). diff --git a/crates/thought-git-probe/src/lib.rs b/crates/thought-git-probe/src/lib.rs index d52657c15..4a13cd988 100644 --- a/crates/thought-git-probe/src/lib.rs +++ b/crates/thought-git-probe/src/lib.rs @@ -1,7 +1,11 @@ //! D-TGB-0 — the pre-registration for thought git //! (`.claude/plans/thought-git-basics-v1.md`, work item D-TGB-0). //! -//! Three things live here, all written before any thought-git code exists: +//! The full thought git API (`thought-git-api-v1.md`) is still a proposal. +//! This probe holds the pre-registration and the first lab piece, the slot +//! pool; it has no production land, rebase or persistence pipeline. +//! +//! The pre-registration, written before any thought-git code existed: //! //! - [`fixtures`]: the thoughts every later D-TGB item is tested on, one per //! Rubicon outcome, a stack of unremarkable thoughts, one thought per From 42505f1ab9e9f924c49c52d38c355704b201495a Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 20 Oct 2026 20:01:55 +0000 Subject: [PATCH 5/6] Address #1633 review: out-of-range constants, full reads, dataset identity, CI paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - A demand constant above u32::MAX now matches nothing on both Lance arms, as on the memory path. Before, Arm A refused it with Lance and Arm B refused it with NotLeadingColumn. - Arm B no longer refuses a side keyed by its second column when it has no centers to look up. Such a side is read in full, so the lane order does not matter; only a lookup needs the leading column. - An attested lane records its dataset's URI. A lane used against another table with the same relation id and version is refused (OtherDataset). Before, its row addresses were applied to the wrong table. - The probe workflow also runs on lance-graph-contract changes. - Plan §2c: the 128-needle disagreement is called benchmark noise, which is what it is. Tests B10-B12. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f --- .../2026-10-20-crepe-rule-optimizer-spike.md | 2 +- .../plans/crepe-rule-optimizer-spike-v1.md | 2 +- .github/workflows/crepe-plan-probe.yml | 5 +- crates/crepe-plan-probe/src/addressed.rs | 46 ++++++++-- crates/crepe-plan-probe/src/lance_store.rs | 9 +- crates/crepe-plan-probe/tests/pipeline.rs | 89 +++++++++++++++++++ 6 files changed, 138 insertions(+), 15 deletions(-) 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 b93dca172..966439d99 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 @@ -18,6 +18,6 @@ - Dynamic MeTTa rules need crepe's analysis as a library: crepe is a proc-macro crate (follow-up A, `crepe-core` split). - 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–6 µs per needle at 50,000 rows and beats the scalar test up to 64 needles; at 128 needles the order differs between runs (re-measured after the #1632 Codex fix). Next step: a column-major relation store. +- 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–6 µs per needle at 50,000 rows and beats the scalar test up to 64 needles; at 128 needles two runs disagree because of benchmark noise (the scalar time moved 1.8× between runs), so that crossover is not pinned (re-measured after the #1632 Codex fix). Next step: a column-major relation store. - Stage 3 over Lance 13 / lancedb 0.40 built (`lance_store`): each side read under a relalg address mask with `with_row_addr_prefilter`; every masked read scans exactly its mask (P1–P4 green). The relalg membership mask holds 4 centers at most (one scratch slot per `eq` atom, `MAX_SLOTS = 8`); above that a scalar test builds it. - Arm B built (`addressed`): lanes in key order, attested with `SealedFacetLane::attest_sorted`, read only at the `seek_from` ranges of the demanded keys. At 50,000 rows it scans `3k` rows where Arm A scans at least 150,000. It is 2.5–12× faster than Arm A up to 1% of centers and pays back its 7 ms attest after 2–3 queries. From 10% of centers up a full scan wins, and with all centers Arm B never breaks even. Open: the selectivity crossover between 1% and 10%; multi-fragment lanes (an append currently forces a compaction); a second lane for demands on the second column. diff --git a/.claude/plans/crepe-rule-optimizer-spike-v1.md b/.claude/plans/crepe-rule-optimizer-spike-v1.md index 0c2835e0b..d49c25473 100644 --- a/.claude/plans/crepe-rule-optimizer-spike-v1.md +++ b/.claude/plans/crepe-rule-optimizer-spike-v1.md @@ -107,7 +107,7 @@ The semijoin filter (and the constant restriction) is a scalar test per row: bin Re-measured 2026-10-20 after the Codex P2 fix on #1632. The contiguous arm used to return the mask; it now collects the kept rows like the other two arms, and the example asserts that its rows equal the scalar test's. Collecting the rows costs at most 1.8 µs per run. - 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–22× faster than the strided ones, and they beat the scalar test up to 64 needles in both runs. At 128 needles the order differs between runs: this run 462 vs 518 µs, the first run 485 vs 375 µs. The scalar time moves by up to 1.8× between runs (17.4 vs 30.9 µs at one needle), so the crossover lies near 64–128 needles and is not stable enough to pin. At 1,000 rows the scalar test wins at 128 (10.0 vs 7.2 µs). +- Over a contiguous column the same compares are 14–22× faster than the strided ones, and they beat the scalar test up to 64 needles in both runs. At 128 needles the two runs disagree: this run 462 vs 518 µs, the first run 485 vs 375 µs. The disagreement is benchmark noise, not a property of the kernels: the scalar time moved by up to 1.8× between runs (17.4 vs 30.9 µs at one needle). The crossover lies near 64–128 needles; pinning it needs a quieter benchmark (more repetitions, pinned CPU). At 1,000 rows the scalar test wins at 128 (10.0 vs 7.2 µs). - 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. diff --git a/.github/workflows/crepe-plan-probe.yml b/.github/workflows/crepe-plan-probe.yml index 2811c8f17..73699ee93 100644 --- a/.github/workflows/crepe-plan-probe.yml +++ b/.github/workflows/crepe-plan-probe.yml @@ -3,12 +3,14 @@ name: Crepe plan probe # git deps on the crepe and MORK forks, lance 13 / lancedb 0.40), so no # workspace job builds it. It runs from its own directory so the cargo config # above it and its own `.cargo/config.toml` apply. It path-deps relalg, -# mask-risc and mork-pathmap-probe, so changes there run it too. +# mask-risc, mork-pathmap-probe and lance-graph-contract (the ordered lane and +# facet keys of arm B), so changes there run it too. on: pull_request: paths: - 'crates/crepe-plan-probe/**' - 'crates/lance-graph-relalg/**' + - 'crates/lance-graph-contract/**' - 'crates/lance-graph-mask-risc/**' - 'crates/mork-pathmap-probe/**' - '.github/workflows/crepe-plan-probe.yml' @@ -18,6 +20,7 @@ on: paths: - 'crates/crepe-plan-probe/**' - 'crates/lance-graph-relalg/**' + - 'crates/lance-graph-contract/**' - 'crates/lance-graph-mask-risc/**' - 'crates/mork-pathmap-probe/**' - '.github/workflows/crepe-plan-probe.yml' diff --git a/crates/crepe-plan-probe/src/addressed.rs b/crates/crepe-plan-probe/src/addressed.rs index c5111c0d2..b7b379213 100644 --- a/crates/crepe-plan-probe/src/addressed.rs +++ b/crates/crepe-plan-probe/src/addressed.rs @@ -15,8 +15,10 @@ //! lane's own positions, in NodeGuid order. //! //! What this arm refuses, by name: -//! - a side whose restriction or demand is on the second column (the lane is -//! ordered by the first; the reverse order is a second lane, not built); +//! - a side whose restriction or lookup is on the second column (the lane is +//! ordered by the first; the reverse order is a second lane, not built). A +//! side read in full, with no centers to look up, may be keyed either way; +//! - a lane attested on another dataset (`OtherDataset`); //! - a lane whose table moved on since it was attested (`Stale`); //! - a table whose rows are not in key order (`NotSorted`, at attest); //! - a table that is not one contiguous fragment (`NotContiguous`, at attest). @@ -64,6 +66,12 @@ pub enum AddressedRefusal { NoLane { rel: RelId, }, + /// The lane was attested on another dataset. + OtherDataset { + rel: RelId, + attested: String, + current: String, + }, /// The side would need the lane ordered by its second column. NotLeadingColumn { side: usize, @@ -79,6 +87,10 @@ impl From for AddressedRefusal { /// One relation's attested lane. pub struct AddressedRelation { pub rel: RelId, + /// The dataset the lane was read from. A lane is only valid against the + /// table it was attested on, not another table with the same relation + /// id and version. + pub uri: String, pub lane: SealedFacetLane, pub witness: OrderedLaneWitness, } @@ -94,7 +106,12 @@ pub async fn attest(store: &LanceStore, rel: RelId) -> Result Result<(Vec, bool), LanceStoreError> { @@ -191,6 +208,13 @@ async fn current_lane<'a>( let rel = plan.sides[i].key.rel; let a = lane_for(lanes, rel)?; let ds = store.dataset(rel).await?; + if ds.uri() != a.uri { + return Err(AddressedRefusal::OtherDataset { + rel, + attested: a.uri.clone(), + current: ds.uri().to_string(), + }); + } let current = ds.version().version; if current != a.lane.version() { return Err(AddressedRefusal::Stale { @@ -227,10 +251,12 @@ pub async fn prepare_addressed( if key.key_col != 1 { return Err(AddressedRefusal::NotLeadingColumn { side: i }); } - let k = u32::try_from(restricted(i).expect("restricted")) - .map_err(|_| AddressedRefusal::NotLeadingColumn { side: i })?; let (a, ds) = current_lane(plan, i, store, lanes).await?; - let (r, n) = ranges(a, &[k])?; + // A constant above `u32::MAX` matches no stored value: no range. + let (r, n) = match u32::try_from(restricted(i).expect("restricted")) { + Ok(k) => ranges(a, &[k])?, + Err(_) => (Vec::new(), 0), + }; let (got, rows_io) = read_rows(&ds, Some(range_mask(&r))).await?; lookups[i] = Lookup { keys: 1, @@ -275,11 +301,13 @@ pub async fn prepare_addressed( for i in (0..n_sides).filter(|&i| restricted(i).is_none()) { let key = plan.sides[i].key; - if key.key_col != 0 { - return Err(AddressedRefusal::NotLeadingColumn { side: i }); - } let (a, ds) = current_lane(plan, i, store, lanes).await?; let (mask, n, sought) = match ¢ers { + // Without centers the side is read in full, so its key column + // may be either; only a lookup needs the leading column. + Some(_) if key.key_col != 0 => { + return Err(AddressedRefusal::NotLeadingColumn { side: i }); + } Some(c) => { let (r, n) = ranges(a, c)?; (Some(range_mask(&r)), n, c.len()) diff --git a/crates/crepe-plan-probe/src/lance_store.rs b/crates/crepe-plan-probe/src/lance_store.rs index 90b2ff778..3317d78bb 100644 --- a/crates/crepe-plan-probe/src/lance_store.rs +++ b/crates/crepe-plan-probe/src/lance_store.rs @@ -303,9 +303,12 @@ pub async fn prepare_lance( let ds = store.dataset(key.rel).await?; let other = 1 - key.key_col as usize; let (values, addr, col_io) = read_column(&ds, other).await?; - let k = u32::try_from(restricted(i).expect("restricted")) - .map_err(|_| LanceStoreError::Lance(format!("restriction outside u32 on side {i}")))?; - let mask = address_mask(&values, &Rel::eq_u32(0, k)).map_err(LanceStoreError::Relalg)?; + // A constant above `u32::MAX` matches no stored value: an empty mask, + // as the memory path finds no row. + let mask = match u32::try_from(restricted(i).expect("restricted")) { + Ok(k) => address_mask(&values, &Rel::eq_u32(0, k)).map_err(LanceStoreError::Relalg)?, + Err(_) => vec![0; words_for(values.len())], + }; let (m, r) = row_addr_mask(&mask, values.len(), &addr); let (got, rows_io) = read_rows(&ds, Some(m)).await?; io[i] = SideIo { diff --git a/crates/crepe-plan-probe/tests/pipeline.rs b/crates/crepe-plan-probe/tests/pipeline.rs index 4d971f727..a1bfbda8d 100644 --- a/crates/crepe-plan-probe/tests/pipeline.rs +++ b/crates/crepe-plan-probe/tests/pipeline.rs @@ -1345,6 +1345,95 @@ mod arm_b { assert_eq!(got.unwrap().0, Outcome::Count(2)); } + /// B10: a demand constant above `u32::MAX` matches nothing on every path + /// (no stored value is that large), instead of being refused. + #[tokio::test(flavor = "multi_thread")] + async fn b10_a_constant_above_u32_matches_nothing() { + let f = family_data(); + let (labels, rules) = resolved(&family::crepe_plan()); + let mem = store(&labels, &f, [1, 1, 1]); + let fx = fixture(&labels, &f).await; + let d = answer_demand(&labels, u64::from(u32::MAX) + 1); + for kind in kinds() { + let (m, _) = evaluate(&rules, &d, kind, Options::ALL, &mem).unwrap(); + let (a, _, _) = evaluate_lance(&rules, &d, kind, Options::ALL, &fx.lance) + .await + .unwrap(); + let (b, _, io, _) = + evaluate_addressed(&rules, &d, kind, Options::ALL, &fx.lance, &fx.lanes) + .await + .unwrap(); + assert_eq!(a, m, "{kind:?}"); + assert_eq!(b, m, "{kind:?}"); + let scanned: usize = io.iter().flatten().map(|s| s.rows.rows_scanned).sum(); + assert_eq!(scanned, 0, "{kind:?}"); + } + // Not vacuous: the memory path answers nothing here. + let (m, _) = evaluate(&rules, &d, ResultKind::Count(BAG), Options::ALL, &mem).unwrap(); + assert_eq!(m, Outcome::Count(0)); + } + + /// B11: with nothing bound there are no centers, every side is read in + /// full, and a side keyed by its second column is served. + #[tokio::test(flavor = "multi_thread")] + async fn b11_an_unbound_demand_reads_every_side_in_full() { + let f = family_data(); + let (labels, rules) = resolved(&family::crepe_plan()); + let mem = store(&labels, &f, [1, 1, 1]); + let fx = fixture(&labels, &f).await; + let d = Demand { + goal: labels.id("Answer").unwrap(), + bound: vec![None, None], + }; + for kind in kinds() { + let (m, _) = evaluate(&rules, &d, kind, Options::ALL, &mem).unwrap(); + let (b, _, _, _) = + evaluate_addressed(&rules, &d, kind, Options::ALL, &fx.lance, &fx.lanes) + .await + .unwrap(); + assert_eq!(b, m, "{kind:?}"); + } + // Not vacuous: Parent(x, y) is keyed by its second column here. + let planned = + crepe_plan_probe::pipeline::plan(&rules, &d, ResultKind::Count(BAG), Options::ALL) + .unwrap(); + assert!( + planned + .stars + .iter() + .any(|s| s.sides.iter().any(|side| side.key.key_col == 1)) + ); + } + + /// B12: a lane attested on one table is refused against another table + /// with the same relation id and version. + #[tokio::test(flavor = "multi_thread")] + async fn b12_a_lane_from_another_dataset_is_refused() { + let f = family_data(); + let (labels, rules) = resolved(&family::crepe_plan()); + let one = fixture(&labels, &f).await; + let mut g = family_data(); + g.parent[0] = [17, 2]; + let two = fixture(&labels, &g).await; + let d = answer_demand(&labels, 17); + let kind = ResultKind::Count(BAG); + // Same relation ids and versions on both stores. + for (l1, l2) in one.lanes.iter().zip(&two.lanes) { + assert_eq!((l1.rel, l1.lane.version()), (l2.rel, l2.lane.version())); + } + let got = evaluate_addressed(&rules, &d, kind, Options::ALL, &two.lance, &one.lanes).await; + assert!( + matches!(&got, Err(PipelineRefusal::Addressed(e)) if e.starts_with("OtherDataset")), + "{got:?}" + ); + // Silence twin: each store with its own lanes is served. + assert!( + evaluate_addressed(&rules, &d, kind, Options::ALL, &two.lance, &two.lanes) + .await + .is_ok() + ); + } + /// B9: the arm adds no program and no foreign hop. #[tokio::test(flavor = "multi_thread")] async fn b9_the_addressed_arm_adds_no_program() { From 2fedea41aeac8fec9cd240eabee843272c792a4a Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 20 Oct 2026 20:09:09 +0000 Subject: [PATCH 6/6] crepe-plan-probe: the relalg Or limit is a setting, 2k-1 slots An Or of k atoms takes 2k - 1 scratch slots, not one per atom: each atom and each OR gets a slot of its own, and no slot is reused inside a fold. MAX_SLOTS = 8 is an r2il setting ("sized with headroom"); mask-risc addresses 65,536 slots. A larger arena, or slot reuse inside the fold, would lift the 4-center limit. The comment, the plan and the board entry now say so. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f --- .../2026-10-20-crepe-rule-optimizer-spike.md | 2 +- .claude/plans/crepe-rule-optimizer-spike-v1.md | 2 +- crates/crepe-plan-probe/src/lance_store.rs | 16 ++++++++++------ 3 files changed, 12 insertions(+), 8 deletions(-) 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 966439d99..f1e980392 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 @@ -19,5 +19,5 @@ - Dynamic MeTTa rules need crepe's analysis as a library: crepe is a proc-macro crate (follow-up A, `crepe-core` split). - 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–6 µs per needle at 50,000 rows and beats the scalar test up to 64 needles; at 128 needles two runs disagree because of benchmark noise (the scalar time moved 1.8× between runs), so that crossover is not pinned (re-measured after the #1632 Codex fix). Next step: a column-major relation store. -- Stage 3 over Lance 13 / lancedb 0.40 built (`lance_store`): each side read under a relalg address mask with `with_row_addr_prefilter`; every masked read scans exactly its mask (P1–P4 green). The relalg membership mask holds 4 centers at most (one scratch slot per `eq` atom, `MAX_SLOTS = 8`); above that a scalar test builds it. +- Stage 3 over Lance 13 / lancedb 0.40 built (`lance_store`): each side read under a relalg address mask with `with_row_addr_prefilter`; every masked read scans exactly its mask (P1–P4 green). The relalg membership mask holds 4 centers at most: `k` centers take `2k - 1` scratch slots (no slot reuse inside a fold), and `MAX_SLOTS = 8` is an r2il setting, not a hard limit (mask-risc addresses 65,536). Above that a scalar test builds it. - Arm B built (`addressed`): lanes in key order, attested with `SealedFacetLane::attest_sorted`, read only at the `seek_from` ranges of the demanded keys. At 50,000 rows it scans `3k` rows where Arm A scans at least 150,000. It is 2.5–12× faster than Arm A up to 1% of centers and pays back its 7 ms attest after 2–3 queries. From 10% of centers up a full scan wins, and with all centers Arm B never breaks even. Open: the selectivity crossover between 1% and 10%; multi-fragment lanes (an append currently forces a compaction); a second lane for demands on the second column. diff --git a/.claude/plans/crepe-rule-optimizer-spike-v1.md b/.claude/plans/crepe-rule-optimizer-spike-v1.md index d49c25473..0b2e2db9a 100644 --- a/.claude/plans/crepe-rule-optimizer-spike-v1.md +++ b/.claude/plans/crepe-rule-optimizer-spike-v1.md @@ -118,7 +118,7 @@ DECISION (operator, 2026-10-20): "ja, bau Stufe 3 mit Lance 13 und CI-Workflow" The relations are stored column-major in Lance: one table per relation, columns `c0`, `c1` (`UInt32`), written through lancedb 0.40 and read through lance 13. Preparation becomes H14d's stages 2 and 3 per side: 1. Read the one column the address mask needs (the restriction column, or the key column for the semijoin), with `_rowaddr`. -2. Build the address mask `R` through relalg over that column as a `U32` lane: `eq_u32(col, k)` for a restriction, `Or` of `eq_u32(key, c)` for the semijoin centers (lowered by mask-risc onto `ndarray::simd::eq_u32_to_mask`). Above 4 centers the mask is built by the scalar membership test instead: each `eq` atom of the `Or` takes one scratch slot and a fold may use at most 8 (`lance_graph_r2il::MAX_SLOTS`), so 5 centers fail with `ScratchBufferTooSmall`. This is relalg's ceiling, not the §2c speed crossover, which is ndarray's row filter. +2. Build the address mask `R` through relalg over that column as a `U32` lane: `eq_u32(col, k)` for a restriction, `Or` of `eq_u32(key, c)` for the semijoin centers (lowered by mask-risc onto `ndarray::simd::eq_u32_to_mask`). Above 4 centers the mask is built by the scalar membership test instead. relalg lowers `Or(a1..ak)` left-associatively, and every atom and every `OR` takes a scratch slot of its own (no reuse inside a fold), so `k` centers take `2k - 1` slots. One fold may use `lance_graph_r2il::MAX_SLOTS = 8`, a setting the crate calls "sized with headroom", so 5 centers (9 slots) fail with `ScratchBufferTooSmall`. mask-risc addresses up to 65,536 slots. Either a larger arena or slot reuse inside the fold (an `Or` chain needs only the accumulator and the current atom) would lift the limit; both are r2il/relalg changes. This is not the §2c speed crossover, which is ndarray's row filter. 3. Read the side's columns with `with_row_addr_prefilter(R)`. 4. Hand those rows to `JoinTable::prepare_star`, as the in-memory path does. diff --git a/crates/crepe-plan-probe/src/lance_store.rs b/crates/crepe-plan-probe/src/lance_store.rs index 3317d78bb..b1d9a7c33 100644 --- a/crates/crepe-plan-probe/src/lance_store.rs +++ b/crates/crepe-plan-probe/src/lance_store.rs @@ -30,12 +30,16 @@ use crate::native::{PreparedStar, StarPlan, prepare_from_rows}; /// Up to this many semijoin centers, the address mask is an `Or` of `eq_u32` /// atoms through relalg; above it, the scalar membership test builds it. /// -/// The limit is relalg's, not a speed crossover: each `eq` atom of the `Or` -/// holds its own scratch slot, and one fold may use at most -/// `lance_graph_r2il::MAX_SLOTS` (8) of them. Measured: 4 centers fit, 5 -/// fail with `ScratchBufferTooSmall { need_words: 145, have_words: 129 }` at -/// 1,000 rows. (The 64-needle crossover in plan §2c is ndarray's row filter, -/// a different mechanism.) +/// The limit comes from a setting, not from the algebra and not from a speed +/// crossover. relalg lowers `Or(a1..ak)` left-associatively, and each atom +/// and each `OR` takes a scratch slot of its own (slots are not reused +/// inside a fold), so `k` centers take `2k - 1` slots. One fold may use +/// `lance_graph_r2il::MAX_SLOTS` (8, "sized with headroom") of them. Measured: +/// 4 centers fit (7 slots); 5 need 9 and fail with +/// `ScratchBufferTooSmall { need_words: 145, have_words: 129 }` at 1,000 rows. +/// mask-risc itself addresses up to 65,536 slots. A larger arena, or slot +/// reuse in the fold, lifts this. (The 64-needle crossover in plan §2c is +/// ndarray's row filter, a different mechanism.) pub const SEMIJOIN_ATOMS_MAX: usize = 4; #[derive(Debug)]