diff --git a/src/ops/query.c b/src/ops/query.c index 7c5f0e61..83324064 100644 --- a/src/ops/query.c +++ b/src/ops/query.c @@ -2006,6 +2006,7 @@ static int expr_contains_call_named(ray_t* expr, const char* name, size_t name_l } static ray_t* query_materialize_parted_col(ray_t* col); +static bool table_has_parted_columns(ray_t* tbl); /* True when a projection's TOP-LEVEL call is a "whole-column verb": a * length-changing / reordering builtin (distinct, asc, desc, reverse) that @@ -11321,6 +11322,35 @@ ray_t* ray_update(ray_t** args, int64_t n) { } if (tbl->type != RAY_TABLE) { int8_t tbl_t = tbl->type; ray_release(tbl); return ray_error("type", "update: `from:` must be a table, got %s", ray_type_name(tbl_t)); } + /* A parted table's data columns carry the RAY_PARTED_BASE wrapper type + * (which `ray_type_name` prints as "?"), and its partition key is + * RAY_MAPCOMMON. The update machinery below reads the original column + * through `ray_vec_new(ct, ...)` / `ray_data(col)` / the per-group gather + * and type-check against the wrapper type, none of which understand the + * parted/segmented shape — so `(update {col: … from: partedT})` failed + * with `expression type I64 does not match ? column` (or the `by:` path + * with `group: argument must be a vector`). `select` solves this by + * materialising parted columns on demand; replicate it here by flattening + * the whole table once so every branch below sees wrapped-free vectors. */ + if (table_has_parted_columns(tbl)) { + ray_t* flat_tbl = ray_table_new(ray_table_ncols(tbl)); + if (!flat_tbl || RAY_IS_ERR(flat_tbl)) { ray_release(tbl); return flat_tbl ? flat_tbl : ray_error("oom", NULL); } + int64_t nc = ray_table_ncols(tbl); + for (int64_t c = 0; c < nc; c++) { + ray_t* col = ray_table_get_col_idx(tbl, c); + ray_t* flat_col = query_materialize_parted_col(col); + if (!flat_col || RAY_IS_ERR(flat_col)) { + ray_release(flat_tbl); ray_release(tbl); + return flat_col ? flat_col : ray_error("oom", NULL); + } + flat_tbl = ray_table_add_col(flat_tbl, ray_table_col_name(tbl, c), flat_col); + ray_release(flat_col); + if (!flat_tbl || RAY_IS_ERR(flat_tbl)) { ray_release(tbl); return flat_tbl ? flat_tbl : ray_error("oom", NULL); } + } + ray_release(tbl); + tbl = flat_tbl; + } + ray_t* where_expr = dict_get(dict, "where"); ray_t* by_expr = dict_get(dict, "by"); @@ -11381,6 +11411,19 @@ ray_t* ray_update(ray_t** args, int64_t n) { if (RAY_IS_ERR(result)) { ray_release(groups); ray_release(tbl); DICT_VIEW_CLOSE(updv); return result; } for (int64_t c = 0; c < ncols; c++) { int64_t cn = ray_table_col_name(tbl, c); + /* Skip columns that the update dict replaces — the aggregate + * loop below adds the computed column for a REPLACED target, and + * ray_table_add_col always appends, so keeping both would create + * a duplicate name ([k v w w]) and `at` would read the stale + * original. Added columns (not in the source schema) fall out + * naturally. */ + int64_t replaced = 0; + for (int64_t d = 0; d + 1 < dict_n; d += 2) { + int64_t kid = dict_elems[d]->i64; + if (kid == from_id || kid == where_id || kid == by_id) continue; + if (kid == cn) { replaced = 1; break; } + } + if (replaced) continue; ray_t* col = ray_table_get_col_idx(tbl, c); ray_retain(col); result = ray_table_add_col(result, cn, col); diff --git a/test/rfl/query/query_update_coverage.rfl b/test/rfl/query/query_update_coverage.rfl index 3d28ee8a..1c9334b2 100644 --- a/test/rfl/query/query_update_coverage.rfl +++ b/test/rfl/query/query_update_coverage.rfl @@ -463,25 +463,31 @@ ;; ──────────────────────────────────────────────────────────────────── ;; I32 key: case RAY_I32 at lines 131-133 (set Tupd_i32by (table [k v] (list (as 'I32 [1 2 1 2 3]) [10 20 30 40 50]))) -;; update by: scatters aggregate back to original 5 rows (count unchanged) -;; Groups: k=1→sum=40, k=2→sum=60, k=3→sum=50. -;; Scatter fills only first occurrence per group; others remain 0. -;; Row values: [40, 60, 0, 0, 50] → sum = 150 +;; update by: broadcasts the aggregate back to every row of its group +;; (kdb `by:` semantics — the group sum lands on ALL member rows, not just +;; the first). Groups: k=1→sum=40, k=2→sum=60, k=3→sum=50. +;; Row values: [40, 60, 40, 60, 50] → sum = 250 +;; +;; USER-FACING FIX (was bug): pre-fix, updating an EXISTING column `v` via +;; `by:` silently did nothing — the aggregate was appended as a duplicate +;; column (schema `[k v v]`), so `(at U 'v)` kept reading the stale +;; original [10 20 30 40 50] → sum 150. From a user's perspective the +;; `by:`-update "succeeded" (no error) but the column never changed, and +;; `(key U)` unexpectedly listed the column name twice. (count (update {v: (sum v) by: k from: Tupd_i32by})) -- 5 -(sum (at (update {v: (sum v) by: k from: Tupd_i32by}) 'v)) -- 150 +(sum (at (update {v: (sum v) by: k from: Tupd_i32by}) 'v)) -- 250 ;; BOOL key: case RAY_BOOL (RAY_U8) at lines 135-136 (set Tupd_boolby (table [k v] (list [true false true false] [10 20 30 40]))) -;; Groups: k=true→sum=40, k=false→sum=60. -;; Scatter fills first occurrence; others remain 0: [40, 60, 0, 0] → sum = 100 +;; Groups: k=true→sum=40, k=false→sum=60 → broadcast [40, 60, 40, 60] → sum = 200 (count (update {v: (sum v) by: k from: Tupd_boolby})) -- 4 -(sum (at (update {v: (sum v) by: k from: Tupd_boolby}) 'v)) -- 100 +(sum (at (update {v: (sum v) by: k from: Tupd_boolby}) 'v)) -- 200 ;; F64 key: case RAY_F64 at line 137 (set Tupd_f64by (table [k v] (list [1.0 2.0 1.0 2.0] [10 20 30 40]))) -;; Groups: k=1.0→sum=40, k=2.0→sum=60. [40, 60, 0, 0] → sum = 100 +;; Groups: k=1.0→sum=40, k=2.0→sum=60 → broadcast [40, 60, 40, 60] → sum = 200 (count (update {v: (sum v) by: k from: Tupd_f64by})) -- 4 -(sum (at (update {v: (sum v) by: k from: Tupd_f64by}) 'v)) -- 100 +(sum (at (update {v: (sum v) by: k from: Tupd_f64by}) 'v)) -- 200 ;; ──────────────────────────────────────────────────────────────────── ;; WHERE-update SYM column with null in expr_vec (line 8707) diff --git a/test/rfl/query/update_parted.rfl b/test/rfl/query/update_parted.rfl new file mode 100644 index 00000000..36e2daa3 --- /dev/null +++ b/test/rfl/query/update_parted.rfl @@ -0,0 +1,93 @@ +;; Regression for `update` over PARTED tables (src/ops/query.c). +;; +;; Two distinct bugs, both hitting a `.db.parted.get` table as the `from:` +;; source: +;; +;; 1. PARTED columns carry the RAY_PARTED_BASE wrapper type (printed as "?") +;; and a RAY_MAPCOMMON partition key. ray_update read the original +;; column through `ray_vec_new(ct, ...)` / `ray_data(col)` / the grouped +;; gather, none of which understood the segmented shape — so MODIFYING an +;; existing column of a parted table failed with +;; `expression type I64 does not match ? column` (and `by:` with +;; `group: argument must be a vector`). Fix: flatten a parted input +;; table once, the way `select` does. +;; +;; 2. The `by:`-UPDATE branch duplicated an existing target column instead +;; of replacing it — `(update {w: (sum v) from: T by: k})` on a table that +;; already has `w` produced schema `[k v w w]` and `at` read the stale +;; original. Affected flat tables too. Fix: skip source columns that +;; the update dict replaces in the initial copy loop. +;; +;; Both are checked against flat-table oracles with identical data. + +;; ────────────── build a 2-partition parted table ────────────── +(.sys.exec "rm -rf /tmp/rfl_update_parted") +(set D1 (table [k v w] (list [1 2 3] [10 20 30] [100 200 300]))) +(set D2 (table [k v w] (list [1 1 2] [40 50 60] [400 500 600]))) +(.db.splayed.set "/tmp/rfl_update_parted/2024.01.01/t/" D1) +(.db.splayed.set "/tmp/rfl_update_parted/2024.01.02/t/" D2) +(set Pt (.db.parted.get "/tmp/rfl_update_parted/" 't)) +(set flat (table [k v w] (list [1 2 3 1 1 2] [10 20 30 40 50 60] [100 200 300 400 500 600]))) + +;; ────────────── fix 1a: modify existing column, no where ────────────── +;; USER-FACING: pre-fix, `(update {v: (+ v 100) from: Pt})` on a parted table +;; aborted with an immediate type error — +;; `error: type: update: expression type I64 does not match ? column` +;; so the user could not modify ANY existing column of a parted table at all. +(at (update {v: (+ v 100) from: Pt}) 'v) -- [110 120 130 140 150 160] +(count (update {v: (+ v 100) from: Pt})) -- 6 + +;; ────────────── fix 1b: scalar broadcast into existing column ────────────── +;; USER-FACING: pre-fix, `(update {v: 5 from: Pt})` failed with the same +;; `expression type I64 does not match ? column` — even a plain constant +;; cannot be written over an existing parted column. +(at (update {v: 5 from: Pt}) 'v) -- [5 5 5 5 5 5] + +;; ────────────── fix 1c: where-masked update of existing column ────────────── +;; USER-FACING: pre-fix, `where:`-masked writes failed with +;; `error: type: vec_new: type must be a positive concrete vector type, got ?` +;; so conditional in-place updates over parted data were impossible. +(at (update {v: 99 from: Pt where: (> k 1)}) 'v) -- [10 99 99 40 50 99] + +;; ────────────── fix 1d: update by: aggregate broadcast over parted ────────────── +;; USER-FACING: pre-fix, `by:`-grouped updates on parted tables errored with +;; `error: type: group: argument must be a vector or list, got ?` +;; while on FLAT tables (bug 2) the write silently appeared to do nothing: +;; the aggregate was appended as a duplicate column, the query "succeeded", +;; `(key U)` suddenly listed the column twice, and `(at U 'w)` kept returning +;; the stale original values. +;; k=1 row v={10,40,50} sum=100; k=2 {20,60} sum=80; k=3 {30} sum=30 +(at (update {w: (sum v) from: Pt by: k}) 'w) -- [100 80 30 100 100 80] +;; fix 2: schema must NOT duplicate the replaced target column +(key (update {w: (sum v) from: Pt by: k})) -- [date k v w] +;; flat oracle — same rows, same answers +(at (update {w: (sum v) from: flat by: k}) 'w) -- [100 80 30 100 100 80] +(key (update {w: (sum v) from: flat by: k})) -- [k v w] + +;; ────────────── fix 1e: mixed update — modify existing + add new over parted ────────────── +;; USER-FACING: any dict naming an EXISTING column tripped the fix-1a error +;; even when it also added new columns (`z`), so mixed single-pass updates +;; over parted tables were not possible. +(set U5 (update {v: (+ v 1) z: (* k 10) from: Pt})) +(at U5 'v) -- [11 21 31 41 51 61] +(at U5 'z) -- [10 20 30 10 10 20] +(key U5) -- [date k v w z] + +;; ────────────── in-place amend of a parted global (from: 'name) ────────────── +;; NOTE: `(update {from: 'G …})` amends the env global G in place and returns a +;; SYM (it does not return a new table) — for both flat and parted, so the +;; caller ignores the return and inspects G afterwards. +;; USER-FACING: pre-fix, even this over a parted global hit the same +;; `does not match ? column` error. +(set G (.db.parted.get "/tmp/rfl_update_parted/" 't)) +(update {from: 'G v: (* v 100)}) ;; amend G in place, ignore the sym return +(at G 'v) -- [1000 2000 3000 4000 5000 6000] +(key G) -- [date k v w] +;; flat oracle — same rows, same answer (in-place amend via symbol) +(set Gflat (table [k v w] (list [1 2 3 1 1 2] [10 20 30 40 50 60] [100 200 300 400 500 600]))) +(update {from: 'Gflat v: (* v 100)}) +(at Gflat 'v) -- [1000 2000 3000 4000 5000 6000] +(key Gflat) -- [k v w] + +(.sys.exec "rm -rf /tmp/rfl_update_parted") +(exit 0) \ No newline at end of file