1use crate::{
7 db::{
8 data::{CanonicalRow, RawDataStoreKey, RawRow},
9 direction::Direction,
10 ordered_overlay::{OrderedOverlayEntry, ordered_overlay_entries},
11 positioned_overlay::{
12 JournalOverlayPosition, PositionedOverlayMetadata, PositionedOverlayRetirement,
13 },
14 },
15 types::EntityTag,
16};
17use ic_memory::{DefaultMemoryImpl, Memory, RuntimeMemory};
18use ic_storage::adapter::StableBTreeMap;
19#[cfg(test)]
20use std::cell::Cell;
21use std::collections::{BTreeMap as HeapBTreeMap, BTreeSet, btree_map::Entry};
22use std::convert::Infallible;
23use std::ops::{Bound, RangeBounds};
24
25#[cfg(test)]
26thread_local! {
27 static DATA_STORE_GET_CALL_COUNT: Cell<u64> = const { Cell::new(0) };
28}
29
30#[cfg(test)]
31fn record_data_store_get_call() {
32 DATA_STORE_GET_CALL_COUNT.with(|count| {
33 count.set(count.get().saturating_add(1));
34 });
35}
36
37pub struct DataStore {
47 backend: DataStoreBackend,
48 generation: u64,
49 entity_cardinality: EntityCardinality,
50}
51
52enum DataStoreBackend {
53 Heap(HeapBTreeMap<RawDataStoreKey, RawRow>),
54 Journaled {
55 canonical: StableBTreeMap<RawDataStoreKey, RawRow, DefaultMemoryImpl>,
56 live: HeapBTreeMap<RawDataStoreKey, RawRow>,
57 tombstones: BTreeSet<RawDataStoreKey>,
58 positions: PositionedOverlayMetadata<RawDataStoreKey>,
59 entity_cardinality_delta: EntityCardinalityDelta,
60 },
61}
62
63#[cfg(feature = "migration")]
65pub(in crate::db) struct PreparedDataPositionPublication {
66 keys: Vec<RawDataStoreKey>,
67 position: JournalOverlayPosition,
68}
69
70pub(in crate::db) struct PreparedDataPositionRetirement {
72 entries: Vec<(RawDataStoreKey, PositionedOverlayRetirement)>,
73}
74
75pub(in crate::db) enum StoredRowRead<'a> {
81 Missing,
82 Borrowed(&'a RawRow),
83 Owned(RawRow),
84}
85
86impl StoredRowRead<'_> {
87 #[must_use]
89 pub(in crate::db) const fn as_row(&self) -> Option<&RawRow> {
90 match self {
91 Self::Missing => None,
92 Self::Borrowed(row) => Some(row),
93 Self::Owned(row) => Some(row),
94 }
95 }
96
97 #[must_use]
99 fn into_owned(self) -> Option<RawRow> {
100 match self {
101 Self::Missing => None,
102 Self::Borrowed(row) => Some(row.clone()),
103 Self::Owned(row) => Some(row),
104 }
105 }
106}
107
108#[derive(Clone, Copy, Debug, Eq, PartialEq)]
110pub(in crate::db) enum StoreVisit {
111 Continue,
112 Stop,
113}
114
115impl StoreVisit {
116 const fn should_stop(self) -> bool {
117 matches!(self, Self::Stop)
118 }
119}
120
121impl DataStore {
122 #[must_use]
124 pub const fn init_heap() -> Self {
125 Self {
126 backend: DataStoreBackend::Heap(HeapBTreeMap::new()),
127 generation: 0,
128 entity_cardinality: EntityCardinality::empty(),
129 }
130 }
131
132 #[must_use]
142 pub fn init_journaled(memory: RuntimeMemory<DefaultMemoryImpl>) -> Self {
143 let canonical = if memory.size() == 0 {
145 StableBTreeMap::init(memory)
146 } else {
147 StableBTreeMap::load(memory)
148 };
149 let entity_cardinality = if canonical.is_empty() {
150 EntityCardinality::empty()
151 } else {
152 EntityCardinality::unavailable()
153 };
154 Self {
155 backend: DataStoreBackend::Journaled {
156 canonical,
157 live: HeapBTreeMap::new(),
158 tombstones: BTreeSet::new(),
159 positions: PositionedOverlayMetadata::new(),
160 entity_cardinality_delta: EntityCardinalityDelta::empty(),
161 },
162 generation: 0,
163 entity_cardinality,
167 }
168 }
169
170 pub(in crate::db) fn insert(
172 &mut self,
173 key: RawDataStoreKey,
174 row: CanonicalRow,
175 ) -> Option<RawRow> {
176 self.insert_raw(key, row.into_raw_row())
177 }
178
179 fn insert_raw(&mut self, key: RawDataStoreKey, row: RawRow) -> Option<RawRow> {
181 let entity = key.entity_tag_prefix();
184 let previous = match &mut self.backend {
185 DataStoreBackend::Heap(map) => map.insert(key, row),
186 DataStoreBackend::Journaled {
187 canonical,
188 live,
189 tombstones,
190 ..
191 } => {
192 let hidden = tombstones.remove(&key);
193 match live.entry(key) {
196 Entry::Occupied(mut entry) => {
197 let previous = entry.insert(row);
198 (!hidden).then_some(previous)
199 }
200 Entry::Vacant(entry) => {
201 let previous = if hidden {
202 None
203 } else {
204 canonical.get(entry.key())
205 };
206 entry.insert(row);
207 previous
208 }
209 }
210 }
211 };
212 self.entity_cardinality
213 .apply_insert(entity, previous.is_some());
214 self.apply_entity_overlay_delta(entity, previous.is_some(), true);
215 self.bump_generation();
216 previous
217 }
218
219 #[cfg(test)]
221 pub(in crate::db) fn insert_raw_for_test(
222 &mut self,
223 key: RawDataStoreKey,
224 row: RawRow,
225 ) -> Option<RawRow> {
226 self.insert_raw(key, row)
227 }
228
229 pub(in crate::db) fn remove(&mut self, key: &RawDataStoreKey) -> Option<RawRow> {
231 let previous = match &mut self.backend {
232 DataStoreBackend::Heap(map) => map.remove(key),
233 DataStoreBackend::Journaled {
234 canonical,
235 live,
236 tombstones,
237 ..
238 } => {
239 let previous = live.remove(key);
240 let previous = if tombstones.contains(key) {
241 None
242 } else {
243 previous.or_else(|| canonical.get(key))
244 };
245 tombstones.insert(key.clone());
246 previous
247 }
248 };
249 self.entity_cardinality
250 .apply_remove(key.entity_tag_prefix(), previous.is_some());
251 self.apply_entity_overlay_delta(key.entity_tag_prefix(), previous.is_some(), false);
252 self.bump_generation();
253 previous
254 }
255
256 pub(in crate::db) fn reset_journaled_live_projection(
259 &mut self,
260 ) -> Result<(), crate::error::InternalError> {
261 let DataStoreBackend::Journaled {
262 canonical,
263 live,
264 tombstones,
265 positions,
266 entity_cardinality_delta,
267 } = &mut self.backend
268 else {
269 return Err(crate::error::InternalError::store_invariant());
270 };
271
272 live.clear();
273 tombstones.clear();
274 positions.clear();
275 *entity_cardinality_delta = EntityCardinalityDelta::empty();
276 self.entity_cardinality = if canonical.is_empty() {
277 EntityCardinality::empty()
278 } else {
279 EntityCardinality::unavailable()
280 };
281 self.bump_generation();
282
283 Ok(())
284 }
285
286 #[cfg(any(test, feature = "migration"))]
288 pub(in crate::db) fn apply_recovered_journal_put(
289 &mut self,
290 key: RawDataStoreKey,
291 row: RawRow,
292 ) -> Result<(), crate::error::InternalError> {
293 let previous_present = self.contains(&key);
295 let DataStoreBackend::Journaled {
296 live, tombstones, ..
297 } = &mut self.backend
298 else {
299 return Err(crate::error::InternalError::store_invariant());
300 };
301
302 tombstones.remove(&key);
303 let entity = key.entity_tag_prefix();
304 live.insert(key, row);
305 self.entity_cardinality
306 .apply_insert(entity, previous_present);
307 self.apply_entity_overlay_delta(entity, previous_present, true);
308 self.bump_generation();
309
310 Ok(())
311 }
312
313 pub(in crate::db) fn publish_preflighted_journal_entry(
315 &mut self,
316 key: RawDataStoreKey,
317 row: Option<RawRow>,
318 position: JournalOverlayPosition,
319 ) -> Result<(), crate::error::InternalError> {
320 let previous_present = self.contains(&key);
322 let DataStoreBackend::Journaled {
323 live,
324 tombstones,
325 positions,
326 entity_cardinality_delta,
327 ..
328 } = &mut self.backend
329 else {
330 return Err(crate::error::InternalError::store_invariant());
331 };
332 let next_present = row.is_some();
333 let entity = key.entity_tag_prefix();
334 if let Some(row) = row {
335 tombstones.remove(&key);
336 live.insert(key.clone(), row);
337 self.entity_cardinality
338 .apply_insert(entity, previous_present);
339 } else {
340 live.remove(&key);
341 tombstones.insert(key.clone());
342 self.entity_cardinality
343 .apply_remove(entity, previous_present);
344 }
345 entity_cardinality_delta.apply_presence_transition(entity, previous_present, next_present);
346 positions.publish_preflighted(key, position);
347 self.bump_generation();
348
349 Ok(())
350 }
351
352 #[cfg(test)]
354 pub(in crate::db) fn publish_positioned_journal_entry(
355 &mut self,
356 key: RawDataStoreKey,
357 row: Option<RawRow>,
358 position: JournalOverlayPosition,
359 ) -> Result<(), crate::error::InternalError> {
360 self.preflight_positioned_journal_entry(&key, position)?;
361 self.publish_preflighted_journal_entry(key, row, position)
362 }
363
364 pub(in crate::db) fn preflight_positioned_journal_entry(
366 &self,
367 key: &RawDataStoreKey,
368 position: JournalOverlayPosition,
369 ) -> Result<(), crate::error::InternalError> {
370 let DataStoreBackend::Journaled { positions, .. } = &self.backend else {
371 return Err(crate::error::InternalError::store_invariant());
372 };
373 positions.preflight_publish(key, position)
374 }
375
376 #[cfg(feature = "migration")]
378 pub(in crate::db) fn prepare_position_publication(
379 &self,
380 keys: impl IntoIterator<Item = RawDataStoreKey>,
381 position: JournalOverlayPosition,
382 ) -> Result<PreparedDataPositionPublication, crate::error::InternalError> {
383 let DataStoreBackend::Journaled { positions, .. } = &self.backend else {
384 return Err(crate::error::InternalError::store_invariant());
385 };
386 let keys = keys.into_iter().collect::<BTreeSet<_>>();
387 for key in &keys {
388 positions.preflight_publish(key, position)?;
389 }
390 Ok(PreparedDataPositionPublication {
391 keys: keys.into_iter().collect(),
392 position,
393 })
394 }
395
396 #[cfg(feature = "migration")]
398 pub(in crate::db) fn publish_prepared_positions(
399 &mut self,
400 prepared: PreparedDataPositionPublication,
401 ) {
402 let DataStoreBackend::Journaled { positions, .. } = &mut self.backend else {
403 debug_assert!(false, "preflighted row positions require a journaled store");
404 return;
405 };
406 for key in prepared.keys {
407 positions.publish_preflighted(key, prepared.position);
408 }
409 }
410
411 pub(in crate::db) fn prepare_position_retirement(
413 &self,
414 keys: impl IntoIterator<Item = RawDataStoreKey>,
415 position: JournalOverlayPosition,
416 ) -> Result<PreparedDataPositionRetirement, crate::error::InternalError> {
417 let DataStoreBackend::Journaled { positions, .. } = &self.backend else {
418 return Err(crate::error::InternalError::store_invariant());
419 };
420 let entries = keys
421 .into_iter()
422 .collect::<BTreeSet<_>>()
423 .into_iter()
424 .map(|key| {
425 positions
426 .preflight_retirement(&key, position)
427 .map(|retirement| (key, retirement))
428 })
429 .collect::<Result<Vec<_>, _>>()?;
430 Ok(PreparedDataPositionRetirement { entries })
431 }
432
433 pub(in crate::db) fn apply_prepared_position_retirement(
435 &mut self,
436 prepared: PreparedDataPositionRetirement,
437 ) {
438 let DataStoreBackend::Journaled {
439 live,
440 tombstones,
441 positions,
442 ..
443 } = &mut self.backend
444 else {
445 debug_assert!(
446 false,
447 "preflighted row retirement requires a journaled store"
448 );
449 return;
450 };
451 positions.retire_overlays_preflighted(prepared.entries, live, tombstones);
452 }
453
454 pub(in crate::db) fn fold_recovered_journal_put(
456 &mut self,
457 key: RawDataStoreKey,
458 row: RawRow,
459 ) -> Result<Option<RawRow>, crate::error::InternalError> {
460 let DataStoreBackend::Journaled {
461 canonical,
462 live,
463 tombstones,
464 ..
465 } = &mut self.backend
466 else {
467 return Err(crate::error::InternalError::store_invariant());
468 };
469
470 let visible = !live.contains_key(&key) && !tombstones.contains(&key);
471 let entity = key.entity_tag_prefix();
472 let previous = canonical.insert(key, row);
473 if visible {
474 self.entity_cardinality
475 .apply_insert(entity, previous.is_some());
476 } else if previous.is_none() {
477 self.apply_entity_overlay_delta(entity, true, false);
478 }
479 self.bump_generation();
480
481 Ok(previous)
482 }
483
484 pub(in crate::db) fn fold_recovered_journal_delete(
486 &mut self,
487 key: &RawDataStoreKey,
488 ) -> Result<Option<RawRow>, crate::error::InternalError> {
489 let DataStoreBackend::Journaled {
490 canonical,
491 live,
492 tombstones,
493 ..
494 } = &mut self.backend
495 else {
496 return Err(crate::error::InternalError::store_invariant());
497 };
498
499 let visible = !live.contains_key(key) && !tombstones.contains(key);
500 let previous = canonical.remove(key);
501 if visible {
502 self.entity_cardinality
503 .apply_remove(key.entity_tag_prefix(), previous.is_some());
504 } else if previous.is_some() {
505 self.apply_entity_overlay_delta(key.entity_tag_prefix(), false, true);
506 }
507 self.bump_generation();
508
509 Ok(previous)
510 }
511
512 pub(in crate::db) fn preflight_fold_recovered_journal(
514 &self,
515 ) -> Result<(), crate::error::InternalError> {
516 match self.backend {
517 DataStoreBackend::Journaled { .. } => Ok(()),
518 DataStoreBackend::Heap(_) => Err(crate::error::InternalError::store_invariant()),
519 }
520 }
521
522 pub(in crate::db) fn get(&self, key: &RawDataStoreKey) -> Option<RawRow> {
524 self.read(key).into_owned()
525 }
526
527 pub(in crate::db) fn get_canonical(&self, key: &RawDataStoreKey) -> Option<RawRow> {
533 match &self.backend {
534 DataStoreBackend::Heap(map) => map.get(key).cloned(),
535 DataStoreBackend::Journaled { canonical, .. } => canonical.get(key),
536 }
537 }
538
539 #[must_use]
542 pub(in crate::db) fn contains_canonical(&self, key: &RawDataStoreKey) -> bool {
543 match &self.backend {
544 DataStoreBackend::Heap(map) => map.contains_key(key),
545 DataStoreBackend::Journaled { canonical, .. } => canonical.contains_key(key),
546 }
547 }
548
549 pub(in crate::db) fn read<'a>(&'a self, key: &RawDataStoreKey) -> StoredRowRead<'a> {
551 #[cfg(test)]
552 record_data_store_get_call();
553
554 match &self.backend {
555 DataStoreBackend::Heap(map) => map
556 .get(key)
557 .map_or(StoredRowRead::Missing, StoredRowRead::Borrowed),
558 DataStoreBackend::Journaled {
559 canonical,
560 live,
561 tombstones,
562 ..
563 } => {
564 if tombstones.contains(key) {
565 StoredRowRead::Missing
566 } else if let Some(row) = live.get(key) {
567 StoredRowRead::Borrowed(row)
568 } else {
569 canonical
570 .get(key)
571 .map_or(StoredRowRead::Missing, StoredRowRead::Owned)
572 }
573 }
574 }
575 }
576
577 #[must_use]
579 pub(in crate::db) fn contains(&self, key: &RawDataStoreKey) -> bool {
580 match &self.backend {
581 DataStoreBackend::Heap(map) => map.contains_key(key),
582 DataStoreBackend::Journaled {
583 canonical,
584 live,
585 tombstones,
586 ..
587 } => {
588 !tombstones.contains(key) && (live.contains_key(key) || canonical.contains_key(key))
589 }
590 }
591 }
592
593 #[must_use]
598 pub(in crate::db) fn is_empty(&self) -> bool {
599 let mut empty = true;
600 let _: Result<(), Infallible> = self.visit_key_range(.., |_key| {
601 empty = false;
602 Ok(StoreVisit::Stop)
603 });
604 empty
605 }
606
607 #[cfg(test)]
609 #[must_use]
610 pub(in crate::db) fn len(&self) -> u64 {
611 match &self.backend {
612 DataStoreBackend::Heap(map) => u64::try_from(map.len()).unwrap_or(u64::MAX),
613 DataStoreBackend::Journaled { .. } => {
614 let mut count = 0_u64;
615 let _: Result<(), Infallible> = self.visit_entries(|_key, _row| {
616 count = count.saturating_add(1);
617 Ok(StoreVisit::Continue)
618 });
619 count
620 }
621 }
622 }
623
624 #[must_use]
626 pub(in crate::db) const fn generation(&self) -> u64 {
627 self.generation
628 }
629
630 #[must_use]
632 pub(in crate::db) fn exact_entity_count(&self, entity: EntityTag) -> Option<u64> {
633 self.entity_cardinality.exact_count(entity)
634 }
635
636 #[must_use]
638 pub(in crate::db) fn exact_entity_cardinality_delta(&self, entity: EntityTag) -> Option<i64> {
639 match &self.backend {
640 DataStoreBackend::Heap(_) => Some(0),
641 DataStoreBackend::Journaled {
642 entity_cardinality_delta,
643 ..
644 } => entity_cardinality_delta.exact_delta(entity),
645 }
646 }
647
648 pub(in crate::db) fn visit_entries<E>(
650 &self,
651 mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
652 ) -> Result<(), E> {
653 match &self.backend {
654 DataStoreBackend::Heap(map) => {
655 for (key, row) in map {
656 if visitor(key, row)?.should_stop() {
657 break;
658 }
659 }
660 }
661 DataStoreBackend::Journaled { .. } => Self::visit_journaled_entries_in_bounds(
662 &self.backend,
663 (Bound::Unbounded, Bound::Unbounded),
664 &mut visitor,
665 )?,
666 }
667
668 Ok(())
669 }
670
671 pub(in crate::db) fn visit_canonical_entries_after(
680 &self,
681 checkpoint: Option<&RawDataStoreKey>,
682 mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<bool, crate::error::InternalError>,
683 ) -> Result<Option<RawDataStoreKey>, crate::error::InternalError> {
684 let lower = checkpoint.map_or(Bound::Unbounded, Bound::Excluded);
685 match &self.backend {
686 DataStoreBackend::Heap(map) => {
687 let mut previous: Option<&RawDataStoreKey> = None;
688 for (key, row) in map.range((lower, Bound::Unbounded)) {
689 if visitor(key, row)? {
690 return Ok(previous.cloned());
691 }
692 previous = Some(key);
693 }
694 Ok(None)
695 }
696 DataStoreBackend::Journaled { canonical, .. } => {
697 let mut previous = None;
698 let mut stopped = false;
699 for entry in canonical.range((lower, Bound::Unbounded)) {
700 if visitor(entry.key(), &entry.value())? {
701 stopped = true;
702 break;
703 }
704 previous = Some(entry);
705 }
706 Ok(if stopped {
707 previous.map(|entry| entry.key().clone())
708 } else {
709 None
710 })
711 }
712 }
713 }
714
715 pub(in crate::db) fn canonical_is_empty(&self) -> Result<bool, crate::error::InternalError> {
720 match &self.backend {
721 DataStoreBackend::Journaled { canonical, .. } => Ok(canonical.is_empty()),
722 DataStoreBackend::Heap(_) => Err(crate::error::InternalError::store_invariant()),
723 }
724 }
725
726 pub(in crate::db) fn visit_range<E>(
728 &self,
729 key_range: impl RangeBounds<RawDataStoreKey>,
730 mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
731 ) -> Result<(), E> {
732 let bounds = (key_range.start_bound(), key_range.end_bound());
733 match &self.backend {
734 DataStoreBackend::Heap(map) => {
735 for (key, row) in map.range(bounds) {
736 if visitor(key, row)?.should_stop() {
737 break;
738 }
739 }
740 }
741 DataStoreBackend::Journaled { .. } => {
742 Self::visit_journaled_entries_in_bounds(&self.backend, bounds, &mut visitor)?;
743 }
744 }
745
746 Ok(())
747 }
748
749 pub(in crate::db) fn try_visit_range_with_row_preflight<E>(
756 &self,
757 key_range: impl RangeBounds<RawDataStoreKey>,
758 direction: Direction,
759 mut preflight: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
760 mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
761 ) -> Result<Option<bool>, E> {
762 let bounds = (key_range.start_bound(), key_range.end_bound());
763 let mut stopped = false;
764 match &self.backend {
765 DataStoreBackend::Heap(map) => {
766 let mut entries = map.range(bounds);
767 while let Some((key, row)) = match direction {
768 Direction::Asc => entries.next(),
769 Direction::Desc => entries.next_back(),
770 } {
771 if preflight(key)?.should_stop() {
772 stopped = true;
773 break;
774 }
775 if visitor(key, row)?.should_stop() {
776 stopped = true;
777 break;
778 }
779 }
780 }
781 DataStoreBackend::Journaled {
782 canonical,
783 live,
784 tombstones,
785 ..
786 } if canonical.is_empty() => {
787 let mut entries = live.range(bounds);
788 while let Some((key, row)) = match direction {
789 Direction::Asc => entries.next(),
790 Direction::Desc => entries.next_back(),
791 } {
792 if tombstones.contains(key) {
793 continue;
794 }
795 if preflight(key)?.should_stop() {
796 stopped = true;
797 break;
798 }
799 if visitor(key, row)?.should_stop() {
800 stopped = true;
801 break;
802 }
803 }
804 }
805 DataStoreBackend::Journaled {
806 canonical,
807 live,
808 tombstones,
809 ..
810 } if live.is_empty() && tombstones.is_empty() => {
811 let mut entries = canonical.range(bounds);
812 while let Some(entry) = match direction {
813 Direction::Asc => entries.next(),
814 Direction::Desc => entries.next_back(),
815 } {
816 if preflight(entry.key())?.should_stop() {
817 stopped = true;
818 break;
819 }
820 if visitor(entry.key(), &entry.value())?.should_stop() {
821 stopped = true;
822 break;
823 }
824 }
825 }
826 DataStoreBackend::Journaled { .. } => return Ok(None),
827 }
828
829 Ok(Some(!stopped))
830 }
831
832 pub(in crate::db) fn visit_key_range<E>(
839 &self,
840 key_range: impl RangeBounds<RawDataStoreKey>,
841 visitor: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
842 ) -> Result<(), E> {
843 self.visit_keys_in_bounds(
844 (key_range.start_bound(), key_range.end_bound()),
845 false,
846 visitor,
847 )
848 }
849
850 pub(in crate::db) fn visit_key_range_rev<E>(
852 &self,
853 key_range: impl RangeBounds<RawDataStoreKey>,
854 visitor: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
855 ) -> Result<(), E> {
856 self.visit_keys_in_bounds(
857 (key_range.start_bound(), key_range.end_bound()),
858 true,
859 visitor,
860 )
861 }
862
863 #[cfg(test)]
865 pub(in crate::db) fn memory_bytes(&self) -> u64 {
866 let mut bytes = 0u64;
868 let _: Result<(), Infallible> = self.visit_entries(|key, row| {
869 bytes = bytes.saturating_add(key.as_bytes().len() as u64 + row.len() as u64);
870 Ok(StoreVisit::Continue)
871 });
872 bytes
873 }
874
875 const fn bump_generation(&mut self) {
876 self.generation = self.generation.saturating_add(1);
877 }
878
879 #[cfg(test)]
880 fn rebuild_entity_cardinality_from_entries(&mut self) {
881 let mut cardinality = EntityCardinality::empty();
882 let _: Result<(), Infallible> = self.visit_entries(|key, _row| {
883 cardinality.apply_present_entity(key.entity_tag_prefix());
884 Ok(StoreVisit::Continue)
885 });
886 self.entity_cardinality = cardinality;
887 }
888
889 fn apply_entity_overlay_delta(
890 &mut self,
891 entity: Option<EntityTag>,
892 previous_present: bool,
893 next_present: bool,
894 ) {
895 let DataStoreBackend::Journaled {
896 entity_cardinality_delta,
897 ..
898 } = &mut self.backend
899 else {
900 return;
901 };
902 entity_cardinality_delta.apply_presence_transition(entity, previous_present, next_present);
903 }
904
905 #[cfg(test)]
907 pub(in crate::db) fn current_get_call_count() -> u64 {
908 DATA_STORE_GET_CALL_COUNT.with(Cell::get)
909 }
910
911 fn visit_keys_in_bounds<E>(
912 &self,
913 bounds: (Bound<&RawDataStoreKey>, Bound<&RawDataStoreKey>),
914 reverse: bool,
915 mut visitor: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
916 ) -> Result<(), E> {
917 match &self.backend {
918 DataStoreBackend::Heap(map) => {
919 if reverse {
920 for (key, _row) in map.range(bounds).rev() {
921 if visitor(key)?.should_stop() {
922 break;
923 }
924 }
925 } else {
926 for (key, _row) in map.range(bounds) {
927 if visitor(key)?.should_stop() {
928 break;
929 }
930 }
931 }
932 }
933 DataStoreBackend::Journaled { .. } => {
934 Self::visit_journaled_keys_in_bounds(&self.backend, bounds, reverse, &mut visitor)?;
935 }
936 }
937
938 Ok(())
939 }
940
941 fn visit_journaled_keys_in_bounds<E>(
944 backend: &DataStoreBackend,
945 bounds: (Bound<&RawDataStoreKey>, Bound<&RawDataStoreKey>),
946 reverse: bool,
947 visitor: &mut dyn FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
948 ) -> Result<(), E> {
949 let DataStoreBackend::Journaled {
950 canonical,
951 live,
952 tombstones,
953 ..
954 } = backend
955 else {
956 return Ok(());
957 };
958
959 if canonical.is_empty() {
960 if reverse {
961 for (key, _row) in live.range(bounds).rev() {
962 if visitor(key)?.should_stop() {
963 return Ok(());
964 }
965 }
966 } else {
967 for (key, _row) in live.range(bounds) {
968 if visitor(key)?.should_stop() {
969 return Ok(());
970 }
971 }
972 }
973 return Ok(());
974 }
975
976 if live.is_empty() && tombstones.is_empty() {
977 if reverse {
978 for entry in canonical.range(bounds).rev() {
979 if visitor(entry.key())?.should_stop() {
980 return Ok(());
981 }
982 }
983 } else {
984 for entry in canonical.range(bounds) {
985 if visitor(entry.key())?.should_stop() {
986 return Ok(());
987 }
988 }
989 }
990 return Ok(());
991 }
992
993 let direction = if reverse {
994 Direction::Desc
995 } else {
996 Direction::Asc
997 };
998 match direction {
999 Direction::Asc => {
1000 for entry in ordered_overlay_entries(
1001 canonical.range(bounds),
1002 live.range(bounds),
1003 direction,
1004 |entry| entry.key(),
1005 |entry| entry.0,
1006 tombstones,
1007 ) {
1008 let visit = match entry {
1009 OrderedOverlayEntry::Canonical(canonical_entry) => {
1010 visitor(canonical_entry.key())?
1011 }
1012 OrderedOverlayEntry::Live((key, _row)) => visitor(key)?,
1013 };
1014 if visit.should_stop() {
1015 return Ok(());
1016 }
1017 }
1018 }
1019 Direction::Desc => {
1020 for entry in ordered_overlay_entries(
1021 canonical.range(bounds).rev(),
1022 live.range(bounds).rev(),
1023 direction,
1024 |entry| entry.key(),
1025 |entry| entry.0,
1026 tombstones,
1027 ) {
1028 let visit = match entry {
1029 OrderedOverlayEntry::Canonical(canonical_entry) => {
1030 visitor(canonical_entry.key())?
1031 }
1032 OrderedOverlayEntry::Live((key, _row)) => visitor(key)?,
1033 };
1034 if visit.should_stop() {
1035 return Ok(());
1036 }
1037 }
1038 }
1039 }
1040
1041 Ok(())
1042 }
1043
1044 fn visit_journaled_entries_in_bounds<E>(
1047 backend: &DataStoreBackend,
1048 bounds: (Bound<&RawDataStoreKey>, Bound<&RawDataStoreKey>),
1049 visitor: &mut dyn FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
1050 ) -> Result<(), E> {
1051 let DataStoreBackend::Journaled {
1052 canonical,
1053 live,
1054 tombstones,
1055 ..
1056 } = backend
1057 else {
1058 return Ok(());
1059 };
1060
1061 if canonical.is_empty() {
1062 for (key, row) in live.range(bounds) {
1063 if visitor(key, row)?.should_stop() {
1064 return Ok(());
1065 }
1066 }
1067 return Ok(());
1068 }
1069
1070 if live.is_empty() && tombstones.is_empty() {
1071 for entry in canonical.range(bounds) {
1072 if visitor(entry.key(), &entry.value())?.should_stop() {
1073 return Ok(());
1074 }
1075 }
1076 return Ok(());
1077 }
1078
1079 for entry in ordered_overlay_entries(
1080 canonical.range(bounds),
1081 live.range(bounds),
1082 Direction::Asc,
1083 |entry| entry.key(),
1084 |entry| entry.0,
1085 tombstones,
1086 ) {
1087 let visit = match entry {
1088 OrderedOverlayEntry::Canonical(canonical_entry) => {
1089 visitor(canonical_entry.key(), &canonical_entry.value())?
1090 }
1091 OrderedOverlayEntry::Live((key, row)) => visitor(key, row)?,
1092 };
1093 if visit.should_stop() {
1094 return Ok(());
1095 }
1096 }
1097
1098 Ok(())
1099 }
1100}
1101
1102#[derive(Clone, Debug)]
1103struct EntityCardinality {
1104 counts: HeapBTreeMap<EntityTag, u64>,
1105 decodable: bool,
1106}
1107
1108#[derive(Clone, Debug)]
1109struct EntityCardinalityDelta {
1110 counts: HeapBTreeMap<EntityTag, i64>,
1111 decodable: bool,
1112}
1113
1114impl EntityCardinalityDelta {
1115 const fn empty() -> Self {
1116 Self {
1117 counts: HeapBTreeMap::new(),
1118 decodable: true,
1119 }
1120 }
1121
1122 fn exact_delta(&self, entity: EntityTag) -> Option<i64> {
1123 self.decodable
1124 .then(|| self.counts.get(&entity).copied().unwrap_or(0))
1125 }
1126
1127 fn apply_presence_transition(
1128 &mut self,
1129 entity: Option<EntityTag>,
1130 previous_present: bool,
1131 next_present: bool,
1132 ) {
1133 if !self.decodable || previous_present == next_present {
1134 return;
1135 }
1136 let Some(entity) = entity else {
1137 self.counts.clear();
1138 self.decodable = false;
1139 return;
1140 };
1141 let delta = if next_present { 1_i64 } else { -1_i64 };
1142 let count = self.counts.entry(entity).or_insert(0);
1143 let Some(next) = count.checked_add(delta) else {
1144 self.counts.clear();
1145 self.decodable = false;
1146 return;
1147 };
1148 *count = next;
1149 if *count == 0 {
1150 self.counts.remove(&entity);
1151 }
1152 }
1153}
1154
1155impl EntityCardinality {
1156 const fn empty() -> Self {
1157 Self {
1158 counts: HeapBTreeMap::new(),
1159 decodable: true,
1160 }
1161 }
1162
1163 const fn unavailable() -> Self {
1164 Self {
1165 counts: HeapBTreeMap::new(),
1166 decodable: false,
1167 }
1168 }
1169
1170 fn exact_count(&self, entity: EntityTag) -> Option<u64> {
1171 self.decodable
1172 .then(|| self.counts.get(&entity).copied().unwrap_or(0))
1173 }
1174
1175 fn apply_insert(&mut self, entity: Option<EntityTag>, previous_present: bool) {
1176 if previous_present {
1177 return;
1178 }
1179 self.apply_present_entity(entity);
1180 }
1181
1182 fn apply_remove(&mut self, entity: Option<EntityTag>, previous_present: bool) {
1183 if !previous_present {
1184 return;
1185 }
1186 self.apply_removed_entity(entity);
1187 }
1188
1189 fn apply_present_entity(&mut self, entity: Option<EntityTag>) {
1190 if !self.decodable {
1191 return;
1192 }
1193 let Some(entity) = entity else {
1194 self.invalidate();
1195 return;
1196 };
1197
1198 let count = self.counts.entry(entity).or_insert(0);
1199 *count = count.saturating_add(1);
1200 }
1201
1202 fn apply_removed_entity(&mut self, entity: Option<EntityTag>) {
1203 if !self.decodable {
1204 return;
1205 }
1206 let Some(entity) = entity else {
1207 self.invalidate();
1208 return;
1209 };
1210
1211 if let Some(count) = self.counts.get_mut(&entity) {
1212 *count = count.saturating_sub(1);
1213 if *count == 0 {
1214 self.counts.remove(&entity);
1215 }
1216 }
1217 }
1218
1219 fn invalidate(&mut self) {
1220 self.counts.clear();
1221 self.decodable = false;
1222 }
1223}
1224
1225#[cfg(test)]
1226mod tests;