From b03682176ffe69b987b2b0a6baab9a58959d4847 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 20 Oct 2026 22:59:45 +0000 Subject: [PATCH] crepe-plan-probe: ARM-B-PREPARE, Arm B's read policy above 1% of centers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Arm B read the located lane ranges exactly at every selectivity. Above about 1% of centers its many short ranges cost Lance more per row than one sequential read (plan §2e). Each side now chooses its read with `addressed::plan_read`, a pure function of the located ranges and the table size: - exact ranges (touching ranges merge and stay exact); - coalesced ranges across gaps of at most `coalesce_gap` rows; - a full read once the ranges cover `full_permille` of the table. Rows read beyond the demand are dropped by the row's own leading value, so no position is taken from another lane's values. `ReadPolicy::EXACT` keeps the old behaviour and stays the default of `evaluate_addressed`; `evaluate_addressed_with` takes a policy. `ReadPolicy::DEFAULT` (gap 16, full at 100 permille) is pinned from examples/addressed_read.rs: - 50,000 rows: exact wins to about 3%, ties at 4-6%, loses from 10% (35.1 vs 18-19 ms at 50%). - 200,000 rows: exact wins to about 5%, ties at 7-10% (128.5 vs 74-79 ms at 50%). - Coalescing is neutral at gap 16 and loses at gap 64 on uniform demand. Tests: 5 unit tests on `plan_read`, B13-B15 against Lance (every policy answers as EXACT, the switch fires and stays silent per side, coalescing hands Lance fewer ranges). 46 pass. Disable runs, each red: no post-read drop, strict gap, never full, never coalesce, touching ranges counted as coalesced. Plan §2f, board entry 2026-10-20-arm-b-prepare, entries index. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f --- .../board/entries/2026-10-20-arm-b-prepare.md | 22 ++ .claude/board/entries/README.md | 3 +- .../plans/crepe-rule-optimizer-spike-v1.md | 47 +++- .../examples/addressed_read.rs | 180 ++++++++++++++++ crates/crepe-plan-probe/src/addressed.rs | 203 +++++++++++++++++- crates/crepe-plan-probe/src/pipeline.rs | 17 +- crates/crepe-plan-probe/tests/pipeline.rs | 152 ++++++++++++- 7 files changed, 611 insertions(+), 13 deletions(-) create mode 100644 .claude/board/entries/2026-10-20-arm-b-prepare.md create mode 100644 crates/crepe-plan-probe/examples/addressed_read.rs diff --git a/.claude/board/entries/2026-10-20-arm-b-prepare.md b/.claude/board/entries/2026-10-20-arm-b-prepare.md new file mode 100644 index 000000000..837bf01b0 --- /dev/null +++ b/.claude/board/entries/2026-10-20-arm-b-prepare.md @@ -0,0 +1,22 @@ +# 2026-10-20 — ARM-B-PREPARE: Arm B's read policy above 1% of centers + +**Status:** MEASURED (lab probe `crates/crepe-plan-probe`). Closes ledger (d) item ARM-B-PREPARE for uniform demand. Plan: `.claude/plans/crepe-rule-optimizer-spike-v1.md` §2f. + +## What landed + +- `addressed::plan_read` (pure, lane positions only) chooses per side between exact ranges, coalesced ranges and a full read; extra rows are dropped by the row's own leading value (co-addressed, no gather). +- `ReadPolicy::{EXACT, FULL, DEFAULT}`; `evaluate_addressed` keeps EXACT, `evaluate_addressed_with` takes a policy. `Lookup` reports the read and the ranges handed to Lance. +- `examples/addressed_read.rs` (`READ_N`, `READ_KS`, `READ_REPS`). + +## Measured + +- 50,000 rows: exact wins to about 3%, ties 4–6%, loses from 10% (50%: 35.1 vs 18–19 ms). +- 200,000 rows: exact wins to about 5%, ties 7–10% (50%: 128.5 vs 74–79 ms). +- DEFAULT (gap 16, full at 10%) is never measurably worse on either size. Coalescing is neutral at gap 16 and loses at gap 64 on uniform demand. +- 46 tests pass; 5 disable runs red. + +## OPEN + +- The crossover moves with table size; a size-aware switch is not modelled. +- Clustered demand (where coalescing should pay) is not measured. +- `evaluate_addressed` still defaults to EXACT; no caller uses DEFAULT by default. diff --git a/.claude/board/entries/README.md b/.claude/board/entries/README.md index 4baa255b9..8119392a2 100644 --- a/.claude/board/entries/README.md +++ b/.claude/board/entries/README.md @@ -25,7 +25,7 @@ index row, (3) no duplicate entry id. Checks 1 and 2 are deliberately opposite directions; the stranding this convention prevents shows up in exactly one of them, never both. -392 entries, 2026-08-06 .. 2026-10-20. +393 entries, 2026-08-06 .. 2026-10-20. | date | entry id | finding | file | |---|---|---|---| @@ -47,6 +47,7 @@ exactly one of them, never both. | 2026-10-20 | `D-RMAP-2-AMX-RERUN` | §5a-AMX re-measured on a second AMX host | [2026-10-20-d-rmap-2-amx-remeasured.md](2026-10-20-d-rmap-2-amx-remeasured.md) | | 2026-10-20 | `D-SRAM-2` | | [2026-10-20-crepe-rule-optimizer-spike.md](2026-10-20-crepe-rule-optimizer-spike.md) | | 2026-10-20 | `ce64-isa-not-storage` | | [2026-10-20-ce64-isa-not-storage.md](2026-10-20-ce64-isa-not-storage.md) | +| 2026-10-20 | `arm-b-prepare` | | [2026-10-20-arm-b-prepare.md](2026-10-20-arm-b-prepare.md) | | 2026-10-19 | `post-merge-recovery-1601-1617` | | [2026-10-19-post-merge-recovery-1601-1617.md](2026-10-19-post-merge-recovery-1601-1617.md) | | 2026-10-19 | `native-join-relationship-table` | | [2026-10-19-native-join-relationship-table.md](2026-10-19-native-join-relationship-table.md) | | 2026-10-19 | `mork-pathmap-schnellverschluss` | | [2026-10-19-mork-pathmap-schnellverschluss.md](2026-10-19-mork-pathmap-schnellverschluss.md) | diff --git a/.claude/plans/crepe-rule-optimizer-spike-v1.md b/.claude/plans/crepe-rule-optimizer-spike-v1.md index 3bdd23da4..7fef516f4 100644 --- a/.claude/plans/crepe-rule-optimizer-spike-v1.md +++ b/.claude/plans/crepe-rule-optimizer-spike-v1.md @@ -31,6 +31,7 @@ Pipeline, one stage per module of the 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) +- [x] ARM-B-PREPARE: per-side read policy (exact ranges, coalesced, full) with a measured switch (§2f, tests B13–B15 and 5 unit tests) ## 1. Integration point @@ -168,6 +169,50 @@ What this says: P5 is measured at 50,000 rows only; 1,000 and 10,000 were not run. +## 2f. ARM-B-PREPARE: when Arm B stops reading ranges (measured 2026-10-20) + +Ledger (d) item ARM-B-PREPARE: Arm B removes the O(n) preparation scan only below about 1% of centers; above that, its many short ranges cost Lance more per row than one sequential read (§2e). + +Each side now chooses its read with `addressed::plan_read`, a pure function of the located lane ranges and the table size (it reads no row): +- `Read::Ranges`: the located ranges (ranges that touch are merged and still count as exact); +- `Read::Coalesced`: ranges at most `coalesce_gap` rows apart are merged, and the extra rows are dropped after the read by the row's own leading value (co-addressed: no position is taken from another lane's values); +- `Read::Full`: once the merged ranges cover `full_permille` of the table, the whole table is read without a row filter and the same drop applies. + +`ReadPolicy::EXACT` keeps the old behaviour and is still what `evaluate_addressed` uses; `evaluate_addressed_with` takes a policy. `ReadPolicy::DEFAULT` = gap 16, full at 100 permille. + +Measured with `examples/addressed_read.rs` (release, rustc 1.99.0; same fixture as §2e: uniform demand, one row per center; every strategy checked against the answer count before timing). Times in ms. + +50,000 rows (median of 9; 1,000–3,000 re-run with 31): + +| k | exact | gap 16 | full | +|---|---|---|---| +| 0 | 0.40 | 0.36 | 2.98 | +| 500 (1%) | 2.72 | 3.20 | 4.09 | +| 1,000 (2%) | 4.55 | 4.06 | 6.50 | +| 2,000 (4%) | 5.66 | 5.10 | 5.48 | +| 3,000 (6%) | 6.53 | 6.58 | 5.79 | +| 5,000 (10%) | 8.87 | 7.27 | 7.46 | +| 25,000 (50%) | 35.10 | 17.97 | 19.06 | + +200,000 rows (`READ_N=200000`, median of 15): + +| k | exact | gap 16 | gap 64 | full | +|---|---|---|---|---| +| 1,000 (0.5%) | 5.11 | 5.66 | 5.41 | 12.09 | +| 2,000 (1%) | 6.94 | 7.62 | 11.70 | 18.08 | +| 10,000 (5%) | 19.82 | 18.11 | 20.13 | 22.15 | +| 14,285 (7%) | 22.22 | 21.40 | 20.72 | 23.21 | +| 100,000 (50%) | 128.48 | 74.29 | 75.62 | 79.37 | + +What this says: +- The crossover is not a fixed fraction of the table: exact ranges win up to about 3% at 50,000 rows and about 5% at 200,000, and tie with the full read at 4–6% and 7–10% respectively. Between the two, the choice is within run-to-run noise (an identical policy measured 3.9 and 6.1 ms at k = 1,000 in one run). +- Above 10% the full read wins: 1.9× at 50% of 50,000 rows, 1.6–1.7× at 50% of 200,000. DEFAULT switches at 10%, which is never measurably worse on either size. +- A full read has a floor (3 ms at 50,000 rows, 6.7 ms at 200,000), so below the switch it would lose up to 3×. +- Coalescing measured neutral at gap 16 and harmful at gap 64 (11.7 vs 6.9 ms at 1% of 200,000 rows) on uniform demand. Clustered demand, where it should help, is not measured. +- Two table sizes only; the crossover's dependence on size is observed, not modelled. + +Tests: `addressed::tests` (5: gap boundary, touching ranges, full-read threshold at the exact row, coalesced rows counted toward it, EXACT/FULL extremes); `arm_b::b13` (every policy answers as EXACT, every kind, hit/sponsor/miss), `b14` (the switch fires on dense demand and stays silent on sparse, per side; a full read scans more rows), `b15` (coalescing hands Lance fewer ranges, answer equal to the memory path). Disable runs, each red: no post-read drop (b13, b14), strict gap (3 unit tests), never full (3 unit tests and b14), never coalesce (3 unit tests and b15), touching ranges counted as coalesced (1 unit test). B15 does not pin the drop: undemanded rows find no join partner there. + ## 3. Soufflé harvest (souffle-lang/souffle `f53dab8`, UPL-1.0; algorithms only, no source copied) | Soufflé | transformation | precondition | in crepe | in lance-graph | taken | @@ -211,7 +256,7 @@ Known limits: rule dedup does not search body permutations; the star must have o - ~~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. +- ~~Selectivity crossover~~ — built, §2f. Measured at 50,000 and 200,000 rows; `ReadPolicy::DEFAULT` switches to the full read at 10% coverage. Clustered demand and a size-aware switch remain open. - 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. diff --git a/crates/crepe-plan-probe/examples/addressed_read.rs b/crates/crepe-plan-probe/examples/addressed_read.rs new file mode 100644 index 000000000..1bbf0e6d8 --- /dev/null +++ b/crates/crepe-plan-probe/examples/addressed_read.rs @@ -0,0 +1,180 @@ +//! Arm B's row read by strategy and demand selectivity, to place the switch +//! from exact ranges to a coalesced or full read. Run in release from this +//! crate's directory: +//! +//! ```text +//! cargo run --release --example addressed_read +//! ``` +//! +//! The fixture is the one of `examples/addressed.rs`: `Answer(17, z)` over +//! `family`, `READ_N` (default 50,000) background edges with distinct `y`, and `k` +//! children of 17 spread over them, so `k` centers and `k` answers. Every +//! strategy is checked against that count before it is timed. +//! +//! Strategies, each a median of `READ_REPS` (default 9) warm runs, lanes +//! attested once. `READ_KS=1000,2000` restricts the sweep to those `k`: +//! - `exact`: the located ranges only (`ReadPolicy::EXACT`); +//! - `gap G`: ranges merged across gaps of at most `G` rows, then the extra +//! rows dropped by their leading value (no full read); +//! - `full`: the whole table, then the same drop (`ReadPolicy::FULL`); +//! - `auto`: the pinned `ReadPolicy::DEFAULT`. + +use std::time::{Duration, Instant}; + +use crepe_plan_probe::addressed::{AddressedRelation, Read, ReadPolicy, attest}; +use crepe_plan_probe::lance_store::LanceStore; +use crepe_plan_probe::logic::{Labels, Rule, resolve}; +use crepe_plan_probe::optimize::{Demand, Options, ResultKind}; +use crepe_plan_probe::pipeline::{Outcome, evaluate_addressed_with}; +use crepe_plan_probe::programs::family; +use mork_pathmap_probe::Semantics; + +/// Background rows (`READ_N`, default 50,000). +fn n_rows() -> u32 { + std::env::var("READ_N") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(50_000) +} +const KIND: ResultKind = ResultKind::Count(Semantics::Bag); + +/// Runs per strategy (`READ_REPS`, default 9). +fn reps() -> usize { + std::env::var("READ_REPS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(9) +} + +fn median(mut v: Vec) -> Duration { + v.sort(); + v[v.len() / 2] +} + +/// Median time, the strategy of the side with the most centers, and the +/// ranges it handed to Lance. +async fn run( + rules: &[Rule], + d: &Demand, + st: &LanceStore, + lanes: &[AddressedRelation], + policy: ReadPolicy, + k: u64, +) -> (Duration, Read, usize) { + let mut t = Vec::with_capacity(reps()); + let mut how = (Read::Ranges, 0); + for _ in 0..reps() { + let s = Instant::now(); + let (out, _, _, lookups) = + evaluate_addressed_with(rules, d, KIND, Options::ALL, st, lanes, policy) + .await + .unwrap(); + t.push(s.elapsed()); + assert_eq!(out, Outcome::Count(k), "policy {policy:?} at k={k}"); + let widest = lookups + .iter() + .flatten() + .max_by_key(|l| l.keys) + .expect("a side"); + how = (widest.read, widest.ranges_read); + } + (median(t), how.0, how.1) +} + +fn ms(d: Duration) -> f64 { + d.as_secs_f64() * 1e3 +} + +#[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 n = n_rows(); + let gaps: [u32; 3] = [4, 16, 64]; + // `READ_KS=1000,2000` measures only those k. + let ks: Vec = match std::env::var("READ_KS") { + Ok(v) => v + .split(',') + .map(|k| k.trim().parse().expect("READ_KS")) + .collect(), + Err(_) => vec![ + 0, + 1, + 8, + n / 200, + n / 100, + n / 50, + n / 33, + n / 20, + n / 14, + n / 10, + n / 2, + n, + ], + }; + + print!("{:>7} | {:>8}", "k", "exact"); + for g in gaps { + print!(" {:>8}", format!("gap {g}")); + } + println!( + " {:>8} | {:>8} {:>10} {:>7}", + "full", "auto", "auto read", "ranges" + ); + for k in ks { + let mut parent: Vec<[u32; 2]> = (0..n).map(|i| [1_000_000 + i, 3_000_000 + i]).collect(); + let step = n.checked_div(k).unwrap_or(1); + parent.extend((0..k).map(|j| [17, 1_000_000 + j * step])); + let allowed: Vec<[u32; 2]> = (0..n).map(|i| [1_000_000 + i, 1_000_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 lanes = Vec::new(); + for &r in &rels { + lanes.push(attest(&st, r).await.unwrap()); + } + let k64 = u64::from(k); + + let (exact, ..) = run(&rules, &d, &st, &lanes, ReadPolicy::EXACT, k64).await; + print!("{:>7} | {:>8.3}", k, ms(exact)); + for g in gaps { + let p = ReadPolicy { + coalesce_gap: g, + full_permille: u32::MAX, + }; + let (t, ..) = run(&rules, &d, &st, &lanes, p, k64).await; + print!(" {:>8.3}", ms(t)); + } + let (full, ..) = run(&rules, &d, &st, &lanes, ReadPolicy::FULL, k64).await; + let (auto, how, nr) = run(&rules, &d, &st, &lanes, ReadPolicy::DEFAULT, k64).await; + println!( + " {:>8.3} | {:>8.3} {:>10} {:>7}", + ms(full), + ms(auto), + format!("{how:?}"), + nr + ); + } +} diff --git a/crates/crepe-plan-probe/src/addressed.rs b/crates/crepe-plan-probe/src/addressed.rs index b7b379213..a072b0db9 100644 --- a/crates/crepe-plan-probe/src/addressed.rs +++ b/crates/crepe-plan-probe/src/addressed.rs @@ -31,7 +31,7 @@ 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::lance_store::{Io, LanceStore, LanceStoreError, SideIo, read_rows}; use crate::logic::RelId; use crate::native::{PreparedStar, StarPlan, prepare_from_rows}; @@ -148,6 +148,87 @@ pub struct Lookup { pub keys: usize, /// Rows inside the located ranges. pub rows_in_ranges: usize, + /// How the side's rows were read. + pub read: Read, + /// Ranges handed to Lance (0 for a full read). + pub ranges_read: usize, +} + +/// How a side's rows are read once its lane ranges are located. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub enum Read { + /// Exactly the located ranges. + #[default] + Ranges, + /// The located ranges with short gaps between them read too. The extra + /// rows are dropped after the read by their own leading value. + Coalesced, + /// The whole table without a row filter, then the same drop. + Full, +} + +/// When a side stops reading exact ranges. Many short ranges cost Lance more +/// per row than one sequential read (§2e: from about 10% of centers every +/// masked read loses to the full one). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ReadPolicy { + /// Two ranges are merged when at most this many rows lie between them. + pub coalesce_gap: u32, + /// The table is read in full when the (merged) ranges cover at least this + /// many rows per 1,000 of the table. Above 1,000, never. + pub full_permille: u32, +} + +impl ReadPolicy { + /// Exact ranges only: Arm B as first measured. + pub const EXACT: Self = Self { + coalesce_gap: 0, + full_permille: u32::MAX, + }; + /// The measured switch (`examples/addressed_read.rs`, release, uniform + /// demand, policy pins rather than derived values): + /// - Full read at 10% coverage. The crossover moved with table size: at + /// 50,000 rows exact ranges win up to about 3% and lose from about 10%; + /// at 200,000 rows they win up to about 5% and tie at 7–10%. Between + /// the two, either choice is within run-to-run noise; above 10% the + /// full read is up to 1.9× faster (50% of centers). + /// - A coalescing gap of 16 rows measured neutral on uniform demand. A + /// gap of 64 lost (11.7 vs 6.9 ms at 1% of 200,000 rows). Clustered + /// demand, where coalescing should help, is not measured. + pub const DEFAULT: Self = Self { + coalesce_gap: 16, + full_permille: 100, + }; + /// Always the whole table. + pub const FULL: Self = Self { + coalesce_gap: 0, + full_permille: 0, + }; +} + +/// Decides how to read `ranges` (sorted, disjoint, inside `0..n_rows`) and +/// returns the ranges to hand to Lance (empty for [`Read::Full`]). The rule +/// reads only the lane positions, never a row: it is part of preparation. +pub fn plan_read(ranges: &[(u32, u32)], n_rows: u32, p: ReadPolicy) -> (Read, Vec<(u32, u32)>) { + let mut merged: Vec<(u32, u32)> = Vec::with_capacity(ranges.len()); + for &(lo, hi) in ranges { + match merged.last_mut() { + Some(last) if lo - last.1 <= p.coalesce_gap => last.1 = hi, + _ => merged.push((lo, hi)), + } + } + let span = |r: &[(u32, u32)]| -> u64 { r.iter().map(|&(lo, hi)| u64::from(hi - lo)).sum() }; + let covered = span(&merged); + if n_rows > 0 && covered * 1000 >= u64::from(n_rows) * u64::from(p.full_permille) { + return (Read::Full, Vec::new()); + } + // Merging ranges that touch reads no extra row: still exact. + let read = if covered > span(ranges) { + Read::Coalesced + } else { + Read::Ranges + }; + (read, merged) } /// The rows of every leading value in `wanted` (sorted, deduplicated) as lane @@ -226,14 +307,48 @@ async fn current_lane<'a>( Ok((a, ds)) } +/// Reads one side's rows for the located `ranges` of the leading values in +/// `wanted` (sorted, deduplicated), as `policy` decides. A coalesced or full +/// read returns rows of other leading values too; they are dropped by the +/// row's own leading value, so the result equals the exact read. +async fn read_side( + ds: &Dataset, + a: &AddressedRelation, + ranges: &[(u32, u32)], + wanted: &[u32], + policy: ReadPolicy, +) -> Result<(Read, usize, Vec<[u32; 2]>, Io), AddressedRefusal> { + let (how, r) = plan_read(ranges, a.lane.n_rows(), policy); + let mask = match how { + Read::Full => None, + Read::Ranges | Read::Coalesced => Some(range_mask(&r)), + }; + let (mut got, rows_io) = read_rows(ds, mask).await?; + if how != Read::Ranges { + got.retain(|row| wanted.binary_search(&row[0]).is_ok()); + } + Ok((how, r.len(), got, rows_io)) +} + +/// [`prepare_addressed_with`] under [`ReadPolicy::EXACT`]. +pub async fn prepare_addressed( + plan: &StarPlan, + store: &LanceStore, + lanes: &[AddressedRelation], +) -> Result<(PreparedStar, Vec, Vec), AddressedRefusal> { + prepare_addressed_with(plan, store, lanes, ReadPolicy::EXACT).await +} + /// 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( +/// demand to address by). `policy` decides per side whether the located +/// ranges are read exactly, coalesced, or replaced by a full read. +pub async fn prepare_addressed_with( plan: &StarPlan, store: &LanceStore, lanes: &[AddressedRelation], + policy: ReadPolicy, ) -> Result<(PreparedStar, Vec, Vec), AddressedRefusal> { let n_sides = plan.sides.len(); let mut rows: Vec, u64)>> = vec![None; n_sides]; @@ -257,10 +372,15 @@ pub async fn prepare_addressed( Ok(k) => ranges(a, &[k])?, Err(_) => (Vec::new(), 0), }; - let (got, rows_io) = read_rows(&ds, Some(range_mask(&r))).await?; + let wanted: Vec = u32::try_from(restricted(i).expect("restricted")) + .into_iter() + .collect(); + let (how, read_ranges, got, rows_io) = read_side(&ds, a, &r, &wanted, policy).await?; lookups[i] = Lookup { keys: 1, rows_in_ranges: n, + read: how, + ranges_read: read_ranges, }; io[i] = SideIo { mask_column: None, @@ -302,7 +422,7 @@ pub async fn prepare_addressed( for i in (0..n_sides).filter(|&i| restricted(i).is_none()) { let key = plan.sides[i].key; let (a, ds) = current_lane(plan, i, store, lanes).await?; - let (mask, n, sought) = match ¢ers { + let (n, sought, how, read_ranges, got, rows_io) = 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 => { @@ -310,14 +430,19 @@ pub async fn prepare_addressed( } Some(c) => { let (r, n) = ranges(a, c)?; - (Some(range_mask(&r)), n, c.len()) + let (how, rr, got, rows_io) = read_side(&ds, a, &r, c, policy).await?; + (n, c.len(), how, rr, got, rows_io) + } + None => { + let (got, rows_io) = read_rows(&ds, None).await?; + (a.lane.n_rows() as usize, 0, Read::Full, 0, got, rows_io) } - 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, + read: how, + ranges_read: read_ranges, }; io[i] = SideIo { mask_column: None, @@ -333,3 +458,65 @@ pub async fn prepare_addressed( .collect(); Ok((prepare_from_rows(plan, owned), io, lookups)) } + +#[cfg(test)] +mod tests { + use super::*; + + const NO_FULL: u32 = u32::MAX; + + fn policy(coalesce_gap: u32, full_permille: u32) -> ReadPolicy { + ReadPolicy { + coalesce_gap, + full_permille, + } + } + + #[test] + fn a_gap_of_exactly_the_limit_merges_and_one_more_does_not() { + let r = [(0, 2), (5, 6)]; // gap of 3 rows + assert_eq!( + plan_read(&r, 1_000, policy(3, NO_FULL)), + (Read::Coalesced, vec![(0, 6)]) + ); + assert_eq!( + plan_read(&r, 1_000, policy(2, NO_FULL)), + (Read::Ranges, r.to_vec()) + ); + } + + #[test] + fn touching_ranges_merge_but_stay_exact() { + let r = [(0, 2), (2, 4), (9, 10)]; + assert_eq!( + plan_read(&r, 1_000, ReadPolicy::EXACT), + (Read::Ranges, vec![(0, 4), (9, 10)]) + ); + } + + #[test] + fn the_full_read_fires_at_the_threshold_and_not_one_row_below() { + // 1,000 rows, 100 permille: 100 covered rows switch, 99 do not. + assert_eq!(plan_read(&[(0, 100)], 1_000, policy(0, 100)).0, Read::Full); + assert_eq!(plan_read(&[(0, 99)], 1_000, policy(0, 100)).0, Read::Ranges); + } + + #[test] + fn the_full_read_counts_the_rows_coalescing_adds() { + // 50 exact rows in 50 one-row ranges two apart: 99 rows once merged. + let r: Vec<(u32, u32)> = (0..50).map(|i| (2 * i, 2 * i + 1)).collect(); + assert_eq!(plan_read(&r, 1_000, policy(0, 99)).0, Read::Ranges); + assert_eq!(plan_read(&r, 1_000, policy(1, 99)).0, Read::Full); + } + + #[test] + fn exact_never_reads_in_full_and_full_always_does() { + assert_eq!( + plan_read(&[(0, 1_000)], 1_000, ReadPolicy::EXACT).0, + Read::Ranges + ); + assert_eq!(plan_read(&[], 1_000, ReadPolicy::FULL).0, Read::Full); + // An empty table is never read "in full": there is nothing to read. + assert_eq!(plan_read(&[], 0, ReadPolicy::FULL).0, Read::Ranges); + } +} diff --git a/crates/crepe-plan-probe/src/pipeline.rs b/crates/crepe-plan-probe/src/pipeline.rs index 85405bb6a..eec8bef6e 100644 --- a/crates/crepe-plan-probe/src/pipeline.rs +++ b/crates/crepe-plan-probe/src/pipeline.rs @@ -7,7 +7,7 @@ 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::addressed::{AddressedRelation, Lookup, ReadPolicy, prepare_addressed_with}; use crate::lance_store::{LanceStore, LanceStoreError, SideIo, prepare_lance}; use crate::logic::{RelId, Rule}; use crate::mork_api::{MorkRefusal, QueryMulti, admit, query_multi}; @@ -279,6 +279,19 @@ pub async fn evaluate_addressed( opts: Options, store: &LanceStore, lanes: &[AddressedRelation], +) -> Result<(Outcome, Report, Vec>, Vec>), PipelineRefusal> { + evaluate_addressed_with(rules, demand, kind, opts, store, lanes, ReadPolicy::EXACT).await +} + +/// [`evaluate_addressed`] with an explicit [`ReadPolicy`] for every side. +pub async fn evaluate_addressed_with( + rules: &[Rule], + demand: &Demand, + kind: ResultKind, + opts: Options, + store: &LanceStore, + lanes: &[AddressedRelation], + policy: ReadPolicy, ) -> Result<(Outcome, Report, Vec>, Vec>), PipelineRefusal> { let planned = plan(rules, demand, kind, opts)?; let lance = |e: LanceStoreError| PipelineRefusal::Lance(format!("{e:?}")); @@ -287,7 +300,7 @@ pub async fn evaluate_addressed( 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) + let (p, side_io, l) = prepare_addressed_with(s, store, lanes, policy) .await .map_err(|e| PipelineRefusal::Addressed(format!("{e:?}")))?; prepared.push(p); diff --git a/crates/crepe-plan-probe/tests/pipeline.rs b/crates/crepe-plan-probe/tests/pipeline.rs index 11687b182..04ae9f5cf 100644 --- a/crates/crepe-plan-probe/tests/pipeline.rs +++ b/crates/crepe-plan-probe/tests/pipeline.rs @@ -1022,8 +1022,11 @@ mod stage3 { mod arm_b { use super::*; use crepe_plan_probe::addressed::{AddressedRefusal, AddressedRelation, attest, key}; + use crepe_plan_probe::addressed::{Lookup, Read, ReadPolicy}; use crepe_plan_probe::lance_store::{LanceStore, SideIo}; - use crepe_plan_probe::pipeline::{PipelineRefusal, evaluate_addressed, evaluate_lance}; + use crepe_plan_probe::pipeline::{ + PipelineRefusal, evaluate_addressed, evaluate_addressed_with, evaluate_lance, + }; use lance_graph_contract::ordered_lane::WitnessError; struct Fixture { @@ -1398,6 +1401,153 @@ mod arm_b { /// B12: a lane attested on one table is refused against another table /// with the same relation id and version. + /// B13: every read policy answers what the exact read answers, for every + /// result kind, on a hit, a sponsor branch and a miss. The bag kinds see + /// the duplicate `Allowed(2, 2)` row, so a coalesced or full read that + /// dropped or doubled a row would show. + #[tokio::test(flavor = "multi_thread")] + async fn b13_every_read_policy_answers_like_the_exact_read() { + let f = family_data(); + let (labels, rules) = resolved(&family::crepe_plan()); + let fx = fixture(&labels, &f).await; + let policies = [ + ReadPolicy::FULL, + ReadPolicy::DEFAULT, + ReadPolicy { + coalesce_gap: 1, + full_permille: u32::MAX, + }, + ReadPolicy { + coalesce_gap: u32::MAX, + full_permille: u32::MAX, + }, + ]; + for x in [17, 5, 999] { + let d = answer_demand(&labels, x); + for kind in kinds() { + let (exact, _, _, _) = evaluate_addressed_with( + &rules, + &d, + kind, + Options::ALL, + &fx.lance, + &fx.lanes, + ReadPolicy::EXACT, + ) + .await + .unwrap(); + for p in policies { + let (got, _, _, _) = evaluate_addressed_with( + &rules, + &d, + kind, + Options::ALL, + &fx.lance, + &fx.lanes, + p, + ) + .await + .unwrap(); + assert_eq!(got, exact, "x={x} {kind:?} {p:?}"); + } + } + } + } + + /// B14: the switch can fire and can stay silent on the same demand. The + /// centers of `x = 17` cover a few rows of a 400-row `Parent`: below a + /// 100-permille threshold the side reads ranges; at 1 permille it reads + /// the table in full, and Lance then scans every row of it. + #[tokio::test(flavor = "multi_thread")] + async fn b14_the_full_read_fires_on_dense_demand_and_not_on_sparse() { + 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 run = |p: ReadPolicy| { + evaluate_addressed_with(&rules, &d, kind, Options::ALL, &fx.lance, &fx.lanes, p) + }; + let (sparse, _, sparse_io, sparse_l) = run(ReadPolicy { + coalesce_gap: 0, + full_permille: 100, + }) + .await + .unwrap(); + let (dense, _, dense_io, dense_l) = run(ReadPolicy { + coalesce_gap: 0, + full_permille: 1, + }) + .await + .unwrap(); + assert_eq!(dense, sparse); + // Sides that sought keys. `Allowed` has 4 rows, so its 3 centers + // cover 750 permille and it reads in full under both policies; + // `Parent` has 400 rows and is the side the threshold separates. + let sought = |l: &[Vec]| -> Vec { + l.iter().flatten().filter(|l| l.keys > 0).copied().collect() + }; + let (s, d) = (sought(&sparse_l), sought(&dense_l)); + assert!(!s.is_empty(), "the fixture must have centers to seek"); + assert!( + s.iter().any(|l| l.read == Read::Ranges), + "sparse demand reads ranges on the large side: {s:?}" + ); + assert!( + d.iter().all(|l| l.read == Read::Full && l.ranges_read == 0), + "dense demand reads every sought side in full: {d:?}" + ); + let scanned = |io: &[Vec]| -> usize { + io.iter().flatten().map(|s| s.rows.rows_scanned).sum() + }; + assert!( + scanned(&dense_io) > scanned(&sparse_io), + "a full read scans more rows than the ranges: {} vs {}", + scanned(&dense_io), + scanned(&sparse_io) + ); + } + + /// B15: a coalesced read hands Lance fewer ranges than the exact read and + /// gives the memory path's answer. Without `Parent(17, 2)` the centers are + /// `{1, 3}`, and key 2 sits between them, undemanded, on both `Parent` and + /// `Allowed` (where it is the duplicate `[2, 2]` pair), so the merged read + /// covers it. This fixture does not show the post-read drop: rows of key 2 + /// find no partner in the join, so the answer is the same with or without + /// it (disable run). B13 and B14 are what pin the drop. + #[tokio::test(flavor = "multi_thread")] + async fn b15_coalescing_hands_lance_fewer_ranges_for_the_same_answer() { + let mut f = family_data(); + f.parent.retain(|r| *r != [17, 2]); + f.allowed = vec![[1, 1], [2, 2], [2, 2], [3, 3], [2000, 2000]]; + 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); + for kind in kinds() { + let (m, _) = evaluate(&rules, &d, kind, Options::ALL, &mem).unwrap(); + let run = |p: ReadPolicy| { + evaluate_addressed_with(&rules, &d, kind, Options::ALL, &fx.lance, &fx.lanes, p) + }; + let (exact, _, _, el) = run(ReadPolicy::EXACT).await.unwrap(); + let (merged, _, _, ml) = run(ReadPolicy { + coalesce_gap: u32::MAX, + full_permille: u32::MAX, + }) + .await + .unwrap(); + assert_eq!(exact, m, "{kind:?}"); + assert_eq!(merged, m, "{kind:?}"); + let exact_ranges: usize = el.iter().flatten().map(|l| l.ranges_read).sum(); + let merged_ranges: usize = ml.iter().flatten().map(|l| l.ranges_read).sum(); + assert!( + merged_ranges < exact_ranges, + "{merged_ranges} vs {exact_ranges}" + ); + assert!(ml.iter().flatten().any(|l| l.read == Read::Coalesced)); + } + } + #[tokio::test(flavor = "multi_thread")] async fn b12_a_lane_from_another_dataset_is_refused() { let f = family_data();