diff --git a/interactive/src/corgi/chunk.rs b/interactive/src/corgi/chunk.rs index bc727e7e7..fb7c7e6ba 100644 --- a/interactive/src/corgi/chunk.rs +++ b/interactive/src/corgi/chunk.rs @@ -283,13 +283,27 @@ where keep: &mut VecDeque, ship: &mut VecDeque, ) { - // One input chunk per call: partition into keep (`>= frontier`) and ship pieces via `gather`. + // Move whole chunks; only a split needs to gather its two payload pieces. let Some(chunk) = input.pop_front() else { return }; + let times = chunk.times(); + if times.is_empty() { return; } + let beyond = times.beyond(frontier); + if beyond.iter().all(|b| !b) { + ship.push_back(chunk); + return; + } + if beyond.iter().all(|b| *b) { + let covered = times.beyond(residual.borrow()); + let fresh: Vec = (0..times.len()).filter(|&i| !covered[i]).collect(); + for i in times.minimal(&fresh) { residual.insert(times.get(i)); } + keep.push_back(chunk); + return; + } let kv = chunk.kv(); - let (times, diffs) = (chunk.times(), chunk.diffs()); + let diffs = chunk.diffs(); let (mut ki, mut si) = (Vec::new(), Vec::new()); - for (i, beyond) in times.beyond(frontier).into_iter().enumerate() { - if beyond { ki.push(i); } else { si.push(i); } + for (i, is_beyond) in beyond.into_iter().enumerate() { + if is_beyond { ki.push(i); } else { si.push(i); } } if !ki.is_empty() { // Only kept rows the residual does not already cover can change it; timestamps are @@ -962,6 +976,52 @@ mod test { } } + #[test] + fn extract_moves_whole_structured_chunks() { + let source = CorgiChunk::from_columns(ints(vec![0, 0, 1, 1]), + CValue::List(corgi::Bounds::Stride(1, 4), Box::new(ints(vec![7, 7, 8, 8]))), + vec![0u64, 1, 2, 3].into_iter().collect(), vec![1i64, -1, 2, -3]); + for frontier in [0, 2, 4] { + for initial in [0, 5] { + let mut input = VecDeque::from([source.clone()]); + let (mut keep, mut ship) = (VecDeque::new(), VecDeque::new()); + let mut residual = Antichain::from_elem(initial); + let mut expected = residual.clone(); + for row in 0..source.len_() { + if source.times().get(row) >= frontier { expected.insert(source.times().get(row)); } + } + CorgiChunk::extract(&mut input, AntichainRef::new(&[frontier]), + &mut residual, &mut keep, &mut ship); + assert!(input.is_empty()); + assert_eq!(residual, expected); + for (output, kept) in [(&keep, true), (&ship, false)] { + let selected: Vec<_> = (0..source.len_()) + .filter(|&row| (source.times().get(row) >= frontier) == kept).collect(); + assert_eq!(output.len(), usize::from(!selected.is_empty())); + if let Some(chunk) = output.front() { + assert_eq!(chunk.len_(), selected.len()); + for (row, &original) in selected.iter().enumerate() { + assert_eq!(compare_at(&chunk.kv(), row, &source.kv(), original), Ordering::Equal); + assert_eq!(chunk.times().get(row), source.times().get(original)); + assert_eq!(chunk.diffs()[row], source.diffs()[original]); + } + if selected.len() == source.len_() { + assert!(Rc::ptr_eq(&source.0, &chunk.0)); + } else { + assert!(!Rc::ptr_eq(&source.0, &chunk.0)); + } + } + } + } + } + let mut input = VecDeque::from([CorgiChunk::::default()]); + let (mut keep, mut ship) = (VecDeque::new(), VecDeque::new()); + let mut residual = Antichain::from_elem(5); + CorgiChunk::extract(&mut input, AntichainRef::new(&[0]), &mut residual, &mut keep, &mut ship); + assert!(input.is_empty() && keep.is_empty() && ship.is_empty()); + assert_eq!(residual, Antichain::from_elem(5)); + } + /// `extract` must split rows at the frontier and extend the residual with exactly the kept /// times, whether a chunk is kept whole, shipped whole, or split. #[test]