use std::{
collections::{BTreeMap, HashMap, HashSet},
ops::{Bound, RangeBounds},
};
use reifydb_core::{
actors::drop::{DropMessage, DropRequest},
common::CommitVersion,
delta::Delta,
encoded::{
key::{EncodedKey, EncodedKeyRange},
row::EncodedRow,
},
event::metric::{MultiCommittedEvent, MultiDelete, MultiWrite},
interface::store::{
EntryKind, MultiVersionBatch, MultiVersionCommit, MultiVersionContains, MultiVersionGet,
MultiVersionGetPrevious, MultiVersionRow, MultiVersionStore, classify_key, classify_range,
is_single_version_semantics_key,
},
};
use reifydb_store::row::page::PageId;
use reifydb_value::{
reifydb_assertions,
util::{cowvec::CowVec, hex},
};
use tracing::{instrument, warn};
use super::StandardMultiStore;
use crate::{
MultiVersionScope, Result,
tier::{
RangeBatch, RangeCursor, TierBatch, TierStorage, VersionedGetResult,
commit::buffer::MultiCommitBufferTier,
persistent::MultiPersistentTier,
read::{MultiReadBufferTier, ServedChunk},
},
};
const TIER_SCAN_CHUNK_SIZE: usize = 32;
pub(crate) const WARM_THRESHOLD: u64 = 4 * TIER_SCAN_CHUNK_SIZE as u64;
impl MultiVersionGet for StandardMultiStore {
fn get(&self, key: &EncodedKey, version: CommitVersion) -> Result<Option<MultiVersionRow>> {
match classify_key(key) {
EntryKind::Operator(_) => self.get_operator(key, version),
EntryKind::Source(_) => self.get_source(key, version),
_ => self.get_multi(key, version),
}
}
}
impl StandardMultiStore {
#[instrument(name = "store::multi::get::operator", level = "trace", skip(self, key), fields(version = version.0))]
fn get_operator(&self, key: &EncodedKey, version: CommitVersion) -> Result<Option<MultiVersionRow>> {
self.get_impl(key, version)
}
#[instrument(name = "store::multi::get::source", level = "trace", skip(self, key), fields(version = version.0))]
fn get_source(&self, key: &EncodedKey, version: CommitVersion) -> Result<Option<MultiVersionRow>> {
self.get_impl(key, version)
}
#[instrument(name = "store::multi::get::multi", level = "trace", skip(self, key), fields(version = version.0))]
fn get_multi(&self, key: &EncodedKey, version: CommitVersion) -> Result<Option<MultiVersionRow>> {
self.get_impl(key, version)
}
#[inline]
fn get_impl(&self, key: &EncodedKey, version: CommitVersion) -> Result<Option<MultiVersionRow>> {
let table = classify_key(key);
if let Some(found) = self.get_probe_commit(table, key, version)? {
return Ok(found);
}
if let Some(found) = self.get_probe_read(key, version) {
return Ok(found);
}
if let Some(found) = self.get_probe_persistent(table, key, version)? {
return Ok(found);
}
Ok(None)
}
}
impl StandardMultiStore {
#[inline]
fn get_probe_commit(
&self,
table: EntryKind,
key: &EncodedKey,
version: CommitVersion,
) -> Result<Option<Option<MultiVersionRow>>> {
let Some(commit) = &self.commit else {
return Ok(None);
};
Ok(match commit.get(table, key.as_ref(), version)? {
VersionedGetResult::Value {
value,
version: v,
} => Some(Some(MultiVersionRow {
key: key.clone(),
row: EncodedRow(value),
version: v,
})),
VersionedGetResult::Tombstone => Some(None),
VersionedGetResult::NotFound => None,
})
}
#[inline]
fn get_probe_read(&self, key: &EncodedKey, version: CommitVersion) -> Option<Option<MultiVersionRow>> {
let read = self.read.as_ref()?;
match read.get(key, version) {
VersionedGetResult::Value {
value,
version: v,
} => Some(Some(MultiVersionRow {
key: key.clone(),
row: EncodedRow(value),
version: v,
})),
VersionedGetResult::Tombstone => Some(None),
VersionedGetResult::NotFound => None,
}
}
#[inline]
fn get_probe_persistent(
&self,
table: EntryKind,
key: &EncodedKey,
version: CommitVersion,
) -> Result<Option<Option<MultiVersionRow>>> {
let Some(persistent) = &self.persistent else {
return Ok(None);
};
Ok(match persistent.get(table, key.as_ref(), version)? {
VersionedGetResult::Value {
value,
version: v,
} => {
if let Some(read) = &self.read {
read.insert(key.clone(), v, Some(value.clone()));
}
Some(Some(MultiVersionRow {
key: key.clone(),
row: EncodedRow(value),
version: v,
}))
}
VersionedGetResult::Tombstone => Some(None),
VersionedGetResult::NotFound => None,
})
}
}
impl MultiVersionContains for StandardMultiStore {
#[instrument(name = "store::multi::contains", level = "trace", skip(self), fields(key_hex = %hex::display(key.as_ref()), version = version.0), ret)]
fn contains(&self, key: &EncodedKey, version: CommitVersion) -> Result<bool> {
Ok(MultiVersionGet::get(self, key, version)?.is_some())
}
}
impl MultiVersionCommit for StandardMultiStore {
#[instrument(name = "store::multi::commit", level = "debug", skip(self, deltas), fields(delta_count = deltas.len(), version = version.0))]
fn commit(&self, deltas: CowVec<Delta>, version: CommitVersion) -> Result<()> {
let classified = classify_deltas(&deltas);
let (operator_drops, source_drops) = partition_drops(classified.explicit_drops);
self.dispatch_drops(build_drop_batch(source_drops, &classified.pending_set_keys, version));
self.update_read_cache_on_commit(version, &classified.batches);
if !self.write_batches(version, classified.batches)? {
return Ok(());
}
self.evict_operator_state(&operator_drops)?;
self.emit_commit_metrics(classified.writes, classified.deletes, version);
Ok(())
}
}
type DropPartition = (Vec<(EntryKind, EncodedKey)>, Vec<(EntryKind, EncodedKey)>);
#[inline]
fn partition_drops(explicit_drops: Vec<(EntryKind, EncodedKey)>) -> DropPartition {
explicit_drops.into_iter().partition(|(table, _)| matches!(table, EntryKind::Operator(_)))
}
struct ClassifiedDeltas {
pending_set_keys: HashSet<EncodedKey>,
writes: Vec<MultiWrite>,
deletes: Vec<MultiDelete>,
batches: TierBatch,
explicit_drops: Vec<(EntryKind, EncodedKey)>,
}
#[inline]
fn classify_deltas(deltas: &CowVec<Delta>) -> ClassifiedDeltas {
let mut pending_set_keys: HashSet<EncodedKey> = HashSet::new();
let mut writes: Vec<MultiWrite> = Vec::new();
let mut deletes: Vec<MultiDelete> = Vec::new();
let mut batches: TierBatch = HashMap::new();
let mut explicit_drops: Vec<(EntryKind, EncodedKey)> = Vec::new();
for delta in deltas.iter() {
let key = delta.key();
let table = classify_key(key);
let is_single_version = is_single_version_semantics_key(key);
match delta {
Delta::Set {
key,
row,
} => {
if is_single_version {
pending_set_keys.insert(key.clone());
}
writes.push(MultiWrite {
key: key.clone(),
value_bytes: row.len() as u64,
});
batches.entry(table).or_default().push((key.clone(), Some(row.0.clone())));
}
Delta::Unset {
key,
row,
} => {
deletes.push(MultiDelete {
key: key.clone(),
value_bytes: row.len() as u64,
});
batches.entry(table).or_default().push((key.clone(), None));
}
Delta::Remove {
key,
} => {
deletes.push(MultiDelete {
key: key.clone(),
value_bytes: 0,
});
batches.entry(table).or_default().push((key.clone(), None));
}
Delta::Drop {
key,
} => {
explicit_drops.push((table, key.clone()));
}
}
}
ClassifiedDeltas {
pending_set_keys,
writes,
deletes,
batches,
explicit_drops,
}
}
#[inline]
fn build_drop_batch(
explicit_drops: Vec<(EntryKind, EncodedKey)>,
pending_set_keys: &HashSet<EncodedKey>,
version: CommitVersion,
) -> Vec<DropRequest> {
let mut drop_batch = Vec::with_capacity(explicit_drops.len() + pending_set_keys.len());
for (table, key) in explicit_drops {
let pending_version = if pending_set_keys.contains(key.as_ref()) {
Some(version)
} else {
None
};
drop_batch.push(DropRequest {
table,
key,
commit_version: version,
pending_version,
});
}
for key in pending_set_keys.iter() {
let encoded = EncodedKey::new(key.to_vec());
let table = classify_key(&encoded);
drop_batch.push(DropRequest {
table,
key: encoded,
commit_version: version,
pending_version: Some(version),
});
}
drop_batch
}
impl StandardMultiStore {
pub fn get_many(
&self,
keys: &[EncodedKey],
version: CommitVersion,
) -> Result<HashMap<EncodedKey, MultiVersionRow>> {
let mut by_table: HashMap<EntryKind, Vec<&EncodedKey>> = HashMap::new();
for key in keys {
by_table.entry(classify_key(key)).or_default().push(key);
}
let mut out: HashMap<EncodedKey, MultiVersionRow> = HashMap::new();
for (table, table_keys) in by_table {
self.get_many_for_table(table, &table_keys, version, &mut out)?;
}
Ok(out)
}
#[inline]
fn get_many_for_table(
&self,
table: EntryKind,
table_keys: &[&EncodedKey],
version: CommitVersion,
out: &mut HashMap<EncodedKey, MultiVersionRow>,
) -> Result<()> {
let key_slices: Vec<&[u8]> = table_keys.iter().map(|k| k.as_ref()).collect();
let commit_results = self.probe_commit_batch(table, &key_slices, version)?;
let (read_aligned, persistent_aligned) = self.resolve_misses_through_read_and_persistent(
table,
table_keys,
&key_slices,
&commit_results,
version,
)?;
reifydb_assertions! {
let n = key_slices.len();
assert!(
commit_results.len() == n && read_aligned.len() == n && persistent_aligned.len() == n,
"per-tier result vectors must stay index-aligned with the table's keys, otherwise collect_resolved_rows \
reads a tier result for the wrong key and returns mismatched rows (keys={n}, commit={}, read={}, persistent={})",
commit_results.len(),
read_aligned.len(),
persistent_aligned.len()
);
}
self.collect_resolved_rows(table_keys, &commit_results, &read_aligned, &persistent_aligned, out);
Ok(())
}
#[inline]
fn probe_commit_batch(
&self,
table: EntryKind,
key_slices: &[&[u8]],
version: CommitVersion,
) -> Result<Vec<VersionedGetResult>> {
match &self.commit {
Some(commit) => commit.get_many(table, key_slices, version),
None => Ok(vec![VersionedGetResult::NotFound; key_slices.len()]),
}
}
#[inline]
fn resolve_misses_through_read_and_persistent(
&self,
table: EntryKind,
table_keys: &[&EncodedKey],
key_slices: &[&[u8]],
commit_results: &[VersionedGetResult],
version: CommitVersion,
) -> Result<(Vec<VersionedGetResult>, Vec<VersionedGetResult>)> {
let mut read_aligned = vec![VersionedGetResult::NotFound; key_slices.len()];
let mut persistent_idx: Vec<usize> = Vec::new();
let mut persistent_slices: Vec<&[u8]> = Vec::new();
for (i, result) in commit_results.iter().enumerate() {
if !matches!(result, VersionedGetResult::NotFound) {
continue;
}
let read_hit = self
.read
.as_ref()
.map(|c| c.get(table_keys[i], version))
.unwrap_or(VersionedGetResult::NotFound);
match read_hit {
VersionedGetResult::Value {
value,
version: v,
} => {
read_aligned[i] = VersionedGetResult::Value {
value,
version: v,
};
}
VersionedGetResult::Tombstone => {
read_aligned[i] = VersionedGetResult::Tombstone;
}
VersionedGetResult::NotFound => {
persistent_idx.push(i);
persistent_slices.push(key_slices[i]);
}
}
}
let mut persistent_aligned = vec![VersionedGetResult::NotFound; key_slices.len()];
if !persistent_slices.is_empty()
&& let Some(persistent) = &self.persistent
{
let persistent_results = persistent.get_many(table, &persistent_slices, version)?;
for (slot, result) in persistent_idx.into_iter().zip(persistent_results) {
if let (
Some(read),
VersionedGetResult::Value {
value,
version: v,
},
) = (&self.read, &result)
{
read.insert(table_keys[slot].clone(), *v, Some(value.clone()));
}
persistent_aligned[slot] = result;
}
}
Ok((read_aligned, persistent_aligned))
}
#[inline]
fn collect_resolved_rows(
&self,
table_keys: &[&EncodedKey],
commit_results: &[VersionedGetResult],
read_aligned: &[VersionedGetResult],
persistent_aligned: &[VersionedGetResult],
out: &mut HashMap<EncodedKey, MultiVersionRow>,
) {
for (i, key) in table_keys.iter().enumerate() {
let resolved = match &commit_results[i] {
VersionedGetResult::Value {
value,
version: v,
} => Some((value.clone(), *v)),
VersionedGetResult::Tombstone => None,
VersionedGetResult::NotFound => match &read_aligned[i] {
VersionedGetResult::Value {
value,
version: v,
} => Some((value.clone(), *v)),
VersionedGetResult::Tombstone => None,
VersionedGetResult::NotFound => match &persistent_aligned[i] {
VersionedGetResult::Value {
value,
version: v,
} => Some((value.clone(), *v)),
_ => None,
},
},
};
if let Some((value, v)) = resolved {
out.insert(
(*key).clone(),
MultiVersionRow {
key: (*key).clone(),
row: EncodedRow(value),
version: v,
},
);
}
}
}
#[inline]
fn dispatch_drops(&self, drop_batch: Vec<DropRequest>) {
if drop_batch.is_empty() {
return;
}
if let Some(actor) = &self.drop_actor
&& actor.send_blocking(DropMessage::Batch(drop_batch)).is_err()
{
warn!("Failed to send drop batch");
}
}
#[inline]
fn update_read_cache_on_commit(&self, version: CommitVersion, batches: &TierBatch) {
let Some(read) = &self.read else {
return;
};
for (table, entries) in batches {
match table {
EntryKind::Operator(_) => {
for (key, value) in entries {
match value {
Some(value) => {
read.insert(key.clone(), version, Some(value.clone()))
}
None => read.invalidate(key),
}
}
}
_ => {
for (key, _) in entries {
read.invalidate(key);
}
}
}
}
}
#[inline]
fn write_batches(&self, version: CommitVersion, batches: TierBatch) -> Result<bool> {
if let Some(commit) = &self.commit {
commit.set(version, batches)?;
} else if let Some(persistent) = &self.persistent {
persistent.set(version, batches)?;
} else {
return Ok(false);
}
Ok(true)
}
fn evict_operator_state(&self, drops: &[(EntryKind, EncodedKey)]) -> Result<()> {
if drops.is_empty() {
return Ok(());
}
self.evict_drops_from_commit(drops)?;
self.delete_drops_from_persistent(drops)?;
self.invalidate_drops_in_read(drops);
Ok(())
}
#[inline]
fn evict_drops_from_commit(&self, drops: &[(EntryKind, EncodedKey)]) -> Result<()> {
let Some(commit) = &self.commit else {
return Ok(());
};
let mut batches: HashMap<EntryKind, Vec<(EncodedKey, CommitVersion)>> = HashMap::new();
for (table, key) in drops {
for (entry_version, _) in commit.get_all_versions(*table, key.as_ref())? {
batches.entry(*table).or_default().push((key.clone(), entry_version));
}
}
if !batches.is_empty() {
commit.drop(batches)?;
}
Ok(())
}
#[inline]
fn delete_drops_from_persistent(&self, drops: &[(EntryKind, EncodedKey)]) -> Result<()> {
let Some(persistent) = &self.persistent else {
return Ok(());
};
let mut by_table: HashMap<EntryKind, Vec<EncodedKey>> = HashMap::new();
for (table, key) in drops {
by_table.entry(*table).or_default().push(key.clone());
}
for (table, keys) in by_table {
persistent.delete_keys(table, &keys)?;
}
Ok(())
}
#[inline]
fn invalidate_drops_in_read(&self, drops: &[(EntryKind, EncodedKey)]) {
let Some(read) = &self.read else {
return;
};
for (_, key) in drops {
read.invalidate(key);
}
}
#[inline]
fn emit_commit_metrics(&self, writes: Vec<MultiWrite>, deletes: Vec<MultiDelete>, version: CommitVersion) {
if writes.is_empty() && deletes.is_empty() {
return;
}
self.event_bus.emit(MultiCommittedEvent::new(writes, deletes, vec![], version));
}
}
#[derive(Debug, Clone, Default)]
pub struct MultiVersionRangeCursor {
pub commit: RangeCursor,
pub persistent: RangeCursor,
pub exhausted: bool,
warm_bucket: Option<PageId>,
warm_consumed: u64,
}
impl MultiVersionRangeCursor {
pub fn new() -> Self {
Self::default()
}
pub fn is_exhausted(&self) -> bool {
self.exhausted
}
}
pub struct TierScanQuery<'a> {
pub table: EntryKind,
pub start: &'a [u8],
pub end: &'a [u8],
pub scope: MultiVersionScope,
pub range: &'a EncodedKeyRange,
}
pub fn scan_tier_chunk<S: TierStorage>(
storage: &S,
cursor: &mut RangeCursor,
scan: &TierScanQuery,
collected: &mut BTreeMap<Vec<u8>, (CommitVersion, Option<CowVec<u8>>)>,
) -> Result<bool> {
let batch = storage.range_next(
scan.table,
cursor,
Bound::Included(scan.start),
Bound::Included(scan.end),
scan.scope,
TIER_SCAN_CHUNK_SIZE,
)?;
merge_tier_batch(batch, scan.range, collected)
}
pub fn scan_tier_chunk_rev<S: TierStorage>(
storage: &S,
cursor: &mut RangeCursor,
scan: &TierScanQuery,
collected: &mut BTreeMap<Vec<u8>, (CommitVersion, Option<CowVec<u8>>)>,
) -> Result<bool> {
let batch = storage.range_rev_next(
scan.table,
cursor,
Bound::Included(scan.start),
Bound::Included(scan.end),
scan.scope,
TIER_SCAN_CHUNK_SIZE,
)?;
merge_tier_batch(batch, scan.range, collected)
}
#[inline]
fn merge_tier_batch(
batch: RangeBatch,
range: &EncodedKeyRange,
collected: &mut BTreeMap<Vec<u8>, (CommitVersion, Option<CowVec<u8>>)>,
) -> Result<bool> {
if batch.entries.is_empty() {
return Ok(false);
}
for entry in batch.entries {
let original_key = entry.key.as_slice().to_vec();
let entry_version = entry.version;
let original_key_encoded = EncodedKey::new(original_key.clone());
if !range.contains(&original_key_encoded) {
continue;
}
let should_update = match collected.get(&original_key) {
None => true,
Some((existing_version, _)) => entry_version > *existing_version,
};
if should_update {
collected.insert(original_key, (entry_version, entry.value));
}
}
Ok(true)
}
#[inline]
pub fn collected_to_batch(
collected: BTreeMap<Vec<u8>, (CommitVersion, Option<CowVec<u8>>)>,
has_more: bool,
) -> MultiVersionBatch {
let items: Vec<MultiVersionRow> = collected
.into_iter()
.filter_map(|(key_bytes, (v, value))| {
value.map(|val| MultiVersionRow {
key: EncodedKey::new(key_bytes),
row: EncodedRow(val),
version: v,
})
})
.collect();
MultiVersionBatch {
items,
has_more,
}
}
#[inline]
fn step_all_tiers(
buffer: Option<&MultiCommitBufferTier>,
buffer_cursor: &mut RangeCursor,
persistent: Option<&MultiPersistentTier>,
persistent_cursor: &mut RangeCursor,
scan: &TierScanQuery,
collected: &mut BTreeMap<Vec<u8>, (CommitVersion, Option<CowVec<u8>>)>,
) -> Result<bool> {
let mut any_progress = false;
if let Some(s) = buffer
&& !buffer_cursor.exhausted
{
any_progress |= scan_tier_chunk(s, buffer_cursor, scan, collected)?;
}
if let Some(s) = persistent
&& !persistent_cursor.exhausted
{
any_progress |= scan_tier_chunk(s, persistent_cursor, scan, collected)?;
}
Ok(any_progress)
}
pub fn scan_tiers_latest(
buffer: Option<&MultiCommitBufferTier>,
persistent: Option<&MultiPersistentTier>,
range: EncodedKeyRange,
scope: MultiVersionScope,
max_keys: usize,
) -> Result<MultiVersionBatch> {
let table = classify_key_range(&range);
let (start, end) = make_range_bounds(&range);
let scan = TierScanQuery {
table,
start: &start,
end: &end,
scope,
range: &range,
};
let mut collected: BTreeMap<Vec<u8>, (CommitVersion, Option<CowVec<u8>>)> = BTreeMap::new();
let mut buffer_cursor = RangeCursor::default();
let mut persistent_cursor = RangeCursor::default();
let mut exhausted = false;
while collected.len() < max_keys {
let progress = step_all_tiers(
buffer,
&mut buffer_cursor,
persistent,
&mut persistent_cursor,
&scan,
&mut collected,
)?;
if !progress {
exhausted = true;
break;
}
}
Ok(collected_to_batch(collected, !exhausted))
}
impl StandardMultiStore {
pub fn range_next(
&self,
cursor: &mut MultiVersionRangeCursor,
range: EncodedKeyRange,
scope: MultiVersionScope,
batch_size: u64,
) -> Result<MultiVersionBatch> {
if cursor.exhausted {
return Ok(MultiVersionBatch {
items: Vec::new(),
has_more: false,
});
}
mark_unconfigured_exhausted(self, cursor);
let table = classify_key_range(&range);
let (start, end) = make_range_bounds(&range);
let batch_size = batch_size as usize;
let scan = TierScanQuery {
table,
start: &start,
end: &end,
scope,
range: &range,
};
let mut collected: BTreeMap<Vec<u8>, (CommitVersion, Option<CowVec<u8>>)> = BTreeMap::new();
while collected.len() < batch_size {
let mut any_progress = false;
if let Some(commit) = &self.commit
&& !cursor.commit.exhausted
{
any_progress |= scan_tier_chunk(commit, &mut cursor.commit, &scan, &mut collected)?;
}
if self.persistent.is_some() && !cursor.persistent.exhausted {
any_progress |= self.step_persistent_cached(&scan, cursor, &mut collected, false)?;
}
if !any_progress {
cursor.exhausted = true;
break;
}
}
apply_forward_horizon(cursor, &mut collected);
let items: Vec<MultiVersionRow> = collected
.into_iter()
.filter_map(|(key_bytes, (v, value))| {
value.map(|val| MultiVersionRow {
key: EncodedKey::new(key_bytes),
row: EncodedRow(val),
version: v,
})
})
.collect();
let has_more = !cursor.exhausted;
Ok(MultiVersionBatch {
items,
has_more,
})
}
pub fn range(
&self,
range: EncodedKeyRange,
scope: MultiVersionScope,
batch_size: usize,
) -> MultiVersionRangeIter {
MultiVersionRangeIter {
store: self.clone(),
cursor: MultiVersionRangeCursor::new(),
range,
scope,
batch_size,
current_batch: Vec::new(),
current_index: 0,
}
}
pub fn range_rev(
&self,
range: EncodedKeyRange,
scope: MultiVersionScope,
batch_size: usize,
) -> MultiVersionRangeRevIter {
MultiVersionRangeRevIter {
store: self.clone(),
cursor: MultiVersionRangeCursor::new(),
range,
scope,
batch_size,
current_batch: Vec::new(),
current_index: 0,
}
}
fn range_rev_next(
&self,
cursor: &mut MultiVersionRangeCursor,
range: EncodedKeyRange,
scope: MultiVersionScope,
batch_size: u64,
) -> Result<MultiVersionBatch> {
if cursor.exhausted {
return Ok(MultiVersionBatch {
items: Vec::new(),
has_more: false,
});
}
mark_unconfigured_exhausted(self, cursor);
let table = classify_key_range(&range);
let (start, end) = make_range_bounds(&range);
let batch_size = batch_size as usize;
let scan = TierScanQuery {
table,
start: &start,
end: &end,
scope,
range: &range,
};
let mut collected: BTreeMap<Vec<u8>, (CommitVersion, Option<CowVec<u8>>)> = BTreeMap::new();
while collected.len() < batch_size {
let mut any_progress = false;
if let Some(commit) = &self.commit
&& !cursor.commit.exhausted
{
any_progress |= scan_tier_chunk_rev(commit, &mut cursor.commit, &scan, &mut collected)?;
}
if self.persistent.is_some() && !cursor.persistent.exhausted {
any_progress |= self.step_persistent_cached(&scan, cursor, &mut collected, true)?;
}
if !any_progress {
cursor.exhausted = true;
break;
}
}
apply_reverse_horizon(cursor, &mut collected);
let items: Vec<MultiVersionRow> = collected
.into_iter()
.rev()
.filter_map(|(key_bytes, (v, value))| {
value.map(|val| MultiVersionRow {
key: EncodedKey::new(key_bytes),
row: EncodedRow(val),
version: v,
})
})
.collect();
let has_more = !cursor.exhausted;
Ok(MultiVersionBatch {
items,
has_more,
})
}
fn step_persistent_cached(
&self,
scan: &TierScanQuery,
cursor: &mut MultiVersionRangeCursor,
collected: &mut BTreeMap<Vec<u8>, (CommitVersion, Option<CowVec<u8>>)>,
descending: bool,
) -> Result<bool> {
let Some(persistent) = &self.persistent else {
return Ok(false);
};
if let Some(served) = self.serve_from_read_cache(scan, cursor, collected, descending) {
return served;
}
let (consumed, progressed) =
self.scan_persistent_chunk(persistent, scan, cursor, collected, descending)?;
self.warm_read_bucket_after_scan(persistent, scan, cursor, consumed)?;
Ok(progressed)
}
#[inline]
fn serve_from_read_cache(
&self,
scan: &TierScanQuery,
cursor: &mut MultiVersionRangeCursor,
collected: &mut BTreeMap<Vec<u8>, (CommitVersion, Option<CowVec<u8>>)>,
descending: bool,
) -> Option<Result<bool>> {
let (Some(read), EntryKind::Source(_)) = (&self.read, scan.table) else {
return None;
};
match read.serve_persistent_chunk(
scan.table,
&mut cursor.persistent,
scan.start,
scan.end,
scan.scope,
TIER_SCAN_CHUNK_SIZE,
descending,
) {
ServedChunk::Served(batch) => Some(merge_tier_batch(batch, scan.range, collected)),
ServedChunk::Gap => None,
}
}
#[inline]
fn scan_persistent_chunk(
&self,
persistent: &MultiPersistentTier,
scan: &TierScanQuery,
cursor: &mut MultiVersionRangeCursor,
collected: &mut BTreeMap<Vec<u8>, (CommitVersion, Option<CowVec<u8>>)>,
descending: bool,
) -> Result<(usize, bool)> {
let batch = if descending {
persistent.range_rev_next(
scan.table,
&mut cursor.persistent,
Bound::Included(scan.start),
Bound::Included(scan.end),
scan.scope,
TIER_SCAN_CHUNK_SIZE,
)?
} else {
persistent.range_next(
scan.table,
&mut cursor.persistent,
Bound::Included(scan.start),
Bound::Included(scan.end),
scan.scope,
TIER_SCAN_CHUNK_SIZE,
)?
};
let consumed = batch.entries.len();
let progressed = merge_tier_batch(batch, scan.range, collected)?;
Ok((consumed, progressed))
}
#[inline]
fn warm_read_bucket_after_scan(
&self,
persistent: &MultiPersistentTier,
scan: &TierScanQuery,
cursor: &mut MultiVersionRangeCursor,
consumed: usize,
) -> Result<()> {
if let (Some(read), EntryKind::Source(_)) = (&self.read, scan.table) {
maybe_warm_bucket(read, persistent, cursor, scan.table, consumed)?;
}
Ok(())
}
}
fn maybe_warm_bucket(
read: &MultiReadBufferTier,
persistent: &MultiPersistentTier,
cursor: &mut MultiVersionRangeCursor,
table: EntryKind,
consumed: usize,
) -> Result<()> {
let page = {
let Some(last) = cursor.persistent.last_key.as_ref() else {
return Ok(());
};
read.page_of_key(last)
};
if !matches!(page.kind, EntryKind::Source(_)) {
return Ok(());
}
if cursor.warm_bucket == Some(page) {
cursor.warm_consumed = cursor.warm_consumed.saturating_add(consumed as u64);
} else {
cursor.warm_bucket = Some(page);
cursor.warm_consumed = consumed as u64;
}
if cursor.warm_consumed <= WARM_THRESHOLD {
return Ok(());
}
let Some(range) = read.page_key_range(page) else {
return Ok(());
};
let (Bound::Included(lo), Bound::Included(hi)) = (range.start, range.end) else {
return Ok(());
};
let entries = persistent.load_range_consistent(
table,
Bound::Included(lo.as_slice()),
Bound::Included(hi.as_slice()),
CommitVersion(u64::MAX),
)?;
read.populate_page(page, entries, true);
cursor.warm_bucket = None;
cursor.warm_consumed = 0;
Ok(())
}
fn mark_unconfigured_exhausted(store: &StandardMultiStore, cursor: &mut MultiVersionRangeCursor) {
if store.commit.is_none() {
cursor.commit.exhausted = true;
}
if store.persistent.is_none() {
cursor.persistent.exhausted = true;
}
}
fn apply_forward_horizon(
cursor: &mut MultiVersionRangeCursor,
collected: &mut BTreeMap<Vec<u8>, (CommitVersion, Option<CowVec<u8>>)>,
) {
let horizon = forward_horizon(cursor);
if let Some(h) = horizon {
collected.retain(|k, _| k.as_slice() <= h.as_slice());
rewind_over_advanced_forward(cursor, &h);
}
}
fn apply_reverse_horizon(
cursor: &mut MultiVersionRangeCursor,
collected: &mut BTreeMap<Vec<u8>, (CommitVersion, Option<CowVec<u8>>)>,
) {
let horizon = reverse_horizon(cursor);
if let Some(h) = horizon {
collected.retain(|k, _| k.as_slice() >= h.as_slice());
rewind_over_advanced_reverse(cursor, &h);
}
}
fn forward_horizon(cursor: &MultiVersionRangeCursor) -> Option<EncodedKey> {
let mut horizon: Option<EncodedKey> = None;
for tier in [&cursor.commit, &cursor.persistent] {
if tier.exhausted {
continue;
}
let last = match &tier.last_key {
Some(k) => k.clone(),
None => return None,
};
horizon = Some(match horizon {
None => last,
Some(prev) => {
if last.as_slice() < prev.as_slice() {
last
} else {
prev
}
}
});
}
horizon
}
fn reverse_horizon(cursor: &MultiVersionRangeCursor) -> Option<EncodedKey> {
let mut horizon: Option<EncodedKey> = None;
for tier in [&cursor.commit, &cursor.persistent] {
if tier.exhausted {
continue;
}
let last = match &tier.last_key {
Some(k) => k.clone(),
None => return None,
};
horizon = Some(match horizon {
None => last,
Some(prev) => {
if last.as_slice() > prev.as_slice() {
last
} else {
prev
}
}
});
}
horizon
}
fn rewind_over_advanced_forward(cursor: &mut MultiVersionRangeCursor, horizon: &EncodedKey) {
for tier in [&mut cursor.commit, &mut cursor.persistent] {
if let Some(last) = &tier.last_key
&& last.as_slice() > horizon.as_slice()
{
tier.last_key = Some(horizon.clone());
tier.exhausted = false;
}
}
}
fn rewind_over_advanced_reverse(cursor: &mut MultiVersionRangeCursor, horizon: &EncodedKey) {
for tier in [&mut cursor.commit, &mut cursor.persistent] {
if let Some(last) = &tier.last_key
&& last.as_slice() < horizon.as_slice()
{
tier.last_key = Some(horizon.clone());
tier.exhausted = false;
}
}
}
impl MultiVersionGetPrevious for StandardMultiStore {
fn get_previous_version(
&self,
key: &EncodedKey,
before_version: CommitVersion,
) -> Result<Option<MultiVersionRow>> {
if before_version.0 == 0 {
return Ok(None);
}
let table = classify_key(key);
reifydb_assertions! {
assert!(
before_version.0 >= 1,
"the before_version==0 guard must precede this subtraction, otherwise before_version.0 - 1 \
wraps to u64::MAX and the probe reads the latest version instead of the previous one \
(before_version={})",
before_version.0
);
}
let prev_version = CommitVersion(before_version.0 - 1);
if let Some(found) = self.previous_probe_commit(table, key, prev_version)? {
return Ok(found);
}
if let Some(found) = self.previous_probe_read(key, prev_version) {
return Ok(found);
}
if let Some(found) = self.previous_probe_persistent(table, key, prev_version)? {
return Ok(found);
}
Ok(None)
}
}
impl StandardMultiStore {
#[inline]
fn previous_probe_commit(
&self,
table: EntryKind,
key: &EncodedKey,
prev_version: CommitVersion,
) -> Result<Option<Option<MultiVersionRow>>> {
let Some(commit) = &self.commit else {
return Ok(None);
};
Ok(match commit.get(table, key.as_ref(), prev_version)? {
VersionedGetResult::Value {
value,
version,
} => Some(Some(MultiVersionRow {
key: key.clone(),
row: EncodedRow(CowVec::new(value.to_vec())),
version,
})),
VersionedGetResult::Tombstone => Some(None),
VersionedGetResult::NotFound => None,
})
}
#[inline]
fn previous_probe_read(
&self,
key: &EncodedKey,
prev_version: CommitVersion,
) -> Option<Option<MultiVersionRow>> {
let read = self.read.as_ref()?;
match read.get(key, prev_version) {
VersionedGetResult::Value {
value,
version,
} => Some(Some(MultiVersionRow {
key: key.clone(),
row: EncodedRow(CowVec::new(value.to_vec())),
version,
})),
VersionedGetResult::Tombstone => Some(None),
VersionedGetResult::NotFound => None,
}
}
#[inline]
fn previous_probe_persistent(
&self,
table: EntryKind,
key: &EncodedKey,
prev_version: CommitVersion,
) -> Result<Option<Option<MultiVersionRow>>> {
let Some(persistent) = &self.persistent else {
return Ok(None);
};
Ok(match persistent.get(table, key.as_ref(), prev_version)? {
VersionedGetResult::Value {
value,
version,
} => {
if let Some(read) = &self.read {
read.insert(key.clone(), version, Some(value.clone()));
}
Some(Some(MultiVersionRow {
key: key.clone(),
row: EncodedRow(CowVec::new(value.to_vec())),
version,
}))
}
VersionedGetResult::Tombstone => Some(None),
VersionedGetResult::NotFound => None,
})
}
}
impl MultiVersionStore for StandardMultiStore {}
pub struct MultiVersionRangeIter {
store: StandardMultiStore,
cursor: MultiVersionRangeCursor,
range: EncodedKeyRange,
scope: MultiVersionScope,
batch_size: usize,
current_batch: Vec<MultiVersionRow>,
current_index: usize,
}
impl Iterator for MultiVersionRangeIter {
type Item = Result<MultiVersionRow>;
fn next(&mut self) -> Option<Self::Item> {
if self.current_index < self.current_batch.len() {
let item = self.current_batch[self.current_index].clone();
self.current_index += 1;
return Some(Ok(item));
}
if self.cursor.exhausted {
return None;
}
match self.store.range_next(&mut self.cursor, self.range.clone(), self.scope, self.batch_size as u64) {
Ok(batch) => {
if batch.items.is_empty() {
if self.cursor.exhausted {
return None;
}
return self.next();
}
self.current_batch = batch.items;
self.current_index = 0;
self.next()
}
Err(e) => Some(Err(e)),
}
}
}
pub struct MultiVersionRangeRevIter {
store: StandardMultiStore,
cursor: MultiVersionRangeCursor,
range: EncodedKeyRange,
scope: MultiVersionScope,
batch_size: usize,
current_batch: Vec<MultiVersionRow>,
current_index: usize,
}
impl Iterator for MultiVersionRangeRevIter {
type Item = Result<MultiVersionRow>;
fn next(&mut self) -> Option<Self::Item> {
if self.current_index < self.current_batch.len() {
let item = self.current_batch[self.current_index].clone();
self.current_index += 1;
return Some(Ok(item));
}
if self.cursor.exhausted {
return None;
}
match self.store.range_rev_next(
&mut self.cursor,
self.range.clone(),
self.scope,
self.batch_size as u64,
) {
Ok(batch) => {
if batch.items.is_empty() {
if self.cursor.exhausted {
return None;
}
return self.next();
}
self.current_batch = batch.items;
self.current_index = 0;
self.next()
}
Err(e) => Some(Err(e)),
}
}
}
fn classify_key_range(range: &EncodedKeyRange) -> EntryKind {
classify_range(range).unwrap_or(EntryKind::Multi)
}
fn make_range_bounds(range: &EncodedKeyRange) -> (Vec<u8>, Vec<u8>) {
let start = match &range.start {
Bound::Included(key) => key.as_ref().to_vec(),
Bound::Excluded(key) => key.as_ref().to_vec(),
Bound::Unbounded => vec![],
};
let end = match &range.end {
Bound::Included(key) => key.as_ref().to_vec(),
Bound::Excluded(key) => key.as_ref().to_vec(),
Bound::Unbounded => vec![0xFFu8; 256],
};
(start, end)
}
#[cfg(all(test, feature = "sqlite", not(target_arch = "wasm32")))]
mod cache_tests {
use std::collections::HashMap;
use reifydb_core::{
common::CommitVersion,
delta::Delta,
encoded::{key::EncodedKey, row::EncodedRow},
interface::{
catalog::{flow::FlowNodeId, id::TableId, shape::ShapeId},
store::{EntryKind, MultiVersionCommit},
},
key::{
EncodableKey, flow_node_internal_state::FlowNodeInternalStateKey,
flow_node_state::FlowNodeStateKey, row::RowKey,
},
};
use reifydb_value::{cow_vec, util::cowvec::CowVec};
use crate::{
MultiVersionScope,
store::{StandardMultiStore, multi::WARM_THRESHOLD},
tier::{RawEntry, TierStorage, VersionedGetResult, commit::buffer::MultiCommitBufferTier},
};
const SHAPE: ShapeId = ShapeId::Table(TableId(1));
fn commit_row(store: &StandardMultiStore, n: u64, version: u64) {
MultiVersionCommit::commit(
store,
cow_vec![Delta::Set {
key: RowKey::encoded(SHAPE, n),
row: EncodedRow(CowVec::new(format!("v{n}").into_bytes())),
}],
CommitVersion(version),
)
.unwrap();
}
fn flush(store: &StandardMultiStore, cutoff: CommitVersion) {
let commit = store.commit().expect("commit tier");
for kind in commit.list_all_entry_kinds().unwrap() {
let (to_persist, to_drop) = match commit {
MultiCommitBufferTier::Memory(s) => s.collect_evictable_below(kind, cutoff),
};
if to_drop.is_empty() {
continue;
}
if !to_persist.is_empty() {
let persistent = store.persistent().expect("persistent tier");
let mut by_version: HashMap<
CommitVersion,
HashMap<EntryKind, Vec<(EncodedKey, Option<CowVec<u8>>)>>,
> = HashMap::new();
for (key, version, value) in to_persist {
by_version
.entry(version)
.or_default()
.entry(kind)
.or_default()
.push((key, value));
}
for (version, batch) in by_version {
persistent.set(version, batch).unwrap();
}
}
for (key, _) in &to_drop {
store.invalidate_read_key(key);
}
commit.drop(HashMap::from([(kind, to_drop)])).unwrap();
}
}
#[test]
fn operator_drop_fully_removes_state_leaving_no_tombstone() {
let (store, _g) = StandardMultiStore::testing_memory_with_persistent_sqlite();
let node = FlowNodeId(7);
let table = EntryKind::Operator(node);
let data_key = FlowNodeStateKey::encoded(node, vec![1u8]);
let internal_key = FlowNodeInternalStateKey::encoded(node, vec![2u8]);
for v in [1u64, 2] {
MultiVersionCommit::commit(
&store,
cow_vec![Delta::Set {
key: data_key.clone(),
row: EncodedRow(CowVec::new(vec![v as u8])),
}],
CommitVersion(v),
)
.unwrap();
}
for v in [3u64, 4] {
MultiVersionCommit::commit(
&store,
cow_vec![Delta::Set {
key: internal_key.clone(),
row: EncodedRow(CowVec::new(vec![v as u8])),
}],
CommitVersion(v),
)
.unwrap();
}
let commit = store.commit().expect("commit tier");
assert!(!commit.get_all_versions(table, data_key.as_ref()).unwrap().is_empty());
assert!(!commit.get_all_versions(table, internal_key.as_ref()).unwrap().is_empty());
MultiVersionCommit::commit(
&store,
cow_vec![
Delta::Drop {
key: data_key.clone(),
},
Delta::Drop {
key: internal_key.clone(),
}
],
CommitVersion(5),
)
.unwrap();
assert!(
commit.get_all_versions(table, data_key.as_ref()).unwrap().is_empty(),
"operator data-state Drop must remove every version, not leave a tombstone"
);
assert!(
commit.get_all_versions(table, internal_key.as_ref()).unwrap().is_empty(),
"operator internal-state Drop must remove every version, not leave a tombstone"
);
}
#[test]
fn operator_remove_leaves_a_tombstone_in_commit_tier() {
let (store, _g) = StandardMultiStore::testing_memory_with_persistent_sqlite();
let node = FlowNodeId(8);
let table = EntryKind::Operator(node);
let key = FlowNodeInternalStateKey::encoded(node, vec![9u8]);
MultiVersionCommit::commit(
&store,
cow_vec![Delta::Set {
key: key.clone(),
row: EncodedRow(CowVec::new(vec![1u8])),
}],
CommitVersion(1),
)
.unwrap();
MultiVersionCommit::commit(
&store,
cow_vec![Delta::Remove {
key: key.clone(),
}],
CommitVersion(2),
)
.unwrap();
let commit = store.commit().expect("commit tier");
let versions = commit.get_all_versions(table, key.as_ref()).unwrap();
assert!(
versions.iter().any(|(_, value)| value.is_none()),
"Remove leaves a tombstone in the commit tier (the path Drop must avoid); versions={versions:?}"
);
}
#[test]
fn operator_state_drop_keeps_keyspace_bounded_under_churn() {
const ROUNDS: u64 = 200;
fn current_count(store: &StandardMultiStore, table: EntryKind) -> u64 {
match store.commit().expect("commit tier") {
MultiCommitBufferTier::Memory(s) => s.count_current(table).unwrap(),
}
}
fn churn(evict_with_drop: bool) -> u64 {
let (store, _g) = StandardMultiStore::testing_memory_with_persistent_sqlite();
let node = FlowNodeId(21);
let table = EntryKind::Operator(node);
let key_at = |round: u64| FlowNodeInternalStateKey::encoded(node, round.to_be_bytes().to_vec());
let mut version = 0u64;
for round in 0..ROUNDS {
version += 1;
MultiVersionCommit::commit(
&store,
cow_vec![Delta::Set {
key: key_at(round),
row: EncodedRow(CowVec::new(vec![1u8])),
}],
CommitVersion(version),
)
.unwrap();
if round > 0 {
version += 1;
let prev = key_at(round - 1);
let delta = if evict_with_drop {
Delta::Drop {
key: prev,
}
} else {
Delta::Remove {
key: prev,
}
};
MultiVersionCommit::commit(&store, cow_vec![delta], CommitVersion(version))
.unwrap();
}
}
current_count(&store, table)
}
let drop_live = churn(true);
let remove_live = churn(false);
assert!(
drop_live <= 2,
"Drop must keep the operator keyspace bounded to the live set; got {drop_live}"
);
assert!(
remove_live >= ROUNDS - 1,
"Remove leaves a tombstone per round (the path Drop avoids); got {remove_live} after {ROUNDS} rounds"
);
}
#[test]
fn warm_threshold_warms_only_buckets_above_threshold() {
const HEAVY: u64 = WARM_THRESHOLD + 64;
const LIGHT: u64 = 20;
let (store, _g) = StandardMultiStore::testing_memory_with_persistent_sqlite();
for n in 1..=HEAVY {
commit_row(&store, n, 1);
}
for n in 0..LIGHT {
commit_row(&store, (1u64 << 16) + n, 1);
}
flush(&store, CommitVersion(1));
let read = store.read.clone().expect("read tier configured");
let heavy_bucket = read.page_of_key(&RowKey::encoded(SHAPE, 1));
let light_bucket = read.page_of_key(&RowKey::encoded(SHAPE, 1u64 << 16));
assert_ne!(heavy_bucket, light_bucket, "the two row groups must land in different buckets");
assert!(!read.page_is_complete(heavy_bucket), "nothing is warm before the scan");
let scanned = store
.range(
RowKey::full_scan(SHAPE),
MultiVersionScope::AsOf {
read: CommitVersion(10),
},
32,
)
.collect::<Result<Vec<_>, _>>()
.unwrap();
assert_eq!(scanned.len() as u64, HEAVY + LIGHT, "the scan returns every row regardless of warming");
assert!(read.page_is_complete(heavy_bucket), "a bucket scanned past the threshold must be warmed");
assert!(
!read.page_is_complete(light_bucket),
"a bucket scanned below the threshold must not be warmed"
);
}
#[test]
fn operator_state_write_through_keeps_read_cache_warm() {
let (store, _g) = StandardMultiStore::testing_memory_with_persistent_sqlite();
let read = store.read.clone().expect("read tier configured");
let opkey = FlowNodeStateKey::new(FlowNodeId(7), vec![1, 2, 3]).encode();
MultiVersionCommit::commit(
&store,
cow_vec![Delta::Set {
key: opkey.clone(),
row: EncodedRow(CowVec::new(b"state-v10".to_vec())),
}],
CommitVersion(10),
)
.unwrap();
match read.get(&opkey, CommitVersion(10)) {
VersionedGetResult::Value {
value,
version,
} => {
assert_eq!(
value.as_ref(),
b"state-v10",
"the cached operator state must be the committed value"
);
assert_eq!(
version,
CommitVersion(10),
"the cached entry must carry the commit version"
);
}
other => {
panic!("operator state must be served from the read cache after commit, got {other:?}")
}
}
assert!(
matches!(read.get(&opkey, CommitVersion(9)), VersionedGetResult::NotFound),
"a pre-write snapshot read must miss the write-through entry, not see the newer value"
);
}
#[test]
fn source_row_write_clears_range_complete_on_its_page() {
let (store, _g) = StandardMultiStore::testing_memory_with_persistent_sqlite();
let read = store.read.clone().expect("read tier configured");
let neighbor = RowKey::encoded(SHAPE, 1);
let page = read.page_of_key(&neighbor);
assert_eq!(
read.page_of_key(&RowKey::encoded(SHAPE, 2)),
page,
"both source rows must share a page for this test to exercise flag-clearing"
);
read.populate_page(
page,
vec![RawEntry {
key: neighbor,
version: CommitVersion(1),
value: Some(CowVec::new(b"neighbor".to_vec())),
}],
true,
);
assert!(read.page_is_complete(page), "the page must start range-complete");
commit_row(&store, 2, 5);
assert!(
!read.page_is_complete(page),
"writing a source row into a range-complete page must clear the flag so the range cache re-warms"
);
}
}