feat(specializer): raw-dict output mode — output='dict' opt-in closes the python_dict gap (TASK-51)

The marshaller already builds the per-row dict the model would consume;
dict mode returns it and skips model construction. Typed default is
untouched; mutually exclusive with output_model; all four emit paths
honor it (marshaller / generic / reentrant / constant, the latter
copying per call). Parity gate extended: dict == typed field-for-field
before any timing. Bench gains the spec_dict row: 25-35% off the typed
path at n=1024, and the handcrafted python_dict floor shrinks from the
standing 1.3-2x caveat to 0.74-1.11x — spec_dict WINS house_prices;
what remains is input marshalling, not output.

TASK-50 Done-flip and TASK-51 ride this branch.

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

Author: ahrzb

backlog/tasks/task-50 - Specializer-SQL-support-join-forms-—-USING-residual-ON-semi-joins-comma-rewrite-wave-4.md

@@ -3,10 +3,10 @@ id: TASK-50
 title: >-
   Specializer SQL support: join forms — USING, residual ON, semi joins, comma
   rewrite (wave 4)
-status: In Progress
+status: Done
 assignee: []
 created_date: '2026-07-26 18:15'
-updated_date: '2026-07-26 19:24'
+updated_date: '2026-07-26 20:17'
 labels: []
 milestone: m-7
 dependencies:

backlog/tasks/task-51 - Specializer-raw-dict-output-mode-—-close-the-python_dict-gap-opt-in.md

@@ -0,0 +1,28 @@
+---
+id: TASK-51
+title: Specializer raw-dict output mode — close the python_dict gap (opt-in)
+status: Done
+assignee: []
+created_date: '2026-07-26 20:17'
+labels: []
+milestone: m-7
+dependencies:
+  - TASK-50
+type: feature
+ordinal: 45000
+---
+
+## Acceptance Criteria
+<!-- AC:BEGIN -->
+- [x] #1 output='dict' opt-in on DuckDBInferFn; typed default untouched; mutually exclusive with output_model; unknown values rejected
+- [x] #2 All four emit paths honor it (marshaller, generic fallback, reentrant, constant engine) with fresh dicts per call
+- [x] #3 Parity: dict mode == typed mode field-for-field, enforced in the serving parity gate + oracle tests
+- [x] #4 Bench: spec_dict engine row lands the python_dict head-to-head
+- [x] #5 gate green
+<!-- AC:END -->
+
+## Final Summary
+
+<!-- SECTION:FINAL_SUMMARY:BEGIN -->
+Raw-dict output shipped as a strict opt-in: DuckDBInferFn(..., output="dict") returns per-row plain dicts, skipping model construction — the marshaller already built the dict the model would consume, so the modes agree field-for-field by construction (enforced in the serving parity gate and tests). All four emit paths honor the mode (generated marshaller, generic baseline, reentrant fallback, constant engine — the latter copying per call so caller mutation cannot leak). Guards: mutually exclusive with output_model, unknown values rejected, output getter exposes the mode. Measured at n=1024: 25-35% off the typed path; the handcrafted python_dict floor shrinks from the standing 1.3-2x caveat to 0.74-1.11x (spec_dict WINS house_prices) — the remaining gap is input marshalling, not output. Gate green (759 + 13 xfail); 6 new tests + parity-gate extension + spec_dict bench engine.
+<!-- SECTION:FINAL_SUMMARY:END -->

benchmarks/bench_serving.py

@@ -5,6 +5,7 @@
 n in {1, 8, 64, 1024}:
 
   spec     — the specializer: cranelift + generated marshaller (ours)
+  spec_dict — the same fn built with output="dict" (raw-dict opt-in mode)
   interp   — interpreter backend, same marshaller (backend control;
              SPECIALIZER_FORCE_INTERP=1)
   generic  — cranelift behind the pre-marshaller boundary (previous
@@ -87,6 +88,10 @@ def build_callers(mod, engines):
         for e in ("spec", "interp", "generic"):
             if e in engines:
                 callers[e] = fn.infer_rows
+    if "spec_dict" in engines:
+        # The raw-dict output mode: same engine, marshaller skips model
+        # construction — python_dict's fair opponent.
+        callers["spec_dict"] = sc.build_spec_fn(mod, statics, output="dict").infer_rows
 
     for eng, cls_path in (
         ("native", "sql_transform._interpreter.InferFn"),
@@ -171,7 +176,15 @@ def orchestrate():
             sys.exit(1)
 
     groups = {
-        "main": ["spec", "native", "codegen", "duckdb", "python", "python_dict"],
+        "main": [
+            "spec",
+            "spec_dict",
+            "native",
+            "codegen",
+            "duckdb",
+            "python",
+            "python_dict",
+        ],
         "interp": ["interp"],
         "generic": ["generic"],
     }
@@ -190,7 +203,16 @@ def orchestrate():
         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"]
+    order = [
+        "spec",
+        "spec_dict",
+        "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))

benchmarks/serving_scenarios/__init__.py

@@ -67,14 +67,15 @@ def rows_table(mod, rows: list[dict]) -> pa.Table:
     return pa.Table.from_pylist(rows, schema=arrow_schema(mod.ROW_SCHEMA))
 
 
-def build_spec_fn(mod, statics: dict[str, pa.Table]):
+def build_spec_fn(mod, statics: dict[str, pa.Table], output: str = "model"):
     """The specializer serving fn (env knobs select backend/boundary)."""
     from sql_transform._interpreter import DuckDBInferFn
 
     return DuckDBInferFn(
         mod.SQL,
         row_tables={"__THIS__": row_model(mod.ROW_SCHEMA)},
         static_tables=statics,
+        output=output,
     )
 
 
@@ -119,6 +120,12 @@ def verify_parity(mod, n: int = 300) -> list[str]:
     if fn.backend != "cranelift":
         problems.append(f"{mod.NAME}: backend is {fn.backend}, not cranelift")
 
+    # The raw-dict mode must agree with the typed mode field-for-field
+    # (positional — same engine, same input order).
+    got_dict = build_spec_fn(mod, statics, output="dict").infer_rows(rows)
+    if got_dict != got_spec:
+        problems.append(f"{mod.NAME}: output='dict' differs from the typed mode")
+
     def norm(row: dict) -> tuple:
         return tuple((k, repr(v)) for k, v in row.items())
 

src/duckdb/mod.rs

@@ -235,6 +235,10 @@ struct Marshaller {
     model: Py<PyAny>,
     /// `output_model.model_validate`, present iff the model was supplied.
     validate: Option<Py<PyAny>>,
+    /// output="dict": emit the row dict itself — no model at all. The
+    /// dict is exactly what validate/slot-fill would have consumed, so
+    /// the modes agree field-for-field by construction.
+    emit_dicts: bool,
     object_new: Py<PyAny>,
     object_setattr: Py<PyAny>,
     s_dict: Py<PyString>,
@@ -252,6 +256,7 @@ impl Marshaller {
         out_cols: &[Col],
         output_model: &Py<PyAny>,
         supplied: bool,
+        emit_dicts: bool,
         fun: &Backend,
     ) -> PyResult<Marshaller> {
         let object = PyModule::import(py, "builtins")?.getattr("object")?;
@@ -270,6 +275,7 @@ impl Marshaller {
             } else {
                 None
             },
+            emit_dicts,
             object_new: object.getattr("__new__")?.unbind(),
             object_setattr: object.getattr("__setattr__")?.unbind(),
             s_dict: PyString::intern(py, "__dict__").unbind(),
@@ -387,6 +393,10 @@ impl Marshaller {
                     }
                 }
             }
+            if self.emit_dicts {
+                out.push(d.unbind().into_any());
+                continue;
+            }
             if let Some(v) = &self.validate {
                 // Supplied model: full pydantic semantics (validators,
                 // defaults, coercion, extra/private) — master parity.
@@ -430,19 +440,38 @@ pub struct DuckDBInferFn {
     row_table: String,
     #[pyo3(get)]
     output_model: Py<PyAny>,
+    output_dicts: bool,
 }
 
 #[pymethods]
 impl DuckDBInferFn {
     #[new]
-    #[pyo3(signature = (sql, row_tables, static_tables, output_model=None))]
+    #[pyo3(signature = (sql, row_tables, static_tables, output_model=None, output=None))]
     fn new(
         py: Python<'_>,
         sql: String,
         row_tables: HashMap<String, Py<PyAny>>,
         static_tables: HashMap<String, Py<PyAny>>,
         output_model: Option<Py<PyAny>>,
+        output: Option<String>,
     ) -> PyResult<Self> {
+        // output="dict" is the opt-in raw-dict mode: same engine, same
+        // lanes, the marshaller just skips model construction. The typed
+        // default is untouched.
+        let output_dicts = match output.as_deref() {
+            None | Some("model") => false,
+            Some("dict") => true,
+            Some(other) => {
+                return Err(pyo3::exceptions::PyValueError::new_err(format!(
+                    "output must be 'model' or 'dict', got '{other}'"
+                )))
+            }
+        };
+        if output_dicts && output_model.is_some() {
+            return Err(pyo3::exceptions::PyValueError::new_err(
+                "output='dict' and output_model are mutually exclusive",
+            ));
+        }
         let (row_table, model) = match row_tables.len() {
             1 => row_tables.into_iter().next().unwrap(),
             n => {
@@ -517,6 +546,7 @@ impl DuckDBInferFn {
                             engine: Engine::Constant { rows },
                             row_table,
                             output_model,
+                            output_dicts,
                         });
                     }
                     Err(_) => return Err(build_err(e.to_string())),
@@ -583,6 +613,7 @@ impl DuckDBInferFn {
                 &prepared.program.out_cols,
                 &output_model,
                 supplied,
+                output_dicts,
                 &fun,
             )?))
         };
@@ -595,9 +626,20 @@ impl DuckDBInferFn {
             },
             row_table,
             output_model,
+            output_dicts,
         })
     }
 
+    /// The output mode: "model" (typed, default) or "dict" (opt-in).
+    #[getter]
+    fn output(&self) -> &'static str {
+        if self.output_dicts {
+            "dict"
+        } else {
+            "model"
+        }
+    }
+
     /// Which engine executes: "cranelift", "interpreter", or "constant".
     #[getter]
     fn backend(&self) -> &'static str {
@@ -661,7 +703,17 @@ impl DuckDBInferFn {
                 let model = self.output_model.bind(py);
                 let mut out = Vec::with_capacity(fixed.len());
                 for r in fixed.iter() {
-                    out.push(model.call_method1("model_validate", (r,))?.unbind());
+                    if self.output_dicts {
+                        // A fresh copy per call: callers may mutate.
+                        let d = r.bind(py).cast::<PyDict>().map_err(|_| {
+                            pyo3::exceptions::PyValueError::new_err(
+                                "internal: constant rows are dicts",
+                            )
+                        })?;
+                        out.push(d.copy()?.unbind().into_any());
+                    } else {
+                        out.push(model.call_method1("model_validate", (r,))?.unbind());
+                    }
                 }
                 return Ok(out);
             }
@@ -778,7 +830,11 @@ impl DuckDBInferFn {
                     }
                 }
             }
-            out.push(model.call_method1("model_validate", (dict,))?.unbind());
+            if self.output_dicts {
+                out.push(dict.unbind().into_any());
+            } else {
+                out.push(model.call_method1("model_validate", (dict,))?.unbind());
+            }
         }
         Ok(out)
     }

tests/test_specializer_dict_output.py

@@ -0,0 +1,99 @@
+"""The raw-dict output mode: output="dict" on DuckDBInferFn.
+
+Opt-in only — the typed pydantic contract is the untouched default. Same
+engine, same lanes; the marshaller skips model construction, returning
+per-row dicts that agree with the typed mode field-for-field (the modes
+share the dict the model would have been built from).
+"""
+
+from __future__ import annotations
+
+import pytest
+from pydantic import create_model
+from test_duckdb_interpreter import duck_check, static
+
+from sql_transform._interpreter import DuckDBInferFn
+
+Row = create_model("Row", a=(int, ...), s=(str | None, None))
+SQL = "SELECT a * 2 AS d, upper(coalesce(s, 'x')) AS u FROM __THIS__ WHERE a > 0"
+ROWS = [
+    {"a": 3, "s": "hi"},
+    {"a": -1, "s": None},
+    {"a": 5, "s": None},
+]
+
+
+def _fns():
+    kw = {"row_tables": {"__THIS__": Row}, "static_tables": {}}
+    return (
+        DuckDBInferFn(SQL, **kw),
+        DuckDBInferFn(SQL, **kw, output="dict"),
+    )
+
+
+def test_dict_mode_agrees_with_typed_mode():
+    m, d = _fns()
+    rows = [Row(**r) for r in ROWS]
+    typed = [r.model_dump() for r in m.infer({"__THIS__": rows})]
+    dicts = d.infer({"__THIS__": rows})
+    assert typed == dicts
+    assert all(type(r) is dict for r in dicts)
+    assert m.output == "model" and d.output == "dict"
+
+
+def test_dict_mode_with_join_and_nulls():
+    dim = static(
+        {"id": "int", "name": "str"},
+        [{"id": 1, "name": "one"}, {"id": 3, "name": "three"}],
+    )
+    fn = DuckDBInferFn(
+        "SELECT k, name FROM __THIS__ LEFT JOIN dim ON k = dim.id",
+        row_tables={"__THIS__": create_model("K", k=(int, ...))},
+        static_tables={"dim": dim},
+        output="dict",
+    )
+    K = create_model("K", k=(int, ...))
+    got = fn.infer({"__THIS__": [K(k=1), K(k=2)]})
+    assert got == [{"k": 1, "name": "one"}, {"k": 2, "name": None}]
+
+
+def test_dict_mode_returns_fresh_dicts_per_call():
+    _, d = _fns()
+    rows = [Row(a=1)]
+    first = d.infer({"__THIS__": rows})
+    first[0]["d"] = "mutated"
+    again = d.infer({"__THIS__": rows})
+    assert again == [{"d": 2, "u": "X"}]
+
+
+def test_dict_mode_on_constant_engine():
+    fn = DuckDBInferFn(
+        "SELECT 1 AS one, 'x' AS s",
+        row_tables={"__THIS__": Row},
+        static_tables={},
+        output="dict",
+    )
+    got = fn.infer({"__THIS__": []})
+    assert all(type(r) is dict for r in got)
+    # Mutating a returned dict must not leak into the next call.
+    if got:
+        got[0]["one"] = "mutated"
+        assert fn.infer({"__THIS__": []}) != got
+
+
+def test_dict_mode_guards():
+    kw = {"row_tables": {"__THIS__": Row}, "static_tables": {}}
+    with pytest.raises(ValueError, match="mutually exclusive"):
+        DuckDBInferFn(SQL, **kw, output="dict", output_model=Row)
+    with pytest.raises(ValueError, match="'model' or 'dict'"):
+        DuckDBInferFn(SQL, **kw, output="bogus")
+
+
+def test_dict_mode_oracle_parity():
+    # Same duck_check oracle discipline: dict-mode dumps == duckdb rows.
+    m, d = _fns()
+    rows = [Row(**r) for r in ROWS]
+    assert d.infer({"__THIS__": rows}) == [
+        r.model_dump() for r in m.infer({"__THIS__": rows})
+    ]
+    duck_check(SQL, {"a": "int", "s": "str?"}, ROWS)