reifydb_store_multi/tier/commit/memory/
entry.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::{
5	cmp::Reverse,
6	collections::{BTreeMap, HashSet},
7	mem::size_of,
8	sync::{
9		Arc,
10		atomic::{AtomicU64, Ordering},
11	},
12};
13
14use reifydb_codec::key::encoded::EncodedKey;
15use reifydb_core::{common::CommitVersion, interface::store::EntryKind};
16use reifydb_runtime::sync::{
17	map::Map,
18	rwlock::{RwLock, RwLockWriteGuard},
19};
20use reifydb_value::util::cowvec::CowVec;
21use tracing::instrument;
22
23pub(super) type Value = Option<CowVec<u8>>;
24
25pub(super) type CurrentMap = BTreeMap<EncodedKey, (CommitVersion, Value)>;
26
27pub(super) type HistoricalMap = BTreeMap<EncodedKey, BTreeMap<Reverse<CommitVersion>, Value>>;
28
29pub(super) type OldestIndex = BTreeMap<CommitVersion, HashSet<EncodedKey>>;
30
31pub(super) const NODE_FILL_DIVISOR: usize = 2;
32
33pub(super) const ENTRY_OVERHEAD: usize =
34	NODE_FILL_DIVISOR * (size_of::<EncodedKey>() + size_of::<CommitVersion>() + size_of::<Value>());
35
36pub(super) fn oldest_version(
37	current: &CurrentMap,
38	historical: &HistoricalMap,
39	key: &EncodedKey,
40) -> Option<CommitVersion> {
41	let hist = historical.get(key).and_then(|m| m.keys().next_back()).map(|r| r.0);
42	let cur = current.get(key).map(|(v, _)| *v);
43	match (hist, cur) {
44		(Some(h), Some(c)) => Some(h.min(c)),
45		(Some(h), None) => Some(h),
46		(None, cur) => cur,
47	}
48}
49
50pub(super) fn reconcile_oldest(
51	index: &mut OldestIndex,
52	key: &EncodedKey,
53	old: Option<CommitVersion>,
54	new: Option<CommitVersion>,
55) {
56	if old == new {
57		return;
58	}
59	if let Some(old_v) = old
60		&& let Some(bucket) = index.get_mut(&old_v)
61	{
62		bucket.remove(key);
63		if bucket.is_empty() {
64			index.remove(&old_v);
65		}
66	}
67	if let Some(new_v) = new {
68		index.entry(new_v).or_default().insert(key.clone());
69	}
70}
71
72pub(super) fn entry_bytes(key: &EncodedKey, value: &Value) -> u64 {
73	entry_bytes_with(key.heap_bytes(), value)
74}
75
76pub(super) fn entry_bytes_with(key_heap: usize, value: &Value) -> u64 {
77	(ENTRY_OVERHEAD + key_heap + value.as_ref().map_or(0, |bytes| bytes.len())) as u64
78}
79
80pub(super) struct EntryBytes {
81	current: AtomicU64,
82	historical: AtomicU64,
83}
84
85impl EntryBytes {
86	fn new() -> Self {
87		Self {
88			current: AtomicU64::new(0),
89			historical: AtomicU64::new(0),
90		}
91	}
92
93	pub fn add_current(&self, bytes: u64) {
94		self.current.fetch_add(bytes, Ordering::Relaxed);
95	}
96
97	pub fn sub_current(&self, bytes: u64) {
98		saturating_sub(&self.current, bytes);
99	}
100
101	pub fn add_historical(&self, bytes: u64) {
102		self.historical.fetch_add(bytes, Ordering::Relaxed);
103	}
104
105	pub fn sub_historical(&self, bytes: u64) {
106		saturating_sub(&self.historical, bytes);
107	}
108
109	pub fn current(&self) -> u64 {
110		self.current.load(Ordering::Relaxed)
111	}
112
113	pub fn historical(&self) -> u64 {
114		self.historical.load(Ordering::Relaxed)
115	}
116
117	pub fn reset(&self) {
118		self.current.store(0, Ordering::Relaxed);
119		self.historical.store(0, Ordering::Relaxed);
120	}
121}
122
123fn saturating_sub(counter: &AtomicU64, amount: u64) {
124	let mut observed = counter.load(Ordering::Relaxed);
125	loop {
126		let next = observed.saturating_sub(amount);
127		match counter.compare_exchange_weak(observed, next, Ordering::Relaxed, Ordering::Relaxed) {
128			Ok(_) => return,
129			Err(actual) => observed = actual,
130		}
131	}
132}
133
134pub(super) struct Entry {
135	pub current: Arc<RwLock<CurrentMap>>,
136
137	pub historical: Arc<RwLock<HistoricalMap>>,
138
139	pub oldest: Arc<RwLock<OldestIndex>>,
140
141	pub bytes: Arc<EntryBytes>,
142}
143
144impl Entry {
145	pub fn new() -> Self {
146		Self {
147			current: Arc::new(RwLock::new(BTreeMap::new())),
148			historical: Arc::new(RwLock::new(BTreeMap::new())),
149			oldest: Arc::new(RwLock::new(BTreeMap::new())),
150			bytes: Arc::new(EntryBytes::new()),
151		}
152	}
153
154	#[instrument(name = "store::multi::memory::write_acquire", level = "debug", skip_all)]
155	pub fn write_pair(&self) -> (RwLockWriteGuard<'_, CurrentMap>, RwLockWriteGuard<'_, HistoricalMap>) {
156		(self.current.write(), self.historical.write())
157	}
158}
159
160impl Clone for Entry {
161	fn clone(&self) -> Self {
162		Self {
163			current: Arc::clone(&self.current),
164			historical: Arc::clone(&self.historical),
165			oldest: Arc::clone(&self.oldest),
166			bytes: Arc::clone(&self.bytes),
167		}
168	}
169}
170
171pub(super) struct Entries {
172	pub(super) data: Map<EntryKind, Entry>,
173}
174
175impl Default for Entries {
176	fn default() -> Self {
177		Self {
178			data: Map::new(),
179		}
180	}
181}
reifydb_store_multi/tier/commit/memory/entry.rs

reifydb_store_multi/tier/commit/memory/
entry.rs
