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

#include "table.h"
#include "mem/heap.h"
#include "ops/ops.h"
#include "lang/format.h"
#include "ops/idxop.h"
#include "store/enccol.h"
#include "vec/vec.h"
#include <string.h>

/* --------------------------------------------------------------------------
 * Data layout — same shape as RAY_DICT.
 *
 * Block header (32 B):  type = RAY_TABLE, len = 2
 *   slot[0] = ray_t* schema    — RAY_I64 vector of column name sym IDs
 *   slot[1] = ray_t* cols      — RAY_LIST of column vectors
 *
 * `tbl->len` is the slot count (always 2).  Use ray_table_ncols() to get
 * the column count, ray_table_nrows() for the row count.
 *
 * The schema vector stays RAY_I64 (rather than RAY_SYM) because the rest
 * of the codebase reads it as `int64_t* ids = ray_data(schema)` in
 * dozens of hot loops; RAY_SYM's adaptive widths (W8/W16/W32/W64) would
 * silently truncate those reads.  RAY_DICT is free to use any keys type;
 * TABLE deliberately pins schema to I64 for that interop.
 * -------------------------------------------------------------------------- */

#define TBL_DATA_SIZE  (2 * sizeof(ray_t*))

static inline ray_t** tbl_slots(ray_t* tbl) {
    return (ray_t**)ray_data(tbl);
}

/* The public accessors below decode `tbl`'s payload as the two-slot
 * {schema, cols} layout, so they are only meaningful for a RAY_TABLE.  An
 * embedder sees one opaque ray_t* and has nothing to check against, so each
 * accessor is total over ray_t: a wrong tag gets the empty answer, never a
 * read through a decoded payload (#567).  Matches ray_dict_keys and
 * ray_parted_nrows, which already check their own tags.  Internal callers
 * establish the type during dispatch and pay one predicted compare against a
 * field already in the same cache line as `type`. */
static inline bool tbl_is_table(ray_t* tbl) {
    return tbl && !RAY_IS_ERR(tbl) && tbl->type == RAY_TABLE;
}

static inline ray_t* tbl_schema(ray_t* tbl) {
    return tbl_slots(tbl)[0];
}

static inline ray_t* tbl_cols(ray_t* tbl) {
    return tbl_slots(tbl)[1];
}

static bool table_col_is_valid(ray_t* col) {
    if (!col || RAY_IS_ERR(col)) return false;
    return col->type == RAY_LIST ||
           ray_is_vec(col) ||
           RAY_IS_PARTED(col->type) ||
           col->type == RAY_MAPCOMMON ||
           col->type == RAY_ENCCOL;          /* single or parted; len is logical */
}

static int64_t table_col_nrows(ray_t* col) {
    if (!col || ray_is_atom(col)) return 0;      /* an atom has no rows (its len field is payload) */
    if (col->type == RAY_ENCCOL) return col->len;   /* no decode */
    if (RAY_IS_PARTED(col->type) || col->type == RAY_MAPCOMMON)
        return ray_parted_nrows(col);
    return col->len;
}

int64_t ray_table_col_nrows(ray_t* col) { return table_col_nrows(col); }

/* --------------------------------------------------------------------------
 * ray_table_new — allocates an empty table with capacity for `ncols`.
 *
 * The schema vector and cols list are pre-sized to avoid early grows.
 * Callers append columns via ray_table_add_col.
 * -------------------------------------------------------------------------- */

ray_t* ray_table_new(int64_t ncols) {
    if (ncols < 0) return ray_error("range", "table_new: column count must be non-negative, got %lld", (long long)ncols);

    ray_t* tbl = ray_alloc(TBL_DATA_SIZE);
    if (!tbl) return ray_error("oom", "table_new(ncols=%lld)", (long long)ncols);
    if (RAY_IS_ERR(tbl)) return tbl;
    tbl->type  = RAY_TABLE;
    tbl->attrs = 0;
    tbl->len   = 2;
    memset(tbl->aux, 0, 16);
    memset(ray_data(tbl), 0, TBL_DATA_SIZE);

    ray_t* schema = ray_vec_new(RAY_I64, ncols);
    if (!schema || RAY_IS_ERR(schema)) {
        ray_free(tbl);
        return schema ? schema : ray_error("oom", NULL);
    }
    ray_t* cols = ray_list_new(ncols);
    if (!cols || RAY_IS_ERR(cols)) {
        ray_release(schema);
        ray_free(tbl);
        return cols ? cols : ray_error("oom", NULL);
    }
    tbl_slots(tbl)[0] = schema;
    tbl_slots(tbl)[1] = cols;
    return tbl;
}

/* --------------------------------------------------------------------------
 * ray_table_add_col — append `col_vec` under name `name_id`.
 *
 * Consumes one ref of the input table; on success returns an owned ref to
 * the (possibly COW'd) result.  Retains `col_vec` internally so the caller
 * keeps its own ref.
 * -------------------------------------------------------------------------- */

ray_t* ray_table_add_col(ray_t* tbl, int64_t name_id, ray_t* col_vec) {
    if (!tbl || RAY_IS_ERR(tbl)) return tbl;
    /* Unlike the accessors, this consumes `tbl` and owns its result, so the
     * wrong-tag answer has to be a typed error rather than an empty one —
     * same consume-and-error contract as the bad-column path below.  Without
     * it a non-table gets ray_cow'd and then has slot pointers written into
     * its payload (#567). */
    if (tbl->type != RAY_TABLE) {
        ray_t* err = ray_error("type", "table add_col: expected table, got %s",
                               ray_type_name(tbl->type));
        ray_release(tbl);
        return err;
    }
    if (!table_col_is_valid(col_vec)) {
        ray_release(tbl);
        return ray_error("domain", "table add_col: column must be list/vector-like, got %s",
                         col_vec ? ray_type_name(col_vec->type) : "null");
    }

    tbl = ray_cow(tbl);
    if (!tbl || RAY_IS_ERR(tbl)) return tbl;

    ray_t** slots = tbl_slots(tbl);
    ray_t* schema = slots[0];
    ray_t* cols   = slots[1];

    /* schema and cols may themselves be shared after cow — append helpers
     * COW them as needed and return the (possibly new) owned ref. */
    ray_t* new_schema = ray_vec_append(schema, &name_id);
    if (!new_schema || RAY_IS_ERR(new_schema)) { ray_release(tbl); return new_schema ? new_schema : ray_error("oom", NULL); }
    slots[0] = new_schema;

    ray_retain(col_vec);
    ray_t* new_cols = ray_list_append(cols, col_vec);
    ray_release(col_vec);
    if (!new_cols || RAY_IS_ERR(new_cols)) { ray_release(tbl); return new_cols ? new_cols : ray_error("oom", NULL); }
    slots[1] = new_cols;

    return tbl;
}

ray_t* ray_table_validate_rectangular(ray_t* tbl, const char* context) {
    const char* where = context ? context : "table";
    if (!tbl) return ray_error("type", "%s: expected table, got null", where);
    if (RAY_IS_ERR(tbl))
        return ray_error(ray_err_code(tbl) ? ray_err_code(tbl) : "error",
                         "%s: expected table, got error", where);
    if (tbl->type != RAY_TABLE)
        return ray_error("type", "%s: expected table, got %s", where, ray_type_name(tbl->type));

    ray_t* schema = tbl_schema(tbl);
    ray_t* cols = tbl_cols(tbl);
    if (!schema || !cols || RAY_IS_ERR(schema) || RAY_IS_ERR(cols))
        return ray_error("domain", "%s: malformed table storage", where);
    if (schema->len != cols->len)
        return ray_error("domain",
                         "%s: schema/column count mismatch (%lld names, %lld columns)",
                         where, (long long)schema->len, (long long)cols->len);

    ray_t** col_ptrs = (ray_t**)ray_data(cols);
    int64_t expected = 0;
    for (int64_t i = 0; i < cols->len; i++) {
        ray_t* col = col_ptrs[i];
        if (!table_col_is_valid(col))
            return ray_error("domain", "%s: column must be list/vector-like, got %s",
                             where, col ? ray_type_name(col->type) : "null");
        int64_t got = table_col_nrows(col);
        if (i == 0) {
            expected = got;
        } else if (got != expected) {
            return ray_error("domain",
                             "%s: ragged columns (column %lld has %lld rows, expected %lld)",
                             where, (long long)i, (long long)got, (long long)expected);
        }
    }

    return NULL;
}

/* --------------------------------------------------------------------------
 * Slot lookups (borrowed slot object, plain or RAY_ENCCOL), then the
 * accessors built on them: metadata (never decodes), plain column
 * (materializes an encoded slot), index, encoded object, and the legacy
 * getters.
 * -------------------------------------------------------------------------- */

static ray_t* tbl_slot_by_name(ray_t* tbl, int64_t name_id) {
    if (!tbl_is_table(tbl)) return NULL;
    ray_t* schema = tbl_schema(tbl);
    ray_t* cols   = tbl_cols(tbl);
    if (!schema || !cols) return NULL;
    int64_t* ids = (int64_t*)ray_data(schema);
    int64_t ncols = schema->len;
    ray_t** col_ptrs = (ray_t**)ray_data(cols);
    for (int64_t i = 0; i < ncols; i++)
        if (ids[i] == name_id) return col_ptrs[i];
    return NULL;
}

static ray_t* tbl_slot_by_idx(ray_t* tbl, int64_t idx) {
    if (!tbl_is_table(tbl)) return NULL;
    ray_t* cols = tbl_cols(tbl);
    if (!cols) return NULL;
    if (idx < 0 || idx >= cols->len) return NULL;
    return ((ray_t**)ray_data(cols))[idx];
}

bool ray_col_meta_of(ray_t* c, ray_col_meta_t* m) {
    memset(m, 0, sizeof *m);
    if (!c || RAY_IS_ERR(c)) return false;
    if (c->type == RAY_ENCCOL) { ray_enccol_meta(c, m); return true; }
    m->type = c->type;
    m->attrs = c->attrs;
    m->len = table_col_nrows(c);
    m->null_count = -1;
    if (RAY_IS_PARTED(c->type)) { m->n_segs = c->len; return true; }
    if (ray_is_vec(c) && ray_index_has(c)) {
        ray_index_t* ix = ray_index_payload(c->index);
        m->index_kind = ix->kind;
        m->unique = (ix->markers & RAY_MARK_UNIQUE) != 0;
    }
    return true;
}

bool ray_table_col_meta(ray_t* tbl, int64_t name_id, ray_col_meta_t* m) {
    return ray_col_meta_of(tbl_slot_by_name(tbl, name_id), m);
}
bool ray_table_col_meta_idx(ray_t* tbl, int64_t idx, ray_col_meta_t* m) {
    return ray_col_meta_of(tbl_slot_by_idx(tbl, idx), m);
}

static ray_t* slot_plain(ray_t* c) {
    if (!c) return NULL;
    return c->type == RAY_ENCCOL ? ray_enccol_materialize(c) : c;
}
/* The result is borrowed and read-only: encoded objects are shared between
 * table copies and share one decoded cache, so never write through it. */
ray_t* ray_table_col(ray_t* tbl, int64_t name_id)  { return slot_plain(tbl_slot_by_name(tbl, name_id)); }
ray_t* ray_table_col_idx(ray_t* tbl, int64_t idx)  { return slot_plain(tbl_slot_by_idx(tbl, idx)); }

static ray_t* slot_index(ray_t* c, ray_t** index) {
    *index = NULL;
    if (!c) return NULL;
    if (c->type != RAY_ENCCOL) {
        if (ray_is_vec(c) && ray_index_has(c)) *index = c->index;
        return NULL;
    }
    if (!ray_enccol_may_have_index(c)) return NULL;    /* parted answers false too */
    ray_t* d = ray_enccol_materialize(c);
    if (RAY_IS_ERR(d)) return d;
    if (ray_index_has(d)) *index = d->index;
    return NULL;
}
ray_t* ray_table_col_index(ray_t* tbl, int64_t name_id, ray_t** index) {
    return slot_index(tbl_slot_by_name(tbl, name_id), index);
}
ray_t* ray_table_col_index_idx(ray_t* tbl, int64_t idx, ray_t** index) {
    return slot_index(tbl_slot_by_idx(tbl, idx), index);
}

ray_t* ray_table_get_col_enc(ray_t* tbl, int64_t name_id) {
    ray_t* c = tbl_slot_by_name(tbl, name_id);
    return (c && c->type == RAY_ENCCOL) ? c : NULL;
}
ray_t* ray_table_get_col_enc_idx(ray_t* tbl, int64_t idx) {
    ray_t* c = tbl_slot_by_idx(tbl, idx);
    return (c && c->type == RAY_ENCCOL) ? c : NULL;
}

ray_t* ray_table_slot(ray_t* tbl, int64_t name_id) { return tbl_slot_by_name(tbl, name_id); }
ray_t* ray_table_slot_idx(ray_t* tbl, int64_t idx) { return tbl_slot_by_idx(tbl, idx); }

static ray_t* slot_parted_seg(ray_t* c, int64_t s) {
    if (!c) return NULL;
    if (c->type == RAY_ENCCOL) return ray_enccol_is_parted(c) ? ray_enccol_seg(c, s) : NULL;
    if (!RAY_IS_PARTED(c->type) || s < 0 || s >= c->len) return NULL;
    return ((ray_t**)ray_data(c))[s];                   /* parted wrapper, not a table */
}
ray_t* ray_table_parted_seg(ray_t* tbl, int64_t name_id, int64_t s) {
    return slot_parted_seg(tbl_slot_by_name(tbl, name_id), s);
}
ray_t* ray_table_parted_seg_idx(ray_t* tbl, int64_t idx, int64_t s) {
    return slot_parted_seg(tbl_slot_by_idx(tbl, idx), s);
}

/* aux[0] of a parted SYM/STR wrapper: its null presence once computed. */
#define PARTED_TEXT_NULLS_KNOWN 0x01
#define PARTED_TEXT_NULLS_HAS   0x02

bool ray_table_col_has_nulls(ray_t* tbl, int64_t name_id) {
    ray_t* c = tbl_slot_by_name(tbl, name_id);
    if (!c) return false;
    ray_col_meta_t m;
    ray_col_meta_of(c, &m);
    if (m.null_count >= 0) return m.null_count > 0;          /* generation 2: directory */
    if (m.enc) return (m.attrs & RAY_ATTR_HAS_NULLS) != 0;
    if (RAY_IS_PARTED(c->type)) {             /* the wrapper carries no flag: ask each segment */
        /* A text wrapper caches its answer in aux[0] (PARTED_TEXT_NULLS_*):
         * the segment scans run once per wrapper, not on every planning
         * call.  A wrapper is never edited in place (an insert builds a new
         * one, aux zeroed) and its loaded segments are read-only.  The byte
         * is written with the same value by any racing caller. */
        int8_t base = (int8_t)RAY_PARTED_BASETYPE(c->type);
        bool text_base = base == RAY_SYM || base == RAY_STR;
        if (text_base) {
            uint8_t k = __atomic_load_n(&c->aux[0], __ATOMIC_RELAXED);
            if (k & PARTED_TEXT_NULLS_KNOWN) return (k & PARTED_TEXT_NULLS_HAS) != 0;
        }
        bool any = false;
        ray_t** segs = (ray_t**)ray_data(c);
        for (int64_t i = 0; i < c->len; i++) {
            ray_col_meta_t sm;
            if (!ray_col_meta_of(segs[i], &sm)) continue;
            bool text = !sm.enc && (sm.type == RAY_SYM || sm.type == RAY_STR);
            bool nulls = sm.null_count >= 0 ? sm.null_count > 0   /* an encoded segment's directory */
                       /* a plain text segment: the exact check, which scans
                        * the payload (text nulls are in-band, no flag) */
                       : text ? ray_vec_has_nulls(segs[i])
                       : (sm.attrs & RAY_ATTR_HAS_NULLS) != 0;
            if (nulls) { any = true; break; }
        }
        if (text_base)
            __atomic_store_n(&c->aux[0], (uint8_t)(PARTED_TEXT_NULLS_KNOWN |
                             (any ? PARTED_TEXT_NULLS_HAS : 0)), __ATOMIC_RELAXED);
        return any;
    }
    return ray_is_vec(c) && ray_vec_has_nulls(c);            /* generation 0: today's scan */
}

/* Legacy getters (public, include/rayforce.h).  A plain slot (the common
 * case) is a pure lookup that leaves the error slot alone; only an encoded
 * slot saves the pending message, clears the slot and decodes.  Success
 * restores the pending message; a failure disposes of the error with
 * ray_error_free and returns NULL with its message left in the slot. */
static ray_t* legacy_slot(ray_t* slot) {
    if (!slot || slot->type != RAY_ENCCOL) return slot;
    char pending[256];
    snprintf(pending, sizeof pending, "%s", ray_error_msg() ? ray_error_msg() : "");
    ray_error_clear();
    ray_t* c = ray_enccol_materialize(slot);
    if (RAY_IS_ERR(c)) { ray_error_free(c); return NULL; }
    if (pending[0]) ray_error_free(ray_error("domain", "%s", pending));
    return c;
}
ray_t* ray_table_get_col(ray_t* tbl, int64_t name_id) {
    return legacy_slot(tbl_slot_by_name(tbl, name_id));
}
ray_t* ray_table_get_col_idx(ray_t* tbl, int64_t idx) {
    return legacy_slot(tbl_slot_by_idx(tbl, idx));
}

/* --------------------------------------------------------------------------
 * ray_table_set_col_idx — replace the column vector at slot `idx`.  COWs the
 * cols list, releases the previous column and retains `col_vec`.  No-op when
 * idx is out of range.  Used to override a column an aggregate emitter built
 * with the wrong representation (e.g. wide STR/GUID min/max post-fill).
 * -------------------------------------------------------------------------- */

void ray_table_set_col_idx(ray_t* tbl, int64_t idx, ray_t* col_vec) {
    if (!tbl_is_table(tbl) || !col_vec) return;
    ray_t** slots = tbl_slots(tbl);
    ray_t* cols = slots[1];
    if (!cols || RAY_IS_ERR(cols)) return;
    if (idx < 0 || idx >= cols->len) return;
    cols = ray_cow(cols);
    if (!cols || RAY_IS_ERR(cols)) return;
    slots[1] = cols;
    ray_t** cv = (ray_t**)ray_data(cols);
    ray_retain(col_vec);
    ray_release(cv[idx]);
    cv[idx] = col_vec;
}

/* --------------------------------------------------------------------------
 * ray_table_col_name — sym id at slot `idx`, -1 on out-of-range.
 * -------------------------------------------------------------------------- */

int64_t ray_table_col_name(ray_t* tbl, int64_t idx) {
    if (!tbl_is_table(tbl)) return -1;
    ray_t* schema = tbl_schema(tbl);
    if (!schema) return -1;
    if (idx < 0 || idx >= schema->len) return -1;
    return ((int64_t*)ray_data(schema))[idx];
}

/* --------------------------------------------------------------------------
 * ray_table_set_col_name — overwrite name at `idx`.  Caller must ensure
 * exclusive ownership (rc==1) before calling.
 * -------------------------------------------------------------------------- */

void ray_table_set_col_name(ray_t* tbl, int64_t idx, int64_t name_id) {
    if (!tbl_is_table(tbl)) return;
    ray_t** slots = tbl_slots(tbl);
    ray_t* schema = slots[0];
    if (!schema || RAY_IS_ERR(schema)) return;
    if (idx < 0 || idx >= schema->len) return;
    schema = ray_cow(schema);
    if (!schema || RAY_IS_ERR(schema)) return;
    slots[0] = schema;
    ((int64_t*)ray_data(schema))[idx] = name_id;
}

/* --------------------------------------------------------------------------
 * ray_table_ncols / ray_table_nrows / ray_table_schema
 * -------------------------------------------------------------------------- */

int64_t ray_table_ncols(ray_t* tbl) {
    if (!tbl_is_table(tbl)) return 0;
    ray_t* schema = tbl_schema(tbl);
    return schema ? schema->len : 0;
}

int64_t ray_table_nrows(ray_t* tbl) {
    if (!tbl_is_table(tbl)) return 0;
    ray_t* cols = tbl_cols(tbl);
    if (!cols || cols->len <= 0) return 0;
    ray_t* first_col = ((ray_t**)ray_data(cols))[0];
    if (!first_col || RAY_IS_ERR(first_col)) return 0;

    return table_col_nrows(first_col);
}

int64_t ray_parted_nrows(ray_t* v) {
    if (!v || RAY_IS_ERR(v)) return 0;
    if (!RAY_IS_PARTED(v->type) && v->type != RAY_MAPCOMMON) return v->len;

    if (v->type == RAY_MAPCOMMON) {
        ray_t** ptrs = (ray_t**)ray_data(v);
        ray_t* counts = ptrs[1];
        if (!counts || RAY_IS_ERR(counts)) return 0;
        int64_t total = 0;
        int64_t* cdata = (int64_t*)ray_data(counts);
        for (int64_t i = 0; i < counts->len; i++)
            total += cdata[i];
        return total;
    }

    int64_t n_segs = v->len;
    ray_t** segs = (ray_t**)ray_data(v);
    int64_t total = 0;
    for (int64_t i = 0; i < n_segs; i++) {
        if (segs[i] && !RAY_IS_ERR(segs[i]))
            total += segs[i]->len;
    }
    return total;
}

ray_t* ray_table_schema(ray_t* tbl) {
    if (!tbl_is_table(tbl)) return NULL;
    return tbl_schema(tbl);
}

bool ray_table_cols_canonical(ray_t* tbl) {
    if (!tbl_is_table(tbl)) return false;
    ray_t* schema = tbl_schema(tbl);
    ray_t* cols = tbl_cols(tbl);
    return schema && !RAY_IS_ERR(schema) && cols && !RAY_IS_ERR(cols) &&
           cols->type == RAY_LIST && cols->attrs == 0 && cols->len == schema->len;
}

ray_t* ray_table_cols_inplace(ray_t* tbl) {
    if (!tbl_is_table(tbl)) return NULL;
    ray_t* cols = tbl_cols(tbl);
    if (!cols || RAY_IS_ERR(cols)) return NULL;
    ray_t** cv = (ray_t**)ray_data(cols);
    for (int64_t i = 0; i < cols->len; i++)
        if (cv[i] && cv[i]->type == RAY_ENCCOL) return NULL;
    return cols;
}

bool ray_table_has_col(ray_t* tbl, int64_t name_id) {
    ray_col_meta_t m;
    return tbl && !RAY_IS_ERR(tbl) && tbl->type == RAY_TABLE && ray_table_col_meta(tbl, name_id, &m);
}