Bump corgi to WIP 4b1d23b

Cargo.toml

@@ -15,7 +15,7 @@ resolver = "2"
 
 [workspace.package]
 edition = "2021"
-rust-version = "1.86"
+rust-version = "1.88"
 
 [workspace.dependencies]
 differential-dataflow = { path = "differential-dataflow", default-features = false, version = "0.25.1" }

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 = "7d46146955d3f3f45c8d69d394684aa9b4e37dcc", features = ["serde"] }
+corgi = { git = "https://github.com/frankmcsherry/WIP", rev = "4b1d23bfc4ffc8a4ee51e2230216426a17f880a4", features = ["serde"] }
 differential-dataflow = { workspace = true }
 serde = { version = "1.0", features = ["derive"] }
 smallvec = "1.15.1"

interactive/src/corgi/logic.rs

@@ -97,6 +97,11 @@ pub fn transcode_owned(rows: Vec<DValue>, shape: &Shape) -> CValue {
             let lane_vals = payloads.into_iter().zip(lanes).map(|(p, lshape)| transcode_owned(p, lshape)).collect();
             CValue::sum(tags, lane_vals)
         }
+        // A reference names rows of a shared list: build the list, and refer to each of its rows.
+        Shape::Ref(list) => {
+            let named: Vec<usize> = (0..rows.len()).collect();
+            CValue::Ref(std::sync::Arc::new(transcode_owned(rows, list)), std::sync::Arc::new(named))
+        }
     }
 }
 
@@ -153,6 +158,11 @@ pub fn untranscode(col: CValue, shape: &Shape) -> Vec<DValue> {
                 })
                 .collect()
         }
+        // A row of references reads as the rows it names.
+        Shape::Ref(list) => match col {
+            CValue::Ref(arena, rows) => untranscode(corgi::arrange::gather(&arena, &rows), list),
+            other => untranscode(other, list),
+        },
     }
 }
 
@@ -298,14 +308,14 @@ pub fn compile(
             }
         }
         Term::Spread(_) => Err("`$n...` spread is only meaningful inside a tuple".into()),
-        // Projection: a tuple field, or a list element (`Get`, faulting out of range as `eval` does).
+        // Projection: a tuple field, or a list element (`Gather`, which reads the zero of the element's shape out of range).
         Term::Proj(t, i) => {
             let id = compile(t, b, env, env_shapes, anchor, None)?;
             match shape_of_term(t, env_shapes, None)? {
                 Shape::List(_) => {
                     let idx = b.add(Op::Lit(CValue::u64(vec![*i as u64])), vec![anchor]);
                     let pair = b.tuple(vec![idx, id]);
-                    Ok(b.add(Op::Get, vec![pair]))
+                    Ok(b.add(Op::Gather, vec![pair]))
                 }
                 _ => Ok(b.add(Op::Field(*i), vec![id])),
             }
@@ -642,7 +652,7 @@ pub fn compile(
                             let mut bb = Builder::<NumOp>::default();
                             let inp = bb.input();
                             let acc = bb.add(Op::Field(0), vec![inp]);
-                            let out = bb.add(ArithOp::AddU64(1), vec![acc]);
+                            let out = bb.add(ArithOp::BinImm(CBinOp::Add, Kind::U, 64, 1), vec![acc]);
                             bb.finish(out)
                         };
                         b.add(Op::Fold(Box::new(body)), vec![seed])
@@ -689,7 +699,16 @@ pub fn compile(
                 lanes.push(b.add(Op::Enlist, vec![e]));
             }
             let count = b.add(Op::Lit(CValue::u64(vec![fields.len() as u64])), vec![anchor]);
-            let mut weave_in = vec![b.add(Op::Iota, vec![count])];
+            // Weave reads its tags as bytes; `iota` counts in `U64`, so each tag is narrowed.
+            let positions = b.add(Op::Iota, vec![count]);
+            let narrow = {
+                let mut bb = Builder::<NumOp>::default();
+                let inp = bb.input();
+                let out = bb.add(Op::Cast(8), vec![inp]);
+                bb.finish(out)
+            };
+            let tags = b.add(Op::MapList(Box::new(narrow)), vec![positions]);
+            let mut weave_in = vec![tags];
             weave_in.extend(lanes);
             let woven_in = b.tuple(weave_in);
             let woven = b.add(Op::Weave, vec![woven_in]);
@@ -751,7 +770,7 @@ fn float_abs(b: &mut Builder<NumOp>, x: usize) -> usize {
 /// A 0/1 mask: is the float leaf NaN? The NaNs are exactly the keys whose magnitude exceeds +inf.
 fn is_nan(b: &mut Builder<NumOp>, x: usize) -> usize {
     let abs = float_abs(b, x);
-    b.add(CmpOp::Gt(float_key(f64::INFINITY)), vec![abs])
+    b.add(CmpOp::RelImm(Pred::Gt, 64, float_key(f64::INFINITY)), vec![abs])
 }
 
 /// Replace `-0.0` by `0.0` in a float leaf. Their keys are adjacent, so this adds the mask.

interactive/src/corgi/reduce.rs

@@ -20,8 +20,9 @@
 //!     `CorgiChunk` batch column-natively.
 //!
 //! Min/Collect values use a segmented structural sort over an order-only columnar view: signed
-//! integer leaves are swizzled, and lists become lexicographic ranks. The winning rows are still
-//! gathered from the original columns.
+//! integer leaves are swizzled, and lists keep their structure, as Corgi orders them
+//! lexicographically just as DDIR does. The winning rows are still gathered from the original
+//! columns.
 //!
 //! The changed-key restriction is honored by presenting only the changed keys: novel batches are
 //! read whole (delta-sized), the accumulated history is scanned and filtered to the changed hashes
@@ -46,7 +47,7 @@ use crate::ir::{Diff, Reducer};
 
 type CBatch<T> = Rc<ChunkBatch<CorgiChunk<T, Diff>>>;
 
-/// Build a sortable view matching DDIR's signed leaves and lexicographic lists.
+/// Build a sortable view matching DDIR's signed leaves.
 ///
 /// DDIR's leaf scalar is `Int`, transcoded into a Corgi primitive as its raw
 /// bits. Corgi's radix sort is unsigned, so XORing each payload leaf's sign bit
@@ -63,57 +64,14 @@ fn signed_order_view(value: CValue) -> CValue {
             // the lane assignment is untouched — only the payload lanes are swizzled.
             CValue::Sum(tags, variants.into_iter().map(signed_order_view).collect())
         }
-        CValue::List(bounds, values) => lexicographic_list_ranks(bounds, signed_order_view(*values)),
+        // Corgi orders lists lexicographically, as DDIR does, so only their elements are swizzled.
+        CValue::List(bounds, values) => CValue::List(bounds, Box::new(signed_order_view(*values))),
         CValue::Unit(len) => CValue::Unit(len),
+        // References order as the rows they name.
+        CValue::Ref(arena, rows) => signed_order_view(corgi::arrange::gather(&arena, &rows)),
     }
 }
 
-/// An order-only integer rank for each list, using DDIR's lexicographic order.
-/// Corgi's general structural order is intentionally length-first. Preserve
-/// that contract and adapt here: rank the element columns, then refine tied
-/// list prefixes in column batches, with end-of-list preceding every element.
-/// No DDIR rows or per-comparison interpreter calls are materialized.
-///
-/// This is a correctness adapter, not a performance-neutral view: it eagerly
-/// ranks all elements, including unused tails, and each prefix round scans all
-/// lists and allocates fresh scratch even when most prefixes are resolved.
-/// Cost therefore grows with total element count and unresolved prefix depth.
-/// Only list-valued subcolumns pay this ranking cost (including strings encoded
-/// as lists); physical arrangement-key ordering is unchanged.
-fn lexicographic_list_ranks(bounds: Bounds, ordered_elements: CValue) -> CValue {
-    let ends = bounds.to_vec();
-    let rows = ends.len();
-    let (element_perm, element_labels) = sort_blocks(&vec![0; ordered_elements.len()], &ordered_elements);
-    let mut element_ranks = vec![0; element_perm.len()];
-    for (i, &row) in element_perm.iter().enumerate() { element_ranks[row] = element_labels[i]; }
-    let starts: Vec<_> = std::iter::once(0).chain(ends.iter().copied()).take(rows).collect();
-    let mut perm: Vec<_> = (0..rows).collect();
-    let mut labels = vec![0; rows];
-    let mut position = 0;
-    loop {
-        let mut present = vec![0; rows];
-        let mut keys = vec![0; rows];
-        let mut active = false;
-        for i in 0..rows {
-            let tied = (i > 0 && labels[i] == labels[i-1]) || (i+1 < rows && labels[i] == labels[i+1]);
-            let row = perm[i];
-            if tied && position < ends[row]-starts[row] {
-                present[i] = 1;
-                keys[i] = element_ranks[starts[row]+position];
-                active = true;
-            }
-        }
-        if !active { break; }
-        let (order, refined) = sort_blocks(&labels, &CValue::Prod(vec![CValue::u64(present), CValue::u64(keys)]));
-        perm = order.into_iter().map(|i| perm[i]).collect();
-        labels = refined;
-        position += 1;
-    }
-    let mut ranks = vec![0; rows];
-    for (i, &row) in perm.iter().enumerate() { ranks[row] = labels[i]; }
-    CValue::u64(ranks)
-}
-
 /// Values named by id: id `i` is row `refs[i].1` of `columns[refs[i].0]`.
 #[derive(Default)]
 struct Pool {