From 7c0d05b947fd582be3bf3c2d3b3299fb5b93cf7c Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Mon, 3 Aug 2026 20:08:04 -0400 Subject: [PATCH 1/2] Widen inferred Sum arities before the corgi chunker compares or gathers `infer_shape_cols` commits only the arms a batch actually contains, so two containers of one DDIR type can disagree on Sum arity: a collection of only `Rare(_)` infers `Sum([Some(_)])`, one of only `Common(_)` infers `Sum([None, Some(_)])`. corgi reads a differing arity as a genuine type error (`shape::join` says so explicitly), and `gather_lanes` then indexes a lane vector shorter than the tags handed to it: index out of bounds: the len is 1 but the index is 1 corgi/src/value.rs:390 within_offsets -> Value::sum_from_prim -> gather_lanes -> chunk::concat_blocks Reachable from a plain program -- concatenate two collections that inject different constructors, and arrange the result. vec renders it fine, so this was a corgi-only parity hole. `tests/programs/sum_skew.ddp` is that program; it panics without this change. Widen to the common arity with uncommitted (bottom) lanes at each site that reads two independently-inferred columns together. Sound by corgi's own rule that an uncommitted lane holds no rows, and it leaves within-variant offsets untouched (they depend only on tags and per-tag cursors). `merge` needs it as much as the gathers do -- `compare_at` has no defined order for a tag with no lane. Gate 13/13 debug and release; scc steady-state 17.49s vs 17.46s baseline (median of 3, baseline spanning 17.44-17.51) -- the walk is over the shape spine, not the rows. Note for review: this reconciles shapes in DDIR, which is arguably data-layer work. Two alternatives, both bigger: teach corgi's `gather_lanes` to join source arities (fixes every caller, needs a pin bump), or carry the `con` declarations into the plan so every batch infers the declared arity and no reconciliation is needed anywhere. The last is the principled fix; this one is the small one that closes the panic. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_0168NJWzHwrLGW2RXToGooYP --- interactive/src/corgi/chunk.rs | 61 ++++++++++++++++++++++--- interactive/tests/corgi_backend.rs | 3 ++ interactive/tests/programs/sum_skew.ddp | 17 +++++++ 3 files changed, 74 insertions(+), 7 deletions(-) create mode 100644 interactive/tests/programs/sum_skew.ddp 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..507477b3f 100644 --- a/interactive/tests/corgi_backend.rs +++ b/interactive/tests/corgi_backend.rs @@ -31,6 +31,8 @@ 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]])], "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 +78,7 @@ 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 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..794b01d2f --- /dev/null +++ b/interactive/tests/programs/sum_skew.ddp @@ -0,0 +1,17 @@ +-- Two collections whose variant columns commit DIFFERENT arms, concatenated into one +-- arrangement. Shapes are inferred per batch from the data, so one side infers +-- `Sum([Some(_)])` (tag 0 only) and 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 chunker must widen with uncommitted (⊥) lanes before it compares or gathers them. +-- +-- `hash` keeps both maps on the ROW-WISE fallback, the only path whose shape comes from +-- the data rather than from the term (the compiled path derives it from `con`). +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); From ebc55b9ed4148d4676f9e3478ce9edcc856a6fec Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Mon, 3 Aug 2026 20:37:38 -0400 Subject: [PATCH 2/2] Cover the compiled path too; correct the fallback-only claim The first fixture's comment said `hash` was load-bearing -- that only the row-wise fallback infers shape from data, so only it could produce a skewed arity. That is wrong, and the new program proves it: with `hash` dropped both maps lower to corgi logic and it still panics identically. `infer_term_shape` gives `Inject(tag, _)` an arity of `tag + 1`, so the compiled path under-approximates the variant universe exactly as `infer_shape_cols` does -- from the term instead of from the data. A single term reconciles its own arms (`If` joins them); two separate operators have nothing to reconcile them. Both programs panic without this PR's widening and pass with it. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_0168NJWzHwrLGW2RXToGooYP --- interactive/tests/corgi_backend.rs | 2 ++ interactive/tests/programs/sum_skew.ddp | 15 ++++++++------- interactive/tests/programs/sum_skew_compiled.ddp | 16 ++++++++++++++++ 3 files changed, 26 insertions(+), 7 deletions(-) create mode 100644 interactive/tests/programs/sum_skew_compiled.ddp diff --git a/interactive/tests/corgi_backend.rs b/interactive/tests/corgi_backend.rs index 507477b3f..ca50f5332 100644 --- a/interactive/tests/corgi_backend.rs +++ b/interactive/tests/corgi_backend.rs @@ -33,6 +33,7 @@ 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![ @@ -79,6 +80,7 @@ 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/sum_skew.ddp b/interactive/tests/programs/sum_skew.ddp index 794b01d2f..2fa9dbb58 100644 --- a/interactive/tests/programs/sum_skew.ddp +++ b/interactive/tests/programs/sum_skew.ddp @@ -1,11 +1,12 @@ --- Two collections whose variant columns commit DIFFERENT arms, concatenated into one --- arrangement. Shapes are inferred per batch from the data, so one side infers --- `Sum([Some(_)])` (tag 0 only) and 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 chunker must widen with uncommitted (⊥) lanes before it compares or gathers them. +-- 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. -- --- `hash` keeps both maps on the ROW-WISE fallback, the only path whose shape comes from --- the data rather than from the term (the compiled path derives it from `con`). +-- 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; 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);