reifydb_store_multi/tier/commit/
buffer.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::{collections::HashMap, ops::Bound};
5
6use reifydb_codec::key::encoded::EncodedKey;
7use reifydb_core::{common::CommitVersion, interface::store::EntryKind};
8use reifydb_value::{Result, util::cowvec::CowVec};
9
10use crate::{
11	MultiVersionScope,
12	tier::{
13		HistoricalCursor, RangeBatch, RangeCursor, TierBackend, TierBatch, TierStorage, VersionedGetResult,
14		commit::memory::storage::MemoryPrimitiveStorage,
15	},
16};
17
18#[derive(Clone)]
19#[repr(u8)]
20pub enum MultiCommitBufferTier {
21	Memory(MemoryPrimitiveStorage) = 0,
22}
23
24impl MultiCommitBufferTier {
25	pub fn memory() -> Self {
26		Self::Memory(MemoryPrimitiveStorage::new())
27	}
28}
29
30impl MultiCommitBufferTier {
31	pub fn maintenance(&self) {
32		match self {
33			Self::Memory(_) => {}
34		}
35	}
36
37	pub fn count_current(&self, table: EntryKind) -> Result<u64> {
38		match self {
39			Self::Memory(s) => s.count_current(table),
40		}
41	}
42
43	pub fn count_historical(&self, table: EntryKind) -> Result<u64> {
44		match self {
45			Self::Memory(s) => s.count_historical(table),
46		}
47	}
48
49	pub fn list_all_entry_kinds(&self) -> Result<Vec<EntryKind>> {
50		match self {
51			Self::Memory(s) => s.list_all_entry_kinds(),
52		}
53	}
54}
55
56impl TierStorage for MultiCommitBufferTier {
57	#[inline]
58	fn get(&self, table: EntryKind, key: &[u8], version: CommitVersion) -> Result<VersionedGetResult> {
59		match self {
60			Self::Memory(s) => s.get(table, key, version),
61		}
62	}
63
64	#[inline]
65	fn contains(&self, table: EntryKind, key: &[u8], version: CommitVersion) -> Result<bool> {
66		match self {
67			Self::Memory(s) => s.contains(table, key, version),
68		}
69	}
70
71	#[inline]
72	fn set(&self, version: CommitVersion, batches: TierBatch) -> Result<()> {
73		match self {
74			Self::Memory(s) => s.set(version, batches),
75		}
76	}
77
78	#[inline]
79	fn range_next(
80		&self,
81		table: EntryKind,
82		cursor: &mut RangeCursor,
83		start: Bound<&[u8]>,
84		end: Bound<&[u8]>,
85		scope: MultiVersionScope,
86		batch_size: usize,
87	) -> Result<RangeBatch> {
88		match self {
89			Self::Memory(s) => s.range_next(table, cursor, start, end, scope, batch_size),
90		}
91	}
92
93	#[inline]
94	fn range_rev_next(
95		&self,
96		table: EntryKind,
97		cursor: &mut RangeCursor,
98		start: Bound<&[u8]>,
99		end: Bound<&[u8]>,
100		scope: MultiVersionScope,
101		batch_size: usize,
102	) -> Result<RangeBatch> {
103		match self {
104			Self::Memory(s) => s.range_rev_next(table, cursor, start, end, scope, batch_size),
105		}
106	}
107
108	#[inline]
109	fn ensure_table(&self, table: EntryKind) -> Result<()> {
110		match self {
111			Self::Memory(s) => s.ensure_table(table),
112		}
113	}
114
115	#[inline]
116	fn clear_table(&self, table: EntryKind) -> Result<()> {
117		match self {
118			Self::Memory(s) => s.clear_table(table),
119		}
120	}
121
122	#[inline]
123	fn drop(&self, batches: HashMap<EntryKind, Vec<(EncodedKey, CommitVersion)>>) -> Result<()> {
124		match self {
125			Self::Memory(s) => s.drop(batches),
126		}
127	}
128
129	#[inline]
130	fn get_all_versions(&self, table: EntryKind, key: &[u8]) -> Result<Vec<(CommitVersion, Option<CowVec<u8>>)>> {
131		match self {
132			Self::Memory(s) => s.get_all_versions(table, key),
133		}
134	}
135
136	#[inline]
137	fn scan_historical_below(
138		&self,
139		table: EntryKind,
140		cutoff: CommitVersion,
141		cursor: &mut HistoricalCursor,
142		batch_size: usize,
143	) -> Result<Vec<(EncodedKey, CommitVersion)>> {
144		match self {
145			Self::Memory(s) => s.scan_historical_below(table, cutoff, cursor, batch_size),
146		}
147	}
148}
149
150impl TierBackend for MultiCommitBufferTier {}
151
152#[cfg(test)]
153pub mod tests {
154	use super::*;
155
156	#[test]
157	fn test_memory_backend() {
158		let storage = MultiCommitBufferTier::memory();
159
160		let key = EncodedKey::new(b"key".to_vec());
161		let version = CommitVersion(1);
162
163		storage.set(
164			version,
165			HashMap::from([(EntryKind::Multi, vec![(key.clone(), Some(CowVec::new(b"value".to_vec())))])]),
166		)
167		.unwrap();
168		assert_eq!(
169			storage.get(EntryKind::Multi, &key, version).unwrap().value().as_deref(),
170			Some(b"value".as_slice())
171		);
172	}
173
174	#[test]
175	fn test_range_next_memory() {
176		let storage = MultiCommitBufferTier::memory();
177
178		let version = CommitVersion(1);
179		storage.set(
180			version,
181			HashMap::from([(
182				EntryKind::Multi,
183				vec![
184					(EncodedKey::new(b"a".to_vec()), Some(CowVec::new(b"1".to_vec()))),
185					(EncodedKey::new(b"b".to_vec()), Some(CowVec::new(b"2".to_vec()))),
186					(EncodedKey::new(b"c".to_vec()), Some(CowVec::new(b"3".to_vec()))),
187				],
188			)]),
189		)
190		.unwrap();
191
192		let mut cursor = RangeCursor::new();
193		let batch = storage
194			.range_next(
195				EntryKind::Multi,
196				&mut cursor,
197				Bound::Unbounded,
198				Bound::Unbounded,
199				MultiVersionScope::AsOf {
200					read: version,
201				},
202				100,
203			)
204			.unwrap();
205
206		assert_eq!(batch.entries.len(), 3);
207		assert!(!batch.has_more);
208		assert!(cursor.exhausted);
209	}
210}
reifydb_store_multi/tier/commit/buffer.rs

reifydb_store_multi/tier/commit/
buffer.rs
