Skip to main content

reifydb_store_multi/tier/persistent/
mod.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::ops::Bound;
5
6use reifydb_codec::key::encoded::EncodedKey;
7use reifydb_core::{
8	common::CommitVersion,
9	interface::store::EntryKind,
10	key::{
11		row::{StoragePartitionedRowKey, StorageRowKey},
12		series::{StoragePartitionedSeriesKey, StorageSeriesKey},
13	},
14};
15use reifydb_runtime::shutdown::Shutdown;
16#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
17use reifydb_sqlite::{SqliteConfig, SqliteTempPathGuard};
18use reifydb_store::{coverage::cursor::Cursor, filter::KeyFilter, metrics::PageCacheMetrics};
19use reifydb_store_commit::{MultiVersionScope, RangeBatch, RangeCursor, RangeStop, TierBatch, VersionedGetResult};
20use reifydb_value::{Result, value::datetime::DateTime};
21
22use crate::{
23	filter::MultiKeys,
24	tier::{TierStorage, range::NarrowLayout},
25};
26
27#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
28pub mod sqlite;
29
30#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
31use sqlite::storage::SqlitePersistentStorage;
32
33pub struct NarrowRangeRequest<'a, K> {
34	pub table: EntryKind,
35	pub start: Bound<&'a K>,
36	pub end: Bound<&'a K>,
37	pub scope: MultiVersionScope,
38	pub batch_size: usize,
39	pub descending: bool,
40}
41
42#[derive(Clone)]
43#[cfg_attr(all(feature = "sqlite", not(target_arch = "wasm32")), repr(u8))]
44pub enum MultiPersistentTier {
45	#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
46	Sqlite(SqlitePersistentStorage) = 0,
47}
48
49#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
50impl MultiPersistentTier {
51	pub(crate) fn range_next_row(
52		&self,
53		cursor: &mut Cursor<RangeStop, StorageRowKey>,
54		request: NarrowRangeRequest<'_, StorageRowKey>,
55	) -> Result<RangeBatch<StorageRowKey>> {
56		match self {
57			Self::Sqlite(s) => s.range_chunk_row(cursor, request),
58		}
59	}
60
61	pub(crate) fn range_next_partitioned_row(
62		&self,
63		cursor: &mut Cursor<RangeStop, StoragePartitionedRowKey>,
64		request: NarrowRangeRequest<'_, StoragePartitionedRowKey>,
65	) -> Result<RangeBatch<StoragePartitionedRowKey>> {
66		match self {
67			Self::Sqlite(s) => s.range_chunk_partitioned(cursor, request),
68		}
69	}
70
71	pub(crate) fn range_next_series(
72		&self,
73		cursor: &mut Cursor<RangeStop, StorageSeriesKey>,
74		request: NarrowRangeRequest<'_, StorageSeriesKey>,
75	) -> Result<RangeBatch<StorageSeriesKey>> {
76		match self {
77			Self::Sqlite(s) => s.range_chunk_series(cursor, request),
78		}
79	}
80
81	pub(crate) fn range_next_partitioned_series(
82		&self,
83		cursor: &mut Cursor<RangeStop, StoragePartitionedSeriesKey>,
84		request: NarrowRangeRequest<'_, StoragePartitionedSeriesKey>,
85	) -> Result<RangeBatch<StoragePartitionedSeriesKey>> {
86		match self {
87			Self::Sqlite(s) => s.range_chunk_partitioned_series(cursor, request),
88		}
89	}
90}
91
92#[cfg(not(all(feature = "sqlite", not(target_arch = "wasm32"))))]
93impl MultiPersistentTier {
94	pub(crate) fn range_next_row(
95		&self,
96		_cursor: &mut Cursor<RangeStop, StorageRowKey>,
97		_request: NarrowRangeRequest<'_, StorageRowKey>,
98	) -> Result<RangeBatch<StorageRowKey>> {
99		match *self {}
100	}
101
102	pub(crate) fn range_next_partitioned_row(
103		&self,
104		_cursor: &mut Cursor<RangeStop, StoragePartitionedRowKey>,
105		_request: NarrowRangeRequest<'_, StoragePartitionedRowKey>,
106	) -> Result<RangeBatch<StoragePartitionedRowKey>> {
107		match *self {}
108	}
109
110	pub(crate) fn range_next_series(
111		&self,
112		_cursor: &mut Cursor<RangeStop, StorageSeriesKey>,
113		_request: NarrowRangeRequest<'_, StorageSeriesKey>,
114	) -> Result<RangeBatch<StorageSeriesKey>> {
115		match *self {}
116	}
117
118	pub(crate) fn range_next_partitioned_series(
119		&self,
120		_cursor: &mut Cursor<RangeStop, StoragePartitionedSeriesKey>,
121		_request: NarrowRangeRequest<'_, StoragePartitionedSeriesKey>,
122	) -> Result<RangeBatch<StoragePartitionedSeriesKey>> {
123		match *self {}
124	}
125}
126
127pub trait PersistentRangeLayout: NarrowLayout {
128	fn range_next(
129		persistent: &MultiPersistentTier,
130		cursor: &mut Cursor<RangeStop, Self>,
131		request: NarrowRangeRequest<'_, Self>,
132	) -> Result<RangeBatch<Self>>;
133}
134
135impl PersistentRangeLayout for StorageRowKey {
136	fn range_next(
137		persistent: &MultiPersistentTier,
138		cursor: &mut Cursor<RangeStop, Self>,
139		request: NarrowRangeRequest<'_, Self>,
140	) -> Result<RangeBatch<Self>> {
141		persistent.range_next_row(cursor, request)
142	}
143}
144
145impl PersistentRangeLayout for StoragePartitionedRowKey {
146	fn range_next(
147		persistent: &MultiPersistentTier,
148		cursor: &mut Cursor<RangeStop, Self>,
149		request: NarrowRangeRequest<'_, Self>,
150	) -> Result<RangeBatch<Self>> {
151		persistent.range_next_partitioned_row(cursor, request)
152	}
153}
154
155impl PersistentRangeLayout for StorageSeriesKey {
156	fn range_next(
157		persistent: &MultiPersistentTier,
158		cursor: &mut Cursor<RangeStop, Self>,
159		request: NarrowRangeRequest<'_, Self>,
160	) -> Result<RangeBatch<Self>> {
161		persistent.range_next_series(cursor, request)
162	}
163}
164
165impl PersistentRangeLayout for StoragePartitionedSeriesKey {
166	fn range_next(
167		persistent: &MultiPersistentTier,
168		cursor: &mut Cursor<RangeStop, Self>,
169		request: NarrowRangeRequest<'_, Self>,
170	) -> Result<RangeBatch<Self>> {
171		persistent.range_next_partitioned_series(cursor, request)
172	}
173}
174
175impl Shutdown for MultiPersistentTier {
176	fn shutdown(&self) {
177		match self {
178			#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
179			Self::Sqlite(s) => s.shutdown(),
180			#[cfg(not(all(feature = "sqlite", not(target_arch = "wasm32"))))]
181			_ => {}
182		}
183	}
184}
185
186#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
187impl MultiPersistentTier {
188	pub fn sqlite(config: SqliteConfig) -> Self {
189		Self::Sqlite(SqlitePersistentStorage::new(config))
190	}
191
192	pub fn sqlite_in_memory() -> (Self, SqliteTempPathGuard) {
193		let (storage, guard) = SqlitePersistentStorage::in_memory();
194		(Self::Sqlite(storage), guard)
195	}
196
197	pub fn sqlite_storage(&self) -> &SqlitePersistentStorage {
198		match self {
199			Self::Sqlite(storage) => storage,
200		}
201	}
202
203	pub fn filter(&self) -> &KeyFilter<MultiKeys> {
204		match self {
205			Self::Sqlite(storage) => storage.filter(),
206		}
207	}
208
209	pub fn page_cache_metrics(&self) -> PageCacheMetrics {
210		match self {
211			Self::Sqlite(storage) => storage.page_cache_metrics(),
212		}
213	}
214
215	pub fn set_checkpoint_threshold(&self, frames: u32) {
216		match self {
217			Self::Sqlite(s) => s.set_checkpoint_threshold(frames),
218		}
219	}
220	pub fn delete_keys(&self, table: EntryKind, keys: &[EncodedKey]) -> Result<u64> {
221		match self {
222			Self::Sqlite(s) => s.delete_keys(table, keys),
223		}
224	}
225
226	pub fn expired_keys(
227		&self,
228		table: EntryKind,
229		cutoff: DateTime,
230		cursor: Option<(DateTime, &[u8])>,
231		limit: usize,
232	) -> Result<Vec<(EncodedKey, DateTime)>> {
233		match self {
234			Self::Sqlite(s) => s.expired_keys(table, cutoff, cursor, limit),
235		}
236	}
237
238	pub fn list_current_entries(&self) -> Result<Vec<EntryKind>> {
239		match self {
240			Self::Sqlite(s) => s.list_current_entries(),
241		}
242	}
243
244	pub fn set_collecting_accepted(&self, version: CommitVersion, batches: TierBatch) -> Result<Vec<EncodedKey>> {
245		match self {
246			Self::Sqlite(s) => s.set_collecting_accepted(version, batches),
247		}
248	}
249
250	pub fn persist_sweep(&self, batches: Vec<(CommitVersion, TierBatch)>) -> Result<Vec<EncodedKey>> {
251		match self {
252			Self::Sqlite(s) => s.persist_sweep(batches),
253		}
254	}
255
256	pub fn install_floor(&self) -> Result<CommitVersion> {
257		match self {
258			Self::Sqlite(s) => s.install_floor(),
259		}
260	}
261}
262
263#[cfg(not(all(feature = "sqlite", not(target_arch = "wasm32"))))]
264impl MultiPersistentTier {
265	pub fn filter(&self) -> &KeyFilter<MultiKeys> {
266		match *self {}
267	}
268
269	pub fn page_cache_metrics(&self) -> PageCacheMetrics {
270		match *self {}
271	}
272
273	pub fn set_checkpoint_threshold(&self, _frames: u32) {
274		match *self {}
275	}
276	pub fn delete_keys(&self, _table: EntryKind, _keys: &[EncodedKey]) -> Result<u64> {
277		match *self {}
278	}
279
280	pub fn expired_keys(
281		&self,
282		_table: EntryKind,
283		_cutoff: DateTime,
284		_cursor: Option<(DateTime, &[u8])>,
285		_limit: usize,
286	) -> Result<Vec<(EncodedKey, DateTime)>> {
287		match *self {}
288	}
289
290	pub fn list_current_entries(&self) -> Result<Vec<EntryKind>> {
291		match *self {}
292	}
293
294	pub fn persist_sweep(&self, _batches: Vec<(CommitVersion, TierBatch)>) -> Result<Vec<EncodedKey>> {
295		match *self {}
296	}
297
298	pub fn install_floor(&self) -> Result<CommitVersion> {
299		match *self {}
300	}
301}
302
303#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
304impl TierStorage for MultiPersistentTier {
305	fn get(&self, table: EntryKind, key: &[u8], version: CommitVersion) -> Result<VersionedGetResult> {
306		match self {
307			Self::Sqlite(s) => s.get(table, key, version),
308		}
309	}
310
311	fn get_many(
312		&self,
313		table: EntryKind,
314		keys: &[&[u8]],
315		version: CommitVersion,
316	) -> Result<Vec<VersionedGetResult>> {
317		match self {
318			Self::Sqlite(s) => s.get_many(table, keys, version),
319		}
320	}
321
322	fn set(&self, version: CommitVersion, batches: TierBatch) -> Result<()> {
323		match self {
324			Self::Sqlite(s) => s.set(version, batches),
325		}
326	}
327
328	fn range_next(
329		&self,
330		table: EntryKind,
331		cursor: &mut RangeCursor,
332		start: Bound<&[u8]>,
333		end: Bound<&[u8]>,
334		scope: MultiVersionScope,
335		batch_size: usize,
336	) -> Result<RangeBatch> {
337		match self {
338			Self::Sqlite(s) => s.range_next(table, cursor, start, end, scope, batch_size),
339		}
340	}
341
342	fn range_rev_next(
343		&self,
344		table: EntryKind,
345		cursor: &mut RangeCursor,
346		start: Bound<&[u8]>,
347		end: Bound<&[u8]>,
348		scope: MultiVersionScope,
349		batch_size: usize,
350	) -> Result<RangeBatch> {
351		match self {
352			Self::Sqlite(s) => s.range_rev_next(table, cursor, start, end, scope, batch_size),
353		}
354	}
355
356	fn ensure_table(&self, table: EntryKind) -> Result<()> {
357		match self {
358			Self::Sqlite(s) => s.ensure_table(table),
359		}
360	}
361
362	fn clear_table(&self, table: EntryKind) -> Result<()> {
363		match self {
364			Self::Sqlite(s) => s.clear_table(table),
365		}
366	}
367}
368
369#[cfg(not(all(feature = "sqlite", not(target_arch = "wasm32"))))]
370impl TierStorage for MultiPersistentTier {
371	fn get(&self, _table: EntryKind, _key: &[u8], _version: CommitVersion) -> Result<VersionedGetResult> {
372		match *self {}
373	}
374
375	fn set(&self, _version: CommitVersion, _batches: TierBatch) -> Result<()> {
376		match *self {}
377	}
378
379	fn range_next(
380		&self,
381		_table: EntryKind,
382		_cursor: &mut RangeCursor,
383		_start: Bound<&[u8]>,
384		_end: Bound<&[u8]>,
385		_scope: MultiVersionScope,
386		_batch_size: usize,
387	) -> Result<RangeBatch> {
388		match *self {}
389	}
390
391	fn range_rev_next(
392		&self,
393		_table: EntryKind,
394		_cursor: &mut RangeCursor,
395		_start: Bound<&[u8]>,
396		_end: Bound<&[u8]>,
397		_scope: MultiVersionScope,
398		_batch_size: usize,
399	) -> Result<RangeBatch> {
400		match *self {}
401	}
402
403	fn ensure_table(&self, _table: EntryKind) -> Result<()> {
404		match *self {}
405	}
406
407	fn clear_table(&self, _table: EntryKind) -> Result<()> {
408		match *self {}
409	}
410}