#include "ops/internal.h"
#include "lang/internal.h"
#include "lang/format.h"
#include "ops/rowsel.h"
#include "table/table.h"
#include "ops/colbind.h"
#include "ops/idxop.h"
#include "store/enccol.h"
#include "store/enceval.h"
#include <stdio.h>
#include <stdlib.h>
static inline uint8_t truthy_intish(double v) {
return v != 0.0 ? 1 : 0;
}
static inline uint8_t truthy_f64ish(double v) {
return (v == v && v != 0.0) ? 1 : 0;
}
static inline bool floor_idiv_i64_checked(int64_t a, int64_t b, int64_t* out) {
if (b == 0 || (a == INT64_MIN && b == -1)) return false;
int64_t q = a / b;
int64_t rem = a % b;
if (rem != 0 && ((a < 0) != (b < 0))) q--;
*out = q;
return true;
}
#define DBL_EXACT_INT_LIM (0x1p53)
static inline bool i64_dbl_exact(int64_t v) {
return v <= (int64_t)DBL_EXACT_INT_LIM && v >= -(int64_t)DBL_EXACT_INT_LIM;
}
static inline bool i64_span_dbl_exact(const int64_t* v, int64_t n) {
int bad = 0;
for (int64_t i = 0; i < n; i++) bad |= !i64_dbl_exact(v[i]);
return !bad;
}
static inline bool i64_span2_dbl_exact(const int64_t* a, const int64_t* b, int64_t n) {
int bad = 0;
for (int64_t i = 0; i < n; i++) {
bad |= !i64_dbl_exact(a[i]);
bad |= !i64_dbl_exact(b[i]);
}
return !bad;
}
static bool atom_to_numeric(ray_t* atom, double* out_f, int64_t* out_i, bool* out_is_f64) {
if (!atom || !ray_is_atom(atom)) return false;
switch (atom->type) {
case -RAY_F64:
*out_f = atom->f64;
*out_i = ray_cast_f64_to_i64_null(atom->f64);
*out_is_f64 = true;
return true;
case -RAY_F32:
*out_f = (double)(float)atom->f64;
*out_i = ray_cast_f64_to_i64_null((double)(float)atom->f64);
*out_is_f64 = true;
return true;
case -RAY_I64:
case -RAY_SYM:
case -RAY_DATE:
case -RAY_TIME:
case -RAY_TIMESTAMP:
*out_i = atom->i64;
*out_f = (double)atom->i64;
*out_is_f64 = false;
return true;
case -RAY_I32:
*out_i = (int64_t)atom->i32;
*out_f = (double)atom->i32;
*out_is_f64 = false;
return true;
case -RAY_I16:
*out_i = (int64_t)atom->i16;
*out_f = (double)atom->i16;
*out_is_f64 = false;
return true;
case -RAY_U8:
case -RAY_BOOL:
*out_i = (int64_t)atom->u8;
*out_f = (double)atom->u8;
*out_is_f64 = false;
return true;
default:
return false;
}
}
static bool eval_const_numeric_expr(ray_graph_t* g, ray_op_t* op,
double* out_f, int64_t* out_i, bool* out_is_f64) {
if (!g || !op || !out_f || !out_i || !out_is_f64) return false;
if (op->opcode == OP_CONST) {
ray_op_ext_t* ext = find_ext(g, op->id);
if (!ext || !ext->literal) return false;
if (RAY_ATOM_IS_NULL(ext->literal)) return false;
return atom_to_numeric(ext->literal, out_f, out_i, out_is_f64);
}
if (op->arity == 1 && op_child(g, op, 0)) {
double af = 0.0;
int64_t ai = 0;
bool a_is_f64 = false;
if (!eval_const_numeric_expr(g, op_child(g, op, 0), &af, &ai, &a_is_f64)) return false;
if (op->opcode == OP_NEG || op->opcode == OP_ABS) {
if (a_is_f64 || op->out_type == RAY_F64) {
double v = a_is_f64 ? af : (double)ai;
double r = (op->opcode == OP_NEG) ? -v : fabs(v);
*out_f = r;
*out_i = (int64_t)r;
*out_is_f64 = true;
return true;
}
int64_t v = ai;
int64_t r = (op->opcode == OP_NEG)
? (int64_t)(-(uint64_t)v)
: (v < 0 ? (int64_t)(-(uint64_t)v) : v);
*out_i = r;
*out_f = (double)r;
*out_is_f64 = false;
return true;
}
double v = a_is_f64 ? af : (double)ai;
double r = 0.0;
switch (op->opcode) {
case OP_SQRT: r = sqrt(v); break;
case OP_LOG: r = log(v); break;
case OP_EXP: r = exp(v); break;
case OP_SIN: r = sin(v); break;
case OP_ASIN: r = asin(v); break;
case OP_COS: r = cos(v); break;
case OP_ACOS: r = acos(v); break;
case OP_TAN: r = tan(v); break;
case OP_ATAN: r = atan(v); break;
case OP_RECIPROCAL: r = v != 0.0 ? 1.0 / v : NAN; break;
case OP_SIGNUM:
*out_i = (v > 0.0) - (v < 0.0);
*out_f = (double)*out_i;
*out_is_f64 = false;
return true;
default:
return false;
}
r = ray_f64_fin(r);
*out_f = r;
*out_i = ray_cast_f64_to_i64_null(r);
*out_is_f64 = true;
return true;
}
if (op->arity != 2 || !op_child(g, op, 0) || !op_child(g, op, 1)) return false;
if ((op->opcode < OP_ADD || op->opcode > OP_MAX2) &&
op->opcode != OP_IDIV && op->opcode != OP_POW) return false;
double lf = 0.0, rf = 0.0;
int64_t li = 0, ri = 0;
bool l_is_f64 = false, r_is_f64 = false;
if (!eval_const_numeric_expr(g, op_child(g, op, 0), &lf, &li, &l_is_f64)) return false;
if (!eval_const_numeric_expr(g, op_child(g, op, 1), &rf, &ri, &r_is_f64)) return false;
if (op->out_type == RAY_F64 || l_is_f64 || r_is_f64 ||
op->opcode == OP_DIV || op->opcode == OP_POW) {
double lv = l_is_f64 ? lf : (double)li;
double rv = r_is_f64 ? rf : (double)ri;
double r = 0.0;
switch (op->opcode) {
case OP_ADD: r = lv + rv; break;
case OP_SUB: r = lv - rv; break;
case OP_MUL: r = lv * rv; break;
case OP_DIV: r = rv != 0.0 ? lv / rv : NAN; break;
case OP_IDIV: r = rv != 0.0 ? floor(lv / rv) : NAN; break;
case OP_MOD: { if (rv != 0.0) { r = fmod(lv, rv); if (r && ((r > 0) != (rv > 0))) r += rv; } else { r = NAN; } } break;
case OP_POW: r = (lv != lv || rv != rv) ? NAN : pow(lv, rv); break;
case OP_MIN2: r = lv < rv ? lv : rv; break;
case OP_MAX2: r = lv > rv ? lv : rv; break;
default: return false;
}
r = ray_f64_fin(r);
*out_f = r;
*out_i = ray_cast_f64_to_i64_null(r);
*out_is_f64 = true;
return true;
}
int64_t r = 0;
switch (op->opcode) {
case OP_ADD: r = (int64_t)((uint64_t)li + (uint64_t)ri); break;
case OP_SUB: r = (int64_t)((uint64_t)li - (uint64_t)ri); break;
case OP_MUL: r = (int64_t)((uint64_t)li * (uint64_t)ri); break;
case OP_DIV:
if (!floor_idiv_i64_checked(li, ri, &r)) return false;
break;
case OP_IDIV:
if (!floor_idiv_i64_checked(li, ri, &r)) return false;
break;
case OP_MOD:
if (ri==0) return false;
r = li%ri; if (r && (r^ri)<0) r+=ri;
break;
case OP_MIN2: r = li < ri ? li : ri; break;
case OP_MAX2: r = li > ri ? li : ri; break;
default: return false;
}
*out_i = r;
*out_f = (double)r;
*out_is_f64 = false;
return true;
}
static bool const_expr_to_i64(ray_graph_t* g, ray_op_t* op, int64_t* out) {
if (!g || !op || !out) return false;
double c_f = 0.0;
int64_t c_i = 0;
bool c_is_f64 = false;
if (!eval_const_numeric_expr(g, op, &c_f, &c_i, &c_is_f64)) return false;
if (c_is_f64) return false;
*out = c_i;
return true;
}
static inline bool type_is_linear_i64_col(int8_t t) {
return t == RAY_I64 || t == RAY_TIMESTAMP ||
t == RAY_I32 || t == RAY_DATE || t == RAY_TIME || t == RAY_I16 ||
t == RAY_U8 || t == RAY_BOOL || RAY_IS_SYM(t);
}
static bool linear_expr_add_term(linear_expr_i64_t* e, int64_t sym, int64_t coeff) {
if (!e) return false;
if (coeff == 0) return true;
for (uint8_t i = 0; i < e->n_terms; i++) {
if (e->syms[i] != sym) continue;
int64_t next = e->coeff_i64[i] + coeff;
if (next != 0) {
e->coeff_i64[i] = next;
return true;
}
for (uint8_t j = i + 1; j < e->n_terms; j++) {
e->syms[j - 1] = e->syms[j];
e->coeff_i64[j - 1] = e->coeff_i64[j];
}
e->n_terms--;
return true;
}
if (e->n_terms >= AGG_LINEAR_MAX_TERMS) return false;
e->syms[e->n_terms] = sym;
e->coeff_i64[e->n_terms] = coeff;
e->n_terms++;
return true;
}
static void linear_expr_scale(linear_expr_i64_t* e, int64_t k) {
if (!e || k == 1) return;
e->bias_i64 *= k;
for (uint8_t i = 0; i < e->n_terms; i++)
e->coeff_i64[i] *= k;
}
static bool linear_expr_add_scaled(linear_expr_i64_t* dst, const linear_expr_i64_t* src, int64_t scale) {
if (!dst || !src) return false;
dst->bias_i64 += src->bias_i64 * scale;
for (uint8_t i = 0; i < src->n_terms; i++) {
if (!linear_expr_add_term(dst, src->syms[i], src->coeff_i64[i] * scale))
return false;
}
return true;
}
static bool parse_linear_i64_expr(ray_graph_t* g, ray_op_t* op, linear_expr_i64_t* out) {
if (!g || !op || !out) return false;
memset(out, 0, sizeof(*out));
int64_t c = 0;
if (const_expr_to_i64(g, op, &c)) {
out->bias_i64 = c;
return true;
}
if (op->opcode == OP_SCAN) {
ray_op_ext_t* ext = find_ext(g, op->id);
if (!ext || ext->base.opcode != OP_SCAN) return false;
out->n_terms = 1;
out->syms[0] = ext->sym;
out->coeff_i64[0] = 1;
return true;
}
if (op->opcode == OP_NEG && op->arity == 1 && op_child(g, op, 0)) {
linear_expr_i64_t inner;
if (!parse_linear_i64_expr(g, op_child(g, op, 0), &inner)) return false;
linear_expr_scale(&inner, -1);
*out = inner;
return true;
}
if ((op->opcode == OP_ADD || op->opcode == OP_SUB) &&
op->arity == 2 && op_child(g, op, 0) && op_child(g, op, 1)) {
linear_expr_i64_t lhs;
linear_expr_i64_t rhs;
if (!parse_linear_i64_expr(g, op_child(g, op, 0), &lhs)) return false;
if (!parse_linear_i64_expr(g, op_child(g, op, 1), &rhs)) return false;
*out = lhs;
return linear_expr_add_scaled(out, &rhs, op->opcode == OP_ADD ? 1 : -1);
}
if (op->opcode == OP_MUL && op->arity == 2 && op_child(g, op, 0) && op_child(g, op, 1)) {
int64_t k = 0;
linear_expr_i64_t side;
if (const_expr_to_i64(g, op_child(g, op, 0), &k) &&
parse_linear_i64_expr(g, op_child(g, op, 1), &side)) {
linear_expr_scale(&side, k);
*out = side;
return true;
}
if (const_expr_to_i64(g, op_child(g, op, 1), &k) &&
parse_linear_i64_expr(g, op_child(g, op, 0), &side)) {
linear_expr_scale(&side, k);
*out = side;
return true;
}
}
return false;
}
bool try_linear_sumavg_input_i64(ray_graph_t* g, ray_t* tbl, ray_op_t* input_op,
agg_linear_t* out_plan, ray_t** err) {
*err = NULL;
if (!g || !tbl || !input_op || !out_plan) return false;
linear_expr_i64_t lin;
if (!parse_linear_i64_expr(g, input_op, &lin)) return false;
memset(out_plan, 0, sizeof(*out_plan));
out_plan->n_terms = lin.n_terms;
out_plan->bias_i64 = lin.bias_i64;
ray_col_src_t* srcs[AGG_LINEAR_MAX_TERMS];
for (uint8_t i = 0; i < lin.n_terms; i++) {
if (ray_col_bind_declines(tbl, lin.syms[i])) return false;
srcs[i] = ray_graph_bind_col(g, tbl, lin.syms[i], err);
if (!srcs[i]) return false;
if (srcs[i]->parted || !type_is_linear_i64_col(srcs[i]->type)) return false;
if (srcs[i]->may_null) return false;
if (!ray_col_src_base(srcs[i])) return false;
}
for (uint8_t i = 0; i < lin.n_terms; i++) {
ray_t* col = NULL;
*err = ray_col_src_plain(srcs[i], &col);
if (*err) return false;
out_plan->term_ptrs[i] = ray_data(col);
out_plan->term_types[i] = srcs[i]->type;
out_plan->coeff_i64[i] = lin.coeff_i64[i];
}
out_plan->enabled = true;
return true;
}
bool try_prod_sumavg_input_f64(ray_graph_t* g, ray_t* tbl, ray_op_t* input_op,
agg_prod_t* out, ray_t** err) {
*err = NULL;
if (!g || !tbl || !input_op || !out) return false;
if (input_op->opcode != OP_MUL || input_op->arity != 2) return false;
ray_op_t* lhs = op_child(g, input_op, 0);
ray_op_t* rhs = op_child(g, input_op, 1);
if (!lhs || !rhs) return false;
if (lhs->opcode != OP_SCAN || rhs->opcode != OP_SCAN) return false;
ray_op_ext_t* lx = find_ext(g, lhs->id);
ray_op_ext_t* rx = find_ext(g, rhs->id);
if (!lx || !rx) return false;
if (ray_col_bind_declines(tbl, lx->sym) || ray_col_bind_declines(tbl, rx->sym))
return false;
ray_col_src_t* sa = ray_graph_bind_col(g, tbl, lx->sym, err);
if (!sa) return false;
ray_col_src_t* sb = ray_graph_bind_col(g, tbl, rx->sym, err);
if (!sb) return false;
if (sa->parted || sb->parted) return false;
if (sa->may_null || sb->may_null)
return false;
if (!ray_col_src_base(sa) || !ray_col_src_base(sb)) return false;
#define PROD_OK_T(t) ((t) == RAY_F64 || (t) == RAY_I64 || (t) == RAY_I32 || \
(t) == RAY_I16 || (t) == RAY_U8 || (t) == RAY_BOOL)
if (!PROD_OK_T(sa->type) || !PROD_OK_T(sb->type)) return false;
#undef PROD_OK_T
if (sa->type != RAY_F64 && sb->type != RAY_F64) return false;
ray_t* ca = NULL;
ray_t* cb = NULL;
*err = ray_col_src_plain(sa, &ca);
if (*err) return false;
*err = ray_col_src_plain(sb, &cb);
if (*err) return false;
out->enabled = true;
out->pa = ray_data(ca); out->ta = sa->type; out->aa = sa->attrs;
out->pb = ray_data(cb); out->tb = sb->type; out->ab = sb->attrs;
return true;
}
static ray_t* affine_base_obj(ray_col_src_t* src) {
ray_t* ready = ray_col_src_ready(src);
return ready ? ready : src->obj;
}
static bool affine_int_f64_bias_exact(int8_t bt, double bias, int64_t nrows,
bool may_null) {
if (may_null || nrows <= 0 || !isfinite(bias) || trunc(bias) != bias)
return false;
long double max_abs;
switch (bt) {
case RAY_I32: max_abs = 2147483648.0L; break;
case RAY_I16: max_abs = 32768.0L; break;
case RAY_U8: max_abs = 255.0L; break;
case RAY_BOOL: max_abs = 1.0L; break;
default: return false;
}
return ((max_abs + fabsl((long double)bias)) * (long double)nrows) <=
9007199254740992.0L;
}
bool try_affine_sumavg_input(ray_graph_t* g, ray_t* tbl, ray_op_t* input_op,
bool f64_bias_ok,
ray_t** out_vec, agg_affine_t* out_affine, ray_t** err) {
*err = NULL;
if (!g || !tbl || !input_op || !out_vec || !out_affine) return false;
if (input_op->opcode != OP_ADD && input_op->opcode != OP_SUB) return false;
if (input_op->arity != 2 || !op_child(g, input_op, 0) || !op_child(g, input_op, 1)) return false;
ray_op_t* lhs = op_child(g, input_op, 0);
ray_op_t* rhs = op_child(g, input_op, 1);
ray_op_t* base_op = NULL;
int sign = 1;
double c_f = 0.0;
int64_t c_i = 0;
bool c_is_f64 = false;
double lhs_f = 0.0, rhs_f = 0.0;
int64_t lhs_i = 0, rhs_i = 0;
bool lhs_is_f64 = false, rhs_is_f64 = false;
bool lhs_const = eval_const_numeric_expr(g, lhs, &lhs_f, &lhs_i, &lhs_is_f64);
bool rhs_const = eval_const_numeric_expr(g, rhs, &rhs_f, &rhs_i, &rhs_is_f64);
if (input_op->opcode == OP_ADD) {
if (lhs_const) {
base_op = rhs;
sign = 1;
c_f = lhs_f;
c_i = lhs_i;
c_is_f64 = lhs_is_f64;
} else if (rhs_const) {
base_op = lhs;
sign = 1;
c_f = rhs_f;
c_i = rhs_i;
c_is_f64 = rhs_is_f64;
}
} else {
if (rhs_const) {
base_op = lhs;
sign = -1;
c_f = rhs_f;
c_i = rhs_i;
c_is_f64 = rhs_is_f64;
}
}
if (!base_op) return false;
ray_op_ext_t* base_ext = find_ext(g, base_op->id);
if (!base_ext || base_ext->base.opcode != OP_SCAN) return false;
if (ray_col_bind_declines(tbl, base_ext->sym)) return false;
ray_col_src_t* base_src = ray_graph_bind_col(g, tbl, base_ext->sym, err);
if (!base_src) return false;
if (base_src->parted) return false;
int8_t bt = base_src->type;
ray_t* base_vec = NULL;
if (bt == RAY_F64) {
base_vec = affine_base_obj(base_src);
out_affine->enabled = true;
out_affine->bias_f64 = (double)sign * (c_is_f64 ? c_f : (double)c_i);
out_affine->bias_i64 = (int64_t)out_affine->bias_f64;
*out_vec = base_vec;
return true;
}
if (bt == RAY_I64 || bt == RAY_TIMESTAMP ||
bt == RAY_I32 || bt == RAY_I16 || bt == RAY_U8 || bt == RAY_BOOL ||
RAY_IS_SYM(bt)) {
if (c_is_f64) {
if (!f64_bias_ok || !affine_int_f64_bias_exact(bt, c_f,
base_src->len, base_src->may_null)) return false;
base_vec = affine_base_obj(base_src);
out_affine->enabled = true;
out_affine->f64 = true;
out_affine->bias_f64 = (double)sign * c_f;
out_affine->bias_i64 = 0;
*out_vec = base_vec;
return true;
}
int64_t c = c_i;
base_vec = affine_base_obj(base_src);
out_affine->enabled = true;
out_affine->bias_i64 = sign > 0 ? c : -c;
out_affine->bias_f64 = (double)out_affine->bias_i64;
*out_vec = base_vec;
return true;
}
return false;
}
uint64_t ray_expr_bail_counts[EXPR_BAIL__N];
uint64_t ray_expr_compile_ok;
bool ray_expr_disable;
#define EXPR_BAIL(reason) do { \
ray_expr_bail_counts[(reason)]++; \
return false; \
} while (0)
static void expr_stats_dump(void) {
static const char* names[EXPR_BAIL__N] = {
"root-shape", "graph-size", "depth", "regs", "ins",
"mapcommon", "str", "guid", "nulls", "slice", "sym-domain",
"const", "null-shape", "other",
};
fprintf(stderr, "expr_compile ok=%llu\n",
(unsigned long long)ray_expr_compile_ok);
for (int i = 0; i < EXPR_BAIL__N; i++)
if (ray_expr_bail_counts[i])
fprintf(stderr, "expr_compile bail %-10s %llu\n", names[i],
(unsigned long long)ray_expr_bail_counts[i]);
}
void ray_expr_stats_init(void) {
if (getenv("RAY_EXPR_STATS")) atexit(expr_stats_dump);
}
static inline bool expr_is_f64_unary_math(uint16_t op) {
return op == OP_SQRT || op == OP_LOG || op == OP_EXP ||
(op >= OP_SIN && op <= OP_RECIPROCAL);
}
static inline bool expr_is_elementwise(uint16_t op) {
return (op >= OP_NEG && op <= OP_CAST) ||
(op >= OP_ADD && op <= OP_MAX2) ||
op == OP_POW || (op >= OP_SIN && op <= OP_SIGNUM);
}
static uint8_t expr_ensure_type(ray_expr_t* out, uint8_t src, int8_t target) {
if (out->regs[src].type == target) return src;
if (out->n_regs >= EXPR_MAX_REGS || out->n_ins >= EXPR_MAX_INS) return src;
uint8_t r = out->n_regs;
out->regs[r].kind = REG_SCRATCH;
out->regs[r].type = target;
out->regs[r].nullable = out->regs[src].nullable;
out->regs[r].null_src = out->regs[src].null_src;
out->n_regs++;
out->n_scratch++;
out->ins[out->n_ins++] = (expr_ins_t){
.opcode = OP_CAST, .dst = r, .src1 = src, .src2 = 0xFF,
.null_aware = out->regs[src].nullable ? 1 : 0,
};
return r;
}
static bool expr_op_generates_null(uint16_t op, int8_t ot, int8_t t1, bool binary) {
if (ot == RAY_F64) {
switch (op) {
case OP_ADD: case OP_SUB: case OP_MUL:
case OP_DIV: case OP_IDIV: case OP_MOD: case OP_POW:
case OP_SQRT: case OP_LOG: case OP_EXP:
case OP_SIN: case OP_ASIN: case OP_COS: case OP_ACOS:
case OP_TAN: case OP_ATAN: case OP_RECIPROCAL:
return true;
default:
return false;
}
}
if (ot == RAY_I64) {
if (binary) return op == OP_DIV || op == OP_IDIV || op == OP_MOD;
return (op == OP_NEG || op == OP_ABS) && t1 == RAY_I64;
}
return false;
}
static inline int8_t expr_cast_lane_type(int8_t t) {
if (t == RAY_TIMESTAMP) return RAY_I64;
if (t == RAY_DATE || t == RAY_TIME) return RAY_I32;
return t;
}
static bool expr_null_capable(uint8_t op, int8_t dt, int8_t t1) {
if (op == OP_CAST) {
dt = expr_cast_lane_type(dt);
t1 = expr_cast_lane_type(t1);
if (dt == RAY_F64 && t1 != RAY_F64) return true;
if (dt == RAY_I64 && t1 == RAY_F64) return true;
if (dt == RAY_I64 && t1 != RAY_F64) return true;
if (dt == RAY_BOOL) return true;
if (dt == RAY_I32 || dt == RAY_I16) return true;
}
if (dt == RAY_BOOL && t1 == RAY_I64 &&
((op >= OP_EQ && op <= OP_GE) || op == OP_AND || op == OP_OR))
return true;
if (op == OP_ISNULL) return true;
if (dt == RAY_I64 && t1 != RAY_F64 &&
((op >= OP_ADD && op <= OP_MOD) || op == OP_MIN2 || op == OP_MAX2 ||
op == OP_IDIV || op == OP_NEG || op == OP_ABS || op == OP_SIGNUM))
return true;
if (dt == RAY_I64 && t1 == RAY_F64 && op == OP_SIGNUM)
return true;
if (dt == RAY_F64 && (op == OP_MIN2 || op == OP_MAX2)) return true;
if (dt == RAY_F64 && op == OP_POW) return true;
return false;
}
static bool expr_lit_null_sym(const ray_t* lit) {
return lit && lit->type == -RAY_SYM && RAY_ATOM_IS_NULL(lit);
}
static bool expr_sym_lit_bytes(ray_graph_t* g, ray_op_t* n, const char** s, size_t* len) {
if (!n || n->opcode != OP_CONST) return false;
ray_op_ext_t* ext = find_ext(g, n->id);
ray_t* lit = ext ? ext->literal : NULL;
if (!lit) return false;
if (lit->type == -RAY_STR) { *s = ray_str_ptr(lit); *len = ray_str_len(lit); return true; }
if (expr_lit_null_sym(lit)) { *s = ""; *len = 0; return true; }
if (lit->type == -RAY_SYM) {
ray_t* str = ray_sym_str(lit->i64);
if (!str) return false;
*s = ray_str_ptr(str); *len = ray_str_len(str);
return true;
}
return false;
}
static bool expr_sym_lit(ray_graph_t* g, ray_t* tbl, ray_op_t* n, ray_op_t* col) {
const char* s; size_t len;
if (!expr_sym_lit_bytes(g, n, &s, &len)) return false;
ray_op_ext_t* ext = find_ext(g, n->id);
if (!expr_lit_null_sym(ext->literal)) return true;
ray_op_ext_t* ce = col && col->opcode == OP_SCAN ? find_ext(g, col->id) : NULL;
ray_col_meta_t cm;
return ce && ray_table_col_meta(tbl, ce->sym, &cm) && cm.enc != NULL;
}
static bool expr_sym_lit_in(ray_graph_t* g, ray_op_t* n, const ray_col_src_t* src, int64_t* id) {
const char* s; size_t len;
if (!expr_sym_lit_bytes(g, n, &s, &len)) return false;
int64_t x = len == 0 ? 0 : ray_col_src_sym_find(src, s, len);
*id = x < 0 ? -1 : x;
return true;
}
bool expr_compile(ray_graph_t* g, ray_t* tbl, ray_op_t* root, ray_expr_t* out, ray_t** err) {
memset(out, 0, sizeof(*out));
*err = NULL;
if (ray_expr_disable) return false;
if (!root || !g || !tbl) EXPR_BAIL(EXPR_BAIL_ROOT);
if (root->opcode == OP_SCAN || root->opcode == OP_CONST) EXPR_BAIL(EXPR_BAIL_ROOT);
if (!expr_is_elementwise(root->opcode)) EXPR_BAIL(EXPR_BAIL_ROOT);
uint32_t nc = g->node_count;
if (nc > 4096) EXPR_BAIL(EXPR_BAIL_SIZE);
uint8_t node_reg[nc];
memset(node_reg, 0xFF, nc * sizeof(uint8_t));
uint8_t sym_cmp_safe[nc];
memset(sym_cmp_safe, 1, nc * sizeof(uint8_t));
uint8_t sym_eq_only[nc];
memset(sym_eq_only, 1, nc * sizeof(uint8_t));
ray_col_src_t* fd_src[EXPR_MAX_REGS];
bool const_unres[EXPR_MAX_REGS];
memset(fd_src, 0, sizeof fd_src);
memset(const_unres, 0, sizeof const_unres);
{
uint8_t seen[nc];
uint32_t stk[nc];
uint32_t ssp = 0;
memset(seen, 0, nc * sizeof(uint8_t));
if (root->id < nc) { seen[root->id] = 1; stk[ssp++] = root->id; }
while (ssp > 0) {
ray_op_t* n = &g->nodes[stk[--ssp]];
int8_t type = n->out_type;
if (RAY_IS_PARTED(type)) type = (int8_t)RAY_PARTED_BASETYPE(type);
if (type == RAY_TIMESPAN) EXPR_BAIL(EXPR_BAIL_ROOT);
bool cmp = n->arity == 2 && n->opcode >= OP_EQ && n->opcode <= OP_GE;
bool eqne = n->arity == 2 && (n->opcode == OP_EQ || n->opcode == OP_NE);
for (int i = 0; i < n->arity; i++) {
ray_op_t* ch = op_child(g, n, i);
if (!ch || ch->id >= nc) continue;
bool safe = false;
if (cmp) {
ray_op_t* other = op_child(g, n, 1 - i);
if (other && other->opcode == OP_CONST) safe = true;
else if (other && other->opcode == OP_SCAN) {
ray_op_ext_t* oe = find_ext(g, other->id);
ray_col_meta_t om;
bool oh = oe && ray_table_col_meta(tbl, oe->sym, &om);
if (oh && (RAY_IS_PARTED(om.type)
? RAY_PARTED_BASETYPE(om.type) == RAY_SYM
: om.type == RAY_SYM))
safe = true;
}
}
if (!safe) sym_cmp_safe[ch->id] = 0;
if (!(eqne && expr_sym_lit(g, tbl, op_child(g, n, 1 - i), ch))) sym_eq_only[ch->id] = 0;
if (!seen[ch->id]) { seen[ch->id] = 1; stk[ssp++] = ch->id; }
}
}
}
typedef struct { ray_op_t* node; uint8_t phase; } dfs_t;
dfs_t dfs[64];
int sp = 0;
dfs[sp++] = (dfs_t){root, 0};
while (sp > 0) {
dfs_t* top = &dfs[sp - 1];
ray_op_t* node = top->node;
if (node->id < nc && node_reg[node->id] != 0xFF) { sp--; continue; }
if (top->phase == 0) {
top->phase = 1;
for (int i = node->arity - 1; i >= 0; i--) {
ray_op_t* ch = op_child(g, node, i);
if (!ch) continue;
if (ch->id < nc && node_reg[ch->id] != 0xFF) continue;
if (sp >= 64) EXPR_BAIL(EXPR_BAIL_DEPTH);
dfs[sp++] = (dfs_t){ch, 0};
}
} else {
sp--;
uint8_t r = out->n_regs;
if (r >= EXPR_MAX_REGS) EXPR_BAIL(EXPR_BAIL_REGS);
if (node->opcode == OP_SCAN) {
ray_op_ext_t* ext = find_ext(g, node->id);
if (!ext) EXPR_BAIL(EXPR_BAIL_OTHER);
ray_col_meta_t cm;
if (!ray_table_col_meta(tbl, ext->sym, &cm)) EXPR_BAIL(EXPR_BAIL_OTHER);
if (ray_col_bind_declines(tbl, ext->sym))
EXPR_BAIL(cm.type == RAY_MAPCOMMON ? EXPR_BAIL_MAPCOMMON : EXPR_BAIL_SLICE);
ray_col_src_t* src = ray_graph_bind_col(g, tbl, ext->sym, err);
if (*err) return false;
if (!src) EXPR_BAIL(EXPR_BAIL_OTHER);
int8_t elem = src->type;
if (elem == RAY_STR) EXPR_BAIL(EXPR_BAIL_STR);
if (elem == RAY_GUID) EXPR_BAIL(EXPR_BAIL_GUID);
if (!((elem >= RAY_BOOL && elem <= RAY_TIMESTAMP) ||
elem == RAY_SYM))
EXPR_BAIL(EXPR_BAIL_OTHER);
if (src->obj->type != RAY_ENCCOL && (src->obj->attrs & RAY_ATTR_SLICE))
EXPR_BAIL(EXPR_BAIL_SLICE);
if (!src->parted) {
int64_t nrows = ray_table_nrows(tbl);
if (src->len != nrows) EXPR_BAIL(EXPR_BAIL_OTHER);
}
if (elem == RAY_SYM && src->dom != ray_sym_runtime_domain()) {
if (node->id >= nc || !sym_eq_only[node->id])
EXPR_BAIL(EXPR_BAIL_SYM_DOMAIN);
fd_src[r] = src;
}
bool need_proof = !(elem == RAY_SYM && node->id < nc &&
sym_cmp_safe[node->id]);
bool enc_src = src->obj->type == RAY_ENCCOL;
bool col_nulls = false;
if (need_proof && enc_src) col_nulls = src->may_null;
else if (need_proof && !src->may_null) col_nulls = false;
else if (need_proof) {
ray_t* c0 = src->obj;
col_nulls = ray_vec_has_nulls(c0);
if (RAY_IS_PARTED(c0->type)) {
ray_t** segs = (ray_t**)ray_data(c0);
for (int64_t s = 0; s < c0->len; s++)
if (segs[s] && ray_vec_has_nulls(segs[s]))
col_nulls = true;
}
}
if (col_nulls && elem == RAY_SYM)
EXPR_BAIL(EXPR_BAIL_NULLS);
ray_t* ready = ray_col_src_ready(src);
bool enc = !src->parted && !ready;
bool enc_parted = src->parted && !ready;
out->regs[r].kind = REG_SCAN;
out->regs[r].src = src;
out->regs[r].enc = enc;
if (enc) {
out->regs[r].col_type = src->type;
out->regs[r].col_attrs = src->attrs;
out->regs[r].data = NULL;
out->regs[r].is_parted = false;
out->regs[r].col_obj = NULL;
out->regs[r].parted_col = NULL;
out->regs[r].type = (src->type == RAY_F64 || src->type == RAY_F32)
? RAY_F64 : RAY_I64;
out->regs[r].nullable = col_nulls;
out->regs[r].null_src = col_nulls;
} else {
ray_t* col = NULL;
if (enc_parted) col = src->obj;
else {
*err = ray_col_src_plain(src, &col);
if (*err) return false;
}
if (src->parted) {
int8_t base = elem;
out->regs[r].col_type = base;
out->regs[r].data = NULL;
out->regs[r].is_parted = true;
out->regs[r].col_obj = NULL;
out->regs[r].parted_col = col;
out->regs[r].type = (base == RAY_F64 || base == RAY_F32)
? RAY_F64 : RAY_I64;
out->regs[r].nullable = col_nulls;
out->regs[r].null_src = col_nulls;
out->has_parted = true;
} else {
out->regs[r].col_type = col->type;
out->regs[r].col_attrs = col->attrs;
out->regs[r].data = ray_data(col);
out->regs[r].is_parted = false;
out->regs[r].col_obj = col;
out->regs[r].parted_col = NULL;
out->regs[r].type = (col->type == RAY_F64 || col->type == RAY_F32)
? RAY_F64 : RAY_I64;
out->regs[r].nullable = col_nulls;
out->regs[r].null_src = col_nulls;
}
}
} else if (node->opcode == OP_CONST) {
ray_op_ext_t* ext = find_ext(g, node->id);
if (!ext || !ext->literal) EXPR_BAIL(EXPR_BAIL_CONST);
double cf; int64_t ci; bool is_f64;
if (RAY_ATOM_IS_NULL(ext->literal)) {
if (ext->literal->type != -RAY_STR && !expr_lit_null_sym(ext->literal))
EXPR_BAIL(EXPR_BAIL_CONST);
const_unres[r] = true;
ci = 0; cf = 0; is_f64 = false;
} else if (!atom_to_numeric(ext->literal, &cf, &ci, &is_f64)) {
if (ext->literal->type == -RAY_STR) {
const char* s = ray_str_ptr(ext->literal);
size_t slen = ray_str_len(ext->literal);
int64_t sid = ray_sym_find(s, slen);
if (sid < 0) { const_unres[r] = true; sid = -1; }
ci = sid;
cf = (double)sid;
is_f64 = false;
} else {
EXPR_BAIL(EXPR_BAIL_CONST);
}
}
out->regs[r].kind = REG_CONST;
out->regs[r].type = is_f64 ? RAY_F64 : RAY_I64;
out->regs[r].const_f64 = cf;
out->regs[r].const_i64 = ci;
} else if (expr_is_elementwise(node->opcode)) {
if (node->out_type == RAY_F32)
EXPR_BAIL(EXPR_BAIL_OTHER);
if (node->opcode == OP_MOD && node->arity >= 2 &&
node->out_type != RAY_I64 && node->out_type != RAY_F64 &&
op_child(g, node, 1) &&
op_child(g, node, 1)->out_type == RAY_U8)
EXPR_BAIL(EXPR_BAIL_NULL_SHAPE);
if (!op_child(g, node, 0)) EXPR_BAIL(EXPR_BAIL_OTHER);
uint8_t s1 = node_reg[node->in_id[0]];
if (s1 == 0xFF) EXPR_BAIL(EXPR_BAIL_OTHER);
uint8_t s2 = 0xFF;
if (node->arity >= 2 && op_child(g, node, 1)) {
s2 = node_reg[node->in_id[1]];
if (s2 == 0xFF) EXPR_BAIL(EXPR_BAIL_OTHER);
}
int8_t t1 = out->regs[s1].type;
int8_t t2 = (s2 != 0xFF) ? out->regs[s2].type : t1;
uint16_t op = node->opcode;
int8_t ot;
if ((op == OP_LT || op == OP_LE || op == OP_GT || op == OP_GE) &&
(op_child(g, node, 0)->out_type == RAY_SYM ||
(s2 != 0xFF && op_child(g, node, 1)->out_type == RAY_SYM)))
EXPR_BAIL(EXPR_BAIL_OTHER);
if (op == OP_CAST)
ot = node->out_type;
else if ((op >= OP_EQ && op <= OP_GE) ||
op == OP_AND || op == OP_OR || op == OP_NOT ||
op == OP_ISNULL)
ot = RAY_BOOL;
else if (op == OP_SIGNUM)
ot = RAY_I64;
else if (t1 == RAY_F64 || t2 == RAY_F64 || op == OP_DIV ||
op == OP_POW ||
expr_is_f64_unary_math(op))
ot = RAY_F64;
else
ot = RAY_I64;
#define EXPR_PROMOTE(sreg, T) do { \
(sreg) = expr_ensure_type(out, (sreg), (T)); \
if (out->regs[(sreg)].type != (T)) \
EXPR_BAIL(EXPR_BAIL_REGS); \
} while (0)
if (op == OP_CAST) {
} else if (ot == RAY_F64) {
EXPR_PROMOTE(s1, RAY_F64);
if (s2 != 0xFF) EXPR_PROMOTE(s2, RAY_F64);
} else if (ot == RAY_I64) {
if (out->regs[s1].type != RAY_F64) EXPR_PROMOTE(s1, RAY_I64);
if (s2 != 0xFF && out->regs[s2].type != RAY_F64) EXPR_PROMOTE(s2, RAY_I64);
} else if (ot == RAY_BOOL && s2 != 0xFF &&
((op >= OP_EQ && op <= OP_GE) || op == OP_AND || op == OP_OR)) {
int8_t pt = (t1 == RAY_F64 || t2 == RAY_F64) ? RAY_F64
: (t1 == RAY_BOOL && t2 == RAY_BOOL) ? RAY_BOOL : RAY_I64;
if (pt != RAY_BOOL) {
EXPR_PROMOTE(s1, pt);
EXPR_PROMOTE(s2, pt);
}
}
#undef EXPR_PROMOTE
if ((op == OP_EQ || op == OP_NE) && s2 != 0xFF &&
(fd_src[s1] || fd_src[s2])) {
bool left = fd_src[s1] != NULL;
ray_op_t* cn = op_child(g, node, left ? 1 : 0);
uint8_t cs = left ? s2 : s1;
if (out->regs[cs].kind != REG_CONST) EXPR_BAIL(EXPR_BAIL_SYM_DOMAIN);
int64_t fid;
if (!expr_sym_lit_in(g, cn, left ? fd_src[s1] : fd_src[s2], &fid))
EXPR_BAIL(EXPR_BAIL_CONST);
uint8_t nr = out->n_regs;
if (nr >= EXPR_MAX_REGS) EXPR_BAIL(EXPR_BAIL_REGS);
out->regs[nr].kind = REG_CONST;
out->regs[nr].type = RAY_I64;
out->regs[nr].const_i64 = fid;
out->regs[nr].const_f64 = (double)fid;
out->n_regs++;
if (left) s2 = nr; else s1 = nr;
}
r = out->n_regs;
if (r >= EXPR_MAX_REGS) EXPR_BAIL(EXPR_BAIL_REGS);
bool in_null = out->regs[s1].nullable ||
(s2 != 0xFF && out->regs[s2].nullable);
bool in_src = out->regs[s1].null_src ||
(s2 != 0xFF && out->regs[s2].null_src);
bool gen_null = expr_op_generates_null(op, ot, out->regs[s1].type,
s2 != 0xFF);
bool ins_null_aware = false;
bool dst_nullable = false;
if (in_null) {
bool cmp = (op >= OP_EQ && op <= OP_GE);
bool boolop = (op == OP_AND || op == OP_OR || op == OP_NOT);
if (cmp || boolop || op == OP_ISNULL) {
ins_null_aware = (out->regs[s1].type != RAY_F64) ||
op == OP_ISNULL ||
op == OP_AND || op == OP_OR || op == OP_NOT;
dst_nullable = false;
} else if (op == OP_CAST) {
dst_nullable = (ot != RAY_BOOL && ot != RAY_U8);
ins_null_aware = true;
} else {
bool f64_ieee_propagates = (ot == RAY_F64) &&
(op == OP_ADD || op == OP_SUB || op == OP_MUL ||
op == OP_DIV || op == OP_MOD ||
op == OP_NEG || op == OP_ABS ||
expr_is_f64_unary_math(op) ||
op == OP_CEIL || op == OP_FLOOR || op == OP_ROUND);
ins_null_aware = !f64_ieee_propagates;
dst_nullable = true;
}
if (ot == RAY_U8)
EXPR_BAIL(EXPR_BAIL_NULL_SHAPE);
}
out->regs[r].kind = REG_SCRATCH;
out->regs[r].type = ot;
out->regs[r].nullable = dst_nullable || gen_null;
out->regs[r].null_src = dst_nullable && in_src;
out->n_scratch++;
if (out->n_ins >= EXPR_MAX_INS) EXPR_BAIL(EXPR_BAIL_INS);
out->ins[out->n_ins++] = (expr_ins_t){
.opcode = (uint8_t)op, .dst = r, .src1 = s1, .src2 = s2,
.null_aware = ins_null_aware ? 1 : 0,
.u8_mod_zero = (op == OP_MOD && ot == RAY_I64 &&
op_child(g, node, 1) &&
op_child(g, node, 1)->out_type == RAY_U8),
};
} else {
EXPR_BAIL(EXPR_BAIL_OTHER);
}
out->n_regs++;
if (node->id < nc) node_reg[node->id] = r;
}
}
if (out->n_regs == 0 || out->n_ins == 0) EXPR_BAIL(EXPR_BAIL_OTHER);
for (uint8_t i = 0; i < out->n_ins; i++)
if (const_unres[out->ins[i].src1] ||
(out->ins[i].src2 != 0xFF && const_unres[out->ins[i].src2]))
EXPR_BAIL(EXPR_BAIL_CONST);
out->out_reg = out->n_regs - 1;
out->out_type = out->regs[out->out_reg].type;
for (uint8_t i = 0; i < out->n_ins; i++) {
const expr_ins_t* in = &out->ins[i];
if (in->null_aware &&
!expr_null_capable(in->opcode, out->regs[in->dst].type,
out->regs[in->src1].type))
EXPR_BAIL(EXPR_BAIL_NULL_SHAPE);
}
ray_expr_compile_ok++;
return true;
}
static void expr_load_i64(int64_t* dst, const void* data, int8_t col_type,
uint8_t col_attrs, bool has_nulls,
int64_t start, int64_t n) {
switch (col_type) {
case RAY_I64: case RAY_TIMESTAMP:
memcpy(dst, (const int64_t*)data + start, (size_t)n * 8);
break;
case RAY_SYM: {
for (int64_t j = 0; j < n; j++)
dst[j] = ray_read_sym(data, start + j, col_type, col_attrs);
} break;
case RAY_I32: case RAY_DATE: case RAY_TIME: {
const int32_t* s = (const int32_t*)data + start;
if (has_nulls)
for (int64_t j = 0; j < n; j++)
dst[j] = (s[j] == NULL_I32) ? NULL_I64 : (int64_t)s[j];
else
for (int64_t j = 0; j < n; j++) dst[j] = s[j];
} break;
case RAY_U8: case RAY_BOOL: {
const uint8_t* s = (const uint8_t*)data + start;
for (int64_t j = 0; j < n; j++) dst[j] = s[j];
} break;
case RAY_I16: {
const int16_t* s = (const int16_t*)data + start;
if (has_nulls)
for (int64_t j = 0; j < n; j++)
dst[j] = (s[j] == NULL_I16) ? NULL_I64 : (int64_t)s[j];
else
for (int64_t j = 0; j < n; j++) dst[j] = s[j];
} break;
default: memset(dst, 0, (size_t)n * 8); break;
}
}
static void expr_load_f64(double* dst, const void* data, int8_t col_type,
uint8_t col_attrs, bool has_nulls,
int64_t start, int64_t n) {
switch (col_type) {
case RAY_F64:
memcpy(dst, (const double*)data + start, (size_t)n * 8);
break;
case RAY_F32: {
const float* s = (const float*)data + start;
if (has_nulls)
for (int64_t j = 0; j < n; j++)
dst[j] = (s[j] != s[j]) ? NULL_F64 : (double)s[j];
else
for (int64_t j = 0; j < n; j++) dst[j] = (double)s[j];
} break;
case RAY_I64: case RAY_TIMESTAMP: {
const int64_t* s = (const int64_t*)data + start;
if (has_nulls)
for (int64_t j = 0; j < n; j++)
dst[j] = (s[j] == NULL_I64) ? NULL_F64 : (double)s[j];
else
for (int64_t j = 0; j < n; j++) dst[j] = (double)s[j];
} break;
case RAY_SYM: {
for (int64_t j = 0; j < n; j++)
dst[j] = (double)ray_read_sym(data, start + j, col_type, col_attrs);
} break;
case RAY_I32: case RAY_DATE: case RAY_TIME: {
const int32_t* s = (const int32_t*)data + start;
if (has_nulls)
for (int64_t j = 0; j < n; j++)
dst[j] = (s[j] == NULL_I32) ? NULL_F64 : (double)s[j];
else
for (int64_t j = 0; j < n; j++) dst[j] = (double)s[j];
} break;
case RAY_U8: case RAY_BOOL: {
const uint8_t* s = (const uint8_t*)data + start;
for (int64_t j = 0; j < n; j++) dst[j] = (double)s[j];
} break;
case RAY_I16: {
const int16_t* s = (const int16_t*)data + start;
if (has_nulls)
for (int64_t j = 0; j < n; j++)
dst[j] = (s[j] == NULL_I16) ? NULL_F64 : (double)s[j];
else
for (int64_t j = 0; j < n; j++) dst[j] = (double)s[j];
} break;
default: memset(dst, 0, (size_t)n * 8); break;
}
}
static void expr_exec_binary(uint8_t opcode, uint8_t null_aware, uint8_t u8_mod_zero,
int8_t dt, void* dp,
int8_t t1, const void* ap,
int8_t t2, const void* bp, int64_t n) {
(void)t2;
if (dt == RAY_F64) {
double* d = (double*)dp;
const double* a = (const double*)ap;
const double* b = (const double*)bp;
if (null_aware && opcode == OP_POW) {
#define F64_ISN(x) ((x) != (x))
for (int64_t j = 0; j < n; j++)
d[j] = (F64_ISN(a[j]) || F64_ISN(b[j]))
? NULL_F64
: ray_f64_fin(pow(a[j], b[j]));
#undef F64_ISN
return;
}
if (null_aware && (opcode == OP_MIN2 || opcode == OP_MAX2)) {
#define F64_ISN(x) ((x) != (x))
if (opcode == OP_MIN2)
for (int64_t j = 0; j < n; j++)
d[j] = (F64_ISN(a[j]) || F64_ISN(b[j])) ? NULL_F64
: (a[j] < b[j] ? a[j] : b[j]);
else
for (int64_t j = 0; j < n; j++)
d[j] = (F64_ISN(a[j]) || F64_ISN(b[j])) ? NULL_F64
: (a[j] > b[j] ? a[j] : b[j]);
#undef F64_ISN
return;
}
switch (opcode) {
case OP_ADD: for (int64_t j = 0; j < n; j++) d[j] = ray_f64_fin(a[j] + b[j]); break;
case OP_SUB: for (int64_t j = 0; j < n; j++) d[j] = ray_f64_fin(a[j] - b[j]); break;
case OP_MUL: for (int64_t j = 0; j < n; j++) d[j] = ray_f64_fin(a[j] * b[j]); break;
case OP_DIV: for (int64_t j = 0; j < n; j++) d[j] = b[j] != 0.0 ? ray_f64_fin(a[j] / b[j]) : NULL_F64; break;
case OP_IDIV: for (int64_t j = 0; j < n; j++) d[j] = b[j] != 0.0 ? ray_f64_fin(floor(a[j] / b[j])) : NULL_F64; break;
case OP_MOD: for (int64_t j = 0; j < n; j++) {
if (b[j] == 0.0) { d[j] = NULL_F64; continue; }
double m = fmod(a[j], b[j]);
m = (m && ((m > 0) != (b[j] > 0))) ? m + b[j] : m;
d[j] = ray_f64_fin(m);
} break;
case OP_POW: for (int64_t j = 0; j < n; j++) d[j] = ray_f64_fin(pow(a[j], b[j])); break;
case OP_MIN2: for (int64_t j = 0; j < n; j++) d[j] = a[j] < b[j] ? a[j] : b[j]; break;
case OP_MAX2: for (int64_t j = 0; j < n; j++) d[j] = a[j] > b[j] ? a[j] : b[j]; break;
default: break;
}
} else if (dt == RAY_I64 || dt == RAY_TIMESTAMP) {
int64_t* d = (int64_t*)dp;
const int64_t* a = (const int64_t*)ap;
const int64_t* b = (const int64_t*)bp;
#define I64_ISN(x) ((x) == NULL_I64)
if (null_aware) {
switch (opcode) {
case OP_ADD: for (int64_t j=0;j<n;j++) d[j]=(I64_ISN(a[j])||I64_ISN(b[j]))?NULL_I64:(int64_t)((uint64_t)a[j]+(uint64_t)b[j]); break;
case OP_SUB: for (int64_t j=0;j<n;j++) d[j]=(I64_ISN(a[j])||I64_ISN(b[j]))?NULL_I64:(int64_t)((uint64_t)a[j]-(uint64_t)b[j]); break;
case OP_MUL: for (int64_t j=0;j<n;j++) d[j]=(I64_ISN(a[j])||I64_ISN(b[j]))?NULL_I64:(int64_t)((uint64_t)a[j]*(uint64_t)b[j]); break;
case OP_DIV:
case OP_IDIV:
if (i64_span2_dbl_exact(a, b, n)) {
for (int64_t j=0;j<n;j++)
d[j] = b[j] != 0 ? (int64_t)floor((double)a[j] / (double)b[j]) : NULL_I64;
} else {
for (int64_t j=0;j<n;j++) {
if (I64_ISN(a[j])||I64_ISN(b[j])) { d[j]=NULL_I64; continue; }
int64_t q = 0;
d[j] = floor_idiv_i64_checked(a[j], b[j], &q) ? q : NULL_I64;
}
}
break;
case OP_MOD: for (int64_t j=0;j<n;j++) {
if (I64_ISN(a[j])||I64_ISN(b[j])) { d[j]=NULL_I64; continue; }
if (b[j]==0) { d[j]=u8_mod_zero ? 0 : NULL_I64; continue; }
if (b[j]==-1&&a[j]==INT64_MIN) { d[j]=NULL_I64; continue; }
int64_t m=a[j]%b[j];
if (m&&(m^b[j])<0) m+=b[j];
d[j]=m;
} break;
case OP_MIN2: for (int64_t j=0;j<n;j++) d[j]=(I64_ISN(a[j])||I64_ISN(b[j]))?NULL_I64:(a[j]<b[j]?a[j]:b[j]); break;
case OP_MAX2: for (int64_t j=0;j<n;j++) d[j]=(I64_ISN(a[j])||I64_ISN(b[j]))?NULL_I64:(a[j]>b[j]?a[j]:b[j]); break;
default: memset(dp, 0, (size_t)n * 8); break;
}
} else {
switch (opcode) {
case OP_ADD: for (int64_t j = 0; j < n; j++) d[j] = (int64_t)((uint64_t)a[j] + (uint64_t)b[j]); break;
case OP_SUB: for (int64_t j = 0; j < n; j++) d[j] = (int64_t)((uint64_t)a[j] - (uint64_t)b[j]); break;
case OP_MUL: for (int64_t j = 0; j < n; j++) d[j] = (int64_t)((uint64_t)a[j] * (uint64_t)b[j]); break;
case OP_DIV:
case OP_IDIV:
if (i64_span2_dbl_exact(a, b, n)) {
for (int64_t j = 0; j < n; j++)
d[j] = b[j] != 0 ? (int64_t)floor((double)a[j] / (double)b[j]) : NULL_I64;
} else {
for (int64_t j = 0; j < n; j++) {
int64_t q = 0;
d[j] = floor_idiv_i64_checked(a[j], b[j], &q) ? q : NULL_I64;
}
}
break;
case OP_MOD: for (int64_t j = 0; j < n; j++) {
if (b[j]==0) { d[j]=u8_mod_zero ? 0 : NULL_I64; continue; }
if (b[j]==-1 && a[j]==INT64_MIN) { d[j]=NULL_I64; continue; }
int64_t m = a[j]%b[j];
if (m && (m^b[j])<0) m+=b[j];
d[j] = m;
} break;
case OP_MIN2: for (int64_t j = 0; j < n; j++) d[j] = a[j] < b[j] ? a[j] : b[j]; break;
case OP_MAX2: for (int64_t j = 0; j < n; j++) d[j] = a[j] > b[j] ? a[j] : b[j]; break;
case OP_AND: for (int64_t j = 0; j < n; j++) d[j] = (a[j] && b[j]) ? 1 : 0; break;
case OP_OR: for (int64_t j = 0; j < n; j++) d[j] = (a[j] || b[j]) ? 1 : 0; break;
default: break;
}
}
#undef I64_ISN
} else if (dt == RAY_I32 || dt == RAY_DATE || dt == RAY_TIME) {
int32_t* d = (int32_t*)dp;
const int32_t* a = (const int32_t*)ap;
const int32_t* b = (const int32_t*)bp;
switch (opcode) {
case OP_ADD: for (int64_t j = 0; j < n; j++) d[j] = (int32_t)((uint32_t)a[j] + (uint32_t)b[j]); break;
case OP_SUB: for (int64_t j = 0; j < n; j++) d[j] = (int32_t)((uint32_t)a[j] - (uint32_t)b[j]); break;
case OP_MUL: for (int64_t j = 0; j < n; j++) d[j] = (int32_t)((uint32_t)a[j] * (uint32_t)b[j]); break;
case OP_DIV: for (int64_t j = 0; j < n; j++) {
if (b[j]==0 || (b[j]==-1 && a[j]==INT32_MIN)) { d[j]=0; continue; }
int32_t q = a[j]/b[j];
if ((a[j]^b[j])<0 && q*b[j]!=a[j]) q--;
d[j] = q;
} break;
case OP_MOD: for (int64_t j = 0; j < n; j++) {
if (b[j]==0 || (b[j]==-1 && a[j]==INT32_MIN)) { d[j]=0; continue; }
int32_t m = a[j]%b[j];
if (m && (m^b[j])<0) m+=b[j];
d[j] = m;
} break;
case OP_MIN2: for (int64_t j = 0; j < n; j++) d[j] = a[j] < b[j] ? a[j] : b[j]; break;
case OP_MAX2: for (int64_t j = 0; j < n; j++) d[j] = a[j] > b[j] ? a[j] : b[j]; break;
default: break;
}
} else if (dt == RAY_I16) {
int16_t* d = (int16_t*)dp;
const int16_t* a = (const int16_t*)ap;
const int16_t* b = (const int16_t*)bp;
switch (opcode) {
case OP_ADD: for (int64_t j = 0; j < n; j++) d[j] = (int16_t)((uint16_t)a[j] + (uint16_t)b[j]); break;
case OP_SUB: for (int64_t j = 0; j < n; j++) d[j] = (int16_t)((uint16_t)a[j] - (uint16_t)b[j]); break;
case OP_MUL: for (int64_t j = 0; j < n; j++) d[j] = (int16_t)((uint16_t)a[j] * (uint16_t)b[j]); break;
case OP_DIV: for (int64_t j = 0; j < n; j++) { d[j] = b[j] ? a[j] / b[j] : 0; } break;
case OP_MOD: for (int64_t j = 0; j < n; j++) { d[j] = b[j] ? a[j] % b[j] : 0; } break;
case OP_MIN2: for (int64_t j = 0; j < n; j++) d[j] = a[j] < b[j] ? a[j] : b[j]; break;
case OP_MAX2: for (int64_t j = 0; j < n; j++) d[j] = a[j] > b[j] ? a[j] : b[j]; break;
default: break;
}
} else if (dt == RAY_U8) {
uint8_t* d2 = (uint8_t*)dp;
const uint8_t* a2 = (const uint8_t*)ap;
const uint8_t* b2 = (const uint8_t*)bp;
switch (opcode) {
case OP_ADD: for (int64_t j = 0; j < n; j++) d2[j] = a2[j] + b2[j]; break;
case OP_SUB: for (int64_t j = 0; j < n; j++) d2[j] = a2[j] - b2[j]; break;
case OP_MUL: for (int64_t j = 0; j < n; j++) d2[j] = a2[j] * b2[j]; break;
case OP_DIV: for (int64_t j = 0; j < n; j++) { d2[j] = b2[j] ? a2[j] / b2[j] : 0; } break;
case OP_MOD: for (int64_t j = 0; j < n; j++) { d2[j] = b2[j] ? a2[j] % b2[j] : 0; } break;
case OP_MIN2: for (int64_t j = 0; j < n; j++) d2[j] = a2[j] < b2[j] ? a2[j] : b2[j]; break;
case OP_MAX2: for (int64_t j = 0; j < n; j++) d2[j] = a2[j] > b2[j] ? a2[j] : b2[j]; break;
default: break;
}
} else if (dt == RAY_BOOL) {
uint8_t* d = (uint8_t*)dp;
if (t1 == RAY_F64) {
const double* a = (const double*)ap;
const double* b = (const double*)bp;
#define F64_ISNAN(x) ((x) != (x))
switch (opcode) {
case OP_EQ: for (int64_t j = 0; j < n; j++) d[j] = (F64_ISNAN(a[j])&&F64_ISNAN(b[j])) ? 1 : (F64_ISNAN(a[j])||F64_ISNAN(b[j])) ? 0 : a[j]==b[j]; break;
case OP_NE: for (int64_t j = 0; j < n; j++) d[j] = (F64_ISNAN(a[j])&&F64_ISNAN(b[j])) ? 0 : (F64_ISNAN(a[j])||F64_ISNAN(b[j])) ? 1 : a[j]!=b[j]; break;
case OP_LT: for (int64_t j = 0; j < n; j++) d[j] = (F64_ISNAN(a[j])&&F64_ISNAN(b[j])) ? 0 : F64_ISNAN(a[j]) ? 1 : F64_ISNAN(b[j]) ? 0 : a[j]<b[j]; break;
case OP_LE: for (int64_t j = 0; j < n; j++) d[j] = (F64_ISNAN(a[j])&&F64_ISNAN(b[j])) ? 1 : F64_ISNAN(a[j]) ? 1 : F64_ISNAN(b[j]) ? 0 : a[j]<=b[j]; break;
case OP_GT: for (int64_t j = 0; j < n; j++) d[j] = (F64_ISNAN(a[j])&&F64_ISNAN(b[j])) ? 0 : F64_ISNAN(b[j]) ? 1 : F64_ISNAN(a[j]) ? 0 : a[j]>b[j]; break;
case OP_GE: for (int64_t j = 0; j < n; j++) d[j] = (F64_ISNAN(a[j])&&F64_ISNAN(b[j])) ? 1 : F64_ISNAN(b[j]) ? 1 : F64_ISNAN(a[j]) ? 0 : a[j]>=b[j]; break;
default: break;
}
#undef F64_ISNAN
} else if (t1 == RAY_I64) {
const int64_t* a = (const int64_t*)ap;
const int64_t* b = (const int64_t*)bp;
#define I64_ISN(x) ((x) == NULL_I64)
if (null_aware) {
switch (opcode) {
case OP_EQ: for (int64_t j=0;j<n;j++) d[j]=(I64_ISN(a[j])&&I64_ISN(b[j]))?1:(I64_ISN(a[j])||I64_ISN(b[j]))?0:a[j]==b[j]; break;
case OP_NE: for (int64_t j=0;j<n;j++) d[j]=(I64_ISN(a[j])&&I64_ISN(b[j]))?0:(I64_ISN(a[j])||I64_ISN(b[j]))?1:a[j]!=b[j]; break;
case OP_LT: for (int64_t j=0;j<n;j++) d[j]=(I64_ISN(a[j])&&I64_ISN(b[j]))?0:I64_ISN(a[j])?1:I64_ISN(b[j])?0:a[j]<b[j]; break;
case OP_LE: for (int64_t j=0;j<n;j++) d[j]=(I64_ISN(a[j])&&I64_ISN(b[j]))?1:I64_ISN(a[j])?1:I64_ISN(b[j])?0:a[j]<=b[j]; break;
case OP_GT: for (int64_t j=0;j<n;j++) d[j]=(I64_ISN(a[j])&&I64_ISN(b[j]))?0:I64_ISN(b[j])?1:I64_ISN(a[j])?0:a[j]>b[j]; break;
case OP_GE: for (int64_t j=0;j<n;j++) d[j]=(I64_ISN(a[j])&&I64_ISN(b[j]))?1:I64_ISN(b[j])?1:I64_ISN(a[j])?0:a[j]>=b[j]; break;
case OP_AND: for (int64_t j=0;j<n;j++) d[j]=(I64_ISN(a[j])||I64_ISN(b[j]))?0:(a[j]&&b[j])?1:0; break;
case OP_OR: for (int64_t j=0;j<n;j++) d[j]=(I64_ISN(a[j])||I64_ISN(b[j]))?0:(a[j]||b[j])?1:0; break;
default: break;
}
} else {
switch (opcode) {
case OP_EQ: for (int64_t j = 0; j < n; j++) d[j] = a[j]==b[j]; break;
case OP_NE: for (int64_t j = 0; j < n; j++) d[j] = a[j]!=b[j]; break;
case OP_LT: for (int64_t j = 0; j < n; j++) d[j] = a[j]<b[j]; break;
case OP_LE: for (int64_t j = 0; j < n; j++) d[j] = a[j]<=b[j]; break;
case OP_GT: for (int64_t j = 0; j < n; j++) d[j] = a[j]>b[j]; break;
case OP_GE: for (int64_t j = 0; j < n; j++) d[j] = a[j]>=b[j]; break;
case OP_AND: for (int64_t j = 0; j < n; j++) d[j] = (a[j] && b[j]) ? 1 : 0; break;
case OP_OR: for (int64_t j = 0; j < n; j++) d[j] = (a[j] || b[j]) ? 1 : 0; break;
default: break;
}
}
#undef I64_ISN
} else {
const uint8_t* a = (const uint8_t*)ap;
const uint8_t* b = (const uint8_t*)bp;
switch (opcode) {
case OP_AND: for (int64_t j = 0; j < n; j++) d[j] = a[j] && b[j]; break;
case OP_OR: for (int64_t j = 0; j < n; j++) d[j] = a[j] || b[j]; break;
default: break;
}
}
}
}
static void expr_exec_unary(uint8_t opcode, uint8_t null_aware, int8_t dt, void* dp,
int8_t t1, const void* ap, int64_t n) {
if (opcode == OP_CAST) {
dt = expr_cast_lane_type(dt);
t1 = expr_cast_lane_type(t1);
}
if (dt == RAY_F64) {
double* d = (double*)dp;
if (t1 == RAY_F64) {
const double* a = (const double*)ap;
switch (opcode) {
case OP_NEG: for (int64_t j = 0; j < n; j++) d[j] = -a[j]; break;
case OP_ABS: for (int64_t j = 0; j < n; j++) d[j] = fabs(a[j]); break;
case OP_SQRT: for (int64_t j = 0; j < n; j++) d[j] = ray_f64_fin(sqrt(a[j])); break;
case OP_LOG: for (int64_t j = 0; j < n; j++) d[j] = ray_f64_fin(log(a[j])); break;
case OP_EXP: for (int64_t j = 0; j < n; j++) d[j] = ray_f64_fin(exp(a[j])); break;
case OP_SIN: for (int64_t j = 0; j < n; j++) d[j] = ray_f64_fin(sin(a[j])); break;
case OP_ASIN: for (int64_t j = 0; j < n; j++) d[j] = ray_f64_fin(asin(a[j])); break;
case OP_COS: for (int64_t j = 0; j < n; j++) d[j] = ray_f64_fin(cos(a[j])); break;
case OP_ACOS: for (int64_t j = 0; j < n; j++) d[j] = ray_f64_fin(acos(a[j])); break;
case OP_TAN: for (int64_t j = 0; j < n; j++) d[j] = ray_f64_fin(tan(a[j])); break;
case OP_ATAN: for (int64_t j = 0; j < n; j++) d[j] = ray_f64_fin(atan(a[j])); break;
case OP_RECIPROCAL:
for (int64_t j = 0; j < n; j++)
d[j] = a[j] != 0.0 ? ray_f64_fin(1.0 / a[j]) : NULL_F64;
break;
case OP_CEIL: for (int64_t j = 0; j < n; j++) d[j] = ceil(a[j]); break;
case OP_FLOOR: for (int64_t j = 0; j < n; j++) d[j] = floor(a[j]); break;
case OP_ROUND: for (int64_t j = 0; j < n; j++) d[j] = round(a[j]); break;
case OP_CAST: memcpy(d, a, (size_t)n * sizeof(double)); break;
default: break;
}
} else if (t1 == RAY_I32) {
const int32_t* a = (const int32_t*)ap;
for (int64_t j = 0; j < n; j++)
d[j] = (null_aware && a[j] == NULL_I32) ? NULL_F64 : (double)a[j];
} else if (t1 == RAY_I16) {
const int16_t* a = (const int16_t*)ap;
for (int64_t j = 0; j < n; j++)
d[j] = (null_aware && a[j] == NULL_I16) ? NULL_F64 : (double)a[j];
} else if (t1 == RAY_BOOL || t1 == RAY_U8) {
const uint8_t* a = (const uint8_t*)ap;
for (int64_t j = 0; j < n; j++) d[j] = (double)a[j];
} else {
const int64_t* a = (const int64_t*)ap;
if (null_aware)
for (int64_t j = 0; j < n; j++)
d[j] = (a[j] == NULL_I64) ? NULL_F64 : (double)a[j];
else
for (int64_t j = 0; j < n; j++) d[j] = (double)a[j];
}
} else if (dt == RAY_I64) {
int64_t* d = (int64_t*)dp;
if (t1 == RAY_I64) {
const int64_t* a = (const int64_t*)ap;
if (null_aware) {
#define I64_ISN(x) ((x) == NULL_I64)
switch (opcode) {
case OP_NEG: for (int64_t j=0;j<n;j++) d[j]=I64_ISN(a[j])?NULL_I64:(int64_t)(-(uint64_t)a[j]); break;
case OP_ABS: for (int64_t j=0;j<n;j++) d[j]=I64_ISN(a[j])?NULL_I64:(a[j]<0?(int64_t)(-(uint64_t)a[j]):a[j]); break;
case OP_SIGNUM: for (int64_t j=0;j<n;j++) d[j]=I64_ISN(a[j])?NULL_I64:(a[j]>0)-(a[j]<0); break;
case OP_CAST: memcpy(d, a, (size_t)n * sizeof(int64_t)); break;
default: break;
}
#undef I64_ISN
return;
}
switch (opcode) {
case OP_NEG: for (int64_t j = 0; j < n; j++) d[j] = (int64_t)(-(uint64_t)a[j]); break;
case OP_ABS: for (int64_t j = 0; j < n; j++) d[j] = a[j] < 0 ? (int64_t)(-(uint64_t)a[j]) : a[j]; break;
case OP_SIGNUM: for (int64_t j = 0; j < n; j++) d[j] = (a[j] > 0) - (a[j] < 0); break;
case OP_CAST: memcpy(d, a, (size_t)n * sizeof(int64_t)); break;
default: break;
}
} else if (t1 == RAY_BOOL || t1 == RAY_U8) {
const uint8_t* a = (const uint8_t*)ap;
for (int64_t j = 0; j < n; j++) d[j] = a[j];
} else if (t1 == RAY_I32 || t1 == RAY_DATE || t1 == RAY_TIME) {
const int32_t* a = (const int32_t*)ap;
if (null_aware)
for (int64_t j = 0; j < n; j++)
d[j] = (a[j] == NULL_I32) ? NULL_I64 : (int64_t)a[j];
else
for (int64_t j = 0; j < n; j++) d[j] = a[j];
} else if (t1 == RAY_I16) {
const int16_t* a = (const int16_t*)ap;
if (null_aware)
for (int64_t j = 0; j < n; j++)
d[j] = (a[j] == NULL_I16) ? NULL_I64 : (int64_t)a[j];
else
for (int64_t j = 0; j < n; j++) d[j] = a[j];
} else {
const double* a = (const double*)ap;
if (opcode == OP_SIGNUM) {
if (null_aware)
for (int64_t j = 0; j < n; j++)
d[j] = (a[j] != a[j]) ? NULL_I64 : (a[j] > 0.0) - (a[j] < 0.0);
else
for (int64_t j = 0; j < n; j++)
d[j] = (a[j] > 0.0) - (a[j] < 0.0);
return;
}
if (null_aware)
for (int64_t j = 0; j < n; j++)
d[j] = ray_cast_f64_to_i64_null(a[j]);
else
for (int64_t j = 0; j < n; j++)
d[j] = ray_cast_f64_to_i64_null(a[j]);
}
} else if (dt == RAY_BOOL) {
uint8_t* d = (uint8_t*)dp;
if (opcode == OP_ISNULL) {
if (t1 == RAY_F64) {
const double* a = (const double*)ap;
for (int64_t j = 0; j < n; j++) d[j] = (a[j] != a[j]) ? 1 : 0;
} else if ((t1 == RAY_I32 || t1 == RAY_I16) && !null_aware) {
memset(d, 0, (size_t)n);
} else if (t1 == RAY_I32) {
const int32_t* a = (const int32_t*)ap;
for (int64_t j = 0; j < n; j++) d[j] = (a[j] == NULL_I32) ? 1 : 0;
} else if (t1 == RAY_I16) {
const int16_t* a = (const int16_t*)ap;
for (int64_t j = 0; j < n; j++) d[j] = (a[j] == NULL_I16) ? 1 : 0;
} else if (t1 == RAY_BOOL || t1 == RAY_U8) {
memset(d, 0, (size_t)n);
} else {
const int64_t* a = (const int64_t*)ap;
for (int64_t j = 0; j < n; j++) d[j] = (a[j] == NULL_I64) ? 1 : 0;
}
} else if (opcode == OP_CAST) {
if (t1 == RAY_F64) {
const double* a = (const double*)ap;
for (int64_t j = 0; j < n; j++)
d[j] = ray_cast_f64_to_bool_null(a[j]);
} else if (t1 == RAY_I32) {
const int32_t* a = (const int32_t*)ap;
for (int64_t j = 0; j < n; j++)
d[j] = a[j] != 0 ? 1 : 0;
} else if (t1 == RAY_I16) {
const int16_t* a = (const int16_t*)ap;
for (int64_t j = 0; j < n; j++)
d[j] = a[j] != 0 ? 1 : 0;
} else if (t1 == RAY_BOOL || t1 == RAY_U8) {
const uint8_t* a = (const uint8_t*)ap;
for (int64_t j = 0; j < n; j++) d[j] = a[j] ? 1 : 0;
} else {
const int64_t* a = (const int64_t*)ap;
for (int64_t j = 0; j < n; j++)
d[j] = a[j] != 0 ? 1 : 0;
}
} else if (opcode == OP_NOT) {
if (t1 == RAY_F64) {
const double* a = (const double*)ap;
for (int64_t j = 0; j < n; j++) d[j] = (a[j] == 0.0) ? 1 : 0;
} else if (t1 == RAY_I64) {
const int64_t* a = (const int64_t*)ap;
for (int64_t j = 0; j < n; j++) d[j] = (a[j] == 0) ? 1 : 0;
} else if (t1 == RAY_I32) {
const int32_t* a = (const int32_t*)ap;
for (int64_t j = 0; j < n; j++) d[j] = (a[j] == 0) ? 1 : 0;
} else if (t1 == RAY_I16) {
const int16_t* a = (const int16_t*)ap;
for (int64_t j = 0; j < n; j++) d[j] = (a[j] == 0) ? 1 : 0;
} else {
const uint8_t* a = (const uint8_t*)ap;
for (int64_t j = 0; j < n; j++) d[j] = !a[j];
}
}
} else if (dt == RAY_I32) {
int32_t* d = (int32_t*)dp;
if (t1 == RAY_F64) {
const double* a = (const double*)ap;
if (null_aware)
for (int64_t j = 0; j < n; j++)
d[j] = ray_cast_f64_to_i32_null(a[j]);
else
for (int64_t j = 0; j < n; j++) d[j] = ray_cast_f64_to_i32_null(a[j]);
} else if (t1 == RAY_I32) {
memcpy(d, ap, (size_t)n * sizeof(int32_t));
} else if (t1 == RAY_I16) {
const int16_t* a = (const int16_t*)ap;
for (int64_t j = 0; j < n; j++)
d[j] = (null_aware && a[j] == NULL_I16) ? NULL_I32 : (int32_t)a[j];
} else if (t1 == RAY_BOOL || t1 == RAY_U8) {
const uint8_t* a = (const uint8_t*)ap;
for (int64_t j = 0; j < n; j++) d[j] = a[j];
} else {
const int64_t* a = (const int64_t*)ap;
if (null_aware)
for (int64_t j = 0; j < n; j++)
d[j] = (a[j] == NULL_I64) ? NULL_I32 : (int32_t)a[j];
else
for (int64_t j = 0; j < n; j++) d[j] = (int32_t)a[j];
}
} else if (dt == RAY_I16) {
int16_t* d = (int16_t*)dp;
if (t1 == RAY_F64) {
const double* a = (const double*)ap;
if (null_aware)
for (int64_t j = 0; j < n; j++)
d[j] = ray_cast_f64_to_i16_null(a[j]);
else
for (int64_t j = 0; j < n; j++) d[j] = ray_cast_f64_to_i16_null(a[j]);
} else if (t1 == RAY_I32) {
const int32_t* a = (const int32_t*)ap;
for (int64_t j = 0; j < n; j++)
d[j] = (null_aware && a[j] == NULL_I32) ? NULL_I16 : (int16_t)a[j];
} else if (t1 == RAY_I16) {
memcpy(d, ap, (size_t)n * sizeof(int16_t));
} else if (t1 == RAY_BOOL || t1 == RAY_U8) {
const uint8_t* a = (const uint8_t*)ap;
for (int64_t j = 0; j < n; j++) d[j] = a[j];
} else {
const int64_t* a = (const int64_t*)ap;
if (null_aware)
for (int64_t j = 0; j < n; j++)
d[j] = (a[j] == NULL_I64) ? NULL_I16 : (int16_t)a[j];
else
for (int64_t j = 0; j < n; j++) d[j] = (int16_t)a[j];
}
} else if (dt == RAY_U8) {
uint8_t* d = (uint8_t*)dp;
if (t1 == RAY_F64) {
const double* a = (const double*)ap;
for (int64_t j = 0; j < n; j++) d[j] = ray_cast_f64_to_u8_null(a[j]);
} else if (t1 == RAY_I32) {
const int32_t* a = (const int32_t*)ap;
for (int64_t j = 0; j < n; j++) d[j] = (uint8_t)a[j];
} else if (t1 == RAY_I16) {
const int16_t* a = (const int16_t*)ap;
for (int64_t j = 0; j < n; j++) d[j] = (uint8_t)a[j];
} else if (t1 == RAY_BOOL || t1 == RAY_U8) {
memcpy(d, ap, (size_t)n);
} else {
const int64_t* a = (const int64_t*)ap;
for (int64_t j = 0; j < n; j++) d[j] = (uint8_t)a[j];
}
}
}
static inline bool expr_src_synthetic(const ray_col_src_t* src) {
return ray_col_src_encoded(src) && ray_enccol_is_synthetic(src->obj);
}
static bool expr_enc_null_fill(const ray_expr_t* expr, uint8_t r, int64_t start, int64_t n, void* dst) {
const __typeof__(expr->regs[0])* rg = &expr->regs[r];
if ((start >> 16) != ((start + n - 1) >> 16) || !expr_src_synthetic(rg->src)) return false;
int8_t ct = rg->col_type, rt = rg->type;
int64_t raw;
switch (ct) {
case RAY_I64: case RAY_TIMESTAMP: raw = NULL_I64; break;
case RAY_I32: case RAY_DATE: case RAY_TIME: raw = NULL_I32; break;
case RAY_I16: raw = NULL_I16; break;
case RAY_F64: case RAY_F32: raw = 0; break;
default: return false;
}
ray_enc_chunk_stat_t st;
if (!ray_col_src_chunk_stat(rg->src, start >> 16, &st) || st.rows <= 0 ||
(int64_t)st.null_count != st.rows) return false;
bool fl = ct == RAY_F64 || ct == RAY_F32;
if (rt == RAY_F64) {
double v = fl || rg->nullable ? NULL_F64 : (double)raw;
for (int64_t j = 0; j < n; j++) ((double*)dst)[j] = v;
return true;
}
if (rt != RAY_I64 || fl) return false;
int64_t v = rg->nullable ? NULL_I64 : raw;
for (int64_t j = 0; j < n; j++) ((int64_t*)dst)[j] = v;
return true;
}
RAY_INLINE ray_t* expr_load_reg(const ray_expr_t* expr, uint8_t r, void** scratch,
ray_col_scratch_t* rsc, int64_t start, int64_t n, void** rptr) {
int8_t rt = expr->regs[r].type;
int8_t ct = expr->regs[r].col_type;
switch (expr->regs[r].kind) {
case REG_SCAN: {
uint8_t ca = expr->regs[r].col_attrs;
bool hn = expr->regs[r].nullable;
bool direct = (rt == RAY_F64 && ct == RAY_F64) ||
(rt == RAY_I64 && (ct == RAY_I64 || ct == RAY_TIMESTAMP)) ||
(rt == RAY_I64 && ct == RAY_SYM && (ca & RAY_SYM_W_MASK) == RAY_SYM_W64);
if (expr->regs[r].enc) {
int64_t at = 0;
*rptr = scratch[r];
if (expr_enc_null_fill(expr, r, start, n, scratch[r])) return NULL;
if (!direct && (rt == RAY_I64 || rt == RAY_F64) &&
(ct == RAY_BOOL || ct == RAY_U8 || ct == RAY_I16 || ct == RAY_I32 ||
ct == RAY_DATE || ct == RAY_TIME)) {
ray_t* e = ray_col_range_wide(expr->regs[r].src, start, n, &rsc[r], rt, scratch[r], &at);
if (e) return e;
}
while (at < n) {
ray_col_range_t v;
ray_t* e = ray_col_range(expr->regs[r].src, start + at, n - at, &rsc[r], &v);
if (e) return e;
if (v.n <= 0) return ray_error("corrupt", "fused scan: short range read");
if (at == 0 && v.n == n && direct) { *rptr = (void*)v.data; break; }
if (direct)
memcpy((char*)scratch[r] + at * 8, v.data, (size_t)v.n * 8);
else if (rt == RAY_F64)
expr_load_f64((double*)scratch[r] + at, v.data, ct, v.attrs, hn, 0, v.n);
else
expr_load_i64((int64_t*)scratch[r] + at, v.data, ct, v.attrs, hn, 0, v.n);
at += v.n;
}
} else if (direct) {
*rptr = (char*)expr->regs[r].data + (size_t)start * 8;
} else {
*rptr = scratch[r];
if (rt == RAY_F64)
expr_load_f64(scratch[r], expr->regs[r].data, ct, ca, hn, start, n);
else
expr_load_i64(scratch[r], expr->regs[r].data, ct, ca, hn, start, n);
}
}
break;
case REG_CONST:
*rptr = scratch[r];
if (rt == RAY_F64) {
double v = expr->regs[r].const_f64;
double* d = (double*)scratch[r];
for (int64_t j = 0; j < n; j++) d[j] = v;
} else {
int64_t v = expr->regs[r].const_i64;
int64_t* d = (int64_t*)scratch[r];
for (int64_t j = 0; j < n; j++) d[j] = v;
}
break;
default:
*rptr = scratch[r];
break;
}
return NULL;
}
static int expr_leaf_on_view(const ray_expr_t* expr, const expr_enc_plan_t* p, uint8_t i,
ray_enc_vview_t* vbuf, void** scratch, void** rptrs,
int64_t start, int64_t n, ray_t** err);
#if defined(__GNUC__) && !defined(__clang__)
#define EXPR_ONE_COPY __attribute__((noinline, noclone))
#else
#define EXPR_ONE_COPY __attribute__((noinline))
#endif
EXPR_ONE_COPY static void* expr_eval_morsel(const ray_expr_t* expr, const expr_enc_plan_t* plan, ray_enc_vview_t* vbuf,
void** scratch, ray_col_scratch_t* rsc, int64_t start, int64_t end,
ray_t** err) {
int64_t n = end - start;
if (n <= 0) return NULL;
void* rptrs[EXPR_MAX_REGS];
bool loaded[EXPR_MAX_REGS];
for (uint8_t r = 0; r < expr->n_regs; r++) {
if (plan && !plan->load_reg[r]) { rptrs[r] = scratch[r]; loaded[r] = false; continue; }
loaded[r] = true;
ray_t* e = expr_load_reg(expr, r, scratch, rsc, start, n, &rptrs[r]);
if (e) { *err = e; return NULL; }
}
for (uint8_t i = 0; i < expr->n_ins; i++) {
const expr_ins_t* ins = &expr->ins[i];
if (plan && plan->leaf_of_ins[i] >= 0) {
int lv = expr_leaf_on_view(expr, plan, i, vbuf, scratch, rptrs, start, n, err);
if (lv < 0) return NULL;
if (lv > 0) continue;
uint8_t lr = plan->leaf[plan->leaf_of_ins[i]].reg;
if (!loaded[lr]) {
ray_t* e = expr_load_reg(expr, lr, scratch, rsc, start, n, &rptrs[lr]);
if (e) { *err = e; return NULL; }
loaded[lr] = true;
}
}
int8_t dt = expr->regs[ins->dst].type;
if (ins->src2 != 0xFF) {
expr_exec_binary(ins->opcode, ins->null_aware, ins->u8_mod_zero,
dt, rptrs[ins->dst],
expr->regs[ins->src1].type, rptrs[ins->src1],
expr->regs[ins->src2].type, rptrs[ins->src2], n);
} else {
expr_exec_unary(ins->opcode, ins->null_aware, dt, rptrs[ins->dst],
expr->regs[ins->src1].type, rptrs[ins->src1], n);
}
}
return rptrs[expr->out_reg];
}
#define ZT_MIXED (-1)
#define ZT_NONE (-2)
static inline int zt_and(int a, int b) {
if (a == 0 || b == 0) return 0;
if (a == 1 && b == 1) return 1;
return ZT_MIXED;
}
static inline int zt_or(int a, int b) {
if (a == 1 || b == 1) return 1;
if (a == 0 && b == 0) return 0;
return ZT_MIXED;
}
static inline uint16_t zone_swap_op(uint16_t op) {
switch (op) {
case OP_LT: return OP_GT;
case OP_GT: return OP_LT;
case OP_LE: return OP_GE;
case OP_GE: return OP_LE;
default: return op;
}
}
static int zone_cmp_decision(const ray_index_t* ix, int64_t ch,
uint16_t cmp_op, int64_t cval) {
const int64_t* mins = (const int64_t*)ray_data(ix->u.chunk_zone.mins);
const int64_t* maxs = (const int64_t*)ray_data(ix->u.chunk_zone.maxs);
const uint8_t* nb = (const uint8_t*)ray_data(ix->u.chunk_zone.null_bits);
bool has_nulls = (nb[ch >> 3] >> (ch & 7)) & 1u;
int d = zone_cmp_core(mins[ch], maxs[ch], true, has_nulls, false, cmp_op, cval);
return d < 0 ? ZT_MIXED : d;
}
static int zone_all_null_decision(uint16_t cmp_op) {
switch (cmp_op) {
case OP_EQ: case OP_GT: case OP_GE: return 0;
case OP_NE: case OP_LT: case OP_LE: return 1;
default: return ZT_MIXED;
}
}
_Static_assert(RAY_ENC_CHUNK == 1 << 16, "the directory gate reads 64K-row chunks");
static int zone_dir_decision(const ray_col_src_t* src, int64_t ch, uint16_t cmp_op, int64_t cval) {
ray_enc_chunk_stat_t st;
if (!ray_col_src_chunk_stat(src, ch, &st)) return ZT_MIXED;
bool all_null = (int64_t)st.null_count == st.rows;
if (all_null && st.rows > 0 && expr_src_synthetic(src)) return zone_all_null_decision(cmp_op);
int d = zone_cmp_core(st.min, st.max, st.has_minmax, st.null_count > 0, all_null, cmp_op, cval);
return d < 0 ? ZT_MIXED : d;
}
static int zone_vec_decision(const ray_col_src_t* src, int64_t v, uint16_t cmp_op, int64_t cval) {
int64_t mn, mx; uint32_t nn;
if (!ray_col_src_vec_stat(src, v, &mn, &mx, &nn)) return ZT_MIXED;
int d = zone_cmp_core(mn, mx, true, nn > 0, false, cmp_op, cval);
return d < 0 ? ZT_MIXED : d;
}
static int zone_sym_dir_decision(const ray_col_src_t* src, int64_t ch, uint16_t cmp_op, int64_t cval) {
if (cval < 0) return zone_sym_decision(0, 0, false, 0, 1, cmp_op, cval);
ray_enc_chunk_stat_t st;
if (!ray_col_src_chunk_stat(src, ch, &st)) return ZT_MIXED;
if ((int64_t)st.null_count == st.rows && !expr_src_synthetic(src)) return ZT_MIXED;
int d = zone_sym_decision(st.min, st.max, st.has_minmax, (int64_t)st.null_count, st.rows, cmp_op, cval);
return d < 0 ? ZT_MIXED : d;
}
static int zone_sym_vec_decision(const ray_col_src_t* src, int64_t v, uint16_t cmp_op, int64_t cval) {
int64_t mn, mx; uint32_t nn;
if (!ray_col_src_vec_stat(src, v, &mn, &mx, &nn)) return ZT_MIXED;
int64_t rows = src->len - v * RAY_ENC_VEC;
if (rows > RAY_ENC_VEC) rows = RAY_ENC_VEC;
int d = zone_sym_decision(mn, mx, true, nn, rows, cmp_op, cval);
return d < 0 ? ZT_MIXED : d;
}
static inline bool zone_f64_type(int8_t t) { return t == RAY_F32 || t == RAY_F64; }
static int zone_dir_decision_f64(const ray_col_src_t* src, int64_t ch, uint16_t cmp_op, double cval) {
ray_enc_chunk_stat_t st;
if (!ray_col_src_chunk_stat(src, ch, &st)) return ZT_MIXED;
double mn, mx;
memcpy(&mn, &st.min, 8); memcpy(&mx, &st.max, 8);
int d = zone_cmp_core_f64(mn, mx, st.has_minmax, st.null_count > 0,
(int64_t)st.null_count == st.rows, cmp_op, cval);
return d < 0 ? ZT_MIXED : d;
}
static int zone_vec_decision_f64(const ray_col_src_t* src, int64_t v, uint16_t cmp_op, double cval) {
int64_t bmn, bmx; uint32_t nn;
if (!ray_col_src_vec_stat(src, v, &bmn, &bmx, &nn)) return ZT_MIXED;
double mn, mx;
memcpy(&mn, &bmn, 8); memcpy(&mx, &bmx, 8);
int d = zone_cmp_core_f64(mn, mx, true, nn > 0, false, cmp_op, cval);
return d < 0 ? ZT_MIXED : d;
}
static inline const ray_index_t* zone_fresh_index(const ray_t* col) {
if (!col || !(col->attrs & RAY_ATTR_HAS_INDEX) || !col->index) return NULL;
const ray_index_t* ix = ray_index_payload(col->index);
if (ix->kind != RAY_IDX_CHUNK_ZONE ||
ix->built_for_len != col->len ||
ix->u.chunk_zone.is_f64) return NULL;
return ix;
}
static bool expr_zone_eligible(const ray_expr_t* expr) {
if (expr->out_type != RAY_BOOL) return false;
for (uint8_t r = 0; r < expr->n_regs; r++)
if (expr->regs[r].kind == REG_SCAN &&
(expr->regs[r].enc ? zone_dir_type(expr->regs[r].col_type) || zone_f64_type(expr->regs[r].col_type) ||
expr->regs[r].col_type == RAY_SYM
: zone_fresh_index(expr->regs[r].col_obj) != NULL))
return true;
return false;
}
static int expr_zone_decide(const ray_expr_t* expr, int64_t ms, int64_t me) {
int tri[EXPR_MAX_REGS];
for (uint8_t r = 0; r < expr->n_regs; r++) tri[r] = ZT_NONE;
for (uint8_t i = 0; i < expr->n_ins; i++) {
const expr_ins_t* in = &expr->ins[i];
uint16_t op = in->opcode;
if (op >= OP_EQ && op <= OP_GE && in->src2 != 0xFF) {
const ray_t* col = NULL; int src_reg = -1; int64_t cval = 0; uint16_t cmp = op;
uint8_t a = in->src1, b = in->src2;
if (expr->regs[a].kind == REG_SCAN &&
expr->regs[b].kind == REG_CONST && expr->regs[b].type == RAY_I64) {
col = expr->regs[a].col_obj; src_reg = a; cval = expr->regs[b].const_i64;
} else if (expr->regs[b].kind == REG_SCAN &&
expr->regs[a].kind == REG_CONST && expr->regs[a].type == RAY_I64) {
col = expr->regs[b].col_obj; src_reg = b; cval = expr->regs[a].const_i64;
cmp = zone_swap_op(op);
}
int d = ZT_MIXED;
int fr = -1; double fc = 0; uint16_t fcmp = op;
if (expr->regs[a].kind == REG_SCAN && expr->regs[a].enc && zone_f64_type(expr->regs[a].col_type) &&
expr->regs[a].type == RAY_F64 && expr->regs[b].kind == REG_CONST && expr->regs[b].type == RAY_F64) {
fr = a; fc = expr->regs[b].const_f64;
} else if (expr->regs[b].kind == REG_SCAN && expr->regs[b].enc && zone_f64_type(expr->regs[b].col_type) &&
expr->regs[b].type == RAY_F64 && expr->regs[a].kind == REG_CONST && expr->regs[a].type == RAY_F64) {
fr = b; fc = expr->regs[a].const_f64; fcmp = zone_swap_op(op);
}
if (fr >= 0 && fc == fc) {
if ((ms >> 16) == ((me - 1) >> 16))
d = zone_dir_decision_f64(expr->regs[fr].src, ms >> 16, fcmp, fc);
if (d == ZT_MIXED && ms / RAY_ENC_VEC == (me - 1) / RAY_ENC_VEC)
d = zone_vec_decision_f64(expr->regs[fr].src, ms / RAY_ENC_VEC, fcmp, fc);
}
if (src_reg >= 0 && cval == NULL_I64) { src_reg = -1; col = NULL; }
if (src_reg >= 0 && expr->regs[src_reg].enc && expr->regs[src_reg].col_type == RAY_SYM) {
const ray_col_src_t* ss = expr->regs[src_reg].src;
if ((ms >> 16) == ((me - 1) >> 16))
d = zone_sym_dir_decision(ss, ms >> 16, cmp, cval);
if (d == ZT_MIXED && ms / RAY_ENC_VEC == (me - 1) / RAY_ENC_VEC)
d = zone_sym_vec_decision(ss, ms / RAY_ENC_VEC, cmp, cval);
} else if (src_reg >= 0 && expr->regs[src_reg].enc) {
if (zone_dir_type(expr->regs[src_reg].col_type) && (ms >> 16) == ((me - 1) >> 16))
d = zone_dir_decision(expr->regs[src_reg].src, ms >> 16, cmp, cval);
if (d == ZT_MIXED && zone_dir_type(expr->regs[src_reg].col_type) &&
ms / RAY_ENC_VEC == (me - 1) / RAY_ENC_VEC)
d = zone_vec_decision(expr->regs[src_reg].src, ms / RAY_ENC_VEC, cmp, cval);
} else {
const ray_index_t* ix = zone_fresh_index(col);
if (ix) {
uint8_t log2 = ix->u.chunk_zone.chunk_log2;
int64_t s_ch = ms >> log2, e_ch = (me - 1) >> log2;
if (s_ch == e_ch && (uint32_t)s_ch < ix->u.chunk_zone.n_chunks)
d = zone_cmp_decision(ix, s_ch, cmp, cval);
}
}
tri[in->dst] = d;
} else if (op == OP_AND && in->src2 != 0xFF) {
tri[in->dst] = zt_and(tri[in->src1], tri[in->src2]);
} else if (op == OP_OR && in->src2 != 0xFF) {
tri[in->dst] = zt_or(tri[in->src1], tri[in->src2]);
} else {
tri[in->dst] = ZT_MIXED;
}
}
int res = tri[expr->out_reg];
return (res == 0 || res == 1) ? res : -1;
}
static int64_t leaf_null_lane_i64(int8_t ct, bool nullable) {
if (nullable) return NULL_I64;
switch (ct) {
case RAY_I32: case RAY_DATE: case RAY_TIME: return (int64_t)NULL_I32;
case RAY_I16: return (int64_t)NULL_I16;
default: return NULL_I64;
}
}
static uint8_t leaf_null_result(const ray_expr_t* expr, const expr_ins_t* in, uint8_t scan_reg) {
uint8_t other = in->src1 == scan_reg ? in->src2 : in->src1;
ray_expr_t one;
memset(&one, 0, sizeof one);
one.n_regs = 3; one.n_ins = 1; one.n_scratch = 1; one.out_reg = 2; one.out_type = RAY_BOOL;
one.regs[0].kind = REG_CONST; one.regs[0].type = expr->regs[scan_reg].type;
one.regs[0].const_i64 = leaf_null_lane_i64(expr->regs[scan_reg].col_type, expr->regs[scan_reg].nullable);
one.regs[0].const_f64 = NULL_F64;
one.regs[1].kind = REG_CONST; one.regs[1].type = expr->regs[other].type;
one.regs[1].const_i64 = expr->regs[other].const_i64; one.regs[1].const_f64 = expr->regs[other].const_f64;
one.regs[2].kind = REG_SCRATCH; one.regs[2].type = RAY_BOOL;
one.ins[0] = (expr_ins_t){ .opcode = in->opcode, .dst = 2, .src1 = in->src1 == scan_reg ? 0 : 1,
.src2 = in->src1 == scan_reg ? 1 : 0, .null_aware = in->null_aware };
int64_t lanes[3] = { 0, 0, 0 };
void* sc[3] = { &lanes[0], &lanes[1], &lanes[2] };
ray_t* err = NULL;
const uint8_t* out = (const uint8_t*)expr_eval_morsel(&one, NULL, NULL, sc, NULL, 0, 1, &err);
if (err) ray_error_free(err);
return out ? out[0] : 0;
}
static bool leaf_reg_ok(const ray_expr_t* expr, uint8_t r, uint16_t op) {
int8_t ct = expr->regs[r].col_type;
if (expr->regs[r].type == RAY_F64) return zone_f64_type(ct);
if (expr->regs[r].type != RAY_I64) return false;
if (ct == RAY_SYM) return op == OP_EQ || op == OP_NE;
return zone_dir_type(ct);
}
static const expr_enc_plan_t* expr_enc_plan_build(const ray_expr_t* expr, expr_enc_plan_t* p) {
bool any = false;
for (uint8_t r = 0; r < expr->n_regs; r++) any |= expr->regs[r].kind == REG_SCAN && expr->regs[r].enc;
if (!any) return NULL;
memset(p, 0, sizeof *p);
memset(p->leaf_of_ins, -1, sizeof p->leaf_of_ins);
for (uint8_t r = 0; r < expr->n_regs; r++)
p->load_reg[r] = !(expr->regs[r].kind == REG_SCAN && expr->regs[r].enc);
if (expr->out_type != RAY_BOOL) {
for (uint8_t r = 0; r < expr->n_regs; r++) p->load_reg[r] = true;
return p;
}
for (uint8_t i = 0; i < expr->n_ins; i++) {
const expr_ins_t* in = &expr->ins[i];
uint8_t a = in->src1, b = in->src2;
bool ea = expr->regs[a].kind == REG_SCAN && expr->regs[a].enc;
bool eb = b != 0xFF && expr->regs[b].kind == REG_SCAN && expr->regs[b].enc;
int sreg = -1, creg = -1;
if (in->opcode >= OP_EQ && in->opcode <= OP_GE && b != 0xFF) {
if (ea && expr->regs[b].kind == REG_CONST && expr->regs[b].type == expr->regs[a].type) { sreg = a; creg = b; }
else if (eb && expr->regs[a].kind == REG_CONST && expr->regs[a].type == expr->regs[b].type) { sreg = b; creg = a; }
}
uint16_t op = sreg == a ? in->opcode : zone_swap_op(in->opcode);
if (sreg >= 0 && !leaf_reg_ok(expr, (uint8_t)sreg, op)) sreg = -1;
if (sreg >= 0) {
expr_enc_leaf_t* l = &p->leaf[p->n_leaves];
l->ins = i; l->reg = (uint8_t)sreg; l->op = op;
l->is_f64 = expr->regs[sreg].type == RAY_F64;
l->ci = expr->regs[creg].const_i64; l->cf = expr->regs[creg].const_f64;
l->null_result = leaf_null_result(expr, in, (uint8_t)sreg);
p->leaf_of_ins[i] = (int8_t)p->n_leaves++;
} else {
if (ea) p->load_reg[a] = true;
if (eb) p->load_reg[b] = true;
}
}
return p;
}
_Static_assert(RAY_MORSEL_ELEMS == RAY_ENC_VEC, "a fused morsel is one vector");
static int expr_leaf_on_view(const ray_expr_t* expr, const expr_enc_plan_t* p, uint8_t i,
ray_enc_vview_t* vbuf, void** scratch, void** rptrs,
int64_t ms, int64_t n, ray_t** err) {
const expr_ins_t* in = &expr->ins[i];
const expr_enc_leaf_t* l = &p->leaf[p->leaf_of_ins[i]];
if (ms % RAY_ENC_VEC) return 0;
int vr = ray_col_src_vview(expr->regs[l->reg].src, ms / RAY_ENC_VEC, vbuf, err);
if (vr <= 0) return vr;
if (n != vbuf->n) return 0;
uint8_t* out = (uint8_t*)scratch[in->dst];
int cd = l->is_f64 ? ray_enc_const_decision_f64(vbuf, l->op, l->cf, l->null_result)
: ray_enc_const_decision(vbuf, l->op, l->ci, l->null_result);
if (cd >= 0) memset(out, cd, (size_t)n);
else if (!(l->is_f64 ? ray_enc_eval_cmp_f64(vbuf, l->op, l->cf, l->null_result, out)
: ray_enc_eval_cmp(vbuf, l->op, l->ci, l->null_result, out))) return 0;
rptrs[in->dst] = out;
return 1;
}
static _Atomic(int64_t) g_expr_leaf_runs;
int64_t ray_expr_test_leaf_runs(void) { return atomic_load_explicit(&g_expr_leaf_runs, memory_order_relaxed); }
static bool expr_single_leaf(const ray_expr_t* expr, const expr_enc_plan_t* p) {
return p && p->n_leaves == 1 && expr->n_ins == 1 && p->leaf_of_ins[0] == 0 &&
expr->ins[0].dst == expr->out_reg;
}
static int expr_leaf_emit(const ray_expr_t* expr, const expr_enc_plan_t* p, ray_enc_vview_t* vbuf,
rowsel_builder_t* sb, uint32_t lseg, uint8_t* out, int64_t ms, int64_t n,
ray_t** err) {
const expr_enc_leaf_t* l = &p->leaf[0];
if (ms % RAY_ENC_VEC) return 0;
int vr = ray_col_src_vview(expr->regs[l->reg].src, ms / RAY_ENC_VEC, vbuf, err);
if (vr <= 0) return vr;
if (n != vbuf->n) return 0;
int cd = l->is_f64 ? ray_enc_const_decision_f64(vbuf, l->op, l->cf, l->null_result)
: ray_enc_const_decision(vbuf, l->op, l->ci, l->null_result);
if (cd >= 0) { rowsel_emit_uniform(sb, lseg, (uint8_t)cd, n); return 1; }
uint8_t pass[RAY_ENC_VEC];
if (l->is_f64 ? ray_enc_eval_cmp_runs_f64(vbuf, l->op, l->cf, pass)
: ray_enc_eval_cmp_runs(vbuf, l->op, l->ci, pass)) {
rowsel_emit_runs(sb, lseg, vbuf->run_ends.h, pass, vbuf->n_runs, n);
atomic_fetch_add_explicit(&g_expr_leaf_runs, 1, memory_order_relaxed);
return 1;
}
if (!(l->is_f64 ? ray_enc_eval_cmp_f64(vbuf, l->op, l->cf, l->null_result, out)
: ray_enc_eval_cmp(vbuf, l->op, l->ci, l->null_result, out))) return 0;
rowsel_emit_segment(sb, lseg, out, n);
return 1;
}
typedef struct {
const ray_expr_t* expr;
void* out_data;
int8_t out_type;
rowsel_builder_t* sel_builders;
uint32_t n_sel_builders;
ray_scan_fail_t* fail;
int64_t key_base;
const expr_enc_plan_t* plan;
} expr_full_ctx_t;
static _Atomic(int64_t) g_expr_zone_skips;
int64_t ray_expr_test_zone_skips(void) { return atomic_load_explicit(&g_expr_zone_skips, memory_order_relaxed); }
static void expr_full_block(void* ctx, uint32_t worker_id, int64_t start, int64_t end) {
(void)worker_id;
expr_full_ctx_t* c = (expr_full_ctx_t*)ctx;
const ray_expr_t* expr = c->expr;
uint8_t esz = ray_elem_size(c->out_type);
int64_t key = c->key_base + start;
if (ray_scan_fail_skip(c->fail, key)) return;
ray_t* scratch_hdr = NULL;
char* scratch_mem = (char*)scratch_alloc(&scratch_hdr,
(size_t)(expr->n_regs ? expr->n_regs : 1) * EXPR_MORSEL * 8);
if (!scratch_mem) { ray_scan_fail_note(c->fail, key, ray_error("oom", "fused expression scratch")); return; }
void* scratch[EXPR_MAX_REGS];
for (uint8_t r = 0; r < expr->n_regs; r++)
scratch[r] = scratch_mem + (size_t)r * EXPR_MORSEL * 8;
ray_col_scratch_t rsc_arr[EXPR_MAX_REGS];
ray_col_scratch_t* rsc = NULL;
for (uint8_t r = 0; r < expr->n_regs; r++)
if (expr->regs[r].kind == REG_SCAN && expr->regs[r].enc) { rsc = rsc_arr; break; }
if (rsc) for (uint8_t r = 0; r < expr->n_regs; r++) ray_col_scratch_init(&rsc[r]);
bool failed = false;
bool zone = expr_zone_eligible(expr);
int64_t skips = 0;
const expr_enc_plan_t* plan = c->plan && c->plan->n_leaves ? c->plan : NULL;
ray_enc_vview_t vbuf;
const bool leaf1 = expr_single_leaf(expr, plan);
rowsel_builder_t* sb = NULL;
uint32_t local_base_seg = 0;
if (c->sel_builders) {
const int64_t grain = (int64_t)RAY_DISPATCH_MORSELS * RAY_MORSEL_ELEMS;
_Static_assert((RAY_DISPATCH_MORSELS * RAY_MORSEL_ELEMS) % RAY_MORSEL_ELEMS == 0,
"TASK_GRAIN must be a whole number of morsels");
assert(start % grain == 0 &&
"fused-sel: task start not grain/morsel aligned");
uint32_t slot = (uint32_t)(start / grain);
assert(slot < c->n_sel_builders && "fused-sel: builder slot OOB");
sb = &c->sel_builders[slot];
local_base_seg = (uint32_t)(start / RAY_MORSEL_ELEMS);
}
for (int64_t ms = start; ms < end; ms += EXPR_MORSEL) {
int64_t me = (ms + EXPR_MORSEL < end) ? ms + EXPR_MORSEL : end;
if (zone) {
int d = expr_zone_decide(expr, ms, me);
if (d >= 0) {
skips++;
if (sb) {
uint32_t lseg = (uint32_t)(ms / RAY_MORSEL_ELEMS) - local_base_seg;
memset(scratch[0], d, (size_t)(me - ms));
rowsel_emit_segment(sb, lseg, (const uint8_t*)scratch[0], me - ms);
} else {
memset((char*)c->out_data + ms * esz, d, (size_t)(me - ms) * esz);
}
continue;
}
}
if (sb && leaf1 && !failed) {
ray_t* err = NULL;
uint32_t lseg = (uint32_t)(ms / RAY_MORSEL_ELEMS) - local_base_seg;
int lr = expr_leaf_emit(expr, plan, &vbuf, sb, lseg, (uint8_t*)scratch[0], ms, me - ms, &err);
if (lr > 0) continue;
if (lr < 0) { ray_scan_fail_note(c->fail, key, err); failed = true; }
}
void* result = NULL;
if (!failed) {
ray_t* err = NULL;
result = expr_eval_morsel(expr, plan, plan ? &vbuf : NULL, scratch, rsc, ms, me, &err);
if (err) { ray_scan_fail_note(c->fail, key, err); failed = true; result = NULL; }
}
if (sb) {
uint32_t lseg = (uint32_t)(ms / RAY_MORSEL_ELEMS) - local_base_seg;
if (!result) { memset(scratch[0], 0, (size_t)(me - ms)); result = scratch[0]; }
rowsel_emit_segment(sb, lseg, (const uint8_t*)result, me - ms);
} else if (result) {
memcpy((char*)c->out_data + ms * esz, result, (size_t)(me - ms) * esz);
}
}
if (rsc) for (uint8_t r = 0; r < expr->n_regs; r++) ray_col_scratch_free(&rsc[r]);
scratch_free(scratch_hdr);
if (skips) atomic_fetch_add_explicit(&g_expr_zone_skips, skips, memory_order_relaxed);
}
static void expr_full_fn(void* ctx, uint32_t worker_id, int64_t start, int64_t end) {
const expr_full_ctx_t* c = (const expr_full_ctx_t*)ctx;
const int64_t grain = (int64_t)RAY_DISPATCH_MORSELS * RAY_MORSEL_ELEMS;
if (!c->sel_builders || c->n_sel_builders <= 1 || end - start <= grain) {
expr_full_block(ctx, worker_id, start, end);
return;
}
for (int64_t b = start; b < end; b += grain)
expr_full_block(ctx, worker_id, b, b + grain < end ? b + grain : end);
}
static void mark_i64_overflow_as_null(ray_t* result, int64_t off, int64_t len) {
int64_t* d = (int64_t*)ray_data(result) + off;
for (int64_t i = 0; i < len; i++) {
if (RAY_UNLIKELY(d[i] == NULL_I64)) {
ray_vec_set_null(result, off + i, true);
}
}
}
static void expr_flag_output_nulls(const ray_expr_t* expr, ray_t* out, int64_t nrows) {
uint8_t o = expr->out_reg;
if (!expr->regs[o].nullable) return;
if (expr->regs[o].null_src) { out->attrs |= RAY_ATTR_HAS_NULLS; return; }
if (expr->out_type == RAY_I64) mark_i64_overflow_as_null(out, 0, nrows);
else if (expr->out_type == RAY_F64) mark_f64_nonfinite_as_null(out, 0, nrows);
}
static bool g_expr_test_parted_all_segs;
void ray_expr_test_parted_all_segments(bool on) { g_expr_test_parted_all_segs = on; }
static ray_t* expr_seg(ray_t* parted_col, int64_t s) {
if (parted_col->type == RAY_ENCCOL) return ray_enccol_seg(parted_col, s);
return ((ray_t**)ray_data(parted_col))[s];
}
static int64_t expr_nsegs(ray_t* parted_col) {
return parted_col->type == RAY_ENCCOL ? ray_enccol_of(parted_col)->wrapper->len : parted_col->len;
}
static ray_t* expr_eval_full_parted(const ray_expr_t* expr, int64_t nrows) {
ray_t* out = ray_vec_new(expr->out_type, nrows);
if (!out || RAY_IS_ERR(out)) {
return out;
}
out->len = nrows;
ray_t* ref_parted = NULL;
for (uint8_t r = 0; r < expr->n_regs; r++) {
if (expr->regs[r].is_parted) {
ref_parted = expr->regs[r].parted_col;
break;
}
}
if (!ref_parted) { ray_release(out); return ray_error("nyi", NULL); }
int64_t n_segs = expr_nsegs(ref_parted);
uint8_t esz = ray_elem_size(expr->out_type);
ray_pool_t* pool = ray_pool_get();
int64_t global_off = 0;
ray_scan_fail_t fail; ray_scan_fail_init(&fail);
bool dispatched = false;
for (int64_t s = 0; s < n_segs; s++) {
int64_t seg_len = 0;
for (uint8_t r = 0; r < expr->n_regs; r++) {
if (expr->regs[r].is_parted) {
ray_t* seg = expr_seg(expr->regs[r].parted_col, s);
if (seg) { seg_len = ray_table_col_nrows(seg); break; }
}
}
if (seg_len <= 0) continue;
ray_expr_t seg_expr = *expr;
ray_col_src_t seg_src[EXPR_MAX_REGS];
uint8_t n_seg_src = 0;
bool seg_ok = true;
for (uint8_t r = 0; r < seg_expr.n_regs; r++) {
if (!seg_expr.regs[r].is_parted) continue;
ray_t* seg = expr_seg(seg_expr.regs[r].parted_col, s);
if (!seg) { seg_ok = false; break; }
int64_t rl = ray_table_col_nrows(seg);
if (rl != seg_len) {
ray_scan_fail_note(&fail, global_off,
ray_error("length", "parted expression: segment %lld has %lld rows, expected %lld",
(long long)s, (long long)rl, (long long)seg_len));
seg_ok = false; break;
}
ray_t* segp = seg;
if (seg->type == RAY_ENCCOL) {
ray_t* pk = ray_enccol_decoded_peek(seg);
if (pk) segp = pk;
}
seg_expr.regs[r].is_parted = false;
if (segp->type == RAY_ENCCOL) {
ray_t* be = ray_col_src_bind(seg, &seg_src[n_seg_src]);
if (be) { ray_scan_fail_note(&fail, global_off, be); seg_ok = false; break; }
seg_expr.regs[r].src = &seg_src[n_seg_src++];
seg_expr.regs[r].enc = true;
seg_expr.regs[r].data = NULL;
seg_expr.regs[r].col_attrs = seg_expr.regs[r].src->attrs;
ray_col_src_advise(seg_expr.regs[r].src, 0, seg_len);
} else {
seg_expr.regs[r].enc = false;
seg_expr.regs[r].data = ray_data(segp);
seg_expr.regs[r].col_attrs = segp->attrs;
}
}
if (seg_ok) {
expr_enc_plan_t plan;
expr_full_ctx_t ctx = {
.expr = &seg_expr,
.out_data = (char*)ray_data(out) + global_off * esz,
.out_type = expr->out_type,
.fail = &fail,
.key_base = global_off,
.plan = expr_enc_plan_build(&seg_expr, &plan),
};
if (pool && seg_len >= RAY_PARALLEL_THRESHOLD) {
ray_pool_dispatch(pool, expr_full_fn, &ctx, seg_len);
dispatched = true;
} else
expr_full_fn(&ctx, 0, 0, seg_len);
} else if (!ray_scan_failed(&fail)) {
memset((char*)ray_data(out) + global_off * esz, 0,
(size_t)seg_len * esz);
}
for (uint8_t i = 0; i < n_seg_src; i++) ray_col_src_release(&seg_src[i]);
global_off += seg_len;
if (ray_scan_failed(&fail) && !g_expr_test_parted_all_segs) break;
if (ray_interrupted() || (pool && atomic_load_explicit(&pool->cancelled, memory_order_acquire)))
break;
}
ray_t* ferr = ray_scan_fail_finish(&fail, dispatched ? pool : NULL);
if (ferr) { ray_release(out); return ferr; }
expr_flag_output_nulls(expr, out, nrows);
return out;
}
ray_t* expr_eval_full(const ray_expr_t* expr, int64_t nrows) {
if (expr->has_parted)
return expr_eval_full_parted(expr, nrows);
ray_t* out = ray_vec_new(expr->out_type, nrows);
if (!out || RAY_IS_ERR(out)) return out;
out->len = nrows;
ray_scan_fail_t fail; ray_scan_fail_init(&fail);
expr_enc_plan_t plan;
expr_full_ctx_t ctx = {
.expr = expr, .out_data = ray_data(out), .out_type = expr->out_type, .fail = &fail,
.plan = expr_enc_plan_build(expr, &plan),
};
for (uint8_t r = 0; r < expr->n_regs; r++)
if (expr->regs[r].kind == REG_SCAN && expr->regs[r].enc)
ray_col_src_advise(expr->regs[r].src, 0, nrows);
ray_pool_t* pool = ray_pool_get();
bool par = pool && nrows >= RAY_PARALLEL_THRESHOLD;
if (par)
ray_pool_dispatch(pool, expr_full_fn, &ctx, nrows);
else
expr_full_fn(&ctx, 0, 0, nrows);
ray_t* ferr = ray_scan_fail_finish(&fail, par ? pool : NULL);
if (ferr) { ray_release(out); return ferr; }
expr_flag_output_nulls(expr, out, nrows);
return out;
}
void ray_expr_test_eval_range(const ray_expr_t* expr, int64_t start, int64_t end,
uint8_t* out, ray_t** err) {
*err = NULL;
ray_t* hdr = NULL;
char* mem = (char*)scratch_alloc(&hdr, (size_t)(expr->n_regs ? expr->n_regs : 1) * EXPR_MORSEL * 8);
if (!mem) { *err = ray_error("oom", NULL); return; }
void* scratch[EXPR_MAX_REGS];
ray_col_scratch_t rsc[EXPR_MAX_REGS];
for (uint8_t r = 0; r < expr->n_regs; r++) { scratch[r] = mem + (size_t)r * EXPR_MORSEL * 8; ray_col_scratch_init(&rsc[r]); }
void* res = expr_eval_morsel(expr, NULL, NULL, scratch, rsc, start, end, err);
if (res && !*err) memcpy(out, res, (size_t)(end - start));
for (uint8_t r = 0; r < expr->n_regs; r++) ray_col_scratch_free(&rsc[r]);
scratch_free(hdr);
}
static rowsel_builder_t* pred_sel_builders_alloc(int64_t nrows, ray_pool_t* pool,
bool* parallel,
uint32_t* n_builders_out) {
const int64_t grain = (int64_t)RAY_DISPATCH_MORSELS * RAY_MORSEL_ELEMS;
int64_t n_segs = (nrows + RAY_MORSEL_ELEMS - 1) / RAY_MORSEL_ELEMS;
bool par = (pool && nrows >= RAY_PARALLEL_THRESHOLD);
uint32_t n_builders = par ? (uint32_t)((nrows + grain - 1) / grain) : 1u;
rowsel_builder_t* builders =
(rowsel_builder_t*)ray_alloc_raw((size_t)n_builders * sizeof(rowsel_builder_t));
if (!builders) return NULL;
if (par) {
for (uint32_t i = 0; i < n_builders; i++) {
int64_t s = (int64_t)i * grain;
int64_t e = s + grain;
if (e > nrows) e = nrows;
uint32_t segs = (uint32_t)((e - s + RAY_MORSEL_ELEMS - 1) / RAY_MORSEL_ELEMS);
rowsel_builder_init(&builders[i], segs);
}
} else {
rowsel_builder_init(&builders[0], (uint32_t)n_segs);
}
*parallel = par;
*n_builders_out = n_builders;
return builders;
}
static ray_t* pred_sel_builders_finish(rowsel_builder_t* builders,
uint32_t n_builders, int64_t nrows,
bool* all_pass) {
ray_t* block = rowsel_builder_finish(builders, n_builders, nrows);
ray_free_raw(builders);
if (!block) return NULL;
ray_rowsel_t* m = ray_rowsel_meta(block);
if (m->total_pass == nrows) {
ray_rowsel_release(block);
*all_pass = true;
return NULL;
}
return block;
}
typedef struct {
pred_sel_fill_fn fill;
void* fill_ctx;
rowsel_builder_t* sel_builders;
uint32_t n_sel_builders;
ray_scan_fail_t* fail;
} pred_sel_drive_ctx_t;
static void pred_sel_block(void* vctx, uint32_t worker_id,
int64_t start, int64_t end) {
(void)worker_id;
pred_sel_drive_ctx_t* c = (pred_sel_drive_ctx_t*)vctx;
if (c->fail && ray_scan_fail_skip(c->fail, start)) return;
ray_t* scratch_hdr = NULL;
uint8_t* scratch = (uint8_t*)scratch_alloc(&scratch_hdr, RAY_MORSEL_ELEMS);
if (!scratch) {
if (c->fail) ray_scan_fail_note(c->fail, start, ray_error("oom", "selection scratch"));
return;
}
ray_col_scratch_t sc;
ray_col_scratch_init(&sc);
const int64_t grain = (int64_t)RAY_DISPATCH_MORSELS * RAY_MORSEL_ELEMS;
_Static_assert((RAY_DISPATCH_MORSELS * RAY_MORSEL_ELEMS) % RAY_MORSEL_ELEMS == 0,
"TASK_GRAIN must be a whole number of morsels");
assert(start % grain == 0 && "pred-sel: task start not grain/morsel aligned");
uint32_t slot = (uint32_t)(start / grain);
assert(slot < c->n_sel_builders && "pred-sel: builder slot OOB");
rowsel_builder_t* sb = &c->sel_builders[slot];
uint32_t local_base_seg = (uint32_t)(start / RAY_MORSEL_ELEMS);
for (int64_t ms = start; ms < end; ms += RAY_MORSEL_ELEMS) {
int64_t me = (ms + RAY_MORSEL_ELEMS < end) ? ms + RAY_MORSEL_ELEMS : end;
c->fill(c->fill_ctx, &sc, ms, me - ms, scratch);
uint32_t lseg = (uint32_t)(ms / RAY_MORSEL_ELEMS) - local_base_seg;
rowsel_emit_segment(sb, lseg, scratch, me - ms);
}
ray_col_scratch_free(&sc);
scratch_free(scratch_hdr);
}
static void pred_sel_worker(void* vctx, uint32_t worker_id, int64_t start, int64_t end) {
const pred_sel_drive_ctx_t* c = (const pred_sel_drive_ctx_t*)vctx;
const int64_t grain = (int64_t)RAY_DISPATCH_MORSELS * RAY_MORSEL_ELEMS;
if (c->n_sel_builders <= 1 || end - start <= grain) { pred_sel_block(vctx, worker_id, start, end); return; }
for (int64_t b = start; b < end; b += grain)
pred_sel_block(vctx, worker_id, b, b + grain < end ? b + grain : end);
}
ray_t* pred_sel_drive(int64_t nrows, pred_sel_fill_fn fill, void* fill_ctx,
ray_scan_fail_t* fail, bool* all_pass) {
*all_pass = false;
if (nrows <= 0) { *all_pass = true; return NULL; }
ray_pool_t* pool = ray_pool_get();
bool parallel;
uint32_t n_builders;
rowsel_builder_t* builders =
pred_sel_builders_alloc(nrows, pool, ¶llel, &n_builders);
if (!builders) return NULL;
pred_sel_drive_ctx_t ctx = {
.fill = fill,
.fill_ctx = fill_ctx,
.sel_builders = builders,
.n_sel_builders = n_builders,
.fail = fail,
};
if (parallel)
ray_pool_dispatch(pool, pred_sel_worker, &ctx, nrows);
else
pred_sel_worker(&ctx, 0, 0, nrows);
bool ap = false;
ray_t* block = pred_sel_builders_finish(builders, n_builders, nrows, &ap);
ray_t* ferr = fail ? ray_scan_fail_finish(fail, parallel ? pool : NULL) : NULL;
if (ferr) {
if (block) ray_rowsel_release(block);
return ferr;
}
*all_pass = ap;
return block;
}
ray_t* exec_pred_to_selection(ray_graph_t* g, ray_op_t* pred, int64_t nrows,
bool* all_pass) {
*all_pass = false;
if (!g || !g->table || !pred) return NULL;
if (nrows <= 0) { *all_pass = true; return NULL; }
if (pred->opcode == OP_EQ || pred->opcode == OP_NE) {
ray_t* ss = exec_str_to_selection(g, pred, nrows, all_pass);
if (ss || *all_pass) return ss;
}
if (pred->opcode == OP_IN || pred->opcode == OP_NOT_IN)
return exec_in_to_selection(g, pred, nrows, all_pass);
ray_expr_t ex;
ray_t* cerr = NULL;
if (!expr_compile(g, g->table, pred, &ex, &cerr)) return cerr;
if (ex.out_type != RAY_BOOL || ex.has_parted) return NULL;
ray_pool_t* pool = ray_pool_get();
bool parallel;
uint32_t n_builders;
rowsel_builder_t* builders =
pred_sel_builders_alloc(nrows, pool, ¶llel, &n_builders);
if (!builders) return NULL;
ray_scan_fail_t fail; ray_scan_fail_init(&fail);
expr_enc_plan_t plan;
expr_full_ctx_t ctx = {
.expr = &ex,
.out_data = NULL,
.out_type = ex.out_type,
.sel_builders = builders,
.n_sel_builders = n_builders,
.fail = &fail,
.plan = expr_enc_plan_build(&ex, &plan),
};
for (uint8_t r = 0; r < ex.n_regs; r++)
if (ex.regs[r].kind == REG_SCAN && ex.regs[r].enc)
ray_col_src_advise(ex.regs[r].src, 0, nrows);
if (parallel)
ray_pool_dispatch(pool, expr_full_fn, &ctx, nrows);
else
expr_full_fn(&ctx, 0, 0, nrows);
bool ap = false;
ray_t* block = pred_sel_builders_finish(builders, n_builders, nrows, &ap);
ray_t* ferr = ray_scan_fail_finish(&fail, parallel ? pool : NULL);
if (ferr) {
if (block) ray_rowsel_release(block);
return ferr;
}
*all_pass = ap;
return block;
}
static ray_t* refine_emit_mix(rowsel_builder_t* b, uint32_t seg, const uint16_t* ix, uint32_t cnt,
const uint8_t* res, int64_t n) {
b->seg_off[seg] = b->idx_len;
b->seg_flags[seg] = RAY_SEL_NONE;
b->seg_off[seg + 1] = b->idx_len;
if (b->idx_len + cnt > b->idx_cap) {
uint32_t newcap = b->idx_cap ? b->idx_cap : 8u;
while (newcap < b->idx_len + cnt) newcap *= 2u;
uint16_t* ni = (uint16_t*)ray_realloc_raw(b->idx, (size_t)newcap * sizeof(uint16_t));
if (!ni) return ray_error("oom", "refinement selection");
b->idx = ni;
b->idx_cap = newcap;
}
uint32_t pc = 0;
for (uint32_t k = 0; k < cnt; k++) {
if (ix[k] >= n)
return ray_error("corrupt", "refinement: selection row %u past its segment", (unsigned)ix[k]);
if (!res || res[ix[k]]) b->idx[b->idx_len + pc++] = ix[k];
}
b->total_pass += pc;
if (pc == 0) b->seg_flags[seg] = RAY_SEL_NONE;
else if ((int64_t)pc == n) b->seg_flags[seg] = RAY_SEL_ALL;
else { b->seg_flags[seg] = RAY_SEL_MIX; b->idx_len += pc; }
b->seg_off[seg + 1] = b->idx_len;
return NULL;
}
typedef struct {
const ray_expr_t* expr;
const expr_enc_plan_t* plan;
const uint8_t* flags;
const uint32_t* offs;
const uint16_t* idx;
rowsel_builder_t* sel_builders;
uint32_t n_sel_builders;
int64_t nrows, n_segs;
ray_scan_fail_t* fail;
} expr_refine_ctx_t;
static void expr_refine_task(expr_refine_ctx_t* c, uint32_t t) {
const ray_expr_t* expr = c->expr;
const int64_t s0 = c->n_segs * t / c->n_sel_builders, s1 = c->n_segs * (t + 1) / c->n_sel_builders;
const int64_t start = s0 * RAY_MORSEL_ELEMS;
const int64_t end = s1 * RAY_MORSEL_ELEMS < c->nrows ? s1 * RAY_MORSEL_ELEMS : c->nrows;
rowsel_builder_t* sb = &c->sel_builders[t];
const uint32_t base_seg = (uint32_t)s0;
if (s0 >= s1) return;
if (ray_scan_fail_skip(c->fail, start)) {
for (int64_t seg = s0; seg < s1; seg++) {
const uint32_t ls = (uint32_t)(seg - s0);
sb->seg_off[ls] = sb->idx_len; sb->seg_flags[ls] = RAY_SEL_NONE; sb->seg_off[ls + 1] = sb->idx_len;
}
return;
}
ray_t* scratch_hdr = NULL;
char* scratch_mem = (char*)scratch_alloc(&scratch_hdr,
(size_t)(expr->n_regs ? expr->n_regs : 1) * EXPR_MORSEL * 8);
bool failed = false;
if (!scratch_mem) { ray_scan_fail_note(c->fail, start, ray_error("oom", "fused expression scratch")); failed = true; }
void* scratch[EXPR_MAX_REGS];
for (uint8_t r = 0; r < expr->n_regs && scratch_mem; r++) scratch[r] = scratch_mem + (size_t)r * EXPR_MORSEL * 8;
ray_col_scratch_t rsc_arr[EXPR_MAX_REGS];
ray_col_scratch_t* rsc = NULL;
for (uint8_t r = 0; r < expr->n_regs; r++)
if (expr->regs[r].kind == REG_SCAN && expr->regs[r].enc) { rsc = rsc_arr; break; }
if (rsc) for (uint8_t r = 0; r < expr->n_regs; r++) ray_col_scratch_init(&rsc[r]);
const expr_enc_plan_t* plan = c->plan && c->plan->n_leaves ? c->plan : NULL;
ray_enc_vview_t vbuf;
for (int64_t ms = start; ms < end; ms += EXPR_MORSEL) {
int64_t me = ms + EXPR_MORSEL < end ? ms + EXPR_MORSEL : end;
const int64_t n = me - ms;
const uint32_t seg = (uint32_t)(ms / RAY_MORSEL_ELEMS);
const uint8_t f = c->flags[seg];
if (f == RAY_SEL_NONE || failed) {
const uint32_t ls = seg - base_seg;
sb->seg_off[ls] = sb->idx_len;
sb->seg_flags[ls] = RAY_SEL_NONE;
sb->seg_off[ls + 1] = sb->idx_len;
continue;
}
const uint8_t* res = NULL;
ray_t* err = NULL;
res = (const uint8_t*)expr_eval_morsel(expr, plan, plan ? &vbuf : NULL, scratch, rsc, ms, me, &err);
if (err) { ray_scan_fail_note(c->fail, start, err); failed = true; res = NULL; }
const uint32_t ls = seg - base_seg;
if (!res) {
sb->seg_off[ls] = sb->idx_len;
sb->seg_flags[ls] = RAY_SEL_NONE;
sb->seg_off[ls + 1] = sb->idx_len;
continue;
}
if (f == RAY_SEL_MIX) {
if ((err = refine_emit_mix(sb, ls, c->idx + c->offs[seg], c->offs[seg + 1] - c->offs[seg], res, n)) != NULL) {
ray_scan_fail_note(c->fail, start, err);
failed = true;
}
continue;
}
rowsel_emit_segment(sb, ls, res, n);
}
if (rsc) for (uint8_t r = 0; r < expr->n_regs; r++) ray_col_scratch_free(&rsc[r]);
scratch_free(scratch_hdr);
}
static void expr_refine_fn(void* ctx, uint32_t worker_id, int64_t start, int64_t end) {
(void)worker_id;
for (int64_t t = start; t < end; t++) expr_refine_task((expr_refine_ctx_t*)ctx, (uint32_t)t);
}
ray_t* exec_pred_refine_selection(ray_graph_t* g, ray_op_t* pred, ray_t* sel, bool* all_pass) {
*all_pass = false;
if (!g || !g->table || !pred || !sel) return NULL;
ray_rowsel_t* m = ray_rowsel_meta(sel);
const int64_t nrows = m->nrows;
if (nrows <= 0 || nrows != ray_table_nrows(g->table)) return NULL;
if (pred->opcode == OP_EQ || pred->opcode == OP_NE) {
ray_t* ss = exec_str_refine_selection(g, pred, sel, all_pass);
if (ss || *all_pass) return ss;
}
ray_expr_t ex;
ray_t* cerr = NULL;
if (!expr_compile(g, g->table, pred, &ex, &cerr)) return cerr;
if (ex.out_type != RAY_BOOL || ex.has_parted) return NULL;
ray_pool_t* pool = ray_pool_get();
const int64_t n_segs = (nrows + RAY_MORSEL_ELEMS - 1) / RAY_MORSEL_ELEMS;
const bool parallel = ray_pool_par_dispatch_ok(pool, nrows, RAY_PARALLEL_THRESHOLD);
int64_t ntask = parallel ? (int64_t)ray_pool_total_workers(pool) * 4 : 1;
if (ntask > n_segs) ntask = n_segs;
const uint32_t n_builders = (uint32_t)(ntask > 0 ? ntask : 1);
rowsel_builder_t* builders = (rowsel_builder_t*)ray_alloc_raw((size_t)n_builders * sizeof(rowsel_builder_t));
if (!builders) return NULL;
for (uint32_t t = 0; t < n_builders; t++)
rowsel_builder_init(&builders[t], (uint32_t)(n_segs * (t + 1) / n_builders - n_segs * t / n_builders));
ray_scan_fail_t fail; ray_scan_fail_init(&fail);
expr_enc_plan_t plan;
expr_refine_ctx_t ctx = { .expr = &ex, .plan = expr_enc_plan_build(&ex, &plan),
.flags = ray_rowsel_flags(sel), .offs = ray_rowsel_offsets(sel), .idx = ray_rowsel_idx(sel),
.sel_builders = builders, .n_sel_builders = n_builders, .nrows = nrows, .n_segs = n_segs, .fail = &fail };
if (n_builders > 1) ray_pool_dispatch_n(pool, expr_refine_fn, &ctx, n_builders);
else expr_refine_fn(&ctx, 0, 0, 1);
bool ap = false;
ray_t* block = pred_sel_builders_finish(builders, n_builders, nrows, &ap);
ray_t* ferr = ray_scan_fail_finish(&fail, parallel ? pool : NULL);
if (ferr) { if (block) ray_rowsel_release(block); return ferr; }
*all_pass = ap;
return block;
}
static void propagate_nulls(ray_t* src, ray_t* dst, int64_t len) {
if (!ray_vec_may_have_nulls(src)) return;
for (int64_t i = 0; i < len; i++) {
if (ray_vec_is_null(src, i))
ray_vec_set_null(dst, i, true);
}
}
static bool op_propagates_null(uint16_t opc) {
return opc < OP_EQ || opc > OP_OR;
}
static bool scalar_is_null(ray_t* x) {
if (ray_is_atom(x)) return RAY_ATOM_IS_NULL(x);
return ray_vec_is_null(x, 0);
}
static bool vec_may_have_nulls(ray_t* v) {
return ray_vec_may_have_nulls(v);
}
static int8_t ew_value_base_type(ray_t* x) {
int8_t t = ray_is_atom(x) ? (int8_t)(-(int)x->type) : x->type;
if (RAY_IS_PARTED(t)) t = (int8_t)RAY_PARTED_BASETYPE(t);
return t;
}
static bool ew_pow_type_admitted(int8_t t) {
return t == RAY_BOOL || t == RAY_U8 || t == RAY_I16 ||
t == RAY_I32 || t == RAY_I64 || t == RAY_F32 || t == RAY_F64;
}
static void* resolve_vec_data(ray_t* v, int64_t* offset) {
if (v->attrs & RAY_ATTR_SLICE) {
*offset += v->slice_offset;
return ray_data(v->slice_parent);
}
return ray_data(v);
}
static void fix_null_comparisons(ray_t* lhs, ray_t* rhs, ray_t* result,
bool l_scalar, bool r_scalar, int64_t len,
uint16_t opcode) {
uint8_t* dst = (uint8_t*)ray_data(result);
bool ln_s = l_scalar && scalar_is_null(lhs);
bool rn_s = r_scalar && scalar_is_null(rhs);
bool l_has = !l_scalar && vec_may_have_nulls(lhs);
bool r_has = !r_scalar && vec_may_have_nulls(rhs);
if (!ln_s && !rn_s && !l_has && !r_has) return;
if (!ln_s && !rn_s && (l_has ^ r_has)) {
ray_t* src = l_has ? lhs : rhs;
bool src_left = l_has;
uint8_t left_bits = (opcode == OP_LT || opcode == OP_LE || opcode == OP_NE);
uint8_t right_bits = (opcode == OP_GT || opcode == OP_GE || opcode == OP_NE);
uint8_t fill = src_left ? left_bits : right_bits;
for (int64_t i = 0; i < len; i++) {
if (ray_vec_is_null(src, i)) dst[i] = fill;
}
return;
}
for (int64_t i = 0; i < len; i++) {
bool ln = ln_s || (l_has && ray_vec_is_null(lhs, i));
bool rn = rn_s || (r_has && ray_vec_is_null(rhs, i));
if (!ln && !rn) continue;
if (ln && rn) {
dst[i] = (opcode == OP_EQ || opcode == OP_LE || opcode == OP_GE) ? 1 : 0;
continue;
}
if (ln) {
dst[i] = (opcode == OP_LT || opcode == OP_LE || opcode == OP_NE) ? 1 : 0;
continue;
}
dst[i] = (opcode == OP_GT || opcode == OP_GE || opcode == OP_NE) ? 1 : 0;
}
}
static void set_all_null(ray_t* result, int64_t len) {
result->attrs |= RAY_ATTR_HAS_NULLS;
switch (result->type) {
case RAY_F64: {
double* d = (double*)ray_data(result);
for (int64_t i = 0; i < len; i++) d[i] = NULL_F64;
break;
}
case RAY_I64: case RAY_TIMESTAMP: {
int64_t* d = (int64_t*)ray_data(result);
for (int64_t i = 0; i < len; i++) d[i] = NULL_I64;
break;
}
case RAY_F32: {
float* d = (float*)ray_data(result);
for (int64_t i = 0; i < len; i++) d[i] = NULL_F32;
break;
}
case RAY_I32: case RAY_DATE: case RAY_TIME: {
int32_t* d = (int32_t*)ray_data(result);
for (int64_t i = 0; i < len; i++) d[i] = NULL_I32;
break;
}
case RAY_I16: {
int16_t* d = (int16_t*)ray_data(result);
for (int64_t i = 0; i < len; i++) d[i] = NULL_I16;
break;
}
case RAY_STR: {
ray_str_t* s = (ray_str_t*)ray_data(result);
memset(s, 0, (size_t)len * sizeof(ray_str_t));
break;
}
case RAY_GUID:
memset(ray_data(result), 0, (size_t)len * 16);
break;
default: break;
}
}
static void propagate_nulls_binary(ray_t* lhs, ray_t* rhs, ray_t* result,
bool l_scalar, bool r_scalar, int64_t len) {
if (l_scalar && scalar_is_null(lhs)) {
set_all_null(result, len);
} else if (r_scalar && scalar_is_null(rhs)) {
set_all_null(result, len);
} else {
if (!l_scalar && vec_may_have_nulls(lhs)) propagate_nulls(lhs, result, len);
if (!r_scalar && vec_may_have_nulls(rhs)) propagate_nulls(rhs, result, len);
}
}
static inline int8_t ew_runtime_out_type(int8_t out_type, int8_t operand_type) {
if (RAY_IS_PARTED(out_type)) return (int8_t)RAY_PARTED_BASETYPE(out_type);
if (out_type == RAY_MAPCOMMON) return operand_type;
return out_type;
}
static bool ew_unary_math_type_admitted(uint16_t opcode, int8_t t) {
if (opcode != OP_SIGNUM && !expr_is_f64_unary_math(opcode)) return true;
if (RAY_IS_PARTED(t)) t = (int8_t)RAY_PARTED_BASETYPE(t);
return t == RAY_BOOL || t == RAY_U8 || t == RAY_I16 ||
t == RAY_I32 || t == RAY_I64 || t == RAY_F32 || t == RAY_F64;
}
ray_t* exec_elementwise_unary(ray_graph_t* g, ray_op_t* op, ray_t* input) {
if (input && !RAY_IS_ERR(input) && (ray_has_timespan(input) || op->out_type == RAY_TIMESPAN)) {
if (op->opcode == OP_CAST) {
const char* name = ray_type_name(op->out_type);
ray_t* target = ray_sym(ray_sym_intern(name, strlen(name)));
if (RAY_IS_ERR(target)) return target;
ray_t* result = ray_cast_fn(target, input);
ray_release(target);
return result;
}
if (op->opcode == OP_NEG) return ray_neg_fn(input);
if (op->opcode == OP_ABS) return ray_abs_fn(input);
if (op->opcode != OP_ISNULL) return ray_error("type", "unsupported duration unary operation");
}
(void)g;
if (!input || RAY_IS_ERR(input)) return input;
if (!ray_is_vec(input)) return ray_error("type", "expr eval: unary op expects a vector, got %s", ray_type_name(input->type));
int64_t len = input->len;
int8_t in_type = input->type;
int8_t out_type = ew_runtime_out_type(op->out_type, in_type);
if (!ew_unary_math_type_admitted(op->opcode, in_type)) {
return ray_error("type", "%s: expects a numeric argument, got %s",
ray_opcode_name(op->opcode), ray_type_name(input->type));
}
ray_t* result = ray_vec_new(out_type, len);
if (!result || RAY_IS_ERR(result)) return result;
result->len = len;
if (op->opcode == OP_ISNULL) {
uint8_t* dst = (uint8_t*)ray_data(result);
if (input->type != RAY_STR && input->type != RAY_SYM &&
!vec_may_have_nulls(input)) {
memset(dst, 0, (size_t)len);
} else {
for (int64_t i = 0; i < len; i++)
dst[i] = ray_vec_is_null(input, i) ? 1 : 0;
}
return result;
}
ray_morsel_t m;
ray_morsel_init(&m, input);
int64_t out_off = 0;
uint16_t opc = op->opcode;
if (opc == OP_CAST && out_type == RAY_F32) {
if (in_type != RAY_F64 && in_type != RAY_F32 &&
in_type != RAY_I64 && in_type != RAY_TIMESTAMP &&
in_type != RAY_I32 && in_type != RAY_DATE && in_type != RAY_TIME &&
in_type != RAY_I16 && in_type != RAY_U8 && in_type != RAY_BOOL) {
ray_release(result);
return ray_error("type", "cast: cannot convert %s to F32",
ray_type_name(in_type));
}
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
float* dst = (float*)ray_data(result) + out_off;
if (in_type == RAY_F64) {
const double* src = (const double*)m.morsel_ptr;
for (int64_t i = 0; i < n; i++) dst[i] = ray_cast_f64_to_f32_null(src[i]);
} else if (in_type == RAY_F32) {
memcpy(dst, m.morsel_ptr, (size_t)n * sizeof(float));
} else if (in_type == RAY_I64 || in_type == RAY_TIMESTAMP) {
const int64_t* src = (const int64_t*)m.morsel_ptr;
for (int64_t i = 0; i < n; i++) dst[i] = (float)src[i];
} else if (in_type == RAY_I32 || in_type == RAY_DATE || in_type == RAY_TIME) {
const int32_t* src = (const int32_t*)m.morsel_ptr;
for (int64_t i = 0; i < n; i++) dst[i] = (float)src[i];
} else if (in_type == RAY_I16) {
const int16_t* src = (const int16_t*)m.morsel_ptr;
for (int64_t i = 0; i < n; i++) dst[i] = (float)src[i];
} else {
const uint8_t* src = (const uint8_t*)m.morsel_ptr;
for (int64_t i = 0; i < n; i++) dst[i] = (float)src[i];
}
out_off += n;
}
} else if (in_type == RAY_F32 && out_type == RAY_F32) {
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
float* src = (float*)m.morsel_ptr;
float* dst = (float*)ray_data(result) + out_off;
switch (opc) {
case OP_NEG: for (int64_t i=0;i<n;i++) dst[i] = -src[i]; break;
case OP_ABS: for (int64_t i=0;i<n;i++) dst[i] = fabsf(src[i]); break;
case OP_CEIL: for (int64_t i=0;i<n;i++) dst[i] = ceilf(src[i]); break;
case OP_FLOOR: for (int64_t i=0;i<n;i++) dst[i] = floorf(src[i]); break;
case OP_ROUND: for (int64_t i=0;i<n;i++) dst[i] = roundf(src[i]); break;
default: for (int64_t i=0;i<n;i++) dst[i] = src[i]; break;
}
out_off += n;
}
} else if (in_type == RAY_F32 && out_type == RAY_F64) {
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
float* src = (float*)m.morsel_ptr;
double* dst = (double*)ray_data(result) + out_off;
switch (opc) {
case OP_SQRT: for (int64_t i=0;i<n;i++) dst[i] = ray_f64_fin(sqrt((double)src[i])); break;
case OP_LOG: for (int64_t i=0;i<n;i++) dst[i] = ray_f64_fin(log((double)src[i])); break;
case OP_EXP: for (int64_t i=0;i<n;i++) dst[i] = ray_f64_fin(exp((double)src[i])); break;
case OP_SIN: for (int64_t i=0;i<n;i++) dst[i] = ray_f64_fin(sin((double)src[i])); break;
case OP_ASIN: for (int64_t i=0;i<n;i++) dst[i] = ray_f64_fin(asin((double)src[i])); break;
case OP_COS: for (int64_t i=0;i<n;i++) dst[i] = ray_f64_fin(cos((double)src[i])); break;
case OP_ACOS: for (int64_t i=0;i<n;i++) dst[i] = ray_f64_fin(acos((double)src[i])); break;
case OP_TAN: for (int64_t i=0;i<n;i++) dst[i] = ray_f64_fin(tan((double)src[i])); break;
case OP_ATAN: for (int64_t i=0;i<n;i++) dst[i] = ray_f64_fin(atan((double)src[i])); break;
case OP_RECIPROCAL:
for (int64_t i=0;i<n;i++) dst[i] = src[i] != 0.0f ? ray_f64_fin(1.0 / (double)src[i]) : NULL_F64;
break;
default: for (int64_t i=0;i<n;i++) dst[i] = (double)src[i]; break;
}
out_off += n;
}
} else if (in_type == RAY_F32 && out_type == RAY_I64) {
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
float* src = (float*)m.morsel_ptr;
int64_t* dst = (int64_t*)ray_data(result) + out_off;
if (opc == OP_SIGNUM)
for (int64_t i=0;i<n;i++) dst[i] = (src[i] > 0.0f) - (src[i] < 0.0f);
else
for (int64_t i=0;i<n;i++) dst[i] = (int64_t)src[i];
out_off += n;
}
} else if (in_type == RAY_F64 && out_type == RAY_F64) {
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
double* src = (double*)m.morsel_ptr;
double* dst = (double*)ray_data(result) + out_off;
switch (opc) {
case OP_NEG: for (int64_t i=0;i<n;i++) dst[i] = -src[i]; break;
case OP_ABS: for (int64_t i=0;i<n;i++) dst[i] = fabs(src[i]); break;
case OP_SQRT: for (int64_t i=0;i<n;i++) dst[i] = ray_f64_fin(sqrt(src[i])); break;
case OP_LOG: for (int64_t i=0;i<n;i++) dst[i] = ray_f64_fin(log(src[i])); break;
case OP_EXP: for (int64_t i=0;i<n;i++) dst[i] = ray_f64_fin(exp(src[i])); break;
case OP_SIN: for (int64_t i=0;i<n;i++) dst[i] = ray_f64_fin(sin(src[i])); break;
case OP_ASIN: for (int64_t i=0;i<n;i++) dst[i] = ray_f64_fin(asin(src[i])); break;
case OP_COS: for (int64_t i=0;i<n;i++) dst[i] = ray_f64_fin(cos(src[i])); break;
case OP_ACOS: for (int64_t i=0;i<n;i++) dst[i] = ray_f64_fin(acos(src[i])); break;
case OP_TAN: for (int64_t i=0;i<n;i++) dst[i] = ray_f64_fin(tan(src[i])); break;
case OP_ATAN: for (int64_t i=0;i<n;i++) dst[i] = ray_f64_fin(atan(src[i])); break;
case OP_RECIPROCAL:
for (int64_t i=0;i<n;i++) dst[i] = src[i] != 0.0 ? ray_f64_fin(1.0 / src[i]) : NULL_F64;
break;
case OP_CEIL: for (int64_t i=0;i<n;i++) dst[i] = ceil(src[i]); break;
case OP_FLOOR: for (int64_t i=0;i<n;i++) dst[i] = floor(src[i]); break;
case OP_ROUND: for (int64_t i=0;i<n;i++) dst[i] = round(src[i]); break;
default: for (int64_t i=0;i<n;i++) dst[i] = src[i]; break;
}
out_off += n;
}
} else if (in_type == RAY_F64 && out_type == RAY_I64) {
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
double* src = (double*)m.morsel_ptr;
int64_t* dst = (int64_t*)((char*)ray_data(result) + out_off * sizeof(int64_t));
switch (opc) {
case OP_NEG: for (int64_t i=0;i<n;i++) dst[i] = (int64_t)(-src[i]); break;
case OP_ABS: for (int64_t i=0;i<n;i++) dst[i] = (int64_t)fabs(src[i]); break;
case OP_SQRT: for (int64_t i=0;i<n;i++) dst[i] = (int64_t)sqrt(src[i]); break;
case OP_LOG: for (int64_t i=0;i<n;i++) dst[i] = (int64_t)log(src[i]); break;
case OP_EXP: for (int64_t i=0;i<n;i++) dst[i] = (int64_t)exp(src[i]); break;
case OP_SIGNUM:for (int64_t i=0;i<n;i++) dst[i] = (src[i] > 0.0) - (src[i] < 0.0); break;
case OP_CEIL: for (int64_t i=0;i<n;i++) dst[i] = (int64_t)ceil(src[i]); break;
case OP_FLOOR: for (int64_t i=0;i<n;i++) dst[i] = (int64_t)floor(src[i]); break;
case OP_ROUND: for (int64_t i=0;i<n;i++) dst[i] = (int64_t)round(src[i]); break;
default: for (int64_t i=0;i<n;i++) dst[i] = ray_cast_f64_to_i64_null(src[i]); break;
}
out_off += n;
}
} else if (in_type == RAY_I64 && out_type == RAY_I64) {
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
int64_t* src = (int64_t*)m.morsel_ptr;
int64_t* dst = (int64_t*)((char*)ray_data(result) + out_off * sizeof(int64_t));
switch (opc) {
case OP_NEG: for (int64_t i=0;i<n;i++) dst[i]=(int64_t)(-(uint64_t)src[i]); break;
case OP_ABS: for (int64_t i=0;i<n;i++) dst[i]=src[i]<0?(int64_t)(-(uint64_t)src[i]):src[i]; break;
case OP_SIGNUM: for (int64_t i=0;i<n;i++) dst[i]=(src[i]>0)-(src[i]<0); break;
default: for (int64_t i=0;i<n;i++) dst[i]=src[i]; break;
}
out_off += n;
}
} else if (in_type == RAY_I64 && out_type == RAY_F64) {
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
int64_t* src = (int64_t*)m.morsel_ptr;
double* dst = (double*)ray_data(result) + out_off;
switch (opc) {
case OP_NEG: for (int64_t i=0;i<n;i++) dst[i]=-(double)src[i]; break;
case OP_SQRT: for (int64_t i=0;i<n;i++) dst[i]=ray_f64_fin(sqrt((double)src[i])); break;
case OP_LOG: for (int64_t i=0;i<n;i++) dst[i]=ray_f64_fin(log((double)src[i])); break;
case OP_EXP: for (int64_t i=0;i<n;i++) dst[i]=ray_f64_fin(exp((double)src[i])); break;
case OP_SIN: for (int64_t i=0;i<n;i++) dst[i]=ray_f64_fin(sin((double)src[i])); break;
case OP_ASIN: for (int64_t i=0;i<n;i++) dst[i]=ray_f64_fin(asin((double)src[i])); break;
case OP_COS: for (int64_t i=0;i<n;i++) dst[i]=ray_f64_fin(cos((double)src[i])); break;
case OP_ACOS: for (int64_t i=0;i<n;i++) dst[i]=ray_f64_fin(acos((double)src[i])); break;
case OP_TAN: for (int64_t i=0;i<n;i++) dst[i]=ray_f64_fin(tan((double)src[i])); break;
case OP_ATAN: for (int64_t i=0;i<n;i++) dst[i]=ray_f64_fin(atan((double)src[i])); break;
case OP_RECIPROCAL:
for (int64_t i=0;i<n;i++) dst[i]=src[i] != 0 ? ray_f64_fin(1.0 / (double)src[i]) : NULL_F64;
break;
default: for (int64_t i=0;i<n;i++) dst[i]=(double)src[i]; break;
}
out_off += n;
}
} else if (out_type == RAY_F64 && (in_type == RAY_I32 || in_type == RAY_I16 ||
in_type == RAY_U8 || in_type == RAY_BOOL)) {
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
double* dst = (double*)ray_data(result) + out_off;
#define SRC_NARROW_D(i) \
(in_type == RAY_I32 ? (double)((int32_t*)m.morsel_ptr)[i] : \
in_type == RAY_I16 ? (double)((int16_t*)m.morsel_ptr)[i] : \
(double)((uint8_t*)m.morsel_ptr)[i])
switch (opc) {
case OP_SIN: for (int64_t i=0;i<n;i++){ double v=SRC_NARROW_D(i); dst[i]=ray_f64_fin(sin(v)); } break;
case OP_ASIN: for (int64_t i=0;i<n;i++){ double v=SRC_NARROW_D(i); dst[i]=ray_f64_fin(asin(v)); } break;
case OP_COS: for (int64_t i=0;i<n;i++){ double v=SRC_NARROW_D(i); dst[i]=ray_f64_fin(cos(v)); } break;
case OP_ACOS: for (int64_t i=0;i<n;i++){ double v=SRC_NARROW_D(i); dst[i]=ray_f64_fin(acos(v)); } break;
case OP_TAN: for (int64_t i=0;i<n;i++){ double v=SRC_NARROW_D(i); dst[i]=ray_f64_fin(tan(v)); } break;
case OP_ATAN: for (int64_t i=0;i<n;i++){ double v=SRC_NARROW_D(i); dst[i]=ray_f64_fin(atan(v)); } break;
case OP_RECIPROCAL:
for (int64_t i=0;i<n;i++){ double v=SRC_NARROW_D(i); dst[i]=v != 0.0 ? ray_f64_fin(1.0 / v) : NULL_F64; }
break;
default:
for (int64_t i=0;i<n;i++) dst[i]=SRC_NARROW_D(i);
break;
}
#undef SRC_NARROW_D
out_off += n;
}
} else if (opc == OP_SIGNUM && out_type == RAY_I64 &&
(in_type == RAY_I32 || in_type == RAY_I16 ||
in_type == RAY_U8 || in_type == RAY_BOOL)) {
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
int64_t* dst = (int64_t*)((char*)ray_data(result) + out_off * sizeof(int64_t));
if (in_type == RAY_I32) {
int32_t* src = (int32_t*)m.morsel_ptr;
for (int64_t i=0;i<n;i++) dst[i]=(src[i]>0)-(src[i]<0);
} else if (in_type == RAY_I16) {
int16_t* src = (int16_t*)m.morsel_ptr;
for (int64_t i=0;i<n;i++) dst[i]=(src[i]>0)-(src[i]<0);
} else {
uint8_t* src = (uint8_t*)m.morsel_ptr;
for (int64_t i=0;i<n;i++) dst[i]=src[i] ? 1 : 0;
}
out_off += n;
}
} else if (in_type == RAY_I64 && out_type == RAY_BOOL && opc == OP_CAST) {
{
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
int64_t* src = (int64_t*)m.morsel_ptr;
uint8_t* dst = (uint8_t*)((char*)ray_data(result) + out_off);
for (int64_t i = 0; i < n; i++)
dst[i] = src[i] != 0 ? 1 : 0;
out_off += n;
}
}
} else if (opc == OP_NOT) {
int8_t it = expr_cast_lane_type(in_type);
if (it != RAY_F64 && it != RAY_F32 && it != RAY_I64 && it != RAY_I32 &&
it != RAY_I16 && it != RAY_U8 && it != RAY_BOOL) {
ray_release(result);
return ray_error("type", "not: expects a numeric argument, got %s",
ray_type_name(in_type));
}
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
uint8_t* dst = (uint8_t*)((char*)ray_data(result) + out_off);
if (it == RAY_F64) {
const double* src = (const double*)m.morsel_ptr;
for (int64_t i = 0; i < n; i++) dst[i] = !(src[i] != 0.0 && src[i] == src[i]);
} else if (it == RAY_F32) {
const float* src = (const float*)m.morsel_ptr;
for (int64_t i = 0; i < n; i++) dst[i] = !(src[i] != 0.0f && src[i] == src[i]);
} else if (it == RAY_I64) {
const int64_t* src = (const int64_t*)m.morsel_ptr;
for (int64_t i = 0; i < n; i++) dst[i] = src[i] == 0 || src[i] == NULL_I64;
} else if (it == RAY_I32) {
const int32_t* src = (const int32_t*)m.morsel_ptr;
for (int64_t i = 0; i < n; i++) dst[i] = src[i] == 0 || src[i] == NULL_I32;
} else if (it == RAY_I16) {
const int16_t* src = (const int16_t*)m.morsel_ptr;
for (int64_t i = 0; i < n; i++) dst[i] = src[i] == 0 || src[i] == NULL_I16;
} else {
const uint8_t* src = (const uint8_t*)m.morsel_ptr;
for (int64_t i = 0; i < n; i++) dst[i] = !src[i];
}
out_off += n;
}
} else if (opc == OP_CAST && (out_type == RAY_I32 || out_type == RAY_I16 ||
out_type == RAY_U8 || out_type == RAY_BOOL) &&
(expr_cast_lane_type(in_type) == RAY_I32 || in_type == RAY_I16 ||
in_type == RAY_U8 || in_type == RAY_BOOL) &&
expr_cast_lane_type(in_type) != expr_cast_lane_type(out_type)) {
int8_t it = expr_cast_lane_type(in_type);
size_t osz = ray_elem_size(out_type);
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
char* dst = (char*)ray_data(result) + (size_t)out_off * osz;
for (int64_t i = 0; i < n; i++) {
int64_t v = it == RAY_I32 ? (int64_t)((const int32_t*)m.morsel_ptr)[i]
: it == RAY_I16 ? (int64_t)((const int16_t*)m.morsel_ptr)[i]
: (int64_t)((const uint8_t*)m.morsel_ptr)[i];
if (out_type == RAY_I32) ((int32_t*)dst)[i] = (int32_t)v;
else if (out_type == RAY_I16) ((int16_t*)dst)[i] = (int16_t)v;
else if (out_type == RAY_U8) ((uint8_t*)dst)[i] = (uint8_t)v;
else
((uint8_t*)dst)[i] = v != 0;
}
out_off += n;
}
} else if (opc == OP_CAST) {
int8_t it = expr_cast_lane_type(in_type);
int8_t ot = expr_cast_lane_type(out_type);
if (it == ot) {
size_t esz = ray_elem_size(ot);
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
memcpy((char*)ray_data(result) + (size_t)out_off * esz,
m.morsel_ptr, (size_t)n * esz);
out_off += n;
}
} else
if (it == RAY_I32) {
if (ot == RAY_I64) {
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
int32_t* src = (int32_t*)m.morsel_ptr;
int64_t* dst = (int64_t*)((char*)ray_data(result) + out_off * sizeof(int64_t));
for (int64_t i = 0; i < n; i++) dst[i] = (int64_t)src[i];
out_off += n;
}
} else {
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
int32_t* src = (int32_t*)m.morsel_ptr;
double* dst = (double*)ray_data(result) + out_off;
for (int64_t i = 0; i < n; i++) dst[i] = (double)src[i];
out_off += n;
}
}
} else if (it == RAY_I16) {
if (ot == RAY_I64) {
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
int16_t* src = (int16_t*)m.morsel_ptr;
int64_t* dst = (int64_t*)((char*)ray_data(result) + out_off * sizeof(int64_t));
for (int64_t i = 0; i < n; i++) dst[i] = (int64_t)src[i];
out_off += n;
}
} else {
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
int16_t* src = (int16_t*)m.morsel_ptr;
double* dst = (double*)ray_data(result) + out_off;
for (int64_t i = 0; i < n; i++) dst[i] = (double)src[i];
out_off += n;
}
}
} else if (it == RAY_U8 || it == RAY_BOOL) {
if (ot == RAY_I64) {
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
uint8_t* src = (uint8_t*)m.morsel_ptr;
int64_t* dst = (int64_t*)((char*)ray_data(result) + out_off * sizeof(int64_t));
for (int64_t i = 0; i < n; i++) dst[i] = (int64_t)src[i];
out_off += n;
}
} else {
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
uint8_t* src = (uint8_t*)m.morsel_ptr;
double* dst = (double*)ray_data(result) + out_off;
for (int64_t i = 0; i < n; i++) dst[i] = (double)src[i];
out_off += n;
}
}
} else if (it == RAY_I64) {
if (ot == RAY_I32) {
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
int64_t* src = (int64_t*)m.morsel_ptr;
int32_t* dst = (int32_t*)((char*)ray_data(result) + out_off * sizeof(int32_t));
for (int64_t i = 0; i < n; i++) dst[i] = (int32_t)src[i];
out_off += n;
}
} else if (ot == RAY_I16) {
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
int64_t* src = (int64_t*)m.morsel_ptr;
int16_t* dst = (int16_t*)((char*)ray_data(result) + out_off * sizeof(int16_t));
for (int64_t i = 0; i < n; i++) dst[i] = (int16_t)src[i];
out_off += n;
}
} else if (ot == RAY_U8 || ot == RAY_BOOL) {
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
int64_t* src = (int64_t*)m.morsel_ptr;
uint8_t* dst = (uint8_t*)((char*)ray_data(result) + out_off);
if (ot == RAY_BOOL)
for (int64_t i = 0; i < n; i++) dst[i] = src[i] ? 1 : 0;
else
for (int64_t i = 0; i < n; i++) dst[i] = (uint8_t)src[i];
out_off += n;
}
}
} else if (it == RAY_F64) {
if (ot == RAY_I32) {
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
double* src = (double*)m.morsel_ptr;
int32_t* dst = (int32_t*)((char*)ray_data(result) + out_off * sizeof(int32_t));
for (int64_t i = 0; i < n; i++) dst[i] = ray_cast_f64_to_i32_null(src[i]);
out_off += n;
}
} else if (ot == RAY_I16) {
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
double* src = (double*)m.morsel_ptr;
int16_t* dst = (int16_t*)((char*)ray_data(result) + out_off * sizeof(int16_t));
for (int64_t i = 0; i < n; i++) dst[i] = ray_cast_f64_to_i16_null(src[i]);
out_off += n;
}
} else if (ot == RAY_U8 || ot == RAY_BOOL) {
while (ray_morsel_next(&m)) {
int64_t n = m.morsel_len;
double* src = (double*)m.morsel_ptr;
uint8_t* dst = (uint8_t*)((char*)ray_data(result) + out_off);
if (ot == RAY_BOOL)
for (int64_t i = 0; i < n; i++)
dst[i] = ray_cast_f64_to_bool_null(src[i]);
else
for (int64_t i = 0; i < n; i++) dst[i] = ray_cast_f64_to_u8_null(src[i]);
out_off += n;
}
}
}
}
if (vec_may_have_nulls(input) &&
!(opc == OP_CAST && (out_type == RAY_BOOL || out_type == RAY_U8)))
propagate_nulls(input, result, len);
if (out_type == RAY_I64 && in_type == RAY_I64 &&
(op->opcode == OP_NEG || op->opcode == OP_ABS))
mark_i64_overflow_as_null(result, 0, len);
if (out_type == RAY_F64 && expr_is_f64_unary_math(op->opcode))
mark_f64_nonfinite_as_null(result, 0, len);
return result;
}
static void binary_range_str(ray_op_t* op, ray_t* lhs, ray_t* rhs, ray_t* result,
bool l_scalar, bool r_scalar,
int64_t start, int64_t end) {
uint8_t* dst = (uint8_t*)ray_data(result) + start;
int64_t n = end - start;
uint16_t opc = op->opcode;
const ray_str_t* l_elems = NULL;
const ray_str_t* r_elems = NULL;
const char* l_pool = NULL;
const char* r_pool = NULL;
if (!l_scalar) { str_resolve(lhs, &l_elems, &l_pool); l_elems += start; }
if (!r_scalar) { str_resolve(rhs, &r_elems, &r_pool); r_elems += start; }
ray_str_t l_scalar_elem = {0}, r_scalar_elem = {0};
const char* l_scalar_pool = NULL;
const char* r_scalar_pool = NULL;
if (l_scalar) {
atom_to_str_t(lhs, &l_scalar_elem, &l_scalar_pool);
l_elems = &l_scalar_elem;
}
if (r_scalar) {
atom_to_str_t(rhs, &r_scalar_elem, &r_scalar_pool);
r_elems = &r_scalar_elem;
}
if ((opc == OP_EQ || opc == OP_NE) &&
((l_scalar && !r_scalar && l_scalar_elem.len == 0) ||
(r_scalar && !l_scalar && r_scalar_elem.len == 0))) {
const ray_str_t* vec = l_scalar ? r_elems : l_elems;
bool is_eq = (opc == OP_EQ);
for (int64_t i = 0; i < n; i++)
dst[i] = (uint8_t)((vec[i].len == 0) == is_eq);
return;
}
const ray_str_t* la = l_scalar ? l_elems : l_elems;
const ray_str_t* lb = r_scalar ? r_elems : r_elems;
const char* pa = l_scalar ? l_scalar_pool : l_pool;
const char* pb = r_scalar ? r_scalar_pool : r_pool;
int step_l = l_scalar ? 0 : 1;
int step_r = r_scalar ? 0 : 1;
#define STR_CMP_LOOP(expr) \
do { \
for (int64_t i = 0; i < n; i++) { \
const ray_str_t* a = la + i * step_l; \
const ray_str_t* b = lb + i * step_r; \
dst[i] = (uint8_t)(expr); \
} \
} while (0)
switch (opc) {
case OP_EQ: STR_CMP_LOOP(ray_str_t_eq(a, pa, b, pb)); break;
case OP_NE: STR_CMP_LOOP(!ray_str_t_eq(a, pa, b, pb)); break;
case OP_LT: STR_CMP_LOOP(ray_str_t_cmp(a, pa, b, pb) < 0); break;
case OP_LE: STR_CMP_LOOP(ray_str_t_cmp(a, pa, b, pb) <= 0); break;
case OP_GT: STR_CMP_LOOP(ray_str_t_cmp(a, pa, b, pb) > 0); break;
case OP_GE: STR_CMP_LOOP(ray_str_t_cmp(a, pa, b, pb) >= 0); break;
default: for (int64_t i = 0; i < n; i++) dst[i] = 0; break;
}
#undef STR_CMP_LOOP
}
typedef struct {
ray_op_t* op;
ray_t* lhs;
ray_t* rhs;
ray_t* result;
bool l_scalar;
bool r_scalar;
} par_binary_str_ctx_t;
static void par_binary_str_fn(void* ctx, uint32_t worker_id, int64_t start, int64_t end) {
(void)worker_id;
par_binary_str_ctx_t* c = (par_binary_str_ctx_t*)ctx;
binary_range_str(c->op, c->lhs, c->rhs, c->result,
c->l_scalar, c->r_scalar, start, end);
}
static void binary_range(ray_op_t* op, int8_t out_type,
ray_t* lhs, ray_t* rhs, ray_t* result, void* rdata,
bool l_scalar, bool r_scalar,
double l_f64, double r_f64,
int64_t l_i64, int64_t r_i64,
int64_t start, int64_t end) {
uint8_t out_esz = ray_elem_size(out_type);
void* dst = (char*)rdata + start * out_esz;
int64_t n = end - start;
bool r_scalar_float = rhs->type == -RAY_F64 || rhs->type == -RAY_F32 ||
rhs->type == RAY_F64 || rhs->type == RAY_F32;
if (out_type == RAY_BOOL && !l_scalar && r_scalar && !r_scalar_float &&
(op->opcode == OP_EQ || op->opcode == OP_NE ||
op->opcode == OP_LT || op->opcode == OP_LE ||
op->opcode == OP_GT || op->opcode == OP_GE) &&
(lhs->type == RAY_I64 || lhs->type == RAY_TIMESTAMP ||
lhs->type == RAY_I32 || lhs->type == RAY_DATE ||
lhs->type == RAY_TIME || lhs->type == RAY_I16 ||
lhs->type == RAY_BOOL || lhs->type == RAY_U8 ||
RAY_IS_SYM(lhs->type)))
{
int64_t l_off = start;
void* l_data = resolve_vec_data(lhs, &l_off);
uint8_t l_esz = ray_sym_elem_size(lhs->type, lhs->attrs);
const uint8_t* lbase = (const uint8_t*)l_data + l_off * l_esz;
uint8_t* odst = (uint8_t*)dst;
int64_t cv = r_i64;
uint16_t opc = op->opcode;
#define BR_FAST(T, READ_T) \
do { \
const T* d = (const T*)lbase; \
T cvt = (T)cv; \
switch (opc) { \
case OP_EQ: for (int64_t i = 0; i < n; i++) odst[i] = (READ_T == cvt); break; \
case OP_NE: for (int64_t i = 0; i < n; i++) odst[i] = (READ_T != cvt); break; \
case OP_LT: for (int64_t i = 0; i < n; i++) odst[i] = (READ_T < cvt); break; \
case OP_LE: for (int64_t i = 0; i < n; i++) odst[i] = (READ_T <= cvt); break; \
case OP_GT: for (int64_t i = 0; i < n; i++) odst[i] = (READ_T > cvt); break; \
case OP_GE: for (int64_t i = 0; i < n; i++) odst[i] = (READ_T >= cvt); break; \
} \
} while (0)
if (l_esz == 8) {
if (RAY_IS_SYM(lhs->type)) {
BR_FAST(int64_t, d[i]);
} else if (lhs->type == RAY_I64 || lhs->type == RAY_TIMESTAMP) {
BR_FAST(int64_t, d[i]);
}
return;
} else if (l_esz == 4) {
if (RAY_IS_SYM(lhs->type)) {
if (opc == OP_EQ || opc == OP_NE) {
const uint32_t* d = (const uint32_t*)lbase;
uint32_t cvu = (uint32_t)cv;
if (opc == OP_EQ)
for (int64_t i = 0; i < n; i++) odst[i] = (d[i] == cvu);
else
for (int64_t i = 0; i < n; i++) odst[i] = (d[i] != cvu);
return;
}
BR_FAST(uint32_t, (int64_t)d[i]);
} else {
BR_FAST(int32_t, (int64_t)d[i]);
}
return;
} else if (l_esz == 2) {
if (RAY_IS_SYM(lhs->type)) {
BR_FAST(uint16_t, (int64_t)d[i]);
} else {
BR_FAST(int16_t, (int64_t)d[i]);
}
return;
} else if (l_esz == 1) {
BR_FAST(uint8_t, (int64_t)d[i]);
return;
}
#undef BR_FAST
}
if (!l_scalar && r_scalar &&
(op->opcode == OP_ADD || op->opcode == OP_SUB ||
op->opcode == OP_MUL || op->opcode == OP_MIN2 ||
op->opcode == OP_MAX2) &&
lhs->type == out_type &&
(out_type == RAY_I64 || out_type == RAY_TIMESTAMP ||
out_type == RAY_I32 || out_type == RAY_DATE || out_type == RAY_TIME ||
out_type == RAY_I16))
{
int64_t l_off = start;
void* l_data = resolve_vec_data(lhs, &l_off);
uint8_t l_esz = ray_sym_elem_size(lhs->type, lhs->attrs);
const uint8_t* lbase = (const uint8_t*)l_data + l_off * l_esz;
int64_t cv = r_i64;
uint16_t opc = op->opcode;
#define BR_AR_FAST(T) do { \
const T* d = (const T*)lbase; \
T* odst = (T*)dst; \
T cvt = (T)cv; \
switch (opc) { \
case OP_ADD: for (int64_t i = 0; i < n; i++) odst[i] = (T)((uint64_t)d[i] + (uint64_t)cvt); break; \
case OP_SUB: for (int64_t i = 0; i < n; i++) odst[i] = (T)((uint64_t)d[i] - (uint64_t)cvt); break; \
case OP_MUL: for (int64_t i = 0; i < n; i++) odst[i] = (T)((uint64_t)d[i] * (uint64_t)cvt); break; \
case OP_MIN2: for (int64_t i = 0; i < n; i++) odst[i] = d[i] < cvt ? d[i] : cvt; break; \
case OP_MAX2: for (int64_t i = 0; i < n; i++) odst[i] = d[i] > cvt ? d[i] : cvt; break; \
} \
} while (0)
if (l_esz == 8) { BR_AR_FAST(int64_t); return; }
if (l_esz == 4) { BR_AR_FAST(int32_t); return; }
if (l_esz == 2) { BR_AR_FAST(int16_t); return; }
#undef BR_AR_FAST
}
double* lp_f64 = NULL; float* lp_f32 = NULL;
int64_t* lp_i64 = NULL; uint8_t* lp_bool = NULL;
double* rp_f64 = NULL; float* rp_f32 = NULL;
int64_t* rp_i64 = NULL; uint8_t* rp_bool = NULL;
int32_t* lp_i32 = NULL; uint32_t* lp_u32 = NULL; int16_t* lp_i16 = NULL;
int32_t* rp_i32 = NULL; uint32_t* rp_u32 = NULL; int16_t* rp_i16 = NULL;
int64_t* lsym_buf = NULL;
int64_t* rsym_buf = NULL;
bool sym_xlate = !l_scalar && !r_scalar &&
RAY_IS_SYM(lhs->type) && RAY_IS_SYM(rhs->type) &&
ray_sym_vec_domain(lhs) != ray_sym_vec_domain(rhs);
if (!l_scalar) {
int64_t l_off = start;
void* l_data = resolve_vec_data(lhs, &l_off);
void* lbase = (char*)l_data + l_off * ray_sym_elem_size(lhs->type, lhs->attrs);
if (lhs->type == RAY_F64) lp_f64 = (double*)lbase;
else if (lhs->type == RAY_F32) lp_f32 = (float*)lbase;
else if (lhs->type == RAY_I64 || lhs->type == RAY_TIMESTAMP) lp_i64 = (int64_t*)lbase;
else if (RAY_IS_SYM(lhs->type)) {
uint8_t w = lhs->attrs & RAY_SYM_W_MASK;
if (w == RAY_SYM_W64) lp_i64 = (int64_t*)lbase;
else if (w == RAY_SYM_W32) lp_u32 = (uint32_t*)lbase;
else {
lsym_buf = (int64_t*)ray_alloc_raw((size_t)n * sizeof(int64_t));
if (!lsym_buf) goto done;
for (int64_t j = 0; j < n; j++) lsym_buf[j] = ray_read_sym(l_data, l_off+j, lhs->type, lhs->attrs);
lp_i64 = lsym_buf;
}
}
else if (lhs->type == RAY_I32 || lhs->type == RAY_DATE || lhs->type == RAY_TIME) lp_i32 = (int32_t*)lbase;
else if (lhs->type == RAY_I16) lp_i16 = (int16_t*)lbase;
else if (lhs->type == RAY_BOOL || lhs->type == RAY_U8) lp_bool = (uint8_t*)lbase;
}
if (!r_scalar) {
int64_t r_off = start;
void* r_data = resolve_vec_data(rhs, &r_off);
void* rbase = (char*)r_data + r_off * ray_sym_elem_size(rhs->type, rhs->attrs);
if (rhs->type == RAY_F64) rp_f64 = (double*)rbase;
else if (rhs->type == RAY_F32) rp_f32 = (float*)rbase;
else if (rhs->type == RAY_I64 || rhs->type == RAY_TIMESTAMP) rp_i64 = (int64_t*)rbase;
else if (RAY_IS_SYM(rhs->type)) {
uint8_t w = rhs->attrs & RAY_SYM_W_MASK;
if (sym_xlate) {
rsym_buf = (int64_t*)ray_alloc_raw((size_t)n * sizeof(int64_t));
if (!rsym_buf) goto done;
struct ray_sym_domain_s* ld = ray_sym_vec_domain(lhs);
struct ray_sym_domain_s* rd = ray_sym_vec_domain(rhs);
for (int64_t j = 0; j < n; j++) {
int64_t rid = ray_read_sym(r_data, r_off+j, rhs->type, rhs->attrs);
ray_t* s = ray_sym_domain_str(rd, rid);
rsym_buf[j] = s ? ray_sym_domain_find(ld, ray_str_ptr(s),
ray_str_len(s)) : -1;
}
rp_i64 = rsym_buf;
}
else if (w == RAY_SYM_W64) rp_i64 = (int64_t*)rbase;
else if (w == RAY_SYM_W32) rp_u32 = (uint32_t*)rbase;
else {
rsym_buf = (int64_t*)ray_alloc_raw((size_t)n * sizeof(int64_t));
if (!rsym_buf) goto done;
for (int64_t j = 0; j < n; j++) rsym_buf[j] = ray_read_sym(r_data, r_off+j, rhs->type, rhs->attrs);
rp_i64 = rsym_buf;
}
}
else if (rhs->type == RAY_I32 || rhs->type == RAY_DATE || rhs->type == RAY_TIME) rp_i32 = (int32_t*)rbase;
else if (rhs->type == RAY_I16) rp_i16 = (int16_t*)rbase;
else if (rhs->type == RAY_BOOL || rhs->type == RAY_U8) rp_bool = (uint8_t*)rbase;
}
#define LV_READ(i) (lp_f64 ? lp_f64[i] : lp_f32 ? (double)lp_f32[i] : lp_i64 ? (double)lp_i64[i] : lp_i32 ? (double)lp_i32[i] : lp_u32 ? (double)lp_u32[i] : lp_i16 ? (double)lp_i16[i] : lp_bool ? (double)lp_bool[i] : (l_scalar && (lhs->type == -RAY_F64 || lhs->type == RAY_F64 || lhs->type == -RAY_F32 || lhs->type == RAY_F32)) ? l_f64 : (double)l_i64)
#define RV_READ(i) (rp_f64 ? rp_f64[i] : rp_f32 ? (double)rp_f32[i] : rp_i64 ? (double)rp_i64[i] : rp_i32 ? (double)rp_i32[i] : rp_u32 ? (double)rp_u32[i] : rp_i16 ? (double)rp_i16[i] : rp_bool ? (double)rp_bool[i] : (r_scalar && (rhs->type == -RAY_F64 || rhs->type == RAY_F64 || rhs->type == -RAY_F32 || rhs->type == RAY_F32)) ? r_f64 : (double)r_i64)
#define LV_READ_I64(i) (lp_i64 ? lp_i64[i] : lp_i32 ? (int64_t)lp_i32[i] : lp_u32 ? (int64_t)lp_u32[i] : lp_i16 ? (int64_t)lp_i16[i] : lp_bool ? (int64_t)lp_bool[i] : l_i64)
#define RV_READ_I64(i) (rp_i64 ? rp_i64[i] : rp_i32 ? (int64_t)rp_i32[i] : rp_u32 ? (int64_t)rp_u32[i] : rp_i16 ? (int64_t)rp_i16[i] : rp_bool ? (int64_t)rp_bool[i] : r_i64)
#define LV_INT(i) (src_is_i64_all ? LV_READ_I64(i) : (int64_t)LV_READ(i))
#define RV_INT(i) (src_is_i64_all ? RV_READ_I64(i) : (int64_t)RV_READ(i))
int l_is_int = !(lp_f64 || lp_f32 || (l_scalar &&
(lhs->type == -RAY_F64 || lhs->type == RAY_F64 ||
lhs->type == -RAY_F32 || lhs->type == RAY_F32)));
int r_is_int = !(rp_f64 || rp_f32 || (r_scalar &&
(rhs->type == -RAY_F64 || rhs->type == RAY_F64 ||
rhs->type == -RAY_F32 || rhs->type == RAY_F32)));
int src_is_i64_all = l_is_int && r_is_int;
bool idiv_i64_small = false;
bool idiv_i64_inline_gate = false;
if (src_is_i64_all && op->opcode == OP_IDIV) {
const int64_t* lscan = lp_i64;
const int64_t* rscan = rp_i64;
bool l_vec = lscan || lp_i32 || lp_u32 || lp_i16 || lp_bool;
bool r_vec = rscan || rp_i32 || rp_u32 || rp_i16 || rp_bool;
idiv_i64_small = (l_vec || i64_dbl_exact(l_i64)) &&
(r_vec || i64_dbl_exact(r_i64));
idiv_i64_inline_gate = idiv_i64_small && lscan && (rscan || !r_vec) &&
(out_type == RAY_I64 || out_type == RAY_TIMESTAMP);
if (idiv_i64_inline_gate) idiv_i64_small = false;
if (idiv_i64_small && !idiv_i64_inline_gate) {
if (lscan && rscan) idiv_i64_small = i64_span2_dbl_exact(lscan, rscan, n);
else if (lscan) idiv_i64_small = i64_span_dbl_exact(lscan, n);
else if (rscan) idiv_i64_small = i64_span_dbl_exact(rscan, n);
}
}
if (out_type == RAY_F64) {
double* odst = (double*)dst;
switch (op->opcode) {
case OP_ADD: for (int64_t i=0;i<n;i++) { double lv=LV_READ(i),rv=RV_READ(i); odst[i]=ray_f64_fin(lv+rv); } break;
case OP_SUB: for (int64_t i=0;i<n;i++) { double lv=LV_READ(i),rv=RV_READ(i); odst[i]=ray_f64_fin(lv-rv); } break;
case OP_MUL: for (int64_t i=0;i<n;i++) { double lv=LV_READ(i),rv=RV_READ(i); odst[i]=ray_f64_fin(lv*rv); } break;
case OP_DIV: for (int64_t i=0;i<n;i++) { double lv=LV_READ(i),rv=RV_READ(i); odst[i]=rv!=0.0?ray_f64_fin(lv/rv):NULL_F64; } break;
case OP_IDIV: for (int64_t i=0;i<n;i++) { double lv=LV_READ(i),rv=RV_READ(i); odst[i]=rv!=0.0?ray_f64_fin(floor(lv/rv)):NULL_F64; } break;
case OP_MOD: for (int64_t i=0;i<n;i++) { double lv=LV_READ(i),rv=RV_READ(i); double r; if(rv!=0.0){r=fmod(lv,rv);if(r&&((r>0)!=(rv>0)))r+=rv; r=ray_f64_fin(r);}else r=NULL_F64; odst[i]=r; } break;
case OP_POW: for (int64_t i=0;i<n;i++) { double lv=LV_READ(i),rv=RV_READ(i); odst[i]=(lv!=lv||rv!=rv)?NULL_F64:ray_f64_fin(pow(lv,rv)); } break;
case OP_MIN2: for (int64_t i=0;i<n;i++) { double lv=LV_READ(i),rv=RV_READ(i); odst[i]=lv<rv?lv:rv; } break;
case OP_MAX2: for (int64_t i=0;i<n;i++) { double lv=LV_READ(i),rv=RV_READ(i); odst[i]=lv>rv?lv:rv; } break;
default: for (int64_t i=0;i<n;i++) odst[i]=0.0; break;
}
if (op->opcode == OP_ADD || op->opcode == OP_SUB || op->opcode == OP_MUL ||
op->opcode == OP_DIV || op->opcode == OP_IDIV || op->opcode == OP_MOD ||
op->opcode == OP_POW) {
int any_nan = 0;
for (int64_t i = 0; i < n; i++) any_nan |= (odst[i] != odst[i]);
if (any_nan)
__atomic_fetch_or(&result->attrs, (uint8_t)RAY_ATTR_HAS_NULLS, __ATOMIC_RELAXED);
}
} else if (out_type == RAY_F32) {
float* odst = (float*)dst;
switch (op->opcode) {
case OP_ADD: for (int64_t i=0;i<n;i++) { float v=(float)(LV_READ(i)+RV_READ(i)); odst[i]=isfinite(v)?v:NULL_F32; } break;
case OP_SUB: for (int64_t i=0;i<n;i++) { float v=(float)(LV_READ(i)-RV_READ(i)); odst[i]=isfinite(v)?v:NULL_F32; } break;
case OP_MUL: for (int64_t i=0;i<n;i++) { float v=(float)(LV_READ(i)*RV_READ(i)); odst[i]=isfinite(v)?v:NULL_F32; } break;
case OP_MOD: for (int64_t i=0;i<n;i++) { float lv=(float)LV_READ(i),rv=(float)RV_READ(i); float v; if(rv!=0.0f){v=fmodf(lv,rv);if(v&&((v>0)!=(rv>0)))v+=rv;v=isfinite(v)?v:NULL_F32;}else v=NULL_F32;odst[i]=v; } break;
case OP_MIN2: for (int64_t i=0;i<n;i++) { float lv=(float)LV_READ(i),rv=(float)RV_READ(i); odst[i]=lv<rv?lv:rv; } break;
case OP_MAX2: for (int64_t i=0;i<n;i++) { float lv=(float)LV_READ(i),rv=(float)RV_READ(i); odst[i]=lv>rv?lv:rv; } break;
default: for (int64_t i=0;i<n;i++) odst[i]=0.0f; break;
}
if (op->opcode == OP_ADD || op->opcode == OP_SUB ||
op->opcode == OP_MUL || op->opcode == OP_MOD) {
int any_nan = 0;
for (int64_t i = 0; i < n; i++) any_nan |= (odst[i] != odst[i]);
if (any_nan)
__atomic_fetch_or(&result->attrs, (uint8_t)RAY_ATTR_HAS_NULLS,
__ATOMIC_RELAXED);
}
} else if (out_type == RAY_I64 || out_type == RAY_TIMESTAMP) {
int64_t* odst = (int64_t*)dst;
switch (op->opcode) {
case OP_ADD: for(int64_t i=0;i<n;i++){int64_t li=LV_INT(i),ri=RV_INT(i);odst[i]=(int64_t)((uint64_t)li+(uint64_t)ri);}break;
case OP_SUB: for(int64_t i=0;i<n;i++){int64_t li=LV_INT(i),ri=RV_INT(i);odst[i]=(int64_t)((uint64_t)li-(uint64_t)ri);}break;
case OP_MUL: for(int64_t i=0;i<n;i++){int64_t li=LV_INT(i),ri=RV_INT(i);odst[i]=(int64_t)((uint64_t)li*(uint64_t)ri);}break;
case OP_DIV: for(int64_t i=0;i<n;i++){int64_t li=LV_INT(i),ri=RV_INT(i),q=0;odst[i]=floor_idiv_i64_checked(li,ri,&q)?q:0;}break;
case OP_IDIV:
if (rp_i16) {
int any_zero = 0;
if (src_is_i64_all && !idiv_i64_small) {
for (int64_t i=0;i<n;i++) {
int64_t li=LV_READ_I64(i), ri=rp_i16[i], q=0;
any_zero |= (ri == 0);
odst[i] = ri == 0 ? NULL_I64
: floor_idiv_i64_checked(li, ri, &q) ? q : 0;
}
} else {
for (int64_t i=0;i<n;i++) {
double lv=LV_READ(i);
int16_t ri=rp_i16[i];
any_zero |= (ri == 0);
odst[i] = ri == 0 ? NULL_I64
: ray_cast_f64_to_i64_null(floor(lv/(double)ri));
}
}
if (any_zero)
__atomic_fetch_or(&result->attrs, (uint8_t)RAY_ATTR_HAS_NULLS,
__ATOMIC_RELAXED);
} else if (idiv_i64_inline_gate) {
int bad = 0;
if (rp_i64) {
const int64_t* lv = lp_i64; const int64_t* rv = rp_i64;
for (int64_t i=0;i<n;i++) {
int64_t li=lv[i], ri=rv[i];
bad |= !i64_dbl_exact(li);
bad |= !i64_dbl_exact(ri);
odst[i] = ri!=0 ? ray_cast_f64_to_i64_null(floor((double)li/(double)ri)) : 0;
}
} else {
const int64_t* lv = lp_i64; double rd = (double)r_i64;
for (int64_t i=0;i<n;i++) {
int64_t li=lv[i];
bad |= !i64_dbl_exact(li);
odst[i] = r_i64!=0 ? ray_cast_f64_to_i64_null(floor((double)li/rd)) : 0;
}
}
if (bad)
for(int64_t i=0;i<n;i++){int64_t li=LV_READ_I64(i),ri=RV_READ_I64(i),q=0;odst[i]=floor_idiv_i64_checked(li,ri,&q)?q:0;}
} else if (src_is_i64_all && !idiv_i64_small) {
for(int64_t i=0;i<n;i++){int64_t li=LV_READ_I64(i),ri=RV_READ_I64(i),q=0;odst[i]=floor_idiv_i64_checked(li,ri,&q)?q:0;}
} else {
for(int64_t i=0;i<n;i++){double lv=LV_READ(i),rv=RV_READ(i);odst[i]=rv!=0.0?ray_cast_f64_to_i64_null(floor(lv/rv)):0;}
}
break;
case OP_MOD: for(int64_t i=0;i<n;i++){int64_t li=LV_INT(i),ri=RV_INT(i);int64_t r;if(ri==0||(ri==-1&&li==INT64_MIN)){r=0;}else{r=li%ri;if(r&&(r^ri)<0)r+=ri;}odst[i]=r;}break;
case OP_MIN2:for(int64_t i=0;i<n;i++){int64_t li=LV_INT(i),ri=RV_INT(i);odst[i]=li<ri?li:ri;}break;
case OP_MAX2:for(int64_t i=0;i<n;i++){int64_t li=LV_INT(i),ri=RV_INT(i);odst[i]=li>ri?li:ri;}break;
default: for(int64_t i=0;i<n;i++)odst[i]=0;break;
}
} else if (out_type == RAY_I32 || out_type == RAY_DATE || out_type == RAY_TIME) {
int32_t* odst = (int32_t*)dst;
switch (op->opcode) {
case OP_ADD: for(int64_t i=0;i<n;i++){int32_t li=(int32_t)LV_INT(i),ri=(int32_t)RV_INT(i);odst[i]=(int32_t)((uint32_t)li+(uint32_t)ri);}break;
case OP_SUB: for(int64_t i=0;i<n;i++){int32_t li=(int32_t)LV_INT(i),ri=(int32_t)RV_INT(i);odst[i]=(int32_t)((uint32_t)li-(uint32_t)ri);}break;
case OP_MUL: for(int64_t i=0;i<n;i++){int32_t li=(int32_t)LV_INT(i),ri=(int32_t)RV_INT(i);odst[i]=(int32_t)((uint32_t)li*(uint32_t)ri);}break;
case OP_IDIV:
if (src_is_i64_all && !idiv_i64_small) {
for(int64_t i=0;i<n;i++){int64_t li=LV_READ_I64(i),ri=RV_READ_I64(i),q=0;odst[i]=floor_idiv_i64_checked(li,ri,&q)?(q>=INT32_MAX?INT32_MAX:q<=INT32_MIN?INT32_MIN+1:(int32_t)q):0;}
} else {
for(int64_t i=0;i<n;i++){double lv=LV_READ(i),rv=RV_READ(i);odst[i]=rv!=0.0?ray_cast_f64_to_i32_null(floor(lv/rv)):0;}
}
break;
case OP_MOD: for(int64_t i=0;i<n;i++){int32_t li=(int32_t)LV_INT(i),ri=(int32_t)RV_INT(i);int32_t r;if(ri==0||(ri==-1&&li==INT32_MIN)){r=0;}else{r=li%ri;if(r&&(r^ri)<0)r+=ri;}odst[i]=r;}break;
case OP_MIN2:for(int64_t i=0;i<n;i++){int32_t li=(int32_t)LV_INT(i),ri=(int32_t)RV_INT(i);odst[i]=li<ri?li:ri;}break;
case OP_MAX2:for(int64_t i=0;i<n;i++){int32_t li=(int32_t)LV_INT(i),ri=(int32_t)RV_INT(i);odst[i]=li>ri?li:ri;}break;
default: for(int64_t i=0;i<n;i++)odst[i]=0;break;
}
} else if (out_type == RAY_I16) {
int16_t* odst = (int16_t*)dst;
switch (op->opcode) {
case OP_ADD: for(int64_t i=0;i<n;i++){int16_t li=(int16_t)LV_INT(i),ri=(int16_t)RV_INT(i);odst[i]=(int16_t)((uint16_t)li+(uint16_t)ri);}break;
case OP_SUB: for(int64_t i=0;i<n;i++){int16_t li=(int16_t)LV_INT(i),ri=(int16_t)RV_INT(i);odst[i]=(int16_t)((uint16_t)li-(uint16_t)ri);}break;
case OP_MUL: for(int64_t i=0;i<n;i++){int16_t li=(int16_t)LV_INT(i),ri=(int16_t)RV_INT(i);odst[i]=(int16_t)((uint16_t)li*(uint16_t)ri);}break;
case OP_IDIV:
if (src_is_i64_all && !idiv_i64_small) {
for(int64_t i=0;i<n;i++){int64_t li=LV_READ_I64(i),ri=RV_READ_I64(i),q=0;odst[i]=floor_idiv_i64_checked(li,ri,&q)?(q>=INT16_MAX?INT16_MAX:q<=INT16_MIN?INT16_MIN+1:(int16_t)q):0;}
} else {
for(int64_t i=0;i<n;i++){double lv=LV_READ(i),rv=RV_READ(i);odst[i]=rv!=0.0?ray_cast_f64_to_i16_null(floor(lv/rv)):0;}
}
break;
case OP_MOD: for(int64_t i=0;i<n;i++){int16_t li=(int16_t)LV_INT(i),ri=(int16_t)RV_INT(i);int16_t r;if(ri==0||(ri==-1&&li==INT16_MIN)){r=0;}else{r=(int16_t)(li%ri);if(r&&(r^ri)<0)r=(int16_t)(r+ri);}odst[i]=r;}break;
case OP_MIN2:for(int64_t i=0;i<n;i++){int16_t li=(int16_t)LV_INT(i),ri=(int16_t)RV_INT(i);odst[i]=li<ri?li:ri;}break;
case OP_MAX2:for(int64_t i=0;i<n;i++){int16_t li=(int16_t)LV_INT(i),ri=(int16_t)RV_INT(i);odst[i]=li>ri?li:ri;}break;
default: for(int64_t i=0;i<n;i++)odst[i]=0;break;
}
} else if (out_type == RAY_U8) {
uint8_t* odst = (uint8_t*)dst;
switch (op->opcode) {
case OP_ADD: for(int64_t i=0;i<n;i++){uint8_t li=(uint8_t)LV_INT(i),ri=(uint8_t)RV_INT(i);odst[i]=li+ri;}break;
case OP_SUB: for(int64_t i=0;i<n;i++){uint8_t li=(uint8_t)LV_INT(i),ri=(uint8_t)RV_INT(i);odst[i]=li-ri;}break;
case OP_MUL: for(int64_t i=0;i<n;i++){uint8_t li=(uint8_t)LV_INT(i),ri=(uint8_t)RV_INT(i);odst[i]=li*ri;}break;
case OP_IDIV:
if (src_is_i64_all && !idiv_i64_small) {
for(int64_t i=0;i<n;i++){int64_t li=LV_READ_I64(i),ri=RV_READ_I64(i),q=0;odst[i]=floor_idiv_i64_checked(li,ri,&q)?(q<=0?0:q>=UINT8_MAX?UINT8_MAX:(uint8_t)q):0;}
} else {
for(int64_t i=0;i<n;i++){double lv=LV_READ(i),rv=RV_READ(i);odst[i]=rv!=0.0?ray_cast_f64_to_u8_null(floor(lv/rv)):0;}
}
break;
case OP_MOD: for(int64_t i=0;i<n;i++){uint8_t li=(uint8_t)LV_INT(i),ri=(uint8_t)RV_INT(i);odst[i]=ri?li%ri:0;}break;
case OP_MIN2:for(int64_t i=0;i<n;i++){uint8_t li=(uint8_t)LV_INT(i),ri=(uint8_t)RV_INT(i);odst[i]=li<ri?li:ri;}break;
case OP_MAX2:for(int64_t i=0;i<n;i++){uint8_t li=(uint8_t)LV_INT(i),ri=(uint8_t)RV_INT(i);odst[i]=li>ri?li:ri;}break;
default: for(int64_t i=0;i<n;i++)odst[i]=0;break;
}
} else if (out_type == RAY_BOOL) {
uint8_t* odst = (uint8_t*)dst;
if (src_is_i64_all) {
switch (op->opcode) {
case OP_EQ: for(int64_t i=0;i<n;i++){int64_t li=LV_INT(i),ri=RV_INT(i);odst[i]=li==ri;}break;
case OP_NE: for(int64_t i=0;i<n;i++){int64_t li=LV_INT(i),ri=RV_INT(i);odst[i]=li!=ri;}break;
case OP_LT: for(int64_t i=0;i<n;i++){int64_t li=LV_INT(i),ri=RV_INT(i);odst[i]=li<ri;}break;
case OP_LE: for(int64_t i=0;i<n;i++){int64_t li=LV_INT(i),ri=RV_INT(i);odst[i]=li<=ri;}break;
case OP_GT: for(int64_t i=0;i<n;i++){int64_t li=LV_INT(i),ri=RV_INT(i);odst[i]=li>ri;}break;
case OP_GE: for(int64_t i=0;i<n;i++){int64_t li=LV_INT(i),ri=RV_INT(i);odst[i]=li>=ri;}break;
case OP_AND: for(int64_t i=0;i<n;i++){uint8_t li=truthy_intish(LV_READ(i)),ri=truthy_intish(RV_READ(i));odst[i]=li&&ri;}break;
case OP_OR: for(int64_t i=0;i<n;i++){uint8_t li=truthy_intish(LV_READ(i)),ri=truthy_intish(RV_READ(i));odst[i]=li||ri;}break;
default: for(int64_t i=0;i<n;i++)odst[i]=0;break;
}
} else {
switch (op->opcode) {
case OP_EQ: for(int64_t i=0;i<n;i++){double lv=LV_READ(i),rv=RV_READ(i);int ln=lv!=lv,rn=rv!=rv;odst[i]=(ln&&rn)?1:(ln||rn)?0:lv==rv;}break;
case OP_NE: for(int64_t i=0;i<n;i++){double lv=LV_READ(i),rv=RV_READ(i);int ln=lv!=lv,rn=rv!=rv;odst[i]=(ln&&rn)?0:(ln||rn)?1:lv!=rv;}break;
case OP_LT: for(int64_t i=0;i<n;i++){double lv=LV_READ(i),rv=RV_READ(i);int ln=lv!=lv,rn=rv!=rv;odst[i]=(ln&&rn)?0:ln?1:rn?0:lv<rv;}break;
case OP_LE: for(int64_t i=0;i<n;i++){double lv=LV_READ(i),rv=RV_READ(i);int ln=lv!=lv,rn=rv!=rv;odst[i]=(ln&&rn)?1:ln?1:rn?0:lv<=rv;}break;
case OP_GT: for(int64_t i=0;i<n;i++){double lv=LV_READ(i),rv=RV_READ(i);int ln=lv!=lv,rn=rv!=rv;odst[i]=(ln&&rn)?0:rn?1:ln?0:lv>rv;}break;
case OP_GE: for(int64_t i=0;i<n;i++){double lv=LV_READ(i),rv=RV_READ(i);int ln=lv!=lv,rn=rv!=rv;odst[i]=(ln&&rn)?1:rn?1:ln?0:lv>=rv;}break;
case OP_AND: for(int64_t i=0;i<n;i++){uint8_t li=truthy_f64ish(LV_READ(i)),ri=truthy_f64ish(RV_READ(i));odst[i]=li&&ri;}break;
case OP_OR: for(int64_t i=0;i<n;i++){uint8_t li=truthy_f64ish(LV_READ(i)),ri=truthy_f64ish(RV_READ(i));odst[i]=li||ri;}break;
default: for(int64_t i=0;i<n;i++)odst[i]=0;break;
}
}
}
#undef LV_READ
#undef RV_READ
#undef LV_READ_I64
#undef RV_READ_I64
#undef LV_INT
#undef RV_INT
done:
if (lsym_buf) ray_free_raw(lsym_buf);
if (rsym_buf) ray_free_raw(rsym_buf);
}
typedef struct {
ray_op_t* op;
int8_t out_type;
ray_t* lhs;
ray_t* rhs;
ray_t* result;
void* result_data;
bool l_scalar;
bool r_scalar;
double l_f64, r_f64;
int64_t l_i64, r_i64;
const uint8_t* sel_flg;
const uint32_t* sel_offs;
const uint16_t* sel_idx;
uint32_t sel_n_segs;
} par_binary_ctx_t;
static void par_binary_fn(void* ctx, uint32_t worker_id, int64_t start, int64_t end) {
(void)worker_id;
par_binary_ctx_t* c = (par_binary_ctx_t*)ctx;
if (!c->sel_flg) {
binary_range(c->op, c->out_type, c->lhs, c->rhs, c->result, c->result_data,
c->l_scalar, c->r_scalar,
c->l_f64, c->r_f64, c->l_i64, c->r_i64,
start, end);
return;
}
uint32_t seg_lo = (uint32_t)(start / RAY_MORSEL_ELEMS);
uint32_t seg_hi = (uint32_t)((end + RAY_MORSEL_ELEMS - 1) / RAY_MORSEL_ELEMS);
if (seg_hi > c->sel_n_segs) seg_hi = c->sel_n_segs;
for (uint32_t seg = seg_lo; seg < seg_hi; seg++) {
int64_t s_lo = (int64_t)seg * RAY_MORSEL_ELEMS;
int64_t s_hi = s_lo + RAY_MORSEL_ELEMS;
if (s_lo < start) s_lo = start;
if (s_hi > end) s_hi = end;
if (c->sel_flg[seg] == RAY_SEL_NONE) continue;
binary_range(c->op, c->out_type, c->lhs, c->rhs, c->result, c->result_data,
c->l_scalar, c->r_scalar,
c->l_f64, c->r_f64, c->l_i64, c->r_i64,
s_lo, s_hi);
}
}
ray_t* exec_elementwise_binary(ray_graph_t* g, ray_op_t* op, ray_t* lhs, ray_t* rhs) {
if (!lhs || RAY_IS_ERR(lhs)) return lhs;
if (!rhs || RAY_IS_ERR(rhs)) return rhs;
int8_t left_type = g->nodes[op->in_id[0]].out_type;
int8_t right_type = g->nodes[op->in_id[1]].out_type;
if (ray_has_timespan(lhs) || ray_has_timespan(rhs) ||
left_type == RAY_TIMESPAN || right_type == RAY_TIMESPAN ||
(op->opcode == OP_SUB && left_type == RAY_TIMESTAMP && right_type == RAY_TIMESTAMP)) {
ray_t left_atom, right_atom;
if (ray_is_atom(lhs) && (left_type == RAY_TIMESPAN || left_type == RAY_TIMESTAMP)) {
left_atom = *lhs; left_atom.type = -left_type; lhs = &left_atom;
}
if (ray_is_atom(rhs) && (right_type == RAY_TIMESPAN || right_type == RAY_TIMESTAMP)) {
right_atom = *rhs; right_atom.type = -right_type; rhs = &right_atom;
}
return ray_timespan_binary(op->opcode, lhs, rhs);
}
bool l_scalar = ray_is_atom(lhs) || (lhs->type > 0 && lhs->len == 1);
bool r_scalar = ray_is_atom(rhs) || (rhs->type > 0 && rhs->len == 1);
int64_t len = 1;
if (!l_scalar && !r_scalar) {
if (lhs->len != rhs->len) return ray_error("length", "expr eval: binary op operand lengths must match, got %lld and %lld", (long long)lhs->len, (long long)rhs->len);
len = lhs->len;
} else if (l_scalar && !r_scalar) {
len = rhs->len;
} else if (!l_scalar && r_scalar) {
len = lhs->len;
}
int8_t lt = lhs->type < 0 ? (int8_t)-lhs->type : lhs->type;
int8_t rt = rhs->type < 0 ? (int8_t)-rhs->type : rhs->type;
if ((op->opcode == OP_LT || op->opcode == OP_LE ||
op->opcode == OP_GT || op->opcode == OP_GE) &&
((lt == RAY_SYM && (rt == RAY_SYM || rt == RAY_STR)) ||
(rt == RAY_SYM && lt == RAY_STR)))
return exec_sym_order(op->opcode, lhs, rhs, l_scalar, r_scalar);
if (op->opcode == OP_POW) {
int8_t lt = ew_value_base_type(lhs);
int8_t rt = ew_value_base_type(rhs);
if (!ew_pow_type_admitted(lt) || !ew_pow_type_admitted(rt)) {
return ray_error("type",
"pow: expects numeric base and exponent, got %s and %s",
ray_type_name(lhs->type), ray_type_name(rhs->type));
}
}
int8_t out_type = ew_runtime_out_type(op->out_type,
!l_scalar ? lhs->type
: !r_scalar ? rhs->type
: (int8_t)(lhs->type < 0 ? -lhs->type : lhs->type));
ray_t* result = ray_vec_new(out_type, len);
if (!result || RAY_IS_ERR(result)) return result;
result->len = len;
{
bool l_is_str = (!l_scalar && lhs->type == RAY_STR);
bool r_is_str = (!r_scalar && rhs->type == RAY_STR);
bool l_atom_str = (l_scalar && (lhs->type == -RAY_STR
|| lhs->type == RAY_STR
|| (RAY_IS_SYM(lhs->type) && ray_is_atom(lhs))));
bool r_atom_str = (r_scalar && (rhs->type == -RAY_STR
|| rhs->type == RAY_STR
|| (RAY_IS_SYM(rhs->type) && ray_is_atom(rhs))));
if (l_is_str || r_is_str || (l_atom_str && r_atom_str)) {
uint16_t opc = op->opcode;
if (opc < OP_EQ || opc > OP_GE) { ray_release(result); return ray_error("type", "expr eval: string operands support only comparison operators, got %s and %s", ray_type_name(lhs->type), ray_type_name(rhs->type)); }
if (l_scalar && !l_atom_str) { ray_release(result); return ray_error("type", "expr eval: string comparison left operand must be a str/sym scalar, got %s", ray_type_name(lhs->type)); }
if (r_scalar && !r_atom_str) { ray_release(result); return ray_error("type", "expr eval: string comparison right operand must be a str/sym scalar, got %s", ray_type_name(rhs->type)); }
if (!l_scalar && !l_is_str) { ray_release(result); return ray_error("type", "expr eval: string comparison left operand must be a str column, got %s", ray_type_name(lhs->type)); }
if (!r_scalar && !r_is_str) { ray_release(result); return ray_error("type", "expr eval: string comparison right operand must be a str column, got %s", ray_type_name(rhs->type)); }
ray_pool_t* pool = ray_pool_get();
if (pool && len >= RAY_PARALLEL_THRESHOLD) {
par_binary_str_ctx_t ctx = {
.op = op, .lhs = lhs, .rhs = rhs, .result = result,
.l_scalar = l_scalar, .r_scalar = r_scalar,
};
ray_pool_dispatch(pool, par_binary_str_fn, &ctx, len);
fix_null_comparisons(lhs, rhs, result, l_scalar, r_scalar, len, op->opcode);
return result;
}
binary_range_str(op, lhs, rhs, result, l_scalar, r_scalar, 0, len);
fix_null_comparisons(lhs, rhs, result, l_scalar, r_scalar, len, op->opcode);
return result;
}
}
{
bool l_guid = (l_scalar ? (lhs->type == -RAY_GUID || lhs->type == RAY_GUID)
: lhs->type == RAY_GUID);
bool r_guid = (r_scalar ? (rhs->type == -RAY_GUID || rhs->type == RAY_GUID)
: rhs->type == RAY_GUID);
if (l_guid || r_guid) {
uint16_t opc = op->opcode;
if (!l_guid || !r_guid) {
ray_release(result);
return ray_error("type", "expr eval: cannot compare %s and %s",
ray_type_name(lhs->type), ray_type_name(rhs->type));
}
if (opc != OP_EQ && opc != OP_NE) {
ray_release(result);
return ray_error("type", "expr eval: guid operands support only == and !=");
}
uint8_t* out = (uint8_t*)ray_data(result);
const uint8_t* lb = (lhs->type == -RAY_GUID)
? (const uint8_t*)ray_data(lhs->obj)
: (const uint8_t*)ray_data(lhs);
const uint8_t* rb = (rhs->type == -RAY_GUID)
? (const uint8_t*)ray_data(rhs->obj)
: (const uint8_t*)ray_data(rhs);
for (int64_t i = 0; i < len; i++) {
const uint8_t* lp = lb + (l_scalar ? 0 : i * 16);
const uint8_t* rp = rb + (r_scalar ? 0 : i * 16);
uint8_t eq = (memcmp(lp, rp, 16) == 0) ? 1 : 0;
out[i] = (opc == OP_EQ) ? eq : (uint8_t)!eq;
}
return result;
}
}
{
bool l_ok = l_scalar || (lhs->type >= RAY_BOOL && lhs->type <= RAY_TIMESTAMP)
|| RAY_IS_SYM(lhs->type);
bool r_ok = r_scalar || (rhs->type >= RAY_BOOL && rhs->type <= RAY_TIMESTAMP)
|| RAY_IS_SYM(rhs->type);
if (!l_ok || !r_ok) {
ray_release(result);
return ray_error("type", "expr eval: unsupported column type in binary op, got %s and %s",
ray_type_name(lhs->type), ray_type_name(rhs->type));
}
}
{
bool l_atom_str = (l_scalar && lhs->type == -RAY_STR);
bool r_atom_str = (r_scalar && rhs->type == -RAY_STR);
if (l_atom_str || r_atom_str) {
uint16_t opc = op->opcode;
bool is_cmp = (opc >= OP_EQ && opc <= OP_GE);
if (!is_cmp && !RAY_IS_SYM(lhs->type) && !RAY_IS_SYM(rhs->type)) {
ray_release(result);
return ray_error("type", "expr eval: string atom is not valid in arithmetic, got %s and %s", ray_type_name(lhs->type), ray_type_name(rhs->type));
}
}
}
bool str_resolved = false;
int64_t resolved_sym_id = 0;
if (r_scalar && rhs->type == -RAY_STR &&
RAY_IS_SYM(lhs->type)) {
const char* s = ray_str_ptr(rhs);
size_t slen = ray_str_len(rhs);
resolved_sym_id = ray_sym_vec_lookup(lhs, s, slen);
str_resolved = true;
} else if (l_scalar && lhs->type == -RAY_STR &&
RAY_IS_SYM(rhs->type)) {
const char* s = ray_str_ptr(lhs);
size_t slen = ray_str_len(lhs);
resolved_sym_id = ray_sym_vec_lookup(rhs, s, slen);
str_resolved = true;
}
double l_f64_val = 0, r_f64_val = 0;
int64_t l_i64_val = 0, r_i64_val = 0;
if (l_scalar) {
if (str_resolved && lhs->type == -RAY_STR)
l_i64_val = resolved_sym_id;
else if (ray_is_atom(lhs)) {
if (lhs->type == -RAY_F64 || lhs->type == -RAY_F32)
l_f64_val = lhs->f64;
else if (lhs->type == -RAY_I32 || lhs->type == -RAY_DATE || lhs->type == -RAY_TIME)
l_i64_val = (int64_t)lhs->i32;
else if (lhs->type == -RAY_I16) l_i64_val = (int64_t)lhs->i16;
else if (lhs->type == -RAY_U8 || lhs->type == -RAY_BOOL) l_i64_val = (int64_t)lhs->u8;
else l_i64_val = lhs->i64;
} else {
int8_t t = lhs->type;
int64_t elem = 0;
void* data = resolve_vec_data(lhs, &elem);
if (t == RAY_F64) l_f64_val = ((double*)data)[elem];
else if (t == RAY_F32) l_f64_val = (double)((float*)data)[elem];
else l_i64_val = read_col_i64(data, elem, t, lhs->attrs);
}
}
if (r_scalar) {
if (str_resolved && rhs->type == -RAY_STR)
r_i64_val = resolved_sym_id;
else if (ray_is_atom(rhs)) {
if (rhs->type == -RAY_F64 || rhs->type == -RAY_F32)
r_f64_val = rhs->f64;
else if (rhs->type == -RAY_I32 || rhs->type == -RAY_DATE || rhs->type == -RAY_TIME)
r_i64_val = (int64_t)rhs->i32;
else if (rhs->type == -RAY_I16) r_i64_val = (int64_t)rhs->i16;
else if (rhs->type == -RAY_U8 || rhs->type == -RAY_BOOL) r_i64_val = (int64_t)rhs->u8;
else r_i64_val = rhs->i64;
} else {
int8_t t = rhs->type;
int64_t elem = 0;
void* data = resolve_vec_data(rhs, &elem);
if (t == RAY_F64) r_f64_val = ((double*)data)[elem];
else if (t == RAY_F32) r_f64_val = (double)((float*)data)[elem];
else r_i64_val = read_col_i64(data, elem, t, rhs->attrs);
}
}
if (!l_scalar && RAY_IS_SYM(lhs->type) && r_scalar &&
(rhs->type == -RAY_SYM || RAY_IS_SYM(rhs->type))) {
struct ray_sym_domain_s* cdom = ray_sym_vec_domain(lhs);
struct ray_sym_domain_s* sdom = (rhs->type == -RAY_SYM)
? ray_sym_runtime_domain() : ray_sym_vec_domain(rhs);
if (cdom != sdom) {
ray_t* s = ray_sym_domain_str(sdom, r_i64_val);
r_i64_val = s ? ray_sym_domain_find(cdom, ray_str_ptr(s),
ray_str_len(s)) : -1;
}
} else if (!r_scalar && RAY_IS_SYM(rhs->type) && l_scalar &&
(lhs->type == -RAY_SYM || RAY_IS_SYM(lhs->type))) {
struct ray_sym_domain_s* cdom = ray_sym_vec_domain(rhs);
struct ray_sym_domain_s* sdom = (lhs->type == -RAY_SYM)
? ray_sym_runtime_domain() : ray_sym_vec_domain(lhs);
if (cdom != sdom) {
ray_t* s = ray_sym_domain_str(sdom, l_i64_val);
l_i64_val = s ? ray_sym_domain_find(cdom, ray_str_ptr(s),
ray_str_len(s)) : -1;
}
}
ray_pool_t* pool = ray_pool_get();
if (pool && len >= RAY_PARALLEL_THRESHOLD) {
par_binary_ctx_t ctx = {
.op = op, .out_type = out_type,
.lhs = lhs, .rhs = rhs, .result = result,
.result_data = ray_data(result),
.l_scalar = l_scalar, .r_scalar = r_scalar,
.l_f64 = l_f64_val, .r_f64 = r_f64_val,
.l_i64 = l_i64_val, .r_i64 = r_i64_val,
.sel_flg = NULL,
.sel_offs = NULL,
.sel_idx = NULL,
.sel_n_segs = 0,
};
if (g && g->selection) {
ray_rowsel_t* sm = ray_rowsel_meta(g->selection);
if (sm->nrows == len) {
ctx.sel_flg = ray_rowsel_flags(g->selection);
ctx.sel_offs = ray_rowsel_offsets(g->selection);
ctx.sel_idx = ray_rowsel_idx(g->selection);
ctx.sel_n_segs = sm->n_segs;
}
}
ray_pool_dispatch(pool, par_binary_fn, &ctx, len);
} else {
binary_range(op, out_type, lhs, rhs, result, ray_data(result),
l_scalar, r_scalar,
l_f64_val, r_f64_val, l_i64_val, r_i64_val,
0, len);
}
if (!str_resolved) {
if (op_propagates_null(op->opcode))
propagate_nulls_binary(lhs, rhs, result, l_scalar, r_scalar, len);
else
fix_null_comparisons(lhs, rhs, result, l_scalar, r_scalar, len, op->opcode);
}
uint16_t opc = op->opcode;
if (opc == OP_DIV || opc == OP_IDIV || opc == OP_MOD) {
if (!r_scalar) {
int8_t rt = rhs->type;
if (rt == RAY_I64 || rt == RAY_TIMESTAMP) {
const int64_t* b = (const int64_t*)ray_data(rhs);
for (int64_t i = 0; i < len; i++)
if (b[i] == 0) ray_vec_set_null(result, i, true);
} else if (rt == RAY_I32 || rt == RAY_DATE || rt == RAY_TIME) {
const int32_t* b = (const int32_t*)ray_data(rhs);
for (int64_t i = 0; i < len; i++)
if (b[i] == 0) ray_vec_set_null(result, i, true);
} else if (rt == RAY_I16 && !(opc == OP_IDIV &&
(result->type == RAY_I64 || result->type == RAY_TIMESTAMP))) {
const int16_t* b = (const int16_t*)ray_data(rhs);
for (int64_t i = 0; i < len; i++)
if (b[i] == 0) ray_vec_set_null(result, i, true);
} else if ((rt == RAY_U8 && opc != OP_MOD) || rt == RAY_BOOL) {
const uint8_t* b = (const uint8_t*)ray_data(rhs);
for (int64_t i = 0; i < len; i++)
if (b[i] == 0) ray_vec_set_null(result, i, true);
} else if (rt == RAY_F64 && result->type != RAY_F64) {
const double* b = (const double*)ray_data(rhs);
for (int64_t i = 0; i < len; i++)
if (b[i] == 0.0) ray_vec_set_null(result, i, true);
} else if (rt == RAY_F32 && result->type != RAY_F32 &&
result->type != RAY_F64) {
const float* b = (const float*)ray_data(rhs);
for (int64_t i = 0; i < len; i++)
if (b[i] == 0.0f) ray_vec_set_null(result, i, true);
}
} else {
bool is_zero = false;
if (rhs->type == -RAY_F64 || rhs->type == RAY_F64 ||
rhs->type == -RAY_F32 || rhs->type == RAY_F32)
is_zero = (r_f64_val == 0.0);
else
is_zero = (r_i64_val == 0);
if (opc == OP_MOD && rt == RAY_U8)
is_zero = false;
if (is_zero) {
for (int64_t i = 0; i < len; i++)
ray_vec_set_null(result, i, true);
}
}
}
return result;
}