diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/common.rs | 25 | ||||
| -rw-r--r-- | src/lib.rs | 92 | ||||
| -rw-r--r-- | src/log_reader.rs | 123 |
3 files changed, 183 insertions, 57 deletions
diff --git a/src/common.rs b/src/common.rs index f7aa755..4922c73 100644 --- a/src/common.rs +++ b/src/common.rs @@ -12,9 +12,28 @@ pub const ACTIVE_LOG_FILENAME: &str = "db"; pub const DEFAULT_READ_BUF_SIZE: usize = 1024 * 1024; // 1 MB pub const FIELD_SEPARATOR: u8 = b'\x1C'; pub const ESCAPE_CHARACTER: u8 = b'\x1D'; -pub const SEQ_RECORD_SEP: &[u8] = &[FIELD_SEPARATOR, FIELD_SEPARATOR, ESCAPE_CHARACTER]; -pub const SEQ_LIT_ESCAPE: &[u8] = &[ESCAPE_CHARACTER, ESCAPE_CHARACTER, ESCAPE_CHARACTER]; -pub const SEQ_LIT_FIELD_SEP: &[u8] = &[ESCAPE_CHARACTER, FIELD_SEPARATOR, ESCAPE_CHARACTER]; + +// Special sequences. Note: these must have the same length! +// Since the log is read both forwards and backwards, we must have a signal +// character (ESCAPE_CHARACTER) on both sides of the special sequence. +pub const SEQ_RECORD_SEP: &[u8] = &[ + ESCAPE_CHARACTER, + FIELD_SEPARATOR, + FIELD_SEPARATOR, + ESCAPE_CHARACTER, +]; +pub const SEQ_LIT_ESCAPE: &[u8] = &[ + ESCAPE_CHARACTER, + ESCAPE_CHARACTER, + ESCAPE_CHARACTER, + ESCAPE_CHARACTER, +]; +pub const SEQ_LIT_FIELD_SEP: &[u8] = &[ + ESCAPE_CHARACTER, + ESCAPE_CHARACTER, + FIELD_SEPARATOR, + ESCAPE_CHARACTER, +]; #[derive(Debug, Eq, PartialEq)] pub enum SpecialSequence { @@ -9,7 +9,8 @@ mod secondary_memtable; pub use common::*; use fs2::FileExt; -pub use log_reader::LogReader; +pub use log_reader::ForwardLogReader; +pub use log_reader::ReverseLogReader; use primary_memtable::PrimaryMemtable; use secondary_memtable::SecondaryMemtable; use std::fmt::Debug; @@ -146,7 +147,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> { } fn initialize(config: &Config<Field>) -> Result<DB<Field>, io::Error> { - info!("Initializing DB"); + info!("Initializing DB..."); // If data_dir does not exist, create it if !fs::exists(&config.data_dir)? { fs::create_dir_all(&config.data_dir)?; @@ -214,13 +215,25 @@ impl<Field: Eq + Clone + Debug> DB<Field> { }) .collect(); - let db = DB::<Field> { + let mut db = DB::<Field> { config: config.clone(), - log_path, + log_path: log_path.clone(), primary_key_index, primary_memtable, secondary_memtables, }; + + info!("Rebuilding memtable indexes..."); + let mut file = fs::OpenOptions::new().read(true).open(&log_path)?; + + let forward_log_reader = ForwardLogReader::new(&mut file); + for record in forward_log_reader { + db.update_primary_index(&record); + db.update_secondary_indexes(&record); + } + + info!("Database ready."); + Ok(db) } @@ -295,34 +308,12 @@ impl<Field: Eq + Clone + Debug> DB<Field> { file.unlock()?; debug!("Record appended to log file, lock released"); - debug!("Updating primary memtable"); - let primary_value = - &record.values[self.primary_key_index] - .as_indexable() - .ok_or(io::Error::new( - io::ErrorKind::InvalidInput, - "Primary key must be an IndexableValue", - ))?; - - self.primary_memtable.set(primary_value, record); + debug!("Updating primary memtable"); + self.update_primary_index(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 key = record.values[index] - .as_indexable() - .expect("Secondary index key was not indexable"); - secondary_memtable.set(&key, record); - } - } - }); + debug!("Updating secondary memtables"); + self.update_secondary_indexes(record); Ok(()) } @@ -372,8 +363,8 @@ impl<Field: Eq + Clone + Debug> DB<Field> { debug!("Lock acquired, searching log file for record"); - let mut log_reader = LogReader::new(&mut file)?; - let result = log_reader.find(|record| { + let mut reverse_log_reader = ReverseLogReader::new(&mut file)?; + let result = reverse_log_reader.find(|record| { let record_key = record.values[self.primary_key_index] .as_indexable() .expect("A non-indexable value was stored at key index"); @@ -391,8 +382,13 @@ impl<Field: Eq + Clone + Debug> DB<Field> { } }; - debug!("Found matching record in log file. Storing result in primary memtable."); - self.primary_memtable.set(&query_key, &result_value); + debug!("Found matching record in log file."); + + debug!("Updating primary memtable"); + self.update_primary_index(&result_value); + + debug!("Updating secondary memtables"); + self.update_secondary_indexes(&result_value); Ok(Some(result_value)) } @@ -472,7 +468,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> { debug!("Lock acquired, searching log file for record"); - let result = LogReader::new(&mut file)? + let result = ReverseLogReader::new(&mut file)? .filter(|record| { let record_key = record.values[key_index] .as_indexable() @@ -496,4 +492,30 @@ impl<Field: Eq + Clone + Debug> DB<Field> { Ok(result) } + + fn update_primary_index(&mut self, record: &Record) { + let key = record.values[self.primary_key_index] + .as_indexable() + .expect("A non-indexable value was stored at key index"); + self.primary_memtable.set(&key, record); + } + + fn update_secondary_indexes(&mut self, record: &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 key = record.values[index] + .as_indexable() + .expect("Secondary index key was not indexable"); + secondary_memtable.set(&key, record); + } + } + }); + } } diff --git a/src/log_reader.rs b/src/log_reader.rs index 0ee0a81..e32ef53 100644 --- a/src/log_reader.rs +++ b/src/log_reader.rs @@ -1,16 +1,30 @@ use super::common::*; use rev_buf_reader::RevBufReader; use std::fs::{self}; -use std::io::{self, BufRead, Read, Seek}; +use std::io::{self, BufRead, Read, Seek, SeekFrom}; -pub struct LogReader<'a> { +/// There are three special characters that need to be handled: +/// Here: SC = escape char, FS = field separator. +/// - FS FS SC -> actual record separator +/// - SC FS SC -> literal FS +/// - SC SC SC -> literal SC +fn validate_special(buf: &[u8]) -> Option<SpecialSequence> { + match buf { + SEQ_RECORD_SEP => Some(SpecialSequence::RecordSeparator), + SEQ_LIT_FIELD_SEP => Some(SpecialSequence::LiteralFieldSeparator), + SEQ_LIT_ESCAPE => Some(SpecialSequence::LiteralEscape), + _ => None, + } +} + +pub struct ReverseLogReader<'a> { rev_reader: RevBufReader<&'a mut fs::File>, } -impl<'a> LogReader<'a> { - pub fn new(file: &mut fs::File) -> Result<LogReader, io::Error> { +impl<'a> ReverseLogReader<'a> { + pub fn new(file: &mut fs::File) -> Result<ReverseLogReader, io::Error> { let rev_reader = RevBufReader::new(file); - Ok(LogReader { rev_reader }) + Ok(ReverseLogReader { rev_reader }) } fn read_record(&mut self) -> Result<Option<Record>, io::Error> { @@ -48,7 +62,7 @@ impl<'a> LogReader<'a> { SpecialSequence::RecordSeparator => { // The record is complete, so we can break out of the loop. // Move the cursor back to the beginning of the special sequence. - self.rev_reader.seek_relative(3)?; + self.rev_reader.seek_relative(SEQ_RECORD_SEP.len() as i64)?; break; } SpecialSequence::LiteralFieldSeparator => { @@ -68,7 +82,7 @@ impl<'a> LogReader<'a> { } fn read_special_sequence(&mut self) -> Result<SpecialSequence, io::Error> { - let mut special_buf: Vec<u8> = vec![0; 3]; + let mut special_buf: Vec<u8> = vec![0; SEQ_RECORD_SEP.len()]; self.rev_reader.read_exact(&mut special_buf)?; match validate_special(&special_buf.as_slice()) { @@ -81,7 +95,7 @@ impl<'a> LogReader<'a> { } } -impl Iterator for LogReader<'_> { +impl Iterator for ReverseLogReader<'_> { type Item = Record; fn next(&mut self) -> Option<Self::Item> { @@ -93,16 +107,87 @@ impl Iterator for LogReader<'_> { } } -/// There are three special characters that need to be handled: -/// Here: SC = escape char, FS = field separator. -/// - FS FS SC -> actual record separator -/// - SC FS SC -> literal FS -/// - SC SC SC -> literal SC -fn validate_special(buf: &[u8]) -> Option<SpecialSequence> { - match buf { - SEQ_RECORD_SEP => Some(SpecialSequence::RecordSeparator), - SEQ_LIT_FIELD_SEP => Some(SpecialSequence::LiteralFieldSeparator), - SEQ_LIT_ESCAPE => Some(SpecialSequence::LiteralEscape), - _ => None, +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<Option<Record>, io::Error> { + // The buffer that stores the bytes read from the file. + let mut read_buf: Vec<u8> = Vec::new(); + // The buffer that stores all the bytes of the record read so far in reverse order. + let mut result_buf: Vec<u8> = 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, 1]; + match self.reader.read_exact(&mut peek_buf) { + Ok(_) => { + // Go back one byte (not sure why you need to seek by -2 here?) + self.reader.seek_relative(-2)?; + } + 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<SpecialSequence, io::Error> { + let mut special_buf: Vec<u8> = 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<Self::Item> { + match self.read_record() { + Ok(Some(record)) => Some(record), + Ok(None) => None, + Err(err) => panic!("Error reading record: {:?}", err), + } } } |