reifydb_store_multi/tier/commit/
buffer.rs

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

reifydb_store_multi/tier/commit/
buffer.rs
