From aff43d99cc36af1828491bca14437c99d10c73d4 Mon Sep 17 00:00:00 2001 From: Vincent Bernardoff Date: Thu, 22 Oct 2026 13:22:09 +0200 Subject: [PATCH 1/2] perf(join): retain left columns for identity output mappings Verify the completed left index mapping before sharing unchanged columns in flat LEFT joins. Retain their existing nulls, string pools and symbol domains without gathering or rewriting metadata on shared objects. Keep duplicate expansion, reordered output and FULL joins on the gather path, and preserve copy-on-write isolation for subsequent updates. Cover ownership, source lifetime, nulls, strings and nested values, plus large joins. Add a standalone probe and matched backtester measurements. Co-Authored-By: GPT-6 --- bench/join_identity/README.md | 62 ++++++++++++++++++++++++++ bench/join_identity/probe.rfl | 15 +++++++ docs/docs/queries/joins.md | 5 +++ src/ops/join.c | 31 ++++++++++--- test/rfl/join/join_identity.rfl | 20 +++++++++ test/test_join_buildside.c | 78 +++++++++++++++++++++++++++++++++ 6 files changed, 205 insertions(+), 6 deletions(-) create mode 100644 bench/join_identity/README.md create mode 100644 bench/join_identity/probe.rfl create mode 100644 test/rfl/join/join_identity.rfl diff --git a/bench/join_identity/README.md b/bench/join_identity/README.md new file mode 100644 index 000000000..3e436eb53 --- /dev/null +++ b/bench/join_identity/README.md @@ -0,0 +1,62 @@ +# Identity left-join gather + +The flat LEFT join executor retains left columns when the completed left +index mapping is exactly `[0, ..., left_rows - 1]`. It avoids allocating +and gathering unchanged columns and preserves their null bitmaps, string +pools and symbol domains. Updates retain the existing copy-on-write +semantics. Duplicate expansion and reordered output use the gather path; +FULL joins also keep that path because right-only rows change the left side. + +## Standalone probe + +Build separate release binaries from the base and candidate, then run each +from this checkout: + +```sh +RAYFORCE_CORES=4 /path/to/base bench/join_identity/probe.rfl +RAYFORCE_CORES=4 /path/to/candidate bench/join_identity/probe.rfl +``` + +The probe builds one million rows with eight left columns and three unique +right keys. It measures one join, excluding input construction, and checks +the left row order, the last payload column, right payload sum and unmatched +null count. Each process prints `BENCH|milliseconds|valid|rows`. Compare fresh +processes in randomized order after discarded warm-ups, with identical build +flags and worker settings. Whole-process timing includes construction and +validation; use the printed join timing when comparing the executor itself. +The probe checks values; C ownership tests verify that sharing actually occurs. + +## Backtester measurement + +Matched local release builds based on `d7153e72`, Clang 23.1.3, default +`make release` flags; the candidate changes only `join.c`. Both variants +use the same projected backtester report. The workload has 100 instruments, +26,304 hourly steps, 2,630,400 bars and 705 fills, using xsmomf. The default +worker configuration is unchanged between variants. + +One discarded warm-up per variant, then three fresh processes each in +shuffled order (seed 22), without concurrent local compilation or tests: + +| Median | Base | Candidate | +|---|---:|---:| +| Post-run checks | 1139.00 ms | 1033.27 ms | +| Whole process | 4.007 s | 3.917 s | +| Peak RSS | 1607.8 MiB | 1588.6 MiB | + +Check ranges: base 1120.70–1146.35 ms, candidate 1013.85–1056.53 ms. +Process ranges: base 3.997–4.017 s, candidate 3.898–3.921 s. +This small experiment on a shared host is evidence for this workload, +not a general speedup guarantee. The backtester and its generated data are +external to this repository; the standalone probe above needs neither. + +## Regression coverage + +`test/test_join_buildside.c` checks pointer sharing, copy-on-write, null +metadata, string-pool lifetime after releasing the source, duplicate +expansion and a large order-preserving join. The large fixture keeps its +build side below radix dispatch: radix partitioning may reorder rows, which +must fail the explicit identity guard even when cardinality is unchanged. + +`test/rfl/join/join_identity.rfl` checks nulls, symbols, nested lists, +source/result update isolation, missing matches, duplicate expansion and +FULL joins. diff --git a/bench/join_identity/probe.rfl b/bench/join_identity/probe.rfl new file mode 100644 index 000000000..acc031d37 --- /dev/null +++ b/bench/join_identity/probe.rfl @@ -0,0 +1,15 @@ +;; One measured join per process; input construction is outside timeit. +(set n 1000000) +(set ids (til n)) +(set left (table [k a b c d e f g] + (list ids (+ ids 1) (+ ids 2) (+ ids 3) + (+ ids 4) (+ ids 5) (+ ids 6) (+ ids 7)))) +(set right (table [k mark] (list [0 500000 999999] [10 20 30]))) +(set elapsed_ms (timeit (set joined (left-join [k] left right)))) +(set valid (and (== (count joined) n) + (all (== (at joined 'k) ids)) + (all (== (at joined 'g) (+ ids 7))) + (== (sum (at joined 'mark)) 60) + (== (sum (map nil? (at joined 'mark))) (- n 3)))) +(if (not valid) (raise "identity-join: incorrect result")) +(println (format "BENCH|%|%|%" elapsed_ms (as 'I64 valid) (count joined))) diff --git a/docs/docs/queries/joins.md b/docs/docs/queries/joins.md index 2bd42f551..3d721af17 100644 --- a/docs/docs/queries/joins.md +++ b/docs/docs/queries/joins.md @@ -56,6 +56,11 @@ string key on the other side. The `left-join` function keeps all rows from the left table. Unmatched rows have null in the right-side columns. +When each left row produces exactly one output row in its original position, +the flat join executor retains the left columns instead of copying them. +Subsequent updates use copy-on-write, preserving the source and result as +independent values. Duplicate matches still expand and gather the rows. + ### Signature `(left-join [keys] left right)` diff --git a/src/ops/join.c b/src/ops/join.c index 9845a852d..9884ff6c0 100644 --- a/src/ops/join.c +++ b/src/ops/join.c @@ -1761,6 +1761,17 @@ join_gather:; * indices), falling back to per-column gather for nullable RIGHT columns. */ int64_t left_ncols = ray_table_ncols(left_table); int64_t right_ncols = ray_table_ncols(right_table); + /* Share left columns only when the LEFT join's actual mapping preserves + * every left row in order. Cardinality alone is not sufficient: radix + * partitioning can reorder even a one-to-one join's output. + * Retained columns already carry their nulls, pools and symbol domains; + * do not rewrite metadata on these shared objects below. */ + bool left_identity = join_type == 1 && pair_count == left_rows; + if (left_identity) { + for (int64_t i = 0; i < pair_count; i++) { + if (l_idx[i] != i) { left_identity = false; break; } + } + } result = ray_table_new(left_ncols + right_ncols); if (!result || RAY_IS_ERR(result)) goto join_cleanup; @@ -1802,9 +1813,15 @@ join_gather:; result = col; goto join_cleanup; } - ray_t* new_col = col->type == RAY_LIST - ? join_gather_list_col(col, l_idx, pair_count) - : col_vec_new(col, pair_count); + ray_t* new_col; + if (left_identity) { + ray_retain(col); + new_col = col; + } else { + new_col = col->type == RAY_LIST + ? join_gather_list_col(col, l_idx, pair_count) + : col_vec_new(col, pair_count); + } if (!new_col || RAY_IS_ERR(new_col)) { ray_t* err = new_col ? new_col : ray_error("oom", NULL); for (int64_t i = 0; i < l_out_count; i++) ray_release(l_out_cols[i]); @@ -1813,7 +1830,7 @@ join_gather:; goto join_cleanup; } if (col->type == RAY_LIST) l_has_list = true; - else new_col->len = pair_count; + else if (!left_identity) new_col->len = pair_count; l_out_cols[l_out_count] = new_col; l_src_cols[l_out_count] = col; l_out_names[l_out_count] = ray_table_col_name(left_table, c); @@ -1863,7 +1880,9 @@ join_gather:; if (pair_count > 0) { /* Left columns: multi_gather (non-nullable for INNER/LEFT) */ bool l_nullable = (join_type == 2); /* only FULL OUTER */ - if (!l_nullable && !l_has_list && + if (left_identity) { + /* Source columns are retained, so no gather is necessary. */ + } else if (!l_nullable && !l_has_list && l_out_count > 1 && l_out_count <= MGATHER_MAX_COLS) { multi_gather_ctx_t mgctx = { .idx = l_idx, .ncols = l_out_count }; for (int64_t si = 0; si < l_out_count; si++) { @@ -1928,7 +1947,7 @@ join_gather:; * adopt that source's domain (sym-domain Phase 2; no-op while all * domains are the runtime singleton). */ { - for (int64_t si = 0; si < l_out_count; si++) { + for (int64_t si = 0; !left_identity && si < l_out_count; si++) { ray_t* col = l_src_cols[si]; col_propagate_str_pool(l_out_cols[si], col); if (col->type != RAY_LIST) diff --git a/test/rfl/join/join_identity.rfl b/test/rfl/join/join_identity.rfl new file mode 100644 index 000000000..64b09a8df --- /dev/null +++ b/test/rfl/join/join_identity.rfl @@ -0,0 +1,20 @@ +;; One-to-one LEFT joins preserve left values and mutation isolation. +(set identity_left (table [k v s nested] (list [1 2 3] [10 0Nl 30] [z a b] (list [1 2] [3] [4 5])))) +(set identity_right (table [k r] (list [1 3] [100 300]))) +(set identity_join (left-join [k] identity_left identity_right)) +(at identity_join 'v) -- [10 0Nl 30] +(at identity_join 's) -- [z a b] +(at identity_join 'nested) -- (list [1 2] [3] [4 5]) +(set identity_changed (update {v: 99 where: (== k 1) from: identity_join})) +(at identity_left 'v) -- [10 0Nl 30] +(at identity_join 'v) -- [10 0Nl 30] +(at identity_changed 'v) -- [99 0Nl 30] +(set identity_left (update {v: 77 where: (== k 3) from: identity_left})) +(at identity_join 'v) -- [10 0Nl 30] +(at identity_left 'v) -- [10 0Nl 77] +;; All unmatched rows still preserve the left side. +(count (left-join [k] identity_left (table [k r] (list [8] [80])))) -- 3 +;; Duplicate matches must expand and gather the left side. +(at (left-join [k] identity_left (table [k r] (list [1 1 3] [4 5 6]))) 'k) -- [1 1 2 3] +;; FULL joins must allocate, since right-only rows change the left side. +(count (full-join [k] identity_left (table [k r] (list [1 8] [4 5])))) -- 4 diff --git a/test/test_join_buildside.c b/test/test_join_buildside.c index a39b94155..f28d2eeb5 100644 --- a/test/test_join_buildside.c +++ b/test/test_join_buildside.c @@ -1251,9 +1251,87 @@ static test_result_t test_jb_nf_differential(void) { return result; } +/* Identity LEFT gathers must retain ownership and fall back on expansion. */ +static test_result_t test_jb_left_identity(void) { + ray_heap_init(); + (void)ray_sym_init(); + const int64_t keys[] = {1, 2, 3}; + const int64_t rhs[] = {1, 3}; + ray_t* lt = jb_table1("lk", keys, 3); + ray_t* rt = jb_table1("rk", rhs, 2); + ray_t* payload = ray_vec_from_raw(RAY_I64, keys, 3); + ray_vec_set_null(payload, 1, true); + lt = ray_table_add_col(lt, ray_sym_intern("payload", 7), payload); + ray_release(payload); + ray_t* strings = ray_vec_new(RAY_STR, 3); + strings = ray_str_vec_append(strings, "alpha", 5); + strings = ray_str_vec_append(strings, "beta", 4); + strings = ray_str_vec_append(strings, "gamma", 5); + lt = ray_table_add_col(lt, ray_sym_intern("strings", 7), strings); + ray_release(strings); + ray_t* got = jb_join(lt, "lk", rt, "rk", 1); + TEST_ASSERT(got && !RAY_IS_ERR(got), "identity left join"); + ray_t* src = ray_table_col_idx(lt, 1); + ray_t* out = ray_table_col_idx(got, 1); + TEST_ASSERT(out == src, "identity left output shares its source"); + TEST_ASSERT(ray_vec_is_null(out, 1), "shared null metadata"); + ray_retain(out); + int64_t value = 99; + ray_t* changed = ray_vec_set(out, 0, &value); + TEST_ASSERT(changed && !RAY_IS_ERR(changed), "copy on write"); + TEST_ASSERT(changed != src, "mutation detaches shared output"); + TEST_ASSERT(*(int64_t*)ray_vec_get(src, 0) == 1, "source remains unchanged"); + ray_release(changed); + ray_release(lt); + TEST_ASSERT(*(int64_t*)ray_vec_get(out, 2) == 3, "output survives source release"); + ray_t* out_strings = ray_table_col_idx(got, 2); + size_t str_len = 0; + const char* str = ray_str_vec_get(out_strings, 2, &str_len); + TEST_ASSERT(str && str_len == 5 && memcmp(str, "gamma", 5) == 0, + "shared string pool survives source release"); + ray_release(got); + ray_release(rt); + + const int64_t duplicates[] = {1, 1, 3}; + lt = jb_table1("lk", keys, 3); + rt = jb_table1("rk", duplicates, 3); + got = jb_join(lt, "lk", rt, "rk", 1); + TEST_ASSERT(got && !RAY_IS_ERR(got), "duplicate left join"); + TEST_ASSERT(ray_table_nrows(got) == 4, "duplicates expand rows"); + TEST_ASSERT(ray_table_col_idx(got, 0) != ray_table_col_idx(lt, 0), + "expanded output must gather"); + ray_release(got); + ray_release(lt); + ray_release(rt); + + const int64_t n = RAY_PARALLEL_THRESHOLD + 5000; + int64_t* large_keys = malloc((size_t)n * sizeof(*large_keys)); + TEST_ASSERT(large_keys, "large identity fixture allocation"); + for (int64_t i = 0; i < n; i++) large_keys[i] = i; + lt = jb_table1("lk", large_keys, n); + /* Keep the build side below radix dispatch: radix partitioning may + * reorder pairs and must fail the explicit identity check. */ + rt = jb_table1("rk", large_keys, 3); + free(large_keys); + got = jb_join(lt, "lk", rt, "rk", 1); + TEST_ASSERT(got && !RAY_IS_ERR(got), "large identity left join"); + TEST_ASSERT(ray_table_nrows(got) == n, "large identity row count"); + TEST_ASSERT(ray_table_col_idx(got, 0) == ray_table_col_idx(lt, 0), + "large identity shares its source"); + TEST_ASSERT(*(int64_t*)ray_vec_get(ray_table_col_idx(got, 0), n - 1) == n - 1, + "large identity preserves final row"); + ray_release(got); + ray_release(lt); + ray_release(rt); + ray_sym_destroy(); + ray_heap_destroy(); + PASS(); +} + /* ── Entry table ─────────────────────────────────────────────────────────── */ const test_entry_t join_buildside_entries[] = { + { "join_buildside/left_identity", test_jb_left_identity, NULL, NULL }, { "join_buildside/force_fallback_ordinary", test_jb_force_fallback_ordinary, NULL, NULL }, { "join_buildside/auto_fallback_dup", test_jb_auto_fallback_dup, NULL, NULL }, { "join_buildside/baseline_radix_inner", test_jb_baseline_radix_inner, NULL, NULL }, From 6217b31d076d8e7beaf8635286f6033135e5de1f Mon Sep 17 00:00:00 2001 From: Vincent Bernardoff Date: Thu, 22 Oct 2026 13:47:26 +0200 Subject: [PATCH 2/2] fix(join): gather slice columns in identity left joins Keep slice-valued left columns on the independent gather path because copy-on-write currently copies only their headers and can mutate the shared parent payload. Retain eligible ordinary columns in the same table, and propagate metadata only for newly gathered output columns. Reproduce the owned C mutation and assert source, parent and original result values. Pin equal-cardinality reordered radix output with matched left/right payloads, and verify removing the mapping guard fails it. Refresh the benchmark measurements after the slice fallback correction. Co-Authored-By: GPT-6 --- bench/join_identity/README.md | 27 +++++++--- docs/docs/queries/joins.md | 4 +- src/ops/join.c | 18 ++++--- test/test_join_buildside.c | 99 +++++++++++++++++++++++++++++++++++ 4 files changed, 131 insertions(+), 17 deletions(-) diff --git a/bench/join_identity/README.md b/bench/join_identity/README.md index 3e436eb53..e2c38cb44 100644 --- a/bench/join_identity/README.md +++ b/bench/join_identity/README.md @@ -6,6 +6,9 @@ and gathering unchanged columns and preserves their null bitmaps, string pools and symbol domains. Updates retain the existing copy-on-write semantics. Duplicate expansion and reordered output use the gather path; FULL joins also keep that path because right-only rows change the left side. +Slice-valued left columns are gathered independently even with an identity +mapping: slice-header COW does not detach their parent. Other columns in the +same table can still be retained. ## Standalone probe @@ -39,12 +42,13 @@ shuffled order (seed 22), without concurrent local compilation or tests: | Median | Base | Candidate | |---|---:|---:| -| Post-run checks | 1139.00 ms | 1033.27 ms | -| Whole process | 4.007 s | 3.917 s | -| Peak RSS | 1607.8 MiB | 1588.6 MiB | +| Post-run checks | 1146.46 ms | 1047.97 ms | +| Whole process | 4.137 s | 4.022 s | +| Peak RSS | 1608.2 MiB | 1590.1 MiB | -Check ranges: base 1120.70–1146.35 ms, candidate 1013.85–1056.53 ms. -Process ranges: base 3.997–4.017 s, candidate 3.898–3.921 s. +These measurements include the slice fallback review correction. +Check ranges: base 1134.08–1163.55 ms, candidate 1043.04–1051.07 ms. +Process ranges: base 4.074–4.326 s, candidate 4.011–4.052 s. This small experiment on a shared host is evidence for this workload, not a general speedup guarantee. The backtester and its generated data are external to this repository; the standalone probe above needs neither. @@ -53,9 +57,16 @@ external to this repository; the standalone probe above needs neither. `test/test_join_buildside.c` checks pointer sharing, copy-on-write, null metadata, string-pool lifetime after releasing the source, duplicate -expansion and a large order-preserving join. The large fixture keeps its -build side below radix dispatch: radix partitioning may reorder rows, which -must fail the explicit identity guard even when cardinality is unchanged. +expansion and a large order-preserving join. The large sharing fixture keeps +its build side below radix dispatch. +A separate equal-cardinality fixture takes the radix route and verifies +that output really is reordered, every left/right payload remains matched, +and reordered columns are gathered. The slice fixture mutates an owned +output and verifies the source slice, its parent and the original join +result retain their values, while an ordinary column still shares. +The slice regression fails on the original PR at the source-value assertion. +Deliberately removing the mapping guard fails the radix test at the +left/right payload-association assertion; restoring the guard passes. `test/rfl/join/join_identity.rfl` checks nulls, symbols, nested lists, source/result update isolation, missing matches, duplicate expansion and diff --git a/docs/docs/queries/joins.md b/docs/docs/queries/joins.md index 3d721af17..fc7d9776d 100644 --- a/docs/docs/queries/joins.md +++ b/docs/docs/queries/joins.md @@ -57,7 +57,9 @@ string key on the other side. The `left-join` function keeps all rows from the left table. Unmatched rows have null in the right-side columns. When each left row produces exactly one output row in its original position, -the flat join executor retains the left columns instead of copying them. +the flat join executor retains ordinary left columns instead of copying them. +Slice-valued columns still use an independent gather because copying a slice +header does not detach its parent payload. Subsequent updates use copy-on-write, preserving the source and result as independent values. Duplicate matches still expand and gather the rows. diff --git a/src/ops/join.c b/src/ops/join.c index 9884ff6c0..0b902707e 100644 --- a/src/ops/join.c +++ b/src/ops/join.c @@ -1814,7 +1814,9 @@ join_gather:; goto join_cleanup; } ray_t* new_col; - if (left_identity) { + /* Slice COW copies only its header, leaving the parent shared. + * Preserve the old independent gather for this representation. */ + if (left_identity && !(col->attrs & RAY_ATTR_SLICE)) { ray_retain(col); new_col = col; } else { @@ -1830,7 +1832,7 @@ join_gather:; goto join_cleanup; } if (col->type == RAY_LIST) l_has_list = true; - else if (!left_identity) new_col->len = pair_count; + else if (new_col != col) new_col->len = pair_count; l_out_cols[l_out_count] = new_col; l_src_cols[l_out_count] = col; l_out_names[l_out_count] = ray_table_col_name(left_table, c); @@ -1880,9 +1882,7 @@ join_gather:; if (pair_count > 0) { /* Left columns: multi_gather (non-nullable for INNER/LEFT) */ bool l_nullable = (join_type == 2); /* only FULL OUTER */ - if (left_identity) { - /* Source columns are retained, so no gather is necessary. */ - } else if (!l_nullable && !l_has_list && + if (!left_identity && !l_nullable && !l_has_list && l_out_count > 1 && l_out_count <= MGATHER_MAX_COLS) { multi_gather_ctx_t mgctx = { .idx = l_idx, .ncols = l_out_count }; for (int64_t si = 0; si < l_out_count; si++) { @@ -1896,10 +1896,11 @@ join_gather:; else multi_gather_fn(&mgctx, 0, 0, pair_count); } else { - /* Fall back to per-column gather for nullable or single column */ + /* Gather newly allocated columns, including slice fallbacks. + * Retained identity outputs already have their source values. */ for (int64_t si = 0; si < l_out_count; si++) { ray_t* col = l_src_cols[si]; - if (col->type != RAY_LIST) { + if (l_out_cols[si] != col && col->type != RAY_LIST) { gather_ctx_t gctx = { .idx = l_idx, .src_col = col, .dst_col = l_out_cols[si], .esz = col_esz(col), .nullable = l_nullable, @@ -1947,8 +1948,9 @@ join_gather:; * adopt that source's domain (sym-domain Phase 2; no-op while all * domains are the runtime singleton). */ { - for (int64_t si = 0; !left_identity && si < l_out_count; si++) { + for (int64_t si = 0; si < l_out_count; si++) { ray_t* col = l_src_cols[si]; + if (l_out_cols[si] == col) continue; /* retained, metadata intact */ col_propagate_str_pool(l_out_cols[si], col); if (col->type != RAY_LIST) col_propagate_nulls_gather(l_out_cols[si], col, l_idx, pair_count); diff --git a/test/test_join_buildside.c b/test/test_join_buildside.c index f28d2eeb5..1ddcbb5fe 100644 --- a/test/test_join_buildside.c +++ b/test/test_join_buildside.c @@ -1328,9 +1328,108 @@ static test_result_t test_jb_left_identity(void) { PASS(); } +/* Slice-header COW does not detach the underlying parent payload. */ +static test_result_t test_jb_left_identity_slice(void) { + ray_heap_init(); + (void)ray_sym_init(); + const int64_t keys[] = {1, 2}; + const int64_t values[] = {10, 20, 30, 40}; + ray_t* lt = jb_table1("lk", keys, 2); + ray_t* rt = jb_table1("rk", keys, 2); + ray_t* parent = ray_vec_from_raw(RAY_I64, values, 4); + ray_t* slice = ray_vec_slice(parent, 1, 2); + TEST_ASSERT(slice && !RAY_IS_ERR(slice), "slice fixture"); + lt = ray_table_add_col(lt, ray_sym_intern("payload", 7), slice); + ray_release(slice); + ray_t* got = jb_join(lt, "lk", rt, "rk", 1); + TEST_ASSERT(got && !RAY_IS_ERR(got), "identity join with slice payload"); + ray_t* src = ray_table_col_idx(lt, 1); + ray_t* out = ray_table_col_idx(got, 1); + ray_retain(out); + int64_t value = 99; + ray_t* changed = ray_vec_set(out, 0, &value); + TEST_ASSERT(changed && !RAY_IS_ERR(changed), "owned output mutation"); + TEST_ASSERT(*(int64_t*)ray_vec_get(changed, 0) == 99, "changed value"); + TEST_ASSERT(*(int64_t*)ray_vec_get(src, 0) == 20, "source slice unchanged"); + TEST_ASSERT(*(int64_t*)ray_vec_get(out, 0) == 20, "original join unchanged"); + for (int64_t i = 0; i < 4; i++) + TEST_ASSERT(*(int64_t*)ray_vec_get(parent, i) == values[i], + "source parent unchanged"); + TEST_ASSERT(!(out->attrs & RAY_ATTR_SLICE), "slice output is gathered"); + TEST_ASSERT(ray_table_col_idx(got, 0) == ray_table_col_idx(lt, 0), + "ordinary column still shares in mixed representation table"); + ray_release(changed); + ray_release(lt); + ray_release(parent); + TEST_ASSERT(*(int64_t*)ray_vec_get(out, 1) == 30, "gathered slice lifetime"); + ray_release(got); + ray_release(rt); + ray_sym_destroy(); + ray_heap_destroy(); + PASS(); +} + +/* Equal cardinality cannot prove identity: radix output is partition ordered. */ +static test_result_t test_jb_left_identity_radix(void) { + ray_heap_init(); + (void)ray_sym_init(); + const int64_t n = RAY_PARALLEL_THRESHOLD + 5000; + int64_t* keys = malloc((size_t)n * sizeof(*keys)); + int64_t* values = malloc((size_t)n * sizeof(*values)); + TEST_ASSERT(keys && values, "radix fixture allocation"); + for (int64_t i = 0; i < n; i++) { keys[i] = i; values[i] = 7 * i + 11; } + ray_t* lt = jb_table1("lk", keys, n); + ray_t* rt = jb_table1("rk", keys, n); + ray_t* right_ids = ray_vec_from_raw(RAY_I64, keys, n); + rt = ray_table_add_col(rt, ray_sym_intern("right_id", 8), right_ids); + ray_release(right_ids); + ray_t* payload = ray_vec_from_raw(RAY_I64, values, n); + lt = ray_table_add_col(lt, ray_sym_intern("payload", 7), payload); + ray_release(payload); + free(keys); + free(values); + ray_t* got = jb_join(lt, "lk", rt, "rk", 1); + TEST_ASSERT(got && !RAY_IS_ERR(got), "radix left join"); + TEST_ASSERT(ray_table_nrows(got) == n, "equal output cardinality"); + ray_t* out_key = ray_table_col_idx(got, 0); + ray_t* out_value = ray_table_col_idx(got, 1); + ray_t* out_right_id = ray_table_col_idx(got, 2); + uint8_t* seen = calloc((size_t)n, 1); + TEST_ASSERT(seen, "radix membership allocation"); + bool reordered = false; + for (int64_t i = 0; i < n; i++) { + int64_t key = *(int64_t*)ray_vec_get(out_key, i); + TEST_ASSERT(key >= 0 && key < n && !seen[key], "unique output key"); + seen[key] = 1; + reordered |= key != i; + TEST_ASSERT(*(int64_t*)ray_vec_get(out_right_id, i) == key, + "left and right payloads remain matched after reorder"); + TEST_ASSERT(*(int64_t*)ray_vec_get(out_value, i) == 7 * key + 11, + "payload follows its original row through radix reorder"); + } + free(seen); + TEST_ASSERT(reordered, "fixture exercises a nonidentity radix mapping"); + TEST_ASSERT(out_key != ray_table_col_idx(lt, 0), "reordered keys gathered"); + TEST_ASSERT(out_value != ray_table_col_idx(lt, 1), "reordered payload gathered"); + for (int64_t i = 0; i < n; i++) { + TEST_ASSERT(*(int64_t*)ray_vec_get(ray_table_col_idx(lt, 0), i) == i, + "source keys unchanged"); + TEST_ASSERT(*(int64_t*)ray_vec_get(ray_table_col_idx(lt, 1), i) == 7 * i + 11, + "source payload unchanged"); + } + ray_release(got); + ray_release(lt); + ray_release(rt); + ray_sym_destroy(); + ray_heap_destroy(); + PASS(); +} + /* ── Entry table ─────────────────────────────────────────────────────────── */ const test_entry_t join_buildside_entries[] = { + { "join_buildside/left_identity_slice", test_jb_left_identity_slice, NULL, NULL }, + { "join_buildside/left_identity_radix", test_jb_left_identity_radix, NULL, NULL }, { "join_buildside/left_identity", test_jb_left_identity, NULL, NULL }, { "join_buildside/force_fallback_ordinary", test_jb_force_fallback_ordinary, NULL, NULL }, { "join_buildside/auto_fallback_dup", test_jb_auto_fallback_dup, NULL, NULL },