aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--log_db/benches/benchmark.rs57
-rw-r--r--log_db/src/common.rs35
-rw-r--r--log_db/src/lib.rs324
-rw-r--r--log_db/src/log_reader_forward.rs18
-rw-r--r--log_db/src/log_reader_reverse.rs366
-rw-r--r--log_db/src/memtable_secondary.rs1
-rw-r--r--log_db/tests/integration.rs182
-rw-r--r--log_db/tests/resources/test_data_1bin255 -> 265 bytes
-rw-r--r--log_db/tests/resources/test_db1bin31 -> 0 bytes
-rw-r--r--log_db/tests/resources/test_db2bin46243 -> 0 bytes
-rw-r--r--log_db/tests/resources/test_db3bin46243 -> 0 bytes
-rw-r--r--log_db/tests/resources/test_metadata_1bin40 -> 40 bytes
-rw-r--r--py_bindings/src/lib.rs1
13 files changed, 304 insertions, 680 deletions
diff --git a/log_db/benches/benchmark.rs b/log_db/benches/benchmark.rs
index 0dcf953..353334a 100644
--- a/log_db/benches/benchmark.rs
+++ b/log_db/benches/benchmark.rs
@@ -2,8 +2,6 @@ mod utils;
use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion};
use log_db::*;
-use std::fs::OpenOptions;
-use std::path::Path;
use tempfile;
use utils::*;
@@ -56,7 +54,7 @@ pub fn upsert_various_initial_sizes_compacted(c: &mut Criterion) {
.expect("Failed to convert tmpdir path to str");
let sample_record = random_record(0, 1);
- let record_length = sample_record.serialize().len() + SEQ_RECORD_SEP.len();
+ let record_length = sample_record.serialize().len();
let mut db = DB::configure()
.data_dir(&data_dir)
@@ -231,58 +229,6 @@ pub fn get_various_memtable_capacities(c: &mut Criterion) {
}
}
-fn reverse_read_file_with_various_buffer_sizes(c: &mut Criterion) {
- let mut group = c.benchmark_group("reverse_read_file_with_various_buffer_sizes");
- group.sample_size(50);
-
- // odd powers of 2
- let buffer_sizes = [128, 512, 2048, 8192, 32768, 131_072, 524_288];
- const PREFILL_N: usize = 100_000;
-
- let data_dir_obj = tempfile::tempdir().expect("Failed to get tmpdir");
- let data_dir = &data_dir_obj
- .path()
- .to_str()
- .expect("Failed to convert tmpdir path to str");
-
- // Create a db instance for prefilling
- let mut db = DB::configure()
- .data_dir(&data_dir)
- .fields(vec![
- (Field::Id, RecordField::int()),
- (Field::Name, RecordField::string()),
- (Field::Data, RecordField::bytes()),
- ])
- .primary_key(Field::Id)
- .initialize()
- .expect("Failed to initialize DB");
-
- prefill_db(&mut db, PREFILL_N, false).expect("Failed to prefill DB");
- drop(db);
-
- for size in buffer_sizes {
- group.bench_with_input(BenchmarkId::from_parameter(size), &size, |b, &_size| {
- let mut file = OpenOptions::new()
- .read(true)
- .open(Path::new(data_dir).join("db"))
- .expect("Failed to open log file");
-
- b.iter(|| {
- let mut rev_reader = ReverseLogReader::new_with_size(&mut file, size)
- .expect("Failed to create ReverseLogReader");
-
- // This is to avoid optimizing out the loop
- let mut i = 0;
- for _ in &mut rev_reader {
- i += 1;
- }
-
- i
- });
- });
- }
-}
-
// Register the benchmark group
criterion_group!(
benches,
@@ -292,6 +238,5 @@ criterion_group!(
get_from_disk_various_initial_sizes,
get_from_disk_various_initial_sizes_compacted,
get_various_memtable_capacities,
- reverse_read_file_with_various_buffer_sizes,
);
criterion_main!(benches);
diff --git a/log_db/src/common.rs b/log_db/src/common.rs
index 448226d..ce1a30f 100644
--- a/log_db/src/common.rs
+++ b/log_db/src/common.rs
@@ -17,6 +17,7 @@ use std::os::windows::fs::MetadataExt;
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 INIT_LOCK_FILENAME: &str = "init_lock";
pub const DEFAULT_READ_BUF_SIZE: usize = 1024 * 1024; // 1 MB
pub const TEST_RESOURCES_DIR: &str = "tests/resources";
@@ -216,32 +217,26 @@ impl RecordValue {
}
RecordValue::Int(i) => {
let mut bytes = vec![1]; // Tag for Int
- let data_bytes = escape_bytes(&i.to_be_bytes());
- bytes.extend(&data_bytes);
+ bytes.extend(&i.to_be_bytes());
bytes
}
RecordValue::Float(f) => {
let mut bytes = vec![2]; // Tag for Float
- let data_bytes = escape_bytes(&f.to_be_bytes());
- bytes.extend(&data_bytes);
+ bytes.extend(&f.to_be_bytes());
bytes
}
RecordValue::String(s) => {
let mut bytes = vec![3]; // Tag for String
let length = s.len() as u64;
- 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.extend(&length.to_be_bytes());
+ bytes.extend(s.as_bytes());
bytes
}
RecordValue::Bytes(b) => {
let mut bytes = vec![4]; // Tag for Bytes
let length = b.len() as u64;
- let length_bytes = escape_bytes(&length.to_be_bytes());
- bytes.extend(&length_bytes);
- let data_bytes = escape_bytes(b);
- bytes.extend(&data_bytes);
+ bytes.extend(&length.to_be_bytes());
+ bytes.extend(b);
bytes
}
}
@@ -319,22 +314,6 @@ impl Record {
}
}
-pub 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
-}
-
/// A path to a log segment file along with its type
pub enum SegmentPath {
/// A symbolic link to the active log file
diff --git a/log_db/src/lib.rs b/log_db/src/lib.rs
index b0de03f..320ec6c 100644
--- a/log_db/src/lib.rs
+++ b/log_db/src/lib.rs
@@ -14,7 +14,6 @@ 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};
use std::io::Seek;
@@ -23,9 +22,6 @@ use std::io::{self, Read, Write};
use std::os::unix::fs::MetadataExt;
use std::path::{Path, PathBuf};
use std::thread;
-use tempfile;
-use tempfile::tempfile_in;
-use tempfile::NamedTempFile;
use uuid::Uuid;
pub struct ConfigBuilder<Field: Eq + Clone + Debug> {
@@ -162,26 +158,43 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
// After creation, the directory should always be in a complete state
// without missing files.
// A tempdir-move strategy is used to achieve one-phase commit.
- if !fs::exists(&config.data_dir)?
- || !fs::exists(&Path::new(&config.data_dir).join(ACTIVE_SYMLINK_FILENAME))?
- {
- let tmpdir = tempfile::tempdir()?;
- let tmpdir_path = tmpdir.into_path();
- let (segment_uuid, _) = DB::<Field>::create_segment_data_file(&tmpdir_path)?;
+ // Ensure the data directory exists
+ let data_dir_path = Path::new(&config.data_dir);
+ match fs::create_dir(&data_dir_path) {
+ Ok(_) => {}
+ Err(e) => {
+ if e.kind() != io::ErrorKind::AlreadyExists {
+ return Err(e);
+ }
+ }
+ }
+
+ // Create an initialize lock file to prevent multiple concurrent initializations
+ let init_lock_file = fs::OpenOptions::new()
+ .create(true)
+ .write(true)
+ .open(&data_dir_path.join(INIT_LOCK_FILENAME))?;
+
+ init_lock_file.lock_exclusive()?;
+
+ // We have acquired the lock, check if the data directory is in a complete state
+ // If not, initialize it, otherwise skip.
+ if !fs::exists(data_dir_path.join(ACTIVE_SYMLINK_FILENAME))? {
+ let (segment_uuid, _) = DB::<Field>::create_segment_data_file(data_dir_path)?;
let (segment_num, _) =
- DB::<Field>::create_segment_metadata_file(&tmpdir_path, &segment_uuid)?;
- DB::<Field>::set_active_segment(&tmpdir_path, segment_num)?;
+ DB::<Field>::create_segment_metadata_file(data_dir_path, &segment_uuid)?;
+ DB::<Field>::set_active_segment(data_dir_path, segment_num)?;
// Create the exclusive lock request file
fs::OpenOptions::new()
.create(true)
.write(true)
- .open(&tmpdir_path.join(EXCL_LOCK_REQUEST_FILENAME))?;
-
- fs::rename(tmpdir_path, &config.data_dir)?;
+ .open(data_dir_path.join(EXCL_LOCK_REQUEST_FILENAME))?;
}
+ init_lock_file.unlock()?;
+
// Calculate the index of the primary value in a record
let primary_key_index = config
.fields
@@ -238,7 +251,6 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
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)?;
@@ -322,9 +334,9 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
debug!("Opening file in append mode and acquiring exclusive lock...");
// Acquire an exclusive lock for writing
- self.request_exclusive_lock()?;
+ self.request_exclusive_lock_on_active()?;
- if self.ensure_correct_file_is_open()? {
+ if self.ensure_active_file_is_open()? {
// The log file has been rotated, so we must try again
return self.upsert(record);
}
@@ -332,18 +344,37 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
debug!("Lock acquired, appending to log file");
// Write the record to the log
- self.log_file.write_all(&record.serialize())?;
+ let serialized = &record.serialize();
+ let record_offset = self.active_data_file.metadata()?.len();
+ let record_length = serialized.len() as u64;
+ self.active_data_file.write_all(serialized)?;
- // Flush and sync to disk
+ // Flush and sync data to disk
if self.config.write_durability == WriteDurability::Flush {
- self.log_file.flush()?;
+ self.active_data_file.flush()?;
}
if self.config.write_durability == WriteDurability::FlushSync {
- self.log_file.flush()?;
- self.log_file.sync_all()?;
+ self.active_data_file.flush()?;
+ self.active_data_file.sync_all()?;
}
- self.log_file.unlock()?;
+ // Write the record metadata to the metadata file
+ let mut metadata_buf = vec![];
+ metadata_buf.extend(&record_offset.to_be_bytes());
+ metadata_buf.extend(&record_length.to_be_bytes());
+ self.active_metadata_file.write_all(&metadata_buf)?;
+
+ // Flush and sync metadata to disk
+ if self.config.write_durability == WriteDurability::Flush {
+ self.active_metadata_file.flush()?;
+ }
+ if self.config.write_durability == WriteDurability::FlushSync {
+ self.active_metadata_file.flush()?;
+ self.active_metadata_file.sync_all()?;
+ }
+
+ self.active_data_file.unlock()?;
+ self.active_metadata_file.unlock()?;
debug!("Record appended to log file, lock released");
@@ -382,65 +413,48 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
"Matching records based on value at primary key index ({})",
&self.primary_key_index
);
- debug!("Opening file in read mode and acquiring shared lock...");
- // Open the file and acquire a shared lock for reading
- let mut file = fs::OpenOptions::new().read(true).open(&self.log_path)?;
+ let data_dir_path = Path::new(&self.config.data_dir);
+ let greatest = DB::<Field>::greatest_segment_number(&data_dir_path)?;
+ debug!("Searching segments {} through 1", greatest);
- self.request_shared_lock(&mut file)?;
+ let mut found_record: Option<Record> = None;
+ for segment_num in (1..=greatest).rev() {
+ let segment_path = data_dir_path.join(format!("metadata.{}", segment_num));
- 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(query_key_original);
- }
+ debug!(
+ "Opening segment {} in read mode and acquiring shared lock...",
+ segment_num
+ );
- debug!("Lock acquired, searching log files for record");
+ let mut metadata_file = fs::OpenOptions::new().read(true).open(&segment_path)?;
- let greatest = DB::<Field>::greatest_segment_number(&Path::new(&self.config.data_dir))?;
- let segment_numbers = (0..=greatest); // FIXME all of this needs fixing
- let mut result: Option<Record> = None;
- for n in segment_numbers {
- if n == 0 {
- debug!("Searching the active log file...");
- result = ReverseLogReader::new(&mut file)?.find(|record| {
- let record_key = record.values[self.primary_key_index]
- .as_indexable()
- .expect("A non-indexable value was stored at key index");
- record_key == query_key
- });
+ self.request_shared_lock(&mut metadata_file)?;
- debug!("Active log file searched, releasing shared lock...");
- file.unlock()?;
- } else {
- debug!("Locking and searching rotated log segment file {}...", n);
- let path = Path::new(&self.config.data_dir)
- .join(ACTIVE_SYMLINK_FILENAME)
- .with_extension(n.to_string());
+ let metadata_header = DB::<Field>::read_metadata_header(&mut metadata_file)?;
+ let data_path = data_dir_path.join(metadata_header.uuid.to_string());
+ let data_file = fs::OpenOptions::new().read(true).open(&data_path)?;
- let mut segm_file = fs::OpenOptions::new().read(true).open(&path)?;
- self.request_shared_lock(&mut segm_file)?;
- result = ReverseLogReader::new(&mut segm_file)?.find(|record| {
- let record_key = record.values[self.primary_key_index]
- .as_indexable()
- .expect("A non-indexable value was stored at key index");
- record_key == query_key
- });
+ // We should not "request_shared_lock()" here because we do not want
+ // to give way to writers at this point. That would possibly lead to a deadlock.
+ data_file.lock_shared()?;
- debug!("Segment file searched, releasing shared lock...");
- segm_file.unlock()?;
- };
+ let mut reader = ReverseLogReader::new(metadata_file, data_file)?;
- if result.is_some() {
+ if let Some(found) = reader.find(|record| {
+ let record_key = record.values[self.primary_key_index]
+ .as_indexable()
+ .expect("Primary key must be indexable");
+ record_key == query_key
+ }) {
+ found_record = Some(found);
break;
}
}
debug!("Record search complete");
- let result_value = match &result {
+ let result_value = match &found_record {
Some(record) => record,
None => {
debug!("No record found for key {:?}", query_key);
@@ -453,7 +467,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
debug!("Updating memtables");
self.insert_to_memtables(&result_value);
- Ok(result)
+ Ok(found_record)
}
/// Get a collection of records based on a field value.
@@ -512,39 +526,52 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
"Key not found in schema after initialize",
))?;
- debug!("Matching key index {}", key_index);
- debug!("Acquiring shared lock...");
+ let data_dir_path = Path::new(&self.config.data_dir);
+ let greatest = DB::<Field>::greatest_segment_number(&data_dir_path)?;
- // Acquire a shared lock for reading
- self.log_file.lock_shared()?;
+ let mut found_records = vec![];
+ for segment_num in (1..=greatest).rev() {
+ let segment_path = data_dir_path.join(format!("metadata.{}", segment_num));
- if self.ensure_correct_file_is_open()? {
- // The log file has been rotated, so we must try again
- return self.find_all(field, query_key_original);
- }
+ debug!(
+ "Opening segment {} in read mode and acquiring shared lock...",
+ segment_num
+ );
+
+ let mut metadata_file = fs::OpenOptions::new().read(true).open(&segment_path)?;
- debug!("Lock acquired, searching log file for record");
+ self.request_shared_lock(&mut metadata_file)?;
- let result = ReverseLogReader::new(&mut self.log_file)?
- .filter(|record| {
+ let metadata_header = DB::<Field>::read_metadata_header(&mut metadata_file)?;
+ let data_path = &data_dir_path.join(metadata_header.uuid.to_string());
+ let data_file = fs::OpenOptions::new().read(true).open(&data_path)?;
+
+ // We should not "request_shared_lock()" here because we do not want
+ // to give way to writers at this point. That would possibly lead to a deadlock.
+ data_file.lock_shared()?;
+
+ let reader = ReverseLogReader::new(metadata_file, data_file)?;
+
+ for record in reader {
let record_key = record.values[key_index]
.as_indexable()
- .expect("A non-indexable value was stored at key index");
- record_key == query_key
- })
- .collect::<Vec<Record>>();
+ .expect("Secondary key must be indexable");
+ if record_key == query_key {
+ found_records.push(record);
+ }
+ }
+ }
- self.log_file.unlock()?;
- debug!("Record search complete, lock released");
+ debug!("Record search complete");
debug!(
"Number of matching records found in log file: {}",
- result.len()
+ found_records.len()
);
if let Some(memtable_index) = found_memtable_index {
debug!("Inserting result set into secondary index");
- let primary_values: Vec<IndexableValue> = result
+ let primary_values: Vec<IndexableValue> = found_records
.iter()
.map(|r| {
r.values[self.primary_key_index]
@@ -555,7 +582,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
self.secondary_memtables[memtable_index].set_all(&query_key, &primary_values);
}
- Ok(result)
+ Ok(found_records)
}
fn get_secondary_memtable_index_by_field(&self, field: &Field) -> Option<usize> {
@@ -565,26 +592,34 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
.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.
+ /// Ensures that the `self.metadata_file` and `self.data_file` handles are still pointing to the correct files.
+ /// If the segment has been rotated, the handle will be closed and reopened.
/// Returns `true` if the file has been rotated and the handle has been reopened.
- fn ensure_correct_file_is_open(&mut self) -> Result<bool, io::Error> {
- if !is_file_same_as_path(&self.log_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. Reopening file and retrying..."
- );
- self.log_file.unlock()?;
+ fn ensure_active_file_is_open(&mut self) -> Result<bool, io::Error> {
+ let data_dir_path = Path::new(&self.config.data_dir);
+ let active_target = fs::read_link(data_dir_path.join("active"))?;
+ let active_metadata_path = data_dir_path.join(active_target);
- self.log_file = fs::OpenOptions::new()
- .create(true)
+ let correct = is_file_same_as_path(&self.active_metadata_file, &active_metadata_path)?;
+ if !correct {
+ debug!("Metadata file has been rotated. Reopening...");
+ let mut metadata_file = fs::OpenOptions::new()
.read(true)
- .append(true)
- .open(&self.log_path)?;
+ .write(true)
+ .open(&active_metadata_path)?;
+
+ self.request_shared_lock(&mut metadata_file)?;
+
+ let metadata_header =
+ DB::<Field>::read_metadata_header(&mut self.active_metadata_file)?;
+ let data_file_path = data_dir_path.join(metadata_header.uuid.to_string());
- Ok(true)
+ self.active_metadata_file = metadata_file;
+ self.active_data_file = fs::OpenOptions::new().append(true).open(&data_file_path)?;
+
+ return Ok(true);
} else {
- Ok(false)
+ return Ok(false);
}
}
@@ -619,7 +654,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
}
}
- fn request_exclusive_lock(&mut self) -> Result<(), io::Error> {
+ fn request_exclusive_lock_on_active(&mut self) -> Result<(), io::Error> {
// Create a lock on the exclusive lock request file to signal to readers that they should wait
let lock_request_path = Path::new(&self.config.data_dir).join(EXCL_LOCK_REQUEST_FILENAME);
let lock_request_file = fs::OpenOptions::new()
@@ -641,8 +676,9 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
// ));
// }
- // Acquire an exclusive lock on the log file
- self.log_file.lock_exclusive()?;
+ // Acquire an exclusive lock on the segment files
+ self.active_metadata_file.lock_exclusive()?;
+ self.active_data_file.lock_exclusive()?;
// Unlock the request file
lock_request_file.unlock()?;
@@ -705,38 +741,32 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
/// 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_SYMLINK_FILENAME);
+ let data_dir = self.config.data_dir.to_owned();
+ let data_dir_path = Path::new(&data_dir);
+ let active_log_path = data_dir_path.join(ACTIVE_SYMLINK_FILENAME);
let active_log_md = fs::metadata(&active_log_path)?;
if active_log_md.size() >= self.config.segment_size as u64 {
// Rotate the active log file
debug!("Starting rotation, requesting exclusive lock...");
- self.request_exclusive_lock()?;
+ self.request_exclusive_lock_on_active()?;
debug!("Exclusive lock acquired, rotating active log file...");
- let next_segment_number =
- DB::<Field>::greatest_segment_number(&Path::new(&self.config.data_dir))? + 1;
- let next_segment_path =
- &active_log_path.with_extension(next_segment_number.to_string());
-
- debug!("Renaming active log file to {:?}", &next_segment_path);
- fs::rename(&active_log_path, &next_segment_path)?;
- // Create a new active log file
- self.log_file = fs::OpenOptions::new()
- .create(true)
- .write(true)
- .append(true)
- .open(&active_log_path)?;
+ // Create a new active log segment
+ let (data_file_uuid, _) = DB::<Field>::create_segment_data_file(data_dir_path)?;
+ let (new_segment_num, new_segment_path) =
+ DB::<Field>::create_segment_metadata_file(data_dir_path, &data_file_uuid)?;
+ DB::<Field>::set_active_segment(data_dir_path, new_segment_num)?;
// The new active log file is not locked by this client so it cannot be touched.
- debug!("Active log file rotated");
+ debug!("Active log file rotated, new segment: {}", new_segment_num);
// Compact the rotated segment without a lock.
// Since the rotated segment and the compacted segment based on it will be
// a) read-only, and b) identical in effective content, there is no need to lock it.
- self.compact_segment(&next_segment_path)?;
+ self.compact_segment(&new_segment_path)?;
debug!("Segment compacted");
}
@@ -745,34 +775,36 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
}
fn compact_segment(&self, path: &Path) -> Result<(), io::Error> {
- debug!("Opening segment file {:?} for compaction", path);
- let mut segment_file = fs::OpenOptions::new().read(true).open(path)?;
+ warn!("TODO compact_segment");
+ // debug!("Opening segment file {:?} for compaction", path);
+ // let mut segment_file = fs::OpenOptions::new().read(true).open(path)?;
- debug!("Reading segment data into a BTreeMap");
- let mut map = BTreeMap::new();
- let forward_log_reader = ForwardLogReader::new(&mut segment_file);
- for entry in forward_log_reader {
- let primary_key = entry.values[self.primary_key_index]
- .as_indexable()
- .expect("Primary key was not indexable");
- map.insert(primary_key, entry);
- }
+ // debug!("Reading segment data into a BTreeMap");
+ // let mut map = BTreeMap::new();
+ // let forward_log_reader = ForwardLogReader::new(&mut segment_file);
+ // for entry in forward_log_reader {
+ // let primary_key = entry.values[self.primary_key_index]
+ // .as_indexable()
+ // .expect("Primary key was not indexable");
+ // map.insert(primary_key, entry);
+ // }
- debug!("Opening temporary file for writing compacted data");
- let temp_file = tempfile::NamedTempFile::new()?;
- let temp_path = temp_file.as_ref();
+ // debug!("Opening temporary file for writing compacted data");
+ // let temp_file = tempfile::NamedTempFile::new()?;
+ // let temp_path = temp_file.as_ref();
- let mut temp_file = fs::OpenOptions::new()
- .create(true)
- .append(true)
- .open(temp_path)?;
+ // let mut temp_file = fs::OpenOptions::new()
+ // .create(true)
+ // .append(true)
+ // .open(temp_path)?;
- for entry in map.values() {
- temp_file.write_all(&entry.serialize())?;
- }
+ // for entry in map.values() {
+ // temp_file.write_all(&entry.serialize())?;
+ // }
- fs::rename(&temp_path, path)?;
+ // fs::rename(&temp_path, path)?;
+ // Ok(())
Ok(())
}
diff --git a/log_db/src/log_reader_forward.rs b/log_db/src/log_reader_forward.rs
index 808d3fa..1b34dfc 100644
--- a/log_db/src/log_reader_forward.rs
+++ b/log_db/src/log_reader_forward.rs
@@ -31,13 +31,26 @@ impl<'a> ForwardLogReader {
}
}
+ debug!("Read 16 bytes from metadata file: {:?}", metadata_entry_buf);
+
// 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());
- self.data_reader.seek(io::SeekFrom::Start(entry_offset))?;
+ debug!(
+ "Read offset {} and length {} from metadata file",
+ entry_offset, entry_length
+ );
+
+ // Use .seek_relative instead of .seek to avoid dropping the BufReader internal buffer when
+ // the seek distance is small
+ let seek_distance = entry_offset - self.data_reader.stream_position()?;
+ self.data_reader.seek_relative(seek_distance as i64)?;
+
let mut result_buf = vec![0; entry_length as usize];
+ debug!("Reading {} bytes from data file", entry_length);
self.data_reader.read_exact(&mut result_buf)?;
+ debug!("Read {} bytes from data file", result_buf.len());
let record = Record::deserialize(&result_buf);
Ok(Some(record))
@@ -63,6 +76,7 @@ mod tests {
#[test]
fn test_forward_log_reader_fixture_db1() {
+ let _ = env_logger::builder().is_test(true).try_init();
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()
@@ -82,7 +96,7 @@ mod tests {
.next()
.expect("Failed to read the first record");
assert!(match first_record.values.as_slice() {
- [RecordValue::Bytes(_)] => true,
+ [RecordValue::Bytes(bytes)] => bytes.len() == 256,
_ => false,
});
diff --git a/log_db/src/log_reader_reverse.rs b/log_db/src/log_reader_reverse.rs
index f042406..ce17e59 100644
--- a/log_db/src/log_reader_reverse.rs
+++ b/log_db/src/log_reader_reverse.rs
@@ -1,345 +1,109 @@
use super::common::*;
use std::fs::{self};
-use std::io::{self, Read, Seek, SeekFrom};
+use std::io::{self, Read, Seek};
-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<u8>,
- /// 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<ReverseLogReader, io::Error> {
- 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<ReverseLogReader, io::Error> {
- 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<Option<Record>, 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<u8> = 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)?;
+pub struct ReverseLogReader {
+ /// The contents of the metadata file are read into this buffer in one go.
+ metadata_buf: Vec<u8>,
- result_buf.extend(&read_buf[..read]);
+ /// The data log file that is read based on offset + length information from the metadata file.
+ data_reader: io::BufReader<fs::File>,
- if self.file.stream_position()? == 0 && self.internal_pos == 0 {
- // We read until the start of the file, we are done
- break;
- }
+ /// The current position in metadata_buf.
+ metadata_pos: usize,
+}
- 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;
- }
- }
- }
+impl ReverseLogReader {
+ pub fn new(
+ mut metadata_file: fs::File,
+ data_file: fs::File,
+ ) -> Result<ReverseLogReader, io::Error> {
+ let mut metadata_buf = vec![];
+ metadata_file.seek(io::SeekFrom::Start(0))?;
+ metadata_file.read_to_end(&mut metadata_buf)?;
+ let len = metadata_buf.len();
- result_buf.reverse();
- Ok(Some(Record::deserialize(&result_buf)))
+ let ret = ReverseLogReader {
+ metadata_buf,
+ data_reader: io::BufReader::new(data_file),
+ metadata_pos: len,
+ };
+ Ok(ret)
}
- /// 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<usize, io::Error> {
- 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",
- ));
- }
- }
+ fn read_record(&mut self) -> Result<Option<Record>, io::Error> {
+ assert!(
+ (self.metadata_pos - METADATA_FILE_HEADER_SIZE) % 16 == 0,
+ "metadata_pos is not aligned"
+ );
- 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;
+ // Return None if we have read the entire metadata file and
+ // reached the end of the header
+ if self.metadata_pos == METADATA_FILE_HEADER_SIZE {
+ return Ok(None);
}
- Ok(read)
- }
-
- fn populate_internal_buf(&mut self) -> Result<usize, io::Error> {
- 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))?;
+ self.metadata_pos -= 16;
+ let i = self.metadata_pos; // shorter alias
- 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<u8>) -> Result<usize, io::Error> {
- // 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);
- }
- }
+ // 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(self.metadata_buf[i..i + 8].try_into().unwrap());
+ let entry_length = u64::from_be_bytes(self.metadata_buf[i + 8..i + 16].try_into().unwrap());
- 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;
- }
- }
- }
+ debug!(
+ "Read offset {} and length {} from metadata file at position {}",
+ entry_offset, entry_length, i
+ );
- fn read_special_sequence(&mut self) -> Result<SpecialSequence, io::Error> {
- let mut special_buf: Vec<u8> = vec![0; SEQ_RECORD_SEP.len()];
- self.read_exact(&mut special_buf)?;
+ 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)?;
- match validate_special(&special_buf.as_slice()) {
- Some(special) => Ok(special),
- None => {
- let pos = self.file.stream_position().unwrap() + self.internal_pos as u64;
+ let record = Record::deserialize(&result_buf);
- Err(io::Error::new(
- io::ErrorKind::InvalidData,
- format!(
- "Not a special sequence: {:?} at pos: {:x}",
- special_buf, pos,
- ),
- ))
- }
- }
+ assert!(
+ (self.metadata_pos - METADATA_FILE_HEADER_SIZE) % 16 == 0,
+ "metadata_pos is not aligned"
+ );
+ Ok(Some(record))
}
}
#[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()
+ let _ = env_logger::builder().is_test(true).try_init();
+ let metadata_path = Path::new(TEST_RESOURCES_DIR).join("test_metadata_1");
+ let metadata_file = fs::OpenOptions::new()
.read(true)
- .open(&db_path)
+ .open(&metadata_path)
.expect("Failed to open file");
- let mut reverse_log_reader = ReverseLogReader::new(&mut file).unwrap();
+ let data_file = fs::OpenOptions::new()
+ .read(true)
+ .open(Path::new(TEST_RESOURCES_DIR).join("test_data_1"))
+ .expect("Failed to open file");
+
+ let mut reverse_log_reader = ReverseLogReader::new(metadata_file, data_file).unwrap();
- // There are two records in the log with "schema": Int, Null
+ // There are two records in the log with "schema": Int
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,
+ [RecordValue::Bytes(bytes)] => bytes.len() == 256,
_ => 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<'_> {
+impl Iterator for ReverseLogReader {
type Item = Record;
fn next(&mut self) -> Option<Self::Item> {
@@ -347,7 +111,7 @@ impl Iterator for ReverseLogReader<'_> {
Ok(Some(record)) => Some(record),
Ok(None) => None,
Err(err) => {
- panic!("Error reading record: {:?}", err,)
+ panic!("Error reading record: {:?}", err)
}
}
}
diff --git a/log_db/src/memtable_secondary.rs b/log_db/src/memtable_secondary.rs
index b4ebb8f..3f9863b 100644
--- a/log_db/src/memtable_secondary.rs
+++ b/log_db/src/memtable_secondary.rs
@@ -1,7 +1,6 @@
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
diff --git a/log_db/tests/integration.rs b/log_db/tests/integration.rs
index 49a7e7b..b43f673 100644
--- a/log_db/tests/integration.rs
+++ b/log_db/tests/integration.rs
@@ -1,10 +1,12 @@
+#[macro_use]
+extern crate log;
extern crate ctor;
extern crate tempfile;
use ctor::ctor;
use env_logger;
use log_db::*;
-use std::fs::{self, OpenOptions};
+use std::fs::{self};
use std::path::Path;
use std::thread;
use std::time::Duration;
@@ -280,77 +282,9 @@ fn test_upsert_and_get_from_secondary_memtable() {
}
#[test]
-fn test_initialize_and_read_from_primary_memtable_fixture_db2() {
- let data_dir = tmp_dir();
- // Copy the fixture DB to the test data directory
- fs::create_dir_all(&data_dir).expect("Failed to create the test data directory");
- fs::copy(
- &Path::new(TEST_RESOURCES_DIR).join("test_db2"),
- &Path::new(&data_dir).join(ACTIVE_SYMLINK_FILENAME),
- )
- .expect("Failed to copy the fixture DB");
-
- let mut db = DB::configure()
- .data_dir(&data_dir)
- .fields(vec![
- (Field::Id, RecordField::int()),
- (Field::Name, RecordField::string()),
- (Field::Data, RecordField::bytes()),
- ])
- .primary_key(Field::Id)
- .initialize()
- .expect("Failed to initialize DB instance");
-
- // Delete the DB so that any results must come from a memtable
- fs::remove_file(Path::new(&data_dir).join(ACTIVE_SYMLINK_FILENAME))
- .expect("Failed to delete the DB log file");
-
- let result = db.get(&RecordValue::Int(1)).unwrap().unwrap();
-
- // Check that the IDs match
- let expected = RecordValue::Int(1);
- assert!(match (&result.values[0], &expected) {
- (RecordValue::Int(a), RecordValue::Int(b)) => a == b,
- _ => false,
- });
-}
-
-#[test]
-fn test_initialize_without_memtables_fixture_db3() {
- let data_dir = tmp_dir();
- // Copy the fixture DB to the test data directory
- fs::create_dir_all(&data_dir).expect("Failed to create the test data directory");
- fs::copy(
- &Path::new(TEST_RESOURCES_DIR).join("test_db3"),
- &Path::new(&data_dir).join(ACTIVE_SYMLINK_FILENAME),
- )
- .expect("Failed to copy the fixture DB");
-
- let mut db = DB::configure()
- .data_dir(&data_dir)
- .fields(vec![
- (Field::Id, RecordField::int()),
- (Field::Name, RecordField::string()),
- (Field::Data, RecordField::bytes()),
- ])
- .memtable_capacity(0)
- .primary_key(Field::Id)
- .initialize()
- .expect("Failed to initialize DB instance");
-
- let result = db.get(&RecordValue::Int(1)).unwrap().unwrap();
-
- // Check that the IDs match
- let expected = RecordValue::Int(1);
- assert!(match (&result.values[0], &expected) {
- (RecordValue::Int(a), RecordValue::Int(b)) => a == b,
- _ => false,
- });
-}
-
-#[test]
fn test_multiple_writing_threads() {
let data_dir = tmp_dir();
+ debug!("Data dir: {:?}", data_dir);
let mut threads = vec![];
let threads_n = 100;
@@ -458,50 +392,14 @@ fn test_one_writer_and_multiple_reading_threads() {
}
#[test]
-fn test_literal_escape_is_escaped() {
- let data_dir = tmp_dir();
-
- let mut db = DB::configure()
- .data_dir(&data_dir)
- .memtable_capacity(0) // disable memtables
- .fields(vec![
- (Field::Id, RecordField::int()),
- (Field::Data, RecordField::bytes()),
- ])
- .primary_key(Field::Id)
- .initialize()
- .expect("Failed to initialize DB instance");
-
- let record = Record {
- values: vec![
- RecordValue::Int(1),
- RecordValue::Bytes(vec![0x1A, 0x1B, 0x1C, 0x1D]),
- ],
- };
-
- db.upsert(&record).expect("Failed to upsert record");
-
- let found = db
- .get(&RecordValue::Int(1))
- .expect("Failed to get record")
- .expect("Record not found");
-
- let received = match &found.values[1] {
- RecordValue::Bytes(bytes) => bytes,
- _ => panic!("Unexpected record value type"),
- };
-
- assert_eq!(received, &vec![0x1A, 0x1B, 0x1C, 0x1D]);
-}
-
-#[test]
fn test_log_is_rotated_when_capacity_reached() {
let data_dir = tmp_dir();
+ let data_dir_path = Path::new(&data_dir);
let record = Record {
values: vec![RecordValue::Int(1), RecordValue::Bytes(vec![1, 2, 3, 4])],
};
- let record_len = &record.serialize().len() + SEQ_RECORD_SEP.len();
+ let record_len = &record.serialize().len();
let mut db = DB::configure()
.data_dir(&data_dir)
@@ -524,46 +422,40 @@ fn test_log_is_rotated_when_capacity_reached() {
}
// Check that the rotated segments exist
- assert!(Path::new(&data_dir)
- .join(ACTIVE_SYMLINK_FILENAME)
- .with_extension("1")
- .exists());
-
- assert!(Path::new(&data_dir)
- .join(ACTIVE_SYMLINK_FILENAME)
- .with_extension("2")
- .exists());
+ assert!(data_dir_path.join("metadata").with_extension("1").exists());
+ assert!(data_dir_path.join("metadata").with_extension("2").exists());
// 3rd segment should not exist (note negation)
- assert!(!Path::new(&data_dir)
- .join(ACTIVE_SYMLINK_FILENAME)
- .with_extension("3")
- .exists());
+ assert!(!data_dir_path.join("metadata").with_extension("3").exists());
- // Check that the active file only contains five rows
- let mut file = OpenOptions::new()
- .read(true)
- .open(Path::new(&data_dir).join(ACTIVE_SYMLINK_FILENAME))
- .expect("File could not be opened");
- let records_in_active_log = ForwardLogReader::new(&mut file).count();
- assert_eq!(records_in_active_log, 5);
+ // TODO re-implement rest of the test
+ // when refactor is done
- // Check that each rotated file contains only 1 record
- // because of compaction
- for i in &[1, 2] {
- let mut file = OpenOptions::new()
- .read(true)
- .open(
- Path::new(&data_dir)
- .join(ACTIVE_SYMLINK_FILENAME)
- .with_extension(i.to_string()),
- )
- .expect("File could not be opened");
- let records_in_rotated_log = ForwardLogReader::new(&mut file).count();
- assert_eq!(records_in_rotated_log, 1);
- }
+ // // Check that the active file only contains five rows
+ // let mut file = OpenOptions::new()
+ // .read(true)
+ // .open(data_dir_path.join(ACTIVE_SYMLINK_FILENAME))
+ // .expect("File could not be opened");
+
+ // let records_in_active_log = ForwardLogReader::new(&mut file).count();
+ // assert_eq!(records_in_active_log, 5);
+
+ // // Check that each rotated file contains only 1 record
+ // // because of compaction
+ // for i in &[1, 2] {
+ // let mut file = OpenOptions::new()
+ // .read(true)
+ // .open(
+ // Path::new(&data_dir)
+ // .join(ACTIVE_SYMLINK_FILENAME)
+ // .with_extension(i.to_string()),
+ // )
+ // .expect("File could not be opened");
+ // let records_in_rotated_log = ForwardLogReader::new(&mut file).count();
+ // assert_eq!(records_in_rotated_log, 1);
+ // }
- // Look for nonexistant record to scan all segment files
- let found = db.get(&RecordValue::Int(2)).expect("Failed to get record");
- assert!(found.is_none());
+ // // Look for nonexistant record to scan all segment files
+ // let found = db.get(&RecordValue::Int(2)).expect("Failed to get record");
+ // assert!(found.is_none());
}
diff --git a/log_db/tests/resources/test_data_1 b/log_db/tests/resources/test_data_1
index d4c6a21..16dd175 100644
--- a/log_db/tests/resources/test_data_1
+++ b/log_db/tests/resources/test_data_1
Binary files differ
diff --git a/log_db/tests/resources/test_db1 b/log_db/tests/resources/test_db1
deleted file mode 100644
index 1f19a86..0000000
--- a/log_db/tests/resources/test_db1
+++ /dev/null
Binary files differ
diff --git a/log_db/tests/resources/test_db2 b/log_db/tests/resources/test_db2
deleted file mode 100644
index 156659f..0000000
--- a/log_db/tests/resources/test_db2
+++ /dev/null
Binary files differ
diff --git a/log_db/tests/resources/test_db3 b/log_db/tests/resources/test_db3
deleted file mode 100644
index 156659f..0000000
--- a/log_db/tests/resources/test_db3
+++ /dev/null
Binary files differ
diff --git a/log_db/tests/resources/test_metadata_1 b/log_db/tests/resources/test_metadata_1
index 9694599..d8dcc88 100644
--- a/log_db/tests/resources/test_metadata_1
+++ b/log_db/tests/resources/test_metadata_1
Binary files differ
diff --git a/py_bindings/src/lib.rs b/py_bindings/src/lib.rs
index 9ad1362..a1e7bf7 100644
--- a/py_bindings/src/lib.rs
+++ b/py_bindings/src/lib.rs
@@ -1,7 +1,6 @@
use log_db;
use pyo3::exceptions::PyException;
use pyo3::prelude::*;
-use pyo3::types::PyTuple;
type Field = String;

Want to contribute? Host your changes somewhere and send me the link: jan (at) sydanviita.fi. Also, check out my website.


forge.jan.systems — powered by cgit