1use 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}