add max_range_size

Author: BohuTANG

src/common/base/src/rangemap/range_merger.rs

@@ -21,10 +21,14 @@ use std::cmp;
 use std::ops::Range;
 
 /// Check overlap.
-fn overlaps(this: &Range<u64>, other: &Range<u64>, hole: u64) -> bool {
-    let end_with_hole = this.end + hole;
-    (other.start >= this.start && other.start <= end_with_hole)
-        || (other.end >= this.start && other.end <= end_with_hole)
+fn overlaps(this: &Range<u64>, other: &Range<u64>, max_gap_size: u64, max_range_size: u64) -> bool {
+    if (this.end - this.start) >= max_range_size {
+        false
+    } else {
+        let end_with_hole = this.end + max_gap_size;
+        (other.start >= this.start && other.start <= end_with_hole)
+            || (other.end >= this.start && other.end <= end_with_hole)
+    }
 }
 
 /// Merge to one.
@@ -35,18 +39,20 @@ fn merge(this: &mut Range<u64>, other: &Range<u64>) {
 
 #[derive(Debug, Clone)]
 pub struct RangeMerger {
-    max_merge_gap: u64,
+    max_gap_size: u64,
+    max_range_size: u64,
     ranges: Vec<Range<u64>>,
 }
 
 impl RangeMerger {
-    pub fn from_iter<I>(iter: I, max_merge_gap: u64) -> Self
+    pub fn from_iter<I>(iter: I, max_gap_size: u64, max_range_size: u64) -> Self
     where I: IntoIterator<Item = std::ops::Range<u64>> {
         let mut raw_ranges: Vec<_> = iter.into_iter().collect();
         raw_ranges.sort_by(|a, b| a.start.cmp(&b.start));
 
         let mut rs = RangeMerger {
-            max_merge_gap,
+            max_gap_size,
+            max_range_size,
             ranges: Vec::with_capacity(raw_ranges.len()),
         };
 
@@ -63,7 +69,7 @@ impl RangeMerger {
 
     fn add(&mut self, range: &Range<u64>) {
         if let Some(last) = self.ranges.last_mut() {
-            if overlaps(last, range, self.max_merge_gap) {
+            if overlaps(last, range, self.max_gap_size, self.max_range_size) {
                 merge(last, range);
                 return;
             }

src/common/base/tests/it/range_merger.rs

@@ -32,7 +32,7 @@ impl fmt::Display for Array {
 fn test_range_merger() -> Result<()> {
     let v = [3..6, 1..5, 7..11, 8..9, 9..12, 4..8, 13..15, 18..20];
 
-    let mr = RangeMerger::from_iter(v, 0);
+    let mr = RangeMerger::from_iter(v, 0, 100);
     let actual = format!("{}", Array(mr.ranges()));
     let expect = "[1,12] [13,15] [18,20] ";
     assert_eq!(actual, expect);
@@ -41,32 +41,40 @@ fn test_range_merger() -> Result<()> {
 }
 
 #[test]
-fn test_range_merger_with_hole() -> Result<()> {
+fn test_range_merger_with_gap() -> Result<()> {
     let v = [3..6, 1..5, 7..11, 8..9, 9..12, 4..8, 13..15, 18..20];
 
-    // Hole = 1
+    // max_gap_size = 1
     {
-        let mr = RangeMerger::from_iter(v.clone(), 1);
+        let mr = RangeMerger::from_iter(v.clone(), 1, 100);
         let actual = format!("{}", Array(mr.ranges()));
         let expect = "[1,15] [18,20] ";
         assert_eq!(actual, expect);
     }
 
-    // Hole = 2
+    // max_gap_size = 2
     {
-        let mr = RangeMerger::from_iter(v.clone(), 2);
+        let mr = RangeMerger::from_iter(v.clone(), 2, 100);
         let actual = format!("{}", Array(mr.ranges()));
         let expect = "[1,15] [18,20] ";
         assert_eq!(actual, expect);
     }
 
-    // Hole = 3
+    // max_gap_size = 3
     {
-        let mr = RangeMerger::from_iter(v, 3);
+        let mr = RangeMerger::from_iter(v.clone(), 3, 100);
         let actual = format!("{}", Array(mr.ranges()));
         let expect = "[1,20] ";
         assert_eq!(actual, expect);
     }
 
+    // max_gap_size = 3, max_range_size = 5
+    {
+        let mr = RangeMerger::from_iter(v, 3, 4);
+        let actual = format!("{}", Array(mr.ranges()));
+        let expect = "[1,5] [3,8] [7,11] [8,12] [13,20] ";
+        assert_eq!(actual, expect);
+    }
+
     Ok(())
 }

src/query/storages/fuse/fuse/src/io/read/block_reader_parquet.rs

@@ -342,7 +342,8 @@ impl BlockReader {
 
     /// Merge overlap io request to one.
     fn merge_io_requests(
-        max_merge_gap: u64,
+        max_gap_size: u64,
+        max_range_size: u64,
         part: &PartInfoPtr,
     ) -> Result<Vec<std::ops::Range<u64>>> {
         let part = FusePartInfo::from_part(part)?;
@@ -351,7 +352,7 @@ impl BlockReader {
             .values()
             .map(|v| (v.offset..v.offset + v.len))
             .collect::<Vec<_>>();
-        Ok(RangeMerger::from_iter(ranges, max_merge_gap).ranges())
+        Ok(RangeMerger::from_iter(ranges, max_gap_size, max_range_size).ranges())
     }
 
     pub async fn read_columns_data(
@@ -380,8 +381,9 @@ impl BlockReader {
         let normal_cost = now.elapsed().unwrap().as_millis();
 
         // Merge io requests.
-        let max_merge_gap = ctx.get_settings().get_max_storage_io_requests_merge_gap()?;
-        let ranges = Self::merge_io_requests(max_merge_gap, &raw_part)?;
+        let max_gap_size = ctx.get_settings().get_max_storage_io_requests_merge_gap()?;
+        let max_range_size = ctx.get_settings().get_max_storage_io_requests_page_size()?;
+        let ranges = Self::merge_io_requests(max_gap_size, max_range_size, &raw_part)?;
         let mut merge_io_handlers = Vec::with_capacity(ranges.len());
         for (index, range) in ranges.iter().enumerate() {
             merge_io_handlers.push(UnlimitedFuture::create(Self::read_range(