hdf5-pure-rust 0.3.7

use crate::error::{Error, Result};
use crate::format::messages::data_layout::{ChunkIndexType, LayoutClass};

use super::{u64_from_usize, usize_from_u64, Dataset, DatasetInfo};

impl Dataset {
    /// Read a subset of the dataset using a selection.
    ///
    /// Example: `ds.read_slice::<f64>(10..20)` reads elements 10-19 from a 1D dataset.
    pub fn read_slice<T: crate::hl::types::H5Type, S: crate::hl::selection::IntoSelection>(
        &self,
        sel: S,
    ) -> Result<Vec<T>> {
        let mut shape = Vec::new();
        self.shape_into(&mut shape)?;
        let selection = sel.into_selection(&shape);
        self.read_slice_alloc::<T>(&shape, &selection)
    }

    fn read_slice_alloc<T: crate::hl::types::H5Type>(
        &self,
        shape: &[u64],
        selection: &crate::hl::selection::Selection,
    ) -> Result<Vec<T>> {
        if matches!(selection, crate::hl::selection::Selection::None) {
            return Ok(Vec::new());
        }
        if let crate::hl::selection::Selection::Points(points) = &selection {
            Self::validate_selection_points(shape, points)?;
            let all_data = self.read_dataset_values::<T>()?;
            return Self::extract_point_selection(&all_data, shape, points);
        }
        // Partial-I/O fast path: a contiguous sub-rectangle selection over a
        // contiguous dataset is read as a single seek+read of just the selected
        // bytes, instead of loading the whole dataset and slicing. Falls back to
        // the whole-read paths below for any selection/layout not yet covered.
        if let Some(raw) = self.try_partial_contiguous_rect(shape, selection)? {
            let info = self.info()?;
            return crate::hl::conversion::ReadConversion::for_dataset::<T>(&info.datatype)?
                .bytes_to_vec(raw);
        }
        // Chunked datasets (BTreeV1 index): decode only the chunks overlapping
        // the selection rectangle, scattering each overlap into the output.
        if let Some(raw) = self.try_partial_chunked_rect(shape, selection)? {
            let info = self.info()?;
            return crate::hl::conversion::ReadConversion::for_dataset::<T>(&info.datatype)?
                .bytes_to_vec(raw);
        }
        if let crate::hl::selection::Selection::Hyperslab(dims) = &selection {
            Self::validate_hyperslab_selection(shape, dims)?;
            let mut out_shape = Vec::new();
            selection.output_shape_into(shape, &mut out_shape);
            let total_out = Self::selection_output_elements(&out_shape)?;
            if total_out == 0 {
                return Ok(Vec::new());
            }
            let total_out_usize = usize_from_u64(total_out, "hyperslab selection element count")?;
            let all_data = self.read_dataset_values::<T>()?;
            return Self::extract_hyperslab_selection(
                &all_data,
                shape,
                dims,
                &out_shape,
                total_out,
                total_out_usize,
            );
        }

        let slices = selection.to_slices(shape);
        Self::validate_selection_slices(shape, &slices)?;
        let mut out_shape = Vec::new();
        selection.output_shape_into(shape, &mut out_shape);
        let total_out = Self::selection_output_elements(&out_shape)?;
        let total_out_usize = usize_from_u64(total_out, "selection element count")?;
        if total_out == 0 {
            return Ok(Vec::new());
        }

        if let Some(values) = self.try_read_slice_direct_1d::<T>(&shape, &slices)? {
            return Ok(values);
        }

        let all_data = self.read_dataset_values::<T>()?;
        if shape.is_empty() {
            return Ok(all_data);
        }
        if Self::selection_slices_cover_shape(shape, &slices) {
            return Ok(all_data);
        }

        if shape.len() == 1 && slices.len() == 1 {
            return Self::extract_1d_selection(&all_data, &slices[0]);
        }

        Self::extract_nd_selection(
            &all_data,
            shape,
            &slices,
            &out_shape,
            total_out,
            total_out_usize,
        )
    }

    /// Read a subset of the dataset into caller-provided storage.
    ///
    /// The output buffer length must exactly match the number of selected
    /// elements.
    pub fn read_slice_into<T: crate::hl::types::H5Type, S: crate::hl::selection::IntoSelection>(
        &self,
        sel: S,
        out: &mut [T],
    ) -> Result<()> {
        let mut shape = Vec::new();
        self.shape_into(&mut shape)?;
        let selection = sel.into_selection(&shape);
        self.read_selection_into(&shape, &selection, out)
    }

    /// Read an already-built selection into caller-provided storage.
    ///
    /// This is the allocation-aware form for callers that already have the
    /// dataset shape and selection object, such as repeated reads over reused
    /// selection metadata.
    pub fn read_selection_into<T: crate::hl::types::H5Type>(
        &self,
        shape: &[u64],
        selection: &crate::hl::selection::Selection,
        out: &mut [T],
    ) -> Result<()> {
        let expected_len = Self::selection_output_len(shape, selection)?;
        if out.len() != expected_len {
            return Err(Error::InvalidFormat(format!(
                "slice output buffer has {} elements, expected {expected_len}",
                out.len()
            )));
        }
        if expected_len == 0 {
            return Ok(());
        }

        if !matches!(
            selection,
            crate::hl::selection::Selection::Points(_)
                | crate::hl::selection::Selection::Hyperslab(_)
        ) {
            let slices = selection.to_slices(shape);
            let elem_size = T::type_size();
            if elem_size == 0 {
                return Err(Error::Other("zero-size type".into()));
            }
            if self.try_read_slice_direct_1d_into(shape, &slices, out)? {
                return Ok(());
            }
        }

        if let Some(raw) = self.try_partial_contiguous_rect(shape, selection)? {
            let info = self.info()?;
            crate::hl::conversion::ReadConversion::for_dataset::<T>(&info.datatype)?
                .bytes_into_slice(&raw, out)?;
            return Ok(());
        }

        if let Some(raw) = self.try_partial_chunked_rect(shape, selection)? {
            let info = self.info()?;
            crate::hl::conversion::ReadConversion::for_dataset::<T>(&info.datatype)?
                .bytes_into_slice(&raw, out)?;
            return Ok(());
        }

        self.read_slice_into_impl(shape, selection, out)
    }

    /// Read a single dataset cell at row-major coordinates.
    pub fn read_cell<T: crate::hl::types::H5Type>(&self, coords: &[u64]) -> Result<T> {
        let selection = crate::hl::selection::Selection::Points(vec![coords.to_vec()]);
        let mut values = self.read_slice::<T, _>(selection)?;
        values
            .pop()
            .ok_or_else(|| Error::InvalidFormat("cell selection produced no value".into()))
    }

    /// Read a single dataset cell into caller-provided scalar storage.
    pub fn read_cell_into<T: crate::hl::types::H5Type>(
        &self,
        coords: &[u64],
        out: &mut T,
    ) -> Result<()> {
        let selection = crate::hl::selection::Selection::Points(vec![coords.to_vec()]);
        let shape = self.shape()?;
        let mut tmp = [*out];
        self.read_selection_into(&shape, &selection, &mut tmp)?;
        *out = tmp[0];
        Ok(())
    }

    fn read_slice_into_impl<T: crate::hl::types::H5Type>(
        &self,
        shape: &[u64],
        selection: &crate::hl::selection::Selection,
        out: &mut [T],
    ) -> Result<()> {
        if matches!(selection, crate::hl::selection::Selection::None) {
            return Ok(());
        }

        let all_data = self.read_dataset_values::<T>()?;
        if let crate::hl::selection::Selection::Points(points) = selection {
            Self::extract_point_selection_into(&all_data, shape, points, out)?;
            return Ok(());
        }
        if let crate::hl::selection::Selection::Hyperslab(dims) = selection {
            let mut out_shape = Vec::new();
            selection.output_shape_into(shape, &mut out_shape);
            let total_out = Self::selection_output_elements(&out_shape)?;
            Self::extract_hyperslab_selection_into(
                &all_data, shape, dims, &out_shape, total_out, out,
            )?;
            return Ok(());
        }

        let slices = selection.to_slices(shape);
        if shape.is_empty() {
            out.copy_from_slice(&all_data);
            return Ok(());
        }
        if shape.len() == 1 && slices.len() == 1 {
            Self::extract_1d_selection_into(&all_data, &slices[0], out)?;
            return Ok(());
        }

        let mut out_shape = Vec::new();
        selection.output_shape_into(shape, &mut out_shape);
        let total_out = Self::selection_output_elements(&out_shape)?;
        Self::extract_nd_selection_into(&all_data, shape, &slices, &out_shape, total_out, out)
    }

    fn read_dataset_values<T: crate::hl::types::H5Type>(&self) -> Result<Vec<T>> {
        let info = self.info()?;
        let conversion = crate::hl::conversion::ReadConversion::for_dataset::<T>(&info.datatype)?;
        let element_size = usize_from_u64(u64::from(info.datatype.size), "datatype size")?;
        if element_size == 0 {
            return Err(Error::InvalidFormat("zero-sized datatype".into()));
        }
        let access = crate::hl::dataset::DatasetAccess::new();
        let total_elements = if info.layout.layout_class == LayoutClass::Virtual {
            let shape = self.virtual_shape_with_info(&info, &access)?;
            Self::dataspace_element_count(info.dataspace.space_type, &shape)?
        } else {
            Self::dataspace_element_count(info.dataspace.space_type, &info.dataspace.dims)?
        };
        let total_elements_usize = usize_from_u64(total_elements, "dimension product")?;
        let total_bytes = total_elements_usize
            .checked_mul(element_size)
            .ok_or_else(|| Error::InvalidFormat("total data size overflow".into()))?;
        if conversion.is_same_size_bytes() {
            let mut values = Vec::<T>::with_capacity(total_elements_usize);
            let raw_out = unsafe {
                std::slice::from_raw_parts_mut(values.as_mut_ptr() as *mut u8, total_bytes)
            };
            self.read_raw_into_with_info(&info, &access, raw_out)?;
            conversion.convert_bytes_in_place(raw_out);
            T::validate_byte_slice(raw_out)?;
            unsafe {
                values.set_len(total_elements_usize);
            }
            return Ok(values);
        }
        let mut raw = vec![0u8; total_bytes];
        self.read_raw_into_with_info(&info, &access, &mut raw)?;
        conversion.bytes_to_vec(raw)
    }

    /// Convert a selection into a per-dimension contiguous `(start, len)`
    /// rectangle, or `None` if the selection is not a contiguous sub-rectangle
    /// (point selections, strided/non-adjacent-block hyperslabs, step != 1
    /// slices, scalars) — in which case the caller falls back to the general
    /// whole-read-then-slice path.
    fn selection_to_rect(
        shape: &[u64],
        selection: &crate::hl::selection::Selection,
    ) -> Option<Vec<(u64, u64)>> {
        use crate::hl::selection::Selection;
        if shape.is_empty() {
            return None; // scalar — let the general path handle it
        }
        match selection {
            Selection::All => Some(shape.iter().map(|&d| (0, d)).collect()),
            Selection::Hyperslab(dims) => {
                if dims.len() != shape.len() {
                    return None;
                }
                let mut rect = Vec::with_capacity(dims.len());
                for (d, hd) in dims.iter().enumerate() {
                    // A run is contiguous iff it is a single block, or its blocks
                    // are adjacent (stride == block).
                    if !(hd.count <= 1 || hd.stride == hd.block) {
                        return None;
                    }
                    let len = hd.count.checked_mul(hd.block)?;
                    if hd.start.checked_add(len)? > shape[d] {
                        return None;
                    }
                    rect.push((hd.start, len));
                }
                Some(rect)
            }
            Selection::Slice(slices) => {
                if slices.len() != shape.len() {
                    return None;
                }
                let mut rect = Vec::with_capacity(slices.len());
                for (d, s) in slices.iter().enumerate() {
                    if s.step != 1 || s.end < s.start || s.end > shape[d] {
                        return None;
                    }
                    rect.push((s.start, s.end - s.start));
                }
                Some(rect)
            }
            Selection::None | Selection::Points(_) => None,
        }
    }

    /// Fast path for a contiguous sub-rectangle of a *contiguous-layout* dataset:
    /// if the selection is a contiguous slab (the first partially-selected
    /// dimension is followed only by fully-selected dimensions), the selected
    /// elements form one contiguous byte range, so read just that range with a
    /// single seek. Returns `Ok(None)` (fall back to whole-read) for any
    /// selection/layout not handled here. Bytes are returned in on-disk element
    /// layout; the caller applies datatype conversion.
    fn try_partial_contiguous_rect(
        &self,
        shape: &[u64],
        selection: &crate::hl::selection::Selection,
    ) -> Result<Option<Vec<u8>>> {
        let Some(rect) = Self::selection_to_rect(shape, selection) else {
            return Ok(None);
        };
        let info = self.info()?;
        if info.layout.layout_class != LayoutClass::Contiguous {
            return Ok(None); // chunked/compact/virtual handled elsewhere
        }
        let element_size = usize_from_u64(u64::from(info.datatype.size), "datatype size")?;
        if element_size == 0 {
            return Ok(None);
        }
        let ndims = shape.len();

        // In row-major layout, a rectangle is one byte run only when every
        // dimension after the first selected dimension with length > 1 is full.
        // Leading dimensions of length 1 simply choose a single outer index.
        if let Some(first_multi) = (0..ndims).find(|&d| rect[d].1 != 1) {
            for d in (first_multi + 1)..ndims {
                if rect[d].1 != shape[d] {
                    return Ok(None); // a "hole" in a trailing dim -> not contiguous
                }
            }
        }

        // Row-major element strides over the dataset shape.
        let mut stride = vec![1u64; ndims];
        for d in (0..ndims.saturating_sub(1)).rev() {
            stride[d] = stride[d + 1]
                .checked_mul(shape[d + 1])
                .ok_or_else(|| Error::InvalidFormat("contiguous stride overflow".into()))?;
        }
        let mut offset_elems = 0u64;
        let mut total_elems = 1u64;
        for d in 0..ndims {
            offset_elems = offset_elems
                .checked_add(
                    rect[d]
                        .0
                        .checked_mul(stride[d])
                        .ok_or_else(|| Error::InvalidFormat("offset overflow".into()))?,
                )
                .ok_or_else(|| Error::InvalidFormat("offset overflow".into()))?;
            total_elems = total_elems
                .checked_mul(rect[d].1)
                .ok_or_else(|| Error::InvalidFormat("selection element count overflow".into()))?;
        }

        let addr = match info.layout.contiguous_addr {
            Some(a) if !crate::io::reader::is_undef_addr(a) => a,
            // Unallocated / external data: fall back (fill values, external lists).
            _ => return Ok(None),
        };
        let offset_bytes = offset_elems
            .checked_mul(element_size as u64)
            .ok_or_else(|| Error::InvalidFormat("byte offset overflow".into()))?;
        let total_elems_usize = usize_from_u64(total_elems, "selection element count")?;
        let len_bytes = total_elems_usize
            .checked_mul(element_size)
            .ok_or_else(|| Error::InvalidFormat("selection byte length overflow".into()))?;

        // Stay within the stored data; otherwise fall back (defensive).
        if let Some(sz) = info.layout.contiguous_size {
            match offset_bytes.checked_add(u64_from_usize(len_bytes, "selection byte length")?) {
                Some(end) if end <= sz => {}
                _ => return Ok(None),
            }
        }

        let mut guard = self.inner.lock();
        let start = addr
            .checked_add(offset_bytes)
            .ok_or_else(|| Error::InvalidFormat("read offset overflow".into()))?;
        let mut buf = vec![0u8; len_bytes];
        guard.reader.seek(start)?;
        guard.reader.read_exact(&mut buf)?;
        Ok(Some(buf))
    }

    /// Partial-I/O fast path for a contiguous sub-rectangle of a *chunked*
    /// (BTreeV1-indexed) dataset: enumerate the chunk records, decode only the
    /// chunks whose extent overlaps the selection, and scatter each overlap into
    /// the selection-shaped output (missing chunks remain at the fill value).
    /// Returns `Ok(None)` to fall back for unsupported index types / selections.
    fn try_partial_chunked_rect(
        &self,
        shape: &[u64],
        selection: &crate::hl::selection::Selection,
    ) -> Result<Option<Vec<u8>>> {
        let Some(rect) = Self::selection_to_rect(shape, selection) else {
            return Ok(None);
        };
        let info = self.info()?;
        if info.layout.layout_class != LayoutClass::Chunked {
            return Ok(None);
        }
        // All chunk index types are handled below; a missing index type is only
        // valid for the v<=3 (B-tree v1) default — otherwise fall back.
        if info.layout.chunk_index_type.is_none() && info.layout.version > 3 {
            return Ok(None);
        }
        let element_size = usize_from_u64(u64::from(info.datatype.size), "datatype size")?;
        let ndims = shape.len();
        if element_size == 0 || ndims == 0 {
            return Ok(None);
        }
        let chunk_dims_raw = match info.layout.chunk_dims.as_ref() {
            Some(c) => c,
            None => return Ok(None),
        };
        let chunk_data_dims = Self::chunk_data_dims(shape, chunk_dims_raw)?;
        if chunk_data_dims.len() != ndims {
            return Ok(None);
        }
        let chunk_bytes = Self::chunk_byte_len(chunk_dims_raw, chunk_data_dims, element_size)?;

        let sel_start: Vec<u64> = rect.iter().map(|r| r.0).collect();
        let sel_len: Vec<u64> = rect.iter().map(|r| r.1).collect();
        let mut sel_total: u64 = 1;
        for &l in &sel_len {
            sel_total = sel_total
                .checked_mul(l)
                .ok_or_else(|| Error::InvalidFormat("selection element count overflow".into()))?;
        }
        let sel_total_usize = usize_from_u64(sel_total, "selection element count")?;
        let out_bytes = sel_total_usize
            .checked_mul(element_size)
            .ok_or_else(|| Error::InvalidFormat("selection byte length overflow".into()))?;
        let mut out = vec![0u8; out_bytes];
        // Pre-fill with the dataset fill value so chunks absent from the index
        // (sparse datasets) read correctly.
        Self::filled_data_into(sel_total_usize, element_size, &info, &mut out)?;

        let idx_addr = match info.layout.chunk_index_addr {
            Some(a) => a,
            None => return Ok(None),
        };
        if crate::io::reader::is_undef_addr(idx_addr) {
            return Ok(Some(out)); // no chunks allocated -> all fill
        }

        let filtered = info
            .filter_pipeline
            .as_ref()
            .map(|p| !p.filters.is_empty())
            .unwrap_or(false);

        let mut guard = self.inner.lock();
        // Unified per-chunk records: (element-offset coords, file addr, on-disk
        // size, filter mask), collected per index type.
        // Helper for the FixedArray / ExtensibleArray collectors, which share the
        // `FixedArrayElement` record and the linear-index -> coords mapping.
        let chunk_size_len = if filtered {
            Self::filtered_chunk_size_len(
                &info,
                chunk_bytes,
                usize::from(guard.reader.sizeof_size()),
            )?
        } else {
            0
        };
        let records: Vec<(Vec<u64>, u64, u64, u32)> = match info.layout.chunk_index_type {
            Some(ChunkIndexType::FixedArray) | Some(ChunkIndexType::ExtensibleArray) => {
                let mut elements = Vec::new();
                if matches!(
                    info.layout.chunk_index_type,
                    Some(ChunkIndexType::FixedArray)
                ) {
                    crate::format::fixed_array::read_fixed_array_chunks_into(
                        &mut guard.reader,
                        idx_addr,
                        filtered,
                        chunk_size_len,
                        &mut elements,
                    )?;
                } else {
                    crate::format::extensible_array::read_extensible_array_chunks_into(
                        &mut guard.reader,
                        idx_addr,
                        filtered,
                        chunk_size_len,
                        &mut elements,
                    )?;
                }
                let chunks_per_dim = Self::chunks_per_dim(shape, chunk_data_dims)?;
                let mut recs = Vec::with_capacity(elements.len());
                let mut coords = Vec::with_capacity(ndims);
                for (i, el) in elements.iter().enumerate() {
                    if crate::io::reader::is_undef_addr(el.addr) {
                        continue; // absent chunk -> stays fill value
                    }
                    Self::implicit_chunk_coords_into(
                        i,
                        chunk_data_dims,
                        &chunks_per_dim,
                        &mut coords,
                    )?;
                    let size = el.nbytes.unwrap_or(chunk_bytes as u64);
                    recs.push((coords.clone(), el.addr, size, el.filter_mask));
                }
                recs
            }
            Some(ChunkIndexType::Implicit) => {
                // Implicit index: contiguous chunks at idx_addr + i*chunk_bytes.
                // The upstream walker rejects filters here, so do the same.
                if filtered {
                    return Ok(None);
                }
                let chunks_per_dim = Self::chunks_per_dim(shape, chunk_data_dims)?;
                let total_chunks: usize = chunks_per_dim
                    .iter()
                    .try_fold(1usize, |a, &c| a.checked_mul(c))
                    .ok_or_else(|| Error::InvalidFormat("chunk count overflow".into()))?;
                let chunk_bytes_u64 = u64_from_usize(chunk_bytes, "implicit chunk byte size")?;
                let mut recs = Vec::with_capacity(total_chunks);
                let mut coords = Vec::with_capacity(ndims);
                for i in 0..total_chunks {
                    Self::implicit_chunk_coords_into(
                        i,
                        chunk_data_dims,
                        &chunks_per_dim,
                        &mut coords,
                    )?;
                    let addr = Self::implicit_chunk_offset(idx_addr, i, chunk_bytes_u64)?;
                    recs.push((coords.clone(), addr, chunk_bytes as u64, 0));
                }
                recs
            }
            Some(ChunkIndexType::SingleChunk) => {
                let size = info
                    .layout
                    .single_chunk_filtered_size
                    .unwrap_or(chunk_bytes as u64);
                let mask = info.layout.single_chunk_filter_mask.unwrap_or(0);
                vec![(vec![0u64; ndims], idx_addr, size, mask)]
            }
            Some(ChunkIndexType::BTreeV2) => {
                let sizeof_addr = usize::from(guard.reader.sizeof_addr());
                let mut raw_records = Vec::new();
                crate::format::btree_v2::collect_all_records_into(
                    &mut guard.reader,
                    idx_addr,
                    &mut raw_records,
                )?;
                let mut recs = Vec::with_capacity(raw_records.len());
                let mut coords = Vec::with_capacity(ndims);
                for record in &raw_records {
                    let (addr, nbytes, filter_mask) = Self::decode_btree_v2_chunk_record_into(
                        record,
                        filtered,
                        chunk_size_len,
                        sizeof_addr,
                        ndims,
                        chunk_bytes,
                        &mut coords,
                    )?;
                    if crate::io::reader::is_undef_addr(addr) {
                        continue;
                    }
                    Self::scale_btree_v2_chunk_coords(&mut coords, chunk_data_dims)?;
                    recs.push((coords.clone(), addr, nbytes, filter_mask));
                }
                recs
            }
            // BTreeV1 explicitly, and the None (v<=3) default which is also v1.
            _ => Self::collect_btree_v1_chunks(&mut guard.reader, idx_addr, ndims)?
                .into_iter()
                .map(|r| (r.coords, r.chunk_addr, r.chunk_size, r.filter_mask))
                .collect(),
        };

        for (coords, addr, on_disk_size, filter_mask) in &records {
            if coords.len() != ndims {
                continue;
            }
            // Skip chunks whose extent does not overlap the selection.
            let mut overlaps = true;
            for d in 0..ndims {
                let clo = coords[d];
                let chi = clo.saturating_add(chunk_data_dims[d]);
                let shi = sel_start[d] + sel_len[d];
                if sel_start[d].max(clo) >= shi.min(chi) {
                    overlaps = false;
                    break;
                }
            }
            if !overlaps {
                continue;
            }
            guard.reader.seek(*addr)?;
            let decoded = if !filtered {
                let mut buf = vec![0u8; chunk_bytes];
                guard.reader.read_exact(&mut buf)?;
                buf
            } else {
                let read_size = usize_from_u64(*on_disk_size, "chunk size")?;
                let mut raw = vec![0u8; read_size];
                guard.reader.read_bytes_into(&mut raw)?;
                let pipeline = info.filter_pipeline.as_ref().unwrap();
                crate::filters::apply_pipeline_reverse_with_mask_expected(
                    &raw,
                    pipeline,
                    element_size,
                    *filter_mask,
                    chunk_bytes,
                )?
            };
            if decoded.len() < chunk_bytes {
                return Err(Error::InvalidFormat(
                    "decoded chunk shorter than expected".into(),
                ));
            }
            Self::copy_rect_overlap(
                &decoded,
                coords,
                chunk_data_dims,
                &mut out,
                &sel_start,
                &sel_len,
                element_size,
            );
        }
        Ok(Some(out))
    }

    /// Copy the overlap of one chunk (extent `[coord, coord+chunk_dims)`, buffer
    /// `src` in row-major chunk layout) into the selection-shaped output `dst`
    /// (rectangle `[sel_start, sel_start+sel_len)`, row-major), element size
    /// `esize` bytes. Copies the innermost contiguous run per outer-index combo.
    fn copy_rect_overlap(
        src: &[u8],
        coord: &[u64],
        chunk_dims: &[u64],
        dst: &mut [u8],
        sel_start: &[u64],
        sel_len: &[u64],
        esize: usize,
    ) {
        let ndims = sel_len.len();
        let mut lo = vec![0u64; ndims];
        let mut hi = vec![0u64; ndims];
        for d in 0..ndims {
            lo[d] = sel_start[d].max(coord[d]);
            hi[d] = (sel_start[d] + sel_len[d]).min(coord[d] + chunk_dims[d]);
            if lo[d] >= hi[d] {
                return; // no overlap
            }
        }
        // Row-major element strides.
        let mut src_stride = vec![1u64; ndims];
        let mut dst_stride = vec![1u64; ndims];
        for d in (0..ndims.saturating_sub(1)).rev() {
            src_stride[d] = src_stride[d + 1] * chunk_dims[d + 1];
            dst_stride[d] = dst_stride[d + 1] * sel_len[d + 1];
        }
        let last = ndims - 1;
        let run = (hi[last] - lo[last]) as usize * esize;
        let mut idx = lo.clone();
        loop {
            let mut src_off = 0u64;
            let mut dst_off = 0u64;
            for d in 0..ndims {
                let c = if d == last { lo[last] } else { idx[d] };
                src_off += (c - coord[d]) * src_stride[d];
                dst_off += (c - sel_start[d]) * dst_stride[d];
            }
            let s = src_off as usize * esize;
            let t = dst_off as usize * esize;
            dst[t..t + run].copy_from_slice(&src[s..s + run]);

            if last == 0 {
                break;
            }
            let mut d = last - 1;
            loop {
                idx[d] += 1;
                if idx[d] < hi[d] {
                    break;
                }
                idx[d] = lo[d];
                if d == 0 {
                    return;
                }
                d -= 1;
            }
        }
    }

    fn selection_output_len(
        shape: &[u64],
        selection: &crate::hl::selection::Selection,
    ) -> Result<usize> {
        match selection {
            crate::hl::selection::Selection::None => Ok(0),
            crate::hl::selection::Selection::Points(points) => {
                Self::validate_selection_points(shape, points)?;
                Ok(points.len())
            }
            crate::hl::selection::Selection::Hyperslab(dims) => {
                Self::validate_hyperslab_selection(shape, dims)?;
                let mut out_shape = Vec::new();
                selection.output_shape_into(shape, &mut out_shape);
                let total_out = Self::selection_output_elements(&out_shape)?;
                usize_from_u64(total_out, "hyperslab selection element count")
            }
            _ => {
                let slices = selection.to_slices(shape);
                Self::validate_selection_slices(shape, &slices)?;
                let mut out_shape = Vec::new();
                selection.output_shape_into(shape, &mut out_shape);
                let total_out = Self::selection_output_elements(&out_shape)?;
                usize_from_u64(total_out, "selection element count")
            }
        }
    }

    fn selection_output_elements(out_shape: &[u64]) -> Result<u64> {
        if out_shape.is_empty() {
            return Ok(1);
        }
        out_shape.iter().try_fold(1u64, |acc, &dim| {
            acc.checked_mul(dim)
                .ok_or_else(|| Error::InvalidFormat("selection element count overflow".into()))
        })
    }

    fn selection_slices_cover_shape(
        shape: &[u64],
        slices: &[crate::hl::selection::SliceInfo],
    ) -> bool {
        shape.len() == slices.len()
            && shape
                .iter()
                .zip(slices)
                .all(|(&extent, slice)| slice.start == 0 && slice.end == extent && slice.step == 1)
    }

    fn validate_selection_slices(
        shape: &[u64],
        slices: &[crate::hl::selection::SliceInfo],
    ) -> Result<()> {
        if slices.len() != shape.len() {
            return Err(Error::InvalidFormat(format!(
                "selection rank {} does not match dataset rank {}",
                slices.len(),
                shape.len()
            )));
        }
        for (dim, (slice, &extent)) in slices.iter().zip(shape).enumerate() {
            if slice.step == 0 {
                return Err(Error::InvalidFormat(format!(
                    "selection dimension {dim} has zero step"
                )));
            }
            if slice.start > extent || slice.end > extent {
                return Err(Error::InvalidFormat(format!(
                    "selection dimension {dim} range {}..{} exceeds extent {extent}",
                    slice.start, slice.end
                )));
            }
        }
        Ok(())
    }

    fn validate_hyperslab_selection(
        shape: &[u64],
        dims: &[crate::hl::selection::HyperslabDim],
    ) -> Result<()> {
        if dims.len() != shape.len() {
            return Err(Error::InvalidFormat(format!(
                "hyperslab rank {} does not match dataset rank {}",
                dims.len(),
                shape.len()
            )));
        }
        for (dim, (selection, &extent)) in dims.iter().zip(shape).enumerate() {
            if selection.stride == 0 {
                return Err(Error::InvalidFormat(format!(
                    "hyperslab dimension {dim} has zero stride"
                )));
            }
            if selection.block == 0 && selection.count != 0 {
                return Err(Error::InvalidFormat(format!(
                    "hyperslab dimension {dim} has zero block"
                )));
            }
            if selection.count == 0 || selection.block == 0 {
                continue;
            }
            let span_start = selection
                .count
                .checked_sub(1)
                .and_then(|count_minus_one| count_minus_one.checked_mul(selection.stride))
                .and_then(|offset| selection.start.checked_add(offset))
                .ok_or_else(|| Error::InvalidFormat("hyperslab extent overflow".into()))?;
            let span_end = span_start
                .checked_add(selection.block)
                .ok_or_else(|| Error::InvalidFormat("hyperslab extent overflow".into()))?;
            if span_end > extent {
                return Err(Error::InvalidFormat(format!(
                    "hyperslab dimension {dim} exceeds extent {extent}"
                )));
            }
        }
        Ok(())
    }

    fn validate_selection_points(shape: &[u64], points: &[Vec<u64>]) -> Result<()> {
        for point in points {
            if point.len() != shape.len() {
                return Err(Error::InvalidFormat(format!(
                    "point selection rank {} does not match dataset rank {}",
                    point.len(),
                    shape.len()
                )));
            }
            for (dim, (&coord, &extent)) in point.iter().zip(shape).enumerate() {
                if coord >= extent {
                    return Err(Error::InvalidFormat(format!(
                        "point selection coordinate {coord} in dimension {dim} exceeds extent {extent}"
                    )));
                }
            }
        }
        Ok(())
    }

    fn extract_point_selection<T: crate::hl::types::H5Type>(
        all_data: &[T],
        shape: &[u64],
        points: &[Vec<u64>],
    ) -> Result<Vec<T>> {
        let strides = Self::row_major_strides(shape)?;
        let mut result = Vec::new();
        result.try_reserve_exact(points.len()).map_err(|err| {
            Error::InvalidFormat(format!("point selection allocation failed: {err}"))
        })?;
        for point in points {
            let index = Self::linear_index(point, &strides)?;
            if index < all_data.len() {
                result.push(all_data[index]);
            }
        }
        Ok(result)
    }

    fn extract_point_selection_into<T: crate::hl::types::H5Type>(
        all_data: &[T],
        shape: &[u64],
        points: &[Vec<u64>],
        out: &mut [T],
    ) -> Result<()> {
        let strides = Self::row_major_strides(shape)?;
        let mut written = 0usize;
        for point in points {
            let index = Self::linear_index(point, &strides)?;
            if index < all_data.len() {
                out[written] = all_data[index];
                written += 1;
            }
        }
        if written == out.len() {
            Ok(())
        } else {
            Err(Error::InvalidFormat(format!(
                "point selection produced {written} elements, expected {}",
                out.len()
            )))
        }
    }

    fn extract_hyperslab_selection<T: crate::hl::types::H5Type>(
        all_data: &[T],
        shape: &[u64],
        dims: &[crate::hl::selection::HyperslabDim],
        out_shape: &[u64],
        total_out: u64,
        total_out_usize: usize,
    ) -> Result<Vec<T>> {
        let strides = Self::row_major_strides(shape)?;
        let ndims = shape.len();
        let mut result = Vec::new();
        result.try_reserve_exact(total_out_usize).map_err(|err| {
            Error::InvalidFormat(format!("hyperslab selection allocation failed: {err}"))
        })?;
        let mut out_idx = vec![0u64; ndims];
        for _ in 0..total_out {
            let mut in_linear = 0usize;
            for dim in 0..ndims {
                let selected_block = out_idx[dim] / dims[dim].block;
                let selected_offset = out_idx[dim] % dims[dim].block;
                let in_d =
                    dims[dim]
                        .start
                        .checked_add(selected_block.checked_mul(dims[dim].stride).ok_or_else(
                            || Error::InvalidFormat("hyperslab coordinate overflow".into()),
                        )?)
                        .and_then(|coord| coord.checked_add(selected_offset))
                        .ok_or_else(|| {
                            Error::InvalidFormat("hyperslab coordinate overflow".into())
                        })?;
                let term = usize_from_u64(in_d, "hyperslab input index")?
                    .checked_mul(strides[dim])
                    .ok_or_else(|| {
                        Error::InvalidFormat("hyperslab linear index overflow".into())
                    })?;
                in_linear = in_linear.checked_add(term).ok_or_else(|| {
                    Error::InvalidFormat("hyperslab linear index overflow".into())
                })?;
            }
            if in_linear < all_data.len() {
                result.push(all_data[in_linear]);
            }
            for dim in (0..ndims).rev() {
                out_idx[dim] = out_idx[dim].checked_add(1).ok_or_else(|| {
                    Error::InvalidFormat("hyperslab output index overflow".into())
                })?;
                if out_idx[dim] < out_shape[dim] {
                    break;
                }
                out_idx[dim] = 0;
            }
        }
        Ok(result)
    }

    fn extract_hyperslab_selection_into<T: crate::hl::types::H5Type>(
        all_data: &[T],
        shape: &[u64],
        dims: &[crate::hl::selection::HyperslabDim],
        out_shape: &[u64],
        total_out: u64,
        out: &mut [T],
    ) -> Result<()> {
        let strides = Self::row_major_strides(shape)?;
        let ndims = shape.len();
        let mut out_idx = vec![0u64; ndims];
        let mut written = 0usize;
        for _ in 0..total_out {
            let mut in_linear = 0usize;
            for dim in 0..ndims {
                let selected_block = out_idx[dim] / dims[dim].block;
                let selected_offset = out_idx[dim] % dims[dim].block;
                let in_d =
                    dims[dim]
                        .start
                        .checked_add(selected_block.checked_mul(dims[dim].stride).ok_or_else(
                            || Error::InvalidFormat("hyperslab coordinate overflow".into()),
                        )?)
                        .and_then(|coord| coord.checked_add(selected_offset))
                        .ok_or_else(|| {
                            Error::InvalidFormat("hyperslab coordinate overflow".into())
                        })?;
                let term = usize_from_u64(in_d, "hyperslab input index")?
                    .checked_mul(strides[dim])
                    .ok_or_else(|| {
                        Error::InvalidFormat("hyperslab linear index overflow".into())
                    })?;
                in_linear = in_linear.checked_add(term).ok_or_else(|| {
                    Error::InvalidFormat("hyperslab linear index overflow".into())
                })?;
            }
            if in_linear < all_data.len() {
                out[written] = all_data[in_linear];
                written += 1;
            }
            for dim in (0..ndims).rev() {
                out_idx[dim] = out_idx[dim].checked_add(1).ok_or_else(|| {
                    Error::InvalidFormat("hyperslab output index overflow".into())
                })?;
                if out_idx[dim] < out_shape[dim] {
                    break;
                }
                out_idx[dim] = 0;
            }
        }
        if written == out.len() {
            Ok(())
        } else {
            Err(Error::InvalidFormat(format!(
                "hyperslab selection produced {written} elements, expected {}",
                out.len()
            )))
        }
    }

    fn try_read_slice_direct_1d<T: crate::hl::types::H5Type>(
        &self,
        shape: &[u64],
        slices: &[crate::hl::selection::SliceInfo],
    ) -> Result<Option<Vec<T>>> {
        if !(shape.len() == 1 && slices.len() == 1 && slices[0].step == 1) {
            return Ok(None);
        }

        let info = self.info()?;
        let conversion = crate::hl::conversion::ReadConversion::for_dataset::<T>(&info.datatype)?;
        if !conversion.is_same_size_bytes() || T::requires_validation() {
            return Ok(None);
        }
        let elem_size = usize_from_u64(u64::from(info.datatype.size), "datatype size")?;
        if elem_size == 0 {
            return Err(Error::Other("zero-size type".into()));
        }
        let nbytes = usize_from_u64(slices[0].count(), "selection count")?
            .checked_mul(elem_size)
            .ok_or_else(|| Error::InvalidFormat("selection byte count overflow".into()))?;
        let count = usize_from_u64(slices[0].count(), "selection count")?;
        let mut values = Vec::<T>::with_capacity(count);
        if nbytes > 0 {
            let raw_out =
                unsafe { std::slice::from_raw_parts_mut(values.as_mut_ptr() as *mut u8, nbytes) };
            match info.layout.layout_class {
                LayoutClass::Contiguous => {
                    let Some(addr) = info.layout.contiguous_addr else {
                        return Ok(None);
                    };
                    if crate::io::reader::is_undef_addr(addr) {
                        return Ok(None);
                    }

                    let start_byte = usize_from_u64(slices[0].start, "selection start")?
                        .checked_mul(elem_size)
                        .ok_or_else(|| {
                            Error::InvalidFormat("selection byte offset overflow".into())
                        })?;
                    let read_addr = addr
                        .checked_add(u64_from_usize(start_byte, "selection byte offset")?)
                        .ok_or_else(|| {
                            Error::InvalidFormat("selection read address overflow".into())
                        })?;

                    let mut guard = self.inner.lock();
                    guard.reader.seek(read_addr)?;
                    guard.reader.read_exact(raw_out)?;
                    conversion.convert_bytes_in_place(raw_out);
                }
                LayoutClass::Chunked => {
                    if !self
                        .try_read_chunked_slice_1d_into(&info, &slices[0], elem_size, raw_out)?
                    {
                        return Ok(None);
                    }
                    conversion.convert_bytes_in_place(raw_out);
                }
                _ => return Ok(None),
            }
        }
        unsafe {
            values.set_len(count);
        }
        Ok(Some(values))
    }

    fn try_read_slice_direct_1d_into<T: crate::hl::types::H5Type>(
        &self,
        shape: &[u64],
        slices: &[crate::hl::selection::SliceInfo],
        out: &mut [T],
    ) -> Result<bool> {
        if !(shape.len() == 1 && slices.len() == 1 && slices[0].step == 1) {
            return Ok(false);
        }

        let info = self.info()?;
        let conversion = crate::hl::conversion::ReadConversion::for_dataset::<T>(&info.datatype)?;
        if !conversion.is_same_size_bytes() || T::requires_validation() {
            return Ok(false);
        }
        let elem_size = usize_from_u64(u64::from(info.datatype.size), "datatype size")?;
        if elem_size == 0 {
            return Err(Error::Other("zero-size type".into()));
        }
        let raw_out = crate::hl::types::slice_as_bytes_mut(out);
        match info.layout.layout_class {
            LayoutClass::Contiguous => {
                let Some(addr) = info.layout.contiguous_addr else {
                    return Ok(false);
                };
                if crate::io::reader::is_undef_addr(addr) {
                    return Ok(false);
                }

                let start_byte = usize_from_u64(slices[0].start, "selection start")?
                    .checked_mul(elem_size)
                    .ok_or_else(|| Error::InvalidFormat("selection byte offset overflow".into()))?;
                let read_addr = addr
                    .checked_add(u64_from_usize(start_byte, "selection byte offset")?)
                    .ok_or_else(|| {
                        Error::InvalidFormat("selection read address overflow".into())
                    })?;

                let mut guard = self.inner.lock();
                guard.reader.seek(read_addr)?;
                guard.reader.read_exact(raw_out)?;
                conversion.convert_bytes_in_place(raw_out);
                Ok(true)
            }
            LayoutClass::Chunked => {
                if self.try_read_chunked_slice_1d_into(&info, &slices[0], elem_size, raw_out)? {
                    conversion.convert_bytes_in_place(raw_out);
                    return Ok(true);
                }
                Ok(false)
            }
            _ => Ok(false),
        }
    }

    fn try_read_chunked_slice_1d_into(
        &self,
        info: &DatasetInfo,
        slice: &crate::hl::selection::SliceInfo,
        elem_size: usize,
        raw_out: &mut [u8],
    ) -> Result<bool> {
        let filtered = info
            .filter_pipeline
            .as_ref()
            .map(|pipeline| !pipeline.filters.is_empty())
            .unwrap_or(false);
        if filtered {
            return Ok(false);
        }

        let raw_chunk_dims = info
            .layout
            .chunk_dims
            .as_ref()
            .ok_or_else(|| Error::InvalidFormat("chunked dataset missing chunk dims".into()))?;
        let chunk_dims = Self::chunk_data_dims(&info.dataspace.dims, raw_chunk_dims)?;
        if info.dataspace.dims.len() != 1 || chunk_dims.len() != 1 {
            return Ok(false);
        }

        Self::filled_data_into(raw_out.len() / elem_size, elem_size, info, raw_out)?;

        let selection_start = usize_from_u64(slice.start, "selection start")?;
        let selection_end = usize_from_u64(slice.end, "selection end")?;
        let chunk_len = usize_from_u64(chunk_dims[0], "chunk dimension")?;
        let data_len = usize_from_u64(info.dataspace.dims[0], "dataset dimension")?;

        let mut reads = Vec::new();
        let mut can_direct = true;
        self.visit_chunk_infos(|offset, filter_mask, addr, size| {
            if offset.len() != 1 || crate::io::reader::is_undef_addr(addr) {
                return Ok(());
            }

            let chunk_start = usize_from_u64(offset[0], "chunk coordinate")?;
            let chunk_end = chunk_start
                .checked_add(chunk_len)
                .ok_or_else(|| Error::InvalidFormat("chunk coordinate overflow".into()))?
                .min(data_len);
            let copy_start = selection_start.max(chunk_start);
            let copy_end = selection_end.min(chunk_end);
            if copy_start >= copy_end {
                return Ok(());
            }
            if filter_mask != 0 {
                can_direct = false;
                return Ok(());
            }

            let src_offset = copy_start
                .checked_sub(chunk_start)
                .and_then(|elements| elements.checked_mul(elem_size))
                .ok_or_else(|| Error::InvalidFormat("chunk slice offset overflow".into()))?;
            let copy_bytes = copy_end
                .checked_sub(copy_start)
                .and_then(|elements| elements.checked_mul(elem_size))
                .ok_or_else(|| Error::InvalidFormat("chunk slice byte count overflow".into()))?;
            let read_size = usize_from_u64(size, "chunk size")?;
            let src_end = src_offset
                .checked_add(copy_bytes)
                .ok_or_else(|| Error::InvalidFormat("chunk slice range overflow".into()))?;
            if src_end > read_size {
                can_direct = false;
                return Ok(());
            }

            let dst_offset = copy_start
                .checked_sub(selection_start)
                .and_then(|elements| elements.checked_mul(elem_size))
                .ok_or_else(|| Error::InvalidFormat("selection output offset overflow".into()))?;
            let read_addr = addr
                .checked_add(u64_from_usize(src_offset, "chunk slice offset")?)
                .ok_or_else(|| Error::InvalidFormat("chunk slice read address overflow".into()))?;
            reads.push((read_addr, dst_offset, copy_bytes));
            Ok(())
        })?;
        if !can_direct {
            return Ok(false);
        }

        let mut guard = self.inner.lock();
        for (addr, dst_offset, copy_bytes) in reads {
            let dst_end = dst_offset
                .checked_add(copy_bytes)
                .ok_or_else(|| Error::InvalidFormat("selection output range overflow".into()))?;
            guard.reader.seek(addr)?;
            guard.reader.read_exact(&mut raw_out[dst_offset..dst_end])?;
        }
        Ok(true)
    }

    fn extract_1d_selection<T: crate::hl::types::H5Type>(
        all_data: &[T],
        slice: &crate::hl::selection::SliceInfo,
    ) -> Result<Vec<T>> {
        let start = usize_from_u64(slice.start, "selection start")?;
        let end = usize_from_u64(slice.end, "selection end")?;
        if start > all_data.len() {
            return Ok(Vec::new());
        }
        let end = end.min(all_data.len());
        if slice.step == 1 {
            return copy_h5_values(&all_data[start..end], "1D selection");
        }
        let step = usize_from_u64(slice.step, "selection step")?;
        let selected = selection_len_for_step(end - start, step)?;
        let mut result = Vec::new();
        result.try_reserve_exact(selected).map_err(|err| {
            Error::InvalidFormat(format!("1D selection allocation failed: {err}"))
        })?;
        result.extend(all_data[start..end].iter().step_by(step).copied());
        Ok(result)
    }

    fn extract_1d_selection_into<T: crate::hl::types::H5Type>(
        all_data: &[T],
        slice: &crate::hl::selection::SliceInfo,
        out: &mut [T],
    ) -> Result<()> {
        let start = usize_from_u64(slice.start, "selection start")?;
        let end = usize_from_u64(slice.end, "selection end")?.min(all_data.len());
        if start > all_data.len() {
            if out.is_empty() {
                return Ok(());
            }
            return Err(Error::InvalidFormat(format!(
                "1D selection produced 0 elements, expected {}",
                out.len()
            )));
        }
        let step = usize_from_u64(slice.step, "selection step")?;
        let mut written = 0usize;
        for value in all_data[start..end].iter().step_by(step) {
            out[written] = *value;
            written += 1;
        }
        if written == out.len() {
            Ok(())
        } else {
            Err(Error::InvalidFormat(format!(
                "1D selection produced {written} elements, expected {}",
                out.len()
            )))
        }
    }

    fn extract_nd_selection<T: crate::hl::types::H5Type>(
        all_data: &[T],
        shape: &[u64],
        slices: &[crate::hl::selection::SliceInfo],
        out_shape: &[u64],
        total_out: u64,
        total_out_usize: usize,
    ) -> Result<Vec<T>> {
        let mut result = Vec::new();
        result
            .try_reserve_exact(total_out_usize)
            .map_err(|err| Error::InvalidFormat(format!("selection allocation failed: {err}")))?;
        let ndims = shape.len();

        let mut in_strides = vec![1usize; ndims];
        for d in (0..ndims - 1).rev() {
            in_strides[d] = in_strides[d + 1]
                .checked_mul(usize_from_u64(shape[d + 1], "selection shape")?)
                .ok_or_else(|| Error::InvalidFormat("selection stride overflow".into()))?;
        }

        let mut out_idx = vec![0u64; ndims];
        for _ in 0..total_out {
            let mut in_linear = 0usize;
            for d in 0..ndims {
                let in_d = out_idx[d]
                    .checked_mul(slices[d].step)
                    .and_then(|offset| slices[d].start.checked_add(offset))
                    .ok_or_else(|| Error::InvalidFormat("selection coordinate overflow".into()))?;
                let term = usize_from_u64(in_d, "selection input index")?
                    .checked_mul(in_strides[d])
                    .ok_or_else(|| {
                        Error::InvalidFormat("selection linear index overflow".into())
                    })?;
                in_linear = in_linear.checked_add(term).ok_or_else(|| {
                    Error::InvalidFormat("selection linear index overflow".into())
                })?;
            }

            if in_linear < all_data.len() {
                result.push(all_data[in_linear]);
            }

            for d in (0..ndims).rev() {
                out_idx[d] = out_idx[d].checked_add(1).ok_or_else(|| {
                    Error::InvalidFormat("selection output index overflow".into())
                })?;
                if out_idx[d] < out_shape[d] {
                    break;
                }
                out_idx[d] = 0;
            }
        }

        Ok(result)
    }

    fn extract_nd_selection_into<T: crate::hl::types::H5Type>(
        all_data: &[T],
        shape: &[u64],
        slices: &[crate::hl::selection::SliceInfo],
        out_shape: &[u64],
        total_out: u64,
        out: &mut [T],
    ) -> Result<()> {
        let ndims = shape.len();

        let mut in_strides = vec![1usize; ndims];
        for d in (0..ndims - 1).rev() {
            in_strides[d] = in_strides[d + 1]
                .checked_mul(usize_from_u64(shape[d + 1], "selection shape")?)
                .ok_or_else(|| Error::InvalidFormat("selection stride overflow".into()))?;
        }

        let mut out_idx = vec![0u64; ndims];
        let mut written = 0usize;
        for _ in 0..total_out {
            let mut in_linear = 0usize;
            for d in 0..ndims {
                let in_d = out_idx[d]
                    .checked_mul(slices[d].step)
                    .and_then(|offset| slices[d].start.checked_add(offset))
                    .ok_or_else(|| Error::InvalidFormat("selection coordinate overflow".into()))?;
                let term = usize_from_u64(in_d, "selection input index")?
                    .checked_mul(in_strides[d])
                    .ok_or_else(|| {
                        Error::InvalidFormat("selection linear index overflow".into())
                    })?;
                in_linear = in_linear.checked_add(term).ok_or_else(|| {
                    Error::InvalidFormat("selection linear index overflow".into())
                })?;
            }

            if in_linear < all_data.len() {
                out[written] = all_data[in_linear];
                written += 1;
            }

            for d in (0..ndims).rev() {
                out_idx[d] = out_idx[d].checked_add(1).ok_or_else(|| {
                    Error::InvalidFormat("selection output index overflow".into())
                })?;
                if out_idx[d] < out_shape[d] {
                    break;
                }
                out_idx[d] = 0;
            }
        }

        if written == out.len() {
            Ok(())
        } else {
            Err(Error::InvalidFormat(format!(
                "selection produced {written} elements, expected {}",
                out.len()
            )))
        }
    }

    pub(super) fn row_major_strides(dims: &[u64]) -> Result<Vec<usize>> {
        let mut strides = vec![1usize; dims.len()];
        for dim in (0..dims.len().saturating_sub(1)).rev() {
            strides[dim] = strides[dim + 1]
                .checked_mul(usize_from_u64(dims[dim + 1], "dataspace dimension")?)
                .ok_or_else(|| Error::InvalidFormat("dataspace stride overflow".into()))?;
        }
        Ok(strides)
    }

    pub(super) fn linear_index(coords: &[u64], strides: &[usize]) -> Result<usize> {
        coords
            .iter()
            .zip(strides)
            .try_fold(0usize, |acc, (&coord, &stride)| {
                acc.checked_add(
                    usize_from_u64(coord, "dataspace coordinate")
                        .ok()?
                        .checked_mul(stride)?,
                )
            })
            .ok_or_else(|| Error::InvalidFormat("linear index overflow".into()))
    }
}

fn copy_h5_values<T: crate::hl::types::H5Type>(values: &[T], context: &str) -> Result<Vec<T>> {
    let mut out = Vec::new();
    out.try_reserve_exact(values.len())
        .map_err(|err| Error::InvalidFormat(format!("{context} allocation failed: {err}")))?;
    out.extend_from_slice(values);
    Ok(out)
}

fn selection_len_for_step(len: usize, step: usize) -> Result<usize> {
    if step == 0 {
        return Err(Error::InvalidFormat(
            "selection step must be non-zero".into(),
        ));
    }
    Ok(if len == 0 { 0 } else { (len - 1) / step + 1 })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::hl::selection::SliceInfo;

    #[test]
    fn nd_slice_extraction_rejects_coordinate_overflow() {
        let slices = [SliceInfo {
            start: u64::MAX - 1,
            end: u64::MAX,
            step: 2,
        }];
        let err = Dataset::extract_nd_selection::<u8>(&[0], &[u64::MAX], &slices, &[2], 2, 2)
            .unwrap_err();
        assert!(
            err.to_string().contains("selection coordinate overflow"),
            "unexpected error: {err}"
        );
    }
}

#[cfg(test)]
mod partial_read_tests {
    use crate::hl::file::File;
    use crate::hl::selection::{HyperslabDim, Selection, SliceInfo};
    use crate::hl::writable_file::WritableFile;
    use tempfile::tempdir;

    fn write_contiguous_3x8(path: &std::path::Path) -> Vec<u16> {
        let data: Vec<u16> = (0..24).collect();
        let mut wf = WritableFile::create(path).unwrap();
        wf.new_dataset_builder("d")
            .shape(&[3, 8])
            .write::<u16>(&data)
            .unwrap(); // no .chunk() -> contiguous layout
        wf.flush().unwrap();
        data
    }

    #[test]
    fn contiguous_slab_fast_path_matches_ground_truth() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("contig.h5");
        let data = write_contiguous_3x8(&path);

        let f = File::open(&path).unwrap();
        let ds = f.dataset("d").unwrap();
        assert_eq!(ds.read::<u16>().unwrap(), data);

        // Rows 1..3, all columns: a leading-dim contiguous slab (fast path).
        let sel = Selection::Hyperslab(vec![
            HyperslabDim::new(1, 1, 2, 1), // start=1, 2 rows
            HyperslabDim::new(0, 1, 8, 1), // full columns
        ]);
        // Ground truth: rows 1 and 2 of a row-major 3x8 = elements 8..24.
        assert_eq!(
            ds.read_slice::<u16, _>(sel.clone()).unwrap(),
            data[8..24].to_vec()
        );
        let mut into = vec![0u16; 16];
        ds.read_slice_into::<u16, _>(sel, &mut into).unwrap();
        assert_eq!(into, data[8..24].to_vec());

        // Same via a step-1 Slice selection.
        let sel2 = Selection::Slice(vec![
            SliceInfo {
                start: 1,
                end: 3,
                step: 1,
            },
            SliceInfo {
                start: 0,
                end: 8,
                step: 1,
            },
        ]);
        assert_eq!(
            ds.read_slice::<u16, _>(sel2.clone()).unwrap(),
            data[8..24].to_vec()
        );
        let mut into = vec![0u16; 16];
        ds.read_slice_into::<u16, _>(sel2, &mut into).unwrap();
        assert_eq!(into, data[8..24].to_vec());
    }

    #[test]
    fn non_contiguous_region_falls_back_correctly() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("contig2.h5");
        write_contiguous_3x8(&path);

        let f = File::open(&path).unwrap();
        let ds = f.dataset("d").unwrap();

        // Sub-region cols 2..5 of rows 1..3 — a "hole" in the trailing dim, so
        // the fast path declines and the whole-read-then-slice fallback runs.
        let sel = Selection::Hyperslab(vec![
            HyperslabDim::new(1, 1, 2, 1), // rows 1,2
            HyperslabDim::new(2, 1, 3, 1), // cols 2,3,4
        ]);
        let got = ds.read_slice::<u16, _>(sel).unwrap();
        // Ground truth: row1 cols2-4 = 10,11,12 ; row2 cols2-4 = 18,19,20.
        assert_eq!(got, vec![10u16, 11, 12, 18, 19, 20]);
    }
    /// Write a 6x8 u16 dataset (chunk 3x4 -> 2x2 chunk grid unless `chunk`
    /// equals the shape) under the given index-selecting options.
    fn write_6x8(
        path: &std::path::Path,
        chunk: &[u64],
        max_shape: Option<&[u64]>,
        deflate: bool,
        shuffle: bool,
    ) -> Vec<u16> {
        let data: Vec<u16> = (0..48).collect();
        let mut wf = WritableFile::create(path).unwrap();
        let mut b = wf.new_dataset_builder("d").shape(&[6, 8]).chunk(chunk);
        if let Some(m) = max_shape {
            b = b.max_shape(m);
        }
        if shuffle {
            b = b.shuffle();
        }
        if deflate {
            b = b.deflate(4);
        }
        b.write::<u16>(&data).unwrap();
        wf.flush().unwrap();
        data
    }

    /// Open `path` and assert partial reads (a 4-chunk-spanning sub-rectangle and
    /// a full-width slab) match the ground truth derived from the row-major data.
    fn assert_partial_reads(path: &std::path::Path, data: &[u16]) {
        let f = File::open(path).unwrap();
        let ds = f.dataset("d").unwrap();
        assert_eq!(ds.read::<u16>().unwrap(), data);

        // Sub-rectangle rows 1..5, cols 2..7.
        let sel = Selection::Hyperslab(vec![
            HyperslabDim::new(1, 1, 4, 1),
            HyperslabDim::new(2, 1, 5, 1),
        ]);
        let mut expect = Vec::new();
        for r in 1..5u64 {
            for c in 2..7u64 {
                expect.push(data[(r * 8 + c) as usize]);
            }
        }
        assert_eq!(ds.read_slice::<u16, _>(sel.clone()).unwrap(), expect);
        let mut into = vec![0u16; expect.len()];
        ds.read_slice_into::<u16, _>(sel, &mut into).unwrap();
        assert_eq!(into, expect);

        // Full-width slab rows 2..4.
        let slab = Selection::Hyperslab(vec![
            HyperslabDim::new(2, 1, 2, 1),
            HyperslabDim::new(0, 1, 8, 1),
        ]);
        assert_eq!(
            ds.read_slice::<u16, _>(slab.clone()).unwrap(),
            data[16..32].to_vec()
        );
        let mut into = vec![0u16; 16];
        ds.read_slice_into::<u16, _>(slab, &mut into).unwrap();
        assert_eq!(into, data[16..32].to_vec());
    }

    #[test]
    fn chunked_fixed_array_partial_read() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("fixed.h5");
        let data = write_6x8(&path, &[3, 4], None, false, false);
        assert_partial_reads(&path, &data);
    }

    #[test]
    fn chunked_filtered_partial_read() {
        // Per-chunk filtered decode (shuffle + deflate).
        let dir = tempdir().unwrap();
        let path = dir.path().join("filtered.h5");
        let data = write_6x8(&path, &[3, 4], None, true, true);
        assert_partial_reads(&path, &data);
    }

    #[test]
    fn chunked_single_chunk_partial_read() {
        // chunk == shape -> SingleChunk index.
        let dir = tempdir().unwrap();
        let path = dir.path().join("single.h5");
        let data = write_6x8(&path, &[6, 8], None, false, false);
        assert_partial_reads(&path, &data);
    }

    #[test]
    fn chunked_extensible_array_partial_read() {
        // one unlimited dimension -> ExtensibleArray index.
        let dir = tempdir().unwrap();
        let path = dir.path().join("ext.h5");
        let data = write_6x8(&path, &[3, 4], Some(&[u64::MAX, 8]), false, false);
        assert_partial_reads(&path, &data);
    }

    #[test]
    fn chunked_btree_v2_partial_read() {
        // two unlimited dimensions -> v2 B-tree index.
        let dir = tempdir().unwrap();
        let path = dir.path().join("btv2.h5");
        let data = write_6x8(&path, &[3, 4], Some(&[u64::MAX, u64::MAX]), false, false);
        assert_partial_reads(&path, &data);
    }

    #[test]
    fn chunked_btree_v2_filtered_partial_read() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("btv2f.h5");
        let data = write_6x8(&path, &[3, 4], Some(&[u64::MAX, u64::MAX]), true, true);
        assert_partial_reads(&path, &data);
    }
}