From fc2cbaf5c58c9cad43a6bdfb0790b4bef738ee6d Mon Sep 17 00:00:00 2001 From: Serhii Savchuk Date: Thu, 17 Sep 2026 10:50:39 +0300 Subject: [PATCH 1/2] fix(sort): derive the radix key width against keys[0] in every task MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit compute_key_nbytes lets each dispatch task OR together `keys[i] ^ keys[start]`, the differences against the task's own first element, and only samples nw-1 cross-task points on the assumption that workers own contiguous chunks. The pool dispatches 8192-row tasks, so a high byte that changes only between tasks — never inside one and never at a sampled row — is dropped from the key width, and asc/desc/xasc silently sort by the low bytes alone. 200000 values where rows 40960..49151 carry bit 16 come back with 65536 in the middle of the zeros; at -c 1 any nearly sorted vector past 65536 rows with one swap sorts by its low 16 bits. Every task now diffs against keys[0], the reference the serial path already uses, which covers every bit exactly; the sampling loop goes. Co-Authored-By: Claude Fable 5.1 --- src/ops/sort.c | 14 +++++--------- test/rfl/sort/asc_key_width.rfl | 31 +++++++++++++++++++++++++++++++ 2 files changed, 36 insertions(+), 9 deletions(-) create mode 100644 test/rfl/sort/asc_key_width.rfl 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/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] From fdb8e5abed09955b7bf3adb2bb912b52b94ed76f Mon Sep 17 00:00:00 2001 From: Serhii Savchuk Date: Thu, 17 Sep 2026 10:50:39 +0300 Subject: [PATCH 2/2] fix(query): keep every aggregate column when a grouped select matches no row The eval-fallback grouping paths build each result column lazily, on the first group: the streaming aggregates allocate `agg_vec` from the first value's type, and nonagg_eval_per_group_core only creates its result inside the per-group loop. With zero groups both stay NULL, so a single-key select silently dropped those columns (17 requested, 2 returned) and a composite key failed with "per-group projection evaluation failed". Probe the expression once over an empty slice (projections) or the aggregate over an empty vector of the source type (streaming) and emit an empty column of that type; when the probe cannot run, an empty LIST column stands in. aggr_unary_per_group_buf's I64 guess takes the same probe. Co-Authored-By: Claude Fable 5.1 --- src/ops/query.c | 76 +++++++++++++++++++-- test/rfl/group/zero_groups_keep_columns.rfl | 30 ++++++++ 2 files changed, 101 insertions(+), 5 deletions(-) create mode 100644 test/rfl/group/zero_groups_keep_columns.rfl 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/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]