use crate::{
db::{
data::{CanonicalRow, RawDataStoreKey, RawRow},
direction::Direction,
ordered_overlay::{OrderedOverlayEntry, OrderedOverlayVisit, visit_ordered_overlay},
positioned_overlay::{
JournalOverlayPosition, PositionedOverlayMetadata, PositionedOverlayRetirement,
},
},
types::EntityTag,
};
use ic_stable_structures::{
BTreeMap as StableBTreeMap, DefaultMemoryImpl, memory_manager::VirtualMemory,
};
#[cfg(all(feature = "sql", feature = "diagnostics"))]
use std::cell::Cell;
use std::collections::{BTreeMap as HeapBTreeMap, BTreeSet};
use std::convert::Infallible;
use std::ops::{Bound, RangeBounds};
#[cfg(all(feature = "sql", feature = "diagnostics"))]
thread_local! {
static DATA_STORE_GET_CALL_COUNT: Cell<u64> = const { Cell::new(0) };
}
#[cfg(all(feature = "sql", feature = "diagnostics"))]
fn record_data_store_get_call() {
DATA_STORE_GET_CALL_COUNT.with(|count| {
count.set(count.get().saturating_add(1));
});
}
pub struct DataStore {
backend: DataStoreBackend,
generation: u64,
entity_cardinality: EntityCardinality,
}
enum DataStoreBackend {
Heap(HeapBTreeMap<RawDataStoreKey, RawRow>),
Journaled {
canonical: StableBTreeMap<RawDataStoreKey, RawRow, VirtualMemory<DefaultMemoryImpl>>,
live: HeapBTreeMap<RawDataStoreKey, RawRow>,
tombstones: BTreeSet<RawDataStoreKey>,
positions: PositionedOverlayMetadata<RawDataStoreKey>,
},
}
#[cfg(any(test, feature = "migration"))]
pub(in crate::db) struct PreparedDataPositionPublication {
keys: Vec<RawDataStoreKey>,
position: JournalOverlayPosition,
}
pub(in crate::db) struct PreparedDataPositionRetirement {
entries: Vec<(RawDataStoreKey, PositionedOverlayRetirement)>,
}
pub(in crate::db) enum StoredRowRead<'a> {
Missing,
Borrowed(&'a RawRow),
Owned(RawRow),
}
impl StoredRowRead<'_> {
#[must_use]
pub(in crate::db) const fn as_row(&self) -> Option<&RawRow> {
match self {
Self::Missing => None,
Self::Borrowed(row) => Some(row),
Self::Owned(row) => Some(row),
}
}
#[must_use]
fn into_owned(self) -> Option<RawRow> {
match self {
Self::Missing => None,
Self::Borrowed(row) => Some(row.clone()),
Self::Owned(row) => Some(row),
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(in crate::db) enum StoreVisit {
Continue,
Stop,
}
impl StoreVisit {
const fn should_stop(self) -> bool {
matches!(self, Self::Stop)
}
}
impl DataStore {
#[must_use]
pub const fn init_heap() -> Self {
Self {
backend: DataStoreBackend::Heap(HeapBTreeMap::new()),
generation: 0,
entity_cardinality: EntityCardinality::empty(),
}
}
#[must_use]
pub fn init_journaled(memory: VirtualMemory<DefaultMemoryImpl>) -> Self {
let canonical = StableBTreeMap::init(memory);
let entity_cardinality = if canonical.is_empty() {
EntityCardinality::empty()
} else {
EntityCardinality::unavailable()
};
Self {
backend: DataStoreBackend::Journaled {
canonical,
live: HeapBTreeMap::new(),
tombstones: BTreeSet::new(),
positions: PositionedOverlayMetadata::new(),
},
generation: 0,
entity_cardinality,
}
}
pub(in crate::db) fn insert(
&mut self,
key: RawDataStoreKey,
row: CanonicalRow,
) -> Option<RawRow> {
let row = row.into_raw_row();
let previous_journaled = if matches!(self.backend, DataStoreBackend::Journaled { .. }) {
self.get(&key)
} else {
None
};
let cardinality_key = key.clone();
let previous = match &mut self.backend {
DataStoreBackend::Heap(map) => map.insert(key, row),
DataStoreBackend::Journaled {
live, tombstones, ..
} => {
tombstones.remove(&key);
live.insert(key, row);
previous_journaled
}
};
self.entity_cardinality
.apply_insert(&cardinality_key, previous.as_ref());
self.bump_generation();
previous
}
#[cfg(test)]
pub(in crate::db) fn insert_raw_for_test(
&mut self,
key: RawDataStoreKey,
row: RawRow,
) -> Option<RawRow> {
let previous_journaled = if matches!(self.backend, DataStoreBackend::Journaled { .. }) {
self.get(&key)
} else {
None
};
let cardinality_key = key.clone();
let previous = match &mut self.backend {
DataStoreBackend::Heap(map) => map.insert(key, row),
DataStoreBackend::Journaled {
live, tombstones, ..
} => {
tombstones.remove(&key);
live.insert(key, row);
previous_journaled
}
};
self.entity_cardinality
.apply_insert(&cardinality_key, previous.as_ref());
self.bump_generation();
previous
}
pub(in crate::db) fn remove(&mut self, key: &RawDataStoreKey) -> Option<RawRow> {
let previous_journaled = if matches!(self.backend, DataStoreBackend::Journaled { .. }) {
self.get(key)
} else {
None
};
let previous = match &mut self.backend {
DataStoreBackend::Heap(map) => map.remove(key),
DataStoreBackend::Journaled {
live, tombstones, ..
} => {
live.remove(key);
tombstones.insert(key.clone());
previous_journaled
}
};
self.entity_cardinality.apply_remove(key, previous.as_ref());
self.bump_generation();
previous
}
pub(in crate::db) fn reset_journaled_live_projection(
&mut self,
) -> Result<(), crate::error::InternalError> {
let DataStoreBackend::Journaled {
canonical,
live,
tombstones,
positions,
} = &mut self.backend
else {
return Err(crate::error::InternalError::store_invariant());
};
live.clear();
tombstones.clear();
positions.clear();
self.entity_cardinality = if canonical.is_empty() {
EntityCardinality::empty()
} else {
EntityCardinality::unavailable()
};
self.bump_generation();
Ok(())
}
pub(in crate::db) fn apply_recovered_journal_put(
&mut self,
key: RawDataStoreKey,
row: RawRow,
) -> Result<Option<RawRow>, crate::error::InternalError> {
let DataStoreBackend::Journaled {
canonical,
live,
tombstones,
..
} = &mut self.backend
else {
return Err(crate::error::InternalError::store_invariant());
};
let previous = if tombstones.contains(&key) {
None
} else {
live.get(&key).cloned().or_else(|| canonical.get(&key))
};
tombstones.remove(&key);
let cardinality_key = key.clone();
live.insert(key, row);
self.entity_cardinality
.apply_insert(&cardinality_key, previous.as_ref());
self.bump_generation();
Ok(previous)
}
pub(in crate::db) fn apply_recovered_journal_delete(
&mut self,
key: &RawDataStoreKey,
) -> Result<Option<RawRow>, crate::error::InternalError> {
let DataStoreBackend::Journaled {
canonical,
live,
tombstones,
..
} = &mut self.backend
else {
return Err(crate::error::InternalError::store_invariant());
};
let previous = if tombstones.contains(key) {
None
} else {
live.get(key).cloned().or_else(|| canonical.get(key))
};
live.remove(key);
tombstones.insert(key.clone());
self.entity_cardinality.apply_remove(key, previous.as_ref());
self.bump_generation();
Ok(previous)
}
pub(in crate::db) fn publish_preflighted_journal_entry(
&mut self,
key: RawDataStoreKey,
row: Option<RawRow>,
position: JournalOverlayPosition,
) -> Result<Option<RawRow>, crate::error::InternalError> {
let DataStoreBackend::Journaled {
canonical,
live,
tombstones,
positions,
} = &mut self.backend
else {
return Err(crate::error::InternalError::store_invariant());
};
let previous = if tombstones.contains(&key) {
None
} else {
live.get(&key).cloned().or_else(|| canonical.get(&key))
};
let cardinality_key = key.clone();
if let Some(row) = row {
tombstones.remove(&key);
live.insert(key.clone(), row);
self.entity_cardinality
.apply_insert(&cardinality_key, previous.as_ref());
} else {
live.remove(&key);
tombstones.insert(key.clone());
self.entity_cardinality
.apply_remove(&cardinality_key, previous.as_ref());
}
positions.publish_preflighted(key, position);
self.bump_generation();
Ok(previous)
}
#[cfg(test)]
pub(in crate::db) fn publish_positioned_journal_entry(
&mut self,
key: RawDataStoreKey,
row: Option<RawRow>,
position: JournalOverlayPosition,
) -> Result<Option<RawRow>, crate::error::InternalError> {
self.preflight_positioned_journal_entry(&key, position)?;
self.publish_preflighted_journal_entry(key, row, position)
}
pub(in crate::db) fn preflight_positioned_journal_entry(
&self,
key: &RawDataStoreKey,
position: JournalOverlayPosition,
) -> Result<(), crate::error::InternalError> {
let DataStoreBackend::Journaled { positions, .. } = &self.backend else {
return Err(crate::error::InternalError::store_invariant());
};
positions.preflight_publish(key, position)
}
#[cfg(any(test, feature = "migration"))]
pub(in crate::db) fn prepare_position_publication(
&self,
keys: impl IntoIterator<Item = RawDataStoreKey>,
position: JournalOverlayPosition,
) -> Result<PreparedDataPositionPublication, crate::error::InternalError> {
let DataStoreBackend::Journaled { positions, .. } = &self.backend else {
return Err(crate::error::InternalError::store_invariant());
};
let keys = keys.into_iter().collect::<BTreeSet<_>>();
for key in &keys {
positions.preflight_publish(key, position)?;
}
Ok(PreparedDataPositionPublication {
keys: keys.into_iter().collect(),
position,
})
}
#[cfg(any(test, feature = "migration"))]
pub(in crate::db) fn publish_prepared_positions(
&mut self,
prepared: PreparedDataPositionPublication,
) {
let DataStoreBackend::Journaled { positions, .. } = &mut self.backend else {
debug_assert!(false, "preflighted row positions require a journaled store");
return;
};
for key in prepared.keys {
positions.publish_preflighted(key, prepared.position);
}
}
pub(in crate::db) fn prepare_position_retirement(
&self,
keys: impl IntoIterator<Item = RawDataStoreKey>,
position: JournalOverlayPosition,
) -> Result<PreparedDataPositionRetirement, crate::error::InternalError> {
let DataStoreBackend::Journaled { positions, .. } = &self.backend else {
return Err(crate::error::InternalError::store_invariant());
};
let entries = keys
.into_iter()
.collect::<BTreeSet<_>>()
.into_iter()
.map(|key| {
positions
.preflight_retirement(&key, position)
.map(|retirement| (key, retirement))
})
.collect::<Result<Vec<_>, _>>()?;
Ok(PreparedDataPositionRetirement { entries })
}
pub(in crate::db) fn apply_prepared_position_retirement(
&mut self,
prepared: PreparedDataPositionRetirement,
) {
let DataStoreBackend::Journaled {
live,
tombstones,
positions,
..
} = &mut self.backend
else {
debug_assert!(
false,
"preflighted row retirement requires a journaled store"
);
return;
};
for (key, retirement) in prepared.entries {
if retirement == PositionedOverlayRetirement::Exact {
live.remove(&key);
tombstones.remove(&key);
positions.retire_preflighted(&key, retirement);
}
}
}
#[cfg(test)]
fn retire_positioned_journal_effect(
&mut self,
key: &RawDataStoreKey,
position: JournalOverlayPosition,
) -> Result<PositionedOverlayRetirement, crate::error::InternalError> {
let DataStoreBackend::Journaled { positions, .. } = &self.backend else {
return Err(crate::error::InternalError::store_invariant());
};
let retirement = positions.preflight_retirement(key, position)?;
let prepared = PreparedDataPositionRetirement {
entries: vec![(key.clone(), retirement)],
};
self.apply_prepared_position_retirement(prepared);
Ok(retirement)
}
pub(in crate::db) fn fold_recovered_journal_put(
&mut self,
key: RawDataStoreKey,
row: RawRow,
) -> Result<Option<RawRow>, crate::error::InternalError> {
let DataStoreBackend::Journaled {
canonical,
live,
tombstones,
..
} = &mut self.backend
else {
return Err(crate::error::InternalError::store_invariant());
};
let visible = !live.contains_key(&key) && !tombstones.contains(&key);
let cardinality_key = key.clone();
let previous = canonical.insert(key, row);
if visible {
self.entity_cardinality
.apply_insert(&cardinality_key, previous.as_ref());
}
self.bump_generation();
Ok(previous)
}
pub(in crate::db) fn fold_recovered_journal_delete(
&mut self,
key: &RawDataStoreKey,
) -> Result<Option<RawRow>, crate::error::InternalError> {
let DataStoreBackend::Journaled {
canonical,
live,
tombstones,
..
} = &mut self.backend
else {
return Err(crate::error::InternalError::store_invariant());
};
let visible = !live.contains_key(key) && !tombstones.contains(key);
let previous = canonical.remove(key);
if visible {
self.entity_cardinality.apply_remove(key, previous.as_ref());
}
self.bump_generation();
Ok(previous)
}
pub(in crate::db) fn preflight_fold_recovered_journal(
&self,
) -> Result<(), crate::error::InternalError> {
match self.backend {
DataStoreBackend::Journaled { .. } => Ok(()),
DataStoreBackend::Heap(_) => Err(crate::error::InternalError::store_invariant()),
}
}
pub(in crate::db) fn get(&self, key: &RawDataStoreKey) -> Option<RawRow> {
self.read(key).into_owned()
}
pub(in crate::db) fn get_canonical(&self, key: &RawDataStoreKey) -> Option<RawRow> {
match &self.backend {
DataStoreBackend::Heap(map) => map.get(key).cloned(),
DataStoreBackend::Journaled { canonical, .. } => canonical.get(key),
}
}
pub(in crate::db) fn read<'a>(&'a self, key: &RawDataStoreKey) -> StoredRowRead<'a> {
#[cfg(all(feature = "sql", feature = "diagnostics"))]
record_data_store_get_call();
match &self.backend {
DataStoreBackend::Heap(map) => map
.get(key)
.map_or(StoredRowRead::Missing, StoredRowRead::Borrowed),
DataStoreBackend::Journaled {
canonical,
live,
tombstones,
..
} => {
if tombstones.contains(key) {
StoredRowRead::Missing
} else if let Some(row) = live.get(key) {
StoredRowRead::Borrowed(row)
} else {
canonical
.get(key)
.map_or(StoredRowRead::Missing, StoredRowRead::Owned)
}
}
}
}
#[must_use]
pub(in crate::db) fn contains(&self, key: &RawDataStoreKey) -> bool {
match &self.backend {
DataStoreBackend::Heap(map) => map.contains_key(key),
DataStoreBackend::Journaled {
canonical,
live,
tombstones,
..
} => {
!tombstones.contains(key)
&& (live.contains_key(key) || canonical.get(key).is_some())
}
}
}
#[must_use]
pub(in crate::db) fn len(&self) -> u64 {
match &self.backend {
DataStoreBackend::Heap(map) => u64::try_from(map.len()).unwrap_or(u64::MAX),
DataStoreBackend::Journaled { .. } => {
let mut count = 0_u64;
let _: Result<(), Infallible> = self.visit_entries(|_key, _row| {
count = count.saturating_add(1);
Ok(StoreVisit::Continue)
});
count
}
}
}
#[must_use]
pub(in crate::db) const fn generation(&self) -> u64 {
self.generation
}
#[must_use]
pub(in crate::db) fn exact_entity_count(&self, entity: EntityTag) -> Option<u64> {
self.entity_cardinality.exact_count(entity)
}
pub(in crate::db) fn visit_entries<E>(
&self,
mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
) -> Result<(), E> {
match &self.backend {
DataStoreBackend::Heap(map) => {
for (key, row) in map {
if visitor(key, row)?.should_stop() {
break;
}
}
}
DataStoreBackend::Journaled { .. } => Self::visit_journaled_entries_in_bounds(
&self.backend,
(Bound::Unbounded, Bound::Unbounded),
visitor,
)?,
}
Ok(())
}
pub(in crate::db) fn visit_range<E>(
&self,
key_range: impl RangeBounds<RawDataStoreKey>,
mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
) -> Result<(), E> {
let bounds = Self::owned_range_bounds(&key_range);
match &self.backend {
DataStoreBackend::Heap(map) => {
for (key, row) in map.range((bounds.0.clone(), bounds.1)) {
if visitor(key, row)?.should_stop() {
break;
}
}
}
DataStoreBackend::Journaled { .. } => {
Self::visit_journaled_entries_in_bounds(&self.backend, bounds, visitor)?;
}
}
Ok(())
}
pub(in crate::db) fn try_visit_range_with_row_preflight<E>(
&self,
key_range: impl RangeBounds<RawDataStoreKey>,
mut preflight: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
) -> Result<Option<bool>, E> {
let bounds = Self::owned_range_bounds(&key_range);
let mut stopped = false;
match &self.backend {
DataStoreBackend::Heap(map) => {
for (key, row) in map.range((bounds.0.clone(), bounds.1)) {
if preflight(key)?.should_stop() {
stopped = true;
break;
}
if visitor(key, row)?.should_stop() {
stopped = true;
break;
}
}
}
DataStoreBackend::Journaled {
canonical,
live,
tombstones,
..
} if canonical.is_empty() => {
for (key, row) in live.range((bounds.0.clone(), bounds.1)) {
if tombstones.contains(key) {
continue;
}
if preflight(key)?.should_stop() {
stopped = true;
break;
}
if visitor(key, row)?.should_stop() {
stopped = true;
break;
}
}
}
DataStoreBackend::Journaled {
canonical,
live,
tombstones,
..
} if live.is_empty() && tombstones.is_empty() => {
for entry in canonical.range((bounds.0.clone(), bounds.1)) {
if preflight(entry.key())?.should_stop() {
stopped = true;
break;
}
if visitor(entry.key(), &entry.value())?.should_stop() {
stopped = true;
break;
}
}
}
DataStoreBackend::Journaled { .. } => return Ok(None),
}
Ok(Some(!stopped))
}
pub(in crate::db) fn visit_key_range<E>(
&self,
key_range: impl RangeBounds<RawDataStoreKey>,
visitor: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
) -> Result<(), E> {
self.visit_keys_in_bounds(Self::owned_range_bounds(&key_range), false, visitor)
}
pub(in crate::db) fn visit_key_range_rev<E>(
&self,
key_range: impl RangeBounds<RawDataStoreKey>,
visitor: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
) -> Result<(), E> {
self.visit_keys_in_bounds(Self::owned_range_bounds(&key_range), true, visitor)
}
pub(in crate::db) fn memory_bytes(&self) -> u64 {
let mut bytes = 0u64;
let _: Result<(), Infallible> = self.visit_entries(|key, row| {
bytes = bytes.saturating_add(key.as_bytes().len() as u64 + row.len() as u64);
Ok(StoreVisit::Continue)
});
bytes
}
const fn bump_generation(&mut self) {
self.generation = self.generation.saturating_add(1);
}
#[cfg(test)]
fn rebuild_entity_cardinality_from_entries(&mut self) {
let mut cardinality = EntityCardinality::empty();
let _: Result<(), Infallible> = self.visit_entries(|key, _row| {
cardinality.apply_present_key(key);
Ok(StoreVisit::Continue)
});
self.entity_cardinality = cardinality;
}
#[cfg(all(feature = "sql", feature = "diagnostics"))]
pub(in crate::db) fn current_get_call_count() -> u64 {
DATA_STORE_GET_CALL_COUNT.with(Cell::get)
}
fn owned_range_bounds(
key_range: &impl RangeBounds<RawDataStoreKey>,
) -> (Bound<RawDataStoreKey>, Bound<RawDataStoreKey>) {
let lower = match key_range.start_bound() {
Bound::Included(key) => Bound::Included(key.clone()),
Bound::Excluded(key) => Bound::Excluded(key.clone()),
Bound::Unbounded => Bound::Unbounded,
};
let upper = match key_range.end_bound() {
Bound::Included(key) => Bound::Included(key.clone()),
Bound::Excluded(key) => Bound::Excluded(key.clone()),
Bound::Unbounded => Bound::Unbounded,
};
(lower, upper)
}
fn visit_keys_in_bounds<E>(
&self,
bounds: (Bound<RawDataStoreKey>, Bound<RawDataStoreKey>),
reverse: bool,
mut visitor: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
) -> Result<(), E> {
match &self.backend {
DataStoreBackend::Heap(map) => {
if reverse {
for (key, _row) in map.range(bounds).rev() {
if visitor(key)?.should_stop() {
break;
}
}
} else {
for (key, _row) in map.range(bounds) {
if visitor(key)?.should_stop() {
break;
}
}
}
}
DataStoreBackend::Journaled { .. } => {
Self::visit_journaled_keys_in_bounds(&self.backend, bounds, reverse, visitor)?;
}
}
Ok(())
}
fn visit_journaled_keys_in_bounds<E>(
backend: &DataStoreBackend,
bounds: (Bound<RawDataStoreKey>, Bound<RawDataStoreKey>),
reverse: bool,
mut visitor: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
) -> Result<(), E> {
let DataStoreBackend::Journaled {
canonical,
live,
tombstones,
..
} = backend
else {
return Ok(());
};
if canonical.is_empty() {
if reverse {
for (key, _row) in live.range(bounds).rev() {
if visitor(key)?.should_stop() {
return Ok(());
}
}
} else {
for (key, _row) in live.range(bounds) {
if visitor(key)?.should_stop() {
return Ok(());
}
}
}
return Ok(());
}
if live.is_empty() && tombstones.is_empty() {
if reverse {
for entry in canonical.range(bounds).rev() {
if visitor(entry.key())?.should_stop() {
return Ok(());
}
}
} else {
for entry in canonical.range(bounds) {
if visitor(entry.key())?.should_stop() {
return Ok(());
}
}
}
return Ok(());
}
let direction = if reverse {
Direction::Desc
} else {
Direction::Asc
};
match direction {
Direction::Asc => visit_ordered_overlay(
canonical.range((bounds.0.clone(), bounds.1.clone())),
live.range((bounds.0, bounds.1)),
direction,
|canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
|canonical_entry| !tombstones.contains(canonical_entry.key()),
|live_entry| !tombstones.contains(live_entry.0),
|entry| {
let visit = match entry {
OrderedOverlayEntry::Canonical(canonical_entry) => {
visitor(canonical_entry.key())?
}
OrderedOverlayEntry::Live((key, _row)) => visitor(key)?,
};
Ok(if visit.should_stop() {
OrderedOverlayVisit::Stop
} else {
OrderedOverlayVisit::Continue
})
},
),
Direction::Desc => visit_ordered_overlay(
canonical.range((bounds.0.clone(), bounds.1.clone())).rev(),
live.range((bounds.0, bounds.1)).rev(),
direction,
|canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
|canonical_entry| !tombstones.contains(canonical_entry.key()),
|live_entry| !tombstones.contains(live_entry.0),
|entry| {
let visit = match entry {
OrderedOverlayEntry::Canonical(canonical_entry) => {
visitor(canonical_entry.key())?
}
OrderedOverlayEntry::Live((key, _row)) => visitor(key)?,
};
Ok(if visit.should_stop() {
OrderedOverlayVisit::Stop
} else {
OrderedOverlayVisit::Continue
})
},
),
}
}
fn visit_journaled_entries_in_bounds<E>(
backend: &DataStoreBackend,
bounds: (Bound<RawDataStoreKey>, Bound<RawDataStoreKey>),
mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
) -> Result<(), E> {
let DataStoreBackend::Journaled {
canonical,
live,
tombstones,
..
} = backend
else {
return Ok(());
};
if canonical.is_empty() {
for (key, row) in live.range(bounds) {
if visitor(key, row)?.should_stop() {
return Ok(());
}
}
return Ok(());
}
if live.is_empty() && tombstones.is_empty() {
for entry in canonical.range(bounds) {
if visitor(entry.key(), &entry.value())?.should_stop() {
return Ok(());
}
}
return Ok(());
}
visit_ordered_overlay(
canonical.range((bounds.0.clone(), bounds.1.clone())),
live.range((bounds.0, bounds.1)),
Direction::Asc,
|canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
|canonical_entry| !tombstones.contains(canonical_entry.key()),
|live_entry| !tombstones.contains(live_entry.0),
|entry| {
let visit = match entry {
OrderedOverlayEntry::Canonical(canonical_entry) => {
visitor(canonical_entry.key(), &canonical_entry.value())?
}
OrderedOverlayEntry::Live((key, row)) => visitor(key, row)?,
};
Ok(if visit.should_stop() {
OrderedOverlayVisit::Stop
} else {
OrderedOverlayVisit::Continue
})
},
)
}
}
#[derive(Clone, Debug)]
struct EntityCardinality {
counts: HeapBTreeMap<EntityTag, u64>,
decodable: bool,
}
impl EntityCardinality {
const fn empty() -> Self {
Self {
counts: HeapBTreeMap::new(),
decodable: true,
}
}
const fn unavailable() -> Self {
Self {
counts: HeapBTreeMap::new(),
decodable: false,
}
}
fn exact_count(&self, entity: EntityTag) -> Option<u64> {
self.decodable
.then(|| self.counts.get(&entity).copied().unwrap_or(0))
}
fn apply_insert(&mut self, key: &RawDataStoreKey, previous: Option<&RawRow>) {
if previous.is_some() {
return;
}
self.apply_present_key(key);
}
fn apply_remove(&mut self, key: &RawDataStoreKey, previous: Option<&RawRow>) {
if previous.is_none() {
return;
}
self.apply_removed_key(key);
}
fn apply_present_key(&mut self, key: &RawDataStoreKey) {
if !self.decodable {
return;
}
let Some(entity) = key.entity_tag_prefix() else {
self.invalidate();
return;
};
let count = self.counts.entry(entity).or_insert(0);
*count = count.saturating_add(1);
}
fn apply_removed_key(&mut self, key: &RawDataStoreKey) {
if !self.decodable {
return;
}
let Some(entity) = key.entity_tag_prefix() else {
self.invalidate();
return;
};
if let Some(count) = self.counts.get_mut(&entity) {
*count = count.saturating_sub(1);
if *count == 0 {
self.counts.remove(&entity);
}
}
}
fn invalidate(&mut self) {
self.counts.clear();
self.decodable = false;
}
}
#[cfg(test)]
mod tests;