| // Copyright 2021 The Fuchsia Authors. All rights reserved. |
| // Use of this source code is governed by a BSD-style license that can be |
| // found in the LICENSE file. |
| |
| use { |
| super::extent_record::{ExtentKey, ExtentValue}, |
| super::object_record::{AttributeKey, ObjectKey, ObjectKeyData, ObjectValue}, |
| crate::lsm_tree::{ |
| merge::{ |
| ItemOp::{Discard, Keep, Replace}, |
| MergeLayerIterator, MergeResult, |
| }, |
| types::Item, |
| }, |
| }; |
| |
| fn merge_extents( |
| object_id: u64, |
| attribute_id: u64, |
| left: &MergeLayerIterator<'_, ObjectKey, ObjectValue>, |
| right: &MergeLayerIterator<'_, ObjectKey, ObjectValue>, |
| left_key: &ExtentKey, |
| right_key: &ExtentKey, |
| left_value: &ExtentValue, |
| right_value: &ExtentValue, |
| ) -> MergeResult<ObjectKey, ObjectValue> { |
| // For now, we don't support/expect two extents with the same key in one layer. |
| // One reason you can't merge deleted extents in non-adjacent layers is because you |
| // can't decide which layer the merge should end up in. |
| // |
| // Consider this scenario: |
| // |X-X-X| |
| // |a-a-a| |
| // |X-X-X| |
| // If you merge the deleted extents here, you might end up with: |
| // |X-X-X-X-X-X| |
| // |a-a-a| |
| // which is clearly incorrect. |
| debug_assert!(right.layer_index != left.layer_index); |
| |
| if let (ExtentValue::None, ExtentValue::None) = (left_value, right_value) { |
| if (left.layer_index as i32 - right.layer_index as i32).abs() == 1 { |
| // Two deletions in adjacent layers can be merged. |
| return merge_deleted_extents( |
| object_id, |
| attribute_id, |
| left_key, |
| right_key, |
| std::cmp::min(left.sequence(), right.sequence()), |
| ); |
| } |
| } |
| |
| if left_key.range.end <= right_key.range.start { |
| // Extents don't overlap. |
| return MergeResult::EmitLeft; |
| } |
| |
| // The start of the left extent is <= the start of the right extent, due to merge key ordering. |
| // |
| // One of the extents has to win. The way we break this tie is by picking the extent from the |
| // newest layer (i.e. the layer with the lowest index). |
| // |
| // Generally, we'll be doing the following: |
| // |
| // Old |----------| |
| // New |----------| |
| // |
| // Turns into |
| // |
| // Emit |------| |
| // Old |--| |
| // New |----------| |
| |
| if right.layer_index < left.layer_index { |
| // Right layer is newer. |
| debug_assert!(left_key.range.start < right_key.range.start); |
| return MergeResult::Other { |
| emit: Some(Item::new_with_sequence( |
| ObjectKey::extent( |
| object_id, |
| attribute_id, |
| left_key.range.start..right_key.range.start, |
| ), |
| ObjectValue::Extent(left_value.shrunk( |
| left_key.range.end - left_key.range.start, |
| right_key.range.start - left_key.range.start, |
| )), |
| std::cmp::min(left.sequence(), right.sequence()), |
| )), |
| left: Replace(Item::new_with_sequence( |
| ObjectKey::extent( |
| object_id, |
| attribute_id, |
| right_key.range.start..left_key.range.end, |
| ), |
| ObjectValue::Extent(left_value.offset_by( |
| right_key.range.start - left_key.range.start, |
| left_key.range.end - left_key.range.start, |
| )), |
| std::cmp::min(left.sequence(), right.sequence()), |
| )), |
| right: Keep, |
| }; |
| } |
| // Left layer is newer. |
| if left_key.range.end >= right_key.range.end { |
| // The left key entirely contains the right key. |
| return MergeResult::Other { emit: None, left: Keep, right: Discard }; |
| } |
| MergeResult::Other { |
| emit: None, |
| left: Keep, |
| right: Replace(Item::new_with_sequence( |
| ObjectKey::extent(object_id, attribute_id, left_key.range.end..right_key.range.end), |
| ObjectValue::Extent(right_value.offset_by( |
| left_key.range.end - right_key.range.start, |
| right_key.range.end - right_key.range.start, |
| )), |
| std::cmp::min(left.sequence(), right.sequence()), |
| )), |
| } |
| } |
| |
| // Assumes that the two extents to be merged are on adjacent layers (i.e. layers N, N+1). |
| fn merge_deleted_extents( |
| object_id: u64, |
| attribute_id: u64, |
| left_key: &ExtentKey, |
| right_key: &ExtentKey, |
| sequence: u64, |
| ) -> MergeResult<ObjectKey, ObjectValue> { |
| if left_key.range.end < right_key.range.start { |
| // The extents are not adjacent or overlapping. |
| return MergeResult::EmitLeft; |
| } |
| // Both of these are deleted extents which are either adjacent or overlapping, which means |
| // we can coalece the records. |
| if left_key.range.end >= right_key.range.end { |
| // The left deletion eclipses the right, so just keep the left. |
| return MergeResult::Other { emit: None, left: Keep, right: Discard }; |
| } |
| MergeResult::Other { |
| emit: None, |
| left: Discard, |
| right: Replace(Item::new_with_sequence( |
| ObjectKey::extent(object_id, attribute_id, left_key.range.start..right_key.range.end), |
| ObjectValue::deleted_extent(), |
| sequence, |
| )), |
| } |
| } |
| |
| /// Merge function for items in the object store. |
| /// |
| /// The most interesting behaviour in this merge function is how extents are handled. Since extents |
| /// can overlap and replace one another, the merge function generally builds up the most |
| /// recent view of the extents in the tree, so that the output of a full merge contains no |
| /// overlapping extents. You can imagine looking down at the extents from the top-most layer. |
| /// |
| /// A brief example: |
| /// |
| /// Layer 0 |a-a-a-a| |b-b-b| |
| /// Layer 1 |c-c-c-c-c| |
| /// Layer 2 |d-d-d-d| |
| /// |
| /// Merged |a-a-a-a|c| |b-b-b|d-d-d| |
| /// |
| /// Adjacent or overlapping extent deletions in two adjacent layers can be merged into single |
| /// records (since they do not have a physical offset, so there's no need to keep the physical |
| /// extents contiguous). We can't merge deletions from non-adjacent layers, since that would |
| /// cause issues in situations like this: |
| /// |
| /// Layer 0 |X-X-X| |
| /// Layer 1 |a-a-a-a-a-a| |
| /// Layer 2 |X-X-X| |
| /// |
| /// Merging the two deletions in layers 0 and 2 would either result in the middle extent being |
| /// fully occluded or not at all (depending on whether we replaced on the left or right layer). |
| pub fn merge( |
| left: &MergeLayerIterator<'_, ObjectKey, ObjectValue>, |
| right: &MergeLayerIterator<'_, ObjectKey, ObjectValue>, |
| ) -> MergeResult<ObjectKey, ObjectValue> { |
| if left.key().object_id != right.key().object_id { |
| return MergeResult::EmitLeft; |
| } |
| match (left.key(), right.key(), left.value(), right.value()) { |
| ( |
| ObjectKey { |
| object_id, |
| data: ObjectKeyData::Attribute(left_attr_id, AttributeKey::Extent(left_extent_key)), |
| }, |
| ObjectKey { |
| object_id: _, |
| data: |
| ObjectKeyData::Attribute(right_attr_id, AttributeKey::Extent(right_extent_key)), |
| }, |
| ObjectValue::Extent(left_extent), |
| ObjectValue::Extent(right_extent), |
| ) if left_attr_id == right_attr_id => { |
| return merge_extents( |
| *object_id, |
| *left_attr_id, |
| left, |
| right, |
| left_extent_key, |
| right_extent_key, |
| left_extent, |
| right_extent, |
| ); |
| } |
| // Tombstones (ObjectKeyData::Object) compare before others, so always appear on left. |
| (ObjectKey { data: ObjectKeyData::Object, .. }, _, ObjectValue::None, _) => { |
| debug_assert!(left.layer_index < right.layer_index); |
| MergeResult::Other { emit: None, left: Keep, right: Discard } |
| } |
| // Note that identical keys are sorted by layer_index, so left is always newer. |
| (left_key, right_key, _, _) if left_key == right_key => { |
| debug_assert!(left.layer_index < right.layer_index); |
| MergeResult::Other { emit: None, left: Keep, right: Discard } |
| } |
| _ => MergeResult::EmitLeft, |
| } |
| } |
| |
| #[cfg(test)] |
| mod tests { |
| use { |
| super::merge, |
| crate::{ |
| lsm_tree::{ |
| types::{Item, LayerIterator, MergeableKey, Value}, |
| LSMTree, |
| }, |
| object_store::extent_record::{Checksums, ExtentValue}, |
| object_store::object_record::{AttributeKey, ObjectKey, ObjectValue, Timestamp}, |
| }, |
| anyhow::Error, |
| fuchsia_async as fasync, |
| std::ops::Bound, |
| }; |
| |
| async fn test_merge<K: MergeableKey, V: Value + PartialEq>( |
| tree: &LSMTree<K, V>, |
| layer0: &[Item<K, V>], |
| layer1: &[Item<K, V>], |
| expected: &[Item<K, V>], |
| ) { |
| for item in layer1 { |
| tree.insert(item.clone()).await; |
| } |
| tree.seal().await; |
| for item in layer0 { |
| tree.insert(item.clone()).await; |
| } |
| let layer_set = tree.layer_set(); |
| let mut merger = layer_set.merger(); |
| let mut iter = merger.seek(Bound::Unbounded).await.expect("seek failed"); |
| for e in expected { |
| assert_eq!(iter.get().expect("get failed"), e.as_item_ref()); |
| iter.advance().await.expect("advance failed"); |
| } |
| assert!(iter.get().is_none()); |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_merge_extents_non_overlapping() -> Result<(), Error> { |
| let object_id = 0; |
| let attr_id = 0; |
| let tree = LSMTree::new(merge); |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 0..512), |
| ObjectValue::Extent(ExtentValue::new(0)), |
| )) |
| .await; |
| tree.seal().await; |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 512..1024), |
| ObjectValue::Extent(ExtentValue::new(16384)), |
| )) |
| .await; |
| |
| let layer_set = tree.layer_set(); |
| let mut merger = layer_set.merger(); |
| let mut iter = merger.seek(std::ops::Bound::Unbounded).await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..512)); |
| iter.advance().await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 512..1024)); |
| iter.advance().await?; |
| assert!(iter.get().is_none()); |
| Ok(()) |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_merge_extents_rewrite_right() -> Result<(), Error> { |
| let object_id = 0; |
| let attr_id = 0; |
| let tree = LSMTree::new(merge); |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 0..1024), |
| ObjectValue::Extent(ExtentValue::new(0)), |
| )) |
| .await; |
| tree.seal().await; |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 512..1024), |
| ObjectValue::Extent(ExtentValue::new(16384)), |
| )) |
| .await; |
| |
| let layer_set = tree.layer_set(); |
| let mut merger = layer_set.merger(); |
| let mut iter = merger.seek(std::ops::Bound::Unbounded).await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..512)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::Extent(ExtentValue::new(0))); |
| iter.advance().await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 512..1024)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::Extent(ExtentValue::new(16384))); |
| iter.advance().await?; |
| assert!(iter.get().is_none()); |
| Ok(()) |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_merge_extents_rewrite_left() -> Result<(), Error> { |
| let object_id = 0; |
| let attr_id = 0; |
| let tree = LSMTree::new(merge); |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 0..1024), |
| ObjectValue::Extent(ExtentValue::with_checksum(0, Checksums::Fletcher(vec![1, 2]))), |
| )) |
| .await; |
| tree.seal().await; |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 0..512), |
| ObjectValue::Extent(ExtentValue::with_checksum(16384, Checksums::Fletcher(vec![3]))), |
| )) |
| .await; |
| |
| let layer_set = tree.layer_set(); |
| let mut merger = layer_set.merger(); |
| let mut iter = merger.seek(std::ops::Bound::Unbounded).await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..512)); |
| assert_eq!( |
| iter.get().unwrap().value, |
| &ObjectValue::Extent(ExtentValue::with_checksum(16384, Checksums::Fletcher(vec![3]))) |
| ); |
| iter.advance().await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 512..1024)); |
| assert_eq!( |
| iter.get().unwrap().value, |
| &ObjectValue::Extent(ExtentValue::with_checksum(512, Checksums::Fletcher(vec![2]))) |
| ); |
| iter.advance().await?; |
| assert!(iter.get().is_none()); |
| Ok(()) |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_merge_extents_rewrite_middle() -> Result<(), Error> { |
| let object_id = 0; |
| let attr_id = 0; |
| let tree = LSMTree::new(merge); |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 0..2048), |
| ObjectValue::Extent(ExtentValue::with_checksum( |
| 0, |
| Checksums::Fletcher(vec![1, 2, 3, 4]), |
| )), |
| )) |
| .await; |
| tree.seal().await; |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 1024..1536), |
| ObjectValue::Extent(ExtentValue::with_checksum(16384, Checksums::Fletcher(vec![5]))), |
| )) |
| .await; |
| |
| let layer_set = tree.layer_set(); |
| let mut merger = layer_set.merger(); |
| let mut iter = merger.seek(std::ops::Bound::Unbounded).await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..1024)); |
| assert_eq!( |
| iter.get().unwrap().value, |
| &ObjectValue::Extent(ExtentValue::with_checksum(0, Checksums::Fletcher(vec![1, 2]))) |
| ); |
| iter.advance().await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 1024..1536)); |
| assert_eq!( |
| iter.get().unwrap().value, |
| &ObjectValue::Extent(ExtentValue::with_checksum(16384, Checksums::Fletcher(vec![5]))) |
| ); |
| iter.advance().await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 1536..2048)); |
| assert_eq!( |
| iter.get().unwrap().value, |
| &ObjectValue::Extent(ExtentValue::with_checksum(1536, Checksums::Fletcher(vec![4]))) |
| ); |
| iter.advance().await?; |
| assert!(iter.get().is_none()); |
| Ok(()) |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_merge_extents_rewrite_eclipses() -> Result<(), Error> { |
| let object_id = 0; |
| let attr_id = 0; |
| let tree = LSMTree::new(merge); |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 1024..1536), |
| ObjectValue::Extent(ExtentValue::new(0)), |
| )) |
| .await; |
| tree.seal().await; |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 0..2048), |
| ObjectValue::Extent(ExtentValue::new(16384)), |
| )) |
| .await; |
| |
| let layer_set = tree.layer_set(); |
| let mut merger = layer_set.merger(); |
| let mut iter = merger.seek(std::ops::Bound::Unbounded).await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..2048)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::Extent(ExtentValue::new(16384))); |
| iter.advance().await?; |
| assert!(iter.get().is_none()); |
| Ok(()) |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_merge_extents_delete_left() -> Result<(), Error> { |
| let object_id = 0; |
| let attr_id = 0; |
| let tree = LSMTree::new(merge); |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 0..1024), |
| ObjectValue::Extent(ExtentValue::new(0)), |
| )) |
| .await; |
| tree.seal().await; |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 0..512), |
| ObjectValue::deleted_extent(), |
| )) |
| .await; |
| |
| let layer_set = tree.layer_set(); |
| let mut merger = layer_set.merger(); |
| let mut iter = merger.seek(std::ops::Bound::Unbounded).await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..512)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent()); |
| iter.advance().await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 512..1024)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::Extent(ExtentValue::new(512))); |
| iter.advance().await?; |
| assert!(iter.get().is_none()); |
| Ok(()) |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_merge_extents_delete_right() -> Result<(), Error> { |
| let object_id = 0; |
| let attr_id = 0; |
| let tree = LSMTree::new(merge); |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 0..1024), |
| ObjectValue::Extent(ExtentValue::new(0)), |
| )) |
| .await; |
| tree.seal().await; |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 512..1024), |
| ObjectValue::deleted_extent(), |
| )) |
| .await; |
| |
| let layer_set = tree.layer_set(); |
| let mut merger = layer_set.merger(); |
| let mut iter = merger.seek(std::ops::Bound::Unbounded).await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..512)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::Extent(ExtentValue::new(0))); |
| iter.advance().await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 512..1024)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent()); |
| iter.advance().await?; |
| assert!(iter.get().is_none()); |
| Ok(()) |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_merge_extents_delete_middle() -> Result<(), Error> { |
| let object_id = 0; |
| let attr_id = 0; |
| let tree = LSMTree::new(merge); |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 0..2048), |
| ObjectValue::Extent(ExtentValue::new(0)), |
| )) |
| .await; |
| tree.seal().await; |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 1024..1536), |
| ObjectValue::deleted_extent(), |
| )) |
| .await; |
| |
| let layer_set = tree.layer_set(); |
| let mut merger = layer_set.merger(); |
| let mut iter = merger.seek(std::ops::Bound::Unbounded).await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..1024)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::Extent(ExtentValue::new(0))); |
| iter.advance().await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 1024..1536)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent()); |
| iter.advance().await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 1536..2048)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::Extent(ExtentValue::new(1536))); |
| iter.advance().await?; |
| assert!(iter.get().is_none()); |
| Ok(()) |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_merge_extents_delete_eclipses() -> Result<(), Error> { |
| let object_id = 0; |
| let attr_id = 0; |
| let tree = LSMTree::new(merge); |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 1024..1536), |
| ObjectValue::Extent(ExtentValue::new(0)), |
| )) |
| .await; |
| tree.seal().await; |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 0..2048), |
| ObjectValue::deleted_extent(), |
| )) |
| .await; |
| |
| let layer_set = tree.layer_set(); |
| let mut merger = layer_set.merger(); |
| let mut iter = merger.seek(std::ops::Bound::Unbounded).await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..2048)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent()); |
| iter.advance().await?; |
| assert!(iter.get().is_none()); |
| Ok(()) |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_merge_deleted_extents_new_layer_joins_two_deletions() -> Result<(), Error> { |
| // Old layer: [----] [----] |
| // New layer: [----] |
| // Merged: [--------------] |
| let object_id = 0; |
| let attr_id = 0; |
| let tree = LSMTree::new(merge); |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 0..512), |
| ObjectValue::deleted_extent(), |
| )) |
| .await; |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 1024..1536), |
| ObjectValue::deleted_extent(), |
| )) |
| .await; |
| tree.seal().await; |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 512..1024), |
| ObjectValue::deleted_extent(), |
| )) |
| .await; |
| tree.seal().await; |
| |
| let layer_set = tree.layer_set(); |
| let mut merger = layer_set.merger(); |
| let mut iter = merger.seek(std::ops::Bound::Unbounded).await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..1536)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent()); |
| iter.advance().await?; |
| assert!(iter.get().is_none()); |
| Ok(()) |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_merge_deleted_extents_new_layer_joined_by_old_deletion() -> Result<(), Error> { |
| // Old layer: [----] |
| // New layer: [----] [----] |
| // Merged: [--------------] |
| let object_id = 0; |
| let attr_id = 0; |
| let tree = LSMTree::new(merge); |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 512..1024), |
| ObjectValue::deleted_extent(), |
| )) |
| .await; |
| tree.seal().await; |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 0..512), |
| ObjectValue::deleted_extent(), |
| )) |
| .await; |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 1024..1536), |
| ObjectValue::deleted_extent(), |
| )) |
| .await; |
| tree.seal().await; |
| |
| let layer_set = tree.layer_set(); |
| let mut merger = layer_set.merger(); |
| let mut iter = merger.seek(std::ops::Bound::Unbounded).await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..1536)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent()); |
| iter.advance().await?; |
| assert!(iter.get().is_none()); |
| Ok(()) |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_merge_deleted_extents_overlapping_newest_on_right() -> Result<(), Error> { |
| let object_id = 0; |
| let attr_id = 0; |
| let tree = LSMTree::new(merge); |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 0..1024), |
| ObjectValue::deleted_extent(), |
| )) |
| .await; |
| tree.seal().await; |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 512..1536), |
| ObjectValue::deleted_extent(), |
| )) |
| .await; |
| |
| let layer_set = tree.layer_set(); |
| let mut merger = layer_set.merger(); |
| let mut iter = merger.seek(std::ops::Bound::Unbounded).await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..1536)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent()); |
| iter.advance().await?; |
| assert!(iter.get().is_none()); |
| Ok(()) |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_merge_deleted_extents_overlapping_newest_on_left() -> Result<(), Error> { |
| let object_id = 0; |
| let attr_id = 0; |
| let tree = LSMTree::new(merge); |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 512..1536), |
| ObjectValue::deleted_extent(), |
| )) |
| .await; |
| tree.seal().await; |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 0..1024), |
| ObjectValue::deleted_extent(), |
| )) |
| .await; |
| |
| let layer_set = tree.layer_set(); |
| let mut merger = layer_set.merger(); |
| let mut iter = merger.seek(std::ops::Bound::Unbounded).await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..1536)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent()); |
| iter.advance().await?; |
| assert!(iter.get().is_none()); |
| Ok(()) |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_merge_deleted_extents_new_layer_contained_in_old() -> Result<(), Error> { |
| // Old layer: [--------------] |
| // New layer: [----] |
| // Merged: [--------------] |
| let object_id = 0; |
| let attr_id = 0; |
| let tree = LSMTree::new(merge); |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 0..1536), |
| ObjectValue::deleted_extent(), |
| )) |
| .await; |
| tree.seal().await; |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 512..1024), |
| ObjectValue::deleted_extent(), |
| )) |
| .await; |
| tree.seal().await; |
| |
| let layer_set = tree.layer_set(); |
| let mut merger = layer_set.merger(); |
| let mut iter = merger.seek(std::ops::Bound::Unbounded).await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..1536)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent()); |
| iter.advance().await?; |
| assert!(iter.get().is_none()); |
| Ok(()) |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_merge_deleted_extents_new_layer_eclipses_old() -> Result<(), Error> { |
| // Old layer: [----] |
| // New layer: [--------------] |
| // Merged: [--------------] |
| let object_id = 0; |
| let attr_id = 0; |
| let tree = LSMTree::new(merge); |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 512..1024), |
| ObjectValue::deleted_extent(), |
| )) |
| .await; |
| tree.seal().await; |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 0..1536), |
| ObjectValue::deleted_extent(), |
| )) |
| .await; |
| tree.seal().await; |
| |
| let layer_set = tree.layer_set(); |
| let mut merger = layer_set.merger(); |
| let mut iter = merger.seek(std::ops::Bound::Unbounded).await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..1536)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent()); |
| iter.advance().await?; |
| assert!(iter.get().is_none()); |
| Ok(()) |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_merge_deleted_extents_does_not_coalesce_if_not_adjacent_layers( |
| ) -> Result<(), Error> { |
| // Layer 0: [XXXXX] |
| // Layer 1: [--------------] |
| // Layer 2: [XXXXXXXX] |
| // Merged: [XXXXX|--------] |
| let object_id = 0; |
| let attr_id = 0; |
| let tree = LSMTree::<ObjectKey, ObjectValue>::new(merge); |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 512..1024), |
| ObjectValue::deleted_extent(), |
| )) |
| .await; |
| tree.seal().await; |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 0..1024), |
| ObjectValue::Extent(ExtentValue::new(0)), |
| )) |
| .await; |
| tree.seal().await; |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 0..512), |
| ObjectValue::deleted_extent(), |
| )) |
| .await; |
| |
| let layer_set = tree.layer_set(); |
| let mut merger = layer_set.merger(); |
| let mut iter = merger.seek(std::ops::Bound::Unbounded).await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..512)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent()); |
| iter.advance().await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 512..1024)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::Extent(ExtentValue::new(512))); |
| iter.advance().await?; |
| assert!(iter.get().is_none()); |
| Ok(()) |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_merge_deleted_extents_does_not_coalesce_if_not_adjacent_deletions( |
| ) -> Result<(), Error> { |
| // Layer 0: [XXXXX|--------] |
| // Layer 1: [XXXXX] |
| // Merged: [XXXXX|--------] |
| let object_id = 0; |
| let attr_id = 0; |
| let tree = LSMTree::<ObjectKey, ObjectValue>::new(merge); |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 1024..1536), |
| ObjectValue::Extent(ExtentValue::new(0)), |
| )) |
| .await; |
| tree.seal().await; |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 0..512), |
| ObjectValue::deleted_extent(), |
| )) |
| .await; |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 512..1536), |
| ObjectValue::Extent(ExtentValue::new(0)), |
| )) |
| .await; |
| |
| let layer_set = tree.layer_set(); |
| let mut merger = layer_set.merger(); |
| let mut iter = merger.seek(std::ops::Bound::Unbounded).await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..512)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent()); |
| iter.advance().await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 512..1536)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::Extent(ExtentValue::new(0))); |
| iter.advance().await?; |
| assert!(iter.get().is_none()); |
| Ok(()) |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_merge_deleted_extent_into_overwrites_extents() -> Result<(), Error> { |
| let object_id = 0; |
| let attr_id = 0; |
| let tree = LSMTree::<ObjectKey, ObjectValue>::new(merge); |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 0..1024), |
| ObjectValue::Extent(ExtentValue::new(0)), |
| )) |
| .await; |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 1024..2048), |
| ObjectValue::Extent(ExtentValue::new(16384)), |
| )) |
| .await; |
| let key = ObjectKey::extent(object_id, attr_id, 512..1536); |
| tree.merge_into( |
| Item::new(key.clone(), ObjectValue::deleted_extent()), |
| &key.key_for_merge_into(), |
| ) |
| .await; |
| |
| let layer_set = tree.layer_set(); |
| let mut merger = layer_set.merger(); |
| let mut iter = merger.seek(std::ops::Bound::Unbounded).await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..512)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::Extent(ExtentValue::new(0))); |
| iter.advance().await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 512..1536)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent()); |
| iter.advance().await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 1536..2048)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::Extent(ExtentValue::new(16896))); |
| iter.advance().await?; |
| assert!(iter.get().is_none()); |
| Ok(()) |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_merge_deleted_extent_into_merges_with_other_deletions() -> Result<(), Error> { |
| let object_id = 0; |
| let attr_id = 0; |
| let tree = LSMTree::<ObjectKey, ObjectValue>::new(merge); |
| |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 0..1024), |
| ObjectValue::deleted_extent(), |
| )) |
| .await; |
| tree.insert(Item::new( |
| ObjectKey::extent(object_id, attr_id, 1024..2048), |
| ObjectValue::deleted_extent(), |
| )) |
| .await; |
| |
| let key = ObjectKey::extent(object_id, attr_id, 512..1536); |
| tree.merge_into( |
| Item::new(key.clone(), ObjectValue::deleted_extent()), |
| &key.key_for_merge_into(), |
| ) |
| .await; |
| |
| let layer_set = tree.layer_set(); |
| let mut merger = layer_set.merger(); |
| let mut iter = merger.seek(std::ops::Bound::Unbounded).await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..2048)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent()); |
| iter.advance().await?; |
| assert!(iter.get().is_none()); |
| Ok(()) |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_merge_size_records() { |
| let left = |
| &[Item::new(ObjectKey::attribute(1, 0, AttributeKey::Size), ObjectValue::attribute(5))]; |
| let right = &[Item::new( |
| ObjectKey::attribute(1, 0, AttributeKey::Size), |
| ObjectValue::attribute(10), |
| )]; |
| let tree = LSMTree::new(merge); |
| test_merge(&tree, left, right, left).await; |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_different_attributes_not_merged() { |
| let left = |
| Item::new(ObjectKey::attribute(1, 0, AttributeKey::Size), ObjectValue::attribute(5)); |
| let right = |
| Item::new(ObjectKey::attribute(1, 1, AttributeKey::Size), ObjectValue::attribute(10)); |
| let tree = LSMTree::new(merge); |
| test_merge(&tree, &[left.clone()], &[right.clone()], &[left, right]).await; |
| |
| let left = |
| Item::new(ObjectKey::extent(1, 0, 0..100), ObjectValue::Extent(ExtentValue::new(0))); |
| let right = |
| Item::new(ObjectKey::extent(1, 1, 0..100), ObjectValue::Extent(ExtentValue::new(1))); |
| let tree = LSMTree::new(merge); |
| test_merge(&tree, &[left.clone()], &[right.clone()], &[left, right]).await; |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_tombstone_discards_all_other_records() { |
| let tombstone = Item::new(ObjectKey::object(1), ObjectValue::None); |
| let other_object = Item::new( |
| ObjectKey::object(2), |
| ObjectValue::file(1, 0, Timestamp::default(), Timestamp::default()), |
| ); |
| let tree = LSMTree::new(merge); |
| test_merge( |
| &tree, |
| &[tombstone.clone()], |
| &[ |
| Item::new( |
| ObjectKey::object(1), |
| ObjectValue::file(1, 100, Timestamp::default(), Timestamp::default()), |
| ), |
| Item::new( |
| ObjectKey::attribute(1, 0, AttributeKey::Size), |
| ObjectValue::attribute(100), |
| ), |
| other_object.clone(), |
| ], |
| &[tombstone, other_object], |
| ) |
| .await; |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_merge_preserves_sequences() -> Result<(), Error> { |
| let object_id = 0; |
| let attr_id = 0; |
| let tree = LSMTree::<ObjectKey, ObjectValue>::new(merge); |
| |
| tree.insert(Item { |
| key: ObjectKey::extent(object_id, attr_id, 0..1024), |
| value: ObjectValue::Extent(ExtentValue::new(0u64)), |
| sequence: 1u64, |
| }) |
| .await; |
| tree.seal().await; |
| tree.insert(Item { |
| key: ObjectKey::extent(object_id, attr_id, 0..512), |
| value: ObjectValue::deleted_extent(), |
| sequence: 2u64, |
| }) |
| .await; |
| tree.insert(Item { |
| key: ObjectKey::extent(object_id, attr_id, 1536..2048), |
| value: ObjectValue::Extent(ExtentValue::new(1536)), |
| sequence: 3u64, |
| }) |
| .await; |
| tree.insert(Item { |
| key: ObjectKey::extent(object_id, attr_id, 768..1024), |
| value: ObjectValue::Extent(ExtentValue::new(12345)), |
| sequence: 4u64, |
| }) |
| .await; |
| |
| let layer_set = tree.layer_set(); |
| let mut merger = layer_set.merger(); |
| let mut iter = merger.seek(std::ops::Bound::Unbounded).await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..512)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent()); |
| assert_eq!(iter.get().unwrap().sequence, 2u64); |
| iter.advance().await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 512..768)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::Extent(ExtentValue::new(512))); |
| assert_eq!(iter.get().unwrap().sequence, 1u64); |
| iter.advance().await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 768..1024)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::Extent(ExtentValue::new(12345))); |
| assert_eq!(iter.get().unwrap().sequence, 4u64); |
| iter.advance().await?; |
| assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 1536..2048)); |
| assert_eq!(iter.get().unwrap().value, &ObjectValue::Extent(ExtentValue::new(1536))); |
| assert_eq!(iter.get().unwrap().sequence, 3u64); |
| iter.advance().await?; |
| assert!(iter.get().is_none()); |
| Ok(()) |
| } |
| } |
