diff --git a/interactive/src/corgi/chunk.rs b/interactive/src/corgi/chunk.rs index 4b33b21a3..fa61db4c7 100644 --- a/interactive/src/corgi/chunk.rs +++ b/interactive/src/corgi/chunk.rs @@ -127,7 +127,8 @@ where /// Concatenate a run of (globally-sorted) chunks into one combined `(kv, times, diffs)`. fn concat(chunks: &[Self]) -> (CValue, ColTimes, Vec) { - let kvs: Vec = chunks.iter().map(Self::kv).collect(); + let mut kvs: Vec = chunks.iter().map(Self::kv).collect(); + unify_all(&mut kvs); let srcs: Vec> = kvs.iter().map(Some).collect(); let total: usize = chunks.iter().map(Self::len_).sum(); let (mut tags, mut offs) = (Vec::with_capacity(total), Vec::with_capacity(total)); @@ -169,7 +170,10 @@ where fn merge(in1: &mut VecDeque, in2: &mut VecDeque, out: &mut VecDeque) { let c1 = in1.pop_front().unwrap(); let c2 = in2.pop_front().unwrap(); - let (kv1, kv2) = (c1.kv(), c2.kv()); + let (mut kv1, mut kv2) = (c1.kv(), c2.kv()); + // Chunks are sealed by separate flushes, so their inferred arities can differ; the + // compare below is undefined on a tag with no lane. + unify_sum_arity(&mut kv1, &mut kv2); let (n1, n2) = (c1.len_(), c2.len_()); let (t1, d1) = (c1.times(), c1.diffs()); let (t2, d2) = (c2.times(), c2.diffs()); @@ -314,7 +318,8 @@ where out, |acc, next| { let (na, nb) = (acc.len_(), next.len_()); - let kvs = [acc.kv(), next.kv()]; + let mut kvs = [acc.kv(), next.kv()]; + unify_all(&mut kvs); let srcs = [Some(&kvs[0]), Some(&kvs[1])]; let mut tags = Vec::with_capacity(na + nb); let mut offs = Vec::with_capacity(na + nb); @@ -472,12 +477,54 @@ impl Default for CorgiChunker { } } +/// Pad two columns' `Sum` lane vectors to their common arity, `None` (`⊥`) filling. +/// +/// Shapes here are inferred per batch by `infer_shape_cols`, which commits only the arms it +/// actually SEES — so two batches of one DDIR type can disagree on arity (a batch of only +/// `Rare(_)` gives `Sum([Some])`, one of only `Common(_)` gives `Sum([None, Some])`). corgi +/// reads that as two unrelated types (`shape::join` calls a differing arity the genuine type +/// error), and `gather_lanes` then indexes a lane vector shorter than the tags it is given. +/// Widening with `⊥` is sound by corgi's own rule that an uncommitted lane holds no rows, and +/// leaves the within-variant offsets untouched (they depend only on tags and per-tag cursors). +fn unify_sum_arity(a: &mut CValue, b: &mut CValue) { + match (a, b) { + (CValue::Prod(xs), CValue::Prod(ys)) => { + for (x, y) in xs.iter_mut().zip(ys.iter_mut()) { unify_sum_arity(x, y); } + } + (CValue::List(_, x), CValue::List(_, y)) => unify_sum_arity(x, y), + (CValue::Sum(_, _, xs), CValue::Sum(_, _, ys)) => { + let k = xs.len().max(ys.len()); + xs.resize(k, None); + ys.resize(k, None); + for (x, y) in xs.iter_mut().zip(ys.iter_mut()) { + if let (Some(x), Some(y)) = (x, y) { unify_sum_arity(x, y); } + } + } + _ => {} + } +} + +/// Bring a set of columns to a common `Sum` arity. Pairwise against column 0 twice: `resize` +/// only grows, so the first pass leaves column 0 at the running max and the second lifts the +/// rest to it. +/// +/// Every site that reads two independently-inferred columns together must call this first — +/// `compare_at` as much as `gather_lanes`, since a tag with no lane has no defined order. +fn unify_all(cols: &mut [CValue]) { + if cols.len() < 2 { return; } + for _ in 0..2 { + let (head, tail) = cols.split_at_mut(1); + for c in tail.iter_mut() { unify_sum_arity(&mut head[0], c); } + } +} + /// Concatenate column blocks into one column (multi-source `gather_lanes`, no sort). -fn concat_blocks(blocks: &[CValue]) -> CValue { +fn concat_blocks(blocks: &mut [CValue]) -> CValue { if blocks.len() == 1 { return blocks[0].clone(); } - let srcs: Vec> = blocks.iter().map(Some).collect(); + unify_all(blocks); + let srcs: Vec> = blocks.iter().map(|b| Some(&*b)).collect(); let (mut tags, mut offs) = (Vec::new(), Vec::new()); for (ti, b) in blocks.iter().enumerate() { for o in 0..b.len() { tags.push(ti); offs.push(o); } @@ -495,8 +542,8 @@ where if self.times.is_empty() { return; } - let keys = concat_blocks(&self.k_blocks); - let vals = concat_blocks(&self.v_blocks); + let keys = concat_blocks(&mut self.k_blocks); + let vals = concat_blocks(&mut self.v_blocks); self.k_blocks.clear(); self.v_blocks.clear(); let times = std::mem::take(&mut self.times); diff --git a/interactive/tests/corgi_backend.rs b/interactive/tests/corgi_backend.rs index 31db6d137..ca50f5332 100644 --- a/interactive/tests/corgi_backend.rs +++ b/interactive/tests/corgi_backend.rs @@ -31,6 +31,9 @@ fn inputs_for(prog: &str) -> Vec> { // scalar_ops: (key, a, b) triples; a values straddle the `> 2` and `= -5` tests. "scalar_ops" => vec![rows(&[&[1, 1, 9], &[1, 4, 8], &[2, 3, 7], &[3, -5, 6], &[3, 2, 5]])], "sum_ops" => vec![rows(&[&[1, 10], &[2, 20], &[2, 21]])], + // 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![ @@ -76,6 +79,8 @@ fn assert_backends_agree(prog: &str) { #[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/programs/sum_skew.ddp b/interactive/tests/programs/sum_skew.ddp new file mode 100644 index 000000000..2fa9dbb58 --- /dev/null +++ b/interactive/tests/programs/sum_skew.ddp @@ -0,0 +1,18 @@ +-- 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. +-- +-- 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. +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])) ); + +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 new file mode 100644 index 000000000..2a3a0251c --- /dev/null +++ b/interactive/tests/programs/sum_skew_compiled.ddp @@ -0,0 +1,16 @@ +-- `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);