diff --git a/interactive/Cargo.toml b/interactive/Cargo.toml index c31ee2212..b3e3a028d 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 = "1301b281501d5e70ab63f9405412770a3250d985" } +corgi = { git = "https://github.com/frankmcsherry/wip", rev = "c4626fce02288594c9806e9b747a19598d680e0a" } differential-dataflow = { workspace = true } mimalloc = "0.1.48" serde = { version = "1.0", features = ["derive"] } diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs index 1435d0a59..68ceacd21 100644 --- a/interactive/src/corgi/reduce.rs +++ b/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 { - 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> = 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, CBatch>, changed: &[u64], cursor: &mut usize) -> Option> { - // 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; }