perf(join): skip speculative hashing for minimum chained tables

bench/join_tiny_ht/.gitignore

@@ -0,0 +1,2 @@
+__pycache__/
+/results.json

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.

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']})

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)

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);

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