From 719813a71f3e00b7c69c934b539b38800abddf64 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 9 Oct 2026 11:08:07 -0400 Subject: [PATCH 1/3] Bump corgi to WIP c068d1a Corgi's surface ops moved since 7d46146; the lowering follows. List indexing uses `Gather`, which is total: an index outside the list reads the zero of the element's shape. `Weave` reads its tags as bytes, so a list literal narrows its `iota` positions with `Cast(8)`. `AddU64` and `CmpOp::Gt` are now the immediate forms `BinImm` and `RelImm`. Corgi's reference values (`Ref`) are read as the rows they name, when transcoding and when ordering reduce values. Co-Authored-By: Claude Opus 5.5 --- interactive/Cargo.toml | 2 +- interactive/src/corgi/logic.rs | 29 ++++++++++++++++++++++++----- interactive/src/corgi/reduce.rs | 2 ++ 3 files changed, 27 insertions(+), 6 deletions(-) diff --git a/interactive/Cargo.toml b/interactive/Cargo.toml index 0019e7d0a..9369c50be 100644 --- a/interactive/Cargo.toml +++ b/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 = "c068d1aedf7b6b48c2713fe7b830d5c2e7ec6d29", features = ["serde"] } differential-dataflow = { workspace = true } serde = { version = "1.0", features = ["derive"] } smallvec = "1.15.1" diff --git a/interactive/src/corgi/logic.rs b/interactive/src/corgi/logic.rs index affb513f9..2045f7f43 100644 --- a/interactive/src/corgi/logic.rs +++ b/interactive/src/corgi/logic.rs @@ -97,6 +97,11 @@ pub fn transcode_owned(rows: Vec, 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 = (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 { }) .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::::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::::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, 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, 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. diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs index 7527812e8..d6eb5d147 100644 --- a/interactive/src/corgi/reduce.rs +++ b/interactive/src/corgi/reduce.rs @@ -65,6 +65,8 @@ fn signed_order_view(value: CValue) -> CValue { } CValue::List(bounds, values) => lexicographic_list_ranks(bounds, 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)), } } From c900684242418c2ed8d6e8312f58b3ef3f0970e8 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 9 Oct 2026 11:33:39 -0400 Subject: [PATCH 2/3] Bump corgi to WIP 4b1d23b; lists order as Corgi orders them Corgi now orders lists lexicographically, as DDIR does. Reduce's `Min` and `Collect` read list values in Corgi's own order, and the adapter that ranked lists lexicographically is removed. Signed integer leaves are still swizzled for the sort. Co-Authored-By: Claude Opus 5.5 --- interactive/Cargo.toml | 2 +- interactive/src/corgi/reduce.rs | 56 ++++----------------------------- 2 files changed, 7 insertions(+), 51 deletions(-) diff --git a/interactive/Cargo.toml b/interactive/Cargo.toml index 9369c50be..37e8347c9 100644 --- a/interactive/Cargo.toml +++ b/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 = "c068d1aedf7b6b48c2713fe7b830d5c2e7ec6d29", 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" diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs index d6eb5d147..68a2864a4 100644 --- a/interactive/src/corgi/reduce.rs +++ b/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 = Rc>>; -/// 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,59 +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 { From 7ab3bc14f1214dd722b49ccbd8e43eb2c7292094 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 9 Oct 2026 12:14:49 -0400 Subject: [PATCH 3/3] Raise the MSRV to 1.88 Corgi uses `std::hint::select_unpredictable`, stable since Rust 1.88. Co-Authored-By: Claude Opus 5.5 --- Cargo.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index 1d1d45622..461a8e6e3 100644 --- a/Cargo.toml +++ b/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" }