Refactor DB into DB and Engine

1 files changed, 747 insertions, 0 deletions

diff --git a/log_db/src/engine.rs b/log_db/src/engine.rs
new file mode 100644
index 0000000..ef73db2
--- /dev/null
+++ b/log_db/src/engine.rs
@@ -0,0 +1,747 @@
+use super::*;
+
+pub struct Engine<R: Recordable> {
+    pub config: Config<R>,
+
+    data_dir: PathBuf,
+    active_metadata_file: fs::File,
+    active_data_file: fs::File,
+    primary_key_index: usize,
+    refresh_next_logkey: LogKey,
+
+    // TODO: these could be made private. Currently they are public for testing in lib.rs.
+    pub primary_memtable: PrimaryMemtable,
+    pub secondary_memtables: Vec<SecondaryMemtable>,
+}
+
+impl<R: Recordable> Engine<R> {
+    pub fn initialize(config: Config<R>) -> DBResult<Engine<R>> {
+        info!("Initializing DB...");
+        // If data_dir does not exist or is empty, create it and any necessary files
+        // 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.
+
+        // Ensure the data directory exists
+        let data_dir = Path::new(&config.data_dir).to_path_buf();
+        match fs::create_dir(&data_dir) {
+            Ok(_) => {}
+            Err(e) => {
+                if e.kind() != io::ErrorKind::AlreadyExists {
+                    return Err(DBError::IOError(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.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.join(ACTIVE_SYMLINK_FILENAME))? {
+            let (segment_uuid, _) = create_segment_data_file(&data_dir)?;
+            let (segment_num, _) = create_segment_metadata_file(&data_dir, &segment_uuid)?;
+            set_active_segment(&data_dir, segment_num)?;
+
+            // Create the exclusive lock request file
+            fs::OpenOptions::new()
+                .create(true)
+                .write(true)
+                .open(data_dir.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
+            .iter()
+            .position(|(field, _)| field == &config.primary_key)
+            .ok_or(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "Primary key not found in schema after initialize",
+            ))?;
+
+        // Join primary key and secondary keys vec into a single vec
+        let mut all_keys = vec![&config.primary_key];
+        all_keys.extend(&config.secondary_keys);
+
+        // If any of the keys is not in the schema or
+        // is not an IndexableValue, return an error
+        for &key in &all_keys {
+            let (_, value_type) = config.fields.iter().find(|(field, _)| field == key).ok_or(
+                DBError::ValidationError("Key must be present in the field schema".to_owned()),
+            )?;
+
+            match value_type.prim_value_type {
+                PrimValueType::Int | PrimValueType::String => {}
+                _ => return Err(DBError::ValidationError("Key must be indexable".to_owned())),
+            }
+        }
+        let primary_memtable = PrimaryMemtable::new();
+        let secondary_memtables = config
+            .secondary_keys
+            .iter()
+            .map(|_| SecondaryMemtable::new())
+            .collect();
+
+        let active_symlink = Path::new(&config.data_dir).join(ACTIVE_SYMLINK_FILENAME);
+
+        let active_target = fs::read_link(&active_symlink)?;
+        let active_metadata_path = Path::new(&config.data_dir).join(active_target);
+        let mut active_metadata_file = APPEND_MODE.open(&active_metadata_path)?;
+
+        let active_metadata_header = read_metadata_header(&mut active_metadata_file)?;
+        validate_metadata_header(&active_metadata_header)?;
+
+        let active_data_path =
+            Path::new(&config.data_dir).join(active_metadata_header.uuid.to_string());
+        let active_data_file = APPEND_MODE.open(&active_data_path)?;
+
+        let mut engine = Engine::<R> {
+            config,
+            data_dir,
+            active_metadata_file,
+            active_data_file,
+            primary_key_index,
+            primary_memtable,
+            secondary_memtables,
+            refresh_next_logkey: LogKey::new(1, 0),
+        };
+
+        info!("Rebuilding memtable indexes...");
+        engine.refresh_indexes()?;
+
+        info!("Database ready.");
+
+        Ok(engine)
+    }
+
+    pub fn refresh_indexes(&mut self) -> DBResult<()> {
+        let active_symlink_path = self.data_dir.join(ACTIVE_SYMLINK_FILENAME);
+        let active_target = fs::read_link(active_symlink_path)?;
+        let active_metadata_path = self.data_dir.join(active_target);
+
+        let to_segnum = parse_segment_number(&active_metadata_path)?;
+        let from_segnum = self.refresh_next_logkey.segment_num();
+        let mut from_index = self.refresh_next_logkey.index();
+
+        for segnum in from_segnum..=to_segnum {
+            let metadata_path = self.data_dir.join(metadata_filename(segnum));
+            let mut metadata_file = READ_MODE.open(&metadata_path)?;
+
+            let metadata_len = metadata_file.seek(SeekFrom::End(0))?;
+            if (metadata_len - METADATA_FILE_HEADER_SIZE as u64) % METADATA_ROW_LENGTH as u64 != 0 {
+                return Err(DBError::ConsistencyError(format!(
+                    "Metadata file {} has invalid size: {}",
+                    metadata_path.display(),
+                    metadata_len
+                )));
+            }
+
+            request_shared_lock(&self.data_dir, &mut metadata_file)?;
+
+            let metadata_header = read_metadata_header(&mut metadata_file)?;
+            validate_metadata_header(&metadata_header)?;
+
+            let data_path = self.data_dir.join(metadata_header.uuid.to_string());
+            let data_file = READ_MODE.open(data_path)?;
+
+            for ForwardLogReaderItem { record, index } in
+                ForwardLogReader::new_with_index(metadata_file, data_file, from_index)
+            {
+                let log_key = LogKey::new(segnum, index);
+
+                if record.tombstone {
+                    self.remove_record_from_memtables(&record);
+                } else {
+                    self.insert_record_to_memtables(log_key, record);
+                }
+
+                // Update from_index in case this is the last iteration: we need to know the next
+                // index that should be read on later invocations of refresh_indexes.
+                from_index = index + 1
+            }
+
+            // If there are still segments to read, set from_index to zero to read them
+            // from beginning. Otherwise we leave from_index as the index of the next record to read.
+            if segnum != to_segnum {
+                from_index = 0
+            }
+        }
+
+        self.refresh_next_logkey = LogKey::new(to_segnum, from_index);
+
+        Ok(())
+    }
+
+    fn insert_record_to_memtables(&mut self, log_key: LogKey, record: Record) {
+        for (sk_index, sk_field) in self.config.secondary_keys.iter().enumerate() {
+            let secondary_memtable = &mut self.secondary_memtables[sk_index];
+            let sk_field_index = self
+                .config
+                .fields
+                .iter()
+                .position(|(f, _)| sk_field == f)
+                .unwrap();
+            let sk = record.at(sk_field_index).as_indexable().unwrap();
+
+            secondary_memtable.set(sk, log_key.clone());
+        }
+
+        // Doing this last because this moves log_key
+        let pk = record.at(self.primary_key_index).as_indexable().unwrap();
+        self.primary_memtable.set(pk, log_key);
+    }
+
+    fn remove_record_from_memtables(&mut self, record: &Record) {
+        let pk = record.at(self.primary_key_index).as_indexable().unwrap();
+
+        if let Some(plk) = self.primary_memtable.remove(&pk) {
+            for (sk_index, sk_field) in self.config.secondary_keys.iter_mut().enumerate() {
+                let secondary_memtable = &mut self.secondary_memtables[sk_index];
+                let sk_field_index = self
+                    .config
+                    .fields
+                    .iter()
+                    .position(|(f, _)| sk_field == f)
+                    .unwrap();
+                let sk = record.at(sk_field_index).as_indexable().unwrap();
+
+                secondary_memtable.remove(&sk, &plk);
+            }
+        }
+    }
+
+    pub fn batch_upsert_records(&mut self, records: impl Iterator<Item = Record>) -> DBResult<()> {
+        debug!("Opening file in append mode and acquiring exclusive lock...");
+
+        // Acquire an exclusive lock for writing
+        request_exclusive_lock(&self.data_dir, &mut self.active_metadata_file)?;
+
+        if !self.ensure_metadata_file_is_active()?
+            || !ensure_active_metadata_is_valid(&self.data_dir, &mut self.active_metadata_file)?
+        {
+            // The log file has been rotated, so we must try again
+            self.active_metadata_file.unlock()?;
+            return self.batch_upsert_records(records);
+        }
+
+        self.active_data_file.lock_exclusive()?;
+
+        let active_symlink_path = self.data_dir.join(ACTIVE_SYMLINK_FILENAME);
+        let active_target = fs::read_link(active_symlink_path)?;
+        let segment_num = parse_segment_number(&active_target)?;
+
+        debug!("Exclusive lock acquired, appending to log file");
+
+        let mut serialized_data: Vec<u8> = vec![];
+        let mut serialized_metadata: Vec<u8> = vec![];
+        let mut pending_memtable_insertions: Vec<(LogKey, Record)> = vec![];
+        for record in records {
+            // Write the record to the log
+            let serialized = &record.serialize();
+            let record_offset = self.active_data_file.seek(SeekFrom::End(0))?;
+            let record_length = serialized.len() as u64;
+            assert!(record_length > 0);
+
+            serialized_data.extend(serialized);
+
+            let metadata_pos = self.active_metadata_file.seek(SeekFrom::End(0))?;
+            let metadata_index =
+                (metadata_pos - METADATA_FILE_HEADER_SIZE as u64) / METADATA_ROW_LENGTH as u64;
+
+            // Write the record metadata to the metadata file
+            let mut metadata_buf = vec![];
+            metadata_buf.extend(record_offset.to_be_bytes().into_iter());
+            metadata_buf.extend(record_length.to_be_bytes().into_iter());
+
+            assert_eq!(metadata_buf.len(), 16);
+
+            serialized_metadata.extend(metadata_buf);
+
+            let log_key = LogKey::new(segment_num, metadata_index);
+
+            pending_memtable_insertions.push((log_key, record));
+        }
+
+        self.active_data_file.write_all(&serialized_data)?;
+        self.active_metadata_file.write_all(&serialized_metadata)?;
+
+        // Flush and sync data and metadata to disk
+        if self.config.write_durability == WriteDurability::Flush {
+            self.active_data_file.flush()?;
+            self.active_metadata_file.flush()?;
+        } else if self.config.write_durability == WriteDurability::FlushSync {
+            self.active_data_file.flush()?;
+            self.active_data_file.sync_all()?;
+            self.active_metadata_file.flush()?;
+            self.active_metadata_file.sync_all()?;
+        }
+
+        debug!("Records appended to log file, releasing locks");
+
+        // Manually release the locks because the file handles are left open
+        self.active_data_file.unlock()?;
+        self.active_metadata_file.unlock()?;
+
+        for (log_key, record) in pending_memtable_insertions {
+            self.insert_record_to_memtables(log_key, record);
+        }
+
+        Ok(())
+    }
+
+    pub fn batch_find_by_records<'a>(
+        &mut self,
+        field: &R::Field,
+        values: impl Iterator<Item = &'a Value>,
+    ) -> DBResult<Vec<(usize, Record)>> {
+        let field_type = self.get_field_type(field).ok_or(DBError::ValidationError(
+            "Field not found in schema".to_owned(),
+        ))?;
+
+        let indexables = values
+            .map(|value| {
+                if type_check(&value, &field_type) {
+                    value.as_indexable().ok_or(DBError::ValidationError(
+                        "Queried value must be indexable".to_owned(),
+                    ))
+                } else {
+                    Err(DBError::ValidationError(format!(
+                        "Queried value {:?} does not match key type: {:?}",
+                        value, field_type
+                    )))
+                }
+            })
+            .collect::<DBResult<Vec<IndexableValue>>>()?;
+
+        // Otherwise, continue with querying secondary indexes.
+        debug!(
+            "Finding all records with fields {:?} = {:?}",
+            field, indexables
+        );
+
+        if self.config.read_consistency == ReadConsistency::Strong {
+            self.refresh_indexes()?;
+        }
+
+        let log_key_batches = indexables
+            .into_iter()
+            .map(|query_key| {
+                if field == &self.config.primary_key {
+                    let opt = self.primary_memtable.get(&query_key);
+                    let log_keys = match opt {
+                        Some(log_key) => vec![log_key],
+                        None => vec![],
+                    };
+                    Ok(log_keys)
+                } else {
+                    let smemtable_index = match get_secondary_memtable_index_by_field(
+                        &self.config.secondary_keys,
+                        field,
+                    ) {
+                        Some(index) => index,
+                        None => {
+                            return Err(DBError::ValidationError(
+                                "Cannot find_by by non-indexed key".to_owned(),
+                            ))
+                        }
+                    };
+
+                    let log_keys = self.secondary_memtables[smemtable_index]
+                        .find_by(&query_key)
+                        .into_iter()
+                        .collect();
+                    Ok(log_keys)
+                }
+            })
+            .collect::<DBResult<Vec<Vec<&LogKey>>>>()?;
+
+        debug!("Found log keys in memtable: {:?}", log_key_batches);
+
+        let mut tagged = vec![];
+        let mut tag: usize = 0;
+        for batch in log_key_batches {
+            let mapped = batch.into_iter().map(|log_key| (tag, log_key));
+            tagged.extend(mapped);
+            tag += 1;
+        }
+
+        let tagged_records = self.read_tagged_log_keys(tagged.into_iter())?;
+
+        debug!("Read {} records", tagged_records.len());
+
+        Ok(tagged_records)
+    }
+
+    /// Read records from segment files based on log keys.
+    /// The log keys are accompanied by an integer tag that can be used to identify and group them later.
+    fn read_tagged_log_keys<'a>(
+        &self,
+        log_keys: impl Iterator<Item = (usize, &'a LogKey)>,
+    ) -> DBResult<Vec<(usize, Record)>> {
+        let mut records = vec![];
+        let mut log_keys_map = BTreeMap::new();
+
+        for (tag, log_key) in log_keys {
+            if !log_keys_map.contains_key(&log_key.segment_num()) {
+                log_keys_map.insert(log_key.segment_num(), vec![(tag, log_key.index())]);
+            } else {
+                log_keys_map
+                    .get_mut(&log_key.segment_num())
+                    .unwrap()
+                    .push((tag, log_key.index()));
+            }
+        }
+
+        for (segment_num, mut segment_indexes) in log_keys_map {
+            segment_indexes.sort_unstable();
+
+            let metadata_path = &self.data_dir.join(metadata_filename(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)?;
+
+            let data_path = &self.data_dir.join(metadata_header.uuid.to_string());
+            let mut data_file = READ_MODE.open(&data_path)?;
+
+            data_file.lock_shared()?;
+
+            let header_size = METADATA_FILE_HEADER_SIZE as i64;
+            let row_length = METADATA_ROW_LENGTH as i64;
+            let mut current_metadata_offset = header_size;
+            for (tag, segment_index) in segment_indexes {
+                let new_metadata_offset = header_size + segment_index as i64 * row_length;
+                metadata_file.seek_relative(new_metadata_offset - current_metadata_offset)?;
+
+                let mut metadata_buf = [0; METADATA_ROW_LENGTH];
+                metadata_file.read_exact(&mut metadata_buf)?;
+
+                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());
+                assert!(data_length > 0);
+
+                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);
+                records.push((tag, record));
+
+                current_metadata_offset = new_metadata_offset + row_length;
+            }
+
+            metadata_file.unlock()?;
+            data_file.unlock()?;
+        }
+
+        Ok(records)
+    }
+
+    pub fn range_by_records<B: RangeBounds<Value>>(
+        &mut self,
+        field: &R::Field,
+        range: B,
+    ) -> DBResult<Vec<Record>> {
+        fn range_bound_to_indexable(
+            bound: Bound<&Value>,
+            field_type: &ValueType,
+        ) -> DBResult<Bound<IndexableValue>> {
+            fn convert(value: &Value, field_type: &ValueType) -> DBResult<IndexableValue> {
+                if !type_check(&value, field_type) {
+                    return Err(DBError::ValidationError(format!(
+                        "Queried value does not match type: {:?}",
+                        field_type
+                    )));
+                }
+                value.as_indexable().ok_or(DBError::ValidationError(
+                    "Queried value must be indexable".to_owned(),
+                ))
+            }
+
+            match bound {
+                Bound::Included(value) => convert(value, field_type).map(Bound::Included),
+                Bound::Excluded(value) => convert(value, field_type).map(Bound::Excluded),
+                Bound::Unbounded => Ok(Bound::Unbounded),
+            }
+        }
+
+        let field_type = self.get_field_type(field).ok_or(DBError::ValidationError(
+            "Field not found in schema".to_owned(),
+        ))?;
+
+        let start_indexable = range_bound_to_indexable(range.start_bound(), field_type)?;
+        let end_indexable = range_bound_to_indexable(range.end_bound(), field_type)?;
+
+        let indexable_bounds = OwnedBounds::new(start_indexable, end_indexable);
+
+        if self.config.read_consistency == ReadConsistency::Strong {
+            self.refresh_indexes()?;
+        }
+
+        let log_keys = if field == &self.config.primary_key {
+            self.primary_memtable.range(indexable_bounds)
+        } else {
+            let index = get_secondary_memtable_index_by_field(&self.config.secondary_keys, field)
+                .ok_or_else(|| {
+                DBError::ValidationError("Cannot range_by by non-indexed key".to_owned())
+            })?;
+
+            self.secondary_memtables[index].range(indexable_bounds)
+        };
+
+        let log_key_batches = log_keys.into_iter().map(|log_key| (0, log_key));
+
+        let tagged_records = self.read_tagged_log_keys(log_key_batches);
+
+        Ok(tagged_records?.into_iter().map(|(_, rec)| rec).collect())
+    }
+
+    /// 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 `false` if the file has been rotated and the handle has been reopened, `true` otherwise.
+    fn ensure_metadata_file_is_active(&mut self) -> DBResult<bool> {
+        let active_target = fs::read_link(&self.data_dir.join(ACTIVE_SYMLINK_FILENAME))?;
+        let active_metadata_path = &self.data_dir.join(active_target);
+
+        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 = APPEND_MODE.open(&active_metadata_path)?;
+
+            request_shared_lock(&self.data_dir, &mut metadata_file)?;
+
+            let metadata_header = read_metadata_header(&mut self.active_metadata_file)?;
+
+            validate_metadata_header(&metadata_header)?;
+
+            let data_file_path = &self.data_dir.join(metadata_header.uuid.to_string());
+
+            self.active_metadata_file = metadata_file;
+            self.active_data_file = APPEND_MODE.open(&data_file_path)?;
+
+            return Ok(false);
+        } else {
+            return Ok(true);
+        }
+    }
+
+    pub fn delete_by_field(&mut self, field: &R::Field, value: &Value) -> DBResult<Vec<Record>> {
+        let value_batch = std::iter::once(value);
+        let recs: Vec<Record> = self
+            .batch_find_by_records(field, value_batch)?
+            .into_iter()
+            .map(|(_, mut rec)| {
+                rec.tombstone = true;
+                rec
+            })
+            .collect();
+
+        request_exclusive_lock(&self.data_dir, &mut self.active_metadata_file)?;
+        self.active_data_file.lock_exclusive()?;
+
+        for record in &recs {
+            let record_serialized = record.serialize();
+
+            let offset = self.active_data_file.seek(SeekFrom::End(0))?;
+            let length = record_serialized.len() as u64;
+
+            self.active_data_file.write_all(&record_serialized)?;
+
+            // Flush and sync data to disk
+            if self.config.write_durability == WriteDurability::Flush {
+                self.active_data_file.flush()?;
+            }
+            if self.config.write_durability == WriteDurability::FlushSync {
+                self.active_data_file.flush()?;
+                self.active_data_file.sync_all()?;
+            }
+
+            let mut metadata_entry = vec![];
+            metadata_entry.extend(offset.to_be_bytes().into_iter());
+            metadata_entry.extend(length.to_be_bytes().into_iter());
+
+            self.active_metadata_file.write_all(&metadata_entry)?;
+
+            // 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.remove_record_from_memtables(&record);
+        }
+
+        self.active_metadata_file.unlock()?;
+        self.active_data_file.unlock()?;
+
+        debug!("Records deleted");
+
+        Ok(recs)
+    }
+
+    pub fn do_maintenance_tasks(&mut self) -> DBResult<()> {
+        request_exclusive_lock(&self.data_dir, &mut self.active_metadata_file)?;
+
+        ensure_active_metadata_is_valid(&self.data_dir, &mut self.active_metadata_file)?;
+
+        let metadata_size = self.active_metadata_file.seek(SeekFrom::End(0))?;
+        if metadata_size >= self.config.segment_size as u64 {
+            self.rotate_and_compact()?;
+        }
+
+        self.active_metadata_file.unlock()?;
+
+        Ok(())
+    }
+
+    fn rotate_and_compact(&mut self) -> DBResult<()> {
+        debug!("Active log size exceeds threshold, starting rotation and compaction...");
+
+        self.active_data_file.lock_shared()?;
+        let original_data_len = self.active_data_file.seek(SeekFrom::End(0))?;
+
+        let active_target = fs::read_link(&self.data_dir.join(ACTIVE_SYMLINK_FILENAME))?;
+        let active_num = parse_segment_number(&active_target)?;
+
+        debug!("Reading segment data into a BTreeMap");
+        let mut pk_to_item_map: BTreeMap<&IndexableValue, &Record> = BTreeMap::new();
+        let forward_read_items: Vec<(IndexableValue, Record)> = ForwardLogReader::new(
+            self.active_metadata_file.try_clone()?,
+            self.active_data_file.try_clone()?,
+        )
+        .map(|item| {
+            (
+                item.record
+                    .at(self.primary_key_index)
+                    .as_indexable()
+                    .expect("Primary key was not indexable"),
+                item.record,
+            )
+        })
+        .collect();
+
+        self.active_data_file.unlock()?;
+
+        for (pk, record) in forward_read_items.iter() {
+            pk_to_item_map.insert(pk, record);
+        }
+
+        debug!(
+            "Read {} records, out of which {} were unique",
+            forward_read_items.len(),
+            pk_to_item_map.len()
+        );
+
+        // Create a new log data file and write it
+        debug!("Opening new data file and writing compacted data");
+        let (new_data_uuid, new_data_path) = create_segment_data_file(&self.data_dir)?;
+        let mut new_data_file = APPEND_MODE.open(&new_data_path)?;
+
+        let mut pk_to_data_map = BTreeMap::new();
+        let mut offset = 0u64;
+        for (pk, record) in pk_to_item_map.into_iter() {
+            let serialized = record.serialize();
+            let len = serialized.len() as u64;
+            new_data_file.write_all(&serialized)?;
+
+            pk_to_data_map.insert(pk, (offset, len));
+            offset += len;
+        }
+
+        // Sync the data file to disk.
+        // This is fine to do without consulting WriteDurability because this is a one-off
+        // operation that is not part of the normal write path.
+        new_data_file.flush()?;
+        new_data_file.sync_all()?;
+
+        let final_data_len = new_data_file.seek(io::SeekFrom::End(0))?;
+        debug!(
+            "Wrote compacted data, reduced data size: {} -> {}",
+            original_data_len, final_data_len
+        );
+
+        // Create a new log metadata file and write it
+        debug!("Opening temp metadata file and writing pointers to compacted data file");
+        let temp_metadata_file = tempfile::NamedTempFile::new()?;
+        let temp_metadata_path = temp_metadata_file.as_ref();
+        let mut temp_metadata_file = WRITE_MODE.open(temp_metadata_path)?;
+
+        let metadata_header = MetadataHeader {
+            version: 1,
+            uuid: new_data_uuid,
+        };
+
+        temp_metadata_file.write_all(&metadata_header.serialize())?;
+
+        for (pk, _) in forward_read_items.iter() {
+            let (offset, len) = pk_to_data_map.get(&pk).unwrap();
+
+            let mut metadata_buf = vec![];
+            metadata_buf.extend(offset.to_be_bytes().into_iter());
+            metadata_buf.extend(len.to_be_bytes().into_iter());
+
+            temp_metadata_file.write_all(&metadata_buf)?;
+        }
+
+        // Sync the metadata file to disk, see comment above about sync.
+        temp_metadata_file.flush()?;
+        temp_metadata_file.sync_all()?;
+
+        debug!("Moving temporary files to their final locations");
+        let new_data_path = &self.data_dir.join(new_data_uuid.to_string());
+        let active_metadata_path = &self.data_dir.join(metadata_filename(active_num)); // overwrite active
+
+        fs::rename(&temp_metadata_path, &active_metadata_path)?;
+
+        debug!("Compaction complete, creating new segment");
+
+        let new_segment_num = active_num + 1;
+        let new_metadata_path = self.data_dir.join(metadata_filename(new_segment_num));
+        let mut new_metadata_file = APPEND_MODE.clone().create(true).open(&new_metadata_path)?;
+
+        let new_metadata_header = MetadataHeader {
+            version: 1,
+            uuid: new_data_uuid,
+        };
+
+        new_metadata_file.write_all(&new_metadata_header.serialize())?;
+
+        set_active_segment(&self.data_dir, new_segment_num)?;
+
+        // Old active metadata file should lose lock by RAII, or by
+        // the manual unlock call in the do_maintenance_tasks method.
+
+        self.active_metadata_file = APPEND_MODE.open(&new_metadata_path)?;
+        self.active_data_file = APPEND_MODE.open(&new_data_path)?;
+
+        // The new active log file is not locked by this client so it cannot be touched.
+        debug!(
+            "Active log file {} rotated and compacted, new segment: {}",
+            active_num, new_segment_num
+        );
+
+        Ok(())
+    }
+
+    #[inline]
+    fn get_field_type(&self, field: &R::Field) -> Option<&ValueType> {
+        self.config
+            .fields
+            .iter()
+            .find(|(f, _)| f == field)
+            .map(|(_, t)| t)
+    }
+}