mz_timely_util/
columnation.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License in the LICENSE file at the
6// root of this repository, or online at
7//
8//     http://www.apache.org/licenses/LICENSE-2.0
9//
10// Unless required by applicable law or agreed to in writing, software
11// distributed under the License is distributed on an "AS IS" BASIS,
12// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13// See the License for the specific language governing permissions and
14// limitations under the License.
15
16//! Columnation-based containers vendored from differential-dataflow.
17//!
18//! This module provides [`ColumnationStack`], a columnar container that stores data using the
19//! columnation library, along with [`ColumnationChunker`] for organizing streams into sorted
20//! chunks, and the [`ColInternalMerger`] [`Merger`] implementation needed by the merge batcher.
21
22use std::collections::VecDeque;
23use std::iter::FromIterator;
24
25use columnation::{Columnation, Region};
26use differential_dataflow::consolidation::consolidate_updates;
27use differential_dataflow::difference::Semigroup;
28use differential_dataflow::lattice::Lattice;
29use differential_dataflow::trace::implementations::BatchContainer;
30use differential_dataflow::trace::implementations::merge_batcher::{Extractor, Merger};
31use differential_dataflow::trace::implementations::ord_neu::BuilderInput;
32use timely::container::{ContainerBuilder, DrainContainer, PushInto, SizableContainer};
33use timely::progress::Timestamp;
34use timely::progress::frontier::{Antichain, AntichainRef};
35use timely::{Accountable, PartialOrder};
36
37// ---------------------------------------------------------------------------
38// ColumnationStack
39// ---------------------------------------------------------------------------
40
41/// An append-only vector that stores records as columns.
42///
43/// This container maintains elements that might conventionally own
44/// memory allocations, but instead the pointers to those allocations
45/// reference larger regions of memory shared with multiple instances
46/// of the type. Elements can be retrieved as references, and care is
47/// taken when this type is dropped to ensure that the correct memory
48/// is returned (rather than the incorrect memory, from running the
49/// elements `Drop` implementations).
50pub struct ColumnationStack<T: Columnation> {
51    local: Vec<T>,
52    inner: T::InnerRegion,
53}
54
55impl<T: Columnation> ColumnationStack<T> {
56    /// Construct a [`ColumnationStack`], reserving space for `capacity` elements.
57    ///
58    /// Note that the associated region is not initialized to a specific capacity
59    /// because we can't generally know how much space would be required.
60    pub fn with_capacity(capacity: usize) -> Self {
61        Self {
62            local: Vec::with_capacity(capacity),
63            inner: T::InnerRegion::default(),
64        }
65    }
66
67    /// Ensures `Self` can absorb `items` without further allocations.
68    ///
69    /// The argument `items` may be cloned and iterated multiple times.
70    /// Please be careful if it contains side effects.
71    #[inline(always)]
72    pub fn reserve_items<'a, I>(&mut self, items: I)
73    where
74        I: Iterator<Item = &'a T> + Clone,
75        T: 'a,
76    {
77        self.local.reserve(items.clone().count());
78        self.inner.reserve_items(items);
79    }
80
81    /// Ensures `Self` can absorb `items` without further allocations.
82    ///
83    /// The argument `items` may be cloned and iterated multiple times.
84    /// Please be careful if it contains side effects.
85    #[inline(always)]
86    pub fn reserve_regions<'a, I>(&mut self, regions: I)
87    where
88        Self: 'a,
89        I: Iterator<Item = &'a Self> + Clone,
90    {
91        self.local
92            .reserve(regions.clone().map(|cs| cs.local.len()).sum());
93        self.inner.reserve_regions(regions.map(|cs| &cs.inner));
94    }
95
96    /// Copies an element into the region.
97    ///
98    /// The element can be read by indexing.
99    pub fn copy(&mut self, item: &T) {
100        unsafe {
101            self.local.push(self.inner.copy(item));
102        }
103    }
104
105    /// Empties the collection.
106    pub fn clear(&mut self) {
107        unsafe {
108            self.local.set_len(0);
109            self.inner.clear();
110        }
111    }
112
113    /// Retain elements that pass a predicate, from a specified offset.
114    ///
115    /// This method may or may not reclaim memory in the inner region.
116    pub fn retain_from<P: FnMut(&T) -> bool>(&mut self, index: usize, mut predicate: P) {
117        let mut write_position = index;
118        for position in index..self.local.len() {
119            if predicate(&self[position]) {
120                self.local.swap(position, write_position);
121                write_position += 1;
122            }
123        }
124        unsafe {
125            // Unsafety justified in that `write_position` is no greater than
126            // `self.local.len()` and so this exposes no invalid data.
127            self.local.set_len(write_position);
128        }
129    }
130
131    /// Unsafe access to `local` data. The slices store data that is backed by a region
132    /// allocation. Therefore, it is undefined behavior to mutate elements of the `local` slice.
133    ///
134    /// # Safety
135    /// Elements within `local` can be reordered, but not mutated, removed and/or dropped.
136    pub unsafe fn local(&mut self) -> &mut [T] {
137        &mut self.local[..]
138    }
139
140    /// Estimate the memory capacity in bytes.
141    #[inline]
142    pub fn heap_size(&self, mut callback: impl FnMut(usize, usize)) {
143        let size_of = std::mem::size_of::<T>();
144        callback(self.local.len() * size_of, self.local.capacity() * size_of);
145        self.inner.heap_size(callback);
146    }
147
148    /// Estimate the consumed memory capacity in bytes, summing both used and total capacity.
149    #[inline]
150    pub fn summed_heap_size(&self) -> (usize, usize) {
151        let (mut length, mut capacity) = (0, 0);
152        self.heap_size(|len, cap| {
153            length += len;
154            capacity += cap
155        });
156        (length, capacity)
157    }
158
159    /// The length in items.
160    #[inline]
161    pub fn len(&self) -> usize {
162        self.local.len()
163    }
164
165    /// Returns `true` if the stack is empty.
166    pub fn is_empty(&self) -> bool {
167        self.local.is_empty()
168    }
169
170    /// The capacity of the local vector.
171    #[inline]
172    pub fn capacity(&self) -> usize {
173        self.local.capacity()
174    }
175
176    /// Reserve space for `additional` elements.
177    #[inline]
178    pub fn reserve(&mut self, additional: usize) {
179        self.local.reserve(additional)
180    }
181}
182
183impl<A: Columnation, B: Columnation> ColumnationStack<(A, B)> {
184    /// Copies a destructured tuple `(A, B)` into this column stack.
185    pub fn copy_destructured(&mut self, t1: &A, t2: &B) {
186        unsafe {
187            self.local.push(self.inner.copy_destructured(t1, t2));
188        }
189    }
190}
191
192impl<A: Columnation, B: Columnation, C: Columnation> ColumnationStack<(A, B, C)> {
193    /// Copies a destructured tuple `(A, B, C)` into this column stack.
194    pub fn copy_destructured(&mut self, r0: &A, r1: &B, r2: &C) {
195        unsafe {
196            self.local.push(self.inner.copy_destructured(r0, r1, r2));
197        }
198    }
199}
200
201impl<T: Columnation> std::ops::Deref for ColumnationStack<T> {
202    type Target = [T];
203    #[inline(always)]
204    fn deref(&self) -> &Self::Target {
205        &self.local[..]
206    }
207}
208
209impl<T: Columnation> Drop for ColumnationStack<T> {
210    fn drop(&mut self) {
211        self.clear();
212    }
213}
214
215impl<T: Columnation> Default for ColumnationStack<T> {
216    fn default() -> Self {
217        Self {
218            local: Vec::new(),
219            inner: T::InnerRegion::default(),
220        }
221    }
222}
223
224impl<'a, A: 'a + Columnation> FromIterator<&'a A> for ColumnationStack<A> {
225    fn from_iter<I: IntoIterator<Item = &'a A>>(iter: I) -> Self {
226        let iter = iter.into_iter();
227        let mut c = ColumnationStack::<A>::with_capacity(iter.size_hint().0);
228        for element in iter {
229            c.copy(element);
230        }
231        c
232    }
233}
234
235impl<T: Columnation + PartialEq> PartialEq for ColumnationStack<T> {
236    fn eq(&self, other: &Self) -> bool {
237        PartialEq::eq(&self[..], &other[..])
238    }
239}
240
241impl<T: Columnation + Eq> Eq for ColumnationStack<T> {}
242
243impl<T: Columnation + std::fmt::Debug> std::fmt::Debug for ColumnationStack<T> {
244    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
245        self[..].fmt(f)
246    }
247}
248
249impl<T: Columnation> Clone for ColumnationStack<T> {
250    fn clone(&self) -> Self {
251        let mut new: Self = Default::default();
252        for item in &self[..] {
253            new.copy(item);
254        }
255        new
256    }
257
258    fn clone_from(&mut self, source: &Self) {
259        self.clear();
260        for item in &source[..] {
261            self.copy(item);
262        }
263    }
264}
265
266impl<T: Columnation> PushInto<T> for ColumnationStack<T> {
267    #[inline]
268    fn push_into(&mut self, item: T) {
269        self.copy(&item);
270    }
271}
272
273impl<T: Columnation> PushInto<&T> for ColumnationStack<T> {
274    #[inline]
275    fn push_into(&mut self, item: &T) {
276        self.copy(item);
277    }
278}
279
280impl<T: Columnation> PushInto<&&T> for ColumnationStack<T> {
281    #[inline]
282    fn push_into(&mut self, item: &&T) {
283        self.copy(*item);
284    }
285}
286
287// Container trait impls
288
289impl<T: Columnation> Accountable for ColumnationStack<T> {
290    #[inline]
291    fn record_count(&self) -> i64 {
292        i64::try_from(self.local.len()).unwrap()
293    }
294    #[inline]
295    fn is_empty(&self) -> bool {
296        self.local.is_empty()
297    }
298}
299
300impl<T: Columnation> DrainContainer for ColumnationStack<T> {
301    type Item<'a>
302        = &'a T
303    where
304        Self: 'a;
305    type DrainIter<'a>
306        = std::slice::Iter<'a, T>
307    where
308        Self: 'a;
309    #[inline]
310    fn drain(&mut self) -> Self::DrainIter<'_> {
311        (*self).iter()
312    }
313}
314
315impl<T: Columnation> SizableContainer for ColumnationStack<T> {
316    fn at_capacity(&self) -> bool {
317        self.len() == self.capacity()
318    }
319    fn ensure_capacity(&mut self, stash: &mut Option<Self>) {
320        if self.capacity() == 0 {
321            *self = stash.take().unwrap_or_default();
322            self.clear();
323        }
324        let preferred = timely::container::buffer::default_capacity::<T>();
325        if self.capacity() < preferred {
326            self.reserve(preferred - self.capacity());
327        }
328    }
329}
330
331impl<T: Clone + Ord + Columnation + 'static> BatchContainer for ColumnationStack<T> {
332    type Owned = T;
333    type ReadItem<'a> = &'a T;
334
335    #[inline(always)]
336    fn into_owned<'a>(item: Self::ReadItem<'a>) -> Self::Owned {
337        item.clone()
338    }
339    #[inline(always)]
340    fn clone_onto<'a>(item: Self::ReadItem<'a>, other: &mut Self::Owned) {
341        other.clone_from(item);
342    }
343
344    fn reborrow<'b, 'a: 'b>(item: Self::ReadItem<'a>) -> Self::ReadItem<'b> {
345        item
346    }
347
348    fn push_ref(&mut self, item: Self::ReadItem<'_>) {
349        self.push_into(item)
350    }
351    fn push_own(&mut self, item: &Self::Owned) {
352        self.push_into(item)
353    }
354
355    fn clear(&mut self) {
356        self.clear()
357    }
358
359    fn with_capacity(size: usize) -> Self {
360        Self::with_capacity(size)
361    }
362    fn merge_capacity(cont1: &Self, cont2: &Self) -> Self {
363        let mut new = Self::default();
364        new.reserve_regions(std::iter::once(cont1).chain(std::iter::once(cont2)));
365        new
366    }
367    fn index(&self, index: usize) -> Self::ReadItem<'_> {
368        &self[index]
369    }
370    fn len(&self) -> usize {
371        self[..].len()
372    }
373}
374
375impl<K, V, T, R> BuilderInput<K, V> for ColumnationStack<((K::Owned, V::Owned), T, R)>
376where
377    K: for<'a> BatchContainer<
378            ReadItem<'a>: PartialEq<&'a K::Owned>,
379            Owned: Ord + Columnation + Clone + 'static,
380        >,
381    V: for<'a> BatchContainer<
382            ReadItem<'a>: PartialEq<&'a V::Owned>,
383            Owned: Ord + Columnation + Clone + 'static,
384        >,
385    T: Timestamp + Lattice + Columnation + Clone + 'static,
386    R: Ord + Clone + Semigroup + Columnation + 'static,
387{
388    type Key<'a> = &'a K::Owned;
389    type Val<'a> = &'a V::Owned;
390    type Time = T;
391    type Diff = R;
392
393    fn into_parts<'a>(
394        ((key, val), time, diff): Self::Item<'a>,
395    ) -> (Self::Key<'a>, Self::Val<'a>, Self::Time, Self::Diff) {
396        (key, val, time.clone(), diff.clone())
397    }
398
399    fn key_eq(this: &&K::Owned, other: K::ReadItem<'_>) -> bool {
400        K::reborrow(other) == *this
401    }
402
403    fn val_eq(this: &&V::Owned, other: V::ReadItem<'_>) -> bool {
404        V::reborrow(other) == *this
405    }
406
407    fn key_val_upd_counts(chain: &[Self]) -> (usize, usize, usize) {
408        let mut keys = 0;
409        let mut vals = 0;
410        let mut upds = 0;
411        let mut prev_keyval = None;
412        for link in chain.iter() {
413            for ((key, val), _, _) in link.iter() {
414                if let Some((p_key, p_val)) = prev_keyval {
415                    if p_key != key {
416                        keys += 1;
417                        vals += 1;
418                    } else if p_val != val {
419                        vals += 1;
420                    }
421                } else {
422                    keys += 1;
423                    vals += 1;
424                }
425                upds += 1;
426                prev_keyval = Some((key, val));
427            }
428        }
429        (keys, vals, upds)
430    }
431}
432
433// ---------------------------------------------------------------------------
434// ColumnationChunker
435// ---------------------------------------------------------------------------
436
437/// Chunk a stream of vectors into chains of columnation stacks.
438///
439/// This chunker accumulates into a `Vec` (not a `ColumnationStack`) for efficient
440/// in-place sorting and consolidation, then copies the consolidated results
441/// into `ColumnationStack` chunks. This avoids the cost of sorting through
442/// columnation indirection.
443pub struct ColumnationChunker<T: Columnation> {
444    pending: Vec<T>,
445    ready: VecDeque<ColumnationStack<T>>,
446    empty: Option<ColumnationStack<T>>,
447}
448
449impl<T: Columnation> Default for ColumnationChunker<T> {
450    fn default() -> Self {
451        Self {
452            pending: Vec::default(),
453            ready: VecDeque::default(),
454            empty: None,
455        }
456    }
457}
458
459impl<D, T, R> ColumnationChunker<(D, T, R)>
460where
461    D: Columnation + Ord,
462    T: Columnation + Ord,
463    R: Columnation + Semigroup,
464{
465    const BUFFER_SIZE_BYTES: usize = 64 << 10;
466
467    fn chunk_capacity() -> usize {
468        let size = std::mem::size_of::<(D, T, R)>();
469        if size == 0 {
470            Self::BUFFER_SIZE_BYTES
471        } else if size <= Self::BUFFER_SIZE_BYTES {
472            Self::BUFFER_SIZE_BYTES / size
473        } else {
474            1
475        }
476    }
477
478    fn form_chunk(&mut self) {
479        consolidate_updates(&mut self.pending);
480        if self.pending.len() >= Self::chunk_capacity() {
481            while self.pending.len() > Self::chunk_capacity() {
482                let mut chunk = ColumnationStack::with_capacity(Self::chunk_capacity());
483                for item in self.pending.drain(..chunk.capacity()) {
484                    chunk.copy(&item);
485                }
486                self.ready.push_back(chunk);
487            }
488        }
489    }
490}
491
492impl<'a, D, T, R> PushInto<&'a mut Vec<(D, T, R)>> for ColumnationChunker<(D, T, R)>
493where
494    D: Columnation + Ord + Clone,
495    T: Columnation + Ord + Clone,
496    R: Columnation + Semigroup + Clone,
497{
498    fn push_into(&mut self, container: &'a mut Vec<(D, T, R)>) {
499        if self.pending.capacity() < Self::chunk_capacity() * 2 {
500            self.pending
501                .reserve(Self::chunk_capacity() * 2 - self.pending.len());
502        }
503
504        let mut drain = container.drain(..).peekable();
505        while drain.peek().is_some() {
506            self.pending
507                .extend((&mut drain).take(self.pending.capacity() - self.pending.len()));
508            if self.pending.len() == self.pending.capacity() {
509                self.form_chunk();
510            }
511        }
512    }
513}
514
515impl<D, T, R> ContainerBuilder for ColumnationChunker<(D, T, R)>
516where
517    D: Columnation + Ord + Clone + 'static,
518    T: Columnation + Ord + Clone + 'static,
519    R: Columnation + Semigroup + Clone + 'static,
520{
521    type Container = ColumnationStack<(D, T, R)>;
522
523    fn extract(&mut self) -> Option<&mut Self::Container> {
524        if let Some(ready) = self.ready.pop_front() {
525            self.empty = Some(ready);
526            self.empty.as_mut()
527        } else {
528            None
529        }
530    }
531
532    fn finish(&mut self) -> Option<&mut Self::Container> {
533        consolidate_updates(&mut self.pending);
534        while !self.pending.is_empty() {
535            let mut chunk = ColumnationStack::with_capacity(Self::chunk_capacity());
536            for item in self
537                .pending
538                .drain(..std::cmp::min(self.pending.len(), chunk.capacity()))
539            {
540                chunk.copy(&item);
541            }
542            self.ready.push_back(chunk);
543        }
544        self.empty = self.ready.pop_front();
545        self.empty.as_mut()
546    }
547}
548
549// ---------------------------------------------------------------------------
550// ColInternalMerger: a `Merger` for `ColumnationStack` chunks
551// ---------------------------------------------------------------------------
552
553/// A [`Merger`] using internal iteration over [`ColumnationStack`] chunks.
554///
555/// Implements differential-dataflow's chunk-list merge-batcher interface by
556/// merging and splitting chains of sorted, consolidated `ColumnationStack`
557/// chunks. Elements are copied through `copy` / `copy_destructured` rather than
558/// moved out of their backing regions, which is why this operates by internal
559/// iteration rather than draining owned tuples like the stock `VecMerger`.
560pub struct ColInternalMerger<D, T, R> {
561    _marker: std::marker::PhantomData<(D, T, R)>,
562}
563
564impl<D, T, R> Default for ColInternalMerger<D, T, R> {
565    fn default() -> Self {
566        Self {
567            _marker: std::marker::PhantomData,
568        }
569    }
570}
571
572impl<D, T, R> ColInternalMerger<D, T, R>
573where
574    D: Ord + Columnation + Clone + 'static,
575    T: Ord + Columnation + Clone + PartialOrder + 'static,
576    R: Default + Semigroup + Columnation + Clone + 'static,
577{
578    /// Target chunk capacity in elements, mirroring [`ColumnationChunker`].
579    fn chunk_capacity() -> usize {
580        const BUFFER_SIZE_BYTES: usize = 64 << 10;
581        let size = std::mem::size_of::<(D, T, R)>();
582        if size == 0 {
583            BUFFER_SIZE_BYTES
584        } else if size <= BUFFER_SIZE_BYTES {
585            BUFFER_SIZE_BYTES / size
586        } else {
587            1
588        }
589    }
590
591    /// Acquire an empty chunk with the target capacity, recycling from `stash`.
592    fn empty(stash: &mut Vec<ColumnationStack<(D, T, R)>>) -> ColumnationStack<(D, T, R)> {
593        let target = Self::chunk_capacity();
594        match stash.pop() {
595            Some(mut chunk) if chunk.capacity() >= target => {
596                chunk.clear();
597                chunk
598            }
599            _ => ColumnationStack::with_capacity(target),
600        }
601    }
602
603    /// Recycle a consumed chunk into `stash`, retaining its allocations.
604    fn recycle(
605        mut chunk: ColumnationStack<(D, T, R)>,
606        stash: &mut Vec<ColumnationStack<(D, T, R)>>,
607    ) {
608        chunk.clear();
609        stash.push(chunk);
610    }
611
612    /// Drain remaining items from one side into `result`/`output`.
613    ///
614    /// Copies the partially-consumed head into `result` (swapping wholesale
615    /// when `result` is empty and the head untouched), then appends remaining
616    /// full chunks directly to `output` without copying.
617    fn drain_side(
618        head: &mut ColumnationStack<(D, T, R)>,
619        pos: &mut usize,
620        list: &mut std::vec::IntoIter<ColumnationStack<(D, T, R)>>,
621        result: &mut ColumnationStack<(D, T, R)>,
622        output: &mut Vec<ColumnationStack<(D, T, R)>>,
623        stash: &mut Vec<ColumnationStack<(D, T, R)>>,
624    ) {
625        // Copy the partially-consumed head into result.
626        if *pos < head[..].len() {
627            if result.is_empty() && *pos == 0 {
628                std::mem::swap(result, head);
629            } else {
630                for i in *pos..head[..].len() {
631                    result.copy(&head[i]);
632                }
633            }
634            *pos = head[..].len();
635        }
636        // Flush result before appending full chunks.
637        if !result.is_empty() {
638            output.push(std::mem::replace(result, Self::empty(stash)));
639        }
640        // Remaining full chunks go directly to output.
641        output.extend(list);
642    }
643}
644
645impl<D, T, R> Merger for ColInternalMerger<D, T, R>
646where
647    D: Ord + Columnation + Clone + 'static,
648    T: Ord + Columnation + Clone + PartialOrder + 'static,
649    R: Default + Semigroup + Columnation + Clone + 'static,
650{
651    type Chunk = ColumnationStack<(D, T, R)>;
652
653    fn merge(
654        &mut self,
655        list1: Vec<Self::Chunk>,
656        list2: Vec<Self::Chunk>,
657        output: &mut Vec<Self::Chunk>,
658        stash: &mut Vec<Self::Chunk>,
659    ) {
660        use std::cmp::Ordering;
661
662        let mut list1 = list1.into_iter();
663        let mut list2 = list2.into_iter();
664        let mut head1 = list1.next().unwrap_or_default();
665        let mut head2 = list2.next().unwrap_or_default();
666        let mut pos1 = 0;
667        let mut pos2 = 0;
668        let mut result = Self::empty(stash);
669        let mut diff = R::default();
670
671        // Main merge loop: both sides have data.
672        while pos1 < head1[..].len() && pos2 < head2[..].len() {
673            // Tight inner loop bounded by the current heads and result capacity.
674            while pos1 < head1[..].len() && pos2 < head2[..].len() && !result.at_capacity() {
675                let (d1, t1, r1) = &head1[pos1];
676                let (d2, t2, r2) = &head2[pos2];
677                match (d1, t1).cmp(&(d2, t2)) {
678                    Ordering::Less => {
679                        result.copy(&head1[pos1]);
680                        pos1 += 1;
681                    }
682                    Ordering::Greater => {
683                        result.copy(&head2[pos2]);
684                        pos2 += 1;
685                    }
686                    Ordering::Equal => {
687                        diff.clone_from(r1);
688                        diff.plus_equals(r2);
689                        if !diff.is_zero() {
690                            result.copy_destructured(d1, t1, &diff);
691                        }
692                        pos1 += 1;
693                        pos2 += 1;
694                    }
695                }
696            }
697            if result.at_capacity() {
698                output.push(std::mem::replace(&mut result, Self::empty(stash)));
699            }
700            if pos1 >= head1[..].len() {
701                let old = std::mem::replace(&mut head1, list1.next().unwrap_or_default());
702                Self::recycle(old, stash);
703                pos1 = 0;
704            }
705            if pos2 >= head2[..].len() {
706                let old = std::mem::replace(&mut head2, list2.next().unwrap_or_default());
707                Self::recycle(old, stash);
708                pos2 = 0;
709            }
710        }
711
712        // After the loop at least one side is exhausted; draining it is a
713        // no-op, and the other contributes its sorted tail in order — the
714        // partial head by copy, remaining full chunks by passthrough.
715        Self::drain_side(
716            &mut head1,
717            &mut pos1,
718            &mut list1,
719            &mut result,
720            output,
721            stash,
722        );
723        Self::drain_side(
724            &mut head2,
725            &mut pos2,
726            &mut list2,
727            &mut result,
728            output,
729            stash,
730        );
731
732        if !result.is_empty() {
733            output.push(result);
734        }
735    }
736
737    fn len(chunk: &Self::Chunk) -> usize {
738        chunk[..].len()
739    }
740
741    fn allocation(chunk: &Self::Chunk) -> (usize, usize, usize) {
742        let (mut size, mut capacity, mut allocations) = (0, 0, 0);
743        chunk.heap_size(|siz, cap| {
744            size += siz;
745            capacity += cap;
746            allocations += 1;
747        });
748        (size, capacity, allocations)
749    }
750}
751
752impl<D, T, R> Extractor for ColInternalMerger<D, T, R>
753where
754    D: Ord + Columnation + Clone + 'static,
755    T: Ord + Columnation + Clone + PartialOrder + 'static,
756    R: Default + Semigroup + Columnation + Clone + 'static,
757{
758    type Time = T;
759
760    fn extract(
761        &mut self,
762        merged: Vec<Self::Chunk>,
763        upper: AntichainRef<T>,
764        frontier: &mut Antichain<T>,
765        readied: &mut Vec<Self::Chunk>,
766        kept: &mut Vec<Self::Chunk>,
767        stash: &mut Vec<Self::Chunk>,
768    ) {
769        let mut keep = Self::empty(stash);
770        let mut ready = Self::empty(stash);
771
772        for chunk in merged {
773            let len = chunk[..].len();
774            for i in 0..len {
775                let (data, time, diff) = &chunk[i];
776                if upper.less_equal(time) {
777                    frontier.insert_with(time, |time| time.clone());
778                    keep.copy_destructured(data, time, diff);
779                } else {
780                    ready.copy_destructured(data, time, diff);
781                }
782                if keep.at_capacity() {
783                    kept.push(std::mem::replace(&mut keep, Self::empty(stash)));
784                }
785                if ready.at_capacity() {
786                    readied.push(std::mem::replace(&mut ready, Self::empty(stash)));
787                }
788            }
789            // Chunk fully consumed; recycle its allocations.
790            Self::recycle(chunk, stash);
791        }
792
793        if !keep.is_empty() {
794            kept.push(keep);
795        }
796        if !ready.is_empty() {
797            readied.push(ready);
798        }
799    }
800}
mz_timely_util/columnation.rs

mz_timely_util/
columnation.rs
