rayforce-sys 1.4.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.
 */

/**   Shared helpers for exec.c split — included by expr.c, filter.c, join.c, etc.
 *   Small hot-path helpers are static inline; larger functions that remain in
 *   exec.c are declared extern.
 */

#ifndef RAY_EXEC_INTERNAL_H
#define RAY_EXEC_INTERNAL_H

#if !defined(RAY_OS_WINDOWS) && !defined(_GNU_SOURCE)
#define _GNU_SOURCE
#endif

#include "exec.h"
#include "core/pool.h"
#include "core/profile.h"
#include "core/runtime.h"   /* ray_err_code — expr_eval_err_fatal */
#include "store/colrange.h"
#include "store/csr.h"
#include "store/hnsw.h"
#include "lftj.h"
#include "mem/heap.h"
#include "table/sym.h"
#include "table/table.h"
#include "vec/str.h"
#include "vec/vec.h"
#include <string.h>
#include <math.h>
#include <stdbool.h>
#include <stdint.h>

/* Shared strict parser for launcher flags and .sys.args.  When core_count is
 * true, values are limited to the heap registry's total worker capacity. */
int parse_sys_int_arg(const char* s, bool nonnegative, bool core_count,
                      int64_t* out);

/* True if every leaf is constant or reduced to a scalar. */
bool op_tree_is_scalar(ray_graph_t* g, uint32_t id);
bool op_tree_is_atom(ray_graph_t* g, uint32_t id);

/* Borrow only while the symbol table stays read-only through the dispatch
 * barrier. Workers compare or copy source-domain codes without interning.
 * File-domain lookup keeps its own publication/lifetime contract. */
typedef struct {
    ray_t** strings;
    uint32_t count;
} ray_group_sym_view_t;
static inline ray_t* ray_group_sym_read(const ray_group_sym_view_t* view, struct ray_sym_domain_s* domain, int64_t id) {
    if (domain == ray_sym_runtime_domain())
        return id >= 0 && (uint64_t)id < view->count ? view->strings[id] : NULL;
    return ray_sym_domain_str(domain, id);
}


#include <stdio.h>
#include <stdlib.h>
#include <float.h>
#include <ctype.h>

/* ══════════════════════════════════════════
 * Parted segment helpers
 * ══════════════════════════════════════════ */

/* Output attrs for a flat vector gathered/flattened from parted segments:
 * the MAX SYM index width across segments (width bits only).  Partitions
 * legitimately carry different widths — each partition's width tracked
 * the shared vocabulary size at ITS save time, so partitions written
 * after the vocabulary crossed the 255/65535 boundaries are wider than
 * older ones.  The flat output must hold the widest segment's ids. */
static inline uint8_t parted_sym_max_attrs(ray_t** segs, int64_t n_segs) {
    uint8_t w = 0;
    for (int64_t i = 0; i < n_segs; i++)
        if (segs[i] && (segs[i]->attrs & RAY_SYM_W_MASK) > w)
            w = segs[i]->attrs & RAY_SYM_W_MASK;
    return w;
}

/* Copy n cells from flat source vec `src` [src_off..src_off+n) into the raw
 * output buffer `out_data` at [dst_off..), where the output was created
 * with element type `base` and attrs `out_attrs`.  Equal element size
 * (always true for non-SYM bases, and for SYM segments saved at the
 * output's width) takes the memcpy fast path.  A SYM source at a
 * different index width takes the per-cell conversion loop — ids are
 * positions in the shared domain, so widening preserves them verbatim.
 * The loop only runs for genuinely mixed-width parted tables. */
static inline void parted_copy_cells(void* out_data, int8_t base,
                                     uint8_t out_attrs, int64_t dst_off,
                                     ray_t* src, int64_t src_off, int64_t n) {
    uint8_t out_esz = ray_sym_elem_size(base, out_attrs);
    if (base != RAY_SYM ||
        ((src->attrs ^ out_attrs) & RAY_SYM_W_MASK) == 0) {
        memcpy((char*)out_data + (size_t)dst_off * out_esz,
               (const char*)ray_data(src) + (size_t)src_off * out_esz,
               (size_t)n * out_esz);
        return;
    }
    const void* src_data = ray_data(src);
    for (int64_t i = 0; i < n; i++)
        ray_write_sym(out_data, dst_off + i,
                      (uint64_t)ray_read_sym(src_data, src_off + i,
                                             RAY_SYM, src->attrs),
                      RAY_SYM, out_attrs);
}

/* ══════════════════════════════════════════
 * Global profiler
 * ══════════════════════════════════════════ */

extern ray_profile_t g_ray_profile;

/* ══════════════════════════════════════════
 * Arena-based scratch allocation helpers
 * ══════════════════════════════════════════ */

/* Allocate zero-initialized scratch buffer, returns data pointer.
 * *hdr_out receives the ray_t* header for later ray_free(). */
static inline void* scratch_calloc(ray_t** hdr_out, size_t nbytes) {
    ray_t* h = ray_alloc(nbytes);
    if (!h) { *hdr_out = NULL; return NULL; }
    void* p = ray_data(h);
    memset(p, 0, nbytes);
    *hdr_out = h;
    return p;
}

/* Allocate uninitialized scratch buffer. */
static inline void* scratch_alloc(ray_t** hdr_out, size_t nbytes) {
    ray_t* h = ray_alloc(nbytes);
    if (!h) { *hdr_out = NULL; return NULL; }
    *hdr_out = h;
    return ray_data(h);
}

/* Reallocate: alloc new, copy old, free old. Returns new data pointer. */
static inline void* scratch_realloc(ray_t** hdr_out, size_t old_bytes, size_t new_bytes) {
    ray_t* old_h = *hdr_out;
    ray_t* new_h = ray_alloc(new_bytes);
    if (!new_h) return NULL;
    void* new_p = ray_data(new_h);
    if (old_h) {
        memcpy(new_p, ray_data(old_h), old_bytes < new_bytes ? old_bytes : new_bytes);
        ray_free(old_h);
    }
    *hdr_out = new_h;
    return new_p;
}

/* Free a scratch buffer (NULL-safe). */
static inline void scratch_free(ray_t* hdr) {
    if (!hdr) return;
    ray_free(hdr);
}

/* ══════════════════════════════════════════
 * Safe sym intern
 * ══════════════════════════════════════════ */

/* Safe sym intern for constant column names in graph algorithm result tables.
 * Falls back to 0 on failure (column name interning should never fail for
 * short constant strings unless ray_sym_init failed). */
static inline int64_t sym_intern_safe(const char* s, size_t len) {
    int64_t id = ray_sym_intern(s, len);
    return id >= 0 ? id : 0;
}

/* ══════════════════════════════════════════
 * Unified column read/write helpers
 * ══════════════════════════════════════════ */

static inline int64_t read_col_i64(const void* data, int64_t row,
                                    int8_t type, uint8_t attrs) {
    switch (type) {
    case RAY_I64: case RAY_TIMESPAN: case RAY_TIMESTAMP:
        return ((const int64_t*)data)[row];
    case RAY_SYM:
        switch (attrs & RAY_SYM_W_MASK) {
        case RAY_SYM_W8:  return (int64_t)((const uint8_t*)data)[row];
        case RAY_SYM_W16: return (int64_t)((const uint16_t*)data)[row];
        case RAY_SYM_W32: return (int64_t)((const uint32_t*)data)[row];
        default:         return ((const int64_t*)data)[row];
        }
    case RAY_I32: case RAY_DATE: case RAY_TIME:
        return (int64_t)((const int32_t*)data)[row];
    case RAY_I16:
        return (int64_t)((const int16_t*)data)[row];
    case RAY_F32: {
        /* The 4-byte lane's bits, zero-extended, with -0.0 canonicalised
         * to +0.0 the way the F64 group-key readers do (group_key_f64_bits,
         * #407): a group hash table stores and compares these bits, so the
         * two zeros must read alike.  Bit-level, as there, because release
         * builds use -fno-signed-zeros.  Before this arm F32 fell to the
         * one-byte default below, and a multi-core group by an F32 key
         * hashed a byte of each value as the key (#760). */
        uint32_t u; memcpy(&u, (const float*)data + row, 4);
        if (u == 0x80000000u) u = 0;
        return (int64_t)u;
    }
    default: /* RAY_BOOL, RAY_U8 */
        return (int64_t)((const uint8_t*)data)[row];
    }
}

static inline void write_col_i64(void* data, int64_t row, int64_t val,
                                  int8_t type, uint8_t attrs) {
    switch (type) {
    case RAY_I64: case RAY_TIMESPAN: case RAY_TIMESTAMP:
        ((int64_t*)data)[row] = val; return;
    case RAY_SYM:
        ray_write_sym(data, row, (uint64_t)val, type, attrs); return;
    case RAY_I32: case RAY_DATE: case RAY_TIME:
        ((int32_t*)data)[row] = (int32_t)val; return;
    case RAY_I16:
        ((int16_t*)data)[row] = (int16_t)val; return;
    case RAY_F32: {
        uint32_t u = (uint32_t)val; memcpy((float*)data + row, &u, 4); return;
    }
    default: /* RAY_BOOL, RAY_U8 */
        ((uint8_t*)data)[row] = (uint8_t)val; return;
    }
}

/* ══════════════════════════════════════════
 * RAY_SYM-aware column helpers
 * ══════════════════════════════════════════ */

static inline uint8_t col_esz(const ray_t* col) {
    return ray_sym_elem_size(col->type, col->attrs);
}

/* Fast key reader for DA/sort hot loops: elem_size is pre-computed and
 * loop-invariant, so the switch is always perfectly predicted.  Avoids the
 * ray_read_sym → type dispatch chain (3+ branches per element).
 *
 * Unsigned widening — appropriate for SYM dictionary IDs (which are
 * unsigned) and for packing keys into composite ints where bit-pattern
 * round-trip matters more than sign.  Do NOT use for arithmetic on
 * signed narrow column values (I16/I32/DATE/TIME) — use
 * `read_signed_by_esz` so a stored -1 reads as -1 and not 65535. */
static inline int64_t read_by_esz(const void* data, int64_t row, uint8_t esz) {
    switch (esz) {
    case 1:  return (int64_t)((const uint8_t*)data)[row];
    case 2:  return (int64_t)((const uint16_t*)data)[row];
    case 4:  return (int64_t)((const uint32_t*)data)[row];
    default: return ((const int64_t*)data)[row];
    }
}

/* Sign-aware reader.  Aggregate kernels (SUM/MIN/MAX/AVG) need the
 * stored value's mathematical magnitude, so signed narrow columns
 * must be sign-extended into the int64 result.  Pass `is_unsigned=1`
 * for U8/BOOL or SYM dict ID reads, `0` for I16/I32/I64 and the
 * temporal types (DATE/TIME/TIMESTAMP) which are stored signed. */
static inline int64_t read_signed_by_esz(const void* data, int64_t row,
                                         uint8_t esz, int is_unsigned) {
    if (is_unsigned) {
        switch (esz) {
        case 1:  return (int64_t)((const uint8_t*)data)[row];
        case 2:  return (int64_t)((const uint16_t*)data)[row];
        case 4:  return (int64_t)((const uint32_t*)data)[row];
        default: return ((const int64_t*)data)[row];
        }
    }
    switch (esz) {
    case 1:  return (int64_t)((const int8_t*)data)[row];
    case 2:  return (int64_t)((const int16_t*)data)[row];
    case 4:  return (int64_t)((const int32_t*)data)[row];
    default: return ((const int64_t*)data)[row];
    }
}

static inline ray_t* col_vec_new(const ray_t* src, int64_t cap) {
    if (src->type == RAY_SYM)
        return ray_sym_vec_new(src->attrs & RAY_SYM_W_MASK, cap);
    return ray_vec_new(src->type, cap);
}

/* Propagate str_pool from source to gathered result.
 * Source may be a slice — resolve to owner's pool. */
static inline void col_propagate_str_pool(ray_t* dst, const ray_t* src) {
    if (src->type != RAY_STR || dst->type != RAY_STR) return;
    const ray_t* owner = (src->attrs & RAY_ATTR_SLICE) ? src->slice_parent : src;
    if (owner->str_pool) {
        if (dst->str_pool) ray_release(dst->str_pool);
        ray_retain(owner->str_pool);
        dst->str_pool = owner->str_pool;
    }
}

/* Propagate str_pool from parted segments to gathered result.
 * All segments must share the same pool for memcpy-gathered results
 * to be valid. For multi-pool cases, callers must use the deep-copy
 * gather path (parted_gather_str_col) instead. */
static inline void col_propagate_str_pool_parted(ray_t* dst, ray_t** segs, int64_t n_segs) {
    if (dst->type != RAY_STR) return;
    for (int64_t i = 0; i < n_segs; i++) {
        if (segs[i] && segs[i]->type == RAY_STR && segs[i]->str_pool) {
            col_propagate_str_pool(dst, segs[i]);
            return;
        }
    }
}

/* ---- Null bitmap propagation helpers ---- */

/* Propagate nulls from src to dst via index array: dst[r] gets src's null
 * bit at indices[r].  indices may contain -1 for LEFT/OUTER join fill rows
 * (those are set null unconditionally). */
static inline void col_propagate_nulls_gather(ray_t* dst, const ray_t* src,
                                               const int64_t* indices,
                                               int64_t count) {
    /* Two loops, the may-have-nulls test hoisted out: this runs once per
     * row per gathered column, and with the test inside, inlined into a
     * large caller (sel_compact), the loop lost its registers and cost ~8%
     * on where-heavy group queries. */
    if (!ray_vec_may_have_nulls(src)) {
        for (int64_t r = 0; r < count; r++)
            if (indices[r] < 0) ray_vec_set_null(dst, r, true);
        return;
    }
    for (int64_t r = 0; r < count; r++) {
        int64_t i = indices[r];
        if (i < 0 || ray_vec_is_null((ray_t*)src, i))
            ray_vec_set_null(dst, r, true);
    }
}

/* Propagate nulls from src[src_off..src_off+count) to dst[dst_off..),
 * for contiguous range copies (HEAD, TAIL, range inserts). */
static inline void col_propagate_nulls_range(ray_t* dst, int64_t dst_off,
                                              const ray_t* src, int64_t src_off,
                                              int64_t count) {
    if (!ray_vec_may_have_nulls(src)) return;
    for (int64_t i = 0; i < count; i++) {
        if (ray_vec_is_null((ray_t*)src, src_off + i))
            ray_vec_set_null(dst, dst_off + i, true);
    }
}

/* Propagate nulls through a boolean filter mask: for each set bit in
 * mask[0..src_len), copy the null bit from src to dst[out_idx++]. */
static inline void col_propagate_nulls_filter(ray_t* dst, const ray_t* src,
                                               const uint8_t* mask,
                                               int64_t src_len) {
    if (!ray_vec_may_have_nulls(src)) return;
    int64_t out = 0;
    for (int64_t i = 0; i < src_len; i++) {
        if (mask[i]) {
            if (ray_vec_is_null((ray_t*)src, i))
                ray_vec_set_null(dst, out, true);
            out++;
        }
    }
}

/* ── Ragged parted tables ─────────────────────────────────────────────────
 * A partition may lack a trailing column (store/part.c): that column's
 * segment for it is NULL and stands for the partition's rows as typed
 * nulls, which is how ser and the formatter read it.  The partition's row
 * count lives in the table's partition-key (MAPCOMMON) column, so every
 * reader that flattens or gathers a parted column takes those counts and a
 * NULL segment contributes its rows as nulls rather than none.  Without
 * counts (a column met away from its table) a NULL segment contributes no
 * rows, as before. */
static inline const int64_t* parted_table_counts(ray_t* tbl, int64_t* n_out) {
    *n_out = 0;
    if (!tbl || RAY_IS_ERR(tbl) || tbl->type != RAY_TABLE) return NULL;
    int64_t nc = ray_table_ncols(tbl);
    for (int64_t c = 0; c < nc; c++) {
        /* metadata first: an encoded column stays encoded; MAPCOMMON is plain */
        ray_col_meta_t m;
        if (!ray_table_col_meta_idx(tbl, c, &m) || m.type != RAY_MAPCOMMON) continue;
        ray_t* col = ray_table_col_idx(tbl, c);
        if (col && !RAY_IS_ERR(col) && col->type == RAY_MAPCOMMON) {
            ray_t* rc = ((ray_t**)ray_data(col))[1];   /* not-a-table */
            if (!rc || RAY_IS_ERR(rc)) return NULL;
            *n_out = rc->len;
            return (const int64_t*)ray_data(rc);
        }
    }
    return NULL;
}

/* Rows segment s stands for: its length, or a NULL segment's partition count. */
static inline int64_t parted_seg_rows(ray_t** segs, int64_t s,
                                      const int64_t* cnt, int64_t ncnt) {
    if (segs[s] && !RAY_IS_ERR(segs[s])) return segs[s]->len;
    return (cnt && s < ncnt) ? cnt[s] : 0;
}

/* Rows of a parted column, NULL segments counted by the partition counts. */
static inline int64_t parted_rows_counted(ray_t* col, const int64_t* cnt, int64_t ncnt) {
    ray_t** segs = (ray_t**)ray_data(col);
    int64_t total = 0;
    for (int64_t s = 0; s < col->len; s++) total += parted_seg_rows(segs, s, cnt, ncnt);
    return total;
}

/* Rows [off, off+n) of a flat fixed-width or SYM vector (len already
 * covering them) as the type's null; BOOL/U8 have none and get 0. */
static inline void parted_fill_nulls(ray_t* flat, int64_t off, int64_t n) {
    if (n <= 0) return;
    if (flat->type == RAY_BOOL || flat->type == RAY_U8) {
        memset((char*)ray_data(flat) + off, 0, (size_t)n);
        return;
    }
    for (int64_t i = 0; i < n; i++) ray_vec_set_null(flat, off + i, true);
}

/* Append one string element from a parted segment, preserving nulls. */
static inline ray_t* parted_str_append_elem(ray_t* out, ray_t* seg,
                                            int64_t local_idx,
                                            const char* pool_base) {
    if (ray_vec_may_have_nulls(seg) && ray_vec_is_null(seg, local_idx)) {
        out = ray_str_vec_append(out, "", 0);
        if (!RAY_IS_ERR(out))
            ray_vec_set_null(out, out->len - 1, true);
    } else {
        ray_str_t* elems = (ray_str_t*)ray_data(seg);
        const char* str = ray_str_t_ptr(&elems[local_idx], pool_base);
        out = ray_str_vec_append(out, str, elems[local_idx].len);
    }
    return out;
}

/* Deep-copy gather from parted RAY_STR segments by row index.
 * Resolves each string from its source segment's pool and appends
 * into the output vector's own pool. Safe for multi-pool segments. */
static inline ray_t* parted_gather_str_rows_c(ray_t** segs, int64_t n_segs,
                                              const int64_t* row_indices,
                                              int64_t count,
                                              const int64_t* cnt, int64_t ncnt) {
    /* Build prefix-sum segment boundaries (a NULL segment spans its
     * partition's rows) */
    int64_t cumul = 0;
    int64_t stack_ends[64];
    int64_t* seg_ends = (n_segs <= 64) ? stack_ends : NULL;
    ray_t* ends_hdr = NULL;
    if (!seg_ends) {
        seg_ends = (int64_t*)scratch_alloc(&ends_hdr, (size_t)n_segs * sizeof(int64_t));
        if (!seg_ends) return ray_error("oom", NULL);
    }
    for (int64_t i = 0; i < n_segs; i++) {
        cumul += parted_seg_rows(segs, i, cnt, ncnt);
        seg_ends[i] = cumul;
    }

    ray_t* out = ray_vec_new(RAY_STR, count);
    if (!out || RAY_IS_ERR(out)) { if (ends_hdr) scratch_free(ends_hdr); return out; }

    int64_t seg = 0;
    for (int64_t i = 0; i < count; i++) {
        int64_t row = row_indices[i];
        while (seg < n_segs - 1 && row >= seg_ends[seg]) seg++;
        if (!segs[seg]) {
            out = ray_str_vec_append(out, "", 0);
            if (!RAY_IS_ERR(out))
                ray_vec_set_null(out, out->len - 1, true);
        } else {
            int64_t seg_start = (seg > 0) ? seg_ends[seg - 1] : 0;
            int64_t local = row - seg_start;
            const char* pool_base = segs[seg]->str_pool
                                  ? (const char*)ray_data(segs[seg]->str_pool) : NULL;
            out = parted_str_append_elem(out, segs[seg], local, pool_base);
        }
        if (RAY_IS_ERR(out)) { if (ends_hdr) scratch_free(ends_hdr); return out; }
    }
    if (ends_hdr) scratch_free(ends_hdr);
    return out;
}

static inline ray_t* parted_gather_str_rows(ray_t** segs, int64_t n_segs,
                                            const int64_t* row_indices,
                                            int64_t count) {
    return parted_gather_str_rows_c(segs, n_segs, row_indices, count, NULL, 0);
}

/* First n rows from parted RAY_STR segments; a NULL segment's rows (its
 * partition count) are null strings. */
static inline ray_t* parted_head_str_c(ray_t** segs, int64_t n_segs, int64_t n,
                                       const int64_t* cnt, int64_t ncnt) {
    ray_t* out = ray_vec_new(RAY_STR, n);
    if (!out || RAY_IS_ERR(out)) return out;
    int64_t remaining = n;
    for (int64_t s = 0; s < n_segs && remaining > 0; s++) {
        int64_t seg_len = parted_seg_rows(segs, s, cnt, ncnt);
        int64_t take = seg_len < remaining ? seg_len : remaining;
        if (!segs[s] || RAY_IS_ERR(segs[s])) {
            for (int64_t i = 0; i < take; i++) {
                out = ray_str_vec_append(out, "", 0);
                if (RAY_IS_ERR(out)) return out;
                ray_vec_set_null(out, out->len - 1, true);
            }
        } else {
            const char* pool_base = segs[s]->str_pool
                                  ? (const char*)ray_data(segs[s]->str_pool) : NULL;
            for (int64_t i = 0; i < take; i++) {
                out = parted_str_append_elem(out, segs[s], i, pool_base);
                if (RAY_IS_ERR(out)) return out;
            }
        }
        remaining -= take;
    }
    return out;
}
static inline ray_t* parted_head_str(ray_t** segs, int64_t n_segs, int64_t n) {
    return parted_head_str_c(segs, n_segs, n, NULL, 0);
}

/* Last n rows from parted RAY_STR segments; a NULL segment's rows (its
 * partition count) are null strings. */
static inline ray_t* parted_tail_str_c(ray_t** segs, int64_t n_segs, int64_t n,
                                       const int64_t* cnt, int64_t ncnt) {
    int64_t total = 0;
    for (int64_t s = 0; s < n_segs; s++) total += parted_seg_rows(segs, s, cnt, ncnt);
    int64_t skip = total - n;
    if (skip < 0) { skip = 0; n = total; }
    ray_t* out = ray_vec_new(RAY_STR, n);
    if (!out || RAY_IS_ERR(out)) return out;
    int64_t skipped = 0;
    for (int64_t s = 0; s < n_segs; s++) {
        int64_t seg_len = parted_seg_rows(segs, s, cnt, ncnt);
        int64_t seg_start = 0;
        if (skipped + seg_len <= skip) { skipped += seg_len; continue; }
        if (skipped < skip) { seg_start = skip - skipped; skipped = skip; }
        if (!segs[s] || RAY_IS_ERR(segs[s])) {
            for (int64_t i = seg_start; i < seg_len; i++) {
                out = ray_str_vec_append(out, "", 0);
                if (RAY_IS_ERR(out)) return out;
                ray_vec_set_null(out, out->len - 1, true);
            }
        } else {
            const char* pool_base = segs[s]->str_pool
                                  ? (const char*)ray_data(segs[s]->str_pool) : NULL;
            for (int64_t i = seg_start; i < seg_len; i++) {
                out = parted_str_append_elem(out, segs[s], i, pool_base);
                if (RAY_IS_ERR(out)) return out;
            }
        }
        skipped += seg_len;
    }
    return out;
}
static inline ray_t* parted_tail_str(ray_t** segs, int64_t n_segs, int64_t n) {
    return parted_tail_str_c(segs, n_segs, n, NULL, 0);
}

/* Deep-copy flatten all rows from parted RAY_STR segments; a NULL segment
 * yields its partition's rows as null (empty) strings. */
static inline ray_t* parted_flatten_str_c(ray_t** segs, int64_t n_segs,
                                          const int64_t* cnt, int64_t ncnt) {
    int64_t total = 0;
    for (int64_t s = 0; s < n_segs; s++) total += parted_seg_rows(segs, s, cnt, ncnt);
    ray_t* out = ray_vec_new(RAY_STR, total);
    if (!out || RAY_IS_ERR(out)) return out;
    for (int64_t s = 0; s < n_segs; s++) {
        if (!segs[s] || RAY_IS_ERR(segs[s])) {
            int64_t n = parted_seg_rows(segs, s, cnt, ncnt);
            for (int64_t i = 0; i < n; i++) {
                out = ray_str_vec_append(out, "", 0);
                if (RAY_IS_ERR(out)) return out;
                ray_vec_set_null(out, out->len - 1, true);
            }
            continue;
        }
        if (segs[s]->len <= 0) continue;
        const char* pool_base = segs[s]->str_pool
                              ? (const char*)ray_data(segs[s]->str_pool) : NULL;
        for (int64_t i = 0; i < segs[s]->len; i++) {
            out = parted_str_append_elem(out, segs[s], i, pool_base);
            if (RAY_IS_ERR(out)) return out;
        }
    }
    return out;
}

/* Deep-copy flatten all rows from parted RAY_STR segments. */
static inline ray_t* parted_flatten_str(ray_t** segs, int64_t n_segs, int64_t total) {
    (void)total;   /* a capacity hint; the segments give the exact count */
    return parted_flatten_str_c(segs, n_segs, NULL, 0);
}

/* ── LIST analogues of the parted STR helpers ────────────────────────────
 * A LIST cell is a ray_t* — every copy below SHARES elements via
 * ray_retain, never byte-copies: a raw memcpy of the 8-byte pointers
 * would alias the segments' elements without a refcount and double-free
 * on release (issue #355).  NULL segments contribute zero rows, matching
 * parted_gather_col's row numbering. */

/* Flatten all rows from parted RAY_LIST segments; a NULL segment yields its
 * partition's rows as the null object. */
static inline ray_t* parted_flatten_list_c(ray_t** segs, int64_t n_segs,
                                           const int64_t* cnt, int64_t ncnt) {
    int64_t total = 0;
    for (int64_t s = 0; s < n_segs; s++) total += parted_seg_rows(segs, s, cnt, ncnt);
    ray_t* out = ray_list_new(total);
    if (!out || RAY_IS_ERR(out)) return out;
    ray_t** dst = (ray_t**)ray_data(out);
    int64_t pos = 0;
    for (int64_t s = 0; s < n_segs; s++) {
        if (!segs[s] || RAY_IS_ERR(segs[s])) {
            int64_t n = parted_seg_rows(segs, s, cnt, ncnt);
            for (int64_t i = 0; i < n && pos < total; i++) {
                ray_retain(RAY_NULL_OBJ);
                dst[pos++] = RAY_NULL_OBJ;
            }
            continue;
        }
        ray_t** src = (ray_t**)ray_data(segs[s]);
        for (int64_t i = 0; i < segs[s]->len && pos < total; i++) {
            ray_t* e = src[i];
            if (e) ray_retain(e);
            dst[pos++] = e;
        }
    }
    out->len = pos;
    return out;
}

/* Flatten all rows from parted RAY_LIST segments. */
static inline ray_t* parted_flatten_list(ray_t** segs, int64_t n_segs,
                                         int64_t total) {
    (void)total;
    return parted_flatten_list_c(segs, n_segs, NULL, 0);
}

/* First n rows from parted RAY_LIST segments; a NULL segment's rows are the
 * null object. */
static inline ray_t* parted_head_list_c(ray_t** segs, int64_t n_segs, int64_t n,
                                        const int64_t* cnt, int64_t ncnt) {
    ray_t* out = ray_list_new(n);
    if (!out || RAY_IS_ERR(out)) return out;
    ray_t** dst = (ray_t**)ray_data(out);
    int64_t pos = 0;
    for (int64_t s = 0; s < n_segs && pos < n; s++) {
        int64_t seg_len = parted_seg_rows(segs, s, cnt, ncnt);
        if (!segs[s] || RAY_IS_ERR(segs[s])) {
            for (int64_t i = 0; i < seg_len && pos < n; i++) {
                ray_retain(RAY_NULL_OBJ);
                dst[pos++] = RAY_NULL_OBJ;
            }
            continue;
        }
        ray_t** src = (ray_t**)ray_data(segs[s]);
        for (int64_t i = 0; i < seg_len && pos < n; i++) {
            ray_t* e = src[i];
            if (e) ray_retain(e);
            dst[pos++] = e;
        }
    }
    out->len = pos;
    return out;
}
static inline ray_t* parted_head_list(ray_t** segs, int64_t n_segs, int64_t n) {
    return parted_head_list_c(segs, n_segs, n, NULL, 0);
}

/* Last n rows from parted RAY_LIST segments; a NULL segment's rows are the
 * null object. */
static inline ray_t* parted_tail_list_c(ray_t** segs, int64_t n_segs, int64_t n,
                                        const int64_t* cnt, int64_t ncnt) {
    int64_t total = 0;
    for (int64_t s = 0; s < n_segs; s++) total += parted_seg_rows(segs, s, cnt, ncnt);
    int64_t skip = total - n;
    if (skip < 0) { skip = 0; n = total; }
    ray_t* out = ray_list_new(n);
    if (!out || RAY_IS_ERR(out)) return out;
    ray_t** dst = (ray_t**)ray_data(out);
    int64_t pos = 0, skipped = 0;
    for (int64_t s = 0; s < n_segs && pos < n; s++) {
        int64_t seg_len = parted_seg_rows(segs, s, cnt, ncnt);
        int64_t seg_start = 0;
        if (skipped + seg_len <= skip) { skipped += seg_len; continue; }
        if (skipped < skip) { seg_start = skip - skipped; skipped = skip; }
        if (!segs[s] || RAY_IS_ERR(segs[s])) {
            for (int64_t i = seg_start; i < seg_len && pos < n; i++) {
                ray_retain(RAY_NULL_OBJ);
                dst[pos++] = RAY_NULL_OBJ;
            }
        } else {
            ray_t** src = (ray_t**)ray_data(segs[s]);
            for (int64_t i = seg_start; i < seg_len && pos < n; i++) {
                ray_t* e = src[i];
                if (e) ray_retain(e);
                dst[pos++] = e;
            }
        }
        skipped += seg_len;
    }
    out->len = pos;
    return out;
}
static inline ray_t* parted_tail_list(ray_t** segs, int64_t n_segs, int64_t n) {
    return parted_tail_list_c(segs, n_segs, n, NULL, 0);
}

/* Gather rows (global row indices, sorted ascending) from parted RAY_LIST
 * segments — LIST counterpart of parted_gather_str_rows. */
static inline ray_t* parted_gather_list_rows_c(ray_t** segs, int64_t n_segs,
                                               const int64_t* idx, int64_t count,
                                               const int64_t* cnt, int64_t ncnt) {
    ray_t* out = ray_list_new(count);
    if (!out || RAY_IS_ERR(out)) return out;
    ray_t** dst = (ray_t**)ray_data(out);
    int64_t seg = 0;
    int64_t seg_start = 0;
    int64_t seg_end = n_segs > 0 ? parted_seg_rows(segs, 0, cnt, ncnt) : 0;
    for (int64_t j = 0; j < count; j++) {
        int64_t r = idx[j];
        while (seg < n_segs - 1 && r >= seg_end) {
            seg_start = seg_end;
            seg++;
            seg_end += parted_seg_rows(segs, seg, cnt, ncnt);
        }
        ray_t* e = NULL;
        if (segs[seg] && r >= seg_start && r - seg_start < segs[seg]->len)
            e = ((ray_t**)ray_data(segs[seg]))[r - seg_start];
        else if (!segs[seg] && r >= seg_start && r < seg_end)
            e = RAY_NULL_OBJ;   /* a missing segment's row */
        if (e) ray_retain(e);
        dst[j] = e;
    }
    out->len = count;
    return out;
}

static inline ray_t* parted_gather_list_rows(ray_t** segs, int64_t n_segs,
                                             const int64_t* idx,
                                             int64_t count) {
    return parted_gather_list_rows_c(segs, n_segs, idx, count, NULL, 0);
}

/* Filter parted RAY_LIST segments by a boolean predicate over the column's
 * global rows; a NULL segment spans its partition's rows, which pass as
 * the null object. */
static inline ray_t* parted_filter_list_c(ray_t** segs, int64_t n_segs,
                                          const uint8_t* pred, int64_t pass_count,
                                          const int64_t* cnt, int64_t ncnt) {
    ray_t* out = ray_list_new(pass_count);
    if (!out || RAY_IS_ERR(out)) return out;
    ray_t** dst = (ray_t**)ray_data(out);
    int64_t off = 0, pos = 0;
    for (int64_t s = 0; s < n_segs; s++) {
        int64_t seg_len = parted_seg_rows(segs, s, cnt, ncnt);
        if (!segs[s] || RAY_IS_ERR(segs[s])) {
            for (int64_t i = 0; i < seg_len && pos < pass_count; i++) {
                if (!pred[off + i]) continue;
                ray_retain(RAY_NULL_OBJ);
                dst[pos++] = RAY_NULL_OBJ;
            }
            off += seg_len;
            continue;
        }
        ray_t** src = (ray_t**)ray_data(segs[s]);
        for (int64_t i = 0; i < seg_len; i++) {
            if (!pred[off + i]) continue;
            if (pos >= pass_count) break;
            ray_t* e = src[i];
            if (e) ray_retain(e);
            dst[pos++] = e;
        }
        off += seg_len;
    }
    out->len = pos;
    return out;
}
static inline ray_t* parted_filter_list(ray_t** segs, int64_t n_segs,
                                        const uint8_t* pred, int64_t pass_count) {
    return parted_filter_list_c(segs, n_segs, pred, pass_count, NULL, 0);
}

/* True for a parted column whose segments are RAY_LIST vectors — such a
 * column must go through the retaining LIST helpers above, never the
 * byte-copy gather kernels. */
static inline bool parted_base_is_list(const ray_t* col) {
    return RAY_IS_PARTED(col->type) &&
           (int8_t)RAY_PARTED_BASETYPE(col->type) == RAY_LIST;
}

/* Same but from explicit type + attrs (for parted base type, etc.) */
static inline ray_t* typed_vec_new(int8_t type, uint8_t attrs, int64_t cap) {
    if (type == RAY_SYM)
        return ray_sym_vec_new(attrs & RAY_SYM_W_MASK, cap);
    return ray_vec_new(type, cap);
}

/* ══════════════════════════════════════════
 * Cancellation check
 * ══════════════════════════════════════════ */

static inline bool pool_cancelled(ray_pool_t* pool) {
    if (RAY_UNLIKELY(ray_interrupted())) return true;
    return pool && RAY_UNLIKELY(atomic_load_explicit(&pool->cancelled,
                                                     memory_order_relaxed));
}

#define CHECK_CANCEL(pool)                                \
    do { if (pool_cancelled(pool))                        \
             return ray_error("cancel", NULL); } while(0)

#define CHECK_CANCEL_GOTO(pool, lbl)                      \
    do { if (pool_cancelled(pool)) {                      \
             result = ray_error("cancel", NULL);          \
             goto lbl;                                    \
         }                                                \
    } while(0)

/* ══════════════════════════════════════════
 * Graph helpers: ext-node allocation + lookup
 * ══════════════════════════════════════════ */

/* Allocate an ext node (64 B + `extra` trailing bytes for inline arrays).
 * Defined in graph.c.  Used by ray_*-style graph constructors that need
 * to embed variable-length data (keys, agg lists, etc.) after the ext
 * struct.  Use EXT_TRAIL() to find the trailing bytes. */
ray_op_ext_t* graph_alloc_ext_node_ex(ray_graph_t* g, size_t extra);

/* Allocate a plain 32-byte node, growing g->nodes if needed.  Defined in
 * graph.c.  Shared with opt.c (no second copy of the realloc logic). */
ray_op_t* graph_alloc_node(ray_graph_t* g);

/* Compile a Rayfall AST sub-expression to a DAG node.  Defined in
 * query.c.  Used by fused_* operators that need to evaluate a small
 * predicate without going through the full select planner. */
ray_op_t* compile_expr_dag(ray_graph_t* g, ray_t* expr);

/* The rows a `where:` keeps, sampled for the distinct-count route gate
 * (query.c; exposed for the contract tests): the table rows surviving the
 * predicate among CD_GATE_WHERE_SAMPLE sampled ones in *rows, *nrows of
 * them, *population the rows the whole predicate would keep.  NULL (the
 * outputs then describe the whole table) when the predicate names no
 * column the gather covers or keeps every sampled row. */
ray_t* query_cd_gate_where_rows(ray_t* tbl, ray_t* where_expr, int64_t n_table,
                                const int64_t** rows, int64_t* nrows, int64_t* population);

/* Pointer to trailing bytes after an ext node. */
#define EXT_TRAIL(ext) ((char*)((ext) + 1))

static inline ray_op_ext_t* find_ext(ray_graph_t* g, uint32_t node_id) {
    for (uint32_t i = 0; i < g->ext_count; i++) {
        if (g->ext_nodes[i] && g->ext_nodes[i]->base.id == node_id)
            return g->ext_nodes[i];
    }
    return NULL;
}

/* Resolve a node id to its ray_op_t* in g->nodes.  Returns NULL for the
 * RAY_OP_NONE sentinel.  Node ids are stable indices, so the returned
 * pointer is only valid until the next g->nodes realloc — never store it
 * across a node allocation; re-resolve from the id instead. */
static inline ray_op_t* op_node(ray_graph_t* g, uint32_t id) {
    return id == RAY_OP_NONE ? NULL : &g->nodes[id];
}

/* Resolve input edge i (0 or 1) of op to its child node, or NULL if unused. */
static inline ray_op_t* op_child(ray_graph_t* g, const ray_op_t* op, int i) {
    return op_node(g, op->in_id[i]);
}

/* ══════════════════════════════════════════
 * String helpers
 * ══════════════════════════════════════════ */

/* Convert an atom (-RAY_STR or RAY_SYM scalar) to ray_str_t for comparison */
static inline void atom_to_str_t(ray_t* atom, ray_str_t* out, const char** out_pool) {
    const char* sp;
    size_t sl;
    if (atom->type == -RAY_STR) {
        sp = ray_str_ptr(atom);
        sl = ray_str_len(atom);
    } else if (atom->type == RAY_STR) {
        /* Length-1 RAY_STR vector used as scalar */
        if (atom->len < 1) {
            memset(out, 0, sizeof(ray_str_t));
            *out_pool = NULL;
            return;
        }
        /* Resolve slice to parent data — slices have no data of their own,
         * and str_pool shares the union with slice_offset. */
        ray_t* src = atom;
        int64_t idx = 0;
        if (atom->attrs & RAY_ATTR_SLICE) {
            src = atom->slice_parent;
            idx = atom->slice_offset;
        }
        const ray_str_t* elems = (const ray_str_t*)ray_data(src);
        *out = elems[idx];
        *out_pool = src->str_pool ? (const char*)ray_data(src->str_pool) : NULL;
        return;
    } else if (RAY_IS_SYM(atom->type) && ray_is_atom(atom)) {
        /* SAFETY: ray_sym_str returns a borrowed pointer into the append-only
         * sym table.  The pointer is valid for the lifetime of the sym table
         * (i.e., the entire query execution).  If the sym table ever gains
         * eviction, this must retain the returned atom. */
        ray_t* s = ray_sym_str(atom->i64);
        sp = s ? ray_str_ptr(s) : "";
        sl = s ? ray_str_len(s) : 0;
    } else {
        sp = ""; sl = 0;
    }
    memset(out, 0, sizeof(ray_str_t));
    out->len = (uint32_t)sl;
    if (sl <= RAY_STR_INLINE_MAX) {
        if (sl > 0) memcpy(out->data, sp, sl);
        *out_pool = NULL;
    } else {
        memcpy(out->prefix, sp, 4);
        out->pool_off = 0;
        *out_pool = sp; /* point directly at atom's string data */
    }
}

/* Resolve RAY_STR vec to data owner, accounting for slices.
 * Returns element pointer (already offset for slices) and pool pointer. */
static inline void str_resolve(const ray_t* v, const ray_str_t** elems,
                               const char** pool) {
    const ray_t* owner = (v->attrs & RAY_ATTR_SLICE) ? v->slice_parent : v;
    int64_t base = (v->attrs & RAY_ATTR_SLICE) ? v->slice_offset : 0;
    *elems = (const ray_str_t*)ray_data((ray_t*)owner) + base;
    *pool = owner->str_pool ? (const char*)ray_data(owner->str_pool) : NULL;
}

/* Helper: resolve sym/enum element to string.  The id is CELL DATA of
 * `input`, so it resolves through the COLUMN's domain (sym-domain
 * Phase 2) — exact no-op while the domain is the runtime singleton
 * (ray_sym_domain_str delegates to ray_sym_str). */
static inline void sym_elem(const ray_t* input, int64_t i,
                            const char** out_str, size_t* out_len) {
    int64_t sym_id = ray_read_sym(ray_data((ray_t*)input), i, input->type, input->attrs);
    ray_t* atom = ray_sym_domain_str(ray_sym_vec_domain((ray_t*)input), sym_id);
    if (!atom) { *out_str = ""; *out_len = 0; return; }
    *out_str = ray_str_ptr(atom);
    *out_len = ray_str_len(atom);
}

/* ══════════════════════════════════════════
 * Shared types — used by expr.c and exec.c
 * ══════════════════════════════════════════ */

typedef struct {
    bool    enabled;
    bool    f64;        /* a FLOAT bias over an integer column: SUM and AVG
                         * emit F64 from the integer sum (bias_i64 unused) */
    double  bias_f64;
    int64_t bias_i64;
} agg_affine_t;

#define AGG_LINEAR_MAX_TERMS 8

typedef struct {
    bool    enabled;
    uint8_t n_terms;
    void*   term_ptrs[AGG_LINEAR_MAX_TERMS];
    int8_t  term_types[AGG_LINEAR_MAX_TERMS];
    int64_t coeff_i64[AGG_LINEAR_MAX_TERMS];
    int64_t bias_i64;
} agg_linear_t;

typedef struct {
    uint8_t n_terms;
    int64_t syms[AGG_LINEAR_MAX_TERMS];
    int64_t coeff_i64[AGG_LINEAR_MAX_TERMS];
    int64_t bias_i64;
} linear_expr_i64_t;

/* SUM/AVG(a * b) fused product input: the group accumulators read both
 * source columns and multiply per row (as F64, exactly the expr path's
 * promote-to-double semantics) instead of materializing the product
 * vector — for a 1M-row group that vector is 8MB of alloc+fault+write
 * per query.  Gated to flat, null-free columns with a float product. */
typedef struct {
    bool        enabled;
    const void* pa; int8_t ta; uint8_t aa;
    const void* pb; int8_t tb; uint8_t ab;
} agg_prod_t;

/* ── Expression compiler types ── */

typedef enum {
    EXPR_BAIL_ROOT = 0, EXPR_BAIL_SIZE, EXPR_BAIL_DEPTH, EXPR_BAIL_REGS,
    EXPR_BAIL_INS, EXPR_BAIL_MAPCOMMON, EXPR_BAIL_STR, EXPR_BAIL_GUID,
    EXPR_BAIL_NULLS,
    EXPR_BAIL_SLICE, EXPR_BAIL_SYM_DOMAIN, EXPR_BAIL_CONST,
    EXPR_BAIL_NULL_SHAPE, /* nullable program hit a not-yet-null-capable instruction */
    EXPR_BAIL_OTHER, EXPR_BAIL__N
} expr_bail_t;

extern uint64_t ray_expr_bail_counts[EXPR_BAIL__N];
extern uint64_t ray_expr_compile_ok;
extern bool     ray_expr_disable;
extern bool     ray_opt_no_group_pushdown;
extern bool     ray_join_no_build_swap;
extern uint64_t ray_join_build_swaps;
extern bool     ray_join_force_dup_fallback;
/* perf-gate bypass: disable the auto dup-fallback to measure the pre-fix O(dup²) build */
extern bool     ray_join_no_dup_fallback;
extern uint64_t ray_join_dup_fallbacks;
extern uint64_t ray_join_null_fallbacks;
extern bool     ray_join_force_null_checks;
extern uint64_t ray_join_nullfree_keys;
extern bool     ray_agg_engine_v2; /* route OP_GROUP through v2 agg engine; default ON (agg_engine.c) */
void ray_expr_stats_init(void);

#define EXPR_MAX_REGS 32
#define EXPR_MAX_INS  96
#define EXPR_MORSEL   RAY_MORSEL_ELEMS

typedef struct {
    uint8_t opcode;     /* OP_ADD, OP_NEG, OP_CAST, etc. */
    uint8_t dst;        /* destination register */
    uint8_t src1;       /* source 1 register */
    uint8_t src2;       /* source 2 register (0xFF for unary) */
    uint8_t null_aware; /* 1 = use sentinel-checking kernel variant */
    uint8_t u8_mod_zero; /* U8 divisor: modulo by zero yields 0 */
} expr_ins_t;

enum { REG_SCAN = 0, REG_CONST = 1, REG_SCRATCH = 2 };

typedef struct {
    uint8_t n_ins;
    uint8_t n_regs;
    uint8_t n_scratch;      /* scratch registers needed */
    uint8_t out_reg;
    int8_t  out_type;       /* RAY_F64, RAY_I64, or RAY_BOOL */
    bool    has_parted;     /* true if any REG_SCAN refs a parted column */
    struct {
        uint8_t     kind;       /* REG_SCAN / REG_CONST / REG_SCRATCH */
        int8_t      type;       /* computational type: RAY_F64 / RAY_I64 / RAY_BOOL */
        int8_t      col_type;   /* original column type (REG_SCAN only) */
        uint8_t     col_attrs;  /* column attrs — RAY_SYM width (REG_SCAN only) */
        bool        is_parted;  /* true if this SCAN refs a parted column */
        bool        nullable;   /* lanes may contain NULL_I64 / NaN */
        bool        null_src;   /* that nullability traces to a nullable column
                                 * (else: only op-generated sentinels) */
        ray_col_src_t* src;     /* bound source (REG_SCAN; borrowed from the graph) */
        bool        enc;        /* src is range-read per morsel (generation 2,
                                 * single file, not materialized; or, rebound
                                 * per segment, an encoded parted segment):
                                 * data is NULL */
        const void* data;       /* column data pointer (REG_SCAN only) */
        ray_t*       col_obj;    /* source column vec (REG_SCAN, non-parted) —
                                  * carries the chunk-zone index for zone-skip */
        ray_t*       parted_col; /* parted wrapper, or a RAY_ENCCOL_PARTED without a
                                  * published materialization (is_parted only):
                                  * expr_eval_full_parted rebinds it per segment */
        double      const_f64;  /* scalar value (REG_CONST) */
        int64_t     const_i64;  /* scalar value (REG_CONST) */
    } regs[EXPR_MAX_REGS];
    expr_ins_t ins[EXPR_MAX_INS];
} ray_expr_t;

/* Encoded leaves of a fused BOOL program (expr.c, expr_enc_plan_build): a
 * comparison of an encoded single-file scan register with a constant of
 * the same lane type, evaluated on the vector's view instead of its
 * decoded values. */
typedef struct {
    uint8_t  ins;          /* the compare instruction */
    uint8_t  reg;          /* its encoded scan register */
    uint16_t op;           /* normalized: the column on the left */
    bool     is_f64;       /* F64 lanes: the scan register and the constant are both F64 */
    int64_t  ci;           /* the constant, I64 lanes */
    double   cf;           /* the constant, F64 lanes */
    uint8_t  null_result;  /* the leaf kernel's result on the lane's null sentinel */
} expr_enc_leaf_t;
typedef struct {
    uint8_t         n_leaves;
    int8_t          leaf_of_ins[EXPR_MAX_INS];
    bool            load_reg[EXPR_MAX_REGS];   /* some non-leaf instruction reads the register */
    expr_enc_leaf_t leaf[EXPR_MAX_INS];
} expr_enc_plan_t;

/* ══════════════════════════════════════════
 * Shared gather types — used by filter.c, exec.c (sort, join)
 * ══════════════════════════════════════════ */

#define MGATHER_MAX_COLS 16

typedef struct {
    const int64_t* idx;
    char*          srcs[MGATHER_MAX_COLS];
    char*          dsts[MGATHER_MAX_COLS];
    uint8_t        esz[MGATHER_MAX_COLS];
    int64_t        ncols;
} multi_gather_ctx_t;

typedef struct {
    int64_t*     idx;
    ray_t*        src_col;
    ray_t*        dst_col;
    uint8_t      esz;
    bool         nullable;  /* true = idx may contain -1 (LEFT JOIN nulls) */
} gather_ctx_t;

/* ══════════════════════════════════════════
 * Shared sort types and constants — used by sort_exec.c, exec.c (window)
 * ══════════════════════════════════════════ */

#define RADIX_SORT_THRESHOLD 4096  /* switch from comparison to radix sort */
#define SMALL_POOL_THRESHOLD 8192  /* skip pool dispatch below this size */
#define NEARLY_SORTED_FRAC   0.05  /* threshold for nearly-sorted detection */
#define MK_PRESCAN_MAX_KEYS  8     /* max sort keys for stack allocation */

/* Where a null sorts, stated once for every sort path.
 *
 * A null is the smallest value, so ascending puts nulls first and descending
 * puts them last.  Any path that needs the default must ask here rather than
 * spell it out: two paths spelling out opposite defaults is what made a null's
 * position depend on the row count and the column type, since which path runs
 * is chosen by both. */
static inline uint8_t sort_nulls_first(uint8_t desc) { return (uint8_t)!desc; }

/* Column types whose directory statistics are ordered for the zone gate
 * in P2: BOOL, U8, I16, I32, I64, DATE, TIME, TIMESTAMP. */
static inline bool zone_dir_type(int8_t t) {
    return t == RAY_BOOL || t == RAY_U8 || t == RAY_I16 || t == RAY_I32 || t == RAY_I64 ||
           t == RAY_DATE || t == RAY_TIME || t == RAY_TIMESTAMP;
}

/* Three-valued decision of (col cmp_op cval) over a range from its
 * non-null extrema: 1 all-pass, 0 all-fail, -1 undecided.  Nulls rank
 * below every value (zone_cmp_decision's rule).  all_null ranges and
 * ranges without extrema are undecided. */
static inline int zone_cmp_core(int64_t cmin, int64_t cmax, bool has_minmax, bool has_nulls,
                                bool all_null, uint16_t cmp_op, int64_t cval) {
    if (all_null || !has_minmax || cmin > cmax) return -1;   /* leave to the kernel */
    switch (cmp_op) {
    case OP_EQ:
        if (cval < cmin || cval > cmax)                 return 0;
        if (!has_nulls && cmin == cmax)                 return 1;
        break;
    case OP_NE:
        if (!has_nulls && (cval < cmin || cval > cmax)) return 1;
        if (!has_nulls && cmin == cmax && cval == cmin) return 0;
        break;
    case OP_LT:
        if (!has_nulls && cmin >= cval)                 return 0;
        if (!has_nulls && cmax <  cval)                 return 1;
        break;
    case OP_LE:
        if (!has_nulls && cmin >  cval)                 return 0;
        if (!has_nulls && cmax <= cval)                 return 1;
        break;
    case OP_GT:
        if (cmax <= cval)                               return 0;
        if (!has_nulls && cmin >  cval)                 return 1;
        break;
    case OP_GE:
        if (cmax <  cval)                               return 0;
        if (!has_nulls && cmin >= cval)                 return 1;
        break;
    default: break;
    }
    return -1;
}

/* The SYM membership rule: (ids == c) or (ids != c) over a range whose
 * statistics exclude id 0 ("", the in-band null): min / max over the
 * nonzero ids (has_minmax false when every row is id 0), n_nulls the id-0
 * rows of `rows`.  1 all-pass, 0 all-fail, -1 undecided.  c is an id of
 * the column's domain, or -1 for a string absent from it (EQ all-fail, NE
 * all-pass, without statistics).  c == 0 is decided by the null count
 * alone, a nonzero c by [min, max].  EQ / NE only: ids are never ordered
 * (#738); any other operator is undecided. */
static inline int zone_sym_decision(int64_t cmin, int64_t cmax, bool has_minmax, int64_t n_nulls,
                                    int64_t rows, uint16_t op, int64_t c) {
    if (op != OP_EQ && op != OP_NE) return -1;
    int eq = -1;
    if (c < 0) eq = 0;
    else if (rows <= 0) eq = -1;
    else if (c == 0) eq = n_nulls == 0 ? 0 : n_nulls == rows ? 1 : -1;
    else if (n_nulls == rows) eq = 0;
    else if (!has_minmax || cmin > cmax) eq = -1;
    else if (c < cmin || c > cmax) eq = 0;
    else if (cmin == cmax && n_nulls == 0) eq = 1;
    if (eq < 0) return -1;
    return op == OP_EQ ? eq : 1 - eq;
}

/* zone_cmp_core over double extrema (F32 / F64 directory and descriptor
 * statistics, NaN excluded as the null): the same table under the
 * predicate comparator, where -0.0 == +0.0 as plain C comparison gives. */
static inline int zone_cmp_core_f64(double cmin, double cmax, bool has_minmax, bool has_nulls,
                                    bool all_null, uint16_t cmp_op, double cval) {
    if (all_null || !has_minmax || !(cmin <= cmax) || cval != cval) return -1;   /* leave to the kernel */
    switch (cmp_op) {
    case OP_EQ:
        if (cval < cmin || cval > cmax)                 return 0;
        if (!has_nulls && cmin == cmax)                 return 1;
        break;
    case OP_NE:
        if (!has_nulls && (cval < cmin || cval > cmax)) return 1;
        if (!has_nulls && cmin == cmax && cval == cmin) return 0;
        break;
    case OP_LT:
        if (!has_nulls && cmin >= cval)                 return 0;
        if (!has_nulls && cmax <  cval)                 return 1;
        break;
    case OP_LE:
        if (!has_nulls && cmin >  cval)                 return 0;
        if (!has_nulls && cmax <= cval)                 return 1;
        break;
    case OP_GT:
        if (cmax <= cval)                               return 0;
        if (!has_nulls && cmin >  cval)                 return 1;
        break;
    case OP_GE:
        if (cmax <  cval)                               return 0;
        if (!has_nulls && cmin >= cval)                 return 1;
        break;
    default: break;
    }
    return -1;
}

typedef struct {
    ray_t**       vecs;
    uint8_t*     desc;
    /* NULL means "the default for each key's direction" — see
     * sort_nulls_first.  A caller with its own policy passes one flag per
     * key. */
    uint8_t*     nulls_first;
    uint32_t     n_sort;
} sort_cmp_ctx_t;

/* Radix pass context (shared across histogram + scatter phases) */
typedef struct {
    const uint64_t*  keys;
    const int64_t*   idx;
    uint64_t*        keys_out;
    int64_t*         idx_out;
    int64_t          n;
    uint8_t          shift;
    uint32_t         n_tasks;
    uint32_t*        hist;       /* flat [n_tasks * 256] */
    int64_t*         offsets;    /* flat [n_tasks * 256] */
} radix_pass_ctx_t;

/* Key-encoding context for parallel encode phase */
typedef struct {
    uint64_t*       keys;      /* output */
    int64_t*        indices;   /* if non-NULL, initialize indices[i]=i (fused iota) */
    /* Single-key fields: */
    const void*     data;      /* raw column data */
    ray_t*          col;       /* source column (for null bitmap access) */
    int8_t          type;      /* column type */
    uint8_t         col_attrs; /* RAY_SYM width attrs */
    bool            desc;
    /* Where this key's nulls go: 1=first, 0=last.  Read by every typed
     * single-key arm, not just the float one; the composite encode does
     * not consult it and declines a nullable F64 key instead. */
    bool            nulls_first;
    /* SYM rank mapping (NULL if not sym): */
    const uint32_t* enum_rank; /* intern_id → sort rank */
    /* Composite-key fields (n_keys > 1): */
    uint8_t         n_keys; /* uint8: filled only under the <=16 radix gates (sort.c/window.c) */
    ray_t**          vecs;
    int64_t         mins[16];
    int64_t         ranges[16];
    uint8_t         bit_shifts[16]; /* bit offset for key k in composite */
    uint8_t         descs[16];
    const uint32_t* enum_ranks[16]; /* per-key rank mappings */
} radix_encode_ctx_t;

/* Parallel multi-key min/max prescan context */
typedef struct {
    ray_t*     const* vecs;
    uint32_t* const* enum_ranks;
    uint32_t         n_keys;
    int64_t          nrows;
    uint32_t         n_workers;
    int64_t*         pw_mins;
    int64_t*         pw_maxs;
} mk_prescan_ctx_t;

/* Parallel sort phase 1 context */
typedef struct {
    const sort_cmp_ctx_t* cmp_ctx;
    int64_t*  indices;
    int64_t*  tmp;
    int64_t   nrows;
    uint32_t  n_chunks;
} sort_phase1_ctx_t;

/* Parallel merge pass context */
typedef struct {
    const sort_cmp_ctx_t* cmp_ctx;
    const int64_t*  src;
    int64_t*        dst;
    int64_t         nrows;
    int64_t         run_size;
} sort_merge_ctx_t;

/* Compute the number of significant bytes for radix sort based on type.
 * Returns 1..8: the number of byte passes radix_sort_run needs. */
static inline uint8_t radix_key_bytes(int8_t type) {
    switch (type) {
    case RAY_BOOL: case RAY_U8:   return 1;
    case RAY_I16:                return 2;
    case RAY_I32: case RAY_F32: case RAY_DATE: case RAY_TIME: return 4;
    default:                    return 8;  /* I64, F64, TIMESTAMP, SYM */
    }
}

/* ══════════════════════════════════════════
 * Extern forward declarations — larger functions in exec.c
 * ══════════════════════════════════════════ */

/* ── exec.c (gather helpers) ── */
void multi_gather_fn(void* raw, uint32_t wid, int64_t start, int64_t end);
void gather_fn(void* raw, uint32_t wid, int64_t start, int64_t end);
void partitioned_gather(ray_pool_t* pool, const int64_t* idx, int64_t n,
                        int64_t src_rows, char** srcs, char** dsts,
                        const uint8_t* esz, int64_t ncols);

/* ── filter.c ── */
/* Scan read-ahead registration (ray_pool_scan_set).  exec_scan_add_col adds
 * `col` when it is a mapped fixed-width vector of the registration's length
 * (the first such column sets it, resident or not) whose pages are not
 * already resident, so a warm query registers nothing to read. */
void exec_scan_init(ray_pool_scan_t* s);
void exec_scan_add_col(ray_pool_scan_t* s, ray_t* col);
/* The running select's filter produced selection `sel` (NULL: every row
 * kept, or a compacted table of `kept` rows): request the columns its
 * registration held back for it (query.c). */
void ray_select_scan_filtered(ray_t* sel, int64_t kept);

ray_t* exec_filter(ray_graph_t* g, ray_op_t* op, ray_t* input, ray_t* pred);
ray_t* exec_filter_head(ray_t* input, ray_t* pred, int64_t limit);
/* The one `where:` rule (see filter.c): consumes `pred`, returns an owned
 * BOOL mask of `tn` rows or an error.  `label` prefixes the error. */
ray_t* ray_where_mask_coerce(ray_t* pred, int64_t tn, const char* label);
/* Typed membership kernel (verdict-LUT / SIMD / pool-parallel); NULL when
 * the col/set shape is unsupported — caller falls back to its own path. */
ray_t* ray_in_vec_exec(ray_t* col, ray_t* set, bool negate);
ray_t* sel_compact(ray_graph_t* g, ray_t* tbl, ray_t* sel,
                   const int64_t* keep_syms, int keep_n);

/* ── expr.c ── */
/* The try_* planners and expr_compile return false for "not this shape".  A
 * failed column bind (or plain request) also returns false but leaves an
 * owned error in *err (ray_error_free); *err is NULL otherwise. */
/* try_affine_sumavg_input's *out_vec is the base scan's column object: the
 * plain column when it is ready, else a RAY_ENCCOL the caller binds
 * (ray_col_src_bind) and reads by range.  try_linear_sumavg_input_i64 and
 * try_prod_sumavg_input_f64 read whole-column pointers: an encoded term not
 * yet materialized declines them (*err NULL). */
bool try_affine_sumavg_input(ray_graph_t* g, ray_t* tbl, ray_op_t* input_op,
                             bool f64_bias_ok,
                             ray_t** out_vec, agg_affine_t* out_affine, ray_t** err);
bool try_linear_sumavg_input_i64(ray_graph_t* g, ray_t* tbl, ray_op_t* input_op,
                                 agg_linear_t* out_plan, ray_t** err);
bool try_prod_sumavg_input_f64(ray_graph_t* g, ray_t* tbl, ray_op_t* input_op,
                               agg_prod_t* out, ray_t** err);
bool expr_compile(ray_graph_t* g, ray_t* tbl, ray_op_t* root, ray_expr_t* out, ray_t** err);
ray_t* expr_eval_full(const ray_expr_t* expr, int64_t nrows);
/* A failed expr_eval_full result: NULL or oom keeps the caller's fallback;
 * any other error (a range read's corrupt, a cancel) is the query's answer,
 * since a fallback would re-read the same column and hide the path. */
static inline bool expr_eval_err_fatal(ray_t* v) {
    return v && RAY_IS_ERR(v) && strcmp(ray_err_code(v), "oom") != 0;
}
/* Test hook: run the morsel evaluator of a BOOL program over [start, end)
 * (at most EXPR_MORSEL rows, any alignment) into out[0 .. end - start). */
void ray_expr_test_eval_range(const ray_expr_t* expr, int64_t start, int64_t end,
                              uint8_t* out, ray_t** err);
/* Test hook: the parted evaluator runs every segment instead of stopping
 * at the first failing one (false restores the stop). */
void ray_expr_test_parted_all_segments(bool on);
/* Monotonic count of fused-expression morsels decided from statistics
 * (zone or directory) without reading their rows.  Tests. */
int64_t ray_expr_test_zone_skips(void);
/* Test hook: selection segments a single-leaf program emitted from an RLE
 * view's runs (expr_leaf_emit), never expanded to bools. */
int64_t ray_expr_test_leaf_runs(void);

/* exec.c test hook: membership tests the sparse filter refinement took
 * (refine_in_keep) since process start. */
uint64_t ray_exec_test_refine_in(void);
/* exec.c test hook: filter chains pruned from their conjuncts' chunk
 * statistics before running (exec_filter_chain_prune), and the selection
 * segments (vectors) they dropped, since process start. */
void ray_exec_test_zone_prune(uint64_t* calls, uint64_t* pruned_segs);
/* exec.c test hook: later `in` conjuncts over a single-file generation-2
 * column answered by range (in_sel_single_within) since process start. */
uint64_t ray_exec_test_in_single_within(void);
/* query.c test hook: counts the (count (select ... where <compare>)) fast
 * path answered since process start. */
uint64_t ray_query_test_count_fast(void);
/* query.c test hook: rows of data[0, n) of `type` (BOOL, U8, I16, I32,
 * I64, DATE, TIME, TIMESTAMP) for which `x op rhs` holds, op 1..6 = == !=
 * < <= > >=; ref: through the reference widening compare instead of the
 * native-width kernels. */
int64_t  ray_query_test_count_compare(int8_t type, const void* data, int64_t n, int64_t rhs,
                                      int op, bool ref);

/* Fused predicate→selection (skips the full BOOL vec).
 *
 * Compiles and evaluates `pred` in selection-output mode (per-task rowsel
 * builders dispatched over [0,nrows), then finished into one rowsel block).
 * Handles the parted/non-BOOL guard.
 * Returns:
 *   - non-NULL rowsel block  → the selection (caller installs / reads it).
 *   - NULL with *all_pass=1  → every row passed (all-pass = "no selection").
 *   - NULL with *all_pass=0  → disabled / unsupported, or an OOM before the
 *                              dispatch (builders, rowsel block) — caller falls
 *                              back to exec_node→ray_rowsel_from_pred unchanged.
 *   - an owned error         → a column bind or a range read failed, a task's
 *                              fused or range scratch ran out of memory (oom),
 *                              or the dispatch was cancelled; the caller
 *                              returns it (no fallback). */
ray_t* exec_pred_to_selection(ray_graph_t* g, ray_op_t* pred, int64_t nrows,
                              bool* all_pass);
/* exec_pred_to_selection of `pred` ANDed with the existing selection `sel`
 * (borrowed): only the segments sel keeps are evaluated, with the fused
 * path's encoded leaves.  The same tri-state return (a block, NULL with
 * *all_pass, NULL unsupported) or an owned error. */
ray_t* exec_pred_refine_selection(ray_graph_t* g, ray_op_t* pred, ray_t* sel, bool* all_pass);

/* Per-morsel "fill this morsel's bool" callback used by pred_sel_drive.
 * Writes out[0..n) (1=selected, 0=not) for the kernel's verdict on global
 * rows [start, start+n).  Null rows must resolve to 0 (not-selected).  `sc`
 * is the calling task's range scratch: initialized at the task's start and
 * freed at its end on the same thread, so a fill that range-reads one
 * source keeps one decode buffer per task, not per morsel. */
typedef void (*pred_sel_fill_fn)(void* ctx, ray_col_scratch_t* sc, int64_t start,
                                 int64_t n, uint8_t* out);

/* Generic selection-output driver: dispatch `fill` over [0,nrows) in the
 * per-task rowsel-builder model, stitch into one block.  Same tri-state
 * return as exec_pred_to_selection (block / NULL+all_pass / NULL).  Used by
 * the standalone predicate kernels (exec_in_to_selection).  With `fail`
 * (initialized by the caller, noted by the fill), the dispatch's verdict is
 * ray_scan_fail_finish's: the lowest noted failure, else `cancel` for a
 * cancelled dispatch, returned as an owned error (no block). */
ray_t* pred_sel_drive(int64_t nrows, pred_sel_fill_fn fill, void* fill_ctx,
                      ray_scan_fail_t* fail, bool* all_pass);

/* `in` / `not-in` selection-output worker: a root OP_IN / OP_NOT_IN over a
 * flat column + literal set, streaming membership bools straight into a
 * rowsel (a generation-2 column is read by range).  Same tri-state return as
 * exec_pred_to_selection; NULL+all_pass=0 for any unsupported shape (non-CONST
 * set, parted/MAPCOMMON/STR/GUID column) → bool-path fallback; an owned
 * error for a bind or range-read failure or a cancel. */
ray_t* exec_in_to_selection(ray_graph_t* g, ray_op_t* pred, int64_t nrows,
                            bool* all_pass);
/* (== s "c") / (!= s "c") on a STR column whose source is generation 2
 * (single file or parted with an encoded segment), as a rowsel.  Byte
 * equality, nulls included ("" is the in-band null).  Per segment: one
 * dictionary lookup; chunks and vectors decided from code bounds; STR_DICT
 * vectors compared on codes; other vectors range-read and compared on
 * bytes.  Same tri-state as exec_pred_to_selection; NULL with *all_pass
 * false for any other shape (generation 0, a published materialization of
 * a single-file column, a SYM constant: their path is unchanged). */
ray_t* exec_str_to_selection(ray_graph_t* g, ray_op_t* pred, int64_t nrows, bool* all_pass);
/* exec_str_to_selection ANDed with the installed selection `sel`
 * (borrowed; a later conjunct, exec_pred_refine_selection): a morsel sel
 * drops is not read.  The same returns. */
ray_t* exec_str_refine_selection(ray_graph_t* g, ray_op_t* pred, ray_t* sel, bool* all_pass);
/* Test hook: STR selections answered (within 0: a first conjunct; 1: a
 * refinement) since process start. */
uint64_t ray_exec_test_str_sel(int within);
ray_t* exec_elementwise_unary(ray_graph_t* g, ray_op_t* op, ray_t* input);
ray_t* exec_elementwise_binary(ray_graph_t* g, ray_op_t* op, ray_t* lhs, ray_t* rhs);
ray_t* exec_sym_order(uint16_t opcode, ray_t* lhs, ray_t* rhs,
                      bool l_scalar, bool r_scalar);

/* ── sort_exec.c ── */
int sort_cmp(const sort_cmp_ctx_t* ctx, int64_t a, int64_t b);
void sort_insertion(const sort_cmp_ctx_t* ctx, int64_t* arr, int64_t n);
void sort_merge_recursive(const sort_cmp_ctx_t* ctx,
                          int64_t* arr, int64_t* tmp, int64_t n);
void sort_phase1_fn(void* arg, uint32_t worker_id, int64_t start, int64_t end);
void sort_merge_fn(void* arg, uint32_t worker_id, int64_t start, int64_t end);
void key_introsort(uint64_t* keys, int64_t* idx, int64_t n);
double detect_sortedness(ray_pool_t* pool, const uint64_t* keys, int64_t n);
uint8_t compute_key_nbytes(ray_pool_t* pool, const uint64_t* keys,
                            int64_t n, uint8_t type_max);
int64_t* radix_sort_run(ray_pool_t* pool, uint64_t* keys, int64_t* indices,
                         uint64_t* keys_tmp, int64_t* idx_tmp,
                         int64_t n, uint8_t n_bytes,
                         uint64_t** sorted_keys_out);
uint64_t* packed_radix_sort_run(ray_pool_t* pool, uint64_t* data,
                                 uint64_t* tmp, int64_t n, uint8_t n_bytes);
int64_t* msd_radix_sort_run(ray_pool_t* pool, uint64_t* keys, int64_t* indices,
                              uint64_t* keys_tmp, int64_t* idx_tmp,
                              int64_t n, uint8_t n_bytes,
                              uint64_t** sorted_keys_out);
void radix_encode_fn(void* arg, uint32_t wid, int64_t start, int64_t end);
/* A double as a composite radix component: the bits read as an i64 with
 * a negative value's low 63 bits flipped, so signed i64 order equals
 * numeric order.  The composite prescans take SIGNED min/max per key; the
 * single-key encoding (bits ^ mask) is unsigned-ordered instead and would
 * put every positive below every negative there.  The encode subtracts the
 * min and so only needs the same form on both sides.  -0.0 reads as +0.0:
 * the comparator holds the two equal, so a tie between them must not be
 * split by sign here either (it was, above the comparator's row count). */
static inline int64_t radix_f64_ord(double x) {
    if (x == 0.0) x = 0.0;
    uint64_t bits;
    memcpy(&bits, &x, 8);
    return (int64_t)(bits ^ (-(bits >> 63) >> 1));
}

/* A float null is NaN.  Its bits would pack above every real number, so a
 * null-bearing float key sorted one way below 64 rows (comparator: a null
 * is the smallest value) and the other above (composite).  Instead the
 * prescans leave nulls out of the key's min/max, radix_key_bounds_finish
 * reserves the slot below the minimum, and the encode maps a null there:
 * first on ascending, last on descending, as the I64 sentinel INT64_MIN
 * already is and as the comparator orders it. */
static inline bool radix_f64_is_null(double x) { return x != x; }

/* Final bounds of key `col` for the composite, once kmin/kmax hold the
 * min/max over its non-null values.  An F64 key that may hold nulls gets
 * one slot below its minimum for them; a key with no non-null value at all
 * (every row null) spans a single slot. */
static inline void radix_key_bounds_finish(ray_t* col, int64_t* kmin, int64_t* kmax) {
    if (col->type != RAY_F64) return;
    if (*kmin > *kmax) { *kmin = 0; *kmax = 0; }
    if (ray_vec_may_have_nulls(col)) *kmin = (int64_t)((uint64_t)*kmin - 1);
}
void mk_prescan_fn(void* arg, uint32_t wid, int64_t start, int64_t end);
uint32_t* build_enum_rank(ray_t* col, int64_t nrows, ray_t** hdr_out);
ray_t* exec_sort(ray_graph_t* g, ray_op_t* op, ray_t* tbl, int64_t limit);
/* HEAD(SORT) of `limit` rows under the lazy selection `sel` (borrowed),
 * late materialization: the keys compacted and top-k'd, then only the
 * winners' rows gathered, the old route's answer row for row.  NULL when
 * the shape is not the top-k shortcut's (compact the table and sort). */
ray_t* exec_sort_head_sel(ray_graph_t* g, ray_op_t* op, ray_t* tbl, ray_t* sel, int64_t limit);
uint64_t ray_sort_test_late_heads(void);   /* test hook: HEAD(SORT)s it answered */

/* ── join.c ── */
ray_t* exec_join(ray_graph_t* g, ray_op_t* op, ray_t* left_table, ray_t* right_table);
ray_t* exec_antijoin(ray_graph_t* g, ray_op_t* op,
                     ray_t* left_table, ray_t* right_table);
ray_t* exec_window_join(ray_graph_t* g, ray_op_t* op,
                        ray_t* left_table, ray_t* right_table);

/* ── group.c ── */
ray_t* exec_reduction(ray_graph_t* g, ray_op_t* op, ray_t* input);
/* Reduction `opcode` (out_type: the op's, 0 for the input's) over a bound
 * encoded source without a selection: the result exec_reduction gives over
 * its materialization, from directory statistics where agg.c answers a
 * generation-0 CHUNK_ZONE column from its zone (ray_zone_dir_answer),
 * else by range reads.  NULL when the type or op is not served here (the
 * caller takes the plain path); otherwise a result or an owned error. */
ray_t* ray_reduce_src(ray_col_src_t* src, uint16_t opcode, int8_t out_type);
/* Test hook (C tests only): 0 normal routing; 1 skip the v2 aggregate
 * engine (legacy ladder); 2 also skip the direct-array route (legacy
 * radix); 3 also skip the slice-group probe.  Disarmed by
 * ray_test_hooks_off. */
void ray_group_test_route(int route);
/* Fail the next scalar chunk-buffer allocation, then disarm automatically. */
void ray_group_test_fail_scalar_alloc(int fail);
/* Test hook (C tests only): the scalar route's chunk walk (scalar_chunk_fn)
 * runs serially in tasks of `rows` rows, any cut (as ray_pool_dispatch's
 * grain after its ring clamp); 0 dispatches as usual.  Disarmed by
 * ray_test_hooks_off. */
void ray_group_test_scalar_grain(int64_t rows);
/* Test hook (C tests only): the scalar route's F64 lane sum of d[0, n) at
 * absolute row row0, as one block (*whole) and as pieces cut at cuts[]
 * through the staged piece function (*cut). */
void ray_group_test_f64_lanes(const double* d, int64_t row0, int64_t n, const int64_t* cuts,
                              int ncuts, double* whole, double* cut);
ray_t* exec_reduction_src(ray_graph_t* g, ray_op_t* op, ray_col_src_t* src);
/* ray_first_fn / ray_last_fn of a bound source for the types they read
 * with collection_elem among those served here (I32, I16, BOOL, U8, DATE,
 * TIME, TIMESTAMP): row 0 (len - 1) itself, a null row a typed null, read
 * by range.  NULL for any other type; else an atom or an owned error. */
ray_t* ray_reduce_src_elem(ray_col_src_t* src, bool last);
/* (collection.c) The value (at tbl 'col) returns eagerly. */
ray_t* ray_table_col_value(ray_t* tbl, int64_t sym);
ray_t* exec_count_distinct(ray_graph_t* g, ray_op_t* op, ray_t* input);

/* Parallel distinct for large fixed-width columns: radix-partitioned
 * dedup (exec_count_distinct's layout) returning an I64 vec of
 * first-occurrence row ids in first-appearance order.  Returns NULL
 * when the kernel doesn't apply (unsupported type, small input, no
 * pool) — caller falls back to the eager hashset.  Errors are
 * RAY_IS_ERR ray_t*. */
ray_t* distinct_radix_first_ids(ray_t* input);

/* Single-pass per-group count(distinct).  Returns I64 vec of length
 * n_groups, or NULL if `src->type` isn't a supported scalar/SYM type
 * (caller falls back to per-group exec_count_distinct).  Errors are
 * returned as RAY_IS_ERR ray_t*. */
ray_t* ray_count_distinct_per_group(ray_t* src, const int64_t* row_gid,
                                    int64_t n_rows, int64_t n_groups);

/* Parallel exact median per group via ray_pool_dispatch_n.  idx_buf is
 * the group-contiguous row-index layout produced by the upstream
 * group-by phase (already prefix-summed; offsets[g]..offsets[g]+
 * grp_cnt[g] is group g's slice).  Returns F64 vec of n_groups, NULL
 * on unsupported source type (caller falls back to serial). */
ray_t* ray_median_per_group_buf(ray_t* src,
                                const int64_t* idx_buf,
                                const int64_t* offsets,
                                const int64_t* grp_cnt,
                                int64_t n_groups);

/* Parallel per-group bounded top-K / bot-K via ray_pool_dispatch_n.
 * Reuses the same idx_buf/offsets/grp_cnt layout as
 * ray_median_per_group_buf.  K must be >= 1; cells shorter than K when
 * grp_cnt[gi] < K (matches the standalone topk_take_vec convention).
 * desc=1 → top (K largest, descending), desc=0 → bot (K smallest,
 * ascending).  Returns ray_list_new(n_groups), each cell is a vec of
 * the same type as `src`.  NULL on unsupported source type. */
ray_t* ray_topk_per_group_buf(ray_t* src,
                              int64_t k,
                              uint8_t desc,
                              const int64_t* idx_buf,
                              const int64_t* offsets,
                              const int64_t* grp_cnt,
                              int64_t n_groups);

/* Per-group min/max/first/last for wide element types (STR/GUID).  Same
 * idx_buf/offsets/grp_cnt layout as the median/topk kernels; materialises the
 * winning element per group into a typed result column.  Returns NULL for a
 * non-wide src (caller falls back to the numeric path). */
ray_t* ray_wide_minmax_per_group_buf(ray_t* src, uint16_t op,
                                     const int64_t* idx_buf,
                                     const int64_t* offsets,
                                     const int64_t* grp_cnt,
                                     int64_t n_groups);

ray_t* exec_group(ray_graph_t* g, ray_op_t* op, ray_t* tbl, int64_t group_limit);

/* Monotonic count of streaming parted-GROUP kernel (exec_group_per_partition)
 * entries.  O(1) per query; surfaced via (.sys.mem)'s "group-perpart-runs". */
int64_t ray_group_perpart_runs(void);
/* Monotonic count of exec_group_per_partition runs that returned their
 * result (not the concat fallback's NULL, not an error).  Tests. */
int64_t ray_group_perpart_done(void);
/* Per-partition non-null count slots built for nullable AVG/VAR/DEV inputs
 * (test hook; a null-free input keeps the row count). */
int64_t ray_group_perpart_nn_slots(void);

/* Monotonic count of exec_group_parted's in-place concat groups (the
 * parted columns grouped from their segments, no flatten).  Tests. */
int64_t ray_group_segstream_runs(void);
/* Monotonic bytes of null fill the parted group built for ragged
 * partitions' stand-ins (bounded per query by the group's transient
 * budget, AGG_COMPACT_BITMAP_BUDGET).  Tests. */
int64_t ray_group_null_fill_bytes(void);
/* Monotonic count of rows the direct-array key prescan walked over a
 * staged (generation-2) key by range (minmax_piece).  Tests. */
int64_t ray_group_test_minmax_staged_rows(void);
/* Monotonic count of per-partition unordered takes answered from the
 * first keys read up front (grp_first_keys_scan).  Tests. */
int64_t ray_group_firstkeys_runs(void);

/* A parted fixed-width column (a plain parted wrapper or a RAY_ENCCOL_PARTED
 * slot; base neither STR nor LIST) copied into one owned flat vector of
 * `total` rows: segments read by range (in place when plain or already
 * materialized, a vector at a time when encoded; never materialized), SYM
 * cells at the widest segment's width over the first segment's domain,
 * HAS_NULLS from the segments.  An owned vector or error. */
ray_t* parted_flatten_fixed(ray_t* col, int64_t total);
/* parted_flatten_fixed reading a hand-built plain wrapper's missing (NULL)
 * segments as their partitions' rows of nulls (cnt / ncnt: the partition-key
 * counts, parted_table_counts). */
ray_t* parted_flatten_fixed_c(ray_t* col, int64_t total, const int64_t* cnt, int64_t ncnt);
/* Test hook: parted_flatten_fixed calls since process start (the flat
 * copies of parted columns the group fallbacks make). */
int64_t ray_group_test_flat_copies(void);
/* The widest SYM width class over a parted SYM column's segments (a plain
 * parted wrapper or a RAY_ENCCOL_PARTED slot), read from their headers: the
 * width a vector holding every segment's ids needs.  0 for anything else. */
uint8_t parted_sym_width(ray_t* col);

/* Monotonic count of streaming parted asof-join per-partition (day) runs
 * bumped once per partition processed by the streaming asof path in query.c.
 * O(1) per partition; surfaced via (.sys.mem)'s "asof-perpart-runs". */
int64_t ray_asof_perpart_runs(void);

/* Monotonic count of per-partition runs taken by either parted ORDER BY
 * streaming path (concat of pre-sorted partitions, or per-partition top-k).
 * Bumped once per partition; surfaced via (.sys.mem)'s "sort-perpart-runs". */
int64_t ray_sort_perpart_runs(void);

/* Monotonic count of per-segment runs taken by the case-a streaming parted
 * window fast-path (PARTITION BY includes the physical/date key).  Bumped
 * once per segment; surfaced via (.sys.mem)'s "window-perpart-runs". */
int64_t ray_window_perpart_runs(void);

/* Monotonic count of per-segment runs taken by the join broadcast-streaming
 * fast-path (exactly one side parted, streamed against the whole other
 * side).  Bumped once per segment; surfaced via (.sys.mem)'s
 * "join-perpart-runs". */
int64_t ray_join_perpart_runs(void);

/* Are these key columns already sorted (single uniform direction) under the
 * sort's lexicographic ordering?  Null-free integer-family / SYM keys only;
 * everything else returns false.  O(nrows) with early bail.  Defined in
 * sort.c; used by the parted ORDER BY streaming path. */
bool ray_key_cols_sorted(ray_t** key_cols, int64_t n_keys, uint8_t descending,
                         int64_t nrows);

/* Build a flat table holding one segment (seg_idx) of a parted table:
 * parted columns yield segs[seg_idx], MAPCOMMON columns are broadcast.
 * Returns an owned flat RAY_TABLE (rc=1), or a RAY_IS_ERR value.  Defined
 * in exec.c; consumed by the streaming parted asof path in query.c. */
ray_t* build_segment_table(ray_t* parted_tbl, int32_t seg_idx);

/* May the DAG rooted at `root` run segment by segment with a concat merge:
 * streamable ops only, scalar constants only, and a scan of the default
 * table.  Defined in exec.c; the parted per-segment sort/take and asof
 * routes in query.c decide a clause's eligibility with it. */
bool dag_can_stream(ray_graph_t* g, ray_op_t* root);
/* Value invariance under order-preserving restriction of the default rows:
 * primitive scalars/row expressions only, with fixed membership-set edges.
 * Caller must separately establish the input/default-table correspondence.
 * Read-only and bounded; false declines a rewrite, not the language program.
 * This does not prove structural-domain or exception-order equivalence. */
bool ray_expr_subset_invariant(ray_graph_t* g, uint32_t root_id);
/* Does node `id` read the default table's rows, through filters, aliases
 * and materializations only?  Defined in opt.c; the correspondence
 * ray_expr_subset_invariant leaves to its caller. */
bool filter_input_default_domain(ray_graph_t* g, uint32_t id);

/* Ordered concat-merge of two partial partition-streamed results (tables:
 * per-column concat; vectors: direct concat).  Returns an OWNED result.
 * Defined in exec.c; consumed by the streaming parted asof path in query.c. */
ray_t* ray_result_merge(ray_t* accum, ray_t* partial);

/* Slice-group fusion probe (exec.c): arm g's slice-group hint instead of
 * executing the WHERE filter when its predicate is exactly in/eq on the
 * single bare group-key column with a fresh CSR hash index.  See the
 * definition for the full contract. */
bool ray_slice_group_probe(ray_graph_t* g, ray_op_t* root, ray_t* by_col);

/* Early hint settle (group.c): fold an armed slice-group hint whose group
 * shape is kernel-ineligible into the equivalent g->selection, ahead of
 * OP_GROUP's sparse pre-compaction.  Returns an error to propagate or
 * NULL. */
ray_t* ray_group_slice_hint_settle(ray_graph_t* g, ray_op_t* op, ray_t* tbl,
                                   int64_t group_limit);

/* Dict-code count-distinct row_gid side-channel: exec_group stashes the
 * per-result-group dict codes so the count-distinct path can build row_gid
 * by remapping codes (cheap int) instead of re-hashing strings. */
typedef struct {
    int32_t*     result_codes;  /* [n_groups] dict code of each result group */
    int64_t      n_groups;
    int64_t      key_sym;
    const ray_t* tbl;
    int64_t      n_distinct;
    uint8_t      valid;
} ray_dict_cd_t;
ray_dict_cd_t ray_dict_cd_get(void);
void ray_dict_cd_clear(void);
/* Runtime teardown, before the heap goes: frees this thread's stash and
 * makes every other thread's stale (dropped, never freed: #865). */
void ray_dict_cd_runtime_gone(void);

/* ── collection.c ── */
ray_t* distinct_vec_eager(ray_t* x);
ray_t* reverse_vec_eager(ray_t* x);
ray_t* lag_vec_eager(ray_t* x);
ray_t* lead_vec_eager(ray_t* x);
ray_t* deltas_vec_eager(ray_t* x);
ray_t* ratios_vec_eager(ray_t* x);
ray_t* fills_vec_eager(ray_t* x);
ray_t* running_vec_eager(ray_t* x, uint16_t opcode);
ray_t* differ_vec_eager(ray_t* x);
ray_t* moving_vec_eager(ray_t* x, uint16_t opcode, int64_t window);

/* ── sort.c ── */
ray_t* asc_vec_eager(ray_t* x);
ray_t* desc_vec_eager(ray_t* x);

/* Group HT types and helpers — shared with pivot (exec.c) */
#define GHT_NEED_SUM     0x01
#define GHT_NEED_MIN     0x02
#define GHT_NEED_MAX     0x04
#define GHT_NEED_SUMSQ   0x08
/* OP_PEARSON_CORR per-group accumulators: x-side piggybacks on SUM and
 * SUMSQ blocks; this flag enables the y-side blocks (Σy, Σy², Σxy). */
#define GHT_NEED_PEARSON 0x10
/* Exact integer AVG: an extra int64 block (off_sum_hi) carries the high
 * word of a 128-bit two's-complement sum next to each off_sum slot, so an
 * integer mean never divides a wrapped int64 (ray_i128_add / ray_i128_to_f64
 * in idxop.h).  Set whenever an OP_AVG agg has a non-float input; SUM keeps
 * reading off_sum alone (int64 wraparound is its contract). */
#define GHT_NEED_SUM128  0x20

/* ── ght_layout_t — inline-or-spill, fixed-size, by-value embeddable ──
 *
 * The per-agg and per-key metadata that used to live in fixed [8] arrays +
 * uint8 bitmasks now lives behind BASE POINTERS.  Widths ≤ GHT_INLINE (8)
 * point the bases at the in-struct inline arrays (same cache lines, same
 * memory locality as the old fixed [8] fields); wider layouts point them at
 * one owned heap spill block (spill_hdr).  Because a struct with pointers to
 * its own members does NOT survive a raw memcpy / by-value return, every
 * by-value materialisation MUST fix the bases: ght_compute_layout aims them
 * at the caller-provided out struct, and ght_layout_copy re-aims them at the
 * destination's own inline arrays (inline) or shares the source's spill
 * (spilled — the ORIGINAL owns/frees it, copies must not outlive it).
 *
 * The eight former per-agg / per-key bitmasks become per-element FLAG BYTES
 * (agg_flags[a] / key_flags[k]); agg_flags_any / any_wide_key / any_inline_str
 * are OR-summaries kept as scalar fast guards for the old `mask != 0` shape
 * tests. */
#define GHT_INLINE 8
/* agg_flags bits */
#define GHT_AF_F64      1u
#define GHT_AF_SYM      2u   /* lex compare for MIN/MAX (sym_lex_lt) */
#define GHT_AF_FIRST    4u
#define GHT_AF_LAST     8u
#define GHT_AF_PROD     16u
#define GHT_AF_BINARY   32u  /* two inputs (OP_PEARSON_CORR): packs (x,y) */
#define GHT_AF_HOLISTIC 64u  /* MEDIAN/QUANTILE/TOP/BOT/wide-mm: no accum slot */
#define GHT_AF_WIDE     128u /* subset of HOLISTIC: wide-element min/max/first/last */
/* agg_flags2 bits (agg_flags is full — 8 bits used) */
#define GHT_AF2_NULLABLE 1u  /* agg input column advertises HAS_NULLS: the row-layout
                              * accumulators skip F64 NaN / NULL_I* sentinels and track
                              * a per-slot non-null count (off_nn) for the divisor and
                              * all-null → typed-null finalization. */
#define GHT_AF2_TRUTHY   2u  /* OP_ALL/OP_ANY: off_sum stores truthy count */
#define GHT_AF2_Y_NULLABLE 4u /* binary agg whose y-side column advertises
                               * HAS_NULLS: the null-aware accumulators skip
                               * the whole (x,y) pair when y is null, matching
                               * the scalar reducer, the v2 engine and the DA
                               * path (pair-skip contract).  Entry packing
                               * canonicalizes int y nulls (NaN in F64-packed
                               * entries, NULL_I64 in int-packed ones). */
/* key_flags bits */
#define GHT_KEYF_WIDE       1u  /* key does not fit in 8 B (RAY_GUID / RAY_STR) */
#define GHT_KEYF_INLINE_STR 2u  /* key stores a 16 B ray_str_t descriptor inline */
typedef struct {
    uint16_t entry_stride;
    uint16_t row_stride;
    uint16_t n_keys;           /* widened from uint8_t (unbounded-ready) */
    uint16_t n_aggs;
    uint16_t n_agg_vals;
    uint8_t  need_flags;
    uint16_t off_sum;
    uint16_t off_min;
    uint16_t off_max;
    uint16_t off_sumsq;
    /* Per-slot source-row bounds for FIRST/LAST under work-stealing
     * dispatch (see ght_compute_layout).  Allocated only when any agg
     * is OP_FIRST or OP_LAST; otherwise both stay 0.  Each block is
     * n_agg_vals * 8 bytes of int64 (initialized in init_accum_from_entry
     * to the source row index of the entry that opened the group, then
     * monotonically updated by accum_from_entry).  The matching entry
     * tail slot at offset (entry_stride - 8) carries the source-row
     * index of the row the entry was built from. */
    uint16_t off_first_row;
    uint16_t off_last_row;
    /* OP_PEARSON_CORR y-side accumulators.  Allocated when
     * GHT_NEED_PEARSON is set; for an OP_PEARSON_CORR agg at slot s the
     * x-side accumulators live at off_sum[s] (Σx) and off_sumsq[s] (Σx²),
     * the y-side at these three offsets at the same slot index. */
    uint16_t off_sum_y;
    uint16_t off_sumsq_y;
    uint16_t off_sumxy;
    /* High words of the 128-bit integer sums (GHT_NEED_SUM128); 0 when the
     * layout carries none. */
    uint16_t off_sum_hi;
    /* Earliest contributing source row for this group.  Every packed entry
     * carries its source row in the tail slot; partition merges retain the
     * minimum so output order is independent of radix partition count. */
    uint16_t off_group_first;
    /* Per-slot non-null count block (n_agg_vals int64 slots), allocated only
     * when any_agg_null (any agg input column HAS_NULLS).  Zero otherwise —
     * finalize then divides by the group row count exactly as before, so
     * every null-free shape is byte-identical.  When present, AVG/VAR/STDDEV
     * use nn[s] as the divisor and MIN/MAX/PROD/FIRST/LAST/AVG/VAR/STDDEV
     * emit a typed null for nn[s]==0 (all-null group), matching the DA path. */
    uint16_t off_nn;
    uint16_t key_region;       /* total key region bytes = key_off[n_keys] + null_words*8 */
    /* Number of int64 null-mask words carried after the keys: ceil(n_keys/64)
     * (floored at 1 so the n_keys==0 fallback keeps a null slot).  Key null
     * bit k lives at word (k>>6), bit (k&63); word w is at key_off[n_keys]+w*8.
     * ≤64 keys use exactly one word — byte-identical to the legacy single-int64
     * null slot, so every reachable (≤8-key) shape is unchanged. */
    uint16_t null_words;
    uint8_t  agg_flags_any;    /* OR of all agg_flags[a] — scalar shape guard */
    uint8_t  any_wide_key;     /* replaces wide_key_mask != 0 */
    uint8_t  any_inline_str;   /* replaces key_inline_str != 0 */
    /* Set when the whole key tuple is plain fixed-width integer lanes: no
     * wide/inline-STR key and no floating-point key.  Such a key region is
     * exactly (n_keys + null_words) 8-byte lanes whose STORED BITS are the
     * identity, so hash and compare may treat it as one packed value.  F64
     * keys are excluded: read_col_i64 (which the packed stagers use) does not
     * decode F64 at all, and keeping them out also keeps the packed lanes free
     * of the -0.0 canonicalisation the F64 builders apply (#407,
     * group_key_f64_bits).  Pinned by group_extra/ght_packed_key_excludes_f64. */
    uint8_t  packed_key;
    uint8_t  any_agg_null;     /* OR of GHT_AF2_NULLABLE over aggs — hoisted hot-loop gate */
    /* ── base pointers: aim at the *_in inline arrays (≤8) or the spill block ── */
    int8_t*  agg_val_slot;     /* [n_aggs] accum slot per agg, -1 = none */
    uint8_t* agg_flags;        /* [n_aggs] GHT_AF_* per agg */
    /* Per-agg SYM resolution domain (sym-domain Phase 2): the domain of
     * agg_vecs[a] when it is a SYM column, NULL otherwise.  Borrowed —
     * the agg input vec outlives every layout use.  accum_from_entry's
     * lex MIN/MAX resolves cell ids through it (cell ids are positions
     * in the COLUMN's domain, not the global intern table). */
    struct ray_sym_domain_s** agg_dom;   /* [n_aggs] */
    /* Per-agg null metadata (populated only when any_agg_null).  agg_flags2[a]
     * bit GHT_AF2_NULLABLE marks a HAS_NULLS agg input; agg_null_sentinel[a]
     * is the NULL_I* sentinel for its integer/temporal type (0 for F64, which
     * detects null via NaN, and for non-nullable aggs).  Same inline-or-spill
     * base-pointer discipline as agg_flags/agg_val_slot. */
    uint8_t*  agg_flags2;      /* [n_aggs] GHT_AF2_* per agg */
    int64_t*  agg_null_sentinel; /* [n_aggs] integer null sentinel per agg */
    /* Byte offset of each key within the entry/HT-row key region (relative to
     * the start of the key region, i.e. entry+8 / row+8).  key_off[n_keys] is
     * the offset of null-mask word 0 (null_words words follow contiguously).
     * Most keys are 8 B (key_off[k]==k*8); a RAY_STR key
     * stores its ray_str_t descriptor INLINE (16 B) so probe/rehash compare it
     * cache-locally instead of chasing key_data[k]+row*16 in the source column. */
    uint16_t* key_off;         /* [n_keys + 1] */
    uint8_t*  key_flags;       /* [n_keys] GHT_KEYF_* per key */
    uint8_t*  wide_key_esz;    /* [n_keys] per-element byte size of source col */
    /* Per wide key, the source element type so resolution can branch:
     * RAY_GUID → fixed wide_key_esz bytes; RAY_STR → ray_str_t descriptor
     * (16 B) resolved via the string pool (key_pool[k]) with SSO-aware
     * hash/eq (inline ≤12 B keys need no pool deref). */
    int8_t*   wide_key_type;   /* [n_keys] */
    /* ── inline storage (used when n_keys ≤ 8 AND n_aggs ≤ 8) ── */
    int8_t    agg_val_slot_in[GHT_INLINE];
    uint8_t   agg_flags_in[GHT_INLINE];
    uint8_t   agg_flags2_in[GHT_INLINE];
    int64_t   agg_null_sentinel_in[GHT_INLINE];
    struct ray_sym_domain_s* agg_dom_in[GHT_INLINE];
    uint16_t  key_off_in[GHT_INLINE + 2];
    uint8_t   key_flags_in[GHT_INLINE];
    uint8_t   wide_key_esz_in[GHT_INLINE];
    int8_t    wide_key_type_in[GHT_INLINE];
    /* Owned heap spill block for wider layouts; NULL when inline OR when
     * this instance is a borrower (a copy of a spilled layout — its bases
     * point into the ORIGINAL owner's spill block, but spill_hdr stays NULL
     * so ght_layout_free is a no-op on it).  Because a borrower's own
     * spill_hdr reads NULL exactly like true-inline storage, ght_layout_copy
     * cannot dispatch on spill_hdr == NULL — it must test STORAGE identity
     * (agg_val_slot == agg_val_slot_in) instead; see ght_layout_copy.  That
     * test is depth-invariant: master -> borrower -> borrower -> ... all
     * correctly re-borrow the one shared block, however many copies deep.
     * Freed by ght_layout_free (the owning master only).  Lifetime rule:
     * the owning master must outlive every borrower transitively copied
     * from it — all borrowers, at any depth, must be done with before the
     * master frees the spill. */
    ray_t*    spill_hdr;
} ght_layout_t;

typedef struct {
    uint32_t*    slots;
    uint32_t     ht_cap;
    char*        rows;
    uint32_t     grp_count;
    uint32_t     grp_cap;
    ght_layout_t layout;
    /* Wide-key source-column resolution (non-NULL slots only when
     * layout.any_wide_key != 0).  key_data[k] = raw data pointer into the
     * original key column (slice-unaware) for group_probe_entry /
     * group_ht_rehash to resolve row-indexed wide keys; key_pool[k] = its
     * str-pool base for SSO-aware STR resolve.  BASE POINTERS (unbounded-
     * slots cut 4): aimed at the in-struct inline [8] arrays when n_keys ≤ 8
     * (byte-identical to the legacy fixed layout), or at one owned heap spill
     * (key_wide_hdr) carved by group_ht_init_sized when n_keys > 8.  Because
     * group_ht_t is only ever memset-then-init'd in place (never memcpy'd by
     * value), self-referential inline bases are safe.  Freed with the HT. */
    void**       key_data;
    const void** key_pool;
    void*        key_data_in[8];
    const void*  key_pool_in[8];
    ray_t*       key_wide_hdr;   /* owned spill for the >8-key key_data+key_pool block */
    ray_t*        _h_slots;
    ray_t*        _h_rows;
    uint8_t       oom;        /* set by group_probe_entry on grow failure */
    /* Optional growth target in SLOTS (0 = plain doubling).  Consumed ONLY by
     * group_ht_rehash — never by group_ht_grow, which keeps the row array,
     * where nearly all the bytes live, at O(groups).
     *
     * What the trigger actually proves: a rehash means the table just crossed
     * ht_cap/2 groups.  That is a lower bound on its cardinality, NOT evidence
     * that it will keep growing — so the caller's target (derived from how
     * many rows the table can still receive) is applied at most ONE extra
     * doubling per rehash rather than in full.  A table that stalls right
     * after a jump therefore over-allocates its slot array by 2x and nothing
     * else, while a genuinely near-unique table still reaches its final size
     * in far fewer rungs — and every rung skipped is one whole re-hash and
     * re-insert of every live group avoided.
     *
     * Idempotent (the jump is a max()), so once ht_cap reaches grow_cap the
     * field stops having any effect.  Placed here to land in the existing tail
     * padding — group_ht_t's size is unchanged. */
    uint32_t      grow_cap;
} group_ht_t;

/* Row-level accessors for group HT rows */
#define HT_SALT(h)  ((uint8_t)((h) >> 56))
#define HT_EMPTY    UINT32_MAX
#define HT_PACK(salt, gid)  (((uint32_t)(uint8_t)(salt) << 24) | ((gid) & 0xFFFFFF))
#define HT_GID(s)   ((s) & 0xFFFFFF)
#define HT_SALT_V(s) ((uint8_t)((s) >> 24))

#define ROW_RD_F64(row, off, slot) (((const double*)((const void*)((row) + (off))))[(slot)])
#define ROW_RD_I64(row, off, slot) (((const int64_t*)((const void*)((row) + (off))))[(slot)])
#define ROW_WR_F64(row, off, slot) (((double*)((void*)((row) + (off))))[(slot)])
#define ROW_WR_I64(row, off, slot) (((int64_t*)((void*)((row) + (off))))[(slot)])

/* Fill *out with the row/entry layout.  Returns false on unrecoverable
 * failure (key stride budget overflow, or spill-block OOM for wide layouts);
 * on false *out carries no owned spill.  On success *out owns its spill_hdr
 * (NULL when inline) and every base pointer aims at *out's own storage. */
/* Per-type integer null sentinel for an aggregation column.  Returns 0 for
 * the other types.  For SYM that is the null itself (cell id 0, the empty
 * symbol), and so it is for a strlen fusion over STR/SYM (length 0, the
 * empty string): the row-layout accumulators skip those zeros.  BOOL/U8 are
 * non-nullable and F64 nulls are NaN, so 0 is never consulted for them.
 * Shared by the group row-layout accumulators/finalizers and the pivot
 * finalizer. */
static inline int64_t agg_int_null_sentinel_for(int8_t t) {
    switch (t) {
        case RAY_I64: case RAY_TIMESPAN: case RAY_TIMESTAMP:            return NULL_I64;
        case RAY_I32: case RAY_DATE: case RAY_TIME:  return (int64_t)NULL_I32;
        case RAY_I16:                                return (int64_t)NULL_I16;
        default:                                     return 0;
    }
}

bool ght_compute_layout(ght_layout_t* out, uint32_t n_keys, uint32_t n_aggs,
                        ray_t** agg_vecs, ray_t** agg_vecs2,
                        uint8_t need_flags,
                        const uint16_t* agg_ops,
                        const int8_t* key_types,
                        ray_t* const* key_vecs);
/* By-value copy that fixes the base pointers.  Dispatches on STORAGE, not
 * ownership (src->spill_hdr == NULL is true both for genuine inline layouts
 * AND for borrowers, so it cannot be the test): when src->agg_val_slot ==
 * src->agg_val_slot_in, src's bases are self-referential inline storage and
 * dst's bases are re-aimed at dst's OWN inline arrays (self-contained).
 * Otherwise dst BORROWS the shared spill block that src's bases already
 * point into (dst->spill_hdr = NULL, so ght_layout_free(dst) is a no-op).
 * This makes the copy depth-invariant: master -> borrower -> borrower -> ...
 * all correctly borrow the one spill block regardless of copy depth.
 * Lifetime rule: the owning master MUST outlive every borrower copied from
 * it, transitively — all borrowers at any depth must be done with / freed
 * before the master frees the spill. */
void ght_layout_copy(ght_layout_t* dst, const ght_layout_t* src);
/* NULL-safe; frees the owned spill block (no-op when inline / already freed). */
void ght_layout_free(ght_layout_t* ly);
bool group_ht_init(group_ht_t* ht, uint32_t cap, const ght_layout_t* ly);
void group_ht_free(group_ht_t* ht);
typedef struct {
    int enabled;
    uint8_t agg_index;
    int64_t min_count_exclusive;
    int64_t top_count_take;
    /* Agg op of the filtered agg.  When 0 (the default for the
     * historical COUNT-only filter), consumers MUST treat it as
     * OP_COUNT.  When non-zero, must equal ext->agg_ops[agg_index].
     * Supported here: OP_COUNT, OP_SUM, OP_MIN, OP_MAX.  AVG and
     * higher-order aggs (STDDEV/VAR/PEARSON/MEDIAN) are excluded
     * because their ordering doesn't reduce to a single int64 read
     * from the row slot — they fall through to the full sort + take. */
    uint16_t agg_op;
    /* Direction: 1 = top-N largest (desc), 0 = top-N smallest (asc).
     * For COUNT/SUM/MAX the natural ordering is largest-first; for
     * MIN it's smallest-first.  Both directions are supported per
     * agg kind so `desc: min_value take: N` (the N groups with the
     * largest min) is also expressible.
     *
     * CONTRACT (#408): every arming site MUST set this explicitly.
     * A zero default is NOT coerced to largest-first for COUNT any
     * more — consumers take .desc at face value, and the desc-only
     * keep-min trims are gated off entirely when it is 0. */
    uint8_t  desc;
    /* Evaluation depth (__VM->eval_depth) of the select whose group node
     * this filter targets.  The filter is thread-local and stays armed while
     * an arming select evaluates its `from:` (the "having" shape arms it for
     * the DIRECT child select), so grouped selects nested deeper would see
     * it too; consumers read it through ray_group_emit_filter_active(),
     * which returns it only at this depth. */
    int32_t  target_depth;
} ray_group_emit_filter_t;
ray_group_emit_filter_t ray_group_emit_filter_get(void);
/* The armed filter when the calling thread's eval depth is its target
 * depth (or no VM is bound); a disabled filter otherwise. */
ray_group_emit_filter_t ray_group_emit_filter_active(void);
void ray_group_emit_filter_set(ray_group_emit_filter_t filter);
/* Hash-aggregate rows [start, end) into ht.
 *
 * When match_idx is non-NULL, the loop iterates `i` in [start, end)
 * and reads `row = match_idx[i]` — start/end index the selection
 * space (number of passing rows), not the source column length.
 * When match_idx is NULL, `row = i` — iterating directly over source
 * column rows (no selection). */
/* agg_vecs2 is the optional y-side input column per agg (NULL when no
 * binary aggs).  Pass 1 packs (x, y) consecutively for binary aggs. */
void group_rows_range(group_ht_t* ht, void** key_data, int8_t* key_types,
                      uint8_t* key_attrs, ray_t** key_vecs, ray_t** agg_vecs,
                      ray_t** agg_vecs2,
                      uint8_t* agg_strlen,
                      ray_t* rowsel,
                      int64_t start, int64_t end,
                      const int64_t* match_idx);

/* ══════════════════════════════════════════
 * Pivot ingest — shared parallel hash-aggregate path.
 *
 * Runs the same radix pipeline exec_group uses (phases 1+2), leaving
 * the result in a set of per-partition HTs with prefix offsets. Phase
 * 3 is left to the caller so pivot can restructure the output. For
 * small inputs or when no thread pool is available, falls back to a
 * single sequential HT transparently — the caller iterates
 * part_hts[0..n_parts) the same way either way.
 * ══════════════════════════════════════════ */

typedef struct {
    group_ht_t* part_hts;       /* n_parts entries */
    uint32_t*   part_offsets;   /* n_parts+1 entries (prefix sums of grp_counts) */
    uint32_t    n_parts;        /* 1 sequential; execution-derived when parallel */
    uint32_t    total_grps;
    uint16_t    row_stride;

    /* Internal cleanup state — do not touch from callers. */
    ray_t*      _part_hts_hdr;
    ray_t*      _offsets_hdr;
    void*       _radix_bufs;    /* radix_buf_t* — allocated only on parallel path */
    ray_t*      _radix_bufs_hdr;
    size_t      _n_bufs;
} pivot_ingest_t;

/* Run parallel (or sequential-fallback) hash aggregation for pivot.
 * Returns true on success, false on unrecoverable OOM. On true the
 * caller must eventually call pivot_ingest_free(). Cancellation is
 * propagated via ray_interrupted() — callers should check that too. */
bool pivot_ingest_run(pivot_ingest_t* out,
                      const ght_layout_t* ly,
                      void** key_data, int8_t* key_types, uint8_t* key_attrs,
                      ray_t** key_vecs, ray_t** agg_vecs,
                      int64_t n_scan);

void pivot_ingest_free(pivot_ingest_t* out);

/* ── window.c ── */
ray_t* exec_window(ray_graph_t* g, ray_op_t* op, ray_t* tbl);

/* ── graph_exec.c ── */
ray_t* exec_expand(ray_graph_t* g, ray_op_t* op, ray_t* src_vec);
ray_t* exec_var_expand(ray_graph_t* g, ray_op_t* op, ray_t* start_vec);
ray_t* exec_shortest_path(ray_graph_t* g, ray_op_t* op,
                          ray_t* src_val, ray_t* dst_val);
ray_t* exec_pagerank(ray_graph_t* g, ray_op_t* op);
ray_t* exec_connected_comp(ray_graph_t* g, ray_op_t* op);
ray_t* exec_dijkstra(ray_graph_t* g, ray_op_t* op,
                     ray_t* src_val, ray_t* dst_val);
ray_t* exec_wco_join(ray_graph_t* g, ray_op_t* op);
ray_t* exec_louvain(ray_graph_t* g, ray_op_t* op);
ray_t* exec_degree_cent(ray_graph_t* g, ray_op_t* op);
ray_t* exec_topsort(ray_graph_t* g, ray_op_t* op);
ray_t* exec_cluster_coeff(ray_graph_t* g, ray_op_t* op);
ray_t* exec_betweenness(ray_graph_t* g, ray_op_t* op);
ray_t* exec_closeness(ray_graph_t* g, ray_op_t* op);
ray_t* exec_mst(ray_graph_t* g, ray_op_t* op);
ray_t* exec_random_walk(ray_graph_t* g, ray_op_t* op, ray_t* src_val);
ray_t* exec_dfs(ray_graph_t* g, ray_op_t* op, ray_t* src_val);
ray_t* exec_astar(ray_graph_t* g, ray_op_t* op,
                  ray_t* src_val, ray_t* dst_val);
ray_t* exec_k_shortest(ray_graph_t* g, ray_op_t* op,
                       ray_t* src_val, ray_t* dst_val);

/* ── pivot_exec.c ── */
ray_t* exec_if(ray_graph_t* g, ray_op_t* op);

/* Is a descriptor view worth rebuilding over its own bytes (string.c)? */
bool ray_str_view_should_compact(uint64_t pooled_bytes, int64_t pool_len);

/* Shared-node memo around a sub-evaluation over a swapped g->table
 * (exec.c): push sets the outer memo aside and arms one for the current
 * table and sub-root; pop tears it down and restores the outer one. */
typedef struct {
    ray_t**   vals;
    uint32_t* uses;
    uint32_t  n;
    ray_t*    hdr;
    ray_t*    table;
} ray_exec_memo_save_t;
void ray_exec_memo_push(ray_graph_t* g, ray_op_t* root, ray_exec_memo_save_t* save);
void ray_exec_memo_pop(ray_graph_t* g, const ray_exec_memo_save_t* save);
ray_t* exec_pivot(ray_graph_t* g, ray_op_t* op, ray_t* tbl);

/* ── embedding_exec.c ── */
ray_t* exec_ann_rerank(ray_graph_t* g, ray_op_t* op, ray_t* src);
ray_t* exec_knn_rerank(ray_graph_t* g, ray_op_t* op, ray_t* src);

/* ── temporal_exec.c ── */
ray_t* exec_extract(ray_graph_t* g, ray_op_t* op);
ray_t* exec_date_trunc(ray_graph_t* g, ray_op_t* op);

/* ── string_exec.c ── */
ray_t* exec_like(ray_graph_t* g, ray_op_t* op);
/* Shared LIKE kernel over a STR/SYM column (see string.c); selection may be NULL. */
ray_t* ray_like_vec(ray_t* input, ray_t* pat_v, ray_t* selection);
ray_t* exec_ilike(ray_graph_t* g, ray_op_t* op);
ray_t* exec_string_unary(ray_graph_t* g, ray_op_t* op);
ray_t* exec_strlen(ray_graph_t* g, ray_op_t* op);
void ray_sym_strlen_into(ray_t* input, ray_t* result);   /* string.c */
ray_t* exec_substr(ray_graph_t* g, ray_op_t* op);
ray_t* exec_replace(ray_graph_t* g, ray_op_t* op);
ray_t* exec_concat(ray_graph_t* g, ray_op_t* op);
ray_t* exec_str_find(ray_graph_t* g, ray_op_t* op);

/* ── exec.c ── */
ray_t* materialize_mapcommon(ray_t* mc);
ray_t* materialize_mapcommon_head(ray_t* mc, int64_t n);
ray_t* materialize_mapcommon_filter(ray_t* mc, ray_t* pred, int64_t pass_count);
ray_t* materialize_mapcommon_gather(ray_t* mc, const int64_t* idx, int64_t n);
ray_t* broadcast_scalar(ray_t* atom, int64_t nrows);
ray_t* exec_node(ray_graph_t* g, ray_op_t* op);

/* ══════════════════════════════════════════
 * Parallel null-sentinel helpers (group/window)
 * ══════════════════════════════════════════ */

/* Write the type-correct NULL_* sentinel into a worker-owned payload slot.
 * Workers must own disjoint slots and leave the shared header unchanged:
 * even an atomic attrs OR races with ray_data()'s ordinary SLICE-bit read.
 * The producer derives HAS_NULLS with par_finalize_nulls after workers join,
 * before exposing the result to metadata consumers.
 *
 * BOOL/U8 are non-nullable (rejected at the producer surface).  SYM/STR/GUID
 * don't appear in parallel aggregation/window output columns and are no-ops
 * here. */
static inline void par_set_null(ray_t* vec, int64_t idx) {
    void* p = ray_data(vec);
    switch (vec->type) {
        case RAY_F64:                          ((double*)p)[idx] = NULL_F64; break;
        case RAY_F32:                          ((float*)p)[idx]  = NULL_F32; break;
        case RAY_I64: case RAY_TIMESPAN: case RAY_TIMESTAMP:      ((int64_t*)p)[idx] = NULL_I64; break;
        case RAY_I32: case RAY_DATE: case RAY_TIME: ((int32_t*)p)[idx] = NULL_I32; break;
        case RAY_I16:                          ((int16_t*)p)[idx] = NULL_I16; break;
        default: return;
    }
}

/* After workers join, derive RAY_ATTR_HAS_NULLS from type-correct NULL_*
 * sentinels before publishing the output. This serial pass is not a repair
 * for racing header accesses: workers must not write attrs in the first
 * place. No-op for non-sentinel types. */
static inline void par_finalize_nulls(ray_t* vec) {
    int64_t n = vec->len;
    const void* p = ray_data(vec);
    switch (vec->type) {
        case RAY_F64: {
            const double* d = (const double*)p;
            for (int64_t i = 0; i < n; i++)
                if (d[i] != d[i]) { vec->attrs |= RAY_ATTR_HAS_NULLS; return; }
            return;
        }
        case RAY_F32: {
            const float* d = (const float*)p;
            for (int64_t i = 0; i < n; i++)
                if (d[i] != d[i]) { vec->attrs |= RAY_ATTR_HAS_NULLS; return; }
            return;
        }
        case RAY_I64: case RAY_TIMESPAN: case RAY_TIMESTAMP: {
            const int64_t* d = (const int64_t*)p;
            for (int64_t i = 0; i < n; i++)
                if (d[i] == NULL_I64) { vec->attrs |= RAY_ATTR_HAS_NULLS; return; }
            return;
        }
        case RAY_I32: case RAY_DATE: case RAY_TIME: {
            const int32_t* d = (const int32_t*)p;
            for (int64_t i = 0; i < n; i++)
                if (d[i] == NULL_I32) { vec->attrs |= RAY_ATTR_HAS_NULLS; return; }
            return;
        }
        case RAY_I16: {
            const int16_t* d = (const int16_t*)p;
            for (int64_t i = 0; i < n; i++)
                if (d[i] == NULL_I16) { vec->attrs |= RAY_ATTR_HAS_NULLS; return; }
            return;
        }
        default: return;
    }
}

/* ══════════════════════════════════════════
 * Single-null float model
 * ══════════════════════════════════════════
 * The F64 value domain is {finite} ∪ {0Nf}.  Any non-finite F64 result —
 * NaN OR ±Inf — canonicalizes to NULL_F64 (= __builtin_nan("")) at the
 * produce site; Inf is NOT a value.  Null detection is `x != x`.  Apply
 * ray_f64_fin() to every F64-RESULT produce site (kernels, finalizers,
 * computed-store sites) so a column never holds a non-finite-non-0Nf value.
 *
 * Written compare+select (no early-out branch) so vector loops stay
 * auto-vectorizable.  For DIV/MOD keep the explicit `divisor==0 → NULL_F64`
 * guard at the call site (mirrors the int kernels, avoids the FPU NaN slow
 * path) and wrap only the value-producing branch. */
static inline double ray_f64_fin(double r) {
    /* finite ⟺ |r| <= DBL_MAX: this comparison is FALSE for ±Inf (|Inf| >
     * DBL_MAX) AND for NaN (every NaN compare is false), so it selects exactly
     * the finite values and maps every non-finite to NULL_F64.  Use this fabs +
     * compare + select, NOT __builtin_isfinite — the latter lowers to a scalar
     * fpclassify-style sequence that breaks auto-vectorization and regressed
     * the hot float kernels ~50% (measured).  fabs is a bitwise-AND mask and
     * the ternary becomes a vector blend, keeping the kernel loops vectorized
     * (measured: within noise of baseline). */
    return (__builtin_fabs(r) <= DBL_MAX) ? r : NULL_F64;
}

/* Post-pass for fused/fallback F64 kernels that may have canonicalized a
 * non-finite result to NULL_F64 (NaN) in-buffer: scan [off,off+len) and set
 * RAY_ATTR_HAS_NULLS if any lane is a NaN sentinel.  Mirrors
 * mark_i64_overflow_as_null (which relies on the sentinel already being in
 * the payload — here NULL_F64 is too, so we only flip the attr).  Single
 * pass, memory-bound, branchless inner test; caller invokes single-threaded
 * after pool dispatch joins. */
static inline void mark_f64_nonfinite_as_null(ray_t* result, int64_t off, int64_t len) {
    const double* d = (const double*)ray_data(result) + off;
    for (int64_t i = 0; i < len; i++) {
        if (RAY_UNLIKELY(d[i] != d[i])) { result->attrs |= RAY_ATTR_HAS_NULLS; return; }
    }
}

/* Canonicalise IEEE 754 -0.0 → +0.0 via bit-level check.
 * Immune to -fno-signed-zeros (which makes `if (f==0) f=0` a no-op).
 * Used at output / hash-key boundaries only — not in hot SIMD loops. */
static inline double clear_neg_zero(double v) {
    uint64_t bits; memcpy(&bits, &v, 8);
    if (bits == UINT64_C(0x8000000000000000)) v = 0.0;
    return v;
}

/* One F64 value as one hash key.  Every NaN payload folds to the same
 * canonical NaN and -0.0 to +0.0, so a null counts once however it was
 * produced and 0.0 keys the same whichever sign it carries.
 *
 * Shared rather than open-coded because agreement between kernels is the
 * point: count-distinct answers from a serial loop, a partitioned one and a
 * per-group buffer depending on the row and group counts, and they have to
 * key a value identically or the count depends on the shape of the data. */
static inline int64_t canon_f64_key(double v) {
    if (v != v) v = (double)NAN;
    else v = clear_neg_zero(v);
    int64_t k;
    memcpy(&k, &v, sizeof(int64_t));
    return k;
}

/* Dispatch contiguous group slices by their row counts. */
void ray_group_dispatch(ray_pool_fn fn, void* context, const int64_t* counts, int64_t groups);

/* Associative winner reductions over stable source-row slices. The callback
 * also merges two non-null partial winners, in original slice order. */
typedef int64_t (*ray_group_winner_fn)(void* context, const int64_t* rows, int64_t count);
void ray_group_winners(ray_group_winner_fn fn, void* context, const int64_t* rows,
    const int64_t* offsets, const int64_t* counts, int64_t groups, int64_t* winners);

/* Gather winning group rows, retaining source domains and native types. */
ray_t* ray_group_gather(ray_t* column, const int64_t* rows, int64_t count);
/* Keep a grouped key column's "may have nulls" flag only when a key value is
 * null (vectorised, pool-split scan); see agg_engine.c. */
void   ray_group_key_settle_nulls(ray_t* kc);


/* A scalar `where:` predicate holds for every row or none.  It needs a truth
 * value: booleans, numbers and temporals have one (zero and null are false);
 * a symbol or a string does not. */
static inline bool ray_pred_atom_type_ok(int8_t t) {
    switch (t) {
    case -RAY_BOOL: case -RAY_U8: case -RAY_I16: case -RAY_I32: case -RAY_I64:
    case -RAY_F32: case -RAY_F64: case -RAY_DATE: case -RAY_TIME: case -RAY_TIMESTAMP:
        return true;
    default:
        return false;
    }
}

/* Owned merged query dict, or NULL when nested select composition is unsafe. */
ray_t* ray_select_merge_plan(ray_t* dict);

#endif /* RAY_EXEC_INTERNAL_H */