aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--log_db/src/common.rs64
-rw-r--r--log_db/src/lib.rs78
-rw-r--r--log_db/src/memtable_secondary.rs36
-rw-r--r--log_db/tests/integration.rs62
4 files changed, 201 insertions, 39 deletions
diff --git a/log_db/src/common.rs b/log_db/src/common.rs
index 4be2b39..a4f253c 100644
--- a/log_db/src/common.rs
+++ b/log_db/src/common.rs
@@ -26,6 +26,16 @@ pub const INIT_LOCK_FILENAME: &str = "init_lock";
pub const DEFAULT_READ_BUF_SIZE: usize = 1024 * 1024; // 1 MB
pub const TEST_RESOURCES_DIR: &str = "tests/resources";
+// Serialized value tags
+pub const B_NULL: u8 = 0x0;
+pub const B_INT: u8 = 0x1;
+pub const B_FLOAT: u8 = 0x2;
+pub const B_STRING: u8 = 0x3;
+pub const B_BYTES: u8 = 0x4;
+// Tombstone marker tags
+pub const B_LIVE: u8 = 0x0;
+pub const B_TOMBSTONE: u8 = 0xFF;
+
pub fn metadata_filename(num: u16) -> String {
format!("metadata.{}", num)
}
@@ -42,6 +52,14 @@ pub enum DBError {
IOError(#[from] io::Error),
}
+#[derive(Debug, Error)]
+pub enum LogKeySetError {
+ #[error("log key not found in set")]
+ NotFoundError,
+ #[error("attempted to remove last element of non-empty set")]
+ RemovingLastElementError,
+}
+
/// LogKey is a packed struct that contains:
/// - a log segment number (16 bits)
/// - a log index within the segment (48 bits)
@@ -102,6 +120,10 @@ impl LogKeySet {
self.set.iter()
}
+ pub fn contains(&self, key: &LogKey) -> bool {
+ self.set.contains(key)
+ }
+
/// The number of LogKeys in the set.
pub fn len(&self) -> usize {
self.set.len()
@@ -113,22 +135,16 @@ impl LogKeySet {
}
/// 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> {
+ /// Return `LogKeySetError::RemovingLastElementError` if trying to remove the last element.
+ /// Return `LogKeySetError::NotFoundError` if the key was not found.
+ pub fn remove(&mut self, key: &LogKey) -> Result<(), LogKeySetError> {
if self.set.len() == 1 {
- return Err(io::Error::new(
- io::ErrorKind::InvalidInput,
- "Cannot remove the last element from LogKeySet",
- ));
+ return Err(LogKeySetError::RemovingLastElementError);
}
let removed = self.set.remove(key);
if !removed {
- return Err(io::Error::new(
- io::ErrorKind::NotFound,
- "LogKey not found in LogKeySet",
- ));
+ return Err(LogKeySetError::NotFoundError);
}
assert!(
@@ -309,27 +325,27 @@ impl Value {
pub fn serialize(&self) -> Vec<u8> {
match self {
Value::Null => {
- vec![0] // Tag for Null
+ vec![B_NULL]
}
Value::Int(i) => {
- let mut bytes = vec![1]; // Tag for Int
+ let mut bytes = vec![B_INT];
bytes.extend(&i.to_be_bytes());
bytes
}
Value::Float(f) => {
- let mut bytes = vec![2]; // Tag for Float
+ let mut bytes = vec![B_FLOAT];
bytes.extend(&f.to_be_bytes());
bytes
}
Value::String(s) => {
- let mut bytes = vec![3]; // Tag for String
+ let mut bytes = vec![B_STRING];
let length = s.len() as u64;
bytes.extend(&length.to_be_bytes());
bytes.extend(s.as_bytes());
bytes
}
Value::Bytes(b) => {
- let mut bytes = vec![4]; // Tag for Bytes
+ let mut bytes = vec![B_BYTES];
let length = b.len() as u64;
bytes.extend(&length.to_be_bytes());
bytes.extend(b);
@@ -342,18 +358,18 @@ impl Value {
/// Returns the deserialized Value and the number of bytes consumed.
pub fn deserialize(bytes: &[u8]) -> (Value, usize) {
match bytes[0] {
- 0 => (Value::Null, 1),
- 1 => {
+ B_NULL => (Value::Null, 1),
+ B_INT => {
let mut int_bytes = [0; 8];
int_bytes.copy_from_slice(&bytes[1..1 + 8]);
(Value::Int(i64::from_be_bytes(int_bytes)), 1 + 8)
}
- 2 => {
+ B_FLOAT => {
let mut float_bytes = [0; 8];
float_bytes.copy_from_slice(&bytes[1..1 + 8]);
(Value::Float(f64::from_be_bytes(float_bytes)), 1 + 8)
}
- 3 => {
+ B_STRING => {
let length_bytes = &bytes[1..1 + 8];
let length = u64::from_be_bytes(length_bytes.try_into().unwrap()) as usize;
(
@@ -363,7 +379,7 @@ impl Value {
1 + 8 + length,
)
}
- 4 => {
+ B_BYTES => {
let length_bytes = &bytes[1..1 + 8];
let length = u64::from_be_bytes(length_bytes.try_into().unwrap()) as usize;
(
@@ -395,9 +411,9 @@ impl Record {
let mut bytes = Vec::new();
if self.tombstone {
- bytes.extend(&[0xFF]);
+ bytes.extend(&[B_TOMBSTONE]);
} else {
- bytes.extend(&[0]);
+ bytes.extend(&[B_LIVE]);
}
for value in &self.values {
@@ -409,7 +425,7 @@ impl Record {
pub fn deserialize(bytes: &[u8]) -> Record {
let mut values = Vec::new();
- let tombstone = bytes[0] == 0xFF;
+ let tombstone = bytes[0] == B_TOMBSTONE;
let mut start = 1;
while start < bytes.len() {
diff --git a/log_db/src/lib.rs b/log_db/src/lib.rs
index 194b6d6..3c3f7bb 100644
--- a/log_db/src/lib.rs
+++ b/log_db/src/lib.rs
@@ -355,19 +355,20 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
fn remove_record_from_memtables(&mut self, record: &Record) {
let pk = record.at(self.primary_key_index).as_indexable().unwrap();
- self.primary_memtable.remove(&pk);
- 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();
+ 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);
+ secondary_memtable.remove(&sk, &plk);
+ }
}
}
@@ -651,9 +652,58 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
}
}
- /// Delete records by a field value.
- pub fn delete(&mut self, field: &Field, value: &Value) -> Result<u64, DBError> {
- todo!()
+ /// Delete record by primary key.
+ pub fn delete(&mut self, pk: &Value) -> Result<Option<Record>, DBError> {
+ let record = match self.get(pk)? {
+ Some(record) => record,
+ None => return Ok(None),
+ };
+
+ // The Record interface does not allow manually setting the tombstone flag,
+ // so we have to serialize the record and manually set the first byte to B_TOMBSTONE.
+ let mut record_serialized = vec![B_TOMBSTONE];
+ record_serialized.extend(&record.serialize()[1..]);
+
+ request_exclusive_lock(&self.data_dir, &mut self.active_metadata_file)?;
+ self.active_data_file.lock_exclusive()?;
+
+ 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().iter());
+ metadata_entry.extend(length.to_be_bytes().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!("Record deleted, returning from delete");
+
+ Ok(Some(record))
}
/// Check if there are any pending tasks and do them. Tasks include:
diff --git a/log_db/src/memtable_secondary.rs b/log_db/src/memtable_secondary.rs
index 4479516..a7a9a8a 100644
--- a/log_db/src/memtable_secondary.rs
+++ b/log_db/src/memtable_secondary.rs
@@ -38,7 +38,41 @@ impl SecondaryMemtable {
}
}
- pub fn remove(&mut self, key: &IndexableValue) -> Option<LogKeySet> {
+ // Remove all log keys associated with the given key
+ pub fn remove_all(&mut self, key: &IndexableValue) -> Option<LogKeySet> {
self.records.remove(key)
}
+
+ // Remove a single log key associated with the given key. Returns `true`
+ // if the log key existed and was removed, `false` otherwise.
+ pub fn remove(&mut self, key: &IndexableValue, log_key: &LogKey) -> bool {
+ let set = match self.records.get_mut(key) {
+ Some(set) => set,
+ None => return false,
+ };
+ if set.len() == 1 && set.contains(log_key) {
+ self.records.remove(key);
+ true
+ } else {
+ return match set.remove(log_key) {
+ Ok(_) => true,
+ Err(LogKeySetError::NotFoundError) => false,
+ Err(e) => panic!("{:?}", e),
+ };
+ }
+ }
+
+ // Remove all log keys associated with the given log key
+ // Note: This is a linear time operation, prefer using the `remove` method
+ // if you know the secondary key associated with the log key.
+ pub fn scan_remove(&mut self, log_key: &LogKey) -> u64 {
+ let mut removed = 0;
+ self.records.iter_mut().for_each(|(_, set)| {
+ if let Ok(_) = set.remove(&log_key) {
+ removed += 1;
+ }
+ });
+
+ removed
+ }
}
diff --git a/log_db/tests/integration.rs b/log_db/tests/integration.rs
index bb4b8ac..daf2745 100644
--- a/log_db/tests/integration.rs
+++ b/log_db/tests/integration.rs
@@ -148,6 +148,24 @@ fn test_upsert_and_get() {
}
#[test]
+fn test_get_nonexistant() {
+ let data_dir = tmp_dir();
+ let mut db = DB::configure()
+ .data_dir(&data_dir)
+ .fields(&[
+ (Field::Id, ValueType::int()),
+ (Field::Name, ValueType::string().nullable()),
+ (Field::Data, ValueType::bytes()),
+ ])
+ .primary_key(Field::Id)
+ .initialize()
+ .expect("Failed to initialize DB instance");
+
+ let result = db.get(&Value::Int(0)).unwrap();
+ assert!(result.is_none());
+}
+
+#[test]
fn test_upsert_fails_on_null_in_non_nullable_field() {
let data_dir = tmp_dir();
let mut db = DB::configure()
@@ -396,3 +414,47 @@ fn test_log_is_rotated_when_capacity_reached() {
// 3rd segment should not exist (note negation)
assert!(!data_dir_path.join("metadata").with_extension("3").exists());
}
+
+#[test]
+fn test_delete() {
+ let data_dir = tmp_dir();
+ let mut db = DB::configure()
+ .data_dir(&data_dir)
+ .fields(&[
+ (Field::Id, ValueType::int()),
+ (Field::Name, ValueType::string()),
+ (Field::Data, ValueType::bytes()),
+ ])
+ .primary_key(Field::Id)
+ .secondary_keys(&[Field::Name])
+ .initialize()
+ .expect("Failed to initialize DB instance");
+
+ // Insert some records
+ let record0 = Record::from(&[
+ Value::Int(0),
+ Value::String("John".to_string()),
+ Value::Bytes(vec![3, 4, 5]),
+ ]);
+ db.upsert(&record0).unwrap();
+
+ let record1 = Record::from(&[
+ Value::Int(1),
+ Value::String("John".to_string()),
+ Value::Bytes(vec![1, 2, 3]),
+ ]);
+ db.upsert(&record1).unwrap();
+
+ db.delete(&Value::Int(0)).unwrap();
+
+ // Check that the record is deleted
+ assert!(db.get(&Value::Int(0)).unwrap().is_none());
+
+ // Check that the secondary index is updated
+ assert_eq!(
+ db.find_all(&Field::Name, &Value::String("John".to_string()))
+ .unwrap()
+ .len(),
+ 1
+ );
+}

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