Start reimplementing memtable changes after revert

3 files changed, 47 insertions, 82 deletions

diff --git a/log_db/src/lib.rs b/log_db/src/lib.rs
index d7b10eb..8b342dd 100644
--- a/log_db/src/lib.rs
+++ b/log_db/src/lib.rs
@@ -16,9 +16,7 @@ use memtable_secondary::SecondaryMemtable;
 use std::collections::BTreeMap;
 use std::fmt::Debug;
 use std::fs::{self};
-use std::io::Seek;
-use std::io::SeekFrom;
-use std::io::{self, Write};
+use std::io::{self, Read, Seek, SeekFrom, Write};
 use std::os::unix::fs::MetadataExt;
 use std::path::{Path, PathBuf};
 use uuid::Uuid;
@@ -362,9 +360,41 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
 
         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()));
+        if let Some(log_key) = found {
+            debug!("Found log_key in primary memtable: {:?}", log_key);
+            let segment_num = log_key.segment_num();
+            let segment_index = log_key.index();
+
+            let metadata_path = &self.data_dir.join(format!("metadata.{}", segment_num));
+            let mut metadata_file = READ_MODE.open(&metadata_path)?;
+
+            request_shared_lock(&self.data_dir, &mut metadata_file)?;
+
+            let metadata_header = read_metadata_header(&mut metadata_file)?;
+
+            metadata_file.seek_relative(segment_index as i64 * 16)?;
+
+            let mut metadata_buf = [0; 2 * 8];
+            metadata_file.read_exact(&mut metadata_buf)?;
+
+            metadata_file.unlock()?;
+
+            let data_offset = u64::from_be_bytes(metadata_buf[0..8].try_into().unwrap());
+            let data_length = u64::from_be_bytes(metadata_buf[8..16].try_into().unwrap());
+
+            let data_path = &self.data_dir.join(metadata_header.uuid.to_string());
+            let mut data_file = READ_MODE.open(&data_path)?;
+
+            request_shared_lock(&self.data_dir, &mut data_file)?;
+
+            data_file.seek(SeekFrom::Start(data_offset))?;
+
+            let mut data_buf = vec![0; data_length as usize];
+            data_file.read_exact(&mut data_buf)?;
+
+            let record = Record::deserialize(&data_buf);
+
+            return Ok(Some(record));
         }
 
         debug!(
@@ -471,10 +501,8 @@ 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(&self.primary_memtable, &query_key);
-            debug!("Found matching key");
-            return Ok(records.iter().map(|record| record.clone()).collect());
+
+            // TODO: Implement secondary memtable search
         }
 
         debug!(
@@ -544,19 +572,6 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
             found_records.len()
         );
 
-        if let Some(memtable_index) = found_memtable_index {
-            debug!("Inserting result set into secondary index");
-            let primary_values: Vec<IndexableValue> = found_records
-                .iter()
-                .map(|r| {
-                    r.at(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(found_records)
     }
 
diff --git a/log_db/src/memtable_primary.rs b/log_db/src/memtable_primary.rs
index 9d7e1e8..6a28f30 100644
--- a/log_db/src/memtable_primary.rs
+++ b/log_db/src/memtable_primary.rs
@@ -9,7 +9,7 @@ pub struct PrimaryMemtable {
     ///
     /// Note: it must be invariant that all memtables (primary and secondary)
     /// contain the same keys.
-    records: BTreeMap<IndexableValue, Record>,
+    records: BTreeMap<IndexableValue, LogKey>,
 }
 
 impl PrimaryMemtable {
@@ -19,15 +19,11 @@ impl PrimaryMemtable {
         }
     }
 
-    pub fn set(&mut self, key: &IndexableValue, value: &Record) {
+    pub fn set(&mut self, key: &IndexableValue, value: &LogKey) {
         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> {
+    pub fn get(&self, key: &IndexableValue) -> Option<&LogKey> {
         self.records.get(key)
     }
 }
diff --git a/log_db/src/memtable_secondary.rs b/log_db/src/memtable_secondary.rs
index 3f9863b..371cb84 100644
--- a/log_db/src/memtable_secondary.rs
+++ b/log_db/src/memtable_secondary.rs
@@ -1,12 +1,11 @@
 use super::*;
 use std::collections::BTreeMap;
-use std::collections::HashSet;
 
 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.
-    records: BTreeMap<IndexableValue, HashSet<IndexableValue>>,
+    records: BTreeMap<IndexableValue, LogKeySet>,
 }
 
 impl SecondaryMemtable {
@@ -17,59 +16,14 @@ impl SecondaryMemtable {
     }
 
     pub fn set(&mut self, key: &IndexableValue, value: &IndexableValue) {
-        debug!(
-            "Inserting/updating record in secondary memtable with key {:?} = {:?}",
-            &key, &value,
-        );
-
-        match self.records.get_mut(key) {
-            Some(existing) => {
-                debug!(
-                    "Existing entry found with {} records in the set",
-                    &existing.len()
-                );
-                existing.insert(value.clone());
-            }
-            None => {
-                debug!("No existing entry found, creating one.");
-                let mut set = HashSet::with_capacity(1);
-                set.insert(value.clone());
-                self.records.insert(key.clone(), set);
-            }
-        }
+        unimplemented!();
     }
 
-    pub fn set_all(&mut self, key: &IndexableValue, values: &[IndexableValue]) {
-        debug!(
-            "Replacing set of records in secondary memtable with key {:?} ({} values)",
-            &key,
-            &values.len(),
-        );
-
-        let mut set = HashSet::with_capacity(values.len());
-        values.iter().for_each(|value| {
-            set.insert(value.clone());
-        });
-
-        self.records.insert(key.clone(), set);
+    pub fn set_all(&mut self, key: &IndexableValue, values: &LogKeySet) {
+        unimplemented!();
     }
 
-    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(|key| {
-                    primary_memtable
-                        .get_without_update(key)
-                        .expect("Record not found")
-                        .clone()
-                })
-                .collect(),
-        }
+    pub fn find_all(&self, key: &IndexableValue) -> &LogKeySet {
+        unimplemented!();
     }
 }