rayforce-sys 1.2.0

/*
 *   Copyright (c) 2025-2026 Anton Kundenko <singaraiona@gmail.com>
 *   All rights reserved.

 *   Permission is hereby granted, free of charge, to any person obtaining a copy
 *   of this software and associated documentation files (the "Software"), to deal
 *   in the Software without restriction, including without limitation the rights
 *   to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 *   copies of the Software, and to permit persons to whom the Software is
 *   furnished to do so, subject to the following conditions:

 *   The above copyright notice and this permission notice shall be included in all
 *   copies or substantial portions of the Software.

 *   THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 *   IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 *   FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 *   AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 *   LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 *   OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 *   SOFTWARE.
 */

/*  Table builtins — extracted from eval.c  */

#include "lang/internal.h"
#include "lang/env.h"
#include "ops/ops.h"
#include "ops/internal.h"
#include "ops/hash.h"
#include "ops/idxop.h"
#include "ops/linkop.h"
#include "table/sym.h"
#include "mem/heap.h"
#include <stdio.h>
#include <inttypes.h>

/* ══════════════════════════════════════════
 * pivot_fn_to_agg_op
 * ══════════════════════════════════════════ */

/* Map a RAY_UNARY agg function pointer to a DAG opcode.
 * Returns 0 if the function is not a known aggregation builtin. */
uint16_t pivot_fn_to_agg_op(ray_t* fn) {
    if (fn->type != RAY_UNARY) return 0;
    ray_unary_fn f = (ray_unary_fn)(uintptr_t)fn->i64;
    if (f == ray_sum_fn)   return OP_SUM;
    if (f == ray_avg_fn)   return OP_AVG;
    if (f == ray_min_fn)      return OP_MIN;
    if (f == ray_max_fn)      return OP_MAX;
    if (f == ray_count_fn) return OP_COUNT;
    if (f == ray_first_fn) return OP_FIRST;
    if (f == ray_last_fn)  return OP_LAST;
    return 0;
}

/* ══════════════════════════════════════════
 * pivot
 * ══════════════════════════════════════════ */

/* Build a shallow copy of `tbl` whose F64 pivot-key column has -0.0 collapsed
 * to +0.0, so 0.0/-0.0 form one pivot group/column.  Done ONCE here (a cold
 * pivot path) over the key column, never per-row in the group-by key-build, so
 * the -fno-signed-zeros vectorization of the hot group-by path stays intact.
 * Returns NULL when the column has no -0.0 (caller uses the original table),
 * or RAY_ERR on allocation failure. */
static ray_t* pivot_normalize_f64_key(ray_t* tbl, int64_t pcol_name) {
    ray_t* pcol = ray_table_get_col(tbl, pcol_name);
    if (!pcol || pcol->type != RAY_F64) return NULL;
    int64_t nr = pcol->len;
    const double* src = (const double*)ray_data(pcol);
    bool any_neg0 = false;
    for (int64_t i = 0; i < nr; i++) {
        double nz = clear_neg_zero(src[i]);
        if (memcmp(&nz, &src[i], sizeof(double)) != 0) { any_neg0 = true; break; }
    }
    if (!any_neg0) return NULL;
    ray_t* norm = ray_vec_new(RAY_F64, nr);
    if (!norm || RAY_IS_ERR(norm)) return norm ? norm : ray_error("oom", NULL);
    norm->len = nr;
    double* dst = (double*)ray_data(norm);
    for (int64_t i = 0; i < nr; i++) dst[i] = clear_neg_zero(src[i]);
    int64_t ncols = ray_table_ncols(tbl);
    ray_t* wt = ray_table_new(ncols);
    if (!wt || RAY_IS_ERR(wt)) { ray_release(norm); return wt ? wt : ray_error("oom", NULL); }
    for (int64_t c = 0; c < ncols && !RAY_IS_ERR(wt); c++) {
        int64_t cn = ray_table_col_name(tbl, c);
        ray_t* cv = (cn == pcol_name) ? norm : ray_table_get_col_idx(tbl, c);
        wt = ray_table_add_col(wt, cn, cv);
    }
    ray_release(norm);
    return wt;
}

/* (pivot table index_col pivot_col value_col agg_fn) — pivot table */
static ray_t* pivot_fn_impl(ray_t* tbl, ray_t* index_arg, ray_t* pivot_col_name,
                            ray_t* value_col_name, ray_t* agg_fn) {

    if (tbl->type != RAY_TABLE)
        return ray_error("type", "pivot: first argument must be a table, got %s", ray_type_name(tbl->type));
    if (pivot_col_name->type != -RAY_SYM)
        return ray_error("type", "pivot: pivot-col must be a symbol, got %s", ray_type_name(pivot_col_name->type));
    if (value_col_name->type != -RAY_SYM)
        return ray_error("type", "pivot: value-col must be a symbol, got %s", ray_type_name(value_col_name->type));
    if (agg_fn->type != RAY_UNARY && agg_fn->type != RAY_LAMBDA &&
        agg_fn->type != RAY_VARY)
        return ray_error("type", "pivot: agg-fn must be a function, got %s", ray_type_name(agg_fn->type));

    /* Determine index columns.  idx_syms are column NAME ids: atoms are
     * runtime-domain by design and collection_elem re-expresses SYM vec
     * cells into runtime atoms (the Task-3 chokepoint), so the
     * ray_sym_str / ray_table_get_col calls below stay global.
     *
     * No count cap: pivot's index-column bound used to mirror exec_pivot's
     * ght layout's fixed [8]-slot key array (pivot.c: n_keys = n_idx + 1 >
     * 8 -> error), a documented spec deviation from cut 3's otherwise-
     * unbounded join/group/distinct key caps.  The cut-4 ght rework widens
     * that layout to spill past 8 keys (same mechanism group.c's
     * exec_group_run relies on), so both this caller-side check and
     * exec_pivot's own gate are gone — n_idx is sized from the real column
     * count below, not compared against a cap. */
    int64_t n_idx;
    if (index_arg->type == -RAY_SYM) {
        n_idx = 1;
    } else if (index_arg->type == RAY_LIST || ray_is_vec(index_arg)) {
        n_idx = ray_len(index_arg);
    } else {
        return ray_error("type", "pivot: index must be a symbol or list of symbols, got %s", ray_type_name(index_arg->type));
    }

    /* n_idx is a USER-SUPPLIED RUNTIME value — the index-arg's length, not
     * an AST-bounded count (unlike group.c's exec_group_run driver arrays,
     * whose element count is capped by query text) — so a caller can pass
     * a multi-million-element index vector.  A stack VLA at that size is a
     * stack-clash SIGSEGV where the old code returned a clean limit error;
     * heap-carve instead.  One consolidated block holds idx_syms (column
     * NAME ids) and icols (the resolved column pointers) — both survive to
     * the very end of this function (the fb_cleanup result-building code
     * below reads both) — so every one of this function's many return
     * points must scratch_free(ic_hdr) exactly once.  Floored at 1 to
     * sidestep zero-size alloc UB (an empty index list is legal: pivot
     * grouped by the pivot column alone). */
    int64_t vla_idx = n_idx > 0 ? n_idx : 1;
    ray_t* ic_hdr = NULL;
    int64_t* idx_syms = (int64_t*)scratch_alloc(&ic_hdr,
            (size_t)vla_idx * (sizeof(int64_t) + sizeof(ray_t*)));
    if (!idx_syms) return ray_error("oom", NULL);
    ray_t** icols = (ray_t**)(idx_syms + vla_idx);
    if (index_arg->type == -RAY_SYM) {
        idx_syms[0] = index_arg->i64;
    } else {
        for (int64_t i = 0; i < n_idx; i++) {
            int alloc = 0;
            ray_t* elem = collection_elem(index_arg, i, &alloc);
            if (RAY_IS_ERR(elem)) { scratch_free(ic_hdr); return elem; }
            if (elem->type != -RAY_SYM) {
                int8_t et = elem->type;
                if (alloc) ray_release(elem);
                scratch_free(ic_hdr);
                return ray_error("type", "pivot: index columns must be symbols, got %s", ray_type_name(et));
            }
            idx_syms[i] = elem->i64;
            if (alloc) ray_release(elem);
        }
    }

    /* Get pivot column, value column */
    ray_t* pcol = ray_table_get_col(tbl, pivot_col_name->i64);
    if (!pcol) { scratch_free(ic_hdr); return ray_error("domain", "pivot: pivot column not found"); }
    ray_t* vcol = ray_table_get_col(tbl, value_col_name->i64);
    if (!vcol) { scratch_free(ic_hdr); return ray_error("domain", "pivot: value column not found"); }

    /* The pivot column's distinct values become output column NAMES (syms);
     * a STR pivot column cannot supply names.  (STR *index* columns ARE now
     * supported — they group via the wide-key path; this restriction is
     * specific to the pivot column, whose values name the unstacked columns.) */
    if (pcol->type == RAY_STR) {
        scratch_free(ic_hdr);
        return ray_error("nyi", "pivot: STR pivot column not supported as column names");
    }

    /* Get index columns (icols carved together with idx_syms above) */
    for (int64_t i = 0; i < n_idx; i++) {
        icols[i] = ray_table_get_col(tbl, idx_syms[i]);
        if (!icols[i]) { scratch_free(ic_hdr); return ray_error("domain", "pivot: index column not found"); }
    }

    int64_t nrows = ray_table_nrows(tbl);
    if (nrows == 0) { scratch_free(ic_hdr); return ray_table_new(0); }

    /* DAG fast path: known agg builtins on hashable columns → OP_PIVOT */
    uint16_t agg_op = pivot_fn_to_agg_op(agg_fn);
    bool dag_ok = (agg_op != 0 && pcol->type != RAY_STR && vcol->type != RAY_STR);
    for (int64_t i = 0; i < n_idx && dag_ok; i++)
        if (icols[i]->type == RAY_STR) dag_ok = false;

    if (dag_ok) {
        ray_graph_t* g = ray_graph_new(tbl);
        if (!g) { scratch_free(ic_hdr); return ray_error("oom", NULL); }
        /* Same user-length hazard as idx_syms/icols above: heap-carve.
         *
         * SECOND, independent hazard uncovered by lifting the index-count
         * cap: ray_scan() appends a node to `g`, which grows g->nodes via
         * realloc once node_count crosses node_cap (GRAPH_INIT_CAP=4096,
         * doubling) — moving the whole array.  Any ray_op_t* returned by an
         * EARLIER ray_scan() call in the loop below is a pointer into the
         * OLD (now freed) g->nodes block once a LATER call in the same
         * loop crosses that boundary — exactly the hazard op_node's own
         * contract warns about ("never store [a node pointer] across a
         * node allocation; re-resolve from the id instead").  n_idx used
         * to be capped well under 4096, so this was unreachable; a 5000+
         * duplicated-index pivot now reaches it and segfaults inside
         * exec_group on a garbage node id read through the stale pointer.
         * Fix: collect each scan's stable ->id through every ray_scan call
         * in this block, and only resolve id -> pointer in ONE pass after
         * the LAST scan call (v_op below) — that resolve reads the FINAL
         * g->nodes block, and nothing after it grows the graph again
         * before ray_pivot_op consumes the pointers. */
        ray_t* iids_hdr = NULL;
        ray_op_t** idx_ops = (ray_op_t**)scratch_alloc(&iids_hdr,
                (size_t)vla_idx * (sizeof(ray_op_t*) + sizeof(uint32_t)));
        if (!idx_ops) { ray_graph_free(g); scratch_free(ic_hdr); return ray_error("oom", NULL); }
        uint32_t* idx_ids = (uint32_t*)(idx_ops + vla_idx);
        bool ok = true;
        for (int64_t i = 0; i < n_idx && ok; i++) {
            ray_t* s = ray_sym_str(idx_syms[i]);
            ray_op_t* op = s ? ray_scan(g, ray_str_ptr(s)) : NULL;
            if (!op) { ok = false; break; }
            idx_ids[i] = op->id;
        }
        ray_t* ps = ray_sym_str(pivot_col_name->i64);
        ray_op_t* p_scan = (ps && ok) ? ray_scan(g, ray_str_ptr(ps)) : NULL;
        uint32_t p_id = p_scan ? p_scan->id : RAY_OP_NONE;
        ray_t* vs = ray_sym_str(value_col_name->i64);
        ray_op_t* v_op = (vs && p_scan) ? ray_scan(g, ray_str_ptr(vs)) : NULL;
        if (v_op) {
            /* Resolve now: v_op (the LAST scan call above) is still fresh;
             * idx_ops/p_op must be re-read from their ids (see comment
             * above — their own earlier pointers may already be stale). */
            for (int64_t i = 0; i < n_idx; i++) idx_ops[i] = op_node(g, idx_ids[i]);
            ray_op_t* p_op = op_node(g, p_id);
            ray_op_t* root = ray_pivot_op(g, idx_ops, (uint32_t)n_idx, p_op, v_op, agg_op);
            scratch_free(iids_hdr);
            if (root) {
                ray_t* result = ray_execute(g, root);
                ray_graph_free(g);
                scratch_free(ic_hdr);
                return result;
            }
        } else {
            scratch_free(iids_hdr);
        }
        ray_graph_free(g);
    }

    /* Generic fallback: use OP_GROUP DAG to group by (index_cols, pivot_col),
     * then apply agg_fn per group and unstack.  Single O(n) hash pass. */

    /* Build GROUP BY (idx0, ..., idxN-1, pivot_col) with COUNT agg via DAG */
    ray_graph_t* g = ray_graph_new(tbl);
    if (!g) { scratch_free(ic_hdr); return ray_error("oom", NULL); }

    uint32_t n_keys = (uint32_t)n_idx + 1;
    /* Same two hazards as the DAG fast path above (user-length heap-carve
     * + stale ray_op_t* across a reallocating ray_scan loop): collect ids
     * through every scan call, resolve to pointers in one pass right
     * before ray_group — val_scan below is the LAST scan call in this
     * function, so it stays fresh without needing the id round-trip. */
    ray_t* kids_hdr = NULL;
    ray_op_t** key_ops = (ray_op_t**)scratch_alloc(&kids_hdr,
            (size_t)n_keys * (sizeof(ray_op_t*) + sizeof(uint32_t)));
    if (!key_ops) { ray_graph_free(g); scratch_free(ic_hdr); return ray_error("oom", NULL); }
    uint32_t* key_ids = (uint32_t*)(key_ops + n_keys);
    bool ok = true;
    for (int64_t i = 0; i < n_idx && ok; i++) {
        ray_t* s = ray_sym_str(idx_syms[i]);
        ray_op_t* op = s ? ray_scan(g, ray_str_ptr(s)) : NULL;
        if (!op) { ok = false; break; }
        key_ids[i] = op->id;
    }
    {
        ray_t* ps = ray_sym_str(pivot_col_name->i64);
        ray_op_t* op = (ps && ok) ? ray_scan(g, ray_str_ptr(ps)) : NULL;
        if (op) key_ids[n_idx] = op->id; else ok = false;
    }
    /* Value column scan for COUNT (just need a column ref for group) */
    ray_t* vs = ray_sym_str(value_col_name->i64);
    ray_op_t* val_scan = (vs && ok) ? ray_scan(g, ray_str_ptr(vs)) : NULL;
    if (!val_scan) { scratch_free(kids_hdr); ray_graph_free(g); scratch_free(ic_hdr); return ray_error("domain", "pivot: failed to build DAG"); }
    for (uint32_t i = 0; i < n_keys; i++) key_ops[i] = op_node(g, key_ids[i]);

    uint16_t grp_agg_ops[1] = { OP_COUNT };
    ray_op_t* grp_agg_ins[1] = { val_scan };
    ray_op_t* grp_root = ray_group(g, key_ops, n_keys, grp_agg_ops, grp_agg_ins, 1);
    scratch_free(kids_hdr);
    if (!grp_root) { ray_graph_free(g); scratch_free(ic_hdr); return ray_error("oom", NULL); }

    ray_t* grouped = ray_execute(g, grp_root);
    ray_graph_free(g);
    if (!grouped || RAY_IS_ERR(grouped)) { scratch_free(ic_hdr); return grouped; }

    /* `grouped` is a table: (idx0, ..., idxN-1, pivot_col, _count).
     * Each row is one (index, pivot) combination.
     * Now for each group, gather the value column subset and apply agg_fn. */
    int64_t n_grps = ray_table_nrows(grouped);

    /* Same user-length hazard as idx_syms/icols: heap-carve.  Lives until
     * the unstack loop below finishes using it (last read is inside the
     * "Find or insert index key" loop over ix_list), then freed there —
     * every return between here and that point must also free it. */
    ray_t* g_icols_hdr = NULL;
    ray_t** g_icols = (ray_t**)scratch_alloc(&g_icols_hdr, (size_t)vla_idx * sizeof(ray_t*));
    if (!g_icols) { ray_release(grouped); scratch_free(ic_hdr); return ray_error("oom", NULL); }
    for (int64_t i = 0; i < n_idx; i++)
        g_icols[i] = ray_table_get_col(grouped, idx_syms[i]);
    ray_t* g_pcol = ray_table_get_col(grouped, pivot_col_name->i64);

    /* Collect distinct pivot values and index keys from grouped table.
     * ray_distinct_fn returns a lazy for concrete vecs; materialise here
     * since pivot C-code needs a concrete result to iterate over. */
    ray_retain(g_pcol);
    ray_t* dvals = ray_distinct_fn(g_pcol);
    ray_release(g_pcol);
    if (RAY_IS_ERR(dvals)) { scratch_free(g_icols_hdr); ray_release(grouped); scratch_free(ic_hdr); return dvals; }
    if (ray_is_lazy(dvals)) {
        dvals = ray_lazy_materialize(dvals);
        if (!dvals || RAY_IS_ERR(dvals)) {
            scratch_free(g_icols_hdr); ray_release(grouped); scratch_free(ic_hdr);
            return dvals ? dvals : ray_error("oom", NULL);
        }
    }
    int64_t n_pv = ray_len(dvals);

    /* Re-scan original table to assign a grouped-row index to each
     * input row.  Previously this was an O(nrows * n_grps) nested loop
     * that hung on any large pivot that took the generic fallback.
     * Replaced with an open-addressed hash table keyed by a cheap row
     * hash of (idx_cols..., pivot_col), giving O(nrows + n_grps) in the
     * common case.  Hash collisions re-verify via atom_eq so unhashable
     * cells (strings, guids) still match correctly.
     *
     * Hash helper: produces the same value when called on two rows with
     * equal cell values for numeric/sym/temporal columns; for strings
     * and guids we under-hash (returning a type-independent constant)
     * and rely entirely on atom_eq for equality. */
    #define FB_ROW_HASH(cols, ncols, pv, rid)                                \
        ({                                                                    \
            uint64_t _h = 0;                                                  \
            for (int64_t _k = 0; _k < (ncols); _k++) {                        \
                ray_t* _c = (cols)[_k];                                       \
                uint64_t _kh;                                                 \
                if (ray_vec_is_null(_c, (rid)))                               \
                    _kh = 0x9E3779B97F4A7C15ULL ^ (uint64_t)(rid);            \
                else if (_c->type == RAY_F64)                                 \
                    _kh = ray_hash_f64(((double*)ray_data(_c))[(rid)]);       \
                else if (_c->type == RAY_STR || _c->type == RAY_GUID)         \
                    _kh = 0xDEADBEEFCAFEBABEULL;                              \
                else                                                           \
                    _kh = ray_hash_i64(read_col_i64(ray_data(_c), (rid),      \
                                                    _c->type, _c->attrs));    \
                _h = (_k == 0) ? _kh : ray_hash_combine(_h, _kh);             \
            }                                                                  \
            ray_t* _pc = (pv);                                                 \
            uint64_t _ph;                                                      \
            if (ray_vec_is_null(_pc, (rid)))                                  \
                _ph = 0x165667B19E3779F9ULL ^ (uint64_t)(rid);                \
            else if (_pc->type == RAY_F64)                                     \
                _ph = ray_hash_f64(((double*)ray_data(_pc))[(rid)]);          \
            else if (_pc->type == RAY_STR || _pc->type == RAY_GUID)            \
                _ph = 0xFEEDFACE12345678ULL;                                   \
            else                                                                \
                _ph = ray_hash_i64(read_col_i64(ray_data(_pc), (rid),         \
                                                 _pc->type, _pc->attrs));      \
            ray_hash_combine(_h, _ph);                                         \
        })

    uint32_t gid_cap = 256;
    while (gid_cap < (uint32_t)n_grps * 2 && gid_cap < (1u << 30)) gid_cap <<= 1;
    ray_t* gid_ht_hdr = ray_alloc((size_t)gid_cap * sizeof(uint32_t));
    if (!gid_ht_hdr) {
        scratch_free(g_icols_hdr); ray_release(dvals); ray_release(grouped); scratch_free(ic_hdr);
        return ray_error("oom", NULL);
    }
    uint32_t* gid_ht = (uint32_t*)ray_data(gid_ht_hdr);
    memset(gid_ht, 0xFF, gid_cap * sizeof(uint32_t));
    uint32_t gid_mask = gid_cap - 1;

    /* Insert each grouped row into the HT (grouped rows are already
     * distinct by construction — no equality check needed on insert). */
    for (int64_t gi = 0; gi < n_grps; gi++) {
        uint64_t h = FB_ROW_HASH(g_icols, n_idx, g_pcol, gi);
        uint32_t slot = (uint32_t)(h & gid_mask);
        while (gid_ht[slot] != UINT32_MAX) slot = (slot + 1) & gid_mask;
        gid_ht[slot] = (uint32_t)gi;
    }

    ray_t* gid_vec = ray_vec_new(RAY_I64, nrows);
    if (!gid_vec || RAY_IS_ERR(gid_vec)) {
        ray_free(gid_ht_hdr); scratch_free(g_icols_hdr); ray_release(dvals); ray_release(grouped);
        scratch_free(ic_hdr);
        return ray_error("oom", NULL);
    }
    gid_vec->len = nrows;
    int64_t* gids = (int64_t*)ray_data(gid_vec);

    /* Probe HT for each input row; on collision fall through to atom_eq. */
    for (int64_t r = 0; r < nrows; r++) {
        uint64_t h = FB_ROW_HASH(icols, n_idx, pcol, r);
        uint32_t slot = (uint32_t)(h & gid_mask);
        int64_t found = -1;
        while (gid_ht[slot] != UINT32_MAX) {
            int64_t gi = gid_ht[slot];
            bool match = true;
            for (int64_t ci = 0; ci < n_idx && match; ci++) {
                int a1 = 0, a2 = 0;
                ray_t* v1 = collection_elem(icols[ci], r, &a1);
                ray_t* v2 = collection_elem(g_icols[ci], gi, &a2);
                if (!atom_eq(v1, v2)) match = false;
                if (a1) ray_release(v1);
                if (a2) ray_release(v2);
            }
            if (match) {
                int a1 = 0, a2 = 0;
                ray_t* v1 = collection_elem(pcol, r, &a1);
                ray_t* v2 = collection_elem(g_pcol, gi, &a2);
                if (!atom_eq(v1, v2)) match = false;
                if (a1) ray_release(v1);
                if (a2) ray_release(v2);
            }
            if (match) { found = gi; break; }
            slot = (slot + 1) & gid_mask;
        }
        gids[r] = found;
    }
    ray_free(gid_ht_hdr);

    /* For each group, gather the value column subset and apply agg_fn.
     * agg_results owns the per-group results as a RAY_LIST, so ray_free
     * on it releases each child exactly once via ray_release_owned_refs.
     * Slots are NULL-initialised so error paths can free agg_results even
     * before all slots are populated. */
    ray_t* agg_results = ray_alloc(n_grps * sizeof(ray_t*));
    if (!agg_results) {
        scratch_free(g_icols_hdr); ray_release(gid_vec); ray_release(dvals); ray_release(grouped);
        scratch_free(ic_hdr);
        return ray_error("oom", NULL);
    }
    agg_results->type = RAY_LIST;
    agg_results->len = n_grps;
    ray_t** ar = (ray_t**)ray_data(agg_results);
    memset(ar, 0, (size_t)n_grps * sizeof(ray_t*));

    /* Counting-sort rows by gid: O(nrows + n_grps) vs the previous
     * O(nrows * n_grps) double-scan per group. */
    ray_t* off_hdr = ray_alloc((size_t)(n_grps + 1) * sizeof(int64_t));
    if (!off_hdr) {
        ray_free(agg_results); scratch_free(g_icols_hdr); ray_release(gid_vec); ray_release(dvals); ray_release(grouped);
        scratch_free(ic_hdr);
        return ray_error("oom", NULL);
    }
    int64_t* offs = (int64_t*)ray_data(off_hdr);
    memset(offs, 0, (size_t)(n_grps + 1) * sizeof(int64_t));
    for (int64_t r = 0; r < nrows; r++) {
        int64_t g = gids[r];
        if (g >= 0) offs[g + 1]++;
    }
    for (int64_t gi = 0; gi < n_grps; gi++) offs[gi + 1] += offs[gi];

    ray_t* sorted_hdr = ray_alloc((size_t)nrows * sizeof(int64_t));
    if (!sorted_hdr) {
        ray_free(off_hdr);
        ray_free(agg_results); scratch_free(g_icols_hdr); ray_release(gid_vec); ray_release(dvals); ray_release(grouped);
        scratch_free(ic_hdr);
        return ray_error("oom", NULL);
    }
    int64_t* sorted = (int64_t*)ray_data(sorted_hdr);
    /* Write-cursor array derived from offs. */
    ray_t* wcur_hdr = ray_alloc((size_t)n_grps * sizeof(int64_t));
    if (!wcur_hdr) {
        ray_free(sorted_hdr); ray_free(off_hdr);
        ray_free(agg_results); scratch_free(g_icols_hdr); ray_release(gid_vec); ray_release(dvals); ray_release(grouped);
        scratch_free(ic_hdr);
        return ray_error("oom", NULL);
    }
    int64_t* wcur = (int64_t*)ray_data(wcur_hdr);
    memcpy(wcur, offs, (size_t)n_grps * sizeof(int64_t));
    for (int64_t r = 0; r < nrows; r++) {
        int64_t g = gids[r];
        if (g >= 0) sorted[wcur[g]++] = r;
    }
    ray_free(wcur_hdr);

    for (int64_t gi = 0; gi < n_grps; gi++) {
        int64_t cnt = offs[gi + 1] - offs[gi];
        ray_t* subset = gather_by_idx(vcol, sorted + offs[gi], cnt);
        if (RAY_IS_ERR(subset)) {
            ray_free(sorted_hdr); ray_free(off_hdr);
            ray_free(agg_results); scratch_free(g_icols_hdr); ray_release(gid_vec); ray_release(dvals); ray_release(grouped);
            scratch_free(ic_hdr);
            return subset;
        }
        ray_t* agg_val = call_fn1(agg_fn, subset);
        ray_release(subset);
        /* The lambda result may be lazy (e.g. (fn [xs] (sum xs)) returns a
         * lazy chain).  Materialise before storing into agg_results so
         * list_to_typed_vec sees a concrete scalar. */
        if (ray_is_lazy(agg_val))
            agg_val = ray_lazy_materialize(agg_val);
        if (RAY_IS_ERR(agg_val)) {
            ray_free(sorted_hdr); ray_free(off_hdr);
            ray_free(agg_results); scratch_free(g_icols_hdr); ray_release(gid_vec); ray_release(dvals); ray_release(grouped);
            scratch_free(ic_hdr);
            return agg_val;
        }
        ar[gi] = agg_val;
    }
    ray_free(sorted_hdr);
    ray_free(off_hdr);
    ray_release(gid_vec);

    /* Unstack: collect distinct index keys, build wide result.
     * Map each group to (ix_idx, pv_idx). */
    ray_t* ix_list = ray_list_new(16);
    ray_t* gmap = ray_alloc(n_grps * 2 * sizeof(int64_t));
    int64_t* gm_ix = (int64_t*)ray_data(gmap);
    int64_t* gm_pv = gm_ix + n_grps;

    for (int64_t gi = 0; gi < n_grps; gi++) {
        /* Find pivot index */
        int a1 = 0;
        ray_t* pv = collection_elem(g_pcol, gi, &a1);
        gm_pv[gi] = -1;
        for (int64_t p = 0; p < n_pv; p++) {
            int a2 = 0;
            ray_t* dv = collection_elem(dvals, p, &a2);
            bool eq = atom_eq(pv, dv);
            if (a2) ray_release(dv);
            if (eq) { gm_pv[gi] = p; break; }
        }
        if (a1) ray_release(pv);

        /* Find or insert index key */
        gm_ix[gi] = -1;
        int64_t n_ix = ray_len(ix_list);
        ray_t** ix_items = (ray_t**)ray_data(ix_list);
        for (int64_t j = 0; j < n_ix; j++) {
            ray_t** ex = (ray_t**)ray_data(ix_items[j]);
            bool match = true;
            for (int64_t ci = 0; ci < n_idx && match; ci++) {
                int a2 = 0;
                ray_t* v = collection_elem(g_icols[ci], gi, &a2);
                if (!atom_eq(ex[ci], v)) match = false;
                if (a2) ray_release(v);
            }
            if (match) { gm_ix[gi] = j; break; }
        }
        if (gm_ix[gi] < 0) {
            gm_ix[gi] = ray_len(ix_list);
            ray_t* tup = ray_list_new((int32_t)n_idx);
            for (int64_t ci = 0; ci < n_idx; ci++) {
                int a2 = 0;
                ray_t* v = collection_elem(g_icols[ci], gi, &a2);
                if (!a2) ray_retain(v);
                tup = ray_list_append(tup, v);
                ray_release(v);
            }
            ix_list = ray_list_append(ix_list, tup);
            ray_release(tup);
        }
    }
    scratch_free(g_icols_hdr); /* last use of g_icols was inside this loop */

    int64_t n_ix = ray_len(ix_list);

    /* Build result table */
    ray_t* result = ray_table_new(n_idx + n_pv);
    if (RAY_IS_ERR(result)) goto fb_cleanup;

    /* Index columns */
    { ray_t** ix_items = (ray_t**)ray_data(ix_list);
    for (int64_t ci = 0; ci < n_idx; ci++) {
        ray_t* col_vals = ray_list_new((int32_t)n_ix);
        for (int64_t r = 0; r < n_ix; r++) {
            ray_t* v = ((ray_t**)ray_data(ix_items[r]))[ci];
            ray_retain(v);
            col_vals = ray_list_append(col_vals, v);
            ray_release(v);
        }
        ray_t* col_vec = list_to_typed_vec(col_vals, icols[ci]->type);
        if (RAY_IS_ERR(col_vec)) { ray_release(result); result = col_vec; goto fb_cleanup; }
        result = ray_table_add_col(result, idx_syms[ci], col_vec);
        ray_release(col_vec);
        if (RAY_IS_ERR(result)) goto fb_cleanup;
    }
    }

    /* Value columns */
    for (int64_t p = 0; p < n_pv; p++) {
        int8_t agg_type = RAY_I64;
        for (int64_t gi = 0; gi < n_grps; gi++) {
            if (gm_pv[gi] != p) continue;
            if (ar[gi]->type == -RAY_F64) { agg_type = RAY_F64; break; }
        }

        ray_t* col_vals = ray_list_new((int32_t)n_ix);
        for (int64_t r = 0; r < n_ix; r++) {
            ray_t* nullv = ray_typed_null((int8_t)-agg_type);
            col_vals = ray_list_append(col_vals, nullv);
            ray_release(nullv);
        }

        for (int64_t gi = 0; gi < n_grps; gi++) {
            if (gm_pv[gi] != p) continue;
            ray_t** cv = (ray_t**)ray_data(col_vals);
            ray_release(cv[gm_ix[gi]]);
            ray_retain(ar[gi]);
            cv[gm_ix[gi]] = ar[gi];
        }

        ray_t* agg_vec = list_to_typed_vec(col_vals, agg_type);
        if (RAY_IS_ERR(agg_vec)) { ray_release(result); result = agg_vec; goto fb_cleanup; }

        /* Column name */
        int a1 = 0;
        ray_t* pval = collection_elem(dvals, p, &a1);
        int64_t col_sym;
        if (pval->type == -RAY_SYM) {
            col_sym = pval->i64;
        } else if (pval->type == -RAY_I64) {
            char buf[64]; int len = snprintf(buf, sizeof(buf), "%" PRId64, pval->i64);
            col_sym = ray_sym_intern(buf, (size_t)len);
        } else if (pval->type == -RAY_F64) {
            double fv = clear_neg_zero(pval->f64);
            char buf[64]; int len = snprintf(buf, sizeof(buf), "%g", fv);
            col_sym = ray_sym_intern(buf, (size_t)len);
        } else if (pval->type == -RAY_BOOL) {
            col_sym = ray_sym_intern(pval->b8 ? "true" : "false", pval->b8 ? 4 : 5);
        } else {
            char buf[64]; int len = snprintf(buf, sizeof(buf), "col%" PRId64, pval->i64);
            col_sym = ray_sym_intern(buf, (size_t)len);
        }
        if (a1) ray_release(pval);

        result = ray_table_add_col(result, col_sym, agg_vec);
        ray_release(agg_vec);
        if (RAY_IS_ERR(result)) goto fb_cleanup;
    }

fb_cleanup:
    ray_free(gmap);
    ray_release(ix_list);
    /* agg_results is a RAY_LIST that owns each ar[gi]; ray_free walks
     * its children via ray_release_owned_refs, so a manual loop here
     * would double-release every slot and corrupt the slab freelist. */
    ray_free(agg_results);
    ray_release(dvals);
    ray_release(grouped);
    scratch_free(ic_hdr); /* idx_syms/icols consolidated carve — last read
                            * just above (icols[ci]->type / idx_syms[ci]) */
    return result;
}

/* (pivot table index_col pivot_col value_col agg_fn) — pivot table.
 * Thin wrapper: collapse -0.0 -> +0.0 in an F64 pivot key ONCE up front (so
 * the DAG fast path and the generic fallback both group 0.0/-0.0 together),
 * then run the pivot on the normalized table. */
ray_t* ray_pivot_fn(ray_t** args, int64_t n) {
    if (n != 5) return ray_error("arity", "pivot expects 5 arguments: table, index, pivot-col, value-col, agg-fn");
    ray_t* tbl            = args[0];
    ray_t* pivot_col_name = args[2];
    if (tbl->type == RAY_TABLE && pivot_col_name->type == -RAY_SYM) {
        ray_t* wt = pivot_normalize_f64_key(tbl, pivot_col_name->i64);
        if (wt) {
            if (RAY_IS_ERR(wt)) return wt;
            ray_t* res = pivot_fn_impl(wt, args[1], pivot_col_name, args[3], args[4]);
            ray_release(wt);
            return res;
        }
    }
    return pivot_fn_impl(tbl, args[1], pivot_col_name, args[3], args[4]);
}

/* ══════════════════════════════════════════
 * table schema utilities
 * ══════════════════════════════════════════ */

static ray_t* table_sym_arg(ray_t* arg, const char* fn,
                            int64_t** out_syms, int64_t* out_n) {
    *out_syms = NULL;
    *out_n = 0;
    if (!arg || RAY_IS_ERR(arg))
        return arg ? arg : ray_error("type", "%s: symbol list is null", fn);

    int64_t n = 0;
    if (arg->type == -RAY_SYM) {
        n = 1;
    } else if (arg->type == RAY_SYM || arg->type == RAY_LIST) {
        n = arg->len;
    } else {
        return ray_error("type", "%s: expected a symbol or symbol list, got %s",
                         fn, ray_type_name(arg->type));
    }

    int64_t* syms = (int64_t*)ray_alloc_raw((size_t)(n > 0 ? n : 1) * sizeof(int64_t));
    if (!syms) return ray_error("oom", NULL);

    if (arg->type == -RAY_SYM) {
        syms[0] = arg->i64;
    } else {
        for (int64_t i = 0; i < n; i++) {
            int alloc = 0;
            ray_t* e = collection_elem(arg, i, &alloc);
            if (!e || RAY_IS_ERR(e)) {
                ray_free_raw(syms);
                return e ? e : ray_error("type", "%s: invalid symbol at index %lld",
                                         fn, (long long)i);
            }
            if (e->type != -RAY_SYM) {
                int8_t et = e->type;
                if (alloc) ray_release(e);
                ray_free_raw(syms);
                return ray_error("type", "%s: expected symbol at index %lld, got %s",
                                 fn, (long long)i, ray_type_name(et));
            }
            syms[i] = e->i64;
            if (alloc) ray_release(e);
        }
    }

    *out_syms = syms;
    *out_n = n;
    return NULL;
}

static bool table_sym_in_list(const int64_t* xs, int64_t n, int64_t sym) {
    for (int64_t i = 0; i < n; i++)
        if (xs[i] == sym) return true;
    return false;
}

static ray_t* table_row_key(ray_t** key_cols, int64_t nkeys, int64_t row) {
    if (nkeys == 1) {
        int alloc = 0;
        ray_t* k = collection_elem(key_cols[0], row, &alloc);
        if (!k || RAY_IS_ERR(k)) return k ? k : ray_error("type", "xkey: invalid key cell");
        if (!alloc) ray_retain(k);
        return k;
    }

    ray_t* k = ray_list_new(nkeys);
    if (!k || RAY_IS_ERR(k)) return k ? k : ray_error("oom", NULL);
    for (int64_t c = 0; c < nkeys; c++) {
        int alloc = 0;
        ray_t* cell = collection_elem(key_cols[c], row, &alloc);
        if (!cell || RAY_IS_ERR(cell)) {
            ray_release(k);
            return cell ? cell : ray_error("type", "xkey: invalid key cell");
        }
        ray_t* nk = ray_list_append(k, cell);
        if (alloc) ray_release(cell);
        if (!nk || RAY_IS_ERR(nk)) {
            ray_release(k);
            return nk ? nk : ray_error("oom", NULL);
        }
        k = nk;
    }
    return k;
}

static ray_t* table_row_value_dict(ray_t* tbl, const int64_t* key_syms,
                                   int64_t nkeys, int64_t row) {
    int64_t ncols = ray_table_ncols(tbl);
    int64_t nvals = 0;
    for (int64_t c = 0; c < ncols; c++)
        if (!table_sym_in_list(key_syms, nkeys, ray_table_col_name(tbl, c)))
            nvals++;

    ray_t* keys = ray_sym_vec_new(RAY_SYM_W64, nvals);
    if (!keys || RAY_IS_ERR(keys)) return keys ? keys : ray_error("oom", NULL);
    ray_t* vals = ray_list_new(nvals);
    if (!vals || RAY_IS_ERR(vals)) { ray_release(keys); return vals ? vals : ray_error("oom", NULL); }

    for (int64_t c = 0; c < ncols; c++) {
        int64_t name = ray_table_col_name(tbl, c);
        if (table_sym_in_list(key_syms, nkeys, name)) continue;
        ray_t* nkeys_obj = ray_vec_append(keys, &name);
        if (!nkeys_obj || RAY_IS_ERR(nkeys_obj)) {
            ray_release(keys);
            ray_release(vals);
            return nkeys_obj ? nkeys_obj : ray_error("oom", NULL);
        }
        keys = nkeys_obj;
        ray_t* col = ray_table_get_col_idx(tbl, c);
        int alloc = 0;
        ray_t* cell = collection_elem(col, row, &alloc);
        if (!cell || RAY_IS_ERR(cell)) {
            ray_release(keys);
            ray_release(vals);
            return cell ? cell : ray_error("type", "xkey: invalid value cell");
        }
        ray_t* nvals_obj = ray_list_append(vals, cell);
        if (alloc) ray_release(cell);
        if (!nvals_obj || RAY_IS_ERR(nvals_obj)) {
            ray_release(keys);
            ray_release(vals);
            return nvals_obj ? nvals_obj : ray_error("oom", NULL);
        }
        vals = nvals_obj;
    }
    return ray_dict_new(keys, vals);
}

/* (cols table) — return column names. */
ray_t* ray_cols_fn(ray_t* tbl) {
    if (!tbl || tbl->type != RAY_TABLE)
        return ray_error("type", "cols expects a table, got %s",
                         tbl ? ray_type_name(tbl->type) : "null");
    return ray_key_fn(tbl);
}

/* (xcol table names) — rename all columns, preserving order and data. */
ray_t* ray_xcol_fn(ray_t* tbl, ray_t* names) {
    if (!tbl || tbl->type != RAY_TABLE)
        return ray_error("type", "xcol: first arg must be a table, got %s",
                         tbl ? ray_type_name(tbl->type) : "null");
    int64_t* syms = NULL;
    int64_t n = 0;
    ray_t* err = table_sym_arg(names, "xcol", &syms, &n);
    if (err) return err;

    int64_t ncols = ray_table_ncols(tbl);
    if (n != ncols) {
        ray_free_raw(syms);
        return ray_error("length", "xcol: expected %lld names, got %lld",
                         (long long)ncols, (long long)n);
    }

    ray_t* out = ray_table_new(ncols);
    if (!out || RAY_IS_ERR(out)) { ray_free_raw(syms); return out ? out : ray_error("oom", NULL); }
    for (int64_t c = 0; c < ncols; c++) {
        ray_t* col = ray_table_get_col_idx(tbl, c);
        out = ray_table_add_col(out, syms[c], col);
        if (!out || RAY_IS_ERR(out)) { ray_free_raw(syms); return out ? out : ray_error("oom", NULL); }
    }
    ray_free_raw(syms);
    return out;
}

/* (xcols table cols) — project/reorder columns exactly as requested. */
ray_t* ray_xcols_fn(ray_t* tbl, ray_t* cols) {
    if (!tbl || tbl->type != RAY_TABLE)
        return ray_error("type", "xcols: first arg must be a table, got %s",
                         tbl ? ray_type_name(tbl->type) : "null");
    int64_t* syms = NULL;
    int64_t n = 0;
    ray_t* err = table_sym_arg(cols, "xcols", &syms, &n);
    if (err) return err;

    ray_t* out = ray_table_new(n);
    if (!out || RAY_IS_ERR(out)) { ray_free_raw(syms); return out ? out : ray_error("oom", NULL); }
    for (int64_t i = 0; i < n; i++) {
        ray_t* col = ray_table_get_col(tbl, syms[i]);
        if (!col) {
            ray_release(out);
            ray_free_raw(syms);
            return ray_error("domain", "xcols: unknown table column");
        }
        out = ray_table_add_col(out, syms[i], col);
        if (!out || RAY_IS_ERR(out)) { ray_free_raw(syms); return out ? out : ray_error("oom", NULL); }
    }
    ray_free_raw(syms);
    return out;
}

/* (xkey table keys) — dictionary projection keyed by one or more columns. */
ray_t* ray_xkey_fn(ray_t* tbl, ray_t* keys_arg) {
    if (!tbl || tbl->type != RAY_TABLE)
        return ray_error("type", "xkey: first arg must be a table, got %s",
                         tbl ? ray_type_name(tbl->type) : "null");
    int64_t* key_syms = NULL;
    int64_t nkeys = 0;
    ray_t* err = table_sym_arg(keys_arg, "xkey", &key_syms, &nkeys);
    if (err) return err;
    if (nkeys <= 0) { ray_free_raw(key_syms); return ray_error("domain", "xkey: at least one key column is required"); }

    ray_t** key_cols = (ray_t**)ray_alloc_raw((size_t)nkeys * sizeof(ray_t*));
    if (!key_cols) { ray_free_raw(key_syms); return ray_error("oom", NULL); }
    for (int64_t k = 0; k < nkeys; k++) {
        key_cols[k] = ray_table_get_col(tbl, key_syms[k]);
        if (!key_cols[k]) {
            ray_free_raw(key_cols);
            ray_free_raw(key_syms);
            return ray_error("domain", "xkey: unknown key column");
        }
    }

    int64_t nrows = ray_table_nrows(tbl);
    ray_t* keys = ray_list_new(nrows);
    ray_t* vals = ray_list_new(nrows);
    if (!keys || RAY_IS_ERR(keys) || !vals || RAY_IS_ERR(vals)) {
        if (keys && !RAY_IS_ERR(keys)) ray_release(keys);
        if (vals && !RAY_IS_ERR(vals)) ray_release(vals);
        ray_free_raw(key_cols);
        ray_free_raw(key_syms);
        return keys && RAY_IS_ERR(keys) ? keys : (vals && RAY_IS_ERR(vals) ? vals : ray_error("oom", NULL));
    }

    /* Duplicate detection is an open-addressed set of indices into `keys`.
     * The old all-prior-keys scan made construction quadratic. */
    size_t ht_cap = 16;
    if ((uint64_t)nrows > SIZE_MAX / 2) {
        ray_release(keys); ray_release(vals);
        ray_free_raw(key_cols); ray_free_raw(key_syms);
        return ray_error("limit", "xkey: table is too large");
    }
    while (nrows > 0 && ht_cap < (size_t)nrows * 2) {
        if (ht_cap > SIZE_MAX / 2) {
            ray_release(keys); ray_release(vals);
            ray_free_raw(key_cols); ray_free_raw(key_syms);
            return ray_error("limit", "xkey: table is too large");
        }
        ht_cap *= 2;
    }
    int64_t* ht = (int64_t*)ray_alloc_raw(ht_cap * sizeof(int64_t));
    if (!ht) {
        ray_release(keys); ray_release(vals);
        ray_free_raw(key_cols); ray_free_raw(key_syms);
        return ray_error("oom", NULL);
    }
    for (size_t i = 0; i < ht_cap; i++) ht[i] = -1;

    for (int64_t r = 0; r < nrows; r++) {
        ray_t* k = table_row_key(key_cols, nkeys, r);
        if (!k || RAY_IS_ERR(k)) { ray_free_raw(ht); ray_release(keys); ray_release(vals); ray_free_raw(key_cols); ray_free_raw(key_syms); return k ? k : ray_error("oom", NULL); }
        ray_t** key_items = (ray_t**)ray_data(keys);
        size_t slot = (size_t)ray_atom_hash(k) & (ht_cap - 1);
        while (ht[slot] >= 0) {
            if (atom_eq(key_items[ht[slot]], k)) {
                ray_release(k);
                ray_free_raw(ht);
                ray_release(keys);
                ray_release(vals);
                ray_free_raw(key_cols);
                ray_free_raw(key_syms);
                return ray_error("domain", "xkey: duplicate key at row %lld", (long long)r);
            }
            slot = (slot + 1) & (ht_cap - 1);
        }
        ray_t* v = table_row_value_dict(tbl, key_syms, nkeys, r);
        if (!v || RAY_IS_ERR(v)) { ray_release(k); ray_free_raw(ht); ray_release(keys); ray_release(vals); ray_free_raw(key_cols); ray_free_raw(key_syms); return v ? v : ray_error("oom", NULL); }
        ray_t* nkeys_obj = ray_list_append(keys, k);
        if (!nkeys_obj || RAY_IS_ERR(nkeys_obj)) {
            ray_release(k);
            ray_release(v);
            ray_free_raw(ht);
            ray_release(keys);
            ray_release(vals);
            ray_free_raw(key_cols);
            ray_free_raw(key_syms);
            return nkeys_obj ? nkeys_obj : ray_error("oom", NULL);
        }
        keys = nkeys_obj;
        ht[slot] = keys->len - 1;

        ray_t* nvals_obj = ray_list_append(vals, v);
        if (!nvals_obj || RAY_IS_ERR(nvals_obj)) {
            ray_release(k);
            ray_release(v);
            ray_free_raw(ht);
            ray_release(keys);
            ray_release(vals);
            ray_free_raw(key_cols);
            ray_free_raw(key_syms);
            return nvals_obj ? nvals_obj : ray_error("oom", NULL);
        }
        vals = nvals_obj;
        ray_release(k);
        ray_release(v);
    }

    ray_free_raw(ht);
    ray_free_raw(key_cols);
    ray_free_raw(key_syms);
    return ray_dict_new(keys, vals);
}

/* (xgroup table keys) — dict from key value/tuple to the matching table slice. */
ray_t* ray_xgroup_fn(ray_t* tbl, ray_t* keys_arg) {
    if (!tbl || tbl->type != RAY_TABLE)
        return ray_error("type", "xgroup: first arg must be a table, got %s",
                         tbl ? ray_type_name(tbl->type) : "null");
    int64_t* key_syms = NULL;
    int64_t nkeys = 0;
    ray_t* err = table_sym_arg(keys_arg, "xgroup", &key_syms, &nkeys);
    if (err) return err;
    if (nkeys <= 0) { ray_free_raw(key_syms); return ray_error("domain", "xgroup: at least one key column is required"); }

    ray_t** key_cols = (ray_t**)ray_alloc_raw((size_t)nkeys * sizeof(ray_t*));
    if (!key_cols) { ray_free_raw(key_syms); return ray_error("oom", NULL); }
    for (int64_t k = 0; k < nkeys; k++) {
        key_cols[k] = ray_table_get_col(tbl, key_syms[k]);
        if (!key_cols[k]) {
            ray_free_raw(key_cols);
            ray_free_raw(key_syms);
            return ray_error("domain", "xgroup: unknown key column");
        }
    }

    int64_t nrows = ray_table_nrows(tbl);
    ray_t* keys = ray_list_new(16);
    if (!keys || RAY_IS_ERR(keys)) {
        ray_free_raw(key_cols);
        ray_free_raw(key_syms);
        return keys ? keys : ray_error("oom", NULL);
    }

    int64_t* gids = NULL;
    int64_t* counts = NULL;
    if (nrows > 0) {
        gids = (int64_t*)ray_alloc_raw((size_t)nrows * sizeof(int64_t));
        counts = (int64_t*)ray_alloc_raw((size_t)nrows * sizeof(int64_t));
        if (!gids || !counts) {
            ray_release(keys);
            ray_free_raw(gids);
            ray_free_raw(counts);
            ray_free_raw(key_cols);
            ray_free_raw(key_syms);
            return ray_error("oom", NULL);
        }
    }

    int64_t ngroups = 0;
    for (int64_t r = 0; r < nrows; r++) {
        ray_t* k = table_row_key(key_cols, nkeys, r);
        if (!k || RAY_IS_ERR(k)) {
            ray_release(keys);
            ray_free_raw(gids);
            ray_free_raw(counts);
            ray_free_raw(key_cols);
            ray_free_raw(key_syms);
            return k ? k : ray_error("oom", NULL);
        }

        int64_t gi = -1;
        ray_t** key_items = (ray_t**)ray_data(keys);
        for (int64_t i = 0; i < ngroups; i++) {
            if (atom_eq(key_items[i], k)) { gi = i; break; }
        }

        if (gi < 0) {
            gi = ngroups++;
            ray_t* nkeys_obj = ray_list_append(keys, k);
            if (!nkeys_obj || RAY_IS_ERR(nkeys_obj)) {
                ray_release(k);
                ray_release(keys);
                ray_free_raw(gids);
                ray_free_raw(counts);
                ray_free_raw(key_cols);
                ray_free_raw(key_syms);
                return nkeys_obj ? nkeys_obj : ray_error("oom", NULL);
            }
            keys = nkeys_obj;
            counts[gi] = 0;
        }

        gids[r] = gi;
        counts[gi]++;
        ray_release(k);
    }

    ray_t* vals = ray_list_new(ngroups);
    if (!vals || RAY_IS_ERR(vals)) {
        ray_release(keys);
        ray_free_raw(gids);
        ray_free_raw(counts);
        ray_free_raw(key_cols);
        ray_free_raw(key_syms);
        return vals ? vals : ray_error("oom", NULL);
    }
    for (int64_t i = 0; i < ngroups; i++) {
        ray_t* ridx = ray_vec_new(RAY_I64, counts[i]);
        if (!ridx || RAY_IS_ERR(ridx)) {
            ray_release(keys);
            ray_release(vals);
            ray_free_raw(gids);
            ray_free_raw(counts);
            ray_free_raw(key_cols);
            ray_free_raw(key_syms);
            return ridx ? ridx : ray_error("oom", NULL);
        }
        int64_t* rid = (int64_t*)ray_data(ridx);
        int64_t pos = 0;
        for (int64_t r = 0; r < nrows; r++) {
            if (gids[r] == i) rid[pos++] = r;
        }
        ridx->len = counts[i];

        ray_t* slice = ray_at_fn(tbl, ridx);
        ray_release(ridx);
        if (!slice || RAY_IS_ERR(slice)) {
            ray_release(keys);
            ray_release(vals);
            ray_free_raw(gids);
            ray_free_raw(counts);
            ray_free_raw(key_cols);
            ray_free_raw(key_syms);
            return slice ? slice : ray_error("oom", NULL);
        }
        ray_t* nvals_obj = ray_list_append(vals, slice);
        ray_release(slice);
        if (!nvals_obj || RAY_IS_ERR(nvals_obj)) {
            ray_release(keys);
            ray_release(vals);
            ray_free_raw(gids);
            ray_free_raw(counts);
            ray_free_raw(key_cols);
            ray_free_raw(key_syms);
            return nvals_obj ? nvals_obj : ray_error("oom", NULL);
        }
        vals = nvals_obj;
    }

    ray_free_raw(gids);
    ray_free_raw(counts);
    ray_free_raw(key_cols);
    ray_free_raw(key_syms);
    return ray_dict_new(keys, vals);
}

/* (fkeys table) — linked-column metadata as col->target dict. */
ray_t* ray_fkeys_fn(ray_t* tbl) {
    if (!tbl || tbl->type != RAY_TABLE)
        return ray_error("type", "fkeys expects a table, got %s",
                         tbl ? ray_type_name(tbl->type) : "null");
    int64_t ncols = ray_table_ncols(tbl);
    int64_t count = 0;
    for (int64_t c = 0; c < ncols; c++) {
        ray_t* col = ray_table_get_col_idx(tbl, c);
        if (ray_link_has(col)) count++;
    }

    ray_t* keys = ray_sym_vec_new(RAY_SYM_W64, count);
    if (!keys || RAY_IS_ERR(keys)) return keys ? keys : ray_error("oom", NULL);
    ray_t* vals = ray_list_new(count);
    if (!vals || RAY_IS_ERR(vals)) { ray_release(keys); return vals ? vals : ray_error("oom", NULL); }

    for (int64_t c = 0; c < ncols; c++) {
        ray_t* col = ray_table_get_col_idx(tbl, c);
        if (!ray_link_has(col)) continue;
        int64_t name = ray_table_col_name(tbl, c);
        int64_t target = ray_link_target_id(col);
        ray_t* nkeys_obj = ray_vec_append(keys, &name);
        if (!nkeys_obj || RAY_IS_ERR(nkeys_obj)) {
            ray_release(keys);
            ray_release(vals);
            return nkeys_obj ? nkeys_obj : ray_error("oom", NULL);
        }
        keys = nkeys_obj;
        ray_t* target_sym = ray_sym(target);
        if (!target_sym || RAY_IS_ERR(target_sym)) {
            ray_release(keys);
            ray_release(vals);
            return target_sym ? target_sym : ray_error("oom", NULL);
        }
        ray_t* nvals_obj = ray_list_append(vals, target_sym);
        ray_release(target_sym);
        if (!nvals_obj || RAY_IS_ERR(nvals_obj)) {
            ray_release(keys);
            ray_release(vals);
            return nvals_obj ? nvals_obj : ray_error("oom", NULL);
        }
        vals = nvals_obj;
    }
    return ray_dict_new(keys, vals);
}

/* ══════════════════════════════════════════
 * modify
 * ══════════════════════════════════════════ */

/* (modify tbl col_name fn) — apply fn to the named column, return new table */
ray_t* ray_modify_fn(ray_t** args, int64_t n) {
    if (n < 3) return ray_error("arity", "modify expects 3 arguments: table, column, function, got %lld", (long long)n);
    ray_t* tbl = args[0];
    ray_t* col_name = args[1];
    ray_t* fn = args[2];

    if (tbl->type != RAY_TABLE)
        return ray_error("type", "modify: first arg must be a table, got %s", ray_type_name(tbl->type));
    if (col_name->type != -RAY_SYM)
        return ray_error("type", "modify: second arg must be a symbol, got %s", ray_type_name(col_name->type));

    int64_t target_sym = col_name->i64;
    ray_t* col = ray_table_get_col(tbl, target_sym);
    if (!col) return ray_error("domain", "modify: column not found");

    /* Apply fn to the entire column vector (atomic fns will map element-wise) */
    ray_t* new_col = call_fn1(fn, col);
    if (RAY_IS_ERR(new_col)) return new_col;

    /* Build new table: copy all columns, replacing the target */
    int64_t ncols = ray_table_ncols(tbl);
    ray_t* result = ray_table_new(ncols);
    if (RAY_IS_ERR(result)) { ray_release(new_col); return result; }

    for (int64_t i = 0; i < ncols; i++) {
        int64_t cname = ray_table_col_name(tbl, i);
        ray_t* cvec = (cname == target_sym) ? new_col : ray_table_get_col_idx(tbl, i);
        result = ray_table_add_col(result, cname, cvec);
        if (RAY_IS_ERR(result)) { ray_release(new_col); return result; }
    }
    ray_release(new_col);
    return result;
}

/* ══════════════════════════════════════════
 * alter
 * ══════════════════════════════════════════ */

/* Cleanup helper for alter set's ray_cow failure paths.  Releases the
 * caller's retain on the original vec plus the eval'd args, then returns
 * an owning RAY_ERROR — substituting "oom" when ray_cow itself returned
 * NULL (RAY_IS_ERR(NULL) is false, but the caller still owes us a
 * structured error to propagate).
 *
 * Exposed (non-static) so test code can pin the contract directly:
 * NULL cow_result must produce an "oom" error, RAY_ERROR cow_result
 * passes through, and `original_var` is released exactly once. */
ray_t* ray_alter_set_cow_fail(ray_t* original_var, ray_t* cow_result,
                              ray_t* idx, ray_t* val, ray_t* name_sym) {
    ray_release(original_var);
    if (idx)      ray_release(idx);
    if (val)      ray_release(val);
    if (name_sym) ray_release(name_sym);
    return cow_result ? cow_result : ray_error("oom", NULL);
}

ray_t* ray_alter_fn(ray_t** args, int64_t n) {
    if (n < 3) return ray_error("domain", "alter: expects at least 3 arguments (var, op, ...), got %lld", (long long)n);
    /* First arg: evaluate to get the symbol */
    ray_t* name_sym = ray_eval(args[0]);
    if (!name_sym || RAY_IS_ERR(name_sym)) return name_sym ? name_sym : ray_error("type", "alter: variable name argument evaluated to null");
    if (name_sym->type != -RAY_SYM) { int8_t nt = name_sym->type; ray_release(name_sym); return ray_error("type", "alter: variable name must be a symbol, got %s", ray_type_name(nt)); }

    /* Resolve the variable */
    ray_t* var = ray_env_get(name_sym->i64);
    if (!var) { ray_release(name_sym); return ray_error("name", NULL); }

    /* Second arg: operation name (unevaluated, must be a name) */
    ray_t* op = args[1];
    if (!op || op->type != -RAY_SYM) { ray_release(name_sym); return ray_error("type", "alter: op must be a symbol (set, concat, or remove), got %s", op ? ray_type_name(op->type) : "null"); }
    ray_t* op_name = ray_sym_str(op->i64);
    if (!op_name) { ray_release(name_sym); return ray_error("domain", "alter: unknown op symbol"); }
    const char* oname = ray_str_ptr(op_name);
    size_t olen = ray_str_len(op_name);

    if (olen == 3 && memcmp(oname, "set", 3) == 0) {
        /* (alter 'v set idx val) — idx can be scalar or vector of indices */
        ray_release(op_name);
        if (n < 4) { ray_release(name_sym); return ray_error("domain", "alter set: expects 4 arguments (var, set, idx, val), got %lld", (long long)n); }
        ray_t* idx = ray_eval(args[2]);
        if (!idx || RAY_IS_ERR(idx)) { ray_release(name_sym); return idx ? idx : ray_error("type", "alter set: index argument evaluated to null"); }
        ray_t* val = ray_eval(args[3]);
        if (!val || RAY_IS_ERR(val)) { ray_release(idx); ray_release(name_sym); return val ? val : ray_error("type", "alter set: value argument evaluated to null"); }
        if (!ray_is_vec(var) && var->type != RAY_LIST) { int8_t vt = var->type; ray_release(idx); ray_release(val); ray_release(name_sym); return ray_error("type", "alter set: target must be a vector or list, got %s", ray_type_name(vt)); }

        /* For LIST types, build a new list with replaced elements */
        if (var->type == RAY_LIST) {
            int64_t vlen = ray_len(var);
            ray_t** elems = (ray_t**)ray_data(var);
            ray_t* new_list = ray_alloc(vlen * sizeof(ray_t*));
            if (!new_list) { ray_release(idx); ray_release(val); ray_release(name_sym); return ray_error("oom", NULL); }
            new_list->type = RAY_LIST;
            new_list->len = vlen;
            ray_t** out = (ray_t**)ray_data(new_list);
            for (int64_t i = 0; i < vlen; i++) { ray_retain(elems[i]); out[i] = elems[i]; }

            if (ray_is_atom(idx) && is_numeric(idx)) {
                int64_t i = as_i64(idx);
                if (i >= 0 && i < vlen) { ray_release(out[i]); ray_retain(val); out[i] = val; }
            } else if (ray_is_vec(idx)) {
                int64_t nidx = idx->len;
                for (int64_t k = 0; k < nidx; k++) {
                    int alloc = 0;
                    ray_t* ie = collection_elem(idx, k, &alloc);
                    int64_t i = as_i64(ie);
                    if (alloc) ray_release(ie);
                    if (i >= 0 && i < vlen) { ray_release(out[i]); ray_retain(val); out[i] = val; }
                }
            }
            ray_release(idx); ray_release(val);
            ray_env_set(name_sym->i64, new_list);
            ray_release(new_list);
            /* In-place amend returns the symbol, not the altered value. */
            return name_sym;
        }

        /* `var` came from ray_env_get as a BORROWED ref.  ray_cow's
         * contract is "I take your owning ref; I give you back a ref"
         * — so calling it on a borrow over-decrements the env's
         * binding when the rc>1 copy path fires (releasing v drops
         * env's count from N to N-1; if some other env binding also
         * pointed at v, that binding now sees an extra under-retain
         * and risks UAF when later replaced).
         *
         * Retain up-front so the ref we hand to ray_cow is genuinely
         * ours.  Track the original pointer so the cow-OOM path
         * (alloc_copy fails before ray_cow's release would have run)
         * can still release the retain — without that, OOM leaks the
         * extra ref. */
        ray_t* original_var = var;
        ray_retain(var);
        /* `alter` amends in place, and ray_cow only copies what is shared —
         * but the retain above is unconditional, so ray_cow's own `rc == 1`
         * test could never be true here and the copy path fired on every
         * call.  A one-element write copied the whole vector, and the cost
         * of an alter scaled with the vector's length rather than with the
         * write.  Sole ownership after our retain is `rc == 2`: the
         * binding's reference and ours.  The other three exclusions keep the
         * previous behaviour exactly — ray_cow returns an arena block as-is,
         * and a slice or a mapped block is not ours to write through. */
        /* A dotted name is NOT a candidate.  ray_env_get resolves `d.x` to the
         * leaf living inside the container, which typically holds the only
         * reference to it — so the leaf passes an rc test that says nothing
         * about whether the CONTAINER is shared.  Writing through it would be
         * seen by every alias of the dict or table, and env_set_dotted's
         * COW-rebuild of the chain cannot undo a write that already
         * happened.  The copy is what made that safe. */
        bool sole = !ray_sym_is_dotted(name_sym->i64) &&
                    var->rc == 2 && var->mmod == 0 &&
                    !(var->attrs & (RAY_ATTR_SLICE | RAY_ATTR_ARENA));
        ray_t* cow_result = sole ? var : ray_cow(var);
        /* ray_cow returns NULL when ray_alloc_copy returned NULL (heap
         * exhaustion past RAY_HEAP_MAX_ORDER) and a RAY_ERROR pointer
         * when alloc_copy hit its own len-overflow guard.  Both leave
         * the input ref untouched, so the cleanup helper releases
         * `original_var` and either propagates the error pointer or
         * synthesizes an "oom" RAY_ERROR for the NULL case (test code
         * pins both branches). */
        if (!cow_result || RAY_IS_ERR(cow_result)) {
            return ray_alter_set_cow_fail(original_var, cow_result, idx, val, name_sym);
        }
        var = cow_result;
        /* A value written in place can break the order a `sorted` marker
         * promises (consumers trust it without re-checking). */
        var->attrs &= (uint8_t)~RAY_ATTR_SORTED;

        /* Validate idx shape + (for the atom case) bounds BEFORE we
         * touch any state.  The accelerator-index drop below would
         * otherwise outlive a failed write. */
        bool idx_is_atom_num = ray_is_atom(idx) && is_numeric(idx);
        bool idx_is_vec      = ray_is_vec(idx);
        if (!idx_is_atom_num && !idx_is_vec) {
            int8_t it = idx->type;
            ray_release(var);
            ray_release(idx); ray_release(val); ray_release(name_sym);
            return ray_error("type", "alter set: index must be a numeric atom or a vector, got %s", ray_type_name(it));
        }
        if (idx_is_atom_num) {
            int64_t i_check = as_i64(idx);
            if (i_check < 0 || i_check >= var->len) {
                ray_release(var);
                ray_release(idx); ray_release(val); ray_release(name_sym);
                return ray_error("index", NULL);
            }
        }

        /* alter's set path writes via store_typed_elem, which bypasses
         * ray_vec_set's mutation guard.  Now that we know the write
         * will reach the data array, drop any attached accelerator
         * index so it can't outlive the mutation. */
        if (var->attrs & RAY_ATTR_HAS_INDEX) {
            ray_t* drop_r = ray_index_drop(&var);
            if (RAY_IS_ERR(drop_r)) {
                ray_release(var);
                ray_release(idx); ray_release(val); ray_release(name_sym);
                return drop_r;
            }
        }

        if (idx_is_atom_num) {
            /* Single index — bounds already validated above. */
            int64_t i = as_i64(idx);
            ray_release(idx);
            store_typed_elem(var, i, val);
        } else {
            /* Vector of indices — set each to val.
             * If val is a vector of same length, set pairwise.
             * If val is scalar or shorter, broadcast. */
            int64_t nidx = idx->len;
            int val_is_vec = ray_is_vec(val) && val->len == nidx;
            for (int64_t k = 0; k < nidx; k++) {
                int alloc = 0;
                ray_t* ie = collection_elem(idx, k, &alloc);
                int64_t i = as_i64(ie);
                if (alloc) ray_release(ie);
                if (i < 0 || i >= var->len) continue;
                if (val_is_vec) {
                    int va = 0;
                    ray_t* ve = collection_elem(val, k, &va);
                    store_typed_elem(var, i, ve);
                    if (va) ray_release(ve);
                } else {
                    store_typed_elem(var, i, val);
                }
            }
            ray_release(idx);
        }
        ray_release(val);
        ray_env_set(name_sym->i64, var);
        /* The retain-first at the top of the set path gave us an owning
         * ref to var; ray_env_set retained its own for the binding, so
         * drop ours.  In-place amend returns the symbol, not the value. */
        ray_release(var);
        return name_sym;
    }
    if (olen == 6 && memcmp(oname, "concat", 6) == 0) {
        /* (alter 'v concat val) */
        ray_release(op_name);
        if (n < 3) { ray_release(name_sym); return ray_error("arity", "alter concat: expects 3 arguments (var, concat, val), got %lld", (long long)n); }
        ray_t* val = ray_eval(args[2]);
        if (!val || RAY_IS_ERR(val)) { ray_release(name_sym); return val ? val : ray_error("type", "alter concat: value argument evaluated to null"); }
        ray_t* new_vec = ray_concat_fn(var, val);
        ray_release(val);
        if (RAY_IS_ERR(new_vec)) { ray_release(name_sym); return new_vec; }
        ray_env_set(name_sym->i64, new_vec);
        ray_release(new_vec);
        /* In-place amend returns the symbol, not the altered value. */
        return name_sym;
    }
    if (olen == 6 && memcmp(oname, "remove", 6) == 0) {
        /* (alter 'v remove idx) — remove element(s) at index/indices */
        ray_release(op_name);
        if (n < 3) { ray_release(name_sym); return ray_error("arity", "alter remove: expects 3 arguments (var, remove, idx), got %lld", (long long)n); }
        ray_t* idx = ray_eval(args[2]);
        if (!idx || RAY_IS_ERR(idx)) { ray_release(name_sym); return idx ? idx : ray_error("type", "alter remove: index argument evaluated to null"); }

        if (!var || var->type != RAY_LIST) {
            ray_release(idx); ray_release(name_sym);
            return ray_error("type", "alter remove: target must be a list, got %s", var ? ray_type_name(var->type) : "null");
        }

        int64_t vlen = ray_len(var);
        ray_t** elems = (ray_t**)ray_data(var);

        /* Build a set of indices to remove */
        int64_t remove_idx[256];
        int64_t nremove = 0;
        if (ray_is_atom(idx) && is_numeric(idx)) {
            remove_idx[0] = as_i64(idx);
            nremove = 1;
        } else if (ray_is_vec(idx)) {
            nremove = idx->len;
            if (nremove > 256) { ray_release(idx); ray_release(name_sym); return ray_error("limit", NULL); }
            for (int64_t i = 0; i < nremove; i++) {
                int alloc = 0;
                ray_t* e = collection_elem(idx, i, &alloc);
                remove_idx[i] = as_i64(e);
                if (alloc) ray_release(e);
            }
        } else {
            int8_t it = idx->type;
            ray_release(idx); ray_release(name_sym);
            return ray_error("type", "alter remove: index must be a numeric atom or a vector, got %s", ray_type_name(it));
        }
        ray_release(idx);

        /* Build new list without the removed indices */
        int64_t new_len = vlen;
        for (int64_t i = 0; i < nremove; i++)
            if (remove_idx[i] >= 0 && remove_idx[i] < vlen) new_len--;

        ray_t* new_list = ray_alloc(new_len * sizeof(ray_t*));
        if (!new_list) { ray_release(name_sym); return ray_error("oom", NULL); }
        new_list->type = RAY_LIST;
        new_list->len = new_len;
        ray_t** out = (ray_t**)ray_data(new_list);
        int64_t j = 0;
        for (int64_t i = 0; i < vlen; i++) {
            int skip = 0;
            for (int64_t k = 0; k < nremove; k++)
                if (remove_idx[k] == i) { skip = 1; break; }
            if (!skip) {
                ray_retain(elems[i]);
                out[j++] = elems[i];
            }
        }
        new_list->len = j;
        ray_env_set(name_sym->i64, new_list);
        ray_release(new_list);
        /* In-place amend returns the symbol, not the altered value. */
        return name_sym;
    }
    ray_release(op_name);
    ray_release(name_sym);
    return ray_error("domain", "alter: unknown op, expected set, concat, or remove");
}

/* ══════════════════════════════════════════
 * del
 * ══════════════════════════════════════════ */

/* (del name) — delete variable from environment (special form, unevaluated arg) */
ray_t* ray_del_fn(ray_t** args, int64_t n) {
    if (n < 1) return ray_error("arity", "del expects 1 argument, got %lld", (long long)n);
    ray_t* name = args[0];
    if (name->type != -RAY_SYM)
        return ray_error("type", "del expects a symbol, got %s", ray_type_name(name->type));
    /* Propagate ray_env_set's failure: silently ignoring the return
     * value would let `(del .sys.gc)` appear to succeed while leaving
     * the builtin intact — a confusing lie.  Emit a precise message
     * per error code rather than blaming every failure on the
     * reserved-namespace guard (OOM on dotted-path upsert, for
     * example, is not a reserve error). */
    ray_err_t err = ray_env_set(name->i64, NULL);
    if (err == RAY_OK) return ray_i64(0);
    const char* nm = ray_str_ptr(ray_sym_str(name->i64));
    if (err == RAY_ERR_RESERVED)
        return ray_error("reserve",
                         "cannot delete reserved binding '%s'", nm);
    return ray_error(ray_err_code_str(err),
                     "del '%s' failed", nm);
}

/* ══════════════════════════════════════════
 * row
 * ══════════════════════════════════════════ */

/* (row table idx) — extract a single row from a table as a dict */
ray_t* ray_row_fn(ray_t* tbl, ray_t* idx) {
    if (tbl->type != RAY_TABLE) return ray_error("type", "row expects a table, got %s", ray_type_name(tbl->type));
    if (!is_numeric(idx)) return ray_error("type", "row index must be a numeric atom, got %s", ray_type_name(idx->type));
    /* Delegate to at — it already handles table integer indexing */
    return ray_at_fn(tbl, idx);
}

/* ══════════════════════════════════════════
 * union-all
 * ══════════════════════════════════════════ */

/* (union-all t1 t2) — concatenate two tables row-wise (same schema) */
ray_t* ray_union_all_fn(ray_t* t1, ray_t* t2) {
    if (t1->type != RAY_TABLE)
        return ray_error("type", "union-all: first arg must be a table, got %s", ray_type_name(t1->type));
    if (t2->type != RAY_TABLE)
        return ray_error("type", "union-all: second arg must be a table, got %s", ray_type_name(t2->type));

    int64_t ncols = ray_table_ncols(t1);
    if (ncols != ray_table_ncols(t2))
        return ray_error("type", "union-all: tables must have same number of columns, got %lld and %lld", (long long)ncols, (long long)ray_table_ncols(t2));

    /* Validate matching column names */
    for (int64_t c = 0; c < ncols; c++) {
        if (ray_table_col_name(t1, c) != ray_table_col_name(t2, c))
            return ray_error("type", "union-all: column names must match at column %lld", (long long)c);
    }

    ray_t* result = ray_table_new(ncols);
    if (!result || RAY_IS_ERR(result)) return result;

    for (int64_t c = 0; c < ncols; c++) {
        int64_t name_id = ray_table_col_name(t1, c);
        ray_t* col1 = ray_table_get_col_idx(t1, c);
        ray_t* col2 = ray_table_get_col_idx(t2, c);

        if (!col1 || !col2) {
            ray_release(result);
            return ray_error("domain", "union-all: missing column at index %lld", (long long)c);
        }

        ray_t* combined = ray_vec_concat(col1, col2);
        if (!combined || RAY_IS_ERR(combined)) {
            ray_release(result);
            return combined ? combined : ray_error("oom", NULL);
        }

        result = ray_table_add_col(result, name_id, combined);
        ray_release(combined);
        if (!result || RAY_IS_ERR(result)) return result;
    }

    return result;
}

/* ══════════════════════════════════════════
 * table-distinct
 * ══════════════════════════════════════════ */

/* (table-distinct t) — remove duplicate rows via DAG group-by */
ray_t* ray_table_distinct_fn(ray_t* tbl) {
    if (tbl->type != RAY_TABLE)
        return ray_error("type", "table-distinct expects a table, got %s", ray_type_name(tbl->type));

    int64_t ncols = ray_table_ncols(tbl);
    if (ncols == 0) { ray_retain(tbl); return tbl; }

    ray_graph_t* g = ray_graph_new(tbl);
    if (!g) return ray_error("oom", NULL);

    /* Exact-size carve (cut-3): ray_distinct's n_keys parameter is uint32_t
     * (already widened, so ncols itself is the only limit), and v2's 0-agg
     * GROUP now serves this shape at any width (agg_v2_can_handle no longer
     * declines n_aggs == 0 — see agg_engine.c) — the old fixed keys[255] plus
     * its ncols > 255 range error retire with it. */
    ray_t* keys_hdr;
    ray_op_t** keys = (ray_op_t**)scratch_alloc(&keys_hdr, (size_t)ncols * sizeof(ray_op_t*));
    if (!keys) { ray_graph_free(g); return ray_error("oom", NULL); }

    for (int64_t c = 0; c < ncols; c++) {
        int64_t name_id = ray_table_col_name(tbl, c);
        ray_t* name_str = ray_sym_str(name_id);
        if (!name_str) { scratch_free(keys_hdr); ray_graph_free(g); return ray_error("domain", "table-distinct: column name symbol not found at index %lld", (long long)c); }
        keys[c] = ray_scan(g, ray_str_ptr(name_str));
        if (!keys[c]) { scratch_free(keys_hdr); ray_graph_free(g); return ray_error("oom", NULL); }
    }

    ray_op_t* root = ray_distinct(g, keys, (uint32_t)ncols);
    scratch_free(keys_hdr);
    if (!root) { ray_graph_free(g); return ray_error("oom", NULL); }

    ray_t* result = ray_execute(g, root);
    ray_graph_free(g);
    return result;
}

/* ══════════════════════════════════════════
 * unify
 * ══════════════════════════════════════════ */

/* (unify a b) — return a 2-element list holding a and b.  Same-type inputs and
 * atoms are wrapped as-is; different vector types are wrapped WITHOUT coercion
 * to a common type (callers that need a shared type convert the operands
 * themselves).  See test/rfl/table/tblop.rfl for the covered branches. */
ray_t* ray_unify_fn(ray_t* a, ray_t* b) {
    /* Build a 2-element list containing both values */
    ray_t* result = ray_list_new(2);
    if (RAY_IS_ERR(result)) return result;

    if (a->type == b->type || ray_is_atom(a) || ray_is_atom(b)) {
        /* Same type or atoms: return as-is */
        ray_retain(a); ray_retain(b);
        result = ray_list_append(result, a); ray_release(a);
        result = ray_list_append(result, b); ray_release(b);
        return result;
    }

    /* Different vector types: wrapped as-is, no coercion to a common type. */
    ray_retain(a); ray_retain(b);
    result = ray_list_append(result, a); ray_release(a);
    result = ray_list_append(result, b); ray_release(b);
    return result;
}