diff --git a/interactive/Cargo.toml b/interactive/Cargo.toml index b730cf48d..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 = { git = "https://github.com/frankmcsherry/wip", rev = "cb26fbd29c223a97781891298a0cc7b5f94a5b2f" } +corgi = { git = "https://github.com/frankmcsherry/wip", rev = "222157885ef424674163aeaf3f97775c9f160293" } differential-dataflow = { workspace = true } mimalloc = "0.1.48" serde = { version = "1.0", features = ["derive"] } 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/src/backend/corgi.rs b/interactive/src/backend/corgi.rs index 8f4065052..6eec6d6fd 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, compile_scalar}; use crate::ir::{Diff, LinearOp, Time, Value as DValue}; use crate::parse::{Projection, Reducer}; use crate::scope_ir as st; @@ -80,11 +80,13 @@ 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 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; @@ -95,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(); @@ -139,62 +140,125 @@ 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) } + // 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 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 } -/// 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 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) } 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/logic.rs b/interactive/src/corgi/logic.rs index 11bb882a1..9aa86e9a3 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 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; @@ -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 @@ -281,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`. @@ -318,12 +321,15 @@ 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. +/// +/// 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, @@ -336,10 +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), - // `Case` deliberately stays false HERE: this shape-free check gates join-INLINE - // projections only, and `Case` needs shapes (arm homogeneity). The join defers such - // projections to a linear stage, whose shape-aware `compile` lowers them there. - _ => false, // List intro, Case (here), data-driven Inject, Hash — see `compile`. + // `Op::Hash` is shape-generic (it folds whatever structure it is handed), so `hash` + // needs no shapes to lower and can answer true here. + Term::Hash(args) => args.iter().all(compilable), + // `List` and `Case` deliberately stay false HERE even though `compile` lowers both: + // each needs shapes to decide homogeneity (a list's elements, a case's arms), and this + // check runs without them. The join defers such projections to a linear stage, whose + // shape-aware `compile` lowers them there. + _ => false, // List intro, Case (here), data-driven Inject — see `compile`. } } @@ -419,13 +429,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)?; @@ -555,7 +562,63 @@ 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])) + } + // 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, } } @@ -570,17 +633,41 @@ fn compile_fold_body(step: &Term, init_shape: &Shape, elem_shape: &Shape) -> Opt Some(bb.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> { +/// 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(cond, &mut b, &[var_k, var_v], &[kshape.clone(), vshape.clone()], input)?; + let out = compile(term, &mut b, &[var_k, var_v], &[kshape.clone(), vshape.clone()], input)?; Some(b.finish(out)) } +/// 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> { + compile_over_kv(cond, kshape, vshape) +} + /// Compile a join projection: key/val Terms over `Var(0)=key`, `Var(1)=val0`, `Var(2)=val1` (with /// their shapes for `Spread`). Input `Prod([key, val0, val1])`; output `Prod([newkey, newval])`. /// Join-inline projections are gated by [`compilable`], so the lowering must succeed. @@ -619,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/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/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..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![ @@ -33,7 +35,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![ @@ -75,12 +76,12 @@ 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"); } #[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/explain.rs b/interactive/tests/explain.rs index 9fd47b54c..3c2218aab 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,201 @@ 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] +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] +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] +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); +} + 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);