aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--Cargo.lock10
-rw-r--r--log_db/Cargo.toml1
-rw-r--r--log_db/src/common.rs116
-rw-r--r--log_db/src/lib.rs170
-rw-r--r--log_db/src/log_reader_forward.rs (renamed from log_db/src/forward_log_reader.rs)0
-rw-r--r--log_db/src/log_reader_reverse.rs (renamed from log_db/src/reverse_log_reader.rs)0
-rw-r--r--log_db/src/memtable_primary.rs33
-rw-r--r--log_db/src/memtable_secondary.rs (renamed from log_db/src/secondary_memtable.rs)43
-rw-r--r--log_db/src/primary_memtable.rs96
-rw-r--r--log_db/tests/integration.rs20
-rw-r--r--py_bindings/src/lib.rs13
11 files changed, 206 insertions, 296 deletions
diff --git a/Cargo.lock b/Cargo.lock
index 3dad95c..8d078fa 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -518,6 +518,7 @@ dependencies = [
"rand",
"serial_test",
"tempfile",
+ "uuid",
]
[[package]]
@@ -1002,6 +1003,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"
[[package]]
+name = "uuid"
+version = "1.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f8c5f0a0af699448548ad1a2fbf920fb4bee257eae39953ba95cb84891a0446a"
+dependencies = [
+ "getrandom",
+]
+
+[[package]]
name = "walkdir"
version = "2.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
diff --git a/log_db/Cargo.toml b/log_db/Cargo.toml
index 050abd9..723d29d 100644
--- a/log_db/Cargo.toml
+++ b/log_db/Cargo.toml
@@ -11,6 +11,7 @@ fs2 = "0.4.3"
log = "0.4.22"
priority-queue = "2.1.1"
tempfile = "3.13.0"
+uuid = { version = "1.11.0", features = ["v4"] }
[dev-dependencies]
ctor = "0.2.8"
diff --git a/log_db/src/common.rs b/log_db/src/common.rs
index c2af766..0a1179e 100644
--- a/log_db/src/common.rs
+++ b/log_db/src/common.rs
@@ -1,7 +1,9 @@
+use std::cmp::Ordering;
+use std::collections::HashSet;
use std::fmt::Display;
use std::fs::{metadata, File};
use std::io::{self};
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
// For Unix-like systems
#[cfg(unix)]
@@ -11,7 +13,7 @@ use std::os::unix::fs::MetadataExt;
#[cfg(windows)]
use std::os::windows::fs::MetadataExt;
-pub const ACTIVE_LOG_FILENAME: &str = "db";
+pub const ACTIVE_SYMLINK_FILENAME: &str = "active";
pub const EXCL_LOCK_REQUEST_FILENAME: &str = "excl_lock_req";
pub const DEFAULT_READ_BUF_SIZE: usize = 1024 * 1024; // 1 MB
pub const FIELD_SEPARATOR: u8 = b'\x1C';
@@ -63,11 +65,93 @@ pub enum SpecialSequence {
LiteralEscape,
}
+/// LogKey is a packed struct that contains:
+/// - a log segment number (16 bits)
+/// - a log index within the segment (48 bits)
+#[derive(Debug, Clone, Eq, PartialEq, Hash, Ord, PartialOrd)]
+pub struct LogKey(u64);
+
+impl LogKey {
+ pub fn new(segment_num: u16, index: u64) -> Self {
+ assert!(index < (1 << 48), "Index must fit in 48 bits");
+ LogKey((segment_num as u64) << 48 | index)
+ }
+
+ pub fn segment_num(&self) -> u16 {
+ (self.0 >> 48) as u16
+ }
+
+ pub fn index(&self) -> u64 {
+ self.0 & 0x0000_FFFF_FFFF_FFFF
+ }
+}
+
+/// LogKeySet is a non-empty set of LogKeys.
#[derive(Debug, Clone, Eq, PartialEq)]
-pub enum MemtableEvictPolicy {
- LeastWritten,
- LeastRead,
- LeastReadOrWritten,
+pub struct LogKeySet {
+ set: HashSet<LogKey>,
+}
+
+impl PartialOrd for LogKeySet {
+ fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+ let self_max_elem = self.set.iter().max()?;
+ let other_max_elem = other.set.iter().max()?;
+ Some(self_max_elem.cmp(other_max_elem))
+ }
+}
+
+impl LogKeySet {
+ /// Create a new LogKeySet with an initial LogKey.
+ /// The initial LogKey is required since LogKeySet must be non-empty.
+ pub fn new_with_initial(key: &LogKey) -> Self {
+ let mut set = HashSet::new();
+ set.insert(key.clone());
+ LogKeySet { set }
+ }
+
+ /// Insert a LogKey into the set.
+ pub fn insert(&mut self, key: LogKey) {
+ self.set.insert(key);
+ }
+
+ /// Remove a LogKey from the set. Return Ok(()) if the key was found and removed.
+ /// Return io::Error::InvalidInput if trying to remove the last element.
+ /// Return io::Error::NotFound if the key was not found.
+ pub fn remove(&mut self, key: &LogKey) -> Result<(), io::Error> {
+ if self.set.len() == 1 {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidInput,
+ "Cannot remove the last element from LogKeySet",
+ ));
+ }
+ let removed = self.set.remove(key);
+
+ if !removed {
+ return Err(io::Error::new(
+ io::ErrorKind::NotFound,
+ "LogKey not found in LogKeySet",
+ ));
+ }
+
+ assert!(
+ self.set.len() > 0,
+ "LogKeySet should not be empty after removal"
+ );
+
+ Ok(())
+ }
+
+ /// Get a reference to the set of LogKeys.
+ pub fn log_keys(&self) -> &HashSet<LogKey> {
+ &self.set
+ }
+}
+
+impl Ord for LogKeySet {
+ fn cmp(&self, other: &Self) -> Ordering {
+ self.partial_cmp(other)
+ .expect("LogKeySet comparison failed, possibly due to empty set")
+ }
}
#[derive(Debug, Clone, Eq, PartialEq)]
@@ -284,6 +368,14 @@ pub fn escape_bytes(buf: &[u8]) -> Vec<u8> {
result
}
+/// A path to a log segment file along with its type
+pub enum SegmentPath {
+ /// A symbolic link to the active log file
+ ActiveSymlink(String),
+ /// A compacted segment that is no longer being written to
+ Compacted(String),
+}
+
pub fn is_file_same_as_path(file: &File, path: &PathBuf) -> io::Result<bool> {
// Get the metadata for the open file handle
let file_metadata = file.metadata()?;
@@ -306,3 +398,15 @@ pub fn is_file_same_as_path(file: &File, path: &PathBuf) -> io::Result<bool> {
&& file_metadata.volume_serial_number() == path_metadata.volume_serial_number())
}
}
+
+pub fn symlink(original: &Path, link: &Path) -> io::Result<()> {
+ #[cfg(unix)]
+ {
+ std::os::unix::fs::symlink(original, link)
+ }
+
+ #[cfg(windows)]
+ {
+ std::os::windows::fs::symlink_file(original, link)
+ }
+}
diff --git a/log_db/src/lib.rs b/log_db/src/lib.rs
index bcdaefe..0f086a5 100644
--- a/log_db/src/lib.rs
+++ b/log_db/src/lib.rs
@@ -2,18 +2,18 @@
extern crate log;
mod common;
-mod forward_log_reader;
-mod primary_memtable;
-mod reverse_log_reader;
-mod secondary_memtable;
+mod log_reader_forward;
+mod log_reader_reverse;
+mod memtable_primary;
+mod memtable_secondary;
pub use common::*;
-pub use forward_log_reader::ForwardLogReader;
use fs2::lock_contended_error;
use fs2::FileExt;
-use primary_memtable::PrimaryMemtable;
-pub use reverse_log_reader::ReverseLogReader;
-use secondary_memtable::SecondaryMemtable;
+pub use log_reader_forward::ForwardLogReader;
+pub use log_reader_reverse::ReverseLogReader;
+use memtable_primary::PrimaryMemtable;
+use memtable_secondary::SecondaryMemtable;
use std::collections::BTreeMap;
use std::fmt::Debug;
use std::fs::{self};
@@ -30,7 +30,6 @@ pub struct ConfigBuilder<Field: Eq + Clone + Debug> {
fields: Option<Vec<(Field, RecordField)>>,
primary_key: Option<Field>,
secondary_keys: Option<Vec<Field>>,
- memtable_evict_policy: Option<MemtableEvictPolicy>,
write_durability: Option<WriteDurability>,
}
@@ -43,7 +42,6 @@ impl<'a, Field: Eq + Clone + Debug> ConfigBuilder<Field> {
fields: None,
primary_key: None,
secondary_keys: None,
- memtable_evict_policy: None,
write_durability: None,
}
}
@@ -91,17 +89,6 @@ impl<'a, Field: Eq + Clone + Debug> ConfigBuilder<Field> {
self
}
- /// The eviction policy for the memtables. Determines which
- /// record will be dropped from a memtable when it reaches
- /// capacity.
- pub fn memtable_evict_policy(
- &mut self,
- memtable_evict_policy: MemtableEvictPolicy,
- ) -> &mut Self {
- self.memtable_evict_policy = Some(memtable_evict_policy);
- self
- }
-
/// The write durability policy for the database.
/// This determines how writes are persisted to disk.
/// The default is WriteDurability::Flush.
@@ -128,10 +115,6 @@ impl<'a, Field: Eq + Clone + Debug> ConfigBuilder<Field> {
"Required config value \"primary_key\" is not set",
))?,
secondary_keys: self.secondary_keys.clone().unwrap_or(Vec::new()),
- memtable_evict_policy: self
- .memtable_evict_policy
- .clone()
- .unwrap_or(MemtableEvictPolicy::LeastReadOrWritten),
write_durability: self
.write_durability
.clone()
@@ -150,7 +133,6 @@ struct Config<Field: Eq + Clone> {
pub fields: Vec<(Field, RecordField)>,
pub primary_key: Field,
pub secondary_keys: Vec<Field>,
- pub memtable_evict_policy: MemtableEvictPolicy,
pub write_durability: WriteDurability,
}
@@ -160,7 +142,7 @@ pub struct DB<Field: Eq + Clone + Debug> {
log_file: fs::File,
primary_key_index: usize,
primary_memtable: PrimaryMemtable,
- secondary_memtables: Vec<SecondaryMemtable<Field>>,
+ secondary_memtables: Vec<SecondaryMemtable>,
}
impl<Field: Eq + Clone + Debug> DB<Field> {
@@ -176,7 +158,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
fs::create_dir_all(&config.data_dir)?;
}
- let log_path = Path::new(&config.data_dir).join(ACTIVE_LOG_FILENAME);
+ let log_path = Path::new(&config.data_dir).join(ACTIVE_SYMLINK_FILENAME);
// Create the log file if it does not exist
let log_file_file = fs::OpenOptions::new()
@@ -227,14 +209,11 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
}
}
}
- let primary_memtable = PrimaryMemtable::new(
- config.memtable_capacity,
- config.memtable_evict_policy.clone(),
- );
+ let primary_memtable = PrimaryMemtable::new();
let secondary_memtables = config
.secondary_keys
.iter()
- .map(|key| SecondaryMemtable::new(&config.fields, key, primary_key_index))
+ .map(|key| SecondaryMemtable::new())
.collect();
let mut db = DB::<Field> {
@@ -419,7 +398,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
} else {
debug!("Locking and searching rotated log segment file {}...", n);
let path = Path::new(&self.config.data_dir)
- .join(ACTIVE_LOG_FILENAME)
+ .join(ACTIVE_SYMLINK_FILENAME)
.with_extension(n.to_string());
let mut segm_file = fs::OpenOptions::new().read(true).open(&path)?;
@@ -485,10 +464,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
))?;
// Try to find a memtable with the queried key
- let found_memtable_index = self
- .secondary_memtables
- .iter_mut()
- .position(|mt| &mt.field == field);
+ let found_memtable_index = self.get_secondary_memtable_index_by_field(field);
if let Some(memtable_index) = found_memtable_index {
debug!(
@@ -563,6 +539,13 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
Ok(result)
}
+ fn get_secondary_memtable_index_by_field(&self, field: &Field) -> Option<usize> {
+ self.config
+ .secondary_keys
+ .iter()
+ .position(|schema_field| schema_field == field)
+ }
+
/// Ensures that the `self.log_file` handle is still pointing to the correct file.
/// If the file has been rotated, the handle will be closed and reopened.
/// Returns `true` if the file has been rotated and the handle has been reopened.
@@ -591,7 +574,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
.as_indexable()
.expect("A non-indexable value was stored at key index");
- if self.primary_memtable.capacity == 0 {
+ if self.config.memtable_capacity == 0 {
return;
}
@@ -600,35 +583,21 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
&key, &record,
);
- if let Some(evicted) = self.primary_memtable.evict_if_necessary() {
- self.secondary_memtables
- .iter_mut()
- .for_each(|secondary_memtable| {
- secondary_memtable.remove(&evicted);
- });
- }
-
self.primary_memtable.set(&key, record);
- self.secondary_memtables
- .iter_mut()
- .for_each(|secondary_memtable| {
- debug!(
- "Updating memtable for index on {:?}",
- &secondary_memtable.field
- );
- for (index, (schema_field, _)) in self.config.fields.iter().enumerate() {
- if schema_field == &secondary_memtable.field {
- let primary_key = record.values[self.primary_key_index]
- .as_indexable()
- .expect("Primary key was not indexable");
- let key = record.values[index]
- .as_indexable()
- .expect("Secondary index key was not indexable");
- secondary_memtable.set(&key, &primary_key);
- }
- }
- });
+ for (field_index, value) in record.values.iter().enumerate() {
+ let field = &self.config.fields[field_index].0;
+ if let Some(smt_index) = self.get_secondary_memtable_index_by_field(field) {
+ debug!("Updating memtable for index on {:?}", field);
+
+ let memtable = &mut self.secondary_memtables[smt_index];
+ let key = value.as_indexable().expect("Primary key was not indexable");
+ let primary_key = record.values[self.primary_key_index]
+ .as_indexable()
+ .expect("Primary key was not indexable");
+ memtable.set(&key, &primary_key);
+ }
+ }
}
fn request_exclusive_lock(&mut self) -> Result<(), io::Error> {
@@ -717,7 +686,7 @@ impl<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_LOG_FILENAME);
+ let active_log_path = Path::new(&self.config.data_dir).join(ACTIVE_SYMLINK_FILENAME);
let active_log_md = fs::metadata(&active_log_path)?;
if active_log_md.size() >= self.config.segment_size as u64 {
@@ -788,7 +757,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
.expect("Failed to convert file name to string")
.to_string();
- if name != ACTIVE_LOG_FILENAME {
+ if name != ACTIVE_SYMLINK_FILENAME {
return None;
}
@@ -849,68 +818,3 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
Ok(())
}
}
-
-#[cfg(test)]
-mod tests {
- use super::*;
- use rand::distributions::Alphanumeric;
- use rand::Rng;
- use std::collections::HashSet;
- use tempfile::tempdir;
-
- #[derive(Eq, PartialEq, Clone, Debug)]
- enum Field {
- Id,
- Name,
- Data,
- }
-
- fn tmp_dir() -> String {
- let dir = tempdir()
- .expect("Failed to create temporary directory")
- .path()
- .to_str()
- .expect("Failed to convert temporary directory path to string")
- .to_string();
- fs::create_dir_all(&dir).expect("Failed to create temporary directory");
- dir
- }
-
- #[test]
- fn memtables_always_have_the_same_primary_keys() {
- let data_dir = tmp_dir();
-
- let mut db = DB::configure()
- .data_dir(&data_dir)
- .fields(vec![
- (Field::Id, RecordField::int()),
- (Field::Name, RecordField::string()),
- ])
- .primary_key(Field::Id)
- .secondary_keys(vec![Field::Name])
- .initialize()
- .expect("Failed to initialize DB instance");
-
- let mut rng = rand::thread_rng();
- for _ in 0..100 {
- let id = rng.gen_range(0..100);
- let name = (0..5).map(|_| rng.sample(Alphanumeric) as char).collect();
-
- let record = Record {
- values: vec![RecordValue::Int(id), RecordValue::String(name)],
- };
- db.upsert(&record).expect("Failed to upsert record");
-
- let p_set: HashSet<&IndexableValue> = db.primary_memtable.records.keys().collect();
- let mut s_set: HashSet<&IndexableValue> = HashSet::new();
-
- for table in db.secondary_memtables.iter() {
- table.records.values().for_each(|r| {
- s_set.extend(r);
- });
- }
-
- assert_eq!(p_set, s_set);
- }
- }
-}
diff --git a/log_db/src/forward_log_reader.rs b/log_db/src/log_reader_forward.rs
index 03726d0..03726d0 100644
--- a/log_db/src/forward_log_reader.rs
+++ b/log_db/src/log_reader_forward.rs
diff --git a/log_db/src/reverse_log_reader.rs b/log_db/src/log_reader_reverse.rs
index f042406..f042406 100644
--- a/log_db/src/reverse_log_reader.rs
+++ b/log_db/src/log_reader_reverse.rs
diff --git a/log_db/src/memtable_primary.rs b/log_db/src/memtable_primary.rs
new file mode 100644
index 0000000..9d7e1e8
--- /dev/null
+++ b/log_db/src/memtable_primary.rs
@@ -0,0 +1,33 @@
+use super::common::*;
+use std::collections::BTreeMap;
+
+pub struct PrimaryMemtable {
+ /// Map of records indexed by key. Used as a shared heap of records
+ /// for all secondary memtables also. Secondary memtables store an
+ /// IndexableValue as their record value, which is used to get
+ /// the actual record from the primary memtable `records` map.
+ ///
+ /// Note: it must be invariant that all memtables (primary and secondary)
+ /// contain the same keys.
+ records: BTreeMap<IndexableValue, Record>,
+}
+
+impl PrimaryMemtable {
+ pub fn new() -> PrimaryMemtable {
+ PrimaryMemtable {
+ records: BTreeMap::new(),
+ }
+ }
+
+ pub fn set(&mut self, key: &IndexableValue, value: &Record) {
+ self.records.insert(key.clone(), value.clone());
+ }
+
+ pub fn get(&mut self, key: &IndexableValue) -> Option<&Record> {
+ self.records.get(key)
+ }
+
+ pub fn get_without_update(&self, key: &IndexableValue) -> Option<&Record> {
+ self.records.get(key)
+ }
+}
diff --git a/log_db/src/secondary_memtable.rs b/log_db/src/memtable_secondary.rs
index 994ee0d..b4ebb8f 100644
--- a/log_db/src/secondary_memtable.rs
+++ b/log_db/src/memtable_secondary.rs
@@ -3,32 +3,16 @@ use std::collections::BTreeMap;
use std::collections::HashSet;
use std::fmt::Debug;
-pub struct SecondaryMemtable<Field: Eq + Clone + Debug> {
- pub field: Field,
- field_index: usize,
- primary_key_index: usize,
-
+pub struct SecondaryMemtable {
/// Map of records indexed by key. The value is the set of primary key values of records
/// that have the secondary key value. The actual `Record` objects are stored in the
/// primary memtable, which acts as the shared heap.
- pub records: BTreeMap<IndexableValue, HashSet<IndexableValue>>,
+ records: BTreeMap<IndexableValue, HashSet<IndexableValue>>,
}
-impl<Field: Eq + Clone + Debug> SecondaryMemtable<Field> {
- pub fn new(
- field_schema: &Vec<(Field, RecordField)>,
- field: &Field,
- primary_key_index: usize,
- ) -> SecondaryMemtable<Field> {
- let field_index = field_schema
- .iter()
- .position(|(f, _)| f == field)
- .expect("Field not found in schema");
-
+impl SecondaryMemtable {
+ pub fn new() -> SecondaryMemtable {
SecondaryMemtable {
- field: field.clone(),
- field_index,
- primary_key_index,
records: BTreeMap::new(),
}
}
@@ -89,23 +73,4 @@ impl<Field: Eq + Clone + Debug> SecondaryMemtable<Field> {
.collect(),
}
}
-
- pub fn remove(&mut self, record: &Record) {
- let key = record.values[self.field_index]
- .as_indexable()
- .expect("Field is not indexable");
-
- let primary_key = record.values[self.primary_key_index]
- .as_indexable()
- .expect("Primary key is not indexable");
-
- match self.records.get_mut(&key) {
- Some(set) => {
- set.remove(&primary_key);
- }
- None => {
- panic!("Record not found in secondary memtable");
- }
- }
- }
}
diff --git a/log_db/src/primary_memtable.rs b/log_db/src/primary_memtable.rs
deleted file mode 100644
index 180ae72..0000000
--- a/log_db/src/primary_memtable.rs
+++ /dev/null
@@ -1,96 +0,0 @@
-use super::common::*;
-use priority_queue::PriorityQueue;
-use std::collections::BTreeMap;
-
-pub struct PrimaryMemtable {
- /// Maximum number of records that can be stored in the memtable
- /// before evicting the oldest records. The oldest record is
- /// determined by the `evict_policy`.
- pub capacity: usize,
- /// Map of records indexed by key. Used as a shared heap of records
- /// for all secondary memtables also. Secondary memtables store an
- /// IndexableValue as their record value, which is used to get
- /// the actual record from the primary memtable `records` map.
- ///
- /// Note: it must be invariant that all memtables (primary and secondary)
- /// contain the same keys.
- pub records: BTreeMap<IndexableValue, Record>,
- /// A max heap priority queue of keys. The record with least priority is evicted
- /// from the primary memtable and any secondary memtables that reference it, when
- /// the memtable reaches capacity.
- ///
- /// Note: n_operations must be negated upon append to evict oldest values first.
- evict_queue: PriorityQueue<IndexableValue, i64>,
- /// Policy for prioritizing records for eviction.
- evict_policy: MemtableEvictPolicy,
- /// Running counter of memtable operations, used as priority
- /// in evict_queue.
- n_operations: u64,
-}
-
-impl PrimaryMemtable {
- pub fn new(capacity: usize, evict_policy: MemtableEvictPolicy) -> PrimaryMemtable {
- PrimaryMemtable {
- capacity,
- n_operations: 0,
- records: BTreeMap::new(),
- evict_queue: PriorityQueue::new(),
- evict_policy,
- }
- }
-
- pub fn set(&mut self, key: &IndexableValue, value: &Record) {
- self.records.insert(key.clone(), value.clone());
-
- if self.evict_policy == MemtableEvictPolicy::LeastWritten
- || self.evict_policy == MemtableEvictPolicy::LeastReadOrWritten
- {
- self.set_priority(&key);
- }
- }
-
- pub fn get(&mut self, key: &IndexableValue) -> Option<&Record> {
- if self.evict_policy == MemtableEvictPolicy::LeastRead
- || self.evict_policy == MemtableEvictPolicy::LeastReadOrWritten
- {
- self.set_priority(&key);
- }
-
- self.records.get(key)
- }
-
- pub fn get_without_update(&self, key: &IndexableValue) -> Option<&Record> {
- self.records.get(key)
- }
-
- fn set_priority(&mut self, key: &IndexableValue) {
- let priority = self.get_and_increment_current_priority();
- match self.evict_queue.get(key) {
- Some(_) => {
- self.evict_queue.change_priority(key, priority);
- }
- None => {
- self.evict_queue.push(key.clone(), priority);
- }
- }
- }
-
- fn get_and_increment_current_priority(&mut self) -> i64 {
- let ret = -(self.n_operations as i64);
- self.n_operations += 1;
- ret
- }
-
- pub fn evict_if_necessary(&mut self) -> Option<Record> {
- if self.records.len() >= self.capacity {
- let (evict_key, _prio) = self.evict_queue.pop().expect("Evict queue was empty");
- let removed = self
- .records
- .remove(&evict_key)
- .expect("Key was not found in records");
- Some(removed)
- } else {
- None
- }
- }
-}
diff --git a/log_db/tests/integration.rs b/log_db/tests/integration.rs
index 2d2a4ca..7e1aca4 100644
--- a/log_db/tests/integration.rs
+++ b/log_db/tests/integration.rs
@@ -268,7 +268,8 @@ fn test_upsert_and_get_from_secondary_memtable() {
db.upsert(&record2).unwrap();
// Delete the DB so that any results must come from a memtable
- fs::remove_file(Path::new(&data_dir).join("db")).expect("Failed to delete the DB log file");
+ fs::remove_file(Path::new(&data_dir).join(ACTIVE_SYMLINK_FILENAME))
+ .expect("Failed to delete the DB log file");
// There should be 2 Johns
let johns = db
@@ -285,7 +286,7 @@ fn test_initialize_and_read_from_primary_memtable_fixture_db2() {
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("db"),
+ &Path::new(&data_dir).join(ACTIVE_SYMLINK_FILENAME),
)
.expect("Failed to copy the fixture DB");
@@ -301,7 +302,8 @@ fn test_initialize_and_read_from_primary_memtable_fixture_db2() {
.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("db")).expect("Failed to delete the DB log file");
+ 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();
@@ -320,7 +322,7 @@ fn test_initialize_without_memtables_fixture_db3() {
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("db"),
+ &Path::new(&data_dir).join(ACTIVE_SYMLINK_FILENAME),
)
.expect("Failed to copy the fixture DB");
@@ -523,25 +525,25 @@ fn test_log_is_rotated_when_capacity_reached() {
// Check that the rotated segments exist
assert!(Path::new(&data_dir)
- .join(ACTIVE_LOG_FILENAME)
+ .join(ACTIVE_SYMLINK_FILENAME)
.with_extension("1")
.exists());
assert!(Path::new(&data_dir)
- .join(ACTIVE_LOG_FILENAME)
+ .join(ACTIVE_SYMLINK_FILENAME)
.with_extension("2")
.exists());
// 3rd segment should not exist (note negation)
assert!(!Path::new(&data_dir)
- .join(ACTIVE_LOG_FILENAME)
+ .join(ACTIVE_SYMLINK_FILENAME)
.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_LOG_FILENAME))
+ .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);
@@ -553,7 +555,7 @@ fn test_log_is_rotated_when_capacity_reached() {
.read(true)
.open(
Path::new(&data_dir)
- .join(ACTIVE_LOG_FILENAME)
+ .join(ACTIVE_SYMLINK_FILENAME)
.with_extension(i.to_string()),
)
.expect("File could not be opened");
diff --git a/py_bindings/src/lib.rs b/py_bindings/src/lib.rs
index fcf9052..9ad1362 100644
--- a/py_bindings/src/lib.rs
+++ b/py_bindings/src/lib.rs
@@ -50,12 +50,6 @@ impl RecordField {
#[pyclass]
#[derive(Clone)]
-struct MemtableEvictPolicy {
- memtable_evict_policy: log_db::MemtableEvictPolicy,
-}
-
-#[pyclass]
-#[derive(Clone)]
struct WriteDurability {
write_durability: log_db::WriteDurability,
}
@@ -75,8 +69,6 @@ struct Config {
#[pyo3(get, set)]
secondary_keys: Option<Vec<Field>>,
#[pyo3(get, set)]
- memtable_evict_policy: Option<MemtableEvictPolicy>,
- #[pyo3(get, set)]
write_durability: Option<WriteDurability>,
}
@@ -107,10 +99,6 @@ impl Config {
let tmp = self.secondary_keys.as_ref().unwrap();
config.secondary_keys(tmp.clone());
}
- if self.memtable_evict_policy.is_some() {
- let tmp = self.memtable_evict_policy.as_ref().unwrap();
- config.memtable_evict_policy(tmp.memtable_evict_policy.clone());
- }
if self.write_durability.is_some() {
let tmp = self.write_durability.as_ref().unwrap();
config.write_durability(tmp.write_durability.clone());
@@ -208,7 +196,6 @@ impl DB {
fields: None,
primary_key: None,
secondary_keys: None,
- memtable_evict_policy: None,
write_durability: None,
}
}

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