DDIR: the sum universe — declared types, corgi as the typer, no shape…

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 = "20f38d69ba0f964e6dd0d904338bf433fea0c642" }
+corgi = { git = "https://github.com/frankmcsherry/WIP", rev = "578ef3fe79fdaa813ba0c721c08f60323f22d420", features = ["serde"] }
 differential-dataflow = { workspace = true }
 mimalloc = "0.1.48"
 serde = { version = "1.0", features = ["derive"] }

interactive/examples/programs/adt.ddp

@@ -2,8 +2,8 @@
 -- carrying the field-sum as payload, then matched back out to a (bucket; sum)
 -- and counted per bucket. (Contrived — exercises sum intro/outro + fold.)
 
-con Fwd(1) = 0;   -- src < dst
-con Bwd(1) = 1;   -- otherwise
+type Dir = Fwd u64   -- src < dst
+         | Bwd u64;  -- otherwise
 
 let edges = input 0 | key($0[0] ; $0[1]);
 

interactive/examples/programs/ast.ddp

@@ -12,8 +12,7 @@
 -- `$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;
+type Dir = Fwd u64 | Bwd u64;
 
 let rows = input 0 | key($0[0] ; $0[1]);
 

interactive/examples/programs/binders.ddp

@@ -1,7 +1,7 @@
 -- Depth-tracking test: a pattern binder `xs` used INSIDE a fold step. Correct
 -- de Bruijn resolution => xs is the case payload at Bound(2) inside the fold;
 -- a wrong index would hit the fold's Int element and panic on Proj.
-con Node(1) = 0;
+type Node = Node List(u64);
 
 let rows = input 0 | key($0[0] ; Node(list($0[0], $0[1])));
 

interactive/examples/programs/tour.ddp

@@ -3,8 +3,7 @@
 -- fixtures under tests/programs/ pin individual lowerings and their fallback
 -- routing; this program pins that the constructs work TOGETHER.
 
-con Fwd(1) = 0;
-con Bwd(1) = 1;
+type Dir = Fwd u64 | Bwd u64;
 
 let edges = input 0 | key($0[0] ; $0[1]);
 let roots = input 1 | key($0[0] ;);

interactive/src/corgi/bytes.rs

@@ -177,11 +177,32 @@ mod test {
         ]
     }
 
+    /// The declared shapes of each family — what a program's schema would pin (a variant carries
+    /// only its tag, so a family with sums cannot be pinned from a row).
+    fn shapes_of(name: &str) -> (corgi::Shape, corgi::Shape) {
+        use corgi::Shape::{List, Prim, Prod, Sum};
+        let u = || Prim(64);
+        let pair = || Prod(vec![u(), u()]);
+        match name {
+            "scalars" => (u(), u()),
+            "tuples" => (pair(), Prod(vec![u()])),
+            "lists" => (u(), List(Box::new(u()))),
+            "variants" => (Sum(vec![u(), pair()]), u()),
+            "nested" => (Prod(vec![u(), Sum(vec![u(), pair()])]), List(Box::new(u()))),
+            other => panic!("no shapes for family {other}"),
+        }
+    }
+
+    fn container_of(name: &str, updates: Vec<((DValue, DValue), Time, Diff)>) -> CorgiContainer<Time, Diff> {
+        let (k, v) = shapes_of(name);
+        CorgiContainer::from_updates(updates, &k, &v)
+    }
+
     /// Every update survives the round trip, with its time and diff, for every shape family.
     #[test]
     fn round_trip_preserves_updates() {
         for (name, updates) in shape_families() {
-            let c = CorgiContainer::<Time, Diff>::from_updates(updates.clone());
+            let c = container_of(name, updates.clone());
             let back = round_trip(&c);
             assert_eq!(back.into_updates(), updates, "{name} did not survive the round trip");
         }
@@ -202,7 +223,7 @@ mod test {
     #[test]
     fn round_trip_preserves_column_hashes() {
         for (name, updates) in shape_families() {
-            let c = CorgiContainer::<Time, Diff>::from_updates(updates);
+            let c = container_of(name, updates);
             let (keys, vals) = (c.keys.clone(), c.vals.clone());
             let back = round_trip(&c);
             assert_eq!(corgi::arrange::hash_rows(&back.keys), corgi::arrange::hash_rows(&keys), "{name} keys");
@@ -218,7 +239,7 @@ mod test {
         let updates: Vec<_> = (0..1000i64)
             .map(|i| ((DValue::Int(i), DValue::Int(i * 2)), time(0, &[]), 1))
             .collect();
-        let c = CorgiContainer::<Time, Diff>::from_updates(updates);
+        let c = CorgiContainer::<Time, Diff>::from_updates_pinned(updates);
         assert_eq!(corgi::bytes::length_in_bytes(&c.keys), 24 + 8 * 1000);
         assert_eq!(corgi::bytes::length_in_bytes(&c.vals), 24 + 8 * 1000);
     }

interactive/src/corgi/chunk.rs

@@ -75,7 +75,7 @@ impl<T: Columnar, R> Default for CorgiChunk<T, R> {
 
 /// Split a `Prod([keys, vals])` corgi value into its two columns.
 fn split_kv(kv: CValue) -> (CValue, CValue) {
-    let mut cols = kv.into_prod("corgi chunk kv");
+    let mut cols = kv.into_prod("corgi chunk kv").unwrap();
     let vals = cols.pop().unwrap();
     let keys = cols.pop().unwrap();
     (keys, vals)
@@ -497,7 +497,7 @@ pub fn present_key(keys: CValue) -> CValue {
     if corgi::arrange::leaf_slice(&keys).is_some() {
         return keys;
     }
-    let hashes = corgi::hash(&keys).into_u64("present_key");
+    let hashes = corgi::hash(&keys).into_u64("present_key").unwrap();
     CValue::Prod(vec![CValue::u64(hashes), keys])
 }
 
@@ -638,8 +638,8 @@ mod test {
     fn read_batch(b: &ChunkBatch<CorgiChunk<u64, i64>>) -> BTreeMap<((u64, u64), u64), i64> {
         let mut m = BTreeMap::new();
         for ch in &b.chunks {
-            let ks = ch.keys().clone().into_u64("k");
-            let vs = ch.vals().clone().into_u64("v");
+            let ks = ch.keys().clone().into_u64("k").unwrap();
+            let vs = ch.vals().clone().into_u64("v").unwrap();
             for i in 0..ch.len_() { *m.entry(((ks[i], vs[i]), ch.times().get(i))).or_insert(0) += ch.diffs()[i]; }
         }
         m.retain(|_, d| *d != 0);

interactive/src/corgi/container.rs

@@ -14,7 +14,7 @@ use timely::progress::{PathSummary, Timestamp};
 use differential_dataflow::collection::containers::{Enter, Leave, Negate, ResultsIn};
 use differential_dataflow::difference::Abelian;
 
-use crate::corgi::logic::{infer_shape_cols, transcode, untranscode};
+use crate::corgi::logic::{transcode, untranscode};
 use crate::ir::Value as DValue;
 
 type Row = DValue;
@@ -58,33 +58,29 @@ impl<T: 'static, R: 'static> Accountable for CorgiContainer<T, R> {
 }
 
 impl<T: Clone + 'static, R: Clone + 'static> CorgiContainer<T, R> {
-    /// Build a container from DDIR row updates — the **ingest boundary** transcode (once per batch).
-    ///
-    /// Intended to have exactly ONE caller: external rows entering the dataflow. The other
-    /// callers (with [`into_updates`](Self::into_updates)) are `apply_ops`' row-wise fallbacks,
-    /// each scheduled to disappear: `Project`/`Filter` with the `Case`/`Inject`/`List`/`Unary`/
-    /// `Hash` lowerings, `EnterAt` with bulk time mutation, `LiftIter` with a columnar
-    /// column-append, `FlatMap` with corgi's list ops. Row↔column round-trips inside the
-    /// dataflow are debt, not design.
-    /// Shapes are inferred by scanning the whole column ([`infer_shape_cols`]) — required so a
-    /// `Variant` column discovers all its arms (a single sample shows only one tag).
-    pub fn from_updates(updates: Vec<((Row, Row), T, R)>) -> Self {
-        if updates.is_empty() {
-            return Self::default();
-        }
+    /// Build a container from DDIR row updates at the collection's PINNED shapes — the **ingest
+    /// boundary** transcode (once per batch). The only rows→columns conversion in the corgi
+    /// backend: inside the dataflow every operator is columnar.
+    pub fn from_updates(updates: Vec<((Row, Row), T, R)>, kshape: &corgi::Shape, vshape: &corgi::Shape) -> Self {
         let keys_rows: Vec<DValue> = updates.iter().map(|u| u.0 .0.clone()).collect();
         let vals_rows: Vec<DValue> = updates.iter().map(|u| u.0 .1.clone()).collect();
-        let kshape = infer_shape_cols(&keys_rows);
-        let vshape = infer_shape_cols(&vals_rows);
         let times = updates.iter().map(|u| u.1.clone()).collect();
         let diffs = updates.iter().map(|u| u.2.clone()).collect();
-        CorgiContainer { keys: transcode(&keys_rows, &kshape), vals: transcode(&vals_rows, &vshape), times, diffs }
+        CorgiContainer { keys: transcode(&keys_rows, kshape), vals: transcode(&vals_rows, vshape), times, diffs }
+    }
+
+    /// Test convenience: build a container from row updates, pinning the shapes from the first
+    /// row (what the ingest operator does with the first batch it sees).
+    #[cfg(test)]
+    pub(crate) fn from_updates_pinned(updates: Vec<((Row, Row), T, R)>) -> Self {
+        use crate::corgi::logic::shape_of_row;
+        let Some(((k, v), _, _)) = updates.first() else { return Self::default() };
+        let (ks, vs) = (shape_of_row(k).unwrap(), shape_of_row(v).unwrap());
+        Self::from_updates(updates, &ks, &vs)
     }
 
     /// Read the container back to DDIR row updates — the **egress boundary** transcode (once).
-    /// corgi `Value` is self-describing, so shapes come from `shape_of_value`. Intended callers
-    /// are inspection/output edges; in-dataflow uses are fallback debt (see
-    /// [`from_updates`](Self::from_updates)).
+    /// corgi `Value` is self-describing, so shapes come from `shape_of_value`.
     pub fn into_updates(self) -> Vec<((Row, Row), T, R)> {
         if self.times.is_empty() {
             return Vec::new();

interactive/src/corgi/exchange.rs

@@ -111,7 +111,7 @@ impl<T: Clone + 'static, R: Clone + 'static> Distributor<CorgiContainer<T, R>> f
         }
         let peers = pushers.len();
 
-        let ids = corgi::hash(&container.keys).into_u64("corgi exchange: key hash");
+        let ids = corgi::hash(&container.keys).into_u64("corgi exchange: key hash").unwrap();
         self.counting_sort(&ids, peers);
 
         // Whole-container fast path. When every row shares a destination — a batch narrower than
@@ -213,7 +213,7 @@ mod test {
 
     /// Partition `updates` across `peers` destinations and read each destination back as rows.
     fn partition(updates: Vec<((DValue, DValue), Time, Diff)>, peers: usize) -> Vec<Vec<((DValue, DValue), Time, Diff)>> {
-        let mut container = CorgiContainer::<Time, Diff>::from_updates(updates);
+        let mut container = CorgiContainer::<Time, Diff>::from_updates_pinned(updates);
         let mut pushers: Vec<Collect<Time, Diff>> = (0..peers).map(|_| Collect::default()).collect();
         let mut distributor = CorgiDistributor::<Time, Diff>::default();
         distributor.partition(&mut container, &timely::progress::Stamp::from_elem(0u64), &mut pushers);
@@ -299,7 +299,7 @@ mod test {
     /// either way so no row is sent twice.
     #[test]
     fn the_input_is_consumed() {
-        let mut container = CorgiContainer::<Time, Diff>::from_updates(scalar_updates(50));
+        let mut container = CorgiContainer::<Time, Diff>::from_updates_pinned(scalar_updates(50));
         let mut pushers: Vec<Collect<Time, Diff>> = (0..4).map(|_| Collect::default()).collect();
         let mut distributor = CorgiDistributor::<Time, Diff>::default();
         distributor.partition(&mut container, &timely::progress::Stamp::from_elem(0u64), &mut pushers);

interactive/src/corgi/join.rs

@@ -174,9 +174,10 @@ impl<T: ColTime> ProxyJoinBackend<T, CBatch<T>, CBatch<T>> for CorgiJoinBackend<
             let kc = recover_key(&gather_lanes(&keys0, &tag0, &off0));
             let v0 = gather_lanes(&vals0, &tag0, &off0);
             let v1 = gather_lanes(&vals1, &tag1, &off1);
-            let proj = compile_join_projection(&self.key, &self.val, &shape_of_value(&kc), &shape_of_value(&v0), &shape_of_value(&v1));
+            let proj = compile_join_projection(&self.key, &self.val, &shape_of_value(&kc), &shape_of_value(&v0), &shape_of_value(&v1))
+                .unwrap_or_else(|e| panic!("join projection: type error: {e}"));
             let projected = corgi::eval_graph(&proj, CValue::Prod(vec![kc, v0, v1]));
-            let mut cols = projected.into_prod("corgi join projection");
+            let mut cols = projected.into_prod("corgi join projection").unwrap();
             let nv = cols.pop().unwrap();
             let nk = cols.pop().unwrap();
             output.push(CorgiContainer {
@@ -236,7 +237,7 @@ fn leaf_valued<T: ColTime>(chunks: &[&CorgiChunk<T, Diff>]) -> bool {
 fn pull_lanes(col: &CValue, idx: &[usize]) -> Vec<Vec<u64>> {
     leaf_lanes(col).expect("pull_lanes: leaf-laned column")
         .into_iter()
-        .map(|lane| gather(lane, idx).into_u64("corgi join lane pull"))
+        .map(|lane| gather(lane, idx).into_u64("corgi join lane pull").unwrap())
         .collect()
 }