fix(agg): reconcile scalar vs DAG aggregation (type-admission, STR/GUID, pearson, median, BOOL) + coverage

include/rayforce.h

@@ -459,6 +459,7 @@ ray_t*       ray_table_get_col(ray_t* tbl, int64_t name_id);
 ray_t*       ray_table_get_col_idx(ray_t* tbl, int64_t idx);
 int64_t     ray_table_col_name(ray_t* tbl, int64_t idx);
 void        ray_table_set_col_name(ray_t* tbl, int64_t idx, int64_t name_id);
+void        ray_table_set_col_idx(ray_t* tbl, int64_t idx, ray_t* col_vec);
 int64_t     ray_table_ncols(ray_t* tbl);
 int64_t     ray_table_nrows(ray_t* tbl);
 ray_t*       ray_table_schema(ray_t* tbl);

src/ops/agg.c

@@ -226,7 +226,9 @@ ray_t* ray_sum_fn(ray_t* x) {
         ray_retain(x); return x;
     }
     if (ray_is_vec(x)) {
-        if (x->type == RAY_DATE) return ray_error("type", NULL);
+        /* Canonical admission: numeric + TIME (duration); DATE/TIMESTAMP are
+         * absolute points and SYM/STR/GUID are non-numeric → type error. */
+        if (!agg_type_admitted(OP_SUM, x->type)) return ray_error("type", NULL);
         /* Narrow/temporal types need specific return constructors that the
          * DAG executor doesn't provide — use scalar path for these. */
         if (x->type == RAY_I32 || x->type == RAY_I16 || x->type == RAY_U8 ||
@@ -309,7 +311,12 @@ ray_t* ray_avg_fn(ray_t* x) {
         if (is_numeric(x)) return make_f64(as_f64(x));
         ray_retain(x); return x;
     }
-    if (ray_is_vec(x)) AGG_VEC_VIA_DAG(x, ray_avg);
+    if (ray_is_vec(x)) {
+        /* Canonical admission: numeric + temporal (→ F64); SYM/STR/GUID are
+         * non-numeric → type error (the DAG path otherwise averaged raw ids). */
+        if (!agg_type_admitted(OP_AVG, x->type)) return ray_error("type", NULL);
+        AGG_VEC_VIA_DAG(x, ray_avg);
+    }
     if (!is_list(x)) return ray_error("type", NULL);
     int64_t len = ray_len(x);
     if (len == 0) return ray_error("domain", NULL);
@@ -537,9 +544,17 @@ static ray_t* vec_to_f64_scratch(ray_t* x, double** out_vals) {
     } else if (x->type == RAY_I16) {
         int16_t* d = (int16_t*)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_U8) {
+    } else if (x->type == RAY_U8 || x->type == RAY_BOOL) {
         uint8_t* d = (uint8_t*)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_DATE || x->type == RAY_TIME) {
+        /* temporal stored as int32 days/ms — avg/var/stddev compute over the
+         * raw counts (result F64), per the canonical admission table. */
+        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]; }
+    } else if (x->type == RAY_TIMESTAMP) {
+        int64_t* d = (int64_t*)ray_data(x);
+        for (int64_t i = 0; i < len; i++) { if (!ray_vec_is_null(x, i)) vals[cnt++] = (double)d[i]; }
     } else {
         ray_release(scratch);
         return ray_error("type", NULL);

src/ops/collection.c

@@ -1591,10 +1591,16 @@ ray_t* ray_take_fn(ray_t* vec, ray_t* n_obj) {
 /* (at vec idx) or (at table 'col) — index into vector or table */
 ray_t* ray_at_fn(ray_t* vec, ray_t* idx) {
     if (ray_is_lazy(vec)) vec = ray_lazy_materialize(vec);
-    /* Table column access by symbol key — return the typed vector directly */
+    /* Table column access by symbol key — return the typed vector directly.
+     * A column loaded from a splayed/parted table (.db.parted.get) is still in
+     * segmented (RAY_IS_PARTED) form; the query path flattens it lazily, but a
+     * direct `(at table 'col)` must materialize it so downstream `(at col i)` /
+     * count / formatting see a dense vector instead of failing with `type`. */
     if (vec->type == RAY_TABLE && idx->type == -RAY_SYM) {
         ray_t* col = ray_table_get_col(vec, idx->i64);
         if (!col) return ray_error("domain", NULL);
+        if (RAY_IS_PARTED(col->type)) return parted_to_flat_vec(col);
+        if (col->type == RAY_MAPCOMMON) return materialize_mapcommon(col);
         ray_retain(col);
         return col;
     }

src/ops/exec.c

@@ -1084,6 +1084,18 @@ static ray_t* exec_node_inner(ray_graph_t* g, ray_op_t* op) {
                 input = compacted;
                 own_input = true;
             }
+            /* Canonical aggregand type-admission — reject non-admitted element
+             * types (SYM/STR/GUID everywhere; DATE/TIMESTAMP for sum) so the
+             * DAG reduction matches the scalar builtins instead of aggregating
+             * raw ids / bytes / date counts. */
+            if (input && input->type != RAY_TABLE) {
+                int8_t et = RAY_IS_PARTED(input->type)
+                          ? (int8_t)RAY_PARTED_BASETYPE(input->type) : input->type;
+                if (et > 0 && !agg_type_admitted(op->opcode, et)) {
+                    if (own_input) ray_release(input);
+                    return ray_error("type", NULL);
+                }
+            }
             ray_t* result = exec_reduction(g, op, input);
             if (own_input) ray_release(input);
             return result;