fxfs/object_store/
merge.rs

1// Copyright 2021 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use super::object_record::{
6    AttributeId, AttributeKey, ObjectKey, ObjectKeyData, ObjectValue, ProjectProperty,
7};
8use super::{Extent, ExtentValue};
9use crate::log::*;
10use crate::lsm_tree::merge::ItemOp::{Discard, Keep, Replace};
11use crate::lsm_tree::merge::{MergeLayerIterator, MergeResult};
12use crate::lsm_tree::types::Item;
13
14fn merge_extents(
15    object_id: u64,
16    attribute_id: AttributeId,
17    left: &MergeLayerIterator<'_, ObjectKey, ObjectValue>,
18    right: &MergeLayerIterator<'_, ObjectKey, ObjectValue>,
19    left_key: &Extent,
20    right_key: &Extent,
21    left_value: &ExtentValue,
22    right_value: &ExtentValue,
23) -> MergeResult<ObjectKey, ObjectValue> {
24    // For now, we don't support/expect two extents with the same key in one layer.
25    // One reason you can't merge deleted extents in non-adjacent layers is because you
26    // can't decide which layer the merge should end up in.
27    //
28    // Consider this scenario:
29    //   |X-X-X|
30    //      |a-a-a|
31    //         |X-X-X|
32    // If you merge the deleted extents here, you might end up with:
33    //   |X-X-X-X-X-X|
34    //      |a-a-a|
35    // which is clearly incorrect.
36    debug_assert!(right.layer_index != left.layer_index);
37
38    if let (ExtentValue::None, ExtentValue::None) = (left_value, right_value) {
39        if (left.layer_index as i32 - right.layer_index as i32).abs() == 1 {
40            // Two deletions in adjacent layers can be merged.
41            return merge_deleted_extents(object_id, attribute_id, left_key, right_key);
42        }
43    }
44
45    if left_key.end <= right_key.start {
46        // Extents don't overlap.
47        return MergeResult::EmitLeft;
48    }
49
50    // The start of the left extent is <= the start of the right extent, due to merge key ordering.
51    //
52    // One of the extents has to win. The way we break this tie is by picking the extent from the
53    // newest layer (i.e. the layer with the lowest index).
54    //
55    // Generally, we'll be doing the following:
56    //
57    //  Old  |----------|
58    //  New          |----------|
59    //
60    // Turns into
61    //
62    //  Emit  |------|
63    //  Old          |--|
64    //  New          |----------|
65
66    if right.layer_index < left.layer_index {
67        // Right layer is newer.
68        debug_assert!(left_key.start < right_key.start);
69        return MergeResult::Other {
70            emit: Some(
71                Item::new(
72                    ObjectKey::extent(object_id, attribute_id, left_key.start..right_key.start),
73                    ObjectValue::Extent(
74                        left_value.shrunk(
75                            left_key.end - left_key.start,
76                            right_key.start - left_key.start,
77                        ),
78                    ),
79                )
80                .boxed(),
81            ),
82            left: Replace(
83                Item::new(
84                    ObjectKey::extent(object_id, attribute_id, right_key.start..left_key.end),
85                    ObjectValue::Extent(left_value.offset_by(
86                        right_key.start - left_key.start,
87                        left_key.end - left_key.start,
88                    )),
89                )
90                .boxed(),
91            ),
92            right: Keep,
93        };
94    }
95    // Left layer is newer.
96    if left_key.end >= right_key.end {
97        // The left key entirely contains the right key.
98        return MergeResult::Other { emit: None, left: Keep, right: Discard };
99    }
100    MergeResult::Other {
101        emit: None,
102        left: Keep,
103        right: Replace(
104            Item::new(
105                ObjectKey::extent(object_id, attribute_id, left_key.end..right_key.end),
106                ObjectValue::Extent(
107                    right_value
108                        .offset_by(left_key.end - right_key.start, right_key.end - right_key.start),
109                ),
110            )
111            .boxed(),
112        ),
113    }
114}
115
116// Assumes that the two extents to be merged are on adjacent layers (i.e. layers N, N+1).
117fn merge_deleted_extents(
118    object_id: u64,
119    attribute_id: AttributeId,
120    left_key: &Extent,
121    right_key: &Extent,
122) -> MergeResult<ObjectKey, ObjectValue> {
123    if left_key.end < right_key.start {
124        // The extents are not adjacent or overlapping.
125        return MergeResult::EmitLeft;
126    }
127    // Both of these are deleted extents which are either adjacent or overlapping, which means
128    // we can coalesce the records.
129    if left_key.end >= right_key.end {
130        // The left deletion eclipses the right, so just keep the left.
131        return MergeResult::Other { emit: None, left: Keep, right: Discard };
132    }
133    MergeResult::Other {
134        emit: None,
135        left: Discard,
136        right: Replace(Box::new(Item::new(
137            ObjectKey::extent(object_id, attribute_id, left_key.start..right_key.end),
138            ObjectValue::deleted_extent(),
139        ))),
140    }
141}
142
143/// Merge function for items in the object store.
144///
145/// The most interesting behaviour in this merge function is how extents are handled. Since extents
146/// can overlap and replace one another, the merge function generally builds up the most
147/// recent view of the extents in the tree, so that the output of a full merge contains no
148/// overlapping extents. You can imagine looking down at the extents from the top-most layer.
149///
150/// A brief example:
151///
152/// Layer 0   |a-a-a-a|     |b-b-b|
153/// Layer 1   |c-c-c-c-c|
154/// Layer 2                     |d-d-d-d|
155///
156/// Merged    |a-a-a-a|c|   |b-b-b|d-d-d|
157///
158/// Adjacent or overlapping extent deletions in two adjacent layers can be merged into single
159/// records (since they do not have a physical offset, so there's no need to keep the physical
160/// extents contiguous). We can't merge deletions from non-adjacent layers, since that would
161/// cause issues in situations like this:
162///
163/// Layer 0         |X-X-X|
164/// Layer 1   |a-a-a-a-a-a|
165/// Layer 2   |X-X-X|
166///
167/// Merging the two deletions in layers 0 and 2 would either result in the middle extent being
168/// fully occluded or not at all (depending on whether we replaced on the left or right layer).
169pub fn merge(
170    left: &MergeLayerIterator<'_, ObjectKey, ObjectValue>,
171    right: &MergeLayerIterator<'_, ObjectKey, ObjectValue>,
172) -> MergeResult<ObjectKey, ObjectValue> {
173    if left.key().object_id != right.key().object_id {
174        return MergeResult::EmitLeft;
175    }
176    match (left.key(), right.key(), left.value(), right.value()) {
177        (
178            ObjectKey {
179                object_id,
180                data: ObjectKeyData::Attribute(left_attr_id, AttributeKey::Extent(left_extent_key)),
181            },
182            ObjectKey {
183                object_id: _,
184                data:
185                    ObjectKeyData::Attribute(right_attr_id, AttributeKey::Extent(right_extent_key)),
186            },
187            ObjectValue::Extent(left_extent),
188            ObjectValue::Extent(right_extent),
189        ) if left_attr_id == right_attr_id => {
190            return merge_extents(
191                *object_id,
192                *left_attr_id,
193                left,
194                right,
195                left_extent_key,
196                right_extent_key,
197                left_extent,
198                right_extent,
199            );
200        }
201        (
202            ObjectKey {
203                object_id: _,
204                data:
205                    ObjectKeyData::Project {
206                        project_id: left_project_id,
207                        property: ProjectProperty::Usage,
208                    },
209            },
210            ObjectKey {
211                object_id: _,
212                data:
213                    ObjectKeyData::Project {
214                        project_id: right_project_id,
215                        property: ProjectProperty::Usage,
216                    },
217            },
218            ObjectValue::BytesAndNodes { bytes: left_bytes, nodes: left_nodes },
219            ObjectValue::BytesAndNodes { bytes: right_bytes, nodes: right_nodes },
220        ) if left_project_id == right_project_id => {
221            let bytes = left_bytes + right_bytes;
222            let nodes = left_nodes + right_nodes;
223            // Tombstone the tracking when it goes to zero.
224            match (bytes, nodes) {
225                (0, 0) => MergeResult::Other { emit: None, left: Discard, right: Discard },
226                _ => MergeResult::Other {
227                    emit: None,
228                    left: Discard,
229                    right: Replace(
230                        Item::new(right.key().clone(), ObjectValue::BytesAndNodes { bytes, nodes })
231                            .boxed(),
232                    ),
233                },
234            }
235        }
236        // Tombstones (ObjectKeyData::Object) compare before others, so always appear on left.
237        (ObjectKey { data: ObjectKeyData::Object, .. }, _, ObjectValue::None, _) => {
238            if left.layer_index > right.layer_index {
239                warn!("Detected inconsistency: record has been inserted after tombstone");
240                MergeResult::EmitLeft
241            } else {
242                MergeResult::Other { emit: None, left: Keep, right: Discard }
243            }
244        }
245        // Note that identical keys are sorted by layer_index, so left is always newer.
246        (left_key, right_key, _, _) if left_key == right_key => {
247            debug_assert!(left.layer_index < right.layer_index);
248            MergeResult::Other { emit: None, left: Keep, right: Discard }
249        }
250        _ => MergeResult::EmitLeft,
251    }
252}
253
254#[cfg(test)]
255mod tests {
256    use super::merge;
257    use crate::checksum::Checksums;
258    use crate::lsm_tree::types::{Item, LayerIterator, MergeableKey, Value};
259    use crate::lsm_tree::{LSMTree, Query};
260    use crate::object_store::extent::MIN_BLOCK_SIZE;
261    use crate::object_store::extent_record::ExtentValue;
262    use crate::object_store::object_record::{AttributeKey, ObjectKey, ObjectValue, Timestamp};
263    use crate::object_store::{AttributeId, ProjectId, VOLUME_DATA_KEY_ID};
264    use anyhow::Error;
265
266    async fn test_merge<K: MergeableKey, V: Value + PartialEq>(
267        tree: &LSMTree<K, V>,
268        layer0: &[Item<K, V>],
269        layer1: &[Item<K, V>],
270        expected: &[Item<K, V>],
271    ) {
272        for item in layer1 {
273            tree.insert(item.clone()).expect("insert error");
274        }
275        tree.seal();
276        for item in layer0 {
277            tree.insert(item.clone()).expect("insert error");
278        }
279        let layer_set = tree.layer_set();
280        let mut merger = layer_set.merger();
281        let mut iter = merger.query(Query::FullScan).await.expect("seek failed");
282        for e in expected {
283            assert_eq!(iter.get().expect("get failed"), e.as_item_ref());
284            iter.advance().await.expect("advance failed");
285        }
286        assert!(iter.get().is_none());
287    }
288
289    #[fuchsia::test]
290    async fn test_merge_extents_non_overlapping() -> Result<(), Error> {
291        let object_id = 0;
292        let attr_id = AttributeId::TEST_ID;
293        let tree = LSMTree::new(merge, None);
294
295        tree.insert(Item::new(
296            ObjectKey::extent(object_id, attr_id, 0..512),
297            ObjectValue::Extent(ExtentValue::new_raw(0, VOLUME_DATA_KEY_ID)),
298        ))
299        .expect("insert error");
300        tree.seal();
301
302        tree.insert(Item::new(
303            ObjectKey::extent(object_id, attr_id, 512..1024),
304            ObjectValue::Extent(ExtentValue::new_raw(16384, VOLUME_DATA_KEY_ID)),
305        ))
306        .expect("insert error");
307
308        let layer_set = tree.layer_set();
309        let mut merger = layer_set.merger();
310        let mut iter = merger.query(Query::FullScan).await?;
311        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..512));
312        iter.advance().await?;
313        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 512..1024));
314        iter.advance().await?;
315        assert!(iter.get().is_none());
316        Ok(())
317    }
318
319    #[fuchsia::test]
320    async fn test_merge_extents_rewrite_right() -> Result<(), Error> {
321        let object_id = 0;
322        let attr_id = AttributeId::TEST_ID;
323        let tree = LSMTree::new(merge, None);
324
325        tree.insert(Item::new(
326            ObjectKey::extent(object_id, attr_id, 0..1024),
327            ObjectValue::Extent(ExtentValue::new_raw(0, VOLUME_DATA_KEY_ID)),
328        ))
329        .expect("insert error");
330        tree.seal();
331
332        tree.insert(Item::new(
333            ObjectKey::extent(object_id, attr_id, 512..1024),
334            ObjectValue::Extent(ExtentValue::new_raw(16384, VOLUME_DATA_KEY_ID)),
335        ))
336        .expect("insert error");
337
338        let layer_set = tree.layer_set();
339        let mut merger = layer_set.merger();
340        let mut iter = merger.query(Query::FullScan).await?;
341        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..512));
342        assert_eq!(
343            iter.get().unwrap().value,
344            &ObjectValue::Extent(ExtentValue::new_raw(0, VOLUME_DATA_KEY_ID))
345        );
346        iter.advance().await?;
347        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 512..1024));
348        assert_eq!(
349            iter.get().unwrap().value,
350            &ObjectValue::Extent(ExtentValue::new_raw(16384, VOLUME_DATA_KEY_ID))
351        );
352        iter.advance().await?;
353        assert!(iter.get().is_none());
354        Ok(())
355    }
356
357    #[fuchsia::test]
358    async fn test_merge_extents_rewrite_left() -> Result<(), Error> {
359        let object_id = 0;
360        let attr_id = AttributeId::TEST_ID;
361        let tree = LSMTree::new(merge, None);
362
363        tree.insert(Item::new(
364            ObjectKey::extent(object_id, attr_id, 0..1024),
365            ObjectValue::Extent(ExtentValue::with_checksum(
366                0,
367                Checksums::fletcher(vec![1, 2]),
368                VOLUME_DATA_KEY_ID,
369            )),
370        ))
371        .expect("insert error");
372        tree.seal();
373
374        tree.insert(Item::new(
375            ObjectKey::extent(object_id, attr_id, 0..512),
376            ObjectValue::Extent(ExtentValue::with_checksum(
377                16384,
378                Checksums::fletcher(vec![3]),
379                VOLUME_DATA_KEY_ID,
380            )),
381        ))
382        .expect("insert error");
383
384        let layer_set = tree.layer_set();
385        let mut merger = layer_set.merger();
386        let mut iter = merger.query(Query::FullScan).await?;
387        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..512));
388        assert_eq!(
389            iter.get().unwrap().value,
390            &ObjectValue::Extent(ExtentValue::with_checksum(
391                16384,
392                Checksums::fletcher(vec![3]),
393                VOLUME_DATA_KEY_ID
394            ))
395        );
396        iter.advance().await?;
397        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 512..1024));
398        assert_eq!(
399            iter.get().unwrap().value,
400            &ObjectValue::Extent(ExtentValue::with_checksum(
401                512,
402                Checksums::fletcher(vec![2]),
403                VOLUME_DATA_KEY_ID
404            ))
405        );
406        iter.advance().await?;
407        assert!(iter.get().is_none());
408        Ok(())
409    }
410
411    #[fuchsia::test]
412    async fn test_merge_extents_rewrite_middle() -> Result<(), Error> {
413        let object_id = 0;
414        let attr_id = AttributeId::TEST_ID;
415        let tree = LSMTree::new(merge, None);
416
417        tree.insert(Item::new(
418            ObjectKey::extent(object_id, attr_id, 0..2048),
419            ObjectValue::Extent(ExtentValue::with_checksum(
420                0,
421                Checksums::fletcher(vec![1, 2, 3, 4]),
422                VOLUME_DATA_KEY_ID,
423            )),
424        ))
425        .expect("insert error");
426        tree.seal();
427
428        tree.insert(Item::new(
429            ObjectKey::extent(object_id, attr_id, 1024..1536),
430            ObjectValue::Extent(ExtentValue::with_checksum(
431                16384,
432                Checksums::fletcher(vec![5]),
433                VOLUME_DATA_KEY_ID,
434            )),
435        ))
436        .expect("insert error");
437
438        let layer_set = tree.layer_set();
439        let mut merger = layer_set.merger();
440        let mut iter = merger.query(Query::FullScan).await?;
441        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..1024));
442        assert_eq!(
443            iter.get().unwrap().value,
444            &ObjectValue::Extent(ExtentValue::with_checksum(
445                0,
446                Checksums::fletcher(vec![1, 2]),
447                VOLUME_DATA_KEY_ID
448            ))
449        );
450        iter.advance().await?;
451        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 1024..1536));
452        assert_eq!(
453            iter.get().unwrap().value,
454            &ObjectValue::Extent(ExtentValue::with_checksum(
455                16384,
456                Checksums::fletcher(vec![5]),
457                VOLUME_DATA_KEY_ID
458            ))
459        );
460        iter.advance().await?;
461        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 1536..2048));
462        assert_eq!(
463            iter.get().unwrap().value,
464            &ObjectValue::Extent(ExtentValue::with_checksum(
465                1536,
466                Checksums::fletcher(vec![4]),
467                VOLUME_DATA_KEY_ID
468            ))
469        );
470        iter.advance().await?;
471        assert!(iter.get().is_none());
472        Ok(())
473    }
474
475    #[fuchsia::test]
476    async fn test_merge_extents_rewrite_eclipses() -> Result<(), Error> {
477        let object_id = 0;
478        let attr_id = AttributeId::TEST_ID;
479        let tree = LSMTree::new(merge, None);
480
481        tree.insert(Item::new(
482            ObjectKey::extent(object_id, attr_id, 1024..1536),
483            ObjectValue::Extent(ExtentValue::new_raw(0, VOLUME_DATA_KEY_ID)),
484        ))
485        .expect("insert error");
486        tree.seal();
487
488        tree.insert(Item::new(
489            ObjectKey::extent(object_id, attr_id, 0..2048),
490            ObjectValue::Extent(ExtentValue::new_raw(16384, VOLUME_DATA_KEY_ID)),
491        ))
492        .expect("insert error");
493
494        let layer_set = tree.layer_set();
495        let mut merger = layer_set.merger();
496        let mut iter = merger.query(Query::FullScan).await?;
497        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..2048));
498        assert_eq!(
499            iter.get().unwrap().value,
500            &ObjectValue::Extent(ExtentValue::new_raw(16384, VOLUME_DATA_KEY_ID))
501        );
502        iter.advance().await?;
503        assert!(iter.get().is_none());
504        Ok(())
505    }
506
507    #[fuchsia::test]
508    async fn test_merge_extents_delete_left() -> Result<(), Error> {
509        let object_id = 0;
510        let attr_id = AttributeId::TEST_ID;
511        let tree = LSMTree::new(merge, None);
512
513        tree.insert(Item::new(
514            ObjectKey::extent(object_id, attr_id, 0..1024),
515            ObjectValue::Extent(ExtentValue::new_raw(0, VOLUME_DATA_KEY_ID)),
516        ))
517        .expect("insert error");
518        tree.seal();
519
520        tree.insert(Item::new(
521            ObjectKey::extent(object_id, attr_id, 0..512),
522            ObjectValue::deleted_extent(),
523        ))
524        .expect("insert error");
525
526        let layer_set = tree.layer_set();
527        let mut merger = layer_set.merger();
528        let mut iter = merger.query(Query::FullScan).await?;
529        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..512));
530        assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent());
531        iter.advance().await?;
532        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 512..1024));
533        assert_eq!(
534            iter.get().unwrap().value,
535            &ObjectValue::Extent(ExtentValue::new_raw(512, VOLUME_DATA_KEY_ID))
536        );
537        iter.advance().await?;
538        assert!(iter.get().is_none());
539        Ok(())
540    }
541
542    #[fuchsia::test]
543    async fn test_merge_extents_delete_right() -> Result<(), Error> {
544        let object_id = 0;
545        let attr_id = AttributeId::TEST_ID;
546        let tree = LSMTree::new(merge, None);
547
548        tree.insert(Item::new(
549            ObjectKey::extent(object_id, attr_id, 0..1024),
550            ObjectValue::Extent(ExtentValue::new_raw(0, VOLUME_DATA_KEY_ID)),
551        ))
552        .expect("insert error");
553        tree.seal();
554
555        tree.insert(Item::new(
556            ObjectKey::extent(object_id, attr_id, 512..1024),
557            ObjectValue::deleted_extent(),
558        ))
559        .expect("insert error");
560
561        let layer_set = tree.layer_set();
562        let mut merger = layer_set.merger();
563        let mut iter = merger.query(Query::FullScan).await?;
564        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..512));
565        assert_eq!(
566            iter.get().unwrap().value,
567            &ObjectValue::Extent(ExtentValue::new_raw(0, VOLUME_DATA_KEY_ID))
568        );
569        iter.advance().await?;
570        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 512..1024));
571        assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent());
572        iter.advance().await?;
573        assert!(iter.get().is_none());
574        Ok(())
575    }
576
577    #[fuchsia::test]
578    async fn test_merge_extents_delete_middle() -> Result<(), Error> {
579        let object_id = 0;
580        let attr_id = AttributeId::TEST_ID;
581        let tree = LSMTree::new(merge, None);
582
583        tree.insert(Item::new(
584            ObjectKey::extent(object_id, attr_id, 0..2048),
585            ObjectValue::Extent(ExtentValue::new_raw(0, VOLUME_DATA_KEY_ID)),
586        ))
587        .expect("insert error");
588        tree.seal();
589
590        tree.insert(Item::new(
591            ObjectKey::extent(object_id, attr_id, 1024..1536),
592            ObjectValue::deleted_extent(),
593        ))
594        .expect("insert error");
595
596        let layer_set = tree.layer_set();
597        let mut merger = layer_set.merger();
598        let mut iter = merger.query(Query::FullScan).await?;
599        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..1024));
600        assert_eq!(
601            iter.get().unwrap().value,
602            &ObjectValue::Extent(ExtentValue::new_raw(0, VOLUME_DATA_KEY_ID))
603        );
604        iter.advance().await?;
605        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 1024..1536));
606        assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent());
607        iter.advance().await?;
608        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 1536..2048));
609        assert_eq!(
610            iter.get().unwrap().value,
611            &ObjectValue::Extent(ExtentValue::new_raw(1536, VOLUME_DATA_KEY_ID))
612        );
613        iter.advance().await?;
614        assert!(iter.get().is_none());
615        Ok(())
616    }
617
618    #[fuchsia::test]
619    async fn test_merge_extents_delete_eclipses() -> Result<(), Error> {
620        let object_id = 0;
621        let attr_id = AttributeId::TEST_ID;
622        let tree = LSMTree::new(merge, None);
623
624        tree.insert(Item::new(
625            ObjectKey::extent(object_id, attr_id, 1024..1536),
626            ObjectValue::Extent(ExtentValue::new_raw(0, VOLUME_DATA_KEY_ID)),
627        ))
628        .expect("insert error");
629        tree.seal();
630
631        tree.insert(Item::new(
632            ObjectKey::extent(object_id, attr_id, 0..2048),
633            ObjectValue::deleted_extent(),
634        ))
635        .expect("insert error");
636
637        let layer_set = tree.layer_set();
638        let mut merger = layer_set.merger();
639        let mut iter = merger.query(Query::FullScan).await?;
640        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..2048));
641        assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent());
642        iter.advance().await?;
643        assert!(iter.get().is_none());
644        Ok(())
645    }
646
647    #[fuchsia::test]
648    async fn test_merge_deleted_extents_new_layer_joins_two_deletions() -> Result<(), Error> {
649        // Old layer:  [----]    [----]
650        // New layer:       [----]
651        // Merged:     [--------------]
652        let object_id = 0;
653        let attr_id = AttributeId::TEST_ID;
654        let tree = LSMTree::new(merge, None);
655
656        tree.insert(Item::new(
657            ObjectKey::extent(object_id, attr_id, 0..512),
658            ObjectValue::deleted_extent(),
659        ))
660        .expect("insert error");
661        tree.insert(Item::new(
662            ObjectKey::extent(object_id, attr_id, 1024..1536),
663            ObjectValue::deleted_extent(),
664        ))
665        .expect("insert error");
666        tree.seal();
667
668        tree.insert(Item::new(
669            ObjectKey::extent(object_id, attr_id, 512..1024),
670            ObjectValue::deleted_extent(),
671        ))
672        .expect("insert error");
673        tree.seal();
674
675        let layer_set = tree.layer_set();
676        let mut merger = layer_set.merger();
677        let mut iter = merger.query(Query::FullScan).await?;
678        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..1536));
679        assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent());
680        iter.advance().await?;
681        assert!(iter.get().is_none());
682        Ok(())
683    }
684
685    #[fuchsia::test]
686    async fn test_merge_deleted_extents_new_layer_joined_by_old_deletion() -> Result<(), Error> {
687        // Old layer:       [----]
688        // New layer:  [----]    [----]
689        // Merged:     [--------------]
690        let object_id = 0;
691        let attr_id = AttributeId::TEST_ID;
692        let tree = LSMTree::new(merge, None);
693
694        tree.insert(Item::new(
695            ObjectKey::extent(object_id, attr_id, 512..1024),
696            ObjectValue::deleted_extent(),
697        ))
698        .expect("insert error");
699        tree.seal();
700
701        tree.insert(Item::new(
702            ObjectKey::extent(object_id, attr_id, 0..512),
703            ObjectValue::deleted_extent(),
704        ))
705        .expect("insert error");
706        tree.insert(Item::new(
707            ObjectKey::extent(object_id, attr_id, 1024..1536),
708            ObjectValue::deleted_extent(),
709        ))
710        .expect("insert error");
711        tree.seal();
712
713        let layer_set = tree.layer_set();
714        let mut merger = layer_set.merger();
715        let mut iter = merger.query(Query::FullScan).await?;
716        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..1536));
717        assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent());
718        iter.advance().await?;
719        assert!(iter.get().is_none());
720        Ok(())
721    }
722
723    #[fuchsia::test]
724    async fn test_merge_deleted_extents_overlapping_newest_on_right() -> Result<(), Error> {
725        let object_id = 0;
726        let attr_id = AttributeId::TEST_ID;
727        let tree = LSMTree::new(merge, None);
728
729        tree.insert(Item::new(
730            ObjectKey::extent(object_id, attr_id, 0..1024),
731            ObjectValue::deleted_extent(),
732        ))
733        .expect("insert error");
734        tree.seal();
735
736        tree.insert(Item::new(
737            ObjectKey::extent(object_id, attr_id, 512..1536),
738            ObjectValue::deleted_extent(),
739        ))
740        .expect("insert error");
741
742        let layer_set = tree.layer_set();
743        let mut merger = layer_set.merger();
744        let mut iter = merger.query(Query::FullScan).await?;
745        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..1536));
746        assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent());
747        iter.advance().await?;
748        assert!(iter.get().is_none());
749        Ok(())
750    }
751
752    #[fuchsia::test]
753    async fn test_merge_deleted_extents_overlapping_newest_on_left() -> Result<(), Error> {
754        let object_id = 0;
755        let attr_id = AttributeId::TEST_ID;
756        let tree = LSMTree::new(merge, None);
757
758        tree.insert(Item::new(
759            ObjectKey::extent(object_id, attr_id, 512..1536),
760            ObjectValue::deleted_extent(),
761        ))
762        .expect("insert error");
763        tree.seal();
764        tree.insert(Item::new(
765            ObjectKey::extent(object_id, attr_id, 0..1024),
766            ObjectValue::deleted_extent(),
767        ))
768        .expect("insert error");
769
770        let layer_set = tree.layer_set();
771        let mut merger = layer_set.merger();
772        let mut iter = merger.query(Query::FullScan).await?;
773        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..1536));
774        assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent());
775        iter.advance().await?;
776        assert!(iter.get().is_none());
777        Ok(())
778    }
779
780    #[fuchsia::test]
781    async fn test_merge_deleted_extents_new_layer_contained_in_old() -> Result<(), Error> {
782        // Old layer:  [--------------]
783        // New layer:       [----]
784        // Merged:     [--------------]
785        let object_id = 0;
786        let attr_id = AttributeId::TEST_ID;
787        let tree = LSMTree::new(merge, None);
788
789        tree.insert(Item::new(
790            ObjectKey::extent(object_id, attr_id, 0..1536),
791            ObjectValue::deleted_extent(),
792        ))
793        .expect("insert error");
794        tree.seal();
795
796        tree.insert(Item::new(
797            ObjectKey::extent(object_id, attr_id, 512..1024),
798            ObjectValue::deleted_extent(),
799        ))
800        .expect("insert error");
801        tree.seal();
802
803        let layer_set = tree.layer_set();
804        let mut merger = layer_set.merger();
805        let mut iter = merger.query(Query::FullScan).await?;
806        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..1536));
807        assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent());
808        iter.advance().await?;
809        assert!(iter.get().is_none());
810        Ok(())
811    }
812
813    #[fuchsia::test]
814    async fn test_merge_deleted_extents_new_layer_eclipses_old() -> Result<(), Error> {
815        // Old layer:       [----]
816        // New layer:  [--------------]
817        // Merged:     [--------------]
818        let object_id = 0;
819        let attr_id = AttributeId::TEST_ID;
820        let tree = LSMTree::new(merge, None);
821
822        tree.insert(Item::new(
823            ObjectKey::extent(object_id, attr_id, 512..1024),
824            ObjectValue::deleted_extent(),
825        ))
826        .expect("insert error");
827        tree.seal();
828
829        tree.insert(Item::new(
830            ObjectKey::extent(object_id, attr_id, 0..1536),
831            ObjectValue::deleted_extent(),
832        ))
833        .expect("insert error");
834        tree.seal();
835
836        let layer_set = tree.layer_set();
837        let mut merger = layer_set.merger();
838        let mut iter = merger.query(Query::FullScan).await?;
839        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..1536));
840        assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent());
841        iter.advance().await?;
842        assert!(iter.get().is_none());
843        Ok(())
844    }
845
846    #[fuchsia::test]
847    async fn test_merge_deleted_extents_does_not_coalesce_if_not_adjacent_layers()
848    -> Result<(), Error> {
849        // Layer 0:  [XXXXX]
850        // Layer 1:  [--------------]
851        // Layer 2:        [XXXXXXXX]
852        //  Merged:  [XXXXX|--------]
853        let object_id = 0;
854        let attr_id = AttributeId::TEST_ID;
855        let tree = LSMTree::<ObjectKey, ObjectValue>::new(merge, None);
856
857        tree.insert(Item::new(
858            ObjectKey::extent(object_id, attr_id, 512..1024),
859            ObjectValue::deleted_extent(),
860        ))
861        .expect("insert error");
862        tree.seal();
863
864        tree.insert(Item::new(
865            ObjectKey::extent(object_id, attr_id, 0..1024),
866            ObjectValue::Extent(ExtentValue::new_raw(0, VOLUME_DATA_KEY_ID)),
867        ))
868        .expect("insert error");
869        tree.seal();
870
871        tree.insert(Item::new(
872            ObjectKey::extent(object_id, attr_id, 0..512),
873            ObjectValue::deleted_extent(),
874        ))
875        .expect("insert error");
876
877        let layer_set = tree.layer_set();
878        let mut merger = layer_set.merger();
879        let mut iter = merger.query(Query::FullScan).await?;
880        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..512));
881        assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent());
882        iter.advance().await?;
883        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 512..1024));
884        assert_eq!(
885            iter.get().unwrap().value,
886            &ObjectValue::Extent(ExtentValue::new_raw(512, VOLUME_DATA_KEY_ID))
887        );
888        iter.advance().await?;
889        assert!(iter.get().is_none());
890        Ok(())
891    }
892
893    #[fuchsia::test]
894    async fn test_merge_deleted_extents_does_not_coalesce_if_not_adjacent_deletions()
895    -> Result<(), Error> {
896        // Layer 0:  [XXXXX|--------]
897        // Layer 1:           [XXXXX]
898        //  Merged:  [XXXXX|--------]
899        let object_id = 0;
900        let attr_id = AttributeId::TEST_ID;
901        let tree = LSMTree::<ObjectKey, ObjectValue>::new(merge, None);
902
903        tree.insert(Item::new(
904            ObjectKey::extent(object_id, attr_id, 1024..1536),
905            ObjectValue::Extent(ExtentValue::new_raw(0, VOLUME_DATA_KEY_ID)),
906        ))
907        .expect("insert error");
908        tree.seal();
909
910        tree.insert(Item::new(
911            ObjectKey::extent(object_id, attr_id, 0..512),
912            ObjectValue::deleted_extent(),
913        ))
914        .expect("insert error");
915        tree.insert(Item::new(
916            ObjectKey::extent(object_id, attr_id, 512..1536),
917            ObjectValue::Extent(ExtentValue::new_raw(0, VOLUME_DATA_KEY_ID)),
918        ))
919        .expect("insert error");
920
921        let layer_set = tree.layer_set();
922        let mut merger = layer_set.merger();
923        let mut iter = merger.query(Query::FullScan).await?;
924        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..512));
925        assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent());
926        iter.advance().await?;
927        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 512..1536));
928        assert_eq!(
929            iter.get().unwrap().value,
930            &ObjectValue::Extent(ExtentValue::new_raw(0, VOLUME_DATA_KEY_ID))
931        );
932        iter.advance().await?;
933        assert!(iter.get().is_none());
934        Ok(())
935    }
936
937    #[fuchsia::test]
938    async fn test_merge_deleted_extent_into_overwrites_extents() -> Result<(), Error> {
939        let object_id = 0;
940        let attr_id = AttributeId::TEST_ID;
941        let tree = LSMTree::<ObjectKey, ObjectValue>::new(merge, None);
942
943        tree.insert(Item::new(
944            ObjectKey::extent(object_id, attr_id, 0..1024),
945            ObjectValue::Extent(ExtentValue::new_raw(0, VOLUME_DATA_KEY_ID)),
946        ))
947        .expect("insert error");
948        tree.insert(Item::new(
949            ObjectKey::extent(object_id, attr_id, 1024..2048),
950            ObjectValue::Extent(ExtentValue::new_raw(16384, VOLUME_DATA_KEY_ID)),
951        ))
952        .expect("insert error");
953        let key = ObjectKey::extent(object_id, attr_id, 512..1536);
954        tree.merge_into(
955            Item::new(key.clone(), ObjectValue::deleted_extent()),
956            &key.key_for_merge_into(),
957        );
958
959        let layer_set = tree.layer_set();
960        let mut merger = layer_set.merger();
961        let mut iter = merger.query(Query::FullScan).await?;
962        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..512));
963        assert_eq!(
964            iter.get().unwrap().value,
965            &ObjectValue::Extent(ExtentValue::new_raw(0, VOLUME_DATA_KEY_ID))
966        );
967        iter.advance().await?;
968        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 512..1536));
969        assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent());
970        iter.advance().await?;
971        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 1536..2048));
972        assert_eq!(
973            iter.get().unwrap().value,
974            &ObjectValue::Extent(ExtentValue::new_raw(16896, VOLUME_DATA_KEY_ID))
975        );
976        iter.advance().await?;
977        assert!(iter.get().is_none());
978        Ok(())
979    }
980
981    #[fuchsia::test]
982    async fn test_merge_deleted_extent_into_merges_with_other_deletions() -> Result<(), Error> {
983        let object_id = 0;
984        let attr_id = AttributeId::TEST_ID;
985        let tree = LSMTree::<ObjectKey, ObjectValue>::new(merge, None);
986
987        tree.insert(Item::new(
988            ObjectKey::extent(object_id, attr_id, 0..1024),
989            ObjectValue::deleted_extent(),
990        ))
991        .expect("insert error");
992        tree.insert(Item::new(
993            ObjectKey::extent(object_id, attr_id, 1024..2048),
994            ObjectValue::deleted_extent(),
995        ))
996        .expect("insert error");
997
998        let key = ObjectKey::extent(object_id, attr_id, 512..1536);
999        tree.merge_into(
1000            Item::new(key.clone(), ObjectValue::deleted_extent()),
1001            &key.key_for_merge_into(),
1002        );
1003
1004        let layer_set = tree.layer_set();
1005        let mut merger = layer_set.merger();
1006        let mut iter = merger.query(Query::FullScan).await?;
1007        assert_eq!(iter.get().unwrap().key, &ObjectKey::extent(object_id, attr_id, 0..2048));
1008        assert_eq!(iter.get().unwrap().value, &ObjectValue::deleted_extent());
1009        iter.advance().await?;
1010        assert!(iter.get().is_none());
1011        Ok(())
1012    }
1013
1014    #[fuchsia::test]
1015    async fn test_merge_size_records() {
1016        let left = &[Item::new(
1017            ObjectKey::attribute(1, AttributeId::TEST_ID, AttributeKey::Attribute),
1018            ObjectValue::attribute(5, false),
1019        )];
1020        let right = &[Item::new(
1021            ObjectKey::attribute(1, AttributeId::TEST_ID, AttributeKey::Attribute),
1022            ObjectValue::attribute(10, false),
1023        )];
1024        let tree = LSMTree::new(merge, None);
1025        test_merge(&tree, left, right, left).await;
1026    }
1027
1028    #[fuchsia::test]
1029    async fn test_different_attributes_not_merged() {
1030        let left = Item::new(
1031            ObjectKey::attribute(1, AttributeId::TEST_ID, AttributeKey::Attribute),
1032            ObjectValue::attribute(5, false),
1033        );
1034        let right = Item::new(
1035            ObjectKey::attribute(1, AttributeId::TEST_ID.next(), AttributeKey::Attribute),
1036            ObjectValue::attribute(10, false),
1037        );
1038        let tree = LSMTree::new(merge, None);
1039        test_merge(&tree, &[left.clone()], &[right.clone()], &[left, right]).await;
1040
1041        let left = Item::new(
1042            ObjectKey::extent(1, AttributeId::TEST_ID, 0..100),
1043            ObjectValue::Extent(ExtentValue::new_raw(0, VOLUME_DATA_KEY_ID)),
1044        );
1045        let right = Item::new(
1046            ObjectKey::extent(1, AttributeId::TEST_ID.next(), 0..100),
1047            ObjectValue::Extent(ExtentValue::new_raw(1, VOLUME_DATA_KEY_ID)),
1048        );
1049        let tree = LSMTree::new(merge, None);
1050        test_merge(&tree, &[left.clone()], &[right.clone()], &[left, right]).await;
1051    }
1052
1053    #[fuchsia::test]
1054    async fn test_tombstone_discards_all_other_records() {
1055        let tombstone = Item::new(ObjectKey::object(1), ObjectValue::None);
1056        let other_object = Item::new(
1057            ObjectKey::object(2),
1058            ObjectValue::file(
1059                1,
1060                0,
1061                Timestamp::default(),
1062                Timestamp::default(),
1063                Timestamp::default(),
1064                Timestamp::default(),
1065                None,
1066                None,
1067            ),
1068        );
1069        let tree = LSMTree::new(merge, None);
1070        test_merge(
1071            &tree,
1072            &[tombstone.clone()],
1073            &[
1074                Item::new(
1075                    ObjectKey::object(1),
1076                    ObjectValue::file(
1077                        1,
1078                        100,
1079                        Timestamp::default(),
1080                        Timestamp::default(),
1081                        Timestamp::default(),
1082                        Timestamp::default(),
1083                        None,
1084                        None,
1085                    ),
1086                ),
1087                Item::new(
1088                    ObjectKey::attribute(1, AttributeId::TEST_ID, AttributeKey::Attribute),
1089                    ObjectValue::attribute(100, false),
1090                ),
1091                other_object.clone(),
1092            ],
1093            &[tombstone, other_object],
1094        )
1095        .await;
1096    }
1097
1098    #[fuchsia::test]
1099    async fn test_extent_overlapping_boundaries() {
1100        use crate::object_store::VOLUME_DATA_KEY_ID;
1101        let object_id = 1;
1102        let attr_id = AttributeId::TEST_ID;
1103        let base = Item::new(
1104            ObjectKey::extent(object_id, attr_id, 50..100),
1105            ObjectValue::Extent(ExtentValue::new_raw(0, VOLUME_DATA_KEY_ID)),
1106        );
1107
1108        // 1. Same end, start off-by-one.
1109        // 49..100 (older, val 2) vs 50..100 (newer, val 1).
1110        // Yields 49..50 (val 2), 50..100 (val 1).
1111        let tree = LSMTree::new(merge, None);
1112        test_merge(
1113            &tree,
1114            &[base.clone()],
1115            &[Item::new(
1116                ObjectKey::extent(object_id, attr_id, 49..100),
1117                ObjectValue::Extent(ExtentValue::new_raw(16384, VOLUME_DATA_KEY_ID)),
1118            )],
1119            &[
1120                Item::new(
1121                    ObjectKey::extent(object_id, attr_id, 49..50),
1122                    ObjectValue::Extent(ExtentValue::new_raw(16384, VOLUME_DATA_KEY_ID)),
1123                ),
1124                base.clone(),
1125            ],
1126        )
1127        .await;
1128
1129        // 51..100 (older, val 2) vs 50..100 (newer, val 1).
1130        // Yields 50..100 (val 1).
1131        let tree = LSMTree::new(merge, None);
1132        test_merge(
1133            &tree,
1134            &[base.clone()],
1135            &[Item::new(
1136                ObjectKey::extent(object_id, attr_id, 51..100),
1137                ObjectValue::Extent(ExtentValue::new_raw(16384, VOLUME_DATA_KEY_ID)),
1138            )],
1139            &[base.clone()],
1140        )
1141        .await;
1142    }
1143
1144    // Tests merging across three LSM tree layers (top, middle, base) with various overlapping
1145    // combinations. `calculate_expected` constructs a 1-D point oracle where the newest layer
1146    // (top = 1, middle = 2, base = 3) wins for each byte position, and aggregates continuous
1147    // segments to verify against the merger iterator output and layer advance depths.
1148    #[fuchsia::test]
1149    async fn test_extent_complex_multi_layer() {
1150        use crate::object_store::extent_record::ExtentValue;
1151        use crate::object_store::object_record::{
1152            AttributeKey, ObjectKey, ObjectKeyData, ObjectValue,
1153        };
1154
1155        let object_id = 1;
1156        let attr_id = AttributeId::TEST_ID;
1157
1158        let top_options = vec![49..50, 50..51, 51..52, 98..99, 99..100, 100..101, 40..95];
1159
1160        let middle_range = 50..100;
1161
1162        let base_options = vec![49..101, 49..100, 50..101, 48..102, 100..101, 100..102, 30..90];
1163
1164        let scale_range =
1165            |r: std::ops::Range<u64>| r.start * MIN_BLOCK_SIZE..r.end * MIN_BLOCK_SIZE;
1166
1167        let calculate_expected = |top: std::ops::Range<u64>,
1168                                  middle: std::ops::Range<u64>,
1169                                  base: std::ops::Range<u64>| {
1170            let mut points = vec![0; 200];
1171            for x in 0u64..200u64 {
1172                if top.contains(&x) {
1173                    points[x as usize] = 1;
1174                } else if middle.contains(&x) {
1175                    points[x as usize] = 2;
1176                } else if base.contains(&x) {
1177                    points[x as usize] = 3;
1178                }
1179            }
1180
1181            let mut result = Vec::new();
1182            let mut current_val = 0;
1183            let mut start = 0;
1184            let mut max_layers_needed = 0;
1185            let mut start_x = None;
1186
1187            for x in 0..200 {
1188                let val = points[x];
1189
1190                let layers_needed = if top.contains(&(x as u64)) {
1191                    1
1192                } else if middle.contains(&(x as u64)) {
1193                    2
1194                } else {
1195                    3
1196                };
1197
1198                if val != current_val {
1199                    if current_val != 0 {
1200                        result.push((start as u64..x as u64, current_val, 3 - max_layers_needed));
1201                    }
1202                    current_val = val;
1203                    start = x;
1204                    if start_x.is_none() && val != 0 {
1205                        start_x = Some(x);
1206                    }
1207                }
1208
1209                if let Some(sx) = start_x {
1210                    if x >= sx {
1211                        max_layers_needed = std::cmp::max(max_layers_needed, layers_needed);
1212                    }
1213                }
1214            }
1215            if current_val != 0 {
1216                result.push((start as u64..200, current_val, 3 - max_layers_needed));
1217            }
1218
1219            result
1220        };
1221
1222        for top_range in top_options {
1223            for base_range in &base_options {
1224                let expected =
1225                    calculate_expected(top_range.clone(), middle_range.clone(), base_range.clone());
1226
1227                let tree = LSMTree::new(merge, None);
1228
1229                // Base layer (Layer 2)
1230                tree.insert(Item::new(
1231                    ObjectKey::extent(object_id, attr_id, scale_range(base_range.clone())),
1232                    ObjectValue::Extent(ExtentValue::new_raw(0, 3)), // key_id = 3
1233                ))
1234                .expect("insert error");
1235                tree.seal();
1236
1237                // Middle layer (Layer 1)
1238                tree.insert(Item::new(
1239                    ObjectKey::extent(object_id, attr_id, scale_range(middle_range.clone())),
1240                    ObjectValue::Extent(ExtentValue::new_raw(0, 2)), // key_id = 2
1241                ))
1242                .expect("insert error");
1243                tree.seal();
1244
1245                // Top layer (Layer 0)
1246                tree.insert(Item::new(
1247                    ObjectKey::extent(object_id, attr_id, scale_range(top_range.clone())),
1248                    ObjectValue::Extent(ExtentValue::new_raw(0, 1)), // key_id = 1
1249                ))
1250                .expect("insert error");
1251
1252                let layer_set = tree.layer_set();
1253
1254                // Start search with full range of expected results.
1255                let mut merger = layer_set.merger();
1256                let mut iter = merger
1257                    .query(Query::LimitedRange(&ObjectKey::extent(
1258                        object_id,
1259                        attr_id,
1260                        scale_range(expected[0].0.start..expected.last().unwrap().0.end),
1261                    )))
1262                    .await
1263                    .expect("seek failed");
1264
1265                assert_eq!(iter.pending_iterators_len(), expected[0].2);
1266
1267                for e in expected {
1268                    let crate::object_store::ItemRef { key, value, .. } =
1269                        iter.get().expect("get failed");
1270                    if let ObjectKeyData::Attribute(aid, AttributeKey::Extent(extent)) = &key.data {
1271                        assert_eq!(aid, &attr_id);
1272                        assert_eq!(&**extent, &scale_range(e.0));
1273                    } else {
1274                        panic!("Unexpected key type");
1275                    }
1276                    if let ObjectValue::Extent(ExtentValue::Some { key_id, .. }) = value {
1277                        assert_eq!(key_id, &e.1);
1278                    } else {
1279                        panic!("Unexpected value type");
1280                    }
1281                    assert_eq!(iter.pending_iterators_len(), e.2);
1282                    iter.advance().await.expect("advance failed");
1283                }
1284                assert_eq!(iter.pending_iterators_len(), 0);
1285                assert!(iter.get().is_none());
1286            }
1287        }
1288    }
1289
1290    #[fuchsia::test]
1291    async fn test_next_key_behavior() -> Result<(), Error> {
1292        let object_id = 1;
1293        let attr_id = AttributeId::TEST_ID;
1294        let tree = LSMTree::new(merge, None);
1295
1296        // Layer 1 (older)
1297        tree.insert(Item::new(
1298            ObjectKey::extent(object_id, attr_id, 50 * MIN_BLOCK_SIZE..101 * MIN_BLOCK_SIZE),
1299            ObjectValue::Extent(ExtentValue::new_raw(0, VOLUME_DATA_KEY_ID)),
1300        ))?;
1301        tree.seal();
1302
1303        // Layer 0 (newer)
1304        tree.insert(Item::new(
1305            ObjectKey::extent(object_id, attr_id, 0..100 * MIN_BLOCK_SIZE),
1306            ObjectValue::Extent(ExtentValue::new_raw(16384, VOLUME_DATA_KEY_ID)),
1307        ))?;
1308
1309        let layer_set = tree.layer_set();
1310        let mut merger = layer_set.merger();
1311
1312        let mut iter = merger
1313            .query(Query::LimitedRange(&ObjectKey::extent(
1314                object_id,
1315                attr_id,
1316                0..100 * MIN_BLOCK_SIZE,
1317            )))
1318            .await?;
1319
1320        let item = iter.get().expect("missing item");
1321        assert_eq!(item.key, &ObjectKey::extent(object_id, attr_id, 0..100 * MIN_BLOCK_SIZE));
1322
1323        iter.advance().await.expect("advance failed");
1324
1325        let item = iter.get().expect("missing item");
1326        assert_eq!(
1327            item.key,
1328            &ObjectKey::extent(object_id, attr_id, 100 * MIN_BLOCK_SIZE..101 * MIN_BLOCK_SIZE,)
1329        );
1330
1331        iter.advance().await.expect("advance failed");
1332        assert!(iter.get().is_none());
1333
1334        Ok(())
1335    }
1336
1337    #[fuchsia::test]
1338    async fn test_merge_project_usage() {
1339        let tree = LSMTree::new(merge, None);
1340        let key = ObjectKey::project_usage(5, ProjectId::new(6).unwrap());
1341
1342        tree.insert(Item::new(key.clone(), ObjectValue::BytesAndNodes { bytes: 100, nodes: 1000 }))
1343            .expect("insert error");
1344        tree.merge_into(
1345            Item::new(key.clone(), ObjectValue::BytesAndNodes { bytes: 4, nodes: 8 }),
1346            &key,
1347        );
1348        tree.seal();
1349
1350        tree.merge_into(
1351            Item::new(key.clone(), ObjectValue::BytesAndNodes { bytes: -1, nodes: -2 }),
1352            &key,
1353        );
1354        tree.seal();
1355
1356        tree.merge_into(
1357            Item::new(key.clone(), ObjectValue::BytesAndNodes { bytes: 16, nodes: 32 }),
1358            &key,
1359        );
1360
1361        let layer_set = tree.layer_set();
1362        let mut merger = layer_set.merger();
1363        let mut iter = merger.query(Query::FullScan).await.unwrap();
1364        assert_eq!(iter.get().unwrap().key, &key);
1365        assert_eq!(
1366            iter.get().unwrap().value,
1367            &ObjectValue::BytesAndNodes { bytes: 119, nodes: 1038 }
1368        );
1369        iter.advance().await.unwrap();
1370        assert!(iter.get().is_none());
1371    }
1372
1373    #[fuchsia::test]
1374    async fn test_merge_project_usage_gap_layer() {
1375        let tree = LSMTree::new(merge, None);
1376        let key = ObjectKey::project_usage(5, ProjectId::new(6).unwrap());
1377        let key2 = ObjectKey::project_usage(5, ProjectId::new(7).unwrap());
1378
1379        tree.insert(Item::new(key.clone(), ObjectValue::BytesAndNodes { bytes: 100, nodes: 1000 }))
1380            .expect("insert error");
1381        tree.merge_into(
1382            Item::new(key.clone(), ObjectValue::BytesAndNodes { bytes: 4, nodes: 8 }),
1383            &key,
1384        );
1385        tree.seal();
1386
1387        assert_eq!(
1388            tree.find_value(&key).await.expect("Find").unwrap(),
1389            ObjectValue::BytesAndNodes { bytes: 104, nodes: 1008 }
1390        );
1391
1392        tree.merge_into(
1393            Item::new(key2.clone(), ObjectValue::BytesAndNodes { bytes: 13, nodes: 17 }),
1394            &key,
1395        );
1396        tree.seal();
1397
1398        assert_eq!(
1399            tree.find_value(&key).await.expect("Find").unwrap(),
1400            ObjectValue::BytesAndNodes { bytes: 104, nodes: 1008 }
1401        );
1402
1403        tree.merge_into(
1404            Item::new(key.clone(), ObjectValue::BytesAndNodes { bytes: 16, nodes: 32 }),
1405            &key,
1406        );
1407
1408        let layer_set = tree.layer_set();
1409        let mut merger = layer_set.merger();
1410        let mut iter = merger.query(Query::FullScan).await.unwrap();
1411        assert_eq!(iter.get().unwrap().key, &key);
1412        assert_eq!(
1413            iter.get().unwrap().value,
1414            &ObjectValue::BytesAndNodes { bytes: 120, nodes: 1040 }
1415        );
1416        iter.advance().await.unwrap();
1417        assert_eq!(iter.get().unwrap().key, &key2);
1418        assert_eq!(iter.get().unwrap().value, &ObjectValue::BytesAndNodes { bytes: 13, nodes: 17 });
1419        iter.advance().await.unwrap();
1420        assert!(iter.get().is_none());
1421
1422        assert_eq!(
1423            tree.find_value(&key).await.expect("Find").unwrap(),
1424            ObjectValue::BytesAndNodes { bytes: 120, nodes: 1040 }
1425        );
1426    }
1427
1428    #[fuchsia::test]
1429    async fn test_merge_project_usage_to_zero() {
1430        let tree = LSMTree::new(merge, None);
1431        let key = ObjectKey::project_usage(5, ProjectId::new(6).unwrap());
1432
1433        tree.insert(Item::new(key.clone(), ObjectValue::BytesAndNodes { bytes: 4, nodes: 8 }))
1434            .expect("insert error");
1435        tree.seal();
1436
1437        tree.merge_into(
1438            Item::new(key.clone(), ObjectValue::BytesAndNodes { bytes: -4, nodes: -8 }),
1439            &key,
1440        );
1441        tree.seal();
1442
1443        let layer_set = tree.layer_set();
1444        let mut merger = layer_set.merger();
1445        let iter = merger.query(Query::FullScan).await.unwrap();
1446        assert!(iter.get().is_none());
1447    }
1448
1449    #[fuchsia::test]
1450    async fn test_merge_project_usage_recover_from_zero() {
1451        let tree = LSMTree::new(merge, None);
1452        let key = ObjectKey::project_usage(5, ProjectId::new(6).unwrap());
1453
1454        tree.insert(Item::new(key.clone(), ObjectValue::BytesAndNodes { bytes: 4, nodes: 8 }))
1455            .expect("insert error");
1456        tree.seal();
1457
1458        tree.merge_into(
1459            Item::new(key.clone(), ObjectValue::BytesAndNodes { bytes: -4, nodes: -8 }),
1460            &key,
1461        );
1462        tree.seal();
1463
1464        tree.merge_into(
1465            Item::new(key.clone(), ObjectValue::BytesAndNodes { bytes: 20, nodes: 40 }),
1466            &key,
1467        );
1468        tree.seal();
1469
1470        let layer_set = tree.layer_set();
1471        let mut merger = layer_set.merger();
1472        let mut iter = merger.query(Query::FullScan).await.unwrap();
1473        assert_eq!(iter.get().unwrap().key, &key);
1474        assert_eq!(iter.get().unwrap().value, &ObjectValue::BytesAndNodes { bytes: 20, nodes: 40 });
1475        iter.advance().await.unwrap();
1476        assert!(iter.get().is_none());
1477    }
1478
1479    #[fuchsia::test]
1480    async fn test_merge_project_usage_layer_merge_to_negative() {
1481        let tree = LSMTree::new(merge, None);
1482        let key = ObjectKey::project_usage(5, ProjectId::new(6).unwrap());
1483
1484        tree.insert(Item::new(key.clone(), ObjectValue::BytesAndNodes { bytes: 4, nodes: 8 }))
1485            .expect("insert error");
1486        tree.seal();
1487
1488        tree.merge_into(
1489            Item::new(key.clone(), ObjectValue::BytesAndNodes { bytes: 16, nodes: 32 }),
1490            &key,
1491        );
1492        tree.seal();
1493
1494        // As we merge from the newest layer down this will drop bytes below zero and nodes to
1495        // exactly zero during the merge process.
1496        tree.merge_into(
1497            Item::new(key.clone(), ObjectValue::BytesAndNodes { bytes: -18, nodes: -32 }),
1498            &key,
1499        );
1500        tree.seal();
1501
1502        let layer_set = tree.layer_set();
1503        let mut merger = layer_set.merger();
1504        let mut iter = merger.query(Query::FullScan).await.unwrap();
1505        assert_eq!(iter.get().unwrap().key, &key);
1506        assert_eq!(iter.get().unwrap().value, &ObjectValue::BytesAndNodes { bytes: 2, nodes: 8 });
1507        iter.advance().await.unwrap();
1508        assert!(iter.get().is_none());
1509    }
1510}
fxfs/object_store/merge.rs

fxfs/object_store/
merge.rs
