Skip to main content

mito2/compaction/
run.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! This file contains code to find sorted runs in a set if ranged items and
16//! along with the best way to merge these items to satisfy the desired run count.
17
18use std::cmp::Ordering;
19use std::collections::BinaryHeap;
20
21use bytes::{Buf, Bytes};
22use common_base::BitVec;
23use common_time::Timestamp;
24
25use crate::sst::file::FileHandle;
26
27/// Trait for any items with specific range (both boundaries are inclusive).
28pub trait Ranged {
29    type BoundType: Ord + Copy;
30
31    /// Returns the inclusive range of item.
32    fn range(&self) -> (Self::BoundType, Self::BoundType);
33
34    fn overlap(&self, other: &Self) -> bool {
35        let (lhs_start, lhs_end) = self.range();
36        let (rhs_start, rhs_end) = other.range();
37
38        lhs_start.max(rhs_start) < lhs_end.min(rhs_end)
39    }
40
41    /// Like `overlap`, but treats touching boundaries as overlapping (inclusive).
42    /// Used by `find_overlapping_items` where shared boundaries count as overlap.
43    fn overlap_inclusive(&self, other: &Self) -> bool {
44        let (lhs_start, lhs_end) = self.range();
45        let (rhs_start, rhs_end) = other.range();
46
47        lhs_start.max(rhs_start) <= lhs_end.min(rhs_end)
48    }
49}
50
51pub(crate) fn primary_key_ranges_overlap(lhs: &(Bytes, Bytes), rhs: &(Bytes, Bytes)) -> bool {
52    lhs.0.chunk().max(rhs.0.chunk()) <= lhs.1.chunk().min(rhs.1.chunk())
53}
54
55pub(crate) fn merge_primary_key_ranges(
56    lhs: Option<(Bytes, Bytes)>,
57    rhs: Option<(Bytes, Bytes)>,
58) -> Option<(Bytes, Bytes)> {
59    match (lhs, rhs) {
60        (Some((lhs_min, lhs_max)), Some((rhs_min, rhs_max))) => {
61            Some((lhs_min.min(rhs_min), lhs_max.max(rhs_max)))
62        }
63        _ => None,
64    }
65}
66
67pub fn find_overlapping_items<T: Item + Clone>(
68    l: &mut SortedRun<T>,
69    r: &mut SortedRun<T>,
70    result: &mut Vec<T>,
71) {
72    if l.items.is_empty() || r.items.is_empty() {
73        return;
74    }
75
76    result.clear();
77    result.reserve(l.items.len() + r.items.len());
78
79    // Sort both arrays by start boundary for more efficient overlap detection
80    if !l.sorted {
81        sort_ranged_items(&mut l.items);
82        l.sorted = true;
83    }
84    if !r.sorted {
85        sort_ranged_items(&mut r.items);
86        r.sorted = true;
87    }
88
89    let mut r_idx = 0;
90
91    let mut selected = BitVec::repeat(false, r.items().len() + l.items.len());
92
93    for (lhs_idx, lhs) in l.items.iter().enumerate() {
94        let (lhs_start, lhs_end) = lhs.range();
95
96        // Skip right elements that end before current left element starts
97        while r_idx < r.items.len() {
98            let (_, rhs_end) = r.items[r_idx].range();
99            if rhs_end < lhs_start {
100                r_idx += 1;
101            } else {
102                break;
103            }
104        }
105
106        // Check for overlaps with remaining right elements
107        let mut j = r_idx;
108        while j < r.items.len() {
109            let (rhs_start, _rhs_end) = r.items[j].range();
110
111            // If right element starts after left element ends, no more overlaps possible
112            if rhs_start > lhs_end {
113                break;
114            }
115
116            // We have an overlap (inclusive: touching boundaries count)
117            if lhs.overlap_inclusive(&r.items[j]) {
118                if !selected[lhs_idx] {
119                    result.push(lhs.clone());
120                    selected.set(lhs_idx, true);
121                }
122
123                let rhs_selected_idx = l.items.len() + j;
124                if !selected[rhs_selected_idx] {
125                    result.push(r.items[j].clone());
126                    selected.set(rhs_selected_idx, true);
127                }
128            }
129
130            j += 1;
131        }
132    }
133}
134
135// Sorts ranges by start asc and end desc.
136fn sort_ranged_items<T: Ranged>(values: &mut [T]) {
137    values.sort_unstable_by(|l, r| {
138        let (l_start, l_end) = l.range();
139        let (r_start, r_end) = r.range();
140        l_start.cmp(&r_start).then(r_end.cmp(&l_end))
141    });
142}
143
144/// Trait for items to merge.
145pub trait Item: Ranged + Clone {
146    /// Size is used to calculate the cost of merging items.
147    fn size(&self) -> usize;
148}
149
150impl Ranged for FileHandle {
151    type BoundType = Timestamp;
152
153    fn range(&self) -> (Self::BoundType, Self::BoundType) {
154        self.time_range()
155    }
156
157    fn overlap(&self, other: &Self) -> bool {
158        let (lhs_start, lhs_end) = self.range();
159        let (rhs_start, rhs_end) = other.range();
160        if lhs_start.max(rhs_start) >= lhs_end.min(rhs_end) {
161            return false;
162        }
163
164        match (&self.primary_key_range(), &other.primary_key_range()) {
165            (Some(lhs), Some(rhs)) => primary_key_ranges_overlap(lhs, rhs),
166            _ => true,
167        }
168    }
169
170    fn overlap_inclusive(&self, other: &Self) -> bool {
171        let (lhs_start, lhs_end) = self.range();
172        let (rhs_start, rhs_end) = other.range();
173        if lhs_start.max(rhs_start) > lhs_end.min(rhs_end) {
174            return false;
175        }
176
177        match (&self.primary_key_range(), &other.primary_key_range()) {
178            (Some(lhs), Some(rhs)) => primary_key_ranges_overlap(lhs, rhs),
179            _ => true,
180        }
181    }
182}
183
184impl Item for FileHandle {
185    fn size(&self) -> usize {
186        self.size() as usize
187    }
188}
189
190/// A set of files with non-overlapping time ranges.
191#[derive(Debug, Clone)]
192pub struct SortedRun<T: Item> {
193    /// Items to merge
194    items: Vec<T>,
195    /// The total size of all items.
196    size: usize,
197    /// The lower bound of all items.
198    start: Option<T::BoundType>,
199    /// The upper bound of all items.
200    end: Option<T::BoundType>,
201    /// Whether items are sorted.
202    sorted: bool,
203}
204
205impl<T: Item> From<Vec<T>> for SortedRun<T> {
206    fn from(items: Vec<T>) -> Self {
207        let mut r = Self {
208            items: Vec::with_capacity(items.len()),
209            size: 0,
210            start: None,
211            end: None,
212            sorted: false,
213        };
214        for item in items {
215            r.push_item(item);
216        }
217
218        r
219    }
220}
221
222impl<T> Default for SortedRun<T>
223where
224    T: Item,
225{
226    fn default() -> Self {
227        Self {
228            items: vec![],
229            size: 0,
230            start: None,
231            end: None,
232            sorted: false,
233        }
234    }
235}
236
237impl<T> SortedRun<T>
238where
239    T: Item,
240{
241    pub fn items(&self) -> &[T] {
242        &self.items
243    }
244
245    fn push_item(&mut self, t: T) {
246        let (file_start, file_end) = t.range();
247        self.size += t.size();
248        self.items.push(t);
249        self.start = Some(self.start.map_or(file_start, |v| v.min(file_start)));
250        self.end = Some(self.end.map_or(file_end, |v| v.max(file_end)));
251    }
252}
253
254/// Finds sorted runs in given items.
255pub fn find_sorted_runs<T>(items: &mut [T]) -> Vec<SortedRun<T>>
256where
257    T: Item,
258{
259    if items.is_empty() {
260        return vec![];
261    }
262    // sort files
263    sort_ranged_items(items);
264
265    let mut current_run = SortedRun::default();
266    let mut runs = vec![];
267    let mut active_run_item_indices = Vec::new();
268
269    let mut selection = BitVec::repeat(false, items.len());
270    while !selection.all() {
271        // until all items are assigned to some sorted run.
272        let mut last_pruned_start = None;
273        for (item, mut selected) in items.iter().zip(selection.iter_mut()) {
274            if *selected {
275                // item is already assigned.
276                continue;
277            }
278            if current_run.items.is_empty() {
279                // current run is empty, just add current_item
280                selected.set(true);
281                current_run.push_item(item.clone());
282                active_run_item_indices.push(current_run.items.len() - 1);
283            } else {
284                // the current item does not overlap with any item in current run,
285                // then it belongs to current run. Because now we introduced primary
286                // key range, we cannot simply use timestamps to check overlapping.
287                let (item_start, _) = item.range();
288                if last_pruned_start != Some(item_start) {
289                    active_run_item_indices.retain(|idx| {
290                        let (_, run_item_end) = current_run.items[*idx].range();
291                        run_item_end > item_start
292                    });
293                    last_pruned_start = Some(item_start);
294                }
295
296                let mut overlaps_any = false;
297                for idx in &active_run_item_indices {
298                    let run_item = &current_run.items[*idx];
299                    if run_item.overlap(item) {
300                        overlaps_any = true;
301                        break;
302                    }
303                }
304                if !overlaps_any {
305                    // does not overlap, push to current run
306                    selected.set(true);
307                    let item_idx = current_run.items.len();
308                    current_run.push_item(item.clone());
309                    active_run_item_indices.push(item_idx);
310                }
311            }
312        }
313        // finished an iteration, we've found a new run.
314        runs.push(std::mem::take(&mut current_run));
315        active_run_item_indices.clear();
316    }
317    runs
318}
319
320#[cfg(any(test, feature = "test", feature = "testing"))]
321pub fn find_sorted_runs_original<T>(items: &mut [T]) -> Vec<SortedRun<T>>
322where
323    T: Item,
324{
325    if items.is_empty() {
326        return vec![];
327    }
328    // sort files
329    sort_ranged_items(items);
330
331    let mut current_run = SortedRun::default();
332    let mut runs = vec![];
333
334    let mut selection = BitVec::repeat(false, items.len());
335    while !selection.all() {
336        // until all items are assigned to some sorted run.
337        for (item, mut selected) in items.iter().zip(selection.iter_mut()) {
338            if *selected {
339                // item is already assigned.
340                continue;
341            }
342            if current_run.items.is_empty() {
343                // current run is empty, just add current_item
344                selected.set(true);
345                current_run.push_item(item.clone());
346            } else {
347                // the current item does not overlap with any item in current run,
348                // then it belongs to current run. Because now we introduced primary
349                // key range, we cannot simply use timestamps to check overlapping.
350                let overlaps_any = current_run.items.iter().any(|i| i.overlap(item));
351                if !overlaps_any {
352                    // does not overlap, push to current run
353                    selected.set(true);
354                    current_run.push_item(item.clone());
355                }
356            }
357        }
358        // finished an iteration, we've found a new run.
359        runs.push(std::mem::take(&mut current_run));
360    }
361    runs
362}
363
364pub(crate) fn find_sorted_runs_by_time_range<T>(items: &mut [T]) -> Vec<SortedRun<T>>
365where
366    T: Item,
367{
368    if items.is_empty() {
369        return vec![];
370    }
371    sort_ranged_items(items);
372
373    use derive_more::{Eq, PartialEq};
374
375    /// `SortedRun` with a creation sequence `i`.
376    #[derive(PartialEq, Eq)]
377    struct Run<T: Item> {
378        i: usize,
379        #[partial_eq(skip)]
380        run: SortedRun<T>,
381    }
382
383    impl<T: Item> Run<T> {
384        fn new(i: usize, item: &T) -> Run<T> {
385            let mut run = SortedRun::default();
386            run.push_item(item.clone());
387            Run { i, run }
388        }
389
390        fn push_item(&mut self, item: &T) {
391            self.run.push_item(item.clone());
392        }
393    }
394
395    impl<T: Item> PartialOrd for Run<T> {
396        fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
397            Some(self.cmp(other))
398        }
399    }
400
401    /// Sort by run's `end` desc then `start` asc.
402    impl<T: Item> Ord for Run<T> {
403        fn cmp(&self, other: &Self) -> Ordering {
404            let l_run = &self.run;
405            let r_run = &other.run;
406
407            // Safety: `start` and `end` must both exist because it's guaranteed that whenever a
408            // `Run` is created, an item is pushed into it immediately (see its `new` method above).
409            // And there are no other ways to create a `Run` beyond its `new` method in this
410            // function's scope.
411            let l_end = l_run.end.unwrap();
412            let r_end = r_run.end.unwrap();
413            r_end
414                .cmp(&l_end)
415                .then_with(|| {
416                    let l_start = l_run.start.unwrap();
417                    let r_start = r_run.start.unwrap();
418                    l_start.cmp(&r_start)
419                })
420                .then_with(|| self.i.cmp(&other.i))
421        }
422    }
423
424    /// Wrapper around the `Run` above, to support sorting them by their creation sequence `i`.
425    #[derive(PartialEq, Eq)]
426    struct Wrapper<T: Item>(Run<T>);
427
428    impl<T: Item> PartialOrd for Wrapper<T> {
429        fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
430            Some(self.cmp(other))
431        }
432    }
433
434    impl<T: Item> Ord for Wrapper<T> {
435        fn cmp(&self, other: &Self) -> Ordering {
436            other.0.i.cmp(&self.0.i)
437        }
438    }
439
440    // Two heaps for finding a run that is both:
441    // 1. not overlapping with item's range,
442    // 2. and is created earliest,
443    // when iterating the items.
444    //
445    // Heap 1 (`runs_sorted_by_end`) is for storing the runs of which top has the minimal "end"
446    // just about to overlap with the current selected item.
447    //
448    // Heap 2 (`runs_sort_by_index`) is for storing the runs that all have "end"s non-overlap with
449    // the current selected item, and of which top is the earliest created run.
450    //
451    // The finding of a suitable run basically works like this:
452    // 1. moves the runs in heap 1 to heap 2, until the top is overlapping with the current item;
453    // 2. now heap 2 has all the runs that can accept the current item, pop its top;
454    // 3. the top is the earliest created run, push the current item;
455    // 4. because the run has changed, push it back to heap 1;
456    // 5. check the next item. Important: we don't need to push the runs in heap 2 to 1, because
457    //    the items are sorted by "start". When checking the next item, heap 2's runs must all have
458    //    "end"s smaller than next item's "start".
459    //
460    // Actually the heap 2 is only for aligning with the runs selection outcomes in the original
461    // `find_sorted_runs` implementation. If we just need the invariant that each run has the
462    // non-overlapping items, we can get rid of heap 2 and make the codes simpler.
463
464    let mut runs_sort_by_end = BinaryHeap::<Run<T>>::new();
465    let mut runs_sort_by_index = BinaryHeap::<Wrapper<T>>::new();
466    let mut i = 0;
467
468    for item in items {
469        let (start, _) = item.range();
470
471        while let Some(run) = runs_sort_by_end.pop_if(|x| x.run.end.unwrap() <= start) {
472            runs_sort_by_index.push(Wrapper(run));
473        }
474
475        let Some(mut run) = runs_sort_by_index.pop() else {
476            i += 1;
477            runs_sort_by_end.push(Run::new(i, item));
478            continue;
479        };
480
481        run.0.push_item(item);
482        runs_sort_by_end.push(run.0);
483    }
484
485    let mut runs = runs_sort_by_end.into_vec();
486    runs.extend(runs_sort_by_index.into_vec().into_iter().map(|x| x.0));
487    runs.sort_unstable_by_key(|run| run.i);
488    runs.into_iter().map(|x| x.run).collect()
489}
490
491#[cfg(test)]
492mod tests {
493    use bytes::Bytes;
494    use store_api::storage::FileId;
495
496    use super::*;
497    use crate::compaction::test_util::new_file_handle_with_size_sequence_and_primary_key_range;
498
499    #[derive(Clone, Debug, PartialEq)]
500    struct MockFile {
501        start: i64,
502        end: i64,
503        size: usize,
504    }
505
506    impl Ranged for MockFile {
507        type BoundType = i64;
508
509        fn range(&self) -> (Self::BoundType, Self::BoundType) {
510            (self.start, self.end)
511        }
512    }
513
514    impl Item for MockFile {
515        fn size(&self) -> usize {
516            self.size
517        }
518    }
519
520    fn build_items(ranges: &[(i64, i64)]) -> Vec<MockFile> {
521        ranges
522            .iter()
523            .map(|(start, end)| MockFile {
524                start: *start,
525                end: *end,
526                size: (*end - *start) as usize,
527            })
528            .collect()
529    }
530
531    fn pk_range(min: &'static [u8], max: &'static [u8]) -> Option<(Bytes, Bytes)> {
532        Some((Bytes::from_static(min), Bytes::from_static(max)))
533    }
534
535    fn check_sorted_runs(
536        ranges: &[(i64, i64)],
537        expected_runs: &[Vec<(i64, i64)>],
538    ) -> Vec<SortedRun<MockFile>> {
539        let mut files = build_items(ranges);
540        let mut files_clone = files.clone();
541
542        let runs = find_sorted_runs(&mut files);
543
544        let result_file_ranges: Vec<Vec<_>> = runs
545            .iter()
546            .map(|r| r.items.iter().map(|f| f.range()).collect())
547            .collect();
548        assert_eq!(&expected_runs, &result_file_ranges);
549
550        let runs_by_time_range = find_sorted_runs_by_time_range(&mut files_clone);
551        let results: Vec<Vec<_>> = runs_by_time_range
552            .iter()
553            .map(|r| r.items.iter().map(|f| f.range()).collect())
554            .collect();
555        assert_eq!(&expected_runs, &results);
556        runs
557    }
558
559    fn sorted_run_ranges<T: Item>(runs: &[SortedRun<T>]) -> Vec<Vec<T::BoundType>> {
560        runs.iter()
561            .map(|r| {
562                r.items
563                    .iter()
564                    .flat_map(|f| {
565                        let (start, end) = f.range();
566                        [start, end]
567                    })
568                    .collect()
569            })
570            .collect()
571    }
572
573    fn check_find_sorted_runs_consistency(ranges: &[(i64, i64)]) {
574        let mut files = build_items(ranges);
575        let mut files_for_original = files.clone();
576
577        let runs = find_sorted_runs(&mut files);
578        let original_runs = find_sorted_runs_original(&mut files_for_original);
579
580        assert_eq!(sorted_run_ranges(&original_runs), sorted_run_ranges(&runs));
581    }
582
583    #[test]
584    fn test_find_sorted_runs() {
585        check_sorted_runs(&[], &[]);
586        check_sorted_runs(&[(1, 1), (2, 2)], &[vec![(1, 1), (2, 2)]]);
587        check_sorted_runs(&[(1, 2)], &[vec![(1, 2)]]);
588        check_sorted_runs(&[(1, 2), (2, 3)], &[vec![(1, 2), (2, 3)]]);
589        check_sorted_runs(&[(1, 2), (3, 4)], &[vec![(1, 2), (3, 4)]]);
590        check_sorted_runs(&[(2, 4), (1, 3)], &[vec![(1, 3)], vec![(2, 4)]]);
591        check_sorted_runs(
592            &[(1, 3), (2, 4), (4, 5)],
593            &[vec![(1, 3), (4, 5)], vec![(2, 4)]],
594        );
595
596        check_sorted_runs(
597            &[(1, 2), (3, 4), (3, 5)],
598            &[vec![(1, 2), (3, 5)], vec![(3, 4)]],
599        );
600
601        check_sorted_runs(
602            &[(1, 3), (2, 4), (5, 6)],
603            &[vec![(1, 3), (5, 6)], vec![(2, 4)]],
604        );
605
606        check_sorted_runs(
607            &[(1, 2), (3, 5), (4, 6)],
608            &[vec![(1, 2), (3, 5)], vec![(4, 6)]],
609        );
610
611        check_sorted_runs(
612            &[(1, 2), (3, 4), (4, 6), (7, 8)],
613            &[vec![(1, 2), (3, 4), (4, 6), (7, 8)]],
614        );
615        check_sorted_runs(
616            &[(1, 2), (3, 4), (5, 6), (3, 6), (7, 8), (8, 9)],
617            &[vec![(1, 2), (3, 6), (7, 8), (8, 9)], vec![(3, 4), (5, 6)]],
618        );
619
620        check_sorted_runs(
621            &[(10, 19), (20, 21), (20, 29), (30, 39)],
622            &[vec![(10, 19), (20, 29), (30, 39)], vec![(20, 21)]],
623        );
624
625        check_sorted_runs(
626            &[(10, 19), (20, 29), (21, 22), (30, 39), (31, 32), (32, 42)],
627            &[
628                vec![(10, 19), (20, 29), (30, 39)],
629                vec![(21, 22), (31, 32), (32, 42)],
630            ],
631        );
632    }
633
634    #[test]
635    fn test_find_sorted_runs_matches_original_impl() {
636        for ranges in [
637            &[][..],
638            &[(1, 1), (2, 2)],
639            &[(1, 2), (2, 3)],
640            &[(2, 4), (1, 3)],
641            &[(1, 3), (2, 4), (4, 5)],
642            &[(1, 2), (3, 4), (3, 5)],
643            &[(1, 3), (2, 4), (5, 6)],
644            &[(1, 2), (3, 5), (4, 6)],
645            &[(1, 2), (3, 4), (4, 6), (7, 8)],
646            &[(1, 2), (3, 4), (5, 6), (3, 6), (7, 8), (8, 9)],
647            &[(10, 19), (20, 21), (20, 29), (30, 39)],
648            &[(10, 19), (20, 29), (21, 22), (30, 39), (31, 32), (32, 42)],
649            &[(32, 42), (10, 19), (31, 32), (20, 29), (21, 22), (30, 39)],
650        ] {
651            check_find_sorted_runs_consistency(ranges);
652        }
653    }
654
655    #[test]
656    fn test_find_overlapping_items() {
657        let mut result = Vec::new();
658
659        // Test empty inputs
660        find_overlapping_items(
661            &mut SortedRun::from(Vec::<MockFile>::new()),
662            &mut SortedRun::from(Vec::<MockFile>::new()),
663            &mut result,
664        );
665        assert_eq!(result, Vec::<MockFile>::new());
666
667        let files1 = build_items(&[(1, 3)]);
668        find_overlapping_items(
669            &mut SortedRun::from(files1.clone()),
670            &mut SortedRun::from(Vec::<MockFile>::new()),
671            &mut result,
672        );
673        assert_eq!(result, Vec::<MockFile>::new());
674
675        find_overlapping_items(
676            &mut SortedRun::from(Vec::<MockFile>::new()),
677            &mut SortedRun::from(files1.clone()),
678            &mut result,
679        );
680        assert_eq!(result, Vec::<MockFile>::new());
681
682        // Test non-overlapping ranges
683        let files1 = build_items(&[(1, 3), (5, 7)]);
684        let files2 = build_items(&[(10, 12), (15, 20)]);
685        find_overlapping_items(
686            &mut SortedRun::from(files1),
687            &mut SortedRun::from(files2),
688            &mut result,
689        );
690        assert_eq!(result, Vec::<MockFile>::new());
691
692        // Test simple overlap
693        let files1 = build_items(&[(1, 5)]);
694        let files2 = build_items(&[(3, 7)]);
695        find_overlapping_items(
696            &mut SortedRun::from(files1),
697            &mut SortedRun::from(files2),
698            &mut result,
699        );
700        assert_eq!(result.len(), 2);
701        assert_eq!(result[0].range(), (1, 5));
702        assert_eq!(result[1].range(), (3, 7));
703
704        // Test multiple overlaps
705        let files1 = build_items(&[(1, 5), (8, 12), (15, 20)]);
706        let files2 = build_items(&[(3, 6), (7, 10), (18, 25)]);
707        find_overlapping_items(
708            &mut SortedRun::from(files1),
709            &mut SortedRun::from(files2),
710            &mut result,
711        );
712        assert_eq!(result.len(), 6);
713
714        // Test boundary cases (touching but not overlapping)
715        let files1 = build_items(&[(1, 5)]);
716        let files2 = build_items(&[(5, 10)]); // Touching at 5
717        find_overlapping_items(
718            &mut SortedRun::from(files1),
719            &mut SortedRun::from(files2),
720            &mut result,
721        );
722        assert_eq!(result.len(), 2); // Should overlap since ranges are inclusive
723
724        // Test completely contained ranges
725        let files1 = build_items(&[(1, 10)]);
726        let files2 = build_items(&[(3, 7)]);
727        find_overlapping_items(
728            &mut SortedRun::from(files1),
729            &mut SortedRun::from(files2),
730            &mut result,
731        );
732        assert_eq!(result.len(), 2);
733
734        // Test identical ranges
735        let files1 = build_items(&[(1, 5)]);
736        let files2 = build_items(&[(1, 5)]);
737        find_overlapping_items(
738            &mut SortedRun::from(files1),
739            &mut SortedRun::from(files2),
740            &mut result,
741        );
742        assert_eq!(result.len(), 2);
743
744        // Test unsorted input handling
745        let files1 = build_items(&[(5, 10), (1, 3)]); // Unsorted
746        let files2 = build_items(&[(2, 7), (8, 12)]); // Unsorted
747        find_overlapping_items(
748            &mut SortedRun::from(files1),
749            &mut SortedRun::from(files2),
750            &mut result,
751        );
752        assert_eq!(result.len(), 4); // Should find both overlaps
753    }
754
755    #[test]
756    fn test_file_overlap_time_overlap_pk_disjoint() {
757        let lhs = new_file_handle_with_size_sequence_and_primary_key_range(
758            FileId::random(),
759            0,
760            100,
761            0,
762            1,
763            10,
764            pk_range(b"a", b"f"),
765        );
766        let rhs = new_file_handle_with_size_sequence_and_primary_key_range(
767            FileId::random(),
768            50,
769            150,
770            0,
771            2,
772            10,
773            pk_range(b"x", b"z"),
774        );
775
776        assert!(!lhs.overlap(&rhs));
777    }
778
779    #[test]
780    fn test_find_sorted_runs_collapses_pk_disjoint_files_into_one_run() {
781        let mut files = vec![
782            new_file_handle_with_size_sequence_and_primary_key_range(
783                FileId::random(),
784                0,
785                100,
786                0,
787                1,
788                10,
789                pk_range(b"a", b"f"),
790            ),
791            new_file_handle_with_size_sequence_and_primary_key_range(
792                FileId::random(),
793                50,
794                150,
795                0,
796                2,
797                10,
798                pk_range(b"x", b"z"),
799            ),
800        ];
801
802        let runs = find_sorted_runs(&mut files);
803
804        assert_eq!(1, runs.len());
805        assert_eq!(2, runs[0].items().len());
806    }
807
808    #[test]
809    fn test_find_sorted_runs_handles_2d_transitivity_break() {
810        let mut files = vec![
811            new_file_handle_with_size_sequence_and_primary_key_range(
812                FileId::random(),
813                0,
814                100,
815                0,
816                1,
817                10,
818                pk_range(b"a", b"f"),
819            ),
820            new_file_handle_with_size_sequence_and_primary_key_range(
821                FileId::random(),
822                50,
823                150,
824                0,
825                2,
826                10,
827                pk_range(b"x", b"z"),
828            ),
829            new_file_handle_with_size_sequence_and_primary_key_range(
830                FileId::random(),
831                50,
832                150,
833                0,
834                3,
835                10,
836                pk_range(b"a", b"f"),
837            ),
838        ];
839
840        let runs = find_sorted_runs(&mut files);
841
842        assert_eq!(2, runs.len());
843        assert_eq!(2, runs[0].items().len());
844        assert_eq!(1, runs[1].items().len());
845    }
846
847    #[test]
848    fn test_find_overlapping_items_skips_pk_disjoint_pairs() {
849        let mut left = SortedRun::from(vec![
850            new_file_handle_with_size_sequence_and_primary_key_range(
851                FileId::random(),
852                0,
853                100,
854                0,
855                1,
856                10,
857                pk_range(b"a", b"f"),
858            ),
859        ]);
860        let mut right = SortedRun::from(vec![
861            new_file_handle_with_size_sequence_and_primary_key_range(
862                FileId::random(),
863                50,
864                150,
865                0,
866                2,
867                10,
868                pk_range(b"x", b"z"),
869            ),
870        ]);
871        let mut result = Vec::new();
872
873        find_overlapping_items(&mut left, &mut right, &mut result);
874
875        assert!(result.is_empty());
876    }
877
878    #[test]
879    fn test_file_touching_time_boundary_with_same_pk_is_not_overlap() {
880        let lhs = new_file_handle_with_size_sequence_and_primary_key_range(
881            FileId::random(),
882            0,
883            100,
884            0,
885            1,
886            10,
887            pk_range(b"a", b"f"),
888        );
889        let rhs = new_file_handle_with_size_sequence_and_primary_key_range(
890            FileId::random(),
891            100,
892            150,
893            0,
894            2,
895            10,
896            pk_range(b"a", b"f"),
897        );
898
899        assert!(!lhs.overlap(&rhs));
900    }
901}