Skip to main content

dbsp/trace/ord/file/
key_batch.rs

1use crate::storage::file::format::BatchMetadata;
2use crate::storage::file::{
3    FilterKind, FilterStats, TouchedWindowCount, TouchedWindowCounter, collect_roaring_metadata,
4};
5use crate::trace::cursor::Position;
6use crate::{
7    DBData, DBWeight, NumEntries, Runtime, Timestamp,
8    dynamic::{
9        DataTrait, DynDataTyped, DynOpt, DynPair, DynUnit, DynVec, DynWeightedPairs, Erase,
10        Factory, LeanVec, WeightTrait, WithFactory,
11    },
12    storage::{
13        buffer_cache::CacheStats,
14        file::{
15            Factories as FileFactories, FilterPlan,
16            reader::{Cursor as FileCursor, Error as ReaderError, Reader},
17            writer::Writer2,
18        },
19    },
20    trace::{
21        Batch, BatchFactories, BatchLocation, BatchReader, BatchReaderFactories, Builder, Cursor,
22        WeightedItem,
23        filter::BatchFilters,
24        ord::{file::UnwrapStorage, merge_batcher::MergeBatcher},
25    },
26    utils::Tup2,
27};
28use feldera_storage::{FileReader, StoragePath};
29use rand::{Rng, seq::index::sample};
30use rkyv::{Archive, Archived, Deserialize, Fallible, Serialize, ser::Serializer};
31use size_of::SizeOf;
32use std::any::TypeId;
33use std::{
34    fmt::{self, Debug},
35    sync::Arc,
36};
37
38pub struct FileKeyBatchFactories<K, T, R>
39where
40    K: DataTrait + ?Sized,
41    T: Timestamp,
42    R: WeightTrait + ?Sized,
43{
44    key_factory: &'static dyn Factory<K>,
45    weight_factory: &'static dyn Factory<R>,
46    weights_factory: &'static dyn Factory<DynVec<R>>,
47    keys_factory: &'static dyn Factory<DynVec<K>>,
48    item_factory: &'static dyn Factory<DynPair<K, DynUnit>>,
49    factories0: FileFactories<K, DynUnit>,
50    factories1: FileFactories<DynDataTyped<T>, R>,
51    opt_key_factory: &'static dyn Factory<DynOpt<K>>,
52    weighted_item_factory: &'static dyn Factory<WeightedItem<K, DynUnit, R>>,
53    weighted_items_factory: &'static dyn Factory<DynWeightedPairs<DynPair<K, DynUnit>, R>>,
54    weighted_vals_factory: &'static dyn Factory<DynWeightedPairs<DynUnit, R>>,
55    pub timediff_factory: &'static dyn Factory<DynWeightedPairs<DynDataTyped<T>, R>>,
56}
57
58impl<K, T, R> Clone for FileKeyBatchFactories<K, T, R>
59where
60    K: DataTrait + ?Sized,
61    T: Timestamp,
62    R: WeightTrait + ?Sized,
63{
64    fn clone(&self) -> Self {
65        Self {
66            key_factory: self.key_factory,
67            weight_factory: self.weight_factory,
68            weights_factory: self.weights_factory,
69            keys_factory: self.keys_factory,
70            item_factory: self.item_factory,
71            factories0: self.factories0.clone(),
72            factories1: self.factories1.clone(),
73            opt_key_factory: self.opt_key_factory,
74            weighted_item_factory: self.weighted_item_factory,
75            weighted_items_factory: self.weighted_items_factory,
76            weighted_vals_factory: self.weighted_vals_factory,
77            timediff_factory: self.timediff_factory,
78        }
79    }
80}
81
82impl<K, T, R> BatchReaderFactories<K, DynUnit, T, R> for FileKeyBatchFactories<K, T, R>
83where
84    K: DataTrait + ?Sized,
85    T: Timestamp,
86    R: WeightTrait + ?Sized,
87{
88    fn new<KType, VType, RType>() -> Self
89    where
90        KType: DBData + Erase<K>,
91        VType: DBData + Erase<DynUnit>,
92        RType: DBWeight + Erase<R>,
93    {
94        Self {
95            key_factory: WithFactory::<KType>::FACTORY,
96            weight_factory: WithFactory::<RType>::FACTORY,
97            weights_factory: WithFactory::<LeanVec<RType>>::FACTORY,
98            keys_factory: WithFactory::<LeanVec<KType>>::FACTORY,
99            item_factory: WithFactory::<Tup2<KType, ()>>::FACTORY,
100            factories0: FileFactories::new::<KType, ()>(),
101            factories1: FileFactories::new::<T, RType>(),
102            opt_key_factory: WithFactory::<Option<KType>>::FACTORY,
103            weighted_item_factory: WithFactory::<Tup2<Tup2<KType, ()>, RType>>::FACTORY,
104            weighted_items_factory: WithFactory::<LeanVec<Tup2<Tup2<KType, ()>, RType>>>::FACTORY,
105            weighted_vals_factory: WithFactory::<LeanVec<Tup2<(), RType>>>::FACTORY,
106            timediff_factory: WithFactory::<LeanVec<Tup2<T, RType>>>::FACTORY,
107        }
108    }
109
110    fn key_factory(&self) -> &'static dyn Factory<K> {
111        self.key_factory
112    }
113
114    fn keys_factory(&self) -> &'static dyn Factory<DynVec<K>> {
115        self.keys_factory
116    }
117
118    fn val_factory(&self) -> &'static dyn Factory<DynUnit> {
119        WithFactory::<()>::FACTORY
120    }
121
122    fn weight_factory(&self) -> &'static dyn Factory<R> {
123        self.weight_factory
124    }
125}
126
127impl<K, T, R> BatchFactories<K, DynUnit, T, R> for FileKeyBatchFactories<K, T, R>
128where
129    K: DataTrait + ?Sized,
130    T: Timestamp,
131    R: WeightTrait + ?Sized,
132{
133    fn item_factory(&self) -> &'static dyn Factory<DynPair<K, DynUnit>> {
134        self.item_factory
135    }
136
137    fn weighted_item_factory(&self) -> &'static dyn Factory<WeightedItem<K, DynUnit, R>> {
138        self.weighted_item_factory
139    }
140
141    fn weighted_items_factory(
142        &self,
143    ) -> &'static dyn Factory<DynWeightedPairs<DynPair<K, DynUnit>, R>> {
144        self.weighted_items_factory
145    }
146
147    fn weighted_vals_factory(&self) -> &'static dyn Factory<DynWeightedPairs<DynUnit, R>> {
148        self.weighted_vals_factory
149    }
150
151    fn time_diffs_factory(
152        &self,
153    ) -> Option<&'static dyn Factory<DynWeightedPairs<DynDataTyped<T>, R>>> {
154        Some(self.timediff_factory)
155    }
156}
157
158/// A batch of keys with weights and times.
159///
160/// Each tuple in `FileKeyBatch<K, T, R>` has key type `K`, value type `()`,
161/// weight type `R`, and time type `R`.
162#[derive(SizeOf)]
163pub struct FileKeyBatch<K, T, R>
164where
165    K: DataTrait + ?Sized,
166    T: Timestamp,
167    R: WeightTrait + ?Sized,
168{
169    #[size_of(skip)]
170    factories: FileKeyBatchFactories<K, T, R>,
171    #[allow(clippy::type_complexity)]
172    file: Arc<
173        Reader<(
174            &'static K,
175            &'static DynUnit,
176            (&'static DynDataTyped<T>, &'static R, ()),
177        )>,
178    >,
179    filters: BatchFilters<K>,
180}
181
182impl<K, T, R> Debug for FileKeyBatch<K, T, R>
183where
184    K: DataTrait + ?Sized,
185    T: Timestamp,
186    R: WeightTrait + ?Sized,
187{
188    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
189        write!(f, "FileKeyBatch {{ data: ")?;
190        let mut cursor = self.cursor();
191        let mut n_keys = 0;
192        while cursor.key_valid() {
193            if n_keys > 0 {
194                write!(f, ", ")?;
195            }
196            write!(f, "{:?}(", cursor.key())?;
197            let mut n_values = 0;
198            cursor.map_times(&mut |time, diff| {
199                if n_values > 0 {
200                    let _ = write!(f, ", ");
201                }
202                let _ = write!(f, "({time:?}, {diff:+?})");
203                n_values += 1;
204            });
205            write!(f, ")")?;
206            n_keys += 1;
207            cursor.step_key();
208        }
209        write!(f, " }}")
210    }
211}
212
213impl<K, T, R> Clone for FileKeyBatch<K, T, R>
214where
215    K: DataTrait + ?Sized,
216    T: Timestamp,
217    R: WeightTrait + ?Sized,
218{
219    fn clone(&self) -> Self {
220        Self {
221            factories: self.factories.clone(),
222            file: self.file.clone(),
223            filters: self.filters.clone(),
224        }
225    }
226}
227
228impl<K, T, R> FileKeyBatch<K, T, R>
229where
230    K: DataTrait + ?Sized,
231    T: Timestamp,
232    R: WeightTrait + ?Sized,
233{
234    fn from_parts(
235        factories: FileKeyBatchFactories<K, T, R>,
236        file: Arc<
237            Reader<(
238                &'static K,
239                &'static DynUnit,
240                (&'static DynDataTyped<T>, &'static R, ()),
241            )>,
242        >,
243        filters: BatchFilters<K>,
244    ) -> Self {
245        Self {
246            factories,
247            file,
248            filters,
249        }
250    }
251}
252
253impl<K, T, R> NumEntries for FileKeyBatch<K, T, R>
254where
255    K: DataTrait + ?Sized,
256    T: Timestamp,
257    R: WeightTrait + ?Sized,
258{
259    const CONST_NUM_ENTRIES: Option<usize> = None;
260
261    fn num_entries_shallow(&self) -> usize {
262        self.file.rows().len() as usize
263    }
264
265    fn num_entries_deep(&self) -> usize {
266        self.file.n_rows(1) as usize
267    }
268}
269
270impl<K, T, R> BatchReader for FileKeyBatch<K, T, R>
271where
272    K: DataTrait + ?Sized,
273    T: Timestamp,
274    R: WeightTrait + ?Sized,
275{
276    type Factories = FileKeyBatchFactories<K, T, R>;
277    type Key = K;
278    type Val = DynUnit;
279    type Time = T;
280    type R = R;
281    type Cursor<'s> = FileKeyCursor<'s, K, T, R>;
282
283    fn factories(&self) -> Self::Factories {
284        self.factories.clone()
285    }
286
287    fn cursor(&self) -> Self::Cursor<'_> {
288        FileKeyCursor::new(self)
289    }
290
291    #[inline]
292    fn key_count(&self) -> usize {
293        self.file.n_rows(0) as usize
294    }
295
296    #[inline]
297    fn len(&self) -> usize {
298        self.file.n_rows(1) as usize
299    }
300
301    fn approximate_byte_size(&self) -> usize {
302        self.file.byte_size().unwrap_storage() as usize
303    }
304
305    fn membership_filter_stats(&self) -> FilterStats {
306        self.filters.stats().membership_filter
307    }
308
309    fn membership_filter_kind(&self) -> FilterKind {
310        self.filters.membership_filter_kind()
311    }
312
313    fn range_filter_stats(&self) -> FilterStats {
314        self.filters.stats().range_filter
315    }
316
317    #[inline]
318    fn location(&self) -> BatchLocation {
319        BatchLocation::Storage
320    }
321
322    fn cache_stats(&self) -> CacheStats {
323        self.file.cache_stats()
324    }
325
326    fn sample_keys<RG>(&self, rng: &mut RG, sample_size: usize, output: &mut DynVec<Self::Key>)
327    where
328        RG: Rng,
329    {
330        let size = self.key_count();
331        let mut cursor = self.cursor();
332        if sample_size >= size {
333            output.reserve(size);
334
335            while cursor.key_valid() {
336                output.push_ref(cursor.key());
337                cursor.step_key();
338            }
339        } else {
340            output.reserve(sample_size);
341
342            let mut indexes = sample(rng, size, sample_size).into_vec();
343            indexes.sort_unstable();
344            for index in indexes.into_iter() {
345                cursor.move_key(|key_cursor| unsafe { key_cursor.move_to_row(index as u64) });
346                output.push_ref(cursor.key());
347            }
348        }
349    }
350}
351
352impl<K, T, R> Batch for FileKeyBatch<K, T, R>
353where
354    K: DataTrait + ?Sized,
355    T: Timestamp,
356    R: WeightTrait + ?Sized,
357{
358    type Timed<T2: Timestamp> = FileKeyBatch<K, T2, R>;
359    type Batcher = MergeBatcher<Self>;
360    type Builder = FileKeyBuilder<K, T, R>;
361
362    fn file_reader(&self) -> Option<Arc<dyn FileReader>> {
363        self.file.mark_for_checkpoint();
364        Some(self.file.file_handle().clone())
365    }
366
367    fn from_path(factories: &Self::Factories, path: &StoragePath) -> Result<Self, ReaderError> {
368        let any_factory0 = factories.factories0.any_factories();
369        let any_factory1 = factories.factories1.any_factories();
370        let (file, membership_filter) = Reader::open_with_filter(
371            &[&any_factory0, &any_factory1],
372            Runtime::buffer_cache,
373            &*Runtime::storage_backend().unwrap_storage(),
374            path,
375        )?;
376        let file = Arc::new(file);
377        let key_range = file.key_range()?.map(Into::into);
378        let filters = BatchFilters::from_file(key_range, membership_filter);
379
380        Ok(Self::from_parts(factories.clone(), file, filters))
381    }
382
383    fn key_bounds(&self) -> Option<(&Self::Key, &Self::Key)> {
384        self.filters.key_bounds()
385    }
386
387    fn negative_weight_count(&self) -> Option<u64> {
388        None
389    }
390
391    fn touched_window_count(&self) -> TouchedWindowCount {
392        self.file.metadata().touched_window_count
393    }
394}
395
396type RawKeyCursor<'s, K, T, R> = FileCursor<
397    's,
398    K,
399    DynUnit,
400    (&'static DynDataTyped<T>, &'static R, ()),
401    (
402        &'static K,
403        &'static DynUnit,
404        (&'static DynDataTyped<T>, &'static R, ()),
405    ),
406>;
407
408/// A cursor for navigating a single layer.
409#[derive(Debug, SizeOf)]
410pub struct FileKeyCursor<'s, K, T, R>
411where
412    K: DataTrait + ?Sized,
413    T: Timestamp,
414    R: WeightTrait + ?Sized,
415{
416    batch: &'s FileKeyBatch<K, T, R>,
417    pub(crate) cursor: RawKeyCursor<'s, K, T, R>,
418    val_valid: bool,
419
420    pub(crate) time: Box<DynDataTyped<T>>,
421    pub(crate) diff: Box<R>,
422}
423
424impl<K, T, R> Clone for FileKeyCursor<'_, K, T, R>
425where
426    K: DataTrait + ?Sized,
427    T: Timestamp,
428    R: WeightTrait + ?Sized,
429{
430    fn clone(&self) -> Self {
431        Self {
432            batch: self.batch,
433            cursor: self.cursor.clone(),
434            val_valid: self.val_valid,
435
436            // These don't need to be cloned because they're only used for
437            // temporary storage.
438            time: self.batch.factories.factories1.key_factory.default_box(),
439            diff: self.batch.factories.weight_factory.default_box(),
440        }
441    }
442}
443
444impl<'s, K, T, R> FileKeyCursor<'s, K, T, R>
445where
446    K: DataTrait + ?Sized,
447    T: Timestamp,
448    R: WeightTrait + ?Sized,
449{
450    fn new_from(batch: &'s FileKeyBatch<K, T, R>, lower_bound: usize) -> Self {
451        let cursor = unsafe {
452            batch
453                .file
454                .rows()
455                .subset(lower_bound as u64..)
456                .first()
457                .unwrap_storage()
458        };
459        let key_valid = cursor.has_value();
460
461        Self {
462            batch,
463            cursor,
464            val_valid: key_valid,
465            time: batch.factories.factories1.key_factory.default_box(),
466            diff: batch.factories.weight_factory.default_box(),
467        }
468    }
469
470    fn new(batch: &'s FileKeyBatch<K, T, R>) -> Self {
471        Self::new_from(batch, 0)
472    }
473
474    fn move_key<F>(&mut self, op: F)
475    where
476        F: Fn(&mut RawKeyCursor<'s, K, T, R>) -> Result<(), ReaderError>,
477    {
478        op(&mut self.cursor).unwrap_storage();
479        self.val_valid = self.cursor.has_value();
480    }
481}
482
483impl<K, T, R> Cursor<K, DynUnit, T, R> for FileKeyCursor<'_, K, T, R>
484where
485    K: DataTrait + ?Sized,
486    T: Timestamp,
487    R: WeightTrait + ?Sized,
488{
489    fn weight_factory(&self) -> &'static dyn Factory<R> {
490        self.batch.factories.weight_factory
491    }
492
493    fn key(&self) -> &K {
494        debug_assert!(self.key_valid());
495        self.cursor.key().unwrap()
496    }
497
498    fn val(&self) -> &DynUnit {
499        &()
500    }
501
502    fn map_times(&mut self, logic: &mut dyn FnMut(&T, &R)) {
503        let mut val_cursor = unsafe {
504            self.cursor
505                .next_column()
506                .unwrap_storage()
507                .first()
508                .unwrap_storage()
509        };
510        while unsafe { val_cursor.item((self.time.as_mut(), &mut self.diff)) }.is_some() {
511            logic(self.time.as_ref(), self.diff.as_ref());
512            unsafe { val_cursor.move_next() }.unwrap_storage();
513        }
514    }
515
516    fn map_times_through(&mut self, upper: &T, logic: &mut dyn FnMut(&T, &R)) {
517        let mut val_cursor = unsafe {
518            self.cursor
519                .next_column()
520                .unwrap_storage()
521                .first()
522                .unwrap_storage()
523        };
524        while unsafe { val_cursor.item((self.time.as_mut(), &mut self.diff)) }.is_some() {
525            if self.time.less_equal(upper) {
526                logic(self.time.as_ref(), self.diff.as_ref());
527            }
528            unsafe { val_cursor.move_next() }.unwrap_storage();
529        }
530    }
531
532    fn map_values(&mut self, logic: &mut dyn FnMut(&DynUnit, &R))
533    where
534        T: PartialEq<()>,
535    {
536        if self.val_valid {
537            logic(&(), self.weight())
538        }
539    }
540
541    fn weight(&mut self) -> &R
542    where
543        T: PartialEq<()>,
544    {
545        self.weight_checked()
546    }
547
548    fn weight_checked(&mut self) -> &R {
549        if TypeId::of::<T>() == TypeId::of::<()>() {
550            let val_cursor = unsafe {
551                self.cursor
552                    .next_column()
553                    .unwrap_storage()
554                    .first()
555                    .unwrap_storage()
556            };
557            unsafe { val_cursor.aux(&mut self.diff) }.unwrap();
558            self.diff.as_ref()
559        } else {
560            panic!("FileKeyCursor::weight_checked called on non-unit timestamp type");
561        }
562    }
563
564    fn key_valid(&self) -> bool {
565        self.cursor.has_value()
566    }
567
568    fn val_valid(&self) -> bool {
569        self.val_valid
570    }
571
572    fn step_key(&mut self) {
573        self.move_key(|key_cursor| unsafe { key_cursor.move_next() });
574    }
575
576    fn step_key_reverse(&mut self) {
577        self.move_key(|key_cursor| unsafe { key_cursor.move_prev() });
578    }
579
580    fn seek_key(&mut self, key: &K) {
581        self.move_key(|key_cursor| unsafe { key_cursor.advance_to_value_or_larger(key) });
582    }
583
584    fn seek_key_exact(&mut self, key: &K, hash: Option<u64>) -> bool {
585        if !self.batch.filters.maybe_contains_key(key, hash) {
586            return false;
587        }
588        self.seek_key(key);
589        self.key_valid() && self.key().eq(key)
590    }
591
592    fn seek_key_with(&mut self, predicate: &dyn Fn(&K) -> bool) {
593        self.move_key(|key_cursor| unsafe { key_cursor.seek_forward_until(predicate) });
594    }
595
596    fn seek_key_with_reverse(&mut self, predicate: &dyn Fn(&K) -> bool) {
597        self.move_key(|key_cursor| unsafe { key_cursor.seek_backward_until(predicate) });
598    }
599
600    fn seek_key_reverse(&mut self, key: &K) {
601        self.move_key(|key_cursor| unsafe { key_cursor.rewind_to_value_or_smaller(key) });
602    }
603
604    fn step_val(&mut self) {
605        self.val_valid = false;
606    }
607
608    fn seek_val(&mut self, _val: &DynUnit) {}
609
610    fn seek_val_with(&mut self, predicate: &dyn Fn(&DynUnit) -> bool) {
611        if !predicate(&()) {
612            self.val_valid = false;
613        }
614    }
615
616    fn rewind_keys(&mut self) {
617        self.move_key(|key_cursor| unsafe { key_cursor.move_first() });
618    }
619
620    fn fast_forward_keys(&mut self) {
621        self.move_key(|key_cursor| unsafe { key_cursor.move_last() });
622    }
623
624    fn rewind_vals(&mut self) {
625        self.val_valid = true;
626    }
627
628    fn step_val_reverse(&mut self) {
629        self.val_valid = false;
630    }
631
632    fn seek_val_reverse(&mut self, _val: &DynUnit) {}
633
634    fn seek_val_with_reverse(&mut self, predicate: &dyn Fn(&DynUnit) -> bool) {
635        if !predicate(&()) {
636            self.val_valid = false;
637        }
638    }
639
640    fn fast_forward_vals(&mut self) {
641        self.val_valid = true;
642    }
643
644    fn position(&self) -> Option<Position> {
645        Some(Position {
646            total: self.cursor.len(),
647            offset: self.cursor.absolute_position(),
648        })
649    }
650}
651
652/// A builder for creating layers from unsorted update tuples.
653#[derive(SizeOf)]
654pub struct FileKeyBuilder<K, T, R>
655where
656    K: DataTrait + ?Sized,
657    T: Timestamp,
658    R: WeightTrait + ?Sized,
659{
660    #[size_of(skip)]
661    factories: FileKeyBatchFactories<K, T, R>,
662    #[size_of(skip)]
663    writer: Writer2<K, DynUnit, DynDataTyped<T>, R>,
664    key: Box<DynOpt<K>>,
665    num_tuples: usize,
666    #[size_of(skip)]
667    stats: BatchMetadata,
668    #[size_of(skip)]
669    touched_window_counter: Option<TouchedWindowCounter>,
670}
671
672impl<K, T, R> Builder<FileKeyBatch<K, T, R>> for FileKeyBuilder<K, T, R>
673where
674    Self: SizeOf,
675    K: DataTrait + ?Sized,
676    T: Timestamp,
677    R: WeightTrait + ?Sized,
678{
679    #[inline]
680    fn with_capacity_in_location(
681        factories: &FileKeyBatchFactories<K, T, R>,
682        key_capacity: usize,
683        _value_capacity: usize,
684        _location: Option<BatchLocation>,
685    ) -> Self {
686        Self {
687            factories: factories.clone(),
688            writer: Writer2::new(
689                &factories.factories0,
690                &factories.factories1,
691                Runtime::buffer_cache,
692                &*Runtime::storage_backend().unwrap_storage(),
693                Runtime::file_writer_parameters(),
694                FilterPlan::<K>::decide_filter(None, key_capacity),
695            )
696            .unwrap_storage(),
697            key: factories.opt_key_factory.default_box(),
698            num_tuples: 0,
699            stats: BatchMetadata::default(),
700            touched_window_counter: collect_roaring_metadata().then(TouchedWindowCounter::default),
701        }
702    }
703
704    fn for_merge<'a, B, I>(
705        factories: &FileKeyBatchFactories<K, T, R>,
706        batches: I,
707        _location: Option<BatchLocation>,
708    ) -> Self
709    where
710        B: Batch<Key = K, Val = DynUnit, Time = T, R = R>,
711        I: IntoIterator<Item = &'a B> + Clone,
712    {
713        let key_capacity = batches.clone().into_iter().map(|b| b.key_count()).sum();
714        let key_filter = if collect_roaring_metadata() {
715            let filter_plan = FilterPlan::from_batches(batches.clone());
716            filter_plan.map_or_else(
717                || FilterPlan::<K>::decide_filter(None, key_capacity),
718                |filter_plan| FilterPlan::decide_filter(Some(&filter_plan), key_capacity),
719            )
720        } else {
721            FilterPlan::<K>::decide_filter(None, key_capacity)
722        };
723        Self {
724            factories: factories.clone(),
725            writer: Writer2::new(
726                &factories.factories0,
727                &factories.factories1,
728                Runtime::buffer_cache,
729                &*Runtime::storage_backend().unwrap_storage(),
730                Runtime::file_writer_parameters(),
731                key_filter,
732            )
733            .unwrap_storage(),
734            key: factories.opt_key_factory.default_box(),
735            num_tuples: 0,
736            stats: BatchMetadata::default(),
737            touched_window_counter: collect_roaring_metadata().then(TouchedWindowCounter::default),
738        }
739    }
740
741    fn push_key(&mut self, key: &K) {
742        if let Some(counter) = self.touched_window_counter.as_mut()
743            && !counter.push_key(key)
744        {
745            self.touched_window_counter = None;
746        }
747        self.writer.write0((key, &())).unwrap_storage();
748    }
749
750    fn push_val(&mut self, _val: &DynUnit) {}
751
752    fn push_time_diff(&mut self, time: &T, weight: &R) {
753        debug_assert!(!weight.is_zero());
754        self.writer.write1((time, weight)).unwrap_storage();
755        self.num_tuples += 1;
756    }
757
758    fn done(mut self) -> FileKeyBatch<K, T, R> {
759        self.stats.touched_window_count = self
760            .touched_window_counter
761            .map(TouchedWindowCounter::finish)
762            .unwrap_or_default();
763        let (file, filters) = self.writer.into_reader(self.stats).unwrap_storage();
764        let file = Arc::new(file);
765        FileKeyBatch::from_parts(self.factories, file, filters)
766    }
767
768    fn num_keys(&self) -> usize {
769        self.writer.n_rows() as usize
770    }
771
772    fn num_tuples(&self) -> usize {
773        self.num_tuples
774    }
775}
776
777impl<K, T, R> Archive for FileKeyBatch<K, T, R>
778where
779    K: DataTrait + ?Sized,
780    T: Timestamp,
781    R: WeightTrait + ?Sized,
782{
783    type Archived = ();
784    type Resolver = ();
785
786    unsafe fn resolve(&self, _pos: usize, _resolver: Self::Resolver, _out: *mut Self::Archived) {
787        unimplemented!();
788    }
789}
790
791impl<K, T, R, S> Serialize<S> for FileKeyBatch<K, T, R>
792where
793    K: DataTrait + ?Sized,
794    T: Timestamp,
795    R: WeightTrait + ?Sized,
796    S: Serializer + ?Sized,
797{
798    fn serialize(&self, _serializer: &mut S) -> Result<Self::Resolver, S::Error> {
799        unimplemented!();
800    }
801}
802
803impl<K, T, R, D> Deserialize<FileKeyBatch<K, T, R>, D> for Archived<FileKeyBatch<K, T, R>>
804where
805    K: DataTrait + ?Sized,
806    T: Timestamp,
807    R: WeightTrait + ?Sized,
808    D: Fallible,
809{
810    fn deserialize(&self, _deserializer: &mut D) -> Result<FileKeyBatch<K, T, R>, D::Error> {
811        unimplemented!();
812    }
813}