feat: add complete MVCC (Multi-Version Concurrency Control) implementation

Add comprehensive MVCC system with sequence-based versioning:

- Multi-namespace support with ViewNameSpace trait
- Read-write transactions via View with Read Committed isolation
- Read-only transactions via ViewReadonly with Snapshot isolation
- Staged changes with commit protocol
- Tombstone-based deletion system
- Stream-based range queries for memory efficiency
- Namespace-scoped views for ergonomic API usage

Key features:
- Type-safe generic design with ViewKey/ViewValue traits
- Table abstraction with insertion constraints and conflict detection
- Utility functions for key-value result sorting and merging
- Comprehensive isolation level documentation
- Test coverage for non-incrementing namespaces (secondary indices)

Author: drmingdrmer

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "map-api"
 description = "Raft state machine"
-version = "0.2.7"
+version = "0.3.0"
 authors = ["Databend Authors <[email protected]>"]
 license = "Apache-2.0"
 edition = "2021"
@@ -14,7 +14,7 @@ deepsize = { version = "0.2.0" }
 futures = "0.3.24"
 futures-util = "0.3.24"
 log = { version = "0.4.21", features = ["serde", "kv_unstable_std"] }
-seq-marked = { version = "0.3.1"}
+seq-marked = { version = "0.3.3"}
 serde = { version = "1.0.164", features = ["derive", "rc"] }
 stream-more = { version = "0.1.3" }
 thiserror = { version = "1" }

src/compact.rs

@@ -26,6 +26,15 @@ use crate::MapApiRO;
 use crate::MapKey;
 use crate::SeqMarked;
 
+/// Return the newer if it is a `not-found` record.
+pub fn compact_seq_marked_pair<T>(newer: SeqMarked<T>, older: SeqMarked<T>) -> SeqMarked<T> {
+    if newer.is_not_found() {
+        older
+    } else {
+        newer
+    }
+}
+
 /// Get a key from multi levels data.
 ///
 /// Returns the first non-tombstone entry.
@@ -116,13 +125,37 @@ mod tests {
 
     use futures_util::TryStreamExt;
 
+    use super::*;
     use crate::compact::compacted_get;
     use crate::compact::compacted_range;
     use crate::impls::immutable::Immutable;
     use crate::impls::level::Level;
     use crate::MapApi;
     use crate::SeqMarked;
 
+    #[test]
+    fn test_compact_seq_marked_pair() {
+        assert_eq!(
+            compact_seq_marked_pair(SeqMarked::new_normal(1, "a"), SeqMarked::new_normal(2, "b")),
+            SeqMarked::new_normal(1, "a")
+        );
+        assert_eq!(
+            compact_seq_marked_pair(SeqMarked::new_normal(1, "a"), SeqMarked::new_tombstone(2)),
+            SeqMarked::new_normal(1, "a")
+        );
+        assert_eq!(
+            compact_seq_marked_pair(SeqMarked::new_tombstone(1), SeqMarked::new_normal(2, "b")),
+            SeqMarked::new_tombstone(1)
+        );
+        assert_eq!(
+            compact_seq_marked_pair(
+                SeqMarked::<()>::new_tombstone(0),
+                SeqMarked::new_tombstone(2)
+            ),
+            SeqMarked::new_tombstone(2)
+        );
+    }
+
     #[tokio::test]
     async fn test_compacted_get() -> anyhow::Result<()> {
         let mut l0 = Level::default();
@@ -248,4 +281,173 @@ mod tests {
     fn b(x: impl ToString) -> Vec<u8> {
         x.to_string().as_bytes().to_vec()
     }
+
+    #[test]
+    fn test_compact_seq_marked_pair_edge_cases() {
+        // Not found newer should return older
+        assert_eq!(
+            compact_seq_marked_pair(
+                SeqMarked::<String>::new_not_found(),
+                SeqMarked::new_normal(5, "older".to_string())
+            ),
+            SeqMarked::new_normal(5, "older".to_string())
+        );
+
+        // Not found newer with tombstone older
+        assert_eq!(
+            compact_seq_marked_pair(
+                SeqMarked::<String>::new_not_found(),
+                SeqMarked::new_tombstone(3)
+            ),
+            SeqMarked::new_tombstone(3)
+        );
+
+        // Not found both
+        assert_eq!(
+            compact_seq_marked_pair(
+                SeqMarked::<String>::new_not_found(),
+                SeqMarked::<String>::new_not_found()
+            ),
+            SeqMarked::<String>::new_not_found()
+        );
+    }
+
+    #[tokio::test]
+    async fn test_compacted_get_empty_levels() -> anyhow::Result<()> {
+        // No levels at all
+        let got = compacted_get::<String, Level, Level>(&s("missing"), [], []).await?;
+        assert!(got.is_not_found());
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_compacted_get_key_not_found() -> anyhow::Result<()> {
+        let mut l0 = Level::default();
+        l0.set(s("a"), Some(b("a"))).await?;
+
+        let mut l1 = l0.new_level();
+        l1.set(s("b"), Some(b("b"))).await?;
+
+        // Key not in any level
+        let got = compacted_get::<String, _, Level>(&s("missing"), [&l0, &l1], []).await?;
+        assert!(got.is_not_found());
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_compacted_get_only_tombstones() -> anyhow::Result<()> {
+        let mut l0 = Level::default();
+        l0.set(s("a"), Some(b("a"))).await?;
+
+        let mut l1 = l0.new_level();
+        l1.set(s("a"), None).await?; // Tombstone
+
+        let mut l2 = l1.new_level();
+        l2.set(s("a"), None).await?; // Another tombstone
+
+        // Should find first tombstone
+        let got = compacted_get::<String, _, Level>(&s("a"), [&l2, &l1], []).await?;
+        assert_eq!(got, SeqMarked::new_tombstone(1));
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_compacted_range_empty_levels() -> anyhow::Result<()> {
+        // No levels at all
+        let got = compacted_range::<String, _, Level, &Level, Level>(s("").., None, [], []).await?;
+        let got = got.try_collect::<Vec<_>>().await?;
+        assert!(got.is_empty());
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_compacted_range_single_level() -> anyhow::Result<()> {
+        let mut l0 = Level::default();
+        l0.set(s("a"), Some(b("a"))).await?;
+        l0.set(s("b"), Some(b("b"))).await?;
+
+        let got = compacted_range::<_, _, Level, _, Level>(s("").., None, [&l0], []).await?;
+        let got = got.try_collect::<Vec<_>>().await?;
+        assert_eq!(got, vec![
+            (s("a"), SeqMarked::new_normal(1, b("a"))),
+            (s("b"), SeqMarked::new_normal(2, b("b"))),
+        ]);
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_compacted_range_with_persisted_only() -> anyhow::Result<()> {
+        let mut l0 = Level::default();
+        l0.set(s("a"), Some(b("a"))).await?;
+        l0.set(s("b"), Some(b("b"))).await?;
+
+        // Only persisted levels, no in-memory levels
+        let got =
+            compacted_range::<String, _, Level, &Level, &Level>(s("").., None, [], [&l0]).await?;
+        let got = got.try_collect::<Vec<_>>().await?;
+        assert_eq!(got, vec![
+            (s("a"), SeqMarked::new_normal(1, b("a"))),
+            (s("b"), SeqMarked::new_normal(2, b("b"))),
+        ]);
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_compacted_range_overlapping_keys() -> anyhow::Result<()> {
+        // Test that newer versions override older ones
+        let mut l0 = Level::default();
+        l0.set(s("a"), Some(b("old_a"))).await?;
+        l0.set(s("c"), Some(b("c"))).await?;
+
+        let mut l1 = l0.new_level();
+        l1.set(s("a"), Some(b("new_a"))).await?; // Override
+        l1.set(s("b"), Some(b("b"))).await?; // New key
+
+        let got = compacted_range::<_, _, Level, _, Level>(s("").., None, [&l1, &l0], []).await?;
+        let got = got.try_collect::<Vec<_>>().await?;
+        assert_eq!(got, vec![
+            (s("a"), SeqMarked::new_normal(3, b("new_a"))), /* Newer version (sequence continues from l0) */
+            (s("b"), SeqMarked::new_normal(4, b("b"))),
+            (s("c"), SeqMarked::new_normal(2, b("c"))),
+        ]);
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_compacted_range_bounded_range() -> anyhow::Result<()> {
+        let mut l0 = Level::default();
+        l0.set(s("a"), Some(b("a"))).await?;
+        l0.set(s("b"), Some(b("b"))).await?;
+        l0.set(s("c"), Some(b("c"))).await?;
+        l0.set(s("d"), Some(b("d"))).await?;
+
+        // Test bounded range
+        let got =
+            compacted_range::<_, _, Level, _, Level>(s("b")..=s("c"), None, [&l0], []).await?;
+        let got = got.try_collect::<Vec<_>>().await?;
+        assert_eq!(got, vec![
+            (s("b"), SeqMarked::new_normal(2, b("b"))),
+            (s("c"), SeqMarked::new_normal(3, b("c"))),
+        ]);
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_compacted_range_all_tombstones() -> anyhow::Result<()> {
+        let mut l0 = Level::default();
+        l0.set(s("a"), Some(b("a"))).await?;
+        l0.set(s("b"), Some(b("b"))).await?;
+
+        let mut l1 = l0.new_level();
+        l1.set(s("a"), None).await?; // Tombstone
+        l1.set(s("b"), None).await?; // Tombstone
+
+        let got = compacted_range::<_, _, Level, _, Level>(s("").., None, [&l1], []).await?;
+        let got = got.try_collect::<Vec<_>>().await?;
+        assert_eq!(got, vec![
+            (s("a"), SeqMarked::new_tombstone(2)), /* Both tombstones get same seq from the implementation */
+            (s("b"), SeqMarked::new_tombstone(2)),
+        ]);
+        Ok(())
+    }
 }

src/impls/immutable.rs

@@ -73,3 +73,130 @@ impl MapApiRO<String> for Immutable<ValueOf<String>> {
         Ok(strm)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use futures_util::TryStreamExt;
+
+    use super::*;
+    use crate::impls::level::Level;
+    use crate::MapApi;
+    use crate::SeqMarked;
+
+    #[test]
+    fn test_immutable_new() {
+        let level = Level::<Vec<u8>>::default();
+        let immutable = Immutable::new(Arc::new(level));
+
+        // Test that we can access the inner level
+        assert_eq!(Arc::strong_count(&immutable.level), 1);
+    }
+
+    #[test]
+    fn test_immutable_new_from_level() {
+        let level = Level::<Vec<u8>>::default();
+        let immutable = Immutable::new_from_level(level);
+
+        // Test that the immutable was created successfully
+        assert_eq!(Arc::strong_count(&immutable.level), 1);
+    }
+
+    #[test]
+    fn test_immutable_as_ref() {
+        let level = Level::<Vec<u8>>::default();
+        let immutable = Immutable::new_from_level(level);
+
+        // Test AsRef trait
+        let level_ref: &Level<Vec<u8>> = immutable.as_ref();
+        let _: &Level<Vec<u8>> = level_ref; // Type assertion
+    }
+
+    #[test]
+    fn test_immutable_deref() {
+        let level = Level::<Vec<u8>>::default();
+        let immutable = Immutable::new_from_level(level);
+
+        // Test Deref trait - we should be able to access Level methods directly
+        let _level_direct: &Level<Vec<u8>> = &immutable;
+    }
+
+    #[tokio::test]
+    async fn test_immutable_map_api_ro_get() {
+        let mut level = Level::default();
+        level
+            .set(
+                "test_key".to_string(),
+                Some("test_value".as_bytes().to_vec()),
+            )
+            .await
+            .unwrap();
+        let immutable = Immutable::new_from_level(level);
+
+        // Test MapApiRO implementation
+        let result = immutable.get(&"test_key".to_string()).await.unwrap();
+        assert!(result.is_normal());
+        assert_eq!(
+            result,
+            SeqMarked::new_normal(1, "test_value".as_bytes().to_vec())
+        );
+    }
+
+    #[tokio::test]
+    async fn test_immutable_map_api_ro_get_nonexistent() {
+        let level = Level::default();
+        let immutable = Immutable::new_from_level(level);
+
+        let result = immutable.get(&"nonexistent".to_string()).await.unwrap();
+        assert!(result.is_not_found());
+    }
+
+    #[tokio::test]
+    async fn test_immutable_map_api_ro_range() {
+        let mut level = Level::default();
+        level
+            .set("key1".to_string(), Some("value1".as_bytes().to_vec()))
+            .await
+            .unwrap();
+        level
+            .set("key2".to_string(), Some("value2".as_bytes().to_vec()))
+            .await
+            .unwrap();
+        let immutable = Immutable::new_from_level(level);
+
+        // Test range operation
+        let range_result = immutable
+            .range("key1".to_string().."key3".to_string())
+            .await
+            .unwrap();
+        let items: Vec<_> = range_result.try_collect().await.unwrap();
+
+        assert_eq!(items.len(), 2);
+        assert_eq!(items[0].0, "key1");
+        assert_eq!(items[1].0, "key2");
+    }
+
+    #[tokio::test]
+    async fn test_immutable_map_api_ro_empty_range() {
+        let level = Level::default();
+        let immutable = Immutable::new_from_level(level);
+
+        let range_result = immutable
+            .range("a".to_string().."z".to_string())
+            .await
+            .unwrap();
+        let items: Vec<_> = range_result.try_collect().await.unwrap();
+
+        assert!(items.is_empty());
+    }
+
+    #[test]
+    fn test_immutable_clone() {
+        let level = Level::<Vec<u8>>::default();
+        let immutable1 = Immutable::new_from_level(level);
+        let immutable2 = immutable1.clone();
+
+        // Test that both point to the same Arc
+        assert!(Arc::ptr_eq(&immutable1.level, &immutable2.level));
+        assert_eq!(Arc::strong_count(&immutable1.level), 2);
+    }
+}

src/lib.rs

@@ -70,6 +70,7 @@ pub mod map_api_ro;
 pub mod map_key;
 pub mod map_value;
 pub mod match_seq;
+pub mod mvcc;
 pub mod seq_value;
 pub mod util;