Stage B part 2: self-joins — the batch as build side (TASK-59)

backlog/tasks/task-59 - Stage-B-join-multiplicity-under-shapemany-—-dup-key-cross-inequality-and-self-joins.md

@@ -3,10 +3,10 @@ id: TASK-59
 title: >-
   Stage B: join multiplicity under shape='many' — dup-key, cross/inequality, and
   self-joins
-status: In Progress
+status: Done
 assignee: []
 created_date: '2026-07-28 01:55'
-updated_date: '2026-07-28 01:57'
+updated_date: '2026-07-28 03:07'
 labels:
   - specializer
   - stage-b
@@ -29,12 +29,12 @@ Hard stop after shipping: columnar-API discussion with the user before any next
 
 ## Acceptance Criteria
 <!-- AC:BEGIN -->
-- [ ] #1 Pins spec for multiplicity semantics (emission order/determinism, LEFT-miss, cross/inequality, self-join star forms, NULL keys among dups) with raw JSONs
-- [ ] #2 All stage-B constructs REJECT under shape='filter'/'map' exactly as today; they build only under shape='many'
-- [ ] #3 1:N equi-joins, cross/inequality joins, and self-joins serve bit-exact (order-insensitive multiset vs DuckDB) under shape='many'
-- [ ] #4 Corpus replay strictly above 529 with zero FAILs (replay builds stage-B cases with shape='many')
-- [ ] #5 Limitations doc + twin updated: stage-B rows move from 'designed, not built' to the shape='many' contract
-- [ ] #6 Full gates green on release build; PR opened (stacked on #44)
+- [x] #1 Pins spec for multiplicity semantics (emission order/determinism, LEFT-miss, cross/inequality, self-join star forms, NULL keys among dups) with raw JSONs
+- [x] #2 All stage-B constructs REJECT under shape='filter'/'map' exactly as today; they build only under shape='many'
+- [x] #3 1:N equi-joins, cross/inequality joins, and self-joins serve bit-exact (order-insensitive multiset vs DuckDB) under shape='many'
+- [x] #4 Corpus replay strictly above 529 with zero FAILs (replay builds stage-B cases with shape='many')
+- [x] #5 Limitations doc + twin updated: stage-B rows move from 'designed, not built' to the shape='many' contract
+- [x] #6 Full gates green on release build; PR opened (stacked on #44)
 <!-- AC:END -->
 
 ## Implementation Plan
@@ -46,3 +46,13 @@ TODAY: both backends are strictly one-emit-per-row. Interpreter: per row, block
 
 DESIGN DIRECTION for multiplicity (validate in spec): (1) emission becomes an explicit operation instead of a terminal side effect — either Inst::EmitRow (appends the current out-lane values) with CTerm::Emit demoted to plain end-of-row, or loop-structured blocks using the EXISTING Brif/Jump machinery (loop header + body + back-edge) with an EmitRow inst in the body. Verify/canonicalize must accept back-edges (check: today's block graph is probably a DAG — verify.rs may assume forward-only; if so that is the first thing to extend). (2) New probe ops: ProbeStart {join} -> cursor, ProbeNext {join, cursor} -> (has: i1, cursor', value lanes) over per-key row LISTS in StaticTy::Map (values become Vec-of-rows per key — flat arena + (off, len) per key is the lean layout). (3) Cranelift: emission via a helper call h_emit_row(cx) that appends to st.out (helpers already write through Cx), loop via normal CLIF blocks — feasible without new architecture; the return code stays for trap/end only. If this turns out heavy, interpreter-first for shape='many' with cranelift fallback is the documented fallback plan (bench guard exempts 'many' from backend-identity tests). (4) Self-join: per-call build of the batch-side map before the row loop (new PreparedStatic variant built in run(), or a batch-map handed via RunState); marshaller unaffected. (5) Cross/inequality: a join with ZERO key columns = single bucket = ProbeStart/Next over the whole static (or batch) + residual predicate in the loop body. (6) LEFT: emit null-extension when the loop body emitted nothing for this row (a per-row 'any_emitted' flag register). (7) shape gate: prepare learns the target shape (prepare_opaque param or Prepared metadata + duckdb/mod.rs check): multiplicity constructs produce a NAMED reject unless shape='many'; corpus replay builds stage-B-shaped cases with shape='many' (detect by retrying on the named errors, or always pass shape='many'? NO — corpus must keep proving the default rejects; retry-with-many on the named needles is the honest harness).
 <!-- SECTION:PLAN:END -->
+
+## Final Summary
+
+<!-- SECTION:FINAL_SUMMARY:BEGIN -->
+Stage B shipped as two stacked PRs: #45 (part 1: dup-key equi-joins, cross/inequality/constant-ON joins) and #46 (part 2: self-joins via the batch as build side). Corpus 529 -> 550 match / 0 FAIL of 678 (81%); the 27-case stage-B pool is fully resolved (21 served + 3 USING-self-join named rejections + 3 rowid self-joins routed to the documented rowid descope).
+
+Everything builds ONLY under shape='many' (TASK-58's fence): default shapes are byte-identical, dup keys and self-joins still reject. Machinery: Term::EmitTo emit-and-continue back-edges (verifier now allows terminating cycles only), StaticTy::MultiMap (stable-sorted equal-key runs; insertion order = the engine's documented 1:N emission order) + ProbeRange/ProbeRead, StaticTy::BatchMap (per-call batch flattening for self-joins, whole ON as per-pair residual — cross-then-filter, pins-proved identical), loop-structured lowering riding state on the live stack with per-block probe-cache reseeds (block splits from CASE machinery), residual-vs-WHERE gate split (residual decides match-ness and the LEFT null-extension; WHERE only gates emission). Interpreter-only; cranelift pre-rejects into the existing fallback.
+
+Central pins finding: DuckDB's join output ORDER is a hash-join accident (LIFO chains, lockstep vector passes, run-to-run divergence at scale) — parity is MULTISET (corpus sorts; duck oracle tests sort), and the engine's own deterministic order (probe outer, insertion inner, null-extension in place) is pinned as a test. Pins: 2026-07-28-stageB-multiplicity-pins.md + pins-stageB/*.json (5 agents, 57 pins). Follow-up left open: USING/NATURAL self-joins (named rejection).
+<!-- SECTION:FINAL_SUMMARY:END -->

docs/known-limitations.md

@@ -9,7 +9,7 @@ forces this document to change with it.
 **The contract.** For any SQL you hand it, the engine does exactly one of:
 
 1. **Serve it bit-for-bit identical to DuckDB** (verified continuously
-   against DuckDB's own test corpus: 546 of 678 statements as of stage B), or
+   against DuckDB's own test corpus: 550 of 678 statements as of stage B), or
 2. **Refuse loudly at BUILD time** — `DuckDBInferFn(...)` raises a
    `ValueError` naming the construct. Nothing is ever silently wrong or
    silently dropped at inference time.
@@ -31,7 +31,7 @@ rejected, permanently by design:
 | Regex patterns must be constants (`regexp_matches(s, pattern_col)` rejects) | `unsupported: non-constant regex pattern (compiled at prepare in v0)` | Regexes compile at prepare; DuckDB compiles per row. Per-row compilation is the opposite of specialization. |
 | Replacement strings / regex options / extract group indexes must be constants | `non-constant regexp_replace replacement` etc. | Same. |
 | Static (join) tables must be provided at build time; under the DEFAULT shapes their keys must be unique | `duplicate map key` | Joins are frozen hash maps baked into the function. Duplicate keys mean 1:N join multiplicity, which SERVES under the opt-in `shape='many'` (TASK-59) — along with cross joins, inequality `ON` predicates, and constant `ON` clauses — with multiset parity vs DuckDB (its join output order is a measured hash-join accident; the engine emits probe order outer, build insertion order inner). Under `filter`/`map` the 1:1 contract is unchanged. NULL *values* serve since TASK-55; NULL *keys* never match. |
-| The dynamic table cannot be joined to itself | `joining the dynamic table to itself` | The batch is the probe side; using it as a build side too is the one stage-B piece still in progress (it will also require `shape='many'`). |
+| Self-joins require `shape='many'` (and the ON form) | `joining the dynamic table to itself` | Under `'many'` the batch itself becomes the build side (assembled per call, the ON as a per-pair residual) — comma/cross and `ON` self-joins serve with multiset parity. `USING`/`NATURAL` self-joins stay a named rejection (follow-up); the default shapes keep the original error. |
 | Exactly one row table drives the query | `the specializer takes exactly one row table`, `must be the dynamic table` | The serving contract is rows-in → rows-out for one entity stream. |
 
 ## 2. Out of scope for row-serving (by decision, not difficulty)
@@ -45,8 +45,9 @@ clause, an INNER join (key misses drop), a static-tables-only constant
 query; `"many"` (0..N) is the multiplicity opt-in (stage B): duplicate-key
 joins, cross joins, and inequality/constant `ON` joins build ONLY under
 it (one join per query for now, named restriction) — multiplicity can
-never sneak into a serving path by default. Self-joins are still
-rejected under every shape (in progress).
+never sneak into a serving path by default. Comma and `ON` self-joins
+serve under `'many'` too; `USING`/`NATURAL` self-joins are a named
+follow-up rejection.
 
 The engine serves **row-at-a-time feature transforms**. Whole-relation
 constructs are out of scope because their output shape is not

src/duckdb/mod.rs

@@ -715,6 +715,11 @@ impl DuckDBInferFn {
         // Program statics and StaticSpecs are both indexed by join id.
         let mut data = Vec::with_capacity(prepared.statics.len());
         for (spec, sty) in prepared.statics.iter().zip(&prepared.program.statics) {
+            if spec.batch {
+                // Stage-B self-join: built per call by the executor.
+                data.push(StaticData::Map(Vec::new()));
+                continue;
+            }
             let (StaticTy::Map { keys, values } | StaticTy::MultiMap { keys, values }) =
                 sty
             else {
@@ -740,6 +745,11 @@ impl DuckDBInferFn {
                 Err(_) => {
                     let mut data = Vec::with_capacity(prepared.statics.len());
                     for (spec, sty) in prepared.statics.iter().zip(&prepared.program.statics) {
+                        if spec.batch {
+                            // Stage-B self-join: built per call by the executor.
+                            data.push(StaticData::Map(Vec::new()));
+                            continue;
+                        }
                         let (StaticTy::Map { keys, values } | StaticTy::MultiMap { keys, values }) =
                             sty
                         else {

src/specializer/exec/cranelift.rs

@@ -957,7 +957,7 @@ pub fn compile(p: &Program, statics: Vec<super::StaticData>) -> Result<Cranelift
     let has_multiplicity = p
         .statics
         .iter()
-        .any(|s| matches!(s, super::super::ir::StaticTy::MultiMap { .. }))
+        .any(|s| matches!(s, super::super::ir::StaticTy::MultiMap { .. } | super::super::ir::StaticTy::BatchMap { .. }))
         || p.blocks.iter().any(|b| {
             matches!(b.term, Term::EmitTo { .. })
                 || b.insts

src/specializer/exec/interp.rs

@@ -90,6 +90,10 @@ struct Ctx<'a> {
     out: &'a mut [OutCol],
     input: &'a Batch,
     statics: &'a [PreparedStatic],
+    /// Stage-B self-join: the batch's rows flattened like multimap values
+    /// (nullable -> validity+payload). Empty unless the program declares a
+    /// batchmap static.
+    batch_rows: &'a [Vec<ScalarVal>],
     row: usize,
 }
 
@@ -108,6 +112,9 @@ pub(super) enum PreparedStatic {
     MultiMap {
         entries: Vec<(Vec<KeyBits>, Vec<ScalarVal>)>,
     },
+    /// Stage-B self-join: the rows come from the BATCH, flattened per call
+    /// in `run` (see `build_batch_rows`) — nothing is prepared here.
+    BatchMap,
 }
 
 struct CBlock {
@@ -147,6 +154,9 @@ pub struct InterpFn {
     statics: Vec<PreparedStatic>,
     in_decl: Vec<(Ty, bool)>,
     out_decl: Vec<Ty>,
+    /// True when a batchmap static exists: `run` flattens the batch's
+    /// rows before the row loop (stage-B self-joins).
+    has_batch_map: bool,
 }
 
 pub fn compile(p: &Program, statics: Vec<StaticData>) -> Result<InterpFn, CompileError> {
@@ -190,6 +200,7 @@ pub fn compile(p: &Program, statics: Vec<StaticData>) -> Result<InterpFn, Compil
         statics: prepared,
         in_decl: p.in_cols.iter().map(|c| (c.ty.ty, c.ty.nullable)).collect(),
         out_decl: p.out_cols.iter().map(|c| c.ty.ty).collect(),
+        has_batch_map: p.statics.iter().any(|s| matches!(s, StaticTy::BatchMap { .. })),
     })
 }
 
@@ -225,6 +236,13 @@ impl InterpFn {
         }
         reserve_out(&mut st.out, input.rows);
 
+        // Stage-B self-joins: flatten the batch ONCE per call into the
+        // multimap-value layout (nullable -> validity+payload).
+        let batch_rows: Vec<Vec<ScalarVal>> = if self.has_batch_map {
+            build_batch_rows(input, &self.in_decl)
+        } else {
+            Vec::new()
+        };
         let mut emitted = 0usize;
         for row in 0..input.rows {
             let mut ctx = Ctx {
@@ -233,6 +251,7 @@ impl InterpFn {
                 out: &mut st.out,
                 input,
                 statics: &self.statics,
+                batch_rows: &batch_rows,
                 row,
             };
             let mut bi = 0usize;
@@ -375,6 +394,45 @@ pub(super) fn reserve_out(out: &mut [OutCol], rows: usize) {
     }
 }
 
+/// Flatten the batch into multimap-value rows for a batchmap static:
+/// per input row, each column contributes payload (non-nullable) or
+/// validity + payload (nullable), in declaration order — the same layout
+/// the lowering declared for the batchmap's values.
+fn build_batch_rows(input: &Batch, in_decl: &[(Ty, bool)]) -> Vec<Vec<ScalarVal>> {
+    let mut rows = Vec::with_capacity(input.rows);
+    for r in 0..input.rows {
+        let mut vals = Vec::new();
+        for (ci, &(ty, nullable)) in in_decl.iter().enumerate() {
+            let c = &input.cols[ci];
+            let valid = col_valid(c, r);
+            if nullable {
+                vals.push(ScalarVal::I1(valid));
+            }
+            let v = if valid {
+                match c {
+                    ColData::I1 { data, .. } => ScalarVal::I1(data[r]),
+                    ColData::I64 { data, .. } => ScalarVal::I64(data[r]),
+                    ColData::F64 { data, .. } => ScalarVal::F64(data[r]),
+                    ColData::Str { buf, spans, .. } => {
+                        let sp = spans[r];
+                        ScalarVal::Str(buf[sp.off as usize..(sp.off + sp.len) as usize].to_string())
+                    }
+                }
+            } else {
+                match ty {
+                    Ty::I1 => ScalarVal::I1(false),
+                    Ty::I64 => ScalarVal::I64(0),
+                    Ty::F64 => ScalarVal::F64(0.0),
+                    Ty::Str => ScalarVal::Str(String::new()),
+                }
+            };
+            vals.push(v);
+        }
+        rows.push(vals);
+    }
+    rows
+}
+
 pub(super) fn prepare_statics(
     p: &Program,
     statics: Vec<StaticData>,
@@ -452,6 +510,19 @@ pub(super) fn prepare_statics(
                 entries.sort_by(|a, b| a.0.cmp(&b.0));
                 prepared.push(PreparedStatic::MultiMap { entries });
             }
+            (StaticTy::BatchMap { .. }, StaticData::Map(entries)) => {
+                if !entries.is_empty() {
+                    return Err(CompileError::Static(format!(
+                        "@{i}: batchmap takes no prepared entries"
+                    )));
+                }
+                prepared.push(PreparedStatic::BatchMap);
+            }
+            (StaticTy::BatchMap { .. }, StaticData::Scalar { .. }) => {
+                return Err(CompileError::Static(format!(
+                    "@{i}: declared batchmap, got scalar data"
+                )))
+            }
             (StaticTy::MultiMap { .. }, StaticData::Scalar { .. }) => {
                 return Err(CompileError::Static(format!(
                     "@{i}: declared multimap, got scalar data"
@@ -2223,19 +2294,25 @@ fn compile_inst(
             let static_id = static_id as usize;
             let (start, end) = (sl(slots, start), sl(slots, end));
             let keys: Vec<usize> = keys.iter().map(|k| sl(slots, *k)).collect();
+            let is_batch = matches!(&p.statics[static_id], StaticTy::BatchMap { .. });
             Box::new(move |ctx| {
-                let PreparedStatic::MultiMap { entries } = &ctx.statics[static_id] else {
-                    unreachable!("static kind checked at compile");
-                };
-                let (lo, hi) = if keys.is_empty() {
-                    (0, entries.len())
+                let (lo, hi) = if is_batch {
+                    (0, ctx.batch_rows.len())
                 } else {
-                    let lo = entries
-                        .partition_point(|(k, _)| cmp_key(k, &keys, ctx) == std::cmp::Ordering::Less);
-                    let hi = entries.partition_point(|(k, _)| {
-                        cmp_key(k, &keys, ctx) != std::cmp::Ordering::Greater
-                    });
-                    (lo, hi)
+                    let PreparedStatic::MultiMap { entries } = &ctx.statics[static_id] else {
+                        unreachable!("static kind checked at compile");
+                    };
+                    if keys.is_empty() {
+                        (0, entries.len())
+                    } else {
+                        let lo = entries.partition_point(|(k, _)| {
+                            cmp_key(k, &keys, ctx) == std::cmp::Ordering::Less
+                        });
+                        let hi = entries.partition_point(|(k, _)| {
+                            cmp_key(k, &keys, ctx) != std::cmp::Ordering::Greater
+                        });
+                        (lo, hi)
+                    }
                 };
                 ctx.regs[start] = RegVal::I64(lo as i64);
                 ctx.regs[end] = RegVal::I64(hi as i64);
@@ -2250,11 +2327,21 @@ fn compile_inst(
             let static_id = static_id as usize;
             let idx = sl(slots, idx);
             let dsts: Vec<usize> = dsts.iter().map(|d| sl(slots, *d)).collect();
+            let is_batch = matches!(&p.statics[static_id], StaticTy::BatchMap { .. });
             Box::new(move |ctx| {
+                let i = as_i64(ctx.regs[idx]) as usize;
+                if is_batch {
+                    let Some(row) = ctx.batch_rows.get(i) else {
+                        return Err(Trap("probe.read index out of range (lowering bug)".into()));
+                    };
+                    for (di, v) in dsts.iter().zip(row.iter()) {
+                        ctx.regs[*di] = scalar_to_reg(v, ctx.arena);
+                    }
+                    return Ok(());
+                }
                 let PreparedStatic::MultiMap { entries } = &ctx.statics[static_id] else {
                     unreachable!("static kind checked at compile");
                 };
-                let i = as_i64(ctx.regs[idx]) as usize;
                 let Some(row) = entries.get(i) else {
                     return Err(Trap("probe.read index out of range (lowering bug)".into()));
                 };

src/specializer/exec/tests.rs

@@ -818,7 +818,7 @@ fn gen_statics(rng: &mut gen::Rng, p: &Program) -> Vec<StaticData> {
                 StaticData::Map(entries)
             }
             // The generator never declares multimaps (see gen.rs).
-            StaticTy::MultiMap { .. } => StaticData::Map(Vec::new()),
+            StaticTy::MultiMap { .. } | StaticTy::BatchMap { .. } => StaticData::Map(Vec::new()),
         })
         .collect()
 }