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.
 */

#ifndef RAY_IDXOP_H
#define RAY_IDXOP_H

/*
 * idxop.h -- Per-vector accelerator indices.
 *
 * A vector with RAY_ATTR_HAS_INDEX set carries a child ray_t of type
 * RAY_INDEX in its aux[0..7] slot.  The index ray_t holds:
 *   - the kind (hash / sort / zone / bloom)
 *   - kind-specific payload (keys vec, perm vec, min/max, bloom bits)
 *   - a snapshot of the parent's original 16-byte aux union plus
 *     the relevant attrs bits, so detach can restore the vector to its
 *     pre-attach state byte-for-byte.
 *
 * Attach precondition: parent must not be a slice, must not already
 * carry an index, must be COW'd to rc==1 by the caller's path.
 *
 * Mutation invalidates: any in-place write to the parent vector must
 * call ray_index_drop() first AND clear RAY_ATTR_SORTED — a stale index
 * or a stale sorted marker is a wrong-answer bug (e.g. an asof-join that
 * trusts a false ordering claim).  Operators that build a fresh result
 * vector get this for free (fresh vectors start at attrs==0, no index).
 */

#include <stdio.h>   /* FILE: ray_index_inline_write_file */
#include <rayforce.h>
#include "mem/heap.h"  /* RAY_ATTR_HAS_INDEX */
#include "vec/str.h"   /* ray_str_t: ray_dict_acc_add_raw */

/* Maximum key columns a RAY_IDX_UKEY map covers.  Chosen so the arm fits
 * the existing union (see the static assert below); a wider key simply gets
 * no index and falls back to the per-call build. */
#define RAY_UKEY_MAX_COLS 8

/* Index kinds.  Stored in ray_index_t.kind. */
typedef enum {
    RAY_IDX_NONE       = 0,
    RAY_IDX_HASH       = 1,
    RAY_IDX_SORT       = 2,
    RAY_IDX_ZONE       = 3,
    RAY_IDX_BLOOM      = 4,
    /* Per-chunk min/max + null bit, one entry per (1 << chunk_log2) rows.
     * The whole-column zone is derivable as
     *   min(chunk_mins)/max(chunk_maxs) over the entries, so this
     *   subsumes RAY_IDX_ZONE wherever it's used in the reduce path.
     * Built at column ingest (csv.read); read by the min/max reduce
     * and by the predicate planner to skip chunks whose [min,max]
     * provably excludes/includes the constant.  See chunk_zone arm
     * of ray_index_t.u below. */
    RAY_IDX_CHUNK_ZONE = 5,
    RAY_IDX_PART       = 6,
    /* Per-column string dictionary: an int32 code per row + the distinct
     * string values (code -> string).  Lets group-by / distinct run on the
     * cheap integer-code path instead of hashing 16-byte ray_str_t descriptors
     * and chasing the string pool.  Built at column ingest / splayed save and
     * persisted inline like the chunk-zone index; the only accelerator index
     * permitted on RAY_STR (it stores codes alongside the descriptors, leaving
     * the column's own representation untouched). */
    RAY_IDX_DICT       = 7,
    /* Key-tuple -> row map for keyed upsert.  RUNTIME-ONLY and attached to
     * the RAY_TABLE object rather than to a column, because the key spans
     * several columns.  Never persisted: serde masks attrs down to
     * HAS_NULLS and col_save strips HAS_INDEX, so it cannot reach a file.
     * See the ukey arm of ray_index_t.u below. */
    RAY_IDX_UKEY       = 8,
} ray_idx_kind_t;

/* Marker bits stored in ray_index_t.markers (block-resident attributes
 * that have no dedicated attrs bit).  sorted lives in attrs (RAY_ATTR_SORTED),
 * not here. */
#define RAY_MARK_UNIQUE  0x01
/* Set when this index and its child vecs are passengers in the parent column's
 * file mapping (restored in-place from the on-disk inline index region).  The
 * parent's single munmap frees them; ray_release_owned_refs must NOT free a
 * passenger index by pointer.  Clear = heap-resident index (freed normally),
 * including a runtime-built index attached to an mmap'd column. */
#define RAY_MARK_MMAP    0x02
/* Hash index whose slot table takes a key's home slot from the TOP bits of
 * its mixed hash (the partitioned builder); clear = the low bits, as every
 * table built before it.  Fixed for the life of the table: a rebuild picks
 * its own. */
#define RAY_MARK_HASH_HIGH 0x04

/* The payload stored inside data[] of a RAY_INDEX ray_t. */
typedef struct {
    uint8_t  kind;            /* ray_idx_kind_t */
    uint8_t  saved_attrs;     /* parent attrs & HAS_NULLS at attach */
    int8_t   parent_type;     /* parent->type (recorded for diagnostics) */
    uint8_t  markers;         /* RAY_MARK_* bits (block-resident attr flags) */
    int64_t  built_for_len;   /* parent->len at attach (mismatch -> stale) */

    /* Raw 16-byte snapshot of parent->aux union at attach time,
     * restored verbatim on detach.  For the numeric vector types that
     * may attach an index (see prepare_attach) this snapshot holds no
     * owned ray_t* refs: bytes 0-7 are unused and bytes 8-15 carry the
     * link_target int64 when HAS_LINK is set. */
    uint8_t  saved_aux[16];

    /* Kind-specific payload.  All ray_t* fields are owning refs. */
    union {
        struct {                /* RAY_IDX_HASH — CSR grouped layout */
            /* Open-addressing bucket table over DISTINCT keys plus a CSR
             * (offsets + contiguous row lists) grouping of the rows:
             *   table[mask+1]  slot -> group id + 1 (0 = empty bucket); a
             *                  key's home slot is the top bits of its mixed
             *                  hash under RAY_MARK_HASH_HIGH (groups then in
             *                  hash order), else the low bits
             *   gkeys[n_groups]     key value per group (i64-canonical;
             *                       SYM stores the column-domain id)
             *   offs[n_groups+1]    rows[] slice bounds per group
             *   rows[n_keys]        row ids, ASCENDING within each group
             * Lookup `k`: slot-walk table comparing gkeys[gid] == k; a hit
             * yields the group's contiguous rows[offs[gid]..offs[gid+1])
             * slice — group size is O(1), first occurrence is rows[offs[gid]],
             * and the ascending order is what rowsel builders and the asof
             * binary search consume directly. */
            ray_t*   table;     /* RAY_I64 vec, capacity entries */
            ray_t*   gkeys;     /* RAY_I64 vec, n_groups entries */
            ray_t*   offs;      /* RAY_I64 vec, n_groups+1 entries */
            ray_t*   rows;      /* RAY_I64 vec, n_keys entries */
            uint64_t mask;      /* capacity - 1 (capacity is power of two) */
            int64_t  n_keys;    /* number of non-null rows indexed */
            int64_t  n_groups;  /* number of distinct keys */
            int64_t  order_sym; /* optional column symbol ordered within each group */
        } hash;
        struct {                /* RAY_IDX_SORT */
            ray_t* perm;        /* RAY_I64 vec, perm[i] = row id at sorted pos i */
        } sort;
        struct {                /* RAY_IDX_ZONE */
            /* A column is integer XOR float, so the int and float extrema share
             * storage (keeps ray_index_t a 32-byte multiple for mmap layout). */
            union { int64_t min_i; double min_f; }; /* min (int: date/time too) */
            union { int64_t max_i; double max_f; }; /* max */
            int64_t n_nulls;    /* number of null rows (0 if no nulls) */
        } zone;
        struct {                /* RAY_IDX_BLOOM */
            ray_t*   bits;      /* RAY_U8 vec, m/8 bytes */
            uint64_t m_mask;    /* m - 1 (m is power of two, m bits total) */
            uint32_t k;         /* number of hash functions */
            uint32_t _pad;
            int64_t  n_keys;    /* number of non-null rows added */
        } bloom;
        struct {                /* RAY_IDX_CHUNK_ZONE */
            /* mins / maxs hold n_chunks entries.  For integer / temporal
             * column types they are RAY_I64 vecs storing the per-chunk
             * extrema as int64; for RAY_F64 columns they are RAY_F64
             * vecs.  is_f64 disambiguates at read time. */
            ray_t*   mins;
            ray_t*   maxs;
            ray_t*   null_bits;   /* RAY_U8 vec, packed: bit i = chunk i has any null */
            uint32_t n_chunks;
            uint8_t  chunk_log2;  /* chunk size = 1 << chunk_log2 (default 16 → 64 K rows) */
            uint8_t  is_f64;
            uint8_t  _pad[2];
            /* Integer / temporal zones only (NULL for float zones and for
             * indexes written before it existed): RAY_I64 vec of
             * 2 * n_chunks — [0, n) the chunk's non-null values summed with
             * int64 wraparound, [n, 2n) its non-null row count.  Whole-column
             * sum / count / avg answer from it in O(n_chunks). */
            ray_t*   aggs;
        } chunk_zone;
        struct {                /* RAY_IDX_PART */
            ray_t*  keys;       /* distinct partition values, in ascending block order */
            ray_t*  starts;     /* RAY_I64, row offset where each part begins */
            ray_t*  lens;       /* RAY_I64, row count of each part */
            int64_t n_parts;
            int64_t order_sym;  /* optional column symbol ordered within each part */
        } part;
        struct {                /* RAY_IDX_DICT */
            /* Both children are RAY_I32 (numeric — the inline persistence stores
             * them verbatim, no nested str_pool).  The distinct STRING values
             * are NOT duplicated: first_occ[c] is the parent-column row index of
             * code c's first occurrence, so code -> string resolves through the
             * parent column itself (ray_str_vec_get(col, first_occ[c])). */
            ray_t*   codes;     /* RAY_I32 vec, parent->len entries: code per row */
            ray_t*   first_occ; /* RAY_I32 vec, n_distinct entries: first-occ row  */
            int64_t  n_distinct;
        } dict;
        struct {                /* RAY_IDX_UKEY */
            /* Open addressing over the whole key tuple.  An occupied slot
             * packs the low 32 bits of the key's hash in its high half and
             * row + 1 in its low half (0 = empty, all-ones = tombstone), so
             * a probe rejects a colliding slot without reading its row, and
             * a map that outgrows its capacity is re-placed from the slots
             * themselves instead of being rebuilt from the key columns.
             * Rows are entered in row order, and the re-placement walks each
             * chain from its beginning, so a probe from the hash slot meets
             * duplicates lowest-row first — the row the linear scan would
             * have picked.
             *
             * `nrows` is the table row count this map describes.  Any
             * mutation that changes the row count leaves it behind and the
             * map is rebuilt; a rewrite of a key cell that keeps the count
             * must drop the index explicitly (upsert itself never rewrites
             * key cells — a matched row updates non-key columns only). */
            ray_t*   slots;     /* RAY_I64 vec, mask+1 entries */
            uint64_t mask;      /* capacity - 1 (capacity is a power of two) */
            int64_t  nrows;
            /* Slots vacated by a deleted row.  They are not free: clearing a
             * slot in the middle of a probe chain would orphan whatever was
             * displaced past it, so a deleted entry becomes UKEY_SLOT_TOMB and
             * probing walks over it.  Counted because a tombstone occupies a
             * slot without being a row, and a capacity test that ignored them
             * would let the array fill completely — at which point the
             * free-slot scan in upsert_map_put would not terminate. */
            int64_t  n_tomb;
            int64_t  nk;        /* number of key columns */
            /* Column positions.  int16 keeps this arm inside the hash arm's
             * 64 bytes now that n_tomb shares it; a table with more than
             * INT16_MAX columns simply gets no map. */
            int16_t  kci[RAY_UKEY_MAX_COLS];
        } ukey;
    } u;
} ray_index_t;

/* The ukey arm is sized to land exactly on the hash arm's 64 bytes so
 * ray_index_t does not grow: it is stored inline in persisted column files,
 * and a larger payload would change that on-disk layout for every kind. */
_Static_assert(sizeof(((ray_index_t*)0)->u.ukey) <= sizeof(((ray_index_t*)0)->u.hash),
               "RAY_IDX_UKEY payload must not grow ray_index_t");

/* On-disk index persistence stores the RAY_INDEX object and its child vecs as
 * contiguous 32-byte-aligned ray_t blocks mmap'd in place (no serialization).
 * For the trailing child blocks to stay 32-aligned, the index payload must be a
 * 32-byte multiple — enforce it so the layout invariant can't silently break. */
_Static_assert(sizeof(ray_index_t) % 32 == 0,
               "ray_index_t must be a 32-byte multiple for in-place mmap layout");

/* Inline accessor — returns ray_index_t* for a RAY_INDEX block. */
static inline ray_index_t* ray_index_payload(ray_t* idx) {
    return (ray_index_t*)idx->data;
}

/* The 64-bit mix every hash index uses (builder, probes, and the
 * generation-2 validator). */
static inline uint64_t ray_index_mix64(uint64_t x) {
    x ^= x >> 30; x *= 0xbf58476d1ce4e5b9ULL;
    x ^= x >> 27; x *= 0x94d049bb133111ebULL;
    x ^= x >> 31; return x;
}

/* ── Routing observability: per-site consult/hit counters ──
 * Diagnostic, unsynchronized (same caveat as ray_expr_bail_counts). */
typedef enum {
    IDX_SITE_FILTER_ZONE = 0, IDX_SITE_FILTER_BLOOM, IDX_SITE_FILTER_HASH,
    IDX_SITE_FILTER_PART, IDX_SITE_FILTER_RANGE, IDX_SITE_IN, IDX_SITE_FIND,
    IDX_SITE_SORT, IDX_SITE_DISTINCT, IDX_SITE_ASOF,
    IDX_SITE_FILTER_EQRANGE, IDX_SITE_GROUP_SLICE, IDX_SITE_ASOF_ORDERED, IDX_SITE__N
} idx_site_t;
extern uint64_t ray_idx_consults[IDX_SITE__N];
extern uint64_t ray_idx_hits[IDX_SITE__N];
void ray_idx_stats_init(void);   /* atexit dump when RAY_IDX_STATS set */

/* Per-index use attribution (diagnostic, for the index policy
 * measurement).  Counts one use of col->index; read and reset through the
 * `.idx.uses` builtin.  No-op for vectors without an index. */
void   ray_idx_note_use(const ray_t* col);
/* A consult answered from encv's directory (zone class or exclusion): counts
 * under the RAY_ENCCOL object itself. */
void   ray_idx_note_use_enc(const ray_t* encv);
ray_t* ray_idx_uses_fn(ray_t* tbl);

/* ===== Lazy in-memory HASH (index policy) =====
 *
 * The converters persist a CHUNK_ZONE, not a HASH.  Point lookups (equality
 * or `in` filters) on an 8-byte integer or timestamp column whose zone cannot
 * prune (the cost test admits no narrower type) build a HASH in memory once
 * they have been repeated enough, and reuse it in the same process.
 * The index lives in a process-wide cache keyed by the plain column object
 * (generation 0: the table's column; generation 2: the encoded object's
 * materialization), never in a table slot and never attached to the column,
 * so no save can see it.  An entry holds one reference to the column and an
 * indexed heap copy (ray_idx_hash_fn, copy-on-write); an entry whose column
 * only the cache still holds is dropped on the next probe.  The bytes held stay
 * within a quarter of ray_sys_ram_limit(); least recently used entries are
 * evicted to make room.  A column the budget or a failed build declines is
 * remembered, so later probes do not retry it.
 *
 * The build waits for proven repetition: the scan answers a wanted column's
 * probes, and the rows they read are counted in a pending entry, until that
 * scan work pays for one build.  Measured on 10M unclustered I64 rows (8
 * cores, release, page cache warm): the parallel in-memory build, column copy
 * included, takes 615 ms (about 61 ns/row); the equality scan it replaces
 * reads the mapped column in 1.9 ms (about 0.19 ns/row).  61 / 0.19 rounds
 * to 320 full-column probes: the 320th probe builds, the first 319 scan.
 * A cold scan costs more and pays sooner, so the warm ratio is the safe one.
 *
 * Threads: the cache takes its own lock, but its entries share column
 * objects with the caller's tables, so a caller outside the pool's workers
 * that probes while another thread uses the same objects must run with
 * ray_rc_sync set (atomic reference counts), as for any shared object. */
#define RAY_IDX_LAZY_BUILD_PROBES 320

/* The random-shape test the converters used: a fresh, integer CHUNK_ZONE of
 * at least 4 non-empty chunks whose mean chunk span exceeds half the global
 * range, and an index costing at most 5x the column's payload. */
bool   ray_idx_lazy_hash_wanted(const ray_t* col);
/* True when col's fresh integer chunk zone proves that no row equals key
 * (key outside its whole-column [min, max]), or, for _set, no row equals any
 * element of the integer vector set.  The caller has ruled out nulls. */
bool   ray_idx_zone_excludes(const ray_t* col, int64_t key);
bool   ray_idx_zone_excludes_set(const ray_t* col, const ray_t* set);
/* Directory twins for a single-file generation-2 column (encv a
 * RAY_ENCCOL): the answer the zone its materialization would synthesize
 * (enc_synth_zone / enc_synth_chunk_zone, src/store/enccol.c) gives,
 * from the same directory entries, never decoding.  UNKNOWN / false for a
 * parted column, another recorded kind, or a type the zone does not cover. */
bool   ray_enc_zone_excludes(ray_t* encv, int64_t key);
bool   ray_enc_zone_excludes_set(ray_t* encv, const ray_t* set);
bool   ray_enc_lazy_hash_wanted(ray_t* encv);
/* For a probe on `col`: NULL with *out = a retained vector equal to col
 * carrying a HASH (from the cache, or built now), or NULL with *out = NULL
 * when no lazy index applies (not wanted, over budget, declined before, or
 * still pending: the caller scans and this probe's work is counted); an
 * owned error on failure to build (the caller falls back to its scan). */
ray_t* ray_idx_lazy_hash(ray_t* col, ray_t** out);
/* ray_idx_lazy_hash for a single-file generation-2 column (encv its
 * RAY_ENCCOL): the probes are counted on the encoded payload (any header
 * copy), and the column is materialized only when the build fires; the
 * built entry is keyed by that materialization, and *out is its indexed
 * copy.  ray_idx_lazy_hash / _lookup / _work on the published
 * materialization meet the same entry. */
ray_t* ray_idx_lazy_hash_enc(ray_t* encv, ray_t** out);
/* The cached indexed copy for col, retained (the caller releases it), or NULL. */
ray_t* ray_idx_lazy_lookup(const ray_t* col);
/* Drop the entry of a column object that is going away (the generation-2
 * object frees its materialization). */
void   ray_idx_lazy_forget(const ray_t* col);
/* An encoded column's payload is freed: its memo (a probe count kept
 * before the column was materialized) goes with it. */
struct ray_enccol_s;
void   ray_idx_lazy_forget_enc(const struct ray_enccol_s* e);
/* Tests: rows the scans of col's pending probes read (0 once built), or -1
 * when col has no entry or was declined. */
int64_t ray_idx_lazy_work(const ray_t* col);
/* Test hook: the next lazy build fails with `code` ("oom" or "cancel");
 * NULL disarms. */
void    ray_idx_lazy_test_fail_build(const char* code);
/* Tests and runtime teardown: drop every entry; indexed entry count; every
 * entry (built, pending, declined); bytes held. */
void    ray_idx_lazy_clear(void);
int64_t ray_idx_lazy_count(void);
int64_t ray_idx_lazy_entries(void);
int64_t ray_idx_lazy_bytes(void);

/* ===== Attach / Detach ===== */

/* Build an accelerator and attach.  Numeric types are supported by every
 * index kind below; hash and part also support SYM domain ids.
 * On success, *vp is the (possibly new) parent vector with HAS_INDEX set.
 * On failure, *vp is unchanged and a RAY_ERROR is returned. */
ray_t* ray_index_attach_zone (ray_t** vp);
ray_t* ray_index_attach_hash (ray_t** vp);
ray_t* ray_index_attach_sort (ray_t** vp);
ray_t* ray_index_attach_bloom(ray_t** vp);
ray_t* ray_index_attach_part (ray_t** vp);
/* Build per-chunk min/max + null bit at chunk_size = 1 << chunk_log2.
 * Passing 0 picks the default (16 → 64 K rows / chunk).  Only valid on
 * numeric and temporal vectors; SYM/STR/GUID return RAY_ERR_NYI. */
ray_t* ray_index_attach_chunk_zone(ray_t** vp, uint8_t chunk_log2);

/* Build a chunk-zone index WITHOUT attaching it — returns a standalone
 * RAY_INDEX object (caller releases).  Used by the splayed-store builder to
 * compute an index for persistence without COWing a shared column. */
ray_t* ray_index_chunk_zone_compute(ray_t* v, uint8_t chunk_log2);

/* ===== Incremental chunk-zone accumulator ===== */

typedef struct {
    int8_t    type;       /* column type; 0 = inactive */
    uint8_t   is_f64;     /* F32 / F64 column: fmins/fmaxs used, no sums */
    int       esz;        /* element size in bytes */
    int64_t   next_row;   /* global row the next ray_zone_acc_add starts at */
    uint32_t  n_chunks;   /* chunks in use so far, indexed by global chunk */
    uint32_t  cap;        /* chunks allocated */
    int64_t*  mins;  int64_t* maxs;    /* integer zones */
    double*   fmins; double*  fmaxs;   /* float zones */
    bool      saw_null;   /* the column holds a sentinel/NaN; the writer must
                           * publish HAS_NULLS (col.c #495 rule) */
    uint8_t*  nulls;      /* one byte per chunk: 1 = any null in chunk */
    uint64_t* sum_lo; int64_t* sum_hi; int64_t* nn;   /* integer zones only */
} ray_zone_acc_t;

/* true when `type` gets a chunk zone (the ten numeric/temporal types). */
bool      ray_zone_acc_supported(int8_t type);
/* Start accumulating rows of `type` from global row `start_row`. */
ray_err_t ray_zone_acc_init(ray_zone_acc_t* a, int8_t type, int64_t start_row);
/* Rows [next_row, next_row + v->len) are v; v->type must equal a->type. */
/* A non-OK return (OOM, CANCEL) leaves the accumulator partial: free/discard it. */
ray_err_t ray_zone_acc_add(ray_zone_acc_t* a, ray_t* v);
/* dst += src, chunk by chunk; both must have the same type. src is untouched. */
ray_err_t ray_zone_acc_merge(ray_zone_acc_t* dst, const ray_zone_acc_t* src);
/* The standalone RAY_IDX_CHUNK_ZONE index for a column of `len` rows
 * (chunk_log2 16), identical to ray_index_chunk_zone_compute(col, 16);
 * NULL when len < 65536. Frees the accumulator either way. */
ray_t*    ray_zone_acc_finish(ray_zone_acc_t* a, int64_t len);
void      ray_zone_acc_free(ray_zone_acc_t* a);

/* 128-bit two's-complement accumulation of int64 values: (hi, lo) += v.
 * The engine's integer avg sums this way — exact for any column, and the
 * same bits whatever the morsel split — and the chunk-zone metadata keeps
 * the per-chunk (hi, lo) so it can answer the same value. */
static inline void ray_i128_add(int64_t* hi, uint64_t* lo, int64_t v) {
    uint64_t l = *lo + (uint64_t)v;
    *hi += (v < 0 ? -1 : 0) + (l < (uint64_t)v ? 1 : 0);
    *lo = l;
}
static inline void ray_i128_add128(int64_t* hi, uint64_t* lo, int64_t vhi, uint64_t vlo) {
    uint64_t l = *lo + vlo;
    *hi += vhi + (l < vlo ? 1 : 0);
    *lo = l;
}
/* Double nearest the 128-bit value (hi, lo): the magnitude is converted
 * (high word scaled by 2^64 plus the low word) and the sign reapplied, so
 * a small negative total is not lost in 2^64 - |s|.  One formula
 * everywhere so every path agrees bit for bit. */
static inline double ray_i128_to_f64(int64_t hi, uint64_t lo) {
    bool neg = hi < 0;
    uint64_t h = (uint64_t)hi, l = lo;
    if (neg) { l = ~l + 1u; h = ~h + (l == 0 ? 1u : 0u); }
    double d = (double)h * 18446744073709551616.0 + (double)l;
    return neg ? -d : d;
}

/* Whole-column sum (int64 wraparound) and non-null count of an integer
 * column from its chunk-zone per-chunk aggregates; false when the column
 * carries none for its current length. */
bool ray_zone_int_sum(ray_t* x, int64_t* sum_out, int64_t* nn_out);
/* The exact 128-bit sum (hi, lo) and non-null count from the same
 * metadata: true when the per-chunk high words are stored, or when no
 * partial sum can wrap int64 (then the wrapped sum is the exact one). */
bool ray_zone_int_sum128(ray_t* x, int64_t* hi_out, uint64_t* lo_out, int64_t* nn_out);
/* OP_SUM / OP_AVG / OP_MIN / OP_MAX of a single-file generation-2 column
 * whose recorded index is CHUNK_ZONE, answered as ray_sum_fn / ray_avg_fn /
 * ray_min_fn / ray_max_fn answer the generation-0 column from its
 * CHUNK_ZONE (the zone the materialization would synthesize), from the
 * directory; NULL when that zone path would not answer. */
ray_t* ray_zone_dir_answer(ray_t* encv, uint16_t opcode);
/* ray_zone_int_sum128's twin for the same columns: the exact 128-bit sum
 * and non-null count from the directory; false when the zone the
 * materialization would synthesize carries no integer sums. */
bool ray_zone_dir_int_sum128(ray_t* encv, int64_t* hi, uint64_t* lo, int64_t* nn);
/* ray_zone_dir_answer's answer for every single-file stored generation-2
 * column, with or without a recorded zone: the execution-time reduction
 * (ray_reduce_src) takes it.  Planning-time answers stay on
 * ray_zone_dir_answer, so they do not widen past generation 0's. */
ray_t* ray_dir_answer(ray_t* encv, uint16_t opcode);
/* The directory totals (a ray_enc_col_totals_t, store/enccol.h) of one
 * parted segment for an integer or temporal SUM / AVG / MIN / MAX: true
 * when the segment is a single-file stored generation-2 column the
 * directory answers `opcode` for, false for a plain, synthetic or float
 * segment (the caller reduces its rows). */
bool ray_dir_seg_totals(ray_t* segv, uint16_t opcode, void* totals_out);

/* Build a RAY_IDX_DICT (codes + distinct values) for STR vector `v` WITHOUT
 * attaching it — standalone RAY_INDEX object (caller releases / attaches).
 * Returns RAY_ERR_NYI for non-STR.  Used at splayed save to persist the dict
 * and by ray_index_attach_dict for the runtime path. */
ray_t* ray_index_dict_compute(ray_t* v);
ray_t* ray_index_attach_dict(ray_t** vp);

/* ===== Incremental STR dictionary ===== */

typedef struct ray_dict_acc_s {
    int32_t*  codes;     int64_t n_rows;     int64_t cap_rows;
    int32_t*  first_occ; int64_t n_distinct; int64_t cap_distinct;
    uint64_t* offs;      /* per distinct: byte offset into pool */
    uint32_t* lens;      /* per distinct: byte length */
    char*     pool;      uint64_t pool_len;  uint64_t pool_cap;
    uint32_t* slot;      uint64_t mask;      /* code + 1, 0 = empty */
    bool      dead;      /* row count overflowed int32 codes: nothing is kept */
} ray_dict_acc_t;
ray_err_t ray_dict_acc_init(ray_dict_acc_t* a);
/* Appends every row of STR vector v. There is no cardinality cap: the
 * result matches ray_index_dict_compute for any number of distinct strings.
 * Only a row count past INT32_MAX (codes are int32) frees the accumulator,
 * sets dead and returns RAY_OK; further adds are no-ops. A non-OK return
 * (OOM, CANCEL) leaves the accumulator partial: free/discard it. */
ray_err_t ray_dict_acc_add(ray_dict_acc_t* a, ray_t* v);
/* Appends n raw STR descriptors whose pooled bytes are relative to `pool`
 * (ray_str_t_ptr): the same codes ray_dict_acc_add assigns to the same
 * strings, with the same dead / error rules. */
ray_err_t ray_dict_acc_add_raw(ray_dict_acc_t* a, const ray_str_t* d, const char* pool, int64_t n);
/* The fewest rows a saved generation-0 STR column carries its DICT index at. */
#define RAY_DICT_MIN_ROWS 65536
/* The standalone RAY_IDX_DICT for a column of `len` rows, identical to
 * ray_index_dict_compute; NULL when dead or len < RAY_DICT_MIN_ROWS. Frees `a`. */
ray_t*    ray_dict_acc_finish(ray_dict_acc_t* a, int64_t len);
void      ray_dict_acc_free(ray_dict_acc_t* a);

/* Attach an already-built standalone RAY_INDEX object (zero-copy on rc=1). */
ray_t* ray_index_attach_built(ray_t** vp, ray_t* idx);
/* Attach a heap-built root to a fresh plain vector.  Consumes `root` on
 * success and on error; refuses (corrupt) a pair the matrix does not
 * admit, a root whose parent_type or built_for_len disagree, a slice, or
 * a vector already indexed. */
ray_t* ray_index_attach_persisted(ray_t** vp, ray_t* root);
/* A zeroed RAY_INDEX block of `kind` for a parent (public wrapper of the
 * allocator used by every builder). */
ray_t* ray_index_new_root(ray_idx_kind_t kind, int8_t parent_type, int64_t parent_len);
/* Spec "Materialization" step 3 admission matrix:
 * numeric and temporal types carry HASH, SORT, BLOOM, PART, ZONE,
 * CHUNK_ZONE and a marker-only root; SYM carries HASH, PART and a
 * marker-only root; STR carries HASH, DICT (persisted for generation 2
 * since P6b) and a marker-only root; GUID carries nothing. */
bool ray_index_kind_admits(ray_idx_kind_t kind, int8_t parent_type);

/* ── Inline on-disk index region (zero-copy mmap) ──
 * ray_index_inline_size: bytes the region occupies (32-aligned ray_t blocks).
 * ray_index_inline_write: serialize `ix` into dst (child ptrs → region offsets).
 * ray_index_inline_map: patch an mmap'd region's child offsets → absolute ptrs
 *   in place and return the RAY_INDEX object (flagged RAY_MARK_MMAP). */
int64_t ray_index_inline_size(const ray_index_t* ix);
void    ray_index_inline_write(uint8_t* dst, const ray_index_t* ix);
/* The same bytes as ray_index_inline_write, streamed to `f` block by block
 * straight from the child vecs: no region-sized staging buffer (a 100 M-row
 * hash region is ~4.5 GB) and one copy fewer.  false on a short write. */
bool    ray_index_inline_write_file(FILE* f, const ray_index_t* ix);
ray_t*  ray_index_inline_map(uint8_t* region, int64_t region_size);

/* ── Hash-index build trace (RAY_CSV_TRACE) ──
 * Per phase of one build: wall time, major faults, the bytes the process
 * read and wrote (/proc/self/io; zero off Linux), and `moved`, the bytes the
 * phase's own passes read and write by design (each array a pass reads or
 * writes in full, once per pass) — the I/O it costs when nothing fits in
 * memory.  count: the column read for the partition histogram; bucket: the
 * column read again into key words and row ids; distinct: the groups of
 * each partition counted; groups: deduped and stored, and the slot table
 * filled alongside when its size is known; table: the slot table after the
 * groups, and the arrays cut to size; append: the region into the file. */
enum { RAY_HXT_COUNT, RAY_HXT_BUCKET, RAY_HXT_DISTINCT,
       RAY_HXT_GROUPS, RAY_HXT_TABLE, RAY_HXT_APPEND, RAY_HXT_N };
typedef struct {
    int64_t ns[RAY_HXT_N], majflt[RAY_HXT_N], rd[RAY_HXT_N], wr[RAY_HXT_N], moved[RAY_HXT_N];
    int64_t cwr[RAY_HXT_N];     /* dirty bytes dropped before their writeback */
    int64_t n_keys, n_groups;   /* the index built */
} ray_hash_trace_t;
typedef struct { int64_t ns, majflt, rd, wr, cwr; } ray_hash_mark_t;
ray_hash_mark_t ray_hash_trace_mark(void);
/* Phase `ph` ran since `since` and moved `moved` bytes. */
void ray_hash_trace_add(ray_hash_trace_t* t, int ph, ray_hash_mark_t since, int64_t moved);
/* One line: each phase that ran, `moved` also as a multiple of `col_bytes`. */
void ray_hash_trace_print(FILE* f, const char* what, const ray_hash_trace_t* t, int64_t col_bytes);

/* Build the hash index of numeric or SYM vector `v` straight into its
 * persisted inline region, without attaching it and without staging the
 * arrays in memory: once the region's size is known, `map(ctx, bytes)`
 * returns `bytes` of zero-filled writable memory (NULL when it cannot),
 * and the build writes there exactly the bytes ray_index_inline_write would
 * write for the index ray_index_attach_hash builds on `v`.  RAY_ERR_NYI for
 * a type the partitioned builder does not take (STR) — nothing was mapped;
 * RAY_ERR_IO when map failed, RAY_ERR_OOM / RAY_ERR_CANCEL when the build
 * did: the mapped memory, if any, holds no complete region.  `tr` (may be
 * NULL) collects the phases. */
typedef uint8_t* (*ray_index_region_fn)(void* ctx, int64_t bytes);
/* The arrays of HASH payload `ix` in the generation-1 layout a
 * generation-2 index region stores: every array I64, the groups numbered by
 * first occurrence (row order of their first rows), each group in the first
 * free slot from mix64(key) & mask (the low-bits home rule): the bytes the
 * generation-1 builders made.  Shared (retained) when ix is that layout
 * already; fresh otherwise.  False on OOM (outputs NULL). */
bool ray_index_hash_legacy(const ray_index_t* ix, ray_t** table, ray_t** gkeys,
                           ray_t** offs, ray_t** rows);

ray_err_t ray_index_hash_build_region(ray_t* v, ray_index_region_fn map, void* ctx,
                                      ray_hash_trace_t* tr);

/* Drop any attached index from *vp.  No-op if none.  Restores the
 * pre-attach aux state byte-for-byte.  Returns *vp. */
ray_t* ray_index_drop(ray_t** vp);

/* ===== Table-resident key index (RAY_IDX_UKEY) =====
 * Attached to a RAY_TABLE, not to a vector, so the generic attach path does
 * not apply: a table's aux is zero-init, there is nothing to snapshot and
 * restore.  Retain/release ride the type-agnostic RAY_ATTR_HAS_INDEX arms in
 * heap.c, so the index dies with the table. */

/* Build a standalone RAY_IDX_UKEY sized for `entries` rows (caller attaches
 * or releases).  The slot vector is allocated zeroed. */
ray_t* ray_index_build_ukey(const int64_t* kci, int64_t nk, int64_t entries);

/* Attach `idx` to `tbl`, taking ownership of the caller's reference.
 * Returns false (and keeps the caller's reference) if `tbl` already carries
 * an index — the caller drops first. */
bool ray_table_ukey_attach(ray_t* tbl, ray_t* idx);

/* The attached ukey payload, or NULL when the table carries no index or one
 * of another kind. */
ray_index_t* ray_table_ukey_get(ray_t* tbl);

/* Detach and release any index on `tbl`.  Safe on a table with none. */
void ray_table_ukey_drop(ray_t* tbl);

/* ===== Introspection ===== */

static inline bool ray_index_has(const ray_t* v) {
    return v && !RAY_IS_ERR((ray_t*)v) &&
           (v->attrs & RAY_ATTR_HAS_INDEX) &&
           v->index != NULL;
}

/* Returns RAY_IDX_NONE if no index is attached. */
static inline ray_idx_kind_t ray_index_kind(const ray_t* v) {
    if (!ray_index_has(v)) return RAY_IDX_NONE;
    return (ray_idx_kind_t)ray_index_payload(v->index)->kind;
}

/* --- Semantic attribute markers (see mem/heap.h: RAY_ATTR_SORTED) --- */

/* True iff v is a vector flagged sorted (non-descending). */
static inline bool ray_attr_is_sorted(const ray_t* v) {
    return v && !RAY_IS_ERR((ray_t*)v) && ray_is_vec(v)
        && (v->attrs & RAY_ATTR_SORTED);
}

/* Returns a fresh RAY_DICT with {kind, length, ...kind-specific...}
 * or RAY_NULL_OBJ when no index is attached. */
ray_t* ray_index_info(ray_t* v);

/* ===== Zone-index all/none classification =====
 *
 * O(1) pre-check: given the zone index on `col` and a comparison
 * (col cmp_op key), decide whether the predicate is definitely true for
 * ALL rows, definitely false for ALL rows (NONE), or UNKNOWN (need scan).
 *
 * cmp_op is the OP_EQ..OP_GE opcode.  key_i is used for integer/temporal
 * column families; key_f for float families.  NaN key_f → UNKNOWN.
 * Returns UNKNOWN when the column is not eligible (no zone index, stale,
 * null-bearing, unsupported type). */
typedef enum { RAY_ZONE_UNKNOWN = 0, RAY_ZONE_ALL = 1, RAY_ZONE_NONE = 2 } ray_zone_class_t;
ray_zone_class_t ray_index_zone_class(ray_t* col, uint16_t cmp_op,
                                      int64_t key_i, double key_f);
/* Its directory twin for a generation-2 column recording ZONE (see
 * ray_enc_zone_excludes): UNKNOWN unless encv is a single-file RAY_ENCCOL
 * without nulls. */
ray_zone_class_t ray_enc_zone_class(ray_t* encv, uint16_t cmp_op, int64_t key_i, double key_f);

/* Build an all-NONE rowsel of `n` rows.  Wraps rowsel_from_sorted_ids(n, NULL, 0).
 * Returns NULL on OOM; returns a valid zero-match rowsel on success. */
ray_t* ray_index_empty_rowsel(int64_t n);

/* ===== Bloom-index definite-absent probe =====
 *
 * Returns true ONLY when the bloom filter PROVES the key is absent
 * (all k probe bits are clear).  Returns false when the key may be
 * present OR when the column is not eligible — caller proceeds to
 * other consults / the scan either way.
 *
 * Integer-family columns only in v1 (mirrors the hash path's
 * conservatism on float equality: F32/F64 NaN/-0 semantics).
 * No range validation is performed: an out-of-range key hashes to
 * some bit positions; if they are set we fall through — the absent
 * proof is still sound (no false absent is possible). */
bool ray_index_bloom_absent(ray_t* col, int64_t key);

/* ===== Hash-index point-lookup probe =====
 *
 * Build a ray_rowsel directly from a hash probe on `col`'s
 * RAY_IDX_HASH for rows where the payload equals `key`.  Bypasses
 * the intermediate BOOL pred vec entirely — touches O(matches)
 * memory instead of O(rows), which is the whole reason to ship
 * this fast path.
 *
 * Returns:
 *   - A fresh rowsel block (rc=1) on success — install on
 *     g->selection.  The block carries per-segment NONE/MIX/ALL
 *     flags and the morsel-local indices for matching rows.
 *     Pure NONE blocks (no matches) are returned as a valid empty
 *     rowsel rather than NULL — NULL is the "all-pass" sentinel
 *     in the consumer and would let every row through.
 *   - NULL when the column is not eligible: no index, wrong kind,
 *     built_for_len mismatch (stale), type mismatch, or out-of-
 *     range key.  Caller must fall back to the full scan path.
 *
 * Eligibility (and the canonical hashing used) match
 * ray_index_attach_hash: BOOL/U8/I16/I32/I64/DATE/TIME/TIMESTAMP.
 * Floats are intentionally not supported — equality on F32/F64
 * has NaN / -0 semantics the unfused compare kernel handles. */
ray_t* ray_index_hash_eq_rowsel(ray_t* col, int64_t key);
/* ray_index_hash_eq_rowsel for a single-file generation-2 column whose
 * recorded HASH region is probed where it is mapped, without materializing
 * the column or copying the region: the root's layout is checked against
 * the region, every slot, group and offset the probe reads is bounded, and
 * the group's rows are checked ascending, inside the column, and holding
 * `key` (read back by range: a region that disagrees with the column is
 * `corrupt`, naming it).  `key` as for ray_index_hash_eq_rowsel (a SYM key
 * a nonzero domain id).  1: *sel the selection (an empty one for an absent
 * key); 0 declined (no such region, a key or type it does not cover, or a
 * group dense enough for the scan, as ray_index_hash_eq_rowsel declines);
 * -1: *err owned. */
int    ray_enc_hash_eq_rowsel(ray_t* encv, int64_t key, ray_t** sel, ray_t** err);
/* Test hook: probes ray_enc_hash_eq_rowsel answered since process start. */
uint64_t ray_enc_hash_test_probes(void);

/* Direct equality selection for a fresh part index.  `key` is a numeric
 * value or a column-domain SYM id.  Dense matches remain eligible because
 * the part payload already stores one contiguous physical span. */
ray_t* ray_index_part_eq_rowsel(ray_t* col, int64_t key);

/* ===== Hash/part-index IN probe =====
 *
 * Build a rowsel of all rows whose value is in set_vec via per-element hash
 * probes or direct physical spans from a part index.  Integer-family and SYM
 * columns (same conservatism as the hash-eq path: F32/F64 NaN/-0 semantics
 * belong to the scan kernel).
 *
 * set_vec must match the column family: an integer-family typed vec for an
 * integer column, a SYM vec for a SYM column (each set symbol is resolved
 * through the COLUMN's domain; domain-absent symbols match no rows).
 * Other combinations → NULL.
 *
 * Returns:
 *   - A fresh rowsel block (rc=1) on success — install on g->selection.
 *     An empty set yields a valid all-NONE rowsel (NOT NULL).
 *   - NULL when the column is not eligible (no hash/part index, stale,
 *     float-family column, unsupported set type, OOM) or a hash-index union
 *     is too dense.  Part-index unions retain dense full morsels efficiently.
 *     Caller falls back to the scan path. */
ray_t* ray_index_in_rowsel(ray_t* col, ray_t* set_vec);

/* ===== Hash-index find (point lookup) =====
 *
 * Returns the minimum row id where the column value equals `key`, or
 * -1 when the index proves the key is absent (provably-no-match).
 * Returns -2 when the column is not eligible: no hash index, stale,
 * float-family column, or out-of-eligibility condition — caller must
 * fall back to the linear scan.
 *
 * Contract:
 *   >= 0  → key found; this is the FIRST (minimum) row id match.
 *   -1    → key provably absent; caller returns find's miss value.
 *   -2    → not eligible; caller falls back to the scan.
 *
 * Uses idx_fresh_nonull: null-bearing columns fall back (-2) so the
 * scan correctly surfaces null-equality searches. */
int64_t ray_index_find_row(ray_t* col, int64_t key);

/* ===== Hash-index find: one atom / a vector of needles =====
 *
 * ray_index_find_atom: like ray_index_find_row but keyed by an ATOM, so it
 * also serves SYM columns (the symbol re-expressed in the column's domain)
 * and STR columns (the bytes).  Same contract: >= 0 first row, -1 provably
 * absent, -2 not eligible (no fresh null-free hash index, null / float /
 * cross-family needle) — caller falls back to the scan.
 *
 * ray_index_find_vec: one probe per needle of the typed vector `needles`
 * into `out[needles->len]` (first matching row, NULL_I64 on a miss; a null
 * needle misses — the index is only consulted on null-free columns).
 * Returns 1 when handled, 0 when not eligible (out is then unspecified and
 * the caller builds its own hash).  Integer-family needles for an
 * integer-family column, SYM for SYM (any domain), STR for STR. */
int64_t ray_index_find_atom(ray_t* col, ray_t* atom);
int ray_index_find_vec(ray_t* col, ray_t* needles, int64_t* out, bool* any_miss);

/* ===== Hash-index carry across an append =====
 *
 * `dst` is a fresh (rc=1, unsliced, unindexed, null-free) vector whose first
 * src->len rows are src's rows and whose tail is appended rows.  When src
 * carries a fresh null-free hash index and every appended row is a NEW key,
 * dst gets an equivalent index in O(groups + appended) without re-hashing the
 * old rows; markers (`unique`) and the within-group order symbol carry over.
 * A repeated key, a float column, a SYM domain change or OOM leave dst
 * unindexed (the caller re-attaches if it wants one).  Best-effort: never
 * fails the append. */
void ray_index_carry_append(ray_t* src, ray_t* dst);

/* Row ids borrowed from a hash index: 64-bit, or unsigned 32-bit when the
 * index was built narrow (any column below 2^32 rows).  Read with
 * ray_idx_rows_at. */
typedef struct {
    const void* p;
    bool        narrow;
} ray_idx_rows_t;

static inline int64_t ray_idx_rows_at(ray_idx_rows_t r, int64_t i) {
    return r.narrow ? (int64_t)((const uint32_t*)r.p)[i]
                    : ((const int64_t*)r.p)[i];
}
/* The ids from position `off` on. */
static inline ray_idx_rows_t ray_idx_rows_from(ray_idx_rows_t r, int64_t off) {
    r.p = (const uint8_t*)r.p + off * (r.narrow ? 4 : 8);
    return r;
}

/* ===== Hash-index group slice (CSR accessor) =====
 *
 * Resolve `key` to its group's contiguous ascending row-id slice.
 * SYM columns take the COLUMN-DOMAIN id as key; numeric columns take the
 * value (canonicalized through the storage width like the eq probe).
 *
 * Returns:
 *    1 → hit: *rows_out holds the ascending row ids, *n_out the size.
 *        The slice borrows the index payload — valid while the index is.
 *    0 → key provably absent (rows_out->p NULL, *n_out 0).
 *   -1 → not eligible (no fresh hash index / bad key) — caller falls back. */
int ray_index_hash_group(ray_t* col, int64_t key,
                         ray_idx_rows_t* rows_out, int64_t* n_out);

/* Build a rowsel from ASCENDING row ids (empty n=0 is valid).  NULL on OOM. */
ray_t* ray_index_rowsel_from_ids(int64_t nrows, const int64_t* ids, int64_t n);
/* Same, from row ids borrowed from an index. */
ray_t* ray_index_rowsel_from_rows(int64_t nrows, ray_idx_rows_t ids, int64_t n);

/* ===== Slice-group resolution (FILTER + GROUP-by-key fusion) =====
 *
 * One surviving group of the fused shape: the key's column-domain id and
 * its CSR row slice (ascending row ids, borrowed from the index payload —
 * valid while the index is). */
typedef struct {
    int64_t        dom;   /* column-domain key id */
    ray_idx_rows_t rows;  /* ascending row ids (borrowed); .p NULL if contiguous */
    int64_t        first; /* first physical row when rows.p == NULL */
    int64_t        n;     /* slice length (> 0) */
} ray_idx_slice_t;

/* Resolve a SYM eq key (sym ATOM, global intern id) or IN set (SYM vec,
 * any domain) against `col`'s fresh hash or part index into per-key slices,
 * canonicalized ascending by column-domain id (duplicates collapsed;
 * absent / rowless keys dropped). Hash slices borrow CSR row ids; part
 * slices use rows==NULL plus their contiguous physical first row. On success
 * returns K >= 0 and, for
 * K > 0, *out points into a fresh alloc block returned via *hdr_out
 * (release with ray_free).  K == 0 → provably no surviving rows (*out
 * and *hdr_out NULL).  Returns -1 when ineligible: stale or missing
 * hash/part index, null-bearing or non-SYM column, wrong key shape. */
int64_t ray_index_sym_slices(ray_t* col, ray_t* keys,
                             ray_idx_slice_t** out, ray_t** hdr_out);

/* ===== Sort-index range probe =====
 *
 * Build a rowsel from a binary search over the sort-index permutation for
 * rows satisfying (col cmp_op key).  Supports EQ/LT/LE/GT/GE.  NE returns
 * NULL (two disjoint spans — unsupported).
 *
 * Returns:
 *   - A fresh rowsel block on success — install on g->selection.
 *     span == 0 yields a valid all-NONE rowsel (NOT NULL).
 *   - NULL when not eligible: no sort index, stale, type mismatch,
 *     span > col->len / IDX_RANGE_MAX_FRAC (selectivity guard), OOM,
 *     or NE operator.  Caller falls back to the scan.
 *
 * key_i is used for integer-family columns; key_f for float-family.
 * is_float must match the column family (mirror of idx_filter_decode
 * contract); mismatch returns NULL defensively. */
ray_t* ray_index_range_rowsel(ray_t* col, uint16_t cmp_op,
                              int64_t key_i, double key_f, bool is_float);

/* Build a rowsel by binary-searching a physically non-descending column.
 * Unlike the sort-index route above, matching row ids are already one
 * contiguous interval, so dense spans are eligible and the rowsel stores
 * only the at-most-two partial boundary morsels.  Requires the verified
 * RAY_ATTR_SORTED marker and a null-free numeric column. */
ray_t* ray_sorted_range_rowsel(ray_t* col, uint16_t cmp_op,
                               int64_t key_i, double key_f, bool is_float);

/* ===== Sort-index ORDER BY permutation probe =====
 *
 * Returns a BORROWED pointer to the ascending permutation vector stored in
 * the sort index on `col` when the column carries a fresh, null-free sort
 * index; NULL otherwise.  The caller MUST NOT release the returned pointer
 * (it is owned by the index payload).  To use it as an owned reference,
 * call ray_retain() immediately after. */
ray_t* ray_index_sort_perm_fresh(ray_t* col);

/* ===== Sort-index distinct fast path =====
 *
 * Walk the sort permutation once to collect the first-occurrence row id of
 * each distinct value (the first element of each equal-value run, which is
 * the minimum row id of the run because the sort is stable with iota
 * tie-breaking).  Returns an OWNED I64 vector of those row ids, sorted
 * ascending by VALUE (matching distinct_vec_eager's qsort-of-indices
 * contract), and sets *out_count to the number of distinct values.
 *
 * Returns NULL if not eligible (no sort index, stale, null-bearing) or on
 * OOM.  The caller is responsible for releasing the returned vector. */
ray_t* ray_index_distinct_ids(ray_t* col, int64_t* out_count);

/* ===== Internal helpers (used by retain/release/detach in heap.c
 * and by mutation paths in vec.c) ===== */

/* Release the saved-aux pointers carried by a RAY_INDEX ray_t.
 * Invoked from ray_release_owned_refs when the index ray_t is freed. */
void ray_index_release_saved(ray_index_t* ix);

/* Retain the saved-aux pointers carried by a RAY_INDEX ray_t.
 * Invoked from ray_retain_owned_refs after a copy of the index ray_t. */
void ray_index_retain_saved(ray_index_t* ix);

/* Release per-kind payload children (keys/table/perm/bits...). */
void ray_index_release_payload(ray_index_t* ix);
/* The payload child blocks of an index (the same set retain/release walk),
 * written to out[0..cap) — returns the count (at most 4).  Footprint
 * accounting (ray_shallow_bytes / ray_retained_bytes) sums them; a
 * RAY_MARK_MMAP index's children live inside the column's mapping and are
 * charged there, so callers skip them. */
int  ray_index_child_blocks(const ray_index_t* ix, ray_t** out, int cap);

/* Retain per-kind payload children. */
void ray_index_retain_payload(ray_index_t* ix);

/* ===== Rayfall builtin entry points (registered from src/lang/eval.c) ===== */

ray_t* ray_idx_zone_fn (ray_t* v);  /* (.idx.zone  v) -> v with zone  attached */
ray_t* ray_idx_hash_fn (ray_t* v);  /* (.idx.hash  v) -> v with hash  attached */
ray_t* ray_idx_sort_fn (ray_t* v);  /* (.idx.sort  v) -> v with sort  attached */
ray_t* ray_idx_bloom_fn(ray_t* v);  /* (.idx.bloom v) -> v with bloom attached */
ray_t* ray_idx_drop_fn (ray_t* v);  /* (.idx.drop  v) -> v with index removed */
ray_t* ray_idx_has_fn  (ray_t* v);  /* (.idx.has?  v) -> 0b/1b */
ray_t* ray_idx_info_fn (ray_t* v);  /* (.idx.info  v) -> dict of metadata */
ray_t* ray_idx_info_vary_fn(ray_t** args, int64_t n);  /* (.idx.info col) | (.idx.info table 'col) */

ray_t* ray_attr_set_fn (ray_t* name, ray_t* v); /* (.attr.set 'name v) */
ray_t* ray_attr_get_fn (ray_t* v);              /* (.attr.get v) -> sym vec */
ray_t* ray_attr_drop_fn(ray_t* v);              /* (.attr.drop v) -> v cleared */

#endif /* RAY_IDXOP_H */