diff --git a/interactive/examples/ddir_vec.rs b/interactive/examples/ddir.rs similarity index 90% rename from interactive/examples/ddir_vec.rs rename to interactive/examples/ddir.rs index 68bd69eb7..a706bd804 100644 --- a/interactive/examples/ddir_vec.rs +++ b/interactive/examples/ddir.rs @@ -1,6 +1,6 @@ -//! DD IR vec-backed driver: parse, lower, render (via `interactive::backend::vec`), execute. +//! The DDIR driver: parse, lower, render on the chosen backend, execute. //! -//! Usage: `ddir_vec [flags] [batch] [rounds] [timely args]` +//! Usage: `ddir [flags] [batch] [rounds] [timely args]` //! //! Inputs: with `EDGES_FILE` set, rows come from that file — one row per line, //! whitespace-separated `i64` fields, assigned round-robin to the program's @@ -12,6 +12,8 @@ //! - `--query=K:V[,q]`: seed the query input with one row (requires --explain). //! - `--debug-demand`: tap every demand collection with an Inspect. //! - `--diag`: serve timely/DD diagnostics on port 51371. +//! - `--backend=vec|corgi`: rendering substrate (default `vec`). The corgi +//! backend is single-worker (its arrange does not exchange). use mimalloc::MiMalloc; @@ -26,6 +28,7 @@ use interactive::lower; use interactive::scope_ir as st; use interactive::ir::{Diff, Value}; use interactive::backend::vec::{render_tree, Row}; +use interactive::backend::corgi::render_tree_rows; #[derive(Clone, Default)] struct Flags { @@ -33,6 +36,7 @@ struct Flags { query: Option, debug_demand: bool, diag: bool, + corgi: bool, } fn run( @@ -55,7 +59,7 @@ fn run( if explain { let source_shapes: Vec<(usize, usize)> = tree.root.imports.iter().map(|imp| match &imp.from { st::Source::Input(_) => (arity, 0usize), - other => panic!("ddir_vec --explain: unsupported source {:?}", other), + other => panic!("ddir --explain: unsupported source {:?}", other), }).collect(); let shape = interactive::explain::export_shape(&tree, &source_shapes); eprintln!("explain: first export shape (k={}, v={}); query is (key[{}] ; val[{}] ++ q)", shape.0, shape.1, shape.0, shape.1); @@ -98,10 +102,14 @@ fn run( let entered: Vec<_> = collections.iter().map(|c| c.clone().enter(inner)).collect(); let root_imports: Vec<_> = tree.root.imports.iter().map(|imp| match &imp.from { st::Source::Input(n) => entered[*n].clone(), - st::Source::Trace(name) => panic!("ddir_vec: Import {:?} not supported in this harness (no trace registry).", name), + st::Source::Trace(name) => panic!("ddir: Import {:?} not supported in this harness (no trace registry).", name), st::Source::Parent(_) => unreachable!("root scope cannot import from a parent"), }).collect(); - let exports = render_tree(&tree.root, inner, 0, root_imports); + let exports = if flags.corgi { + render_tree_rows(&tree.root, inner, 0, root_imports) + } else { + render_tree(&tree.root, inner, 0, root_imports) + }; exports[tree_export_idx].clone().leave(scope) }); output.probe_with(&mut probe); @@ -201,6 +209,9 @@ fn main() { else if let Some(q) = a.strip_prefix("--query=") { flags.query = Some(q.to_string()); } else if a == "--debug-demand" { flags.debug_demand = true; } else if a == "--diag" { flags.diag = true; } + else if let Some(b) = a.strip_prefix("--backend=") { + flags.corgi = match b { "corgi" => true, "vec" => false, other => panic!("unknown backend {other:?} (vec|corgi)") }; + } else { rest.push(a); } } let mut out = vec![prog]; out.extend(rest); @@ -221,7 +232,7 @@ fn main() { }; let (n_inputs, imports) = interactive::survey_sources(&stmts); if !imports.is_empty() { - panic!("ddir_vec: program references imports {:?} but this harness has no trace registry.", imports); + panic!("ddir: program references imports {:?} but this harness has no trace registry.", imports); } let name = std::path::Path::new(&program).file_stem().map(|s| s.to_string_lossy().into_owned()).unwrap_or(program.clone()); let timely_args: Vec = args.iter().skip(4).cloned().collect(); diff --git a/interactive/examples/programs/tour.ddp b/interactive/examples/programs/tour.ddp new file mode 100644 index 000000000..29e9cb790 --- /dev/null +++ b/interactive/examples/programs/tour.ddp @@ -0,0 +1,46 @@ +-- A tour of the DDIR surface: the constructs, composed. Doubles as the grammar's +-- worked example and, in the gate, as the integration check — the per-feature +-- 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; + +let edges = input 0 | key($0[0] ; $0[1]); +let roots = input 1 | key($0[0] ;); + +-- Scalars: arithmetic, (signed) compares, unaries, `hash`, `if`. +let scored = edges + | map($0 ; $1[0] * 2, -$1[0], if($1[0] > 2, 1, 0), hash(97, $0[0])) + | filter(not($1[1] == 0 - 99)); + +-- Sums: tagged intro, `case` with binder + captured var + default, `istag`. +let tagged = edges + | map($0 ; if($0[0] < $1[0], Fwd($1[0]), Bwd($1[0]))) + | map($0 ; case $1[0] { Fwd(x) => x + $0[0], _ => 0 - 1 }, istag(1, $1[0])); + +-- Lists: intro, `fold` (binders ^0/^1), `len`; `flatmap`/`collect` round trip. +let folded = edges + | map($0 ; list($0[0], $1[0], 7)) + | map($0 ; fold($1[0], 0, ^0 + ^1), len($1[0])); +let nested = edges | flatmap(list($1[0], $0[0])) | collect; + +-- Reducers. +let counted = edges | count; +let least = edges | min; +let uniq = edges | distinct; + +-- Iteration: reach, with concat (+) and distinct. +reach: { + let proposals = reach | join(edges, ($2 ;)); + var reach = roots + proposals | distinct; +} + +export "scored" = scored | arrange | inspect(total); +export "sums" = tagged | arrange | inspect(total); +export "folded" = folded | arrange | inspect(total); +export "nested" = nested | arrange | inspect(nest); +export "counts" = counted | arrange | inspect(total); +export "least" = least | arrange | inspect(total); +export "uniq" = uniq | arrange | inspect(total); +export "reach" = reach::reach | key(;) | arrange | inspect(total); diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs index 32805c00a..8435c6827 100644 --- a/interactive/src/backend/corgi.rs +++ b/interactive/src/backend/corgi.rs @@ -41,6 +41,45 @@ type Upd = ((Row, Row), Time, Diff); type CC = CorgiContainer; type CTrace = differential_dataflow::trace::chunk::ChunkSpine>; +/// Rebase a join-projection term from the join environment (`$0`=key, `$1`=left val, +/// `$2`=right val) onto the row environment of the identity join's output +/// (`$0`=key, `$1`=(left val, right val)): `$1 -> $1[0]`, `$2 -> $1[1]`, structurally +/// everywhere. `Bound` binders are scope-relative and pass through untouched. +fn rebase_join_term(t: &crate::parse::Term) -> crate::parse::Term { + use crate::parse::Term::*; + match t { + Var(0) => Var(0), + Var(1) => Proj(Box::new(Var(1)), 0), + Var(2) => Proj(Box::new(Var(1)), 1), + Var(n) => panic!("join projection references ${n}"), + Bound(k) => Bound(*k), + Int(n) => Int(*n), + Tuple(fs) => Tuple(fs.iter().map(rebase_join_term).collect()), + 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))), + Case { scrutinee, arms, default } => Case { + scrutinee: Box::new(rebase_join_term(scrutinee)), + arms: arms.iter().map(rebase_join_term).collect(), + default: default.as_ref().map(|d| Box::new(rebase_join_term(d))), + }, + Fold { list, init, step } => Fold { + list: Box::new(rebase_join_term(list)), + init: Box::new(rebase_join_term(init)), + step: Box::new(rebase_join_term(step)), + }, + If { cond, then, els } => If { + cond: Box::new(rebase_join_term(cond)), + then: Box::new(rebase_join_term(then)), + els: Box::new(rebase_join_term(els)), + }, + Binary(op, l, r) => Binary(*op, Box::new(rebase_join_term(l)), Box::new(rebase_join_term(r))), + Unary(op, inner) => Unary(*op, Box::new(rebase_join_term(inner))), + Hash(args) => Hash(args.iter().map(rebase_join_term).collect()), + } +} + /// Apply a `LinearOp` chain to one corgi container (the corgi-native row-wise compute per batch). /// Project = corgi `eval_graph`; Filter = corgi mask + `gather`; Negate = Rust — all columnar. /// The time/list-shaping ops (EnterAt/LiftIter/FlatMap) take a correctness-first row-wise path @@ -53,46 +92,46 @@ fn apply_ops(mut c: CC, ops: &[LinearOp], level: usize) -> CC { for op in ops { c = match op { - LinearOp::Project(p) if compilable(&p.key) && compilable(&p.val) => { - let (kshape, vshape) = (corgi::shape_of_value(&c.keys), corgi::shape_of_value(&c.vals)); - let 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 } - } - LinearOp::Filter(cond) if compilable(cond) => { - let (kshape, vshape) = (corgi::shape_of_value(&c.keys), corgi::shape_of_value(&c.vals)); - let 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 } - } - // Row-wise fallback (`ir::eval`, parity with `backend::vec`) for terms whose LOWERING - // isn't written yet: `Case`/`Inject`, `List`, `Unary`, `Hash`. Corgi itself models sums - // and lists (`Branch`/`MapSum`/`CapSum`/`Unwrap`, `Enlist`/`MapList`/`Fold`); only - // list-intro (and possibly hash/len ops) may need kernels. See `logic::compilable`. LinearOp::Project(p) => { - 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)); + 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 (`Case` with conflicting arms, list + // intro, `hash`...). Corgi models sums and lists; `hash` is the one kernel gap + // (splitmix parity needs lane-wise xor and integer rem). + 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) } - CorgiContainer::from_updates(out) } LinearOp::Filter(cond) => { - 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)); } + 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) } - CorgiContainer::from_updates(out) } LinearOp::Negate => { for d in c.diffs.iter_mut() { @@ -185,6 +224,10 @@ impl Backend for CorgiBackend { } fn arrange<'s>(c: Collection<'s, Time, CC>) -> Self::Arr<'s> { + // Single-worker guard: this arrange is `Pipeline` (no key exchange), so multi-worker + // execution would MIS-PLACE keys — silently wrong, not slow. A columnar exchange + // (radix partition by key hash) lifts this; it pairs with the stored-hash-column work. + assert_eq!(c.inner.scope().peers(), 1, "the corgi backend is single-worker: arrange does not exchange keys"); // Column-native ingest: `CorgiChunker` sort-consolidates each input `CorgiContainer`'s // columns straight into a `CorgiChunk` (no drain-to-rows), then the standard chunk batcher + // builder. No columns→rows→columns round-trip at the arrangement boundary. @@ -226,8 +269,23 @@ impl Backend for CorgiBackend { // The proxy-join seam drives the backend blockwise under the driver's fuel; the backend // compiles the projection per container (shape-directed, for `Spread`) and emits corgi // columns directly as `CorgiContainer`s — column-native, no row round-trip. - let tactic = ProxyJoinTactic::new(CorgiJoinBackend::new(projection.key.clone(), projection.val.clone())); - join_with_tactic::<_, _, _, CC>(l, r, tactic).as_collection() + if compilable(&projection.key) && compilable(&projection.val) { + let tactic = ProxyJoinTactic::new(CorgiJoinBackend::new(projection.key.clone(), projection.val.clone())); + join_with_tactic::<_, _, _, CC>(l, r, tactic).as_collection() + } else { + // Projections the lowering can't compile take the same shape as `linear`'s gate: + // join with the identity projection (compilable by construction), then apply the + // original terms as a row-wise `Project`, rebased from the join env + // `[$0=key, $1=left val, $2=right val]` onto the row env `[$0=key, $1=(lv, rv)]`. + // Capability never depends on the lowering's coverage; only speed does. + use crate::parse::Term; + let key = Term::Var(0); + let val = Term::Tuple(vec![Term::Var(1), Term::Var(2)]); + let tactic = ProxyJoinTactic::new(CorgiJoinBackend::new(key, val)); + let joined = join_with_tactic::<_, _, _, CC>(l, r, tactic).as_collection(); + let rebased = Projection { key: rebase_join_term(&projection.key), val: rebase_join_term(&projection.val) }; + Self::linear(joined, vec![LinearOp::Project(rebased)], 0) + } } fn reduce<'s>(a: Self::Arr<'s>, reducer: &Reducer) -> Self::Arr<'s> { @@ -297,6 +355,49 @@ pub fn render_tree<'s>( crate::backend::render_tree::(s, scope, depth, imports) } +/// Render `s` with the corgi substrate over ROW collections: each import converts to corgi +/// containers at the boundary (`ToCorgi`), the tree renders columnar, and each export +/// converts back (`FromCorgi`). Signature-compatible with +/// [`vec::render_tree`](crate::backend::vec::render_tree) (hence the `vec::Col` alias), so a +/// row-speaking driver switches backends by switching this one call. +pub fn render_tree_rows<'s>( + s: &st::Scope, + scope: Scope<'s, Time>, + depth: usize, + imports: Vec>, +) -> Vec> { + let corgi_imports: Vec> = imports + .into_iter() + .map(|c| { + c.inner + .unary(Pipeline, "ToCorgi", |_, _| { + |input, output| { + input.for_each(|cap, data| { + let mut cc = CorgiContainer::from_updates(std::mem::take(data)); + output.session(&cap).give_container(&mut cc); + }); + } + }) + .as_collection() + }) + .collect(); + render_tree(s, scope, depth, corgi_imports) + .into_iter() + .map(|c| { + c.inner + .unary(Pipeline, "FromCorgi", |_, _| { + |input, output| { + input.for_each(|cap, data| { + let mut rows = std::mem::take(data).into_updates(); + output.session(&cap).give_container(&mut rows); + }); + } + }) + .as_collection() + }) + .collect() +} + /// Evaluate `program` on explicit inputs via the **corgi** backend (mirrors [`crate::backend::vec::evaluate`]). /// /// Inputs/exports cross the iterative-scope boundary as Vec rows (which support refinement @@ -309,11 +410,8 @@ pub fn evaluate( use std::collections::BTreeMap; use std::sync::mpsc::channel; use timely::dataflow::operators::core::capture::{Capture, Event}; - use timely::dataflow::operators::generic::Operator; use differential_dataflow::input::Input; - use differential_dataflow::AsCollection; use differential_dataflow::dynamic::pointstamp::PointStamp; - use crate::corgi::container::CorgiContainer; let names: Vec = program.root.exports.iter().map(|e| e.name.clone()).collect(); let mut txs = Vec::new(); @@ -336,49 +434,22 @@ pub fn evaluate( collections.push(c); } let exports = scope.iterative::, _, _>(|inner| { - // Enter Vec collections (refinement), then convert each to a corgi container. - let mut corgi_imports = Vec::new(); - for c in collections.iter().map(|c| c.clone().enter(inner)) { - let cs = c - .inner - .unary(Pipeline, "ToCorgi", |_, _| { - |input, output| { - input.for_each(|cap, data| { - let mut cc = CorgiContainer::from_updates(std::mem::take(data)); - output.session(&cap).give_container(&mut cc); - }); - } - }) - .as_collection(); - corgi_imports.push(cs); - } + // Enter row collections (refinement); rows convert to corgi containers and + // back inside `render_tree_rows`. + let entered: Vec<_> = collections.iter().map(|c| c.clone().enter(inner)).collect(); let root_imports: Vec<_> = program .root .imports .iter() .map(|imp| match &imp.from { - st::Source::Input(n) => corgi_imports[*n].clone(), + st::Source::Input(n) => entered[*n].clone(), other => panic!("corgi evaluate: unsupported source {other:?}"), }) .collect(); - let exports = render_tree(&program.root, inner.clone(), 0, root_imports); - // Convert corgi exports back to Vec rows, then leave the scope. - let mut leaved = Vec::new(); - for c in exports { - let rows = c - .inner - .unary(Pipeline, "FromCorgi", |_, _| { - |input, output| { - input.for_each(|cap, data| { - let mut rows = std::mem::take(data).into_updates(); - output.session(&cap).give_container(&mut rows); - }); - } - }) - .as_collection(); - leaved.push(rows.leave(scope)); - } - leaved + render_tree_rows(&program.root, inner.clone(), 0, root_imports) + .into_iter() + .map(|rows| rows.leave(scope)) + .collect::>() }); for (col, tx) in exports.into_iter().zip(txs) { col.inner.capture_into(tx); diff --git a/interactive/src/corgi/join.rs b/interactive/src/corgi/join.rs index 9ac4f6424..82e514075 100644 --- a/interactive/src/corgi/join.rs +++ b/interactive/src/corgi/join.rs @@ -87,10 +87,10 @@ impl ProxyJoinBackend, CBatch> for CorgiJoinBackend *from = None; return; } - if single_lane_keyed(&chunks0) && single_lane_keyed(&chunks1) { - advance_leaf(&chunks0, &chunks1, &instance.lower, from, bridge0, bridge1); - } else { - advance_structured(&chunks0, &chunks1, &instance.lower, from, bridge0, bridge1); + match (leaf_key_lanes(&chunks0), leaf_key_lanes(&chunks1)) { + (Some(1), Some(1)) => advance_leaf(&chunks0, &chunks1, &instance.lower, from, bridge0, bridge1), + (Some(_), Some(_)) => advance_lanes(&chunks0, &chunks1, &instance.lower, from, bridge0, bridge1), + _ => advance_structured(&chunks0, &chunks1, &instance.lower, from, bridge0, bridge1), } } @@ -163,10 +163,21 @@ fn leaf_lanes(col: &CValue) -> Option> { if walk(col, &mut out) { Some(out) } else { None } } -/// Whether every nonempty chunk's key column is a single `u64` leaf lane, making the key's -/// own value the group token (chunk order IS `u64` order for a one-lane lexicographic key). -fn single_lane_keyed(chunks: &[&CorgiChunk]) -> bool { - chunks.iter().filter(|c| c.len() > 0).all(|c| leaf_lanes(c.keys()).is_some_and(|l| l.len() == 1)) +/// The key columns' common leaf-lane count: `Some(n)` when every nonempty chunk's key +/// flattens to exactly `n` 64-bit lanes. `Some(1)` keys use their own value as the group +/// token (chunk order IS `u64` order); `Some(n>1)` keys walk the lane-tuple path (ordinal +/// tokens, one block); `None` keys (sums/lists in the key) take the structural walk. +fn leaf_key_lanes(chunks: &[&CorgiChunk]) -> Option { + let mut lanes: Option = None; + for c in chunks.iter().filter(|c| c.len() > 0) { + let n = leaf_lanes(c.keys())?.len(); + match lanes { + None => lanes = Some(n), + Some(m) if m == n => {} + _ => return None, + } + } + lanes } /// Whether every nonempty chunk's val column flattens to `u64` leaf lanes. @@ -667,10 +678,210 @@ fn leaf_merge<'a, T: ColTime>( } } -/// Fallback `advance` for structured keys, which have no order-preserving `u64` embedding to -/// resume by: the whole intersection in one block, group tokens an ordinal counter (block- -/// scoped; both sides named by this single walk). Output is still cut at `TARGET_OUT` by -/// `cross`; only the bounded-bridge property is forgone. +/// A fully-pulled view over one chunk for the LANE-TUPLE key path: every key lane (and, +/// when leaf-shaped, val lane) as `u64` buffers, so the key walk is lexicographic machine +/// compares — no per-row structural dispatch. `side` routes emission; tuples have no +/// order-preserving `u64` embedding, so this path runs as ONE block with ordinal tokens +/// (resumable blocking for tuples would need a digest scheme; the walk, not the blocking, +/// is what scales). +struct LaneView<'a, T: ColTime> { + chunk: &'a CorgiChunk, + cid: usize, + side: usize, + keys: Vec>, + vals: Option>>, + cur: usize, +} + +impl<'a, T: ColTime> LaneView<'a, T> { + fn new(chunk: &'a CorgiChunk, cid: usize, side: usize, leaf_vals: bool) -> Self { + let idx: Vec = (0..chunk.len()).collect(); + let keys = pull_lanes(chunk.keys(), &idx); + let vals = leaf_vals.then(|| pull_lanes(chunk.vals(), &idx)); + LaneView { chunk, cid, side, keys, vals, cur: 0 } + } + fn exhausted(&self) -> bool { + self.cur >= self.chunk.len() + } + fn run_ref(&self, s: usize, e: usize) -> RunRef<'_, T> { + RunRef { chunk: self.chunk, cid: self.cid, s, e, vals: self.vals.as_ref().map(|lanes| (&lanes[..], 0)) } + } +} + +/// Lexicographic compare of `views[a]`'s row `ai` against `views[b]`'s row `bi`. +fn lane_key_cmp(views: &[LaneView<'_, T>], a: usize, ai: usize, b: usize, bi: usize) -> Ordering { + for (la, lb) in views[a].keys.iter().zip(views[b].keys.iter()) { + match la[ai].cmp(&lb[bi]) { + Ordering::Equal => {} + other => return other, + } + } + Ordering::Equal +} + +/// One past the end of the run of rows equal to row `s` in `views[v]`. +fn lane_run_end(views: &[LaneView<'_, T>], v: usize, s: usize) -> usize { + let len = views[v].chunk.len(); + let mut e = s + 1; + while e < len && lane_key_cmp(views, v, e, v, s) == Ordering::Equal { + e += 1; + } + e +} + +/// `advance` for lane-tuple keys: the whole intersection in one call (ordinal group tokens, +/// ascending with key order), with the same two regimes as the single-lane path — a much +/// smaller side DRIVES and the other is probed at the driver's keys (`find_ranges` with +/// gathered tuple needles, so per-round cost tracks the delta); comparable sides merge +/// symmetrically on the pulled lane buffers. +fn advance_lanes( + chunks0: &[&CorgiChunk], + chunks1: &[&CorgiChunk], + lower: &T, + from: &mut Option, + bridge0: &mut ProxyBridge, + bridge1: &mut ProxyBridge, +) { + *from = None; // one block: tuples have no resumable `u64` embedding + let (r0, r1): (usize, usize) = (chunks0.iter().map(|c| c.len()).sum(), chunks1.iter().map(|c| c.len()).sum()); + if r0 == 0 || r1 == 0 { + return; + } + fn views_of<'a, T: ColTime>(chunks: &[&'a CorgiChunk], side: usize) -> Vec> { + let leaf_vals = leaf_valued(chunks); + chunks.iter().enumerate().filter(|(_, c)| c.len() > 0).map(|(cid, c)| LaneView::new(c, cid, side, leaf_vals)).collect() + } + + if r0.max(r1) >= 2 * r0.min(r1) { + // Lopsided: walk only the DRIVER's keys; probe the other side wholesale. + let drive0 = r0 <= r1; + let (dchunks, pchunks) = if drive0 { (chunks0, chunks1) } else { (chunks1, chunks0) }; + let mut dviews = views_of(dchunks, 0); + // Collect the driver's distinct keys (as gather coordinates for the needle column) + // and each key's runs. + let mut needle_tags: Vec = Vec::new(); + let mut needle_offs: Vec = Vec::new(); + let mut druns: Vec<(usize, usize, usize, usize)> = Vec::new(); // (key idx, view, s, e) + loop { + let mut min: Option = None; + for v in 0..dviews.len() { + if dviews[v].exhausted() { + continue; + } + min = Some(match min { + None => v, + Some(m) if lane_key_cmp(&dviews, v, dviews[v].cur, m, dviews[m].cur) == Ordering::Less => v, + Some(m) => m, + }); + } + let Some(m) = min else { break }; + let j = needle_tags.len(); + needle_tags.push(m); + needle_offs.push(dviews[m].cur); + let ends: Vec<(usize, usize, usize)> = (0..dviews.len()) + .filter(|&v| !dviews[v].exhausted() && lane_key_cmp(&dviews, v, dviews[v].cur, m, dviews[m].cur) == Ordering::Equal) + .map(|v| (v, dviews[v].cur, lane_run_end(&dviews, v, dviews[v].cur))) + .collect(); + for (v, ss, e) in ends { + druns.push((j, v, ss, e)); + dviews[v].cur = e; + } + } + if needle_tags.is_empty() { + return; + } + // One tuple-shaped needle column, one batched probe per probee chunk. + let key_srcs: Vec> = dviews.iter().map(|v| Some(v.chunk.keys())).collect(); + let needles = gather_lanes(&key_srcs, &needle_tags, &needle_offs); + let pvleaf = leaf_valued(pchunks); + let probes: Vec> = pchunks.iter().enumerate() + .filter(|(_, c)| c.len() > 0) + .map(|(cid, c)| Probe::new(c, cid, &needles, pvleaf)) + .collect(); + let (mut sd, mut sp) = (SideScratch::new(), SideScratch::new()); + let (bd, bp) = if drive0 { (bridge0, bridge1) } else { (bridge1, bridge0) }; + let mut drun_at = 0usize; + let mut refs: Vec> = Vec::new(); + let mut token = 0u64; + for j in 0..needle_tags.len() { + refs.clear(); + while drun_at < druns.len() && druns[drun_at].0 == j { + let (_, v, ss, e) = druns[drun_at]; + refs.push(dviews[v].run_ref(ss, e)); + drun_at += 1; + } + let dref_count = refs.len(); + refs.extend(probes.iter().filter_map(|p| p.run_ref(j))); + if refs.len() == dref_count { + continue; + } + let (drefs, prefs) = refs.split_at(dref_count); + sd.stage_runs(drefs, lower); + sp.stage_runs(prefs, lower); + if sd.entries.is_empty() || sp.entries.is_empty() { + continue; + } + sd.emit(token, bd); + sp.emit(token, bp); + token += 1; + } + } else { + // Comparable sides: one tagged view set, symmetric lexicographic merge. + let mut views = views_of(chunks0, 0); + views.extend(views_of(chunks1, 1)); + let (mut s0, mut s1) = (SideScratch::new(), SideScratch::new()); + let mut token = 0u64; + loop { + let mut min: Option = None; + for v in 0..views.len() { + if views[v].exhausted() { + continue; + } + min = Some(match min { + None => v, + Some(m) if lane_key_cmp(&views, v, views[v].cur, m, views[m].cur) == Ordering::Less => v, + Some(m) => m, + }); + } + let Some(m) = min else { break }; + let ends: Vec<(usize, usize, usize)> = (0..views.len()) + .filter(|&v| !views[v].exhausted() && lane_key_cmp(&views, v, views[v].cur, m, views[m].cur) == Ordering::Equal) + .map(|v| (v, views[v].cur, lane_run_end(&views, v, views[v].cur))) + .collect(); + let both_sides = { + // Reads only: the run refs borrow `views` and end with this block. + let mut refs0: Vec> = Vec::new(); + let mut refs1: Vec> = Vec::new(); + for &(v, ss, e) in &ends { + let r = views[v].run_ref(ss, e); + if views[v].side == 0 { refs0.push(r) } else { refs1.push(r) } + } + let both = !refs0.is_empty() && !refs1.is_empty(); + if both { + s0.stage_runs(&refs0, lower); + s1.stage_runs(&refs1, lower); + } + both + }; + for (v, _, e) in ends { + views[v].cur = e; + } + if !both_sides { + continue; + } + if s0.entries.is_empty() || s1.entries.is_empty() { + continue; + } + s0.emit(token, bridge0); + s1.emit(token, bridge1); + token += 1; + } + } +} + +/// Last-resort `advance` for keys that do not flatten to integer lanes at all (sums or +/// lists in the key): the whole intersection in one block, ordinal tokens, and a structural +/// (`compare_at`) walk. Rare by construction — tuple keys take [`advance_lanes`]. fn advance_structured( chunks0: &[&CorgiChunk], chunks1: &[&CorgiChunk], diff --git a/interactive/src/corgi/logic.rs b/interactive/src/corgi/logic.rs index 227ad7e0e..5866d51f4 100644 --- a/interactive/src/corgi/logic.rs +++ b/interactive/src/corgi/logic.rs @@ -2,13 +2,15 @@ //! transcode DDIR rows (`ir::Value`) to/from corgi columnar `Value`, directed by a `Shape` //! inferred from the data (the dynamic-typing primitive). //! -//! The compiler (`compile`) covers Var/Bound/Int/Tuple(+Spread)/Proj/Binary/If/Fold over non-negative -//! ints; terms it can't lower (List, Case/Inject, Unary, Hash — see `compilable`) fall back to row-wise -//! `ir::eval` in the backend. The transcode layer is total over `Shape` (Prim/Unit/Prod/List/Sum), so a +//! The compiler (`compile`) covers Var/Bound/Int/Tuple(+Spread)/Proj/Binary/If/Fold and the +//! Neg/Not/Len 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). Terms it can't lower (List, Case/Inject, +//! IsTag, Hash — see `compilable`) 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). use crate::ir::Value as DValue; -use crate::parse::{BinOp, Term}; +use crate::parse::{BinOp, Term, UnOp}; use corgi::{ArithOp, BinOp as CBinOp, Builder, CmpOp, Graph, Kind, NumOp, Op, Pred, Shape, Value as CValue}; @@ -253,17 +255,87 @@ fn infer_term_shape(t: &Term, env_shapes: &[Shape]) -> Shape { } if fs.is_empty() { Shape::Unit } else { Shape::Prod(fs) } } - Term::If { then, .. } => infer_term_shape(then, env_shapes), + 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::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)) + } + 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), } } +/// Whether a shape contains a `Sum` anywhere (see the `If` lowering's engine caveat). +fn shape_has_sum(s: &Shape) -> bool { + match s { + Shape::Sum(_) => true, + Shape::Prod(fs) => fs.iter().any(shape_has_sum), + Shape::List(e) => shape_has_sum(e), + _ => false, + } +} + +/// 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)) + } + _ => None, + } +} + /// Whether [`compile`] can lower this term to a corgi graph. Terms whose lowering is not yet -/// written — `Inject`/`Case` (corgi has `Branch`/`MapSum`/`CapSum`/`Unwrap`), `List` (intro may -/// need a kernel), `Unary`, `Hash` — return false, and the backend falls back to row-wise -/// `ir::eval` (parity with `backend::vec`). The gap is compiler debt here, not expressiveness -/// in corgi. +/// written — `Inject`/`Case` and `IsTag` (corgi has `Branch`/`MapSum`/`CapSum`/`Unwrap`), +/// `List` (intro may need a kernel) — return false, and the backend falls back to row-wise +/// `ir::eval` (parity with `backend::vec`); that gap is compiler debt here, not expressiveness +/// in corgi. `Hash` is the one true kernel gap: exact splitmix64 parity with `ir::eval` needs +/// lane-wise xor and integer rem, which corgi's arithmetic does not yet have. pub fn compilable(t: &Term) -> bool { match t { Term::Var(_) | Term::Bound(_) | Term::Int(_) => true, @@ -272,24 +344,34 @@ pub fn compilable(t: &Term) -> bool { Term::Binary(_, l, r) => compilable(l) && compilable(r), Term::If { cond, then, els } => compilable(cond) && compilable(then) && compilable(els), Term::Fold { list, init, step } => compilable(list) && compilable(init) && compilable(step), - _ => false, // List, Inject, Case, Unary, Hash — row-wise fallback. + 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), + // `Case` deliberately stays false HERE: this shape-free check gates join-INLINE + // projections only, and `Case` needs shapes (arm homogeneity). The join defers such + // projections to a linear stage, whose shape-aware `compile` lowers them there. + _ => false, // List intro, Case (here), data-driven Inject, Hash — 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) -> usize { +pub fn compile(term: &Term, b: &mut Builder, env: &[usize], env_shapes: &[Shape], anchor: usize) -> Option { match term { - Term::Var(i) => env[*i], - Term::Bound(k) => env[env.len() - 1 - *k], - Term::Int(n) => b.add(Op::Lit(CValue::u64(vec![*n as u64])), vec![anchor]), + // 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::Tuple(fields) => { // A `Spread(place)` child splices the place'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); + let node = compile(inner, b, env, env_shapes, anchor)?; match shape_of_place(inner, env_shapes) { Shape::Prod(fs) => { for i in 0..fs.len() { @@ -300,26 +382,26 @@ 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)?), } } // 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() { - b.add(Op::Unit, vec![anchor]) + Some(b.add(Op::Unit, vec![anchor])) } else { - b.tuple(ids) + Some(b.tuple(ids)) } } Term::Proj(t, i) => { - let id = compile(t, b, env, env_shapes, anchor); - 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)?; + let rid = compile(r, b, env, env_shapes, anchor)?; let pair = |b: &mut Builder, x, y| b.tuple(vec![x, y]); - match op { + Some(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]) } @@ -335,20 +417,32 @@ pub fn compile(term: &Term, b: &mut Builder, env: &[usize], env_shapes: & b.add(CmpOp::Rel(pred), vec![p]) } } - BinOp::Lt => { let p = pair(b, lid, rid); b.add(CmpOp::Rel(Pred::Lt), vec![p]) } - BinOp::Le => { let p = pair(b, lid, rid); b.add(CmpOp::Rel(Pred::Le), vec![p]) } - BinOp::Gt => { let p = pair(b, rid, lid); b.add(CmpOp::Rel(Pred::Lt), vec![p]) } - BinOp::Ge => { let p = pair(b, rid, lid); b.add(CmpOp::Rel(Pred::Le), vec![p]) } + // Ordered compares go through `ToSigned` (XOR the sign bit: the order-preserving + // signed encoding), so they agree with `ir::eval`'s signed semantics for negative + // ints too. `Eq`/`Ne` are bit-equality — sign-safe as raw bits. + BinOp::Lt => { let (ls, rs) = (b.add(ArithOp::ToSigned, vec![lid]), b.add(ArithOp::ToSigned, vec![rid])); let p = pair(b, ls, rs); b.add(CmpOp::Rel(Pred::Lt), vec![p]) } + BinOp::Le => { let (ls, rs) = (b.add(ArithOp::ToSigned, vec![lid]), b.add(ArithOp::ToSigned, vec![rid])); let p = pair(b, ls, rs); b.add(CmpOp::Rel(Pred::Le), vec![p]) } + BinOp::Gt => { let (ls, rs) = (b.add(ArithOp::ToSigned, vec![lid]), b.add(ArithOp::ToSigned, vec![rid])); let p = pair(b, rs, ls); b.add(CmpOp::Rel(Pred::Lt), vec![p]) } + BinOp::Ge => { let (ls, rs) = (b.add(ArithOp::ToSigned, vec![lid]), b.add(ArithOp::ToSigned, vec![rid])); let p = pair(b, rs, ls); b.add(CmpOp::Rel(Pred::Le), vec![p]) } BinOp::And => { let p = pair(b, lid, rid); b.add(CmpOp::Min, vec![p]) } BinOp::Or => { let p = pair(b, lid, rid); b.add(CmpOp::Max, vec![p]) } - } + }) } Term::If { cond, then, els } => { - 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, but the branches must agree up to + // ⊥ lanes; genuinely conflicting branch shapes (dynamic typing) defer to rows. + // Sum-shaped results also defer for now: merging sum columns that commit different + // lanes trips an offset bug in the pinned engine's lane merge (engine.rs + // `sum_from_prim` path) — revisit at the next corgi pin bump. + let joined = shape_join(&infer_term_shape(then, env_shapes), &infer_term_shape(els, env_shapes))?; + if shape_has_sum(&joined) { + return None; + } + 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)?; let sel = b.tuple(vec![c, t, e]); - b.add(Op::Select, vec![sel]) + Some(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 @@ -356,49 +450,152 @@ pub fn compile(term: &Term, b: &mut Builder, env: &[usize], env_shapes: & // Restriction: the step references only its binders (monoid-style), not outer // Vars — corgi closes the body; an outer reference would need CapList capture. 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 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); - b.add(Op::Fold(Box::new(body)), vec![pair]) + let body = compile_fold_body(step, &init_shape, &elem)?; + Some(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])) } - other => panic!("compile: unsupported Term: {other:?}"), + // 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). + 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 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 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 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)) + } else { + return None; + }; + result = Some(match result { None => out_shape, Some(prev) => shape_join(&prev, &out_shape)? }); + bodies.push((i, bb.finish(out))); + } + if bodies.is_empty() { + return None; // an all-⊥ scrutinee shape: nothing to map + } + let mapped = b.add(Op::MapSum(bodies), vec![cap]); + Some(b.add(Op::Unwrap, vec![mapped])) + } + Term::Unary(op, inner) => { + let id = compile(inner, b, env, env_shapes, anchor)?; + Some(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) { + Shape::Prim(_) => { + let zero = b.add(Op::Lit(CValue::u64(vec![0])), vec![anchor]); + let p = b.tuple(vec![id, zero]); + b.add(CmpOp::Rel(Pred::Eq), vec![p]) + } + _ => b.add(Op::Lit(CValue::u64(vec![1])), vec![anchor]), + }, + // 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) { + 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(_) => { + let zero = b.add(Op::Lit(CValue::u64(vec![0])), vec![anchor]); + let seed = b.tuple(vec![zero, id]); + let body = { + let mut bb = Builder::::default(); + let inp = bb.input(); + let acc = bb.add(Op::Field(0), vec![inp]); + let out = bb.add(ArithOp::AddU64(1), vec![acc]); + bb.finish(out) + }; + b.add(Op::Fold(Box::new(body)), vec![seed]) + } + _ => return None, + }, + // 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) { + 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)) + }) + }) + .collect(); + let mapped = b.add(Op::MapSum(arms), vec![id]); + b.add(Op::Unwrap, vec![mapped]) + } + _ => b.add(Op::Lit(CValue::u64(vec![0])), vec![anchor]), + }, + }) + } + _ => None, // List intro, Hash: see `compilable`'s accounting } } /// 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) -> Graph { +fn compile_fold_body(step: &Term, init_shape: &Shape, elem_shape: &Shape) -> Option> { 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]); - // Monoid fold bodies use only binders (no Spread/Proj-on-list), so no env shapes are needed. - let out = compile(step, &mut bb, &[acc, elem], &[], inp); - bb.finish(out) -} - -/// Compile a single `Term` whose `Var(0)` is the whole input row/column. (Spread-free terms only — -/// the bench/chain/fold examples — so no env shape is needed.) -pub fn compile_term_single(term: &Term) -> Graph { - let mut b = Builder::::default(); - let input = b.input(); - let out = compile(term, &mut b, &[input], &[], input); - b.finish(out) + let out = compile(step, &mut bb, &[acc, elem], &[init_shape.clone(), elem_shape.clone()], inp)?; + Some(bb.finish(out)) } /// Compile a `Filter` predicate over `Var(0)=key` (shape `kshape`), `Var(1)=val` (`vshape`) → mask. -pub fn compile_predicate(cond: &Term, kshape: &Shape, vshape: &Shape) -> Graph { +/// `None` when the term (with these shapes) has no lowering; the caller falls back to rows. +pub fn compile_predicate(cond: &Term, kshape: &Shape, vshape: &Shape) -> Option> { 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(cond, &mut b, &[var_k, var_v], &[kshape.clone(), vshape.clone()], input); - b.finish(out) + let out = compile(cond, &mut b, &[var_k, var_v], &[kshape.clone(), vshape.clone()], input)?; + Some(b.finish(out)) } /// 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 { let mut b = Builder::::default(); let input = b.input(); @@ -407,25 +604,26 @@ 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); - let nv = compile(val, &mut b, &env, &shapes, input); + 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 out = b.tuple(vec![nk, nv]); b.finish(out) } /// Compile a DDIR `Projection` over `Var(0)=key` (`kshape`), `Var(1)=val` (`vshape`). -/// Input `Prod([key, val])`; output `Prod([newkey, newval])`. -pub fn compile_projection(key: &Term, val: &Term, kshape: &Shape, vshape: &Shape) -> Graph { +/// 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> { 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)?; + let nv = compile(val, &mut b, &env, &shapes, input)?; let out = b.tuple(vec![nk, nv]); - b.finish(out) + Some(b.finish(out)) } #[cfg(test)] diff --git a/interactive/tests/corgi_backend.rs b/interactive/tests/corgi_backend.rs index d3faad679..31db6d137 100644 --- a/interactive/tests/corgi_backend.rs +++ b/interactive/tests/corgi_backend.rs @@ -23,13 +23,39 @@ fn inputs_for(prog: &str) -> Vec> { "unnest" => vec![rows(&[&[1, 2], &[3, 4]])], "adt" => vec![edges], "binders" => vec![rows(&[&[1, 2], &[3, 4]])], + // join_fallback: two keyed relations with overlapping keys (incl. a key with fanout). + "join_fallback" => vec![ + rows(&[&[1, 10], &[2, 20], &[2, 21], &[3, 30]]), + rows(&[&[1, 5], &[2, 6], &[4, 7]]), + ], + // scalar_ops: (key, a, b) triples; a values straddle the `> 2` and `= -5` tests. + "scalar_ops" => vec![rows(&[&[1, 1, 9], &[1, 4, 8], &[2, 3, 7], &[3, -5, 6], &[3, 2, 5]])], + "sum_ops" => vec![rows(&[&[1, 10], &[2, 20], &[2, 21]])], + "case_ops" => vec![rows(&[&[1, 10], &[2, 20], &[3, 14], &[3, 30]])], + // pair_keys: composite keys with overlap, fanout, and one-sided keys on both sides. + "pair_keys" => vec![ + rows(&[&[1, 1, 10], &[1, 2, 20], &[2, 1, 30], &[2, 1, 31], &[9, 9, 90]]), + rows(&[&[1, 1, 5], &[2, 1, 6], &[3, 3, 7]]), + ], + // tour: edges (with a cycle and a chord) + roots. + "tour" => vec![ + rows(&[&[1, 2], &[2, 3], &[3, 1], &[3, 4], &[5, 2]]), + rows(&[&[1], &[5]]), + ], other => panic!("no inputs configured for {other}"), } } /// Evaluate `prog` through both backends and assert the outputs match. fn assert_backends_agree(prog: &str) { - let path = format!("{}/examples/programs/{prog}.ddp", env!("CARGO_MANIFEST_DIR")); + // Fixtures pinning individual lowerings live with the gate (tests/programs); the + // algorithm programs double as examples and stay in examples/programs. + let fixture = format!("{}/tests/programs/{prog}.ddp", env!("CARGO_MANIFEST_DIR")); + let path = if std::path::Path::new(&fixture).exists() { + fixture + } else { + format!("{}/examples/programs/{prog}.ddp", env!("CARGO_MANIFEST_DIR")) + }; let src = interactive::load_program(&path); let mut tree = lower::lower_tree(parse::pipe::parse(&src)); tree.optimize(); @@ -47,3 +73,9 @@ fn assert_backends_agree(prog: &str) { #[test] fn unnest() { assert_backends_agree("unnest"); } #[test] fn adt() { assert_backends_agree("adt"); } #[test] fn binders() { assert_backends_agree("binders"); } +#[test] fn join_fallback() { assert_backends_agree("join_fallback"); } +#[test] fn scalar_ops() { assert_backends_agree("scalar_ops"); } +#[test] fn sum_ops() { assert_backends_agree("sum_ops"); } +#[test] fn case_ops() { assert_backends_agree("case_ops"); } +#[test] fn tour() { assert_backends_agree("tour"); } +#[test] fn pair_keys() { assert_backends_agree("pair_keys"); } diff --git a/interactive/tests/programs/case_ops.ddp b/interactive/tests/programs/case_ops.ddp new file mode 100644 index 000000000..e703d27f5 --- /dev/null +++ b/interactive/tests/programs/case_ops.ddp @@ -0,0 +1,19 @@ +-- `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. + +con Small(1) = 0; +con Big(1) = 1; + +let pairs = input 0 | key($0[0] ; $0[1]); +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) }); + +export "picked" = picked | arrange | inspect(total); +export "clash" = clash | arrange | inspect(adt); diff --git a/interactive/tests/programs/join_fallback.ddp b/interactive/tests/programs/join_fallback.ddp new file mode 100644 index 000000000..c2b4d8d94 --- /dev/null +++ b/interactive/tests/programs/join_fallback.ddp @@ -0,0 +1,9 @@ +-- A join whose projection the corgi lowering cannot compile (`hash`, reading both +-- sides): exercises the identity-join + row-wise-Project fallback. The gate is +-- agreement with the vec backend. + +let left = input 0 | key($0[0] ; $0[1]); +let right = input 1 | key($0[0] ; $0[1]); +let out = left | join(right, ($0 ; hash(0, $1[0], $2[0]), $2[0])); + +export "result" = out | arrange | inspect(total); diff --git a/interactive/tests/programs/pair_keys.ddp b/interactive/tests/programs/pair_keys.ddp new file mode 100644 index 000000000..24d680e0e --- /dev/null +++ b/interactive/tests/programs/pair_keys.ddp @@ -0,0 +1,9 @@ +-- A join keyed on a 2-tuple: exercises the lane-tuple key path (lexicographic +-- lane-buffer walk + tuple-needle probe, ordinal tokens) rather than the +-- single-lane or structural walks. The gate is agreement with vec. + +let left = input 0 | key($0[0], $0[1] ; $0[2]); +let right = input 1 | key($0[0], $0[1] ; $0[2]); +let out = left | join(right, ($0 ; $1[0] + $2[0])); + +export "result" = out | arrange | inspect(total); diff --git a/interactive/tests/programs/scalar_ops.ddp b/interactive/tests/programs/scalar_ops.ddp new file mode 100644 index 000000000..dc7e3faf1 --- /dev/null +++ b/interactive/tests/programs/scalar_ops.ddp @@ -0,0 +1,13 @@ +-- Scalar unaries on the corgi compiled path, checked against the vec backend: +-- negation, `not` (scalar and — constant-folded — non-scalar), tuple `len` +-- (static arity), and list `len` (folded along `collect`ed values). + +let pairs = input 0 | key($0[0] ; $0[1], $0[2]); +let mapped = pairs + | map($0 ; -$1[0], not($1[0] > 2), len($1)) + | filter(not($1[0] == 0 - 5)); + +let lengths = pairs | map($0 ; $1[0]) | collect | map($0 ; len($1)); + +export "mapped" = mapped | arrange | inspect(total); +export "lengths" = lengths | arrange | inspect(total); diff --git a/interactive/tests/programs/sum_ops.ddp b/interactive/tests/programs/sum_ops.ddp new file mode 100644 index 000000000..1858acd62 --- /dev/null +++ b/interactive/tests/programs/sum_ops.ddp @@ -0,0 +1,9 @@ +-- 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). + +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)); + +export "result" = tagged | arrange | inspect(total);