add w2d7 Task 2: Block Checksum

Author: Zhou-jw

Cargo.lock

@@ -19,6 +19,7 @@ crossbeam-channel = "0.5.11"
 serde_json = { version = "1.0" }
 serde = { version = "1.0", features = ["derive"] }
 farmhash = "1"
+crc32fast = "1.4.2"
 nom = "7.1.3"
 rustyline = "13.0.0"
 

mini-lsm-starter/Cargo.toml

@@ -9,7 +9,7 @@ use std::fs::File;
 use std::path::Path;
 use std::sync::Arc;
 
-use anyhow::{anyhow, Ok, Result};
+use anyhow::{anyhow, bail, Ok, Result};
 pub use builder::SsTableBuilder;
 use bytes::{Buf, BufMut};
 pub use iterator::SsTableIterator;
@@ -200,11 +200,17 @@ impl SsTable {
             .block_meta
             .get(block_idx + 1)
             .map_or(self.block_meta_offset, |x| x.offset);
-        let block_data = self
+        let block_len = offset_end - offset - 4;
+        let block_data_chksum = self
             .file
             .read(offset as u64, (offset_end - offset) as u64)?;
-
-        Ok(Arc::new(Block::decode(&block_data[..])))
+        let block_data = &block_data_chksum[..block_len];
+        let read_checksum = (&block_data_chksum[block_len..]).get_u32();
+        let checksum = crc32fast::hash(block_data);
+        if checksum != read_checksum {
+            bail!("mismatched block checksum !");
+        }
+        Ok(Arc::new(Block::decode(block_data)))
     }
 
     /// Read a block from disk, with block cache. (Day 4)

mini-lsm-starter/src/table.rs

@@ -74,13 +74,14 @@ impl SsTableBuilder {
         // std::mem::replace() is really helpful
         let block = std::mem::replace(&mut self.builder, BlockBuilder::new(self.block_size));
         let encoded_data = block.build().encode();
-        // let x = std::mem::take(&mut self.first_key).as_slice();
+        let check_sum = crc32fast::hash(&encoded_data);
         self.meta.push(BlockMeta {
             offset: self.data.len(),
             first_key: KeyVec::from_vec(std::mem::take(&mut self.first_key)).into_key_bytes(),
             last_key: KeyVec::from_vec(std::mem::take(&mut self.last_key)).into_key_bytes(),
         });
         self.data.extend(encoded_data); //TODO why return a Bytes and extend it to data instead of returning a Vec?
+        self.data.put_u32(check_sum);
     }
 
     /// Get the estimated size of the SSTable.

mini-lsm-starter/src/table/builder.rs

@@ -14,3 +14,4 @@ mod week2_day2;
 mod week2_day3;
 mod week2_day4;
 mod week2_day5;
+mod week2_day6;

mini-lsm-starter/src/tests.rs

@@ -0,0 +1,78 @@
+use tempfile::tempdir;
+
+use crate::{
+    compact::{
+        CompactionOptions, LeveledCompactionOptions, SimpleLeveledCompactionOptions,
+        TieredCompactionOptions,
+    },
+    lsm_storage::{LsmStorageOptions, MiniLsm},
+    tests::harness::dump_files_in_dir,
+};
+
+#[test]
+fn test_integration_leveled() {
+    test_integration(CompactionOptions::Leveled(LeveledCompactionOptions {
+        level_size_multiplier: 2,
+        level0_file_num_compaction_trigger: 2,
+        max_levels: 3,
+        base_level_size_mb: 1,
+    }))
+}
+
+#[test]
+fn test_integration_tiered() {
+    test_integration(CompactionOptions::Tiered(TieredCompactionOptions {
+        num_tiers: 3,
+        max_size_amplification_percent: 200,
+        size_ratio: 1,
+        min_merge_width: 3,
+        max_merge_width: None,
+    }))
+}
+
+#[test]
+fn test_integration_simple() {
+    test_integration(CompactionOptions::Simple(SimpleLeveledCompactionOptions {
+        size_ratio_percent: 200,
+        level0_file_num_compaction_trigger: 2,
+        max_levels: 3,
+    }));
+}
+
+fn test_integration(compaction_options: CompactionOptions) {
+    let dir = tempdir().unwrap();
+    let mut options = LsmStorageOptions::default_for_week2_test(compaction_options);
+    options.enable_wal = true;
+    let storage = MiniLsm::open(&dir, options.clone()).unwrap();
+    for i in 0..=20 {
+        storage.put(b"0", format!("v{}", i).as_bytes()).unwrap();
+        if i % 2 == 0 {
+            storage.put(b"1", format!("v{}", i).as_bytes()).unwrap();
+        } else {
+            storage.delete(b"1").unwrap();
+        }
+        if i % 2 == 1 {
+            storage.put(b"2", format!("v{}", i).as_bytes()).unwrap();
+        } else {
+            storage.delete(b"2").unwrap();
+        }
+        storage
+            .inner
+            .force_freeze_memtable(&storage.inner.state_lock.lock())
+            .unwrap();
+    }
+    storage.close().unwrap();
+    // ensure some SSTs are not flushed
+    assert!(
+        !storage.inner.state.read().memtable.is_empty()
+            || !storage.inner.state.read().imm_memtables.is_empty()
+    );
+    storage.dump_structure();
+    drop(storage);
+    dump_files_in_dir(&dir);
+
+    let storage = MiniLsm::open(&dir, options).unwrap();
+    assert_eq!(&storage.get(b"0").unwrap().unwrap()[..], b"v20".as_slice());
+    assert_eq!(&storage.get(b"1").unwrap().unwrap()[..], b"v20".as_slice());
+    assert_eq!(storage.get(b"2").unwrap(), None);
+}