diff options
Diffstat (limited to 'log_db/src')
| -rw-r--r-- | log_db/src/lib.rs | 29 | ||||
| -rw-r--r-- | log_db/src/primary_memtable.rs | 28 | ||||
| -rw-r--r-- | log_db/src/secondary_memtable.rs | 145 |
3 files changed, 63 insertions, 139 deletions
diff --git a/log_db/src/lib.rs b/log_db/src/lib.rs index 49caabc..2e77e19 100644 --- a/log_db/src/lib.rs +++ b/log_db/src/lib.rs @@ -232,14 +232,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> { let secondary_memtables = config .secondary_keys .iter() - .map(|key| { - SecondaryMemtable::new( - key, - primary_key_index, - config.memtable_capacity, - config.memtable_evict_policy.clone(), - ) - }) + .map(|key| SecondaryMemtable::new(key)) .collect(); let mut db = DB::<Field> { @@ -507,9 +500,10 @@ impl<Field: Eq + Clone + Debug> DB<Field> { "Found suitable secondary index. Looking up key {:?} in the memtable", query_key ); - let records = self.secondary_memtables[memtable_index].find_all(&query_key); + let records = self.secondary_memtables[memtable_index] + .find_all(&self.primary_memtable, &query_key); debug!("Found matching key"); - return Ok(records.iter().map(|&record| record.clone()).collect()); + return Ok(records.iter().map(|record| record.clone()).collect()); } debug!( @@ -560,7 +554,15 @@ impl<Field: Eq + Clone + Debug> DB<Field> { if let Some(memtable_index) = found_memtable_index { debug!("Inserting result set into secondary index"); - self.secondary_memtables[memtable_index].set_all(&query_key, &result); + let primary_values: Vec<IndexableValue> = result + .iter() + .map(|r| { + r.values[self.primary_key_index] + .as_indexable() + .expect("A non-indexable value was stored at primary key index") + }) + .collect(); + self.secondary_memtables[memtable_index].set_all(&query_key, &primary_values); } Ok(result) @@ -606,10 +608,13 @@ impl<Field: Eq + Clone + Debug> DB<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, record); + secondary_memtable.set(&key, &primary_key); } } }); diff --git a/log_db/src/primary_memtable.rs b/log_db/src/primary_memtable.rs index 0c94ceb..141238f 100644 --- a/log_db/src/primary_memtable.rs +++ b/log_db/src/primary_memtable.rs @@ -3,15 +3,29 @@ 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`. capacity: usize, - /// Running counter of memtable operations, used as priority - /// in evict_queue. - n_operations: u64, + /// 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>, - /// A max heap priority queue of keys. Note: n_operations must - /// be negated upon append to evict oldest values first. + /// 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 { @@ -59,6 +73,10 @@ impl PrimaryMemtable { 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) { diff --git a/log_db/src/secondary_memtable.rs b/log_db/src/secondary_memtable.rs index fcf4a7f..194d4a0 100644 --- a/log_db/src/secondary_memtable.rs +++ b/log_db/src/secondary_memtable.rs @@ -1,178 +1,79 @@ -use super::common::*; -use priority_queue::PriorityQueue; +use super::*; use std::collections::BTreeMap; use std::collections::HashSet; use std::fmt::Debug; -use std::hash::{Hash, Hasher}; - -struct UniqueRecord { - /// The value of the record's primary key field - primary_value: IndexableValue, - /// The record itself. - record: Record, -} - -impl PartialEq for UniqueRecord { - fn eq(&self, other: &UniqueRecord) -> bool { - self.primary_value == other.primary_value - } -} - -impl Eq for UniqueRecord {} - -impl Hash for UniqueRecord { - fn hash<H>(&self, state: &mut H) - where - H: Hasher, - { - self.primary_value.hash(state) - } -} pub struct SecondaryMemtable<Field: Eq + Clone + Debug> { pub field: Field, - pub primary_field_index: usize, - capacity: usize, - /// Running counter of memtable operations, used as priority - /// in evict_queue. - n_operations: u64, - records: BTreeMap<IndexableValue, HashSet<UniqueRecord>>, - /// A max heap priority queue of keys. Note: n_operations must - /// be negated upon append to evict oldest values first. - evict_queue: PriorityQueue<IndexableValue, i64>, - evict_policy: MemtableEvictPolicy, + + /// 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. + records: BTreeMap<IndexableValue, HashSet<IndexableValue>>, } impl<Field: Eq + Clone + Debug> SecondaryMemtable<Field> { - pub fn new( - field: &Field, - primary_field_index: usize, - capacity: usize, - evict_policy: MemtableEvictPolicy, - ) -> SecondaryMemtable<Field> { + pub fn new(field: &Field) -> SecondaryMemtable<Field> { SecondaryMemtable { field: field.clone(), - primary_field_index, - capacity, - n_operations: 0, records: BTreeMap::new(), - evict_queue: PriorityQueue::new(), - evict_policy, } } - pub fn set(&mut self, key: &IndexableValue, value: &Record) { - if self.capacity == 0 { - return; - } - + pub fn set(&mut self, key: &IndexableValue, value: &IndexableValue) { debug!( "Inserting/updating record in secondary memtable with key {:?} = {:?}", &key, &value, ); - if self.records.len() >= self.capacity { - let (evict_key, _prio) = self.evict_queue.pop().expect("Evict queue was empty"); - self.records.remove(&evict_key); - } - - let unique_record = UniqueRecord { - primary_value: value.values[self.primary_field_index] - .as_indexable() - .expect("Value at primary field index was not indexable"), - record: value.clone(), - }; - match self.records.get_mut(key) { Some(existing) => { debug!( "Existing entry found with {} records in the set", &existing.len() ); - existing.insert(unique_record); + existing.insert(value.clone()); } None => { debug!("No existing entry found, creating one."); let mut set = HashSet::with_capacity(1); - set.insert(unique_record); + set.insert(value.clone()); self.records.insert(key.clone(), set); } } - - if self.evict_policy == MemtableEvictPolicy::LeastWritten - || self.evict_policy == MemtableEvictPolicy::LeastReadOrWritten - { - self.set_priority(&key); - } } - pub fn set_all(&mut self, key: &IndexableValue, values: &[Record]) { - if self.capacity == 0 { - return; - } - + pub fn set_all(&mut self, key: &IndexableValue, values: &[IndexableValue]) { debug!( "Replacing set of records in secondary memtable with key {:?} ({} values)", &key, &values.len(), ); - if self.records.len() >= self.capacity { - let (evict_key, _prio) = self.evict_queue.pop().expect("Evict queue was empty"); - self.records.remove(&evict_key); - } - let mut set = HashSet::with_capacity(values.len()); values.iter().for_each(|value| { - let unique_record = UniqueRecord { - primary_value: value.values[self.primary_field_index] - .as_indexable() - .expect("Value at primary field index was not indexable"), - record: value.clone(), - }; - set.insert(unique_record); + set.insert(value.clone()); }); self.records.insert(key.clone(), set); - - if self.evict_policy == MemtableEvictPolicy::LeastWritten - || self.evict_policy == MemtableEvictPolicy::LeastReadOrWritten - { - self.set_priority(&key); - } } - pub fn find_all(&mut self, key: &IndexableValue) -> Vec<&Record> { - if self.evict_policy == MemtableEvictPolicy::LeastRead - || self.evict_policy == MemtableEvictPolicy::LeastReadOrWritten - { - self.set_priority(&key); - } - + pub fn find_all( + &mut self, + primary_memtable: &PrimaryMemtable, + key: &IndexableValue, + ) -> Vec<Record> { match self.records.get(key) { None => vec![], Some(set) => set .iter() - .map(|unique_record| &unique_record.record) + .map(|key| { + primary_memtable + .get_without_update(key) + .expect("Record not found") + .clone() + }) .collect(), } } - - 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 - } } |