[db] Do in place shrinking of values in Storage #1783

agdb/src/storage.rs

@@ -545,12 +545,31 @@ impl<D: StorageData> Storage<D> {
             self.set_size(record.index, new_size);
             self.truncate(record.end())
         } else {
+            let empty_size = record.size - new_size;
             record.size = new_size;
             self.set_size(record.index, new_size);
-            self.move_to_end(record, new_size)
+
+            if empty_size >= Self::record_serialized_size() {
+                self.data.write(
+                    record.pos + record.index.serialized_size(),
+                    &new_size.serialize(),
+                )?;
+                let empty_pos = record.value_start() + new_size;
+                let empty_data_size = empty_size - Self::record_serialized_size();
+                self.write_empty(empty_pos, empty_data_size)
+            } else {
+                self.move_to_end(record, new_size)
+            }
         }
     }
 
+    fn write_empty(&mut self, pos: u64, data_size: u64) -> Result<(), DbError> {
+        let mut bytes = Vec::with_capacity(Self::record_serialized_size() as usize);
+        bytes.extend(0_u64.serialize());
+        bytes.extend(data_size.serialize());
+        self.data.write(pos, &bytes)
+    }
+
     fn truncate(&mut self, size: u64) -> Result<(), DbError> {
         let current_size = self.len();
 

agdb/src/storage/file_storage.rs

@@ -1451,4 +1451,29 @@ mod tests {
         assert_eq!(vec, value1);
         assert_eq!(storage.version(), 1);
     }
+
+    #[test]
+    fn file_storage_with_non_empty_hole_is_readable() {
+        let test_file = TestFile::new();
+
+        let index1;
+        let index2;
+        let index3;
+
+        {
+            let mut storage = Storage::<FileStorage>::new(test_file.file_name()).unwrap();
+            index1 = storage
+                .insert(&vec![1_i64, 2_i64, 3_i64, 4_i64, 5_i64])
+                .unwrap();
+            index2 = storage.insert(&42_i64).unwrap();
+            index3 = storage.insert(&43_i64).unwrap();
+            storage.resize_value(index1, 16).unwrap();
+        }
+
+        let storage = Storage::<FileStorage>::new(test_file.file_name()).unwrap();
+
+        assert_eq!(storage.value_size(index1).unwrap(), 16);
+        assert_eq!(storage.value::<i64>(index2), Ok(42_i64));
+        assert_eq!(storage.value::<i64>(index3), Ok(43_i64));
+    }
 }

agdb/src/storage/memory_storage.rs

@@ -582,6 +582,83 @@ mod tests {
         );
     }
 
+    #[test]
+    fn shrink_value_in_place_does_not_grow_storage() {
+        let mut storage = Storage::<MemoryStorage>::new("storage").unwrap();
+
+        let index1 = storage
+            .insert(&vec![1_i64, 2_i64, 3_i64, 4_i64, 5_i64])
+            .unwrap();
+        let index2 = storage.insert(&10_i64).unwrap();
+        let index3 = storage.insert(&20_i64).unwrap();
+
+        let size_before = storage.len();
+
+        storage.resize_value(index1, 24).unwrap();
+
+        assert_eq!(storage.len(), size_before);
+        assert_eq!(storage.value_size(index1).unwrap(), 24);
+
+        assert_eq!(storage.value::<i64>(index2), Ok(10_i64));
+        assert_eq!(storage.value::<i64>(index3), Ok(20_i64));
+    }
+
+    #[test]
+    fn shrink_value_at_end_truncates_storage() {
+        let mut storage = Storage::<MemoryStorage>::new("storage").unwrap();
+
+        let index1 = storage
+            .insert(&vec![1_i64, 2_i64, 3_i64, 4_i64, 5_i64])
+            .unwrap();
+
+        let size_before = storage.len();
+
+        storage.resize_value(index1, 24).unwrap();
+
+        assert_eq!(storage.len(), size_before - 24);
+        assert_eq!(storage.value_size(index1).unwrap(), 24);
+    }
+
+    #[test]
+    fn shrink_value_small_reclaim_falls_back_to_move_to_end() {
+        let mut storage = Storage::<MemoryStorage>::new("storage").unwrap();
+
+        let index1 = storage.insert(&vec![1_i64, 2_i64]).unwrap(); // size=24
+        let index2 = storage.insert(&99_i64).unwrap();
+
+        let size_before = storage.len();
+
+        storage.resize_value(index1, 16).unwrap();
+
+        assert!(storage.len() > size_before);
+        assert_eq!(storage.value_size(index1).unwrap(), 16);
+        assert_eq!(storage.value::<i64>(index2), Ok(99_i64));
+    }
+
+    #[test]
+    fn shrink_value_in_place_reclaimed_by_optimize_storage() {
+        let mut storage = Storage::<MemoryStorage>::new("storage").unwrap();
+
+        let index1 = storage
+            .insert(&vec![1_i64, 2_i64, 3_i64, 4_i64, 5_i64])
+            .unwrap();
+        let index2 = storage.insert(&42_i64).unwrap();
+        let index3 = storage.insert(&43_i64).unwrap();
+
+        let size_before = storage.value_size(index1).unwrap();
+        storage.resize_value(index1, 16).unwrap();
+        let reclaimed_size = size_before - 16;
+        let before_optimize = storage.len();
+
+        storage.optimize_storage().unwrap();
+
+        assert_eq!(storage.len(), before_optimize - reclaimed_size);
+
+        assert_eq!(storage.value_size(index1).unwrap(), 16);
+        assert_eq!(storage.value::<i64>(index2), Ok(42_i64));
+        assert_eq!(storage.value::<i64>(index3), Ok(43_i64));
+    }
+
     #[test]
     fn shrink_to_fit() {
         let mut storage = Storage::<MemoryStorage>::new("storage").unwrap();