feat(fuse): try to improve object storage read, merge io requests

BohuTANG · BohuTANG · commit 64082b50afed · 2022-12-22T17:35:51.000+08:00
diff --git a/src/common/base/src/rangemap/mod.rs b/src/common/base/src/rangemap/mod.rs
@@ -14,6 +14,8 @@
 
 mod range_map;
 mod range_map_key;
+mod range_merger;
 
 pub use range_map::RangeMap;
 pub use range_map_key::RangeMapKey;
+pub use range_merger::RangeMerger;
diff --git a/src/common/base/src/rangemap/range_merger.rs b/src/common/base/src/rangemap/range_merger.rs
@@ -0,0 +1,80 @@
+// Copyright 2022 Datafuse Labs.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+/// Merge the overlap range.
+/// If we have 3 ranges: [1,3], [2,4], [6,8]
+/// The final range stack will be: [1,4], [6,8]
+/// If we set the max_merge_gap to 2:
+/// The final range stack will be: [1,8]
+use std::cmp;
+use std::ops::Range;
+
+/// Check overlap.
+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.
+fn merge(this: &mut Range<u64>, other: &Range<u64>) {
+    this.start = cmp::min(this.start, other.start);
+    this.end = cmp::max(this.end, other.end);
+}
+
+#[derive(Debug, Clone)]
+pub struct RangeMerger {
+    max_gap_size: u64,
+    max_range_size: u64,
+    ranges: Vec<Range<u64>>,
+}
+
+impl RangeMerger {
+    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_gap_size,
+            max_range_size,
+            ranges: Vec::with_capacity(raw_ranges.len()),
+        };
+
+        for range in &raw_ranges {
+            rs.add(range);
+        }
+
+        rs
+    }
+
+    pub fn ranges(&self) -> Vec<Range<u64>> {
+        self.ranges.clone()
+    }
+
+    fn add(&mut self, range: &Range<u64>) {
+        if let Some(last) = self.ranges.last_mut() {
+            if overlaps(last, range, self.max_gap_size, self.max_range_size) {
+                merge(last, range);
+                return;
+            }
+        }
+
+        self.ranges.push(range.clone());
+    }
+}
diff --git a/src/common/base/tests/it/main.rs b/src/common/base/tests/it/main.rs
@@ -15,6 +15,7 @@
 mod pool;
 mod pool_retry;
 mod progress;
+mod range_merger;
 mod runtime;
 mod runtime_tracker;
 mod stoppable;
diff --git a/src/common/base/tests/it/range_merger.rs b/src/common/base/tests/it/range_merger.rs
@@ -0,0 +1,80 @@
+// Copyright 2022 Datafuse Labs.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::fmt;
+use std::fmt::Formatter;
+
+use common_base::rangemap::RangeMerger;
+use common_exception::Result;
+
+struct Array(Vec<std::ops::Range<u64>>);
+impl fmt::Display for Array {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        for range in &self.0 {
+            write!(f, "[{},{}] ", range.start, range.end)?;
+        }
+        Ok(())
+    }
+}
+
+#[test]
+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, 100);
+    let actual = format!("{}", Array(mr.ranges()));
+    let expect = "[1,12] [13,15] [18,20] ";
+    assert_eq!(actual, expect);
+
+    Ok(())
+}
+
+#[test]
+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];
+
+    // max_gap_size = 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);
+    }
+
+    // max_gap_size = 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);
+    }
+
+    // max_gap_size = 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(())
+}
diff --git a/src/query/settings/src/lib.rs b/src/query/settings/src/lib.rs
@@ -181,6 +181,28 @@ impl Settings {
                 desc: "The maximum number of concurrent IO requests. By default the value is determined automatically.",
                 possible_values: None,
             },
+            // max_storage_io_requests_merge_gap
+            SettingValue {
+                default_value: UserSettingValue::UInt64(16 * 1024),
+                user_setting: UserSetting::create(
+                    "max_storage_io_requests_merge_gap",
+                    UserSettingValue::UInt64(16 * 1024),
+                ),
+                level: ScopeLevel::Session,
+                desc: "The maximum gap bytes of adjusting to merge two IO requests. By default the value is 16KB",
+                possible_values: None,
+            },
+            // max_storage_io_requests_page_size
+            SettingValue {
+                default_value: UserSettingValue::UInt64(256 * 1024),
+                user_setting: UserSetting::create(
+                    "max_storage_io_requests_page_size",
+                    UserSettingValue::UInt64(256 * 1024),
+                ),
+                level: ScopeLevel::Session,
+                desc: "The maximum bytes of one IO request read. By default the value is 256KB",
+                possible_values: None,
+            },
             // flight_client_timeout
             SettingValue {
                 default_value: UserSettingValue::UInt64(60),
@@ -561,6 +583,26 @@ impl Settings {
         self.try_set_u64(key, val, false)
     }
 
+    pub fn get_max_storage_io_requests_merge_gap(&self) -> Result<u64> {
+        let key = "max_storage_io_requests_merge_gap";
+        self.try_get_u64(key)
+    }
+
+    pub fn set_max_storage_io_requests_merge_gap(&self, val: u64) -> Result<()> {
+        let key = "max_storage_io_requests_merge_gap";
+        self.try_set_u64(key, val, false)
+    }
+
+    pub fn get_max_storage_io_requests_page_size(&self) -> Result<u64> {
+        let key = "max_storage_io_requests_page_size";
+        self.try_get_u64(key)
+    }
+
+    pub fn set_max_storage_io_requests_page_size(&self, val: u64) -> Result<()> {
+        let key = "max_storage_io_requests_page_size";
+        self.try_set_u64(key, val, false)
+    }
+
     // Get max_execute_time.
     pub fn get_max_execute_time(&self) -> Result<u64> {
         self.try_get_u64("max_execute_time")
diff --git a/src/query/storages/fuse/src/io/read/block_reader_parquet.rs b/src/query/storages/fuse/src/io/read/block_reader_parquet.rs
@@ -15,6 +15,7 @@
 use std::collections::hash_map::Entry;
 use std::collections::HashMap;
 use std::sync::Arc;
+use std::time::SystemTime;
 
 use common_arrow::arrow::array::Array;
 use common_arrow::arrow::chunk::Chunk;
@@ -28,9 +29,11 @@ use common_arrow::parquet::metadata::ColumnDescriptor;
 use common_arrow::parquet::read::BasicDecompressor;
 use common_arrow::parquet::read::PageMetaData;
 use common_arrow::parquet::read::PageReader;
+use common_base::rangemap::RangeMerger;
 use common_base::runtime::UnlimitedFuture;
 use common_catalog::plan::PartInfoPtr;
 use common_catalog::plan::Projection;
+use common_catalog::table_context::TableContext;
 use common_datablocks::DataBlock;
 use common_datavalues::DataSchemaRef;
 use common_exception::ErrorCode;
@@ -45,6 +48,7 @@ use futures::TryStreamExt;
 use opendal::Object;
 use opendal::Operator;
 use tracing::debug_span;
+use tracing::info;
 use tracing::Instrument;
 
 use crate::fuse_part::FusePartInfo;
@@ -270,8 +274,27 @@ impl BlockReader {
         self.try_next_block(&mut deserializer)
     }
 
-    pub async fn read_columns_data(&self, part: PartInfoPtr) -> Result<Vec<(usize, Vec<u8>)>> {
-        let part = FusePartInfo::from_part(&part)?;
+    /// Merge overlap io request to one.
+    fn merge_io_requests(
+        max_gap_size: u64,
+        max_range_size: u64,
+        part: &PartInfoPtr,
+    ) -> Result<Vec<std::ops::Range<u64>>> {
+        let part = FusePartInfo::from_part(part)?;
+        let ranges = part
+            .columns_meta
+            .values()
+            .map(|v| (v.offset..v.offset + v.len))
+            .collect::<Vec<_>>();
+        Ok(RangeMerger::from_iter(ranges, max_gap_size, max_range_size).ranges())
+    }
+
+    pub async fn read_columns_data(
+        &self,
+        ctx: Arc<dyn TableContext>,
+        raw_part: PartInfoPtr,
+    ) -> Result<Vec<(usize, Vec<u8>)>> {
+        let part = FusePartInfo::from_part(&raw_part)?;
         let columns = self.projection.project_column_leaves(&self.column_leaves)?;
         let indices = Self::build_projection_indices(&columns);
         let mut join_handlers = Vec::with_capacity(indices.len());
@@ -286,7 +309,41 @@ impl BlockReader {
             )));
         }
 
-        futures::future::try_join_all(join_handlers).await
+        let now = SystemTime::now();
+        let res = futures::future::try_join_all(join_handlers).await;
+        let normal_cost = now.elapsed().unwrap().as_millis();
+
+        // Merge io requests.
+        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(
+                self.operator.object(&part.location),
+                index,
+                range.start,
+                range.end,
+            )));
+        }
+        let now = SystemTime::now();
+        let _ = futures::future::try_join_all(merge_io_handlers).await;
+        let merge_cost = now.elapsed().unwrap().as_millis();
+
+        info!(
+            "async read normal partition={},  count={}, took:{} ms",
+            part.location,
+            part.columns_meta.len(),
+            normal_cost,
+        );
+        info!(
+            "async read merge partition={},  count={}, took:{} ms",
+            part.location,
+            ranges.len(),
+            merge_cost,
+        );
+
+        res
     }
 
     pub fn support_blocking_api(&self) -> bool {
@@ -316,6 +373,17 @@ impl BlockReader {
         Ok(results)
     }
 
+    pub async fn read_range(
+        o: Object,
+        index: usize,
+        start: u64,
+        end: u64,
+    ) -> Result<(usize, Vec<u8>)> {
+        let chunk = o.range_read(start..end).await?;
+
+        Ok((index, chunk))
+    }
+
     pub async fn read_column(
         o: Object,
         index: usize,
diff --git a/src/query/storages/fuse/src/operations/fuse_parquet_source.rs b/src/query/storages/fuse/src/operations/fuse_parquet_source.rs
@@ -283,7 +283,10 @@ impl Processor for FuseParquetSource {
     async fn async_process(&mut self) -> Result<()> {
         match std::mem::replace(&mut self.state, State::Finish) {
             State::ReadDataPrewhere(Some(part)) => {
-                let chunks = self.prewhere_reader.read_columns_data(part.clone()).await?;
+                let chunks = self
+                    .prewhere_reader
+                    .read_columns_data(self.ctx.clone(), part.clone())
+                    .await?;
 
                 if self.prewhere_filter.is_some() {
                     self.state = State::PrewhereFilter(part, chunks);
@@ -295,7 +298,9 @@ impl Processor for FuseParquetSource {
             }
             State::ReadDataRemain(part, prewhere_data) => {
                 if let Some(remain_reader) = self.remain_reader.as_ref() {
-                    let chunks = remain_reader.read_columns_data(part.clone()).await?;
+                    let chunks = remain_reader
+                        .read_columns_data(self.ctx.clone(), part.clone())
+                        .await?;
                     self.state = State::Deserialize(part, chunks, Some(prewhere_data));
                     Ok(())
                 } else {
diff --git a/src/query/storages/fuse/src/operations/mutation/deletion/deletion_source.rs b/src/query/storages/fuse/src/operations/mutation/deletion/deletion_source.rs
@@ -298,7 +298,10 @@ impl Processor for DeletionSource {
                 self.index = deletion_part.index;
                 self.origin_stats = deletion_part.cluster_stats.clone();
                 let part = deletion_part.inner_part.clone();
-                let chunks = self.block_reader.read_columns_data(part.clone()).await?;
+                let chunks = self
+                    .block_reader
+                    .read_columns_data(self.ctx.clone(), part.clone())
+                    .await?;
                 self.state = State::FilterData(part, chunks);
             }
             State::ReadRemain {
@@ -307,7 +310,9 @@ impl Processor for DeletionSource {
                 filter,
             } => {
                 if let Some(remain_reader) = self.remain_reader.as_ref() {
-                    let chunks = remain_reader.read_columns_data(part.clone()).await?;
+                    let chunks = remain_reader
+                        .read_columns_data(self.ctx.clone(), part.clone())
+                        .await?;
                     self.state = State::MergeRemain {
                         part,
                         chunks,
