perf: parallelize serial stages around group-by; unify binary-agg null semantics

src/core/pool.c

@@ -156,7 +156,9 @@ static ray_err_t ray_pool_create_impl(ray_pool_t* pool, uint32_t n_workers,
     atomic_store_explicit(&pool->shutdown, 0, memory_order_relaxed);
 
     /* Allocate task ring */
-    pool->task_cap = 1024;
+    _Static_assert(RAY_POOL_INIT_TASKS <= RAY_POOL_MAX_TASKS,
+                   "initial ring capacity must fit the ring ceiling");
+    pool->task_cap = RAY_POOL_INIT_TASKS;
     if (pool->task_cap < MAX_RING_CAP) {
         /* Will grow if needed in dispatch */
     }

src/core/pool.h

@@ -87,6 +87,24 @@ struct ray_pool {
  * in pool.c. */
 #define RAY_POOL_MAX_TASKS  (1u << 16)
 
+/* Initial task-ring capacity (power of 2).  ray_pool_dispatch_n callers that
+ * must never trigger ring growth (its clamp-on-grow-failure silently DROPS
+ * tasks) cap their task count at this value — one shared constant instead of
+ * a hardcoded 1024 at every site. */
+#define RAY_POOL_INIT_TASKS 1024u
+
+/* True when a data-parallel dispatch is both worthwhile AND safe: a live
+ * pool with background workers (at -c 1 the pool exists with n_workers==0),
+ * at least min_elems elements to amortize task overhead, and NOT already
+ * inside an in-flight dispatch — ray_pool_dispatch/_n are single-producer,
+ * so a nested dispatch from a worker thread would corrupt the task ring
+ * (ray_parallel_flag is raised for the duration of a dispatch). */
+static inline bool ray_pool_par_dispatch_ok(ray_pool_t* p, int64_t n,
+                                            int64_t min_elems) {
+    return p && p->n_workers > 0 && n >= min_elems &&
+           atomic_load_explicit(&ray_parallel_flag, memory_order_relaxed) == 0;
+}
+
 /* Initialize pool with n_workers background threads.
  * Pass 0 to auto-detect (nproc - 1). */
 ray_err_t ray_pool_create(ray_pool_t* pool, uint32_t n_workers);

src/lang/eval.c

@@ -1432,6 +1432,41 @@ ray_t* call_fn2(ray_t* fn, ray_t* a, ray_t* b) {
  * Sorting builtins
  * ══════════════════════════════════════════ */
 
+/* Parallel fixed-width index gather for gather_by_idx: value memcpy loops
+ * split across the worker pool.  Null-bit propagation stays serial (bit
+ * writes within a shared word would race across range boundaries).
+ *
+ * gather_by_idx is a LEAF utility with 35+ call sites; unlike op-level
+ * dispatch points it cannot know its execution context — the shared
+ * ray_pool_par_dispatch_ok reentrancy check (core/pool.h) is what makes
+ * dispatching from here safe. */
+static inline bool gbi_par_ok(ray_pool_t* p, int64_t n) {
+    return ray_pool_par_dispatch_ok(p, n, RAY_PARALLEL_THRESHOLD);
+}
+
+typedef struct {
+    const char*    src;
+    char*          dst;
+    const int64_t* idx;
+    uint8_t        esz;
+} gbi_ctx_t;
+
+static void gbi_fill_fn(void* ctxp, uint32_t wid, int64_t start, int64_t end) {
+    (void)wid;
+    gbi_ctx_t* c = (gbi_ctx_t*)ctxp;
+    const char* src = c->src;
+    char* dst = c->dst;
+    const int64_t* idx = c->idx;
+    switch (c->esz) {
+    case 8: for (int64_t i = start; i < end; i++) memcpy(dst + i*8, src + idx[i]*8, 8); break;
+    case 4: for (int64_t i = start; i < end; i++) memcpy(dst + i*4, src + idx[i]*4, 4); break;
+    case 2: for (int64_t i = start; i < end; i++) memcpy(dst + i*2, src + idx[i]*2, 2); break;
+    case 1: for (int64_t i = start; i < end; i++) dst[i] = src[idx[i]]; break;
+    case 16: for (int64_t i = start; i < end; i++) memcpy(dst + i*16, src + idx[i]*16, 16); break;
+    default: for (int64_t i = start; i < end; i++) memcpy(dst + i*c->esz, src + idx[i]*c->esz, c->esz); break;
+    }
+}
+
 /* Reorder vector elements by an index array */
 ray_t* gather_by_idx(ray_t* vec, int64_t* idx, int64_t n) {
     int8_t type = vec->type;
@@ -1465,15 +1500,14 @@ ray_t* gather_by_idx(ray_t* vec, int64_t* idx, int64_t n) {
         if (RAY_IS_ERR(result)) return result;
         result->len = n;
         uint8_t esz = (uint8_t)RAY_SYM_ELEM(w);
-        char* src = (char*)ray_data(vec);
-        char* dst = (char*)ray_data(result);
-        switch (esz) {
-        case 8: for (int64_t i = 0; i < n; i++) memcpy(dst + i*8, src + idx[i]*8, 8); break;
-        case 4: for (int64_t i = 0; i < n; i++) memcpy(dst + i*4, src + idx[i]*4, 4); break;
-        case 2: for (int64_t i = 0; i < n; i++) memcpy(dst + i*2, src + idx[i]*2, 2); break;
-        case 1: for (int64_t i = 0; i < n; i++) dst[i] = src[idx[i]]; break;
-        default: for (int64_t i = 0; i < n; i++) memcpy(dst + i*esz, src + idx[i]*esz, esz); break;
-        }
+        gbi_ctx_t gc = { .src = (const char*)ray_data(vec),
+                         .dst = (char*)ray_data(result),
+                         .idx = idx, .esz = esz };
+        ray_pool_t* gpool = ray_pool_get();
+        if (gbi_par_ok(gpool, n))
+            ray_pool_dispatch(gpool, gbi_fill_fn, &gc, n);
+        else
+            gbi_fill_fn(&gc, 0, 0, n);
         if (has_nulls) {
             for (int64_t i = 0; i < n; i++)
                 if (ray_vec_is_null(vec, idx[i]))
@@ -1506,17 +1540,16 @@ ray_t* gather_by_idx(ray_t* vec, int64_t* idx, int64_t n) {
     if (RAY_IS_ERR(result)) return result;
     result->len = n;
     uint8_t esz = ray_type_sizes[type];
-    char* src = (char*)ray_data(vec);
-    char* dst = (char*)ray_data(result);
-    /* Typed gather — compiler constant esz enables vectorization, alias-safe */
-    switch (esz) {
-    case 8: for (int64_t i = 0; i < n; i++) memcpy(dst + i*8, src + idx[i]*8, 8); break;
-    case 4: for (int64_t i = 0; i < n; i++) memcpy(dst + i*4, src + idx[i]*4, 4); break;
-    case 2: for (int64_t i = 0; i < n; i++) memcpy(dst + i*2, src + idx[i]*2, 2); break;
-    case 1: for (int64_t i = 0; i < n; i++) dst[i] = src[idx[i]]; break;
-    default: for (int64_t i = 0; i < n; i++) memcpy(dst + i*esz, src + idx[i]*esz, esz); break;
-    case 16: for (int64_t i = 0; i < n; i++) memcpy(dst + i*16, src + idx[i]*16, 16); break;
-    }
+    /* Typed gather — compiler constant esz enables vectorization, alias-safe;
+     * pool-parallel over disjoint output ranges for large gathers. */
+    gbi_ctx_t gc = { .src = (const char*)ray_data(vec),
+                     .dst = (char*)ray_data(result),
+                     .idx = idx, .esz = esz };
+    ray_pool_t* gpool = ray_pool_get();
+    if (gbi_par_ok(gpool, n))
+        ray_pool_dispatch(gpool, gbi_fill_fn, &gc, n);
+    else
+        gbi_fill_fn(&gc, 0, 0, n);
 
     /* Propagate null bitmap */
     if (has_nulls) {

src/ops/builtins.c

@@ -2100,11 +2100,99 @@ ray_t* ray_nil_fn(ray_t* x) {
 }
 
 /* (where bool-vec) -> indices of true values */
+
+/* Parallel where: 3-phase chunk compaction (parallel per-chunk popcount →
+ * tiny serial prefix → parallel per-chunk emit at disjoint offsets). */
+#define WHERE_CHUNK_ROWS 65536
+
+typedef struct {
+    const uint8_t* data;
+    int64_t  n;
+    int64_t  chunk_rows;
+    int64_t* chunk_counts;   /* popcount, then exclusive-prefix offsets */
+    int64_t* out;            /* emit-phase output (NULL during count) */
+} where_ctx_t;
+
+/* Dispatched via ray_pool_dispatch_n — each task is ONE chunk ([i,i+1)). */
+static void where_count_fn(void* ctxp, uint32_t wid, int64_t start, int64_t end) {
+    (void)wid;
+    where_ctx_t* c = (where_ctx_t*)ctxp;
+    for (int64_t ch = start; ch < end; ch++) {
+        int64_t lo = ch * c->chunk_rows;
+        int64_t hi = lo + c->chunk_rows;
+        if (hi > c->n) hi = c->n;
+        int64_t cnt = 0;
+        for (int64_t i = lo; i < hi; i++) cnt += (c->data[i] != 0);
+        c->chunk_counts[ch] = cnt;
+    }
+}
+
+/* Word-at-a-time zero skip: sparse masks — the common WHERE shape —
+ * run at memory speed, falling to per-byte only inside a live word. */
+static void where_emit_fn(void* ctxp, uint32_t wid, int64_t start, int64_t end) {
+    (void)wid;
+    where_ctx_t* c = (where_ctx_t*)ctxp;
+    for (int64_t ch = start; ch < end; ch++) {
+        int64_t lo = ch * c->chunk_rows;
+        int64_t hi = lo + c->chunk_rows;
+        if (hi > c->n) hi = c->n;
+        int64_t j = c->chunk_counts[ch];
+        int64_t i = lo;
+        int64_t h8 = lo + ((hi - lo) & ~7LL);
+        for (; i < h8; i += 8) {
+            uint64_t w;
+            memcpy(&w, c->data + i, 8);
+            if (!w) continue;
+            for (int64_t k = 0; k < 8; k++)
+                if (c->data[i + k]) c->out[j++] = i + k;
+        }
+        for (; i < hi; i++)
+            if (c->data[i]) c->out[j++] = i;
+    }
+}
+
 ray_t* ray_where_fn(ray_t* x) {
     if (!ray_is_vec(x) || x->type != RAY_BOOL)
         return ray_error("type", "where: argument must be a b8 vector, got %s", ray_type_name(x->type));
     const uint8_t* data = (const uint8_t*)ray_data(x);
     int64_t n = x->len;
+
+    ray_pool_t* pool = ray_pool_get();
+    if (ray_pool_par_dispatch_ok(pool, n, RAY_PARALLEL_THRESHOLD)) {
+        /* One task per chunk (ray_pool_dispatch_n).  Chunk count is capped
+         * at 1024 — the pool's INITIAL ring capacity — so dispatch_n never
+         * needs to grow the ring (its clamp-on-grow-failure silently DROPS
+         * tasks, which here would mean uninitialized prefix entries and
+         * out-of-bounds emit offsets).  1024 tasks is ample granularity. */
+        int64_t chunk_rows = WHERE_CHUNK_ROWS;
+        while ((n + chunk_rows - 1) / chunk_rows > RAY_POOL_INIT_TASKS) chunk_rows *= 2;
+        int64_t n_chunks = (n + chunk_rows - 1) / chunk_rows;
+        /* ops/internal.h's scratch_alloc idiom is unavailable here (that
+         * header clashes with builtins.c-local helpers); ray_alloc +
+         * ray_release below is its exact expansion. */
+        ray_t* cc_hdr = ray_alloc((size_t)n_chunks * sizeof(int64_t));
+        int64_t* chunk_counts = cc_hdr ? (int64_t*)ray_data(cc_hdr) : NULL;
+        if (chunk_counts) {
+            where_ctx_t wc = { .data = data, .n = n, .chunk_rows = chunk_rows,
+                               .chunk_counts = chunk_counts, .out = NULL };
+            ray_pool_dispatch_n(pool, where_count_fn, &wc, (uint32_t)n_chunks);
+            int64_t cum = 0;
+            for (int64_t ch = 0; ch < n_chunks; ch++) {
+                int64_t v = chunk_counts[ch];
+                chunk_counts[ch] = cum;
+                cum += v;
+            }
+            ray_t* result = ray_vec_new(RAY_I64, cum);
+            if (RAY_IS_ERR(result)) { ray_release(cc_hdr); return result; }
+            wc.out = (int64_t*)ray_data(result);
+            ray_pool_dispatch_n(pool, where_emit_fn, &wc, (uint32_t)n_chunks);
+            result->len = cum;
+            ray_release(cc_hdr);
+            return result;
+        }
+        /* scratch OOM → sequential fallback below */
+    }
+
     /* Count pass: `cnt += byte != 0` vectorizes; the branchy form costs
      * a mispredict-prone compare per element. */
     int64_t cnt = 0;