perf(ops): ClickBench bottleneck fixes — top-K, grouped count(distinct), LIKE on dict SYM

Lands the four findings + bonus from RAYFORCE_BOTTLENECKS.md, taking
ClickBench hot-run total from ~1.6 M ms to ~14 K ms across 40
measurable queries (≈99% reduction).

* Fused `select { … asc/desc: c take: K }` lowers to bounded-heap
  top-K when k << nrows and keys resolve to plain column refs.
  Single-key uses the radix-encoded fast path; multi-key falls back
  to the comparator-based heap.  Q26 SearchPhrase: 5 186 → 72 ms.

* Grouped `count(distinct)` no longer routed through per-group
  eval-fallback — the fused OP_COUNT_DISTINCT runs per group-slice.
  Scaling moves from 94×/decade to ≈4.6×/decade between 100 K and
  1 M rows (essentially linear).

* LIKE on dict-encoded SYM scans the dictionary once and lifts the
  result through the codes vector instead of re-evaluating per row.
  Low-card SYM (54-unique BrowserCountry): 52 → 3.65 ms (14×).
  High-card SYM (1.73 M-unique URL): 498 → 220 ms (2.3×).

* Unifies the previously-divergent glob matchers (eval used `*?[abc]`,
  DAG used SQL `%_`; one variant blew up exponentially on
  `a*a*…a*b` against an a-only string) behind a single iterative
  two-pointer implementation in src/ops/glob.{c,h}.  Both call sites
  delegate.

* Bonus: `(at table (iasc table.col))` no longer crashes on tables —
  re-indexes each column to return a TABLE.

Tests: query_coverage / read_csv / reserved_namespace updated for the
new dispatch paths; cross_type_workout / collection/at extended.

Author: singaraiona

include/rayforce.h

@@ -359,6 +359,14 @@ int64_t  ray_sym_intern(const char* str, size_t len);
 int64_t  ray_sym_find(const char* str, size_t len);
 ray_t*    ray_sym_str(int64_t id);
 uint32_t ray_sym_count(void);
+
+/* Borrow a snapshot of the sym → string array.  Returns a pointer to
+ * the underlying ray_t** strings table along with its length; valid
+ * only while no concurrent ray_sym_intern occurs (i.e. read-only
+ * execution phases).  Lock is taken once for the snapshot and dropped
+ * before return — caller may iterate freely.  Both *out_strings and
+ * *out_count must be non-NULL. */
+void ray_sym_strings_borrow(ray_t*** out_strings, uint32_t* out_count);
 bool     ray_sym_ensure_cap(uint32_t needed);
 ray_err_t ray_sym_save(const char* path);
 ray_err_t ray_sym_load(const char* path);

src/lang/eval.c

@@ -875,10 +875,6 @@ ray_t* gather_by_idx(ray_t* vec, int64_t* idx, int64_t n) {
         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;
         }
-        if (vec->sym_dict) {
-            ray_retain(vec->sym_dict);
-            result->sym_dict = vec->sym_dict;
-        }
         if (has_nulls) {
             for (int64_t i = 0; i < n; i++)
                 if (ray_vec_is_null(vec, idx[i]))
@@ -2280,7 +2276,12 @@ static void ray_register_builtins(void) {
     register_vary("update",    RAY_FN_SPECIAL_FORM | RAY_FN_RESTRICTED, ray_update_fn);
     register_vary("insert",    RAY_FN_SPECIAL_FORM | RAY_FN_RESTRICTED, ray_insert_fn);
     register_vary("upsert",    RAY_FN_SPECIAL_FORM | RAY_FN_RESTRICTED, ray_upsert_fn);
-    register_binary("xbar",    RAY_FN_ATOMIC, ray_xbar_fn);
+    /* xbar is registered NON-atomic so the call path lands in
+     * ray_xbar_fn(VEC, scalar) directly.  ray_xbar_fn handles the
+     * vector fast path itself (tight per-element loop, no per-atom
+     * allocation) and recurses through atomic_map_binary for the rare
+     * (collection, collection) zip case. */
+    register_binary("xbar",    RAY_FN_NONE,   ray_xbar_fn);
 
     /* Join operations */
     register_vary("left-join",   RAY_FN_NONE, ray_left_join_fn);
@@ -2294,6 +2295,8 @@ static void ray_register_builtins(void) {
     register_vary("println",    RAY_FN_NONE, ray_println_fn);
     register_vary("show",       RAY_FN_NONE, ray_show_fn);
     register_vary("format",     RAY_FN_NONE, ray_format_fn);
+    register_vary("read-csv",   RAY_FN_RESTRICTED, ray_read_csv_fn);
+    register_vary("write-csv",  RAY_FN_RESTRICTED, ray_write_csv_fn);
     register_vary(".csv.read",  RAY_FN_RESTRICTED, ray_read_csv_fn);
     register_vary(".csv.write", RAY_FN_RESTRICTED, ray_write_csv_fn);
     register_binary("as",       RAY_FN_NONE, ray_cast_fn);

src/ops/collection.c

@@ -1554,6 +1554,39 @@ ray_t* ray_at_fn(ray_t* vec, ray_t* idx) {
         return ray_dict_new(keys, vals);
     }
 
+    /* Table row selection by index vector: apply the row ids to each
+     * column and return a table.  Keep this before the generic collection
+     * fallback; otherwise a table indexed by millions of row ids becomes
+     * a LIST of row dictionaries. */
+    if (vec->type == RAY_TABLE && idx->type == RAY_I64) {
+        int64_t nrows = ray_table_nrows(vec);
+        int64_t nidx = ray_len(idx);
+        int64_t* ids = (int64_t*)ray_data(idx);
+        for (int64_t i = 0; i < nidx; i++) {
+            if (ids[i] < 0 || ids[i] >= nrows)
+                return ray_error("domain", NULL);
+        }
+
+        int64_t ncols = ray_table_ncols(vec);
+        ray_t* result = ray_table_new(ncols);
+        if (!result || RAY_IS_ERR(result)) return result ? result : ray_error("oom", NULL);
+        for (int64_t c = 0; c < ncols; c++) {
+            ray_t* col = ray_table_get_col_idx(vec, c);
+            int64_t name = ray_table_col_name(vec, c);
+            if (!col) continue;
+            ray_t* gathered = gather_by_idx(col, ids, nidx);
+            if (!gathered || RAY_IS_ERR(gathered)) {
+                ray_release(result);
+                return gathered ? gathered : ray_error("oom", NULL);
+            }
+            result = ray_table_add_col(result, name, gathered);
+            ray_release(gathered);
+            if (!result || RAY_IS_ERR(result))
+                return result ? result : ray_error("oom", NULL);
+        }
+        return result;
+    }
+
     /* Dict key access: (at dict key) → value or 0Nl if missing */
     if (vec->type == RAY_DICT) {
         ray_t* v = ray_dict_get(vec, idx);

src/ops/glob.c

@@ -13,6 +13,9 @@
 
 #include "ops/glob.h"
 
+#define _GNU_SOURCE
+#include <string.h>
+
 /* Lowercase an ASCII byte; non-ASCII passes through unchanged. */
 static inline char to_lower(char c) {
     return (c >= 'A' && c <= 'Z') ? (char)(c + 32) : c;
@@ -100,3 +103,96 @@ bool ray_glob_match(const char* s, size_t sn, const char* p, size_t pn) {
 bool ray_glob_match_ci(const char* s, size_t sn, const char* p, size_t pn) {
     return glob_impl(s, sn, p, pn, true);
 }
+
+ray_glob_compiled_t ray_glob_compile(const char* p, size_t pn) {
+    ray_glob_compiled_t c = { RAY_GLOB_SHAPE_NONE, NULL, 0 };
+
+    if (pn == 0) {
+        c.shape = RAY_GLOB_SHAPE_EXACT;
+        c.lit = p; c.lit_len = 0;
+        return c;
+    }
+
+    /* Strip a single leading and trailing '*'; classify by the residual
+     * pattern.  Any other glob metachar (`?`, `[`, or interior `*`)
+     * forces the general matcher. */
+    size_t lo = 0, hi = pn;
+    bool leading_star  = (p[0] == '*');
+    bool trailing_star = (pn > 0 && p[pn - 1] == '*' &&
+                          /* don't double-count single '*' as both */
+                          (pn > 1 || !leading_star));
+    if (leading_star)  lo = 1;
+    if (trailing_star) hi = pn - 1;
+
+    /* Ensure the residual has no glob metacharacters. */
+    for (size_t i = lo; i < hi; i++) {
+        char ch = p[i];
+        if (ch == '*' || ch == '?' || ch == '[') {
+            c.shape = RAY_GLOB_SHAPE_NONE;
+            return c;
+        }
+    }
+
+    c.lit     = p + lo;
+    c.lit_len = hi - lo;
+
+    if (leading_star && trailing_star) {
+        c.shape = (c.lit_len == 0) ? RAY_GLOB_SHAPE_ANY
+                                   : RAY_GLOB_SHAPE_CONTAINS;
+    } else if (leading_star) {
+        c.shape = RAY_GLOB_SHAPE_SUFFIX;
+    } else if (trailing_star) {
+        c.shape = RAY_GLOB_SHAPE_PREFIX;
+    } else {
+        c.shape = RAY_GLOB_SHAPE_EXACT;
+    }
+    return c;
+}
+
+bool ray_glob_match_compiled(const ray_glob_compiled_t* c,
+                             const char* s, size_t sn) {
+    switch (c->shape) {
+    case RAY_GLOB_SHAPE_ANY:
+        return true;
+    case RAY_GLOB_SHAPE_EXACT:
+        return sn == c->lit_len &&
+               (c->lit_len == 0 || memcmp(s, c->lit, c->lit_len) == 0);
+    case RAY_GLOB_SHAPE_PREFIX:
+        return sn >= c->lit_len &&
+               (c->lit_len == 0 || memcmp(s, c->lit, c->lit_len) == 0);
+    case RAY_GLOB_SHAPE_SUFFIX:
+        return sn >= c->lit_len &&
+               (c->lit_len == 0 ||
+                memcmp(s + sn - c->lit_len, c->lit, c->lit_len) == 0);
+    case RAY_GLOB_SHAPE_CONTAINS:
+        if (c->lit_len == 0) return true;
+        if (sn < c->lit_len) return false;
+        /* glibc's memmem is SIMD-accelerated; use it where available.
+         * Falls back to a portable Boyer-Moore-Horspool when not. */
+#if defined(__GLIBC__) || defined(__APPLE__) || defined(__FreeBSD__)
+        return memmem(s, sn, c->lit, c->lit_len) != NULL;
+#else
+        {
+            /* Portable fallback: short-needle byte scan with memchr. */
+            const char first = c->lit[0];
+            const char* haystack = s;
+            size_t remaining = sn;
+            while (remaining >= c->lit_len) {
+                const char* hit = (const char*)memchr(haystack, first,
+                                                      remaining - c->lit_len + 1);
+                if (!hit) return false;
+                if (memcmp(hit, c->lit, c->lit_len) == 0) return true;
+                size_t adv = (size_t)(hit - haystack) + 1;
+                haystack = hit + 1;
+                remaining -= adv;
+            }
+            return false;
+        }
+#endif
+    case RAY_GLOB_SHAPE_NONE:
+    default:
+        /* Caller contract violation — fall through to false rather than
+         * silently matching everything. */
+        return false;
+    }
+}

src/ops/glob.h

@@ -40,4 +40,47 @@
 bool ray_glob_match(const char* s, size_t sn, const char* p, size_t pn);
 bool ray_glob_match_ci(const char* s, size_t sn, const char* p, size_t pn);
 
+/* ---- Pre-compiled pattern fast path -------------------------------------
+ * Many LIKE workloads have very simple patterns (e.g. `*google*`).  When
+ * the pattern has no metacharacters except (optionally) a leading `*`
+ * and/or a trailing `*`, the match collapses to a literal substring /
+ * prefix / suffix / equality test that we can drive with memcmp /
+ * memmem — both libc-vectorised on modern glibc.  Detect the shape once
+ * up front, then run the entire dictionary (or row vector) through a
+ * single tight loop.
+ *
+ * Shapes:
+ *   RAY_GLOB_SHAPE_NONE     — pattern needs the full glob matcher
+ *   RAY_GLOB_SHAPE_EXACT    — no `*`/`?`/`[` — literal equality
+ *   RAY_GLOB_SHAPE_PREFIX   — `<lit>*`        — strncmp prefix
+ *   RAY_GLOB_SHAPE_SUFFIX   — `*<lit>`        — tail equality
+ *   RAY_GLOB_SHAPE_CONTAINS — `*<lit>*`       — memmem
+ *   RAY_GLOB_SHAPE_ANY      — pattern is "*"  — always true
+ * The compiled struct caches a pointer/length into the original
+ * pattern buffer, so the caller must keep the pattern alive while the
+ * compiled view is in use. */
+typedef enum {
+    RAY_GLOB_SHAPE_NONE = 0,
+    RAY_GLOB_SHAPE_EXACT,
+    RAY_GLOB_SHAPE_PREFIX,
+    RAY_GLOB_SHAPE_SUFFIX,
+    RAY_GLOB_SHAPE_CONTAINS,
+    RAY_GLOB_SHAPE_ANY,
+} ray_glob_shape_t;
+
+typedef struct {
+    ray_glob_shape_t shape;
+    const char*      lit;     /* literal substring inside the pattern */
+    size_t           lit_len;
+} ray_glob_compiled_t;
+
+/* Classify a pattern.  Returns the simplest matching shape; falls back
+ * to RAY_GLOB_SHAPE_NONE when the pattern needs the general matcher. */
+ray_glob_compiled_t ray_glob_compile(const char* p, size_t pn);
+
+/* Match a single string against a compiled simple-shape pattern.
+ * Caller must guarantee shape != RAY_GLOB_SHAPE_NONE. */
+bool ray_glob_match_compiled(const ray_glob_compiled_t* c,
+                             const char* s, size_t sn);
+
 #endif /* RAY_OPS_GLOB_H */