feat: SQLProjection — marginalization over __THIS__

Rename SQLTransform to SQLProjection (strict projections only) and
implement the fit half: window aggregates over __THIS__ are marginalized
into materialized params tables plus a rewritten serving_sql that LEFT
JOINs them via IS NOT DISTINCT FROM. Parsing is DuckDB's own
json_serialize_sql/json_deserialize_sql — the oracle's grammar and
printer, no second SQL dialect. Confit is untouched.

Fit is two-stage, both stages fuzz-derived: windows_sql reruns the
original computation with each window re-projected (GROUP BY and solo
window queries drift by an ulp — DuckDB sums floats in operator-chain-
dependent order), then per-keyset SELECT DISTINCT collapses the
materialized values. Fit pins threads=1: DuckDB's parallel window
aggregation is not bit-deterministic (measured 1/500).

The gate is differential: original SQL over train == serving_sql joined
to fitted params, both executed by DuckDB, bit-exact including schema —
plus a seeded 500-case fuzz, AST shape pins, and a named-refusal table.

Serving (infer/infer_batch through Confit) stays NotImplementedError —
the named next loop.

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

Author: 2 people

README.md

@@ -10,13 +10,7 @@ This repository is a workspace of two packages:
 | package | what it is |
 |---|---|
 | [`packages/confit`](packages/confit) | **Confit** — the serving engine. SQL plus static tables frozen at fit time are partially evaluated, once, into a native function. Serves bit-exact with DuckDB or refuses at build time. Usable on its own. |
-| [`packages/sql-transform`](packages/sql-transform) | The authoring surface: `SQLTransform`. **Not implemented** — signatures only, being rebuilt on Confit. |
-
-> **sql-transform is a stub.** Its implementation was deliberately removed --
-> the DataFusion engine, the codegen backend, the batch `transform()` path and
-> the fit pipeline are all gone. What remains is the interface, raising
-> `NotImplementedError`, as the contract for a rebuild on Confit.
-> **Confit itself is complete and unaffected.**
+| [`packages/sql-transform`](packages/sql-transform) | The authoring surface: `SQLProjection`. The **fit half works**: window aggregates over `__THIS__` are marginalized into materialized params tables plus a rewritten serving SQL. The serving half (through Confit) is a later loop. |
 
 ## Installation
 
@@ -67,23 +61,35 @@ fn.infer_rows([Row(age=40.0)])          # row objects in, row objects out
 fn.infer_arrow(pa.table({"age": [40.0]}))  # pa.Table in, pa.Table out
 ```
 
-`sql_transform.SQLTransform` is the intended authoring layer on top of it --
-write SQL with window aggregates, `fit()` to freeze them, then serve. It is
-currently **not implemented**; see [packages/sql-transform](packages/sql-transform).
+`sql_transform.SQLProjection` is the authoring layer on top of it — write SQL
+with window aggregates, `fit()` to freeze them. The fit half works today; see
+[packages/sql-transform](packages/sql-transform).
+
+```python
+from sql_transform import SQLProjection
+
+p = SQLProjection(
+    "SELECT (age - avg(age) OVER (PARTITION BY country)) AS d FROM __THIS__"
+).fit(train)
+p.serving_sql   # the rewritten projection: params joins instead of aggregates
+p.params        # {"__CF_PARAMS_0__": <pyarrow.Table>}
+```
 
 ## Architecture
 
-The intended two-phase shape, of which **only the second phase exists today**:
+The two-phase shape — **fit works, the wiring between the phases is the next
+loop**:
 
 ```
 SQL over __THIS__
       │
       ▼
-   fit(train) ── freeze each window aggregate (e.g. MEAN(age)) into a typed
-      │          __STATE__ table and rewrite the SQL to reference it instead
-      │          of recomputing.                         [NOT IMPLEMENTED]
+   fit(train) ── marginalize each window aggregate (e.g. avg(age) OVER
+      │          (PARTITION BY country)) into a materialized params table and
+      │          rewrite the SQL to LEFT JOIN it instead of recomputing.
+      │          Bit-exact with DuckDB by differential gate.      [WORKS]
       │
-      │  rewritten SQL + frozen state
+      │  rewritten SQL + frozen params          [wiring: NOT IMPLEMENTED]
       ▼
    Confit ── partially evaluates the pair into a native function: binding-time
       │      analysis collapses every static lookup into a prepare-time probe,
@@ -105,9 +111,10 @@ The expression surface, joins to static tables, the row-shape contract
 mined from DuckDB's own test suite replay bit-exact, with the remainder clean,
 named build-time rejections.
 
-The authoring layer on top — window-aggregate `fit`, typed I/O, sklearn
-transformer references, and composing one transform into another — is what the
-rebuild has to supply.
+The authoring layer's window-aggregate `fit` (marginalization) works; typed
+I/O, serving through Confit, static tables in authored SQL, and the broader
+DRAFT-20 program (fitted model artifacts as params tables) are the named next
+loops.
 
 ## Reports
 

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

@@ -23,24 +23,52 @@ FROM __THIS__
 
 ```sql
 -- serving_sql (generated)
-SELECT (t.age - p0.__cf_a0) / p0.__cf_a1 AS age_z,
-       t.fare - p1.__cf_a0 AS fare_c
-FROM __THIS__ AS t
-LEFT JOIN __CF_PARAMS_0__ AS p0 ON t.country IS NOT DISTINCT FROM p0.country
-CROSS JOIN __CF_PARAMS_1__ AS p1
+SELECT ((__cf_t.age - __cf_p0.__cf_a0) / __cf_p0.__cf_a1) AS age_z,
+       (__cf_t.fare - __cf_p1.__cf_a0) AS fare_c
+FROM __THIS__ AS __cf_t
+LEFT JOIN __CF_PARAMS_0__ AS __cf_p0
+  ON ((__cf_t.country IS NOT DISTINCT FROM __cf_p0.country))
+LEFT JOIN __CF_PARAMS_1__ AS __cf_p1 ON ((1 = 1))
 ```
 
 ```sql
--- params[0].fit_sql   (keys = [country])
-SELECT country, avg(age) AS __cf_a0, stddev_samp(age) AS __cf_a1
-FROM __THIS__ GROUP BY country
+-- windows_sql: ONE fit-time execution — the original select list verbatim
+-- (chain-pinning, discarded), each distinct window re-projected, the keys:
+SELECT (age - avg(age) OVER (PARTITION BY country))
+       / stddev_samp(age) OVER (PARTITION BY country) AS __cf_o0,
+       (fare - avg(fare) OVER ()) AS __cf_o1,
+       avg(age) OVER (PARTITION BY country) AS __cf_w0,
+       stddev_samp(age) OVER (PARTITION BY country) AS __cf_w1,
+       avg(fare) OVER () AS __cf_w2,
+       country AS __cf_k0
+FROM __THIS__
+-- params[0].fit_sql   (keys = [country]) — pure value picking, no arithmetic:
+SELECT DISTINCT __cf_k0 AS country, __cf_w0 AS __cf_a0, __cf_w1 AS __cf_a1
+FROM __CF_WINDOWS__
 -- params[1].fit_sql   (keys = [])
-SELECT avg(fare) AS __cf_a0 FROM __THIS__
+SELECT DISTINCT __cf_w2 AS __cf_a0 FROM __CF_WINDOWS__
 ```
 
-`fit(train)` runs each `fit_sql` over the training table via DuckDB and stores
-the resulting tables. Serving (a later slice) hands `serving_sql` + params to
-Confit as static tables.
+`fit(train)` registers the training table, runs `windows_sql` once
+(single-threaded, see below), registers the materialized result as
+`__CF_WINDOWS__`, and collapses each params table out of it with SELECT
+DISTINCT — every allowlisted aggregate is deterministic per group, so the
+tuple is constant within a group and DISTINCT yields exactly one row per
+group. Serving (a later slice) hands `serving_sql` + params to Confit as
+static tables.
+
+**Why fit is two-stage — measured, not designed.** The differential fuzz
+killed two simpler forms. A `GROUP BY` fit drifts from the original text by
+an ulp: DuckDB's group-by and window aggregates sum floats in different
+orders. A standalone per-keyset window query drifts too: DuckDB chains window
+operators, each reordering rows for the next, so a float aggregate's
+summation order depends on which *other* windows share the query. Keeping the
+original select items in `windows_sql` pins the operator chain, and
+re-projecting an already-present window is CSE'd — the original text's value,
+bit-exactly. The last ulp source is parallelism itself: DuckDB's parallel
+window aggregation is schedule-dependent (measured 1/500 cases), so fit pins
+`SET threads = 1`, making params deterministic and machine-reproducible; the
+gate compares both sides single-threaded.
 
 ## Architecture: parse with the oracle
 
@@ -80,16 +108,23 @@ regardless of parser.
    normalization) within a key set dedupe to one column.
 2. **Join predicate is `IS NOT DISTINCT FROM`, never `USING`/`=`.** Window
    `PARTITION BY` groups NULL keys into one partition; equality joins would
-   drop them. `OVER ()` (empty key set) becomes `CROSS JOIN` against a
-   one-row table.
+   drop them. `OVER ()` (empty key set) joins its one-row table via
+   `LEFT JOIN … ON (1 = 1)` — never a CROSS join, because the oracle prints
+   CROSS joins in comma form, which re-parses with different associativity
+   once another join follows (fuzz-found).
 3. **Multiplicity by construction:** `fit_sql` is `GROUP BY keys` ⇒ keys are
    unique ⇒ LEFT JOIN matches ≤ 1 ⇒ exactly one row out per row in. Map shape
    is proven, not tested.
 4. **`__CF_` is a reserved prefix** (params tables `__CF_PARAMS_N__`, columns
    `__cf_aN`). Input SQL containing it (case-insensitive) is refused — same
    idiom as Confit's `__glob_pat` reservation.
-5. `__THIS__` gains the alias `t` in `serving_sql`; base-column references are
-   qualified through it.
+5. `__THIS__` gains the alias `__cf_t` in `serving_sql`; base-column
+   references are qualified through it (a short alias like `t` could collide
+   with a user column; the reserved prefix cannot). Unaliased non-column
+   select items get their derived name frozen as an explicit alias first —
+   DuckDB names such columns by their own printed text, which qualification
+   would change. Plain column refs are exempt: their name is the last path
+   part, untouched by qualification.
 
 ## Accepted surface and refusals
 
@@ -100,11 +135,13 @@ refusal raises a named, positioned error — serve-or-refuse, no third mode.
 `agg(expr…) OVER ()`, where:
 
 - `agg` is on the **allowlist** (grown one measured entry at a time, like the
-  corpus): `avg`, `sum`, `count`, `min`, `max`, `stddev`, `stddev_pop`,
-  `stddev_samp`, `var_pop`, `var_samp`, `variance`, `median`. Order-sensitive
-  aggregates (`first`, `last`, `string_agg`, `array_agg`, …) are refused by
+  corpus): `avg`, `sum`, `count`, `count_star`, `min`, `max`, `stddev`,
+  `stddev_pop`, `stddev_samp`, `var_pop`, `var_samp`, `variance`, `median`.
+  Order-sensitive aggregates (`string_agg`, `array_agg`, …) are refused by
   absence: their per-group value is nondeterministic, which would make both
-  fit and the differential gate flaky.
+  fit and the differential gate flaky. (`first`/`last` never reach the
+  allowlist check — the oracle classifies them as their own window types, so
+  they are refused as position-dependent window functions.)
 - Partition keys are plain `__THIS__` columns (expressions: next loop).
 - Aggregate arguments are arbitrary expressions over `__THIS__` columns —
   computed training-side, no restriction beyond what DuckDB accepts.
@@ -148,14 +185,23 @@ class SQLProjection:
     def boundary(self) -> str: ...
 ```
 
-Internally, `marginalize(sql) -> Marginalized(serving_sql, params_specs)` is a
-pure function (parse + rewrite + plan, no execution); `fit` = marginalize +
-materialize. `this_model` is accepted for signature stability but unused this
-slice (the training table brings its own schema).
+Internally, `marginalize(sql) -> Marginalized(serving_sql, windows_sql,
+params)` is a pure function (parse + rewrite + plan, no execution); `fit` =
+marginalize + materialize (two stages, see above). `this_model` is accepted
+for signature stability but unused this slice (the training table brings its
+own schema).
+
+The raw AST dicts are handled under a two-tier typing discipline: subtrees
+merely carried pass through as opaque `Node` dicts, while every node the
+module *interprets* is read through a pydantic view that validates shape at
+the read site — a DuckDB format change fails as one named "AST shape drift"
+error, not a `KeyError` mid-walk.
 
 ## The bulletproof gate
 
-DuckDB-vs-DuckDB differential — no inference code anywhere:
+DuckDB-vs-DuckDB differential — no inference code anywhere, both sides at
+`threads = 1` (the only setting where the oracle's own float window
+aggregation is bit-deterministic):
 
 ```python
 original  = duck(sql, __THIS__=train)
@@ -172,6 +218,13 @@ unicode/quoted identifiers. Plus:
 - **AST pins:** executed `json_serialize_sql` examples for every node shape the
   walker relies on, so a DuckDB bump that moves the format fails loudly.
 - Column-order and dtype equality in the differential, not just values.
+- **Seeded differential fuzz** (`MARGINALIZE_FUZZ_N`, default 25; deep runs at
+  500+): random typed tables with NULLs everywhere × random projections. The
+  fuzz found every deep bug in this slice: the CROSS-join comma-form
+  associativity trap, both float summation-order drifts, and the degenerate
+  `pa.null()`-typed-column coercion (an untyped training column is rejected
+  territory for a later loop — the generator now types its columns
+  explicitly).
 
 ## Out of scope (named next loops)
 

packages/sql-transform/README.md

@@ -1,29 +1,47 @@
 # sql-transform
 
-The authoring surface for SQL feature transforms — **not implemented**.
+`SQLProjection` — projections over `__THIS__`, fit once, serve row-at-a-time.
+
+The **fit half works today**: window aggregates over `__THIS__` are
+*marginalized* — computed once over training data, materialized into params
+tables, and the SQL rewritten to join them instead of recomputing:
 
 ```python
-from sql_transform import SQLTransform
+from sql_transform import SQLProjection
+
+p = SQLProjection(
+    "SELECT (age - avg(age) OVER (PARTITION BY country))"
+    " / stddev_samp(age) OVER (PARTITION BY country) AS age_z FROM __THIS__"
+).fit(train)   # train: pyarrow.Table
 
-t = SQLTransform(sql)   # NotImplementedError
+p.serving_sql
+# SELECT (__cf_t.age - __cf_p0.__cf_a0) / __cf_p0.__cf_a1 AS age_z
+# FROM __THIS__ AS __cf_t
+# LEFT JOIN __CF_PARAMS_0__ AS __cf_p0
+#   ON (__cf_t.country IS NOT DISTINCT FROM __cf_p0.country)
+p.params
+# {"__CF_PARAMS_0__": <pyarrow.Table: country, __cf_a0, __cf_a1>}
 ```
 
-Every method raises `NotImplementedError`. The signatures, defaults and return
-types are all that remain, and they are the contract a rebuild has to satisfy:
-
-| method | intended behaviour |
-|---|---|
-| `SQLTransform(sql)` | accept SQL (or a t-string) over `__THIS__` |
-| `from_file(path)` | same, read from a file |
-| `fit(table, this_model=None)` | freeze window-aggregate state into static tables, rewrite the SQL to reference them, specialize via Confit; returns self |
-| `infer(row)` / `infer_batch(rows)` | serve rows through the specialized function |
-| `backend` / `boundary` | report which engine and boundary the fitted function uses |
-
-The serving half already exists and is unaffected: [Confit](../confit) takes SQL
-plus frozen static tables and partially evaluates them into a native function,
-bit-exact with DuckDB or refusing at build time. What was removed here is the
-fit half, the DataFusion batch engine, the codegen backend, and transform
-composition.
-
-Tests in this package assert only that the surface exists and refuses honestly —
-they fail if a method is dropped, renamed, or quietly starts returning something.
+The serving half (`infer`/`infer_batch` through [Confit](../confit)) is a later
+loop and still raises `NotImplementedError`.
+
+## The contract
+
+Strict projection: `SELECT <exprs> FROM __THIS__`. Everything else — WHERE,
+GROUP BY, joins, subqueries, CTEs, running windows, order-sensitive
+aggregates — is refused at construction with a named error. Serve-or-refuse,
+no third mode.
+
+Parsing is DuckDB's own (`json_serialize_sql` / `json_deserialize_sql`): the
+oracle's grammar and the oracle's printer, no second SQL dialect anywhere.
+
+The correctness gate is differential: for every accepted projection, the
+original SQL over training data must be **bit-exact** with `serving_sql`
+joined against the fitted params — both executed by DuckDB (single-threaded:
+DuckDB's parallel float window aggregation is not bit-deterministic even
+against itself). NULL partition keys are one partition and join back via
+`IS NOT DISTINCT FROM`; join multiplicity is provable, not tested.
+
+See the design spec:
+[2026-07-29-sql-projection-marginalization-design.md](../../docs/superpowers/specs/2026-07-29-sql-projection-marginalization-design.md).

packages/sql-transform/pyproject.toml

@@ -1,12 +1,13 @@
 [project]
 name = "sql-transform"
 version = "0.1.0"
-description = "SQL feature transforms — authoring surface, being rebuilt on Confit"
+description = "SQL projections over __THIS__ — fit marginalizes aggregates via DuckDB, serving via Confit"
 readme = "README.md"
 requires-python = ">=3.14"
-# Only what the surface itself references. The fit-time engine (datafusion,
-# sqlglot, numpy) went with the implementation; add back what the rebuild needs.
+# duckdb is the fit-time engine and the parser (json_serialize_sql); serving
+# will come through confit and never needs it.
 dependencies = [
+    "duckdb>=1.5.5",
     "pyarrow>=19.0",
     "pydantic>=2.0,<3.0",
 ]

packages/sql-transform/sql_transform/__init__.py

@@ -1,15 +1,25 @@
-"""SQLTransform — the authoring surface, awaiting reimplementation.
+"""SQLProjection — projections over ``__THIS__``, fit once, serve row-at-a-time.
 
-Every method raises `NotImplementedError`. The signatures are the contract a
-rebuild has to satisfy; the previous implementation (a DataFusion batch engine,
-a codegen backend, and transform composition) was deleted deliberately.
-
-The serving half already exists and is unaffected: confit takes SQL plus static
-tables frozen at fit time and partially evaluates them into a native function.
+The fit half works today: ``fit(table)`` marginalizes every window aggregate
+into materialized params tables plus a rewritten ``serving_sql``. The serving
+half (``infer``/``infer_batch`` through Confit) is a later loop and raises
+``NotImplementedError``.
 """
 
 from __future__ import annotations
 
-from sql_transform._transform import SQLTransform
+from sql_transform._marginalize import (
+    Marginalized,
+    MarginalizeError,
+    ParamsSpec,
+    marginalize,
+)
+from sql_transform._projection import SQLProjection
 
-__all__ = ["SQLTransform"]
+__all__ = [
+    "Marginalized",
+    "MarginalizeError",
+    "ParamsSpec",
+    "SQLProjection",
+    "marginalize",
+]