From 62e44476b0f585352b7d71c96a04a9485c90b5c3 Mon Sep 17 00:00:00 2001 From: Anton Date: Sun, 20 Sep 2026 11:37:36 +0200 Subject: [PATCH] perf(join): skip per-cell key null tests on provably null-free columns MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ray_vec_is_null is out-of-line (no LTO) and the join called it once per key column per row in hash_row_keys — on the build side, the probe side, and the prefetch lookahead — plus twice per key column per hash-chain step in join_keys_eq, across both the count and the fill pass. A reported profile put 13.28% of a service's samples there, on a join keyed by two SYM columns that structurally never hold a null. Prove once per join that no key column can hold a null and drop the call. SYM/STR nulls are canonical empty payloads (id 0 / length 0) that HAS_NULLS does not track, so text columns are proven by the chunked zero-scan from #533 rather than by the flag; everything else reads the flag through slices and takes a set bit at face value, so the proof stays O(n) and never degrades into ray_vec_has_nulls' per-element walk. Flag-readable columns are settled first, so a nullable numeric key short-circuits before any text column is scanned. OP_CONST (atom) key slots are refused as unprovable. Also skip the #458 null-run pre-scan under the proof. It gates on ray_vec_may_have_nulls, which is unconditionally true for SYM/STR, so a SYM-keyed join ran a full per-key-per-build-row ray_vec_is_null scan before the join proper on every execution; a null-free key set cannot contain an all-null row, so the scan is dead. Measured with bench/join_nullfree (release, 1:1 book join, 4M probe x 500K build): two SYM keys 275 -> 249 ms (-9.5% median, -7.8% min); single I64 key 106 -> 99 ms (-6.3% median, -7.9% min). A proof that fails on a text key costs a partial scan (~8ms on a 4M-row SYM column) and gains nothing — only nullable SYM/STR key columns pay it. ray_join_force_null_checks forces the null-aware loops so the differential tests and the perf gate can compare both paths in one binary; ray_join_nullfree_keys counts the joins that took the fast path. Closes #597 --- bench/join_nullfree/main.c | 307 +++++++++++++++++++++++++++++++++++++ src/ops/internal.h | 2 + src/ops/join.c | 129 +++++++++++++--- test/test_join_buildside.c | 184 ++++++++++++++++++++++ 4 files changed, 599 insertions(+), 23 deletions(-) create mode 100644 bench/join_nullfree/main.c diff --git a/bench/join_nullfree/main.c b/bench/join_nullfree/main.c new file mode 100644 index 000000000..bdb3252f6 --- /dev/null +++ b/bench/join_nullfree/main.c @@ -0,0 +1,307 @@ +/* Null-free join key perf gate (#597). + * + * The join tests every key cell for null on every row: once per key column + * in hash_row_keys (build and probe, plus the prefetch lookahead) and twice + * per key column per hash-chain step in join_keys_eq, across both the count + * and the fill pass. ray_vec_is_null is out-of-line (no LTO), so each test + * is a call. A reporter profiled 13.28% of a service's samples there, on a + * join keyed by two SYM columns that structurally never hold a null. + * + * join_keys_nullfree proves once per join that no key column can hold a + * null and the loops drop the call. This measures what that is worth. + * + * Cases (all two-key SYM joins, mirroring the reported venue+instrument + * book shape): + * SYM2 right=500K (unique venue+instrument pairs), left=4M + * drawn from those pairs, both key columns null-free, so + * each left row matches exactly one right row — the shape + * of a book join, not a fan-out. The fast path must fire. + * SYM2-NULL identical, except one left venue cell is the SYM null. + * The fast path must NOT fire; both sides must time alike, + * bounding what the proof scan itself costs. + * I64 right=500K, left=4M, single I64 key, HAS_NULLS clear. + * Fast path fires via the attrs bit rather than a scan. + * + * Mechanism: ray_join_nullfree_keys must advance on SYM2 and I64 and must + * not advance on SYM2-NULL. ray_join_force_null_checks supplies the + * null-aware baseline in the same binary. + * + * Timing: CLOCK_MONOTONIC around ray_execute only. Tables built once + * outside the timed loop; graph rebuilt per rep; sides interleaved per rep + * so drift hits both equally. + */ +#if defined(__APPLE__) +# define _DARWIN_C_SOURCE +#else +# define _POSIX_C_SOURCE 200809L +#endif + +#include +#include "mem/heap.h" +#include "ops/ops.h" +#include "ops/internal.h" +#include "table/sym.h" +#include +#include +#include +#include +#include +#include + +/* ---------- timing ---------- */ +static double now_ms(void) { + struct timespec ts; + clock_gettime(CLOCK_MONOTONIC, &ts); + return (double)ts.tv_sec * 1e3 + (double)ts.tv_nsec * 1e-6; +} + +static int cmp_double(const void* a, const void* b) { + double x = *(const double*)a, y = *(const double*)b; + return (x > y) - (x < y); +} +static double medianN(double arr[], int n) { + double tmp[64]; + memcpy(tmp, arr, (size_t)n * sizeof(double)); + qsort(tmp, (size_t)n, sizeof(double), cmp_double); + return tmp[n / 2]; +} +static double minN(double arr[], int n) { + double m = arr[0]; + for (int i = 1; i < n; i++) if (arr[i] < m) m = arr[i]; + return m; +} + +/* ---------- SYM column over a vocabulary of `vocab` interned symbols ------ + * Cell i takes vocabulary entry (i * stride) % vocab. null_at, when >= 0, + * is written as SYM id 0 — the canonical SYM null, which HAS_NULLS does not + * track, so only a payload scan can see it. */ +static ray_t* make_sym_col(const char* prefix, int64_t n, int64_t vocab, + int64_t stride, int64_t null_at) { + ray_t* col = ray_sym_vec_new(RAY_SYM_W64, n); + if (!col || RAY_IS_ERR(col)) { fprintf(stderr, "make_sym_col: alloc\n"); abort(); } + col->len = n; + + int64_t* ids = (int64_t*)malloc((size_t)vocab * sizeof(int64_t)); + if (!ids) { fprintf(stderr, "make_sym_col: OOM vocab\n"); abort(); } + for (int64_t v = 0; v < vocab; v++) { + char b[32]; + int m = snprintf(b, sizeof(b), "%s%lld", prefix, (long long)v); + ids[v] = ray_sym_intern(b, (size_t)m); + } + for (int64_t i = 0; i < n; i++) + ray_write_sym(ray_data(col), i, (uint64_t)ids[(i * stride) % vocab], + RAY_SYM, col->attrs); + if (null_at >= 0 && null_at < n) + ray_write_sym(ray_data(col), null_at, 0, RAY_SYM, col->attrs); + free(ids); + return col; +} + +static ray_t* make_i64_col(int64_t n, int64_t mod) { + int64_t* v = (int64_t*)malloc((size_t)n * sizeof(int64_t)); + if (!v) { fprintf(stderr, "make_i64_col: OOM\n"); abort(); } + for (int64_t i = 0; i < n; i++) v[i] = i % mod; + ray_t* col = ray_vec_from_raw(RAY_I64, v, n); + free(v); + if (!col || RAY_IS_ERR(col)) { fprintf(stderr, "make_i64_col: from_raw\n"); abort(); } + return col; +} + +static ray_t* table_of(const char* const* names, ray_t* const* cols, int64_t ncols) { + ray_t* tbl = ray_table_new(ncols); + for (int64_t c = 0; c < ncols; c++) + tbl = ray_table_add_col(tbl, ray_sym_intern(names[c], strlen(names[c])), cols[c]); + if (!tbl || RAY_IS_ERR(tbl)) { fprintf(stderr, "table_of: add_col\n"); abort(); } + return tbl; +} + +/* ---------- one inner-join rep ---------- */ +static double run_join_rep(ray_t* lt, const char* const* lkeys, + ray_t* rt, const char* const* rkeys, + uint32_t n_keys, int64_t* rows_out) { + ray_graph_t* g = ray_graph_new(lt); + if (!g) { fprintf(stderr, "run_join_rep: graph alloc\n"); abort(); } + + ray_op_t* lt_node = ray_const_table(g, lt); + ray_op_t* rt_node = ray_const_table(g, rt); + ray_op_t* lk_arr[4]; + ray_op_t* rk_arr[4]; + for (uint32_t k = 0; k < n_keys; k++) { + lk_arr[k] = ray_scan(g, lkeys[k]); + rk_arr[k] = ray_scan(g, rkeys[k]); + if (!lk_arr[k] || !rk_arr[k]) { fprintf(stderr, "run_join_rep: key node\n"); abort(); } + } + if (!lt_node || !rt_node) { fprintf(stderr, "run_join_rep: node alloc\n"); abort(); } + + ray_op_t* jn = ray_join(g, lt_node, lk_arr, rt_node, rk_arr, n_keys, 0); + if (!jn) { fprintf(stderr, "run_join_rep: join node\n"); abort(); } + jn = ray_optimize(g, jn); + + double t0 = now_ms(); + ray_t* result = ray_execute(g, jn); + double t1 = now_ms(); + + if (!result || RAY_IS_ERR(result)) { + fprintf(stderr, "run_join_rep: execute returned error\n"); abort(); + } + if (rows_out) *rows_out = ray_table_nrows(result); + ray_release(result); + ray_graph_free(g); + return t1 - t0; +} + +#define NREPS 11 + +typedef struct { + const char* name; + double fast_ms[NREPS]; /* knob off — null-free proof allowed */ + double base_ms[NREPS]; /* knob on — null-aware loops (pre-#597) */ + int64_t rows_out; +} case_result_t; + +static void run_case(const char* name, + ray_t* lt, const char* const* lkeys, + ray_t* rt, const char* const* rkeys, + uint32_t n_keys, bool expect_fast, + case_result_t* cr) { + cr->name = name; + cr->rows_out = -1; + + printf("Running case %-12s (%d reps)...\n", name, NREPS); + fflush(stdout); + + uint64_t nf_before = ray_join_nullfree_keys; + + for (int rep = 0; rep < NREPS; rep++) { + ray_join_force_null_checks = false; + int64_t rows_f = -1; + cr->fast_ms[rep] = run_join_rep(lt, lkeys, rt, rkeys, n_keys, &rows_f); + + ray_join_force_null_checks = true; + int64_t rows_b = -1; + cr->base_ms[rep] = run_join_rep(lt, lkeys, rt, rkeys, n_keys, &rows_b); + ray_join_force_null_checks = false; + + if (rows_f != rows_b) { + fprintf(stderr, + "CORRECTNESS FAILURE case %s rep %d: fast=%lld rows, baseline=%lld rows\n", + name, rep, (long long)rows_f, (long long)rows_b); + abort(); + } + cr->rows_out = rows_f; + } + + bool fired = ray_join_nullfree_keys > nf_before; + if (expect_fast != fired) { + fprintf(stderr, + "MECHANISM FAILURE case %s: expected null-free path %s " + "(before=%llu after=%llu)\n", + name, expect_fast ? "to fire" : "NOT to fire", + (unsigned long long)nf_before, + (unsigned long long)ray_join_nullfree_keys); + abort(); + } + + printf(" nullfree counter: before=%llu after=%llu fired=%s rows=%lld\n", + (unsigned long long)nf_before, + (unsigned long long)ray_join_nullfree_keys, + fired ? "YES" : "NO", + (long long)cr->rows_out); + fflush(stdout); +} + +static void report(const case_result_t* cr) { + double fmed = medianN((double*)cr->fast_ms, NREPS); + double bmed = medianN((double*)cr->base_ms, NREPS); + double fmin = minN((double*)cr->fast_ms, NREPS); + double bmin = minN((double*)cr->base_ms, NREPS); + printf("%-12s median %8.2f -> %8.2f ms (%+6.1f%%) min %8.2f -> %8.2f ms (%+6.1f%%)\n", + cr->name, bmed, fmed, 100.0 * (fmed - bmed) / bmed, + bmin, fmin, 100.0 * (fmin - bmin) / bmin); +} + +int main(void) { + ray_heap_init(); + (void)ray_sym_init(); + ray_join_force_null_checks = false; + + printf("=== bench-join-nullfree (#597) ===\n"); + printf("NREPS=%d RAY_PARALLEL_THRESHOLD=%d\n\n", + NREPS, (int)RAY_PARALLEL_THRESHOLD); + fflush(stdout); + + const int64_t nl = 4000000L; /* probe side */ + const int64_t nr = 500000L; /* build side */ + + /* ---- SYM2: two null-free SYM key columns ---- */ + printf("Building SYM2 tables (left=%lld, right=%lld)...\n", + (long long)nl, (long long)nr); + fflush(stdout); + { + /* Right: instrument unique per row, venue spread over 64 — 500K + * distinct (venue, instrument) pairs. Left: the same pairs cycled, + * so every left row has exactly one match. */ + ray_t* rv = make_sym_col("venue", nr, nr, 1, -1); + ray_t* ri = make_sym_col("inst", nr, nr, 1, -1); + ray_t* lv = make_sym_col("venue", nl, nr, 1, -1); + ray_t* li = make_sym_col("inst", nl, nr, 1, -1); + + /* SYM2-NULL shares the build side and differs only in one left cell. */ + ray_t* lv_null = make_sym_col("venue", nl, nr, 1, nl / 2); + ray_t* li_null = make_sym_col("inst", nl, nr, 1, -1); + + static const char* const lnames[] = { "lvenue", "linst" }; + static const char* const rnames[] = { "rvenue", "rinst" }; + static const char* const lkeys[] = { "lvenue", "linst" }; + static const char* const rkeys[] = { "rvenue", "rinst" }; + + ray_t* lcols[2] = { lv, li }; + ray_t* rcols[2] = { rv, ri }; + ray_t* lcols_n[2] = { lv_null, li_null }; + ray_t* lt = table_of(lnames, lcols, 2); + ray_t* rt = table_of(rnames, rcols, 2); + ray_t* lt_null = table_of(lnames, lcols_n, 2); + ray_release(lv); ray_release(li); ray_release(rv); ray_release(ri); + ray_release(lv_null); ray_release(li_null); + + case_result_t cr_sym2, cr_sym2n; + run_case("SYM2", lt, lkeys, rt, rkeys, 2, true, &cr_sym2); + run_case("SYM2-NULL", lt_null, lkeys, rt, rkeys, 2, false, &cr_sym2n); + + printf("\n--- results (baseline = forced null-aware loops) ---\n"); + report(&cr_sym2); + report(&cr_sym2n); + + ray_release(lt); ray_release(rt); ray_release(lt_null); + } + + /* ---- I64: single key, proof is the HAS_NULLS bit ---- */ + printf("\nBuilding I64 tables (left=%lld, right=%lld)...\n", + (long long)nl, (long long)nr); + fflush(stdout); + { + ray_t* lc = make_i64_col(nl, nr); + ray_t* rc = make_i64_col(nr, nr); + static const char* const lnames[] = { "lk" }; + static const char* const rnames[] = { "rk" }; + static const char* const lkeys[] = { "lk" }; + static const char* const rkeys[] = { "rk" }; + ray_t* lcols[1] = { lc }; + ray_t* rcols[1] = { rc }; + ray_t* lt = table_of(lnames, lcols, 1); + ray_t* rt = table_of(rnames, rcols, 1); + ray_release(lc); ray_release(rc); + + case_result_t cr_i64; + run_case("I64", lt, lkeys, rt, rkeys, 1, true, &cr_i64); + printf("\n--- results (baseline = forced null-aware loops) ---\n"); + report(&cr_i64); + + ray_release(lt); ray_release(rt); + } + + printf("\n(negative %% = faster with the null-free path)\n"); + ray_sym_destroy(); + ray_heap_destroy(); + return 0; +} diff --git a/src/ops/internal.h b/src/ops/internal.h index 4ff400598..2bb8dd9e9 100644 --- a/src/ops/internal.h +++ b/src/ops/internal.h @@ -804,6 +804,8 @@ extern bool ray_join_force_dup_fallback; extern bool ray_join_no_dup_fallback; extern uint64_t ray_join_dup_fallbacks; extern uint64_t ray_join_null_fallbacks; +extern bool ray_join_force_null_checks; +extern uint64_t ray_join_nullfree_keys; extern bool ray_agg_engine_v2; /* route OP_GROUP through v2 agg engine; default ON (agg_engine.c) */ void ray_expr_stats_init(void); diff --git a/src/ops/join.c b/src/ops/join.c index bc8e744c0..7a53f57f3 100644 --- a/src/ops/join.c +++ b/src/ops/join.c @@ -44,6 +44,67 @@ uint64_t ray_join_dup_fallbacks = 0; /* Diagnostic: radix joins routed to the chained path upfront because the * build side carries more all-null key rows than RADIX_DUP_RUN_MAX (#458). */ uint64_t ray_join_null_fallbacks = 0; +/* Test knob: suppress the null-free key fast path (#597) so the differential + * harness can compare it against the null-aware loops in one binary. */ +bool ray_join_force_null_checks = false; +/* Diagnostic: joins whose key columns were all proven null-free. */ +uint64_t ray_join_nullfree_keys = 0; + +/* ── #597: null-free key proof ─────────────────────────────────────────── + * ray_vec_is_null is an out-of-line call (no LTO), and the join makes one + * per key column per row in hash_row_keys plus two per key column per + * hash-chain step in join_keys_eq — on both the count and the fill pass. + * A profiled service spent 13% of its samples there. Prove ONCE per join + * that no key column can hold a null and the loops drop the call entirely. + * + * SYM/STR nulls are canonical empty payloads (id 0 / length 0) that + * HAS_NULLS does not track, so text columns are proven by the chunked, + * vectorized zero-scan from #533 rather than by the flag — which is why + * the flag alone would not have helped a SYM-keyed join. Everything else + * is the flag, read through slices by ray_vec_may_have_nulls; a set flag + * is taken at face value (a column that merely may hold a null keeps the + * null-aware path) so the proof stays O(n) and never degrades into the + * per-element walk ray_vec_has_nulls would do. + * + * An absent key column is not a null cell: hash_row_keys skips it and + * join_keys_eq rejects the pair before either reaches the null test, so + * it cannot affect the proof. */ +static bool join_key_col_nullfree(const ray_t* v) { + if (!v) return true; + /* A key slot may hold an OP_CONST literal, i.e. an atom, whose null + * state is RAY_ATOM_IS_NULL and not the vector attrs the proof reads. + * Nothing in the current tree builds such a key, so this is a guard + * rather than a fix: refuse to prove what this function cannot see. */ + if (ray_is_atom(v)) return false; + if (v->type == RAY_SYM || v->type == RAY_STR) return !ray_vec_text_has_nulls(v); + return !ray_vec_may_have_nulls(v); +} + +/* Both sides must be clean: join_keys_eq tests the left and the right cell. + * + * Flag-readable columns are settled first, in one O(1) pass, so a nullable + * numeric key short-circuits the whole proof before any text column is + * scanned. A failed proof on a text column still costs a partial scan + * (~8ms on a 4M-row SYM column, measured) with nothing to show for it — + * that is the price of a payload-encoded null, and only nullable SYM/STR + * keys pay it. */ +static bool join_key_col_scanned(const ray_t* v) { + return v && !ray_is_atom(v) && (v->type == RAY_SYM || v->type == RAY_STR); +} + +static bool join_keys_nullfree(ray_t* const* l_vecs, ray_t* const* r_vecs, + uint32_t n_keys) { + if (ray_join_force_null_checks) return false; + for (uint32_t k = 0; k < n_keys; k++) { + if (!join_key_col_scanned(l_vecs[k]) && !join_key_col_nullfree(l_vecs[k])) return false; + if (!join_key_col_scanned(r_vecs[k]) && !join_key_col_nullfree(r_vecs[k])) return false; + } + for (uint32_t k = 0; k < n_keys; k++) { + if (join_key_col_scanned(l_vecs[k]) && !join_key_col_nullfree(l_vecs[k])) return false; + if (join_key_col_scanned(r_vecs[k]) && !join_key_col_nullfree(r_vecs[k])) return false; + } + return true; +} static int join_store_key_cell(ray_t* dst, int64_t dst_row, ray_t* src, int64_t src_row) { @@ -112,13 +173,14 @@ static inline bool join_str_eq_hashed(const ray_str_t* a, const char* pool_a, * real key's hash is resolved by join_keys_eq, as for any other hash. */ #define JOIN_NULL_KEY_HASH INT64_C(0x5B7A6D3F2E1C0A94) -static uint64_t hash_row_keys(ray_t** key_vecs, uint32_t n_keys, int64_t row) { +static uint64_t hash_row_keys(ray_t** key_vecs, uint32_t n_keys, int64_t row, + bool nullfree) { uint64_t h = 0; for (uint32_t k = 0; k < n_keys; k++) { ray_t* col = key_vecs[k]; if (!col) continue; uint64_t kh; - if (ray_vec_is_null(col, row)) { + if (!nullfree && ray_vec_is_null(col, row)) { /* null == null: hash every null cell alike instead of giving the * row a private hash. A per-row hash made a null key unmatchable * even against itself, so `anti-join [c] X X` came back non-empty @@ -224,6 +286,7 @@ typedef struct { uint32_t* hashes; /* output: hash[row] */ const ray_str_t* str_desc; /* one-key STR specialization, else NULL */ const char* str_pool; + bool nullfree; /* #597: no key column can hold a null */ } join_radix_hash_ctx_t; static void join_radix_hash_fn(void* raw, uint32_t wid, int64_t start, int64_t end) { @@ -241,14 +304,14 @@ static void join_radix_hash_fn(void* raw, uint32_t wid, int64_t start, int64_t e return; } for (int64_t r = start; r < end; r++) - c->hashes[r] = (uint32_t)hash_row_keys(c->key_vecs, c->n_keys, r); + c->hashes[r] = (uint32_t)hash_row_keys(c->key_vecs, c->n_keys, r, c->nullfree); } static join_radix_hash_ctx_t join_radix_hash_ctx(ray_t** keys, uint32_t n_keys, - uint32_t* hashes) { + uint32_t* hashes, bool nullfree) { join_radix_hash_ctx_t c = { .key_vecs = keys, .n_keys = n_keys, .hashes = hashes, - .str_desc = NULL, .str_pool = NULL, + .str_desc = NULL, .str_pool = NULL, .nullfree = nullfree, }; if (n_keys == 1 && keys[0] && keys[0]->type == RAY_STR) { ray_t* col = keys[0]; @@ -584,7 +647,7 @@ static ray_t* join_gather_col_serial(ray_t* src, const int64_t* idx, /* Key equality helper — shared by count + fill phases */ static inline bool join_keys_eq(ray_t* const* l_vecs, ray_t* const* r_vecs, uint32_t n_keys, - int64_t l, int64_t r) { + int64_t l, int64_t r, bool nullfree) { for (uint32_t k = 0; k < n_keys; k++) { ray_t* lc = l_vecs[k]; ray_t* rc = r_vecs[k]; @@ -598,8 +661,8 @@ static inline bool join_keys_eq(ray_t* const* l_vecs, ray_t* const* r_vecs, uint * whenever c held a null and made left-join miss a null-keyed match. * As-of join keeps its own documented NULLs-never-match rule; it * does not route through here. */ - bool l_null = ray_vec_is_null(lc, l); - bool r_null = ray_vec_is_null(rc, r); + bool l_null = !nullfree && ray_vec_is_null(lc, l); + bool r_null = !nullfree && ray_vec_is_null(rc, r); if (l_null || r_null) { if (l_null != r_null) return false; /* No looser than the value arms below: they pair STR only with @@ -667,6 +730,7 @@ typedef struct { const char* l_str_pool; const char* r_str_pool; uint8_t join_type; + bool nullfree; /* #597: no key column can hold a null */ /* Per-partition output: pp_l[p], pp_r[p] are local buffers */ int32_t** pp_l; /* per-partition left indices (int32_t) */ int32_t** pp_r; /* per-partition right indices (int32_t) */ @@ -858,7 +922,7 @@ static void join_radix_build_probe_fn(void* raw, uint32_t wid, int64_t task_star ? join_str_eq_hashed(&c->l_str_desc[lr], c->l_str_pool, &c->r_str_desc[rr], c->r_str_pool) : join_keys_eq(c->l_key_vecs, c->r_key_vecs, c->n_keys, - (int64_t)lr, (int64_t)rr); + (int64_t)lr, (int64_t)rr, c->nullfree); if (keys_equal) { if (!bp_grow_bufs(c, p, &pl, &pr, &cap, cnt)) goto done; @@ -908,6 +972,7 @@ typedef struct { /* ASP-Join: semijoin filter from factorized left side (NULL if N/A) */ uint64_t* asp_bits; int64_t asp_key_max; + bool nullfree; /* #597: no key column can hold a null */ } join_build_ctx_t; static void join_build_fn(void* raw, uint32_t wid, int64_t start, int64_t end) { @@ -930,10 +995,10 @@ static void join_build_fn(void* raw, uint32_t wid, int64_t start, int64_t end) { continue; } if (r + 8 < end) { - uint64_t pf_h = hash_row_keys(c->r_key_vecs, c->n_keys, r + 8); + uint64_t pf_h = hash_row_keys(c->r_key_vecs, c->n_keys, r + 8, c->nullfree); __builtin_prefetch(&heads[(uint32_t)(pf_h & mask)], 1, 1); } - uint64_t h = hash_row_keys(c->r_key_vecs, c->n_keys, r); + uint64_t h = hash_row_keys(c->r_key_vecs, c->n_keys, r, c->nullfree); uint32_t slot = (uint32_t)(h & mask); uint32_t row32 = (uint32_t)r; uint32_t old = atomic_load_explicit(&heads[slot], memory_order_relaxed); @@ -966,6 +1031,7 @@ typedef struct { /* S-Join: semijoin filter bitmap (NULL if not applicable) */ uint64_t* sjoin_bits; int64_t sjoin_key_max; + bool nullfree; /* #597: no key column can hold a null */ } join_probe_ctx_t; /* Pass 2a: count matches per morsel */ @@ -993,14 +1059,14 @@ static void join_count_fn(void* raw, uint32_t wid, int64_t task_start, int64_t t } if (l + 8 < row_end) { - uint64_t pf_h = hash_row_keys(c->l_key_vecs, c->n_keys, l + 8); + uint64_t pf_h = hash_row_keys(c->l_key_vecs, c->n_keys, l + 8, c->nullfree); __builtin_prefetch(&c->ht_heads[(uint32_t)(pf_h & ht_mask)], 0, 1); } - uint64_t h = hash_row_keys(c->l_key_vecs, c->n_keys, l); + uint64_t h = hash_row_keys(c->l_key_vecs, c->n_keys, l, c->nullfree); uint32_t slot = (uint32_t)(h & ht_mask); bool matched = false; for (uint32_t r = c->ht_heads[slot]; r != JHT_EMPTY; r = c->ht_next[r]) { - if (join_keys_eq(c->l_key_vecs, c->r_key_vecs, c->n_keys, l, (int64_t)r)) { + if (join_keys_eq(c->l_key_vecs, c->r_key_vecs, c->n_keys, l, (int64_t)r, c->nullfree)) { count++; matched = true; } @@ -1042,14 +1108,14 @@ static void join_fill_fn(void* raw, uint32_t wid, int64_t task_start, int64_t ta } if (l + 8 < row_end) { - uint64_t pf_h = hash_row_keys(c->l_key_vecs, c->n_keys, l + 8); + uint64_t pf_h = hash_row_keys(c->l_key_vecs, c->n_keys, l + 8, c->nullfree); __builtin_prefetch(&c->ht_heads[(uint32_t)(pf_h & ht_mask)], 0, 1); } - uint64_t h = hash_row_keys(c->l_key_vecs, c->n_keys, l); + uint64_t h = hash_row_keys(c->l_key_vecs, c->n_keys, l, c->nullfree); uint32_t slot = (uint32_t)(h & ht_mask); bool matched = false; for (uint32_t r = c->ht_heads[slot]; r != JHT_EMPTY; r = c->ht_next[r]) { - if (join_keys_eq(c->l_key_vecs, c->r_key_vecs, c->n_keys, l, (int64_t)r)) { + if (join_keys_eq(c->l_key_vecs, c->r_key_vecs, c->n_keys, l, (int64_t)r, c->nullfree)) { li[off] = l; ri[off] = (int64_t)r; off++; @@ -1156,6 +1222,10 @@ static ray_t* exec_join_flat(ray_graph_t* g, ray_op_t* op, ray_t* left_table, ra return ray_error("oom", "join: sym domain runtime-id LUT build failed"); } + /* #597: one proof for the whole join — see join_keys_nullfree. */ + bool keys_nullfree = join_keys_nullfree(l_key_vecs, r_key_vecs, n_keys); + if (keys_nullfree) ray_join_nullfree_keys++; + ray_pool_t* pool = ray_pool_get(); /* Shared output state — used by both radix and chained HT paths */ @@ -1203,8 +1273,14 @@ static ray_t* exec_join_flat(ray_graph_t* g, ray_op_t* op, ray_t* left_table, ra * hash_row_keys (so it counts exactly what would form the run), and * stops at the threshold. */ { + /* #597: ray_vec_may_have_nulls is unconditionally true for + * SYM/STR (their nulls are in the payload, not in attrs), so a + * SYM-keyed join always reached the row scan below — a second + * out-of-line ray_vec_is_null per key per build row, before the + * join proper, every execution. A proven null-free key set + * cannot contain an all-null row, so the whole scan is dead. */ bool any_nullable = false; - for (uint32_t k = 0; k < n_keys && !any_nullable; k++) + for (uint32_t k = 0; k < n_keys && !any_nullable && !keys_nullfree; k++) if (build_keys[k] && ray_vec_may_have_nulls(build_keys[k])) any_nullable = true; if (any_nullable) { @@ -1236,8 +1312,8 @@ static ray_t* exec_join_flat(ray_graph_t* g, ray_op_t* op, ray_t* left_table, ra if (l_hash_hdr) scratch_free(l_hash_hdr); goto chained_ht_fallback; } - join_radix_hash_ctx_t rhctx = join_radix_hash_ctx(build_keys, n_keys, r_hashes); - join_radix_hash_ctx_t lhctx = join_radix_hash_ctx(probe_keys, n_keys, l_hashes); + join_radix_hash_ctx_t rhctx = join_radix_hash_ctx(build_keys, n_keys, r_hashes, keys_nullfree); + join_radix_hash_ctx_t lhctx = join_radix_hash_ctx(probe_keys, n_keys, l_hashes, keys_nullfree); if (pool) { ray_pool_dispatch(pool, join_radix_hash_fn, &rhctx, build_rows); ray_pool_dispatch(pool, join_radix_hash_fn, &lhctx, probe_rows); @@ -1331,7 +1407,7 @@ static ray_t* exec_join_flat(ray_graph_t* g, ray_op_t* op, ray_t* left_table, ra join_radix_bp_ctx_t bp_ctx = { .l_parts = l_parts, .r_parts = r_parts, .l_key_vecs = probe_keys, .r_key_vecs = build_keys, - .n_keys = n_keys, .join_type = join_type, + .n_keys = n_keys, .join_type = join_type, .nullfree = keys_nullfree, .l_str_desc = lhctx.str_desc, .r_str_desc = rhctx.str_desc, .l_str_pool = lhctx.str_pool, .r_str_pool = rhctx.str_pool, .pp_l = pp_l, .pp_r = pp_r, @@ -1508,6 +1584,7 @@ chained_ht_fallback:; .n_keys = n_keys, .asp_bits = asp_bits, .asp_key_max = asp_key_max, + .nullfree = keys_nullfree, }; if (pool && right_rows > RAY_PARALLEL_THRESHOLD) ray_pool_dispatch(pool, join_build_fn, &bctx, right_rows); @@ -1585,6 +1662,7 @@ chained_ht_fallback:; .matched_right = matched_right, .sjoin_bits = sjoin_bits, .sjoin_key_max = sjoin_key_max, + .nullfree = keys_nullfree, }; /* 2a: Count matches per morsel */ @@ -2100,6 +2178,10 @@ static ray_t* exec_antijoin_flat(ray_graph_t* g, ray_op_t* op, return ray_error("oom", "join: sym domain runtime-id LUT build failed"); } + /* #597: one proof for the whole anti-join — see join_keys_nullfree. */ + bool keys_nullfree = join_keys_nullfree(l_key_vecs, r_key_vecs, n_keys); + if (keys_nullfree) ray_join_nullfree_keys++; + /* Build chained hash table from right side */ ray_t* ht_next_hdr = NULL; ray_t* ht_heads_hdr = NULL; @@ -2133,6 +2215,7 @@ static ray_t* exec_antijoin_flat(ray_graph_t* g, ray_op_t* op, .n_keys = n_keys, .asp_bits = NULL, .asp_key_max = 0, + .nullfree = keys_nullfree, }; if (pool && right_rows > RAY_PARALLEL_THRESHOLD) ray_pool_dispatch(pool, join_build_fn, &bctx, right_rows); @@ -2171,11 +2254,11 @@ static ray_t* exec_antijoin_flat(ray_graph_t* g, ray_op_t* op, scratch_free(key_vecs_hdr); return ray_error("cancel", NULL); } - uint64_t h = hash_row_keys(l_key_vecs, n_keys, l); + uint64_t h = hash_row_keys(l_key_vecs, n_keys, l, keys_nullfree); uint32_t slot = (uint32_t)(h & ht_mask); bool matched = false; for (uint32_t r = ht_heads[slot]; r != JHT_EMPTY; r = ht_next[r]) { - if (join_keys_eq(l_key_vecs, r_key_vecs, n_keys, l, (int64_t)r)) { + if (join_keys_eq(l_key_vecs, r_key_vecs, n_keys, l, (int64_t)r, keys_nullfree)) { matched = true; break; /* anti-join: one match is enough to exclude */ } diff --git a/test/test_join_buildside.c b/test/test_join_buildside.c index 30ebccaf6..a39b94155 100644 --- a/test/test_join_buildside.c +++ b/test/test_join_buildside.c @@ -1071,6 +1071,186 @@ static test_result_t test_jb_mixed_type_radix(void) { PASS(); } + +/* ── #597: null-free key-column fast path ───────────────────────────────── + * A join proves once, per key column, whether the column can hold a null + * (payload scan for SYM/STR, the HAS_NULLS bit otherwise) and skips the + * per-cell null test in hash_row_keys / join_keys_eq when no key column can. + * + * ray_join_nullfree_keys counts the joins that took that path; + * ray_join_force_null_checks suppresses it, giving the differential tests a + * null-aware oracle inside one binary (same shape as the build-swap knob). + * ──────────────────────────────────────────────────────────────────────── */ + +/* SYM table; a "" entry is the canonical SYM null (id 0). */ +static ray_t* jb_sym_table1(const char* name, const char* const* vals, int64_t n) { + ray_t* col = ray_sym_vec_new(RAY_SYM_W64, n); + if (!col || RAY_IS_ERR(col)) return col; + col->len = n; + for (int64_t i = 0; i < n; i++) { + int64_t id = vals[i][0] ? ray_sym_intern(vals[i], strlen(vals[i])) : 0; + ray_write_sym(ray_data(col), i, (uint64_t)id, RAY_SYM, col->attrs); + } + ray_t* tbl = ray_table_new(1); + int64_t sym = ray_sym_intern(name, strlen(name)); + tbl = ray_table_add_col(tbl, sym, col); + ray_release(col); + return tbl; +} + +/* I64 table with an optional null at `null_at` (-1 for none). */ +static ray_t* jb_table1_null(const char* name, const int64_t* vals, int64_t n, + int64_t null_at) { + ray_t* col = ray_vec_from_raw(RAY_I64, vals, n); + if (!col || RAY_IS_ERR(col)) return col; + if (null_at >= 0) ray_vec_set_null(col, null_at, true); + ray_t* tbl = ray_table_new(1); + int64_t sym = ray_sym_intern(name, strlen(name)); + tbl = ray_table_add_col(tbl, sym, col); + ray_release(col); + return tbl; +} + +/* SYM key columns with no null cell take the null-free path. */ +static test_result_t test_jb_nf_sym_keys_prove(void) { + ray_heap_init(); + (void)ray_sym_init(); + + static const char* const lv[] = { "NYSE", "LSE", "NYSE", "TSE" }; + static const char* const rv[] = { "LSE", "NYSE", "XETR" }; + ray_t* lt = jb_sym_table1("venue", lv, 4); + ray_t* rt = jb_sym_table1("rvenue", rv, 3); + TEST_ASSERT(lt && !RAY_IS_ERR(lt), "left SYM table"); + TEST_ASSERT(rt && !RAY_IS_ERR(rt), "right SYM table"); + + uint64_t before = ray_join_nullfree_keys; + ray_t* got = jb_inner_join(lt, "venue", rt, "rvenue"); + TEST_ASSERT(got && !RAY_IS_ERR(got), "join execution"); + TEST_ASSERT(ray_join_nullfree_keys > before, + "null-free SYM key columns must take the null-free path"); + TEST_ASSERT_EQ_I(ray_table_nrows(got), 3); + + ray_release(got); + ray_release(lt); + ray_release(rt); + ray_sym_destroy(); + ray_heap_destroy(); + PASS(); +} + +/* One null SYM cell on the left blocks the fast path. SYM nulls are id 0 + * in the payload and are NOT reflected in HAS_NULLS, so this case is the + * reason the proof scans the payload instead of reading the flag. */ +static test_result_t test_jb_nf_sym_null_blocks(void) { + ray_heap_init(); + (void)ray_sym_init(); + + static const char* const lv[] = { "NYSE", "", "NYSE", "TSE" }; + static const char* const rv[] = { "LSE", "NYSE", "XETR" }; + ray_t* lt = jb_sym_table1("venue", lv, 4); + ray_t* rt = jb_sym_table1("rvenue", rv, 3); + TEST_ASSERT(lt && !RAY_IS_ERR(lt), "left SYM table"); + TEST_ASSERT(rt && !RAY_IS_ERR(rt), "right SYM table"); + TEST_ASSERT_FALSE(ray_table_get_col_idx(lt, 0)->attrs & RAY_ATTR_HAS_NULLS); + + uint64_t before = ray_join_nullfree_keys; + ray_t* got = jb_inner_join(lt, "venue", rt, "rvenue"); + TEST_ASSERT(got && !RAY_IS_ERR(got), "join execution"); + TEST_ASSERT(ray_join_nullfree_keys == before, + "a null SYM key cell must block the null-free path"); + + ray_release(got); + ray_release(lt); + ray_release(rt); + ray_sym_destroy(); + ray_heap_destroy(); + PASS(); +} + +/* I64 keys: the HAS_NULLS bit decides. */ +static test_result_t test_jb_nf_i64_flag_decides(void) { + ray_heap_init(); + (void)ray_sym_init(); + + static const int64_t lv[] = { 1, 2, 3, 4 }; + static const int64_t rv[] = { 2, 3, 9 }; + + ray_t* lt_clean = jb_table1_null("lk", lv, 4, -1); + ray_t* rt = jb_table1_null("rk", rv, 3, -1); + uint64_t before = ray_join_nullfree_keys; + ray_t* got = jb_inner_join(lt_clean, "lk", rt, "rk"); + TEST_ASSERT(got && !RAY_IS_ERR(got), "clean join execution"); + TEST_ASSERT(ray_join_nullfree_keys > before, + "HAS_NULLS-clear I64 keys must take the null-free path"); + ray_release(got); + ray_release(lt_clean); + + ray_t* lt_null = jb_table1_null("lk", lv, 4, 1); + TEST_ASSERT_TRUE(ray_table_get_col_idx(lt_null, 0)->attrs & RAY_ATTR_HAS_NULLS); + before = ray_join_nullfree_keys; + got = jb_inner_join(lt_null, "lk", rt, "rk"); + TEST_ASSERT(got && !RAY_IS_ERR(got), "nullable join execution"); + TEST_ASSERT(ray_join_nullfree_keys == before, + "a HAS_NULLS I64 key column must block the null-free path"); + + ray_release(got); + ray_release(lt_null); + ray_release(rt); + ray_sym_destroy(); + ray_heap_destroy(); + PASS(); +} + +/* The null-free path must not change any answer: run every join type, at + * both the chained and the radix size, with and without null keys, against + * the forced null-aware oracle. */ +static test_result_t test_jb_nf_differential(void) { + ray_heap_init(); + (void)ray_sym_init(); + + const int64_t sizes[] = { 64, RAY_PARALLEL_THRESHOLD + 5000 }; + const uint8_t types[] = { 0, 1, 2 }; + test_result_t result = (test_result_t){ TEST_PASS, NULL }; + + for (size_t s = 0; s < sizeof sizes / sizeof *sizes && result.status == TEST_PASS; s++) { + int64_t n = sizes[s]; + int64_t* lv = malloc((size_t)n * sizeof(*lv)); + int64_t* rv = malloc((size_t)n * sizeof(*rv)); + if (!lv || !rv) { free(lv); free(rv); ray_sym_destroy(); ray_heap_destroy(); + return (test_result_t){ TEST_FAIL, "malloc" }; } + for (int64_t i = 0; i < n; i++) { lv[i] = i % (n / 4 + 1); rv[i] = i % (n / 3 + 1); } + + for (int nulls = 0; nulls < 2 && result.status == TEST_PASS; nulls++) { + ray_t* lt = jb_table1_null("lk", lv, n, nulls ? n / 2 : -1); + ray_t* rt = jb_table1_null("rk", rv, n, nulls ? n / 3 : -1); + + for (size_t t = 0; t < sizeof types / sizeof *types && result.status == TEST_PASS; t++) { + ray_join_force_null_checks = true; + ray_t* oracle = jb_join(lt, "lk", rt, "rk", types[t]); + ray_join_force_null_checks = false; + ray_t* fast = jb_join(lt, "lk", rt, "rk", types[t]); + + if (!oracle || RAY_IS_ERR(oracle) || !fast || RAY_IS_ERR(fast)) { + result = (test_result_t){ TEST_FAIL, "differential join execution" }; + } else { + result = jb_results_equal(fast, oracle); + } + ray_release(oracle); + ray_release(fast); + } + ray_release(lt); + ray_release(rt); + } + free(lv); + free(rv); + } + + ray_join_force_null_checks = false; + ray_sym_destroy(); + ray_heap_destroy(); + return result; +} + /* ── Entry table ─────────────────────────────────────────────────────────── */ const test_entry_t join_buildside_entries[] = { @@ -1093,5 +1273,9 @@ const test_entry_t join_buildside_entries[] = { { "join_buildside/not_sticky", test_jb_not_sticky, NULL, NULL }, { "join_buildside/mixed_type_radix", test_jb_mixed_type_radix, NULL, NULL }, { "join_buildside/null_run_upfront_fallback", test_jb_null_run_upfront_fallback, NULL, NULL }, + { "join_buildside/nf_sym_keys_prove", test_jb_nf_sym_keys_prove, NULL, NULL }, + { "join_buildside/nf_sym_null_blocks", test_jb_nf_sym_null_blocks, NULL, NULL }, + { "join_buildside/nf_i64_flag_decides", test_jb_nf_i64_flag_decides, NULL, NULL }, + { "join_buildside/nf_differential", test_jb_nf_differential, NULL, NULL }, { NULL, NULL, NULL, NULL }, };