diff --git a/bench/join_tiny_ht/.gitignore b/bench/join_tiny_ht/.gitignore new file mode 100644 index 000000000..2a27e17ac --- /dev/null +++ b/bench/join_tiny_ht/.gitignore @@ -0,0 +1,2 @@ +__pycache__/ +/results.json diff --git a/bench/join_tiny_ht/README.md b/bench/join_tiny_ht/README.md new file mode 100644 index 000000000..b4658ec7b --- /dev/null +++ b/bench/join_tiny_ht/README.md @@ -0,0 +1,53 @@ +# Tiny-build chained join probe + +A fixed-key version of the issue #599 shape: 200 inner joins of a 130,000-row, +773-instrument table against three unique composite keys. Each run verifies +101,400 output rows by an independent arithmetic oracle. This is a count check, +not a full answer-equivalence proof; correctness fixtures check exact pair IDs. + +Build baseline and candidate **release** binaries in separate worktrees and +preserve copies. `make test` overwrites the worktree CLI with a sanitizer build; +do not accidentally compare that binary against release. + +```sh +python3 bench/join_tiny_ht/compare.py /path/base-release /path/candidate-release \ + --cores 1 4 --trials 5 --output /tmp/join-tiny-results.json +``` + +The Unix-only, standard-library Python wrapper shuffles fresh processes, records +binary and probe SHA256s, and explicitly selects the existing steal scheduler. +`-c N` is the CLI's **total participants** (main included); this is not the +background-worker count used by positive `RAYFORCE_CORES`. The script clears that +environment override and tracing. There is no explicit warmup. Process user/sys +CPU and process wall time include startup/setup; `loop_ms` measures the join loop. +Avoid concurrent local builds/tests and report hardware, compiler, affinity and +noise/ranges when sharing measurements. The wrapper does not certify build flags +or exclusive machine idleness. + +This is a narrow workload, not a general join or compression benchmark, a pool +policy recommendation, or complete resolution of issue #599. User CPU is not an +instruction count: clocks, SMT, topology and waiting can change thread-seconds. +For causal analysis, measure instructions/cycles separately with helper-thread +inheritance and multiplexing checks. Do not mix instrumented timing with the +uninstrumented comparison. + +## What this optimization changes + +The minimum chained table has 256 heads (1 KiB). Its count and fill passes do +not hash a future left row merely to prefetch a head in that small table. Actual +current-row hashing, equality and output are unchanged. Capacity, not distinct +key count, selects the mode once per pass; the 128/129 right-row boundary has +256/512 heads. Both serial and pooled calls use the selected callback. Shared +inline bodies keep the semantics in one place while the two constant-mode +wrappers avoid a release per-row capacity-mode test in the inspected Clang build. +The larger-table prefetch and build-side prefetch remain. The head array's size +is not proof that the entire working set stays resident. + +Exact pair-ID fixtures separately cover duplicates, nulls and misses for inner, +left and full joins, the capacity boundary, and probe rows spanning a morsel. +The benchmark's row-count oracle does not replace those checks. Choose core +counts appropriate for the machine; requested counts do not prove which physical +helpers executed tasks. The example uses one and four total participants rather +than silently oversubscribing a four-core host. Retain any unfavorable larger, +radix or long-string controls when reporting the narrow gain. Callback +specialization can change code size/layout even on a retained-prefetch path. diff --git a/bench/join_tiny_ht/compare.py b/bench/join_tiny_ht/compare.py new file mode 100644 index 000000000..4289466ef --- /dev/null +++ b/bench/join_tiny_ht/compare.py @@ -0,0 +1,68 @@ +#!/usr/bin/env python3 +"""Unix-only, stdlib benchmark. Does not build or modify either binary.""" +import argparse +import hashlib +import json +import os +from pathlib import Path +import random +import re +import resource +import statistics +import subprocess +import time + +parser = argparse.ArgumentParser() +parser.add_argument('baseline', type=Path) +parser.add_argument('candidate', type=Path) +parser.add_argument('--cores', type=int, nargs='+', default=[1, 8, 16]) +parser.add_argument('--trials', type=int, default=3) +parser.add_argument('--output', type=Path, default=Path(__file__).with_name('results.json')) +args = parser.parse_args() +if args.trials < 1 or any(c < 1 for c in args.cores): + parser.error('trials and total-participant core counts must be positive') +binaries = {name: path.resolve() for name, path in + [('baseline', args.baseline), ('candidate', args.candidate)]} +probe = Path(__file__).with_name('probe.rfl').resolve() +metadata = {name: {'path': str(path), 'sha256': hashlib.sha256(path.read_bytes()).hexdigest()} + for name, path in binaries.items()} +jobs = [(name, cores, trial) for name in binaries for cores in args.cores + for trial in range(args.trials)] +random.Random(599).shuffle(jobs) +env = os.environ.copy() +for name in ['RAYFORCE_CORES', 'RAY_POOL_TRACE']: + env.pop(name, None) +env['RAY_POOL_STEAL'] = '1' +rows = [] +for name, cores, trial in jobs: + before = resource.getrusage(resource.RUSAGE_CHILDREN) + start = time.monotonic() + proc = subprocess.run([str(binaries[name]), '-c', str(cores), str(probe)], + stdin=subprocess.DEVNULL, capture_output=True, + text=True, env=env, timeout=180) + wall = time.monotonic() - start + after = resource.getrusage(resource.RUSAGE_CHILDREN) + if proc.returncode: + raise RuntimeError(f'{name}, -c {cores}: {proc.stdout}\n{proc.stderr}') + if 'keys: ["INST0" "INST1" "INST2"]' not in proc.stdout: + raise RuntimeError(f'unexpected build keys: {proc.stdout}') + count = re.search(r'^result: (\d+)$', proc.stdout, re.M) + # 130000 rows cycling773 values: each of keys0/1/2 occurs169 times. + expected = 200 * sum(130000 // 773 + (i < 130000 % 773) for i in range(3)) + elapsed = re.search(r'^loop_ms: ([\d.e+-]+)$', proc.stdout, re.M) + if not count or int(count.group(1)) != expected or not elapsed: + raise RuntimeError(f'output/oracle mismatch: {proc.stdout}') + row = dict(binary=name, cores=cores, trial=trial, wall_s=wall, + user_s=after.ru_utime-before.ru_utime, + sys_s=after.ru_stime-before.ru_stime, + loop_ms=float(elapsed.group(1)), result=expected) + rows.append(row) + print(json.dumps(row), flush=True) +args.output.write_text(json.dumps(dict(binaries=metadata, + probe_sha256=hashlib.sha256(probe.read_bytes()).hexdigest(), + environment=dict(scheduler='steal', explicit_warmup=False), rows=rows), indent=2)) +for name in binaries: + for cores in args.cores: + group = [r for r in rows if r['binary'] == name and r['cores'] == cores] + print(name, cores, {key: statistics.median(r[key] for r in group) + for key in ['user_s', 'sys_s', 'wall_s', 'loop_ms']}) diff --git a/bench/join_tiny_ht/probe.rfl b/bench/join_tiny_ht/probe.rfl new file mode 100644 index 000000000..0e2b7b457 --- /dev/null +++ b/bench/join_tiny_ht/probe.rfl @@ -0,0 +1,10 @@ +;; Fixed-key version of the issue599 join shape; no explicit warmup. +(set N 130000) +(set syms (map (fn [i] (parse (format "INST%" i))) (til 773))) +(set t (table [venue instrument session message_id px] (list (take ['BFU] N) (take syms N) (take ['S1] N) (til N) (til N)))) +(set ks (select {from: t by: [venue instrument] n: (count px)})) +(set k3 (select {from: ks where: (in instrument ['INST0 'INST1 'INST2]) asc: instrument})) +(println "keys: %" (as 'STR (at k3 'instrument))) +(set n 0) +(println "loop_ms: %" (timeit (times 200 (set n (+ n (count (inner-join [venue instrument] t k3))))))) +(println "result: %" n) diff --git a/src/ops/join.c b/src/ops/join.c index 0b902707e..2fc3fa169 100644 --- a/src/ops/join.c +++ b/src/ops/join.c @@ -962,6 +962,7 @@ static void join_radix_build_probe_fn(void* raw, uint32_t wid, int64_t task_star /* ht_heads is accessed atomically from multiple workers during join build. * Using _Atomic(uint32_t)* for C11-compliant atomic access. */ +#define JHT_MIN_CAP 256u #define JHT_EMPTY UINT32_MAX /* sentinel for empty HT slot/chain end */ typedef struct { @@ -1036,7 +1037,7 @@ typedef struct { } join_probe_ctx_t; /* Pass 2a: count matches per morsel */ -static void join_count_fn(void* raw, uint32_t wid, int64_t task_start, int64_t task_end) { +RAY_INLINE void join_count_rows(void* raw, uint32_t wid, int64_t task_start, int64_t task_end, bool prefetch_heads) { (void)wid; (void)task_end; join_probe_ctx_t* c = (join_probe_ctx_t*)raw; uint32_t tid = (uint32_t)task_start; @@ -1059,7 +1060,7 @@ static void join_count_fn(void* raw, uint32_t wid, int64_t task_start, int64_t t continue; } - if (l + 8 < row_end) { + if (prefetch_heads && l + 8 < row_end) { 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); } @@ -1077,8 +1078,16 @@ static void join_count_fn(void* raw, uint32_t wid, int64_t task_start, int64_t t c->morsel_counts[tid] = count; } +/* Compile both modes from one body; select once per pass, not per row. */ +static void join_count_fn(void* raw, uint32_t wid, int64_t start, int64_t end) { + join_count_rows(raw, wid, start, end, true); +} +static void join_count_tiny_fn(void* raw, uint32_t wid, int64_t start, int64_t end) { + join_count_rows(raw, wid, start, end, false); +} + /* Pass 2b: fill match pairs using pre-computed offsets */ -static void join_fill_fn(void* raw, uint32_t wid, int64_t task_start, int64_t task_end) { +RAY_INLINE void join_fill_rows(void* raw, uint32_t wid, int64_t task_start, int64_t task_end, bool prefetch_heads) { (void)wid; (void)task_end; join_probe_ctx_t* c = (join_probe_ctx_t*)raw; uint32_t tid = (uint32_t)task_start; @@ -1108,7 +1117,7 @@ static void join_fill_fn(void* raw, uint32_t wid, int64_t task_start, int64_t ta continue; } - if (l + 8 < row_end) { + if (prefetch_heads && l + 8 < row_end) { 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); } @@ -1133,6 +1142,14 @@ static void join_fill_fn(void* raw, uint32_t wid, int64_t task_start, int64_t ta } } +/* Compile both modes from one body; select once per pass, not per row. */ +static void join_fill_fn(void* raw, uint32_t wid, int64_t start, int64_t end) { + join_fill_rows(raw, wid, start, end, true); +} +static void join_fill_tiny_fn(void* raw, uint32_t wid, int64_t start, int64_t end) { + join_fill_rows(raw, wid, start, end, false); +} + /* Segment count of a parted table = length of any parted column's segs * array (mirrors build_segment_table's own read, exec.c:3377-3384, and * window.c's window_parted_seg_count). */ @@ -1541,7 +1558,7 @@ static ray_t* exec_join_flat(ray_graph_t* g, ray_op_t* op, ray_t* left_table, ra chained_ht_fallback:; /* ── Chained HT path (small joins / radix OOM fallback) ────────── */ - uint64_t ht_cap64 = 256; + uint64_t ht_cap64 = JHT_MIN_CAP; uint64_t target = (uint64_t)right_rows * 2; while (ht_cap64 < target) ht_cap64 *= 2; if (ht_cap64 > UINT32_MAX) ht_cap64 = (uint64_t)1 << 31; @@ -1679,12 +1696,16 @@ chained_ht_fallback:; .nullfree = keys_nullfree, }; + /* Minimum heads occupy only 1 KiB: choose no-lookahead callbacks here. + * Bodies share their implementation; constant modes remove the row-loop + * capacity decision instead of relying on compiler loop unswitching. */ + const ray_pool_fn count_probe = ht_cap > JHT_MIN_CAP ? join_count_fn : join_count_tiny_fn; /* 2a: Count matches per morsel */ if (pool && n_tasks > 1) - ray_pool_dispatch_n(pool, join_count_fn, &probe_ctx, n_tasks); + ray_pool_dispatch_n(pool, count_probe, &probe_ctx, n_tasks); else for (uint32_t t = 0; t < n_tasks; t++) - join_count_fn(&probe_ctx, 0, t, t + 1); + count_probe(&probe_ctx, 0, t, t + 1); /* Prefix sum → morsel_offsets (reuse counts array as offsets) */ pair_count = 0; @@ -1707,11 +1728,12 @@ chained_ht_fallback:; probe_ctx.r_idx = r_idx; if (pair_count > 0) { + const ray_pool_fn fill_probe = ht_cap > JHT_MIN_CAP ? join_fill_fn : join_fill_tiny_fn; if (pool && n_tasks > 1) - ray_pool_dispatch_n(pool, join_fill_fn, &probe_ctx, n_tasks); + ray_pool_dispatch_n(pool, fill_probe, &probe_ctx, n_tasks); else for (uint32_t t = 0; t < n_tasks; t++) - join_fill_fn(&probe_ctx, 0, t, t + 1); + fill_probe(&probe_ctx, 0, t, t + 1); } CHECK_CANCEL_GOTO(pool, join_cleanup); diff --git a/test/rfl/join/tiny_chained_prefetch.rfl b/test/rfl/join/tiny_chained_prefetch.rfl new file mode 100644 index 000000000..02ed22b12 --- /dev/null +++ b/test/rfl/join/tiny_chained_prefetch.rfl @@ -0,0 +1,84 @@ +;; Exact pair-ID correctness coverage for tiny and boundary-sized build tables. +;; These cases also pass without the prefetch optimization. +;; Expected pairs were independently enumerated with tuple equality, null==null. +;; Pair IDs normalized to -1 for absent rows, then lexicographically sorted. +;; Normalization avoids assuming null sort order. IDs themselves are nonnegative. +;; No disk paths, pool knobs, cancellation or execution-policy changes. + +;; Tiny mixed SYM/I64 composite keys: duplicate Cartesian pairs, partial/all +;; nulls, near misses on the other component, and unmatched rows on both sides. +(set JPTL (table [a b li] (list (as 'SYM ["a" "a" "b" "" "a" "" "x" "a" "b" "" "z" "a"]) [1 1 2 1 0N 0N 9 2 2 2 0 1] (til 12)))) +(set JPTR (table [a b ri] (list (as 'SYM ["a" "a" "b" "" "a" "" "q" "" "a"]) [1 1 2 1 0N 0N 7 3 2] (+ 100 (til 9))))) +(set JPJ (inner-join [a b] JPTL JPTR)) +(set JPP (xasc (table [li ri] (list (fill -1 (at JPJ 'li)) (fill -1 (at JPJ 'ri)))) [li ri])) +(count JPP) -- 12 +(at JPP 'li) -- [0 0 1 1 2 3 4 5 7 8 11 11] +(at JPP 'ri) -- [100 101 100 101 102 103 104 105 108 102 100 101] +(set JPJ (left-join [a b] JPTL JPTR)) +(set JPP (xasc (table [li ri] (list (fill -1 (at JPJ 'li)) (fill -1 (at JPJ 'ri)))) [li ri])) +(count JPP) -- 15 +(at JPP 'li) -- [0 0 1 1 2 3 4 5 6 7 8 9 10 11 11] +(at JPP 'ri) -- [100 101 100 101 102 103 104 105 -1 108 102 -1 -1 100 101] +(set JPJ (full-join [a b] JPTL JPTR)) +(set JPP (xasc (table [li ri] (list (fill -1 (at JPJ 'li)) (fill -1 (at JPJ 'ri)))) [li ri])) +(count JPP) -- 17 +(at JPP 'li) -- [-1 -1 0 0 1 1 2 3 4 5 6 7 8 9 10 11 11] +(at JPP 'ri) -- [106 107 100 101 100 101 102 103 104 105 -1 108 102 -1 -1 100 101] + +;; Right build size 128: capacity starts 256 and doubles until >= 2*n. +(set JPB128L (table [a b li] (list (til 140) (take 7 140) (til 140)))) +(set JPB128R (table [a b ri] (list (til 128) (take 7 128) (+ 1000 (til 128))))) +(set JPJ (inner-join [a b] JPB128L JPB128R)) +(set JPP (xasc (table [li ri] (list (fill -1 (at JPJ 'li)) (fill -1 (at JPJ 'ri)))) [li ri])) +(count JPP) -- 128 +(all (== (at JPP 'li) (til 128))) -- true +(all (== (at JPP 'ri) (+ 1000 (til 128)))) -- true +(set JPJ (left-join [a b] JPB128L JPB128R)) +(set JPP (xasc (table [li ri] (list (fill -1 (at JPJ 'li)) (fill -1 (at JPJ 'ri)))) [li ri])) +(count JPP) -- 140 +(all (== (at JPP 'li) (til 140))) -- true +(all (== (at JPP 'ri) (concat (+ 1000 (til 128)) (take -1 12)))) -- true +(set JPJ (full-join [a b] JPB128L JPB128R)) +(set JPP (xasc (table [li ri] (list (fill -1 (at JPJ 'li)) (fill -1 (at JPJ 'ri)))) [li ri])) +(count JPP) -- 140 +(all (== (at JPP 'li) (til 140))) -- true +(all (== (at JPP 'ri) (concat (+ 1000 (til 128)) (take -1 12)))) -- true + +;; Right build size 129: capacity starts 256 and doubles until >= 2*n. +(set JPB129L (table [a b li] (list (til 140) (take 7 140) (til 140)))) +(set JPB129R (table [a b ri] (list (til 129) (take 7 129) (+ 1000 (til 129))))) +(set JPJ (inner-join [a b] JPB129L JPB129R)) +(set JPP (xasc (table [li ri] (list (fill -1 (at JPJ 'li)) (fill -1 (at JPJ 'ri)))) [li ri])) +(count JPP) -- 129 +(all (== (at JPP 'li) (til 129))) -- true +(all (== (at JPP 'ri) (+ 1000 (til 129)))) -- true +(set JPJ (left-join [a b] JPB129L JPB129R)) +(set JPP (xasc (table [li ri] (list (fill -1 (at JPJ 'li)) (fill -1 (at JPJ 'ri)))) [li ri])) +(count JPP) -- 140 +(all (== (at JPP 'li) (til 140))) -- true +(all (== (at JPP 'ri) (concat (+ 1000 (til 129)) (take -1 11)))) -- true +(set JPJ (full-join [a b] JPB129L JPB129R)) +(set JPP (xasc (table [li ri] (list (fill -1 (at JPJ 'li)) (fill -1 (at JPJ 'ri)))) [li ri])) +(count JPP) -- 140 +(all (== (at JPP 'li) (til 140))) -- true +(all (== (at JPP 'ri) (concat (+ 1000 (til 129)) (take -1 11)))) -- true + +;; 8197 probes: matches at start, both sides of morsel boundary, and tail; +;; right key 8192 duplicates, right-only 9000. Exact all-row IDs, not sums. +(set JPML (table [a b li] (list (til 8197) (take 7 8197) (til 8197)))) +(set JPMR (table [a b ri] (list [0 8191 8192 8192 8196 9000] (take 7 6) (+ 100 (til 6))))) +(set JPJ (inner-join [a b] JPML JPMR)) +(set JPP (xasc (table [li ri] (list (fill -1 (at JPJ 'li)) (fill -1 (at JPJ 'ri)))) [li ri])) +(count JPP) -- 5 +(at JPP 'li) -- [0 8191 8192 8192 8196] +(at JPP 'ri) -- [100 101 102 103 104] +(set JPJ (left-join [a b] JPML JPMR)) +(set JPP (xasc (table [li ri] (list (fill -1 (at JPJ 'li)) (fill -1 (at JPJ 'ri)))) [li ri])) +(count JPP) -- 8198 +(all (== (at JPP 'li) (concat (til 8193) (+ 8192 (til 5))))) -- true +(all (== (at JPP 'ri) (concat (concat (concat (concat [100] (take -1 8190)) (+ 101 (til 3))) (take -1 3)) [104]))) -- true +(set JPJ (full-join [a b] JPML JPMR)) +(set JPP (xasc (table [li ri] (list (fill -1 (at JPJ 'li)) (fill -1 (at JPJ 'ri)))) [li ri])) +(count JPP) -- 8199 +(all (== (at JPP 'li) (concat (+ -1 (til 8194)) (+ 8192 (til 5))))) -- true +(all (== (at JPP 'ri) (concat (concat (concat (concat (concat [105] [100]) (take -1 8190)) (+ 101 (til 3))) (take -1 3)) [104]))) -- true