From f8fa0b035f4fc56abbc17256ff98f504df861110 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Thu, 13 Aug 2026 14:52:24 -0400 Subject: [PATCH 01/11] B4 ast-compute: the row-wise-fallback measurement harness MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ports the B4 benchmark and its samply target from fable/work, where they were written but never landed: list build with arithmetic, flatmap explosion, variant tag + case + fold per element, then a tiny min reduce (16 keys, so the reduce is not the subject). A closed-form oracle checks vec, corgi checks against vec, and a hand-written native twin checks against the oracle. The program is chosen to sit entirely on `apply_ops`' row-wise fallback path, which is what it measures. On master-next today (arm64, mimalloc): n=1m native 1.08s vec 6.56s (6.08x nat) corgi 5.96s (5.52x nat, 0.91x vec) n=4m native 4.45s vec 32.45s (7.29x nat) corgi 25.70s (5.77x nat, 0.79x vec) Profile at n=1m: apply_ops is 55.9% of the run, of which ir::eval 27.1%, from_updates 18.0%, into_updates 5.5% — the untranscode/eval/re-transcode round trip is 50.6% of total. Value::clone (12.8%) plus drop_in_place (10.9%) is 23.7% spent on DValue trees that exist only for the fallback. corgi::eval_graph draws zero samples: every map declines. `Term::List` has no lowering, and the decline propagates up through Fold/Inject/If/Case, so the general Case lowering from #811 never engages on this program. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01YH7iq9JoXmf7ATpq1gZaQS --- interactive/examples/corgi_ast_compute.rs | 156 ++++++++++++++++++++++ interactive/examples/corgi_ast_prof.rs | 62 +++++++++ 2 files changed, 218 insertions(+) create mode 100644 interactive/examples/corgi_ast_compute.rs create mode 100644 interactive/examples/corgi_ast_prof.rs diff --git a/interactive/examples/corgi_ast_compute.rs b/interactive/examples/corgi_ast_compute.rs new file mode 100644 index 000000000..8e3f4775d --- /dev/null +++ b/interactive/examples/corgi_ast_compute.rs @@ -0,0 +1,156 @@ +//! B4 "ast-compute": the compute-heavy AST-style bookend — list build with arithmetic, +//! flatmap explosion, variant tag + case + fold per element, then a tiny min reduce. +//! This is corgi's home turf (wide per-row compute, no joins/recursion); it is a bookend, +//! not the headline. +//! +//! N=1000000,4000000 cargo run --release --example corgi_ast_compute +//! +//! Per row (a, b < 2^15 — keeps every intermediate non-negative and inside i64; corgi's +//! structural order is unsigned at the integer leaf, so signed values are out of contract +//! for `min`): build list [a, b, a+b, a*b, a-b+2^15, b*b, a*a, a+b*b], explode to +//! (pos, elem), bucket = pos + 8*(elem < H ? 0 : 1), payload = fold over +//! [a, elem, a*elem, elem*elem] wrapped in a Fwd/Bwd variant and matched back out, +//! then min(payload) per bucket (min so the compute cannot be dead-code eliminated, +//! while the reduce itself stays tiny — 16 keys). +//! +//! Correctness: closed-form Rust oracle checked against vec, corgi == vec, and the +//! hand-written native twin checked against the same oracle. `CHECK=1` forces. + +// The suite runs on mimalloc (as a real deployment would — `ddir_server` does): the +// system allocator was 27-28% of both DDIR backends' SCC profiles. One binary per +// benchmark, so every column (native/vec/corgi) shares the same allocator. +#[global_allocator] +static ALLOC: mimalloc::MiMalloc = mimalloc::MiMalloc; + +use std::time::{Duration, Instant}; + +use interactive::backend::{corgi, vec}; +use interactive::ir::Value; +use interactive::{lower, parse}; + +use differential_dataflow::input::Input; +use timely::dataflow::operators::probe::Handle; + +const H: i64 = 500_000; + +const AST_SRC: &str = r#" + con Fwd(1) = 0; + con Bwd(1) = 1; + + let rows = input 0 | key($0[0] ; $0[1]); + let lists = rows | map($0 ; list($0[0], $1[0], $0[0] + $1[0], $0[0] * $1[0], $0[0] - $1[0] + 32768, $1[0] * $1[0], $0[0] * $0[0], $0[0] + $1[0] * $1[0])); + let exploded = lists | flatmap($1[0]); + let tagged = exploded + | map( $1[0] + 8 * if($1[1] < 500000, 0, 1) + ; case if($1[1] < 250000, + Fwd(fold(list($0[0], $1[1], $0[0] * $1[1], $1[1] * $1[1]), 0, ^0 + ^1)), + Bwd(fold(list($0[0], $1[1], $0[0] * $1[1], $1[1] * $1[1]), 0, ^0 + ^1))) + { + Fwd(s) => s, + Bwd(s) => s, + } ); + let buckets = tagged | min; + export "result" = buckets | arrange; +"#; + +fn xorshift(s: &mut u64) -> u64 { *s ^= *s << 13; *s ^= *s >> 7; *s ^= *s << 17; *s } + +fn tup(fields: &[i64]) -> Value { Value::Tuple(fields.iter().map(|&n| Value::Int(n)).collect()) } + +/// The per-row logic, shared by the oracle and the native twin. +fn explode(a: i64, b: i64) -> impl Iterator { + let elems = [a, b, a + b, a * b, a - b + 32768, b * b, a * a, a + b * b]; + elems.into_iter().enumerate().map(move |(pos, e)| { + let bucket = pos as i64 + 8 * if e < H { 0 } else { 1 }; + let payload = a + e + a * e + e * e; + (bucket, payload) + }) +} + +fn native_ast_once(rows: &[(i64, i64)], capture: bool) -> Option> { + use timely::dataflow::operators::capture::{Capture, Event, Extract}; + let rows = rows.to_vec(); + let (tx, rx) = std::sync::mpsc::channel::>>(); + + timely::execute_directly(move |worker| { + let mut probe = Handle::new(); + let mut input = worker.dataflow::(|scope| { + let (input, data) = scope.new_collection::<(i64, i64), isize>(); + let buckets = data + .flat_map(|(a, b)| explode(a, b)) + .reduce(|_k, s, t| t.push((*s[0].0, 1isize))); + buckets.clone().probe_with(&mut probe); + if capture { buckets.inner.capture_into(tx); } + input + }); + for &(a, b) in &rows { input.insert((a, b)); } + input.advance_to(1); + input.flush(); + while probe.less_than(input.time()) { worker.step(); } + }); + + if capture { + let mut out: std::collections::BTreeMap<(i64, i64), isize> = Default::default(); + for (_, batch) in rx.extract() { + for (d, _, r) in batch { *out.entry(d).or_insert(0) += r; } + } + Some(out.into_iter().filter(|(_, r)| *r != 0).collect()) + } else { + None + } +} + +fn once(mut f: F) -> Duration { let t = Instant::now(); f(); t.elapsed() } + +fn consolidated(export: &[((Value, Value), i64)]) -> Vec<((Value, Value), i64)> { + let mut map: std::collections::BTreeMap<(Value, Value), i64> = Default::default(); + for ((k, v), d) in export { *map.entry((k.clone(), v.clone())).or_insert(0) += d; } + map.into_iter().filter(|(_, d)| *d != 0).collect() +} + +fn main() { + let mut p = lower::lower_tree(parse::pipe::parse(AST_SRC)); + p.optimize(); + let sizes: Vec = std::env::var("N").ok() + .map(|s| s.split(',').filter_map(|x| x.trim().parse().ok()).collect()) + .unwrap_or_else(|| vec![1_000_000, 4_000_000]); + + println!("ast-compute (B4) — list build + flatmap + variant/case/fold + min (8 elems/row):"); + for (i, &n_rows) in sizes.iter().enumerate() { + let mut seed = 0xfeed_f00d_u64; + let rows: Vec<(i64, i64)> = (0..n_rows) + .map(|_| ((xorshift(&mut seed) % 32_768) as i64, (xorshift(&mut seed) % 32_768) as i64)) + .collect(); + let ddir_rows: Vec<(Value, Value)> = rows.iter().map(|&(a, b)| (tup(&[a, b]), Value::unit())).collect(); + let inputs = vec![ddir_rows]; + + let check = std::env::var("CHECK").ok().map(|s| s != "0").unwrap_or(i == 0); + if check { + // Closed-form oracle: min payload per bucket. + let mut mins: std::collections::BTreeMap = Default::default(); + for &(a, b) in &rows { + for (bucket, payload) in explode(a, b) { + mins.entry(bucket).and_modify(|m| *m = (*m).min(payload)).or_insert(payload); + } + } + let expect: Vec<((Value, Value), i64)> = + mins.iter().map(|(&k, &v)| ((tup(&[k]), tup(&[v])), 1)).collect(); + let expect_nat: Vec<((i64, i64), isize)> = + mins.iter().map(|(&k, &v)| ((k, v), 1)).collect(); + + let vec_out = vec::evaluate(&p, &inputs); + assert_eq!(consolidated(&vec_out["result"]), expect, "vec != oracle at n={n_rows}"); + assert_eq!(corgi::evaluate(&p, &inputs), vec_out, "corgi != vec at n={n_rows}"); + assert_eq!(native_ast_once(&rows, true).unwrap(), expect_nat, "native != oracle at n={n_rows}"); + } + + let nt = once(|| { native_ast_once(&rows, false); }); + let vt = once(|| { std::hint::black_box(vec::evaluate(&p, &inputs)); }); + let ct = once(|| { std::hint::black_box(corgi::evaluate(&p, &inputs)); }); + let (nf, vf, cf) = (nt.as_secs_f64(), vt.as_secs_f64(), ct.as_secs_f64()); + println!( + " n={n_rows:<9} native {nt:>8.2?} vec-DDIR {vt:>8.2?} ({:.2}x nat) corgi-DDIR {ct:>8.2?} ({:.2}x nat, {:.2}x vec){}", + vf / nf, cf / nf, cf / vf, if check { " [checked]" } else { "" }, + ); + } +} diff --git a/interactive/examples/corgi_ast_prof.rs b/interactive/examples/corgi_ast_prof.rs new file mode 100644 index 000000000..a1b9fbef2 --- /dev/null +++ b/interactive/examples/corgi_ast_prof.rs @@ -0,0 +1,62 @@ +//! Profiling target for B4 ast-compute: loop one backend at a fixed size for samply. +//! The program is the one [`corgi_ast_compute`](../corgi_ast_compute.rs) measures; this +//! binary drops the oracle, the native twin, and the other backend so a profile carries +//! one backend's stacks and nothing else. +//! +//! N=1000000 ITERS=1 BACKEND=corgi samply record --save-only -o /tmp/p.json.gz -- \ +//! target/release/examples/corgi_ast_prof + +// Same allocator as the measurement binary — see `corgi_ast_compute`. +#[global_allocator] +static ALLOC: mimalloc::MiMalloc = mimalloc::MiMalloc; + +use interactive::backend::{corgi, vec}; +use interactive::ir::Value; +use interactive::{lower, parse}; + +const AST_SRC: &str = r#" + con Fwd(1) = 0; + con Bwd(1) = 1; + + let rows = input 0 | key($0[0] ; $0[1]); + let lists = rows | map($0 ; list($0[0], $1[0], $0[0] + $1[0], $0[0] * $1[0], $0[0] - $1[0] + 32768, $1[0] * $1[0], $0[0] * $0[0], $0[0] + $1[0] * $1[0])); + let exploded = lists | flatmap($1[0]); + let tagged = exploded + | map( $1[0] + 8 * if($1[1] < 500000, 0, 1) + ; case if($1[1] < 250000, + Fwd(fold(list($0[0], $1[1], $0[0] * $1[1], $1[1] * $1[1]), 0, ^0 + ^1)), + Bwd(fold(list($0[0], $1[1], $0[0] * $1[1], $1[1] * $1[1]), 0, ^0 + ^1))) + { + Fwd(s) => s, + Bwd(s) => s, + } ); + let buckets = tagged | min; + export "result" = buckets | arrange; +"#; + +fn xorshift(s: &mut u64) -> u64 { *s ^= *s << 13; *s ^= *s >> 7; *s ^= *s << 17; *s } + +fn main() { + let mut p = lower::lower_tree(parse::pipe::parse(AST_SRC)); + p.optimize(); + let n_rows: u64 = std::env::var("N").ok().and_then(|s| s.parse().ok()).unwrap_or(1_000_000); + let iters: usize = std::env::var("ITERS").ok().and_then(|s| s.parse().ok()).unwrap_or(1); + let backend = std::env::var("BACKEND").unwrap_or_else(|_| "corgi".into()); + let mut seed = 0xfeed_f00d_u64; + let rows: Vec<(Value, Value)> = (0..n_rows) + .map(|_| { + let a = (xorshift(&mut seed) % 32_768) as i64; + let b = (xorshift(&mut seed) % 32_768) as i64; + (Value::Tuple(vec![Value::Int(a), Value::Int(b)]), Value::unit()) + }) + .collect(); + let inputs = vec![rows]; + let mut acc = 0usize; + for _ in 0..iters { + acc += match backend.as_str() { + "vec" => std::hint::black_box(vec::evaluate(&p, &inputs)).len(), + _ => std::hint::black_box(corgi::evaluate(&p, &inputs)).len(), + }; + } + eprintln!("done ast backend={backend} n={n_rows} iters={iters} (acc={acc})"); +} From 21891d6bcd7c1fe07a7ef00a4ff278dce7e0f4ba Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Thu, 13 Aug 2026 15:06:32 -0400 Subject: [PATCH 02/11] Columnar list-literal intro: the keystone lowering A homogeneous `list(e0, .., ek)` becomes k columns and a length-k list per row through corgi's 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. Empty and heterogeneous literals decline (Weave needs a lane, Unwrap needs the lanes to join) and rows handle them. `infer_term_shape` gains the matching arm, which is what makes `fold` over a literal work: the Fold lowering reads the list's shape to find its element shape, so without it a `fold(list(..), ..)` declined even though both halves were already written. That is the keystone property in general -- a `list(..)` subterm anywhere made the WHOLE projection fall back to rows, which is why ast's case and fold lowerings (landed in #811) had never once engaged. B4 at n=1m: eval_graph goes from 0 samples to 59, and corgi 0.91x -> 0.85x vec. `compilable` still answers false for `List`, like `Case`: it is the shape-free gate for join-INLINE projections and cannot judge homogeneity. Its doc said these were unwritten lowerings; it now says what the gate is actually for. Gate 14/14, 29 lib tests, B4 checked against its oracle at every size. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01YH7iq9JoXmf7ATpq1gZaQS --- interactive/src/corgi/logic.rs | 78 +++++++++++++++++++++++++++------- 1 file changed, 63 insertions(+), 15 deletions(-) diff --git a/interactive/src/corgi/logic.rs b/interactive/src/corgi/logic.rs index 11bb882a1..71d70f865 100644 --- a/interactive/src/corgi/logic.rs +++ b/interactive/src/corgi/logic.rs @@ -2,11 +2,13 @@ //! 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 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 +//! 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 the one term with no lowering, and 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). use crate::ir::Value as DValue; @@ -243,6 +245,17 @@ fn infer_term_shape(t: &Term, env_shapes: &[Shape]) -> Shape { } 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 @@ -318,12 +331,14 @@ fn shape_join(a: &Shape, b: &Shape) -> Option { } } -/// Whether [`compile`] can lower this term to a corgi graph. Terms whose lowering is not yet -/// 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. +/// 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. +/// +/// `Hash` is the one term with no lowering anywhere: 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, @@ -336,9 +351,10 @@ pub fn compilable(t: &Term) -> bool { // 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. + // `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, Hash — see `compile`. } } @@ -555,7 +571,39 @@ pub fn compile(term: &Term, b: &mut Builder, env: &[usize], env_shapes: & }, }) } - _ => None, // List intro, Hash: see `compilable`'s accounting + // Homogeneous list literal: `k` element columns become a length-`k` list per row through + // 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. + 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)) + })?; + let mut lanes = Vec::with_capacity(fields.len()); + for f in fields { + let e = compile(f, b, env, env_shapes, anchor)?; + lanes.push(b.add(Op::Enlist, vec![e])); + } + let count = b.add(Op::Lit(CValue::u64(vec![fields.len() as u64])), vec![anchor]); + let mut weave_in = vec![b.add(Op::Iota, vec![count])]; + weave_in.extend(lanes); + let woven_in = b.tuple(weave_in); + let woven = b.add(Op::Weave, vec![woven_in]); + let unwrap_body = { + let mut bb = Builder::::default(); + let inp = bb.input(); + let out = bb.add(Op::Unwrap, vec![inp]); + bb.finish(out) + }; + Some(b.add(Op::MapList(Box::new(unwrap_body)), vec![woven])) + } + _ => None, // Hash: see `compilable`'s accounting } } From 65e0dc9d2a471415c40e3d0319db609f67d52d36 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Thu, 13 Aug 2026 15:07:11 -0400 Subject: [PATCH 03/11] Drop the sum-If gate: the pin it was written against moved in #817 The `If` lowering refused any sum-shaped result, on the grounds that merging sum columns committing different lanes tripped an offset bug in the pinned engine's lane merge. The comment said "revisit at the next corgi pin bump". The bump happened: #817 moved the pin from c4626fc to cb26fbd, whose sole commit IS the fix -- "gather_lanes: commit every lane any source does, not source 0's (#10)", with a regression test for the exact differing-arity case. `Op::Select` is implemented by `gather_lanes`, so the gate has been dead weight since Aug 11. It was not a cheap gate. `if(c, Fwd(x), Bwd(y))` is the shape of every conditional constructor, so a `case` over one fell back to rows -- which, together with the list-literal gap, is why B4's whole compute chain ran row-wise. B4 (n, corgi vs vec, checked against the oracle at each size): 100k 0.94x -> 0.54x 1m 0.91x -> 0.59x 4m 0.79x -> 0.54x Against the compiled-DD twin, 5.5-5.8x -> 3.6-3.9x. Profile at n=1m, share of total: apply_ops 55.9% -> 32.6%, ir::eval 27.1% -> 0.5%, from_updates 18.0% -> 11.8%, into_updates 5.5% -> 2.2%, eval_graph 0% -> 13.7%. The untranscode/eval/retranscode round trip falls from 50.6% to 14.5%, and what remains is flatmap, which is still row-wise. Gate 14/14 (adt, case_ops, sum_ops, sum_skew, sum_skew_compiled and tour all exercise sums), 29 lib tests. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01YH7iq9JoXmf7ATpq1gZaQS --- interactive/src/corgi/logic.rs | 21 ++++----------------- 1 file changed, 4 insertions(+), 17 deletions(-) diff --git a/interactive/src/corgi/logic.rs b/interactive/src/corgi/logic.rs index 71d70f865..8f0acdf07 100644 --- a/interactive/src/corgi/logic.rs +++ b/interactive/src/corgi/logic.rs @@ -294,16 +294,6 @@ fn infer_term_shape(t: &Term, env_shapes: &[Shape]) -> Shape { } } -/// 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`. @@ -435,13 +425,10 @@ 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 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; - } + // 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)?; From f59593d2b0809ae466bad23a70e3893a6de803b3 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Thu, 13 Aug 2026 16:05:39 -0400 Subject: [PATCH 04/11] Columnar flatmap: explode the list column structurally `compile_flatmap` lowers the list term over [key, val] to a corgi List column, declining when it will not lower or is not list-shaped (the explode needs real bounds, where `ir::eval` would accept any List value it happened to produce). The backend then explodes that column WITHOUT moving any element: the list's flat element storage already IS the new value column, and each row's span in the bounds yields both the within-row position DDIR pairs with it and a repeat map that carries key/time/diff across. No per-row eval, no transcode. The row-wise path moves to `apply_flatmap_rows` and stays the fallback. B4 (n, corgi vs vec, checked against the oracle at each size): 100k 0.54x -> 0.42x 1m 0.59x -> 0.48x 4m 0.54x -> 0.45x Against the compiled-DD twin, 3.6-3.9x -> 3.0-3.2x. That finishes the row-wise fallback on this program. Profile at n=1m, share of total, from where this branch started: apply_ops 55.9% -> 17.2% ir::eval 27.1% -> ~0% eval_graph 0% -> 16.0% from_updates 18.0% -> 1.2% into_updates 5.5% -> ~0% apply_ops is now essentially eval_graph alone, and the remaining `from_updates` is the `ToCorgi` import unary -- the I/O boundary, which is where transcoding belongs. `Value::clone` + `drop_in_place` fall from 23.7% to 2.0%. Gate 14/14, with unnest and tour verified to drive the columnar path rather than the fallback (instrumented run: zero fallbacks across the gate). Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01YH7iq9JoXmf7ATpq1gZaQS --- interactive/src/backend/corgi.rs | 74 ++++++++++++++++++++++++-------- interactive/src/corgi/logic.rs | 18 ++++++++ 2 files changed, 75 insertions(+), 17 deletions(-) diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs index 8f4065052..8bb584262 100644 --- a/interactive/src/backend/corgi.rs +++ b/interactive/src/backend/corgi.rs @@ -29,7 +29,7 @@ use crate::corgi::container::CorgiContainer; use crate::corgi::join::CorgiJoinBackend; use crate::corgi::reduce::CorgiReduceBackend; use differential_dataflow::operators::int_proxy::{ProxyJoinTactic, ProxyReduceTactic}; -use crate::corgi::logic::{compilable, compile_predicate, compile_projection}; +use crate::corgi::logic::{compilable, compile_flatmap, compile_predicate, compile_projection}; use crate::ir::{Diff, LinearOp, Time, Value as DValue}; use crate::parse::{Projection, Reducer}; use crate::scope_ir as st; @@ -80,11 +80,14 @@ fn rebase_join_term(t: &crate::parse::Term) -> crate::parse::Term { } } -/// 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 -/// (untranscode → vec-style transform → `from_updates`), matching `backend::vec` exactly; a columnar -/// fast-path is future work. `level` is the scope depth (locates the iteration coordinate). +/// 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. +/// The time-shaping ops (EnterAt/LiftIter) are still row-wise throughout (untranscode → vec-style +/// transform → `from_updates`), matching `backend::vec` exactly; their columnar forms touch only +/// times, so they are coverage rather than speed. `level` is the scope depth (locates the +/// iteration coordinate). fn apply_ops(mut c: CC, ops: &[LinearOp], level: usize) -> CC { use timely::order::Product; use differential_dataflow::lattice::Lattice; @@ -168,26 +171,63 @@ fn apply_ops(mut c: CC, ops: &[LinearOp], level: usize) -> CC { CorgiContainer::from_updates(out) } LinearOp::FlatMap(list_term) => { - 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:?}"), - } + 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(), }; - for (pos, elem) in elems.into_iter().enumerate() { - out.push(((k.clone(), DValue::Tuple(vec![DValue::Int(pos as i64), elem])), t.clone(), d)); + 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(), + } + } else { + apply_flatmap_rows(c, list_term) } - CorgiContainer::from_updates(out) } }; } 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) +} + /// Append the user-iter coordinate to a value (mirrors `backend::vec::append_iter`): extend a `Tuple`, /// or wrap any other value as `(value, iter)`. fn append_iter(val: DValue, iter: i64) -> DValue { diff --git a/interactive/src/corgi/logic.rs b/interactive/src/corgi/logic.rs index 8f0acdf07..c55d549fb 100644 --- a/interactive/src/corgi/logic.rs +++ b/interactive/src/corgi/logic.rs @@ -605,6 +605,24 @@ fn compile_fold_body(step: &Term, init_shape: &Shape, elem_shape: &Shape) -> Opt Some(bb.finish(out)) } +/// Compile a `FlatMap`'s list term over `Var(0)=key` (shape `kshape`), `Var(1)=val` (`vshape`) → +/// a corgi `List` column, one list per input row. `None` when the term has no lowering with these +/// shapes, or when it 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. +/// The caller falls back to rows in both cases. +pub fn compile_flatmap(list_term: &Term, kshape: &Shape, vshape: &Shape) -> Option> { + let shapes = [kshape.clone(), vshape.clone()]; + if !matches!(infer_term_shape(list_term, &shapes), Shape::List(_)) { + return None; + } + 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(list_term, &mut b, &[var_k, var_v], &shapes, input)?; + Some(b.finish(out)) +} + /// Compile a `Filter` predicate over `Var(0)=key` (shape `kshape`), `Var(1)=val` (`vshape`) → mask. /// `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> { From 4723d1b17b9583139012d052e161b2caa7d24d6c Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Thu, 13 Aug 2026 16:44:14 -0400 Subject: [PATCH 05/11] Columnar EnterAt: read the delay as a column, join it into times in place `enter_at`'s key and val columns are the identity -- only times change -- but the row-wise path untranscoded the WHOLE container to get at one integer per row. `compile_scalar` now lowers the delay field to a `U64` column (declining when it is not `Prim`-shaped, since the delay has no per-row reading then), and the times are adjusted in place. The delta never has to be built: joining `Product(0, PointStamp([0,..,delay]))` is coordinate-wise `max` at index `level-1` and identity elsewhere, because 0 is u64's minimum. `PointStamp::new` re-strips the trailing minimums the resize can add, so a zero delay leaves the representation canonical. `compile_flatmap`/`compile_scalar`/`compile_predicate` now share one `compile_over_kv` -- they differed only in the shape they demand of the result. As sized earlier this is coverage, not speed: enter_at's input collection changes once per epoch, so it is not per-iteration hot. Verified engaged rather than assumed -- an instrumented run shows the gate taking the columnar branch twice (scc.ddp's two enter_at sites) and the fallback zero times. scc agrees with vec beyond the gate's tiny inputs: 500/1500 -> 1319/-34 on both, and 5000/15000 -> 13281/-39 on both (corgi 348ms vs vec 410ms). Gate 14/14, 29 lib tests, 10 explain tests. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01YH7iq9JoXmf7ATpq1gZaQS --- interactive/src/backend/corgi.rs | 60 ++++++++++++++++++++++---------- interactive/src/corgi/logic.rs | 44 +++++++++++++---------- 2 files changed, 66 insertions(+), 38 deletions(-) diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs index 8bb584262..13b1a7b4b 100644 --- a/interactive/src/backend/corgi.rs +++ b/interactive/src/backend/corgi.rs @@ -29,7 +29,7 @@ use crate::corgi::container::CorgiContainer; 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}; +use crate::corgi::logic::{compilable, compile_flatmap, compile_predicate, compile_projection, compile_scalar}; use crate::ir::{Diff, LinearOp, Time, Value as DValue}; use crate::parse::{Projection, Reducer}; use crate::scope_ir as st; @@ -84,10 +84,9 @@ fn rebase_join_term(t: &crate::parse::Term) -> crate::parse::Term { /// 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. -/// The time-shaping ops (EnterAt/LiftIter) are still row-wise throughout (untranscode → vec-style -/// transform → `from_updates`), matching `backend::vec` exactly; their columnar forms touch only -/// times, so they are coverage rather than speed. `level` is the scope depth (locates the -/// iteration coordinate). +/// EnterAt reads its delay field columnar and joins it into `times` in place. `LiftIter` is the +/// one op still row-wise throughout (untranscode → vec-style transform → `from_updates`), matching +/// `backend::vec` exactly. `level` is the scope depth (locates the iteration coordinate). fn apply_ops(mut c: CC, ops: &[LinearOp], level: usize) -> CC { use timely::order::Product; use differential_dataflow::lattice::Lattice; @@ -142,23 +141,46 @@ fn apply_ops(mut c: CC, ops: &[LinearOp], level: usize) -> CC { } c } - // Row-wise ops (parity with `backend::vec::render_linear`). LinearOp::EnterAt(field) => { - 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 (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)); + } + CorgiContainer::from_updates(out) } - CorgiContainer::from_updates(out) } + // Row-wise ops (parity with `backend::vec::render_linear`). LinearOp::LiftIter => { let mut out: Vec = Vec::new(); for ((k, v), t, d) in c.into_updates() { diff --git a/interactive/src/corgi/logic.rs b/interactive/src/corgi/logic.rs index c55d549fb..e6cfe381f 100644 --- a/interactive/src/corgi/logic.rs +++ b/interactive/src/corgi/logic.rs @@ -605,33 +605,39 @@ fn compile_fold_body(step: &Term, init_shape: &Shape, elem_shape: &Shape) -> Opt Some(bb.finish(out)) } -/// Compile a `FlatMap`'s list term over `Var(0)=key` (shape `kshape`), `Var(1)=val` (`vshape`) → -/// a corgi `List` column, one list per input row. `None` when the term has no lowering with these -/// shapes, or when it 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. -/// The caller falls back to rows in both cases. -pub fn compile_flatmap(list_term: &Term, kshape: &Shape, vshape: &Shape) -> Option> { - let shapes = [kshape.clone(), vshape.clone()]; - if !matches!(infer_term_shape(list_term, &shapes), Shape::List(_)) { - return None; - } +/// 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> { 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(list_term, &mut b, &[var_k, var_v], &shapes, input)?; + let out = compile(term, &mut b, &[var_k, var_v], &[kshape.clone(), vshape.clone()], input)?; Some(b.finish(out)) } -/// Compile a `Filter` predicate over `Var(0)=key` (shape `kshape`), `Var(1)=val` (`vshape`) → mask. -/// `None` when the term (with these shapes) has no lowering; the caller falls back to rows. +/// 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 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 `Filter` predicate → a mask column (nonzero keeps the row). 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)?; - Some(b.finish(out)) + compile_over_kv(cond, kshape, vshape) } /// Compile a join projection: key/val Terms over `Var(0)=key`, `Var(1)=val0`, `Var(2)=val1` (with From 7434cf71c55929202fe67f56ce59b90bba85c48f Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Thu, 13 Aug 2026 16:53:06 -0400 Subject: [PATCH 06/11] Explain on corgi: the rewrite's own dataflows, cross-checked columnar `LinearOp::LiftIter` is synthesized by the explanation rewrite -- every `$host:` export is a LiftIter Linear, and it panics if it appears in a user program -- so an explained program is the only thing that exercises it. tests/explain.rs ran on `vec` alone, leaving the entire explain surface unexercised on corgi. These tests render the SAME rewritten dataflow on both backends and require every export to agree, which is the coverage a columnar LiftIter needs to be safe to write. Seven agree: reach, tc, flatmap, collect, a depth-1 `enter_at` loop, a depth-1 `min` loop, and an SCC-shaped program (depth-2 nesting, min, three joins, filtered feedback). Three do not, and the divergence PREDATES this branch -- reproduced at 78d75b05 in a scratch worktree, so it is neither the list work nor the columnar enter_at. On an explained program whose iterative feedback is NEGATED, corgi reports a strict SUBSET of vec's `demand:input0` (2 rows of 10 on the small scc instance). Isolated to the negation alone: `corgi_agrees_on_explained_scc_one_scope` and `explained_scc_one_scope_negated` differ in exactly one line (`var trim = trim_fwd | filter(..)` vs `var trim = trim_fwd - edges`) and only the negated one fails. Ruled out: `enter_at` (the depth-1 delay loop agrees, and forcing the row-wise enter_at path changes nothing), `min`, and depth-2 nesting on its own. The three are `#[ignore]`d with that reason rather than deleted, and the one-line pair is kept as the minimal reproducer. Note the plain (unexplained) programs do NOT diverge -- the corgi gate's scc passes, and scc at 5000/15000 matches vec exactly -- so the bug needs both the negated feedback and the rewrite. `query_rows` is factored out of `demand_for_queries` for reuse. Explain suite: 17 passed, 7 ignored (3 new known-divergence + 4 pre-existing sweeps). Gate 14/14, 29 lib tests. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01YH7iq9JoXmf7ATpq1gZaQS --- interactive/tests/explain.rs | 235 ++++++++++++++++++++++++++++++++--- 1 file changed, 221 insertions(+), 14 deletions(-) diff --git a/interactive/tests/explain.rs b/interactive/tests/explain.rs index 9fd47b54c..e75a47437 100644 --- a/interactive/tests/explain.rs +++ b/interactive/tests/explain.rs @@ -1,5 +1,8 @@ //! End-to-end semantic tests for the explanation rewrite, built on -//! `backend::vec::evaluate` (explicit inputs in, every export out). +//! `backend::vec::evaluate` (explicit inputs in, every export out). The +//! sufficiency properties are checked against `vec`, the correctness reference; +//! a final section cross-checks that the rewritten programs render identically +//! on the corgi backend. //! //! The central property is *sufficiency*: for a query against a program's //! output, the original inputs *restricted to* the demand-sets the rewritten @@ -10,6 +13,7 @@ use std::collections::BTreeSet; +use interactive::backend::corgi::evaluate as corgi_evaluate; use interactive::backend::vec::{evaluate, Row}; use interactive::ir::Value; use interactive::scope_ir::Program; @@ -118,6 +122,21 @@ fn optimized(src: &str) -> Program { p } +/// The query input's rows: each query is the flat demand envelope +/// `(K ; Tuple([V…, chain…, q]))` — V's fields, then the chain (empty, since the +/// first export is at depth 0), then the query id. +fn query_rows(queries: &[(Row, Row)]) -> Vec<(Row, Row)> { + queries + .iter() + .enumerate() + .map(|(q, (k, v))| { + let mut fields = match v { Value::Tuple(xs) => xs.clone(), other => vec![other.clone()] }; + fields.push(Value::Int(q as i64)); + (k.clone(), Value::Tuple(fields)) + }) + .collect() +} + /// The per-input demand-sets for a batch of query rows (q ids assigned in /// order) against `src`'s first export, with `src` run on `inputs`. fn demand_for_queries( @@ -129,20 +148,8 @@ fn demand_for_queries( let tree = lowered(src); let mut ex = explain::explain(&tree, shapes); ex.optimize(); - // A query row is the flat demand envelope `(K ; Tuple([V…, chain…, q]))`: - // V's fields, then the chain (empty — the first export is at depth 0), then - // the query id. - let query_rows: Vec<(Row, Row)> = queries - .iter() - .enumerate() - .map(|(q, (k, v))| { - let mut fields = match v { Value::Tuple(xs) => xs.clone(), other => vec![other.clone()] }; - fields.push(Value::Int(q as i64)); - (k.clone(), Value::Tuple(fields)) - }) - .collect(); let mut ex_inputs: Vec> = inputs.to_vec(); - ex_inputs.push(query_rows); + ex_inputs.push(query_rows(queries)); let exports = evaluate(&ex, &ex_inputs); (0..inputs.len()) .map(|i| { @@ -464,3 +471,203 @@ fn collect_explanations_sufficient_small() { let sizes = assert_all_rows_sufficient(COLLECT_ROW, COLLECT_SHAPES, &[gen_edges(20, 22)]); assert!(!sizes.is_empty(), "expected some collected lists to explain"); } + +// --------------------------------------------------------------------------- +// The corgi cross-check. +// +// `LinearOp::LiftIter` is SYNTHESIZED by the rewrite — every `$host:` export is +// a LiftIter Linear, and it panics if it appears in a user program — so an +// explained program is the only thing that exercises it. The rest of this file +// runs on `vec` alone, which left the whole explain surface unexercised on the +// columnar backend. These tests render the SAME rewritten dataflow on both and +// require every export to agree. +// +// Two of them are `#[ignore]`d against a divergence that PREDATES this coverage +// (reproduced at 78d75b05): on an explained program whose iterative feedback is +// NEGATED, corgi reports a strict subset of vec's `demand:input0`. The trigger +// is isolated below to the negation alone — `explained_scc_one_scope_negated` +// and `corgi_agrees_on_explained_scc_one_scope` differ in exactly that one line, +// and only the negated one fails. Ruled out along the way: `enter_at` (a depth-1 +// reach with a delay agrees, and forcing the row-wise enter_at path changes +// nothing), `min` (a depth-1 min/join loop agrees), and depth-2 nesting on its +// own (the negation-free two-scope program agrees). +// --------------------------------------------------------------------------- + +/// Run `src`'s explanation program on `inputs` + `queries` under both backends +/// and require identical exports. +fn assert_explained_backends_agree( + src: &str, + shapes: &[(usize, usize)], + inputs: &[Vec<(Row, Row)>], + queries: &[(Row, Row)], +) { + let tree = lowered(src); + let mut ex = explain::explain(&tree, shapes); + ex.optimize(); + let mut ex_inputs: Vec> = inputs.to_vec(); + ex_inputs.push(query_rows(queries)); + + let by_vec = evaluate(&ex, &ex_inputs); + let by_corgi = corgi_evaluate(&ex, &ex_inputs); + assert_eq!( + by_vec.keys().collect::>(), + by_corgi.keys().collect::>(), + "backends disagree on the export names" + ); + for (name, rows) in &by_vec { + assert_eq!(rows, &by_corgi[name], "export {name:?} differs between backends"); + } +} + +/// The first output row of `src` on `inputs` — a query that makes the +/// explanation dataflow do real work. +fn first_output_row(src: &str, inputs: &[Vec<(Row, Row)>]) -> (Row, Row) { + let p = optimized(src); + export_rows(&p, inputs, "result").into_iter().next().expect("a row to query") +} + +/// Two inputs (edges and roots), so the rewrite reports two demand exports. +#[test] +fn corgi_agrees_on_reach_explanation() { + let inputs = vec![gen_edges(50, 55), vec![(row(&[0]), Value::unit())]]; + let q = first_output_row(REACH_ROW, &inputs); + assert_explained_backends_agree(REACH_ROW, REACH_SHAPES, &inputs, &[q]); +} + +/// Transitive closure: iteration with `distinct` rather than `min`. +#[test] +fn corgi_agrees_on_tc_explanation() { + let inputs = vec![gen_edges(20, 22)]; + let q = first_output_row(TC_ROW, &inputs); + assert_explained_backends_agree(TC_ROW, TC_SHAPES, &inputs, &[q]); +} + +/// The list ops through the rewrite: `flatmap`'s reverse rule (intro + explode) +/// and `collect`'s (a List-valued reducer output). +#[test] +fn corgi_agrees_on_flatmap_explanation() { + let inputs = vec![gen_edges(20, 22)]; + let q = first_output_row(FLATMAP_ROW, &inputs); + assert_explained_backends_agree(FLATMAP_ROW, FLATMAP_SHAPES, &inputs, &[q]); +} + +#[test] +fn corgi_agrees_on_collect_explanation() { + let inputs = vec![gen_edges(20, 22)]; + let q = first_output_row(COLLECT_ROW, &inputs); + assert_explained_backends_agree(COLLECT_ROW, COLLECT_SHAPES, &inputs, &[q]); +} + +/// `enter_at` through the rewrite, at depth 1 so the negation trigger is absent. +const REACH_ENTER_AT: &str = r#" + let edges = input 0 | key($0[0] ; $0[1]); + let roots = input 1 | key($0[0] ;); + reach: { + let seeds = roots | enter_at($0[0]); + let proposals = reach | join(edges, ($2 ;)); + var reach = seeds + proposals | distinct; + } + export "result" = reach::reach; +"#; + +#[test] +fn corgi_agrees_on_enter_at_explanation() { + let inputs = vec![gen_edges(50, 55), vec![(row(&[0]), Value::unit())]]; + let q = first_output_row(REACH_ENTER_AT, &inputs); + assert_explained_backends_agree(REACH_ENTER_AT, REACH_SHAPES, &inputs, &[q]); +} + +/// A `min` reducer in an iterative loop, at depth 1. +const MIN_LOOP: &str = r#" + let edges = input 0 | key($0[0] ; $0[1]); + outer: { + let nodes = edges | key($1 ; $1); + let labels = proposals + nodes | min; + var proposals = labels | join(edges, ($2 ; $1)); + } + export "result" = outer::labels; +"#; + +#[test] +fn corgi_agrees_on_min_loop_explanation() { + let inputs = vec![gen_edges(50, 55)]; + let q = first_output_row(MIN_LOOP, &inputs); + assert_explained_backends_agree(MIN_LOOP, SCC_SHAPES, &inputs, &[q]); +} + +/// SCC's shape — depth-2 nesting, `min`, three joins, a filtered feedback — with +/// the feedback NOT negated. Agrees; the twin below differs only in that line. +const SCC_ONE_SCOPE: &str = r#" + let edges = input 0 | key($0[0] ; $0[1]); + outer: { + let scc = edges + trim; + fwd: { + let nodes = edges | key($1 ; $1); + let labels = proposals + nodes | min; + var proposals = labels | join(scc, ($2 ; $1)); + } + let trim_fwd = edges + | join(fwd::labels, ($1 ; $0, $2)) + | join(fwd::labels, ($0 ; $1, $2)) + | filter($1[1] == $1[2]) + | key($0 ; $1[0]); + var trim = trim_fwd | filter($0[0] > 1000000); + } + export "result" = outer::scc; +"#; + +/// The same program with `var trim = trim_fwd - edges` — the one-line minimal +/// reproducer for the demand divergence. +const SCC_ONE_SCOPE_NEGATED: &str = r#" + let edges = input 0 | key($0[0] ; $0[1]); + outer: { + let scc = edges + trim; + fwd: { + let nodes = edges | key($1 ; $1); + let labels = proposals + nodes | min; + var proposals = labels | join(scc, ($2 ; $1)); + } + let trim_fwd = edges + | join(fwd::labels, ($1 ; $0, $2)) + | join(fwd::labels, ($0 ; $1, $2)) + | filter($1[1] == $1[2]) + | key($0 ; $1[0]); + var trim = trim_fwd - edges; + } + export "result" = outer::scc; +"#; + +#[test] +fn corgi_agrees_on_explained_scc_one_scope() { + let inputs = vec![gen_edges(50, 55)]; + let q = first_output_row(SCC_ONE_SCOPE, &inputs); + assert_explained_backends_agree(SCC_ONE_SCOPE, SCC_SHAPES, &inputs, &[q]); +} + +#[test] +#[ignore = "known: corgi under-reports explain demand when the feedback is negated (pre-existing at 78d75b05)"] +fn explained_scc_one_scope_negated() { + let inputs = vec![gen_edges(50, 55)]; + let q = first_output_row(SCC_ONE_SCOPE_NEGATED, &inputs); + assert_explained_backends_agree(SCC_ONE_SCOPE_NEGATED, SCC_SHAPES, &inputs, &[q]); +} + +/// Iterative + `enter_at` + negated feedback: the real SCC. +#[test] +#[ignore = "known: corgi under-reports explain demand when the feedback is negated (pre-existing at 78d75b05)"] +fn corgi_agrees_on_scc_explanation() { + let inputs = vec![gen_edges(50, 55)]; + let q = first_output_row(SCC_ROW, &inputs); + assert_explained_backends_agree(SCC_ROW, SCC_SHAPES, &inputs, &[q]); +} + +/// Two queries at once, so the query input carries more than one envelope. +#[test] +#[ignore = "known: corgi under-reports explain demand when the feedback is negated (pre-existing at 78d75b05)"] +fn corgi_agrees_on_two_query_explanation() { + let inputs = vec![gen_edges(50, 55)]; + let p = optimized(SCC_ROW); + let qs: Vec<(Row, Row)> = export_rows(&p, &inputs, "result").into_iter().take(2).collect(); + assert_eq!(qs.len(), 2, "expected at least two scc edges to query"); + assert_explained_backends_agree(SCC_ROW, SCC_SHAPES, &inputs, &qs); +} From 19af697dc49185454d4b9a420ede0d773a716746 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Thu, 13 Aug 2026 16:57:09 -0400 Subject: [PATCH 07/11] Columnar LiftIter: one integer field, read out of the time The inverse of EnterAt. `LiftIter` reads each row's iteration coordinate out of its time and appends it to the value, so keys, times and diffs are untouched and no term is compiled -- which makes this path TOTAL. There is no gate and no fallback, unlike every other columnar op here. The empty product is where the two representations part company, and it is not a corner case: DDIR unit IS `Tuple([])`, which `append_iter` extends to `Tuple([iter])`, but columnar it arrives as `CValue::Unit`, not an empty `Prod`. Emitting `Prod([Unit, iter])` would be a silent one-field-too-many divergence from `backend::vec`, so `Unit` maps to `Prod([iter])`. The explain suite hits that branch 38 times (and the `Prod` branch 128) at the previous commit's coverage, so this is checked rather than argued. `append_iter` goes with it -- corgi has no remaining row-wise caller, and `backend::vec` keeps its own. Explain 17 passed / 7 ignored, gate 14/14, 29 lib tests. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01YH7iq9JoXmf7ATpq1gZaQS --- interactive/src/backend/corgi.rs | 47 +++++++++++++++++--------------- 1 file changed, 25 insertions(+), 22 deletions(-) diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs index 13b1a7b4b..ffb4e0290 100644 --- a/interactive/src/backend/corgi.rs +++ b/interactive/src/backend/corgi.rs @@ -84,9 +84,9 @@ fn rebase_join_term(t: &crate::parse::Term) -> crate::parse::Term { /// 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. -/// EnterAt reads its delay field columnar and joins it into `times` in place. `LiftIter` is the -/// one op still row-wise throughout (untranscode → vec-style transform → `from_updates`), matching -/// `backend::vec` exactly. `level` is the scope depth (locates the iteration coordinate). +/// 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; @@ -180,18 +180,30 @@ fn apply_ops(mut c: CC, ops: &[LinearOp], level: usize) -> CC { CorgiContainer::from_updates(out) } } - // Row-wise ops (parity with `backend::vec::render_linear`). + // 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. + // + // 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 + // to `Tuple([iter])`, but columnar it arrives as `CValue::Unit`, not an empty `Prod`. + // So `Unit` must become `Prod([iter])` — `Prod([Unit, iter])` would be a silent + // one-field-too-many divergence from `backend::vec`. LinearOp::LiftIter => { - let mut out: Vec = Vec::new(); - for ((k, v), t, d) in c.into_updates() { - let iter = level - .checked_sub(1) - .and_then(|idx| t.inner.get(idx).copied()) - .unwrap_or(0) as i64; - out.push(((k, append_iter(v, iter)), t, d)); - } - CorgiContainer::from_updates(out) + let iters: Vec = c + .times + .iter() + .map(|t| level.checked_sub(1).and_then(|idx| t.inner.get(idx).copied()).unwrap_or(0)) + .collect(); + let lane = CValue::u64(iters); + let vals = match c.vals { + CValue::Prod(mut fields) => { fields.push(lane); CValue::Prod(fields) } + CValue::Unit(_) => CValue::Prod(vec![lane]), + other => CValue::Prod(vec![other, lane]), + }; + 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) { @@ -250,15 +262,6 @@ fn apply_flatmap_rows(c: CC, list_term: &crate::parse::Term) -> CC { CorgiContainer::from_updates(out) } -/// Append the user-iter coordinate to a value (mirrors `backend::vec::append_iter`): extend a `Tuple`, -/// or wrap any other value as `(value, iter)`. -fn append_iter(val: DValue, iter: i64) -> DValue { - match val { - DValue::Tuple(mut xs) => { xs.push(DValue::Int(iter)); DValue::Tuple(xs) } - other => DValue::Tuple(vec![other, DValue::Int(iter)]), - } -} - /// 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 From 23728490855f42715ff7c6a3c8c06e51d72fc45f Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Thu, 13 Aug 2026 17:10:22 -0400 Subject: [PATCH 08/11] DDIR's hash IS corgi's hash; the gates that assumed otherwise go MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Nothing ever required the two backends' hash to be a PARTICULAR function -- only that they agree, since `hash` is an observable program value that lands in keys and gets exported. Rust makes no cross-run promise for `Hash` either. So DDIR's `hash` becomes corgi's structural hash, and `ir::eval` follows corgi rather than the reverse: `ir::structural_hash` is a row-at-a-time transcription of `corgi::hash`'s fold, salts and all, pinned by five tests that hash the same values through both paths (scalars, tuples, units, lists, variants, nesting). `ir::eval` no longer calls `hash_u64`, which goes back to being what it is -- the benchmark row generator's mixer, unrelated to the language. The lowering is `Op::Hash` over the arguments as one tuple, `Shr(1)` for the sign bit, `Rem` by the bound. The bound guard needs no `Select`: `Rem`'s total `x % 0 = x` gives `bound == 0` the identity, and a NEGATIVE bound reads as a u64 at or above 2^63 -- above the shifted hash -- so it is the identity too. Both match `ir::eval`'s `if bound > 0` exactly. `compilable` can admit `hash` shape-free, since `Op::Hash` folds whatever structure it is handed. REQUIRES corgi's `BinOp::Rem` (~/Projects/WIP/corgi, branch `ddir-rem`, commit 2221578, NOT pushed). The pin here is a local path so this branch is buildable and testable; it must become a rev before this merges. The two gates keyed to hash being unlowerable are gone, per Frank: a test that asserts a thing we intend to fix is the bug. * `join_fallback.ddp` used `hash` purely as an unlowerable term. It now uses a `case`, which drives the SAME path for a permanent reason: `compilable` is the shape-free gate, it runs before any container exists, and it will always decline shape-dependent terms. Verified still driving it (instrumented: one ROWFALL join, and no project fallback after, so the rebased projection compiles columnar). * `sum_skew.ddp` and `sum_skew_compiled.ddp` were the same program twice, one wrapped in `hash` to force the row-wise path. They now compile identically, so they are one program: the surviving `sum_skew.ddp` keeps the property (differing Sum arity reconciled through ⊥ lanes) and the accounting of both derivations. Gate 14 -> 13. Audit: instrumenting every fallback and running the whole suite leaves exactly two, both correct declines rather than gaps -- a `filter` whose `Case` arms have genuinely conflicting shapes (Prim vs Prod), and a `project` whose `Case` arm is a `Fold` whose step reads `Bound(2)`, past the closed body's env. No term now falls back for want of a lowering. B4 unmoved at 0.50x vec. 34 lib, 13 gate, 17 explain. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01YH7iq9JoXmf7ATpq1gZaQS --- interactive/Cargo.toml | 2 +- interactive/src/backend/corgi.rs | 5 +- interactive/src/corgi/logic.rs | 96 +++++++++++++++++-- interactive/src/ir.rs | 54 +++++++++-- interactive/tests/corgi_backend.rs | 2 - interactive/tests/programs/join_fallback.ddp | 16 +++- interactive/tests/programs/sum_skew.ddp | 25 +++-- .../tests/programs/sum_skew_compiled.ddp | 16 ---- 8 files changed, 166 insertions(+), 50 deletions(-) delete mode 100644 interactive/tests/programs/sum_skew_compiled.ddp diff --git a/interactive/Cargo.toml b/interactive/Cargo.toml index b730cf48d..227fb6253 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 = "cb26fbd29c223a97781891298a0cc7b5f94a5b2f" } +corgi = { path = "/Users/mcsherry/Projects/WIP/corgi" } # TEMPORARY local pin for validation differential-dataflow = { workspace = true } mimalloc = "0.1.48" serde = { version = "1.0", features = ["derive"] } diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs index ffb4e0290..6eec6d6fd 100644 --- a/interactive/src/backend/corgi.rs +++ b/interactive/src/backend/corgi.rs @@ -97,9 +97,8 @@ fn apply_ops(mut c: CC, ops: &[LinearOp], level: usize) -> CC { 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 (`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). + // 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(); diff --git a/interactive/src/corgi/logic.rs b/interactive/src/corgi/logic.rs index e6cfe381f..9aa86e9a3 100644 --- a/interactive/src/corgi/logic.rs +++ b/interactive/src/corgi/logic.rs @@ -6,9 +6,9 @@ //! 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 the one term with no lowering, and 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 +//! `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). use crate::ir::Value as DValue; @@ -327,8 +327,9 @@ fn shape_join(a: &Shape, b: &Shape) -> Option { /// `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. /// -/// `Hash` is the one term with no lowering anywhere: exact splitmix64 parity with `ir::eval` -/// needs lane-wise xor and integer rem, which corgi's arithmetic does not yet have. +/// 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. pub fn compilable(t: &Term) -> bool { match t { Term::Var(_) | Term::Bound(_) | Term::Int(_) => true, @@ -341,11 +342,14 @@ pub fn compilable(t: &Term) -> bool { // 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), + // `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, Hash — see `compile`. + _ => false, // List intro, Case (here), data-driven Inject — see `compile`. } } @@ -590,7 +594,31 @@ pub fn compile(term: &Term, b: &mut Builder, env: &[usize], env_shapes: & }; Some(b.add(Op::MapList(Box::new(unwrap_body)), vec![woven])) } - _ => None, // Hash: see `compilable`'s accounting + // 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. + // + // The bound guard is pure arithmetic — no `Select`. `Rem`'s total `x % 0 = x` gives + // `bound == 0` the identity, and a NEGATIVE bound reads as a `u64` at or above 2^63, + // 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 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)?); + } + 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])) + } + _ => None, } } @@ -678,6 +706,60 @@ mod tests { use super::*; use crate::ir::Value as V; + /// 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); + let col = transcode(&rows, &shape); + let columnar = corgi::hash(&col).into_u64("hash"); + 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)]); + } + + #[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()]); + // 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)])), + crate::ir::structural_hash(&V::Int(7)) + ); + } + + #[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)]), + ]); + } + + #[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))), + ]); + } + + #[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)])))]), + ]); + } + /// Round-trip a column of rows through infer_shape_cols → transcode → untranscode. fn roundtrip(rows: Vec) { let shape = infer_shape_cols(&rows); diff --git a/interactive/src/ir.rs b/interactive/src/ir.rs index c3e4f13e5..ce763357d 100644 --- a/interactive/src/ir.rs +++ b/interactive/src/ir.rs @@ -52,6 +52,49 @@ pub enum LinearOp { FlatMap(Term), } +// DDIR's `hash` IS corgi's structural hash, evaluated a row at a time here and a column at a +// time in the corgi backend. The two must agree bit for bit — they are the same program value, +// and the backends are checked against each other — so this is a transcription of +// `corgi::hash`'s fold, not an independent design. `roundtrip_hash_matches_corgi` pins it. +// +// The values are DDIR's; the shapes they transcode to are corgi's, and the fold follows those: +// `Int` is a `Prim` leaf, the empty `Tuple` is `Unit` (NOT a fieldless `Prod`), a `Tuple` is a +// `Prod`, a `List` folds its length then its elements, and a `Variant` folds its tag then its +// payload. The salts and multipliers are corgi's constants; changing one re-ids everything. +// +// No cross-run or cross-implementation stability is promised, exactly as Rust's own `Hash` makes +// no such promise: nothing may test a literal hash value or an order derived from one. +const HASH_PROD: u64 = 0x243f_6a88_85a3_08d3; +const HASH_SUM: u64 = 0x1319_8a2e_0370_7344; +const HASH_LIST: u64 = 0xa409_3822_299f_31d0; +const HASH_UNIT: u64 = 0x082e_fa98_ec4e_6c89; + +/// splitmix64's finalizer — the one bit-mixing primitive, as in `corgi::hash::mix64`. +fn mix64(mut z: u64) -> u64 { + z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9); + z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb); + z ^ (z >> 31) +} + +/// Fold one child hash into an accumulator — order-sensitive, so field order, element order and +/// tag position all matter. +fn hash_combine(acc: u64, x: u64) -> u64 { + (acc ^ mix64(x)).wrapping_mul(0x9e37_79b9_7f4a_7c15) +} + +/// The stable structural hash of one DDIR value. Row-wise twin of `corgi::hash`. +pub fn structural_hash(v: &Value) -> u64 { + match v { + Value::Int(x) => mix64(*x as u64), + Value::Tuple(fs) if fs.is_empty() => HASH_UNIT, + Value::Tuple(fs) => fs.iter().fold(HASH_PROD, |a, f| hash_combine(a, structural_hash(f))), + Value::List(xs) => xs + .iter() + .fold(hash_combine(HASH_LIST, xs.len() as u64), |a, x| hash_combine(a, structural_hash(x))), + Value::Variant(t, p) => hash_combine(hash_combine(HASH_SUM, *t as u64), structural_hash(p)), + } +} + /// Evaluate a scalar `Term` against an environment of `Value`s. /// /// `env` holds the operator's input rows at the bottom (`Var(i)` indexes it @@ -123,15 +166,10 @@ pub fn eval(term: &Term, env: &mut Vec) -> Value { if eval(cond, env).truthy() { eval(then, env) } else { eval(els, env) } } Term::Hash(args) => { - // args[0] is the (exclusive) bound; the rest are mixed into a - // deterministic non-negative draw via splitmix64. + // args[0] is the (exclusive) bound; the rest are hashed as one tuple. let bound = eval(&args[0], env).as_int(); - let mut acc: u64 = 0xcbf29ce484222325; - for a in &args[1..] { - acc ^= eval(a, env).as_int() as u64; - acc = crate::hash_u64(acc); - } - let h = (acc >> 1) as i64; // non-negative + let payload = Value::Tuple(args[1..].iter().map(|a| eval(a, env)).collect()); + let h = (structural_hash(&payload) >> 1) as i64; // non-negative Value::Int(if bound > 0 { h % bound } else { h }) } Term::Unary(op, t) => eval_unary(*op, eval(t, env)), diff --git a/interactive/tests/corgi_backend.rs b/interactive/tests/corgi_backend.rs index ca50f5332..507477b3f 100644 --- a/interactive/tests/corgi_backend.rs +++ b/interactive/tests/corgi_backend.rs @@ -33,7 +33,6 @@ fn inputs_for(prog: &str) -> Vec> { "sum_ops" => vec![rows(&[&[1, 10], &[2, 20], &[2, 21]])], // sum_skew: any keyed pairs — the skew is in the program, not the data. "sum_skew" => vec![rows(&[&[1, 10], &[2, 20], &[2, 21], &[3, 30]])], - "sum_skew_compiled" => vec![rows(&[&[1, 10], &[2, 20], &[2, 21], &[3, 30]])], "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![ @@ -80,7 +79,6 @@ fn assert_backends_agree(prog: &str) { #[test] fn scalar_ops() { assert_backends_agree("scalar_ops"); } #[test] fn sum_ops() { assert_backends_agree("sum_ops"); } #[test] fn sum_skew() { assert_backends_agree("sum_skew"); } -#[test] fn sum_skew_compiled() { assert_backends_agree("sum_skew_compiled"); } #[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/join_fallback.ddp b/interactive/tests/programs/join_fallback.ddp index c2b4d8d94..59206e5d2 100644 --- a/interactive/tests/programs/join_fallback.ddp +++ b/interactive/tests/programs/join_fallback.ddp @@ -1,9 +1,17 @@ --- 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. +-- A join whose projection the SHAPE-FREE gate cannot admit: `compilable` runs before any +-- container is in hand, so it declines every shape-dependent term — here a `case`, whose arm +-- homogeneity is a fact about the data. The join therefore joins with the identity projection +-- and applies the original terms as a rebased row-environment `Project`, which the shape-aware +-- lowering then compiles columnar. The gate is agreement with the vec backend. +-- +-- This is a PERMANENT path, not a gap: `compilable` will always be narrower than `compile`, +-- 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; 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])); +let out = left | join(right, ($0 ; case L($1[0] + $2[0]) { L(s) => s }, $2[0])); export "result" = out | arrange | inspect(total); diff --git a/interactive/tests/programs/sum_skew.ddp b/interactive/tests/programs/sum_skew.ddp index 2fa9dbb58..23bb961e8 100644 --- a/interactive/tests/programs/sum_skew.ddp +++ b/interactive/tests/programs/sum_skew.ddp @@ -1,18 +1,25 @@ -- 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. +-- 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). -- --- This is the ROW-WISE FALLBACK case: `hash` is a term corgi does not lower, so the shapes --- here come from `infer_shape_cols` scanning the data. See `sum_skew_compiled.ddp` for the --- same defect on the compiled path — both under-approximate, just from different sources. +-- 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; let pairs = input 0 | key($0[0] ; $0[1]); -let onlyRare = pairs | map( $0[0] ; Rare(hash(1000000, $1[0])) ); -let onlyCommon = pairs | map( $0[0] ; Common(hash(1000000, $1[0])) ); +let onlyRare = pairs | map( $0[0] ; Rare($1[0]) ); +let onlyCommon = pairs | map( $0[0] ; Common($1[0]) ); export "result" = (onlyRare + onlyCommon) | arrange | inspect(total); diff --git a/interactive/tests/programs/sum_skew_compiled.ddp b/interactive/tests/programs/sum_skew_compiled.ddp deleted file mode 100644 index 2a3a0251c..000000000 --- a/interactive/tests/programs/sum_skew_compiled.ddp +++ /dev/null @@ -1,16 +0,0 @@ --- `sum_skew.ddp` on the COMPILED path: no `hash`, so both maps lower to corgi logic. --- --- The compiled path derives shape from the TERM rather than the data, but that is no less --- of an under-approximation: `infer_term_shape` gives `Inject(tag, _)` an arity of `tag + 1`, --- so `Rare(_)` compiles to one lane and `Common(_)` to two. 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. -con Rare(1) = 0; -con Common(1) = 1; - -let pairs = input 0 | key($0[0] ; $0[1]); - -let onlyRare = pairs | map( $0[0] ; Rare($1[0]) ); -let onlyCommon = pairs | map( $0[0] ; Common($1[0]) ); - -export "result" = (onlyRare + onlyCommon) | arrange | inspect(total); From 83a47478fdf3144e685bf55b34ffca8ef07c0591 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Thu, 13 Aug 2026 17:59:49 -0400 Subject: [PATCH 09/11] Pin corgi to the pushed `Rem` rev `frankmcsherry/wip` @ 2221578 ("Integer Rem, with a total zero divisor"), branched off the previously pinned cb26fbd. Replaces the local-path pin the previous commit used to be testable before the push. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01YH7iq9JoXmf7ATpq1gZaQS --- interactive/Cargo.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/interactive/Cargo.toml b/interactive/Cargo.toml index 227fb6253..59ea7384d 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 = { path = "/Users/mcsherry/Projects/WIP/corgi" } # TEMPORARY local pin for validation +corgi = { git = "https://github.com/frankmcsherry/wip", rev = "222157885ef424674163aeaf3f97775c9f160293" } differential-dataflow = { workspace = true } mimalloc = "0.1.48" serde = { version = "1.0", features = ["derive"] } From 352350f6ef13a464b21df5c81876d507a1afa8f1 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Thu, 13 Aug 2026 19:01:22 -0400 Subject: [PATCH 10/11] corgi reduce: present values by NON-ZERO net, not positive net `Distinct` and `Min` selected the values they present with `d > 0`. DD's `reduce` presents every value whose accumulation is non-zero, negatives included, and `backend::vec` -- the correctness reference -- acts on exactly what DD hands it: its `Distinct` pushes `1` without looking at the diffs at all, and its `Min` takes `vals.iter().map(|(v, _)| v).min()` over all of them. So corgi silently dropped any key whose values all accumulated negative, and could pick a different minimum in a bracket that mixed signs. Only a negated collection produces a negative accumulation, and in the six canonical programs the reducers only ever see non-negative data -- which is why this survived until the explain rewrite, whose demand dataflow subtracts. `Count` already tested the SUM (`c > 0`, matching vec) and `Collect` already emitted per bracket with `d.max(0)` items; both were right and are untouched. Found by bisection on operator-level traces of the two backends: the wave front was `n165 = n164 | Distinct` in the explain scope, whose input agreed exactly on both backends at that time -- one value accumulating to -1 -- and whose output had the row on vec and not on corgi. Everything previously reported about this bug (the demand loop stalling at iteration 2, a join yielding multiplicity 1 instead of 3, the nested clone diverging while the forward clone did not) was downstream of that one row. The three `#[ignore]`d divergence tests from the explain-on-corgi commit now pass and are un-ignored, including the one-line minimal reproducer pair. Ruled out along the way and left as a pinned invariant: `ColTimes`'s columnar time order matches the owned `Ord` (property test over mixed-length PointStamps, `[3]` vs `[3, 1]` and friends) -- the chunk layer sorts and merges by the former while every other layer reasons with the latter. 35 lib, 13 gate, 20 explain, plus all 4 heavy sweeps (fuzz, scc-100, join-partner-time, demand-excess). B4 0.47x vec, scc 5000/15000 identical to vec (13281/-39) at 369ms vs 408ms. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01YH7iq9JoXmf7ATpq1gZaQS --- interactive/src/corgi/col_times.rs | 39 ++++++++++++++++++++++++++++++ interactive/src/corgi/reduce.rs | 20 ++++++++++----- interactive/tests/explain.rs | 4 +-- 3 files changed, 54 insertions(+), 9 deletions(-) diff --git a/interactive/src/corgi/col_times.rs b/interactive/src/corgi/col_times.rs index 83275c8ac..dc66c4feb 100644 --- a/interactive/src/corgi/col_times.rs +++ b/interactive/src/corgi/col_times.rs @@ -147,3 +147,42 @@ impl FromIterator for ColTimes { ColTimes { store } } } + +#[cfg(test)] +mod cmp_agreement_tests { + use super::*; + use differential_dataflow::dynamic::pointstamp::PointStamp; + use timely::order::Product; + + type T = Product>; + + fn t(outer: u64, coords: &[u64]) -> T { + Product::new(outer, PointStamp::new(coords.iter().copied().collect())) + } + + /// `ColTime::cmp_refs` (the derived `Ord` on the columnar `Ref`) must agree with the + /// timestamp's OWN `Ord` for every pair — the chunk layer sorts and merges by the former + /// and every other layer reasons with the latter. + #[test] + fn col_times_order_matches_owned_order() { + let times: Vec = vec![ + t(0, &[]), t(0, &[0]), t(0, &[1]), t(0, &[2]), t(0, &[3]), + t(0, &[1, 1]), t(0, &[1, 2]), t(0, &[2, 1]), t(0, &[3, 1]), t(0, &[3, 2]), + t(0, &[1, 1, 1]), t(0, &[3, 1, 2]), t(0, &[3, 2, 1]), + t(1, &[]), t(1, &[3]), t(1, &[3, 1]), + ]; + let mut store = ColTimes::::new(); + for x in × { store.push(x); } + for i in 0..times.len() { + for j in 0..times.len() { + let owned = times[i].cmp(×[j]); + let col = store.cmp(i, j); + assert_eq!( + owned, col, + "order disagrees for {:?} vs {:?}: owned {:?}, columnar {:?}", + times[i], times[j], owned, col + ); + } + } + } +} diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs index 1accf08b2..83effce22 100644 --- a/interactive/src/corgi/reduce.rs +++ b/interactive/src/corgi/reduce.rs @@ -346,11 +346,15 @@ where self.register_vals(col, &out_ids); } Reducer::Distinct => { - // Present iff any value has positive net; output value is unit (a `Unit` column). + // Present iff any value has NON-ZERO net -- the sign does not matter. DD's `reduce` + // presents every value whose accumulation is non-zero, negatives included, and + // `backend::vec`'s Distinct then emits `1` without looking at the diffs at all. A + // `> 0` test here silently drops a key whose values all accumulate negative, which + // is exactly what a negated collection produces. Output value is unit (a `Unit` column). let mut present = 0usize; let mut start = 0; for &end in ends { - if input[start..end].iter().any(|&(_, d)| d > 0) { + if input[start..end].iter().any(|&(_, d)| d != 0) { present += 1; out_diffs.push(1); } @@ -365,9 +369,13 @@ where self.register_vals(col, &out_ids); } Reducer::Min => { - // The DDIR `min` over the positive-diff values, in corgi's structural order (== DDIR - // `Ord` for the non-negative scalar/tuple values these reductions see; see module doc). - // Gather all positive-diff candidates across brackets into one column, segment by + // The DDIR `min` over the values with NON-ZERO net, in corgi's structural order + // (== DDIR `Ord` for the non-negative scalar/tuple values these reductions see; see + // module doc). The sign does not select candidates: `backend::vec` takes `min` over + // every value DD presents, and DD presents every non-zero accumulation. Filtering to + // `> 0` here both dropped all-negative keys and could pick a different minimum when a + // bracket mixed signs. + // Gather all candidates across brackets into one column, segment by // bracket, and one corgi `sort_blocks` gives every bracket's argmin at once // (`perm[block_start]`). The winning ROW is taken columnar and reuses its input value id. let mut cand_reps: Vec = Vec::new(); // input presentation rep index per candidate @@ -378,7 +386,7 @@ where let lo = cand_reps.len(); let seg = block_starts.len() as u64; for k in start..end { - if input[k].1 > 0 { + if input[k].1 != 0 { cand_reps.push(input[k].0); labels.push(seg); } diff --git a/interactive/tests/explain.rs b/interactive/tests/explain.rs index e75a47437..3c2218aab 100644 --- a/interactive/tests/explain.rs +++ b/interactive/tests/explain.rs @@ -645,7 +645,6 @@ fn corgi_agrees_on_explained_scc_one_scope() { } #[test] -#[ignore = "known: corgi under-reports explain demand when the feedback is negated (pre-existing at 78d75b05)"] fn explained_scc_one_scope_negated() { let inputs = vec![gen_edges(50, 55)]; let q = first_output_row(SCC_ONE_SCOPE_NEGATED, &inputs); @@ -654,7 +653,6 @@ fn explained_scc_one_scope_negated() { /// Iterative + `enter_at` + negated feedback: the real SCC. #[test] -#[ignore = "known: corgi under-reports explain demand when the feedback is negated (pre-existing at 78d75b05)"] fn corgi_agrees_on_scc_explanation() { let inputs = vec![gen_edges(50, 55)]; let q = first_output_row(SCC_ROW, &inputs); @@ -663,7 +661,6 @@ fn corgi_agrees_on_scc_explanation() { /// Two queries at once, so the query input carries more than one envelope. #[test] -#[ignore = "known: corgi under-reports explain demand when the feedback is negated (pre-existing at 78d75b05)"] fn corgi_agrees_on_two_query_explanation() { let inputs = vec![gen_edges(50, 55)]; let p = optimized(SCC_ROW); @@ -671,3 +668,4 @@ fn corgi_agrees_on_two_query_explanation() { assert_eq!(qs.len(), 2, "expected at least two scc edges to query"); assert_explained_backends_agree(SCC_ROW, SCC_SHAPES, &inputs, &qs); } + From 42ccaf84ba850886df3d83466d8bd0bf02b84e49 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Thu, 13 Aug 2026 20:55:23 -0400 Subject: [PATCH 11/11] The ast benchmark becomes a DDIR program, not two Rust binaries `interactive/examples` holds three Rust files, all generic drivers, plus DDIR programs. The B4 harness added two more binaries, each carrying its DDIR program as a string literal -- exactly what the convention exists to prevent. `corgi_ast_prof` was wholly redundant: `ddir` already loads a `.ddp`, takes `--backend=vec|corgi`, synthesizes rows and reports timings, so profiling is `samply record -- ddir --backend=corgi programs/ast.ddp ..` with no new binary. `corgi_ast_compute` duplicated that too; what was genuinely its own -- a closed-form oracle and a hand-written compiled-DD twin -- is a benchmarking concern, and compiled baselines already live in `diagnostics/examples`, not here. Both go; a benchmarking pass can reintroduce a baseline deliberately. The program itself is worth keeping and is now `examples/programs/ast.ddp`, where `ddir` can run it and the gate can cover it (14 programs). It is the only one exercising list intro, columnar flatmap, `case` over an `if`-selected constructor and `fold` together, and since a `list(..)` subterm anywhere makes the whole projection fall back to rows, a regression in one shows up as the others going row-wise too. Its input contract (non-negative, `$0[0] - $1[0] + 32768` non-negative) is written down in the file. The corgi-vs-vec claim stays reproducible from the tree without the harness: `ddir --backend=vec|corgi programs/ast.ddp 2 32768 200000 200000 1` gives vec 4.01s vs corgi 1.86s (0.46x), matching the 0.47-0.49x the deleted harness measured at n=1m. What is not reproducible without a baseline binary is the ratio to compiled DD. Branch diff drops from +856/-146 to +658/-146. 35 lib, 14 gate, 20 explain. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01YH7iq9JoXmf7ATpq1gZaQS --- interactive/examples/corgi_ast_compute.rs | 156 ---------------------- interactive/examples/corgi_ast_prof.rs | 62 --------- interactive/examples/programs/ast.ddp | 40 ++++++ interactive/tests/corgi_backend.rs | 5 +- 4 files changed, 44 insertions(+), 219 deletions(-) delete mode 100644 interactive/examples/corgi_ast_compute.rs delete mode 100644 interactive/examples/corgi_ast_prof.rs create mode 100644 interactive/examples/programs/ast.ddp diff --git a/interactive/examples/corgi_ast_compute.rs b/interactive/examples/corgi_ast_compute.rs deleted file mode 100644 index 8e3f4775d..000000000 --- a/interactive/examples/corgi_ast_compute.rs +++ /dev/null @@ -1,156 +0,0 @@ -//! B4 "ast-compute": the compute-heavy AST-style bookend — list build with arithmetic, -//! flatmap explosion, variant tag + case + fold per element, then a tiny min reduce. -//! This is corgi's home turf (wide per-row compute, no joins/recursion); it is a bookend, -//! not the headline. -//! -//! N=1000000,4000000 cargo run --release --example corgi_ast_compute -//! -//! Per row (a, b < 2^15 — keeps every intermediate non-negative and inside i64; corgi's -//! structural order is unsigned at the integer leaf, so signed values are out of contract -//! for `min`): build list [a, b, a+b, a*b, a-b+2^15, b*b, a*a, a+b*b], explode to -//! (pos, elem), bucket = pos + 8*(elem < H ? 0 : 1), payload = fold over -//! [a, elem, a*elem, elem*elem] wrapped in a Fwd/Bwd variant and matched back out, -//! then min(payload) per bucket (min so the compute cannot be dead-code eliminated, -//! while the reduce itself stays tiny — 16 keys). -//! -//! Correctness: closed-form Rust oracle checked against vec, corgi == vec, and the -//! hand-written native twin checked against the same oracle. `CHECK=1` forces. - -// The suite runs on mimalloc (as a real deployment would — `ddir_server` does): the -// system allocator was 27-28% of both DDIR backends' SCC profiles. One binary per -// benchmark, so every column (native/vec/corgi) shares the same allocator. -#[global_allocator] -static ALLOC: mimalloc::MiMalloc = mimalloc::MiMalloc; - -use std::time::{Duration, Instant}; - -use interactive::backend::{corgi, vec}; -use interactive::ir::Value; -use interactive::{lower, parse}; - -use differential_dataflow::input::Input; -use timely::dataflow::operators::probe::Handle; - -const H: i64 = 500_000; - -const AST_SRC: &str = r#" - con Fwd(1) = 0; - con Bwd(1) = 1; - - let rows = input 0 | key($0[0] ; $0[1]); - let lists = rows | map($0 ; list($0[0], $1[0], $0[0] + $1[0], $0[0] * $1[0], $0[0] - $1[0] + 32768, $1[0] * $1[0], $0[0] * $0[0], $0[0] + $1[0] * $1[0])); - let exploded = lists | flatmap($1[0]); - let tagged = exploded - | map( $1[0] + 8 * if($1[1] < 500000, 0, 1) - ; case if($1[1] < 250000, - Fwd(fold(list($0[0], $1[1], $0[0] * $1[1], $1[1] * $1[1]), 0, ^0 + ^1)), - Bwd(fold(list($0[0], $1[1], $0[0] * $1[1], $1[1] * $1[1]), 0, ^0 + ^1))) - { - Fwd(s) => s, - Bwd(s) => s, - } ); - let buckets = tagged | min; - export "result" = buckets | arrange; -"#; - -fn xorshift(s: &mut u64) -> u64 { *s ^= *s << 13; *s ^= *s >> 7; *s ^= *s << 17; *s } - -fn tup(fields: &[i64]) -> Value { Value::Tuple(fields.iter().map(|&n| Value::Int(n)).collect()) } - -/// The per-row logic, shared by the oracle and the native twin. -fn explode(a: i64, b: i64) -> impl Iterator { - let elems = [a, b, a + b, a * b, a - b + 32768, b * b, a * a, a + b * b]; - elems.into_iter().enumerate().map(move |(pos, e)| { - let bucket = pos as i64 + 8 * if e < H { 0 } else { 1 }; - let payload = a + e + a * e + e * e; - (bucket, payload) - }) -} - -fn native_ast_once(rows: &[(i64, i64)], capture: bool) -> Option> { - use timely::dataflow::operators::capture::{Capture, Event, Extract}; - let rows = rows.to_vec(); - let (tx, rx) = std::sync::mpsc::channel::>>(); - - timely::execute_directly(move |worker| { - let mut probe = Handle::new(); - let mut input = worker.dataflow::(|scope| { - let (input, data) = scope.new_collection::<(i64, i64), isize>(); - let buckets = data - .flat_map(|(a, b)| explode(a, b)) - .reduce(|_k, s, t| t.push((*s[0].0, 1isize))); - buckets.clone().probe_with(&mut probe); - if capture { buckets.inner.capture_into(tx); } - input - }); - for &(a, b) in &rows { input.insert((a, b)); } - input.advance_to(1); - input.flush(); - while probe.less_than(input.time()) { worker.step(); } - }); - - if capture { - let mut out: std::collections::BTreeMap<(i64, i64), isize> = Default::default(); - for (_, batch) in rx.extract() { - for (d, _, r) in batch { *out.entry(d).or_insert(0) += r; } - } - Some(out.into_iter().filter(|(_, r)| *r != 0).collect()) - } else { - None - } -} - -fn once(mut f: F) -> Duration { let t = Instant::now(); f(); t.elapsed() } - -fn consolidated(export: &[((Value, Value), i64)]) -> Vec<((Value, Value), i64)> { - let mut map: std::collections::BTreeMap<(Value, Value), i64> = Default::default(); - for ((k, v), d) in export { *map.entry((k.clone(), v.clone())).or_insert(0) += d; } - map.into_iter().filter(|(_, d)| *d != 0).collect() -} - -fn main() { - let mut p = lower::lower_tree(parse::pipe::parse(AST_SRC)); - p.optimize(); - let sizes: Vec = std::env::var("N").ok() - .map(|s| s.split(',').filter_map(|x| x.trim().parse().ok()).collect()) - .unwrap_or_else(|| vec![1_000_000, 4_000_000]); - - println!("ast-compute (B4) — list build + flatmap + variant/case/fold + min (8 elems/row):"); - for (i, &n_rows) in sizes.iter().enumerate() { - let mut seed = 0xfeed_f00d_u64; - let rows: Vec<(i64, i64)> = (0..n_rows) - .map(|_| ((xorshift(&mut seed) % 32_768) as i64, (xorshift(&mut seed) % 32_768) as i64)) - .collect(); - let ddir_rows: Vec<(Value, Value)> = rows.iter().map(|&(a, b)| (tup(&[a, b]), Value::unit())).collect(); - let inputs = vec![ddir_rows]; - - let check = std::env::var("CHECK").ok().map(|s| s != "0").unwrap_or(i == 0); - if check { - // Closed-form oracle: min payload per bucket. - let mut mins: std::collections::BTreeMap = Default::default(); - for &(a, b) in &rows { - for (bucket, payload) in explode(a, b) { - mins.entry(bucket).and_modify(|m| *m = (*m).min(payload)).or_insert(payload); - } - } - let expect: Vec<((Value, Value), i64)> = - mins.iter().map(|(&k, &v)| ((tup(&[k]), tup(&[v])), 1)).collect(); - let expect_nat: Vec<((i64, i64), isize)> = - mins.iter().map(|(&k, &v)| ((k, v), 1)).collect(); - - let vec_out = vec::evaluate(&p, &inputs); - assert_eq!(consolidated(&vec_out["result"]), expect, "vec != oracle at n={n_rows}"); - assert_eq!(corgi::evaluate(&p, &inputs), vec_out, "corgi != vec at n={n_rows}"); - assert_eq!(native_ast_once(&rows, true).unwrap(), expect_nat, "native != oracle at n={n_rows}"); - } - - let nt = once(|| { native_ast_once(&rows, false); }); - let vt = once(|| { std::hint::black_box(vec::evaluate(&p, &inputs)); }); - let ct = once(|| { std::hint::black_box(corgi::evaluate(&p, &inputs)); }); - let (nf, vf, cf) = (nt.as_secs_f64(), vt.as_secs_f64(), ct.as_secs_f64()); - println!( - " n={n_rows:<9} native {nt:>8.2?} vec-DDIR {vt:>8.2?} ({:.2}x nat) corgi-DDIR {ct:>8.2?} ({:.2}x nat, {:.2}x vec){}", - vf / nf, cf / nf, cf / vf, if check { " [checked]" } else { "" }, - ); - } -} diff --git a/interactive/examples/corgi_ast_prof.rs b/interactive/examples/corgi_ast_prof.rs deleted file mode 100644 index a1b9fbef2..000000000 --- a/interactive/examples/corgi_ast_prof.rs +++ /dev/null @@ -1,62 +0,0 @@ -//! Profiling target for B4 ast-compute: loop one backend at a fixed size for samply. -//! The program is the one [`corgi_ast_compute`](../corgi_ast_compute.rs) measures; this -//! binary drops the oracle, the native twin, and the other backend so a profile carries -//! one backend's stacks and nothing else. -//! -//! N=1000000 ITERS=1 BACKEND=corgi samply record --save-only -o /tmp/p.json.gz -- \ -//! target/release/examples/corgi_ast_prof - -// Same allocator as the measurement binary — see `corgi_ast_compute`. -#[global_allocator] -static ALLOC: mimalloc::MiMalloc = mimalloc::MiMalloc; - -use interactive::backend::{corgi, vec}; -use interactive::ir::Value; -use interactive::{lower, parse}; - -const AST_SRC: &str = r#" - con Fwd(1) = 0; - con Bwd(1) = 1; - - let rows = input 0 | key($0[0] ; $0[1]); - let lists = rows | map($0 ; list($0[0], $1[0], $0[0] + $1[0], $0[0] * $1[0], $0[0] - $1[0] + 32768, $1[0] * $1[0], $0[0] * $0[0], $0[0] + $1[0] * $1[0])); - let exploded = lists | flatmap($1[0]); - let tagged = exploded - | map( $1[0] + 8 * if($1[1] < 500000, 0, 1) - ; case if($1[1] < 250000, - Fwd(fold(list($0[0], $1[1], $0[0] * $1[1], $1[1] * $1[1]), 0, ^0 + ^1)), - Bwd(fold(list($0[0], $1[1], $0[0] * $1[1], $1[1] * $1[1]), 0, ^0 + ^1))) - { - Fwd(s) => s, - Bwd(s) => s, - } ); - let buckets = tagged | min; - export "result" = buckets | arrange; -"#; - -fn xorshift(s: &mut u64) -> u64 { *s ^= *s << 13; *s ^= *s >> 7; *s ^= *s << 17; *s } - -fn main() { - let mut p = lower::lower_tree(parse::pipe::parse(AST_SRC)); - p.optimize(); - let n_rows: u64 = std::env::var("N").ok().and_then(|s| s.parse().ok()).unwrap_or(1_000_000); - let iters: usize = std::env::var("ITERS").ok().and_then(|s| s.parse().ok()).unwrap_or(1); - let backend = std::env::var("BACKEND").unwrap_or_else(|_| "corgi".into()); - let mut seed = 0xfeed_f00d_u64; - let rows: Vec<(Value, Value)> = (0..n_rows) - .map(|_| { - let a = (xorshift(&mut seed) % 32_768) as i64; - let b = (xorshift(&mut seed) % 32_768) as i64; - (Value::Tuple(vec![Value::Int(a), Value::Int(b)]), Value::unit()) - }) - .collect(); - let inputs = vec![rows]; - let mut acc = 0usize; - for _ in 0..iters { - acc += match backend.as_str() { - "vec" => std::hint::black_box(vec::evaluate(&p, &inputs)).len(), - _ => std::hint::black_box(corgi::evaluate(&p, &inputs)).len(), - }; - } - eprintln!("done ast backend={backend} n={n_rows} iters={iters} (acc={acc})"); -} diff --git a/interactive/examples/programs/ast.ddp b/interactive/examples/programs/ast.ddp new file mode 100644 index 000000000..da7d98f96 --- /dev/null +++ b/interactive/examples/programs/ast.ddp @@ -0,0 +1,40 @@ +-- AST-style compute: the wide per-row bookend. Build a list with arithmetic, explode it, +-- tag each element as a variant, fold the payload, match it back out, and reduce. No joins +-- and no recursion -- this is a compute program, where the columnar backend's scalar logic +-- is the whole cost and the differential machinery is not. +-- +-- Every lowering it needs landed together: `list` intro, columnar `flatmap`, `case` over an +-- `if`-selected constructor, and `fold`. A `list(..)` subterm anywhere makes the WHOLE +-- projection fall back to rows, so a regression in any one of them shows up here as the +-- others going row-wise too. +-- +-- Contract on the inputs: fields must be non-negative and small enough that +-- `$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; + +let rows = input 0 | key($0[0] ; $0[1]); + +-- eight derived values per row +let lists = rows | map($0 ; list($0[0], $1[0], $0[0] + $1[0], $0[0] * $1[0], $0[0] - $1[0] + 32768, $1[0] * $1[0], $0[0] * $0[0], $0[0] + $1[0] * $1[0])); + +-- one row per element, carrying its position +let exploded = lists | flatmap($1[0]); + +-- bucket by (position, magnitude); payload folds a four-element list, wrapped in a variant +-- and matched straight back out (the tag is the point, not the payload). +let tagged = exploded + | map( $1[0] + 8 * if($1[1] < 500000, 0, 1) + ; case if($1[1] < 250000, + Fwd(fold(list($0[0], $1[1], $0[0] * $1[1], $1[1] * $1[1]), 0, ^0 + ^1)), + Bwd(fold(list($0[0], $1[1], $0[0] * $1[1], $1[1] * $1[1]), 0, ^0 + ^1))) + { + Fwd(s) => s, + Bwd(s) => s, + } ); + +let buckets = tagged | min; + +export "result" = buckets | arrange; diff --git a/interactive/tests/corgi_backend.rs b/interactive/tests/corgi_backend.rs index 507477b3f..dfc517de2 100644 --- a/interactive/tests/corgi_backend.rs +++ b/interactive/tests/corgi_backend.rs @@ -21,7 +21,9 @@ fn inputs_for(prog: &str) -> Vec> { // stable: edges (l_node, l_pref, r_node, r_pref) "stable" => vec![rows(&[&[1, 1, 10, 1], &[1, 2, 11, 1], &[2, 1, 10, 2], &[2, 2, 11, 2]])], "unnest" => vec![rows(&[&[1, 2], &[3, 4]])], - "adt" => vec![edges], + "adt" => vec![edges.clone()], + // ast: pairs; small and non-negative, per the program's stated input contract. + "ast" => 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![ @@ -74,6 +76,7 @@ fn assert_backends_agree(prog: &str) { #[test] fn stable() { assert_backends_agree("stable"); } #[test] fn unnest() { assert_backends_agree("unnest"); } #[test] fn adt() { assert_backends_agree("adt"); } +#[test] fn ast() { assert_backends_agree("ast"); } #[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"); }