Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
73 changes: 73 additions & 0 deletions bench/join_identity/README.md
Original file line number Diff line number Diff line change
@@ -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.
15 changes: 15 additions & 0 deletions bench/join_identity/probe.rfl
Original file line number Diff line number Diff line change
@@ -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)))
7 changes: 7 additions & 0 deletions docs/docs/queries/joins.md
Original file line number Diff line number Diff line change
Expand Up @@ -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)`
Expand Down
35 changes: 28 additions & 7 deletions src/ops/join.c
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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]);
Expand All @@ -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);
Expand Down Expand Up @@ -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++) {
Expand All @@ -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,
Expand Down Expand Up @@ -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);
Expand Down
20 changes: 20 additions & 0 deletions test/rfl/join/join_identity.rfl
Original file line number Diff line number Diff line change
@@ -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
177 changes: 177 additions & 0 deletions test/test_join_buildside.c
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Comment thread
morsel-ray marked this conversation as resolved.
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 },
Expand Down
Loading