simd: eq_u32_via_to_mask — an equality predicate read through a foreign key

`out[i] = fk[i] < foreign.len() && foreign[fk[i]] == v`, one pass, zero
fallback on an out-of-range key. This is the join filter in factored
form: the consumer needs neither a predicate plane over the foreign table
nor a gathered mask over its own rows. Parity check 0xD40 in group 13.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GXUahz73MZxtxWcfpHp9dG

Author: claude

.claude/blackboard.md

@@ -7,6 +7,7 @@ Parity: `check_gather_scatter_group` (0xDxx) in `crates/simd-masking-parity`, ag
 `masked_group_sum_i32_via(mask, index, remap, values, out)` — the same one-pass keyed sum with the key read through a foreign-key hop (`SUM(line.amount) GROUP BY partner.country`); the indirection is fused so no remapped key lane of N is ever materialised. Fourth arm of the same parity group (0xD3x).
 Consumer: lance-graph-mask-risc `Gather`/`ScatterOr`/`GroupSum` (landing next).
 ⊘ CORRECTED (operator ruling, same day): `mask_scatter_or_u32`, `masked_group_sum_i32`, and `masked_group_sum_i32_via` no longer zero `out`/`out_words` on entry — a fold kernel accumulates into the caller's demanded sink, and the caller zeroes it once. `mask_gather_u32` is unaffected (its destination is its own output tile). Each function's doc comment now states "accumulates into `out`; the caller zeroes `out` once before the first call"; the surplus/unreferenced-slot assertions in the tail tests were reworded to "untouched, stays as the caller left it" rather than "cleared". Every unit test whose `out` buffer relied on the old zeroing was given an explicit zeroed prefill in its arrange step, and each of the three functions got two new two-sided tests (a preloaded slot/bit that survives untouched alongside the call's own contribution landing correctly; a second call with a different mask summing/unioning on top of the first) — disable-verified red-then-green by temporarily restoring the whole-buffer zero in each function in turn. The parity crate's `check_gather_scatter_group` (0xDxx) now zeroes `out`/`out2` explicitly before its from-zero reference checks and adds one accumulation check per function (0xD11/0xD22/0xD32) that preloads `out` and asserts the call adds on top.
+Fifth arm, same parity group: `eq_u32_via_to_mask(fk, foreign, v, out_words)` — the same fk lane evaluated as a join-filter PREDICATE packed into a mask rather than folded into a sum (`fk[i] < foreign.len() && foreign[fk[i]] == v`, zero-fallback at the out-of-range hop, full overwrite of tail/surplus like `mask_gather_u32`); parity check `0xD40`, disable-verified red-then-green by flipping `==` to `!=` in the kernel and back.
 
 ## 2026-09-17 (19) — G8 named: a tree-depth column (`lzcnt(bswap(x)) >> 2`) is the missing primitive for basin-local ranking; popcount is only its tie-break
 

crates/simd-masking-parity/src/lib.rs

@@ -25,22 +25,24 @@
 //! permutation/scatter family (`mask_gather_u32`/`mask_scatter_or_u32`/
 //! `masked_group_sum_i32`/`masked_group_sum_i32_via`, for
 //! lance-graph-mask-risc's Gather/ScatterOr/GroupSum verbs); `0xD3x` the
-//! fk-indirected `masked_group_sum_i32_via` (two-hop zero-fallback). `main.rs` (native / qemu) and
+//! fk-indirected `masked_group_sum_i32_via` (two-hop zero-fallback); `0xD4x`
+//! `eq_u32_via_to_mask` (the same fk lane, packed as a predicate rather than
+//! folded into a sum). `main.rs` (native / qemu) and
 //! `selfcheck()` (the wasm cdylib export, driven by `run.mjs`) both call
 //! [`run`].
 
 use ndarray::simd::{
     blend_i32, eq_i32_to_mask, eq_i32_to_mask_under, eq_u32_strided_to_mask, eq_u32_to_mask, eq_u32_to_mask_under,
-    eq_u64_to_mask, eq_u8_to_mask, ge_i32_to_mask, ge_i32_to_mask_under, ge_u64_to_mask, ge_u8_to_mask, gt_i32_to_mask,
-    gt_i32_to_mask_under, gt_u64_to_mask, gt_u8_to_mask, le_i32_to_mask, le_i32_to_mask_under, le_u64_to_mask,
-    le_u8_to_mask, lt_i32_to_mask, lt_i32_to_mask_under, lt_u64_to_mask, lt_u8_to_mask, mask_all, mask_and,
-    mask_and_assign, mask_andnot, mask_andnot_assign, mask_any, mask_gather_u32, mask_not, mask_not_assign, mask_or,
-    mask_or_assign, mask_scatter_or_u32, mask_set_range, mask_shift_morton, mask_ternlog, mask_ternlog_assign,
-    mask_xor, mask_xor_assign, masked_group_sum_i32, masked_group_sum_i32_via, masked_max_i32, masked_min_i32,
-    masked_strided_group_sum, masked_sum_i32, ne_i32_to_mask, ne_i32_to_mask_under, ne_u32_to_mask,
-    ne_u32_to_mask_under, ne_u64_to_mask, ne_u8_to_mask, ternary_match_strided_to_mask, ternary_match_u32_to_mask,
-    ternary_match_u32_to_mask_under, ternary_match_u64_to_mask, ternary_match_u64_to_mask_under, ternlog, I32x16,
-    MortonDir, U32x16, U64x8,
+    eq_u32_via_to_mask, eq_u64_to_mask, eq_u8_to_mask, ge_i32_to_mask, ge_i32_to_mask_under, ge_u64_to_mask,
+    ge_u8_to_mask, gt_i32_to_mask, gt_i32_to_mask_under, gt_u64_to_mask, gt_u8_to_mask, le_i32_to_mask,
+    le_i32_to_mask_under, le_u64_to_mask, le_u8_to_mask, lt_i32_to_mask, lt_i32_to_mask_under, lt_u64_to_mask,
+    lt_u8_to_mask, mask_all, mask_and, mask_and_assign, mask_andnot, mask_andnot_assign, mask_any, mask_gather_u32,
+    mask_not, mask_not_assign, mask_or, mask_or_assign, mask_scatter_or_u32, mask_set_range, mask_shift_morton,
+    mask_ternlog, mask_ternlog_assign, mask_xor, mask_xor_assign, masked_group_sum_i32, masked_group_sum_i32_via,
+    masked_max_i32, masked_min_i32, masked_strided_group_sum, masked_sum_i32, ne_i32_to_mask, ne_i32_to_mask_under,
+    ne_u32_to_mask, ne_u32_to_mask_under, ne_u64_to_mask, ne_u8_to_mask, ternary_match_strided_to_mask,
+    ternary_match_u32_to_mask, ternary_match_u32_to_mask_under, ternary_match_u64_to_mask,
+    ternary_match_u64_to_mask_under, ternlog, I32x16, MortonDir, U32x16, U64x8,
 };
 
 /// Number of check groups [`run`] executes (for the log line only).
@@ -1331,6 +1333,33 @@ fn check_gather_scatter_group() -> Result<(), u32> {
                 return Err(0xD32);
             }
         }
+
+        // ── eq_u32_via_to_mask ────────────────────────────────────────────
+        // A predicate evaluated through the same fk lane `masked_group_sum_i32_via`
+        // uses for its key, but packed into a bitmask rather than folded into a
+        // sum: `fk[i] < foreign.len() && foreign[fk[i]] == v`.
+        let foreign_len = 9usize;
+        let foreign: Vec<u32> = (0..foreign_len).map(|_| (rng.next() % 5) as u32).collect();
+        let v = 2u32;
+        // Every fourth key is deliberately out of range for `foreign`.
+        let fk: Vec<u32> = (0..n)
+            .map(|i| {
+                if i % 4 == 0 {
+                    (foreign_len as u64 + 6 + i as u64) as u32
+                } else {
+                    (rng.next() % foreign_len as u64) as u32
+                }
+            })
+            .collect();
+        let mut via_mask = vec![u64::MAX; out_len]; // dirty, over-long
+        eq_u32_via_to_mask(&fk, &foreign, v, &mut via_mask);
+        let want_via_mask = reference_mask(n, out_len, |i| {
+            let k = fk[i] as usize;
+            k < foreign.len() && foreign[k] == v
+        });
+        if via_mask != want_via_mask {
+            return Err(0xD40);
+        }
     }
     Ok(())
 }

src/simd.rs

@@ -789,6 +789,7 @@ pub use crate::simd_masking_ops::{
     eq_u32_strided_to_mask,
     eq_u32_to_mask,
     eq_u32_to_mask_under,
+    eq_u32_via_to_mask,
     eq_u64_to_mask,
     eq_u8_to_mask,
     ge_i32_to_mask,

src/simd_masking_ops.rs

@@ -1207,6 +1207,94 @@ pub fn masked_group_sum_i32_via(mask_words: &[u64], index: &[u32], remap: &[u32]
     }
 }
 
+/// Packs `fk[i] < foreign.len() && foreign[fk[i]] == v` into `out_words`,
+/// one bit per row `i < fk.len()`, LSB-first — a join-filter predicate
+/// evaluated **through a foreign key**, with no gathered mask and no
+/// materialised foreign predicate plane in between.
+///
+/// `WHERE partner.country = v` filtered from the `line` side, without
+/// first computing `country_of_line[i] = country[partner_id[i]]` into its
+/// own array and then comparing that: `fk = line.partner_id`,
+/// `foreign = partner.country`, and this fuses the gather-then-compare
+/// into one pass over `fk`, the same fusion [`masked_group_sum_i32_via`]
+/// applies to a fk-indirected group key rather than a fk-indirected
+/// predicate.
+///
+/// **Zero-fallback, same rule as [`mask_gather_u32`]'s out-of-range
+/// read**: `fk[i] >= foreign.len()` means row `i`'s key names no foreign
+/// row, so it does not match — not an error, not a panic, just `false` for
+/// that bit. A key that names no row is not a match, the same way an
+/// unminted classid is not a class.
+///
+/// `out_words` is **fully overwritten**, not OR-ed into; trailing bits
+/// beyond `fk.len()`, and any surplus words past `mask_words_for(fk.len())`,
+/// are written `0` — this writes exactly its own output tile and nothing
+/// past it, the same contract as [`mask_gather_u32`].
+///
+/// # Why this lives HERE
+///
+/// [`mask_gather_u32`] reads a *mask bit* through an index; this reads an
+/// *equality predicate* through an index, and belongs beside it for the
+/// same reason: a consumer hand-rolling `foreign[fk[i]] == v` in a loop is
+/// exactly the polyfill bypass the "all SIMD from `ndarray::simd`"
+/// invariant exists to prevent, and it is the gather half of
+/// [`masked_group_sum_i32_via`]'s two-hop shape applied to a predicate
+/// instead of a sum.
+///
+/// # Vectorisation, honestly
+///
+/// **Scalar, and by necessity, not oversight** — same shape as
+/// [`mask_gather_u32`]: the address read from `foreign` is `fk[i]`, a
+/// value out of another array, so there is no vector gather over
+/// individual predicate results on any of this crate's backends. Cost is
+/// `O(fk.len())`; there is no input mask to skip zero words against.
+///
+/// # Panics
+///
+/// Panics if `out_words.len() < mask_words_for(fk.len())`.
+///
+/// # Examples
+///
+/// ```
+/// use ndarray::simd::eq_u32_via_to_mask;
+///
+/// // partner 0 -> country 7, partner 1 -> country 3; row 2's fk (5) is
+/// // out of range for `foreign` and never matches.
+/// let fk = [0u32, 1, 5, 0];
+/// let foreign = [7u32, 3];
+/// let mut out = [u64::MAX]; // dirty tail must be overwritten
+/// eq_u32_via_to_mask(&fk, &foreign, 7, &mut out);
+/// // rows 0 and 3 resolve to country 7; row 1 resolves to 3; row 2 drops.
+/// assert_eq!(out[0], 0b1001);
+/// ```
+#[inline]
+pub fn eq_u32_via_to_mask(fk: &[u32], foreign: &[u32], v: u32, out_words: &mut [u64]) {
+    let n = fk.len();
+    let words = mask_words_for(n);
+    assert!(
+        out_words.len() >= words,
+        "eq_u32_via_to_mask: out_words.len()={} < required {}",
+        out_words.len(),
+        words
+    );
+
+    for (w, out_word) in out_words.iter_mut().enumerate().take(words) {
+        let base = w * 64;
+        let live = (n - base).min(64);
+        let mut acc = 0u64;
+        for lane in 0..live {
+            let key = fk[base + lane] as usize;
+            if key < foreign.len() && foreign[key] == v {
+                acc |= 1u64 << lane;
+            }
+        }
+        *out_word = acc;
+    }
+    for w in out_words.iter_mut().skip(words) {
+        *w = 0;
+    }
+}
+
 // ────────────────────────────────────────────────────────────────────────
 // The closed comparison family + mask complement/xor/any + care-masked
 // register match + masked min/max + blend (the DuckDB-vector-execution set,
@@ -4582,6 +4670,99 @@ mod tests {
         masked_group_sum_i32_via(&mask, &index, &remap, &values, &mut out);
     }
 
+    // ── eq_u32_via_to_mask ──
+
+    fn naive_eq_via(fk: &[u32], foreign: &[u32], v: u32) -> Vec<bool> {
+        fk.iter()
+            .map(|&k| {
+                let k = k as usize;
+                k < foreign.len() && foreign[k] == v
+            })
+            .collect()
+    }
+
+    #[test]
+    fn eq_u32_via_to_mask_matches_naive_reference_across_the_tail() {
+        for &n in &[0usize, 1, 63, 64, 65, 130, 1000] {
+            let mut seed = 0xACE1_2345_6789_BEEFu64;
+            let foreign_len = 17usize;
+            let foreign: Vec<u32> = (0..foreign_len)
+                .map(|_| (splitmix(&mut seed) % 5) as u32)
+                .collect();
+            let v = 2u32;
+            // A third of keys are deliberately out of range; the rest hit
+            // `foreign`, so both the match and no-match arms are genuinely
+            // exercised (not merely plausible).
+            let fk: Vec<u32> = (0..n)
+                .map(|i| {
+                    if i % 3 == 0 {
+                        (foreign_len as u64 + 3 + i as u64) as u32
+                    } else {
+                        (splitmix(&mut seed) % foreign_len as u64) as u32
+                    }
+                })
+                .collect();
+            let want_bits = naive_eq_via(&fk, &foreign, v);
+            let want = bits_to_words(&want_bits);
+            let out_words = n.div_ceil(64).max(1);
+            let mut out = vec![0xFFFF_FFFF_FFFF_FFFFu64; out_words + 1]; // dirty, over-long
+            eq_u32_via_to_mask(&fk, &foreign, v, &mut out);
+            assert_eq!(&out[..want.len()], &want[..], "eq_u32_via_to_mask mismatch at n={n}");
+            assert_eq!(out[out_words], 0, "surplus word must be cleared at n={n}");
+            if n >= 10 {
+                // Anti-vacuity (skipped at tiny n, where a single fixture
+                // cannot be relied on to hit both arms): this fixture must
+                // actually contain both a match and a non-match, or the
+                // comparison above proves nothing about which arm is
+                // exercised.
+                assert!(want_bits.iter().any(|&b| b), "fixture at n={n} has no matching row at all");
+                assert!(want_bits.iter().any(|&b| !b), "fixture at n={n} has no non-matching row at all");
+            }
+        }
+    }
+
+    #[test]
+    fn eq_u32_via_to_mask_out_of_range_fk_never_matches_even_when_foreign_0_equals_v() {
+        let fk = [5u32, 10, 100, u32::MAX];
+        let foreign = [7u32]; // foreign[0] == v, but every fk above is >= 1
+        let mut out = [0u64; 1];
+        eq_u32_via_to_mask(&fk, &foreign, 7, &mut out);
+        assert_eq!(out[0], 0, "every fk names no row in `foreign`, so nothing may match");
+    }
+
+    #[test]
+    fn eq_u32_via_to_mask_tail_and_surplus_words_are_cleared_not_left_dirty() {
+        let fk = [0u32, 0, 0, 0, 0]; // n = 5, one word; foreign[0] == v for all
+        let foreign = [9u32];
+        let mut out = [0xFFFF_FFFF_FFFF_FFFFu64; 3]; // one live word + two surplus
+        eq_u32_via_to_mask(&fk, &foreign, 9, &mut out);
+        assert_eq!(out[0], 0b11111, "the five live rows should be set");
+        assert_eq!(out[0] & !0b11111, 0, "bits past n=5 in the live word must be zero, not dirty");
+        assert_eq!(out[1], 0, "surplus word 1 must be cleared");
+        assert_eq!(out[2], 0, "surplus word 2 must be cleared");
+    }
+
+    #[test]
+    fn eq_u32_via_to_mask_empty_foreign_yields_all_zero_mask_for_nonempty_fk() {
+        // Every fk names a row, but `foreign` is empty, so every key is out
+        // of range: the whole mask must be false, not a panic and not a
+        // vacuous "unreachable, so anything goes".
+        let fk = [0u32, 1, 2, 3, 4, 5, 6, 7];
+        let foreign: [u32; 0] = [];
+        let mut out = [0xFFFF_FFFF_FFFF_FFFFu64; 1];
+        eq_u32_via_to_mask(&fk, &foreign, 0, &mut out);
+        assert_eq!(out[0], 0, "an empty foreign table matches nothing");
+    }
+
+    #[test]
+    #[should_panic(expected = "out_words.len()")]
+    fn eq_u32_via_to_mask_rejects_short_out_buffer() {
+        let fk = vec![0u32; 65]; // needs 2 words
+        let foreign = [0u32];
+        let mut out = [0u64; 1];
+        eq_u32_via_to_mask(&fk, &foreign, 0, &mut out);
+    }
+
     // ── 2026-09-13 additions: the closed comparison family, complement/xor/
     //    any/all, care-masked register match, masked min/max, blend ──