From 048534348f93ce338bbed7f808a03dc79a0d5881 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Tue, 1 Sep 2026 21:37:49 -0400 Subject: [PATCH 1/4] =?UTF-8?q?DDIR:=20the=20sum=20universe=20=E2=80=94=20?= =?UTF-8?q?declared=20types,=20corgi=20as=20the=20typer,=20no=20shape=20in?= =?UTF-8?q?ference?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sums in DDIR had no type: `con Name(arity) = tag` was parse-time sugar, and three mechanisms guessed each sum's shape from data (infer_shape_cols per batch, infer_term_shape per term with an arity of tag+1, and shape_join, a private bottom lattice reconciling them). The corgi bump removes bottom, and with it the room to guess. Now: * `type Kind = Rare u64 | Common (u64, u64) | Empty;` declares a sum. Tags are positions, scoped to the type; payload shapes are u64/int, (), tuples, List(..), Option(..), Result(..), or an earlier type. Constructors build into the whole declared sum (`Term::Inject { tag, payload, sum }` carries every lane's shape; `Type::Ctor` disambiguates), patterns bind a tuple payload's fields or the whole payload, and `variant(Type, tag, payload)` is the data-driven form (corgi's Branch, over a homogeneous type). * Built-in `Option`/`Result` with `Some`/`None`/`Ok`/`Err`; the lane a payload cannot fix comes from the other branch of an `if` or the other arms of a `case` (a compile-time hole, never a runtime lane). * The typer is corgi's: `shape_of_term` lowers a term into a scratch graph and asks `corgi::shape_of`. infer_shape_cols, infer_term_shape, shape_of_place and shape_join are gone. * Shapes are pinned. An input collection's shape comes from its first row (`shape_of_row`; a variant or an empty list on input needs a declared schema, which is a follow-up) and every later batch transcodes against it. Each linear op compiles once, on its first non-empty batch (`Plan`), and a later batch of another shape is the invariant violation. * No row-wise path inside the dataflow. A term that does not lower is a type error with corgi's message, not a fallback: `Fold` steps that read the enclosing environment are closed by `CapList` capture, and list projection lowers to the new `Op::Get`. Rows exist only at the ingest and egress boundaries. Programs: `con` decls become `type` decls (sum_skew is an ordinary program now — both sides build the same declared sum); sum_ops uses the typed variant form; case_ops' shape-conflicted filter becomes a well-typed one and gains an `Option` round trip. All 14 backend programs agree with vec. corgi is pinned at WIP 578ef3f (the streamline merge, #16), with the `serde` feature DDIR's Term needs. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_012k2GSwxmvD2LvckkoXi6GK --- interactive/Cargo.toml | 2 +- interactive/examples/programs/adt.ddp | 4 +- interactive/examples/programs/ast.ddp | 3 +- interactive/examples/programs/binders.ddp | 2 +- interactive/examples/programs/tour.ddp | 3 +- interactive/src/backend/corgi.rs | 273 +++--- interactive/src/corgi/bytes.rs | 27 +- interactive/src/corgi/chunk.rs | 8 +- interactive/src/corgi/container.rs | 38 +- interactive/src/corgi/exchange.rs | 6 +- interactive/src/corgi/join.rs | 7 +- interactive/src/corgi/logic.rs | 899 +++++++++---------- interactive/src/corgi/reduce.rs | 6 +- interactive/src/ir.rs | 2 +- interactive/src/parse/mod.rs | 26 +- interactive/src/parse/pipe.rs | 282 ++++-- interactive/tests/programs/case_ops.ddp | 22 +- interactive/tests/programs/join_fallback.ddp | 2 +- interactive/tests/programs/sum_ops.ddp | 9 +- interactive/tests/programs/sum_skew.ddp | 24 +- 20 files changed, 880 insertions(+), 765 deletions(-) diff --git a/interactive/Cargo.toml b/interactive/Cargo.toml index f15835d90..1673d1440 100644 --- a/interactive/Cargo.toml +++ b/interactive/Cargo.toml @@ -14,7 +14,7 @@ workspace = true [dependencies] columnar = { workspace = true } # The columnar kernels for the interpreted backend, pinned by git rev. -corgi = { git = "https://github.com/frankmcsherry/wip", rev = "20f38d69ba0f964e6dd0d904338bf433fea0c642" } +corgi = { git = "https://github.com/frankmcsherry/WIP", rev = "578ef3fe79fdaa813ba0c721c08f60323f22d420", features = ["serde"] } differential-dataflow = { workspace = true } mimalloc = "0.1.48" serde = { version = "1.0", features = ["derive"] } diff --git a/interactive/examples/programs/adt.ddp b/interactive/examples/programs/adt.ddp index fb11c80e4..6abebc36b 100644 --- a/interactive/examples/programs/adt.ddp +++ b/interactive/examples/programs/adt.ddp @@ -2,8 +2,8 @@ -- carrying the field-sum as payload, then matched back out to a (bucket; sum) -- and counted per bucket. (Contrived — exercises sum intro/outro + fold.) -con Fwd(1) = 0; -- src < dst -con Bwd(1) = 1; -- otherwise +type Dir = Fwd u64 -- src < dst + | Bwd u64; -- otherwise let edges = input 0 | key($0[0] ; $0[1]); diff --git a/interactive/examples/programs/ast.ddp b/interactive/examples/programs/ast.ddp index da7d98f96..563dd0287 100644 --- a/interactive/examples/programs/ast.ddp +++ b/interactive/examples/programs/ast.ddp @@ -12,8 +12,7 @@ -- `$0[0] - $1[0] + 32768` stays non-negative -- corgi's structural order is unsigned at the -- integer leaf, so signed values are out of contract for `min`. -con Fwd(1) = 0; -con Bwd(1) = 1; +type Dir = Fwd u64 | Bwd u64; let rows = input 0 | key($0[0] ; $0[1]); diff --git a/interactive/examples/programs/binders.ddp b/interactive/examples/programs/binders.ddp index e79dfd0fa..fc9429214 100644 --- a/interactive/examples/programs/binders.ddp +++ b/interactive/examples/programs/binders.ddp @@ -1,7 +1,7 @@ -- Depth-tracking test: a pattern binder `xs` used INSIDE a fold step. Correct -- de Bruijn resolution => xs is the case payload at Bound(2) inside the fold; -- a wrong index would hit the fold's Int element and panic on Proj. -con Node(1) = 0; +type Node = Node List(u64); let rows = input 0 | key($0[0] ; Node(list($0[0], $0[1]))); diff --git a/interactive/examples/programs/tour.ddp b/interactive/examples/programs/tour.ddp index 29e9cb790..3d2a3bedf 100644 --- a/interactive/examples/programs/tour.ddp +++ b/interactive/examples/programs/tour.ddp @@ -3,8 +3,7 @@ -- fixtures under tests/programs/ pin individual lowerings and their fallback -- routing; this program pins that the constructs work TOGETHER. -con Fwd(1) = 0; -con Bwd(1) = 1; +type Dir = Fwd u64 | Bwd u64; let edges = input 0 | key($0[0] ; $0[1]); let roots = input 1 | key($0[0] ;); diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs index e940b4feb..4422b235c 100644 --- a/interactive/src/backend/corgi.rs +++ b/interactive/src/backend/corgi.rs @@ -30,7 +30,8 @@ use crate::corgi::exchange::CorgiPact; use crate::corgi::join::CorgiJoinBackend; use crate::corgi::reduce::CorgiReduceBackend; use differential_dataflow::operators::int_proxy::{ProxyJoinTactic, ProxyReduceTactic}; -use crate::corgi::logic::{compilable, compile_flatmap, compile_predicate, compile_projection, compile_scalar}; +use crate::corgi::logic::{compilable, compile_flatmap, compile_predicate, compile_projection, compile_scalar, shape_of_row}; +use corgi::{Graph, NumOp, Shape}; use crate::ir::{Diff, LinearOp, Time, Value as DValue}; use crate::parse::{Projection, Reducer}; use crate::scope_ir as st; @@ -38,7 +39,6 @@ use crate::scope_ir as st; /// A DDIR row, an update, the corgi container on dataflow edges, and the columnar trace — /// the shorthands the `Backend` methods below are written in terms of. type Row = DValue; -type Upd = ((Row, Row), Time, Diff); type CC = CorgiContainer; type CTrace = differential_dataflow::trace::chunk::ChunkSpine>; @@ -59,7 +59,7 @@ fn rebase_join_term(t: &crate::parse::Term) -> crate::parse::Term { List(fs) => List(fs.iter().map(rebase_join_term).collect()), Spread(inner) => Spread(Box::new(rebase_join_term(inner))), Proj(inner, i) => Proj(Box::new(rebase_join_term(inner)), *i), - Inject(tag, payload) => Inject(Box::new(rebase_join_term(tag)), Box::new(rebase_join_term(payload))), + Inject { tag, payload, sum } => Inject { tag: Box::new(rebase_join_term(tag)), payload: Box::new(rebase_join_term(payload)), sum: sum.clone() }, Case { scrutinee, arms, default } => Case { scrutinee: Box::new(rebase_join_term(scrutinee)), arms: arms.iter().map(rebase_join_term).collect(), @@ -81,84 +81,68 @@ fn rebase_join_term(t: &crate::parse::Term) -> crate::parse::Term { } } +/// The compiled form of one `LinearOp`, pinned to the shapes it was compiled against. Shapes are +/// static per collection, so a chain compiles ONCE, on the first non-empty batch, and every later +/// batch reuses the graph; a batch of a different shape is the invariant violation, not a +/// recompile. +#[derive(Default)] +pub struct Plan { + compiled: Option<(Shape, Shape, Graph)>, +} + +impl Plan { + /// The graph for a container of these shapes, compiling on first use. A type error is a + /// panic with corgi's message: a program that typechecks never reaches it. + fn graph(&mut self, what: &str, kshape: Shape, vshape: Shape, compile: impl FnOnce(&Shape, &Shape) -> Result, String>) -> &Graph { + if self.compiled.is_none() { + let g = compile(&kshape, &vshape).unwrap_or_else(|e| panic!("{what}: type error at shapes ({kshape}, {vshape}): {e}")); + self.compiled = Some((kshape.clone(), vshape.clone(), g)); + } + let (k, v, g) = self.compiled.as_ref().unwrap(); + assert!(*k == kshape && *v == vshape, "{what}: a batch of shape ({kshape}, {vshape}) reached an operator pinned at ({k}, {v})"); + g + } +} + /// Apply a `LinearOp` chain to one corgi container (the corgi-native compute per batch). /// Project = corgi `eval_graph`; Filter = corgi mask + `gather`; FlatMap = `eval_graph` to a list -/// column + a structural explode; Negate = Rust. Each falls back to rows when the term has no -/// lowering with this container's shapes, so capability never depends on the compiler's coverage. +/// column + a structural explode; Negate = Rust. Every term is columnar — there is no row-wise +/// path inside the dataflow — and each op's graph is compiled once (`plans`). An empty batch +/// passes through untouched: it carries no shape to compile against and no rows to compute. /// The two data<->time ops are columnar and total: EnterAt reads its delay field as a column and /// joins it into `times` in place; LiftIter reads the iteration coordinate out of `times` and /// appends it to `vals`. `level` is the scope depth (it locates that coordinate). -fn apply_ops(mut c: CC, ops: &[LinearOp], level: usize) -> CC { - use timely::order::Product; - use differential_dataflow::lattice::Lattice; +fn apply_ops(mut c: CC, ops: &[LinearOp], level: usize, plans: &mut [Plan]) -> CC { use differential_dataflow::dynamic::pointstamp::PointStamp; - // A container with no rows has no SHAPE either, and the ops below are shape-directed. - // - // Every path that rebuilds a container from rows (`from_updates`) infers its columns by - // scanning them, so with nothing to scan it returns the shape-erased default — `Unit` keys and - // `Unit` vals. Feed that to the next op and `compile_projection` cheerfully lowers `$1[0]` - // against a `Unit`, because a `Field` is legal in the abstract; `eval_graph` then meets a - // column that is not a product and panics. A filter that empties a batch mid-chain is enough - // to arrange this: `... | filter(len($1) == 2) | map(; $1[0][0] * $1[1][0])`. - // - // Skipping is not a shortcut, it is the whole answer: every `LinearOp` here maps zero rows to - // zero rows, so the only thing the ops could contribute is the output shape, and the input's - // shape is already gone. Nothing downstream reads it either — `CorgiChunker::push_into` drops - // empty containers before they reach `concat_blocks`, which is the one place shapes must - // agree. - // - // Latent since the backend was written; a single worker rarely empties a batch it was given, - // and the key-hash exchange makes it routine. - // - // The check belongs at the top of the LOOP, not before it. A chain empties itself: a filter - // takes the row-wise path, drops every row, and hands the erased container to the very next - // op in the same `ops` slice. Guarding only the entry catches the batch that arrives empty - // and misses the one that becomes empty, which is the common case. - for op in ops { + // A container with no rows has no SHAPE either, and the ops below are shape-directed: an + // empty batch passes through untouched (every `LinearOp` maps zero rows to zero rows), and + // nothing downstream reads its shape — `CorgiChunker::push_into` drops empty containers + // before they reach `concat_blocks`, the one place shapes must agree. The check belongs at + // the top of the LOOP, not before it: a filter can empty a batch mid-chain, and the next op + // in the same `ops` slice must not compile against the erased container. + for (op, plan) in ops.iter().zip(plans.iter_mut()) { if c.times.is_empty() { - break; + return c; } + let (kshape, vshape) = (corgi::shape_of_value(&c.keys), corgi::shape_of_value(&c.vals)); c = match op { LinearOp::Project(p) => { - let (kshape, vshape) = (corgi::shape_of_value(&c.keys), corgi::shape_of_value(&c.vals)); - // The shape-aware gate: attempt the lowering with this container's shapes and - // fall back to rows only when it declines — a heterogeneous list literal, a - // `Case` whose arms disagree, a data-driven tag. - if let Some(g) = compile_projection(&p.key, &p.val, &kshape, &vshape) { - let mut cols = corgi::eval_graph(&g, CValue::Prod(vec![c.keys, c.vals])).into_prod("linear project"); - let vals = cols.pop().unwrap(); - let keys = cols.pop().unwrap(); - CorgiContainer { keys, vals, times: c.times, diffs: c.diffs } - } else { - let mut out: Vec = Vec::new(); - for ((k, v), t, d) in c.into_updates() { - let mut env = vec![k, v]; - let nk = crate::ir::eval(&p.key, &mut env); - let nv = crate::ir::eval(&p.val, &mut env); - out.push(((nk, nv), t, d)); - } - CorgiContainer::from_updates(out) - } + let g = plan.graph("map", kshape, vshape, |k, v| compile_projection(&p.key, &p.val, k, v)); + let mut cols = corgi::eval_graph(g, CValue::Prod(vec![c.keys, c.vals])).into_prod("linear project").unwrap(); + let vals = cols.pop().unwrap(); + let keys = cols.pop().unwrap(); + CorgiContainer { keys, vals, times: c.times, diffs: c.diffs } } LinearOp::Filter(cond) => { - let (kshape, vshape) = (corgi::shape_of_value(&c.keys), corgi::shape_of_value(&c.vals)); - if let Some(g) = compile_predicate(cond, &kshape, &vshape) { - let mask = corgi::eval_graph(&g, CValue::Prod(vec![c.keys.clone(), c.vals.clone()])).into_u64("filter mask"); - let keep: Vec = (0..mask.len()).filter(|&i| mask[i] != 0).collect(); - let keys = gather(&c.keys, &keep); - let vals = gather(&c.vals, &keep); - let times = keep.iter().map(|&i| c.times[i].clone()).collect(); - let diffs = keep.iter().map(|&i| c.diffs[i]).collect(); - CorgiContainer { keys, vals, times, diffs } - } else { - let mut out: Vec = Vec::new(); - for ((k, v), t, d) in c.into_updates() { - let keep = { let mut env = vec![k.clone(), v.clone()]; crate::ir::eval(cond, &mut env).truthy() }; - if keep { out.push(((k, v), t, d)); } - } - CorgiContainer::from_updates(out) - } + let g = plan.graph("filter", kshape, vshape, |k, v| compile_predicate(cond, k, v)); + let mask = corgi::eval_graph(g, CValue::Prod(vec![c.keys.clone(), c.vals.clone()])).into_u64("filter mask").unwrap(); + let keep: Vec = (0..mask.len()).filter(|&i| mask[i] != 0).collect(); + let keys = gather(&c.keys, &keep); + let vals = gather(&c.vals, &keep); + let times = keep.iter().map(|&i| c.times[i].clone()).collect(); + let diffs = keep.iter().map(|&i| c.diffs[i]).collect(); + CorgiContainer { keys, vals, times, diffs } } LinearOp::Negate => { for d in c.diffs.iter_mut() { @@ -167,47 +151,30 @@ fn apply_ops(mut c: CC, ops: &[LinearOp], level: usize) -> CC { c } LinearOp::EnterAt(field) => { - let (kshape, vshape) = (corgi::shape_of_value(&c.keys), corgi::shape_of_value(&c.vals)); - if let Some(g) = compile_scalar(field, &kshape, &vshape) { - // The key and val columns are IDENTITY here — only times change. Evaluate the - // delay field to a `U64` column and join it into each time in place. Joining - // `Product(0, PointStamp([0,..,0, delay]))` is, coordinate-wise, `max` at index - // `level-1` and identity everywhere else (u64's minimum is 0), so the delta - // never has to be built. `PointStamp::new` re-strips the trailing minimums the - // resize may add, keeping the representation canonical for a zero delay. - let raw = corgi::eval_graph(&g, CValue::Prod(vec![c.keys.clone(), c.vals.clone()])) - .into_u64("enter_at delay"); - let idx = level.saturating_sub(1); - for (t, &r) in c.times.iter_mut().zip(raw.iter()) { - let delay = 256 * (64 - r.leading_zeros() as u64); - let mut coords = std::mem::take(&mut t.inner).into_inner(); - if coords.len() <= idx { - coords.resize(idx + 1, 0); - } - coords[idx] = coords[idx].max(delay); - t.inner = PointStamp::new(coords); - } - c - } else { - let mut out: Vec = Vec::new(); - for ((k, v), t, d) in c.into_updates() { - let delay = { - let mut env = vec![k.clone(), v.clone()]; - let raw = crate::ir::eval(field, &mut env).as_int() as u64; - 256 * (64 - raw.leading_zeros() as u64) - }; - let mut coords = smallvec::SmallVec::<[u64; 1]>::new(); - for _ in 0..level.saturating_sub(1) { coords.push(0); } - coords.push(delay); - let delta = Product::new(0u64, PointStamp::new(coords)); - out.push(((k, v), t.join(&delta), d)); + let g = plan.graph("enter_at", kshape, vshape, |k, v| compile_scalar(field, k, v)); + // The key and val columns are IDENTITY here — only times change. Evaluate the + // delay field to a `U64` column and join it into each time in place. Joining + // `Product(0, PointStamp([0,..,0, delay]))` is, coordinate-wise, `max` at index + // `level-1` and identity everywhere else (u64's minimum is 0), so the delta + // never has to be built. `PointStamp::new` re-strips the trailing minimums the + // resize may add, keeping the representation canonical for a zero delay. + let raw = corgi::eval_graph(g, CValue::Prod(vec![c.keys.clone(), c.vals.clone()])) + .into_u64("enter_at delay") + .unwrap(); + let idx = level.saturating_sub(1); + for (t, &r) in c.times.iter_mut().zip(raw.iter()) { + let delay = 256 * (64 - r.leading_zeros() as u64); + let mut coords = std::mem::take(&mut t.inner).into_inner(); + if coords.len() <= idx { + coords.resize(idx + 1, 0); } - CorgiContainer::from_updates(out) + coords[idx] = coords[idx].max(delay); + t.inner = PointStamp::new(coords); } + c } // The inverse of `EnterAt`: a value read OUT of each row's time. Vals gain one - // integer field; keys, times and diffs are untouched, and no term is compiled, so - // this path is total — there is no fallback to fall back to. + // integer field; keys, times and diffs are untouched, and no term is compiled. // // It mirrors [`append_iter`] shape for shape, and the empty product is where the two // representations part company: DDIR unit IS `Tuple([])`, which `append_iter` extends @@ -228,38 +195,33 @@ fn apply_ops(mut c: CC, ops: &[LinearOp], level: usize) -> CC { }; CorgiContainer { keys: c.keys, vals, times: c.times, diffs: c.diffs } } - // Row-wise ops (parity with `backend::vec::render_linear`). LinearOp::FlatMap(list_term) => { - let (kshape, vshape) = (corgi::shape_of_value(&c.keys), corgi::shape_of_value(&c.vals)); - if let Some(g) = compile_flatmap(list_term, &kshape, &vshape) { - // Structural explode: the evaluated list column's FLAT element storage already - // IS the new value column, so the elements never move. Each row's span in the - // bounds gives both the within-row position (DDIR's `$1[0]`) and a repeat map - // carrying key/time/diff across. No per-row eval, no transcode. - let (bounds, elems) = - corgi::eval_graph(&g, CValue::Prod(vec![c.keys.clone(), c.vals])).into_list("flatmap list"); - let ends: Vec = match &bounds { - corgi::Bounds::Offsets(v) => v.clone(), - corgi::Bounds::Stride(k, rows) => (1..=*rows).map(|i| i * k).collect(), - }; - let total = ends.last().copied().unwrap_or(0); - let (mut reps, mut pos) = (Vec::with_capacity(total), Vec::with_capacity(total)); - let mut start = 0usize; - for (row, end) in ends.into_iter().enumerate() { - for p in 0..(end - start) { - reps.push(row); - pos.push(p as u64); - } - start = end; - } - CorgiContainer { - keys: gather(&c.keys, &reps), - vals: CValue::Prod(vec![CValue::u64(pos), elems]), - times: reps.iter().map(|&r| c.times[r].clone()).collect(), - diffs: reps.iter().map(|&r| c.diffs[r]).collect(), + let g = plan.graph("flatmap", kshape, vshape, |k, v| compile_flatmap(list_term, k, v)); + // Structural explode: the evaluated list column's FLAT element storage already + // IS the new value column, so the elements never move. Each row's span in the + // bounds gives both the within-row position (DDIR's `$1[0]`) and a repeat map + // carrying key/time/diff across. No per-row eval, no transcode. + let (bounds, elems) = + corgi::eval_graph(g, CValue::Prod(vec![c.keys.clone(), c.vals])).into_list("flatmap list").unwrap(); + let ends: Vec = match &bounds { + corgi::Bounds::Offsets(v) => v.clone(), + corgi::Bounds::Stride(k, rows) => (1..=*rows).map(|i| i * k).collect(), + }; + let total = ends.last().copied().unwrap_or(0); + let (mut reps, mut pos) = (Vec::with_capacity(total), Vec::with_capacity(total)); + let mut start = 0usize; + for (row, end) in ends.into_iter().enumerate() { + for p in 0..(end - start) { + reps.push(row); + pos.push(p as u64); } - } else { - apply_flatmap_rows(c, list_term) + start = end; + } + CorgiContainer { + keys: gather(&c.keys, &reps), + vals: CValue::Prod(vec![CValue::u64(pos), elems]), + times: reps.iter().map(|&r| c.times[r].clone()).collect(), + diffs: reps.iter().map(|&r| c.diffs[r]).collect(), } } }; @@ -267,27 +229,6 @@ fn apply_ops(mut c: CC, ops: &[LinearOp], level: usize) -> CC { c } -/// The row-wise `FlatMap`: untranscode, `ir::eval` the list term per row, explode, re-transcode. -/// Parity with `backend::vec::render_linear`, and the fallback when the list term has no columnar -/// lowering with this container's shapes. -fn apply_flatmap_rows(c: CC, list_term: &crate::parse::Term) -> CC { - let mut out: Vec = Vec::new(); - for ((k, v), t, d) in c.into_updates() { - let elems = { - let mut env = vec![k.clone(), v.clone()]; - match crate::ir::eval(list_term, &mut env) { - DValue::List(xs) => xs, - other => panic!("flatmap: expected a List, got {other:?}"), - } - }; - for (pos, elem) in elems.into_iter().enumerate() { - out.push(((k.clone(), DValue::Tuple(vec![DValue::Int(pos as i64), elem])), t.clone(), d)); - } - } - CorgiContainer::from_updates(out) -} - - /// The corgi rendering substrate. An uninhabited type used only as a type-level tag: it /// carries the [`Backend`] impl (a namespace of rendering functions selected by type) and is /// never a value — rendering goes through `render_tree::`. The empty enum (vs a @@ -303,9 +244,10 @@ impl Backend for CorgiBackend { // `level` is the scope depth (locates the iteration coordinate for LiftIter/EnterAt). c.inner .unary(Pipeline, "CorgiLinear", move |_, _| { + let mut plans: Vec = ops.iter().map(|_| Plan::default()).collect(); move |input, output| { input.for_each(|cap, data| { - let mut out = apply_ops(std::mem::take(data), &ops, level); + let mut out = apply_ops(std::mem::take(data), &ops, level, &mut plans); output.session(&cap).give_container(&mut out); }); } @@ -469,9 +411,22 @@ pub fn render_tree_rows<'s>( .map(|c| { c.inner .unary(Pipeline, "ToCorgi", |_, _| { - |input, output| { + // The collection's shape, pinned from its first row: every later batch + // transcodes against it (a misfit row is a panic in `transcode`). + let mut pinned: Option<(Shape, Shape)> = None; + move |input, output| { input.for_each(|cap, data| { - let mut cc = CorgiContainer::from_updates(std::mem::take(data)); + let rows = std::mem::take(data); + let mut cc = match rows.first() { + None => CorgiContainer::default(), + Some(((k, v), _, _)) => { + let (ks, vs) = pinned.get_or_insert_with(|| { + let pin = |r: &Row, what: &str| shape_of_row(r).unwrap_or_else(|e| panic!("input {what}: {e}")); + (pin(k, "key"), pin(v, "value")) + }); + CorgiContainer::from_updates(rows, ks, vs) + } + }; output.session(&cap).give_container(&mut cc); }); } diff --git a/interactive/src/corgi/bytes.rs b/interactive/src/corgi/bytes.rs index 388521183..3275a9dfe 100644 --- a/interactive/src/corgi/bytes.rs +++ b/interactive/src/corgi/bytes.rs @@ -177,11 +177,32 @@ mod test { ] } + /// The declared shapes of each family — what a program's schema would pin (a variant carries + /// only its tag, so a family with sums cannot be pinned from a row). + fn shapes_of(name: &str) -> (corgi::Shape, corgi::Shape) { + use corgi::Shape::{List, Prim, Prod, Sum}; + let u = || Prim(64); + let pair = || Prod(vec![u(), u()]); + match name { + "scalars" => (u(), u()), + "tuples" => (pair(), Prod(vec![u()])), + "lists" => (u(), List(Box::new(u()))), + "variants" => (Sum(vec![u(), pair()]), u()), + "nested" => (Prod(vec![u(), Sum(vec![u(), pair()])]), List(Box::new(u()))), + other => panic!("no shapes for family {other}"), + } + } + + fn container_of(name: &str, updates: Vec<((DValue, DValue), Time, Diff)>) -> CorgiContainer { + let (k, v) = shapes_of(name); + CorgiContainer::from_updates(updates, &k, &v) + } + /// Every update survives the round trip, with its time and diff, for every shape family. #[test] fn round_trip_preserves_updates() { for (name, updates) in shape_families() { - let c = CorgiContainer::::from_updates(updates.clone()); + let c = container_of(name, updates.clone()); let back = round_trip(&c); assert_eq!(back.into_updates(), updates, "{name} did not survive the round trip"); } @@ -202,7 +223,7 @@ mod test { #[test] fn round_trip_preserves_column_hashes() { for (name, updates) in shape_families() { - let c = CorgiContainer::::from_updates(updates); + let c = container_of(name, updates); let (keys, vals) = (c.keys.clone(), c.vals.clone()); let back = round_trip(&c); assert_eq!(corgi::arrange::hash_rows(&back.keys), corgi::arrange::hash_rows(&keys), "{name} keys"); @@ -218,7 +239,7 @@ mod test { let updates: Vec<_> = (0..1000i64) .map(|i| ((DValue::Int(i), DValue::Int(i * 2)), time(0, &[]), 1)) .collect(); - let c = CorgiContainer::::from_updates(updates); + let c = CorgiContainer::::from_updates_pinned(updates); assert_eq!(corgi::bytes::length_in_bytes(&c.keys), 24 + 8 * 1000); assert_eq!(corgi::bytes::length_in_bytes(&c.vals), 24 + 8 * 1000); } diff --git a/interactive/src/corgi/chunk.rs b/interactive/src/corgi/chunk.rs index d3aa17318..a7aea6b13 100644 --- a/interactive/src/corgi/chunk.rs +++ b/interactive/src/corgi/chunk.rs @@ -75,7 +75,7 @@ impl Default for CorgiChunk { /// Split a `Prod([keys, vals])` corgi value into its two columns. fn split_kv(kv: CValue) -> (CValue, CValue) { - let mut cols = kv.into_prod("corgi chunk kv"); + let mut cols = kv.into_prod("corgi chunk kv").unwrap(); let vals = cols.pop().unwrap(); let keys = cols.pop().unwrap(); (keys, vals) @@ -497,7 +497,7 @@ pub fn present_key(keys: CValue) -> CValue { if corgi::arrange::leaf_slice(&keys).is_some() { return keys; } - let hashes = corgi::hash(&keys).into_u64("present_key"); + let hashes = corgi::hash(&keys).into_u64("present_key").unwrap(); CValue::Prod(vec![CValue::u64(hashes), keys]) } @@ -638,8 +638,8 @@ mod test { fn read_batch(b: &ChunkBatch>) -> BTreeMap<((u64, u64), u64), i64> { let mut m = BTreeMap::new(); for ch in &b.chunks { - let ks = ch.keys().clone().into_u64("k"); - let vs = ch.vals().clone().into_u64("v"); + let ks = ch.keys().clone().into_u64("k").unwrap(); + let vs = ch.vals().clone().into_u64("v").unwrap(); for i in 0..ch.len_() { *m.entry(((ks[i], vs[i]), ch.times().get(i))).or_insert(0) += ch.diffs()[i]; } } m.retain(|_, d| *d != 0); diff --git a/interactive/src/corgi/container.rs b/interactive/src/corgi/container.rs index 0b84c1108..d808a31e5 100644 --- a/interactive/src/corgi/container.rs +++ b/interactive/src/corgi/container.rs @@ -14,7 +14,7 @@ use timely::progress::{PathSummary, Timestamp}; use differential_dataflow::collection::containers::{Enter, Leave, Negate, ResultsIn}; use differential_dataflow::difference::Abelian; -use crate::corgi::logic::{infer_shape_cols, transcode, untranscode}; +use crate::corgi::logic::{transcode, untranscode}; use crate::ir::Value as DValue; type Row = DValue; @@ -58,33 +58,29 @@ impl Accountable for CorgiContainer { } impl CorgiContainer { - /// Build a container from DDIR row updates — the **ingest boundary** transcode (once per batch). - /// - /// Intended to have exactly ONE caller: external rows entering the dataflow. The other - /// callers (with [`into_updates`](Self::into_updates)) are `apply_ops`' row-wise fallbacks, - /// each scheduled to disappear: `Project`/`Filter` with the `Case`/`Inject`/`List`/`Unary`/ - /// `Hash` lowerings, `EnterAt` with bulk time mutation, `LiftIter` with a columnar - /// column-append, `FlatMap` with corgi's list ops. Row↔column round-trips inside the - /// dataflow are debt, not design. - /// Shapes are inferred by scanning the whole column ([`infer_shape_cols`]) — required so a - /// `Variant` column discovers all its arms (a single sample shows only one tag). - pub fn from_updates(updates: Vec<((Row, Row), T, R)>) -> Self { - if updates.is_empty() { - return Self::default(); - } + /// Build a container from DDIR row updates at the collection's PINNED shapes — the **ingest + /// boundary** transcode (once per batch). The only rows→columns conversion in the corgi + /// backend: inside the dataflow every operator is columnar. + pub fn from_updates(updates: Vec<((Row, Row), T, R)>, kshape: &corgi::Shape, vshape: &corgi::Shape) -> Self { let keys_rows: Vec = updates.iter().map(|u| u.0 .0.clone()).collect(); let vals_rows: Vec = updates.iter().map(|u| u.0 .1.clone()).collect(); - let kshape = infer_shape_cols(&keys_rows); - let vshape = infer_shape_cols(&vals_rows); let times = updates.iter().map(|u| u.1.clone()).collect(); let diffs = updates.iter().map(|u| u.2.clone()).collect(); - CorgiContainer { keys: transcode(&keys_rows, &kshape), vals: transcode(&vals_rows, &vshape), times, diffs } + CorgiContainer { keys: transcode(&keys_rows, kshape), vals: transcode(&vals_rows, vshape), times, diffs } + } + + /// Test convenience: build a container from row updates, pinning the shapes from the first + /// row (what the ingest operator does with the first batch it sees). + #[cfg(test)] + pub(crate) fn from_updates_pinned(updates: Vec<((Row, Row), T, R)>) -> Self { + use crate::corgi::logic::shape_of_row; + let Some(((k, v), _, _)) = updates.first() else { return Self::default() }; + let (ks, vs) = (shape_of_row(k).unwrap(), shape_of_row(v).unwrap()); + Self::from_updates(updates, &ks, &vs) } /// Read the container back to DDIR row updates — the **egress boundary** transcode (once). - /// corgi `Value` is self-describing, so shapes come from `shape_of_value`. Intended callers - /// are inspection/output edges; in-dataflow uses are fallback debt (see - /// [`from_updates`](Self::from_updates)). + /// corgi `Value` is self-describing, so shapes come from `shape_of_value`. pub fn into_updates(self) -> Vec<((Row, Row), T, R)> { if self.times.is_empty() { return Vec::new(); diff --git a/interactive/src/corgi/exchange.rs b/interactive/src/corgi/exchange.rs index d8c659cec..aaa875ff9 100644 --- a/interactive/src/corgi/exchange.rs +++ b/interactive/src/corgi/exchange.rs @@ -111,7 +111,7 @@ impl Distributor> f } let peers = pushers.len(); - let ids = corgi::hash(&container.keys).into_u64("corgi exchange: key hash"); + let ids = corgi::hash(&container.keys).into_u64("corgi exchange: key hash").unwrap(); self.counting_sort(&ids, peers); // Whole-container fast path. When every row shares a destination — a batch narrower than @@ -213,7 +213,7 @@ mod test { /// Partition `updates` across `peers` destinations and read each destination back as rows. fn partition(updates: Vec<((DValue, DValue), Time, Diff)>, peers: usize) -> Vec> { - let mut container = CorgiContainer::::from_updates(updates); + let mut container = CorgiContainer::::from_updates_pinned(updates); let mut pushers: Vec> = (0..peers).map(|_| Collect::default()).collect(); let mut distributor = CorgiDistributor::::default(); distributor.partition(&mut container, &timely::progress::Stamp::from_elem(0u64), &mut pushers); @@ -299,7 +299,7 @@ mod test { /// either way so no row is sent twice. #[test] fn the_input_is_consumed() { - let mut container = CorgiContainer::::from_updates(scalar_updates(50)); + let mut container = CorgiContainer::::from_updates_pinned(scalar_updates(50)); let mut pushers: Vec> = (0..4).map(|_| Collect::default()).collect(); let mut distributor = CorgiDistributor::::default(); distributor.partition(&mut container, &timely::progress::Stamp::from_elem(0u64), &mut pushers); diff --git a/interactive/src/corgi/join.rs b/interactive/src/corgi/join.rs index 27c6c454e..b189dad9c 100644 --- a/interactive/src/corgi/join.rs +++ b/interactive/src/corgi/join.rs @@ -174,9 +174,10 @@ impl ProxyJoinBackend, CBatch> for CorgiJoinBackend< let kc = recover_key(&gather_lanes(&keys0, &tag0, &off0)); let v0 = gather_lanes(&vals0, &tag0, &off0); let v1 = gather_lanes(&vals1, &tag1, &off1); - let proj = compile_join_projection(&self.key, &self.val, &shape_of_value(&kc), &shape_of_value(&v0), &shape_of_value(&v1)); + let proj = compile_join_projection(&self.key, &self.val, &shape_of_value(&kc), &shape_of_value(&v0), &shape_of_value(&v1)) + .unwrap_or_else(|e| panic!("join projection: type error: {e}")); let projected = corgi::eval_graph(&proj, CValue::Prod(vec![kc, v0, v1])); - let mut cols = projected.into_prod("corgi join projection"); + let mut cols = projected.into_prod("corgi join projection").unwrap(); let nv = cols.pop().unwrap(); let nk = cols.pop().unwrap(); output.push(CorgiContainer { @@ -236,7 +237,7 @@ fn leaf_valued(chunks: &[&CorgiChunk]) -> bool { fn pull_lanes(col: &CValue, idx: &[usize]) -> Vec> { leaf_lanes(col).expect("pull_lanes: leaf-laned column") .into_iter() - .map(|lane| gather(lane, idx).into_u64("corgi join lane pull")) + .map(|lane| gather(lane, idx).into_u64("corgi join lane pull").unwrap()) .collect() } diff --git a/interactive/src/corgi/logic.rs b/interactive/src/corgi/logic.rs index fee0ff47c..e00cd603a 100644 --- a/interactive/src/corgi/logic.rs +++ b/interactive/src/corgi/logic.rs @@ -1,114 +1,47 @@ -//! corgi as DDIR's columnar scalar logic: compile a `Term` to a corgi `Graph`, -//! transcode DDIR rows (`ir::Value`) to/from corgi columnar `Value`, directed by a `Shape` -//! inferred from the data (the dynamic-typing primitive). +//! corgi as DDIR's columnar scalar logic: compile a `Term` to a corgi `Graph`, and +//! transcode DDIR rows (`ir::Value`) to/from corgi columnar `Value` at the I/O boundaries. //! -//! The compiler (`compile`) covers Var/Bound/Int/Tuple(+Spread)/Proj/Binary/If/Fold, list and -//! sum intro (`List`/`Inject`), sum elimination (`Case`), and the Neg/Not/Len/IsTag unaries. -//! Ordered compares are signed-correct (`ToSigned`); the residual non-negative-int assumption is -//! confined to order-SENSITIVE contexts (the `Min` reducer and structural sort order compare raw -//! `u64` bits). `hash` is corgi's structural `Op::Hash`, the same function `ir::eval` folds row-wise. -//! Only shape-dependent cases decline (heterogeneous lists, conflicting `Case` arms, data-driven -//! tags); those fall back to row-wise `ir::eval` in the backend. The transcode layer is total over `Shape` (Prim/Unit/Prod/List/Sum), so a -//! `Variant` column round-trips via corgi `Sum` (see `infer_shape_cols` for the all-rows arm scan). +//! Shapes are STATIC. A collection's shape is fixed when it is first seen — pinned from its first +//! row at ingest ([`shape_of_row`]), or computed from its operator's term — and never re-derived +//! from data. Sums are the case where a row cannot tell: a `Variant` carries a tag, not its type, +//! so every sum a program builds is one it declared (`Term::Inject` carries the whole sum's lane +//! shapes), and a sum arriving on input needs a declared schema. +//! +//! The typer is corgi's. [`shape_of_term`] compiles a term into a scratch graph and asks +//! `corgi::shape_of` (its evaluator on zero rows) for the result shape, so every shape rule — +//! which lanes a `case` sees, what an `if` may blend, whether two operands compare — is the one +//! the kernels enforce, and a program that lowers is a program that runs. The compiler covers +//! Var/Bound/Int/Tuple(+Spread)/Proj/Binary/If/Fold, list and sum intro (`List`/`Inject`), sum +//! elimination (`Case`), and the Neg/Not/Len/IsTag unaries; `Err` is a type error, reported with +//! corgi's message. Ordered compares are signed-correct (`ToSigned`); `hash` is corgi's structural +//! `Op::Hash`, the same function `ir::eval` folds row-wise. use crate::ir::Value as DValue; -use crate::parse::{BinOp, Term, UnOp}; +use crate::parse::{BinOp, SumTy, Term, UnOp}; use corgi::{ArithOp, BinOp as CBinOp, Builder, CmpOp, Graph, Kind, NumOp, Op, Pred, Shape, Value as CValue}; -/// Dynamic typing over a whole COLUMN: infer a `Shape` by scanning every row, not just a sample. -/// Required for sum types — a `Variant` column's shape is the union of all arms that appear, which a -/// single sample can't reveal (it shows only one tag), so the scan is over every row, recursing -/// column-wise to cover nested variants too. -pub fn infer_shape_cols(rows: &[DValue]) -> Shape { - assert_uniform(rows); - let Some(first) = rows.first() else { return Shape::Unit }; - match first { - DValue::Int(_) => Shape::Prim(64), - DValue::Tuple(xs) if xs.is_empty() => Shape::Unit, - DValue::Tuple(xs) => { - let n = xs.len(); - Shape::Prod( - (0..n) - .map(|i| { - let col: Vec = rows - .iter() - .map(|r| match r { - DValue::Tuple(f) => f[i].clone(), - other => panic!("infer_shape_cols: expected Tuple, got {other:?}"), - }) - .collect(); - infer_shape_cols(&col) - }) - .collect(), - ) - } - DValue::List(_) => { - let flat: Vec = rows - .iter() - .flat_map(|r| match r { - DValue::List(xs) => xs.clone(), - other => panic!("infer_shape_cols: expected List, got {other:?}"), - }) - .collect(); - Shape::List(Box::new(infer_shape_cols(&flat))) - } - DValue::Variant(..) => { - // One lane per variant 0..=max_tag; a lane present in the data gets its arm's shape (from - // that arm's rows), an absent tag stays `None` (⊥, uncommitted — adopts a sibling on join). - let max_tag = rows - .iter() - .map(|r| match r { - DValue::Variant(t, _) => *t as usize, - other => panic!("infer_shape_cols: expected Variant, got {other:?}"), - }) - .max() - .unwrap(); - let lanes = (0..=max_tag) - .map(|tag| { - let payloads: Vec = rows - .iter() - .filter_map(|r| match r { - DValue::Variant(t, p) if *t as usize == tag => Some((**p).clone()), - _ => None, - }) - .collect(); - (!payloads.is_empty()).then(|| infer_shape_cols(&payloads)) - }) - .collect(); - Shape::Sum(lanes) - } - } -} +type Res = Result; -/// The uniformity contract, enforced LOUDLY at the ingest boundary: corgi columns are SoA -/// and carry ONE shape, while DDIR rows are dynamically typed. Without this check a mixed -/// column either truncated silently (rows wider than row 0 lost their extra fields — silent -/// wrong answers) or panicked with a bare `index out of bounds` — both worse than saying why. -/// Nested levels are covered by `infer_shape_cols`' recursion (each recursive call re-checks). -fn assert_uniform(rows: &[DValue]) { - let Some(first) = rows.first() else { return }; - fn kind(r: &DValue) -> (&'static str, usize) { - match r { - DValue::Int(_) => ("Int", 0), - DValue::Tuple(xs) => ("Tuple", xs.len()), - DValue::List(_) => ("List", 0), - DValue::Variant(..) => ("Variant", 0), - } - } - let k0 = kind(first); - for (i, r) in rows.iter().enumerate().skip(1) { - let ki = kind(r); - assert!( - ki == k0, - "corgi transcode: heterogeneous column — row 0 is {}/{} but row {} is {}/{}: \ - the corgi backend requires shape-uniform columns (pad rows to one arity, or use --backend=vec)", - k0.0, k0.1, i, ki.0, ki.1 - ); +/// The shape of one input row — what a collection is pinned to when its first row arrives. An +/// `Int`, a `Tuple`, or a non-empty `List` determines its shape alone; a `Variant` (whose type a +/// bare tag cannot name) or an empty `List` (whose element shape nothing supplies) needs the +/// input's schema declared. +pub fn shape_of_row(row: &DValue) -> Res { + match row { + DValue::Int(_) => Ok(Shape::Prim(64)), + DValue::Tuple(xs) if xs.is_empty() => Ok(Shape::Unit), + DValue::Tuple(xs) => Ok(Shape::Prod(xs.iter().map(shape_of_row).collect::>()?)), + DValue::List(xs) => match xs.first() { + Some(x) => Ok(Shape::List(Box::new(shape_of_row(x)?))), + None => Err("an empty list on input has no element shape; declare the input's schema".into()), + }, + DValue::Variant(..) => Err("a variant on input has no type; declare the input's schema".into()), } } -/// AoS rows -> SoA corgi columns, directed by `shape`. +/// AoS rows -> SoA corgi columns, directed by `shape`. A row that does not fit the shape is a +/// panic: the shape was pinned from a row of this collection, so a misfit is an ingest error. pub fn transcode(rows: &[DValue], shape: &Shape) -> CValue { match shape { Shape::Prim(_) => CValue::u64(rows.iter().map(|r| r.as_int() as u64).collect()), @@ -146,8 +79,8 @@ pub fn transcode(rows: &[DValue], shape: &Shape) -> CValue { CValue::List(ends.into(), Box::new(transcode(&flat, elem))) } Shape::Sum(lanes) => { - // Per-row tag, plus one packed lane per committed variant (its arm's rows in row order; - // `Value::sum_opt` derives each row's within-lane offset). Absent arms stay `⊥` (None). + // Per-row tag, plus one packed lane per variant (its arm's rows in row order; a + // variant no row uses is an empty column of its declared shape). let tags: Vec = rows .iter() .map(|r| match r { @@ -155,23 +88,24 @@ pub fn transcode(rows: &[DValue], shape: &Shape) -> CValue { other => panic!("transcode: expected Variant, got {other:?}"), }) .collect(); - let lane_vals: Vec> = lanes + if let Some(t) = tags.iter().find(|&&t| t >= lanes.len()) { + panic!("transcode: tag {t} is outside the declared {}-variant sum", lanes.len()); + } + let lane_vals: Vec = lanes .iter() .enumerate() - .map(|(tag, ls)| { - ls.as_ref().map(|lshape| { - let payloads: Vec = rows - .iter() - .filter_map(|r| match r { - DValue::Variant(t, p) if *t as usize == tag => Some((**p).clone()), - _ => None, - }) - .collect(); - transcode(&payloads, lshape) - }) + .map(|(tag, lshape)| { + let payloads: Vec = rows + .iter() + .filter_map(|r| match r { + DValue::Variant(t, p) if *t as usize == tag => Some((**p).clone()), + _ => None, + }) + .collect(); + transcode(&payloads, lshape) }) .collect(); - CValue::sum_opt(tags, lane_vals) + CValue::sum(tags, lane_vals) } } } @@ -179,10 +113,10 @@ pub fn transcode(rows: &[DValue], shape: &Shape) -> CValue { /// SoA corgi columns -> AoS rows, directed by `shape`. Inverse of [`transcode`]. pub fn untranscode(col: CValue, shape: &Shape) -> Vec { match shape { - Shape::Prim(_) => col.into_u64("untranscode").into_iter().map(|x| DValue::Int(x as i64)).collect(), + Shape::Prim(_) => col.into_u64("untranscode").unwrap().into_iter().map(|x| DValue::Int(x as i64)).collect(), Shape::Unit => vec![DValue::unit(); col.len()], Shape::Prod(fs) => { - let cols = col.into_prod("untranscode"); + let cols = col.into_prod("untranscode").unwrap(); let n = if cols.is_empty() { 0 } else { cols[0].len() }; let per_field: Vec> = cols.into_iter().zip(fs.iter()).map(|(c, fsi)| untranscode(c, fsi)).collect(); @@ -209,171 +143,59 @@ pub fn untranscode(col: CValue, shape: &Shape) -> Vec { out } Shape::Sum(lanes) => { - // Inverse of transcode's Sum: untranscode each committed lane, then for each row pull its - // payload from its lane at the recorded within-lane OFFSET (robust to row reordering from - // a prior gather/merge — not a sequential cursor). - let (tags, offsets, variant_vals) = col.into_sum("untranscode"); - let lane_rows: Vec>> = variant_vals - .into_iter() - .zip(lanes.iter()) - .map(|(v, ls)| match (v, ls) { - (Some(cv), Some(lshape)) => Some(untranscode(cv, lshape)), - _ => None, - }) - .collect(); + // Inverse of transcode's Sum: untranscode each lane, then for each row pull its payload + // from its lane at the recorded within-lane OFFSET (robust to row reordering from a + // prior gather/merge — not a sequential cursor). + let (tags, offsets, variant_vals) = col.into_sum("untranscode").unwrap(); + let lane_rows: Vec> = + variant_vals.into_iter().zip(lanes.iter()).map(|(v, ls)| untranscode(v, ls)).collect(); tags.iter() .zip(&offsets) - .map(|(&tag, &off)| { - let payload = lane_rows[tag].as_ref().expect("untranscode: committed lane")[off].clone(); - DValue::Variant(tag as u32, Box::new(payload)) - }) + .map(|(&tag, &off)| DValue::Variant(tag as u32, Box::new(lane_rows[tag][off].clone()))) .collect() } } } -/// Structural shape of a "place" Term (Var/Proj chain) given the env vars' shapes — used to resolve -/// `Spread`'s arity (how many fields to splice) and (later) Proj-on-list vs Proj-on-tuple. -pub fn shape_of_place(t: &Term, env_shapes: &[Shape]) -> Shape { - match t { - Term::Var(i) => env_shapes[*i].clone(), - Term::Proj(inner, i) => match shape_of_place(inner, env_shapes) { - Shape::Prod(fs) => fs[*i].clone(), - Shape::List(e) => *e, - other => panic!("shape_of_place: Proj on non-aggregate {other:?}"), - }, - other => panic!("shape_of_place: unsupported place {other:?}"), - } -} - -/// TOTAL best-effort shape of a `Proj` operand: resolve the `Var`/`Bound`/`Proj` spine -/// through KNOWN shapes only; `None` means "can't tell" (never a panic), and the caller -/// keeps the old lowering. The non-panicking cousin of [`shape_of_place`]. -fn place_shape_opt(t: &Term, env_shapes: &[Shape]) -> Option { - match t { - Term::Var(i) => env_shapes.get(*i).cloned(), - Term::Bound(k) => env_shapes.get(env_shapes.len().checked_sub(1 + *k)?).cloned(), - Term::Proj(inner, i) => match place_shape_opt(inner, env_shapes)? { - Shape::Prod(fs) => fs.get(*i).cloned(), - Shape::List(e) => Some(*e), - _ => None, - }, - _ => None, - } +/// The shape of a term in an environment — corgi's typer, asked through a scratch lowering. +/// `expected` is the shape an enclosing `if` or `case` arm already fixed; it fills the lane a +/// built-in `None`/`Ok`/`Err` cannot fix from its payload. +pub fn shape_of_term(t: &Term, env_shapes: &[Shape], expected: Option<&Shape>) -> Res { + let mut b = Builder::::default(); + let inp = b.input(); + let env: Vec = (0..env_shapes.len()).map(|i| b.add(Op::Field(i), vec![inp])).collect(); + let out = compile(t, &mut b, &env, env_shapes, inp, expected)?; + corgi::shape_of(&b.finish(out), &Shape::Prod(env_shapes.to_vec())) } -/// Best-effort structural `Shape` of a (non-place) compiled `Term`, used to decide cross-shape -/// `Eq`/`Ne`: DDIR compares `Value`s structurally, so e.g. `Tuple != Int` is *always* true (the -/// variants differ) — but corgi's `CmpOp::Rel` requires matching shapes. When the operand shapes -/// differ we fold the comparison to a constant instead of emitting a (panicking) `Rel`. -fn infer_term_shape(t: &Term, env_shapes: &[Shape]) -> Shape { +/// Does `t` read anything outside its own `depth` innermost binders — an input row (`Var`) or an +/// enclosing binder? A fold step that does needs its environment captured into the fold. +fn mentions_env(t: &Term, depth: usize) -> bool { match t { - Term::Var(i) => env_shapes[*i].clone(), - Term::Int(_) => Shape::Prim(64), - Term::Proj(inner, i) => match infer_term_shape(inner, env_shapes) { - Shape::Prod(fs) => fs[*i].clone(), - Shape::List(e) => *e, - other => panic!("infer_term_shape: Proj on non-aggregate {other:?}"), - }, - Term::Tuple(fields) => { - let mut fs = Vec::new(); - for f in fields { - match f { - Term::Spread(inner) => match infer_term_shape(inner, env_shapes) { - Shape::Prod(inner_fs) => fs.extend(inner_fs), - Shape::Unit => {} - other => fs.push(other), - }, - _ => fs.push(infer_term_shape(f, env_shapes)), - } - } - if fs.is_empty() { Shape::Unit } else { Shape::Prod(fs) } - } - // A list literal is a `List` of its elements' joined shape. `Fold` reads this to find its - // element shape, so a literal folded in place (`fold(list(..), ..)`) depends on it; a - // heterogeneous literal keeps the first field's shape here and declines in `compile`. - Term::List(fields) => { - let elem = fields - .iter() - .map(|f| infer_term_shape(f, env_shapes)) - .reduce(|acc, s| shape_join(&acc, &s).unwrap_or(acc)) - .unwrap_or(Shape::Unit); - Shape::List(Box::new(elem)) - } - Term::Bound(k) => env_shapes.get(env_shapes.len().wrapping_sub(1 + *k)).cloned().unwrap_or(Shape::Prim(64)), - Term::If { then, els, .. } => { - // Join the branch shapes (⊥ sum lanes unify), so a downstream `Case` sees every - // lane either branch can commit; under-approximating lanes would leave runtime - // rows unmapped. - let t = infer_term_shape(then, env_shapes); - shape_join(&t, &infer_term_shape(els, env_shapes)).unwrap_or(t) - } + Term::Var(_) => true, + Term::Bound(k) => *k >= depth, + Term::Int(_) => false, + Term::Tuple(fs) | Term::List(fs) | Term::Hash(fs) => fs.iter().any(|f| mentions_env(f, depth)), + Term::Spread(inner) | Term::Proj(inner, _) | Term::Unary(_, inner) => mentions_env(inner, depth), + Term::Inject { tag, payload, .. } => mentions_env(tag, depth) || mentions_env(payload, depth), Term::Case { scrutinee, arms, default } => { - // The joined shape of the reachable arms (the committed scrutinee lanes). - let lanes = match infer_term_shape(scrutinee, env_shapes) { Shape::Sum(l) => l, _ => Vec::new() }; - let mut shape: Option = None; - for (i, lane) in lanes.iter().enumerate() { - let Some(lane_shape) = lane else { continue }; - let s = if i < arms.len() { - let mut es = env_shapes.to_vec(); - es.push(lane_shape.clone()); - infer_term_shape(&arms[i], &es) - } else if let Some(d) = default { - infer_term_shape(d, env_shapes) - } else { - continue; - }; - shape = Some(match shape { None => s, Some(prev) => shape_join(&prev, &s).unwrap_or(prev) }); - } - shape.unwrap_or(Shape::Prim(64)) + mentions_env(scrutinee, depth) + || arms.iter().any(|a| mentions_env(a, depth + 1)) + || default.as_ref().is_some_and(|d| mentions_env(d, depth)) } - Term::Inject(tag, payload) => { - let t = match &**tag { Term::Int(t) => *t as usize, _ => 0 }; - let mut lanes: Vec> = vec![None; t + 1]; - lanes[t] = Some(infer_term_shape(payload, env_shapes)); - Shape::Sum(lanes) - } - // Arithmetic, comparisons, and anything else scalar-ish reduce to a primitive column. - _ => Shape::Prim(64), - } -} - -/// The ⊥-tolerant join of two shapes: `Sum` lanes unify lane-wise with an uncommitted (`None`) -/// lane adopting its sibling; `None` (the function's) means the shapes genuinely conflict. -/// Local until corgi exports its `shape::join`. -fn shape_join(a: &Shape, b: &Shape) -> Option { - match (a, b) { - (Shape::Prim(x), Shape::Prim(y)) if x == y => Some(Shape::Prim(*x)), - (Shape::Unit, Shape::Unit) => Some(Shape::Unit), - (Shape::Prod(xs), Shape::Prod(ys)) if xs.len() == ys.len() => { - let fs: Option> = xs.iter().zip(ys).map(|(x, y)| shape_join(x, y)).collect(); - Some(Shape::Prod(fs?)) - } - (Shape::List(x), Shape::List(y)) => Some(Shape::List(Box::new(shape_join(x, y)?))), - (Shape::Sum(xs), Shape::Sum(ys)) => { - let n = xs.len().max(ys.len()); - let mut lanes = Vec::with_capacity(n); - for i in 0..n { - lanes.push(match (xs.get(i).cloned().flatten(), ys.get(i).cloned().flatten()) { - (Some(x), Some(y)) => Some(shape_join(&x, &y)?), - (x, y) => x.or(y), - }); - } - Some(Shape::Sum(lanes)) + Term::Fold { list, init, step } => { + mentions_env(list, depth) || mentions_env(init, depth) || mentions_env(step, depth + 2) } - _ => None, + Term::If { cond, then, els } => mentions_env(cond, depth) || mentions_env(then, depth) || mentions_env(els, depth), + Term::Binary(_, l, r) => mentions_env(l, depth) || mentions_env(r, depth), } } /// Whether [`compile`] can lower this term WITHOUT knowing its operands' shapes — the gate for /// join-INLINE projections, which are compiled before any container is in hand. It is therefore -/// deliberately narrower than `compile`: every shape-dependent form (`List`, `Case`, data-driven -/// `Inject`) answers false here and is compiled by the linear stage the join defers it to, which -/// does have shapes. Capability never depends on this; only where the work happens. -/// -/// The only term with no lowering at all is a data-driven `Inject` tag, which has no static lane -/// count — and no surface syntax either, so nothing a program can write reaches the row-wise -/// fallback on shape-free grounds. +/// deliberately narrower than `compile`: every shape-dependent form (`List`, `Case`, a built-in +/// sum whose lane the payload fixes, a data-driven tag) answers false here and is compiled by the +/// linear stage the join defers it to, which does have shapes. pub fn compilable(t: &Term) -> bool { match t { Term::Var(_) | Term::Bound(_) | Term::Int(_) => true, @@ -383,38 +205,92 @@ pub fn compilable(t: &Term) -> bool { Term::If { cond, then, els } => compilable(cond) && compilable(then) && compilable(els), Term::Fold { list, init, step } => compilable(list) && compilable(init) && compilable(step), Term::Unary(op, inner) => matches!(op, UnOp::Neg | UnOp::Not | UnOp::Len | UnOp::IsTag(_)) && compilable(inner), - // Literal-tag sum intro lowers (`Op::Inject`); a data-driven tag has no static lane - // count, so it stays row-wise. - Term::Inject(tag, payload) => matches!(&**tag, Term::Int(_)) && compilable(payload), + // A literal tag into a declared type knows its whole sum; the built-ins and a data-driven + // tag need the payload's shape. + Term::Inject { tag, payload, sum } => { + matches!(&**tag, Term::Int(_)) && matches!(sum, SumTy::Declared(_)) && compilable(payload) + } // `Op::Hash` is shape-generic (it folds whatever structure it is handed), so `hash` // needs no shapes to lower and can answer true here. Term::Hash(args) => args.iter().all(compilable), - // `List` and `Case` deliberately stay false HERE even though `compile` lowers both: - // each needs shapes to decide homogeneity (a list's elements, a case's arms), and this - // check runs without them. The join defers such projections to a linear stage, whose - // shape-aware `compile` lowers them there. - _ => false, // List intro, Case (here), data-driven Inject — see `compile`. + _ => false, // List intro, Case — see `compile`. } } -/// Compile a `Term` to a corgi node. `env[i]` = node for `Var(i)`; `env_shapes[i]` = its shape -/// (for `Spread`). Binders push on top (read by `Bound(k)`). `anchor` sizes `Lit` broadcasts. -pub fn compile(term: &Term, b: &mut Builder, env: &[usize], env_shapes: &[Shape], anchor: usize) -> Option { +/// The lane shapes of the sum an `Inject` builds: the declaration's, or a built-in's with the +/// payload in its lane and the other lane from `expected`. +fn lanes_of(sum: &SumTy, tag: usize, payload: &Shape, expected: Option<&Shape>) -> Res> { + let other = |k: usize, what: &str| -> Res { + match expected { + Some(Shape::Sum(ls)) if ls.len() == 2 => Ok(ls[k].clone()), + _ => Err(format!( + "cannot infer the {what} lane of this built-in sum here; use a declared type, or an `if` whose other branch fixes it" + )), + } + }; + match sum { + SumTy::Declared(lanes) => { + if tag >= lanes.len() { + return Err(format!("constructor tag {tag} is outside the declared {}-variant sum", lanes.len())); + } + Ok(lanes.clone()) + } + SumTy::Option => match tag { + 0 => Ok(vec![Shape::Unit, other(1, "Some")?]), + 1 => Ok(vec![Shape::Unit, payload.clone()]), + _ => Err("Option has two variants".into()), + }, + SumTy::Result => match tag { + 0 => Ok(vec![payload.clone(), other(1, "Err")?]), + 1 => Ok(vec![other(0, "Ok")?, payload.clone()]), + _ => Err("Result has two variants".into()), + }, + } +} + +/// The shapes of an `if`'s two branches: each is typed on its own, and a branch that cannot be +/// (a bare `None`/`Ok`/`Err`) borrows the other's shape. +fn branch_shapes(then: &Term, els: &Term, env_shapes: &[Shape], expected: Option<&Shape>) -> Res<(Shape, Shape)> { + match (shape_of_term(then, env_shapes, expected), shape_of_term(els, env_shapes, expected)) { + (Ok(t), Ok(e)) => Ok((t, e)), + (Ok(t), Err(_)) => { + let e = shape_of_term(els, env_shapes, Some(&t))?; + Ok((t, e)) + } + (Err(_), Ok(e)) => { + let t = shape_of_term(then, env_shapes, Some(&e))?; + Ok((t, e)) + } + (Err(e), Err(_)) => Err(e), + } +} + +/// Compile a `Term` to a corgi node. `env[i]` = node for `Var(i)`; `env_shapes[i]` = its shape. +/// Binders push on top (read by `Bound(k)`). `anchor` sizes `Lit` broadcasts. `expected` is the +/// shape the context already fixed for this term, if any (see [`shape_of_term`]). `Err` is a +/// type error: the term has no meaning at these shapes. +pub fn compile( + term: &Term, + b: &mut Builder, + env: &[usize], + env_shapes: &[Shape], + anchor: usize, + expected: Option<&Shape>, +) -> Res { match term { - // Out-of-range env references decline rather than panic: closed bodies (fold steps, - // case arms) truncate the environment by design, and a term reaching past it is the - // documented restriction speaking — rows handle it. - Term::Var(i) => env.get(*i).copied(), - Term::Bound(k) => env.len().checked_sub(1 + *k).map(|i| env[i]), - Term::Int(n) => Some(b.add(Op::Lit(CValue::u64(vec![*n as u64])), vec![anchor])), + Term::Var(i) => env.get(*i).copied().ok_or_else(|| format!("`${i}` is not in scope here")), + Term::Bound(k) => { + env.len().checked_sub(1 + *k).map(|i| env[i]).ok_or_else(|| format!("binder `^{k}` is not in scope here")) + } + Term::Int(n) => Ok(b.add(Op::Lit(CValue::u64(vec![*n as u64])), vec![anchor])), Term::Tuple(fields) => { - // A `Spread(place)` child splices the place's `Prod` fields in place (the flat-row model). + // A `Spread(t)` child splices `t`'s `Prod` fields in place (the flat-row model). let mut ids: Vec = Vec::new(); for f in fields { match f { Term::Spread(inner) => { - let node = compile(inner, b, env, env_shapes, anchor)?; - match shape_of_place(inner, env_shapes) { + let node = compile(inner, b, env, env_shapes, anchor, None)?; + match shape_of_term(inner, env_shapes, None)? { Shape::Prod(fs) => { for i in 0..fs.len() { ids.push(b.add(Op::Field(i), vec![node])); @@ -424,41 +300,42 @@ pub fn compile(term: &Term, b: &mut Builder, env: &[usize], env_shapes: & _ => ids.push(node), // scalar: splice the value itself } } - _ => ids.push(compile(f, b, env, env_shapes, anchor)?), + _ => ids.push(compile(f, b, env, env_shapes, anchor, None)?), } } // An empty field list is DDIR unit: emit a length-carrying `Unit` column over the anchor, // NOT `Prod([])` (an empty product has no rows to count, so the row count would be lost). if ids.is_empty() { - Some(b.add(Op::Unit, vec![anchor])) + Ok(b.add(Op::Unit, vec![anchor])) } else { - Some(b.tuple(ids)) + Ok(b.tuple(ids)) } } + Term::Spread(_) => Err("`$n...` spread is only meaningful inside a tuple".into()), + // Projection: a tuple field, or a list element (`Get`, faulting out of range as `eval` does). Term::Proj(t, i) => { - // Shape-directed: `Op::Field` is PRODUCT elimination only. On a `List` operand, - // DDIR's Proj means indexing (`ir::eval` ir.rs:112), which has no columnar - // lowering here yet (corgi's Get tier) — decline, and rows handle it. The check - // is the TOTAL place resolver (never panics); an unresolved operand compiles as - // before (`Field`, with corgi's runtime shape check behind it). - if matches!(place_shape_opt(t, env_shapes), Some(Shape::List(_))) { - return None; + let id = compile(t, b, env, env_shapes, anchor, None)?; + match shape_of_term(t, env_shapes, None)? { + Shape::List(_) => { + let idx = b.add(Op::Lit(CValue::u64(vec![*i as u64])), vec![anchor]); + let pair = b.tuple(vec![idx, id]); + Ok(b.add(Op::Get, vec![pair])) + } + _ => Ok(b.add(Op::Field(*i), vec![id])), } - let id = compile(t, b, env, env_shapes, anchor)?; - Some(b.add(Op::Field(*i), vec![id])) } Term::Binary(op, l, r) => { - let lid = compile(l, b, env, env_shapes, anchor)?; - let rid = compile(r, b, env, env_shapes, anchor)?; + let lid = compile(l, b, env, env_shapes, anchor, None)?; + let rid = compile(r, b, env, env_shapes, anchor, None)?; let pair = |b: &mut Builder, x, y| b.tuple(vec![x, y]); - Some(match op { + Ok(match op { BinOp::Add => { let p = pair(b, lid, rid); b.add(ArithOp::Bin(CBinOp::Add, Kind::U, 64), vec![p]) } BinOp::Sub => { let p = pair(b, lid, rid); b.add(ArithOp::Bin(CBinOp::Sub, Kind::U, 64), vec![p]) } BinOp::Mul => { let p = pair(b, lid, rid); b.add(ArithOp::Bin(CBinOp::Mul, Kind::U, 64), vec![p]) } BinOp::Eq | BinOp::Ne => { // Cross-shape structural compare folds to a constant (Eq→0, Ne→1) over `anchor`; // same-shape emits a real corgi `Rel`. - if infer_term_shape(l, env_shapes) != infer_term_shape(r, env_shapes) { + if shape_of_term(l, env_shapes, None)? != shape_of_term(r, env_shapes, None)? { let v = if matches!(op, BinOp::Ne) { 1u64 } else { 0u64 }; b.add(Op::Lit(CValue::u64(vec![v])), vec![anchor]) } else { @@ -479,90 +356,134 @@ pub fn compile(term: &Term, b: &mut Builder, env: &[usize], env_shapes: & }) } Term::If { cond, then, els } => { - // `Select` blends per row and is shape-generic, but the branches must agree up to - // ⊥ lanes; genuinely conflicting branch shapes (dynamic typing) defer to rows. - // Sum-shaped results included: `Select` gathers lanes, and the branches commit - // different ones (`if(c, Fwd(x), Bwd(y))` is the shape of every conditional - // constructor), which is what the pin bumped for in #817. - shape_join(&infer_term_shape(then, env_shapes), &infer_term_shape(els, env_shapes))?; - let c = compile(cond, b, env, env_shapes, anchor)?; - let t = compile(then, b, env, env_shapes, anchor)?; - let e = compile(els, b, env, env_shapes, anchor)?; + // `Select` blends per row and is shape-generic; the branches must share one shape, + // and a branch that cannot fix its own (a bare `None`) takes the other's. + let (ts, es) = branch_shapes(then, els, env_shapes, expected)?; + if ts != es { + return Err(format!("if: the branches differ in shape, {ts} vs {es}")); + } + let c = compile(cond, b, env, env_shapes, anchor, None)?; + let t = compile(then, b, env, env_shapes, anchor, Some(&ts))?; + let e = compile(els, b, env, env_shapes, anchor, Some(&es))?; let sel = b.tuple(vec![c, t, e]); - Some(b.add(Op::Select, vec![sel])) + Ok(b.add(Op::Select, vec![sel])) } // Fold over a List. corgi `Op::Fold` consumes `Prod([seed, List])` and folds each row's // list; its body is a closed sub-graph over `Prod([acc, elem])`. DDIR's step sees - // elem=Bound(0), acc=Bound(1), so the body env is [acc, elem] (Bound counts from the top). - // Restriction: the step references only its binders (monoid-style), not outer - // Vars — corgi closes the body; an outer reference would need CapList capture. + // elem=Bound(0), acc=Bound(1). A step that reads outside its two binders gets the + // environment captured into the list first (`CapList`: every element paired with the + // context), so the closed body can see it — the same closure conversion `Case` does. Term::Fold { list, init, step } => { - let init_id = compile(init, b, env, env_shapes, anchor)?; - let list_id = compile(list, b, env, env_shapes, anchor)?; - let elem = match infer_term_shape(list, env_shapes) { Shape::List(e) => *e, _ => return None }; - let init_shape = infer_term_shape(init, env_shapes); - let pair = b.tuple(vec![init_id, list_id]); - let body = compile_fold_body(step, &init_shape, &elem)?; - Some(b.add(Op::Fold(Box::new(body)), vec![pair])) + let init_id = compile(init, b, env, env_shapes, anchor, expected)?; + let list_id = compile(list, b, env, env_shapes, anchor, None)?; + let elem = match shape_of_term(list, env_shapes, None)? { + Shape::List(e) => *e, + other => return Err(format!("fold over a non-list: {other}")), + }; + let init_shape = shape_of_term(init, env_shapes, expected)?; + if mentions_env(step, 2) { + let ctx = b.tuple(env.to_vec()); + let cap_in = b.tuple(vec![ctx, list_id]); + let cap = b.add(Op::CapList, vec![cap_in]); + let pair = b.tuple(vec![init_id, cap]); + let body = compile_fold_body(step, Some(env_shapes), &init_shape, &elem)?; + Ok(b.add(Op::Fold(Box::new(body)), vec![pair])) + } else { + let pair = b.tuple(vec![init_id, list_id]); + let body = compile_fold_body(step, None, &init_shape, &elem)?; + Ok(b.add(Op::Fold(Box::new(body)), vec![pair])) + } } - // Literal-tag sum intro is `Op::Inject` (lane t of a t+1-lane sum); a data-driven tag - // has no static lane count, so it defers to rows. - Term::Inject(tag, payload) => { - let Term::Int(t) = &**tag else { return None }; - let pid = compile(payload, b, env, env_shapes, anchor)?; - Some(b.add(Op::Inject(*t as usize, *t as usize + 1), vec![pid])) + // Sum intro. A literal tag is corgi's `Inject` into the whole declared sum (the lanes the + // payload does not fill are built empty). A data-driven tag is a demux (`Branch`), which + // needs every lane to share the payload's shape. + Term::Inject { tag, payload, sum } => { + let pid = compile(payload, b, env, env_shapes, anchor, None)?; + let pshape = shape_of_term(payload, env_shapes, None)?; + match &**tag { + Term::Int(t) => { + let t = usize::try_from(*t).map_err(|_| format!("constructor tag {t} is negative"))?; + let lanes = lanes_of(sum, t, &pshape, expected)?; + Ok(b.add(Op::Inject(t, lanes), vec![pid])) + } + _ => { + let SumTy::Declared(lanes) = sum else { + return Err("a data-driven variant tag needs a declared type".into()); + }; + if let Some(l) = lanes.iter().find(|l| **l != pshape) { + return Err(format!("variant: every lane must have the payload's shape {pshape}, but one is {l}")); + } + let tid = compile(tag, b, env, env_shapes, anchor, None)?; + let pair = b.tuple(vec![pid, tid]); + Ok(b.add(Op::Branch(lanes.len()), vec![pair])) + } + } } - // Sum elimination: distribute the environment into each committed lane (`CapSum`), run - // each arm as a closed body over `Prod([ctx, payload])` (`MapSum`), and collapse the - // homogeneous result (`Unwrap`). Arms see the outer env plus the payload as the top - // binder; a `default` runs WITHOUT the payload binder (matching `eval`). Arms whose - // result shapes genuinely conflict (dynamic typing) defer to rows, as does a lane with - // neither arm nor default (where `eval` panics). + // Sum elimination: distribute the environment into each lane (`CapSum`), run each arm as a + // closed body over `Prod([ctx, payload])` (`MapSum`), and collapse the homogeneous result + // (`Unwrap`, which is where arms that disagree are reported). Arms see the outer env plus + // the payload as the top binder; a `default` runs WITHOUT the payload binder (matching + // `eval`). An arm that cannot fix its own shape (a bare `None`) takes the first that can. Term::Case { scrutinee, arms, default } => { - let Shape::Sum(lanes) = infer_term_shape(scrutinee, env_shapes) else { return None }; - let sid = compile(scrutinee, b, env, env_shapes, anchor)?; + let lanes = match shape_of_term(scrutinee, env_shapes, None)? { + Shape::Sum(lanes) => lanes, + other => return Err(format!("case on a non-sum: {other}")), + }; + let sid = compile(scrutinee, b, env, env_shapes, anchor, None)?; let ctx = b.tuple(env.to_vec()); let cap_in = b.tuple(vec![ctx, sid]); let cap = b.add(Op::CapSum, vec![cap_in]); - let mut bodies: Vec<(usize, Graph)> = Vec::new(); - let mut result: Option = None; - for (i, lane) in lanes.iter().enumerate() { - let Some(lane_shape) = lane else { continue }; + let arm_graph = |i: usize, exp: Option<&Shape>| -> Res<(Graph, Shape)> { let mut bb = Builder::::default(); let inp = bb.input(); let cnode = bb.add(Op::Field(0), vec![inp]); let mut env2: Vec = (0..env.len()).map(|j| bb.add(Op::Field(j), vec![cnode])).collect(); let mut shapes2: Vec = env_shapes.to_vec(); - let (out, out_shape) = if i < arms.len() { + let term = if i < arms.len() { let pnode = bb.add(Op::Field(1), vec![inp]); env2.push(pnode); - shapes2.push(lane_shape.clone()); - (compile(&arms[i], &mut bb, &env2, &shapes2, inp)?, infer_term_shape(&arms[i], &shapes2)) - } else if let Some(d) = default { - (compile(d, &mut bb, &env2, &shapes2, inp)?, infer_term_shape(d, &shapes2)) + shapes2.push(lanes[i].clone()); + &arms[i] } else { - return None; + default.as_deref().ok_or_else(|| format!("case: no arm for tag {i} and no `_` default"))? }; - result = Some(match result { None => out_shape, Some(prev) => shape_join(&prev, &out_shape)? }); - bodies.push((i, bb.finish(out))); + let out = compile(term, &mut bb, &env2, &shapes2, inp, exp)?; + let g = bb.finish(out); + let in_shape = Shape::Prod(vec![Shape::Prod(env_shapes.to_vec()), lanes[i].clone()]); + let s = corgi::shape_of(&g, &in_shape)?; + Ok((g, s)) + }; + let mut exp: Option = expected.cloned(); + let mut bodies: Vec<(usize, Graph)> = Vec::with_capacity(lanes.len()); + let mut deferred: Vec<(usize, String)> = Vec::new(); + for i in 0..lanes.len() { + match arm_graph(i, exp.as_ref()) { + Ok((g, s)) => { + bodies.push((i, g)); + exp.get_or_insert(s); + } + Err(e) => deferred.push((i, e)), + } } - if bodies.is_empty() { - return None; // an all-⊥ scrutinee shape: nothing to map + for (i, e) in deferred { + let Some(s) = exp.as_ref() else { return Err(e) }; + let (g, _) = arm_graph(i, Some(s))?; + bodies.push((i, g)); } + bodies.sort_by_key(|(i, _)| *i); let mapped = b.add(Op::MapSum(bodies), vec![cap]); - Some(b.add(Op::Unwrap, vec![mapped])) + Ok(b.add(Op::Unwrap, vec![mapped])) } Term::Unary(op, inner) => { - let id = compile(inner, b, env, env_shapes, anchor)?; - Some(match op { + let id = compile(inner, b, env, env_shapes, anchor, None)?; + let shape = shape_of_term(inner, env_shapes, None)?; + Ok(match op { // Wrapping negate on the raw two's-complement bits — exactly `-as_int()`. - // (Order-sensitive use of negatives inherits the crate-wide non-negative-int - // comparison contract; `Neg` adds no new exposure over `Sub` below zero.) UnOp::Neg => b.add(ArithOp::Neg(Kind::U, 64), vec![id]), // `truthy` is "nonzero Int": scalars compare against zero; non-`Int` values // are never truthy, so their `not` folds to the constant 1 (the cross-shape // `Eq` fold's precedent). - UnOp::Not => match infer_term_shape(inner, env_shapes) { + UnOp::Not => match shape { Shape::Prim(_) => { let zero = b.add(Op::Lit(CValue::u64(vec![0])), vec![anchor]); let p = b.tuple(vec![id, zero]); @@ -572,7 +493,7 @@ pub fn compile(term: &Term, b: &mut Builder, env: &[usize], env_shapes: & }, // Tuple arity is static (a shape fact); list length folds `acc + 1` along // each row's list; anything else is the program error `eval` reports. - UnOp::Len => match infer_term_shape(inner, env_shapes) { + UnOp::Len => match shape { Shape::Prod(fs) => b.add(Op::Lit(CValue::u64(vec![fs.len() as u64])), vec![anchor]), Shape::Unit => b.add(Op::Lit(CValue::u64(vec![0])), vec![anchor]), Shape::List(_) => { @@ -587,24 +508,20 @@ pub fn compile(term: &Term, b: &mut Builder, env: &[usize], env_shapes: & }; b.add(Op::Fold(Box::new(body)), vec![seed]) } - _ => return None, + other => return Err(format!("len of a {other}")), }, - // On a sum, every committed lane maps to its constant answer and the result - // unwraps (lanes are homogeneous `U64`); on any other shape, `istag` is - // constantly 0 (matching `eval`'s "non-Variant is never the tag"). - UnOp::IsTag(t) => match infer_term_shape(inner, env_shapes) { + // On a sum, every lane maps to its constant answer and the result unwraps + // (lanes are homogeneous `U64`); on any other shape, `istag` is constantly 0 + // (matching `eval`'s "non-Variant is never the tag"). + UnOp::IsTag(t) => match shape { Shape::Sum(lanes) => { - let arms: Vec<(usize, Graph)> = lanes - .iter() - .enumerate() - .filter_map(|(i, lane)| { - lane.as_ref().map(|_| { - let mut bb = Builder::::default(); - let inp = bb.input(); - let v = (i as u32 == *t) as u64; - let out = bb.add(Op::Lit(CValue::u64(vec![v])), vec![inp]); - (i, bb.finish(out)) - }) + let arms: Vec<(usize, Graph)> = (0..lanes.len()) + .map(|i| { + let mut bb = Builder::::default(); + let inp = bb.input(); + let v = (i as u32 == *t) as u64; + let out = bb.add(Op::Lit(CValue::u64(vec![v])), vec![inp]); + (i, bb.finish(out)) }) .collect(); let mapped = b.add(Op::MapSum(arms), vec![id]); @@ -618,19 +535,15 @@ pub fn compile(term: &Term, b: &mut Builder, env: &[usize], env_shapes: & // the existing kernel matrix, with no per-row work and no new corgi op — `Enlist` each // element (a length-1 lane per row), `Iota` a per-row `[0..k)` tag list, `Weave` // interleaves the lanes in field order into `List`, and `MapList(Unwrap)` - // strips the now-homogeneous sum. A fused list-intro kernel is corgi's call if this - // composition ever profiles hot. - // - // Empty and heterogeneous literals decline: `Weave` needs at least one lane, and - // `Unwrap` needs the committed lanes to join. Rows handle both. + // strips the now-homogeneous sum (and reports a heterogeneous literal). A fused + // list-intro kernel is corgi's call if this composition ever profiles hot. Term::List(fields) => { - let (first, rest) = fields.split_first()?; - rest.iter().try_fold(infer_term_shape(first, env_shapes), |acc, f| { - shape_join(&acc, &infer_term_shape(f, env_shapes)) - })?; + if fields.is_empty() { + return Err("an empty list literal has no element shape".into()); + } let mut lanes = Vec::with_capacity(fields.len()); for f in fields { - let e = compile(f, b, env, env_shapes, anchor)?; + let e = compile(f, b, env, env_shapes, anchor, None)?; lanes.push(b.add(Op::Enlist, vec![e])); } let count = b.add(Op::Lit(CValue::u64(vec![fields.len() as u64])), vec![anchor]); @@ -644,7 +557,7 @@ pub fn compile(term: &Term, b: &mut Builder, env: &[usize], env_shapes: & let out = bb.add(Op::Unwrap, vec![inp]); bb.finish(out) }; - Some(b.add(Op::MapList(Box::new(unwrap_body)), vec![woven])) + Ok(b.add(Op::MapList(Box::new(unwrap_body)), vec![woven])) } // DDIR's `hash` IS corgi's `Op::Hash` (`ir::structural_hash` is the row-wise twin): hash // the arguments as one tuple, shift out the sign bit, reduce by the bound. @@ -654,76 +567,94 @@ pub fn compile(term: &Term, b: &mut Builder, env: &[usize], env_shapes: & // which is larger than the shifted hash, so it reduces to the identity too. Both are // exactly what `ir::eval`'s `if bound > 0` produces. Term::Hash(args) => { - let (bound, rest) = args.split_first()?; - let bid = compile(bound, b, env, env_shapes, anchor)?; + let (bound, rest) = args.split_first().ok_or("hash needs a bound")?; + let bid = compile(bound, b, env, env_shapes, anchor, None)?; let payload = if rest.is_empty() { b.add(Op::Unit, vec![anchor]) } else { let mut ids = Vec::with_capacity(rest.len()); for a in rest { - ids.push(compile(a, b, env, env_shapes, anchor)?); + ids.push(compile(a, b, env, env_shapes, anchor, None)?); } b.tuple(ids) }; let h = b.add(Op::Hash, vec![payload]); let shifted = b.add(ArithOp::Shr(1), vec![h]); let pair = b.tuple(vec![shifted, bid]); - Some(b.add(ArithOp::Bin(CBinOp::Rem, Kind::U, 64), vec![pair])) + Ok(b.add(ArithOp::Bin(CBinOp::Rem, Kind::U, 64), vec![pair])) } - _ => None, } } -/// Compile a `Fold` step into a closed corgi sub-graph over `Prod([acc, elem])`. -/// Env `[acc, elem]` so `Bound(0)`=elem (top), `Bound(1)`=acc — matching `ir::eval`'s Fold. -fn compile_fold_body(step: &Term, init_shape: &Shape, elem_shape: &Shape) -> Option> { +/// Compile a `Fold` step into a closed corgi sub-graph. Without capture the body's input is +/// `Prod([acc, elem])`; with it, `Prod([acc, (ctx, elem)])` where `ctx` is the captured +/// environment (its fields come first, so `Var(i)` and outer `Bound`s resolve as they do in +/// `ir::eval`'s stack). Either way `Bound(0)` = elem, `Bound(1)` = acc. +fn compile_fold_body(step: &Term, ctx: Option<&[Shape]>, init_shape: &Shape, elem_shape: &Shape) -> Res> { let mut bb = Builder::::default(); let inp = bb.input(); let acc = bb.add(Op::Field(0), vec![inp]); - let elem = bb.add(Op::Field(1), vec![inp]); - let out = compile(step, &mut bb, &[acc, elem], &[init_shape.clone(), elem_shape.clone()], inp)?; - Some(bb.finish(out)) + let (mut env, mut shapes) = (Vec::new(), Vec::new()); + let elem = match ctx { + Some(cs) => { + let ce = bb.add(Op::Field(1), vec![inp]); + let c = bb.add(Op::Field(0), vec![ce]); + for (j, s) in cs.iter().enumerate() { + env.push(bb.add(Op::Field(j), vec![c])); + shapes.push(s.clone()); + } + bb.add(Op::Field(1), vec![ce]) + } + None => bb.add(Op::Field(1), vec![inp]), + }; + env.push(acc); + env.push(elem); + shapes.push(init_shape.clone()); + shapes.push(elem_shape.clone()); + let out = compile(step, &mut bb, &env, &shapes, inp, Some(init_shape))?; + Ok(bb.finish(out)) } /// Compile a term in the row environment `Var(0)=key` (shape `kshape`), `Var(1)=val` (`vshape`) — -/// the environment every `LinearOp` reads. The graph's input is `Prod([key, val])`. `None` when -/// the term has no lowering with these shapes; every caller falls back to rows there. -fn compile_over_kv(term: &Term, kshape: &Shape, vshape: &Shape) -> Option> { +/// the environment every `LinearOp` reads. The graph's input is `Prod([key, val])`, and it is +/// typechecked once here, so an `Ok` graph runs on every batch of these shapes. +fn compile_over_kv(term: &Term, kshape: &Shape, vshape: &Shape) -> Res> { let mut b = Builder::::default(); let input = b.input(); let var_k = b.add(Op::Field(0), vec![input]); let var_v = b.add(Op::Field(1), vec![input]); - let out = compile(term, &mut b, &[var_k, var_v], &[kshape.clone(), vshape.clone()], input)?; - Some(b.finish(out)) + let out = compile(term, &mut b, &[var_k, var_v], &[kshape.clone(), vshape.clone()], input, None)?; + let g = b.finish(out); + corgi::shape_of(&g, &Shape::Prod(vec![kshape.clone(), vshape.clone()]))?; + Ok(g) } -/// Compile a `FlatMap`'s list term → a corgi `List` column, one list per input row. Declines when -/// the term is not list-shaped: the backend explodes the column structurally and needs real list -/// bounds to do it, where `ir::eval` would take any `List` value it happened to produce. -pub fn compile_flatmap(list_term: &Term, kshape: &Shape, vshape: &Shape) -> Option> { - matches!(infer_term_shape(list_term, &[kshape.clone(), vshape.clone()]), Shape::List(_)) - .then(|| compile_over_kv(list_term, kshape, vshape)) - .flatten() +/// Compile a `FlatMap`'s list term → a corgi `List` column, one list per input row. A term that is +/// not list-shaped is the type error: the backend explodes the column structurally. +pub fn compile_flatmap(list_term: &Term, kshape: &Shape, vshape: &Shape) -> Res> { + match shape_of_term(list_term, &[kshape.clone(), vshape.clone()], None)? { + Shape::List(_) => compile_over_kv(list_term, kshape, vshape), + other => Err(format!("flatmap over a non-list: {other}")), + } } -/// Compile a scalar term (`EnterAt`'s delay field) → a `U64` column. Declines when the term is not -/// `Prim`-shaped: the delay is read as one integer per row, which a `Prod`/`List`/`Sum` column has -/// no reading of. -pub fn compile_scalar(term: &Term, kshape: &Shape, vshape: &Shape) -> Option> { - matches!(infer_term_shape(term, &[kshape.clone(), vshape.clone()]), Shape::Prim(_)) - .then(|| compile_over_kv(term, kshape, vshape)) - .flatten() +/// Compile a scalar term (`EnterAt`'s delay field) → a `U64` column; a non-integer term is the +/// type error (the delay is read as one integer per row). +pub fn compile_scalar(term: &Term, kshape: &Shape, vshape: &Shape) -> Res> { + match shape_of_term(term, &[kshape.clone(), vshape.clone()], None)? { + Shape::Prim(_) => compile_over_kv(term, kshape, vshape), + other => Err(format!("enter_at delay is not an integer: {other}")), + } } /// Compile a `Filter` predicate → a mask column (nonzero keeps the row). -pub fn compile_predicate(cond: &Term, kshape: &Shape, vshape: &Shape) -> Option> { +pub fn compile_predicate(cond: &Term, kshape: &Shape, vshape: &Shape) -> Res> { compile_over_kv(cond, kshape, vshape) } /// Compile a join projection: key/val Terms over `Var(0)=key`, `Var(1)=val0`, `Var(2)=val1` (with -/// their shapes for `Spread`). Input `Prod([key, val0, val1])`; output `Prod([newkey, newval])`. -/// Join-inline projections are gated by [`compilable`], so the lowering must succeed. -pub fn compile_join_projection(key: &Term, val: &Term, kshape: &Shape, v0shape: &Shape, v1shape: &Shape) -> Graph { +/// their shapes). Input `Prod([key, val0, val1])`; output `Prod([newkey, newval])`. +pub fn compile_join_projection(key: &Term, val: &Term, kshape: &Shape, v0shape: &Shape, v1shape: &Shape) -> Res> { let mut b = Builder::::default(); let input = b.input(); let var_k = b.add(Op::Field(0), vec![input]); @@ -731,26 +662,29 @@ pub fn compile_join_projection(key: &Term, val: &Term, kshape: &Shape, v0shape: let var_1 = b.add(Op::Field(2), vec![input]); let env = [var_k, var_0, var_1]; let shapes = [kshape.clone(), v0shape.clone(), v1shape.clone()]; - let nk = compile(key, &mut b, &env, &shapes, input).expect("join-inline projections are gated by `compilable`"); - let nv = compile(val, &mut b, &env, &shapes, input).expect("join-inline projections are gated by `compilable`"); + let nk = compile(key, &mut b, &env, &shapes, input, None)?; + let nv = compile(val, &mut b, &env, &shapes, input, None)?; let out = b.tuple(vec![nk, nv]); - b.finish(out) + let g = b.finish(out); + corgi::shape_of(&g, &Shape::Prod(shapes.to_vec()))?; + Ok(g) } /// Compile a DDIR `Projection` over `Var(0)=key` (`kshape`), `Var(1)=val` (`vshape`). -/// Input `Prod([key, val])`; output `Prod([newkey, newval])`. `None` when either term (with -/// these shapes) has no lowering; the caller falls back to rows. -pub fn compile_projection(key: &Term, val: &Term, kshape: &Shape, vshape: &Shape) -> Option> { +/// Input `Prod([key, val])`; output `Prod([newkey, newval])`. +pub fn compile_projection(key: &Term, val: &Term, kshape: &Shape, vshape: &Shape) -> Res> { let mut b = Builder::::default(); let input = b.input(); let var_k = b.add(Op::Field(0), vec![input]); let var_v = b.add(Op::Field(1), vec![input]); let env = [var_k, var_v]; let shapes = [kshape.clone(), vshape.clone()]; - let nk = compile(key, &mut b, &env, &shapes, input)?; - let nv = compile(val, &mut b, &env, &shapes, input)?; + let nk = compile(key, &mut b, &env, &shapes, input, None)?; + let nv = compile(val, &mut b, &env, &shapes, input, None)?; let out = b.tuple(vec![nk, nv]); - Some(b.finish(out)) + let g = b.finish(out); + corgi::shape_of(&g, &Shape::Prod(shapes.to_vec()))?; + Ok(g) } #[cfg(test)] @@ -758,27 +692,31 @@ mod tests { use super::*; use crate::ir::Value as V; + fn u64s() -> Shape { Shape::Prim(64) } + fn sum(lanes: Vec) -> Shape { Shape::Sum(lanes) } + /// The pin on DDIR's `hash`: `ir::structural_hash` is a row-at-a-time transcription of /// `corgi::hash`, and the two backends compute the SAME program value, so they must agree /// bit for bit on every shape the transcode layer covers. If corgi's salts or fold change, /// this is what fails. - fn hash_agrees(rows: Vec) { - let shape = infer_shape_cols(&rows); + fn hash_agrees(rows: Vec, shape: Shape) { let col = transcode(&rows, &shape); - let columnar = corgi::hash(&col).into_u64("hash"); + let columnar = corgi::hash(&col).into_u64("hash").unwrap(); let row_wise: Vec = rows.iter().map(crate::ir::structural_hash).collect(); assert_eq!(columnar, row_wise, "hash disagrees (shape {shape:?})"); } #[test] fn hash_matches_corgi_on_scalars() { - hash_agrees(vec![V::Int(0), V::Int(1), V::Int(-1), V::Int(i64::MIN), V::Int(i64::MAX)]); + hash_agrees(vec![V::Int(0), V::Int(1), V::Int(-1), V::Int(i64::MIN), V::Int(i64::MAX)], u64s()); } #[test] fn hash_matches_corgi_on_tuples_and_units() { - hash_agrees(vec![V::Tuple(vec![V::Int(1), V::Int(2)]), V::Tuple(vec![V::Int(2), V::Int(1)])]); - hash_agrees(vec![V::unit(), V::unit()]); + let rows = vec![V::Tuple(vec![V::Int(1), V::Int(2)]), V::Tuple(vec![V::Int(2), V::Int(1)])]; + let shape = shape_of_row(&rows[0]).unwrap(); + hash_agrees(rows, shape); + hash_agrees(vec![V::unit(), V::unit()], Shape::Unit); // A 1-tuple must not collapse onto its scalar, nor a unit onto an empty anything. assert_ne!( crate::ir::structural_hash(&V::Tuple(vec![V::Int(7)])), @@ -788,34 +726,36 @@ mod tests { #[test] fn hash_matches_corgi_on_lists() { - hash_agrees(vec![ - V::List(vec![V::Int(1), V::Int(2), V::Int(3)]), - V::List(vec![]), - V::List(vec![V::Int(3), V::Int(2), V::Int(1)]), - ]); + hash_agrees( + vec![V::List(vec![V::Int(1), V::Int(2), V::Int(3)]), V::List(vec![]), V::List(vec![V::Int(3), V::Int(2), V::Int(1)])], + Shape::List(Box::new(u64s())), + ); } #[test] fn hash_matches_corgi_on_variants() { - hash_agrees(vec![ - V::Variant(0, Box::new(V::Int(5))), - V::Variant(1, Box::new(V::Int(5))), - V::Variant(0, Box::new(V::Int(6))), - ]); + hash_agrees( + vec![V::Variant(0, Box::new(V::Int(5))), V::Variant(1, Box::new(V::Int(5))), V::Variant(0, Box::new(V::Int(6)))], + sum(vec![u64s(), u64s()]), + ); } #[test] fn hash_matches_corgi_on_nesting() { - hash_agrees(vec![ - V::Tuple(vec![V::List(vec![V::Int(1)]), V::Variant(0, Box::new(V::Tuple(vec![V::Int(2), V::Int(3)])))]), - V::Tuple(vec![V::List(vec![V::Int(1), V::Int(1)]), V::Variant(0, Box::new(V::Tuple(vec![V::Int(2), V::Int(4)])))]), - ]); + let pair = Shape::Prod(vec![u64s(), u64s()]); + hash_agrees( + vec![ + V::Tuple(vec![V::List(vec![V::Int(1)]), V::Variant(0, Box::new(V::Tuple(vec![V::Int(2), V::Int(3)])))]), + V::Tuple(vec![V::List(vec![V::Int(1), V::Int(1)]), V::Variant(0, Box::new(V::Tuple(vec![V::Int(2), V::Int(4)])))]), + ], + Shape::Prod(vec![Shape::List(Box::new(u64s())), sum(vec![pair, Shape::Unit])]), + ); } - /// Round-trip a column of rows through infer_shape_cols → transcode → untranscode. - fn roundtrip(rows: Vec) { - let shape = infer_shape_cols(&rows); + /// Round-trip a column of rows through transcode → untranscode at a given shape. + fn roundtrip(rows: Vec, shape: Shape) { let col = transcode(&rows, &shape); + assert_eq!(corgi::shape_of_value(&col), shape, "transcode builds the declared shape"); let back = untranscode(col, &shape); assert_eq!(back, rows, "roundtrip mismatch (shape {shape:?})"); } @@ -823,40 +763,55 @@ mod tests { #[test] fn roundtrip_variant_single_arm() { // binders-style: a single constructor wrapping a list. - roundtrip(vec![ - V::Variant(0, Box::new(V::List(vec![V::Int(1), V::Int(2)]))), - V::Variant(0, Box::new(V::List(vec![V::Int(3)]))), - V::Variant(0, Box::new(V::List(vec![]))), - ]); + roundtrip( + vec![ + V::Variant(0, Box::new(V::List(vec![V::Int(1), V::Int(2)]))), + V::Variant(0, Box::new(V::List(vec![V::Int(3)]))), + V::Variant(0, Box::new(V::List(vec![]))), + ], + sum(vec![Shape::List(Box::new(u64s()))]), + ); } #[test] fn roundtrip_variant_multi_arm() { // adt-style: two arms, interleaved; payloads of different shape per arm. - roundtrip(vec![ - V::Variant(0, Box::new(V::Int(10))), - V::Variant(1, Box::new(V::Tuple(vec![V::Int(1), V::Int(2)]))), - V::Variant(0, Box::new(V::Int(20))), - V::Variant(1, Box::new(V::Tuple(vec![V::Int(3), V::Int(4)]))), - V::Variant(0, Box::new(V::Int(30))), - ]); + roundtrip( + vec![ + V::Variant(0, Box::new(V::Int(10))), + V::Variant(1, Box::new(V::Tuple(vec![V::Int(1), V::Int(2)]))), + V::Variant(0, Box::new(V::Int(20))), + V::Variant(1, Box::new(V::Tuple(vec![V::Int(3), V::Int(4)]))), + V::Variant(0, Box::new(V::Int(30))), + ], + sum(vec![u64s(), Shape::Prod(vec![u64s(), u64s()])]), + ); } #[test] - fn roundtrip_variant_absent_arm_is_bottom() { - // tags {0, 2} present, arm 1 absent → a ⊥ lane; round-trip must still reconstruct. - roundtrip(vec![ - V::Variant(0, Box::new(V::Int(1))), - V::Variant(2, Box::new(V::Int(2))), - V::Variant(0, Box::new(V::Int(3))), - ]); + fn roundtrip_variant_absent_arm_is_an_empty_lane() { + // tags {0, 2} present, arm 1 absent: its lane is an empty column of the declared shape. + roundtrip( + vec![V::Variant(0, Box::new(V::Int(1))), V::Variant(2, Box::new(V::Int(2))), V::Variant(0, Box::new(V::Int(3)))], + sum(vec![u64s(), Shape::List(Box::new(u64s())), u64s()]), + ); } #[test] fn roundtrip_nested_variant_in_tuple() { - roundtrip(vec![ - V::Tuple(vec![V::Int(1), V::Variant(0, Box::new(V::Int(7)))]), - V::Tuple(vec![V::Int(2), V::Variant(1, Box::new(V::unit()))]), - ]); + roundtrip( + vec![ + V::Tuple(vec![V::Int(1), V::Variant(0, Box::new(V::Int(7)))]), + V::Tuple(vec![V::Int(2), V::Variant(1, Box::new(V::unit()))]), + ], + Shape::Prod(vec![u64s(), sum(vec![u64s(), Shape::Unit])]), + ); + } + + #[test] + fn shape_of_row_pins_what_a_row_can_say() { + assert_eq!(shape_of_row(&V::Tuple(vec![V::Int(1), V::List(vec![V::Int(2)])])).unwrap(), Shape::Prod(vec![u64s(), Shape::List(Box::new(u64s()))])); + assert!(shape_of_row(&V::List(vec![])).is_err()); + assert!(shape_of_row(&V::Variant(0, Box::new(V::Int(1)))).is_err()); } } diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs index 7d5aa3d0a..54ae3e5ce 100644 --- a/interactive/src/corgi/reduce.rs +++ b/interactive/src/corgi/reduce.rs @@ -53,7 +53,7 @@ type CBatch = Rc>>; /// columns, allowing Corgi to swizzle their buffers in place when unshared. fn signed_order_view(value: CValue) -> CValue { match value { - value @ CValue::Prim(_) => NumOp::from(ArithOp::ToSigned).eval(value), + value @ CValue::Prim(_) => NumOp::from(ArithOp::ToSigned).eval(value).expect("ToSigned on a leaf"), CValue::Prod(fields) => { CValue::Prod(fields.into_iter().map(signed_order_view).collect()) } @@ -62,7 +62,7 @@ fn signed_order_view(value: CValue) -> CValue { within, variants .into_iter() - .map(|variant| variant.map(signed_order_view)) + .map(signed_order_view) .collect(), ), CValue::List(bounds, values) => { @@ -200,7 +200,7 @@ fn ids(col: &CValue) -> Vec { if let Some(sl) = corgi::arrange::leaf_slice(col) { return sl.to_vec(); } - corgi::hash(col).into_u64("ids") + corgi::hash(col).into_u64("ids").unwrap() } /// Concatenate the records of the `changed` keys across a run of chunks into parallel diff --git a/interactive/src/ir.rs b/interactive/src/ir.rs index 3ff615c1d..78c82aaeb 100644 --- a/interactive/src/ir.rs +++ b/interactive/src/ir.rs @@ -133,7 +133,7 @@ pub fn eval(term: &Term, env: &mut Vec) -> Value { } } } - Term::Inject(tag, t) => Value::Variant(eval(tag, env).as_int() as u32, Box::new(eval(t, env))), + Term::Inject { tag, payload, .. } => Value::Variant(eval(tag, env).as_int() as u32, Box::new(eval(payload, env))), Term::Case { scrutinee, arms, default } => { let Value::Variant(tag, payload) = eval(scrutinee, env) else { panic!("Case scrutinee is not a Variant") diff --git a/interactive/src/parse/mod.rs b/interactive/src/parse/mod.rs index b2bb8f35a..e60cf8798 100644 --- a/interactive/src/parse/mod.rs +++ b/interactive/src/parse/mod.rs @@ -42,10 +42,13 @@ pub enum Term { Spread(Box), /// Product/list elimination: index into a `Tuple` or `List`. Proj(Box, usize), - /// Sum intro: tag a payload. The tag is a `Term` (evaluated to an `Int`), - /// so the constructor can be data-driven — essential for ASTs/JSON whose - /// node kind comes from the data, not a literal. - Inject(Box, Box), + /// Sum intro: tag a payload into a KNOWN sum type. `sum` names every lane's + /// shape (from a `type` declaration, or a built-in `Option`/`Result`), so a + /// column of these is one concrete columnar sum whichever lanes its rows + /// happen to use. The tag is a `Term`: a literal for a constructor call, + /// or data-driven (`variant(Type, tag, payload)`) when every lane of the + /// type shares the payload's shape. + Inject { tag: Box, payload: Box, sum: SumTy }, /// Sum elimination. The scrutinee's payload is pushed as `Bound(0)` for /// the chosen arm. `arms[t]` handles tag `t`; `default` handles the rest. Case { scrutinee: Box, arms: Vec, default: Option> }, @@ -62,6 +65,19 @@ pub enum Term { Hash(Vec), } +/// The sum type an `Inject` builds into. `Declared` carries the full lane shapes of a `type` +/// declaration. The built-ins are shape constructors whose parameter is filled in at compile +/// time from the payload (`Some(x)`, `Ok(x)`, `Err(e)`) or from the other branch of an enclosing +/// `if` (`None`, and the lane the payload does not fill). +#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)] +pub enum SumTy { + Declared(Vec), + /// `Option(T)` = `Sum{ () | T }`: `None` is lane 0, `Some` lane 1. + Option, + /// `Result(T, E)` = `Sum{ T | E }`: `Ok` is lane 0, `Err` lane 1. + Result, +} + #[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)] pub enum UnOp { /// Integer negation. @@ -148,7 +164,7 @@ pub(crate) fn build_builtin(name: &str, args: &mut Vec) -> Term { match name { "tuple" => Term::Tuple(std::mem::take(args)), "list" => Term::List(std::mem::take(args)), - "inject" | "variant" => { assert_eq!(args.len(), 2, "{}(tag, payload)", name); let payload = Box::new(args.remove(1)); Term::Inject(Box::new(args.remove(0)), payload) } + "inject" | "variant" => panic!("`{name}(tag, payload)` has no type: write `variant(Type, tag, payload)` after a `type` declaration, or a constructor `Ctor(payload)`"), "case" => { assert!(args.len() >= 2, "case(scrutinee, arm0, ...)"); let scrutinee = Box::new(args.remove(0)); Term::Case { scrutinee, arms: std::mem::take(args), default: None } } "fold" => { assert_eq!(args.len(), 3, "fold(list, init, step)"); let step = Box::new(args.remove(2)); let init = Box::new(args.remove(1)); let list = Box::new(args.remove(0)); Term::Fold { list, init, step } } "proj" => { assert_eq!(args.len(), 2, "proj(value, index)"); let i = int_arg(&args[1]) as usize; Term::Proj(Box::new(args.remove(0)), i) } diff --git a/interactive/src/parse/pipe.rs b/interactive/src/parse/pipe.rs index 39e9dbea2..061e8cba6 100644 --- a/interactive/src/parse/pipe.rs +++ b/interactive/src/parse/pipe.rs @@ -24,8 +24,8 @@ //! //! Statements: `let x = …;`, `var x = …;` (a feedback variable, for //! recursion), `name: { … }` (a nested scope), `export "name" = …;` (a -//! program output, root scope only). `con Name(arity) = tag;` declares a -//! constructor (see below); it is parse-time only and emits no IR. +//! program output, root scope only). `type Name = V0 shape? | V1 shape? …;` +//! declares a sum type (see below); it is parse-time only and emits no IR. //! //! # Scalar language (`Term`) //! @@ -38,11 +38,18 @@ //! `or(a, b)`, `not(x)`, unary `-x`. //! - Products: `tuple(a, …)`; index with `v[i]` or `proj(v, i)`; `len(v)`. //! - Lists: `list(a, …)`; eliminated by `flatmap` / `collect` / `fold`. -//! - Sums: `inject(tag, payload)` (alias `variant`); test with `istag(tag, v)`. -//! - Constructors (sugar): after `con Name(arity) = tag;`, a call `Name(a, b)` -//! desugars to `inject(tag, tuple(a, b))`. -//! - Pattern match: `case scrut { Ctor(a, b) => arm, …, _ => default }` — each -//! arm binds the matched variant's payload fields by name. +//! - Sums: every sum is a declared type, `type Size = Small u64 | Big (u64, u64) +//! | Empty;` — tags are positions, scoped to the type; a payload shape is +//! `u64`/`int`, `()` (the default when omitted), `(a, b, …)`, `List(a)`, +//! `Option(a)`, `Result(a, b)`, or an earlier type's name. A constructor call +//! `Small(x)` / `Big(a, b)` / `Empty` builds the sum (`Type::Ctor` when two +//! types share a name); `variant(Type, tag, payload)` takes a data-driven tag +//! (every lane must then share the payload's shape); `istag(tag, v)` tests. +//! The built-ins `Some(x)`/`None` and `Ok(x)`/`Err(e)` build `Option`/`Result`; +//! a `None`/`Ok`/`Err` learns its other lane from the other branch of an `if` +//! or the other arms of a `case`. +//! - Pattern match: `case scrut { Ctor(a, b) => arm, …, _ => default }` — an arm +//! names the payload's tuple fields, or the whole payload with one name. //! - Fold: `fold(list, init, step)` — in `step`, `^0` is the element and `^1` //! the accumulator. //! - Binders: `^k` refers to the k-th enclosing `case`/`fold` binder (de @@ -56,7 +63,7 @@ use super::*; #[derive(Debug, Clone, PartialEq)] enum Token { - Let, Var, Export, Con, Case, Fold, + Let, Var, Export, Type, Case, Fold, Input, Import, Key, Map, Join, Min, Distinct, Count, Collect, Arrange, Negate, Filter, EnterAt, LiftIter, FlatMap, Inspect, Ident(String), Int(i64), Str(String), Dollar, Caret, LParen, RParen, LBrace, RBrace, LBracket, RBracket, @@ -116,7 +123,7 @@ fn tokenize(input: &str) -> Vec { while let Some(&c) = chars.peek() { if c.is_ascii_alphanumeric() || c == '_' { ident.push(c); chars.next(); } else { break; } } tokens.push(match ident.as_str() { "let" => Token::Let, "var" => Token::Var, "export" => Token::Export, - "con" => Token::Con, "case" => Token::Case, "fold" => Token::Fold, + "type" => Token::Type, "case" => Token::Case, "fold" => Token::Fold, "input" => Token::Input, "import" => Token::Import, "key" => Token::Key, "map" => Token::Map, "join" => Token::Join, "min" => Token::Min, "distinct" => Token::Distinct, @@ -138,20 +145,24 @@ fn tokenize(input: &str) -> Vec { struct Parser { tokens: Vec, pos: usize, - /// Declared constructors: name -> (tag, arity). Populated by `con` decls, - /// used to desugar `Name(args)` and pattern `case` arms. - cons: std::collections::HashMap, - /// In-scope pattern binders, innermost last: (name, binder-depth, field). - /// A use resolves to `Proj(Bound(cur_depth - binder_depth), field)`, so the - /// de Bruijn index tracks nesting (a fold/case between binding and use). - binders: Vec<(String, usize, usize)>, + /// Declared sum types: name -> the variants in tag order, each with its payload shape. + /// Populated by `type` decls; tags are positions, scoped to their type. + types: std::collections::HashMap>, + /// Constructor name -> every (type, tag) that declares it, for unqualified use. A name + /// declared by two types must be written `Type::Name`. + ctors: std::collections::HashMap>, + /// In-scope pattern binders, innermost last: (name, binder-depth, field). A use resolves to + /// `Bound(cur_depth - binder_depth)`, projected to `field` when the payload is a tuple whose + /// fields the pattern named, so the de Bruijn index tracks nesting (a fold/case between + /// binding and use). + binders: Vec<(String, usize, Option)>, /// Number of `case`/`fold` binders currently in scope. depth: usize, } impl Parser { fn new(tokens: Vec) -> Self { - Parser { tokens, pos: 0, cons: std::collections::HashMap::new(), binders: Vec::new(), depth: 0 } + Parser { tokens, pos: 0, types: Default::default(), ctors: Default::default(), binders: Vec::new(), depth: 0 } } fn peek(&self) -> &Token { &self.tokens[self.pos] } fn next(&mut self) -> Token { let t = self.tokens[self.pos].clone(); self.pos += 1; t } @@ -160,17 +171,24 @@ impl Parser { fn parse_program(&mut self) -> Vec { let mut stmts = Vec::new(); while *self.peek() != Token::Eof && *self.peek() != Token::RBrace { - // `con Name(arity) = tag;` is a parse-time declaration (no IR). - if *self.peek() == Token::Con { + // `type Name = V0 shape? | V1 shape? | …;` is a parse-time declaration (no IR): the + // sum universe. Tags are positions; an omitted payload shape is `()`. + if *self.peek() == Token::Type { self.next(); let name = self.parse_ident(); - self.expect(&Token::LParen); - let arity = match self.next() { Token::Int(n) => n as usize, o => panic!("con: expected arity, got {:?}", o) }; - self.expect(&Token::RParen); self.expect(&Token::Eq); - let tag = match self.next() { Token::Int(n) => n, o => panic!("con: expected tag, got {:?}", o) }; + let mut variants = vec![self.parse_variant_decl()]; + while *self.peek() == Token::Pipe { + self.next(); + variants.push(self.parse_variant_decl()); + } self.expect(&Token::Semi); - self.cons.insert(name, (tag, arity)); + assert!(!self.types.contains_key(&name), "type `{name}` declared twice"); + for (tag, (v, _)) in variants.iter().enumerate() { + assert!(!variants[..tag].iter().any(|(w, _)| w == v), "type `{name}` declares `{v}` twice"); + self.ctors.entry(v.clone()).or_default().push((name.clone(), tag)); + } + self.types.insert(name, variants); continue; } stmts.push(self.parse_stmt()); @@ -181,10 +199,128 @@ impl Parser { /// Resolve a bare name to a pattern binder, if any. fn resolve_binder(&self, name: &str) -> Option { self.binders.iter().rev().find(|(n, _, _)| n == name).map(|&(_, level, field)| { - Term::Proj(Box::new(Term::Bound(self.depth - level)), field) + let bound = Term::Bound(self.depth - level); + match field { + Some(f) => Term::Proj(Box::new(bound), f), + None => bound, + } }) } + /// One `Name shape?` of a `type` declaration. + fn parse_variant_decl(&mut self) -> (String, corgi::Shape) { + let name = self.parse_ident(); + let shape = match self.peek() { + Token::Pipe | Token::Semi => corgi::Shape::Unit, + _ => self.parse_shape(), + }; + (name, shape) + } + + /// A payload shape: `u64`/`int`, `()`, `(a, b, ..)`, `List(a)`, `Option(a)`, `Result(a, b)`, + /// or an earlier `type`'s name (so sums nest; never recursively). + fn parse_shape(&mut self) -> corgi::Shape { + use corgi::Shape; + match self.next() { + Token::LParen => { + if *self.peek() == Token::RParen { + self.next(); + return Shape::Unit; + } + let mut fields = vec![self.parse_shape()]; + while *self.peek() == Token::Comma { + self.next(); + fields.push(self.parse_shape()); + } + self.expect(&Token::RParen); + Shape::Prod(fields) + } + Token::Ident(k) => match k.as_str() { + "u64" | "int" => Shape::Prim(64), + "List" => { + self.expect(&Token::LParen); + let inner = self.parse_shape(); + self.expect(&Token::RParen); + Shape::List(Box::new(inner)) + } + "Option" => { + self.expect(&Token::LParen); + let inner = self.parse_shape(); + self.expect(&Token::RParen); + Shape::Sum(vec![Shape::Unit, inner]) + } + "Result" => { + self.expect(&Token::LParen); + let ok = self.parse_shape(); + self.expect(&Token::Comma); + let err = self.parse_shape(); + self.expect(&Token::RParen); + Shape::Sum(vec![ok, err]) + } + name => Shape::Sum(self.type_lanes(name)), + }, + other => panic!("expected a shape, got {:?}", other), + } + } + + /// The lane shapes of a declared type. + fn type_lanes(&self, name: &str) -> Vec { + self.types + .get(name) + .unwrap_or_else(|| panic!("unknown type `{name}`")) + .iter() + .map(|(_, s)| s.clone()) + .collect() + } + + /// Resolve a constructor name (optionally qualified by its type) to the sum it builds, its + /// tag, and its declared payload shape (`None` for a built-in whose lane the payload fixes). + fn resolve_ctor(&self, ty: Option<&str>, name: &str) -> Option<(SumTy, usize, Option)> { + let (ty, tag) = match ty { + Some(t) => { + let vs = self.types.get(t).unwrap_or_else(|| panic!("unknown type `{t}`")); + let tag = vs.iter().position(|(v, _)| v == name).unwrap_or_else(|| panic!("type `{t}` has no constructor `{name}`")); + (t.to_string(), tag) + } + None => match self.ctors.get(name).map(Vec::as_slice) { + Some([(t, tag)]) => (t.clone(), *tag), + Some(many) => panic!( + "constructor `{name}` is declared by {} types ({}); qualify it as `Type::{name}`", + many.len(), + many.iter().map(|(t, _)| t.as_str()).collect::>().join(", ") + ), + None => { + // the built-ins, unless a declaration shadows them. + return match name { + "None" => Some((SumTy::Option, 0, Some(corgi::Shape::Unit))), + "Some" => Some((SumTy::Option, 1, None)), + "Ok" => Some((SumTy::Result, 0, None)), + "Err" => Some((SumTy::Result, 1, None)), + _ => None, + }; + } + }, + }; + let shape = self.types[&ty][tag].1.clone(); + Some((SumTy::Declared(self.type_lanes(&ty)), tag, Some(shape))) + } + + /// A constructor's payload from its call arguments: a tuple-shaped lane takes one argument + /// per field (or one tuple), a unit lane takes none, any other lane takes exactly one. + fn ctor_payload(name: &str, shape: Option<&corgi::Shape>, mut args: Vec) -> Term { + match shape { + Some(corgi::Shape::Unit) => { + assert!(args.is_empty(), "constructor `{name}` takes no payload, got {} args", args.len()); + Term::Tuple(Vec::new()) + } + Some(corgi::Shape::Prod(fs)) if args.len() == fs.len() && fs.len() != 1 => Term::Tuple(args), + _ => { + assert_eq!(args.len(), 1, "constructor `{name}` takes one payload, got {} args", args.len()); + args.pop().unwrap() + } + } + } + fn parse_stmt(&mut self) -> Stmt { match self.peek().clone() { Token::Let => { self.next(); let n = self.parse_ident(); self.expect(&Token::Eq); let e = self.parse_pipe_expr(); self.expect(&Token::Semi); Stmt::Let(n, e) }, @@ -379,18 +515,24 @@ impl Parser { Token::Fold => self.parse_fold(), Token::Ident(name) => { self.next(); - if *self.peek() == Token::LParen { - if let Some(&(tag, arity)) = self.cons.get(&name) { - // Named constructor: `Name(a, b)` => inject(tag, tuple(a, b)). - let args = self.parse_args(); - assert_eq!(args.len(), arity, "constructor `{}` expects {} args, got {}", name, arity, args.len()); - Term::Inject(Box::new(Term::Int(tag)), Box::new(Term::Tuple(args))) - } else { - self.parse_builtin(&name) - } + // `Type::Ctor` names a constructor by its type. + let (ty, name) = if *self.peek() == Token::ColonColon { + self.next(); + (Some(name), self.parse_ident()) + } else { + (None, name) + }; + if let Some(binder) = ty.is_none().then(|| self.resolve_binder(&name)).flatten() { + binder + } else if let Some((sum, tag, shape)) = self.resolve_ctor(ty.as_deref(), &name) { + // Constructor: `Ctor(a, b)` / `Ctor(x)` / bare `Ctor` => inject into its type. + let args = if *self.peek() == Token::LParen { self.parse_args() } else { Vec::new() }; + let payload = Self::ctor_payload(&name, shape.as_ref(), args); + Term::Inject { tag: Box::new(Term::Int(tag as i64)), payload: Box::new(payload), sum } + } else if *self.peek() == Token::LParen { + self.parse_builtin(&name) } else { - self.resolve_binder(&name) - .unwrap_or_else(|| panic!("unknown name in term: `{}` (not a binder or builtin)", name)) + panic!("unknown name in term: `{}` (not a binder, constructor, or builtin)", name) } } other => panic!("Unexpected token in term: {:?}", other), @@ -428,29 +570,54 @@ impl Parser { self.expect(&Token::Case); let scrutinee = Box::new(self.parse_term()); self.expect(&Token::LBrace); - let mut tagged: Vec<(i64, Term)> = Vec::new(); + let mut tagged: Vec<(usize, Term)> = Vec::new(); let mut default: Option> = None; + let mut lanes: Option = None; // the matched type's arity, from its first arm while *self.peek() != Token::RBrace { let name = self.parse_ident(); if name == "_" { self.expect(&Token::FatArrow); default = Some(Box::new(self.parse_term())); } else { - let (tag, arity) = *self.cons.get(&name) + let (ty, name) = if *self.peek() == Token::ColonColon { + self.next(); + (Some(name), self.parse_ident()) + } else { + (None, name) + }; + let (sum, tag, shape) = self + .resolve_ctor(ty.as_deref(), &name) .unwrap_or_else(|| panic!("unknown constructor in pattern: `{}`", name)); - self.expect(&Token::LParen); + match &lanes { + None => lanes = Some(match &sum { SumTy::Declared(ls) => ls.len(), SumTy::Option | SumTy::Result => 2 }), + Some(n) => assert_eq!(*n, match &sum { SumTy::Declared(ls) => ls.len(), _ => 2 }, "case arms mix types (`{name}`)"), + } let mut names = Vec::new(); - if *self.peek() != Token::RParen { - names.push(self.parse_ident()); - while *self.peek() == Token::Comma { self.next(); names.push(self.parse_ident()); } + if *self.peek() == Token::LParen { + self.next(); + if *self.peek() != Token::RParen { + names.push(self.parse_ident()); + while *self.peek() == Token::Comma { self.next(); names.push(self.parse_ident()); } + } + self.expect(&Token::RParen); } - self.expect(&Token::RParen); - assert_eq!(names.len(), arity, "pattern `{}` expects {} binders, got {}", name, arity, names.len()); + // The binders: one per field of a tuple payload (each a projection), else the + // one name is the payload itself; a unit payload binds nothing. + let fields: Vec> = match shape { + Some(corgi::Shape::Unit) => { + assert!(names.is_empty(), "pattern `{name}` binds nothing, got {} names", names.len()); + Vec::new() + } + Some(corgi::Shape::Prod(fs)) if names.len() == fs.len() && fs.len() != 1 => (0..fs.len()).map(Some).collect(), + _ => { + assert_eq!(names.len(), 1, "pattern `{name}` binds its one payload, got {} names", names.len()); + vec![None] + } + }; self.expect(&Token::FatArrow); - // The matched arm binds the payload; its fields are named. self.depth += 1; let level = self.depth; - for (i, n) in names.iter().enumerate() { self.binders.push((n.clone(), level, i)); } + for (n, f) in names.iter().zip(fields) { self.binders.push((n.clone(), level, f)); } let arm = self.parse_term(); for _ in 0..names.len() { self.binders.pop(); } self.depth -= 1; @@ -459,11 +626,11 @@ impl Parser { if *self.peek() == Token::Comma { self.next(); } } self.expect(&Token::RBrace); - // `Case` indexes arms by tag; fill any gaps with the default (which must - // exist if the matched tags aren't contiguous from 0). - let max_tag = tagged.iter().map(|(t, _)| *t).max().unwrap_or(-1); + // `Case` indexes arms by tag over the WHOLE type; a missing arm takes the default, + // which must then exist. + let n = lanes.unwrap_or_else(|| panic!("case needs at least one constructor arm")); let mut arms: Vec = Vec::new(); - for t in 0..=max_tag { + for t in 0..n { match tagged.iter().find(|(tt, _)| *tt == t) { Some((_, arm)) => arms.push(arm.clone()), None => match &default { @@ -487,9 +654,22 @@ impl Parser { args } - /// Function-call style ADT operators: `tuple`, `list`, `inject`, `case`, + /// Function-call style ADT operators: `tuple`, `list`, `variant`, `case`, /// `fold`, `proj`, `len`, `istag`, `not`, `or`, `if`. fn parse_builtin(&mut self, name: &str) -> Term { + if name == "variant" || name == "inject" { + // `variant(Type, tag, payload)`: a data-driven tag into a declared type. Every lane + // of the type must share the payload's shape (the columnar form is a demux). + self.expect(&Token::LParen); + let ty = self.parse_ident(); + let lanes = self.type_lanes(&ty); + self.expect(&Token::Comma); + let tag = Box::new(self.parse_term()); + self.expect(&Token::Comma); + let payload = Box::new(self.parse_term()); + self.expect(&Token::RParen); + return Term::Inject { tag, payload, sum: SumTy::Declared(lanes) }; + } let mut args = self.parse_args(); super::build_builtin(name, &mut args) } diff --git a/interactive/tests/programs/case_ops.ddp b/interactive/tests/programs/case_ops.ddp index e703d27f5..7b7413047 100644 --- a/interactive/tests/programs/case_ops.ddp +++ b/interactive/tests/programs/case_ops.ddp @@ -1,10 +1,9 @@ --- `case` on the compiled path (CapSum/MapSum/Unwrap), checked against vec: --- an arm using both the payload binder and a captured outer var, a `_ =>` --- default (evaluated without the binder), and a shape-conflicted case (in a --- filter, so the program stays valid) that defers to rows. +-- `case` on the compiled path (CapSum/MapSum/Unwrap), checked against vec: an arm using +-- both the payload binder and a captured outer var, a `_ =>` default (evaluated without the +-- binder), a `case` inside a filter whose arms are both predicates, and the built-in +-- `Option`: `Some`/`None` in the two branches of an `if` (the `if` fixes `None`'s type). -con Small(1) = 0; -con Big(1) = 1; +type Size = Small u64 | Big u64; let pairs = input 0 | key($0[0] ; $0[1]); let tagged = pairs | map($0 ; if($1[0] < 15, Small($1[0]), Big($1[0]))); @@ -12,8 +11,13 @@ let tagged = pairs | map($0 ; if($1[0] < 15, Small($1[0]), Big($1[0]))); let picked = tagged | map($0 ; case $1[0] { Small(x) => x + $0[0], _ => 0 - 1 }); -let clash = tagged - | filter(case $1[0] { Small(x) => x < 12, Big(x) => tuple(x, x) }); +let sieved = tagged + | filter(case $1[0] { Small(x) => x < 12, Big(x) => x > 25 }); + +let maybe = pairs + | map($0 ; if($1[0] < 15, Some($1[0]), None)) + | map($0 ; case $1[0] { Some(x) => x * 2, None => 0 }); export "picked" = picked | arrange | inspect(total); -export "clash" = clash | arrange | inspect(adt); +export "sieved" = sieved | arrange | inspect(adt); +export "maybe" = maybe | arrange | inspect(total); diff --git a/interactive/tests/programs/join_fallback.ddp b/interactive/tests/programs/join_fallback.ddp index 59206e5d2..12aec3e6a 100644 --- a/interactive/tests/programs/join_fallback.ddp +++ b/interactive/tests/programs/join_fallback.ddp @@ -8,7 +8,7 @@ -- because the join has no shapes to reason with. (It previously used `hash`, which stopped -- driving the fallback once `hash` became corgi's structural hash — a test keyed to a hole in -- the compiler rather than to a property of the design.) -con L(1) = 0; +type L = L u64; let left = input 0 | key($0[0] ; $0[1]); let right = input 1 | key($0[0] ; $0[1]); diff --git a/interactive/tests/programs/sum_ops.ddp b/interactive/tests/programs/sum_ops.ddp index 1858acd62..a0f8a507e 100644 --- a/interactive/tests/programs/sum_ops.ddp +++ b/interactive/tests/programs/sum_ops.ddp @@ -1,9 +1,10 @@ --- Sum intro and tag tests on the corgi compiled path, checked against vec: --- literal-tag `variant` (Op::Inject), `istag` on a sum (MapSum + Unwrap, incl. --- an uncommitted-lane tag), and `istag` on a non-sum (constant 0). +-- Sum intro and tag tests on the corgi compiled path, checked against vec: a data-driven +-- `variant(Type, tag, payload)` (Op::Branch over a homogeneous declared type), `istag` on a +-- sum (MapSum + Unwrap, incl. a tag no row carries), and `istag` on a non-sum (constant 0). +type Three = A u64 | B u64 | C u64; let pairs = input 0 | key($0[0] ; $0[1]); let tagged = pairs - | map($0 ; variant(2, $1[0]), istag(2, variant(2, $1[0])), istag(1, variant(2, $1[0])), istag(0, $1)); + | map($0 ; variant(Three, 2, $1[0]), istag(2, variant(Three, 2, $1[0])), istag(1, variant(Three, 2, $1[0])), istag(0, $1)); export "result" = tagged | arrange | inspect(total); diff --git a/interactive/tests/programs/sum_skew.ddp b/interactive/tests/programs/sum_skew.ddp index 23bb961e8..ba3ef99f6 100644 --- a/interactive/tests/programs/sum_skew.ddp +++ b/interactive/tests/programs/sum_skew.ddp @@ -1,21 +1,9 @@ --- Two collections that commit DIFFERENT variant arms, concatenated into one arrangement. --- Neither side's shape names the whole variant universe: one infers `Sum([Some(_)])` (tag 0 --- only), the other `Sum([None, Some(_)])` (tag 1 only) — two arities for one DDIR type. corgi --- reads a differing Sum arity as a type error, so the two must be reconciled with uncommitted --- (⊥) lanes before anything compares or gathers them (corgi's `gather_lanes` fix, DDIR #817). --- --- Both derivations under-approximate, and neither is avoidable here. `infer_term_shape` gives --- `Inject(tag, _)` an arity of `tag + 1`, so `Rare(_)` compiles to one lane and `Common(_)` to --- two; `infer_shape_cols` scanning the data reaches the same place from the other side. A --- single term reconciles its own arms (`If` joins them), but two separate operators have --- nothing to reconcile them, and the declared universe (`con`) that would is discarded at parse. --- --- (There used to be a second copy of this program whose maps wrapped `hash(..)`, on the --- premise that `hash` forced the row-wise path and so exercised the data-derived derivation --- separately. `hash` now lowers, the two copies became one program twice, and the surviving --- one is this.) -con Rare(1) = 0; -con Common(1) = 1; +-- Two collections that build DIFFERENT arms of one declared sum, concatenated into one +-- arrangement. The declaration is the universe: both sides transcode to the whole `Kind` +-- (the arm a side never builds is an empty lane of its declared shape), so the two batches +-- have one shape and merge without reconciliation. This program used to need corgi's +-- uncommitted (⊥) lanes to typecheck (DDIR #817); now it is an ordinary program. +type Kind = Rare u64 | Common u64; let pairs = input 0 | key($0[0] ; $0[1]); From 05efda9d5740154232449d65e2ebaac649104bf7 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Tue, 1 Sep 2026 21:50:58 -0400 Subject: [PATCH 2/4] DDIR corgi reduce: a Collect bracket emits only when its key has input The Collect reducer emitted a row for every bracket, an empty list when all diffs were <= 0. A bracket whose values all cancelled is a key with no input, for which DD invokes no reducer, so the vec backend emits nothing; the corgi backend kept a stale empty list per retracted key. Under iteration this leaked: the e-graph's signature table kept signatures for minted nodes that had since been retracted, and the class map disagreed with vec by exactly those ids. The same all-cancelled bracket was also the source of a `List<()>` column (an empty list with no element shape) that the next batch's `List` could not be concatenated with; the element column now always carries the input values' shape. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_012k2GSwxmvD2LvckkoXi6GK --- interactive/src/corgi/reduce.rs | 29 +++++++++++++++++++---------- 1 file changed, 19 insertions(+), 10 deletions(-) diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs index 54ae3e5ce..85e247bd5 100644 --- a/interactive/src/corgi/reduce.rs +++ b/interactive/src/corgi/reduce.rs @@ -497,22 +497,27 @@ where Reducer::Collect => { // One row per bracket: the values sorted in corgi structural order, // each repeated by its diff, as a `List`. One `sort_blocks` orders every bracket's - // entries at once; element rows are then taken columnar. Every bracket emits (empty - // list if all diffs ≤ 0), matching the row reducer. + // entries at once; element rows are then taken columnar. A bracket emits iff some + // value has NON-ZERO net (as Distinct/Min: DD invokes the reducer only for a key + // with input, and the row reducer then lists the positive copies — an empty list + // when every net is negative). A bracket whose values all cancelled is a key with + // no input: it must emit nothing, or a retracted key keeps a stale (empty) list. let mut entry_reps: Vec = Vec::new(); let mut entry_diffs: Vec = Vec::new(); let mut labels: Vec = Vec::new(); let mut blocks: Vec<(usize, usize)> = Vec::with_capacity(ends.len()); let mut start = 0; for (bi, &end) in ends.iter().enumerate() { - let lo = entry_reps.len(); - for k in start..end { - entry_reps.push(input[k].0); - entry_diffs.push(input[k].1); - labels.push(bi as u64); + if input[start..end].iter().any(|&(_, d)| d != 0) { + let lo = entry_reps.len(); + for k in start..end { + entry_reps.push(input[k].0); + entry_diffs.push(input[k].1); + labels.push(bi as u64); + } + blocks.push((lo, entry_reps.len())); + out_diffs.push(1); } - blocks.push((lo, entry_reps.len())); - out_diffs.push(1); out_ends.push(out_diffs.len()); start = end; } @@ -532,7 +537,11 @@ where } bracket_ends.push(elem_reps.len()); } - let elems = if elem_reps.is_empty() { CValue::Unit(0) } else { gather(&self.in_vals, &elem_reps) }; + // A window whose lists are all empty still has an element SHAPE — the input + // values' — and the column must carry it, or this batch's `List<()>` meets the + // next batch's `List` where the two are concatenated. `gather` at no indices + // is the empty column of that shape. + let elems = gather(&self.in_vals, &elem_reps); let col = CValue::List(Bounds::Offsets(bracket_ends), Box::new(elems)); out_ids = ids(&col); self.register_vals(col, &out_ids); From 4e7016f39f6a45fbb5d42559119fe3797ad88ba8 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Tue, 1 Sep 2026 22:00:41 -0400 Subject: [PATCH 3/4] DDIR: keep the untyped inject(tag, payload) as a value literal for closed terms The server's `feed` writes ADT values as closed terms, `inject(2, tuple(3, 4))`, which the sum-universe commit had turned into a parse panic. It is now `SumTy::Dynamic`: a `Value::Variant` literal the row interpreter evaluates, naming no sum, so the columnar lowering reports it as untyped rather than guessing a shape. Programs build sums from declared types as before. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_012k2GSwxmvD2LvckkoXi6GK --- interactive/src/corgi/logic.rs | 3 ++- interactive/src/parse/mod.rs | 6 +++++- interactive/src/parse/pipe.rs | 11 +++++++---- 3 files changed, 14 insertions(+), 6 deletions(-) diff --git a/interactive/src/corgi/logic.rs b/interactive/src/corgi/logic.rs index e00cd603a..055625be6 100644 --- a/interactive/src/corgi/logic.rs +++ b/interactive/src/corgi/logic.rs @@ -245,6 +245,7 @@ fn lanes_of(sum: &SumTy, tag: usize, payload: &Shape, expected: Option<&Shape>) 1 => Ok(vec![other(0, "Ok")?, payload.clone()]), _ => Err("Result has two variants".into()), }, + SumTy::Dynamic => Err("an untyped `inject(tag, payload)` names no sum; declare a `type` and use its constructor".into()), } } @@ -408,7 +409,7 @@ pub fn compile( } _ => { let SumTy::Declared(lanes) = sum else { - return Err("a data-driven variant tag needs a declared type".into()); + return Err("a data-driven variant tag needs a declared type: `variant(Type, tag, payload)`".into()); }; if let Some(l) = lanes.iter().find(|l| **l != pshape) { return Err(format!("variant: every lane must have the payload's shape {pshape}, but one is {l}")); diff --git a/interactive/src/parse/mod.rs b/interactive/src/parse/mod.rs index e60cf8798..4fb5846a5 100644 --- a/interactive/src/parse/mod.rs +++ b/interactive/src/parse/mod.rs @@ -76,6 +76,10 @@ pub enum SumTy { Option, /// `Result(T, E)` = `Sum{ T | E }`: `Ok` is lane 0, `Err` lane 1. Result, + /// An untyped literal, `inject(tag, payload)`: a `Value::Variant` written as a constant, for + /// closed terms fed to the row interpreter (the server's `feed`). It names no sum, so the + /// columnar lowering rejects it — a program builds sums from declared types. + Dynamic, } #[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)] @@ -164,7 +168,7 @@ pub(crate) fn build_builtin(name: &str, args: &mut Vec) -> Term { match name { "tuple" => Term::Tuple(std::mem::take(args)), "list" => Term::List(std::mem::take(args)), - "inject" | "variant" => panic!("`{name}(tag, payload)` has no type: write `variant(Type, tag, payload)` after a `type` declaration, or a constructor `Ctor(payload)`"), + "inject" | "variant" => { assert_eq!(args.len(), 2, "{}(tag, payload)", name); let payload = Box::new(args.remove(1)); Term::Inject { tag: Box::new(args.remove(0)), payload, sum: SumTy::Dynamic } } "case" => { assert!(args.len() >= 2, "case(scrutinee, arm0, ...)"); let scrutinee = Box::new(args.remove(0)); Term::Case { scrutinee, arms: std::mem::take(args), default: None } } "fold" => { assert_eq!(args.len(), 3, "fold(list, init, step)"); let step = Box::new(args.remove(2)); let init = Box::new(args.remove(1)); let list = Box::new(args.remove(0)); Term::Fold { list, init, step } } "proj" => { assert_eq!(args.len(), 2, "proj(value, index)"); let i = int_arg(&args[1]) as usize; Term::Proj(Box::new(args.remove(0)), i) } diff --git a/interactive/src/parse/pipe.rs b/interactive/src/parse/pipe.rs index 061e8cba6..adb4bf301 100644 --- a/interactive/src/parse/pipe.rs +++ b/interactive/src/parse/pipe.rs @@ -47,7 +47,9 @@ //! (every lane must then share the payload's shape); `istag(tag, v)` tests. //! The built-ins `Some(x)`/`None` and `Ok(x)`/`Err(e)` build `Option`/`Result`; //! a `None`/`Ok`/`Err` learns its other lane from the other branch of an `if` -//! or the other arms of a `case`. +//! or the other arms of a `case`. The untyped `inject(tag, payload)` remains as +//! a VALUE literal for closed terms (the server's `feed`): it names no sum, so +//! the row interpreter takes it and the columnar lowering rejects it. //! - Pattern match: `case scrut { Ctor(a, b) => arm, …, _ => default }` — an arm //! names the payload's tuple fields, or the whole payload with one name. //! - Fold: `fold(list, init, step)` — in `step`, `^0` is the element and `^1` @@ -589,7 +591,7 @@ impl Parser { .resolve_ctor(ty.as_deref(), &name) .unwrap_or_else(|| panic!("unknown constructor in pattern: `{}`", name)); match &lanes { - None => lanes = Some(match &sum { SumTy::Declared(ls) => ls.len(), SumTy::Option | SumTy::Result => 2 }), + None => lanes = Some(match &sum { SumTy::Declared(ls) => ls.len(), SumTy::Option | SumTy::Result => 2, SumTy::Dynamic => unreachable!("patterns name a constructor") }), Some(n) => assert_eq!(*n, match &sum { SumTy::Declared(ls) => ls.len(), _ => 2 }, "case arms mix types (`{name}`)"), } let mut names = Vec::new(); @@ -657,9 +659,10 @@ impl Parser { /// Function-call style ADT operators: `tuple`, `list`, `variant`, `case`, /// `fold`, `proj`, `len`, `istag`, `not`, `or`, `if`. fn parse_builtin(&mut self, name: &str) -> Term { - if name == "variant" || name == "inject" { + if (name == "variant" || name == "inject") && matches!(self.tokens.get(self.pos + 1), Some(Token::Ident(t)) if self.types.contains_key(t)) { // `variant(Type, tag, payload)`: a data-driven tag into a declared type. Every lane - // of the type must share the payload's shape (the columnar form is a demux). + // of the type must share the payload's shape (the columnar form is a demux). The + // two-argument `inject(tag, payload)` stays the untyped value literal (`SumTy::Dynamic`). self.expect(&Token::LParen); let ty = self.parse_ident(); let lanes = self.type_lanes(&ty); From 862b1f1917c32abf4219391494d55c12dec3db1c Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Tue, 1 Sep 2026 22:02:31 -0400 Subject: [PATCH 4/4] DDIR: rename a shadowed binding in the case lowering Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_012k2GSwxmvD2LvckkoXi6GK --- interactive/src/corgi/logic.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/interactive/src/corgi/logic.rs b/interactive/src/corgi/logic.rs index 055625be6..5e20aafb2 100644 --- a/interactive/src/corgi/logic.rs +++ b/interactive/src/corgi/logic.rs @@ -440,7 +440,7 @@ pub fn compile( let cnode = bb.add(Op::Field(0), vec![inp]); let mut env2: Vec = (0..env.len()).map(|j| bb.add(Op::Field(j), vec![cnode])).collect(); let mut shapes2: Vec = env_shapes.to_vec(); - let term = if i < arms.len() { + let arm = if i < arms.len() { let pnode = bb.add(Op::Field(1), vec![inp]); env2.push(pnode); shapes2.push(lanes[i].clone()); @@ -448,7 +448,7 @@ pub fn compile( } else { default.as_deref().ok_or_else(|| format!("case: no arm for tag {i} and no `_` default"))? }; - let out = compile(term, &mut bb, &env2, &shapes2, inp, exp)?; + let out = compile(arm, &mut bb, &env2, &shapes2, inp, exp)?; let g = bb.finish(out); let in_shape = Shape::Prod(vec![Shape::Prod(env_shapes.to_vec()), lanes[i].clone()]); let s = corgi::shape_of(&g, &in_shape)?;