diff options
| -rw-r--r-- | log_db/src/common.rs | 46 | ||||
| -rw-r--r-- | log_db/src/lib.rs | 61 | ||||
| -rw-r--r-- | log_db/src/log_reader_forward.rs | 113 | ||||
| -rw-r--r-- | log_db/tests/resources/test_data_1 | 254 | ||||
| -rw-r--r-- | log_db/tests/resources/test_metadata_1 | bin | 0 -> 40 bytes |
5 files changed, 337 insertions, 137 deletions
diff --git a/log_db/src/common.rs b/log_db/src/common.rs index c08f212..448226d 100644 --- a/log_db/src/common.rs +++ b/log_db/src/common.rs @@ -4,6 +4,7 @@ use std::fmt::Display; use std::fs::{metadata, File}; use std::io::{self}; use std::path::{Path, PathBuf}; +use uuid::Uuid; // For Unix-like systems #[cfg(unix)] @@ -17,48 +18,8 @@ pub const ACTIVE_SYMLINK_FILENAME: &str = "active"; pub const METADATA_FILE_HEADER_SIZE: usize = 24; 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'; -pub const ESCAPE_CHARACTER: u8 = b'\x1D'; pub const TEST_RESOURCES_DIR: &str = "tests/resources"; -// 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, -]; - -/// There are three special sequences that need to be handled: -/// Here: SC = escape char, FS = field separator. -/// - SC FS FS SC -> actual record separator -/// - SC SC FS SC -> literal FS -/// - SC SC SC SC -> literal SC -/// -/// Returns SpecialSequence or None if not valid. -pub 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, - } -} - #[derive(Debug, Eq, PartialEq)] pub enum SpecialSequence { RecordSeparator, @@ -155,6 +116,11 @@ impl Ord for LogKeySet { } } +pub struct MetadataHeader { + pub version: u8, + pub uuid: Uuid, +} + #[derive(Debug, Clone, Eq, PartialEq)] pub enum WriteDurability { /// Changes are written to an application-level write buffer without flushing to the OS write buffer or syncing to disk. diff --git a/log_db/src/lib.rs b/log_db/src/lib.rs index 868d0c9..b0de03f 100644 --- a/log_db/src/lib.rs +++ b/log_db/src/lib.rs @@ -17,7 +17,9 @@ use memtable_secondary::SecondaryMemtable; use std::collections::BTreeMap; use std::fmt::Debug; use std::fs::{self}; -use std::io::{self, Write}; +use std::io::Seek; +use std::io::SeekFrom; +use std::io::{self, Read, Write}; use std::os::unix::fs::MetadataExt; use std::path::{Path, PathBuf}; use std::thread; @@ -141,8 +143,8 @@ struct Config<Field: Eq + Clone> { pub struct DB<Field: Eq + Clone + Debug> { config: Config<Field>, - log_path: PathBuf, - log_file: fs::File, + active_metadata_file: fs::File, + active_data_file: fs::File, primary_key_index: usize, primary_memtable: PrimaryMemtable, secondary_memtables: Vec<SecondaryMemtable>, @@ -224,32 +226,33 @@ impl<Field: Eq + Clone + Debug> DB<Field> { .collect(); let active_symlink = Path::new(&config.data_dir).join(ACTIVE_SYMLINK_FILENAME); - fs::read_dir(&config.data_dir)?.for_each(|entry| if let Ok(entry) = entry {}); let active_target = fs::read_link(&active_symlink)?; - let log_path = Path::new(&config.data_dir).join(active_target); - let log_file = fs::OpenOptions::new() + let active_metadata_path = Path::new(&config.data_dir).join(active_target); + let mut active_metadata_file = fs::OpenOptions::new() .read(true) .append(true) - .open(&log_path)?; + .open(&active_metadata_path)?; - let mut memtable_file = fs::OpenOptions::new().read(true).open(&log_path)?; + let active_metadata_header = DB::<Field>::read_metadata_header(&mut active_metadata_file)?; + let active_data_path = + Path::new(&config.data_dir).join(active_metadata_header.uuid.to_string()); + let active_data_file = fs::OpenOptions::new() + .read(true) + .append(true) + .open(&active_data_path)?; - let mut db = DB::<Field> { + let db = DB::<Field> { config: config.clone(), - log_path, - log_file, + active_metadata_file, + active_data_file, primary_key_index, primary_memtable, secondary_memtables, }; - info!("Rebuilding memtable indexes..."); - - let forward_log_reader = ForwardLogReader::new(&mut memtable_file); - for record in forward_log_reader { - db.insert_to_memtables(&record); - } + // info!("Rebuilding memtable indexes..."); + // TODO FIXME build memtable indexes info!("Database ready."); @@ -329,10 +332,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> { debug!("Lock acquired, appending to log file"); // Write the record to the log - // Each serialized row is suffixed with the field separator character sequence - let mut serialized_record = record.serialize(); - serialized_record.extend(SEQ_RECORD_SEP); - self.log_file.write_all(&serialized_record)?; + self.log_file.write_all(&record.serialize())?; // Flush and sync to disk if self.config.write_durability == WriteDurability::Flush { @@ -768,9 +768,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> { .open(temp_path)?; for entry in map.values() { - let mut serialized = entry.serialize(); - serialized.extend(SEQ_RECORD_SEP); - temp_file.write_all(&serialized)?; + temp_file.write_all(&entry.serialize())?; } fs::rename(&temp_path, path)?; @@ -874,4 +872,19 @@ impl<Field: Eq + Clone + Debug> DB<Field> { Ok(()) } + + /// Reads the metadata header from the metadata file. + /// Leaves the file seek head at the beginning of the records, after the header. + fn read_metadata_header(metadata_file: &mut fs::File) -> Result<MetadataHeader, io::Error> { + let mut version_buf = vec![0u8; 1]; + metadata_file.seek(SeekFrom::Start(0))?; + metadata_file.read_exact(&mut version_buf)?; + let version = version_buf[0]; + + let mut uuid_buf = vec![0u8; 16]; + metadata_file.seek_relative(7)?; // skip over padding + metadata_file.read_exact(&mut uuid_buf)?; + let uuid = Uuid::from_slice(&uuid_buf).expect("Invalid UUID"); + Ok(MetadataHeader { version, uuid }) + } } diff --git a/log_db/src/log_reader_forward.rs b/log_db/src/log_reader_forward.rs index 03726d0..808d3fa 100644 --- a/log_db/src/log_reader_forward.rs +++ b/log_db/src/log_reader_forward.rs @@ -1,81 +1,50 @@ use super::common::*; use std::fs::{self}; -use std::io::{self, BufRead, Read}; +use std::io::{self, Read, Seek}; -pub struct ForwardLogReader<'a> { - reader: io::BufReader<&'a mut fs::File>, +pub struct ForwardLogReader { + metadata_reader: io::BufReader<fs::File>, + data_reader: io::BufReader<fs::File>, } -impl<'a> ForwardLogReader<'a> { - pub fn new(file: &mut fs::File) -> ForwardLogReader { - let reader = io::BufReader::new(file); - ForwardLogReader { reader } +impl<'a> ForwardLogReader { + pub fn new(metadata_file: fs::File, data_file: fs::File) -> ForwardLogReader { + let mut ret = ForwardLogReader { + metadata_reader: io::BufReader::new(metadata_file), + data_reader: io::BufReader::new(data_file), + }; + + ret.metadata_reader + .seek(io::SeekFrom::Start(METADATA_FILE_HEADER_SIZE as u64)) + .expect("Seek failed"); + + ret } 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]; - 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 => { + let mut metadata_entry_buf = vec![0; 16]; // 2x u64 + if let Err(e) = self.metadata_reader.read_exact(&mut metadata_entry_buf) { + if e.kind() == io::ErrorKind::UnexpectedEof { return Ok(None); - } - Err(e) => { + } else { 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]); + // First u64 is the offset of the record in the data file, second is the length of the record + let entry_offset = u64::from_be_bytes(metadata_entry_buf[0..8].try_into().unwrap()); + let entry_length = u64::from_be_bytes(metadata_entry_buf[8..16].try_into().unwrap()); - // 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); - } - } - } + self.data_reader.seek(io::SeekFrom::Start(entry_offset))?; + let mut result_buf = vec![0; entry_length as usize]; + self.data_reader.read_exact(&mut result_buf)?; 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<'_> { +impl Iterator for ForwardLogReader { type Item = Record; fn next(&mut self) -> Option<Self::Item> { @@ -94,28 +63,26 @@ mod tests { #[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() + let metadata_path = Path::new(TEST_RESOURCES_DIR).join("test_metadata_1"); + let data_path = Path::new(TEST_RESOURCES_DIR).join("test_data_1"); + let metadata_file = fs::OpenOptions::new() .read(true) - .open(&db_path) - .expect("Failed to open file"); - let mut forward_log_reader = ForwardLogReader::new(&mut file); + .open(&metadata_path) + .expect("Failed to open metadata file"); + let data_file = fs::OpenOptions::new() + .read(true) + .open(&data_path) + .expect("Failed to open data file"); + + let mut forward_log_reader = ForwardLogReader::new(metadata_file, data_file); - // There are two records in the log with "schema": Int, Null + // There are two records in the log with "schema" with one field: Bytes 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, + [RecordValue::Bytes(_)] => true, _ => false, }); diff --git a/log_db/tests/resources/test_data_1 b/log_db/tests/resources/test_data_1 new file mode 100644 index 0000000..d4c6a21 --- /dev/null +++ b/log_db/tests/resources/test_data_1 @@ -0,0 +1,254 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/log_db/tests/resources/test_metadata_1 b/log_db/tests/resources/test_metadata_1 Binary files differnew file mode 100644 index 0000000..9694599 --- /dev/null +++ b/log_db/tests/resources/test_metadata_1 |