|
20 | 20 | //! on equal prefixes, and emits the sorted, consolidated chain in one copy pass. The first chunk |
21 | 21 | //! carrying a second time hands everything held to a [`MergeBatcher`], and the batcher stays on |
22 | 22 | //! that path from then on, so steady-state behaviour is the merge batcher's. |
| 23 | +//! |
| 24 | +//! [`UnsortedChunker`] pairs with it: chunks arrive in arrival order, since sorting them ahead |
| 25 | +//! of a batcher that sorts everything at `seal` is wasted, and the batcher sorts and |
| 26 | +//! consolidates each chunk itself before handing it to the merge batcher on the fallback path. |
23 | 27 |
|
24 | 28 | use std::cmp::Ordering; |
| 29 | +use std::collections::VecDeque; |
25 | 30 |
|
| 31 | +use columnar::{Columnar, Index, Len}; |
26 | 32 | use columnation::Columnation; |
27 | 33 | use differential_dataflow::difference::Semigroup; |
28 | 34 | use differential_dataflow::lattice::Lattice; |
29 | 35 | use differential_dataflow::logging::Logger; |
30 | 36 | use differential_dataflow::trace::implementations::merge_batcher::MergeBatcher; |
31 | 37 | use differential_dataflow::trace::{Batcher, Description}; |
32 | 38 | use mz_repr::Row; |
| 39 | +use mz_timely_util::columnar::Column; |
33 | 40 | use mz_timely_util::columnation::{ColInternalMerger, ColumnationStack}; |
34 | | -use timely::container::PushInto; |
| 41 | +use timely::container::{ContainerBuilder, PushInto}; |
35 | 42 | use timely::progress::Timestamp; |
36 | 43 | use timely::progress::frontier::{Antichain, AntichainRef}; |
37 | 44 |
|
@@ -103,6 +110,35 @@ where |
103 | 110 | } |
104 | 111 | } |
105 | 112 |
|
| 113 | + /// Sort and consolidate one chunk, for handing to the general path, which requires it. |
| 114 | + fn sort_chunk(chunk: Chunk<D, T, R>) -> Chunk<D, T, R> { |
| 115 | + let mut index: Vec<usize> = (0..chunk.len()).collect(); |
| 116 | + index.sort_unstable_by(|&a, &b| { |
| 117 | + let (da, ta, _) = &chunk[a]; |
| 118 | + let (db, tb, _) = &chunk[b]; |
| 119 | + (da, ta).cmp(&(db, tb)) |
| 120 | + }); |
| 121 | + let mut out: Chunk<D, T, R> = ColumnationStack::with_capacity(chunk.len()); |
| 122 | + let mut iter = index.iter().peekable(); |
| 123 | + while let Some(&i) = iter.next() { |
| 124 | + let (d, t, r) = &chunk[i]; |
| 125 | + let mut diff = r.clone(); |
| 126 | + while let Some(&&j) = iter.peek() { |
| 127 | + let (d2, t2, r2) = &chunk[j]; |
| 128 | + if (d2, t2).cmp(&(d, t)) == Ordering::Equal { |
| 129 | + diff.plus_equals(r2); |
| 130 | + iter.next(); |
| 131 | + } else { |
| 132 | + break; |
| 133 | + } |
| 134 | + } |
| 135 | + if !diff.is_zero() { |
| 136 | + out.copy_destructured(d, t, &diff); |
| 137 | + } |
| 138 | + } |
| 139 | + out |
| 140 | + } |
| 141 | + |
106 | 142 | /// Sort and consolidate everything pending into a chain of chunks, in one copy pass. |
107 | 143 | fn sort_pending(&mut self) -> Vec<Chunk<D, T, R>> { |
108 | 144 | let mut index: Vec<(u64, u32, u32)> = Vec::with_capacity(self.pending_len); |
@@ -243,15 +279,132 @@ where |
243 | 279 | self.pending.push(chunk); |
244 | 280 | return; |
245 | 281 | } |
246 | | - // A second time: hand everything held to the general path, for good. |
| 282 | + // A second time: hand everything held to the general path, for good. The merge |
| 283 | + // batcher needs sorted, consolidated chunks, which the chunker did not provide. |
247 | 284 | self.general = true; |
248 | 285 | self.time = None; |
249 | 286 | self.pending_len = 0; |
250 | 287 | for held in self.pending.drain(..) { |
251 | | - self.inner.push_into(held); |
| 288 | + self.inner.push_into(Self::sort_chunk(held)); |
| 289 | + } |
| 290 | + } |
| 291 | + self.inner.push_into(Self::sort_chunk(chunk)); |
| 292 | + } |
| 293 | +} |
| 294 | + |
| 295 | +/// A chunker that packs incoming updates into [`ColumnationStack`] chunks in arrival order. |
| 296 | +/// |
| 297 | +/// The sorting chunkers sort and consolidate every input container before the batcher sees |
| 298 | +/// it. A [`SnapshotBatcher`] sorts everything it holds at `seal`, so that work is wasted on |
| 299 | +/// its fast path; on the fallback path it sorts the held chunks itself (see |
| 300 | +/// [`SnapshotBatcher::sort_chunk`]). Only pair this chunker with that batcher. |
| 301 | +pub struct UnsortedChunker<D, T, R> |
| 302 | +where |
| 303 | + D: Columnation, |
| 304 | + T: Columnation, |
| 305 | + R: Columnation, |
| 306 | +{ |
| 307 | + pending: Vec<(D, T, R)>, |
| 308 | + ready: VecDeque<ColumnationStack<(D, T, R)>>, |
| 309 | + empty: Option<ColumnationStack<(D, T, R)>>, |
| 310 | +} |
| 311 | + |
| 312 | +impl<D, T, R> Default for UnsortedChunker<D, T, R> |
| 313 | +where |
| 314 | + D: Columnation, |
| 315 | + T: Columnation, |
| 316 | + R: Columnation, |
| 317 | +{ |
| 318 | + fn default() -> Self { |
| 319 | + Self { |
| 320 | + pending: Vec::new(), |
| 321 | + ready: VecDeque::new(), |
| 322 | + empty: None, |
| 323 | + } |
| 324 | + } |
| 325 | +} |
| 326 | + |
| 327 | +impl<D, T, R> UnsortedChunker<D, T, R> |
| 328 | +where |
| 329 | + D: Columnation, |
| 330 | + T: Columnation, |
| 331 | + R: Columnation, |
| 332 | +{ |
| 333 | + /// Records per chunk, matching the merge batcher's 64 KiB chunks. |
| 334 | + fn chunk_capacity() -> usize { |
| 335 | + const BUFFER_SIZE_BYTES: usize = 64 << 10; |
| 336 | + let size = std::mem::size_of::<(D, T, R)>(); |
| 337 | + if size == 0 { |
| 338 | + BUFFER_SIZE_BYTES |
| 339 | + } else if size <= BUFFER_SIZE_BYTES { |
| 340 | + BUFFER_SIZE_BYTES / size |
| 341 | + } else { |
| 342 | + 1 |
| 343 | + } |
| 344 | + } |
| 345 | + |
| 346 | + fn form_chunks(&mut self, all: bool) { |
| 347 | + let cap = Self::chunk_capacity(); |
| 348 | + while self.pending.len() >= cap || (all && !self.pending.is_empty()) { |
| 349 | + let take = std::cmp::min(self.pending.len(), cap); |
| 350 | + let mut chunk = ColumnationStack::with_capacity(cap); |
| 351 | + for item in self.pending.drain(..take) { |
| 352 | + chunk.copy(&item); |
252 | 353 | } |
| 354 | + self.ready.push_back(chunk); |
| 355 | + } |
| 356 | + } |
| 357 | +} |
| 358 | + |
| 359 | +impl<'a, D, T, R> PushInto<&'a mut Vec<(D, T, R)>> for UnsortedChunker<D, T, R> |
| 360 | +where |
| 361 | + D: Columnation, |
| 362 | + T: Columnation, |
| 363 | + R: Columnation, |
| 364 | +{ |
| 365 | + fn push_into(&mut self, container: &'a mut Vec<(D, T, R)>) { |
| 366 | + self.pending.append(container); |
| 367 | + self.form_chunks(false); |
| 368 | + } |
| 369 | +} |
| 370 | + |
| 371 | +impl<'a, D, T, R> PushInto<&'a mut Column<(D, T, R)>> for UnsortedChunker<D, T, R> |
| 372 | +where |
| 373 | + D: Columnar + Columnation, |
| 374 | + T: Columnar + Columnation, |
| 375 | + R: Columnar + Columnation, |
| 376 | +{ |
| 377 | + fn push_into(&mut self, container: &'a mut Column<(D, T, R)>) { |
| 378 | + let borrowed = container.borrow(); |
| 379 | + self.pending.reserve(borrowed.len()); |
| 380 | + for (d, t, r) in borrowed.into_index_iter() { |
| 381 | + self.pending |
| 382 | + .push((D::into_owned(d), T::into_owned(t), R::into_owned(r))); |
253 | 383 | } |
254 | | - self.inner.push_into(chunk); |
| 384 | + self.form_chunks(false); |
| 385 | + } |
| 386 | +} |
| 387 | + |
| 388 | +impl<D, T, R> ContainerBuilder for UnsortedChunker<D, T, R> |
| 389 | +where |
| 390 | + D: Columnation + Clone + 'static, |
| 391 | + T: Columnation + Clone + 'static, |
| 392 | + R: Columnation + Clone + 'static, |
| 393 | +{ |
| 394 | + type Container = ColumnationStack<(D, T, R)>; |
| 395 | + |
| 396 | + fn extract(&mut self) -> Option<&mut Self::Container> { |
| 397 | + if let Some(ready) = self.ready.pop_front() { |
| 398 | + self.empty = Some(ready); |
| 399 | + self.empty.as_mut() |
| 400 | + } else { |
| 401 | + None |
| 402 | + } |
| 403 | + } |
| 404 | + |
| 405 | + fn finish(&mut self) -> Option<&mut Self::Container> { |
| 406 | + self.form_chunks(true); |
| 407 | + self.extract() |
255 | 408 | } |
256 | 409 | } |
257 | 410 |
|
@@ -350,6 +503,45 @@ mod tests { |
350 | 503 | assert!(!b.general); |
351 | 504 | } |
352 | 505 |
|
| 506 | + /// A chunk in arrival order, as [`UnsortedChunker`] produces. |
| 507 | + fn unsorted(updates: &[(i64, u64, i64)]) -> Chunk<(Row, ()), u64, i64> { |
| 508 | + let mut out = ColumnationStack::with_capacity(updates.len()); |
| 509 | + for (k, t, r) in updates { |
| 510 | + out.copy(&((Row::pack_slice(&[Datum::Int64(*k)]), ()), *t, *r)); |
| 511 | + } |
| 512 | + out |
| 513 | + } |
| 514 | + |
| 515 | + #[mz_ore::test] |
| 516 | + fn unsorted_chunks_on_both_paths() { |
| 517 | + let mut b = B::new(None, 0); |
| 518 | + b.push_into(unsorted(&[(3, 1, 1), (1, 1, 1), (3, 1, 1)])); |
| 519 | + let (chain, _) = b.seal(upper(2)); |
| 520 | + assert_eq!(collect(&chain), vec![(1, 1, 1), (3, 1, 2)]); |
| 521 | + b.push_into(unsorted(&[(9, 3, 1), (2, 3, 1), (9, 3, -1)])); |
| 522 | + b.push_into(unsorted(&[(5, 4, 1), (2, 3, 1)])); |
| 523 | + let (chain, _) = b.seal(upper(5)); |
| 524 | + assert!(b.general); |
| 525 | + assert_eq!(collect(&chain), vec![(2, 3, 2), (5, 4, 1)]); |
| 526 | + } |
| 527 | + |
| 528 | + #[mz_ore::test] |
| 529 | + fn unsorted_chunker_keeps_arrival_order_and_chunk_size() { |
| 530 | + let mut c: UnsortedChunker<(Row, ()), u64, i64> = UnsortedChunker::default(); |
| 531 | + let mut input: Vec<((Row, ()), u64, i64)> = (0..5) |
| 532 | + .rev() |
| 533 | + .map(|k| ((Row::pack_slice(&[Datum::Int64(k)]), ()), 1, 1)) |
| 534 | + .collect(); |
| 535 | + c.push_into(&mut input); |
| 536 | + assert!(c.extract().is_none(), "below chunk capacity, nothing ready"); |
| 537 | + let chunk = c.finish().expect("finish flushes"); |
| 538 | + let keys: Vec<i64> = chunk |
| 539 | + .iter() |
| 540 | + .map(|((k, ()), _, _)| k.iter().next().unwrap().unwrap_int64()) |
| 541 | + .collect(); |
| 542 | + assert_eq!(keys, vec![4, 3, 2, 1, 0]); |
| 543 | + } |
| 544 | + |
353 | 545 | #[mz_ore::test] |
354 | 546 | fn sort_prefix_agrees_with_row_order() { |
355 | 547 | let rows: Vec<Row> = [ |
|
0 commit comments