diff --git a/interactive/src/corgi/chunk.rs b/interactive/src/corgi/chunk.rs index 784aab4c9..738d30ce5 100644 --- a/interactive/src/corgi/chunk.rs +++ b/interactive/src/corgi/chunk.rs @@ -215,6 +215,14 @@ where } } } + // Equal (key, val) classes can continue in the next chunk. Once either + // whole input chunk is spent, its last timestamp is the shared horizon: + // retain the other side's suffix, including the rest of this class. + if i == n1 || j == n2 { + p1 = i; + p2 = j; + break; + } if i < a_hi { copy(&mut tags, &mut offs, &mut times, &mut diffs, 0, i, a_hi); } if j < b_hi { copy(&mut tags, &mut offs, &mut times, &mut diffs, 1, j, b_hi); } } diff --git a/interactive/tests/corgi_chunk_merge.rs b/interactive/tests/corgi_chunk_merge.rs new file mode 100644 index 000000000..7e5abcfe3 --- /dev/null +++ b/interactive/tests/corgi_chunk_merge.rs @@ -0,0 +1,37 @@ +//! Regression coverage for the sorted/consolidated chunk-chain merge contract. +use corgi::Value; +use differential_dataflow::batcher::merge::Merger; +use differential_dataflow::trace::chunk::{Chunk, ChunkMerger}; +use interactive::corgi::chunk::CorgiChunk; + +fn chunk(times: &[u64]) -> CorgiChunk { + CorgiChunk::from_columns( + Value::u64(vec![7; times.len()]), + Value::u64(vec![8; times.len()]), + times.to_vec(), + vec![1; times.len()], + ) +} + +fn check_horizon(prefix: usize) { + let n = prefix as u64; + let left = vec![chunk(&(0..n).collect::>()), chunk(&[n + 1])]; + let right = vec![chunk(&[n, n + 2])]; + let mut output = Vec::new(); + ChunkMerger::default().merge(left, right, &mut output, &mut Vec::new()); + let times: Vec<_> = output.iter().flat_map(|c| (0..c.times().len()).map(|i| c.times().get(i))).collect(); + assert_eq!(times.len(), prefix + 3); + assert_eq!(×[prefix - 1..], [n - 1, n, n + 1, n + 2]); + assert!(times.windows(2).all(|pair| pair[0] < pair[1])); +} + +#[test] +fn merge_keeps_time_order_when_an_equal_value_class_crosses_a_chunk_boundary() { + check_horizon(1); +} + +#[test] +#[ignore = "scale confirmation with fully graded inputs; the small case tests the same merge contract"] +fn merge_horizon_with_graded_input_chains() { + check_horizon(CorgiChunk::::TARGET); +}