diff --git "a/backlog/tasks/task-59 - Stage-B-join-multiplicity-under-shapemany-\342\200\224-dup-key-cross-inequality-and-self-joins.md" "b/backlog/tasks/task-59 - Stage-B-join-multiplicity-under-shapemany-\342\200\224-dup-key-cross-inequality-and-self-joins.md" index 38c7418..65231fb 100644 --- "a/backlog/tasks/task-59 - Stage-B-join-multiplicity-under-shapemany-\342\200\224-dup-key-cross-inequality-and-self-joins.md" +++ "b/backlog/tasks/task-59 - Stage-B-join-multiplicity-under-shapemany-\342\200\224-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 -- [ ] #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) ## 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). + +## Final Summary + + +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). + diff --git a/docs/known-limitations.md b/docs/known-limitations.md index 3282e5b..dd84706 100644 --- a/docs/known-limitations.md +++ b/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 diff --git a/src/duckdb/mod.rs b/src/duckdb/mod.rs index e8a5570..4136272 100644 --- a/src/duckdb/mod.rs +++ b/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 { diff --git a/src/specializer/exec/cranelift.rs b/src/specializer/exec/cranelift.rs index 7938431..fb4346a 100644 --- a/src/specializer/exec/cranelift.rs +++ b/src/specializer/exec/cranelift.rs @@ -957,7 +957,7 @@ pub fn compile(p: &Program, statics: Vec) -> Result { 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], row: usize, } @@ -108,6 +112,9 @@ pub(super) enum PreparedStatic { MultiMap { entries: Vec<(Vec, Vec)>, }, + /// 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, in_decl: Vec<(Ty, bool)>, out_decl: Vec, + /// 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) -> Result { @@ -190,6 +200,7 @@ pub fn compile(p: &Program, statics: Vec) -> Result validity+payload). + let batch_rows: Vec> = 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> { + 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, @@ -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 = 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 = 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())); }; diff --git a/src/specializer/exec/tests.rs b/src/specializer/exec/tests.rs index 94c8f6a..8d57f21 100644 --- a/src/specializer/exec/tests.rs +++ b/src/specializer/exec/tests.rs @@ -818,7 +818,7 @@ fn gen_statics(rng: &mut gen::Rng, p: &Program) -> Vec { 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() } diff --git a/src/specializer/frontend.rs b/src/specializer/frontend.rs index 58317ee..7b63d2a 100644 --- a/src/specializer/frontend.rs +++ b/src/specializer/frontend.rs @@ -69,6 +69,7 @@ pub fn frontend( opaque: &[(usize, String)], structs: &[super::plan::StructCol], statics: &[StaticTable], + many: bool, ) -> Result<(Rel, Vec, Vec, Vec), PrepareError> { // GenericDialect, not DuckDbDialect: measured as a strict superset for // the forms we serve (adds ^@, * ILIKE, * RENAME) and matches the oracle @@ -131,7 +132,7 @@ pub fn frontend( } let (binder, joins, leftover_where) = - bind_from(select, this_name, in_cols, opaque, structs, statics)?; + bind_from(select, this_name, in_cols, opaque, structs, statics, many)?; let mut out_cols = Vec::new(); let mut exprs = Vec::new(); @@ -252,6 +253,7 @@ fn bind_from<'a>( opaque: &'a [(usize, String)], structs: &'a [super::plan::StructCol], statics: &'a [StaticTable], + many: bool, ) -> Result<(Binder<'a>, Vec, Option), PrepareError> { // Plain scalar columns occupy in_cols[..n_plain]; struct leaf lanes // follow and are addressable ONLY through their struct paths. @@ -363,7 +365,64 @@ fn bind_from<'a>( other => return Err(unsup(format!("JOIN {other}"))), }; if raw_name.eq_ignore_ascii_case(this_name) { - return Err(unsup("joining the dynamic table to itself")); + if !many { + return Err(unsup("joining the dynamic table to itself")); + } + if !opaque.is_empty() || !structs.is_empty() { + return Err(unsup( + "self-join over a row model with non-scalar columns", + )); + } + if binder.this_name.eq_ignore_ascii_case(&scope_name) + || binder + .joins + .iter() + .any(|j| j.name.eq_ignore_ascii_case(&scope_name)) + { + return Err(PrepareError::Bind(format!( + "duplicate table name '{scope_name}' in FROM" + ))); + } + // Stage-B self-join: the build side is the BATCH — a keyless + // batchmap (built per call) with the WHOLE ON as residual. + let on = match constraint { + JoinConstraint::On(e) => Some(e), + JoinConstraint::Using(_) | JoinConstraint::Natural => { + return Err(unsup( + "self-join USING/NATURAL (stage-B follow-up; use ON)", + )) + } + JoinConstraint::None => { + return Err(unsup("JOIN without ON (cross join)")) + } + }; + let n_batch = binder.n_plain as u32; + binder.joins.push(ScopeJoin { + name: scope_name.clone(), + table: std::borrow::Cow::Owned(StaticTable { + name: scope_name, + cols: in_cols[..binder.n_plain].to_vec(), + }), + kind, + key_cols: Vec::new(), + val_cols: (0..n_batch).collect(), + keys: Vec::new(), + using: false, + }); + let residual = match on { + None => None, + Some(e) => Some(fold(bool_context(binder.expr(e)?, "JOIN condition")?)), + }; + specs.push(JoinSpec { + table: 0, + batch: true, + kind, + keys: Vec::new(), + key_cols: Vec::new(), + val_cols: (0..n_batch).collect(), + residual, + }); + continue; } let table_idx = resolve_static(statics, &raw_name)?; if binder.this_name.eq_ignore_ascii_case(&scope_name) @@ -448,7 +507,7 @@ fn bind_from<'a>( binder.joins.push(ScopeJoin { name: scope_name, - table: st, + table: std::borrow::Cow::Borrowed(st), kind, key_cols: key_cols.clone(), val_cols: val_cols.clone(), @@ -460,6 +519,7 @@ fn bind_from<'a>( let residual = bind_residual(&binder, j, &residual_raw)?; specs.push(JoinSpec { table: table_idx, + batch: false, kind, keys, key_cols, @@ -497,7 +557,50 @@ fn bind_from<'a>( other => return Err(unsup(format!("FROM {other}"))), }; if raw_name.eq_ignore_ascii_case(this_name) { - return Err(unsup("joining the dynamic table to itself")); + if !many { + return Err(unsup("joining the dynamic table to itself")); + } + if !opaque.is_empty() || !structs.is_empty() { + return Err(unsup( + "self-join over a row model with non-scalar columns", + )); + } + if binder.this_name.eq_ignore_ascii_case(&scope_name) + || binder + .joins + .iter() + .any(|j| j.name.eq_ignore_ascii_case(&scope_name)) + { + return Err(PrepareError::Bind(format!( + "duplicate table name '{scope_name}' in FROM" + ))); + } + // Comma self-join = pure cross against the batch; equi + // conjuncts stay in WHERE (cross-then-filter is bit-identical + // under multiplicity — measured, pins-stageB). + let n_batch = binder.n_plain as u32; + binder.joins.push(ScopeJoin { + name: scope_name.clone(), + table: std::borrow::Cow::Owned(StaticTable { + name: scope_name, + cols: in_cols[..binder.n_plain].to_vec(), + }), + kind: JoinKind::Inner, + key_cols: Vec::new(), + val_cols: (0..n_batch).collect(), + keys: Vec::new(), + using: false, + }); + specs.push(JoinSpec { + table: 0, + batch: true, + kind: JoinKind::Inner, + keys: Vec::new(), + key_cols: Vec::new(), + val_cols: (0..n_batch).collect(), + residual: None, + }); + continue; } // Unresolvable comma tables (schema-qualified names, table // functions we didn't get as statics) stay CLEAN. @@ -561,7 +664,7 @@ fn bind_from<'a>( .collect(); binder.joins.push(ScopeJoin { name: scope_name, - table: st, + table: std::borrow::Cow::Borrowed(st), kind: JoinKind::Inner, key_cols: key_cols.clone(), val_cols: val_cols.clone(), @@ -570,6 +673,7 @@ fn bind_from<'a>( }); specs.push(JoinSpec { table: table_idx, + batch: false, kind: JoinKind::Inner, keys, key_cols, @@ -868,7 +972,7 @@ fn default_name(e: &SqlExpr) -> String { /// dynamic side: `r.id` ≡ CASE match THEN dyn-key ELSE NULL — wave-4). struct ScopeJoin<'a> { name: String, - table: &'a StaticTable, + table: std::borrow::Cow<'a, StaticTable>, kind: JoinKind, key_cols: Vec, val_cols: Vec, @@ -2669,6 +2773,9 @@ impl Binder<'_> { sc.name ))); } + if hit.is_none() && name.eq_ignore_ascii_case("rowid") { + return Err(unsup("rowid pseudo-column")); + } let (i, c) = hit.ok_or_else(|| { PrepareError::Bind(format!("column '{name}' does not exist in '{table}'")) })?; @@ -2705,6 +2812,9 @@ impl Binder<'_> { return Ok(self.key_lane(j, kp)); } } + if name.eq_ignore_ascii_case("rowid") { + return Err(unsup("rowid pseudo-column")); + } return Err(PrepareError::Bind(format!( "column '{name}' does not exist in '{table}'" ))); diff --git a/src/specializer/ir/gen.rs b/src/specializer/ir/gen.rs index d31d509..821859d 100644 --- a/src/specializer/ir/gen.rs +++ b/src/specializer/ir/gen.rs @@ -216,7 +216,7 @@ fn load_all( } // The random-program generator doesn't emit multiplicity loops // (stage-B programs are exercised by targeted tests instead). - StaticTy::MultiMap { .. } => { + StaticTy::MultiMap { .. } | StaticTy::BatchMap { .. } => { } } } diff --git a/src/specializer/ir/mod.rs b/src/specializer/ir/mod.rs index f75d9b4..68f3ea8 100644 --- a/src/specializer/ir/mod.rs +++ b/src/specializer/ir/mod.rs @@ -183,6 +183,11 @@ pub enum StaticTy { /// flat row range (ProbeRange -> [start, end), ProbeRead per index). /// Zero keys = the keyless one-bucket join (cross/inequality). MultiMap { keys: Vec, values: Vec }, + /// Stage-B self-join: the BATCH as build side, assembled per call by + /// the executor (always keyless; the ON is the join's residual). + /// `values` = the batch's columns flattened like multimap values + /// (nullable -> validity+payload pairs). + BatchMap { values: Vec }, } /// SSA value id. Presentation names are not stored; the printer derives diff --git a/src/specializer/ir/parse.rs b/src/specializer/ir/parse.rs index f708f0a..6c2d3a7 100644 --- a/src/specializer/ir/parse.rs +++ b/src/specializer/ir/parse.rs @@ -669,6 +669,19 @@ impl Parser { self.expect(Tok::RParen)?; Ok(StaticTy::MultiMap { keys, values }) } + "batchmap" => { + self.expect(Tok::LParen)?; + self.expect(Tok::RParen)?; + self.expect(Tok::Arrow)?; + self.expect(Tok::LParen)?; + let values = if *self.peek() == Tok::RParen { + Vec::new() + } else { + self.ty_list()? + }; + self.expect(Tok::RParen)?; + Ok(StaticTy::BatchMap { values }) + } other => Err(self.err(format!("expected 'scalar' or 'map', found '{other}'"))), } } diff --git a/src/specializer/ir/print.rs b/src/specializer/ir/print.rs index 217a005..61932f1 100644 --- a/src/specializer/ir/print.rs +++ b/src/specializer/ir/print.rs @@ -22,6 +22,9 @@ pub fn print(p: &Program) -> String { StaticTy::MultiMap { keys, values } => { let _ = writeln!(s, "multimap({}) -> ({})", tys(keys), tys(values)); } + StaticTy::BatchMap { values } => { + let _ = writeln!(s, "batchmap() -> ({})", tys(values)); + } } } for (i, re) in p.regexes.iter().enumerate() { diff --git a/src/specializer/ir/verify.rs b/src/specializer/ir/verify.rs index b03c976..f515bad 100644 --- a/src/specializer/ir/verify.rs +++ b/src/specializer/ir/verify.rs @@ -228,7 +228,10 @@ fn dst_types(p: &Program, inst: &Inst) -> Vec<(Value, Ty)> { static_id, dsts, .. } => { let mut v = Vec::new(); - if let Some(StaticTy::MultiMap { values, .. }) = p.statics.get(*static_id as usize) { + if let Some( + StaticTy::MultiMap { values, .. } | StaticTy::BatchMap { values }, + ) = p.statics.get(*static_id as usize) + { for (d, ty) in dsts.iter().zip(values.iter()) { v.push((*d, *ty)); } @@ -578,7 +581,7 @@ fn check_block( ); } } - Some(StaticTy::MultiMap { .. }) => err( + Some(StaticTy::MultiMap { .. } | StaticTy::BatchMap { .. }) => err( errs, Some(bi), i, @@ -589,6 +592,16 @@ fn check_block( static_id, keys, .. } => match p.statics.get(*static_id as usize) { None => err(errs, Some(bi), i, format!("unknown static @{static_id}")), + Some(StaticTy::BatchMap { .. }) => { + if !keys.is_empty() { + err( + errs, + Some(bi), + i, + format!("@{static_id} is a batchmap: probe.range takes no keys"), + ); + } + } Some(StaticTy::MultiMap { keys: kts, .. }) => { if keys.len() != kts.len() { err( @@ -620,7 +633,7 @@ fn check_block( dsts, } => match p.statics.get(*static_id as usize) { None => err(errs, Some(bi), i, format!("unknown static @{static_id}")), - Some(StaticTy::MultiMap { values: vts, .. }) => { + Some(StaticTy::MultiMap { values: vts, .. } | StaticTy::BatchMap { values: vts }) => { want(&in_scope, def_types, *idx, Ty::I64, "probe index", bi, i, errs); if dsts.len() != vts.len() { err( @@ -644,7 +657,7 @@ fn check_block( }, Inst::Sload { static_id, .. } => match p.statics.get(*static_id as usize) { None => err(errs, Some(bi), i, format!("unknown static @{static_id}")), - Some(StaticTy::Map { .. }) | Some(StaticTy::MultiMap { .. }) => err( + Some(StaticTy::Map { .. } | StaticTy::MultiMap { .. } | StaticTy::BatchMap { .. }) => err( errs, Some(bi), i, @@ -660,7 +673,7 @@ fn check_block( }, Inst::SloadOpt { static_id, .. } => match p.statics.get(*static_id as usize) { None => err(errs, Some(bi), i, format!("unknown static @{static_id}")), - Some(StaticTy::Map { .. }) | Some(StaticTy::MultiMap { .. }) => err( + Some(StaticTy::Map { .. } | StaticTy::MultiMap { .. } | StaticTy::BatchMap { .. }) => err( errs, Some(bi), i, diff --git a/src/specializer/lower.rs b/src/specializer/lower.rs index 175f65d..c5d0a5b 100644 --- a/src/specializer/lower.rs +++ b/src/specializer/lower.rs @@ -74,20 +74,28 @@ pub fn lower( } if many && joins.len() == 1 { fb.lower_many_loop(exprs, filter_pred, &out_cols)?; - let statics = vec![StaticTy::MultiMap { - keys: joins[0].keys.iter().map(|k| k.ty).collect(), - values: joins[0] - .val_cols + let flat = |cols: &[Col], val_cols: &[u32]| -> Vec { + val_cols .iter() .flat_map(|&c| { - let ct = catalog[joins[0].table].cols[c as usize].ty; + let ct = cols[c as usize].ty; if ct.nullable { vec![Ty::I1, ct.ty] } else { vec![ct.ty] } }) - .collect(), + .collect() + }; + let statics = vec![if joins[0].batch { + StaticTy::BatchMap { + values: flat(in_cols, &joins[0].val_cols), + } + } else { + StaticTy::MultiMap { + keys: joins[0].keys.iter().map(|k| k.ty).collect(), + values: flat(&catalog[joins[0].table].cols, &joins[0].val_cols), + } }]; return fb.finish(name, statics, in_cols, out_cols, regexes); } @@ -1015,10 +1023,15 @@ impl<'a> FB<'a> { /// (TASK-55), mirroring the StaticTy::Map flattening. fn val_slots(&self, j: u32) -> Vec<(Option, usize)> { let spec = &self.joins[j as usize]; + let cols: &[Col] = if spec.batch { + self.in_cols + } else { + &self.catalog[spec.table].cols + }; let mut out = Vec::with_capacity(spec.val_cols.len()); let mut i = 0usize; for &c in &spec.val_cols { - if self.catalog[spec.table].cols[c as usize].ty.nullable { + if cols[c as usize].ty.nullable { out.push((Some(i), i + 1)); i += 2; } else { @@ -1032,10 +1045,15 @@ impl<'a> FB<'a> { /// Flattened probe-dst TYPES for join `j` (same order as val_slots). fn val_flat_tys(&self, j: u32) -> Vec { let spec = &self.joins[j as usize]; + let cols: &[Col] = if spec.batch { + self.in_cols + } else { + &self.catalog[spec.table].cols + }; spec.val_cols .iter() .flat_map(|&c| { - let ct = self.catalog[spec.table].cols[c as usize].ty; + let ct = cols[c as usize].ty; if ct.nullable { vec![Ty::I1, ct.ty] } else { diff --git a/src/specializer/mod.rs b/src/specializer/mod.rs index d427b79..cdd66e9 100644 --- a/src/specializer/mod.rs +++ b/src/specializer/mod.rs @@ -28,6 +28,9 @@ pub use frontend::PrepareError; /// becomes f64 here), valued by `val_cols`. Rows with a NULL key are dropped /// (a NULL never equi-matches); a NULL in a value column is an error. pub struct StaticSpec { + /// Stage-B self-join: no materialization — the build side is the + /// BATCH, assembled per call by the executor. + pub batch: bool, pub table: String, pub key_cols: Vec, pub val_cols: Vec, @@ -78,7 +81,7 @@ pub fn prepare_opaque( many: bool, ) -> Result { let (rel, joins, out_cols, regexes) = - frontend::frontend(sql, this_name, in_cols, opaque, structs, statics)?; + frontend::frontend(sql, this_name, in_cols, opaque, structs, statics, many)?; let one_row_blocker = one_row_blocker(&rel, &joins, statics); let mut program = lower::lower(&rel, &joins, statics, in_cols, out_cols, regexes, "run", many)?; @@ -95,8 +98,18 @@ pub fn prepare_opaque( let specs = joins .iter() .map(|j| { + if j.batch { + return StaticSpec { + batch: true, + table: String::new(), + key_cols: Vec::new(), + val_cols: Vec::new(), + val_nullable: Vec::new(), + }; + } let t = &statics[j.table]; StaticSpec { + batch: false, table: t.name.clone(), key_cols: j .key_cols @@ -144,11 +157,18 @@ fn one_row_blocker( } for j in joins { if j.kind != plan::JoinKind::Left { + if j.batch { + return Some("an INNER self-join drops rows on a miss".to_string()); + } return Some(format!( "INNER JOIN '{}' drops rows on a key miss (use LEFT JOIN)", statics[j.table].name )); } + if j.batch { + // A LEFT self-join still multiplies rows — never exactly-one. + return Some("a self-join multiplies rows".to_string()); + } } None } diff --git a/src/specializer/plan.rs b/src/specializer/plan.rs index f01ced0..ebf6b73 100644 --- a/src/specializer/plan.rs +++ b/src/specializer/plan.rs @@ -26,6 +26,7 @@ pub enum Rel { /// Value-column nullability is deliberately ignored here: arrow schemas /// default to nullable, so the real check — no NULL in a value column — /// happens against the data at materialization. +#[derive(Clone)] pub struct StaticTable { pub name: String, pub cols: Vec, @@ -85,7 +86,11 @@ pub enum JoinKind { /// key column split comes from the ON clause. pub struct JoinSpec { /// Index into the static-table catalog handed to `prepare`. + /// MEANINGLESS when `batch` is true. pub table: usize, + /// Stage-B self-join: the build side is the BATCH itself (a keyless + /// batchmap built per call; the whole ON rides in `residual`). + pub batch: bool, pub kind: JoinKind, /// Dynamic-side key expressions, one per key column, already promoted /// to the map's key types. diff --git a/src/specializer/tests.rs b/src/specializer/tests.rs index 31e5ed6..188ee41 100644 --- a/src/specializer/tests.rs +++ b/src/specializer/tests.rs @@ -2337,3 +2337,86 @@ fn many_shape_keyless_and_inequality_joins() { rows(&[&["1", "NULL"], &["2", "NULL"], &["3", "NULL"]]) ); } + +#[test] +fn many_shape_self_joins() { + // Stage-B self-joins: the batch is BOTH sides — a keyless batchmap + // built per call, the whole ON as residual (cross-then-filter is + // bit-identical under multiplicity; pins-stageB). + let schema = cols(&[("i", Ty::I64, false), ("j", Ty::I64, false)]); + let prep_many = |sql: &str| { + super::prepare_opaque(sql, "__THIS__", &schema, &[], &[], &[], true) + }; + let input = || { + batch( + 2, + vec![c_i64(&[Some(1), Some(2)]), c_i64(&[Some(10), Some(20)])], + ) + }; + let run_many = |sql: &str| -> Result>, String> { + let p = prep_many(sql).map_err(|e| e.to_string())?; + let f = compile(&p.program, vec![StaticData::Map(Vec::new())]) + .map_err(|e| e.to_string())?; + run_snapshot(&f, &input()).map_err(|e| e.to_string()) + }; + + // Comma cross self-join: 2x2 in probe-outer/batch-insertion order. + assert_eq!( + run_many("SELECT i1.i, i2.j FROM __THIS__ i1, __THIS__ i2").unwrap(), + rows(&[&["1", "10"], &["1", "20"], &["2", "10"], &["2", "20"]]) + ); + // Equi conjuncts stay WHERE (cross-then-filter). + assert_eq!( + run_many("SELECT i1.i, i2.j FROM __THIS__ i1, __THIS__ i2 WHERE i1.i = i2.i") + .unwrap(), + rows(&[&["1", "10"], &["2", "20"]]) + ); + // ON self-join, inequality + LEFT null-extension. + assert_eq!( + run_many("SELECT i1.i, i2.i FROM __THIS__ i1 JOIN __THIS__ i2 ON i1.i > i2.i") + .unwrap(), + rows(&[&["2", "1"]]) + ); + assert_eq!( + run_many( + "SELECT i1.i, i2.i FROM __THIS__ i1 LEFT JOIN __THIS__ i2 ON i1.i > i2.i" + ) + .unwrap(), + rows(&[&["1", "NULL"], &["2", "1"]]) + ); + // Star EXCLUDE over the self-join (the corpus shapes): unqualified + // strips both copies; qualified strips one side's. + let p = prep_many("SELECT * EXCLUDE (i) FROM __THIS__ i1, __THIS__ i2").unwrap(); + assert_eq!( + p.program + .out_cols + .iter() + .map(|c| c.name.as_str()) + .collect::>(), + ["j", "j_1"] // dup names go through the boundary rename + ); + let p = prep_many("SELECT i1.* EXCLUDE (i), i2.* EXCLUDE (j) FROM __THIS__ i1, __THIS__ i2") + .unwrap(); + assert_eq!( + p.program + .out_cols + .iter() + .map(|c| c.name.as_str()) + .collect::>(), + ["j", "i"] + ); + // Default shapes: still the named rejection. + let e = prep( + "SELECT i1.i FROM __THIS__ i1, __THIS__ i2", + &schema, + ) + .unwrap_err() + .to_string(); + assert!(e.contains("dynamic table"), "{e}"); + // USING self-join: named stage-B follow-up. + let e = match prep_many("SELECT * FROM __THIS__ i1 JOIN __THIS__ i2 USING (i)") { + Err(e) => e.to_string(), + Ok(_) => panic!("USING self-join must stay a named rejection"), + }; + assert!(e.contains("USING/NATURAL"), "{e}"); +} diff --git a/tests/test_corpus_replay.py b/tests/test_corpus_replay.py index 748c211..e1fdcc4 100644 --- a/tests/test_corpus_replay.py +++ b/tests/test_corpus_replay.py @@ -163,7 +163,7 @@ def _replay(case: dict) -> tuple[str, str]: # while letting the corpus exercise the multiplicity path; rows # compare as a sorted multiset below, which is exactly the pinned # parity contract (DuckDB's join order is a hash-join accident). - if "duplicate map key" in msg: + if "duplicate map key" in msg or "dynamic table to itself" in msg: try: fn = DuckDBInferFn( case["sql"], diff --git a/tests/test_known_limitations.py b/tests/test_known_limitations.py index 9a75d4b..e21ddae 100644 --- a/tests/test_known_limitations.py +++ b/tests/test_known_limitations.py @@ -69,10 +69,29 @@ def test_static_tables_are_frozen_unique_key_maps(): def test_dynamic_self_join_rejects(): + # Default shapes: the original named rejection. Under shape='many' the + # batch becomes the build side and ON self-joins SERVE (stage B). rejects( "SELECT t2.a FROM __THIS__ JOIN __THIS__ t2 ON __THIS__.a = t2.a", "dynamic table", ) + fn = DuckDBInferFn( + "SELECT t2.a FROM __THIS__ JOIN __THIS__ t2 ON __THIS__.a = t2.a", + row_tables={"__THIS__": T}, + static_tables={}, + output="dict", + shape="many", + ) + got = sorted(r["a"] for r in fn.infer({"__THIS__": [T(a=1), T(a=2)]})) + assert got == [1, 2] + # USING/NATURAL self-joins stay a named follow-up rejection. + with pytest.raises(ValueError, match="USING/NATURAL"): + DuckDBInferFn( + "SELECT * FROM __THIS__ t1 JOIN __THIS__ t2 USING (a)", + row_tables={"__THIS__": T}, + static_tables={}, + shape="many", + ) # ---- 2. Out of scope for row-serving --------------------------------------