Widen inferred Sum arities before the corgi chunker compares or gathers

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<T>, Vec<R>) {
-        let kvs: Vec<CValue> = chunks.iter().map(Self::kv).collect();
+        let mut kvs: Vec<CValue> = chunks.iter().map(Self::kv).collect();
+        unify_all(&mut kvs);
         let srcs: Vec<Option<&CValue>> = 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<Self>, in2: &mut VecDeque<Self>, out: &mut VecDeque<Self>) {
         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<T: Columnar, R> Default for CorgiChunker<T, R> {
     }
 }
 
+/// 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<Option<&CValue>> = blocks.iter().map(Some).collect();
+    unify_all(blocks);
+    let srcs: Vec<Option<&CValue>> = 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);

interactive/tests/corgi_backend.rs

@@ -31,6 +31,9 @@ fn inputs_for(prog: &str) -> Vec<Vec<(Value, Value)>> {
         // 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"); }

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);

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);