Corgi-DDIR work

interactive/Cargo.toml

@@ -13,6 +13,8 @@ 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 = "1301b281501d5e70ab63f9405412770a3250d985" }
 differential-dataflow = { workspace = true }
 mimalloc = "0.1.48"
 serde = { version = "1.0", features = ["derive"] }

interactive/src/backend/mod.rs

@@ -11,6 +11,7 @@
 //! that pick a backend and call [`render_tree`].
 
 pub mod vec;
+pub mod corgi;
 // The columnar substrate is deferred on the Value-model port (it needs a
 // `Columnar`/columnar-storage story for `Value`); see `backend::vec`.
 // pub mod col;

interactive/src/corgi/col_times.rs

@@ -0,0 +1,149 @@
+//! Columnar per-tuple times for `CorgiChunk`.
+//!
+//! `Inner.times` was `Vec<T>` — one heap-ish `PointStamp` (`SmallVec`) per row, the dominant SCC
+//! allocation (~40% of the profile). Since our `T = Product<u64, PointStamp<u64>>` already derives
+//! `columnar::Columnar` (Product, PointStamp, and u64 all do), the same times live in SoA form in
+//! `<T as Columnar>::Container` — one pair of allocations (offsets + values) for the whole column
+//! instead of `n` `SmallVec`s.
+//!
+//! Times are compared IN PLACE via the container's derived `Ord` on `Ref` (both `Product` and
+//! `PointStamp` carry `#[columnar(derive(Ord, PartialOrd))]`), so merge/sort never materialize a
+//! `T`. An owned `T` is reconstructed (`get`) only where a `Lattice` op is unavoidable — `join` in
+//! the join cross-product, `advance_by` in compaction — or at the emit boundary handing `T` back to
+//! DD. Range copies (`emit`/`concat`) push `Ref`s straight across (`push_ref`), also no `T`.
+//!
+//! This is the O(data) time store; DD's `Chunk` boundary only ever sees whole chunks + frontier
+//! antichains (control complexity), so this stays entirely inside the backend — no DD change.
+
+use std::cmp::Ordering;
+
+use columnar::{Borrow, Clear, Columnar, Index, Len, Push};
+
+use differential_dataflow::lattice::Lattice;
+use timely::progress::Timestamp;
+
+/// A timestamp usable as a columnar time column: `Timestamp + Lattice` (DD's algebra) plus
+/// `Columnar` with an *ordered* `Ref` (so times compare in their SoA form). Our
+/// `Product<u64, PointStamp<u64>>` satisfies it — every layer derives `Columnar` with
+/// `#[columnar(derive(Ord, PartialOrd))]`.
+///
+/// The `Ref: Ord` requirement is an HRTB on a projection (`for<'a> Ref<'a, Self>: Ord`), which Rust
+/// does NOT imply from a plain `T: ColTime` bound at use sites — so it would otherwise go viral
+/// across every fn driving the `Chunk` methods. We discharge it ONCE here, in the blanket impl,
+/// behind [`ColTime::cmp_refs`]; downstream code compares times through that method and needs only
+/// `T: ColTime`.
+pub trait ColTime: Timestamp + Lattice + Columnar {
+    /// Order two SoA time references (via the derived `Ref: Ord`, lifetimes reborrowed to unify).
+    fn cmp_refs(a: columnar::Ref<'_, Self>, b: columnar::Ref<'_, Self>) -> Ordering;
+}
+
+impl<T> ColTime for T
+where
+    T: Timestamp + Lattice + Columnar,
+    for<'a> columnar::Ref<'a, T>: Ord,
+{
+    #[inline]
+    fn cmp_refs(a: columnar::Ref<'_, T>, b: columnar::Ref<'_, T>) -> Ordering {
+        let a = <<T as Columnar>::Container as Borrow>::reborrow_ref(a);
+        let b = <<T as Columnar>::Container as Borrow>::reborrow_ref(b);
+        a.cmp(&b)
+    }
+}
+
+/// SoA column of per-tuple times, backed by `<T as Columnar>::Container`.
+#[derive(Default)]
+pub struct ColTimes<T: Columnar> {
+    store: <T as Columnar>::Container,
+}
+
+impl<T: Columnar> ColTimes<T> {
+    #[inline]
+    pub fn new() -> Self {
+        ColTimes { store: Default::default() }
+    }
+
+    #[inline]
+    pub fn len(&self) -> usize {
+        self.store.borrow().len()
+    }
+
+    #[inline]
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+
+    /// Empty the column, retaining allocation (the emit-flush reuse in `advance`).
+    #[inline]
+    pub fn clear(&mut self) {
+        self.store.clear();
+    }
+
+    /// Append an owned time (columnar `Push<&T>`); the value is stored SoA, not cloned whole.
+    #[inline]
+    pub fn push(&mut self, t: &T) {
+        self.store.push(t);
+    }
+
+    /// Append `other`'s row `i` by pushing its `Ref` straight across — no `T` materialized. Used by
+    /// the range copies in `emit`/`concat` (`Container: Push<Self::Ref>`).
+    #[inline]
+    pub fn push_ref(&mut self, other: &ColTimes<T>, i: usize) {
+        self.store.push(other.store.borrow().get(i));
+    }
+
+    /// The owned time at row `i` — materializes a `T`. Reserve for `Lattice` ops and the emit
+    /// boundary; use `cmp`/`cmp_cross` for ordering.
+    #[inline]
+    pub fn get(&self, i: usize) -> T {
+        <T as Columnar>::into_owned(self.store.borrow().get(i))
+    }
+
+    /// Append rows `[s, e)` of `other`, pushing `Ref`s straight across — no `T` materialized. The
+    /// range copy used by `emit`/`concat`/merge-suffix.
+    #[inline]
+    pub fn push_range(&mut self, other: &ColTimes<T>, s: usize, e: usize) {
+        let b = other.store.borrow();
+        for i in s..e {
+            self.store.push(b.get(i));
+        }
+    }
+
+    /// Materialize the whole column to `Vec<T>` — the egress boundary (`SortedRun` for the join,
+    /// `CorgiContainer` for `as_collection`), where owned `T` is wanted anyway.
+    pub fn to_vec(&self) -> Vec<T> {
+        let b = self.store.borrow();
+        (0..b.len()).map(|i| <T as Columnar>::into_owned(b.get(i))).collect()
+    }
+
+    /// Order rows `i` and `j` within this column, in place via [`ColTime::cmp_refs`] (no `T`).
+    #[inline]
+    pub fn cmp(&self, i: usize, j: usize) -> Ordering
+    where
+        T: ColTime,
+    {
+        let b = self.store.borrow();
+        T::cmp_refs(b.get(i), b.get(j))
+    }
+
+    /// Order this column's row `i` against `other`'s row `j` (the two-pointer merge compare).
+    #[inline]
+    pub fn cmp_cross(&self, i: usize, other: &ColTimes<T>, j: usize) -> Ordering
+    where
+        T: ColTime,
+    {
+        let ba = self.store.borrow();
+        let bb = other.store.borrow();
+        T::cmp_refs(ba.get(i), bb.get(j))
+    }
+}
+
+/// Build a column from an iterator of owned times (the `FromIterator` path used at construction).
+impl<T: Columnar> FromIterator<T> for ColTimes<T> {
+    fn from_iter<I: IntoIterator<Item = T>>(iter: I) -> Self {
+        let mut store: <T as Columnar>::Container = Default::default();
+        for t in iter {
+            store.push(&t);
+        }
+        ColTimes { store }
+    }
+}

interactive/src/corgi/container.rs

@@ -0,0 +1,149 @@
+//! Phase 2 — the corgi-native container: corgi columns for the (key,val) payload, plain Rust
+//! Vecs for time/diff (corgi never touches the lattice). This is what flows on dataflow edges in
+//! the corgi backend; operators transform block→block via `eval_graph` with NO per-op transcode.
+//! Conversion to/from DDIR rows happens only at I/O boundaries (`from_updates`/`into_updates`).
+//!
+//! Minimal trait surface for a pipelined (single-worker) dataflow: `Accountable + Default + Clone`
+//! (= `timely::Container`). `Negate`/`Enter`/`Leave`/`ResultsIn` (iterative scopes) come with M3.
+
+use timely::Accountable;
+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::ir::Value as DValue;
+
+type Row = DValue;
+
+/// A batch of `(key, val, time, diff)` updates: payload columnar (corgi), time/diff native Rust.
+pub struct CorgiContainer<T, R> {
+    /// Key column (corgi columnar `Value`).
+    pub keys: corgi::Value,
+    /// Val column (corgi columnar `Value`).
+    pub vals: corgi::Value,
+    /// Per-update times (corgi never reads these; the Rust side keeps the lattice algebra).
+    pub times: Vec<T>,
+    /// Per-update diffs.
+    pub diffs: Vec<R>,
+}
+
+impl<T, R> Default for CorgiContainer<T, R> {
+    fn default() -> Self {
+        // Empty sentinel columns (shape-agnostic length-0 unit columns).
+        CorgiContainer { keys: corgi::Value::Unit(0), vals: corgi::Value::Unit(0), times: Vec::new(), diffs: Vec::new() }
+    }
+}
+
+impl<T: Clone, R: Clone> Clone for CorgiContainer<T, R> {
+    fn clone(&self) -> Self {
+        // corgi `Value` clone is an Arc bump on the leaf buffers — columns are shared, not copied.
+        CorgiContainer { keys: self.keys.clone(), vals: self.vals.clone(), times: self.times.clone(), diffs: self.diffs.clone() }
+    }
+}
+
+impl<T: 'static, R: 'static> Accountable for CorgiContainer<T, R> {
+    #[inline]
+    fn record_count(&self) -> i64 {
+        self.times.len() as i64
+    }
+}
+
+/// Draining a `CorgiContainer` yields its DDIR row updates (untranscode) — lets the standard
+/// `ContainerChunker<Vec<((key,val),T,R)>>` chunk a corgi-container stream into row chains for the
+/// reused `MergeBatcher`, which the `CorgiBatchBuilder` then transcodes back to corgi columns at
+/// arrangement build (the one ingest-boundary round-trip; reduce/join read corgi columns).
+impl<T: Clone + 'static, R: Clone + 'static> timely::container::DrainContainer for CorgiContainer<T, R> {
+    type Item<'a> = ((Row, Row), T, R) where Self: 'a;
+    type DrainIter<'a> = std::vec::IntoIter<((Row, Row), T, R)> where Self: 'a;
+    fn drain(&mut self) -> Self::DrainIter<'_> {
+        std::mem::take(self).into_updates().into_iter()
+    }
+}
+
+impl<T: Clone + 'static, R: Clone + 'static> CorgiContainer<T, R> {
+    /// Build a container from DDIR row updates — the **ingest boundary** transcode (once per batch).
+    /// 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();
+        }
+        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 }
+    }
+
+    /// 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)> {
+        if self.times.is_empty() {
+            return Vec::new();
+        }
+        let kshape = corgi::shape_of_value(&self.keys);
+        let vshape = corgi::shape_of_value(&self.vals);
+        let keys_rows = untranscode(self.keys, &kshape);
+        let vals_rows = untranscode(self.vals, &vshape);
+        keys_rows
+            .into_iter()
+            .zip(vals_rows)
+            .zip(self.times)
+            .zip(self.diffs)
+            .map(|(((k, v), t), d)| ((k, v), t, d))
+            .collect()
+    }
+}
+
+// --- Container traits required by the `Backend` bound + iterative scopes ---
+// time/diff live in Rust, so these are plain Rust passes; the corgi key/val columns only move
+// (`gather`) when `ResultsIn` drops rows. `Enter`/`Leave` are identity for DDIR's same-Time dynamic
+// timestamp model (region entry doesn't change the time type; `leave_dynamic` pops the coord).
+
+impl<T: Timestamp, R: Abelian + 'static> Negate for CorgiContainer<T, R> {
+    fn negate(mut self) -> Self {
+        for d in self.diffs.iter_mut() {
+            d.negate();
+        }
+        self
+    }
+}
+
+impl<T: Timestamp, R: 'static> Enter<T, T> for CorgiContainer<T, R> {
+    type InnerContainer = Self;
+    fn enter(self) -> Self {
+        self
+    }
+}
+
+impl<T: Timestamp, R: 'static> Leave<T, T> for CorgiContainer<T, R> {
+    type OuterContainer = Self;
+    fn leave(self) -> Self {
+        self
+    }
+}
+
+impl<T: Timestamp, R: Clone + 'static> ResultsIn<T::Summary> for CorgiContainer<T, R> {
+    fn results_in(self, step: &T::Summary) -> Self {
+        let n = self.times.len();
+        let mut keep = Vec::with_capacity(n);
+        let mut new_times = Vec::with_capacity(n);
+        for (i, t) in self.times.iter().enumerate() {
+            if let Some(nt) = step.results_in(t) {
+                keep.push(i);
+                new_times.push(nt);
+            }
+        }
+        if keep.len() == n {
+            return CorgiContainer { keys: self.keys, vals: self.vals, times: new_times, diffs: self.diffs };
+        }
+        let keys = corgi::arrange::gather(&self.keys, &keep);
+        let vals = corgi::arrange::gather(&self.vals, &keep);
+        let diffs = keep.iter().map(|&i| self.diffs[i].clone()).collect();
+        CorgiContainer { keys, vals, times: new_times, diffs }
+    }
+}