diff --git a/differential-dataflow/src/operators/int_proxy/join.rs b/differential-dataflow/src/operators/int_proxy/join.rs index 401cad3fe..a1ceaaa22 100644 --- a/differential-dataflow/src/operators/int_proxy/join.rs +++ b/differential-dataflow/src/operators/int_proxy/join.rs @@ -45,7 +45,10 @@ pub trait ProxyJoinBackend> { bridge1: &mut ProxyBridge, ); - /// Interpret a list of matching identifiers, translate them to outputs, and place them in `output`. + /// Interpret matches derived from the immediately preceding [`Self::advance`] call and place + /// them in `output`. The iterator calls `cross` before another `advance`, so a backend may keep + /// block-local interpretation state between the two calls. `cross` may be skipped when the + /// block produced no matches, in which case the next `advance` may overwrite that state. fn cross( &mut self, instance: &JoinInstance, diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs index 75bc95450..345a4a0ac 100644 --- a/differential-dataflow/src/operators/int_proxy/reduce.rs +++ b/differential-dataflow/src/operators/int_proxy/reduce.rs @@ -147,7 +147,7 @@ impl ProxyReduceTactic { } } -fn assert_pending_frontier(pending: &BTreeMap>, maintained: &Antichain) { +fn debug_assert_pending_frontier(pending: &BTreeMap>, maintained: &Antichain) { debug_assert!({ let mut expected = Antichain::new(); for time in pending.values().flatten() { expected.insert_ref(time); } @@ -214,7 +214,7 @@ where // beyond `upper` can remain when nothing is due, and releasing their capabilities would // strand them (see the frontier clause of the `ReduceTactic::retire` contract). if changed.is_empty() && instance.input_batches.iter().all(|b| b.is_empty()) { - assert_pending_frontier(&self.pending, &pending_frontier); + debug_assert_pending_frontier(&self.pending, &pending_frontier); return (Vec::new(), pending_frontier); } @@ -397,7 +397,7 @@ where } let produced: Vec<(B1::Time, B2)> = tile_held.into_iter().zip(self.backend.finish()).collect(); - assert_pending_frontier(&self.pending, &pending_frontier); + debug_assert_pending_frontier(&self.pending, &pending_frontier); (produced, pending_frontier) } } diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs index 6eec6d6fd..8b1d3efd7 100644 --- a/interactive/src/backend/corgi.rs +++ b/interactive/src/backend/corgi.rs @@ -24,7 +24,7 @@ use corgi::arrange::gather; use corgi::Value as CValue; use crate::backend::Backend; -use crate::corgi::chunk::{CorgiChunk, CorgiChunker}; +use crate::corgi::chunk::{recover_key, CorgiChunk, CorgiChunker}; use crate::corgi::container::CorgiContainer; use crate::corgi::join::CorgiJoinBackend; use crate::corgi::reduce::CorgiReduceBackend; @@ -315,7 +315,9 @@ impl Backend for CorgiBackend { for batch in data.iter() { for ch in batch.chunks.iter().filter(|c| c.len() > 0) { let mut c = CorgiContainer { - keys: ch.keys().clone(), + // Drop the arrangement's leading identifier lane: edges carry + // the key the program wrote, so `$0` indexes what it always did. + keys: recover_key(ch.keys()), vals: ch.vals().clone(), times: ch.times().to_vec(), diffs: ch.diffs().to_vec(), diff --git a/interactive/src/corgi/chunk.rs b/interactive/src/corgi/chunk.rs index 4b33b21a3..35c617a6c 100644 --- a/interactive/src/corgi/chunk.rs +++ b/interactive/src/corgi/chunk.rs @@ -416,6 +416,10 @@ where /// ingest — no transcode). pub fn from_columns(keys: CValue, vals: CValue, times: Vec, diffs: Vec) -> Self { let (keys, vals, times, diffs) = sort_consolidate(keys, vals, times, diffs); + debug_assert!({ + let lane = corgi::arrange::leaf_slice(key_lane(&keys)); + lane.is_some_and(|ids| ids.windows(2).all(|pair| pair[0] <= pair[1])) + }, "arrangement key must lead with a sorted u64 identifier lane"); Self::from_parts(keys, vals, ColTimes::from_iter(times), diffs) } @@ -472,6 +476,71 @@ impl Default for CorgiChunker { } } +/// An arrangement's key column, in the form every consumer of a `CorgiChunk` can rely on: +/// **it leads with an integer, and the chunk is sorted by that integer.** +/// +/// A key that is already a primitive integer (a bare 64-bit `Prim`, or the 1-field `Prod` that +/// [`corgi::arrange::leaf_slice`] also reads through) is used as it stands — the value IS the +/// identifier, injectively, and a hash lane would cost 8 bytes a row to say the same thing. +/// Any other key shape — multi-field `Prod`, `List`, `Sum`, `Unit` — is hashed and the hash is +/// PREPENDED, so the key becomes `Prod([hash, key])`. `CorgiChunk::from_columns` then sorts +/// lexicographically over lanes, which is hash order with the real key as tie-break. +/// +/// The original key stays in the column, which is what makes the hash safe: colliding keys land +/// adjacent and sub-sorted, so they are told apart by comparison rather than by luck, and reads +/// recover the real key by [`recover_key`]. The two forms are distinguishable after the fact +/// (`leaf_slice` succeeds on exactly the un-prepended one) because no compound key reaches an +/// arrangement un-prepended. +/// +/// The hash is computed ONCE here, at ingest, and thereafter moves as data: `merge`, `advance` +/// and `settle` permute key columns with `gather_lanes`, so no transducer recomputes it. +pub fn present_key(keys: CValue) -> CValue { + if corgi::arrange::leaf_slice(&keys).is_some() { + return keys; + } + let hashes = corgi::hash(&keys).into_u64("present_key"); + CValue::Prod(vec![CValue::u64(hashes), keys]) +} + +/// The integer identifier of each row of a [`present_key`] column: the key's own values when it is +/// a primitive integer, and the prepended hash lane otherwise. Never re-hashes. +pub fn key_ids(keys: &CValue) -> Vec { + if let Some(sl) = corgi::arrange::leaf_slice(keys) { + return sl.to_vec(); + } + corgi::arrange::leaf_slice(key_lane(keys)).expect("a prepended hash lane is a u64 leaf").to_vec() +} + +/// The single column an arrangement is sorted by, for seeking: the key itself when it is a +/// primitive integer, else the prepended hash lane. Always a bare `u64` leaf, so `find_ranges` +/// over it takes corgi's `u64` fast path whatever the underlying key shape. +/// +/// One rule covers both forms, because [`present_key`] leaves exactly three possibilities: a bare +/// `Prim`, the 1-field `Prod` that also counts as primitive, or a prepended `Prod([hash, key])`. +/// The leading field is the identifier in all three. +pub fn key_lane(keys: &CValue) -> &CValue { + match keys { + CValue::Prod(cols) => &cols[0], + _ => keys, + } +} + +/// Whether [`present_key`] prepended a hash to this key — i.e. whether rows sharing an identifier +/// may hold DIFFERENT keys. False for primitive-integer keys, whose identifier is injective, so +/// readers can skip the checks that guard against collisions entirely. +pub fn key_is_hashed(keys: &CValue) -> bool { + corgi::arrange::leaf_slice(keys).is_none() +} + +/// Undo [`present_key`]: the key as the rest of the system knows it. A corgi clone is an `Arc` +/// bump, so dropping the hash lane costs nothing. +pub fn recover_key(keys: &CValue) -> CValue { + match keys { + CValue::Prod(cols) if corgi::arrange::leaf_slice(keys).is_none() => cols[1].clone(), + _ => keys.clone(), + } +} + /// Concatenate column blocks into one column (multi-source `gather_lanes`, no sort). fn concat_blocks(blocks: &[CValue]) -> CValue { if blocks.len() == 1 { @@ -495,7 +564,7 @@ where if self.times.is_empty() { return; } - let keys = concat_blocks(&self.k_blocks); + let keys = present_key(concat_blocks(&self.k_blocks)); let vals = concat_blocks(&self.v_blocks); self.k_blocks.clear(); self.v_blocks.clear(); diff --git a/interactive/src/corgi/join.rs b/interactive/src/corgi/join.rs index 82e514075..71df9d95d 100644 --- a/interactive/src/corgi/join.rs +++ b/interactive/src/corgi/join.rs @@ -8,13 +8,11 @@ //! //! # Tokens //! -//! * The **group token** is the key's own `u64` when the key column is a leaf (DDIR `Int` -//! keys transcode to a `U64` leaf, and chunk order IS `u64` order), which makes `from` -//! seekable (`find_ranges` with a one-row needle) and blocks resumable. Structured keys -//! have no order-preserving `u64` embedding, so they take a fallback: the whole -//! intersection in ONE block, tokens an ordinal counter (block-scoped, both sides -//! assigned by the same walk). Containers are still cut at `TARGET_OUT` either way — -//! the fallback forgoes only the bounded-bridge property, not bounded output. +//! * The **group token** is the arrangement's leading `u64` identifier: the key itself for +//! primitive integer keys, or the carried content hash for structured keys. Chunk order is +//! identifier order, which makes `from` seekable and every key shape resumably blockable. +//! A hash collision remains under that token through proxy matching; only then does `cross` +//! compare the recorded real-key coordinates and discard unequal pairs. //! //! * The **value token** is a *canonical coordinate*: `(chunk << 48) | row` of the value's //! first occurrence among its side's chunks. Coordinates redeem against the instance @@ -40,7 +38,7 @@ use differential_dataflow::trace::chunk::{Chunk, ChunkBatch}; use corgi::arrange::{compare_at, find_ranges, gather, gather_lanes}; use corgi::{shape_of_value, Shape, Value as CValue}; -use crate::corgi::chunk::CorgiChunk; +use crate::corgi::chunk::{key_is_hashed, key_lane, recover_key, CorgiChunk}; use crate::corgi::col_times::ColTime; use crate::corgi::container::CorgiContainer; use crate::corgi::logic::compile_join_projection; @@ -59,12 +57,16 @@ const COORD_BITS: u32 = 48; pub struct CorgiJoinBackend { key: Term, val: Term, + /// Identifier tokens in the current block that cover more than one real key. `advance` writes + /// this and the immediately following `cross` reads it before the next `advance`; only matches + /// under these astronomically rare tokens need a real-key comparison. + colliding: Vec, _t: PhantomData, } impl CorgiJoinBackend { pub fn new(key: Term, val: Term) -> Self { - CorgiJoinBackend { key, val, _t: PhantomData } + CorgiJoinBackend { key, val, colliding: Vec::new(), _t: PhantomData } } } @@ -81,17 +83,36 @@ impl ProxyJoinBackend, CBatch> for CorgiJoinBackend bridge0: &mut ProxyBridge, bridge1: &mut ProxyBridge, ) { + self.colliding.clear(); let chunks0 = side_chunks(&instance.batches0); let chunks1 = side_chunks(&instance.batches1); - if chunks0.iter().all(|c| c.len() == 0) || chunks1.iter().all(|c| c.len() == 0) { + if chunks0.is_empty() || chunks1.is_empty() { *from = None; return; } - match (leaf_key_lanes(&chunks0), leaf_key_lanes(&chunks1)) { - (Some(1), Some(1)) => advance_leaf(&chunks0, &chunks1, &instance.lower, from, bridge0, bridge1), - (Some(_), Some(_)) => advance_lanes(&chunks0, &chunks1, &instance.lower, from, bridge0, bridge1), - _ => advance_structured(&chunks0, &chunks1, &instance.lower, from, bridge0, bridge1), - } + debug_assert_eq!( + key_is_hashed(chunks0[0].keys()), + key_is_hashed(chunks1[0].keys()), + "join sides disagree on whether the arrangement key carries a hash lane", + ); + debug_assert!( + chunks0.iter().all(|chunk| key_is_hashed(chunk.keys()) == key_is_hashed(chunks0[0].keys())) + && chunks1.iter().all(|chunk| key_is_hashed(chunk.keys()) == key_is_hashed(chunks1[0].keys())), + "chunks of one join key disagree on whether they carry a hash lane", + ); + // Every arrangement key leads with its identifier and is sorted by it, so one resumable + // leaf walk applies to every key shape. A hash collision is retained in proxy space and + // recorded here; `cross` then filters only that token's unequal real-key pairs. Restarting + // through a whole-key walk is not valid after an earlier block has already retired. + advance_leaf( + &chunks0, + &chunks1, + &instance.lower, + from, + bridge0, + bridge1, + &mut self.colliding, + ); } fn cross( @@ -113,13 +134,44 @@ impl ProxyJoinBackend, CBatch> for CorgiJoinBackend while start < n { let end = (start + TARGET_OUT).min(n); tag0.clear(); off0.clear(); tag1.clear(); off1.clear(); - for (_, (c0, c1)) in &matches.ids[start..end] { - tag0.push((c0 >> COORD_BITS) as usize); - off0.push((c0 & ((1 << COORD_BITS) - 1)) as usize); - tag1.push((c1 >> COORD_BITS) as usize); - off1.push((c1 & ((1 << COORD_BITS) - 1)) as usize); + let kept = if self.colliding.is_empty() { + // The universal hot path: preserve the old allocation-free slice copies. + for (_, (c0, c1)) in &matches.ids[start..end] { + tag0.push((c0 >> COORD_BITS) as usize); + off0.push((c0 & ((1 << COORD_BITS) - 1)) as usize); + tag1.push((c1 >> COORD_BITS) as usize); + off1.push((c1 & ((1 << COORD_BITS) - 1)) as usize); + } + None + } else { + let mut kept = Vec::with_capacity(end - start); + for (index, (token, (c0, c1))) in matches.ids[start..end].iter().enumerate() { + let c0_chunk = (c0 >> COORD_BITS) as usize; + let c0_row = (c0 & ((1 << COORD_BITS) - 1)) as usize; + let c1_chunk = (c1 >> COORD_BITS) as usize; + let c1_row = (c1 & ((1 << COORD_BITS) - 1)) as usize; + if self.colliding.binary_search(token).is_ok() + && compare_at(chunks0[c0_chunk].keys(), c0_row, chunks1[c1_chunk].keys(), c1_row) + != Ordering::Equal + { + continue; + } + kept.push(start + index); + tag0.push(c0_chunk); + off0.push(c0_row); + tag1.push(c1_chunk); + off1.push(c1_row); + } + Some(kept) + }; + if kept.as_ref().is_some_and(Vec::is_empty) { + start = end; + continue; } - let kc = gather_lanes(&keys0, &tag0, &off0); + // The join's projection is written against the key the program declared, so drop + // the arrangement's leading identifier lane before evaluating it. The output goes to + // an arrange, which re-derives the identifier for the new key. + let kc = recover_key(&gather_lanes(&keys0, &tag0, &off0)); let v0 = gather_lanes(&vals0, &tag0, &off0); let v1 = gather_lanes(&vals1, &tag1, &off1); let proj = compile_join_projection(&self.key, &self.val, &shape_of_value(&kc), &shape_of_value(&v0), &shape_of_value(&v1)); @@ -130,8 +182,14 @@ impl ProxyJoinBackend, CBatch> for CorgiJoinBackend output.push(CorgiContainer { keys: nk, vals: nv, - times: matches.times[start..end].to_vec(), - diffs: matches.diffs[start..end].to_vec(), + times: kept.as_ref().map_or_else( + || matches.times[start..end].to_vec(), + |kept| kept.iter().map(|&index| matches.times[index].clone()).collect(), + ), + diffs: kept.as_ref().map_or_else( + || matches.diffs[start..end].to_vec(), + |kept| kept.iter().map(|&index| matches.diffs[index]).collect(), + ), }); start = end; } @@ -141,7 +199,13 @@ impl ProxyJoinBackend, CBatch> for CorgiJoinBackend /// The instance's chunks on one side, in the deterministic order coordinates index /// (batches in order, chunks in order) — `advance` and `cross` must agree on it. fn side_chunks(batches: &[CBatch]) -> Vec<&CorgiChunk> { - let chunks: Vec<&CorgiChunk> = batches.iter().flat_map(|b| b.chunks.iter()).collect(); + // Coordinates use this same filtered ordering in `advance` and `cross`. Excluding empties here + // makes the convention explicit and keeps `ident` from inspecting a shape-less `Unit(0)` key. + let chunks: Vec<&CorgiChunk> = batches + .iter() + .flat_map(|batch| batch.chunks.iter()) + .filter(|chunk| chunk.len() > 0) + .collect(); assert!(chunks.len() < (1 << (64 - COORD_BITS)), "too many chunks for coordinate packing"); chunks } @@ -163,38 +227,11 @@ fn leaf_lanes(col: &CValue) -> Option> { if walk(col, &mut out) { Some(out) } else { None } } -/// The key columns' common leaf-lane count: `Some(n)` when every nonempty chunk's key -/// flattens to exactly `n` 64-bit lanes. `Some(1)` keys use their own value as the group -/// token (chunk order IS `u64` order); `Some(n>1)` keys walk the lane-tuple path (ordinal -/// tokens, one block); `None` keys (sums/lists in the key) take the structural walk. -fn leaf_key_lanes(chunks: &[&CorgiChunk]) -> Option { - let mut lanes: Option = None; - for c in chunks.iter().filter(|c| c.len() > 0) { - let n = leaf_lanes(c.keys())?.len(); - match lanes { - None => lanes = Some(n), - Some(m) if m == n => {} - _ => return None, - } - } - lanes -} - /// Whether every nonempty chunk's val column flattens to `u64` leaf lanes. fn leaf_valued(chunks: &[&CorgiChunk]) -> bool { chunks.iter().filter(|c| c.len() > 0).all(|c| leaf_lanes(c.vals()).is_some()) } -/// A needle column of `keys`, shaped like `col` (a single-leaf-lane key column). -fn needle_like(col: &CValue, keys: &[u64]) -> CValue { - match col { - CValue::Prim(_) => CValue::u64(keys.to_vec()), - CValue::Prod(fields) => CValue::Prod(fields.iter().map(|f| needle_like(f, keys)).collect()), - CValue::Unit(_) => CValue::Unit(keys.len()), - _ => unreachable!("needle_like: single-leaf key columns only"), - } -} - /// Pull rows `idx` of the column's leaf lanes as `u64` buffers. fn pull_lanes(col: &CValue, idx: &[usize]) -> Vec> { leaf_lanes(col).expect("pull_lanes: leaf-laned column") @@ -203,24 +240,6 @@ fn pull_lanes(col: &CValue, idx: &[usize]) -> Vec> { .collect() } -/// One past the end of the key run beginning at `start` (gallop + binary search, same-column -/// structural compares). -fn run_end(keys: &CValue, start: usize, len: usize) -> usize { - let same = |i: usize| compare_at(keys, i, keys, start) == Ordering::Equal; - let mut step = 1usize; - let mut lo = start; // last known-equal - while lo + step < len && same(lo + step) { - lo += step; - step <<= 1; - } - let mut hi = (lo + step).min(len); // first known-beyond (or len) - while lo + 1 < hi { - let mid = lo + (hi - lo) / 2; - if same(mid) { lo = mid; } else { hi = mid; } - } - hi -} - /// One key's records in one chunk: absolute rows `[s, e)`. When the vals are leaf-laned, /// `vals = (lanes, pos)` gives row `r`'s tuple as `lanes[.][pos + r - s]`; `None` falls /// back to structural compares against the chunk itself. @@ -349,97 +368,72 @@ impl SideScratch { } } -/// The equal-key runs at the current positions: for each chunk whose next key equals the -/// candidate at `(cand_chunk, cand_row)`, its `(chunk, start, end)` — advancing positions -/// past every run taken. -fn take_runs( - chunks: &[&CorgiChunk], - pos: &mut [usize], - cand: (&CorgiChunk, usize), - runs: &mut Vec<(usize, usize, usize)>, -) { - runs.clear(); - for (c, chunk) in chunks.iter().enumerate() { - let p = pos[c]; - if p < chunk.len() && compare_at(chunk.keys(), p, cand.0.keys(), cand.1) == Ordering::Equal { - let e = run_end(chunk.keys(), p, chunk.len()); - runs.push((c, p, e)); - pos[c] = e; - } - } +/// The identifier column of a chunk's keys, as a sorted `u64` slice. Every arrangement key +/// leads with its identifier ([`present_key`](crate::corgi::chunk::present_key)), so this is +/// always available and always ordered — no gather, no copy. +fn ident<'a, T: ColTime>(chunk: &'a CorgiChunk) -> &'a [u64] { + corgi::arrange::leaf_slice(key_lane(chunk.keys())).expect("the identifier lane is a u64 leaf") } -/// The chunk holding the least current key across both sides, or `None` when exhausted. -fn min_key<'a, T: ColTime>( - chunks0: &[&'a CorgiChunk], - pos0: &[usize], - chunks1: &[&'a CorgiChunk], - pos1: &[usize], -) -> Option<(&'a CorgiChunk, usize)> { - let mut best: Option<(&CorgiChunk, usize)> = None; - for (&chunk, &p) in chunks0.iter().zip(pos0.iter()).chain(chunks1.iter().zip(pos1.iter())) { - if p < chunk.len() { - match best { - None => best = Some((chunk, p)), - Some((bc, bp)) => { - if compare_at(chunk.keys(), p, bc.keys(), bp) == Ordering::Less { - best = Some((chunk, p)); - } - } - } +/// The block's exclusive identifier bound: for each chunk with more than `budget` rows left, +/// take the identifier `budget` rows in and bump past its whole run, then choose the least of +/// those bounds. A chunk may therefore contribute `budget + run length` rows rather than obeying +/// a hard cap; the expansion is what keeps one identifier wholly within one block. `None` when no +/// chunk is over budget — the block runs to the end and the walk is exhausted. +/// +/// This is the whole point of a totally ordered identifier: the block's extent is decided by +/// reading ONE value per chunk, before anything is read in bulk, so each chunk can then be +/// read exactly once over exactly the rows the block needs. Deciding it the other way round — +/// read a fixed number of rows, then discover how far the block can reach — leaves whatever +/// was read past the bound to be discarded and read again next block, and the further apart +/// the chunks' key densities are, the more that is. +fn block_horizon(chunks: &[&CorgiChunk], starts: &[usize], budget: usize) -> Option { + let mut horizon: Option = None; + for (c, &s) in chunks.iter().zip(starts) { + let lane = ident(c); + if lane.len() - s <= budget { + continue; } + // Past the whole run of the identifier at the budget, so the block ends at a key + // boundary. `None` on overflow means nothing can exceed it: run to the end. + let Some(bound) = lane[s + budget].checked_add(1) else { continue }; + horizon = Some(horizon.map_or(bound, |h: u64| h.min(bound))); } - best + horizon +} + +/// Where each chunk's block ends: the first row at or past `horizon`, or its end. +fn block_ends(chunks: &[&CorgiChunk], horizon: Option) -> Vec { + chunks.iter().map(|c| match horizon { + Some(h) => ident(c).partition_point(|&x| x < h), + None => c.len(), + }).collect() } -/// Pulled-buffer view over one leaf-keyed chunk of the DRIVER side: the block's key rows -/// extracted once per `advance` call as `u64` buffers, so the key merge runs on slices. +/// View over one leaf-keyed chunk's rows for THIS block, `[base, end)`. The identifiers are +/// borrowed from the chunk's own lane — the block reads them, it does not copy them — and the +/// vals are gathered once, over exactly those rows. struct LeafView<'a, T: ColTime> { chunk: &'a CorgiChunk, cid: usize, /// Absolute row of `keys[0]`. base: usize, - keys: Vec, + keys: &'a [u64], /// Leaf-laned vals over the same rows; `None` when vals are structured. vals: Option>>, /// Cursor within `keys`. cur: usize, } -/// Rows pulled per driver chunk per `advance` call. Unprocessed pulled rows are re-pulled -/// by the next call (bounded per-call waste, in exchange for pull-once simplicity). +/// Rows per chunk per block. Bounds the block's working set; nothing outside it is read. const PULL: usize = 1 << 14; impl<'a, T: ColTime> LeafView<'a, T> { - fn new(chunk: &'a CorgiChunk, cid: usize, start: usize, leaf_vals: bool) -> Self { - let mut view = LeafView { chunk, cid, base: start, keys: Vec::new(), vals: if leaf_vals { Some(Vec::new()) } else { None }, cur: 0 }; - view.extend_pull(PULL); - view - } - fn abs_end(&self) -> usize { - self.base + self.keys.len() - } - fn fully_pulled(&self) -> bool { - self.abs_end() == self.chunk.len() - } - /// Append up to `more` further rows to the buffers. - fn extend_pull(&mut self, more: usize) { - let s = self.abs_end(); - let e = (s + more).min(self.chunk.len()); - if s == e { return; } - let idx: Vec = (s..e).collect(); - let key_lane = leaf_lanes(self.chunk.keys()).expect("single-lane keys")[0]; - self.keys.extend(gather(key_lane, &idx).into_u64("corgi join key pull")); - if let Some(vals) = self.vals.as_mut() { - let pulled = pull_lanes(self.chunk.vals(), &idx); - if vals.is_empty() { - *vals = pulled; - } else { - for (lane, more) in vals.iter_mut().zip(pulled) { lane.extend(more); } - } - } + fn new(chunk: &'a CorgiChunk, cid: usize, start: usize, end: usize, leaf_vals: bool) -> Self { + let vals = (leaf_vals && end > start).then(|| pull_lanes(chunk.vals(), &(start..end).collect::>())); + LeafView { chunk, cid, base: start, keys: &ident(chunk)[start..end], vals, cur: 0 } } - /// The key under the cursor, if any remains in the buffer. + /// The key under the cursor, if any remains in this block. fn cur_key(&self) -> Option { self.keys.get(self.cur).copied() } @@ -477,7 +471,7 @@ struct Probe<'a, T: ColTime> { impl<'a, T: ColTime> Probe<'a, T> { fn new(chunk: &'a CorgiChunk, cid: usize, needles: &CValue, leaf_vals: bool) -> Self { - let (lo, hi) = find_ranges(needles, chunk.keys()); + let (lo, hi) = find_ranges(needles, key_lane(chunk.keys())); let mut off = Vec::with_capacity(lo.len() + 1); let mut idx: Vec = Vec::new(); off.push(0); @@ -506,8 +500,78 @@ impl<'a, T: ColTime> Probe<'a, T> { } } -/// Blockwise `advance` for leaf-keyed inputs: group token = the key's own `u64` (chunk order -/// IS `u64` order), so blocks resume by seeking `from` and end at key boundaries. +/// Whether every row covered by `refs` carries the same KEY, not merely the same identifier. +/// +/// The identifier is injective for a primitive-integer key, but a hashed key can in principle put +/// two distinct keys under one identifier, and the leaf path would then cross-product them. Runs +/// are contiguous and sub-sorted by the real key ([`present_key`](crate::corgi::chunk::present_key) +/// keeps the key in the column, after the hash), so a run holds one key exactly when its first and +/// last rows agree — a handful of `compare_at` per matched key, never per row. `refs` spans both +/// sides, so one pass also confirms the two sides matched on the key and not just the hash. +fn one_key(a: &[RunRef<'_, T>], b: &[RunRef<'_, T>]) -> bool { + let Some(first) = a.first().or_else(|| b.first()) else { return true }; + let (rk, ri) = (first.chunk.keys(), first.s); + a.iter().chain(b).all(|r| { + compare_at(r.chunk.keys(), r.s, rk, ri) == Ordering::Equal + && compare_at(r.chunk.keys(), r.e - 1, rk, ri) == Ordering::Equal + }) +} + +/// Stage a colliding identifier without allowing equal values from different real keys to +/// consolidate together. Each real key is staged independently, but all retain the identifier as +/// their proxy token; `cross` uses the coordinates to discard unequal-key pairs afterward. +fn stage_collision( + runs: &[RunRef<'_, T>], + lower: &T, + token: u64, + bridge: &mut ProxyBridge, +) { + let mut positions: Vec = runs.iter().map(|run| run.s).collect(); + let mut scratch = SideScratch::new(); + let mut staged = Vec::new(); + loop { + let Some(min) = positions + .iter() + .enumerate() + .filter(|(index, position)| **position < runs[*index].e) + .min_by(|(ai, ap), (bi, bp)| { + compare_at(runs[*ai].chunk.keys(), **ap, runs[*bi].chunk.keys(), **bp) + }) + .map(|(index, _)| index) + else { + break; + }; + let (reference, row) = (runs[min].chunk.keys(), positions[min]); + let mut equal_runs = Vec::new(); + for (index, run) in runs.iter().enumerate() { + let start = positions[index]; + if start == run.e + || compare_at(run.chunk.keys(), start, reference, row) != Ordering::Equal + { + continue; + } + let end = start + + (start..run.e) + .position(|candidate| { + compare_at(run.chunk.keys(), candidate, reference, row) != Ordering::Equal + }) + .unwrap_or(run.e - start); + let vals = run + .vals + .map(|(lanes, offset)| (lanes, offset + start - run.s)); + equal_runs.push(RunRef { chunk: run.chunk, cid: run.cid, s: start, e: end, vals }); + positions[index] = end; + } + scratch.stage_runs(&equal_runs, lower); + staged.extend(scratch.entries.drain(..).map(|(coord, time, diff)| ((token, coord), time, diff))); + } + staged.sort_by(|a, b| (a.0, &a.1).cmp(&(b.0, &b.1))); + bridge.extend(staged); +} + +/// Blockwise `advance` over every arrangement key shape: group token = the leading identifier +/// lane (the key itself or its carried hash), so blocks resume by seeking `from` and end at +/// identifier boundaries. /// /// Two regimes: when one side is much smaller (the fresh delta against an accumulated /// trace), the small side DRIVES and the large side is presented only at the driver's keys @@ -521,10 +585,14 @@ fn advance_leaf( from: &mut Option, bridge0: &mut ProxyBridge, bridge1: &mut ProxyBridge, + colliding: &mut Vec, ) { let start = from.expect("advance called on an exhausted unit"); + let hashed = key_is_hashed(chunks0[0].keys()); + // Resume: the first row of each chunk at or past `start`, by binary search on its identifier + // lane. The lane is a sorted `u64` slice, so this is a slice operation, not a column probe. let seek = |chunks: &[&CorgiChunk]| -> Vec { - chunks.iter().map(|c| if c.len() == 0 { 0 } else { find_ranges(&needle_like(c.keys(), &[start]), c.keys()).0[0] }).collect() + chunks.iter().map(|c| ident(c).partition_point(|&x| x < start)).collect() }; let start0 = seek(chunks0); let start1 = seek(chunks1); @@ -532,40 +600,30 @@ fn advance_leaf( chunks.iter().zip(starts).map(|(c, &s)| c.len() - s).sum() }; let (r0, r1) = (remaining(chunks0, &start0), remaining(chunks1, &start1)); - fn views<'a, T: ColTime>(chunks: &[&'a CorgiChunk], starts: &[usize]) -> Vec> { + fn views<'a, T: ColTime>(chunks: &[&'a CorgiChunk], starts: &[usize], ends: &[usize]) -> Vec> { let leaf_vals = leaf_valued(chunks); chunks.iter().enumerate() - .filter(|(_, c)| c.len() > 0) - .map(|(cid, c)| LeafView::new(c, cid, starts[cid], leaf_vals)) + .filter(|(cid, _)| ends[*cid] > starts[*cid]) + .map(|(cid, c)| LeafView::new(c, cid, starts[cid], ends[cid], leaf_vals)) .collect() } if r0.max(r1) >= 2 * r0.min(r1) { + // Lopsided: only the driver is read in bulk, so only the driver bounds the block. let drive0 = r0 <= r1; - let (dviews, pchunks) = if drive0 { (views(chunks0, &start0), chunks1) } else { (views(chunks1, &start1), chunks0) }; + let (dchunks, dstarts, pchunks) = if drive0 { (chunks0, &start0, chunks1) } else { (chunks1, &start1, chunks0) }; + let h = block_horizon(dchunks, dstarts, PULL); + let dviews = views(dchunks, dstarts, &block_ends(dchunks, h)); let (bd, bp) = if drive0 { (bridge0, bridge1) } else { (bridge1, bridge0) }; - leaf_probe(dviews, pchunks, lower, start, from, bd, bp); + leaf_probe(dviews, pchunks, lower, h, from, bd, bp, hashed, colliding) } else { - leaf_merge(views(chunks0, &start0), views(chunks1, &start1), lower, start, from, bridge0, bridge1); - } -} - -/// The pull horizon: keys at or beyond the least last-pulled key of a partially-pulled -/// view may have rows outside its buffer, so the block stops there. Extended pulls push -/// it out when it would deny all progress (a run longer than the pull). -fn pull_horizon<'a, T: ColTime>(views0: &mut [LeafView<'a, T>], views1: &mut [LeafView<'a, T>], start: u64) -> Option { - let horizon = |a: &[LeafView<'a, T>], b: &[LeafView<'a, T>]| -> Option { - a.iter().chain(b).filter(|v| !v.fully_pulled()).map(|v| *v.keys.last().unwrap()).min() - }; - let mut h = horizon(views0, views1); - while h == Some(start) { - for v in views0.iter_mut().chain(views1.iter_mut()) { - if !v.fully_pulled() && *v.keys.last().unwrap() == start { - v.extend_pull(v.keys.len()); - } - } - h = horizon(views0, views1); + // Symmetric: both sides are read, so both bound the block. + let h = block_horizon(chunks0, &start0, PULL).into_iter() + .chain(block_horizon(chunks1, &start1, PULL)) + .min(); + let views0 = views(chunks0, &start0, &block_ends(chunks0, h)); + let views1 = views(chunks1, &start1, &block_ends(chunks1, h)); + leaf_merge(views0, views1, lower, h, from, bridge0, bridge1, hashed, colliding) } - h } /// Lopsided regime: the driver views are walked; the probee is presented only at the @@ -574,13 +632,13 @@ fn leaf_probe<'a, T: ColTime>( mut dviews: Vec>, pchunks: &[&CorgiChunk], lower: &T, - start: u64, + h: Option, from: &mut Option, bridge_d: &mut ProxyBridge, bridge_p: &mut ProxyBridge, + hashed: bool, + colliding: &mut Vec, ) { - let h = pull_horizon(&mut dviews, &mut [], start); - // Merge the driver block's keys (strictly below the horizon) into the distinct key // list and each key's runs. let mut keyset: Vec = Vec::new(); @@ -606,7 +664,7 @@ fn leaf_probe<'a, T: ColTime>( let pvleaf = leaf_valued(pchunks); let probes: Vec> = pchunks.iter().enumerate() .filter(|(_, c)| c.len() > 0) - .map(|(cid, c)| Probe::new(c, cid, &needle_like(c.keys(), &keyset), pvleaf)) + .map(|(cid, c)| Probe::new(c, cid, &CValue::u64(keyset.clone()), pvleaf)) .collect(); // Walk the keys in order, staging both sides and emitting the survivors. @@ -625,7 +683,21 @@ fn leaf_probe<'a, T: ColTime>( if refs.len() == dref_count { continue; // key absent from the probee: no matches } + let collision = hashed && !one_key(&refs, &[]); + if collision { + colliding.push(k); + } let (drefs, prefs) = refs.split_at(dref_count); + if collision { + let (mut staged_d, mut staged_p) = (Vec::new(), Vec::new()); + stage_collision(drefs, lower, k, &mut staged_d); + stage_collision(prefs, lower, k, &mut staged_p); + if !staged_d.is_empty() && !staged_p.is_empty() { + bridge_d.extend(staged_d); + bridge_p.extend(staged_p); + } + continue; + } sd.stage_runs(drefs, lower); sp.stage_runs(prefs, lower); if sd.entries.is_empty() || sp.entries.is_empty() { @@ -645,12 +717,13 @@ fn leaf_merge<'a, T: ColTime>( mut views0: Vec>, mut views1: Vec>, lower: &T, - start: u64, + h: Option, from: &mut Option, bridge0: &mut ProxyBridge, bridge1: &mut ProxyBridge, + hashed: bool, + colliding: &mut Vec, ) { - let h = pull_horizon(&mut views0, &mut views1, start); let (mut s0, mut s1) = (SideScratch::new(), SideScratch::new()); let (mut refs0, mut refs1): (Vec<(usize, usize, usize)>, Vec<(usize, usize, usize)>) = (Vec::new(), Vec::new()); loop { @@ -668,6 +741,20 @@ fn leaf_merge<'a, T: ColTime>( } let r0: Vec> = refs0.iter().map(|&(vi, s, e)| views0[vi].run_ref(s, e)).collect(); let r1: Vec> = refs1.iter().map(|&(vi, s, e)| views1[vi].run_ref(s, e)).collect(); + let collision = hashed && !one_key(&r0, &r1); + if collision { + colliding.push(k); + } + if collision { + let (mut staged0, mut staged1) = (Vec::new(), Vec::new()); + stage_collision(&r0, lower, k, &mut staged0); + stage_collision(&r1, lower, k, &mut staged1); + if !staged0.is_empty() && !staged1.is_empty() { + bridge0.extend(staged0); + bridge1.extend(staged1); + } + continue; + } s0.stage_runs(&r0, lower); s1.stage_runs(&r1, lower); if s0.entries.is_empty() || s1.entries.is_empty() { @@ -678,240 +765,104 @@ fn leaf_merge<'a, T: ColTime>( } } -/// A fully-pulled view over one chunk for the LANE-TUPLE key path: every key lane (and, -/// when leaf-shaped, val lane) as `u64` buffers, so the key walk is lexicographic machine -/// compares — no per-row structural dispatch. `side` routes emission; tuples have no -/// order-preserving `u64` embedding, so this path runs as ONE block with ordinal tokens -/// (resumable blocking for tuples would need a digest scheme; the walk, not the blocking, -/// is what scales). -struct LaneView<'a, T: ColTime> { - chunk: &'a CorgiChunk, - cid: usize, - side: usize, - keys: Vec>, - vals: Option>>, - cur: usize, -} - -impl<'a, T: ColTime> LaneView<'a, T> { - fn new(chunk: &'a CorgiChunk, cid: usize, side: usize, leaf_vals: bool) -> Self { - let idx: Vec = (0..chunk.len()).collect(); - let keys = pull_lanes(chunk.keys(), &idx); - let vals = leaf_vals.then(|| pull_lanes(chunk.vals(), &idx)); - LaneView { chunk, cid, side, keys, vals, cur: 0 } - } - fn exhausted(&self) -> bool { - self.cur >= self.chunk.len() - } - fn run_ref(&self, s: usize, e: usize) -> RunRef<'_, T> { - RunRef { chunk: self.chunk, cid: self.cid, s, e, vals: self.vals.as_ref().map(|lanes| (&lanes[..], 0)) } - } -} - -/// Lexicographic compare of `views[a]`'s row `ai` against `views[b]`'s row `bi`. -fn lane_key_cmp(views: &[LaneView<'_, T>], a: usize, ai: usize, b: usize, bi: usize) -> Ordering { - for (la, lb) in views[a].keys.iter().zip(views[b].keys.iter()) { - match la[ai].cmp(&lb[bi]) { - Ordering::Equal => {} - other => return other, - } - } - Ordering::Equal -} - -/// One past the end of the run of rows equal to row `s` in `views[v]`. -fn lane_run_end(views: &[LaneView<'_, T>], v: usize, s: usize) -> usize { - let len = views[v].chunk.len(); - let mut e = s + 1; - while e < len && lane_key_cmp(views, v, e, v, s) == Ordering::Equal { - e += 1; - } - e -} - -/// `advance` for lane-tuple keys: the whole intersection in one call (ordinal group tokens, -/// ascending with key order), with the same two regimes as the single-lane path — a much -/// smaller side DRIVES and the other is probed at the driver's keys (`find_ranges` with -/// gathered tuple needles, so per-round cost tracks the delta); comparable sides merge -/// symmetrically on the pulled lane buffers. -fn advance_lanes( - chunks0: &[&CorgiChunk], - chunks1: &[&CorgiChunk], - lower: &T, - from: &mut Option, - bridge0: &mut ProxyBridge, - bridge1: &mut ProxyBridge, -) { - *from = None; // one block: tuples have no resumable `u64` embedding - let (r0, r1): (usize, usize) = (chunks0.iter().map(|c| c.len()).sum(), chunks1.iter().map(|c| c.len()).sum()); - if r0 == 0 || r1 == 0 { - return; - } - fn views_of<'a, T: ColTime>(chunks: &[&'a CorgiChunk], side: usize) -> Vec> { - let leaf_vals = leaf_valued(chunks); - chunks.iter().enumerate().filter(|(_, c)| c.len() > 0).map(|(cid, c)| LaneView::new(c, cid, side, leaf_vals)).collect() - } - - if r0.max(r1) >= 2 * r0.min(r1) { - // Lopsided: walk only the DRIVER's keys; probe the other side wholesale. - let drive0 = r0 <= r1; - let (dchunks, pchunks) = if drive0 { (chunks0, chunks1) } else { (chunks1, chunks0) }; - let mut dviews = views_of(dchunks, 0); - // Collect the driver's distinct keys (as gather coordinates for the needle column) - // and each key's runs. - let mut needle_tags: Vec = Vec::new(); - let mut needle_offs: Vec = Vec::new(); - let mut druns: Vec<(usize, usize, usize, usize)> = Vec::new(); // (key idx, view, s, e) - loop { - let mut min: Option = None; - for v in 0..dviews.len() { - if dviews[v].exhausted() { - continue; - } - min = Some(match min { - None => v, - Some(m) if lane_key_cmp(&dviews, v, dviews[v].cur, m, dviews[m].cur) == Ordering::Less => v, - Some(m) => m, - }); - } - let Some(m) = min else { break }; - let j = needle_tags.len(); - needle_tags.push(m); - needle_offs.push(dviews[m].cur); - let ends: Vec<(usize, usize, usize)> = (0..dviews.len()) - .filter(|&v| !dviews[v].exhausted() && lane_key_cmp(&dviews, v, dviews[v].cur, m, dviews[m].cur) == Ordering::Equal) - .map(|v| (v, dviews[v].cur, lane_run_end(&dviews, v, dviews[v].cur))) - .collect(); - for (v, ss, e) in ends { - druns.push((j, v, ss, e)); - dviews[v].cur = e; - } - } - if needle_tags.is_empty() { - return; - } - // One tuple-shaped needle column, one batched probe per probee chunk. - let key_srcs: Vec> = dviews.iter().map(|v| Some(v.chunk.keys())).collect(); - let needles = gather_lanes(&key_srcs, &needle_tags, &needle_offs); - let pvleaf = leaf_valued(pchunks); - let probes: Vec> = pchunks.iter().enumerate() - .filter(|(_, c)| c.len() > 0) - .map(|(cid, c)| Probe::new(c, cid, &needles, pvleaf)) - .collect(); - let (mut sd, mut sp) = (SideScratch::new(), SideScratch::new()); - let (bd, bp) = if drive0 { (bridge0, bridge1) } else { (bridge1, bridge0) }; - let mut drun_at = 0usize; - let mut refs: Vec> = Vec::new(); - let mut token = 0u64; - for j in 0..needle_tags.len() { - refs.clear(); - while drun_at < druns.len() && druns[drun_at].0 == j { - let (_, v, ss, e) = druns[drun_at]; - refs.push(dviews[v].run_ref(ss, e)); - drun_at += 1; - } - let dref_count = refs.len(); - refs.extend(probes.iter().filter_map(|p| p.run_ref(j))); - if refs.len() == dref_count { - continue; - } - let (drefs, prefs) = refs.split_at(dref_count); - sd.stage_runs(drefs, lower); - sp.stage_runs(prefs, lower); - if sd.entries.is_empty() || sp.entries.is_empty() { - continue; - } - sd.emit(token, bd); - sp.emit(token, bp); - token += 1; - } - } else { - // Comparable sides: one tagged view set, symmetric lexicographic merge. - let mut views = views_of(chunks0, 0); - views.extend(views_of(chunks1, 1)); - let (mut s0, mut s1) = (SideScratch::new(), SideScratch::new()); - let mut token = 0u64; - loop { - let mut min: Option = None; - for v in 0..views.len() { - if views[v].exhausted() { - continue; +#[cfg(test)] +mod tests { + use super::*; + use differential_dataflow::trace::Description; + use timely::progress::Antichain; + + fn batch(rows: &[(u64, u64, u64)]) -> CBatch { + let keys = CValue::Prod(vec![ + CValue::u64(rows.iter().map(|row| row.0).collect()), + CValue::Prod(vec![ + CValue::u64(rows.iter().map(|row| row.1).collect()), + CValue::u64(rows.iter().map(|row| row.2).collect()), + ]), + ]); + // Deliberately equal across different real keys: collision staging must not consolidate + // these together before `cross` has a chance to compare their keys. + let vals = CValue::u64(vec![0; rows.len()]); + let chunk = CorgiChunk::from_columns(keys, vals, vec![0; rows.len()], vec![1; rows.len()]); + Rc::new(ChunkBatch::new( + vec![chunk], + Description::new( + Antichain::from_elem(0), + Antichain::from_elem(1), + Antichain::from_elem(0), + ), + )) + } + + fn backend() -> CorgiJoinBackend { + CorgiJoinBackend::new( + Term::Var(0), + Term::Tuple(vec![Term::Var(1), Term::Var(2)]), + ) + } + + fn cross_bridges( + backend: &mut CorgiJoinBackend, + instance: &JoinInstance, CBatch>, + left: &ProxyBridge, + right: &ProxyBridge, + ) -> usize { + let mut matches = JoinMatches::default(); + for a in left { + for b in right { + if a.0.0 == b.0.0 { + matches.ids.push((a.0.0, (a.0.1, b.0.1))); + matches.times.push(a.1.max(b.1)); + matches.diffs.push(a.2 * b.2); } - min = Some(match min { - None => v, - Some(m) if lane_key_cmp(&views, v, views[v].cur, m, views[m].cur) == Ordering::Less => v, - Some(m) => m, - }); } - let Some(m) = min else { break }; - let ends: Vec<(usize, usize, usize)> = (0..views.len()) - .filter(|&v| !views[v].exhausted() && lane_key_cmp(&views, v, views[v].cur, m, views[m].cur) == Ordering::Equal) - .map(|v| (v, views[v].cur, lane_run_end(&views, v, views[v].cur))) - .collect(); - let both_sides = { - // Reads only: the run refs borrow `views` and end with this block. - let mut refs0: Vec> = Vec::new(); - let mut refs1: Vec> = Vec::new(); - for &(v, ss, e) in &ends { - let r = views[v].run_ref(ss, e); - if views[v].side == 0 { refs0.push(r) } else { refs1.push(r) } - } - let both = !refs0.is_empty() && !refs1.is_empty(); - if both { - s0.stage_runs(&refs0, lower); - s1.stage_runs(&refs1, lower); - } - both - }; - for (v, _, e) in ends { - views[v].cur = e; - } - if !both_sides { - continue; - } - if s0.entries.is_empty() || s1.entries.is_empty() { - continue; - } - s0.emit(token, bridge0); - s1.emit(token, bridge1); - token += 1; } - } -} + let mut output = Vec::new(); + backend.cross(instance, &mut matches, &mut output); + output.iter().map(|container| container.diffs.len()).sum() + } + + #[test] + fn collision_after_completed_block_filters_real_keys_without_restarting() { + let collision = PULL as u64 + 1; + let mut rows: Vec<_> = (0..collision).map(|id| (id, id, 0)).collect(); + rows.extend([(collision, 7, 0), (collision, 8, 0)]); + let instance = JoinInstance { + batches0: vec![batch(&rows)], + batches1: vec![batch(&rows)], + lower: 0, + }; + let mut backend = backend(); + let mut from = Some(0); + let (mut left, mut right) = (Vec::new(), Vec::new()); + + backend.advance(&instance, &mut from, &mut left, &mut right); + assert!(from.is_some(), "the first block must leave work for the collision block"); + assert!(backend.colliding.is_empty()); + + left.clear(); + right.clear(); + backend.advance(&instance, &mut from, &mut left, &mut right); + assert_eq!(backend.colliding, vec![collision]); + assert_eq!(cross_bridges(&mut backend, &instance, &left, &right), 2); + } + + #[test] + fn lopsided_collision_filters_real_keys() { + let collision = 42; + let left_rows = [(collision, 7, 0), (collision, 8, 0)]; + let mut right_rows: Vec<_> = (0..8).map(|id| (id, id, 0)).collect(); + right_rows.extend([(collision, 7, 0), (collision, 8, 0)]); + let instance = JoinInstance { + batches0: vec![batch(&left_rows)], + batches1: vec![batch(&right_rows)], + lower: 0, + }; + let mut backend = backend(); + let mut from = Some(0); + let (mut left, mut right) = (Vec::new(), Vec::new()); -/// Last-resort `advance` for keys that do not flatten to integer lanes at all (sums or -/// lists in the key): the whole intersection in one block, ordinal tokens, and a structural -/// (`compare_at`) walk. Rare by construction — tuple keys take [`advance_lanes`]. -fn advance_structured( - chunks0: &[&CorgiChunk], - chunks1: &[&CorgiChunk], - lower: &T, - from: &mut Option, - bridge0: &mut ProxyBridge, - bridge1: &mut ProxyBridge, -) { - let mut pos0 = vec![0usize; chunks0.len()]; - let mut pos1 = vec![0usize; chunks1.len()]; - let (mut s0, mut s1) = (SideScratch::new(), SideScratch::new()); - let (mut runs0, mut runs1) = (Vec::new(), Vec::new()); - let mut token = 0u64; - while let Some(cand) = min_key(chunks0, &pos0, chunks1, &pos1) { - take_runs(chunks0, &mut pos0, cand, &mut runs0); - take_runs(chunks1, &mut pos1, cand, &mut runs1); - if runs0.is_empty() || runs1.is_empty() { - continue; - } - fn refs<'a, T: ColTime>(chunks: &[&'a CorgiChunk], runs: &[(usize, usize, usize)]) -> Vec> { - runs.iter().map(|&(c, s, e)| RunRef { chunk: chunks[c], cid: c, s, e, vals: None }).collect() - } - s0.stage_runs(&refs(chunks0, &runs0), lower); - s1.stage_runs(&refs(chunks1, &runs1), lower); - if s0.entries.is_empty() || s1.entries.is_empty() { - continue; - } - s0.emit(token, bridge0); - s1.emit(token, bridge1); - token += 1; + backend.advance(&instance, &mut from, &mut left, &mut right); + assert_eq!(backend.colliding, vec![collision]); + assert_eq!(cross_bridges(&mut backend, &instance, &left, &right), 2); } - *from = None; } diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs index c53d882ab..eb44620fc 100644 --- a/interactive/src/corgi/reduce.rs +++ b/interactive/src/corgi/reduce.rs @@ -38,11 +38,11 @@ use differential_dataflow::trace::chunk::ChunkBatch; use differential_dataflow::operators::int_proxy::ProxyBridge; use differential_dataflow::operators::int_proxy::reduce::{ProxyReduceBackend, ReduceInstance, ReduceWindow}; -use corgi::arrange::{find_ranges, gather, gather_lanes, sort_blocks}; -use corgi::{Bounds, Shape, Value as CValue}; +use corgi::arrange::{compare_at, find_ranges, gather, gather_lanes, sort_blocks}; +use corgi::{Bounds, Value as CValue}; use crate::corgi::col_times::ColTime; -use crate::corgi::chunk::{columns_to_batch, CorgiChunk}; +use crate::corgi::chunk::{columns_to_batch, key_ids, key_lane, CorgiChunk}; use crate::ir::Diff; use crate::parse::Reducer; @@ -158,7 +158,7 @@ fn concat_columns(blocks: &[CValue]) -> CValue { } } -/// Id column for a key/value column. For a PRIMITIVE column — a bare 64-bit `Prim`, or a 1-field +/// Id column for a VALUE column. For a PRIMITIVE column — a bare 64-bit `Prim`, or a 1-field /// `Prod([Prim(64)])` — the value itself is already a collision-free id (`i64 as u64` is a bijection), /// so pass it straight through and skip the content hash. Compound shapes (Unit / List / Sum / /// multi-field `Prod`) hash via the CANONICAL native `corgi::hash` (the designed boundary-id fold, @@ -166,7 +166,7 @@ fn concat_columns(blocks: &[CValue]) -> CValue { /// transcodes every leaf to `u64`, so width-blindness is a no-op for us and there is no cross-path /// hash comparison (value-as-id and native hash are never used for the same value: shape is uniform /// per column). Compound ids are used only for identity, but the leaf fast path additionally relies -/// on raw-id order matching corgi's unsigned leaf order: `seek_needles` searches in that order and +/// on raw-id order matching corgi's unsigned leaf order: the stored key lane is searched in that order and /// `merge_present` merges chunk runs in it. Raw two's-complement `u64` therefore remains correct for /// negative ints (no swizzle); changing the leaf encoding must also revisit those ordered paths. /// Applied CONSISTENTLY at every id site (both value presentations AND the freshly-produced @@ -181,43 +181,30 @@ fn ids(col: &CValue) -> Vec { corgi::hash(col).into_u64("ids") } -/// The `changed` set as a needle column in the chunks' own key shape — possible exactly -/// when `ids` uses key VALUES (a bare `u64` leaf, or a 1-tuple of one); the hashed ids of -/// structural keys cannot be inverted into needles. -fn seek_needles(sample: &CValue, changed: &[u64]) -> Option { - match corgi::shape_of_value(sample) { - Shape::Prim(64) => Some(CValue::u64(changed.to_vec())), - Shape::Prod(ref fs) if fs.len() == 1 && matches!(fs[0], Shape::Prim(64)) => { - Some(CValue::Prod(vec![CValue::u64(changed.to_vec())])) - } - _ => None, - } -} - /// Concatenate the records of the `changed` keys across a run of chunks into parallel /// `(keys_col, vals_col)` corgi columns plus per-record `(key_hash, time, diff)`. `changed` is the -/// ASCENDING set of changed key hashes; a row is kept iff its key hash is in it. +/// ASCENDING set of changed key ids; a row is kept iff its key id is in it. /// -/// Seek-vs-scan, decided per retire, now that the sizes are known: seeking the changed keys -/// (`find_ranges`, O(|changed|·log rows) per chunk, no key hashing at all) wins when the -/// changed set is narrow — the steady incremental case; the full scan (O(rows) per chunk, -/// plus each chunk's key hashes re-derived) wins for broad churn — loads and label-cascade -/// retires, where most keys change and a gallop per key only adds overhead. Seeking requires -/// ids that ARE key values (single-leaf keys, `ids`' fast paths): hashed ids of structural -/// keys cannot be inverted into needles, so those always scan. +/// Seek or scan, decided per retire now that the sizes are known: seeking the changed keys wins +/// when the changed set is narrow (the steady incremental case), and a flat membership scan wins +/// for broad churn — loads and label-cascade retires, where most keys change and a gallop per key +/// only adds overhead. /// -/// TODO: the scan's per-row work can still batch: `ids` re-derives (and copies) each chunk's -/// key hashes every retire (memoize per chunk, or a stored hash column), and each hit -/// materializes an owned time (`times().get`); kept RANGES could move via `push_range`. +/// What the stored identifier changed is that BOTH branches are now available, and cheap, for every +/// key shape. An arrangement's key leads with its identifier and is sorted by it +/// ([`present_key`](crate::corgi::chunk::present_key)), so [`key_lane`] is a sorted `u64` leaf: the +/// seek is `find_ranges` over it (corgi's `u64` fast path) and the scan borrows it outright, with no +/// hashing and no allocation on either side. Previously a structural key could do neither — its +/// identifier was hashed per chunk per retire, and a hash derived on the fly cannot be inverted into +/// a needle, so those keys were forced onto the scan and forced to materialize to take it. fn collect_present(chunks: &[&CorgiChunk], changed: &[u64]) -> (CValue, CValue, Vec, Vec, Vec, Vec) where T: ColTime, { - /// Seek only when the changed set is at least this many times narrower than the - /// presented rows: a `find_ranges` probe is a structurally-dispatched binary search - /// (~log(rows) compares, each far costlier than the scan's flat membership test), so - /// marginal seeks LOSE to the scan — measured, not modeled; 16 regressed load-shaped - /// retires before this was widened. + /// Seek only when the changed set is at least this many times narrower than the presented + /// rows: a `find_ranges` probe costs ~log(rows) compares against the scan's flat membership + /// test, so marginal seeks LOSE — measured, not modeled; 16 regressed load-shaped retires + /// before this was widened. const SEEK_ADVANTAGE: usize = 64; let key_srcs: Vec> = chunks.iter().map(|c| Some(c.keys())).collect(); @@ -226,20 +213,17 @@ where let (mut khs, mut times, mut diffs) = (Vec::new(), Vec::new(), Vec::new()); let mut run_ends = Vec::new(); let total: usize = chunks.iter().map(|c| c.diffs().len()).sum(); - let needles = if changed.len().saturating_mul(SEEK_ADVANTAGE) < total { - chunks.iter().find(|c| c.diffs().len() > 0).and_then(|c| seek_needles(c.keys(), changed)) - } else { - None - }; - if let Some(needles) = needles { - // Narrow changed set over seekable keys: gallop each chunk once per changed key. - // Chunks are key-ordered and `changed` ascends, so emission order matches the scan's. - for (ci, ch) in chunks.iter().enumerate() { - let before = khs.len(); - if ch.diffs().is_empty() { - continue; - } - let (lo, hi) = find_ranges(&needles, ch.keys()); + let seek = changed.len().saturating_mul(SEEK_ADVANTAGE) < total; + let needles = CValue::u64(changed.to_vec()); + // Chunks are id-ordered and `changed` ascends, so either branch emits in merged order. + for (ci, ch) in chunks.iter().enumerate() { + let before = khs.len(); + if ch.diffs().is_empty() { + continue; + } + let lane = key_lane(ch.keys()); + if seek { + let (lo, hi) = find_ranges(&needles, lane); for (j, (&l, &h)) in lo.iter().zip(hi.iter()).enumerate() { for i in l..h { tags.push(ci); @@ -249,21 +233,9 @@ where diffs.push(ch.diffs()[i]); } } - if khs.len() > before { run_ends.push(khs.len()); } - } - } else { - for (ci, ch) in chunks.iter().enumerate() { - let before = khs.len(); - // 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"), - }; + } else { + // Borrowed, never materialized: the identifier lane is a `u64` leaf whatever the key. + let kh = corgi::arrange::leaf_slice(lane).expect("the identifier lane is a u64 leaf"); for i in 0..kh.len() { if changed.binary_search(&kh[i]).is_ok() { tags.push(ci); @@ -273,8 +245,8 @@ where diffs.push(ch.diffs()[i]); } } - if khs.len() > before { run_ends.push(khs.len()); } } + if khs.len() > before { run_ends.push(khs.len()); } } if tags.is_empty() { return (CValue::Unit(0), CValue::Unit(0), khs, times, diffs, run_ends); @@ -284,17 +256,29 @@ where (keys_col, vals_col, khs, times, diffs, run_ends) } -/// Merge already-ordered selected chunk runs directly into an empty proxy bridge. Identity-id leaf -/// columns preserve the chunks' structural order; a debug assertion audits the resulting -/// `(key_id, value_id, time)` order against that inference. Returns false for structural columns or -/// a nonempty bridge, so the caller can append and consolidate the whole bridge normally. +/// Merge already-ordered selected chunk runs directly into an empty proxy bridge. Leaf values +/// preserve value-id order. Keys may either be identity-id leaves or carried-hash columns, provided +/// no one chunk run contains two real keys under the same hash; in the latter case the real-key +/// tie-break would interrupt proxy `(key_id, value_id, time)` order, so we fall back to ordinary +/// consolidation. A debug assertion audits the inferred order. Returns false when the inference +/// does not hold or the bridge is nonempty. fn merge_present( keys_col: &CValue, vals_col: &CValue, khs: &[u64], vids: &[u64], times: &[T], diffs: &[Diff], run_ends: &[usize], bridge: &mut ProxyBridge, ) -> bool { - let ordered_ids = corgi::arrange::leaf_slice(keys_col).is_some() - && corgi::arrange::leaf_slice(vals_col).is_some(); + let ordered_keys = corgi::arrange::leaf_slice(keys_col).is_some() || { + let mut start = 0usize; + run_ends.iter().all(|&end| { + let one_real_key_per_id = (start + 1..end).all(|index| { + khs[index - 1] != khs[index] + || compare_at(keys_col, index - 1, keys_col, index) == std::cmp::Ordering::Equal + }); + start = end; + one_real_key_per_id + }) && start == khs.len() + }; + let ordered_ids = ordered_keys && corgi::arrange::leaf_slice(vals_col).is_some(); if !ordered_ids || !bridge.is_empty() { return false; } @@ -577,7 +561,7 @@ where // key hashes come from the scan the key list needs anyway. let mut seeds: Vec<(u64, T)> = Vec::new(); for ch in novel_chunks.iter() { - let khs = ids(ch.keys()); + let khs = key_ids(ch.keys()); let times = ch.times(); for (i, kh) in khs.iter().enumerate() { seeds.push((*kh, times.get(i))); @@ -698,3 +682,39 @@ where }).collect() } } + +#[cfg(test)] +mod tests { + use super::*; + + fn compound_keys(hashes: Vec, real: Vec) -> CValue { + CValue::Prod(vec![CValue::u64(hashes), CValue::Prod(vec![CValue::u64(real), CValue::u64(vec![0, 0])])]) + } + + #[test] + fn merge_present_accepts_ordered_compound_keys() { + let keys = compound_keys(vec![1, 2], vec![7, 8]); + let vals = CValue::u64(vec![10, 20]); + let mut bridge = Vec::new(); + assert!(merge_present( + &keys, &vals, &[1, 2], &[10, 20], &[0u64, 0], &[1, 1], &[2], &mut bridge, + )); + assert_eq!(bridge.len(), 2); + } + + #[test] + fn merge_present_rejects_a_compound_hash_collision_within_a_run() { + let keys = compound_keys(vec![1, 1], vec![7, 8]); + let vals = CValue::u64(vec![10, 20]); + assert!(!merge_present( + &keys, + &vals, + &[1, 1], + &[10, 20], + &[0u64, 0], + &[1, 1], + &[2], + &mut Vec::new(), + )); + } +}