diff options
Diffstat (limited to 'log_db/src/secondary_memtable.rs')
| -rw-r--r-- | log_db/src/secondary_memtable.rs | 145 |
1 files changed, 23 insertions, 122 deletions
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 - } } |
