diff --git a/bench/join_identity/README.md b/bench/join_identity/README.md new file mode 100644 index 000000000..e2c38cb44 --- /dev/null +++ b/bench/join_identity/README.md @@ -0,0 +1,73 @@ +# 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. +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 + +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 | 1146.46 ms | 1047.97 ms | +| Whole process | 4.137 s | 4.022 s | +| Peak RSS | 1608.2 MiB | 1590.1 MiB | + +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. + +## 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 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 +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..fc7d9776d 100644 --- a/docs/docs/queries/joins.md +++ b/docs/docs/queries/joins.md @@ -56,6 +56,13 @@ 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 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. + ### Signature `(left-join [keys] left right)` diff --git a/src/ops/join.c b/src/ops/join.c index 9845a852d..0b902707e 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,17 @@ 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; + /* 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 { + 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 +1832,7 @@ join_gather:; goto join_cleanup; } if (col->type == RAY_LIST) l_has_list = true; - else 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); @@ -1863,7 +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 (!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++) { @@ -1877,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, @@ -1930,6 +1950,7 @@ join_gather:; { 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/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..1ddcbb5fe 100644 --- a/test/test_join_buildside.c +++ b/test/test_join_buildside.c @@ -1251,9 +1251,186 @@ 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(); +} + +/* 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 }, { "join_buildside/baseline_radix_inner", test_jb_baseline_radix_inner, NULL, NULL },