chore: Bump ver: 0.4.0

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "map-api"
 description = "Raft state machine"
-version = "0.3.2"
+version = "0.4.0"
 authors = ["Databend Authors <opensource@datafuselabs.com>"]
 license = "Apache-2.0"
 edition = "2021"

src/mvcc/commit.rs

@@ -41,3 +41,232 @@ where
         changes: BTreeMap<S, Table<K, V>>,
     ) -> Result<(), io::Error>;
 }
+
+#[async_trait::async_trait]
+impl<S, K, V> Commit<S, K, V> for BTreeMap<S, Table<K, V>>
+where
+    Self: Send + Sync,
+    S: ViewNamespace,
+    K: ViewKey,
+    V: ViewValue,
+{
+    async fn commit(
+        &mut self,
+        last_seq: InternalSeq,
+        changes: BTreeMap<S, Table<K, V>>,
+    ) -> Result<(), io::Error> {
+        let _ = last_seq;
+        for (space, table) in changes {
+            // insert the table if absent
+            self.entry(space).or_default();
+
+            let dst = self.get_mut(&space).unwrap();
+
+            dst.apply(table);
+        }
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::BTreeMap;
+
+    use seq_marked::InternalSeq;
+    use seq_marked::SeqMarked;
+
+    use super::*;
+
+    #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+    enum TestSpace {
+        A,
+        B,
+    }
+
+    impl ViewNamespace for TestSpace {
+        fn increments_seq(&self) -> bool {
+            true
+        }
+    }
+
+    #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
+    struct TestKey(String);
+
+    #[derive(Debug, Clone, PartialEq, Eq)]
+    struct TestValue(String);
+
+    fn k(s: &str) -> TestKey {
+        TestKey(s.to_string())
+    }
+
+    fn v(s: &str) -> TestValue {
+        TestValue(s.to_string())
+    }
+
+    fn test_btreemap() -> BTreeMap<TestSpace, Table<TestKey, TestValue>> {
+        let mut table = Table::new();
+        table.insert(k("k1"), 1, v("v1")).unwrap();
+        table.insert(k("k2"), 2, v("v2")).unwrap();
+
+        let mut map = BTreeMap::new();
+        map.insert(TestSpace::A, table);
+        map
+    }
+
+    #[tokio::test]
+    async fn test_commit_empty() {
+        let mut map = test_btreemap();
+
+        map.commit(InternalSeq::new(0), BTreeMap::new())
+            .await
+            .unwrap();
+
+        assert_eq!(map.len(), 1);
+        let table = &map[&TestSpace::A];
+        assert_eq!(table.get(k("k1"), 100), SeqMarked::new_normal(1, &v("v1")));
+    }
+
+    #[tokio::test]
+    async fn test_commit_new_space() {
+        let mut map = test_btreemap();
+        let mut table = Table::new();
+        table.insert(k("k3"), 5, v("v3")).unwrap();
+
+        let mut changes = BTreeMap::new();
+        changes.insert(TestSpace::B, table);
+
+        map.commit(InternalSeq::new(10), changes).await.unwrap();
+
+        assert_eq!(map.len(), 2);
+        let table_b = &map[&TestSpace::B];
+        assert_eq!(
+            table_b.get(k("k3"), 100),
+            SeqMarked::new_normal(5, &v("v3"))
+        );
+    }
+
+    #[tokio::test]
+    async fn test_commit_existing_space() {
+        let mut map = test_btreemap();
+        let mut table = Table::new();
+        table.insert(k("k1"), 5, v("v1_new")).unwrap();
+        table.insert(k("k3"), 6, v("v3")).unwrap();
+        table.insert_tombstone(k("k2"), 7).unwrap();
+
+        let mut changes = BTreeMap::new();
+        changes.insert(TestSpace::A, table);
+
+        map.commit(InternalSeq::new(10), changes).await.unwrap();
+
+        let table_a = &map[&TestSpace::A];
+        assert_eq!(
+            table_a.get(k("k1"), 100),
+            SeqMarked::new_normal(5, &v("v1_new"))
+        );
+        assert_eq!(table_a.get(k("k2"), 100), SeqMarked::new_tombstone(7));
+        assert_eq!(
+            table_a.get(k("k3"), 100),
+            SeqMarked::new_normal(6, &v("v3"))
+        );
+    }
+
+    #[tokio::test]
+    async fn test_commit_multiple_spaces() {
+        let mut map = test_btreemap();
+
+        let mut table_a = Table::new();
+        table_a.insert(k("k3"), 10, v("v3")).unwrap();
+
+        let mut table_b = Table::new();
+        table_b.insert(k("k4"), 11, v("v4")).unwrap();
+
+        let mut changes = BTreeMap::new();
+        changes.insert(TestSpace::A, table_a);
+        changes.insert(TestSpace::B, table_b);
+
+        map.commit(InternalSeq::new(15), changes).await.unwrap();
+
+        assert_eq!(map.len(), 2);
+        let table_a = &map[&TestSpace::A];
+        let table_b = &map[&TestSpace::B];
+
+        assert_eq!(
+            table_a.get(k("k3"), 100),
+            SeqMarked::new_normal(10, &v("v3"))
+        );
+        assert_eq!(
+            table_b.get(k("k4"), 100),
+            SeqMarked::new_normal(11, &v("v4"))
+        );
+    }
+
+    #[tokio::test]
+    async fn test_commit_versioning() {
+        let mut map = test_btreemap();
+        let mut table = Table::new();
+        table.insert(k("k1"), 10, v("new")).unwrap();
+
+        let mut changes = BTreeMap::new();
+        changes.insert(TestSpace::A, table);
+
+        map.commit(InternalSeq::new(15), changes).await.unwrap();
+
+        let table_a = &map[&TestSpace::A];
+        assert_eq!(
+            table_a.get(k("k1"), 100),
+            SeqMarked::new_normal(10, &v("new"))
+        );
+        assert_eq!(table_a.get(k("k1"), 5), SeqMarked::new_normal(1, &v("v1")));
+    }
+
+    #[tokio::test]
+    #[should_panic(expected = "assertion failed: self.last_seq <= last_seq")]
+    async fn test_commit_sequence_conflict() {
+        let mut map = test_btreemap();
+
+        let mut table1 = Table::new();
+        table1.insert(k("k3"), 10, v("v3")).unwrap();
+        let mut changes1 = BTreeMap::new();
+        changes1.insert(TestSpace::A, table1);
+        map.commit(InternalSeq::new(15), changes1).await.unwrap();
+
+        let mut table2 = Table::new();
+        table2.insert(k("k4"), 5, v("v4")).unwrap();
+        let mut changes2 = BTreeMap::new();
+        changes2.insert(TestSpace::A, table2);
+        map.commit(InternalSeq::new(20), changes2).await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_commit_empty_tables() {
+        let mut map = BTreeMap::new();
+        let empty_table = Table::<TestKey, TestValue>::new();
+
+        let mut changes = BTreeMap::new();
+        changes.insert(TestSpace::A, empty_table);
+
+        map.commit(InternalSeq::new(0), changes).await.unwrap();
+
+        assert_eq!(map.len(), 1);
+        let table_a = &map[&TestSpace::A];
+        assert_eq!(table_a.last_seq, SeqMarked::zero());
+        assert!(table_a.get(k("any"), 100).is_not_found());
+    }
+
+    #[tokio::test]
+    async fn test_commit_only_tombstones() {
+        let mut map = test_btreemap();
+        let mut table = Table::new();
+        table.insert_tombstone(k("k1"), 5).unwrap();
+        table.insert_tombstone(k("k_new"), 6).unwrap();
+
+        let mut changes = BTreeMap::new();
+        changes.insert(TestSpace::A, table);
+
+        map.commit(InternalSeq::new(10), changes).await.unwrap();
+
+        let table_a = &map[&TestSpace::A];
+        assert_eq!(table_a.get(k("k1"), 100), SeqMarked::new_tombstone(5));
+        assert_eq!(table_a.get(k("k_new"), 100), SeqMarked::new_tombstone(6));
+    }
+}

src/mvcc/key.rs

@@ -16,7 +16,20 @@ use std::fmt;
 
 /// Trait for types that can be used as keys in MVCC operations.
 ///
-/// Keys must be orderable for range queries and serializable for storage.
+/// # Requirements
+///
+/// Keys must satisfy multiple constraints to work within the MVCC system:
+/// - **Ordering**: `Ord` enables range queries and consistent iteration
+/// - **Cloning**: `Clone` supports versioning and snapshot operations
+/// - **Threading**: `Send + Sync` allows concurrent access across threads
+/// - **Debugging**: `Debug` provides troubleshooting capabilities
+/// - **Async**: `Unpin` enables use in async contexts
+///
+/// # Automatic Implementation
+///
+/// This trait is automatically implemented for any type that meets the trait bounds.
+/// Common key types include `String`, `u64`, `Vec<u8>`, and custom structs that derive
+/// the required traits.
 pub trait ViewKey
 where Self: Clone + Ord + fmt::Debug + Send + Sync + Unpin + 'static
 {

src/mvcc/mod.rs

@@ -12,65 +12,86 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! Multi-Version Concurrency Control (MVCC) implementation.
+//! Multi-Version Concurrency Control (MVCC) for versioned key-value storage.
 //!
-//! Provides snapshot isolation for concurrent access to versioned key-value data.
-//! Operations are organized by namespaces and support both read-only and read-write access patterns.
+//! Provides **Read Committed** isolation with atomic transactions and snapshot consistency.
+//! Data is organized by namespaces with support for concurrent read-write operations.
 //!
-//! # Core Components
+//! # Architecture
 //!
-//! - **[`Table`]**: In-memory storage for versioned key-value pairs
-//! - **[`View`]**: Read-write transactional view with staged changes  
-//! - **Snapshot traits**: Read-only access at specific sequence points
-//! - **Scoped traits**: Namespace-bound operations for convenience
+//! - **[`Table`]**: In-memory versioned storage with sequence-based ordering
+//! - **[`View`]**: Read-write transaction with staged changes and commit capability
+//! - **[`Snapshot`]**: Read-only point-in-time view with fixed sequence boundary
+//! - **Scoped APIs**: Namespace-bound convenience methods ([`ScopedApi`], [`ScopedGet`], etc.)
+//!
+//! # Key Features
+//!
+//! - **Snapshot Isolation**: Each transaction sees consistent data from start time
+//! - **Atomic Commits**: All changes in a transaction commit together or fail together
+//! - **Namespace Partitioning**: Logical separation of data domains
+//! - **Streaming Range Queries**: Memory-efficient iteration over large datasets
 //!
 //! # Usage
 //!
 //! ```rust,ignore
-//! // Create storage and view
+//! // Basic transaction workflow
 //! let table = Table::new();
 //! let mut view = View::new(table);
 //!
-//! // Perform operations
-//! view.set(namespace, key, Some(value));
-//! let result = view.get(namespace, key).await?;
+//! // Stage changes
+//! view.set(namespace, "key1".to_string(), Some("value1".to_string()));
+//! view.set(namespace, "key2".to_string(), None); // deletion
+//!
+//! // Read with staged changes visible
+//! let current = view.get(namespace, "key1".to_string()).await?;
 //!
-//! // Commit changes
+//! // Atomic commit
 //! let updated_table = view.commit().await?;
 //! ```
 
 pub mod commit;
 pub mod key;
-pub mod namespace_view;
-pub mod scoped_snapshot_get;
-pub mod scoped_snapshot_into_range;
-pub mod scoped_snapshot_range;
-pub mod scoped_snapshot_range_iter;
-pub mod scoped_view;
-pub mod scoped_view_readonly;
-pub mod snapshot_get;
-pub mod snapshot_into_range;
-pub mod snapshot_range_iter;
+pub mod scoped_api;
+pub mod scoped_get;
+pub mod scoped_range;
+pub mod scoped_read;
+pub mod scoped_seq_bounded_get;
+pub mod scoped_seq_bounded_into_range;
+pub mod scoped_seq_bounded_range;
+pub mod scoped_seq_bounded_range_iter;
+pub mod scoped_set;
+pub mod seq_bounded_get;
+pub mod seq_bounded_into_range;
+pub mod seq_bounded_range;
+pub mod seq_bounded_range_iter;
+pub mod seq_bounded_read;
+pub mod snapshot;
+pub mod snapshot_seq;
 pub mod table;
 pub mod value;
 pub mod view;
 pub mod view_namespace;
-pub mod view_readonly;
 
 #[cfg(test)]
 mod namespace_view_no_seq_increase_test;
 
 pub use self::commit::Commit;
 pub use self::key::ViewKey;
-pub use self::scoped_snapshot_get::ScopedSnapshotGet;
-pub use self::scoped_snapshot_into_range::ScopedSnapshotIntoRange;
-pub use self::scoped_snapshot_range::ScopedSnapshotRange;
-pub use self::scoped_snapshot_range_iter::ScopedSnapshotRangeIter;
-pub use self::scoped_view::ScopedView;
-pub use self::scoped_view_readonly::ScopedViewReadonly;
+pub use self::scoped_api::ScopedApi;
+pub use self::scoped_get::ScopedGet;
+pub use self::scoped_range::ScopedRange;
+pub use self::scoped_read::ScopedRead;
+pub use self::scoped_seq_bounded_get::ScopedSeqBoundedGet;
+pub use self::scoped_seq_bounded_into_range::ScopedSnapshotIntoRange;
+pub use self::scoped_seq_bounded_range::ScopedSeqBoundedRange;
+pub use self::scoped_seq_bounded_range_iter::ScopedSeqBoundedRangeIter;
+pub use self::scoped_set::ScopedSet;
+pub use self::seq_bounded_get::SeqBoundedGet;
+pub use self::seq_bounded_range::SeqBoundedRange;
+pub use self::snapshot::Snapshot;
+pub use self::snapshot_seq::SnapshotSeq;
 pub use self::table::Table;
-pub use self::table::TableViewReadonly;
+pub use self::table::TablesSnapshot;
 pub use self::value::ViewValue;
 pub use self::view::View;
 pub use self::view_namespace::ViewNamespace;
-pub use self::view_readonly::ViewReadonly;