From 7070b9ade189bb53d4d1e966705fc0171d6b1ccb Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Wed, 21 Oct 2026 08:06:54 +0000
Subject: [PATCH 1/6] U4: co-addressed join operator over one JoinDomain (WIP)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f
---
 crates/crepe-plan-probe/src/coaddressed.rs | 711 +++++++++++++++++++++
 crates/crepe-plan-probe/src/lib.rs         |   1 +
 crates/mork-pathmap-probe/src/join.rs      |  11 +
 3 files changed, 723 insertions(+)
 create mode 100644 crates/crepe-plan-probe/src/coaddressed.rs

diff --git a/crates/crepe-plan-probe/src/coaddressed.rs b/crates/crepe-plan-probe/src/coaddressed.rs
new file mode 100644
index 000000000..a96513394
--- /dev/null
+++ b/crates/crepe-plan-probe/src/coaddressed.rs
@@ -0,0 +1,711 @@
+//! The co-addressed join, as a named operator (D-RU-4 / U4).
+//!
+//! STATUS: PROBE, like [`crate::native`]: the table underneath
+//! (`mork_pathmap_probe::join`) is measured but not admitted.
+//!
+//! # What it is
+//!
+//! A conjunction of per-table [`Rel`]s over one [`JoinDomain`]:
+//!
+//! ```text
+//! CoJoin(table, [rel_0, …, rel_k])
+//!   rows_i  = rel_i over side i's own rows            (relalg fold, per side)
+//!   count_i = |rows_i| per slot                       (GroupReduce Count, keyed
+//!                                                      by side i's slot lane)
+//!   joint   = ∧_i (count_i > 0)                       (relalg And of planes)
+//! ```
+//!
+//! [`crate::native`] applies a side's restriction cold: the rows are filtered
+//! before the side is laid out, so every new constant needs a new prepared
+//! table. Here the restriction is a `Rel` evaluated at query time over one
+//! table prepared without it, and it can be any `Rel` relalg lowers (equality,
+//! inequality, `Or`, `Not`), not only one constant.
+//!
+//! # Why it is not a gather
+//!
+//! Each side's rows are laid out by slot, so row `j` of side `i` sits in the
+//! run `offsets[s]..offsets[s + 1]` of its slot `s`. The slot lane is built
+//! from those offsets, the side's own layout, and is non-decreasing. The
+//! projection `rows_i → count_i` adds each selected row to the run it already
+//! sits in. The joint is an `And` of planes over the one domain. Tuple
+//! enumeration reads each side at the slot the joint names and tests the row
+//! bits in that slot's run. No operand is read at a position taken from
+//! another operand's values. [`CoEval::programs`] keeps every mask-risc
+//! program the operator ran, so a caller can check that with
+//! [`crate::native::has_foreign_hop`].
+//!
+//! # What is counted
+//!
+//! [`CoCost`]: programs and ops run, side rows read (only for sides whose
+//! `Rel` is not `All`), and domain slots scanned. A side with `Rel::All`
+//! costs nothing beyond the star path: its presence plane and offsets are
+//! read as prepared.
+
+use lance_graph_mask_risc::{
+    words_for, Foreign, GroupFold, GroupKey, LaneRef, Operand, Out, Planes, Program, Scratch,
+    Terminal, execute_into,
+};
+use lance_graph_mask_risc::ExecError;
+use lance_graph_relalg::{Measure, Query, Rel, RelError, Run, normalize, run_with_program};
+use mork_pathmap_probe::RelationView;
+use mork_pathmap_probe::join::{JoinDomain, JoinRefusal, JoinTable, Side, for_each_set};
+
+use crate::native::{HeadTerm, SideKey, StarPlan, Store, fingerprint};
+
+/// The lane a side's `Rel` names: the side's other column.
+pub const VALUE_LANE: u16 = 0;
+
+/// One side: the laid-out rows and their slot lane.
+#[derive(Clone, Debug)]
+pub struct CoSide {
+    pub side: Side,
+    /// `slot[j]`: the slot row `j` sits in. Built from `side.offsets()`, so
+    /// non-decreasing.
+    slot: Vec<u32>,
+}
+
+impl CoSide {
+    pub fn new(side: Side) -> CoSide {
+        let off = side.offsets();
+        let mut slot = Vec::with_capacity(side.values().len());
+        for s in 0..off.len() - 1 {
+            for _ in off[s]..off[s + 1] {
+                slot.push(s as u32);
+            }
+        }
+        CoSide { side, slot }
+    }
+
+    pub fn n_rows(&self) -> usize {
+        self.side.values().len()
+    }
+
+    /// Rows of this side at slot `s`, as a row range.
+    fn run(&self, s: usize) -> std::ops::Range<usize> {
+        let off = self.side.offsets();
+        off[s] as usize..off[s + 1] as usize
+    }
+}
+
+/// One domain and the sides laid out over it, without restrictions.
+pub struct CoTable {
+    pub domain: JoinDomain,
+    pub sides: Vec<CoSide>,
+    /// The sides' keys, with `restrict: None`: a restriction is a query-time
+    /// `Rel` here, not part of the layout.
+    pub keys: Vec<SideKey>,
+    pub domain_fp: u64,
+}
+
+/// `key` without its restriction.
+pub fn layout_key(key: SideKey) -> SideKey {
+    SideKey {
+        restrict: None,
+        ..key
+    }
+}
+
+impl CoTable {
+    /// Lays out every side of `plan` from all its rows, ignoring restrictions.
+    pub fn prepare(plan: &StarPlan, store: &Store<'_>) -> CoTable {
+        let inputs: Vec<(RelationView<'_>, usize)> = plan
+            .sides
+            .iter()
+            .map(|s| {
+                let (rows, generation) = store.view(s.key.rel);
+                (RelationView { rows, generation }, s.key.key_col as usize)
+            })
+            .collect();
+        let table = JoinTable::prepare_star(&inputs);
+        CoTable {
+            domain_fp: fingerprint(&table),
+            keys: plan.sides.iter().map(|s| layout_key(s.key)).collect(),
+            sides: table.sides.into_iter().map(CoSide::new).collect(),
+            domain: table.domain,
+        }
+    }
+
+    pub fn rows(&self) -> usize {
+        self.sides.iter().map(CoSide::n_rows).sum()
+    }
+}
+
+/// The per-side `Rel`s that express `plan`'s restrictions: `value == k`, or
+/// `Empty` for a constant no `u32` value can equal; `All` when unrestricted.
+pub fn plan_rels(plan: &StarPlan) -> Vec<Rel> {
+    plan.sides
+        .iter()
+        .map(|s| match s.key.restrict {
+            _ if s.key.unary => Rel::All,
+            None => Rel::All,
+            Some(k) => match u32::try_from(k) {
+                Ok(k) => Rel::eq_u32(VALUE_LANE, k),
+                Err(_) => Rel::Empty,
+            },
+        })
+        .collect()
+}
+
+/// What the operator cost, counted from what it ran.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct CoCost {
+    /// mask-risc programs executed.
+    pub programs: usize,
+    /// mask-risc ops executed.
+    pub passes: usize,
+    /// Sides whose `Rel` was not answered from the description alone.
+    pub sides_filtered: usize,
+    /// Rows of those sides read (each read twice: the value lane by the fold,
+    /// the slot lane by the projection).
+    pub side_rows_read: usize,
+    /// Domain slots scanned to turn counts into planes.
+    pub slots_scanned: usize,
+}
+
+/// Why the operator did not answer.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum CoRefusal {
+    /// The table holds other sides than the plan reads.
+    NoCoordinateWitness,
+    /// One `Rel` per side is needed.
+    Rels { sides: usize, rels: usize },
+    /// One generation per side is needed.
+    Generations { sides: usize, given: usize },
+    /// A side is stale.
+    Table(JoinRefusal),
+    Relalg(RelError),
+    Exec(ExecError),
+}
+
+/// Rows of one side the `Rel` selected.
+#[derive(Debug, Clone, PartialEq, Eq)]
+enum Selected {
+    /// Every row (`Rel` normalized to `All`): read the side as prepared.
+    All,
+    /// The rows in a mask over the side's rows, and the count per slot.
+    Some { rows: Vec<u64>, counts: Vec<i64> },
+}
+
+/// The operator's result: the joint slots, each side's selection, and cost.
+pub struct CoEval {
+    pub joint: Vec<u64>,
+    selected: Vec<Selected>,
+    pub cost: CoCost,
+    /// Every mask-risc program the operator ran.
+    pub programs: Vec<Program>,
+}
+
+/// Runs `rels[i]` over side `i` of `table` and intersects the sides on the
+/// shared domain. Refused unless the table holds exactly `plan`'s sides at
+/// the generations in `current`.
+pub fn eval(
+    table: &CoTable,
+    plan: &StarPlan,
+    rels: &[Rel],
+    current: &[u64],
+) -> Result<CoEval, CoRefusal> {
+    let wanted: Vec<SideKey> = plan.sides.iter().map(|s| layout_key(s.key)).collect();
+    if wanted != table.keys {
+        return Err(CoRefusal::NoCoordinateWitness);
+    }
+    let k = table.sides.len();
+    if rels.len() != k {
+        return Err(CoRefusal::Rels {
+            sides: k,
+            rels: rels.len(),
+        });
+    }
+    if current.len() != k {
+        return Err(CoRefusal::Generations {
+            sides: k,
+            given: current.len(),
+        });
+    }
+    for (i, (s, &now)) in table.sides.iter().zip(current).enumerate() {
+        if s.side.generation != now {
+            return Err(CoRefusal::Table(JoinRefusal::Stale {
+                side: i,
+                bound: s.side.generation,
+                current: now,
+            }));
+        }
+    }
+    let n = table.domain.len();
+    let words = words_for(n);
+    let mut cost = CoCost::default();
+    let mut programs = Vec::new();
+    let mut selected = Vec::with_capacity(k);
+    let mut planes: Vec<Vec<u64>> = Vec::with_capacity(k);
+    for (s, rel) in table.sides.iter().zip(rels) {
+        let (sel, plane) = select(s, rel, n, &mut cost, &mut programs)?;
+        selected.push(sel);
+        planes.push(plane);
+    }
+    let masks: Vec<&[u64]> = table
+        .sides
+        .iter()
+        .zip(&planes)
+        .zip(&selected)
+        .map(|((s, p), sel)| match sel {
+            Selected::All => s.side.present(),
+            Selected::Some { .. } => p.as_slice(),
+        })
+        .collect();
+    let view = Planes {
+        n_rows: n,
+        masks: &masks,
+        lanes: &[],
+    };
+    let rel = Rel::And((0..k as u16).map(Rel::Plane).collect());
+    let mut joint = vec![0u64; words];
+    let (run, program) = run_with_program(
+        &Query {
+            rel,
+            measure: Measure::Keep,
+        },
+        &view,
+        Some(&mut joint),
+    )
+    .map_err(CoRefusal::Relalg)?;
+    add_run(&mut cost, run, program, &mut programs);
+    Ok(CoEval {
+        joint,
+        selected,
+        cost,
+        programs,
+    })
+}
+
+fn add_run(cost: &mut CoCost, run: Run, program: Option<Program>, programs: &mut Vec<Program>) {
+    cost.programs += run.programs;
+    cost.passes += run.passes;
+    if let Some(p) = program {
+        programs.push(p);
+    }
+}
+
+/// One side's selection and, when filtered, its slot plane.
+fn select(
+    s: &CoSide,
+    rel: &Rel,
+    n: usize,
+    cost: &mut CoCost,
+    programs: &mut Vec<Program>,
+) -> Result<(Selected, Vec<u64>), CoRefusal> {
+    let rel = normalize(rel);
+    if rel == Rel::All {
+        return Ok((Selected::All, Vec::new()));
+    }
+    let rows_n = s.n_rows();
+    let row_words = words_for(rows_n);
+    let mut rows = vec![0u64; row_words];
+    let mut counts = vec![0i64; n];
+    let mut plane = vec![0u64; words_for(n)];
+    if rel != Rel::Empty && rows_n > 0 {
+        let lanes = [LaneRef::U32(s.side.values())];
+        let view = Planes {
+            n_rows: rows_n,
+            masks: &[],
+            lanes: &lanes,
+        };
+        let (run, program) = run_with_program(
+            &Query {
+                rel,
+                measure: Measure::Keep,
+            },
+            &view,
+            Some(&mut rows),
+        )
+        .map_err(CoRefusal::Relalg)?;
+        add_run(cost, run, program, programs);
+        project(s, &rows, &mut counts, cost, programs)?;
+        cost.sides_filtered += 1;
+        cost.side_rows_read += rows_n;
+        cost.slots_scanned += n;
+        for (slot, &c) in counts.iter().enumerate() {
+            if c > 0 {
+                plane[slot >> 6] |= 1 << (slot & 63);
+            }
+        }
+    }
+    Ok((Selected::Some { rows, counts }, plane))
+}
+
+/// `counts[slot] = |{ selected rows in slot's run }|`, one mask-risc
+/// `GroupReduce Count` keyed by the side's own slot lane.
+fn project(
+    s: &CoSide,
+    rows: &[u64],
+    counts: &mut [i64],
+    cost: &mut CoCost,
+    programs: &mut Vec<Program>,
+) -> Result<(), CoRefusal> {
+    let program = Program::new(
+        Vec::new(),
+        Terminal::GroupReduce {
+            mask: Operand::Plane(0),
+            key: GroupKey::Lane(0),
+            fold: GroupFold::Count,
+        },
+    );
+    let masks = [rows];
+    let lanes = [LaneRef::U32(&s.slot)];
+    let view = Planes {
+        n_rows: s.n_rows(),
+        masks: &masks,
+        lanes: &lanes,
+    };
+    let mut scratch = Scratch::new(words_for(s.n_rows()), (program.scratch_slots as usize).max(1));
+    execute_into(&program, &view, &Foreign::NONE, &mut scratch, Out::I64(counts))
+        .map_err(CoRefusal::Exec)?;
+    cost.programs += 1;
+    cost.passes += program.ops.len() + 1;
+    programs.push(program);
+    Ok(())
+}
+
+impl CoEval {
+    /// Selected rows of side `i` at slot `s`.
+    fn mult(&self, table: &CoTable, i: usize, s: usize) -> u64 {
+        match &self.selected[i] {
+            Selected::All => table.sides[i].run(s).len() as u64,
+            Selected::Some { counts, .. } => counts[s] as u64,
+        }
+    }
+
+    /// Selected values of side `i` at slot `s`, in row order.
+    fn values_at<'a>(
+        &'a self,
+        table: &'a CoTable,
+        i: usize,
+        s: usize,
+    ) -> impl Iterator<Item = u32> + 'a {
+        let side = &table.sides[i];
+        let rows = match &self.selected[i] {
+            Selected::All => None,
+            Selected::Some { rows, .. } => Some(rows.as_slice()),
+        };
+        side.run(s).filter_map(move |j| {
+            let keep = rows.is_none_or(|m| m[j >> 6] >> (j & 63) & 1 == 1);
+            keep.then(|| side.side.values()[j])
+        })
+    }
+
+    /// Distinct center values every side has a selected row at.
+    pub fn distinct_centers(&self) -> u64 {
+        self.joint.iter().map(|w| u64::from(w.count_ones())).sum()
+    }
+
+    /// Bag answers of the star, `Σ_slot Π_i mult_i(slot)`: a co-addressed dot
+    /// product of the per-side counts, without enumerating a row.
+    pub fn bag_count(&self, table: &CoTable) -> u64 {
+        let mut total = 0u64;
+        for_each_set(&self.joint, table.domain.len(), |s| {
+            total += (0..table.sides.len())
+                .map(|i| self.mult(table, i, s))
+                .product::<u64>();
+        });
+        total
+    }
+
+    /// Output tuples of `plan` with multiplicities, sorted, in the shape of
+    /// [`crate::native::bindings`]. Without `bag`, every multiplicity is 1.
+    pub fn bindings(&self, table: &CoTable, plan: &StarPlan, bag: bool) -> Vec<(Vec<u64>, u64)> {
+        let mut out: Vec<(Vec<u64>, u64)> = Vec::new();
+        for_each_set(&self.joint, table.domain.len(), |slot| {
+            let center = u64::from(table.domain.id(slot));
+            let mut tuples: Vec<(Vec<Option<u64>>, u64)> = vec![(vec![None; plan.sides.len()], 1)];
+            for (i, s) in plan.sides.iter().enumerate() {
+                if s.emit.is_some() {
+                    let vals: Vec<u32> = self.values_at(table, i, slot).collect();
+                    tuples = tuples
+                        .into_iter()
+                        .flat_map(|(t, m)| {
+                            vals.iter().map(move |&v| {
+                                let mut t = t.clone();
+                                t[i] = Some(u64::from(v));
+                                (t, m)
+                            })
+                        })
+                        .collect();
+                } else {
+                    let k = self.mult(table, i, slot);
+                    for t in &mut tuples {
+                        t.1 *= k;
+                    }
+                }
+            }
+            for (t, m) in tuples {
+                let row = plan
+                    .head
+                    .iter()
+                    .map(|h| match *h {
+                        HeadTerm::Const(c) => c,
+                        HeadTerm::Center => center,
+                        HeadTerm::Side(i) => t[i].expect("emitting side"),
+                    })
+                    .collect();
+                out.push((row, m));
+            }
+        });
+        out.sort();
+        let mut merged: Vec<(Vec<u64>, u64)> = Vec::new();
+        for (row, m) in out {
+            match merged.last_mut() {
+                Some((r, c)) if *r == row => *c += m,
+                _ => merged.push((row, m)),
+            }
+        }
+        if !bag {
+            for e in &mut merged {
+                e.1 = 1;
+            }
+        }
+        merged
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::logic::{RelId, VarId};
+    use crate::native::{self, SidePlan, has_foreign_hop};
+
+    const R: RelId = RelId(0);
+    const S: RelId = RelId(1);
+    const T: RelId = RelId(2);
+
+    /// `Out(x, c, z) <- R(x, c), S(c, z), T(c, k)`: three binary sides on `c`.
+    fn plan(r: Option<u64>, s: Option<u64>, t: Option<u64>) -> StarPlan {
+        let side = |rel, key_col, restrict, emit| SidePlan {
+            key: SideKey {
+                rel,
+                key_col,
+                restrict,
+                unary: false,
+            },
+            emit,
+        };
+        StarPlan {
+            center: VarId(1),
+            sides: vec![
+                side(R, 1, r, r.is_none().then_some(VarId(0))),
+                side(S, 0, s, s.is_none().then_some(VarId(2))),
+                side(T, 0, t, None),
+            ],
+            head: [
+                r.is_none().then_some(HeadTerm::Side(0)),
+                Some(HeadTerm::Center),
+                s.is_none().then_some(HeadTerm::Side(1)),
+            ]
+            .into_iter()
+            .flatten()
+            .collect(),
+        }
+    }
+
+    /// Deterministic rows: `n` rows, both columns below `m`.
+    fn rows(seed: u64, n: usize, m: u32) -> Vec<[u32; 2]> {
+        let mut x = seed;
+        let mut next = || {
+            x ^= x << 13;
+            x ^= x >> 7;
+            x ^= x << 17;
+            (x % u64::from(m)) as u32
+        };
+        (0..n).map(|_| [next(), next()]).collect()
+    }
+
+    struct Fixture {
+        r: Vec<[u32; 2]>,
+        s: Vec<[u32; 2]>,
+        t: Vec<[u32; 2]>,
+    }
+
+    impl Fixture {
+        fn new() -> Self {
+            Fixture {
+                r: rows(0x9E37_79B9, 600, 40),
+                s: rows(0x7F4A_7C15, 500, 40),
+                t: rows(0x2545_F491, 400, 40),
+            }
+        }
+        fn store(&self) -> Store<'_> {
+            Store {
+                relations: vec![(R, &self.r, 1), (S, &self.s, 2), (T, &self.t, 3)],
+            }
+        }
+    }
+
+    /// Nested loops over the rows: the reference both paths must equal.
+    fn brute(
+        f: &Fixture,
+        p: &StarPlan,
+        keep: [&dyn Fn(u32) -> bool; 3],
+        bag: bool,
+    ) -> Vec<(Vec<u64>, u64)> {
+        let mut out: Vec<(Vec<u64>, u64)> = Vec::new();
+        for r in &f.r {
+            for s in &f.s {
+                for t in &f.t {
+                    let c = r[1];
+                    if s[0] != c || t[0] != c || !keep[0](r[0]) || !keep[1](s[1]) || !keep[2](t[1])
+                    {
+                        continue;
+                    }
+                    let vals = [r[0], s[1], t[1]];
+                    let row = p
+                        .head
+                        .iter()
+                        .map(|h| match *h {
+                            HeadTerm::Const(k) => k,
+                            HeadTerm::Center => u64::from(c),
+                            HeadTerm::Side(i) => u64::from(vals[i]),
+                        })
+                        .collect();
+                    out.push((row, 1));
+                }
+            }
+        }
+        out.sort();
+        let mut merged: Vec<(Vec<u64>, u64)> = Vec::new();
+        for (row, m) in out {
+            match merged.last_mut() {
+                Some((r, c)) if *r == row => *c += m,
+                _ => merged.push((row, m)),
+            }
+        }
+        if !bag {
+            for e in &mut merged {
+                e.1 = 1;
+            }
+        }
+        merged
+    }
+
+    const GENS: [u64; 3] = [1, 2, 3];
+
+    /// The operator over one unrestricted table equals the star path prepared
+    /// with each restriction, for every restriction combination tried, in
+    /// bindings (bag and set), bag count and distinct centers.
+    #[test]
+    fn parity_with_the_star_path() {
+        let f = Fixture::new();
+        let store = f.store();
+        let table = CoTable::prepare(&plan(None, None, None), &store);
+        let mut nonempty = 0;
+        for (r, s, t) in [
+            (None, None, None),
+            (Some(3), None, None),
+            (None, Some(7), Some(11)),
+            (Some(5), Some(9), Some(13)),
+            (Some(u64::from(u32::MAX) + 1), None, None),
+        ] {
+            let p = plan(r, s, t);
+            let ev = eval(&table, &p, &plan_rels(&p), &GENS).expect("eval");
+            let star = native::prepare(&p, &store);
+            for bag in [true, false] {
+                let want = native::bindings(&star, &p, &GENS, bag).expect("star");
+                assert_eq!(ev.bindings(&table, &p, bag), want, "{r:?} {s:?} {t:?} bag={bag}");
+            }
+            assert_eq!(
+                ev.distinct_centers(),
+                native::distinct_centers(&star, &p, &GENS).expect("star")
+            );
+            let bag_rows: u64 = native::bindings(&star, &p, &GENS, true)
+                .expect("star")
+                .iter()
+                .map(|(_, m)| m)
+                .sum();
+            assert_eq!(ev.bag_count(&table), bag_rows);
+            nonempty += usize::from(bag_rows > 0);
+        }
+        // Not vacuous: most cases have answers, so equality compares rows.
+        assert!(nonempty >= 4, "only {nonempty} cases had answers");
+    }
+
+    /// A restriction the star path cannot express (`Or`, `Not`) matches the
+    /// nested-loop reference.
+    #[test]
+    fn disjunctive_restrictions_match_the_reference() {
+        let f = Fixture::new();
+        let store = f.store();
+        let p = plan(None, None, None);
+        let table = CoTable::prepare(&p, &store);
+        let rels = vec![
+            Rel::Or(vec![Rel::eq_u32(VALUE_LANE, 1), Rel::eq_u32(VALUE_LANE, 2)]),
+            Rel::Not(Box::new(Rel::eq_u32(VALUE_LANE, 4))),
+            Rel::ne_u32(VALUE_LANE, 0),
+        ];
+        let ev = eval(&table, &p, &rels, &GENS).expect("eval");
+        let keep: [&dyn Fn(u32) -> bool; 3] = [&|x| x == 1 || x == 2, &|z| z != 4, &|k| k != 0];
+        for bag in [true, false] {
+            let want = brute(&f, &p, keep, bag);
+            assert!(!want.is_empty(), "fixture must have answers");
+            assert_eq!(ev.bindings(&table, &p, bag), want);
+        }
+        let total: u64 = brute(&f, &p, keep, true).iter().map(|(_, m)| m).sum();
+        assert_eq!(ev.bag_count(&table), total);
+        assert_eq!(ev.cost.sides_filtered, 3);
+        assert_eq!(ev.cost.side_rows_read, table.rows());
+    }
+
+    /// No program the operator runs reads at a position taken from another
+    /// operand's values, and with a filtered side it does run programs.
+    #[test]
+    fn no_gather() {
+        let f = Fixture::new();
+        let store = f.store();
+        let p = plan(Some(3), Some(7), None);
+        let table = CoTable::prepare(&plan(None, None, None), &store);
+        let ev = eval(&table, &p, &plan_rels(&p), &GENS).expect("eval");
+        // Two filtered sides: a fold and a projection each, plus the joint.
+        assert_eq!(ev.programs.len(), 5, "{:?}", ev.programs);
+        for prog in &ev.programs {
+            assert!(!has_foreign_hop(prog), "foreign hop in {prog:?}");
+        }
+    }
+
+    /// Unrestricted sides cost nothing past the star path: one program, the
+    /// joint, and no side row read.
+    #[test]
+    fn all_sides_read_as_prepared() {
+        let f = Fixture::new();
+        let store = f.store();
+        let p = plan(None, None, None);
+        let table = CoTable::prepare(&p, &store);
+        let ev = eval(&table, &p, &plan_rels(&p), &GENS).expect("eval");
+        assert_eq!(ev.cost.side_rows_read, 0);
+        assert_eq!(ev.cost.sides_filtered, 0);
+        assert_eq!(ev.programs.len(), 1);
+    }
+
+    #[test]
+    fn refusals() {
+        let f = Fixture::new();
+        let store = f.store();
+        let p = plan(Some(3), None, None);
+        let table = CoTable::prepare(&p, &store);
+        let rels = plan_rels(&p);
+        assert!(matches!(
+            eval(&table, &p, &rels, &[1, 2, 4]),
+            Err(CoRefusal::Table(JoinRefusal::Stale { side: 2, .. }))
+        ));
+        assert_eq!(
+            eval(&table, &p, &rels, &[1, 2]).err(),
+            Some(CoRefusal::Generations { sides: 3, given: 2 })
+        );
+        assert_eq!(
+            eval(&table, &p, &rels[..2], &GENS).err(),
+            Some(CoRefusal::Rels { sides: 3, rels: 2 })
+        );
+        let mut other = p.clone();
+        other.sides[2].key.rel = R;
+        assert_eq!(
+            eval(&table, &other, &rels, &GENS).err(),
+            Some(CoRefusal::NoCoordinateWitness)
+        );
+        // A restriction is not part of the layout: the same table answers it.
+        assert!(eval(&table, &plan(None, Some(1), None), &rels, &GENS).is_ok());
+    }
+}
diff --git a/crates/crepe-plan-probe/src/lib.rs b/crates/crepe-plan-probe/src/lib.rs
index cb75078a4..0cad5c44d 100644
--- a/crates/crepe-plan-probe/src/lib.rs
+++ b/crates/crepe-plan-probe/src/lib.rs
@@ -31,6 +31,7 @@
 //!    are the other oracles (tests only).
 
 pub mod addressed;
+pub mod coaddressed;
 pub mod cse;
 pub mod lance_store;
 pub mod logic;
diff --git a/crates/mork-pathmap-probe/src/join.rs b/crates/mork-pathmap-probe/src/join.rs
index 9ce202f7a..c9f3cd3c5 100644
--- a/crates/mork-pathmap-probe/src/join.rs
+++ b/crates/mork-pathmap-probe/src/join.rs
@@ -159,6 +159,17 @@ impl Side {
         &self.present
     }
 
+    /// The other column, in slot order: the side's own row order. Row `i`
+    /// sits in slot `s` for `offsets()[s] <= i < offsets()[s + 1]`.
+    pub fn values(&self) -> &[u32] {
+        &self.vals
+    }
+
+    /// `off`: the start of each slot's run, `domain.len() + 1` entries.
+    pub fn offsets(&self) -> &[u32] {
+        &self.off
+    }
+
     pub fn bytes(&self) -> usize {
         (self.present.len() * 8) + (self.off.len() * 4) + (self.vals.len() * 4)
     }

From 86d8322030663e695a4e8b347da981391512c2be Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Wed, 21 Oct 2026 08:28:59 +0000
Subject: [PATCH 2/6] U4: benchmark and fixture for the co-addressed join

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f
---
 .../crepe-plan-probe/examples/coaddressed.rs  | 212 ++++++++++++++++++
 1 file changed, 212 insertions(+)
 create mode 100644 crates/crepe-plan-probe/examples/coaddressed.rs

diff --git a/crates/crepe-plan-probe/examples/coaddressed.rs b/crates/crepe-plan-probe/examples/coaddressed.rs
new file mode 100644
index 000000000..9378b89c2
--- /dev/null
+++ b/crates/crepe-plan-probe/examples/coaddressed.rs
@@ -0,0 +1,212 @@
+//! D-RU-4 (U4): the co-addressed join operator against the star path, over a
+//! family of restricted queries on one star. Run from this crate's directory:
+//!
+//! ```text
+//! cargo run --release --example coaddressed    # the benchmark
+//! cargo test --example coaddressed             # the correctness fixture
+//! ```
+//!
+//! # The queries
+//!
+//! Query `q` (one per `q` in `0..QUERIES`):
+//!
+//! ```text
+//! Out(x, c, z) <- R(x, c), S(c, z), T(c, q)
+//! ```
+//!
+//! # Arms
+//!
+//! * **C** — [`native::prepare`] per query: the restriction `T.k == q` is
+//!   applied to the rows before layout, so every query lays out all three
+//!   sides again.
+//! * **CS** — [`native::prepare_semijoin`] per query: the same, and R and S
+//!   keep only the centers the restricted T has.
+//! * **O** — [`coaddressed::eval`]: the table is laid out once, without the
+//!   restriction; each query runs `T.k == q` as a `Rel` over T's rows and
+//!   intersects on the shared domain.
+//!
+//! Every arm answers the bag count and the distinct-center count of every
+//! query, and the answers are compared.
+
+use std::time::{Duration, Instant};
+
+use crepe_plan_probe::coaddressed::{self, CoTable, plan_rels};
+use crepe_plan_probe::logic::{RelId, VarId};
+use crepe_plan_probe::native::{self, HeadTerm, SideKey, SidePlan, StarPlan, Store};
+use mork_pathmap_probe::join::JoinExec;
+
+const R: RelId = RelId(0);
+const S: RelId = RelId(1);
+const T: RelId = RelId(2);
+const GENS: [u64; 3] = [1, 2, 3];
+
+fn env(name: &str, default: usize) -> usize {
+    std::env::var(name)
+        .ok()
+        .and_then(|v| v.parse().ok())
+        .unwrap_or(default)
+}
+
+fn plan(q: Option<u64>) -> StarPlan {
+    let side = |rel, key_col, restrict, emit| SidePlan {
+        key: SideKey {
+            rel,
+            key_col,
+            restrict,
+            unary: false,
+        },
+        emit,
+    };
+    StarPlan {
+        center: VarId(1),
+        sides: vec![
+            side(R, 1, None, Some(VarId(0))),
+            side(S, 0, None, Some(VarId(2))),
+            side(T, 0, q, None),
+        ],
+        head: vec![HeadTerm::Side(0), HeadTerm::Center, HeadTerm::Side(1)],
+    }
+}
+
+/// `n` rows: column `0` below `a`, column `1` below `b`.
+fn rows(seed: u64, n: usize, a: u32, b: u32) -> Vec<[u32; 2]> {
+    let mut x = seed;
+    let mut next = |m: u32| {
+        x ^= x << 13;
+        x ^= x >> 7;
+        x ^= x << 17;
+        (x % u64::from(m)) as u32
+    };
+    (0..n).map(|_| [next(a), next(b)]).collect()
+}
+
+struct Data {
+    r: Vec<[u32; 2]>,
+    s: Vec<[u32; 2]>,
+    t: Vec<[u32; 2]>,
+}
+
+impl Data {
+    fn new(n: usize, centers: u32, queries: u32) -> Data {
+        Data {
+            r: rows(0x9E37_79B9, n, 1_000_000, centers),
+            s: rows(0x7F4A_7C15, n, centers, 1_000_000),
+            t: rows(0x2545_F491, n, centers, queries),
+        }
+    }
+    fn store(&self) -> Store<'_> {
+        Store {
+            relations: vec![(R, &self.r, 1), (S, &self.s, 2), (T, &self.t, 3)],
+        }
+    }
+}
+
+/// `(bag count, distinct centers)` per query, rows laid out or read, time.
+type ArmRun = (Vec<(u64, u64)>, usize, Duration);
+
+fn cold(data: &Data, queries: u32, semijoin: bool) -> ArmRun {
+    let store = data.store();
+    let start = Instant::now();
+    let mut out = Vec::new();
+    let mut rows = 0;
+    for q in 0..queries {
+        let p = plan(Some(u64::from(q)));
+        let prep = if semijoin {
+            native::prepare_semijoin(&p, &store)
+        } else {
+            native::prepare(&p, &store)
+        };
+        rows += prep.rows_prepared;
+        let bag = JoinExec::new(&prep.table)
+            .bag_count(&prep.table, &GENS, None)
+            .expect("bag");
+        let centers = native::distinct_centers(&prep, &p, &GENS).expect("centers");
+        out.push((bag, centers));
+    }
+    (out, rows, start.elapsed())
+}
+
+/// `(answers, side rows read, eval time)` plus the one layout's time and rows.
+fn operator(data: &Data, queries: u32) -> (ArmRun, Duration, usize) {
+    let store = data.store();
+    let start = Instant::now();
+    let table = CoTable::prepare(&plan(None), &store);
+    let prepare = start.elapsed();
+    let start = Instant::now();
+    let mut out = Vec::new();
+    let mut rows = 0;
+    for q in 0..queries {
+        let p = plan(Some(u64::from(q)));
+        let ev = coaddressed::eval(&table, &p, &plan_rels(&p), &GENS).expect("eval");
+        assert!(ev.programs.iter().all(|p| !native::has_foreign_hop(p)));
+        rows += ev.cost.side_rows_read;
+        out.push((ev.bag_count(&table), ev.distinct_centers()));
+    }
+    ((out, rows, start.elapsed()), prepare, table.rows())
+}
+
+fn ms(d: Duration) -> f64 {
+    d.as_secs_f64() * 1e3
+}
+
+fn main() {
+    let n = env("BENCH_ROWS", 200_000);
+    let centers = env("BENCH_CENTERS", 20_000) as u32;
+    let queries = env("BENCH_QUERIES", 64) as u32;
+    let reps = env("BENCH_REPS", 3);
+    println!(
+        "simd: avx512f={}  rows/relation={n}  centers={centers}  queries={queries}  reps={reps}",
+        cfg!(target_feature = "avx512f")
+    );
+    let data = Data::new(n, centers, queries);
+    let mut best: [Option<(f64, usize)>; 3] = [None; 3];
+    let mut layout = (f64::MAX, 0);
+    for _ in 0..reps {
+        let c = cold(&data, queries, false);
+        let cs = cold(&data, queries, true);
+        let ((o, o_rows, o_t), prep, prep_rows) = operator(&data, queries);
+        assert_eq!(c.0, o, "C and O disagree");
+        assert_eq!(cs.0, o, "CS and O disagree");
+        for (slot, (t, r)) in best
+            .iter_mut()
+            .zip([(ms(c.2), c.1), (ms(cs.2), cs.1), (ms(o_t), o_rows)])
+        {
+            if slot.is_none_or(|(b, _)| t < b) {
+                *slot = Some((t, r));
+            }
+        }
+        layout.0 = layout.0.min(ms(prep));
+        layout.1 = prep_rows;
+    }
+    let bag: u64 = cold(&data, queries, false).0.iter().map(|a| a.0).sum();
+    println!("answers equal across arms; total bag rows over all queries: {bag}");
+    println!("{:<4} {:>12} {:>16}", "arm", "ms (best)", "rows touched");
+    for (name, b) in ["C", "CS", "O"].iter().zip(best) {
+        let (t, r) = b.expect("ran");
+        println!("{name:<4} {t:>12.2} {r:>16}");
+    }
+    println!(
+        "O layout (once): {:.2} ms, {} rows; O total incl. layout: {:.2} ms",
+        layout.0,
+        layout.1,
+        layout.0 + best[2].expect("ran").0
+    );
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// All three arms give the same per-query answers, and the fixture has
+    /// answers to compare.
+    #[test]
+    fn arms_agree() {
+        let data = Data::new(3_000, 200, 8);
+        let c = cold(&data, 8, false);
+        let cs = cold(&data, 8, true);
+        let ((o, _, _), _, _) = operator(&data, 8);
+        assert_eq!(c.0, o);
+        assert_eq!(cs.0, o);
+        assert!(o.iter().filter(|a| a.0 > 0).count() >= 6, "{o:?}");
+    }
+}

From 9ad0ed34777ae0bbc628ed9752576dc1184eec00 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Wed, 21 Oct 2026 08:47:38 +0000
Subject: [PATCH 3/6] U4 (D-RU-4): co-addressed join, measured; CI step and
 board

Per-side Rels at query time over one table laid out without them.
64 restricted star queries at 200k rows: 1423 ms laying out per query,
776 ms with semijoin, 12.4 ms + 30.4 ms one layout here. Answers equal
the star path and a nested-loop reference; no program has a foreign hop.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f
---
 .claude/board/STATUS_BOARD.md                 |  2 +-
 .../entries/2026-10-21-u4-coaddressed-join.md | 53 +++++++++++++++++++
 .claude/board/entries/README.md               |  3 +-
 .claude/plans/d-relalg-unified-v1.md          |  2 +-
 .github/workflows/crepe-plan-probe.yml        |  4 ++
 .../crepe-plan-probe/examples/coaddressed.rs  |  6 +--
 crates/crepe-plan-probe/src/coaddressed.rs    | 37 +++++++++----
 7 files changed, 92 insertions(+), 15 deletions(-)
 create mode 100644 .claude/board/entries/2026-10-21-u4-coaddressed-join.md

diff --git a/.claude/board/STATUS_BOARD.md b/.claude/board/STATUS_BOARD.md
index 71c00a051..287aa2042 100644
--- a/.claude/board/STATUS_BOARD.md
+++ b/.claude/board/STATUS_BOARD.md
@@ -8,7 +8,7 @@ Plan: `.claude/plans/d-relalg-unified-v1.md`. Decisions recorded there: ADMISSIO
 | **D-RU-1** | U1: JC statistical batch, distinct (selection, key, lane pair) triples over shared power-sum terminals | MEASURED (`crates/jc-batch-probe`; `entries/2026-10-20-u1-jc-batch.md`): 252 requests → 2 selection + 42 cross passes; at 1M rows 6.4× faster, 11.5× fewer rows traversed, 8.5× fewer values read, 126× fewer mask bytes; bit-identical to per-request; no R reference (R absent) | rows traversed and terminal passes below the per-request baseline; results bit-exact on integer lanes; overflow refused, not wrapped; R references |
 | **D-RU-2** | U2: `let`-bound planes on the contract IR, immutable `Plane(i)`, versioned coordinates, no DAG inside a fold | MEASURED (`contract::set_program`, `relalg::program`; `entries/2026-10-21-u2-let-planes.md`): 16 queries sharing one conjunction at 1M rows: 3.1× faster, 14 vs 59 lane reads, 2 planes (250 kB) materialized, ≤ 2 scratch slots; answers equal recompute and a row oracle; other versions refused | rows traversed and planes materialized against recompute; width test ≤ 3 slots; answers and versions unchanged |
 | **D-RU-3** | cross-rule CSE in crepe-plan-probe (256-rule benchmark, separate from the hop benchmark) | MEASURED (`crepe-plan-probe/src/cse.rs`, `examples/rule_cse.rs`; `entries/2026-10-21-u3-cross-rule-cse.md`): 256 rules, 90 % shared core, 200k centers: rows laid out 15.3 M → 6.2 M, prepare 1.78 s → 1.22 s; lets cut ops 512 → 283 but folds are ≤ 2.4 ms and `share` costs 7 ms; one table for all rules is slower than per-rule when nothing is shared, so tables are per group of rules sharing a side; answers equal crepe, ascent and the reference | answers == crepe and ascent; traversals and materializations measured |
-| **D-RU-4** | co-addressed join operator over `JoinDomain` | Queued (after D-RU-2) | parity with the star path; no position taken from another table's values |
+| **D-RU-4** | co-addressed join operator over `JoinDomain` | MEASURED (`crepe-plan-probe/src/coaddressed.rs`, `examples/coaddressed.rs`; `entries/2026-10-21-u4-coaddressed-join.md`): per-side `Rel`s at query time over one unrestricted table, `GroupReduce Count` on each side's own slot lane, `And` of slot planes; 64 restricted star queries at 200k rows: 1423 ms (re-layout per query) / 776 ms (with semijoin) → 12.4 ms + 30.4 ms one layout; answers equal the star path, the nested-loop reference and every arm; no program has a foreign hop (disable-verified) | parity with the star path; no position taken from another table's values |
 | **D-RU-5** | Quack physical pins (dense vs sorted, materialize vs recompute) | Queued (after D-RU-3/4) | can-fire and can-stay-silent per pin |
 | **D-RU-6** | recursion driver over BIND tiles (256-hop benchmark) | Queued/Lab (slot assignment must be argued as an attested frozen-codebook lookup under ADMISSION = B). Lab probe MEASURED (`crepe-plan-probe/examples/walk_tile.rs`, `entries/2026-10-21-d-ru-6-tile-walk-probe.md`): tuple sets equal ascent, crepe and walk_mask; execution gather-free; binding witness probe-local, so ADMISSION B is NOT demonstrated; count-only 1.0–9.3× slower than walk_mask, end to end 1.1–2.0× slower. Block-sparse arm C MEASURED (`examples/walk_sparse.rs`, `entries/2026-10-21-d-ru-6-sparse-compute.md`): faster than the dense tile path on every fixture (0.11–0.96× count-only); against walk_mask end to end it wins only on deep low-fan-out recursion (chain(230) 0.82×, chain(700) 0.59× UNCHECKED), is near parity on sparse graphs (1.03–1.07×) and loses on dense, cyclic, clustered and mixed-length graphs (1.20–2.43×). Arm A′ (walk_mask with a dirty-frontier skip, `examples/walk_dirty.rs`, `entries/2026-10-21-d-ru-6-dirty-frontier.md`) does not remove C's chain lead; the chain sweep (two runs, distinct fan-out, like-for-like timing) confines C's end-to-end win over A′ to unbranched chains of about 192 nodes or more (3–30 %); with fan-out 2 or 4 C loses end to end at most points. Recommendation: defer the codebook attestation | least fixpoint == datafrog; ≥ 200 iterations asserted; cycles terminate. Benchmark requirement (operator, 2026-10-21; refs D-RMAP-2 §5b, #1642): tuple materialization, count-only evaluation, memory growth, address binding and value-derived reads by origin reported separately; tuple-set equality with crepe and ascent wherever they run |
 | **D-RU-7** | prefetch and staging of Arm B ranges | Queued (after D-RU-6 or a non-recursive multi-stage plan) | critical-path overlap measured; bytes read unchanged |
diff --git a/.claude/board/entries/2026-10-21-u4-coaddressed-join.md b/.claude/board/entries/2026-10-21-u4-coaddressed-join.md
new file mode 100644
index 000000000..670225870
--- /dev/null
+++ b/.claude/board/entries/2026-10-21-u4-coaddressed-join.md
@@ -0,0 +1,53 @@
+# 2026-10-21 — D-RU-4 (U4): the co-addressed join as a named operator
+
+**Status:** MEASURED (lab; `crates/crepe-plan-probe/src/coaddressed.rs`, `examples/coaddressed.rs`). Plan: `.claude/plans/d-relalg-unified-v1.md` §5.3, §6, stage table (U4). Builds on U2 (`relalg::run_with_program`) and the D-MPS-8 table (`mork_pathmap_probe::join`).
+
+## What landed
+
+- `coaddressed::eval(table, plan, rels, current)`: a conjunction of per-table `Rel`s over one `JoinDomain`. For each side `i`:
+  - `rel_i` runs as a relalg fold over side `i`'s own rows (lane 0 = the side's other column);
+  - one mask-risc `GroupReduce Count` keyed by the side's slot lane turns the selected rows into per-slot counts;
+  - a side whose `Rel` normalizes to `All` reads its presence plane and offsets as prepared, with no row read.
+- The joint is relalg `And(Plane(i))` over the per-side slot planes. Answers:
+  - `distinct_centers`, the popcount of the joint;
+  - `bag_count`, a co-addressed dot product of the per-side counts (`Σ_slot Π_i count_i[slot]`) that enumerates no row;
+  - `bindings`, the tuples in `native::bindings`' shape, reading each side at the joint's slot and testing the row bits in that slot's run.
+- `CoTable::prepare` lays out every side without its restriction. `plan_rels` turns a `StarPlan`'s restrictions into `value == k` (or `Empty` for a constant above `u32::MAX`). A restriction is therefore a query-time `Rel` and no longer part of the layout. The `Rel` can be anything relalg lowers on a `U32` lane (`==`, `!=`, `Or`, `Not`), which the star path cannot express.
+- The slot lane is built from the side's own offsets, so it is non-decreasing. A row is counted in the run it already sits in, and no operand is read at a position taken from another operand's values. `CoEval::programs` keeps every program run so a caller can check that with `native::has_foreign_hop`.
+- Refusals: a table holding other sides than the plan reads, the wrong number of `Rel`s or generations, a stale side.
+- `mork_pathmap_probe::join::Side` gains two read-only accessors, `values()` and `offsets()`.
+
+## Measured
+
+Workload: 64 queries `Out(x, c, z) <- R(x, c), S(c, z), T(c, q)` for `q` in `0..64`. 200,000 rows per relation, 20,000 centers. Release build, AVX-512, best of 3.
+- Arms: C is `native::prepare` per query, which lays out all three sides again with T restricted; CS adds the semijoin; O is one `CoTable` plus `eval` per query.
+
+| arm | time, 64 queries | rows touched |
+|---|---|---|
+| C | 1423 ms | 25.8 M laid out |
+| CS | 776 ms | 3.9 M laid out |
+| O | 12.4 ms (+ 30.4 ms one layout) | 12.8 M read (T's value and slot lanes, 64 × 200k) |
+
+**Findings.**
+- With the restriction moved out of the layout, O answers all 64 queries in 42.8 ms including its one layout: 33× faster than C and 18× faster than CS.
+- CS lays out fewer rows than O reads, but is still 63× slower per query. Its row count does not include the scan that filters 600k rows per query before layout, nor the sort. So "rows touched" here does not cover the same work across arms, and only the times are comparable.
+- This workload restricts one side. O reads only that side's rows per query; R and S are read as prepared. A query restricting all three sides would read all three per query.
+
+**Correctness.**
+- `parity_with_the_star_path`: for five restriction combinations (none; one; two; three; a constant above `u32::MAX`), bindings (bag and set), the bag count and distinct centers equal `native::prepare` + `native::bindings` / `distinct_centers`. At least four of the five cases have answers.
+- `disjunctive_restrictions_match_the_reference`: `Or`, `Not` and `!=` restrictions equal a nested-loop reference, bag and set.
+- The benchmark asserts that the answers of C, CS and O are equal on every rep, and `cargo test --example coaddressed` runs the same comparison on a small fixture in CI.
+- **Disables:** five, each red on the test it targets:
+  - a projection that scatters by the slot lane (`ScatterOrU32`, answer-correct) fails only `no_gather`, at the gather assertion;
+  - row bits ignored in tuple enumeration;
+  - per-slot counts replaced by run lengths;
+  - witness check removed;
+  - stale-generation check removed.
+- The first scatter disable used `ScatterCountU32` with an `Out::I64`. It failed on an `ExecError` before the gather assertion ran, so it proved nothing about that assertion. It was replaced with the answer-correct `ScatterOrU32` version above.
+
+## Open
+
+- The projection writes per-slot counts as `i64` for every filtered side (`8 × domain` bytes), and builds the slot plane from them in a Rust loop over the domain. A fused "count into plane" terminal would remove both; not built, not measured.
+- Each filtered side runs two programs (fold, then projection) and keeps the row mask for tuple enumeration. When only counts are asked, the fold and the projection could be one program; not measured.
+- No restriction on the center itself (a `Rel` over the domain) yet.
+- §6 factorization beyond the star (a chain through two domains) is not built.
diff --git a/.claude/board/entries/README.md b/.claude/board/entries/README.md
index 4f8798545..ee32789a5 100644
--- a/.claude/board/entries/README.md
+++ b/.claude/board/entries/README.md
@@ -25,10 +25,11 @@ 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.
 
-406 entries, 2026-08-06 .. 2026-10-21.
+407 entries, 2026-08-06 .. 2026-10-21.
 
 | date | entry id | finding | file |
 |---|---|---|---|
+| 2026-10-21 | `D-RU-4` |  | [2026-10-21-u4-coaddressed-join.md](2026-10-21-u4-coaddressed-join.md) |
 | 2026-10-21 | `D-RU-3` |  | [2026-10-21-u3-cross-rule-cse.md](2026-10-21-u3-cross-rule-cse.md) |
 | 2026-10-21 | `D-RU-2` |  | [2026-10-21-u2-let-planes.md](2026-10-21-u2-let-planes.md) |
 | 2026-10-21 | `D-RU-6-LAB-1` |  | [2026-10-21-d-ru-6-tile-walk-probe.md](2026-10-21-d-ru-6-tile-walk-probe.md) |
diff --git a/.claude/plans/d-relalg-unified-v1.md b/.claude/plans/d-relalg-unified-v1.md
index 73502a75c..6b8e95a3d 100644
--- a/.claude/plans/d-relalg-unified-v1.md
+++ b/.claude/plans/d-relalg-unified-v1.md
@@ -298,4 +298,4 @@ One workload, run under three toggles. Each toggle changes one mechanism only.
 | U7 | prefetch and staging of Arm B ranges | U6 or a non-recursive multi-stage plan | critical-path overlap measured; bytes read unchanged |
 | U8 | Wikidata and SAP-shaped fixtures; completeness refusals | U6, U7 | refusal tests; witness carried end to end |
 
-U1 needs nothing new and comes first. **U1 MEASURED (2026-10-20, `crates/jc-batch-probe`, `entries/2026-10-20-u1-jc-batch.md`):** 252 requests reduce to 2 selection and 42 cross passes; at 1M rows the batch is 6.4× faster with 11.5× fewer rows traversed, 8.5× fewer values read and 126× fewer mask bytes, and its answers equal the per-request answers bit for bit. The R reference is replaced by JC's slice functions because R is not installed. **U2 MEASURED (2026-10-21, `contract::set_program` + `relalg::program`, `entries/2026-10-21-u2-let-planes.md`):** each shared junction subset is evaluated once into a scratch plane; for 16 queries sharing one conjunction at 1M rows the program is 3.1× faster than recompute and reads 14 lanes instead of 59, at the cost of 2 materialized planes (250 kB). Answers equal recompute and a row oracle, and a program for another version is refused. **U3 MEASURED (2026-10-21, `crepe-plan-probe/src/cse.rs` + `examples/rule_cse.rs`, `entries/2026-10-21-u3-cross-rule-cse.md`):** the §7 claim "one core join plus 256 ANDs" holds for the layout, not for the folds. At 90 % shared core the core is laid out once (15.3 M → 6.2 M rows, prepare 1.78 s → 1.22 s); the lets cut ops 512 → 283, but the folds cost ≤ 2.4 ms and the subset search 7 ms. One table for all rules is slower than per-rule tables when nothing is shared, so tables are formed per group of rules that share a side. Both are accepted only on measured row traversals, materializations and total cost (§ Decisions), not on operator counts.
+U1 needs nothing new and comes first. **U1 MEASURED (2026-10-20, `crates/jc-batch-probe`, `entries/2026-10-20-u1-jc-batch.md`):** 252 requests reduce to 2 selection and 42 cross passes; at 1M rows the batch is 6.4× faster with 11.5× fewer rows traversed, 8.5× fewer values read and 126× fewer mask bytes, and its answers equal the per-request answers bit for bit. The R reference is replaced by JC's slice functions because R is not installed. **U2 MEASURED (2026-10-21, `contract::set_program` + `relalg::program`, `entries/2026-10-21-u2-let-planes.md`):** each shared junction subset is evaluated once into a scratch plane; for 16 queries sharing one conjunction at 1M rows the program is 3.1× faster than recompute and reads 14 lanes instead of 59, at the cost of 2 materialized planes (250 kB). Answers equal recompute and a row oracle, and a program for another version is refused. **U3 MEASURED (2026-10-21, `crepe-plan-probe/src/cse.rs` + `examples/rule_cse.rs`, `entries/2026-10-21-u3-cross-rule-cse.md`):** the §7 claim "one core join plus 256 ANDs" holds for the layout, not for the folds. At 90 % shared core the core is laid out once (15.3 M → 6.2 M rows, prepare 1.78 s → 1.22 s); the lets cut ops 512 → 283, but the folds cost ≤ 2.4 ms and the subset search 7 ms. One table for all rules is slower than per-rule tables when nothing is shared, so tables are formed per group of rules that share a side. **U4 MEASURED (2026-10-21, `crepe-plan-probe/src/coaddressed.rs` + `examples/coaddressed.rs`, `entries/2026-10-21-u4-coaddressed-join.md`):** the co-addressed join is a named operator, a conjunction of per-table `Rel`s over one `JoinDomain`. A side's restriction runs at query time over one table laid out without it, so 64 restricted queries cost 12.4 ms after one 30.4 ms layout, where laying out per query costs 1423 ms (776 ms with semijoin). Answers equal the star path; no program reads at a position taken from another operand's values. Both are accepted only on measured row traversals, materializations and total cost (§ Decisions), not on operator counts.
diff --git a/.github/workflows/crepe-plan-probe.yml b/.github/workflows/crepe-plan-probe.yml
index 2eb9ec389..5661d30d8 100644
--- a/.github/workflows/crepe-plan-probe.yml
+++ b/.github/workflows/crepe-plan-probe.yml
@@ -63,6 +63,10 @@ jobs:
       # The benchmark (`cargo run --release --example rule_cse`) is not run
       # in CI.
       - run: cargo test --example rule_cse
+      # D-RU-4 (U4): the co-addressed join operator against the star path.
+      # The benchmark (`cargo run --release --example coaddressed`) is not
+      # run in CI.
+      - run: cargo test --example coaddressed
       # D-RU-6 arm C: the block-sparse product, against arms A and B and the
       # oracles, its refusals and the semi-naive bound.
       - run: cargo test --example walk_sparse
diff --git a/crates/crepe-plan-probe/examples/coaddressed.rs b/crates/crepe-plan-probe/examples/coaddressed.rs
index 9378b89c2..81fb9bce6 100644
--- a/crates/crepe-plan-probe/examples/coaddressed.rs
+++ b/crates/crepe-plan-probe/examples/coaddressed.rs
@@ -167,9 +167,9 @@ fn main() {
         let ((o, o_rows, o_t), prep, prep_rows) = operator(&data, queries);
         assert_eq!(c.0, o, "C and O disagree");
         assert_eq!(cs.0, o, "CS and O disagree");
-        for (slot, (t, r)) in best
-            .iter_mut()
-            .zip([(ms(c.2), c.1), (ms(cs.2), cs.1), (ms(o_t), o_rows)])
+        for (slot, (t, r)) in
+            best.iter_mut()
+                .zip([(ms(c.2), c.1), (ms(cs.2), cs.1), (ms(o_t), o_rows)])
         {
             if slot.is_none_or(|(b, _)| t < b) {
                 *slot = Some((t, r));
diff --git a/crates/crepe-plan-probe/src/coaddressed.rs b/crates/crepe-plan-probe/src/coaddressed.rs
index a96513394..b2fcd3b9a 100644
--- a/crates/crepe-plan-probe/src/coaddressed.rs
+++ b/crates/crepe-plan-probe/src/coaddressed.rs
@@ -41,11 +41,11 @@
 //! costs nothing beyond the star path: its presence plane and offsets are
 //! read as prepared.
 
+use lance_graph_mask_risc::ExecError;
 use lance_graph_mask_risc::{
-    words_for, Foreign, GroupFold, GroupKey, LaneRef, Operand, Out, Planes, Program, Scratch,
-    Terminal, execute_into,
+    Foreign, GroupFold, GroupKey, LaneRef, Operand, Out, Planes, Program, Scratch, Terminal,
+    execute_into, words_for,
 };
-use lance_graph_mask_risc::ExecError;
 use lance_graph_relalg::{Measure, Query, Rel, RelError, Run, normalize, run_with_program};
 use mork_pathmap_probe::RelationView;
 use mork_pathmap_probe::join::{JoinDomain, JoinRefusal, JoinTable, Side, for_each_set};
@@ -168,9 +168,15 @@ pub enum CoRefusal {
     /// The table holds other sides than the plan reads.
     NoCoordinateWitness,
     /// One `Rel` per side is needed.
-    Rels { sides: usize, rels: usize },
+    Rels {
+        sides: usize,
+        rels: usize,
+    },
     /// One generation per side is needed.
-    Generations { sides: usize, given: usize },
+    Generations {
+        sides: usize,
+        given: usize,
+    },
     /// A side is stale.
     Table(JoinRefusal),
     Relalg(RelError),
@@ -355,9 +361,18 @@ fn project(
         masks: &masks,
         lanes: &lanes,
     };
-    let mut scratch = Scratch::new(words_for(s.n_rows()), (program.scratch_slots as usize).max(1));
-    execute_into(&program, &view, &Foreign::NONE, &mut scratch, Out::I64(counts))
-        .map_err(CoRefusal::Exec)?;
+    let mut scratch = Scratch::new(
+        words_for(s.n_rows()),
+        (program.scratch_slots as usize).max(1),
+    );
+    execute_into(
+        &program,
+        &view,
+        &Foreign::NONE,
+        &mut scratch,
+        Out::I64(counts),
+    )
+    .map_err(CoRefusal::Exec)?;
     cost.programs += 1;
     cost.passes += program.ops.len() + 1;
     programs.push(program);
@@ -606,7 +621,11 @@ mod tests {
             let star = native::prepare(&p, &store);
             for bag in [true, false] {
                 let want = native::bindings(&star, &p, &GENS, bag).expect("star");
-                assert_eq!(ev.bindings(&table, &p, bag), want, "{r:?} {s:?} {t:?} bag={bag}");
+                assert_eq!(
+                    ev.bindings(&table, &p, bag),
+                    want,
+                    "{r:?} {s:?} {t:?} bag={bag}"
+                );
             }
             assert_eq!(
                 ev.distinct_centers(),

From 3edd980bfbdd7af6e78bcdaa2125b2adba257ad1 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Wed, 21 Oct 2026 09:02:13 +0000
Subject: [PATCH 4/6] coaddressed: CoTable fields private, so no side over
 another domain can be installed

Codex P2 on #1658: with public fields a caller could put in a side laid out
over another JoinDomain. Keys and generations would still match, and equal
slots would then name different centers. prepare is now the only constructor;
the fields are read through accessors.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f
---
 crates/crepe-plan-probe/src/coaddressed.rs | 31 +++++++++++++++++++---
 1 file changed, 27 insertions(+), 4 deletions(-)

diff --git a/crates/crepe-plan-probe/src/coaddressed.rs b/crates/crepe-plan-probe/src/coaddressed.rs
index b2fcd3b9a..03f83c6c2 100644
--- a/crates/crepe-plan-probe/src/coaddressed.rs
+++ b/crates/crepe-plan-probe/src/coaddressed.rs
@@ -88,13 +88,18 @@ impl CoSide {
 }
 
 /// One domain and the sides laid out over it, without restrictions.
+///
+/// The fields are private: [`CoTable::prepare`] is the only way to build one,
+/// so every side is laid out over this table's domain. Equal slot numbers over
+/// two different domains name different centers, and nothing in [`eval`]
+/// could tell them apart.
 pub struct CoTable {
-    pub domain: JoinDomain,
-    pub sides: Vec<CoSide>,
+    domain: JoinDomain,
+    sides: Vec<CoSide>,
     /// The sides' keys, with `restrict: None`: a restriction is a query-time
     /// `Rel` here, not part of the layout.
-    pub keys: Vec<SideKey>,
-    pub domain_fp: u64,
+    keys: Vec<SideKey>,
+    domain_fp: u64,
 }
 
 /// `key` without its restriction.
@@ -125,6 +130,24 @@ impl CoTable {
         }
     }
 
+    pub fn domain(&self) -> &JoinDomain {
+        &self.domain
+    }
+
+    pub fn sides(&self) -> &[CoSide] {
+        &self.sides
+    }
+
+    pub fn keys(&self) -> &[SideKey] {
+        &self.keys
+    }
+
+    /// Fingerprint of the domain every side is laid out over
+    /// ([`crate::native::fingerprint`]).
+    pub fn domain_fp(&self) -> u64 {
+        self.domain_fp
+    }
+
     pub fn rows(&self) -> usize {
         self.sides.iter().map(CoSide::n_rows).sum()
     }

From a044f298da49fe977de4708ccedaae10a5a0ab42 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Wed, 21 Oct 2026 11:20:48 +0000
Subject: [PATCH 5/6] coaddressed: drop a joint check that cannot change an
 answer

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01G22yT6htkcdyXsihxxXdrg
---
 crates/crepe-plan-probe/src/coaddressed.rs | 8 +++-----
 1 file changed, 3 insertions(+), 5 deletions(-)

diff --git a/crates/crepe-plan-probe/src/coaddressed.rs b/crates/crepe-plan-probe/src/coaddressed.rs
index 9a18ecdc3..20671ae4e 100644
--- a/crates/crepe-plan-probe/src/coaddressed.rs
+++ b/crates/crepe-plan-probe/src/coaddressed.rs
@@ -488,11 +488,9 @@ impl CoEval {
             let Some(s) = u32::try_from(v).ok().and_then(|y| table.domain.slot_of(y)) else {
                 return 0;
             };
-            return if self.joint[s >> 6] >> (s & 63) & 1 == 1 {
-                at(s)
-            } else {
-                0
-            };
+            // A slot outside the joint has a side with no selected row, so
+            // its product is already 0.
+            return at(s);
         }
         let mut total = 0u64;
         for_each_set(&self.joint, table.domain.len(), |s| total += at(s));

From faa0837c3e6b9de28726dcf6746328efe51e1f5e Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Wed, 21 Oct 2026 11:23:52 +0000
Subject: [PATCH 6/6] board: U4 is a failed gate on speed under the no-ms rule;
 timings withdrawn

Every arm of examples/coaddressed.rs ran in milliseconds, the star path's
included. Under CLAUDE.md P0 (2026-10-21) that is a failed gate, so the
entry, the STATUS_BOARD row and the plan note no longer report timings. They
name where the operator leaves the mask-risc terminals: a kept row mask per
filtered side, an i64 sink of domain size, a scalar loop building the plane,
and scalar per-slot products. Correctness (parity, no foreign hop, the
disables) stands. bindings -> matching after 1d1c0b77.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VncKoyG6JkpsoqhqTGTv3f
---
 .claude/board/STATUS_BOARD.md                 |  2 +-
 .../entries/2026-10-21-u4-coaddressed-join.md | 31 +++++++++----------
 .claude/plans/d-relalg-unified-v1.md          |  2 +-
 3 files changed, 17 insertions(+), 18 deletions(-)

diff --git a/.claude/board/STATUS_BOARD.md b/.claude/board/STATUS_BOARD.md
index 0a0bc26ba..81360e227 100644
--- a/.claude/board/STATUS_BOARD.md
+++ b/.claude/board/STATUS_BOARD.md
@@ -8,7 +8,7 @@ Plan: `.claude/plans/d-relalg-unified-v1.md`. Decisions recorded there: ADMISSIO
 | **D-RU-1** | U1: JC statistical batch, distinct (selection, key, lane pair) triples over shared power-sum terminals | MEASURED (`crates/jc-batch-probe`; `entries/2026-10-20-u1-jc-batch.md`): 252 requests → 2 selection + 42 cross passes; at 1M rows 6.4× faster, 11.5× fewer rows traversed, 8.5× fewer values read, 126× fewer mask bytes; bit-identical to per-request; no R reference (R absent) | rows traversed and terminal passes below the per-request baseline; results bit-exact on integer lanes; overflow refused, not wrapped; R references |
 | **D-RU-2** | U2: `let`-bound planes on the contract IR, immutable `Plane(i)`, versioned coordinates, no DAG inside a fold | MEASURED (`contract::set_program`, `relalg::program`; `entries/2026-10-21-u2-let-planes.md`): 16 queries sharing one conjunction at 1M rows: 3.1× faster, 14 vs 59 lane reads, 2 planes (250 kB) materialized, ≤ 2 scratch slots; answers equal recompute and a row oracle; other versions refused | rows traversed and planes materialized against recompute; width test ≤ 3 slots; answers and versions unchanged |
 | **D-RU-3** | cross-rule CSE in crepe-plan-probe (256-rule benchmark, separate from the hop benchmark) | MEASURED (`crepe-plan-probe/src/cse.rs`, `examples/rule_cse.rs`; `entries/2026-10-21-u3-cross-rule-cse.md`): 256 rules, 90 % shared core, 200k centers: rows laid out 15.3 M → 6.2 M, prepare 1.78 s → 1.22 s; lets cut ops 512 → 283 but folds are ≤ 2.4 ms and `share` costs 7 ms; one table for all rules is slower than per-rule when nothing is shared, so tables are per group of rules sharing a side; answers equal crepe, ascent and the reference | answers == crepe and ascent; traversals and materializations measured |
-| **D-RU-4** | co-addressed join operator over `JoinDomain` | MEASURED (`crepe-plan-probe/src/coaddressed.rs`, `examples/coaddressed.rs`; `entries/2026-10-21-u4-coaddressed-join.md`): per-side `Rel`s at query time over one unrestricted table, `GroupReduce Count` on each side's own slot lane, `And` of slot planes; 64 restricted star queries at 200k rows: 1423 ms (re-layout per query) / 776 ms (with semijoin) → 12.4 ms + 30.4 ms one layout; answers equal the star path, the nested-loop reference and every arm; no program has a foreign hop (disable-verified) | parity with the star path; no position taken from another table's values |
+| **D-RU-4** | co-addressed join operator over `JoinDomain` | FAILED GATE — speed (no-ms rule, `CLAUDE.md` P0); correctness MEASURED (`crepe-plan-probe/src/coaddressed.rs`, `examples/coaddressed.rs`; `entries/2026-10-21-u4-coaddressed-join.md`): per-side `Rel`s at query time over one unrestricted table, `GroupReduce Count` on each side's own slot lane, `And` of slot planes; answers equal the star path, the nested-loop reference and every arm; no program has a foreign hop (disable-verified). Every arm of the benchmark, the star path's included, ran in milliseconds; open: fold the per-side `Rel` into the projection terminal | parity with the star path; no position taken from another table's values |
 | **D-RU-5** | Quack physical pins (dense vs sorted, materialize vs recompute) | Queued (after D-RU-3/4) | can-fire and can-stay-silent per pin |
 | **D-RU-6** | recursion driver over BIND tiles (256-hop benchmark) | Queued/Lab (slot assignment must be argued as an attested frozen-codebook lookup under ADMISSION = B). Lab probe MEASURED (`crepe-plan-probe/examples/walk_tile.rs`, `entries/2026-10-21-d-ru-6-tile-walk-probe.md`): tuple sets equal ascent, crepe and walk_mask; execution gather-free; binding witness probe-local, so ADMISSION B is NOT demonstrated; count-only 1.0–9.3× slower than walk_mask, end to end 1.1–2.0× slower. Block-sparse arm C MEASURED (`examples/walk_sparse.rs`, `entries/2026-10-21-d-ru-6-sparse-compute.md`): faster than the dense tile path on every fixture (0.11–0.96× count-only); against walk_mask end to end it wins only on deep low-fan-out recursion (chain(230) 0.82×, chain(700) 0.59× UNCHECKED), is near parity on sparse graphs (1.03–1.07×) and loses on dense, cyclic, clustered and mixed-length graphs (1.20–2.43×). Arm A′ (walk_mask with a dirty-frontier skip, `examples/walk_dirty.rs`, `entries/2026-10-21-d-ru-6-dirty-frontier.md`) does not remove C's chain lead; the chain sweep (two runs, distinct fan-out, like-for-like timing) confines C's end-to-end win over A′ to unbranched chains of about 192 nodes or more (3–30 %); with fan-out 2 or 4 C loses end to end at most points. Recommendation: defer the codebook attestation | least fixpoint == datafrog; ≥ 200 iterations asserted; cycles terminate. Benchmark requirement (operator, 2026-10-21; refs D-RMAP-2 §5b, #1642): tuple materialization, count-only evaluation, memory growth, address binding and value-derived reads by origin reported separately; tuple-set equality with crepe and ascent wherever they run |
 | **D-RU-7** | prefetch and staging of Arm B ranges | Queued (after D-RU-6 or a non-recursive multi-stage plan) | critical-path overlap measured; bytes read unchanged |
diff --git a/.claude/board/entries/2026-10-21-u4-coaddressed-join.md b/.claude/board/entries/2026-10-21-u4-coaddressed-join.md
index 670225870..a1670a7b3 100644
--- a/.claude/board/entries/2026-10-21-u4-coaddressed-join.md
+++ b/.claude/board/entries/2026-10-21-u4-coaddressed-join.md
@@ -1,6 +1,6 @@
 # 2026-10-21 — D-RU-4 (U4): the co-addressed join as a named operator
 
-**Status:** MEASURED (lab; `crates/crepe-plan-probe/src/coaddressed.rs`, `examples/coaddressed.rs`). Plan: `.claude/plans/d-relalg-unified-v1.md` §5.3, §6, stage table (U4). Builds on U2 (`relalg::run_with_program`) and the D-MPS-8 table (`mork_pathmap_probe::join`).
+**Status:** FAILED GATE (no-ms rule, `CLAUDE.md` P0, 2026-10-21) — correctness MEASURED, speed not (lab; `crates/crepe-plan-probe/src/coaddressed.rs`, `examples/coaddressed.rs`). Plan: `.claude/plans/d-relalg-unified-v1.md` §5.3, §6, stage table (U4). Builds on U2 (`relalg::run_with_program`) and the D-MPS-8 table (`mork_pathmap_probe::join`).
 
 ## What landed
 
@@ -11,32 +11,29 @@
 - The joint is relalg `And(Plane(i))` over the per-side slot planes. Answers:
   - `distinct_centers`, the popcount of the joint;
   - `bag_count`, a co-addressed dot product of the per-side counts (`Σ_slot Π_i count_i[slot]`) that enumerates no row;
-  - `bindings`, the tuples in `native::bindings`' shape, reading each side at the joint's slot and testing the row bits in that slot's run.
+  - `matching(pattern)`, the derivations whose head reads as a pattern: a fold over the joint, no row built. (This replaced the first version's `bindings`, a tuple builder, in `1d1c0b77` after #1657 removed `native::bindings`.)
 - `CoTable::prepare` lays out every side without its restriction. `plan_rels` turns a `StarPlan`'s restrictions into `value == k` (or `Empty` for a constant above `u32::MAX`). A restriction is therefore a query-time `Rel` and no longer part of the layout. The `Rel` can be anything relalg lowers on a `U32` lane (`==`, `!=`, `Or`, `Not`), which the star path cannot express.
 - The slot lane is built from the side's own offsets, so it is non-decreasing. A row is counted in the run it already sits in, and no operand is read at a position taken from another operand's values. `CoEval::programs` keeps every program run so a caller can check that with `native::has_foreign_hop`.
 - Refusals: a table holding other sides than the plan reads, the wrong number of `Rel`s or generations, a stale side.
 - `mork_pathmap_probe::join::Side` gains two read-only accessors, `values()` and `offsets()`.
 
-## Measured
+## Measured — failed gate
 
-Workload: 64 queries `Out(x, c, z) <- R(x, c), S(c, z), T(c, q)` for `q` in `0..64`. 200,000 rows per relation, 20,000 centers. Release build, AVX-512, best of 3.
-- Arms: C is `native::prepare` per query, which lays out all three sides again with T restricted; CS adds the semijoin; O is one `CoTable` plus `eval` per query.
+The benchmark (`examples/coaddressed.rs`: 64 queries `Out(x, c, z) <- R(x, c), S(c, z), T(c, q)`, 200,000 rows per relation, 20,000 centers) produced results in milliseconds for every arm: the per-query layout of the star path (`native::prepare`, with and without semijoin) and this operator alike. Under the `CLAUDE.md` P0 rule (2026-10-21) a millisecond result is a failed gate, not a measurement, so no timing is reported here. The first version of this entry and of #1658's body reported them; they are withdrawn.
 
-| arm | time, 64 queries | rows touched |
-|---|---|---|
-| C | 1423 ms | 25.8 M laid out |
-| CS | 776 ms | 3.9 M laid out |
-| O | 12.4 ms (+ 30.4 ms one layout) | 12.8 M read (T's value and slot lanes, 64 × 200k) |
+What stays true and is not a timing: the operator lays the table out once and reads only the rows of sides whose `Rel` is not `All` (here T's value and slot lanes, 200k rows per query); the star path lays out all sides again per query.
 
-**Findings.**
-- With the restriction moved out of the layout, O answers all 64 queries in 42.8 ms including its one layout: 33× faster than C and 18× faster than CS.
-- CS lays out fewer rows than O reads, but is still 63× slower per query. Its row count does not include the scan that filters 600k rows per query before layout, nor the sort. So "rows touched" here does not cover the same work across arms, and only the times are comparable.
-- This workload restricts one side. O reads only that side's rows per query; R and S are read as prepared. A query restricting all three sides would read all three per query.
+**Where the operator leaves the mask-risc terminals** (from the code; not attributed by measurement):
+- each filtered side keeps a full row mask (`Measure::Keep`) instead of folding into the terminal;
+- the projection's `GroupReduce Count` sinks into an `i64` buffer of domain size (`8 × domain` bytes);
+- the slot plane is built from those counts by a scalar Rust loop over the domain;
+- `bag_count` / `matching` walk the joint bit by bit and multiply side counts in scalar code per slot;
+- each filtered side runs two programs (fold, then projection).
 
 **Correctness.**
-- `parity_with_the_star_path`: for five restriction combinations (none; one; two; three; a constant above `u32::MAX`), bindings (bag and set), the bag count and distinct centers equal `native::prepare` + `native::bindings` / `distinct_centers`. At least four of the five cases have answers.
+- `parity_with_the_star_path`: for five restriction combinations (none; one; two; three; a constant above `u32::MAX`), the bag count, distinct centers and pinned per-derivation counts equal the star path (`native::matching` / `distinct_centers`). At least four of the five cases have answers.
 - `disjunctive_restrictions_match_the_reference`: `Or`, `Not` and `!=` restrictions equal a nested-loop reference, bag and set.
-- The benchmark asserts that the answers of C, CS and O are equal on every rep, and `cargo test --example coaddressed` runs the same comparison on a small fixture in CI.
+- The benchmark asserts that the answers of every arm are equal on every rep, and `cargo test --example coaddressed` runs the same comparison on a small fixture in CI.
 - **Disables:** five, each red on the test it targets:
   - a projection that scatters by the slot lane (`ScatterOrU32`, answer-correct) fails only `no_gather`, at the gather assertion;
   - row bits ignored in tuple enumeration;
@@ -47,6 +44,8 @@ Workload: 64 queries `Out(x, c, z) <- R(x, c), S(c, z), T(c, q)` for `q` in `0..
 
 ## Open
 
+- **The speed gate.** Bring the per-query path to the mask-risc terminals: fold each side's `Rel` straight into the slot projection, sink into a plane rather than an `i64` buffer, and answer counts through a terminal over the kept joint. Until that runs in nanoseconds per operation (microseconds per batch), D-RU-4 is not shipped.
+
 - The projection writes per-slot counts as `i64` for every filtered side (`8 × domain` bytes), and builds the slot plane from them in a Rust loop over the domain. A fused "count into plane" terminal would remove both; not built, not measured.
 - Each filtered side runs two programs (fold, then projection) and keeps the row mask for tuple enumeration. When only counts are asked, the fold and the projection could be one program; not measured.
 - No restriction on the center itself (a `Rel` over the domain) yet.
diff --git a/.claude/plans/d-relalg-unified-v1.md b/.claude/plans/d-relalg-unified-v1.md
index 6b8e95a3d..27efb66d0 100644
--- a/.claude/plans/d-relalg-unified-v1.md
+++ b/.claude/plans/d-relalg-unified-v1.md
@@ -298,4 +298,4 @@ One workload, run under three toggles. Each toggle changes one mechanism only.
 | U7 | prefetch and staging of Arm B ranges | U6 or a non-recursive multi-stage plan | critical-path overlap measured; bytes read unchanged |
 | U8 | Wikidata and SAP-shaped fixtures; completeness refusals | U6, U7 | refusal tests; witness carried end to end |
 
-U1 needs nothing new and comes first. **U1 MEASURED (2026-10-20, `crates/jc-batch-probe`, `entries/2026-10-20-u1-jc-batch.md`):** 252 requests reduce to 2 selection and 42 cross passes; at 1M rows the batch is 6.4× faster with 11.5× fewer rows traversed, 8.5× fewer values read and 126× fewer mask bytes, and its answers equal the per-request answers bit for bit. The R reference is replaced by JC's slice functions because R is not installed. **U2 MEASURED (2026-10-21, `contract::set_program` + `relalg::program`, `entries/2026-10-21-u2-let-planes.md`):** each shared junction subset is evaluated once into a scratch plane; for 16 queries sharing one conjunction at 1M rows the program is 3.1× faster than recompute and reads 14 lanes instead of 59, at the cost of 2 materialized planes (250 kB). Answers equal recompute and a row oracle, and a program for another version is refused. **U3 MEASURED (2026-10-21, `crepe-plan-probe/src/cse.rs` + `examples/rule_cse.rs`, `entries/2026-10-21-u3-cross-rule-cse.md`):** the §7 claim "one core join plus 256 ANDs" holds for the layout, not for the folds. At 90 % shared core the core is laid out once (15.3 M → 6.2 M rows, prepare 1.78 s → 1.22 s); the lets cut ops 512 → 283, but the folds cost ≤ 2.4 ms and the subset search 7 ms. One table for all rules is slower than per-rule tables when nothing is shared, so tables are formed per group of rules that share a side. **U4 MEASURED (2026-10-21, `crepe-plan-probe/src/coaddressed.rs` + `examples/coaddressed.rs`, `entries/2026-10-21-u4-coaddressed-join.md`):** the co-addressed join is a named operator, a conjunction of per-table `Rel`s over one `JoinDomain`. A side's restriction runs at query time over one table laid out without it, so 64 restricted queries cost 12.4 ms after one 30.4 ms layout, where laying out per query costs 1423 ms (776 ms with semijoin). Answers equal the star path; no program reads at a position taken from another operand's values. Both are accepted only on measured row traversals, materializations and total cost (§ Decisions), not on operator counts.
+U1 needs nothing new and comes first. **U1 MEASURED (2026-10-20, `crates/jc-batch-probe`, `entries/2026-10-20-u1-jc-batch.md`):** 252 requests reduce to 2 selection and 42 cross passes; at 1M rows the batch is 6.4× faster with 11.5× fewer rows traversed, 8.5× fewer values read and 126× fewer mask bytes, and its answers equal the per-request answers bit for bit. The R reference is replaced by JC's slice functions because R is not installed. **U2 MEASURED (2026-10-21, `contract::set_program` + `relalg::program`, `entries/2026-10-21-u2-let-planes.md`):** each shared junction subset is evaluated once into a scratch plane; for 16 queries sharing one conjunction at 1M rows the program is 3.1× faster than recompute and reads 14 lanes instead of 59, at the cost of 2 materialized planes (250 kB). Answers equal recompute and a row oracle, and a program for another version is refused. **U3 MEASURED (2026-10-21, `crepe-plan-probe/src/cse.rs` + `examples/rule_cse.rs`, `entries/2026-10-21-u3-cross-rule-cse.md`):** the §7 claim "one core join plus 256 ANDs" holds for the layout, not for the folds. At 90 % shared core the core is laid out once (15.3 M → 6.2 M rows, prepare 1.78 s → 1.22 s); the lets cut ops 512 → 283, but the folds cost ≤ 2.4 ms and the subset search 7 ms. One table for all rules is slower than per-rule tables when nothing is shared, so tables are formed per group of rules that share a side. **U4 FAILED GATE on speed (2026-10-21, `crepe-plan-probe/src/coaddressed.rs` + `examples/coaddressed.rs`, `entries/2026-10-21-u4-coaddressed-join.md`):** the co-addressed join is a named operator, a conjunction of per-table `Rel`s over one `JoinDomain`, and a side's restriction runs at query time over one table laid out without it. Answers equal the star path and no program reads at a position taken from another operand's values, but every arm of its benchmark ran in milliseconds, which the no-ms rule makes a failed gate; the per-side `Rel` must fold into the projection terminal. Both are accepted only on measured row traversals, materializations and total cost (§ Decisions), not on operator counts.