Added biggest key index

.vscode/settings.json

@@ -4,6 +4,7 @@
         "./Cargo.toml",
         "./Cargo.toml",
         "./Cargo.toml",
+        "./Cargo.toml",
         "./Cargo.toml"
     ]
 }

src/block/block.rs

@@ -50,7 +50,7 @@
 
 use tokio::{
     fs::File,
-    io::{self, AsyncSeekExt, AsyncWriteExt},
+    io::{self, AsyncWriteExt},
 };
 
 use err::StorageEngineError::*;
@@ -59,7 +59,7 @@ use crate::{
     consts::{SIZE_OF_U32, SIZE_OF_U64, SIZE_OF_U8},
     err::{self, StorageEngineError},
 };
-const BLOCK_SIZE: usize = 8 * 1024; // 4KB
+const BLOCK_SIZE: usize = 64 * 1024; // 4KB
 
 #[derive(Debug, Clone)]
 pub struct Block {
@@ -129,7 +129,7 @@ impl Block {
     }
 
     pub async fn write_to_file(&self, file: &mut File) -> Result<(), StorageEngineError> {
-        for  entry in &self.data {
+        for entry in &self.data {
             let entry_len = entry.key.len() + SIZE_OF_U32 + SIZE_OF_U32 + SIZE_OF_U64 + SIZE_OF_U8;
             let mut entry_vec = Vec::with_capacity(entry_len);
 

src/bloom_filter/bf.rs

@@ -2,6 +2,7 @@ use bit_vec::BitVec;
 use std::{
     collections::hash_map::DefaultHasher,
     hash::{Hash, Hasher},
+    path::{self, PathBuf},
     sync::{
         atomic::{AtomicU32, Ordering},
         Arc, Mutex,
@@ -67,8 +68,23 @@ impl BloomFilter {
         self.sstable_path = Some(sstable_path);
     }
 
+    pub fn filter_by_sstable_paths<'a>(
+        bloom_filters: &'a Vec<BloomFilter>,
+        paths: Vec<&'a PathBuf>,
+    ) -> Vec<&'a BloomFilter> {
+        let mut filtered_bfs = Vec::new();
+        paths.into_iter().for_each(|p| {
+            bloom_filters.iter().for_each(|b| {
+                if b.get_sstable_path().data_file_path.as_path() == p.as_path() {
+                    filtered_bfs.push(b)
+                }
+            })
+        });
+        filtered_bfs
+    }
+
     pub fn get_sstable_paths_that_contains_key<T: Hash>(
-        bloom_filters: &Vec<BloomFilter>,
+        bloom_filters: Vec<&BloomFilter>,
         key: &T,
     ) -> Option<Vec<SSTablePath>> {
         let mut sstables: Vec<SSTablePath> = Vec::new();

src/compaction/bucket_coordinator.rs

@@ -28,6 +28,7 @@ use StorageEngineError::*;
 pub trait IndexWithSizeInBytes {
     fn get_index(&self) -> Arc<SkipMap<Vec<u8>, (usize, u64, bool)>>; // usize for value offset, u64 to store entry creation date in milliseconds
     fn size(&self) -> usize;
+    fn find_biggest_key_from_table(&self) -> Result<Vec<u8>, StorageEngineError>;
 }
 
 impl Bucket {

src/compaction/compactor.rs

@@ -10,6 +10,7 @@ use crate::{
     bloom_filter::BloomFilter,
     consts::TOMB_STONE_TTL,
     err::StorageEngineError,
+    key_offseter::TableBiggestKeys,
     memtable::Entry,
     sstable::{SSTable, SSTablePath},
 };
@@ -53,6 +54,7 @@ impl Compactor {
         &mut self,
         buckets: &mut BucketMap,
         bloom_filters: &mut Vec<BloomFilter>,
+        biggest_key_index: &mut TableBiggestKeys,
     ) -> Result<bool, StorageEngineError> {
         let mut number_of_compactions = 0;
         // The compaction loop will keep running until there
@@ -92,26 +94,43 @@ impl Compactor {
                                     sst_file_path.data_file_path, sst_file_path.index_file_path
                                 );
                                 // Step 4: Map this bloom filter to its sstable file path
+                                let sstable_data_file_path = sst_file_path.get_data_file_path();
                                 m.bloom_filter.set_sstable_path(sst_file_path);
 
                                 // Step 5: Store the bloom filter in the bloom filters vector
                                 bloom_filters.push(m.bloom_filter);
+                                let biggest_key = m.sstable.find_biggest_key()?;
+                                if biggest_key.is_empty() {
+                                    return Err(BiggestKeyIndexError);
+                                }
+                                biggest_key_index.set(sstable_data_file_path, biggest_key);
                                 actual_number_of_sstables_written_to_disk += 1;
                             }
                             Err(err) => {
                                 // Step 6: Trigger recovery in case compaction failed at any point
 
                                 // Ensure that bloom filter is restored to the previous state by removing entries added so far in
                                 // the compaction process and also remove merged sstables written to disk so far to prevent unstable state
-                                while let Some(bf) = bloom_filters.pop() {
-                                    match fs::remove_dir_all(bf.get_sstable_path().dir.clone())
-                                        .await
-                                    {
-                                        Ok(()) => info!("Stale SSTable File successfully deleted."),
-                                        Err(e) => error!("Stale SSTable File not deleted. {}", e),
+                                while actual_number_of_sstables_written_to_disk > 0 {
+                                    if let Some(bf) = bloom_filters.pop() {
+                                        match fs::remove_dir_all(bf.get_sstable_path().dir.clone())
+                                            .await
+                                        {
+                                            Ok(()) => {
+                                                biggest_key_index.remove(
+                                                    bf.get_sstable_path().data_file_path.clone(),
+                                                );
+                                                info!("Stale SSTable File successfully deleted.")
+                                            }
+                                            Err(e) => {
+                                                error!("Stale SSTable File not deleted. {}", e)
+                                            }
+                                        }
                                     }
+
                                     actual_number_of_sstables_written_to_disk -= 1;
                                 }
+
                                 error!("merged SSTable was not written to disk  {}", err);
                                 return Err(CompactionFailed(err.to_string()));
                             }
@@ -130,6 +149,7 @@ impl Compactor {
                             buckets,
                             &sstables_files_to_remove,
                             bloom_filters,
+                            biggest_key_index,
                         );
                         match bloom_filter_updated_opt.await {
                             Some(is_bloom_filter_updated) => {
@@ -159,9 +179,15 @@ impl Compactor {
         buckets: &mut BucketMap,
         sstables_to_delete: &Vec<(Uuid, Vec<SSTablePath>)>,
         bloom_filters_with_both_old_and_new_sstables: &mut Vec<BloomFilter>,
+        biggest_key_index: &mut TableBiggestKeys,
     ) -> Option<bool> {
+        // Remove obsolete keys from biggest keys index
+        sstables_to_delete.iter().for_each(|(_, sstables)| {
+            sstables.iter().for_each(|s| {
+                biggest_key_index.remove(s.get_data_file_path());
+            })
+        });
         let all_sstables_deleted = buckets.delete_sstables(sstables_to_delete).await;
-
         // if all sstables were not deleted then don't remove the associated bloom filters
         // although this can lead to redundancy bloom filters are in-memory and its also less costly
         // since keys are represented in bits
@@ -217,7 +243,7 @@ impl Compactor {
                 )
                 .await
                 .map_err(|err| CompactionFailed(err.to_string()))?;
-            
+
                 match sst_opt {
                     Some(sst) => {
                         merged_sstable = self

src/consts/mod.rs

@@ -2,7 +2,7 @@ use crate::storage_engine::SizeUnit;
 
 pub const GC_THREAD_COUNT: u32 = 5;
 
-pub const DEFAULT_MEMTABLE_CAPACITY: usize = SizeUnit::Kilobytes.to_bytes(30);
+pub const DEFAULT_MEMTABLE_CAPACITY: usize = SizeUnit::Megabytes.to_bytes(1);
 
 pub const DEFAULT_FALSE_POSITIVE_RATE: f64 = 1e-200;
 

src/err/mod.rs

@@ -149,11 +149,17 @@ pub enum StorageEngineError {
     #[error("Index file write error")]
     IndexFileWriteError(#[source] io::Error),
 
-     /// Error while reading from index file
-     #[error("Index file read error")]
-     IndexFileReadError(#[source] io::Error),
+    /// Error while reading from index file
+    #[error("Index file read error")]
+    IndexFileReadError(#[source] io::Error),
 
     /// Error while flushing write to disk for index file
     #[error("Index file flush error")]
     IndexFileFlushError(#[source] io::Error),
+
+    #[error("Error finding biggest key in memtable (None was returned)")]
+    BiggestKeyIndexError,
+
+    #[error("All bloom filters return false for all sstables")]
+    KeyNotFoundByAnyBloomFilterError,
 }

src/key_offseter/key_offseter.rs

@@ -0,0 +1,33 @@
+use std::{cmp::Ordering, collections::HashMap, path::PathBuf};
+
+#[derive(Clone, Debug)]
+pub struct TableBiggestKeys {
+    pub sstables: HashMap<PathBuf, Vec<u8>>,
+}
+impl TableBiggestKeys {
+    pub fn new() -> Self {
+        Self {
+            sstables: HashMap::new(),
+        }
+    }
+
+    pub fn set(&mut self, sst_path: PathBuf, biggest_key: Vec<u8>) -> bool {
+        self.sstables.insert(sst_path, biggest_key).is_some()
+    }
+
+    pub fn remove(&mut self, sst_path: PathBuf) -> bool {
+        self.sstables.remove(&sst_path).is_some()
+    }
+
+    // Returns SSTables whose last key is greater than the supplied key parameter
+    pub fn filter_sstables_by_biggest_key(&self, key: &Vec<u8>) -> Vec<&PathBuf> {
+        self.sstables
+            .iter()
+            .filter(|(_, key_prefix)| {
+                key_prefix.as_slice().cmp(key) == Ordering::Greater
+                    || key_prefix.as_slice().cmp(key) == Ordering::Equal
+            })
+            .map(|(path, _)| return path)
+            .collect()
+    }
+}

src/key_offseter/mod.rs

@@ -0,0 +1,2 @@
+mod key_offseter;
+pub use key_offseter::TableBiggestKeys;

src/lib.rs

@@ -10,4 +10,5 @@ pub mod storage_engine;
 pub mod value_log;
 pub mod block;
 pub mod garbage_collector;
-pub mod sparse_index;
+pub mod sparse_index;
+pub mod key_offseter;

src/memtable/inmemory.rs

@@ -1,13 +1,15 @@
 use crate::bloom_filter::BloomFilter;
 use crate::compaction::IndexWithSizeInBytes;
-use crate::consts::{DEFAULT_FALSE_POSITIVE_RATE, DEFAULT_MEMTABLE_CAPACITY};
+use crate::consts::{
+    DEFAULT_FALSE_POSITIVE_RATE, DEFAULT_MEMTABLE_CAPACITY, SIZE_OF_U32, SIZE_OF_U64, SIZE_OF_U8,
+};
 use crate::err::StorageEngineError;
 //use crate::memtable::val_option::ValueOption;
 use crate::storage_engine::SizeUnit;
 use chrono::{DateTime, Utc};
 use crossbeam_skiplist::SkipMap;
-
 use std::cmp;
+use StorageEngineError::*;
 
 use std::{hash::Hash, sync::Arc};
 
@@ -36,6 +38,9 @@ impl IndexWithSizeInBytes for InMemoryTable<Vec<u8>> {
     fn size(&self) -> usize {
         self.size
     }
+    fn find_biggest_key_from_table(&self) -> Result<Vec<u8>, StorageEngineError> {
+        self.find_biggest_key()
+    }
 }
 
 impl Entry<Vec<u8>, usize> {
@@ -48,7 +53,6 @@ impl Entry<Vec<u8>, usize> {
         }
     }
 
-
     pub fn has_expired(&self, ttl: u64) -> bool {
         // Current time
         let current_time = Utc::now();
@@ -104,7 +108,7 @@ impl InMemoryTable<Vec<u8>> {
 
             // key length + value offset length + date created length
             // it takes 4 bytes to store a 32 bit integer ande 1 byte for tombstone checker
-            let entry_length_byte = entry.key.len() + 4 + 8 + 1;
+            let entry_length_byte = entry.key.len() + SIZE_OF_U32 + SIZE_OF_U64 + SIZE_OF_U8;
             self.size += entry_length_byte;
             return Ok(());
         }
@@ -115,7 +119,7 @@ impl InMemoryTable<Vec<u8>> {
         );
         // key length + value offset length + date created length
         // it takes 4 bytes to store a 32 bit integer since 8 bits makes 1 byte
-        let entry_length_byte = entry.key.len() + 4 + 8 + 1;
+        let entry_length_byte = entry.key.len() + SIZE_OF_U32 + SIZE_OF_U64 + SIZE_OF_U8;
         self.size += entry_length_byte;
         Ok(())
     }
@@ -131,7 +135,7 @@ impl InMemoryTable<Vec<u8>> {
 
     pub fn update(&mut self, entry: &Entry<Vec<u8>, usize>) -> Result<(), StorageEngineError> {
         if !self.bloom_filter.contains(&entry.key) {
-            return Err(StorageEngineError::KeyNotFoundInMemTable);
+            return Err(KeyNotFoundInMemTable);
         }
         // If the key already exist in the bloom filter then just insert into the entry alone
         self.index.insert(
@@ -147,16 +151,29 @@ impl InMemoryTable<Vec<u8>> {
 
     pub fn delete(&mut self, entry: &Entry<Vec<u8>, usize>) -> Result<(), StorageEngineError> {
         if !self.bloom_filter.contains(&entry.key) {
-            return Err(StorageEngineError::KeyNotFoundInMemTable);
+            return Err(KeyNotFoundInMemTable);
         }
         let created_at = Utc::now();
         // Insert thumb stone to indicate deletion
         self.index.insert(
             entry.key.to_vec(),
-            (entry.val_offset, created_at.timestamp_millis() as u64, entry.is_tombstone),
+            (
+                entry.val_offset,
+                created_at.timestamp_millis() as u64,
+                entry.is_tombstone,
+            ),
         );
         Ok(())
     }
+
+    // Find the biggest element in the skip list
+    pub fn find_biggest_key(&self) -> Result<Vec<u8>, StorageEngineError> {
+        let largest_entry = self.index.iter().next_back();
+        match largest_entry {
+            Some(e) => return Ok(e.key().to_vec()),
+            None => Err(BiggestKeyIndexError),
+        }
+    }
     pub fn false_positive_rate(&mut self) -> f64 {
         self.false_positive_rate
     }

src/sstable/sst.rs

@@ -44,6 +44,9 @@ impl IndexWithSizeInBytes for SSTable {
     fn size(&self) -> usize {
         self.size
     }
+    fn find_biggest_key_from_table(&self) -> Result<Vec<u8>, StorageEngineError>{
+        self.find_biggest_key()
+    }
 }
 
 #[derive(Debug, Clone)]
@@ -134,7 +137,14 @@ impl SSTable {
             created_at: created_at.timestamp_millis() as u64,
         }
     }
-
+    // Find the biggest element in the skip list
+    pub fn find_biggest_key(&self) -> Result<Vec<u8>, StorageEngineError> {
+        let largest_entry = self.index.iter().next_back();
+        match largest_entry {
+            Some(e) => return Ok(e.key().to_vec()),
+            None => Err(BiggestKeyIndexError),
+        }
+    }
     pub(crate) async fn write_to_file(&self) -> Result<(), StorageEngineError> {
         // Open the file in write mode with the append flag.
         let data_file_path = &self.data_file_path;