interactive: pin corgi at master; ids() borrows via leaf_slice

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

interactive/src/corgi/reduce.rs

@@ -169,14 +169,13 @@ fn concat_columns(blocks: &[CValue]) -> CValue {
 /// Applied CONSISTENTLY at every id site (both value presentations AND the freshly-produced
 /// `reduce_brackets` outputs), else `desired − current` nets across mismatched ids for the same value.
 fn ids(col: &CValue) -> Vec<u64> {
-    match corgi::shape_of_value(col) {
-        Shape::Prim(64) => col.clone().into_u64("ids"),
-        Shape::Prod(ref fs) if fs.len() == 1 && matches!(fs[0], Shape::Prim(64)) => match col {
-            CValue::Prod(fields) => fields[0].clone().into_u64("ids"),
-            _ => unreachable!("shape Prod but value not Prod"),
-        },
-        _ => corgi::hash(col).into_u64("ids"),
+    // Value-as-id: borrow the leaf and copy once, rather than `clone().into_u64()` — the
+    // clone bumps the `Arc`, so `into_u64`'s try-unwrap always fails and copies anyway,
+    // even for a freshly-gathered column with one holder.
+    if let Some(sl) = corgi::arrange::leaf_slice(col) {
+        return sl.to_vec();
     }
+    corgi::hash(col).into_u64("ids")
 }
 
 /// The `changed` set as a needle column in the chunks' own key shape — possible exactly
@@ -248,7 +247,16 @@ where
         }
     } else {
         for (ci, ch) in chunks.iter().enumerate() {
-            let kh = ids(ch.keys());
+            // Borrow the key leaf when there is one (`ids`' value-as-id fast paths); only
+            // structural keys need the hash, and only they pay a materialization. A shared
+            // column's `Arc` cannot be unwrapped, so `ids` would copy the whole key column
+            // here, once per chunk per retire, to read values it never mutates.
+            let hashed: Option<Vec<u64>> = corgi::arrange::leaf_slice(ch.keys()).is_none().then(|| ids(ch.keys()));
+            let kh: &[u64] = match (&hashed, corgi::arrange::leaf_slice(ch.keys())) {
+                (Some(v), _) => &v[..],
+                (None, Some(sl)) => sl,
+                (None, None) => unreachable!("leaf_slice absent implies hashed present"),
+            };
             for i in 0..kh.len() {
                 if changed.binary_search(&kh[i]).is_ok() {
                     tags.push(ci);
@@ -446,8 +454,13 @@ where
     }
 
     fn next_window(&mut self, instance: &ReduceInstance<'_, CBatch<T>, CBatch<T>>, changed: &[u64], cursor: &mut usize) -> Option<ReduceWindow<T, Diff, Diff>> {
-        // Single window: present ALL remaining changed keys at once (bounded-memory windowing is a
-        // later refinement). `changed` is ascending, so `binary_search` is the changed-key filter.
+        // Single window: present ALL remaining changed keys at once. This is NOT a deferred
+        // refinement — bounded windows were measured and rejected: at WINDOW = 1<<14, scc
+        // (100 rounds x batch 100) cost 84.4s against 63.7s, a 33% regression, while peak RSS
+        // fell only 356MB -> 340MB. Two reasons: the per-window, per-chunk seek setup is a
+        // fixed cost that multiplies by the window count, and the presentation is not the
+        // memory peak in the first place (the trace is). `changed` is ascending, so
+        // `binary_search` is the changed-key filter.
         if *cursor >= changed.len() {
             return None;
         }