fix(object_store): make get_ranges coalescing configurable

core/core/src/raw/ops.rs

@@ -479,8 +479,11 @@ impl OpReader {
     }
 
     /// Set the gap size.
+    ///
+    /// Set to `0` to disable merging ranges separated by a gap. Overlapping or
+    /// adjacent ranges are still merged.
     pub fn with_gap(mut self, gap: usize) -> Self {
-        self.gap = Some(gap.max(1));
+        self.gap = Some(gap);
         self
     }
 

core/core/src/types/operator/operator_futures.rs

@@ -508,16 +508,19 @@ impl<F: Future<Output = Result<Reader>>> FutureReader<F> {
 
     /// Controls the optimization strategy for range reads in [`Reader::fetch`].
     ///
-    /// When performing range reads, if the gap between two requested ranges is smaller than
-    /// the configured `gap` size, OpenDAL will merge these ranges into a single read request
+    /// When performing range reads, if the gap between two requested ranges is less than or
+    /// equal to the configured `gap` size, OpenDAL will merge these ranges into a single read request
     /// and discard the unrequested data in between. This helps reduce the number of API calls
     /// to remote storage services.
     ///
     /// This optimization is particularly useful when performing multiple small range reads
     /// that are close to each other, as it reduces the overhead of multiple network requests
     /// at the cost of transferring some additional data.
     ///
-    /// In this example, if two requested ranges are separated by less than 1MiB,
+    /// Set to `0` to disable merging ranges separated by a gap. Overlapping or adjacent ranges
+    /// are still merged.
+    ///
+    /// In this example, if two requested ranges are separated by no more than 1 MiB,
     /// they will be merged into a single read request:
     ///
     /// ```

core/core/src/types/options.rs

@@ -134,11 +134,14 @@ pub struct ReadOptions {
     pub chunk: Option<usize>,
     /// Controls the optimization strategy for range reads in [`crate::Reader::fetch`].
     ///
-    /// When performing range reads, if the gap between two requested ranges is smaller than
-    /// the configured `gap` size, OpenDAL will merge these ranges into a single read request
+    /// When performing range reads, if the gap between two requested ranges is less than or
+    /// equal to the configured `gap` size, OpenDAL will merge these ranges into a single read request
     /// and discard the unrequested data in between. This helps reduce the number of API calls
     /// to remote storage services.
     ///
+    /// Set to `0` to disable merging ranges separated by a gap. Overlapping or adjacent ranges
+    /// are still merged.
+    ///
     /// This optimization is particularly useful when performing multiple small range reads
     /// that are close to each other, as it reduces the overhead of multiple network requests
     /// at the cost of transferring some additional data.
@@ -228,11 +231,14 @@ pub struct ReaderOptions {
     pub chunk: Option<usize>,
     /// Controls the optimization strategy for range reads in [`crate::Reader::fetch`].
     ///
-    /// When performing range reads, if the gap between two requested ranges is smaller than
-    /// the configured `gap` size, OpenDAL will merge these ranges into a single read request
+    /// When performing range reads, if the gap between two requested ranges is less than or
+    /// equal to the configured `gap` size, OpenDAL will merge these ranges into a single read request
     /// and discard the unrequested data in between. This helps reduce the number of API calls
     /// to remote storage services.
     ///
+    /// Set to `0` to disable merging ranges separated by a gap. Overlapping or adjacent ranges
+    /// are still merged.
+    ///
     /// This optimization is particularly useful when performing multiple small range reads
     /// that are close to each other, as it reduces the overhead of multiple network requests
     /// at the cost of transferring some additional data.

core/core/src/types/read/reader.rs

@@ -165,6 +165,8 @@ impl Reader {
     /// non-overlapping ranges. Users may also specify a `gap` to merge
     /// close ranges. Merged ranges will be split by `chunk`, then executed
     /// with `concurrent` and `prefetch`.
+    /// Set `gap` to `0` to avoid merging ranges separated by any bytes.
+    /// Overlapping or adjacent ranges are still merged.
     ///
     /// The returning `Buffer` may share the same underlying memory without
     /// any extra copy.
@@ -909,6 +911,13 @@ mod tests {
         let ranges = vec![0..10, 10..20, 21..30, 40..50, 40..60, 45..59];
         let merged = reader.merge_ranges(ranges);
         assert_eq!(merged, vec![0..30, 40..60]);
+
+        let reader = op.reader_with(path).gap(0).await.unwrap();
+        let ranges = vec![0..10, 10..20, 21..30, 40..50, 40..60, 45..59];
+        let merged = reader.merge_ranges(ranges);
+        assert_eq!(merged, vec![0..20, 21..30, 40..60]);
+
+        assert_eq!(OpReader::new().with_gap(0).gap(), Some(0));
         Ok(())
     }
 

integrations/object_store/README.md

@@ -86,6 +86,21 @@ async fn main() {
 }
 ```
 
+### Tune vectored reads
+
+`OpendalStore` coalesces ranges separated by up to 1 MiB by default, matching
+`object_store::OBJECT_STORE_COALESCE_DEFAULT`. Workloads that need to limit
+over-fetching can configure the gap when constructing the store:
+
+```rust
+let object_store = OpendalStore::new(operator).with_get_ranges_gap(0);
+```
+
+A gap of `0` disables merging ranges separated by bytes. Overlapping or adjacent
+ranges are still merged without over-fetching. Set an intermediate value to trade
+fewer backend requests for additional bytes read. A gap of `0` can be useful for
+HDFS scans that are sensitive to retaining large coalesced backing buffers.
+
 ## WASM support
 
 To build with `wasm32-unknown-unknown` target, you need to enable the `send_wrapper` feature:

integrations/object_store/src/store.rs

@@ -58,6 +58,7 @@ use opendal::{Operator, OperatorInfo};
 use std::collections::HashMap;
 
 const DEFAULT_CONCURRENT: usize = 8;
+const DEFAULT_GET_RANGES_GAP: usize = object_store::OBJECT_STORE_COALESCE_DEFAULT as usize;
 
 fn format_object_attributes(meta: &opendal::Metadata) -> Attributes {
     let mut attributes = Attributes::new();
@@ -181,6 +182,7 @@ fn format_without_stat_error(err: opendal::Error, path: &str) -> object_store::E
 pub struct OpendalStore {
     info: Arc<OperatorInfo>,
     inner: Operator,
+    get_ranges_gap: usize,
 }
 
 impl OpendalStore {
@@ -189,9 +191,20 @@ impl OpendalStore {
         Self {
             info: op.info().into(),
             inner: op,
+            get_ranges_gap: DEFAULT_GET_RANGES_GAP,
         }
     }
 
+    /// Set the maximum gap that [`ObjectStore::get_ranges`] coalesces.
+    ///
+    /// The default is [`object_store::OBJECT_STORE_COALESCE_DEFAULT`] (1 MiB).
+    /// Set this value to `0` to avoid merging ranges separated by any bytes.
+    /// Overlapping or adjacent ranges are still merged without over-fetching.
+    pub fn with_get_ranges_gap(mut self, gap: usize) -> Self {
+        self.get_ranges_gap = gap;
+        self
+    }
+
     /// Get the Operator info.
     pub fn info(&self) -> &OperatorInfo {
         self.info.as_ref()
@@ -391,6 +404,7 @@ impl Debug for OpendalStore {
             .field("name", &self.info.name())
             .field("root", &self.info.root())
             .field("capability", &self.info.capability())
+            .field("get_ranges_gap", &self.get_ranges_gap)
             .finish()
     }
 }
@@ -555,28 +569,20 @@ impl ObjectStore for OpendalStore {
         let raw_location = percent_decode_path(location.as_ref());
         let reader = self
             .inner
-            .reader(&raw_location)
+            .reader_with(&raw_location)
+            .concurrent(DEFAULT_CONCURRENT)
+            .gap(self.get_ranges_gap)
             .into_send()
             .await
             .map_err(|err| format_object_store_error(err, location.as_ref()))?;
 
-        let location_ref: Arc<str> = Arc::from(location.as_ref());
-        futures::stream::iter(ranges.iter().cloned())
-            .map(|range| {
-                let reader = reader.clone();
-                let location_ref = location_ref.clone();
-                async move {
-                    reader
-                        .read(range)
-                        .into_send()
-                        .await
-                        .map(|buf| buf.to_bytes())
-                        .map_err(|err| format_object_store_error(err, &location_ref))
-                }
-            })
-            .buffered(DEFAULT_CONCURRENT)
-            .try_collect()
+        let buffers = reader
+            .fetch(ranges.to_vec())
+            .into_send()
             .await
+            .map_err(|err| format_object_store_error(err, location.as_ref()))?;
+
+        Ok(buffers.into_iter().map(|buf| buf.to_bytes()).collect())
     }
 
     fn delete_stream(
@@ -1014,19 +1020,29 @@ mod tests {
         );
     }
 
-    /// Custom layer that counts stat operations for testing
+    /// Custom layer that counts stat operations and optionally records read ranges.
     mod stat_counter {
         use super::*;
+        use std::sync::Mutex as StdMutex;
         use std::sync::atomic::{AtomicUsize, Ordering};
 
         #[derive(Debug, Clone)]
         pub struct StatCounterLayer {
             count: Arc<AtomicUsize>,
+            read_ranges: Option<Arc<StdMutex<Vec<BytesRange>>>>,
         }
 
         impl StatCounterLayer {
             pub fn new(count: Arc<AtomicUsize>) -> Self {
-                Self { count }
+                Self {
+                    count,
+                    read_ranges: None,
+                }
+            }
+
+            pub fn with_read_ranges(mut self, read_ranges: Arc<StdMutex<Vec<BytesRange>>>) -> Self {
+                self.read_ranges = Some(read_ranges);
+                self
             }
         }
 
@@ -1035,6 +1051,7 @@ mod tests {
                 Arc::new(StatCounterService {
                     srv,
                     count: self.count.clone(),
+                    read_ranges: self.read_ranges.clone(),
                 })
             }
         }
@@ -1043,6 +1060,32 @@ mod tests {
         pub struct StatCounterService {
             srv: opendal::raw::Servicer,
             count: Arc<AtomicUsize>,
+            read_ranges: Option<Arc<StdMutex<Vec<BytesRange>>>>,
+        }
+
+        struct ReadCounter {
+            inner: opendal::raw::oio::Reader,
+            ranges: Arc<StdMutex<Vec<BytesRange>>>,
+        }
+
+        impl opendal::raw::oio::Read for ReadCounter {
+            async fn open(
+                &self,
+                range: BytesRange,
+            ) -> opendal::Result<(
+                opendal::raw::RpRead,
+                Box<dyn opendal::raw::oio::ReadStreamDyn>,
+            )> {
+                self.inner.open(range).await
+            }
+
+            async fn read(
+                &self,
+                range: BytesRange,
+            ) -> opendal::Result<(opendal::raw::RpRead, Buffer)> {
+                self.ranges.lock().unwrap().push(range);
+                self.inner.read(range).await
+            }
         }
 
         impl opendal::raw::Service for StatCounterService {
@@ -1085,7 +1128,14 @@ mod tests {
                 path: &str,
                 args: opendal::raw::OpRead,
             ) -> opendal::Result<Self::Reader> {
-                self.srv.read(ctx, path, args)
+                let reader = self.srv.read(ctx, path, args)?;
+                match &self.read_ranges {
+                    Some(ranges) => Ok(Box::new(ReadCounter {
+                        inner: reader,
+                        ranges: ranges.clone(),
+                    })),
+                    None => Ok(reader),
+                }
             }
 
             fn write(
@@ -1142,6 +1192,72 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn test_get_ranges_gap_configuration() {
+        use std::sync::Mutex as StdMutex;
+        use std::sync::atomic::AtomicUsize;
+
+        let read_ranges = Arc::new(StdMutex::new(Vec::new()));
+        let op = Operator::new(opendal::services::Memory::default())
+            .unwrap()
+            .layer(
+                stat_counter::StatCounterLayer::new(Arc::new(AtomicUsize::new(0)))
+                    .with_read_ranges(read_ranges.clone()),
+            );
+        let store = OpendalStore::new(op);
+        let location = "test_get_ranges_fetch.txt".into();
+        store
+            .put(&location, Bytes::from_static(b"0123456789abcdefgh").into())
+            .await
+            .unwrap();
+
+        let ranges = [15..17, 0..4, 8..10];
+        let result = store.get_ranges(&location, &ranges).await.unwrap();
+        assert_eq!(
+            result,
+            vec![
+                Bytes::from_static(b"fg"),
+                Bytes::from_static(b"0123"),
+                Bytes::from_static(b"89"),
+            ]
+        );
+        assert_eq!(
+            read_ranges.lock().unwrap().as_slice(),
+            &[BytesRange::new(0, Some(17))]
+        );
+
+        read_ranges.lock().unwrap().clear();
+        store
+            .clone()
+            .with_get_ranges_gap(0)
+            .get_ranges(&location, &ranges)
+            .await
+            .unwrap();
+        let mut actual = read_ranges.lock().unwrap().clone();
+        actual.sort_unstable_by_key(BytesRange::offset);
+        assert_eq!(
+            actual,
+            vec![
+                BytesRange::new(0, Some(4)),
+                BytesRange::new(8, Some(2)),
+                BytesRange::new(15, Some(2)),
+            ]
+        );
+
+        read_ranges.lock().unwrap().clear();
+        store
+            .with_get_ranges_gap(4)
+            .get_ranges(&location, &ranges)
+            .await
+            .unwrap();
+        let mut actual = read_ranges.lock().unwrap().clone();
+        actual.sort_unstable_by_key(BytesRange::offset);
+        assert_eq!(
+            actual,
+            vec![BytesRange::new(0, Some(10)), BytesRange::new(15, Some(2))]
+        );
+    }
+
     #[tokio::test]
     async fn test_get_range_no_stat() {
         use std::sync::atomic::{AtomicUsize, Ordering};