diff --git a/docs/docs/storage/index.md b/docs/docs/storage/index.md index c5741660..62b72116 100644 --- a/docs/docs/storage/index.md +++ b/docs/docs/storage/index.md @@ -129,6 +129,9 @@ ray_t* trades = ray_read_parted("db", "trades"); !!! note "Rayfall builtin" Use `.db.parted.get` from Rayfall to load partitioned tables: `(.db.parted.get "db" 'trades)`. See the [Rayfall Storage Builtins](#rayfall-storage) section below. +!!! warning "`update` over a partitioned table materializes in memory" + `update` on a partitioned table flattens the whole table into memory first — the parted/`MAPCOMMON` columns cannot be mutated in place — so the result is an ordinary in-memory table. It is **not** written back to the store: re-reading the root returns the original values, and the returned table loses its parted / memory-mapped identity. + ### Partition Pruning The query optimizer recognizes predicates on the `MAPCOMMON` column and eliminates entire partitions from the scan plan. This means a query filtering on a single date in a year of data only touches 1/365th of the files on disk — with zero per-row cost for the pruned partitions. diff --git a/src/ops/query.c b/src/ops/query.c index deea9da7..c39f030a 100644 --- a/src/ops/query.c +++ b/src/ops/query.c @@ -2073,6 +2073,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 @@ -11492,6 +11493,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"); @@ -11546,37 +11576,55 @@ ray_t* ray_update(ray_t** args, int64_t n) { if (RAY_IS_ERR(groups)) { ray_release(tbl); DICT_VIEW_CLOSE(updv); return groups; } } - /* Start with a copy of the original table */ int64_t ncols = ray_table_ncols(tbl); - ray_t* result = ray_table_new((int32_t)ncols); - 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); - ray_t* col = ray_table_get_col_idx(tbl, c); - ray_retain(col); - result = ray_table_add_col(result, cn, col); - ray_release(col); - if (RAY_IS_ERR(result)) { ray_release(groups); ray_release(tbl); DICT_VIEW_CLOSE(updv); return result; } - } + int64_t ngroups = groups->len / 2; + ray_t** gdata = (ray_t**)ray_data(groups); + if (!gdata) { ray_release(groups); ray_release(tbl); DICT_VIEW_CLOSE(updv); return ray_error("oom", NULL); } + + int64_t n_updates = 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; + n_updates++; + } + size_t upd_slots = (size_t)(n_updates ? n_updates : 1); + ray_t* upd_hdr = NULL; + int64_t* upd_names = (int64_t*)scratch_calloc(&upd_hdr, + upd_slots * sizeof(int64_t) + + upd_slots * sizeof(ray_t*) + + upd_slots * sizeof(uint8_t)); + if (!upd_names) { ray_release(groups); ray_release(tbl); DICT_VIEW_CLOSE(updv); return ray_error("oom", NULL); } + ray_t** upd_cols = (ray_t**)(upd_names + upd_slots); + uint8_t* upd_used = (uint8_t*)(upd_cols + upd_slots); + + #define UPDATE_BY_CLEANUP_COLS() do { \ + for (int64_t _ui = 0; _ui < n_updates; _ui++) \ + if (upd_cols[_ui]) ray_release(upd_cols[_ui]); \ + scratch_free(upd_hdr); \ + } while (0) /* For each aggregate expression, compute per group and broadcast */ + int64_t upd_i = 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; ray_t* agg_expr = dict_elems[d + 1]; - /* Evaluate the aggregate for each group and broadcast */ - ray_t* grp_items = (ray_t**)ray_data(groups) ? groups : NULL; - if (!grp_items) { ray_release(result); ray_release(groups); ray_release(tbl); DICT_VIEW_CLOSE(updv); return ray_error("oom", NULL); } - int64_t ngroups = groups->len / 2; - ray_t** gdata = (ray_t**)ray_data(groups); - /* We need to evaluate the aggregate per group. * Build the result column by evaluating the expression on each group's subset. */ - ray_t* out_col = ray_vec_new(RAY_I64, nrows2); /* will be resized to correct type */ - if (RAY_IS_ERR(out_col)) { ray_release(result); ray_release(groups); ray_release(tbl); DICT_VIEW_CLOSE(updv); return out_col; } - int8_t out_type = RAY_I64; + ray_t* target_col = ray_table_get_col(tbl, kid); + if (ngroups == 0 && target_col) out_type = target_col->type; + ray_t* out_col = ray_vec_new(out_type, nrows2); /* resized to expression type on first group */ + if (RAY_IS_ERR(out_col)) { UPDATE_BY_CLEANUP_COLS(); ray_release(groups); ray_release(tbl); DICT_VIEW_CLOSE(updv); return out_col; } + if (ngroups == 0) { + out_col->len = nrows2; + /* Match the per-group path's zero-fill so an unwritten buffer + * never carries allocator garbage (unreachable with rows today + * — a non-empty table always yields a group — but uniform). */ + memset(ray_data(out_col), 0, + (size_t)nrows2 * (size_t)ray_sym_elem_size(out_col->type, out_col->attrs)); + } int first_group = 1; for (int64_t gi = 0; gi < ngroups; gi++) { @@ -11585,7 +11633,7 @@ ray_t* ray_update(ray_t** args, int64_t n) { /* Build a sub-table for this group */ ray_t* sub_tbl = ray_table_new((int32_t)ncols); - if (RAY_IS_ERR(sub_tbl)) { ray_release(out_col); ray_release(result); ray_release(groups); ray_release(tbl); DICT_VIEW_CLOSE(updv); return sub_tbl; } + if (RAY_IS_ERR(sub_tbl)) { ray_release(out_col); UPDATE_BY_CLEANUP_COLS(); ray_release(groups); ray_release(tbl); DICT_VIEW_CLOSE(updv); return sub_tbl; } for (int64_t c = 0; c < ncols; c++) { int64_t cn = ray_table_col_name(tbl, c); ray_t* full_col = ray_table_get_col_idx(tbl, c); @@ -11599,7 +11647,7 @@ ray_t* ray_update(ray_t** args, int64_t n) { ray_t* sub_col = (ct == RAY_SYM) ? ray_sym_vec_new(full_col->attrs & RAY_SYM_W_MASK, gsize) : ray_vec_new(ct, gsize); - if (RAY_IS_ERR(sub_col)) { ray_release(sub_tbl); ray_release(out_col); ray_release(result); ray_release(groups); ray_release(tbl); DICT_VIEW_CLOSE(updv); return sub_col; } + if (RAY_IS_ERR(sub_col)) { ray_release(sub_tbl); ray_release(out_col); UPDATE_BY_CLEANUP_COLS(); ray_release(groups); ray_release(tbl); DICT_VIEW_CLOSE(updv); return sub_col; } /* per-group gather raw-copies cell ids from ONE * source column — keep its dictionary */ if (ct == RAY_SYM) @@ -11613,19 +11661,19 @@ ray_t* ray_update(ray_t** args, int64_t n) { memcpy(dst + r * esz, src + idxs[r] * esz, esz); sub_tbl = ray_table_add_col(sub_tbl, cn, sub_col); ray_release(sub_col); - if (RAY_IS_ERR(sub_tbl)) { ray_release(out_col); ray_release(result); ray_release(groups); ray_release(tbl); DICT_VIEW_CLOSE(updv); return sub_tbl; } + if (RAY_IS_ERR(sub_tbl)) { ray_release(out_col); UPDATE_BY_CLEANUP_COLS(); ray_release(groups); ray_release(tbl); DICT_VIEW_CLOSE(updv); return sub_tbl; } } /* Evaluate expression on sub-table via DAG */ ray_graph_t* ug = ray_graph_new(sub_tbl); ray_op_t* expr_op = compile_expr_dag(ug, agg_expr); - if (!expr_op) { ray_graph_free(ug); ray_release(sub_tbl); ray_release(out_col); ray_release(result); ray_release(groups); ray_release(tbl); DICT_VIEW_CLOSE(updv); return ray_error("domain", "update by: failed to compile aggregate expression"); } + if (!expr_op) { ray_graph_free(ug); ray_release(sub_tbl); ray_release(out_col); UPDATE_BY_CLEANUP_COLS(); ray_release(groups); ray_release(tbl); DICT_VIEW_CLOSE(updv); return ray_error("domain", "update by: failed to compile aggregate expression"); } expr_op = ray_optimize(ug, expr_op); ray_t* agg_result = ray_execute(ug, expr_op); ray_graph_free(ug); ray_release(sub_tbl); - if (RAY_IS_ERR(agg_result)) { ray_release(out_col); ray_release(result); ray_release(groups); ray_release(tbl); DICT_VIEW_CLOSE(updv); return agg_result; } + if (RAY_IS_ERR(agg_result)) { ray_release(out_col); UPDATE_BY_CLEANUP_COLS(); ray_release(groups); ray_release(tbl); DICT_VIEW_CLOSE(updv); return agg_result; } /* Determine output type from first group */ if (first_group) { @@ -11633,25 +11681,77 @@ ray_t* ray_update(ray_t** args, int64_t n) { else if (ray_is_vec(agg_result)) out_type = agg_result->type; ray_release(out_col); out_col = ray_vec_new(out_type, nrows2); - if (RAY_IS_ERR(out_col)) { ray_release(agg_result); ray_release(result); ray_release(groups); ray_release(tbl); DICT_VIEW_CLOSE(updv); return out_col; } + if (RAY_IS_ERR(out_col)) { ray_release(agg_result); UPDATE_BY_CLEANUP_COLS(); ray_release(groups); ray_release(tbl); DICT_VIEW_CLOSE(updv); return out_col; } out_col->len = nrows2; + memset(ray_data(out_col), 0, + (size_t)nrows2 * (size_t)ray_sym_elem_size(out_col->type, out_col->attrs)); first_group = 0; } - /* Broadcast aggregate value to all rows in this group */ + /* Scatter the group's result back to its rows. An atom + * broadcasts to every row of the group; a per-row vector + * (valid kdb, e.g. `update v: 2*v by k`) scatters elementwise + * through idxs[r], symmetric with the atom branch. Any other + * shape — a vector whose length is neither 1 nor the group + * size — has no row-aligned meaning, so decline loudly rather + * than leave the memset zeros in place (silent data loss). */ int64_t* idxs = (int64_t*)ray_data(idx_vec); if (ray_is_atom(agg_result)) { for (int64_t r = 0; r < gsize; r++) store_typed_elem(out_col, idxs[r], agg_result); + } else if (ray_is_vec(agg_result) && ray_len(agg_result) == gsize) { + for (int64_t r = 0; r < gsize; r++) { + int alloc = 0; + ray_t* cell = collection_elem(agg_result, r, &alloc); + store_typed_elem(out_col, idxs[r], cell); + if (alloc) ray_release(cell); + } + } else { + int64_t got = ray_is_vec(agg_result) ? ray_len(agg_result) : -1; + ray_release(agg_result); ray_release(out_col); + UPDATE_BY_CLEANUP_COLS(); ray_release(groups); ray_release(tbl); DICT_VIEW_CLOSE(updv); + return ray_error("length", "update by: expression result length %lld does not match group size %lld", (long long)got, (long long)gsize); } ray_release(agg_result); } - /* Add the new column to the result table */ - result = ray_table_add_col(result, kid, out_col); - ray_release(out_col); - if (RAY_IS_ERR(result)) { ray_release(groups); ray_release(tbl); DICT_VIEW_CLOSE(updv); return result; } + upd_names[upd_i] = kid; + upd_cols[upd_i] = out_col; + upd_i++; + } + + /* Build result in schema order: replace existing targets in place, then + * append genuinely new update columns in dict order. */ + ray_t* result = ray_table_new((int32_t)(ncols + n_updates)); + if (RAY_IS_ERR(result)) { UPDATE_BY_CLEANUP_COLS(); 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); + int64_t ui = -1; + for (int64_t u = 0; u < n_updates; u++) { + if (upd_names[u] == cn) { ui = u; break; } + } + if (ui >= 0) { + result = ray_table_add_col(result, cn, upd_cols[ui]); + ray_release(upd_cols[ui]); + upd_cols[ui] = NULL; + upd_used[ui] = 1; + } else { + ray_t* col = ray_table_get_col_idx(tbl, c); + ray_retain(col); + result = ray_table_add_col(result, cn, col); + ray_release(col); + } + if (RAY_IS_ERR(result)) { UPDATE_BY_CLEANUP_COLS(); ray_release(groups); ray_release(tbl); DICT_VIEW_CLOSE(updv); return result; } + } + for (int64_t u = 0; u < n_updates; u++) { + if (upd_used[u]) continue; + result = ray_table_add_col(result, upd_names[u], upd_cols[u]); + ray_release(upd_cols[u]); + upd_cols[u] = NULL; + if (RAY_IS_ERR(result)) { UPDATE_BY_CLEANUP_COLS(); ray_release(groups); ray_release(tbl); DICT_VIEW_CLOSE(updv); return result; } } + UPDATE_BY_CLEANUP_COLS(); + #undef UPDATE_BY_CLEANUP_COLS ray_release(groups); /* Store in-place and return the symbol if amending by name. */ diff --git a/test/rfl/query/query_update_coverage.rfl b/test/rfl/query/query_update_coverage.rfl index 3d28ee8a..4f2cb493 100644 --- a/test/rfl/query/query_update_coverage.rfl +++ b/test/rfl/query/query_update_coverage.rfl @@ -386,23 +386,24 @@ (at (window-join [Sym Time] wji_null wjt_null wjq_i64null {l: (last Price)}) 'l) -- [400] ;; ──────────────────────────────────────────────────────────────────── -;; update by: with vector-returning expression (line 8376): -;; When agg_result from exec (sub-table expression) is a vector (not atom), -;; ray_is_vec(agg_result) = true at line 8376. -;; The sub-table expression (* v 2) on each group returns a vector, -;; so line 8376 fires for the first group's result. -;; NOTE: the by-group vector result path has no broadcast logic (only -;; atoms are broadcast at 8387-8388), so new_v fills with zeros. -;; This is observable behavior, not an error. +;; update by: with a per-row VECTOR expression (valid kdb: `update v:2*v by k`). +;; The sub-table expression (* v 2) returns a vector of the group's size, so +;; each group's result is scattered elementwise back to its rows through +;; idxs[r] — symmetric with the atom broadcast. Regression: this path used to +;; write nothing, leaving the memset'd column all-zero (silent data loss). ;; ──────────────────────────────────────────────────────────────────── (set Tupd_by_vec (table [k v] (list (list "a" "b" "a" "b") [10 20 30 40]))) -;; update by: k where expression (* v 2) returns a vector per group -;; line 8376 fires: ray_is_vec(agg_result) = true for first group -;; (vector result has no broadcast → new_v column filled with 0) +;; NEW column: group "a"=rows 0,2 -> [20 60]; group "b"=rows 1,3 -> [40 80] (count (update {new_v: (* v 2) by: k from: Tupd_by_vec})) -- 4 -;; Vector group result is NOT broadcast → new_v column stays all-zero. -;; Assert the actual modified cells, not just the row count. -(at (update {new_v: (* v 2) by: k from: Tupd_by_vec}) 'new_v) -- [40 80 0 0] +(at (update {new_v: (* v 2) by: k from: Tupd_by_vec}) 'new_v) -- [20 40 60 80] +;; REPLACING an existing column in place must scatter too, not zero it out +(at (update {v: (* v 2) by: k from: Tupd_by_vec}) 'v) -- [20 40 60 80] +;; the reviewer's exact repro: integer keys, an untouched trailing column +(set Tupd_by_vec3 (table [k v w] (list [1 2 1 2] [10 20 30 40] [7 7 7 7]))) +(at (update {v: (* v 2) from: Tupd_by_vec3 by: k}) 'v) -- [20 40 60 80] +(at (update {v: (* v 2) from: Tupd_by_vec3 by: k}) 'w) -- [7 7 7 7] +;; a vector result whose length != the group size has no row mapping -> length error +(update {new_v: [1 2 3] by: k from: Tupd_by_vec}) !- length ;; ──────────────────────────────────────────────────────────────────── ;; update WHERE with LIST-type expression → type error (lines 8682-8684): @@ -463,25 +464,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..3510cc22 --- /dev/null +++ b/test/rfl/query/update_parted.rfl @@ -0,0 +1,99 @@ +;; 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: substitute existing target +;; columns in their original slots and append only new columns. +;; +;; 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] +;; replaced columns keep their original schema position, even when not last +(key (update {v: (sum v) from: flat by: k})) -- [k v w] +(at (update {v: (sum v) from: flat by: k}) 'v) -- [100 80 30 100 100 80] +;; empty grouped updates preserve an existing target column's type +(set E (table [k v w] (list (as 'I64 []) (as 'F64 []) (as 'F64 [])))) +(key (update {w: (sum v) from: E by: k})) -- [k v w] +(type (at (update {w: (sum v) from: E by: k}) 'w)) -- 'F64 + +;; ────────────── 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")