finished week 3 day 7

Q&A.md

@@ -318,5 +318,9 @@ OCC checks for conflicts at commit time. Pessimistic mechanisms would take a loc
 
 ##  Week 3 Day 6
 * If you have some experience with building a relational database, you may think about the following question: assume that we build a database based on Mini-LSM where we store each row in the relation table as a key-value pair (key: primary key, value: serialized row) and enable serializable verification, does the database system directly gain ANSI serializable isolation level capability? Why or why not?
+Keep forgetting the definition of ANSI serializable isolation but it's unlikely
+
 * The thing we implement here is actually write snapshot-isolation (see A critique of snapshot isolation) that guarantees serializable. Is there any cases where the execution is serializable, but will be rejected by the write snapshot-isolation validation?
+
+
 * There are databases that claim they have serializable snapshot isolation support by only tracking the keys accessed in gets and scans (instead of key range). Do they really prevent write skews caused by phantoms? (Okay... Actually, I'm talking about BadgerDB.)

mini-lsm-starter/src/compact.rs

@@ -20,7 +20,7 @@ use crate::iterators::merge_iterator::MergeIterator;
 use crate::iterators::two_merge_iterator::TwoMergeIterator;
 use crate::iterators::StorageIterator;
 use crate::key;
-use crate::lsm_storage::{LsmStorageInner, LsmStorageState};
+use crate::lsm_storage::{CompactionFilter, LsmStorageInner, LsmStorageState};
 use crate::manifest::ManifestRecord;
 use crate::table::{SsTable, SsTableBuilder, SsTableIterator};
 
@@ -115,6 +115,13 @@ pub enum CompactionOptions {
 }
 
 impl LsmStorageInner {
+    fn does_comp_filter_prefix_match(&self, key: &[u8]) -> bool {
+        let compaction_filters = self.compaction_filters.lock();
+        compaction_filters.iter().any(|filter| match filter {
+            CompactionFilter::Prefix(bytes) => key.starts_with(bytes),
+        })
+    }
+
     fn build_sstables_from_iterator(
         &self,
         mut iter: impl for<'a> StorageIterator<KeyType<'a> = key::Key<&'a [u8]>>,
@@ -147,6 +154,9 @@ impl LsmStorageInner {
                 if !hit_watermark {
                     //  first version of this key at or below watermark
                     hit_watermark = true;
+                    if self.does_comp_filter_prefix_match(iter.key().key_ref()) {
+                        continue;
+                    }
                     if compact_to_bottom_level && !iter.value().is_empty() {
                         sstable_builder.add(iter.key(), iter.value());
                     }

mini-lsm-starter/src/tests.rs

@@ -20,3 +20,4 @@ mod week3_day3;
 mod week3_day4;
 mod week3_day5;
 mod week3_day6;
+mod week3_day7;

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

@@ -0,0 +1,70 @@
+use bytes::Bytes;
+use tempfile::tempdir;
+
+use crate::{
+    compact::CompactionOptions,
+    lsm_storage::{CompactionFilter, LsmStorageOptions, MiniLsm, WriteBatchRecord},
+};
+
+use super::harness::{check_iter_result_by_key, construct_merge_iterator_over_storage};
+
+#[test]
+fn test_task3_mvcc_compaction() {
+    let dir = tempdir().unwrap();
+    let options = LsmStorageOptions::default_for_week2_test(CompactionOptions::NoCompaction);
+    let storage = MiniLsm::open(&dir, options.clone()).unwrap();
+    storage
+        .write_batch(&[
+            WriteBatchRecord::Put("table1_a", "1"),
+            WriteBatchRecord::Put("table1_b", "1"),
+            WriteBatchRecord::Put("table1_c", "1"),
+            WriteBatchRecord::Put("table2_a", "1"),
+            WriteBatchRecord::Put("table2_b", "1"),
+            WriteBatchRecord::Put("table2_c", "1"),
+        ])
+        .unwrap();
+    storage.force_flush().unwrap();
+    let snapshot0 = storage.new_txn().unwrap();
+    storage
+        .write_batch(&[
+            WriteBatchRecord::Put("table1_a", "2"),
+            WriteBatchRecord::Del("table1_b"),
+            WriteBatchRecord::Put("table1_c", "2"),
+            WriteBatchRecord::Put("table2_a", "2"),
+            WriteBatchRecord::Del("table2_b"),
+            WriteBatchRecord::Put("table2_c", "2"),
+        ])
+        .unwrap();
+    storage.force_flush().unwrap();
+    storage.add_compaction_filter(CompactionFilter::Prefix(Bytes::from("table2_")));
+    storage.force_full_compaction().unwrap();
+
+    let mut iter = construct_merge_iterator_over_storage(&storage.inner.state.read());
+    check_iter_result_by_key(
+        &mut iter,
+        vec![
+            (Bytes::from("table1_a"), Bytes::from("2")),
+            (Bytes::from("table1_a"), Bytes::from("1")),
+            (Bytes::from("table1_b"), Bytes::new()),
+            (Bytes::from("table1_b"), Bytes::from("1")),
+            (Bytes::from("table1_c"), Bytes::from("2")),
+            (Bytes::from("table1_c"), Bytes::from("1")),
+            (Bytes::from("table2_a"), Bytes::from("2")),
+            (Bytes::from("table2_b"), Bytes::new()),
+            (Bytes::from("table2_c"), Bytes::from("2")),
+        ],
+    );
+
+    drop(snapshot0);
+
+    storage.force_full_compaction().unwrap();
+
+    let mut iter = construct_merge_iterator_over_storage(&storage.inner.state.read());
+    check_iter_result_by_key(
+        &mut iter,
+        vec![
+            (Bytes::from("table1_a"), Bytes::from("2")),
+            (Bytes::from("table1_c"), Bytes::from("2")),
+        ],
+    );
+}