Skip to content

Agent/indexed statistical groups #326

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
singaraiona merged 12 commits into from
Jul 11, 2026
Merged

Agent/indexed statistical groups #326

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
2281d81
perf(aggr): fuse indexed statistical groups
singaraiona Jul 9, 2026
4203946
feat(mem): add scoped query measurements
singaraiona Jul 10, 2026
ab4e864
perf(query): accelerate indexed filters and parted groups
singaraiona Jul 10, 2026
6b2182b
fix(audit): restore linear running windows
singaraiona Jul 10, 2026
ac38240
Merge branch 'dev' into agent/indexed-statistical-groups
singaraiona Jul 10, 2026
e0ffa67
perf(aggr): remove fixed grouping limits
singaraiona Jul 10, 2026
48eefc8
fix(core): make persisted F32 queryable
singaraiona Jul 10, 2026
5763645
fix(rowsel): initialize mixed intersections
singaraiona Jul 10, 2026
8b063d9
fix(csv): serialize F32 columns
singaraiona Jul 10, 2026
6879ff3
fix(ops): complete F32 query execution
singaraiona Jul 10, 2026
e5741a4
fix(lang): compile null keyword in lambdas
singaraiona Jul 10, 2026
b9a08f7
chore(core): merge dev into indexed statistical groups
singaraiona Jul 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 21 additions & 0 deletions src/io/csv.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1037,6 +1037,7 @@ static void csv_parse_fn(void* arg, uint32_t worker_id,
case CSV_TYPE_I16: ((int16_t*)ctx->col_data[c])[row] = NULL_I16; break;
case CSV_TYPE_I32: ((int32_t*)ctx->col_data[c])[row] = NULL_I32; break;
case CSV_TYPE_I64: ((int64_t*)ctx->col_data[c])[row] = NULL_I64; break;
case CSV_TYPE_F32: ((float*)ctx->col_data[c])[row] = NULL_F32; break;
case CSV_TYPE_F64: ((double*)ctx->col_data[c])[row] = NULL_F64; break;
case CSV_TYPE_DATE: ((int32_t*)ctx->col_data[c])[row] = NULL_I32; break;
case CSV_TYPE_TIME: ((int32_t*)ctx->col_data[c])[row] = NULL_I32; break;
Expand Down Expand Up @@ -1114,6 +1115,13 @@ static void csv_parse_fn(void* arg, uint32_t worker_id,
if (is_null) my_had_null[c] = true;
break;
}
case CSV_TYPE_F32: {
bool is_null;
double v = fast_f64(fld, flen, &is_null);
((float*)ctx->col_data[c])[row] = is_null ? NULL_F32 : (float)v;
if (is_null) my_had_null[c] = true;
break;
}
case CSV_TYPE_DATE: {
bool is_null;
int32_t v = fast_date(fld, flen, &is_null);
Expand Down Expand Up @@ -1206,6 +1214,7 @@ static void csv_parse_serial(const char* buf, size_t buf_size,
case CSV_TYPE_I16: ((int16_t*)col_data[c])[row] = NULL_I16; break;
case CSV_TYPE_I32: ((int32_t*)col_data[c])[row] = NULL_I32; break;
case CSV_TYPE_I64: ((int64_t*)col_data[c])[row] = NULL_I64; break;
case CSV_TYPE_F32: ((float*)col_data[c])[row] = NULL_F32; break;
case CSV_TYPE_F64: ((double*)col_data[c])[row] = NULL_F64; break;
case CSV_TYPE_DATE: ((int32_t*)col_data[c])[row] = NULL_I32; break;
case CSV_TYPE_TIME: ((int32_t*)col_data[c])[row] = NULL_I32; break;
Expand Down Expand Up @@ -1281,6 +1290,13 @@ static void csv_parse_serial(const char* buf, size_t buf_size,
if (is_null) col_had_null[c] = true;
break;
}
case CSV_TYPE_F32: {
bool is_null;
double v = fast_f64(fld, flen, &is_null);
((float*)col_data[c])[row] = is_null ? NULL_F32 : (float)v;
if (is_null) col_had_null[c] = true;
break;
}
case CSV_TYPE_DATE: {
bool is_null;
int32_t v = fast_date(fld, flen, &is_null);
Expand Down Expand Up @@ -1629,6 +1645,7 @@ static ray_t* csv_materialize_rows(const char* buf, size_t file_size,
case RAY_I16: parse_types[c] = CSV_TYPE_I16; break;
case RAY_I32: parse_types[c] = CSV_TYPE_I32; break;
case RAY_I64: parse_types[c] = CSV_TYPE_I64; break;
case RAY_F32: parse_types[c] = CSV_TYPE_F32; break;
case RAY_F64: parse_types[c] = CSV_TYPE_F64; break;
case RAY_DATE: parse_types[c] = CSV_TYPE_DATE; break;
case RAY_TIME: parse_types[c] = CSV_TYPE_TIME; break;
Expand Down Expand Up @@ -2037,6 +2054,7 @@ ray_t* ray_read_csv_named_opts(const char* path, char delimiter, bool header,
case RAY_I16: parse_types[c] = CSV_TYPE_I16; break;
case RAY_I32: parse_types[c] = CSV_TYPE_I32; break;
case RAY_I64: parse_types[c] = CSV_TYPE_I64; break;
case RAY_F32: parse_types[c] = CSV_TYPE_F32; break;
case RAY_F64: parse_types[c] = CSV_TYPE_F64; break;
case RAY_DATE: parse_types[c] = CSV_TYPE_DATE; break;
case RAY_TIME: parse_types[c] = CSV_TYPE_TIME; break;
Expand Down Expand Up @@ -3128,6 +3146,9 @@ static void csv_write_cell(csv_writer_t* w, const csv_col_info_t* ci, int64_t r)
case RAY_F64:
csv_write_f64(w, ((const double*)d)[dr]);
break;
case RAY_F32:
csv_write_f64(w, (double)((const float*)d)[dr]);
break;
case RAY_DATE:
csv_write_date(w, ((const int32_t*)d)[dr]);
break;
Expand Down
1 change: 1 addition & 0 deletions src/io/csv.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@ typedef enum {
CSV_TYPE_U8,
CSV_TYPE_I16,
CSV_TYPE_I32,
CSV_TYPE_F32,
/* Marker for "pick the narrowest integer width automatically" — used only
* during schema processing, never returned by the resolver. */
CSV_TYPE_AUTO
Expand Down
13 changes: 11 additions & 2 deletions src/lang/compile.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -192,7 +192,7 @@ static void patch_jump(compiler_t *c, int32_t pos) {
}

/* Cached sym IDs for special forms */
static _Thread_local int64_t sf_set = -1, sf_let = -1, sf_if = -1, sf_do = -1, sf_fn = -1, sf_self = -1, sf_try = -1, sf_return = -1;
static _Thread_local int64_t sf_set = -1, sf_let = -1, sf_if = -1, sf_do = -1, sf_fn = -1, sf_self = -1, sf_try = -1, sf_return = -1, sf_null = -1;
static _Thread_local int64_t sf_eval = -1, sf_resolve = -1;

static void init_sf_syms(void) {
Expand All @@ -205,6 +205,7 @@ static void init_sf_syms(void) {
sf_self = ray_sym_intern("self", 4);
sf_try = ray_sym_intern("try", 3);
sf_return = ray_sym_intern("return", 6);
sf_null = ray_sym_intern("null", 4);
sf_eval = ray_sym_intern("eval", 4);
sf_resolve = ray_sym_intern("resolve", 7);
}
Expand Down Expand Up @@ -522,6 +523,14 @@ static void compile_expr(compiler_t *c, ray_t *ast) {

if (ray_is_atom(ast)) {
if (ast->type == -RAY_SYM && !(ast->attrs & ATTR_QUOTED)) {
/* The tree walker treats bare `null` as the singleton value, not
* an environment lookup. Compiled lambdas must do the same;
* otherwise any branch returning null fails with a name error. */
init_sf_syms();
if (ast->i64 == sf_null) {
emit_const(c, add_constant(c, RAY_NULL_OBJ));
return;
}
int32_t slot = find_local(c, ast->i64);
if (slot >= 0) {
emit(c, OP_LOADENV);
Expand Down Expand Up @@ -639,6 +648,6 @@ ray_span_t ray_bc_dbg_get(ray_t* dbg, int32_t ip) {
}

void ray_compile_reset(void) {
sf_set = sf_let = sf_if = sf_do = sf_fn = sf_self = sf_try = sf_return = -1;
sf_set = sf_let = sf_if = sf_do = sf_fn = sf_self = sf_try = sf_return = sf_null = -1;
sf_eval = sf_resolve = -1;
}
6 changes: 6 additions & 0 deletions src/lang/internal.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,6 +138,10 @@ static inline int is_float_op(ray_t* a, ray_t* b) {
a->type == -RAY_F32 || b->type == -RAY_F32;
}

static inline int both_f32_atoms(ray_t* a, ray_t* b) {
return a->type == -RAY_F32 && b->type == -RAY_F32;
}

/* ══════════════════════════════════════════
* Null/type helpers
* ══════════════════════════════════════════ */
Expand All @@ -148,6 +152,8 @@ static inline int is_float_op(ray_t* a, ray_t* b) {
static inline ray_t* null_for_promoted(ray_t* a, ray_t* b) {
if (a->type == -RAY_F64 || b->type == -RAY_F64)
return ray_typed_null(-RAY_F64);
if (a->type == -RAY_F32 || b->type == -RAY_F32)
return ray_typed_null(both_f32_atoms(a, b) ? -RAY_F32 : -RAY_F64);
if (a->type == -RAY_I64 || b->type == -RAY_I64)
return ray_typed_null(-RAY_I64);
if (a->type == -RAY_I32 || b->type == -RAY_I32)
Expand Down
53 changes: 32 additions & 21 deletions src/ops/agg.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,7 @@ static void nth_element_dbl(double* a, int64_t lo, int64_t hi, int64_t k) {

static int agg_parted_numeric_base(int8_t t) {
return t == RAY_BOOL || t == RAY_U8 || t == RAY_I16 ||
t == RAY_I32 || t == RAY_I64 || t == RAY_F64 ||
t == RAY_I32 || t == RAY_I64 || t == RAY_F32 || t == RAY_F64 ||
t == RAY_DATE || t == RAY_TIME || t == RAY_TIMESTAMP;
}

Expand Down Expand Up @@ -130,15 +130,17 @@ static ray_t* agg_parted_sum(ray_t* x) {
if (!agg_parted_numeric_base(base) || base == RAY_DATE)
return ray_error("type", "sum expects a numeric or time-duration parted column, got %s", ray_type_name(base));
ray_t** segs = (ray_t**)ray_data(x);
if (base == RAY_F64) {
if (base == RAY_F64 || base == RAY_F32) {
double sum = 0.0;
for (int64_t s = 0; s < x->len; s++) {
ray_t* seg = segs[s];
if (!seg) continue;
double* d = (double*)ray_data(seg);
int has_nulls = (seg->attrs & RAY_ATTR_HAS_NULLS) != 0;
for (int64_t i = 0; i < seg->len; i++)
if (!has_nulls || !ray_vec_is_null(seg, i)) sum += d[i];
if (!has_nulls || !ray_vec_is_null(seg, i)) {
if (base == RAY_F64) sum += ((double*)ray_data(seg))[i];
else sum += (double)((float*)ray_data(seg))[i];
}
}
return make_f64(sum);
}
Expand All @@ -165,11 +167,12 @@ static ray_t* agg_parted_avg(ray_t* x) {
ray_t* seg = segs[s];
if (!seg) continue;
int has_nulls = (seg->attrs & RAY_ATTR_HAS_NULLS) != 0;
if (base == RAY_F64) {
double* d = (double*)ray_data(seg);
if (base == RAY_F64 || base == RAY_F32) {
for (int64_t i = 0; i < seg->len; i++) {
if (has_nulls && ray_vec_is_null(seg, i)) continue;
sum += d[i]; cnt++;
if (base == RAY_F64) sum += ((double*)ray_data(seg))[i];
else sum += (double)((float*)ray_data(seg))[i];
cnt++;
}
} else {
for (int64_t i = 0; i < seg->len; i++) {
Expand All @@ -189,16 +192,16 @@ static ray_t* agg_parted_prod(ray_t* x) {
return ray_error("type", "prod expects a numeric parted column, got %s", ray_type_name(base));
}
ray_t** segs = (ray_t**)ray_data(x);
if (base == RAY_F64) {
if (base == RAY_F64 || base == RAY_F32) {
double prod = 1.0;
for (int64_t s = 0; s < x->len; s++) {
ray_t* seg = segs[s];
if (!seg) continue;
double* d = (double*)ray_data(seg);
int has_nulls = (seg->attrs & RAY_ATTR_HAS_NULLS) != 0;
for (int64_t i = 0; i < seg->len; i++) {
if (has_nulls && ray_vec_is_null(seg, i)) continue;
prod *= d[i];
if (base == RAY_F64) prod *= ((double*)ray_data(seg))[i];
else prod *= (double)((float*)ray_data(seg))[i];
}
}
return make_f64(prod);
Expand Down Expand Up @@ -228,11 +231,12 @@ static ray_t* agg_parted_minmax(ray_t* x, int want_max) {
ray_t* seg = segs[s];
if (!seg) continue;
int has_nulls = (seg->attrs & RAY_ATTR_HAS_NULLS) != 0;
if (base == RAY_F64) {
double* d = (double*)ray_data(seg);
if (base == RAY_F64 || base == RAY_F32) {
for (int64_t i = 0; i < seg->len; i++) {
if (has_nulls && ray_vec_is_null(seg, i)) continue;
double v = d[i];
double v = base == RAY_F64
? ((double*)ray_data(seg))[i]
: (double)((float*)ray_data(seg))[i];
if (!found || (want_max ? v > best_f : v < best_f)) {
best_f = v; found = 1;
}
Expand All @@ -249,18 +253,20 @@ static ray_t* agg_parted_minmax(ray_t* x, int want_max) {
}
if (!found) return ray_typed_null(-base);
if (base == RAY_F64) return make_f64(best_f);
if (base == RAY_F32) return ray_f32((float)best_f);
return agg_atom_i64_for_type(base, best_i);
}

static int agg_numeric_flat_type(int8_t t) {
return t == RAY_BOOL || t == RAY_U8 || t == RAY_I16 ||
t == RAY_I32 || t == RAY_I64 || t == RAY_F64;
t == RAY_I32 || t == RAY_I64 || t == RAY_F32 || t == RAY_F64;
}

static int agg_read_vec_f64(ray_t* v, int64_t i, double* out) {
void* d = ray_data(v);
switch (v->type) {
case RAY_F64: *out = ((double*)d)[i]; return 1;
case RAY_F32: *out = (double)((float*)d)[i]; return 1;
case RAY_I64: *out = (double)((int64_t*)d)[i]; return 1;
case RAY_I32: *out = (double)((int32_t*)d)[i]; return 1;
case RAY_I16: *out = (double)((int16_t*)d)[i]; return 1;
Expand Down Expand Up @@ -362,7 +368,9 @@ static ray_t* agg_pair_vec(ray_t* x, ray_t* y, uint16_t op) {
double xv = 0.0, yv = 0.0;
agg_read_vec_f64(x, i, &xv);
agg_read_vec_f64(y, i, &yv);
sx += xv; sy += yv; sxx += xv * xv; syy += yv * yv; sxy += xv * yv; cnt++;
sx += xv; sy += yv; sxx += xv * xv; syy += yv * yv;
sxy += xv * yv;
cnt++;
}

if (op == OP_WSUM) return ray_f64(ray_f64_fin(sxy));
Expand Down Expand Up @@ -449,10 +457,10 @@ ray_t* ray_sum_fn(ray_t* x) {
for (int64_t i = 0; i < len; i++) {
if (!is_numeric(elems[i])) return ray_error("type", "sum: list elements must be numeric, got %s", ray_type_name(elems[i]->type));
if (RAY_ATOM_IS_NULL(elems[i])) {
if (elems[i]->type == -RAY_F64) has_float = 1;
if (elems[i]->type == -RAY_F64 || elems[i]->type == -RAY_F32) has_float = 1;
continue;
}
if (elems[i]->type == -RAY_F64) { has_float = 1; fsum += elems[i]->f64; }
if (elems[i]->type == -RAY_F64 || elems[i]->type == -RAY_F32) { has_float = 1; fsum += elems[i]->f64; }
else if (elems[i]->type == -RAY_I64) { isum += elems[i]->i64; fsum += (double)elems[i]->i64; }
else { int64_t v = (int64_t)as_f64(elems[i]); isum += v; fsum += (double)v; }
}
Expand All @@ -479,10 +487,10 @@ ray_t* ray_prod_fn(ray_t* x) {
if (!is_numeric(elems[i]))
return ray_error("type", "prod: list elements must be numeric, got %s", ray_type_name(elems[i]->type));
if (RAY_ATOM_IS_NULL(elems[i])) {
if (elems[i]->type == -RAY_F64) has_float = 1;
if (elems[i]->type == -RAY_F64 || elems[i]->type == -RAY_F32) has_float = 1;
continue;
}
if (elems[i]->type == -RAY_F64) {
if (elems[i]->type == -RAY_F64 || elems[i]->type == -RAY_F32) {
has_float = 1;
fprod *= elems[i]->f64;
} else {
Expand Down Expand Up @@ -613,7 +621,7 @@ ray_t* ray_min_fn(ray_t* x) {
double fmin = 0; int64_t imin = 0;
for (int64_t i = 0; i < len; i++) {
if (!is_numeric(elems[i])) return ray_error("type", "min: list elements must be numeric, got %s", ray_type_name(elems[i]->type));
if (elems[i]->type == -RAY_F64) has_float = 1;
if (elems[i]->type == -RAY_F64 || elems[i]->type == -RAY_F32) has_float = 1;
if (RAY_ATOM_IS_NULL(elems[i])) continue;
double v = as_f64(elems[i]);
if (!found || v < fmin) { fmin = v; imin = elems[i]->type == -RAY_I64 ? elems[i]->i64 : 0; found = 1; }
Expand Down Expand Up @@ -667,7 +675,7 @@ ray_t* ray_max_fn(ray_t* x) {
double fmax = 0; int64_t imax = 0;
for (int64_t i = 0; i < len; i++) {
if (!is_numeric(elems[i])) return ray_error("type", "max: list elements must be numeric, got %s", ray_type_name(elems[i]->type));
if (elems[i]->type == -RAY_F64) has_float = 1;
if (elems[i]->type == -RAY_F64 || elems[i]->type == -RAY_F32) has_float = 1;
if (RAY_ATOM_IS_NULL(elems[i])) continue;
double v = as_f64(elems[i]);
if (!found || v > fmax) { fmax = v; imax = elems[i]->type == -RAY_I64 ? elems[i]->i64 : 0; found = 1; }
Expand Down Expand Up @@ -770,6 +778,9 @@ static ray_t* vec_to_f64_scratch(ray_t* x, double** out_vals) {
} else if (x->type == RAY_F64) {
double* d = (double*)ray_data(x);
for (int64_t i = 0; i < len; i++) { if (!ray_vec_is_null(x, i)) vals[cnt++] = d[i]; }
} else if (x->type == RAY_F32) {
float* d = (float*)ray_data(x);
for (int64_t i = 0; i < len; i++) { if (!ray_vec_is_null(x, i)) vals[cnt++] = (double)d[i]; }
} else if (x->type == RAY_I32) {
int32_t* d = (int32_t*)ray_data(x);
for (int64_t i = 0; i < len; i++) { if (!ray_vec_is_null(x, i)) vals[cnt++] = (double)d[i]; }
Expand Down
Loading
Loading