diff --git a/src/ops/query.c b/src/ops/query.c index 6196d599..26300fe1 100644 --- a/src/ops/query.c +++ b/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; diff --git a/src/ops/sort.c b/src/ops/sort.c index daed7243..6118baea 100644 --- a/src/ops/sort.c +++ b/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]; diff --git a/test/rfl/group/zero_groups_keep_columns.rfl b/test/rfl/group/zero_groups_keep_columns.rfl new file mode 100644 index 00000000..7b1bc593 --- /dev/null +++ b/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] diff --git a/test/rfl/sort/asc_key_width.rfl b/test/rfl/sort/asc_key_width.rfl new file mode 100644 index 00000000..ad5b7698 --- /dev/null +++ b/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]