fix(sort,query): radix key width across dispatch tasks; keep aggregate columns on zero groups

src/ops/query.c

@@ -2883,6 +2883,39 @@ static ray_t* nonagg_eval_per_group_core(ray_t* expr, ray_t* tbl,
     ray_t* result = NULL;       /* typed vec OR list col */
     int direct_typed = 0;       /* non-zero → result is a typed vec */
     int8_t typed_t = 0;         /* atom type sentinel for the typed path */
+    if (n_groups == 0) {
+        /* No group to walk, but the column must still exist.  Probe the
+         * expression once over an EMPTY slice of every referenced column so
+         * the result keeps the type a non-empty result would have had; an
+         * expression that cannot evaluate on zero rows yields a LIST column,
+         * the same shape the per-group path falls back to. */
+        ray_t* empty_idx = ray_vec_new(RAY_I64, 0);
+        int8_t vt = 0;
+        if (empty_idx && !RAY_IS_ERR(empty_idx)) {
+            empty_idx->len = 0;
+            int bound = 1;
+            for (int i = 0; i < n_cols && bound; i++) {
+                ray_t* err = bind_col_slice(col_syms[i], cols[i], empty_idx);
+                if (err) { ray_error_free(err); bound = 0; }
+            }
+            if (bound) {
+                ray_t* cell = ray_eval(expr);
+                if (cell && !RAY_IS_ERR(cell) && ray_is_lazy(cell)) cell = ray_lazy_materialize(cell);
+                if (cell && !RAY_IS_ERR(cell)) {
+                    int8_t t = cell->type;
+                    if (t < 0 && t != -RAY_SYM && t != -RAY_STR && t != -RAY_GUID) vt = (int8_t)(-t);
+                    ray_release(cell);
+                } else if (cell) ray_error_free(cell);
+            }
+            ray_release(empty_idx);
+        } else if (empty_idx) ray_error_free(empty_idx);
+        result = vt ? ray_vec_new(vt, 0) : ray_list_new(0);
+        if (result && !RAY_IS_ERR(result)) result->len = 0;
+        g_active_query_table = _aqt;
+        ray_env_pop_scope();
+        scratch_free(refs_hdr);
+        return result ? result : ray_error("oom", NULL);
+    }
 
     for (int64_t gi = 0; gi < n_groups; gi++) {
         ray_t* idx_list = feeder(gi, fstate);
@@ -3201,6 +3234,32 @@ static ray_t* eval_expr_whole_column(ray_t* expr, ray_t* tbl) {
     return result;
 }
 
+/* Zero groups still need the aggregate's column.  Run the aggregate once over
+ * an empty vector of the source's type so the empty column keeps the type a
+ * non-empty result would have had; if that probe cannot run, an empty LIST
+ * column stands in, the shape the per-group path itself falls back to. */
+static ray_t* empty_agg_column(ray_t* fn_name, ray_t* src) {
+    ray_t* out = NULL;
+    ray_t* fn_obj = fn_name ? ray_env_get(fn_name->i64) : NULL;
+    if (fn_obj && fn_obj->type == RAY_UNARY && src && ray_is_vec(src) && src->type != RAY_LIST) {
+        ray_t* empty = ray_vec_new(src->type, 0);
+        if (empty && !RAY_IS_ERR(empty)) {
+            empty->len = 0;
+            ray_t* v = ((ray_unary_fn)(uintptr_t)fn_obj->i64)(empty);
+            if (v && !RAY_IS_ERR(v) && ray_is_lazy(v)) v = ray_lazy_materialize(v);
+            if (v && !RAY_IS_ERR(v)) {
+                int8_t t = v->type;
+                if (t < 0 && t != -RAY_SYM && t != -RAY_STR && t != -RAY_GUID) out = ray_vec_new((int8_t)(-t), 0);
+                ray_release(v);
+            } else if (v) ray_error_free(v);
+            ray_release(empty);
+        } else if (empty) ray_error_free(empty);
+    }
+    if (!out || RAY_IS_ERR(out)) { if (out) ray_error_free(out); out = ray_list_new(0); }
+    if (out && !RAY_IS_ERR(out)) out->len = 0;
+    return out;
+}
+
 /* Streaming-style per-group AGG body, DAG flavor.  For an expression
  * like `(med v)` (head is RAY_FN_AGGR + RAY_UNARY, second elem is a
  * column ref or full-table-eval-able sub-expression), slice src per
@@ -3298,13 +3357,18 @@ static ray_t* aggr_unary_per_group_buf(ray_t* expr, ray_t* tbl,
         ray_release(agg_val);
     }
 
-    ray_release(idx_vec); ray_release(src);
+    ray_release(idx_vec);
     if (!agg_vec) {
-        /* No groups produced a value (all empty?) — return an empty typed
-         * vec sized n_groups; default to I64 for lack of a better guess. */
-        agg_vec = ray_vec_new(RAY_I64, n_groups);
-        if (agg_vec && !RAY_IS_ERR(agg_vec)) agg_vec->len = n_groups;
+        /* No group produced a value.  With zero groups the column must still
+         * carry the type a value would have had; with groups that were all
+         * empty keep the I64 default sized to n_groups. */
+        if (n_groups == 0) agg_vec = empty_agg_column(fn_name, src);
+        else {
+            agg_vec = ray_vec_new(RAY_I64, n_groups);
+            if (agg_vec && !RAY_IS_ERR(agg_vec)) agg_vec->len = n_groups;
+        }
     }
+    ray_release(src);
     return agg_vec;
 }
 
@@ -7826,6 +7890,7 @@ ray_t* ray_select(ray_t** args, int64_t n) {
                                 ray_release(agg_val);
                             }
                         }
+                        if (!agg_vec && out_groups == 0) agg_vec = empty_agg_column(agg_fn_name, src_col_val);
                         ray_release(src_col_val);
                         agg_names[n_agg_out] = kid;
                         agg_results[n_agg_out] = agg_vec;
@@ -8368,6 +8433,7 @@ ray_t* ray_select(ray_t** args, int64_t n) {
                         store_typed_elem(agg_vec, gi, agg_val);
                         ray_release(agg_val);
                     }
+                    if (!agg_vec && n_groups == 0) agg_vec = empty_agg_column(agg_fn_name, src_col_val);
                     ray_release(src_col_val);
                     agg_names[n_agg_out] = kid;
                     agg_results[n_agg_out] = agg_vec;

src/ops/sort.c

@@ -301,11 +301,15 @@ typedef struct {
     uint64_t*       pw_or;   /* per-worker XOR-diff accumulator */
 } key_range_ctx_t;
 
+/* Every task diffs against keys[0], the same reference the serial path uses.
+ * Diffing against the task's own first element only covers differences
+ * INSIDE a task; a byte that changes only between tasks is then invisible
+ * and the radix sort silently drops it. */
 static void key_range_fn(void* arg, uint32_t wid, int64_t start, int64_t end) {
     key_range_ctx_t* c = (key_range_ctx_t*)arg;
     const uint64_t* keys = c->keys;
     uint64_t local_or = c->pw_or[wid];
-    uint64_t first = keys[start];
+    uint64_t first = keys[0];
     for (int64_t i = start; i < end; i++)
         local_or |= keys[i] ^ first;
     c->pw_or[wid] = local_or;
@@ -323,14 +327,6 @@ uint8_t compute_key_nbytes(ray_pool_t* pool, const uint64_t* keys,
         ray_pool_dispatch(pool, key_range_fn, &ctx, n);
         diff = 0;
         for (uint32_t w = 0; w < nw; w++) diff |= pw_or[w];
-        /* Also XOR the first element from different worker ranges to
-         * catch cross-worker differences (workers' "first" may differ) */
-        uint64_t first = keys[0];
-        int64_t chunk = (n + nw - 1) / nw;
-        for (uint32_t w = 1; w < nw; w++) {
-            int64_t wstart = (int64_t)w * chunk;
-            if (wstart < n) diff |= keys[wstart] ^ first;
-        }
     } else {
         diff = 0;
         uint64_t first = keys[0];

test/rfl/group/zero_groups_keep_columns.rfl

@@ -0,0 +1,30 @@
+;; A grouped select whose selection matches no row must still return every
+;; requested column, whatever aggregate family produces it, for a single key
+;; and for a composite key alike.  Before the fix the streaming aggregates and
+;; the per-group projections were silently dropped (or the composite query
+;; failed outright) because their result columns were only created on the
+;; first group.
+(set T (table [k s v tm sy] (list ["a" "b" "c"] (as 'I64 [1 2 3]) [1.5 2.5 3.5] (as 'TIME [1 2 3]) ['x 'y 'z])))
+(set Z (select {from: T by: k where: (== k "missing") s: (sum s) m: (med v) q: (quantile v 0.5) mo: (mode v) f: (first tm) l: (last sy) tp: (top v 3) bt: (bot s 2) d: (distinct v) cd: (count (distinct v)) n: (count v) mn: (min sy) mx: (max s) av: (avg s) fs: (first s) ls: (last k)}))
+(set F (select {from: T by: k s: (sum s) m: (med v) q: (quantile v 0.5) mo: (mode v) f: (first tm) l: (last sy) tp: (top v 3) bt: (bot s 2) d: (distinct v) cd: (count (distinct v)) n: (count v) mn: (min sy) mx: (max s) av: (avg s) fs: (first s) ls: (last k)}))
+(count Z) -- 0
+(cols Z) -- (cols F)
+(count (cols Z)) -- 17
+;; the empty columns carry the type their non-empty counterparts have
+(type (at Z 's)) -- (type (at F 's))
+(type (at Z 'm)) -- (type (at F 'm))
+(type (at Z 'n)) -- (type (at F 'n))
+(type (at Z 'f)) -- (type (at F 'f))
+(type (at Z 'av)) -- (type (at F 'av))
+;; composite key, nothing selected
+(set ZC (select {from: T by: [k sy] where: (== k "missing") s: (sum s) n: (count v) f: (first v) m: (med v) d: (distinct v) cd: (count (distinct v)) tp: (top v 2)}))
+(set FC (select {from: T by: [k sy] s: (sum s) n: (count v) f: (first v) m: (med v) d: (distinct v) cd: (count (distinct v)) tp: (top v 2)}))
+(count ZC) -- 0
+(cols ZC) -- (cols FC)
+(count (cols ZC)) -- 9
+(type (at ZC 'n)) -- (type (at FC 'n))
+(type (at ZC 'f)) -- (type (at FC 'f))
+;; a non-empty selection is untouched
+(set P (select {from: T by: [k sy] where: (> s 1) s: (sum s) n: (count v) f: (first v) d: (distinct v)}))
+(count P) -- 2
+(cols P) -- [k sy s n f d]

test/rfl/sort/asc_key_width.rfl

@@ -0,0 +1,31 @@
+;; The radix key width is derived from the data range.  A high byte that only
+;; differs BETWEEN dispatch tasks (8192 rows each) must still widen the key;
+;; if it is missed, asc/desc silently sort by the low bytes alone.
+
+;; rows 40960..49151 carry bit 16, everything else lives in [0, 8192)
+(set N 200000)
+(set X (% (* 7919 (til N)) 8192))
+(set HI (as 'I64 (>= (til N) 40960)))
+(set HI (* HI (as 'I64 (< (til N) 49152))))
+(set V (+ X (* 65536 HI)))
+
+(set SA (asc V))
+(min (>= (- (at SA (+ 1 (til (- N 1)))) (at SA (til (- N 1)))) 0)) -- true
+(count SA) -- N
+(sum SA) -- (sum V)
+
+(set SD (desc V))
+(min (<= (- (at SD (+ 1 (til (- N 1)))) (at SD (til (- N 1)))) 0)) -- true
+
+(set T (table [v] (list V)))
+(set SV (at (select {from: T asc: v}) 'v))
+(min (>= (- (at SV (+ 1 (til (- N 1)))) (at SV (til (- N 1)))) 0)) -- true
+
+;; a nearly sorted vector: one swap past the 65536th element
+(set W (til 70000))
+(set W (at W (at (til 70000) (til 70000))))
+(set W (+ W (* (as 'I64 (== (til 70000) 66000)) 1)))
+(set W (- W (* (as 'I64 (== (til 70000) 66001)) 1)))
+(set SW (asc W))
+(min (>= (- (at SW (+ 1 (til 69999))) (at SW (til 69999))) 0)) -- true
+(at SW (+ 65990 (til 4))) -- [65990 65991 65992 65993]