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#[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#[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 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#[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}