From 912c32f762ee7585e20fde57c7c5dca1f9b0478d Mon Sep 17 00:00:00 2001 From: Jan Tuomi Date: Sat, 2 Nov 2024 16:58:01 +0200 Subject: Refactor in preparation to larger changes --- log_db/src/common.rs | 116 ++++++++++++- log_db/src/forward_log_reader.rs | 124 -------------- log_db/src/lib.rs | 170 ++++--------------- log_db/src/log_reader_forward.rs | 124 ++++++++++++++ log_db/src/log_reader_reverse.rs | 354 +++++++++++++++++++++++++++++++++++++++ log_db/src/memtable_primary.rs | 33 ++++ log_db/src/memtable_secondary.rs | 76 +++++++++ log_db/src/primary_memtable.rs | 96 ----------- log_db/src/reverse_log_reader.rs | 354 --------------------------------------- log_db/src/secondary_memtable.rs | 111 ------------ 10 files changed, 734 insertions(+), 824 deletions(-) delete mode 100644 log_db/src/forward_log_reader.rs create mode 100644 log_db/src/log_reader_forward.rs create mode 100644 log_db/src/log_reader_reverse.rs create mode 100644 log_db/src/memtable_primary.rs create mode 100644 log_db/src/memtable_secondary.rs delete mode 100644 log_db/src/primary_memtable.rs delete mode 100644 log_db/src/reverse_log_reader.rs delete mode 100644 log_db/src/secondary_memtable.rs (limited to 'log_db/src') diff --git a/log_db/src/common.rs b/log_db/src/common.rs index c2af766..0a1179e 100644 --- a/log_db/src/common.rs +++ b/log_db/src/common.rs @@ -1,7 +1,9 @@ +use std::cmp::Ordering; +use std::collections::HashSet; use std::fmt::Display; use std::fs::{metadata, File}; use std::io::{self}; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; // For Unix-like systems #[cfg(unix)] @@ -11,7 +13,7 @@ use std::os::unix::fs::MetadataExt; #[cfg(windows)] use std::os::windows::fs::MetadataExt; -pub const ACTIVE_LOG_FILENAME: &str = "db"; +pub const ACTIVE_SYMLINK_FILENAME: &str = "active"; pub const EXCL_LOCK_REQUEST_FILENAME: &str = "excl_lock_req"; pub const DEFAULT_READ_BUF_SIZE: usize = 1024 * 1024; // 1 MB pub const FIELD_SEPARATOR: u8 = b'\x1C'; @@ -63,11 +65,93 @@ pub enum SpecialSequence { LiteralEscape, } +/// LogKey is a packed struct that contains: +/// - a log segment number (16 bits) +/// - a log index within the segment (48 bits) +#[derive(Debug, Clone, Eq, PartialEq, Hash, Ord, PartialOrd)] +pub struct LogKey(u64); + +impl LogKey { + pub fn new(segment_num: u16, index: u64) -> Self { + assert!(index < (1 << 48), "Index must fit in 48 bits"); + LogKey((segment_num as u64) << 48 | index) + } + + pub fn segment_num(&self) -> u16 { + (self.0 >> 48) as u16 + } + + pub fn index(&self) -> u64 { + self.0 & 0x0000_FFFF_FFFF_FFFF + } +} + +/// LogKeySet is a non-empty set of LogKeys. #[derive(Debug, Clone, Eq, PartialEq)] -pub enum MemtableEvictPolicy { - LeastWritten, - LeastRead, - LeastReadOrWritten, +pub struct LogKeySet { + set: HashSet, +} + +impl PartialOrd for LogKeySet { + fn partial_cmp(&self, other: &Self) -> Option { + let self_max_elem = self.set.iter().max()?; + let other_max_elem = other.set.iter().max()?; + Some(self_max_elem.cmp(other_max_elem)) + } +} + +impl LogKeySet { + /// Create a new LogKeySet with an initial LogKey. + /// The initial LogKey is required since LogKeySet must be non-empty. + pub fn new_with_initial(key: &LogKey) -> Self { + let mut set = HashSet::new(); + set.insert(key.clone()); + LogKeySet { set } + } + + /// Insert a LogKey into the set. + pub fn insert(&mut self, key: LogKey) { + self.set.insert(key); + } + + /// Remove a LogKey from the set. Return Ok(()) if the key was found and removed. + /// Return io::Error::InvalidInput if trying to remove the last element. + /// Return io::Error::NotFound if the key was not found. + pub fn remove(&mut self, key: &LogKey) -> Result<(), io::Error> { + if self.set.len() == 1 { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "Cannot remove the last element from LogKeySet", + )); + } + let removed = self.set.remove(key); + + if !removed { + return Err(io::Error::new( + io::ErrorKind::NotFound, + "LogKey not found in LogKeySet", + )); + } + + assert!( + self.set.len() > 0, + "LogKeySet should not be empty after removal" + ); + + Ok(()) + } + + /// Get a reference to the set of LogKeys. + pub fn log_keys(&self) -> &HashSet { + &self.set + } +} + +impl Ord for LogKeySet { + fn cmp(&self, other: &Self) -> Ordering { + self.partial_cmp(other) + .expect("LogKeySet comparison failed, possibly due to empty set") + } } #[derive(Debug, Clone, Eq, PartialEq)] @@ -284,6 +368,14 @@ pub fn escape_bytes(buf: &[u8]) -> Vec { result } +/// A path to a log segment file along with its type +pub enum SegmentPath { + /// A symbolic link to the active log file + ActiveSymlink(String), + /// A compacted segment that is no longer being written to + Compacted(String), +} + pub fn is_file_same_as_path(file: &File, path: &PathBuf) -> io::Result { // Get the metadata for the open file handle let file_metadata = file.metadata()?; @@ -306,3 +398,15 @@ pub fn is_file_same_as_path(file: &File, path: &PathBuf) -> io::Result { && file_metadata.volume_serial_number() == path_metadata.volume_serial_number()) } } + +pub fn symlink(original: &Path, link: &Path) -> io::Result<()> { + #[cfg(unix)] + { + std::os::unix::fs::symlink(original, link) + } + + #[cfg(windows)] + { + std::os::windows::fs::symlink_file(original, link) + } +} diff --git a/log_db/src/forward_log_reader.rs b/log_db/src/forward_log_reader.rs deleted file mode 100644 index 03726d0..0000000 --- a/log_db/src/forward_log_reader.rs +++ /dev/null @@ -1,124 +0,0 @@ -use super::common::*; -use std::fs::{self}; -use std::io::{self, BufRead, Read}; - -pub struct ForwardLogReader<'a> { - reader: io::BufReader<&'a mut fs::File>, -} - -impl<'a> ForwardLogReader<'a> { - pub fn new(file: &mut fs::File) -> ForwardLogReader { - let reader = io::BufReader::new(file); - ForwardLogReader { reader } - } - - fn read_record(&mut self) -> Result, io::Error> { - // The buffer that stores the bytes read from the file. - let mut read_buf: Vec = Vec::new(); - // The buffer that stores all the bytes of the record read so far in reverse order. - let mut result_buf: Vec = Vec::new(); - - // Try reading a byte from the file. - // If we've reached the end of the file, return None. - let mut peek_buf = vec![0]; - match self.reader.read_exact(&mut peek_buf) { - Ok(_) => { - // Go back one byte - self.reader.seek_relative(-1)?; - } - Err(ref e) if e.kind() == io::ErrorKind::UnexpectedEof => { - return Ok(None); - } - Err(e) => { - return Err(e); - } - } - - loop { - read_buf.clear(); - self.reader.read_until(ESCAPE_CHARACTER, &mut read_buf)?; - self.reader.seek_relative(-1)?; - result_buf.extend(&read_buf[..read_buf.len() - 1]); - - // Otherwise, we must have encountered an escape character. - match self.read_special_sequence()? { - SpecialSequence::RecordSeparator => { - // The record is complete, so we can break out of the loop. - break; - } - SpecialSequence::LiteralFieldSeparator => { - // The field separator is escaped, so we need to add it to the result buffer. - result_buf.push(FIELD_SEPARATOR); - } - SpecialSequence::LiteralEscape => { - // The escape character is escaped, so we need to add it to the result buffer. - result_buf.push(ESCAPE_CHARACTER); - } - } - } - - let record = Record::deserialize(&result_buf); - Ok(Some(record)) - } - - fn read_special_sequence(&mut self) -> Result { - let mut special_buf: Vec = vec![0; SEQ_RECORD_SEP.len()]; - self.reader.read_exact(&mut special_buf)?; - - match validate_special(&special_buf.as_slice()) { - Some(special) => Ok(special), - None => Err(io::Error::new( - io::ErrorKind::InvalidData, - "Not a special sequence", - )), - } - } -} - -impl Iterator for ForwardLogReader<'_> { - type Item = Record; - - fn next(&mut self) -> Option { - match self.read_record() { - Ok(Some(record)) => Some(record), - Ok(None) => None, - Err(err) => panic!("Error reading record: {:?}", err), - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - use std::path::Path; - - #[test] - fn test_forward_log_reader_fixture_db1() { - let db_path = Path::new(TEST_RESOURCES_DIR).join("test_db1"); - let mut file = fs::OpenOptions::new() - .read(true) - .open(&db_path) - .expect("Failed to open file"); - let mut forward_log_reader = ForwardLogReader::new(&mut file); - - // There are two records in the log with "schema": Int, Null - - let first_record = forward_log_reader - .next() - .expect("Failed to read the first record"); - assert!(match first_record.values.as_slice() { - [RecordValue::Int(0x1D), RecordValue::Null] => true, - _ => false, - }); - - let last_record = forward_log_reader - .next() - .expect("Failed to read the last record"); - assert!(match last_record.values.as_slice() { - [RecordValue::Int(10), RecordValue::Null] => true, - _ => false, - }); - - assert!(forward_log_reader.next().is_none()); - } -} diff --git a/log_db/src/lib.rs b/log_db/src/lib.rs index bcdaefe..0f086a5 100644 --- a/log_db/src/lib.rs +++ b/log_db/src/lib.rs @@ -2,18 +2,18 @@ extern crate log; mod common; -mod forward_log_reader; -mod primary_memtable; -mod reverse_log_reader; -mod secondary_memtable; +mod log_reader_forward; +mod log_reader_reverse; +mod memtable_primary; +mod memtable_secondary; pub use common::*; -pub use forward_log_reader::ForwardLogReader; use fs2::lock_contended_error; use fs2::FileExt; -use primary_memtable::PrimaryMemtable; -pub use reverse_log_reader::ReverseLogReader; -use secondary_memtable::SecondaryMemtable; +pub use log_reader_forward::ForwardLogReader; +pub use log_reader_reverse::ReverseLogReader; +use memtable_primary::PrimaryMemtable; +use memtable_secondary::SecondaryMemtable; use std::collections::BTreeMap; use std::fmt::Debug; use std::fs::{self}; @@ -30,7 +30,6 @@ pub struct ConfigBuilder { fields: Option>, primary_key: Option, secondary_keys: Option>, - memtable_evict_policy: Option, write_durability: Option, } @@ -43,7 +42,6 @@ impl<'a, Field: Eq + Clone + Debug> ConfigBuilder { fields: None, primary_key: None, secondary_keys: None, - memtable_evict_policy: None, write_durability: None, } } @@ -91,17 +89,6 @@ impl<'a, Field: Eq + Clone + Debug> ConfigBuilder { self } - /// The eviction policy for the memtables. Determines which - /// record will be dropped from a memtable when it reaches - /// capacity. - pub fn memtable_evict_policy( - &mut self, - memtable_evict_policy: MemtableEvictPolicy, - ) -> &mut Self { - self.memtable_evict_policy = Some(memtable_evict_policy); - self - } - /// The write durability policy for the database. /// This determines how writes are persisted to disk. /// The default is WriteDurability::Flush. @@ -128,10 +115,6 @@ impl<'a, Field: Eq + Clone + Debug> ConfigBuilder { "Required config value \"primary_key\" is not set", ))?, secondary_keys: self.secondary_keys.clone().unwrap_or(Vec::new()), - memtable_evict_policy: self - .memtable_evict_policy - .clone() - .unwrap_or(MemtableEvictPolicy::LeastReadOrWritten), write_durability: self .write_durability .clone() @@ -150,7 +133,6 @@ struct Config { pub fields: Vec<(Field, RecordField)>, pub primary_key: Field, pub secondary_keys: Vec, - pub memtable_evict_policy: MemtableEvictPolicy, pub write_durability: WriteDurability, } @@ -160,7 +142,7 @@ pub struct DB { log_file: fs::File, primary_key_index: usize, primary_memtable: PrimaryMemtable, - secondary_memtables: Vec>, + secondary_memtables: Vec, } impl DB { @@ -176,7 +158,7 @@ impl DB { fs::create_dir_all(&config.data_dir)?; } - let log_path = Path::new(&config.data_dir).join(ACTIVE_LOG_FILENAME); + let log_path = Path::new(&config.data_dir).join(ACTIVE_SYMLINK_FILENAME); // Create the log file if it does not exist let log_file_file = fs::OpenOptions::new() @@ -227,14 +209,11 @@ impl DB { } } } - let primary_memtable = PrimaryMemtable::new( - config.memtable_capacity, - config.memtable_evict_policy.clone(), - ); + let primary_memtable = PrimaryMemtable::new(); let secondary_memtables = config .secondary_keys .iter() - .map(|key| SecondaryMemtable::new(&config.fields, key, primary_key_index)) + .map(|key| SecondaryMemtable::new()) .collect(); let mut db = DB:: { @@ -419,7 +398,7 @@ impl DB { } else { debug!("Locking and searching rotated log segment file {}...", n); let path = Path::new(&self.config.data_dir) - .join(ACTIVE_LOG_FILENAME) + .join(ACTIVE_SYMLINK_FILENAME) .with_extension(n.to_string()); let mut segm_file = fs::OpenOptions::new().read(true).open(&path)?; @@ -485,10 +464,7 @@ impl DB { ))?; // Try to find a memtable with the queried key - let found_memtable_index = self - .secondary_memtables - .iter_mut() - .position(|mt| &mt.field == field); + let found_memtable_index = self.get_secondary_memtable_index_by_field(field); if let Some(memtable_index) = found_memtable_index { debug!( @@ -563,6 +539,13 @@ impl DB { Ok(result) } + fn get_secondary_memtable_index_by_field(&self, field: &Field) -> Option { + self.config + .secondary_keys + .iter() + .position(|schema_field| schema_field == field) + } + /// Ensures that the `self.log_file` handle is still pointing to the correct file. /// If the file has been rotated, the handle will be closed and reopened. /// Returns `true` if the file has been rotated and the handle has been reopened. @@ -591,7 +574,7 @@ impl DB { .as_indexable() .expect("A non-indexable value was stored at key index"); - if self.primary_memtable.capacity == 0 { + if self.config.memtable_capacity == 0 { return; } @@ -600,35 +583,21 @@ impl DB { &key, &record, ); - if let Some(evicted) = self.primary_memtable.evict_if_necessary() { - self.secondary_memtables - .iter_mut() - .for_each(|secondary_memtable| { - secondary_memtable.remove(&evicted); - }); - } - self.primary_memtable.set(&key, record); - self.secondary_memtables - .iter_mut() - .for_each(|secondary_memtable| { - debug!( - "Updating memtable for index on {:?}", - &secondary_memtable.field - ); - for (index, (schema_field, _)) in self.config.fields.iter().enumerate() { - if schema_field == &secondary_memtable.field { - let primary_key = record.values[self.primary_key_index] - .as_indexable() - .expect("Primary key was not indexable"); - let key = record.values[index] - .as_indexable() - .expect("Secondary index key was not indexable"); - secondary_memtable.set(&key, &primary_key); - } - } - }); + for (field_index, value) in record.values.iter().enumerate() { + let field = &self.config.fields[field_index].0; + if let Some(smt_index) = self.get_secondary_memtable_index_by_field(field) { + debug!("Updating memtable for index on {:?}", field); + + let memtable = &mut self.secondary_memtables[smt_index]; + let key = value.as_indexable().expect("Primary key was not indexable"); + let primary_key = record.values[self.primary_key_index] + .as_indexable() + .expect("Primary key was not indexable"); + memtable.set(&key, &primary_key); + } + } } fn request_exclusive_lock(&mut self) -> Result<(), io::Error> { @@ -717,7 +686,7 @@ impl DB { /// You may call this function in a separate thread or process to avoid blocking the main thread. /// However, the database will be exclusively locked, so all writes will be blocked during the tasks. pub fn do_maintenance_tasks(&mut self) -> Result<(), io::Error> { - let active_log_path = Path::new(&self.config.data_dir).join(ACTIVE_LOG_FILENAME); + let active_log_path = Path::new(&self.config.data_dir).join(ACTIVE_SYMLINK_FILENAME); let active_log_md = fs::metadata(&active_log_path)?; if active_log_md.size() >= self.config.segment_size as u64 { @@ -788,7 +757,7 @@ impl DB { .expect("Failed to convert file name to string") .to_string(); - if name != ACTIVE_LOG_FILENAME { + if name != ACTIVE_SYMLINK_FILENAME { return None; } @@ -849,68 +818,3 @@ impl DB { Ok(()) } } - -#[cfg(test)] -mod tests { - use super::*; - use rand::distributions::Alphanumeric; - use rand::Rng; - use std::collections::HashSet; - use tempfile::tempdir; - - #[derive(Eq, PartialEq, Clone, Debug)] - enum Field { - Id, - Name, - Data, - } - - fn tmp_dir() -> String { - let dir = tempdir() - .expect("Failed to create temporary directory") - .path() - .to_str() - .expect("Failed to convert temporary directory path to string") - .to_string(); - fs::create_dir_all(&dir).expect("Failed to create temporary directory"); - dir - } - - #[test] - fn memtables_always_have_the_same_primary_keys() { - let data_dir = tmp_dir(); - - let mut db = DB::configure() - .data_dir(&data_dir) - .fields(vec![ - (Field::Id, RecordField::int()), - (Field::Name, RecordField::string()), - ]) - .primary_key(Field::Id) - .secondary_keys(vec![Field::Name]) - .initialize() - .expect("Failed to initialize DB instance"); - - let mut rng = rand::thread_rng(); - for _ in 0..100 { - let id = rng.gen_range(0..100); - let name = (0..5).map(|_| rng.sample(Alphanumeric) as char).collect(); - - let record = Record { - values: vec![RecordValue::Int(id), RecordValue::String(name)], - }; - db.upsert(&record).expect("Failed to upsert record"); - - let p_set: HashSet<&IndexableValue> = db.primary_memtable.records.keys().collect(); - let mut s_set: HashSet<&IndexableValue> = HashSet::new(); - - for table in db.secondary_memtables.iter() { - table.records.values().for_each(|r| { - s_set.extend(r); - }); - } - - assert_eq!(p_set, s_set); - } - } -} diff --git a/log_db/src/log_reader_forward.rs b/log_db/src/log_reader_forward.rs new file mode 100644 index 0000000..03726d0 --- /dev/null +++ b/log_db/src/log_reader_forward.rs @@ -0,0 +1,124 @@ +use super::common::*; +use std::fs::{self}; +use std::io::{self, BufRead, Read}; + +pub struct ForwardLogReader<'a> { + reader: io::BufReader<&'a mut fs::File>, +} + +impl<'a> ForwardLogReader<'a> { + pub fn new(file: &mut fs::File) -> ForwardLogReader { + let reader = io::BufReader::new(file); + ForwardLogReader { reader } + } + + fn read_record(&mut self) -> Result, io::Error> { + // The buffer that stores the bytes read from the file. + let mut read_buf: Vec = Vec::new(); + // The buffer that stores all the bytes of the record read so far in reverse order. + let mut result_buf: Vec = Vec::new(); + + // Try reading a byte from the file. + // If we've reached the end of the file, return None. + let mut peek_buf = vec![0]; + match self.reader.read_exact(&mut peek_buf) { + Ok(_) => { + // Go back one byte + self.reader.seek_relative(-1)?; + } + Err(ref e) if e.kind() == io::ErrorKind::UnexpectedEof => { + return Ok(None); + } + Err(e) => { + return Err(e); + } + } + + loop { + read_buf.clear(); + self.reader.read_until(ESCAPE_CHARACTER, &mut read_buf)?; + self.reader.seek_relative(-1)?; + result_buf.extend(&read_buf[..read_buf.len() - 1]); + + // Otherwise, we must have encountered an escape character. + match self.read_special_sequence()? { + SpecialSequence::RecordSeparator => { + // The record is complete, so we can break out of the loop. + break; + } + SpecialSequence::LiteralFieldSeparator => { + // The field separator is escaped, so we need to add it to the result buffer. + result_buf.push(FIELD_SEPARATOR); + } + SpecialSequence::LiteralEscape => { + // The escape character is escaped, so we need to add it to the result buffer. + result_buf.push(ESCAPE_CHARACTER); + } + } + } + + let record = Record::deserialize(&result_buf); + Ok(Some(record)) + } + + fn read_special_sequence(&mut self) -> Result { + let mut special_buf: Vec = vec![0; SEQ_RECORD_SEP.len()]; + self.reader.read_exact(&mut special_buf)?; + + match validate_special(&special_buf.as_slice()) { + Some(special) => Ok(special), + None => Err(io::Error::new( + io::ErrorKind::InvalidData, + "Not a special sequence", + )), + } + } +} + +impl Iterator for ForwardLogReader<'_> { + type Item = Record; + + fn next(&mut self) -> Option { + match self.read_record() { + Ok(Some(record)) => Some(record), + Ok(None) => None, + Err(err) => panic!("Error reading record: {:?}", err), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::path::Path; + + #[test] + fn test_forward_log_reader_fixture_db1() { + let db_path = Path::new(TEST_RESOURCES_DIR).join("test_db1"); + let mut file = fs::OpenOptions::new() + .read(true) + .open(&db_path) + .expect("Failed to open file"); + let mut forward_log_reader = ForwardLogReader::new(&mut file); + + // There are two records in the log with "schema": Int, Null + + let first_record = forward_log_reader + .next() + .expect("Failed to read the first record"); + assert!(match first_record.values.as_slice() { + [RecordValue::Int(0x1D), RecordValue::Null] => true, + _ => false, + }); + + let last_record = forward_log_reader + .next() + .expect("Failed to read the last record"); + assert!(match last_record.values.as_slice() { + [RecordValue::Int(10), RecordValue::Null] => true, + _ => false, + }); + + assert!(forward_log_reader.next().is_none()); + } +} diff --git a/log_db/src/log_reader_reverse.rs b/log_db/src/log_reader_reverse.rs new file mode 100644 index 0000000..f042406 --- /dev/null +++ b/log_db/src/log_reader_reverse.rs @@ -0,0 +1,354 @@ +use super::common::*; +use std::fs::{self}; +use std::io::{self, Read, Seek, SeekFrom}; + +pub struct ReverseLogReader<'a> { + /// The file to read from end to beginning. + file: &'a mut fs::File, + /// The internal buffer used to read from the file. + /// It is populated with the last INTERNAL_BUF_SIZE bytes read from the file + /// and is used to read records in reverse order + internal_buf: Vec, + /// The current position in the internal buffer. It is decremented as bytes are read + /// from the buffer. When a read is requested and the internal position is 0, the buffer + /// is populated with the next (= closer to the start of the file) INTERNAL_BUF_SIZE bytes from the file. + /// Note: This is the index of the next byte to be read from the internal buffer + 1 + internal_pos: usize, + /// A flag indicating whether the record separator at the cursor position has been consumed. + /// Useful to avoid consuming the separator once when reading until an escape character, and + /// a second time when reading a new record and validating it ends in a separator. + consumed_record_sep: bool, +} + +// This value is based on the reverse_read_file_with_various_buffer_sizes benchmark. +// Greater values yield little to no performance improvement. +const DEFAULT_INTERNAL_BUF_SIZE: usize = 32768; +impl<'a> ReverseLogReader<'a> { + pub fn new(file: &mut fs::File) -> Result { + file.seek(SeekFrom::End(0))?; + Ok(ReverseLogReader { + file, + internal_buf: vec![0; DEFAULT_INTERNAL_BUF_SIZE], + internal_pos: 0, + consumed_record_sep: false, + }) + } + + pub fn new_with_size( + file: &mut fs::File, + internal_buf_size: usize, + ) -> Result { + file.seek(SeekFrom::End(0))?; + Ok(ReverseLogReader { + file, + internal_buf: vec![0; internal_buf_size], + internal_pos: 0, + consumed_record_sep: false, + }) + } + + pub fn read_record(&mut self) -> Result, io::Error> { + if self.file.stream_position()? == 0 && self.internal_pos == 0 { + return Ok(None); + } + + if !self.consumed_record_sep { + // Check that record ends with a record separator + match self.read_special_sequence()? { + SpecialSequence::RecordSeparator => {} + _ => { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "Record does not end in record separator", + )); + } + } + } + self.consumed_record_sep = false; + + let mut result_buf: Vec = vec![]; + let mut read_buf = Vec::with_capacity(self.internal_buf.len()); + loop { + read_buf.clear(); + let read = self.read_until(ESCAPE_CHARACTER, &mut read_buf)?; + + result_buf.extend(&read_buf[..read]); + + if self.file.stream_position()? == 0 && self.internal_pos == 0 { + // We read until the start of the file, we are done + break; + } + + match self.read_special_sequence()? { + SpecialSequence::LiteralEscape => { + result_buf.push(ESCAPE_CHARACTER); + } + SpecialSequence::LiteralFieldSeparator => { + result_buf.push(FIELD_SEPARATOR); + } + SpecialSequence::RecordSeparator => { + self.consumed_record_sep = true; + break; + } + } + } + + result_buf.reverse(); + Ok(Some(Record::deserialize(&result_buf))) + } + + /// Read exactly `buf.len()` bytes from the file, return an error if the file is exhausted. + /// The bytes are returned in start -> end order. + /// If an error is returned, the contents of `buf` are in an undefined state. + fn read_exact(&mut self, buf: &mut [u8]) -> Result { + let mut read = 0; + while read < buf.len() { + if self.internal_pos == 0 { + let populated_n = self.populate_internal_buf()?; + if populated_n == 0 { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "Unexpected end of file", + )); + } + } + + let end = buf.len() - read; + let n = std::cmp::min(self.internal_pos, end); + buf[end - n..end] + .copy_from_slice(&self.internal_buf[self.internal_pos - n..self.internal_pos]); + read += n; + self.internal_pos -= n; + } + + Ok(read) + } + + fn populate_internal_buf(&mut self) -> Result { + let current_seek_pos = self.file.stream_position()? as usize; + + // Seek back by the size of the internal buffer or to the beginning of the file + let seek_length = if current_seek_pos > self.internal_buf.len() { + self.internal_pos = self.internal_buf.len(); + self.internal_buf.len() + } else { + self.internal_buf = vec![0; current_seek_pos as usize]; + self.internal_pos = current_seek_pos as usize; + current_seek_pos + }; + + self.file.seek_relative(-(seek_length as i64))?; + self.file.read_exact(&mut self.internal_buf)?; + self.file.seek_relative(-(seek_length as i64))?; + + Ok(seek_length as usize) + } + + /// Iterate over the internal buffer with internal_pos as the index. + /// If the byte is found or the file has been exhausted, we return the number of bytes read. + /// The `buf` parameter is used to store the bytes read from the internal buffer, excluding the found byte, + /// in reverse order. + fn read_until(&mut self, byte: u8, buf: &mut Vec) -> Result { + // TODO: optimize this + let mut read = 0; + loop { + // If we reach internal_pos == 0, we need to populate the internal buffer. + if self.internal_pos == 0 { + let populated_n = self.populate_internal_buf()?; + if populated_n == 0 { + return Ok(read); + } + } + + while self.internal_pos > 0 { + let index = self.internal_pos - 1; + if self.internal_buf[index] == byte { + return Ok(read); + } + buf.push(self.internal_buf[index]); + read += 1; + self.internal_pos -= 1; + } + } + } + + fn read_special_sequence(&mut self) -> Result { + let mut special_buf: Vec = vec![0; SEQ_RECORD_SEP.len()]; + self.read_exact(&mut special_buf)?; + + match validate_special(&special_buf.as_slice()) { + Some(special) => Ok(special), + None => { + let pos = self.file.stream_position().unwrap() + self.internal_pos as u64; + + Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "Not a special sequence: {:?} at pos: {:x}", + special_buf, pos, + ), + )) + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Write; + use std::path::Path; + + #[test] + fn test_read_until_found() { + let mut file = tempfile::tempfile().unwrap(); + file.write_all(b"hello,world").unwrap(); + let mut reader = ReverseLogReader::new(&mut file).unwrap(); + let mut buf = vec![]; + assert_eq!(reader.read_until(b',', &mut buf).unwrap(), 5); + assert_eq!(buf, b"dlrow"); + } + + #[test] + fn test_read_until_not_found() { + let mut file = tempfile::tempfile().unwrap(); + file.write_all(b"hello,world").unwrap(); + let mut reader = ReverseLogReader::new(&mut file).unwrap(); + let mut buf = vec![]; + let read = reader.read_until(b'!', &mut buf).unwrap(); + assert_eq!(buf, b"dlrow,olleh"); + assert_eq!(read, 11); + } + + #[test] + fn test_read_special_sequence() { + let mut file = tempfile::tempfile().unwrap(); + let mut buf = vec![]; + buf.extend(SEQ_RECORD_SEP); + buf.extend(SEQ_LIT_ESCAPE); + buf.extend(SEQ_LIT_FIELD_SEP); + // Note: written in start -> end order, read end -> start + file.write_all(&buf).unwrap(); + + let mut reader = ReverseLogReader::new(&mut file).unwrap(); + assert_eq!( + reader.read_special_sequence().unwrap(), + SpecialSequence::LiteralFieldSeparator + ); + assert_eq!( + reader.read_special_sequence().unwrap(), + SpecialSequence::LiteralEscape + ); + assert_eq!( + reader.read_special_sequence().unwrap(), + SpecialSequence::RecordSeparator + ); + } + + #[test] + fn test_populate_internal_buf() { + let mut file = tempfile::tempfile().unwrap(); + file.write_all(b"hello,world").unwrap(); + let mut reader = ReverseLogReader::new_with_size(&mut file, 3).unwrap(); + + reader.populate_internal_buf().unwrap(); + assert_eq!( + String::from_utf8(reader.internal_buf.clone()).unwrap(), + "rld".to_string() + ); + + reader.populate_internal_buf().unwrap(); + assert_eq!( + String::from_utf8(reader.internal_buf.clone()).unwrap(), + ",wo".to_string() + ); + + reader.populate_internal_buf().unwrap(); + assert_eq!( + String::from_utf8(reader.internal_buf.clone()).unwrap(), + "llo".to_string() + ); + + reader.populate_internal_buf().unwrap(); + assert_eq!( + String::from_utf8(reader.internal_buf.clone()).unwrap(), + "he".to_string() + ); + } + + #[test] + fn test_reverse_log_reader_fixture_db1() { + let db_path = Path::new(TEST_RESOURCES_DIR).join("test_db1"); + let mut file = fs::OpenOptions::new() + .read(true) + .open(&db_path) + .expect("Failed to open file"); + let mut reverse_log_reader = ReverseLogReader::new(&mut file).unwrap(); + + // There are two records in the log with "schema": Int, Null + + let last_record = reverse_log_reader + .next() + .expect("Failed to read the last record"); + assert!(match last_record.values.as_slice() { + [RecordValue::Int(10), RecordValue::Null] => true, + _ => false, + }); + + let first_record = reverse_log_reader + .next() + .expect("Failed to read the first record"); + assert!(match first_record.values.as_slice() { + // Note: the int value is equal to the escape byte + [RecordValue::Int(0x1D), RecordValue::Null] => true, + _ => false, + }); + + assert!(reverse_log_reader.next().is_none()); + } + + #[test] + fn test_read_exact() { + let mut file = tempfile::tempfile().unwrap(); + file.write_all(b"hello,world").unwrap(); + let mut reader = ReverseLogReader::new(&mut file).unwrap(); + let mut buf = vec![0; 3]; + + let read = reader.read_exact(&mut buf).unwrap(); + assert_eq!(buf, b"rld"); + assert_eq!(read, 3); + + let read = reader.read_exact(&mut buf).unwrap(); + assert_eq!(buf, b",wo"); + assert_eq!(read, 3); + + let read = reader.read_exact(&mut buf).unwrap(); + assert_eq!(buf, b"llo"); + assert_eq!(read, 3); + + assert!(reader.read_exact(&mut buf).unwrap_err().kind() == io::ErrorKind::UnexpectedEof); + } + + #[test] + fn test_read_exact_insufficient_bytes() { + let mut file = tempfile::tempfile().unwrap(); + file.write_all(b"hello").unwrap(); + let mut reader = ReverseLogReader::new(&mut file).unwrap(); + let mut buf = vec![0; 10]; + assert!(reader.read_exact(&mut buf).unwrap_err().kind() == io::ErrorKind::UnexpectedEof); + } +} + +impl Iterator for ReverseLogReader<'_> { + type Item = Record; + + fn next(&mut self) -> Option { + match self.read_record() { + Ok(Some(record)) => Some(record), + Ok(None) => None, + Err(err) => { + panic!("Error reading record: {:?}", err,) + } + } + } +} diff --git a/log_db/src/memtable_primary.rs b/log_db/src/memtable_primary.rs new file mode 100644 index 0000000..9d7e1e8 --- /dev/null +++ b/log_db/src/memtable_primary.rs @@ -0,0 +1,33 @@ +use super::common::*; +use std::collections::BTreeMap; + +pub struct PrimaryMemtable { + /// Map of records indexed by key. Used as a shared heap of records + /// for all secondary memtables also. Secondary memtables store an + /// IndexableValue as their record value, which is used to get + /// the actual record from the primary memtable `records` map. + /// + /// Note: it must be invariant that all memtables (primary and secondary) + /// contain the same keys. + records: BTreeMap, +} + +impl PrimaryMemtable { + pub fn new() -> PrimaryMemtable { + PrimaryMemtable { + records: BTreeMap::new(), + } + } + + pub fn set(&mut self, key: &IndexableValue, value: &Record) { + self.records.insert(key.clone(), value.clone()); + } + + pub fn get(&mut self, key: &IndexableValue) -> Option<&Record> { + self.records.get(key) + } + + pub fn get_without_update(&self, key: &IndexableValue) -> Option<&Record> { + self.records.get(key) + } +} diff --git a/log_db/src/memtable_secondary.rs b/log_db/src/memtable_secondary.rs new file mode 100644 index 0000000..b4ebb8f --- /dev/null +++ b/log_db/src/memtable_secondary.rs @@ -0,0 +1,76 @@ +use super::*; +use std::collections::BTreeMap; +use std::collections::HashSet; +use std::fmt::Debug; + +pub struct SecondaryMemtable { + /// Map of records indexed by key. The value is the set of primary key values of records + /// that have the secondary key value. The actual `Record` objects are stored in the + /// primary memtable, which acts as the shared heap. + records: BTreeMap>, +} + +impl SecondaryMemtable { + pub fn new() -> SecondaryMemtable { + SecondaryMemtable { + records: BTreeMap::new(), + } + } + + pub fn set(&mut self, key: &IndexableValue, value: &IndexableValue) { + debug!( + "Inserting/updating record in secondary memtable with key {:?} = {:?}", + &key, &value, + ); + + match self.records.get_mut(key) { + Some(existing) => { + debug!( + "Existing entry found with {} records in the set", + &existing.len() + ); + existing.insert(value.clone()); + } + None => { + debug!("No existing entry found, creating one."); + let mut set = HashSet::with_capacity(1); + set.insert(value.clone()); + self.records.insert(key.clone(), set); + } + } + } + + pub fn set_all(&mut self, key: &IndexableValue, values: &[IndexableValue]) { + debug!( + "Replacing set of records in secondary memtable with key {:?} ({} values)", + &key, + &values.len(), + ); + + let mut set = HashSet::with_capacity(values.len()); + values.iter().for_each(|value| { + set.insert(value.clone()); + }); + + self.records.insert(key.clone(), set); + } + + pub fn find_all( + &mut self, + primary_memtable: &PrimaryMemtable, + key: &IndexableValue, + ) -> Vec { + match self.records.get(key) { + None => vec![], + Some(set) => set + .iter() + .map(|key| { + primary_memtable + .get_without_update(key) + .expect("Record not found") + .clone() + }) + .collect(), + } + } +} diff --git a/log_db/src/primary_memtable.rs b/log_db/src/primary_memtable.rs deleted file mode 100644 index 180ae72..0000000 --- a/log_db/src/primary_memtable.rs +++ /dev/null @@ -1,96 +0,0 @@ -use super::common::*; -use priority_queue::PriorityQueue; -use std::collections::BTreeMap; - -pub struct PrimaryMemtable { - /// Maximum number of records that can be stored in the memtable - /// before evicting the oldest records. The oldest record is - /// determined by the `evict_policy`. - pub capacity: usize, - /// Map of records indexed by key. Used as a shared heap of records - /// for all secondary memtables also. Secondary memtables store an - /// IndexableValue as their record value, which is used to get - /// the actual record from the primary memtable `records` map. - /// - /// Note: it must be invariant that all memtables (primary and secondary) - /// contain the same keys. - pub records: BTreeMap, - /// A max heap priority queue of keys. The record with least priority is evicted - /// from the primary memtable and any secondary memtables that reference it, when - /// the memtable reaches capacity. - /// - /// Note: n_operations must be negated upon append to evict oldest values first. - evict_queue: PriorityQueue, - /// Policy for prioritizing records for eviction. - evict_policy: MemtableEvictPolicy, - /// Running counter of memtable operations, used as priority - /// in evict_queue. - n_operations: u64, -} - -impl PrimaryMemtable { - pub fn new(capacity: usize, evict_policy: MemtableEvictPolicy) -> PrimaryMemtable { - PrimaryMemtable { - capacity, - n_operations: 0, - records: BTreeMap::new(), - evict_queue: PriorityQueue::new(), - evict_policy, - } - } - - pub fn set(&mut self, key: &IndexableValue, value: &Record) { - self.records.insert(key.clone(), value.clone()); - - if self.evict_policy == MemtableEvictPolicy::LeastWritten - || self.evict_policy == MemtableEvictPolicy::LeastReadOrWritten - { - self.set_priority(&key); - } - } - - pub fn get(&mut self, key: &IndexableValue) -> Option<&Record> { - if self.evict_policy == MemtableEvictPolicy::LeastRead - || self.evict_policy == MemtableEvictPolicy::LeastReadOrWritten - { - self.set_priority(&key); - } - - self.records.get(key) - } - - pub fn get_without_update(&self, key: &IndexableValue) -> Option<&Record> { - self.records.get(key) - } - - fn set_priority(&mut self, key: &IndexableValue) { - let priority = self.get_and_increment_current_priority(); - match self.evict_queue.get(key) { - Some(_) => { - self.evict_queue.change_priority(key, priority); - } - None => { - self.evict_queue.push(key.clone(), priority); - } - } - } - - fn get_and_increment_current_priority(&mut self) -> i64 { - let ret = -(self.n_operations as i64); - self.n_operations += 1; - ret - } - - pub fn evict_if_necessary(&mut self) -> Option { - if self.records.len() >= self.capacity { - let (evict_key, _prio) = self.evict_queue.pop().expect("Evict queue was empty"); - let removed = self - .records - .remove(&evict_key) - .expect("Key was not found in records"); - Some(removed) - } else { - None - } - } -} diff --git a/log_db/src/reverse_log_reader.rs b/log_db/src/reverse_log_reader.rs deleted file mode 100644 index f042406..0000000 --- a/log_db/src/reverse_log_reader.rs +++ /dev/null @@ -1,354 +0,0 @@ -use super::common::*; -use std::fs::{self}; -use std::io::{self, Read, Seek, SeekFrom}; - -pub struct ReverseLogReader<'a> { - /// The file to read from end to beginning. - file: &'a mut fs::File, - /// The internal buffer used to read from the file. - /// It is populated with the last INTERNAL_BUF_SIZE bytes read from the file - /// and is used to read records in reverse order - internal_buf: Vec, - /// The current position in the internal buffer. It is decremented as bytes are read - /// from the buffer. When a read is requested and the internal position is 0, the buffer - /// is populated with the next (= closer to the start of the file) INTERNAL_BUF_SIZE bytes from the file. - /// Note: This is the index of the next byte to be read from the internal buffer + 1 - internal_pos: usize, - /// A flag indicating whether the record separator at the cursor position has been consumed. - /// Useful to avoid consuming the separator once when reading until an escape character, and - /// a second time when reading a new record and validating it ends in a separator. - consumed_record_sep: bool, -} - -// This value is based on the reverse_read_file_with_various_buffer_sizes benchmark. -// Greater values yield little to no performance improvement. -const DEFAULT_INTERNAL_BUF_SIZE: usize = 32768; -impl<'a> ReverseLogReader<'a> { - pub fn new(file: &mut fs::File) -> Result { - file.seek(SeekFrom::End(0))?; - Ok(ReverseLogReader { - file, - internal_buf: vec![0; DEFAULT_INTERNAL_BUF_SIZE], - internal_pos: 0, - consumed_record_sep: false, - }) - } - - pub fn new_with_size( - file: &mut fs::File, - internal_buf_size: usize, - ) -> Result { - file.seek(SeekFrom::End(0))?; - Ok(ReverseLogReader { - file, - internal_buf: vec![0; internal_buf_size], - internal_pos: 0, - consumed_record_sep: false, - }) - } - - pub fn read_record(&mut self) -> Result, io::Error> { - if self.file.stream_position()? == 0 && self.internal_pos == 0 { - return Ok(None); - } - - if !self.consumed_record_sep { - // Check that record ends with a record separator - match self.read_special_sequence()? { - SpecialSequence::RecordSeparator => {} - _ => { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - "Record does not end in record separator", - )); - } - } - } - self.consumed_record_sep = false; - - let mut result_buf: Vec = vec![]; - let mut read_buf = Vec::with_capacity(self.internal_buf.len()); - loop { - read_buf.clear(); - let read = self.read_until(ESCAPE_CHARACTER, &mut read_buf)?; - - result_buf.extend(&read_buf[..read]); - - if self.file.stream_position()? == 0 && self.internal_pos == 0 { - // We read until the start of the file, we are done - break; - } - - match self.read_special_sequence()? { - SpecialSequence::LiteralEscape => { - result_buf.push(ESCAPE_CHARACTER); - } - SpecialSequence::LiteralFieldSeparator => { - result_buf.push(FIELD_SEPARATOR); - } - SpecialSequence::RecordSeparator => { - self.consumed_record_sep = true; - break; - } - } - } - - result_buf.reverse(); - Ok(Some(Record::deserialize(&result_buf))) - } - - /// Read exactly `buf.len()` bytes from the file, return an error if the file is exhausted. - /// The bytes are returned in start -> end order. - /// If an error is returned, the contents of `buf` are in an undefined state. - fn read_exact(&mut self, buf: &mut [u8]) -> Result { - let mut read = 0; - while read < buf.len() { - if self.internal_pos == 0 { - let populated_n = self.populate_internal_buf()?; - if populated_n == 0 { - return Err(io::Error::new( - io::ErrorKind::UnexpectedEof, - "Unexpected end of file", - )); - } - } - - let end = buf.len() - read; - let n = std::cmp::min(self.internal_pos, end); - buf[end - n..end] - .copy_from_slice(&self.internal_buf[self.internal_pos - n..self.internal_pos]); - read += n; - self.internal_pos -= n; - } - - Ok(read) - } - - fn populate_internal_buf(&mut self) -> Result { - let current_seek_pos = self.file.stream_position()? as usize; - - // Seek back by the size of the internal buffer or to the beginning of the file - let seek_length = if current_seek_pos > self.internal_buf.len() { - self.internal_pos = self.internal_buf.len(); - self.internal_buf.len() - } else { - self.internal_buf = vec![0; current_seek_pos as usize]; - self.internal_pos = current_seek_pos as usize; - current_seek_pos - }; - - self.file.seek_relative(-(seek_length as i64))?; - self.file.read_exact(&mut self.internal_buf)?; - self.file.seek_relative(-(seek_length as i64))?; - - Ok(seek_length as usize) - } - - /// Iterate over the internal buffer with internal_pos as the index. - /// If the byte is found or the file has been exhausted, we return the number of bytes read. - /// The `buf` parameter is used to store the bytes read from the internal buffer, excluding the found byte, - /// in reverse order. - fn read_until(&mut self, byte: u8, buf: &mut Vec) -> Result { - // TODO: optimize this - let mut read = 0; - loop { - // If we reach internal_pos == 0, we need to populate the internal buffer. - if self.internal_pos == 0 { - let populated_n = self.populate_internal_buf()?; - if populated_n == 0 { - return Ok(read); - } - } - - while self.internal_pos > 0 { - let index = self.internal_pos - 1; - if self.internal_buf[index] == byte { - return Ok(read); - } - buf.push(self.internal_buf[index]); - read += 1; - self.internal_pos -= 1; - } - } - } - - fn read_special_sequence(&mut self) -> Result { - let mut special_buf: Vec = vec![0; SEQ_RECORD_SEP.len()]; - self.read_exact(&mut special_buf)?; - - match validate_special(&special_buf.as_slice()) { - Some(special) => Ok(special), - None => { - let pos = self.file.stream_position().unwrap() + self.internal_pos as u64; - - Err(io::Error::new( - io::ErrorKind::InvalidData, - format!( - "Not a special sequence: {:?} at pos: {:x}", - special_buf, pos, - ), - )) - } - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - use std::io::Write; - use std::path::Path; - - #[test] - fn test_read_until_found() { - let mut file = tempfile::tempfile().unwrap(); - file.write_all(b"hello,world").unwrap(); - let mut reader = ReverseLogReader::new(&mut file).unwrap(); - let mut buf = vec![]; - assert_eq!(reader.read_until(b',', &mut buf).unwrap(), 5); - assert_eq!(buf, b"dlrow"); - } - - #[test] - fn test_read_until_not_found() { - let mut file = tempfile::tempfile().unwrap(); - file.write_all(b"hello,world").unwrap(); - let mut reader = ReverseLogReader::new(&mut file).unwrap(); - let mut buf = vec![]; - let read = reader.read_until(b'!', &mut buf).unwrap(); - assert_eq!(buf, b"dlrow,olleh"); - assert_eq!(read, 11); - } - - #[test] - fn test_read_special_sequence() { - let mut file = tempfile::tempfile().unwrap(); - let mut buf = vec![]; - buf.extend(SEQ_RECORD_SEP); - buf.extend(SEQ_LIT_ESCAPE); - buf.extend(SEQ_LIT_FIELD_SEP); - // Note: written in start -> end order, read end -> start - file.write_all(&buf).unwrap(); - - let mut reader = ReverseLogReader::new(&mut file).unwrap(); - assert_eq!( - reader.read_special_sequence().unwrap(), - SpecialSequence::LiteralFieldSeparator - ); - assert_eq!( - reader.read_special_sequence().unwrap(), - SpecialSequence::LiteralEscape - ); - assert_eq!( - reader.read_special_sequence().unwrap(), - SpecialSequence::RecordSeparator - ); - } - - #[test] - fn test_populate_internal_buf() { - let mut file = tempfile::tempfile().unwrap(); - file.write_all(b"hello,world").unwrap(); - let mut reader = ReverseLogReader::new_with_size(&mut file, 3).unwrap(); - - reader.populate_internal_buf().unwrap(); - assert_eq!( - String::from_utf8(reader.internal_buf.clone()).unwrap(), - "rld".to_string() - ); - - reader.populate_internal_buf().unwrap(); - assert_eq!( - String::from_utf8(reader.internal_buf.clone()).unwrap(), - ",wo".to_string() - ); - - reader.populate_internal_buf().unwrap(); - assert_eq!( - String::from_utf8(reader.internal_buf.clone()).unwrap(), - "llo".to_string() - ); - - reader.populate_internal_buf().unwrap(); - assert_eq!( - String::from_utf8(reader.internal_buf.clone()).unwrap(), - "he".to_string() - ); - } - - #[test] - fn test_reverse_log_reader_fixture_db1() { - let db_path = Path::new(TEST_RESOURCES_DIR).join("test_db1"); - let mut file = fs::OpenOptions::new() - .read(true) - .open(&db_path) - .expect("Failed to open file"); - let mut reverse_log_reader = ReverseLogReader::new(&mut file).unwrap(); - - // There are two records in the log with "schema": Int, Null - - let last_record = reverse_log_reader - .next() - .expect("Failed to read the last record"); - assert!(match last_record.values.as_slice() { - [RecordValue::Int(10), RecordValue::Null] => true, - _ => false, - }); - - let first_record = reverse_log_reader - .next() - .expect("Failed to read the first record"); - assert!(match first_record.values.as_slice() { - // Note: the int value is equal to the escape byte - [RecordValue::Int(0x1D), RecordValue::Null] => true, - _ => false, - }); - - assert!(reverse_log_reader.next().is_none()); - } - - #[test] - fn test_read_exact() { - let mut file = tempfile::tempfile().unwrap(); - file.write_all(b"hello,world").unwrap(); - let mut reader = ReverseLogReader::new(&mut file).unwrap(); - let mut buf = vec![0; 3]; - - let read = reader.read_exact(&mut buf).unwrap(); - assert_eq!(buf, b"rld"); - assert_eq!(read, 3); - - let read = reader.read_exact(&mut buf).unwrap(); - assert_eq!(buf, b",wo"); - assert_eq!(read, 3); - - let read = reader.read_exact(&mut buf).unwrap(); - assert_eq!(buf, b"llo"); - assert_eq!(read, 3); - - assert!(reader.read_exact(&mut buf).unwrap_err().kind() == io::ErrorKind::UnexpectedEof); - } - - #[test] - fn test_read_exact_insufficient_bytes() { - let mut file = tempfile::tempfile().unwrap(); - file.write_all(b"hello").unwrap(); - let mut reader = ReverseLogReader::new(&mut file).unwrap(); - let mut buf = vec![0; 10]; - assert!(reader.read_exact(&mut buf).unwrap_err().kind() == io::ErrorKind::UnexpectedEof); - } -} - -impl Iterator for ReverseLogReader<'_> { - type Item = Record; - - fn next(&mut self) -> Option { - match self.read_record() { - Ok(Some(record)) => Some(record), - Ok(None) => None, - Err(err) => { - panic!("Error reading record: {:?}", err,) - } - } - } -} diff --git a/log_db/src/secondary_memtable.rs b/log_db/src/secondary_memtable.rs deleted file mode 100644 index 994ee0d..0000000 --- a/log_db/src/secondary_memtable.rs +++ /dev/null @@ -1,111 +0,0 @@ -use super::*; -use std::collections::BTreeMap; -use std::collections::HashSet; -use std::fmt::Debug; - -pub struct SecondaryMemtable { - pub field: Field, - field_index: usize, - primary_key_index: usize, - - /// Map of records indexed by key. The value is the set of primary key values of records - /// that have the secondary key value. The actual `Record` objects are stored in the - /// primary memtable, which acts as the shared heap. - pub records: BTreeMap>, -} - -impl SecondaryMemtable { - pub fn new( - field_schema: &Vec<(Field, RecordField)>, - field: &Field, - primary_key_index: usize, - ) -> SecondaryMemtable { - let field_index = field_schema - .iter() - .position(|(f, _)| f == field) - .expect("Field not found in schema"); - - SecondaryMemtable { - field: field.clone(), - field_index, - primary_key_index, - records: BTreeMap::new(), - } - } - - pub fn set(&mut self, key: &IndexableValue, value: &IndexableValue) { - debug!( - "Inserting/updating record in secondary memtable with key {:?} = {:?}", - &key, &value, - ); - - match self.records.get_mut(key) { - Some(existing) => { - debug!( - "Existing entry found with {} records in the set", - &existing.len() - ); - existing.insert(value.clone()); - } - None => { - debug!("No existing entry found, creating one."); - let mut set = HashSet::with_capacity(1); - set.insert(value.clone()); - self.records.insert(key.clone(), set); - } - } - } - - pub fn set_all(&mut self, key: &IndexableValue, values: &[IndexableValue]) { - debug!( - "Replacing set of records in secondary memtable with key {:?} ({} values)", - &key, - &values.len(), - ); - - let mut set = HashSet::with_capacity(values.len()); - values.iter().for_each(|value| { - set.insert(value.clone()); - }); - - self.records.insert(key.clone(), set); - } - - pub fn find_all( - &mut self, - primary_memtable: &PrimaryMemtable, - key: &IndexableValue, - ) -> Vec { - match self.records.get(key) { - None => vec![], - Some(set) => set - .iter() - .map(|key| { - primary_memtable - .get_without_update(key) - .expect("Record not found") - .clone() - }) - .collect(), - } - } - - pub fn remove(&mut self, record: &Record) { - let key = record.values[self.field_index] - .as_indexable() - .expect("Field is not indexable"); - - let primary_key = record.values[self.primary_key_index] - .as_indexable() - .expect("Primary key is not indexable"); - - match self.records.get_mut(&key) { - Some(set) => { - set.remove(&primary_key); - } - None => { - panic!("Record not found in secondary memtable"); - } - } - } -} -- cgit v1.3