Specializer M-boundary: generated row marshaller + SpecializedTransform (TASK-45)

backlog/tasks/task-44 - Specializer-M-cranelift-codegen-backend-behind-the-same-interface.md

@@ -1,10 +1,10 @@
 ---
 id: TASK-44
 title: 'Specializer M-cranelift: codegen backend behind the same interface'
-status: In Progress
+status: Done
 assignee: []
 created_date: '2026-07-25 02:31'
-updated_date: '2026-07-26 06:05'
+updated_date: '2026-07-26 07:55'
 labels: []
 milestone: m-7
 dependencies:
@@ -23,10 +23,10 @@ Cranelift-jit backend for the imperative IR (design doc §7; cranelift-jit 0.126
 
 ## Acceptance Criteria
 <!-- AC:BEGIN -->
-- [ ] #1 Every v0-subset prepared query compiles under cranelift and agrees with the interpreter backend on the corpus and on randomized inputs
-- [ ] #2 Uncovered ops fall back to the interpreter backend rather than failing prepare
-- [ ] #3 p50/p99 ns/call reported at n in {1, 8, 64, 1024} against the interpreter control and the existing native + codegen engines
-- [ ] #4 mise gate-specializer green
+- [x] #1 Every v0-subset prepared query compiles under cranelift and agrees with the interpreter backend on the corpus and on randomized inputs
+- [x] #2 Uncovered ops fall back to the interpreter backend rather than failing prepare
+- [x] #3 p50/p99 ns/call reported at n in {1, 8, 64, 1024} against the interpreter control and the existing native + codegen engines
+- [x] #4 mise gate-specializer green
 <!-- AC:END -->
 
 ## Implementation Plan
@@ -38,3 +38,15 @@ Stretch plan (recorded 2026-07-26, design doc §7 + §10):
 3. Differential: gen.rs random-IR fuzz interpreter-vs-cranelift (same seeds, byte-identical outputs incl. NaN/-0.0/arena strings); corpus replay + duck_check suite through the cranelift backend; gate green.
 4. Bench per §10: baseline the boundary with a no-op f first, then p50/p99 ns/call at n in {1,8,64,1024}, interpreter as control, next to the existing native + codegen engines; committed as a script, numbers into the task.
 <!-- SECTION:PLAN:END -->
+
+## Implementation Notes
+
+<!-- SECTION:NOTES:BEGIN -->
+All four stretches landed in one session (commits 6fd137b/0362b37/222138c on claude/specializer-m-cranelift, PR #27). Backend shape: one JIT'd fn per program, called per row — extern "C" fn(*mut Cx) -> i64 (0 emit / 1 skip / 2 helper trap / 3+k term trap); Cx is repr(C), the generated code reads only trap_flag at offset 0 (checked after every fallible helper). Coverage-first: consts/float arith/int cmp/logic/select/itof/fabs inline in CLIF; everything nontrivial calls extern helpers delegating to the interpreter's own semantic fns (casemap, substr_window, duck_fcmp, DuckF64, arena) — backends cannot drift where they share code. Strings = (off,len) i64 pairs; IR blocks map 1:1 to CLIF blocks+params. CraneliftFn owns a compiled InterpFn (checks, statics, fallback). The 500-seed random-IR differential caught two real bugs pre-landing: load.opt and sload.opt must normalize payloads to type defaults under a false flag. DuckDBInferFn compiles cranelift-first with interpreter fallback (AC #2) + SPECIALIZER_FORCE_INTERP knob + .backend getter; the whole pytest suite (607 + corpus 53/625/0) now exercises the JIT. Bench (scripts/bench_specializer.py, §10 discipline): pydantic boundary dominates end-to-end — noop passthrough within ~20% of a full join; cranelift==interp through Python; both beat native/codegen ~1.5-2x at larger n. Raw compute isolation (backend_compute_bench, ignored test): cranelift 13.2 ns/row vs interp 37.6 ns/row = 2.8x — the JIT win is real and waiting behind the boundary (M-boundary's job). MILESTONE COMPLETE pending review; hard stop before M-boundary.
+<!-- SECTION:NOTES:END -->
+
+## Final Summary
+
+<!-- SECTION:FINAL_SUMMARY:BEGIN -->
+M-cranelift delivered on claude/specializer-m-cranelift (PR #27): cranelift-jit 0.126 backend with FULL IR instruction coverage behind the same interface. Per-row extern "C" ABI; coverage-first op strategy (inline CLIF where trivially safe, shared-semantics extern helpers elsewhere — the two backends physically share the semantic functions). 500-seed interpreter-vs-cranelift random-IR differential green (caught two real payload-normalization bugs before landing); whole pytest suite incl. 678-case corpus replay (53 match / 625 clean-unsupported / 0 FAIL) runs on the JIT; interpreter fallback + force knob wired (AC #2). Bench per design doc §10 at n in {1,8,64,1024} vs interp control + native + codegen: boundary dominates end-to-end (JIT==interp through Python, both ~1.5-2x ahead of the existing engines); raw compute isolated: 13.2 vs 37.6 ns/row = 2.8x for cranelift. Gate green (cargo + 607 pytest, 12 xfail). Next milestone (M-boundary, generated row marshaller) is where the compute win becomes end-to-end visible.
+<!-- SECTION:FINAL_SUMMARY:END -->

backlog/tasks/task-45 - Specializer-M-boundary-generated-row-marshaller-Python-API.md

@@ -1,9 +1,10 @@
 ---
 id: TASK-45
 title: 'Specializer M-boundary: generated row marshaller + Python API'
-status: To Do
+status: In Progress
 assignee: []
 created_date: '2026-07-25 02:32'
+updated_date: '2026-07-26 09:10'
 labels: []
 milestone: m-7
 dependencies:
@@ -22,9 +23,26 @@ Wire the specializer into the Python surface: SpecializedTransform (SQLTransform
 
 ## Acceptance Criteria
 <!-- AC:BEGIN -->
-- [ ] #1 SpecializedTransform fit/infer/infer_batch works end-to-end on the v0 subset with dict and pydantic-model rows
-- [ ] #2 No-op-f boundary baseline reported: generic pydantic path vs generated marshaller, p50/p99 at n in {1, 8, 64, 1024}
-- [ ] #3 End-to-end p50/p99 vs the current native and codegen engines reported
-- [ ] #4 Steady-state hot path allocates nothing per call beyond the output objects; arena reset only
-- [ ] #5 mise gate-specializer green
+- [x] #1 SpecializedTransform fit/infer/infer_batch works end-to-end on the v0 subset with dict and pydantic-model rows
+- [x] #2 No-op-f boundary baseline reported: generic pydantic path vs generated marshaller, p50/p99 at n in {1, 8, 64, 1024}
+- [x] #3 End-to-end p50/p99 vs the current native and codegen engines reported
+- [x] #4 Steady-state hot path allocates nothing per call beyond the output objects; arena reset only
+- [x] #5 mise gate-specializer green
 <!-- AC:END -->
+
+## Implementation Plan
+
+<!-- SECTION:PLAN:BEGIN -->
+Stretch plan (recorded 2026-07-26, design doc §3 flag 1 + §10). Measured targets from the M-cranelift bench: per-cell getattr with a fresh name string, per-call buffer allocs, and model_validate per output row — those three ARE the boundary that dominates end-to-end.
+1. Marshaller core (src/duckdb, the boundary module): prepare-time interned PyString field names for input attrs and output keys; input rows accepted as dict (get_item on interned key) or pydantic model (getattr on interned name), unboxed type-directed straight into the existing Batch columns — SoA stays: both backends read Batch, the batch is L1-resident at this n, and the doc's own flag-1 argument makes layout irrelevant here (deviation from the AoS row-struct line, noted deliberately); output built model_construct-style (cached bound method + interned keys), never model_validate; RunState + input buffers owned by the fn object behind a Mutex, cleared not dropped per call. SPECIALIZER_GENERIC_BOUNDARY env knob keeps the old generic path runnable for the baseline; a .boundary getter mirrors .backend.
+2. SpecializedTransform (sql_transform package): SQLTransform API minus transformer refs — ctor(sql | Template), fit(table, this_model=None) reusing desugar/inline_references/build_state_tables/rewrite_sql then preparing DuckDBInferFn (clear error on transformer refs; records output only, dense raises); infer/infer_batch pass dicts/models straight through, no SimpleNamespace hop. Window aggregates rewrite into static-table equi-joins = v0 subset, so they ride along where the rewrite output parses.
+3. Bench per §10: extend scripts/bench_specializer.py — no-op f through generic vs marshalled boundary (AC #2, the marshaller's win as a measured number), end-to-end p50/p99 at n in {1,8,64,1024} vs native + codegen (AC #3); numbers into this ticket.
+4. Zero-alloc steady state (AC #4): counting-global-allocator Rust test around the reused-state run path asserting no Rust-side allocation on the second call (arena reset only); marshaller buffer reuse asserted the same way. Gate green (AC #5).
+<!-- SECTION:PLAN:END -->
+
+## Implementation Notes
+
+<!-- SECTION:NOTES:BEGIN -->
+All four stretches landed on claude/specializer-m-boundary. The marshaller (src/duckdb/mod.rs) does at prepare time everything knowable at prepare time: interned attribute-name PyStrings in fixed field order, input buffers + RunState owned and cleared-not-dropped per call, dict rows via get_item and model rows via getattr on the interned names, output rows by direct pydantic-v2 slot fill. Two assumptions died by measurement: (1) pydantic's literal model_construct API is pure-Python and SLOWER than model_validate (1432 vs 882 ns/row on 2.13) — the shipped path is object.__new__ + object.__setattr__ of __dict__/__pydantic_fields_set__/__pydantic_extra__/__pydantic_private__ at 491 ns, semantically equal (eq, fields_set, assignment all verified); (2) the design doc's AoS row structs buy nothing at L1-resident n — the marshaller fills the existing SoA Batch directly (deliberate deviation, noted in the plan). AC #4 forced real work: ColData::Str became one flat buffer + spans (killing per-cell Strings AND a hidden per-load clone in the JIT's h_load_str), substr/trim became pure sub-span arithmetic, case mapping streams into the arena (Arena::case_map), number→text formats via Arena::push_fmt with DuckF64 on a stack buffer, and h_probe emits without its per-call Vec + ScalarVal clones — all shared between backends, pinned by counting-allocator tests over a probe/arith fixture and a string-heavy program on BOTH backends (the remaining per-call allocs are the output objects themselves plus pyo3's input-list Vec, i.e. the AC's "beyond the output objects"). SpecializedTransform (sql_transform/_specialized.py) reuses the whole fit pipeline minus transformer refs (ValueError at ctor), so window aggregates ride the equi-join rewrite onto cranelift — parity with SQLTransform asserted row-for-row; WHERE stays rejected at the authoring surface (parse_and_validate), while raw DuckDBInferFn keeps it. Bench (scripts/bench_specializer.py, +generic engine via SPECIALIZER_GENERIC_BOUNDARY): noop p50 marshaller vs generic = 1.1/2.1µs at n=1, 468/1192µs at n=1024 (1.9-2.5x, AC #2); vs shipping engines at n=1024 the specializer is 3.7-4.2x faster than native and 2.4-3.8x than codegen (AC #3); cranelift-vs-interp is now visible end-to-end (arith 339 vs 391µs at n=1024). SPECIALIZER_GENERIC_BOUNDARY + .boundary getter mirror the FORCE_INTERP pattern; infer_rows() is the direct hot entry SpecializedTransform uses.
+<!-- SECTION:NOTES:END -->
+

benchmarks/bench_serving.py

@@ -0,0 +1,195 @@
+"""Realistic serving-path bench — wide tables, involved feature engineering.
+
+The scenarios (benchmarks/serving_scenarios/) reproduce the inference paths
+of famous tabular-ML problems. Engines compared, p50/p99 ns per call at
+n in {1, 8, 64, 1024}:
+
+  spec     — the specializer: cranelift + generated marshaller (ours)
+  interp   — interpreter backend, same marshaller (backend control;
+             SPECIALIZER_FORCE_INTERP=1)
+  generic  — cranelift behind the pre-marshaller boundary (previous
+             boundary architecture; SPECIALIZER_GENERIC_BOUNDARY=1)
+  native   — the previous DataFusion-semantics InferFn
+  codegen  — the previous Python-codegen engine
+  duckdb   — DuckDB itself per call (statics pre-materialized as native
+             tables; per call: Arrow batch from row dicts -> register ->
+             execute -> fetch)
+  python   — the handcrafted twin: what an engineer would hand-write for a
+             microservice, returning the same typed output models
+  python_dict — same, returning plain dicts (the absolute floor; JSON only)
+
+A three-way parity gate (specializer == DuckDB == handcrafted) runs before
+any timing; a scenario that disagrees aborts the bench.
+
+Run: uv run python -m benchmarks.bench_serving [--json out.json]
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import time
+
+from benchmarks import serving_scenarios as sc
+
+NS = [1, 8, 64, 1024]
+
+ENGINE_ENV = {
+    "spec": {},
+    "interp": {"SPECIALIZER_FORCE_INTERP": "1"},
+    "generic": {"SPECIALIZER_GENERIC_BOUNDARY": "1"},
+}
+
+
+def scenario_names():
+    """All scenarios, or a comma-separated BENCH_SCENARIOS subset."""
+    sub = os.environ.get("BENCH_SCENARIOS")
+    return sub.split(",") if sub else sc.NAMES
+
+
+def quantiles(samples):
+    s = sorted(samples)
+    return s[len(s) // 2], s[min(len(s) - 1, int(len(s) * 0.99))]
+
+
+def bench_call(f, n):
+    iters = max(30, min(2000, 2_000_000 // max(n, 1)))
+    f()  # warm
+    out = []
+    # 3s budget per cell (min 30 samples) — keeps the ms-scale engines
+    # (duckdb per call) from dominating wall time.
+    deadline = time.perf_counter_ns() + 3_000_000_000
+    for _ in range(iters):
+        t0 = time.perf_counter_ns()
+        f()
+        out.append(time.perf_counter_ns() - t0)
+        if len(out) >= 30 and time.perf_counter_ns() > deadline:
+            break
+    p50, p99 = quantiles(out)
+    return {"p50_ns": p50, "p99_ns": p99}
+
+
+def build_callers(mod, engines):
+    """-> {engine: callable(rows) or None if the engine can't serve it}."""
+    statics = mod.make_statics(sc.SEED)
+    model = sc.row_model(mod.ROW_SCHEMA)
+    callers = {}
+
+    if any(e in engines for e in ("spec", "interp", "generic")):
+        fn = sc.build_spec_fn(mod, statics)
+        for e in ("spec", "interp", "generic"):
+            if e in engines:
+                callers[e] = fn.infer_rows
+
+    for eng, cls_path in (
+        ("native", "sql_transform._interpreter.InferFn"),
+        ("codegen", "sql_transform._codegen.CodegenFn"),
+    ):
+        if eng not in engines:
+            continue
+        mod_path, cls_name = cls_path.rsplit(".", 1)
+        cls = getattr(__import__(mod_path, fromlist=[cls_name]), cls_name)
+        try:
+            f = cls(mod.SQL, row_tables={"__THIS__": model}, static_tables=statics)
+            callers[eng] = lambda rows, f=f: f.infer({"__THIS__": rows})
+        except Exception as e:  # noqa: BLE001 -- engines differ in coverage
+            callers[eng] = str(e)[:140]
+
+    if "duckdb" in engines:
+        duck = sc.duckdb_server(mod, statics)
+        callers["duckdb"] = duck
+
+    if "python" in engines or "python_dict" in engines:
+        hand = mod.handcrafted(statics)
+        out_model = sc.build_spec_fn(mod, statics).output_model
+        if "python" in engines:
+            callers["python"] = lambda rows: [out_model(**hand(r)) for r in rows]
+        if "python_dict" in engines:
+            callers["python_dict"] = lambda rows: [hand(r) for r in rows]
+
+    return callers
+
+
+def engine_run(engines):
+    results = {}
+    for mod in map(sc.load, scenario_names()):
+        callers = build_callers(mod, engines)
+        per = {}
+        for eng, call in callers.items():
+            if isinstance(call, str):
+                per[eng] = {"error": call}
+                continue
+            per[eng] = {}
+            for n in NS:
+                rows = mod.make_rows(sc.SEED + 2, n)
+                # Engines take model objects on their classic surface;
+                # spec-family and duckdb/python take dicts natively. Feed
+                # each what its real caller would: dict rows for
+                # spec/duckdb/python, model objects for native/codegen
+                # (their only supported input shape).
+                if eng in ("native", "codegen"):
+                    model = sc.row_model(mod.ROW_SCHEMA)
+                    rows = [model(**r) for r in rows]
+                per[eng][n] = bench_call(lambda c=call, r=rows: c(r), n)
+        results[mod.NAME] = per
+    print(json.dumps(results))
+
+
+def orchestrate():
+    print("parity gate:", flush=True)
+    for mod in map(sc.load, scenario_names()):
+        problems = sc.verify_parity(mod)
+        tag = "ok" if not problems else "FAIL"
+        print(f"  {mod.NAME:<14} {tag}")
+        if problems:
+            print("\n".join("    " + p for p in problems))
+            sys.exit(1)
+
+    groups = {
+        "main": ["spec", "native", "codegen", "duckdb", "python", "python_dict"],
+        "interp": ["interp"],
+        "generic": ["generic"],
+    }
+    merged: dict = {}
+    for group, engines in groups.items():
+        env = os.environ.copy()
+        env.update(ENGINE_ENV.get(group if group != "main" else "spec", {}))
+        env["BENCH_ENGINES"] = ",".join(engines)
+        out = subprocess.run(  # noqa: S603 -- fixed argv
+            [sys.executable, "-m", "benchmarks.bench_serving", "--engine-run"],
+            env=env,
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+        for scenario, per in json.loads(out.stdout.strip().splitlines()[-1]).items():
+            merged.setdefault(scenario, {}).update(per)
+
+    order = ["spec", "interp", "generic", "native", "codegen", "duckdb", "python"]
+    hdr = f"{'scenario':<14} {'engine':<9}" + "".join(f"{f'n={n}':>15}" for n in NS)
+    print("\n" + hdr)
+    print("-" * len(hdr))
+    for scenario, per in merged.items():
+        for eng in order:
+            r = per.get(eng)
+            if r is None:
+                continue
+            if "error" in r:
+                print(f"{scenario:<14} {eng:<9}  unsupported: {r['error'][:70]}")
+                continue
+            cells = "".join(f"{r[str(n)]['p50_ns']:>13,}ns" for n in NS)
+            print(f"{scenario:<14} {eng:<9}{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=1)
+        print(f"wrote {path}")
+
+
+if __name__ == "__main__":
+    if "--engine-run" in sys.argv:
+        engine_run(os.environ["BENCH_ENGINES"].split(","))
+    else:
+        orchestrate()