impl w3d5 Transaction and Optimistic Concurrency Control

Author: Zhou-jw

mini-lsm-starter/src/lsm_iterator.rs

@@ -67,7 +67,6 @@ impl LsmIterator {
             while self.inner.is_valid() && self.prev_key == self.inner.key().key_ref() {
                 self.inner_next()?;
             }
-
             if !self.inner.is_valid() {
                 break;
             }
@@ -82,12 +81,16 @@ impl LsmIterator {
             {
                 self.inner_next()?;
             }
-
             // after call of self.inner_next(), self.inner may be invalid, so we should check once call self.inner_next()
             if !self.inner.is_valid() {
                 break;
             }
 
+            // if seek to next different key, repeat the process
+            if self.prev_key != self.inner.key().key_ref() {
+                continue;
+            }
+
             if !self.inner.value().is_empty() {
                 break;
             }

mini-lsm-starter/src/mvcc/txn.rs

@@ -33,6 +33,9 @@ impl Transaction {
     pub fn get(&self, key: &[u8]) -> Result<Option<Bytes>> {
         // first probe the local_sotrage
         if let Some(entry) = self.local_storage.get(key) {
+            if entry.value().is_empty() {
+                return Ok(None);
+            }
             return Ok(Some(entry.value().clone()));
         }
         self.inner.get_with_ts(key, self.read_ts)
@@ -83,7 +86,8 @@ impl Transaction {
                 }
             })
             .collect::<Vec<_>>();
-        Ok(())
+
+        self.inner.write_batch(&batch)
     }
 }
 

mini-lsm-starter/src/tests.rs

@@ -19,3 +19,4 @@ mod week3_day1;
 mod week3_day2;
 mod week3_day3;
 mod week3_day4;
+mod week3_day5;

mini-lsm-starter/src/tests/week3_day5.rs

@@ -0,0 +1,75 @@
+use std::ops::Bound;
+
+use bytes::Bytes;
+use tempfile::tempdir;
+
+use crate::{
+    compact::CompactionOptions,
+    lsm_storage::{LsmStorageOptions, MiniLsm},
+    tests::harness::check_lsm_iter_result_by_key,
+};
+
+#[test]
+fn test_txn_integration() {
+    let dir = tempdir().unwrap();
+    let options = LsmStorageOptions::default_for_week2_test(CompactionOptions::NoCompaction);
+    let storage = MiniLsm::open(&dir, options.clone()).unwrap();
+    let txn1 = storage.new_txn().unwrap();
+    let txn2 = storage.new_txn().unwrap();
+    txn1.put(b"test1", b"233");
+    txn2.put(b"test2", b"233");
+    check_lsm_iter_result_by_key(
+        &mut txn1.scan(Bound::Unbounded, Bound::Unbounded).unwrap(),
+        vec![(Bytes::from("test1"), Bytes::from("233"))],
+    );
+    check_lsm_iter_result_by_key(
+        &mut txn2.scan(Bound::Unbounded, Bound::Unbounded).unwrap(),
+        vec![(Bytes::from("test2"), Bytes::from("233"))],
+    );
+    let txn3 = storage.new_txn().unwrap();
+    check_lsm_iter_result_by_key(
+        &mut txn3.scan(Bound::Unbounded, Bound::Unbounded).unwrap(),
+        vec![],
+    );
+    txn1.commit().unwrap();
+    txn2.commit().unwrap();
+    check_lsm_iter_result_by_key(
+        &mut txn3.scan(Bound::Unbounded, Bound::Unbounded).unwrap(),
+        vec![],
+    );
+    drop(txn3);
+    check_lsm_iter_result_by_key(
+        &mut storage.scan(Bound::Unbounded, Bound::Unbounded).unwrap(),
+        vec![
+            (Bytes::from("test1"), Bytes::from("233")),
+            (Bytes::from("test2"), Bytes::from("233")),
+        ],
+    );
+    let txn4 = storage.new_txn().unwrap();
+    assert_eq!(txn4.get(b"test1").unwrap(), Some(Bytes::from("233")));
+    assert_eq!(txn4.get(b"test2").unwrap(), Some(Bytes::from("233")));
+    check_lsm_iter_result_by_key(
+        &mut txn4.scan(Bound::Unbounded, Bound::Unbounded).unwrap(),
+        vec![
+            (Bytes::from("test1"), Bytes::from("233")),
+            (Bytes::from("test2"), Bytes::from("233")),
+        ],
+    );
+    txn4.put(b"test2", b"2333");
+    assert_eq!(txn4.get(b"test1").unwrap(), Some(Bytes::from("233")));
+    assert_eq!(txn4.get(b"test2").unwrap(), Some(Bytes::from("2333")));
+    check_lsm_iter_result_by_key(
+        &mut txn4.scan(Bound::Unbounded, Bound::Unbounded).unwrap(),
+        vec![
+            (Bytes::from("test1"), Bytes::from("233")),
+            (Bytes::from("test2"), Bytes::from("2333")),
+        ],
+    );
+    txn4.delete(b"test2");
+    assert_eq!(txn4.get(b"test1").unwrap(), Some(Bytes::from("233")));
+    assert_eq!(txn4.get(b"test2").unwrap(), None);
+    check_lsm_iter_result_by_key(
+        &mut txn4.scan(Bound::Unbounded, Bound::Unbounded).unwrap(),
+        vec![(Bytes::from("test1"), Bytes::from("233"))],
+    );
+}