fix: OP_IN null-aware row and set handling

Regression: exec_in treated typed nulls as ordinary data values.
On a column containing 0Nl (i64 null), the null-sentinel integer
read back from the column didn't match the filter set `[1]`, so
`not-in k [1]` happily returned the null rows as "true" — leaking
4 rows instead of the 2 that should pass.

Fix:
 - Read the column's and set's HAS_NULLS flag + null bitmap.
 - Before probing, drop any null elements from the set's probe
   buffer — a non-null col value must never match a null set cell
   via sentinel-collision.
 - In the per-row loop, if the col row is null (atom or bitmap),
   emit 0 unconditionally.  Both `in` and `not-in` are defined
   to be false for null inputs (SQL-style UNKNOWN → false in
   boolean context).
 - Also handle atom null via RAY_ATOM_IS_NULL for the degenerate
   scalar-col case.

Regression test /eval/select_where_in_nulls asserts both `in` and
`not-in` correctly exclude null rows from a 5-row column with two
nulls.

621/621 tests pass in debug and release.

Author: singaraiona

src/ops/exec.c

@@ -606,6 +606,16 @@ static ray_t* exec_in(ray_graph_t* g, ray_op_t* op, ray_t* col, ray_t* set) {
     out->len = col_len;
     uint8_t* ob = (uint8_t*)ray_data(out);
 
+    /* Null-aware: null rows in the column never pass either `in` or
+     * `not-in`.  Mirrors SQL-style semantics where NULL IN (…) and
+     * NULL NOT IN (…) both yield UNKNOWN / false in a boolean
+     * context.  Also skip null elements when building the probe
+     * buffer so a non-null col row doesn't accidentally match the
+     * sentinel value of a null set element. */
+    bool col_has_nulls = !ray_is_atom(col) && (col->attrs & RAY_ATTR_HAS_NULLS);
+    bool col_atom_null = ray_is_atom(col) && RAY_ATOM_IS_NULL(col);
+    bool set_has_nulls = !ray_is_atom(set) && (set->attrs & RAY_ATTR_HAS_NULLS);
+
     #define READ_I64(dst, vec, type, idx) do {                             \
         const void* _d = ray_data(vec);                                    \
         switch (type) {                                                    \
@@ -636,6 +646,10 @@ static ray_t* exec_in(ray_graph_t* g, ray_op_t* op, ray_t* col, ray_t* set) {
         }                                                                  \
     } while (0)
 
+    /* Compact probe buffer: drop null set elements up front so the
+     * inner loop doesn't special-case them. */
+    int64_t sv_len = 0;
+
     if (use_double) {
         double sv_stack[32];
         ray_t* sv_hdr = NULL;
@@ -646,21 +660,28 @@ static ray_t* exec_in(ray_graph_t* g, ray_op_t* op, ray_t* col, ray_t* set) {
             sv = (double*)ray_data(sv_hdr);
         }
         if (ray_is_atom(set)) {
-            if (st == RAY_F64)  sv[0] = set->f64;
-            else                sv[0] = (double)set->i64;
+            if (!RAY_ATOM_IS_NULL(set)) {
+                sv[0] = (st == RAY_F64) ? set->f64 : (double)set->i64;
+                sv_len = 1;
+            }
         } else {
-            for (int64_t i = 0; i < set_len; i++) READ_F64(sv[i], set, st, i);
+            for (int64_t i = 0; i < set_len; i++) {
+                if (set_has_nulls && ray_vec_is_null(set, i)) continue;
+                READ_F64(sv[sv_len], set, st, i);
+                sv_len++;
+            }
         }
         for (int64_t i = 0; i < col_len; i++) {
+            /* Null col rows never pass either in or not-in. */
+            bool row_null = col_atom_null ||
+                            (col_has_nulls && !ray_is_atom(col) &&
+                             ray_vec_is_null(col, i));
+            if (row_null) { ob[i] = 0; continue; }
             double cv;
-            if (ray_is_atom(col)) {
-                if (ct == RAY_F64)  cv = col->f64;
-                else                cv = (double)col->i64;
-            } else {
-                READ_F64(cv, col, ct, i);
-            }
+            if (ray_is_atom(col)) cv = (ct == RAY_F64) ? col->f64 : (double)col->i64;
+            else READ_F64(cv, col, ct, i);
             int found = 0;
-            for (int64_t j = 0; j < set_len; j++) {
+            for (int64_t j = 0; j < sv_len; j++) {
                 if (cv == sv[j]) { found = 1; break; }
             }
             ob[i] = (uint8_t)(found ^ negate);
@@ -676,15 +697,26 @@ static ray_t* exec_in(ray_graph_t* g, ray_op_t* op, ray_t* col, ray_t* set) {
             if (!sv_hdr) { ray_release(out); return ray_error("oom", NULL); }
             sv = (int64_t*)ray_data(sv_hdr);
         }
-        if (ray_is_atom(set)) sv[0] = set->i64;
-        else for (int64_t i = 0; i < set_len; i++) READ_I64(sv[i], set, st, i);
+        if (ray_is_atom(set)) {
+            if (!RAY_ATOM_IS_NULL(set)) { sv[0] = set->i64; sv_len = 1; }
+        } else {
+            for (int64_t i = 0; i < set_len; i++) {
+                if (set_has_nulls && ray_vec_is_null(set, i)) continue;
+                READ_I64(sv[sv_len], set, st, i);
+                sv_len++;
+            }
+        }
 
         for (int64_t i = 0; i < col_len; i++) {
+            bool row_null = col_atom_null ||
+                            (col_has_nulls && !ray_is_atom(col) &&
+                             ray_vec_is_null(col, i));
+            if (row_null) { ob[i] = 0; continue; }
             int64_t cv;
             if (ray_is_atom(col)) cv = col->i64;
             else READ_I64(cv, col, ct, i);
             int found = 0;
-            for (int64_t j = 0; j < set_len; j++) {
+            for (int64_t j = 0; j < sv_len; j++) {
                 if (cv == sv[j]) { found = 1; break; }
             }
             ob[i] = (uint8_t)(found ^ negate);

test/test_lang.c

@@ -1008,6 +1008,38 @@ static MunitResult test_eval_select_where_in_sym(const void* params, void* fixtu
     return MUNIT_OK;
 }
 
+/* ---- Test: OP_IN must not leak null rows through ----
+ * Regression: exec_in originally treated the raw null-sentinel
+ * value as a normal data value.  That meant `(not-in k [1])`
+ * on a column containing 0Nl nulls returned the null rows as
+ * "true" (because null-sentinel != 1), leaking them through.
+ * Null rows must never pass either `in` or `not-in`. */
+static MunitResult test_eval_select_where_in_nulls(const void* params, void* fixture) {
+    (void)params; (void)fixture;
+
+    /* not-in with null rows: source has [1 0Nl 3 0Nl 5], filter
+     * `not-in [1]` should return only 3 and 5 — not the two nulls. */
+    ray_t* r1 = ray_eval_str(
+        "(do (set t (table ['k] (list [1 0Nl 3 0Nl 5]))) "
+        "(select {from: t where: (not-in k [1])}))");
+    munit_assert_ptr_not_null(r1);
+    munit_assert_false(RAY_IS_ERR(r1));
+    munit_assert_int(ray_table_nrows(r1), ==, 2);
+    ray_release(r1);
+
+    /* in with null rows: source has [1 0Nl 3 0Nl 5], filter
+     * `in [1 3]` should return 1 and 3 — not the nulls. */
+    ray_t* r2 = ray_eval_str(
+        "(do (set t (table ['k] (list [1 0Nl 3 0Nl 5]))) "
+        "(select {from: t where: (in k [1 3])}))");
+    munit_assert_ptr_not_null(r2);
+    munit_assert_false(RAY_IS_ERR(r2));
+    munit_assert_int(ray_table_nrows(r2), ==, 2);
+    ray_release(r2);
+
+    return MUNIT_OK;
+}
+
 /* ---- Test: `in` with mixed numeric types (F64 col vs I64 set) ----
  * Regression: exec_in originally compared bit patterns which made
  * `(in price_f64 [1 2 3])` match nothing.  Now we promote to double
@@ -2761,6 +2793,7 @@ static MunitTest lang_tests[] = {
     { "/eval/select_where_in_sym",   test_eval_select_where_in_sym,   lang_setup, lang_teardown, 0, NULL },
     { "/eval/select_where_in_i64",   test_eval_select_where_in_i64,   lang_setup, lang_teardown, 0, NULL },
     { "/eval/select_where_in_mixed_numeric", test_eval_select_where_in_mixed_numeric, lang_setup, lang_teardown, 0, NULL },
+    { "/eval/select_where_in_nulls", test_eval_select_where_in_nulls, lang_setup, lang_teardown, 0, NULL },
     { "/eval/select_by_where_filters", test_eval_select_by_where_filters, lang_setup, lang_teardown, 0, NULL },
     { "/eval/select_by_where_in",    test_eval_select_by_where_in,    lang_setup, lang_teardown, 0, NULL },
     { "/eval/select_if",             test_eval_select_if,             lang_setup, lang_teardown, 0, NULL },