mwmr_core/
lib.rs

1#![allow(clippy::type_complexity)]
2
3use std::{collections::BTreeMap, hash::BuildHasher};
4
5use either::Either;
6use indexmap::IndexMap;
7use smallvec_wrapper::OneOrMore;
8
9/// Types
10pub mod types {
11  use super::*;
12  use cheap_clone::CheapClone;
13
14  /// A reference to a key.
15  pub struct KeyRef<'a, K: 'a> {
16    /// The key.
17    pub key: &'a K,
18    /// The version of the key.
19    pub version: u64,
20  }
21
22  impl<'a, K: 'a> KeyRef<'a, K> {
23    /// Returns the key.
24    pub fn key(&self) -> &K {
25      self.key
26    }
27
28    /// Returns the version of the key.
29    ///
30    /// This version is useful when you want to implement MVCC.
31    pub fn version(&self) -> u64 {
32      self.version
33    }
34  }
35
36  /// An item that is either prefetched or fetched from the database.
37  pub enum Item<'a, K, V, B, O> {
38    /// A pending item, which means that this item is still in a
39    /// transaction, and not yet commit to the database.
40    Pending(EntryRef<'a, K, V>),
41    /// An item comes from the database, some db implementation is
42    /// Copy-on-Write, so this enum varint allows such kind of behavior
43    Borrowed(B),
44    /// An item comes from the database, some db implementation is not
45    /// Copy-on-Write, so this enum varint allows such kind of behavior
46    Owned(O),
47  }
48
49  impl<'a, K, V, B, O> Item<'a, K, V, B, O> {
50    /// Returns the prefetched item.
51    ///
52    /// # Panic
53    /// If the item is not prefetched
54    pub fn unwrap_pending(&self) -> EntryRef<'a, K, V> {
55      match self {
56        Item::Pending(item) => *item,
57        _ => panic!("expected pending item"),
58      }
59    }
60
61    /// Returns the borrowed item.
62    ///
63    /// # Panic
64    /// If the item is not borrowed
65    pub fn unwrap_borrow(&self) -> &B {
66      match self {
67        Item::Borrowed(item) => item,
68        _ => panic!("expected borrowed item"),
69      }
70    }
71
72    /// Returns the owned item.
73    ///
74    /// # Panic
75    /// If the item is not owned
76    pub fn unwrap_owned(self) -> O {
77      match self {
78        Item::Owned(item) => item,
79        _ => panic!("expected owned item"),
80      }
81    }
82
83    /// Returns the owned item ref.
84    ///
85    /// # Panic
86    /// If the item is not owned
87    pub fn unwrap_owned_ref(&self) -> &O {
88      match self {
89        Item::Owned(item) => item,
90        _ => panic!("expected owned item"),
91      }
92    }
93
94    /// Returns the committed item.
95    ///
96    /// # Panic
97    /// If the item is not committed
98    pub fn unwrap_committed(self) -> Either<B, O> {
99      match self {
100        Item::Borrowed(item) => Either::Left(item),
101        Item::Owned(item) => Either::Right(item),
102        _ => panic!("expected committed item"),
103      }
104    }
105
106    /// Returns the committed item ref.
107    ///
108    /// # Panic
109    /// If the item is not committed
110    pub fn unwrap_committed_ref(&self) -> Either<&B, &O> {
111      match self {
112        Item::Borrowed(item) => Either::Left(item),
113        Item::Owned(item) => Either::Right(item),
114        _ => panic!("expected committed item"),
115      }
116    }
117  }
118
119  /// The reference of the [`Entry`].
120  pub struct EntryRef<'a, K, V> {
121    pub data: EntryDataRef<'a, K, V>,
122    pub version: u64,
123  }
124
125  impl<'a, K: core::fmt::Debug, V: core::fmt::Debug> core::fmt::Debug for EntryRef<'a, K, V> {
126    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
127      f.debug_struct("EntryRef")
128        .field("version", &self.version)
129        .field("data", &self.data)
130        .finish()
131    }
132  }
133
134  impl<'a, K, V> Clone for EntryRef<'a, K, V> {
135    fn clone(&self) -> Self {
136      *self
137    }
138  }
139
140  impl<'a, K, V> Copy for EntryRef<'a, K, V> {}
141
142  impl<'a, K, V> EntryRef<'a, K, V> {
143    /// Get the data of the entry.
144    #[inline]
145    pub const fn data(&self) -> &EntryDataRef<'a, K, V> {
146      &self.data
147    }
148
149    /// Get the value of the entry, if None, it means the entry is removed.
150    #[inline]
151    pub const fn value(&self) -> Option<&V> {
152      match self.data {
153        EntryDataRef::Insert { value, .. } => Some(value),
154        EntryDataRef::Remove(_) => None,
155      }
156    }
157
158    /// Returns the version of the entry.
159    ///
160    /// This version is useful when you want to implement MVCC.
161    #[inline]
162    pub const fn version(&self) -> u64 {
163      self.version
164    }
165  }
166
167  /// The reference of the [`EntryData`].
168  pub enum EntryDataRef<'a, K, V> {
169    /// Insert the key and the value.
170    Insert {
171      /// key of the entry.
172      key: &'a K,
173      /// value of the entry.
174      value: &'a V,
175    },
176    /// Remove the key.
177    Remove(&'a K),
178  }
179
180  impl<'a, K: core::fmt::Debug, V: core::fmt::Debug> core::fmt::Debug for EntryDataRef<'a, K, V> {
181    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
182      match self {
183        Self::Insert { key, value } => f
184          .debug_struct("Insert")
185          .field("key", key)
186          .field("value", value)
187          .finish(),
188        Self::Remove(key) => f.debug_tuple("Remove").field(key).finish(),
189      }
190    }
191  }
192
193  impl<'a, K, V> Clone for EntryDataRef<'a, K, V> {
194    fn clone(&self) -> Self {
195      *self
196    }
197  }
198
199  impl<'a, K, V> Copy for EntryDataRef<'a, K, V> {}
200
201  /// The data of the [`Entry`].
202  pub enum EntryData<K, V> {
203    /// Insert the key and the value.
204    Insert {
205      /// key of the entry.
206      key: K,
207      /// value of the entry.
208      value: V,
209    },
210    /// Remove the key.
211    Remove(K),
212  }
213
214  impl<K, V> EntryData<K, V> {
215    /// Returns the key of the entry.
216    #[inline]
217    pub const fn key(&self) -> &K {
218      match self {
219        Self::Insert { key, .. } => key,
220        Self::Remove(key) => key,
221      }
222    }
223  }
224
225  impl<K: core::fmt::Debug, V: core::fmt::Debug> core::fmt::Debug for EntryData<K, V> {
226    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
227      match self {
228        Self::Insert { key, value } => f
229          .debug_struct("Insert")
230          .field("key", key)
231          .field("value", value)
232          .finish(),
233        Self::Remove(key) => f.debug_tuple("Remove").field(key).finish(),
234      }
235    }
236  }
237
238  impl<K, V> Clone for EntryData<K, V>
239  where
240    K: Clone,
241    V: Clone,
242  {
243    fn clone(&self) -> Self {
244      match self {
245        Self::Insert { key, value } => Self::Insert {
246          key: key.clone(),
247          value: value.clone(),
248        },
249        Self::Remove(key) => Self::Remove(key.clone()),
250      }
251    }
252  }
253
254  impl<K, V> CheapClone for EntryData<K, V>
255  where
256    K: CheapClone,
257    V: CheapClone,
258  {
259    fn cheap_clone(&self) -> Self {
260      match self {
261        Self::Insert { key, value } => Self::Insert {
262          key: key.cheap_clone(),
263          value: value.cheap_clone(),
264        },
265        Self::Remove(key) => Self::Remove(key.cheap_clone()),
266      }
267    }
268  }
269
270  /// An entry can be persisted to the database.
271  pub struct Entry<K, V> {
272    pub version: u64,
273    pub data: EntryData<K, V>,
274  }
275
276  impl<K: core::fmt::Debug, V: core::fmt::Debug> core::fmt::Debug for Entry<K, V> {
277    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
278      f.debug_struct("Entry")
279        .field("version", &self.version)
280        .field("data", &self.data)
281        .finish()
282    }
283  }
284
285  impl<K, V> Clone for Entry<K, V>
286  where
287    K: Clone,
288    V: Clone,
289  {
290    fn clone(&self) -> Self {
291      Self {
292        version: self.version,
293        data: self.data.clone(),
294      }
295    }
296  }
297
298  impl<K, V> CheapClone for Entry<K, V>
299  where
300    K: CheapClone,
301    V: CheapClone,
302  {
303    fn cheap_clone(&self) -> Self {
304      Self {
305        version: self.version,
306        data: self.data.cheap_clone(),
307      }
308    }
309  }
310
311  impl<K, V> Entry<K, V> {
312    /// Returns the data contained by the entry.
313    #[inline]
314    pub const fn data(&self) -> &EntryData<K, V> {
315      &self.data
316    }
317
318    /// Returns the version (can also be tought as transaction timestamp) of the entry.
319    #[inline]
320    pub const fn version(&self) -> u64 {
321      self.version
322    }
323
324    /// Consumes the entry and returns the version and the entry data.
325    #[inline]
326    pub fn into_components(self) -> (u64, EntryData<K, V>) {
327      (self.version, self.data)
328    }
329
330    /// Returns the key of the entry.
331    #[inline]
332    pub fn key(&self) -> &K {
333      match &self.data {
334        EntryData::Insert { key, .. } => key,
335        EntryData::Remove(key) => key,
336      }
337    }
338
339    /// Split the entry into its key and [`EntryValue`].
340    pub fn split(self) -> (K, EntryValue<V>) {
341      let Entry { data, version } = self;
342
343      let (key, value) = match data {
344        EntryData::Insert { key, value } => (key, Some(value)),
345        EntryData::Remove(key) => (key, None),
346      };
347      (key, EntryValue { value, version })
348    }
349
350    /// Unsplit the key and [`EntryValue`] into an entry.
351    pub fn unsplit(key: K, value: EntryValue<V>) -> Self {
352      let EntryValue { value, version } = value;
353      Entry {
354        data: match value {
355          Some(value) => EntryData::Insert { key, value },
356          None => EntryData::Remove(key),
357        },
358        version,
359      }
360    }
361  }
362
363  /// A entry value
364  pub struct EntryValue<V> {
365    /// The version of the entry.
366    pub version: u64,
367    /// The value of the entry.
368    pub value: Option<V>,
369  }
370
371  impl<V: core::fmt::Debug> core::fmt::Debug for EntryValue<V> {
372    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
373      f.debug_struct("EntryValue")
374        .field("version", &self.version)
375        .field("value", &self.value)
376        .finish()
377    }
378  }
379
380  impl<V> Clone for EntryValue<V>
381  where
382    V: Clone,
383  {
384    fn clone(&self) -> Self {
385      Self {
386        version: self.version,
387        value: self.value.clone(),
388      }
389    }
390  }
391
392  impl<V> CheapClone for EntryValue<V>
393  where
394    V: CheapClone,
395  {
396    fn cheap_clone(&self) -> Self {
397      Self {
398        version: self.version,
399        value: self.value.cheap_clone(),
400      }
401    }
402  }
403
404  /// Used to set options when iterating over key-value
405  /// stores.
406  #[derive(Debug, Copy, Clone, Eq, PartialEq, Hash)]
407  pub struct KeysOptions {
408    /// The number of KV pairs to prefetch while iterating.
409    ///
410    /// Some databases optimize iteration by prefetching
411    pub prefetch_size: usize,
412    /// Direction of iteration. False is forward, true is backward.
413    pub reverse: bool,
414    /// Fetch all valid versions of the same key.
415    pub all_versions: bool,
416    /// Only read data that has version > `since_version`.
417    pub since_version: u64,
418  }
419
420  impl Default for KeysOptions {
421    fn default() -> Self {
422      Self::new()
423    }
424  }
425
426  impl KeysOptions {
427    /// Create a new iterator options with default values.
428    #[inline]
429    pub const fn new() -> Self {
430      Self {
431        prefetch_size: 0,
432        reverse: false,
433        all_versions: false,
434        since_version: 0,
435      }
436    }
437
438    /// Set the number of KV pairs to prefetch while iterating.
439    #[inline]
440    pub fn set_prefetch_size(&mut self, prefetch_size: usize) -> &mut Self {
441      self.prefetch_size = prefetch_size;
442      self
443    }
444
445    /// Set the number of KV pairs to prefetch while iterating.
446    #[inline]
447    pub const fn with_prefetch_size(mut self, prefetch_size: usize) -> Self {
448      self.prefetch_size = prefetch_size;
449      self
450    }
451
452    /// Set the direction of iteration. False is forward, true is backward.
453    #[inline]
454    pub fn set_reverse(&mut self, reverse: bool) -> &mut Self {
455      self.reverse = reverse;
456      self
457    }
458
459    /// Set the direction of iteration. False is forward, true is backward.
460    #[inline]
461    pub const fn with_reverse(mut self, reverse: bool) -> Self {
462      self.reverse = reverse;
463      self
464    }
465
466    /// Set whether to fetch all valid versions of the same key.
467    #[inline]
468    pub fn set_all_versions(&mut self, all_versions: bool) -> &mut Self {
469      self.all_versions = all_versions;
470      self
471    }
472
473    /// Set whether to fetch all valid versions of the same key.
474    #[inline]
475    pub const fn with_all_versions(mut self, all_versions: bool) -> Self {
476      self.all_versions = all_versions;
477      self
478    }
479
480    /// Set the version to start reading from.
481    #[inline]
482    pub fn set_since_version(&mut self, since_version: u64) -> &mut Self {
483      self.since_version = since_version;
484      self
485    }
486
487    /// Set the version to start reading from.
488    #[inline]
489    pub const fn with_since_version(mut self, since_version: u64) -> Self {
490      self.since_version = since_version;
491      self
492    }
493  }
494
495  /// Used to set options when iterating over key-value
496  /// stores.
497  #[derive(Debug, Copy, Clone, Eq, PartialEq, Hash)]
498  pub struct IteratorOptions {
499    /// The number of KV pairs to prefetch while iterating.
500    ///
501    /// Some databases optimize iteration by prefetching
502    pub prefetch_size: usize,
503    /// Indicates whether we should prefetch values during
504    /// iteration and store them.
505    ///
506    /// Some databases use key-value separation for optimization
507    /// and this option can be used to prefetch values.
508    pub prefetch_values: bool,
509    /// Direction of iteration. False is forward, true is backward.
510    pub reverse: bool,
511    /// Fetch all valid versions of the same key.
512    pub all_versions: bool,
513    /// Only read data that has version > `since_version`.
514    pub since_version: u64,
515  }
516
517  impl Default for IteratorOptions {
518    fn default() -> Self {
519      Self::new()
520    }
521  }
522
523  impl IteratorOptions {
524    /// Create a new iterator options with default values.
525    #[inline]
526    pub const fn new() -> Self {
527      Self {
528        prefetch_size: 0,
529        prefetch_values: false,
530        reverse: false,
531        all_versions: false,
532        since_version: 0,
533      }
534    }
535
536    /// Set the number of KV pairs to prefetch while iterating.
537    #[inline]
538    pub fn set_prefetch_size(&mut self, prefetch_size: usize) -> &mut Self {
539      self.prefetch_size = prefetch_size;
540      self
541    }
542
543    /// Set the number of KV pairs to prefetch while iterating.
544    #[inline]
545    pub const fn with_prefetch_size(mut self, prefetch_size: usize) -> Self {
546      self.prefetch_size = prefetch_size;
547      self
548    }
549
550    /// Set whether we should prefetch values during iteration and store them.
551    #[inline]
552    pub fn set_prefetch_values(&mut self, prefetch_values: bool) -> &mut Self {
553      self.prefetch_values = prefetch_values;
554      self
555    }
556
557    /// Set whether we should prefetch values during iteration and store them.
558    #[inline]
559    pub const fn with_prefetch_values(mut self, prefetch_values: bool) -> Self {
560      self.prefetch_values = prefetch_values;
561      self
562    }
563
564    /// Set the direction of iteration. False is forward, true is backward.
565    #[inline]
566    pub fn set_reverse(&mut self, reverse: bool) -> &mut Self {
567      self.reverse = reverse;
568      self
569    }
570
571    /// Set the direction of iteration. False is forward, true is backward.
572    #[inline]
573    pub const fn with_reverse(mut self, reverse: bool) -> Self {
574      self.reverse = reverse;
575      self
576    }
577
578    /// Set whether to fetch all valid versions of the same key.
579    #[inline]
580    pub fn set_all_versions(&mut self, all_versions: bool) -> &mut Self {
581      self.all_versions = all_versions;
582      self
583    }
584
585    /// Set whether to fetch all valid versions of the same key.
586    #[inline]
587    pub const fn with_all_versions(mut self, all_versions: bool) -> Self {
588      self.all_versions = all_versions;
589      self
590    }
591
592    /// Set the version to start reading from.
593    #[inline]
594    pub fn set_since_version(&mut self, since_version: u64) -> &mut Self {
595      self.since_version = since_version;
596      self
597    }
598
599    /// Set the version to start reading from.
600    #[inline]
601    pub const fn with_since_version(mut self, since_version: u64) -> Self {
602      self.since_version = since_version;
603      self
604    }
605  }
606}
607
608/// Traits for synchronization.
609pub mod sync {
610  use super::{types::*, *};
611
612  /// A pending writes manager that can be used to store pending writes in a transaction.
613  ///
614  /// By default, there are two implementations of this trait:
615  /// - [`IndexMap`]: A hash map with consistent ordering and fast lookups.
616  /// - [`BTreeMap`]: A balanced binary tree with ordered keys and fast lookups.
617  ///
618  /// But, users can create their own implementations by implementing this trait.
619  /// e.g. if you want to implement a recovery transaction manager, you can use a persistent
620  /// storage to store the pending writes.
621  pub trait PendingManager: 'static {
622    /// The error type returned by the pending manager.
623    type Error: std::error::Error + 'static;
624    /// The key type.
625    type Key: 'static;
626    /// The value type.
627    type Value: 'static;
628
629    /// Returns true if the buffer is empty.
630    fn is_empty(&self) -> bool;
631
632    /// Returns the number of elements in the buffer.
633    fn len(&self) -> usize;
634
635    /// Returns a reference to the value corresponding to the key.
636    fn get(&self, key: &Self::Key) -> Result<Option<&EntryValue<Self::Value>>, Self::Error>;
637
638    /// Inserts a key-value pair into the buffer.
639    fn insert(&mut self, key: Self::Key, value: EntryValue<Self::Value>)
640      -> Result<(), Self::Error>;
641
642    /// Removes a key from the buffer, returning the key-value pair if the key was previously in the buffer.
643    fn remove_entry(
644      &mut self,
645      key: &Self::Key,
646    ) -> Result<Option<(Self::Key, EntryValue<Self::Value>)>, Self::Error>;
647
648    /// Returns an iterator over the keys in the buffer.
649    fn keys(&self) -> impl Iterator<Item = &'_ Self::Key>;
650
651    /// Returns an iterator over the key-value pairs in the buffer.
652    fn iter(&self) -> impl Iterator<Item = (&'_ Self::Key, &'_ EntryValue<Self::Value>)>;
653
654    /// Returns an iterator that consumes the buffer.
655    fn into_iter(self) -> impl Iterator<Item = (Self::Key, EntryValue<Self::Value>)>;
656  }
657
658  /// A type alias for [`PendingManager`] that based on the [`IndexMap`].
659  pub type IndexMapManager<K, V, S = std::hash::RandomState> = IndexMap<K, EntryValue<V>, S>;
660  /// A type alias for [`PendingManager`] that based on the [`BTreeMap`].
661  pub type BTreeMapManager<K, V> = BTreeMap<K, EntryValue<V>>;
662
663  impl<K, V, S> PendingManager for IndexMap<K, EntryValue<V>, S>
664  where
665    K: Eq + core::hash::Hash + 'static,
666    V: 'static,
667    S: BuildHasher + Default + 'static,
668  {
669    type Error = std::convert::Infallible;
670    type Key = K;
671    type Value = V;
672
673    fn is_empty(&self) -> bool {
674      self.is_empty()
675    }
676
677    fn len(&self) -> usize {
678      self.len()
679    }
680
681    fn get(&self, key: &K) -> Result<Option<&EntryValue<V>>, Self::Error> {
682      Ok(self.get(key))
683    }
684
685    fn insert(&mut self, key: K, value: EntryValue<V>) -> Result<(), Self::Error> {
686      self.insert(key, value);
687      Ok(())
688    }
689
690    fn remove_entry(&mut self, key: &K) -> Result<Option<(K, EntryValue<V>)>, Self::Error> {
691      Ok(self.shift_remove_entry(key))
692    }
693
694    fn keys(&self) -> impl Iterator<Item = &K> {
695      self.keys()
696    }
697
698    fn iter(&self) -> impl Iterator<Item = (&K, &EntryValue<V>)> {
699      self.iter()
700    }
701
702    fn into_iter(self) -> impl Iterator<Item = (K, EntryValue<V>)> {
703      core::iter::IntoIterator::into_iter(self)
704    }
705  }
706
707  impl<K, V> PendingManager for BTreeMap<K, EntryValue<V>>
708  where
709    K: Eq + core::hash::Hash + Ord + 'static,
710    V: 'static,
711  {
712    type Error = std::convert::Infallible;
713    type Key = K;
714    type Value = V;
715
716    fn is_empty(&self) -> bool {
717      self.is_empty()
718    }
719
720    fn len(&self) -> usize {
721      self.len()
722    }
723
724    fn get(&self, key: &K) -> Result<Option<&EntryValue<Self::Value>>, Self::Error> {
725      Ok(self.get(key))
726    }
727
728    fn insert(&mut self, key: K, value: EntryValue<Self::Value>) -> Result<(), Self::Error> {
729      self.insert(key, value);
730      Ok(())
731    }
732
733    fn remove_entry(
734      &mut self,
735      key: &K,
736    ) -> Result<Option<(K, EntryValue<Self::Value>)>, Self::Error> {
737      Ok(self.remove_entry(key))
738    }
739
740    fn keys(&self) -> impl Iterator<Item = &K> {
741      self.keys()
742    }
743
744    fn iter(&self) -> impl Iterator<Item = (&K, &EntryValue<Self::Value>)> {
745      self.iter()
746    }
747
748    fn into_iter(self) -> impl Iterator<Item = (K, EntryValue<Self::Value>)> {
749      core::iter::IntoIterator::into_iter(self)
750    }
751  }
752
753  /// An abstraction of database which can be managed by the [`TransactionDB`].
754  pub trait Database: Sized + 'static {
755    /// The error type returned by the database.
756    type Error: std::error::Error + 'static;
757    /// The options type of the database, which used to create the database.
758    type Options;
759    /// The key type of the database.
760    type Key: core::fmt::Debug + 'static;
761    /// The value type of the database.
762    type Value: core::fmt::Debug + 'static;
763    /// The owned item type can be returned by `get`.
764    type Item: 'static;
765    /// The reference item type can be returned by `get`.
766    type ItemRef<'a>
767    where
768      Self: 'a;
769    /// The iterator type of the database.
770    type Iterator<'a>
771    where
772      Self: 'a;
773    /// The key iterator type of the database.
774    type Keys<'a>
775    where
776      Self: 'a;
777
778    /// Returns the maximum batch size in bytes
779    fn max_batch_size(&self) -> u64;
780
781    /// Returns the maximum entries in batch
782    fn max_batch_entries(&self) -> u64;
783
784    /// Returns the estimated size of the entry in bytes when persisted in the database.
785    fn estimate_size(&self, entry: &Entry<Self::Key, Self::Value>) -> u64;
786
787    /// Validate if the entry is valid for this database.
788    ///
789    /// e.g.
790    /// - If the entry is expired
791    /// - If the key or the value is too large
792    /// - If the key or the value is empty
793    /// - If the key or the value contains invalid characters
794    /// - and etc.
795    fn validate_entry(&self, entry: &Entry<Self::Key, Self::Value>) -> Result<(), Self::Error>;
796
797    /// Returns the maximum version of the entry in the database, if you are not implementing MVCC, you can just ignore this method.
798    fn maximum_version(&self) -> u64;
799
800    /// Returns the options of the database.
801    fn options(&self) -> &Self::Options;
802
803    /// Open the database with the given options.
804    fn open(opts: Self::Options) -> Result<Self, Self::Error>;
805
806    /// Returns the fingerprint of key.
807    ///
808    /// Implementors should ensure that the fingerprint is consistent for the same key.
809    fn fingerprint(&self, k: &Self::Key) -> u64;
810
811    /// Applies a series of entries to the database. This method will be invoked in [`Mwmr::commit`] method.
812    ///
813    /// This method is responsible for persisting a batch of entries to the database. It is called
814    /// after the entries have been prepared and serialized by a higher-level [`Mwmr`] transaction manager,
815    /// ensuring that consistency and isolation requirements are already satisfied. Users of this
816    /// method do not need to handle these concerns; they can simply pass the entries to be applied.
817    ///
818    /// # Implementation Notes
819    ///
820    /// Implementors of this method must ensure atomicity of the apply operation; either all entries
821    /// are applied successfully, or none are, to prevent partial updates to the database state. It is
822    /// assumed that the consistency and isolation levels required for the entries have been managed
823    /// by a higher-level transaction manager before invocation of this method.
824    fn apply(&self, entries: OneOrMore<Entry<Self::Key, Self::Value>>) -> Result<(), Self::Error>;
825
826    /// Get the item from the database by the key and the version (version can be used for MVCC).
827    fn get(
828      &self,
829      k: &Self::Key,
830      version: u64,
831    ) -> Result<Option<Either<Self::ItemRef<'_>, Self::Item>>, Self::Error>;
832
833    /// Accepts an iterator of pending and returns an combined iterator.
834    ///
835    /// It depends on the database implementation to decide how to handle the `pending` and construct
836    /// the final conbined iterator.
837    ///
838    /// The order of the items in the iterator depends on the [`PendingManager`] of the [`WriteTransaction`].
839    ///
840    /// e.g.
841    /// - if users create [`WriteTransaction`] with [`IndexMap`] as the [`PendingManager`], the order of the
842    /// entries in the iterator will be the same as the insertion order.
843    /// - if users create [`WriteTransaction`] with [`BTreeCache`] as the [`PendingManager`], the order of the
844    /// entires in the iterator will be sorted by key.
845    fn iter<'a, 'b: 'a>(
846      &'a self,
847      pending: impl Iterator<Item = EntryRef<'b, Self::Key, Self::Value>> + 'b,
848      transaction_version: u64,
849      opts: IteratorOptions,
850    ) -> Self::Iterator<'a>;
851
852    /// Accepts an iterator of pending keys and returns an combined iterator.
853    ///
854    /// It depends on the database implementation to decide how to handle the `pending` and construct
855    /// the final conbined iterator.
856    ///
857    /// The order of the items in the iterator depends on the [`PendingManager`] of the [`WriteTransaction`].
858    ///
859    /// e.g.
860    /// - if users create [`WriteTransaction`] with [`IndexMap`] as the [`PendingManager`], the order of the
861    /// keys in the iterator will be the same as the insertion order.
862    /// - if users create [`WriteTransaction`] with [`BTreeCache`] as the [`PendingManager`], the order of the
863    /// keys in the iterator will be sorted by key.
864    fn keys<'a, 'b: 'a>(
865      &'a self,
866      pending: impl Iterator<Item = KeyRef<'b, Self::Key>> + 'b,
867      transaction_version: u64,
868      opts: KeysOptions,
869    ) -> Self::Keys<'a>;
870  }
871}
872
873/// Traits for asynchronous.
874pub mod future {
875  use super::{types::*, *};
876  use core::future::Future;
877
878  /// A pending writes manager that can be used to store pending writes in a transaction.
879  ///
880  /// By default, there are two implementations of this trait:
881  /// - [`IndexMap`]: A hash map with consistent ordering and fast lookups.
882  /// - [`BTreeMap`]: A balanced binary tree with ordered keys and fast lookups.
883  ///
884  /// But, users can create their own implementations by implementing this trait.
885  /// e.g. if you want to implement a recovery transaction manager, you can use a persistent
886  /// storage to store the pending writes.
887  pub trait AsyncPendingManager: Send + Sync + 'static {
888    /// The error type returned by the pending manager.
889    type Error: std::error::Error + Send + Sync + 'static;
890    /// The key type.
891    type Key: Send + Sync + 'static;
892    /// The value type.
893    type Value: Send + Sync + 'static;
894
895    /// Returns true if the buffer is empty.
896    fn is_empty(&self) -> bool;
897
898    /// Returns the number of elements in the buffer.
899    fn len(&self) -> usize;
900
901    /// Returns a reference to the value corresponding to the key.
902    fn get(
903      &self,
904      key: &Self::Key,
905    ) -> impl Future<Output = Result<Option<&EntryValue<Self::Value>>, Self::Error>> + Send;
906
907    /// Inserts a key-value pair into the buffer.
908    fn insert(
909      &mut self,
910      key: Self::Key,
911      value: EntryValue<Self::Value>,
912    ) -> impl Future<Output = Result<(), Self::Error>> + Send;
913
914    /// Removes a key from the buffer, returning the key-value pair if the key was previously in the buffer.
915    fn remove_entry(
916      &mut self,
917      key: &Self::Key,
918    ) -> impl Future<Output = Result<Option<(Self::Key, EntryValue<Self::Value>)>, Self::Error>> + Send;
919
920    /// Returns an iterator over the keys in the buffer.
921    fn keys(&self) -> impl Future<Output = impl Iterator<Item = &'_ Self::Key>> + Send;
922
923    /// Returns an iterator over the key-value pairs in the buffer.
924    fn iter(
925      &self,
926    ) -> impl Future<Output = impl Iterator<Item = (&'_ Self::Key, &'_ EntryValue<Self::Value>)>> + Send;
927
928    /// Returns an iterator that consumes the buffer.
929    fn into_iter(
930      self,
931    ) -> impl Future<Output = impl Iterator<Item = (Self::Key, EntryValue<Self::Value>)>> + Send;
932  }
933
934  /// A type alias for [`PendingManager`] that based on the [`IndexMap`].
935  pub type AsyncIndexMapManager<K, V, S = std::hash::RandomState> = IndexMap<K, EntryValue<V>, S>;
936  /// A type alias for [`PendingManager`] that based on the [`BTreeMap`].
937  pub type AsyncBTreeMapManager<K, V> = BTreeMap<K, EntryValue<V>>;
938
939  impl<K, V, S> AsyncPendingManager for IndexMap<K, EntryValue<V>, S>
940  where
941    K: Eq + core::hash::Hash + Send + Sync + 'static,
942    V: Send + Sync + 'static,
943    S: BuildHasher + Default + Send + Sync + 'static,
944  {
945    type Error = std::convert::Infallible;
946    type Key = K;
947    type Value = V;
948
949    fn is_empty(&self) -> bool {
950      self.is_empty()
951    }
952
953    fn len(&self) -> usize {
954      self.len()
955    }
956
957    async fn get(&self, key: &K) -> Result<Option<&EntryValue<V>>, Self::Error> {
958      Ok(self.get(key))
959    }
960
961    async fn insert(&mut self, key: K, value: EntryValue<V>) -> Result<(), Self::Error> {
962      self.insert(key, value);
963      Ok(())
964    }
965
966    async fn remove_entry(&mut self, key: &K) -> Result<Option<(K, EntryValue<V>)>, Self::Error> {
967      Ok(self.shift_remove_entry(key))
968    }
969
970    async fn keys(&self) -> impl Iterator<Item = &K> {
971      self.keys()
972    }
973
974    async fn iter(&self) -> impl Iterator<Item = (&K, &EntryValue<V>)> {
975      self.iter()
976    }
977
978    async fn into_iter(self) -> impl Iterator<Item = (K, EntryValue<V>)> {
979      core::iter::IntoIterator::into_iter(self)
980    }
981  }
982
983  impl<K, V> AsyncPendingManager for BTreeMap<K, EntryValue<V>>
984  where
985    K: Eq + core::hash::Hash + Ord + Send + Sync + 'static,
986    V: Send + Sync + 'static,
987  {
988    type Error = std::convert::Infallible;
989    type Key = K;
990    type Value = V;
991
992    fn is_empty(&self) -> bool {
993      self.is_empty()
994    }
995
996    fn len(&self) -> usize {
997      self.len()
998    }
999
1000    async fn get(&self, key: &K) -> Result<Option<&EntryValue<Self::Value>>, Self::Error> {
1001      Ok(self.get(key))
1002    }
1003
1004    async fn insert(&mut self, key: K, value: EntryValue<Self::Value>) -> Result<(), Self::Error> {
1005      self.insert(key, value);
1006      Ok(())
1007    }
1008
1009    async fn remove_entry(
1010      &mut self,
1011      key: &K,
1012    ) -> Result<Option<(K, EntryValue<Self::Value>)>, Self::Error> {
1013      Ok(self.remove_entry(key))
1014    }
1015
1016    async fn keys(&self) -> impl Iterator<Item = &K> {
1017      self.keys()
1018    }
1019
1020    async fn iter(&self) -> impl Iterator<Item = (&K, &EntryValue<Self::Value>)> {
1021      self.iter()
1022    }
1023
1024    async fn into_iter(self) -> impl Iterator<Item = (K, EntryValue<Self::Value>)> {
1025      core::iter::IntoIterator::into_iter(self)
1026    }
1027  }
1028
1029  /// An abstraction of database which can be managed by the [`TransactionDB`].
1030  pub trait AsyncDatabase: Sized + Send + Sync + 'static {
1031    /// The error type returned by the database.
1032    type Error: std::error::Error + Send + Sync + 'static;
1033    /// The options type of the database, which used to create the database.
1034    type Options;
1035    /// The key type of the database.
1036    type Key: core::fmt::Debug + Send + Sync + 'static;
1037    /// The value type of the database.
1038    type Value: core::fmt::Debug + Send + Sync + 'static;
1039    /// The owned item type can be returned by `get`.
1040    type Item: Send + Sync + 'static;
1041    /// The reference item type can be returned by `get`.
1042    type ItemRef<'a>: Send + Sync
1043    where
1044      Self: 'a;
1045    /// The iterator type of the database.
1046    type Iterator<'a>: Send + Sync
1047    where
1048      Self: 'a;
1049    /// The key iterator type of the database.
1050    type Keys<'a>: Send + Sync
1051    where
1052      Self: 'a;
1053
1054    /// Returns the maximum batch size in bytes
1055    fn max_batch_size(&self) -> u64;
1056
1057    /// Returns the maximum entries in batch
1058    fn max_batch_entries(&self) -> u64;
1059
1060    /// Returns the estimated size of the entry in bytes when persisted in the database.
1061    fn estimate_size(&self, entry: &Entry<Self::Key, Self::Value>) -> u64;
1062
1063    /// Validate if the entry is valid for this database.
1064    ///
1065    /// e.g.
1066    /// - If the entry is expired
1067    /// - If the key or the value is too large
1068    /// - If the key or the value is empty
1069    /// - If the key or the value contains invalid characters
1070    /// - and etc.
1071    fn validate_entry(&self, entry: &Entry<Self::Key, Self::Value>) -> Result<(), Self::Error>;
1072
1073    /// Returns the maximum version of the entry in the database, if you are not implementing MVCC, you can just ignore this method.
1074    fn maximum_version(&self) -> u64;
1075
1076    /// Returns the options of the database.
1077    fn options(&self) -> &Self::Options;
1078
1079    /// Open the database with the given options.
1080    fn open(opts: Self::Options) -> impl Future<Output = Result<Self, Self::Error>> + Send;
1081
1082    /// Returns the fingerprint of key.
1083    ///
1084    /// Implementors should ensure that the fingerprint is consistent for the same key.
1085    fn fingerprint(&self, k: &Self::Key) -> u64;
1086
1087    /// Applies a series of entries to the database. This method will be invoked in [`Mwmr::commit`] method.
1088    ///
1089    /// This method is responsible for persisting a batch of entries to the database. It is called
1090    /// after the entries have been prepared and serialized by a higher-level [`Mwmr`] transaction manager,
1091    /// ensuring that consistency and isolation requirements are already satisfied. Users of this
1092    /// method do not need to handle these concerns; they can simply pass the entries to be applied.
1093    ///
1094    /// # Implementation Notes
1095    ///
1096    /// Implementors of this method must ensure atomicity of the apply operation; either all entries
1097    /// are applied successfully, or none are, to prevent partial updates to the database state. It is
1098    /// assumed that the consistency and isolation levels required for the entries have been managed
1099    /// by a higher-level transaction manager before invocation of this method.
1100    fn apply(
1101      &self,
1102      entries: OneOrMore<Entry<Self::Key, Self::Value>>,
1103    ) -> impl Future<Output = Result<(), Self::Error>> + Send;
1104
1105    /// Get the item from the database by the key and the version (version can be used for MVCC).
1106    fn get(
1107      &self,
1108      k: &Self::Key,
1109      version: u64,
1110    ) -> impl Future<Output = Result<Option<Either<Self::ItemRef<'_>, Self::Item>>, Self::Error>> + Send;
1111
1112    /// Accepts an iterator of pending and returns an combined iterator.
1113    ///
1114    /// It depends on the database implementation to decide how to handle the `pending` and construct
1115    /// the final conbined iterator.
1116    ///
1117    /// The order of the items in the iterator depends on the [`PendingManager`] of the [`WriteTransaction`].
1118    ///
1119    /// e.g.
1120    /// - if users create [`WriteTransaction`] with [`IndexMap`] as the [`PendingManager`], the order of the
1121    /// entries in the iterator will be the same as the insertion order.
1122    /// - if users create [`WriteTransaction`] with [`BTreeCache`] as the [`PendingManager`], the order of the
1123    /// entires in the iterator will be sorted by key.
1124    fn iter<'a, 'b: 'a>(
1125      &'a self,
1126      pending: impl Iterator<Item = EntryRef<'b, Self::Key, Self::Value>> + 'b,
1127      transaction_version: u64,
1128      opts: IteratorOptions,
1129    ) -> impl Future<Output = Self::Iterator<'a>> + 'a;
1130
1131    /// Accepts an iterator of pending keys and returns an combined iterator.
1132    ///
1133    /// It depends on the database implementation to decide how to handle the `pending` and construct
1134    /// the final conbined iterator.
1135    ///
1136    /// The order of the items in the iterator depends on the [`PendingManager`] of the [`WriteTransaction`].
1137    ///
1138    /// e.g.
1139    /// - if users create [`WriteTransaction`] with [`IndexMap`] as the [`PendingManager`], the order of the
1140    /// keys in the iterator will be the same as the insertion order.
1141    /// - if users create [`WriteTransaction`] with [`BTreeCache`] as the [`PendingManager`], the order of the
1142    /// keys in the iterator will be sorted by key.
1143    fn keys<'a, 'b: 'a>(
1144      &'a self,
1145      pending: impl Iterator<Item = KeyRef<'b, Self::Key>> + 'b,
1146      transaction_version: u64,
1147      opts: KeysOptions,
1148    ) -> impl Future<Output = Self::Keys<'a>> + 'a;
1149  }
1150}
mwmr_core/lib.rs

mwmr_core/
lib.rs
