Serializable Snapshot Isolation (SSI)

Cargo.toml

@@ -17,11 +17,12 @@ name = "fjall"
 path = "src/lib.rs"
 
 [features]
-default = ["bloom", "single_writer_tx", "lz4"]
+default = ["bloom", "ssi_tx", "lz4"]
 lz4 = ["lsm-tree/lz4"]
 miniz = ["lsm-tree/miniz"]
 bloom = ["lsm-tree/bloom"]
 single_writer_tx = []
+ssi_tx = ["wmark"]
 __internal_whitebox = []
 
 [dependencies]
@@ -33,6 +34,7 @@ tempfile = "3.10.1"
 path-absolutize = "3.1.1"
 dashmap = "6.0.1"
 xxhash-rust = { version = "0.8.12", features = ["xxh3"] }
+wmark = { version = "0.1.4", default-features = false, features = ["std"], optional = true }
 
 [dev-dependencies]
 criterion = { version = "0.5.1", features = ["html_reports"] }

README.md

@@ -20,7 +20,7 @@ Fjall is an LSM-based embeddable key-value storage engine written in Rust. It fe
 - Automatic background maintenance
 - Partitions (a.k.a. column families) with cross-partition atomic semantics
 - Built-in compression (default = LZ4)
-- Single-writer, multi-reader transactions (optional)
+- Serializable transactions (optional)
 - Key-value separation for large blob use cases (optional)
 
 Each `Keyspace` is a single logical database and is split into `partitions` (a.k.a. column families) - you should probably only use a single keyspace for your application. Each partition is physically a single LSM-tree and its own logical collection; however, write operations across partitions are atomic as they are persisted in a single keyspace-level journal, which will be recovered on restart.
@@ -117,6 +117,13 @@ Allows opening a transactional Keyspace for single-writer (serialized) transacti
 
 *Enabled by default.*
 
+### ssi_tx
+
+Allows opening a transactional Keyspace for multi-writer, serializable transactions, allowing RYOW (read-your-own-write), fetch-and-update and other atomic operations.
+Conflict checking is done using opimistic concurrency control.
+
+*Disabled by default.*
+
 ## Stable disk format
 
 The disk format is stable as of 1.0.0.

src/config.rs

@@ -235,7 +235,7 @@ impl Config {
     /// # Errors
     ///
     /// Will return `Err` if an IO error occurs.
-    #[cfg(feature = "single_writer_tx")]
+    #[cfg(any(feature = "single_writer_tx", feature = "ssi_tx"))]
     pub fn open_transactional(self) -> crate::Result<crate::TxKeyspace> {
         crate::TxKeyspace::open(self)
     }

src/error.rs

@@ -2,6 +2,8 @@
 // This source code is licensed under both the Apache 2.0 and MIT License
 // (found in the LICENSE-* files in the repository)
 
+#[cfg(any(feature = "single_writer_tx", feature = "ssi_tx"))]
+use crate::tx::write::ssi::Error as SsiError;
 use crate::{journal::error::RecoveryError as JournalRecoveryError, version::Version};
 use lsm_tree::{DecodeError, EncodeError};
 
@@ -37,6 +39,10 @@ pub enum Error {
 
     /// Partition is deleted
     PartitionDeleted,
+
+    #[cfg(feature = "ssi_tx")]
+    /// SSI transaction-related error
+    Ssi(SsiError),
 }
 
 impl std::fmt::Display for Error {
@@ -69,6 +75,13 @@ impl From<lsm_tree::Error> for Error {
     }
 }
 
+#[cfg(feature = "ssi_tx")]
+impl From<SsiError> for Error {
+    fn from(value: SsiError) -> Self {
+        Self::Ssi(value)
+    }
+}
+
 impl std::error::Error for Error {}
 
 /// Result helper type

src/lib.rs

@@ -103,7 +103,7 @@ mod snapshot_nonce;
 mod snapshot_tracker;
 mod tracked_snapshot;
 
-#[cfg(feature = "single_writer_tx")]
+#[cfg(any(feature = "single_writer_tx", feature = "ssi_tx"))]
 mod tx;
 
 mod version;
@@ -127,7 +127,7 @@ pub use {
     version::Version,
 };
 
-#[cfg(feature = "single_writer_tx")]
+#[cfg(any(feature = "single_writer_tx", feature = "ssi_tx"))]
 pub use tx::{
     keyspace::{TransactionalKeyspace, TxKeyspace},
     partition::TransactionalPartitionHandle,
@@ -142,15 +142,15 @@ pub type WriteBatch = Batch;
 pub type Partition = PartitionHandle;
 
 /// Alias for [`TransactionalPartitionHandle`]
-#[cfg(feature = "single_writer_tx")]
+#[cfg(any(feature = "single_writer_tx", feature = "ssi_tx"))]
 pub type TxPartition = TransactionalPartitionHandle;
 
 /// Alias for [`TransactionalPartitionHandle`]
-#[cfg(feature = "single_writer_tx")]
+#[cfg(any(feature = "single_writer_tx", feature = "ssi_tx"))]
 pub type TxPartitionHandle = TransactionalPartitionHandle;
 
 /// Alias for [`TransactionalPartitionHandle`]
-#[cfg(feature = "single_writer_tx")]
+#[cfg(any(feature = "single_writer_tx", feature = "ssi_tx"))]
 pub type TransactionalPartition = TransactionalPartitionHandle;
 
 /// A snapshot moment

src/tx/conflict_manager.rs

@@ -0,0 +1,218 @@
+use crate::batch::PartitionKey;
+use core::ops::Bound;
+use lsm_tree::Slice;
+use std::{
+    collections::{BTreeMap, BTreeSet},
+    ops::RangeBounds,
+    sync::Mutex,
+};
+
+#[derive(Clone, Debug)]
+enum Read {
+    Single(Slice),
+    Range {
+        start: Bound<Slice>,
+        end: Bound<Slice>,
+    },
+    All,
+}
+
+#[derive(Default, Debug)]
+pub struct ConflictManager {
+    reads: Mutex<BTreeMap<PartitionKey, Vec<Read>>>,
+    conflict_keys: Mutex<BTreeMap<PartitionKey, BTreeSet<Slice>>>,
+}
+
+impl ConflictManager {
+    fn push_read(&self, partition: &PartitionKey, read: Read) {
+        let mut map = self.reads.lock().expect("poisoned reads lock");
+        if let Some(tbl) = map.get_mut(partition) {
+            tbl.push(read);
+        } else {
+            map.entry(partition.clone()).or_default().push(read);
+        }
+    }
+
+    pub fn mark_read(&self, partition: &PartitionKey, key: &Slice) {
+        self.push_read(partition, Read::Single(key.clone()));
+    }
+
+    pub fn mark_conflict(&self, partition: &PartitionKey, key: &Slice) {
+        let mut map = self
+            .conflict_keys
+            .lock()
+            .expect("poisoned conflict keys lock");
+        if let Some(tbl) = map.get_mut(partition) {
+            tbl.insert(key.clone());
+        } else {
+            map.entry(partition.clone())
+                .or_default()
+                .insert(key.clone());
+        }
+    }
+
+    pub fn mark_range(&self, partition: &PartitionKey, range: impl RangeBounds<Slice>) {
+        let start = match range.start_bound() {
+            Bound::Included(k) => Bound::Included(k.clone()),
+            Bound::Excluded(k) => Bound::Excluded(k.clone()),
+            Bound::Unbounded => Bound::Unbounded,
+        };
+
+        let end = match range.end_bound() {
+            Bound::Included(k) => Bound::Included(k.clone()),
+            Bound::Excluded(k) => Bound::Excluded(k.clone()),
+            Bound::Unbounded => Bound::Unbounded,
+        };
+
+        let read = if start == Bound::Unbounded && end == Bound::Unbounded {
+            Read::All
+        } else {
+            Read::Range { start, end }
+        };
+
+        self.push_read(partition, read);
+    }
+}
+
+pub struct ConflictChecker {
+    reads: BTreeMap<PartitionKey, Vec<Read>>,
+    conflict_keys: BTreeMap<PartitionKey, BTreeSet<Slice>>,
+}
+
+impl ConflictChecker {
+    #[allow(clippy::too_many_lines)]
+    pub fn has_conflict(&self, other: &Self) -> bool {
+        if self.reads.is_empty() {
+            return false;
+        }
+
+        for (partition, keys) in &self.reads {
+            if let Some(other_conflict_keys) = other.conflict_keys.get(partition) {
+                for ro in keys {
+                    match ro {
+                        Read::Single(k) => {
+                            if other_conflict_keys.contains(k) {
+                                return true;
+                            }
+                        }
+                        Read::Range { start, end } => match (start, end) {
+                            (Bound::Included(start), Bound::Included(end)) => {
+                                if other_conflict_keys
+                                    .range::<Slice, _>((
+                                        Bound::Included(start),
+                                        Bound::Included(end),
+                                    ))
+                                    .next()
+                                    .is_some()
+                                {
+                                    return true;
+                                }
+                            }
+                            (Bound::Included(start), Bound::Excluded(end)) => {
+                                if other_conflict_keys
+                                    .range::<Slice, _>((
+                                        Bound::Included(start),
+                                        Bound::Excluded(end),
+                                    ))
+                                    .next()
+                                    .is_some()
+                                {
+                                    return true;
+                                }
+                            }
+                            (Bound::Included(start), Bound::Unbounded) => {
+                                if other_conflict_keys
+                                    .range::<Slice, _>((Bound::Included(start), Bound::Unbounded))
+                                    .next()
+                                    .is_some()
+                                {
+                                    return true;
+                                }
+                            }
+                            (Bound::Excluded(start), Bound::Included(end)) => {
+                                if other_conflict_keys
+                                    .range::<Slice, _>((
+                                        Bound::Excluded(start),
+                                        Bound::Included(end),
+                                    ))
+                                    .next()
+                                    .is_some()
+                                {
+                                    return true;
+                                }
+                            }
+                            (Bound::Excluded(start), Bound::Excluded(end)) => {
+                                if other_conflict_keys
+                                    .range::<Slice, _>((
+                                        Bound::Excluded(start),
+                                        Bound::Excluded(end),
+                                    ))
+                                    .next()
+                                    .is_some()
+                                {
+                                    return true;
+                                }
+                            }
+                            (Bound::Excluded(start), Bound::Unbounded) => {
+                                if other_conflict_keys
+                                    .range::<Slice, _>((Bound::Excluded(start), Bound::Unbounded))
+                                    .next()
+                                    .is_some()
+                                {
+                                    return true;
+                                }
+                            }
+                            (Bound::Unbounded, Bound::Included(end)) => {
+                                let range = ..=end;
+                                for write in other_conflict_keys {
+                                    if range.contains(&write) {
+                                        return true;
+                                    }
+                                }
+                            }
+                            (Bound::Unbounded, Bound::Excluded(end)) => {
+                                let range = ..end;
+                                for write in other_conflict_keys {
+                                    if range.contains(&write) {
+                                        return true;
+                                    }
+                                }
+                            }
+                            (Bound::Unbounded, Bound::Unbounded) => unreachable!(),
+                        },
+                        Read::All => {
+                            if !other_conflict_keys.is_empty() {
+                                return true;
+                            }
+                        }
+                    }
+                }
+            }
+        }
+
+        false
+    }
+}
+
+impl From<ConflictManager> for ConflictChecker {
+    fn from(value: ConflictManager) -> Self {
+        Self {
+            reads: std::mem::take(&mut value.reads.lock().expect("reads lock poisoned")),
+            conflict_keys: std::mem::take(
+                &mut value
+                    .conflict_keys
+                    .lock()
+                    .expect("conflict_keys lock poisoned"),
+            ),
+        }
+    }
+}
+
+impl From<ConflictChecker> for ConflictManager {
+    fn from(mut value: ConflictChecker) -> Self {
+        Self {
+            reads: Mutex::new(std::mem::take(&mut value.reads)),
+            conflict_keys: Mutex::new(std::mem::take(&mut value.conflict_keys)),
+        }
+    }
+}