#include <inttypes.h>
#include <stdatomic.h>
#include "ops/internal.h"
#include "ops/hash.h"
#include "ops/idxop.h"
#include "ops/rowsel.h"
#include "table/domain.h"
#include "core/pool.h"
#include "lang/format.h"
static inline ray_t* sym_scalar_str(ray_t* v) {
return ray_is_atom(v) ? ray_sym_str(v->i64) : ray_sym_vec_cell(v, 0);
}
static inline int64_t sym_id_runtime(ray_t* col, int64_t id) {
struct ray_sym_domain_s* dom = ray_sym_vec_domain(col);
if (dom == ray_sym_runtime_domain()) return id;
const int64_t* lut = ray_sym_domain_runtime_lut(dom);
if (!lut || id < 0 || id >= ray_sym_domain_count(dom)) return -1;
return lut[id];
}
static inline int64_t sym_cell_runtime_id(ray_t* v, int64_t i) {
return sym_id_runtime(v, ray_read_sym(ray_data(v), i, v->type, v->attrs));
}
static inline int64_t sym_scalar_runtime_id(ray_t* v) {
return ray_is_atom(v) ? v->i64 : sym_cell_runtime_id(v, 0);
}
static inline ray_t* str_vec_pool_obj(ray_t* v) {
if (!v || v->type != RAY_STR) return NULL;
ray_t* owner = (v->attrs & RAY_ATTR_SLICE) ? v->slice_parent : v;
return owner ? owner->str_pool : NULL;
}
enum { IF_BRANCH_MAX_SCAN_SYMS = 64 };
typedef struct {
int64_t syms[IF_BRANCH_MAX_SCAN_SYMS];
int n_syms;
} if_branch_plan_t;
static bool if_plan_add_sym(if_branch_plan_t* p, int64_t sym) {
for (int i = 0; i < p->n_syms; i++)
if (p->syms[i] == sym) return true;
if (p->n_syms >= IF_BRANCH_MAX_SCAN_SYMS) return false;
p->syms[p->n_syms++] = sym;
return true;
}
static bool if_op_rowwise(uint16_t opc) {
switch (opc) {
case OP_SCAN: case OP_CONST:
case OP_NEG: case OP_ABS: case OP_NOT: case OP_SQRT:
case OP_LOG: case OP_EXP: case OP_CEIL: case OP_FLOOR: case OP_ROUND:
case OP_SIN: case OP_ASIN: case OP_COS: case OP_ACOS:
case OP_TAN: case OP_ATAN: case OP_RECIPROCAL: case OP_SIGNUM:
case OP_ISNULL: case OP_CAST:
case OP_ADD: case OP_SUB: case OP_MUL: case OP_DIV: case OP_IDIV: case OP_MOD:
case OP_POW:
case OP_EQ: case OP_NE: case OP_LT: case OP_LE:
case OP_GT: case OP_GE: case OP_AND: case OP_OR:
case OP_MIN2: case OP_MAX2: case OP_IF:
case OP_LIKE: case OP_ILIKE: case OP_UPPER: case OP_LOWER:
case OP_STRLEN: case OP_SUBSTR: case OP_REPLACE: case OP_TRIM:
case OP_CONCAT: case OP_STR_FIND:
case OP_EXTRACT: case OP_DATE_TRUNC:
return true;
default:
return false;
}
}
static bool if_collect_branch(ray_graph_t* g, ray_op_t* op,
if_branch_plan_t* plan, uint64_t* visited) {
if (!g || !op) return false;
uint32_t nid = op->id;
if (nid < g->node_count) {
uint64_t bit = (uint64_t)1 << (nid & 63);
uint64_t* word = &visited[nid >> 6];
if (*word & bit) return true;
*word |= bit;
}
uint16_t opc = op->opcode;
if (opc == OP_CONST) {
ray_op_ext_t* ext = find_ext(g, op->id);
return ext && ext->literal && ray_is_atom(ext->literal);
}
if (opc == OP_SCAN) {
ray_op_ext_t* ext = find_ext(g, op->id);
if (!ext) return false;
uint16_t stored_table_id = 0;
memcpy(&stored_table_id, ext->base.pad, sizeof(uint16_t));
if (stored_table_id != 0) return false;
return if_plan_add_sym(plan, ext->sym);
}
if (!if_op_rowwise(opc)) return false;
for (uint8_t i = 0; i < op->arity && i < 2; i++) {
if (op->in_id[i] == RAY_OP_NONE) continue;
if (!if_collect_branch(g, op_child(g, op, i), plan, visited))
return false;
}
ray_op_ext_t* ext = find_ext(g, op->id);
if ((opc == OP_IF || opc == OP_SUBSTR || opc == OP_REPLACE) && ext) {
if (ext->third_in != RAY_OP_NONE &&
!if_collect_branch(g, op_node(g, ext->third_in), plan, visited))
return false;
} else if (opc == OP_CONCAT && ext) {
int n_args = (int)ext->sym;
uint32_t* trail = (uint32_t*)((char*)(ext + 1));
for (int i = 2; i < n_args; i++) {
if (!if_collect_branch(g, op_node(g, trail[i - 2]), plan, visited))
return false;
}
}
return true;
}
static bool if_make_branch_plan(ray_graph_t* g, ray_op_t* root,
if_branch_plan_t* plan) {
memset(plan, 0, sizeof(*plan));
if (!g || !root || g->node_count == 0) return false;
ray_t* hdr = NULL;
size_t words = ((size_t)g->node_count + 63u) >> 6;
uint64_t* visited = (uint64_t*)scratch_calloc(&hdr, words * sizeof(uint64_t));
if (!visited) return false;
bool ok = if_collect_branch(g, root, plan, visited);
scratch_free(hdr);
return ok;
}
static int64_t if_selected_true_count(const uint8_t* cond, int64_t nrows,
ray_t* outer_sel) {
int64_t total = 0;
if (!outer_sel) {
for (int64_t i = 0; i < nrows; i++) total += cond[i] ? 1 : 0;
return total;
}
ray_rowsel_t* m = ray_rowsel_meta(outer_sel);
const uint8_t* flags = ray_rowsel_flags(outer_sel);
const uint32_t* offsets = ray_rowsel_offsets(outer_sel);
const uint16_t* idx = ray_rowsel_idx(outer_sel);
for (uint32_t seg = 0; seg < m->n_segs; seg++) {
uint8_t f = flags[seg];
if (f == RAY_SEL_NONE) continue;
int64_t base = (int64_t)seg * RAY_MORSEL_ELEMS;
int64_t end = base + RAY_MORSEL_ELEMS;
if (end > nrows) end = nrows;
if (f == RAY_SEL_ALL) {
for (int64_t r = base; r < end; r++) total += cond[r] ? 1 : 0;
} else {
uint32_t off = offsets[seg];
uint32_t cnt = offsets[seg + 1] - off;
for (uint32_t i = 0; i < cnt; i++) {
int64_t r = base + idx[off + i];
total += cond[r] ? 1 : 0;
}
}
}
return total;
}
static void if_fill_branch_ids(const uint8_t* cond, int64_t nrows,
ray_t* outer_sel, int64_t* true_ids,
int64_t* false_ids) {
int64_t ti = 0, fi = 0;
#define IF_ADD_ROW(R) do { int64_t _r = (R); if (cond[_r]) true_ids[ti++] = _r; else false_ids[fi++] = _r; } while (0)
if (!outer_sel) {
for (int64_t r = 0; r < nrows; r++) IF_ADD_ROW(r);
} else {
ray_rowsel_t* m = ray_rowsel_meta(outer_sel);
const uint8_t* flags = ray_rowsel_flags(outer_sel);
const uint32_t* offsets = ray_rowsel_offsets(outer_sel);
const uint16_t* idx = ray_rowsel_idx(outer_sel);
for (uint32_t seg = 0; seg < m->n_segs; seg++) {
uint8_t f = flags[seg];
if (f == RAY_SEL_NONE) continue;
int64_t base = (int64_t)seg * RAY_MORSEL_ELEMS;
int64_t end = base + RAY_MORSEL_ELEMS;
if (end > nrows) end = nrows;
if (f == RAY_SEL_ALL) {
for (int64_t r = base; r < end; r++) IF_ADD_ROW(r);
} else {
uint32_t off = offsets[seg];
uint32_t cnt = offsets[seg + 1] - off;
for (uint32_t i = 0; i < cnt; i++) IF_ADD_ROW(base + idx[off + i]);
}
}
}
#undef IF_ADD_ROW
}
static ray_t* if_eval_branch(ray_graph_t* g, ray_op_t* branch,
const if_branch_plan_t* plan,
int64_t* ids, int64_t count, int64_t nrows) {
if (count <= 0) return NULL;
ray_t* saved_table = g->table;
ray_t* saved_sel = g->selection;
ray_t* sub = NULL;
ray_t* sel = NULL;
if (plan->n_syms > 0 && count != nrows) {
sel = ray_index_rowsel_from_ids(nrows, ids, count);
if (!sel) return ray_error("oom", NULL);
sub = sel_compact(g, saved_table, sel, plan->syms, plan->n_syms);
ray_release(sel);
if (!sub || RAY_IS_ERR(sub)) return sub ? sub : ray_error("oom", NULL);
} else {
sub = saved_table;
ray_retain(sub);
}
g->table = sub;
g->selection = NULL;
ray_exec_memo_save_t memo_save;
ray_exec_memo_push(g, branch, &memo_save);
ray_t* value = exec_node(g, branch);
ray_exec_memo_pop(g, &memo_save);
if (g->selection) {
ray_release(g->selection);
g->selection = NULL;
}
g->table = saved_table;
g->selection = saved_sel;
ray_release(sub);
return value;
}
static inline bool if_value_scalar(ray_t* v) {
return ray_is_atom(v) || (v->type > 0 && v->len == 1);
}
static int64_t if_value_index(ray_t* v, int64_t* ids, int64_t j,
int64_t count, int64_t nrows) {
if (if_value_scalar(v)) return 0;
if (v->len == count) return j;
if (v->len == nrows) return ids[j];
return -1;
}
static inline int64_t if_f64_to_i64(double x) {
if (!(x >= -9223372036854775808.0 && x < 9223372036854775808.0))
return NULL_I64;
return (int64_t)x;
}
static int64_t if_atom_i64(ray_t* v) {
switch (-v->type) {
case RAY_I64: case RAY_TIMESTAMP: return v->i64;
case RAY_I32: case RAY_DATE: case RAY_TIME: return v->i32;
case RAY_I16: return v->i16;
case RAY_BOOL: case RAY_U8: return v->b8;
case RAY_F64: case RAY_F32: return if_f64_to_i64(v->f64);
default: return 0;
}
}
static double if_atom_f64(ray_t* v) {
if (RAY_ATOM_IS_NULL(v)) return (double)NAN;
return (v->type == -RAY_F64 || v->type == -RAY_F32)
? v->f64 : (double)if_atom_i64(v);
}
static bool if_val_has_null(ray_t* v) {
if (!v) return false;
if (!ray_is_atom(v)) return (v->attrs & RAY_ATTR_HAS_NULLS) != 0;
return RAY_ATOM_IS_NULL(v);
}
static int64_t if_vec_i64(ray_t* v, int64_t idx) {
if (v->type == RAY_F64) return if_f64_to_i64(((double*)ray_data(v))[idx]);
if (v->type == RAY_F32) return if_f64_to_i64((double)((float*)ray_data(v))[idx]);
return read_col_i64(ray_data(v), idx, v->type, v->attrs);
}
static double if_vec_f64(ray_t* v, int64_t idx) {
if (v->type == RAY_F64) return ((double*)ray_data(v))[idx];
if (v->type == RAY_F32) return (double)((float*)ray_data(v))[idx];
return (double)if_vec_i64(v, idx);
}
static int64_t if_scalar_i64(ray_t* v) {
return ray_is_atom(v) ? if_atom_i64(v) : if_vec_i64(v, 0);
}
static double if_scalar_f64(ray_t* v) {
return ray_is_atom(v) ? if_atom_f64(v) : if_vec_f64(v, 0);
}
static int8_t if_value_type(ray_t* v) {
if (!v) return 0;
if (ray_is_atom(v) && v->type == -RAY_STR) return RAY_SYM;
int8_t t = ray_is_atom(v) ? (int8_t)(-(int)v->type) : v->type;
return (t == RAY_F32) ? RAY_F64 : t;
}
static int if_type_rank(int8_t t) {
switch (t) {
case RAY_BOOL: return 1;
case RAY_U8: return 2;
case RAY_I16: return 3;
case RAY_I32: return 4;
case RAY_I64: return 5;
case RAY_F64: return 6;
default: return 0;
}
}
static int8_t if_selected_type(ray_t* then_v, ray_t* else_v, int8_t declared,
bool eager_possible) {
int8_t tt = if_value_type(then_v);
int8_t et = if_value_type(else_v);
if (!eager_possible && (!then_v || !else_v)) {
int8_t live = then_v ? tt : et;
int lr = if_type_rank(live);
bool float_branch_absent = declared == RAY_F64 &&
lr && lr < if_type_rank(RAY_F64);
bool string_branch_absent = (declared == RAY_STR || declared == RAY_SYM) &&
lr > 0;
if (float_branch_absent || string_branch_absent) return live;
}
int dr = if_type_rank(declared);
if (dr) {
int8_t widest = declared;
int wr = dr;
if (if_type_rank(tt) > wr) { widest = tt; wr = if_type_rank(tt); }
if (if_type_rank(et) > wr) { widest = et; }
return widest;
}
return declared;
}
static bool if_scatter_numeric(ray_t* result, ray_t* value, int64_t* ids,
int64_t count, int64_t nrows, int8_t out_type) {
if (!value) return true;
for (int64_t j = 0; j < count; j++) {
int64_t src = if_value_index(value, ids, j, count, nrows);
if (src < 0) return false;
int64_t dst = ids[j];
if (out_type == RAY_F64) {
double x = ray_is_atom(value) ? if_atom_f64(value)
: if_vec_f64(value, src);
((double*)ray_data(result))[dst] = x;
} else if (out_type == RAY_I64 || out_type == RAY_TIMESTAMP) {
int64_t x = ray_is_atom(value) ? if_atom_i64(value) : if_vec_i64(value, src);
((int64_t*)ray_data(result))[dst] = x;
} else if (out_type == RAY_I32 || out_type == RAY_DATE || out_type == RAY_TIME) {
int64_t x = ray_is_atom(value) ? if_atom_i64(value) : if_vec_i64(value, src);
((int32_t*)ray_data(result))[dst] = (int32_t)x;
} else if (out_type == RAY_I16) {
int64_t x = ray_is_atom(value) ? if_atom_i64(value) : if_vec_i64(value, src);
((int16_t*)ray_data(result))[dst] = (int16_t)x;
} else {
int64_t x = ray_is_atom(value) ? if_atom_i64(value) : if_vec_i64(value, src);
((uint8_t*)ray_data(result))[dst] = (uint8_t)x;
}
}
return true;
}
static bool if_string_cell(ray_t* value, int64_t src,
const char** sp, size_t* sl) {
if (value->type == -RAY_STR) {
*sp = ray_str_ptr(value); *sl = ray_str_len(value); return true;
}
if (value->type == RAY_STR) {
*sp = ray_str_vec_get(value, src, sl);
if (!*sp) { *sp = ""; *sl = 0; }
return true;
}
if (value->type == -RAY_SYM || RAY_IS_SYM(value->type)) {
ray_t* s = ray_is_atom(value) ? sym_scalar_str(value)
: ray_sym_vec_cell(value, src);
*sp = s ? ray_str_ptr(s) : "";
*sl = s ? ray_str_len(s) : 0;
return true;
}
*sp = ""; *sl = 0;
return true;
}
static ray_t* if_scatter_str(ray_t* result, ray_t* value, int64_t* ids,
int64_t count, int64_t nrows) {
if (!value) return result;
for (int64_t j = 0; j < count; j++) {
int64_t src = if_value_index(value, ids, j, count, nrows);
if (src < 0) { ray_release(result); return ray_error("length", "if: branch length mismatch"); }
const char* sp = NULL;
size_t sl = 0;
if_string_cell(value, src, &sp, &sl);
ray_t* next = ray_str_vec_set(result, ids[j], sp ? sp : "", sp ? sl : 0);
if (!next || RAY_IS_ERR(next)) {
ray_release(result);
return next ? next : ray_error("oom", NULL);
}
result = next;
}
return result;
}
typedef struct {
ray_t* v;
int64_t* ids;
int64_t n;
bool scalar;
bool full;
const ray_str_t* desc;
const char* bytes;
ray_t* pool;
const char* sp;
size_t sl;
} if_str_side_t;
static bool if_str_side_init(ray_t* v, int64_t* ids, int64_t n, int64_t nrows,
if_str_side_t* s) {
memset(s, 0, sizeof(*s));
s->ids = ids; s->n = n;
if (!v || n <= 0) return true;
s->v = v;
if (v->type == -RAY_STR) {
s->scalar = true; s->sp = ray_str_ptr(v); s->sl = ray_str_len(v);
return true;
}
if (v->type != RAY_STR) return false;
if (v->len == 1) {
s->scalar = true;
s->sp = ray_str_vec_get(v, 0, &s->sl);
if (!s->sp) { s->sp = ""; s->sl = 0; }
return true;
}
if (v->len == n) s->full = false;
else if (v->len == nrows) s->full = true;
else return false;
str_resolve(v, &s->desc, &s->bytes);
s->pool = str_vec_pool_obj(v);
if (s->pool && RAY_IS_ERR(s->pool)) return false;
return true;
}
static inline ray_str_t if_str_side_desc(const if_str_side_t* s, int64_t j) {
return s->full ? s->desc[s->ids[j]] : s->desc[j];
}
static uint64_t if_str_side_pooled_bytes(const if_str_side_t* s) {
if (!s->v || s->scalar) return 0;
uint64_t sum = 0;
for (int64_t j = 0; j < s->n; j++) {
ray_str_t d = if_str_side_desc(s, j);
if (!ray_str_is_inline(&d)) sum += d.len;
}
return sum;
}
static void if_str_side_offsets(const if_str_side_t* s, uint32_t* newoff, uint64_t* run) {
for (int64_t j = 0; j < s->n; j++) {
ray_str_t d = if_str_side_desc(s, j);
if (ray_str_is_inline(&d)) continue;
newoff[j] = (uint32_t)*run;
*run += d.len;
}
}
typedef struct {
const int64_t* ids;
const ray_str_t* src;
bool scalar;
bool full;
ray_str_t sd;
ray_str_t* dst;
const uint32_t* newoff;
const char* src_bytes;
char* dst_bytes;
} if_str_scatter_ctx_t;
static void if_str_scatter_fn(void* vctx, uint32_t wid, int64_t start, int64_t end) {
(void)wid;
const if_str_scatter_ctx_t* c = (const if_str_scatter_ctx_t*)vctx;
if (c->scalar) {
for (int64_t j = start; j < end; j++) c->dst[c->ids[j]] = c->sd;
return;
}
for (int64_t j = start; j < end; j++) {
ray_str_t d = c->full ? c->src[c->ids[j]] : c->src[j];
if (c->newoff && !ray_str_is_inline(&d)) {
memcpy(c->dst_bytes + c->newoff[j], c->src_bytes + d.pool_off, d.len);
d.pool_off = c->newoff[j];
}
c->dst[c->ids[j]] = d;
}
}
static void if_str_scatter_side(const if_str_side_t* s, const uint32_t* newoff, char* dst_bytes,
uint32_t scalar_off, ray_str_t* dst) {
if (!s->v) return;
if_str_scatter_ctx_t c = {
.ids = s->ids, .src = s->desc, .scalar = s->scalar, .full = s->full,
.dst = dst, .newoff = newoff, .src_bytes = s->bytes, .dst_bytes = dst_bytes,
};
if (s->scalar) {
memset(&c.sd, 0, sizeof(c.sd));
c.sd.len = (uint32_t)s->sl;
if (s->sl <= RAY_STR_INLINE_MAX) {
if (s->sl) memcpy(c.sd.data, s->sp, s->sl);
} else {
memcpy(c.sd.prefix, s->sp, 4);
c.sd.pool_off = scalar_off;
}
}
ray_pool_t* pool = ray_pool_get();
if (ray_pool_par_dispatch_ok(pool, s->n, RAY_PARALLEL_THRESHOLD))
ray_pool_dispatch(pool, if_str_scatter_fn, &c, s->n);
else
if_str_scatter_fn(&c, 0, 0, s->n);
}
static ray_t* if_scatter_str_desc(ray_t* result,
ray_t* then_v, int64_t* t_ids, int64_t t_n,
ray_t* else_v, int64_t* e_ids, int64_t e_n,
int64_t nrows) {
if_str_side_t t, e;
if (!if_str_side_init(then_v, t_ids, t_n, nrows, &t)) return NULL;
if (!if_str_side_init(else_v, e_ids, e_n, nrows, &e)) return NULL;
ray_t* pt = (t.v && !t.scalar) ? t.pool : NULL;
ray_t* pe = (e.v && !e.scalar) ? e.pool : NULL;
bool t_big = t.v && t.scalar && t.sl > RAY_STR_INLINE_MAX;
bool e_big = e.v && e.scalar && e.sl > RAY_STR_INLINE_MAX;
uint64_t t_ref = if_str_side_pooled_bytes(&t);
uint64_t e_ref = if_str_side_pooled_bytes(&e);
ray_str_t* dst = (ray_str_t*)ray_data(result);
bool one_pool = (pt == pe) || !pt || !pe;
ray_t* shared = pt ? pt : pe;
if (!t_big && !e_big && one_pool &&
(!shared || !ray_str_view_should_compact(t_ref + e_ref, shared->len))) {
if (shared) { ray_retain(shared); result->str_pool = shared; }
if_str_scatter_side(&t, NULL, NULL, 0, dst);
if_str_scatter_side(&e, NULL, NULL, 0, dst);
return result;
}
uint64_t total = t_ref + e_ref + (t_big ? (uint64_t)t.sl : 0) + (e_big ? (uint64_t)e.sl : 0);
if (total > UINT32_MAX) return NULL;
int64_t n_off = (t.v && !t.scalar ? t.n : 0) + (e.v && !e.scalar ? e.n : 0);
ray_t* off_hdr = NULL;
uint32_t* newoff = (uint32_t*)scratch_alloc(&off_hdr, (size_t)(n_off > 0 ? n_off : 1) * sizeof(uint32_t));
if (!newoff) return NULL;
uint32_t* t_off = (t.v && !t.scalar) ? newoff : NULL;
uint32_t* e_off = (e.v && !e.scalar) ? newoff + (t.v && !t.scalar ? t.n : 0) : NULL;
uint64_t run = 0;
if (t_off) if_str_side_offsets(&t, t_off, &run);
if (e_off) if_str_side_offsets(&e, e_off, &run);
uint32_t ts_off = 0, es_off = 0;
ray_t* np = ray_alloc(total > 0 ? (size_t)total : 1);
if (!np || RAY_IS_ERR(np)) { scratch_free(off_hdr); return NULL; }
np->type = RAY_U8;
np->len = (int64_t)total;
char* dst_bytes = (char*)ray_data(np);
if (t_big) { memcpy(dst_bytes + run, t.sp, t.sl); ts_off = (uint32_t)run; run += t.sl; }
if (e_big) { memcpy(dst_bytes + run, e.sp, e.sl); es_off = (uint32_t)run; run += e.sl; }
result->str_pool = np;
if_str_scatter_side(&t, t_off, dst_bytes, ts_off, dst);
if_str_scatter_side(&e, e_off, dst_bytes, es_off, dst);
scratch_free(off_hdr);
return result;
}
typedef struct {
ray_t* v;
struct ray_sym_domain_s* dom;
int64_t dn;
int32_t* lut;
ray_t* lut_hdr;
ray_sym_domain_raw_t raw;
bool raw_ok;
} if_sym_side_t;
static bool if_sym_side_init(if_sym_side_t* side, ray_t* v) {
memset(side, 0, sizeof(*side));
side->v = v;
if (!v || ray_is_atom(v) || !RAY_IS_SYM(v->type)) return true;
struct ray_sym_domain_s* dom = ray_sym_vec_domain(v);
if (!dom || dom == ray_sym_runtime_domain()) return true;
int64_t dn = ray_sym_domain_count(dom);
if (dn <= 0) return true;
side->lut = (int32_t*)scratch_alloc(&side->lut_hdr, (size_t)dn * sizeof(int32_t));
if (!side->lut) return false;
memset(side->lut, 0xff, (size_t)dn * sizeof(int32_t));
side->dom = dom;
side->dn = dn;
side->raw_ok = ray_sym_domain_raw_pin(dom, &side->raw);
return true;
}
static void if_sym_side_free(if_sym_side_t* side) {
if (side->raw_ok) ray_sym_domain_raw_unpin(side->dom);
scratch_free(side->lut_hdr);
side->lut = NULL;
side->lut_hdr = NULL;
}
static int64_t if_sym_side_resolve(if_sym_side_t* side, int64_t pos) {
if (pos < 0 || pos >= side->dn) return -1;
int64_t id = side->lut[pos];
if (id >= 0) return id;
const char* sp = NULL;
size_t sl = 0;
if (side->raw_ok && pos < side->raw.count) {
sp = ray_sym_domain_raw_str(&side->raw, pos, &sl);
} else {
ray_t* s = ray_sym_domain_str(side->dom, pos);
if (s) { sp = ray_str_ptr(s); sl = ray_str_len(s); }
}
id = ray_sym_intern(sp ? sp : "", sp ? sl : 0);
if (id >= 0 && id <= INT32_MAX) side->lut[pos] = (int32_t)id;
return id;
}
static inline int64_t if_sym_side_lookup(const if_sym_side_t* side, int64_t pos) {
if (pos < 0 || pos >= side->dn) return 0;
int32_t id = side->lut[pos];
return id >= 0 ? (int64_t)id : 0;
}
static void if_sym_side_prepare(if_sym_side_t* side, const uint8_t* cond,
int64_t len, uint8_t want) {
if (!side->dom) return;
const void* base = ray_data(side->v);
for (int64_t i = 0; i < len; i++) {
if ((cond[i] != 0) != (want != 0)) continue;
int64_t pos = ray_read_sym(base, i, side->v->type, side->v->attrs);
if (pos >= 0 && pos < side->dn && side->lut[pos] < 0)
(void)if_sym_side_resolve(side, pos);
}
}
static int64_t if_sym_cell_value(if_sym_side_t* side, int64_t src) {
ray_t* value = side->v;
if (value->type == -RAY_STR) return ray_sym_intern(ray_str_ptr(value), ray_str_len(value));
if (value->type == RAY_STR) {
size_t sl = 0;
const char* sp = ray_str_vec_get(value, src, &sl);
return ray_sym_intern(sp ? sp : "", sp ? sl : 0);
}
if (ray_is_atom(value)) return sym_scalar_runtime_id(value);
if (!side->dom) return sym_cell_runtime_id(value, src);
int64_t pos = ray_read_sym(ray_data(value), src, value->type, value->attrs);
if (pos >= 0 && pos < side->dn && side->lut[pos] >= 0) return side->lut[pos];
return if_sym_side_resolve(side, pos);
}
static bool if_scatter_sym(ray_t* result, ray_t* value, int64_t* ids,
int64_t count, int64_t nrows) {
if (!value) return true;
if_sym_side_t side;
if (!if_sym_side_init(&side, value)) return false;
int64_t* dst = (int64_t*)ray_data(result);
bool ok = true;
for (int64_t j = 0; j < count && ok; j++) {
int64_t src = if_value_index(value, ids, j, count, nrows);
if (src < 0) { ok = false; break; }
dst[ids[j]] = if_sym_cell_value(&side, src);
}
if_sym_side_free(&side);
return ok;
}
static bool if_lazy_supported_type(int8_t out_type) {
return out_type == RAY_F64 || out_type == RAY_I64 ||
out_type == RAY_I32 || out_type == RAY_I16 ||
out_type == RAY_BOOL || out_type == RAY_U8 ||
out_type == RAY_TIMESTAMP || out_type == RAY_TIME ||
out_type == RAY_DATE || out_type == RAY_STR ||
out_type == RAY_SYM;
}
static bool if_branch_trivial(ray_graph_t* g, ray_op_t* op) {
if (!op) return false;
if (op->opcode == OP_ALIAS || op->opcode == OP_MATERIALIZE)
return if_branch_trivial(g, op_child(g, op, 0));
if (op->opcode == OP_SCAN) return true;
if (op->opcode == OP_CONST) {
ray_op_ext_t* ext = find_ext(g, op->id);
return ext && ext->literal && ray_is_atom(ext->literal);
}
return false;
}
static bool if_branch_cheap(ray_graph_t* g, ray_op_t* op, int depth) {
if (!op || depth > 32) return false;
switch (op->opcode) {
case OP_ALIAS: case OP_MATERIALIZE:
return if_branch_cheap(g, op_child(g, op, 0), depth + 1);
case OP_SCAN: return true;
case OP_CONST: {
ray_op_ext_t* ext = find_ext(g, op->id);
return ext && ext->literal && ray_is_atom(ext->literal);
}
case OP_SUBSTR: case OP_IF: {
ray_op_ext_t* ext = find_ext(g, op->id);
if (!ext || ext->third_in >= g->node_count) return false;
for (int k = 0; k < op->arity && k < 2; k++)
if (!if_branch_cheap(g, op_child(g, op, k), depth + 1)) return false;
return if_branch_cheap(g, op_node(g, ext->third_in), depth + 1);
}
case OP_STRLEN: case OP_UPPER: case OP_LOWER: case OP_TRIM:
case OP_ADD: case OP_SUB: case OP_MUL: case OP_NEG: case OP_ABS:
case OP_EQ: case OP_NE: case OP_LT: case OP_LE: case OP_GT: case OP_GE:
case OP_AND: case OP_OR: case OP_NOT: case OP_ISNULL:
for (int k = 0; k < op->arity && k < 2; k++)
if (!if_branch_cheap(g, op_child(g, op, k), depth + 1)) return false;
return true;
default:
return false;
}
}
static bool if_type_eager_ok(int8_t bt, int8_t out) {
int8_t b = bt < 0 ? (int8_t)-bt : bt;
if (out == RAY_SYM)
return b == RAY_SYM || bt == -RAY_STR;
switch (b) {
case RAY_F64: case RAY_I64: case RAY_I32: case RAY_I16:
case RAY_U8: case RAY_BOOL:
case RAY_TIMESTAMP: case RAY_TIME: case RAY_DATE: case RAY_GUID:
return true;
default: return false;
}
}
static ray_t* exec_if_selected(ray_graph_t* g, ray_op_t* op, ray_t* cond_v) {
if (!g || !g->table || !cond_v || cond_v->type != RAY_BOOL)
return NULL;
int64_t nrows = ray_table_nrows(g->table);
if (cond_v->len != nrows) return NULL;
if (!if_lazy_supported_type(op->out_type)) return NULL;
ray_op_t* then_op = op_child(g, op, 1);
ray_op_ext_t* ext = find_ext(g, op->id);
ray_op_t* else_op = ext ? op_node(g, ext->third_in) : NULL;
if (!then_op || !else_op) return NULL;
ray_t* outer_sel = g->selection;
if (outer_sel) {
ray_rowsel_t* sm = ray_rowsel_meta(outer_sel);
if (!sm || sm->nrows != nrows) return NULL;
}
int64_t selected = outer_sel ? ray_rowsel_meta(outer_sel)->total_pass : nrows;
if (selected < 0 || selected > nrows) return NULL;
bool sparse_sel = outer_sel && selected * 4 <= nrows;
bool near_full_sel = !outer_sel || selected * 10 >= nrows * 9;
bool eager_possible = (op->out_type != RAY_STR &&
if_branch_trivial(g, then_op) &&
if_branch_trivial(g, else_op) &&
if_type_eager_ok(then_op->out_type, op->out_type) &&
if_type_eager_ok(else_op->out_type, op->out_type)) ||
(op->out_type == RAY_STR && !sparse_sel &&
then_op->out_type == RAY_STR && else_op->out_type == RAY_STR &&
if_branch_cheap(g, then_op, 0) && if_branch_cheap(g, else_op, 0)) ||
(op->out_type == RAY_STR && near_full_sel && nrows >= RAY_PARALLEL_THRESHOLD &&
(then_op->out_type == RAY_SYM || else_op->out_type == RAY_SYM) &&
(then_op->out_type == RAY_SYM || then_op->out_type == RAY_STR) &&
(else_op->out_type == RAY_SYM || else_op->out_type == RAY_STR) &&
if_branch_trivial(g, then_op) && if_branch_trivial(g, else_op));
{
ray_pool_t* rp = ray_pool_get();
if (rp && rp->n_workers > 0 && eager_possible)
return NULL;
}
if_branch_plan_t then_plan, else_plan;
if (!if_make_branch_plan(g, then_op, &then_plan) ||
!if_make_branch_plan(g, else_op, &else_plan))
return NULL;
uint8_t* cond = (uint8_t*)ray_data(cond_v);
int64_t true_count = if_selected_true_count(cond, nrows, outer_sel);
int64_t false_count = selected - true_count;
if (true_count < 0 || false_count < 0) return NULL;
ray_t* ids_hdr = NULL;
int64_t* ids = NULL;
if (selected > 0) {
ids = (int64_t*)scratch_alloc(&ids_hdr, (size_t)selected * sizeof(int64_t));
if (!ids) return ray_error("oom", NULL);
if_fill_branch_ids(cond, nrows, outer_sel, ids, ids + true_count);
}
int64_t* true_ids = ids;
int64_t* false_ids = ids ? ids + true_count : NULL;
ray_t* then_v = if_eval_branch(g, then_op, &then_plan, true_ids, true_count, nrows);
if (then_v && RAY_IS_ERR(then_v)) { scratch_free(ids_hdr); return then_v; }
ray_t* else_v = if_eval_branch(g, else_op, &else_plan, false_ids, false_count, nrows);
if (else_v && RAY_IS_ERR(else_v)) {
if (then_v) ray_release(then_v);
scratch_free(ids_hdr);
return else_v;
}
int8_t out_type = if_selected_type(then_v, else_v, op->out_type, eager_possible);
if (!if_lazy_supported_type(out_type)) {
if (then_v) ray_release(then_v);
if (else_v) ray_release(else_v);
scratch_free(ids_hdr);
return NULL;
}
ray_t* result = ray_vec_new(out_type, nrows);
if (!result || RAY_IS_ERR(result)) {
if (then_v) ray_release(then_v);
if (else_v) ray_release(else_v);
scratch_free(ids_hdr);
return result;
}
result->len = nrows;
if (if_val_has_null(then_v) || if_val_has_null(else_v))
result->attrs |= RAY_ATTR_HAS_NULLS;
if (out_type == RAY_STR)
memset(ray_data(result), 0, (size_t)nrows * sizeof(ray_str_t));
else
memset(ray_data(result), 0, (size_t)nrows * ray_sym_elem_size(out_type, result->attrs));
bool ok = true;
if (out_type == RAY_STR) {
ray_t* fast = if_scatter_str_desc(result, then_v, true_ids, true_count,
else_v, false_ids, false_count, nrows);
if (fast) {
result = fast;
} else {
result = if_scatter_str(result, then_v, true_ids, true_count, nrows);
if (result && !RAY_IS_ERR(result))
result = if_scatter_str(result, else_v, false_ids, false_count, nrows);
}
} else if (out_type == RAY_SYM) {
ok = if_scatter_sym(result, then_v, true_ids, true_count, nrows) &&
if_scatter_sym(result, else_v, false_ids, false_count, nrows);
} else {
ok = if_scatter_numeric(result, then_v, true_ids, true_count, nrows, out_type) &&
if_scatter_numeric(result, else_v, false_ids, false_count, nrows, out_type);
}
if (then_v) ray_release(then_v);
if (else_v) ray_release(else_v);
scratch_free(ids_hdr);
if (!result || RAY_IS_ERR(result)) return result;
if (!ok) {
ray_release(result);
return ray_error("length", "if: branch length mismatch");
}
return result;
}
typedef struct {
const uint8_t* cond;
ray_t* then_v;
ray_t* else_v;
bool then_scalar;
bool else_scalar;
double t_f64, e_f64;
int64_t t_i64, e_i64;
void* dst;
int8_t out_type;
if_sym_side_t* t_side;
if_sym_side_t* e_side;
bool sides_prepared;
} if_fill_ctx_t;
static inline int64_t if_fill_sym_cell(const if_fill_ctx_t* c, if_sym_side_t* side,
int64_t i) {
ray_t* v = side->v;
if (c->sides_prepared && side->dom) {
int64_t pos = ray_read_sym(ray_data(v), i, v->type, v->attrs);
return if_sym_side_lookup(side, pos);
}
return if_sym_cell_value(side, i);
}
static void if_fill_range(const if_fill_ctx_t* c, int64_t i0, int64_t i1) {
const uint8_t* cond_p = c->cond;
switch (c->out_type) {
case RAY_F64: {
double* dst = (double*)c->dst;
for (int64_t i = i0; i < i1; i++) {
double tv = c->then_scalar ? c->t_f64 : if_vec_f64(c->then_v, i);
double ev = c->else_scalar ? c->e_f64 : if_vec_f64(c->else_v, i);
dst[i] = cond_p[i] ? tv : ev;
}
break; }
case RAY_I64: case RAY_TIMESTAMP: {
int64_t* dst = (int64_t*)c->dst;
for (int64_t i = i0; i < i1; i++) {
int64_t tv = c->then_scalar ? c->t_i64 : if_vec_i64(c->then_v, i);
int64_t ev = c->else_scalar ? c->e_i64 : if_vec_i64(c->else_v, i);
dst[i] = cond_p[i] ? tv : ev;
}
break; }
case RAY_SYM: {
int64_t* dst = (int64_t*)c->dst;
for (int64_t i = i0; i < i1; i++) {
if (cond_p[i])
dst[i] = c->then_scalar ? c->t_i64 : if_fill_sym_cell(c, c->t_side, i);
else
dst[i] = c->else_scalar ? c->e_i64 : if_fill_sym_cell(c, c->e_side, i);
}
break; }
case RAY_I32: case RAY_TIME: case RAY_DATE: {
int32_t* dst = (int32_t*)c->dst;
for (int64_t i = i0; i < i1; i++) {
int32_t tv = c->then_scalar ? (int32_t)c->t_i64 : (int32_t)if_vec_i64(c->then_v, i);
int32_t ev = c->else_scalar ? (int32_t)c->e_i64 : (int32_t)if_vec_i64(c->else_v, i);
dst[i] = cond_p[i] ? tv : ev;
}
break; }
case RAY_I16: {
int16_t* dst = (int16_t*)c->dst;
for (int64_t i = i0; i < i1; i++) {
int16_t tv = c->then_scalar ? (int16_t)c->t_i64 : (int16_t)if_vec_i64(c->then_v, i);
int16_t ev = c->else_scalar ? (int16_t)c->e_i64 : (int16_t)if_vec_i64(c->else_v, i);
dst[i] = cond_p[i] ? tv : ev;
}
break; }
case RAY_BOOL: case RAY_U8: {
uint8_t* dst = (uint8_t*)c->dst;
for (int64_t i = i0; i < i1; i++) {
uint8_t tv = c->then_scalar ? (uint8_t)c->t_i64 : (uint8_t)if_vec_i64(c->then_v, i);
uint8_t ev = c->else_scalar ? (uint8_t)c->e_i64 : (uint8_t)if_vec_i64(c->else_v, i);
dst[i] = cond_p[i] ? tv : ev;
}
break; }
case RAY_GUID: {
uint8_t* dst = (uint8_t*)c->dst;
const uint8_t* tb = ray_is_atom(c->then_v) ? (const uint8_t*)ray_data(c->then_v->obj)
: (const uint8_t*)ray_data(c->then_v);
const uint8_t* eb = ray_is_atom(c->else_v) ? (const uint8_t*)ray_data(c->else_v->obj)
: (const uint8_t*)ray_data(c->else_v);
for (int64_t i = i0; i < i1; i++) {
const uint8_t* src = cond_p[i]
? tb + (c->then_scalar ? 0 : i) * 16
: eb + (c->else_scalar ? 0 : i) * 16;
memcpy(dst + i * 16, src, 16);
}
break; }
default: break;
}
}
static void if_fill_par_fn(void* ctx, uint32_t wid, int64_t start, int64_t end) {
(void)wid;
if_fill_range((const if_fill_ctx_t*)ctx, start, end);
}
typedef struct {
const uint8_t* cond;
const ray_str_t* t;
const ray_str_t* e;
ray_str_t* dst;
const uint32_t* newoff;
const char* t_bytes;
const char* e_bytes;
char* dst_bytes;
} if_str_desc_ctx_t;
static void if_str_desc_fn(void* vctx, uint32_t worker_id, int64_t start, int64_t end) {
(void)worker_id;
const if_str_desc_ctx_t* c = (const if_str_desc_ctx_t*)vctx;
for (int64_t r = start; r < end; r++) {
ray_str_t d = c->cond[r] ? c->t[r] : c->e[r];
if (c->newoff && !ray_str_is_inline(&d)) {
const char* src = c->cond[r] ? c->t_bytes : c->e_bytes;
memcpy(c->dst_bytes + c->newoff[r], src + d.pool_off, d.len);
d.pool_off = c->newoff[r];
}
c->dst[r] = d;
}
}
static uint8_t if_cond_cell(ray_t* c, int64_t i) {
const void* d = ray_data(c);
switch (c->type) {
case RAY_BOOL: case RAY_U8:
return ((const uint8_t*)d)[i] != 0;
default: break;
}
if (ray_vec_is_null(c, i)) return 0;
switch (c->type) {
case RAY_I16: return ((const int16_t*)d)[i] != 0;
case RAY_I32: case RAY_DATE: case RAY_TIME:
return ((const int32_t*)d)[i] != 0;
case RAY_I64: case RAY_TIMESTAMP:
return ((const int64_t*)d)[i] != 0;
case RAY_F32: { float f = ((const float*)d)[i]; return f != 0.0f && f == f; }
case RAY_F64: { double f = ((const double*)d)[i]; return f != 0.0 && f == f; }
default: return 0;
}
}
static uint8_t if_cond_atom(ray_t* x) {
if (RAY_ATOM_IS_NULL(x)) return 0;
switch (x->type) {
case -RAY_BOOL: return x->b8 != 0;
case -RAY_U8: return x->u8 != 0;
case -RAY_I16: return x->i16 != 0;
case -RAY_I32: return x->i32 != 0;
case -RAY_I64: case -RAY_DATE: case -RAY_TIME: case -RAY_TIMESTAMP:
return x->i64 != 0;
case -RAY_F32: case -RAY_F64:
return x->f64 != 0.0 && x->f64 == x->f64;
default:
return 1;
}
}
static bool if_cond_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;
}
}
enum { IF_TXT_SCALAR = 1, IF_TXT_STR, IF_TXT_SYM_RT, IF_TXT_SYM_FILE };
typedef struct {
uint8_t kind;
const char* sp;
uint32_t sl;
const ray_str_t* desc;
const char* bytes;
const void* sym;
int8_t sym_type;
uint8_t sym_attrs;
ray_t** rt_strings;
uint32_t rt_count;
struct ray_sym_domain_s* dom;
ray_sym_domain_raw_t raw;
} if_txt_side_t;
static bool if_txt_side_init(if_txt_side_t* s, ray_t* v, bool scalar, int64_t len) {
memset(s, 0, sizeof(*s));
if (scalar) {
const char* sp = "";
size_t sl = 0;
if (v->type == -RAY_STR) {
sp = ray_str_ptr(v); sl = ray_str_len(v);
} else if (v->type == RAY_STR) {
sp = ray_str_vec_get(v, 0, &sl);
if (!sp) { sp = ""; sl = 0; }
} else if (v->type == -RAY_SYM || RAY_IS_SYM(v->type)) {
ray_t* a = sym_scalar_str(v);
sp = a ? ray_str_ptr(a) : "";
sl = a ? ray_str_len(a) : 0;
} else {
return false;
}
if (sl > UINT32_MAX) return false;
s->kind = IF_TXT_SCALAR; s->sp = sp; s->sl = (uint32_t)sl;
return true;
}
if (v->len < len) return false;
if (v->type == RAY_STR) {
s->kind = IF_TXT_STR;
str_resolve(v, &s->desc, &s->bytes);
return true;
}
if (!RAY_IS_SYM(v->type)) return false;
s->sym = ray_data(v);
s->sym_type = v->type;
s->sym_attrs = v->attrs;
struct ray_sym_domain_s* dom = ray_sym_vec_domain(v);
if (dom == ray_sym_runtime_domain()) {
ray_sym_strings_borrow(&s->rt_strings, &s->rt_count);
if (!s->rt_strings) return false;
s->kind = IF_TXT_SYM_RT;
return true;
}
if (!ray_sym_domain_raw_pin(dom, &s->raw)) return false;
s->dom = dom;
s->kind = IF_TXT_SYM_FILE;
return true;
}
static void if_txt_side_free(if_txt_side_t* s) {
if (s->kind == IF_TXT_SYM_FILE) ray_sym_domain_raw_unpin(s->dom);
s->kind = 0;
}
static inline bool if_txt_get(const if_txt_side_t* s, int64_t r,
const char** sp, uint32_t* sl) {
switch (s->kind) {
case IF_TXT_SCALAR:
*sp = s->sp; *sl = s->sl;
return true;
case IF_TXT_STR: {
const ray_str_t* d = &s->desc[r];
*sl = d->len;
*sp = d->len == 0 ? "" : ray_str_is_inline(d) ? d->data : s->bytes + d->pool_off;
return true;
}
case IF_TXT_SYM_RT: {
int64_t id = ray_read_sym(s->sym, r, s->sym_type, s->sym_attrs);
if (id < 0 || id >= (int64_t)s->rt_count || !s->rt_strings[id]) return false;
ray_t* a = s->rt_strings[id];
*sp = ray_str_ptr(a); *sl = (uint32_t)ray_str_len(a);
return true;
}
case IF_TXT_SYM_FILE: {
int64_t pos = ray_read_sym(s->sym, r, s->sym_type, s->sym_attrs);
if (pos < 0 || pos >= s->raw.count) return false;
size_t l = 0;
*sp = ray_sym_domain_raw_str(&s->raw, pos, &l);
*sl = (uint32_t)l;
return true;
}
default:
return false;
}
}
typedef struct {
const uint8_t* cond;
const if_txt_side_t* t;
const if_txt_side_t* e;
uint32_t* off;
ray_str_t* dst;
char* pool;
_Atomic(int) miss;
} if_txt_ctx_t;
static void if_txt_len_fn(void* vctx, uint32_t wid, int64_t start, int64_t end) {
(void)wid;
if_txt_ctx_t* c = (if_txt_ctx_t*)vctx;
for (int64_t r = start; r < end; r++) {
const char* sp; uint32_t sl;
if (!if_txt_get(c->cond[r] ? c->t : c->e, r, &sp, &sl)) {
atomic_store_explicit(&c->miss, 1, memory_order_relaxed);
return;
}
c->off[r] = sl > RAY_STR_INLINE_MAX ? sl : 0;
}
}
static void if_txt_fill_fn(void* vctx, uint32_t wid, int64_t start, int64_t end) {
(void)wid;
if_txt_ctx_t* c = (if_txt_ctx_t*)vctx;
for (int64_t r = start; r < end; r++) {
const char* sp = ""; uint32_t sl = 0;
if (!if_txt_get(c->cond[r] ? c->t : c->e, r, &sp, &sl)) { sp = ""; sl = 0; }
ray_str_t* d = &c->dst[r];
memset(d, 0, sizeof(*d));
d->len = sl;
if (sl <= RAY_STR_INLINE_MAX) {
if (sl) memcpy(d->data, sp, sl);
} else {
memcpy(d->prefix, sp, 4);
d->pool_off = c->off[r];
memcpy(c->pool + c->off[r], sp, sl);
}
}
}
static ray_t* if_str_text_build(ray_t* result, const uint8_t* cond, int64_t len,
ray_t* then_v, bool then_scalar,
ray_t* else_v, bool else_scalar) {
bool t_sym = then_v->type == -RAY_SYM || RAY_IS_SYM(then_v->type);
bool e_sym = else_v->type == -RAY_SYM || RAY_IS_SYM(else_v->type);
if (!t_sym && !e_sym) return NULL;
if (len <= 0) return NULL;
if_txt_side_t t, e;
if (!if_txt_side_init(&t, then_v, then_scalar, len)) {
if_txt_side_free(&t);
return NULL;
}
if (!if_txt_side_init(&e, else_v, else_scalar, len)) {
if_txt_side_free(&t); if_txt_side_free(&e);
return NULL;
}
ray_t* off_hdr = NULL;
uint32_t* off = (uint32_t*)scratch_alloc(&off_hdr, (size_t)len * sizeof(uint32_t));
if (!off) { if_txt_side_free(&t); if_txt_side_free(&e); return NULL; }
if_txt_ctx_t c = { .cond = cond, .t = &t, .e = &e, .off = off,
.dst = (ray_str_t*)ray_data(result), .pool = NULL };
atomic_init(&c.miss, 0);
ray_pool_t* pool = ray_pool_get();
bool par = ray_pool_par_dispatch_ok(pool, len, RAY_PARALLEL_THRESHOLD);
if (par) ray_pool_dispatch(pool, if_txt_len_fn, &c, len);
else if_txt_len_fn(&c, 0, 0, len);
uint64_t run = 0;
bool ok = atomic_load_explicit(&c.miss, memory_order_relaxed) == 0;
if (ok) {
for (int64_t r = 0; r < len; r++) {
uint32_t l = off[r];
off[r] = (uint32_t)run;
run += l;
if (run > UINT32_MAX) { ok = false; break; }
}
}
ray_t* np = NULL;
if (ok && run > 0) {
np = ray_alloc((size_t)run);
if (!np || RAY_IS_ERR(np)) { np = NULL; ok = false; }
else { np->type = RAY_U8; np->len = (int64_t)run; c.pool = (char*)ray_data(np); }
}
if (ok) {
if (par) ray_pool_dispatch(pool, if_txt_fill_fn, &c, len);
else if_txt_fill_fn(&c, 0, 0, len);
result->str_pool = np;
}
scratch_free(off_hdr);
if_txt_side_free(&t);
if_txt_side_free(&e);
return ok ? result : NULL;
}
static ray_t* exec_if_eager(ray_graph_t* g, ray_op_t* op, ray_t* cond_in) {
ray_t* cond_v = cond_in ? cond_in : exec_node(g, op_child(g, op, 0));
ray_t* then_v = exec_node(g, op_child(g, op, 1));
ray_op_ext_t* ext = find_ext(g, op->id);
uint32_t else_id = ext->third_in;
ray_t* else_v = exec_node(g, op_node(g, else_id));
if (!cond_v || RAY_IS_ERR(cond_v)) {
if (then_v && !RAY_IS_ERR(then_v)) ray_release(then_v);
if (else_v && !RAY_IS_ERR(else_v)) ray_release(else_v);
return cond_v;
}
if (!then_v || RAY_IS_ERR(then_v)) {
ray_release(cond_v);
if (else_v && !RAY_IS_ERR(else_v)) ray_release(else_v);
return then_v;
}
if (!else_v || RAY_IS_ERR(else_v)) {
ray_release(cond_v); ray_release(then_v);
return else_v;
}
int64_t branch_len = 1;
if (!ray_is_atom(then_v) && then_v->len > branch_len) branch_len = then_v->len;
if (!ray_is_atom(else_v) && else_v->len > branch_len) branch_len = else_v->len;
if (!ray_is_atom(cond_v) && !if_cond_type_ok(cond_v->type)) {
int8_t ct = cond_v->type;
g->if_refused = true;
ray_release(cond_v); ray_release(then_v); ray_release(else_v);
return ray_error("type", "if: condition must be boolean, numeric or temporal, got %s",
ray_type_name(ct));
}
bool cond_bcast = ray_is_atom(cond_v) || (cond_v->len == 1 && branch_len > 1);
if (cond_bcast || cond_v->type != RAY_BOOL) {
int64_t n = cond_bcast ? branch_len : cond_v->len;
ray_t* mask = ray_vec_new(RAY_BOOL, n);
if (!mask || RAY_IS_ERR(mask)) {
ray_release(cond_v); ray_release(then_v); ray_release(else_v);
return mask ? mask : ray_error("oom", NULL);
}
uint8_t* m = (uint8_t*)ray_data(mask);
if (cond_bcast)
memset(m, ray_is_atom(cond_v) ? if_cond_atom(cond_v)
: if_cond_cell(cond_v, 0), (size_t)n);
else
for (int64_t i = 0; i < n; i++) m[i] = if_cond_cell(cond_v, i);
mask->len = n;
ray_release(cond_v);
cond_v = mask;
}
int64_t len = cond_v->len;
bool then_scalar = ray_is_atom(then_v) || (then_v->type > 0 && then_v->len == 1);
bool else_scalar = ray_is_atom(else_v) || (else_v->type > 0 && else_v->len == 1);
if (then_scalar && !else_scalar) len = else_v->len;
if (!then_scalar) len = then_v->len;
int8_t out_type = op->out_type;
ray_t* result = ray_vec_new(out_type, len);
if (!result || RAY_IS_ERR(result)) {
ray_release(cond_v); ray_release(then_v); ray_release(else_v);
return result;
}
result->len = len;
if (if_val_has_null(then_v) || if_val_has_null(else_v))
result->attrs |= RAY_ATTR_HAS_NULLS;
uint8_t* cond_p = (uint8_t*)ray_data(cond_v);
if (out_type == RAY_STR) {
if (if_str_text_build(result, cond_p, len, then_v, then_scalar,
else_v, else_scalar)) {
ray_release(cond_v); ray_release(then_v); ray_release(else_v);
return result;
}
if (!then_scalar && !else_scalar &&
then_v->type == RAY_STR && else_v->type == RAY_STR &&
len <= then_v->len && len <= else_v->len) {
ray_t* then_pool = str_vec_pool_obj(then_v);
ray_t* else_pool = str_vec_pool_obj(else_v);
const ray_str_t* t_desc = NULL;
const ray_str_t* e_desc = NULL;
const char* t_bytes = NULL;
const char* e_bytes = NULL;
str_resolve(then_v, &t_desc, &t_bytes);
str_resolve(else_v, &e_desc, &e_bytes);
bool ok = true;
ray_t* off_hdr = NULL;
uint32_t* newoff = NULL;
if (then_pool == else_pool || !then_pool || !else_pool) {
ray_t* out_pool = then_pool ? then_pool : else_pool;
if (out_pool && !RAY_IS_ERR(out_pool)) {
ray_retain(out_pool);
result->str_pool = out_pool;
}
} else if (RAY_IS_ERR(then_pool) || RAY_IS_ERR(else_pool)) {
ok = false;
} else {
newoff = (uint32_t*)scratch_alloc(&off_hdr, (size_t)(len > 0 ? len : 1) * sizeof(uint32_t));
if (!newoff) {
ok = false;
} else {
uint64_t run = 0;
for (int64_t r = 0; r < len; r++) {
const ray_str_t* d = cond_p[r] ? &t_desc[r] : &e_desc[r];
if (ray_str_is_inline(d)) continue;
newoff[r] = (uint32_t)run;
run += d->len;
}
ray_t* np = (run <= UINT32_MAX) ? ray_alloc(run > 0 ? (size_t)run : 1) : NULL;
if (!np || RAY_IS_ERR(np)) {
ok = false;
} else {
np->type = RAY_U8;
np->len = (int64_t)run;
result->str_pool = np;
}
}
if (!ok && off_hdr) { scratch_free(off_hdr); off_hdr = NULL; newoff = NULL; }
}
if (ok) {
if_str_desc_ctx_t dctx = {
.cond = cond_p, .t = t_desc, .e = e_desc,
.dst = (ray_str_t*)ray_data(result),
.newoff = newoff, .t_bytes = t_bytes, .e_bytes = e_bytes,
.dst_bytes = result->str_pool ? (char*)ray_data(result->str_pool) : NULL,
};
ray_pool_t* pool = ray_pool_get();
if (ray_pool_par_dispatch_ok(pool, len, RAY_PARALLEL_THRESHOLD))
ray_pool_dispatch(pool, if_str_desc_fn, &dctx, len);
else
if_str_desc_fn(&dctx, 0, 0, len);
if (off_hdr) scratch_free(off_hdr);
if (ray_vec_may_have_nulls(then_v) || ray_vec_may_have_nulls(else_v))
result->attrs |= RAY_ATTR_HAS_NULLS;
ray_release(cond_v); ray_release(then_v); ray_release(else_v);
return result;
}
}
result->len = 0;
for (int64_t i = 0; i < len; i++) {
const char* sp;
size_t sl;
if (cond_p[i]) {
if (then_scalar) {
if (then_v->type == -RAY_STR) {
sp = ray_str_ptr(then_v);
sl = ray_str_len(then_v);
} else if (then_v->type == RAY_STR) {
sp = ray_str_vec_get(then_v, 0, &sl);
if (!sp) { sp = ""; sl = 0; }
} else if (RAY_IS_SYM(then_v->type) || then_v->type == -RAY_SYM) {
ray_t* s = sym_scalar_str(then_v);
sp = s ? ray_str_ptr(s) : "";
sl = s ? ray_str_len(s) : 0;
} else { sp = ""; sl = 0; }
} else if (then_v->type == RAY_STR) {
sp = ray_str_vec_get(then_v, i, &sl);
if (!sp) { sp = ""; sl = 0; }
} else {
ray_t* sa = ray_sym_vec_cell(then_v, i);
sp = sa ? ray_str_ptr(sa) : "";
sl = sa ? ray_str_len(sa) : 0;
}
} else {
if (else_scalar) {
if (else_v->type == -RAY_STR) {
sp = ray_str_ptr(else_v);
sl = ray_str_len(else_v);
} else if (else_v->type == RAY_STR) {
sp = ray_str_vec_get(else_v, 0, &sl);
if (!sp) { sp = ""; sl = 0; }
} else if (RAY_IS_SYM(else_v->type) || else_v->type == -RAY_SYM) {
ray_t* s = sym_scalar_str(else_v);
sp = s ? ray_str_ptr(s) : "";
sl = s ? ray_str_len(s) : 0;
} else { sp = ""; sl = 0; }
} else if (else_v->type == RAY_STR) {
sp = ray_str_vec_get(else_v, i, &sl);
if (!sp) { sp = ""; sl = 0; }
} else {
ray_t* sa = ray_sym_vec_cell(else_v, i);
sp = sa ? ray_str_ptr(sa) : "";
sl = sa ? ray_str_len(sa) : 0;
}
}
result = ray_str_vec_append(result, sp, sl);
if (RAY_IS_ERR(result)) break;
}
} else {
if_fill_ctx_t fc = {
.cond = cond_p, .then_v = then_v, .else_v = else_v,
.then_scalar = then_scalar, .else_scalar = else_scalar,
.t_f64 = 0.0, .e_f64 = 0.0, .t_i64 = 0, .e_i64 = 0,
.dst = ray_data(result), .out_type = out_type,
};
if (out_type == RAY_SYM) {
if (then_scalar)
fc.t_i64 = (then_v->type == -RAY_STR)
? ray_sym_intern(ray_str_ptr(then_v), ray_str_len(then_v))
: sym_scalar_runtime_id(then_v);
if (else_scalar)
fc.e_i64 = (else_v->type == -RAY_STR)
? ray_sym_intern(ray_str_ptr(else_v), ray_str_len(else_v))
: sym_scalar_runtime_id(else_v);
} else {
if (then_scalar) { fc.t_f64 = if_scalar_f64(then_v); fc.t_i64 = if_scalar_i64(then_v); }
if (else_scalar) { fc.e_f64 = if_scalar_f64(else_v); fc.e_i64 = if_scalar_i64(else_v); }
}
ray_pool_t* pool = ray_pool_get();
bool par = ray_pool_par_dispatch_ok(pool, len, RAY_PARALLEL_THRESHOLD);
if_sym_side_t t_side, e_side;
bool sides_ok = true;
if (out_type == RAY_SYM) {
bool t_ok = if_sym_side_init(&t_side, then_scalar ? NULL : then_v);
bool e_ok = if_sym_side_init(&e_side, else_scalar ? NULL : else_v);
sides_ok = t_ok && e_ok;
fc.t_side = &t_side;
fc.e_side = &e_side;
}
if (!sides_ok) {
if (out_type == RAY_SYM) { if_sym_side_free(&t_side); if_sym_side_free(&e_side); }
ray_release(cond_v); ray_release(then_v); ray_release(else_v);
ray_release(result);
return ray_error("oom", NULL);
}
if (par && out_type == RAY_SYM) {
ray_t* sides[2] = { then_scalar ? NULL : then_v,
else_scalar ? NULL : else_v };
for (int s = 0; s < 2 && par; s++) {
if (!sides[s]) continue;
if (sides[s]->type != RAY_SYM) { par = false; break; }
}
if (par) {
if_sym_side_prepare(&t_side, cond_p, len, 1);
if_sym_side_prepare(&e_side, cond_p, len, 0);
fc.sides_prepared = true;
}
}
if (par)
ray_pool_dispatch(pool, if_fill_par_fn, &fc, len);
else
if_fill_range(&fc, 0, len);
if (out_type == RAY_SYM) { if_sym_side_free(&t_side); if_sym_side_free(&e_side); }
}
ray_release(cond_v); ray_release(then_v); ray_release(else_v);
return result;
}
static bool if_type_is_text(int8_t t) { return t == RAY_STR || t == RAY_SYM; }
static bool if_branch_text_ok(ray_graph_t* g, ray_op_t* b) {
if (!b) return true;
int8_t t = b->out_type;
if (t <= 0 || t == RAY_LIST || if_type_is_text(t)) return true;
if (b->opcode == OP_CONST) {
ray_op_ext_t* ext = find_ext(g, b->id);
if (ext && ext->literal && ray_is_atom(ext->literal) &&
RAY_ATOM_IS_NULL(ext->literal))
return true;
}
return false;
}
ray_t* exec_if(ray_graph_t* g, ray_op_t* op) {
if (if_type_is_text(op->out_type)) {
ray_op_t* then_op = op_child(g, op, 1);
ray_op_ext_t* ext = find_ext(g, op->id);
ray_op_t* else_op = ext ? op_node(g, ext->third_in) : NULL;
int8_t tt = then_op ? then_op->out_type : 0;
int8_t et = else_op ? else_op->out_type : 0;
if (!if_branch_text_ok(g, then_op) || !if_branch_text_ok(g, else_op)) {
g->if_refused = true;
return ray_error("type", "if: branches mix %s and %s",
ray_type_name(tt), ray_type_name(et));
}
}
ray_t* cond_v = exec_node(g, op_child(g, op, 0));
if (!cond_v || RAY_IS_ERR(cond_v)) return cond_v;
ray_t* selected = exec_if_selected(g, op, cond_v);
if (selected) {
ray_release(cond_v);
return selected;
}
return exec_if_eager(g, op, cond_v);
}
static inline uint64_t pivot_hash_null_words(uint64_t h, const int64_t* nullw,
uint32_t null_words) {
for (uint32_t w = 0; w < null_words; w++)
if (nullw[w]) h = ray_hash_combine(h, ray_hash_i64(nullw[w]));
return h;
}
static int64_t pivot_parted_seg_count(ray_t* tbl) {
int64_t nc = ray_table_ncols(tbl);
for (int64_t c = 0; c < nc; c++) {
ray_t* col = ray_table_get_col_idx(tbl, c);
if (col && RAY_IS_PARTED(col->type)) return col->len;
}
return 0;
}
static bool pivot_table_has_parted_col(ray_t* tbl) {
int64_t nc = ray_table_ncols(tbl);
for (int64_t c = 0; c < nc; c++) {
ray_t* col = ray_table_get_col_idx(tbl, c);
if (col && (RAY_IS_PARTED(col->type) || col->type == RAY_MAPCOMMON))
return true;
}
return false;
}
ray_t* exec_pivot(ray_graph_t* g, ray_op_t* op, ray_t* tbl) {
if (!tbl || RAY_IS_ERR(tbl)) return tbl;
if (pivot_table_has_parted_col(tbl)) {
int64_t nseg = pivot_parted_seg_count(tbl);
ray_t* flat = NULL;
for (int64_t s = 0; s < nseg; s++) {
ray_t* seg = build_segment_table(tbl, (int32_t)s);
if (!seg || RAY_IS_ERR(seg)) { if (flat) ray_release(flat); return seg ? seg : ray_error("oom", NULL); }
if (!flat) flat = seg;
else {
ray_t* m = ray_result_merge(flat, seg);
ray_release(flat); ray_release(seg);
if (!m || RAY_IS_ERR(m)) return m ? m : ray_error("oom", NULL);
flat = m;
}
}
ray_t* r = exec_pivot(g, op, flat);
ray_release(flat);
return r;
}
ray_op_ext_t* ext = find_ext(g, op->id);
if (!ext) return ray_error("nyi", NULL);
uint32_t n_idx = ext->pivot.n_index;
uint16_t agg_op = ext->pivot.agg_op;
int64_t nrows = ray_table_nrows(tbl);
uint32_t n_idx_cap = n_idx ? n_idx : 1;
uint32_t n_keys = n_idx + 1;
size_t key_ptr_bytes = (size_t)n_idx_cap * sizeof(ray_t*)
+ (size_t)n_keys * sizeof(void*)
+ (size_t)n_keys * sizeof(ray_t*);
size_t key_byte_bytes = (size_t)n_idx_cap * sizeof(bool)
+ (size_t)n_keys * sizeof(int8_t)
+ (size_t)n_keys * sizeof(uint8_t);
ray_t* key_hdr = NULL;
uint8_t* key_base = (uint8_t*)scratch_alloc(&key_hdr, key_ptr_bytes + key_byte_bytes);
if (!key_base) return ray_error("oom", NULL);
ray_t** idx_vecs = (ray_t**)key_base;
void** key_data = (void**)(idx_vecs + n_idx_cap);
ray_t** key_vecs = (ray_t**)(key_data + n_keys);
bool* idx_wide = (bool*)(key_vecs + n_keys);
int8_t* key_types = (int8_t*)(idx_wide + n_idx_cap);
uint8_t* key_attrs = (uint8_t*)(key_types + n_keys);
memset(idx_wide, 0, (size_t)n_idx_cap * sizeof(bool));
for (uint32_t i = 0; i < n_idx; i++) {
ray_op_ext_t* ie = find_ext(g, ext->pivot.index_cols[i]);
idx_vecs[i] = (ie && ie->base.opcode == OP_SCAN)
? ray_table_get_col(tbl, ie->sym) : NULL;
if (!idx_vecs[i]) { scratch_free(key_hdr); return ray_error("domain", "pivot: index column not found"); }
}
ray_op_ext_t* pe = find_ext(g, ext->pivot.pivot_col);
ray_t* pcol = (pe && pe->base.opcode == OP_SCAN)
? ray_table_get_col(tbl, pe->sym) : NULL;
if (!pcol) { scratch_free(key_hdr); return ray_error("domain", "pivot: pivot column not found"); }
ray_op_ext_t* ve = find_ext(g, ext->pivot.value_col);
ray_t* vcol = (ve && ve->base.opcode == OP_SCAN)
? ray_table_get_col(tbl, ve->sym) : NULL;
if (!vcol) { scratch_free(key_hdr); return ray_error("domain", "pivot: value column not found"); }
if (nrows == 0) { scratch_free(key_hdr); return ray_table_new(0); }
for (uint32_t k = 0; k < n_idx; k++)
idx_wide[k] = (idx_vecs[k]->type == RAY_GUID);
bool pvt_wide = (pcol->type == RAY_GUID);
for (uint32_t k = 0; k < n_idx; k++) {
key_data[k] = ray_data(idx_vecs[k]);
key_types[k] = idx_vecs[k]->type;
key_attrs[k] = idx_vecs[k]->attrs;
key_vecs[k] = idx_vecs[k];
}
key_data[n_idx] = ray_data(pcol);
key_types[n_idx] = pcol->type;
key_attrs[n_idx] = pcol->attrs;
key_vecs[n_idx] = pcol;
ray_t* agg_vecs[1] = { vcol };
uint16_t agg_ops[1] = { agg_op };
uint8_t need_flags = GHT_NEED_SUM;
if (agg_op == OP_MIN) need_flags |= GHT_NEED_MIN;
if (agg_op == OP_MAX) need_flags |= GHT_NEED_MAX;
if (agg_op == OP_AVG && (vcol->type == RAY_I64 || vcol->type == RAY_TIMESTAMP ||
(vcol->type != RAY_F64 && vcol->type != RAY_F32 &&
nrows >= ((int64_t)1 << 31))))
need_flags |= GHT_NEED_SUM128;
ght_layout_t ly;
if (!ght_compute_layout(&ly, n_keys, 1, agg_vecs, NULL,
need_flags, agg_ops, key_types, NULL)) {
scratch_free(key_hdr);
return ray_error("limit", "pivot: key stride budget exceeded");
}
ray_progress_update("pivot", "hash-aggregate", 0, (uint64_t)nrows);
pivot_ingest_t pg;
if (!pivot_ingest_run(&pg, &ly, key_data, key_types, key_attrs,
key_vecs, agg_vecs, nrows)) {
pivot_ingest_free(&pg);
ght_layout_free(&ly);
scratch_free(key_hdr);
return ray_error("oom", NULL);
}
ray_progress_update("pivot", "dedupe", 0, (uint64_t)pg.total_grps);
if (ray_interrupted()) {
pivot_ingest_free(&pg);
ght_layout_free(&ly);
scratch_free(key_hdr);
return ray_error("cancel", "interrupted");
}
uint32_t grp_count = pg.total_grps;
if (grp_count == 0) {
pivot_ingest_free(&pg);
ght_layout_free(&ly);
scratch_free(key_hdr);
return ray_table_new(0);
}
const uint32_t null_words = ly.null_words;
const uint32_t pvt_null_word = n_idx >> 6;
const int64_t pvt_null_bit = (int64_t)((uint64_t)1 << (n_idx & 63));
ray_t* pv_first_hdr = NULL;
uint32_t pv_first_cap = 64;
int64_t* pv_first = (int64_t*)scratch_alloc(&pv_first_hdr, pv_first_cap * sizeof(int64_t));
if (!pv_first) { pivot_ingest_free(&pg); ght_layout_free(&ly); scratch_free(key_hdr); return ray_error("oom", NULL); }
uint32_t pv_cap = 64, pv_count = 0;
ray_t* pv_hdr = NULL;
int64_t* pv_vals = (int64_t*)scratch_alloc(&pv_hdr, pv_cap * sizeof(int64_t));
if (!pv_vals) { scratch_free(pv_first_hdr); pivot_ingest_free(&pg); ght_layout_free(&ly); scratch_free(key_hdr); return ray_error("oom", NULL); }
const char* pvt_base = pvt_wide ? (const char*)key_data[n_idx] : NULL;
for (uint32_t _p = 0; _p < pg.n_parts; _p++) {
group_ht_t* ph = &pg.part_hts[_p];
uint32_t pcount = ph->grp_count;
for (uint32_t gi_local = 0; gi_local < pcount; gi_local++) {
const char* row = ph->rows + (size_t)gi_local * pg.row_stride;
const int64_t* rkeys = (const int64_t*)(row + 8);
if (rkeys[n_keys + pvt_null_word] & pvt_null_bit) continue;
int64_t pval = rkeys[n_idx];
int64_t gfirst;
memcpy(&gfirst, row + ly.off_group_first, 8);
bool found = false;
for (uint32_t p = 0; p < pv_count; p++) {
if (pvt_wide) {
if (memcmp(pvt_base + (size_t)pv_vals[p] * 16,
pvt_base + (size_t)pval * 16, 16) == 0) { found = true; if (gfirst < pv_first[p]) pv_first[p] = gfirst; break; }
} else {
if (pv_vals[p] == pval) { found = true; if (gfirst < pv_first[p]) pv_first[p] = gfirst; break; }
}
}
if (!found) {
if (pv_count >= pv_cap) {
uint32_t new_cap = pv_cap * 2;
int64_t* new_pv = (int64_t*)scratch_realloc(&pv_hdr,
pv_cap * sizeof(int64_t), new_cap * sizeof(int64_t));
int64_t* new_pf = (int64_t*)scratch_realloc(&pv_first_hdr,
pv_first_cap * sizeof(int64_t), new_cap * sizeof(int64_t));
if (!new_pv || !new_pf) { scratch_free(pv_first_hdr); pivot_ingest_free(&pg); ght_layout_free(&ly); scratch_free(key_hdr); return ray_error("oom", NULL); }
pv_vals = new_pv; pv_first = new_pf;
pv_cap = new_cap; pv_first_cap = new_cap;
}
pv_first[pv_count] = gfirst;
pv_vals[pv_count++] = pval;
}
}
}
uint32_t ix_cap = 256, ix_count = 0;
ray_t* ix_hdr = NULL;
size_t ix_entry = 8 + (size_t)n_idx * 8 + (size_t)null_words * 8;
char* ix_rows = (char*)scratch_alloc(&ix_hdr, ix_cap * ix_entry);
if (!ix_rows) { scratch_free(pv_first_hdr); scratch_free(pv_hdr); pivot_ingest_free(&pg); ght_layout_free(&ly); scratch_free(key_hdr); return ray_error("oom", NULL); }
ray_t* ix_first_hdr = NULL;
int64_t* ix_first = (int64_t*)scratch_alloc(&ix_first_hdr, ix_cap * sizeof(int64_t));
if (!ix_first) { scratch_free(ix_hdr); scratch_free(pv_first_hdr); scratch_free(pv_hdr); pivot_ingest_free(&pg); ght_layout_free(&ly); scratch_free(key_hdr); return ray_error("oom", NULL); }
uint32_t ix_ht_cap = 256;
while (ix_ht_cap < (uint32_t)grp_count * 2 && ix_ht_cap < (1u << 30)) ix_ht_cap <<= 1;
ray_t* ix_ht_hdr = NULL;
uint32_t* ix_ht = (uint32_t*)scratch_alloc(&ix_ht_hdr, ix_ht_cap * sizeof(uint32_t));
if (!ix_ht) {
scratch_free(ix_hdr); scratch_free(pv_hdr); pivot_ingest_free(&pg); ght_layout_free(&ly);
scratch_free(key_hdr);
scratch_free(ix_first_hdr); scratch_free(pv_first_hdr);
return ray_error("oom", NULL);
}
memset(ix_ht, 0xFF, ix_ht_cap * sizeof(uint32_t));
uint32_t ix_ht_mask = ix_ht_cap - 1;
ray_t* map_hdr = NULL;
uint32_t* grp_ix = (uint32_t*)scratch_alloc(&map_hdr, grp_count * 2 * sizeof(uint32_t));
if (!grp_ix) {
scratch_free(ix_ht_hdr); scratch_free(ix_hdr); scratch_free(pv_hdr);
pivot_ingest_free(&pg); ght_layout_free(&ly);
scratch_free(key_hdr);
scratch_free(ix_first_hdr); scratch_free(pv_first_hdr);
return ray_error("oom", NULL);
}
uint32_t* grp_pv = grp_ix + grp_count;
for (uint32_t _p = 0; _p < pg.n_parts; _p++) {
group_ht_t* ph = &pg.part_hts[_p];
uint32_t pcount = ph->grp_count;
uint32_t gi_base = pg.part_offsets[_p];
ray_progress_update(NULL, NULL, gi_base, (uint64_t)grp_count);
for (uint32_t gi_local = 0; gi_local < pcount; gi_local++) {
uint32_t gi = gi_base + gi_local;
const char* row = ph->rows + (size_t)gi_local * pg.row_stride;
const int64_t* keys = (const int64_t*)(row + 8);
if (keys[n_keys + pvt_null_word] & pvt_null_bit) {
grp_ix[gi] = UINT32_MAX;
grp_pv[gi] = UINT32_MAX;
continue;
}
int64_t gfirst;
memcpy(&gfirst, row + ly.off_group_first, 8);
const int64_t* idx_nwords = keys + n_keys;
uint64_t ih = 0;
for (uint32_t k = 0; k < n_idx; k++) {
uint64_t kh;
if (idx_wide[k]) {
const char* base = (const char*)key_data[k];
kh = ray_hash_bytes(base + (size_t)keys[k] * 16, 16);
} else if (key_types[k] == RAY_F64) {
double kd;
memcpy(&kd, &keys[k], sizeof kd);
kh = ray_hash_f64(kd);
} else {
kh = ray_hash_i64(keys[k]);
}
ih = (k == 0) ? kh : ray_hash_combine(ih, kh);
}
ih = pivot_hash_null_words(ih, idx_nwords, null_words);
uint32_t ix_row = UINT32_MAX;
uint32_t slot = (uint32_t)(ih & ix_ht_mask);
for (;;) {
uint32_t ent = ix_ht[slot];
if (ent == UINT32_MAX) break;
const char* ix_entry_p = ix_rows + (size_t)ent * ix_entry;
if (*(const uint64_t*)ix_entry_p == ih) {
const int64_t* ekeys = (const int64_t*)(ix_entry_p + 8);
bool eq = true;
for (uint32_t k = 0; k < n_idx && eq; k++) {
if (idx_wide[k]) {
const char* base = (const char*)key_data[k];
eq = (memcmp(base + (size_t)ekeys[k] * 16,
base + (size_t)keys[k] * 16, 16) == 0);
} else {
eq = (ekeys[k] == keys[k]);
}
}
if (eq) {
const int64_t* ent_nwords = (const int64_t*)(ix_entry_p + 8 + (size_t)n_idx * 8);
eq = (memcmp(ent_nwords, idx_nwords, (size_t)null_words * 8) == 0);
}
if (eq) { ix_row = ent; if (gfirst < ix_first[ent]) ix_first[ent] = gfirst; break; }
}
slot = (slot + 1) & ix_ht_mask;
}
if (ix_row == UINT32_MAX) {
if (ix_count >= ix_cap) {
uint32_t new_cap = ix_cap * 2;
char* new_rows = (char*)scratch_realloc(&ix_hdr,
ix_cap * ix_entry, new_cap * ix_entry);
if (!new_rows) {
scratch_free(map_hdr); scratch_free(ix_ht_hdr);
scratch_free(pv_hdr); pivot_ingest_free(&pg); ght_layout_free(&ly);
scratch_free(key_hdr);
scratch_free(ix_first_hdr); scratch_free(pv_first_hdr);
return ray_error("oom", NULL);
}
int64_t* new_first = (int64_t*)scratch_realloc(&ix_first_hdr,
ix_cap * sizeof(int64_t), new_cap * sizeof(int64_t));
if (!new_first) {
scratch_free(map_hdr); scratch_free(ix_ht_hdr); scratch_free(ix_first_hdr);
scratch_free(ix_hdr); scratch_free(pv_first_hdr); scratch_free(pv_hdr);
pivot_ingest_free(&pg); ght_layout_free(&ly); scratch_free(key_hdr);
return ray_error("oom", NULL);
}
ix_first = new_first;
ix_rows = new_rows;
ix_cap = new_cap;
}
ix_row = ix_count++;
ix_first[ix_row] = gfirst;
char* dst = ix_rows + (size_t)ix_row * ix_entry;
*(uint64_t*)dst = ih;
memcpy(dst + 8, keys, (size_t)n_idx * 8);
memcpy(dst + 8 + (size_t)n_idx * 8, idx_nwords, (size_t)null_words * 8);
ix_ht[slot] = ix_row;
}
int64_t pval = keys[n_idx];
uint32_t pv_idx = UINT32_MAX;
for (uint32_t p = 0; p < pv_count; p++) {
if (pvt_wide) {
if (memcmp(pvt_base + (size_t)pv_vals[p] * 16,
pvt_base + (size_t)pval * 16, 16) == 0) { pv_idx = p; break; }
} else {
if (pv_vals[p] == pval) { pv_idx = p; break; }
}
}
grp_ix[gi] = ix_row;
grp_pv[gi] = pv_idx;
}
}
{
ray_t* rank_hdr = NULL;
int64_t* pv_tmp = (int64_t*)scratch_alloc(&rank_hdr,
((size_t)pv_count + 1) * sizeof(int64_t) + ((size_t)pv_count + (size_t)ix_count + 1) * sizeof(uint32_t));
uint32_t* pv_rank = pv_tmp ? (uint32_t*)(pv_tmp + pv_count + 1) : NULL;
if (!pv_rank) {
scratch_free(map_hdr); scratch_free(ix_ht_hdr); scratch_free(ix_first_hdr);
scratch_free(ix_hdr); scratch_free(pv_first_hdr); scratch_free(pv_hdr);
pivot_ingest_free(&pg); ght_layout_free(&ly); scratch_free(key_hdr);
return ray_error("oom", NULL);
}
uint32_t* ix_rank = pv_rank + pv_count;
for (uint32_t p = 0; p < pv_count; p++) pv_rank[p] = p;
for (uint32_t i = 1; i < pv_count; i++) {
uint32_t v = pv_rank[i]; uint32_t j = i;
while (j > 0 && pv_first[pv_rank[j - 1]] > pv_first[v]) { pv_rank[j] = pv_rank[j - 1]; j--; }
pv_rank[j] = v;
}
for (uint32_t r = 0; r < pv_count; r++) pv_tmp[r] = pv_vals[pv_rank[r]];
memcpy(pv_vals, pv_tmp, (size_t)pv_count * sizeof(int64_t));
for (uint32_t r = 0; r < pv_count; r++) pv_tmp[pv_rank[r]] = r;
for (uint32_t p = 0; p < pv_count; p++) pv_rank[p] = (uint32_t)pv_tmp[p];
bool ranked = false;
if ((uint64_t)ix_count * 16 >= (uint64_t)nrows && nrows > 0) {
ray_t* pos_hdr = NULL;
uint32_t* pos = (uint32_t*)scratch_alloc(&pos_hdr, (size_t)nrows * sizeof(uint32_t));
if (pos) {
memset(pos, 0xFF, (size_t)nrows * sizeof(uint32_t));
bool ok = true;
for (uint32_t i = 0; i < ix_count && ok; i++) {
int64_t f = ix_first[i];
if (f < 0 || f >= nrows || pos[f] != UINT32_MAX) ok = false; else pos[f] = i;
}
if (ok) {
uint32_t next = 0;
for (int64_t r = 0; r < nrows; r++)
if (pos[r] != UINT32_MAX) ix_rank[pos[r]] = next++;
ranked = (next == ix_count);
}
scratch_free(pos_hdr);
}
}
if (!ranked && ix_count > 0) {
ray_t* tmp_hdr = NULL;
uint32_t* buf = (uint32_t*)scratch_alloc(&tmp_hdr, 2 * (size_t)ix_count * sizeof(uint32_t));
if (!buf) {
scratch_free(rank_hdr); scratch_free(map_hdr); scratch_free(ix_ht_hdr); scratch_free(ix_first_hdr);
scratch_free(ix_hdr); scratch_free(pv_first_hdr); scratch_free(pv_hdr);
pivot_ingest_free(&pg); ght_layout_free(&ly); scratch_free(key_hdr);
return ray_error("oom", NULL);
}
uint32_t* src = buf; uint32_t* dst = buf + ix_count;
for (uint32_t i = 0; i < ix_count; i++) src[i] = i;
for (uint32_t w = 1; w < ix_count; w *= 2) {
for (uint32_t lo = 0; lo < ix_count; lo += 2 * w) {
uint32_t mid = lo + w < ix_count ? lo + w : ix_count;
uint32_t hi = lo + 2 * w < ix_count ? lo + 2 * w : ix_count;
uint32_t a = lo, b = mid, o = lo;
while (a < mid && b < hi) dst[o++] = (ix_first[src[b]] < ix_first[src[a]]) ? src[b++] : src[a++];
while (a < mid) dst[o++] = src[a++];
while (b < hi) dst[o++] = src[b++];
}
uint32_t* t = src; src = dst; dst = t;
}
for (uint32_t r = 0; r < ix_count; r++) ix_rank[src[r]] = r;
scratch_free(tmp_hdr);
}
if (ix_count > 0) {
ray_t* perm_hdr = NULL;
char* perm = (char*)scratch_alloc(&perm_hdr, (size_t)ix_count * ix_entry);
if (!perm) {
scratch_free(rank_hdr); scratch_free(map_hdr); scratch_free(ix_ht_hdr); scratch_free(ix_first_hdr);
scratch_free(ix_hdr); scratch_free(pv_first_hdr); scratch_free(pv_hdr);
pivot_ingest_free(&pg); ght_layout_free(&ly); scratch_free(key_hdr);
return ray_error("oom", NULL);
}
for (uint32_t i = 0; i < ix_count; i++)
memcpy(perm + (size_t)ix_rank[i] * ix_entry, ix_rows + (size_t)i * ix_entry, ix_entry);
memcpy(ix_rows, perm, (size_t)ix_count * ix_entry);
scratch_free(perm_hdr);
}
for (uint32_t gi = 0; gi < grp_count; gi++) {
if (grp_ix[gi] != UINT32_MAX) grp_ix[gi] = ix_rank[grp_ix[gi]];
if (grp_pv[gi] != UINT32_MAX) grp_pv[gi] = pv_rank[grp_pv[gi]];
}
scratch_free(rank_hdr);
}
scratch_free(ix_first_hdr);
scratch_free(pv_first_hdr);
ray_progress_update("pivot", "scatter", 0, (uint64_t)pv_count);
bool val_is_f64 = vcol->type == RAY_F64;
int8_t out_agg_type;
switch (agg_op) {
case OP_AVG: out_agg_type = RAY_F64; break;
case OP_COUNT: out_agg_type = RAY_I64; break;
case OP_SUM: out_agg_type = val_is_f64 ? RAY_F64 : RAY_I64; break;
default: out_agg_type = vcol->type; break;
}
int64_t out_ncols = (int64_t)n_idx + (int64_t)pv_count;
ray_t* result = ray_table_new(out_ncols);
if (!result || RAY_IS_ERR(result)) goto pivot_cleanup;
for (uint32_t k = 0; k < n_idx; k++) {
ray_t* new_col = col_vec_new(idx_vecs[k], (int64_t)ix_count);
if (!new_col || RAY_IS_ERR(new_col)) { ray_release(result); result = ray_error("oom", NULL); goto pivot_cleanup; }
new_col->len = (int64_t)ix_count;
uint8_t esz = col_esz(idx_vecs[k]);
int8_t kt = idx_vecs[k]->type;
const char* src_base = idx_wide[k] ? (const char*)key_data[k] : NULL;
const uint32_t null_kw = k >> 6;
const int64_t null_kbit = (int64_t)((uint64_t)1 << (k & 63));
for (uint32_t r = 0; r < ix_count; r++) {
const char* ix_entry_p = ix_rows + r * ix_entry;
int64_t kv = ((const int64_t*)(ix_entry_p + 8))[k];
const int64_t* ent_nwords = (const int64_t*)(ix_entry_p + 8 + (size_t)n_idx * 8);
if (ent_nwords[null_kw] & null_kbit) {
ray_vec_set_null(new_col, (int64_t)r, true);
switch (kt) {
case RAY_F64:
((double*)ray_data(new_col))[r] = NULL_F64; break;
case RAY_I64: case RAY_TIMESTAMP:
((int64_t*)ray_data(new_col))[r] = NULL_I64; break;
case RAY_I32: case RAY_DATE: case RAY_TIME:
((int32_t*)ray_data(new_col))[r] = NULL_I32; break;
case RAY_I16:
((int16_t*)ray_data(new_col))[r] = NULL_I16; break;
default: break;
}
continue;
}
if (idx_wide[k]) {
memcpy((char*)ray_data(new_col) + (size_t)r * esz,
src_base + (size_t)kv * 16, 16);
} else if (kt == RAY_F64) {
memcpy((char*)ray_data(new_col) + (size_t)r * esz, &kv, 8);
} else {
write_col_i64(ray_data(new_col), (int64_t)r, kv, kt, new_col->attrs);
}
}
if (idx_vecs[k]->type == RAY_STR)
col_propagate_str_pool(new_col, idx_vecs[k]);
if (new_col->type == RAY_SYM)
ray_sym_vec_adopt_domain(new_col, idx_vecs[k]);
ray_op_ext_t* ie = find_ext(g, ext->pivot.index_cols[k]);
result = ray_table_add_col(result, ie->sym, new_col);
ray_release(new_col);
if (RAY_IS_ERR(result)) goto pivot_cleanup;
}
{
int8_t s = ly.agg_val_slot[0];
for (uint32_t p = 0; p < pv_count; p++) {
ray_t* new_col = (out_agg_type == vcol->type)
? col_vec_new(vcol, (int64_t)ix_count)
: ray_vec_new(out_agg_type, (int64_t)ix_count);
if (!new_col || RAY_IS_ERR(new_col)) { ray_release(result); result = ray_error("oom", NULL); goto pivot_cleanup; }
new_col->len = (int64_t)ix_count;
switch (new_col->type) {
case RAY_F64: {
double* d = (double*)ray_data(new_col);
for (int64_t r = 0; r < (int64_t)ix_count; r++) d[r] = NULL_F64;
break;
}
case RAY_F32: {
float* d = (float*)ray_data(new_col);
for (int64_t r = 0; r < (int64_t)ix_count; r++) d[r] = NULL_F32;
break;
}
case RAY_I64: case RAY_TIMESTAMP: {
int64_t* d = (int64_t*)ray_data(new_col);
for (int64_t r = 0; r < (int64_t)ix_count; r++) d[r] = NULL_I64;
break;
}
case RAY_I32: case RAY_DATE: case RAY_TIME: {
int32_t* d = (int32_t*)ray_data(new_col);
for (int64_t r = 0; r < (int64_t)ix_count; r++) d[r] = NULL_I32;
break;
}
case RAY_I16: {
int16_t* d = (int16_t*)ray_data(new_col);
for (int64_t r = 0; r < (int64_t)ix_count; r++) d[r] = NULL_I16;
break;
}
default:
memset(ray_data(new_col), 0, (size_t)ix_count * (size_t)col_esz(new_col));
break;
}
for (uint32_t _pp = 0; _pp < pg.n_parts; _pp++) {
group_ht_t* ph = &pg.part_hts[_pp];
uint32_t pcount = ph->grp_count;
uint32_t gi_base = pg.part_offsets[_pp];
for (uint32_t gi_local = 0; gi_local < pcount; gi_local++) {
uint32_t gi = gi_base + gi_local;
if (grp_pv[gi] != p) continue;
uint32_t r = grp_ix[gi];
const char* row = ph->rows + (size_t)gi_local * pg.row_stride;
int64_t cnt = *(const int64_t*)(const void*)row;
int64_t nn = (ly.off_nn && s >= 0)
? ((const int64_t*)(const void*)(row + ly.off_nn))[s] : cnt;
if (out_agg_type == RAY_F64) {
double v;
switch (agg_op) {
case OP_SUM:
v = val_is_f64 ? ROW_RD_F64(row, ly.off_sum, s)
: (double)ROW_RD_I64(row, ly.off_sum, s);
break;
case OP_AVG:
if (nn == 0) { v = NULL_F64; ray_vec_set_null(new_col, (int64_t)r, true); break; }
v = val_is_f64 ? ROW_RD_F64(row, ly.off_sum, s) / nn
: (ly.need_flags & GHT_NEED_SUM128)
? ray_i128_to_f64(ROW_RD_I64(row, ly.off_sum_hi, s),
(uint64_t)ROW_RD_I64(row, ly.off_sum, s)) / nn
: (double)ROW_RD_I64(row, ly.off_sum, s) / nn;
break;
case OP_MIN:
if (nn == 0) { v = NULL_F64; ray_vec_set_null(new_col, (int64_t)r, true); break; }
v = val_is_f64 ? ROW_RD_F64(row, ly.off_min, s)
: (double)ROW_RD_I64(row, ly.off_min, s);
break;
case OP_MAX:
if (nn == 0) { v = NULL_F64; ray_vec_set_null(new_col, (int64_t)r, true); break; }
v = val_is_f64 ? ROW_RD_F64(row, ly.off_max, s)
: (double)ROW_RD_I64(row, ly.off_max, s);
break;
case OP_FIRST: case OP_LAST:
if (nn == 0) { v = NULL_F64; ray_vec_set_null(new_col, (int64_t)r, true); break; }
v = val_is_f64 ? ROW_RD_F64(row, ly.off_sum, s)
: (double)ROW_RD_I64(row, ly.off_sum, s);
break;
default: v = 0.0; break;
}
((double*)ray_data(new_col))[r] = ray_f64_fin(v);
} else {
int64_t v;
int64_t int_null = agg_int_null_sentinel_for(out_agg_type);
switch (agg_op) {
case OP_SUM: v = ROW_RD_I64(row, ly.off_sum, s); break;
case OP_COUNT: v = cnt; break;
case OP_MIN:
if (nn == 0) { v = int_null; ray_vec_set_null(new_col, (int64_t)r, true); break; }
v = ROW_RD_I64(row, ly.off_min, s); break;
case OP_MAX:
if (nn == 0) { v = int_null; ray_vec_set_null(new_col, (int64_t)r, true); break; }
v = ROW_RD_I64(row, ly.off_max, s); break;
case OP_FIRST: case OP_LAST:
if (nn == 0) { v = int_null; ray_vec_set_null(new_col, (int64_t)r, true); break; }
v = ROW_RD_I64(row, ly.off_sum, s); break;
default: v = 0; break;
}
write_col_i64(ray_data(new_col), (int64_t)r, v, out_agg_type, new_col->attrs);
}
}
}
if (new_col->type == RAY_SYM)
ray_sym_vec_adopt_domain(new_col, vcol);
int64_t pval = pv_vals[p];
int64_t col_sym;
if (pcol->type == RAY_SYM) {
col_sym = sym_id_runtime(pcol, pval);
} else if (pvt_wide) {
static const char hex[] = "0123456789abcdef";
static const int groups[] = {4, 2, 2, 2, 6};
char buf[37];
const uint8_t* bytes = (const uint8_t*)pvt_base + (size_t)pval * 16;
int pos = 0, bpos = 0;
for (int g = 0; g < 5; g++) {
if (g > 0) buf[bpos++] = '-';
for (int j = 0; j < groups[g]; j++) {
buf[bpos++] = hex[bytes[pos] >> 4];
buf[bpos++] = hex[bytes[pos] & 0x0F];
pos++;
}
}
col_sym = ray_sym_intern(buf, (size_t)bpos);
} else {
char buf[128];
int len = 0;
int8_t pt = key_types[n_idx];
if (pt == RAY_F64) {
double fv;
memcpy(&fv, &pval, 8);
fv = clear_neg_zero(fv);
len = snprintf(buf, sizeof(buf), "%g", fv);
} else if (pt == RAY_BOOL) {
len = snprintf(buf, sizeof(buf), "%s", pval ? "true" : "false");
} else if (pt == RAY_I64 || pt == RAY_I32 || pt == RAY_I16 ||
pt == RAY_DATE || pt == RAY_TIME || pt == RAY_TIMESTAMP) {
len = snprintf(buf, sizeof(buf), "%" PRId64, pval);
} else {
len = snprintf(buf, sizeof(buf), "col%" PRId64, pval);
}
col_sym = ray_sym_intern(buf, (size_t)len);
}
par_finalize_nulls(new_col);
result = ray_table_add_col(result, col_sym, new_col);
ray_release(new_col);
if (RAY_IS_ERR(result)) goto pivot_cleanup;
}
}
pivot_cleanup:
ght_layout_free(&ly);
scratch_free(map_hdr);
scratch_free(ix_ht_hdr);
scratch_free(ix_hdr);
scratch_free(pv_hdr);
scratch_free(key_hdr);
pivot_ingest_free(&pg);
return result;
}