diff options
| -rw-r--r-- | Cargo.lock | 154 | ||||
| -rw-r--r-- | Cargo.toml | 7 | ||||
| -rw-r--r-- | src/lib.rs | 376 | ||||
| -rw-r--r-- | tests/integration.rs | 129 | ||||
| -rw-r--r-- | tests/resources/test_db1 | bin | 0 -> 28 bytes |
5 files changed, 607 insertions, 59 deletions
@@ -3,6 +3,64 @@ version = 3 [[package]] +name = "aho-corasick" +version = "1.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e60d3430d3a69478ad0993f19238d2df97c507009a52b3c10addcd7f6bcb916" +dependencies = [ + "memchr", +] + +[[package]] +name = "anstream" +version = "0.6.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "64e15c1ab1f89faffbf04a634d5e1962e9074f2741eef6d97f3c4e322426d526" +dependencies = [ + "anstyle", + "anstyle-parse", + "anstyle-query", + "anstyle-wincon", + "colorchoice", + "is_terminal_polyfill", + "utf8parse", +] + +[[package]] +name = "anstyle" +version = "1.0.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1bec1de6f59aedf83baf9ff929c98f2ad654b97c9510f4e70cf6f661d49fd5b1" + +[[package]] +name = "anstyle-parse" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eb47de1e80c2b463c735db5b217a0ddc39d612e7ac9e2e96a5aed1f57616c1cb" +dependencies = [ + "utf8parse", +] + +[[package]] +name = "anstyle-query" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6d36fc52c7f6c869915e99412912f22093507da8d9e942ceaf66fe4b7c14422a" +dependencies = [ + "windows-sys", +] + +[[package]] +name = "anstyle-wincon" +version = "3.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5bf74e1b6e971609db8ca7a9ce79fd5768ab6ae46441c572e46cf596f59e57f8" +dependencies = [ + "anstyle", + "windows-sys", +] + +[[package]] name = "autocfg" version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -21,6 +79,35 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd" [[package]] +name = "colorchoice" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3fd119d74b830634cea2a0f58bbd0d54540518a14397557951e79340abc28c0" + +[[package]] +name = "env_filter" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4f2c92ceda6ceec50f43169f9ee8424fe2db276791afde7b2cd8bc084cb376ab" +dependencies = [ + "log", + "regex", +] + +[[package]] +name = "env_logger" +version = "0.11.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e13fa619b91fb2381732789fc5de83b45675e882f66623b7d8cb4f643017018d" +dependencies = [ + "anstream", + "anstyle", + "env_filter", + "humantime", + "log", +] + +[[package]] name = "equivalent" version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -120,6 +207,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" [[package]] +name = "humantime" +version = "2.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a3a5bfb195931eeb336b2a7b4d761daec841b97f947d34394601737a7bba5e4" + +[[package]] name = "indexmap" version = "2.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -130,6 +223,12 @@ dependencies = [ ] [[package]] +name = "is_terminal_polyfill" +version = "1.70.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7943c866cc5cd64cbc25b2e01621d07fa8eb2a1a23160ee81ce38704e97b8ecf" + +[[package]] name = "libc" version = "0.2.159" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -155,9 +254,11 @@ checksum = "a7a70ba024b9dc04c27ea2f0c0548feb474ec5c54bba33a7f72f873a39d07b24" name = "log_db" version = "0.1.0" dependencies = [ + "env_logger", "fs2", "log", "priority-queue", + "rev_buf_reader", "serial_test", ] @@ -247,6 +348,44 @@ dependencies = [ ] [[package]] +name = "regex" +version = "1.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "38200e5ee88914975b69f657f0801b6f6dccafd44fd9326302a4aaeecfacb1d8" +dependencies = [ + "aho-corasick", + "memchr", + "regex-automata", + "regex-syntax", +] + +[[package]] +name = "regex-automata" +version = "0.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "368758f23274712b504848e9d5a6f010445cc8b87a7cdb4d7cbee666c1288da3" +dependencies = [ + "aho-corasick", + "memchr", + "regex-syntax", +] + +[[package]] +name = "regex-syntax" +version = "0.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b15c43186be67a4fd63bee50d0303afffcef381492ebe2c5d87f324e1b8815c" + +[[package]] +name = "rev_buf_reader" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8c0f2e47e00e29920959826e2e1784728a3780d1a784247be5257258cc75f910" +dependencies = [ + "memchr", +] + +[[package]] name = "scc" version = "2.1.17" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -325,6 +464,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e91b56cd4cadaeb79bbf1a5645f6b4f8dc5bde8834ad5894a8db35fda9efa1fe" [[package]] +name = "utf8parse" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" + +[[package]] name = "winapi" version = "0.3.9" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -347,6 +492,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" [[package]] +name = "windows-sys" +version = "0.52.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" +dependencies = [ + "windows-targets", +] + +[[package]] name = "windows-targets" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -7,10 +7,11 @@ edition = "2021" fs2 = "0.4.3" log = "0.4.22" priority-queue = "2.1.1" -serial_test = "3.1.1" +rev_buf_reader = "0.3.0" [lib] crate-type = ["lib"] -[package.metadata] -cargo-test-no-parallel = true +[dev-dependencies] +env_logger = "0.11.5" +serial_test = "3.1.1" @@ -1,15 +1,30 @@ #[macro_use] extern crate log; +extern crate rev_buf_reader; + pub mod log_db { use fs2::FileExt; use priority_queue::PriorityQueue; + use rev_buf_reader::RevBufReader; use std::collections::{BTreeMap, HashSet}; - use std::fs; - use std::io; - use std::io::Write; + use std::fmt::Debug; + use std::fs::{self, metadata, File}; + use std::io::{self, BufRead, Read, Seek, Write}; use std::path::{Path, PathBuf}; - const ACTIVE_LOG_FILENAME: &str = "db"; + #[cfg(unix)] + use std::os::unix::fs::MetadataExt; // For Unix-like systems + + #[cfg(windows)] + use std::os::windows::fs::MetadataExt; // For Windows + + 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]; #[derive(Debug, Clone, Ord, PartialOrd, Eq, PartialEq)] pub enum IndexableValue { @@ -42,55 +57,71 @@ pub mod log_db { } RecordValue::Int(i) => { let mut bytes = vec![1]; // Tag for Int - bytes.extend(&i.to_be_bytes()); + let data_bytes = escape_bytes(&i.to_be_bytes()); + bytes.extend(&data_bytes); bytes } RecordValue::Float(f) => { let mut bytes = vec![2]; // Tag for Float - bytes.extend(&f.to_be_bytes()); + let data_bytes = escape_bytes(&f.to_be_bytes()); + bytes.extend(&data_bytes); bytes } RecordValue::String(s) => { let mut bytes = vec![3]; // Tag for String let length = s.len() as u64; - bytes.extend(&length.to_be_bytes()); - bytes.extend(s.as_bytes()); + let length_bytes = escape_bytes(&length.to_be_bytes()); + bytes.extend(&length_bytes); + let data_bytes = escape_bytes(s.as_bytes()); + bytes.extend(&data_bytes); bytes } RecordValue::Bytes(b) => { - let mut bytes = vec![3]; // Tag for Bytes + let mut bytes = vec![4]; // Tag for Bytes let length = b.len() as u64; - bytes.extend(&length.to_be_bytes()); - bytes.extend(b); + let length_bytes = escape_bytes(&length.to_be_bytes()); + bytes.extend(&length_bytes); + let data_bytes = escape_bytes(b); + bytes.extend(&data_bytes); bytes } } } - fn deserialize(bytes: &[u8]) -> RecordValue { + /// Deserialize a RecordValue from a byte slice. + /// Returns the deserialized RecordValue and the number of bytes consumed. + fn deserialize(bytes: &[u8]) -> (RecordValue, usize) { match bytes[0] { - 0 => RecordValue::Null, + 0 => (RecordValue::Null, 1), 1 => { let mut int_bytes = [0; 8]; - int_bytes.copy_from_slice(&bytes[1..9]); - RecordValue::Int(i64::from_be_bytes(int_bytes)) + int_bytes.copy_from_slice(&bytes[1..1 + 8]); + (RecordValue::Int(i64::from_be_bytes(int_bytes)), 1 + 8) } 2 => { let mut float_bytes = [0; 8]; - float_bytes.copy_from_slice(&bytes[1..9]); - RecordValue::Float(f64::from_be_bytes(float_bytes)) + float_bytes.copy_from_slice(&bytes[1..1 + 8]); + (RecordValue::Float(f64::from_be_bytes(float_bytes)), 1 + 8) } 3 => { - let length_bytes = &bytes[1..9]; + let length_bytes = &bytes[1..1 + 8]; let length = u64::from_be_bytes(length_bytes.try_into().unwrap()) as usize; - RecordValue::String(String::from_utf8(bytes[9..9 + length].to_vec()).unwrap()) + ( + RecordValue::String( + String::from_utf8(bytes[1 + 8..1 + 8 + length].to_vec()).unwrap(), + ), + 1 + 8 + length, + ) } 4 => { - let length_bytes = &bytes[1..9]; + let length_bytes = &bytes[1..1 + 8]; let length = u64::from_be_bytes(length_bytes.try_into().unwrap()) as usize; - RecordValue::Bytes(bytes[9..9 + length].to_vec()) + ( + RecordValue::Bytes(bytes[1 + 8..1 + 8 + length].to_vec()), + 1 + 8 + length, + ) } - _ => panic!("Invalid tag"), + _ => panic!("Invalid tag: {}", bytes[0]), } } @@ -108,6 +139,27 @@ pub mod log_db { pub values: Vec<RecordValue>, } + impl Record { + pub fn serialize(&self) -> Vec<u8> { + let mut bytes = Vec::new(); + for value in &self.values { + bytes.extend(value.serialize()); + } + bytes + } + + pub fn deserialize(bytes: &[u8]) -> Record { + let mut values = Vec::new(); + let mut start = 0; + while start < bytes.len() { + let (rv, consumed) = RecordValue::deserialize(&bytes[start..]); + values.push(rv); + start += consumed; + } + Record { values } + } + } + #[derive(Clone)] pub struct Config<Field: Eq + Clone> { /// Directory where the database will store its data. @@ -115,10 +167,10 @@ pub mod log_db { /// The maximum size of a segment file in bytes. /// Once a segment file reaches this size, it is closed and a new one is created. /// Closed segment files can be compacted. - pub segment_size: u64, + pub segment_size: usize, /// The maximum size of a single memtable in bytes. /// Note that each secondary index will have its own memtable. - pub memtable_size: u64, + pub memtable_size: usize, /// The field schema of the database. pub fields: Vec<(Field, RecordFieldType)>, /// The primary key of the database, used to construct @@ -130,14 +182,14 @@ pub mod log_db { pub secondary_keys: Vec<Field>, } - pub struct DB<Field: Eq + Clone> { + pub struct DB<Field: Eq + Clone + Debug> { config: Config<Field>, log_path: PathBuf, primary_memtable: BTreeMap<IndexableValue, Record>, secondary_memtables: Vec<BTreeMap<IndexableValue, HashSet<Record>>>, } - impl<Field: Eq + Clone> DB<Field> { + impl<Field: Eq + Clone + Debug> DB<Field> { pub fn initialize(config: &Config<Field>) -> Result<DB<Field>, io::Error> { info!("Initializing DB"); // If data_dir does not exist, create it @@ -197,6 +249,7 @@ pub mod log_db { } pub fn upsert(&mut self, record: &Record) -> Result<(), io::Error> { + debug!("Upserting record: {:?}", record); // Validate the record length if record.values.len() != self.config.fields.len() { return Err(io::Error::new( @@ -222,13 +275,16 @@ pub mod log_db { io::ErrorKind::InvalidInput, format!( "Record field {} has incorrect type: {:?}, expected {:?}", - i, record.values[i], field_type + &i, &record.values[i], &field_type ), )) } } } + debug!("Record is valid"); + debug!("Opening file in append mode and acquiring exclusive lock..."); + // Open the log file in append mode let mut file = fs::OpenOptions::new() .create(true) @@ -238,15 +294,21 @@ pub mod log_db { // Acquire an exclusive lock for writing file.lock_exclusive()?; + if !is_file_same_as_path(&file, &self.log_path)? { + // The log file has been rotated, so we must try again + debug!("Lock acquired, but the log file has been rotated. Retrying upsert..."); + file.unlock()?; + drop(file); + return self.upsert(record); + } + + debug!("Lock acquired, appending to log file"); + // Write the record to the log - // Each serialized row is suffixed with the field separator character - let mut serialized = record - .values - .iter() - .flat_map(|field| field.serialize()) - .collect::<Vec<u8>>(); - serialized.extend(vec![b'\x1C']); - file.write_all(&serialized)?; + // Each serialized row is suffixed with the field separator character sequence + let mut serialized_record = record.serialize(); + serialized_record.extend(SEQ_RECORD_SEP); + file.write_all(&serialized_record)?; // Sync to disk file.flush()?; @@ -254,6 +316,9 @@ pub mod log_db { file.unlock()?; + debug!("Record appended to log file, lock released"); + debug!("Updating memtables"); + // Update the primary memtable let primary_key_index = self .config @@ -272,51 +337,262 @@ pub mod log_db { "Primary key must be an IndexableValue", ))?; - self.primary_memtable - .insert(primary_value.clone(), record.clone()); + if self.primary_memtable.len() < self.config.memtable_size { + // TODO: handle capacity better, remove oldest records + debug!( + "Inserting record into primary memtable with key {:?} = {:?}", + &self.config.primary_key, &primary_value, + ); + self.primary_memtable + .insert(primary_value.clone(), record.clone()); + } else { + debug!("Primary memtable is full, not inserting"); + } // TODO: Update secondary memtables Ok(()) } - pub fn get(&self, field: Field, key: RecordValue) -> Result<Option<Record>, io::Error> { - // If the requested field is the primary key, look up the value in the primary memtable - if field == self.config.primary_key { - let primary_value = key.as_indexable().ok_or(io::Error::new( - io::ErrorKind::InvalidInput, - "Primary key must be an IndexableValue", - ))?; + pub fn get( + &mut self, + field: &Field, + query_key: &RecordValue, + ) -> Result<Option<Record>, io::Error> { + let query_key_original = query_key; + debug!("Getting record with field {:?} = {:?}", field, query_key); + let query_key = query_key_original.as_indexable().ok_or(io::Error::new( + io::ErrorKind::InvalidInput, + "Queried value must be indexable", + ))?; - let found = self.primary_memtable.get(&primary_value); + // If the requested field is the primary key, look up the value in the primary memtable + if *field == self.config.primary_key { + debug!("Looking up key {:?} in primary memtable", query_key); + let found = self.primary_memtable.get(&query_key); if let Some(record) = found { + debug!("Found record in primary memtable: {:?}", record); return Ok(Some(record.clone())); } } // TODO: query secondary memtables + debug!( + "No memtable entry found, looking up key {:?} in log file", + query_key + ); + // Get the index of the requested field let key_index = self .config .fields .iter() - .position(|(schema_field, _)| schema_field == &field) + .position(|(schema_field, _)| schema_field == field) .ok_or(io::Error::new( io::ErrorKind::InvalidInput, "Key not found in schema after initialize", ))?; + debug!("Matching key index {}", key_index); + debug!("Opening file in read mode and acquiring shared lock..."); + // Open the file and acquire a shared lock for reading - let file = fs::OpenOptions::new().read(true).open(&self.log_path)?; + let mut file = fs::OpenOptions::new().read(true).open(&self.log_path)?; file.lock_shared()?; - // Do a log file scan from the bottom up to find the record - // TODO + if !is_file_same_as_path(&file, &self.log_path)? { + // The log file has been rotated, so we must try again + debug!("Lock acquired, but the log file has been rotated. Retrying get..."); + file.unlock()?; + drop(file); + return self.get(field, query_key_original); + } + + debug!("Lock acquired, searching log file for record"); + + let mut log_reader = LogReader::new(&mut file)?; + let result = log_reader.find(|record| { + let record_key = record.values[key_index].as_indexable().unwrap(); + record_key == query_key + }); file.unlock()?; + debug!("Record search complete, lock released"); + + let result_value = match result { + Some(record) => record, + None => { + debug!("No record found for key {:?}", query_key); + return Ok(None); + } + }; + + debug!("Found matching record in log file"); + + if self.primary_memtable.len() < self.config.memtable_size { + if *field == self.config.primary_key { + // TODO: handle capacity better, remove oldest records + debug!( + "Inserting record into primary memtable with key {:?} = {:?}", + &field, &query_key, + ); + self.primary_memtable + .insert(query_key.clone(), result_value.clone()); + } + } else { + debug!("Primary memtable is full, not inserting"); + } + + Ok(Some(result_value)) + } + } + + #[derive(Debug, Eq, PartialEq)] + enum SpecialSequence { + RecordSeparator, + LiteralFieldSeparator, + LiteralEscape, + } + + pub struct LogReader<'a> { + rev_reader: RevBufReader<&'a mut fs::File>, + } + + impl<'a> LogReader<'a> { + pub fn new(file: &mut fs::File) -> Result<LogReader, io::Error> { + let rev_reader = RevBufReader::new(file); + Ok(LogReader { rev_reader }) + } + + fn read_record(&mut self) -> Result<Option<Record>, io::Error> { + if self.rev_reader.stream_position()? == 0 { + return Ok(None); + } + + // Check that the record starts with the record separator + if self.read_special_sequence()? != SpecialSequence::RecordSeparator { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "Record candidate does not end with record separator", + )); + } + + // The buffer that stores all the bytes of the record read so far in reverse order. + let mut result_buf: Vec<u8> = Vec::new(); + // The buffer that stores the bytes read from the file. + let mut read_buf: Vec<u8> = Vec::new(); + + loop { + read_buf.clear(); + self.rev_reader + .read_until(ESCAPE_CHARACTER, &mut read_buf)?; + + result_buf.extend(read_buf.iter().rev()); + + if self.rev_reader.stream_position()? == 0 { + // If we've reached the beginning of the file, we've read the entire record. + break; + } + + // 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. + // Move the cursor back to the beginning of the special sequence. + self.rev_reader.seek_relative(3)?; + 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); + } + } + } + + result_buf.reverse(); + 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; 3]; + self.rev_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 LogReader<'_> { + 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), + } + } + } + + /// 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, + } + } + + fn escape_bytes(buf: &[u8]) -> Vec<u8> { + let mut result = Vec::new(); + for byte in buf { + match byte { + &FIELD_SEPARATOR => { + result.extend(SEQ_LIT_FIELD_SEP); + } + &ESCAPE_CHARACTER => { + result.extend(SEQ_LIT_ESCAPE); + } + _ => result.push(*byte), + } + } + result + } + + fn is_file_same_as_path(file: &File, path: &PathBuf) -> io::Result<bool> { + // Get the metadata for the open file handle + let file_metadata = file.metadata()?; + + // Get the metadata for the file at the specified path + let path_metadata = metadata(path)?; + + // Platform-specific comparison + #[cfg(unix)] + { + Ok(file_metadata.dev() == path_metadata.dev() + && file_metadata.ino() == path_metadata.ino()) + } - Ok(None) + #[cfg(windows)] + { + Ok(file_metadata.file_index() == path_metadata.file_index() + && file_metadata.volume_serial_number() == path_metadata.volume_serial_number()) } } } diff --git a/tests/integration.rs b/tests/integration.rs index 2c0a958..7fb3229 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -1,11 +1,16 @@ -use log_db::log_db::{Config, Record, RecordFieldType, RecordValue, DB}; +use env_logger; +use log_db; +use log_db::log_db::{Config, LogReader, Record, RecordFieldType, RecordValue, DB}; use serial_test::serial; +use std::fs; +use std::path::Path; const TEST_DATA_DIR: &str = "test_db_data"; -const TEST_SEGMENT_SIZE: u64 = 1024 * 1024; // 1 MB -const TEST_MEMTABLE_SIZE: u64 = 1024 * 1024; // 1 MB +const TEST_RESOURCES_DIR: &str = "tests/resources"; +const TEST_SEGMENT_SIZE: usize = 1024 * 1024; // 1 MB +const TEST_MEMTABLE_SIZE: usize = 1024 * 1024; // 1 MB -#[derive(Eq, PartialEq, Clone)] +#[derive(Eq, PartialEq, Clone, Debug)] enum Field { Id, Name, @@ -35,7 +40,7 @@ fn test_initialize() { #[test] #[serial] -fn test_upsert_to_empty_db() { +fn test_upsert_and_get_with_memtable() { let mut db = DB::initialize(&Config { data_dir: TEST_DATA_DIR.to_string(), segment_size: TEST_SEGMENT_SIZE, @@ -59,7 +64,7 @@ fn test_upsert_to_empty_db() { }; db.upsert(&record).unwrap(); - let result = db.get(Field::Id, RecordValue::Int(1)).unwrap().unwrap(); + let result = db.get(&Field::Id, &RecordValue::Int(1)).unwrap().unwrap(); // Check that the IDs match assert!(match (&result.values[0], &record.values[0]) { @@ -73,6 +78,91 @@ fn test_upsert_to_empty_db() { #[test] #[serial] +fn test_upsert_and_get_without_memtable() { + let _ = env_logger::builder().is_test(true).try_init(); + let mut db = DB::initialize(&Config { + data_dir: TEST_DATA_DIR.to_string(), + segment_size: TEST_SEGMENT_SIZE, + memtable_size: 0, // Force log reads + fields: vec![ + (Field::Id, RecordFieldType::Int), + (Field::Name, RecordFieldType::String), + (Field::Data, RecordFieldType::Bytes), + ], + primary_key: Field::Id, + secondary_keys: vec![], + }) + .unwrap(); + + // Insert some records + let record0 = Record { + values: vec![ + RecordValue::Int(0), + RecordValue::String("John".to_string()), + RecordValue::Bytes(vec![3, 4, 5]), + ], + }; + db.upsert(&record0).unwrap(); + + let record1 = Record { + values: vec![ + RecordValue::Int(1), + RecordValue::String("Alice".to_string()), + RecordValue::Bytes(vec![0, 1, 2]), + ], + }; + db.upsert(&record1).unwrap(); + + let record2 = Record { + values: vec![ + RecordValue::Int(1), + RecordValue::String("Bob".to_string()), + RecordValue::Bytes(vec![0, 1, 2]), + ], + }; + db.upsert(&record2).unwrap(); + + let record3 = Record { + values: vec![ + RecordValue::Int(2), + RecordValue::String("George".to_string()), + RecordValue::Bytes(vec![]), + ], + }; + db.upsert(&record3).unwrap(); + + // Get with ID = 0 + let result = db.get(&Field::Id, &RecordValue::Int(0)).unwrap().unwrap(); + + // Should match record0 + assert!(match (&result.values[0], &record0.values[0]) { + (RecordValue::Int(a), RecordValue::Int(b)) => a == b, + _ => false, + }); + assert!(match (&result.values[1], &record0.values[1]) { + (RecordValue::String(a), RecordValue::String(b)) => a == b, + _ => false, + }); + + // Get with ID = 1 + let result = db.get(&Field::Id, &RecordValue::Int(1)).unwrap().unwrap(); + + // Should match record2 + assert!(match (&result.values[0], &record2.values[0]) { + (RecordValue::Int(a), RecordValue::Int(b)) => a == b, + _ => false, + }); + assert!(match (&result.values[1], &record2.values[1]) { + (RecordValue::String(a), RecordValue::String(b)) => a == b, + _ => false, + }); + + // Clean up + std::fs::remove_dir_all(TEST_DATA_DIR.to_string()).unwrap(); +} + +#[test] +#[serial] fn test_upsert_fails_on_invalid_number_of_values() { let mut db = DB::initialize(&Config { data_dir: TEST_DATA_DIR.to_string(), @@ -130,3 +220,30 @@ fn test_upsert_fails_on_invalid_value_type() { // Clean up std::fs::remove_dir_all(TEST_DATA_DIR.to_string()).unwrap(); } + +#[test] +#[serial] +fn test_log_reader_fixture_db1() { + let _ = env_logger::builder().is_test(true).try_init(); + 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 log_reader = LogReader::new(&mut file).unwrap(); + + // There are two records in the log with "schema": Int, Null + + let last_record = 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 = 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, + }); +} diff --git a/tests/resources/test_db1 b/tests/resources/test_db1 Binary files differnew file mode 100644 index 0000000..e75edf5 --- /dev/null +++ b/tests/resources/test_db1 |