feat(specializer): ns/call bench per §10 — boundary baseline, four engines, raw-compute isolation (TASK-44 stretch 4)

scripts/bench_specializer.py reports p50/p99 ns/call at n in
{1,8,64,1024} for the no-op boundary baseline, cranelift, the
interpreter control (SPECIALIZER_FORCE_INTERP env knob, also a
debugging escape hatch), and the existing native + codegen engines.
Measured on this machine: the pydantic boundary dominates end-to-end
(a no-op passthrough costs within ~20% of a full join), cranelift and
interp are at parity through Python, and both beat native/codegen by
~1.5-2x at larger n. The ignored backend_compute_bench isolates raw
compute without the boundary: cranelift 13.2 ns/row vs interpreter
37.6 ns/row (2.8x) — the JIT win is real and waiting behind the
boundary, which is M-boundary's job.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: 2 people

scripts/bench_specializer.py

@@ -0,0 +1,150 @@
+"""Specializer ns/call bench — the design doc §10 measurement discipline.
+
+Reports p50/p99 ns per call at n in {1, 8, 64, 1024} for:
+  boundary   — a no-op passthrough (SELECT a FROM __THIS__) through the
+               generic pydantic path: the floor every engine pays today
+  cranelift  — the specializer's JIT backend (default)
+  interp     — the same programs on the interpreter backend (the control;
+               SPECIALIZER_FORCE_INTERP=1 in a subprocess)
+  native     — the existing DataFusion-semantics InferFn
+  codegen    — the existing Python codegen engine
+
+Run: `uv run python scripts/bench_specializer.py [--json out.json]`
+The interp control re-execs this script in a subprocess so the env knob is
+set before the module builds its functions.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import time
+
+import pyarrow as pa
+from pydantic import create_model
+
+NS = [1, 8, 64, 1024]
+CASES = {
+    "noop": "SELECT a FROM __THIS__",
+    "arith+where": "SELECT a * 2 + 1 AS x, b / 2 AS h FROM __THIS__ WHERE a % 3 <> 0",
+    "strings": "SELECT upper(s) || '-' || a AS t, substr(s, 2, 3) AS m FROM __THIS__",
+    "join": "SELECT a, name FROM __THIS__ JOIN dim ON a = dim.id",
+}
+
+
+def rows_of(n):
+    return [{"a": i % 60, "b": (i % 7) / 3.0, "s": f"item{i}"} for i in range(n)]
+
+
+def quantiles(samples_ns):
+    s = sorted(samples_ns)
+    return s[len(s) // 2], s[min(len(s) - 1, int(len(s) * 0.99))]
+
+
+def bench_fn(f, n):
+    iters = max(30, min(3000, 3_000_000 // max(n, 1)))
+    f()  # warm
+    out = []
+    for _ in range(iters):
+        t0 = time.perf_counter_ns()
+        f()
+        out.append(time.perf_counter_ns() - t0)
+    return quantiles(out)
+
+
+def build_specializer(sql, model, dim):
+    from sql_transform._interpreter import DuckDBInferFn
+
+    return DuckDBInferFn(
+        sql, row_tables={"__THIS__": model}, static_tables={"dim": dim}
+    )
+
+
+def build_native(sql, model, dim):
+    from sql_transform._interpreter import InferFn
+
+    return InferFn(sql, row_tables={"__THIS__": model}, static_tables={"dim": dim})
+
+
+def build_codegen(sql, model, dim):
+    from sql_transform._codegen import CodegenFn
+
+    return CodegenFn(sql, row_tables={"__THIS__": model}, static_tables={"dim": dim})
+
+
+def main():
+    model = create_model("Row", a=(int, ...), b=(float, ...), s=(str, ...))
+    dim = pa.table({"id": list(range(60)), "name": [f"n{i}" for i in range(60)]})
+    engine = os.environ.get("BENCH_ENGINE", "cranelift")
+
+    builders = {"cranelift": build_specializer, "interp": build_specializer}
+    if engine in ("native", "codegen"):
+        builders = {"native": build_native, "codegen": build_codegen}
+
+    results = {}
+    for case, sql in CASES.items():
+        try:
+            fn = builders[engine](sql, model, dim)
+        except Exception as e:  # noqa: BLE001 -- engines differ in coverage
+            results[case] = {"error": str(e)[:120]}
+            continue
+        if engine in ("cranelift", "interp"):
+            want = "interpreter" if engine == "interp" else "cranelift"
+            assert fn.backend == want, f"{case}: backend is {fn.backend}, wanted {want}"
+        per_n = {}
+        for n in NS:
+            objs = [model(**r) for r in rows_of(n)]
+            p50, p99 = bench_fn(lambda f=fn, o=objs: f.infer({"__THIS__": o}), n)
+            per_n[n] = {"p50_ns": p50, "p99_ns": p99}
+        results[case] = per_n
+    print(json.dumps({engine: results}))
+
+
+def orchestrate():
+    """Run every engine (interp in a subprocess for the env knob), merge."""
+    merged = {}
+    for engine in ("cranelift", "interp", "native", "codegen"):
+        env = os.environ.copy()
+        env["BENCH_ENGINE"] = engine
+        if engine == "interp":
+            env["SPECIALIZER_FORCE_INTERP"] = "1"
+        out = subprocess.run(  # noqa: S603 -- fixed argv
+            [sys.executable, __file__, "--engine-run"],
+            env=env,
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+        merged.update(json.loads(out.stdout.strip().splitlines()[-1]))
+
+    # Human table: p50 ns/call (p99 in parens), one row per case+engine.
+    hdr = f"{'case':<12} {'engine':<10}" + "".join(f"{f'n={n}':>16}" for n in NS)
+    print(hdr)
+    print("-" * len(hdr))
+    for case in CASES:
+        for engine, per_case in merged.items():
+            r = per_case.get(case)
+            if r is None:
+                continue
+            if "error" in r:
+                print(f"{case:<12} {engine:<10}  unsupported: {r['error'][:60]}")
+                continue
+            cells = "".join(
+                f"{r[str(n)]['p50_ns'] if str(n) in r else r[n]['p50_ns']:>13,}ns"
+                for n in NS
+            )
+            print(f"{case:<12} {engine:<10}{cells}")
+    if "--json" in sys.argv:
+        path = sys.argv[sys.argv.index("--json") + 1]
+        with open(path, "w", encoding="utf-8") as f:
+            json.dump(merged, f, indent=2)
+        print(f"wrote {path}")
+
+
+if __name__ == "__main__":
+    if "--engine-run" in sys.argv:
+        main()
+    else:
+        orchestrate()

src/duckdb/mod.rs

@@ -324,25 +324,33 @@ impl DuckDBInferFn {
             data.push(materialize_map(py, table, spec, keys, values)?);
         }
 
-        let fun = match cranelift::compile(&prepared.program, data) {
-            Ok(f) => Backend::Cranelift(f),
-            // The failed attempt consumed the static data; rematerialize on
-            // this cold path and fall back to the interpreter.
-            Err(_) => {
-                let mut data = Vec::with_capacity(prepared.statics.len());
-                for (spec, sty) in prepared.statics.iter().zip(&prepared.program.statics) {
-                    let StaticTy::Map { keys, values } = sty else {
-                        return Err(build_err("internal: v0 lowering emits only map statics"));
-                    };
-                    let table = static_tables
-                        .get(&spec.table)
-                        .expect("spec names come from the catalog");
-                    data.push(materialize_map(py, table, spec, keys, values)?);
+        // SPECIALIZER_FORCE_INTERP pins the interpreter — the bench control
+        // and a debugging escape hatch.
+        let force_interp = std::env::var_os("SPECIALIZER_FORCE_INTERP").is_some();
+        let fun = match (force_interp, data) {
+            (true, data) => Backend::Interp(
+                compile(&prepared.program, data).map_err(|e| build_err(e.to_string()))?,
+            ),
+            (false, data) => match cranelift::compile(&prepared.program, data) {
+                Ok(f) => Backend::Cranelift(f),
+                // The failed attempt consumed the static data; rematerialize
+                // on this cold path and fall back to the interpreter.
+                Err(_) => {
+                    let mut data = Vec::with_capacity(prepared.statics.len());
+                    for (spec, sty) in prepared.statics.iter().zip(&prepared.program.statics) {
+                        let StaticTy::Map { keys, values } = sty else {
+                            return Err(build_err("internal: v0 lowering emits only map statics"));
+                        };
+                        let table = static_tables
+                            .get(&spec.table)
+                            .expect("spec names come from the catalog");
+                        data.push(materialize_map(py, table, spec, keys, values)?);
+                    }
+                    Backend::Interp(
+                        compile(&prepared.program, data).map_err(|e| build_err(e.to_string()))?,
+                    )
                 }
-                Backend::Interp(
-                    compile(&prepared.program, data).map_err(|e| build_err(e.to_string()))?,
-                )
-            }
+            },
         };
         let output_model = match output_model {
             // Supplied models are trusted as-is in v0 (no shape validation).

src/specializer/exec/tests.rs

@@ -850,3 +850,60 @@ fn fuzz_cranelift_agrees_with_interpreter() {
         }
     }
 }
+
+/// Raw compute, no Python boundary: `cargo test --release backend_compute_
+/// bench -- --ignored --nocapture`. Informational, not a gate assertion.
+#[test]
+#[ignore = "bench, run explicitly"]
+fn backend_compute_bench() {
+    use super::cranelift;
+    use std::time::Instant;
+    let p = parse(
+        r#"fn f(in: batch{a: i64, b: f64}, out: batch{x: i64, h: f64, k: i1}) {
+entry:
+  %a = load in.a
+  %b = load in.b
+  %two = const.i64 2
+  %one = const.i64 1
+  %m = imul %a, %two
+  %x = iadd %m, %one
+  %hf = const.f64 0.5
+  %h = fmul %b, %hf
+  %z = const.f64 10.0
+  %k = fcmp.gt %h, %z
+  store out.x, %x
+  store out.h, %h
+  store out.k, %k
+  emit
+}"#,
+    )
+    .unwrap();
+    let n = 100_000usize;
+    let a: Vec<Option<i64>> = (0..n).map(|i| Some(i as i64 % 1000)).collect();
+    let b: Vec<Option<f64>> = (0..n).map(|i| Some(i as f64 / 7.0)).collect();
+    let input = batch(n, vec![c_i64(&a), c_f64(&b)]);
+
+    let fi = compile(&p, vec![]).unwrap();
+    let fc = cranelift::compile(&p, vec![]).unwrap();
+    let mut sti = fi.new_state();
+    let mut stc = fc.new_state();
+    for (name, run) in [
+        (
+            "interp",
+            &mut (|| fi.run(&input, &mut sti).unwrap()) as &mut dyn FnMut(),
+        ),
+        ("cranelift", &mut (|| fc.run(&input, &mut stc).unwrap())),
+    ] {
+        run(); // warm
+        let mut best = u128::MAX;
+        for _ in 0..20 {
+            let t = Instant::now();
+            run();
+            best = best.min(t.elapsed().as_nanos());
+        }
+        println!(
+            "{name:>10}: {:>6.2} ns/row (best of 20, n={n})",
+            best as f64 / n as f64
+        );
+    }
+}