feat(specializer): row-shape contract flag — shape='map' | 'filter' | 'many' (TASK-58)

User-requested serving fence ahead of stage B: shape='map' statically
PROVES exactly one output row per input row (out[i] <-> in[i]) at build
time — rejects WHERE, INNER joins (key misses drop), and static-only
constant queries, each named; LEFT joins and scalar projections serve
(unique keys are already the map contract). shape='filter' (default) is
byte-identical to before; shape='many' is reserved and will be the ONLY
shape under which stage-B multiplicity constructs build. Introspectable
via fn.shape. Doc note in known-limitations §2.

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

Author: 2 people

backlog/tasks/task-58 - Row-shape-contract-flag-shape-map-filter-many-on-DuckDBInferFn.md

@@ -0,0 +1,33 @@
+---
+id: TASK-58
+title: 'Row-shape contract flag: shape="map" | "filter" | "many" on DuckDBInferFn'
+status: In Progress
+assignee: []
+created_date: '2026-07-28 01:50'
+labels:
+  - specializer
+  - api
+dependencies: []
+type: feature
+ordinal: 52000
+---
+
+## Description
+
+<!-- SECTION:DESCRIPTION:BEGIN -->
+User-requested (2026-07-28 morning) safety fence ahead of stage-B: serving paths need a BUILD-TIME guarantee that a query is a true projection (exactly one output row per input row, out[i] <-> in[i]).
+
+API: DuckDBInferFn(..., shape="filter") default = today's 0..1 behavior, unchanged. shape="map" = static proof of exactly-1: rejects WHERE (can drop), INNER JOIN (key miss drops), the static-only constant path (output unrelated to input rows), and every stage-B form; allows scalar exprs + LEFT JOIN (unique keys are already enforced). shape="many" = reserved now (named rejection pointing at stage-B), becomes the ONLY way multiplicity constructs build once stage-B lands.
+
+The proof is static (query shape), not a runtime row-count assertion. Rejection messages name the blocking construct ("shape='map': WHERE clause can drop rows").
+<!-- SECTION:DESCRIPTION:END -->
+
+## Acceptance Criteria
+<!-- AC:BEGIN -->
+- [ ] #1 shape param accepted with 'filter' default; invalid values are named ValueErrors
+- [ ] #2 shape='map' statically rejects WHERE, INNER JOIN, and constant-path queries with messages naming the construct; serves scalar projections and LEFT JOINs
+- [ ] #3 shape='many' is a named reserved rejection until stage-B
+- [ ] #4 Default behavior byte-identical to today (full gate green, corpus 529 unchanged)
+- [ ] #5 Tests cover accept/reject matrix; docs note in known-limitations SS2
+- [ ] #6 PR opened
+<!-- AC:END -->

docs/known-limitations.md

@@ -36,6 +36,17 @@ rejected, permanently by design:
 
 ## 2. Out of scope for row-serving (by decision, not difficulty)
 
+**The row-shape contract** (TASK-58): `DuckDBInferFn(..., shape=...)`
+declares how many output rows each input row may produce, checked at
+build time. `"filter"` (the default) is the engine's native 0..1;
+`"map"` statically PROVES exactly-one (`out[i] ↔ in[i]`, the strict
+serving guarantee) by rejecting anything that can drop a row — a WHERE
+clause, an INNER join (key misses drop), a static-tables-only constant
+query; `"many"` (0..N) is reserved for join multiplicity (stage B) and
+will be the only shape under which duplicate-key joins, cross/inequality
+joins, and self-joins ever build — multiplicity can never sneak into a
+serving path by default.
+
 The engine serves **row-at-a-time feature transforms**. Whole-relation
 constructs are out of scope because their output shape is not
 one-row-in/one-row-out:

src/duckdb/mod.rs

@@ -495,20 +495,41 @@ pub struct DuckDBInferFn {
     #[pyo3(get)]
     output_model: Py<PyAny>,
     output_dicts: bool,
+    shape_map: bool,
 }
 
 #[pymethods]
 impl DuckDBInferFn {
     #[new]
-    #[pyo3(signature = (sql, row_tables, static_tables, output_model=None, output=None))]
+    #[pyo3(signature = (sql, row_tables, static_tables, output_model=None, output=None, shape=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>,
+        shape: Option<String>,
     ) -> PyResult<Self> {
+        // The row-shape contract (TASK-58): "filter" (default) is today's
+        // 0..1 rows out per row in; "map" statically PROVES exactly-one
+        // (out[i] <-> in[i]) or refuses at build; "many" is reserved for
+        // join multiplicity (stage B) and is the only shape under which
+        // those constructs will ever build.
+        let strict_map = match shape.as_deref() {
+            None | Some("filter") => false,
+            Some("map") => true,
+            Some("many") => {
+                return Err(pyo3::exceptions::PyValueError::new_err(
+                    "shape='many' is reserved for join multiplicity (stage B — not built yet)",
+                ))
+            }
+            Some(other) => {
+                return Err(pyo3::exceptions::PyValueError::new_err(format!(
+                    "shape must be 'map', 'filter', or 'many', got '{other}'"
+                )))
+            }
+        };
         // output="dict" is the opt-in raw-dict mode: same engine, same
         // lanes, the marshaller just skips model construction. The typed
         // default is untouched.
@@ -653,6 +674,14 @@ impl DuckDBInferFn {
             Err(e @ (PrepareError::Unsupported(_) | PrepareError::Parse(_))) => {
                 match eval_static_only(py, &sql, &static_tables) {
                     Ok((rows, fields)) => {
+                        if strict_map {
+                            // Fixed rows regardless of input — the exact
+                            // opposite of out[i] <-> in[i].
+                            return Err(pyo3::exceptions::PyValueError::new_err(
+                                "shape='map': a static-tables-only query emits fixed \
+                                 rows unrelated to the input rows",
+                            ));
+                        }
                         let output_model = match output_model {
                             Some(m) => m,
                             None => model_from_fields(py, fields)?,
@@ -662,13 +691,21 @@ impl DuckDBInferFn {
                             row_table,
                             output_model,
                             output_dicts,
+                            shape_map: false,
                         });
                     }
                     Err(_) => return Err(build_err(e.to_string())),
                 }
             }
             Err(e) => return Err(build_err(e.to_string())),
         };
+        if strict_map {
+            if let Some(blocker) = &prepared.one_row_blocker {
+                return Err(pyo3::exceptions::PyValueError::new_err(format!(
+                    "shape='map': {blocker}"
+                )));
+            }
+        }
 
         // Program statics and StaticSpecs are both indexed by join id.
         let mut data = Vec::with_capacity(prepared.statics.len());
@@ -742,9 +779,20 @@ impl DuckDBInferFn {
             row_table,
             output_model,
             output_dicts,
+            shape_map: strict_map,
         })
     }
 
+    /// The row-shape contract: "map" (proven exactly-1) or "filter" (0..1).
+    #[getter]
+    fn shape(&self) -> &'static str {
+        if self.shape_map {
+            "map"
+        } else {
+            "filter"
+        }
+    }
+
     /// The output mode: "model" (typed, default) or "dict" (opt-in).
     #[getter]
     fn output(&self) -> &'static str {

src/specializer/mod.rs

@@ -42,6 +42,12 @@ pub struct StaticSpec {
 pub struct Prepared {
     pub program: ir::Program,
     pub statics: Vec<StaticSpec>,
+    /// `None` when the query provably emits EXACTLY one output row per
+    /// input row (out[i] <-> in[i]); otherwise names the first construct
+    /// that can drop a row. The static proof behind `shape="map"`
+    /// (TASK-58): no WHERE, and every join is LEFT (unique keys are
+    /// already the map contract, so LEFT never drops or duplicates).
+    pub one_row_blocker: Option<String>,
 }
 
 /// STAGE 1 for the v0 ribbon: SQL text + the dynamic table's name and schema
@@ -72,6 +78,7 @@ pub fn prepare_opaque(
 ) -> Result<Prepared, PrepareError> {
     let (rel, joins, out_cols, regexes) =
         frontend::frontend(sql, this_name, in_cols, opaque, structs, statics)?;
+    let one_row_blocker = one_row_blocker(&rel, &joins, statics);
     let mut program = lower::lower(&rel, &joins, statics, in_cols, out_cols, regexes, "run")?;
     // Block-splitting lowerings mint ids out of text order; renumber so
     // every prepared program is exactly canonical (parse(print(p)) == p).
@@ -110,5 +117,36 @@ pub fn prepare_opaque(
     Ok(Prepared {
         program,
         statics: specs,
+        one_row_blocker,
     })
 }
+
+/// The static exactly-one-row proof behind `shape="map"`: a Filter node or
+/// a non-LEFT join can drop input rows; everything else the engine serves
+/// is row-preserving (unique join keys are already the map contract, so a
+/// LEFT join never drops or duplicates).
+fn one_row_blocker(
+    rel: &plan::Rel,
+    joins: &[plan::JoinSpec],
+    statics: &[plan::StaticTable],
+) -> Option<String> {
+    fn has_filter(r: &plan::Rel) -> bool {
+        match r {
+            plan::Rel::Filter { .. } => true,
+            plan::Rel::Project { input, .. } => has_filter(input),
+            plan::Rel::Scan => false,
+        }
+    }
+    if has_filter(rel) {
+        return Some("a WHERE clause can drop rows".into());
+    }
+    for j in joins {
+        if j.kind != plan::JoinKind::Left {
+            return Some(format!(
+                "INNER JOIN '{}' drops rows on a key miss (use LEFT JOIN)",
+                statics[j.table].name
+            ));
+        }
+    }
+    None
+}

tests/test_shape_contract.py

@@ -0,0 +1,72 @@
+"""The row-shape contract flag (TASK-58): shape="map" | "filter" | "many".
+
+"map" is a STATIC build-time proof of exactly one output row per input row
+(out[i] <-> in[i]) — the serving-path guarantee. "filter" is the default
+0..1 behavior, byte-identical to before the flag existed. "many" is
+reserved for join multiplicity (stage B) and is the only shape under which
+those constructs will ever build.
+"""
+
+from __future__ import annotations
+
+import pyarrow as pa
+import pytest
+from pydantic import create_model
+
+from sql_transform._interpreter import DuckDBInferFn
+
+T = create_model("T", a=(int, ...), s=(str | None, None))
+DIM = pa.table({"id": [1, 2], "v": [10, 20]})
+
+
+def build(sql, shape=None, statics=None):
+    kwargs = {"output": "dict"}
+    if shape is not None:
+        kwargs["shape"] = shape
+    return DuckDBInferFn(
+        sql, row_tables={"__THIS__": T}, static_tables=statics or {}, **kwargs
+    )
+
+
+def test_map_serves_projections_and_left_joins():
+    fn = build("SELECT a + 1 AS b, upper(s) AS u FROM __THIS__", shape="map")
+    assert fn.shape == "map"
+    got = fn.infer({"__THIS__": [T(a=1, s="x"), T(a=2, s=None)]})
+    assert [r["b"] for r in got] == [2, 3]  # exactly one out per in, in order
+
+    fn = build(
+        "SELECT a, v FROM __THIS__ LEFT JOIN d ON a = d.id",
+        shape="map",
+        statics={"d": DIM},
+    )
+    got = fn.infer({"__THIS__": [T(a=1), T(a=99)]})
+    assert [r["v"] for r in got] == [10, None]  # a miss maps, never drops
+
+
+def test_map_rejects_row_dropping_constructs():
+    with pytest.raises(ValueError, match="shape='map'.*WHERE"):
+        build("SELECT a FROM __THIS__ WHERE a > 0", shape="map")
+    with pytest.raises(ValueError, match="shape='map'.*INNER JOIN 'd'"):
+        build(
+            "SELECT a, v FROM __THIS__ JOIN d ON a = d.id",
+            shape="map",
+            statics={"d": DIM},
+        )
+    # A static-only query emits fixed rows unrelated to the input.
+    with pytest.raises(ValueError, match="shape='map'.*static-tables-only"):
+        build("SELECT max(id) FROM d", shape="map", statics={"d": DIM})
+
+
+def test_filter_default_unchanged():
+    for kwargs in [{}, {"shape": "filter"}]:
+        fn = build("SELECT a FROM __THIS__ WHERE a > 1", **kwargs)
+        assert fn.shape == "filter"
+        got = fn.infer({"__THIS__": [T(a=1), T(a=2)]})
+        assert [r["a"] for r in got] == [2]
+
+
+def test_many_is_reserved_and_bad_values_are_named():
+    with pytest.raises(ValueError, match="stage B"):
+        build("SELECT a FROM __THIS__", shape="many")
+    with pytest.raises(ValueError, match="must be 'map', 'filter', or 'many'"):
+        build("SELECT a FROM __THIS__", shape="projection")