1use crate::{
7 db::{
8 data::{CanonicalRow, RawDataStoreKey, RawRow},
9 direction::Direction,
10 ordered_overlay::{OrderedOverlayEntry, OrderedOverlayVisit, visit_ordered_overlay},
11 positioned_overlay::{
12 JournalOverlayPosition, PositionedOverlayMetadata, PositionedOverlayRetirement,
13 },
14 },
15 types::EntityTag,
16};
17use ic_stable_structures::{
18 BTreeMap as StableBTreeMap, DefaultMemoryImpl, memory_manager::VirtualMemory,
19};
20#[cfg(all(feature = "sql", feature = "diagnostics"))]
21use std::cell::Cell;
22use std::collections::{BTreeMap as HeapBTreeMap, BTreeSet};
23use std::convert::Infallible;
24use std::ops::{Bound, RangeBounds};
25
26#[cfg(all(feature = "sql", feature = "diagnostics"))]
27thread_local! {
28 static DATA_STORE_GET_CALL_COUNT: Cell<u64> = const { Cell::new(0) };
29}
30
31#[cfg(all(feature = "sql", feature = "diagnostics"))]
32fn record_data_store_get_call() {
33 DATA_STORE_GET_CALL_COUNT.with(|count| {
34 count.set(count.get().saturating_add(1));
35 });
36}
37
38pub struct DataStore {
48 backend: DataStoreBackend,
49 generation: u64,
50 entity_cardinality: EntityCardinality,
51}
52
53enum DataStoreBackend {
54 Heap(HeapBTreeMap<RawDataStoreKey, RawRow>),
55 Journaled {
56 canonical: StableBTreeMap<RawDataStoreKey, RawRow, VirtualMemory<DefaultMemoryImpl>>,
57 live: HeapBTreeMap<RawDataStoreKey, RawRow>,
58 tombstones: BTreeSet<RawDataStoreKey>,
59 positions: PositionedOverlayMetadata<RawDataStoreKey>,
60 },
61}
62
63#[cfg(any(test, 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]
138 pub fn init_journaled(memory: VirtualMemory<DefaultMemoryImpl>) -> Self {
139 let canonical = StableBTreeMap::init(memory);
140 let entity_cardinality = if canonical.is_empty() {
141 EntityCardinality::empty()
142 } else {
143 EntityCardinality::unavailable()
144 };
145 Self {
146 backend: DataStoreBackend::Journaled {
147 canonical,
148 live: HeapBTreeMap::new(),
149 tombstones: BTreeSet::new(),
150 positions: PositionedOverlayMetadata::new(),
151 },
152 generation: 0,
153 entity_cardinality,
157 }
158 }
159
160 pub(in crate::db) fn insert(
162 &mut self,
163 key: RawDataStoreKey,
164 row: CanonicalRow,
165 ) -> Option<RawRow> {
166 let row = row.into_raw_row();
167 let previous_journaled = if matches!(self.backend, DataStoreBackend::Journaled { .. }) {
168 self.get(&key)
169 } else {
170 None
171 };
172 let cardinality_key = key.clone();
173 let previous = match &mut self.backend {
174 DataStoreBackend::Heap(map) => map.insert(key, row),
175 DataStoreBackend::Journaled {
176 live, tombstones, ..
177 } => {
178 tombstones.remove(&key);
179 live.insert(key, row);
180 previous_journaled
181 }
182 };
183 self.entity_cardinality
184 .apply_insert(&cardinality_key, previous.as_ref());
185 self.bump_generation();
186 previous
187 }
188
189 #[cfg(test)]
191 pub(in crate::db) fn insert_raw_for_test(
192 &mut self,
193 key: RawDataStoreKey,
194 row: RawRow,
195 ) -> Option<RawRow> {
196 let previous_journaled = if matches!(self.backend, DataStoreBackend::Journaled { .. }) {
197 self.get(&key)
198 } else {
199 None
200 };
201 let cardinality_key = key.clone();
202 let previous = match &mut self.backend {
203 DataStoreBackend::Heap(map) => map.insert(key, row),
204 DataStoreBackend::Journaled {
205 live, tombstones, ..
206 } => {
207 tombstones.remove(&key);
208 live.insert(key, row);
209 previous_journaled
210 }
211 };
212 self.entity_cardinality
213 .apply_insert(&cardinality_key, previous.as_ref());
214 self.bump_generation();
215 previous
216 }
217
218 pub(in crate::db) fn remove(&mut self, key: &RawDataStoreKey) -> Option<RawRow> {
220 let previous_journaled = if matches!(self.backend, DataStoreBackend::Journaled { .. }) {
221 self.get(key)
222 } else {
223 None
224 };
225 let previous = match &mut self.backend {
226 DataStoreBackend::Heap(map) => map.remove(key),
227 DataStoreBackend::Journaled {
228 live, tombstones, ..
229 } => {
230 live.remove(key);
231 tombstones.insert(key.clone());
232 previous_journaled
233 }
234 };
235 self.entity_cardinality.apply_remove(key, previous.as_ref());
236 self.bump_generation();
237 previous
238 }
239
240 pub(in crate::db) fn reset_journaled_live_projection(
243 &mut self,
244 ) -> Result<(), crate::error::InternalError> {
245 let DataStoreBackend::Journaled {
246 canonical,
247 live,
248 tombstones,
249 positions,
250 } = &mut self.backend
251 else {
252 return Err(crate::error::InternalError::store_invariant());
253 };
254
255 live.clear();
256 tombstones.clear();
257 positions.clear();
258 self.entity_cardinality = if canonical.is_empty() {
259 EntityCardinality::empty()
260 } else {
261 EntityCardinality::unavailable()
262 };
263 self.bump_generation();
264
265 Ok(())
266 }
267
268 pub(in crate::db) fn apply_recovered_journal_put(
270 &mut self,
271 key: RawDataStoreKey,
272 row: RawRow,
273 ) -> Result<Option<RawRow>, crate::error::InternalError> {
274 let DataStoreBackend::Journaled {
275 canonical,
276 live,
277 tombstones,
278 ..
279 } = &mut self.backend
280 else {
281 return Err(crate::error::InternalError::store_invariant());
282 };
283
284 let previous = if tombstones.contains(&key) {
285 None
286 } else {
287 live.get(&key).cloned().or_else(|| canonical.get(&key))
288 };
289 tombstones.remove(&key);
290 let cardinality_key = key.clone();
291 live.insert(key, row);
292 self.entity_cardinality
293 .apply_insert(&cardinality_key, previous.as_ref());
294 self.bump_generation();
295
296 Ok(previous)
297 }
298
299 pub(in crate::db) fn apply_recovered_journal_delete(
301 &mut self,
302 key: &RawDataStoreKey,
303 ) -> Result<Option<RawRow>, crate::error::InternalError> {
304 let DataStoreBackend::Journaled {
305 canonical,
306 live,
307 tombstones,
308 ..
309 } = &mut self.backend
310 else {
311 return Err(crate::error::InternalError::store_invariant());
312 };
313
314 let previous = if tombstones.contains(key) {
315 None
316 } else {
317 live.get(key).cloned().or_else(|| canonical.get(key))
318 };
319 live.remove(key);
320 tombstones.insert(key.clone());
321 self.entity_cardinality.apply_remove(key, previous.as_ref());
322 self.bump_generation();
323
324 Ok(previous)
325 }
326
327 pub(in crate::db) fn publish_preflighted_journal_entry(
329 &mut self,
330 key: RawDataStoreKey,
331 row: Option<RawRow>,
332 position: JournalOverlayPosition,
333 ) -> Result<Option<RawRow>, crate::error::InternalError> {
334 let DataStoreBackend::Journaled {
335 canonical,
336 live,
337 tombstones,
338 positions,
339 } = &mut self.backend
340 else {
341 return Err(crate::error::InternalError::store_invariant());
342 };
343 let previous = if tombstones.contains(&key) {
344 None
345 } else {
346 live.get(&key).cloned().or_else(|| canonical.get(&key))
347 };
348 let cardinality_key = key.clone();
349 if let Some(row) = row {
350 tombstones.remove(&key);
351 live.insert(key.clone(), row);
352 self.entity_cardinality
353 .apply_insert(&cardinality_key, previous.as_ref());
354 } else {
355 live.remove(&key);
356 tombstones.insert(key.clone());
357 self.entity_cardinality
358 .apply_remove(&cardinality_key, previous.as_ref());
359 }
360 positions.publish_preflighted(key, position);
361 self.bump_generation();
362
363 Ok(previous)
364 }
365
366 #[cfg(test)]
368 pub(in crate::db) fn publish_positioned_journal_entry(
369 &mut self,
370 key: RawDataStoreKey,
371 row: Option<RawRow>,
372 position: JournalOverlayPosition,
373 ) -> Result<Option<RawRow>, crate::error::InternalError> {
374 self.preflight_positioned_journal_entry(&key, position)?;
375 self.publish_preflighted_journal_entry(key, row, position)
376 }
377
378 pub(in crate::db) fn preflight_positioned_journal_entry(
380 &self,
381 key: &RawDataStoreKey,
382 position: JournalOverlayPosition,
383 ) -> Result<(), crate::error::InternalError> {
384 let DataStoreBackend::Journaled { positions, .. } = &self.backend else {
385 return Err(crate::error::InternalError::store_invariant());
386 };
387 positions.preflight_publish(key, position)
388 }
389
390 #[cfg(any(test, feature = "migration"))]
392 pub(in crate::db) fn prepare_position_publication(
393 &self,
394 keys: impl IntoIterator<Item = RawDataStoreKey>,
395 position: JournalOverlayPosition,
396 ) -> Result<PreparedDataPositionPublication, crate::error::InternalError> {
397 let DataStoreBackend::Journaled { positions, .. } = &self.backend else {
398 return Err(crate::error::InternalError::store_invariant());
399 };
400 let keys = keys.into_iter().collect::<BTreeSet<_>>();
401 for key in &keys {
402 positions.preflight_publish(key, position)?;
403 }
404 Ok(PreparedDataPositionPublication {
405 keys: keys.into_iter().collect(),
406 position,
407 })
408 }
409
410 #[cfg(any(test, feature = "migration"))]
412 pub(in crate::db) fn publish_prepared_positions(
413 &mut self,
414 prepared: PreparedDataPositionPublication,
415 ) {
416 let DataStoreBackend::Journaled { positions, .. } = &mut self.backend else {
417 debug_assert!(false, "preflighted row positions require a journaled store");
418 return;
419 };
420 for key in prepared.keys {
421 positions.publish_preflighted(key, prepared.position);
422 }
423 }
424
425 pub(in crate::db) fn prepare_position_retirement(
427 &self,
428 keys: impl IntoIterator<Item = RawDataStoreKey>,
429 position: JournalOverlayPosition,
430 ) -> Result<PreparedDataPositionRetirement, crate::error::InternalError> {
431 let DataStoreBackend::Journaled { positions, .. } = &self.backend else {
432 return Err(crate::error::InternalError::store_invariant());
433 };
434 let entries = keys
435 .into_iter()
436 .collect::<BTreeSet<_>>()
437 .into_iter()
438 .map(|key| {
439 positions
440 .preflight_retirement(&key, position)
441 .map(|retirement| (key, retirement))
442 })
443 .collect::<Result<Vec<_>, _>>()?;
444 Ok(PreparedDataPositionRetirement { entries })
445 }
446
447 pub(in crate::db) fn apply_prepared_position_retirement(
449 &mut self,
450 prepared: PreparedDataPositionRetirement,
451 ) {
452 let DataStoreBackend::Journaled {
453 live,
454 tombstones,
455 positions,
456 ..
457 } = &mut self.backend
458 else {
459 debug_assert!(
460 false,
461 "preflighted row retirement requires a journaled store"
462 );
463 return;
464 };
465 for (key, retirement) in prepared.entries {
466 if retirement == PositionedOverlayRetirement::Exact {
467 live.remove(&key);
468 tombstones.remove(&key);
469 positions.retire_preflighted(&key, retirement);
470 }
471 }
472 }
473
474 #[cfg(test)]
475 fn retire_positioned_journal_effect(
476 &mut self,
477 key: &RawDataStoreKey,
478 position: JournalOverlayPosition,
479 ) -> Result<PositionedOverlayRetirement, crate::error::InternalError> {
480 let DataStoreBackend::Journaled { positions, .. } = &self.backend else {
481 return Err(crate::error::InternalError::store_invariant());
482 };
483 let retirement = positions.preflight_retirement(key, position)?;
484 let prepared = PreparedDataPositionRetirement {
485 entries: vec![(key.clone(), retirement)],
486 };
487 self.apply_prepared_position_retirement(prepared);
488 Ok(retirement)
489 }
490
491 pub(in crate::db) fn fold_recovered_journal_put(
493 &mut self,
494 key: RawDataStoreKey,
495 row: RawRow,
496 ) -> Result<Option<RawRow>, crate::error::InternalError> {
497 let DataStoreBackend::Journaled {
498 canonical,
499 live,
500 tombstones,
501 ..
502 } = &mut self.backend
503 else {
504 return Err(crate::error::InternalError::store_invariant());
505 };
506
507 let visible = !live.contains_key(&key) && !tombstones.contains(&key);
508 let cardinality_key = key.clone();
509 let previous = canonical.insert(key, row);
510 if visible {
511 self.entity_cardinality
512 .apply_insert(&cardinality_key, previous.as_ref());
513 }
514 self.bump_generation();
515
516 Ok(previous)
517 }
518
519 pub(in crate::db) fn fold_recovered_journal_delete(
521 &mut self,
522 key: &RawDataStoreKey,
523 ) -> Result<Option<RawRow>, crate::error::InternalError> {
524 let DataStoreBackend::Journaled {
525 canonical,
526 live,
527 tombstones,
528 ..
529 } = &mut self.backend
530 else {
531 return Err(crate::error::InternalError::store_invariant());
532 };
533
534 let visible = !live.contains_key(key) && !tombstones.contains(key);
535 let previous = canonical.remove(key);
536 if visible {
537 self.entity_cardinality.apply_remove(key, previous.as_ref());
538 }
539 self.bump_generation();
540
541 Ok(previous)
542 }
543
544 pub(in crate::db) fn preflight_fold_recovered_journal(
546 &self,
547 ) -> Result<(), crate::error::InternalError> {
548 match self.backend {
549 DataStoreBackend::Journaled { .. } => Ok(()),
550 DataStoreBackend::Heap(_) => Err(crate::error::InternalError::store_invariant()),
551 }
552 }
553
554 pub(in crate::db) fn get(&self, key: &RawDataStoreKey) -> Option<RawRow> {
556 self.read(key).into_owned()
557 }
558
559 pub(in crate::db) fn get_canonical(&self, key: &RawDataStoreKey) -> Option<RawRow> {
565 match &self.backend {
566 DataStoreBackend::Heap(map) => map.get(key).cloned(),
567 DataStoreBackend::Journaled { canonical, .. } => canonical.get(key),
568 }
569 }
570
571 pub(in crate::db) fn read<'a>(&'a self, key: &RawDataStoreKey) -> StoredRowRead<'a> {
573 #[cfg(all(feature = "sql", feature = "diagnostics"))]
574 record_data_store_get_call();
575
576 match &self.backend {
577 DataStoreBackend::Heap(map) => map
578 .get(key)
579 .map_or(StoredRowRead::Missing, StoredRowRead::Borrowed),
580 DataStoreBackend::Journaled {
581 canonical,
582 live,
583 tombstones,
584 ..
585 } => {
586 if tombstones.contains(key) {
587 StoredRowRead::Missing
588 } else if let Some(row) = live.get(key) {
589 StoredRowRead::Borrowed(row)
590 } else {
591 canonical
592 .get(key)
593 .map_or(StoredRowRead::Missing, StoredRowRead::Owned)
594 }
595 }
596 }
597 }
598
599 #[must_use]
601 pub(in crate::db) fn contains(&self, key: &RawDataStoreKey) -> bool {
602 match &self.backend {
603 DataStoreBackend::Heap(map) => map.contains_key(key),
604 DataStoreBackend::Journaled {
605 canonical,
606 live,
607 tombstones,
608 ..
609 } => {
610 !tombstones.contains(key)
611 && (live.contains_key(key) || canonical.get(key).is_some())
612 }
613 }
614 }
615
616 #[must_use]
618 pub(in crate::db) fn len(&self) -> u64 {
619 match &self.backend {
620 DataStoreBackend::Heap(map) => u64::try_from(map.len()).unwrap_or(u64::MAX),
621 DataStoreBackend::Journaled { .. } => {
622 let mut count = 0_u64;
623 let _: Result<(), Infallible> = self.visit_entries(|_key, _row| {
624 count = count.saturating_add(1);
625 Ok(StoreVisit::Continue)
626 });
627 count
628 }
629 }
630 }
631
632 #[must_use]
634 pub(in crate::db) const fn generation(&self) -> u64 {
635 self.generation
636 }
637
638 #[must_use]
640 pub(in crate::db) fn exact_entity_count(&self, entity: EntityTag) -> Option<u64> {
641 self.entity_cardinality.exact_count(entity)
642 }
643
644 pub(in crate::db) fn visit_entries<E>(
646 &self,
647 mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
648 ) -> Result<(), E> {
649 match &self.backend {
650 DataStoreBackend::Heap(map) => {
651 for (key, row) in map {
652 if visitor(key, row)?.should_stop() {
653 break;
654 }
655 }
656 }
657 DataStoreBackend::Journaled { .. } => Self::visit_journaled_entries_in_bounds(
658 &self.backend,
659 (Bound::Unbounded, Bound::Unbounded),
660 visitor,
661 )?,
662 }
663
664 Ok(())
665 }
666
667 pub(in crate::db) fn visit_range<E>(
669 &self,
670 key_range: impl RangeBounds<RawDataStoreKey>,
671 mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
672 ) -> Result<(), E> {
673 let bounds = Self::owned_range_bounds(&key_range);
674 match &self.backend {
675 DataStoreBackend::Heap(map) => {
676 for (key, row) in map.range((bounds.0.clone(), bounds.1)) {
677 if visitor(key, row)?.should_stop() {
678 break;
679 }
680 }
681 }
682 DataStoreBackend::Journaled { .. } => {
683 Self::visit_journaled_entries_in_bounds(&self.backend, bounds, visitor)?;
684 }
685 }
686
687 Ok(())
688 }
689
690 pub(in crate::db) fn try_visit_range_with_row_preflight<E>(
697 &self,
698 key_range: impl RangeBounds<RawDataStoreKey>,
699 mut preflight: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
700 mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
701 ) -> Result<Option<bool>, E> {
702 let bounds = Self::owned_range_bounds(&key_range);
703 let mut stopped = false;
704 match &self.backend {
705 DataStoreBackend::Heap(map) => {
706 for (key, row) in map.range((bounds.0.clone(), bounds.1)) {
707 if preflight(key)?.should_stop() {
708 stopped = true;
709 break;
710 }
711 if visitor(key, row)?.should_stop() {
712 stopped = true;
713 break;
714 }
715 }
716 }
717 DataStoreBackend::Journaled {
718 canonical,
719 live,
720 tombstones,
721 ..
722 } if canonical.is_empty() => {
723 for (key, row) in live.range((bounds.0.clone(), bounds.1)) {
724 if tombstones.contains(key) {
725 continue;
726 }
727 if preflight(key)?.should_stop() {
728 stopped = true;
729 break;
730 }
731 if visitor(key, row)?.should_stop() {
732 stopped = true;
733 break;
734 }
735 }
736 }
737 DataStoreBackend::Journaled {
738 canonical,
739 live,
740 tombstones,
741 ..
742 } if live.is_empty() && tombstones.is_empty() => {
743 for entry in canonical.range((bounds.0.clone(), bounds.1)) {
744 if preflight(entry.key())?.should_stop() {
745 stopped = true;
746 break;
747 }
748 if visitor(entry.key(), &entry.value())?.should_stop() {
749 stopped = true;
750 break;
751 }
752 }
753 }
754 DataStoreBackend::Journaled { .. } => return Ok(None),
755 }
756
757 Ok(Some(!stopped))
758 }
759
760 pub(in crate::db) fn visit_key_range<E>(
767 &self,
768 key_range: impl RangeBounds<RawDataStoreKey>,
769 visitor: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
770 ) -> Result<(), E> {
771 self.visit_keys_in_bounds(Self::owned_range_bounds(&key_range), false, visitor)
772 }
773
774 pub(in crate::db) fn visit_key_range_rev<E>(
776 &self,
777 key_range: impl RangeBounds<RawDataStoreKey>,
778 visitor: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
779 ) -> Result<(), E> {
780 self.visit_keys_in_bounds(Self::owned_range_bounds(&key_range), true, visitor)
781 }
782
783 pub(in crate::db) fn memory_bytes(&self) -> u64 {
785 let mut bytes = 0u64;
787 let _: Result<(), Infallible> = self.visit_entries(|key, row| {
788 bytes = bytes.saturating_add(key.as_bytes().len() as u64 + row.len() as u64);
789 Ok(StoreVisit::Continue)
790 });
791 bytes
792 }
793
794 const fn bump_generation(&mut self) {
795 self.generation = self.generation.saturating_add(1);
796 }
797
798 #[cfg(test)]
799 fn rebuild_entity_cardinality_from_entries(&mut self) {
800 let mut cardinality = EntityCardinality::empty();
801 let _: Result<(), Infallible> = self.visit_entries(|key, _row| {
802 cardinality.apply_present_key(key);
803 Ok(StoreVisit::Continue)
804 });
805 self.entity_cardinality = cardinality;
806 }
807
808 #[cfg(all(feature = "sql", feature = "diagnostics"))]
810 pub(in crate::db) fn current_get_call_count() -> u64 {
811 DATA_STORE_GET_CALL_COUNT.with(Cell::get)
812 }
813
814 fn owned_range_bounds(
815 key_range: &impl RangeBounds<RawDataStoreKey>,
816 ) -> (Bound<RawDataStoreKey>, Bound<RawDataStoreKey>) {
817 let lower = match key_range.start_bound() {
818 Bound::Included(key) => Bound::Included(key.clone()),
819 Bound::Excluded(key) => Bound::Excluded(key.clone()),
820 Bound::Unbounded => Bound::Unbounded,
821 };
822 let upper = match key_range.end_bound() {
823 Bound::Included(key) => Bound::Included(key.clone()),
824 Bound::Excluded(key) => Bound::Excluded(key.clone()),
825 Bound::Unbounded => Bound::Unbounded,
826 };
827
828 (lower, upper)
829 }
830
831 fn visit_keys_in_bounds<E>(
832 &self,
833 bounds: (Bound<RawDataStoreKey>, Bound<RawDataStoreKey>),
834 reverse: bool,
835 mut visitor: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
836 ) -> Result<(), E> {
837 match &self.backend {
838 DataStoreBackend::Heap(map) => {
839 if reverse {
840 for (key, _row) in map.range(bounds).rev() {
841 if visitor(key)?.should_stop() {
842 break;
843 }
844 }
845 } else {
846 for (key, _row) in map.range(bounds) {
847 if visitor(key)?.should_stop() {
848 break;
849 }
850 }
851 }
852 }
853 DataStoreBackend::Journaled { .. } => {
854 Self::visit_journaled_keys_in_bounds(&self.backend, bounds, reverse, visitor)?;
855 }
856 }
857
858 Ok(())
859 }
860
861 fn visit_journaled_keys_in_bounds<E>(
862 backend: &DataStoreBackend,
863 bounds: (Bound<RawDataStoreKey>, Bound<RawDataStoreKey>),
864 reverse: bool,
865 mut visitor: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
866 ) -> Result<(), E> {
867 let DataStoreBackend::Journaled {
868 canonical,
869 live,
870 tombstones,
871 ..
872 } = backend
873 else {
874 return Ok(());
875 };
876
877 if canonical.is_empty() {
878 if reverse {
879 for (key, _row) in live.range(bounds).rev() {
880 if visitor(key)?.should_stop() {
881 return Ok(());
882 }
883 }
884 } else {
885 for (key, _row) in live.range(bounds) {
886 if visitor(key)?.should_stop() {
887 return Ok(());
888 }
889 }
890 }
891 return Ok(());
892 }
893
894 if live.is_empty() && tombstones.is_empty() {
895 if reverse {
896 for entry in canonical.range(bounds).rev() {
897 if visitor(entry.key())?.should_stop() {
898 return Ok(());
899 }
900 }
901 } else {
902 for entry in canonical.range(bounds) {
903 if visitor(entry.key())?.should_stop() {
904 return Ok(());
905 }
906 }
907 }
908 return Ok(());
909 }
910
911 let direction = if reverse {
912 Direction::Desc
913 } else {
914 Direction::Asc
915 };
916 match direction {
917 Direction::Asc => visit_ordered_overlay(
918 canonical.range((bounds.0.clone(), bounds.1.clone())),
919 live.range((bounds.0, bounds.1)),
920 direction,
921 |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
922 |canonical_entry| !tombstones.contains(canonical_entry.key()),
923 |live_entry| !tombstones.contains(live_entry.0),
924 |entry| {
925 let visit = match entry {
926 OrderedOverlayEntry::Canonical(canonical_entry) => {
927 visitor(canonical_entry.key())?
928 }
929 OrderedOverlayEntry::Live((key, _row)) => visitor(key)?,
930 };
931 Ok(if visit.should_stop() {
932 OrderedOverlayVisit::Stop
933 } else {
934 OrderedOverlayVisit::Continue
935 })
936 },
937 ),
938 Direction::Desc => visit_ordered_overlay(
939 canonical.range((bounds.0.clone(), bounds.1.clone())).rev(),
940 live.range((bounds.0, bounds.1)).rev(),
941 direction,
942 |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
943 |canonical_entry| !tombstones.contains(canonical_entry.key()),
944 |live_entry| !tombstones.contains(live_entry.0),
945 |entry| {
946 let visit = match entry {
947 OrderedOverlayEntry::Canonical(canonical_entry) => {
948 visitor(canonical_entry.key())?
949 }
950 OrderedOverlayEntry::Live((key, _row)) => visitor(key)?,
951 };
952 Ok(if visit.should_stop() {
953 OrderedOverlayVisit::Stop
954 } else {
955 OrderedOverlayVisit::Continue
956 })
957 },
958 ),
959 }
960 }
961
962 fn visit_journaled_entries_in_bounds<E>(
963 backend: &DataStoreBackend,
964 bounds: (Bound<RawDataStoreKey>, Bound<RawDataStoreKey>),
965 mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
966 ) -> Result<(), E> {
967 let DataStoreBackend::Journaled {
968 canonical,
969 live,
970 tombstones,
971 ..
972 } = backend
973 else {
974 return Ok(());
975 };
976
977 if canonical.is_empty() {
978 for (key, row) in live.range(bounds) {
979 if visitor(key, row)?.should_stop() {
980 return Ok(());
981 }
982 }
983 return Ok(());
984 }
985
986 if live.is_empty() && tombstones.is_empty() {
987 for entry in canonical.range(bounds) {
988 if visitor(entry.key(), &entry.value())?.should_stop() {
989 return Ok(());
990 }
991 }
992 return Ok(());
993 }
994
995 visit_ordered_overlay(
996 canonical.range((bounds.0.clone(), bounds.1.clone())),
997 live.range((bounds.0, bounds.1)),
998 Direction::Asc,
999 |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
1000 |canonical_entry| !tombstones.contains(canonical_entry.key()),
1001 |live_entry| !tombstones.contains(live_entry.0),
1002 |entry| {
1003 let visit = match entry {
1004 OrderedOverlayEntry::Canonical(canonical_entry) => {
1005 visitor(canonical_entry.key(), &canonical_entry.value())?
1006 }
1007 OrderedOverlayEntry::Live((key, row)) => visitor(key, row)?,
1008 };
1009 Ok(if visit.should_stop() {
1010 OrderedOverlayVisit::Stop
1011 } else {
1012 OrderedOverlayVisit::Continue
1013 })
1014 },
1015 )
1016 }
1017}
1018
1019#[derive(Clone, Debug)]
1020struct EntityCardinality {
1021 counts: HeapBTreeMap<EntityTag, u64>,
1022 decodable: bool,
1023}
1024
1025impl EntityCardinality {
1026 const fn empty() -> Self {
1027 Self {
1028 counts: HeapBTreeMap::new(),
1029 decodable: true,
1030 }
1031 }
1032
1033 const fn unavailable() -> Self {
1034 Self {
1035 counts: HeapBTreeMap::new(),
1036 decodable: false,
1037 }
1038 }
1039
1040 fn exact_count(&self, entity: EntityTag) -> Option<u64> {
1041 self.decodable
1042 .then(|| self.counts.get(&entity).copied().unwrap_or(0))
1043 }
1044
1045 fn apply_insert(&mut self, key: &RawDataStoreKey, previous: Option<&RawRow>) {
1046 if previous.is_some() {
1047 return;
1048 }
1049 self.apply_present_key(key);
1050 }
1051
1052 fn apply_remove(&mut self, key: &RawDataStoreKey, previous: Option<&RawRow>) {
1053 if previous.is_none() {
1054 return;
1055 }
1056 self.apply_removed_key(key);
1057 }
1058
1059 fn apply_present_key(&mut self, key: &RawDataStoreKey) {
1060 if !self.decodable {
1061 return;
1062 }
1063 let Some(entity) = key.entity_tag_prefix() else {
1064 self.invalidate();
1065 return;
1066 };
1067
1068 let count = self.counts.entry(entity).or_insert(0);
1069 *count = count.saturating_add(1);
1070 }
1071
1072 fn apply_removed_key(&mut self, key: &RawDataStoreKey) {
1073 if !self.decodable {
1074 return;
1075 }
1076 let Some(entity) = key.entity_tag_prefix() else {
1077 self.invalidate();
1078 return;
1079 };
1080
1081 if let Some(count) = self.counts.get_mut(&entity) {
1082 *count = count.saturating_sub(1);
1083 if *count == 0 {
1084 self.counts.remove(&entity);
1085 }
1086 }
1087 }
1088
1089 fn invalidate(&mut self) {
1090 self.counts.clear();
1091 self.decodable = false;
1092 }
1093}
1094
1095#[cfg(test)]
1096mod tests;