From a3f2990a6cb5636504a748d2ac23111035f772e2 Mon Sep 17 00:00:00 2001 From: Xuanwo Date: Thu, 13 Aug 2026 13:39:34 +0800 Subject: [PATCH] fix(object_store): make range coalescing configurable --- core/core/src/raw/ops.rs | 5 +- .../src/types/operator/operator_futures.rs | 9 +- core/core/src/types/options.rs | 14 +- core/core/src/types/read/reader.rs | 9 + integrations/object_store/README.md | 15 ++ integrations/object_store/src/store.rs | 156 +++++++++++++++--- 6 files changed, 180 insertions(+), 28 deletions(-) diff --git a/core/core/src/raw/ops.rs b/core/core/src/raw/ops.rs index eee51bed2fda..99cdd2ddd16b 100644 --- a/core/core/src/raw/ops.rs +++ b/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 } diff --git a/core/core/src/types/operator/operator_futures.rs b/core/core/src/types/operator/operator_futures.rs index e08b77e86bda..d5353ddebabf 100644 --- a/core/core/src/types/operator/operator_futures.rs +++ b/core/core/src/types/operator/operator_futures.rs @@ -508,8 +508,8 @@ impl>> FutureReader { /// 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. /// @@ -517,7 +517,10 @@ impl>> FutureReader { /// 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: /// /// ``` diff --git a/core/core/src/types/options.rs b/core/core/src/types/options.rs index 4d16e777ed40..5972183c07be 100644 --- a/core/core/src/types/options.rs +++ b/core/core/src/types/options.rs @@ -134,11 +134,14 @@ pub struct ReadOptions { pub chunk: Option, /// 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, /// 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. diff --git a/core/core/src/types/read/reader.rs b/core/core/src/types/read/reader.rs index 274fae936e17..d69e9b64f1e7 100644 --- a/core/core/src/types/read/reader.rs +++ b/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(()) } diff --git a/integrations/object_store/README.md b/integrations/object_store/README.md index 5b5f67e11d35..0c8f3d70ce55 100644 --- a/integrations/object_store/README.md +++ b/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: diff --git a/integrations/object_store/src/store.rs b/integrations/object_store/src/store.rs index 7c92de87bc95..642c9bfb96b5 100644 --- a/integrations/object_store/src/store.rs +++ b/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, 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 = 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, + read_ranges: Option>>>, } impl StatCounterLayer { pub fn new(count: Arc) -> Self { - Self { count } + Self { + count, + read_ranges: None, + } + } + + pub fn with_read_ranges(mut self, read_ranges: Arc>>) -> 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, + read_ranges: Option>>>, + } + + struct ReadCounter { + inner: opendal::raw::oio::Reader, + ranges: Arc>>, + } + + impl opendal::raw::oio::Read for ReadCounter { + async fn open( + &self, + range: BytesRange, + ) -> opendal::Result<( + opendal::raw::RpRead, + Box, + )> { + 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.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};