fixup! Make the DDIR server parallel, timer-free, and Corgi-capable (#853)

Author: frankmcsherry

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/container.rs

@@ -69,6 +69,16 @@ impl<T: Clone + 'static, R: Clone + 'static> CorgiContainer<T, R> {
         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`.
     pub fn into_updates(self) -> Vec<((Row, Row), T, R)> {

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/reduce.rs

@@ -53,7 +53,7 @@ type CBatch<T> = Rc<ChunkBatch<CorgiChunk<T, Diff>>>;
 /// columns, allowing Corgi to swizzle their buffers in place when unshared.
 fn signed_order_view(value: CValue) -> CValue {
     match value {
-        value @ CValue::Prim(_) => NumOp::from(ArithOp::ToSigned).eval(value),
+        value @ CValue::Prim(_) => NumOp::from(ArithOp::ToSigned).eval(value).expect("ToSigned on a leaf"),
         CValue::Prod(fields) => {
             CValue::Prod(fields.into_iter().map(signed_order_view).collect())
         }
@@ -62,7 +62,7 @@ fn signed_order_view(value: CValue) -> CValue {
             within,
             variants
                 .into_iter()
-                .map(|variant| variant.map(signed_order_view))
+                .map(signed_order_view)
                 .collect(),
         ),
         CValue::List(bounds, values) => {