From 3fa5ed1dda7088c6a90757e399142bc789859545 Mon Sep 17 00:00:00 2001
From: osobh <om.sobh@gmail.com>
Date: Sat, 26 Sep 2026 16:28:01 -0500
Subject: [PATCH] format: raw data, VDS and VL data over Storage

Every raw-data path has a generic *_in core, with the &[u8] functions as
thin wrappers: data_read (read_raw_data*, read_raw_data_selection,
read_chunked_native), chunked_read (the v1 B-tree chunk index, list_chunks,
the full, cached, sweep and indexed reads), parallel_read, partial_read,
fill_value (read_full_with_fill, apply_to_unallocated_chunks; and
dataset_fill_value_from_storage is now generic), vds (the virtual file
through Storage, external sources still through the resolver),
vl_data (VlResolver<'a, S = [u8]>, read_vl_strings_in, read_vl_bytes_in),
AttributeMessage::read_vl_strings_in and provenance::verify_dataset_in.

With the whole file in memory nothing changes: chunks and contiguous data
are sliced from it as before. Otherwise a chunked read lists its chunks,
fetches their stored bytes with one Storage::read_ranges call per 64 MiB
batch (chunks the cache already holds are not fetched), then decodes as
today; a selection fetches only the chunks it overlaps, and a contiguous
selection only its runs. Each extent's bounds error is the one the slice
code gave, reported when that extent is reached, so errors keep their
order.

Tests: the equivalence harness now reads every dataset's values (whole,
fill-aware, cached, indexed, three selections, VDS, VL strings and
sequences) through the read_at-only storage and requires the slice
results (all 653 corpus files agree); a misbehaving storage (a failing
Nth read, short reads) only ever yields errors or the right values; and
chunked reads are checked to use one read_ranges call.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 crates/clawhdf5-format/src/attribute.rs       |  12 +-
 crates/clawhdf5-format/src/chunked_read.rs    | 437 ++++++++++++---
 crates/clawhdf5-format/src/data_read.rs       | 215 +++++++-
 crates/clawhdf5-format/src/fill_value.rs      |  65 ++-
 crates/clawhdf5-format/src/gather.rs          | 129 +++++
 crates/clawhdf5-format/src/global_heap.rs     |   2 +-
 crates/clawhdf5-format/src/parallel_read.rs   | 105 ++--
 crates/clawhdf5-format/src/partial_read.rs    | 106 +++-
 crates/clawhdf5-format/src/provenance.rs      |  17 +-
 crates/clawhdf5-format/src/storage.rs         | 153 ++++++
 crates/clawhdf5-format/src/vds.rs             | 123 ++++-
 crates/clawhdf5-format/src/vl_data.rs         | 204 ++++---
 .../tests/storage_equivalence.rs              | 514 +++++++++++++++++-
 13 files changed, 1803 insertions(+), 279 deletions(-)

diff --git a/crates/clawhdf5-format/src/attribute.rs b/crates/clawhdf5-format/src/attribute.rs
index 4b30fac..c8f69b5 100644
--- a/crates/clawhdf5-format/src/attribute.rs
+++ b/crates/clawhdf5-format/src/attribute.rs
@@ -336,9 +336,19 @@ impl AttributeMessage {
         file_data: &[u8],
         offset_size: u8,
         length_size: u8,
+    ) -> Result<Vec<String>, FormatError> {
+        self.read_vl_strings_in(file_data, offset_size, length_size)
+    }
+
+    /// [`Self::read_vl_strings`] over any [`Storage`].
+    pub fn read_vl_strings_in<S: Storage + ?Sized>(
+        &self,
+        file_data: &S,
+        offset_size: u8,
+        length_size: u8,
     ) -> Result<Vec<String>, FormatError> {
         let num_elements = self.dataspace.num_elements();
-        vl_data::read_vl_strings(
+        vl_data::read_vl_strings_in(
             file_data,
             &self.raw_data,
             num_elements,
diff --git a/crates/clawhdf5-format/src/chunked_read.rs b/crates/clawhdf5-format/src/chunked_read.rs
index e249168..140cf37 100644
--- a/crates/clawhdf5-format/src/chunked_read.rs
+++ b/crates/clawhdf5-format/src/chunked_read.rs
@@ -6,19 +6,20 @@ extern crate alloc;
 #[cfg(not(feature = "std"))]
 use alloc::{format, vec, vec::Vec};
 
-use crate::addr::to_usize;
+use crate::addr::{checked_addr, to_usize};
 #[cfg(feature = "std")]
 use crate::chunk_cache::{CacheAlignedBuffer, ChunkCache};
 use crate::data_layout::DataLayout;
 use crate::dataspace::Dataspace;
 use crate::datatype::Datatype;
 use crate::error::FormatError;
-use crate::extensible_array::{ExtensibleArrayHeader, read_extensible_array_chunks};
+use crate::extensible_array::{ExtensibleArrayHeader, read_extensible_array_chunks_in};
 use crate::filter_pipeline::FilterPipeline;
 use crate::filters::{DecodeScratch, decompress_chunk_exact_with};
 #[cfg(feature = "std")]
 use crate::filters::{all_filters_skipped, decompress_chunk_exact};
-use crate::fixed_array::{FixedArrayHeader, read_fixed_array_chunks};
+use crate::fixed_array::{FixedArrayHeader, read_fixed_array_chunks_in};
+use crate::storage::{ExtentBytes, Storage, Window, raw_batches, read_exact_at};
 #[cfg(feature = "std")]
 use std::sync::Arc;
 
@@ -238,22 +239,101 @@ type CacheUse<'a> = Option<&'a core::convert::Infallible>;
 /// ([`parallel_read::run_with_helpers`]), so the caller never waits on a
 /// busy pool. The error returned is the first failing chunk's, in `chunks`
 /// order.
+///
+/// With the whole file in memory the chunks are sliced from it. Otherwise
+/// their stored bytes are fetched first, with one
+/// [`Storage::read_ranges`] call per batch of up to
+/// [`crate::storage::RAW_BATCH_BYTES`] (all of them, for most datasets),
+/// and then decoded as above; chunks the cache already holds are not
+/// fetched.
 #[allow(clippy::too_many_arguments)]
-fn fill_from_chunks(
-    file_data: &[u8],
+fn fill_from_chunks<S: Storage + ?Sized>(
+    file_data: &S,
     chunks: &[ChunkInfo],
     pipeline: Option<&FilterPipeline>,
     placer: &ChunkPlacer,
     chunk_total_bytes: usize,
     cache: CacheUse<'_>,
     output: &mut [u8],
+) -> Result<(), FormatError> {
+    let contiguous = file_data.as_contiguous().is_some();
+    let out = OutBuf::new(output);
+    let batches = raw_batches(chunks.len(), contiguous, |i| chunks[i].chunk_size as usize);
+    for batch in batches {
+        fill_batch(
+            file_data,
+            &chunks[batch],
+            pipeline,
+            placer,
+            chunk_total_bytes,
+            cache,
+            &out,
+        )?;
+    }
+    Ok(())
+}
+
+/// Whether a full read looks chunk `c` up in the cache (see
+/// [`fill_from_chunks`]): a filtered chunk of a dataset the cache keeps.
+#[cfg(feature = "std")]
+fn uses_cache(cache: CacheUse<'_>, pipeline: Option<&FilterPipeline>, c: &ChunkInfo) -> bool {
+    matches!(cache, Some((_, _, true)))
+        && pipeline.is_some_and(|pl| !all_filters_skipped(pl, c.filter_mask))
+}
+
+/// [`fill_from_chunks`] for one batch of chunks.
+#[allow(clippy::too_many_arguments)]
+fn fill_batch<S: Storage + ?Sized>(
+    file_data: &S,
+    chunks: &[ChunkInfo],
+    pipeline: Option<&FilterPipeline>,
+    placer: &ChunkPlacer,
+    chunk_total_bytes: usize,
+    cache: CacheUse<'_>,
+    out: &OutBuf<'_>,
 ) -> Result<(), FormatError> {
     let rank = placer.rank;
     let elem_size = placer.elem_size as u32;
-    let out = OutBuf::new(output);
     #[cfg(not(feature = "std"))]
     let _ = cache;
 
+    // Over a backend without the whole file in memory, the decoded chunks
+    // the cache holds are taken now (so a chunk evicted before it is placed
+    // is not left without bytes), and only the others are fetched.
+    #[cfg(feature = "std")]
+    let hits: Vec<Option<Arc<CacheAlignedBuffer>>> = match cache {
+        Some((cache, key, true)) if file_data.as_contiguous().is_none() => chunks
+            .iter()
+            .map(|c| {
+                if c.offsets.len() >= rank && uses_cache(Some((cache, key, true)), pipeline, c) {
+                    cache.get_decompressed_in(key, &c.offsets[..rank])
+                } else {
+                    None
+                }
+            })
+            .collect(),
+        _ => Vec::new(),
+    };
+    let extents: Vec<(u64, usize, bool)> = if file_data.as_contiguous().is_some() {
+        Vec::new()
+    } else {
+        chunks
+            .iter()
+            .enumerate()
+            .map(|(i, c)| {
+                #[cfg(feature = "std")]
+                let wanted = hits.get(i).is_none_or(Option::is_none);
+                #[cfg(not(feature = "std"))]
+                let wanted = {
+                    let _ = i;
+                    true
+                };
+                (c.address, c.chunk_size as usize, wanted)
+            })
+            .collect()
+    };
+    let raw_bytes = ExtentBytes::fetch(file_data, &extents)?;
+
     // Decode chunk `i` and place it. Its callers below run it either on one
     // thread, or on several for chunks whose regions are pairwise disjoint,
     // each chunk once: no two threads ever write the same bytes.
@@ -266,13 +346,18 @@ fn fill_from_chunks(
             )));
         }
         let offsets = &c.offsets[..rank];
-        let c_addr = to_usize(c.address)?;
         let size = c.chunk_size as usize;
-        ensure_len(file_data, c_addr, size)?;
-        let raw = &file_data[c_addr..c_addr + size];
+        #[cfg(feature = "std")]
+        if let Some(Some(hit)) = hits.get(i) {
+            raw_bytes.check(i, c.address, size)?;
+            // SAFETY: see above.
+            unsafe { placer.place(hit, offsets, out) };
+            return Ok(());
+        }
+        let raw = raw_bytes.get(i, c.address, size)?;
         let Some(pl) = pipeline else {
             // SAFETY: see above.
-            unsafe { placer.place(raw, offsets, &out) };
+            unsafe { placer.place(raw, offsets, out) };
             return Ok(());
         };
         // A chunk stored as-is (every filter skipped) is checked and placed
@@ -296,7 +381,7 @@ fn fill_from_chunks(
                 }
             };
             // SAFETY: see above.
-            unsafe { placer.place(&cached, offsets, &out) };
+            unsafe { placer.place(&cached, offsets, out) };
             return Ok(());
         }
         let data = decompress_chunk_exact_with(
@@ -309,7 +394,7 @@ fn fill_from_chunks(
             scratch,
         )?;
         // SAFETY: see above.
-        unsafe { placer.place(data, offsets, &out) };
+        unsafe { placer.place(data, offsets, out) };
         Ok(())
     };
 
@@ -370,7 +455,29 @@ pub fn decompress_all_chunks_with_stats(
     seed: u64,
     num_lanes: Option<usize>,
 ) -> Result<(Vec<Vec<u8>>, PartitionStats), FormatError> {
-    parallel_read::decompress_chunks_lane_partitioned(
+    decompress_all_chunks_with_stats_in(
+        file_data,
+        chunks,
+        pipeline,
+        chunk_total_bytes,
+        element_size,
+        seed,
+        num_lanes,
+    )
+}
+
+/// [`decompress_all_chunks_with_stats`] over any [`Storage`].
+#[cfg(feature = "parallel")]
+pub fn decompress_all_chunks_with_stats_in<S: Storage + ?Sized>(
+    file_data: &S,
+    chunks: &[ChunkInfo],
+    pipeline: &FilterPipeline,
+    chunk_total_bytes: usize,
+    element_size: u32,
+    seed: u64,
+    num_lanes: Option<usize>,
+) -> Result<(Vec<Vec<u8>>, PartitionStats), FormatError> {
+    parallel_read::decompress_chunks_lane_partitioned_in(
         file_data,
         chunks,
         pipeline,
@@ -394,21 +501,6 @@ pub struct ChunkInfo {
     pub address: u64,
 }
 
-/// Checks that `[offset, offset + needed)` fits within `data`, guarding the
-/// addition against `usize` overflow from a crafted near-`usize::MAX` offset.
-fn ensure_len(data: &[u8], offset: usize, needed: usize) -> Result<(), FormatError> {
-    if offset
-        .checked_add(needed)
-        .is_none_or(|end| end > data.len())
-    {
-        return Err(FormatError::UnexpectedEof {
-            expected: offset.saturating_add(needed),
-            available: data.len(),
-        });
-    }
-    Ok(())
-}
-
 /// `elements * elem_size` for sizes that come from the file. Dataspace and
 /// chunk dimensions are untrusted 64-bit fields, so a crafted file can make
 /// the plain product wrap to a small number (or to something enormous).
@@ -588,6 +680,17 @@ pub fn collect_chunk_info(
     ndims: usize,
     offset_size: u8,
     length_size: u8,
+) -> Result<Vec<ChunkInfo>, FormatError> {
+    collect_chunk_info_in(file_data, btree_address, ndims, offset_size, length_size)
+}
+
+/// [`collect_chunk_info`] over any [`Storage`].
+pub fn collect_chunk_info_in<S: Storage + ?Sized>(
+    file_data: &S,
+    btree_address: u64,
+    ndims: usize,
+    offset_size: u8,
+    length_size: u8,
 ) -> Result<Vec<ChunkInfo>, FormatError> {
     let _ = length_size;
     let mut chunks = Vec::new();
@@ -630,6 +733,23 @@ pub fn collect_chunk_info_checked(
     chunk_dimensions: &[u32],
     offset_size: u8,
     length_size: u8,
+) -> Result<Vec<ChunkInfo>, FormatError> {
+    collect_chunk_info_checked_in(
+        file_data,
+        btree_address,
+        chunk_dimensions,
+        offset_size,
+        length_size,
+    )
+}
+
+/// [`collect_chunk_info_checked`] over any [`Storage`].
+pub fn collect_chunk_info_checked_in<S: Storage + ?Sized>(
+    file_data: &S,
+    btree_address: u64,
+    chunk_dimensions: &[u32],
+    offset_size: u8,
+    length_size: u8,
 ) -> Result<Vec<ChunkInfo>, FormatError> {
     let _ = length_size;
     let ndims = chunk_dimensions.len();
@@ -813,8 +933,8 @@ const MAX_CHUNK_BTREE_DEPTH: usize = 64;
 
 /// Parse the v1 B-tree chunk index node at `btree_address` and its
 /// subtree, appending its chunks to `stored` in tree order.
-fn parse_chunk_node(
-    file_data: &[u8],
+fn parse_chunk_node<S: Storage + ?Sized>(
+    file_data: &S,
     btree_address: u64,
     ndims: usize,
     chunk_dimensions: Option<&[u32]>,
@@ -831,21 +951,24 @@ fn parse_chunk_node(
 
     // Parse B-tree v1 header
     let header_size = 8 + os * 2;
-    ensure_len(file_data, offset, header_size)?;
+    let head = Window::read(file_data, offset as u64, header_size)?;
+    head.ensure(0, header_size)?;
+    let h = &head.bytes;
 
-    if &file_data[offset..offset + 4] != b"TREE" {
+    if &h[..4] != b"TREE" {
         return Err(FormatError::InvalidBTreeSignature);
     }
 
-    let node_type = file_data[offset + 4];
+    let node_type = h[4];
     if node_type != 1 {
         return Err(FormatError::InvalidBTreeNodeType(node_type));
     }
 
-    let node_level = file_data[offset + 5];
-    let entries_used = u16::from_le_bytes([file_data[offset + 6], file_data[offset + 7]]) as usize;
+    let node_level = h[5];
+    let entries_used = u16::from_le_bytes([h[6], h[7]]) as usize;
 
-    let mut pos = offset + 8 + os * 2; // skip left/right sibling
+    // Positions below are relative to the node's start.
+    let mut pos = header_size; // skip left/right sibling
 
     // Key: chunk_size(4) + filter_mask(4) + one offset per dimension. The
     // offsets are always 8 bytes each — they are dataset coordinates, not file
@@ -858,28 +981,20 @@ fn parse_chunk_node(
 
     // key[0], child[0], key[1], child[1], ..., key[N-1], child[N-1], key[N]
     let needed = entries_used * (key_size + os) + key_size;
-    ensure_len(file_data, pos, needed)?;
+    let node = Window::read(file_data, offset as u64, header_size + needed)?;
+    node.ensure(pos, needed)?;
+    let d = &node.bytes;
 
     let mut keys = Vec::with_capacity((entries_used + 1) * ndims);
     let mut chunks = Vec::new();
     let mut child_addrs = Vec::new();
     for _ in 0..entries_used {
-        let chunk_size = u32::from_le_bytes([
-            file_data[pos],
-            file_data[pos + 1],
-            file_data[pos + 2],
-            file_data[pos + 3],
-        ]);
-        let filter_mask = u32::from_le_bytes([
-            file_data[pos + 4],
-            file_data[pos + 5],
-            file_data[pos + 6],
-            file_data[pos + 7],
-        ]);
+        let chunk_size = u32::from_le_bytes([d[pos], d[pos + 1], d[pos + 2], d[pos + 3]]);
+        let filter_mask = u32::from_le_bytes([d[pos + 4], d[pos + 5], d[pos + 6], d[pos + 7]]);
         let k = keys.len();
-        read_key_offsets(file_data, pos, ndims, chunk_dimensions, &mut keys)?;
+        read_key_offsets(d, pos, ndims, chunk_dimensions, &mut keys)?;
         pos += key_size;
-        let address = read_offset(file_data, pos, offset_size)?;
+        let address = read_offset(d, pos, offset_size)?;
         pos += os;
         if node_level == 0 {
             chunks.push(stored.len());
@@ -894,7 +1009,7 @@ fn parse_chunk_node(
         }
     }
     // The final key only bounds the node; libhdf5 still checks it.
-    read_key_offsets(file_data, pos, ndims, chunk_dimensions, &mut keys)?;
+    read_key_offsets(d, pos, ndims, chunk_dimensions, &mut keys)?;
 
     let children = if node_level == 0 {
         ChunkChildren::Chunks(chunks)
@@ -984,18 +1099,18 @@ const BT2_CHUNK_FILTERED: u8 = 11;
 /// The width of the stored-size field depends on the largest possible chunk;
 /// rather than re-derive the library's formula it is taken from the record
 /// size the tree header declares, which is what actually governs the bytes.
-fn read_btree_v2_chunks(
-    file_data: &[u8],
+fn read_btree_v2_chunks<S: Storage + ?Sized>(
+    file_data: &S,
     addr: u64,
     chunk_dims: &[usize],
     elem_size: usize,
     offset_size: u8,
     length_size: u8,
 ) -> Result<Vec<ChunkInfo>, FormatError> {
-    use crate::btree_v2::{BTreeV2Header, collect_btree_v2_records};
+    use crate::btree_v2::{BTreeV2Header, collect_btree_v2_records_in};
 
     let bad = |what: &str| FormatError::ChunkedReadError(format!("B-tree v2 chunk index: {what}"));
-    let header = BTreeV2Header::parse(file_data, to_usize(addr)?, offset_size, length_size)?;
+    let header = BTreeV2Header::parse_in(file_data, checked_addr(addr)?, offset_size, length_size)?;
     let rank = chunk_dims.len();
     let os = offset_size as usize;
     let record_size = header.record_size as usize;
@@ -1022,7 +1137,7 @@ fn read_btree_v2_chunks(
     let unfiltered_bytes =
         u32::try_from(unfiltered_bytes).map_err(|_| bad("chunk larger than 4 GiB"))?;
 
-    let records = collect_btree_v2_records(file_data, &header, offset_size, length_size)?;
+    let records = collect_btree_v2_records_in(file_data, &header, offset_size, length_size)?;
     let mut chunks = Vec::with_capacity(records.len());
     for record in &records {
         let data = record.data.as_slice();
@@ -1085,6 +1200,25 @@ pub fn list_chunks(
     elem_size: usize,
     offset_size: u8,
     length_size: u8,
+) -> Result<(Vec<ChunkInfo>, Vec<usize>), FormatError> {
+    list_chunks_in(
+        file_data,
+        layout,
+        dataspace,
+        elem_size,
+        offset_size,
+        length_size,
+    )
+}
+
+/// [`list_chunks`] over any [`Storage`].
+pub fn list_chunks_in<S: Storage + ?Sized>(
+    file_data: &S,
+    layout: &DataLayout,
+    dataspace: &Dataspace,
+    elem_size: usize,
+    offset_size: u8,
+    length_size: u8,
 ) -> Result<(Vec<ChunkInfo>, Vec<usize>), FormatError> {
     let (
         chunk_dimensions,
@@ -1132,9 +1266,13 @@ pub fn list_chunks(
 
     // Collect chunks based on version and index type
     let mut chunks = match (version, chunk_index_type) {
-        (3, _) => {
-            collect_chunk_info_checked(file_data, addr, chunk_dimensions, offset_size, length_size)?
-        }
+        (3, _) => collect_chunk_info_checked_in(
+            file_data,
+            addr,
+            chunk_dimensions,
+            offset_size,
+            length_size,
+        )?,
         (4, Some(1)) => {
             // Single chunk — one chunk covering the entire dataset
             let chunk_byte_size = checked_chunk_byte_len(&chunk_dims, elem_size)?;
@@ -1163,9 +1301,13 @@ pub fn list_chunks(
         (4, Some(3)) => {
             // Fixed Array — use spatial chunk dims only
             let spatial_chunk_dims: &[u32] = &chunk_dimensions[..rank];
-            let header =
-                FixedArrayHeader::parse(file_data, to_usize(addr)?, offset_size, length_size)?;
-            read_fixed_array_chunks(
+            let header = FixedArrayHeader::parse_in(
+                file_data,
+                checked_addr(addr)?,
+                offset_size,
+                length_size,
+            )?;
+            read_fixed_array_chunks_in(
                 file_data,
                 &header,
                 &dataspace.dimensions,
@@ -1179,9 +1321,13 @@ pub fn list_chunks(
         (4, Some(4)) => {
             // Extensible Array — use spatial chunk dims only
             let spatial_chunk_dims: &[u32] = &chunk_dimensions[..rank];
-            let header =
-                ExtensibleArrayHeader::parse(file_data, to_usize(addr)?, offset_size, length_size)?;
-            read_extensible_array_chunks(
+            let header = ExtensibleArrayHeader::parse_in(
+                file_data,
+                checked_addr(addr)?,
+                offset_size,
+                length_size,
+            )?;
+            read_extensible_array_chunks_in(
                 file_data,
                 &header,
                 &dataspace.dimensions,
@@ -1248,7 +1394,28 @@ pub fn list_chunks_for_read(
     offset_size: u8,
     length_size: u8,
 ) -> Result<(Vec<ChunkInfo>, Vec<usize>), FormatError> {
-    let (chunks, chunk_dims) = list_chunks(
+    list_chunks_for_read_in(
+        file_data,
+        layout,
+        dataspace,
+        elem_size,
+        pipeline,
+        offset_size,
+        length_size,
+    )
+}
+
+/// [`list_chunks_for_read`] over any [`Storage`].
+pub fn list_chunks_for_read_in<S: Storage + ?Sized>(
+    file_data: &S,
+    layout: &DataLayout,
+    dataspace: &Dataspace,
+    elem_size: usize,
+    pipeline: Option<&FilterPipeline>,
+    offset_size: u8,
+    length_size: u8,
+) -> Result<(Vec<ChunkInfo>, Vec<usize>), FormatError> {
+    let (chunks, chunk_dims) = list_chunks_in(
         file_data,
         layout,
         dataspace,
@@ -1284,8 +1451,8 @@ pub(crate) type CacheRef<'a> = Option<&'a core::convert::Infallible>;
 /// output with `alloc` (zeroed, `total_bytes` long, as bytes through
 /// `bytes`), and decode every chunk straight into it.
 #[allow(clippy::too_many_arguments)]
-pub(crate) fn read_chunked_full<O>(
-    file_data: &[u8],
+pub(crate) fn read_chunked_full<O, S: Storage + ?Sized>(
+    file_data: &S,
     layout: &DataLayout,
     dataspace: &Dataspace,
     datatype: &Datatype,
@@ -1299,7 +1466,7 @@ pub(crate) fn read_chunked_full<O>(
     check_chunk_element_size(layout, datatype, offset_size)?;
     let elem_size = datatype.type_size() as usize;
     let list = || {
-        list_chunks_for_read(
+        list_chunks_for_read_in(
             file_data,
             layout,
             dataspace,
@@ -1390,6 +1557,27 @@ pub fn read_chunked_data(
     pipeline: Option<&FilterPipeline>,
     offset_size: u8,
     length_size: u8,
+) -> Result<Vec<u8>, FormatError> {
+    read_chunked_data_in(
+        file_data,
+        layout,
+        dataspace,
+        datatype,
+        pipeline,
+        offset_size,
+        length_size,
+    )
+}
+
+/// [`read_chunked_data`] over any [`Storage`].
+pub fn read_chunked_data_in<S: Storage + ?Sized>(
+    file_data: &S,
+    layout: &DataLayout,
+    dataspace: &Dataspace,
+    datatype: &Datatype,
+    pipeline: Option<&FilterPipeline>,
+    offset_size: u8,
+    length_size: u8,
 ) -> Result<Vec<u8>, FormatError> {
     read_chunked_full(
         file_data,
@@ -1422,6 +1610,31 @@ pub fn read_chunked_data_cached(
     offset_size: u8,
     length_size: u8,
     cache: &ChunkCache,
+) -> Result<Vec<u8>, FormatError> {
+    read_chunked_data_cached_in(
+        file_data,
+        layout,
+        dataspace,
+        datatype,
+        pipeline,
+        offset_size,
+        length_size,
+        cache,
+    )
+}
+
+/// [`read_chunked_data_cached`] over any [`Storage`].
+#[cfg(feature = "std")]
+#[allow(clippy::too_many_arguments)]
+pub fn read_chunked_data_cached_in<S: Storage + ?Sized>(
+    file_data: &S,
+    layout: &DataLayout,
+    dataspace: &Dataspace,
+    datatype: &Datatype,
+    pipeline: Option<&FilterPipeline>,
+    offset_size: u8,
+    length_size: u8,
+    cache: &ChunkCache,
 ) -> Result<Vec<u8>, FormatError> {
     read_chunked_full(
         file_data,
@@ -1592,6 +1805,33 @@ pub fn read_chunked_data_sweep(
     length_size: u8,
     cache: &ChunkCache,
     sweep: &mut SweepContext,
+) -> Result<Vec<u8>, FormatError> {
+    read_chunked_data_sweep_in(
+        file_data,
+        layout,
+        dataspace,
+        datatype,
+        pipeline,
+        offset_size,
+        length_size,
+        cache,
+        sweep,
+    )
+}
+
+/// [`read_chunked_data_sweep`] over any [`Storage`].
+#[cfg(feature = "std")]
+#[allow(clippy::too_many_arguments)]
+pub fn read_chunked_data_sweep_in<S: Storage + ?Sized>(
+    file_data: &S,
+    layout: &DataLayout,
+    dataspace: &Dataspace,
+    datatype: &Datatype,
+    pipeline: Option<&FilterPipeline>,
+    offset_size: u8,
+    length_size: u8,
+    cache: &ChunkCache,
+    sweep: &mut SweepContext,
 ) -> Result<Vec<u8>, FormatError> {
     let (chunk_dimensions, version, addr_opt) = match layout {
         DataLayout::Chunked {
@@ -1623,7 +1863,7 @@ pub fn read_chunked_data_sweep(
     // lookup is keyed by this dataset's chunk-index address, so another
     // dataset's index or chunks are never used for this read.
     let chunks = cache.chunks_for(addr, rank, || {
-        list_chunks_for_read(
+        list_chunks_for_read_in(
             file_data,
             layout,
             dataspace,
@@ -1675,8 +1915,8 @@ pub fn read_chunked_data_sweep(
             // Decompress from file
             let c_addr = to_usize(chunk_info.address)?;
             let size = chunk_info.chunk_size as usize;
-            ensure_len(file_data, c_addr, size)?;
-            let raw_chunk = &file_data[c_addr..c_addr + size];
+            let raw_chunk = read_exact_at(file_data, c_addr as u64, size)?;
+            let raw_chunk = &*raw_chunk;
             let dec = if let Some(pl) = pipeline {
                 decompress_chunk_exact(
                     raw_chunk,
@@ -1736,6 +1976,31 @@ pub fn read_chunked_data_indexed(
     offset_size: u8,
     length_size: u8,
     cache: &ChunkCache,
+) -> Result<Vec<u8>, FormatError> {
+    read_chunked_data_indexed_in(
+        file_data,
+        layout,
+        dataspace,
+        datatype,
+        pipeline,
+        offset_size,
+        length_size,
+        cache,
+    )
+}
+
+/// [`read_chunked_data_indexed`] over any [`Storage`].
+#[cfg(feature = "std")]
+#[allow(clippy::too_many_arguments)]
+pub fn read_chunked_data_indexed_in<S: Storage + ?Sized>(
+    file_data: &S,
+    layout: &DataLayout,
+    dataspace: &Dataspace,
+    datatype: &Datatype,
+    pipeline: Option<&FilterPipeline>,
+    offset_size: u8,
+    length_size: u8,
+    cache: &ChunkCache,
 ) -> Result<Vec<u8>, FormatError> {
     let (chunk_dimensions, version, addr_opt) = match layout {
         DataLayout::Chunked {
@@ -1769,7 +2034,7 @@ pub fn read_chunked_data_indexed(
         addr,
         rank,
         || {
-            list_chunks_for_read(
+            list_chunks_for_read_in(
                 file_data,
                 layout,
                 dataspace,
@@ -1786,18 +2051,30 @@ pub fn read_chunked_data_indexed(
     )?;
     let chunk_total_bytes = plan.chunk_total_bytes;
 
+    // The decoded chunks the cache holds, and the stored bytes of the
+    // others: fetched in one batch when the file is not in memory.
+    let hits: Vec<Option<Arc<CacheAlignedBuffer>>> = plan
+        .mappings
+        .iter()
+        .map(|m| cache.get_decompressed_in(addr, &m.coord))
+        .collect();
+    let extents: Vec<(u64, usize, bool)> = plan
+        .mappings
+        .iter()
+        .zip(&hits)
+        .map(|(m, hit)| (m.file_offset, m.file_size as usize, hit.is_none()))
+        .collect();
+    let raw_bytes = ExtentBytes::fetch(file_data, &extents)?;
+
     // Decompress chunks (using LRU cache where possible)
     let mut chunk_buffers: Vec<Arc<CacheAlignedBuffer>> = Vec::with_capacity(plan.mappings.len());
-    for m in &plan.mappings {
+    for (i, (m, hit)) in plan.mappings.iter().zip(hits).enumerate() {
         let (coord, file_offset, file_size, filter_mask) =
             (&m.coord, &m.file_offset, &m.file_size, &m.filter_mask);
-        if let Some(cached) = cache.get_decompressed_in(addr, coord) {
+        if let Some(cached) = hit {
             chunk_buffers.push(cached);
         } else {
-            let c_addr = to_usize(*file_offset)?;
-            let size = *file_size as usize;
-            ensure_len(file_data, c_addr, size)?;
-            let raw_chunk = &file_data[c_addr..c_addr + size];
+            let raw_chunk = raw_bytes.get(i, *file_offset, *file_size as usize)?;
             let decompressed = if let Some(pl) = pipeline {
                 decompress_chunk_exact(
                     raw_chunk,
diff --git a/crates/clawhdf5-format/src/data_read.rs b/crates/clawhdf5-format/src/data_read.rs
index bcf1a18..28bc4f2 100644
--- a/crates/clawhdf5-format/src/data_read.rs
+++ b/crates/clawhdf5-format/src/data_read.rs
@@ -1,7 +1,9 @@
 //! Raw data reading and typed conversion for HDF5 datasets.
 
 #[cfg(not(feature = "std"))]
-use alloc::{collections::BTreeMap, format, string::String, vec, vec::Vec};
+use alloc::{borrow::Cow, collections::BTreeMap, format, string::String, vec, vec::Vec};
+#[cfg(feature = "std")]
+use std::borrow::Cow;
 
 #[cfg(feature = "std")]
 use std::collections::BTreeMap;
@@ -9,14 +11,15 @@ use std::collections::BTreeMap;
 use crate::addr::to_usize;
 #[cfg(feature = "std")]
 use crate::chunk_cache::ChunkCache;
-use crate::chunked_read::read_chunked_data;
+use crate::chunked_read::read_chunked_data_in;
 #[cfg(feature = "std")]
-use crate::chunked_read::{read_chunked_data_cached, read_chunked_data_indexed};
+use crate::chunked_read::{read_chunked_data_cached_in, read_chunked_data_indexed_in};
 use crate::data_layout::DataLayout;
 use crate::dataspace::Dataspace;
 use crate::datatype::{Datatype, DatatypeByteOrder};
 use crate::error::FormatError;
 use crate::filter_pipeline::FilterPipeline;
+use crate::storage::{Storage, read_exact_at};
 
 /// Checks that `[offset, offset + needed)` fits within `data`, guarding the
 /// addition against `usize` overflow from a crafted near-`usize::MAX` offset.
@@ -149,7 +152,17 @@ pub fn read_raw_data(
     dataspace: &Dataspace,
     datatype: &Datatype,
 ) -> Result<Vec<u8>, FormatError> {
-    read_raw_data_full(file_data, layout, dataspace, datatype, None, 8, 8)
+    read_raw_data_in(file_data, layout, dataspace, datatype)
+}
+
+/// [`read_raw_data`] over any [`Storage`].
+pub fn read_raw_data_in<S: Storage + ?Sized>(
+    file_data: &S,
+    layout: &DataLayout,
+    dataspace: &Dataspace,
+    datatype: &Datatype,
+) -> Result<Vec<u8>, FormatError> {
+    read_raw_data_full_in(file_data, layout, dataspace, datatype, None, 8, 8)
 }
 
 /// Resolves a Virtual Dataset source **file name** (as stored in the mapping,
@@ -171,6 +184,27 @@ pub fn read_raw_data_full(
     pipeline: Option<&FilterPipeline>,
     offset_size: u8,
     length_size: u8,
+) -> Result<Vec<u8>, FormatError> {
+    read_raw_data_full_in(
+        file_data,
+        layout,
+        dataspace,
+        datatype,
+        pipeline,
+        offset_size,
+        length_size,
+    )
+}
+
+/// [`read_raw_data_full`] over any [`Storage`].
+pub fn read_raw_data_full_in<S: Storage + ?Sized>(
+    file_data: &S,
+    layout: &DataLayout,
+    dataspace: &Dataspace,
+    datatype: &Datatype,
+    pipeline: Option<&FilterPipeline>,
+    offset_size: u8,
+    length_size: u8,
 ) -> Result<Vec<u8>, FormatError> {
     read_raw_data_full_impl(
         file_data,
@@ -196,6 +230,30 @@ pub fn read_raw_data_full_with_resolver(
     offset_size: u8,
     length_size: u8,
     resolver: Option<&VdsSourceResolver>,
+) -> Result<Vec<u8>, FormatError> {
+    read_raw_data_full_with_resolver_in(
+        file_data,
+        layout,
+        dataspace,
+        datatype,
+        pipeline,
+        offset_size,
+        length_size,
+        resolver,
+    )
+}
+
+/// [`read_raw_data_full_with_resolver`] over any [`Storage`].
+#[allow(clippy::too_many_arguments)]
+pub fn read_raw_data_full_with_resolver_in<S: Storage + ?Sized>(
+    file_data: &S,
+    layout: &DataLayout,
+    dataspace: &Dataspace,
+    datatype: &Datatype,
+    pipeline: Option<&FilterPipeline>,
+    offset_size: u8,
+    length_size: u8,
+    resolver: Option<&VdsSourceResolver>,
 ) -> Result<Vec<u8>, FormatError> {
     read_raw_data_full_impl(
         file_data,
@@ -210,8 +268,8 @@ pub fn read_raw_data_full_with_resolver(
 }
 
 #[allow(clippy::too_many_arguments)]
-fn read_raw_data_full_impl(
-    file_data: &[u8],
+fn read_raw_data_full_impl<S: Storage + ?Sized>(
+    file_data: &S,
     layout: &DataLayout,
     dataspace: &Dataspace,
     datatype: &Datatype,
@@ -242,12 +300,17 @@ fn read_raw_data_full_impl(
             let addr = address.ok_or(FormatError::NoDataAllocated)?;
             let addr = to_usize(addr)?;
             let sz = contiguous_read_len(*size, expected_size)?;
-            ensure_len(file_data, addr, sz)?;
-            let mut out = crate::bulk_alloc::vec_for_bulk(sz);
-            out.extend_from_slice(&file_data[addr..addr + sz]);
-            Ok(out)
+            match read_exact_at(file_data, addr as u64, sz)? {
+                Cow::Borrowed(bytes) => {
+                    let mut out = crate::bulk_alloc::vec_for_bulk(sz);
+                    out.extend_from_slice(bytes);
+                    Ok(out)
+                }
+                // Fetched for this read: already the caller's copy.
+                Cow::Owned(out) => Ok(out),
+            }
         }
-        DataLayout::Chunked { .. } => read_chunked_data(
+        DataLayout::Chunked { .. } => read_chunked_data_in(
             file_data,
             layout,
             dataspace,
@@ -284,9 +347,34 @@ pub fn read_raw_data_cached(
     offset_size: u8,
     length_size: u8,
     cache: &ChunkCache,
+) -> Result<Vec<u8>, FormatError> {
+    read_raw_data_cached_in(
+        file_data,
+        layout,
+        dataspace,
+        datatype,
+        pipeline,
+        offset_size,
+        length_size,
+        cache,
+    )
+}
+
+/// [`read_raw_data_cached`] over any [`Storage`].
+#[cfg(feature = "std")]
+#[allow(clippy::too_many_arguments)]
+pub fn read_raw_data_cached_in<S: Storage + ?Sized>(
+    file_data: &S,
+    layout: &DataLayout,
+    dataspace: &Dataspace,
+    datatype: &Datatype,
+    pipeline: Option<&FilterPipeline>,
+    offset_size: u8,
+    length_size: u8,
+    cache: &ChunkCache,
 ) -> Result<Vec<u8>, FormatError> {
     match layout {
-        DataLayout::Chunked { .. } => read_chunked_data_cached(
+        DataLayout::Chunked { .. } => read_chunked_data_cached_in(
             file_data,
             layout,
             dataspace,
@@ -296,7 +384,7 @@ pub fn read_raw_data_cached(
             length_size,
             cache,
         ),
-        _ => read_raw_data_full(
+        _ => read_raw_data_full_in(
             file_data,
             layout,
             dataspace,
@@ -325,9 +413,34 @@ pub fn read_raw_data_indexed(
     offset_size: u8,
     length_size: u8,
     cache: &ChunkCache,
+) -> Result<Vec<u8>, FormatError> {
+    read_raw_data_indexed_in(
+        file_data,
+        layout,
+        dataspace,
+        datatype,
+        pipeline,
+        offset_size,
+        length_size,
+        cache,
+    )
+}
+
+/// [`read_raw_data_indexed`] over any [`Storage`].
+#[cfg(feature = "std")]
+#[allow(clippy::too_many_arguments)]
+pub fn read_raw_data_indexed_in<S: Storage + ?Sized>(
+    file_data: &S,
+    layout: &DataLayout,
+    dataspace: &Dataspace,
+    datatype: &Datatype,
+    pipeline: Option<&FilterPipeline>,
+    offset_size: u8,
+    length_size: u8,
+    cache: &ChunkCache,
 ) -> Result<Vec<u8>, FormatError> {
     match layout {
-        DataLayout::Chunked { .. } => read_chunked_data_indexed(
+        DataLayout::Chunked { .. } => read_chunked_data_indexed_in(
             file_data,
             layout,
             dataspace,
@@ -337,7 +450,7 @@ pub fn read_raw_data_indexed(
             length_size,
             cache,
         ),
-        _ => read_raw_data_full(
+        _ => read_raw_data_full_in(
             file_data,
             layout,
             dataspace,
@@ -366,6 +479,30 @@ pub fn read_raw_data_selection(
     offset_size: u8,
     length_size: u8,
     selection: &crate::selection::Selection,
+) -> Result<Vec<u8>, FormatError> {
+    read_raw_data_selection_in(
+        file_data,
+        layout,
+        dataspace,
+        datatype,
+        pipeline,
+        offset_size,
+        length_size,
+        selection,
+    )
+}
+
+/// [`read_raw_data_selection`] over any [`Storage`].
+#[allow(clippy::too_many_arguments)]
+pub fn read_raw_data_selection_in<S: Storage + ?Sized>(
+    file_data: &S,
+    layout: &DataLayout,
+    dataspace: &Dataspace,
+    datatype: &Datatype,
+    pipeline: Option<&FilterPipeline>,
+    offset_size: u8,
+    length_size: u8,
+    selection: &crate::selection::Selection,
 ) -> Result<Vec<u8>, FormatError> {
     use crate::selection::Selection;
 
@@ -375,7 +512,7 @@ pub fn read_raw_data_selection(
     // Read only what the selection's bounding box touches when that is
     // possible; everything below is the decode-everything-then-pick path,
     // kept for the cases `partial_read` declines.
-    if let Some(selected) = crate::partial_read::read_selection(
+    if let Some(selected) = crate::partial_read::read_selection_in(
         file_data,
         layout,
         dataspace,
@@ -390,7 +527,7 @@ pub fn read_raw_data_selection(
 
     match selection {
         Selection::All => {
-            return read_raw_data_full(
+            return read_raw_data_full_in(
                 file_data,
                 layout,
                 dataspace,
@@ -410,7 +547,7 @@ pub fn read_raw_data_selection(
     match layout {
         DataLayout::Compact { .. } | DataLayout::Contiguous { .. } => {
             // Read all data, then extract the selection
-            let full_data = read_raw_data_full(
+            let full_data = read_raw_data_full_in(
                 file_data,
                 layout,
                 dataspace,
@@ -434,7 +571,7 @@ pub fn read_raw_data_selection(
             // implicit-index generator, which then indexed past the rank and
             // panicked — only to decode the full dataset anyway.
             crate::chunked_read::chunk_geometry(chunk_dimensions, *version, dataspace, elem_size)?;
-            let full_data = read_raw_data_full(
+            let full_data = read_raw_data_full_in(
                 file_data,
                 layout,
                 dataspace,
@@ -447,7 +584,7 @@ pub fn read_raw_data_selection(
         }
         DataLayout::Virtual { .. } => {
             // Assemble the full virtual dataset, then apply the read selection.
-            let full_data = read_raw_data_full(
+            let full_data = read_raw_data_full_in(
                 file_data,
                 layout,
                 dataspace,
@@ -471,8 +608,8 @@ pub fn read_raw_data_selection(
 /// would report differently from the stored dataspace (unlimited mappings).
 /// Use [`crate::vds::read_virtual_dataset`] to read those.
 #[allow(clippy::too_many_arguments)]
-fn read_virtual_data(
-    file_data: &[u8],
+fn read_virtual_data<S: Storage + ?Sized>(
+    file_data: &S,
     layout: &DataLayout,
     dataspace: &Dataspace,
     datatype: &Datatype,
@@ -483,7 +620,7 @@ fn read_virtual_data(
     let wrapped =
         resolver.map(|r| move |name: &str| -> Result<Option<Vec<u8>>, FormatError> { Ok(r(name)) });
     let wrapped_ref = wrapped.as_ref().map(|w| w as &crate::vds::VdsFileResolver);
-    let v = crate::vds::read_virtual_dataset(
+    let v = crate::vds::read_virtual_dataset_in(
         file_data,
         layout,
         dataspace,
@@ -885,6 +1022,33 @@ pub fn read_chunked_native<T: NativeElement>(
     offset_size: u8,
     length_size: u8,
     cache: Option<&ChunkCache>,
+) -> Result<Option<Vec<T>>, FormatError> {
+    read_chunked_native_in(
+        messages,
+        file_data,
+        layout,
+        dataspace,
+        datatype,
+        pipeline,
+        offset_size,
+        length_size,
+        cache,
+    )
+}
+
+/// [`read_chunked_native`] over any [`Storage`].
+#[cfg(feature = "std")]
+#[allow(clippy::too_many_arguments)]
+pub fn read_chunked_native_in<T: NativeElement, S: Storage + ?Sized>(
+    messages: &[crate::object_header::HeaderMessage],
+    file_data: &S,
+    layout: &DataLayout,
+    dataspace: &Dataspace,
+    datatype: &Datatype,
+    pipeline: Option<&FilterPipeline>,
+    offset_size: u8,
+    length_size: u8,
+    cache: Option<&ChunkCache>,
 ) -> Result<Option<Vec<T>>, FormatError> {
     use crate::fill_value;
     use crate::message_type::MessageType;
@@ -919,8 +1083,9 @@ pub fn read_chunked_native<T: NativeElement>(
         },
         |values| bytes_of_mut(values),
     )?;
-    let fill = fill_value::dataset_fill_value_in(file_data, messages, offset_size, length_size)?;
-    fill_value::apply_to_unallocated_chunks(
+    let fill =
+        fill_value::dataset_fill_value_from_storage(file_data, messages, offset_size, length_size)?;
+    fill_value::apply_to_unallocated_chunks_in(
         bytes_of_mut(&mut values),
         file_data,
         layout,
diff --git a/crates/clawhdf5-format/src/fill_value.rs b/crates/clawhdf5-format/src/fill_value.rs
index f0f442e..ac23044 100644
--- a/crates/clawhdf5-format/src/fill_value.rs
+++ b/crates/clawhdf5-format/src/fill_value.rs
@@ -13,7 +13,7 @@
 use alloc::{format, vec, vec::Vec};
 
 use crate::addr::to_usize;
-use crate::chunked_read::{alloc_output, checked_byte_len, list_chunks};
+use crate::chunked_read::{alloc_output, checked_byte_len, list_chunks_in};
 use crate::data_layout::DataLayout;
 use crate::dataspace::Dataspace;
 use crate::error::FormatError;
@@ -122,10 +122,11 @@ pub fn dataset_fill_value_in(
 }
 
 /// [`dataset_fill_value_in`] with the file behind any
-/// [`Storage`](crate::storage::Storage). (The trait is not imported here:
-/// its `len` would shadow the slice method in this module.)
-pub fn dataset_fill_value_from_storage(
-    file: &dyn crate::storage::Storage,
+/// [`Storage`](crate::storage::Storage) (a `&dyn Storage` too). (The trait
+/// is not imported here: its `len` would shadow the slice method in this
+/// module.)
+pub fn dataset_fill_value_from_storage<S: crate::storage::Storage + ?Sized>(
+    file: &S,
     messages: &[HeaderMessage],
     offset_size: u8,
     length_size: u8,
@@ -212,6 +213,30 @@ pub fn read_full_with_fill<E: From<FormatError>>(
     offset_size: u8,
     length_size: u8,
     read: impl FnOnce() -> Result<Vec<u8>, E>,
+) -> Result<Vec<u8>, E> {
+    read_full_with_fill_in(
+        messages,
+        file_data,
+        layout,
+        dataspace,
+        elem_size,
+        offset_size,
+        length_size,
+        read,
+    )
+}
+
+/// [`read_full_with_fill`] over any [`Storage`](crate::storage::Storage).
+#[allow(clippy::too_many_arguments)]
+pub fn read_full_with_fill_in<E: From<FormatError>, S: crate::storage::Storage + ?Sized>(
+    messages: &[HeaderMessage],
+    file_data: &S,
+    layout: &DataLayout,
+    dataspace: &Dataspace,
+    elem_size: usize,
+    offset_size: u8,
+    length_size: u8,
+    read: impl FnOnce() -> Result<Vec<u8>, E>,
 ) -> Result<Vec<u8>, E> {
     // A dataset with external raw data also has no data address in this
     // file. It is NOT unallocated — its values live elsewhere — so it must
@@ -222,12 +247,12 @@ pub fn read_full_with_fill<E: From<FormatError>>(
     {
         return Err(FormatError::ExternalDataFilesUnsupported.into());
     }
-    let fill = dataset_fill_value_in(file_data, messages, offset_size, length_size)?;
+    let fill = dataset_fill_value_from_storage(file_data, messages, offset_size, length_size)?;
     if !has_storage(layout) {
         return Ok(filled_dataset(dataspace, elem_size, fill.as_deref())?);
     }
     let mut output = read()?;
-    apply_to_unallocated_chunks(
+    apply_to_unallocated_chunks_in(
         &mut output,
         file_data,
         layout,
@@ -253,6 +278,30 @@ pub fn apply_to_unallocated_chunks(
     fill: Option<&[u8]>,
     offset_size: u8,
     length_size: u8,
+) -> Result<(), FormatError> {
+    apply_to_unallocated_chunks_in(
+        output,
+        file_data,
+        layout,
+        dataspace,
+        elem_size,
+        fill,
+        offset_size,
+        length_size,
+    )
+}
+
+/// [`apply_to_unallocated_chunks`] over any [`Storage`](crate::storage::Storage).
+#[allow(clippy::too_many_arguments)]
+pub fn apply_to_unallocated_chunks_in<S: crate::storage::Storage + ?Sized>(
+    output: &mut [u8],
+    file_data: &S,
+    layout: &DataLayout,
+    dataspace: &Dataspace,
+    elem_size: usize,
+    fill: Option<&[u8]>,
+    offset_size: u8,
+    length_size: u8,
 ) -> Result<(), FormatError> {
     let Some(fill) = fill.filter(|f| f.len() == elem_size && !is_default(Some(f))) else {
         return Ok(());
@@ -260,7 +309,7 @@ pub fn apply_to_unallocated_chunks(
     if !matches!(layout, DataLayout::Chunked { .. }) || elem_size == 0 {
         return Ok(());
     }
-    let (chunks, chunk_dims) = list_chunks(
+    let (chunks, chunk_dims) = list_chunks_in(
         file_data,
         layout,
         dataspace,
diff --git a/crates/clawhdf5-format/src/gather.rs b/crates/clawhdf5-format/src/gather.rs
index 22d2ae7..2b8a4ab 100644
--- a/crates/clawhdf5-format/src/gather.rs
+++ b/crates/clawhdf5-format/src/gather.rs
@@ -13,6 +13,7 @@ use alloc::{vec, vec::Vec};
 use crate::data_read::NativeElement;
 use crate::error::FormatError;
 use crate::selection::Selection;
+use crate::storage::Storage;
 
 /// Row-major element strides of `dims` (the last dimension has stride 1).
 fn strides(dims: &[u64]) -> Vec<u64> {
@@ -261,6 +262,134 @@ pub(crate) fn gather<T: NativeElement>(
     Ok(out)
 }
 
+/// [`gather`] of bytes (`T = u8`) from a dataset that is not in memory: the
+/// dataset's `src_len` bytes start at `base` in `file`, which must hold all
+/// of them (the caller checks). The selection's runs are collected first,
+/// adjacent ones merged, and fetched with one [`Storage::read_ranges`] call,
+/// so only the selected bytes are read. Same checks and errors as
+/// [`gather`].
+pub(crate) fn gather_storage<S: Storage + ?Sized>(
+    file: &S,
+    base: u64,
+    src_len: usize,
+    dims: &[u64],
+    elem_size: usize,
+    selection: &Selection,
+) -> Result<Vec<u8>, FormatError> {
+    if elem_size == 0 {
+        return Err(FormatError::DataSizeMismatch {
+            expected: 1,
+            actual: elem_size,
+        });
+    }
+    let n_elements = match selection {
+        Selection::None => 0,
+        Selection::Hyperslab { count, block, .. } => count
+            .iter()
+            .zip(block)
+            .try_fold(1u64, |acc, (&c, &b)| acc.checked_mul(c.checked_mul(b)?))
+            .ok_or_else(|| FormatError::Overflow("hyperslab count x block overflows".into()))?,
+        Selection::Points(points) => points.len() as u64,
+        Selection::All => {
+            return Err(FormatError::SelectionOutOfBounds(
+                "gather does not take Selection::All".into(),
+            ));
+        }
+    };
+    let out_bytes = crate::chunked_read::checked_byte_len(n_elements, elem_size)?;
+    // Runs as (byte offset in the dataset, byte length), in output order.
+    let mut runs: Vec<(usize, usize)> = Vec::new();
+    let mut total = 0usize;
+    let mut failed = false;
+    let mut collect = |first: u64, n: u64| {
+        if failed {
+            return;
+        }
+        let range = usize::try_from(first)
+            .ok()
+            .and_then(|f| f.checked_mul(elem_size))
+            .zip(
+                usize::try_from(n)
+                    .ok()
+                    .and_then(|n| n.checked_mul(elem_size)),
+            )
+            .and_then(|(at, len)| Some((at, len, at.checked_add(len)?)));
+        match range {
+            Some((at, len, end)) if end <= src_len && total + len <= out_bytes => {
+                match runs.last_mut() {
+                    Some((a, l)) if *a + *l == at => *l += len,
+                    _ => runs.push((at, len)),
+                }
+                total += len;
+            }
+            _ => failed = true,
+        }
+    };
+    let mut bad_point = false;
+    match selection {
+        Selection::Hyperslab {
+            start,
+            stride,
+            count,
+            block,
+        } => {
+            let rank = dims.len();
+            if [start.len(), stride.len(), count.len(), block.len()] != [rank; 4] {
+                return Err(FormatError::SelectionOutOfBounds(
+                    "hyperslab rank does not match dataset rank".into(),
+                ));
+            }
+            hyperslab_runs(dims, start, stride, count, block, &mut collect);
+        }
+        Selection::Points(points) => {
+            let strides = strides(dims);
+            let mut coalesce = Coalesce {
+                start: 0,
+                len: 0,
+                emit: &mut collect,
+            };
+            for p in points {
+                if p.len() != dims.len() || p.iter().zip(dims).any(|(c, n)| c >= n) {
+                    bad_point = true;
+                    break;
+                }
+                let at = p
+                    .iter()
+                    .zip(&strides)
+                    .fold(0u64, |acc, (c, s)| acc.wrapping_add(c.wrapping_mul(*s)));
+                coalesce.push(at, 1);
+            }
+            coalesce.flush();
+        }
+        Selection::None | Selection::All => {}
+    }
+    if failed || bad_point || total != out_bytes {
+        return Err(FormatError::SelectionOutOfBounds(
+            "selection addresses elements outside the dataset".into(),
+        ));
+    }
+    let ranges: Vec<core::ops::Range<u64>> = runs
+        .iter()
+        .map(|&(at, len)| base + at as u64..base + (at + len) as u64)
+        .collect();
+    let fetched = file.read_ranges(&ranges)?;
+    if fetched.len() != ranges.len() {
+        return Err(FormatError::Storage(
+            "read_ranges returned the wrong number of ranges".into(),
+        ));
+    }
+    let mut out = crate::bulk_alloc::vec_for_bulk(out_bytes);
+    for (bytes, &(_, len)) in fetched.iter().zip(&runs) {
+        let Some(b) = bytes.get(..len) else {
+            return Err(FormatError::Storage(
+                "short read inside the file (the storage shrank or the backend failed)".into(),
+            ));
+        };
+        out.extend_from_slice(b);
+    }
+    Ok(out)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/crates/clawhdf5-format/src/global_heap.rs b/crates/clawhdf5-format/src/global_heap.rs
index 9861e8f..be69189 100644
--- a/crates/clawhdf5-format/src/global_heap.rs
+++ b/crates/clawhdf5-format/src/global_heap.rs
@@ -152,7 +152,7 @@ impl GlobalHeapCollection {
     /// Read the collection at `offset` and index its objects: the
     /// collection's bytes, its offset as a `usize`, and the index (with
     /// file offsets).
-    fn read_collection<S: Storage + ?Sized>(
+    pub(crate) fn read_collection<S: Storage + ?Sized>(
         file: &S,
         offset: u64,
         length_size: u8,
diff --git a/crates/clawhdf5-format/src/parallel_read.rs b/crates/clawhdf5-format/src/parallel_read.rs
index 92fb1fa..75a4bd3 100644
--- a/crates/clawhdf5-format/src/parallel_read.rs
+++ b/crates/clawhdf5-format/src/parallel_read.rs
@@ -7,12 +7,30 @@
 //! The lane assignment is seeded by dataset metadata so repeated reads of
 //! the same region produce identical partitions (cache-friendly, reproducible).
 
-use crate::addr::to_usize;
 use crate::chunked_read::ChunkInfo;
 use crate::error::FormatError;
 use crate::filter_pipeline::FilterPipeline;
 use crate::filters::decompress_chunk_exact;
 use crate::lane_partition::{self, LaneStats, PartitionStats};
+use crate::storage::{ExtentBytes, Storage};
+
+/// The stored bytes of every chunk in `chunks`, fetched in one
+/// [`Storage::read_ranges`] call when the file is not in memory (each
+/// chunk's bounds error is reported when that chunk is decoded, as before).
+fn fetch_all<'a, S: Storage + ?Sized>(
+    file_data: &'a S,
+    chunks: &[ChunkInfo],
+) -> Result<ExtentBytes<'a>, FormatError> {
+    let extents: Vec<(u64, usize, bool)> = if file_data.as_contiguous().is_some() {
+        Vec::new()
+    } else {
+        chunks
+            .iter()
+            .map(|c| (c.address, c.chunk_size as usize, true))
+            .collect()
+    };
+    ExtentBytes::fetch(file_data, &extents)
+}
 
 /// Threshold: only use parallel decompression when chunk count exceeds this.
 const PARALLEL_THRESHOLD: usize = 4;
@@ -190,6 +208,27 @@ pub fn decompress_chunks_lane_partitioned(
     element_size: u32,
     seed: u64,
     num_lanes: Option<usize>,
+) -> Result<(Vec<Vec<u8>>, PartitionStats), FormatError> {
+    decompress_chunks_lane_partitioned_in(
+        file_data,
+        chunks,
+        pipeline,
+        chunk_total_bytes,
+        element_size,
+        seed,
+        num_lanes,
+    )
+}
+
+/// [`decompress_chunks_lane_partitioned`] over any [`Storage`].
+pub fn decompress_chunks_lane_partitioned_in<S: Storage + ?Sized>(
+    file_data: &S,
+    chunks: &[ChunkInfo],
+    pipeline: &FilterPipeline,
+    chunk_total_bytes: usize,
+    element_size: u32,
+    seed: u64,
+    num_lanes: Option<usize>,
 ) -> Result<(Vec<Vec<u8>>, PartitionStats), FormatError> {
     use rayon::prelude::*;
 
@@ -199,6 +238,7 @@ pub fn decompress_chunks_lane_partitioned(
             .unwrap_or(1)
     });
 
+    let raw_bytes = fetch_all(file_data, chunks)?;
     let assignments = lane_partition::partition_chunks(chunks.len(), lanes, seed);
     let num_lanes = assignments.len();
 
@@ -211,19 +251,8 @@ pub fn decompress_chunks_lane_partitioned(
 
             for &index in &indices {
                 let chunk_info = &chunks[index];
-                let c_addr = to_usize(chunk_info.address)?;
                 let size = chunk_info.chunk_size as usize;
-
-                if c_addr
-                    .checked_add(size)
-                    .is_none_or(|end| end > file_data.len())
-                {
-                    return Err(FormatError::UnexpectedEof {
-                        expected: c_addr.saturating_add(size),
-                        available: file_data.len(),
-                    });
-                }
-                let raw_chunk = &file_data[c_addr..c_addr + size];
+                let raw_chunk = raw_bytes.get(index, chunk_info.address, size)?;
 
                 let decompressed = decompress_chunk_exact(
                     raw_chunk,
@@ -282,25 +311,27 @@ pub fn decompress_chunks_parallel(
     pipeline: &FilterPipeline,
     chunk_total_bytes: usize,
     element_size: u32,
+) -> Result<Vec<Vec<u8>>, FormatError> {
+    decompress_chunks_parallel_in(file_data, chunks, pipeline, chunk_total_bytes, element_size)
+}
+
+/// [`decompress_chunks_parallel`] over any [`Storage`].
+pub fn decompress_chunks_parallel_in<S: Storage + ?Sized>(
+    file_data: &S,
+    chunks: &[ChunkInfo],
+    pipeline: &FilterPipeline,
+    chunk_total_bytes: usize,
+    element_size: u32,
 ) -> Result<Vec<Vec<u8>>, FormatError> {
     use rayon::prelude::*;
 
+    let raw_bytes = fetch_all(file_data, chunks)?;
     let results: Result<Vec<DecompressedChunk>, FormatError> = chunks
         .par_iter()
         .enumerate()
         .map(|(index, chunk_info)| {
-            let c_addr = to_usize(chunk_info.address)?;
             let size = chunk_info.chunk_size as usize;
-            if c_addr
-                .checked_add(size)
-                .is_none_or(|end| end > file_data.len())
-            {
-                return Err(FormatError::UnexpectedEof {
-                    expected: c_addr.saturating_add(size),
-                    available: file_data.len(),
-                });
-            }
-            let raw_chunk = &file_data[c_addr..c_addr + size];
+            let raw_chunk = raw_bytes.get(index, chunk_info.address, size)?;
 
             let decompressed = decompress_chunk_exact(
                 raw_chunk,
@@ -331,20 +362,22 @@ pub fn decompress_chunks_sequential(
     chunk_total_bytes: usize,
     element_size: u32,
 ) -> Result<Vec<Vec<u8>>, FormatError> {
+    decompress_chunks_sequential_in(file_data, chunks, pipeline, chunk_total_bytes, element_size)
+}
+
+/// [`decompress_chunks_sequential`] over any [`Storage`].
+pub fn decompress_chunks_sequential_in<S: Storage + ?Sized>(
+    file_data: &S,
+    chunks: &[ChunkInfo],
+    pipeline: Option<&FilterPipeline>,
+    chunk_total_bytes: usize,
+    element_size: u32,
+) -> Result<Vec<Vec<u8>>, FormatError> {
+    let raw_bytes = fetch_all(file_data, chunks)?;
     let mut result = Vec::with_capacity(chunks.len());
-    for chunk_info in chunks {
-        let c_addr = to_usize(chunk_info.address)?;
+    for (i, chunk_info) in chunks.iter().enumerate() {
         let size = chunk_info.chunk_size as usize;
-        if c_addr
-            .checked_add(size)
-            .is_none_or(|end| end > file_data.len())
-        {
-            return Err(FormatError::UnexpectedEof {
-                expected: c_addr.saturating_add(size),
-                available: file_data.len(),
-            });
-        }
-        let raw_chunk = &file_data[c_addr..c_addr + size];
+        let raw_chunk = raw_bytes.get(i, chunk_info.address, size)?;
 
         let decompressed = if let Some(pl) = pipeline {
             decompress_chunk_exact(
diff --git a/crates/clawhdf5-format/src/partial_read.rs b/crates/clawhdf5-format/src/partial_read.rs
index ad5dc8e..a9f630c 100644
--- a/crates/clawhdf5-format/src/partial_read.rs
+++ b/crates/clawhdf5-format/src/partial_read.rs
@@ -18,7 +18,7 @@ use alloc::{format, vec, vec::Vec};
 #[cfg(feature = "std")]
 use std::string as alloc_or_std;
 
-use crate::chunked_read::{alloc_output, checked_byte_len, list_chunks_for_read};
+use crate::chunked_read::{alloc_output, checked_byte_len, list_chunks_for_read_in};
 use crate::data_layout::DataLayout;
 use crate::data_read::extract_selection_from_buffer;
 use crate::dataspace::Dataspace;
@@ -26,6 +26,7 @@ use crate::error::FormatError;
 use crate::filter_pipeline::FilterPipeline;
 use crate::filters::{all_filters_skipped, decompress_chunk_exact_with};
 use crate::selection::Selection;
+use crate::storage::{ExtentBytes, Storage};
 
 /// The smallest axis-aligned box containing every selected element, as
 /// `(start, extent)` per dimension. `None` when there is nothing to gain or
@@ -256,6 +257,30 @@ pub fn read_selection(
     offset_size: u8,
     length_size: u8,
     selection: &Selection,
+) -> Result<Option<Vec<u8>>, FormatError> {
+    read_selection_in(
+        file_data,
+        layout,
+        dataspace,
+        elem_size,
+        pipeline,
+        offset_size,
+        length_size,
+        selection,
+    )
+}
+
+/// [`read_selection`] over any [`Storage`].
+#[allow(clippy::too_many_arguments)]
+pub fn read_selection_in<S: Storage + ?Sized>(
+    file_data: &S,
+    layout: &DataLayout,
+    dataspace: &Dataspace,
+    elem_size: usize,
+    pipeline: Option<&FilterPipeline>,
+    offset_size: u8,
+    length_size: u8,
+    selection: &Selection,
 ) -> Result<Option<Vec<u8>>, FormatError> {
     let dims = &dataspace.dimensions;
     if dims.is_empty() || elem_size == 0 {
@@ -276,14 +301,33 @@ pub fn read_selection(
         validate(selection, dims)?;
         let base = usize::try_from(*address)
             .map_err(|_| FormatError::Overflow("data address exceeds usize".into()))?;
-        let data = file_data
-            .get(base..)
-            .and_then(|d| d.get(..checked_byte_len(total, elem_size).ok()?))
-            .ok_or(FormatError::UnexpectedEof {
-                expected: base,
-                available: file_data.len(),
-            })?;
-        return crate::gather::gather::<u8>(data, dims, elem_size, selection).map(Some);
+        let file_len = crate::storage::len_usize(file_data);
+        let eof = FormatError::UnexpectedEof {
+            expected: base,
+            available: file_len,
+        };
+        if let Some(all) = file_data.as_contiguous() {
+            let data = all
+                .get(base..)
+                .and_then(|d| d.get(..checked_byte_len(total, elem_size).ok()?))
+                .ok_or(eof)?;
+            return crate::gather::gather::<u8>(data, dims, elem_size, selection).map(Some);
+        }
+        // Not in memory: the same bounds check, then only the selected runs
+        // are read.
+        let len = checked_byte_len(total, elem_size)
+            .ok()
+            .filter(|&len| base <= file_len && len <= file_len - base)
+            .ok_or(eof)?;
+        return crate::gather::gather_storage(
+            file_data,
+            base as u64,
+            len,
+            dims,
+            elem_size,
+            selection,
+        )
+        .map(Some);
     }
     let Some((box_start, box_extent)) = bounding_box(selection, dims) else {
         return Ok(None);
@@ -303,7 +347,7 @@ pub fn read_selection(
             btree_address: Some(_),
             ..
         } => {
-            let (chunks, chunk_dims) = list_chunks_for_read(
+            let (chunks, chunk_dims) = list_chunks_for_read_in(
                 file_data,
                 layout,
                 dataspace,
@@ -315,29 +359,37 @@ pub fn read_selection(
             let rank = dims.len();
             let chunk_shape: Vec<u64> = chunk_dims.iter().map(|&d| d as u64).collect();
             let chunk_bytes = crate::chunked_read::checked_chunk_byte_len(&chunk_dims, elem_size)?;
-            // Chunks are decoded into this thread's reusable buffers.
-            crate::chunked_read::with_scratch(|scratch| -> Result<(), FormatError> {
-                for chunk in &chunks {
+            // The chunks overlapping the box, in index order.
+            let wanted: Vec<&crate::chunked_read::ChunkInfo> = chunks
+                .iter()
+                .filter(|chunk| {
                     if chunk.offsets.len() < rank || chunk.address == u64::MAX {
-                        continue;
+                        return false;
                     }
                     let origin = &chunk.offsets[..rank];
-                    let overlaps = (0..rank).all(|d| {
+                    (0..rank).all(|d| {
                         origin[d] < box_start[d] + box_extent[d]
                             && origin[d].saturating_add(chunk_shape[d]) > box_start[d]
-                    });
-                    if !overlaps {
-                        continue;
-                    }
-                    let at = usize::try_from(chunk.address)
+                    })
+                })
+                .collect();
+            // Their stored bytes, in one batch when the file is not in memory.
+            let extents: Vec<(u64, usize, bool)> = if file_data.as_contiguous().is_some() {
+                Vec::new()
+            } else {
+                wanted
+                    .iter()
+                    .map(|c| (c.address, c.chunk_size as usize, true))
+                    .collect()
+            };
+            let raw_bytes = ExtentBytes::fetch(file_data, &extents)?;
+            // Chunks are decoded into this thread's reusable buffers.
+            crate::chunked_read::with_scratch(|scratch| -> Result<(), FormatError> {
+                for (i, chunk) in wanted.iter().enumerate() {
+                    let origin = &chunk.offsets[..rank];
+                    usize::try_from(chunk.address)
                         .map_err(|_| FormatError::Overflow("chunk address exceeds usize".into()))?;
-                    let raw = at
-                        .checked_add(chunk.chunk_size as usize)
-                        .and_then(|end| file_data.get(at..end))
-                        .ok_or(FormatError::UnexpectedEof {
-                            expected: at.saturating_add(chunk.chunk_size as usize),
-                            available: file_data.len(),
-                        })?;
+                    let raw = raw_bytes.get(i, chunk.address, chunk.chunk_size as usize)?;
                     // Mirrors the full-read path: filter-mask bit i set means
                     // filter i was not applied to this chunk.
                     let data: &[u8] = match pipeline {
diff --git a/crates/clawhdf5-format/src/provenance.rs b/crates/clawhdf5-format/src/provenance.rs
index ee74994..0bdccab 100644
--- a/crates/clawhdf5-format/src/provenance.rs
+++ b/crates/clawhdf5-format/src/provenance.rs
@@ -13,7 +13,6 @@ use sha2::{Digest, Sha256};
 
 use crate::attribute::AttributeMessage;
 use crate::data_layout::DataLayout;
-use crate::data_read::read_raw_data;
 use crate::dataspace::Dataspace;
 use crate::datatype::Datatype;
 use crate::error::FormatError;
@@ -128,10 +127,20 @@ pub fn verify_dataset(
     header: &ObjectHeader,
     offset_size: u8,
     length_size: u8,
+) -> Result<VerifyResult, FormatError> {
+    verify_dataset_in(file_data, header, offset_size, length_size)
+}
+
+/// [`verify_dataset`] over any [`Storage`](crate::storage::Storage).
+pub fn verify_dataset_in<S: crate::storage::Storage + ?Sized>(
+    file_data: &S,
+    header: &ObjectHeader,
+    offset_size: u8,
+    length_size: u8,
 ) -> Result<VerifyResult, FormatError> {
     // 1. Extract all attributes (compact + dense).
     let attrs =
-        crate::attribute::extract_attributes_full(file_data, header, offset_size, length_size)?;
+        crate::attribute::extract_attributes_full_in(file_data, header, offset_size, length_size)?;
 
     // 2. Find the stored hash.
     let stored_hash = attrs
@@ -174,7 +183,7 @@ pub fn verify_dataset(
         .transpose()?;
 
     let raw = match &dl {
-        DataLayout::Chunked { .. } => crate::chunked_read::read_chunked_data(
+        DataLayout::Chunked { .. } => crate::chunked_read::read_chunked_data_in(
             file_data,
             &dl,
             &ds,
@@ -183,7 +192,7 @@ pub fn verify_dataset(
             offset_size,
             length_size,
         )?,
-        _ => read_raw_data(file_data, &dl, &ds, &dt)?,
+        _ => crate::data_read::read_raw_data_in(file_data, &dl, &ds, &dt)?,
     };
 
     // 4. Compare.
diff --git a/crates/clawhdf5-format/src/storage.rs b/crates/clawhdf5-format/src/storage.rs
index 555ac78..0dd0540 100644
--- a/crates/clawhdf5-format/src/storage.rs
+++ b/crates/clawhdf5-format/src/storage.rs
@@ -336,6 +336,159 @@ pub fn read_upto<S: Storage + ?Sized>(
     Ok(bytes)
 }
 
+/// Most stored bytes fetched by one [`Storage::read_ranges`] call when a
+/// read gathers many extents (a chunked dataset's chunks): a larger read is
+/// fetched and decoded batch by batch, so a remote backend never holds more
+/// than this much undecoded data per read.
+pub(crate) const RAW_BATCH_BYTES: usize = 64 << 20;
+
+/// The stored bytes of a list of extents (chunks, contiguous runs), fetched
+/// together: [`Storage::read_ranges`] is called once for all of them, so a
+/// remote backend can coalesce and parallelise the requests.
+///
+/// With the whole file in memory nothing is fetched: [`Self::get`] slices
+/// it, as the slice readers did. Either way an extent that does not lie in
+/// the file is the error the slice readers gave for it
+/// ([`FormatError::UnexpectedEof`] with its end and the file length, or
+/// [`FormatError::Overflow`] for an address past this platform's `usize`),
+/// reported when that extent is asked for — so a read reports the first
+/// failing extent in its own order, whatever fails after it.
+pub(crate) enum ExtentBytes<'a> {
+    /// The whole file.
+    Contiguous(&'a [u8]),
+    /// Each extent's bytes, or its bounds error.
+    Fetched(Vec<Extent<'a>>),
+}
+
+/// One extent of [`ExtentBytes::Fetched`].
+pub(crate) enum Extent<'a> {
+    /// Its bytes.
+    Bytes(Cow<'a, [u8]>),
+    /// In the file, but not fetched (the caller did not want its bytes).
+    NotFetched,
+    /// The error reading it gives.
+    Err(FormatError),
+}
+
+impl<'a> ExtentBytes<'a> {
+    /// Fetch `extents` (`(address, length, wanted)`): the bytes of those
+    /// `wanted`, and the bounds check of all of them.
+    pub(crate) fn fetch<S: Storage + ?Sized>(
+        file: &'a S,
+        extents: &[(u64, usize, bool)],
+    ) -> Result<Self, FormatError> {
+        if let Some(all) = file.as_contiguous() {
+            return Ok(ExtentBytes::Contiguous(all));
+        }
+        let file_len = len_usize(file);
+        let mut ranges = Vec::new();
+        let mut out = Vec::with_capacity(extents.len());
+        // Positions in `out` of the extents being read, in `ranges` order.
+        let mut slots = Vec::new();
+        for &(addr, len, wanted) in extents {
+            let checked =
+                crate::addr::to_usize(addr).and_then(|start| match start.checked_add(len) {
+                    Some(end) if end <= file_len => Ok(()),
+                    _ => Err(FormatError::UnexpectedEof {
+                        expected: start.saturating_add(len),
+                        available: file_len,
+                    }),
+                });
+            match checked {
+                Ok(()) if wanted => {
+                    slots.push(out.len());
+                    ranges.push(addr..addr + len as u64);
+                    out.push(Extent::NotFetched);
+                }
+                Ok(()) => out.push(Extent::NotFetched),
+                Err(e) => out.push(Extent::Err(e)),
+            }
+        }
+        if !ranges.is_empty() {
+            let got = file.read_ranges(&ranges)?;
+            if got.len() != ranges.len() {
+                return Err(FormatError::Storage(
+                    "read_ranges returned the wrong number of ranges".into(),
+                ));
+            }
+            for ((slot, bytes), r) in slots.into_iter().zip(got).zip(&ranges) {
+                if (bytes.len() as u64) < r.end - r.start {
+                    return Err(short_read());
+                }
+                out[slot] = Extent::Bytes(bytes);
+            }
+        }
+        Ok(ExtentBytes::Fetched(out))
+    }
+
+    /// Whether extent `i` (at `addr`, `len` bytes, as passed to
+    /// [`Self::fetch`]) lies in the file: its bounds error if not.
+    pub(crate) fn check(&self, i: usize, addr: u64, len: usize) -> Result<(), FormatError> {
+        match self {
+            ExtentBytes::Contiguous(_) => self.get(i, addr, len).map(|_| ()),
+            ExtentBytes::Fetched(v) => match v.get(i) {
+                Some(Extent::Err(e)) => Err(e.clone()),
+                Some(_) => Ok(()),
+                None => Err(not_fetched()),
+            },
+        }
+    }
+
+    /// Extent `i`'s bytes (at `addr`, `len` bytes, as passed to
+    /// [`Self::fetch`]).
+    pub(crate) fn get(&self, i: usize, addr: u64, len: usize) -> Result<&[u8], FormatError> {
+        match self {
+            ExtentBytes::Contiguous(all) => {
+                let start = crate::addr::to_usize(addr)?;
+                start
+                    .checked_add(len)
+                    .and_then(|end| all.get(start..end))
+                    .ok_or(FormatError::UnexpectedEof {
+                        expected: start.saturating_add(len),
+                        available: <[u8]>::len(all),
+                    })
+            }
+            ExtentBytes::Fetched(v) => match v.get(i) {
+                Some(Extent::Bytes(b)) => Ok(b),
+                Some(Extent::Err(e)) => Err(e.clone()),
+                _ => Err(not_fetched()),
+            },
+        }
+    }
+}
+
+#[cold]
+fn not_fetched() -> FormatError {
+    FormatError::Storage("an extent that was not fetched was asked for".into())
+}
+
+/// Split `n` extents, whose sizes `size(i)` gives, into consecutive batches
+/// of at most [`RAW_BATCH_BYTES`] (at least one extent each): the ranges of
+/// `0..n` to fetch together. With the whole file in memory (`contiguous`)
+/// there is nothing to fetch, and one batch.
+pub(crate) fn raw_batches(
+    n: usize,
+    contiguous: bool,
+    size: impl Fn(usize) -> usize,
+) -> Vec<Range<usize>> {
+    if contiguous || n == 0 {
+        return core::iter::once(0..n).collect();
+    }
+    let mut out = Vec::new();
+    let (mut start, mut bytes) = (0, 0usize);
+    for i in 0..n {
+        let s = size(i);
+        if i > start && bytes.saturating_add(s) > RAW_BATCH_BYTES {
+            out.push(start..i);
+            start = i;
+            bytes = 0;
+        }
+        bytes = bytes.saturating_add(s);
+    }
+    out.push(start..n);
+    out
+}
+
 /// Borrow the whole file for a code path that has not been converted to
 /// [`Storage`] yet. On a backend without a contiguous view this is the
 /// clean [`FormatError::ContiguousStorageRequired`] error, never a guess.
diff --git a/crates/clawhdf5-format/src/vds.rs b/crates/clawhdf5-format/src/vds.rs
index df1b068..92ad4e7 100644
--- a/crates/clawhdf5-format/src/vds.rs
+++ b/crates/clawhdf5-format/src/vds.rs
@@ -15,12 +15,13 @@
 #[cfg(not(feature = "std"))]
 use alloc::{format, string::String, vec, vec::Vec};
 
-use crate::addr::to_usize;
+use crate::addr::{checked_addr, to_usize};
 use crate::data_layout::{DataLayout, VdsMapping, parse_vds_mappings};
 use crate::dataspace::Dataspace;
 use crate::datatype::Datatype;
 use crate::error::FormatError;
 use crate::selection::{SerializedSelection, UNLIMITED};
+use crate::storage::Storage;
 
 /// Resolves the name of an external VDS source file, as stored in the
 /// mapping, to that file's bytes.
@@ -192,8 +193,8 @@ fn non_unlimited_elements(sel: &SerializedSelection, skip: usize) -> Option<u64>
 }
 
 /// Load and decode the mapping list of a virtual layout.
-fn load_mappings(
-    file_data: &[u8],
+fn load_mappings<S: Storage + ?Sized>(
+    file_data: &S,
     layout: &DataLayout,
     length_size: u8,
 ) -> Result<Vec<Mapping>, FormatError> {
@@ -208,8 +209,11 @@ fn load_mappings(
     let Some(addr) = *global_heap_address else {
         return Ok(Vec::new());
     };
-    let coll =
-        crate::global_heap::GlobalHeapCollection::parse(file_data, to_usize(addr)?, length_size)?;
+    let coll = crate::global_heap::GlobalHeapCollection::parse_in(
+        file_data,
+        checked_addr(addr)?,
+        length_size,
+    )?;
     let index = u16::try_from(*global_heap_index)
         .map_err(|_| vds_err("VDS mapping heap index out of range"))?;
     let obj = coll
@@ -330,7 +334,11 @@ enum Step {
 /// Work out the extent libhdf5 gives the virtual dataset
 /// (`H5D__virtual_set_extent_unlim`, default view `H5D_VDS_LAST_AVAILABLE`
 /// with a printf gap of 0) and how much of each unlimited mapping is read.
-fn plan(mappings: &[Mapping], stored: &[u64], sources: &mut Sources) -> Result<Plan, FormatError> {
+fn plan<S: Storage + ?Sized>(
+    mappings: &[Mapping],
+    stored: &[u64],
+    sources: &mut Sources<'_, '_, S>,
+) -> Result<Plan, FormatError> {
     let overflow = || FormatError::Overflow("VDS extent overflow".into());
     let rank = stored.len();
     let mut new_dims: Vec<Option<u64>> = vec![None; rank];
@@ -472,6 +480,25 @@ pub fn virtual_dataset_extent(
     _offset_size: u8,
     length_size: u8,
     resolver: Option<&VdsFileResolver>,
+) -> Result<Vec<u64>, FormatError> {
+    virtual_dataset_extent_in(
+        file_data,
+        layout,
+        dataspace,
+        _offset_size,
+        length_size,
+        resolver,
+    )
+}
+
+/// [`virtual_dataset_extent`] over any [`Storage`].
+pub fn virtual_dataset_extent_in<S: Storage + ?Sized>(
+    file_data: &S,
+    layout: &DataLayout,
+    dataspace: &Dataspace,
+    _offset_size: u8,
+    length_size: u8,
+    resolver: Option<&VdsFileResolver>,
 ) -> Result<Vec<u64>, FormatError> {
     let mappings = load_mappings(file_data, layout, length_size)?;
     if mappings.iter().all(|m| m.kind == Kind::Fixed) {
@@ -498,6 +525,30 @@ pub fn read_virtual_dataset(
     _offset_size: u8,
     length_size: u8,
     resolver: Option<&VdsFileResolver>,
+) -> Result<VirtualData, FormatError> {
+    read_virtual_dataset_in(
+        file_data,
+        layout,
+        dataspace,
+        datatype,
+        fill,
+        _offset_size,
+        length_size,
+        resolver,
+    )
+}
+
+/// [`read_virtual_dataset`] over any [`Storage`].
+#[allow(clippy::too_many_arguments)]
+pub fn read_virtual_dataset_in<S: Storage + ?Sized>(
+    file_data: &S,
+    layout: &DataLayout,
+    dataspace: &Dataspace,
+    datatype: &Datatype,
+    fill: Option<&[u8]>,
+    _offset_size: u8,
+    length_size: u8,
+    resolver: Option<&VdsFileResolver>,
 ) -> Result<VirtualData, FormatError> {
     let mappings = load_mappings(file_data, layout, length_size)?;
     let mut sources = Sources::new(file_data, resolver);
@@ -511,7 +562,7 @@ pub fn read_virtual_dataset(
     let mut data = crate::chunked_read::alloc_output(crate::chunked_read::checked_byte_len(
         total, elem_size,
     )?)?;
-    if let Some(fill) = fill.filter(|f| f.len() == elem_size && f.iter().any(|&b| b != 0)) {
+    if let Some(fill) = fill.filter(|f| <[u8]>::len(f) == elem_size && f.iter().any(|&b| b != 0)) {
         for element in data.chunks_exact_mut(elem_size) {
             element.copy_from_slice(fill);
         }
@@ -780,14 +831,17 @@ struct SourceData {
 
 /// Source files and datasets, fetched on demand. The most recently used
 /// external file is kept, since consecutive mappings usually share one.
-struct Sources<'a, 'r> {
-    file_data: &'a [u8],
+///
+/// The virtual dataset's own file (`"."`) is read through its [`Storage`];
+/// an external source file is loaded whole, through the resolver.
+struct Sources<'a, 'r, S: Storage + ?Sized> {
+    file_data: &'a S,
     resolver: Option<&'r VdsFileResolver<'r>>,
     cached_file: Option<(String, Option<Vec<u8>>)>,
 }
 
-impl<'a, 'r> Sources<'a, 'r> {
-    fn new(file_data: &'a [u8], resolver: Option<&'r VdsFileResolver<'r>>) -> Self {
+impl<'a, 'r, S: Storage + ?Sized> Sources<'a, 'r, S> {
+    fn new(file_data: &'a S, resolver: Option<&'r VdsFileResolver<'r>>) -> Self {
         Sources {
             file_data,
             resolver,
@@ -795,11 +849,9 @@ impl<'a, 'r> Sources<'a, 'r> {
         }
     }
 
-    /// The bytes of source file `name`, or `None` if it does not exist.
-    fn file(&mut self, name: &str) -> Result<Option<&[u8]>, FormatError> {
-        if name == "." {
-            return Ok(Some(self.file_data));
-        }
+    /// The bytes of external source file `name` (not `"."`), or `None` if it
+    /// does not exist.
+    fn external(&mut self, name: &str) -> Result<Option<&[u8]>, FormatError> {
         if self.cached_file.as_ref().is_none_or(|(n, _)| n != name) {
             let resolver = self.resolver.ok_or_else(|| {
                 vds_err("external-file virtual dataset sources require a file resolver")
@@ -821,7 +873,10 @@ impl<'a, 'r> Sources<'a, 'r> {
     /// The extent of source dataset `path` in file `file`, or `None` when
     /// either does not exist.
     fn dims(&mut self, file: &str, path: &str) -> Result<Option<Vec<u64>>, FormatError> {
-        let Some(bytes) = self.file(file)? else {
+        if file == "." {
+            return Ok(open_source(self.file_data, path)?.map(|s| s.dataspace.dimensions));
+        }
+        let Some(bytes) = self.external(file)? else {
             return Ok(None);
         };
         Ok(open_source(bytes, path)?.map(|s| s.dataspace.dimensions))
@@ -846,7 +901,13 @@ impl<'a, 'r> Sources<'a, 'r> {
                  from another file is not supported"
             )));
         }
-        let Some(bytes) = self.file(file)? else {
+        if file == "." {
+            let Some(src) = open_source(self.file_data, path)? else {
+                return Ok(None);
+            };
+            return read_source(self.file_data, src, path, datatype).map(Some);
+        }
+        let Some(bytes) = self.external(file)? else {
             return Ok(None);
         };
         let Some(src) = open_source(bytes, path)? else {
@@ -910,19 +971,23 @@ fn source_message<'h>(
 
 /// Open source dataset `path` of the file in `file_data`, or `None` if there
 /// is no such object (libhdf5 reads a missing source as fill).
-fn open_source(file_data: &[u8], path: &str) -> Result<Option<OpenSource>, FormatError> {
+fn open_source<S: Storage + ?Sized>(
+    file_data: &S,
+    path: &str,
+) -> Result<Option<OpenSource>, FormatError> {
     use crate::message_type::MessageType;
-    use crate::shared_message::message_data_with_sohm;
+    use crate::shared_message::message_data_with_sohm_in as message_data_with_sohm;
 
-    // `file_data` starts at the superblock (see `Sources::file`).
-    let sb = crate::superblock::Superblock::parse(file_data, 0)?;
+    // `file_data` starts at the superblock (see `Sources::external`).
+    let sb = crate::superblock::Superblock::parse_in(file_data, 0)?;
     let (os, ls) = (sb.offset_size, sb.length_size);
-    let addr = match crate::group_v2::resolve_path_any(file_data, &sb, path) {
+    let addr = match crate::group_v2::resolve_path_any_in(file_data, &sb, path) {
         Ok(a) => a,
         Err(FormatError::PathNotFound(_)) => return Ok(None),
         Err(e) => return Err(e),
     };
-    let header = crate::object_header::ObjectHeader::parse(file_data, to_usize(addr)?, os, ls)?;
+    let header =
+        crate::object_header::ObjectHeader::parse_in(file_data, checked_addr(addr)?, os, ls)?;
     let mut src = OpenSource {
         offset_size: os,
         length_size: ls,
@@ -941,15 +1006,15 @@ fn open_source(file_data: &[u8], path: &str) -> Result<Option<OpenSource>, Forma
 
 /// Read an opened source dataset in full (its own fill value applied to
 /// unallocated chunks).
-fn read_source(
-    file_data: &[u8],
+fn read_source<S: Storage + ?Sized>(
+    file_data: &S,
     src: OpenSource,
     path: &str,
     datatype: &Datatype,
 ) -> Result<SourceData, FormatError> {
     use crate::filter_pipeline::FilterPipeline;
     use crate::message_type::MessageType;
-    use crate::shared_message::message_data_with_sohm;
+    use crate::shared_message::message_data_with_sohm_in as message_data_with_sohm;
 
     let (os, ls) = (src.offset_size, src.length_size);
     let dt_msg = source_message(&src, path, MessageType::Datatype)?;
@@ -987,7 +1052,7 @@ fn read_source(
             message_data_with_sohm(file_data, m, os, ls).and_then(|d| FilterPipeline::parse(&d))
         })
         .transpose()?;
-    let raw = crate::fill_value::read_full_with_fill(
+    let raw = crate::fill_value::read_full_with_fill_in(
         &src.header.messages,
         file_data,
         &layout,
@@ -996,7 +1061,7 @@ fn read_source(
         os,
         ls,
         || {
-            crate::data_read::read_raw_data_full(
+            crate::data_read::read_raw_data_full_in(
                 file_data,
                 &layout,
                 &src.dataspace,
diff --git a/crates/clawhdf5-format/src/vl_data.rs b/crates/clawhdf5-format/src/vl_data.rs
index a998a62..f37638b 100644
--- a/crates/clawhdf5-format/src/vl_data.rs
+++ b/crates/clawhdf5-format/src/vl_data.rs
@@ -5,9 +5,9 @@
 //! `sequence_length(4 LE) + collection_address(offset_size LE) + object_index(4 LE)`.
 
 #[cfg(not(feature = "std"))]
-use alloc::{collections::BTreeMap, format, string::String, vec, vec::Vec};
+use alloc::{borrow::Cow, collections::BTreeMap, format, string::String, vec, vec::Vec};
 #[cfg(feature = "std")]
-use std::collections::BTreeMap;
+use std::{borrow::Cow, collections::BTreeMap};
 
 use crate::addr::to_usize;
 use crate::error::FormatError;
@@ -137,13 +137,15 @@ pub fn check_element_size(stored_size: u32, offset_size: u8) -> Result<(), Forma
 
 /// A collection's objects, located in the file data but not copied:
 /// `(index, offset, size)` of the first object with each index, sorted by
-/// index.
-struct CachedCollection {
+/// index. Over a storage without the whole file in memory, also the
+/// collection's bytes (`(offset, bytes)`), read once when it is indexed.
+struct CachedCollection<'a> {
     objects: Vec<(u16, usize, usize)>,
+    bytes: Option<(usize, Cow<'a, [u8]>)>,
 }
 
-impl CachedCollection {
-    fn new(index: GlobalHeapIndex) -> Self {
+impl<'a> CachedCollection<'a> {
+    fn new(index: GlobalHeapIndex, bytes: Option<(usize, Cow<'a, [u8]>)>) -> Self {
         let mut objects: Vec<(u16, usize, usize)> = index
             .objects
             .iter()
@@ -152,12 +154,16 @@ impl CachedCollection {
         // Stable, so the first object with a repeated index is kept.
         objects.sort_by_key(|o| o.0);
         objects.dedup_by_key(|o| o.0);
-        Self { objects }
+        Self { objects, bytes }
     }
 
     /// What this entry costs to keep, in bytes (roughly).
     fn cost(&self) -> usize {
-        64 + self.objects.len() * core::mem::size_of::<(u16, usize, usize)>()
+        let held = match &self.bytes {
+            Some((_, Cow::Owned(b))) => b.len(),
+            _ => 0,
+        };
+        64 + self.objects.len() * core::mem::size_of::<(u16, usize, usize)>() + held
     }
 
     fn get(&self, index: u32) -> Option<(usize, usize)> {
@@ -170,6 +176,8 @@ impl CachedCollection {
 /// How many bytes of collection indexes a [`VlResolver`] keeps before it
 /// drops them and starts again. Values are never copied into the cache, so
 /// this bounds what a read retains however many collections it visits.
+/// (Over a storage without the whole file in memory the collections' bytes
+/// are kept too, and count against this.)
 const CACHE_BUDGET: usize = 32 << 20;
 
 /// Resolves variable-length elements against a file's global heap, parsing
@@ -185,11 +193,18 @@ const CACHE_BUDGET: usize = 32 << 20;
 /// that overlap one another are refused (libhdf5 never writes them), so a
 /// file cannot make the resolver parse the same bytes as the objects of
 /// many collections.
-pub struct VlResolver<'a> {
-    file_data: &'a [u8],
+///
+/// The file is any [`Storage`](crate::storage::Storage) (`S`, a slice by default). Over one without
+/// the whole file in memory each collection is read once, when first used,
+/// and kept (within the budget above); [`Self::strings`],
+/// [`Self::string_bytes`] and [`Self::sequences`] work over any storage,
+/// [`Self::element`] and [`Self::string_element`], which borrow from the
+/// file, over a slice.
+pub struct VlResolver<'a, S: crate::storage::Storage + ?Sized = [u8]> {
+    file_data: &'a S,
     offset_size: u8,
     length_size: u8,
-    cache: BTreeMap<u64, CachedCollection>,
+    cache: BTreeMap<u64, CachedCollection<'a>>,
     cached_bytes: usize,
     budget: usize,
     /// Start → end of every collection parsed so far (kept when the cache
@@ -201,6 +216,56 @@ impl<'a> VlResolver<'a> {
     /// A resolver over `file_data` (the file from its superblock on), with
     /// the superblock's offset and length sizes.
     pub fn new(file_data: &'a [u8], offset_size: u8, length_size: u8) -> Self {
+        Self::new_in(file_data, offset_size, length_size)
+    }
+
+    /// One element (the first [`element_size`](Self::element_size) bytes of
+    /// `elem`) of a variable-length sequence whose base type is `base_size`
+    /// bytes: its `length × base_size` bytes, or `None` for a null element
+    /// (heap address 0).
+    pub fn element(
+        &mut self,
+        elem: &[u8],
+        base_size: usize,
+    ) -> Result<Option<&'a [u8]>, FormatError> {
+        let vl = parse_vl_references(elem, 1, self.offset_size)?;
+        let vl = &vl[0];
+        if vl.collection_address == 0 {
+            return Ok(None);
+        }
+        let (start, size) = self.locate(vl)?;
+        let data = &self.file_data[start..start + size];
+        check_object_size(vl, data.len(), base_size)?;
+        Ok(Some(data))
+    }
+
+    /// One variable-length string element: its bytes up to the first NUL,
+    /// or `None` for a null element (h5dump prints it as `NULL`, h5py
+    /// returns it as empty).
+    pub fn string_element(&mut self, elem: &[u8]) -> Result<Option<&'a [u8]>, FormatError> {
+        Ok(self.element(elem, 1)?.map(cut_at_nul))
+    }
+}
+
+/// `data_len`, the size of `vl`'s heap object, against the `length ×
+/// base_size` bytes the element says it holds.
+fn check_object_size(vl: &VlElement, data_len: usize, base_size: usize) -> Result<(), FormatError> {
+    let expected = (vl.length as usize)
+        .checked_mul(base_size)
+        .ok_or_else(|| FormatError::Overflow("variable-length element size".into()))?;
+    if data_len != expected {
+        return Err(FormatError::VlDataError(format!(
+            "global heap object {} in the collection at {} holds {data_len} bytes; the element \
+             says {} × {base_size}",
+            vl.object_index, vl.collection_address, vl.length
+        )));
+    }
+    Ok(())
+}
+
+impl<'a, S: crate::storage::Storage + ?Sized> VlResolver<'a, S> {
+    /// [`VlResolver::new`] over any [`Storage`](crate::storage::Storage).
+    pub fn new_in(file_data: &'a S, offset_size: u8, length_size: u8) -> Self {
         Self {
             file_data,
             offset_size,
@@ -231,51 +296,15 @@ impl<'a> VlResolver<'a> {
 
     /// The bytes of one element: `length × base_size` bytes from the heap,
     /// or `None` for a null element.
-    fn resolve(
-        &mut self,
-        vl: &VlElement,
-        base_size: usize,
-    ) -> Result<Option<&'a [u8]>, FormatError> {
-        let addr = vl.collection_address;
-        if addr == 0 {
+    fn resolve(&mut self, vl: &VlElement, base_size: usize) -> Result<Option<&[u8]>, FormatError> {
+        if vl.collection_address == 0 {
             return Ok(None);
         }
         let data = self.object(vl)?;
-        let expected = (vl.length as usize)
-            .checked_mul(base_size)
-            .ok_or_else(|| FormatError::Overflow("variable-length element size".into()))?;
-        if data.len() != expected {
-            return Err(FormatError::VlDataError(format!(
-                "global heap object {} in the collection at {addr} holds {} bytes; the element \
-                 says {} × {base_size}",
-                vl.object_index,
-                data.len(),
-                vl.length
-            )));
-        }
+        check_object_size(vl, data.len(), base_size)?;
         Ok(Some(data))
     }
 
-    /// One element (the first [`element_size`](Self::element_size) bytes of
-    /// `elem`) of a variable-length sequence whose base type is `base_size`
-    /// bytes: its `length × base_size` bytes, or `None` for a null element
-    /// (heap address 0).
-    pub fn element(
-        &mut self,
-        elem: &[u8],
-        base_size: usize,
-    ) -> Result<Option<&'a [u8]>, FormatError> {
-        let vl = parse_vl_references(elem, 1, self.offset_size)?;
-        self.resolve(&vl[0], base_size)
-    }
-
-    /// One variable-length string element: its bytes up to the first NUL,
-    /// or `None` for a null element (h5dump prints it as `NULL`, h5py
-    /// returns it as empty).
-    pub fn string_element(&mut self, elem: &[u8]) -> Result<Option<&'a [u8]>, FormatError> {
-        Ok(self.element(elem, 1)?.map(cut_at_nul))
-    }
-
     /// The strings of the variable-length string elements in `raw`, as
     /// bytes. A string ends at its first NUL, as libhdf5 returns it (it
     /// converts each to a C string); a null element is empty.
@@ -330,9 +359,20 @@ pub fn read_vl_strings(
     num_elements: u64,
     offset_size: u8,
     length_size: u8,
+) -> Result<Vec<String>, FormatError> {
+    read_vl_strings_in(file_data, raw_data, num_elements, offset_size, length_size)
+}
+
+/// [`read_vl_strings`] over any [`Storage`](crate::storage::Storage).
+pub fn read_vl_strings_in<S: crate::storage::Storage + ?Sized>(
+    file_data: &S,
+    raw_data: &[u8],
+    num_elements: u64,
+    offset_size: u8,
+    length_size: u8,
 ) -> Result<Vec<String>, FormatError> {
     let raw = first_elements(raw_data, num_elements, offset_size)?;
-    VlResolver::new(file_data, offset_size, length_size).strings(raw)
+    VlResolver::new_in(file_data, offset_size, length_size).strings(raw)
 }
 
 /// The first `num_elements` elements of `raw`, or an error if it is shorter.
@@ -364,9 +404,20 @@ pub fn read_vl_bytes(
     num_elements: u64,
     offset_size: u8,
     length_size: u8,
+) -> Result<Vec<Vec<u8>>, FormatError> {
+    read_vl_bytes_in(file_data, raw_data, num_elements, offset_size, length_size)
+}
+
+/// [`read_vl_bytes`] over any [`Storage`](crate::storage::Storage).
+pub fn read_vl_bytes_in<S: crate::storage::Storage + ?Sized>(
+    file_data: &S,
+    raw_data: &[u8],
+    num_elements: u64,
+    offset_size: u8,
+    length_size: u8,
 ) -> Result<Vec<Vec<u8>>, FormatError> {
     let refs = parse_vl_references(raw_data, num_elements, offset_size)?;
-    let mut resolver = VlResolver::new(file_data, offset_size, length_size);
+    let mut resolver = VlResolver::new_in(file_data, offset_size, length_size);
     let mut result = Vec::with_capacity(refs.len());
 
     for vl in &refs {
@@ -385,10 +436,10 @@ pub fn read_vl_bytes(
     Ok(result)
 }
 
-impl<'a> VlResolver<'a> {
-    /// The heap object `vl` points to, whatever its size; its collection is
-    /// parsed on first use.
-    fn object(&mut self, vl: &VlElement) -> Result<&'a [u8], FormatError> {
+impl<'a, S: crate::storage::Storage + ?Sized> VlResolver<'a, S> {
+    /// Where the heap object `vl` points to lies in the file, whatever its
+    /// size (`(offset, size)`); its collection is parsed on first use.
+    fn locate(&mut self, vl: &VlElement) -> Result<(usize, usize), FormatError> {
         let addr = vl.collection_address;
         // libhdf5 writes a null element with address 0, never the undefined
         // address, and fails to read one ("addr undefined") even when its
@@ -402,14 +453,20 @@ impl<'a> VlResolver<'a> {
         if !self.cache.contains_key(&addr) {
             let offset = usize::try_from(addr).map_err(|_| FormatError::UnexpectedEof {
                 expected: usize::MAX,
-                available: self.file_data.len(),
+                available: crate::storage::len_usize(self.file_data),
             })?;
-            let index =
-                GlobalHeapCollection::parse_index(self.file_data, offset, self.length_size)?;
-            // parse_index checked that the collection lies in the file.
+            let (bytes, base, index) =
+                GlobalHeapCollection::read_collection(self.file_data, addr, self.length_size)?;
+            // read_collection checked that the collection lies in the file.
             let end = offset + to_usize(index.collection_size)?;
             self.check_overlap(offset, end)?;
-            let coll = CachedCollection::new(index);
+            // With the whole file in memory the objects are sliced from it;
+            // otherwise the collection's bytes are kept.
+            let bytes = match self.file_data.as_contiguous() {
+                Some(_) => None,
+                None => Some((base, bytes)),
+            };
+            let coll = CachedCollection::new(index, bytes);
             if self.cached_bytes.saturating_add(coll.cost()) > self.budget {
                 self.cache.clear();
                 self.cached_bytes = 0;
@@ -417,13 +474,27 @@ impl<'a> VlResolver<'a> {
             self.cached_bytes += coll.cost();
             self.cache.insert(addr, coll);
         }
-        let (start, size) = self.cache[&addr].get(vl.object_index).ok_or(
-            FormatError::GlobalHeapObjectNotFound {
+        self.cache[&addr]
+            .get(vl.object_index)
+            .ok_or(FormatError::GlobalHeapObjectNotFound {
                 collection_address: addr,
                 index: vl.object_index as u16,
-            },
-        )?;
-        Ok(&self.file_data[start..start + size])
+            })
+    }
+
+    /// The heap object `vl` points to, whatever its size; its collection is
+    /// parsed on first use.
+    fn object(&mut self, vl: &VlElement) -> Result<&[u8], FormatError> {
+        let (start, size) = self.locate(vl)?;
+        if let Some(all) = self.file_data.as_contiguous() {
+            return Ok(&all[start..start + size]);
+        }
+        match &self.cache[&vl.collection_address].bytes {
+            Some((base, bytes)) => Ok(&bytes[start - base..start - base + size]),
+            None => Err(FormatError::Storage(
+                "global heap collection bytes were not kept".into(),
+            )),
+        }
     }
 
     /// Record the collection at `start..end`, refusing one that overlaps a
@@ -709,6 +780,7 @@ mod tests {
         let mut r = VlResolver::new(&file_data, 8, 8);
         let one = CachedCollection {
             objects: vec![(0, 0, 0); 3],
+            bytes: None,
         }
         .cost();
         r.budget = 2 * one + 1;
diff --git a/crates/clawhdf5-format/tests/storage_equivalence.rs b/crates/clawhdf5-format/tests/storage_equivalence.rs
index 02a4f17..71b0dc0 100644
--- a/crates/clawhdf5-format/tests/storage_equivalence.rs
+++ b/crates/clawhdf5-format/tests/storage_equivalence.rs
@@ -37,14 +37,25 @@ use clawhdf5_format::btree_v2::{
     BTreeV2Header, collect_btree_v2_records, collect_btree_v2_records_in, find_btree_v2_records,
     find_btree_v2_records_in,
 };
+use clawhdf5_format::chunk_cache::ChunkCache;
+use clawhdf5_format::chunked_read::{list_chunks, list_chunks_in};
 use clawhdf5_format::data_layout::DataLayout;
+use clawhdf5_format::data_read::{
+    read_raw_data_cached, read_raw_data_cached_in, read_raw_data_full, read_raw_data_full_in,
+    read_raw_data_indexed, read_raw_data_indexed_in, read_raw_data_selection,
+    read_raw_data_selection_in,
+};
 use clawhdf5_format::dataspace::Dataspace;
 use clawhdf5_format::datatype::Datatype;
 use clawhdf5_format::error::FormatError;
 use clawhdf5_format::extensible_array::{
     ExtensibleArrayHeader, read_extensible_array_chunks, read_extensible_array_chunks_in,
 };
-use clawhdf5_format::fill_value::{dataset_fill_value_from_storage, dataset_fill_value_in};
+use clawhdf5_format::fill_value::{
+    dataset_fill_value_from_storage, dataset_fill_value_in, read_full_with_fill,
+    read_full_with_fill_in,
+};
+use clawhdf5_format::filter_pipeline::FilterPipeline;
 use clawhdf5_format::fixed_array::{
     FixedArrayHeader, read_fixed_array_chunks, read_fixed_array_chunks_in,
 };
@@ -54,6 +65,7 @@ use clawhdf5_format::link_info::LinkInfoMessage;
 use clawhdf5_format::local_heap::LocalHeap;
 use clawhdf5_format::message_type::MessageType;
 use clawhdf5_format::object_header::ObjectHeader;
+use clawhdf5_format::selection::Selection;
 use clawhdf5_format::shared_message::{
     self, load_sohm_table, load_sohm_table_in, message_data_with_sohm, message_data_with_sohm_in,
     parse_sohm_btree_entries, parse_sohm_btree_entries_in, parse_sohm_list, parse_sohm_list_in,
@@ -66,11 +78,18 @@ use clawhdf5_format::superblock_ext::{
     read_superblock_extension_in,
 };
 use clawhdf5_format::symbol_table::{SymbolTableMessage, SymbolTableNode};
+use clawhdf5_format::vds::{
+    read_virtual_dataset, read_virtual_dataset_in, virtual_dataset_extent,
+    virtual_dataset_extent_in,
+};
+use clawhdf5_format::vl_data::{VlResolver, read_vl_bytes, read_vl_bytes_in};
 
 /// Objects visited per file, heap objects read per heap: enough to cover
 /// every structure kind while keeping a 35 000-group file fast.
 const MAX_OBJECTS: usize = 1500;
 const MAX_HEAP_IDS: usize = 200;
+/// Datasets larger than this are not read (their chunk indexes still are).
+const MAX_DATA_BYTES: u64 = 16 << 20;
 
 #[derive(Default, Debug)]
 struct Tally {
@@ -89,6 +108,8 @@ struct Walk<'a> {
     slice: &'a [u8],
     storage: &'a CountingStorage,
     name: String,
+    /// The file's directory, for external VDS sources.
+    dir: Option<PathBuf>,
     tally: &'a mut Tally,
 }
 
@@ -264,6 +285,7 @@ impl Walk<'_> {
             }
         }
         self.check_layout(&header, os, ls);
+        self.check_data(&header, os, ls);
     }
 
     /// A symbol-table group: its local heap, B-tree, nodes and names.
@@ -352,6 +374,179 @@ impl Walk<'_> {
         }
     }
 
+    /// A dataset's values through every raw-data path: whole reads (plain,
+    /// cached, indexed, fill-aware, virtual), chunk listings, selections
+    /// (a box, a strided hyperslab, points), VL strings and sequences.
+    fn check_data(&mut self, header: &ObjectHeader, os: u8, ls: u8) {
+        let slice = self.slice;
+        let find = |t: MessageType| {
+            header
+                .messages
+                .iter()
+                .find(|m| m.msg_type == t)
+                .and_then(|m| shared_message::message_data_with_sohm(slice, m, os, ls).ok())
+        };
+        let (Some(dt), Some(ds), Some(dl)) = (
+            find(MessageType::Datatype),
+            find(MessageType::Dataspace),
+            find(MessageType::DataLayout),
+        ) else {
+            return;
+        };
+        let (Ok((dt, _)), Ok(ds), Ok(dl)) = (
+            Datatype::parse(&dt),
+            Dataspace::parse(&ds, ls),
+            DataLayout::parse(&dl, os, ls),
+        ) else {
+            return;
+        };
+        let pipeline = match find(MessageType::FilterPipeline).map(|p| FilterPipeline::parse(&p)) {
+            Some(Ok(p)) => Some(p),
+            Some(Err(_)) => return,
+            None => None,
+        };
+        let pl = pipeline.as_ref();
+        let elem = dt.type_size() as u64;
+        let bytes = ds
+            .dimensions
+            .iter()
+            .try_fold(elem, |a, &d| a.checked_mul(d));
+        if bytes.is_none_or(|b| b > MAX_DATA_BYTES) {
+            return;
+        }
+
+        if matches!(dl, DataLayout::Virtual { .. }) {
+            let resolver = self.resolver();
+            let r: &clawhdf5_format::vds::VdsFileResolver = &resolver;
+            let want = virtual_dataset_extent(slice, &dl, &ds, os, ls, Some(r));
+            let got = virtual_dataset_extent_in(self.st(), &dl, &ds, os, ls, Some(r));
+            self.same("VDS extent", &want, &got);
+            let want = read_virtual_dataset(slice, &dl, &ds, &dt, None, os, ls, Some(r));
+            let got = read_virtual_dataset_in(self.st(), &dl, &ds, &dt, None, os, ls, Some(r));
+            self.same("VDS read", &want, &got);
+            return;
+        }
+
+        let want = read_raw_data_full(slice, &dl, &ds, &dt, pl, os, ls);
+        let got = read_raw_data_full_in(self.st(), &dl, &ds, &dt, pl, os, ls);
+        self.same("raw data", &want, &got);
+        let want_fill = read_full_with_fill(
+            &header.messages,
+            slice,
+            &dl,
+            &ds,
+            elem as usize,
+            os,
+            ls,
+            || read_raw_data_full(slice, &dl, &ds, &dt, pl, os, ls),
+        );
+        let got_fill = read_full_with_fill_in(
+            &header.messages,
+            self.st(),
+            &dl,
+            &ds,
+            elem as usize,
+            os,
+            ls,
+            || read_raw_data_full_in(self.st(), &dl, &ds, &dt, pl, os, ls),
+        );
+        self.same("raw data with fill", &want_fill, &got_fill);
+
+        if matches!(dl, DataLayout::Chunked { .. }) {
+            let want = list_chunks(slice, &dl, &ds, elem as usize, os, ls);
+            let got = list_chunks_in(self.st(), &dl, &ds, elem as usize, os, ls);
+            self.same("chunk list", &want, &got);
+            // Through a chunk cache, twice (the second read is served from
+            // it), and through the indexed path.
+            let (c1, c2) = (ChunkCache::new(), ChunkCache::new());
+            for _ in 0..2 {
+                let want = read_raw_data_cached(slice, &dl, &ds, &dt, pl, os, ls, &c1);
+                let got = read_raw_data_cached_in(self.st(), &dl, &ds, &dt, pl, os, ls, &c2);
+                // A cache lists the chunks in hash-map order (see below).
+                if want.is_err() && got.is_err() {
+                    self.tally.checks += 1;
+                } else {
+                    self.same("raw data (cached)", &want, &got);
+                }
+            }
+            let (c1, c2) = (ChunkCache::new(), ChunkCache::new());
+            let want = read_raw_data_indexed(slice, &dl, &ds, &dt, pl, os, ls, &c1);
+            let got = read_raw_data_indexed_in(self.st(), &dl, &ds, &dt, pl, os, ls, &c2);
+            // The indexed path decodes chunks in hash-map order, so which
+            // failing chunk it reports varies between two caches (with the
+            // slice alone, too): only whether it fails must agree.
+            if want.is_err() && got.is_err() {
+                self.tally.checks += 1;
+            } else {
+                self.same("raw data (indexed)", &want, &got);
+            }
+        }
+
+        let dims = &ds.dimensions;
+        if !dims.is_empty() && dims.iter().all(|&d| d > 0) {
+            let rank = dims.len();
+            let ones = vec![1u64; rank];
+            let quarter = Selection::Hyperslab {
+                start: dims.iter().map(|&d| d / 4).collect(),
+                stride: ones.clone(),
+                count: dims.iter().map(|&d| (d / 3).max(1)).collect(),
+                block: ones.clone(),
+            };
+            let mut stride = ones.clone();
+            stride[rank - 1] = 2;
+            let mut count = dims.clone();
+            count[rank - 1] = dims[rank - 1].div_ceil(2);
+            let strided = Selection::Hyperslab {
+                start: vec![0; rank],
+                stride,
+                count,
+                block: ones.clone(),
+            };
+            let points = Selection::Points(vec![
+                dims.iter().map(|&d| d - 1).collect(),
+                vec![0; rank],
+                dims.iter().map(|&d| d / 2).collect(),
+            ]);
+            for (what, sel) in [
+                ("selection (box)", &quarter),
+                ("selection (strided)", &strided),
+                ("selection (points)", &points),
+            ] {
+                let want = read_raw_data_selection(slice, &dl, &ds, &dt, pl, os, ls, sel);
+                let got = read_raw_data_selection_in(self.st(), &dl, &ds, &dt, pl, os, ls, sel);
+                self.same(what, &want, &got);
+            }
+        }
+
+        // Variable-length strings and sequences, resolved in the global heap.
+        if let (Datatype::VariableLength { base_type, .. }, Ok(raw)) = (&dt, &want) {
+            let n = raw.len() / clawhdf5_format::vl_data::element_size(os).max(1);
+            let raw = &raw[..n * clawhdf5_format::vl_data::element_size(os)];
+            let want = VlResolver::new(slice, os, ls).string_bytes(raw);
+            let got = VlResolver::new_in(self.st(), os, ls).string_bytes(raw);
+            self.same("VL strings", &want, &got);
+            let base = base_type.type_size() as usize;
+            let want = VlResolver::new(slice, os, ls).sequences(raw, base);
+            let got = VlResolver::new_in(self.st(), os, ls).sequences(raw, base);
+            self.same("VL sequences", &want, &got);
+            let want = read_vl_bytes(slice, raw, n as u64, os, ls);
+            let got = read_vl_bytes_in(self.st(), raw, n as u64, os, ls);
+            self.same("VL bytes", &want, &got);
+        }
+    }
+
+    /// External VDS source files: siblings of the file being walked.
+    fn resolver(&self) -> impl Fn(&str) -> Result<Option<Vec<u8>>, FormatError> + use<> {
+        let dir = self.dir.clone();
+        move |name: &str| {
+            let (Some(dir), false) = (dir.as_ref(), name.contains("..") || name.starts_with('/'))
+            else {
+                return Ok(None);
+            };
+            Ok(std::fs::read(dir.join(name)).ok())
+        }
+    }
+
     /// A dataset's layout: VDS mappings, and fixed/extensible array chunk
     /// indexes.
     fn check_layout(&mut self, header: &ObjectHeader, os: u8, ls: u8) {
@@ -461,10 +656,19 @@ fn check_file(path: &Path, tally: &mut Tally) {
     let Ok(bytes) = std::fs::read(path) else {
         return;
     };
-    check_bytes(&path.display().to_string(), &bytes, tally);
+    check_bytes_in(
+        &path.display().to_string(),
+        &bytes,
+        path.parent().map(Path::to_path_buf),
+        tally,
+    );
 }
 
 fn check_bytes(name: &str, bytes: &[u8], tally: &mut Tally) {
+    check_bytes_in(name, bytes, None, tally);
+}
+
+fn check_bytes_in(name: &str, bytes: &[u8], dir: Option<PathBuf>, tally: &mut Tally) {
     let Ok((_, hdf5)) = split_user_block(bytes) else {
         return;
     };
@@ -474,6 +678,7 @@ fn check_bytes(name: &str, bytes: &[u8], tally: &mut Tally) {
         slice: hdf5,
         storage: &storage,
         name: name.to_string(),
+        dir,
         tally,
     };
     walk.run();
@@ -548,6 +753,311 @@ fn corpus_parses_identically_through_storage() {
     assert!(tally.files > 0);
 }
 
+/// A storage that misbehaves: fails its `fail_at`-th read (1-based, `0`
+/// never), and, with `short`, serves one byte less than asked for inside
+/// the file (a truncated response).
+struct Adversary {
+    data: Vec<u8>,
+    reads: std::sync::atomic::AtomicUsize,
+    fail_at: usize,
+    short: bool,
+}
+
+impl Storage for Adversary {
+    fn read_at(&self, offset: u64, len: usize) -> Result<std::borrow::Cow<'_, [u8]>, FormatError> {
+        let n = self
+            .reads
+            .fetch_add(1, std::sync::atomic::Ordering::Relaxed)
+            + 1;
+        if n == self.fail_at {
+            return Err(FormatError::Storage(format!(
+                "injected failure of read {n}"
+            )));
+        }
+        let got = self.data.as_slice().read_at(offset, len)?;
+        let mut v = got.into_owned();
+        if self.short && v.len() > 1 {
+            v.pop();
+        }
+        Ok(std::borrow::Cow::Owned(v))
+    }
+
+    fn len(&self) -> u64 {
+        self.data.len() as u64
+    }
+}
+
+/// Every group listing and every dataset's values (whole, fill-aware and
+/// through a selection) read through a storage that fails or serves short
+/// reads: each result is an error or exactly the in-memory result, never
+/// other data; and a failing read is reported as that failure.
+#[test]
+fn misbehaving_storage_never_returns_wrong_data() {
+    let dir = Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures");
+    let mut files = Vec::new();
+    hdf5_files(&dir, &mut files);
+    files.sort();
+    let (mut compared, mut failures_seen) = (0usize, 0usize);
+    for path in &files {
+        let Ok(bytes) = std::fs::read(path) else {
+            continue;
+        };
+        let Ok((_, hdf5)) = split_user_block(&bytes) else {
+            continue;
+        };
+        let Ok(sb) = Superblock::parse(hdf5, 0) else {
+            continue;
+        };
+        // The whole read of every object, as one result to compare.
+        let everything = |file: &dyn Storage| -> Result<String, FormatError> {
+            let (os, ls) = (sb.offset_size, sb.length_size);
+            let mut out = String::new();
+            let mut queue = VecDeque::from([sb.root_group_address]);
+            let mut seen = HashSet::new();
+            while let Some(addr) = queue.pop_front() {
+                if seen.len() > 200 || !seen.insert(addr) {
+                    continue;
+                }
+                let header = ObjectHeader::parse_in(file, addr, os, ls)?;
+                let find = |t: MessageType| {
+                    header
+                        .messages
+                        .iter()
+                        .find(|m| m.msg_type == t)
+                        .map(|m| message_data_with_sohm_in(file, m, os, ls))
+                        .transpose()
+                };
+                if let (Some(dt), Some(ds), Some(dl)) = (
+                    find(MessageType::Datatype)?,
+                    find(MessageType::Dataspace)?,
+                    find(MessageType::DataLayout)?,
+                ) {
+                    let dt = Datatype::parse(&dt)?.0;
+                    let ds = Dataspace::parse(&ds, ls)?;
+                    let dl = DataLayout::parse(&dl, os, ls)?;
+                    let pl = find(MessageType::FilterPipeline)?
+                        .map(|p| FilterPipeline::parse(&p))
+                        .transpose()?;
+                    let data = read_full_with_fill_in(
+                        &header.messages,
+                        file,
+                        &dl,
+                        &ds,
+                        dt.type_size() as usize,
+                        os,
+                        ls,
+                        || read_raw_data_full_in(file, &dl, &ds, &dt, pl.as_ref(), os, ls),
+                    );
+                    out.push_str(&format!("{addr}: {data:?}\n"));
+                    if let Some(&d0) = ds.dimensions.first() {
+                        let rank = ds.dimensions.len();
+                        let sel = Selection::Hyperslab {
+                            start: vec![0; rank],
+                            stride: vec![1; rank],
+                            count: std::iter::once(d0.div_ceil(2))
+                                .chain(ds.dimensions[1..].iter().copied())
+                                .collect(),
+                            block: vec![1; rank],
+                        };
+                        let part = read_raw_data_selection_in(
+                            file,
+                            &dl,
+                            &ds,
+                            &dt,
+                            pl.as_ref(),
+                            os,
+                            ls,
+                            &sel,
+                        );
+                        out.push_str(&format!("{addr} half: {part:?}\n"));
+                    }
+                }
+                let children = group_v2::resolve_group_children_in(file, &sb, addr);
+                out.push_str(&format!("{addr} children: {children:?}\n"));
+                if let Ok(c) = children {
+                    queue.extend(c.iter().map(|c| c.object_header_address));
+                }
+            }
+            Ok(out)
+        };
+        let want = everything(&hdf5);
+        let counting = CountingStorage::new(hdf5.to_vec());
+        assert_eq!(
+            format!("{:?}", everything(&counting)),
+            format!("{want:?}"),
+            "{}",
+            path.display()
+        );
+        let total = counting.reads() as usize;
+        let step = (total / 25).max(1);
+        for fail_at in (1..=total).step_by(step) {
+            for short in [false, true] {
+                if short && fail_at != 1 {
+                    continue;
+                }
+                let adv = Adversary {
+                    data: hdf5.to_vec(),
+                    reads: Default::default(),
+                    fail_at: if short { 0 } else { fail_at },
+                    short,
+                };
+                let got = everything(&adv);
+                compared += 1;
+                match (&got, &want) {
+                    (Ok(g), Ok(w)) => {
+                        // Per-object results inside may be errors; values
+                        // that were read must be the right ones.
+                        for (gl, wl) in g.lines().zip(w.lines()) {
+                            if gl != wl {
+                                assert!(
+                                    gl.contains("Err("),
+                                    "{}: fail_at {fail_at} short {short}:\n got  {gl}\n want {wl}",
+                                    path.display()
+                                );
+                                failures_seen += 1;
+                                // A listing that failed ends the walk
+                                // differently from here on.
+                                if gl.contains("children: Err(") {
+                                    break;
+                                }
+                            }
+                        }
+                    }
+                    (Err(FormatError::Storage(_)), _) => failures_seen += 1,
+                    (Err(e), Ok(_)) => panic!(
+                        "{}: fail_at {fail_at} short {short}: {e:?} instead of a storage error",
+                        path.display()
+                    ),
+                    (Err(_), Err(_)) => {}
+                    // A listing failed, so the walk never reached the
+                    // object that fails in memory.
+                    (Ok(g), Err(e)) => assert!(
+                        g.contains("children: Err(Storage"),
+                        "{}: fail_at {fail_at} short {short}: read where memory fails ({e:?})",
+                        path.display()
+                    ),
+                }
+            }
+        }
+    }
+    eprintln!("misbehaving storage: {compared} runs, {failures_seen} failures reported");
+    assert!(
+        compared > 500 && failures_seen > 100,
+        "{compared} {failures_seen}"
+    );
+}
+
+/// A read_at-only storage that also counts `read_ranges` calls and ranges.
+struct BatchCounting {
+    inner: CountingStorage,
+    batches: std::sync::atomic::AtomicUsize,
+    ranges: std::sync::atomic::AtomicUsize,
+}
+
+impl Storage for BatchCounting {
+    fn read_at(&self, offset: u64, len: usize) -> Result<std::borrow::Cow<'_, [u8]>, FormatError> {
+        self.inner.read_at(offset, len)
+    }
+
+    fn len(&self) -> u64 {
+        self.inner.len()
+    }
+
+    fn read_ranges(
+        &self,
+        ranges: &[std::ops::Range<u64>],
+    ) -> Result<Vec<std::borrow::Cow<'_, [u8]>>, FormatError> {
+        use std::sync::atomic::Ordering::Relaxed;
+        self.batches.fetch_add(1, Relaxed);
+        self.ranges.fetch_add(ranges.len(), Relaxed);
+        ranges
+            .iter()
+            .map(|r| self.inner.read_at(r.start, (r.end - r.start) as usize))
+            .collect()
+    }
+}
+
+/// A chunked read lists its chunks, then fetches all their bytes with one
+/// `read_ranges` call (a remote backend coalesces and parallelises it), and
+/// a selection fetches only the chunks it overlaps, in one call too.
+#[test]
+fn chunked_reads_fetch_their_chunks_in_one_batch() {
+    use std::sync::atomic::Ordering::Relaxed;
+    let dir = Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures");
+    let mut datasets = 0;
+    for name in [
+        "chunked_large.h5",
+        "chunked_deflate.h5",
+        "chunked_2d.h5",
+        "v4_fixed_array.h5",
+    ] {
+        let bytes = std::fs::read(dir.join(name)).unwrap();
+        let sb = Superblock::parse(&bytes, 0).unwrap();
+        let (os, ls) = (sb.offset_size, sb.length_size);
+        let st = BatchCounting {
+            inner: CountingStorage::new(bytes.clone()),
+            batches: Default::default(),
+            ranges: Default::default(),
+        };
+        for child in group_v2::resolve_group_children(&bytes, &sb, sb.root_group_address).unwrap() {
+            let header =
+                ObjectHeader::parse(&bytes, child.object_header_address as usize, os, ls).unwrap();
+            let msg = |t: MessageType| {
+                header
+                    .messages
+                    .iter()
+                    .find(|m| m.msg_type == t)
+                    .map(|m| m.data.clone())
+            };
+            let Some(dl) = msg(MessageType::DataLayout) else {
+                continue;
+            };
+            let dl = DataLayout::parse(&dl, os, ls).unwrap();
+            if !matches!(dl, DataLayout::Chunked { .. }) {
+                continue;
+            }
+            let dt = Datatype::parse(&msg(MessageType::Datatype).unwrap())
+                .unwrap()
+                .0;
+            let ds = Dataspace::parse(&msg(MessageType::Dataspace).unwrap(), ls).unwrap();
+            let pl = msg(MessageType::FilterPipeline).map(|p| FilterPipeline::parse(&p).unwrap());
+            let es = dt.type_size() as usize;
+            let (chunks, _) = list_chunks(&bytes, &dl, &ds, es, os, ls).unwrap();
+            let want = read_raw_data_full(&bytes, &dl, &ds, &dt, pl.as_ref(), os, ls).unwrap();
+            st.batches.store(0, Relaxed);
+            st.ranges.store(0, Relaxed);
+            let got = read_raw_data_full_in(&st, &dl, &ds, &dt, pl.as_ref(), os, ls).unwrap();
+            assert_eq!(got, want, "{name} {}", child.name);
+            assert_eq!(st.batches.load(Relaxed), 1, "{name} {}", child.name);
+            assert_eq!(
+                st.ranges.load(Relaxed),
+                chunks.len(),
+                "{name} {}",
+                child.name
+            );
+            // The first chunk only.
+            let rank = ds.dimensions.len();
+            let sel = Selection::Hyperslab {
+                start: vec![0; rank],
+                stride: vec![1; rank],
+                count: vec![1; rank],
+                block: vec![1; rank],
+            };
+            let want = read_raw_data_selection(&bytes, &dl, &ds, &dt, pl.as_ref(), os, ls, &sel);
+            st.batches.store(0, Relaxed);
+            st.ranges.store(0, Relaxed);
+            let got = read_raw_data_selection_in(&st, &dl, &ds, &dt, pl.as_ref(), os, ls, &sel);
+            assert_eq!(got, want, "{name} {}", child.name);
+            if chunks.len() > 2 {
+                assert_eq!(st.batches.load(Relaxed), 1, "{name} {}", child.name);
+                assert_eq!(st.ranges.load(Relaxed), 1, "{name} {}", child.name);
+            }
+            datasets += 1;
+        }
+    }
+    assert!(datasets >= 4, "{datasets}");
+}
+
 fn python() -> String {
     std::env::var("CLAWHDF5_PYTHON").unwrap_or_else(|_| "python3".to_string())
 }