crepe-plan-probe: ARM-B-PREPARE, Arm B's read policy above 1% of centers

.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.

.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) |

.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.

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>) -> 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<Rule> = 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<u32> = 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
+        );
+    }
+}