Skip to main content

icydb_core/db/data/
store.rs

1//! Module: data::store
2//! Responsibility: journaled-or-heap row storage behind the data-store boundary.
3//! Does not own: key/row validation policy beyond type boundaries.
4//! Boundary: commit/executor call into this layer after prevalidation.
5
6use crate::{
7    db::{
8        data::{CanonicalRow, RawDataStoreKey, RawRow},
9        direction::Direction,
10        ordered_overlay::{OrderedOverlayEntry, OrderedOverlayVisit, visit_ordered_overlay},
11    },
12    types::EntityTag,
13};
14use ic_stable_structures::{
15    BTreeMap as StableBTreeMap, DefaultMemoryImpl, memory_manager::VirtualMemory,
16};
17#[cfg(all(feature = "sql", feature = "diagnostics"))]
18use std::cell::Cell;
19use std::collections::{BTreeMap as HeapBTreeMap, BTreeSet};
20use std::convert::Infallible;
21use std::ops::{Bound, RangeBounds};
22
23#[cfg(all(feature = "sql", feature = "diagnostics"))]
24thread_local! {
25    static DATA_STORE_GET_CALL_COUNT: Cell<u64> = const { Cell::new(0) };
26}
27
28#[cfg(all(feature = "sql", feature = "diagnostics"))]
29fn record_data_store_get_call() {
30    DATA_STORE_GET_CALL_COUNT.with(|count| {
31        count.set(count.get().saturating_add(1));
32    });
33}
34
35///
36/// DataStore
37///
38/// Thin persistence wrapper over one journaled or heap BTreeMap.
39///
40/// Invariant: callers provide already-validated `RawDataStoreKey` and canonical row bytes.
41/// This type intentionally does not enforce commit-phase ordering.
42///
43
44pub struct DataStore {
45    backend: DataStoreBackend,
46    generation: u64,
47    entity_cardinality: EntityCardinality,
48}
49
50enum DataStoreBackend {
51    Heap(HeapBTreeMap<RawDataStoreKey, RawRow>),
52    Journaled {
53        canonical: StableBTreeMap<RawDataStoreKey, RawRow, VirtualMemory<DefaultMemoryImpl>>,
54        live: HeapBTreeMap<RawDataStoreKey, RawRow>,
55        tombstones: BTreeSet<RawDataStoreKey>,
56    },
57}
58
59/// One visible row read that borrows heap/live state and owns stable state.
60///
61/// Callers that only need selected fields can evaluate a borrowed row while
62/// the store handle is active. Stable-structure reads remain owned because
63/// that backend cannot expose a value reference beyond its storage call.
64pub(in crate::db) enum StoredRowRead<'a> {
65    Missing,
66    Borrowed(&'a RawRow),
67    Owned(RawRow),
68}
69
70impl StoredRowRead<'_> {
71    /// Borrow the visible row regardless of its physical backing.
72    #[must_use]
73    pub(in crate::db) const fn as_row(&self) -> Option<&RawRow> {
74        match self {
75            Self::Missing => None,
76            Self::Borrowed(row) => Some(row),
77            Self::Owned(row) => Some(row),
78        }
79    }
80
81    /// Convert the visible row into the existing owned get contract.
82    #[must_use]
83    fn into_owned(self) -> Option<RawRow> {
84        match self {
85            Self::Missing => None,
86            Self::Borrowed(row) => Some(row.clone()),
87            Self::Owned(row) => Some(row),
88        }
89    }
90}
91
92/// Control-flow result for store traversal visitors.
93#[derive(Clone, Copy, Debug, Eq, PartialEq)]
94pub(in crate::db) enum StoreVisit {
95    Continue,
96    Stop,
97}
98
99impl StoreVisit {
100    const fn should_stop(self) -> bool {
101        matches!(self, Self::Stop)
102    }
103}
104
105impl DataStore {
106    /// Initialize a volatile heap-backed data store.
107    #[must_use]
108    pub const fn init_heap() -> Self {
109        Self {
110            backend: DataStoreBackend::Heap(HeapBTreeMap::new()),
111            generation: 0,
112            entity_cardinality: EntityCardinality::empty(),
113        }
114    }
115
116    /// Initialize a journaled cached-stable data store.
117    ///
118    /// Normal writes update only the live projection. The canonical stable map
119    /// is the future fold target and is not mutated by this wrapper's write
120    /// methods.
121    #[must_use]
122    pub fn init_journaled(memory: VirtualMemory<DefaultMemoryImpl>) -> Self {
123        let canonical = StableBTreeMap::init(memory);
124        let entity_cardinality = if canonical.is_empty() {
125            EntityCardinality::empty()
126        } else {
127            EntityCardinality::unavailable()
128        };
129        Self {
130            backend: DataStoreBackend::Journaled {
131                canonical,
132                live: HeapBTreeMap::new(),
133                tombstones: BTreeSet::new(),
134            },
135            generation: 0,
136            // Stable rows remain authoritative after reinitialization. Exact
137            // zero cardinality is still known for an empty canonical map;
138            // populated maps remain unavailable without a startup scan.
139            entity_cardinality,
140        }
141    }
142
143    /// Insert or replace one row by raw key.
144    pub(in crate::db) fn insert(
145        &mut self,
146        key: RawDataStoreKey,
147        row: CanonicalRow,
148    ) -> Option<RawRow> {
149        let row = row.into_raw_row();
150        let previous_journaled = if matches!(self.backend, DataStoreBackend::Journaled { .. }) {
151            self.get(&key)
152        } else {
153            None
154        };
155        let cardinality_key = key.clone();
156        let previous = match &mut self.backend {
157            DataStoreBackend::Heap(map) => map.insert(key, row),
158            DataStoreBackend::Journaled {
159                live, tombstones, ..
160            } => {
161                tombstones.remove(&key);
162                live.insert(key, row);
163                previous_journaled
164            }
165        };
166        self.entity_cardinality
167            .apply_insert(&cardinality_key, previous.as_ref());
168        self.bump_generation();
169        previous
170    }
171
172    /// Insert one raw row directly for corruption-focused test setup only.
173    #[cfg(test)]
174    pub(in crate::db) fn insert_raw_for_test(
175        &mut self,
176        key: RawDataStoreKey,
177        row: RawRow,
178    ) -> Option<RawRow> {
179        let previous_journaled = if matches!(self.backend, DataStoreBackend::Journaled { .. }) {
180            self.get(&key)
181        } else {
182            None
183        };
184        let cardinality_key = key.clone();
185        let previous = match &mut self.backend {
186            DataStoreBackend::Heap(map) => map.insert(key, row),
187            DataStoreBackend::Journaled {
188                live, tombstones, ..
189            } => {
190                tombstones.remove(&key);
191                live.insert(key, row);
192                previous_journaled
193            }
194        };
195        self.entity_cardinality
196            .apply_insert(&cardinality_key, previous.as_ref());
197        self.bump_generation();
198        previous
199    }
200
201    /// Remove one row by raw key.
202    pub(in crate::db) fn remove(&mut self, key: &RawDataStoreKey) -> Option<RawRow> {
203        let previous_journaled = if matches!(self.backend, DataStoreBackend::Journaled { .. }) {
204            self.get(key)
205        } else {
206            None
207        };
208        let previous = match &mut self.backend {
209            DataStoreBackend::Heap(map) => map.remove(key),
210            DataStoreBackend::Journaled {
211                live, tombstones, ..
212            } => {
213                live.remove(key);
214                tombstones.insert(key.clone());
215                previous_journaled
216            }
217        };
218        self.entity_cardinality.apply_remove(key, previous.as_ref());
219        self.bump_generation();
220        previous
221    }
222
223    /// Reset the volatile projection for journaled recovery without mutating
224    /// the canonical stable base.
225    pub(in crate::db) fn reset_journaled_live_projection(
226        &mut self,
227    ) -> Result<(), crate::error::InternalError> {
228        let DataStoreBackend::Journaled {
229            canonical,
230            live,
231            tombstones,
232        } = &mut self.backend
233        else {
234            return Err(crate::error::InternalError::store_invariant());
235        };
236
237        live.clear();
238        tombstones.clear();
239        self.entity_cardinality = if canonical.is_empty() {
240            EntityCardinality::empty()
241        } else {
242            EntityCardinality::unavailable()
243        };
244        self.bump_generation();
245
246        Ok(())
247    }
248
249    /// Apply one recovered journal row put into the volatile projection.
250    pub(in crate::db) fn apply_recovered_journal_put(
251        &mut self,
252        key: RawDataStoreKey,
253        row: RawRow,
254    ) -> Result<Option<RawRow>, crate::error::InternalError> {
255        let DataStoreBackend::Journaled {
256            canonical,
257            live,
258            tombstones,
259        } = &mut self.backend
260        else {
261            return Err(crate::error::InternalError::store_invariant());
262        };
263
264        let previous = if tombstones.contains(&key) {
265            None
266        } else {
267            live.get(&key).cloned().or_else(|| canonical.get(&key))
268        };
269        tombstones.remove(&key);
270        let cardinality_key = key.clone();
271        live.insert(key, row);
272        self.entity_cardinality
273            .apply_insert(&cardinality_key, previous.as_ref());
274        self.bump_generation();
275
276        Ok(previous)
277    }
278
279    /// Apply one recovered journal row delete into the volatile projection.
280    pub(in crate::db) fn apply_recovered_journal_delete(
281        &mut self,
282        key: &RawDataStoreKey,
283    ) -> Result<Option<RawRow>, crate::error::InternalError> {
284        let DataStoreBackend::Journaled {
285            canonical,
286            live,
287            tombstones,
288        } = &mut self.backend
289        else {
290            return Err(crate::error::InternalError::store_invariant());
291        };
292
293        let previous = if tombstones.contains(key) {
294            None
295        } else {
296            live.get(key).cloned().or_else(|| canonical.get(key))
297        };
298        live.remove(key);
299        tombstones.insert(key.clone());
300        self.entity_cardinality.apply_remove(key, previous.as_ref());
301        self.bump_generation();
302
303        Ok(previous)
304    }
305
306    /// Apply one folded journal row put into the canonical stable base.
307    pub(in crate::db) fn fold_recovered_journal_put(
308        &mut self,
309        key: RawDataStoreKey,
310        row: RawRow,
311    ) -> Result<Option<RawRow>, crate::error::InternalError> {
312        let DataStoreBackend::Journaled {
313            canonical,
314            live,
315            tombstones,
316        } = &mut self.backend
317        else {
318            return Err(crate::error::InternalError::store_invariant());
319        };
320
321        let visible = !live.contains_key(&key) && !tombstones.contains(&key);
322        let cardinality_key = key.clone();
323        let previous = canonical.insert(key, row);
324        if visible {
325            self.entity_cardinality
326                .apply_insert(&cardinality_key, previous.as_ref());
327        }
328        self.bump_generation();
329
330        Ok(previous)
331    }
332
333    /// Apply one folded journal row delete into the canonical stable base.
334    pub(in crate::db) fn fold_recovered_journal_delete(
335        &mut self,
336        key: &RawDataStoreKey,
337    ) -> Result<Option<RawRow>, crate::error::InternalError> {
338        let DataStoreBackend::Journaled {
339            canonical,
340            live,
341            tombstones,
342        } = &mut self.backend
343        else {
344            return Err(crate::error::InternalError::store_invariant());
345        };
346
347        let visible = !live.contains_key(key) && !tombstones.contains(key);
348        let previous = canonical.remove(key);
349        if visible {
350            self.entity_cardinality.apply_remove(key, previous.as_ref());
351        }
352        self.bump_generation();
353
354        Ok(previous)
355    }
356
357    /// Prove that recovered journal rows can be folded into canonical storage.
358    pub(in crate::db) fn preflight_fold_recovered_journal(
359        &self,
360    ) -> Result<(), crate::error::InternalError> {
361        match self.backend {
362            DataStoreBackend::Journaled { .. } => Ok(()),
363            DataStoreBackend::Heap(_) => Err(crate::error::InternalError::store_invariant()),
364        }
365    }
366
367    /// Load one row by raw key.
368    pub(in crate::db) fn get(&self, key: &RawDataStoreKey) -> Option<RawRow> {
369        self.read(key).into_owned()
370    }
371
372    /// Read one visible row without cloning heap/live payloads.
373    pub(in crate::db) fn read<'a>(&'a self, key: &RawDataStoreKey) -> StoredRowRead<'a> {
374        #[cfg(all(feature = "sql", feature = "diagnostics"))]
375        record_data_store_get_call();
376
377        match &self.backend {
378            DataStoreBackend::Heap(map) => map
379                .get(key)
380                .map_or(StoredRowRead::Missing, StoredRowRead::Borrowed),
381            DataStoreBackend::Journaled {
382                canonical,
383                live,
384                tombstones,
385            } => {
386                if tombstones.contains(key) {
387                    StoredRowRead::Missing
388                } else if let Some(row) = live.get(key) {
389                    StoredRowRead::Borrowed(row)
390                } else {
391                    canonical
392                        .get(key)
393                        .map_or(StoredRowRead::Missing, StoredRowRead::Owned)
394                }
395            }
396        }
397    }
398
399    /// Return whether one raw key exists without cloning the row payload.
400    #[must_use]
401    pub(in crate::db) fn contains(&self, key: &RawDataStoreKey) -> bool {
402        match &self.backend {
403            DataStoreBackend::Heap(map) => map.contains_key(key),
404            DataStoreBackend::Journaled {
405                canonical,
406                live,
407                tombstones,
408            } => {
409                !tombstones.contains(key)
410                    && (live.contains_key(key) || canonical.get(key).is_some())
411            }
412        }
413    }
414
415    /// Return the number of stored rows without exposing the backing map.
416    #[must_use]
417    pub(in crate::db) fn len(&self) -> u64 {
418        match &self.backend {
419            DataStoreBackend::Heap(map) => u64::try_from(map.len()).unwrap_or(u64::MAX),
420            DataStoreBackend::Journaled { .. } => {
421                let mut count = 0_u64;
422                let _: Result<(), Infallible> = self.visit_entries(|_key, _row| {
423                    count = count.saturating_add(1);
424                    Ok(StoreVisit::Continue)
425                });
426                count
427            }
428        }
429    }
430
431    /// Return the row-store generation used to prove index metadata freshness.
432    #[must_use]
433    pub(in crate::db) const fn generation(&self) -> u64 {
434        self.generation
435    }
436
437    /// Return an exact current row count for one entity when store metadata is valid.
438    #[must_use]
439    pub(in crate::db) fn exact_entity_count(&self, entity: EntityTag) -> Option<u64> {
440        self.entity_cardinality.exact_count(entity)
441    }
442
443    /// Visit raw row entries in canonical storage order.
444    pub(in crate::db) fn visit_entries<E>(
445        &self,
446        mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
447    ) -> Result<(), E> {
448        match &self.backend {
449            DataStoreBackend::Heap(map) => {
450                for (key, row) in map {
451                    if visitor(key, row)?.should_stop() {
452                        break;
453                    }
454                }
455            }
456            DataStoreBackend::Journaled {
457                canonical: _,
458                live: _,
459                tombstones: _,
460            } => Self::visit_journaled_entries_in_bounds(
461                &self.backend,
462                (Bound::Unbounded, Bound::Unbounded),
463                visitor,
464            )?,
465        }
466
467        Ok(())
468    }
469
470    /// Visit raw row entries whose keys belong to the provided storage range.
471    pub(in crate::db) fn visit_range<E>(
472        &self,
473        key_range: impl RangeBounds<RawDataStoreKey>,
474        mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
475    ) -> Result<(), E> {
476        let bounds = Self::owned_range_bounds(&key_range);
477        match &self.backend {
478            DataStoreBackend::Heap(map) => {
479                for (key, row) in map.range((bounds.0.clone(), bounds.1)) {
480                    if visitor(key, row)?.should_stop() {
481                        break;
482                    }
483                }
484            }
485            DataStoreBackend::Journaled {
486                canonical: _,
487                live: _,
488                tombstones: _,
489            } => Self::visit_journaled_entries_in_bounds(&self.backend, bounds, visitor)?,
490        }
491
492        Ok(())
493    }
494
495    /// Visit one ascending row range while allowing the caller to stop after
496    /// seeing the key but before a stable row payload is materialized.
497    ///
498    /// Mixed journal overlays retain the ordinary key-then-point-read path;
499    /// one physical backing can keep the range iterator open for the complete
500    /// scan and avoid a second tree lookup per row.
501    pub(in crate::db) fn try_visit_range_with_row_preflight<E>(
502        &self,
503        key_range: impl RangeBounds<RawDataStoreKey>,
504        mut preflight: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
505        mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
506    ) -> Result<Option<bool>, E> {
507        let bounds = Self::owned_range_bounds(&key_range);
508        let mut stopped = false;
509        match &self.backend {
510            DataStoreBackend::Heap(map) => {
511                for (key, row) in map.range((bounds.0.clone(), bounds.1)) {
512                    if preflight(key)?.should_stop() {
513                        stopped = true;
514                        break;
515                    }
516                    if visitor(key, row)?.should_stop() {
517                        stopped = true;
518                        break;
519                    }
520                }
521            }
522            DataStoreBackend::Journaled {
523                canonical,
524                live,
525                tombstones,
526            } if canonical.is_empty() => {
527                for (key, row) in live.range((bounds.0.clone(), bounds.1)) {
528                    if tombstones.contains(key) {
529                        continue;
530                    }
531                    if preflight(key)?.should_stop() {
532                        stopped = true;
533                        break;
534                    }
535                    if visitor(key, row)?.should_stop() {
536                        stopped = true;
537                        break;
538                    }
539                }
540            }
541            DataStoreBackend::Journaled {
542                canonical,
543                live,
544                tombstones,
545            } if live.is_empty() && tombstones.is_empty() => {
546                for entry in canonical.range((bounds.0.clone(), bounds.1)) {
547                    if preflight(entry.key())?.should_stop() {
548                        stopped = true;
549                        break;
550                    }
551                    if visitor(entry.key(), &entry.value())?.should_stop() {
552                        stopped = true;
553                        break;
554                    }
555                }
556            }
557            DataStoreBackend::Journaled { .. } => return Ok(None),
558        }
559
560        Ok(Some(!stopped))
561    }
562
563    /// Visit only raw keys in storage order without fetching row payloads.
564    ///
565    /// Primary-key access streams use this boundary to discover candidate
566    /// identities before the terminal row runtime decides whether the payload
567    /// is needed. Journaled traversal merges canonical and live keys while
568    /// preserving live overrides and tombstones without reading stable values.
569    pub(in crate::db) fn visit_key_range<E>(
570        &self,
571        key_range: impl RangeBounds<RawDataStoreKey>,
572        visitor: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
573    ) -> Result<(), E> {
574        self.visit_keys_in_bounds(Self::owned_range_bounds(&key_range), false, visitor)
575    }
576
577    /// Visit only raw keys in reverse storage order without fetching row payloads.
578    pub(in crate::db) fn visit_key_range_rev<E>(
579        &self,
580        key_range: impl RangeBounds<RawDataStoreKey>,
581        visitor: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
582    ) -> Result<(), E> {
583        self.visit_keys_in_bounds(Self::owned_range_bounds(&key_range), true, visitor)
584    }
585
586    /// Sum of bytes used by all stored rows.
587    pub(in crate::db) fn memory_bytes(&self) -> u64 {
588        // Report map footprint as key bytes + row bytes per entry.
589        let mut bytes = 0u64;
590        let _: Result<(), Infallible> = self.visit_entries(|key, row| {
591            bytes = bytes.saturating_add(key.as_bytes().len() as u64 + row.len() as u64);
592            Ok(StoreVisit::Continue)
593        });
594        bytes
595    }
596
597    const fn bump_generation(&mut self) {
598        self.generation = self.generation.saturating_add(1);
599    }
600
601    #[cfg(test)]
602    fn rebuild_entity_cardinality_from_entries(&mut self) {
603        let mut cardinality = EntityCardinality::empty();
604        let _: Result<(), Infallible> = self.visit_entries(|key, _row| {
605            cardinality.apply_present_key(key);
606            Ok(StoreVisit::Continue)
607        });
608        self.entity_cardinality = cardinality;
609    }
610
611    /// Return the monotonic perf-only count of stable row fetches seen by this process.
612    #[cfg(all(feature = "sql", feature = "diagnostics"))]
613    pub(in crate::db) fn current_get_call_count() -> u64 {
614        DATA_STORE_GET_CALL_COUNT.with(Cell::get)
615    }
616
617    fn owned_range_bounds(
618        key_range: &impl RangeBounds<RawDataStoreKey>,
619    ) -> (Bound<RawDataStoreKey>, Bound<RawDataStoreKey>) {
620        let lower = match key_range.start_bound() {
621            Bound::Included(key) => Bound::Included(key.clone()),
622            Bound::Excluded(key) => Bound::Excluded(key.clone()),
623            Bound::Unbounded => Bound::Unbounded,
624        };
625        let upper = match key_range.end_bound() {
626            Bound::Included(key) => Bound::Included(key.clone()),
627            Bound::Excluded(key) => Bound::Excluded(key.clone()),
628            Bound::Unbounded => Bound::Unbounded,
629        };
630
631        (lower, upper)
632    }
633
634    fn visit_keys_in_bounds<E>(
635        &self,
636        bounds: (Bound<RawDataStoreKey>, Bound<RawDataStoreKey>),
637        reverse: bool,
638        mut visitor: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
639    ) -> Result<(), E> {
640        match &self.backend {
641            DataStoreBackend::Heap(map) => {
642                if reverse {
643                    for (key, _row) in map.range(bounds).rev() {
644                        if visitor(key)?.should_stop() {
645                            break;
646                        }
647                    }
648                } else {
649                    for (key, _row) in map.range(bounds) {
650                        if visitor(key)?.should_stop() {
651                            break;
652                        }
653                    }
654                }
655            }
656            DataStoreBackend::Journaled { .. } => {
657                Self::visit_journaled_keys_in_bounds(&self.backend, bounds, reverse, visitor)?;
658            }
659        }
660
661        Ok(())
662    }
663
664    fn visit_journaled_keys_in_bounds<E>(
665        backend: &DataStoreBackend,
666        bounds: (Bound<RawDataStoreKey>, Bound<RawDataStoreKey>),
667        reverse: bool,
668        mut visitor: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
669    ) -> Result<(), E> {
670        let DataStoreBackend::Journaled {
671            canonical,
672            live,
673            tombstones,
674        } = backend
675        else {
676            return Ok(());
677        };
678
679        if canonical.is_empty() {
680            if reverse {
681                for (key, _row) in live.range(bounds).rev() {
682                    if visitor(key)?.should_stop() {
683                        return Ok(());
684                    }
685                }
686            } else {
687                for (key, _row) in live.range(bounds) {
688                    if visitor(key)?.should_stop() {
689                        return Ok(());
690                    }
691                }
692            }
693            return Ok(());
694        }
695
696        if live.is_empty() && tombstones.is_empty() {
697            if reverse {
698                for entry in canonical.range(bounds).rev() {
699                    if visitor(entry.key())?.should_stop() {
700                        return Ok(());
701                    }
702                }
703            } else {
704                for entry in canonical.range(bounds) {
705                    if visitor(entry.key())?.should_stop() {
706                        return Ok(());
707                    }
708                }
709            }
710            return Ok(());
711        }
712
713        let direction = if reverse {
714            Direction::Desc
715        } else {
716            Direction::Asc
717        };
718        match direction {
719            Direction::Asc => visit_ordered_overlay(
720                canonical.range((bounds.0.clone(), bounds.1.clone())),
721                live.range((bounds.0, bounds.1)),
722                direction,
723                |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
724                |canonical_entry| !tombstones.contains(canonical_entry.key()),
725                |live_entry| !tombstones.contains(live_entry.0),
726                |entry| {
727                    let visit = match entry {
728                        OrderedOverlayEntry::Canonical(canonical_entry) => {
729                            visitor(canonical_entry.key())?
730                        }
731                        OrderedOverlayEntry::Live((key, _row)) => visitor(key)?,
732                    };
733                    Ok(if visit.should_stop() {
734                        OrderedOverlayVisit::Stop
735                    } else {
736                        OrderedOverlayVisit::Continue
737                    })
738                },
739            ),
740            Direction::Desc => visit_ordered_overlay(
741                canonical.range((bounds.0.clone(), bounds.1.clone())).rev(),
742                live.range((bounds.0, bounds.1)).rev(),
743                direction,
744                |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
745                |canonical_entry| !tombstones.contains(canonical_entry.key()),
746                |live_entry| !tombstones.contains(live_entry.0),
747                |entry| {
748                    let visit = match entry {
749                        OrderedOverlayEntry::Canonical(canonical_entry) => {
750                            visitor(canonical_entry.key())?
751                        }
752                        OrderedOverlayEntry::Live((key, _row)) => visitor(key)?,
753                    };
754                    Ok(if visit.should_stop() {
755                        OrderedOverlayVisit::Stop
756                    } else {
757                        OrderedOverlayVisit::Continue
758                    })
759                },
760            ),
761        }
762    }
763
764    fn visit_journaled_entries_in_bounds<E>(
765        backend: &DataStoreBackend,
766        bounds: (Bound<RawDataStoreKey>, Bound<RawDataStoreKey>),
767        mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
768    ) -> Result<(), E> {
769        let DataStoreBackend::Journaled {
770            canonical,
771            live,
772            tombstones,
773        } = backend
774        else {
775            return Ok(());
776        };
777
778        if canonical.is_empty() {
779            for (key, row) in live.range(bounds) {
780                if visitor(key, row)?.should_stop() {
781                    return Ok(());
782                }
783            }
784            return Ok(());
785        }
786
787        if live.is_empty() && tombstones.is_empty() {
788            for entry in canonical.range(bounds) {
789                if visitor(entry.key(), &entry.value())?.should_stop() {
790                    return Ok(());
791                }
792            }
793            return Ok(());
794        }
795
796        visit_ordered_overlay(
797            canonical.range((bounds.0.clone(), bounds.1.clone())),
798            live.range((bounds.0, bounds.1)),
799            Direction::Asc,
800            |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
801            |canonical_entry| !tombstones.contains(canonical_entry.key()),
802            |live_entry| !tombstones.contains(live_entry.0),
803            |entry| {
804                let visit = match entry {
805                    OrderedOverlayEntry::Canonical(canonical_entry) => {
806                        visitor(canonical_entry.key(), &canonical_entry.value())?
807                    }
808                    OrderedOverlayEntry::Live((key, row)) => visitor(key, row)?,
809                };
810                Ok(if visit.should_stop() {
811                    OrderedOverlayVisit::Stop
812                } else {
813                    OrderedOverlayVisit::Continue
814                })
815            },
816        )
817    }
818}
819
820#[derive(Clone, Debug)]
821struct EntityCardinality {
822    counts: HeapBTreeMap<EntityTag, u64>,
823    decodable: bool,
824}
825
826impl EntityCardinality {
827    const fn empty() -> Self {
828        Self {
829            counts: HeapBTreeMap::new(),
830            decodable: true,
831        }
832    }
833
834    const fn unavailable() -> Self {
835        Self {
836            counts: HeapBTreeMap::new(),
837            decodable: false,
838        }
839    }
840
841    fn exact_count(&self, entity: EntityTag) -> Option<u64> {
842        self.decodable
843            .then(|| self.counts.get(&entity).copied().unwrap_or(0))
844    }
845
846    fn apply_insert(&mut self, key: &RawDataStoreKey, previous: Option<&RawRow>) {
847        if previous.is_some() {
848            return;
849        }
850        self.apply_present_key(key);
851    }
852
853    fn apply_remove(&mut self, key: &RawDataStoreKey, previous: Option<&RawRow>) {
854        if previous.is_none() {
855            return;
856        }
857        self.apply_removed_key(key);
858    }
859
860    fn apply_present_key(&mut self, key: &RawDataStoreKey) {
861        if !self.decodable {
862            return;
863        }
864        let Some(entity) = key.entity_tag_prefix() else {
865            self.invalidate();
866            return;
867        };
868
869        let count = self.counts.entry(entity).or_insert(0);
870        *count = count.saturating_add(1);
871    }
872
873    fn apply_removed_key(&mut self, key: &RawDataStoreKey) {
874        if !self.decodable {
875            return;
876        }
877        let Some(entity) = key.entity_tag_prefix() else {
878            self.invalidate();
879            return;
880        };
881
882        if let Some(count) = self.counts.get_mut(&entity) {
883            *count = count.saturating_sub(1);
884            if *count == 0 {
885                self.counts.remove(&entity);
886            }
887        }
888    }
889
890    fn invalidate(&mut self) {
891        self.counts.clear();
892        self.decodable = false;
893    }
894}
895
896#[cfg(test)]
897mod tests;