feat: SQLProjection.infer/infer_batch through Confit (DRAFT-22 closed)

One artifact, two bindings: transform() stays the DuckDB oracle; infer/
infer_batch lazily prepare a DuckDBInferFn from the same serving_sql +
params + UDF objects (udfs=, shape="map"). Serving row model derives from
the training table's arrow schema. backend/boundary/output_model delegate;
unfitted use refuses by name. Gate: infer_batch == transform value-for-
value across aggregates, transformers (width 1 and 2), author UDFs, chains,
unseen-group NULLs.

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

Author: ahrzb

docs/superpowers/specs/2026-07-29-sql-projection-confit-serving-design.md

@@ -0,0 +1,43 @@
+# SQLProjection serving through Confit (DRAFT-22, the integration loop)
+
+The last leg of DRAFT-22: `infer`/`infer_batch` stop raising and bind the
+fitted artifact to Confit. Requires steps 1-3 (PRs #62, #63), both merged.
+
+## What
+
+One artifact, two bindings:
+
+```python
+p = SQLProjection(sql, transformers={...}).fit(train)
+p.transform(table)         # DuckDB batch: register UDFs, run serving_sql
+p.infer(row)               # Confit row-at-a-time: the same three pieces —
+p.infer_batch(rows)        #   serving_sql + params tables + UDF objects —
+                           #   passed to DuckDBInferFn(udfs=..., shape="map")
+```
+
+- `_serving_fn()` prepares the `DuckDBInferFn` lazily after fit and caches
+  it; refit invalidates. `backend`/`boundary`/`output_model` delegate.
+- The serving **row model derives from the training table's arrow schema**
+  (real types, guaranteed), not from `this_model` (which may carry
+  `object` fields and stays authoritative only for names/order at
+  marginalize time). Unmappable columns become opaque fields — Confit
+  accepts them unless the SQL references them. Consequence (v0 contract):
+  serving rows have the training table's shape.
+- `shape="map"`: every serving query is provably one-row-per-row (LEFT
+  params joins only), so `infer` returns exactly one typed instance.
+
+## Gate
+
+`_serving_test.py`: for every admitted family (aggregates keyed/keyless,
+transformer width-1 partitioned, width-2 global with struct bundle, author
+UDFs, CTE chains, unseen-group NULL), `infer_batch(train rows)` equals
+`transform(train)` value-for-value — the Confit binding vs the DuckDB
+oracle binding of the same artifact. Dict rows and model rows agree.
+
+## Deliberately out
+
+- Pickle/serialization of the fitted projection (artifact = serving_sql +
+  Arrow params + UDF objects; the pickle boundary is a later decision).
+- Native `confit.udfs.*` entries (DRAFT-23) — swap into the same `udfs=`
+  list when they land.
+- Serving-row schema narrowed to referenced columns only.

packages/sql-transform/sql_transform/__init__.py

@@ -1,10 +1,10 @@
 """SQLProjection — projections over ``__THIS__``, fit once, serve row-at-a-time.
 
-The fit half works today: ``fit(table)`` marginalizes every window aggregate
-into materialized params tables plus a rewritten ``serving_sql``; transformers
-and author UDFs become scalar calls that ``transform(table)`` serves by
-registering them on the connection. The serving half (``infer``/``infer_batch``
-through Confit) is a later loop and raises ``NotImplementedError``.
+``fit(table)`` marginalizes every window aggregate into materialized params
+tables plus a rewritten ``serving_sql``; transformers fit per group into
+``PythonTransform`` UDFs. One artifact, two bindings: ``transform(table)``
+runs it through DuckDB (batch, the oracle), ``infer``/``infer_batch`` run it
+through Confit (row-at-a-time, bit-exact with the DuckDB path).
 """
 
 from __future__ import annotations

packages/sql-transform/sql_transform/_projection.py

@@ -1,11 +1,13 @@
 """SQLProjection — projections over ``__THIS__``, fit once, serve row-at-a-time.
 
-The fit half works: window aggregates over ``__THIS__`` are marginalized into
+The fit half: window aggregates over ``__THIS__`` are marginalized into
 materialized params tables plus a rewritten ``serving_sql`` (see
-``_marginalize``). Transformers and author UDFs serve as scalar calls in that
-SQL — ``transform`` registers them on the connection and runs; there is no
-post-processing. The serving half (``infer``/``infer_batch`` through Confit)
-is a later loop and still raises ``NotImplementedError``.
+``_marginalize``); transformers fit per group into ``PythonTransform`` UDFs.
+The serving half is two bindings of the same artifact: ``transform`` runs
+``serving_sql`` through DuckDB with the UDFs registered (batch, the oracle),
+and ``infer``/``infer_batch`` run it through Confit's ``DuckDBInferFn`` with
+the same UDF objects passed as ``udfs=`` (row-at-a-time, bit-exact with the
+DuckDB path by Confit's contract).
 """
 
 from __future__ import annotations
@@ -32,7 +34,28 @@ def _feature_matrix(table: pa.Table, cols: list[str]):
     return mat
 
 
-_TODO = "SQLProjection serving is a later loop; {} has no implementation yet"
+def _model_from_arrow(schema: pa.Schema) -> type[BaseModel]:
+    """The serving row model, derived from the training table's schema (real
+    types, guaranteed — a declared ``this_model`` may carry ``object``
+    fields). Unmappable types become opaque ``object`` fields: Confit
+    accepts those unless the SQL references them."""
+    import pydantic
+
+    fields: dict[str, Any] = {}
+    for f in schema:
+        t = f.type
+        if pa.types.is_floating(t):
+            p: type = float
+        elif pa.types.is_integer(t):
+            p = int
+        elif pa.types.is_boolean(t):
+            p = bool
+        elif pa.types.is_string(t) or pa.types.is_large_string(t):
+            p = str
+        else:
+            p = object
+        fields[f.name] = (p | None if p is not object else Any, None)
+    return pydantic.create_model("Row", **fields)
 
 
 class SQLProjection:
@@ -90,6 +113,8 @@ def _resolve(name: str):
         self._marginalized = marginalize(sql, self._columns, _resolve)
         self._params: dict[str, pa.Table] | None = None
         self._udfs: dict[str, PythonTransform] | None = None
+        self._row_model: type[BaseModel] | None = None
+        self._fn: Any = None  # confit.DuckDBInferFn, built lazily post-fit
 
     @classmethod
     def from_file(cls, path: str) -> SQLProjection:
@@ -147,6 +172,8 @@ def fit(self, table: pa.Table, /) -> SQLProjection:
             self._params = {spec.name: materialized[spec.name] for spec in m.params}
         finally:
             con.close()
+        self._row_model = _model_from_arrow(table.schema)
+        self._fn = None  # refit invalidates the prepared serving function
         return self
 
     def _fit_step(self, step, table: pa.Table) -> tuple[pa.Table, PythonTransform]:
@@ -257,20 +284,47 @@ def udfs(self) -> dict[str, UDF]:
         out.update(self._udfs)
         return out
 
+    def _serving_fn(self):
+        """The Confit binding of the fitted artifact, prepared once: the same
+        ``serving_sql``, params tables, and UDF objects the DuckDB path uses —
+        Confit's contract makes the two bit-exact or refuses by name."""
+        if self._params is None or self._udfs is None or self._row_model is None:
+            raise MarginalizeError("not fitted: call fit(table) first")
+        if self._fn is None:
+            from confit import DuckDBInferFn
+
+            m = self._marginalized
+            udfs = [self.transformers[n] for n in m.scalar_udfs]
+            udfs += list(self._udfs.values())
+            self._fn = DuckDBInferFn(
+                m.serving_sql,
+                row_tables={"__THIS__": self._row_model},
+                static_tables=dict(self._params),
+                udfs=udfs,
+                shape="map",
+            )
+        return self._fn
+
     @property
     def backend(self) -> str:
         """Execution backend: "cranelift", "interpreter", or "constant"."""
-        raise NotImplementedError(_TODO.format("backend"))
+        return self._serving_fn().backend
 
     @property
     def boundary(self) -> str:
         """Boundary path: "marshaller", "generic", or "constant"."""
-        raise NotImplementedError(_TODO.format("boundary"))
+        return self._serving_fn().boundary
+
+    @property
+    def output_model(self) -> type[BaseModel]:
+        """The typed output row model ``infer`` returns instances of."""
+        return self._serving_fn().output_model
 
     def infer(self, row: dict[str, Any] | BaseModel, /) -> BaseModel:
         """Single-row inference; returns the typed output model instance."""
-        raise NotImplementedError(_TODO.format("infer"))
+        (out,) = self._serving_fn().infer_rows([row])  # shape="map": exactly one
+        return out
 
     def infer_batch(self, rows: list[dict[str, Any] | BaseModel], /) -> list[BaseModel]:
         """Many-rows inference; returns typed output model instances."""
-        raise NotImplementedError(_TODO.format("infer_batch"))
+        return self._serving_fn().infer_rows(rows)

packages/sql-transform/sql_transform/_projection_test.py

@@ -501,19 +501,14 @@ def test_template_input_is_a_later_loop():
         SQLProjection(t"SELECT 1 AS x FROM __THIS__")
 
 
-@pytest.mark.parametrize("method", ["infer", "infer_batch"])
-def test_serving_methods_still_raise(method):
+@pytest.mark.parametrize("attr", ["infer", "infer_batch", "backend", "boundary"])
+def test_serving_surface_requires_fit(attr):
+    # The serving half is live (see _serving_test.py); unfitted use refuses.
     p = SQLProjection("SELECT age + 1 AS b FROM __THIS__")
-    with pytest.raises(NotImplementedError, match=method):
-        getattr(p, method)(None)
-
-
-@pytest.mark.parametrize("prop", ["backend", "boundary"])
-def test_serving_properties_still_raise(prop):
-    p = SQLProjection("SELECT age + 1 AS b FROM __THIS__")
-    assert isinstance(inspect.getattr_static(SQLProjection, prop), property)
-    with pytest.raises(NotImplementedError, match=prop):
-        getattr(p, prop)
+    with pytest.raises(MarginalizeError, match="not fitted"):
+        v = getattr(p, attr)
+        if callable(v):
+            v(None)
 
 
 def test_signatures_are_stable():

packages/sql-transform/sql_transform/_serving_test.py

@@ -0,0 +1,110 @@
+"""The serving gate: infer/infer_batch (Confit) == transform (DuckDB).
+
+Both bindings run the same artifact — serving_sql, params tables, UDF
+objects — so their outputs must agree value-for-value on every admitted
+query. ``transform`` is the DuckDB oracle side; ``infer_batch`` is the
+row-at-a-time path this project exists for.
+"""
+
+import pyarrow as pa
+import pytest
+from sklearn.decomposition import PCA
+from sklearn.pipeline import Pipeline
+from sklearn.preprocessing import StandardScaler
+
+from sql_transform import MarginalizeError, PythonUDF, SQLProjection
+
+TRAIN = pa.table(
+    {
+        "country": ["US", "US", None, "DE", None, "DE", "FR"],
+        "age": [40.0, 30.0, 20.0, 25.0, 50.0, 45.0, 35.0],
+        "fare": [7, 8, 6, 9, 10, 11, 5],
+        "name": ["x", "y", "z", "w", "v", "u", "t"],
+    }
+)
+
+
+def serve_gate(sql: str, table: pa.Table = TRAIN, **kwargs) -> SQLProjection:
+    """Fit, then assert the Confit row path equals the DuckDB batch path."""
+    p = SQLProjection(sql, **kwargs).fit(table)
+    want = p.transform(table).to_pylist()
+    got = [r.model_dump() for r in p.infer_batch(table.to_pylist())]
+    assert got == want, f"\n{got}\n!=\n{want}"
+    return p
+
+
+def test_aggregates_only():
+    p = serve_gate(
+        "SELECT age - avg(age) OVER (PARTITION BY country) AS d, name FROM __THIS__"
+    )
+    assert p.backend in ("cranelift", "interpreter")
+    assert p.boundary == "marshaller"
+
+
+def test_keyless_aggregate_and_int_column():
+    serve_gate("SELECT fare - avg(fare) OVER () AS d, name FROM __THIS__")
+
+
+def test_transformer_partitioned_width1():
+    sc = StandardScaler()
+    p = serve_gate(
+        "SELECT sc(age) OVER (PARTITION BY country) * 10 + 1 AS z, name FROM __THIS__",
+        transformers={"sc": sc},
+    )
+    out = p.infer({"country": "US", "age": 33.0, "fare": 1, "name": "q"})
+    assert isinstance(out.z, float)
+
+
+def test_transformer_global_width2_list_field():
+    embed = Pipeline([("s", StandardScaler()), ("p", PCA(n_components=2))])
+    p = serve_gate(
+        "SELECT embed(struct_pack(a := age, f := fare)) OVER () AS e, name"
+        " FROM __THIS__",
+        transformers={"embed": embed},
+    )
+    out = p.infer({"country": "US", "age": 33.0, "fare": 4, "name": "q"})
+    assert isinstance(out.e, list) and len(out.e) == 2
+
+
+def test_unseen_group_is_null_row_at_a_time():
+    sc = StandardScaler()
+    p = serve_gate(
+        "SELECT sc(age) OVER (PARTITION BY country) AS z, name FROM __THIS__",
+        transformers={"sc": sc},
+    )
+    out = p.infer({"country": "JP", "age": 1.0, "fare": 1, "name": "q"})
+    assert out.z is None
+
+
+def test_author_udf_serves():
+    halve = PythonUDF(
+        "halve", lambda x: None if x is None else x / 2.0, ("f64",), ("f64",)
+    )
+    serve_gate(
+        "SELECT halve(age) - avg(halve(age)) OVER () AS d, name FROM __THIS__",
+        transformers={"halve": halve},
+    )
+
+
+def test_chain_with_transformer():
+    sc = StandardScaler()
+    serve_gate(
+        "WITH a AS (SELECT age * 2 AS age2, country, name FROM __THIS__)"
+        " SELECT sc(age2) OVER (PARTITION BY country) - 1 AS z, name FROM a",
+        transformers={"sc": sc},
+    )
+
+
+def test_model_rows_and_dict_rows_agree():
+    p = serve_gate("SELECT age - avg(age) OVER () AS d FROM __THIS__")
+    row = {"country": "US", "age": 33.0, "fare": 1, "name": "q"}
+    model_row = p._row_model(**row)
+    assert p.infer(row).model_dump() == p.infer(model_row).model_dump()
+
+
+def test_not_fitted_refuses():
+    p = SQLProjection("SELECT age - avg(age) OVER () AS d FROM __THIS__")
+    with pytest.raises(MarginalizeError, match="not fitted"):
+        p.infer({"age": 1.0})
+    with pytest.raises(MarginalizeError, match="not fitted"):
+        _ = p.backend