From b86a8d291609ea27587820aa202105d07c3f7089 Mon Sep 17 00:00:00 2001 From: AmirHossein Roozbahani Date: Tue, 28 Jul 2026 03:59:50 +0200 Subject: [PATCH 1/7] docs(proposal): the columnar-path decision document (for the post-stage-B discussion) Co-Authored-By: Claude Fable 5 --- docs/proposals/2026-07-28-columnar-path.md | 97 ++++++++++++++++++++++ 1 file changed, 97 insertions(+) create mode 100644 docs/proposals/2026-07-28-columnar-path.md diff --git a/docs/proposals/2026-07-28-columnar-path.md b/docs/proposals/2026-07-28-columnar-path.md new file mode 100644 index 0000000..22f01e8 --- /dev/null +++ b/docs/proposals/2026-07-28-columnar-path.md @@ -0,0 +1,97 @@ +# Proposal: own the columnar path + +Status: **for discussion** (user + engine). Blocked work: none — this is +the candidate *next* wave after stage B. Decision owner: AmirHossein. + +## Why (measured, not vibes) + +TASK-57 closed with a decomposition of where serving time actually goes +(titanic scenario, 10 input cols, 31 output cols, release build, p50): + +| component | cost | +|---|---| +| whole boundary floor (trivial 1-col query, incl. 10-col ingest) | ~262 ns/row | +| output emission | ~37 ns per output column → ~1.15 µs at 31 cols | +| compute (the compiled program) | ~1.7 µs/row | +| handcrafted python twin — EVERYTHING | ~2.2 µs/row | + +Two conclusions. First, the input side is already near-free — a `__dict__` +fast path benched **neutral** and was deleted. Second, the remaining +overhead vs the twin is *per-Python-object work*: every output value is +boxed into a fresh PyObject, every output row into a dict/model, while the +twin pointer-copies passthrough fields. No row-at-a-time API can remove +that floor — it is the API, not the implementation. + +Meanwhile the engine is **already columnar inside**: `ColData` lanes in, +`OutCol` vectors out. The row boundary is a conversion layer we bolt onto +both ends of a columnar core. The proposal is to stop converting. + +## The proposal (option A, recommended) + +```python +fn = DuckDBInferFn(sql, row_tables=..., static_tables=..., shape="map") + +out: pa.Table = fn.infer_arrow(batch: pa.Table) # arrow in, arrow out +``` + +- Input: one `pa.Table` (or `RecordBatch`) whose columns map 1:1 onto the + engine's lanes (int64/float64/utf8/bool + validity = exactly our four + types + null lanes). Ingest = buffer walks, zero per-value PyObjects. +- Output: arrow arrays built directly from `OutCol` vectors — one + allocation per *column*, not per value. +- The row APIs (`infer`, `infer_rows`, pydantic models) stay untouched. + This is an additive fast lane, not a migration. + +### Why it composes with everything we just built + +- **`shape="map"` is the natural contract here**: `out[i] ↔ in[i]` means + the output table aligns positionally with the input table — the exact + guarantee vectorized feature pipelines want. `shape="filter"`/`"many"` + still work (output row count differs; optionally expose a + `__row_index__` output column so callers can re-align — decision below). +- **The multi-language thesis**: the WASM spike proved one Rust build runs + near-native in Go (wazero) and Java (chicory) with batching amortizing + the boundary. Arrow IPC as the wire format + this columnar entry point + is that story's missing half — the same engine serves Python, Go, and + Java with no per-language marshalling code. + +### Expected win (to be verified by bench, not promised) + +Eliminates the ~1.4 µs/row of boxing/emit at titanic's width and the +per-row call machinery; the remaining cost is compute. That should put +`spec` decisively **ahead** of the handcrafted python twin at every batch +size — the twin cannot go columnar without becoming numpy-vectorized +code, which is exactly the authoring burden this library exists to remove. + +## Costs and risks + +- **Dependency**: arrow ingestion in Rust. Two routes: `arrow-rs` (heavy + crate, full ecosystem) vs the Arrow **C Data Interface** via pyo3 + (lean — a few hundred lines for our four types + validity). Lean route + recommended; we only need 4 primitive layouts + utf8 offsets. +- **Strings**: input utf8 can be read zero-copy from arrow buffers; + output strings copy once from the arena into an arrow buffer (already + cheaper than one PyUnicode per value). +- **Chunked tables**: accept only single-chunk input at v0 (`combine_chunks()` + is the caller's one-liner) — named rejection otherwise. +- **API surface pre-1.0**: v0-no-compat makes this cheap to adjust later. +- **Bench honesty**: the comparison baseline for `infer_arrow` must be a + *vectorized numpy/pandas twin*, not the per-row python twin — new bench + rows, same three-way parity gate. + +## Open questions for the discussion + +1. **Entry point shape**: `infer_arrow(pa.Table) -> pa.Table`, or also + accept/return a dict of numpy arrays? (Arrow-only keeps one code path; + numpy is a `pa.table(...)` call away for callers.) +2. **Non-map shapes**: for `"filter"`/`"many"`, is an output-only + `__row_index__` column the right re-alignment tool, or do we return + `(table, indices)`? +3. **Zero-copy ambition**: v1 copies input buffers into `ColData` (simple, + already a big win) vs true zero-copy lanes reading arrow buffers in + place (bigger surgery in exec). Recommendation: copy first, measure, + then decide if zero-copy is worth it. +4. **Does the columnar path also want `output_model`-style typed schemas** + (arrow schema declaration up front), or is the derived schema enough? +5. **Multi-language sequencing**: is Python-only columnar the first ship, + with the WASM/arrow-IPC endpoint as its own later proposal? From 586e31086f1f726eba9c8129d01bd7ea1cb38fcc Mon Sep 17 00:00:00 2001 From: AmirHossein Roozbahani Date: Tue, 28 Jul 2026 04:08:11 +0200 Subject: [PATCH 2/7] =?UTF-8?q?docs(specializer):=20stage-B=20multiplicity?= =?UTF-8?q?=20pins=20+=20spec=20=E2=80=94=20multiset=20parity,=20engine-de?= =?UTF-8?q?fined=20order=20(TASK-59)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five-agent fleet, 57 pins. Row-SET contract: per-pair emission with full per-key cross products; LEFT = max(1, surviving matches) incl. the all-matches-filtered null-extension; NULL keys match nothing (even <>); keyless joins are cross-then-filter; USING carries the LEFT value on misses. Central 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, and the engine defines its own documented deterministic order. The 1172 family turns out to be pure NULL-semantics (no fan-out at all). Co-Authored-By: Claude Fable 5 --- ...up-key-cross-inequality-and-self-joins.md" | 48 ++++++++++ .../2026-07-28-stageB-multiplicity-pins.md | 72 +++++++++++++++ .../specs/pins-stageB/cross-inequality.json | 84 +++++++++++++++++ .../specs/pins-stageB/dup-key-equi.json | 59 ++++++++++++ .../specs/pins-stageB/left-join-1172.json | 53 +++++++++++ .../specs/pins-stageB/order-contract.json | 71 +++++++++++++++ .../specs/pins-stageB/self-join-stars.json | 89 +++++++++++++++++++ 7 files changed, 476 insertions(+) create mode 100644 "backlog/tasks/task-59 - Stage-B-join-multiplicity-under-shapemany-\342\200\224-dup-key-cross-inequality-and-self-joins.md" create mode 100644 docs/superpowers/specs/2026-07-28-stageB-multiplicity-pins.md create mode 100644 docs/superpowers/specs/pins-stageB/cross-inequality.json create mode 100644 docs/superpowers/specs/pins-stageB/dup-key-equi.json create mode 100644 docs/superpowers/specs/pins-stageB/left-join-1172.json create mode 100644 docs/superpowers/specs/pins-stageB/order-contract.json create mode 100644 docs/superpowers/specs/pins-stageB/self-join-stars.json 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" new file mode 100644 index 0000000..38c7418 --- /dev/null +++ "b/backlog/tasks/task-59 - Stage-B-join-multiplicity-under-shapemany-\342\200\224-dup-key-cross-inequality-and-self-joins.md" @@ -0,0 +1,48 @@ +--- +id: TASK-59 +title: >- + Stage B: join multiplicity under shape='many' — dup-key, cross/inequality, and + self-joins +status: In Progress +assignee: [] +created_date: '2026-07-28 01:55' +updated_date: '2026-07-28 01:57' +labels: + - specializer + - stage-b +dependencies: [] +type: feature +ordinal: 53000 +--- + +## Description + + +User go received 2026-07-28 ("feel free to implement the feature gap"), sequenced behind the TASK-58 shape flag: every stage-B construct builds ONLY under shape='many' (0..N rows out per row in) — the serving fence the user asked for. + +Constituency (census, pins-waveA/census-all-nonmatches.json): 27 cases = 17 'duplicate map key' (true 1:N equi-joins vs statics with repeated keys, PLUS keyless shapes that reduce to one bucket: comma/cross joins with range residuals, inequality ON predicates) + 10 self-joins (the batch as both probe and build side; mostly EXCLUDE/USING star semantics). Target: corpus 529 -> ~556 of 678 (82%). After this ships, everything left is out-of-scope-by-decision. + +Pins-first (fleet): 1:N emission ORDER and determinism, LEFT-miss null-extension under multiplicity, cross/inequality semantics, self-join star expansion + dup-name renames + USING merges, NULL keys among duplicate keys. Design per TASK-50 notes: per-key row lists in the frozen maps, inner emit loop per probe hit, bucket scan + residual predicates for keyless joins, per-call batch-side map build for self-joins. Backend decision (interpreter-first vs cranelift emit loops) goes in the spec, justified. + +Hard stop after shipping: columnar-API discussion with the user before any next wave. + + +## 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) + + +## Implementation Plan + + +Emit-machinery study (pre-spec, 2026-07-28): + +TODAY: both backends are strictly one-emit-per-row. Interpreter: per row, block walk ends at CTerm::Emit (emitted += 1, output values already written to st.out by instructions) or CTerm::Skip. Cranelift: compiled row_fn returns 0=emitted / 1=skip / 2+=trap — the RETURN CODE is the emit signal, so N-per-row cannot be expressed by return codes. + +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). + diff --git a/docs/superpowers/specs/2026-07-28-stageB-multiplicity-pins.md b/docs/superpowers/specs/2026-07-28-stageB-multiplicity-pins.md new file mode 100644 index 0000000..91551f2 --- /dev/null +++ b/docs/superpowers/specs/2026-07-28-stageB-multiplicity-pins.md @@ -0,0 +1,72 @@ +# Stage B: join multiplicity under shape='many' — pins + design (TASK-59) + +Measured against DuckDB 1.5.5, 2026-07-28. Raw pins: `pins-stageB/*.json` +(5 fleet agents, 57 pins). Gate: every construct here builds ONLY under +`shape='many'` (TASK-58); under `filter`/`map` the rejections stay exactly +as today. + +## The row-SET contract (stable everywhere — this is what we serve) + +1. **Inner equi-join**: one output row per (left row, matching right row) + pair; full per-key cross product when BOTH sides have duplicate keys + (2×3 = 6 verified). +2. **LEFT**: per left row, `max(1, surviving matches)` rows — exactly one + null-extended row when zero matches survive, INCLUDING when a residual + ON predicate filtered every match away. +3. **NULL keys match nothing** — under `=`, `>`, and crucially `<>` + (NULL <> NULL is NULL, not TRUE). A NULL-keyed left row = unmatched: + dropped by INNER, one null-extension by LEFT. (The entire 9-case + left_join_issue_1172 corpus family reduces to this — probe is NULL, so + every case is a single null-extended row; the dup keys are inert.) +4. **Residual ON predicates** filter per match-pair; WHERE composes + orthogonally (left rows removed first). +5. **Keyless joins** (comma, CROSS, `ON` inequality / one-sided / + constant): exactly cross-product-then-filter, verified bit-for-bit. + `ON NULL = 2` → zero matches (LEFT: null-extends every left row). + Empty right side: INNER → 0 rows, LEFT → all null-extended. +6. **USING** merges the key into ONE leading output column carrying the + LEFT value (visible on null-extended rows: the probe key, not NULL). + With ON, both copies appear; the model/dict boundary applies the + wave-5 dedup rename (`id, id_1`) — same convention as DuckDB's own + `.df()`. +7. **Self-join stars**: `SELECT *` = left cols then right cols in + declaration order; unqualified EXCLUDE strips EVERY copy; qualified + EXCLUDE strips one side's (USING survivor's position depends on which + copy went — pinned); excluding all of one side's star is legal; bare + names (and rowid) are ambiguity errors. + +## ORDER: an accident, not a contract (the fleet's central finding) + +DuckDB's join output order is a hash-join artifact on three independent +axes: the optimizer picks the streamed side by cost (not FROM order, not +reliably size), dup-key matches emit in REVERSE build-insertion order +(LIFO chains) in per-2048-row lockstep passes, and at threads>1 with +~500k+ rows the order differs RUN-TO-RUN on the same connection. Even +single-threaded, one probe row's matches land ~2048 rows apart — a +row-at-a-time engine cannot naturally reproduce it and must not try. + +**Decision**: parity for shape='many' is **multiset** (sort both sides +before comparing — the corpus already does; duck_check gains a sorted +mode for 'many' tests). The engine defines its OWN deterministic order, +documented: probe rows in input order; all matches for one probe row +contiguous; matches in build INSERTION order; the LEFT null-extension +in place. This intentionally diverges from DuckDB's accidental order — +SQL without ORDER BY promises none, and we reject ORDER BY anyway. + +## Engine design (validated against the emit-machinery study, TASK-59 plan) + +- Emission becomes explicit: `Inst::EmitRow` appends the out-lane values; + `CTerm::Emit` retires to end-of-row. Loops use the existing Brif/Jump + blocks with back-edges (verify must accept them — first change). +- `StaticTy::Map` values become per-key row LISTS (flat arena + + (offset, len) per key). New ops `ProbeStart`/`ProbeNext` iterate them; + keyless joins are the degenerate one-bucket scan with the residual in + the loop body; LEFT keeps an `any_emitted` flag and emits the + null-extension when the loop closes dry. +- Self-joins build the batch-side map per call before the row loop. +- Cranelift: emission via an `h_emit_row` helper (helpers already write + through Cx), loops as normal CLIF blocks; the row_fn return code stays + for end/trap only. Fallback if heavy: interpreter-first for 'many' + (documented, backend-identity tests exempt 'many'). +- Corpus harness: replay retries with shape='many' ONLY on the named + multiplicity rejections, so the default-shape rejections stay proven. diff --git a/docs/superpowers/specs/pins-stageB/cross-inequality.json b/docs/superpowers/specs/pins-stageB/cross-inequality.json new file mode 100644 index 0000000..2b50673 --- /dev/null +++ b/docs/superpowers/specs/pins-stageB/cross-inequality.json @@ -0,0 +1,84 @@ +{ + "topic": "cross joins and inequality-ON joins (keyless multiplicity)", + "duckdb_version": "1.5.5", + "pins": [ + { + "q": "select * from t1 left join t2 on t1.id > t2.id;", + "result": "cols ['id','id']; rows [(3, 2), (1, None), (2, None)]", + "note": "t1=(1,2,3), t2=(2,3,4). Only t1.id=3 matches (3>2); t1.id=1 AND t1.id=2 both null-extend (2 is not > any of 2,3,4). Matched rows stream FIRST, then all null-extended rows as a trailing block in t1 insertion order. Stable across 3 reruns and a fresh connection." + }, + { + "q": "SELECT * FROM t1, t2 WHERE t1.id > t2.id;", + "result": "cols ['id','id']; rows [(3, 2)]", + "note": "INNER inequality via comma join: non-matchers dropped, single surviving pair. Stable order across reruns/fresh connection." + }, + { + "q": "SELECT * FROM t1, t2;", + "result": "cols ['id','id']; rows [(1, 2), (2, 2), (3, 2), (1, 3), (2, 3), (3, 3), (1, 4), (2, 4), (3, 4)]", + "note": "3x3 cross join order is RHS-outer / LHS-inner (t1 varies fastest), NOT t1-outer. Duplicate column name 'id' appears twice: left cols then right cols. Deterministic across 3 reruns + fresh connection." + }, + { + "q": "SELECT * FROM big, small; -- big=range(500), small=(100,200,300,400)", + "result": "2000 rows; first 12 all (a,100) for a=0..11; b changes exactly at row indexes 500/1000/1500; matches t2-outer(t1-inner) exactly", + "note": "500x4: the small (4-row) RHS is tuple-iterated (outer) while the 500-row LHS chunk streams. Deterministic across reruns." + }, + { + "q": "SELECT * FROM small, big; -- args swapped", + "result": "2000 rows; first 12 all (100,b) for b=0..11; small col changes at 500/1000/1500", + "note": "Swapped operands: now the small LHS is tuple-iterated (outer) and big RHS varies fastest. So the OUTER loop side is chosen by chunk size (the smaller chunk is row-iterated), not by syntactic left/right. Compiler cannot assume left-outer nested-loop order." + }, + { + "q": "SELECT * FROM big, small; -- big=range(5000) (3 vectors), small=(10,20,30)", + "result": "15000 rows in 9 blocks: (b=10,a=0..2047),(b=20,a=0..2047),(b=30,a=0..2047),(b=10,a=2048..4095),(20,...),(30,...),(10,a=4096..4999),(20,...),(30,...)", + "note": "Multi-chunk: order is CHUNK-BLOCKED — for each 2048-row LHS vector, iterate all RHS rows, then advance to next LHS vector. Not a global nested loop over whole tables. Deterministic incl. fresh connection." + }, + { + "q": "SELECT * FROM t1 JOIN t2 ON t1.id > 2;", + "result": "cols ['id','id']; rows [(3, 2), (3, 3), (3, 4)]", + "note": "ON referencing ONLY the left side is legal: acts as filter on t1 then cross with all of t2 (t2 in insertion order). 1 x 3 = 3 rows." + }, + { + "q": "SELECT * FROM t1 JOIN t2 ON 1 = 1;", + "result": "cols ['id','id']; rows [(1, 2), (2, 2), (3, 2), (1, 3), (2, 3), (3, 3), (1, 4), (2, 4), (3, 4)]", + "note": "ON referencing NEITHER side, always-true: identical rows AND identical order to bare cross join." + }, + { + "q": "SELECT * FROM t1 JOIN t2 ON NULL = 2;", + "result": "cols ['id','id']; rows []", + "note": "Corpus case. NULL ON-predicate = no matches at all -> 0 rows. No error, no NULL-matching." + }, + { + "q": "SELECT * FROM t1 LEFT JOIN t2 ON NULL = 2;", + "result": "cols ['id','id']; rows [(1, None), (2, None), (3, None)]", + "note": "LEFT variant of NULL ON: every t1 row null-extends exactly once, t1 insertion order preserved." + }, + { + "q": "SELECT * FROM t1 LEFT JOIN t2 ON t1.id > 2;", + "result": "cols ['id','id']; rows [(3, 2), (3, 3), (3, 4), (1, None), (2, None)]", + "note": "LEFT with left-only ON: t1.id=3 matches all 3 t2 rows; t1.id=1,2 null-extend once each. Again matched block first, null-extended block last in t1 order." + }, + { + "q": "SELECT * FROM vals1, vals2 WHERE i>9 AND k>=j AND j>=i AND l>=k", + "result": "cols ['i','j','k','l']; rows [(10, 20, 20, 22), (12, 12, 20, 22), (10, 14, 20, 22), (12, 12, 13, 13)]", + "note": "vals1=(10,20),(5,30),(12,12),(10,14),(NULL,8); vals2=(11,25),(20,22),(10,20),(13,13),(NULL,50). EXACTLY equal (rows AND order) to the explicit filtered cross join SELECT * FROM (SELECT * FROM vals1 CROSS JOIN vals2) sub WHERE .... NULL i or NULL k rows are eliminated by the comparisons. Stable across reruns/fresh conn." + }, + { + "q": "SELECT * FROM t1, t2; -- t2 empty", + "result": "rows []", + "note": "Empty side -> 0 rows for cross join and for JOIN ON 1=1. Also 0 rows when t1 is the empty side." + }, + { + "q": "SELECT * FROM t1 LEFT JOIN t2 ON t1.id > t2.id; -- t2 empty", + "result": "rows [(1, None), (2, None), (3, None)]", + "note": "LEFT vs empty right side: every t1 row null-extended once, t1 insertion order. Same result for LEFT JOIN ON 1=1 vs empty t2. Empty t1 LEFT anything = []." + } + ], + "surprises": [ + "Cross-join output order is NOT left-outer nested loop: DuckDB row-iterates whichever side's chunk is SMALLER and streams the bigger chunk, so 't1,t2' (3x3) comes out RHS-outer/LHS-inner, while 500x4 vs 4x500 BOTH come out small-side-outer regardless of which operand is written first.", + "Above ~2048 rows the order becomes chunk-blocked: for each 2048-row LHS vector, all RHS rows are iterated, then the next LHS vector — a global nested-loop order assumption breaks exactly at vector boundaries.", + "LEFT inequality join emits ALL matched rows first and ALL null-extended rows as one trailing block (t1 insertion order), not interleaved per-probe-row.", + "Every order observed was deterministic across 3 reruns and a fresh connection (single-threaded-sized data), but the ordering rule is an implementation artifact of chunk sizes, not a spec guarantee.", + "ON predicates referencing one side or neither side are legal filters: ON t1.id>2 = filter-then-cross, ON 1=1 = exact cross join (same order too), ON NULL=2 = zero matches (LEFT: null-extend everything)." + ], + "summary": "Rules for a compiler writer (DuckDB 1.5.5, keyless multiplicity):\n1) SEMANTICS — comma join + WHERE, CROSS JOIN + filter, and JOIN ON are all exactly cross-product-then-filter; verified bit-for-bit (rows AND order) on the range-join corpus shape. INNER inequality drops non-matchers; each surviving (left,right) pair appears exactly once.\n2) MULTIPLICITY — inner/cross: |out| = matches (3x3 cross = 9; ON 1=1 = 9; ON NULL=2 = 0; any empty side = 0). LEFT: every left row appears max(matches,1) times; zero-match left rows null-extend exactly once (t2 columns -> None), including when t2 is empty or the ON is constant-NULL/false. ON may reference only one side (filter semantics) or neither (constant), no error.\n3) ORDER (observed artifact, deterministic but unspecified) — cross-product order is chunk-blocked and size-driven: per LHS vector (2048 rows), DuckDB tuple-iterates the smaller chunk side and streams the larger; at 3x3 this yields RHS-outer/LHS-inner (left column varies fastest); at 500x4 the 4-row side is outer regardless of operand order; at 5000x3 blocks are (LHS-chunk x RHS-row). LEFT inequality joins emit all matched rows first, then all null-extended rows as a trailing block in left insertion order. All orders repeated identically across 3 reruns and fresh connections — safe to pin in golden tests at this scale, but do NOT hard-code left-outer nested-loop order into the compiler; treat order as engine-defined.\n4) SCHEMA — SELECT * = left table's columns then right table's columns, duplicate names kept as-is (['id','id'])." +} \ No newline at end of file diff --git a/docs/superpowers/specs/pins-stageB/dup-key-equi.json b/docs/superpowers/specs/pins-stageB/dup-key-equi.json new file mode 100644 index 0000000..dab511a --- /dev/null +++ b/docs/superpowers/specs/pins-stageB/dup-key-equi.json @@ -0,0 +1,59 @@ +{ + "topic": "1:N equi-join semantics — duplicate build-side keys (emission order, LEFT null-extension, NULL keys, residual ON predicates, USING, duplicate output names)", + "duckdb_version": "1.5.5", + "pins": [ + { + "q": "SELECT * FROM probe LEFT JOIN dim ON probe.id = dim.id;", + "result": "cols=['id', 'tag', 'id', 'v'] rows=[(1, 'p1', 1, 'a'), (2, 'p2', 2, 'b'), (1, 'p1', 1, 'c'), (2, 'p2', 2, 'd'), (1, 'p1', 1, 'e'), (3, 'p3', None, None)]", + "note": "3-row probe / 5-row dim. Identical across 3 same-conn runs + fresh conn. Output is in DIM insertion order (a,b,c,d,e), NOT grouped by probe row — optimizer built the hash table on the smaller table (probe) and streamed dim as the physical probe side. Zero-match row (3,'p3') emits exactly one null-extended row, at the very END of the stream. cursor.description has BOTH columns literally named 'id'." + }, + { + "q": "SELECT * FROM probe JOIN dim ON probe.id = dim.id;", + "result": "cols=['id', 'tag', 'id', 'v'] rows=[(1, 'p1', 1, 'a'), (2, 'p2', 2, 'b'), (1, 'p1', 1, 'c'), (2, 'p2', 2, 'd'), (1, 'p1', 1, 'e')]", + "note": "INNER: zero-match probe row 3 dropped (verified). 5 rows = 1 probe row x 3 dim matches for key 1 + 1x2 for key 2. Same dim-scan-order emission. Deterministic 3x + fresh conn." + }, + { + "q": "SELECT * FROM probe JOIN dim ON probe.id = dim.id; -- after INSERT INTO probe VALUES (1,'p1b')", + "result": "rows=[(1, 'p1b', 1, 'a'), (2, 'p2', 2, 'b'), (1, 'p1b', 1, 'c'), (2, 'p2', 2, 'd'), (1, 'p1b', 1, 'e'), (1, 'p1', 1, 'a'), (1, 'p1', 1, 'c'), (1, 'p1', 1, 'e')]", + "note": "Dup keys BOTH sides: key 1 = 2 probe rows x 3 dim rows = 6 output rows (full cross product, verified). Order artifact: for each dim row the LAST-inserted probe row ('p1b') matches FIRST (build-side hash chains are LIFO), then a second pass emits 'p1' matches. LEFT variant identical + (3,'p3',None,None) appended last. Deterministic 3x fresh conns." + }, + { + "q": "SELECT * FROM probe LEFT JOIN dim ON probe.id = dim.id; -- after INSERT INTO dim VALUES (NULL,'n'); INSERT INTO probe VALUES (NULL,'pn')", + "result": "rows=[(1, 'p1', 1, 'a'), (2, 'p2', 2, 'b'), (1, 'p1', 1, 'c'), (2, 'p2', 2, 'd'), (1, 'p1', 1, 'e'), (None, 'pn', None, None), (3, 'p3', None, None)]", + "note": "NULL never matches NULL: dim(NULL,'n') produced zero matches; probe(NULL,'pn') got a LEFT null-extension exactly like a no-match key. INNER drops both (verified: same 5 rows as baseline). Order artifact: unmatched rows emitted ('pn' BEFORE 'p3') — NOT probe insertion order. Deterministic 3x." + }, + { + "q": "SELECT * FROM probe LEFT JOIN dim ON probe.id = dim.id AND dim.v > 'b'", + "result": "rows=[(1, 'p1', 1, 'e'), (2, 'p2', 2, 'd'), (1, 'p1', 1, 'c'), (3, 'p3', None, None)]", + "note": "Residual predicate in ON filters dim-side matches only ('a','b' removed); key semantics unchanged; key 3 still null-extends. CRITICAL order flip vs baseline: adding the residual changed the plan (dim became build side) so per-key order is now e,c (LIFO), interleaved by probe-vector rounds: first-match-per-row [(1,e),(2,d)], then second matches [(1,c)], then unmatched [(3,null)]. No probe row lost: a key whose matches are ALL filtered would null-extend (key 3 pattern). Deterministic 3x + fresh." + }, + { + "q": "SELECT * FROM probe JOIN dim USING (id)", + "result": "cols=['id', 'tag', 'v'] rows=[(1, 'p1', 'a'), (2, 'p2', 'b'), (1, 'p1', 'c'), (2, 'p2', 'd'), (1, 'p1', 'e')]", + "note": "USING merges the key: 3 output columns (id, tag, v), one 'id'. Multiplicity identical to ON form. LEFT JOIN USING: same + (3, 'p3', None) — merged id shows the PROBE value (3), not NULL. Deterministic." + }, + { + "q": "SELECT * FROM probe LEFT JOIN dim ON probe.id = dim.id -- via .df()", + "result": "df columns: ['id', 'tag', 'id_1', 'v']", + "note": "Name duplication: raw cursor.description gives ['id','tag','id','v'] (both literally 'id'); .fetchall() is positional so no conflict; .df() auto-renames the second to 'id_1'. A compiler emitting named outputs must handle both conventions." + }, + { + "q": "SELECT * FROM probe LEFT JOIN dim ON probe.id = dim.id; -- probe grown to 1003 rows (INSERT SELECT (i%10)+1, 'x'||i FROM range(1000))", + "result": "first 12: [(1,'p1',1,'e'),(2,'p2',2,'d'),(1,'x0',1,'e'),(2,'x1',2,'d'),...]; every key-1 probe row sees v-order ('e','c','a'); 801 null-extended rows contiguous at positions 505..1305; 3 fresh conns identical; threads=12", + "note": "Plan flipped again (big probe = physical probe side, dim = build): per-key match order is now e,c,a — the exact REVERSE of the 3-row case's a,c,e. Emission is round-based within a vector: first match for every probe row, then second, then third; nulls after. Same data, same query, different table SIZE => different per-key order." + }, + { + "q": "SELECT * FROM probe LEFT JOIN dim ON probe.id = dim.id; -- probe = 10,000 and 200,000 rows", + "result": "n=10000: 13000 rows, 8000 nulls, first null at index 1025, NOT contiguous at end, 9 null/match block transitions, 3 fresh runs identical. n=200000: 260000 rows, first null at 1025, 145 transitions, 3 fresh runs NOT identical (order differs run-to-run)", + "note": "Multi-vector reality: null-extensions interleave per-chunk (first null at 1025 ~ vector boundary of 2048/2... appears after the first processed chunk), NOT all-at-end. At 200k rows with threads=12, global row order is NON-DETERMINISTIC across runs (parallel hash join). Row SET is stable; row ORDER is not." + } + ], + "surprises": [ + "Output of the 3-row corpus query is in DIM insertion order, not probe order — DuckDB flipped the LEFT join to build on the smaller left table, so the SQL-left table is not necessarily the physical probe side.", + "Per-key match order REVERSES between small and large probe tables (a,c,e vs e,c,a): build-side hash chains are LIFO, and which side gets built depends on optimizer size estimates. Adding a residual ON predicate alone also flipped it.", + "Unmatched LEFT rows are not emitted in probe insertion order even in tiny cases (NULL-key row 'pn' came before 'p3'), and in multi-chunk cases null-extensions interleave per-chunk rather than trailing at the end.", + "At 200k input rows (threads=12) three identical fresh-connection runs returned DIFFERENT row orders — determinism observed at small scale is an artifact of single-chunk execution, not a guarantee.", + "Duplicate output column names: cursor.description literally reports two columns named 'id'; only .df() disambiguates (id_1)." + ], + "summary": "Rules for the compiler writer (DuckDB 1.5.5, unordered-query semantics):\n\n1. MULTIPLICITY (the real contract): inner equi-join emits one row per (left row, matching right row) pair — full per-key cross product when both sides have duplicate keys (2x3=6 verified). LEFT JOIN adds exactly ONE null-extended row per left row with zero surviving matches; INNER drops such rows. This row-SET contract is stable everywhere.\n\n2. NULL keys: NULL = NULL never matches (probe NULL vs dim NULL included). A NULL-keyed left row behaves exactly like an unmatched key: dropped by INNER, single null-extension by LEFT.\n\n3. RESIDUAL ON predicates (LEFT JOIN ... ON key AND pred): the predicate filters which matches survive, per match-pair; if ALL of a left row's matches are filtered out, the row still emits one null-extended row. Filters in WHERE on the left side compose orthogonally (they just remove left rows first).\n\n4. USING merges the join column into one output column carrying the LEFT value (visible on null-extended rows: id=3, not NULL); SELECT * = merged key + remaining left cols + remaining right cols. With ON, both id columns appear and cursor.description repeats the name 'id' verbatim; pandas .df() renames the second to 'id_1'.\n\n5. ORDER — do not compile against it. Observed order is a hash-join artifact: rows stream in the PHYSICAL probe-side scan order (the optimizer picks build/probe by estimated size — the SQL-left table is often the BUILD side), duplicate build-side matches emit in LIFO chain order in per-vector rounds (all first matches, then all second matches...), and LEFT null-extensions flush per-chunk, not at the end. The same query's per-key order reversed (a,c,e -> e,c,a) purely from table size, and flipped again from a residual predicate. Small single-chunk results are bit-identical across runs and connections, but at 200k rows with 12 threads run-to-run order diverges. Verdict: treat join output as an unordered multiset; any order the specializer needs must come from an explicit ORDER BY, and differential tests against DuckDB must compare sorted rows (or sort keys) — never raw fetch order." +} \ No newline at end of file diff --git a/docs/superpowers/specs/pins-stageB/left-join-1172.json b/docs/superpowers/specs/pins-stageB/left-join-1172.json new file mode 100644 index 0000000..80098bb --- /dev/null +++ b/docs/superpowers/specs/pins-stageB/left-join-1172.json @@ -0,0 +1,53 @@ +{ + "topic": "left_join_issue_1172.test corpus family (9 cases): LEFT JOIN null-extension with an all-NULL probe side, dup-key build side, predicates =, >, <>", + "duckdb_version": "1.5.5", + "pins": [ + { + "q": "setup (cases 0-2): drop table if exists t1; drop table if exists t2; create table t1 (id string); create table t2 (id string); insert into t1 values\n(NULL); insert into t2 values (1), (1);", + "result": "t1=[(None,)] t2=[('1',), ('1',)] (typeof(id)='VARCHAR' — the bare 1 literals are coerced to string '1')", + "note": "Stage-1 setup. Cases 3-5 add: insert into t2 values (1); Cases 6-8 further add: insert into t2 values (NULL), (NULL);" + }, + { + "q": "select * from t1 left join t2 on t1.id = t2.id;", + "result": "cols ['id','id'] rows [(None, None)]", + "note": "Case 0 (stage-1 setup). Identical result for case 3 (t2={1,1,1}) and case 6 (t2={1,1,1,NULL,NULL}). NULL probe row matches nothing (NULL = anything is not TRUE, including NULL = NULL), so exactly one null-extended row. Deterministic: 3x same connection + fresh connection all identical." + }, + { + "q": "select * from t1 left join t2 on t1.id > t2.id;", + "result": "cols ['id','id'] rows [(None, None)]", + "note": "Cases 1, 4, 7 — all three t2 stages give the same single null-extended row. NULL > x is never TRUE. Deterministic 3x + fresh conn." + }, + { + "q": "select * from t1 left join t2 on t1.id <> t2.id;", + "result": "cols ['id','id'] rows [(None, None)]", + "note": "Cases 2, 5, 8 — the trap the original DuckDB issue #1172 was about: <> does NOT mean 'NULL differs from 1'. NULL <> '1' is NULL, not TRUE, so still zero matches and one null-extended row even in stage 3 where t2 also has NULLs (NULL <> NULL is NULL too). Deterministic 3x + fresh conn." + }, + { + "q": "select count(*) from t1 left join t2 on / select count(*) from t1 join t2 on (all 3 stages x all 3 preds)", + "result": "LEFT count = 1 and INNER count = 0 for every stage/pred combination; matches predicted sum(max(1,matches))=1 and sum(matches)=0 exactly (18/18 checks ok)", + "note": "Row-count arithmetic verified: LEFT emits max(1, matches) per probe row; INNER emits matches. With a NULL-only probe side, matches is always 0 under =, >, <>." + }, + { + "q": ".df() on select * from t1 left join t2 on t1.id = t2.id (stage-3 setup, both sides named 'id')", + "result": "df.columns == ['id', 'id_1'] while cursor.description == ['id', 'id'] and fetchall == [(None, None)]", + "note": "Pandas dedupes duplicate column names by suffixing _1; the raw cursor keeps both as 'id'. A compiler comparing via .df() must expect the _1 suffix; via fetchall it must tolerate duplicate names." + }, + { + "q": "VARIANT (not in corpus) — same stage-3 setup plus insert into t1 values ('1'); then select * from t1 left join t2 on t1.id = t2.id;", + "result": "rows [('1', '1'), ('1', '1'), ('1', '1'), (None, None)] — stable across 3 fresh connections", + "note": "Exercises the fan-out branch the corpus family never triggers: probe '1' hits 3 dup build keys -> 3 rows; NULL probe -> 1 null-extended row. ORDER: matched rows come FIRST and the null-extension LAST, even though NULL was inserted into t1 first — LEFT hash-join output is NOT probe order." + }, + { + "q": "VARIANT — same setup, select * from t1 left join t2 on t1.id <> t2.id;", + "result": "rows [(None, None), ('1', None)] — stable across 3 fresh connections", + "note": "'1' <> '1' is FALSE and '1' <> NULL is NULL, so BOTH probe rows are unmatched. Here the output IS probe insertion order (inequality join path), unlike the = case above where null-extensions trailed." + } + ], + "surprises": [ + "The whole 9-case family never exercises fan-out at all: every case is matches=0 -> exactly 1 null-extended row, because the only probe row is NULL. The dup keys in t2 (and the t2 NULLs in stage 3) are deliberately inert — the file is a pure NULL-semantics regression test, not a multiplicity test.", + "Row order is join-algorithm dependent: with the extra '1' probe row, the equality join emitted the null-extended row LAST (after all matches, not in probe order), while the <> join emitted rows in probe order. Both orders were stable across reruns and fresh connections, but a compiler must not assume LEFT JOIN preserves probe order in general.", + ".df() silently renames the duplicate right-side 'id' to 'id_1', while cursor.description reports two identical 'id' names — two different truths about the same result depending on the fetch API.", + "insert into t2 values (1) into a STRING column stores VARCHAR '1' (implicit int->varchar cast), so t1.id > t2.id is a string comparison — irrelevant here (probe is NULL) but a trap if the family is extended." + ], + "summary": "Rules for the compiler from this family: (1) LEFT JOIN multiplicity is per-probe-row max(1, matches); INNER is sum(matches) — verified exactly against DuckDB for all 3 setups x 3 predicates (LEFT=1, INNER=0 everywhere). (2) A NULL probe value matches NOTHING under any comparison predicate: =, >, and crucially <> all evaluate to NULL (not TRUE) against every build value including NULL; NULL <> NULL is NULL. So all 9 corpus cases reduce to one null-extended output row [(None, None)] with cols ['id','id'] — the only stage-B machinery they require is LEFT null-extension on zero matches; the build-side dup keys never fan out. (3) All 9 results are single-row so ordering is trivially deterministic (verified 3x + fresh connection); but the fan-out variant shows equality LEFT joins emit matched rows before trailing null-extensions (not probe order) while the inequality path emitted probe order — treat LEFT JOIN output order as unspecified. (4) Column naming: SELECT * over a shared column name yields duplicate 'id','id' in cursor.description but 'id','id_1' in .df(). (5) The string column coerces bare integer literals: values (1) stores VARCHAR '1'." +} \ No newline at end of file diff --git a/docs/superpowers/specs/pins-stageB/order-contract.json b/docs/superpowers/specs/pins-stageB/order-contract.json new file mode 100644 index 0000000..264b7bc --- /dev/null +++ b/docs/superpowers/specs/pins-stageB/order-contract.json @@ -0,0 +1,71 @@ +{ + "topic": "Join output order: contract or accident? (stage B: join multiplicity)", + "duckdb_version": "1.5.5", + "pins": [ + { + "q": "SELECT * FROM probe JOIN dim USING (k)", + "result": "cols=['k','ins']; 10000 rows; first 8: [(38, 0), (67, 1), (73, 2), (8, 3), (38, 4), (18, 5), (76, 6), (27, 7)]; last 3: [(46, 9997), (5, 9998), (19, 9999)]; md5(full repr row-stream)=db999840974adc89e08af76903d2ca69", + "note": "Setup: probe(k)=1..100 inserted in order; dim(k,ins)=keys 1..100 x100 shuffled with python random.Random(42), ins=insertion position 0..9999. RESULT ORDER = dim's shuffled INSERTION order exactly (ins is 0..9999 ascending; k-seq == dim's shuffled key seq; NOT probe-major). DuckDB streamed the 10k-row dim and BUILT the hash table on the 100-row probe (EXPLAIN: probe is the right/build child). Stability: 5 runs same con + 2 fresh connections + threads=1 + threads=4 + preserve_insertion_order=false — ALL byte-identical (default threads=12). Deceptive: stable only because 10k rows never engages parallelism." + }, + { + "q": "SELECT * FROM dim JOIN probe USING (k)", + "result": "cols=['k','ins']; 10000 rows; md5=db999840974adc89e08af76903d2ca69 — byte-identical to 'probe JOIN dim USING (k)' including row order AND column order", + "note": "FROM order is completely irrelevant to both row order and (with USING) column order. The optimizer picks stream/build sides by cost, not syntax. 5 runs identical." + }, + { + "q": "SELECT * FROM probe JOIN dim ON probe.k = dim.k", + "result": "cols=['k','k','ins'] (duplicate name, probe.k then dim.k); first 8: [(38, 38, 0), (67, 67, 1), (73, 73, 2), (8, 8, 3), (38, 38, 4), (18, 18, 5), (76, 76, 6), (27, 27, 7)]; md5=15f31f9f389719beaff5b84b73465984", + "note": "Same ROW order as the USING version (verified (dim.k,ins) sequence == USING output), different column shape: ON keeps both key columns, USING merges them. SELECT * FROM probe, dim WHERE probe.k = dim.k gives the identical fingerprint 15f31f9f3897... — comma+WHERE == JOIN ON in both order and columns." + }, + { + "q": "SELECT * FROM ps JOIN ds USING (k)", + "result": "cols=['k','tag']; rows=[(2, 'x1'), (1, 'y1'), (2, 'x2'), (1, 'y2')]", + "note": "Setup: ps(k): INSERT VALUES (1),(2); ds(k,tag): INSERT VALUES (2,'x1'),(1,'y1'),(2,'x2'),(1,'y2'). Output = ds scan/insertion order (ds streamed, 2-row ps built). 'SELECT * FROM ds JOIN ps USING (k)' returns the identical rows. 3 runs + threads 1/4 + fresh connection all identical." + }, + { + "q": "SELECT * FROM s JOIN b USING (k)", + "result": "cols=['k','pid','tag']; rows=[(1, 50, 'r1'), (2, 40, 'r2'), (1, 50, 'r3'), (2, 40, 'r4'), (1, 30, 'r1'), (2, 20, 'r2'), (1, 30, 'r3'), (2, 20, 'r4'), (1, 10, 'r1'), (1, 10, 'r3')]", + "note": "Setup (threads=1): s(k,pid): VALUES (1,10),(2,20),(1,30),(2,40),(1,50); b(k,tag): VALUES (1,'r1'),(2,'r2'),(1,'r3'),(2,'r4'). SURPRISE: DuckDB built on s (5 rows) and STREAMED b (4 rows) — the LARGER side became the build side (EXPLAIN confirms), so 'smaller side builds' is not a rule either; it is a cost-model choice. Shows vectorized LOCKSTEP: matches for one streamed row are NOT contiguous — iteration 1 emits the 1st match for every streamed row in the vector (r1->50, r2->40, r3->50, r4->40), iteration 2 the 2nd (30/20...), iteration 3 the 3rd. Per-key match order from the build side = REVERSE insertion (50,30,10 for key 1 inserted 10,30,50). 3 runs identical." + }, + { + "q": "SELECT * FROM s3 JOIN b3 USING (k)", + "result": "cols=['k','pid','tag']; rows=[(1, 100, 'w'), (1, 200, 'w'), (1, 100, 'v'), (1, 200, 'v'), (1, 100, 'u'), (1, 200, 'u')]", + "note": "Setup (threads=1): b3(k,tag) inserted (1,'u'),(2,'zz'),(1,'v'),(1,'w'); s3 streams rows pid 100,200 (k=1) plus 10 non-matching filler rows to keep s3 the streamed side. Build-side dup-key matches come out w,v,u = exact REVERSE of build insertion order (LIFO hash chains), interleaved lockstep across the streamed vector." + }, + { + "q": "SELECT * FROM sv JOIN bv USING (k)", + "result": "10000 rows; tag run-length encoding: [('v', 2048), ('u', 2048), ('v', 2048), ('u', 2048), ('v', 904), ('u', 904)]; rows around index 2048: [(1, 2046, 'v'), (1, 2047, 'v'), (1, 0, 'u'), (1, 1, 'u')]; per-pid tag order always ('v','u'); pid sequence NOT globally nondecreasing", + "note": "Setup (threads=1): sv = 5000 rows all k=1, pid=0..4999; bv inserted (1,'u'),(1,'v'). Nails the vector geometry: per 2048-row input vector the join emits [1st-match x 2048][2nd-match x 2048]; the 5000-row scan chunks as 2048/2048/904. 'v' (later-inserted) always first = reverse chain. threads=4 3 runs == threads=1 here (too small to parallelize). Output order is therefore a function of DuckDB's VECTOR SIZE (2048) — unreproducible by any natural row-at-a-time loop." + }, + { + "q": "SELECT * FROM p JOIN d1 USING (k) JOIN d2 USING (k)", + "result": "cols=['k','a','b']; rows=[(2, 'a4', 'b1'), (1, 'a3', 'b2'), (2, 'a4', 'b3'), (1, 'a3', 'b4'), (2, 'a2', 'b1'), (1, 'a1', 'b2'), (2, 'a2', 'b3'), (1, 'a1', 'b4')]", + "note": "Setup: p: (1),(2); d1(k,a): (1,'a1'),(2,'a2'),(1,'a3'),(2,'a4'); d2(k,b): (2,'b1'),(1,'b2'),(2,'b3'),(1,'b4'). Chained-join nesting is optimizer-chosen, not FROM-first: plan = HASH_JOIN(stream=SEQ_SCAN d2, build=HASH_JOIN(stream=d1, build=p)). Output = d2-major (b1,b2,b3,b4 lockstep) with d1 matches in reverse insertion (a4/a3 pass then a2/a1 pass). p — the FROM-first table — is the innermost build. Identical across 3 runs, threads 1/4, fresh connection." + }, + { + "q": "SELECT * FROM bigprobe JOIN dim2 USING (k)", + "result": "1,000,000 rows. threads=1: 3 runs byte-identical, md5 prefix 4c9d94b1f543, first=(1, 0, 1756), last=(1000, 499999, 460). threads=default(12): 3 runs on SAME connection produced DIFFERENT fingerprints (dc5847002a5f first run), last=(640, 368639, 698). threads=4: 3 runs also mutually different (a725f8bc27d9 first run); fresh connection at threads=4 matched none of them. pid nondecreasing = False in ALL configs.", + "note": "Setup: bigprobe = 500k rows, k=range%1000+1, pid=range (insertion order); dim2 = keys 1..1000 x2 shuffled with Random(7), ins=position. Here the BIG side streams (dim2 builds). VERDICT PIN: multi-threaded DuckDB join output order is NONDETERMINISTIC run-to-run on the same connection and data. threads=1 is deterministic but interleaved (pid=0's two matches are ins [1756, 427] = reverse of dim2's k=1 insertion positions [427, 1756], ~2048 rows apart). 'SELECT * FROM dim2 JOIN bigprobe USING (k)' at threads=1 = same row sequence modulo column permutation (cols follow FROM order: k,ins,pid vs k,pid,ins)." + }, + { + "q": "SELECT * FROM lp LEFT JOIN ld USING (k)", + "result": "cols=['k','pid','tag']; rows=[(1, 1, 'm3'), (2, 3, 'm1'), (1, 5, 'm3'), (1, 1, 'm2'), (1, 5, 'm2'), (3, 2, None), (3, 4, None)]", + "note": "Setup (threads=1): lp(k,pid): (1,1),(3,2),(2,3),(3,4),(1,5); ld(k,tag): (2,'m1'),(1,'m2'),(1,'m3'). LEFT JOIN placement: all match iterations first (lockstep, reverse-chain m3 before m2), then the NULL-padded unmatched rows appended at the END of the chunk in streamed order. Unmatched rows do NOT sit at their input position. 3 runs + fresh connection identical." + }, + { + "q": "SELECT * FROM t_small JOIN t_big USING (k)", + "result": "cols=['k','tag','bid']; rows=[(1, 's3', 0), (2, 's1', 1), (1, 's3', 2), (2, 's1', 3), (1, 's3', 4), (2, 's1', 5), (1, 's2', 0), (1, 's2', 2), (1, 's2', 4)]", + "note": "Setup (threads=1): t_small(k,tag): (2,'s1'),(1,'s2'),(1,'s3'); t_big: 6 rows k=range%2+1, bid=range. FROM-first t_small is the BUILD side; t_big streams (bid 0..5 lockstep). t_small's k=1 dup matches come s3-then-s2 = reverse insertion. Confirms: output order tracks the optimizer-chosen streamed side, never the FROM-first table per se." + } + ], + "surprises": [ + "Multi-threaded nondeterminism is real and same-connection: at 500k streamed rows / 1M output, threads=12 and threads=4 each gave 3 DIFFERENT orderings across 3 consecutive runs of the identical query on the same connection. Any order pin against default-config DuckDB is broken by construction.", + "Even single-threaded order is unreproducible row-at-a-time: matches for one input row are NOT contiguous. The hash join emits per-2048-row-vector lockstep passes ([1st match for all rows in vector][2nd match for all rows]...), so a probe row's k matches land ~2048 rows apart. Order depends on DuckDB's internal vector size.", + "Dup-key build matches come in REVERSE insertion order (LIFO hash chains): build rows inserted u,v,w match as w,v,u — 'dim insertion order' is exactly wrong.", + "The streamed (order-defining) side is a cost-model choice, not FROM position — and not even reliably the bigger side: with 4 vs 5 rows DuckDB streamed the 4-row table and built the 5-row one. You cannot predict which table's order dominates without running DuckDB's optimizer.", + "Small joins look like a contract but aren't: the 10k-row join was byte-stable across 5 runs, fresh connections, threads=1/4/12, and preserve_insertion_order=false — stability there is just parallelism failing to engage, and it silently evaporates at scale.", + "LEFT JOIN unmatched rows are appended after all match passes of the chunk (NULL rows last), not emitted at their input position.", + "USING vs ON vs comma+WHERE: identical row order; they differ only in column shape (USING merges the key column; ON/comma keep both; column order follows FROM order)." + ], + "summary": "DuckDB join output order is an ACCIDENT, not a contract — three independent reasons: (1) which table's scan order dominates is the optimizer's stream/build choice (never FROM order, not even reliably 'smaller builds'); (2) even at threads=1 the vectorized hash join interleaves matches in 2048-row lockstep passes with dup-key matches in REVERSE build-insertion order, so no row-at-a-time loop can naturally reproduce it; (3) at threads>1 with ~500k+ streamed rows, order differs RUN-TO-RUN on the same connection. Rules for the compiler: pin join semantics as MULTISET parity (sort both sides before comparing against the DuckDB oracle; never compare join output ordered). Give our row-at-a-time engine its own documented deterministic order — natural choice: left/probe input rows in input order, all matches for one input row contiguous, matches in dim/build insertion order, LEFT JOIN NULL row emitted in place — and document that this intentionally diverges from DuckDB even single-threaded. The only DuckDB configuration that is deterministic is threads=1 (byte-stable across runs and fresh connections at all tested scales), so if an order-sensitive differential test is ever unavoidable, it must SET threads=1 AND still expect DuckDB's lockstep/reverse-chain order, which is not worth reproducing. One clean special case exists but is too fragile to rely on: single-threaded + unique-keyed build side gives output exactly in streamed-side scan order with no interleaving. Column-shape rules measured in passing: USING merges the key column, ON/comma keep both copies, output column order follows FROM order while row order does not depend on FROM order at all." +} \ No newline at end of file diff --git a/docs/superpowers/specs/pins-stageB/self-join-stars.json b/docs/superpowers/specs/pins-stageB/self-join-stars.json new file mode 100644 index 0000000..0dbd414 --- /dev/null +++ b/docs/superpowers/specs/pins-stageB/self-join-stars.json @@ -0,0 +1,89 @@ +{ + "topic": "self-joins — star expansion, EXCLUDE, USING (stage-B self-join corpus cases)", + "duckdb_version": "1.5.5", + "pins": [ + { + "q": "SELECT * EXCLUDE (i, j) FROM integers i1, integers i2", + "result": "cols=['k', 'k']; rows=[(100, 100), (200, 100), (100, 200), (200, 200)]", + "note": "CORPUS VERBATIM. Unqualified EXCLUDE strips the named columns from BOTH sides of the self-join. Duplicate output name 'k' survives in cursor.description. Row order = i1 varies fastest, i2 slowest. Stable 3x + fresh conn." + }, + { + "q": "SELECT i1.* EXCLUDE (i, j), i2.* EXCLUDE (i, j, k) FROM integers i1, integers i2", + "result": "cols=['k']; rows=[(100,), (100,), (200,), (200,)]", + "note": "CORPUS VERBATIM. Qualified star EXCLUDE can strip ALL columns of one side (i2 contributes zero columns — legal, no error). NOTE ROW ORDER: here i1 varies SLOWEST (100,100,200,200) — opposite nesting vs plain star. Stable 3x + fresh conn." + }, + { + "q": "SELECT i1.* EXCLUDE (i, j), i2.* EXCLUDE (k) FROM integers i1, integers i2", + "result": "cols=['k', 'i', 'j']; rows=[(100, 1, 10), (100, 2, 20), (200, 1, 10), (200, 2, 20)]", + "note": "CORPUS VERBATIM. Per-side EXCLUDE lists are independent; remaining columns keep table order within each star. Row order again i1-slowest here. Stable 3x + fresh conn." + }, + { + "q": "SELECT * EXCLUDE (i) FROM integers i1 JOIN integers i2 USING (i)", + "result": "cols=['j', 'k', 'j', 'k']; rows=[(10, 100, 10, 100), (20, 200, 20, 200)]", + "note": "CORPUS VERBATIM. USING merges i into ONE leading column; unqualified EXCLUDE (i) removes that merged column. Leftovers: i1.j, i1.k, i2.j, i2.k with duplicate names. Stable 3x + fresh conn." + }, + { + "q": "SELECT * EXCLUDE (i1.i, i2.i) FROM integers i1 JOIN integers i2 USING (i)", + "result": "cols=['j', 'k', 'j', 'k']; rows=[(10, 100, 10, 100), (20, 200, 20, 200)]", + "note": "CORPUS VERBATIM. Excluding both qualified copies of the USING column is identical to unqualified EXCLUDE (i) — merged column fully removed. Stable 3x + fresh conn." + }, + { + "q": "SELECT * FROM integers i1, integers i2", + "result": "cols=['i', 'j', 'k', 'i', 'j', 'k']; rows=[(1, 10, 100, 1, 10, 100), (2, 20, 200, 1, 10, 100), (1, 10, 100, 2, 20, 200), (2, 20, 200, 2, 20, 200)]", + "note": "Star over cross self-join: 6 columns with DUPLICATE names in cursor.description (no renaming, no qualification). Row order: i1 cycles FASTEST, i2 slowest (i.e. i2 is the outer loop). Stable 3x + fresh conn. Via .df() pandas dedups to ['i','j','k','i_1','j_1','k_1'] — the _1 suffix is a pandas-conversion artifact, NOT SQL-level naming." + }, + { + "q": "SELECT * FROM integers i1 JOIN integers i2 USING (i)", + "result": "cols=['i', 'j', 'k', 'j', 'k']; rows=[(1, 10, 100, 10, 100), (2, 20, 200, 20, 200)]", + "note": "USING (i) self-join star = 5 columns: merged i ONCE (first), then i1's leftovers (j,k), then i2's leftovers (j,k) with duplicate names. .df() gives ['i','j','k','j_1','k_1']. Stable 3x + fresh conn." + }, + { + "q": "SELECT i FROM integers i1, integers i2", + "result": "BinderException: Binder Error: Ambiguous reference to column name \"i\" (use: \"i1.i\" or \"i2.i\")\\n\\nLINE 1: SELECT i FROM integers i1, integers i2\\n ^", + "note": "Bare column over a comma self-join is a binder error; message suggests both qualifications." + }, + { + "q": "SELECT i FROM integers i1 JOIN integers i2 USING (i)", + "result": "cols=['i']; rows=[(1,), (2,)]", + "note": "After USING (i), bare i RESOLVES (no ambiguity) to the merged column. Stable 3x + fresh conn." + }, + { + "q": "SELECT i1.rowid FROM integers i1, integers i2", + "result": "cols=['rowid']; rows=[(0,), (1,), (0,), (1,)]", + "note": "Qualified rowid works on self-join; values are the base-table row ids per side (i1 cycling fastest, matching plain-star order). Unqualified 'SELECT rowid FROM integers i1, integers i2' is BinderException: Ambiguous reference to column name \"rowid\" (use: \"i1.rowid\" or \"i2.rowid\"). Stable 3x + fresh conn." + }, + { + "q": "SELECT * EXCLUDE (i1.i) FROM integers i1, integers i2", + "result": "cols=['j', 'k', 'i', 'j', 'k']; rows=[(10, 100, 1, 10, 100), (10, 100, 2, 20, 200), (20, 200, 1, 10, 100), (20, 200, 2, 20, 200)]", + "note": "Qualified EXCLUDE strips only the named side's copy; i2's i survives in i2's position. SURPRISE: row order here is i1-SLOWEST — differs from plain star's i1-fastest on the same FROM clause. Deterministic (10 fresh conns identical) but plan/projection-dependent." + }, + { + "q": "SELECT * EXCLUDE (i2.i) FROM integers i1, integers i2", + "result": "cols=['i', 'j', 'k', 'j', 'k']; rows=[(1, 10, 100, 10, 100), (2, 20, 200, 10, 100), (1, 10, 100, 20, 200), (2, 20, 200, 20, 200)]", + "note": "Mirror case: i1's i survives in leading position. Row order matches plain star (i1-fastest). Stable 3x + fresh conn." + }, + { + "q": "SELECT * EXCLUDE (i1.i) FROM integers i1 JOIN integers i2 USING (i)", + "result": "cols=['j', 'k', 'i', 'j', 'k']; rows=[(10, 100, 1, 10, 100), (20, 200, 2, 20, 200)]", + "note": "On a USING join, EXCLUDE (i1.i) removes the merged column from the LEADING slot and re-materializes i at i2's slot (position 3). Contrast: EXCLUDE (i2.i) keeps i at leading position 0 (cols=['i','j','k','j','k']). So qualified EXCLUDE of a USING column controls WHERE the survivor sits." + }, + { + "q": "SELECT i1.i, i2.i FROM integers i1 JOIN integers i2 USING (i)", + "result": "cols=['i', 'i']; rows=[(1, 1), (2, 2)]", + "note": "USING does NOT delete the per-side columns — both qualified copies remain addressable (and equal on inner join). Stable 3x + fresh conn." + }, + { + "q": "SELECT i1.* FROM integers i1 JOIN integers i2 USING (i)", + "result": "cols=['i', 'j', 'k']; rows=[(1, 10, 100), (2, 20, 200)]", + "note": "Qualified star after USING still includes the join column i (both i1.* and i2.* return full 3-col rows). The merge-to-one only affects unqualified *." + } + ], + "surprises": [ + "ROW ORDER IS PROJECTION-DEPENDENT on the same FROM clause: SELECT * FROM integers i1, integers i2 yields i1-cycling-fastest order, but SELECT * EXCLUDE (i1.i) over the identical cross join yields i1-cycling-SLOWEST order (optimizer picks a different build/probe side based on projected columns). Each individual query is deterministic (3x same-conn + 10 fresh conns identical), but a compiler must NOT assume one canonical nesting order for un-ORDER-BY'd cross self-joins.", + "cursor.description happily returns DUPLICATE column names (['k','k'], ['i','j','k','i','j','k']) — no renaming at the SQL level; the i_1/j_1/k_1 suffixes seen in .df() are pandas-conversion artifacts only.", + "Qualified EXCLUDE of a USING join column relocates the survivor: EXCLUDE (i1.i) puts i at i2's positional slot (position 3 of 5), while EXCLUDE (i2.i) leaves it at the leading merged position 0 — same values, different column order.", + "i2.* EXCLUDE (i, j, k) — excluding EVERY column of one star — is legal and simply contributes zero columns (no error).", + "Unqualified rowid is ambiguous on a self-join (BinderException naming i1.rowid/i2.rowid), exactly like a regular column; qualified i1.rowid serves base-table row ids 0,1 per side." + ], + "summary": "Rules for the compiler (DuckDB 1.5.5, table integers(i,j,k) rows (1,10,100),(2,20,200)):\n\n1. STAR EXPANSION over comma/cross self-join concatenates left columns then right columns in table-definition order, keeping duplicate names verbatim in cursor.description ('i','j','k','i','j','k'). No auto-renaming; pandas .df() dedups with _1 suffixes but that is client-side only.\n\n2. UNQUALIFIED EXCLUDE (col,...) removes EVERY copy of each named column across all sides (confirmed for self-joins: EXCLUDE (i,j) over the cross join leaves ['k','k']). QUALIFIED EXCLUDE (i1.i) removes only that side's copy; the other side's copy survives at its own position. Excluding all columns of one qualified star is legal (side contributes nothing).\n\n3. USING (i) semantics: unqualified * emits the merged join column ONCE, FIRST, then i1's remaining columns, then i2's remaining columns (['i','j','k','j','k']). Bare i resolves (no ambiguity) to the merged column. Both i1.i and i2.i remain addressable, and i1.*/i2.* still contain i — the merge affects only unqualified star expansion and bare-name resolution. EXCLUDE (i) == EXCLUDE (i1.i, i2.i) == drop the merged column entirely -> ['j','k','j','k']. EXCLUDE of exactly one qualified copy keeps i but its POSITION depends on which copy was excluded: EXCLUDE (i2.i) keeps it leading (position 0); EXCLUDE (i1.i) moves it to i2's slot (position 3).\n\n4. AMBIGUITY: bare column (and bare rowid) over a comma self-join raises BinderException 'Ambiguous reference to column name \\\"X\\\" (use: \\\"i1.X\\\" or \\\"i2.X\\\")'. Qualified rowid works and returns per-side base-table row ids.\n\n5. ORDER: every query above is deterministic across reruns and fresh connections, BUT cross-join row order flips between i1-fastest and i1-slowest depending on the projection (plain * vs EXCLUDE variants) — plan-dependent. Compiler must treat un-ORDER-BY'd multi-row output order as unspecified and compare stage-B results order-insensitively (or sort) unless it replicates DuckDB's plan choice." +} \ No newline at end of file From 1bfa50b456c5e8d1a935876ebcfd3e3270ee646a Mon Sep 17 00:00:00 2001 From: AmirHossein Roozbahani Date: Tue, 28 Jul 2026 04:21:49 +0200 Subject: [PATCH 3/7] =?UTF-8?q?docs(proposal):=20measured=20us-vs-duckdb-o?= =?UTF-8?q?n-arrow=20numbers=20=E2=80=94=20serving=20regime=20ours=20by=20?= =?UTF-8?q?1-3=20orders,=20crossover=20~2-3k=20rows/call=20today?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 --- docs/proposals/2026-07-28-columnar-path.md | 27 ++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/docs/proposals/2026-07-28-columnar-path.md b/docs/proposals/2026-07-28-columnar-path.md index 22f01e8..2b5ed9b 100644 --- a/docs/proposals/2026-07-28-columnar-path.md +++ b/docs/proposals/2026-07-28-columnar-path.md @@ -95,3 +95,30 @@ code, which is exactly the authoring burden this library exists to remove. (arrow schema declaration up front), or is the derived schema enough? 5. **Multi-language sequencing**: is Python-only columnar the first ship, with the WASM/arrow-IPC endpoint as its own later proposal? + +## Measured: us vs DuckDB-on-pyarrow (2026-07-28, titanic, p50 per call) + +DuckDB gets a PRE-BUILT arrow table each call (register+execute+fetch +arrow = serving-realistic; "floor" = re-execute on an already-registered +table, its absolute best). We run today's ROW path (spec_dict) — an upper +bound for any columnar path of ours. + +| n/call | duckdb | duckdb floor | us (row path, today) | us vs duckdb | +|---|---|---|---|---| +| 1 | 6.58 ms | 5.52 ms | 3.3 µs | **2055× faster** | +| 8 | 6.28 ms | 5.86 ms | 24 µs | **265×** | +| 64 | 6.75 ms | 5.94 ms | 206 µs | **33×** | +| 1024 | 7.75 ms | 6.93 ms | 3.42 ms | **2.3×** | +| 16384 | 20.4 ms | 18.7 ms | 60.5 ms | 0.3× (duckdb wins) | +| 131072 | 105 ms | 98 ms | 543 ms | 0.2× (duckdb wins) | + +Reading: DuckDB pays ~5.5–7 ms of per-QUERY cost every call regardless of +batch size — we pay it once at build. The serving regime (1–1k rows/call) +is ours by 1–3 orders of magnitude ALREADY, on the row path. Today's +crossover is ~2–3k rows/call. The columnar path moves that crossover, not +the small-batch story: boundary-only (compute stays ~1.7 µs/row) puts +16k-row calls at ~29 ms vs DuckDB's 20 ms (near-tie); a vectorized +columnar CORE (2–3× on compute) would put the crossover at ~100k+ +rows/call — competitive everywhere except true analytic scans, where +DuckDB's parallelism should keep the crown (and that's fine: that's not +serving). From aae960e76e9d56abd3068f43d070e79294e8b9e8 Mon Sep 17 00:00:00 2001 From: AmirHossein Roozbahani Date: Tue, 28 Jul 2026 04:35:52 +0200 Subject: [PATCH 4/7] =?UTF-8?q?feat(specializer):=20stage-B=20IR=20machine?= =?UTF-8?q?ry=20=E2=80=94=20emit.to=20loops,=20multimap=20statics,=20range?= =?UTF-8?q?=20probes=20(TASK-59=20stage=20B-1)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Term::EmitTo { to, args }: emit the completed row AND continue at a block — the multiplicity loop's back-edge. Verifier: cycles are now legal iff every block still reaches a row-ENDING terminator (emit/skip/trap), which keeps rejecting degenerate spin loops; the store dataflow excludes back-edges from its topo order and EmitTo propagates the RESET all-zero store state to the loop header. - StaticTy::MultiMap + Inst::ProbeRange/ProbeRead: duplicate-key maps probed as a flat equal-key row range (stable sort keeps INSERTION order among equal keys — the engine's documented 1:N emission order); zero keys = the whole table (keyless cross/inequality joins). - Interpreter executes all of it; cranelift pre-rejects multiplicity programs so the existing interp fallback is the documented 'many' path. Print/parse roundtrip (multimap header, probe.range/probe.read, emit.to) + two machinery fixtures: dup-key fan-out with skip-on-miss, and the keyless cross join. Co-Authored-By: Claude Fable 5 --- src/specializer/exec/cranelift.rs | 25 ++++ src/specializer/exec/interp.rs | 97 ++++++++++++++ src/specializer/exec/tests.rs | 66 ++++++++++ src/specializer/ir/fixtures.rs | 26 ++++ src/specializer/ir/gen.rs | 4 + src/specializer/ir/mod.rs | 52 ++++++++ src/specializer/ir/parse.rs | 62 +++++++++ src/specializer/ir/print.rs | 21 +++ src/specializer/ir/verify.rs | 205 +++++++++++++++++++++++++++--- 9 files changed, 539 insertions(+), 19 deletions(-) diff --git a/src/specializer/exec/cranelift.rs b/src/specializer/exec/cranelift.rs index 75b65f2..7938431 100644 --- a/src/specializer/exec/cranelift.rs +++ b/src/specializer/exec/cranelift.rs @@ -950,6 +950,25 @@ fn clif_ty(ty: Ty) -> types::Type { } pub fn compile(p: &Program, statics: Vec) -> Result { + // Stage-B multiplicity programs (EmitTo loops / multimap probes) are + // interpreter-only for now: rejecting here makes the caller's existing + // interp fallback the documented 'many' path. The match arms below can + // then treat these constructs as unreachable. + let has_multiplicity = p + .statics + .iter() + .any(|s| matches!(s, super::super::ir::StaticTy::MultiMap { .. })) + || p.blocks.iter().any(|b| { + matches!(b.term, Term::EmitTo { .. }) + || b.insts + .iter() + .any(|i| matches!(i, Inst::ProbeRange { .. } | Inst::ProbeRead { .. })) + }); + if has_multiplicity { + return Err(CompileError::Static( + "multiplicity programs run on the interpreter (stage B)".into(), + )); + } // The interpreter compile also runs verify + prepare_statics; its // prepared statics are the ones the helpers read. let interp = interp::compile(p, statics)?; @@ -1105,6 +1124,9 @@ pub fn compile(p: &Program, statics: Vec) -> Result { + unreachable!("multiplicity programs are rejected before codegen") + } Term::Jump { to, args } => { let a = arg_list(&vals, args); b.ins().jump(blocks[to.0 as usize], &a); @@ -1253,6 +1275,9 @@ fn translate_inst( let icon = |b: &mut FunctionBuilder, v: i64| b.ins().iconst(types::I64, v); match inst { + Inst::ProbeRange { .. } | Inst::ProbeRead { .. } => { + unreachable!("multiplicity programs are rejected before codegen") + } Inst::Const { dst, lit } => { let v = match lit { Lit::I1(x) => V::S(b.ins().iconst(types::I8, *x as i64)), diff --git a/src/specializer/exec/interp.rs b/src/specializer/exec/interp.rs index ae3b07f..14a5340 100644 --- a/src/specializer/exec/interp.rs +++ b/src/specializer/exec/interp.rs @@ -102,6 +102,12 @@ pub(super) enum PreparedStatic { Map { entries: Vec<(Vec, Vec)>, }, + /// Stage-B: sorted by key with DUPLICATES ADJACENT; probe.range finds + /// the equal-key run, probe.read indexes into it. Keyless multimaps + /// (cross/inequality joins) range over the whole table. + MultiMap { + entries: Vec<(Vec, Vec)>, + }, } struct CBlock { @@ -126,6 +132,11 @@ enum CTerm { else_moves: Vec<(u32, u32)>, }, Emit, + /// Emit the completed row and continue at `to` (multiplicity loops). + EmitTo { + to: usize, + moves: Vec<(u32, u32)>, + }, Skip, Trap(String), } @@ -253,6 +264,11 @@ impl InterpFn { emitted += 1; break; } + CTerm::EmitTo { to, moves } => { + emitted += 1; + do_moves(ctx.regs, moves); + bi = *to; + } CTerm::Skip => break, CTerm::Trap(msg) => return Err(Trap(msg.clone())), } @@ -414,6 +430,33 @@ pub(super) fn prepare_statics( } prepared.push(PreparedStatic::Map { entries }); } + (StaticTy::MultiMap { keys, values }, StaticData::Map(mut entries)) => { + for (ei, (k, v)) in entries.iter().enumerate() { + let kt: Vec = k.iter().map(|kb| kb.ty()).collect(); + let vt: Vec = v.iter().map(|sv| sv.ty()).collect(); + if kt != *keys || vt != *values { + return Err(CompileError::Static(format!( + "@{i}: entry {ei} has shape ({kt:?}) -> ({vt:?}), declared ({keys:?}) -> ({values:?})" + ))); + } + } + for (k, _) in entries.iter_mut() { + for kb in k.iter_mut() { + if let KeyBits::F64(bits) = kb { + *bits = super::canon_f64_bits(f64::from_bits(*bits)); + } + } + } + // Stable sort: equal keys keep INSERTION order — the + // engine's documented emission order for 1:N matches. + entries.sort_by(|a, b| a.0.cmp(&b.0)); + prepared.push(PreparedStatic::MultiMap { entries }); + } + (StaticTy::MultiMap { .. }, StaticData::Scalar { .. }) => { + return Err(CompileError::Static(format!( + "@{i}: declared multimap, got scalar data" + ))) + } (StaticTy::Scalar(_), StaticData::Map(_)) => { return Err(CompileError::Static(format!( "@{i}: declared scalar, got map data" @@ -513,6 +556,10 @@ fn compile_term(p: &Program, t: &Term, slots: &HashMap) -> CTerm { else_moves, } } + Term::EmitTo { to, args } => { + let (to, moves) = mk_moves(*to, args); + CTerm::EmitTo { to, moves } + } Term::Emit => CTerm::Emit, Term::Skip => CTerm::Skip, Term::Trap { msg } => CTerm::Trap(msg.clone()), @@ -2167,6 +2214,56 @@ fn compile_inst( Ok(()) }) } + Inst::ProbeRange { + static_id, + start, + end, + keys, + } => { + 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(); + 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()) + } 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); + Ok(()) + }) + } + Inst::ProbeRead { + static_id, + idx, + dsts, + } => { + let static_id = static_id as usize; + let idx = sl(slots, idx); + let dsts: Vec = dsts.iter().map(|d| sl(slots, *d)).collect(); + Box::new(move |ctx| { + 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())); + }; + for (di, v) in dsts.iter().zip(row.1.iter()) { + ctx.regs[*di] = scalar_to_reg(v, ctx.arena); + } + Ok(()) + }) + } Inst::Sload { static_id, dst } => { let (static_id, dst) = (static_id as usize, sl(slots, dst)); Box::new(move |ctx| { diff --git a/src/specializer/exec/tests.rs b/src/specializer/exec/tests.rs index 7988624..94c8f6a 100644 --- a/src/specializer/exec/tests.rs +++ b/src/specializer/exec/tests.rs @@ -817,6 +817,8 @@ fn gen_statics(rng: &mut gen::Rng, p: &Program) -> Vec { .collect(); StaticData::Map(entries) } + // The generator never declares multimaps (see gen.rs). + StaticTy::MultiMap { .. } => StaticData::Map(Vec::new()), }) .collect() } @@ -1009,3 +1011,67 @@ entry: ); } } + +#[test] +fn multimap_expand_fixture_executes() { + // Stage-B machinery end to end: dup keys fan out one output row per + // match (probe order outer, INSERTION order inner — the stable sort + // keeps equal keys in materialization order), zero matches skip, and + // the cyclic CFG verifies + roundtrips through the text format. + let p = built(fixtures::MULTI_EXPAND); + let printed = super::super::ir::print::print(&p); + assert_eq!(parse(&printed).unwrap(), p, "print/parse roundtrip"); + + // Entries deliberately unsorted; key 1's values inserted 10 then 11. + let data = StaticData::Map(vec![ + (vec![KeyBits::I64(2)], vec![ScalarVal::I64(20)]), + (vec![KeyBits::I64(1)], vec![ScalarVal::I64(10)]), + (vec![KeyBits::I64(1)], vec![ScalarVal::I64(11)]), + ]); + let f = compile(&p, vec![data]).unwrap(); + let got = run_snapshot(&f, &batch(3, vec![c_i64(&[Some(1), Some(2), Some(3)])])).unwrap(); + assert_eq!( + got, + rows(&[&["1", "10"], &["1", "11"], &["2", "20"]]), + "id=1 fans out to both matches in insertion order; id=3 skips" + ); +} + +#[test] +fn keyless_multimap_is_a_cross_join() { + // Zero-key probe.range covers the whole table — the cross/inequality + // join primitive. + let text = r#" +static @0: multimap() -> (i64) + +fn cross(in: batch{id: i64}, out: batch{id: i64, v: i64}) { +entry: + %id = load in.id + %lo, %hi = probe.range @0 + jump head(%lo, %hi, %id) +head(%i: i64, %end: i64, %rid: i64): + %more = icmp.lt %i, %end + brif %more, body(%i, %end, %rid), done +body(%j: i64, %e: i64, %rid2: i64): + %v = probe.read @0, %j + store out.id, %rid2 + store out.v, %v + %one = const.i64 1 + %j2 = iadd %j, %one + emit.to head(%j2, %e, %rid2) +done: + skip +} +"#; + let p = built(text); + let data = StaticData::Map(vec![ + (vec![], vec![ScalarVal::I64(7)]), + (vec![], vec![ScalarVal::I64(8)]), + ]); + let f = compile(&p, vec![data]).unwrap(); + let got = run_snapshot(&f, &batch(2, vec![c_i64(&[Some(1), Some(2)])])).unwrap(); + assert_eq!( + got, + rows(&[&["1", "7"], &["1", "8"], &["2", "7"], &["2", "8"]]) + ); +} diff --git a/src/specializer/ir/fixtures.rs b/src/specializer/ir/fixtures.rs index 69ad1dc..5e6435e 100644 --- a/src/specializer/ir/fixtures.rs +++ b/src/specializer/ir/fixtures.rs @@ -145,3 +145,29 @@ pub fn all() -> Vec<(&'static str, &'static str)> { ("kitchen", KITCHEN), ] } + +/// Stage-B multiplicity: a multimap probe loop — per input row, one output +/// row per matching entry (emit.to back-edge), zero matches skip. Covers: +/// multimap statics, probe.range/probe.read, emit.to, a legal CFG cycle. +pub const MULTI_EXPAND: &str = r#" +static @0: multimap(i64) -> (i64) + +fn expand(in: batch{id: i64}, out: batch{id: i64, v: i64}) { +entry: + %id = load in.id + %lo, %hi = probe.range @0, %id + jump head(%lo, %hi, %id) +head(%i: i64, %end: i64, %rid: i64): + %more = icmp.lt %i, %end + brif %more, body(%i, %end, %rid), done +body(%j: i64, %e: i64, %rid2: i64): + %v = probe.read @0, %j + store out.id, %rid2 + store out.v, %v + %one = const.i64 1 + %j2 = iadd %j, %one + emit.to head(%j2, %e, %rid2) +done: + skip +} +"#; diff --git a/src/specializer/ir/gen.rs b/src/specializer/ir/gen.rs index 0a4874c..d31d509 100644 --- a/src/specializer/ir/gen.rs +++ b/src/specializer/ir/gen.rs @@ -214,6 +214,10 @@ fn load_all( keys: key_vals, }); } + // The random-program generator doesn't emit multiplicity loops + // (stage-B programs are exercised by targeted tests instead). + StaticTy::MultiMap { .. } => { + } } } } diff --git a/src/specializer/ir/mod.rs b/src/specializer/ir/mod.rs index 083cd74..f75d9b4 100644 --- a/src/specializer/ir/mod.rs +++ b/src/specializer/ir/mod.rs @@ -179,6 +179,10 @@ pub struct Col { pub enum StaticTy { Scalar(ColTy), Map { keys: Vec, values: Vec }, + /// Stage-B (shape='many'): a map whose keys may REPEAT — probed as a + /// flat row range (ProbeRange -> [start, end), ProbeRead per index). + /// Zero keys = the keyless one-bucket join (cross/inequality). + MultiMap { keys: Vec, values: Vec }, } /// SSA value id. Presentation names are not stored; the printer derives @@ -761,6 +765,24 @@ pub enum Inst { dsts: Vec, keys: Vec, }, + /// Probe a MultiMap static: the flat row range matching `keys` + /// (`[start, end)`, both I64; empty range on a miss). Zero keys = + /// the whole table (keyless join). TOTAL. + ProbeRange { + static_id: u32, + start: Value, + end: Value, + keys: Vec, + }, + /// Read row `idx` of a MultiMap's flat value store — one dst per + /// declared value lane. `idx` MUST come from a ProbeRange of the same + /// static (verifier-checked bounds are the range's; out-of-range is a + /// program bug, trapped at run). + ProbeRead { + static_id: u32, + idx: Value, + dsts: Vec, + }, /// Read a `scalar` static. Sload { static_id: u32, @@ -812,6 +834,13 @@ pub enum Term { }, /// Output row complete; advance both cursors. Emit, + /// Stage-B: emit the completed output row AND continue at `to` (the + /// multiplicity loop's back-edge). The stored-output state resets — + /// blocks after an EmitTo store the NEXT output row. + EmitTo { + to: BlockId, + args: Vec, + }, /// Drop this input row; nothing may have been stored on this path. Skip, /// Abort the whole call with a runtime error. @@ -891,6 +920,8 @@ impl Inst { all.extend(dsts.iter().copied()); all } + Inst::ProbeRange { start, end, .. } => vec![*start, *end], + Inst::ProbeRead { dsts, .. } => dsts.clone(), Inst::Store { .. } | Inst::StoreOpt { .. } => vec![], } } @@ -911,6 +942,7 @@ impl Term { (*then_to, then_args.as_slice()), (*else_to, else_args.as_slice()), ], + Term::EmitTo { to, args } => vec![(*to, args.as_slice())], Term::Emit | Term::Skip | Term::Trap { .. } => vec![], } } @@ -1007,6 +1039,21 @@ impl Inst { *flag = m(*flag); *val = m(*val); } + Inst::ProbeRange { + start, end, keys, .. + } => { + *start = m(*start); + *end = m(*end); + for k in keys { + *k = m(*k); + } + } + Inst::ProbeRead { idx, dsts, .. } => { + *idx = m(*idx); + for d in dsts { + *d = m(*d); + } + } Inst::Probe { hit, dsts, keys, .. } => { @@ -1042,6 +1089,11 @@ impl Term { *a = m(*a); } } + Term::EmitTo { args, .. } => { + for a in args.iter_mut() { + *a = m(*a); + } + } Term::Emit | Term::Skip | Term::Trap { .. } => {} } } diff --git a/src/specializer/ir/parse.rs b/src/specializer/ir/parse.rs index f3f989c..f708f0a 100644 --- a/src/specializer/ir/parse.rs +++ b/src/specializer/ir/parse.rs @@ -378,6 +378,7 @@ struct RawBlock { enum RawTerm { Jump(String, Vec), + EmitTo(String, Vec), Brif(Value, (String, Vec), (String, Vec)), Emit, Skip, @@ -587,6 +588,10 @@ impl Parser { else_args: ea, }, RawTerm::Emit => Term::Emit, + RawTerm::EmitTo(label, args) => Term::EmitTo { + to: resolve(&label, rb.line)?, + args, + }, RawTerm::Skip => Term::Skip, RawTerm::Trap(msg) => Term::Trap { msg }, }; @@ -645,6 +650,25 @@ impl Parser { self.expect(Tok::RParen)?; Ok(StaticTy::Map { keys, values }) } + "multimap" => { + // Stage-B: duplicate keys legal; empty keys = keyless join. + self.expect(Tok::LParen)?; + let keys = if *self.peek() == Tok::RParen { + Vec::new() + } else { + self.ty_list()? + }; + 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::MultiMap { keys, values }) + } other => Err(self.err(format!("expected 'scalar' or 'map', found '{other}'"))), } } @@ -768,6 +792,15 @@ impl Parser { match kw.as_str() { "emit" => { self.bump(); + if *self.peek() == Tok::Dot { + self.bump(); + let sub = self.ident("'to' after 'emit.'")?; + if sub != "to" { + return Err(self.err(format!("unknown terminator 'emit.{sub}'"))); + } + let (label, args) = self.target()?; + return Ok(Some(RawTerm::EmitTo(label, args))); + } Ok(Some(RawTerm::Emit)) } "skip" => { @@ -1312,6 +1345,35 @@ impl Parser { keys, } } + "probe.range" => { + want_dsts(2, self)?; + let static_id = self.static_ref(statics)?; + let mut keys = Vec::new(); + while *self.peek() == Tok::Comma { + self.bump(); + keys.push(self.use_value()?); + } + Inst::ProbeRange { + static_id, + start: def!(0), + end: def!(1), + keys, + } + } + "probe.read" => { + let static_id = self.static_ref(statics)?; + self.expect(Tok::Comma)?; + let idx = self.use_value()?; + let mut dsts = Vec::with_capacity(dst_names.len()); + for i in 0..dst_names.len() { + dsts.push(def!(i)); + } + Inst::ProbeRead { + static_id, + idx, + dsts, + } + } "sload" => { want_dsts(1, self)?; let static_id = self.static_ref(statics)?; diff --git a/src/specializer/ir/print.rs b/src/specializer/ir/print.rs index ff140d9..217a005 100644 --- a/src/specializer/ir/print.rs +++ b/src/specializer/ir/print.rs @@ -19,6 +19,9 @@ pub fn print(p: &Program) -> String { StaticTy::Map { keys, values } => { let _ = writeln!(s, "map({}) -> ({})", tys(keys), tys(values)); } + StaticTy::MultiMap { keys, values } => { + let _ = writeln!(s, "multimap({}) -> ({})", tys(keys), tys(values)); + } } } for (i, re) in p.regexes.iter().enumerate() { @@ -213,6 +216,21 @@ fn print_inst(s: &mut String, p: &Program, inst: &Inst) { let ks: Vec = keys.iter().map(|k| val(*k)).collect(); let _ = write!(s, "probe @{static_id}, {}", ks.join(", ")); } + Inst::ProbeRange { + static_id, keys, .. + } => { + let ks: Vec = keys.iter().map(|k| val(*k)).collect(); + if ks.is_empty() { + let _ = write!(s, "probe.range @{static_id}"); + } else { + let _ = write!(s, "probe.range @{static_id}, {}", ks.join(", ")); + } + } + Inst::ProbeRead { + static_id, idx, .. + } => { + let _ = write!(s, "probe.read @{static_id}, {}", val(*idx)); + } Inst::Sload { static_id, .. } => { let _ = write!(s, "sload @{static_id}"); } @@ -253,6 +271,9 @@ fn print_term(s: &mut String, b: &Block) { ); } Term::Emit => s.push_str("emit"), + Term::EmitTo { to, args } => { + let _ = write!(s, "emit.to {}", target(to.0, args)); + } Term::Skip => s.push_str("skip"), Term::Trap { msg } => { let _ = write!(s, "trap {}", quote(msg)); diff --git a/src/specializer/ir/verify.rs b/src/specializer/ir/verify.rs index a6dfd73..b03c976 100644 --- a/src/specializer/ir/verify.rs +++ b/src/specializer/ir/verify.rs @@ -16,7 +16,8 @@ //! ones (the null lane can be neither skipped nor invented). //! 4. Statics: every `@N` resolves; probe/sload match the static's kind, //! arity, and types. -//! 5. CFG: all blocks reachable from entry; no cycles (v0); branch args +//! 5. CFG: all blocks reachable from entry; branch args (cycles are legal +//! since stage-B multiplicity loops; see the back-edge notes below); //! match target params in count and type. //! 6. Stores: no path stores a column twice, whatever its terminator //! (including `trap` — a double store is always a lowering bug); paths @@ -222,6 +223,18 @@ fn dst_types(p: &Program, inst: &Inst) -> Vec<(Value, Ty)> { // check reports the arity mismatch. v } + Inst::ProbeRange { start, end, .. } => vec![(*start, Ty::I64), (*end, Ty::I64)], + Inst::ProbeRead { + static_id, dsts, .. + } => { + let mut v = Vec::new(); + if let Some(StaticTy::MultiMap { values, .. }) = p.statics.get(*static_id as usize) { + for (d, ty) in dsts.iter().zip(values.iter()) { + v.push((*d, *ty)); + } + } + v + } Inst::Sload { static_id, dst } => vec![(*dst, scalar_ty(*static_id))], Inst::SloadOpt { static_id, @@ -565,10 +578,73 @@ fn check_block( ); } } + Some(StaticTy::MultiMap { .. }) => err( + errs, + Some(bi), + i, + format!("@{static_id} is a multimap: use probe.range"), + ), + }, + Inst::ProbeRange { + static_id, keys, .. + } => match p.statics.get(*static_id as usize) { + None => err(errs, Some(bi), i, format!("unknown static @{static_id}")), + Some(StaticTy::MultiMap { keys: kts, .. }) => { + if keys.len() != kts.len() { + err( + errs, + Some(bi), + i, + format!( + "@{static_id} has {} key(s), probe.range passes {}", + kts.len(), + keys.len() + ), + ); + } else { + for (k, kt) in keys.iter().zip(kts.iter()) { + want(&in_scope, def_types, *k, *kt, "probe key", bi, i, errs); + } + } + } + Some(_) => err( + errs, + Some(bi), + i, + format!("@{static_id} is not a multimap: probe.range needs one"), + ), + }, + Inst::ProbeRead { + static_id, + idx, + 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, .. }) => { + want(&in_scope, def_types, *idx, Ty::I64, "probe index", bi, i, errs); + if dsts.len() != vts.len() { + err( + errs, + Some(bi), + i, + format!( + "@{static_id} has {} value column(s), probe.read defines {}", + vts.len(), + dsts.len() + ), + ); + } + } + Some(_) => err( + errs, + Some(bi), + i, + format!("@{static_id} is not a multimap: probe.read needs one"), + ), }, 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 { .. }) => err( + Some(StaticTy::Map { .. }) | Some(StaticTy::MultiMap { .. }) => err( errs, Some(bi), i, @@ -584,7 +660,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 { .. }) => err( + Some(StaticTy::Map { .. }) | Some(StaticTy::MultiMap { .. }) => err( errs, Some(bi), i, @@ -671,12 +747,21 @@ fn check_cfg_and_stores(p: &Program, errs: &mut Vec) { return; } - // Iterative DFS for reachability + cycle detection (0 white, 1 gray, + // Iterative DFS for reachability + BACK-EDGE detection (0 white, 1 gray, // 2 black). Explicit stack, NOT recursion: a deep-but-legal CFG (large // CASE/decision-tree lowerings) must not abort the process — recursion // stack-overflowed at ~8k blocks under adversarial fuzzing. + // + // Stage-B: cycles are LEGAL (multiplicity loops jump back to their + // header via EmitTo — emit-and-continue — or a plain Jump on a + // residual-filtered iteration). Back-edges are excluded from the topo + // order below; the store dataflow stays sound because a back-edge's + // state must MATCH the header's already-known entry state (EmitTo + // propagates the RESET all-zero state; a filtered-continue Jump must + // arrive store-free), so no fixpoint iteration is ever needed. let mut color = vec![0u8; n]; - let mut cyclic = false; + let mut back_edges: std::collections::HashSet<(usize, usize)> = + std::collections::HashSet::new(); let mut stack: Vec<(usize, usize)> = vec![(0, 0)]; // (block, next successor index) color[0] = 1; while let Some(frame) = stack.last_mut() { @@ -693,7 +778,9 @@ fn check_cfg_and_stores(p: &Program, errs: &mut Vec) { color[s] = 1; stack.push((s, 0)); } - 1 => cyclic = true, + 1 => { + back_edges.insert((b, s)); + } _ => {} } } else { @@ -701,15 +788,6 @@ fn check_cfg_and_stores(p: &Program, errs: &mut Vec) { stack.pop(); } } - if cyclic { - err( - errs, - None, - None, - "control-flow cycle (v0 CFGs must be acyclic)".to_string(), - ); - return; // the store dataflow below needs a DAG - } let reachable: Vec = color.iter().map(|c| *c != 0).collect(); for (bi, r) in reachable.iter().enumerate() { if !r { @@ -717,6 +795,56 @@ fn check_cfg_and_stores(p: &Program, errs: &mut Vec) { } } + // Cycles must TERMINATE: every reachable block must reach a row-ENDING + // terminator (emit/skip/trap — emit.to continues, so it doesn't count). + // This is the old acyclicity rule relaxed exactly enough for + // multiplicity loops; a cycle with no exit still errors. + let mut preds: Vec> = vec![Vec::new(); n]; + for (bi, b) in p.blocks.iter().enumerate() { + for (succ, _) in b.term.successors() { + let s = succ.0 as usize; + if s < n { + preds[s].push(bi); + } + } + } + let mut reaches_end = vec![false; n]; + let mut work: Vec = (0..n) + .filter(|&b| { + matches!( + p.blocks[b].term, + Term::Emit | Term::Skip | Term::Trap { .. } + ) + }) + .collect(); + for &b in &work { + reaches_end[b] = true; + } + while let Some(b) = work.pop() { + for &pb in &preds[b] { + if !reaches_end[pb] { + reaches_end[pb] = true; + work.push(pb); + } + } + } + let mut nonterminating = false; + for bi in 0..n { + if reachable[bi] && !reaches_end[bi] { + nonterminating = true; + err( + errs, + Some(bi), + None, + "control-flow cycle with no path to emit/skip/trap (cannot terminate)" + .to_string(), + ); + } + } + if nonterminating { + return; // the store dataflow's topo order needs terminating loops + } + // Store dataflow over the DAG in topological order. State: per out // column, how many times it has been stored on every path reaching this // point; all paths into a join must agree. @@ -724,7 +852,7 @@ fn check_cfg_and_stores(p: &Program, errs: &mut Vec) { let mut entry_state: Vec>> = vec![None; n]; entry_state[0] = Some(vec![0; ncols]); - for bi in topo_order(p, n, &reachable) { + for bi in topo_order(p, n, &reachable, &back_edges) { let Some(state) = entry_state[bi].clone() else { continue; // unreachable; already reported }; @@ -762,6 +890,40 @@ fn check_cfg_and_stores(p: &Program, errs: &mut Vec) { } } } + Term::EmitTo { .. } => { + for (ci, count) in state_out.iter().enumerate() { + if *count == 0 { + err( + errs, + Some(bi), + None, + format!("emit without storing out.{}", p.out_cols[ci].name), + ); + } + } + // The emitted row is complete; the CONTINUATION starts the + // next output row from scratch. + let zero = vec![0u8; state_out.len()]; + for (succ, _) in p.blocks[bi].term.successors() { + let s = succ.0 as usize; + if s >= n { + continue; + } + match &entry_state[s] { + None => entry_state[s] = Some(zero.clone()), + Some(existing) if *existing != zero => { + err( + errs, + Some(s), + None, + "paths joining here disagree on which out columns are stored" + .to_string(), + ); + } + Some(_) => {} + } + } + } Term::Skip => { if state_out.iter().any(|c| *c > 0) { err( @@ -804,7 +966,12 @@ fn check_cfg_and_stores(p: &Program, errs: &mut Vec) { /// never drain) must not starve a reachable join of its dataflow visit — /// that would mask the join's store errors behind the island's /// unreachable-block errors and make them reappear one fix later. -fn topo_order(p: &Program, n: usize, reachable: &[bool]) -> Vec { +fn topo_order( + p: &Program, + n: usize, + reachable: &[bool], + back_edges: &std::collections::HashSet<(usize, usize)>, +) -> Vec { let mut indegree = vec![0usize; n]; for (bi, b) in p.blocks.iter().enumerate() { if !reachable[bi] { @@ -812,7 +979,7 @@ fn topo_order(p: &Program, n: usize, reachable: &[bool]) -> Vec { } for (succ, _) in b.term.successors() { let s = succ.0 as usize; - if s < n { + if s < n && !back_edges.contains(&(bi, s)) { indegree[s] += 1; } } @@ -825,7 +992,7 @@ fn topo_order(p: &Program, n: usize, reachable: &[bool]) -> Vec { order.push(b); for (succ, _) in p.blocks[b].term.successors() { let s = succ.0 as usize; - if s < n { + if s < n && !back_edges.contains(&(b, s)) { indegree[s] -= 1; if indegree[s] == 0 { stack.push(s); From 9089e5e38290735591c393b1f22ad43e8351633e Mon Sep 17 00:00:00 2001 From: AmirHossein Roozbahani Date: Tue, 28 Jul 2026 04:44:44 +0200 Subject: [PATCH 5/7] =?UTF-8?q?feat(specializer):=20stage-B=20loop=20lower?= =?UTF-8?q?ing=20=E2=80=94=20dup-key=20equi-joins=20serve=20under=20shape?= =?UTF-8?q?=3D'many'=20(TASK-59=20stage=20B-2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit lower() gains many: bool (one join per query in stage B, named restriction). The join lowers as a multiplicity loop: entry evaluates keys (NULL keys force an empty range) + ProbeRange; header/body iterate the equal-key run with ProbeRead lanes seeded into the block probe cache; the RESIDUAL gates match-ness (and the LEFT null-extension flag) while WHERE only gates emission — a match killed by WHERE still suppresses the null-extension, per DuckDB's join-then-filter order; the back-edge is emit.to. LEFT emits one null-extended row when nothing matched, itself WHERE-gated with default lanes. The multimap static declaration allows duplicate keys at materialization ONLY on this path; the default shapes keep the 1:1 map contract byte-identical (dup keys still error). E2E interp tests cover fan-out order, INNER drops, residual-filters-all, WHERE composition, and both named restrictions. Co-Authored-By: Claude Fable 5 --- src/duckdb/mod.rs | 20 +-- src/specializer/lower.rs | 293 +++++++++++++++++++++++++++++++++++++++ src/specializer/mod.rs | 6 +- src/specializer/tests.rs | 111 ++++++++++++++- 4 files changed, 416 insertions(+), 14 deletions(-) diff --git a/src/duckdb/mod.rs b/src/duckdb/mod.rs index 57d989e..b153e2c 100644 --- a/src/duckdb/mod.rs +++ b/src/duckdb/mod.rs @@ -516,14 +516,11 @@ impl DuckDBInferFn { // (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 many = shape.as_deref() == Some("many"); 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("many") => false, // multiplicity: `many` below Some(other) => { return Err(pyo3::exceptions::PyValueError::new_err(format!( "shape must be 'map', 'filter', or 'many', got '{other}'" @@ -662,8 +659,9 @@ impl DuckDBInferFn { } use super::specializer::PrepareError; - let prepared = - match prepare_opaque(&sql, &row_table, &in_cols, &opaque, &structs, &catalog) { + let prepared = match prepare_opaque( + &sql, &row_table, &in_cols, &opaque, &structs, &catalog, many, + ) { Ok(p) => p, // Unsupported/unparseable SQL might still be a static-tables-only // query (static driving table, aggregation, ORDER BY, DuckDB @@ -710,7 +708,9 @@ 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) { - let StaticTy::Map { keys, values } = sty else { + let (StaticTy::Map { keys, values } | StaticTy::MultiMap { keys, values }) = + sty + else { return Err(build_err("internal: v0 lowering emits only map statics")); }; let table = static_tables @@ -733,7 +733,9 @@ impl DuckDBInferFn { Err(_) => { let mut data = Vec::with_capacity(prepared.statics.len()); for (spec, sty) in prepared.statics.iter().zip(&prepared.program.statics) { - let StaticTy::Map { keys, values } = sty else { + let (StaticTy::Map { keys, values } | StaticTy::MultiMap { keys, values }) = + sty + else { return Err(build_err("internal: v0 lowering emits only map statics")); }; let table = static_tables diff --git a/src/specializer/lower.rs b/src/specializer/lower.rs index 5a68b90..a810f50 100644 --- a/src/specializer/lower.rs +++ b/src/specializer/lower.rs @@ -41,6 +41,7 @@ pub fn lower( out_cols: Vec, regexes: Vec, name: &str, + many: bool, ) -> Result { let (exprs, filter_pred) = match rel { Rel::Project { input, exprs } => match input.as_ref() { @@ -62,6 +63,35 @@ pub fn lower( let mut fb = FB::new(in_cols, joins, catalog); + // shape='many' (stage B): joins lower as multiplicity LOOPS over + // multimap row ranges — 0..N output rows per input row. One join per + // query for now; a map's key-uniqueness is unknown at prepare, so + // under 'many' every join takes the loop form. + if many && joins.len() > 1 { + return Err(PrepareError::Unsupported( + "multiple joins under shape='many' (one join per query in stage B)".to_string(), + )); + } + 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 + .iter() + .flat_map(|&c| { + let ct = catalog[joins[0].table].cols[c as usize].ty; + if ct.nullable { + vec![Ty::I1, ct.ty] + } else { + vec![ct.ty] + } + }) + .collect(), + }]; + return fb.finish(name, statics, in_cols, out_cols, regexes); + } + // Joins run before WHERE (SQL order), each in FROM order: probe, and for // INNER skip the row on a miss. A LEFT join's probe is also forced here // so its key expressions are evaluated (and can trap) for every row that @@ -1015,6 +1045,269 @@ impl<'a> FB<'a> { .collect() } + /// Stage-B loop lowering for the (single) join under shape='many': + /// + /// entry: key exprs (trap per input row), ProbeRange -> [lo, hi) + /// (NULL keys force an EMPTY range — a NULL never matches), + /// jump header(lo, hi, any=false) + /// header(i, end, any): i < end ? body(i, end, any) : after(any) + /// body: ProbeRead lanes; residual AND WHERE (3VL) gate BEFORE any + /// store (the continue path must be store-free); pass -> + /// store_blk(lanes.., i+1, end) -> stores -> emit.to header + /// (any=true); fail -> header(i+1, end, any) + /// after: INNER -> skip. LEFT -> any ? skip : null-extension row + /// (default lanes, hit=false) gated by WHERE, then emit. + /// + /// The engine's documented emission order falls out: probe rows in + /// input order, matches contiguous in build INSERTION order. + fn lower_many_loop( + &mut self, + exprs: &[(String, SExpr)], + filter_pred: Option<&SExpr>, + out_cols: &[Col], + ) -> Result<(), PrepareError> { + let j = 0u32; + let kind = self.joins[0].kind; + let residual = self.joins[0].residual.clone(); + let keys_expr = self.joins[0].keys.clone(); + let dst_tys: Vec = self.val_flat_tys(j); + + // entry: keys + range. + let mut live = Vec::new(); + let mut keys_valid: Option = None; + let mut key_vals = Vec::with_capacity(keys_expr.len()); + for key in &keys_expr { + let lane = self.emit(key, &mut live)?; + key_vals.push(lane.val); + if let Some(fl) = lane.flag { + keys_valid = self.combine_flags(keys_valid, Some(fl)); + } + } + let lo = self.fresh(); + let hi = self.fresh(); + self.inst(Inst::ProbeRange { + static_id: j, + start: lo, + end: hi, + keys: key_vals, + }); + let (lo, hi) = match keys_valid { + None => (lo, hi), + Some(fl) => { + let zero = self.const_lit(Lit::I64(0)); + let l2 = self.fresh(); + self.inst(Inst::Select { + dst: l2, + cond: fl, + a: lo, + b: zero, + }); + let h2 = self.fresh(); + self.inst(Inst::Select { + dst: h2, + cond: fl, + a: hi, + b: zero, + }); + (l2, h2) + } + }; + + let (header, hp) = self.create_block(&[Ty::I64, Ty::I64, Ty::I1]); + let f0 = self.const_i1(false); + self.term(Term::Jump { + to: BlockId(header as u32), + args: vec![lo, hi, f0], + }); + + self.switch(header); + let (h_i, h_end, h_any) = (hp[0], hp[1], hp[2]); + let more = self.fresh(); + self.inst(Inst::Cmp { + pred: CmpPred::Lt, + ty: Ty::I64, + dst: more, + a: h_i, + b: h_end, + }); + let (body, bp) = self.create_block(&[Ty::I64, Ty::I64, Ty::I1]); + let (after, ap) = self.create_block(&[Ty::I1]); + self.term(Term::Brif { + cond: more, + then_to: BlockId(body as u32), + then_args: vec![h_i, h_end, h_any], + else_to: BlockId(after as u32), + else_args: vec![h_any], + }); + + // body: lanes + gate. + self.switch(body); + let (b_i, b_end, b_any) = (bp[0], bp[1], bp[2]); + let dsts: Vec = (0..dst_tys.len()).map(|_| self.fresh()).collect(); + self.inst(Inst::ProbeRead { + static_id: j, + idx: b_i, + dsts: dsts.clone(), + }); + let t1 = self.const_i1(true); + self.blocks[self.cur].probes.insert(j, (t1, dsts.clone())); + let one = self.const_lit(Lit::I64(1)); + let inext = self.bin(BinOp::Iadd, b_i, one); + // Gate 1 — the RESIDUAL decides match-ness (and therefore `any`); + // a residual-failed candidate continues with `any` UNCHANGED. + let mut match_tys = dst_tys.clone(); + match_tys.push(Ty::I64); // inext + match_tys.push(Ty::I64); // end + let (matched_blk, mp) = self.create_block(&match_tys); + let mut match_args: Vec = dsts.clone(); + match_args.push(inext); + match_args.push(b_end); + match &residual { + None => self.term(Term::Jump { + to: BlockId(matched_blk as u32), + args: match_args, + }), + Some(res) => { + let mut live = Vec::new(); + let rl = self.emit(res, &mut live)?; + let rv = self.truthy(rl); + self.term(Term::Brif { + cond: rv, + then_to: BlockId(matched_blk as u32), + then_args: match_args, + else_to: BlockId(header as u32), + else_args: vec![inext, b_end, b_any], + }); + } + } + + // matched_blk: `any` is TRUE from here on, WHERE only gates the + // emission (DuckDB joins first, filters second — a match killed by + // WHERE still suppresses the LEFT null-extension). + self.switch(matched_blk); + let m_dsts: Vec = mp[..dst_tys.len()].to_vec(); + let (m_inext, m_end) = (mp[dst_tys.len()], mp[dst_tys.len() + 1]); + let tm = self.const_i1(true); + self.blocks[self.cur].probes.insert(j, (tm, m_dsts.clone())); + let mut store_tys = dst_tys.clone(); + store_tys.push(Ty::I64); // inext + store_tys.push(Ty::I64); // end + let (store_blk, sp) = self.create_block(&store_tys); + let mut store_args: Vec = m_dsts.clone(); + store_args.push(m_inext); + store_args.push(m_end); + match filter_pred { + None => self.term(Term::Jump { + to: BlockId(store_blk as u32), + args: store_args, + }), + Some(pred) => { + let mut live = Vec::new(); + let pl = self.emit(pred, &mut live)?; + let pv = self.truthy(pl); + let tt = self.const_i1(true); + self.term(Term::Brif { + cond: pv, + then_to: BlockId(store_blk as u32), + then_args: store_args, + else_to: BlockId(header as u32), + else_args: vec![m_inext, m_end, tt], + }); + } + } + + // store_blk: seeded lanes -> stores -> emit.to header(any=true). + self.switch(store_blk); + let s_dsts: Vec = sp[..dst_tys.len()].to_vec(); + let (s_inext, s_end) = (sp[dst_tys.len()], sp[dst_tys.len() + 1]); + let t2 = self.const_i1(true); + self.blocks[self.cur].probes.insert(j, (t2, s_dsts)); + self.store_out_row(exprs, out_cols)?; + let t3 = self.const_i1(true); + self.term(Term::EmitTo { + to: BlockId(header as u32), + args: vec![s_inext, s_end, t3], + }); + + // after: LEFT null-extension or skip. + self.switch(after); + let a_any = ap[0]; + if kind == JoinKind::Left { + let (done, _) = self.create_block(&[]); + let (miss, _) = self.create_block(&[]); + self.term(Term::Brif { + cond: a_any, + then_to: BlockId(done as u32), + then_args: vec![], + else_to: BlockId(miss as u32), + else_args: vec![], + }); + self.switch(miss); + if let Some(pred) = filter_pred { + // WHERE sees the null-extended row too (measured). + let fmiss = self.const_i1(false); + let defaults: Vec = + dst_tys.iter().map(|&ty| self.default_of(ty)).collect(); + self.blocks[self.cur].probes.insert(j, (fmiss, defaults)); + let mut live = Vec::new(); + let pl = self.emit(pred, &mut live)?; + let pv = self.truthy(pl); + let (keep, _) = self.create_block(&[]); + let (drop, _) = self.create_block(&[]); + self.term(Term::Brif { + cond: pv, + then_to: BlockId(keep as u32), + then_args: vec![], + else_to: BlockId(drop as u32), + else_args: vec![], + }); + self.switch(drop); + self.term(Term::Skip); + self.switch(keep); + } + let fmiss2 = self.const_i1(false); + let defaults2: Vec = dst_tys.iter().map(|&ty| self.default_of(ty)).collect(); + self.blocks[self.cur].probes.insert(j, (fmiss2, defaults2)); + self.store_out_row(exprs, out_cols)?; + self.term(Term::Emit); + self.switch(done); + self.term(Term::Skip); + } else { + self.term(Term::Skip); + } + Ok(()) + } + + /// Emit every output expression and store it (the shared tail of both + /// lowerings) — the current block's caches must already be seeded. + fn store_out_row( + &mut self, + exprs: &[(String, SExpr)], + out_cols: &[Col], + ) -> Result<(), PrepareError> { + let mut live = Vec::new(); + for (ci, (_, e)) in exprs.iter().enumerate() { + debug_assert!(live.is_empty()); + let lane = self.emit(e, &mut live)?; + let col = ci as u32; + if out_cols[ci].ty.nullable { + let flag = match lane.flag { + Some(fl) => fl, + None => self.const_i1(true), + }; + self.inst(Inst::StoreOpt { + col, + flag, + val: lane.val, + }); + } else { + debug_assert!(lane.flag.is_none(), "non-nullable column with a flag lane"); + self.inst(Inst::Store { col, val: lane.val }); + } + } + Ok(()) + } + fn emit_probe(&mut self, j: u32, live: &mut Live) -> Result<(Value, Vec), PrepareError> { if let Some((valid_hit, dsts)) = self.blocks[self.cur].probes.get(&j) { return Ok((*valid_hit, dsts.clone())); diff --git a/src/specializer/mod.rs b/src/specializer/mod.rs index 65b3d80..d427b79 100644 --- a/src/specializer/mod.rs +++ b/src/specializer/mod.rs @@ -60,7 +60,7 @@ pub fn prepare( in_cols: &[ir::Col], statics: &[plan::StaticTable], ) -> Result { - prepare_opaque(sql, this_name, in_cols, &[], &[], statics) + prepare_opaque(sql, this_name, in_cols, &[], &[], statics, false) } /// [`prepare`] plus the row-model columns that have no plain scalar lane: @@ -75,11 +75,13 @@ pub fn prepare_opaque( opaque: &[(usize, String)], structs: &[plan::StructCol], statics: &[plan::StaticTable], + many: bool, ) -> Result { 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")?; + let mut program = + lower::lower(&rel, &joins, statics, in_cols, out_cols, regexes, "run", many)?; // Block-splitting lowerings mint ids out of text order; renumber so // every prepared program is exactly canonical (parse(print(p)) == p). ir::canonicalize(&mut program); diff --git a/src/specializer/tests.rs b/src/specializer/tests.rs index c3a3881..46dd788 100644 --- a/src/specializer/tests.rs +++ b/src/specializer/tests.rs @@ -1724,7 +1724,7 @@ fn opaque_row_columns_reject_only_on_reference() { let schema = cols(&[("a", Ty::I64, false), ("s", Ty::Str, true)]); let opaque = vec![(1usize, "d".to_string())]; let prep_o = |sql: &str| { - super::prepare_opaque(sql, "__THIS__", &schema, &opaque, &[], &[]).map(|p| p.program) + super::prepare_opaque(sql, "__THIS__", &schema, &opaque, &[], &[], false).map(|p| p.program) }; // Untouched -> serves end to end. @@ -2035,7 +2035,7 @@ fn structs_flatten_to_lanes() { ], }]; let prep_s = |sql: &str| { - super::prepare_opaque(sql, "__THIS__", &schema, &[], &structs, &[]).map(|p| p.program) + super::prepare_opaque(sql, "__THIS__", &schema, &[], &structs, &[], false).map(|p| p.program) }; let run = |sql: &str| -> Result>, String> { let p = prep_s(sql).map_err(|e| e.to_string())?; @@ -2144,7 +2144,7 @@ fn nested_struct_resolution_matches_pins() { ))))), }]; let prep_s = |sql: &str| { - super::prepare_opaque(sql, "t", &schema, &[], &structs, &[]).map(|p| p.program) + super::prepare_opaque(sql, "t", &schema, &[], &structs, &[], false).map(|p| p.program) }; // 8 parts = schema.table.column + 5 field extracts (corpus verbatim). let p = prep_s("SELECT t.t.t.t.t.t.t.t FROM t.t").unwrap(); @@ -2168,3 +2168,108 @@ fn nested_struct_resolution_matches_pins() { let p = prep_s("SELECT z.t.t.t.t.t.t FROM t.t AS z").unwrap(); assert_eq!(p.out_cols[0].name, "t"); } + +#[test] +fn many_shape_dup_key_joins_fan_out() { + // Stage-B loop lowering (pins-stageB): per-pair emission in probe + // order outer / build INSERTION order inner; LEFT null-extends + // zero-match rows (NULL keys and residual-filters-all included); + // WHERE composes per emitted candidate, null-extension included. + let schema = cols(&[("pid", Ty::I64, true)]); + let dim = stat("d", &[("id", Ty::I64, false), ("v", Ty::I64, false)]); + let prep_many = |sql: &str| { + super::prepare_opaque( + sql, + "__THIS__", + &schema, + &[], + &[], + std::slice::from_ref(&dim), + true, + ) + }; + let data = || { + StaticData::Map(vec![ + (vec![KeyBits::I64(1)], vec![ScalarVal::I64(10)]), + (vec![KeyBits::I64(2)], vec![ScalarVal::I64(20)]), + (vec![KeyBits::I64(1)], vec![ScalarVal::I64(11)]), + ]) + }; + let input = || batch(4, vec![c_i64(&[Some(1), Some(2), Some(3), None])]); + let run_many = |sql: &str| -> Vec> { + let p = prep_many(sql).unwrap(); + let f = compile(&p.program, vec![data()]).unwrap(); + run_snapshot(&f, &input()).unwrap() + }; + + // LEFT: dup-key fan-out + null-extension for the miss and NULL key. + assert_eq!( + run_many("SELECT pid, v FROM __THIS__ LEFT JOIN d ON pid = d.id"), + rows(&[ + &["1", "10"], + &["1", "11"], + &["2", "20"], + &["3", "NULL"], + &["NULL", "NULL"], + ]) + ); + // INNER: zero-match rows drop. + assert_eq!( + run_many("SELECT pid, v FROM __THIS__ JOIN d ON pid = d.id"), + rows(&[&["1", "10"], &["1", "11"], &["2", "20"]]) + ); + // Residual filters PER MATCH; a row whose matches are all filtered + // still null-extends (measured). + assert_eq!( + run_many("SELECT pid, v FROM __THIS__ LEFT JOIN d ON pid = d.id AND d.v > 10"), + rows(&[&["1", "11"], &["2", "20"], &["3", "NULL"], &["NULL", "NULL"]]) + ); + assert_eq!( + run_many("SELECT pid, v FROM __THIS__ LEFT JOIN d ON pid = d.id AND d.v > 100"), + rows(&[ + &["1", "NULL"], + &["2", "NULL"], + &["3", "NULL"], + &["NULL", "NULL"], + ]) + ); + // WHERE applies to every emitted candidate incl. the null-extension. + assert_eq!( + run_many("SELECT pid, v FROM __THIS__ LEFT JOIN d ON pid = d.id WHERE v IS NULL"), + rows(&[&["3", "NULL"], &["NULL", "NULL"]]) + ); + assert_eq!( + run_many("SELECT pid, v FROM __THIS__ LEFT JOIN d ON pid = d.id WHERE v > 10"), + rows(&[&["1", "11"], &["2", "20"]]) + ); + // Under the DEFAULT shape the same dup-key data still errors at + // materialization (the 1:1 map contract is untouched). + let p = prepare( + "SELECT pid, v FROM __THIS__ LEFT JOIN d ON pid = d.id", + "__THIS__", + &schema, + std::slice::from_ref(&dim), + ) + .unwrap(); + let e = match compile(&p.program, vec![data()]) { + Err(e) => e.to_string(), + Ok(_) => panic!("dup keys under the default shape must error"), + }; + assert!(e.contains("duplicate map key"), "{e}"); + // Multi-join under 'many' is the named stage-B restriction. + let dim_b = stat("d", &[("id", Ty::I64, false), ("v", Ty::I64, false)]); + let dim2 = stat("d2", &[("id", Ty::I64, false), ("w", Ty::I64, false)]); + let e = match super::prepare_opaque( + "SELECT pid FROM __THIS__ LEFT JOIN d ON pid = d.id LEFT JOIN d2 ON pid = d2.id", + "__THIS__", + &schema, + &[], + &[], + &[dim_b, dim2], + true, + ) { + Err(e) => e.to_string(), + Ok(_) => panic!("multi-join under many must be the named restriction"), + }; + assert!(e.contains("one join per query"), "{e}"); +} From d8a21940e98e590139dc623df6f6ca1e0772a3ae Mon Sep 17 00:00:00 2001 From: AmirHossein Roozbahani Date: Tue, 28 Jul 2026 04:51:36 +0200 Subject: [PATCH 6/7] =?UTF-8?q?feat(specializer):=20stage-B=20live-stack?= =?UTF-8?q?=20loop=20lowering=20+=20corpus=20harness=20retry=20=E2=80=94?= =?UTF-8?q?=20529=20->=20546,=200=20FAIL=20(TASK-59=20stage=20B-3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Predicate/expression emission may SPLIT blocks (CASE machinery), so the multiplicity loop now rides its state and probe lanes on the LIVE stack (auto-rebound across splits; lanes are always the trailing entries) and re-seeds the per-block probe cache before every emission that can reference join columns — fixes the three corpus binder failures. The keyless path needed nothing: wave-4's empty-key binding already covers cross joins, inequality ON, and constant ON under 'many' (test added). Corpus replay retries 'duplicate map key' rejections with shape='many' (the default's rejection stays proven); fn.shape now reports 'many'. Corpus: 546 match / 132 clean-unsupported / 0 FAIL of 678 (was 529 — the full dup-key + cross/inequality bucket). Self-joins are B-4. Co-Authored-By: Claude Fable 5 --- src/duckdb/mod.rs | 23 ++-- src/specializer/lower.rs | 207 ++++++++++++++++++++---------------- src/specializer/tests.rs | 64 +++++++++++ tests/test_corpus_replay.py | 23 +++- 4 files changed, 218 insertions(+), 99 deletions(-) diff --git a/src/duckdb/mod.rs b/src/duckdb/mod.rs index b153e2c..e8a5570 100644 --- a/src/duckdb/mod.rs +++ b/src/duckdb/mod.rs @@ -495,7 +495,8 @@ pub struct DuckDBInferFn { #[pyo3(get)] output_model: Py, output_dicts: bool, - shape_map: bool, + /// 0 = filter, 1 = map, 2 = many (the declared row-shape contract). + shape_kind: u8, } #[pymethods] @@ -517,6 +518,12 @@ impl DuckDBInferFn { // join multiplicity (stage B) and is the only shape under which // those constructs will ever build. let many = shape.as_deref() == Some("many"); + let shape_kind: u8 = match shape.as_deref() { + None | Some("filter") => 0, + Some("map") => 1, + Some("many") => 2, + Some(_) => 0, // rejected below + }; let strict_map = match shape.as_deref() { None | Some("filter") => false, Some("map") => true, @@ -689,7 +696,7 @@ impl DuckDBInferFn { row_table, output_model, output_dicts, - shape_map: false, + shape_kind, }); } Err(_) => return Err(build_err(e.to_string())), @@ -781,17 +788,17 @@ impl DuckDBInferFn { row_table, output_model, output_dicts, - shape_map: strict_map, + shape_kind, }) } - /// The row-shape contract: "map" (proven exactly-1) or "filter" (0..1). + /// The declared row-shape contract: "map", "filter", or "many". #[getter] fn shape(&self) -> &'static str { - if self.shape_map { - "map" - } else { - "filter" + match self.shape_kind { + 1 => "map", + 2 => "many", + _ => "filter", } } diff --git a/src/specializer/lower.rs b/src/specializer/lower.rs index a810f50..175f65d 100644 --- a/src/specializer/lower.rs +++ b/src/specializer/lower.rs @@ -1050,16 +1050,19 @@ impl<'a> FB<'a> { /// entry: key exprs (trap per input row), ProbeRange -> [lo, hi) /// (NULL keys force an EMPTY range — a NULL never matches), /// jump header(lo, hi, any=false) - /// header(i, end, any): i < end ? body(i, end, any) : after(any) - /// body: ProbeRead lanes; residual AND WHERE (3VL) gate BEFORE any - /// store (the continue path must be store-free); pass -> - /// store_blk(lanes.., i+1, end) -> stores -> emit.to header - /// (any=true); fail -> header(i+1, end, any) + /// header(i, end, any): i < end ? body : after(any) + /// body: ProbeRead lanes; the RESIDUAL gates match-ness (and the + /// `any` flag); WHERE only gates emission (DuckDB joins + /// first, filters second); pass -> stores -> emit.to header + /// (any=true); fail -> continue. /// after: INNER -> skip. LEFT -> any ? skip : null-extension row /// (default lanes, hit=false) gated by WHERE, then emit. /// - /// The engine's documented emission order falls out: probe rows in - /// input order, matches contiguous in build INSERTION order. + /// Predicate/expression emission may SPLIT blocks (CASE machinery), so + /// the loop state and probe lanes ride the LIVE stack (auto-rebound + /// across splits) with the invariant: the probe lanes are always the + /// LAST `nd` live entries; the per-block probe cache is re-seeded from + /// them before every emission that can reference join columns. fn lower_many_loop( &mut self, exprs: &[(String, SExpr)], @@ -1071,9 +1074,10 @@ impl<'a> FB<'a> { let residual = self.joins[0].residual.clone(); let keys_expr = self.joins[0].keys.clone(); let dst_tys: Vec = self.val_flat_tys(j); + let nd = dst_tys.len(); // entry: keys + range. - let mut live = Vec::new(); + let mut live: Live = Vec::new(); let mut keys_valid: Option = None; let mut key_vals = Vec::with_capacity(keys_expr.len()); for key in &keys_expr { @@ -1140,93 +1144,99 @@ impl<'a> FB<'a> { else_args: vec![h_any], }); - // body: lanes + gate. + // body: lanes + the two gates, everything riding live as + // [inext, end, any, dsts...]. self.switch(body); let (b_i, b_end, b_any) = (bp[0], bp[1], bp[2]); - let dsts: Vec = (0..dst_tys.len()).map(|_| self.fresh()).collect(); + let dsts: Vec = (0..nd).map(|_| self.fresh()).collect(); self.inst(Inst::ProbeRead { static_id: j, idx: b_i, dsts: dsts.clone(), }); - let t1 = self.const_i1(true); - self.blocks[self.cur].probes.insert(j, (t1, dsts.clone())); let one = self.const_lit(Lit::I64(1)); let inext = self.bin(BinOp::Iadd, b_i, one); - // Gate 1 — the RESIDUAL decides match-ness (and therefore `any`); - // a residual-failed candidate continues with `any` UNCHANGED. - let mut match_tys = dst_tys.clone(); - match_tys.push(Ty::I64); // inext - match_tys.push(Ty::I64); // end + let mut live: Live = Vec::new(); + live.push((Lane { flag: None, val: inext }, Ty::I64)); + live.push((Lane { flag: None, val: b_end }, Ty::I64)); + live.push((Lane { flag: None, val: b_any }, Ty::I1)); + for (&d, &ty) in dsts.iter().zip(dst_tys.iter()) { + live.push((Lane { flag: None, val: d }, ty)); + } + let rv = match &residual { + None => None, + Some(res) => { + self.reseed_many(j, true, &live, nd); + let rl = self.emit(res, &mut live)?; + Some(self.truthy(rl)) + } + }; + let vals: Vec = live.iter().map(|(l, _)| l.val).collect(); + let (r_inext, r_end, r_any) = (vals[0], vals[1], vals[2]); + let r_dsts: Vec = vals[3..].to_vec(); + live.clear(); + let mut match_tys = vec![Ty::I64, Ty::I64]; + match_tys.extend(dst_tys.iter().copied()); let (matched_blk, mp) = self.create_block(&match_tys); - let mut match_args: Vec = dsts.clone(); - match_args.push(inext); - match_args.push(b_end); - match &residual { + let mut match_args = vec![r_inext, r_end]; + match_args.extend(r_dsts.iter().copied()); + match rv { None => self.term(Term::Jump { to: BlockId(matched_blk as u32), args: match_args, }), - Some(res) => { - let mut live = Vec::new(); - let rl = self.emit(res, &mut live)?; - let rv = self.truthy(rl); - self.term(Term::Brif { - cond: rv, - then_to: BlockId(matched_blk as u32), - then_args: match_args, - else_to: BlockId(header as u32), - else_args: vec![inext, b_end, b_any], - }); - } + Some(cond) => self.term(Term::Brif { + cond, + then_to: BlockId(matched_blk as u32), + then_args: match_args, + else_to: BlockId(header as u32), + else_args: vec![r_inext, r_end, r_any], + }), } - // matched_blk: `any` is TRUE from here on, WHERE only gates the - // emission (DuckDB joins first, filters second — a match killed by - // WHERE still suppresses the LEFT null-extension). + // matched_blk: `any` is TRUE from here on; WHERE gates emission + // only. live = [inext, end, dsts...]. self.switch(matched_blk); - let m_dsts: Vec = mp[..dst_tys.len()].to_vec(); - let (m_inext, m_end) = (mp[dst_tys.len()], mp[dst_tys.len() + 1]); - let tm = self.const_i1(true); - self.blocks[self.cur].probes.insert(j, (tm, m_dsts.clone())); - let mut store_tys = dst_tys.clone(); - store_tys.push(Ty::I64); // inext - store_tys.push(Ty::I64); // end - let (store_blk, sp) = self.create_block(&store_tys); - let mut store_args: Vec = m_dsts.clone(); - store_args.push(m_inext); - store_args.push(m_end); - match filter_pred { - None => self.term(Term::Jump { - to: BlockId(store_blk as u32), - args: store_args, - }), - Some(pred) => { - let mut live = Vec::new(); - let pl = self.emit(pred, &mut live)?; - let pv = self.truthy(pl); - let tt = self.const_i1(true); - self.term(Term::Brif { - cond: pv, - then_to: BlockId(store_blk as u32), - then_args: store_args, - else_to: BlockId(header as u32), - else_args: vec![m_inext, m_end, tt], - }); + let mut live: Live = Vec::new(); + live.push((Lane { flag: None, val: mp[0] }, Ty::I64)); + live.push((Lane { flag: None, val: mp[1] }, Ty::I64)); + for (&d, &ty) in mp[2..].iter().zip(dst_tys.iter()) { + live.push((Lane { flag: None, val: d }, ty)); + } + if let Some(pred) = filter_pred { + self.reseed_many(j, true, &live, nd); + let pl = self.emit(pred, &mut live)?; + let pv = self.truthy(pl); + let vals: Vec = live.iter().map(|(l, _)| l.val).collect(); + let (keep, kp) = { + let mut tys = vec![Ty::I64, Ty::I64]; + tys.extend(dst_tys.iter().copied()); + self.create_block(&tys) + }; + let mut keep_args = vec![vals[0], vals[1]]; + keep_args.extend(vals[2..].iter().copied()); + let tt = self.const_i1(true); + self.term(Term::Brif { + cond: pv, + then_to: BlockId(keep as u32), + then_args: keep_args, + else_to: BlockId(header as u32), + else_args: vec![vals[0], vals[1], tt], + }); + self.switch(keep); + live.clear(); + live.push((Lane { flag: None, val: kp[0] }, Ty::I64)); + live.push((Lane { flag: None, val: kp[1] }, Ty::I64)); + for (&d, &ty) in kp[2..].iter().zip(dst_tys.iter()) { + live.push((Lane { flag: None, val: d }, ty)); } } - - // store_blk: seeded lanes -> stores -> emit.to header(any=true). - self.switch(store_blk); - let s_dsts: Vec = sp[..dst_tys.len()].to_vec(); - let (s_inext, s_end) = (sp[dst_tys.len()], sp[dst_tys.len() + 1]); - let t2 = self.const_i1(true); - self.blocks[self.cur].probes.insert(j, (t2, s_dsts)); - self.store_out_row(exprs, out_cols)?; + self.store_out_row(exprs, out_cols, Some((j, true)), &mut live, nd)?; + let vals: Vec = live.iter().map(|(l, _)| l.val).collect(); let t3 = self.const_i1(true); self.term(Term::EmitTo { to: BlockId(header as u32), - args: vec![s_inext, s_end, t3], + args: vec![vals[0], vals[1], t3], }); // after: LEFT null-extension or skip. @@ -1243,32 +1253,36 @@ impl<'a> FB<'a> { else_args: vec![], }); self.switch(miss); + // live = [dsts(defaults)...] only. + let mut live: Live = Vec::new(); + for &ty in dst_tys.iter() { + let d = self.default_of(ty); + live.push((Lane { flag: None, val: d }, ty)); + } if let Some(pred) = filter_pred { // WHERE sees the null-extended row too (measured). - let fmiss = self.const_i1(false); - let defaults: Vec = - dst_tys.iter().map(|&ty| self.default_of(ty)).collect(); - self.blocks[self.cur].probes.insert(j, (fmiss, defaults)); - let mut live = Vec::new(); + self.reseed_many(j, false, &live, nd); let pl = self.emit(pred, &mut live)?; let pv = self.truthy(pl); - let (keep, _) = self.create_block(&[]); + let vals: Vec = live.iter().map(|(l, _)| l.val).collect(); + let (keep, kp) = self.create_block(&dst_tys); let (drop, _) = self.create_block(&[]); self.term(Term::Brif { cond: pv, then_to: BlockId(keep as u32), - then_args: vec![], + then_args: vals, else_to: BlockId(drop as u32), else_args: vec![], }); self.switch(drop); self.term(Term::Skip); self.switch(keep); + live.clear(); + for (&d, &ty) in kp.iter().zip(dst_tys.iter()) { + live.push((Lane { flag: None, val: d }, ty)); + } } - let fmiss2 = self.const_i1(false); - let defaults2: Vec = dst_tys.iter().map(|&ty| self.default_of(ty)).collect(); - self.blocks[self.cur].probes.insert(j, (fmiss2, defaults2)); - self.store_out_row(exprs, out_cols)?; + self.store_out_row(exprs, out_cols, Some((j, false)), &mut live, nd)?; self.term(Term::Emit); self.switch(done); self.term(Term::Skip); @@ -1278,17 +1292,32 @@ impl<'a> FB<'a> { Ok(()) } - /// Emit every output expression and store it (the shared tail of both - /// lowerings) — the current block's caches must already be seeded. + /// Re-seed the CURRENT block's probe cache for many-join `j` from the + /// live stack's trailing `nd` lanes (the invariant of the loop + /// lowering) with a fresh hit constant. + fn reseed_many(&mut self, j: u32, hit: bool, live: &Live, nd: usize) { + let h = self.const_i1(hit); + let ds: Vec = live[live.len() - nd..].iter().map(|(l, _)| l.val).collect(); + self.blocks[self.cur].probes.insert(j, (h, ds)); + } + + /// Emit every output expression and store it. Under a many-join + /// (`seed` present) the probe cache is re-seeded from `live`'s + /// trailing `nd` lanes before EVERY expression — emission may split + /// blocks, and each new block starts with an empty cache. fn store_out_row( &mut self, exprs: &[(String, SExpr)], out_cols: &[Col], + seed: Option<(u32, bool)>, + live: &mut Live, + nd: usize, ) -> Result<(), PrepareError> { - let mut live = Vec::new(); for (ci, (_, e)) in exprs.iter().enumerate() { - debug_assert!(live.is_empty()); - let lane = self.emit(e, &mut live)?; + if let Some((j, hit)) = seed { + self.reseed_many(j, hit, live, nd); + } + let lane = self.emit(e, live)?; let col = ci as u32; if out_cols[ci].ty.nullable { let flag = match lane.flag { diff --git a/src/specializer/tests.rs b/src/specializer/tests.rs index 46dd788..31e5ed6 100644 --- a/src/specializer/tests.rs +++ b/src/specializer/tests.rs @@ -2273,3 +2273,67 @@ fn many_shape_dup_key_joins_fan_out() { }; assert!(e.contains("one join per query"), "{e}"); } + +#[test] +fn many_shape_keyless_and_inequality_joins() { + // pins-stageB/cross-inequality.json: keyless joins are cross-product- + // then-filter; LEFT null-extends zero-match rows; ON NULL = 2 matches + // nothing. All ride the empty-key multimap (whole-table range). + let schema = cols(&[("pid", Ty::I64, false)]); + let dim = stat("d", &[("id", Ty::I64, false)]); + let prep_many = |sql: &str| { + super::prepare_opaque( + sql, + "__THIS__", + &schema, + &[], + &[], + std::slice::from_ref(&dim), + true, + ) + }; + let data = || { + StaticData::Map(vec![ + (vec![], vec![ScalarVal::I64(2)]), + (vec![], vec![ScalarVal::I64(3)]), + (vec![], vec![ScalarVal::I64(4)]), + ]) + }; + let input = || batch(3, vec![c_i64(&[Some(1), Some(2), Some(3)])]); + let run_many = |sql: &str| -> Result>, String> { + let p = prep_many(sql).map_err(|e| e.to_string())?; + let f = compile(&p.program, vec![data()]).map_err(|e| e.to_string())?; + run_snapshot(&f, &input()).map_err(|e| e.to_string()) + }; + + // Plain comma cross join: 3x3. + assert_eq!( + run_many("SELECT pid, id FROM __THIS__, d").unwrap(), + rows(&[ + &["1", "2"], + &["1", "3"], + &["1", "4"], + &["2", "2"], + &["2", "3"], + &["2", "4"], + &["3", "2"], + &["3", "3"], + &["3", "4"], + ]) + ); + // Inequality INNER: cross + filter. + assert_eq!( + run_many("SELECT pid, id FROM __THIS__ JOIN d ON pid > d.id").unwrap(), + rows(&[&["3", "2"]]) + ); + // Inequality LEFT: null-extension for rows with no match. + assert_eq!( + run_many("SELECT pid, id FROM __THIS__ LEFT JOIN d ON pid > d.id").unwrap(), + rows(&[&["1", "NULL"], &["2", "NULL"], &["3", "2"]]) + ); + // Constant-NULL ON: matches nothing; LEFT null-extends everything. + assert_eq!( + run_many("SELECT pid, id FROM __THIS__ LEFT JOIN d ON NULL = 2").unwrap(), + rows(&[&["1", "NULL"], &["2", "NULL"], &["3", "NULL"]]) + ); +} diff --git a/tests/test_corpus_replay.py b/tests/test_corpus_replay.py index 376edae..748c211 100644 --- a/tests/test_corpus_replay.py +++ b/tests/test_corpus_replay.py @@ -158,9 +158,28 @@ def _replay(case: dict) -> tuple[str, str]: ) except Exception as e: # noqa: BLE001 -- classification, not control flow msg = str(e) - if any(n in msg for n in _CLEAN): + # Stage-B (TASK-59): multiplicity constructs build only under the + # opt-in shape='many'. The retry keeps proving the DEFAULT rejects + # 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: + try: + fn = DuckDBInferFn( + case["sql"], + row_tables={driving: model}, + static_tables=statics, + shape="many", + ) + except Exception as e2: # noqa: BLE001 + msg2 = str(e2) + if any(n in msg2 for n in _CLEAN): + return "unsupported", msg2 + return "FAIL", f"build error under shape='many': {msg2}" + elif any(n in msg for n in _CLEAN): return "unsupported", msg - return "FAIL", f"build error: {type(e).__name__}: {msg}" + else: + return "FAIL", f"build error: {type(e).__name__}: {msg}" if case.get("source") in _KNOWN_DIVERGENT_SOURCES: return "unsupported", "known oracle divergence (see _KNOWN_DIVERGENT_SOURCES)" From 3172a457dd1a28cf42ea556cbaeca9be5f4ab2d1 Mon Sep 17 00:00:00 2001 From: AmirHossein Roozbahani Date: Tue, 28 Jul 2026 04:54:35 +0200 Subject: [PATCH 7/7] =?UTF-8?q?feat(specializer):=20stage-B=20contract=20s?= =?UTF-8?q?urface=20=E2=80=94=20oracle=20tests,=20limitations=20doc/twin,?= =?UTF-8?q?=20corpus=20546=20(TASK-59=20stage=20B-5)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - tests/test_duckdb_stageb_many.py: sorted-multiset oracle tests (dup-key fan-out, residual/WHERE combos, cross + inequality + constant ON) plus the engine's OWN order contract pinned as a test. - known-limitations §1/§2: dup-key/cross/inequality joins now documented as serving under the shape='many' opt-in (multiset parity, engine- defined order); self-join row marked in-progress; corpus count 546. - Twin + shape tests flipped: dup keys serve under many, reject under the defaults; fn.shape == 'many'. Full gate: cargo 167, pytest 616 + 13 xfail, corpus 546/0 FAIL. Co-Authored-By: Claude Fable 5 --- docs/known-limitations.md | 15 +++--- tests/test_duckdb_stageb_many.py | 80 ++++++++++++++++++++++++++++++++ tests/test_known_limitations.py | 12 ++++- tests/test_shape_contract.py | 16 +++++-- 4 files changed, 112 insertions(+), 11 deletions(-) create mode 100644 tests/test_duckdb_stageb_many.py diff --git a/docs/known-limitations.md b/docs/known-limitations.md index e4bb3ea..3282e5b 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: 505 of 678 statements as of wave B), or + against DuckDB's own test corpus: 546 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. @@ -30,8 +30,8 @@ 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, with unique keys | `duplicate map key` | Joins are frozen hash maps baked into the function. Duplicate keys mean 1:N join multiplicity — a designed extension (stage B, TASK-50 notes) that is deliberately not built yet. NULL *values* serve since TASK-55 (they flow through as NULL); NULL *keys* drop the row, matching equi-join semantics. | -| 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 needs stage-B machinery. | +| 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'`). | | 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) @@ -42,10 +42,11 @@ 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. +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). 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/tests/test_duckdb_stageb_many.py b/tests/test_duckdb_stageb_many.py new file mode 100644 index 0000000..892da4e --- /dev/null +++ b/tests/test_duckdb_stageb_many.py @@ -0,0 +1,80 @@ +"""Stage-B multiplicity vs the duckdb oracle (TASK-59), multiset parity. + +Pins: docs/superpowers/specs/2026-07-28-stageB-multiplicity-pins.md — +DuckDB's join output ORDER is a hash-join accident, so comparison is +SORTED; the engine's own order (probe outer, insertion inner) is a +documented contract of its own. +""" + +from __future__ import annotations + +import duckdb +import pyarrow as pa +from pydantic import create_model + +from sql_transform._interpreter import DuckDBInferFn + +T = create_model("T", pid=(int | None, None)) +ROWS = [{"pid": 1}, {"pid": 2}, {"pid": 3}, {"pid": None}] +DIM = pa.table({"id": [1, 2, 1, 2, 1], "v": ["a", "b", "c", "d", "e"]}) + + +def _many_check(sql: str): + """Engine (shape='many') vs DuckDB, sorted-row multiset.""" + fn = DuckDBInferFn( + sql.replace("__THIS__", "__THIS__"), + row_tables={"__THIS__": T}, + static_tables={"d": DIM}, + output="dict", + shape="many", + ) + got = [tuple(r.values()) for r in fn.infer({"__THIS__": [T(**r) for r in ROWS]})] + + con = duckdb.connect() + con.execute("CREATE TABLE __THIS__ (pid BIGINT)") + for r in ROWS: + con.execute("INSERT INTO __THIS__ VALUES (?)", [r["pid"]]) + con.register("__arrow_d", DIM) + con.execute('CREATE TABLE d AS SELECT * FROM "__arrow_d"') + want = con.execute(sql).fetchall() + key = lambda t: tuple((x is None, x) for x in t) # noqa: E731 + assert sorted(got, key=key) == sorted(want, key=key), f"{sql}\n{got}\n{want}" + + +def test_dup_key_fanout_vs_oracle(): + _many_check("SELECT pid, v FROM __THIS__ JOIN d ON pid = d.id") + _many_check("SELECT pid, v FROM __THIS__ LEFT JOIN d ON pid = d.id") + _many_check("SELECT pid, v FROM __THIS__ LEFT JOIN d ON pid = d.id AND d.v > 'b'") + _many_check("SELECT pid, v FROM __THIS__ LEFT JOIN d ON pid = d.id WHERE v IS NULL") + _many_check("SELECT upper(v) AS u FROM __THIS__ JOIN d ON pid = d.id WHERE pid > 1") + + +def test_cross_and_inequality_vs_oracle(): + _many_check("SELECT pid, id, v FROM __THIS__, d") + _many_check("SELECT pid, id FROM __THIS__ JOIN d ON pid > d.id") + _many_check("SELECT pid, id FROM __THIS__ LEFT JOIN d ON pid > d.id") + _many_check("SELECT pid, id FROM __THIS__ LEFT JOIN d ON NULL = 2") + _many_check("SELECT pid, id FROM __THIS__, d WHERE pid >= id AND v <> 'c'") + + +def test_engine_order_contract(): + # The engine's OWN documented deterministic order: probe rows in input + # order, matches contiguous in build INSERTION order, null-extension + # in place. + fn = DuckDBInferFn( + "SELECT pid, v FROM __THIS__ LEFT JOIN d ON pid = d.id", + row_tables={"__THIS__": T}, + static_tables={"d": DIM}, + output="dict", + shape="many", + ) + got = [tuple(r.values()) for r in fn.infer({"__THIS__": [T(**r) for r in ROWS]})] + assert got == [ + (1, "a"), + (1, "c"), + (1, "e"), + (2, "b"), + (2, "d"), + (3, None), + (None, None), + ] diff --git a/tests/test_known_limitations.py b/tests/test_known_limitations.py index 398c21f..9a75d4b 100644 --- a/tests/test_known_limitations.py +++ b/tests/test_known_limitations.py @@ -41,12 +41,22 @@ def test_non_constant_regex_pattern_rejects(): def test_static_tables_are_frozen_unique_key_maps(): dup = static({"id": "int", "v": "int"}, [{"id": 1, "v": 1}, {"id": 1, "v": 2}]) - # Duplicate keys = 1:N multiplicity — designed (stage B), not built. + # Duplicate keys = 1:N multiplicity: rejected under the DEFAULT shapes, + # served under the opt-in shape='many' (stage B, TASK-59). rejects( "SELECT v FROM __THIS__ JOIN d ON a = d.id", "duplicate map key", {"d": dup}, ) + fn = DuckDBInferFn( + "SELECT v FROM __THIS__ JOIN d ON a = d.id", + row_tables={"__THIS__": T}, + static_tables={"d": dup}, + output="dict", + shape="many", + ) + got = sorted(r["v"] for r in fn.infer({"__THIS__": [T(a=1)]})) + assert got == [1, 2] # NULL VALUES serve since TASK-55 (they ride as validity+payload pairs); # only NULL keys keep the drop rule (a NULL never equi-matches). withnull = static({"id": "int", "v": "int?"}, [{"id": 1, "v": None}]) diff --git a/tests/test_shape_contract.py b/tests/test_shape_contract.py index d384174..059d7bb 100644 --- a/tests/test_shape_contract.py +++ b/tests/test_shape_contract.py @@ -65,8 +65,18 @@ def test_filter_default_unchanged(): 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") +def test_many_enables_multiplicity_and_bad_values_are_named(): + # 'many' is the stage-B opt-in: dup-key joins build ONLY under it. + dup = pa.table({"id": [1, 1], "v": [10, 11]}) + with pytest.raises(ValueError, match="duplicate map key"): + build("SELECT a, v FROM __THIS__ JOIN d ON a = d.id", statics={"d": dup}) + fn = build( + "SELECT a, v FROM __THIS__ JOIN d ON a = d.id", + shape="many", + statics={"d": dup}, + ) + assert fn.shape == "many" + got = fn.infer({"__THIS__": [T(a=1), T(a=2)]}) + assert sorted(r["v"] for r in got) == [10, 11] with pytest.raises(ValueError, match="must be 'map', 'filter', or 'many'"): build("SELECT a FROM __THIS__", shape="projection")