fix(in): admit the typed kernel unless both sides carry a null

src/ops/collection.c

@@ -1486,10 +1486,18 @@ ray_t* ray_in_fn(ray_t* val, ray_t* vec) {
              * on a 351k-row column (#593).  vec.h says as much where it
              * defines the two — "paths requiring null-free data use has_nulls
              * below".  The exact check costs one pass and is not on a per-row
-             * path; a null-bearing operand still falls through, because the
-             * kernel's null-matches-nothing semantics differ from this path's
-             * null-equals-null. */
-            if (!ray_vec_has_nulls(val) && !ray_vec_has_nulls(vec)) {
+             * path.
+             *
+             * Only BOTH sides null-bearing must fall through.  The kernel's
+             * null-matches-nothing and this path's null-equals-null disagree
+             * on exactly one question — does a null row match a null set
+             * element — and that needs a null on each side.  A null row
+             * against a null-free set, or a null set element against a
+             * null-free column, is false under both.  Requiring both sides
+             * null-free sent any nullable column to the hashset: one null
+             * row cost ~25x.  The set is tested first: it is usually the
+             * small side, and a null-free set skips the column scan. */
+            if (!(ray_vec_has_nulls(vec) && ray_vec_has_nulls(val))) {
                 ray_t* fast = ray_in_vec_exec(val, vec, false);
                 if (fast) return fast;
             }

src/ops/exec.c

@@ -27,6 +27,7 @@
 #include "ops/rowsel.h"
 #include "ops/fused_group.h"
 #include "ops/idxop.h"
+#include "ops/hash.h"
 #include "mem/heap.h"
 #include "mem/sys.h"
 #include "core/qstats.h"   /* per-worker parallelism stats for profile spans */
@@ -611,13 +612,68 @@ typedef struct {
      * per-row linear set scan into one byte load — any set size, width-
      * specialized, vectorizable.  NULL when not applicable. */
     const uint8_t* symlut;
+    /* Needle hash, built when the set outgrows the SIMD small-set path
+     * (IN_SIMD_SET): open addressing over the live probe values, so a
+     * row costs one probe instead of a scan of the whole set.  hti holds
+     * int64 values (empty = INT64_MIN; a real INT64_MIN needle sets
+     * ht_has_min), htf holds f64 bit patterns (empty = IN_HTF_EMPTY, a
+     * NaN, which is never inserted).  Both NULL = linear scan. */
+    const int64_t*  hti;
+    const uint64_t* htf;
+    int64_t        ht_mask;
+    bool           ht_has_min;
     bool           col_has_nulls;
     bool           col_atom_null;
     bool           col_is_atom;
     bool           use_double;
     bool           negate;
 } in_worker_ctx_t;
 
+/* Sets up to this many live elements take the unrolled SIMD compare in
+ * exec_in_worker; larger ones get the needle hash (in_build_worker_ctx). */
+#define IN_SIMD_SET 8
+#define IN_HTF_EMPTY UINT64_C(0x7ff8000000000000)
+
+/* -0.0 and +0.0 compare equal, so they must share a slot. */
+static inline uint64_t in_f64_key(double v) {
+    uint64_t b;
+    memcpy(&b, &v, sizeof(b));
+    return b == UINT64_C(0x8000000000000000) ? 0 : b;
+}
+
+static inline int in_set_has_i(const in_worker_ctx_t* c, int64_t v) {
+    if (c->hti) {
+        if (v == INT64_MIN) return c->ht_has_min;
+        int64_t s = (int64_t)(ray_hash_i64(v) & (uint64_t)c->ht_mask);
+        for (;;) {
+            int64_t k = c->hti[s];
+            if (k == v) return 1;
+            if (k == INT64_MIN) return 0;
+            s = (s + 1) & c->ht_mask;
+        }
+    }
+    for (int64_t j = 0; j < c->sv_len; j++)
+        if (v == c->svi[j]) return 1;
+    return 0;
+}
+
+static inline int in_set_has_f(const in_worker_ctx_t* c, double v) {
+    if (c->htf) {
+        if (v != v) return 0;  /* NaN equals nothing, as in the scan */
+        uint64_t b = in_f64_key(v);
+        int64_t s = (int64_t)(ray_hash_i64((int64_t)b) & (uint64_t)c->ht_mask);
+        for (;;) {
+            uint64_t k = c->htf[s];
+            if (k == b) return 1;
+            if (k == IN_HTF_EMPTY) return 0;
+            s = (s + 1) & c->ht_mask;
+        }
+    }
+    for (int64_t j = 0; j < c->sv_len; j++)
+        if (v == c->svf[j]) return 1;
+    return 0;
+}
+
 static void exec_in_worker(void* vctx, uint32_t worker_id,
                            int64_t start, int64_t end) {
     (void)worker_id;
@@ -705,7 +761,7 @@ static void exec_in_worker(void* vctx, uint32_t worker_id,
         return;
     }
 
-    if (!c->col_is_atom && !c->use_double && sv_len >= 1 && sv_len <= 8) {
+    if (!c->col_is_atom && !c->use_double && sv_len >= 1 && sv_len <= IN_SIMD_SET) {
         const int64_t* svi = c->svi;
         uint8_t neg = (uint8_t)negate;
         #define IN_FAST(CTYPE, FITS, SENT, HASNULL) do {                     \
@@ -765,7 +821,6 @@ static void exec_in_worker(void* vctx, uint32_t worker_id,
     }
 
     if (c->use_double) {
-        const double* svf = c->svf;
         if (c->col_atom_null) {
             /* All elements are null — fill zeros */
             for (int64_t i = start; i < end; i++) ob[i - ob_base] = 0;
@@ -775,24 +830,17 @@ static void exec_in_worker(void* vctx, uint32_t worker_id,
                 double cv;
                 if (c->col_is_atom) cv = (ct == RAY_F64) ? col->f64 : (double)col->i64;
                 else IN_READ_F64(cv, i);
-                int found = 0;
-                for (int64_t j = 0; j < sv_len; j++)
-                    if (cv == svf[j]) { found = 1; break; }
-                ob[i - ob_base] = (uint8_t)(found ^ negate);
+                ob[i - ob_base] = (uint8_t)(in_set_has_f(c, cv) ^ negate);
             }
         } else {
             for (int64_t i = start; i < end; i++) {
                 double cv;
                 if (c->col_is_atom) cv = (ct == RAY_F64) ? col->f64 : (double)col->i64;
                 else IN_READ_F64(cv, i);
-                int found = 0;
-                for (int64_t j = 0; j < sv_len; j++)
-                    if (cv == svf[j]) { found = 1; break; }
-                ob[i - ob_base] = (uint8_t)(found ^ negate);
+                ob[i - ob_base] = (uint8_t)(in_set_has_f(c, cv) ^ negate);
             }
         }
     } else {
-        const int64_t* svi = c->svi;
         if (c->col_atom_null) {
             for (int64_t i = start; i < end; i++) ob[i - ob_base] = 0;
         } else if (vec_has_nulls) {
@@ -801,20 +849,14 @@ static void exec_in_worker(void* vctx, uint32_t worker_id,
                 int64_t cv;
                 if (c->col_is_atom) cv = col->i64;
                 else IN_READ_I64(cv, i);
-                int found = 0;
-                for (int64_t j = 0; j < sv_len; j++)
-                    if (cv == svi[j]) { found = 1; break; }
-                ob[i - ob_base] = (uint8_t)(found ^ negate);
+                ob[i - ob_base] = (uint8_t)(in_set_has_i(c, cv) ^ negate);
             }
         } else {
             for (int64_t i = start; i < end; i++) {
                 int64_t cv;
                 if (c->col_is_atom) cv = col->i64;
                 else IN_READ_I64(cv, i);
-                int found = 0;
-                for (int64_t j = 0; j < sv_len; j++)
-                    if (cv == svi[j]) { found = 1; break; }
-                ob[i - ob_base] = (uint8_t)(found ^ negate);
+                ob[i - ob_base] = (uint8_t)(in_set_has_i(c, cv) ^ negate);
             }
         }
     }
@@ -885,6 +927,17 @@ static in_ctx_status_t in_build_worker_ctx(ray_t* col, ray_t* set, bool negate,
     int col_class = CLASSIFY(ct);
     int set_class = CLASSIFY(st);
 
+    /* A temporal value equals only its own type: atom_eq — and so the
+     * hashset behind find/except and the null-bearing `in` — never matches
+     * a DATE to an int or to a TIMESTAMP.  Comparing raw payloads here
+     * matched day 0 to 0i, and a DATE's day count to a TIMESTAMP's
+     * nanoseconds.  Empty probe, exactly like SYM vs non-SYM below. */
+    #define IS_TEMPORAL(t) \
+        ((t) == RAY_DATE || (t) == RAY_TIME || (t) == RAY_TIMESTAMP)
+    if ((IS_TEMPORAL(ct) || IS_TEMPORAL(st)) && ct != st)
+        set_len = 0;
+    #undef IS_TEMPORAL
+
     /* Mixed SYM vs non-SYM → treat as an empty probe.  A SYM set
      * containing resolved sym IDs has no meaning when compared to a
      * raw integer column, so nothing can match — but we still drop
@@ -1043,10 +1096,52 @@ static in_ctx_status_t in_build_worker_ctx(ray_t* col, ray_t* set, bool negate,
         }
     }
 
+    /* Large set on a non-LUT column: hash the live needles once so each row
+     * costs one probe.  The linear scan was O(rows × needles) — a nullable
+     * I64 column against 100k needles took over a second.  An allocation
+     * failure keeps the scan: slower, same answer. */
+    const int64_t* hti = NULL;
+    const uint64_t* htf = NULL;
+    int64_t ht_mask = 0;
+    bool ht_has_min = false;
+    if (!ray_is_atom(col) && !symlut && sv_len > IN_SIMD_SET) {
+        int64_t cap = 16;
+        while (cap < sv_len * 2) cap <<= 1;
+        ray_t* hh = ray_alloc((size_t)cap * sizeof(int64_t));
+        if (hh) {
+            ht_mask = cap - 1;
+            if (use_double) {
+                uint64_t* t = (uint64_t*)ray_data(hh);
+                for (int64_t k = 0; k < cap; k++) t[k] = IN_HTF_EMPTY;
+                for (int64_t j = 0; j < sv_len; j++) {
+                    if (svf[j] != svf[j]) continue;  /* NaN matches nothing */
+                    uint64_t b = in_f64_key(svf[j]);
+                    int64_t sl = (int64_t)(ray_hash_i64((int64_t)b) & (uint64_t)ht_mask);
+                    while (t[sl] != IN_HTF_EMPTY && t[sl] != b) sl = (sl + 1) & ht_mask;
+                    t[sl] = b;
+                }
+                htf = t;
+            } else {
+                int64_t* t = (int64_t*)ray_data(hh);
+                for (int64_t k = 0; k < cap; k++) t[k] = INT64_MIN;
+                for (int64_t j = 0; j < sv_len; j++) {
+                    int64_t v = svi[j];
+                    if (v == INT64_MIN) { ht_has_min = true; continue; }
+                    int64_t sl = (int64_t)(ray_hash_i64(v) & (uint64_t)ht_mask);
+                    while (t[sl] != INT64_MIN && t[sl] != v) sl = (sl + 1) & ht_mask;
+                    t[sl] = v;
+                }
+                hti = t;
+            }
+            *lut_hdr_out = hh;  /* freed by every caller with the LUT */
+        }
+    }
+
     *out_ctx = (in_worker_ctx_t){
         .col = col,
         .svf = svf, .svi = svi, .sv_len = sv_len,
         .symlut = symlut,
+        .hti = hti, .htf = htf, .ht_mask = ht_mask, .ht_has_min = ht_has_min,
         .ob = NULL, .ob_base = 0, .ct = ct,
         .col_has_nulls = col_has_nulls,
         .col_atom_null = col_atom_null,

test/rfl/collection/in.rfl

@@ -86,9 +86,10 @@
 (in (list) [1h 2h]) -- (list)
 
 ;; ========== TEXT (SYM/STR) NULL SEMANTICS (#593) ==========
-;; `in` admits the typed verdict-LUT kernel only when BOTH sides are exactly
-;; null-free, because that kernel uses null-matches-nothing semantics while
-;; this path is null-equals-null.  The gate used to ask
+;; `in` admits the typed verdict-LUT kernel unless BOTH sides carry a null:
+;; the kernel uses null-matches-nothing semantics while the hashset path is
+;; null-equals-null, and the two disagree only on a null row against a null
+;; set element.  The gate used to ask
 ;; ray_vec_may_have_nulls, which is unconditionally true for SYM and STR
 ;; (their null is a payload value, not an attribute bit) — so the kernel was
 ;; unreachable for text columns and every symbol `in` fell through to the
@@ -102,8 +103,8 @@
 ;; null in the column, non-null needle — the null matches nothing else
 (in (as 'SYM ["a" "" "b"]) (as 'SYM ["a"])) -- [true false false]
 
-;; null only in the NEEDLES: the column is null-free but the gate must still
-;; reject, since a null needle needs null-equals-null against nothing
+;; null only in the NEEDLES: a null-free column has no row for the null
+;; needle to equal, so both semantics agree and the kernel is admitted
 (in (as 'SYM ["a" "b"]) (as 'SYM ["" "a"])) -- [true false]
 
 ;; nulls on both sides
@@ -113,6 +114,38 @@
 (in (as 'SYM ["a" "b" "c"]) (as 'SYM ["b" "c"])) -- [false true true]
 (in ["a" "b" "c"] ["b" "c"]) -- [false true true]
 
+;; ========== ONE-SIDED NULLS REACH THE KERNEL (#593) ==========
+;; A nullable column against null-free needles (or the reverse) is admitted:
+;; a null row can only match a null needle.  Before, one null row sent the
+;; whole column to the hashset, whose i64/f64 hashes disagree across numeric
+;; types — so the mixed-type rows below also answered wrongly.
+
+;; nullable column, null-free needles, every null-bearing width
+(in [1 0Nl 3] [1 3]) -- [true false true]
+(in [1i 0Ni 3i] [1 3]) -- [true false true]
+(in [1h 0Nh 3h] [1h]) -- [true false false]
+(in [1.0 0Nf 3.0] [1.0 3.0]) -- [true false true]
+(in [2024.01.01 0Nd] [2024.01.01]) -- [true false]
+(in ["a" "" "b"] ["a"]) -- [true false false]
+
+;; the i32 null sentinel is INT32_MIN: a non-null needle of that value must
+;; not match the null row
+(in [1i 0Ni 3i] [-2147483648 3]) -- [false false true]
+
+;; null-free column, null needles: the null needle matches nothing
+(in [1 2 3] [0Nl 3]) -- [false false true]
+(in [1.0 2.0 3.0] [0Nf 3.0]) -- [false false true]
+(in [2024.01.01 2024.01.02] [0Nd 2024.01.02]) -- [false true]
+(in ["a" "b"] ["" "a"]) -- [true false]
+
+;; mixed int/float with a null on one side — was [true false false] and
+;; [false false false] on the hashset path
+(in [1.0 0Nf 3.0] [1 3]) -- [true false true]
+(in [1 0Nl 3] [1.0 3.0]) -- [true false true]
+
+;; nulls on both sides still take the null-equals-null path
+(in [1 0Nl 3] [0Nl 3]) -- [false true true]
+
 ;; empty needle set over a null-free symbol column
 (in (as 'SYM ["a" "b"]) (as 'SYM [])) -- [false false]
 
@@ -123,3 +156,64 @@
 
 ;; duplicate needles must not double-count or change the verdict
 (in (as 'SYM ["a" "b"]) (as 'SYM ["a" "a" "a"])) -- [true false]
+
+;; ========== TEMPORAL EQUALS ONLY ITS OWN TYPE (#593 review) ==========
+;; atom_eq never matches a DATE to an int or to a TIMESTAMP, and neither do
+;; find / except or the hashset `in`.  The typed kernel compared raw
+;; payloads — day 0 matched 0i, a DATE's day count matched a TIMESTAMP's
+;; nanoseconds — so widening its gate made the answer depend on where the
+;; nulls were.  Every shape below answers the same with nulls on either
+;; side, none, or (for the hashset) both.
+
+;; temporal column, int needles: nothing matches
+(in [2000.01.01 2000.01.03] [0i 2i]) -- [false false]
+(in [2000.01.01 0Nd 2000.01.03] [0i 2i]) -- [false false false]
+(in [2000.01.01 2000.01.03] [0Ni 2i]) -- [false false]
+(in [2000.01.01 0Nd] [0]) -- [false false]
+(in [00:00:01.000 0Nt] [1000i]) -- [false false]
+(in [2000.01.01D00:00:00.000000000 0Np] [0]) -- [false false]
+
+;; int column, temporal needles: nothing matches
+(in [0i 2i] [2000.01.01 2000.01.03]) -- [false false]
+(in [0 0Nl] [2000.01.01]) -- [false false]
+(in [0 1] [2000.01.01 0Nd]) -- [false false]
+
+;; distinct temporal types never match — not even by raw payload
+(in [2000.01.02] [2000.01.01D00:00:00.000000001]) -- [false]
+(in [2000.01.02 0Nd] [2000.01.02D00:00:00.000000000]) -- [false false]
+(in [2000.01.01D00:00:00.000000000 0Np] [2000.01.01]) -- [false false]
+(in [00:00:00.001] [2000.01.01D00:00:00.000000001]) -- [false]
+
+;; same with nulls on both sides: only null equals null
+(in [2000.01.01 0Nd] [0Ni 0i]) -- [false true]
+
+;; same-type temporal still matches, with or without nulls
+(in [2000.01.01 0Nd 2000.01.03] [2000.01.03]) -- [false false true]
+(in [2000.01.01 2000.01.03] [2000.01.01 0Nd]) -- [true false]
+
+;; the fused where: and find agree with the bare in
+(count (select {from: (table [d] (list [2000.01.01 2000.01.03 0Nd])) where: (in d [0i 2i])})) -- 0
+(find [2000.01.01 2000.01.03] [0i]) -- [0Nl]
+
+;; ========== LARGE NEEDLE SETS HASH INSIDE THE KERNEL ==========
+;; Past the SIMD small-set size (8) the kernel hashes the needles instead of
+;; scanning them per row, so a nullable column against thousands of needles
+;; stays O(rows + needles).  Pins the boundary and the table's edge values.
+
+;; 8 needles (SIMD path) and 9 (hash path) answer alike
+(in [1 2 3 0Nl 20] [1 3 5 7 9 11 13 15]) -- [true false true false false]
+(in [1 2 3 0Nl 20] [1 3 5 7 9 11 13 15 20]) -- [true false true false true]
+(in [1.0 2.0 0Nf 20.0] [1.0 3.0 5.0 7.0 9.0 11.0 13.0 15.0 20.0]) -- [true false false true]
+
+;; -0.0 and +0.0 share a slot; NaN / null never match
+(in [-0.0 0.0 1.5] (as 'F64 (til 20))) -- [true true false]
+(in [0Nf 2.5] (concat (as 'F64 (til 20)) [2.5])) -- [false true]
+
+;; duplicates in a large set, and a large set against a nullable column
+(sum (in (concat (til 1000) [0Nl]) (take (til 100) 5000))) -- 100
+(sum (in (concat (til 1000) [0Nl]) (* 2 (til 1000)))) -- 500
+(sum (in (concat (as 'F64 (til 1000)) [0Nf]) (as 'F64 (* 2 (til 1000))))) -- 500
+
+;; a large set of narrow ints against a narrow column
+(sum (in (as 'I32 (til 1000)) (as 'I32 (* 3 (til 100))))) -- 100
+(sum (in (as 'I16 (til 1000)) (as 'I16 (* 3 (til 100))))) -- 100